diff options
author | Costa Tsaousis <costa@netdata.cloud> | 2023-06-26 14:00:59 +0300 |
---|---|---|
committer | GitHub <noreply@github.com> | 2023-06-26 14:00:59 +0300 |
commit | 0d61c11b5f4772a4762ede1d8204290b94bb08e7 (patch) | |
tree | 49c97d67e0d2a4846a4b379345f53ef8d93a6aec /streaming | |
parent | f90d56f18d29c2835bc278f6a22e840230b9ca86 (diff) |
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
Diffstat (limited to 'streaming')
-rw-r--r-- | streaming/compression.c | 299 | ||||
-rw-r--r-- | streaming/receiver.c | 231 | ||||
-rw-r--r-- | streaming/rrdpush.c | 24 | ||||
-rw-r--r-- | streaming/rrdpush.h | 131 | ||||
-rw-r--r-- | streaming/sender.c | 66 |
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]; |