snapbuild.c File Reference

#include "postgres.h"
#include <sys/stat.h>
#include <unistd.h>
#include "access/heapam_xlog.h"
#include "access/transam.h"
#include "access/xact.h"
#include "common/file_utils.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "replication/logical.h"
#include "replication/reorderbuffer.h"
#include "replication/snapbuild.h"
#include "replication/snapbuild_internal.h"
#include "storage/fd.h"
#include "storage/lmgr.h"
#include "storage/proc.h"
#include "storage/procarray.h"
#include "storage/standby.h"
#include "utils/builtins.h"
#include "utils/memutils.h"
#include "utils/snapmgr.h"
#include "utils/snapshot.h"
#include "utils/wait_event.h"

Include dependency graph for snapbuild.c:

Go to the source code of this file.

Macros
#define	SnapBuildOnDiskConstantSize    offsetof(SnapBuildOnDisk, builder)
#define	SnapBuildOnDiskNotChecksummedSize    offsetof(SnapBuildOnDisk, version)
#define	SNAPBUILD_MAGIC   0x51A1E001
#define	SNAPBUILD_VERSION   6

Functions
static void	SnapBuildPurgeOlderTxn (SnapBuild *builder)
static Snapshot	SnapBuildBuildSnapshot (SnapBuild *builder)
static void	SnapBuildFreeSnapshot (Snapshot snap)
static void	SnapBuildSnapIncRefcount (Snapshot snap)
static void	SnapBuildDistributeSnapshotAndInval (SnapBuild *builder, XLogRecPtr lsn, TransactionId xid)
static bool	SnapBuildXidHasCatalogChanges (SnapBuild *builder, TransactionId xid, uint32 xinfo)
static bool	SnapBuildFindSnapshot (SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *running)
static void	SnapBuildWaitSnapshot (xl_running_xacts *running, TransactionId cutoff)
static void	SnapBuildSerialize (SnapBuild *builder, XLogRecPtr lsn)
static bool	SnapBuildRestore (SnapBuild *builder, XLogRecPtr lsn)
static void	SnapBuildRestoreContents (int fd, void *dest, Size size, const char *path)
SnapBuild *	AllocateSnapshotBuilder (ReorderBuffer *reorder, TransactionId xmin_horizon, XLogRecPtr start_lsn, bool need_full_snapshot, bool in_slot_creation, XLogRecPtr two_phase_at)
void	FreeSnapshotBuilder (SnapBuild *builder)
SnapBuildState	SnapBuildCurrentState (SnapBuild *builder)
XLogRecPtr	SnapBuildGetTwoPhaseAt (SnapBuild *builder)
void	SnapBuildSetTwoPhaseAt (SnapBuild *builder, XLogRecPtr ptr)
bool	SnapBuildXactNeedsSkip (SnapBuild *builder, XLogRecPtr ptr)
void	SnapBuildSnapDecRefcount (Snapshot snap)
Snapshot	SnapBuildInitialSnapshot (SnapBuild *builder)
const char *	SnapBuildExportSnapshot (SnapBuild *builder)
Snapshot	SnapBuildGetOrBuildSnapshot (SnapBuild *builder)
void	SnapBuildClearExportedSnapshot (void)
void	SnapBuildResetExportedSnapshotState (void)
bool	SnapBuildProcessChange (SnapBuild *builder, TransactionId xid, XLogRecPtr lsn)
void	SnapBuildProcessNewCid (SnapBuild *builder, TransactionId xid, XLogRecPtr lsn, xl_heap_new_cid *xlrec)
static void	SnapBuildAddCommittedTxn (SnapBuild *builder, TransactionId xid)
void	SnapBuildCommitTxn (SnapBuild *builder, XLogRecPtr lsn, TransactionId xid, int nsubxacts, TransactionId *subxacts, uint32 xinfo)
void	SnapBuildProcessRunningXacts (SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *running)
void	SnapBuildSerializationPoint (SnapBuild *builder, XLogRecPtr lsn)
bool	SnapBuildRestoreSnapshot (SnapBuildOnDisk *ondisk, XLogRecPtr lsn, MemoryContext context, bool missing_ok)
void	CheckPointSnapBuild (void)
bool	SnapBuildSnapshotExists (XLogRecPtr lsn)

Variables
static ResourceOwner	SavedResourceOwnerDuringExport = NULL
static bool	ExportInProgress = false

Macro Definition Documentation

SNAPBUILD_MAGIC

#define SNAPBUILD_MAGIC   0x51A1E001

Definition at line 1478 of file snapbuild.c.

SNAPBUILD_VERSION

#define SNAPBUILD_VERSION   6

Definition at line 1479 of file snapbuild.c.

SnapBuildOnDiskConstantSize

#define SnapBuildOnDiskConstantSize    offsetof(SnapBuildOnDisk, builder)

Definition at line 1473 of file snapbuild.c.

{
    if (builder->state < SNAPBUILD_CONSISTENT)
        SnapBuildRestore(builder, lsn);
    else
        SnapBuildSerialize(builder, lsn);
}
 
/*
 * Serialize the snapshot 'builder' at the location 'lsn' if it hasn't already
 * been done by another decoding process.
 */
static void
SnapBuildSerialize(SnapBuild *builder, XLogRecPtr lsn)
{
    Size        needed_length;
    SnapBuildOnDisk *ondisk = NULL;
    TransactionId *catchange_xip = NULL;
    MemoryContext old_ctx;
    size_t      catchange_xcnt;
    char       *ondisk_c;
    int         fd;
    char        tmppath[MAXPGPATH];
    char        path[MAXPGPATH];
    int         ret;
    struct stat stat_buf;
    Size        sz;
 
    Assert(XLogRecPtrIsValid(lsn));
    Assert(!XLogRecPtrIsValid(builder->last_serialized_snapshot) ||
           builder->last_serialized_snapshot <= lsn);
 
    /*
     * no point in serializing if we cannot continue to work immediately after
     * restoring the snapshot
     */
    if (builder->state < SNAPBUILD_CONSISTENT)
        return;
 
    /* consistent snapshots have no next phase */
    Assert(builder->next_phase_at == InvalidTransactionId);
 
    /*
     * We identify snapshots by the LSN they are valid for. We don't need to
     * include timelines in the name as each LSN maps to exactly one timeline
     * unless the user used pg_resetwal or similar. If a user did so, there's
     * no hope continuing to decode anyway.
     */
    sprintf(path, "%s/%X-%X.snap",
            PG_LOGICAL_SNAPSHOTS_DIR,
            LSN_FORMAT_ARGS(lsn));
 
    /*
     * first check whether some other backend already has written the snapshot
     * for this LSN. It's perfectly fine if there's none, so we accept ENOENT
     * as a valid state. Everything else is an unexpected error.
     */
    ret = stat(path, &stat_buf);
 
    if (ret != 0 && errno != ENOENT)
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not stat file \"%s\": %m", path)));
 
    else if (ret == 0)
    {
        /*
         * somebody else has already serialized to this point, don't overwrite
         * but remember location, so we don't need to read old data again.
         *
         * To be sure it has been synced to disk after the rename() from the
         * tempfile filename to the real filename, we just repeat the fsync.
         * That ought to be cheap because in most scenarios it should already
         * be safely on disk.
         */
        fsync_fname(path, false);
        fsync_fname(PG_LOGICAL_SNAPSHOTS_DIR, true);
 
        builder->last_serialized_snapshot = lsn;
        goto out;
    }
 
    /*
     * there is an obvious race condition here between the time we stat(2) the
     * file and us writing the file. But we rename the file into place
     * atomically and all files created need to contain the same data anyway,
     * so this is perfectly fine, although a bit of a resource waste. Locking
     * seems like pointless complication.
     */
    elog(DEBUG1, "serializing snapshot to %s", path);
 
    /* to make sure only we will write to this tempfile, include pid */
    sprintf(tmppath, "%s/%X-%X.snap.%d.tmp",
            PG_LOGICAL_SNAPSHOTS_DIR,
            LSN_FORMAT_ARGS(lsn), MyProcPid);
 
    /*
     * Unlink temporary file if it already exists, needs to have been before a
     * crash/error since we won't enter this function twice from within a
     * single decoding slot/backend and the temporary file contains the pid of
     * the current process.
     */
    if (unlink(tmppath) != 0 && errno != ENOENT)
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not remove file \"%s\": %m", tmppath)));
 
    old_ctx = MemoryContextSwitchTo(builder->context);
 
    /* Get the catalog modifying transactions that are yet not committed */
    catchange_xip = ReorderBufferGetCatalogChangesXacts(builder->reorder);
    catchange_xcnt = dclist_count(&builder->reorder->catchange_txns);
 
    needed_length = sizeof(SnapBuildOnDisk) +
        sizeof(TransactionId) * (builder->committed.xcnt + catchange_xcnt);
 
    ondisk_c = palloc0(needed_length);
    ondisk = (SnapBuildOnDisk *) ondisk_c;
    ondisk->magic = SNAPBUILD_MAGIC;
    ondisk->version = SNAPBUILD_VERSION;
    ondisk->length = needed_length;
    INIT_CRC32C(ondisk->checksum);
    COMP_CRC32C(ondisk->checksum,
                ((char *) ondisk) + SnapBuildOnDiskNotChecksummedSize,
                SnapBuildOnDiskConstantSize - SnapBuildOnDiskNotChecksummedSize);
    ondisk_c += sizeof(SnapBuildOnDisk);
 
    memcpy(&ondisk->builder, builder, sizeof(SnapBuild));
    /* NULL-ify memory-only data */
    ondisk->builder.context = NULL;
    ondisk->builder.snapshot = NULL;
    ondisk->builder.reorder = NULL;
    ondisk->builder.committed.xip = NULL;
    ondisk->builder.catchange.xip = NULL;
    /* update catchange only on disk data */
    ondisk->builder.catchange.xcnt = catchange_xcnt;
 
    COMP_CRC32C(ondisk->checksum,
                &ondisk->builder,
                sizeof(SnapBuild));
 
    /* copy committed xacts */
    if (builder->committed.xcnt > 0)
    {
        sz = sizeof(TransactionId) * builder->committed.xcnt;
        memcpy(ondisk_c, builder->committed.xip, sz);
        COMP_CRC32C(ondisk->checksum, ondisk_c, sz);
        ondisk_c += sz;
    }
 
    /* copy catalog modifying xacts */
    if (catchange_xcnt > 0)
    {
        sz = sizeof(TransactionId) * catchange_xcnt;
        memcpy(ondisk_c, catchange_xip, sz);
        COMP_CRC32C(ondisk->checksum, ondisk_c, sz);
        ondisk_c += sz;
    }
 
    FIN_CRC32C(ondisk->checksum);
 
    /* we have valid data now, open tempfile and write it there */
    fd = OpenTransientFile(tmppath,
                           O_CREAT | O_EXCL | O_WRONLY | PG_BINARY);
    if (fd < 0)
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not open file \"%s\": %m", tmppath)));
 
    errno = 0;
    pgstat_report_wait_start(WAIT_EVENT_SNAPBUILD_WRITE);
    if ((write(fd, ondisk, needed_length)) != needed_length)
    {
        int         save_errno = errno;
 
        CloseTransientFile(fd);
 
        /* if write didn't set errno, assume problem is no disk space */
        errno = save_errno ? save_errno : ENOSPC;
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not write to file \"%s\": %m", tmppath)));
    }
    pgstat_report_wait_end();
 
    /*
     * fsync the file before renaming so that even if we crash after this we
     * have either a fully valid file or nothing.
     *
     * It's safe to just ERROR on fsync() here because we'll retry the whole
     * operation including the writes.
     *
     * TODO: Do the fsync() via checkpoints/restartpoints, doing it here has
     * some noticeable overhead since it's performed synchronously during
     * decoding?
     */
    pgstat_report_wait_start(WAIT_EVENT_SNAPBUILD_SYNC);
    if (pg_fsync(fd) != 0)
    {
        int         save_errno = errno;
 
        CloseTransientFile(fd);
        errno = save_errno;
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not fsync file \"%s\": %m", tmppath)));
    }
    pgstat_report_wait_end();
 
    if (CloseTransientFile(fd) != 0)
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not close file \"%s\": %m", tmppath)));
 
    fsync_fname(PG_LOGICAL_SNAPSHOTS_DIR, true);
 
    /*
     * We may overwrite the work from some other backend, but that's ok, our
     * snapshot is valid as well, we'll just have done some superfluous work.
     */
    if (rename(tmppath, path) != 0)
    {
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not rename file \"%s\" to \"%s\": %m",
                        tmppath, path)));
    }
 
    /* make sure we persist */
    fsync_fname(path, false);
    fsync_fname(PG_LOGICAL_SNAPSHOTS_DIR, true);
 
    /*
     * Now there's no way we can lose the dumped state anymore, remember this
     * as a serialization point.
     */
    builder->last_serialized_snapshot = lsn;
 
    MemoryContextSwitchTo(old_ctx);
 
