Branch data Line data Source code
1 : : /*-------------------------------------------------------------------------
2 : : *
3 : : * astreamer_lz4.c
4 : : *
5 : : * Archive streamers that deal with data compressed using lz4.
6 : : * astreamer_lz4_compressor applies lz4 compression to the input stream,
7 : : * and astreamer_lz4_decompressor does the reverse.
8 : : *
9 : : * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group
10 : : *
11 : : * IDENTIFICATION
12 : : * src/fe_utils/astreamer_lz4.c
13 : : *-------------------------------------------------------------------------
14 : : */
15 : :
16 : : #include "postgres_fe.h"
17 : :
18 : : #include <unistd.h>
19 : :
20 : : #ifdef USE_LZ4
21 : : #include <lz4frame.h>
22 : : #endif
23 : :
24 : : #include "common/logging.h"
25 : : #include "fe_utils/astreamer.h"
26 : :
27 : : #ifdef USE_LZ4
28 : : typedef struct astreamer_lz4_frame
29 : : {
30 : : astreamer base;
31 : :
32 : : LZ4F_compressionContext_t cctx;
33 : : LZ4F_decompressionContext_t dctx;
34 : : LZ4F_preferences_t prefs;
35 : :
36 : : size_t bytes_written;
37 : : bool header_written;
38 : : astreamer_decompression_state state;
39 : : } astreamer_lz4_frame;
40 : :
41 : : static void astreamer_lz4_compressor_content(astreamer *streamer,
42 : : astreamer_member *member,
43 : : const char *data, int len,
44 : : astreamer_archive_context context);
45 : : static void astreamer_lz4_compressor_finalize(astreamer *streamer);
46 : : static void astreamer_lz4_compressor_free(astreamer *streamer);
47 : :
48 : : static const astreamer_ops astreamer_lz4_compressor_ops = {
49 : : .content = astreamer_lz4_compressor_content,
50 : : .finalize = astreamer_lz4_compressor_finalize,
51 : : .free = astreamer_lz4_compressor_free
52 : : };
53 : :
54 : : static void astreamer_lz4_decompressor_content(astreamer *streamer,
55 : : astreamer_member *member,
56 : : const char *data, int len,
57 : : astreamer_archive_context context);
58 : : static void astreamer_lz4_decompressor_finalize(astreamer *streamer);
59 : : static void astreamer_lz4_decompressor_free(astreamer *streamer);
60 : :
61 : : static const astreamer_ops astreamer_lz4_decompressor_ops = {
62 : : .content = astreamer_lz4_decompressor_content,
63 : : .finalize = astreamer_lz4_decompressor_finalize,
64 : : .free = astreamer_lz4_decompressor_free
65 : : };
66 : : #endif
67 : :
68 : : /*
69 : : * Create a new base backup streamer that performs lz4 compression of tar
70 : : * blocks.
71 : : */
72 : : astreamer *
73 : 2 : astreamer_lz4_compressor_new(astreamer *next, pg_compress_specification *compress)
74 : : {
75 : : #ifdef USE_LZ4
76 : : astreamer_lz4_frame *streamer;
77 : : LZ4F_errorCode_t ctxError;
78 : : LZ4F_preferences_t *prefs;
79 : :
80 : : Assert(next != NULL);
81 : :
82 : 2 : streamer = palloc0_object(astreamer_lz4_frame);
83 : 2 : *((const astreamer_ops **) &streamer->base.bbs_ops) =
84 : : &astreamer_lz4_compressor_ops;
85 : :
86 : 2 : streamer->base.bbs_next = next;
87 : 2 : initStringInfo(&streamer->base.bbs_buffer);
88 : 2 : streamer->header_written = false;
89 : :
90 : : /* Initialize stream compression preferences */
91 : 2 : prefs = &streamer->prefs;
92 : 2 : memset(prefs, 0, sizeof(LZ4F_preferences_t));
93 : 2 : prefs->frameInfo.blockSizeID = LZ4F_max256KB;
94 : 2 : prefs->compressionLevel = compress->level;
95 : :
96 : 2 : ctxError = LZ4F_createCompressionContext(&streamer->cctx, LZ4F_VERSION);
97 [ - + ]: 2 : if (LZ4F_isError(ctxError))
98 : 0 : pg_fatal("could not create lz4 compression context: %s",
99 : : LZ4F_getErrorName(ctxError));
100 : :
101 : 2 : return &streamer->base;
102 : : #else
103 : : pg_fatal("this build does not support compression with %s", "LZ4");
104 : : return NULL; /* keep compiler quiet */
105 : : #endif
106 : : }
107 : :
108 : : #ifdef USE_LZ4
109 : : /*
110 : : * Compress the input data to output buffer.
111 : : *
112 : : * Find out the compression bound based on input data length for each
113 : : * invocation to make sure that output buffer has enough capacity to
114 : : * accommodate the compressed data. In case if the output buffer
115 : : * capacity falls short of compression bound then forward the content
116 : : * of output buffer to next streamer and empty the buffer.
117 : : */
118 : : static void
119 : 5458 : astreamer_lz4_compressor_content(astreamer *streamer,
120 : : astreamer_member *member,
121 : : const char *data, int len,
122 : : astreamer_archive_context context)
123 : : {
124 : : astreamer_lz4_frame *mystreamer;
125 : : const uint8 *next_in;
126 : : uint8 *next_out;
127 : : size_t out_bound,
128 : : compressed_size,
129 : : avail_out;
130 : :
131 : 5458 : mystreamer = (astreamer_lz4_frame *) streamer;
132 : 5458 : next_in = (const uint8 *) data;
133 : :
134 : : /* Write header before processing the first input chunk. */
135 [ + + ]: 5458 : if (!mystreamer->header_written)
136 : : {
137 : 2 : compressed_size = LZ4F_compressBegin(mystreamer->cctx,
138 : 2 : (uint8 *) mystreamer->base.bbs_buffer.data,
139 : 2 : mystreamer->base.bbs_buffer.maxlen,
140 : 2 : &mystreamer->prefs);
141 : :
142 [ - + ]: 2 : if (LZ4F_isError(compressed_size))
143 : 0 : pg_fatal("could not write lz4 header: %s",
144 : : LZ4F_getErrorName(compressed_size));
145 : :
146 : 2 : mystreamer->bytes_written += compressed_size;
147 : 2 : mystreamer->header_written = true;
148 : : }
149 : :
150 : : /*
151 : : * Update the offset and capacity of output buffer based on number of
152 : : * bytes written to output buffer.
153 : : */
154 : 5458 : next_out = (uint8 *) mystreamer->base.bbs_buffer.data + mystreamer->bytes_written;
155 : 5458 : avail_out = mystreamer->base.bbs_buffer.maxlen - mystreamer->bytes_written;
156 : :
157 : : /*
158 : : * Find out the compression bound and make sure that output buffer has the
159 : : * required capacity for the success of LZ4F_compressUpdate. If needed
160 : : * forward the content to next streamer and empty the buffer.
161 : : */
162 : 5458 : out_bound = LZ4F_compressBound(len, &mystreamer->prefs);
163 [ + + ]: 5458 : if (avail_out < out_bound)
164 : : {
165 : 33 : astreamer_content(mystreamer->base.bbs_next, member,
166 : 33 : mystreamer->base.bbs_buffer.data,
167 : 33 : mystreamer->bytes_written,
168 : : context);
169 : :
170 : : /* Enlarge buffer if it falls short of out bound. */
171 [ + + ]: 33 : if (mystreamer->base.bbs_buffer.maxlen < out_bound)
172 : 2 : enlargeStringInfo(&mystreamer->base.bbs_buffer, out_bound);
173 : :
174 : 33 : avail_out = mystreamer->base.bbs_buffer.maxlen;
175 : 33 : mystreamer->bytes_written = 0;
176 : 33 : next_out = (uint8 *) mystreamer->base.bbs_buffer.data;
177 : : }
178 : :
179 : : /*
180 : : * This call compresses the data starting at next_in and generates the
181 : : * output starting at next_out. It expects the caller to provide the size
182 : : * of input buffer and capacity of output buffer by providing parameters
183 : : * len and avail_out.
184 : : *
185 : : * It returns the number of bytes compressed to output buffer.
186 : : */
187 : 5458 : compressed_size = LZ4F_compressUpdate(mystreamer->cctx,
188 : : next_out, avail_out,
189 : : next_in, len, NULL);
190 : :
191 [ - + ]: 5458 : if (LZ4F_isError(compressed_size))
192 : 0 : pg_fatal("could not compress data: %s",
193 : : LZ4F_getErrorName(compressed_size));
194 : :
195 : 5458 : mystreamer->bytes_written += compressed_size;
196 : 5458 : }
197 : :
198 : : /*
199 : : * End-of-stream processing.
200 : : */
201 : : static void
202 : 2 : astreamer_lz4_compressor_finalize(astreamer *streamer)
203 : : {
204 : : astreamer_lz4_frame *mystreamer;
205 : : uint8 *next_out;
206 : : size_t footer_bound,
207 : : compressed_size,
208 : : avail_out;
209 : :
210 : 2 : mystreamer = (astreamer_lz4_frame *) streamer;
211 : :
212 : : /* Find out the footer bound and update the output buffer. */
213 : 2 : footer_bound = LZ4F_compressBound(0, &mystreamer->prefs);
214 [ - + ]: 2 : if ((mystreamer->base.bbs_buffer.maxlen - mystreamer->bytes_written) <
215 : : footer_bound)
216 : : {
217 : 0 : astreamer_content(mystreamer->base.bbs_next, NULL,
218 : 0 : mystreamer->base.bbs_buffer.data,
219 : 0 : mystreamer->bytes_written,
220 : : ASTREAMER_UNKNOWN);
221 : :
222 : : /* Enlarge buffer if it falls short of footer bound. */
223 [ # # ]: 0 : if (mystreamer->base.bbs_buffer.maxlen < footer_bound)
224 : 0 : enlargeStringInfo(&mystreamer->base.bbs_buffer, footer_bound);
225 : :
226 : 0 : avail_out = mystreamer->base.bbs_buffer.maxlen;
227 : 0 : mystreamer->bytes_written = 0;
228 : 0 : next_out = (uint8 *) mystreamer->base.bbs_buffer.data;
229 : : }
230 : : else
231 : : {
232 : 2 : next_out = (uint8 *) mystreamer->base.bbs_buffer.data + mystreamer->bytes_written;
233 : 2 : avail_out = mystreamer->base.bbs_buffer.maxlen - mystreamer->bytes_written;
234 : : }
235 : :
236 : : /*
237 : : * Finalize the frame and flush whatever data remaining in compression
238 : : * context.
239 : : */
240 : 2 : compressed_size = LZ4F_compressEnd(mystreamer->cctx,
241 : : next_out, avail_out, NULL);
242 : :
243 [ - + ]: 2 : if (LZ4F_isError(compressed_size))
244 : 0 : pg_fatal("could not end lz4 compression: %s",
245 : : LZ4F_getErrorName(compressed_size));
246 : :
247 : 2 : mystreamer->bytes_written += compressed_size;
248 : :
249 : 2 : astreamer_content(mystreamer->base.bbs_next, NULL,
250 : 2 : mystreamer->base.bbs_buffer.data,
251 : 2 : mystreamer->bytes_written,
252 : : ASTREAMER_UNKNOWN);
253 : :
254 : 2 : astreamer_finalize(mystreamer->base.bbs_next);
255 : 2 : }
256 : :
257 : : /*
258 : : * Free memory.
259 : : */
260 : : static void
261 : 2 : astreamer_lz4_compressor_free(astreamer *streamer)
262 : : {
263 : : astreamer_lz4_frame *mystreamer;
264 : :
265 : 2 : mystreamer = (astreamer_lz4_frame *) streamer;
266 : 2 : astreamer_free(streamer->bbs_next);
267 : 2 : LZ4F_freeCompressionContext(mystreamer->cctx);
268 : 2 : pfree(streamer->bbs_buffer.data);
269 : 2 : pfree(streamer);
270 : 2 : }
271 : : #endif
272 : :
273 : : /*
274 : : * Create a new base backup streamer that performs decompression of lz4
275 : : * compressed blocks.
276 : : */
277 : : astreamer *
278 : 13 : astreamer_lz4_decompressor_new(astreamer *next)
279 : : {
280 : : #ifdef USE_LZ4
281 : : astreamer_lz4_frame *streamer;
282 : : LZ4F_errorCode_t ctxError;
283 : :
284 : : Assert(next != NULL);
285 : :
286 : 13 : streamer = palloc0_object(astreamer_lz4_frame);
287 : 13 : *((const astreamer_ops **) &streamer->base.bbs_ops) =
288 : : &astreamer_lz4_decompressor_ops;
289 : :
290 : 13 : streamer->base.bbs_next = next;
291 : 13 : initStringInfo(&streamer->base.bbs_buffer);
292 : : /* Use a buffer size comparable to the compressor's */
293 : 13 : enlargeStringInfo(&streamer->base.bbs_buffer, 256 * 1024 - 1);
294 : :
295 : : /* Initialize internal stream state for decompression */
296 : 13 : ctxError = LZ4F_createDecompressionContext(&streamer->dctx, LZ4F_VERSION);
297 [ - + ]: 13 : if (LZ4F_isError(ctxError))
298 : 0 : pg_fatal("could not initialize compression library: %s",
299 : : LZ4F_getErrorName(ctxError));
300 : :
301 : 13 : streamer->state = ASTREAMER_STREAM_NEW;
302 : 13 : return &streamer->base;
303 : : #else
304 : : pg_fatal("this build does not support compression with %s", "LZ4");
305 : : return NULL; /* keep compiler quiet */
306 : : #endif
307 : : }
308 : :
309 : : #ifdef USE_LZ4
310 : : /*
311 : : * Decompress the input data to output buffer until we run out of input
312 : : * data. Each time the output buffer is full, pass on the decompressed data
313 : : * to the next streamer.
314 : : */
315 : : static void
316 : 467 : astreamer_lz4_decompressor_content(astreamer *streamer,
317 : : astreamer_member *member,
318 : : const char *data, int len,
319 : : astreamer_archive_context context)
320 : : {
321 : : astreamer_lz4_frame *mystreamer;
322 : : const uint8 *next_in;
323 : : uint8 *next_out;
324 : : size_t avail_in,
325 : : avail_out;
326 : :
327 : 467 : mystreamer = (astreamer_lz4_frame *) streamer;
328 : 467 : next_in = (const uint8 *) data;
329 : 467 : next_out = (uint8 *) mystreamer->base.bbs_buffer.data + mystreamer->bytes_written;
330 : 467 : avail_in = len;
331 : 467 : avail_out = mystreamer->base.bbs_buffer.maxlen - mystreamer->bytes_written;
332 : :
333 [ + + ]: 2571 : while (avail_in > 0)
334 : : {
335 : : size_t ret,
336 : : read_size,
337 : : out_size;
338 : :
339 : 2104 : read_size = avail_in;
340 : 2104 : out_size = avail_out;
341 : :
342 : : /*
343 : : * This call decompresses the data starting at next_in and generates
344 : : * the output data starting at next_out. It expects the caller to
345 : : * provide size of the input buffer and total capacity of the output
346 : : * buffer by providing the read_size and out_size parameters
347 : : * respectively.
348 : : *
349 : : * Per the documentation of LZ4, parameters read_size and out_size
350 : : * behaves as dual parameters. On return, the number of bytes consumed
351 : : * from the input buffer will be written back to read_size and the
352 : : * number of bytes decompressed to output buffer will be written back
353 : : * to out_size respectively.
354 : : */
355 : 2104 : ret = LZ4F_decompress(mystreamer->dctx,
356 : : next_out, &out_size,
357 : : next_in, &read_size, NULL);
358 : :
359 [ - + ]: 2104 : if (LZ4F_isError(ret))
360 : 0 : pg_fatal("could not decompress data: %s",
361 : : LZ4F_getErrorName(ret));
362 : :
363 : : /* The frame is only done when LZ4F_decompress returns 0 */
364 [ + + ]: 2104 : if (ret)
365 : 2093 : mystreamer->state = ASTREAMER_FRAME_INCOMPLETE;
366 : : else
367 : 11 : mystreamer->state = ASTREAMER_FRAME_COMPLETE;
368 : :
369 : : /* Update input buffer based on number of bytes consumed */
370 : 2104 : avail_in -= read_size;
371 : 2104 : next_in += read_size;
372 : :
373 : : /* Update output buffer based on number of bytes produced */
374 : 2104 : avail_out -= out_size;
375 : 2104 : next_out += out_size;
376 : 2104 : mystreamer->bytes_written += out_size;
377 : :
378 : : /*
379 : : * If output buffer is full then forward the content to next streamer
380 : : * and reset the output buffer.
381 : : */
382 [ + + ]: 2104 : if (mystreamer->bytes_written >= mystreamer->base.bbs_buffer.maxlen)
383 : : {
384 : 1736 : astreamer_content(mystreamer->base.bbs_next, member,
385 : 1736 : mystreamer->base.bbs_buffer.data,
386 : : mystreamer->base.bbs_buffer.maxlen,
387 : : context);
388 : :
389 : 1736 : avail_out = mystreamer->base.bbs_buffer.maxlen;
390 : 1736 : next_out = (uint8 *) mystreamer->base.bbs_buffer.data;
391 : 1736 : mystreamer->bytes_written = 0;
392 : : }
393 : : }
394 : 467 : }
395 : :
396 : : /*
397 : : * End-of-stream processing.
398 : : */
399 : : static void
400 : 9 : astreamer_lz4_decompressor_finalize(astreamer *streamer)
401 : : {
402 : : astreamer_lz4_frame *mystreamer;
403 : :
404 : 9 : mystreamer = (astreamer_lz4_frame *) streamer;
405 : :
406 [ - + ]: 9 : if (unlikely(mystreamer->state == ASTREAMER_STREAM_NEW))
407 : 0 : pg_fatal("could not decompress data: compressed stream is empty");
408 [ + + ]: 9 : else if (mystreamer->state != ASTREAMER_FRAME_COMPLETE)
409 : 2 : pg_fatal("could not decompress data: compressed stream is incomplete");
410 : :
411 : : /*
412 : : * End of the stream, if there is some pending data in output buffers then
413 : : * we must forward it to next streamer.
414 : : */
415 [ + - ]: 7 : if (mystreamer->bytes_written > 0)
416 : 7 : astreamer_content(mystreamer->base.bbs_next, NULL,
417 : 7 : mystreamer->base.bbs_buffer.data,
418 : 7 : mystreamer->bytes_written,
419 : : ASTREAMER_UNKNOWN);
420 : :
421 : 7 : astreamer_finalize(mystreamer->base.bbs_next);
422 : 7 : }
423 : :
424 : : /*
425 : : * Free memory.
426 : : */
427 : : static void
428 : 11 : astreamer_lz4_decompressor_free(astreamer *streamer)
429 : : {
430 : : astreamer_lz4_frame *mystreamer;
431 : :
432 : 11 : mystreamer = (astreamer_lz4_frame *) streamer;
433 : 11 : astreamer_free(streamer->bbs_next);
434 : 11 : LZ4F_freeDecompressionContext(mystreamer->dctx);
435 : 11 : pfree(streamer->bbs_buffer.data);
436 : 11 : pfree(streamer);
437 : 11 : }
438 : : #endif
|
