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 "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 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	replorigin_check_prerequisites (bool check_origins, bool recoveryOK)
static bool	IsReservedOriginName (const char *name)
RepOriginId	replorigin_by_name (const char *roname, bool missing_ok)
RepOriginId	replorigin_create (const char *roname)
static void	replorigin_state_clear (RepOriginId roident, bool nowait)
void	replorigin_drop_by_name (const char *name, bool missing_ok, bool nowait)
bool	replorigin_by_oid (RepOriginId roident, bool missing_ok, char **roname)
Size	ReplicationOriginShmemSize (void)
void	ReplicationOriginShmemInit (void)
void	CheckPointReplicationOrigin (void)
void	StartupReplicationOrigin (void)
void	replorigin_redo (XLogReaderState *record)
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)
static void	ReplicationOriginExitCleanup (int code, Datum arg)
void	replorigin_session_setup (RepOriginId 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)
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
RepOriginId	replorigin_session_origin = InvalidRepOriginId
XLogRecPtr	replorigin_session_origin_lsn = InvalidXLogRecPtr
TimestampTz	replorigin_session_origin_timestamp = 0
static ReplicationState *	replication_states
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 100 of file origin.c.

PG_REPLORIGIN_CHECKPOINT_TMPFILE

#define PG_REPLORIGIN_CHECKPOINT_TMPFILE   PG_REPLORIGIN_CHECKPOINT_FILENAME ".tmp"

Definition at line 101 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 187 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 596 of file origin.c.

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

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

Referenced by CheckPointGuts().

IsReservedOriginName()

static bool IsReservedOriginName ( const char *  name )

static

Definition at line 209 of file origin.c.

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

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

