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
|
// SPDX-License-Identifier: GPL-3.0-or-later
#include "compression_brotli.h"
#ifdef ENABLE_BROTLI
#include <brotli/encode.h>
#include <brotli/decode.h>
void rrdpush_compressor_init_brotli(struct compressor_state *state) {
if (!state->initialized) {
state->initialized = true;
state->stream = BrotliEncoderCreateInstance(NULL, NULL, NULL);
if (state->level < BROTLI_MIN_QUALITY) {
state->level = BROTLI_MIN_QUALITY;
} else if (state->level > BROTLI_MAX_QUALITY) {
state->level = BROTLI_MAX_QUALITY;
}
BrotliEncoderSetParameter(state->stream, BROTLI_PARAM_QUALITY, state->level);
}
}
void rrdpush_compressor_destroy_brotli(struct compressor_state *state) {
if (state->stream) {
BrotliEncoderDestroyInstance(state->stream);
state->stream = NULL;
}
}
size_t rrdpush_compress_brotli(struct compressor_state *state, const char *data, size_t size, const char **out) {
if (unlikely(!state || !size || !out))
return 0;
simple_ring_buffer_make_room(&state->output, MAX(BrotliEncoderMaxCompressedSize(size), COMPRESSION_MAX_CHUNK));
size_t available_out = state->output.size;
size_t available_in = size;
const uint8_t *next_in = (const uint8_t *)data;
uint8_t *next_out = (uint8_t *)state->output.data;
if (!BrotliEncoderCompressStream(state->stream, BROTLI_OPERATION_FLUSH, &available_in, &next_in, &available_out, &next_out, NULL)) {
netdata_log_error("STREAM: Brotli compression failed.");
return 0;
}
if(available_in != 0) {
netdata_log_error("STREAM: BrotliEncoderCompressStream() did not use all the input buffer, %zu bytes out of %zu remain",
available_in, size);
return 0;
}
size_t compressed_size = state->output.size - available_out;
if(available_out == 0) {
netdata_log_error("STREAM: BrotliEncoderCompressStream() needs a bigger output buffer than the one we provided "
"(output buffer %zu bytes, compressed payload %zu bytes)",
state->output.size, size);
return 0;
}
if(compressed_size == 0) {
netdata_log_error("STREAM: BrotliEncoderCompressStream() did not produce any output from the input provided "
"(input buffer %zu bytes)",
size);
return 0;
}
state->sender_locked.total_compressions++;
state->sender_locked.total_uncompressed += size - available_in;
state->sender_locked.total_compressed += compressed_size;
*out = state->output.data;
return compressed_size;
}
void rrdpush_decompressor_init_brotli(struct decompressor_state *state) {
if (!state->initialized) {
state->initialized = true;
state->stream = BrotliDecoderCreateInstance(NULL, NULL, NULL);
simple_ring_buffer_make_room(&state->output, COMPRESSION_MAX_CHUNK);
}
}
void rrdpush_decompressor_destroy_brotli(struct decompressor_state *state) {
if (state->stream) {
BrotliDecoderDestroyInstance(state->stream);
state->stream = NULL;
}
}
size_t rrdpush_decompress_brotli(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.
size_t available_out = state->output.size;
size_t available_in = compressed_size;
const uint8_t *next_in = (const uint8_t *)compressed_data;
uint8_t *next_out = (uint8_t *)state->output.data;
if (BrotliDecoderDecompressStream(state->stream, &available_in, &next_in, &available_out, &next_out, NULL) == BROTLI_DECODER_RESULT_ERROR) {
netdata_log_error("STREAM: Brotli decompression failed.");
return 0;
}
if(available_in != 0) {
netdata_log_error("STREAM: BrotliDecoderDecompressStream() did not use all the input buffer, %zu bytes out of %zu remain",
available_in, compressed_size);
return 0;
}
size_t decompressed_size = state->output.size - available_out;
if(available_out == 0) {
netdata_log_error("STREAM: BrotliDecoderDecompressStream() needs a bigger output buffer than the one we provided "
"(output buffer %zu bytes, compressed payload %zu bytes)",
state->output.size, compressed_size);
return 0;
}
if(decompressed_size == 0) {
netdata_log_error("STREAM: BrotliDecoderDecompressStream() did not produce any output from the input provided "
"(input buffer %zu bytes)",
compressed_size);
return 0;
}
state->output.read_pos = 0;
state->output.write_pos = decompressed_size;
state->total_compressed += compressed_size - available_in;
state->total_uncompressed += decompressed_size;
state->total_compressions++;
return decompressed_size;
}
#endif // ENABLE_BROTLI
|
