walmethods.c File Reference

#include "postgres_fe.h"
#include <fcntl.h>
#include <sys/stat.h>
#include <time.h>
#include <unistd.h>
#include "common/file_perm.h"
#include "common/file_utils.h"
#include "common/logging.h"
#include "pgtar.h"
#include "walmethods.h"

Include dependency graph for walmethods.c:

Go to the source code of this file.

Data Structures
struct	DirectoryMethodData
struct	DirectoryMethodFile
struct	TarMethodFile
struct	TarMethodData

Macros
#define	ZLIB_OUT_SIZE   4096
#define	LZ4_IN_SIZE   4096
#define	clear_error(wwmethod)    ((wwmethod)->lasterrstring = NULL, (wwmethod)->lasterrno = 0)

Typedefs
typedef struct DirectoryMethodData	DirectoryMethodData
typedef struct DirectoryMethodFile	DirectoryMethodFile
typedef struct TarMethodFile	TarMethodFile
typedef struct TarMethodData	TarMethodData

Functions
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)
WalWriteMethod *	CreateWalDirectoryMethod (const char *basedir, pg_compress_algorithm compression_algorithm, int compression_level, bool sync)
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)
static bool	tar_write_padding_data (TarMethodFile *f, size_t bytes)
WalWriteMethod *	CreateWalTarMethod (const char *tarbase, pg_compress_algorithm compression_algorithm, int compression_level, bool sync)
const char *	GetLastWalMethodError (WalWriteMethod *wwmethod)

Variables
static const WalWriteMethodOps	WalDirectoryMethodOps
static const WalWriteMethodOps	WalTarMethodOps

Macro Definition Documentation

clear_error

#define clear_error

(

wwmethod

)

((wwmethod)->lasterrstring = NULL, (wwmethod)->lasterrno = 0)

Definition at line 98 of file walmethods.c.

LZ4_IN_SIZE

#define LZ4_IN_SIZE   4096

Definition at line 36 of file walmethods.c.

ZLIB_OUT_SIZE

#define ZLIB_OUT_SIZE   4096

Definition at line 33 of file walmethods.c.

Typedef Documentation

DirectoryMethodData

typedef struct DirectoryMethodData DirectoryMethodData

DirectoryMethodFile

typedef struct DirectoryMethodFile DirectoryMethodFile

TarMethodData

typedef struct TarMethodData TarMethodData

TarMethodFile

typedef struct TarMethodFile TarMethodFile

Function Documentation

CreateWalDirectoryMethod()

WalWriteMethod * CreateWalDirectoryMethod	(	const char *	basedir,
		pg_compress_algorithm	compression_algorithm,
		int	compression_level,
		bool	sync
	)

Definition at line 640 of file walmethods.c.

{
    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;
}

References DirectoryMethodData::base, basedir, DirectoryMethodData::basedir, clear_error, compression_algorithm, WalWriteMethod::compression_algorithm, WalWriteMethod::compression_level, WalWriteMethod::ops, pg_malloc0(), pg_strdup(), WalWriteMethod::sync, and WalDirectoryMethodOps.

Referenced by LogStreamerMain(), and StreamLog().

CreateWalTarMethod()

WalWriteMethod * CreateWalTarMethod	(	const char *	tarbase,
		pg_compress_algorithm	compression_algorithm,
		int	compression_level,
		bool	sync
	)

Definition at line 1355 of file walmethods.c.

{
    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;
}

References TarMethodData::base, clear_error, compression_algorithm, WalWriteMethod::compression_algorithm, WalWriteMethod::compression_level, TarMethodData::fd, WalWriteMethod::ops, PG_COMPRESSION_GZIP, pg_malloc(), pg_malloc0(), sprintf, WalWriteMethod::sync, TarMethodData::tarfilename, WalTarMethodOps, and ZLIB_OUT_SIZE.

Referenced by LogStreamerMain().

dir_close()

static int dir_close ( Walfile *  f, WalCloseMethod  method  )

static

Definition at line 385 of file walmethods.c.

