origin.h File Reference

#include "access/xlog.h"
#include "access/xlogdefs.h"
#include "access/xlogreader.h"
#include "catalog/pg_replication_origin.h"

Include dependency graph for origin.h:

This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Data Structures
struct	xl_replorigin_set
struct	xl_replorigin_drop
struct	ReplOriginXactState

Macros
#define	XLOG_REPLORIGIN_SET   0x00
#define	XLOG_REPLORIGIN_DROP   0x10
#define	InvalidReplOriginId   0
#define	DoNotReplicateId   PG_UINT16_MAX
#define	MAX_RONAME_LEN   512

Typedefs
typedef struct xl_replorigin_set	xl_replorigin_set
typedef struct xl_replorigin_drop	xl_replorigin_drop
typedef struct ReplOriginXactState	ReplOriginXactState

Functions
ReplOriginId	replorigin_by_name (const char *roname, bool missing_ok)
ReplOriginId	replorigin_create (const char *roname)
void	replorigin_drop_by_name (const char *name, bool missing_ok, bool nowait)
bool	replorigin_by_oid (ReplOriginId roident, bool missing_ok, char **roname)
void	replorigin_advance (ReplOriginId node, XLogRecPtr remote_commit, XLogRecPtr local_commit, bool go_backward, bool wal_log)
XLogRecPtr	replorigin_get_progress (ReplOriginId node, bool flush)
void	replorigin_session_advance (XLogRecPtr remote_commit, XLogRecPtr local_commit)
void	replorigin_session_setup (ReplOriginId node, int acquired_by)
void	replorigin_session_reset (void)
XLogRecPtr	replorigin_session_get_progress (bool flush)
void	replorigin_xact_clear (bool clear_origin)
void	CheckPointReplicationOrigin (void)
void	StartupReplicationOrigin (void)
void	replorigin_redo (XLogReaderState *record)
void	replorigin_desc (StringInfo buf, XLogReaderState *record)
const char *	replorigin_identify (uint8 info)

Variables
PGDLLIMPORT ReplOriginXactState	replorigin_xact_state
PGDLLIMPORT int	max_active_replication_origins

Macro Definition Documentation

DoNotReplicateId

#define DoNotReplicateId   PG_UINT16_MAX

Definition at line 34 of file origin.h.

InvalidReplOriginId

#define InvalidReplOriginId   0

Definition at line 33 of file origin.h.

MAX_RONAME_LEN

#define MAX_RONAME_LEN   512

Definition at line 41 of file origin.h.

XLOG_REPLORIGIN_DROP

#define XLOG_REPLORIGIN_DROP   0x10

Definition at line 31 of file origin.h.

XLOG_REPLORIGIN_SET

#define XLOG_REPLORIGIN_SET   0x00

Definition at line 30 of file origin.h.

Typedef Documentation

ReplOriginXactState

typedef struct ReplOriginXactState ReplOriginXactState

xl_replorigin_drop

typedef struct xl_replorigin_drop xl_replorigin_drop

xl_replorigin_set

typedef struct xl_replorigin_set xl_replorigin_set

Function Documentation

CheckPointReplicationOrigin()

void CheckPointReplicationOrigin ( void  )

extern

Definition at line 614 of file origin.c.

{
    const char *tmppath = PG_REPLORIGIN_CHECKPOINT_TMPFILE;
    const char *path = PG_REPLORIGIN_CHECKPOINT_FILENAME;
    int         tmpfd;
    int         i;
    uint32      magic = REPLICATION_STATE_MAGIC;
    pg_crc32c   crc;
 
    if (max_active_replication_origins == 0)
        return;
 
    INIT_CRC32C(crc);
 
    /* make sure no old temp file is remaining */
    if (unlink(tmppath) < 0 && errno != ENOENT)
        ereport(PANIC,
                (errcode_for_file_access(),
                 errmsg("could not remove file \"%s\": %m",
                        tmppath)));
 
    /*
     * no other backend can perform this at the same time; only one checkpoint
     * can happen at a time.
     */
    tmpfd = OpenTransientFile(tmppath,
                              O_CREAT | O_EXCL | O_WRONLY | PG_BINARY);
    if (tmpfd < 0)
        ereport(PANIC,
                (errcode_for_file_access(),
                 errmsg("could not create file \"%s\": %m",
                        tmppath)));
 
    /* write magic */
    errno = 0;
    if ((write(tmpfd, &magic, sizeof(magic))) != sizeof(magic))
    {
        /* if write didn't set errno, assume problem is no disk space */
        if (errno == 0)
            errno = ENOSPC;
        ereport(PANIC,
                (errcode_for_file_access(),
                 errmsg("could not write to file \"%s\": %m",
                        tmppath)));
    }
    COMP_CRC32C(crc, &magic, sizeof(magic));
 
    /* prevent concurrent creations/drops */
    LWLockAcquire(ReplicationOriginLock, LW_SHARED);
 
    /* write actual data */
    for (i = 0; i < max_active_replication_origins; i++)
    {
        ReplicationStateOnDisk disk_state;
        ReplicationState *curstate = &replication_states[i];
        XLogRecPtr  local_lsn;
 
        if (curstate->roident == InvalidReplOriginId)
            continue;
 
        /* zero, to avoid uninitialized padding bytes */
        memset(&disk_state, 0, sizeof(disk_state));
 
        LWLockAcquire(&curstate->lock, LW_SHARED);
 
        disk_state.roident = curstate->roident;
 
        disk_state.remote_lsn = curstate->remote_lsn;
        local_lsn = curstate->local_lsn;
 
        LWLockRelease(&curstate->lock);
 
        /* make sure we only write out a commit that's persistent */
        XLogFlush(local_lsn);
 
        errno = 0;
        if ((write(tmpfd, &disk_state, sizeof(disk_state))) !=
            sizeof(disk_state))
        {
            /* if write didn't set errno, assume problem is no disk space */
            if (errno == 0)
                errno = ENOSPC;
            ereport(PANIC,
                    (errcode_for_file_access(),
                     errmsg("could not write to file \"%s\": %m",
                            tmppath)));
        }
 
        COMP_CRC32C(crc, &disk_state, sizeof(disk_state));
    }
 
    LWLockRelease(ReplicationOriginLock);
 
    /* write out the CRC */
    FIN_CRC32C(crc);
    errno = 0;
    if ((write(tmpfd, &crc, sizeof(crc))) != sizeof(crc))
    {
        /* if write didn't set errno, assume problem is no disk space */
        if (errno == 0)
            errno = ENOSPC;
        ereport(PANIC,
                (errcode_for_file_access(),
                 errmsg("could not write to file \"%s\": %m",
                        tmppath)));
    }
 
    if (CloseTransientFile(tmpfd) != 0)
        ereport(PANIC,
                (errcode_for_file_access(),
                 errmsg("could not close file \"%s\": %m",
                        tmppath)));
 
    /* fsync, rename to permanent file, fsync file and directory */
    durable_rename(tmppath, path, PANIC);
}

