walmethods.c

Go to the documentation of this file.

/*-------------------------------------------------------------------------
 *
 * walmethods.c - implementations of different ways to write received wal
 *
 * Portions Copyright (c) 1996-2023, PostgreSQL Global Development Group
 *
 * IDENTIFICATION
 *        src/bin/pg_basebackup/walmethods.c
 *-------------------------------------------------------------------------
 */
 
#include "postgres_fe.h"
 
#include <sys/stat.h>
#include <time.h>
#include <unistd.h>
 
#ifdef USE_LZ4
#include <lz4frame.h>
#endif
#ifdef HAVE_LIBZ
#include <zlib.h>
#endif
 
#include "common/file_perm.h"
#include "common/file_utils.h"
#include "common/logging.h"
#include "pgtar.h"
#include "receivelog.h"
#include "streamutil.h"
 
/* Size of zlib buffer for .tar.gz */
#define ZLIB_OUT_SIZE 4096
 
/* Size of LZ4 input chunk for .lz4 */
#define LZ4_IN_SIZE  4096
 
/*-------------------------------------------------------------------------
 * WalDirectoryMethod - write wal to a directory looking like pg_wal
 *-------------------------------------------------------------------------
 */
 
static Walfile *dir_open_for_write(WalWriteMethod *wwmethod,
                                   const char *pathname,
                                   const char *temp_suffix,
                                   size_t pad_to_size);
static int dir_close(Walfile *f, WalCloseMethod method);
static bool dir_existsfile(WalWriteMethod *wwmethod, const char *pathname);
static ssize_t dir_get_file_size(WalWriteMethod *wwmethod,
                                 const char *pathname);
static char *dir_get_file_name(WalWriteMethod *wwmethod,
                               const char *pathname, const char *temp_suffix);
static ssize_t dir_write(Walfile *f, const void *buf, size_t count);
static int dir_sync(Walfile *f);
static bool dir_finish(WalWriteMethod *wwmethod);
static void dir_free(WalWriteMethod *wwmethod);
 
const WalWriteMethodOps WalDirectoryMethodOps = {
    .open_for_write = dir_open_for_write,
    .close = dir_close,
    .existsfile = dir_existsfile,
    .get_file_size = dir_get_file_size,
    .get_file_name = dir_get_file_name,
    .write = dir_write,
    .sync = dir_sync,
    .finish = dir_finish,
    .free = dir_free
};
 
/*
 * Global static data for this method
 */
typedef struct DirectoryMethodData
{
    WalWriteMethod  base;
    char       *basedir;
} DirectoryMethodData;
 
/*
 * Local file handle
 */
typedef struct DirectoryMethodFile
{
    Walfile     base;
    int         fd;
    char       *fullpath;
    char       *temp_suffix;
#ifdef HAVE_LIBZ
    gzFile      gzfp;
#endif
#ifdef USE_LZ4
    LZ4F_compressionContext_t ctx;
    size_t      lz4bufsize;
    void       *lz4buf;
#endif
} DirectoryMethodFile;
 
#define clear_error(wwmethod) \
    ((wwmethod)->lasterrstring = NULL, (wwmethod)->lasterrno = 0)
 
static char *
dir_get_file_name(WalWriteMethod *wwmethod,
                  const char *pathname, const char *temp_suffix)
{
    char       *filename = pg_malloc0(MAXPGPATH * sizeof(char));
 
    snprintf(filename, MAXPGPATH, "%s%s%s",
             pathname,
             wwmethod->compression_algorithm == PG_COMPRESSION_GZIP ? ".gz" :
             wwmethod->compression_algorithm == PG_COMPRESSION_LZ4 ? ".lz4" : "",
             temp_suffix ? temp_suffix : "");
 
    return filename;
}
 