out:
    ReorderBufferSetRestartPoint(builder->reorder,
                                 builder->last_serialized_snapshot);
    /* be tidy */
    if (ondisk)
        pfree(ondisk);
    if (catchange_xip)
        pfree(catchange_xip);
}
 
/*
 * Restore the logical snapshot file contents to 'ondisk'.
 *
 * 'context' is the memory context where the catalog modifying/committed xid
 * will live.
 * If 'missing_ok' is true, will not throw an error if the file is not found.
 */
bool
SnapBuildRestoreSnapshot(SnapBuildOnDisk *ondisk, XLogRecPtr lsn,
                         MemoryContext context, bool missing_ok)
{
    int         fd;
    pg_crc32c   checksum;
    Size        sz;
    char        path[MAXPGPATH];
 
    sprintf(path, "%s/%X-%X.snap",
            PG_LOGICAL_SNAPSHOTS_DIR,
            LSN_FORMAT_ARGS(lsn));
 
    fd = OpenTransientFile(path, O_RDONLY | PG_BINARY);
 
    if (fd < 0)
    {
        if (missing_ok && errno == ENOENT)
            return false;
 
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not open file \"%s\": %m", path)));
    }
 
    /* ----
     * Make sure the snapshot had been stored safely to disk, that's normally
     * cheap.
     * Note that we do not need PANIC here, nobody will be able to use the
     * slot without fsyncing, and saving it won't succeed without an fsync()
     * either...
     * ----
     */
    fsync_fname(path, false);
    fsync_fname(PG_LOGICAL_SNAPSHOTS_DIR, true);
 
    /* read statically sized portion of snapshot */
    SnapBuildRestoreContents(fd, ondisk, SnapBuildOnDiskConstantSize, path);
 
    if (ondisk->magic != SNAPBUILD_MAGIC)
        ereport(ERROR,
                (errcode(ERRCODE_DATA_CORRUPTED),
                 errmsg("snapbuild state file \"%s\" has wrong magic number: %u instead of %u",
                        path, ondisk->magic, SNAPBUILD_MAGIC)));
 
    if (ondisk->version != SNAPBUILD_VERSION)
        ereport(ERROR,
                (errcode(ERRCODE_DATA_CORRUPTED),
                 errmsg("snapbuild state file \"%s\" has unsupported version: %u instead of %u",
                        path, ondisk->version, SNAPBUILD_VERSION)));
 
    INIT_CRC32C(checksum);
    COMP_CRC32C(checksum,
                ((char *) ondisk) + SnapBuildOnDiskNotChecksummedSize,
                SnapBuildOnDiskConstantSize - SnapBuildOnDiskNotChecksummedSize);
 
    /* read SnapBuild */
    SnapBuildRestoreContents(fd, &ondisk->builder, sizeof(SnapBuild), path);
    COMP_CRC32C(checksum, &ondisk->builder, sizeof(SnapBuild));
 
    /* restore committed xacts information */
    if (ondisk->builder.committed.xcnt > 0)
    {
        sz = sizeof(TransactionId) * ondisk->builder.committed.xcnt;
        ondisk->builder.committed.xip = MemoryContextAllocZero(context, sz);
        SnapBuildRestoreContents(fd, ondisk->builder.committed.xip, sz, path);
        COMP_CRC32C(checksum, ondisk->builder.committed.xip, sz);
    }
 
    /* restore catalog modifying xacts information */
    if (ondisk->builder.catchange.xcnt > 0)
    {
        sz = sizeof(TransactionId) * ondisk->builder.catchange.xcnt;
        ondisk->builder.catchange.xip = MemoryContextAllocZero(context, sz);
        SnapBuildRestoreContents(fd, ondisk->builder.catchange.xip, sz, path);
        COMP_CRC32C(checksum, ondisk->builder.catchange.xip, sz);
    }
 
    if (CloseTransientFile(fd) != 0)
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not close file \"%s\": %m", path)));
 
    FIN_CRC32C(checksum);
 
    /* verify checksum of what we've read */
    if (!EQ_CRC32C(checksum, ondisk->checksum))
        ereport(ERROR,
                (errcode(ERRCODE_DATA_CORRUPTED),
                 errmsg("checksum mismatch for snapbuild state file \"%s\": is %u, should be %u",
                        path, checksum, ondisk->checksum)));
 
    return true;
}
 
/*
 * Restore a snapshot into 'builder' if previously one has been stored at the
 * location indicated by 'lsn'. Returns true if successful, false otherwise.
 */
static bool
SnapBuildRestore(SnapBuild *builder, XLogRecPtr lsn)
{
    SnapBuildOnDisk ondisk;
 
    /* no point in loading a snapshot if we're already there */
    if (builder->state == SNAPBUILD_CONSISTENT)
        return false;
 
    /* validate and restore the snapshot to 'ondisk' */
    if (!SnapBuildRestoreSnapshot(&ondisk, lsn, builder->context, true))
        return false;
 
    /*
     * ok, we now have a sensible snapshot here, figure out if it has more
     * information than we have.
     */
 
    /*
     * We are only interested in consistent snapshots for now, comparing
     * whether one incomplete snapshot is more "advanced" seems to be
     * unnecessarily complex.
     */
    if (ondisk.builder.state < SNAPBUILD_CONSISTENT)
        goto snapshot_not_interesting;
 
    /*
     * Don't use a snapshot that requires an xmin that we cannot guarantee to
     * be available.
     */
    if (TransactionIdPrecedes(ondisk.builder.xmin, builder->initial_xmin_horizon))
        goto snapshot_not_interesting;
 
    /*
     * Consistent snapshots have no next phase. Reset next_phase_at as it is
     * possible that an old value may remain.
     */
    Assert(ondisk.builder.next_phase_at == InvalidTransactionId);
    builder->next_phase_at = InvalidTransactionId;
 
    /* ok, we think the snapshot is sensible, copy over everything important */
    builder->xmin = ondisk.builder.xmin;
    builder->xmax = ondisk.builder.xmax;
    builder->state = ondisk.builder.state;
 
    builder->committed.xcnt = ondisk.builder.committed.xcnt;
    /* We only allocated/stored xcnt, not xcnt_space xids ! */
    /* don't overwrite preallocated xip, if we don't have anything here */
    if (builder->committed.xcnt > 0)
    {
        pfree(builder->committed.xip);
        builder->committed.xcnt_space = ondisk.builder.committed.xcnt;
        builder->committed.xip = ondisk.builder.committed.xip;
    }
    ondisk.builder.committed.xip = NULL;
 
    /* set catalog modifying transactions */
    if (builder->catchange.xip)
        pfree(builder->catchange.xip);
    builder->catchange.xcnt = ondisk.builder.catchange.xcnt;
    builder->catchange.xip = ondisk.builder.catchange.xip;
    ondisk.builder.catchange.xip = NULL;
 
    /* our snapshot is not interesting anymore, build a new one */
    if (builder->snapshot != NULL)
    {
        SnapBuildSnapDecRefcount(builder->snapshot);
    }
    builder->snapshot = SnapBuildBuildSnapshot(builder);
    SnapBuildSnapIncRefcount(builder->snapshot);
 
    ReorderBufferSetRestartPoint(builder->reorder, lsn);
 
    Assert(builder->state == SNAPBUILD_CONSISTENT);
 
    ereport(LogicalDecodingLogLevel(),
            errmsg("logical decoding found consistent point at %X/%08X",
                   LSN_FORMAT_ARGS(lsn)),
            errdetail("Logical decoding will begin using saved snapshot."));
    return true;
 
snapshot_not_interesting:
    if (ondisk.builder.committed.xip != NULL)
        pfree(ondisk.builder.committed.xip);
    if (ondisk.builder.catchange.xip != NULL)
        pfree(ondisk.builder.catchange.xip);
    return false;
}
 
/*
 * Read the contents of the serialized snapshot to 'dest'.
 */
static void
SnapBuildRestoreContents(int fd, void *dest, Size size, const char *path)
{
    int         readBytes;
 
    pgstat_report_wait_start(WAIT_EVENT_SNAPBUILD_READ);
    readBytes = read(fd, dest, size);
    pgstat_report_wait_end();
    if (readBytes != size)
    {
        int         save_errno = errno;
 
        CloseTransientFile(fd);
 
        if (readBytes < 0)
        {
            errno = save_errno;
            ereport(ERROR,
                    (errcode_for_file_access(),
                     errmsg("could not read file \"%s\": %m", path)));
        }
        else
            ereport(ERROR,
                    (errcode(ERRCODE_DATA_CORRUPTED),
                     errmsg("could not read file \"%s\": read %d of %zu",
                            path, readBytes, size)));
    }
}
 
/*
 * Remove all serialized snapshots that are not required anymore because no
 * slot can need them. This doesn't actually have to run during a checkpoint,
 * but it's a convenient point to schedule this.
 *
 * NB: We run this during checkpoints even if logical decoding is disabled so
 * we cleanup old slots at some point after it got disabled.
 */
void
CheckPointSnapBuild(void)
{
    XLogRecPtr  cutoff;
    XLogRecPtr  redo;
    DIR        *snap_dir;
    struct dirent *snap_de;
    char        path[MAXPGPATH + sizeof(PG_LOGICAL_SNAPSHOTS_DIR)];
 
    /*
     * We start off with a minimum of the last redo pointer. No new
     * replication slot will start before that, so that's a safe upper bound
     * for removal.
     */
    redo = GetRedoRecPtr();
 
    /* now check for the restart ptrs from existing slots */
    cutoff = ReplicationSlotsComputeLogicalRestartLSN();
 
    /* don't start earlier than the restart lsn */
    if (redo < cutoff)
        cutoff = redo;
 
    snap_dir = AllocateDir(PG_LOGICAL_SNAPSHOTS_DIR);
    while ((snap_de = ReadDir(snap_dir, PG_LOGICAL_SNAPSHOTS_DIR)) != NULL)
    {
        uint32      hi;
        uint32      lo;
        XLogRecPtr  lsn;
        PGFileType  de_type;
 
        if (strcmp(snap_de->d_name, ".") == 0 ||
            strcmp(snap_de->d_name, "..") == 0)
            continue;
 
        snprintf(path, sizeof(path), "%s/%s", PG_LOGICAL_SNAPSHOTS_DIR, snap_de->d_name);
        de_type = get_dirent_type(path, snap_de, false, DEBUG1);
 
        if (de_type != PGFILETYPE_ERROR && de_type != PGFILETYPE_REG)
        {
            elog(DEBUG1, "only regular files expected: %s", path);
            continue;
        }
 
        /*
         * temporary filenames from SnapBuildSerialize() include the LSN and
         * everything but are postfixed by .$pid.tmp. We can just remove them
         * the same as other files because there can be none that are
         * currently being written that are older than cutoff.
         *
         * We just log a message if a file doesn't fit the pattern, it's
         * probably some editors lock/state file or similar...
         */
        if (sscanf(snap_de->d_name, "%X-%X.snap", &hi, &lo) != 2)
        {
            ereport(LOG,
                    (errmsg("could not parse file name \"%s\"", path)));
            continue;
        }
 
        lsn = ((uint64) hi) << 32 | lo;
 
        /* check whether we still need it */
        if (lsn < cutoff || !XLogRecPtrIsValid(cutoff))
        {
            elog(DEBUG1, "removing snapbuild snapshot %s", path);
 
            /*
             * It's not particularly harmful, though strange, if we can't
             * remove the file here. Don't prevent the checkpoint from
             * completing, that'd be a cure worse than the disease.
             */
            if (unlink(path) < 0)
            {
                ereport(LOG,
                        (errcode_for_file_access(),
                         errmsg("could not remove file \"%s\": %m",
                                path)));
                continue;
            }
        }
    }
    FreeDir(snap_dir);
}
 
/*
 * Check if a logical snapshot at the specified point has been serialized.
 */
bool
SnapBuildSnapshotExists(XLogRecPtr lsn)
{
    char        path[MAXPGPATH];
    int         ret;
    struct stat stat_buf;
 
    sprintf(path, "%s/%X-%X.snap",
            PG_LOGICAL_SNAPSHOTS_DIR,
            LSN_FORMAT_ARGS(lsn));
 
    ret = stat(path, &stat_buf);
 
    if (ret != 0 && errno != ENOENT)
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not stat file \"%s\": %m", path)));
 
    return ret == 0;
}

SnapBuildOnDiskNotChecksummedSize

#define SnapBuildOnDiskNotChecksummedSize    offsetof(SnapBuildOnDisk, version)

Definition at line 1475 of file snapbuild.c.

Function Documentation

AllocateSnapshotBuilder()

SnapBuild * AllocateSnapshotBuilder	(	ReorderBuffer *	reorder,
		TransactionId	xmin_horizon,
		XLogRecPtr	start_lsn,
		bool	need_full_snapshot,
		bool	in_slot_creation,
		XLogRecPtr	two_phase_at
	)

Definition at line 189 of file snapbuild.c.

