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

Macros
#define	XLOG_REPLORIGIN_SET   0x00
#define	XLOG_REPLORIGIN_DROP   0x10
#define	InvalidRepOriginId   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

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

Variables
PGDLLIMPORT RepOriginId	replorigin_session_origin
PGDLLIMPORT XLogRecPtr	replorigin_session_origin_lsn
PGDLLIMPORT TimestampTz	replorigin_session_origin_timestamp
PGDLLIMPORT int	max_active_replication_origins

Macro Definition Documentation

DoNotReplicateId

#define DoNotReplicateId   PG_UINT16_MAX

Definition at line 34 of file origin.h.

InvalidRepOriginId

#define InvalidRepOriginId   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

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

)

Definition at line 596 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 == InvalidRepOriginId)
            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(), FIN_CRC32C, i, INIT_CRC32C, InvalidRepOriginId, ReplicationState::local_lsn, ReplicationState::lock, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_active_replication_origins, OpenTransientFile(), PANIC, PG_BINARY, PG_REPLORIGIN_CHECKPOINT_FILENAME, PG_REPLORIGIN_CHECKPOINT_TMPFILE, ReplicationState::remote_lsn, ReplicationStateOnDisk::remote_lsn, REPLICATION_STATE_MAGIC, replication_states, ReplicationState::roident, ReplicationStateOnDisk::roident, write, and XLogFlush().

Referenced by CheckPointGuts().

ReplicationOriginShmemInit()

void ReplicationOriginShmemInit

(

void

)

Definition at line 549 of file origin.c.

{
    bool        found;
 
    if (max_active_replication_origins == 0)
        return;
 
    replication_states_ctl = (ReplicationStateCtl *)
        ShmemInitStruct("ReplicationOriginState",
                        ReplicationOriginShmemSize(),
                        &found);
    replication_states = replication_states_ctl->states;
 
    if (!found)
    {
        int         i;
 
        MemSet(replication_states_ctl, 0, ReplicationOriginShmemSize());
 
        replication_states_ctl->tranche_id = LWTRANCHE_REPLICATION_ORIGIN_STATE;
 
        for (i = 0; i < max_active_replication_origins; i++)
        {
            LWLockInitialize(&replication_states[i].lock,
                             replication_states_ctl->tranche_id);
            ConditionVariableInit(&replication_states[i].origin_cv);
        }
    }
}

References ConditionVariableInit(), i, LWLockInitialize(), LWTRANCHE_REPLICATION_ORIGIN_STATE, max_active_replication_origins, MemSet, replication_states, replication_states_ctl, ReplicationOriginShmemSize(), ShmemInitStruct(), ReplicationStateCtl::states, and ReplicationStateCtl::tranche_id.

Referenced by CreateOrAttachShmemStructs().

ReplicationOriginShmemSize()

Size ReplicationOriginShmemSize

(

void

)

Definition at line 534 of file origin.c.

{
    Size        size = 0;
 
    if (max_active_replication_origins == 0)
        return size;
 
    size = add_size(size, offsetof(ReplicationStateCtl, states));
 
    size = add_size(size,
                    mul_size(max_active_replication_origins, sizeof(ReplicationState)));
    return size;
}

References add_size(), max_active_replication_origins, and mul_size().

Referenced by CalculateShmemSize(), and ReplicationOriginShmemInit().

replorigin_advance()

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

Definition at line 911 of file origin.c.