{
    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;
}

References Assert(), DirectoryMethodFile::base, DirectoryMethodData::basedir, clear_error, close, CLOSE_NORMAL, CLOSE_UNLINK, WalWriteMethod::compression_algorithm, dir_get_file_name(), durable_rename(), DirectoryMethodFile::fd, filename, fsync_fname(), fsync_parent_path(), DirectoryMethodFile::fullpath, WalWriteMethod::lasterrno, WalWriteMethod::lasterrstring, MAXPGPATH, Walfile::pathname, PG_COMPRESSION_GZIP, PG_COMPRESSION_LZ4, pg_free(), pg_log_error, snprintf, WalWriteMethod::sync, DirectoryMethodFile::temp_suffix, write, and Walfile::wwmethod.

dir_existsfile()

static bool dir_existsfile ( WalWriteMethod *  wwmethod, const char *  pathname  )

static

Definition at line 584 of file walmethods.c.

{
    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)
 
        /*
         * Skip setting dir_data->lasterrno here because we are only checking
         * for existence.
         */
        return false;
    close(fd);
    return true;
}

References DirectoryMethodData::basedir, clear_error, close, fd(), MAXPGPATH, PG_BINARY, and snprintf.

dir_finish()

static bool dir_finish ( WalWriteMethod *  wwmethod )

static

Definition at line 608 of file walmethods.c.

{
    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;
}

References DirectoryMethodData::basedir, clear_error, fsync_fname(), WalWriteMethod::lasterrno, and WalWriteMethod::sync.

dir_free()

static void dir_free ( WalWriteMethod *  wwmethod )

static

Definition at line 630 of file walmethods.c.

{
    DirectoryMethodData *dir_data = (DirectoryMethodData *) wwmethod;
 
    pg_free(dir_data->basedir);
    pg_free(wwmethod);
}

References DirectoryMethodData::basedir, and pg_free().

dir_get_file_name()

static char * dir_get_file_name ( WalWriteMethod *  wwmethod, const char *  pathname, const char *  temp_suffix  )

static

Definition at line 102 of file walmethods.c.

{
    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;
}

References WalWriteMethod::compression_algorithm, filename, MAXPGPATH, PG_COMPRESSION_GZIP, PG_COMPRESSION_LZ4, pg_malloc0(), and snprintf.

Referenced by dir_close(), and dir_open_for_write().

dir_get_file_size()

static ssize_t dir_get_file_size ( WalWriteMethod *  wwmethod, const char *  pathname  )

static

Definition at line 565 of file walmethods.c.

{
    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;
}

References DirectoryMethodData::basedir, WalWriteMethod::lasterrno, MAXPGPATH, snprintf, stat::st_size, and stat.

dir_open_for_write()

static Walfile * dir_open_for_write ( WalWriteMethod *  wwmethod, const char *  pathname, const char *  temp_suffix, size_t  pad_to_size  )

static

Definition at line 117 of file walmethods.c.

{
    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;
}

References DirectoryMethodFile::base, DirectoryMethodData::basedir, clear_error, close, WalWriteMethod::compression_algorithm, WalWriteMethod::compression_level, Walfile::currpos, dir_get_file_name(), DirectoryMethodFile::fd, fd(), filename, fsync_fname(), fsync_parent_path(), DirectoryMethodFile::fullpath, WalWriteMethod::lasterrno, WalWriteMethod::lasterrstring, LZ4_IN_SIZE, MAXPGPATH, Walfile::pathname, PG_BINARY, PG_COMPRESSION_GZIP, PG_COMPRESSION_LZ4, PG_COMPRESSION_NONE, pg_file_create_mode, pg_free(), pg_malloc0(), pg_pwrite_zeros(), pg_strdup(), snprintf, WalWriteMethod::sync, DirectoryMethodFile::temp_suffix, write, and Walfile::wwmethod.

dir_sync()

static int dir_sync ( Walfile *  f )

static

Definition at line 514 of file walmethods.c.

{
    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;
}