static Walfile *
dir_open_for_write(WalWriteMethod *wwmethod, const char *pathname,
                   const char *temp_suffix, size_t pad_to_size)
{
    DirectoryMethodData *dir_data = (DirectoryMethodData *) wwmethod;
    char        tmppath[MAXPGPATH];
    char       *filename;
    int         fd;
    DirectoryMethodFile *f;
#ifdef HAVE_LIBZ
    gzFile      gzfp = NULL;
#endif
#ifdef USE_LZ4
    LZ4F_compressionContext_t ctx = NULL;
    size_t      lz4bufsize = 0;
    void       *lz4buf = NULL;
#endif
 
    clear_error(wwmethod);
 
    filename = dir_get_file_name(wwmethod, pathname, temp_suffix);
    snprintf(tmppath, sizeof(tmppath), "%s/%s",
             dir_data->basedir, filename);
    pg_free(filename);
 
    /*
     * Open a file for non-compressed as well as compressed files. Tracking
     * the file descriptor is important for dir_sync() method as gzflush()
     * does not do any system calls to fsync() to make changes permanent on
     * disk.
     */
    fd = open(tmppath, O_WRONLY | O_CREAT | PG_BINARY, pg_file_create_mode);
    if (fd < 0)
    {
        wwmethod->lasterrno = errno;
        return NULL;
    }
 
#ifdef HAVE_LIBZ
    if (wwmethod->compression_algorithm == PG_COMPRESSION_GZIP)
    {
        gzfp = gzdopen(fd, "wb");
        if (gzfp == NULL)
        {
            wwmethod->lasterrno = errno;
            close(fd);
            return NULL;
        }
 
        if (gzsetparams(gzfp, wwmethod->compression_level,
                        Z_DEFAULT_STRATEGY) != Z_OK)
        {
            wwmethod->lasterrno = errno;
            gzclose(gzfp);
            return NULL;
        }
    }
#endif
#ifdef USE_LZ4
    if (wwmethod->compression_algorithm == PG_COMPRESSION_LZ4)
    {
        size_t      ctx_out;
        size_t      header_size;
        LZ4F_preferences_t prefs;
 
        ctx_out = LZ4F_createCompressionContext(&ctx, LZ4F_VERSION);
        if (LZ4F_isError(ctx_out))
        {
            wwmethod->lasterrstring = LZ4F_getErrorName(ctx_out);
            close(fd);
            return NULL;
        }
 
        lz4bufsize = LZ4F_compressBound(LZ4_IN_SIZE, NULL);
        lz4buf = pg_malloc0(lz4bufsize);
 
        /* assign the compression level, default is 0 */
        memset(&prefs, 0, sizeof(prefs));
        prefs.compressionLevel = wwmethod->compression_level;
 
        /* add the header */
        header_size = LZ4F_compressBegin(ctx, lz4buf, lz4bufsize, &prefs);
        if (LZ4F_isError(header_size))
        {
            wwmethod->lasterrstring = LZ4F_getErrorName(header_size);
            (void) LZ4F_freeCompressionContext(ctx);
            pg_free(lz4buf);
            close(fd);
            return NULL;
        }
 
        errno = 0;
        if (write(fd, lz4buf, header_size) != header_size)
        {
            /* If write didn't set errno, assume problem is no disk space */
            wwmethod->lasterrno = errno ? errno : ENOSPC;
            (void) LZ4F_freeCompressionContext(ctx);
            pg_free(lz4buf);
            close(fd);
            return NULL;
        }
    }
#endif
 
    /* Do pre-padding on non-compressed files */
    if (pad_to_size && wwmethod->compression_algorithm == PG_COMPRESSION_NONE)
    {
        ssize_t     rc;
 
        rc = pg_pwrite_zeros(fd, pad_to_size, 0);
 
        if (rc < 0)
        {
            wwmethod->lasterrno = errno;
            close(fd);
            return NULL;
        }
 
        /*
         * pg_pwrite() (called via pg_pwrite_zeros()) may have moved the file
         * position, so reset it (see win32pwrite.c).
         */
        if (lseek(fd, 0, SEEK_SET) != 0)
        {
            wwmethod->lasterrno = errno;
            close(fd);
            return NULL;
        }
    }
 
    /*
     * fsync WAL file and containing directory, to ensure the file is
     * persistently created and zeroed (if padded). That's particularly
     * important when using synchronous mode, where the file is modified and
     * fsynced in-place, without a directory fsync.
     */
    if (wwmethod->sync)
    {
        if (fsync_fname(tmppath, false) != 0 ||
            fsync_parent_path(tmppath) != 0)
        {
            wwmethod->lasterrno = errno;
#ifdef HAVE_LIBZ
            if (wwmethod->compression_algorithm == PG_COMPRESSION_GZIP)
                gzclose(gzfp);
            else
#endif
#ifdef USE_LZ4
            if (wwmethod->compression_algorithm == PG_COMPRESSION_LZ4)
            {
                (void) LZ4F_compressEnd(ctx, lz4buf, lz4bufsize, NULL);
                (void) LZ4F_freeCompressionContext(ctx);
                pg_free(lz4buf);
                close(fd);
            }
            else
#endif
                close(fd);
            return NULL;
        }
    }
 
    f = pg_malloc0(sizeof(DirectoryMethodFile));
#ifdef HAVE_LIBZ
    if (wwmethod->compression_algorithm == PG_COMPRESSION_GZIP)
        f->gzfp = gzfp;
#endif
#ifdef USE_LZ4
    if (wwmethod->compression_algorithm == PG_COMPRESSION_LZ4)
    {
        f->ctx = ctx;
        f->lz4buf = lz4buf;
        f->lz4bufsize = lz4bufsize;
    }
#endif
 
    f->base.wwmethod = wwmethod;
    f->base.currpos = 0;
    f->base.pathname = pg_strdup(pathname);
    f->fd = fd;
    f->fullpath = pg_strdup(tmppath);
    if (temp_suffix)
        f->temp_suffix = pg_strdup(temp_suffix);
 
    return &f->base;
}
 