References CloseTransientFile(), COMP_CRC32C, crc, durable_rename(), ereport, errcode_for_file_access(), errmsg, fb(), FIN_CRC32C, i, INIT_CRC32C, InvalidReplOriginId, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_active_replication_origins, OpenTransientFile(), PANIC, PG_BINARY, PG_REPLORIGIN_CHECKPOINT_FILENAME, PG_REPLORIGIN_CHECKPOINT_TMPFILE, REPLICATION_STATE_MAGIC, replication_states, write, and XLogFlush().

Referenced by CheckPointGuts().

replorigin_advance()

void replorigin_advance ( ReplOriginId  node, XLogRecPtr  remote_commit, XLogRecPtr  local_commit, bool  go_backward, bool  wal_log  )

extern

Definition at line 928 of file origin.c.

{
    int         i;
    ReplicationState *replication_state = NULL;
    ReplicationState *free_state = NULL;
 
    Assert(node != InvalidReplOriginId);
 
    /* we don't track DoNotReplicateId */
    if (node == DoNotReplicateId)
        return;
 
    /*
     * XXX: For the case where this is called by WAL replay, it'd be more
     * efficient to restore into a backend local hashtable and only dump into
     * shmem after recovery is finished. Let's wait with implementing that
     * till it's shown to be a measurable expense
     */
 
    /* Lock exclusively, as we may have to create a new table entry. */
    LWLockAcquire(ReplicationOriginLock, LW_EXCLUSIVE);
 
    /*
     * Search for either an existing slot for the origin, or a free one we can
     * use.
     */
    for (i = 0; i < max_active_replication_origins; i++)
    {
        ReplicationState *curstate = &replication_states[i];
 
        /* remember where to insert if necessary */
        if (curstate->roident == InvalidReplOriginId &&
            free_state == NULL)
        {
            free_state = curstate;
            continue;
        }
 
        /* not our slot */
        if (curstate->roident != node)
        {
            continue;
        }
 
        /* ok, found slot */
        replication_state = curstate;
 
        LWLockAcquire(&replication_state->lock, LW_EXCLUSIVE);
 
        /* Make sure it's not used by somebody else */
        if (replication_state->refcount > 0)
        {
            ereport(ERROR,
                    (errcode(ERRCODE_OBJECT_IN_USE),
                     (replication_state->acquired_by != 0)
                     ? errmsg("replication origin with ID %d is already active for PID %d",
                              replication_state->roident,
                              replication_state->acquired_by)
                     : errmsg("replication origin with ID %d is already active in another process",
                              replication_state->roident)));
        }
 
        break;
    }
 
    if (replication_state == NULL && free_state == NULL)
        ereport(ERROR,
                (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
                 errmsg("could not find free replication state slot for replication origin with ID %d",
                        node),
                 errhint("Increase \"max_active_replication_origins\" and try again.")));
 
    if (replication_state == NULL)
    {
        /* initialize new slot */
        LWLockAcquire(&free_state->lock, LW_EXCLUSIVE);
        replication_state = free_state;
        Assert(!XLogRecPtrIsValid(replication_state->remote_lsn));
        Assert(!XLogRecPtrIsValid(replication_state->local_lsn));
        replication_state->roident = node;
    }
 
    Assert(replication_state->roident != InvalidReplOriginId);
 
    /*
     * If somebody "forcefully" sets this slot, WAL log it, so it's durable
     * and the standby gets the message. Primarily this will be called during
     * WAL replay (of commit records) where no WAL logging is necessary.
     */
    if (wal_log)
    {
        xl_replorigin_set xlrec;
 
        xlrec.remote_lsn = remote_commit;
        xlrec.node_id = node;
        xlrec.force = go_backward;
 
        XLogBeginInsert();
        XLogRegisterData(&xlrec, sizeof(xlrec));
 
        XLogInsert(RM_REPLORIGIN_ID, XLOG_REPLORIGIN_SET);
    }
 
    /*
     * Due to - harmless - race conditions during a checkpoint we could see
     * values here that are older than the ones we already have in memory. We
     * could also see older values for prepared transactions when the prepare
     * is sent at a later point of time along with commit prepared and there
     * are other transactions commits between prepare and commit prepared. See
     * ReorderBufferFinishPrepared. Don't overwrite those.
     */
    if (go_backward || replication_state->remote_lsn < remote_commit)
        replication_state->remote_lsn = remote_commit;
    if (XLogRecPtrIsValid(local_commit) &&
        (go_backward || replication_state->local_lsn < local_commit))
        replication_state->local_lsn = local_commit;
    LWLockRelease(&replication_state->lock);
 
    /*
     * Release *after* changing the LSNs, slot isn't acquired and thus could
     * otherwise be dropped anytime.
     */
    LWLockRelease(ReplicationOriginLock);
}

