Branch data Line data Source code
1 : : /*-------------------------------------------------------------------------
2 : : *
3 : : * astreamer_gzip.c
4 : : *
5 : : * Archive streamers that deal with data compressed using gzip.
6 : : * astreamer_gzip_writer applies gzip compression to the input data
7 : : * and writes the result to a file. astreamer_gzip_decompressor assumes
8 : : * that the input stream is compressed using gzip and decompresses it.
9 : : *
10 : : * Note that the code in this file is asymmetric with what we do for
11 : : * other compression types: for lz4 and zstd, there is a compressor and
12 : : * a decompressor, rather than a writer and a decompressor. The approach
13 : : * taken here is less flexible, because a writer can only write to a file,
14 : : * while a compressor can write to a subsequent astreamer which is free
15 : : * to do whatever it likes. The reason it's like this is because this
16 : : * code was adapted from old, less-modular pg_basebackup code that used
17 : : * the same APIs that astreamer_gzip_writer now uses, and it didn't seem
18 : : * necessary to change anything at the time.
19 : : *
20 : : * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group
21 : : *
22 : : * IDENTIFICATION
23 : : * src/fe_utils/astreamer_gzip.c
24 : : *-------------------------------------------------------------------------
25 : : */
26 : :
27 : : #include "postgres_fe.h"
28 : :
29 : : #include <unistd.h>
30 : :
31 : : #ifdef HAVE_LIBZ
32 : : #include <zlib.h>
33 : : #endif
34 : :
35 : : #include "common/logging.h"
36 : : #include "fe_utils/astreamer.h"
37 : :
38 : : #ifdef HAVE_LIBZ
39 : : typedef struct astreamer_gzip_writer
40 : : {
41 : : astreamer base;
42 : : char *pathname;
43 : : gzFile gzfile;
44 : : } astreamer_gzip_writer;
45 : :
46 : : typedef struct astreamer_gzip_decompressor
47 : : {
48 : : astreamer base;
49 : : z_stream zstream;
50 : : size_t bytes_written;
51 : : astreamer_decompression_state state;
52 : : } astreamer_gzip_decompressor;
53 : :
54 : : static void astreamer_gzip_writer_content(astreamer *streamer,
55 : : astreamer_member *member,
56 : : const char *data, int len,
57 : : astreamer_archive_context context);
58 : : static void astreamer_gzip_writer_finalize(astreamer *streamer);
59 : : static void astreamer_gzip_writer_free(astreamer *streamer);
60 : : static const char *get_gz_error(gzFile gzf);
61 : :
62 : : static const astreamer_ops astreamer_gzip_writer_ops = {
63 : : .content = astreamer_gzip_writer_content,
64 : : .finalize = astreamer_gzip_writer_finalize,
65 : : .free = astreamer_gzip_writer_free
66 : : };
67 : :
68 : : static void astreamer_gzip_decompressor_content(astreamer *streamer,
69 : : astreamer_member *member,
70 : : const char *data, int len,
71 : : astreamer_archive_context context);
72 : : static void astreamer_gzip_decompressor_finalize(astreamer *streamer);
73 : : static void astreamer_gzip_decompressor_free(astreamer *streamer);
74 : : static void *gzip_palloc(void *opaque, unsigned items, unsigned size);
75 : : static void gzip_pfree(void *opaque, void *address);
76 : :
77 : : static const astreamer_ops astreamer_gzip_decompressor_ops = {
78 : : .content = astreamer_gzip_decompressor_content,
79 : : .finalize = astreamer_gzip_decompressor_finalize,
80 : : .free = astreamer_gzip_decompressor_free
81 : : };
82 : : #endif
83 : :
84 : : /*
85 : : * Create a astreamer that just compresses data using gzip, and then writes
86 : : * it to a file.
87 : : *
88 : : * The caller must specify a pathname and may specify a file. The pathname is
89 : : * used for error-reporting purposes either way. If file is NULL, the pathname
90 : : * also identifies the file to which the data should be written: it is opened
91 : : * for writing and closed when done. If file is not NULL, the data is written
92 : : * there.
93 : : *
94 : : * Note that zlib does not use the FILE interface, but operates directly on
95 : : * a duplicate of the underlying fd. Hence, callers must take care if they
96 : : * plan to write any other data to the same FILE, either before or after using
97 : : * this.
98 : : */
99 : : astreamer *
100 : 5 : astreamer_gzip_writer_new(char *pathname, FILE *file,
101 : : pg_compress_specification *compress)
102 : : {
103 : : #ifdef HAVE_LIBZ
104 : : astreamer_gzip_writer *streamer;
105 : :
106 : 5 : streamer = palloc0_object(astreamer_gzip_writer);
107 : 5 : *((const astreamer_ops **) &streamer->base.bbs_ops) =
108 : : &astreamer_gzip_writer_ops;
109 : :
110 : 5 : streamer->pathname = pstrdup(pathname);
111 : :
112 [ + - ]: 5 : if (file == NULL)
113 : : {
114 : 5 : streamer->gzfile = gzopen(pathname, "wb");
115 [ - + ]: 5 : if (streamer->gzfile == NULL)
116 : 0 : pg_fatal("could not create compressed file \"%s\": %m",
117 : : pathname);
118 : : }
119 : : else
120 : : {
121 : : /*
122 : : * We must dup the file handle so that gzclose doesn't break the
123 : : * caller's FILE. See comment for astreamer_gzip_writer_finalize.
124 : : */
125 : 0 : int fd = dup(fileno(file));
126 : :
127 [ # # ]: 0 : if (fd < 0)
128 : 0 : pg_fatal("could not duplicate stdout: %m");
129 : :
130 : 0 : streamer->gzfile = gzdopen(fd, "wb");
131 [ # # ]: 0 : if (streamer->gzfile == NULL)
132 : 0 : pg_fatal("could not open output file: %m");
133 : : }
134 : :
135 [ - + ]: 5 : if (gzsetparams(streamer->gzfile, compress->level, Z_DEFAULT_STRATEGY) != Z_OK)
136 : 0 : pg_fatal("could not set compression level %d: %s",
137 : : compress->level, get_gz_error(streamer->gzfile));
138 : :
139 : 5 : return &streamer->base;
140 : : #else
141 : : pg_fatal("this build does not support compression with %s", "gzip");
142 : : return NULL; /* keep compiler quiet */
143 : : #endif
144 : : }
145 : :
146 : : #ifdef HAVE_LIBZ
147 : : /*
148 : : * Write archive content to gzip file.
149 : : */
150 : : static void
151 : 11517 : astreamer_gzip_writer_content(astreamer *streamer,
152 : : astreamer_member *member, const char *data,
153 : : int len, astreamer_archive_context context)
154 : : {
155 : : astreamer_gzip_writer *mystreamer;
156 : :
157 : 11517 : mystreamer = (astreamer_gzip_writer *) streamer;
158 : :
159 [ - + ]: 11517 : if (len == 0)
160 : 0 : return;
161 : :
162 : 11517 : errno = 0;
163 [ - + ]: 11517 : if (gzwrite(mystreamer->gzfile, data, len) != len)
164 : : {
165 : : /* if write didn't set errno, assume problem is no disk space */
166 [ # # ]: 0 : if (errno == 0)
167 : 0 : errno = ENOSPC;
168 : 0 : pg_fatal("could not write to compressed file \"%s\": %s",
169 : : mystreamer->pathname, get_gz_error(mystreamer->gzfile));
170 : : }
171 : : }
172 : :
173 : : /*
174 : : * End-of-archive processing when writing to a gzip file consists of just
175 : : * calling gzclose.
176 : : *
177 : : * It makes no difference whether we opened the file or the caller did it,
178 : : * because libz provides no way of avoiding a close on the underlying file
179 : : * handle. Notice, however, that astreamer_gzip_writer_new() uses dup() to
180 : : * work around this issue, so that the behavior from the caller's viewpoint
181 : : * is the same as for astreamer_plain_writer.
182 : : */
183 : : static void
184 : 5 : astreamer_gzip_writer_finalize(astreamer *streamer)
185 : : {
186 : : astreamer_gzip_writer *mystreamer;
187 : :
188 : 5 : mystreamer = (astreamer_gzip_writer *) streamer;
189 : :
190 : 5 : errno = 0; /* in case gzclose() doesn't set it */
191 [ - + ]: 5 : if (gzclose(mystreamer->gzfile) != 0)
192 : 0 : pg_fatal("could not close compressed file \"%s\": %m",
193 : : mystreamer->pathname);
194 : :
195 : 5 : mystreamer->gzfile = NULL;
196 : 5 : }
197 : :
198 : : /*
199 : : * Free memory associated with this astreamer.
200 : : */
201 : : static void
202 : 5 : astreamer_gzip_writer_free(astreamer *streamer)
203 : : {
204 : : astreamer_gzip_writer *mystreamer;
205 : :
206 : 5 : mystreamer = (astreamer_gzip_writer *) streamer;
207 : :
208 : : Assert(mystreamer->base.bbs_next == NULL);
209 : : Assert(mystreamer->gzfile == NULL);
210 : :
211 : 5 : pfree(mystreamer->pathname);
212 : 5 : pfree(mystreamer);
213 : 5 : }
214 : :
215 : : /*
216 : : * Helper function for libz error reporting.
217 : : */
218 : : static const char *
219 : 0 : get_gz_error(gzFile gzf)
220 : : {
221 : : int errnum;
222 : : const char *errmsg;
223 : :
224 : 0 : errmsg = gzerror(gzf, &errnum);
225 [ # # ]: 0 : if (errnum == Z_ERRNO)
226 : 0 : return strerror(errno);
227 : : else
228 : 0 : return errmsg;
229 : : }
230 : : #endif
231 : :
232 : : /*
233 : : * Create a new base backup streamer that performs decompression of gzip
234 : : * compressed blocks.
235 : : */
236 : : astreamer *
237 : 24 : astreamer_gzip_decompressor_new(astreamer *next)
238 : : {
239 : : #ifdef HAVE_LIBZ
240 : : astreamer_gzip_decompressor *streamer;
241 : : z_stream *zs;
242 : :
243 : : Assert(next != NULL);
244 : :
245 : 24 : streamer = palloc0_object(astreamer_gzip_decompressor);
246 : 24 : *((const astreamer_ops **) &streamer->base.bbs_ops) =
247 : : &astreamer_gzip_decompressor_ops;
248 : :
249 : 24 : streamer->base.bbs_next = next;
250 : 24 : initStringInfo(&streamer->base.bbs_buffer);
251 : : /* Use a buffer size comparable to the other decompressors */
252 : 24 : enlargeStringInfo(&streamer->base.bbs_buffer, 256 * 1024 - 1);
253 : :
254 : : /* Initialize internal stream state for decompression */
255 : 24 : zs = &streamer->zstream;
256 : 24 : zs->zalloc = gzip_palloc;
257 : 24 : zs->zfree = gzip_pfree;
258 : 24 : zs->next_out = (uint8 *) streamer->base.bbs_buffer.data;
259 : 24 : zs->avail_out = streamer->base.bbs_buffer.maxlen;
260 : :
261 : : /*
262 : : * Data compression was initialized using deflateInit2 to request a gzip
263 : : * header. Similarly, we are using inflateInit2 to initialize data
264 : : * decompression.
265 : : *
266 : : * Per the documentation for inflateInit2, the second argument is
267 : : * "windowBits" and its value must be greater than or equal to the value
268 : : * provided while compressing the data, so we are using the maximum
269 : : * possible value for safety.
270 : : */
271 [ - + ]: 24 : if (inflateInit2(zs, 15 + 16) != Z_OK)
272 : 0 : pg_fatal("could not initialize compression library");
273 : :
274 : 24 : streamer->state = ASTREAMER_STREAM_NEW;
275 : 24 : return &streamer->base;
276 : : #else
277 : : pg_fatal("this build does not support compression with %s", "gzip");
278 : : return NULL; /* keep compiler quiet */
279 : : #endif
280 : : }
281 : :
282 : : #ifdef HAVE_LIBZ
283 : : /*
284 : : * Decompress the input data to output buffer until we run out of input
285 : : * data. Each time the output buffer is full, pass on the decompressed data
286 : : * to the next streamer.
287 : : */
288 : : static void
289 : 596 : astreamer_gzip_decompressor_content(astreamer *streamer,
290 : : astreamer_member *member,
291 : : const char *data, int len,
292 : : astreamer_archive_context context)
293 : : {
294 : : astreamer_gzip_decompressor *mystreamer;
295 : : z_stream *zs;
296 : :
297 : 596 : mystreamer = (astreamer_gzip_decompressor *) streamer;
298 : :
299 : 596 : zs = &mystreamer->zstream;
300 : 596 : zs->next_in = (const uint8 *) data;
301 : 596 : zs->avail_in = len;
302 : :
303 : : /* Process the current chunk */
304 [ + + ]: 5299 : while (zs->avail_in > 0)
305 : : {
306 : : int res;
307 : :
308 : : Assert(mystreamer->bytes_written < mystreamer->base.bbs_buffer.maxlen);
309 : :
310 : 4107 : zs->next_out = (uint8 *)
311 : 4107 : mystreamer->base.bbs_buffer.data + mystreamer->bytes_written;
312 : 4107 : zs->avail_out =
313 : 4107 : mystreamer->base.bbs_buffer.maxlen - mystreamer->bytes_written;
314 : :
315 : : /*
316 : : * This call decompresses data starting at zs->next_in and updates
317 : : * zs->next_in * and zs->avail_in. It generates output data starting
318 : : * at zs->next_out and updates zs->next_out and zs->avail_out
319 : : * accordingly.
320 : : */
321 : 4107 : res = inflate(zs, Z_NO_FLUSH);
322 : :
323 [ + + ]: 4107 : if (res == Z_STREAM_END)
324 : 29 : mystreamer->state = ASTREAMER_FRAME_COMPLETE;
325 [ - + - - ]: 4078 : else if (res == Z_OK || res == Z_BUF_ERROR)
326 : 4078 : mystreamer->state = ASTREAMER_FRAME_INCOMPLETE;
327 : : else
328 [ # # ]: 0 : pg_fatal("could not decompress data: %s",
329 : : zs->msg ? zs->msg : "unknown error");
330 : :
331 : 4107 : mystreamer->bytes_written =
332 : 4107 : mystreamer->base.bbs_buffer.maxlen - zs->avail_out;
333 : :
334 : : /* If output buffer is full then pass data to next streamer */
335 [ + + ]: 4107 : if (mystreamer->bytes_written >= mystreamer->base.bbs_buffer.maxlen)
336 : : {
337 : 3502 : astreamer_content(mystreamer->base.bbs_next, member,
338 : 3502 : mystreamer->base.bbs_buffer.data,
339 : : mystreamer->base.bbs_buffer.maxlen, context);
340 : 3502 : mystreamer->bytes_written = 0;
341 : : }
342 : :
343 : : /*
344 : : * The input can be several gzip members one after another, which is
345 : : * what pg_basebackup writes for pg_wal.tar.gz. Get ready for the
346 : : * next one.
347 : : */
348 [ + + - + ]: 4107 : if (res == Z_STREAM_END && inflateReset(zs) != Z_OK)
349 : 0 : pg_fatal("could not reset compression stream");
350 : : }
351 : 596 : }
352 : :
353 : : /*
354 : : * End-of-stream processing.
355 : : */
356 : : static void
357 : 6 : astreamer_gzip_decompressor_finalize(astreamer *streamer)
358 : : {
359 : : astreamer_gzip_decompressor *mystreamer;
360 : :
361 : 6 : mystreamer = (astreamer_gzip_decompressor *) streamer;
362 : :
363 [ - + ]: 6 : if (unlikely(mystreamer->state == ASTREAMER_STREAM_NEW))
364 : 0 : pg_fatal("could not decompress data: compressed stream is empty");
365 [ + + ]: 6 : else if (mystreamer->state != ASTREAMER_FRAME_COMPLETE)
366 : 1 : pg_fatal("could not decompress data: compressed stream is incomplete");
367 : :
368 : : /*
369 : : * End of the stream, if there is some pending data in output buffers then
370 : : * we must forward it to next streamer.
371 : : */
372 [ + - ]: 5 : if (mystreamer->bytes_written > 0)
373 : 5 : astreamer_content(mystreamer->base.bbs_next, NULL,
374 : 5 : mystreamer->base.bbs_buffer.data,
375 : 5 : mystreamer->bytes_written,
376 : : ASTREAMER_UNKNOWN);
377 : :
378 : 5 : astreamer_finalize(mystreamer->base.bbs_next);
379 : 5 : }
380 : :
381 : : /*
382 : : * Free memory.
383 : : */
384 : : static void
385 : 21 : astreamer_gzip_decompressor_free(astreamer *streamer)
386 : : {
387 : : astreamer_gzip_decompressor *mystreamer;
388 : :
389 : 21 : mystreamer = (astreamer_gzip_decompressor *) streamer;
390 : :
391 : 21 : astreamer_free(streamer->bbs_next);
392 : 21 : inflateEnd(&mystreamer->zstream);
393 : 21 : pfree(streamer->bbs_buffer.data);
394 : 21 : pfree(streamer);
395 : 21 : }
396 : :
397 : : /*
398 : : * Wrapper function to adjust the signature of palloc to match what libz
399 : : * expects.
400 : : */
401 : : static void *
402 : 47 : gzip_palloc(void *opaque, unsigned items, unsigned size)
403 : : {
404 : 47 : return palloc(items * size);
405 : : }
406 : :
407 : : /*
408 : : * Wrapper function to adjust the signature of pfree to match what libz
409 : : * expects.
410 : : */
411 : : static void
412 : 41 : gzip_pfree(void *opaque, void *address)
413 : : {
414 : 41 : pfree(address);
415 : 41 : }
416 : : #endif
|
