Branch data Line data Source code
1 : : /*-------------------------------------------------------------------------
2 : : *
3 : : * walmethods.c - implementations of different ways to write received wal
4 : : *
5 : : * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group
6 : : *
7 : : * IDENTIFICATION
8 : : * src/bin/pg_basebackup/walmethods.c
9 : : *-------------------------------------------------------------------------
10 : : */
11 : :
12 : : #include "postgres_fe.h"
13 : :
14 : : #include <fcntl.h>
15 : : #include <sys/stat.h>
16 : : #include <time.h>
17 : : #include <unistd.h>
18 : :
19 : : #ifdef USE_LZ4
20 : : #include <lz4frame.h>
21 : : #endif
22 : : #ifdef HAVE_LIBZ
23 : : #include <zlib.h>
24 : : #endif
25 : :
26 : : #include "common/file_perm.h"
27 : : #include "common/file_utils.h"
28 : : #include "common/logging.h"
29 : : #include "pgtar.h"
30 : : #include "walmethods.h"
31 : :
32 : : /* Size of zlib buffer for .tar.gz */
33 : : #define ZLIB_OUT_SIZE 4096
34 : :
35 : : /* Size of LZ4 input chunk for .lz4 */
36 : : #define LZ4_IN_SIZE 4096
37 : :
38 : : /*-------------------------------------------------------------------------
39 : : * WalDirectoryMethod - write wal to a directory looking like pg_wal
40 : : *-------------------------------------------------------------------------
41 : : */
42 : :
43 : : static Walfile *dir_open_for_write(WalWriteMethod *wwmethod,
44 : : const char *pathname,
45 : : const char *temp_suffix,
46 : : size_t pad_to_size);
47 : : static int dir_close(Walfile *f, WalCloseMethod method);
48 : : static bool dir_existsfile(WalWriteMethod *wwmethod, const char *pathname);
49 : : static ssize_t dir_get_file_size(WalWriteMethod *wwmethod,
50 : : const char *pathname);
51 : : static char *dir_get_file_name(WalWriteMethod *wwmethod,
52 : : const char *pathname, const char *temp_suffix);
53 : : static ssize_t dir_write(Walfile *f, const void *buf, size_t count);
54 : : static int dir_sync(Walfile *f);
55 : : static bool dir_finish(WalWriteMethod *wwmethod);
56 : : static void dir_free(WalWriteMethod *wwmethod);
57 : :
58 : : static const WalWriteMethodOps WalDirectoryMethodOps = {
59 : : .open_for_write = dir_open_for_write,
60 : : .close = dir_close,
61 : : .existsfile = dir_existsfile,
62 : : .get_file_size = dir_get_file_size,
63 : : .get_file_name = dir_get_file_name,
64 : : .write = dir_write,
65 : : .sync = dir_sync,
66 : : .finish = dir_finish,
67 : : .free = dir_free
68 : : };
69 : :
70 : : /*
71 : : * Global static data for this method
72 : : */
73 : : typedef struct DirectoryMethodData
74 : : {
75 : : WalWriteMethod base;
76 : : char *basedir;
77 : : } DirectoryMethodData;
78 : :
79 : : /*
80 : : * Local file handle
81 : : */
82 : : typedef struct DirectoryMethodFile
83 : : {
84 : : Walfile base;
85 : : int fd;
86 : : char *fullpath;
87 : : char *temp_suffix;
88 : : #ifdef HAVE_LIBZ
89 : : gzFile gzfp;
90 : : #endif
91 : : #ifdef USE_LZ4
92 : : LZ4F_compressionContext_t ctx;
93 : : size_t lz4bufsize;
94 : : void *lz4buf;
95 : : #endif
96 : : } DirectoryMethodFile;
97 : :
98 : : #define clear_error(wwmethod) \
99 : : ((wwmethod)->lasterrstring = NULL, (wwmethod)->lasterrno = 0)
100 : :
101 : : static char *
102 : 481 : dir_get_file_name(WalWriteMethod *wwmethod,
103 : : const char *pathname, const char *temp_suffix)
104 : : {
105 : 481 : char *filename = pg_malloc0_array(char, MAXPGPATH);
106 : :
107 [ + + ]: 962 : snprintf(filename, MAXPGPATH, "%s%s%s",
108 : : pathname,
109 [ + + ]: 481 : wwmethod->compression_algorithm == PG_COMPRESSION_GZIP ? ".gz" :
110 [ + + ]: 473 : wwmethod->compression_algorithm == PG_COMPRESSION_LZ4 ? ".lz4" : "",
111 : : temp_suffix ? temp_suffix : "");
112 : :
113 : 481 : return filename;
114 : : }
115 : :
116 : : static Walfile *
117 : 163 : dir_open_for_write(WalWriteMethod *wwmethod, const char *pathname,
118 : : const char *temp_suffix, size_t pad_to_size)
119 : : {
120 : 163 : DirectoryMethodData *dir_data = (DirectoryMethodData *) wwmethod;
121 : : char tmppath[MAXPGPATH];
122 : : char *filename;
123 : : int fd;
124 : : DirectoryMethodFile *f;
125 : : #ifdef HAVE_LIBZ
126 : 163 : gzFile gzfp = NULL;
127 : : #endif
128 : : #ifdef USE_LZ4
129 : 163 : LZ4F_compressionContext_t ctx = NULL;
130 : 163 : size_t lz4bufsize = 0;
131 : 163 : void *lz4buf = NULL;
132 : : #endif
133 : :
134 : 163 : clear_error(wwmethod);
135 : :
136 : 163 : filename = dir_get_file_name(wwmethod, pathname, temp_suffix);
137 : 163 : snprintf(tmppath, sizeof(tmppath), "%s/%s",
138 : : dir_data->basedir, filename);
139 : 163 : pg_free(filename);
140 : :
141 : : /*
142 : : * Open a file for non-compressed as well as compressed files. Tracking
143 : : * the file descriptor is important for dir_sync() method as gzflush()
144 : : * does not do any system calls to fsync() to make changes permanent on
145 : : * disk.
146 : : */
147 : 163 : fd = open(tmppath, O_WRONLY | O_CREAT | PG_BINARY, pg_file_create_mode);
148 [ - + ]: 163 : if (fd < 0)
149 : : {
150 : 0 : wwmethod->lasterrno = errno;
151 : 0 : return NULL;
152 : : }
153 : :
154 : : #ifdef HAVE_LIBZ
155 [ + + ]: 163 : if (wwmethod->compression_algorithm == PG_COMPRESSION_GZIP)
156 : : {
157 : 2 : gzfp = gzdopen(fd, "wb");
158 [ - + ]: 2 : if (gzfp == NULL)
159 : : {
160 : 0 : wwmethod->lasterrno = errno;
161 : 0 : close(fd);
162 : 0 : return NULL;
163 : : }
164 : :
165 [ - + ]: 2 : if (gzsetparams(gzfp, wwmethod->compression_level,
166 : : Z_DEFAULT_STRATEGY) != Z_OK)
167 : : {
168 : 0 : wwmethod->lasterrno = errno;
169 : 0 : gzclose(gzfp);
170 : 0 : return NULL;
171 : : }
172 : : }
173 : : #endif
174 : : #ifdef USE_LZ4
175 [ + + ]: 163 : if (wwmethod->compression_algorithm == PG_COMPRESSION_LZ4)
176 : : {
177 : : size_t ctx_out;
178 : : size_t header_size;
179 : : LZ4F_preferences_t prefs;
180 : :
181 : 2 : ctx_out = LZ4F_createCompressionContext(&ctx, LZ4F_VERSION);
182 [ - + ]: 2 : if (LZ4F_isError(ctx_out))
183 : : {
184 : 0 : wwmethod->lasterrstring = LZ4F_getErrorName(ctx_out);
185 : 0 : close(fd);
186 : 0 : return NULL;
187 : : }
188 : :
189 : 2 : lz4bufsize = LZ4F_compressBound(LZ4_IN_SIZE, NULL);
190 : 2 : lz4buf = pg_malloc0(lz4bufsize);
191 : :
192 : : /* assign the compression level, default is 0 */
193 : 2 : memset(&prefs, 0, sizeof(prefs));
194 : 2 : prefs.compressionLevel = wwmethod->compression_level;
195 : :
196 : : /* add the header */
197 : 2 : header_size = LZ4F_compressBegin(ctx, lz4buf, lz4bufsize, &prefs);
198 [ - + ]: 2 : if (LZ4F_isError(header_size))
199 : : {
200 : 0 : wwmethod->lasterrstring = LZ4F_getErrorName(header_size);
201 : 0 : (void) LZ4F_freeCompressionContext(ctx);
202 : 0 : pg_free(lz4buf);
203 : 0 : close(fd);
204 : 0 : return NULL;
205 : : }
206 : :
207 : 2 : errno = 0;
208 [ - + ]: 2 : if (write(fd, lz4buf, header_size) != header_size)
209 : : {
210 : : /* If write didn't set errno, assume problem is no disk space */
211 [ # # ]: 0 : wwmethod->lasterrno = errno ? errno : ENOSPC;
212 : 0 : (void) LZ4F_freeCompressionContext(ctx);
213 : 0 : pg_free(lz4buf);
214 : 0 : close(fd);
215 : 0 : return NULL;
216 : : }
217 : : }
218 : : #endif
219 : :
220 : : /* Do pre-padding on non-compressed files */
221 [ + + + + ]: 163 : if (pad_to_size && wwmethod->compression_algorithm == PG_COMPRESSION_NONE)
222 : : {
223 : : ssize_t rc;
224 : :
225 : 145 : rc = pg_pwrite_zeros(fd, pad_to_size, 0);
226 : :
227 [ - + ]: 145 : if (rc < 0)
228 : : {
229 : 0 : wwmethod->lasterrno = errno;
230 : 0 : close(fd);
231 : 0 : return NULL;
232 : : }
233 : :
234 : : /*
235 : : * pg_pwrite() (called via pg_pwrite_zeros()) may have moved the file
236 : : * position, so reset it (see win32pwrite.c).
237 : : */
238 [ - + ]: 145 : if (lseek(fd, 0, SEEK_SET) != 0)
239 : : {
240 : 0 : wwmethod->lasterrno = errno;
241 : 0 : close(fd);
242 : 0 : return NULL;
243 : : }
244 : : }
245 : :
246 : : /*
247 : : * fsync WAL file and containing directory, to ensure the file is
248 : : * persistently created and zeroed (if padded). That's particularly
249 : : * important when using synchronous mode, where the file is modified and
250 : : * fsynced in-place, without a directory fsync.
251 : : */
252 [ + + ]: 163 : if (wwmethod->sync)
253 : : {
254 [ + - - + ]: 14 : if (fsync_fname(tmppath, false) != 0 ||
255 : 7 : fsync_parent_path(tmppath) != 0)
256 : : {
257 : 0 : wwmethod->lasterrno = errno;
258 : : #ifdef HAVE_LIBZ
259 [ # # ]: 0 : if (wwmethod->compression_algorithm == PG_COMPRESSION_GZIP)
260 : 0 : gzclose(gzfp);
261 : : else
262 : : #endif
263 : : #ifdef USE_LZ4
264 [ # # ]: 0 : if (wwmethod->compression_algorithm == PG_COMPRESSION_LZ4)
265 : : {
266 : 0 : (void) LZ4F_compressEnd(ctx, lz4buf, lz4bufsize, NULL);
267 : 0 : (void) LZ4F_freeCompressionContext(ctx);
268 : 0 : pg_free(lz4buf);
269 : 0 : close(fd);
270 : : }
271 : : else
272 : : #endif
273 : 0 : close(fd);
274 : 0 : return NULL;
275 : : }
276 : : }
277 : :
278 : 163 : f = pg_malloc0_object(DirectoryMethodFile);
279 : : #ifdef HAVE_LIBZ
280 [ + + ]: 163 : if (wwmethod->compression_algorithm == PG_COMPRESSION_GZIP)
281 : 2 : f->gzfp = gzfp;
282 : : #endif
283 : : #ifdef USE_LZ4
284 [ + + ]: 163 : if (wwmethod->compression_algorithm == PG_COMPRESSION_LZ4)
285 : : {
286 : 2 : f->ctx = ctx;
287 : 2 : f->lz4buf = lz4buf;
288 : 2 : f->lz4bufsize = lz4bufsize;
289 : : }
290 : : #endif
291 : :
292 : 163 : f->base.wwmethod = wwmethod;
293 : 163 : f->base.currpos = 0;
294 : 163 : f->base.pathname = pg_strdup(pathname);
295 : 163 : f->fd = fd;
296 : 163 : f->fullpath = pg_strdup(tmppath);
297 [ + + ]: 163 : if (temp_suffix)
298 : 16 : f->temp_suffix = pg_strdup(temp_suffix);
299 : :
300 : 163 : return &f->base;
301 : : }
302 : :
303 : : static ssize_t
304 : 1574 : dir_write(Walfile *f, const void *buf, size_t count)
305 : : {
306 : : ssize_t r;
307 : 1574 : DirectoryMethodFile *df = (DirectoryMethodFile *) f;
308 : :
309 : : Assert(f != NULL);
310 : 1574 : clear_error(f->wwmethod);
311 : :
312 : : #ifdef HAVE_LIBZ
313 [ + + ]: 1574 : if (f->wwmethod->compression_algorithm == PG_COMPRESSION_GZIP)
314 : : {
315 : 9 : errno = 0;
316 : 9 : r = (ssize_t) gzwrite(df->gzfp, buf, count);
317 [ - + ]: 9 : if (r != count)
318 : : {
319 : : /* If write didn't set errno, assume problem is no disk space */
320 [ # # ]: 0 : f->wwmethod->lasterrno = errno ? errno : ENOSPC;
321 : : }
322 : : }
323 : : else
324 : : #endif
325 : : #ifdef USE_LZ4
326 [ + + ]: 1565 : if (f->wwmethod->compression_algorithm == PG_COMPRESSION_LZ4)
327 : : {
328 : : size_t chunk;
329 : : size_t remaining;
330 : 9 : const void *inbuf = buf;
331 : :
332 : 9 : remaining = count;
333 [ + + ]: 266 : while (remaining > 0)
334 : : {
335 : : size_t compressed;
336 : :
337 [ + + ]: 257 : if (remaining > LZ4_IN_SIZE)
338 : 248 : chunk = LZ4_IN_SIZE;
339 : : else
340 : 9 : chunk = remaining;
341 : :
342 : 257 : remaining -= chunk;
343 : 257 : compressed = LZ4F_compressUpdate(df->ctx,
344 : : df->lz4buf, df->lz4bufsize,
345 : : inbuf, chunk,
346 : : NULL);
347 : :
348 [ - + ]: 257 : if (LZ4F_isError(compressed))
349 : : {
350 : 0 : f->wwmethod->lasterrstring = LZ4F_getErrorName(compressed);
351 : 0 : return -1;
352 : : }
353 : :
354 : 257 : errno = 0;
355 [ - + ]: 257 : if (write(df->fd, df->lz4buf, compressed) != compressed)
356 : : {
357 : : /* If write didn't set errno, assume problem is no disk space */
358 [ # # ]: 0 : f->wwmethod->lasterrno = errno ? errno : ENOSPC;
359 : 0 : return -1;
360 : : }
361 : :
362 : 257 : inbuf = ((const char *) inbuf) + chunk;
363 : : }
364 : :
365 : : /* Our caller keeps track of the uncompressed size. */
366 : 9 : r = (ssize_t) count;
367 : : }
368 : : else
369 : : #endif
370 : : {
371 : 1556 : errno = 0;
372 : 1556 : r = write(df->fd, buf, count);
373 [ - + ]: 1556 : if (r != count)
374 : : {
375 : : /* If write didn't set errno, assume problem is no disk space */
376 [ # # ]: 0 : f->wwmethod->lasterrno = errno ? errno : ENOSPC;
377 : : }
378 : : }
379 [ + - ]: 1574 : if (r > 0)
380 : 1574 : df->base.currpos += r;
381 : 1574 : return r;
382 : : }
383 : :
384 : : static int
385 : 163 : dir_close(Walfile *f, WalCloseMethod method)
386 : : {
387 : : int r;
388 : 163 : DirectoryMethodFile *df = (DirectoryMethodFile *) f;
389 : 163 : DirectoryMethodData *dir_data = (DirectoryMethodData *) f->wwmethod;
390 : : char tmppath[MAXPGPATH];
391 : : char tmppath2[MAXPGPATH];
392 : :
393 : : Assert(f != NULL);
394 : 163 : clear_error(f->wwmethod);
395 : :
396 : : #ifdef HAVE_LIBZ
397 [ + + ]: 163 : if (f->wwmethod->compression_algorithm == PG_COMPRESSION_GZIP)
398 : : {
399 : 2 : errno = 0; /* in case gzclose() doesn't set it */
400 : 2 : r = gzclose(df->gzfp);
401 : : }
402 : : else
403 : : #endif
404 : : #ifdef USE_LZ4
405 [ + + ]: 161 : if (f->wwmethod->compression_algorithm == PG_COMPRESSION_LZ4)
406 : : {
407 : : size_t compressed;
408 : :
409 : 2 : compressed = LZ4F_compressEnd(df->ctx,
410 : : df->lz4buf, df->lz4bufsize,
411 : : NULL);
412 : :
413 [ - + ]: 2 : if (LZ4F_isError(compressed))
414 : : {
415 : 0 : f->wwmethod->lasterrstring = LZ4F_getErrorName(compressed);
416 : 0 : return -1;
417 : : }
418 : :
419 : 2 : errno = 0;
420 [ - + ]: 2 : if (write(df->fd, df->lz4buf, compressed) != compressed)
421 : : {
422 : : /* If write didn't set errno, assume problem is no disk space */
423 [ # # ]: 0 : f->wwmethod->lasterrno = errno ? errno : ENOSPC;
424 : 0 : return -1;
425 : : }
426 : :
427 : 2 : r = close(df->fd);
428 : : }
429 : : else
430 : : #endif
431 : 159 : r = close(df->fd);
432 : :
433 [ + - ]: 163 : if (r == 0)
434 : : {
435 : : /* Build path to the current version of the file */
436 [ + + + + ]: 163 : if (method == CLOSE_NORMAL && df->temp_suffix)
437 : 10 : {
438 : : char *filename;
439 : : char *filename2;
440 : :
441 : : /*
442 : : * If we have a temp prefix, normal operation is to rename the
443 : : * file.
444 : : */
445 : 10 : filename = dir_get_file_name(f->wwmethod, df->base.pathname,
446 : 10 : df->temp_suffix);
447 : 10 : snprintf(tmppath, sizeof(tmppath), "%s/%s",
448 : : dir_data->basedir, filename);
449 : 10 : pg_free(filename);
450 : :
451 : : /* permanent name, so no need for the prefix */
452 : 10 : filename2 = dir_get_file_name(f->wwmethod, df->base.pathname, NULL);
453 : 10 : snprintf(tmppath2, sizeof(tmppath2), "%s/%s",
454 : : dir_data->basedir, filename2);
455 : 10 : pg_free(filename2);
456 [ + + ]: 10 : if (f->wwmethod->sync)
457 : 3 : r = durable_rename(tmppath, tmppath2);
458 : : else
459 : : {
460 [ - + ]: 7 : if (rename(tmppath, tmppath2) != 0)
461 : : {
462 : 0 : pg_log_error("could not rename file \"%s\" to \"%s\": %m",
463 : : tmppath, tmppath2);
464 : 0 : r = -1;
465 : : }
466 : : }
467 : : }
468 [ - + ]: 153 : else if (method == CLOSE_UNLINK)
469 : : {
470 : : char *filename;
471 : :
472 : : /* Unlink the file once it's closed */
473 : 0 : filename = dir_get_file_name(f->wwmethod, df->base.pathname,
474 : 0 : df->temp_suffix);
475 : 0 : snprintf(tmppath, sizeof(tmppath), "%s/%s",
476 : : dir_data->basedir, filename);
477 : 0 : pg_free(filename);
478 : 0 : r = unlink(tmppath);
479 : : }
480 : : else
481 : : {
482 : : /*
483 : : * Else either CLOSE_NORMAL and no temp suffix, or
484 : : * CLOSE_NO_RENAME. In this case, fsync the file and containing
485 : : * directory if sync mode is requested.
486 : : */
487 [ + + ]: 153 : if (f->wwmethod->sync)
488 : : {
489 : 4 : r = fsync_fname(df->fullpath, false);
490 [ + - ]: 4 : if (r == 0)
491 : 4 : r = fsync_parent_path(df->fullpath);
492 : : }
493 : : }
494 : : }
495 : :
496 [ - + ]: 163 : if (r != 0)
497 : 0 : f->wwmethod->lasterrno = errno;
498 : :
499 : : #ifdef USE_LZ4
500 : 163 : pg_free(df->lz4buf);
501 : : /* supports free on NULL */
502 : 163 : LZ4F_freeCompressionContext(df->ctx);
503 : : #endif
504 : :
505 : 163 : pg_free(df->base.pathname);
506 : 163 : pg_free(df->fullpath);
507 : 163 : pg_free(df->temp_suffix);
508 : 163 : pg_free(df);
509 : :
510 : 163 : return r;
511 : : }
512 : :
513 : : static int
514 : 0 : dir_sync(Walfile *f)
515 : : {
516 : : int r;
517 : :
518 : : Assert(f != NULL);
519 : 0 : clear_error(f->wwmethod);
520 : :
521 [ # # ]: 0 : if (!f->wwmethod->sync)
522 : 0 : return 0;
523 : :
524 : : #ifdef HAVE_LIBZ
525 [ # # ]: 0 : if (f->wwmethod->compression_algorithm == PG_COMPRESSION_GZIP)
526 : : {
527 [ # # ]: 0 : if (gzflush(((DirectoryMethodFile *) f)->gzfp, Z_SYNC_FLUSH) != Z_OK)
528 : : {
529 : 0 : f->wwmethod->lasterrno = errno;
530 : 0 : return -1;
531 : : }
532 : : }
533 : : #endif
534 : : #ifdef USE_LZ4
535 [ # # ]: 0 : if (f->wwmethod->compression_algorithm == PG_COMPRESSION_LZ4)
536 : : {
537 : 0 : DirectoryMethodFile *df = (DirectoryMethodFile *) f;
538 : : size_t compressed;
539 : :
540 : : /* Flush any internal buffers */
541 : 0 : compressed = LZ4F_flush(df->ctx, df->lz4buf, df->lz4bufsize, NULL);
542 [ # # ]: 0 : if (LZ4F_isError(compressed))
543 : : {
544 : 0 : f->wwmethod->lasterrstring = LZ4F_getErrorName(compressed);
545 : 0 : return -1;
546 : : }
547 : :
548 : 0 : errno = 0;
549 [ # # ]: 0 : if (write(df->fd, df->lz4buf, compressed) != compressed)
550 : : {
551 : : /* If write didn't set errno, assume problem is no disk space */
552 [ # # ]: 0 : f->wwmethod->lasterrno = errno ? errno : ENOSPC;
553 : 0 : return -1;
554 : : }
555 : : }
556 : : #endif
557 : :
558 : 0 : r = fsync(((DirectoryMethodFile *) f)->fd);
559 [ # # ]: 0 : if (r < 0)
560 : 0 : f->wwmethod->lasterrno = errno;
561 : 0 : return r;
562 : : }
563 : :
564 : : static ssize_t
565 : 0 : dir_get_file_size(WalWriteMethod *wwmethod, const char *pathname)
566 : : {
567 : 0 : DirectoryMethodData *dir_data = (DirectoryMethodData *) wwmethod;
568 : : struct stat statbuf;
569 : : char tmppath[MAXPGPATH];
570 : :
571 : 0 : snprintf(tmppath, sizeof(tmppath), "%s/%s",
572 : : dir_data->basedir, pathname);
573 : :
574 [ # # ]: 0 : if (stat(tmppath, &statbuf) != 0)
575 : : {
576 : 0 : wwmethod->lasterrno = errno;
577 : 0 : return -1;
578 : : }
579 : :
580 : 0 : return statbuf.st_size;
581 : : }
582 : :
583 : : static bool
584 : 149 : dir_existsfile(WalWriteMethod *wwmethod, const char *pathname)
585 : : {
586 : 149 : DirectoryMethodData *dir_data = (DirectoryMethodData *) wwmethod;
587 : : char tmppath[MAXPGPATH];
588 : : int fd;
589 : :
590 : 149 : clear_error(wwmethod);
591 : :
592 : 149 : snprintf(tmppath, sizeof(tmppath), "%s/%s",
593 : : dir_data->basedir, pathname);
594 : :
595 : 149 : fd = open(tmppath, O_RDONLY | PG_BINARY, 0);
596 [ + - ]: 149 : if (fd < 0)
597 : :
598 : : /*
599 : : * Skip setting dir_data->lasterrno here because we are only checking
600 : : * for existence.
601 : : */
602 : 149 : return false;
603 : 0 : close(fd);
604 : 0 : return true;
605 : : }
606 : :
607 : : static bool
608 : 141 : dir_finish(WalWriteMethod *wwmethod)
609 : : {
610 : 141 : clear_error(wwmethod);
611 : :
612 [ + + ]: 141 : if (wwmethod->sync)
613 : : {
614 : 4 : DirectoryMethodData *dir_data = (DirectoryMethodData *) wwmethod;
615 : :
616 : : /*
617 : : * Files are fsynced when they are closed, but we need to fsync the
618 : : * directory entry here as well.
619 : : */
620 [ - + ]: 4 : if (fsync_fname(dir_data->basedir, true) != 0)
621 : : {
622 : 0 : wwmethod->lasterrno = errno;
623 : 0 : return false;
624 : : }
625 : : }
626 : 141 : return true;
627 : : }
628 : :
629 : : static void
630 : 141 : dir_free(WalWriteMethod *wwmethod)
631 : : {
632 : 141 : DirectoryMethodData *dir_data = (DirectoryMethodData *) wwmethod;
633 : :
634 : 141 : pg_free(dir_data->basedir);
635 : 141 : pg_free(wwmethod);
636 : 141 : }
637 : :
638 : :
639 : : WalWriteMethod *
640 : 143 : CreateWalDirectoryMethod(const char *basedir,
641 : : pg_compress_algorithm compression_algorithm,
642 : : int compression_level, bool sync)
643 : : {
644 : : DirectoryMethodData *wwmethod;
645 : :
646 : 143 : wwmethod = pg_malloc0_object(DirectoryMethodData);
647 : 143 : *((const WalWriteMethodOps **) &wwmethod->base.ops) =
648 : : &WalDirectoryMethodOps;
649 : 143 : wwmethod->base.compression_algorithm = compression_algorithm;
650 : 143 : wwmethod->base.compression_level = compression_level;
651 : 143 : wwmethod->base.sync = sync;
652 : 143 : clear_error(&wwmethod->base);
653 : 143 : wwmethod->basedir = pg_strdup(basedir);
654 : :
655 : 143 : return &wwmethod->base;
656 : : }
657 : :
658 : :
659 : : /*-------------------------------------------------------------------------
660 : : * WalTarMethod - write wal to a tar file containing pg_wal contents
661 : : *-------------------------------------------------------------------------
662 : : */
663 : :
664 : : static Walfile *tar_open_for_write(WalWriteMethod *wwmethod,
665 : : const char *pathname,
666 : : const char *temp_suffix,
667 : : size_t pad_to_size);
668 : : static int tar_close(Walfile *f, WalCloseMethod method);
669 : : static bool tar_existsfile(WalWriteMethod *wwmethod, const char *pathname);
670 : : static ssize_t tar_get_file_size(WalWriteMethod *wwmethod,
671 : : const char *pathname);
672 : : static char *tar_get_file_name(WalWriteMethod *wwmethod,
673 : : const char *pathname, const char *temp_suffix);
674 : : static ssize_t tar_write(Walfile *f, const void *buf, size_t count);
675 : : static int tar_sync(Walfile *f);
676 : : static bool tar_finish(WalWriteMethod *wwmethod);
677 : : static void tar_free(WalWriteMethod *wwmethod);
678 : :
679 : : static const WalWriteMethodOps WalTarMethodOps = {
680 : : .open_for_write = tar_open_for_write,
681 : : .close = tar_close,
682 : : .existsfile = tar_existsfile,
683 : : .get_file_size = tar_get_file_size,
684 : : .get_file_name = tar_get_file_name,
685 : : .write = tar_write,
686 : : .sync = tar_sync,
687 : : .finish = tar_finish,
688 : : .free = tar_free
689 : : };
690 : :
691 : : typedef struct TarMethodFile
692 : : {
693 : : Walfile base;
694 : : pgoff_t ofs_start; /* Where does the *header* for this file start */
695 : : char header[TAR_BLOCK_SIZE];
696 : : size_t pad_to_size;
697 : : } TarMethodFile;
698 : :
699 : : typedef struct TarMethodData
700 : : {
701 : : WalWriteMethod base;
702 : : char *tarfilename;
703 : : int fd;
704 : : TarMethodFile *currentfile;
705 : : #ifdef HAVE_LIBZ
706 : : z_streamp zp;
707 : : void *zlibOut;
708 : : #endif
709 : : } TarMethodData;
710 : :
711 : : #ifdef HAVE_LIBZ
712 : : static bool
713 : 6042 : tar_write_compressed_data(TarMethodData *tar_data, const void *buf, size_t count,
714 : : bool flush)
715 : : {
716 : 6042 : tar_data->zp->next_in = buf;
717 : 6042 : tar_data->zp->avail_in = count;
718 : :
719 [ + + + + ]: 12098 : while (tar_data->zp->avail_in || flush)
720 : : {
721 : : int r;
722 : :
723 [ + + ]: 6068 : r = deflate(tar_data->zp, flush ? Z_FINISH : Z_NO_FLUSH);
724 [ - + ]: 6068 : if (r == Z_STREAM_ERROR)
725 : : {
726 : 0 : tar_data->base.lasterrstring = _("could not compress data");
727 : 0 : return false;
728 : : }
729 : :
730 [ + + ]: 6068 : if (tar_data->zp->avail_out < ZLIB_OUT_SIZE)
731 : : {
732 : 62 : size_t len = ZLIB_OUT_SIZE - tar_data->zp->avail_out;
733 : :
734 : 62 : errno = 0;
735 [ - + ]: 62 : if (write(tar_data->fd, tar_data->zlibOut, len) != len)
736 : : {
737 : : /* If write didn't set errno, assume problem is no disk space */
738 [ # # ]: 0 : tar_data->base.lasterrno = errno ? errno : ENOSPC;
739 : 0 : return false;
740 : : }
741 : :
742 : 62 : tar_data->zp->next_out = tar_data->zlibOut;
743 : 62 : tar_data->zp->avail_out = ZLIB_OUT_SIZE;
744 : : }
745 : :
746 [ + + ]: 6068 : if (r == Z_STREAM_END)
747 : 12 : break;
748 : : }
749 : :
750 [ + + ]: 6042 : if (flush)
751 : : {
752 : : /* Reset the stream for writing */
753 [ - + ]: 12 : if (deflateReset(tar_data->zp) != Z_OK)
754 : : {
755 : 0 : tar_data->base.lasterrstring = _("could not reset compression stream");
756 : 0 : return false;
757 : : }
758 : : }
759 : :
760 : 6042 : return true;
761 : : }
762 : : #endif
763 : :
764 : : static ssize_t
765 : 28617 : tar_write(Walfile *f, const void *buf, size_t count)
766 : : {
767 : 28617 : TarMethodData *tar_data = (TarMethodData *) f->wwmethod;
768 : : ssize_t r;
769 : :
770 : : Assert(f != NULL);
771 : 28617 : clear_error(f->wwmethod);
772 : :
773 : : /* Tarfile will always be positioned at the end */
774 [ + + ]: 28617 : if (f->wwmethod->compression_algorithm == PG_COMPRESSION_NONE)
775 : : {
776 : 22590 : errno = 0;
777 : 22590 : r = write(tar_data->fd, buf, count);
778 [ - + ]: 22590 : if (r != count)
779 : : {
780 : : /* If write didn't set errno, assume problem is no disk space */
781 [ # # ]: 0 : f->wwmethod->lasterrno = errno ? errno : ENOSPC;
782 : 0 : return -1;
783 : : }
784 : 22590 : f->currpos += r;
785 : 22590 : return r;
786 : : }
787 : : #ifdef HAVE_LIBZ
788 [ + - ]: 6027 : else if (f->wwmethod->compression_algorithm == PG_COMPRESSION_GZIP)
789 : : {
790 [ - + ]: 6027 : if (!tar_write_compressed_data(tar_data, buf, count, false))
791 : 0 : return -1;
792 : 6027 : f->currpos += count;
793 : 6027 : return count;
794 : : }
795 : : #endif
796 : : else
797 : : {
798 : : /* Can't happen - compression enabled with no method set */
799 : 0 : f->wwmethod->lasterrno = ENOSYS;
800 : 0 : return -1;
801 : : }
802 : : }
803 : :
804 : : static bool
805 : 14 : tar_write_padding_data(TarMethodFile *f, size_t bytes)
806 : : {
807 : : PGAlignedXLogBlock zerobuf;
808 : 14 : size_t bytesleft = bytes;
809 : :
810 : 14 : memset(zerobuf.data, 0, XLOG_BLCKSZ);
811 [ + + ]: 28558 : while (bytesleft)
812 : : {
813 : 28544 : size_t bytestowrite = Min(bytesleft, XLOG_BLCKSZ);
814 : 28544 : ssize_t r = tar_write(&f->base, zerobuf.data, bytestowrite);
815 : :
816 [ - + ]: 28544 : if (r < 0)
817 : 0 : return false;
818 : 28544 : bytesleft -= r;
819 : : }
820 : :
821 : 14 : return true;
822 : : }
823 : :
824 : : static char *
825 : 44 : tar_get_file_name(WalWriteMethod *wwmethod, const char *pathname,
826 : : const char *temp_suffix)
827 : : {
828 : 44 : char *filename = pg_malloc0_array(char, MAXPGPATH);
829 : :
830 [ + + ]: 44 : snprintf(filename, MAXPGPATH, "%s%s",
831 : : pathname, temp_suffix ? temp_suffix : "");
832 : :
833 : 44 : return filename;
834 : : }
835 : :
836 : : static Walfile *
837 : 16 : tar_open_for_write(WalWriteMethod *wwmethod, const char *pathname,
838 : : const char *temp_suffix, size_t pad_to_size)
839 : : {
840 : 16 : TarMethodData *tar_data = (TarMethodData *) wwmethod;
841 : : char *tmppath;
842 : :
843 : 16 : clear_error(wwmethod);
844 : :
845 [ + + ]: 16 : if (tar_data->fd < 0)
846 : : {
847 : : /*
848 : : * We open the tar file only when we first try to write to it.
849 : : */
850 : 14 : tar_data->fd = open(tar_data->tarfilename,
851 : : O_WRONLY | O_CREAT | PG_BINARY,
852 : : pg_file_create_mode);
853 [ - + ]: 14 : if (tar_data->fd < 0)
854 : : {
855 : 0 : wwmethod->lasterrno = errno;
856 : 0 : return NULL;
857 : : }
858 : :
859 : : #ifdef HAVE_LIBZ
860 [ + + ]: 14 : if (wwmethod->compression_algorithm == PG_COMPRESSION_GZIP)
861 : : {
862 : 3 : tar_data->zp = pg_malloc_object(z_stream);
863 : 3 : tar_data->zp->zalloc = Z_NULL;
864 : 3 : tar_data->zp->zfree = Z_NULL;
865 : 3 : tar_data->zp->opaque = Z_NULL;
866 : 3 : tar_data->zp->next_out = tar_data->zlibOut;
867 : 3 : tar_data->zp->avail_out = ZLIB_OUT_SIZE;
868 : :
869 : : /*
870 : : * Initialize deflation library. Adding the magic value 16 to the
871 : : * default 15 for the windowBits parameter makes the output be
872 : : * gzip instead of zlib.
873 : : */
874 [ - + ]: 3 : if (deflateInit2(tar_data->zp, wwmethod->compression_level,
875 : : Z_DEFLATED, 15 + 16, 8, Z_DEFAULT_STRATEGY) != Z_OK)
876 : : {
877 : 0 : pg_free(tar_data->zp);
878 : 0 : tar_data->zp = NULL;
879 : 0 : wwmethod->lasterrstring =
880 : 0 : _("could not initialize compression library");
881 : 0 : return NULL;
882 : : }
883 : : }
884 : : #endif
885 : :
886 : : /* There's no tar header itself, the file starts with regular files */
887 : : }
888 : :
889 [ - + ]: 16 : if (tar_data->currentfile != NULL)
890 : : {
891 : 0 : wwmethod->lasterrstring =
892 : 0 : _("implementation error: tar files can't have more than one open file");
893 : 0 : return NULL;
894 : : }
895 : :
896 : 16 : tar_data->currentfile = pg_malloc0_object(TarMethodFile);
897 : 16 : tar_data->currentfile->base.wwmethod = wwmethod;
898 : :
899 : 16 : tmppath = tar_get_file_name(wwmethod, pathname, temp_suffix);
900 : :
901 : : /* Create a header with size set to 0 - we will fill out the size on close */
902 [ - + ]: 16 : if (tarCreateHeader(tar_data->currentfile->header, tmppath, NULL, 0, S_IRUSR | S_IWUSR, 0, 0, time(NULL)) != TAR_OK)
903 : : {
904 : 0 : pg_free(tar_data->currentfile);
905 : 0 : pg_free(tmppath);
906 : 0 : tar_data->currentfile = NULL;
907 : 0 : wwmethod->lasterrstring = _("could not create tar header");
908 : 0 : return NULL;
909 : : }
910 : :
911 : 16 : pg_free(tmppath);
912 : :
913 : : #ifdef HAVE_LIBZ
914 [ + + ]: 16 : if (wwmethod->compression_algorithm == PG_COMPRESSION_GZIP)
915 : : {
916 : : /* Flush existing data */
917 [ - + ]: 3 : if (!tar_write_compressed_data(tar_data, NULL, 0, true))
918 : 0 : return NULL;
919 : :
920 : : /* Turn off compression for header */
921 [ - + ]: 3 : if (deflateParams(tar_data->zp, 0, Z_DEFAULT_STRATEGY) != Z_OK)
922 : : {
923 : 0 : wwmethod->lasterrstring =
924 : 0 : _("could not change compression parameters");
925 : 0 : return NULL;
926 : : }
927 : : }
928 : : #endif
929 : :
930 : 16 : tar_data->currentfile->ofs_start = lseek(tar_data->fd, 0, SEEK_CUR);
931 [ - + ]: 16 : if (tar_data->currentfile->ofs_start == -1)
932 : : {
933 : 0 : wwmethod->lasterrno = errno;
934 : 0 : pg_free(tar_data->currentfile);
935 : 0 : tar_data->currentfile = NULL;
936 : 0 : return NULL;
937 : : }
938 : 16 : tar_data->currentfile->base.currpos = 0;
939 : :
940 [ + + ]: 16 : if (wwmethod->compression_algorithm == PG_COMPRESSION_NONE)
941 : : {
942 : 13 : errno = 0;
943 [ - + ]: 13 : if (write(tar_data->fd, tar_data->currentfile->header,
944 : : TAR_BLOCK_SIZE) != TAR_BLOCK_SIZE)
945 : : {
946 : : /* If write didn't set errno, assume problem is no disk space */
947 [ # # ]: 0 : wwmethod->lasterrno = errno ? errno : ENOSPC;
948 : 0 : pg_free(tar_data->currentfile);
949 : 0 : tar_data->currentfile = NULL;
950 : 0 : return NULL;
951 : : }
952 : : }
953 : : #ifdef HAVE_LIBZ
954 [ + - ]: 3 : else if (wwmethod->compression_algorithm == PG_COMPRESSION_GZIP)
955 : : {
956 : : /* Write header through the zlib APIs but with no compression */
957 [ - + ]: 3 : if (!tar_write_compressed_data(tar_data, tar_data->currentfile->header,
958 : : TAR_BLOCK_SIZE, true))
959 : 0 : return NULL;
960 : :
961 : : /* Re-enable compression for the rest of the file */
962 [ - + ]: 3 : if (deflateParams(tar_data->zp, wwmethod->compression_level,
963 : : Z_DEFAULT_STRATEGY) != Z_OK)
964 : : {
965 : 0 : wwmethod->lasterrstring = _("could not change compression parameters");
966 : 0 : return NULL;
967 : : }
968 : : }
969 : : #endif
970 : : else
971 : : {
972 : : /* not reachable */
973 : : Assert(false);
974 : : }
975 : :
976 : 16 : tar_data->currentfile->base.pathname = pg_strdup(pathname);
977 : :
978 : : /*
979 : : * Uncompressed files are padded on creation, but for compression we can't
980 : : * do that
981 : : */
982 [ + + ]: 16 : if (pad_to_size)
983 : : {
984 : 14 : tar_data->currentfile->pad_to_size = pad_to_size;
985 [ + + ]: 14 : if (wwmethod->compression_algorithm == PG_COMPRESSION_NONE)
986 : : {
987 : : /* Uncompressed, so pad now */
988 [ - + ]: 11 : if (!tar_write_padding_data(tar_data->currentfile, pad_to_size))
989 : 0 : return NULL;
990 : : /* Seek back to start */
991 : 22 : if (lseek(tar_data->fd,
992 : 11 : tar_data->currentfile->ofs_start + TAR_BLOCK_SIZE,
993 [ - + ]: 11 : SEEK_SET) != tar_data->currentfile->ofs_start + TAR_BLOCK_SIZE)
994 : : {
995 : 0 : wwmethod->lasterrno = errno;
996 : 0 : return NULL;
997 : : }
998 : :
999 : 11 : tar_data->currentfile->base.currpos = 0;
1000 : : }
1001 : : }
1002 : :
1003 : 16 : return &tar_data->currentfile->base;
1004 : : }
1005 : :
1006 : : static ssize_t
1007 : 0 : tar_get_file_size(WalWriteMethod *wwmethod, const char *pathname)
1008 : : {
1009 : 0 : clear_error(wwmethod);
1010 : :
1011 : : /* Currently not used, so not supported */
1012 : 0 : wwmethod->lasterrno = ENOSYS;
1013 : 0 : return -1;
1014 : : }
1015 : :
1016 : : static int
1017 : 16 : tar_sync(Walfile *f)
1018 : : {
1019 : 16 : TarMethodData *tar_data = (TarMethodData *) f->wwmethod;
1020 : : int r;
1021 : :
1022 : : Assert(f != NULL);
1023 : 16 : clear_error(f->wwmethod);
1024 : :
1025 [ + - ]: 16 : if (!f->wwmethod->sync)
1026 : 16 : return 0;
1027 : :
1028 : : /*
1029 : : * Always sync the whole tarfile, because that's all we can do. This makes
1030 : : * no sense on compressed files, so just ignore those.
1031 : : */
1032 [ # # ]: 0 : if (f->wwmethod->compression_algorithm != PG_COMPRESSION_NONE)
1033 : 0 : return 0;
1034 : :
1035 : 0 : r = fsync(tar_data->fd);
1036 [ # # ]: 0 : if (r < 0)
1037 : 0 : f->wwmethod->lasterrno = errno;
1038 : 0 : return r;
1039 : : }
1040 : :
1041 : : static int
1042 : 16 : tar_close(Walfile *f, WalCloseMethod method)
1043 : : {
1044 : : ssize_t filesize;
1045 : : int padding;
1046 : 16 : TarMethodData *tar_data = (TarMethodData *) f->wwmethod;
1047 : 16 : TarMethodFile *tf = (TarMethodFile *) f;
1048 : :
1049 : : Assert(f != NULL);
1050 : 16 : clear_error(f->wwmethod);
1051 : :
1052 [ - + ]: 16 : if (method == CLOSE_UNLINK)
1053 : : {
1054 [ # # ]: 0 : if (f->wwmethod->compression_algorithm != PG_COMPRESSION_NONE)
1055 : : {
1056 : 0 : f->wwmethod->lasterrstring = _("unlink not supported with compression");
1057 : 0 : return -1;
1058 : : }
1059 : :
1060 : : /*
1061 : : * Unlink the file that we just wrote to the tar. We do this by
1062 : : * truncating it to the start of the header. This is safe as we only
1063 : : * allow writing of the very last file.
1064 : : */
1065 [ # # ]: 0 : if (ftruncate(tar_data->fd, tf->ofs_start) != 0)
1066 : : {
1067 : 0 : f->wwmethod->lasterrno = errno;
1068 : 0 : return -1;
1069 : : }
1070 : :
1071 : 0 : pg_free(tf->base.pathname);
1072 : 0 : pg_free(tf);
1073 : 0 : tar_data->currentfile = NULL;
1074 : :
1075 : 0 : return 0;
1076 : : }
1077 : :
1078 : : /*
1079 : : * Pad the file itself with zeroes if necessary. Note that this is
1080 : : * different from the tar format padding -- this is the padding we asked
1081 : : * for when the file was opened.
1082 : : */
1083 [ + + ]: 16 : if (tf->pad_to_size)
1084 : : {
1085 [ + + ]: 14 : if (f->wwmethod->compression_algorithm == PG_COMPRESSION_GZIP)
1086 : : {
1087 : : /*
1088 : : * A compressed tarfile is padded on close since we cannot know
1089 : : * the size of the compressed output until the end.
1090 : : */
1091 : 3 : size_t sizeleft = tf->pad_to_size - tf->base.currpos;
1092 : :
1093 [ + - ]: 3 : if (sizeleft)
1094 : : {
1095 [ - + ]: 3 : if (!tar_write_padding_data(tf, sizeleft))
1096 : 0 : return -1;
1097 : : }
1098 : : }
1099 : : else
1100 : : {
1101 : : /*
1102 : : * An uncompressed tarfile was padded on creation, so just adjust
1103 : : * the current position as if we seeked to the end.
1104 : : */
1105 : 11 : tf->base.currpos = tf->pad_to_size;
1106 : : }
1107 : : }
1108 : :
1109 : : /*
1110 : : * Get the size of the file, and pad out to a multiple of the tar block
1111 : : * size.
1112 : : */
1113 : 16 : filesize = f->currpos;
1114 : 16 : padding = tarPaddingBytesRequired(filesize);
1115 [ + + ]: 16 : if (padding)
1116 : : {
1117 : 1 : char zerobuf[TAR_BLOCK_SIZE] = {0};
1118 : :
1119 [ - + ]: 1 : if (tar_write(f, zerobuf, padding) != padding)
1120 : 0 : return -1;
1121 : : }
1122 : :
1123 : :
1124 : : #ifdef HAVE_LIBZ
1125 [ + + ]: 16 : if (f->wwmethod->compression_algorithm == PG_COMPRESSION_GZIP)
1126 : : {
1127 : : /* Flush the current buffer */
1128 [ - + ]: 3 : if (!tar_write_compressed_data(tar_data, NULL, 0, true))
1129 : 0 : return -1;
1130 : : }
1131 : : #endif
1132 : :
1133 : : /*
1134 : : * Now go back and update the header with the correct filesize and
1135 : : * possibly also renaming the file. We overwrite the entire current header
1136 : : * when done, including the checksum.
1137 : : */
1138 : 16 : print_tar_number(&(tf->header[TAR_OFFSET_SIZE]), 12, filesize);
1139 : :
1140 [ + - ]: 16 : if (method == CLOSE_NORMAL)
1141 : :
1142 : : /*
1143 : : * We overwrite it with what it was before if we have no tempname,
1144 : : * since we're going to write the buffer anyway.
1145 : : */
1146 : 16 : strlcpy(&(tf->header[TAR_OFFSET_NAME]), tf->base.pathname, 100);
1147 : :
1148 : 16 : print_tar_number(&(tf->header[TAR_OFFSET_CHECKSUM]), 8,
1149 : 16 : tarChecksum(((TarMethodFile *) f)->header));
1150 [ - + ]: 16 : if (lseek(tar_data->fd, tf->ofs_start, SEEK_SET) != ((TarMethodFile *) f)->ofs_start)
1151 : : {
1152 : 0 : f->wwmethod->lasterrno = errno;
1153 : 0 : return -1;
1154 : : }
1155 [ + + ]: 16 : if (f->wwmethod->compression_algorithm == PG_COMPRESSION_NONE)
1156 : : {
1157 : 13 : errno = 0;
1158 [ - + ]: 13 : if (write(tar_data->fd, tf->header, TAR_BLOCK_SIZE) != TAR_BLOCK_SIZE)
1159 : : {
1160 : : /* If write didn't set errno, assume problem is no disk space */
1161 [ # # ]: 0 : f->wwmethod->lasterrno = errno ? errno : ENOSPC;
1162 : 0 : return -1;
1163 : : }
1164 : : }
1165 : : #ifdef HAVE_LIBZ
1166 [ + - ]: 3 : else if (f->wwmethod->compression_algorithm == PG_COMPRESSION_GZIP)
1167 : : {
1168 : : /* Turn off compression */
1169 [ - + ]: 3 : if (deflateParams(tar_data->zp, 0, Z_DEFAULT_STRATEGY) != Z_OK)
1170 : : {
1171 : 0 : f->wwmethod->lasterrstring = _("could not change compression parameters");
1172 : 0 : return -1;
1173 : : }
1174 : :
1175 : : /* Overwrite the header, assuming the size will be the same */
1176 [ - + ]: 3 : if (!tar_write_compressed_data(tar_data, tar_data->currentfile->header,
1177 : : TAR_BLOCK_SIZE, true))
1178 : 0 : return -1;
1179 : :
1180 : : /* Turn compression back on */
1181 [ - + ]: 3 : if (deflateParams(tar_data->zp, f->wwmethod->compression_level,
1182 : : Z_DEFAULT_STRATEGY) != Z_OK)
1183 : : {
1184 : 0 : f->wwmethod->lasterrstring = _("could not change compression parameters");
1185 : 0 : return -1;
1186 : : }
1187 : : }
1188 : : #endif
1189 : : else
1190 : : {
1191 : : /* not reachable */
1192 : : Assert(false);
1193 : : }
1194 : :
1195 : : /* Move file pointer back down to end, so we can write the next file */
1196 [ - + ]: 16 : if (lseek(tar_data->fd, 0, SEEK_END) < 0)
1197 : : {
1198 : 0 : f->wwmethod->lasterrno = errno;
1199 : 0 : return -1;
1200 : : }
1201 : :
1202 : : /* Always fsync on close, so the padding gets fsynced */
1203 [ - + ]: 16 : if (tar_sync(f) < 0)
1204 : : {
1205 : : /* XXX this seems pretty bogus; why is only this case fatal? */
1206 : 0 : pg_fatal("could not fsync file \"%s\": %s",
1207 : : tf->base.pathname, GetLastWalMethodError(f->wwmethod));
1208 : : }
1209 : :
1210 : : /* Clean up and done */
1211 : 16 : pg_free(tf->base.pathname);
1212 : 16 : pg_free(tf);
1213 : 16 : tar_data->currentfile = NULL;
1214 : :
1215 : 16 : return 0;
1216 : : }
1217 : :
1218 : : static bool
1219 : 12 : tar_existsfile(WalWriteMethod *wwmethod, const char *pathname)
1220 : : {
1221 : 12 : clear_error(wwmethod);
1222 : : /* We only deal with new tarfiles, so nothing externally created exists */
1223 : 12 : return false;
1224 : : }
1225 : :
1226 : : static bool
1227 : 14 : tar_finish(WalWriteMethod *wwmethod)
1228 : : {
1229 : 14 : TarMethodData *tar_data = (TarMethodData *) wwmethod;
1230 : 14 : char zerobuf[1024] = {0};
1231 : :
1232 : 14 : clear_error(wwmethod);
1233 : :
1234 [ - + ]: 14 : if (tar_data->currentfile)
1235 : : {
1236 [ # # ]: 0 : if (tar_close(&tar_data->currentfile->base, CLOSE_NORMAL) != 0)
1237 : 0 : return false;
1238 : : }
1239 : :
1240 : : /* A tarfile always ends with two empty blocks */
1241 [ + + ]: 14 : if (wwmethod->compression_algorithm == PG_COMPRESSION_NONE)
1242 : : {
1243 : 11 : errno = 0;
1244 [ - + ]: 11 : if (write(tar_data->fd, zerobuf, sizeof(zerobuf)) != sizeof(zerobuf))
1245 : : {
1246 : : /* If write didn't set errno, assume problem is no disk space */
1247 [ # # ]: 0 : wwmethod->lasterrno = errno ? errno : ENOSPC;
1248 : 0 : return false;
1249 : : }
1250 : : }
1251 : : #ifdef HAVE_LIBZ
1252 [ + - ]: 3 : else if (wwmethod->compression_algorithm == PG_COMPRESSION_GZIP)
1253 : : {
1254 [ - + ]: 3 : if (!tar_write_compressed_data(tar_data, zerobuf, sizeof(zerobuf),
1255 : : false))
1256 : 0 : return false;
1257 : :
1258 : : /* Also flush all data to make sure the gzip stream is finished */
1259 : 3 : tar_data->zp->next_in = NULL;
1260 : 3 : tar_data->zp->avail_in = 0;
1261 : : while (true)
1262 : 0 : {
1263 : : int r;
1264 : :
1265 : 3 : r = deflate(tar_data->zp, Z_FINISH);
1266 : :
1267 [ - + ]: 3 : if (r == Z_STREAM_ERROR)
1268 : : {
1269 : 0 : wwmethod->lasterrstring = _("could not compress data");
1270 : 0 : return false;
1271 : : }
1272 [ + - ]: 3 : if (tar_data->zp->avail_out < ZLIB_OUT_SIZE)
1273 : : {
1274 : 3 : size_t len = ZLIB_OUT_SIZE - tar_data->zp->avail_out;
1275 : :
1276 : 3 : errno = 0;
1277 [ - + ]: 3 : if (write(tar_data->fd, tar_data->zlibOut, len) != len)
1278 : : {
1279 : : /*
1280 : : * If write didn't set errno, assume problem is no disk
1281 : : * space.
1282 : : */
1283 [ # # ]: 0 : wwmethod->lasterrno = errno ? errno : ENOSPC;
1284 : 0 : return false;
1285 : : }
1286 : : }
1287 [ + - ]: 3 : if (r == Z_STREAM_END)
1288 : 3 : break;
1289 : : }
1290 : :
1291 [ - + ]: 3 : if (deflateEnd(tar_data->zp) != Z_OK)
1292 : : {
1293 : 0 : wwmethod->lasterrstring = _("could not close compression stream");
1294 : 0 : return false;
1295 : : }
1296 : : }
1297 : : #endif
1298 : : else
1299 : : {
1300 : : /* not reachable */
1301 : : Assert(false);
1302 : : }
1303 : :
1304 : : /* sync the empty blocks as well, since they're after the last file */
1305 [ - + ]: 14 : if (wwmethod->sync)
1306 : : {
1307 [ # # ]: 0 : if (fsync(tar_data->fd) != 0)
1308 : : {
1309 : 0 : wwmethod->lasterrno = errno;
1310 : 0 : return false;
1311 : : }
1312 : : }
1313 : :
1314 [ - + ]: 14 : if (close(tar_data->fd) != 0)
1315 : : {
1316 : 0 : wwmethod->lasterrno = errno;
1317 : 0 : return false;
1318 : : }
1319 : :
1320 : 14 : tar_data->fd = -1;
1321 : :
1322 [ - + ]: 14 : if (wwmethod->sync)
1323 : : {
1324 [ # # # # ]: 0 : if (fsync_fname(tar_data->tarfilename, false) != 0 ||
1325 : 0 : fsync_parent_path(tar_data->tarfilename) != 0)
1326 : : {
1327 : 0 : wwmethod->lasterrno = errno;
1328 : 0 : return false;
1329 : : }
1330 : : }
1331 : :
1332 : 14 : return true;
1333 : : }
1334 : :
1335 : : static void
1336 : 14 : tar_free(WalWriteMethod *wwmethod)
1337 : : {
1338 : 14 : TarMethodData *tar_data = (TarMethodData *) wwmethod;
1339 : :
1340 : 14 : pg_free(tar_data->tarfilename);
1341 : : #ifdef HAVE_LIBZ
1342 [ + + ]: 14 : if (wwmethod->compression_algorithm == PG_COMPRESSION_GZIP)
1343 : 3 : pg_free(tar_data->zlibOut);
1344 : : #endif
1345 : 14 : pg_free(wwmethod);
1346 : 14 : }
1347 : :
1348 : : /*
1349 : : * The argument compression_algorithm is currently ignored. It is in place for
1350 : : * symmetry with CreateWalDirectoryMethod which uses it for distinguishing
1351 : : * between the different compression methods. CreateWalTarMethod and its family
1352 : : * of functions handle only zlib compression.
1353 : : */
1354 : : WalWriteMethod *
1355 : 14 : CreateWalTarMethod(const char *tarbase,
1356 : : pg_compress_algorithm compression_algorithm,
1357 : : int compression_level, bool sync)
1358 : : {
1359 : : TarMethodData *wwmethod;
1360 : 14 : const char *suffix = (compression_algorithm == PG_COMPRESSION_GZIP) ?
1361 [ + + ]: 14 : ".tar.gz" : ".tar";
1362 : :
1363 : 14 : wwmethod = pg_malloc0_object(TarMethodData);
1364 : 14 : *((const WalWriteMethodOps **) &wwmethod->base.ops) =
1365 : : &WalTarMethodOps;
1366 : 14 : wwmethod->base.compression_algorithm = compression_algorithm;
1367 : 14 : wwmethod->base.compression_level = compression_level;
1368 : 14 : wwmethod->base.sync = sync;
1369 : 14 : clear_error(&wwmethod->base);
1370 : :
1371 : 14 : wwmethod->tarfilename = pg_malloc0(strlen(tarbase) + strlen(suffix) + 1);
1372 : 14 : sprintf(wwmethod->tarfilename, "%s%s", tarbase, suffix);
1373 : 14 : wwmethod->fd = -1;
1374 : : #ifdef HAVE_LIBZ
1375 [ + + ]: 14 : if (compression_algorithm == PG_COMPRESSION_GZIP)
1376 : 3 : wwmethod->zlibOut = (char *) pg_malloc(ZLIB_OUT_SIZE + 1);
1377 : : #endif
1378 : :
1379 : 14 : return &wwmethod->base;
1380 : : }
1381 : :
1382 : : const char *
1383 : 0 : GetLastWalMethodError(WalWriteMethod *wwmethod)
1384 : : {
1385 [ # # ]: 0 : if (wwmethod->lasterrstring)
1386 : 0 : return wwmethod->lasterrstring;
1387 : 0 : return strerror(wwmethod->lasterrno);
1388 : : }
|
