@cryptotaxi247 / netdata
Code Commits Issues Pulls Sessions CI/CD
master
master
c 164 lines 5.55 KB
Raw
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 #include "gzip.h"
4 #include <zlib.h>
5
6 void stream_compressor_init_gzip(struct compressor_state *state) {
7 if (!state->initialized) {
8 state->initialized = true;
9
10 // Initialize deflate stream
11 z_stream *strm = state->stream = (z_stream *) mallocz(sizeof(z_stream));
12 strm->zalloc = Z_NULL;
13 strm->zfree = Z_NULL;
14 strm->opaque = Z_NULL;
15
16 if(state->level < Z_BEST_SPEED)
17 state->level = Z_BEST_SPEED;
18
19 if(state->level > Z_BEST_COMPRESSION)
20 state->level = Z_BEST_COMPRESSION;
21
22 // int r = deflateInit2(strm, Z_BEST_COMPRESSION, Z_DEFLATED, 15 + 16, 8, Z_DEFAULT_STRATEGY);
23 int r = deflateInit2(strm, state->level, Z_DEFLATED, 15 + 16, 8, Z_DEFAULT_STRATEGY);
24 if (r != Z_OK) {
25 netdata_log_error("STREAM_COMPRESS: Failed to initialize deflate with error: %d", r);
26 freez(state->stream);
27 state->initialized = false;
28 return;
29 }
30
31 }
32 }
33
34 void stream_compressor_destroy_gzip(struct compressor_state *state) {
35 if (state->stream) {
36 deflateEnd(state->stream);
37 freez(state->stream);
38 state->stream = NULL;
39 }
40 }
41
42 size_t stream_compress_gzip(struct compressor_state *state, const char *data, size_t size, const char **out) {
43 if (unlikely(!state || !size || !out))
44 return 0;
45
46 simple_ring_buffer_make_room(&state->output, deflateBound(state->stream, size));
47
48 z_stream *strm = state->stream;
49 strm->avail_in = (uInt)size;
50 strm->next_in = (Bytef *)data;
51 strm->avail_out = (uInt)state->output.size;
52 strm->next_out = (Bytef *)state->output.data;
53
54 int ret = deflate(strm, Z_SYNC_FLUSH);
55 if (ret != Z_OK && ret != Z_STREAM_END) {
56 netdata_log_error("STREAM_COMPRESS: deflate() failed with error %d", ret);
57 return 0;
58 }
59
60 if(strm->avail_in != 0) {
61 netdata_log_error("STREAM_COMPRESS: deflate() did not use all the input buffer, %u bytes out of %zu remain",
62 strm->avail_in, size);
63 return 0;
64 }
65
66 if(strm->avail_out == 0) {
67 netdata_log_error("STREAM_COMPRESS: deflate() needs a bigger output buffer than the one we provided "
68 "(output buffer %zu bytes, compressed payload %zu bytes)",
69 state->output.size, size);
70 return 0;
71 }
72
73 size_t compressed_data_size = state->output.size - strm->avail_out;
74
75 if(compressed_data_size == 0) {
76 netdata_log_error("STREAM_COMPRESS: deflate() did not produce any output "
77 "(output buffer %zu bytes, compressed payload %zu bytes)",
78 state->output.size, size);
79 return 0;
80 }
81
82 state->sender_locked.total_compressions++;
83 state->sender_locked.total_uncompressed += size;
84 state->sender_locked.total_compressed += compressed_data_size;
85
86 *out = state->output.data;
87 return compressed_data_size;
88 }
89
90 void stream_decompressor_init_gzip(struct decompressor_state *state) {
91 if (!state->initialized) {
92 state->initialized = true;
93
94 // Initialize inflate stream
95 z_stream *strm = state->stream = (z_stream *)mallocz(sizeof(z_stream));
96 strm->zalloc = Z_NULL;
97 strm->zfree = Z_NULL;
98 strm->opaque = Z_NULL;
99
100 int r = inflateInit2(strm, 15 + 16);
101 if (r != Z_OK) {
102 netdata_log_error("STREAM_DECOMPRESS: Failed to initialize inflateInit2() with error: %d", r);
103 freez(state->stream);
104 state->initialized = false;
105 return;
106 }
107
108 simple_ring_buffer_make_room(&state->output, COMPRESSION_MAX_CHUNK);
109 }
110 }
111
112 void stream_decompressor_destroy_gzip(struct decompressor_state *state) {
113 if (state->stream) {
114 inflateEnd(state->stream);
115 freez(state->stream);
116 state->stream = NULL;
117 }
118 }
119
120 size_t stream_decompress_gzip(struct decompressor_state *state, const char *compressed_data, size_t compressed_size) {
121 if (unlikely(!state || !compressed_data || !compressed_size))
122 return 0;
123
124 // The state.output ring buffer is always EMPTY at this point,
125 // meaning that (state->output.read_pos == state->output.write_pos)
126 // However, THEY ARE NOT ZERO.
127
128 z_stream *strm = state->stream;
129 strm->avail_in = (uInt)compressed_size;
130 strm->next_in = (Bytef *)compressed_data;
131 strm->avail_out = (uInt)state->output.size;
132 strm->next_out = (Bytef *)state->output.data;
133
134 int ret = inflate(strm, Z_SYNC_FLUSH);
135 if (ret != Z_STREAM_END && ret != Z_OK) {
136 netdata_log_error("STREAM_DECOMPRESS: inflate() failed with error %d", ret);
137 return 0;
138 }
139
140 if(strm->avail_in != 0) {
141 netdata_log_error("STREAM_DECOMPRESS: inflate() did not use all compressed data we provided "
142 "(compressed payload %zu bytes, remaining to be uncompressed %u)"
143 , compressed_size, strm->avail_in);
144 return 0;
145 }
146
147 if(strm->avail_out == 0) {
148 netdata_log_error("STREAM_DECOMPRESS: inflate() needs a bigger output buffer than the one we provided "
149 "(compressed payload %zu bytes, output buffer size %zu bytes)"
150 , compressed_size, state->output.size);
151 return 0;
152 }
153
154 size_t decompressed_size = state->output.size - strm->avail_out;
155
156 state->output.read_pos = 0;
157 state->output.write_pos = decompressed_size;
158
159 state->total_compressed += compressed_size;
160 state->total_uncompressed += decompressed_size;
161 state->total_compressions++;
162
163 return decompressed_size;
164 }