References Assert, DoNotReplicateId, ereport, errcode(), errhint(), errmsg, ERROR, fb(), i, InvalidReplOriginId, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), max_active_replication_origins, xl_replorigin_set::remote_lsn, replication_states, XLOG_REPLORIGIN_SET, XLogBeginInsert(), XLogInsert(), XLogRecPtrIsValid, and XLogRegisterData().

Referenced by binary_upgrade_replorigin_advance(), LogicalRepSyncTableStart(), pg_replication_origin_advance(), PrepareRedoAdd(), replorigin_redo(), xact_redo_abort(), and xact_redo_commit().

replorigin_by_name()

ReplOriginId replorigin_by_name ( const char *  roname, bool  missing_ok  )

extern

Definition at line 243 of file origin.c.

{
    Form_pg_replication_origin ident;
    Oid         roident = InvalidOid;
    HeapTuple   tuple;
    Datum       roname_d;
 
    roname_d = CStringGetTextDatum(roname);
 
    tuple = SearchSysCache1(REPLORIGNAME, roname_d);
    if (HeapTupleIsValid(tuple))
    {
        ident = (Form_pg_replication_origin) GETSTRUCT(tuple);
        roident = ident->roident;
        ReleaseSysCache(tuple);
    }
    else if (!missing_ok)
        ereport(ERROR,
                (errcode(ERRCODE_UNDEFINED_OBJECT),
                 errmsg("replication origin \"%s\" does not exist",
                        roname)));
 
    return roident;
}

References CStringGetTextDatum, ereport, errcode(), errmsg, ERROR, fb(), Form_pg_replication_origin, GETSTRUCT(), HeapTupleIsValid, ident, InvalidOid, ReleaseSysCache(), and SearchSysCache1().

Referenced by AlterSubscription(), binary_upgrade_replorigin_advance(), LogicalRepSyncTableStart(), ParallelApplyWorkerMain(), pg_replication_origin_advance(), pg_replication_origin_oid(), pg_replication_origin_progress(), pg_replication_origin_session_setup(), replorigin_drop_by_name(), and run_apply_worker().

replorigin_by_oid()

bool replorigin_by_oid ( ReplOriginId  roident, bool  missing_ok, char **  roname  )

extern

Definition at line 513 of file origin.c.

{
    HeapTuple   tuple;
    Form_pg_replication_origin ric;
 
    Assert(OidIsValid((Oid) roident));
    Assert(roident != InvalidReplOriginId);
    Assert(roident != DoNotReplicateId);
 
    tuple = SearchSysCache1(REPLORIGIDENT,
                            ObjectIdGetDatum((Oid) roident));
 
    if (HeapTupleIsValid(tuple))
    {
        ric = (Form_pg_replication_origin) GETSTRUCT(tuple);
        *roname = text_to_cstring(&ric->roname);
        ReleaseSysCache(tuple);
 
        return true;
    }
    else
    {
        *roname = NULL;
 
        if (!missing_ok)
            ereport(ERROR,
                    (errcode(ERRCODE_UNDEFINED_OBJECT),
                     errmsg("replication origin with ID %d does not exist",
                            roident)));
 
        return false;
    }
}

References Assert, DoNotReplicateId, ereport, errcode(), errmsg, ERROR, fb(), Form_pg_replication_origin, GETSTRUCT(), HeapTupleIsValid, InvalidReplOriginId, ObjectIdGetDatum(), OidIsValid, ReleaseSysCache(), SearchSysCache1(), and text_to_cstring().

Referenced by errdetail_apply_conflict(), pg_show_replication_origin_status(), and send_repl_origin().

replorigin_create()

ReplOriginId replorigin_create ( const char *  roname )

extern

Definition at line 274 of file origin.c.

