@cryptotaxi247 / netdata-1
Code Commits Issues Pulls Sessions CI/CD
master
master
c 698 lines 23.8 KB
Raw
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 #include "compression.h"
4 #include "../stream-conf.h"
5 #include "../stream-receiver-internals.h"
6 #include "../stream-sender-internals.h"
7
8 #include "gzip.h"
9
10 #ifdef ENABLE_LZ4
11 #include "lz4.h"
12 #endif
13
14 #ifdef ENABLE_ZSTD
15 #include "zstd.h"
16 #endif
17
18 #ifdef ENABLE_BROTLI
19 #include "brotli.h"
20 #endif
21
22 void stream_parse_compression_order(struct stream_receiver_config *config, const char *order) {
23 // empty all slots
24 for(size_t i = 0; i < COMPRESSION_ALGORITHM_MAX ;i++)
25 config->compression.priorities[i] = STREAM_CAP_NONE;
26
27 char *s = strdupz(order);
28
29 char *words[COMPRESSION_ALGORITHM_MAX + 100] = { NULL };
30 size_t num_words = quoted_strings_splitter_whitespace(s, words, COMPRESSION_ALGORITHM_MAX + 100);
31 size_t slot = 0;
32 STREAM_CAPABILITIES added = STREAM_CAP_NONE;
33 for(size_t i = 0; i < num_words && slot < COMPRESSION_ALGORITHM_MAX ;i++) {
34 if((STREAM_CAP_ZSTD_AVAILABLE) && strcasecmp(words[i], "zstd") == 0 && !(added & STREAM_CAP_ZSTD)) {
35 config->compression.priorities[slot++] = STREAM_CAP_ZSTD;
36 added |= STREAM_CAP_ZSTD;
37 }
38 else if((STREAM_CAP_LZ4_AVAILABLE) && strcasecmp(words[i], "lz4") == 0 && !(added & STREAM_CAP_LZ4)) {
39 config->compression.priorities[slot++] = STREAM_CAP_LZ4;
40 added |= STREAM_CAP_LZ4;
41 }
42 else if((STREAM_CAP_BROTLI_AVAILABLE) && strcasecmp(words[i], "brotli") == 0 && !(added & STREAM_CAP_BROTLI)) {
43 config->compression.priorities[slot++] = STREAM_CAP_BROTLI;
44 added |= STREAM_CAP_BROTLI;
45 }
46 else if(strcasecmp(words[i], "gzip") == 0 && !(added & STREAM_CAP_GZIP)) {
47 config->compression.priorities[slot++] = STREAM_CAP_GZIP;
48 added |= STREAM_CAP_GZIP;
49 }
50 }
51
52 freez(s);
53
54 // make sure all participate
55 if((STREAM_CAP_ZSTD_AVAILABLE) && slot < COMPRESSION_ALGORITHM_MAX && !(added & STREAM_CAP_ZSTD))
56 config->compression.priorities[slot++] = STREAM_CAP_ZSTD;
57 if((STREAM_CAP_LZ4_AVAILABLE) && slot < COMPRESSION_ALGORITHM_MAX && !(added & STREAM_CAP_LZ4))
58 config->compression.priorities[slot++] = STREAM_CAP_LZ4;
59 if((STREAM_CAP_BROTLI_AVAILABLE) && slot < COMPRESSION_ALGORITHM_MAX && !(added & STREAM_CAP_BROTLI))
60 config->compression.priorities[slot++] = STREAM_CAP_BROTLI;
61 if(slot < COMPRESSION_ALGORITHM_MAX && !(added & STREAM_CAP_GZIP))
62 config->compression.priorities[slot++] = STREAM_CAP_GZIP;
63 }
64
65 void stream_select_receiver_compression_algorithm(struct receiver_state *rpt) {
66 if (!rpt->config.compression.enabled)
67 rpt->capabilities &= ~STREAM_CAP_COMPRESSIONS_AVAILABLE;
68
69 // select the right compression before sending our capabilities to the child
70 if(stream_has_more_than_one_capability_of(rpt->capabilities, STREAM_CAP_COMPRESSIONS_AVAILABLE)) {
71 STREAM_CAPABILITIES compressions = rpt->capabilities & STREAM_CAP_COMPRESSIONS_AVAILABLE;
72 for(int i = 0; i < COMPRESSION_ALGORITHM_MAX; i++) {
73 STREAM_CAPABILITIES c = rpt->config.compression.priorities[i];
74
75 if(!(c & STREAM_CAP_COMPRESSIONS_AVAILABLE))
76 continue;
77
78 if(compressions & c) {
79 STREAM_CAPABILITIES exclude = compressions;
80 exclude &= ~c;
81
82 rpt->capabilities &= ~exclude;
83 break;
84 }
85 }
86 }
87 }
88
89 bool stream_compression_initialize(struct sender_state *s) {
90 stream_compressor_destroy(&s->thread.compressor);
91
92 // IMPORTANT
93 // KEEP THE SAME ORDER IN DECOMPRESSION
94
95 if(stream_has_capability(s, STREAM_CAP_ZSTD))
96 s->thread.compressor.algorithm = COMPRESSION_ALGORITHM_ZSTD;
97 else if(stream_has_capability(s, STREAM_CAP_LZ4))
98 s->thread.compressor.algorithm = COMPRESSION_ALGORITHM_LZ4;
99 else if(stream_has_capability(s, STREAM_CAP_BROTLI))
100 s->thread.compressor.algorithm = COMPRESSION_ALGORITHM_BROTLI;
101 else if(stream_has_capability(s, STREAM_CAP_GZIP))
102 s->thread.compressor.algorithm = COMPRESSION_ALGORITHM_GZIP;
103 else
104 s->thread.compressor.algorithm = COMPRESSION_ALGORITHM_NONE;
105
106 if(s->thread.compressor.algorithm != COMPRESSION_ALGORITHM_NONE) {
107 s->thread.compressor.level = stream_send.compression.levels[s->thread.compressor.algorithm];
108 stream_compressor_init(&s->thread.compressor);
109 return true;
110 }
111
112 return false;
113 }
114
115 bool stream_decompression_initialize(struct receiver_state *rpt) {
116 stream_decompressor_destroy(&rpt->thread.compressed.decompressor);
117
118 // IMPORTANT
119 // KEEP THE SAME ORDER IN COMPRESSION
120
121 if(stream_has_capability(rpt, STREAM_CAP_ZSTD))
122 rpt->thread.compressed.decompressor.algorithm = COMPRESSION_ALGORITHM_ZSTD;
123 else if(stream_has_capability(rpt, STREAM_CAP_LZ4))
124 rpt->thread.compressed.decompressor.algorithm = COMPRESSION_ALGORITHM_LZ4;
125 else if(stream_has_capability(rpt, STREAM_CAP_BROTLI))
126 rpt->thread.compressed.decompressor.algorithm = COMPRESSION_ALGORITHM_BROTLI;
127 else if(stream_has_capability(rpt, STREAM_CAP_GZIP))
128 rpt->thread.compressed.decompressor.algorithm = COMPRESSION_ALGORITHM_GZIP;
129 else
130 rpt->thread.compressed.decompressor.algorithm = COMPRESSION_ALGORITHM_NONE;
131
132 if(rpt->thread.compressed.decompressor.algorithm != COMPRESSION_ALGORITHM_NONE) {
133 stream_decompressor_init(&rpt->thread.compressed.decompressor);
134 return true;
135 }
136
137 return false;
138 }
139
140 /*
141 * In case of stream compression buffer overflow
142 * Inform the user through the error log file and
143 * deactivate compression by downgrading the stream protocol.
144 */
145 void stream_compression_deactivate(struct sender_state *s) {
146 switch(s->thread.compressor.algorithm) {
147 case COMPRESSION_ALGORITHM_MAX:
148 case COMPRESSION_ALGORITHM_NONE:
149 netdata_log_error("STREAM_COMPRESSION: compression error on 'host:%s' without any compression enabled. Ignoring error.",
150 rrdhost_hostname(s->host));
151 break;
152
153 case COMPRESSION_ALGORITHM_GZIP:
154 netdata_log_error("STREAM_COMPRESSION: GZIP compression error on 'host:%s'. Disabling GZIP for this node.",
155 rrdhost_hostname(s->host));
156 s->disabled_capabilities |= STREAM_CAP_GZIP;
157 break;
158
159 case COMPRESSION_ALGORITHM_LZ4:
160 netdata_log_error("STREAM_COMPRESSION: LZ4 compression error on 'host:%s'. Disabling ZSTD for this node.",
161 rrdhost_hostname(s->host));
162 s->disabled_capabilities |= STREAM_CAP_LZ4;
163 break;
164
165 case COMPRESSION_ALGORITHM_ZSTD:
166 netdata_log_error("STREAM_COMPRESSION: ZSTD compression error on 'host:%s'. Disabling ZSTD for this node.",
167 rrdhost_hostname(s->host));
168 s->disabled_capabilities |= STREAM_CAP_ZSTD;
169 break;
170
171 case COMPRESSION_ALGORITHM_BROTLI:
172 netdata_log_error("STREAM_COMPRESSION: BROTLI compression error on 'host:%s'. Disabling BROTLI for this node.",
173 rrdhost_hostname(s->host));
174 s->disabled_capabilities |= STREAM_CAP_BROTLI;
175 break;
176 }
177 }
178
179 // ----------------------------------------------------------------------------
180 // compressor public API
181
182 void stream_compressor_init(struct compressor_state *state) {
183 switch(state->algorithm) {
184 #ifdef ENABLE_ZSTD
185 case COMPRESSION_ALGORITHM_ZSTD:
186 stream_compressor_init_zstd(state);
187 break;
188 #endif
189
190 #ifdef ENABLE_LZ4
191 case COMPRESSION_ALGORITHM_LZ4:
192 stream_compressor_init_lz4(state);
193 break;
194 #endif
195
196 #ifdef ENABLE_BROTLI
197 case COMPRESSION_ALGORITHM_BROTLI:
198 stream_compressor_init_brotli(state);
199 break;
200 #endif
201
202 default:
203 case COMPRESSION_ALGORITHM_GZIP:
204 stream_compressor_init_gzip(state);
205 break;
206 }
207
208 simple_ring_buffer_reset(&state->input);
209 simple_ring_buffer_reset(&state->output);
210 }
211
212 void stream_compressor_destroy(struct compressor_state *state) {
213 switch(state->algorithm) {
214 #ifdef ENABLE_ZSTD
215 case COMPRESSION_ALGORITHM_ZSTD:
216 stream_compressor_destroy_zstd(state);
217 break;
218 #endif
219
220 #ifdef ENABLE_LZ4
221 case COMPRESSION_ALGORITHM_LZ4:
222 stream_compressor_destroy_lz4(state);
223 break;
224 #endif
225
226 #ifdef ENABLE_BROTLI
227 case COMPRESSION_ALGORITHM_BROTLI:
228 stream_compressor_destroy_brotli(state);
229 break;
230 #endif
231
232 default:
233 case COMPRESSION_ALGORITHM_GZIP:
234 stream_compressor_destroy_gzip(state);
235 break;
236 }
237
238 state->initialized = false;
239
240 simple_ring_buffer_destroy(&state->input);
241 simple_ring_buffer_destroy(&state->output);
242 }
243
244 size_t stream_compress(struct compressor_state *state, const char *data, size_t size, const char **out) {
245 size_t ret = 0;
246
247 switch(state->algorithm) {
248 #ifdef ENABLE_ZSTD
249 case COMPRESSION_ALGORITHM_ZSTD:
250 ret = stream_compress_zstd(state, data, size, out);
251 break;
252 #endif
253
254 #ifdef ENABLE_LZ4
255 case COMPRESSION_ALGORITHM_LZ4:
256 ret = stream_compress_lz4(state, data, size, out);
257 break;
258 #endif
259
260 #ifdef ENABLE_BROTLI
261 case COMPRESSION_ALGORITHM_BROTLI:
262 ret = stream_compress_brotli(state, data, size, out);
263 break;
264 #endif
265
266 default:
267 case COMPRESSION_ALGORITHM_GZIP:
268 ret = stream_compress_gzip(state, data, size, out);
269 break;
270 }
271
272 if(unlikely(ret >= COMPRESSION_MAX_CHUNK)) {
273 netdata_log_error("STREAM_COMPRESS: compressed data is %zu bytes, which is >= than the max chunk size %d",
274 ret, COMPRESSION_MAX_CHUNK);
275 return 0;
276 }
277
278 return ret;
279 }
280
281 // ----------------------------------------------------------------------------
282 // decompressor public API
283
284 void stream_decompressor_destroy(struct decompressor_state *state) {
285 if(unlikely(!state->initialized))
286 return;
287
288 switch(state->algorithm) {
289 #ifdef ENABLE_ZSTD
290 case COMPRESSION_ALGORITHM_ZSTD:
291 stream_decompressor_destroy_zstd(state);
292 break;
293 #endif
294
295 #ifdef ENABLE_LZ4
296 case COMPRESSION_ALGORITHM_LZ4:
297 stream_decompressor_destroy_lz4(state);
298 break;
299 #endif
300
301 #ifdef ENABLE_BROTLI
302 case COMPRESSION_ALGORITHM_BROTLI:
303 stream_decompressor_destroy_brotli(state);
304 break;
305 #endif
306
307 default:
308 case COMPRESSION_ALGORITHM_GZIP:
309 stream_decompressor_destroy_gzip(state);
310 break;
311 }
312
313 simple_ring_buffer_destroy(&state->output);
314
315 state->initialized = false;
316 }
317
318 void stream_decompressor_init(struct decompressor_state *state) {
319 switch(state->algorithm) {
320 #ifdef ENABLE_ZSTD
321 case COMPRESSION_ALGORITHM_ZSTD:
322 stream_decompressor_init_zstd(state);
323 break;
324 #endif
325
326 #ifdef ENABLE_LZ4
327 case COMPRESSION_ALGORITHM_LZ4:
328 stream_decompressor_init_lz4(state);
329 break;
330 #endif
331
332 #ifdef ENABLE_BROTLI
333 case COMPRESSION_ALGORITHM_BROTLI:
334 stream_decompressor_init_brotli(state);
335 break;
336 #endif
337
338 default:
339 case COMPRESSION_ALGORITHM_GZIP:
340 stream_decompressor_init_gzip(state);
341 break;
342 }
343
344 state->signature_size = STREAM_COMPRESSION_SIGNATURE_SIZE;
345 simple_ring_buffer_reset(&state->output);
346 }
347
348 size_t stream_decompress(struct decompressor_state *state, const char *compressed_data, size_t compressed_size) {
349 if (unlikely(state->output.read_pos != state->output.write_pos))
350 fatal("STREAM_DECOMPRESS: asked to decompress new data, while there are unread data in the decompression buffer!");
351
352 size_t ret = 0;
353
354 switch(state->algorithm) {
355 #ifdef ENABLE_ZSTD
356 case COMPRESSION_ALGORITHM_ZSTD:
357 ret = stream_decompress_zstd(state, compressed_data, compressed_size);
358 break;
359 #endif
360
361 #ifdef ENABLE_LZ4
362 case COMPRESSION_ALGORITHM_LZ4:
363 ret = stream_decompress_lz4(state, compressed_data, compressed_size);
364 break;
365 #endif
366
367 #ifdef ENABLE_BROTLI
368 case COMPRESSION_ALGORITHM_BROTLI:
369 ret = stream_decompress_brotli(state, compressed_data, compressed_size);
370 break;
371 #endif
372
373 default:
374 case COMPRESSION_ALGORITHM_GZIP:
375 ret = stream_decompress_gzip(state, compressed_data, compressed_size);
376 break;
377 }
378
379 // for backwards compatibility we cannot check for COMPRESSION_MAX_MSG_SIZE,
380 // because old children may send this big payloads.
381 if(unlikely(ret > COMPRESSION_MAX_CHUNK)) {
382 netdata_log_error("STREAM_DECOMPRESS: decompressed data is %zu bytes, which is bigger than the max msg size %d",
383 ret, COMPRESSION_MAX_CHUNK);
384 return 0;
385 }
386
387 return ret;
388 }
389
390 // ----------------------------------------------------------------------------
391 // unit test
392
393 void unittest_generate_random_name(char *dst, size_t size) {
394 if(size < 7)
395 size = 7;
396
397 size_t len = 5 + os_random32() % (size - 6);
398
399 for(size_t i = 0; i < len ; i++) {
400 if(os_random8() % 2 == 0)
401 dst[i] = 'A' + os_random8() % 26;
402 else
403 dst[i] = 'a' + os_random8() % 26;
404 }
405
406 dst[len] = '\0';
407 }
408
409 void unittest_generate_message(BUFFER *wb, time_t now_s, size_t counter) {
410 bool with_slots = true;
411 NUMBER_ENCODING integer_encoding = NUMBER_ENCODING_BASE64;
412 NUMBER_ENCODING doubles_encoding = NUMBER_ENCODING_BASE64;
413 time_t update_every = 1;
414 time_t point_end_time_s = now_s;
415 time_t wall_clock_time_s = now_s;
416 size_t chart_slot = counter + 1;
417 size_t dimensions = 2 + os_random8() % 5;
418 char chart[RRD_ID_LENGTH_MAX + 1] = "name";
419 unittest_generate_random_name(chart, 5 + os_random8() % 30);
420
421 buffer_fast_strcat(wb, PLUGINSD_KEYWORD_BEGIN_V2, sizeof(PLUGINSD_KEYWORD_BEGIN_V2) - 1);
422
423 if(with_slots) {
424 buffer_fast_strcat(wb, " "PLUGINSD_KEYWORD_SLOT":", sizeof(PLUGINSD_KEYWORD_SLOT) - 1 + 2);
425 buffer_print_uint64_encoded(wb, integer_encoding, chart_slot);
426 }
427
428 buffer_fast_strcat(wb, " '", 2);
429 buffer_strcat(wb, chart);
430 buffer_fast_strcat(wb, "' ", 2);
431 buffer_print_uint64_encoded(wb, integer_encoding, update_every);
432 buffer_fast_strcat(wb, " ", 1);
433 buffer_print_uint64_encoded(wb, integer_encoding, point_end_time_s);
434 buffer_fast_strcat(wb, " ", 1);
435 if(point_end_time_s == wall_clock_time_s)
436 buffer_fast_strcat(wb, "#", 1);
437 else
438 buffer_print_uint64_encoded(wb, integer_encoding, wall_clock_time_s);
439 buffer_fast_strcat(wb, "\n", 1);
440
441
442 for(size_t d = 0; d < dimensions ;d++) {
443 size_t dim_slot = d + 1;
444 char dim_id[RRD_ID_LENGTH_MAX + 1] = "dimension";
445 unittest_generate_random_name(dim_id, 10 + os_random8() % 20);
446 int64_t last_collected_value = (os_random8() % 2 == 0) ? (int64_t)(counter + d) : (int64_t)os_random32();
447 NETDATA_DOUBLE value = (os_random8() % 2 == 0) ? (NETDATA_DOUBLE)os_random64() / ((NETDATA_DOUBLE)os_random64() + 1) : (NETDATA_DOUBLE)last_collected_value;
448 SN_FLAGS flags = (os_random16() % 1000 == 0) ? SN_FLAG_NONE : SN_FLAG_NOT_ANOMALOUS;
449
450 buffer_fast_strcat(wb, PLUGINSD_KEYWORD_SET_V2, sizeof(PLUGINSD_KEYWORD_SET_V2) - 1);
451
452 if(with_slots) {
453 buffer_fast_strcat(wb, " "PLUGINSD_KEYWORD_SLOT":", sizeof(PLUGINSD_KEYWORD_SLOT) - 1 + 2);
454 buffer_print_uint64_encoded(wb, integer_encoding, dim_slot);
455 }
456
457 buffer_fast_strcat(wb, " '", 2);
458 buffer_strcat(wb, dim_id);
459 buffer_fast_strcat(wb, "' ", 2);
460 buffer_print_int64_encoded(wb, integer_encoding, last_collected_value);
461 buffer_fast_strcat(wb, " ", 1);
462
463 if((NETDATA_DOUBLE)last_collected_value == value)
464 buffer_fast_strcat(wb, "#", 1);
465 else
466 buffer_print_netdata_double_encoded(wb, doubles_encoding, value);
467
468 buffer_fast_strcat(wb, " ", 1);
469 buffer_print_sn_flags(wb, flags, true);
470 buffer_fast_strcat(wb, "\n", 1);
471 }
472
473 buffer_fast_strcat(wb, PLUGINSD_KEYWORD_END_V2 "\n", sizeof(PLUGINSD_KEYWORD_END_V2) - 1 + 1);
474 }
475
476 int unittest_stream_compression_speed(compression_algorithm_t algorithm, const char *name) {
477 fprintf(stderr, "\nTesting streaming compression speed with %s\n", name);
478
479 struct compressor_state cctx = {
480 .initialized = false,
481 .algorithm = algorithm,
482 };
483 struct decompressor_state dctx = {
484 .initialized = false,
485 .algorithm = algorithm,
486 };
487
488 stream_compressor_init(&cctx);
489 stream_decompressor_init(&dctx);
490
491 int errors = 0;
492
493 BUFFER *wb = buffer_create(COMPRESSION_MAX_MSG_SIZE, NULL);
494 time_t now_s = now_realtime_sec();
495 usec_t compression_ut = 0;
496 usec_t decompression_ut = 0;
497 size_t bytes_compressed = 0;
498 size_t bytes_uncompressed = 0;
499
500 usec_t compression_started_ut = now_monotonic_usec();
501 usec_t decompression_started_ut = compression_started_ut;
502
503 for(int i = 0; i < 10000 ;i++) {
504 compression_started_ut = now_monotonic_usec();
505 decompression_ut += compression_started_ut - decompression_started_ut;
506
507 buffer_flush(wb);
508 while(buffer_strlen(wb) < COMPRESSION_MAX_MSG_SIZE - 1024)
509 unittest_generate_message(wb, now_s, i);
510
511 const char *txt = buffer_tostring(wb);
512 size_t txt_len = buffer_strlen(wb);
513 bytes_uncompressed += txt_len;
514
515 const char *out;
516 size_t size = stream_compress(&cctx, txt, txt_len, &out);
517
518 bytes_compressed += size;
519 decompression_started_ut = now_monotonic_usec();
520 compression_ut += decompression_started_ut - compression_started_ut;
521
522 if(size == 0) {
523 fprintf(stderr, "iteration %d: compressed size %zu is zero\n",
524 i, size);
525 errors++;
526 goto cleanup;
527 }
528 else if(size >= COMPRESSION_MAX_CHUNK) {
529 fprintf(stderr, "iteration %d: compressed size %zu exceeds max allowed size\n",
530 i, size);
531 errors++;
532 goto cleanup;
533 }
534 else {
535 size_t dtxt_len = stream_decompress(&dctx, out, size);
536 char *dtxt = (char *) &dctx.output.data[dctx.output.read_pos];
537
538 if(stream_decompressed_bytes_in_buffer(&dctx) != dtxt_len) {
539 fprintf(stderr, "iteration %d: decompressed size %zu does not stream_decompressed_bytes_in_buffer() %zu\n",
540 i, dtxt_len, stream_decompressed_bytes_in_buffer(&dctx));
541 errors++;
542 goto cleanup;
543 }
544
545 if(!dtxt_len) {
546 fprintf(stderr, "iteration %d: decompressed size is zero\n", i);
547 errors++;
548 goto cleanup;
549 }
550 else if(dtxt_len != txt_len) {
551 fprintf(stderr, "iteration %d: decompressed size %zu does not match original size %zu\n",
552 i, dtxt_len, txt_len
553 );
554 errors++;
555 goto cleanup;
556 }
557 else {
558 if(memcmp(txt, dtxt, txt_len) != 0) {
559 fprintf(stderr, "iteration %d: decompressed data '%s' do not match original data length %zu\n",
560 i, dtxt, txt_len);
561 errors++;
562 goto cleanup;
563 }
564 }
565 }
566
567 // here we are supposed to copy the data and advance the position
568 dctx.output.read_pos += stream_decompressed_bytes_in_buffer(&dctx);
569 }
570
571 cleanup:
572 stream_compressor_destroy(&cctx);
573 stream_decompressor_destroy(&dctx);
574
575 if(errors)
576 fprintf(stderr, "Compression with %s: FAILED (%d errors)\n", name, errors);
577 else
578 fprintf(stderr, "Compression with %s: OK "
579 "(compression %llu usec, decompression %llu usec, bytes raw %zu, compressed %zu, savings ratio %0.2f%%)\n",
580 name, (long long unsigned)compression_ut, (long long unsigned)decompression_ut,
581 bytes_uncompressed, bytes_compressed,
582 100.0 - (double)bytes_compressed * 100.0 / (double)bytes_uncompressed);
583
584 return errors;
585 }
586
587 int unittest_stream_compression(compression_algorithm_t algorithm, const char *name) {
588 fprintf(stderr, "\nTesting streaming compression with %s\n", name);
589
590 struct compressor_state cctx = {
591 .initialized = false,
592 .algorithm = algorithm,
593 };
594 struct decompressor_state dctx = {
595 .initialized = false,
596 .algorithm = algorithm,
597 };
598
599 char txt[COMPRESSION_MAX_MSG_SIZE];
600
601 stream_compressor_init(&cctx);
602 stream_decompressor_init(&dctx);
603
604 int errors = 0;
605
606 memset(txt, '=', COMPRESSION_MAX_MSG_SIZE);
607
608 for(int i = 0; i < COMPRESSION_MAX_MSG_SIZE ;i++) {
609 txt[i] = 'A' + (i % 26);
610 size_t txt_len = i + 1;
611
612 const char *out;
613 size_t size = stream_compress(&cctx, txt, txt_len, &out);
614
615 if(size == 0) {
616 fprintf(stderr, "iteration %d: compressed size %zu is zero\n",
617 i, size);
618 errors++;
619 goto cleanup;
620 }
621 else if(size >= COMPRESSION_MAX_CHUNK) {
622 fprintf(stderr, "iteration %d: compressed size %zu exceeds max allowed size\n",
623 i, size);
624 errors++;
625 goto cleanup;
626 }
627 else {
628 size_t dtxt_len = stream_decompress(&dctx, out, size);
629 char *dtxt = (char *) &dctx.output.data[dctx.output.read_pos];
630
631 if(stream_decompressed_bytes_in_buffer(&dctx) != dtxt_len) {
632 fprintf(stderr, "iteration %d: decompressed size %zu does not stream_decompressed_bytes_in_buffer() %zu\n",
633 i, dtxt_len,
634 stream_decompressed_bytes_in_buffer(&dctx)
635 );
636 errors++;
637 goto cleanup;
638 }
639
640 if(!dtxt_len) {
641 fprintf(stderr, "iteration %d: decompressed size is zero\n", i);
642 errors++;
643 goto cleanup;
644 }
645 else if(dtxt_len != txt_len) {
646 fprintf(stderr, "iteration %d: decompressed size %zu does not match original size %zu\n",
647 i, dtxt_len, txt_len
648 );
649 errors++;
650 goto cleanup;
651 }
652 else {
653 if(memcmp(txt, dtxt, txt_len) != 0) {
654 txt[txt_len] = '\0';
655 dtxt[txt_len + 5] = '\0';
656
657 fprintf(stderr, "iteration %d: decompressed data '%s' do not match original data '%s' of length %zu\n",
658 i, dtxt, txt, txt_len);
659 errors++;
660 goto cleanup;
661 }
662 }
663 }
664
665 // fill the compressed buffer with garbage
666 memset((void *)out, 'x', size);
667
668 // here we are supposed to copy the data and advance the position
669 dctx.output.read_pos += stream_decompressed_bytes_in_buffer(&dctx);
670 }
671
672 cleanup:
673 stream_compressor_destroy(&cctx);
674 stream_decompressor_destroy(&dctx);
675
676 if(errors)
677 fprintf(stderr, "Compression with %s: FAILED (%d errors)\n", name, errors);
678 else
679 fprintf(stderr, "Compression with %s: OK\n", name);
680
681 return errors;
682 }
683
684 int unittest_stream_compressions(void) {
685 int ret = 0;
686
687 ret += unittest_stream_compression(COMPRESSION_ALGORITHM_ZSTD, "ZSTD");
688 ret += unittest_stream_compression(COMPRESSION_ALGORITHM_LZ4, "LZ4");
689 ret += unittest_stream_compression(COMPRESSION_ALGORITHM_BROTLI, "BROTLI");
690 ret += unittest_stream_compression(COMPRESSION_ALGORITHM_GZIP, "GZIP");
691
692 ret += unittest_stream_compression_speed(COMPRESSION_ALGORITHM_ZSTD, "ZSTD");
693 ret += unittest_stream_compression_speed(COMPRESSION_ALGORITHM_LZ4, "LZ4");
694 ret += unittest_stream_compression_speed(COMPRESSION_ALGORITHM_BROTLI, "BROTLI");
695 ret += unittest_stream_compression_speed(COMPRESSION_ALGORITHM_GZIP, "GZIP");
696
697 return ret;
698 }