origin.c File Reference

#include "postgres.h"
#include <unistd.h>
#include <sys/stat.h>
#include "access/genam.h"
#include "access/htup_details.h"
#include "access/table.h"
#include "access/xact.h"
#include "access/xloginsert.h"
#include "catalog/catalog.h"
#include "catalog/indexing.h"
#include "catalog/pg_subscription.h"
#include "funcapi.h"
#include "miscadmin.h"
#include "nodes/execnodes.h"
#include "pgstat.h"
#include "replication/origin.h"
#include "replication/slot.h"
#include "storage/condition_variable.h"
#include "storage/fd.h"
#include "storage/ipc.h"
#include "storage/lmgr.h"
#include "storage/subsystems.h"
#include "utils/builtins.h"
#include "utils/fmgroids.h"
#include "utils/guc.h"
#include "utils/pg_lsn.h"
#include "utils/rel.h"
#include "utils/snapmgr.h"
#include "utils/syscache.h"
#include "utils/wait_event.h"

Include dependency graph for origin.c:

Go to the source code of this file.

Data Structures
struct	ReplicationState
struct	ReplicationStateOnDisk
struct	ReplicationStateCtl

Macros
#define	PG_REPLORIGIN_CHECKPOINT_FILENAME   PG_LOGICAL_DIR "/replorigin_checkpoint"
#define	PG_REPLORIGIN_CHECKPOINT_TMPFILE   PG_REPLORIGIN_CHECKPOINT_FILENAME ".tmp"
#define	REPLICATION_STATE_MAGIC   ((uint32) 0x1257DADE)
#define	REPLICATION_ORIGIN_PROGRESS_COLS   4

Typedefs
typedef struct ReplicationState	ReplicationState
typedef struct ReplicationStateOnDisk	ReplicationStateOnDisk
typedef struct ReplicationStateCtl	ReplicationStateCtl

Functions
static void	ReplicationOriginShmemRequest (void *arg)
static void	ReplicationOriginShmemInit (void *arg)
static void	ReplicationOriginShmemAttach (void *arg)
static void	replorigin_check_prerequisites (bool check_origins, bool recoveryOK)
static bool	IsReservedOriginName (const char *name)
ReplOriginId	replorigin_by_name (const char *roname, bool missing_ok)
ReplOriginId	replorigin_create (const char *roname)
static void	replorigin_state_clear (ReplOriginId roident, bool nowait)
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	CheckPointReplicationOrigin (void)
void	StartupReplicationOrigin (void)
void	replorigin_redo (XLogReaderState *record)
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)
static void	replorigin_session_reset_internal (void)
static void	ReplicationOriginExitCleanup (int code, Datum arg)
void	replorigin_session_setup (ReplOriginId node, int acquired_by)
void	replorigin_session_reset (void)
void	replorigin_session_advance (XLogRecPtr remote_commit, XLogRecPtr local_commit)
XLogRecPtr	replorigin_session_get_progress (bool flush)
void	replorigin_xact_clear (bool clear_origin)
Datum	pg_replication_origin_create (PG_FUNCTION_ARGS)
Datum	pg_replication_origin_drop (PG_FUNCTION_ARGS)
Datum	pg_replication_origin_oid (PG_FUNCTION_ARGS)
Datum	pg_replication_origin_session_setup (PG_FUNCTION_ARGS)
Datum	pg_replication_origin_session_reset (PG_FUNCTION_ARGS)
Datum	pg_replication_origin_session_is_setup (PG_FUNCTION_ARGS)
Datum	pg_replication_origin_session_progress (PG_FUNCTION_ARGS)
Datum	pg_replication_origin_xact_setup (PG_FUNCTION_ARGS)
Datum	pg_replication_origin_xact_reset (PG_FUNCTION_ARGS)
Datum	pg_replication_origin_advance (PG_FUNCTION_ARGS)
Datum	pg_replication_origin_progress (PG_FUNCTION_ARGS)
Datum	pg_show_replication_origin_status (PG_FUNCTION_ARGS)