{
    Oid         roident;
    HeapTuple   tuple = NULL;
    Relation    rel;
    Datum       roname_d;
    SnapshotData SnapshotDirty;
    SysScanDesc scan;
    ScanKeyData key;
 
    /*
     * To avoid needing a TOAST table for pg_replication_origin, we limit
     * replication origin names to 512 bytes.  This should be more than enough
     * for all practical use.
     */
    if (strlen(roname) > MAX_RONAME_LEN)
        ereport(ERROR,
                (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
                 errmsg("replication origin name is too long"),
                 errdetail("Replication origin names must be no longer than %d bytes.",
                           MAX_RONAME_LEN)));
 
    roname_d = CStringGetTextDatum(roname);
 
    Assert(IsTransactionState());
 
    /*
     * We need the numeric replication origin to be 16bit wide, so we cannot
     * rely on the normal oid allocation. Instead we simply scan
     * pg_replication_origin for the first unused id. That's not particularly
     * efficient, but this should be a fairly infrequent operation - we can
     * easily spend a bit more code on this when it turns out it needs to be
     * faster.
     *
     * We handle concurrency by taking an exclusive lock (allowing reads!)
     * over the table for the duration of the search. Because we use a "dirty
     * snapshot" we can read rows that other in-progress sessions have
     * written, even though they would be invisible with normal snapshots. Due
     * to the exclusive lock there's no danger that new rows can appear while
     * we're checking.
     */
    InitDirtySnapshot(SnapshotDirty);
 
    rel = table_open(ReplicationOriginRelationId, ExclusiveLock);
 
    /*
     * We want to be able to access pg_replication_origin without setting up a
     * snapshot.  To make that safe, it needs to not have a TOAST table, since
     * TOASTed data cannot be fetched without a snapshot.  As of this writing,
     * its only varlena column is roname, which we limit to 512 bytes to avoid
     * needing out-of-line storage.  If you add a TOAST table to this catalog,
     * be sure to set up a snapshot everywhere it might be needed.  For more
     * information, see https://postgr.es/m/ZvMSUPOqUU-VNADN%40nathan.
     */
    Assert(!OidIsValid(rel->rd_rel->reltoastrelid));
 
    for (roident = InvalidOid + 1; roident < PG_UINT16_MAX; roident++)
    {
        bool        nulls[Natts_pg_replication_origin];
        Datum       values[Natts_pg_replication_origin];
        bool        collides;
 
        CHECK_FOR_INTERRUPTS();
 
        ScanKeyInit(&key,
                    Anum_pg_replication_origin_roident,
                    BTEqualStrategyNumber, F_OIDEQ,
                    ObjectIdGetDatum(roident));
 
        scan = systable_beginscan(rel, ReplicationOriginIdentIndex,
                                  true /* indexOK */ ,
                                  &SnapshotDirty,
                                  1, &key);
 
        collides = HeapTupleIsValid(systable_getnext(scan));
 
        systable_endscan(scan);
 
        if (!collides)
        {
            /*
             * Ok, found an unused roident, insert the new row and do a CCI,
             * so our callers can look it up if they want to.
             */
            memset(&nulls, 0, sizeof(nulls));
 
            values[Anum_pg_replication_origin_roident - 1] = ObjectIdGetDatum(roident);
            values[Anum_pg_replication_origin_roname - 1] = roname_d;
 
            tuple = heap_form_tuple(RelationGetDescr(rel), values, nulls);
            CatalogTupleInsert(rel, tuple);
            CommandCounterIncrement();
            break;
        }
    }
 
    /* now release lock again,  */
    table_close(rel, ExclusiveLock);
 
    if (tuple == NULL)
        ereport(ERROR,
                (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
                 errmsg("could not find free replication origin ID")));
 
    heap_freetuple(tuple);
    return roident;
}

References Assert, BTEqualStrategyNumber, CatalogTupleInsert(), CHECK_FOR_INTERRUPTS, CommandCounterIncrement(), CStringGetTextDatum, ereport, errcode(), errdetail(), errmsg, ERROR, ExclusiveLock, fb(), heap_form_tuple(), heap_freetuple(), HeapTupleIsValid, InitDirtySnapshot, InvalidOid, IsTransactionState(), MAX_RONAME_LEN, ObjectIdGetDatum(), OidIsValid, PG_UINT16_MAX, RelationData::rd_rel, RelationGetDescr, ScanKeyInit(), systable_beginscan(), systable_endscan(), systable_getnext(), table_close(), table_open(), and values.

Referenced by CreateSubscription(), LogicalRepSyncTableStart(), pg_replication_origin_create(), and run_apply_worker().

replorigin_desc()

void replorigin_desc ( StringInfo  buf, XLogReaderState *  record  )

extern

Definition at line 19 of file replorigindesc.c.

{
    char       *rec = XLogRecGetData(record);
    uint8       info = XLogRecGetInfo(record) & ~XLR_INFO_MASK;
 
    switch (info)
    {
        case XLOG_REPLORIGIN_SET:
            {
                xl_replorigin_set *xlrec;
 
                xlrec = (xl_replorigin_set *) rec;
 
                appendStringInfo(buf, "set %u; lsn %X/%08X; force: %d",
                                 xlrec->node_id,
                                 LSN_FORMAT_ARGS(xlrec->remote_lsn),
                                 xlrec->force);
                break;
            }
        case XLOG_REPLORIGIN_DROP:
            {
                xl_replorigin_drop *xlrec;
 
                xlrec = (xl_replorigin_drop *) rec;
 
                appendStringInfo(buf, "drop %u", xlrec->node_id);
                break;
            }
    }
}