{
    text       *name = PG_GETARG_TEXT_PP(0);
    XLogRecPtr  remote_commit = PG_GETARG_LSN(1);
    RepOriginId 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 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 1292 of file origin.c.

{
    char       *name;
    RepOriginId 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, 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 1333 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 1352 of file origin.c.

{
    char       *name;
    RepOriginId 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 1514 of file origin.c.

{
    char       *name;
    bool        flush;
    RepOriginId 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 (remote_lsn == InvalidXLogRecPtr)
        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(), and text_to_cstring().

pg_replication_origin_session_is_setup()

Datum pg_replication_origin_session_is_setup

(

PG_FUNCTION_ARGS

)

Definition at line 1412 of file origin.c.

{
    replorigin_check_prerequisites(false, false);
 
    PG_RETURN_BOOL(replorigin_session_origin != InvalidRepOriginId);
}

References InvalidRepOriginId, PG_RETURN_BOOL, replorigin_check_prerequisites(), and replorigin_session_origin.

pg_replication_origin_session_progress()

Datum pg_replication_origin_session_progress

(

PG_FUNCTION_ARGS

)

Definition at line 1428 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 (remote_lsn == InvalidXLogRecPtr)
        PG_RETURN_NULL();
 
    PG_RETURN_LSN(remote_lsn);
}

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

pg_replication_origin_session_reset()

Datum pg_replication_origin_session_reset

(

PG_FUNCTION_ARGS

)

Definition at line 1395 of file origin.c.

{
    replorigin_check_prerequisites(true, false);
 
    replorigin_session_reset();
 
    replorigin_session_origin = InvalidRepOriginId;
    replorigin_session_origin_lsn = InvalidXLogRecPtr;
    replorigin_session_origin_timestamp = 0;
 
    PG_RETURN_VOID();
}

References InvalidRepOriginId, InvalidXLogRecPtr, PG_RETURN_VOID, replorigin_check_prerequisites(), replorigin_session_origin, replorigin_session_origin_lsn, replorigin_session_origin_timestamp, and replorigin_session_reset().

pg_replication_origin_session_setup()

Datum pg_replication_origin_session_setup

(

PG_FUNCTION_ARGS

)

Definition at line 1373 of file origin.c.

{
    char       *name;
    RepOriginId origin;
 
    replorigin_check_prerequisites(true, false);
 
    name = text_to_cstring((text *) DatumGetPointer(PG_GETARG_DATUM(0)));
    origin = replorigin_by_name(name, false);
    replorigin_session_setup(origin, 0);
 
    replorigin_session_origin = origin;
 
    pfree(name);
 
    PG_RETURN_VOID();
}

References DatumGetPointer(), name, pfree(), PG_GETARG_DATUM, PG_RETURN_VOID, replorigin_by_name(), replorigin_check_prerequisites(), replorigin_session_origin, replorigin_session_setup(), and text_to_cstring().

pg_replication_origin_xact_reset()

Datum pg_replication_origin_xact_reset

(

PG_FUNCTION_ARGS

)

Definition at line 1467 of file origin.c.

{
    replorigin_check_prerequisites(true, false);
 
    replorigin_session_origin_lsn = InvalidXLogRecPtr;
    replorigin_session_origin_timestamp = 0;
 
    PG_RETURN_VOID();
}

References InvalidXLogRecPtr, PG_RETURN_VOID, replorigin_check_prerequisites(), replorigin_session_origin_lsn, and replorigin_session_origin_timestamp.

pg_replication_origin_xact_setup()

Datum pg_replication_origin_xact_setup

(

PG_FUNCTION_ARGS

)

Definition at line 1449 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_session_origin_lsn = location;
    replorigin_session_origin_timestamp = PG_GETARG_TIMESTAMPTZ(1);
 
    PG_RETURN_VOID();
}

References ereport, errcode(), errmsg(), ERROR, PG_GETARG_LSN, PG_GETARG_TIMESTAMPTZ, PG_RETURN_VOID, replorigin_check_prerequisites(), replorigin_session_origin_lsn, replorigin_session_origin_timestamp, and session_replication_state.

pg_show_replication_origin_status()

Datum pg_show_replication_origin_status

(

PG_FUNCTION_ARGS

)

Definition at line 1539 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 == InvalidRepOriginId)
            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, i, InitMaterializedSRF(), InvalidRepOriginId, LSNGetDatum(), LW_SHARED, LWLockAcquire(), LWLockRelease(), max_active_replication_origins, ObjectIdGetDatum(), REPLICATION_ORIGIN_PROGRESS_COLS, replication_states, replorigin_by_oid(), replorigin_check_prerequisites(), ReturnSetInfo::setDesc, ReturnSetInfo::setResult, tuplestore_putvalues(), and values.

ReplicationOriginExitCleanup()

static void ReplicationOriginExitCleanup ( int  code, Datum  arg  )

static

Definition at line 1078 of file origin.c.

{
    ConditionVariable *cv = NULL;
 
    if (session_replication_state == NULL)
        return;
 
    LWLockAcquire(ReplicationOriginLock, LW_EXCLUSIVE);
 
    if (session_replication_state->acquired_by == MyProcPid)
    {
        cv = &session_replication_state->origin_cv;
 
        session_replication_state->acquired_by = 0;
        session_replication_state = NULL;
    }
 
    LWLockRelease(ReplicationOriginLock);
 
    if (cv)
        ConditionVariableBroadcast(cv);
}

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

Referenced by replorigin_session_setup().

ReplicationOriginShmemInit()

void ReplicationOriginShmemInit

(

void

)

Definition at line 549 of file origin.c.

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

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

Referenced by CreateOrAttachShmemStructs().

ReplicationOriginShmemSize()

Size ReplicationOriginShmemSize

(

void

)

Definition at line 534 of file origin.c.

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

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

Referenced by CalculateShmemSize(), and ReplicationOriginShmemInit().

replorigin_advance()

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

Definition at line 911 of file origin.c.

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

References ReplicationState::acquired_by, Assert(), DoNotReplicateId, ereport, errcode(), errhint(), errmsg(), ERROR, xl_replorigin_set::force, i, InvalidRepOriginId, InvalidXLogRecPtr, ReplicationState::local_lsn, ReplicationState::lock, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), max_active_replication_origins, xl_replorigin_set::node_id, ReplicationState::remote_lsn, xl_replorigin_set::remote_lsn, replication_states, ReplicationState::roident, XLOG_REPLORIGIN_SET, XLogBeginInsert(), XLogInsert(), and XLogRegisterData().

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

