streaming compression, query planner and replication fixes (#14023)

* streaming compression, query planner and replication fixes

* remove journal v2 stats from global statistics

* disable sql for checking past sql UUIDs

* single threaded replication

* final replication thread using dictionaries and JudyL for sorting the pending requests

* do not timeout the sending socket when there are pending replication requests

* streaming receiver using read() instead of fread()

* remove FILE * from streaming - now using posix read() and write()

* increase timeouts to 10 minutes

* apply sender timeout only when there are metrics that are supposed to be streamed

* error handling in replication

* remove retries on socket read timeout; better error messages

* take into account inbound traffic too to detect that a connection is stale

* remove race conditions from replication thread

* make sure deleted entries are marked as executed, so that even if deletion fails, they will not be executed

* 2 minutes timeout to retry streaming to a parent that already has this node

* remove unecessary condition check

* fix compilation warnings

* include judy in replication

* wrappers to handle retries for SSL_read and SSL_write

* compressed bytes read monitoring

* recursive locks on replication to make it faster during flush or cleanup

* replication completion chart at the receiver side

* simplified recursive mutex

* simplified recursive mutex again

7 files changed, 912 insertions, 551 deletions

diff --git a/streaming/compression.c b/streaming/compression.c
index 1fddc02b91..7ba9dbf198 100644
--- a/streaming/compression.c
+++ b/streaming/compression.c
@@ -5,6 +5,7 @@
 
 #define STREAM_COMPRESSION_MSG "STREAM_COMPRESSION"
 
+// signature MUST end with a newline
 #define SIGNATURE ((uint32_t)('z' | 0x80) | (0x80 << 8) | (0x80 << 16) | ('\n' << 24))
 #define SIGNATURE_MASK ((uint32_t)0xff | (0x80 << 8) | (0x80 << 16) | (0xff << 24))
 #define SIGNATURE_SIZE 4
@@ -29,7 +30,7 @@ static void lz4_compressor_reset(struct compressor_state *state)
     if (state->data) {
         if (state->data->stream) {
             LZ4_resetStream_fast(state->data->stream);            
-            info("%s: Compressor Reset", STREAM_COMPRESSION_MSG);
+            internal_error(true, "%s: compressor reset", STREAM_COMPRESSION_MSG);
         }
         state->data->input_ring_buffer_pos = 0;
     }
@@ -139,11 +140,12 @@ struct compressor_state *create_compressor()
 /*
  * LZ4 streaming API decompressor specific data
  */
-struct decompressor_data {
-    LZ4_streamDecode_t *stream;
-    char *stream_buffer;
-    size_t stream_buffer_size;
-    size_t stream_buffer_pos;
+struct decompressor_stream {
+    LZ4_streamDecode_t *lz4_stream;
+    char *buffer;
+    size_t size;
+    size_t write_at;
+    size_t read_at;
 };
 
 /*
@@ -151,12 +153,12 @@ struct decompressor_data {
  */
 static void lz4_decompressor_reset(struct decompressor_state *state)
 {
-    if (state->data) {
-        if (state->data->stream)
-           LZ4_setStreamDecode(state->data->stream, NULL, 0);
-        state->data->stream_buffer_pos = 0;
-        state->buffer_len = 0;
-        state->out_buffer_len = 0;
+    if (state->stream) {
+        if (state->stream->lz4_stream)
+           LZ4_setStreamDecode(state->stream->lz4_stream, NULL, 0);
+
+        state->stream->write_at = 0;
+        state->stream->read_at = 0;
     }
 }
 
@@ -167,177 +169,129 @@ static void lz4_decompressor_destroy(struct decompressor_state **state)
 {
     if (state && *state) {
         struct decompressor_state *s = *state;
-        if (s->data) {
+        if (s->stream) {
             debug(D_STREAM, "%s: Destroying decompressor.", STREAM_COMPRESSION_MSG);
-            if (s->data->stream)
-                LZ4_freeStreamDecode(s->data->stream);
-            freez(s->data->stream_buffer);
-            freez(s->data);
+            if (s->stream->lz4_stream)
+                LZ4_freeStreamDecode(s->stream->lz4_stream);
+            freez(s->stream->buffer);
+            freez(s->stream);
         }
-        freez(s->buffer);
         freez(s);
         *state = NULL;
     }
 }
 
-static size_t decode_compress_header(const char *data, size_t data_size)
-{
-    if (!data || !data_size)
+static size_t decode_compress_header(const char *data, size_t data_size) {
+    if (unlikely(!data || !data_size))
         return 0;
-    if (data_size < SIGNATURE_SIZE)
+
+    if (unlikely(data_size != SIGNATURE_SIZE))
         return 0;
+
     uint32_t sign = *(uint32_t *)data;
-    if ((sign & SIGNATURE_MASK) != SIGNATURE)
+    if (unlikely((sign & SIGNATURE_MASK) != SIGNATURE))
         return 0;
+
     size_t length = ((sign >> 8) & 0x7f) | ((sign >> 9) & (0x7f << 7));
     return length;
 }
 
 /*
- * Check input data for the compression header
- * Return the size of compressed data or 0 for uncompressed data
- */
-size_t is_compressed_data(const char *data, size_t data_size)
-{
-    return decode_compress_header(data, data_size);
-}
-
-/*
  * Start the collection of compressed data in an internal buffer
  * Return the size of compressed data or 0 for uncompressed data
  */
-static size_t lz4_decompressor_start(struct decompressor_state *state, const char *header, size_t header_size)
-{
-    size_t length = decode_compress_header(header, header_size);
-    if (!length)
-        return 0;
-
-    if (!state->buffer) {
-        state->buffer = mallocz(length);
-        state->buffer_size = length;
-    } else if (state->buffer_size < length) {
-        state->buffer = reallocz(state->buffer, length);
-        state->buffer_size = length;
-    }
-    state->buffer_len = length;
-    state->buffer_pos = 0;
-    state->out_buffer_pos = 0;
-    state->out_buffer_len = 0;
-    return length;
-}
-
-/*
- * Add a chunk of compressed data to the internal buffer
- * Return the current size of compressed data or 0 for error
- */
-static size_t lz4_decompressor_put(struct decompressor_state *state, const char *data, size_t size)
-{
-    if (!state || !size || !data)
-        return 0;
-    if (!state->buffer)
-        fatal("STREAM: No decompressor buffer allocated");
-
-    if (state->buffer_pos + size > state->buffer_len) {
-        error("STREAM: Decompressor buffer overflow %lu + %lu > %lu",
-                    (long unsigned int) state->buffer_pos, (long unsigned int) size,
-                    (long unsigned int) state->buffer_len);
-        size = state->buffer_len - state->buffer_pos;
-    }
-    memcpy(state->buffer + state->buffer_pos, data, size);
-    state->buffer_pos += size;
-    return state->buffer_pos;
-}
+static size_t lz4_decompressor_start(struct decompressor_state *state __maybe_unused, const char *header, size_t header_size) {
+    if(unlikely(state->stream->read_at != state->stream->write_at))
+        fatal("%s: asked to decompress new data, while there are unread data in the decompression buffer!"
+        , STREAM_COMPRESSION_MSG);
 
-static size_t saving_percent(size_t comp_len, size_t src_len)
-{
-    if (comp_len > src_len)
-        comp_len = src_len;
-    if (!src_len)
-        return 0;
-    return 100 - comp_len * 100 / src_len;
+    return decode_compress_header(header, header_size);
 }
 
 /*
  * Decompress the compressed data in the internal buffer
  * Return the size of uncompressed data or 0 for error
  */
-static size_t lz4_decompressor_decompress(struct decompressor_state *state)
-{
-    if (!state)
-        return 0;
-    if (!state->buffer) {
-        error("%s: No decompressor buffer allocated", STREAM_COMPRESSION_MSG);
+static size_t lz4_decompressor_decompress(struct decompressor_state *state, const char *compressed_data, size_t compressed_size) {
+    if (unlikely(!state || !compressed_data || !compressed_size))
         return 0;
+
+    if(unlikely(state->stream->read_at != state->stream->write_at))
+        fatal("%s: asked to decompress new data, while there are unread data in the decompression buffer!"
+              , STREAM_COMPRESSION_MSG);
+
+    if (unlikely(state->stream->write_at >= state->stream->size / 2)) {
+        state->stream->write_at = 0;
+        state->stream->read_at = 0;
     }
-    
-    long int decompressed_size = LZ4_decompress_safe_continue(state->data->stream, state->buffer,
-            state->data->stream_buffer + state->data->stream_buffer_pos,
-            state->buffer_len, state->data->stream_buffer_size - state->data->stream_buffer_pos);
-    if (decompressed_size < 0) {
-        error("%s: Decompressor error %ld", STREAM_COMPRESSION_MSG, decompressed_size);
+
+    long int decompressed_size = LZ4_decompress_safe_continue(
+            state->stream->lz4_stream
+            , compressed_data
+            , state->stream->buffer + state->stream->write_at
+            , (int)compressed_size
+            , (int)(state->stream->size - state->stream->write_at)
+            );
+
+    if (unlikely(decompressed_size < 0)) {
+        error("%s: decompressor returned negative decompressed bytes: %ld", STREAM_COMPRESSION_MSG, decompressed_size);
         return 0;
     }
 
-    state->out_buffer = state->data->stream_buffer + state->data->stream_buffer_pos;
-    state->data->stream_buffer_pos += decompressed_size;
-    if (state->data->stream_buffer_pos >= state->data->stream_buffer_size - COMPRESSION_MAX_MSG_SIZE)
-        state->data->stream_buffer_pos = 0;
-    state->out_buffer_len = decompressed_size;
-    state->out_buffer_pos = 0;
+    if(unlikely(decompressed_size + state->stream->write_at > state->stream->size))
+        fatal("%s: decompressor overflown the stream_buffer. size: %zu, pos: %zu, added: %ld, exceeding the buffer by %zu"
+              , STREAM_COMPRESSION_MSG
+              , state->stream->size
+              , state->stream->write_at
+              , decompressed_size
+              , state->stream->write_at + decompressed_size - state->stream->size
+              );
 
-    // Some compression statistics
-    size_t old_avg_saving = saving_percent(state->total_compressed, state->total_uncompressed);
-    size_t old_avg_size = state->packet_count ? state->total_uncompressed / state->packet_count : 0;
+    state->stream->write_at += decompressed_size;
 
-    state->total_compressed += state->buffer_len + SIGNATURE_SIZE;
+    // statistics
+    state->total_compressed += compressed_size + SIGNATURE_SIZE;
     state->total_uncompressed += decompressed_size;
     state->packet_count++;
 
-    size_t saving = saving_percent(state->buffer_len, decompressed_size);
-    size_t avg_saving = saving_percent(state->total_compressed, state->total_uncompressed);
-    size_t avg_size = state->total_uncompressed / state->packet_count;
-
-    (void)saving;
-
-    if (old_avg_saving != avg_saving || old_avg_size != avg_size){
-        debug(D_STREAM, "%s: Saving: %lu%% (avg. %lu%%), avg.size: %lu", STREAM_COMPRESSION_MSG,
-              (long unsigned int) saving, (long unsigned int) avg_saving, (long unsigned int) avg_size);
-    }
     return decompressed_size;
 }
 
 /*
  * Return the size of uncompressed data left in the internal buffer or 0 for error
  */
-static size_t lz4_decompressor_decompressed_bytes_in_buffer(struct decompressor_state *state)
-{
-    return state->out_buffer_len ?
-        state->out_buffer_len - state->out_buffer_pos : 0;
+static size_t lz4_decompressor_decompressed_bytes_in_buffer(struct decompressor_state *state) {
+    if(unlikely(state->stream->read_at > state->stream->write_at))
+        fatal("%s: invalid read/write stream positions"
+        , STREAM_COMPRESSION_MSG);
+
+    return state->stream->write_at - state->stream->read_at;
 }
 
 /*
  * Fill the buffer provided with uncompressed data from the internal buffer
  * Return the size of uncompressed data copied or 0 for error
  */
-static size_t lz4_decompressor_get(struct decompressor_state *state, char *data, size_t size)
-{
-    if (!state || !size || !data)
+static size_t lz4_decompressor_get(struct decompressor_state *state, char *dst, size_t size) {
+    if (unlikely(!state || !size || !dst))
         return 0;
-    if (!state->out_buffer)
-        fatal("%s: No decompressor output buffer allocated", STREAM_COMPRESSION_MSG);
-    if (state->out_buffer_pos + size > state->out_buffer_len)
-        size = state->out_buffer_len - state->out_buffer_pos;
-    
-    char *p = state->out_buffer + state->out_buffer_pos, *endp = p + size, *last_lf = NULL;
-    for (; p < endp; ++p)
-        if (*p == '\n' || *p == 0)
-            last_lf = p;
-    if (last_lf)
-        size = last_lf + 1 - (state->out_buffer + state->out_buffer_pos);
-
-    memcpy(data, state->out_buffer + state->out_buffer_pos, size);
-    state->out_buffer_pos += size;
-    return size;
+
+    size_t remaining = lz4_decompressor_decompressed_bytes_in_buffer(state);
+    if(unlikely(!remaining))
+        return 0;
+
+    size_t bytes_to_return = size;
+    if(bytes_to_return > remaining)
+        bytes_to_return = remaining;
+
+    memcpy(dst, state->stream->buffer + state->stream->read_at, bytes_to_return);
+    state->stream->read_at += bytes_to_return;
+
+    if(unlikely(state->stream->read_at > state->stream->write_at))
+        fatal("%s: invalid read/write stream positions"
+        , STREAM_COMPRESSION_MSG);
+
+    return bytes_to_return;
 }
 
 /*
@@ -347,20 +301,20 @@ static size_t lz4_decompressor_get(struct decompressor_state *state, char *data,
 struct decompressor_state *create_decompressor()
 {
     struct decompressor_state *state = callocz(1, sizeof(struct decompressor_state));
+    state->signature_size = SIGNATURE_SIZE;
     state->reset = lz4_decompressor_reset;
     state->start = lz4_decompressor_start;
-    state->put = lz4_decompressor_put;
     state->decompress = lz4_decompressor_decompress;
     state->get = lz4_decompressor_get;
     state->decompressed_bytes_in_buffer = lz4_decompressor_decompressed_bytes_in_buffer;
     state->destroy = lz4_decompressor_destroy;
 
-    state->data = callocz(1, sizeof(struct decompressor_data));
-    fatal_assert(state->data);
-    state->data->stream = LZ4_createStreamDecode();
-    state->data->stream_buffer_size = LZ4_decoderRingBufferSize(COMPRESSION_MAX_MSG_SIZE);
-    state->data->stream_buffer = mallocz(state->data->stream_buffer_size);
-    fatal_assert(state->data->stream_buffer);
+    state->stream = callocz(1, sizeof(struct decompressor_stream));
+    fatal_assert(state->stream);
+    state->stream->lz4_stream = LZ4_createStreamDecode();
+    state->stream->size = LZ4_decoderRingBufferSize(COMPRESSION_MAX_MSG_SIZE) * 2;
+    state->stream->buffer = mallocz(state->stream->size);
+    fatal_assert(state->stream->buffer);
     state->reset(state);
     debug(D_STREAM, "%s: Initialize streaming decompression!", STREAM_COMPRESSION_MSG);
     return state;
diff --git a/streaming/receiver.c b/streaming/receiver.c
index 40673f05b4..32b510e7ce 100644
--- a/streaming/receiver.c
+++ b/streaming/receiver.c
@@ -3,7 +3,12 @@
 #include "rrdpush.h"
 #include "parser/parser.h"
 
+// IMPORTANT: to add workers, you have to edit WORKER_PARSER_FIRST_JOB accordingly
 #define WORKER_RECEIVER_JOB_BYTES_READ (WORKER_PARSER_FIRST_JOB - 1)
+#define WORKER_RECEIVER_JOB_BYTES_UNCOMPRESSED (WORKER_PARSER_FIRST_JOB - 2)
+
+// this has to be the same at parser.h
+#define WORKER_RECEIVER_JOB_REPLICATION_COMPLETION (WORKER_PARSER_FIRST_JOB - 3)
 
 #if WORKER_PARSER_FIRST_JOB < 1
 #error The define WORKER_PARSER_FIRST_JOB needs to be at least 1
@@ -110,185 +115,182 @@ PARSER_RC streaming_claimed_id(char **words, size_t num_words, void *user)
     return PARSER_RC_OK;
 }
 
+static int read_stream(struct receiver_state *r, char* buffer, size_t size) {
+    if(unlikely(!size)) {
+        internal_error(true, "%s() asked to read zero bytes", __FUNCTION__);
+        return 0;
+    }
 
-#ifndef ENABLE_COMPRESSION
-/* The receiver socket is blocking, perform a single read into a buffer so that we can reassemble lines for parsing.
- */
-static int receiver_read(struct receiver_state *r, FILE *fp) {
 #ifdef ENABLE_HTTPS
-    if (r->ssl.conn && !r->ssl.flags) {
-        ERR_clear_error();
-        int desired = sizeof(r->read_buffer) - r->read_len - 1;
-        int ret = SSL_read(r->ssl.conn, r->read_buffer + r->read_len, desired);
-        if (ret > 0 ) {
-            r->read_len += ret;
-            worker_set_metric(WORKER_RECEIVER_JOB_BYTES_READ, ret);
-            return 0;
-        }
-        // Don't treat SSL_ERROR_WANT_READ or SSL_ERROR_WANT_WRITE differently on blocking socket
-        u_long err;
-        char buf[256];
-        while ((err = ERR_get_error()) != 0) {
-            ERR_error_string_n(err, buf, sizeof(buf));
-            error("STREAM %s [receive from %s] ssl error: %s", r->hostname, r->client_ip, buf);
-        }
-        return 1;
-    }
+    if (r->ssl.conn && r->ssl.flags == NETDATA_SSL_HANDSHAKE_COMPLETE)
+        return (int)netdata_ssl_read(r->ssl.conn, buffer, size);
 #endif
-    if (!fgets(r->read_buffer, sizeof(r->read_buffer), fp))
-        return 1;
-    r->read_len = strlen(r->read_buffer);
-    worker_set_metric(WORKER_RECEIVER_JOB_BYTES_READ, r->read_len);
-    return 0;
-}
-#else
-/*
- * The receiver socket is blocking, perform a single read into a buffer so that we can reassemble lines for parsing.
- * if SSL encryption is on, then use SSL API for reading stream data.
- * Use line oriented fgets() in buffer from receiver_state is provided.
- * In other cases use fread to read binary data from socket.
- * Return zero on success and the number of bytes were read using pointer in the last argument.
- */
-static int read_stream(struct receiver_state *r, FILE *fp, char* buffer, size_t size, int* ret) {
-    if (!ret)
-        return 1;
-    *ret = 0;
-#ifdef ENABLE_HTTPS
-    if (r->ssl.conn && !r->ssl.flags) {
-        ERR_clear_error();
-        if (buffer != r->read_buffer + r->read_len) {
-            *ret = SSL_read(r->ssl.conn, buffer, size);
-            if (*ret > 0 ) 
-                return 0;
-        } else {
-            // we need to receive data with LF to parse compression header
-            size_t ofs = 0;
-            int res = 0;
-            errno = 0;
-            while (ofs < size) {
-                do {
-                    res = SSL_read(r->ssl.conn, buffer + ofs, 1);
-                    // When either SSL_ERROR_SYSCALL (OpenSSL < 3.0) or SSL_ERROR_SSL(OpenSSL > 3.0) happens,
-                    // the connection was lost https://www.openssl.org/docs/man3.0/man3/SSL_get_error.html,
-                    // without the test we will have an infinite loop https://github.com/netdata/netdata/issues/13092
-                    int local_ssl_err = SSL_get_error(r->ssl.conn, res);
-                    if (local_ssl_err == SSL_ERROR_SYSCALL || local_ssl_err == SSL_ERROR_SSL) {
-                        error("The SSL connection has error SSL_ERROR_SYSCALL(%d) and system is registering errno = %d",
-                              local_ssl_err, errno);
-                        return 1;
-                    }
-                } while (res == 0);
-
-                if (res < 0)
-                    break;
-                if (buffer[ofs] == '\n')
-                    break;
-                ofs += res;
-            }
-            if (res > 0) {
-                ofs += res;
-                *ret = ofs;
-                buffer[ofs] = 0;
-                return 0;
-            }
-        }
-        // Don't treat SSL_ERROR_WANT_READ or SSL_ERROR_WANT_WRITE differently on blocking socket
-        u_long err;
-        char buf[256];
-        while ((err = ERR_get_error()) != 0) {
-            ERR_error_string_n(err, buf, sizeof(buf));
-            error("STREAM %s [receive from %s] ssl error: %s", r->hostname, r->client_ip, buf);
-        }
-        return 1;
+
+    ssize_t bytes_read = read(r->fd, buffer, size);
+    if(bytes_read == 0 && (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINPROGRESS)) {
+        error("STREAM: %s(): timeout while waiting for data on socket!", __FUNCTION__);
+        bytes_read = -3;
     }
-#endif
-    if (buffer != r->read_buffer + r->read_len) {
-        // read to external buffer
-        *ret = fread(buffer, 1, size, fp);
-        if (!*ret)
-            return 1;
-    } else {
-        if (!fgets(r->read_buffer, sizeof(r->read_buffer), fp))
-            return 1;
-        *ret = strlen(r->read_buffer);
+    else if (bytes_read == 0) {
+        error("STREAM: %s(): EOF while reading data from socket!", __FUNCTION__);
+        bytes_read = -1;
     }
-    return 0;
+    else if (bytes_read < 0) {
+        error("STREAM: %s() failed to read from socket!", __FUNCTION__);
+        bytes_read = -2;
+    }
+
+//    do {
+//        bytes_read = (int) fread(buffer, 1, size, fp);
+//        if (unlikely(bytes_read <= 0)) {
+//            if(feof(fp)) {
+//                internal_error(true, "%s(): fread() failed with EOF", __FUNCTION__);
+//                bytes_read = -2;
+//            }
+//            else if(ferror(fp)) {
+//                internal_error(true, "%s(): fread() failed with ERROR", __FUNCTION__);
+//                bytes_read = -3;
+//            }
+//            else bytes_read = 0;
+//        }
+//        else
+//            worker_set_metric(WORKER_RECEIVER_JOB_BYTES_READ, bytes_read);
+//    } while(bytes_read == 0);
+
+    return (int)bytes_read;
 }
 
-/*
- * Get the next line of data for parsing.
- * Return data from the decompressor buffer if available.
- * Otherwise read next line from the socket and check for compression header.
- * Return the line was read If no compression header was found.
- * Otherwise read the entire block of compressed data, decompress it
- * and return it in receiver_state buffer.
- * Return zero on success.
- */
-static int receiver_read(struct receiver_state *r, FILE *fp) {
-    // check any decompressed data  present
-    if (r->decompressor && r->decompressor->decompressed_bytes_in_buffer(r->decompressor)) {
-        size_t available = sizeof(r->read_buffer) - r->read_len;
+static bool receiver_read_uncompressed(struct receiver_state *r) {
+#ifdef NETDATA_INTERNAL_CHECKS
+    if(r->read_buffer[r->read_len] != '\0')
+        fatal("%s(): read_buffer does not start with zero", __FUNCTION__ );
+#endif
+
+    int bytes_read = read_stream(r, r->read_buffer + r->read_len, sizeof(r->read_buffer) - r->read_len - 1);
+    if(unlikely(bytes_read <= 0))
+        return false;
+
+    worker_set_metric(WORKER_RECEIVER_JOB_BYTES_READ, (NETDATA_DOUBLE)bytes_read);
+    worker_set_metric(WORKER_RECEIVER_JOB_BYTES_UNCOMPRESSED, (NETDATA_DOUBLE)bytes_read);
+
+    r->read_len += bytes_read;
+    r->read_buffer[r->read_len] = '\0';
+
+    return true;
+}
+
+#ifdef ENABLE_COMPRESSION
+static bool receiver_read_compressed(struct receiver_state *r) {
+
+#ifdef NETDATA_INTERNAL_CHECKS
+    if(r->read_buffer[r->read_len] != '\0')
+        fatal("%s: read_buffer does not start with zero #2", __FUNCTION__ );
+#endif
+
+    // first use any available uncompressed data
+    if (r->decompressor->decompressed_bytes_in_buffer(r->decompressor)) {
+        size_t available = sizeof(r->read_buffer) - r->read_len - 1;
         if (available) {
             size_t len = r->decompressor->get(r->decompressor, r->read_buffer + r->read_len, available);
             if (!len) {
-                internal_error(true, "decompressor returned zero length");
-                return 1;
+                internal_error(true, "decompressor returned zero length #1");
+                return false;
             }
 
-            r->read_len += len;
+            r->read_len += (int)len;
+            r->read_buffer[r->read_len] = '\0';
         }
-        return 0;
+        else
+            internal_error(true, "The line to read is too big! Already have %d bytes in read_buffer.", r->read_len);
+
+        return true;
     }
 
-    int ret = 0;
-    if (read_stream(r, fp, r->read_buffer + r->read_len, sizeof(r->read_buffer) - r->read_len - 1, &ret)) {
-        internal_error(true, "read_stream() failed (1).");
-        return 1;
+    // no decompressed data available
+    // read the compression signature of the next block
+
+    if(unlikely(r->read_len + r->decompressor->signature_size > sizeof(r->read_buffer) - 1)) {
+        internal_error(true, "The last incomplete line does not leave enough room for the next compression header! Already have %d bytes in read_buffer.", r->read_len);
+        return false;
     }
 
-    worker_set_metric(WORKER_RECEIVER_JOB_BYTES_READ, ret);
+    // read the compression signature from the stream
+    // we have to do a loop here, because read_stream() may return less than the data we need
+    int bytes_read = 0;
+    do {
+        int ret = read_stream(r, r->read_buffer + r->read_len + bytes_read, r->decompressor->signature_size - bytes_read);
+        if (unlikely(ret <= 0))
+            return false;
 
-    if (!is_compressed_data(r->read_buffer, ret)) {
-        r->read_len += ret;
-        return 0;
+        bytes_read += ret;
+    } while(unlikely(bytes_read < (int)r->decompressor->signature_size));
+
+    worker_set_metric(WORKER_RECEIVER_JOB_BYTES_READ, (NETDATA_DOUBLE)bytes_read);
+
+    if(unlikely(bytes_read != (int)r->decompressor->signature_size))
+        fatal("read %d bytes, but expected compression signature of size %zu", bytes_read, r->decompressor->signature_size);
+
+    size_t compressed_message_size = r->decompressor->start(r->decompressor, r->read_buffer + r->read_len, bytes_read);
+    if (unlikely(!compressed_message_size)) {
+        internal_error(true, "multiplexed uncompressed data in compressed stream!");
+        r->read_len += bytes_read;
+        r->read_buffer[r->read_len] = '\0';
+        return true;
+    }
+
+    if(unlikely(compressed_message_size > COMPRESSION_MAX_MSG_SIZE)) {
+        error("received a compressed message of %zu bytes, which is bigger than the max compressed message size supported of %zu. Ignoring message.",
+              compressed_message_size, (size_t)COMPRESSION_MAX_MSG_SIZE);
+        return false;
     }
 
-    if (unlikely(!r->decompressor)) 
-        r->decompressor = create_decompressor();
-    
-    size_t bytes_to_read = r->decompressor->start(r->decompressor, r->read_buffer, ret);
+    // delete compression header from our read buffer
+    r->read_buffer[r->read_len] = '\0';
 
-    // Read the entire block of compressed data because
-    // we're unable to decompress incomplete block
-    char compressed[bytes_to_read];
+    // Read the entire compressed block of compressed data
+    char compressed[compressed_message_size];
+    size_t compressed_bytes_read = 0;
     do {
-        if (read_stream(r, fp, compressed, bytes_to_read, &ret)) {
-            internal_error(true, "read_stream() failed (2).");
-            return 1;
+        size_t start = compressed_bytes_read;
+        size_t remaining = compressed_message_size - start;
+
+        int last_read_bytes = read_stream(r, &compressed[start], remaining);
+        if (unlikely(last_read_bytes <= 0)) {
+            internal_error(true, "read_stream() failed #2, with code %d", last_read_bytes);
+            return false;
         }
 
-        worker_set_metric(WORKER_RECEIVER_JOB_BYTES_READ, ret);
+        compressed_bytes_read += last_read_bytes;
 
-        // Send input data to decompressor
-        if (ret)
-            r->decompressor->put(r->decompressor, compressed, ret);
+    } while(unlikely(compressed_message_size > compressed_bytes_read));
 
-        bytes_to_read -= ret;
-    } while (bytes_to_read > 0);
+    worker_set_metric(WORKER_RECEIVER_JOB_BYTES_READ, (NETDATA_DOUBLE)compressed_bytes_read);
 
-    // Decompress
-    size_t bytes_to_parse = r->decompressor->decompress(r->decompressor);
+    // decompress the compressed block
+    size_t bytes_to_parse = r->decompressor->decompress(r->decompressor, compressed, compressed_bytes_read);
     if (!bytes_to_parse) {
         internal_error(true, "no bytes to parse.");
-        return 1;
+        return false;
     }
 
-    // Fill read buffer with decompressed data
-    r->read_len = r->decompressor->get(r->decompressor, r->read_buffer, sizeof(r->read_buffer));
-    return 0;
-}
+    worker_set_metric(WORKER_RECEIVER_JOB_BYTES_UNCOMPRESSED, (NETDATA_DOUBLE)bytes_to_parse);
 
-#endif
+    // fill read buffer with decompressed data
+    size_t len = (int)r->decompressor->get(r->decompressor, r->read_buffer + r->read_len, sizeof(r->read_buffer) - r->read_len - 1);
+    if (!len) {
+        internal_error(true, "decompressor returned zero length #2");
+        return false;
+    }
+    r->read_len += (int)len;
+    r->read_buffer[r->read_len] = '\0';
+
+    return true;
+}
+#else // !ENABLE_COMPRESSION
+static bool receiver_read_compressed(struct receiver_state *r) {
+    return receiver_read_uncompressed(r);
+}
+#endif // ENABLE_COMPRESSION
 
 /* Produce a full line if one exists, statefully return where we start next time.
  * When we hit the end of the buffer with a partial line move it to the beginning for the next fill.
@@ -302,7 +304,10 @@ static char *receiver_next_line(struct receiver_state *r, char *buffer, size_t b
     char *de = &buffer[buffer_length - 2];
 
     if(ss >= se) {
+        *ds = '\0';
+        *pos = 0;
         r->read_len = 0;
+        r->read_buffer[r->read_len] = '\0';
         return NULL;
     }
 
@@ -333,6 +338,9 @@ static char *receiver_next_line(struct receiver_state *r, char *buffer, size_t b
     // move everything to the beginning
     memmove(r->read_buffer, &r->read_buffer[start], r->read_len - start);
     r->read_len -= (int)start;
+    r->read_buffer[r->read_len] = '\0';
+    *ds = '\0';
+    *pos = 0;
     return NULL;
 }
 
@@ -342,7 +350,7 @@ static void streaming_parser_thread_cleanup(void *ptr) {
     parser_destroy(parser);
 }
 
-static size_t streaming_parser(struct receiver_state *rpt, struct plugind *cd, FILE *fp_in, FILE *fp_out, void *ssl) {
+static size_t streaming_parser(struct receiver_state *rpt, struct plugind *cd, int fd, void *ssl) {
     size_t result;
 
     PARSER_USER_OBJECT user = {
@@ -353,7 +361,7 @@ static size_t streaming_parser(struct receiver_state *rpt, struct plugind *cd, F
         .trust_durations = 1
     };
 
-    PARSER *parser = parser_init(rpt->host, &user, fp_in, fp_out, PARSER_INPUT_SPLIT, ssl);
+    PARSER *parser = parser_init(rpt->host, &user, NULL, NULL, fd, PARSER_INPUT_SPLIT, ssl);
 
     rrd_collector_started();
 
@@ -365,36 +373,56 @@ static size_t streaming_parser(struct receiver_state *rpt, struct plugind *cd, F
 
     user.parser = parser;
 
+    bool compressed_connection = false;
 #ifdef ENABLE_COMPRESSION
-    if (rpt->decompressor)
-        rpt->decompressor->reset(rpt->decompressor);
+    if(stream_has_capability(rpt, STREAM_CAP_COMPRESSION)) {
+        compressed_connection = true;
+
+        if (!rpt->decompressor)
+            rpt->decompressor = create_decompressor();
+        else
+            rpt->decompressor->reset(rpt->decompressor);
+    }
 #endif
 
-    char buffer[PLUGINSD_LINE_MAX + 2];
-    do {
-        if(receiver_read(rpt, fp_in)) break;
+    rpt->read_buffer[0] = '\0';
+    rpt->read_len = 0;
 
-        size_t pos = 0;
-        while(receiver_next_line(rpt, buffer, PLUGINSD_LINE_MAX + 2, &pos)) {
-            if(unlikely(netdata_exit)) {
-                internal_error(true, "exiting...");
-                goto done;
-            }
-            if(unlikely(rpt->shutdown)) {
-                internal_error(true, "parser shutdown...");
-                goto done;
-            }
-            if (unlikely(parser_action(parser,  buffer))) {
-                internal_error(true, "parser_action() failed...");
-                goto done;
-            }
+    size_t read_buffer_start = 0;
+    char buffer[PLUGINSD_LINE_MAX + 2] = "";
+    while(!netdata_exit) {
+        if(!receiver_next_line(rpt, buffer, PLUGINSD_LINE_MAX + 2, &read_buffer_start)) {
+            bool have_new_data;
+            if(compressed_connection)
+                have_new_data = receiver_read_compressed(rpt);
+            else
+                have_new_data = receiver_read_uncompressed(rpt);
+
+            if(!have_new_data)
+                break;
+
+            rpt->last_msg_t = now_realtime_sec();
+            continue;
         }
 
-        rpt->last_msg_t = now_realtime_sec();
+        if(unlikely(netdata_exit)) {
+            internal_error(true, "exiting...");
+            goto done;
+        }
+        if(unlikely(rpt->shutdown)) {
+            internal_error(true, "parser shutdown...");
+            goto done;
+        }
+
+        if (unlikely(parser_action(parser,  buffer))) {
+            internal_error(true, "parser_action() failed on keyword '%s'.", buffer);
+            break;
+        }
     }
-    while(!netdata_exit);
 
 done:
+    internal_error(true, "Streaming receiver thread stopping...");
+
     result = user.count;
 
     // free parser with the pop function
@@ -644,42 +672,11 @@ static int rrdpush_receive(struct receiver_state *rpt)
         error("STREAM %s [receive from [%s]:%s]: cannot remove the non-blocking flag from socket %d", rrdhost_hostname(rpt->host), rpt->client_ip, rpt->client_port, rpt->fd);
 
     struct timeval timeout;
-    timeout.tv_sec = 120;
+    timeout.tv_sec = 600;
     timeout.tv_usec = 0;
     if (unlikely(setsockopt(rpt->fd, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof timeout) != 0))
         error("STREAM %s [receive from [%s]:%s]: cannot set timeout for socket %d", rrdhost_hostname(rpt->host), rpt->client_ip, rpt->client_port, rpt->fd);
 
-    // convert the socket to a FILE *
-    // It seems that the same FILE * cannot be used for both reading and writing.
-    // (reads and writes seem to interfere with each other, with undefined results).
-
-    int fd_in = rpt->fd;
-    int fd_out = fcntl(rpt->fd, F_DUPFD_CLOEXEC, 0);