{
    MemoryContext context;
    MemoryContext oldcontext;
    SnapBuild  *builder;
 
    /* allocate memory in own context, to have better accountability */
    context = AllocSetContextCreate(CurrentMemoryContext,
                                    "snapshot builder context",
                                    ALLOCSET_DEFAULT_SIZES);
    oldcontext = MemoryContextSwitchTo(context);
 
    builder = palloc0_object(SnapBuild);
 
    builder->state = SNAPBUILD_START;
    builder->context = context;
    builder->reorder = reorder;
    /* Other struct members initialized by zeroing via palloc0 above */
 
    builder->committed.xcnt = 0;
    builder->committed.xcnt_space = 128;    /* arbitrary number */
    builder->committed.xip =
        palloc0_array(TransactionId, builder->committed.xcnt_space);
    builder->committed.includes_all_transactions = true;
 
    builder->catchange.xcnt = 0;
    builder->catchange.xip = NULL;
 
    builder->initial_xmin_horizon = xmin_horizon;
    builder->start_decoding_at = start_lsn;
    builder->in_slot_creation = in_slot_creation;
    builder->building_full_snapshot = need_full_snapshot;
    builder->two_phase_at = two_phase_at;
 
    MemoryContextSwitchTo(oldcontext);
 
    return builder;
}

References ALLOCSET_DEFAULT_SIZES, AllocSetContextCreate, SnapBuild::building_full_snapshot, SnapBuild::catchange, SnapBuild::committed, SnapBuild::context, CurrentMemoryContext, fb(), SnapBuild::in_slot_creation, SnapBuild::includes_all_transactions, SnapBuild::initial_xmin_horizon, MemoryContextSwitchTo(), palloc0_array, palloc0_object, SnapBuild::reorder, SNAPBUILD_START, SnapBuild::start_decoding_at, SnapBuild::state, SnapBuild::two_phase_at, SnapBuild::xcnt, SnapBuild::xcnt_space, and SnapBuild::xip.

Referenced by StartupDecodingContext().

CheckPointSnapBuild()

void CheckPointSnapBuild

(

void

)

Definition at line 1974 of file snapbuild.c.

{
    XLogRecPtr  cutoff;
    XLogRecPtr  redo;
    DIR        *snap_dir;
    struct dirent *snap_de;
    char        path[MAXPGPATH + sizeof(PG_LOGICAL_SNAPSHOTS_DIR)];
 
    /*
     * We start off with a minimum of the last redo pointer. No new
     * replication slot will start before that, so that's a safe upper bound
     * for removal.
     */
    redo = GetRedoRecPtr();
 
    /* now check for the restart ptrs from existing slots */
    cutoff = ReplicationSlotsComputeLogicalRestartLSN();
 
    /* don't start earlier than the restart lsn */
    if (redo < cutoff)
        cutoff = redo;
 
    snap_dir = AllocateDir(PG_LOGICAL_SNAPSHOTS_DIR);
    while ((snap_de = ReadDir(snap_dir, PG_LOGICAL_SNAPSHOTS_DIR)) != NULL)
    {
        uint32      hi;
        uint32      lo;
        XLogRecPtr  lsn;
        PGFileType  de_type;
 
        if (strcmp(snap_de->d_name, ".") == 0 ||
            strcmp(snap_de->d_name, "..") == 0)
            continue;
 
        snprintf(path, sizeof(path), "%s/%s", PG_LOGICAL_SNAPSHOTS_DIR, snap_de->d_name);
        de_type = get_dirent_type(path, snap_de, false, DEBUG1);
 
        if (de_type != PGFILETYPE_ERROR && de_type != PGFILETYPE_REG)
        {
            elog(DEBUG1, "only regular files expected: %s", path);
            continue;
        }
 
        /*
         * temporary filenames from SnapBuildSerialize() include the LSN and
         * everything but are postfixed by .$pid.tmp. We can just remove them
         * the same as other files because there can be none that are
         * currently being written that are older than cutoff.
         *
         * We just log a message if a file doesn't fit the pattern, it's
         * probably some editors lock/state file or similar...
         */
        if (sscanf(snap_de->d_name, "%X-%X.snap", &hi, &lo) != 2)
        {
            ereport(LOG,
                    (errmsg("could not parse file name \"%s\"", path)));
            continue;
        }
 
        lsn = ((uint64) hi) << 32 | lo;
 
        /* check whether we still need it */
        if (lsn < cutoff || !XLogRecPtrIsValid(cutoff))
        {
            elog(DEBUG1, "removing snapbuild snapshot %s", path);
 
            /*
             * It's not particularly harmful, though strange, if we can't
             * remove the file here. Don't prevent the checkpoint from
             * completing, that'd be a cure worse than the disease.
             */
            if (unlink(path) < 0)
            {
                ereport(LOG,
                        (errcode_for_file_access(),
                         errmsg("could not remove file \"%s\": %m",
                                path)));
                continue;
            }
        }
    }
    FreeDir(snap_dir);
}

References AllocateDir(), DEBUG1, elog, ereport, errcode_for_file_access(), errmsg, fb(), FreeDir(), get_dirent_type(), GetRedoRecPtr(), LOG, MAXPGPATH, PG_LOGICAL_SNAPSHOTS_DIR, PGFILETYPE_ERROR, PGFILETYPE_REG, ReadDir(), ReplicationSlotsComputeLogicalRestartLSN(), snprintf, and XLogRecPtrIsValid.

Referenced by CheckPointGuts().

FreeSnapshotBuilder()

void FreeSnapshotBuilder

(

SnapBuild *

builder

)

Definition at line 237 of file snapbuild.c.

{
    MemoryContext context = builder->context;
 
    /* free snapshot explicitly, that contains some error checking */
    if (builder->snapshot != NULL)
    {
        SnapBuildSnapDecRefcount(builder->snapshot);
        builder->snapshot = NULL;
    }
 
    /* other resources are deallocated via memory context reset */
    MemoryContextDelete(context);
}

References SnapBuild::context, fb(), MemoryContextDelete(), SnapBuildSnapDecRefcount(), and SnapBuild::snapshot.

Referenced by FreeDecodingContext().

SnapBuildAddCommittedTxn()

static void SnapBuildAddCommittedTxn ( SnapBuild *  builder, TransactionId  xid  )

static

Definition at line 832 of file snapbuild.c.

{
    Assert(TransactionIdIsValid(xid));
 
    if (builder->committed.xcnt == builder->committed.xcnt_space)
    {
        builder->committed.xcnt_space = builder->committed.xcnt_space * 2 + 1;
 
        elog(DEBUG1, "increasing space for committed transactions to %u",
             (uint32) builder->committed.xcnt_space);
 
        builder->committed.xip = repalloc_array(builder->committed.xip,
                                                TransactionId,
                                                builder->committed.xcnt_space);
    }
 
    /*
     * TODO: It might make sense to keep the array sorted here instead of
     * doing it every time we build a new snapshot. On the other hand this
     * gets called repeatedly when a transaction with subtransactions commits.
     */
    builder->committed.xip[builder->committed.xcnt++] = xid;
}

References Assert, SnapBuild::committed, DEBUG1, elog, repalloc_array, TransactionIdIsValid, SnapBuild::xcnt, SnapBuild::xcnt_space, and SnapBuild::xip.

Referenced by SnapBuildCommitTxn().

SnapBuildBuildSnapshot()

static Snapshot SnapBuildBuildSnapshot ( SnapBuild *  builder )

static

Definition at line 364 of file snapbuild.c.

{
    Snapshot    snapshot;
    Size        ssize;
 
    Assert(builder->state >= SNAPBUILD_FULL_SNAPSHOT);
 
    ssize = sizeof(SnapshotData)
        + sizeof(TransactionId) * builder->committed.xcnt
        + sizeof(TransactionId) * 1 /* toplevel xid */ ;
 
    snapshot = MemoryContextAllocZero(builder->context, ssize);
 
    snapshot->snapshot_type = SNAPSHOT_HISTORIC_MVCC;
 
    /*
     * We misuse the original meaning of SnapshotData's xip and subxip fields
     * to make the more fitting for our needs.
     *
     * In the 'xip' array we store transactions that have to be treated as
     * committed. Since we will only ever look at tuples from transactions
     * that have modified the catalog it's more efficient to store those few
     * that exist between xmin and xmax (frequently there are none).
     *
     * Snapshots that are used in transactions that have modified the catalog
     * also use the 'subxip' array to store their toplevel xid and all the
     * subtransaction xids so we can recognize when we need to treat rows as
     * visible that are not in xip but still need to be visible. Subxip only
     * gets filled when the transaction is copied into the context of a
     * catalog modifying transaction since we otherwise share a snapshot
     * between transactions. As long as a txn hasn't modified the catalog it
     * doesn't need to treat any uncommitted rows as visible, so there is no
     * need for those xids.
     *
     * Both arrays are qsort'ed so that we can use bsearch() on them.
     */
    Assert(TransactionIdIsNormal(builder->xmin));
    Assert(TransactionIdIsNormal(builder->xmax));
 
    snapshot->xmin = builder->xmin;
    snapshot->xmax = builder->xmax;
 
    /* store all transactions to be treated as committed by this snapshot */
    snapshot->xip =
        (TransactionId *) ((char *) snapshot + sizeof(SnapshotData));
    snapshot->xcnt = builder->committed.xcnt;
    memcpy(snapshot->xip,
           builder->committed.xip,
           builder->committed.xcnt * sizeof(TransactionId));
 
    /* sort so we can bsearch() */
    qsort(snapshot->xip, snapshot->xcnt, sizeof(TransactionId), xidComparator);
 
    /*
     * Initially, subxip is empty, i.e. it's a snapshot to be used by
     * transactions that don't modify the catalog. Will be filled by
     * ReorderBufferCopySnap() if necessary.
     */
    snapshot->subxcnt = 0;
    snapshot->subxip = NULL;
 
    snapshot->suboverflowed = false;
    snapshot->takenDuringRecovery = false;
    snapshot->copied = false;
    snapshot->curcid = FirstCommandId;
    snapshot->active_count = 0;
    snapshot->regd_count = 0;
    snapshot->snapXactCompletionCount = 0;
 
    return snapshot;
}

References SnapshotData::active_count, Assert, SnapBuild::committed, SnapBuild::context, SnapshotData::copied, SnapshotData::curcid, fb(), FirstCommandId, memcpy(), MemoryContextAllocZero(), qsort, SnapshotData::regd_count, SNAPBUILD_FULL_SNAPSHOT, SNAPSHOT_HISTORIC_MVCC, SnapshotData::snapshot_type, SnapshotData::snapXactCompletionCount, SnapBuild::state, SnapshotData::suboverflowed, SnapshotData::subxcnt, SnapshotData::subxip, SnapshotData::takenDuringRecovery, TransactionIdIsNormal, SnapBuild::xcnt, SnapshotData::xcnt, xidComparator(), SnapBuild::xip, SnapshotData::xip, SnapBuild::xmax, SnapshotData::xmax, SnapBuild::xmin, and SnapshotData::xmin.

Referenced by SnapBuildCommitTxn(), SnapBuildGetOrBuildSnapshot(), SnapBuildInitialSnapshot(), SnapBuildProcessChange(), and SnapBuildRestore().

SnapBuildClearExportedSnapshot()

void SnapBuildClearExportedSnapshot

(

void

)

Definition at line 603 of file snapbuild.c.

{
    ResourceOwner tmpResOwner;
 
    /* nothing exported, that is the usual case */
    if (!ExportInProgress)
        return;
 
    if (!IsTransactionState())
        elog(ERROR, "clearing exported snapshot in wrong transaction state");
 
    /*
     * AbortCurrentTransaction() takes care of resetting the snapshot state,
     * so remember SavedResourceOwnerDuringExport.
     */
    tmpResOwner = SavedResourceOwnerDuringExport;
 
    /* make sure nothing could have ever happened */
    AbortCurrentTransaction();
 
    CurrentResourceOwner = tmpResOwner;
}

References AbortCurrentTransaction(), CurrentResourceOwner, elog, ERROR, ExportInProgress, fb(), IsTransactionState(), and SavedResourceOwnerDuringExport.

Referenced by exec_replication_command().

SnapBuildCommitTxn()

void SnapBuildCommitTxn	(	SnapBuild *	builder,
		XLogRecPtr	lsn,
		TransactionId	xid,
		int	nsubxacts,
		TransactionId *	subxacts,
		uint32	xinfo
	)

Definition at line 944 of file snapbuild.c.