static ssize_t
dir_write(Walfile *f, const void *buf, size_t count)
{
    ssize_t     r;
    DirectoryMethodFile *df = (DirectoryMethodFile *) f;
 
    Assert(f != NULL);
    clear_error(f->wwmethod);
 
#ifdef HAVE_LIBZ
    if (f->wwmethod->compression_algorithm == PG_COMPRESSION_GZIP)
    {
        errno = 0;
        r = (ssize_t) gzwrite(df->gzfp, buf, count);
        if (r != count)
        {
            /* If write didn't set errno, assume problem is no disk space */
            f->wwmethod->lasterrno = errno ? errno : ENOSPC;
        }
    }
    else
#endif
#ifdef USE_LZ4
    if (f->wwmethod->compression_algorithm == PG_COMPRESSION_LZ4)
    {
        size_t      chunk;
        size_t      remaining;
        const void *inbuf = buf;
 
        remaining = count;
        while (remaining > 0)
        {
            size_t      compressed;
 
            if (remaining > LZ4_IN_SIZE)
                chunk = LZ4_IN_SIZE;
            else
                chunk = remaining;
 
            remaining -= chunk;
            compressed = LZ4F_compressUpdate(df->ctx,
                                             df->lz4buf, df->lz4bufsize,
                                             inbuf, chunk,
                                             NULL);
 
            if (LZ4F_isError(compressed))
            {
                f->wwmethod->lasterrstring = LZ4F_getErrorName(compressed);
                return -1;
            }
 
            errno = 0;
            if (write(df->fd, df->lz4buf, compressed) != compressed)
            {
                /* If write didn't set errno, assume problem is no disk space */
                f->wwmethod->lasterrno = errno ? errno : ENOSPC;
                return -1;
            }
 
            inbuf = ((char *) inbuf) + chunk;
        }
 
        /* Our caller keeps track of the uncompressed size. */
        r = (ssize_t) count;
    }
    else
#endif
    {
        errno = 0;
        r = write(df->fd, buf, count);
        if (r != count)
        {
            /* If write didn't set errno, assume problem is no disk space */
            f->wwmethod->lasterrno = errno ? errno : ENOSPC;
        }
    }
    if (r > 0)
        df->base.currpos += r;
    return r;
}
 
static int
dir_close(Walfile *f, WalCloseMethod method)
{
    int         r;
    DirectoryMethodFile *df = (DirectoryMethodFile *) f;
    DirectoryMethodData *dir_data = (DirectoryMethodData *) f->wwmethod;
    char        tmppath[MAXPGPATH];
    char        tmppath2[MAXPGPATH];
 
    Assert(f != NULL);
    clear_error(f->wwmethod);
 
#ifdef HAVE_LIBZ
    if (f->wwmethod->compression_algorithm == PG_COMPRESSION_GZIP)
    {
        errno = 0;              /* in case gzclose() doesn't set it */
        r = gzclose(df->gzfp);
    }
    else
#endif
#ifdef USE_LZ4
    if (f->wwmethod->compression_algorithm == PG_COMPRESSION_LZ4)
    {
        size_t      compressed;
 
        compressed = LZ4F_compressEnd(df->ctx,
                                      df->lz4buf, df->lz4bufsize,
                                      NULL);
 
        if (LZ4F_isError(compressed))
        {
            f->wwmethod->lasterrstring = LZ4F_getErrorName(compressed);
            return -1;
        }
 
        errno = 0;
        if (write(df->fd, df->lz4buf, compressed) != compressed)
        {
            /* If write didn't set errno, assume problem is no disk space */
            f->wwmethod->lasterrno = errno ? errno : ENOSPC;
            return -1;
        }
 
        r = close(df->fd);
    }
    else
#endif
        r = close(df->fd);
 
    if (r == 0)
    {
        /* Build path to the current version of the file */
        if (method == CLOSE_NORMAL && df->temp_suffix)
        {
            char       *filename;
            char       *filename2;
 
            /*
             * If we have a temp prefix, normal operation is to rename the
             * file.
             */
            filename = dir_get_file_name(f->wwmethod, df->base.pathname,
                                         df->temp_suffix);
            snprintf(tmppath, sizeof(tmppath), "%s/%s",
                     dir_data->basedir, filename);
            pg_free(filename);
 
            /* permanent name, so no need for the prefix */
            filename2 = dir_get_file_name(f->wwmethod, df->base.pathname, NULL);
            snprintf(tmppath2, sizeof(tmppath2), "%s/%s",
                     dir_data->basedir, filename2);
            pg_free(filename2);
            if (f->wwmethod->sync)
                r = durable_rename(tmppath, tmppath2);
            else
            {
                if (rename(tmppath, tmppath2) != 0)
                {
                    pg_log_error("could not rename file \"%s\" to \"%s\": %m",
                                 tmppath, tmppath2);
                    r = -1;
                }
            }
        }
        else if (method == CLOSE_UNLINK)
        {
            char       *filename;
 
            /* Unlink the file once it's closed */
            filename = dir_get_file_name(f->wwmethod, df->base.pathname,
                                         df->temp_suffix);
            snprintf(tmppath, sizeof(tmppath), "%s/%s",
                     dir_data->basedir, filename);
            pg_free(filename);
            r = unlink(tmppath);
        }
        else
        {
            /*
             * Else either CLOSE_NORMAL and no temp suffix, or
             * CLOSE_NO_RENAME. In this case, fsync the file and containing
             * directory if sync mode is requested.
             */
            if (f->wwmethod->sync)
            {
                r = fsync_fname(df->fullpath, false);
                if (r == 0)
                    r = fsync_parent_path(df->fullpath);
            }
        }
    }
 
    if (r != 0)
        f->wwmethod->lasterrno = errno;
 
#ifdef USE_LZ4
    pg_free(df->lz4buf);
    /* supports free on NULL */
    LZ4F_freeCompressionContext(df->ctx);
#endif
 
    pg_free(df->base.pathname);
    pg_free(df->fullpath);
    pg_free(df->temp_suffix);
    pg_free(df);
 
    return r;
}
 
