Age Owner Branch data TLA Line data Source code
1 : : /*-------------------------------------------------------------------------
2 : : *
3 : : * compress_lz4.c
4 : : * Routines for archivers to write a LZ4 compressed data stream.
5 : : *
6 : : * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group
7 : : * Portions Copyright (c) 1994, Regents of the University of California
8 : : *
9 : : * IDENTIFICATION
10 : : * src/bin/pg_dump/compress_lz4.c
11 : : *
12 : : *-------------------------------------------------------------------------
13 : : */
14 : : #include "postgres_fe.h"
15 : : #include <unistd.h>
16 : :
17 : : #include "compress_lz4.h"
18 : : #include "pg_backup_utils.h"
19 : :
20 : : #ifdef USE_LZ4
21 : : #include <lz4frame.h>
22 : :
23 : : /*
24 : : * LZ4F_HEADER_SIZE_MAX first appeared in v1.7.5 of the library.
25 : : * Redefine it for installations with a lesser version.
26 : : */
27 : : #ifndef LZ4F_HEADER_SIZE_MAX
28 : : #define LZ4F_HEADER_SIZE_MAX 32
29 : : #endif
30 : :
31 : : /*---------------------------------
32 : : * Common to both compression APIs
33 : : *---------------------------------
34 : : */
35 : :
36 : : /*
37 : : * (de)compression state used by both the Compressor and Stream APIs.
38 : : */
39 : : typedef struct LZ4State
40 : : {
41 : : /*
42 : : * Used by the Stream API to keep track of the file stream.
43 : : */
44 : : FILE *fp;
45 : :
46 : : LZ4F_preferences_t prefs;
47 : :
48 : : LZ4F_compressionContext_t ctx;
49 : : LZ4F_decompressionContext_t dtx;
50 : :
51 : : /*
52 : : * Used by the Stream API's lazy initialization.
53 : : */
54 : : bool inited;
55 : :
56 : : /*
57 : : * Used by the Stream API to distinguish between compression and
58 : : * decompression operations.
59 : : */
60 : : bool compressing;
61 : : bool frame_finished;
62 : :
63 : : /*
64 : : * I/O buffer area.
65 : : */
66 : : char *buffer; /* buffer for compressed data */
67 : : size_t buflen; /* allocated size of buffer */
68 : : size_t bufdata; /* amount of valid data currently in buffer */
69 : : /* These fields are used only while decompressing: */
70 : : size_t bufnext; /* next buffer position to decompress */
71 : : char *outbuf; /* buffer for decompressed data */
72 : : size_t outbuflen; /* allocated size of outbuf */
73 : : size_t outbufdata; /* amount of valid data currently in outbuf */
74 : : size_t outbufnext; /* next outbuf position to return */
75 : :
76 : : /*
77 : : * Used by both APIs to keep track of error codes.
78 : : */
79 : : size_t errcode;
80 : : } LZ4State;
81 : :
82 : : /*
83 : : * LZ4State_compression_init
84 : : * Initialize the required LZ4State members for compression.
85 : : *
86 : : * Write the LZ4 frame header in a buffer keeping track of its length. Users of
87 : : * this function can choose when and how to write the header to a file stream.
88 : : *
89 : : * Returns true on success. In case of a failure returns false, and stores the
90 : : * error code in state->errcode.
91 : : */
92 : : static bool
1290 tomas.vondra@postgre 93 :CBC 7 : LZ4State_compression_init(LZ4State *state)
94 : : {
95 : : size_t status;
96 : :
97 : : /*
98 : : * Compute size needed for buffer, assuming we will present at most
99 : : * DEFAULT_IO_BUFFER_SIZE input bytes at a time.
100 : : */
101 : 7 : state->buflen = LZ4F_compressBound(DEFAULT_IO_BUFFER_SIZE, &state->prefs);
102 : :
103 : : /*
104 : : * Add some slop to ensure we're not forced to flush every time.
105 : : *
106 : : * The present slop factor of 50% is chosen so that the typical output
107 : : * block size is about 128K when DEFAULT_IO_BUFFER_SIZE = 128K. We might
108 : : * need a different slop factor to maintain that equivalence if
109 : : * DEFAULT_IO_BUFFER_SIZE is changed dramatically.
110 : : */
361 tgl@sss.pgh.pa.us 111 : 7 : state->buflen += state->buflen / 2;
112 : :
113 : : /*
114 : : * LZ4F_compressBegin requires a buffer that is greater or equal to
115 : : * LZ4F_HEADER_SIZE_MAX. Verify that the requirement is met.
116 : : */
1290 tomas.vondra@postgre 117 [ - + ]: 7 : if (state->buflen < LZ4F_HEADER_SIZE_MAX)
1290 tomas.vondra@postgre 118 :UBC 0 : state->buflen = LZ4F_HEADER_SIZE_MAX;
119 : :
1290 tomas.vondra@postgre 120 :CBC 7 : status = LZ4F_createCompressionContext(&state->ctx, LZ4F_VERSION);
121 [ - + ]: 7 : if (LZ4F_isError(status))
122 : : {
1290 tomas.vondra@postgre 123 :UBC 0 : state->errcode = status;
124 : 0 : return false;
125 : : }
126 : :
1290 tomas.vondra@postgre 127 :CBC 7 : state->buffer = pg_malloc(state->buflen);
128 : :
129 : : /*
130 : : * Insert LZ4 header into buffer.
131 : : */
132 : 7 : status = LZ4F_compressBegin(state->ctx,
133 : 7 : state->buffer, state->buflen,
134 : 7 : &state->prefs);
135 [ - + ]: 7 : if (LZ4F_isError(status))
136 : : {
1290 tomas.vondra@postgre 137 :UBC 0 : state->errcode = status;
138 : 0 : return false;
139 : : }
140 : :
364 tgl@sss.pgh.pa.us 141 :CBC 7 : state->bufdata = status;
142 : :
1290 tomas.vondra@postgre 143 : 7 : return true;
144 : : }
145 : :
146 : : /*----------------------
147 : : * Compressor API
148 : : *----------------------
149 : : */
150 : :
151 : : /* Private routines that support LZ4 compressed data I/O */
152 : :
153 : : static void
1327 154 : 4 : ReadDataFromArchiveLZ4(ArchiveHandle *AH, CompressorState *cs)
155 : : {
156 : : size_t r;
157 : : size_t readbuflen;
158 : : char *outbuf;
159 : : char *readbuf;
1290 160 : 4 : LZ4F_decompressionContext_t ctx = NULL;
161 : : LZ4F_decompressOptions_t dec_opt;
162 : : LZ4F_errorCode_t status;
163 : :
164 : : /*
165 : : * cs->private_data is an LZ4State for compression, whereas this function
166 : : * uses a short-lived decompression context. Keep its state local.
167 : : */
3 dgustafsson@postgres 168 : 4 : bool dec_done = false;
169 : :
1290 tomas.vondra@postgre 170 : 4 : memset(&dec_opt, 0, sizeof(dec_opt));
171 : 4 : status = LZ4F_createDecompressionContext(&ctx, LZ4F_VERSION);
172 [ - + ]: 4 : if (LZ4F_isError(status))
1290 tomas.vondra@postgre 173 :UBC 0 : pg_fatal("could not create LZ4 decompression context: %s",
174 : : LZ4F_getErrorName(status));
175 : :
364 tgl@sss.pgh.pa.us 176 :CBC 4 : outbuf = pg_malloc(DEFAULT_IO_BUFFER_SIZE);
177 : 4 : readbuf = pg_malloc(DEFAULT_IO_BUFFER_SIZE);
1290 tomas.vondra@postgre 178 : 4 : readbuflen = DEFAULT_IO_BUFFER_SIZE;
179 [ + + ]: 9 : while ((r = cs->readF(AH, &readbuf, &readbuflen)) > 0)
180 : : {
181 : : char *readp;
182 : : char *readend;
183 : :
184 : : /* Process one chunk */
185 : 5 : readp = readbuf;
186 : 5 : readend = readbuf + r;
187 [ + + ]: 11 : while (readp < readend)
188 : : {
189 : 6 : size_t out_size = DEFAULT_IO_BUFFER_SIZE;
190 : 6 : size_t read_size = readend - readp;
191 : :
192 : 6 : status = LZ4F_decompress(ctx, outbuf, &out_size,
193 : : readp, &read_size, &dec_opt);
194 [ - + ]: 6 : if (LZ4F_isError(status))
1290 tomas.vondra@postgre 195 :UBC 0 : pg_fatal("could not decompress: %s",
196 : : LZ4F_getErrorName(status));
3 dgustafsson@postgres 197 :CBC 6 : dec_done = (status == 0);
198 : :
1290 tomas.vondra@postgre 199 : 6 : ahwrite(outbuf, 1, out_size, AH);
200 : 6 : readp += read_size;
201 : : }
202 : : }
203 : :
3 dgustafsson@postgres 204 [ + + ]: 4 : if (!dec_done)
205 : 1 : pg_fatal("could not decompress data: compressed stream is incomplete");
206 : :
1290 tomas.vondra@postgre 207 : 3 : pg_free(outbuf);
208 : 3 : pg_free(readbuf);
209 : :
210 : 3 : status = LZ4F_freeDecompressionContext(ctx);
211 [ - + ]: 3 : if (LZ4F_isError(status))
1290 tomas.vondra@postgre 212 :UBC 0 : pg_fatal("could not free LZ4 decompression context: %s",
213 : : LZ4F_getErrorName(status));
1327 tomas.vondra@postgre 214 :CBC 3 : }
215 : :
216 : : static void
217 : 4 : WriteDataToArchiveLZ4(ArchiveHandle *AH, CompressorState *cs,
218 : : const void *data, size_t dLen)
219 : : {
1290 220 : 4 : LZ4State *state = (LZ4State *) cs->private_data;
221 : 4 : size_t remaining = dLen;
222 : :
223 [ + + ]: 10 : while (remaining > 0)
224 : : {
225 : : size_t chunk;
226 : : size_t required;
227 : : size_t status;
228 : :
229 : : /* We don't try to present more than DEFAULT_IO_BUFFER_SIZE bytes */
364 tgl@sss.pgh.pa.us 230 : 6 : chunk = Min(remaining, (size_t) DEFAULT_IO_BUFFER_SIZE);
231 : :
232 : : /* If not enough space, must flush buffer */
233 : 6 : required = LZ4F_compressBound(chunk, &state->prefs);
234 [ + + ]: 6 : if (required > state->buflen - state->bufdata)
235 : : {
236 : 1 : cs->writeF(AH, state->buffer, state->bufdata);
237 : 1 : state->bufdata = 0;
238 : : }
239 : :
1290 tomas.vondra@postgre 240 : 6 : status = LZ4F_compressUpdate(state->ctx,
364 tgl@sss.pgh.pa.us 241 : 6 : state->buffer + state->bufdata,
242 : 6 : state->buflen - state->bufdata,
243 : : data, chunk, NULL);
244 : :
1290 tomas.vondra@postgre 245 [ - + ]: 6 : if (LZ4F_isError(status))
1242 peter@eisentraut.org 246 :UBC 0 : pg_fatal("could not compress data: %s",
247 : : LZ4F_getErrorName(status));
248 : :
364 tgl@sss.pgh.pa.us 249 :CBC 6 : state->bufdata += status;
250 : :
251 : 6 : data = ((const char *) data) + chunk;
252 : 6 : remaining -= chunk;
253 : : }
1327 tomas.vondra@postgre 254 : 4 : }
255 : :
256 : : static void
257 : 6 : EndCompressorLZ4(ArchiveHandle *AH, CompressorState *cs)
258 : : {
1290 259 : 6 : LZ4State *state = (LZ4State *) cs->private_data;
260 : : size_t required;
261 : : size_t status;
262 : :
263 : : /* Nothing needs to be done */
264 [ + + ]: 6 : if (!state)
265 : 3 : return;
266 : :
267 : : /* We might need to flush the buffer to make room for LZ4F_compressEnd */
364 tgl@sss.pgh.pa.us 268 : 3 : required = LZ4F_compressBound(0, &state->prefs);
269 [ - + ]: 3 : if (required > state->buflen - state->bufdata)
270 : : {
364 tgl@sss.pgh.pa.us 271 :UBC 0 : cs->writeF(AH, state->buffer, state->bufdata);
272 : 0 : state->bufdata = 0;
273 : : }
274 : :
1290 tomas.vondra@postgre 275 :CBC 3 : status = LZ4F_compressEnd(state->ctx,
364 tgl@sss.pgh.pa.us 276 : 3 : state->buffer + state->bufdata,
277 : 3 : state->buflen - state->bufdata,
278 : : NULL);
1290 tomas.vondra@postgre 279 [ - + ]: 3 : if (LZ4F_isError(status))
1242 peter@eisentraut.org 280 :UBC 0 : pg_fatal("could not end compression: %s",
281 : : LZ4F_getErrorName(status));
364 tgl@sss.pgh.pa.us 282 :CBC 3 : state->bufdata += status;
283 : :
284 : : /* Write the final bufferload */
285 : 3 : cs->writeF(AH, state->buffer, state->bufdata);
286 : :
1290 tomas.vondra@postgre 287 : 3 : status = LZ4F_freeCompressionContext(state->ctx);
288 [ - + ]: 3 : if (LZ4F_isError(status))
1242 peter@eisentraut.org 289 :UBC 0 : pg_fatal("could not end compression: %s",
290 : : LZ4F_getErrorName(status));
291 : :
1290 tomas.vondra@postgre 292 :CBC 3 : pg_free(state->buffer);
293 : 3 : pg_free(state);
294 : :
295 : 3 : cs->private_data = NULL;
296 : : }
297 : :
298 : : /*
299 : : * Public routines that support LZ4 compressed data I/O
300 : : */
301 : : void
1327 302 : 7 : InitCompressorLZ4(CompressorState *cs, const pg_compress_specification compression_spec)
303 : : {
304 : : LZ4State *state;
305 : :
306 : 7 : cs->readData = ReadDataFromArchiveLZ4;
307 : 7 : cs->writeData = WriteDataToArchiveLZ4;
308 : 7 : cs->end = EndCompressorLZ4;
309 : :
310 : 7 : cs->compression_spec = compression_spec;
311 : :
312 : : /*
313 : : * Read operations have access to the whole input. No state needs to be
314 : : * carried between calls.
315 : : */
1290 316 [ + + ]: 7 : if (cs->readF)
317 : 4 : return;
318 : :
241 michael@paquier.xyz 319 : 3 : state = pg_malloc0_object(LZ4State);
1290 tomas.vondra@postgre 320 [ + - ]: 3 : if (cs->compression_spec.level >= 0)
321 : 3 : state->prefs.compressionLevel = cs->compression_spec.level;
322 : :
323 [ - + ]: 3 : if (!LZ4State_compression_init(state))
1290 tomas.vondra@postgre 324 :UBC 0 : pg_fatal("could not initialize LZ4 compression: %s",
325 : : LZ4F_getErrorName(state->errcode));
326 : :
1290 tomas.vondra@postgre 327 :CBC 3 : cs->private_data = state;
328 : : }
329 : :
330 : : /*----------------------
331 : : * Compress Stream API
332 : : *----------------------
333 : : */
334 : :
335 : :
336 : : /*
337 : : * LZ4 equivalent to feof() or gzeof(). Return true iff there is no
338 : : * more buffered data and the end of the input file has been reached.
339 : : */
340 : : static bool
1290 tomas.vondra@postgre 341 :UBC 0 : LZ4Stream_eof(CompressFileHandle *CFH)
342 : : {
343 : 0 : LZ4State *state = (LZ4State *) CFH->private_data;
344 : :
364 tgl@sss.pgh.pa.us 345 : 0 : return state->outbufnext >= state->outbufdata &&
346 [ # # # # : 0 : state->bufnext >= state->bufdata &&
# # ]
347 : 0 : feof(state->fp);
348 : : }
349 : :
350 : : static const char *
1290 tomas.vondra@postgre 351 :CBC 1 : LZ4Stream_get_error(CompressFileHandle *CFH)
352 : : {
353 : 1 : LZ4State *state = (LZ4State *) CFH->private_data;
354 : : const char *errmsg;
355 : :
356 [ - + ]: 1 : if (LZ4F_isError(state->errcode))
1290 tomas.vondra@postgre 357 :UBC 0 : errmsg = LZ4F_getErrorName(state->errcode);
358 : : else
1327 tomas.vondra@postgre 359 :CBC 1 : errmsg = strerror(errno);
360 : :
361 : 1 : return errmsg;
362 : : }
363 : :
364 : : /*
365 : : * Initialize an already alloc'ed LZ4State struct for subsequent calls.
366 : : *
367 : : * Creates the necessary contexts for either compression or decompression. When
368 : : * compressing data (indicated by compressing=true), it additionally writes the
369 : : * LZ4 header in the output buffer.
370 : : *
371 : : * It's expected that a not-yet-initialized LZ4State will be zero-filled.
372 : : *
373 : : * Returns true on success. In case of a failure returns false, and stores the
374 : : * error code in state->errcode.
375 : : */
376 : : static bool
364 tgl@sss.pgh.pa.us 377 : 94 : LZ4Stream_init(LZ4State *state, bool compressing)
378 : : {
379 : : size_t status;
380 : :
1290 tomas.vondra@postgre 381 [ + + ]: 94 : if (state->inited)
1299 382 : 89 : return true;
383 : :
1290 384 : 5 : state->compressing = compressing;
385 : :
386 [ + + ]: 5 : if (state->compressing)
387 : : {
388 [ - + ]: 4 : if (!LZ4State_compression_init(state))
1299 tomas.vondra@postgre 389 :UBC 0 : return false;
390 : : }
391 : : else
392 : : {
1290 tomas.vondra@postgre 393 :CBC 1 : status = LZ4F_createDecompressionContext(&state->dtx, LZ4F_VERSION);
1327 394 [ - + ]: 1 : if (LZ4F_isError(status))
395 : : {
1290 tomas.vondra@postgre 396 :UBC 0 : state->errcode = status;
1299 397 : 0 : return false;
398 : : }
399 : :
364 tgl@sss.pgh.pa.us 400 :CBC 1 : state->buflen = DEFAULT_IO_BUFFER_SIZE;
1290 tomas.vondra@postgre 401 : 1 : state->buffer = pg_malloc(state->buflen);
364 tgl@sss.pgh.pa.us 402 : 1 : state->outbuflen = DEFAULT_IO_BUFFER_SIZE;
403 : 1 : state->outbuf = pg_malloc(state->outbuflen);
404 : : }
405 : :
409 dgustafsson@postgres 406 : 5 : state->inited = true;
1299 tomas.vondra@postgre 407 : 5 : return true;
408 : : }
409 : :
410 : : /*
411 : : * The workhorse for reading decompressed content out of an LZ4 compressed
412 : : * stream.
413 : : *
414 : : * It will read up to 'ptrsize' decompressed content, or up to the new line
415 : : * char if one is found first when the eol_flag is set.
416 : : *
417 : : * Returns the number of bytes of decompressed data copied into the ptr
418 : : * buffer, or -1 in case of error.
419 : : */
420 : : static int
1290 421 : 1 : LZ4Stream_read_internal(LZ4State *state, void *ptr, int ptrsize, bool eol_flag)
422 : : {
1299 423 : 1 : int dsize = 0;
364 tgl@sss.pgh.pa.us 424 : 1 : int remaining = ptrsize;
425 : :
426 : : /* Lazy init */
427 [ - + ]: 1 : if (!LZ4Stream_init(state, false /* decompressing */ ))
428 : : {
409 dgustafsson@postgres 429 :UBC 0 : pg_log_error("unable to initialize LZ4 library: %s",
430 : : LZ4F_getErrorName(state->errcode));
1327 tomas.vondra@postgre 431 : 0 : return -1;
432 : : }
433 : :
434 : : /* Loop until postcondition is satisfied */
364 tgl@sss.pgh.pa.us 435 [ + - ]:CBC 2 : while (remaining > 0)
436 : : {
437 : : /*
438 : : * If we already have some decompressed data, return that.
439 : : */
440 [ + + ]: 2 : if (state->outbufnext < state->outbufdata)
441 : : {
442 : 1 : char *outptr = state->outbuf + state->outbufnext;
443 : 1 : size_t readlen = state->outbufdata - state->outbufnext;
444 : 1 : bool eol_found = false;
445 : :
446 [ - + ]: 1 : if (readlen > remaining)
364 tgl@sss.pgh.pa.us 447 :UBC 0 : readlen = remaining;
448 : : /* If eol_flag is set, don't read beyond a newline */
364 tgl@sss.pgh.pa.us 449 [ - + ]:CBC 1 : if (eol_flag)
450 : : {
364 tgl@sss.pgh.pa.us 451 :UBC 0 : char *eolptr = memchr(outptr, '\n', readlen);
452 : :
453 [ # # ]: 0 : if (eolptr)
454 : : {
455 : 0 : readlen = eolptr - outptr + 1;
456 : 0 : eol_found = true;
457 : : }
458 : : }
364 tgl@sss.pgh.pa.us 459 :CBC 1 : memcpy(ptr, outptr, readlen);
460 : 1 : ptr = ((char *) ptr) + readlen;
461 : 1 : state->outbufnext += readlen;
462 : 1 : dsize += readlen;
463 : 1 : remaining -= readlen;
464 [ + - + - ]: 1 : if (eol_found || remaining == 0)
465 : : break;
466 : : /* We must have emptied outbuf */
467 [ - + ]: 1 : Assert(state->outbufnext >= state->outbufdata);
468 : : }
469 : :
470 : : /*
471 : : * If we don't have any pending compressed data, load more into
472 : : * state->buffer.
473 : : */
474 [ + - ]: 2 : if (state->bufnext >= state->bufdata)
475 : : {
476 : : size_t rsize;
477 : :
478 : 2 : rsize = fread(state->buffer, 1, state->buflen, state->fp);
479 [ + - - + ]: 2 : if (rsize < state->buflen && !feof(state->fp))
480 : : {
364 tgl@sss.pgh.pa.us 481 :UBC 0 : pg_log_error("could not read from input file: %m");
482 : 0 : return -1;
483 : : }
484 : :
364 tgl@sss.pgh.pa.us 485 [ + + ]:CBC 2 : if (rsize == 0)
486 : : {
3 dgustafsson@postgres 487 [ + - ]: 1 : if (!state->frame_finished)
488 : : {
489 : 1 : errno = EIO;
490 : 1 : pg_log_error("could not read from input file: %s",
491 : : strerror(errno));
492 : 1 : return -1;
493 : : }
364 tgl@sss.pgh.pa.us 494 :UBC 0 : break; /* must be EOF */
495 : : }
364 tgl@sss.pgh.pa.us 496 :CBC 1 : state->bufdata = rsize;
497 : 1 : state->bufnext = 0;
498 : : }
499 : :
500 : : /*
501 : : * Decompress some data into state->outbuf.
502 : : */
503 : : {
504 : : size_t status;
505 : 1 : size_t outlen = state->outbuflen;
506 : 1 : size_t inlen = state->bufdata - state->bufnext;
507 : :
508 : 1 : status = LZ4F_decompress(state->dtx,
509 : 1 : state->outbuf, &outlen,
510 : 1 : state->buffer + state->bufnext,
511 : : &inlen,
512 : : NULL);
1327 tomas.vondra@postgre 513 [ - + ]: 1 : if (LZ4F_isError(status))
514 : : {
1290 tomas.vondra@postgre 515 :UBC 0 : state->errcode = status;
409 dgustafsson@postgres 516 : 0 : pg_log_error("could not read from input file: %s",
517 : : LZ4F_getErrorName(state->errcode));
1327 tomas.vondra@postgre 518 : 0 : return -1;
519 : : }
3 dgustafsson@postgres 520 :CBC 1 : state->frame_finished = (status == 0);
364 tgl@sss.pgh.pa.us 521 : 1 : state->bufnext += inlen;
522 : 1 : state->outbufdata = outlen;
523 : 1 : state->outbufnext = 0;
524 : : }
525 : : }
526 : :
1299 tomas.vondra@postgre 527 :UBC 0 : return dsize;
528 : : }
529 : :
530 : : /*
531 : : * Compress size bytes from ptr and write them to the stream.
532 : : */
533 : : static void
1290 tomas.vondra@postgre 534 :CBC 93 : LZ4Stream_write(const void *ptr, size_t size, CompressFileHandle *CFH)
535 : : {
536 : 93 : LZ4State *state = (LZ4State *) CFH->private_data;
364 tgl@sss.pgh.pa.us 537 : 93 : size_t remaining = size;
538 : :
539 : : /* Lazy init */
540 [ - + ]: 93 : if (!LZ4Stream_init(state, true))
409 dgustafsson@postgres 541 :UBC 0 : pg_fatal("unable to initialize LZ4 library: %s",
542 : : LZ4F_getErrorName(state->errcode));
543 : :
1327 tomas.vondra@postgre 544 [ + + ]:CBC 190 : while (remaining > 0)
545 : : {
546 : : size_t chunk;
547 : : size_t required;
548 : : size_t status;
549 : :
550 : : /* We don't try to present more than DEFAULT_IO_BUFFER_SIZE bytes */
364 tgl@sss.pgh.pa.us 551 : 97 : chunk = Min(remaining, (size_t) DEFAULT_IO_BUFFER_SIZE);
552 : :
553 : : /* If not enough space, must flush buffer */
554 : 97 : required = LZ4F_compressBound(chunk, &state->prefs);
555 [ + + ]: 97 : if (required > state->buflen - state->bufdata)
556 : : {
557 : 2 : errno = 0;
558 [ - + ]: 2 : if (fwrite(state->buffer, 1, state->bufdata, state->fp) != state->bufdata)
559 : : {
364 tgl@sss.pgh.pa.us 560 [ # # ]:UBC 0 : errno = (errno) ? errno : ENOSPC;
561 : 0 : pg_fatal("error during writing: %m");
562 : : }
364 tgl@sss.pgh.pa.us 563 :CBC 2 : state->bufdata = 0;
564 : : }
565 : :
566 : 97 : status = LZ4F_compressUpdate(state->ctx,
567 : 97 : state->buffer + state->bufdata,
568 : 97 : state->buflen - state->bufdata,
569 : : ptr, chunk, NULL);
1327 tomas.vondra@postgre 570 [ - + ]: 97 : if (LZ4F_isError(status))
409 dgustafsson@postgres 571 :UBC 0 : pg_fatal("error during writing: %s", LZ4F_getErrorName(status));
364 tgl@sss.pgh.pa.us 572 :CBC 97 : state->bufdata += status;
573 : :
1244 tomas.vondra@postgre 574 : 97 : ptr = ((const char *) ptr) + chunk;
364 tgl@sss.pgh.pa.us 575 : 97 : remaining -= chunk;
576 : : }
1327 tomas.vondra@postgre 577 : 93 : }
578 : :
579 : : /*
580 : : * fread() equivalent implementation for LZ4 compressed files.
581 : : */
582 : : static size_t
409 dgustafsson@postgres 583 : 1 : LZ4Stream_read(void *ptr, size_t size, CompressFileHandle *CFH)
584 : : {
1290 tomas.vondra@postgre 585 : 1 : LZ4State *state = (LZ4State *) CFH->private_data;
586 : : int ret;
587 : :
588 [ + - ]: 1 : if ((ret = LZ4Stream_read_internal(state, ptr, size, false)) < 0)
589 : 1 : pg_fatal("could not read from input file: %s", LZ4Stream_get_error(CFH));
590 : :
409 dgustafsson@postgres 591 :UBC 0 : return (size_t) ret;
592 : : }
593 : :
594 : : /*
595 : : * fgetc() equivalent implementation for LZ4 compressed files.
596 : : */
597 : : static int
1290 tomas.vondra@postgre 598 : 0 : LZ4Stream_getc(CompressFileHandle *CFH)
599 : : {
600 : 0 : LZ4State *state = (LZ4State *) CFH->private_data;
601 : : unsigned char c;
602 : :
603 [ # # ]: 0 : if (LZ4Stream_read_internal(state, &c, 1, false) <= 0)
604 : : {
605 [ # # ]: 0 : if (!LZ4Stream_eof(CFH))
606 : 0 : pg_fatal("could not read from input file: %s", LZ4Stream_get_error(CFH));
607 : : else
1327 608 : 0 : pg_fatal("could not read from input file: end of file");
609 : : }
610 : :
611 : 0 : return c;
612 : : }
613 : :
614 : : /*
615 : : * fgets() equivalent implementation for LZ4 compressed files.
616 : : */
617 : : static char *
1290 618 : 0 : LZ4Stream_gets(char *ptr, int size, CompressFileHandle *CFH)
619 : : {
620 : 0 : LZ4State *state = (LZ4State *) CFH->private_data;
621 : : int ret;
622 : :
1244 623 : 0 : ret = LZ4Stream_read_internal(state, ptr, size - 1, true);
624 : :
625 : : /*
626 : : * LZ4Stream_read_internal returning 0 or -1 means that it was either an
627 : : * EOF or an error, but gets_func is defined to return NULL in either case
628 : : * so we can treat both the same here. LZ4Stream_read_internal is expected
629 : : * to have performed relevent error logging already.
630 : : */
409 dgustafsson@postgres 631 [ # # ]: 0 : if (ret <= 0)
1327 tomas.vondra@postgre 632 : 0 : return NULL;
633 : :
634 : : /*
635 : : * Our caller expects the return string to be NULL terminated and we know
636 : : * that ret is greater than zero.
637 : : */
1244 638 : 0 : ptr[ret - 1] = '\0';
639 : :
1327 640 : 0 : return ptr;
641 : : }
642 : :
643 : : /*
644 : : * Finalize (de)compression of a stream. When compressing it will write any
645 : : * remaining content and/or generated footer from the LZ4 API.
646 : : */
647 : : static bool
1290 tomas.vondra@postgre 648 :CBC 4 : LZ4Stream_close(CompressFileHandle *CFH)
649 : : {
650 : : FILE *fp;
651 : 4 : LZ4State *state = (LZ4State *) CFH->private_data;
652 : : size_t required;
653 : : size_t status;
654 : : int ret;
221 tgl@sss.pgh.pa.us 655 : 4 : bool success = true;
656 : :
1290 tomas.vondra@postgre 657 : 4 : fp = state->fp;
658 [ + - ]: 4 : if (state->inited)
659 : : {
660 [ + - ]: 4 : if (state->compressing)
661 : : {
662 : : /* We might need to flush the buffer to make room */
364 tgl@sss.pgh.pa.us 663 : 4 : required = LZ4F_compressBound(0, &state->prefs);
664 [ - + ]: 4 : if (required > state->buflen - state->bufdata)
665 : : {
364 tgl@sss.pgh.pa.us 666 :UBC 0 : errno = 0;
667 [ # # ]: 0 : if (fwrite(state->buffer, 1, state->bufdata, state->fp) != state->bufdata)
668 : : {
669 [ # # ]: 0 : errno = (errno) ? errno : ENOSPC;
670 : 0 : pg_log_error("could not write to output file: %m");
221 671 : 0 : success = false;
672 : : }
364 673 : 0 : state->bufdata = 0;
674 : : }
675 : :
364 tgl@sss.pgh.pa.us 676 :CBC 4 : status = LZ4F_compressEnd(state->ctx,
677 : 4 : state->buffer + state->bufdata,
678 : 4 : state->buflen - state->bufdata,
679 : : NULL);
1327 tomas.vondra@postgre 680 [ - + ]: 4 : if (LZ4F_isError(status))
681 : : {
409 dgustafsson@postgres 682 :UBC 0 : pg_log_error("could not end compression: %s",
683 : : LZ4F_getErrorName(status));
221 tgl@sss.pgh.pa.us 684 : 0 : success = false;
685 : : }
686 : : else
364 tgl@sss.pgh.pa.us 687 :CBC 4 : state->bufdata += status;
688 : :
689 : 4 : errno = 0;
690 [ - + ]: 4 : if (fwrite(state->buffer, 1, state->bufdata, state->fp) != state->bufdata)
691 : : {
364 tgl@sss.pgh.pa.us 692 [ # # ]:UBC 0 : errno = (errno) ? errno : ENOSPC;
693 : 0 : pg_log_error("could not write to output file: %m");
221 694 : 0 : success = false;
695 : : }
696 : :
1290 tomas.vondra@postgre 697 :CBC 4 : status = LZ4F_freeCompressionContext(state->ctx);
1327 698 [ - + ]: 4 : if (LZ4F_isError(status))
699 : : {
409 dgustafsson@postgres 700 :UBC 0 : pg_log_error("could not end compression: %s",
701 : : LZ4F_getErrorName(status));
221 tgl@sss.pgh.pa.us 702 : 0 : success = false;
703 : : }
704 : : }
705 : : else
706 : : {
3 dgustafsson@postgres 707 [ # # ]: 0 : if (!state->frame_finished)
708 : : {
709 : 0 : errno = EIO;
710 : 0 : success = false;
711 : : }
1290 tomas.vondra@postgre 712 : 0 : status = LZ4F_freeDecompressionContext(state->dtx);
1327 713 [ # # ]: 0 : if (LZ4F_isError(status))
714 : : {
409 dgustafsson@postgres 715 : 0 : pg_log_error("could not end decompression: %s",
716 : : LZ4F_getErrorName(status));
221 tgl@sss.pgh.pa.us 717 : 0 : success = false;
718 : : }
364 719 : 0 : pg_free(state->outbuf);
720 : : }
721 : :
1290 tomas.vondra@postgre 722 :CBC 4 : pg_free(state->buffer);
723 : : }
724 : :
725 : 4 : pg_free(state);
409 dgustafsson@postgres 726 : 4 : CFH->private_data = NULL;
727 : :
728 : 4 : errno = 0;
729 : 4 : ret = fclose(fp);
730 [ - + ]: 4 : if (ret != 0)
731 : : {
409 dgustafsson@postgres 732 :UBC 0 : pg_log_error("could not close file: %m");
221 tgl@sss.pgh.pa.us 733 : 0 : success = false;
734 : : }
735 : :
221 tgl@sss.pgh.pa.us 736 :CBC 4 : return success;
737 : : }
738 : :
739 : : static bool
1290 tomas.vondra@postgre 740 : 5 : LZ4Stream_open(const char *path, int fd, const char *mode,
741 : : CompressFileHandle *CFH)
742 : : {
743 : 5 : LZ4State *state = (LZ4State *) CFH->private_data;
744 : :
1327 745 [ - + ]: 5 : if (fd >= 0)
746 : : {
207 tgl@sss.pgh.pa.us 747 :UBC 0 : int dup_fd = dup(fd);
748 : :
749 [ # # ]: 0 : if (dup_fd < 0)
750 : : {
751 : 0 : state->errcode = errno;
752 : 0 : return false;
753 : : }
754 : 0 : state->fp = fdopen(dup_fd, mode);
755 [ # # ]: 0 : if (state->fp == NULL)
756 : : {
757 : 0 : state->errcode = errno;
758 : 0 : close(dup_fd);
759 : 0 : return false;
760 : : }
761 : : }
762 : : else
763 : : {
207 tgl@sss.pgh.pa.us 764 :CBC 5 : state->fp = fopen(path, mode);
765 [ - + ]: 5 : if (state->fp == NULL)
766 : : {
207 tgl@sss.pgh.pa.us 767 :UBC 0 : state->errcode = errno;
768 : 0 : return false;
769 : : }
770 : : }
771 : :
1299 tomas.vondra@postgre 772 :CBC 5 : return true;
773 : : }
774 : :
775 : : static bool
1290 776 : 3 : LZ4Stream_open_write(const char *path, const char *mode, CompressFileHandle *CFH)
777 : : {
778 : : char *fname;
779 : : int save_errno;
780 : : bool ret;
781 : :
1327 782 : 3 : fname = psprintf("%s.lz4", path);
783 : 3 : ret = CFH->open_func(fname, -1, mode, CFH);
784 : :
1299 785 : 3 : save_errno = errno;
103 peter@eisentraut.org 786 :GNC 3 : pfree(fname);
1299 tomas.vondra@postgre 787 :CBC 3 : errno = save_errno;
788 : :
1327 789 : 3 : return ret;
790 : : }
791 : :
792 : : /*
793 : : * Public routines
794 : : */
795 : : void
796 : 5 : InitCompressFileHandleLZ4(CompressFileHandle *CFH,
797 : : const pg_compress_specification compression_spec)
798 : : {
799 : : LZ4State *state;
800 : :
1290 801 : 5 : CFH->open_func = LZ4Stream_open;
802 : 5 : CFH->open_write_func = LZ4Stream_open_write;
803 : 5 : CFH->read_func = LZ4Stream_read;
804 : 5 : CFH->write_func = LZ4Stream_write;
805 : 5 : CFH->gets_func = LZ4Stream_gets;
806 : 5 : CFH->getc_func = LZ4Stream_getc;
807 : 5 : CFH->eof_func = LZ4Stream_eof;
808 : 5 : CFH->close_func = LZ4Stream_close;
809 : 5 : CFH->get_error_func = LZ4Stream_get_error;
810 : :
1327 811 : 5 : CFH->compression_spec = compression_spec;
241 michael@paquier.xyz 812 : 5 : state = pg_malloc0_object(LZ4State);
1327 tomas.vondra@postgre 813 [ + - ]: 5 : if (CFH->compression_spec.level >= 0)
1290 814 : 5 : state->prefs.compressionLevel = CFH->compression_spec.level;
815 : :
816 : 5 : CFH->private_data = state;
1327 817 : 5 : }
818 : : #else /* USE_LZ4 */
819 : : void
820 : : InitCompressorLZ4(CompressorState *cs,
821 : : const pg_compress_specification compression_spec)
822 : : {
823 : : pg_fatal("this build does not support compression with %s", "LZ4");
824 : : }
825 : :
826 : : void
827 : : InitCompressFileHandleLZ4(CompressFileHandle *CFH,
828 : : const pg_compress_specification compression_spec)
829 : : {
830 : : pg_fatal("this build does not support compression with %s", "LZ4");
831 : : }
832 : : #endif /* USE_LZ4 */
|