{
    int         nxact;
 
    bool        needs_snapshot = false;
    bool        needs_timetravel = false;
    bool        sub_needs_timetravel = false;
 
    TransactionId xmax = xid;
 
    /*
     * Transactions preceding BUILDING_SNAPSHOT will neither be decoded, nor
     * will they be part of a snapshot.  So we don't need to record anything.
     */
    if (builder->state == SNAPBUILD_START ||
        (builder->state == SNAPBUILD_BUILDING_SNAPSHOT &&
         TransactionIdPrecedes(xid, builder->next_phase_at)))
    {
        /* ensure that only commits after this are getting replayed */
        if (builder->start_decoding_at <= lsn)
            builder->start_decoding_at = lsn + 1;
        return;
    }
 
    if (builder->state < SNAPBUILD_CONSISTENT)
    {
        /* ensure that only commits after this are getting replayed */
        if (builder->start_decoding_at <= lsn)
            builder->start_decoding_at = lsn + 1;
 
        /*
         * If building an exportable snapshot, force xid to be tracked, even
         * if the transaction didn't modify the catalog.
         */
        if (builder->building_full_snapshot)
        {
            needs_timetravel = true;
        }
    }
 
    for (nxact = 0; nxact < nsubxacts; nxact++)
    {
        TransactionId subxid = subxacts[nxact];
 
        /*
         * Add subtransaction to base snapshot if catalog modifying, we don't
         * distinguish to toplevel transactions there.
         */
        if (SnapBuildXidHasCatalogChanges(builder, subxid, xinfo))
        {
            sub_needs_timetravel = true;
            needs_snapshot = true;
 
            elog(DEBUG1, "found subtransaction %u:%u with catalog changes",
                 xid, subxid);
 
            SnapBuildAddCommittedTxn(builder, subxid);
 
            if (NormalTransactionIdFollows(subxid, xmax))
                xmax = subxid;
        }
 
        /*
         * If we're forcing timetravel we also need visibility information
         * about subtransaction, so keep track of subtransaction's state, even
         * if not catalog modifying.  Don't need to distribute a snapshot in
         * that case.
         */
        else if (needs_timetravel)
        {
            SnapBuildAddCommittedTxn(builder, subxid);
            if (NormalTransactionIdFollows(subxid, xmax))
                xmax = subxid;
        }
    }
 
    /* if top-level modified catalog, it'll need a snapshot */
    if (SnapBuildXidHasCatalogChanges(builder, xid, xinfo))
    {
        elog(DEBUG2, "found top level transaction %u, with catalog changes",
             xid);
        needs_snapshot = true;
        needs_timetravel = true;
        SnapBuildAddCommittedTxn(builder, xid);
    }
    else if (sub_needs_timetravel)
    {
        /* track toplevel txn as well, subxact alone isn't meaningful */
        elog(DEBUG2, "forced transaction %u to do timetravel due to one of its subtransactions",
             xid);
        needs_timetravel = true;
        SnapBuildAddCommittedTxn(builder, xid);
    }
    else if (needs_timetravel)
    {
        elog(DEBUG2, "forced transaction %u to do timetravel", xid);
 
        SnapBuildAddCommittedTxn(builder, xid);
    }
 
    if (!needs_timetravel)
    {
        /* record that we cannot export a general snapshot anymore */
        builder->committed.includes_all_transactions = false;
    }
 
    Assert(!needs_snapshot || needs_timetravel);
 
    /*
     * Adjust xmax of the snapshot builder, we only do that for committed,
     * catalog modifying, transactions, everything else isn't interesting for
     * us since we'll never look at the respective rows.
     */
    if (needs_timetravel &&
        (!TransactionIdIsValid(builder->xmax) ||
         TransactionIdFollowsOrEquals(xmax, builder->xmax)))
    {
        builder->xmax = xmax;
        TransactionIdAdvance(builder->xmax);
    }
 
    /* if there's any reason to build a historic snapshot, do so now */
    if (needs_snapshot)
    {
        /*
         * If we haven't built a complete snapshot yet there's no need to hand
         * it out, it wouldn't (and couldn't) be used anyway.
         */
        if (builder->state < SNAPBUILD_FULL_SNAPSHOT)
            return;
 
        /*
         * Decrease the snapshot builder's refcount of the old snapshot, note
         * that it still will be used if it has been handed out to the
         * reorderbuffer earlier.
         */
        if (builder->snapshot)
            SnapBuildSnapDecRefcount(builder->snapshot);
 
        builder->snapshot = SnapBuildBuildSnapshot(builder);
 
        /* we might need to execute invalidations, add snapshot */
        if (!ReorderBufferXidHasBaseSnapshot(builder->reorder, xid))
        {
            SnapBuildSnapIncRefcount(builder->snapshot);
            ReorderBufferSetBaseSnapshot(builder->reorder, xid, lsn,
                                         builder->snapshot);
        }
 
        /* refcount of the snapshot builder for the new snapshot */
        SnapBuildSnapIncRefcount(builder->snapshot);
 
        /*
         * Add a new catalog snapshot and invalidations messages to all
         * currently running transactions.
         */
        SnapBuildDistributeSnapshotAndInval(builder, lsn, xid);
    }
}

References Assert, SnapBuild::building_full_snapshot, SnapBuild::committed, DEBUG1, DEBUG2, elog, fb(), SnapBuild::includes_all_transactions, SnapBuild::next_phase_at, NormalTransactionIdFollows, SnapBuild::reorder, ReorderBufferSetBaseSnapshot(), ReorderBufferXidHasBaseSnapshot(), SNAPBUILD_BUILDING_SNAPSHOT, SNAPBUILD_CONSISTENT, SNAPBUILD_FULL_SNAPSHOT, SNAPBUILD_START, SnapBuildAddCommittedTxn(), SnapBuildBuildSnapshot(), SnapBuildDistributeSnapshotAndInval(), SnapBuildSnapDecRefcount(), SnapBuildSnapIncRefcount(), SnapBuildXidHasCatalogChanges(), SnapBuild::snapshot, SnapBuild::start_decoding_at, SnapBuild::state, TransactionIdAdvance, TransactionIdFollowsOrEquals(), TransactionIdIsValid, TransactionIdPrecedes(), and SnapBuild::xmax.

Referenced by DecodeCommit().

SnapBuildCurrentState()

SnapBuildState SnapBuildCurrentState

(

SnapBuild *

builder

)

Definition at line 281 of file snapbuild.c.

{
    return builder->state;
}

References SnapBuild::state.

Referenced by DecodePrepare(), DecodingContextReady(), heap2_decode(), heap_decode(), logicalmsg_decode(), ReorderBufferCanStartStreaming(), and xact_decode().

SnapBuildDistributeSnapshotAndInval()

static void SnapBuildDistributeSnapshotAndInval ( SnapBuild *  builder, XLogRecPtr  lsn, TransactionId  xid  )

static

Definition at line 734 of file snapbuild.c.

{
    dlist_iter  txn_i;
    ReorderBufferTXN *txn;
 
    /*
     * Iterate through all toplevel transactions. This can include
     * subtransactions which we just don't yet know to be that, but that's
     * fine, they will just get an unnecessary snapshot and invalidations
     * queued.
     */
    dlist_foreach(txn_i, &builder->reorder->toplevel_by_lsn)
    {
        txn = dlist_container(ReorderBufferTXN, node, txn_i.cur);
 
        Assert(TransactionIdIsValid(txn->xid));
 
        /*
         * If we don't have a base snapshot yet, there are no changes in this
         * transaction which in turn implies we don't yet need a snapshot at
         * all. We'll add a snapshot when the first change gets queued.
         *
         * Similarly, we don't need to add invalidations to a transaction
         * whose base snapshot is not yet set. Once a base snapshot is built,
         * it will include the xids of committed transactions that have
         * modified the catalog, thus reflecting the new catalog contents. The
         * existing catalog cache will have already been invalidated after
         * processing the invalidations in the transaction that modified
         * catalogs, ensuring that a fresh cache is constructed during
         * decoding.
         *
         * NB: This works correctly even for subtransactions because
         * ReorderBufferAssignChild() takes care to transfer the base snapshot
         * to the top-level transaction, and while iterating the changequeue
         * we'll get the change from the subtxn.
         */
        if (!ReorderBufferXidHasBaseSnapshot(builder->reorder, txn->xid))
            continue;
 
        /*
         * We don't need to add snapshot or invalidations to prepared
         * transactions as they should not see the new catalog contents.
         */
        if (rbtxn_is_prepared(txn))
            continue;
 
        elog(DEBUG2, "adding a new snapshot and invalidations to %u at %X/%08X",
             txn->xid, LSN_FORMAT_ARGS(lsn));
 
        /*
         * increase the snapshot's refcount for the transaction we are handing
         * it out to
         */
        SnapBuildSnapIncRefcount(builder->snapshot);
        ReorderBufferAddSnapshot(builder->reorder, txn->xid, lsn,
                                 builder->snapshot);
 
        /*
         * Add invalidation messages to the reorder buffer of in-progress
         * transactions except the current committed transaction, for which we
         * will execute invalidations at the end.
         *
         * It is required, otherwise, we will end up using the stale catcache
         * contents built by the current transaction even after its decoding,
         * which should have been invalidated due to concurrent catalog
         * changing transaction.
         *
         * Distribute only the invalidation messages generated by the current
         * committed transaction. Invalidation messages received from other
         * transactions would have already been propagated to the relevant
         * in-progress transactions. This transaction would have processed
         * those invalidations, ensuring that subsequent transactions observe
         * a consistent cache state.
         */
        if (txn->xid != xid)
        {
            uint32      ninvalidations;
            SharedInvalidationMessage *msgs = NULL;
 
            ninvalidations = ReorderBufferGetInvalidations(builder->reorder,
                                                           xid, &msgs);
 
            if (ninvalidations > 0)
            {
                Assert(msgs != NULL);
 
                ReorderBufferAddDistributedInvalidations(builder->reorder,
                                                         txn->xid, lsn,
                                                         ninvalidations, msgs);
            }
        }
    }
}

References Assert, DEBUG2, dlist_container, dlist_foreach, elog, fb(), LSN_FORMAT_ARGS, rbtxn_is_prepared, SnapBuild::reorder, ReorderBufferAddDistributedInvalidations(), ReorderBufferAddSnapshot(), ReorderBufferGetInvalidations(), ReorderBufferXidHasBaseSnapshot(), SnapBuildSnapIncRefcount(), SnapBuild::snapshot, ReorderBuffer::toplevel_by_lsn, TransactionIdIsValid, and ReorderBufferTXN::xid.

Referenced by SnapBuildCommitTxn().

SnapBuildExportSnapshot()

const char * SnapBuildExportSnapshot

(

SnapBuild *

builder

)

Definition at line 542 of file snapbuild.c.

{
    Snapshot    snap;
    char       *snapname;
 
    if (IsTransactionOrTransactionBlock())
        elog(ERROR, "cannot export a snapshot from within a transaction");
 
    if (SavedResourceOwnerDuringExport)
        elog(ERROR, "can only export one snapshot at a time");
 
    SavedResourceOwnerDuringExport = CurrentResourceOwner;
    ExportInProgress = true;
 
    StartTransactionCommand();
 
    /* There doesn't seem to a nice API to set these */
    XactIsoLevel = XACT_REPEATABLE_READ;
    XactReadOnly = true;
 
    snap = SnapBuildInitialSnapshot(builder);
 
    /*
     * now that we've built a plain snapshot, make it active and use the
     * normal mechanisms for exporting it
     */
    snapname = ExportSnapshot(snap);
 
    ereport(LOG,
            (errmsg_plural("exported logical decoding snapshot: \"%s\" with %u transaction ID",
                           "exported logical decoding snapshot: \"%s\" with %u transaction IDs",
                           snap->xcnt,
                           snapname, snap->xcnt)));
    return snapname;
}

References CurrentResourceOwner, elog, ereport, errmsg_plural(), ERROR, ExportInProgress, ExportSnapshot(), fb(), IsTransactionOrTransactionBlock(), LOG, SavedResourceOwnerDuringExport, SnapBuildInitialSnapshot(), StartTransactionCommand(), XACT_REPEATABLE_READ, XactIsoLevel, and XactReadOnly.

Referenced by CreateReplicationSlot().

SnapBuildFindSnapshot()

static bool SnapBuildFindSnapshot ( SnapBuild *  builder, XLogRecPtr  lsn, xl_running_xacts *  running  )

static

Definition at line 1242 of file snapbuild.c.

