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

Typedefs
typedef struct xl_replorigin_set	xl_replorigin_set
typedef struct xl_replorigin_drop	xl_replorigin_drop

Functions
RepOriginId	replorigin_by_name (char *name, bool missing_ok)
RepOriginId	replorigin_create (char *name)
void	replorigin_drop_by_name (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)
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

Macro Definition Documentation

DoNotReplicateId

#define DoNotReplicateId   PG_UINT16_MAX

Definition at line 34 of file origin.h.

Referenced by EndPrepare(), RecordTransactionAbortPrepared(), RecordTransactionCommit(), RecordTransactionCommitPrepared(), replorigin_advance(), and replorigin_by_oid().

InvalidRepOriginId

#define InvalidRepOriginId   0

Definition at line 33 of file origin.h.

Referenced by CheckPointReplicationOrigin(), CommitTsShmemInit(), DeactivateCommitTs(), DecodeXLogRecord(), EndPrepare(), pg_decode_filter(), pg_replication_origin_session_is_setup(), pg_replication_origin_session_reset(), pg_show_replication_origin_status(), pgoutput_begin_txn(), pgoutput_stream_start(), PrepareRedoAdd(), RecordTransactionAbortPrepared(), RecordTransactionCommit(), RecordTransactionCommitPrepared(), replorigin_advance(), replorigin_by_oid(), replorigin_drop_guts(), replorigin_redo(), replorigin_session_advance(), replorigin_session_setup(), restoreTwoPhaseData(), TransactionIdGetCommitTsData(), xact_redo_commit(), XactLogAbortRecord(), XactLogCommitRecord(), and XLogRecordAssemble().

XLOG_REPLORIGIN_DROP

#define XLOG_REPLORIGIN_DROP   0x10

Definition at line 31 of file origin.h.

Referenced by replorigin_desc(), replorigin_drop_guts(), replorigin_identify(), and replorigin_redo().

XLOG_REPLORIGIN_SET

#define XLOG_REPLORIGIN_SET   0x00

Definition at line 30 of file origin.h.

Referenced by replorigin_advance(), replorigin_desc(), replorigin_identify(), and replorigin_redo().

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 557 of file origin.c.

References CloseTransientFile(), COMP_CRC32C, durable_rename(), ereport, errcode_for_file_access(), errmsg(), FIN_CRC32C, i, INIT_CRC32C, InvalidRepOriginId, ReplicationState::local_lsn, ReplicationState::lock, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_replication_slots, OpenTransientFile(), PANIC, PG_BINARY, ReplicationState::remote_lsn, ReplicationStateOnDisk::remote_lsn, REPLICATION_STATE_MAGIC, ReplicationState::roident, ReplicationStateOnDisk::roident, write, and XLogFlush().

Referenced by CheckPointGuts().

{
    const char *tmppath = "pg_logical/replorigin_checkpoint.tmp";
    const char *path = "pg_logical/replorigin_checkpoint";
    int         tmpfd;
    int         i;
    uint32      magic = REPLICATION_STATE_MAGIC;
    pg_crc32c   crc;

    if (max_replication_slots == 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_replication_slots; 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);
}

ReplicationOriginShmemInit()

void ReplicationOriginShmemInit

(

void

)

Definition at line 510 of file origin.c.

References ConditionVariableInit(), i, ReplicationState::lock, LWLockInitialize(), LWTRANCHE_REPLICATION_ORIGIN_STATE, max_replication_slots, MemSet, ReplicationState::origin_cv, ReplicationOriginShmemSize(), ShmemInitStruct(), ReplicationStateCtl::states, and ReplicationStateCtl::tranche_id.

Referenced by CreateSharedMemoryAndSemaphores().

{
    bool        found;

    if (max_replication_slots == 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_replication_slots; i++)
        {
            LWLockInitialize(&replication_states[i].lock,
                             replication_states_ctl->tranche_id);
            ConditionVariableInit(&replication_states[i].origin_cv);
        }
    }
}

ReplicationOriginShmemSize()

Size ReplicationOriginShmemSize

(

void

)

Definition at line 490 of file origin.c.

References add_size(), max_replication_slots, mul_size(), and offsetof.

Referenced by CreateSharedMemoryAndSemaphores(), and ReplicationOriginShmemInit().

{
    Size        size = 0;

    /*
     * XXX: max_replication_slots is arguably the wrong thing to use, as here
     * we keep the replay state of *remote* transactions. But for now it seems
     * sufficient to reuse it, rather than introduce a separate GUC.
     */
    if (max_replication_slots == 0)
        return size;

    size = add_size(size, offsetof(ReplicationStateCtl, states));

    size = add_size(size,
                    mul_size(max_replication_slots, sizeof(ReplicationState)));
    return size;
}

replorigin_advance()

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

Definition at line 872 of file origin.c.

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_replication_slots, xl_replorigin_set::node_id, xl_replorigin_set::remote_lsn, ReplicationState::remote_lsn, ReplicationState::roident, XLOG_REPLORIGIN_SET, XLogBeginInsert(), XLogInsert(), and XLogRegisterData().

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

