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