Variables
int	max_active_replication_origins = 10
ReplOriginXactState	replorigin_xact_state
static ReplicationState *	replication_states
const ShmemCallbacks	ReplicationOriginShmemCallbacks
static ReplicationStateCtl *	replication_states_ctl
static ReplicationState *	session_replication_state = NULL

Macro Definition Documentation

PG_REPLORIGIN_CHECKPOINT_FILENAME

#define PG_REPLORIGIN_CHECKPOINT_FILENAME   PG_LOGICAL_DIR "/replorigin_checkpoint"

Definition at line 102 of file origin.c.

PG_REPLORIGIN_CHECKPOINT_TMPFILE

#define PG_REPLORIGIN_CHECKPOINT_TMPFILE   PG_REPLORIGIN_CHECKPOINT_FILENAME ".tmp"

Definition at line 103 of file origin.c.

REPLICATION_ORIGIN_PROGRESS_COLS

#define REPLICATION_ORIGIN_PROGRESS_COLS   4

REPLICATION_STATE_MAGIC

#define REPLICATION_STATE_MAGIC   ((uint32) 0x1257DADE)

Definition at line 204 of file origin.c.

Typedef Documentation

ReplicationState

typedef struct ReplicationState ReplicationState

ReplicationStateCtl

typedef struct ReplicationStateCtl ReplicationStateCtl

ReplicationStateOnDisk

typedef struct ReplicationStateOnDisk ReplicationStateOnDisk

Function Documentation

CheckPointReplicationOrigin()

void CheckPointReplicationOrigin

(

void

)

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().

IsReservedOriginName()

static bool IsReservedOriginName ( const char *  name )

static

Definition at line 226 of file origin.c.

{
    return ((pg_strcasecmp(name, LOGICALREP_ORIGIN_NONE) == 0) ||
            (pg_strcasecmp(name, LOGICALREP_ORIGIN_ANY) == 0));
}

References fb(), name, and pg_strcasecmp().

Referenced by pg_replication_origin_create().

pg_replication_origin_advance()

Datum pg_replication_origin_advance

(

PG_FUNCTION_ARGS

)

Definition at line 1585 of file origin.c.

{
    text       *name = PG_GETARG_TEXT_PP(0);
    XLogRecPtr  remote_commit = PG_GETARG_LSN(1);
    ReplOriginId node;
 
    replorigin_check_prerequisites(true, false);
 
    /* lock to prevent the replication origin from vanishing */
    LockRelationOid(ReplicationOriginRelationId, RowExclusiveLock);
 
    node = replorigin_by_name(text_to_cstring(name), false);
 
    /*
     * Can't sensibly pass a local commit to be flushed at checkpoint - this
     * xact hasn't committed yet. This is why this function should be used to
     * set up the initial replication state, but not for replay.
     */
    replorigin_advance(node, remote_commit, InvalidXLogRecPtr,
                       true /* go backward */ , true /* WAL log */ );
 
    UnlockRelationOid(ReplicationOriginRelationId, RowExclusiveLock);
 
    PG_RETURN_VOID();
}

References fb(), InvalidXLogRecPtr, LockRelationOid(), name, PG_GETARG_LSN, PG_GETARG_TEXT_PP, PG_RETURN_VOID, replorigin_advance(), replorigin_by_name(), replorigin_check_prerequisites(), RowExclusiveLock, text_to_cstring(), and UnlockRelationOid().

pg_replication_origin_create()

Datum pg_replication_origin_create

(

PG_FUNCTION_ARGS

)

Definition at line 1398 of file origin.c.