References Assert(), clear_error, WalWriteMethod::compression_algorithm, DirectoryMethodFile::fd, fd(), fsync, WalWriteMethod::lasterrno, WalWriteMethod::lasterrstring, PG_COMPRESSION_GZIP, PG_COMPRESSION_LZ4, WalWriteMethod::sync, write, and Walfile::wwmethod.

dir_write()

static ssize_t dir_write ( Walfile *  f, const void *  buf, size_t  count  )

static

Definition at line 304 of file walmethods.c.

{
    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;
}

References Assert(), DirectoryMethodFile::base, buf, clear_error, WalWriteMethod::compression_algorithm, Walfile::currpos, DirectoryMethodFile::fd, WalWriteMethod::lasterrno, WalWriteMethod::lasterrstring, LZ4_IN_SIZE, PG_COMPRESSION_GZIP, PG_COMPRESSION_LZ4, remaining, write, and Walfile::wwmethod.

GetLastWalMethodError()

const char * GetLastWalMethodError

(

WalWriteMethod *

wwmethod

)

Definition at line 1383 of file walmethods.c.

{
    if (wwmethod->lasterrstring)
        return wwmethod->lasterrstring;
    return strerror(wwmethod->lasterrno);
}

References WalWriteMethod::lasterrno, WalWriteMethod::lasterrstring, and strerror.

Referenced by close_walfile(), HandleCopyStream(), mark_file_as_archived(), open_walfile(), ProcessKeepaliveMsg(), ProcessWALDataMsg(), ReceiveXlogStream(), tar_close(), and writeTimeLineHistoryFile().

tar_close()

static int tar_close ( Walfile *  f, WalCloseMethod  method  )

static

Definition at line 1042 of file walmethods.c.

{
    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[TAR_OFFSET_SIZE]), 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[TAR_OFFSET_NAME]), tf->base.pathname, 100);
 
    print_tar_number(&(tf->header[TAR_OFFSET_CHECKSUM]), 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;
}

References _, Assert(), clear_error, CLOSE_NORMAL, CLOSE_UNLINK, WalWriteMethod::compression_algorithm, WalWriteMethod::compression_level, TarMethodData::currentfile, Walfile::currpos, TarMethodData::fd, GetLastWalMethodError(), TarMethodFile::header, WalWriteMethod::lasterrno, WalWriteMethod::lasterrstring, TarMethodFile::pad_to_size, PG_COMPRESSION_GZIP, PG_COMPRESSION_NONE, pg_fatal, pg_free(), print_tar_number(), strlcpy(), TAR_BLOCK_SIZE, TAR_OFFSET_CHECKSUM, TAR_OFFSET_NAME, TAR_OFFSET_SIZE, tar_sync(), tar_write(), tar_write_padding_data(), tarChecksum(), tarPaddingBytesRequired(), write, and Walfile::wwmethod.

Referenced by tar_finish().

tar_existsfile()

static bool tar_existsfile ( WalWriteMethod *  wwmethod, const char *  pathname  )

static

Definition at line 1219 of file walmethods.c.

{
    clear_error(wwmethod);
    /* We only deal with new tarfiles, so nothing externally created exists */
    return false;
}

References clear_error.

tar_finish()

static bool tar_finish ( WalWriteMethod *  wwmethod )

static

Definition at line 1227 of file walmethods.c.

{
    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;
}

References _, Assert(), TarMethodFile::base, clear_error, close, CLOSE_NORMAL, WalWriteMethod::compression_algorithm, TarMethodData::currentfile, TarMethodData::fd, fsync, fsync_fname(), fsync_parent_path(), WalWriteMethod::lasterrno, WalWriteMethod::lasterrstring, len, PG_COMPRESSION_GZIP, PG_COMPRESSION_NONE, WalWriteMethod::sync, tar_close(), TarMethodData::tarfilename, write, and ZLIB_OUT_SIZE.

tar_free()

static void tar_free ( WalWriteMethod *  wwmethod )

static

Definition at line 1336 of file walmethods.c.