static int
dir_sync(Walfile *f)
{
    int         r;
 
    Assert(f != NULL);
    clear_error(f->wwmethod);
 
    if (!f->wwmethod->sync)
        return 0;
 
#ifdef HAVE_LIBZ
    if (f->wwmethod->compression_algorithm == PG_COMPRESSION_GZIP)
    {
        if (gzflush(((DirectoryMethodFile *) f)->gzfp, Z_SYNC_FLUSH) != Z_OK)
        {
            f->wwmethod->lasterrno = errno;
            return -1;
        }
    }
#endif
#ifdef USE_LZ4
    if (f->wwmethod->compression_algorithm == PG_COMPRESSION_LZ4)
    {
        DirectoryMethodFile *df = (DirectoryMethodFile *) f;
        size_t      compressed;
 
        /* Flush any internal buffers */
        compressed = LZ4F_flush(df->ctx, df->lz4buf, df->lz4bufsize, NULL);
        if (LZ4F_isError(compressed))
        {
            f->wwmethod->lasterrstring = LZ4F_getErrorName(compressed);
            return -1;
        }
 
        errno = 0;
        if (write(df->fd, df->lz4buf, compressed) != compressed)
        {
            /* If write didn't set errno, assume problem is no disk space */
            f->wwmethod->lasterrno = errno ? errno : ENOSPC;
            return -1;
        }
    }
#endif
 
    r = fsync(((DirectoryMethodFile *) f)->fd);
    if (r < 0)
        f->wwmethod->lasterrno = errno;
    return r;
}
 
static ssize_t
dir_get_file_size(WalWriteMethod *wwmethod, const char *pathname)
{
    DirectoryMethodData *dir_data = (DirectoryMethodData *) wwmethod;
    struct stat statbuf;
    char        tmppath[MAXPGPATH];
 
    snprintf(tmppath, sizeof(tmppath), "%s/%s",
             dir_data->basedir, pathname);
 
    if (stat(tmppath, &statbuf) != 0)
    {
        wwmethod->lasterrno = errno;
        return -1;
    }
 
    return statbuf.st_size;
}
 
static bool
dir_existsfile(WalWriteMethod *wwmethod, const char *pathname)
{
    DirectoryMethodData *dir_data = (DirectoryMethodData *) wwmethod;
    char        tmppath[MAXPGPATH];
    int         fd;
 
    clear_error(wwmethod);
 
    snprintf(tmppath, sizeof(tmppath), "%s/%s",
             dir_data->basedir, pathname);
 
    fd = open(tmppath, O_RDONLY | PG_BINARY, 0);
    if (fd < 0)
        return false;
    close(fd);
    return true;
}
 
static bool
dir_finish(WalWriteMethod *wwmethod)
{
    clear_error(wwmethod);
 
    if (wwmethod->sync)
    {
        DirectoryMethodData *dir_data = (DirectoryMethodData *) wwmethod;
 
        /*
         * Files are fsynced when they are closed, but we need to fsync the
         * directory entry here as well.
         */
        if (fsync_fname(dir_data->basedir, true) != 0)
        {
            wwmethod->lasterrno = errno;
            return false;
        }
    }
    return true;
}
 
static void
dir_free(WalWriteMethod *wwmethod)
{
    DirectoryMethodData *dir_data = (DirectoryMethodData *) wwmethod;
 
    pg_free(dir_data->basedir);
    pg_free(wwmethod);
}
 
 
WalWriteMethod *
CreateWalDirectoryMethod(const char *basedir,
                         pg_compress_algorithm compression_algorithm,
                         int compression_level, bool sync)
{
    DirectoryMethodData *wwmethod;
 
    wwmethod = pg_malloc0(sizeof(DirectoryMethodData));
    *((const WalWriteMethodOps **) &wwmethod->base.ops) =
        &WalDirectoryMethodOps;
    wwmethod->base.compression_algorithm = compression_algorithm;
    wwmethod->base.compression_level = compression_level;
    wwmethod->base.sync = sync;
    clear_error(&wwmethod->base);
    wwmethod->basedir = pg_strdup(basedir);
 
    return &wwmethod->base;
}
 
 
/*-------------------------------------------------------------------------
 * WalTarMethod - write wal to a tar file containing pg_wal contents
 *-------------------------------------------------------------------------
 */
 
static Walfile *tar_open_for_write(WalWriteMethod *wwmethod,
                                   const char *pathname,
                                   const char *temp_suffix,
                                   size_t pad_to_size);
static int tar_close(Walfile *f, WalCloseMethod method);
static bool tar_existsfile(WalWriteMethod *wwmethod, const char *pathname);
static ssize_t tar_get_file_size(WalWriteMethod *wwmethod,
                                 const char *pathname);
static char *tar_get_file_name(WalWriteMethod *wwmethod,
                               const char *pathname, const char *temp_suffix);
static ssize_t tar_write(Walfile *f, const void *buf, size_t count);
static int tar_sync(Walfile *f);
static bool tar_finish(WalWriteMethod *wwmethod);
static void tar_free(WalWriteMethod *wwmethod);
 
const WalWriteMethodOps WalTarMethodOps = {
    .open_for_write = tar_open_for_write,
    .close = tar_close,
    .existsfile = tar_existsfile,
    .get_file_size = tar_get_file_size,
    .get_file_name = tar_get_file_name,
    .write = tar_write,
    .sync = tar_sync,
    .finish = tar_finish,
    .free = tar_free
};
 