replorigin_by_name()

RepOriginId replorigin_by_name	(	const char *	roname,
		bool	missing_ok
	)

Definition at line 226 of file origin.c.

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

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

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

replorigin_by_oid()

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

Definition at line 493 of file origin.c.

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

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

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

replorigin_check_prerequisites()

static void replorigin_check_prerequisites ( bool  check_origins, bool  recoveryOK  )

static

Definition at line 190 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, 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()

RepOriginId replorigin_create

(

const char *

roname

)

Definition at line 257 of file origin.c.

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

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

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

replorigin_drop_by_name()

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

Definition at line 439 of file origin.c.

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

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

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

replorigin_get_progress()

XLogRecPtr replorigin_get_progress	(	RepOriginId	node,
		bool	flush
	)

Definition at line 1037 of file origin.c.

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

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

Referenced by AlterSubscription(), and pg_replication_origin_progress().

replorigin_redo()

void replorigin_redo

(

XLogReaderState *

record

)

Definition at line 850 of file origin.c.

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

References elog, XLogReaderState::EndRecPtr, xl_replorigin_set::force, i, InvalidRepOriginId, InvalidXLogRecPtr, max_active_replication_origins, xl_replorigin_set::node_id, xl_replorigin_drop::node_id, PANIC, xl_replorigin_set::remote_lsn, replication_states, replorigin_advance(), XLOG_REPLORIGIN_DROP, XLOG_REPLORIGIN_SET, XLogRecGetData, and XLogRecGetInfo.

replorigin_session_advance()

void replorigin_session_advance	(	XLogRecPtr	remote_commit,
		XLogRecPtr	local_commit
	)

Definition at line 1242 of file origin.c.

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

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

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

replorigin_session_get_progress()

XLogRecPtr replorigin_session_get_progress

(

bool

flush

)

Definition at line 1260 of file origin.c.

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

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

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

replorigin_session_reset()

void replorigin_session_reset

(

void

)

Definition at line 1213 of file origin.c.

{
    ConditionVariable *cv;
 
    Assert(max_active_replication_origins != 0);
 
    if (session_replication_state == NULL)
        ereport(ERROR,
                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                 errmsg("no replication origin is configured")));
 
    LWLockAcquire(ReplicationOriginLock, LW_EXCLUSIVE);
 
    session_replication_state->acquired_by = 0;
    cv = &session_replication_state->origin_cv;
    session_replication_state = NULL;
 
    LWLockRelease(ReplicationOriginLock);
 
    ConditionVariableBroadcast(cv);
}

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

Referenced by pg_replication_origin_session_reset(), and process_syncing_tables_for_sync().

replorigin_session_setup()

void replorigin_session_setup	(	RepOriginId	node,
		int	acquired_by
	)

Definition at line 1120 of file origin.c.

