diff options
author | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-11-25 14:45:37 +0000 |
---|---|---|
committer | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-11-25 14:48:03 +0000 |
commit | e55403ed71282d7bfd8b56df219de3c28a8af064 (patch) | |
tree | 524889e5becb81643bf8741e3082955dca076f09 /src/streaming/compression.c | |
parent | Releasing debian version 1.47.5-1. (diff) | |
download | netdata-e55403ed71282d7bfd8b56df219de3c28a8af064.tar.xz netdata-e55403ed71282d7bfd8b56df219de3c28a8af064.zip |
Merging upstream version 2.0.3+dfsg:
- does not include dygraphs anymore (Closes: #923993)
- does not include pako anymore (Closes: #1042533)
- does not include dashboard binaries anymore (Closes: #1045145)
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to 'src/streaming/compression.c')
-rw-r--r-- | src/streaming/compression.c | 707 |
1 files changed, 0 insertions, 707 deletions
diff --git a/src/streaming/compression.c b/src/streaming/compression.c deleted file mode 100644 index a94c8a0a6..000000000 --- a/src/streaming/compression.c +++ /dev/null @@ -1,707 +0,0 @@ -// SPDX-License-Identifier: GPL-3.0-or-later - -#include "compression.h" - -#include "compression_gzip.h" - -#ifdef ENABLE_LZ4 -#include "compression_lz4.h" -#endif - -#ifdef ENABLE_ZSTD -#include "compression_zstd.h" -#endif - -#ifdef ENABLE_BROTLI -#include "compression_brotli.h" -#endif - -int rrdpush_compression_levels[COMPRESSION_ALGORITHM_MAX] = { - [COMPRESSION_ALGORITHM_NONE] = 0, - [COMPRESSION_ALGORITHM_ZSTD] = 3, // 1 (faster) - 22 (smaller) - [COMPRESSION_ALGORITHM_LZ4] = 1, // 1 (smaller) - 9 (faster) - [COMPRESSION_ALGORITHM_BROTLI] = 3, // 0 (faster) - 11 (smaller) - [COMPRESSION_ALGORITHM_GZIP] = 1, // 1 (faster) - 9 (smaller) -}; - -void rrdpush_parse_compression_order(struct receiver_state *rpt, const char *order) { - // empty all slots - for(size_t i = 0; i < COMPRESSION_ALGORITHM_MAX ;i++) - rpt->config.compression_priorities[i] = STREAM_CAP_NONE; - - char *s = strdupz(order); - - char *words[COMPRESSION_ALGORITHM_MAX + 100] = { NULL }; - size_t num_words = quoted_strings_splitter_pluginsd(s, words, COMPRESSION_ALGORITHM_MAX + 100); - size_t slot = 0; - STREAM_CAPABILITIES added = STREAM_CAP_NONE; - for(size_t i = 0; i < num_words && slot < COMPRESSION_ALGORITHM_MAX ;i++) { - if((STREAM_CAP_ZSTD_AVAILABLE) && strcasecmp(words[i], "zstd") == 0 && !(added & STREAM_CAP_ZSTD)) { - rpt->config.compression_priorities[slot++] = STREAM_CAP_ZSTD; - added |= STREAM_CAP_ZSTD; - } - else if((STREAM_CAP_LZ4_AVAILABLE) && strcasecmp(words[i], "lz4") == 0 && !(added & STREAM_CAP_LZ4)) { - rpt->config.compression_priorities[slot++] = STREAM_CAP_LZ4; - added |= STREAM_CAP_LZ4; - } - else if((STREAM_CAP_BROTLI_AVAILABLE) && strcasecmp(words[i], "brotli") == 0 && !(added & STREAM_CAP_BROTLI)) { - rpt->config.compression_priorities[slot++] = STREAM_CAP_BROTLI; - added |= STREAM_CAP_BROTLI; - } - else if(strcasecmp(words[i], "gzip") == 0 && !(added & STREAM_CAP_GZIP)) { - rpt->config.compression_priorities[slot++] = STREAM_CAP_GZIP; - added |= STREAM_CAP_GZIP; - } - } - - freez(s); - - // make sure all participate - if((STREAM_CAP_ZSTD_AVAILABLE) && slot < COMPRESSION_ALGORITHM_MAX && !(added & STREAM_CAP_ZSTD)) - rpt->config.compression_priorities[slot++] = STREAM_CAP_ZSTD; - if((STREAM_CAP_LZ4_AVAILABLE) && slot < COMPRESSION_ALGORITHM_MAX && !(added & STREAM_CAP_LZ4)) - rpt->config.compression_priorities[slot++] = STREAM_CAP_LZ4; - if((STREAM_CAP_BROTLI_AVAILABLE) && slot < COMPRESSION_ALGORITHM_MAX && !(added & STREAM_CAP_BROTLI)) - rpt->config.compression_priorities[slot++] = STREAM_CAP_BROTLI; - if(slot < COMPRESSION_ALGORITHM_MAX && !(added & STREAM_CAP_GZIP)) - rpt->config.compression_priorities[slot++] = STREAM_CAP_GZIP; -} - -void rrdpush_select_receiver_compression_algorithm(struct receiver_state *rpt) { - if (!rpt->config.rrdpush_compression) - rpt->capabilities &= ~STREAM_CAP_COMPRESSIONS_AVAILABLE; - - // select the right compression before sending our capabilities to the child - if(stream_has_more_than_one_capability_of(rpt->capabilities, STREAM_CAP_COMPRESSIONS_AVAILABLE)) { - STREAM_CAPABILITIES compressions = rpt->capabilities & STREAM_CAP_COMPRESSIONS_AVAILABLE; - for(int i = 0; i < COMPRESSION_ALGORITHM_MAX; i++) { - STREAM_CAPABILITIES c = rpt->config.compression_priorities[i]; - - if(!(c & STREAM_CAP_COMPRESSIONS_AVAILABLE)) - continue; - - if(compressions & c) { - STREAM_CAPABILITIES exclude = compressions; - exclude &= ~c; - - rpt->capabilities &= ~exclude; - break; - } - } - } -} - -bool rrdpush_compression_initialize(struct sender_state *s) { - rrdpush_compressor_destroy(&s->compressor); - - // IMPORTANT - // KEEP THE SAME ORDER IN DECOMPRESSION - - if(stream_has_capability(s, STREAM_CAP_ZSTD)) - s->compressor.algorithm = COMPRESSION_ALGORITHM_ZSTD; - else if(stream_has_capability(s, STREAM_CAP_LZ4)) - s->compressor.algorithm = COMPRESSION_ALGORITHM_LZ4; - else if(stream_has_capability(s, STREAM_CAP_BROTLI)) - s->compressor.algorithm = COMPRESSION_ALGORITHM_BROTLI; - else if(stream_has_capability(s, STREAM_CAP_GZIP)) - s->compressor.algorithm = COMPRESSION_ALGORITHM_GZIP; - else - s->compressor.algorithm = COMPRESSION_ALGORITHM_NONE; - - if(s->compressor.algorithm != COMPRESSION_ALGORITHM_NONE) { - s->compressor.level = rrdpush_compression_levels[s->compressor.algorithm]; - rrdpush_compressor_init(&s->compressor); - return true; - } - - return false; -} - -bool rrdpush_decompression_initialize(struct receiver_state *rpt) { - rrdpush_decompressor_destroy(&rpt->decompressor); - - // IMPORTANT - // KEEP THE SAME ORDER IN COMPRESSION - - if(stream_has_capability(rpt, STREAM_CAP_ZSTD)) - rpt->decompressor.algorithm = COMPRESSION_ALGORITHM_ZSTD; - else if(stream_has_capability(rpt, STREAM_CAP_LZ4)) - rpt->decompressor.algorithm = COMPRESSION_ALGORITHM_LZ4; - else if(stream_has_capability(rpt, STREAM_CAP_BROTLI)) - rpt->decompressor.algorithm = COMPRESSION_ALGORITHM_BROTLI; - else if(stream_has_capability(rpt, STREAM_CAP_GZIP)) - rpt->decompressor.algorithm = COMPRESSION_ALGORITHM_GZIP; - else - rpt->decompressor.algorithm = COMPRESSION_ALGORITHM_NONE; - - if(rpt->decompressor.algorithm != COMPRESSION_ALGORITHM_NONE) { - rrdpush_decompressor_init(&rpt->decompressor); - return true; - } - - return false; -} - -/* -* In case of stream compression buffer overflow -* Inform the user through the error log file and -* deactivate compression by downgrading the stream protocol. -*/ -void rrdpush_compression_deactivate(struct sender_state *s) { - switch(s->compressor.algorithm) { - case COMPRESSION_ALGORITHM_MAX: - case COMPRESSION_ALGORITHM_NONE: - netdata_log_error("STREAM_COMPRESSION: compression error on 'host:%s' without any compression enabled. Ignoring error.", - rrdhost_hostname(s->host)); - break; - - case COMPRESSION_ALGORITHM_GZIP: - netdata_log_error("STREAM_COMPRESSION: GZIP compression error on 'host:%s'. Disabling GZIP for this node.", - rrdhost_hostname(s->host)); - s->disabled_capabilities |= STREAM_CAP_GZIP; - break; - - case COMPRESSION_ALGORITHM_LZ4: - netdata_log_error("STREAM_COMPRESSION: LZ4 compression error on 'host:%s'. Disabling ZSTD for this node.", - rrdhost_hostname(s->host)); - s->disabled_capabilities |= STREAM_CAP_LZ4; - break; - - case COMPRESSION_ALGORITHM_ZSTD: - netdata_log_error("STREAM_COMPRESSION: ZSTD compression error on 'host:%s'. Disabling ZSTD for this node.", - rrdhost_hostname(s->host)); - s->disabled_capabilities |= STREAM_CAP_ZSTD; - break; - - case COMPRESSION_ALGORITHM_BROTLI: - netdata_log_error("STREAM_COMPRESSION: BROTLI compression error on 'host:%s'. Disabling BROTLI for this node.", - rrdhost_hostname(s->host)); - s->disabled_capabilities |= STREAM_CAP_BROTLI; - break; - } -} - -// ---------------------------------------------------------------------------- -// compressor public API - -void rrdpush_compressor_init(struct compressor_state *state) { - switch(state->algorithm) { -#ifdef ENABLE_ZSTD - case COMPRESSION_ALGORITHM_ZSTD: - rrdpush_compressor_init_zstd(state); - break; -#endif - -#ifdef ENABLE_LZ4 - case COMPRESSION_ALGORITHM_LZ4: - rrdpush_compressor_init_lz4(state); - break; -#endif - -#ifdef ENABLE_BROTLI - case COMPRESSION_ALGORITHM_BROTLI: - rrdpush_compressor_init_brotli(state); - break; -#endif - - default: - case COMPRESSION_ALGORITHM_GZIP: - rrdpush_compressor_init_gzip(state); - break; - } - - simple_ring_buffer_reset(&state->input); - simple_ring_buffer_reset(&state->output); -} - -void rrdpush_compressor_destroy(struct compressor_state *state) { - switch(state->algorithm) { -#ifdef ENABLE_ZSTD - case COMPRESSION_ALGORITHM_ZSTD: - rrdpush_compressor_destroy_zstd(state); - break; -#endif - -#ifdef ENABLE_LZ4 - case COMPRESSION_ALGORITHM_LZ4: - rrdpush_compressor_destroy_lz4(state); - break; -#endif - -#ifdef ENABLE_BROTLI - case COMPRESSION_ALGORITHM_BROTLI: - rrdpush_compressor_destroy_brotli(state); - break; -#endif - - default: - case COMPRESSION_ALGORITHM_GZIP: - rrdpush_compressor_destroy_gzip(state); - break; - } - - state->initialized = false; - - simple_ring_buffer_destroy(&state->input); - simple_ring_buffer_destroy(&state->output); -} - -size_t rrdpush_compress(struct compressor_state *state, const char *data, size_t size, const char **out) { - size_t ret = 0; - - switch(state->algorithm) { -#ifdef ENABLE_ZSTD - case COMPRESSION_ALGORITHM_ZSTD: - ret = rrdpush_compress_zstd(state, data, size, out); - break; -#endif - -#ifdef ENABLE_LZ4 - case COMPRESSION_ALGORITHM_LZ4: - ret = rrdpush_compress_lz4(state, data, size, out); - break; -#endif - -#ifdef ENABLE_BROTLI - case COMPRESSION_ALGORITHM_BROTLI: - ret = rrdpush_compress_brotli(state, data, size, out); - break; -#endif - - default: - case COMPRESSION_ALGORITHM_GZIP: - ret = rrdpush_compress_gzip(state, data, size, out); - break; - } - - if(unlikely(ret >= COMPRESSION_MAX_CHUNK)) { - netdata_log_error("RRDPUSH_COMPRESS: compressed data is %zu bytes, which is >= than the max chunk size %d", - ret, COMPRESSION_MAX_CHUNK); - return 0; - } - - return ret; -} - -// ---------------------------------------------------------------------------- -// decompressor public API - -void rrdpush_decompressor_destroy(struct decompressor_state *state) { - if(unlikely(!state->initialized)) - return; - - switch(state->algorithm) { -#ifdef ENABLE_ZSTD - case COMPRESSION_ALGORITHM_ZSTD: - rrdpush_decompressor_destroy_zstd(state); - break; -#endif - -#ifdef ENABLE_LZ4 - case COMPRESSION_ALGORITHM_LZ4: - rrdpush_decompressor_destroy_lz4(state); - break; -#endif - -#ifdef ENABLE_BROTLI - case COMPRESSION_ALGORITHM_BROTLI: - rrdpush_decompressor_destroy_brotli(state); - break; -#endif - - default: - case COMPRESSION_ALGORITHM_GZIP: - rrdpush_decompressor_destroy_gzip(state); - break; - } - - simple_ring_buffer_destroy(&state->output); - - state->initialized = false; -} - -void rrdpush_decompressor_init(struct decompressor_state *state) { - switch(state->algorithm) { -#ifdef ENABLE_ZSTD - case COMPRESSION_ALGORITHM_ZSTD: - rrdpush_decompressor_init_zstd(state); - break; -#endif - -#ifdef ENABLE_LZ4 - case COMPRESSION_ALGORITHM_LZ4: - rrdpush_decompressor_init_lz4(state); - break; -#endif - -#ifdef ENABLE_BROTLI - case COMPRESSION_ALGORITHM_BROTLI: - rrdpush_decompressor_init_brotli(state); - break; -#endif - - default: - case COMPRESSION_ALGORITHM_GZIP: - rrdpush_decompressor_init_gzip(state); - break; - } - - state->signature_size = RRDPUSH_COMPRESSION_SIGNATURE_SIZE; - simple_ring_buffer_reset(&state->output); -} - -size_t rrdpush_decompress(struct decompressor_state *state, const char *compressed_data, size_t compressed_size) { - if (unlikely(state->output.read_pos != state->output.write_pos)) - fatal("RRDPUSH_DECOMPRESS: asked to decompress new data, while there are unread data in the decompression buffer!"); - - size_t ret = 0; - - switch(state->algorithm) { -#ifdef ENABLE_ZSTD - case COMPRESSION_ALGORITHM_ZSTD: - ret = rrdpush_decompress_zstd(state, compressed_data, compressed_size); - break; -#endif - -#ifdef ENABLE_LZ4 - case COMPRESSION_ALGORITHM_LZ4: - ret = rrdpush_decompress_lz4(state, compressed_data, compressed_size); - break; -#endif - -#ifdef ENABLE_BROTLI - case COMPRESSION_ALGORITHM_BROTLI: - ret = rrdpush_decompress_brotli(state, compressed_data, compressed_size); - break; -#endif - - default: - case COMPRESSION_ALGORITHM_GZIP: - ret = rrdpush_decompress_gzip(state, compressed_data, compressed_size); - break; - } - - // for backwards compatibility we cannot check for COMPRESSION_MAX_MSG_SIZE, - // because old children may send this big payloads. - if(unlikely(ret > COMPRESSION_MAX_CHUNK)) { - netdata_log_error("RRDPUSH_DECOMPRESS: decompressed data is %zu bytes, which is bigger than the max msg size %d", - ret, COMPRESSION_MAX_CHUNK); - return 0; - } - - return ret; -} - -// ---------------------------------------------------------------------------- -// unit test - -static inline long int my_random (void) { - return random(); -} - -void unittest_generate_random_name(char *dst, size_t size) { - if(size < 7) - size = 7; - - size_t len = 5 + my_random() % (size - 6); - - for(size_t i = 0; i < len ; i++) { - if(my_random() % 2 == 0) - dst[i] = 'A' + my_random() % 26; - else - dst[i] = 'a' + my_random() % 26; - } - - dst[len] = '\0'; -} - -void unittest_generate_message(BUFFER *wb, time_t now_s, size_t counter) { - bool with_slots = true; - NUMBER_ENCODING integer_encoding = NUMBER_ENCODING_BASE64; - NUMBER_ENCODING doubles_encoding = NUMBER_ENCODING_BASE64; - time_t update_every = 1; - time_t point_end_time_s = now_s; - time_t wall_clock_time_s = now_s; - size_t chart_slot = counter + 1; - size_t dimensions = 2 + my_random() % 5; - char chart[RRD_ID_LENGTH_MAX + 1] = "name"; - unittest_generate_random_name(chart, 5 + my_random() % 30); - - buffer_fast_strcat(wb, PLUGINSD_KEYWORD_BEGIN_V2, sizeof(PLUGINSD_KEYWORD_BEGIN_V2) - 1); - - if(with_slots) { - buffer_fast_strcat(wb, " "PLUGINSD_KEYWORD_SLOT":", sizeof(PLUGINSD_KEYWORD_SLOT) - 1 + 2); - buffer_print_uint64_encoded(wb, integer_encoding, chart_slot); - } - - buffer_fast_strcat(wb, " '", 2); - buffer_strcat(wb, chart); - buffer_fast_strcat(wb, "' ", 2); - buffer_print_uint64_encoded(wb, integer_encoding, update_every); - buffer_fast_strcat(wb, " ", 1); - buffer_print_uint64_encoded(wb, integer_encoding, point_end_time_s); - buffer_fast_strcat(wb, " ", 1); - if(point_end_time_s == wall_clock_time_s) - buffer_fast_strcat(wb, "#", 1); - else - buffer_print_uint64_encoded(wb, integer_encoding, wall_clock_time_s); - buffer_fast_strcat(wb, "\n", 1); - - - for(size_t d = 0; d < dimensions ;d++) { - size_t dim_slot = d + 1; - char dim_id[RRD_ID_LENGTH_MAX + 1] = "dimension"; - unittest_generate_random_name(dim_id, 10 + my_random() % 20); - int64_t last_collected_value = (my_random() % 2 == 0) ? (int64_t)(counter + d) : (int64_t)my_random(); - NETDATA_DOUBLE value = (my_random() % 2 == 0) ? (NETDATA_DOUBLE)my_random() / ((NETDATA_DOUBLE)my_random() + 1) : (NETDATA_DOUBLE)last_collected_value; - SN_FLAGS flags = (my_random() % 1000 == 0) ? SN_FLAG_NONE : SN_FLAG_NOT_ANOMALOUS; - - buffer_fast_strcat(wb, PLUGINSD_KEYWORD_SET_V2, sizeof(PLUGINSD_KEYWORD_SET_V2) - 1); - - if(with_slots) { - buffer_fast_strcat(wb, " "PLUGINSD_KEYWORD_SLOT":", sizeof(PLUGINSD_KEYWORD_SLOT) - 1 + 2); - buffer_print_uint64_encoded(wb, integer_encoding, dim_slot); - } - - buffer_fast_strcat(wb, " '", 2); - buffer_strcat(wb, dim_id); - buffer_fast_strcat(wb, "' ", 2); - buffer_print_int64_encoded(wb, integer_encoding, last_collected_value); - buffer_fast_strcat(wb, " ", 1); - - if((NETDATA_DOUBLE)last_collected_value == value) - buffer_fast_strcat(wb, "#", 1); - else - buffer_print_netdata_double_encoded(wb, doubles_encoding, value); - - buffer_fast_strcat(wb, " ", 1); - buffer_print_sn_flags(wb, flags, true); - buffer_fast_strcat(wb, "\n", 1); - } - - buffer_fast_strcat(wb, PLUGINSD_KEYWORD_END_V2 "\n", sizeof(PLUGINSD_KEYWORD_END_V2) - 1 + 1); -} - -int unittest_rrdpush_compression_speed(compression_algorithm_t algorithm, const char *name) { - fprintf(stderr, "\nTesting streaming compression speed with %s\n", name); - - struct compressor_state cctx = { - .initialized = false, - .algorithm = algorithm, - }; - struct decompressor_state dctx = { - .initialized = false, - .algorithm = algorithm, - }; - - rrdpush_compressor_init(&cctx); - rrdpush_decompressor_init(&dctx); - - int errors = 0; - - BUFFER *wb = buffer_create(COMPRESSION_MAX_MSG_SIZE, NULL); - time_t now_s = now_realtime_sec(); - usec_t compression_ut = 0; - usec_t decompression_ut = 0; - size_t bytes_compressed = 0; - size_t bytes_uncompressed = 0; - - usec_t compression_started_ut = now_monotonic_usec(); - usec_t decompression_started_ut = compression_started_ut; - - for(int i = 0; i < 10000 ;i++) { - compression_started_ut = now_monotonic_usec(); - decompression_ut += compression_started_ut - decompression_started_ut; - - buffer_flush(wb); - while(buffer_strlen(wb) < COMPRESSION_MAX_MSG_SIZE - 1024) - unittest_generate_message(wb, now_s, i); - - const char *txt = buffer_tostring(wb); - size_t txt_len = buffer_strlen(wb); - bytes_uncompressed += txt_len; - - const char *out; - size_t size = rrdpush_compress(&cctx, txt, txt_len, &out); - - bytes_compressed += size; - decompression_started_ut = now_monotonic_usec(); - compression_ut += decompression_started_ut - compression_started_ut; - - if(size == 0) { - fprintf(stderr, "iteration %d: compressed size %zu is zero\n", - i, size); - errors++; - goto cleanup; - } - else if(size >= COMPRESSION_MAX_CHUNK) { - fprintf(stderr, "iteration %d: compressed size %zu exceeds max allowed size\n", - i, size); - errors++; - goto cleanup; - } - else { - size_t dtxt_len = rrdpush_decompress(&dctx, out, size); - char *dtxt = (char *) &dctx.output.data[dctx.output.read_pos]; - - if(rrdpush_decompressed_bytes_in_buffer(&dctx) != dtxt_len) { - fprintf(stderr, "iteration %d: decompressed size %zu does not rrdpush_decompressed_bytes_in_buffer() %zu\n", - i, dtxt_len, rrdpush_decompressed_bytes_in_buffer(&dctx) - ); - errors++; - goto cleanup; - } - - if(!dtxt_len) { - fprintf(stderr, "iteration %d: decompressed size is zero\n", i); - errors++; - goto cleanup; - } - else if(dtxt_len != txt_len) { - fprintf(stderr, "iteration %d: decompressed size %zu does not match original size %zu\n", - i, dtxt_len, txt_len - ); - errors++; - goto cleanup; - } - else { - if(memcmp(txt, dtxt, txt_len) != 0) { - fprintf(stderr, "iteration %d: decompressed data '%s' do not match original data length %zu\n", - i, dtxt, txt_len); - errors++; - goto cleanup; - } - } - } - - // here we are supposed to copy the data and advance the position - dctx.output.read_pos += rrdpush_decompressed_bytes_in_buffer(&dctx); - } - -cleanup: - rrdpush_compressor_destroy(&cctx); - rrdpush_decompressor_destroy(&dctx); - - if(errors) - fprintf(stderr, "Compression with %s: FAILED (%d errors)\n", name, errors); - else - fprintf(stderr, "Compression with %s: OK " - "(compression %zu usec, decompression %zu usec, bytes raw %zu, compressed %zu, savings ratio %0.2f%%)\n", - name, compression_ut, decompression_ut, - bytes_uncompressed, bytes_compressed, - 100.0 - (double)bytes_compressed * 100.0 / (double)bytes_uncompressed); - - return errors; -} - -int unittest_rrdpush_compression(compression_algorithm_t algorithm, const char *name) { - fprintf(stderr, "\nTesting streaming compression with %s\n", name); - - struct compressor_state cctx = { - .initialized = false, - .algorithm = algorithm, - }; - struct decompressor_state dctx = { - .initialized = false, - .algorithm = algorithm, - }; - - char txt[COMPRESSION_MAX_MSG_SIZE]; - - rrdpush_compressor_init(&cctx); - rrdpush_decompressor_init(&dctx); - - int errors = 0; - - memset(txt, '=', COMPRESSION_MAX_MSG_SIZE); - - for(int i = 0; i < COMPRESSION_MAX_MSG_SIZE ;i++) { - txt[i] = 'A' + (i % 26); - size_t txt_len = i + 1; - - const char *out; - size_t size = rrdpush_compress(&cctx, txt, txt_len, &out); - - if(size == 0) { - fprintf(stderr, "iteration %d: compressed size %zu is zero\n", - i, size); - errors++; - goto cleanup; - } - else if(size >= COMPRESSION_MAX_CHUNK) { - fprintf(stderr, "iteration %d: compressed size %zu exceeds max allowed size\n", - i, size); - errors++; - goto cleanup; - } - else { - size_t dtxt_len = rrdpush_decompress(&dctx, out, size); - char *dtxt = (char *) &dctx.output.data[dctx.output.read_pos]; - - if(rrdpush_decompressed_bytes_in_buffer(&dctx) != dtxt_len) { - fprintf(stderr, "iteration %d: decompressed size %zu does not rrdpush_decompressed_bytes_in_buffer() %zu\n", - i, dtxt_len, rrdpush_decompressed_bytes_in_buffer(&dctx) - ); - errors++; - goto cleanup; - } - - if(!dtxt_len) { - fprintf(stderr, "iteration %d: decompressed size is zero\n", i); - errors++; - goto cleanup; - } - else if(dtxt_len != txt_len) { - fprintf(stderr, "iteration %d: decompressed size %zu does not match original size %zu\n", - i, dtxt_len, txt_len - ); - errors++; - goto cleanup; - } - else { - if(memcmp(txt, dtxt, txt_len) != 0) { - txt[txt_len] = '\0'; - dtxt[txt_len + 5] = '\0'; - - fprintf(stderr, "iteration %d: decompressed data '%s' do not match original data '%s' of length %zu\n", - i, dtxt, txt, txt_len); - errors++; - goto cleanup; - } - } - } - - // fill the compressed buffer with garbage - memset((void *)out, 'x', size); - - // here we are supposed to copy the data and advance the position - dctx.output.read_pos += rrdpush_decompressed_bytes_in_buffer(&dctx); - } - -cleanup: - rrdpush_compressor_destroy(&cctx); - rrdpush_decompressor_destroy(&dctx); - - if(errors) - fprintf(stderr, "Compression with %s: FAILED (%d errors)\n", name, errors); - else - fprintf(stderr, "Compression with %s: OK\n", name); - - return errors; -} - -int unittest_rrdpush_compressions(void) { - int ret = 0; - - ret += unittest_rrdpush_compression(COMPRESSION_ALGORITHM_ZSTD, "ZSTD"); - ret += unittest_rrdpush_compression(COMPRESSION_ALGORITHM_LZ4, "LZ4"); - ret += unittest_rrdpush_compression(COMPRESSION_ALGORITHM_BROTLI, "BROTLI"); - ret += unittest_rrdpush_compression(COMPRESSION_ALGORITHM_GZIP, "GZIP"); - - ret += unittest_rrdpush_compression_speed(COMPRESSION_ALGORITHM_ZSTD, "ZSTD"); - ret += unittest_rrdpush_compression_speed(COMPRESSION_ALGORITHM_LZ4, "LZ4"); - ret += unittest_rrdpush_compression_speed(COMPRESSION_ALGORITHM_BROTLI, "BROTLI"); - ret += unittest_rrdpush_compression_speed(COMPRESSION_ALGORITHM_GZIP, "GZIP"); - - return ret; -} |