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

Include dependency graph for origin.h:

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

Go to the source code of this file.

Data Structures
struct	xl_replorigin_set

struct	xl_replorigin_drop

Macros
#define	XLOG_REPLORIGIN_SET 0x00

#define	XLOG_REPLORIGIN_DROP 0x10

#define	InvalidRepOriginId 0

#define	DoNotReplicateId PG_UINT16_MAX

Typedefs
typedef struct xl_replorigin_set	xl_replorigin_set

typedef struct xl_replorigin_drop	xl_replorigin_drop

Functions
RepOriginId	replorigin_by_name (char *name, bool missing_ok)

RepOriginId	replorigin_create (char *name)

void	replorigin_drop (RepOriginId roident)

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

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

XLogRecPtr	replorigin_get_progress (RepOriginId node, bool flush)

void	replorigin_session_advance (XLogRecPtr remote_commit, XLogRecPtr local_commit)

void	replorigin_session_setup (RepOriginId node)

void	replorigin_session_reset (void)

XLogRecPtr	replorigin_session_get_progress (bool flush)

void	CheckPointReplicationOrigin (void)

void	StartupReplicationOrigin (void)

void	replorigin_redo (XLogReaderState *record)

void	replorigin_desc (StringInfo buf, XLogReaderState *record)

const char *	replorigin_identify (uint8 info)

Size	ReplicationOriginShmemSize (void)

void	ReplicationOriginShmemInit (void)

Variables
PGDLLIMPORT RepOriginId	replorigin_session_origin

PGDLLIMPORT XLogRecPtr	replorigin_session_origin_lsn

PGDLLIMPORT TimestampTz	replorigin_session_origin_timestamp

Macro Definition Documentation

#define DoNotReplicateId PG_UINT16_MAX

Definition at line 35 of file origin.h.

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

#define XLOG_REPLORIGIN_DROP 0x10

Definition at line 32 of file origin.h.

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

#define XLOG_REPLORIGIN_SET 0x00

Definition at line 31 of file origin.h.

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

Typedef Documentation

typedef struct xl_replorigin_drop xl_replorigin_drop

typedef struct xl_replorigin_set xl_replorigin_set

Function Documentation

void CheckPointReplicationOrigin ( void )

Definition at line 504 of file origin.c.

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

Referenced by CheckPointGuts().

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

void ReplicationOriginShmemInit ( void )

Definition at line 457 of file origin.c.

References i, LWLockInitialize(), LWLockRegisterTranche(), LWTRANCHE_REPLICATION_ORIGIN, max_replication_slots, MemSet, ReplicationOriginShmemSize(), ShmemInitStruct(), ReplicationStateCtl::states, and ReplicationStateCtl::tranche_id.

Referenced by CreateSharedMemoryAndSemaphores().

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

Size ReplicationOriginShmemSize ( void )

Definition at line 437 of file origin.c.

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

Referenced by CreateSharedMemoryAndSemaphores(), and ReplicationOriginShmemInit().

 {
     Size        size = 0;
 
     /*
      * XXX: max_replication_slots is arguably the wrong thing to use, as here
      * we keep the replay state of *remote* transactions. But for now it seems
      * sufficient to reuse it, lest we introduce a separate GUC.
      */
     if (max_replication_slots == 0)
         return size;
 
     size = add_size(size, offsetof(ReplicationStateCtl, states));
 
     size = add_size(size,
                     mul_size(max_replication_slots, sizeof(ReplicationState)));
     return size;
 }

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

Definition at line 794 of file origin.c.

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

Referenced by pg_replication_origin_advance(), replorigin_redo(), and xact_redo_commit().

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

RepOriginId replorigin_by_name	(	char *	name,
		bool	missing_ok
	)

Definition at line 206 of file origin.c.

References CStringGetTextDatum, elog, ERROR, GETSTRUCT, HeapTupleIsValid, InvalidOid, ReleaseSysCache(), REPLORIGNAME, and SearchSysCache1.

Referenced by ApplyWorkerMain(), DropSubscription(), pg_replication_origin_advance(), pg_replication_origin_drop(), pg_replication_origin_oid(), pg_replication_origin_progress(), and pg_replication_origin_session_setup().

 {
     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)
         elog(ERROR, "cache lookup failed for replication origin '%s'",
              roname);
 
     return roident;
 }

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

Definition at line 398 of file origin.c.

References Assert, DoNotReplicateId, elog, ERROR, GETSTRUCT, HeapTupleIsValid, InvalidRepOriginId, NULL, ObjectIdGetDatum, OidIsValid, ReleaseSysCache(), REPLORIGIDENT, SearchSysCache1, and text_to_cstring().

Referenced by pg_show_replication_origin_status(), and pgoutput_begin_txn().

 {
     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)
             elog(ERROR, "cache lookup failed for replication origin with oid %u",
                  roident);
 
         return false;
     }
 }

RepOriginId replorigin_create ( char * name )

Definition at line 235 of file origin.c.