typedef struct TarMethodFile
{
    Walfile     base;
    off_t       ofs_start;      /* Where does the *header* for this file start */
    char        header[TAR_BLOCK_SIZE];
    size_t      pad_to_size;
} TarMethodFile;
 
typedef struct TarMethodData
{
    WalWriteMethod  base;
    char       *tarfilename;
    int         fd;
    TarMethodFile *currentfile;
#ifdef HAVE_LIBZ
    z_streamp   zp;
    void       *zlibOut;
#endif
} TarMethodData;
 
#ifdef HAVE_LIBZ
static bool
tar_write_compressed_data(TarMethodData *tar_data, void *buf, size_t count,
                          bool flush)
{
    tar_data->zp->next_in = buf;
    tar_data->zp->avail_in = count;
 
    while (tar_data->zp->avail_in || flush)
    {
        int         r;
 
        r = deflate(tar_data->zp, flush ? Z_FINISH : Z_NO_FLUSH);
        if (r == Z_STREAM_ERROR)
        {
            tar_data->base.lasterrstring = "could not compress data";
            return false;
        }
 
        if (tar_data->zp->avail_out < ZLIB_OUT_SIZE)
        {
            size_t      len = ZLIB_OUT_SIZE - tar_data->zp->avail_out;
 
            errno = 0;
            if (write(tar_data->fd, tar_data->zlibOut, len) != len)
            {
                /* If write didn't set errno, assume problem is no disk space */
                tar_data->base.lasterrno = errno ? errno : ENOSPC;
                return false;
            }
 
            tar_data->zp->next_out = tar_data->zlibOut;
            tar_data->zp->avail_out = ZLIB_OUT_SIZE;
        }
 
        if (r == Z_STREAM_END)
            break;
    }
 
    if (flush)
    {
        /* Reset the stream for writing */
        if (deflateReset(tar_data->zp) != Z_OK)
        {
            tar_data->base.lasterrstring = "could not reset compression stream";
            return false;
        }
    }
 
    return true;
}
#endif
 
static ssize_t
tar_write(Walfile *f, const void *buf, size_t count)
{
    TarMethodData *tar_data = (TarMethodData *) f->wwmethod;
    ssize_t     r;
 
    Assert(f != NULL);
    clear_error(f->wwmethod);
 
    /* Tarfile will always be positioned at the end */
    if (f->wwmethod->compression_algorithm == PG_COMPRESSION_NONE)
    {
        errno = 0;
        r = write(tar_data->fd, buf, count);
        if (r != count)
        {
            /* If write didn't set errno, assume problem is no disk space */
            f->wwmethod->lasterrno = errno ? errno : ENOSPC;
            return -1;
        }
        f->currpos += r;
        return r;
    }
#ifdef HAVE_LIBZ
    else if (f->wwmethod->compression_algorithm == PG_COMPRESSION_GZIP)
    {
        if (!tar_write_compressed_data(tar_data, unconstify(void *, buf),
                                       count, false))
            return -1;
        f->currpos += count;
        return count;
    }
#endif
    else
    {
        /* Can't happen - compression enabled with no method set */
        f->wwmethod->lasterrno = ENOSYS;
        return -1;
    }
}
 
static bool
tar_write_padding_data(TarMethodFile *f, size_t bytes)
{
    PGAlignedXLogBlock zerobuf;
    size_t      bytesleft = bytes;
 
    memset(zerobuf.data, 0, XLOG_BLCKSZ);
    while (bytesleft)
    {
        size_t      bytestowrite = Min(bytesleft, XLOG_BLCKSZ);
        ssize_t     r = tar_write(&f->base, zerobuf.data, bytestowrite);
 
        if (r < 0)
            return false;
        bytesleft -= r;
    }
 
    return true;
}
 
static char *
tar_get_file_name(WalWriteMethod *wwmethod, const char *pathname,
                  const char *temp_suffix)
{
    char       *filename = pg_malloc0(MAXPGPATH * sizeof(char));
 
    snprintf(filename, MAXPGPATH, "%s%s",
             pathname, temp_suffix ? temp_suffix : "");
 
    return filename;
}
 
