From ab1bb5b7f1c3c3a7b240ab7fc8661459ecd7decb Mon Sep 17 00:00:00 2001 From: Daniel Baumann Date: Thu, 20 Jul 2023 06:49:55 +0200 Subject: Adding upstream version 1.41.0. Signed-off-by: Daniel Baumann --- streaming/compression.c | 305 +++++++++++++----------------------------------- 1 file changed, 82 insertions(+), 223 deletions(-) (limited to 'streaming/compression.c') diff --git a/streaming/compression.c b/streaming/compression.c index 8f2517a8e..6d4a128b0 100644 --- a/streaming/compression.c +++ b/streaming/compression.c @@ -1,59 +1,44 @@ #include "rrdpush.h" -#ifdef ENABLE_COMPRESSION +#ifdef ENABLE_RRDPUSH_COMPRESSION #include "lz4.h" #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); + netdata_log_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,238 +70,112 @@ 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) { - error("Data compression error: %ld", compressed_data_size); + netdata_log_error("Data compression error: %ld", compressed_data_size); return 0; } // 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); + netdata_log_debug(D_STREAM, "%s: Compressed data header: %ld", STREAM_COMPRESSION_MSG, compressed_data_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); + netdata_log_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 -- cgit v1.2.3