{
    char       *name;
    ReplOriginId roident;
 
    replorigin_check_prerequisites(false, false);
 
    name = text_to_cstring((text *) DatumGetPointer(PG_GETARG_DATUM(0)));
 
    /*
     * Replication origins "any and "none" are reserved for system options.
     * The origins "pg_xxx" are reserved for internal use.
     */
    if (IsReservedName(name) || IsReservedOriginName(name))
        ereport(ERROR,
                (errcode(ERRCODE_RESERVED_NAME),
                 errmsg("replication origin name \"%s\" is reserved",
                        name),
                 errdetail("Origin names \"%s\", \"%s\", and names starting with \"pg_\" are reserved.",
                           LOGICALREP_ORIGIN_ANY, LOGICALREP_ORIGIN_NONE)));
 
    /*
     * If built with appropriate switch, whine when regression-testing
     * conventions for replication origin names are violated.
     */
#ifdef ENFORCE_REGRESSION_TEST_NAME_RESTRICTIONS
    if (strncmp(name, "regress_", 8) != 0)
        elog(WARNING, "replication origins created by regression test cases should have names starting with \"regress_\"");
#endif
 
    roident = replorigin_create(name);
 
    pfree(name);
 
    PG_RETURN_OID(roident);
}

References DatumGetPointer(), elog, ereport, errcode(), errdetail(), errmsg, ERROR, fb(), IsReservedName(), IsReservedOriginName(), name, pfree(), PG_GETARG_DATUM, PG_RETURN_OID, replorigin_check_prerequisites(), replorigin_create(), text_to_cstring(), and WARNING.

pg_replication_origin_drop()

Datum pg_replication_origin_drop

(

PG_FUNCTION_ARGS

)

Definition at line 1439 of file origin.c.

{
    char       *name;
 
    replorigin_check_prerequisites(false, false);
 
    name = text_to_cstring((text *) DatumGetPointer(PG_GETARG_DATUM(0)));
 
    replorigin_drop_by_name(name, false, true);
 
    pfree(name);
 
    PG_RETURN_VOID();
}

References DatumGetPointer(), name, pfree(), PG_GETARG_DATUM, PG_RETURN_VOID, replorigin_check_prerequisites(), replorigin_drop_by_name(), and text_to_cstring().

pg_replication_origin_oid()

Datum pg_replication_origin_oid

(

PG_FUNCTION_ARGS

)

Definition at line 1458 of file origin.c.

{
    char       *name;
    ReplOriginId roident;
 
    replorigin_check_prerequisites(false, false);
 
    name = text_to_cstring((text *) DatumGetPointer(PG_GETARG_DATUM(0)));
    roident = replorigin_by_name(name, true);
 
    pfree(name);
 
    if (OidIsValid(roident))
        PG_RETURN_OID(roident);
    PG_RETURN_NULL();
}

References DatumGetPointer(), name, OidIsValid, pfree(), PG_GETARG_DATUM, PG_RETURN_NULL, PG_RETURN_OID, replorigin_by_name(), replorigin_check_prerequisites(), and text_to_cstring().

pg_replication_origin_progress()

Datum pg_replication_origin_progress

(

PG_FUNCTION_ARGS

)

Definition at line 1620 of file origin.c.

{
    char       *name;
    bool        flush;
    ReplOriginId roident;
    XLogRecPtr  remote_lsn = InvalidXLogRecPtr;
 
    replorigin_check_prerequisites(true, true);
 
    name = text_to_cstring((text *) DatumGetPointer(PG_GETARG_DATUM(0)));
    flush = PG_GETARG_BOOL(1);
 
    roident = replorigin_by_name(name, false);
    Assert(OidIsValid(roident));
 
    remote_lsn = replorigin_get_progress(roident, flush);
 
    if (!XLogRecPtrIsValid(remote_lsn))
        PG_RETURN_NULL();
 
    PG_RETURN_LSN(remote_lsn);
}

References Assert, DatumGetPointer(), InvalidXLogRecPtr, name, OidIsValid, PG_GETARG_BOOL, PG_GETARG_DATUM, PG_RETURN_LSN, PG_RETURN_NULL, replorigin_by_name(), replorigin_check_prerequisites(), replorigin_get_progress(), text_to_cstring(), and XLogRecPtrIsValid.

pg_replication_origin_session_is_setup()

Datum pg_replication_origin_session_is_setup

(

PG_FUNCTION_ARGS

)

Definition at line 1518 of file origin.c.

{
    replorigin_check_prerequisites(false, false);
 
    PG_RETURN_BOOL(replorigin_xact_state.origin != InvalidReplOriginId);
}