static Walfile *
tar_open_for_write(WalWriteMethod *wwmethod, const char *pathname,
                   const char *temp_suffix, size_t pad_to_size)
{
    TarMethodData *tar_data = (TarMethodData *) wwmethod;
    char       *tmppath;
 
    clear_error(wwmethod);
 
    if (tar_data->fd < 0)
    {
        /*
         * We open the tar file only when we first try to write to it.
         */
        tar_data->fd = open(tar_data->tarfilename,
                            O_WRONLY | O_CREAT | PG_BINARY,
                            pg_file_create_mode);
        if (tar_data->fd < 0)
        {
            wwmethod->lasterrno = errno;
            return NULL;
        }
 
#ifdef HAVE_LIBZ
        if (wwmethod->compression_algorithm == PG_COMPRESSION_GZIP)
        {
            tar_data->zp = (z_streamp) pg_malloc(sizeof(z_stream));
            tar_data->zp->zalloc = Z_NULL;
            tar_data->zp->zfree = Z_NULL;
            tar_data->zp->opaque = Z_NULL;
            tar_data->zp->next_out = tar_data->zlibOut;
            tar_data->zp->avail_out = ZLIB_OUT_SIZE;
 
            /*
             * Initialize deflation library. Adding the magic value 16 to the
             * default 15 for the windowBits parameter makes the output be
             * gzip instead of zlib.
             */
            if (deflateInit2(tar_data->zp, wwmethod->compression_level,
                             Z_DEFLATED, 15 + 16, 8, Z_DEFAULT_STRATEGY) != Z_OK)
            {
                pg_free(tar_data->zp);
                tar_data->zp = NULL;
                wwmethod->lasterrstring =
                    "could not initialize compression library";
                return NULL;
            }
        }
#endif
 
        /* There's no tar header itself, the file starts with regular files */
    }
 
    if (tar_data->currentfile != NULL)
    {
        wwmethod->lasterrstring =
            "implementation error: tar files can't have more than one open file";
        return NULL;
    }
 
    tar_data->currentfile = pg_malloc0(sizeof(TarMethodFile));
    tar_data->currentfile->base.wwmethod = wwmethod;
 
    tmppath = tar_get_file_name(wwmethod, pathname, temp_suffix);
 
    /* Create a header with size set to 0 - we will fill out the size on close */
    if (tarCreateHeader(tar_data->currentfile->header, tmppath, NULL, 0, S_IRUSR | S_IWUSR, 0, 0, time(NULL)) != TAR_OK)
    {
        pg_free(tar_data->currentfile);
        pg_free(tmppath);
        tar_data->currentfile = NULL;
        wwmethod->lasterrstring = "could not create tar header";
        return NULL;
    }
 
    pg_free(tmppath);
 
#ifdef HAVE_LIBZ
    if (wwmethod->compression_algorithm == PG_COMPRESSION_GZIP)
    {
        /* Flush existing data */
        if (!tar_write_compressed_data(tar_data, NULL, 0, true))
            return NULL;
 
        /* Turn off compression for header */
        if (deflateParams(tar_data->zp, 0, Z_DEFAULT_STRATEGY) != Z_OK)
        {
            wwmethod->lasterrstring =
                "could not change compression parameters";
            return NULL;
        }
    }
#endif
 
    tar_data->currentfile->ofs_start = lseek(tar_data->fd, 0, SEEK_CUR);
    if (tar_data->currentfile->ofs_start == -1)
    {
        wwmethod->lasterrno = errno;
        pg_free(tar_data->currentfile);
        tar_data->currentfile = NULL;
        return NULL;
    }
    tar_data->currentfile->base.currpos = 0;
 
    if (wwmethod->compression_algorithm == PG_COMPRESSION_NONE)
    {
        errno = 0;
        if (write(tar_data->fd, tar_data->currentfile->header,
                  TAR_BLOCK_SIZE) != TAR_BLOCK_SIZE)
        {
            /* If write didn't set errno, assume problem is no disk space */
            wwmethod->lasterrno = errno ? errno : ENOSPC;
            pg_free(tar_data->currentfile);
            tar_data->currentfile = NULL;
            return NULL;
        }
    }
#ifdef HAVE_LIBZ
    else if (wwmethod->compression_algorithm == PG_COMPRESSION_GZIP)
    {
        /* Write header through the zlib APIs but with no compression */
        if (!tar_write_compressed_data(tar_data, tar_data->currentfile->header,
                                       TAR_BLOCK_SIZE, true))
            return NULL;
 
        /* Re-enable compression for the rest of the file */
        if (deflateParams(tar_data->zp, wwmethod->compression_level,
                          Z_DEFAULT_STRATEGY) != Z_OK)
        {
            wwmethod->lasterrstring = "could not change compression parameters";
            return NULL;
        }
    }
#endif
    else
    {
        /* not reachable */
        Assert(false);
    }
 
    tar_data->currentfile->base.pathname = pg_strdup(pathname);
 
    /*
     * Uncompressed files are padded on creation, but for compression we can't
     * do that
     */
    if (pad_to_size)
    {
        tar_data->currentfile->pad_to_size = pad_to_size;
        if (wwmethod->compression_algorithm == PG_COMPRESSION_NONE)
        {
            /* Uncompressed, so pad now */
            if (!tar_write_padding_data(tar_data->currentfile, pad_to_size))
                return NULL;
            /* Seek back to start */
            if (lseek(tar_data->fd,
                      tar_data->currentfile->ofs_start + TAR_BLOCK_SIZE,
                      SEEK_SET) != tar_data->currentfile->ofs_start + TAR_BLOCK_SIZE)
            {
                wwmethod->lasterrno = errno;
                return NULL;
            }
 
            tar_data->currentfile->base.currpos = 0;
        }
    }
 
    return &tar_data->currentfile->base;
}
 
static ssize_t
tar_get_file_size(WalWriteMethod *wwmethod, const char *pathname)
{
    clear_error(wwmethod);
 
    /* Currently not used, so not supported */
    wwmethod->lasterrno = ENOSYS;
    return -1;
}
 
static int
tar_sync(Walfile *f)
{
    TarMethodData *tar_data = (TarMethodData *) f->wwmethod;
    int         r;
 
    Assert(f != NULL);
    clear_error(f->wwmethod);
 
    if (!f->wwmethod->sync)
        return 0;
 
    /*
     * Always sync the whole tarfile, because that's all we can do. This makes
     * no sense on compressed files, so just ignore those.
     */
    if (f->wwmethod->compression_algorithm != PG_COMPRESSION_NONE)
        return 0;
 
    r = fsync(tar_data->fd);
    if (r < 0)
        f->wwmethod->lasterrno = errno;
    return r;
}
 
