LCOV - code coverage report
Current view: top level - src/backend/replication/logical - origin.c (source / functions) Hit Total Coverage
Test: PostgreSQL 13beta1 Lines: 319 457 69.8 %
Date: 2020-05-31 23:07:13 Functions: 24 29 82.8 %
Legend: Lines: hit not hit

          Line data    Source code
       1             : /*-------------------------------------------------------------------------
       2             :  *
       3             :  * origin.c
       4             :  *    Logical replication progress tracking support.
       5             :  *
       6             :  * Copyright (c) 2013-2020, PostgreSQL Global Development Group
       7             :  *
       8             :  * IDENTIFICATION
       9             :  *    src/backend/replication/logical/origin.c
      10             :  *
      11             :  * NOTES
      12             :  *
      13             :  * This file provides the following:
      14             :  * * An infrastructure to name nodes in a replication setup
      15             :  * * A facility to efficiently store and persist replication progress in an
      16             :  *   efficient and durable manner.
      17             :  *
      18             :  * Replication origin consist out of a descriptive, user defined, external
      19             :  * name and a short, thus space efficient, internal 2 byte one. This split
      20             :  * exists because replication origin have to be stored in WAL and shared
      21             :  * memory and long descriptors would be inefficient.  For now only use 2 bytes
      22             :  * for the internal id of a replication origin as it seems unlikely that there
      23             :  * soon will be more than 65k nodes in one replication setup; and using only
      24             :  * two bytes allow us to be more space efficient.
      25             :  *
      26             :  * Replication progress is tracked in a shared memory table
      27             :  * (ReplicationState) that's dumped to disk every checkpoint. Entries
      28             :  * ('slots') in this table are identified by the internal id. That's the case
      29             :  * because it allows to increase replication progress during crash
      30             :  * recovery. To allow doing so we store the original LSN (from the originating
      31             :  * system) of a transaction in the commit record. That allows to recover the
      32             :  * precise replayed state after crash recovery; without requiring synchronous
      33             :  * commits. Allowing logical replication to use asynchronous commit is
      34             :  * generally good for performance, but especially important as it allows a
      35             :  * single threaded replay process to keep up with a source that has multiple
      36             :  * backends generating changes concurrently.  For efficiency and simplicity
      37             :  * reasons a backend can setup one replication origin that's from then used as
      38             :  * the source of changes produced by the backend, until reset again.
      39             :  *
      40             :  * This infrastructure is intended to be used in cooperation with logical
      41             :  * decoding. When replaying from a remote system the configured origin is
      42             :  * provided to output plugins, allowing prevention of replication loops and
      43             :  * other filtering.
      44             :  *
      45             :  * There are several levels of locking at work:
      46             :  *
      47             :  * * To create and drop replication origins an exclusive lock on
      48             :  *   pg_replication_slot is required for the duration. That allows us to
      49             :  *   safely and conflict free assign new origins using a dirty snapshot.
      50             :  *
      51             :  * * When creating an in-memory replication progress slot the ReplicationOrigin
      52             :  *   LWLock has to be held exclusively; when iterating over the replication
      53             :  *   progress a shared lock has to be held, the same when advancing the
      54             :  *   replication progress of an individual backend that has not setup as the
      55             :  *   session's replication origin.
      56             :  *
      57             :  * * When manipulating or looking at the remote_lsn and local_lsn fields of a
      58             :  *   replication progress slot that slot's lwlock has to be held. That's
      59             :  *   primarily because we do not assume 8 byte writes (the LSN) is atomic on
      60             :  *   all our platforms, but it also simplifies memory ordering concerns
      61             :  *   between the remote and local lsn. We use a lwlock instead of a spinlock
      62             :  *   so it's less harmful to hold the lock over a WAL write
      63             :  *   (cf. AdvanceReplicationProgress).
      64             :  *
      65             :  * ---------------------------------------------------------------------------
      66             :  */
      67             : 
      68             : #include "postgres.h"
      69             : 
      70             : #include <unistd.h>
      71             : #include <sys/stat.h>
      72             : 
      73             : #include "access/genam.h"
      74             : #include "access/htup_details.h"
      75             : #include "access/table.h"
      76             : #include "access/xact.h"
      77             : #include "catalog/catalog.h"
      78             : #include "catalog/indexing.h"
      79             : #include "funcapi.h"
      80             : #include "miscadmin.h"
      81             : #include "nodes/execnodes.h"
      82             : #include "pgstat.h"
      83             : #include "replication/logical.h"
      84             : #include "replication/origin.h"
      85             : #include "storage/condition_variable.h"
      86             : #include "storage/copydir.h"
      87             : #include "storage/fd.h"
      88             : #include "storage/ipc.h"
      89             : #include "storage/lmgr.h"
      90             : #include "utils/builtins.h"
      91             : #include "utils/fmgroids.h"
      92             : #include "utils/pg_lsn.h"
      93             : #include "utils/rel.h"
      94             : #include "utils/snapmgr.h"
      95             : #include "utils/syscache.h"
      96             : 
      97             : /*
      98             :  * Replay progress of a single remote node.
      99             :  */
     100             : typedef struct ReplicationState
     101             : {
     102             :     /*
     103             :      * Local identifier for the remote node.
     104             :      */
     105             :     RepOriginId roident;
     106             : 
     107             :     /*
     108             :      * Location of the latest commit from the remote side.
     109             :      */
     110             :     XLogRecPtr  remote_lsn;
     111             : 
     112             :     /*
     113             :      * Remember the local lsn of the commit record so we can XLogFlush() to it
     114             :      * during a checkpoint so we know the commit record actually is safe on
     115             :      * disk.
     116             :      */
     117             :     XLogRecPtr  local_lsn;
     118             : 
     119             :     /*
     120             :      * PID of backend that's acquired slot, or 0 if none.
     121             :      */
     122             :     int         acquired_by;
     123             : 
     124             :     /*
     125             :      * Condition variable that's signalled when acquired_by changes.
     126             :      */
     127             :     ConditionVariable origin_cv;
     128             : 
     129             :     /*
     130             :      * Lock protecting remote_lsn and local_lsn.
     131             :      */
     132             :     LWLock      lock;
     133             : } ReplicationState;
     134             : 
     135             : /*
     136             :  * On disk version of ReplicationState.
     137             :  */
     138             : typedef struct ReplicationStateOnDisk
     139             : {
     140             :     RepOriginId roident;
     141             :     XLogRecPtr  remote_lsn;
     142             : } ReplicationStateOnDisk;
     143             : 
     144             : 
     145             : typedef struct ReplicationStateCtl
     146             : {
     147             :     /* Tranche to use for per-origin LWLocks */
     148             :     int         tranche_id;
     149             :     /* Array of length max_replication_slots */
     150             :     ReplicationState states[FLEXIBLE_ARRAY_MEMBER];
     151             : } ReplicationStateCtl;
     152             : 
     153             : /* external variables */
     154             : RepOriginId replorigin_session_origin = InvalidRepOriginId; /* assumed identity */
     155             : XLogRecPtr  replorigin_session_origin_lsn = InvalidXLogRecPtr;
     156             : TimestampTz replorigin_session_origin_timestamp = 0;
     157             : 
     158             : /*
     159             :  * Base address into a shared memory array of replication states of size
     160             :  * max_replication_slots.
     161             :  *
     162             :  * XXX: Should we use a separate variable to size this rather than
     163             :  * max_replication_slots?
     164             :  */
     165             : static ReplicationState *replication_states;
     166             : 
     167             : /*
     168             :  * Actual shared memory block (replication_states[] is now part of this).
     169             :  */
     170             : static ReplicationStateCtl *replication_states_ctl;
     171             : 
     172             : /*
     173             :  * Backend-local, cached element from ReplicationState for use in a backend
     174             :  * replaying remote commits, so we don't have to search ReplicationState for
     175             :  * the backends current RepOriginId.
     176             :  */
     177             : static ReplicationState *session_replication_state = NULL;
     178             : 
     179             : /* Magic for on disk files. */
     180             : #define REPLICATION_STATE_MAGIC ((uint32) 0x1257DADE)
     181             : 
     182             : static void
     183          38 : replorigin_check_prerequisites(bool check_slots, bool recoveryOK)
     184             : {
     185          38 :     if (!superuser())
     186           0 :         ereport(ERROR,
     187             :                 (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
     188             :                  errmsg("only superusers can query or manipulate replication origins")));
     189             : 
     190          38 :     if (check_slots && max_replication_slots == 0)
     191           0 :         ereport(ERROR,
     192             :                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
     193             :                  errmsg("cannot query or manipulate replication origin when max_replication_slots = 0")));
     194             : 
     195          38 :     if (!recoveryOK && RecoveryInProgress())
     196           0 :         ereport(ERROR,
     197             :                 (errcode(ERRCODE_READ_ONLY_SQL_TRANSACTION),
     198             :                  errmsg("cannot manipulate replication origins during recovery")));
     199             : 
     200          38 : }
     201             : 
     202             : 
     203             : /* ---------------------------------------------------------------------------
     204             :  * Functions for working with replication origins themselves.
     205             :  * ---------------------------------------------------------------------------
     206             :  */
     207             : 
     208             : /*
     209             :  * Check for a persistent replication origin identified by name.
     210             :  *
     211             :  * Returns InvalidOid if the node isn't known yet and missing_ok is true.
     212             :  */
     213             : RepOriginId
     214         106 : replorigin_by_name(char *roname, bool missing_ok)
     215             : {
     216             :     Form_pg_replication_origin ident;
     217         106 :     Oid         roident = InvalidOid;
     218             :     HeapTuple   tuple;
     219             :     Datum       roname_d;
     220             : 
     221         106 :     roname_d = CStringGetTextDatum(roname);
     222             : 
     223         106 :     tuple = SearchSysCache1(REPLORIGNAME, roname_d);
     224         106 :     if (HeapTupleIsValid(tuple))
     225             :     {
     226          98 :         ident = (Form_pg_replication_origin) GETSTRUCT(tuple);
     227          98 :         roident = ident->roident;
     228          98 :         ReleaseSysCache(tuple);
     229             :     }
     230           8 :     else if (!missing_ok)
     231           8 :         ereport(ERROR,
     232             :                 (errcode(ERRCODE_UNDEFINED_OBJECT),
     233             :                  errmsg("replication origin \"%s\" does not exist",
     234             :                         roname)));
     235             : 
     236          98 :     return roident;
     237             : }
     238             : 
     239             : /*
     240             :  * Create a replication origin.
     241             :  *
     242             :  * Needs to be called in a transaction.
     243             :  */
     244             : RepOriginId
     245          62 : replorigin_create(char *roname)
     246             : {
     247             :     Oid         roident;
     248          62 :     HeapTuple   tuple = NULL;
     249             :     Relation    rel;
     250             :     Datum       roname_d;
     251             :     SnapshotData SnapshotDirty;
     252             :     SysScanDesc scan;
     253             :     ScanKeyData key;
     254             : 
     255          62 :     roname_d = CStringGetTextDatum(roname);
     256             : 
     257             :     Assert(IsTransactionState());
     258             : 
     259             :     /*
     260             :      * We need the numeric replication origin to be 16bit wide, so we cannot
     261             :      * rely on the normal oid allocation. Instead we simply scan
     262             :      * pg_replication_origin for the first unused id. That's not particularly
     263             :      * efficient, but this should be a fairly infrequent operation - we can
     264             :      * easily spend a bit more code on this when it turns out it needs to be
     265             :      * faster.
     266             :      *
     267             :      * We handle concurrency by taking an exclusive lock (allowing reads!)
     268             :      * over the table for the duration of the search. Because we use a "dirty
     269             :      * snapshot" we can read rows that other in-progress sessions have
     270             :      * written, even though they would be invisible with normal snapshots. Due
     271             :      * to the exclusive lock there's no danger that new rows can appear while
     272             :      * we're checking.
     273             :      */
     274          62 :     InitDirtySnapshot(SnapshotDirty);
     275             : 
     276          62 :     rel = table_open(ReplicationOriginRelationId, ExclusiveLock);
     277             : 
     278          78 :     for (roident = InvalidOid + 1; roident < PG_UINT16_MAX; roident++)
     279             :     {
     280             :         bool        nulls[Natts_pg_replication_origin];
     281             :         Datum       values[Natts_pg_replication_origin];
     282             :         bool        collides;
     283             : 
     284          78 :         CHECK_FOR_INTERRUPTS();
     285             : 
     286          78 :         ScanKeyInit(&key,
     287             :                     Anum_pg_replication_origin_roident,
     288             :                     BTEqualStrategyNumber, F_OIDEQ,
     289             :                     ObjectIdGetDatum(roident));
     290             : 
     291          78 :         scan = systable_beginscan(rel, ReplicationOriginIdentIndex,
     292             :                                   true /* indexOK */ ,
     293             :                                   &SnapshotDirty,
     294             :                                   1, &key);
     295             : 
     296          78 :         collides = HeapTupleIsValid(systable_getnext(scan));
     297             : 
     298          78 :         systable_endscan(scan);
     299             : 
     300          78 :         if (!collides)
     301             :         {
     302             :             /*
     303             :              * Ok, found an unused roident, insert the new row and do a CCI,
     304             :              * so our callers can look it up if they want to.
     305             :              */
     306          62 :             memset(&nulls, 0, sizeof(nulls));
     307             : 
     308          62 :             values[Anum_pg_replication_origin_roident - 1] = ObjectIdGetDatum(roident);
     309          62 :             values[Anum_pg_replication_origin_roname - 1] = roname_d;
     310             : 
     311          62 :             tuple = heap_form_tuple(RelationGetDescr(rel), values, nulls);
     312          62 :             CatalogTupleInsert(rel, tuple);
     313          60 :             CommandCounterIncrement();
     314          60 :             break;
     315             :         }
     316             :     }
     317             : 
     318             :     /* now release lock again,  */
     319          60 :     table_close(rel, ExclusiveLock);
     320             : 
     321          60 :     if (tuple == NULL)
     322           0 :         ereport(ERROR,
     323             :                 (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
     324             :                  errmsg("could not find free replication origin OID")));
     325             : 
     326          60 :     heap_freetuple(tuple);
     327          60 :     return roident;
     328             : }
     329             : 
     330             : 
     331             : /*
     332             :  * Drop replication origin.
     333             :  *
     334             :  * Needs to be called in a transaction.
     335             :  */
     336             : void
     337          34 : replorigin_drop(RepOriginId roident, bool nowait)
     338             : {
     339             :     HeapTuple   tuple;
     340             :     Relation    rel;
     341             :     int         i;
     342             : 
     343             :     Assert(IsTransactionState());
     344             : 
     345             :     /*
     346             :      * To interlock against concurrent drops, we hold ExclusiveLock on
     347             :      * pg_replication_origin throughout this function.
     348             :      */
     349          34 :     rel = table_open(ReplicationOriginRelationId, ExclusiveLock);
     350             : 
     351             :     /*
     352             :      * First, clean up the slot state info, if there is any matching slot.
     353             :      */
     354          34 : restart:
     355          34 :     tuple = NULL;
     356          34 :     LWLockAcquire(ReplicationOriginLock, LW_EXCLUSIVE);
     357             : 
     358         186 :     for (i = 0; i < max_replication_slots; i++)
     359             :     {
     360         170 :         ReplicationState *state = &replication_states[i];
     361             : 
     362         170 :         if (state->roident == roident)
     363             :         {
     364             :             /* found our slot, is it busy? */
     365          18 :             if (state->acquired_by != 0)
     366             :             {
     367             :                 ConditionVariable *cv;
     368             : 
     369           0 :                 if (nowait)
     370           0 :                     ereport(ERROR,
     371             :                             (errcode(ERRCODE_OBJECT_IN_USE),
     372             :                              errmsg("could not drop replication origin with OID %d, in use by PID %d",
     373             :                                     state->roident,
     374             :                                     state->acquired_by)));
     375             : 
     376             :                 /*
     377             :                  * We must wait and then retry.  Since we don't know which CV
     378             :                  * to wait on until here, we can't readily use
     379             :                  * ConditionVariablePrepareToSleep (calling it here would be
     380             :                  * wrong, since we could miss the signal if we did so); just
     381             :                  * use ConditionVariableSleep directly.
     382             :                  */
     383           0 :                 cv = &state->origin_cv;
     384             : 
     385           0 :                 LWLockRelease(ReplicationOriginLock);
     386             : 
     387           0 :                 ConditionVariableSleep(cv, WAIT_EVENT_REPLICATION_ORIGIN_DROP);
     388           0 :                 goto restart;
     389             :             }
     390             : 
     391             :             /* first make a WAL log entry */
     392             :             {
     393             :                 xl_replorigin_drop xlrec;
     394             : 
     395          18 :                 xlrec.node_id = roident;
     396          18 :                 XLogBeginInsert();
     397          18 :                 XLogRegisterData((char *) (&xlrec), sizeof(xlrec));
     398          18 :                 XLogInsert(RM_REPLORIGIN_ID, XLOG_REPLORIGIN_DROP);
     399             :             }
     400             : 
     401             :             /* then clear the in-memory slot */
     402          18 :             state->roident = InvalidRepOriginId;
     403          18 :             state->remote_lsn = InvalidXLogRecPtr;
     404          18 :             state->local_lsn = InvalidXLogRecPtr;
     405          18 :             break;
     406             :         }
     407             :     }
     408          34 :     LWLockRelease(ReplicationOriginLock);
     409          34 :     ConditionVariableCancelSleep();
     410             : 
     411             :     /*
     412             :      * Now, we can delete the catalog entry.
     413             :      */
     414          34 :     tuple = SearchSysCache1(REPLORIGIDENT, ObjectIdGetDatum(roident));
     415          34 :     if (!HeapTupleIsValid(tuple))
     416           0 :         elog(ERROR, "cache lookup failed for replication origin with oid %u",
     417             :              roident);
     418             : 
     419          34 :     CatalogTupleDelete(rel, &tuple->t_self);
     420          34 :     ReleaseSysCache(tuple);
     421             : 
     422          34 :     CommandCounterIncrement();
     423             : 
     424             :     /* now release lock again */
     425          34 :     table_close(rel, ExclusiveLock);
     426          34 : }
     427             : 
     428             : 
     429             : /*
     430             :  * Lookup replication origin via it's oid and return the name.
     431             :  *
     432             :  * The external name is palloc'd in the calling context.
     433             :  *
     434             :  * Returns true if the origin is known, false otherwise.
     435             :  */
     436             : bool
     437           2 : replorigin_by_oid(RepOriginId roident, bool missing_ok, char **roname)
     438             : {
     439             :     HeapTuple   tuple;
     440             :     Form_pg_replication_origin ric;
     441             : 
     442             :     Assert(OidIsValid((Oid) roident));
     443             :     Assert(roident != InvalidRepOriginId);
     444             :     Assert(roident != DoNotReplicateId);
     445             : 
     446           2 :     tuple = SearchSysCache1(REPLORIGIDENT,
     447             :                             ObjectIdGetDatum((Oid) roident));
     448             : 
     449           2 :     if (HeapTupleIsValid(tuple))
     450             :     {
     451           2 :         ric = (Form_pg_replication_origin) GETSTRUCT(tuple);
     452           2 :         *roname = text_to_cstring(&ric->roname);
     453           2 :         ReleaseSysCache(tuple);
     454             : 
     455           2 :         return true;
     456             :     }
     457             :     else
     458             :     {
     459           0 :         *roname = NULL;
     460             : 
     461           0 :         if (!missing_ok)
     462           0 :             ereport(ERROR,
     463             :                     (errcode(ERRCODE_UNDEFINED_OBJECT),
     464             :                      errmsg("replication origin with OID %u does not exist",
     465             :                             roident)));
     466             : 
     467           0 :         return false;
     468             :     }
     469             : }
     470             : 
     471             : 
     472             : /* ---------------------------------------------------------------------------
     473             :  * Functions for handling replication progress.
     474             :  * ---------------------------------------------------------------------------
     475             :  */
     476             : 
     477             : Size
     478        6514 : ReplicationOriginShmemSize(void)
     479             : {
     480        6514 :     Size        size = 0;
     481             : 
     482             :     /*
     483             :      * XXX: max_replication_slots is arguably the wrong thing to use, as here
     484             :      * we keep the replay state of *remote* transactions. But for now it seems
     485             :      * sufficient to reuse it, rather than introduce a separate GUC.
     486             :      */
     487        6514 :     if (max_replication_slots == 0)
     488           0 :         return size;
     489             : 
     490        6514 :     size = add_size(size, offsetof(ReplicationStateCtl, states));
     491             : 
     492        6514 :     size = add_size(size,
     493             :                     mul_size(max_replication_slots, sizeof(ReplicationState)));
     494        6514 :     return size;
     495             : }
     496             : 
     497             : void
     498        2170 : ReplicationOriginShmemInit(void)
     499             : {
     500             :     bool        found;
     501             : 
     502        2170 :     if (max_replication_slots == 0)
     503           0 :         return;
     504             : 
     505        2170 :     replication_states_ctl = (ReplicationStateCtl *)
     506        2170 :         ShmemInitStruct("ReplicationOriginState",
     507             :                         ReplicationOriginShmemSize(),
     508             :                         &found);
     509        2170 :     replication_states = replication_states_ctl->states;
     510             : 
     511        2170 :     if (!found)
     512             :     {
     513             :         int         i;
     514             : 
     515      143290 :         MemSet(replication_states_ctl, 0, ReplicationOriginShmemSize());
     516             : 
     517        2170 :         replication_states_ctl->tranche_id = LWTRANCHE_REPLICATION_ORIGIN_STATE;
     518             : 
     519       22020 :         for (i = 0; i < max_replication_slots; i++)
     520             :         {
     521       19850 :             LWLockInitialize(&replication_states[i].lock,
     522       19850 :                              replication_states_ctl->tranche_id);
     523       19850 :             ConditionVariableInit(&replication_states[i].origin_cv);
     524             :         }
     525             :     }
     526             : }
     527             : 
     528             : /* ---------------------------------------------------------------------------
     529             :  * Perform a checkpoint of each replication origin's progress with respect to
     530             :  * the replayed remote_lsn. Make sure that all transactions we refer to in the
     531             :  * checkpoint (local_lsn) are actually on-disk. This might not yet be the case
     532             :  * if the transactions were originally committed asynchronously.
     533             :  *
     534             :  * We store checkpoints in the following format:
     535             :  * +-------+------------------------+------------------+-----+--------+
     536             :  * | MAGIC | ReplicationStateOnDisk | struct Replic... | ... | CRC32C | EOF
     537             :  * +-------+------------------------+------------------+-----+--------+
     538             :  *
     539             :  * So its just the magic, followed by the statically sized
     540             :  * ReplicationStateOnDisk structs. Note that the maximum number of
     541             :  * ReplicationState is determined by max_replication_slots.
     542             :  * ---------------------------------------------------------------------------
     543             :  */
     544             : void
     545        3172 : CheckPointReplicationOrigin(void)
     546             : {
     547        3172 :     const char *tmppath = "pg_logical/replorigin_checkpoint.tmp";
     548        3172 :     const char *path = "pg_logical/replorigin_checkpoint";
     549             :     int         tmpfd;
     550             :     int         i;
     551        3172 :     uint32      magic = REPLICATION_STATE_MAGIC;
     552             :     pg_crc32c   crc;
     553             : 
     554        3172 :     if (max_replication_slots == 0)
     555           0 :         return;
     556             : 
     557        3172 :     INIT_CRC32C(crc);
     558             : 
     559             :     /* make sure no old temp file is remaining */
     560        3172 :     if (unlink(tmppath) < 0 && errno != ENOENT)
     561           0 :         ereport(PANIC,
     562             :                 (errcode_for_file_access(),
     563             :                  errmsg("could not remove file \"%s\": %m",
     564             :                         tmppath)));
     565             : 
     566             :     /*
     567             :      * no other backend can perform this at the same time, we're protected by
     568             :      * CheckpointLock.
     569             :      */
     570        3172 :     tmpfd = OpenTransientFile(tmppath,
     571             :                               O_CREAT | O_EXCL | O_WRONLY | PG_BINARY);
     572        3172 :     if (tmpfd < 0)
     573           0 :         ereport(PANIC,
     574             :                 (errcode_for_file_access(),
     575             :                  errmsg("could not create file \"%s\": %m",
     576             :                         tmppath)));
     577             : 
     578             :     /* write magic */
     579        3172 :     errno = 0;
     580        3172 :     if ((write(tmpfd, &magic, sizeof(magic))) != sizeof(magic))
     581             :     {
     582             :         /* if write didn't set errno, assume problem is no disk space */
     583           0 :         if (errno == 0)
     584           0 :             errno = ENOSPC;
     585           0 :         ereport(PANIC,
     586             :                 (errcode_for_file_access(),
     587             :                  errmsg("could not write to file \"%s\": %m",
     588             :                         tmppath)));
     589             :     }
     590        3172 :     COMP_CRC32C(crc, &magic, sizeof(magic));
     591             : 
     592             :     /* prevent concurrent creations/drops */
     593        3172 :     LWLockAcquire(ReplicationOriginLock, LW_SHARED);
     594             : 
     595             :     /* write actual data */
     596       33384 :     for (i = 0; i < max_replication_slots; i++)
     597             :     {
     598             :         ReplicationStateOnDisk disk_state;
     599       30212 :         ReplicationState *curstate = &replication_states[i];
     600             :         XLogRecPtr  local_lsn;
     601             : 
     602       30212 :         if (curstate->roident == InvalidRepOriginId)
     603       30198 :             continue;
     604             : 
     605             :         /* zero, to avoid uninitialized padding bytes */
     606          14 :         memset(&disk_state, 0, sizeof(disk_state));
     607             : 
     608          14 :         LWLockAcquire(&curstate->lock, LW_SHARED);
     609             : 
     610          14 :         disk_state.roident = curstate->roident;
     611             : 
     612          14 :         disk_state.remote_lsn = curstate->remote_lsn;
     613          14 :         local_lsn = curstate->local_lsn;
     614             : 
     615          14 :         LWLockRelease(&curstate->lock);
     616             : 
     617             :         /* make sure we only write out a commit that's persistent */
     618          14 :         XLogFlush(local_lsn);
     619             : 
     620          14 :         errno = 0;
     621          14 :         if ((write(tmpfd, &disk_state, sizeof(disk_state))) !=
     622             :             sizeof(disk_state))
     623             :         {
     624             :             /* if write didn't set errno, assume problem is no disk space */
     625           0 :             if (errno == 0)
     626           0 :                 errno = ENOSPC;
     627           0 :             ereport(PANIC,
     628             :                     (errcode_for_file_access(),
     629             :                      errmsg("could not write to file \"%s\": %m",
     630             :                             tmppath)));
     631             :         }
     632             : 
     633          14 :         COMP_CRC32C(crc, &disk_state, sizeof(disk_state));
     634             :     }
     635             : 
     636        3172 :     LWLockRelease(ReplicationOriginLock);
     637             : 
     638             :     /* write out the CRC */
     639        3172 :     FIN_CRC32C(crc);
     640        3172 :     errno = 0;
     641        3172 :     if ((write(tmpfd, &crc, sizeof(crc))) != sizeof(crc))
     642             :     {
     643             :         /* if write didn't set errno, assume problem is no disk space */
     644           0 :         if (errno == 0)
     645           0 :             errno = ENOSPC;
     646           0 :         ereport(PANIC,
     647             :                 (errcode_for_file_access(),
     648             :                  errmsg("could not write to file \"%s\": %m",
     649             :                         tmppath)));
     650             :     }
     651             : 
     652        3172 :     if (CloseTransientFile(tmpfd) != 0)
     653           0 :         ereport(PANIC,
     654             :                 (errcode_for_file_access(),
     655             :                  errmsg("could not close file \"%s\": %m",
     656             :                         tmppath)));
     657             : 
     658             :     /* fsync, rename to permanent file, fsync file and directory */
     659        3172 :     durable_rename(tmppath, path, PANIC);
     660             : }
     661             : 
     662             : /*
     663             :  * Recover replication replay status from checkpoint data saved earlier by
     664             :  * CheckPointReplicationOrigin.
     665             :  *
     666             :  * This only needs to be called at startup and *not* during every checkpoint
     667             :  * read during recovery (e.g. in HS or PITR from a base backup) afterwards. All
     668             :  * state thereafter can be recovered by looking at commit records.
     669             :  */
     670             : void
     671        1390 : StartupReplicationOrigin(void)
     672             : {
     673        1390 :     const char *path = "pg_logical/replorigin_checkpoint";
     674             :     int         fd;
     675             :     int         readBytes;
     676        1390 :     uint32      magic = REPLICATION_STATE_MAGIC;
     677        1390 :     int         last_state = 0;
     678             :     pg_crc32c   file_crc;
     679             :     pg_crc32c   crc;
     680             : 
     681             :     /* don't want to overwrite already existing state */
     682             : #ifdef USE_ASSERT_CHECKING
     683             :     static bool already_started = false;
     684             : 
     685             :     Assert(!already_started);
     686             :     already_started = true;
     687             : #endif
     688             : 
     689        1390 :     if (max_replication_slots == 0)
     690         358 :         return;
     691             : 
     692        1390 :     INIT_CRC32C(crc);
     693             : 
     694        1390 :     elog(DEBUG2, "starting up replication origin progress state");
     695             : 
     696        1390 :     fd = OpenTransientFile(path, O_RDONLY | PG_BINARY);
     697             : 
     698             :     /*
     699             :      * might have had max_replication_slots == 0 last run, or we just brought
     700             :      * up a standby.
     701             :      */
     702        1390 :     if (fd < 0 && errno == ENOENT)
     703         358 :         return;
     704        1032 :     else if (fd < 0)
     705           0 :         ereport(PANIC,
     706             :                 (errcode_for_file_access(),
     707             :                  errmsg("could not open file \"%s\": %m",
     708             :                         path)));
     709             : 
     710             :     /* verify magic, that is written even if nothing was active */
     711        1032 :     readBytes = read(fd, &magic, sizeof(magic));
     712        1032 :     if (readBytes != sizeof(magic))
     713             :     {
     714           0 :         if (readBytes < 0)
     715           0 :             ereport(PANIC,
     716             :                     (errcode_for_file_access(),
     717             :                      errmsg("could not read file \"%s\": %m",
     718             :                             path)));
     719             :         else
     720           0 :             ereport(PANIC,
     721             :                     (errcode(ERRCODE_DATA_CORRUPTED),
     722             :                      errmsg("could not read file \"%s\": read %d of %zu",
     723             :                             path, readBytes, sizeof(magic))));
     724             :     }
     725        1032 :     COMP_CRC32C(crc, &magic, sizeof(magic));
     726             : 
     727        1032 :     if (magic != REPLICATION_STATE_MAGIC)
     728           0 :         ereport(PANIC,
     729             :                 (errmsg("replication checkpoint has wrong magic %u instead of %u",
     730             :                         magic, REPLICATION_STATE_MAGIC)));
     731             : 
     732             :     /* we can skip locking here, no other access is possible */
     733             : 
     734             :     /* recover individual states, until there are no more to be found */
     735             :     while (true)
     736           0 :     {
     737             :         ReplicationStateOnDisk disk_state;
     738             : 
     739        1032 :         readBytes = read(fd, &disk_state, sizeof(disk_state));
     740             : 
     741             :         /* no further data */
     742        1032 :         if (readBytes == sizeof(crc))
     743             :         {
     744             :             /* not pretty, but simple ... */
     745        1032 :             file_crc = *(pg_crc32c *) &disk_state;
     746        1032 :             break;
     747             :         }
     748             : 
     749           0 :         if (readBytes < 0)
     750             :         {
     751           0 :             ereport(PANIC,
     752             :                     (errcode_for_file_access(),
     753             :                      errmsg("could not read file \"%s\": %m",
     754             :                             path)));
     755             :         }
     756             : 
     757           0 :         if (readBytes != sizeof(disk_state))
     758             :         {
     759           0 :             ereport(PANIC,
     760             :                     (errcode_for_file_access(),
     761             :                      errmsg("could not read file \"%s\": read %d of %zu",
     762             :                             path, readBytes, sizeof(disk_state))));
     763             :         }
     764             : 
     765           0 :         COMP_CRC32C(crc, &disk_state, sizeof(disk_state));
     766             : 
     767           0 :         if (last_state == max_replication_slots)
     768           0 :             ereport(PANIC,
     769             :                     (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
     770             :                      errmsg("could not find free replication state, increase max_replication_slots")));
     771             : 
     772             :         /* copy data to shared memory */
     773           0 :         replication_states[last_state].roident = disk_state.roident;
     774           0 :         replication_states[last_state].remote_lsn = disk_state.remote_lsn;
     775           0 :         last_state++;
     776             : 
     777           0 :         elog(LOG, "recovered replication state of node %u to %X/%X",
     778             :              disk_state.roident,
     779             :              (uint32) (disk_state.remote_lsn >> 32),
     780             :              (uint32) disk_state.remote_lsn);
     781             :     }
     782             : 
     783             :     /* now check checksum */
     784        1032 :     FIN_CRC32C(crc);
     785        1032 :     if (file_crc != crc)
     786           0 :         ereport(PANIC,
     787             :                 (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
     788             :                  errmsg("replication slot checkpoint has wrong checksum %u, expected %u",
     789             :                         crc, file_crc)));
     790             : 
     791        1032 :     if (CloseTransientFile(fd) != 0)
     792           0 :         ereport(PANIC,
     793             :                 (errcode_for_file_access(),
     794             :                  errmsg("could not close file \"%s\": %m",
     795             :                         path)));
     796             : }
     797             : 
     798             : void
     799           0 : replorigin_redo(XLogReaderState *record)
     800             : {
     801           0 :     uint8       info = XLogRecGetInfo(record) & ~XLR_INFO_MASK;
     802             : 
     803           0 :     switch (info)
     804             :     {
     805           0 :         case XLOG_REPLORIGIN_SET:
     806             :             {
     807           0 :                 xl_replorigin_set *xlrec =
     808             :                 (xl_replorigin_set *) XLogRecGetData(record);
     809             : 
     810           0 :                 replorigin_advance(xlrec->node_id,
     811             :                                    xlrec->remote_lsn, record->EndRecPtr,
     812           0 :                                    xlrec->force /* backward */ ,
     813             :                                    false /* WAL log */ );
     814           0 :                 break;
     815             :             }
     816           0 :         case XLOG_REPLORIGIN_DROP:
     817             :             {
     818             :                 xl_replorigin_drop *xlrec;
     819             :                 int         i;
     820             : 
     821           0 :                 xlrec = (xl_replorigin_drop *) XLogRecGetData(record);
     822             : 
     823           0 :                 for (i = 0; i < max_replication_slots; i++)
     824             :                 {
     825           0 :                     ReplicationState *state = &replication_states[i];
     826             : 
     827             :                     /* found our slot */
     828           0 :                     if (state->roident == xlrec->node_id)
     829             :                     {
     830             :                         /* reset entry */
     831           0 :                         state->roident = InvalidRepOriginId;
     832           0 :                         state->remote_lsn = InvalidXLogRecPtr;
     833           0 :                         state->local_lsn = InvalidXLogRecPtr;
     834           0 :                         break;
     835             :                     }
     836             :                 }
     837           0 :                 break;
     838             :             }
     839           0 :         default:
     840           0 :             elog(PANIC, "replorigin_redo: unknown op code %u", info);
     841             :     }
     842           0 : }
     843             : 
     844             : 
     845             : /*
     846             :  * Tell the replication origin progress machinery that a commit from 'node'
     847             :  * that originated at the LSN remote_commit on the remote node was replayed
     848             :  * successfully and that we don't need to do so again. In combination with
     849             :  * setting up replorigin_session_origin_lsn and replorigin_session_origin
     850             :  * that ensures we won't loose knowledge about that after a crash if the
     851             :  * transaction had a persistent effect (think of asynchronous commits).
     852             :  *
     853             :  * local_commit needs to be a local LSN of the commit so that we can make sure
     854             :  * upon a checkpoint that enough WAL has been persisted to disk.
     855             :  *
     856             :  * Needs to be called with a RowExclusiveLock on pg_replication_origin,
     857             :  * unless running in recovery.
     858             :  */
     859             : void
     860           0 : replorigin_advance(RepOriginId node,
     861             :                    XLogRecPtr remote_commit, XLogRecPtr local_commit,
     862             :                    bool go_backward, bool wal_log)
     863             : {
     864             :     int         i;
     865           0 :     ReplicationState *replication_state = NULL;
     866           0 :     ReplicationState *free_state = NULL;
     867             : 
     868             :     Assert(node != InvalidRepOriginId);
     869             : 
     870             :     /* we don't track DoNotReplicateId */
     871           0 :     if (node == DoNotReplicateId)
     872           0 :         return;
     873             : 
     874             :     /*
     875             :      * XXX: For the case where this is called by WAL replay, it'd be more
     876             :      * efficient to restore into a backend local hashtable and only dump into
     877             :      * shmem after recovery is finished. Let's wait with implementing that
     878             :      * till it's shown to be a measurable expense
     879             :      */
     880             : 
     881             :     /* Lock exclusively, as we may have to create a new table entry. */
     882           0 :     LWLockAcquire(ReplicationOriginLock, LW_EXCLUSIVE);
     883             : 
     884             :     /*
     885             :      * Search for either an existing slot for the origin, or a free one we can
     886             :      * use.
     887             :      */
     888           0 :     for (i = 0; i < max_replication_slots; i++)
     889             :     {
     890           0 :         ReplicationState *curstate = &replication_states[i];
     891             : 
     892             :         /* remember where to insert if necessary */
     893           0 :         if (curstate->roident == InvalidRepOriginId &&
     894             :             free_state == NULL)
     895             :         {
     896           0 :             free_state = curstate;
     897           0 :             continue;
     898             :         }
     899             : 
     900             :         /* not our slot */
     901           0 :         if (curstate->roident != node)
     902             :         {
     903           0 :             continue;
     904             :         }
     905             : 
     906             :         /* ok, found slot */
     907           0 :         replication_state = curstate;
     908             : 
     909           0 :         LWLockAcquire(&replication_state->lock, LW_EXCLUSIVE);
     910             : 
     911             :         /* Make sure it's not used by somebody else */
     912           0 :         if (replication_state->acquired_by != 0)
     913             :         {
     914           0 :             ereport(ERROR,
     915             :                     (errcode(ERRCODE_OBJECT_IN_USE),
     916             :                      errmsg("replication origin with OID %d is already active for PID %d",
     917             :                             replication_state->roident,
     918             :                             replication_state->acquired_by)));
     919             :         }
     920             : 
     921           0 :         break;
     922             :     }
     923             : 
     924           0 :     if (replication_state == NULL && free_state == NULL)
     925           0 :         ereport(ERROR,
     926             :                 (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
     927             :                  errmsg("could not find free replication state slot for replication origin with OID %u",
     928             :                         node),
     929             :                  errhint("Increase max_replication_slots and try again.")));
     930             : 
     931           0 :     if (replication_state == NULL)
     932             :     {
     933             :         /* initialize new slot */
     934           0 :         LWLockAcquire(&free_state->lock, LW_EXCLUSIVE);
     935           0 :         replication_state = free_state;
     936             :         Assert(replication_state->remote_lsn == InvalidXLogRecPtr);
     937             :         Assert(replication_state->local_lsn == InvalidXLogRecPtr);
     938           0 :         replication_state->roident = node;
     939             :     }
     940             : 
     941             :     Assert(replication_state->roident != InvalidRepOriginId);
     942             : 
     943             :     /*
     944             :      * If somebody "forcefully" sets this slot, WAL log it, so it's durable
     945             :      * and the standby gets the message. Primarily this will be called during
     946             :      * WAL replay (of commit records) where no WAL logging is necessary.
     947             :      */
     948           0 :     if (wal_log)
     949             :     {
     950             :         xl_replorigin_set xlrec;
     951             : 
     952           0 :         xlrec.remote_lsn = remote_commit;
     953           0 :         xlrec.node_id = node;
     954           0 :         xlrec.force = go_backward;
     955             : 
     956           0 :         XLogBeginInsert();
     957           0 :         XLogRegisterData((char *) (&xlrec), sizeof(xlrec));
     958             : 
     959           0 :         XLogInsert(RM_REPLORIGIN_ID, XLOG_REPLORIGIN_SET);
     960             :     }
     961             : 
     962             :     /*
     963             :      * Due to - harmless - race conditions during a checkpoint we could see
     964             :      * values here that are older than the ones we already have in memory.
     965             :      * Don't overwrite those.
     966             :      */
     967           0 :     if (go_backward || replication_state->remote_lsn < remote_commit)
     968           0 :         replication_state->remote_lsn = remote_commit;
     969           0 :     if (local_commit != InvalidXLogRecPtr &&
     970           0 :         (go_backward || replication_state->local_lsn < local_commit))
     971           0 :         replication_state->local_lsn = local_commit;
     972           0 :     LWLockRelease(&replication_state->lock);
     973             : 
     974             :     /*
     975             :      * Release *after* changing the LSNs, slot isn't acquired and thus could
     976             :      * otherwise be dropped anytime.
     977             :      */
     978           0 :     LWLockRelease(ReplicationOriginLock);
     979             : }
     980             : 
     981             : 
     982             : XLogRecPtr
     983           4 : replorigin_get_progress(RepOriginId node, bool flush)
     984             : {
     985             :     int         i;
     986           4 :     XLogRecPtr  local_lsn = InvalidXLogRecPtr;
     987           4 :     XLogRecPtr  remote_lsn = InvalidXLogRecPtr;
     988             : 
     989             :     /* prevent slots from being concurrently dropped */
     990           4 :     LWLockAcquire(ReplicationOriginLock, LW_SHARED);
     991             : 
     992           4 :     for (i = 0; i < max_replication_slots; i++)
     993             :     {
     994             :         ReplicationState *state;
     995             : 
     996           4 :         state = &replication_states[i];
     997             : 
     998           4 :         if (state->roident == node)
     999             :         {
    1000           4 :             LWLockAcquire(&state->lock, LW_SHARED);
    1001             : 
    1002           4 :             remote_lsn = state->remote_lsn;
    1003           4 :             local_lsn = state->local_lsn;
    1004             : 
    1005           4 :             LWLockRelease(&state->lock);
    1006             : 
    1007           4 :             break;
    1008             :         }
    1009             :     }
    1010             : 
    1011           4 :     LWLockRelease(ReplicationOriginLock);
    1012             : 
    1013           4 :     if (flush && local_lsn != InvalidXLogRecPtr)
    1014           2 :         XLogFlush(local_lsn);
    1015             : 
    1016           4 :     return remote_lsn;
    1017             : }
    1018             : 
    1019             : /*
    1020             :  * Tear down a (possibly) configured session replication origin during process
    1021             :  * exit.
    1022             :  */
    1023             : static void
    1024          58 : ReplicationOriginExitCleanup(int code, Datum arg)
    1025             : {
    1026          58 :     ConditionVariable *cv = NULL;
    1027             : 
    1028          58 :     LWLockAcquire(ReplicationOriginLock, LW_EXCLUSIVE);
    1029             : 
    1030          58 :     if (session_replication_state != NULL &&
    1031          56 :         session_replication_state->acquired_by == MyProcPid)
    1032             :     {
    1033          56 :         cv = &session_replication_state->origin_cv;
    1034             : 
    1035          56 :         session_replication_state->acquired_by = 0;
    1036          56 :         session_replication_state = NULL;
    1037             :     }
    1038             : 
    1039          58 :     LWLockRelease(ReplicationOriginLock);
    1040             : 
    1041          58 :     if (cv)
    1042          56 :         ConditionVariableBroadcast(cv);
    1043          58 : }
    1044             : 
    1045             : /*
    1046             :  * Setup a replication origin in the shared memory struct if it doesn't
    1047             :  * already exists and cache access to the specific ReplicationSlot so the
    1048             :  * array doesn't have to be searched when calling
    1049             :  * replorigin_session_advance().
    1050             :  *
    1051             :  * Obviously only one such cached origin can exist per process and the current
    1052             :  * cached value can only be set again after the previous value is torn down
    1053             :  * with replorigin_session_reset().
    1054             :  */
    1055             : void
    1056          60 : replorigin_session_setup(RepOriginId node)
    1057             : {
    1058             :     static bool registered_cleanup;
    1059             :     int         i;
    1060          60 :     int         free_slot = -1;
    1061             : 
    1062          60 :     if (!registered_cleanup)
    1063             :     {
    1064          58 :         on_shmem_exit(ReplicationOriginExitCleanup, 0);
    1065          58 :         registered_cleanup = true;
    1066             :     }
    1067             : 
    1068             :     Assert(max_replication_slots > 0);
    1069             : 
    1070          60 :     if (session_replication_state != NULL)
    1071           2 :         ereport(ERROR,
    1072             :                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
    1073             :                  errmsg("cannot setup replication origin when one is already setup")));
    1074             : 
    1075             :     /* Lock exclusively, as we may have to create a new table entry. */
    1076          58 :     LWLockAcquire(ReplicationOriginLock, LW_EXCLUSIVE);
    1077             : 
    1078             :     /*
    1079             :      * Search for either an existing slot for the origin, or a free one we can
    1080             :      * use.
    1081             :      */
    1082         346 :     for (i = 0; i < max_replication_slots; i++)
    1083             :     {
    1084         288 :         ReplicationState *curstate = &replication_states[i];
    1085             : 
    1086             :         /* remember where to insert if necessary */
    1087         288 :         if (curstate->roident == InvalidRepOriginId &&
    1088             :             free_slot == -1)
    1089             :         {
    1090          58 :             free_slot = i;
    1091          58 :             continue;
    1092             :         }
    1093             : 
    1094             :         /* not our slot */
    1095         230 :         if (curstate->roident != node)
    1096         214 :             continue;
    1097             : 
    1098          16 :         else if (curstate->acquired_by != 0)
    1099             :         {
    1100           0 :             ereport(ERROR,
    1101             :                     (errcode(ERRCODE_OBJECT_IN_USE),
    1102             :                      errmsg("replication origin with OID %d is already active for PID %d",
    1103             :                             curstate->roident, curstate->acquired_by)));
    1104             :         }
    1105             : 
    1106             :         /* ok, found slot */
    1107          16 :         session_replication_state = curstate;
    1108             :     }
    1109             : 
    1110             : 
    1111          58 :     if (session_replication_state == NULL && free_slot == -1)
    1112           0 :         ereport(ERROR,
    1113             :                 (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
    1114             :                  errmsg("could not find free replication state slot for replication origin with OID %u",
    1115             :                         node),
    1116             :                  errhint("Increase max_replication_slots and try again.")));
    1117          58 :     else if (session_replication_state == NULL)
    1118             :     {
    1119             :         /* initialize new slot */
    1120          42 :         session_replication_state = &replication_states[free_slot];
    1121             :         Assert(session_replication_state->remote_lsn == InvalidXLogRecPtr);
    1122             :         Assert(session_replication_state->local_lsn == InvalidXLogRecPtr);
    1123          42 :         session_replication_state->roident = node;
    1124             :     }
    1125             : 
    1126             : 
    1127             :     Assert(session_replication_state->roident != InvalidRepOriginId);
    1128             : 
    1129          58 :     session_replication_state->acquired_by = MyProcPid;
    1130             : 
    1131          58 :     LWLockRelease(ReplicationOriginLock);
    1132             : 
    1133             :     /* probably this one is pointless */
    1134          58 :     ConditionVariableBroadcast(&session_replication_state->origin_cv);
    1135          58 : }
    1136             : 
    1137             : /*
    1138             :  * Reset replay state previously setup in this session.
    1139             :  *
    1140             :  * This function may only be called if an origin was setup with
    1141             :  * replorigin_session_setup().
    1142             :  */
    1143             : void
    1144           4 : replorigin_session_reset(void)
    1145             : {
    1146             :     ConditionVariable *cv;
    1147             : 
    1148             :     Assert(max_replication_slots != 0);
    1149             : 
    1150           4 :     if (session_replication_state == NULL)
    1151           2 :         ereport(ERROR,
    1152             :                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
    1153             :                  errmsg("no replication origin is configured")));
    1154             : 
    1155           2 :     LWLockAcquire(ReplicationOriginLock, LW_EXCLUSIVE);
    1156             : 
    1157           2 :     session_replication_state->acquired_by = 0;
    1158           2 :     cv = &session_replication_state->origin_cv;
    1159           2 :     session_replication_state = NULL;
    1160             : 
    1161           2 :     LWLockRelease(ReplicationOriginLock);
    1162             : 
    1163           2 :     ConditionVariableBroadcast(cv);
    1164           2 : }
    1165             : 
    1166             : /*
    1167             :  * Do the same work replorigin_advance() does, just on the session's
    1168             :  * configured origin.
    1169             :  *
    1170             :  * This is noticeably cheaper than using replorigin_advance().
    1171             :  */
    1172             : void
    1173         438 : replorigin_session_advance(XLogRecPtr remote_commit, XLogRecPtr local_commit)
    1174             : {
    1175             :     Assert(session_replication_state != NULL);
    1176             :     Assert(session_replication_state->roident != InvalidRepOriginId);
    1177             : 
    1178         438 :     LWLockAcquire(&session_replication_state->lock, LW_EXCLUSIVE);
    1179         438 :     if (session_replication_state->local_lsn < local_commit)
    1180         438 :         session_replication_state->local_lsn = local_commit;
    1181         438 :     if (session_replication_state->remote_lsn < remote_commit)
    1182         342 :         session_replication_state->remote_lsn = remote_commit;
    1183         438 :     LWLockRelease(&session_replication_state->lock);
    1184         438 : }
    1185             : 
    1186             : /*
    1187             :  * Ask the machinery about the point up to which we successfully replayed
    1188             :  * changes from an already setup replication origin.
    1189             :  */
    1190             : XLogRecPtr
    1191          60 : replorigin_session_get_progress(bool flush)
    1192             : {
    1193             :     XLogRecPtr  remote_lsn;
    1194             :     XLogRecPtr  local_lsn;
    1195             : 
    1196             :     Assert(session_replication_state != NULL);
    1197             : 
    1198          60 :     LWLockAcquire(&session_replication_state->lock, LW_SHARED);
    1199          60 :     remote_lsn = session_replication_state->remote_lsn;
    1200          60 :     local_lsn = session_replication_state->local_lsn;
    1201          60 :     LWLockRelease(&session_replication_state->lock);
    1202             : 
    1203          60 :     if (flush && local_lsn != InvalidXLogRecPtr)
    1204           2 :         XLogFlush(local_lsn);
    1205             : 
    1206          60 :     return remote_lsn;
    1207             : }
    1208             : 
    1209             : 
    1210             : 
    1211             : /* ---------------------------------------------------------------------------
    1212             :  * SQL functions for working with replication origin.
    1213             :  *
    1214             :  * These mostly should be fairly short wrappers around more generic functions.
    1215             :  * ---------------------------------------------------------------------------
    1216             :  */
    1217             : 
    1218             : /*
    1219             :  * Create replication origin for the passed in name, and return the assigned
    1220             :  * oid.
    1221             :  */
    1222             : Datum
    1223           6 : pg_replication_origin_create(PG_FUNCTION_ARGS)
    1224             : {
    1225             :     char       *name;
    1226             :     RepOriginId roident;
    1227             : 
    1228           6 :     replorigin_check_prerequisites(false, false);
    1229             : 
    1230           6 :     name = text_to_cstring((text *) DatumGetPointer(PG_GETARG_DATUM(0)));
    1231             : 
    1232             :     /* Replication origins "pg_xxx" are reserved for internal use */
    1233           6 :     if (IsReservedName(name))
    1234           0 :         ereport(ERROR,
    1235             :                 (errcode(ERRCODE_RESERVED_NAME),
    1236             :                  errmsg("replication origin name \"%s\" is reserved",
    1237             :                         name),
    1238             :                  errdetail("Origin names starting with \"pg_\" are reserved.")));
    1239             : 
    1240             :     /*
    1241             :      * If built with appropriate switch, whine when regression-testing
    1242             :      * conventions for replication origin names are violated.
    1243             :      */
    1244             : #ifdef ENFORCE_REGRESSION_TEST_NAME_RESTRICTIONS
    1245             :     if (strncmp(name, "regress_", 8) != 0)
    1246             :         elog(WARNING, "replication origins created by regression test cases should have names starting with \"regress_\"");
    1247             : #endif
    1248             : 
    1249           6 :     roident = replorigin_create(name);
    1250             : 
    1251           4 :     pfree(name);
    1252             : 
    1253           4 :     PG_RETURN_OID(roident);
    1254             : }
    1255             : 
    1256             : /*
    1257             :  * Drop replication origin.
    1258             :  */
    1259             : Datum
    1260           6 : pg_replication_origin_drop(PG_FUNCTION_ARGS)
    1261             : {
    1262             :     char       *name;
    1263             :     RepOriginId roident;
    1264             : 
    1265           6 :     replorigin_check_prerequisites(false, false);
    1266             : 
    1267           6 :     name = text_to_cstring((text *) DatumGetPointer(PG_GETARG_DATUM(0)));
    1268             : 
    1269           6 :     roident = replorigin_by_name(name, false);
    1270             :     Assert(OidIsValid(roident));
    1271             : 
    1272           4 :     replorigin_drop(roident, true);
    1273             : 
    1274           4 :     pfree(name);
    1275             : 
    1276           4 :     PG_RETURN_VOID();
    1277             : }
    1278             : 
    1279             : /*
    1280             :  * Return oid of a replication origin.
    1281             :  */
    1282             : Datum
    1283           0 : pg_replication_origin_oid(PG_FUNCTION_ARGS)
    1284             : {
    1285             :     char       *name;
    1286             :     RepOriginId roident;
    1287             : 
    1288           0 :     replorigin_check_prerequisites(false, false);
    1289             : 
    1290           0 :     name = text_to_cstring((text *) DatumGetPointer(PG_GETARG_DATUM(0)));
    1291           0 :     roident = replorigin_by_name(name, true);
    1292             : 
    1293           0 :     pfree(name);
    1294             : 
    1295           0 :     if (OidIsValid(roident))
    1296           0 :         PG_RETURN_OID(roident);
    1297           0 :     PG_RETURN_NULL();
    1298             : }
    1299             : 
    1300             : /*
    1301             :  * Setup a replication origin for this session.
    1302             :  */
    1303             : Datum
    1304           6 : pg_replication_origin_session_setup(PG_FUNCTION_ARGS)
    1305             : {
    1306             :     char       *name;
    1307             :     RepOriginId origin;
    1308             : 
    1309           6 :     replorigin_check_prerequisites(true, false);
    1310             : 
    1311           6 :     name = text_to_cstring((text *) DatumGetPointer(PG_GETARG_DATUM(0)));
    1312           6 :     origin = replorigin_by_name(name, false);
    1313           4 :     replorigin_session_setup(origin);
    1314             : 
    1315           2 :     replorigin_session_origin = origin;
    1316             : 
    1317           2 :     pfree(name);
    1318             : 
    1319           2 :     PG_RETURN_VOID();
    1320             : }
    1321             : 
    1322             : /*
    1323             :  * Reset previously setup origin in this session
    1324             :  */
    1325             : Datum
    1326           4 : pg_replication_origin_session_reset(PG_FUNCTION_ARGS)
    1327             : {
    1328           4 :     replorigin_check_prerequisites(true, false);
    1329             : 
    1330           4 :     replorigin_session_reset();
    1331             : 
    1332           2 :     replorigin_session_origin = InvalidRepOriginId;
    1333           2 :     replorigin_session_origin_lsn = InvalidXLogRecPtr;
    1334           2 :     replorigin_session_origin_timestamp = 0;
    1335             : 
    1336           2 :     PG_RETURN_VOID();
    1337             : }
    1338             : 
    1339             : /*
    1340             :  * Has a replication origin been setup for this session.
    1341             :  */
    1342             : Datum
    1343           0 : pg_replication_origin_session_is_setup(PG_FUNCTION_ARGS)
    1344             : {
    1345           0 :     replorigin_check_prerequisites(false, false);
    1346             : 
    1347           0 :     PG_RETURN_BOOL(replorigin_session_origin != InvalidRepOriginId);
    1348             : }
    1349             : 
    1350             : 
    1351             : /*
    1352             :  * Return the replication progress for origin setup in the current session.
    1353             :  *
    1354             :  * If 'flush' is set to true it is ensured that the returned value corresponds
    1355             :  * to a local transaction that has been flushed. This is useful if asynchronous
    1356             :  * commits are used when replaying replicated transactions.
    1357             :  */
    1358             : Datum
    1359           4 : pg_replication_origin_session_progress(PG_FUNCTION_ARGS)
    1360             : {
    1361           4 :     XLogRecPtr  remote_lsn = InvalidXLogRecPtr;
    1362           4 :     bool        flush = PG_GETARG_BOOL(0);
    1363             : 
    1364           4 :     replorigin_check_prerequisites(true, false);
    1365             : 
    1366           4 :     if (session_replication_state == NULL)
    1367           0 :         ereport(ERROR,
    1368             :                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
    1369             :                  errmsg("no replication origin is configured")));
    1370             : 
    1371           4 :     remote_lsn = replorigin_session_get_progress(flush);
    1372             : 
    1373           4 :     if (remote_lsn == InvalidXLogRecPtr)
    1374           0 :         PG_RETURN_NULL();
    1375             : 
    1376           4 :     PG_RETURN_LSN(remote_lsn);
    1377             : }
    1378             : 
    1379             : Datum
    1380           2 : pg_replication_origin_xact_setup(PG_FUNCTION_ARGS)
    1381             : {
    1382           2 :     XLogRecPtr  location = PG_GETARG_LSN(0);
    1383             : 
    1384           2 :     replorigin_check_prerequisites(true, false);
    1385             : 
    1386           2 :     if (session_replication_state == NULL)
    1387           0 :         ereport(ERROR,
    1388             :                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
    1389             :                  errmsg("no replication origin is configured")));
    1390             : 
    1391           2 :     replorigin_session_origin_lsn = location;
    1392           2 :     replorigin_session_origin_timestamp = PG_GETARG_TIMESTAMPTZ(1);
    1393             : 
    1394           2 :     PG_RETURN_VOID();
    1395             : }
    1396             : 
    1397             : Datum
    1398           0 : pg_replication_origin_xact_reset(PG_FUNCTION_ARGS)
    1399             : {
    1400           0 :     replorigin_check_prerequisites(true, false);
    1401             : 
    1402           0 :     replorigin_session_origin_lsn = InvalidXLogRecPtr;
    1403           0 :     replorigin_session_origin_timestamp = 0;
    1404             : 
    1405           0 :     PG_RETURN_VOID();
    1406             : }
    1407             : 
    1408             : 
    1409             : Datum
    1410           2 : pg_replication_origin_advance(PG_FUNCTION_ARGS)
    1411             : {
    1412           2 :     text       *name = PG_GETARG_TEXT_PP(0);
    1413           2 :     XLogRecPtr  remote_commit = PG_GETARG_LSN(1);
    1414             :     RepOriginId node;
    1415             : 
    1416           2 :     replorigin_check_prerequisites(true, false);
    1417             : 
    1418             :     /* lock to prevent the replication origin from vanishing */
    1419           2 :     LockRelationOid(ReplicationOriginRelationId, RowExclusiveLock);
    1420             : 
    1421           2 :     node = replorigin_by_name(text_to_cstring(name), false);
    1422             : 
    1423             :     /*
    1424             :      * Can't sensibly pass a local commit to be flushed at checkpoint - this
    1425             :      * xact hasn't committed yet. This is why this function should be used to
    1426             :      * set up the initial replication state, but not for replay.
    1427             :      */
    1428           0 :     replorigin_advance(node, remote_commit, InvalidXLogRecPtr,
    1429             :                        true /* go backward */ , true /* WAL log */ );
    1430             : 
    1431           0 :     UnlockRelationOid(ReplicationOriginRelationId, RowExclusiveLock);
    1432             : 
    1433           0 :     PG_RETURN_VOID();
    1434             : }
    1435             : 
    1436             : 
    1437             : /*
    1438             :  * Return the replication progress for an individual replication origin.
    1439             :  *
    1440             :  * If 'flush' is set to true it is ensured that the returned value corresponds
    1441             :  * to a local transaction that has been flushed. This is useful if asynchronous
    1442             :  * commits are used when replaying replicated transactions.
    1443             :  */
    1444             : Datum
    1445           6 : pg_replication_origin_progress(PG_FUNCTION_ARGS)
    1446             : {
    1447             :     char       *name;
    1448             :     bool        flush;
    1449             :     RepOriginId roident;
    1450           6 :     XLogRecPtr  remote_lsn = InvalidXLogRecPtr;
    1451             : 
    1452           6 :     replorigin_check_prerequisites(true, true);
    1453             : 
    1454           6 :     name = text_to_cstring((text *) DatumGetPointer(PG_GETARG_DATUM(0)));
    1455           6 :     flush = PG_GETARG_BOOL(1);
    1456             : 
    1457           6 :     roident = replorigin_by_name(name, false);
    1458             :     Assert(OidIsValid(roident));
    1459             : 
    1460           4 :     remote_lsn = replorigin_get_progress(roident, flush);
    1461             : 
    1462           4 :     if (remote_lsn == InvalidXLogRecPtr)
    1463           0 :         PG_RETURN_NULL();
    1464             : 
    1465           4 :     PG_RETURN_LSN(remote_lsn);
    1466             : }
    1467             : 
    1468             : 
    1469             : Datum
    1470           2 : pg_show_replication_origin_status(PG_FUNCTION_ARGS)
    1471             : {
    1472           2 :     ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
    1473             :     TupleDesc   tupdesc;
    1474             :     Tuplestorestate *tupstore;
    1475             :     MemoryContext per_query_ctx;
    1476             :     MemoryContext oldcontext;
    1477             :     int         i;
    1478             : #define REPLICATION_ORIGIN_PROGRESS_COLS 4
    1479             : 
    1480             :     /* we want to return 0 rows if slot is set to zero */
    1481           2 :     replorigin_check_prerequisites(false, true);
    1482             : 
    1483           2 :     if (rsinfo == NULL || !IsA(rsinfo, ReturnSetInfo))
    1484           0 :         ereport(ERROR,
    1485             :                 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
    1486             :                  errmsg("set-valued function called in context that cannot accept a set")));
    1487           2 :     if (!(rsinfo->allowedModes & SFRM_Materialize))
    1488           0 :         ereport(ERROR,
    1489             :                 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
    1490             :                  errmsg("materialize mode required, but it is not allowed in this context")));
    1491           2 :     if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
    1492           0 :         elog(ERROR, "return type must be a row type");
    1493             : 
    1494           2 :     if (tupdesc->natts != REPLICATION_ORIGIN_PROGRESS_COLS)
    1495           0 :         elog(ERROR, "wrong function definition");
    1496             : 
    1497           2 :     per_query_ctx = rsinfo->econtext->ecxt_per_query_memory;
    1498           2 :     oldcontext = MemoryContextSwitchTo(per_query_ctx);
    1499             : 
    1500           2 :     tupstore = tuplestore_begin_heap(true, false, work_mem);
    1501           2 :     rsinfo->returnMode = SFRM_Materialize;
    1502           2 :     rsinfo->setResult = tupstore;
    1503           2 :     rsinfo->setDesc = tupdesc;
    1504             : 
    1505           2 :     MemoryContextSwitchTo(oldcontext);
    1506             : 
    1507             : 
    1508             :     /* prevent slots from being concurrently dropped */
    1509           2 :     LWLockAcquire(ReplicationOriginLock, LW_SHARED);
    1510             : 
    1511             :     /*
    1512             :      * Iterate through all possible replication_states, display if they are
    1513             :      * filled. Note that we do not take any locks, so slightly corrupted/out
    1514             :      * of date values are a possibility.
    1515             :      */
    1516          10 :     for (i = 0; i < max_replication_slots; i++)
    1517             :     {
    1518             :         ReplicationState *state;
    1519             :         Datum       values[REPLICATION_ORIGIN_PROGRESS_COLS];
    1520             :         bool        nulls[REPLICATION_ORIGIN_PROGRESS_COLS];
    1521             :         char       *roname;
    1522             : 
    1523           8 :         state = &replication_states[i];
    1524             : 
    1525             :         /* unused slot, nothing to display */
    1526           8 :         if (state->roident == InvalidRepOriginId)
    1527           6 :             continue;
    1528             : 
    1529           2 :         memset(values, 0, sizeof(values));
    1530           2 :         memset(nulls, 1, sizeof(nulls));
    1531             : 
    1532           2 :         values[0] = ObjectIdGetDatum(state->roident);
    1533           2 :         nulls[0] = false;
    1534             : 
    1535             :         /*
    1536             :          * We're not preventing the origin to be dropped concurrently, so
    1537             :          * silently accept that it might be gone.
    1538             :          */
    1539           2 :         if (replorigin_by_oid(state->roident, true,
    1540             :                               &roname))
    1541             :         {
    1542           2 :             values[1] = CStringGetTextDatum(roname);
    1543           2 :             nulls[1] = false;
    1544             :         }
    1545             : 
    1546           2 :         LWLockAcquire(&state->lock, LW_SHARED);
    1547             : 
    1548           2 :         values[2] = LSNGetDatum(state->remote_lsn);
    1549           2 :         nulls[2] = false;
    1550             : 
    1551           2 :         values[3] = LSNGetDatum(state->local_lsn);
    1552           2 :         nulls[3] = false;
    1553             : 
    1554           2 :         LWLockRelease(&state->lock);
    1555             : 
    1556           2 :         tuplestore_putvalues(tupstore, tupdesc, values, nulls);
    1557             :     }
    1558             : 
    1559             :     tuplestore_donestoring(tupstore);
    1560             : 
    1561           2 :     LWLockRelease(ReplicationOriginLock);
    1562             : 
    1563             : #undef REPLICATION_ORIGIN_PROGRESS_COLS
    1564             : 
    1565           2 :     return (Datum) 0;
    1566             : }

Generated by: LCOV version 1.13