References appendStringInfo(), buf, fb(), LSN_FORMAT_ARGS, XLOG_REPLORIGIN_DROP, XLOG_REPLORIGIN_SET, XLogRecGetData, and XLogRecGetInfo.

replorigin_drop_by_name()

void replorigin_drop_by_name ( const char *  name, bool  missing_ok, bool  nowait  )

extern

Definition at line 459 of file origin.c.

{
    ReplOriginId roident;
    Relation    rel;
    HeapTuple   tuple;
 
    Assert(IsTransactionState());
 
    rel = table_open(ReplicationOriginRelationId, RowExclusiveLock);
 
    roident = replorigin_by_name(name, missing_ok);
 
    /* Lock the origin to prevent concurrent drops. */
    LockSharedObject(ReplicationOriginRelationId, roident, 0,
                     AccessExclusiveLock);
 
    tuple = SearchSysCache1(REPLORIGIDENT, ObjectIdGetDatum(roident));
    if (!HeapTupleIsValid(tuple))
    {
        if (!missing_ok)
            elog(ERROR, "cache lookup failed for replication origin with ID %d",
                 roident);
 
        /*
         * We don't need to retain the locks if the origin is already dropped.
         */
        UnlockSharedObject(ReplicationOriginRelationId, roident, 0,
                           AccessExclusiveLock);
        table_close(rel, RowExclusiveLock);
        return;
    }
 
    replorigin_state_clear(roident, nowait);
 
    /*
     * Now, we can delete the catalog entry.
     */
    CatalogTupleDelete(rel, &tuple->t_self);
    ReleaseSysCache(tuple);
 
    CommandCounterIncrement();
 
    /* We keep the lock on pg_replication_origin until commit */
    table_close(rel, NoLock);
}

References AccessExclusiveLock, Assert, CatalogTupleDelete(), CommandCounterIncrement(), elog, ERROR, fb(), HeapTupleIsValid, IsTransactionState(), LockSharedObject(), name, NoLock, ObjectIdGetDatum(), ReleaseSysCache(), replorigin_by_name(), replorigin_state_clear(), RowExclusiveLock, SearchSysCache1(), HeapTupleData::t_self, table_close(), table_open(), and UnlockSharedObject().

Referenced by AlterSubscription_refresh(), DropSubscription(), pg_replication_origin_drop(), ProcessSyncingTablesForApply(), and ProcessSyncingTablesForSync().

replorigin_get_progress()

XLogRecPtr replorigin_get_progress ( ReplOriginId  node, bool  flush  )

extern

Definition at line 1057 of file origin.c.

{
    int         i;
    XLogRecPtr  local_lsn = InvalidXLogRecPtr;
    XLogRecPtr  remote_lsn = InvalidXLogRecPtr;
 
    /* prevent slots from being concurrently dropped */
    LWLockAcquire(ReplicationOriginLock, LW_SHARED);
 
    for (i = 0; i < max_active_replication_origins; i++)
    {
        ReplicationState *state;
 
        state = &replication_states[i];
 
        if (state->roident == node)
        {
            LWLockAcquire(&state->lock, LW_SHARED);
 
            remote_lsn = state->remote_lsn;
            local_lsn = state->local_lsn;
 
            LWLockRelease(&state->lock);
 
            break;
        }
    }
 
    LWLockRelease(ReplicationOriginLock);
 
    if (flush && XLogRecPtrIsValid(local_lsn))
        XLogFlush(local_lsn);
 
    return remote_lsn;
}

References fb(), i, InvalidXLogRecPtr, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_active_replication_origins, replication_states, XLogFlush(), and XLogRecPtrIsValid.

Referenced by AlterSubscription(), and pg_replication_origin_progress().

replorigin_identify()

const char * replorigin_identify ( uint8  info )

extern

Definition at line 51 of file replorigindesc.c.

{
    switch (info)
    {
        case XLOG_REPLORIGIN_SET:
            return "SET";
        case XLOG_REPLORIGIN_DROP:
            return "DROP";
        default:
            return NULL;
    }
}

References fb(), XLOG_REPLORIGIN_DROP, and XLOG_REPLORIGIN_SET.

replorigin_redo()

void replorigin_redo ( XLogReaderState *  record )

extern

Definition at line 867 of file origin.c.

{
    uint8       info = XLogRecGetInfo(record) & ~XLR_INFO_MASK;
 
    switch (info)
    {
        case XLOG_REPLORIGIN_SET:
            {
                xl_replorigin_set *xlrec =
                    (xl_replorigin_set *) XLogRecGetData(record);
 
                replorigin_advance(xlrec->node_id,
                                   xlrec->remote_lsn, record->EndRecPtr,
                                   xlrec->force /* backward */ ,
                                   false /* WAL log */ );
                break;
            }
        case XLOG_REPLORIGIN_DROP:
            {
                xl_replorigin_drop *xlrec;
                int         i;
 
                xlrec = (xl_replorigin_drop *) XLogRecGetData(record);
 
                for (i = 0; i < max_active_replication_origins; i++)
                {
                    ReplicationState *state = &replication_states[i];
 
                    /* found our slot */
                    if (state->roident == xlrec->node_id)
                    {
                        /* reset entry */
                        state->roident = InvalidReplOriginId;
                        state->remote_lsn = InvalidXLogRecPtr;
                        state->local_lsn = InvalidXLogRecPtr;
                        break;
                    }
                }
                break;
            }
        default:
            elog(PANIC, "replorigin_redo: unknown op code %u", info);
    }
}

