@cryptotaxi247 / netdata-1
Code Commits Issues Pulls Sessions CI/CD
master
master
c 144 lines 5.02 KB
Raw
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 #include "libnetdata/libnetdata.h"
4 #include "lz4.h"
5
6 #ifdef ENABLE_LZ4
7 #include <lz4.h>
8
9 // ----------------------------------------------------------------------------
10 // compress
11
12 void stream_compressor_init_lz4(struct compressor_state *state) {
13 if(!state->initialized) {
14 state->initialized = true;
15 state->stream = LZ4_createStream();
16
17 // LZ4 needs access to the last 64KB of source data
18 // so, we keep twice the size of each message
19 simple_ring_buffer_make_room(&state->input, 65536 + COMPRESSION_MAX_CHUNK * 2);
20 }
21 }
22
23 void stream_compressor_destroy_lz4(struct compressor_state *state) {
24 if (state->stream) {
25 LZ4_freeStream(state->stream);
26 state->stream = NULL;
27 }
28 }
29
30 /*
31 * Compress the given block of data
32 * Compressed data will remain in the internal buffer until the next invocation
33 * Return the size of compressed data block as result and the pointer to internal buffer using the last argument
34 * or 0 in case of error
35 */
36 size_t stream_compress_lz4(struct compressor_state *state, const char *data, size_t size, const char **out) {
37 if(unlikely(!state || !size || !out))
38 return 0;
39
40 // we need to keep the last 64K of our previous source data
41 // as they were in the ring buffer
42
43 simple_ring_buffer_make_room(&state->output, LZ4_COMPRESSBOUND(size));
44
45 if(state->input.write_pos + size > state->input.size)
46 // the input buffer cannot fit out data, restart from zero
47 simple_ring_buffer_reset(&state->input);
48
49 simple_ring_buffer_append_data(&state->input, data, size);
50
51 long int compressed_data_size = LZ4_compress_fast_continue(
52 state->stream,
53 state->input.data + state->input.read_pos,
54 (char *)state->output.data,
55 (int)(state->input.write_pos - state->input.read_pos),
56 (int)state->output.size,
57 state->level);
58
59 if (compressed_data_size <= 0) {
60 netdata_log_error("STREAM_COMPRESS: LZ4_compress_fast_continue() returned %ld "
61 "(source is %zu bytes, output buffer can fit %zu bytes)",
62 compressed_data_size, size, state->output.size);
63 return 0;
64 }
65
66 state->input.read_pos = state->input.write_pos;
67
68 state->sender_locked.total_compressions++;
69 state->sender_locked.total_uncompressed += size;
70 state->sender_locked.total_compressed += compressed_data_size;
71
72 *out = state->output.data;
73 return compressed_data_size;
74 }
75
76 // ----------------------------------------------------------------------------
77 // decompress
78
79 void stream_decompressor_init_lz4(struct decompressor_state *state) {
80 if(!state->initialized) {
81 state->initialized = true;
82 state->stream = LZ4_createStreamDecode();
83 simple_ring_buffer_make_room(&state->output, 65536 + COMPRESSION_MAX_CHUNK * 2);
84 }
85 }
86
87 void stream_decompressor_destroy_lz4(struct decompressor_state *state) {
88 if (state->stream) {
89 LZ4_freeStreamDecode(state->stream);
90 state->stream = NULL;
91 }
92 }
93
94 /*
95 * Decompress the compressed data in the internal buffer
96 * Return the size of uncompressed data or 0 for error
97 */
98 size_t stream_decompress_lz4(struct decompressor_state *state, const char *compressed_data, size_t compressed_size) {
99 if (unlikely(!state || !compressed_data || !compressed_size))
100 return 0;
101
102 // The state.output ring buffer is always EMPTY at this point,
103 // meaning that (state->output.read_pos == state->output.write_pos)
104 // However, THEY ARE NOT ZERO.
105
106 if (unlikely(state->output.write_pos + COMPRESSION_MAX_CHUNK > state->output.size))
107 // the input buffer cannot fit out data, restart from zero
108 simple_ring_buffer_reset(&state->output);
109
110 long int decompressed_size = LZ4_decompress_safe_continue(
111 state->stream
112 , compressed_data
113 , (char *)(state->output.data + state->output.write_pos)
114 , (int)compressed_size
115 , (int)(state->output.size - state->output.write_pos)
116 );
117
118 if (unlikely(decompressed_size < 0)) {
119 netdata_log_error("STREAM_DECOMPRESS: LZ4_decompress_safe_continue() returned negative value: %ld "
120 "(compressed chunk is %zu bytes)"
121 , decompressed_size, compressed_size);
122 return 0;
123 }
124
125 if(unlikely(decompressed_size + state->output.write_pos > state->output.size))
126 fatal("STREAM_DECOMPRESS: LZ4_decompress_safe_continue() overflown the stream_buffer "
127 "(size: %zu, pos: %zu, added: %ld, exceeding the buffer by %zu)"
128 , state->output.size
129 , state->output.write_pos
130 , decompressed_size
131 , (size_t)(state->output.write_pos + decompressed_size - state->output.size)
132 );
133
134 state->output.write_pos += decompressed_size;
135
136 // statistics
137 state->total_compressed += compressed_size;
138 state->total_uncompressed += decompressed_size;
139 state->total_compressions++;
140
141 return decompressed_size;
142 }
143
144 #endif // ENABLE_LZ4