src/bin/pg_basebackup/walmethods.c

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