References elog, XLogReaderState::EndRecPtr, fb(), i, InvalidReplOriginId, InvalidXLogRecPtr, max_active_replication_origins, PANIC, replication_states, replorigin_advance(), XLOG_REPLORIGIN_DROP, XLOG_REPLORIGIN_SET, XLogRecGetData, and XLogRecGetInfo.

replorigin_session_advance()

void replorigin_session_advance ( XLogRecPtr  remote_commit, XLogRecPtr  local_commit  )

extern

Definition at line 1335 of file origin.c.

{
    Assert(session_replication_state != NULL);
    Assert(session_replication_state->roident != InvalidReplOriginId);
 
    LWLockAcquire(&session_replication_state->lock, LW_EXCLUSIVE);
    if (session_replication_state->local_lsn < local_commit)
        session_replication_state->local_lsn = local_commit;
    if (session_replication_state->remote_lsn < remote_commit)
        session_replication_state->remote_lsn = remote_commit;
    LWLockRelease(&session_replication_state->lock);
}

References Assert, fb(), InvalidReplOriginId, ReplicationState::local_lsn, ReplicationState::lock, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), ReplicationState::remote_lsn, ReplicationState::roident, and session_replication_state.

Referenced by EndPrepare(), RecordTransactionAbort(), RecordTransactionAbortPrepared(), RecordTransactionCommit(), and RecordTransactionCommitPrepared().

replorigin_session_get_progress()

XLogRecPtr replorigin_session_get_progress ( bool  flush )

extern

Definition at line 1353 of file origin.c.

{
    XLogRecPtr  remote_lsn;
    XLogRecPtr  local_lsn;
 
    Assert(session_replication_state != NULL);
 
    LWLockAcquire(&session_replication_state->lock, LW_SHARED);
    remote_lsn = session_replication_state->remote_lsn;
    local_lsn = session_replication_state->local_lsn;
    LWLockRelease(&session_replication_state->lock);
 
    if (flush && XLogRecPtrIsValid(local_lsn))
        XLogFlush(local_lsn);
 
    return remote_lsn;
}

References Assert, fb(), ReplicationState::local_lsn, ReplicationState::lock, LW_SHARED, LWLockAcquire(), LWLockRelease(), ReplicationState::remote_lsn, session_replication_state, XLogFlush(), and XLogRecPtrIsValid.

Referenced by LogicalRepSyncTableStart(), pg_replication_origin_session_progress(), and run_apply_worker().

replorigin_session_reset()

void replorigin_session_reset ( void  )

extern

Definition at line 1301 of file origin.c.

{
    Assert(max_active_replication_origins != 0);
 
    if (session_replication_state == NULL)
        ereport(ERROR,
                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                 errmsg("no replication origin is configured")));
 
    /*
     * Restrict explicit resetting of the replication origin if it was first
     * acquired by this process and others are still using it. While the
     * system handles this safely (as happens if the first session exits
     * without calling reset), it is best to avoid doing so.
     */
    if (session_replication_state->acquired_by == MyProcPid &&
        session_replication_state->refcount > 1)
        ereport(ERROR,
                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                 errmsg("cannot reset replication origin with ID %d because it is still in use by other processes",
                        session_replication_state->roident),
                 errdetail("This session is the first process for this replication origin, and other processes are currently sharing it."),
                 errhint("Reset the replication origin in all other processes before retrying.")));
 
    replorigin_session_reset_internal();
}

References ReplicationState::acquired_by, Assert, ereport, errcode(), errdetail(), errhint(), errmsg, ERROR, fb(), max_active_replication_origins, MyProcPid, ReplicationState::refcount, replorigin_session_reset_internal(), ReplicationState::roident, and session_replication_state.

Referenced by pg_replication_origin_session_reset(), and ProcessSyncingTablesForSync().

replorigin_session_setup()

void replorigin_session_setup ( ReplOriginId  node, int  acquired_by  )

extern

Definition at line 1156 of file origin.c.