{
    int         i;
    ReplicationState *replication_state = NULL;
    ReplicationState *free_state = NULL;
 
    Assert(node != InvalidRepOriginId);
 
    /* 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 == InvalidRepOriginId &&
            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->acquired_by != 0)
        {
            ereport(ERROR,
                    (errcode(ERRCODE_OBJECT_IN_USE),
                     errmsg("replication origin with ID %d is already active for PID %d",
                            replication_state->roident,
                            replication_state->acquired_by)));
        }
 
        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(replication_state->remote_lsn == InvalidXLogRecPtr);
        Assert(replication_state->local_lsn == InvalidXLogRecPtr);
        replication_state->roident = node;
    }
 
    Assert(replication_state->roident != InvalidRepOriginId);
 
    /*
     * 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 (local_commit != InvalidXLogRecPtr &&
        (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 ReplicationState::acquired_by, Assert(), DoNotReplicateId, ereport, errcode(), errhint(), errmsg(), ERROR, xl_replorigin_set::force, i, InvalidRepOriginId, InvalidXLogRecPtr, ReplicationState::local_lsn, ReplicationState::lock, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), max_active_replication_origins, xl_replorigin_set::node_id, ReplicationState::remote_lsn, xl_replorigin_set::remote_lsn, replication_states, ReplicationState::roident, XLOG_REPLORIGIN_SET, XLogBeginInsert(), XLogInsert(), 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()

RepOriginId replorigin_by_name	(	const char *	roname,
		bool	missing_ok
	)

Definition at line 226 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, 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	(	RepOriginId	roident,
		bool	missing_ok,
		char **	roname
	)

Definition at line 493 of file origin.c.

{
    HeapTuple   tuple;
    Form_pg_replication_origin ric;
 
    Assert(OidIsValid((Oid) roident));
    Assert(roident != InvalidRepOriginId);
    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, GETSTRUCT(), HeapTupleIsValid, InvalidRepOriginId, ObjectIdGetDatum(), OidIsValid, ReleaseSysCache(), SearchSysCache1(), and text_to_cstring().

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

replorigin_create()

RepOriginId replorigin_create

(

const char *

roname

)

Definition at line 257 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, heap_form_tuple(), heap_freetuple(), HeapTupleIsValid, InitDirtySnapshot, InvalidOid, IsTransactionState(), sort-test::key, 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
	)

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/%X; 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, xl_replorigin_set::force, LSN_FORMAT_ARGS, xl_replorigin_set::node_id, xl_replorigin_drop::node_id, xl_replorigin_set::remote_lsn, 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
	)

Definition at line 439 of file origin.c.

{
    RepOriginId 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, 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(), process_syncing_tables_for_apply(), and process_syncing_tables_for_sync().

replorigin_get_progress()

XLogRecPtr replorigin_get_progress	(	RepOriginId	node,
		bool	flush
	)

Definition at line 1037 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 && local_lsn != InvalidXLogRecPtr)
        XLogFlush(local_lsn);
 
    return remote_lsn;
}

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

Referenced by AlterSubscription(), and pg_replication_origin_progress().

replorigin_identify()

const char * replorigin_identify

(

uint8

info

)

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 XLOG_REPLORIGIN_DROP, and XLOG_REPLORIGIN_SET.

replorigin_redo()

void replorigin_redo

(

XLogReaderState *

record

)

Definition at line 850 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 = InvalidRepOriginId;
                        state->remote_lsn = InvalidXLogRecPtr;
                        state->local_lsn = InvalidXLogRecPtr;
                        break;
                    }
                }
                break;
            }
        default:
            elog(PANIC, "replorigin_redo: unknown op code %u", info);
    }
}

References elog, XLogReaderState::EndRecPtr, xl_replorigin_set::force, i, InvalidRepOriginId, InvalidXLogRecPtr, max_active_replication_origins, xl_replorigin_set::node_id, xl_replorigin_drop::node_id, PANIC, xl_replorigin_set::remote_lsn, 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
	)

Definition at line 1242 of file origin.c.

{
    Assert(session_replication_state != NULL);
    Assert(session_replication_state->roident != InvalidRepOriginId);
 
    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(), InvalidRepOriginId, 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

)

Definition at line 1260 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 && local_lsn != InvalidXLogRecPtr)
        XLogFlush(local_lsn);
 
    return remote_lsn;
}

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

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

replorigin_session_reset()

void replorigin_session_reset

(

void

)

Definition at line 1213 of file origin.c.

{
    ConditionVariable *cv;
 
    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")));
 
    LWLockAcquire(ReplicationOriginLock, LW_EXCLUSIVE);
 
    session_replication_state->acquired_by = 0;
    cv = &session_replication_state->origin_cv;
    session_replication_state = NULL;
 
    LWLockRelease(ReplicationOriginLock);
 
    ConditionVariableBroadcast(cv);
}

References ReplicationState::acquired_by, Assert(), ConditionVariableBroadcast(), ereport, errcode(), errmsg(), ERROR, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), max_active_replication_origins, ReplicationState::origin_cv, and session_replication_state.

Referenced by pg_replication_origin_session_reset(), and process_syncing_tables_for_sync().

replorigin_session_setup()

void replorigin_session_setup	(	RepOriginId	node,
		int	acquired_by
	)

Definition at line 1120 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 == InvalidRepOriginId &&
            free_slot == -1)
        {
            free_slot = i;
            continue;
        }
 
        /* not our slot */
        if (curstate->roident != node)
            continue;
 
        else if (curstate->acquired_by != 0 && 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)));
        }
 
        /* ok, found slot */
        session_replication_state = curstate;
        break;
    }
 
 
    if (session_replication_state == NULL && 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.")));
    else if (session_replication_state == NULL)
    {
        /* initialize new slot */
        session_replication_state = &replication_states[free_slot];
        Assert(session_replication_state->remote_lsn == InvalidXLogRecPtr);
        Assert(session_replication_state->local_lsn == InvalidXLogRecPtr);
        session_replication_state->roident = node;
    }
 
 
    Assert(session_replication_state->roident != InvalidRepOriginId);
 
    if (acquired_by == 0)
        session_replication_state->acquired_by = MyProcPid;
    else if (session_replication_state->acquired_by != acquired_by)
        elog(ERROR, "could not find replication state slot for replication origin with OID %u which was acquired by %d",
             node, acquired_by);
 
    LWLockRelease(ReplicationOriginLock);
 
    /* probably this one is pointless */
    ConditionVariableBroadcast(&session_replication_state->origin_cv);
}

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

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