static int
tar_close(Walfile *f, WalCloseMethod method)
{
    ssize_t     filesize;
    int         padding;
    TarMethodData *tar_data = (TarMethodData *) f->wwmethod;
    TarMethodFile *tf = (TarMethodFile *) f;
 
    Assert(f != NULL);
    clear_error(f->wwmethod);
 
    if (method == CLOSE_UNLINK)
    {
        if (f->wwmethod->compression_algorithm != PG_COMPRESSION_NONE)
        {
            f->wwmethod->lasterrstring = "unlink not supported with compression";
            return -1;
        }
 
        /*
         * Unlink the file that we just wrote to the tar. We do this by
         * truncating it to the start of the header. This is safe as we only
         * allow writing of the very last file.
         */
        if (ftruncate(tar_data->fd, tf->ofs_start) != 0)
        {
            f->wwmethod->lasterrno = errno;
            return -1;
        }
 
        pg_free(tf->base.pathname);
        pg_free(tf);
        tar_data->currentfile = NULL;
 
        return 0;
    }
 
    /*
     * Pad the file itself with zeroes if necessary. Note that this is
     * different from the tar format padding -- this is the padding we asked
     * for when the file was opened.
     */
    if (tf->pad_to_size)
    {
        if (f->wwmethod->compression_algorithm == PG_COMPRESSION_GZIP)
        {
            /*
             * A compressed tarfile is padded on close since we cannot know
             * the size of the compressed output until the end.
             */
            size_t      sizeleft = tf->pad_to_size - tf->base.currpos;
 
            if (sizeleft)
            {
                if (!tar_write_padding_data(tf, sizeleft))
                    return -1;
            }
        }
        else
        {
            /*
             * An uncompressed tarfile was padded on creation, so just adjust
             * the current position as if we seeked to the end.
             */
            tf->base.currpos = tf->pad_to_size;
        }
    }
 
    /*
     * Get the size of the file, and pad out to a multiple of the tar block
     * size.
     */
    filesize = f->currpos;
    padding = tarPaddingBytesRequired(filesize);
    if (padding)
    {
        char        zerobuf[TAR_BLOCK_SIZE] = {0};
 
        if (tar_write(f, zerobuf, padding) != padding)
            return -1;
    }
 
 
#ifdef HAVE_LIBZ
    if (f->wwmethod->compression_algorithm == PG_COMPRESSION_GZIP)
    {
        /* Flush the current buffer */
        if (!tar_write_compressed_data(tar_data, NULL, 0, true))
            return -1;
    }
#endif
 
    /*
     * Now go back and update the header with the correct filesize and
     * possibly also renaming the file. We overwrite the entire current header
     * when done, including the checksum.
     */
    print_tar_number(&(tf->header[124]), 12, filesize);
 
    if (method == CLOSE_NORMAL)
 
        /*
         * We overwrite it with what it was before if we have no tempname,
         * since we're going to write the buffer anyway.
         */
        strlcpy(&(tf->header[0]), tf->base.pathname, 100);
 
    print_tar_number(&(tf->header[148]), 8, tarChecksum(((TarMethodFile *) f)->header));
    if (lseek(tar_data->fd, tf->ofs_start, SEEK_SET) != ((TarMethodFile *) f)->ofs_start)
    {
        f->wwmethod->lasterrno = errno;
        return -1;
    }
    if (f->wwmethod->compression_algorithm == PG_COMPRESSION_NONE)
    {
        errno = 0;
        if (write(tar_data->fd, tf->header, TAR_BLOCK_SIZE) != TAR_BLOCK_SIZE)
        {
            /* If write didn't set errno, assume problem is no disk space */
            f->wwmethod->lasterrno = errno ? errno : ENOSPC;
            return -1;
        }
    }
#ifdef HAVE_LIBZ
    else if (f->wwmethod->compression_algorithm == PG_COMPRESSION_GZIP)
    {
        /* Turn off compression */
        if (deflateParams(tar_data->zp, 0, Z_DEFAULT_STRATEGY) != Z_OK)
        {
            f->wwmethod->lasterrstring = "could not change compression parameters";
            return -1;
        }
 
        /* Overwrite the header, assuming the size will be the same */
        if (!tar_write_compressed_data(tar_data, tar_data->currentfile->header,
                                       TAR_BLOCK_SIZE, true))
            return -1;
 
        /* Turn compression back on */
        if (deflateParams(tar_data->zp, f->wwmethod->compression_level,
                          Z_DEFAULT_STRATEGY) != Z_OK)
        {
            f->wwmethod->lasterrstring = "could not change compression parameters";
            return -1;
        }
    }
#endif
    else
    {
        /* not reachable */
        Assert(false);
    }
 
    /* Move file pointer back down to end, so we can write the next file */
    if (lseek(tar_data->fd, 0, SEEK_END) < 0)
    {
        f->wwmethod->lasterrno = errno;
        return -1;
    }
 
    /* Always fsync on close, so the padding gets fsynced */
    if (tar_sync(f) < 0)
    {
        /* XXX this seems pretty bogus; why is only this case fatal? */
        pg_fatal("could not fsync file \"%s\": %s",
                 tf->base.pathname, GetLastWalMethodError(f->wwmethod));
    }
 
    /* Clean up and done */
    pg_free(tf->base.pathname);
    pg_free(tf);
    tar_data->currentfile = NULL;
 
    return 0;
}
 