{
    /* ---
     * Build catalog decoding snapshot incrementally using information about
     * the currently running transactions. There are several ways to do that:
     *
     * a) There were no running transactions when the xl_running_xacts record
     *    was inserted, jump to CONSISTENT immediately. We might find such a
     *    state while waiting on c)'s sub-states.
     *
     * b) This (in a previous run) or another decoding slot serialized a
     *    snapshot to disk that we can use. Can't use this method while finding
     *    the start point for decoding changes as the restart LSN would be an
     *    arbitrary LSN but we need to find the start point to extract changes
     *    where we won't see the data for partial transactions. Also, we cannot
     *    use this method when a slot needs a full snapshot for export or direct
     *    use, as that snapshot will only contain catalog modifying transactions.
     *
     * c) First incrementally build a snapshot for catalog tuples
     *    (BUILDING_SNAPSHOT), that requires all, already in-progress,
     *    transactions to finish.  Every transaction starting after that
     *    (FULL_SNAPSHOT state), has enough information to be decoded.  But
     *    for older running transactions no viable snapshot exists yet, so
     *    CONSISTENT will only be reached once all of those have finished.
     * ---
     */
 
    /*
     * xl_running_xacts record is older than what we can use, we might not
     * have all necessary catalog rows anymore.
     */
    if (TransactionIdIsNormal(builder->initial_xmin_horizon) &&
        NormalTransactionIdPrecedes(running->oldestRunningXid,
                                    builder->initial_xmin_horizon))
    {
        ereport(DEBUG1,
                errmsg_internal("skipping snapshot at %X/%08X while building logical decoding snapshot, xmin horizon too low",
                                LSN_FORMAT_ARGS(lsn)),
                errdetail_internal("initial xmin horizon of %u vs the snapshot's %u",
                                   builder->initial_xmin_horizon, running->oldestRunningXid));
 
 
        SnapBuildWaitSnapshot(running, builder->initial_xmin_horizon);
 
        return true;
    }
 
    /*
     * a) No transaction were running, we can jump to consistent.
     *
     * This is not affected by races around xl_running_xacts, because we can
     * miss transaction commits, but currently not transactions starting.
     *
     * NB: We might have already started to incrementally assemble a snapshot,
     * so we need to be careful to deal with that.
     */
    if (running->oldestRunningXid == running->nextXid)
    {
        if (!XLogRecPtrIsValid(builder->start_decoding_at) ||
            builder->start_decoding_at <= lsn)
            /* can decode everything after this */
            builder->start_decoding_at = lsn + 1;
 
        /* As no transactions were running xmin/xmax can be trivially set. */
        builder->xmin = running->nextXid;   /* < are finished */
        builder->xmax = running->nextXid;   /* >= are running */
 
        /* so we can safely use the faster comparisons */
        Assert(TransactionIdIsNormal(builder->xmin));
        Assert(TransactionIdIsNormal(builder->xmax));
 
        builder->state = SNAPBUILD_CONSISTENT;
        builder->next_phase_at = InvalidTransactionId;
 
        ereport(LogicalDecodingLogLevel(),
                errmsg("logical decoding found consistent point at %X/%08X",
                       LSN_FORMAT_ARGS(lsn)),
                errdetail("There are no running transactions."));
 
        return false;
    }
 
    /*
     * b) valid on disk state and while neither building full snapshot nor
     * creating a slot.
     */
    else if (!builder->building_full_snapshot &&
             !builder->in_slot_creation &&
             SnapBuildRestore(builder, lsn))
    {
        /* there won't be any state to cleanup */
        return false;
    }
 
    /*
     * c) transition from START to BUILDING_SNAPSHOT.
     *
     * In START state, and a xl_running_xacts record with running xacts is
     * encountered.  In that case, switch to BUILDING_SNAPSHOT state, and
     * record xl_running_xacts->nextXid.  Once all running xacts have finished
     * (i.e. they're all >= nextXid), we have a complete catalog snapshot.  It
     * might look that we could use xl_running_xacts's ->xids information to
     * get there quicker, but that is problematic because transactions marked
     * as running, might already have inserted their commit record - it's
     * infeasible to change that with locking.
     */
    else if (builder->state == SNAPBUILD_START)
    {
        builder->state = SNAPBUILD_BUILDING_SNAPSHOT;
        builder->next_phase_at = running->nextXid;
 
        /*
         * Start with an xmin/xmax that's correct for future, when all the
         * currently running transactions have finished. We'll update both
         * while waiting for the pending transactions to finish.
         */
        builder->xmin = running->nextXid;   /* < are finished */
        builder->xmax = running->nextXid;   /* >= are running */
 
        /* so we can safely use the faster comparisons */
        Assert(TransactionIdIsNormal(builder->xmin));
        Assert(TransactionIdIsNormal(builder->xmax));
 
        ereport(LOG,
                errmsg("logical decoding found initial starting point at %X/%08X",
                       LSN_FORMAT_ARGS(lsn)),
                errdetail("Waiting for transactions (approximately %d) older than %u to end.",
                          running->xcnt, running->nextXid));
 
        SnapBuildWaitSnapshot(running, running->nextXid);
    }
 
    /*
     * c) transition from BUILDING_SNAPSHOT to FULL_SNAPSHOT.
     *
     * In BUILDING_SNAPSHOT state, and this xl_running_xacts' oldestRunningXid
     * is >= than nextXid from when we switched to BUILDING_SNAPSHOT.  This
     * means all transactions starting afterwards have enough information to
     * be decoded.  Switch to FULL_SNAPSHOT.
     */
    else if (builder->state == SNAPBUILD_BUILDING_SNAPSHOT &&
             TransactionIdPrecedesOrEquals(builder->next_phase_at,
                                           running->oldestRunningXid))
    {
        builder->state = SNAPBUILD_FULL_SNAPSHOT;
        builder->next_phase_at = running->nextXid;
 
        ereport(LOG,
                errmsg("logical decoding found initial consistent point at %X/%08X",
                       LSN_FORMAT_ARGS(lsn)),
                errdetail("Waiting for transactions (approximately %d) older than %u to end.",
                          running->xcnt, running->nextXid));
 
        SnapBuildWaitSnapshot(running, running->nextXid);
    }
 
    /*
     * c) transition from FULL_SNAPSHOT to CONSISTENT.
     *
     * In FULL_SNAPSHOT state, and this xl_running_xacts' oldestRunningXid is
     * >= than nextXid from when we switched to FULL_SNAPSHOT.  This means all
     * transactions that are currently in progress have a catalog snapshot,
     * and all their changes have been collected.  Switch to CONSISTENT.
     */
    else if (builder->state == SNAPBUILD_FULL_SNAPSHOT &&
             TransactionIdPrecedesOrEquals(builder->next_phase_at,
                                           running->oldestRunningXid))
    {
        builder->state = SNAPBUILD_CONSISTENT;
        builder->next_phase_at = InvalidTransactionId;
 
        ereport(LogicalDecodingLogLevel(),
                errmsg("logical decoding found consistent point at %X/%08X",
                       LSN_FORMAT_ARGS(lsn)),
                errdetail("There are no old transactions anymore."));
    }
 
    /*
     * We already started to track running xacts and need to wait for all
     * in-progress ones to finish. We fall through to the normal processing of
     * records so incremental cleanup can be performed.
     */
    return true;
}

References Assert, SnapBuild::building_full_snapshot, DEBUG1, ereport, errdetail(), errdetail_internal(), errmsg, errmsg_internal(), SnapBuild::in_slot_creation, SnapBuild::initial_xmin_horizon, InvalidTransactionId, LOG, LogicalDecodingLogLevel, LSN_FORMAT_ARGS, SnapBuild::next_phase_at, xl_running_xacts::nextXid, NormalTransactionIdPrecedes, xl_running_xacts::oldestRunningXid, SNAPBUILD_BUILDING_SNAPSHOT, SNAPBUILD_CONSISTENT, SNAPBUILD_FULL_SNAPSHOT, SNAPBUILD_START, SnapBuildRestore(), SnapBuildWaitSnapshot(), SnapBuild::start_decoding_at, SnapBuild::state, TransactionIdIsNormal, TransactionIdPrecedesOrEquals(), xl_running_xacts::xcnt, XLogRecPtrIsValid, SnapBuild::xmax, and SnapBuild::xmin.

Referenced by SnapBuildProcessRunningXacts().

SnapBuildFreeSnapshot()

static void SnapBuildFreeSnapshot ( Snapshot  snap )

static

Definition at line 256 of file snapbuild.c.

{
    /* make sure we don't get passed an external snapshot */
    Assert(snap->snapshot_type == SNAPSHOT_HISTORIC_MVCC);
 
    /* make sure nobody modified our snapshot */
    Assert(snap->curcid == FirstCommandId);
    Assert(!snap->suboverflowed);
    Assert(!snap->takenDuringRecovery);
    Assert(snap->regd_count == 0);
 
    /* slightly more likely, so it's checked even without c-asserts */
    if (snap->copied)
        elog(ERROR, "cannot free a copied snapshot");
 
    if (snap->active_count)
        elog(ERROR, "cannot free an active snapshot");
 
    pfree(snap);
}

References Assert, elog, ERROR, fb(), FirstCommandId, pfree(), and SNAPSHOT_HISTORIC_MVCC.

Referenced by SnapBuildSnapDecRefcount().

SnapBuildGetOrBuildSnapshot()

Snapshot SnapBuildGetOrBuildSnapshot

(

SnapBuild *

builder

)

Definition at line 582 of file snapbuild.c.

{
    Assert(builder->state == SNAPBUILD_CONSISTENT);
 
    /* only build a new snapshot if we don't have a prebuilt one */
    if (builder->snapshot == NULL)
    {
        builder->snapshot = SnapBuildBuildSnapshot(builder);
        /* increase refcount for the snapshot builder */
        SnapBuildSnapIncRefcount(builder->snapshot);
    }
 
    return builder->snapshot;
}

References Assert, fb(), SNAPBUILD_CONSISTENT, SnapBuildBuildSnapshot(), SnapBuildSnapIncRefcount(), SnapBuild::snapshot, and SnapBuild::state.

Referenced by logicalmsg_decode().

SnapBuildGetTwoPhaseAt()

XLogRecPtr SnapBuildGetTwoPhaseAt

(

SnapBuild *

builder

)

Definition at line 290 of file snapbuild.c.

{
    return builder->two_phase_at;
}

References SnapBuild::two_phase_at.

Referenced by DecodeCommit().

SnapBuildInitialSnapshot()

Snapshot SnapBuildInitialSnapshot

(

SnapBuild *

builder

)

Definition at line 444 of file snapbuild.c.

{
    Snapshot    snap;
    TransactionId xid;
    TransactionId safeXid;
    TransactionId *newxip;
    int         newxcnt = 0;
 
    Assert(XactIsoLevel == XACT_REPEATABLE_READ);
    Assert(builder->building_full_snapshot);
 
    /* don't allow older snapshots */
    InvalidateCatalogSnapshot();    /* about to overwrite MyProc->xmin */
    if (HaveRegisteredOrActiveSnapshot())
        elog(ERROR, "cannot build an initial slot snapshot when snapshots exist");
    Assert(!HistoricSnapshotActive());
 
    if (builder->state != SNAPBUILD_CONSISTENT)
        elog(ERROR, "cannot build an initial slot snapshot before reaching a consistent state");
 
    if (!builder->committed.includes_all_transactions)
        elog(ERROR, "cannot build an initial slot snapshot, not all transactions are monitored anymore");
 
    /* so we don't overwrite the existing value */
    if (TransactionIdIsValid(MyProc->xmin))
        elog(ERROR, "cannot build an initial slot snapshot when MyProc->xmin already is valid");
 
    snap = SnapBuildBuildSnapshot(builder);
 
    /*
     * We know that snap->xmin is alive, enforced by the logical xmin
     * mechanism. Due to that we can do this without locks, we're only
     * changing our own value.
     *
     * Building an initial snapshot is expensive and an unenforced xmin
     * horizon would have bad consequences, therefore always double-check that
     * the horizon is enforced.
     */
    LWLockAcquire(ProcArrayLock, LW_SHARED);
    safeXid = GetOldestSafeDecodingTransactionId(false);
    LWLockRelease(ProcArrayLock);
 
    if (TransactionIdFollows(safeXid, snap->xmin))
        elog(ERROR, "cannot build an initial slot snapshot as oldest safe xid %u follows snapshot's xmin %u",
             safeXid, snap->xmin);
 
    MyProc->xmin = snap->xmin;
 
    /* allocate in transaction context */
    newxip = palloc_array(TransactionId, GetMaxSnapshotXidCount());
 
    /*
     * snapbuild.c builds transactions in an "inverted" manner, which means it
     * stores committed transactions in ->xip, not ones in progress. Build a
     * classical snapshot by marking all non-committed transactions as
     * in-progress. This can be expensive.
     */
    for (xid = snap->xmin; NormalTransactionIdPrecedes(xid, snap->xmax);)
    {
        void       *test;
 
        /*
         * Check whether transaction committed using the decoding snapshot
         * meaning of ->xip.
         */
        test = bsearch(&xid, snap->xip, snap->xcnt,
                       sizeof(TransactionId), xidComparator);
 
        if (test == NULL)
        {
            if (newxcnt >= GetMaxSnapshotXidCount())
                ereport(ERROR,
                        (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
                         errmsg("initial slot snapshot too large")));
 
            newxip[newxcnt++] = xid;
        }
 
        TransactionIdAdvance(xid);
    }
 
    /* adjust remaining snapshot fields as needed */
    snap->snapshot_type = SNAPSHOT_MVCC;
    snap->xcnt = newxcnt;
    snap->xip = newxip;
 
    return snap;
}

References Assert, SnapBuild::building_full_snapshot, SnapBuild::committed, elog, ereport, errcode(), ERRCODE_T_R_SERIALIZATION_FAILURE, errmsg, ERROR, fb(), GetMaxSnapshotXidCount(), GetOldestSafeDecodingTransactionId(), HaveRegisteredOrActiveSnapshot(), HistoricSnapshotActive(), SnapBuild::includes_all_transactions, InvalidateCatalogSnapshot(), LW_SHARED, LWLockAcquire(), LWLockRelease(), MyProc, NormalTransactionIdPrecedes, palloc_array, SNAPBUILD_CONSISTENT, SnapBuildBuildSnapshot(), SNAPSHOT_MVCC, SnapBuild::state, test(), TransactionIdAdvance, TransactionIdFollows(), TransactionIdIsValid, XACT_REPEATABLE_READ, XactIsoLevel, xidComparator(), and PGPROC::xmin.