{
    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);
}

References WalWriteMethod::compression_algorithm, PG_COMPRESSION_GZIP, pg_free(), and TarMethodData::tarfilename.

tar_get_file_name()

static char * tar_get_file_name ( WalWriteMethod *  wwmethod, const char *  pathname, const char *  temp_suffix  )

static

Definition at line 825 of file walmethods.c.

{
    char       *filename = pg_malloc0(MAXPGPATH * sizeof(char));
 
    snprintf(filename, MAXPGPATH, "%s%s",
             pathname, temp_suffix ? temp_suffix : "");
 
    return filename;
}

References filename, MAXPGPATH, pg_malloc0(), and snprintf.

Referenced by tar_open_for_write().

tar_get_file_size()

static ssize_t tar_get_file_size ( WalWriteMethod *  wwmethod, const char *  pathname  )

static

Definition at line 1007 of file walmethods.c.

{
    clear_error(wwmethod);
 
    /* Currently not used, so not supported */
    wwmethod->lasterrno = ENOSYS;
    return -1;
}

References clear_error, and WalWriteMethod::lasterrno.

tar_open_for_write()

static Walfile * tar_open_for_write ( WalWriteMethod *  wwmethod, const char *  pathname, const char *  temp_suffix, size_t  pad_to_size  )

static

Definition at line 837 of file walmethods.c.

{
    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;
}

References _, Assert(), TarMethodFile::base, clear_error, WalWriteMethod::compression_algorithm, WalWriteMethod::compression_level, TarMethodData::currentfile, Walfile::currpos, TarMethodData::fd, TarMethodFile::header, WalWriteMethod::lasterrno, WalWriteMethod::lasterrstring, TarMethodFile::ofs_start, TarMethodFile::pad_to_size, Walfile::pathname, PG_BINARY, PG_COMPRESSION_GZIP, PG_COMPRESSION_NONE, pg_file_create_mode, pg_free(), pg_malloc(), pg_malloc0(), pg_strdup(), S_IRUSR, S_IWUSR, TAR_BLOCK_SIZE, tar_get_file_name(), TAR_OK, tar_write_padding_data(), tarCreateHeader(), TarMethodData::tarfilename, write, Walfile::wwmethod, and ZLIB_OUT_SIZE.

tar_sync()

static int tar_sync ( Walfile *  f )

static

Definition at line 1017 of file walmethods.c.

{
    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;
}

References Assert(), clear_error, WalWriteMethod::compression_algorithm, TarMethodData::fd, fsync, WalWriteMethod::lasterrno, PG_COMPRESSION_NONE, WalWriteMethod::sync, and Walfile::wwmethod.

Referenced by tar_close().

tar_write()

static ssize_t tar_write ( Walfile *  f, const void *  buf, size_t  count  )

static

Definition at line 765 of file walmethods.c.

{
    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, 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;
    }
}

References Assert(), buf, clear_error, WalWriteMethod::compression_algorithm, Walfile::currpos, TarMethodData::fd, WalWriteMethod::lasterrno, PG_COMPRESSION_GZIP, PG_COMPRESSION_NONE, write, and Walfile::wwmethod.

Referenced by tar_close(), and tar_write_padding_data().

tar_write_padding_data()

static bool tar_write_padding_data ( TarMethodFile *  f, size_t  bytes  )

static

Definition at line 805 of file walmethods.c.

{
    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;
}

References TarMethodFile::base, PGAlignedXLogBlock::data, Min, and tar_write().

Referenced by tar_close(), and tar_open_for_write().

Variable Documentation

WalDirectoryMethodOps

const WalWriteMethodOps WalDirectoryMethodOps

static

Initial value:

= {
    .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
}

Definition at line 58 of file walmethods.c.

Referenced by CreateWalDirectoryMethod().

WalTarMethodOps

const WalWriteMethodOps WalTarMethodOps

static

Initial value:

= {
    .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
}

Definition at line 679 of file walmethods.c.

Referenced by CreateWalTarMethod().