static bool
tar_existsfile(WalWriteMethod *wwmethod, const char *pathname)
{
    clear_error(wwmethod);
    /* We only deal with new tarfiles, so nothing externally created exists */
    return false;
}
 
static bool
tar_finish(WalWriteMethod *wwmethod)
{
    TarMethodData *tar_data = (TarMethodData *) wwmethod;
    char        zerobuf[1024] = {0};
 
    clear_error(wwmethod);
 
    if (tar_data->currentfile)
    {
        if (tar_close(&tar_data->currentfile->base, CLOSE_NORMAL) != 0)
            return false;
    }
 
    /* A tarfile always ends with two empty blocks */
    if (wwmethod->compression_algorithm == PG_COMPRESSION_NONE)
    {
        errno = 0;
        if (write(tar_data->fd, zerobuf, sizeof(zerobuf)) != sizeof(zerobuf))
        {
            /* If write didn't set errno, assume problem is no disk space */
            wwmethod->lasterrno = errno ? errno : ENOSPC;
            return false;
        }
    }
#ifdef HAVE_LIBZ
    else if (wwmethod->compression_algorithm == PG_COMPRESSION_GZIP)
    {
        if (!tar_write_compressed_data(tar_data, zerobuf, sizeof(zerobuf),
                                       false))
            return false;
 
        /* Also flush all data to make sure the gzip stream is finished */
        tar_data->zp->next_in = NULL;
        tar_data->zp->avail_in = 0;
        while (true)
        {
            int         r;
 
            r = deflate(tar_data->zp, Z_FINISH);
 
            if (r == Z_STREAM_ERROR)
            {
                wwmethod->lasterrstring = "could not compress data";
                return false;
            }
            if (tar_data->zp->avail_out < ZLIB_OUT_SIZE)
            {
                size_t      len = ZLIB_OUT_SIZE - tar_data->zp->avail_out;
 
                errno = 0;
                if (write(tar_data->fd, tar_data->zlibOut, len) != len)
                {
                    /*
                     * If write didn't set errno, assume problem is no disk
                     * space.
                     */
                    wwmethod->lasterrno = errno ? errno : ENOSPC;
                    return false;
                }
            }
            if (r == Z_STREAM_END)
                break;
        }
 
        if (deflateEnd(tar_data->zp) != Z_OK)
        {
            wwmethod->lasterrstring = "could not close compression stream";
            return false;
        }
    }
#endif
    else
    {
        /* not reachable */
        Assert(false);
    }
 
    /* sync the empty blocks as well, since they're after the last file */
    if (wwmethod->sync)
    {
        if (fsync(tar_data->fd) != 0)
        {
            wwmethod->lasterrno = errno;
            return false;
        }
    }
 
    if (close(tar_data->fd) != 0)
    {
        wwmethod->lasterrno = errno;
        return false;
    }
 
    tar_data->fd = -1;
 
    if (wwmethod->sync)
    {
        if (fsync_fname(tar_data->tarfilename, false) != 0 ||
            fsync_parent_path(tar_data->tarfilename) != 0)
        {
            wwmethod->lasterrno = errno;
            return false;
        }
    }
 
    return true;
}
 
static void
tar_free(WalWriteMethod *wwmethod)
{
    TarMethodData *tar_data = (TarMethodData *) wwmethod;
 
    pg_free(tar_data->tarfilename);
#ifdef HAVE_LIBZ
    if (wwmethod->compression_algorithm == PG_COMPRESSION_GZIP)
        pg_free(tar_data->zlibOut);
#endif
    pg_free(wwmethod);
}
 
/*
 * The argument compression_algorithm is currently ignored. It is in place for
 * symmetry with CreateWalDirectoryMethod which uses it for distinguishing
 * between the different compression methods. CreateWalTarMethod and its family
 * of functions handle only zlib compression.
 */
WalWriteMethod *
CreateWalTarMethod(const char *tarbase,
                   pg_compress_algorithm compression_algorithm,
                   int compression_level, bool sync)
{
    TarMethodData *wwmethod;
    const char *suffix = (compression_algorithm == PG_COMPRESSION_GZIP) ?
    ".tar.gz" : ".tar";
 
    wwmethod = pg_malloc0(sizeof(TarMethodData));
    *((const WalWriteMethodOps **) &wwmethod->base.ops) =
        &WalTarMethodOps;
    wwmethod->base.compression_algorithm = compression_algorithm;
    wwmethod->base.compression_level = compression_level;
    wwmethod->base.sync = sync;
    clear_error(&wwmethod->base);
 
    wwmethod->tarfilename = pg_malloc0(strlen(tarbase) + strlen(suffix) + 1);
    sprintf(wwmethod->tarfilename, "%s%s", tarbase, suffix);
    wwmethod->fd = -1;
#ifdef HAVE_LIBZ
    if (compression_algorithm == PG_COMPRESSION_GZIP)
        wwmethod->zlibOut = (char *) pg_malloc(ZLIB_OUT_SIZE + 1);
#endif
 
    return &wwmethod->base;
}
 
const char *
GetLastWalMethodError(WalWriteMethod *wwmethod)
{
    if (wwmethod->lasterrstring)
        return wwmethod->lasterrstring;
    return strerror(wwmethod->lasterrno);
}