Referenced by CreateReplicationSlot(), RepackWorkerMain(), and SnapBuildExportSnapshot().

SnapBuildProcessChange()

bool SnapBuildProcessChange	(	SnapBuild *	builder,
		TransactionId	xid,
		XLogRecPtr	lsn
	)

Definition at line 642 of file snapbuild.c.

{
    /*
     * We can't handle data in transactions if we haven't built a snapshot
     * yet, so don't store them.
     */
    if (builder->state < SNAPBUILD_FULL_SNAPSHOT)
        return false;
 
    /*
     * No point in keeping track of changes in transactions that we don't have
     * enough information about to decode. This means that they started before
     * we got into the SNAPBUILD_FULL_SNAPSHOT state.
     */
    if (builder->state < SNAPBUILD_CONSISTENT &&
        TransactionIdPrecedes(xid, builder->next_phase_at))
        return false;
 
    /*
     * If the reorderbuffer doesn't yet have a snapshot, add one now, it will
     * be needed to decode the change we're currently processing.
     */
    if (!ReorderBufferXidHasBaseSnapshot(builder->reorder, xid))
    {
        /* only build a new snapshot if we don't have a prebuilt one */
        if (builder->snapshot == NULL)
        {
            builder->snapshot = SnapBuildBuildSnapshot(builder);
            /* increase refcount for the snapshot builder */
            SnapBuildSnapIncRefcount(builder->snapshot);
        }
 
        /*
         * Increase refcount for the transaction we're handing the snapshot
         * out to.
         */
        SnapBuildSnapIncRefcount(builder->snapshot);
        ReorderBufferSetBaseSnapshot(builder->reorder, xid, lsn,
                                     builder->snapshot);
    }
 
    return true;
}

References fb(), SnapBuild::next_phase_at, SnapBuild::reorder, ReorderBufferSetBaseSnapshot(), ReorderBufferXidHasBaseSnapshot(), SNAPBUILD_CONSISTENT, SNAPBUILD_FULL_SNAPSHOT, SnapBuildBuildSnapshot(), SnapBuildSnapIncRefcount(), SnapBuild::snapshot, SnapBuild::state, and TransactionIdPrecedes().

Referenced by heap2_decode(), heap_decode(), and logicalmsg_decode().

SnapBuildProcessNewCid()

void SnapBuildProcessNewCid	(	SnapBuild *	builder,
		TransactionId	xid,
		XLogRecPtr	lsn,
		xl_heap_new_cid *	xlrec
	)

Definition at line 692 of file snapbuild.c.

{
    CommandId   cid;
 
    /*
     * we only log new_cid's if a catalog tuple was modified, so mark the
     * transaction as containing catalog modifications
     */
    ReorderBufferXidSetCatalogChanges(builder->reorder, xid, lsn);
 
    ReorderBufferAddNewTupleCids(builder->reorder, xlrec->top_xid, lsn,
                                 xlrec->target_locator, xlrec->target_tid,
                                 xlrec->cmin, xlrec->cmax,
                                 xlrec->combocid);
 
    /* figure out new command id */
    if (xlrec->cmin != InvalidCommandId &&
        xlrec->cmax != InvalidCommandId)
        cid = Max(xlrec->cmin, xlrec->cmax);
    else if (xlrec->cmax != InvalidCommandId)
        cid = xlrec->cmax;
    else if (xlrec->cmin != InvalidCommandId)
        cid = xlrec->cmin;
    else
    {
        cid = InvalidCommandId; /* silence compiler */
        elog(ERROR, "xl_heap_new_cid record without a valid CommandId");
    }
 
    ReorderBufferAddNewCommandId(builder->reorder, xid, lsn, cid + 1);
}

References elog, ERROR, fb(), InvalidCommandId, Max, SnapBuild::reorder, ReorderBufferAddNewCommandId(), ReorderBufferAddNewTupleCids(), and ReorderBufferXidSetCatalogChanges().

Referenced by heap2_decode().

SnapBuildProcessRunningXacts()

void SnapBuildProcessRunningXacts	(	SnapBuild *	builder,
		XLogRecPtr	lsn,
		xl_running_xacts *	running
	)

Definition at line 1140 of file snapbuild.c.

{
    ReorderBufferTXN *txn;
    TransactionId xmin;
 
    /*
     * If we're not consistent yet, inspect the record to see whether it
     * allows to get closer to being consistent. If we are consistent, dump
     * our snapshot so others or we, after a restart, can use it.
     */
    if (builder->state < SNAPBUILD_CONSISTENT)
    {
        /* returns false if there's no point in performing cleanup just yet */
        if (!SnapBuildFindSnapshot(builder, lsn, running))
            return;
    }
    else
        SnapBuildSerialize(builder, lsn);
 
    /*
     * Update range of interesting xids based on the running xacts
     * information. We don't increase ->xmax using it, because once we are in
     * a consistent state we can do that ourselves and much more efficiently
     * so, because we only need to do it for catalog transactions since we
     * only ever look at those.
     *
     * NB: We only increase xmax when a catalog modifying transaction commits
     * (see SnapBuildCommitTxn).  Because of this, xmax can be lower than
     * xmin, which looks odd but is correct and actually more efficient, since
     * we hit fast paths in heapam_visibility.c.
     */
    builder->xmin = running->oldestRunningXid;
 
    /* Remove transactions we don't need to keep track off anymore */
    SnapBuildPurgeOlderTxn(builder);
 
    /*
     * Advance the xmin limit for the current replication slot, to allow
     * vacuum to clean up the tuples this slot has been protecting.
     *
     * The reorderbuffer might have an xmin among the currently running
     * snapshots; use it if so.  If not, we need only consider the snapshots
     * we'll produce later, which can't be less than the oldest running xid in
     * the record we're reading now.
     */
    xmin = ReorderBufferGetOldestXmin(builder->reorder);
    if (xmin == InvalidTransactionId)
        xmin = running->oldestRunningXid;
    elog(DEBUG3, "xmin: %u, xmax: %u, oldest running: %u, oldest xmin: %u",
         builder->xmin, builder->xmax, running->oldestRunningXid, xmin);
    LogicalIncreaseXminForSlot(lsn, xmin);
 
    /*
     * Also tell the slot where we can restart decoding from. We don't want to
     * do that after every commit because changing that implies an fsync of
     * the logical slot's state file, so we only do it every time we see a
     * running xacts record.
     *
     * Do so by looking for the oldest in progress transaction (determined by
     * the first LSN of any of its relevant records). Every transaction
     * remembers the last location we stored the snapshot to disk before its
     * beginning. That point is where we can restart from.
     */
 
    /*
     * Can't know about a serialized snapshot's location if we're not
     * consistent.
     */
    if (builder->state < SNAPBUILD_CONSISTENT)
        return;
 
    txn = ReorderBufferGetOldestTXN(builder->reorder);
 
    /*
     * oldest ongoing txn might have started when we didn't yet serialize
     * anything because we hadn't reached a consistent state yet.
     */
    if (txn != NULL && XLogRecPtrIsValid(txn->restart_decoding_lsn))
        LogicalIncreaseRestartDecodingForSlot(lsn, txn->restart_decoding_lsn);
 
    /*
     * No in-progress transaction, can reuse the last serialized snapshot if
     * we have one.
     */
    else if (txn == NULL &&
             XLogRecPtrIsValid(builder->reorder->current_restart_decoding_lsn) &&
             XLogRecPtrIsValid(builder->last_serialized_snapshot))
        LogicalIncreaseRestartDecodingForSlot(lsn,
                                              builder->last_serialized_snapshot);
}

References ReorderBuffer::current_restart_decoding_lsn, DEBUG3, elog, fb(), InvalidTransactionId, SnapBuild::last_serialized_snapshot, LogicalIncreaseRestartDecodingForSlot(), LogicalIncreaseXminForSlot(), xl_running_xacts::oldestRunningXid, SnapBuild::reorder, ReorderBufferGetOldestTXN(), ReorderBufferGetOldestXmin(), ReorderBufferTXN::restart_decoding_lsn, SNAPBUILD_CONSISTENT, SnapBuildFindSnapshot(), SnapBuildPurgeOlderTxn(), SnapBuildSerialize(), SnapBuild::state, XLogRecPtrIsValid, SnapBuild::xmax, and SnapBuild::xmin.

Referenced by standby_decode().

SnapBuildPurgeOlderTxn()

static void SnapBuildPurgeOlderTxn ( SnapBuild *  builder )

static

Definition at line 867 of file snapbuild.c.

{
    int         off;
    TransactionId *workspace;
    int         surviving_xids = 0;
 
    /* not ready yet */
    if (!TransactionIdIsNormal(builder->xmin))
        return;
 
    /* TODO: Neater algorithm than just copying and iterating? */
    workspace =
        MemoryContextAlloc(builder->context,
                           builder->committed.xcnt * sizeof(TransactionId));
 
    /* copy xids that still are interesting to workspace */
    for (off = 0; off < builder->committed.xcnt; off++)
    {
        if (NormalTransactionIdPrecedes(builder->committed.xip[off],
                                        builder->xmin))
            ;                   /* remove */
        else
            workspace[surviving_xids++] = builder->committed.xip[off];
    }
 
    /* copy workspace back to persistent state */
    memcpy(builder->committed.xip, workspace,
           surviving_xids * sizeof(TransactionId));
 
    elog(DEBUG3, "purged committed transactions from %u to %u, xmin: %u, xmax: %u",
         (uint32) builder->committed.xcnt, (uint32) surviving_xids,
         builder->xmin, builder->xmax);
    builder->committed.xcnt = surviving_xids;
 
    pfree(workspace);
 
    /*
     * Purge xids in ->catchange as well. The purged array must also be sorted
     * in xidComparator order.
     */
    if (builder->catchange.xcnt > 0)
    {
        /*
         * Since catchange.xip is sorted, we find the lower bound of xids that
         * are still interesting.
         */
        for (off = 0; off < builder->catchange.xcnt; off++)
        {
            if (TransactionIdFollowsOrEquals(builder->catchange.xip[off],
                                             builder->xmin))
                break;
        }
 
        surviving_xids = builder->catchange.xcnt - off;
 
        if (surviving_xids > 0)
        {
            memmove(builder->catchange.xip, &(builder->catchange.xip[off]),
                    surviving_xids * sizeof(TransactionId));
        }
        else
        {
            pfree(builder->catchange.xip);
            builder->catchange.xip = NULL;
        }
 
        elog(DEBUG3, "purged catalog modifying transactions from %u to %u, xmin: %u, xmax: %u",
             (uint32) builder->catchange.xcnt, (uint32) surviving_xids,
             builder->xmin, builder->xmax);
        builder->catchange.xcnt = surviving_xids;
    }
}

References SnapBuild::catchange, SnapBuild::committed, SnapBuild::context, DEBUG3, elog, fb(), memcpy(), MemoryContextAlloc(), NormalTransactionIdPrecedes, pfree(), TransactionIdFollowsOrEquals(), TransactionIdIsNormal, SnapBuild::xcnt, SnapBuild::xip, SnapBuild::xmax, and SnapBuild::xmin.

Referenced by SnapBuildProcessRunningXacts().

SnapBuildResetExportedSnapshotState()

void SnapBuildResetExportedSnapshotState

(

void

)

Definition at line 630 of file snapbuild.c.

{
    SavedResourceOwnerDuringExport = NULL;
    ExportInProgress = false;
}

References ExportInProgress, fb(), and SavedResourceOwnerDuringExport.

Referenced by AbortTransaction().

SnapBuildRestore()

static bool SnapBuildRestore ( SnapBuild *  builder, XLogRecPtr  lsn  )

static

Definition at line 1845 of file snapbuild.c.