References InvalidReplOriginId, ReplOriginXactState::origin, PG_RETURN_BOOL, replorigin_check_prerequisites(), and replorigin_xact_state.

pg_replication_origin_session_progress()

Datum pg_replication_origin_session_progress

(

PG_FUNCTION_ARGS

)

Definition at line 1534 of file origin.c.

{
    XLogRecPtr  remote_lsn = InvalidXLogRecPtr;
    bool        flush = PG_GETARG_BOOL(0);
 
    replorigin_check_prerequisites(true, false);
 
    if (session_replication_state == NULL)
        ereport(ERROR,
                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                 errmsg("no replication origin is configured")));
 
    remote_lsn = replorigin_session_get_progress(flush);
 
    if (!XLogRecPtrIsValid(remote_lsn))
        PG_RETURN_NULL();
 
    PG_RETURN_LSN(remote_lsn);
}

References ereport, errcode(), errmsg, ERROR, fb(), InvalidXLogRecPtr, PG_GETARG_BOOL, PG_RETURN_LSN, PG_RETURN_NULL, replorigin_check_prerequisites(), replorigin_session_get_progress(), session_replication_state, and XLogRecPtrIsValid.

pg_replication_origin_session_reset()

Datum pg_replication_origin_session_reset

(

PG_FUNCTION_ARGS

)

Definition at line 1503 of file origin.c.

{
    replorigin_check_prerequisites(true, false);
 
    replorigin_session_reset();
 
    replorigin_xact_clear(true);
 
    PG_RETURN_VOID();
}

References PG_RETURN_VOID, replorigin_check_prerequisites(), replorigin_session_reset(), and replorigin_xact_clear().

pg_replication_origin_session_setup()

Datum pg_replication_origin_session_setup

(

PG_FUNCTION_ARGS

)

Definition at line 1479 of file origin.c.

{
    char       *name;
    ReplOriginId origin;
    int         pid;
 
    replorigin_check_prerequisites(true, false);
 
    name = text_to_cstring((text *) DatumGetPointer(PG_GETARG_DATUM(0)));
    origin = replorigin_by_name(name, false);
    pid = PG_GETARG_INT32(1);
    replorigin_session_setup(origin, pid);
 
    replorigin_xact_state.origin = origin;
 
    pfree(name);
 
    PG_RETURN_VOID();
}

References DatumGetPointer(), name, ReplOriginXactState::origin, pfree(), PG_GETARG_DATUM, PG_GETARG_INT32, PG_RETURN_VOID, replorigin_by_name(), replorigin_check_prerequisites(), replorigin_session_setup(), replorigin_xact_state, and text_to_cstring().

pg_replication_origin_xact_reset()

Datum pg_replication_origin_xact_reset

(

PG_FUNCTION_ARGS

)

Definition at line 1573 of file origin.c.

{
    replorigin_check_prerequisites(true, false);
 
    /* Do not clear the session origin */
    replorigin_xact_clear(false);
 
    PG_RETURN_VOID();
}

References PG_RETURN_VOID, replorigin_check_prerequisites(), and replorigin_xact_clear().

pg_replication_origin_xact_setup()

Datum pg_replication_origin_xact_setup

(

PG_FUNCTION_ARGS

)

Definition at line 1555 of file origin.c.

{
    XLogRecPtr  location = PG_GETARG_LSN(0);
 
    replorigin_check_prerequisites(true, false);
 
    if (session_replication_state == NULL)
        ereport(ERROR,
                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                 errmsg("no replication origin is configured")));
 
    replorigin_xact_state.origin_lsn = location;
    replorigin_xact_state.origin_timestamp = PG_GETARG_TIMESTAMPTZ(1);
 
    PG_RETURN_VOID();
}

References ereport, errcode(), errmsg, ERROR, fb(), ReplOriginXactState::origin_lsn, ReplOriginXactState::origin_timestamp, PG_GETARG_LSN, PG_GETARG_TIMESTAMPTZ, PG_RETURN_VOID, replorigin_check_prerequisites(), replorigin_xact_state, and session_replication_state.

