use gperf for the pluginsd/streaming parser hashtable (#15251)

* use gperf for the pluginsd parser

* simplify pluginsd_parser by removing void pointers to user

* pluginsd_split_words() with inlined pluginsd_space()

* quoted_string_splitter() now uses a map instead of a function for determining spaces

* add stress test for pluginsd parser

* optimized BITMAP256

* optimized rrdpush receiver reception

* optimized rrdpush sender compression

* renames and cleanup

* remove wrong negation

* unify handshake and disconnection reasons

* use parser_find_keyword

* register job names only for the current repertoire

5 files changed, 324 insertions, 427 deletions

diff --git a/streaming/compression.c b/streaming/compression.c
index 8f2517a8e3..29bf88af37 100644
--- a/streaming/compression.c
+++ b/streaming/compression.c
@@ -5,55 +5,40 @@
 
 #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
-
-
-/*
- * LZ4 streaming API compressor specific data
- */
-struct compressor_data {
-    LZ4_stream_t *stream;
-    char *input_ring_buffer;
-    size_t input_ring_buffer_size;
-    size_t input_ring_buffer_pos;
-};
-
-
 /*
  * Reset compressor state for a new stream
  */
-static void lz4_compressor_reset(struct compressor_state *state)
-{
-    if (state->data) {
-        if (state->data->stream) {
-            LZ4_resetStream_fast(state->data->stream);            
-            internal_error(true, "%s: compressor reset", STREAM_COMPRESSION_MSG);
-        }
-        state->data->input_ring_buffer_pos = 0;
+void rrdpush_compressor_reset(struct compressor_state *state) {
+    if(!state->initialized) {
+        state->initialized = true;
+
+        state->stream.lz4_stream = LZ4_createStream();
+        state->stream.input_ring_buffer_size = LZ4_DECODER_RING_BUFFER_SIZE(COMPRESSION_MAX_MSG_SIZE * 2);
+        state->stream.input_ring_buffer = callocz(1, state->stream.input_ring_buffer_size);
+        state->compression_result_buffer_size = 0;
     }
+
+    LZ4_resetStream_fast(state->stream.lz4_stream);
+
+    state->stream.input_ring_buffer_pos = 0;
 }
 
 /*
  * Destroy compressor state and all related data
  */
-static void lz4_compressor_destroy(struct compressor_state **state)
-{
-    if (state && *state) {
-        struct compressor_state *s = *state;
-        if (s->data) {
-            if (s->data->stream)
-                LZ4_freeStream(s->data->stream);
-            freez(s->data->input_ring_buffer);
-            freez(s->data);
-        }
-        freez(s->compression_result_buffer);
-        freez(s);
-        *state = NULL;
-        debug(D_STREAM, "%s: Compressor Destroyed.", STREAM_COMPRESSION_MSG);
+void rrdpush_compressor_destroy(struct compressor_state *state) {
+    if (state->stream.lz4_stream) {
+        LZ4_freeStream(state->stream.lz4_stream);
+        state->stream.lz4_stream = NULL;
     }
+
+    freez(state->stream.input_ring_buffer);
+    state->stream.input_ring_buffer = NULL;
+
+    freez(state->compression_result_buffer);
+    state->compression_result_buffer = NULL;
+
+    state->initialized = false;
 }
 
 /*
@@ -62,18 +47,18 @@ static void lz4_compressor_destroy(struct compressor_state **state)
  * Return the size of compressed data block as result and the pointer to internal buffer using the last argument
  * or 0 in case of error
  */
-static size_t lz4_compressor_compress(struct compressor_state *state, const char *data, size_t size, char **out)
-{
+size_t rrdpush_compress(struct compressor_state *state, const char *data, size_t size, char **out) {
     if(unlikely(!state || !size || !out))
         return 0;
 
     if(unlikely(size > COMPRESSION_MAX_MSG_SIZE)) {
-        error("%s: Compression Failed - Message size %lu above compression buffer limit: %d", STREAM_COMPRESSION_MSG, (long unsigned int)size, COMPRESSION_MAX_MSG_SIZE);
+        error("RRDPUSH COMPRESS: Compression Failed - Message size %lu above compression buffer limit: %d",
+              (long unsigned int)size, COMPRESSION_MAX_MSG_SIZE);
         return 0;
     }
 
     size_t max_dst_size = LZ4_COMPRESSBOUND(size);
-    size_t data_size = max_dst_size + SIGNATURE_SIZE;
+    size_t data_size = max_dst_size + RRDPUSH_COMPRESSION_SIGNATURE_SIZE;
 
     if (!state->compression_result_buffer) {
         state->compression_result_buffer = mallocz(data_size);
@@ -85,16 +70,16 @@ static size_t lz4_compressor_compress(struct compressor_state *state, const char
     }
 
     // the ring buffer always has space for LZ4_MAX_MSG_SIZE
-    memcpy(state->data->input_ring_buffer + state->data->input_ring_buffer_pos, data, size);
+    memcpy(state->stream.input_ring_buffer + state->stream.input_ring_buffer_pos, data, size);
 
     // this call needs the last 64K of our previous data
     // they are available in the ring buffer
     long int compressed_data_size = LZ4_compress_fast_continue(
-        state->data->stream,
-        state->data->input_ring_buffer + state->data->input_ring_buffer_pos,
-        state->compression_result_buffer + SIGNATURE_SIZE,
-        size,
-        max_dst_size,
+        state->stream.lz4_stream,
+        state->stream.input_ring_buffer + state->stream.input_ring_buffer_pos,
+        state->compression_result_buffer + RRDPUSH_COMPRESSION_SIGNATURE_SIZE,
+        (int)size,
+        (int)max_dst_size,
         1);
 
     if (compressed_data_size < 0) {
@@ -103,220 +88,94 @@ static size_t lz4_compressor_compress(struct compressor_state *state, const char
     }
 
     // update the next writing position of the ring buffer
-    state->data->input_ring_buffer_pos += size;
-    if(unlikely(state->data->input_ring_buffer_pos >= state->data->input_ring_buffer_size - COMPRESSION_MAX_MSG_SIZE))
-        state->data->input_ring_buffer_pos = 0;
+    state->stream.input_ring_buffer_pos += size;
+    if(unlikely(state->stream.input_ring_buffer_pos >= state->stream.input_ring_buffer_size - COMPRESSION_MAX_MSG_SIZE))
+        state->stream.input_ring_buffer_pos = 0;
 
     // update the signature header
     uint32_t len = ((compressed_data_size & 0x7f) | 0x80 | (((compressed_data_size & (0x7f << 7)) << 1) | 0x8000)) << 8;
-    *(uint32_t *)state->compression_result_buffer = len | SIGNATURE;
+    *(uint32_t *)state->compression_result_buffer = len | RRDPUSH_COMPRESSION_SIGNATURE;
     *out = state->compression_result_buffer;
     debug(D_STREAM, "%s: Compressed data header: %ld", STREAM_COMPRESSION_MSG, compressed_data_size);
-    return compressed_data_size + SIGNATURE_SIZE;
-}
-
-/*
- * Create and initialize compressor state
- * Return the pointer to compressor_state structure created
- */
-struct compressor_state *create_compressor()
-{
-    struct compressor_state *state = callocz(1, sizeof(struct compressor_state));
-
-    state->reset = lz4_compressor_reset;
-    state->compress = lz4_compressor_compress;
-    state->destroy = lz4_compressor_destroy;
-
-    state->data = callocz(1, sizeof(struct compressor_data));
-    state->data->stream = LZ4_createStream();
-    state->data->input_ring_buffer_size = LZ4_DECODER_RING_BUFFER_SIZE(COMPRESSION_MAX_MSG_SIZE * 2);
-    state->data->input_ring_buffer = callocz(1, state->data->input_ring_buffer_size);
-    state->compression_result_buffer_size = 0;
-    state->reset(state);
-    debug(D_STREAM, "%s: Initialize streaming compression!", STREAM_COMPRESSION_MSG);
-    return state;
-}
-
-/*
- * LZ4 streaming API decompressor specific data
- */
-struct decompressor_stream {
-    LZ4_streamDecode_t *lz4_stream;
-    char *buffer;
-    size_t size;
-    size_t write_at;
-    size_t read_at;
-};
-
-/*
- * Reset decompressor state for a new stream
- */
-static void lz4_decompressor_reset(struct decompressor_state *state)
-{
-    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;
-    }
-}
-
-/*
- * Destroy decompressor state and all related data
- */
-static void lz4_decompressor_destroy(struct decompressor_state **state)
-{
-    if (state && *state) {
-        struct decompressor_state *s = *state;
-        if (s->stream) {
-            debug(D_STREAM, "%s: Destroying decompressor.", STREAM_COMPRESSION_MSG);
-            if (s->stream->lz4_stream)
-                LZ4_freeStreamDecode(s->stream->lz4_stream);
-            freez(s->stream->buffer);
-            freez(s->stream);
-        }
-        freez(s);
-        *state = NULL;
-    }
-}
-
-static size_t decode_compress_header(const char *data, size_t data_size) {
-    if (unlikely(!data || !data_size))
-        return 0;
-
-    if (unlikely(data_size != SIGNATURE_SIZE))
-        return 0;
-
-    uint32_t sign = *(uint32_t *)data;
-    if (unlikely((sign & SIGNATURE_MASK) != SIGNATURE))
-        return 0;
-
-    size_t length = ((sign >> 8) & 0x7f) | ((sign >> 9) & (0x7f << 7));
-    return length;
-}
-
-/*
- * 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 __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);
-
-    return decode_compress_header(header, header_size);
+    return compressed_data_size + RRDPUSH_COMPRESSION_SIGNATURE_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, const char *compressed_data, size_t compressed_size) {
+size_t rrdpush_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.read_at != state->stream.write_at))
+        fatal("RRDPUSH_DECOMPRESS: asked to decompress new data, while there are unread data in the decompression buffer!");
 
-    if (unlikely(state->stream->write_at >= state->stream->size / 2)) {
-        state->stream->write_at = 0;
-        state->stream->read_at = 0;
+    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->stream->lz4_stream
+            state->stream.lz4_stream
             , compressed_data
-            , state->stream->buffer + state->stream->write_at
+            , state->stream.buffer + state->stream.write_at
             , (int)compressed_size
-            , (int)(state->stream->size - state->stream->write_at)
+            , (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);
+        error("RRDPUSH DECOMPRESS: decompressor returned negative decompressed bytes: %ld", decompressed_size);
         return 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
+    if(unlikely(decompressed_size + state->stream.write_at > state->stream.size))
+        fatal("RRDPUSH DECOMPRESS: decompressor overflown the stream_buffer. size: %zu, pos: %zu, added: %ld, "
+              "exceeding the buffer by %zu"
+              , state->stream.size
+              , state->stream.write_at
               , decompressed_size
-              , (size_t)(state->stream->write_at + decompressed_size - state->stream->size)
+              , (size_t)(state->stream.write_at + decompressed_size - state->stream.size)
               );
 
-    state->stream->write_at += decompressed_size;
+    state->stream.write_at += decompressed_size;
 
     // statistics
-    state->total_compressed += compressed_size + SIGNATURE_SIZE;
+    state->total_compressed += compressed_size + RRDPUSH_COMPRESSION_SIGNATURE_SIZE;
     state->total_uncompressed += decompressed_size;
     state->packet_count++;
 
     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) {
-    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;
-}
+void rrdpush_decompressor_reset(struct decompressor_state *state) {
+    if(!state->initialized) {
+        state->initialized = true;
+        state->stream.lz4_stream = LZ4_createStreamDecode();
+        state->stream.size = LZ4_decoderRingBufferSize(COMPRESSION_MAX_MSG_SIZE) * 2;
+        state->stream.buffer = mallocz(state->stream.size);
+    }
 
-/*
- * 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 *dst, size_t size) {
-    if (unlikely(!state || !size || !dst))
-        return 0;
+    LZ4_setStreamDecode(state->stream.lz4_stream, NULL, 0);
 
-    size_t remaining = lz4_decompressor_decompressed_bytes_in_buffer(state);
-    if(unlikely(!remaining))
-        return 0;
+    state->signature_size = RRDPUSH_COMPRESSION_SIGNATURE_SIZE;
+    state->stream.write_at = 0;
+    state->stream.read_at = 0;
+}
 
-    size_t bytes_to_return = size;
-    if(bytes_to_return > remaining)
-        bytes_to_return = remaining;
+void rrdpush_decompressor_destroy(struct decompressor_state *state) {
+    if(unlikely(!state->initialized))
+        return;
 
-    memcpy(dst, state->stream->buffer + state->stream->read_at, bytes_to_return);
-    state->stream->read_at += bytes_to_return;
+    if (state->stream.lz4_stream) {
+        LZ4_freeStreamDecode(state->stream.lz4_stream);
+        state->stream.lz4_stream = NULL;
+    }
 
-    if(unlikely(state->stream->read_at > state->stream->write_at))
-        fatal("%s: invalid read/write stream positions"
-        , STREAM_COMPRESSION_MSG);
+    freez(state->stream.buffer);
+    state->stream.buffer = NULL;
 
-    return bytes_to_return;
+    state->initialized = false;
 }
 
-/*
- * Create and initialize decompressor state
- * Return the pointer to decompressor_state structure created
- */
-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->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->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;
-}
 #endif
diff --git a/streaming/receiver.c b/streaming/receiver.c
index 237345cc9c..ce48019683 100644
--- a/streaming/receiver.c
+++ b/streaming/receiver.c
@@ -2,17 +2,6 @@
 
 #include "rrdpush.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
-#endif
-
 extern struct config stream_config;
 
 void receiver_state_free(struct receiver_state *rpt) {
@@ -40,12 +29,11 @@ void receiver_state_free(struct receiver_state *rpt) {
     }
 
 #ifdef ENABLE_COMPRESSION
-    if (rpt->decompressor)
-        rpt->decompressor->destroy(&rpt->decompressor);
+    rrdpush_decompressor_destroy(&rpt->decompressor);
 #endif
 
     if(rpt->system_info)
-        rrdhost_system_info_free(rpt->system_info);
+         rrdhost_system_info_free(rpt->system_info);
 
     __atomic_sub_fetch(&netdata_buffers_statistics.rrdhost_receivers, sizeof(*rpt), __ATOMIC_RELAXED);
 
@@ -54,51 +42,18 @@ void receiver_state_free(struct receiver_state *rpt) {
 
 #include "collectors/plugins.d/pluginsd_parser.h"
 
-PARSER_RC streaming_claimed_id(char **words, size_t num_words, void *user)
-{
-    const char *host_uuid_str = get_word(words, num_words, 1);
-    const char *claim_id_str = get_word(words, num_words, 2);
-
-    if (!host_uuid_str || !claim_id_str) {
-        error("Command CLAIMED_ID came malformed, uuid = '%s', claim_id = '%s'",
-              host_uuid_str ? host_uuid_str : "[unset]",
-              claim_id_str ? claim_id_str : "[unset]");
-        return PARSER_RC_ERROR;
-    }
-
-    uuid_t uuid;
-    RRDHOST *host = ((PARSER_USER_OBJECT *)user)->host;
-
-    // We don't need the parsed UUID
-    // just do it to check the format
-    if(uuid_parse(host_uuid_str, uuid)) {
-        error("1st parameter (host GUID) to CLAIMED_ID command is not valid GUID. Received: \"%s\".", host_uuid_str);
-        return PARSER_RC_ERROR;
-    }
-    if(uuid_parse(claim_id_str, uuid) && strcmp(claim_id_str, "NULL")) {
-        error("2nd parameter (Claim ID) to CLAIMED_ID command is not valid GUID. Received: \"%s\".", claim_id_str);
-        return PARSER_RC_ERROR;
-    }
-
-    if(strcmp(host_uuid_str, host->machine_guid)) {
-        error("Claim ID is for host \"%s\" but it came over connection for \"%s\"", host_uuid_str, host->machine_guid);
-        return PARSER_RC_OK; //the message is OK problem must be somewhere else
-    }
-
-    rrdhost_aclk_state_lock(host);
-    if (host->aclk_state.claimed_id)
-        freez(host->aclk_state.claimed_id);
-    host->aclk_state.claimed_id = strcmp(claim_id_str, "NULL") ? strdupz(claim_id_str) : NULL;
-    rrdhost_aclk_state_unlock(host);
-
-    rrdhost_flag_set(host, RRDHOST_FLAG_METADATA_CLAIMID |RRDHOST_FLAG_METADATA_UPDATE);
+// 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)
 
-    rrdpush_send_claimed_id(host);
+// this has to be the same at parser.h
+#define WORKER_RECEIVER_JOB_REPLICATION_COMPLETION (WORKER_PARSER_FIRST_JOB - 3)
 
-    return PARSER_RC_OK;
-}
+#if WORKER_PARSER_FIRST_JOB < 1
+#error The define WORKER_PARSER_FIRST_JOB needs to be at least 1
+#endif
 
-static int read_stream(struct receiver_state *r, char* buffer, size_t size) {
+static inline 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;
@@ -137,7 +92,7 @@ static int read_stream(struct receiver_state *r, char* buffer, size_t size) {
     return (int)bytes_read;
 }
 
-static bool receiver_read_uncompressed(struct receiver_state *r) {
+static inline 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__ );
@@ -157,19 +112,17 @@ static bool receiver_read_uncompressed(struct receiver_state *r) {
 }
 
 #ifdef ENABLE_COMPRESSION
-static bool receiver_read_compressed(struct receiver_state *r) {
+static inline 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
+    internal_fatal(r->read_buffer[r->read_len] != '\0',
+                   "%s: read_buffer does not start with zero #2", __FUNCTION__ );
 
     // first use any available uncompressed data
-    if (r->decompressor->decompressed_bytes_in_buffer(r->decompressor)) {
+    if (likely(rrdpush_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) {
+        if (likely(available)) {
+            size_t len = rrdpush_decompressor_get(&r->decompressor, r->read_buffer + r->read_len, available);
+            if (unlikely(!len)) {
                 internal_error(true, "decompressor returned zero length #1");
                 return false;
             }
@@ -178,7 +131,7 @@ static bool receiver_read_compressed(struct receiver_state *r) {
             r->read_buffer[r->read_len] = '\0';
         }
         else
-            internal_error(true, "The line to read is too big! Already have %d bytes in read_buffer.", r->read_len);
+            internal_fatal(true, "The line to read is too big! Already have %d bytes in read_buffer.", r->read_len);
 
         return true;
     }
@@ -186,8 +139,9 @@ static bool receiver_read_compressed(struct receiver_state *r) {
     // 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);
+    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;
     }
 
@@ -195,19 +149,19 @@ static bool receiver_read_compressed(struct receiver_state *r) {
     // 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);
+        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;
 
         bytes_read += ret;
-    } while(unlikely(bytes_read < (int)r->decompressor->signature_size));
+    } 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);
+    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);
+    size_t compressed_message_size = rrdpush_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;
@@ -244,8 +198,8 @@ static bool receiver_read_compressed(struct receiver_state *r) {
     worker_set_metric(WORKER_RECEIVER_JOB_BYTES_READ, (NETDATA_DOUBLE)compressed_bytes_read);
 
     // decompress the compressed block
-    size_t bytes_to_parse = r->decompressor->decompress(r->decompressor, compressed, compressed_bytes_read);
-    if (!bytes_to_parse) {
+    size_t bytes_to_parse = rrdpush_decompress(&r->decompressor, compressed, compressed_bytes_read);
+    if (unlikely(!bytes_to_parse)) {
         internal_error(true, "no bytes to parse.");
         return false;
     }
@@ -253,8 +207,8 @@ static bool receiver_read_compressed(struct receiver_state *r) {
     worker_set_metric(WORKER_RECEIVER_JOB_BYTES_UNCOMPRESSED, (NETDATA_DOUBLE)bytes_to_parse);
 
     // 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) {
+    size_t len = (int) rrdpush_decompressor_get(&r->decompressor, r->read_buffer + r->read_len, sizeof(r->read_buffer) - r->read_len - 1);
+    if (unlikely(!len)) {
         internal_error(true, "decompressor returned zero length #2");
         return false;
     }
@@ -264,7 +218,7 @@ static bool receiver_read_compressed(struct receiver_state *r) {
     return true;
 }
 #else // !ENABLE_COMPRESSION
-static bool receiver_read_compressed(struct receiver_state *r) {
+static inline bool receiver_read_compressed(struct receiver_state *r) {
     return receiver_read_uncompressed(r);
 }
 #endif // ENABLE_COMPRESSION
@@ -272,7 +226,7 @@ static bool receiver_read_compressed(struct receiver_state *r) {
 /* 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.
  */
-static char *receiver_next_line(struct receiver_state *r, char *buffer, size_t buffer_length, size_t *pos) {
+static inline char *receiver_next_line(struct receiver_state *r, char *buffer, size_t buffer_length, size_t *pos) {
     size_t start = *pos;
 
     char *ss = &r->read_buffer[start];
@@ -323,20 +277,50 @@ static char *receiver_next_line(struct receiver_state *r, char *buffer, size_t b
 
 bool plugin_is_enabled(struct plugind *cd);
 
+static void receiver_set_exit_reason(struct receiver_state *rpt, STREAM_HANDSHAKE reason, bool force) {
+    if(force || !rpt->exit.reason)
+        rpt->exit.reason = reason;
+}
+
+static inline bool receiver_should_continue(struct receiver_state *rpt) {
+    static __thread size_t counter = 0;
+
+    if(unlikely(rpt->exit.shutdown)) {
+        receiver_set_exit_reason(rpt, STREAM_HANDSHAKE_DISCONNECT_SHUTDOWN, false);
+        return false;
+    }
+
+    // check every 1000 lines read
+    if((counter++ % 1000) != 0) return true;
+
+    if(unlikely(!service_running(SERVICE_STREAMING))) {
+        receiver_set_exit_reason(rpt, STREAM_HANDSHAKE_DISCONNECT_NETDATA_EXIT, false);
+        return false;
+    }
+
+    netdata_thread_testcancel();
+
+    rpt->last_msg_t = now_monotonic_sec();
+
+    return true;
+}
+
 static size_t streaming_parser(struct receiver_state *rpt, struct plugind *cd, int fd, void *ssl) {
     size_t result;
 
-    PARSER_USER_OBJECT user = {
-        .enabled = plugin_is_enabled(cd),
-        .host = rpt->host,
-        .opaque = rpt,
-        .cd = cd,
-        .trust_durations = 1,
-        .capabilities = rpt->capabilities,
-    };
-
-    PARSER *parser = parser_init(&user, NULL, NULL, fd,
-                                 PARSER_INPUT_SPLIT, ssl);
+    PARSER *parser = NULL;
+    {
+        PARSER_USER_OBJECT user = {
+                .enabled = plugin_is_enabled(cd),
+                .host = rpt->host,
+                .opaque = rpt,
+                .cd = cd,
+                .trust_durations = 1,
+                .capabilities = rpt->capabilities,
+        };
+
+        parser = parser_init(&user, NULL, NULL, fd, PARSER_INPUT_SPLIT, ssl);
+    }
 
     pluginsd_keywords_init(parser, PARSER_INIT_STREAMING);
 
@@ -346,20 +330,15 @@ static size_t streaming_parser(struct receiver_state *rpt, struct plugind *cd, i
     // so, parser needs to be allocated before pushing it
     netdata_thread_cleanup_push(pluginsd_process_thread_cleanup, parser);
 
-    parser_add_keyword(parser, "CLAIMED_ID", streaming_claimed_id);
-
-    user.parser = parser;
-
     bool compressed_connection = false;
+
 #ifdef ENABLE_COMPRESSION
     if(stream_has_capability(rpt, STREAM_CAP_COMPRESSION)) {
         compressed_connection = true;
-
-        if (!rpt->decompressor)
-            rpt->decompressor = create_decompressor();
-        else
-            rpt->decompressor->reset(rpt->decompressor);
+        rrdpush_decompressor_reset(&rpt->decompressor);
     }
+    else
+        rrdpush_decompressor_destroy(&rpt->decompressor);
 #endif
 
     rpt->read_buffer[0] = '\0';
@@ -367,51 +346,27 @@ static size_t streaming_parser(struct receiver_state *rpt, struct plugind *cd, i
 
     size_t read_buffer_start = 0;
     char buffer[PLUGINSD_LINE_MAX + 2] = "";
-    while(service_running(SERVICE_STREAMING)) {
-        netdata_thread_testcancel();
+    while(receiver_should_continue(rpt)) {
 
         if(!receiver_next_line(rpt, buffer, PLUGINSD_LINE_MAX + 2, &read_buffer_start)) {
-            bool have_new_data;
-            if(likely(compressed_connection))
-                have_new_data = receiver_read_compressed(rpt);
-            else
-                have_new_data = receiver_read_uncompressed(rpt);
+            bool have_new_data = compressed_connection ? receiver_read_compressed(rpt) : receiver_read_uncompressed(rpt);
 
             if(unlikely(!have_new_data)) {
-                if(!rpt->exit.reason)
-                    rpt->exit.reason = "SOCKET READ ERROR";
-
+                receiver_set_exit_reason(rpt, STREAM_HANDSHAKE_DISCONNECT_SOCKET_READ_ERROR, false);
                 break;
             }
 
-            rpt->last_msg_t = now_realtime_sec();
             continue;
         }
 
-        if(unlikely(!service_running(SERVICE_STREAMING))) {
-            if(!rpt->exit.reason)
-                rpt->exit.reason = "NETDATA EXIT";
-            goto done;
-        }
-        if(unlikely(rpt->exit.shutdown)) {
-            if(!rpt->exit.reason)
-                rpt->exit.reason = "SHUTDOWN REQUESTED";
-
-            goto done;
-        }
-
         if (unlikely(parser_action(parser,  buffer))) {
             internal_error(true, "parser_action() failed on keyword '%s'.", buffer);
-
-            if(!rpt->exit.reason)
-                rpt->exit.reason = "PARSER FAILED";
-
+            receiver_set_exit_reason(rpt, STREAM_HANDSHAKE_DISCONNECT_PARSER_FAILED, false);
             break;
         }
     }
 
-done:
-    result = user.data_collections_count;
+    result = parser ? parser->user.data_collections_count : 0;
 
     // free parser with the pop function
     netdata_thread_cleanup_pop(1);
@@ -501,7 +456,7 @@ static void rrdhost_clear_receiver(struct receiver_state *rpt) {
             if (rpt->config.health_enabled == CONFIG_BOOLEAN_AUTO)
                 host->health.health_enabled = 0;
 
-            rrdpush_sender_thread_stop(host, "RECEIVER LEFT", false);
+            rrdpush_sender_thread_stop(host, STREAM_HANDSHAKE_DISCONNECT_RECEIVER_LEFT, false);
 
             signal_rrdcontext = true;
             rrdpush_receiver_replication_reset(host);
@@ -520,7 +475,7 @@ static void rrdhost_clear_receiver(struct receiver_state *rpt) {
     }
 }
 
-bool stop_streaming_receiver(RRDHOST *host, const char *reason) {
+bool stop_streaming_receiver(RRDHOST *host, STREAM_HANDSHAKE reason) {
     bool ret = false;
 
     netdata_mutex_lock(&host->receiver_lock);
@@ -528,7 +483,7 @@ bool stop_streaming_receiver(RRDHOST *host, const char *reason) {
     if(host->receiver) {
         if(!host->receiver->exit.shutdown) {
             host->receiver->exit.shutdown = true;
-            host->receiver->exit.reason = reason;
+            receiver_set_exit_reason(host->receiver, reason, true);
             shutdown(host->receiver->fd, SHUT_RDWR);
         }
 
@@ -586,9 +541,9 @@ void rrdpush_receive_log_status(struct receiver_state *rpt, const char *msg, con
           , rpt->client_ip, rpt->client_port
           , msg
           , status
-          , rpt->exit.reason?" (":""
-          , rpt->exit.reason?rpt->exit.reason:""
-          , rpt->exit.reason?")":""
+          , rpt->exit.reason != STREAM_HANDSHAKE_NEVER?" (":""
+          , stream_handshake_error_to_string(rpt->exit.reason)
+          , rpt->exit.reason != STREAM_HANDSHAKE_NEVER?")":""
     );
 
 }
@@ -661,7 +616,6 @@ static void rrdpush_receive(struct receiver_state *rpt)
     rpt->config.rrdpush_compression = default_compression_enabled;
     rpt->config.rrdpush_compression = appconfig_get_boolean(&stream_config, rpt->key, "enable compression", rpt->config.rrdpush_compression);
     rpt->config.rrdpush_compression = appconfig_get_boolean(&stream_config, rpt->machine_guid, "enable compression", rpt->config.rrdpush_compression);
-    rpt->rrdpush_compression = (rpt->config.rrdpush_compression && default_compression_enabled);
 #endif  //ENABLE_COMPRESSION
 
     (void)appconfig_set_default(&stream_config, rpt->machine_guid, "host tags", (rpt->tags)?rpt->tags:"");
@@ -758,7 +712,7 @@ static void rrdpush_receive(struct receiver_state *rpt)
 
 #ifdef ENABLE_COMPRESSION
     if (stream_has_capability(rpt, STREAM_CAP_COMPRESSION)) {
-        if (!rpt->rrdpush_compression)
+        if (!rpt->config.rrdpush_compression)
             rpt->capabilities &= ~STREAM_CAP_COMPRESSION;
     }
 #endif
@@ -839,8 +793,7 @@ static void rrdpush_receive(struct receiver_state *rpt)
 #endif
                                     );
 
-    if(!rpt->exit.reason)
-        rpt->exit.reason = "PARSER EXIT";
+    receiver_set_exit_reason(rpt, STREAM_HANDSHAKE_DISCONNECT_PARSER_EXIT, false);
 
     {
         char msg[100 + 1];
diff --git a/streaming/rrdpush.c b/streaming/rrdpush.c
index f231fcab3a..b2da305787 100644
--- a/