{
    SnapBuildOnDisk ondisk;
 
    /* no point in loading a snapshot if we're already there */
    if (builder->state == SNAPBUILD_CONSISTENT)
        return false;
 
    /* validate and restore the snapshot to 'ondisk' */
    if (!SnapBuildRestoreSnapshot(&ondisk, lsn, builder->context, true))
        return false;
 
    /*
     * ok, we now have a sensible snapshot here, figure out if it has more
     * information than we have.
     */
 
    /*
     * We are only interested in consistent snapshots for now, comparing
     * whether one incomplete snapshot is more "advanced" seems to be
     * unnecessarily complex.
     */
    if (ondisk.builder.state < SNAPBUILD_CONSISTENT)
        goto snapshot_not_interesting;
 
    /*
     * Don't use a snapshot that requires an xmin that we cannot guarantee to
     * be available.
     */
    if (TransactionIdPrecedes(ondisk.builder.xmin, builder->initial_xmin_horizon))
        goto snapshot_not_interesting;
 
    /*
     * Consistent snapshots have no next phase. Reset next_phase_at as it is
     * possible that an old value may remain.
     */
    Assert(ondisk.builder.next_phase_at == InvalidTransactionId);
    builder->next_phase_at = InvalidTransactionId;
 
    /* ok, we think the snapshot is sensible, copy over everything important */
    builder->xmin = ondisk.builder.xmin;
    builder->xmax = ondisk.builder.xmax;
    builder->state = ondisk.builder.state;
 
    builder->committed.xcnt = ondisk.builder.committed.xcnt;
    /* We only allocated/stored xcnt, not xcnt_space xids ! */
    /* don't overwrite preallocated xip, if we don't have anything here */
    if (builder->committed.xcnt > 0)
    {
        pfree(builder->committed.xip);
        builder->committed.xcnt_space = ondisk.builder.committed.xcnt;
        builder->committed.xip = ondisk.builder.committed.xip;
    }
    ondisk.builder.committed.xip = NULL;
 
    /* set catalog modifying transactions */
    if (builder->catchange.xip)
        pfree(builder->catchange.xip);
    builder->catchange.xcnt = ondisk.builder.catchange.xcnt;
    builder->catchange.xip = ondisk.builder.catchange.xip;
    ondisk.builder.catchange.xip = NULL;
 
    /* our snapshot is not interesting anymore, build a new one */
    if (builder->snapshot != NULL)
    {
        SnapBuildSnapDecRefcount(builder->snapshot);
    }
    builder->snapshot = SnapBuildBuildSnapshot(builder);
    SnapBuildSnapIncRefcount(builder->snapshot);
 
    ReorderBufferSetRestartPoint(builder->reorder, lsn);
 
    Assert(builder->state == SNAPBUILD_CONSISTENT);
 
    ereport(LogicalDecodingLogLevel(),
            errmsg("logical decoding found consistent point at %X/%08X",
                   LSN_FORMAT_ARGS(lsn)),
            errdetail("Logical decoding will begin using saved snapshot."));
    return true;
 
snapshot_not_interesting:
    if (ondisk.builder.committed.xip != NULL)
        pfree(ondisk.builder.committed.xip);
    if (ondisk.builder.catchange.xip != NULL)
        pfree(ondisk.builder.catchange.xip);
    return false;
}

References Assert, SnapBuildOnDisk::builder, SnapBuild::catchange, SnapBuild::committed, SnapBuild::context, ereport, errdetail(), errmsg, fb(), SnapBuild::initial_xmin_horizon, InvalidTransactionId, LogicalDecodingLogLevel, LSN_FORMAT_ARGS, SnapBuild::next_phase_at, pfree(), SnapBuild::reorder, ReorderBufferSetRestartPoint(), SNAPBUILD_CONSISTENT, SnapBuildBuildSnapshot(), SnapBuildRestoreSnapshot(), SnapBuildSnapDecRefcount(), SnapBuildSnapIncRefcount(), SnapBuild::snapshot, SnapBuild::state, TransactionIdPrecedes(), SnapBuild::xcnt, SnapBuild::xcnt_space, SnapBuild::xip, SnapBuild::xmax, and SnapBuild::xmin.

Referenced by SnapBuildFindSnapshot(), and SnapBuildSerializationPoint().

SnapBuildRestoreContents()

static void SnapBuildRestoreContents ( int  fd, void *  dest, Size  size, const char *  path  )

static

Definition at line 1937 of file snapbuild.c.

{
    int         readBytes;
 
    pgstat_report_wait_start(WAIT_EVENT_SNAPBUILD_READ);
    readBytes = read(fd, dest, size);
    pgstat_report_wait_end();
    if (readBytes != size)
    {
        int         save_errno = errno;
 
        CloseTransientFile(fd);
 
        if (readBytes < 0)
        {
            errno = save_errno;
            ereport(ERROR,
                    (errcode_for_file_access(),
                     errmsg("could not read file \"%s\": %m", path)));
        }
        else
            ereport(ERROR,
                    (errcode(ERRCODE_DATA_CORRUPTED),
                     errmsg("could not read file \"%s\": read %d of %zu",
                            path, readBytes, size)));
    }
}

References CloseTransientFile(), ereport, errcode(), ERRCODE_DATA_CORRUPTED, errcode_for_file_access(), errmsg, ERROR, fb(), fd(), pgstat_report_wait_end(), pgstat_report_wait_start(), and read.

Referenced by SnapBuildRestoreSnapshot().

SnapBuildRestoreSnapshot()

bool SnapBuildRestoreSnapshot	(	SnapBuildOnDisk *	ondisk,
		XLogRecPtr	lsn,
		MemoryContext	context,
		bool	missing_ok
	)

Definition at line 1746 of file snapbuild.c.

{
    int         fd;
    pg_crc32c   checksum;
    Size        sz;
    char        path[MAXPGPATH];
 
    sprintf(path, "%s/%X-%X.snap",
            PG_LOGICAL_SNAPSHOTS_DIR,
            LSN_FORMAT_ARGS(lsn));
 
    fd = OpenTransientFile(path, O_RDONLY | PG_BINARY);
 
    if (fd < 0)
    {
        if (missing_ok && errno == ENOENT)
            return false;
 
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not open file \"%s\": %m", path)));
    }
 
    /* ----
     * Make sure the snapshot had been stored safely to disk, that's normally
     * cheap.
     * Note that we do not need PANIC here, nobody will be able to use the
     * slot without fsyncing, and saving it won't succeed without an fsync()
     * either...
     * ----
     */
    fsync_fname(path, false);
    fsync_fname(PG_LOGICAL_SNAPSHOTS_DIR, true);
 
    /* read statically sized portion of snapshot */
    SnapBuildRestoreContents(fd, ondisk, SnapBuildOnDiskConstantSize, path);
 
    if (ondisk->magic != SNAPBUILD_MAGIC)
        ereport(ERROR,
                (errcode(ERRCODE_DATA_CORRUPTED),
                 errmsg("snapbuild state file \"%s\" has wrong magic number: %u instead of %u",
                        path, ondisk->magic, SNAPBUILD_MAGIC)));
 
    if (ondisk->version != SNAPBUILD_VERSION)
        ereport(ERROR,
                (errcode(ERRCODE_DATA_CORRUPTED),
                 errmsg("snapbuild state file \"%s\" has unsupported version: %u instead of %u",
                        path, ondisk->version, SNAPBUILD_VERSION)));
 
    INIT_CRC32C(checksum);
    COMP_CRC32C(checksum,
                ((char *) ondisk) + SnapBuildOnDiskNotChecksummedSize,
                SnapBuildOnDiskConstantSize - SnapBuildOnDiskNotChecksummedSize);
 
    /* read SnapBuild */
    SnapBuildRestoreContents(fd, &ondisk->builder, sizeof(SnapBuild), path);
    COMP_CRC32C(checksum, &ondisk->builder, sizeof(SnapBuild));
 
    /* restore committed xacts information */
    if (ondisk->builder.committed.xcnt > 0)
    {
        sz = sizeof(TransactionId) * ondisk->builder.committed.xcnt;
        ondisk->builder.committed.xip = MemoryContextAllocZero(context, sz);
        SnapBuildRestoreContents(fd, ondisk->builder.committed.xip, sz, path);
        COMP_CRC32C(checksum, ondisk->builder.committed.xip, sz);
    }
 
    /* restore catalog modifying xacts information */
    if (ondisk->builder.catchange.xcnt > 0)
    {
        sz = sizeof(TransactionId) * ondisk->builder.catchange.xcnt;
        ondisk->builder.catchange.xip = MemoryContextAllocZero(context, sz);
        SnapBuildRestoreContents(fd, ondisk->builder.catchange.xip, sz, path);
        COMP_CRC32C(checksum, ondisk->builder.catchange.xip, sz);
    }
 
    if (CloseTransientFile(fd) != 0)
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not close file \"%s\": %m", path)));
 
    FIN_CRC32C(checksum);
 
    /* verify checksum of what we've read */
    if (!EQ_CRC32C(checksum, ondisk->checksum))
        ereport(ERROR,
                (errcode(ERRCODE_DATA_CORRUPTED),
                 errmsg("checksum mismatch for snapbuild state file \"%s\": is %u, should be %u",
                        path, checksum, ondisk->checksum)));
 
    return true;
}

References SnapBuildOnDisk::builder, SnapBuild::catchange, SnapBuildOnDisk::checksum, CloseTransientFile(), SnapBuild::committed, COMP_CRC32C, EQ_CRC32C, ereport, errcode(), ERRCODE_DATA_CORRUPTED, errcode_for_file_access(), errmsg, ERROR, fb(), fd(), FIN_CRC32C, fsync_fname(), INIT_CRC32C, LSN_FORMAT_ARGS, SnapBuildOnDisk::magic, MAXPGPATH, MemoryContextAllocZero(), OpenTransientFile(), PG_BINARY, PG_LOGICAL_SNAPSHOTS_DIR, SNAPBUILD_MAGIC, SNAPBUILD_VERSION, SnapBuildOnDiskConstantSize, SnapBuildOnDiskNotChecksummedSize, SnapBuildRestoreContents(), sprintf, SnapBuildOnDisk::version, SnapBuild::xcnt, and SnapBuild::xip.

Referenced by pg_get_logical_snapshot_info(), pg_get_logical_snapshot_meta(), and SnapBuildRestore().

SnapBuildSerializationPoint()

void SnapBuildSerializationPoint	(	SnapBuild *	builder,
		XLogRecPtr	lsn
	)

Definition at line 1488 of file snapbuild.c.

{
    if (builder->state < SNAPBUILD_CONSISTENT)
        SnapBuildRestore(builder, lsn);
    else
        SnapBuildSerialize(builder, lsn);
}

References SNAPBUILD_CONSISTENT, SnapBuildRestore(), SnapBuildSerialize(), and SnapBuild::state.

Referenced by xlog_decode().

SnapBuildSerialize()

static void SnapBuildSerialize ( SnapBuild *  builder, XLogRecPtr  lsn  )

static

Definition at line 1501 of file snapbuild.c.