pg_show_replication_origin_status()

Datum pg_show_replication_origin_status

(

PG_FUNCTION_ARGS

)

Definition at line 1645 of file origin.c.

{
    ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
    int         i;
#define REPLICATION_ORIGIN_PROGRESS_COLS 4
 
    /* we want to return 0 rows if slot is set to zero */
    replorigin_check_prerequisites(false, true);
 
    InitMaterializedSRF(fcinfo, 0);
 
    /* prevent slots from being concurrently dropped */
    LWLockAcquire(ReplicationOriginLock, LW_SHARED);
 
    /*
     * Iterate through all possible replication_states, display if they are
     * filled. Note that we do not take any locks, so slightly corrupted/out
     * of date values are a possibility.
     */
    for (i = 0; i < max_active_replication_origins; i++)
    {
        ReplicationState *state;
        Datum       values[REPLICATION_ORIGIN_PROGRESS_COLS];
        bool        nulls[REPLICATION_ORIGIN_PROGRESS_COLS];
        char       *roname;
 
        state = &replication_states[i];
 
        /* unused slot, nothing to display */
        if (state->roident == InvalidReplOriginId)
            continue;
 
        memset(values, 0, sizeof(values));
        memset(nulls, 1, sizeof(nulls));
 
        values[0] = ObjectIdGetDatum(state->roident);
        nulls[0] = false;
 
        /*
         * We're not preventing the origin to be dropped concurrently, so
         * silently accept that it might be gone.
         */
        if (replorigin_by_oid(state->roident, true,
                              &roname))
        {
            values[1] = CStringGetTextDatum(roname);
            nulls[1] = false;
        }
 
        LWLockAcquire(&state->lock, LW_SHARED);
 
        values[2] = LSNGetDatum(state->remote_lsn);
        nulls[2] = false;
 
        values[3] = LSNGetDatum(state->local_lsn);
        nulls[3] = false;
 
        LWLockRelease(&state->lock);
 
        tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc,
                             values, nulls);
    }
 
    LWLockRelease(ReplicationOriginLock);
 
#undef REPLICATION_ORIGIN_PROGRESS_COLS
 
    return (Datum) 0;
}

References CStringGetTextDatum, fb(), i, InitMaterializedSRF(), InvalidReplOriginId, LSNGetDatum(), LW_SHARED, LWLockAcquire(), LWLockRelease(), max_active_replication_origins, ObjectIdGetDatum(), REPLICATION_ORIGIN_PROGRESS_COLS, replication_states, replorigin_by_oid(), replorigin_check_prerequisites(), tuplestore_putvalues(), and values.

ReplicationOriginExitCleanup()

static void ReplicationOriginExitCleanup ( int  code, Datum  arg  )

static

Definition at line 1129 of file origin.c.

{
    if (session_replication_state == NULL)
        return;
 
    replorigin_session_reset_internal();
}

References fb(), replorigin_session_reset_internal(), and session_replication_state.

Referenced by replorigin_session_setup().

ReplicationOriginShmemAttach()

static void ReplicationOriginShmemAttach ( void *  arg )

static

Definition at line 589 of file origin.c.

{
    if (max_active_replication_origins == 0)
        return;
 
    replication_states = replication_states_ctl->states;
}

References max_active_replication_origins, replication_states, replication_states_ctl, and ReplicationStateCtl::states.

ReplicationOriginShmemInit()

static void ReplicationOriginShmemInit ( void *  arg )

static

Definition at line 571 of file origin.c.