References Anum_pg_replication_origin_roident, Anum_pg_replication_origin_roname, Assert, BTEqualStrategyNumber, CatalogTupleInsert(), CHECK_FOR_INTERRUPTS, CommandCounterIncrement(), CStringGetTextDatum, ereport, errcode(), errmsg(), ERROR, ExclusiveLock, heap_close, heap_form_tuple(), heap_freetuple(), heap_open(), HeapTupleIsValid, InitDirtySnapshot, InvalidOid, IsTransactionState(), Natts_pg_replication_origin, NULL, ObjectIdGetDatum, PG_UINT16_MAX, RelationGetDescr, ReplicationOriginIdentIndex, ReplicationOriginRelationId, ScanKeyInit(), systable_beginscan(), systable_endscan(), systable_getnext(), and values.

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

 {
     Oid         roident;
     HeapTuple   tuple = NULL;
     Relation    rel;
     Datum       roname_d;
     SnapshotData SnapshotDirty;
     SysScanDesc scan;
     ScanKeyData key;
 
     roname_d = CStringGetTextDatum(roname);
 
     Assert(IsTransactionState());
 
     /*
      * We need the numeric replication origin to be 16bit wide, so we cannot
      * rely on the normal oid allocation. Instead we simply scan
      * pg_replication_origin for the first unused id. That's not particularly
      * efficient, but this should be a fairly infrequent operation - we can
      * easily spend a bit more code on this when it turns out it needs to be
      * faster.
      *
      * We handle concurrency by taking an exclusive lock (allowing reads!)
      * over the table for the duration of the search. Because we use a "dirty
      * snapshot" we can read rows that other in-progress sessions have
      * written, even though they would be invisible with normal snapshots. Due
      * to the exclusive lock there's no danger that new rows can appear while
      * we're checking.
      */
     InitDirtySnapshot(SnapshotDirty);
 
     rel = heap_open(ReplicationOriginRelationId, ExclusiveLock);
 
     for (roident = InvalidOid + 1; roident < PG_UINT16_MAX; roident++)
     {
         bool        nulls[Natts_pg_replication_origin];
         Datum       values[Natts_pg_replication_origin];
         bool        collides;
 
         CHECK_FOR_INTERRUPTS();
 
         ScanKeyInit(&key,
                     Anum_pg_replication_origin_roident,
                     BTEqualStrategyNumber, F_OIDEQ,
                     ObjectIdGetDatum(roident));
 
         scan = systable_beginscan(rel, ReplicationOriginIdentIndex,
                                   true /* indexOK */ ,
                                   &SnapshotDirty,
                                   1, &key);
 
         collides = HeapTupleIsValid(systable_getnext(scan));
 
         systable_endscan(scan);
 
         if (!collides)
         {
             /*
              * Ok, found an unused roident, insert the new row and do a CCI,
              * so our callers can look it up if they want to.
              */
             memset(&nulls, 0, sizeof(nulls));
 
             values[Anum_pg_replication_origin_roident - 1] = ObjectIdGetDatum(roident);
             values[Anum_pg_replication_origin_roname - 1] = roname_d;
 
             tuple = heap_form_tuple(RelationGetDescr(rel), values, nulls);
             CatalogTupleInsert(rel, tuple);
             CommandCounterIncrement();
             break;
         }
     }
 
     /* now release lock again,  */
     heap_close(rel, ExclusiveLock);
 
     if (tuple == NULL)
         ereport(ERROR,
                 (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
                  errmsg("could not find free replication origin OID")));
 
     heap_freetuple(tuple);
     return roident;
 }

void replorigin_desc	(	StringInfo	buf,
		XLogReaderState *	record
	)

Definition at line 19 of file replorigindesc.c.

References appendStringInfo(), xl_replorigin_set::force, xl_replorigin_set::node_id, xl_replorigin_drop::node_id, xl_replorigin_set::remote_lsn, XLOG_REPLORIGIN_DROP, XLOG_REPLORIGIN_SET, XLogRecGetData, XLogRecGetInfo, and XLR_INFO_MASK.

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

void replorigin_drop ( RepOriginId roident )

Definition at line 327 of file origin.c.

References ReplicationState::acquired_by, Assert, CatalogTupleDelete(), CommandCounterIncrement(), elog, ereport, errcode(), errmsg(), ERROR, ExclusiveLock, heap_close, heap_open(), HeapTupleIsValid, i, InvalidRepOriginId, InvalidXLogRecPtr, IsTransactionState(), ReplicationState::local_lsn, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), max_replication_slots, xl_replorigin_drop::node_id, NULL, ObjectIdGetDatum, ReleaseSysCache(), ReplicationState::remote_lsn, ReplicationOriginRelationId, REPLORIGIDENT, ReplicationState::roident, SearchSysCache1, HeapTupleData::t_self, XLOG_REPLORIGIN_DROP, XLogBeginInsert(), XLogInsert(), and XLogRegisterData().