{
    Size        needed_length;
    SnapBuildOnDisk *ondisk = NULL;
    TransactionId *catchange_xip = NULL;
    MemoryContext old_ctx;
    size_t      catchange_xcnt;
    char       *ondisk_c;
    int         fd;
    char        tmppath[MAXPGPATH];
    char        path[MAXPGPATH];
    int         ret;
    struct stat stat_buf;
    Size        sz;
 
    Assert(XLogRecPtrIsValid(lsn));
    Assert(!XLogRecPtrIsValid(builder->last_serialized_snapshot) ||
           builder->last_serialized_snapshot <= lsn);
 
    /*
     * no point in serializing if we cannot continue to work immediately after
     * restoring the snapshot
     */
    if (builder->state < SNAPBUILD_CONSISTENT)
        return;
 
    /* consistent snapshots have no next phase */
    Assert(builder->next_phase_at == InvalidTransactionId);
 
    /*
     * We identify snapshots by the LSN they are valid for. We don't need to
     * include timelines in the name as each LSN maps to exactly one timeline
     * unless the user used pg_resetwal or similar. If a user did so, there's
     * no hope continuing to decode anyway.
     */
    sprintf(path, "%s/%X-%X.snap",
            PG_LOGICAL_SNAPSHOTS_DIR,
            LSN_FORMAT_ARGS(lsn));
 
    /*
     * first check whether some other backend already has written the snapshot
     * for this LSN. It's perfectly fine if there's none, so we accept ENOENT
     * as a valid state. Everything else is an unexpected error.
     */
    ret = stat(path, &stat_buf);
 
    if (ret != 0 && errno != ENOENT)
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not stat file \"%s\": %m", path)));
 
    else if (ret == 0)
    {
        /*
         * somebody else has already serialized to this point, don't overwrite
         * but remember location, so we don't need to read old data again.
         *
         * To be sure it has been synced to disk after the rename() from the
         * tempfile filename to the real filename, we just repeat the fsync.
         * That ought to be cheap because in most scenarios it should already
         * be safely on disk.
         */
        fsync_fname(path, false);
        fsync_fname(PG_LOGICAL_SNAPSHOTS_DIR, true);
 
        builder->last_serialized_snapshot = lsn;
        goto out;
    }
 
    /*
     * there is an obvious race condition here between the time we stat(2) the
     * file and us writing the file. But we rename the file into place
     * atomically and all files created need to contain the same data anyway,
     * so this is perfectly fine, although a bit of a resource waste. Locking
     * seems like pointless complication.
     */
    elog(DEBUG1, "serializing snapshot to %s", path);
 
    /* to make sure only we will write to this tempfile, include pid */
    sprintf(tmppath, "%s/%X-%X.snap.%d.tmp",
            PG_LOGICAL_SNAPSHOTS_DIR,
            LSN_FORMAT_ARGS(lsn), MyProcPid);
 
    /*
     * Unlink temporary file if it already exists, needs to have been before a
     * crash/error since we won't enter this function twice from within a
     * single decoding slot/backend and the temporary file contains the pid of
     * the current process.
     */
    if (unlink(tmppath) != 0 && errno != ENOENT)
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not remove file \"%s\": %m", tmppath)));
 
    old_ctx = MemoryContextSwitchTo(builder->context);
 
    /* Get the catalog modifying transactions that are yet not committed */
    catchange_xip = ReorderBufferGetCatalogChangesXacts(builder->reorder);
    catchange_xcnt = dclist_count(&builder->reorder->catchange_txns);
 
    needed_length = sizeof(SnapBuildOnDisk) +
        sizeof(TransactionId) * (builder->committed.xcnt + catchange_xcnt);
 
    ondisk_c = palloc0(needed_length);
    ondisk = (SnapBuildOnDisk *) ondisk_c;
    ondisk->magic = SNAPBUILD_MAGIC;
    ondisk->version = SNAPBUILD_VERSION;
    ondisk->length = needed_length;
    INIT_CRC32C(ondisk->checksum);
    COMP_CRC32C(ondisk->checksum,
                ((char *) ondisk) + SnapBuildOnDiskNotChecksummedSize,
                SnapBuildOnDiskConstantSize - SnapBuildOnDiskNotChecksummedSize);
    ondisk_c += sizeof(SnapBuildOnDisk);
 
    memcpy(&ondisk->builder, builder, sizeof(SnapBuild));
    /* NULL-ify memory-only data */
    ondisk->builder.context = NULL;
    ondisk->builder.snapshot = NULL;
    ondisk->builder.reorder = NULL;
    ondisk->builder.committed.xip = NULL;
    ondisk->builder.catchange.xip = NULL;
    /* update catchange only on disk data */
    ondisk->builder.catchange.xcnt = catchange_xcnt;
 
    COMP_CRC32C(ondisk->checksum,
                &ondisk->builder,
                sizeof(SnapBuild));
 
    /* copy committed xacts */
    if (builder->committed.xcnt > 0)
    {
        sz = sizeof(TransactionId) * builder->committed.xcnt;
        memcpy(ondisk_c, builder->committed.xip, sz);
        COMP_CRC32C(ondisk->checksum, ondisk_c, sz);
        ondisk_c += sz;
    }
 
    /* copy catalog modifying xacts */
    if (catchange_xcnt > 0)
    {
        sz = sizeof(TransactionId) * catchange_xcnt;
        memcpy(ondisk_c, catchange_xip, sz);
        COMP_CRC32C(ondisk->checksum, ondisk_c, sz);
        ondisk_c += sz;
    }
 
    FIN_CRC32C(ondisk->checksum);
 
    /* we have valid data now, open tempfile and write it there */
    fd = OpenTransientFile(tmppath,
                           O_CREAT | O_EXCL | O_WRONLY | PG_BINARY);
    if (fd < 0)
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not open file \"%s\": %m", tmppath)));
 
    errno = 0;
    pgstat_report_wait_start(WAIT_EVENT_SNAPBUILD_WRITE);
    if ((write(fd, ondisk, needed_length)) != needed_length)
    {
        int         save_errno = errno;
 
        CloseTransientFile(fd);
 
        /* if write didn't set errno, assume problem is no disk space */
        errno = save_errno ? save_errno : ENOSPC;
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not write to file \"%s\": %m", tmppath)));
    }
    pgstat_report_wait_end();
 
    /*
     * fsync the file before renaming so that even if we crash after this we
     * have either a fully valid file or nothing.
     *
     * It's safe to just ERROR on fsync() here because we'll retry the whole
     * operation including the writes.
     *
     * TODO: Do the fsync() via checkpoints/restartpoints, doing it here has
     * some noticeable overhead since it's performed synchronously during
     * decoding?
     */
    pgstat_report_wait_start(WAIT_EVENT_SNAPBUILD_SYNC);
    if (pg_fsync(fd) != 0)
    {
        int         save_errno = errno;
 
        CloseTransientFile(fd);
        errno = save_errno;
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not fsync file \"%s\": %m", tmppath)));
    }
    pgstat_report_wait_end();
 
    if (CloseTransientFile(fd) != 0)
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not close file \"%s\": %m", tmppath)));
 
    fsync_fname(PG_LOGICAL_SNAPSHOTS_DIR, true);
 
    /*
     * We may overwrite the work from some other backend, but that's ok, our
     * snapshot is valid as well, we'll just have done some superfluous work.
     */
    if (rename(tmppath, path) != 0)
    {
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not rename file \"%s\" to \"%s\": %m",
                        tmppath, path)));
    }
 
    /* make sure we persist */
    fsync_fname(path, false);
    fsync_fname(PG_LOGICAL_SNAPSHOTS_DIR, true);
 
    /*
     * Now there's no way we can lose the dumped state anymore, remember this
     * as a serialization point.
     */
    builder->last_serialized_snapshot = lsn;
 
    MemoryContextSwitchTo(old_ctx);
 
out:
    ReorderBufferSetRestartPoint(builder->reorder,
                                 builder->last_serialized_snapshot);
    /* be tidy */
    if (ondisk)
        pfree(ondisk);
    if (catchange_xip)
        pfree(catchange_xip);
}

References Assert, SnapBuildOnDisk::builder, SnapBuild::catchange, ReorderBuffer::catchange_txns, SnapBuildOnDisk::checksum, CloseTransientFile(), SnapBuild::committed, COMP_CRC32C, SnapBuild::context, dclist_count(), DEBUG1, elog, ereport, errcode_for_file_access(), errmsg, ERROR, fb(), fd(), FIN_CRC32C, fsync_fname(), INIT_CRC32C, InvalidTransactionId, SnapBuild::last_serialized_snapshot, SnapBuildOnDisk::length, LSN_FORMAT_ARGS, SnapBuildOnDisk::magic, MAXPGPATH, memcpy(), MemoryContextSwitchTo(), MyProcPid, SnapBuild::next_phase_at, OpenTransientFile(), palloc0(), pfree(), PG_BINARY, pg_fsync(), PG_LOGICAL_SNAPSHOTS_DIR, pgstat_report_wait_end(), pgstat_report_wait_start(), SnapBuild::reorder, ReorderBufferGetCatalogChangesXacts(), ReorderBufferSetRestartPoint(), SNAPBUILD_CONSISTENT, SNAPBUILD_MAGIC, SNAPBUILD_VERSION, SnapBuildOnDiskConstantSize, SnapBuildOnDiskNotChecksummedSize, SnapBuild::snapshot, sprintf, stat, SnapBuild::state, SnapBuildOnDisk::version, write, SnapBuild::xcnt, SnapBuild::xip, and XLogRecPtrIsValid.

Referenced by SnapBuildProcessRunningXacts(), and SnapBuildSerializationPoint().

SnapBuildSetTwoPhaseAt()

void SnapBuildSetTwoPhaseAt	(	SnapBuild *	builder,
		XLogRecPtr	ptr
	)

Definition at line 299 of file snapbuild.c.

{
    builder->two_phase_at = ptr;
}

References SnapBuild::two_phase_at.

Referenced by CreateDecodingContext().

SnapBuildSnapDecRefcount()

void SnapBuildSnapDecRefcount

(

Snapshot

snap

)

Definition at line 332 of file snapbuild.c.

{
    /* make sure we don't get passed an external snapshot */
    Assert(snap->snapshot_type == SNAPSHOT_HISTORIC_MVCC);
 
    /* make sure nobody modified our snapshot */
    Assert(snap->curcid == FirstCommandId);
    Assert(!snap->suboverflowed);
    Assert(!snap->takenDuringRecovery);
 
    Assert(snap->regd_count == 0);
 
    Assert(snap->active_count > 0);
 
    /* slightly more likely, so it's checked even without casserts */
    if (snap->copied)
        elog(ERROR, "cannot free a copied snapshot");
 
    snap->active_count--;
    if (snap->active_count == 0)
        SnapBuildFreeSnapshot(snap);
}

References Assert, elog, ERROR, fb(), FirstCommandId, SnapBuildFreeSnapshot(), and SNAPSHOT_HISTORIC_MVCC.

Referenced by FreeSnapshotBuilder(), ReorderBufferCleanupTXN(), ReorderBufferFreeSnap(), ReorderBufferTransferSnapToParent(), SnapBuildCommitTxn(), and SnapBuildRestore().

SnapBuildSnapIncRefcount()

static void SnapBuildSnapIncRefcount ( Snapshot  snap )

static

Definition at line 320 of file snapbuild.c.

{
    snap->active_count++;
}

References fb().

Referenced by SnapBuildCommitTxn(), SnapBuildDistributeSnapshotAndInval(), SnapBuildGetOrBuildSnapshot(), SnapBuildProcessChange(), and SnapBuildRestore().

SnapBuildSnapshotExists()

bool SnapBuildSnapshotExists

(

XLogRecPtr

lsn

)

Definition at line 2062 of file snapbuild.c.

{
    char        path[MAXPGPATH];
    int         ret;
    struct stat stat_buf;
 
    sprintf(path, "%s/%X-%X.snap",
            PG_LOGICAL_SNAPSHOTS_DIR,
            LSN_FORMAT_ARGS(lsn));
 
    ret = stat(path, &stat_buf);
 
    if (ret != 0 && errno != ENOENT)
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not stat file \"%s\": %m", path)));
 
    return ret == 0;
}

References ereport, errcode_for_file_access(), errmsg, ERROR, fb(), LSN_FORMAT_ARGS, MAXPGPATH, PG_LOGICAL_SNAPSHOTS_DIR, sprintf, and stat.

Referenced by update_local_synced_slot().

SnapBuildWaitSnapshot()

static void SnapBuildWaitSnapshot ( xl_running_xacts *  running, TransactionId  cutoff  )

static

Definition at line 1439 of file snapbuild.c.

{
    int         off;
 
    for (off = 0; off < running->xcnt; off++)
    {
        TransactionId xid = running->xids[off];
 
        /*
         * Upper layers should prevent that we ever need to wait on ourselves.
         * Check anyway, since failing to do so would either result in an
         * endless wait or an Assert() failure.
         */
        if (TransactionIdIsCurrentTransactionId(xid))
            elog(ERROR, "waiting for ourselves");
 
        if (TransactionIdFollows(xid, cutoff))
            continue;
 
        XactLockTableWait(xid, NULL, NULL, XLTW_None);
    }
 
    /*
     * All transactions we needed to finish finished - try to ensure there is
     * another xl_running_xacts record in a timely manner, without having to
     * wait for bgwriter or checkpointer to log one.  During recovery we can't
     * enforce that, so we'll have to wait.
     */
    if (!RecoveryInProgress())
    {
        LogStandbySnapshot();
    }
}

References elog, ERROR, fb(), LogStandbySnapshot(), RecoveryInProgress(), TransactionIdFollows(), TransactionIdIsCurrentTransactionId(), XactLockTableWait(), xl_running_xacts::xcnt, xl_running_xacts::xids, and XLTW_None.

Referenced by SnapBuildFindSnapshot().

SnapBuildXactNeedsSkip()

bool SnapBuildXactNeedsSkip	(	SnapBuild *	builder,
		XLogRecPtr	ptr
	)

Definition at line 308 of file snapbuild.c.

{
    return ptr < builder->start_decoding_at;
}

References SnapBuild::start_decoding_at.

Referenced by AssertTXNLsnOrder(), DecodeTXNNeedSkip(), logicalmsg_decode(), and ReorderBufferCanStartStreaming().

SnapBuildXidHasCatalogChanges()

static bool SnapBuildXidHasCatalogChanges ( SnapBuild *  builder, TransactionId  xid, uint32  xinfo  )

inlinestatic

Definition at line 1110 of file snapbuild.c.

{
    if (ReorderBufferXidHasCatalogChanges(builder->reorder, xid))
        return true;
 
    /*
     * The transactions that have changed catalogs must have invalidation
     * info.
     */
    if (!(xinfo & XACT_XINFO_HAS_INVALS))
        return false;
 
    /* Check the catchange XID array */
    return ((builder->catchange.xcnt > 0) &&
            (bsearch(&xid, builder->catchange.xip, builder->catchange.xcnt,
                     sizeof(TransactionId), xidComparator) != NULL));
}

References SnapBuild::catchange, fb(), SnapBuild::reorder, ReorderBufferXidHasCatalogChanges(), XACT_XINFO_HAS_INVALS, SnapBuild::xcnt, xidComparator(), and SnapBuild::xip.

Referenced by SnapBuildCommitTxn().

Variable Documentation

ExportInProgress

bool ExportInProgress = false

static

Definition at line 155 of file snapbuild.c.

Referenced by SnapBuildClearExportedSnapshot(), SnapBuildExportSnapshot(), and SnapBuildResetExportedSnapshotState().

SavedResourceOwnerDuringExport

ResourceOwner SavedResourceOwnerDuringExport = NULL

static

Definition at line 154 of file snapbuild.c.

Referenced by SnapBuildClearExportedSnapshot(), SnapBuildExportSnapshot(), and SnapBuildResetExportedSnapshotState().