{
    if (max_active_replication_origins == 0)
        return;
 
    replication_states = replication_states_ctl->states;
 
    replication_states_ctl->tranche_id = LWTRANCHE_REPLICATION_ORIGIN_STATE;
 
    for (int 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(), fb(), i, LWLockInitialize(), max_active_replication_origins, replication_states, replication_states_ctl, ReplicationStateCtl::states, and ReplicationStateCtl::tranche_id.

ReplicationOriginShmemRequest()

static void ReplicationOriginShmemRequest ( void *  arg )

static

Definition at line 554 of file origin.c.

{
    Size        size = 0;
 
    if (max_active_replication_origins == 0)
        return;
 
    size = add_size(size, offsetof(ReplicationStateCtl, states));
    size = add_size(size,
                    mul_size(max_active_replication_origins, sizeof(ReplicationState)));
    ShmemRequestStruct(.name = "ReplicationOriginState",
                       .size = size,
                       .ptr = (void **) &replication_states_ctl,
        );
}

References add_size(), fb(), max_active_replication_origins, mul_size(), name, replication_states_ctl, and ShmemRequestStruct.

replorigin_advance()

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

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
	)

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
	)

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_check_prerequisites()

static void replorigin_check_prerequisites ( bool  check_origins, bool  recoveryOK  )

static

Definition at line 207 of file origin.c.

{
    if (check_origins && max_active_replication_origins == 0)
        ereport(ERROR,
                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                 errmsg("cannot query or manipulate replication origin when \"max_active_replication_origins\" is 0")));
 
    if (!recoveryOK && RecoveryInProgress())
        ereport(ERROR,
                (errcode(ERRCODE_READ_ONLY_SQL_TRANSACTION),
                 errmsg("cannot manipulate replication origins during recovery")));
}

References ereport, errcode(), errmsg, ERROR, fb(), max_active_replication_origins, and RecoveryInProgress().

Referenced by pg_replication_origin_advance(), pg_replication_origin_create(), pg_replication_origin_drop(), pg_replication_origin_oid(), pg_replication_origin_progress(), pg_replication_origin_session_is_setup(), pg_replication_origin_session_progress(), pg_replication_origin_session_reset(), pg_replication_origin_session_setup(), pg_replication_origin_xact_reset(), pg_replication_origin_xact_setup(), and pg_show_replication_origin_status().

replorigin_create()

ReplOriginId replorigin_create

(

const char *

roname

)

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_drop_by_name()

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

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
	)

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_redo()

void replorigin_redo

(

XLogReaderState *

record

)

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
	)

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

)

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

)

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_reset_internal()

static void replorigin_session_reset_internal ( void  )

static

Definition at line 1095 of file origin.c.

{
    ConditionVariable *cv;
 
    Assert(session_replication_state != NULL);
 
    LWLockAcquire(ReplicationOriginLock, LW_EXCLUSIVE);
 
    /* The origin must be held by at least one process at this point. */
    Assert(session_replication_state->refcount > 0);
 
    /*
     * Reset the PID only if the current session is the first to set up this
     * origin. This avoids clearing the first process's PID when any other
     * session releases the origin.
     */
    if (session_replication_state->acquired_by == MyProcPid)
        session_replication_state->acquired_by = 0;
 
    session_replication_state->refcount--;
 
    cv = &session_replication_state->origin_cv;
    session_replication_state = NULL;
 
    LWLockRelease(ReplicationOriginLock);
 
    ConditionVariableBroadcast(cv);
}

References ReplicationState::acquired_by, Assert, ConditionVariableBroadcast(), fb(), LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MyProcPid, ReplicationState::origin_cv, ReplicationState::refcount, and session_replication_state.

Referenced by ReplicationOriginExitCleanup(), and replorigin_session_reset().

replorigin_session_setup()

void replorigin_session_setup	(	ReplOriginId	node,
		int	acquired_by
	)

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_state_clear()

static void replorigin_state_clear ( ReplOriginId  roident, bool  nowait  )

static

Definition at line 386 of file origin.c.

