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