{
    static bool registered_cleanup;
    int         i;
    int         free_slot = -1;
 
    if (!registered_cleanup)
    {
        on_shmem_exit(ReplicationOriginExitCleanup, 0);
        registered_cleanup = true;
    }
 
    Assert(max_active_replication_origins > 0);
 
    if (session_replication_state != NULL)
        ereport(ERROR,
                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                 errmsg("cannot setup replication origin when one is already setup")));
 
    /* Lock exclusively, as we may have to create a new table entry. */
    LWLockAcquire(ReplicationOriginLock, LW_EXCLUSIVE);
 
    /*
     * Search for either an existing slot for the origin, or a free one we can
     * use.
     */
    for (i = 0; i < max_active_replication_origins; i++)
    {
        ReplicationState *curstate = &replication_states[i];
 
        /* remember where to insert if necessary */
        if (curstate->roident == InvalidReplOriginId &&
            free_slot == -1)
        {
            free_slot = i;
            continue;
        }
 
        /* not our slot */
        if (curstate->roident != node)
            continue;
 
        if (acquired_by == 0)
        {
            /* With acquired_by == 0, we need the origin to be free */
            if (curstate->acquired_by != 0)
            {
                ereport(ERROR,
                        (errcode(ERRCODE_OBJECT_IN_USE),
                         errmsg("replication origin with ID %d is already active for PID %d",
                                curstate->roident, curstate->acquired_by)));
            }
            else if (curstate->refcount > 0)
            {
                /*
                 * The origin is in use, but PID is not recorded. This can
                 * happen if the process that originally acquired the origin
                 * exited without releasing it. To ensure correctness, other
                 * processes cannot acquire the origin until all processes
                 * currently using it have released it.
                 */
                ereport(ERROR,
                        (errcode(ERRCODE_OBJECT_IN_USE),
                         errmsg("replication origin with ID %d is already active in another process",
                                curstate->roident)));
            }
        }
        else
        {
            /*
             * With acquired_by != 0, we need the origin to be active by the
             * given PID
             */
            if (curstate->acquired_by != acquired_by)
                ereport(ERROR,
                        (errcode(ERRCODE_OBJECT_IN_USE),
                         errmsg("replication origin with ID %d is not active for PID %d",
                                curstate->roident, acquired_by)));
 
            /*
             * Here, it is okay to have refcount > 0 as more than one process
             * can safely re-use the origin.
             */
        }
 
        /* ok, found slot */
        session_replication_state = curstate;
        break;
    }
 
    if (session_replication_state == NULL)
    {
        if (acquired_by != 0)
            ereport(ERROR,
                    (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                     errmsg("cannot use PID %d for inactive replication origin with ID %d",
                            acquired_by, node)));
 
        /* initialize new slot */
        if (free_slot == -1)
            ereport(ERROR,
                    (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
                     errmsg("could not find free replication state slot for replication origin with ID %d",
                            node),
                     errhint("Increase \"max_active_replication_origins\" and try again.")));
 
        session_replication_state = &replication_states[free_slot];
        Assert(!XLogRecPtrIsValid(session_replication_state->remote_lsn));
        Assert(!XLogRecPtrIsValid(session_replication_state->local_lsn));
        session_replication_state->roident = node;
    }
 
 
    Assert(session_replication_state->roident != InvalidReplOriginId);
 
    if (acquired_by == 0)
    {
        session_replication_state->acquired_by = MyProcPid;
        Assert(session_replication_state->refcount == 0);
    }
    else
    {
        /*
         * Sanity check: the origin must already be acquired by the process
         * passed as input, and at least one process must be using it.
         */
        Assert(session_replication_state->acquired_by == acquired_by);
        Assert(session_replication_state->refcount > 0);
    }
 
    session_replication_state->refcount++;
 
    LWLockRelease(ReplicationOriginLock);
 
    /* probably this one is pointless */
    ConditionVariableBroadcast(&session_replication_state->origin_cv);
}

References ReplicationState::acquired_by, Assert, ConditionVariableBroadcast(), ereport, errcode(), errhint(), errmsg, ERROR, fb(), i, InvalidReplOriginId, ReplicationState::local_lsn, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), max_active_replication_origins, MyProcPid, on_shmem_exit(), ReplicationState::origin_cv, ReplicationState::refcount, ReplicationState::remote_lsn, replication_states, ReplicationOriginExitCleanup(), ReplicationState::roident, session_replication_state, and XLogRecPtrIsValid.

Referenced by LogicalRepSyncTableStart(), ParallelApplyWorkerMain(), pg_replication_origin_session_setup(), and run_apply_worker().

replorigin_xact_clear()

void replorigin_xact_clear ( bool  clear_origin )

extern

Definition at line 1377 of file origin.c.

{
    replorigin_xact_state.origin_lsn = InvalidXLogRecPtr;
    replorigin_xact_state.origin_timestamp = 0;
    if (clear_origin)
        replorigin_xact_state.origin = InvalidReplOriginId;
}

References fb(), InvalidReplOriginId, InvalidXLogRecPtr, ReplOriginXactState::origin, ReplOriginXactState::origin_lsn, ReplOriginXactState::origin_timestamp, and replorigin_xact_state.

Referenced by on_exit_clear_xact_state(), pg_replication_origin_session_reset(), pg_replication_origin_xact_reset(), ProcessSyncingTablesForSync(), and start_apply().

StartupReplicationOrigin()

void StartupReplicationOrigin ( void  )

extern

Definition at line 740 of file origin.c.