{
    int         i;
 
    /*
     * Clean up the slot state info, if there is any matching slot.
     */
restart:
    LWLockAcquire(ReplicationOriginLock, LW_EXCLUSIVE);
 
    for (i = 0; i < max_active_replication_origins; i++)
    {
        ReplicationState *state = &replication_states[i];
 
        if (state->roident == roident)
        {
            /* found our slot, is it busy? */
            if (state->refcount > 0)
            {
                ConditionVariable *cv;
 
                if (nowait)
                    ereport(ERROR,
                            (errcode(ERRCODE_OBJECT_IN_USE),
                             (state->acquired_by != 0)
                             ? errmsg("could not drop replication origin with ID %d, in use by PID %d",
                                      state->roident,
                                      state->acquired_by)
                             : errmsg("could not drop replication origin with ID %d, in use by another process",
                                      state->roident)));
 
                /*
                 * We must wait and then retry.  Since we don't know which CV
                 * to wait on until here, we can't readily use
                 * ConditionVariablePrepareToSleep (calling it here would be
                 * wrong, since we could miss the signal if we did so); just
                 * use ConditionVariableSleep directly.
                 */
                cv = &state->origin_cv;
 
                LWLockRelease(ReplicationOriginLock);
 
                ConditionVariableSleep(cv, WAIT_EVENT_REPLICATION_ORIGIN_DROP);
                goto restart;
            }
 
            /* first make a WAL log entry */
            {
                xl_replorigin_drop xlrec;
 
                xlrec.node_id = roident;
                XLogBeginInsert();
                XLogRegisterData(&xlrec, sizeof(xlrec));
                XLogInsert(RM_REPLORIGIN_ID, XLOG_REPLORIGIN_DROP);
            }
 
            /* then clear the in-memory slot */
            state->roident = InvalidReplOriginId;
            state->remote_lsn = InvalidXLogRecPtr;
            state->local_lsn = InvalidXLogRecPtr;
            break;
        }
    }
    LWLockRelease(ReplicationOriginLock);
    ConditionVariableCancelSleep();
}

References ConditionVariableCancelSleep(), ConditionVariableSleep(), ereport, errcode(), errmsg, ERROR, fb(), i, InvalidReplOriginId, InvalidXLogRecPtr, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), max_active_replication_origins, xl_replorigin_drop::node_id, replication_states, XLOG_REPLORIGIN_DROP, XLogBeginInsert(), XLogInsert(), and XLogRegisterData().

Referenced by replorigin_drop_by_name().

replorigin_xact_clear()

void replorigin_xact_clear

(

bool

clear_origin

)

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

)

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

int max_active_replication_origins = 10

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().

replication_states

ReplicationState* replication_states

static

Definition at line 178 of file origin.c.

Referenced by CheckPointReplicationOrigin(), pg_show_replication_origin_status(), ReplicationOriginShmemAttach(), ReplicationOriginShmemInit(), replorigin_advance(), replorigin_get_progress(), replorigin_redo(), replorigin_session_setup(), replorigin_state_clear(), and StartupReplicationOrigin().

replication_states_ctl

ReplicationStateCtl* replication_states_ctl

static

Definition at line 193 of file origin.c.

Referenced by ReplicationOriginShmemAttach(), ReplicationOriginShmemInit(), and ReplicationOriginShmemRequest().

ReplicationOriginShmemCallbacks

const ShmemCallbacks ReplicationOriginShmemCallbacks

Initial value:

= {
    .request_fn = ReplicationOriginShmemRequest,
    .init_fn = ReplicationOriginShmemInit,
    .attach_fn = ReplicationOriginShmemAttach,
}

Definition at line 184 of file origin.c.

                                                       {
    .request_fn = ReplicationOriginShmemRequest,
    .init_fn = ReplicationOriginShmemInit,
    .attach_fn = ReplicationOriginShmemAttach,
};

replorigin_xact_state

ReplOriginXactState replorigin_xact_state

Initial value:

= {
    .origin = InvalidReplOriginId,  
    .origin_lsn = InvalidXLogRecPtr,
    .origin_timestamp = 0
}

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().

session_replication_state

ReplicationState* session_replication_state = NULL

static

Definition at line 201 of file origin.c.

Referenced by pg_replication_origin_session_progress(), pg_replication_origin_xact_setup(), ReplicationOriginExitCleanup(), replorigin_session_advance(), replorigin_session_get_progress(), replorigin_session_reset(), replorigin_session_reset_internal(), and replorigin_session_setup().