{
    static bool registered_cleanup;
    int         i;
    int         free_slot = -1;
 
    if (!registered_cleanup)
    {
        on_shmem_exit(ReplicationOriginExitCleanup, 0);
        registered_cleanup = true;
    }
 
    Assert(max_active_replication_origins > 0);
 
    if (session_replication_state != NULL)
        ereport(ERROR,
                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                 errmsg("cannot setup replication origin when one is already setup")));
 
    /* Lock exclusively, as we may have to create a new table entry. */
    LWLockAcquire(ReplicationOriginLock, LW_EXCLUSIVE);
 
    /*
     * Search for either an existing slot for the origin, or a free one we can
     * use.
     */
    for (i = 0; i < max_active_replication_origins; i++)
    {
        ReplicationState *curstate = &replication_states[i];
 
        /* remember where to insert if necessary */
        if (curstate->roident == InvalidRepOriginId &&
            free_slot == -1)
        {
            free_slot = i;
            continue;
        }
 
        /* not our slot */
        if (curstate->roident != node)
            continue;
 
        else if (curstate->acquired_by != 0 && acquired_by == 0)
        {
            ereport(ERROR,
                    (errcode(ERRCODE_OBJECT_IN_USE),
                     errmsg("replication origin with ID %d is already active for PID %d",
                            curstate->roident, curstate->acquired_by)));
        }
 
        /* ok, found slot */
        session_replication_state = curstate;
        break;
    }
 
 
    if (session_replication_state == NULL && free_slot == -1)
        ereport(ERROR,
                (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
                 errmsg("could not find free replication state slot for replication origin with ID %d",
                        node),
                 errhint("Increase \"max_active_replication_origins\" and try again.")));
    else if (session_replication_state == NULL)
    {
        /* initialize new slot */
        session_replication_state = &replication_states[free_slot];
        Assert(session_replication_state->remote_lsn == InvalidXLogRecPtr);
        Assert(session_replication_state->local_lsn == InvalidXLogRecPtr);
        session_replication_state->roident = node;
    }
 
 
    Assert(session_replication_state->roident != InvalidRepOriginId);
 
    if (acquired_by == 0)
        session_replication_state->acquired_by = MyProcPid;
    else if (session_replication_state->acquired_by != acquired_by)
        elog(ERROR, "could not find replication state slot for replication origin with OID %u which was acquired by %d",
             node, acquired_by);
 
    LWLockRelease(ReplicationOriginLock);
 
    /* probably this one is pointless */
    ConditionVariableBroadcast(&session_replication_state->origin_cv);
}

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

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

replorigin_state_clear()

static void replorigin_state_clear ( RepOriginId  roident, bool  nowait  )

static

Definition at line 369 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->acquired_by != 0)
            {
                ConditionVariable *cv;
 
                if (nowait)
                    ereport(ERROR,
                            (errcode(ERRCODE_OBJECT_IN_USE),
                             errmsg("could not drop replication origin with ID %d, in use by PID %d",
                                    state->roident,
                                    state->acquired_by)));
 
                /*
                 * 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 = InvalidRepOriginId;
            state->remote_lsn = InvalidXLogRecPtr;
            state->local_lsn = InvalidXLogRecPtr;
            break;
        }
    }
    LWLockRelease(ReplicationOriginLock);
    ConditionVariableCancelSleep();
}

References ConditionVariableCancelSleep(), ConditionVariableSleep(), ereport, errcode(), errmsg(), ERROR, i, InvalidRepOriginId, 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().

StartupReplicationOrigin()

void StartupReplicationOrigin

(

void

)

Definition at line 722 of file origin.c.

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

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

Referenced by StartupXLOG().

Variable Documentation

max_active_replication_origins

int max_active_replication_origins = 10

Definition at line 104 of file origin.c.

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

replication_states

ReplicationState* replication_states

static

Definition at line 171 of file origin.c.

Referenced by CheckPointReplicationOrigin(), pg_show_replication_origin_status(), 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 176 of file origin.c.

Referenced by ReplicationOriginShmemInit().

replorigin_session_origin

RepOriginId replorigin_session_origin = InvalidRepOriginId

Definition at line 163 of file origin.c.

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

replorigin_session_origin_lsn

XLogRecPtr replorigin_session_origin_lsn = InvalidXLogRecPtr

Definition at line 164 of file origin.c.

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

replorigin_session_origin_timestamp

TimestampTz replorigin_session_origin_timestamp = 0

Definition at line 165 of file origin.c.

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

session_replication_state

ReplicationState* session_replication_state = NULL

static

Definition at line 184 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(), and replorigin_session_setup().