Referenced by DropSubscription(), and pg_replication_origin_drop().

 {
     HeapTuple   tuple = NULL;
     Relation    rel;
     int         i;
 
     Assert(IsTransactionState());
 
     rel = heap_open(ReplicationOriginRelationId, ExclusiveLock);
 
     /* cleanup the slot state info */
     LWLockAcquire(ReplicationOriginLock, LW_EXCLUSIVE);
 
     for (i = 0; i < max_replication_slots; i++)
     {
         ReplicationState *state = &replication_states[i];
 
         /* found our slot */
         if (state->roident == roident)
         {
             if (state->acquired_by != 0)
             {
                 ereport(ERROR,
                         (errcode(ERRCODE_OBJECT_IN_USE),
                          errmsg("could not drop replication origin with OID %d, in use by PID %d",
                                 state->roident,
                                 state->acquired_by)));
             }
 
             /* first WAL log */
             {
                 xl_replorigin_drop xlrec;
 
                 xlrec.node_id = roident;
                 XLogBeginInsert();
                 XLogRegisterData((char *) (&xlrec), sizeof(xlrec));
                 XLogInsert(RM_REPLORIGIN_ID, XLOG_REPLORIGIN_DROP);
             }
 
             /* then reset the in-memory entry */
             state->roident = InvalidRepOriginId;
             state->remote_lsn = InvalidXLogRecPtr;
             state->local_lsn = InvalidXLogRecPtr;
             break;
         }
     }
     LWLockRelease(ReplicationOriginLock);
 
     tuple = SearchSysCache1(REPLORIGIDENT, ObjectIdGetDatum(roident));
     if (!HeapTupleIsValid(tuple))
         elog(ERROR, "cache lookup failed for replication origin with oid %u",
              roident);
 
     CatalogTupleDelete(rel, &tuple->t_self);
     ReleaseSysCache(tuple);
 
     CommandCounterIncrement();
 
     /* now release lock again,  */
     heap_close(rel, ExclusiveLock);
 }

XLogRecPtr replorigin_get_progress	(	RepOriginId	node,
		bool	flush
	)

Definition at line 917 of file origin.c.

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

Referenced by pg_replication_origin_progress().

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

const char* replorigin_identify ( uint8 info )

Definition at line 52 of file replorigindesc.c.

References NULL, XLOG_REPLORIGIN_DROP, and XLOG_REPLORIGIN_SET.

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

void replorigin_redo ( XLogReaderState * record )

Definition at line 733 of file origin.c.

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

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

void replorigin_session_advance	(	XLogRecPtr	remote_commit,
		XLogRecPtr	local_commit
	)

Definition at line 1092 of file origin.c.

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

Referenced by RecordTransactionCommit(), and RecordTransactionCommitPrepared().

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

XLogRecPtr replorigin_session_get_progress ( bool flush )

Definition at line 1110 of file origin.c.

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

Referenced by ApplyWorkerMain(), and pg_replication_origin_session_progress().

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

void replorigin_session_reset ( void )

Definition at line 1068 of file origin.c.

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

Referenced by pg_replication_origin_session_reset().

 {
     Assert(max_replication_slots != 0);
 
     if (session_replication_state == NULL)
         ereport(ERROR,
                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                  errmsg("no replication origin is configured")));
 
     LWLockAcquire(ReplicationOriginLock, LW_EXCLUSIVE);
 
     session_replication_state->acquired_by = 0;
     session_replication_state = NULL;
 
     LWLockRelease(ReplicationOriginLock);
 }

void replorigin_session_setup ( RepOriginId node )

Definition at line 983 of file origin.c.

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

Referenced by ApplyWorkerMain(), and pg_replication_origin_session_setup().

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

void StartupReplicationOrigin ( void )

Definition at line 618 of file origin.c.

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

Referenced by StartupXLOG().

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

Variable Documentation

PGDLLIMPORT RepOriginId replorigin_session_origin

Definition at line 150 of file origin.c.

Referenced by ApplyWorkerMain(), pg_replication_origin_session_is_setup(), pg_replication_origin_session_reset(), pg_replication_origin_session_setup(), RecordTransactionCommit(), RecordTransactionCommitPrepared(), XactLogCommitRecord(), and XLogRecordAssemble().

PGDLLIMPORT XLogRecPtr replorigin_session_origin_lsn

Definition at line 151 of file origin.c.

Referenced by apply_handle_commit(), pg_replication_origin_session_reset(), pg_replication_origin_xact_reset(), pg_replication_origin_xact_setup(), RecordTransactionCommit(), RecordTransactionCommitPrepared(), and XactLogCommitRecord().

PGDLLIMPORT TimestampTz replorigin_session_origin_timestamp

Definition at line 152 of file origin.c.

Referenced by apply_handle_commit(), pg_replication_origin_session_reset(), pg_replication_origin_xact_reset(), pg_replication_origin_xact_setup(), RecordTransactionCommit(), RecordTransactionCommitPrepared(), and XactLogCommitRecord().

Data Structures

Macros

Typedefs

Functions

Variables

Macro Definition Documentation

Typedef Documentation

Function Documentation

Variable Documentation