{
    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_replication_slots; 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 OID %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 OID %u",
                        node),
                 errhint("Increase max_replication_slots 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((char *) (&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.
     * 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);
}

replorigin_by_name()

RepOriginId replorigin_by_name	(	char *	name,
		bool	missing_ok
	)

Definition at line 209 of file origin.c.

References CStringGetTextDatum, ereport, errcode(), errmsg(), ERROR, GETSTRUCT, HeapTupleIsValid, InvalidOid, ReleaseSysCache(), REPLORIGNAME, ReplicationState::roident, and SearchSysCache1().

Referenced by ApplyWorkerMain(), LogicalRepSyncTableStart(), pg_replication_origin_advance(), pg_replication_origin_oid(), pg_replication_origin_progress(), pg_replication_origin_session_setup(), and replorigin_drop_by_name().

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

replorigin_by_oid()

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

Definition at line 449 of file origin.c.

References Assert, DoNotReplicateId, ereport, errcode(), errmsg(), ERROR, GETSTRUCT, HeapTupleIsValid, InvalidRepOriginId, ObjectIdGetDatum, OidIsValid, ReleaseSysCache(), REPLORIGIDENT, SearchSysCache1(), and text_to_cstring().

Referenced by pg_show_replication_origin_status(), pgoutput_begin_txn(), and pgoutput_stream_start().

{
    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 OID %u does not exist",
                            roident)));

        return false;
    }
}

replorigin_create()

RepOriginId replorigin_create

(

char *

name

)

Definition at line 240 of file origin.c.

References Assert, BTEqualStrategyNumber, CatalogTupleInsert(), CHECK_FOR_INTERRUPTS, CommandCounterIncrement(), CStringGetTextDatum, ereport, errcode(), errmsg(), ERROR, ExclusiveLock, heap_form_tuple(), heap_freetuple(), HeapTupleIsValid, InitDirtySnapshot, InvalidOid, IsTransactionState(), sort-test::key, ObjectIdGetDatum, PG_UINT16_MAX, RelationGetDescr, ReplicationOriginIdentIndex, ReplicationState::roident, ScanKeyInit(), systable_beginscan(), systable_endscan(), systable_getnext(), table_close(), table_open(), and values.

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

{
    Oid         roident;
    HeapTuple   tuple = NULL;
    Relation    rel;
    Datum       roname_d;
    SnapshotData SnapshotDirty;
    SysScanDesc scan;
    ScanKeyData key;

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

    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 OID")));

    heap_freetuple(tuple);
    return roident;
}

replorigin_desc()

void replorigin_desc	(	StringInfo	buf,
		XLogReaderState *	record
	)

Definition at line 19 of file replorigindesc.c.

References appendStringInfo(), 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, XLogRecGetInfo, and XLR_INFO_MASK.

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

replorigin_drop_by_name()

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

Definition at line 414 of file origin.c.

References Assert, ExclusiveLock, IsTransactionState(), NoLock, OidIsValid, replorigin_by_name(), replorigin_drop_guts(), ReplicationState::roident, table_close(), and table_open().

Referenced by AlterSubscription_refresh(), DropSubscription(), pg_replication_origin_drop(), and process_syncing_tables_for_apply().

{
    RepOriginId roident;
    Relation    rel;

    Assert(IsTransactionState());

    /*
     * To interlock against concurrent drops, we hold ExclusiveLock on
     * pg_replication_origin till xact commit.
     *
     * XXX We can optimize this by acquiring the lock on a specific origin by
     * using LockSharedObject if required. However, for that, we first to
     * acquire a lock on ReplicationOriginRelationId, get the origin_id, lock
     * the specific origin and then re-check if the origin still exists.
     */
    rel = table_open(ReplicationOriginRelationId, ExclusiveLock);

    roident = replorigin_by_name(name, missing_ok);

    if (OidIsValid(roident))
        replorigin_drop_guts(rel, roident, nowait);

    /* We keep the lock on pg_replication_origin until commit */
    table_close(rel, NoLock);
}

replorigin_get_progress()

XLogRecPtr replorigin_get_progress	(	RepOriginId	node,
		bool	flush
	)

Definition at line 995 of file origin.c.

References i, InvalidXLogRecPtr, ReplicationState::local_lsn, ReplicationState::lock, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_replication_slots, ReplicationState::remote_lsn, ReplicationState::roident, and XLogFlush().

Referenced by pg_replication_origin_progress().

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

replorigin_identify()

const char* replorigin_identify

(

uint8

info

)

Definition at line 51 of file replorigindesc.c.

References XLOG_REPLORIGIN_DROP, and XLOG_REPLORIGIN_SET.

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

replorigin_redo()

void replorigin_redo

(

XLogReaderState *

record

)

Definition at line 811 of file origin.c.

