summaryrefslogtreecommitdiffstats
path: root/streaming/compression.c
diff options
context:
space:
mode:
authorDaniel Baumann <daniel.baumann@progress-linux.org>2022-11-30 18:47:05 +0000
committerDaniel Baumann <daniel.baumann@progress-linux.org>2022-11-30 18:47:05 +0000
commit97e01009d69b8fbebfebf68f51e3d126d0ed43fc (patch)
tree02e8b836c3a9d89806f3e67d4a5fe9f52dbb0061 /streaming/compression.c
parentReleasing debian version 1.36.1-1. (diff)
downloadnetdata-97e01009d69b8fbebfebf68f51e3d126d0ed43fc.tar.xz
netdata-97e01009d69b8fbebfebf68f51e3d126d0ed43fc.zip
Merging upstream version 1.37.0.
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to 'streaming/compression.c')
-rw-r--r--streaming/compression.c301
1 files changed, 137 insertions, 164 deletions
diff --git a/streaming/compression.c b/streaming/compression.c
index d6178d6c3..7ba9dbf19 100644
--- a/streaming/compression.c
+++ b/streaming/compression.c
@@ -5,9 +5,7 @@
#define STREAM_COMPRESSION_MSG "STREAM_COMPRESSION"
-#define LZ4_MAX_MSG_SIZE 0x4000
-#define LZ4_STREAM_BUFFER_SIZE (0x10000 + LZ4_MAX_MSG_SIZE)
-
+// 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
@@ -18,8 +16,9 @@
*/
struct compressor_data {
LZ4_stream_t *stream;
- char *stream_buffer;
- size_t stream_buffer_pos;
+ char *input_ring_buffer;
+ size_t input_ring_buffer_size;
+ size_t input_ring_buffer_pos;
};
@@ -31,9 +30,9 @@ 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->stream_buffer_pos = 0;
+ state->data->input_ring_buffer_pos = 0;
}
}
@@ -47,10 +46,10 @@ static void lz4_compressor_destroy(struct compressor_state **state)
if (s->data) {
if (s->data->stream)
LZ4_freeStream(s->data->stream);
- freez(s->data->stream_buffer);
+ freez(s->data->input_ring_buffer);
freez(s->data);
}
- freez(s->buffer);
+ freez(s->compression_result_buffer);
freez(s);
*state = NULL;
debug(D_STREAM, "%s: Compressor Destroyed.", STREAM_COMPRESSION_MSG);
@@ -65,37 +64,53 @@ static void lz4_compressor_destroy(struct compressor_state **state)
*/
static size_t lz4_compressor_compress(struct compressor_state *state, const char *data, size_t size, char **out)
{
- if (!state || !size || !out)
+ if(unlikely(!state || !size || !out))
return 0;
- if (size > LZ4_MAX_MSG_SIZE) {
- error("%s: Compression Failed - Message size %lu above compression buffer limit: %d", STREAM_COMPRESSION_MSG, size, LZ4_MAX_MSG_SIZE);
+
+ 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);
return 0;
}
+
size_t max_dst_size = LZ4_COMPRESSBOUND(size);
size_t data_size = max_dst_size + SIGNATURE_SIZE;
- if (!state->buffer) {
- state->buffer = mallocz(data_size);
- state->buffer_size = data_size;
- } else if (state->buffer_size < data_size) {
- state->buffer = reallocz(state->buffer, data_size);
- state->buffer_size = data_size;
+ if (!state->compression_result_buffer) {
+ state->compression_result_buffer = mallocz(data_size);
+ state->compression_result_buffer_size = data_size;
+ }
+ else if(unlikely(state->compression_result_buffer_size < data_size)) {
+ state->compression_result_buffer = reallocz(state->compression_result_buffer, data_size);
+ state->compression_result_buffer_size = data_size;
}
- memcpy(state->data->stream_buffer + state->data->stream_buffer_pos, data, size);
- long int compressed_data_size = LZ4_compress_fast_continue(state->data->stream,
- state->data->stream_buffer + state->data->stream_buffer_pos,
- state->buffer + SIGNATURE_SIZE, size, max_dst_size, 1);
+ // 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);
+
+ // 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,
+ 1);
+
if (compressed_data_size < 0) {
error("Data compression error: %ld", compressed_data_size);
return 0;
}
- state->data->stream_buffer_pos += size;
- if (state->data->stream_buffer_pos >= LZ4_STREAM_BUFFER_SIZE - LZ4_MAX_MSG_SIZE)
- state->data->stream_buffer_pos = 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;
+
+ // update the signature header
uint32_t len = ((compressed_data_size & 0x7f) | 0x80 | (((compressed_data_size & (0x7f << 7)) << 1) | 0x8000)) << 8;
- *(uint32_t *)state->buffer = len | SIGNATURE;
- *out = state->buffer;
+ *(uint32_t *)state->compression_result_buffer = len | 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;
}
@@ -114,8 +129,9 @@ struct compressor_state *create_compressor()
state->data = callocz(1, sizeof(struct compressor_data));
state->data->stream = LZ4_createStream();
- state->data->stream_buffer = callocz(1, LZ4_DECODER_RING_BUFFER_SIZE(LZ4_MAX_MSG_SIZE));
- state->buffer_size = LZ4_STREAM_BUFFER_SIZE;
+ 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;
@@ -124,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;
};
/*
@@ -136,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;
}
}
@@ -152,173 +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;
+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);
- 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;
+ return decode_compress_header(header, header_size);
}
/*
- * Add a chunk of compressed data to the internal buffer
- * Return the current size of compressed data or 0 for error
+ * Decompress the compressed data in the internal buffer
+ * Return the size of uncompressed 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)
+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 (!state->buffer)
- fatal("STREAM: No decompressor buffer allocated");
- if (state->buffer_pos + size > state->buffer_len) {
- error("STREAM: Decompressor buffer overflow %lu + %lu > %lu",
- state->buffer_pos, size, state->buffer_len);
- size = state->buffer_len - state->buffer_pos;
+ 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;
}
- memcpy(state->buffer + state->buffer_pos, data, size);
- state->buffer_pos += size;
- return state->buffer_pos;
-}
-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;
-}
+ 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)
+ );
-/*
- * 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);
- return 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);
+ 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 - LZ4_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;
-
- 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, saving, avg_saving, 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;
}
/*
@@ -328,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(LZ4_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;