{
    const char *path = PG_REPLORIGIN_CHECKPOINT_FILENAME;
    int         fd;
    int         readBytes;
    uint32      magic = REPLICATION_STATE_MAGIC;
    int         last_state = 0;
    pg_crc32c   file_crc;
    pg_crc32c   crc;
 
    /* don't want to overwrite already existing state */
#ifdef USE_ASSERT_CHECKING
    static bool already_started = false;
 
    Assert(!already_started);
    already_started = true;
#endif
 
    if (max_active_replication_origins == 0)
        return;
 
    INIT_CRC32C(crc);
 
    elog(DEBUG2, "starting up replication origin progress state");
 
    fd = OpenTransientFile(path, O_RDONLY | PG_BINARY);
 
    /*
     * might have had max_active_replication_origins == 0 last run, or we just
     * brought up a standby.
     */
    if (fd < 0 && errno == ENOENT)
        return;
    else if (fd < 0)
        ereport(PANIC,
                (errcode_for_file_access(),
                 errmsg("could not open file \"%s\": %m",
                        path)));
 
    /* verify magic, that is written even if nothing was active */
    readBytes = read(fd, &magic, sizeof(magic));
    if (readBytes != sizeof(magic))
    {
        if (readBytes < 0)
            ereport(PANIC,
                    (errcode_for_file_access(),
                     errmsg("could not read file \"%s\": %m",
                            path)));
        else
            ereport(PANIC,
                    (errcode(ERRCODE_DATA_CORRUPTED),
                     errmsg("could not read file \"%s\": read %d of %zu",
                            path, readBytes, sizeof(magic))));
    }
    COMP_CRC32C(crc, &magic, sizeof(magic));
 
    if (magic != REPLICATION_STATE_MAGIC)
        ereport(PANIC,
                (errmsg("replication checkpoint has wrong magic %u instead of %u",
                        magic, REPLICATION_STATE_MAGIC)));
 
    /* we can skip locking here, no other access is possible */
 
    /* recover individual states, until there are no more to be found */
    while (true)
    {
        ReplicationStateOnDisk disk_state;
 
        readBytes = read(fd, &disk_state, sizeof(disk_state));
 
        if (readBytes < 0)
        {
            ereport(PANIC,
                    (errcode_for_file_access(),
                     errmsg("could not read file \"%s\": %m",
                            path)));
        }
 
        /* no further data */
        if (readBytes == sizeof(crc))
        {
            memcpy(&file_crc, &disk_state, sizeof(file_crc));
            break;
        }
 
        if (readBytes != sizeof(disk_state))
        {
            ereport(PANIC,
                    (errcode_for_file_access(),
                     errmsg("could not read file \"%s\": read %d of %zu",
                            path, readBytes, sizeof(disk_state))));
        }
 
        COMP_CRC32C(crc, &disk_state, sizeof(disk_state));
 
        if (last_state == max_active_replication_origins)
            ereport(PANIC,
                    (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
                     errmsg("could not find free replication state, increase \"max_active_replication_origins\"")));
 
        /* copy data to shared memory */
        replication_states[last_state].roident = disk_state.roident;
        replication_states[last_state].remote_lsn = disk_state.remote_lsn;
        last_state++;
 
        ereport(LOG,
                errmsg("recovered replication state of node %d to %X/%08X",
                       disk_state.roident,
                       LSN_FORMAT_ARGS(disk_state.remote_lsn)));
    }
 
    /* now check checksum */
    FIN_CRC32C(crc);
    if (file_crc != crc)
        ereport(PANIC,
                (errcode(ERRCODE_DATA_CORRUPTED),
                 errmsg("replication slot checkpoint has wrong checksum %u, expected %u",
                        crc, file_crc)));
 
    if (CloseTransientFile(fd) != 0)
        ereport(PANIC,
                (errcode_for_file_access(),
                 errmsg("could not close file \"%s\": %m",
                        path)));
}

References Assert, CloseTransientFile(), COMP_CRC32C, crc, DEBUG2, elog, ereport, errcode(), ERRCODE_DATA_CORRUPTED, errcode_for_file_access(), errmsg, fb(), fd(), FIN_CRC32C, INIT_CRC32C, LOG, LSN_FORMAT_ARGS, max_active_replication_origins, memcpy(), OpenTransientFile(), PANIC, PG_BINARY, PG_REPLORIGIN_CHECKPOINT_FILENAME, read, ReplicationState::remote_lsn, REPLICATION_STATE_MAGIC, replication_states, and ReplicationState::roident.

Referenced by StartupXLOG().

Variable Documentation

max_active_replication_origins

PGDLLIMPORT int max_active_replication_origins

extern

Definition at line 106 of file origin.c.

Referenced by check_new_cluster_subscription_configuration(), CheckPointReplicationOrigin(), logicalrep_worker_launch(), pg_show_replication_origin_status(), ReplicationOriginShmemAttach(), ReplicationOriginShmemInit(), ReplicationOriginShmemRequest(), replorigin_advance(), replorigin_check_prerequisites(), replorigin_get_progress(), replorigin_redo(), replorigin_session_reset(), replorigin_session_setup(), replorigin_state_clear(), and StartupReplicationOrigin().

replorigin_xact_state

PGDLLIMPORT ReplOriginXactState replorigin_xact_state

extern

Definition at line 168 of file origin.c.

                                            {
    .origin = InvalidReplOriginId,  /* assumed identity */
    .origin_lsn = InvalidXLogRecPtr,
    .origin_timestamp = 0
};

Referenced by apply_handle_commit_internal(), apply_handle_commit_prepared(), apply_handle_delete_internal(), apply_handle_prepare_internal(), apply_handle_rollback_prepared(), apply_handle_tuple_routing(), apply_handle_update_internal(), EndPrepare(), LogicalRepSyncTableStart(), pa_stream_abort(), ParallelApplyWorkerMain(), pg_replication_origin_session_is_setup(), pg_replication_origin_session_setup(), pg_replication_origin_xact_setup(), RecordTransactionAbort(), RecordTransactionAbortPrepared(), RecordTransactionCommit(), RecordTransactionCommitPrepared(), replorigin_xact_clear(), run_apply_worker(), XactLogAbortRecord(), XactLogCommitRecord(), and XLogRecordAssemble().