References elog, XLogReaderState::EndRecPtr, xl_replorigin_set::force, i, InvalidRepOriginId, InvalidXLogRecPtr, ReplicationState::local_lsn, max_replication_slots, xl_replorigin_set::node_id, xl_replorigin_drop::node_id, PANIC, xl_replorigin_set::remote_lsn, ReplicationState::remote_lsn, replorigin_advance(), ReplicationState::roident, XLOG_REPLORIGIN_DROP, XLOG_REPLORIGIN_SET, XLogRecGetData, XLogRecGetInfo, and XLR_INFO_MASK.

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

replorigin_session_advance()

void replorigin_session_advance	(	XLogRecPtr	remote_commit,
		XLogRecPtr	local_commit
	)

Definition at line 1185 of file origin.c.

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

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

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

replorigin_session_get_progress()

XLogRecPtr replorigin_session_get_progress

(

bool

flush

)

Definition at line 1203 of file origin.c.

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

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

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

replorigin_session_reset()

void replorigin_session_reset

(

void

)

Definition at line 1156 of file origin.c.

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

Referenced by pg_replication_origin_session_reset().

{
    ConditionVariable *cv;

    Assert(max_replication_slots != 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);
}

replorigin_session_setup()

void replorigin_session_setup

(

RepOriginId

node

)

Definition at line 1068 of file origin.c.

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

Referenced by ApplyWorkerMain(), LogicalRepSyncTableStart(), and pg_replication_origin_session_setup().

{
    static bool registered_cleanup;
    int         i;
    int         free_slot = -1;

    if (!registered_cleanup)
    {
        on_shmem_exit(ReplicationOriginExitCleanup, 0);
        registered_cleanup = true;
    }

    Assert(max_replication_slots > 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_replication_slots; 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)
        {
            ereport(ERROR,
                    (errcode(ERRCODE_OBJECT_IN_USE),
                     errmsg("replication origin with OID %d is already active for PID %d",
                            curstate->roident, curstate->acquired_by)));
        }

        /* ok, found slot */
        session_replication_state = curstate;
    }


    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 OID %u",
                        node),
                 errhint("Increase max_replication_slots 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);

    session_replication_state->acquired_by = MyProcPid;

    LWLockRelease(ReplicationOriginLock);

    /* probably this one is pointless */
    ConditionVariableBroadcast(&session_replication_state->origin_cv);
}

StartupReplicationOrigin()

void StartupReplicationOrigin

(

void

)

Definition at line 683 of file origin.c.

References Assert, CloseTransientFile(), COMP_CRC32C, DEBUG2, elog, ereport, errcode(), ERRCODE_DATA_CORRUPTED, errcode_for_file_access(), errmsg(), fd(), FIN_CRC32C, INIT_CRC32C, LOG, LSN_FORMAT_ARGS, max_replication_slots, OpenTransientFile(), PANIC, PG_BINARY, read, ReplicationState::remote_lsn, ReplicationStateOnDisk::remote_lsn, REPLICATION_STATE_MAGIC, ReplicationState::roident, and ReplicationStateOnDisk::roident.

Referenced by StartupXLOG().

{
    const char *path = "pg_logical/replorigin_checkpoint";
    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_replication_slots == 0)
        return;

    INIT_CRC32C(crc);

    elog(DEBUG2, "starting up replication origin progress state");

    fd = OpenTransientFile(path, O_RDONLY | PG_BINARY);

    /*
     * might have had max_replication_slots == 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_replication_slots)
            ereport(PANIC,
                    (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
                     errmsg("could not find free replication state, increase max_replication_slots")));

        /* 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 %u 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_CONFIGURATION_LIMIT_EXCEEDED),
                 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)));
}

Variable Documentation

replorigin_session_origin

PGDLLIMPORT RepOriginId replorigin_session_origin

Definition at line 154 of file origin.c.

Referenced by ApplyWorkerMain(), EndPrepare(), LogicalRepSyncTableStart(), pg_replication_origin_session_is_setup(), pg_replication_origin_session_reset(), pg_replication_origin_session_setup(), RecordTransactionAbortPrepared(), RecordTransactionCommit(), RecordTransactionCommitPrepared(), XactLogAbortRecord(), XactLogCommitRecord(), and XLogRecordAssemble().

replorigin_session_origin_lsn

PGDLLIMPORT XLogRecPtr replorigin_session_origin_lsn

Definition at line 155 of file origin.c.

Referenced by apply_handle_commit_internal(), EndPrepare(), pg_replication_origin_session_reset(), pg_replication_origin_xact_reset(), pg_replication_origin_xact_setup(), RecordTransactionAbortPrepared(), RecordTransactionCommit(), RecordTransactionCommitPrepared(), XactLogAbortRecord(), and XactLogCommitRecord().

replorigin_session_origin_timestamp

PGDLLIMPORT TimestampTz replorigin_session_origin_timestamp

Definition at line 156 of file origin.c.

Referenced by apply_handle_commit_internal(), EndPrepare(), pg_replication_origin_session_reset(), pg_replication_origin_xact_reset(), pg_replication_origin_xact_setup(), RecordTransactionCommit(), RecordTransactionCommitPrepared(), XactLogAbortRecord(), and XactLogCommitRecord().