Branch data Line data Source code
1 : : /*-------------------------------------------------------------------------
2 : : *
3 : : * basebackup_gzip.c
4 : : * Basebackup sink implementing gzip compression.
5 : : *
6 : : * Portions Copyright (c) 2010-2026, PostgreSQL Global Development Group
7 : : *
8 : : * IDENTIFICATION
9 : : * src/backend/backup/basebackup_gzip.c
10 : : *
11 : : *-------------------------------------------------------------------------
12 : : */
13 : : #include "postgres.h"
14 : :
15 : : #ifdef HAVE_LIBZ
16 : : #include <zlib.h>
17 : : #endif
18 : :
19 : : #include "backup/basebackup_sink.h"
20 : :
21 : : #ifdef HAVE_LIBZ
22 : : typedef struct bbsink_gzip
23 : : {
24 : : /* Common information for all types of sink. */
25 : : bbsink base;
26 : :
27 : : /* Compression level. */
28 : : int compresslevel;
29 : :
30 : : /* Compressed data stream. */
31 : : z_stream zstream;
32 : :
33 : : /* Number of bytes staged in output buffer. */
34 : : size_t bytes_written;
35 : :
36 : : /* Has the zstream been initialized? */
37 : : bool zstream_initialized;
38 : : } bbsink_gzip;
39 : :
40 : : static void bbsink_gzip_begin_backup(bbsink *sink);
41 : : static void bbsink_gzip_begin_archive(bbsink *sink, const char *archive_name);
42 : : static void bbsink_gzip_archive_contents(bbsink *sink, size_t len);
43 : : static void bbsink_gzip_manifest_contents(bbsink *sink, size_t len);
44 : : static void bbsink_gzip_end_archive(bbsink *sink);
45 : : static void bbsink_gzip_cleanup(bbsink *sink);
46 : : static void *gzip_palloc(void *opaque, unsigned items, unsigned size);
47 : : static void gzip_pfree(void *opaque, void *address);
48 : :
49 : : static const bbsink_ops bbsink_gzip_ops = {
50 : : .begin_backup = bbsink_gzip_begin_backup,
51 : : .begin_archive = bbsink_gzip_begin_archive,
52 : : .archive_contents = bbsink_gzip_archive_contents,
53 : : .end_archive = bbsink_gzip_end_archive,
54 : : .begin_manifest = bbsink_forward_begin_manifest,
55 : : .manifest_contents = bbsink_gzip_manifest_contents,
56 : : .end_manifest = bbsink_forward_end_manifest,
57 : : .end_backup = bbsink_forward_end_backup,
58 : : .cleanup = bbsink_gzip_cleanup
59 : : };
60 : : #endif
61 : :
62 : : /*
63 : : * Create a new basebackup sink that performs gzip compression.
64 : : */
65 : : bbsink *
66 : 2 : bbsink_gzip_new(bbsink *next, pg_compress_specification *compress)
67 : : {
68 : : #ifndef HAVE_LIBZ
69 : : ereport(ERROR,
70 : : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
71 : : errmsg("gzip compression is not supported by this build")));
72 : : return NULL; /* keep compiler quiet */
73 : : #else
74 : : bbsink_gzip *sink;
75 : : int compresslevel;
76 : :
77 : : Assert(next != NULL);
78 : :
79 : 2 : compresslevel = compress->level;
80 : : Assert((compresslevel >= 1 && compresslevel <= 9) ||
81 : : compresslevel == Z_DEFAULT_COMPRESSION);
82 : :
83 : 2 : sink = palloc0_object(bbsink_gzip);
84 : 2 : *((const bbsink_ops **) &sink->base.bbs_ops) = &bbsink_gzip_ops;
85 : 2 : sink->base.bbs_next = next;
86 : 2 : sink->compresslevel = compresslevel;
87 : :
88 : 2 : return &sink->base;
89 : : #endif
90 : : }
91 : :
92 : : #ifdef HAVE_LIBZ
93 : :
94 : : /*
95 : : * Begin backup.
96 : : */
97 : : static void
98 : 2 : bbsink_gzip_begin_backup(bbsink *sink)
99 : : {
100 : : /*
101 : : * We need our own buffer, because we're going to pass different data to
102 : : * the next sink than what gets passed to us.
103 : : */
104 : 2 : sink->bbs_buffer = palloc(sink->bbs_buffer_length);
105 : :
106 : : /*
107 : : * Since deflate() doesn't require the output buffer to be of any
108 : : * particular size, we can just make it the same size as the input buffer.
109 : : */
110 : 2 : bbsink_begin_backup(sink->bbs_next, sink->bbs_state,
111 : 2 : sink->bbs_buffer_length);
112 : 2 : }
113 : :
114 : : /*
115 : : * Prepare to compress the next archive.
116 : : */
117 : : static void
118 : 3 : bbsink_gzip_begin_archive(bbsink *sink, const char *archive_name)
119 : : {
120 : 3 : bbsink_gzip *mysink = (bbsink_gzip *) sink;
121 : : char *gz_archive_name;
122 : 3 : z_stream *zs = &mysink->zstream;
123 : :
124 : : /* Initialize compressor object. */
125 : 3 : memset(zs, 0, sizeof(z_stream));
126 : 3 : zs->zalloc = gzip_palloc;
127 : 3 : zs->zfree = gzip_pfree;
128 : 3 : zs->next_out = (uint8 *) sink->bbs_next->bbs_buffer;
129 : 3 : zs->avail_out = sink->bbs_next->bbs_buffer_length;
130 : :
131 : : /*
132 : : * We need to use deflateInit2() rather than deflateInit() here so that we
133 : : * can request a gzip header rather than a zlib header. Otherwise, we want
134 : : * to supply the same values that would have been used by default if we
135 : : * had just called deflateInit().
136 : : *
137 : : * Per the documentation for deflateInit2, the third argument must be
138 : : * Z_DEFLATED; the fourth argument is the number of "window bits", by
139 : : * default 15, but adding 16 gets you a gzip header rather than a zlib
140 : : * header; the fifth argument controls memory usage, and 8 is the default;
141 : : * and likewise Z_DEFAULT_STRATEGY is the default for the sixth argument.
142 : : */
143 [ - + ]: 3 : if (deflateInit2(zs, mysink->compresslevel, Z_DEFLATED, 15 + 16, 8,
144 : : Z_DEFAULT_STRATEGY) != Z_OK)
145 [ # # ]: 0 : ereport(ERROR,
146 : : errcode(ERRCODE_INTERNAL_ERROR),
147 : : errmsg("could not initialize compression library"));
148 : 3 : mysink->zstream_initialized = true;
149 : :
150 : : /*
151 : : * Add ".gz" to the archive name. Note that the pg_basebackup -z produces
152 : : * archives named ".tar.gz" rather than ".tgz", so we match that here.
153 : : */
154 : 3 : gz_archive_name = psprintf("%s.gz", archive_name);
155 : : Assert(sink->bbs_next != NULL);
156 : 3 : bbsink_begin_archive(sink->bbs_next, gz_archive_name);
157 : 3 : pfree(gz_archive_name);
158 : 3 : }
159 : :
160 : : /*
161 : : * Compress the input data to the output buffer until we run out of input
162 : : * data. Each time the output buffer fills up, invoke the archive_contents()
163 : : * method for then next sink.
164 : : *
165 : : * Note that since we're compressing the input, it may very commonly happen
166 : : * that we consume all the input data without filling the output buffer. In
167 : : * that case, the compressed representation of the current input data won't
168 : : * actually be sent to the next bbsink until a later call to this function,
169 : : * or perhaps even not until bbsink_gzip_end_archive() is invoked.
170 : : */
171 : : static void
172 : 5665 : bbsink_gzip_archive_contents(bbsink *sink, size_t len)
173 : : {
174 : 5665 : bbsink_gzip *mysink = (bbsink_gzip *) sink;
175 : 5665 : z_stream *zs = &mysink->zstream;
176 : :
177 : : /* Compress data from input buffer. */
178 : 5665 : zs->next_in = (uint8 *) mysink->base.bbs_buffer;
179 : 5665 : zs->avail_in = len;
180 : :
181 [ + + ]: 17101 : while (zs->avail_in > 0)
182 : : {
183 : : int res;
184 : :
185 : : /* Write output data into unused portion of output buffer. */
186 : : Assert(mysink->bytes_written < mysink->base.bbs_next->bbs_buffer_length);
187 : 5771 : zs->next_out = (uint8 *)
188 : 5771 : mysink->base.bbs_next->bbs_buffer + mysink->bytes_written;
189 : 5771 : zs->avail_out =
190 : 5771 : mysink->base.bbs_next->bbs_buffer_length - mysink->bytes_written;
191 : :
192 : : /*
193 : : * Try to compress. Note that this will update zs->next_in and
194 : : * zs->avail_in according to how much input data was consumed, and
195 : : * zs->next_out and zs->avail_out according to how many output bytes
196 : : * were produced.
197 : : *
198 : : * According to the zlib documentation, Z_STREAM_ERROR should only
199 : : * occur if we've made a programming error, or if say there's been a
200 : : * memory clobber; we use elog() rather than Assert() here out of an
201 : : * abundance of caution.
202 : : */
203 : 5771 : res = deflate(zs, Z_NO_FLUSH);
204 [ - + ]: 5771 : if (res == Z_STREAM_ERROR)
205 [ # # ]: 0 : elog(ERROR, "could not compress data: %s", zs->msg);
206 : :
207 : : /* Update our notion of how many bytes we've written. */
208 : 5771 : mysink->bytes_written =
209 : 5771 : mysink->base.bbs_next->bbs_buffer_length - zs->avail_out;
210 : :
211 : : /*
212 : : * If the output buffer is full, it's time for the next sink to
213 : : * process the contents.
214 : : */
215 [ + + ]: 5771 : if (mysink->bytes_written >= mysink->base.bbs_next->bbs_buffer_length)
216 : : {
217 : 208 : bbsink_archive_contents(sink->bbs_next, mysink->bytes_written);
218 : 208 : mysink->bytes_written = 0;
219 : : }
220 : : }
221 : 5665 : }
222 : :
223 : : /*
224 : : * There might be some data inside zlib's internal buffers; we need to get
225 : : * that flushed out and forwarded to the successor sink as archive content.
226 : : *
227 : : * Then we can end processing for this archive.
228 : : */
229 : : static void
230 : 3 : bbsink_gzip_end_archive(bbsink *sink)
231 : : {
232 : 3 : bbsink_gzip *mysink = (bbsink_gzip *) sink;
233 : 3 : z_stream *zs = &mysink->zstream;
234 : :
235 : : /* There is no more data available. */
236 : 3 : zs->next_in = (uint8 *) mysink->base.bbs_buffer;
237 : 3 : zs->avail_in = 0;
238 : :
239 : : while (1)
240 : 3 : {
241 : : int res;
242 : :
243 : : /* Write output data into unused portion of output buffer. */
244 : : Assert(mysink->bytes_written < mysink->base.bbs_next->bbs_buffer_length);
245 : 6 : zs->next_out = (uint8 *)
246 : 6 : mysink->base.bbs_next->bbs_buffer + mysink->bytes_written;
247 : 6 : zs->avail_out =
248 : 6 : mysink->base.bbs_next->bbs_buffer_length - mysink->bytes_written;
249 : :
250 : : /*
251 : : * As bbsink_gzip_archive_contents, but pass Z_FINISH since there is
252 : : * no more input.
253 : : */
254 : 6 : res = deflate(zs, Z_FINISH);
255 [ - + ]: 6 : if (res == Z_STREAM_ERROR)
256 [ # # ]: 0 : elog(ERROR, "could not compress data: %s", zs->msg);
257 : :
258 : : /* Update our notion of how many bytes we've written. */
259 : 6 : mysink->bytes_written =
260 : 6 : mysink->base.bbs_next->bbs_buffer_length - zs->avail_out;
261 : :
262 : : /*
263 : : * Apparently we had no data in the output buffer and deflate() was
264 : : * not able to add any. We must be done.
265 : : */
266 [ + + ]: 6 : if (mysink->bytes_written == 0)
267 : 3 : break;
268 : :
269 : : /* Send whatever accumulated output bytes we have. */
270 : 3 : bbsink_archive_contents(sink->bbs_next, mysink->bytes_written);
271 : 3 : mysink->bytes_written = 0;
272 : : }
273 : :
274 : : /* Release the compression resources. */
275 : 3 : deflateEnd(zs);
276 : 3 : mysink->zstream_initialized = false;
277 : :
278 : : /* Must also pass on the information that this archive has ended. */
279 : 3 : bbsink_forward_end_archive(sink);
280 : 3 : }
281 : :
282 : : /*
283 : : * Manifest contents are not compressed, but we do need to copy them into
284 : : * the successor sink's buffer, because we have our own.
285 : : */
286 : : static void
287 : 10 : bbsink_gzip_manifest_contents(bbsink *sink, size_t len)
288 : : {
289 : 10 : memcpy(sink->bbs_next->bbs_buffer, sink->bbs_buffer, len);
290 : 10 : bbsink_manifest_contents(sink->bbs_next, len);
291 : 10 : }
292 : :
293 : : /*
294 : : * Wrapper function to adjust the signature of palloc to match what libz
295 : : * expects.
296 : : */
297 : : static void *
298 : 15 : gzip_palloc(void *opaque, unsigned items, unsigned size)
299 : : {
300 : 15 : return palloc(items * size);
301 : : }
302 : :
303 : : /*
304 : : * Wrapper function to adjust the signature of pfree to match what libz
305 : : * expects.
306 : : */
307 : : static void
308 : 15 : gzip_pfree(void *opaque, void *address)
309 : : {
310 : 15 : pfree(address);
311 : 15 : }
312 : :
313 : : /*
314 : : * In case the backup fails, make sure we free the compression context by
315 : : * calling deflateEnd() if needed to avoid a resource leak.
316 : : */
317 : : static void
318 : 2 : bbsink_gzip_cleanup(bbsink *sink)
319 : : {
320 : 2 : bbsink_gzip *mysink = (bbsink_gzip *) sink;
321 : :
322 [ - + ]: 2 : if (mysink->zstream_initialized)
323 : : {
324 : 0 : deflateEnd(&mysink->zstream);
325 : 0 : mysink->zstream_initialized = false;
326 : : }
327 : 2 : }
328 : :
329 : : #endif
|
