Line data Source code
1 : /*-------------------------------------------------------------------------
2 : *
3 : * walmethods.c - implementations of different ways to write received wal
4 : *
5 : * NOTE! The caller must ensure that only one method is instantiated in
6 : * any given program, and that it's only instantiated once!
7 : *
8 : * Portions Copyright (c) 1996-2020, PostgreSQL Global Development Group
9 : *
10 : * IDENTIFICATION
11 : * src/bin/pg_basebackup/walmethods.c
12 : *-------------------------------------------------------------------------
13 : */
14 :
15 : #include "postgres_fe.h"
16 :
17 : #include <sys/stat.h>
18 : #include <time.h>
19 : #include <unistd.h>
20 : #ifdef HAVE_LIBZ
21 : #include <zlib.h>
22 : #endif
23 :
24 : #include "common/file_perm.h"
25 : #include "common/file_utils.h"
26 : #include "pgtar.h"
27 : #include "receivelog.h"
28 : #include "streamutil.h"
29 :
30 : /* Size of zlib buffer for .tar.gz */
31 : #define ZLIB_OUT_SIZE 4096
32 :
33 : /*-------------------------------------------------------------------------
34 : * WalDirectoryMethod - write wal to a directory looking like pg_wal
35 : *-------------------------------------------------------------------------
36 : */
37 :
38 : /*
39 : * Global static data for this method
40 : */
41 : typedef struct DirectoryMethodData
42 : {
43 : char *basedir;
44 : int compression;
45 : bool sync;
46 : } DirectoryMethodData;
47 : static DirectoryMethodData *dir_data = NULL;
48 :
49 : /*
50 : * Local file handle
51 : */
52 : typedef struct DirectoryMethodFile
53 : {
54 : int fd;
55 : off_t currpos;
56 : char *pathname;
57 : char *fullpath;
58 : char *temp_suffix;
59 : #ifdef HAVE_LIBZ
60 : gzFile gzfp;
61 : #endif
62 : } DirectoryMethodFile;
63 :
64 : static const char *
65 0 : dir_getlasterror(void)
66 : {
67 : /* Directory method always sets errno, so just use strerror */
68 0 : return strerror(errno);
69 : }
70 :
71 : static Walfile
72 136 : dir_open_for_write(const char *pathname, const char *temp_suffix, size_t pad_to_size)
73 : {
74 : static char tmppath[MAXPGPATH];
75 : int fd;
76 : DirectoryMethodFile *f;
77 : #ifdef HAVE_LIBZ
78 136 : gzFile gzfp = NULL;
79 : #endif
80 :
81 272 : snprintf(tmppath, sizeof(tmppath), "%s/%s%s%s",
82 136 : dir_data->basedir, pathname,
83 136 : dir_data->compression > 0 ? ".gz" : "",
84 : temp_suffix ? temp_suffix : "");
85 :
86 : /*
87 : * Open a file for non-compressed as well as compressed files. Tracking
88 : * the file descriptor is important for dir_sync() method as gzflush()
89 : * does not do any system calls to fsync() to make changes permanent on
90 : * disk.
91 : */
92 136 : fd = open(tmppath, O_WRONLY | O_CREAT | PG_BINARY, pg_file_create_mode);
93 136 : if (fd < 0)
94 0 : return NULL;
95 :
96 : #ifdef HAVE_LIBZ
97 136 : if (dir_data->compression > 0)
98 : {
99 0 : gzfp = gzdopen(fd, "wb");
100 0 : if (gzfp == NULL)
101 : {
102 0 : close(fd);
103 0 : return NULL;
104 : }
105 :
106 0 : if (gzsetparams(gzfp, dir_data->compression,
107 : Z_DEFAULT_STRATEGY) != Z_OK)
108 : {
109 0 : gzclose(gzfp);
110 0 : return NULL;
111 : }
112 : }
113 : #endif
114 :
115 : /* Do pre-padding on non-compressed files */
116 136 : if (pad_to_size && dir_data->compression == 0)
117 : {
118 : PGAlignedXLogBlock zerobuf;
119 : int bytes;
120 :
121 120 : memset(zerobuf.data, 0, XLOG_BLCKSZ);
122 242040 : for (bytes = 0; bytes < pad_to_size; bytes += XLOG_BLCKSZ)
123 : {
124 241920 : errno = 0;
125 241920 : if (write(fd, zerobuf.data, XLOG_BLCKSZ) != XLOG_BLCKSZ)
126 : {
127 0 : int save_errno = errno;
128 :
129 0 : close(fd);
130 :
131 : /*
132 : * If write didn't set errno, assume problem is no disk space.
133 : */
134 0 : errno = save_errno ? save_errno : ENOSPC;
135 0 : return NULL;
136 : }
137 : }
138 :
139 120 : if (lseek(fd, 0, SEEK_SET) != 0)
140 : {
141 0 : int save_errno = errno;
142 :
143 0 : close(fd);
144 0 : errno = save_errno;
145 0 : return NULL;
146 : }
147 : }
148 :
149 : /*
150 : * fsync WAL file and containing directory, to ensure the file is
151 : * persistently created and zeroed (if padded). That's particularly
152 : * important when using synchronous mode, where the file is modified and
153 : * fsynced in-place, without a directory fsync.
154 : */
155 136 : if (dir_data->sync)
156 : {
157 4 : if (fsync_fname(tmppath, false) != 0 ||
158 2 : fsync_parent_path(tmppath) != 0)
159 : {
160 : #ifdef HAVE_LIBZ
161 0 : if (dir_data->compression > 0)
162 0 : gzclose(gzfp);
163 : else
164 : #endif
165 0 : close(fd);
166 0 : return NULL;
167 : }
168 : }
169 :
170 136 : f = pg_malloc0(sizeof(DirectoryMethodFile));
171 : #ifdef HAVE_LIBZ
172 136 : if (dir_data->compression > 0)
173 0 : f->gzfp = gzfp;
174 : #endif
175 136 : f->fd = fd;
176 136 : f->currpos = 0;
177 136 : f->pathname = pg_strdup(pathname);
178 136 : f->fullpath = pg_strdup(tmppath);
179 136 : if (temp_suffix)
180 2 : f->temp_suffix = pg_strdup(temp_suffix);
181 :
182 136 : return f;
183 : }
184 :
185 : static ssize_t
186 4388 : dir_write(Walfile f, const void *buf, size_t count)
187 : {
188 : ssize_t r;
189 4388 : DirectoryMethodFile *df = (DirectoryMethodFile *) f;
190 :
191 : Assert(f != NULL);
192 :
193 : #ifdef HAVE_LIBZ
194 4388 : if (dir_data->compression > 0)
195 0 : r = (ssize_t) gzwrite(df->gzfp, buf, count);
196 : else
197 : #endif
198 4388 : r = write(df->fd, buf, count);
199 4388 : if (r > 0)
200 4388 : df->currpos += r;
201 4388 : return r;
202 : }
203 :
204 : static off_t
205 4388 : dir_get_current_pos(Walfile f)
206 : {
207 : Assert(f != NULL);
208 :
209 : /* Use a cached value to prevent lots of reseeks */
210 4388 : return ((DirectoryMethodFile *) f)->currpos;
211 : }
212 :
213 : static int
214 136 : dir_close(Walfile f, WalCloseMethod method)
215 : {
216 : int r;
217 136 : DirectoryMethodFile *df = (DirectoryMethodFile *) f;
218 : static char tmppath[MAXPGPATH];
219 : static char tmppath2[MAXPGPATH];
220 :
221 : Assert(f != NULL);
222 :
223 : #ifdef HAVE_LIBZ
224 136 : if (dir_data->compression > 0)
225 0 : r = gzclose(df->gzfp);
226 : else
227 : #endif
228 136 : r = close(df->fd);
229 :
230 136 : if (r == 0)
231 : {
232 : /* Build path to the current version of the file */
233 136 : if (method == CLOSE_NORMAL && df->temp_suffix)
234 : {
235 : /*
236 : * If we have a temp prefix, normal operation is to rename the
237 : * file.
238 : */
239 0 : snprintf(tmppath, sizeof(tmppath), "%s/%s%s%s",
240 0 : dir_data->basedir, df->pathname,
241 0 : dir_data->compression > 0 ? ".gz" : "",
242 : df->temp_suffix);
243 0 : snprintf(tmppath2, sizeof(tmppath2), "%s/%s%s",
244 0 : dir_data->basedir, df->pathname,
245 0 : dir_data->compression > 0 ? ".gz" : "");
246 0 : r = durable_rename(tmppath, tmppath2);
247 : }
248 136 : else if (method == CLOSE_UNLINK)
249 : {
250 : /* Unlink the file once it's closed */
251 0 : snprintf(tmppath, sizeof(tmppath), "%s/%s%s%s",
252 0 : dir_data->basedir, df->pathname,
253 0 : dir_data->compression > 0 ? ".gz" : "",
254 0 : df->temp_suffix ? df->temp_suffix : "");
255 0 : r = unlink(tmppath);
256 : }
257 : else
258 : {
259 : /*
260 : * Else either CLOSE_NORMAL and no temp suffix, or
261 : * CLOSE_NO_RENAME. In this case, fsync the file and containing
262 : * directory if sync mode is requested.
263 : */
264 136 : if (dir_data->sync)
265 : {
266 2 : r = fsync_fname(df->fullpath, false);
267 2 : if (r == 0)
268 2 : r = fsync_parent_path(df->fullpath);
269 : }
270 : }
271 : }
272 :
273 136 : pg_free(df->pathname);
274 136 : pg_free(df->fullpath);
275 136 : if (df->temp_suffix)
276 2 : pg_free(df->temp_suffix);
277 136 : pg_free(df);
278 :
279 136 : return r;
280 : }
281 :
282 : static int
283 0 : dir_sync(Walfile f)
284 : {
285 : Assert(f != NULL);
286 :
287 0 : if (!dir_data->sync)
288 0 : return 0;
289 :
290 : #ifdef HAVE_LIBZ
291 0 : if (dir_data->compression > 0)
292 : {
293 0 : if (gzflush(((DirectoryMethodFile *) f)->gzfp, Z_SYNC_FLUSH) != Z_OK)
294 0 : return -1;
295 : }
296 : #endif
297 :
298 0 : return fsync(((DirectoryMethodFile *) f)->fd);
299 : }
300 :
301 : static ssize_t
302 0 : dir_get_file_size(const char *pathname)
303 : {
304 : struct stat statbuf;
305 : static char tmppath[MAXPGPATH];
306 :
307 0 : snprintf(tmppath, sizeof(tmppath), "%s/%s",
308 0 : dir_data->basedir, pathname);
309 :
310 0 : if (stat(tmppath, &statbuf) != 0)
311 0 : return -1;
312 :
313 0 : return statbuf.st_size;
314 : }
315 :
316 : static bool
317 120 : dir_existsfile(const char *pathname)
318 : {
319 : static char tmppath[MAXPGPATH];
320 : int fd;
321 :
322 120 : snprintf(tmppath, sizeof(tmppath), "%s/%s",
323 120 : dir_data->basedir, pathname);
324 :
325 120 : fd = open(tmppath, O_RDONLY | PG_BINARY, 0);
326 120 : if (fd < 0)
327 120 : return false;
328 0 : close(fd);
329 0 : return true;
330 : }
331 :
332 : static bool
333 120 : dir_finish(void)
334 : {
335 120 : if (dir_data->sync)
336 : {
337 : /*
338 : * Files are fsynced when they are closed, but we need to fsync the
339 : * directory entry here as well.
340 : */
341 2 : if (fsync_fname(dir_data->basedir, true) != 0)
342 0 : return false;
343 : }
344 120 : return true;
345 : }
346 :
347 :
348 : WalWriteMethod *
349 122 : CreateWalDirectoryMethod(const char *basedir, int compression, bool sync)
350 : {
351 : WalWriteMethod *method;
352 :
353 122 : method = pg_malloc0(sizeof(WalWriteMethod));
354 122 : method->open_for_write = dir_open_for_write;
355 122 : method->write = dir_write;
356 122 : method->get_current_pos = dir_get_current_pos;
357 122 : method->get_file_size = dir_get_file_size;
358 122 : method->close = dir_close;
359 122 : method->sync = dir_sync;
360 122 : method->existsfile = dir_existsfile;
361 122 : method->finish = dir_finish;
362 122 : method->getlasterror = dir_getlasterror;
363 :
364 122 : dir_data = pg_malloc0(sizeof(DirectoryMethodData));
365 122 : dir_data->compression = compression;
366 122 : dir_data->basedir = pg_strdup(basedir);
367 122 : dir_data->sync = sync;
368 :
369 122 : return method;
370 : }
371 :
372 : void
373 120 : FreeWalDirectoryMethod(void)
374 : {
375 120 : pg_free(dir_data->basedir);
376 120 : pg_free(dir_data);
377 120 : }
378 :
379 :
380 : /*-------------------------------------------------------------------------
381 : * WalTarMethod - write wal to a tar file containing pg_wal contents
382 : *-------------------------------------------------------------------------
383 : */
384 :
385 : typedef struct TarMethodFile
386 : {
387 : off_t ofs_start; /* Where does the *header* for this file start */
388 : off_t currpos;
389 : char header[512];
390 : char *pathname;
391 : size_t pad_to_size;
392 : } TarMethodFile;
393 :
394 : typedef struct TarMethodData
395 : {
396 : char *tarfilename;
397 : int fd;
398 : int compression;
399 : bool sync;
400 : TarMethodFile *currentfile;
401 : char lasterror[1024];
402 : #ifdef HAVE_LIBZ
403 : z_streamp zp;
404 : void *zlibOut;
405 : #endif
406 : } TarMethodData;
407 : static TarMethodData *tar_data = NULL;
408 :
409 : #define tar_clear_error() tar_data->lasterror[0] = '\0'
410 : #define tar_set_error(msg) strlcpy(tar_data->lasterror, _(msg), sizeof(tar_data->lasterror))
411 :
412 : static const char *
413 0 : tar_getlasterror(void)
414 : {
415 : /*
416 : * If a custom error is set, return that one. Otherwise, assume errno is
417 : * set and return that one.
418 : */
419 0 : if (tar_data->lasterror[0])
420 0 : return tar_data->lasterror;
421 0 : return strerror(errno);
422 : }
423 :
424 : #ifdef HAVE_LIBZ
425 : static bool
426 0 : tar_write_compressed_data(void *buf, size_t count, bool flush)
427 : {
428 0 : tar_data->zp->next_in = buf;
429 0 : tar_data->zp->avail_in = count;
430 :
431 0 : while (tar_data->zp->avail_in || flush)
432 : {
433 : int r;
434 :
435 0 : r = deflate(tar_data->zp, flush ? Z_FINISH : Z_NO_FLUSH);
436 0 : if (r == Z_STREAM_ERROR)
437 : {
438 0 : tar_set_error("could not compress data");
439 0 : return false;
440 : }
441 :
442 0 : if (tar_data->zp->avail_out < ZLIB_OUT_SIZE)
443 : {
444 0 : size_t len = ZLIB_OUT_SIZE - tar_data->zp->avail_out;
445 :
446 0 : errno = 0;
447 0 : if (write(tar_data->fd, tar_data->zlibOut, len) != len)
448 : {
449 : /*
450 : * If write didn't set errno, assume problem is no disk space.
451 : */
452 0 : if (errno == 0)
453 0 : errno = ENOSPC;
454 0 : return false;
455 : }
456 :
457 0 : tar_data->zp->next_out = tar_data->zlibOut;
458 0 : tar_data->zp->avail_out = ZLIB_OUT_SIZE;
459 : }
460 :
461 0 : if (r == Z_STREAM_END)
462 0 : break;
463 : }
464 :
465 0 : if (flush)
466 : {
467 : /* Reset the stream for writing */
468 0 : if (deflateReset(tar_data->zp) != Z_OK)
469 : {
470 0 : tar_set_error("could not reset compression stream");
471 0 : return false;
472 : }
473 : }
474 :
475 0 : return true;
476 : }
477 : #endif
478 :
479 : static ssize_t
480 16478 : tar_write(Walfile f, const void *buf, size_t count)
481 : {
482 : ssize_t r;
483 :
484 : Assert(f != NULL);
485 16478 : tar_clear_error();
486 :
487 : /* Tarfile will always be positioned at the end */
488 16478 : if (!tar_data->compression)
489 : {
490 16478 : r = write(tar_data->fd, buf, count);
491 16478 : if (r > 0)
492 16478 : ((TarMethodFile *) f)->currpos += r;
493 16478 : return r;
494 : }
495 : #ifdef HAVE_LIBZ
496 : else
497 : {
498 0 : if (!tar_write_compressed_data(unconstify(void *, buf), count, false))
499 0 : return -1;
500 0 : ((TarMethodFile *) f)->currpos += count;
501 0 : return count;
502 : }
503 : #else
504 : else
505 : /* Can't happen - compression enabled with no libz */
506 : return -1;
507 : #endif
508 : }
509 :
510 : static bool
511 8 : tar_write_padding_data(TarMethodFile *f, size_t bytes)
512 : {
513 : PGAlignedXLogBlock zerobuf;
514 8 : size_t bytesleft = bytes;
515 :
516 8 : memset(zerobuf.data, 0, XLOG_BLCKSZ);
517 16392 : while (bytesleft)
518 : {
519 16384 : size_t bytestowrite = Min(bytesleft, XLOG_BLCKSZ);
520 16384 : ssize_t r = tar_write(f, zerobuf.data, bytestowrite);
521 :
522 16384 : if (r < 0)
523 0 : return false;
524 16384 : bytesleft -= r;
525 : }
526 :
527 8 : return true;
528 : }
529 :
530 : static Walfile
531 8 : tar_open_for_write(const char *pathname, const char *temp_suffix, size_t pad_to_size)
532 : {
533 : int save_errno;
534 : static char tmppath[MAXPGPATH];
535 :
536 8 : tar_clear_error();
537 :
538 8 : if (tar_data->fd < 0)
539 : {
540 : /*
541 : * We open the tar file only when we first try to write to it.
542 : */
543 8 : tar_data->fd = open(tar_data->tarfilename,
544 : O_WRONLY | O_CREAT | PG_BINARY,
545 : pg_file_create_mode);
546 8 : if (tar_data->fd < 0)
547 0 : return NULL;
548 :
549 : #ifdef HAVE_LIBZ
550 8 : if (tar_data->compression)
551 : {
552 0 : tar_data->zp = (z_streamp) pg_malloc(sizeof(z_stream));
553 0 : tar_data->zp->zalloc = Z_NULL;
554 0 : tar_data->zp->zfree = Z_NULL;
555 0 : tar_data->zp->opaque = Z_NULL;
556 0 : tar_data->zp->next_out = tar_data->zlibOut;
557 0 : tar_data->zp->avail_out = ZLIB_OUT_SIZE;
558 :
559 : /*
560 : * Initialize deflation library. Adding the magic value 16 to the
561 : * default 15 for the windowBits parameter makes the output be
562 : * gzip instead of zlib.
563 : */
564 0 : if (deflateInit2(tar_data->zp, tar_data->compression, Z_DEFLATED, 15 + 16, 8, Z_DEFAULT_STRATEGY) != Z_OK)
565 : {
566 0 : pg_free(tar_data->zp);
567 0 : tar_data->zp = NULL;
568 0 : tar_set_error("could not initialize compression library");
569 0 : return NULL;
570 : }
571 : }
572 : #endif
573 :
574 : /* There's no tar header itself, the file starts with regular files */
575 : }
576 :
577 : Assert(tar_data->currentfile == NULL);
578 8 : if (tar_data->currentfile != NULL)
579 : {
580 0 : tar_set_error("implementation error: tar files can't have more than one open file");
581 0 : return NULL;
582 : }
583 :
584 8 : tar_data->currentfile = pg_malloc0(sizeof(TarMethodFile));
585 :
586 8 : snprintf(tmppath, sizeof(tmppath), "%s%s",
587 : pathname, temp_suffix ? temp_suffix : "");
588 :
589 : /* Create a header with size set to 0 - we will fill out the size on close */
590 8 : if (tarCreateHeader(tar_data->currentfile->header, tmppath, NULL, 0, S_IRUSR | S_IWUSR, 0, 0, time(NULL)) != TAR_OK)
591 : {
592 0 : pg_free(tar_data->currentfile);
593 0 : tar_data->currentfile = NULL;
594 0 : tar_set_error("could not create tar header");
595 0 : return NULL;
596 : }
597 :
598 : #ifdef HAVE_LIBZ
599 8 : if (tar_data->compression)
600 : {
601 : /* Flush existing data */
602 0 : if (!tar_write_compressed_data(NULL, 0, true))
603 0 : return NULL;
604 :
605 : /* Turn off compression for header */
606 0 : if (deflateParams(tar_data->zp, 0, 0) != Z_OK)
607 : {
608 0 : tar_set_error("could not change compression parameters");
609 0 : return NULL;
610 : }
611 : }
612 : #endif
613 :
614 8 : tar_data->currentfile->ofs_start = lseek(tar_data->fd, 0, SEEK_CUR);
615 8 : if (tar_data->currentfile->ofs_start == -1)
616 : {
617 0 : save_errno = errno;
618 0 : pg_free(tar_data->currentfile);
619 0 : tar_data->currentfile = NULL;
620 0 : errno = save_errno;
621 0 : return NULL;
622 : }
623 8 : tar_data->currentfile->currpos = 0;
624 :
625 8 : if (!tar_data->compression)
626 : {
627 8 : errno = 0;
628 8 : if (write(tar_data->fd, tar_data->currentfile->header, 512) != 512)
629 : {
630 0 : save_errno = errno;
631 0 : pg_free(tar_data->currentfile);
632 0 : tar_data->currentfile = NULL;
633 : /* if write didn't set errno, assume problem is no disk space */
634 0 : errno = save_errno ? save_errno : ENOSPC;
635 0 : return NULL;
636 : }
637 : }
638 : #ifdef HAVE_LIBZ
639 : else
640 : {
641 : /* Write header through the zlib APIs but with no compression */
642 0 : if (!tar_write_compressed_data(tar_data->currentfile->header, 512, true))
643 0 : return NULL;
644 :
645 : /* Re-enable compression for the rest of the file */
646 0 : if (deflateParams(tar_data->zp, tar_data->compression, 0) != Z_OK)
647 : {
648 0 : tar_set_error("could not change compression parameters");
649 0 : return NULL;
650 : }
651 : }
652 : #endif
653 :
654 8 : tar_data->currentfile->pathname = pg_strdup(pathname);
655 :
656 : /*
657 : * Uncompressed files are padded on creation, but for compression we can't
658 : * do that
659 : */
660 8 : if (pad_to_size)
661 : {
662 8 : tar_data->currentfile->pad_to_size = pad_to_size;
663 8 : if (!tar_data->compression)
664 : {
665 : /* Uncompressed, so pad now */
666 8 : tar_write_padding_data(tar_data->currentfile, pad_to_size);
667 : /* Seek back to start */
668 8 : if (lseek(tar_data->fd, tar_data->currentfile->ofs_start + 512, SEEK_SET) != tar_data->currentfile->ofs_start + 512)
669 0 : return NULL;
670 :
671 8 : tar_data->currentfile->currpos = 0;
672 : }
673 : }
674 :
675 8 : return tar_data->currentfile;
676 : }
677 :
678 : static ssize_t
679 0 : tar_get_file_size(const char *pathname)
680 : {
681 0 : tar_clear_error();
682 :
683 : /* Currently not used, so not supported */
684 0 : errno = ENOSYS;
685 0 : return -1;
686 : }
687 :
688 : static off_t
689 102 : tar_get_current_pos(Walfile f)
690 : {
691 : Assert(f != NULL);
692 102 : tar_clear_error();
693 :
694 102 : return ((TarMethodFile *) f)->currpos;
695 : }
696 :
697 : static int
698 8 : tar_sync(Walfile f)
699 : {
700 : Assert(f != NULL);
701 8 : tar_clear_error();
702 :
703 8 : if (!tar_data->sync)
704 8 : return 0;
705 :
706 : /*
707 : * Always sync the whole tarfile, because that's all we can do. This makes
708 : * no sense on compressed files, so just ignore those.
709 : */
710 0 : if (tar_data->compression)
711 0 : return 0;
712 :
713 0 : return fsync(tar_data->fd);
714 : }
715 :
716 : static int
717 8 : tar_close(Walfile f, WalCloseMethod method)
718 : {
719 : ssize_t filesize;
720 : int padding;
721 8 : TarMethodFile *tf = (TarMethodFile *) f;
722 :
723 : Assert(f != NULL);
724 8 : tar_clear_error();
725 :
726 8 : if (method == CLOSE_UNLINK)
727 : {
728 0 : if (tar_data->compression)
729 : {
730 0 : tar_set_error("unlink not supported with compression");
731 0 : return -1;
732 : }
733 :
734 : /*
735 : * Unlink the file that we just wrote to the tar. We do this by
736 : * truncating it to the start of the header. This is safe as we only
737 : * allow writing of the very last file.
738 : */
739 0 : if (ftruncate(tar_data->fd, tf->ofs_start) != 0)
740 0 : return -1;
741 :
742 0 : pg_free(tf->pathname);
743 0 : pg_free(tf);
744 0 : tar_data->currentfile = NULL;
745 :
746 0 : return 0;
747 : }
748 :
749 : /*
750 : * Pad the file itself with zeroes if necessary. Note that this is
751 : * different from the tar format padding -- this is the padding we asked
752 : * for when the file was opened.
753 : */
754 8 : if (tf->pad_to_size)
755 : {
756 8 : if (tar_data->compression)
757 : {
758 : /*
759 : * A compressed tarfile is padded on close since we cannot know
760 : * the size of the compressed output until the end.
761 : */
762 0 : size_t sizeleft = tf->pad_to_size - tf->currpos;
763 :
764 0 : if (sizeleft)
765 : {
766 0 : if (!tar_write_padding_data(tf, sizeleft))
767 0 : return -1;
768 : }
769 : }
770 : else
771 : {
772 : /*
773 : * An uncompressed tarfile was padded on creation, so just adjust
774 : * the current position as if we seeked to the end.
775 : */
776 8 : tf->currpos = tf->pad_to_size;
777 : }
778 : }
779 :
780 : /*
781 : * Get the size of the file, and pad the current data up to the nearest
782 : * 512 byte boundary.
783 : */
784 8 : filesize = tar_get_current_pos(f);
785 8 : padding = ((filesize + 511) & ~511) - filesize;
786 8 : if (padding)
787 : {
788 : char zerobuf[512];
789 :
790 0 : MemSet(zerobuf, 0, padding);
791 0 : if (tar_write(f, zerobuf, padding) != padding)
792 0 : return -1;
793 : }
794 :
795 :
796 : #ifdef HAVE_LIBZ
797 8 : if (tar_data->compression)
798 : {
799 : /* Flush the current buffer */
800 0 : if (!tar_write_compressed_data(NULL, 0, true))
801 : {
802 0 : errno = EINVAL;
803 0 : return -1;
804 : }
805 : }
806 : #endif
807 :
808 : /*
809 : * Now go back and update the header with the correct filesize and
810 : * possibly also renaming the file. We overwrite the entire current header
811 : * when done, including the checksum.
812 : */
813 8 : print_tar_number(&(tf->header[124]), 12, filesize);
814 :
815 8 : if (method == CLOSE_NORMAL)
816 :
817 : /*
818 : * We overwrite it with what it was before if we have no tempname,
819 : * since we're going to write the buffer anyway.
820 : */
821 8 : strlcpy(&(tf->header[0]), tf->pathname, 100);
822 :
823 8 : print_tar_number(&(tf->header[148]), 8, tarChecksum(((TarMethodFile *) f)->header));
824 8 : if (lseek(tar_data->fd, tf->ofs_start, SEEK_SET) != ((TarMethodFile *) f)->ofs_start)
825 0 : return -1;
826 8 : if (!tar_data->compression)
827 : {
828 8 : errno = 0;
829 8 : if (write(tar_data->fd, tf->header, 512) != 512)
830 : {
831 : /* if write didn't set errno, assume problem is no disk space */
832 0 : if (errno == 0)
833 0 : errno = ENOSPC;
834 0 : return -1;
835 : }
836 : }
837 : #ifdef HAVE_LIBZ
838 : else
839 : {
840 : /* Turn off compression */
841 0 : if (deflateParams(tar_data->zp, 0, 0) != Z_OK)
842 : {
843 0 : tar_set_error("could not change compression parameters");
844 0 : return -1;
845 : }
846 :
847 : /* Overwrite the header, assuming the size will be the same */
848 0 : if (!tar_write_compressed_data(tar_data->currentfile->header, 512, true))
849 0 : return -1;
850 :
851 : /* Turn compression back on */
852 0 : if (deflateParams(tar_data->zp, tar_data->compression, 0) != Z_OK)
853 : {
854 0 : tar_set_error("could not change compression parameters");
855 0 : return -1;
856 : }
857 : }
858 : #endif
859 :
860 : /* Move file pointer back down to end, so we can write the next file */
861 8 : if (lseek(tar_data->fd, 0, SEEK_END) < 0)
862 0 : return -1;
863 :
864 : /* Always fsync on close, so the padding gets fsynced */
865 8 : if (tar_sync(f) < 0)
866 0 : exit(1);
867 :
868 : /* Clean up and done */
869 8 : pg_free(tf->pathname);
870 8 : pg_free(tf);
871 8 : tar_data->currentfile = NULL;
872 :
873 8 : return 0;
874 : }
875 :
876 : static bool
877 8 : tar_existsfile(const char *pathname)
878 : {
879 8 : tar_clear_error();
880 : /* We only deal with new tarfiles, so nothing externally created exists */
881 8 : return false;
882 : }
883 :
884 : static bool
885 8 : tar_finish(void)
886 : {
887 : char zerobuf[1024];
888 :
889 8 : tar_clear_error();
890 :
891 8 : if (tar_data->currentfile)
892 : {
893 0 : if (tar_close(tar_data->currentfile, CLOSE_NORMAL) != 0)
894 0 : return false;
895 : }
896 :
897 : /* A tarfile always ends with two empty blocks */
898 1032 : MemSet(zerobuf, 0, sizeof(zerobuf));
899 8 : if (!tar_data->compression)
900 : {
901 8 : errno = 0;
902 8 : if (write(tar_data->fd, zerobuf, sizeof(zerobuf)) != sizeof(zerobuf))
903 : {
904 : /* if write didn't set errno, assume problem is no disk space */
905 0 : if (errno == 0)
906 0 : errno = ENOSPC;
907 0 : return false;
908 : }
909 : }
910 : #ifdef HAVE_LIBZ
911 : else
912 : {
913 0 : if (!tar_write_compressed_data(zerobuf, sizeof(zerobuf), false))
914 0 : return false;
915 :
916 : /* Also flush all data to make sure the gzip stream is finished */
917 0 : tar_data->zp->next_in = NULL;
918 0 : tar_data->zp->avail_in = 0;
919 : while (true)
920 0 : {
921 : int r;
922 :
923 0 : r = deflate(tar_data->zp, Z_FINISH);
924 :
925 0 : if (r == Z_STREAM_ERROR)
926 : {
927 0 : tar_set_error("could not compress data");
928 0 : return false;
929 : }
930 0 : if (tar_data->zp->avail_out < ZLIB_OUT_SIZE)
931 : {
932 0 : size_t len = ZLIB_OUT_SIZE - tar_data->zp->avail_out;
933 :
934 0 : errno = 0;
935 0 : if (write(tar_data->fd, tar_data->zlibOut, len) != len)
936 : {
937 : /*
938 : * If write didn't set errno, assume problem is no disk
939 : * space.
940 : */
941 0 : if (errno == 0)
942 0 : errno = ENOSPC;
943 0 : return false;
944 : }
945 : }
946 0 : if (r == Z_STREAM_END)
947 0 : break;
948 : }
949 :
950 0 : if (deflateEnd(tar_data->zp) != Z_OK)
951 : {
952 0 : tar_set_error("could not close compression stream");
953 0 : return false;
954 : }
955 : }
956 : #endif
957 :
958 : /* sync the empty blocks as well, since they're after the last file */
959 8 : if (tar_data->sync)
960 : {
961 0 : if (fsync(tar_data->fd) != 0)
962 0 : return false;
963 : }
964 :
965 8 : if (close(tar_data->fd) != 0)
966 0 : return false;
967 :
968 8 : tar_data->fd = -1;
969 :
970 8 : if (tar_data->sync)
971 : {
972 0 : if (fsync_fname(tar_data->tarfilename, false) != 0)
973 0 : return false;
974 0 : if (fsync_parent_path(tar_data->tarfilename) != 0)
975 0 : return false;
976 : }
977 :
978 8 : return true;
979 : }
980 :
981 : WalWriteMethod *
982 8 : CreateWalTarMethod(const char *tarbase, int compression, bool sync)
983 : {
984 : WalWriteMethod *method;
985 8 : const char *suffix = (compression != 0) ? ".tar.gz" : ".tar";
986 :
987 8 : method = pg_malloc0(sizeof(WalWriteMethod));
988 8 : method->open_for_write = tar_open_for_write;
989 8 : method->write = tar_write;
990 8 : method->get_current_pos = tar_get_current_pos;
991 8 : method->get_file_size = tar_get_file_size;
992 8 : method->close = tar_close;
993 8 : method->sync = tar_sync;
994 8 : method->existsfile = tar_existsfile;
995 8 : method->finish = tar_finish;
996 8 : method->getlasterror = tar_getlasterror;
997 :
998 8 : tar_data = pg_malloc0(sizeof(TarMethodData));
999 8 : tar_data->tarfilename = pg_malloc0(strlen(tarbase) + strlen(suffix) + 1);
1000 8 : sprintf(tar_data->tarfilename, "%s%s", tarbase, suffix);
1001 8 : tar_data->fd = -1;
1002 8 : tar_data->compression = compression;
1003 8 : tar_data->sync = sync;
1004 : #ifdef HAVE_LIBZ
1005 8 : if (compression)
1006 0 : tar_data->zlibOut = (char *) pg_malloc(ZLIB_OUT_SIZE + 1);
1007 : #endif
1008 :
1009 8 : return method;
1010 : }
1011 :
1012 : void
1013 8 : FreeWalTarMethod(void)
1014 : {
1015 8 : pg_free(tar_data->tarfilename);
1016 : #ifdef HAVE_LIBZ
1017 8 : if (tar_data->compression)
1018 0 : pg_free(tar_data->zlibOut);
1019 : #endif
1020 8 : pg_free(tar_data);
1021 8 : }
|