StartupReplicationOrigin()

void StartupReplicationOrigin

(

void

)

Definition at line 722 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));
 
        /* no further data */
        if (readBytes == sizeof(crc))
        {
            /* not pretty, but simple ... */
            file_crc = *(pg_crc32c *) &disk_state;
            break;
        }
 
        if (readBytes < 0)
        {
            ereport(PANIC,
                    (errcode_for_file_access(),
                     errmsg("could not read file \"%s\": %m",
                            path)));
        }
 
        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/%X",
                        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(), fd(), FIN_CRC32C, INIT_CRC32C, LOG, LSN_FORMAT_ARGS, max_active_replication_origins, OpenTransientFile(), PANIC, PG_BINARY, PG_REPLORIGIN_CHECKPOINT_FILENAME, read, ReplicationState::remote_lsn, ReplicationStateOnDisk::remote_lsn, REPLICATION_STATE_MAGIC, replication_states, ReplicationState::roident, and ReplicationStateOnDisk::roident.

Referenced by StartupXLOG().

Variable Documentation

max_active_replication_origins

PGDLLIMPORT int max_active_replication_origins

extern

Definition at line 104 of file origin.c.

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

replorigin_session_origin

PGDLLIMPORT RepOriginId replorigin_session_origin

extern

Definition at line 163 of file origin.c.

Referenced by apply_handle_delete_internal(), apply_handle_tuple_routing(), apply_handle_update_internal(), EndPrepare(), LogicalRepSyncTableStart(), ParallelApplyWorkerMain(), pg_replication_origin_session_is_setup(), pg_replication_origin_session_reset(), pg_replication_origin_session_setup(), process_syncing_tables_for_sync(), RecordTransactionAbort(), RecordTransactionAbortPrepared(), RecordTransactionCommit(), RecordTransactionCommitPrepared(), replorigin_reset(), run_apply_worker(), XactLogAbortRecord(), XactLogCommitRecord(), and XLogRecordAssemble().

replorigin_session_origin_lsn

PGDLLIMPORT XLogRecPtr replorigin_session_origin_lsn

extern

Definition at line 164 of file origin.c.

Referenced by apply_handle_commit_internal(), apply_handle_commit_prepared(), apply_handle_prepare_internal(), apply_handle_rollback_prepared(), EndPrepare(), pa_stream_abort(), pg_replication_origin_session_reset(), pg_replication_origin_xact_reset(), pg_replication_origin_xact_setup(), process_syncing_tables_for_sync(), RecordTransactionAbort(), RecordTransactionAbortPrepared(), RecordTransactionCommit(), RecordTransactionCommitPrepared(), replorigin_reset(), XactLogAbortRecord(), and XactLogCommitRecord().

replorigin_session_origin_timestamp

PGDLLIMPORT TimestampTz replorigin_session_origin_timestamp

extern

Definition at line 165 of file origin.c.

Referenced by apply_handle_commit_internal(), apply_handle_commit_prepared(), apply_handle_prepare_internal(), apply_handle_rollback_prepared(), EndPrepare(), pa_stream_abort(), pg_replication_origin_session_reset(), pg_replication_origin_xact_reset(), pg_replication_origin_xact_setup(), process_syncing_tables_for_sync(), RecordTransactionCommit(), RecordTransactionCommitPrepared(), replorigin_reset(), XactLogAbortRecord(), and XactLogCommitRecord().