PostgreSQL Source Code  git master
slot.h File Reference
Include dependency graph for slot.h:
This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Data Structures

struct  ReplicationSlotPersistentData
 
struct  ReplicationSlot
 
struct  ReplicationSlotCtlData
 

Macros

#define PG_REPLSLOT_DIR   "pg_replslot"
 
#define SlotIsPhysical(slot)   ((slot)->data.database == InvalidOid)
 
#define SlotIsLogical(slot)   ((slot)->data.database != InvalidOid)
 

Typedefs

typedef enum ReplicationSlotPersistency ReplicationSlotPersistency
 
typedef enum ReplicationSlotInvalidationCause ReplicationSlotInvalidationCause
 
typedef struct ReplicationSlotPersistentData ReplicationSlotPersistentData
 
typedef struct ReplicationSlot ReplicationSlot
 
typedef struct ReplicationSlotCtlData ReplicationSlotCtlData
 

Enumerations

enum  ReplicationSlotPersistency { RS_PERSISTENT , RS_EPHEMERAL , RS_TEMPORARY }
 
enum  ReplicationSlotInvalidationCause { RS_INVAL_NONE , RS_INVAL_WAL_REMOVED , RS_INVAL_HORIZON , RS_INVAL_WAL_LEVEL }
 

Functions

Size ReplicationSlotsShmemSize (void)
 
void ReplicationSlotsShmemInit (void)
 
void ReplicationSlotCreate (const char *name, bool db_specific, ReplicationSlotPersistency persistency, bool two_phase, bool failover, bool synced)
 
void ReplicationSlotPersist (void)
 
void ReplicationSlotDrop (const char *name, bool nowait)
 
void ReplicationSlotDropAcquired (void)
 
void ReplicationSlotAlter (const char *name, const bool *failover, const bool *two_phase)
 
void ReplicationSlotAcquire (const char *name, bool nowait)
 
void ReplicationSlotRelease (void)
 
void ReplicationSlotCleanup (bool synced_only)
 
void ReplicationSlotSave (void)
 
void ReplicationSlotMarkDirty (void)
 
void ReplicationSlotInitialize (void)
 
bool ReplicationSlotValidateName (const char *name, int elevel)
 
void ReplicationSlotReserveWal (void)
 
void ReplicationSlotsComputeRequiredXmin (bool already_locked)
 
void ReplicationSlotsComputeRequiredLSN (void)
 
XLogRecPtr ReplicationSlotsComputeLogicalRestartLSN (void)
 
bool ReplicationSlotsCountDBSlots (Oid dboid, int *nslots, int *nactive)
 
void ReplicationSlotsDropDBSlots (Oid dboid)
 
bool InvalidateObsoleteReplicationSlots (ReplicationSlotInvalidationCause cause, XLogSegNo oldestSegno, Oid dboid, TransactionId snapshotConflictHorizon)
 
ReplicationSlotSearchNamedReplicationSlot (const char *name, bool need_lock)
 
int ReplicationSlotIndex (ReplicationSlot *slot)
 
bool ReplicationSlotName (int index, Name name)
 
void ReplicationSlotNameForTablesync (Oid suboid, Oid relid, char *syncslotname, Size szslot)
 
void ReplicationSlotDropAtPubNode (WalReceiverConn *wrconn, char *slotname, bool missing_ok)
 
void StartupReplicationSlots (void)
 
void CheckPointReplicationSlots (bool is_shutdown)
 
void CheckSlotRequirements (void)
 
void CheckSlotPermissions (void)
 
ReplicationSlotInvalidationCause GetSlotInvalidationCause (const char *invalidation_reason)
 
bool SlotExistsInSyncStandbySlots (const char *slot_name)
 
bool StandbySlotsHaveCaughtup (XLogRecPtr wait_for_lsn, int elevel)
 
void WaitForStandbyConfirmation (XLogRecPtr wait_for_lsn)
 

Variables

PGDLLIMPORT const char *const SlotInvalidationCauses []
 
PGDLLIMPORT ReplicationSlotCtlDataReplicationSlotCtl
 
PGDLLIMPORT ReplicationSlotMyReplicationSlot
 
PGDLLIMPORT int max_replication_slots
 
PGDLLIMPORT char * synchronized_standby_slots
 

Macro Definition Documentation

◆ PG_REPLSLOT_DIR

#define PG_REPLSLOT_DIR   "pg_replslot"

Definition at line 21 of file slot.h.

◆ SlotIsLogical

#define SlotIsLogical (   slot)    ((slot)->data.database != InvalidOid)

Definition at line 217 of file slot.h.

◆ SlotIsPhysical

#define SlotIsPhysical (   slot)    ((slot)->data.database == InvalidOid)

Definition at line 216 of file slot.h.

Typedef Documentation

◆ ReplicationSlot

◆ ReplicationSlotCtlData

◆ ReplicationSlotInvalidationCause

◆ ReplicationSlotPersistency

◆ ReplicationSlotPersistentData

Enumeration Type Documentation

◆ ReplicationSlotInvalidationCause

Enumerator
RS_INVAL_NONE 
RS_INVAL_WAL_REMOVED 
RS_INVAL_HORIZON 
RS_INVAL_WAL_LEVEL 

Definition at line 50 of file slot.h.

51 {
53  /* required WAL has been removed */
55  /* required rows have been removed */
57  /* wal_level insufficient for slot */
ReplicationSlotInvalidationCause
Definition: slot.h:51
@ RS_INVAL_WAL_REMOVED
Definition: slot.h:54
@ RS_INVAL_HORIZON
Definition: slot.h:56
@ RS_INVAL_WAL_LEVEL
Definition: slot.h:58
@ RS_INVAL_NONE
Definition: slot.h:52

◆ ReplicationSlotPersistency

Enumerator
RS_PERSISTENT 
RS_EPHEMERAL 
RS_TEMPORARY 

Definition at line 36 of file slot.h.

37 {
ReplicationSlotPersistency
Definition: slot.h:37
@ RS_PERSISTENT
Definition: slot.h:38
@ RS_EPHEMERAL
Definition: slot.h:39
@ RS_TEMPORARY
Definition: slot.h:40

Function Documentation

◆ CheckPointReplicationSlots()

void CheckPointReplicationSlots ( bool  is_shutdown)

Definition at line 1867 of file slot.c.

1868 {
1869  int i;
1870 
1871  elog(DEBUG1, "performing replication slot checkpoint");
1872 
1873  /*
1874  * Prevent any slot from being created/dropped while we're active. As we
1875  * explicitly do *not* want to block iterating over replication_slots or
1876  * acquiring a slot we cannot take the control lock - but that's OK,
1877  * because holding ReplicationSlotAllocationLock is strictly stronger, and
1878  * enough to guarantee that nobody can change the in_use bits on us.
1879  */
1880  LWLockAcquire(ReplicationSlotAllocationLock, LW_SHARED);
1881 
1882  for (i = 0; i < max_replication_slots; i++)
1883  {
1885  char path[MAXPGPATH];
1886 
1887  if (!s->in_use)
1888  continue;
1889 
1890  /* save the slot to disk, locking is handled in SaveSlotToPath() */
1891  sprintf(path, "%s/%s", PG_REPLSLOT_DIR, NameStr(s->data.name));
1892 
1893  /*
1894  * Slot's data is not flushed each time the confirmed_flush LSN is
1895  * updated as that could lead to frequent writes. However, we decide
1896  * to force a flush of all logical slot's data at the time of shutdown
1897  * if the confirmed_flush LSN is changed since we last flushed it to
1898  * disk. This helps in avoiding an unnecessary retreat of the
1899  * confirmed_flush LSN after restart.
1900  */
1901  if (is_shutdown && SlotIsLogical(s))
1902  {
1903  SpinLockAcquire(&s->mutex);
1904 
1905  if (s->data.invalidated == RS_INVAL_NONE &&
1907  {
1908  s->just_dirtied = true;
1909  s->dirty = true;
1910  }
1911  SpinLockRelease(&s->mutex);
1912  }
1913 
1914  SaveSlotToPath(s, path, LOG);
1915  }
1916  LWLockRelease(ReplicationSlotAllocationLock);
1917 }
#define NameStr(name)
Definition: c.h:700
#define LOG
Definition: elog.h:31
#define DEBUG1
Definition: elog.h:30
#define elog(elevel,...)
Definition: elog.h:225
int i
Definition: isn.c:72
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1168
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1781
@ LW_SHARED
Definition: lwlock.h:115
#define MAXPGPATH
#define sprintf
Definition: port.h:240
static void SaveSlotToPath(ReplicationSlot *slot, const char *dir, int elevel)
Definition: slot.c:2046
int max_replication_slots
Definition: slot.c:141
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:135
#define PG_REPLSLOT_DIR
Definition: slot.h:21
#define SlotIsLogical(slot)
Definition: slot.h:217
#define SpinLockRelease(lock)
Definition: spin.h:61
#define SpinLockAcquire(lock)
Definition: spin.h:59
ReplicationSlot replication_slots[1]
Definition: slot.h:228
XLogRecPtr confirmed_flush
Definition: slot.h:107
ReplicationSlotInvalidationCause invalidated
Definition: slot.h:99
slock_t mutex
Definition: slot.h:154
XLogRecPtr last_saved_confirmed_flush
Definition: slot.h:206
bool in_use
Definition: slot.h:157
bool just_dirtied
Definition: slot.h:163
bool dirty
Definition: slot.h:164
ReplicationSlotPersistentData data
Definition: slot.h:181

References ReplicationSlotPersistentData::confirmed_flush, ReplicationSlot::data, DEBUG1, ReplicationSlot::dirty, elog, i, ReplicationSlot::in_use, ReplicationSlotPersistentData::invalidated, ReplicationSlot::just_dirtied, ReplicationSlot::last_saved_confirmed_flush, LOG, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_replication_slots, MAXPGPATH, ReplicationSlot::mutex, ReplicationSlotPersistentData::name, NameStr, PG_REPLSLOT_DIR, ReplicationSlotCtlData::replication_slots, ReplicationSlotCtl, RS_INVAL_NONE, SaveSlotToPath(), SlotIsLogical, SpinLockAcquire, SpinLockRelease, and sprintf.

Referenced by CheckPointGuts().

◆ CheckSlotPermissions()

void CheckSlotPermissions ( void  )

Definition at line 1412 of file slot.c.

1413 {
1414  if (!has_rolreplication(GetUserId()))
1415  ereport(ERROR,
1416  (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
1417  errmsg("permission denied to use replication slots"),
1418  errdetail("Only roles with the %s attribute may use replication slots.",
1419  "REPLICATION")));
1420 }
int errdetail(const char *fmt,...)
Definition: elog.c:1203
int errcode(int sqlerrcode)
Definition: elog.c:853
int errmsg(const char *fmt,...)
Definition: elog.c:1070
#define ERROR
Definition: elog.h:39
#define ereport(elevel,...)
Definition: elog.h:149
Oid GetUserId(void)
Definition: miscinit.c:524
bool has_rolreplication(Oid roleid)
Definition: miscinit.c:743

References ereport, errcode(), errdetail(), errmsg(), ERROR, GetUserId(), and has_rolreplication().

Referenced by copy_replication_slot(), pg_create_logical_replication_slot(), pg_create_physical_replication_slot(), pg_drop_replication_slot(), pg_logical_slot_get_changes_guts(), pg_replication_slot_advance(), and pg_sync_replication_slots().

◆ CheckSlotRequirements()

void CheckSlotRequirements ( void  )

Definition at line 1390 of file slot.c.

1391 {
1392  /*
1393  * NB: Adding a new requirement likely means that RestoreSlotFromDisk()
1394  * needs the same check.
1395  */
1396 
1397  if (max_replication_slots == 0)
1398  ereport(ERROR,
1399  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1400  errmsg("replication slots can only be used if \"max_replication_slots\" > 0")));
1401 
1403  ereport(ERROR,
1404  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1405  errmsg("replication slots can only be used if \"wal_level\" >= \"replica\"")));
1406 }
int wal_level
Definition: xlog.c:131
@ WAL_LEVEL_REPLICA
Definition: xlog.h:75

References ereport, errcode(), errmsg(), ERROR, max_replication_slots, wal_level, and WAL_LEVEL_REPLICA.

Referenced by CheckLogicalDecodingRequirements(), copy_replication_slot(), pg_create_physical_replication_slot(), and pg_drop_replication_slot().

◆ GetSlotInvalidationCause()

ReplicationSlotInvalidationCause GetSlotInvalidationCause ( const char *  invalidation_reason)

Definition at line 2420 of file slot.c.

2421 {
2424  bool found PG_USED_FOR_ASSERTS_ONLY = false;
2425 
2426  Assert(invalidation_reason);
2427 
2428  for (cause = RS_INVAL_NONE; cause <= RS_INVAL_MAX_CAUSES; cause++)
2429  {
2430  if (strcmp(SlotInvalidationCauses[cause], invalidation_reason) == 0)
2431  {
2432  found = true;
2433  result = cause;
2434  break;
2435  }
2436  }
2437 
2438  Assert(found);
2439  return result;
2440 }
#define PG_USED_FOR_ASSERTS_ONLY
Definition: c.h:201
#define Assert(condition)
Definition: c.h:812
#define RS_INVAL_MAX_CAUSES
Definition: slot.c:113
const char *const SlotInvalidationCauses[]
Definition: slot.c:105

References Assert, PG_USED_FOR_ASSERTS_ONLY, RS_INVAL_MAX_CAUSES, RS_INVAL_NONE, and SlotInvalidationCauses.

Referenced by synchronize_slots().

◆ InvalidateObsoleteReplicationSlots()

bool InvalidateObsoleteReplicationSlots ( ReplicationSlotInvalidationCause  cause,
XLogSegNo  oldestSegno,
Oid  dboid,
TransactionId  snapshotConflictHorizon 
)

Definition at line 1811 of file slot.c.

1814 {
1815  XLogRecPtr oldestLSN;
1816  bool invalidated = false;
1817 
1818  Assert(cause != RS_INVAL_HORIZON || TransactionIdIsValid(snapshotConflictHorizon));
1819  Assert(cause != RS_INVAL_WAL_REMOVED || oldestSegno > 0);
1820  Assert(cause != RS_INVAL_NONE);
1821 
1822  if (max_replication_slots == 0)
1823  return invalidated;
1824 
1825  XLogSegNoOffsetToRecPtr(oldestSegno, 0, wal_segment_size, oldestLSN);
1826 
1827 restart:
1828  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
1829  for (int i = 0; i < max_replication_slots; i++)
1830  {
1832 
1833  if (!s->in_use)
1834  continue;
1835 
1836  if (InvalidatePossiblyObsoleteSlot(cause, s, oldestLSN, dboid,
1837  snapshotConflictHorizon,
1838  &invalidated))
1839  {
1840  /* if the lock was released, start from scratch */
1841  goto restart;
1842  }
1843  }
1844  LWLockRelease(ReplicationSlotControlLock);
1845 
1846  /*
1847  * If any slots have been invalidated, recalculate the resource limits.
1848  */
1849  if (invalidated)
1850  {
1853  }
1854 
1855  return invalidated;
1856 }
void ReplicationSlotsComputeRequiredXmin(bool already_locked)
Definition: slot.c:1077
void ReplicationSlotsComputeRequiredLSN(void)
Definition: slot.c:1133
static bool InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause, ReplicationSlot *s, XLogRecPtr oldestLSN, Oid dboid, TransactionId snapshotConflictHorizon, bool *invalidated)
Definition: slot.c:1571
#define TransactionIdIsValid(xid)
Definition: transam.h:41
int wal_segment_size
Definition: xlog.c:143
#define XLogSegNoOffsetToRecPtr(segno, offset, wal_segsz_bytes, dest)
uint64 XLogRecPtr
Definition: xlogdefs.h:21

References Assert, i, ReplicationSlot::in_use, InvalidatePossiblyObsoleteSlot(), LW_SHARED, LWLockAcquire(), LWLockRelease(), max_replication_slots, ReplicationSlotCtlData::replication_slots, ReplicationSlotCtl, ReplicationSlotsComputeRequiredLSN(), ReplicationSlotsComputeRequiredXmin(), RS_INVAL_HORIZON, RS_INVAL_NONE, RS_INVAL_WAL_REMOVED, TransactionIdIsValid, wal_segment_size, and XLogSegNoOffsetToRecPtr.

Referenced by CreateCheckPoint(), CreateRestartPoint(), ResolveRecoveryConflictWithSnapshot(), and xlog_redo().

◆ ReplicationSlotAcquire()

void ReplicationSlotAcquire ( const char *  name,
bool  nowait 
)

Definition at line 540 of file slot.c.

541 {
542  ReplicationSlot *s;
543  int active_pid;
544 
545  Assert(name != NULL);
546 
547 retry:
548  Assert(MyReplicationSlot == NULL);
549 
550  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
551 
552  /* Check if the slot exits with the given name. */
553  s = SearchNamedReplicationSlot(name, false);
554  if (s == NULL || !s->in_use)
555  {
556  LWLockRelease(ReplicationSlotControlLock);
557 
558  ereport(ERROR,
559  (errcode(ERRCODE_UNDEFINED_OBJECT),
560  errmsg("replication slot \"%s\" does not exist",
561  name)));
562  }
563 
564  /*
565  * This is the slot we want; check if it's active under some other
566  * process. In single user mode, we don't need this check.
567  */
568  if (IsUnderPostmaster)
569  {
570  /*
571  * Get ready to sleep on the slot in case it is active. (We may end
572  * up not sleeping, but we don't want to do this while holding the
573  * spinlock.)
574  */
575  if (!nowait)
577 
578  SpinLockAcquire(&s->mutex);
579  if (s->active_pid == 0)
580  s->active_pid = MyProcPid;
581  active_pid = s->active_pid;
582  SpinLockRelease(&s->mutex);
583  }
584  else
585  active_pid = MyProcPid;
586  LWLockRelease(ReplicationSlotControlLock);
587 
588  /*
589  * If we found the slot but it's already active in another process, we
590  * wait until the owning process signals us that it's been released, or
591  * error out.
592  */
593  if (active_pid != MyProcPid)
594  {
595  if (!nowait)
596  {
597  /* Wait here until we get signaled, and then restart */
599  WAIT_EVENT_REPLICATION_SLOT_DROP);
601  goto retry;
602  }
603 
604  ereport(ERROR,
605  (errcode(ERRCODE_OBJECT_IN_USE),
606  errmsg("replication slot \"%s\" is active for PID %d",
607  NameStr(s->data.name), active_pid)));
608  }
609  else if (!nowait)
610  ConditionVariableCancelSleep(); /* no sleep needed after all */
611 
612  /* Let everybody know we've modified this slot */
614 
615  /* We made this slot active, so it's ours now. */
616  MyReplicationSlot = s;
617 
618  /*
619  * The call to pgstat_acquire_replslot() protects against stats for a
620  * different slot, from before a restart or such, being present during
621  * pgstat_report_replslot().
622  */
623  if (SlotIsLogical(s))
625 
626  /*
627  * Reset the time since the slot has become inactive as the slot is active
628  * now.
629  */
630  SpinLockAcquire(&s->mutex);
631  s->inactive_since = 0;
632  SpinLockRelease(&s->mutex);
633 
634  if (am_walsender)
635  {
637  SlotIsLogical(s)
638  ? errmsg("acquired logical replication slot \"%s\"",
639  NameStr(s->data.name))
640  : errmsg("acquired physical replication slot \"%s\"",
641  NameStr(s->data.name)));
642  }
643 }
bool ConditionVariableCancelSleep(void)
void ConditionVariableBroadcast(ConditionVariable *cv)
void ConditionVariablePrepareToSleep(ConditionVariable *cv)
void ConditionVariableSleep(ConditionVariable *cv, uint32 wait_event_info)
int MyProcPid
Definition: globals.c:46
bool IsUnderPostmaster
Definition: globals.c:119
void pgstat_acquire_replslot(ReplicationSlot *slot)
ReplicationSlot * SearchNamedReplicationSlot(const char *name, bool need_lock)
Definition: slot.c:464
ReplicationSlot * MyReplicationSlot
Definition: slot.c:138
pid_t active_pid
Definition: slot.h:160
ConditionVariable active_cv
Definition: slot.h:187
TimestampTz inactive_since
Definition: slot.h:213
const char * name
bool am_walsender
Definition: walsender.c:115
bool log_replication_commands
Definition: walsender.c:125

References ReplicationSlot::active_cv, ReplicationSlot::active_pid, am_walsender, Assert, ConditionVariableBroadcast(), ConditionVariableCancelSleep(), ConditionVariablePrepareToSleep(), ConditionVariableSleep(), ReplicationSlot::data, DEBUG1, ereport, errcode(), errmsg(), ERROR, ReplicationSlot::in_use, ReplicationSlot::inactive_since, IsUnderPostmaster, LOG, log_replication_commands, LW_SHARED, LWLockAcquire(), LWLockRelease(), ReplicationSlot::mutex, MyProcPid, MyReplicationSlot, name, ReplicationSlotPersistentData::name, NameStr, pgstat_acquire_replslot(), SearchNamedReplicationSlot(), SlotIsLogical, SpinLockAcquire, and SpinLockRelease.

Referenced by binary_upgrade_logical_slot_has_caught_up(), drop_local_obsolete_slots(), pg_logical_slot_get_changes_guts(), pg_replication_slot_advance(), ReplicationSlotAlter(), ReplicationSlotDrop(), StartLogicalReplication(), StartReplication(), and synchronize_one_slot().

◆ ReplicationSlotAlter()

void ReplicationSlotAlter ( const char *  name,
const bool *  failover,
const bool *  two_phase 
)

Definition at line 807 of file slot.c.

809 {
810  bool update_slot = false;
811 
812  Assert(MyReplicationSlot == NULL);
813  Assert(failover || two_phase);
814 
816 
818  ereport(ERROR,
819  errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
820  errmsg("cannot use %s with a physical replication slot",
821  "ALTER_REPLICATION_SLOT"));
822 
824  ereport(ERROR,
825  errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
826  errmsg("cannot alter invalid replication slot \"%s\"", name),
827  errdetail("This replication slot has been invalidated due to \"%s\".",
829 
830  if (RecoveryInProgress())
831  {
832  /*
833  * Do not allow users to alter the slots which are currently being
834  * synced from the primary to the standby.
835  */
837  ereport(ERROR,
838  errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
839  errmsg("cannot alter replication slot \"%s\"", name),
840  errdetail("This replication slot is being synchronized from the primary server."));
841 
842  /*
843  * Do not allow users to enable failover on the standby as we do not
844  * support sync to the cascading standby.
845  */
846  if (failover && *failover)
847  ereport(ERROR,
848  errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
849  errmsg("cannot enable failover for a replication slot"
850  " on the standby"));
851  }
852 
853  if (failover)
854  {
855  /*
856  * Do not allow users to enable failover for temporary slots as we do
857  * not support syncing temporary slots to the standby.
858  */
859  if (*failover && MyReplicationSlot->data.persistency == RS_TEMPORARY)
860  ereport(ERROR,
861  errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
862  errmsg("cannot enable failover for a temporary replication slot"));
863 
864  if (MyReplicationSlot->data.failover != *failover)
865  {
867  MyReplicationSlot->data.failover = *failover;
869 
870  update_slot = true;
871  }
872  }
873 
875  {
879 
880  update_slot = true;
881  }
882 
883  if (update_slot)
884  {
887  }
888 
890 }
static bool two_phase
void ReplicationSlotMarkDirty(void)
Definition: slot.c:1038
void ReplicationSlotAcquire(const char *name, bool nowait)
Definition: slot.c:540
void ReplicationSlotSave(void)
Definition: slot.c:1020
void ReplicationSlotRelease(void)
Definition: slot.c:652
#define SlotIsPhysical(slot)
Definition: slot.h:216
ReplicationSlotPersistency persistency
Definition: slot.h:77
bool RecoveryInProgress(void)
Definition: xlog.c:6334

References Assert, ReplicationSlot::data, ereport, errcode(), errdetail(), errmsg(), ERROR, ReplicationSlotPersistentData::failover, ReplicationSlotPersistentData::invalidated, ReplicationSlot::mutex, MyReplicationSlot, name, ReplicationSlotPersistentData::persistency, RecoveryInProgress(), ReplicationSlotAcquire(), ReplicationSlotMarkDirty(), ReplicationSlotRelease(), ReplicationSlotSave(), RS_INVAL_NONE, RS_TEMPORARY, SlotInvalidationCauses, SlotIsPhysical, SpinLockAcquire, SpinLockRelease, ReplicationSlotPersistentData::synced, two_phase, and ReplicationSlotPersistentData::two_phase.

Referenced by AlterReplicationSlot().

◆ ReplicationSlotCleanup()

void ReplicationSlotCleanup ( bool  synced_only)

Definition at line 745 of file slot.c.

746 {
747  int i;
748 
749  Assert(MyReplicationSlot == NULL);
750 
751 restart:
752  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
753  for (i = 0; i < max_replication_slots; i++)
754  {
756 
757  if (!s->in_use)
758  continue;
759 
760  SpinLockAcquire(&s->mutex);
761  if ((s->active_pid == MyProcPid &&
762  (!synced_only || s->data.synced)))
763  {
765  SpinLockRelease(&s->mutex);
766  LWLockRelease(ReplicationSlotControlLock); /* avoid deadlock */
767 
769 
771  goto restart;
772  }
773  else
774  SpinLockRelease(&s->mutex);
775  }
776 
777  LWLockRelease(ReplicationSlotControlLock);
778 }
static void ReplicationSlotDropPtr(ReplicationSlot *slot)
Definition: slot.c:913

References ReplicationSlot::active_cv, ReplicationSlot::active_pid, Assert, ConditionVariableBroadcast(), ReplicationSlot::data, i, ReplicationSlot::in_use, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_replication_slots, ReplicationSlot::mutex, MyProcPid, MyReplicationSlot, ReplicationSlotPersistentData::persistency, ReplicationSlotCtlData::replication_slots, ReplicationSlotCtl, ReplicationSlotDropPtr(), RS_TEMPORARY, SpinLockAcquire, SpinLockRelease, and ReplicationSlotPersistentData::synced.

Referenced by PostgresMain(), ReplicationSlotShmemExit(), slotsync_failure_callback(), slotsync_worker_onexit(), SyncReplicationSlots(), and WalSndErrorCleanup().

◆ ReplicationSlotCreate()

void ReplicationSlotCreate ( const char *  name,
bool  db_specific,
ReplicationSlotPersistency  persistency,
bool  two_phase,
bool  failover,
bool  synced 
)

Definition at line 309 of file slot.c.

312 {
313  ReplicationSlot *slot = NULL;
314  int i;
315 
316  Assert(MyReplicationSlot == NULL);
317 
319 
320  if (failover)
321  {
322  /*
323  * Do not allow users to create the failover enabled slots on the
324  * standby as we do not support sync to the cascading standby.
325  *
326  * However, failover enabled slots can be created during slot
327  * synchronization because we need to retain the same values as the
328  * remote slot.
329  */
331  ereport(ERROR,
332  errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
333  errmsg("cannot enable failover for a replication slot created on the standby"));
334 
335  /*
336  * Do not allow users to create failover enabled temporary slots,
337  * because temporary slots will not be synced to the standby.
338  *
339  * However, failover enabled temporary slots can be created during
340  * slot synchronization. See the comments atop slotsync.c for details.
341  */
342  if (persistency == RS_TEMPORARY && !IsSyncingReplicationSlots())
343  ereport(ERROR,
344  errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
345  errmsg("cannot enable failover for a temporary replication slot"));
346  }
347 
348  /*
349  * If some other backend ran this code concurrently with us, we'd likely
350  * both allocate the same slot, and that would be bad. We'd also be at
351  * risk of missing a name collision. Also, we don't want to try to create
352  * a new slot while somebody's busy cleaning up an old one, because we
353  * might both be monkeying with the same directory.
354  */
355  LWLockAcquire(ReplicationSlotAllocationLock, LW_EXCLUSIVE);
356 
357  /*
358  * Check for name collision, and identify an allocatable slot. We need to
359  * hold ReplicationSlotControlLock in shared mode for this, so that nobody
360  * else can change the in_use flags while we're looking at them.
361  */
362  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
363  for (i = 0; i < max_replication_slots; i++)
364  {
366 
367  if (s->in_use && strcmp(name, NameStr(s->data.name)) == 0)
368  ereport(ERROR,
370  errmsg("replication slot \"%s\" already exists", name)));
371  if (!s->in_use && slot == NULL)
372  slot = s;
373  }
374  LWLockRelease(ReplicationSlotControlLock);
375 
376  /* If all slots are in use, we're out of luck. */
377  if (slot == NULL)
378  ereport(ERROR,
379  (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
380  errmsg("all replication slots are in use"),
381  errhint("Free one or increase \"max_replication_slots\".")));
382 
383  /*
384  * Since this slot is not in use, nobody should be looking at any part of
385  * it other than the in_use field unless they're trying to allocate it.
386  * And since we hold ReplicationSlotAllocationLock, nobody except us can
387  * be doing that. So it's safe to initialize the slot.
388  */
389  Assert(!slot->in_use);
390  Assert(slot->active_pid == 0);
391 
392  /* first initialize persistent data */
393  memset(&slot->data, 0, sizeof(ReplicationSlotPersistentData));
394  namestrcpy(&slot->data.name, name);
395  slot->data.database = db_specific ? MyDatabaseId : InvalidOid;
396  slot->data.persistency = persistency;
397  slot->data.two_phase = two_phase;
399  slot->data.failover = failover;
400  slot->data.synced = synced;
401 
402  /* and then data only present in shared memory */
403  slot->just_dirtied = false;
404  slot->dirty = false;
412  slot->inactive_since = 0;
413 
414  /*
415  * Create the slot on disk. We haven't actually marked the slot allocated
416  * yet, so no special cleanup is required if this errors out.
417  */
418  CreateSlotOnDisk(slot);
419 
420  /*
421  * We need to briefly prevent any other backend from iterating over the
422  * slots while we flip the in_use flag. We also need to set the active
423  * flag while holding the ControlLock as otherwise a concurrent
424  * ReplicationSlotAcquire() could acquire the slot as well.
425  */
426  LWLockAcquire(ReplicationSlotControlLock, LW_EXCLUSIVE);
427 
428  slot->in_use = true;
429 
430  /* We can now mark the slot active, and that makes it our slot. */
431  SpinLockAcquire(&slot->mutex);
432  Assert(slot->active_pid == 0);
433  slot->active_pid = MyProcPid;
434  SpinLockRelease(&slot->mutex);
435  MyReplicationSlot = slot;
436 
437  LWLockRelease(ReplicationSlotControlLock);
438 
439  /*
440  * Create statistics entry for the new logical slot. We don't collect any
441  * stats for physical slots, so no need to create an entry for the same.
442  * See ReplicationSlotDropPtr for why we need to do this before releasing
443  * ReplicationSlotAllocationLock.
444  */
445  if (SlotIsLogical(slot))
447 
448  /*
449  * Now that the slot has been marked as in_use and active, it's safe to
450  * let somebody else try to allocate a slot.
451  */
452  LWLockRelease(ReplicationSlotAllocationLock);
453 
454  /* Let everybody know we've modified this slot */
456 }
int errhint(const char *fmt,...)
Definition: elog.c:1317
Oid MyDatabaseId
Definition: globals.c:93
@ LW_EXCLUSIVE
Definition: lwlock.h:114
void namestrcpy(Name name, const char *str)
Definition: name.c:233
void pgstat_create_replslot(ReplicationSlot *slot)
#define InvalidOid
Definition: postgres_ext.h:36
static void CreateSlotOnDisk(ReplicationSlot *slot)
Definition: slot.c:1985
bool ReplicationSlotValidateName(const char *name, int elevel)
Definition: slot.c:252
bool IsSyncingReplicationSlots(void)
Definition: slotsync.c:1649
#define ERRCODE_DUPLICATE_OBJECT
Definition: streamutil.c:30
XLogRecPtr candidate_xmin_lsn
Definition: slot.h:197
TransactionId effective_catalog_xmin
Definition: slot.h:178
XLogRecPtr candidate_restart_valid
Definition: slot.h:198
TransactionId effective_xmin
Definition: slot.h:177
XLogRecPtr candidate_restart_lsn
Definition: slot.h:199
TransactionId candidate_catalog_xmin
Definition: slot.h:196
#define InvalidTransactionId
Definition: transam.h:31
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28

References ReplicationSlot::active_cv, ReplicationSlot::active_pid, Assert, ReplicationSlot::candidate_catalog_xmin, ReplicationSlot::candidate_restart_lsn, ReplicationSlot::candidate_restart_valid, ReplicationSlot::candidate_xmin_lsn, ConditionVariableBroadcast(), CreateSlotOnDisk(), ReplicationSlot::data, ReplicationSlotPersistentData::database, ReplicationSlot::dirty, ReplicationSlot::effective_catalog_xmin, ReplicationSlot::effective_xmin, ereport, errcode(), ERRCODE_DUPLICATE_OBJECT, errhint(), errmsg(), ERROR, ReplicationSlotPersistentData::failover, i, ReplicationSlot::in_use, ReplicationSlot::inactive_since, InvalidOid, InvalidTransactionId, InvalidXLogRecPtr, IsSyncingReplicationSlots(), ReplicationSlot::just_dirtied, ReplicationSlot::last_saved_confirmed_flush, LW_EXCLUSIVE, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_replication_slots, ReplicationSlot::mutex, MyDatabaseId, MyProcPid, MyReplicationSlot, name, ReplicationSlotPersistentData::name, NameStr, namestrcpy(), ReplicationSlotPersistentData::persistency, pgstat_create_replslot(), RecoveryInProgress(), ReplicationSlotCtlData::replication_slots, ReplicationSlotCtl, ReplicationSlotValidateName(), RS_TEMPORARY, SlotIsLogical, SpinLockAcquire, SpinLockRelease, ReplicationSlotPersistentData::synced, two_phase, ReplicationSlotPersistentData::two_phase, and ReplicationSlotPersistentData::two_phase_at.

Referenced by create_logical_replication_slot(), create_physical_replication_slot(), CreateReplicationSlot(), and synchronize_one_slot().

◆ ReplicationSlotDrop()

void ReplicationSlotDrop ( const char *  name,
bool  nowait 
)

Definition at line 784 of file slot.c.

785 {
786  Assert(MyReplicationSlot == NULL);
787 
788  ReplicationSlotAcquire(name, nowait);
789 
790  /*
791  * Do not allow users to drop the slots which are currently being synced
792  * from the primary to the standby.
793  */
795  ereport(ERROR,
796  errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
797  errmsg("cannot drop replication slot \"%s\"", name),
798  errdetail("This replication slot is being synchronized from the primary server."));
799 
801 }
void ReplicationSlotDropAcquired(void)
Definition: slot.c:896

References Assert, ReplicationSlot::data, ereport, errcode(), errdetail(), errmsg(), ERROR, MyReplicationSlot, name, RecoveryInProgress(), ReplicationSlotAcquire(), ReplicationSlotDropAcquired(), and ReplicationSlotPersistentData::synced.

Referenced by DropReplicationSlot(), and pg_drop_replication_slot().

◆ ReplicationSlotDropAcquired()

void ReplicationSlotDropAcquired ( void  )

Definition at line 896 of file slot.c.

897 {
899 
900  Assert(MyReplicationSlot != NULL);
901 
902  /* slot isn't acquired anymore */
903  MyReplicationSlot = NULL;
904 
906 }

References Assert, MyReplicationSlot, and ReplicationSlotDropPtr().

Referenced by drop_local_obsolete_slots(), ReplicationSlotDrop(), ReplicationSlotRelease(), and ReplicationSlotsDropDBSlots().

◆ ReplicationSlotDropAtPubNode()

void ReplicationSlotDropAtPubNode ( WalReceiverConn wrconn,
char *  slotname,
bool  missing_ok 
)

Definition at line 1913 of file subscriptioncmds.c.

1914 {
1915  StringInfoData cmd;
1916 
1917  Assert(wrconn);
1918 
1919  load_file("libpqwalreceiver", false);
1920 
1921  initStringInfo(&cmd);
1922  appendStringInfo(&cmd, "DROP_REPLICATION_SLOT %s WAIT", quote_identifier(slotname));
1923 
1924  PG_TRY();
1925  {
1927 
1928  res = walrcv_exec(wrconn, cmd.data, 0, NULL);
1929 
1930  if (res->status == WALRCV_OK_COMMAND)
1931  {
1932  /* NOTICE. Success. */
1933  ereport(NOTICE,
1934  (errmsg("dropped replication slot \"%s\" on publisher",
1935  slotname)));
1936  }
1937  else if (res->status == WALRCV_ERROR &&
1938  missing_ok &&
1939  res->sqlstate == ERRCODE_UNDEFINED_OBJECT)
1940  {
1941  /* LOG. Error, but missing_ok = true. */
1942  ereport(LOG,
1943  (errmsg("could not drop replication slot \"%s\" on publisher: %s",
1944  slotname, res->err)));
1945  }
1946  else
1947  {
1948  /* ERROR. */
1949  ereport(ERROR,
1950  (errcode(ERRCODE_CONNECTION_FAILURE),
1951  errmsg("could not drop replication slot \"%s\" on publisher: %s",
1952  slotname, res->err)));
1953  }
1954 
1956  }
1957  PG_FINALLY();
1958  {
1959  pfree(cmd.data);
1960  }
1961  PG_END_TRY();
1962 }
void load_file(const char *filename, bool restricted)
Definition: dfmgr.c:134
#define PG_TRY(...)
Definition: elog.h:371
#define PG_END_TRY(...)
Definition: elog.h:396
#define NOTICE
Definition: elog.h:35
#define PG_FINALLY(...)
Definition: elog.h:388
void pfree(void *pointer)
Definition: mcxt.c:1521
const char * quote_identifier(const char *ident)
Definition: ruleutils.c:12868
void appendStringInfo(StringInfo str, const char *fmt,...)
Definition: stringinfo.c:94
void initStringInfo(StringInfo str)
Definition: stringinfo.c:56
static WalReceiverConn * wrconn
Definition: walreceiver.c:92
@ WALRCV_OK_COMMAND
Definition: walreceiver.h:205
@ WALRCV_ERROR
Definition: walreceiver.h:204
static void walrcv_clear_result(WalRcvExecResult *walres)
Definition: walreceiver.h:471
#define walrcv_exec(conn, exec, nRetTypes, retTypes)
Definition: walreceiver.h:465

References appendStringInfo(), Assert, StringInfoData::data, ereport, errcode(), errmsg(), ERROR, initStringInfo(), load_file(), LOG, NOTICE, pfree(), PG_END_TRY, PG_FINALLY, PG_TRY, quote_identifier(), res, walrcv_clear_result(), WALRCV_ERROR, walrcv_exec, WALRCV_OK_COMMAND, and wrconn.

Referenced by AlterSubscription_refresh(), DropSubscription(), LogicalRepSyncTableStart(), and process_syncing_tables_for_sync().

◆ ReplicationSlotIndex()

int ReplicationSlotIndex ( ReplicationSlot slot)

◆ ReplicationSlotInitialize()

void ReplicationSlotInitialize ( void  )

Definition at line 224 of file slot.c.

225 {
227 }
void before_shmem_exit(pg_on_exit_callback function, Datum arg)
Definition: ipc.c:337
static void ReplicationSlotShmemExit(int code, Datum arg)
Definition: slot.c:233

References before_shmem_exit(), and ReplicationSlotShmemExit().

Referenced by BaseInit().

◆ ReplicationSlotMarkDirty()

◆ ReplicationSlotName()

bool ReplicationSlotName ( int  index,
Name  name 
)

Definition at line 513 of file slot.c.

514 {
515  ReplicationSlot *slot;
516  bool found;
517 
519 
520  /*
521  * Ensure that the slot cannot be dropped while we copy the name. Don't
522  * need the spinlock as the name of an existing slot cannot change.
523  */
524  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
525  found = slot->in_use;
526  if (slot->in_use)
527  namestrcpy(name, NameStr(slot->data.name));
528  LWLockRelease(ReplicationSlotControlLock);
529 
530  return found;
531 }
Definition: type.h:96

References ReplicationSlot::data, ReplicationSlot::in_use, LW_SHARED, LWLockAcquire(), LWLockRelease(), name, ReplicationSlotPersistentData::name, NameStr, namestrcpy(), ReplicationSlotCtlData::replication_slots, and ReplicationSlotCtl.

Referenced by pgstat_replslot_to_serialized_name_cb().

◆ ReplicationSlotNameForTablesync()

void ReplicationSlotNameForTablesync ( Oid  suboid,
Oid  relid,
char *  syncslotname,
Size  szslot 
)

Definition at line 1274 of file tablesync.c.

1276 {
1277  snprintf(syncslotname, szslot, "pg_%u_sync_%u_" UINT64_FORMAT, suboid,
1278  relid, GetSystemIdentifier());
1279 }
#define UINT64_FORMAT
Definition: c.h:504
#define snprintf
Definition: port.h:238
uint64 GetSystemIdentifier(void)
Definition: xlog.c:4568

References GetSystemIdentifier(), snprintf, and UINT64_FORMAT.

Referenced by AlterSubscription_refresh(), DropSubscription(), LogicalRepSyncTableStart(), process_syncing_tables_for_sync(), and ReportSlotConnectionError().

◆ ReplicationSlotPersist()

◆ ReplicationSlotRelease()

void ReplicationSlotRelease ( void  )

Definition at line 652 of file slot.c.

653 {
655  char *slotname = NULL; /* keep compiler quiet */
656  bool is_logical = false; /* keep compiler quiet */
657  TimestampTz now = 0;
658 
659  Assert(slot != NULL && slot->active_pid != 0);
660 
661  if (am_walsender)
662  {
663  slotname = pstrdup(NameStr(slot->data.name));
664  is_logical = SlotIsLogical(slot);
665  }
666 
667  if (slot->data.persistency == RS_EPHEMERAL)
668  {
669  /*
670  * Delete the slot. There is no !PANIC case where this is allowed to
671  * fail, all that may happen is an incomplete cleanup of the on-disk
672  * data.
673  */
675  }
676 
677  /*
678  * If slot needed to temporarily restrain both data and catalog xmin to
679  * create the catalog snapshot, remove that temporary constraint.
680  * Snapshots can only be exported while the initial snapshot is still
681  * acquired.
682  */
683  if (!TransactionIdIsValid(slot->data.xmin) &&
685  {
686  SpinLockAcquire(&slot->mutex);
688  SpinLockRelease(&slot->mutex);
690  }
691 
692  /*
693  * Set the time since the slot has become inactive. We get the current
694  * time beforehand to avoid system call while holding the spinlock.
695  */
697 
698  if (slot->data.persistency == RS_PERSISTENT)
699  {
700  /*
701  * Mark persistent slot inactive. We're not freeing it, just
702  * disconnecting, but wake up others that may be waiting for it.
703  */
704  SpinLockAcquire(&slot->mutex);
705  slot->active_pid = 0;
706  slot->inactive_since = now;
707  SpinLockRelease(&slot->mutex);
709  }
710  else
711  {
712  SpinLockAcquire(&slot->mutex);
713  slot->inactive_since = now;
714  SpinLockRelease(&slot->mutex);
715  }
716 
717  MyReplicationSlot = NULL;
718 
719  /* might not have been set when we've been a plain slot */
720  LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
723  LWLockRelease(ProcArrayLock);
724 
725  if (am_walsender)
726  {
728  is_logical
729  ? errmsg("released logical replication slot \"%s\"",
730  slotname)
731  : errmsg("released physical replication slot \"%s\"",
732  slotname));
733 
734  pfree(slotname);
735  }
736 }
TimestampTz GetCurrentTimestamp(void)
Definition: timestamp.c:1644
Datum now(PG_FUNCTION_ARGS)
Definition: timestamp.c:1608
int64 TimestampTz
Definition: timestamp.h:39
char * pstrdup(const char *in)
Definition: mcxt.c:1696
#define PROC_IN_LOGICAL_DECODING
Definition: proc.h:61
PGPROC * MyProc
Definition: proc.c:66
PROC_HDR * ProcGlobal
Definition: proc.c:78
uint8 statusFlags
Definition: proc.h:242
int pgxactoff
Definition: proc.h:184
uint8 * statusFlags
Definition: proc.h:399
TransactionId xmin
Definition: slot.h:85

References ReplicationSlot::active_cv, ReplicationSlot::active_pid, am_walsender, Assert, ConditionVariableBroadcast(), ReplicationSlot::data, DEBUG1, ReplicationSlot::effective_xmin, ereport, errmsg(), GetCurrentTimestamp(), ReplicationSlot::inactive_since, InvalidTransactionId, LOG, log_replication_commands, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), ReplicationSlot::mutex, MyProc, MyReplicationSlot, ReplicationSlotPersistentData::name, NameStr, now(), ReplicationSlotPersistentData::persistency, pfree(), PGPROC::pgxactoff, PROC_IN_LOGICAL_DECODING, ProcGlobal, pstrdup(), ReplicationSlotDropAcquired(), ReplicationSlotsComputeRequiredXmin(), RS_EPHEMERAL, RS_PERSISTENT, SlotIsLogical, SpinLockAcquire, SpinLockRelease, PGPROC::statusFlags, PROC_HDR::statusFlags, TransactionIdIsValid, and ReplicationSlotPersistentData::xmin.

Referenced by binary_upgrade_logical_slot_has_caught_up(), copy_replication_slot(), CreateReplicationSlot(), InvalidatePossiblyObsoleteSlot(), pg_create_logical_replication_slot(), pg_create_physical_replication_slot(), pg_logical_slot_get_changes_guts(), pg_replication_slot_advance(), PostgresMain(), ReplicationSlotAlter(), ReplicationSlotShmemExit(), slotsync_failure_callback(), slotsync_worker_onexit(), StartLogicalReplication(), StartReplication(), synchronize_one_slot(), and WalSndErrorCleanup().

◆ ReplicationSlotReserveWal()

void ReplicationSlotReserveWal ( void  )

Definition at line 1429 of file slot.c.

1430 {
1432 
1433  Assert(slot != NULL);
1435 
1436  /*
1437  * The replication slot mechanism is used to prevent removal of required
1438  * WAL. As there is no interlock between this routine and checkpoints, WAL
1439  * segments could concurrently be removed when a now stale return value of
1440  * ReplicationSlotsComputeRequiredLSN() is used. In the unlikely case that
1441  * this happens we'll just retry.
1442  */
1443  while (true)
1444  {
1445  XLogSegNo segno;
1446  XLogRecPtr restart_lsn;
1447 
1448  /*
1449  * For logical slots log a standby snapshot and start logical decoding
1450  * at exactly that position. That allows the slot to start up more
1451  * quickly. But on a standby we cannot do WAL writes, so just use the
1452  * replay pointer; effectively, an attempt to create a logical slot on
1453  * standby will cause it to wait for an xl_running_xact record to be
1454  * logged independently on the primary, so that a snapshot can be
1455  * built using the record.
1456  *
1457  * None of this is needed (or indeed helpful) for physical slots as
1458  * they'll start replay at the last logged checkpoint anyway. Instead
1459  * return the location of the last redo LSN. While that slightly
1460  * increases the chance that we have to retry, it's where a base
1461  * backup has to start replay at.
1462  */
1463  if (SlotIsPhysical(slot))
1464  restart_lsn = GetRedoRecPtr();
1465  else if (RecoveryInProgress())
1466  restart_lsn = GetXLogReplayRecPtr(NULL);
1467  else
1468  restart_lsn = GetXLogInsertRecPtr();
1469 
1470  SpinLockAcquire(&slot->mutex);
1471  slot->data.restart_lsn = restart_lsn;
1472  SpinLockRelease(&slot->mutex);
1473 
1474  /* prevent WAL removal as fast as possible */
1476 
1477  /*
1478  * If all required WAL is still there, great, otherwise retry. The
1479  * slot should prevent further removal of WAL, unless there's a
1480  * concurrent ReplicationSlotsComputeRequiredLSN() after we've written
1481  * the new restart_lsn above, so normally we should never need to loop
1482  * more than twice.
1483  */
1485  if (XLogGetLastRemovedSegno() < segno)
1486  break;
1487  }
1488 
1489  if (!RecoveryInProgress() && SlotIsLogical(slot))
1490  {
1491  XLogRecPtr flushptr;
1492 
1493  /* make sure we have enough information to start */
1494  flushptr = LogStandbySnapshot();
1495 
1496  /* and make sure it's fsynced to disk */
1497  XLogFlush(flushptr);
1498  }
1499 }
XLogRecPtr LogStandbySnapshot(void)
Definition: standby.c:1281
XLogRecPtr restart_lsn
Definition: slot.h:96
XLogSegNo XLogGetLastRemovedSegno(void)
Definition: xlog.c:3758
XLogRecPtr GetRedoRecPtr(void)
Definition: xlog.c:6437
XLogRecPtr GetXLogInsertRecPtr(void)
Definition: xlog.c:9435
void XLogFlush(XLogRecPtr record)
Definition: xlog.c:2802
#define XLByteToSeg(xlrp, logSegNo, wal_segsz_bytes)
uint64 XLogSegNo
Definition: xlogdefs.h:48
XLogRecPtr GetXLogReplayRecPtr(TimeLineID *replayTLI)

References Assert, ReplicationSlot::data, GetRedoRecPtr(), GetXLogInsertRecPtr(), GetXLogReplayRecPtr(), InvalidXLogRecPtr, LogStandbySnapshot(), ReplicationSlot::mutex, MyReplicationSlot, RecoveryInProgress(), ReplicationSlotsComputeRequiredLSN(), ReplicationSlotPersistentData::restart_lsn, SlotIsLogical, SlotIsPhysical, SpinLockAcquire, SpinLockRelease, wal_segment_size, XLByteToSeg, XLogFlush(), and XLogGetLastRemovedSegno().

Referenced by create_physical_replication_slot(), CreateInitDecodingContext(), and CreateReplicationSlot().

◆ ReplicationSlotSave()

◆ ReplicationSlotsComputeLogicalRestartLSN()

XLogRecPtr ReplicationSlotsComputeLogicalRestartLSN ( void  )

Definition at line 1182 of file slot.c.

1183 {
1184  XLogRecPtr result = InvalidXLogRecPtr;
1185  int i;
1186 
1187  if (max_replication_slots <= 0)
1188  return InvalidXLogRecPtr;
1189 
1190  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
1191 
1192  for (i = 0; i < max_replication_slots; i++)
1193  {
1194  ReplicationSlot *s;
1195  XLogRecPtr restart_lsn;
1196  bool invalidated;
1197 
1199 
1200  /* cannot change while ReplicationSlotCtlLock is held */
1201  if (!s->in_use)
1202  continue;
1203 
1204  /* we're only interested in logical slots */
1205  if (!SlotIsLogical(s))
1206  continue;
1207 
1208  /* read once, it's ok if it increases while we're checking */
1209  SpinLockAcquire(&s->mutex);
1210  restart_lsn = s->data.restart_lsn;
1211  invalidated = s->data.invalidated != RS_INVAL_NONE;
1212  SpinLockRelease(&s->mutex);
1213 
1214  /* invalidated slots need not apply */
1215  if (invalidated)
1216  continue;
1217 
1218  if (restart_lsn == InvalidXLogRecPtr)
1219  continue;
1220 
1221  if (result == InvalidXLogRecPtr ||
1222  restart_lsn < result)
1223  result = restart_lsn;
1224  }
1225 
1226  LWLockRelease(ReplicationSlotControlLock);
1227 
1228  return result;
1229 }

References ReplicationSlot::data, i, ReplicationSlot::in_use, ReplicationSlotPersistentData::invalidated, InvalidXLogRecPtr, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_replication_slots, ReplicationSlot::mutex, ReplicationSlotCtlData::replication_slots, ReplicationSlotCtl, ReplicationSlotPersistentData::restart_lsn, RS_INVAL_NONE, SlotIsLogical, SpinLockAcquire, and SpinLockRelease.

Referenced by CheckPointLogicalRewriteHeap(), and CheckPointSnapBuild().

◆ ReplicationSlotsComputeRequiredLSN()

void ReplicationSlotsComputeRequiredLSN ( void  )

Definition at line 1133 of file slot.c.

1134 {
1135  int i;
1136  XLogRecPtr min_required = InvalidXLogRecPtr;
1137 
1138  Assert(ReplicationSlotCtl != NULL);
1139 
1140  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
1141  for (i = 0; i < max_replication_slots; i++)
1142  {
1144  XLogRecPtr restart_lsn;
1145  bool invalidated;
1146 
1147  if (!s->in_use)
1148  continue;
1149 
1150  SpinLockAcquire(&s->mutex);
1151  restart_lsn = s->data.restart_lsn;
1152  invalidated = s->data.invalidated != RS_INVAL_NONE;
1153  SpinLockRelease(&s->mutex);
1154 
1155  /* invalidated slots need not apply */
1156  if (invalidated)
1157  continue;
1158 
1159  if (restart_lsn != InvalidXLogRecPtr &&
1160  (min_required == InvalidXLogRecPtr ||
1161  restart_lsn < min_required))
1162  min_required = restart_lsn;
1163  }
1164  LWLockRelease(ReplicationSlotControlLock);
1165 
1166  XLogSetReplicationSlotMinimumLSN(min_required);
1167 }
void XLogSetReplicationSlotMinimumLSN(XLogRecPtr lsn)
Definition: xlog.c:2688

References Assert, ReplicationSlot::data, i, ReplicationSlot::in_use, ReplicationSlotPersistentData::invalidated, InvalidXLogRecPtr, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_replication_slots, ReplicationSlot::mutex, ReplicationSlotCtlData::replication_slots, ReplicationSlotCtl, ReplicationSlotPersistentData::restart_lsn, RS_INVAL_NONE, SpinLockAcquire, SpinLockRelease, and XLogSetReplicationSlotMinimumLSN().

Referenced by copy_replication_slot(), InvalidateObsoleteReplicationSlots(), LogicalConfirmReceivedLocation(), pg_replication_slot_advance(), PhysicalConfirmReceivedLocation(), ReplicationSlotDropPtr(), ReplicationSlotReserveWal(), reserve_wal_for_local_slot(), StartupReplicationSlots(), and update_local_synced_slot().

◆ ReplicationSlotsComputeRequiredXmin()

void ReplicationSlotsComputeRequiredXmin ( bool  already_locked)

Definition at line 1077 of file slot.c.

1078 {
1079  int i;
1081  TransactionId agg_catalog_xmin = InvalidTransactionId;
1082 
1083  Assert(ReplicationSlotCtl != NULL);
1084 
1085  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
1086 
1087  for (i = 0; i < max_replication_slots; i++)
1088  {
1090  TransactionId effective_xmin;
1091  TransactionId effective_catalog_xmin;
1092  bool invalidated;
1093 
1094  if (!s->in_use)
1095  continue;
1096 
1097  SpinLockAcquire(&s->mutex);
1098  effective_xmin = s->effective_xmin;
1099  effective_catalog_xmin = s->effective_catalog_xmin;
1100  invalidated = s->data.invalidated != RS_INVAL_NONE;
1101  SpinLockRelease(&s->mutex);
1102 
1103  /* invalidated slots need not apply */
1104  if (invalidated)
1105  continue;
1106 
1107  /* check the data xmin */
1108  if (TransactionIdIsValid(effective_xmin) &&
1109  (!TransactionIdIsValid(agg_xmin) ||
1110  TransactionIdPrecedes(effective_xmin, agg_xmin)))
1111  agg_xmin = effective_xmin;
1112 
1113  /* check the catalog xmin */
1114  if (TransactionIdIsValid(effective_catalog_xmin) &&
1115  (!TransactionIdIsValid(agg_catalog_xmin) ||
1116  TransactionIdPrecedes(effective_catalog_xmin, agg_catalog_xmin)))
1117  agg_catalog_xmin = effective_catalog_xmin;
1118  }
1119 
1120  LWLockRelease(ReplicationSlotControlLock);
1121 
1122  ProcArraySetReplicationSlotXmin(agg_xmin, agg_catalog_xmin, already_locked);
1123 }
uint32 TransactionId
Definition: c.h:606
void ProcArraySetReplicationSlotXmin(TransactionId xmin, TransactionId catalog_xmin, bool already_locked)
Definition: procarray.c:3947
bool TransactionIdPrecedes(TransactionId id1, TransactionId id2)
Definition: transam.c:280

References Assert, ReplicationSlot::data, ReplicationSlot::effective_catalog_xmin, ReplicationSlot::effective_xmin, i, ReplicationSlot::in_use, ReplicationSlotPersistentData::invalidated, InvalidTransactionId, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_replication_slots, ReplicationSlot::mutex, ProcArraySetReplicationSlotXmin(), ReplicationSlotCtlData::replication_slots, ReplicationSlotCtl, RS_INVAL_NONE, SpinLockAcquire, SpinLockRelease, TransactionIdIsValid, and TransactionIdPrecedes().

Referenced by copy_replication_slot(), CreateInitDecodingContext(), InvalidateObsoleteReplicationSlots(), LogicalConfirmReceivedLocation(), pg_replication_slot_advance(), PhysicalReplicationSlotNewXmin(), ReplicationSlotDropPtr(), ReplicationSlotRelease(), StartupReplicationSlots(), synchronize_one_slot(), and update_local_synced_slot().

◆ ReplicationSlotsCountDBSlots()

bool ReplicationSlotsCountDBSlots ( Oid  dboid,
int *  nslots,
int *  nactive 
)

Definition at line 1240 of file slot.c.

1241 {
1242  int i;
1243 
1244  *nslots = *nactive = 0;
1245 
1246  if (max_replication_slots <= 0)
1247  return false;
1248 
1249  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
1250  for (i = 0; i < max_replication_slots; i++)
1251  {
1252  ReplicationSlot *s;
1253 
1255 
1256  /* cannot change while ReplicationSlotCtlLock is held */
1257  if (!s->in_use)
1258  continue;
1259 
1260  /* only logical slots are database specific, skip */
1261  if (!SlotIsLogical(s))
1262  continue;
1263 
1264  /* not our database, skip */
1265  if (s->data.database != dboid)
1266  continue;
1267 
1268  /* NB: intentionally counting invalidated slots */
1269 
1270  /* count slots with spinlock held */
1271  SpinLockAcquire(&s->mutex);
1272  (*nslots)++;
1273  if (s->active_pid != 0)
1274  (*nactive)++;
1275  SpinLockRelease(&s->mutex);
1276  }
1277  LWLockRelease(ReplicationSlotControlLock);
1278 
1279  if (*nslots > 0)
1280  return true;
1281  return false;
1282 }

References ReplicationSlot::active_pid, ReplicationSlot::data, ReplicationSlotPersistentData::database, i, ReplicationSlot::in_use, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_replication_slots, ReplicationSlot::mutex, ReplicationSlotCtlData::replication_slots, ReplicationSlotCtl, SlotIsLogical, SpinLockAcquire, and SpinLockRelease.

Referenced by dropdb().

◆ ReplicationSlotsDropDBSlots()

void ReplicationSlotsDropDBSlots ( Oid  dboid)

Definition at line 1298 of file slot.c.

1299 {
1300  int i;
1301 
1302  if (max_replication_slots <= 0)
1303  return;
1304 
1305 restart:
1306  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
1307  for (i = 0; i < max_replication_slots; i++)
1308  {
1309  ReplicationSlot *s;
1310  char *slotname;
1311  int active_pid;
1312 
1314 
1315  /* cannot change while ReplicationSlotCtlLock is held */
1316  if (!s->in_use)
1317  continue;
1318 
1319  /* only logical slots are database specific, skip */
1320  if (!SlotIsLogical(s))
1321  continue;
1322 
1323  /* not our database, skip */
1324  if (s->data.database != dboid)
1325  continue;
1326 
1327  /* NB: intentionally including invalidated slots */
1328 
1329  /* acquire slot, so ReplicationSlotDropAcquired can be reused */
1330  SpinLockAcquire(&s->mutex);
1331  /* can't change while ReplicationSlotControlLock is held */
1332  slotname = NameStr(s->data.name);
1333  active_pid = s->active_pid;
1334  if (active_pid == 0)
1335  {
1336  MyReplicationSlot = s;
1337  s->active_pid = MyProcPid;
1338  }
1339  SpinLockRelease(&s->mutex);
1340 
1341  /*
1342  * Even though we hold an exclusive lock on the database object a
1343  * logical slot for that DB can still be active, e.g. if it's
1344  * concurrently being dropped by a backend connected to another DB.
1345  *
1346  * That's fairly unlikely in practice, so we'll just bail out.
1347  *
1348  * The slot sync worker holds a shared lock on the database before
1349  * operating on synced logical slots to avoid conflict with the drop
1350  * happening here. The persistent synced slots are thus safe but there
1351  * is a possibility that the slot sync worker has created a temporary
1352  * slot (which stays active even on release) and we are trying to drop
1353  * that here. In practice, the chances of hitting this scenario are
1354  * less as during slot synchronization, the temporary slot is
1355  * immediately converted to persistent and thus is safe due to the
1356  * shared lock taken on the database. So, we'll just bail out in such
1357  * a case.
1358  *
1359  * XXX: We can consider shutting down the slot sync worker before
1360  * trying to drop synced temporary slots here.
1361  */
1362  if (active_pid)
1363  ereport(ERROR,
1364  (errcode(ERRCODE_OBJECT_IN_USE),
1365  errmsg("replication slot \"%s\" is active for PID %d",
1366  slotname, active_pid)));
1367 
1368  /*
1369  * To avoid duplicating ReplicationSlotDropAcquired() and to avoid
1370  * holding ReplicationSlotControlLock over filesystem operations,
1371  * release ReplicationSlotControlLock and use
1372  * ReplicationSlotDropAcquired.
1373  *
1374  * As that means the set of slots could change, restart scan from the
1375  * beginning each time we release the lock.
1376  */
1377  LWLockRelease(ReplicationSlotControlLock);
1379  goto restart;
1380  }
1381  LWLockRelease(ReplicationSlotControlLock);
1382 }

References ReplicationSlot::active_pid, ReplicationSlot::data, ReplicationSlotPersistentData::database, ereport, errcode(), errmsg(), ERROR, i, ReplicationSlot::in_use, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_replication_slots, ReplicationSlot::mutex, MyProcPid, MyReplicationSlot, ReplicationSlotPersistentData::name, NameStr, ReplicationSlotCtlData::replication_slots, ReplicationSlotCtl, ReplicationSlotDropAcquired(), SlotIsLogical, SpinLockAcquire, and SpinLockRelease.

Referenced by dbase_redo(), and dropdb().

◆ ReplicationSlotsShmemInit()

void ReplicationSlotsShmemInit ( void  )

Definition at line 189 of file slot.c.

190 {
191  bool found;
192 
193  if (max_replication_slots == 0)
194  return;
195 
197  ShmemInitStruct("ReplicationSlot Ctl", ReplicationSlotsShmemSize(),
198  &found);
199 
200  if (!found)
201  {
202  int i;
203 
204  /* First time through, so initialize */
206 
207  for (i = 0; i < max_replication_slots; i++)
208  {
210 
211  /* everything else is zeroed by the memset above */
212  SpinLockInit(&slot->mutex);
216  }
217  }
218 }
#define MemSet(start, val, len)
Definition: c.h:974
void ConditionVariableInit(ConditionVariable *cv)
void LWLockInitialize(LWLock *lock, int tranche_id)
Definition: lwlock.c:707
@ LWTRANCHE_REPLICATION_SLOT_IO
Definition: lwlock.h:189
void * ShmemInitStruct(const char *name, Size size, bool *foundPtr)
Definition: shmem.c:382
Size ReplicationSlotsShmemSize(void)
Definition: slot.c:171
#define SpinLockInit(lock)
Definition: spin.h:57
LWLock io_in_progress_lock
Definition: slot.h:184

References ReplicationSlot::active_cv, ConditionVariableInit(), i, ReplicationSlot::io_in_progress_lock, LWLockInitialize(), LWTRANCHE_REPLICATION_SLOT_IO, max_replication_slots, MemSet, ReplicationSlot::mutex, ReplicationSlotCtlData::replication_slots, ReplicationSlotCtl, ReplicationSlotsShmemSize(), ShmemInitStruct(), and SpinLockInit.

Referenced by CreateOrAttachShmemStructs().

◆ ReplicationSlotsShmemSize()

Size ReplicationSlotsShmemSize ( void  )

Definition at line 171 of file slot.c.

172 {
173  Size size = 0;
174 
175  if (max_replication_slots == 0)
176  return size;
177 
178  size = offsetof(ReplicationSlotCtlData, replication_slots);
179  size = add_size(size,
181 
182  return size;
183 }
size_t Size
Definition: c.h:559
Size add_size(Size s1, Size s2)
Definition: shmem.c:488
Size mul_size(Size s1, Size s2)
Definition: shmem.c:505
static pg_noinline void Size size
Definition: slab.c:607

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

Referenced by CalculateShmemSize(), and ReplicationSlotsShmemInit().

◆ ReplicationSlotValidateName()

bool ReplicationSlotValidateName ( const char *  name,
int  elevel 
)

Definition at line 252 of file slot.c.

253 {
254  const char *cp;
255 
256  if (strlen(name) == 0)
257  {
258  ereport(elevel,
259  (errcode(ERRCODE_INVALID_NAME),
260  errmsg("replication slot name \"%s\" is too short",
261  name)));
262  return false;
263  }
264 
265  if (strlen(name) >= NAMEDATALEN)
266  {
267  ereport(elevel,
268  (errcode(ERRCODE_NAME_TOO_LONG),
269  errmsg("replication slot name \"%s\" is too long",
270  name)));
271  return false;
272  }
273 
274  for (cp = name; *cp; cp++)
275  {
276  if (!((*cp >= 'a' && *cp <= 'z')
277  || (*cp >= '0' && *cp <= '9')
278  || (*cp == '_')))
279  {
280  ereport(elevel,
281  (errcode(ERRCODE_INVALID_NAME),
282  errmsg("replication slot name \"%s\" contains invalid character",
283  name),
284  errhint("Replication slot names may only contain lower case letters, numbers, and the underscore character.")));
285  return false;
286  }
287  }
288  return true;
289 }
#define NAMEDATALEN

References ereport, errcode(), errhint(), errmsg(), name, and NAMEDATALEN.

Referenced by check_primary_slot_name(), parse_subscription_options(), ReplicationSlotCreate(), and StartupReorderBuffer().

◆ SearchNamedReplicationSlot()

ReplicationSlot* SearchNamedReplicationSlot ( const char *  name,
bool  need_lock 
)

Definition at line 464 of file slot.c.

465 {
466  int i;
467  ReplicationSlot *slot = NULL;
468 
469  if (need_lock)
470  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
471 
472  for (i = 0; i < max_replication_slots; i++)
473  {
475 
476  if (s->in_use && strcmp(name, NameStr(s->data.name)) == 0)
477  {
478  slot = s;
479  break;
480  }
481  }
482 
483  if (need_lock)
484  LWLockRelease(ReplicationSlotControlLock);
485 
486  return slot;
487 }

References ReplicationSlot::data, i, ReplicationSlot::in_use, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_replication_slots, name, ReplicationSlotPersistentData::name, NameStr, ReplicationSlotCtlData::replication_slots, and ReplicationSlotCtl.

Referenced by get_replslot_index(), pg_ls_replslotdir(), pgstat_reset_replslot(), ReadReplicationSlot(), ReplicationSlotAcquire(), StandbySlotsHaveCaughtup(), synchronize_one_slot(), and validate_sync_standby_slots().

◆ SlotExistsInSyncStandbySlots()

bool SlotExistsInSyncStandbySlots ( const char *  slot_name)

Definition at line 2572 of file slot.c.

2573 {
2574  const char *standby_slot_name;
2575 
2576  /* Return false if there is no value in synchronized_standby_slots */
2578  return false;
2579 
2580  /*
2581  * XXX: We are not expecting this list to be long so a linear search
2582  * shouldn't hurt but if that turns out not to be true then we can cache
2583  * this information for each WalSender as well.
2584  */
2585  standby_slot_name = synchronized_standby_slots_config->slot_names;
2586  for (int i = 0; i < synchronized_standby_slots_config->nslotnames; i++)
2587  {
2588  if (strcmp(standby_slot_name, slot_name) == 0)
2589  return true;
2590 
2591  standby_slot_name += strlen(standby_slot_name) + 1;
2592  }
2593 
2594  return false;
2595 }
static SyncStandbySlotsConfigData * synchronized_standby_slots_config
Definition: slot.c:151
char slot_names[FLEXIBLE_ARRAY_MEMBER]
Definition: slot.c:99

References i, SyncStandbySlotsConfigData::nslotnames, SyncStandbySlotsConfigData::slot_names, and synchronized_standby_slots_config.

Referenced by PhysicalWakeupLogicalWalSnd().

◆ StandbySlotsHaveCaughtup()

bool StandbySlotsHaveCaughtup ( XLogRecPtr  wait_for_lsn,
int  elevel 
)

Definition at line 2605 of file slot.c.

2606 {
2607  const char *name;
2608  int caught_up_slot_num = 0;
2609  XLogRecPtr min_restart_lsn = InvalidXLogRecPtr;
2610 
2611  /*
2612  * Don't need to wait for the standbys to catch up if there is no value in
2613  * synchronized_standby_slots.
2614  */
2616  return true;
2617 
2618  /*
2619  * Don't need to wait for the standbys to catch up if we are on a standby
2620  * server, since we do not support syncing slots to cascading standbys.
2621  */
2622  if (RecoveryInProgress())
2623  return true;
2624 
2625  /*
2626  * Don't need to wait for the standbys to catch up if they are already
2627  * beyond the specified WAL location.
2628  */
2630  ss_oldest_flush_lsn >= wait_for_lsn)
2631  return true;
2632 
2633  /*
2634  * To prevent concurrent slot dropping and creation while filtering the
2635  * slots, take the ReplicationSlotControlLock outside of the loop.
2636  */
2637  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
2638 
2640  for (int i = 0; i < synchronized_standby_slots_config->nslotnames; i++)
2641  {
2642  XLogRecPtr restart_lsn;
2643  bool invalidated;
2644  bool inactive;
2645  ReplicationSlot *slot;
2646 
2647  slot = SearchNamedReplicationSlot(name, false);
2648 
2649  /*
2650  * If a slot name provided in synchronized_standby_slots does not
2651  * exist, report a message and exit the loop.
2652  *
2653  * Though validate_sync_standby_slots (the GUC check_hook) tries to
2654  * avoid this, it can nonetheless happen because the user can specify
2655  * a nonexistent slot name before server startup. That function cannot
2656  * validate such a slot during startup, as ReplicationSlotCtl is not
2657  * initialized by then. Also, the user might have dropped one slot.
2658  */
2659  if (!slot)
2660  {
2661  ereport(elevel,
2662  errcode(ERRCODE_INVALID_PARAMETER_VALUE),
2663  errmsg("replication slot \"%s\" specified in parameter \"%s\" does not exist",
2664  name, "synchronized_standby_slots"),
2665  errdetail("Logical replication is waiting on the standby associated with replication slot \"%s\".",
2666  name),
2667  errhint("Create the replication slot \"%s\" or amend parameter \"%s\".",
2668  name, "synchronized_standby_slots"));
2669  break;
2670  }
2671 
2672  /* Same as above: if a slot is not physical, exit the loop. */
2673  if (SlotIsLogical(slot))
2674  {
2675  ereport(elevel,
2676  errcode(ERRCODE_INVALID_PARAMETER_VALUE),
2677  errmsg("cannot specify logical replication slot \"%s\" in parameter \"%s\"",
2678  name, "synchronized_standby_slots"),
2679  errdetail("Logical replication is waiting for correction on replication slot \"%s\".",
2680  name),
2681  errhint("Remove the logical replication slot \"%s\" from parameter \"%s\".",
2682  name, "synchronized_standby_slots"));
2683  break;
2684  }
2685 
2686  SpinLockAcquire(&slot->mutex);
2687  restart_lsn = slot->data.restart_lsn;
2688  invalidated = slot->data.invalidated != RS_INVAL_NONE;
2689  inactive = slot->active_pid == 0;
2690  SpinLockRelease(&slot->mutex);
2691 
2692  if (invalidated)
2693  {
2694  /* Specified physical slot has been invalidated */
2695  ereport(elevel,
2696  errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
2697  errmsg("physical replication slot \"%s\" specified in parameter \"%s\" has been invalidated",
2698  name, "synchronized_standby_slots"),
2699  errdetail("Logical replication is waiting on the standby associated with replication slot \"%s\".",
2700  name),
2701  errhint("Drop and recreate the replication slot \"%s\", or amend parameter \"%s\".",
2702  name, "synchronized_standby_slots"));
2703  break;
2704  }
2705 
2706  if (XLogRecPtrIsInvalid(restart_lsn) || restart_lsn < wait_for_lsn)
2707  {
2708  /* Log a message if no active_pid for this physical slot */
2709  if (inactive)
2710  ereport(elevel,
2711  errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
2712  errmsg("replication slot \"%s\" specified in parameter \"%s\" does not have active_pid",
2713  name, "synchronized_standby_slots"),
2714  errdetail("Logical replication is waiting on the standby associated with replication slot \"%s\".",
2715  name),
2716  errhint("Start the standby associated with the replication slot \"%s\", or amend parameter \"%s\".",
2717  name, "synchronized_standby_slots"));
2718 
2719  /* Continue if the current slot hasn't caught up. */
2720  break;
2721  }
2722 
2723  Assert(restart_lsn >= wait_for_lsn);
2724 
2725  if (XLogRecPtrIsInvalid(min_restart_lsn) ||
2726  min_restart_lsn > restart_lsn)
2727  min_restart_lsn = restart_lsn;
2728 
2729  caught_up_slot_num++;
2730 
2731  name += strlen(name) + 1;
2732  }
2733 
2734  LWLockRelease(ReplicationSlotControlLock);
2735 
2736  /*
2737  * Return false if not all the standbys have caught up to the specified
2738  * WAL location.
2739  */
2740  if (caught_up_slot_num != synchronized_standby_slots_config->nslotnames)
2741  return false;
2742 
2743  /* The ss_oldest_flush_lsn must not retreat. */
2745  min_restart_lsn >= ss_oldest_flush_lsn);
2746 
2747  ss_oldest_flush_lsn = min_restart_lsn;
2748 
2749  return true;
2750 }
static XLogRecPtr ss_oldest_flush_lsn
Definition: slot.c:157
#define XLogRecPtrIsInvalid(r)
Definition: xlogdefs.h:29

References ReplicationSlot::active_pid, Assert, ReplicationSlot::data, ereport, errcode(), errdetail(), errhint(), errmsg(), i, ReplicationSlotPersistentData::invalidated, InvalidXLogRecPtr, LW_SHARED, LWLockAcquire(), LWLockRelease(), ReplicationSlot::mutex, name, SyncStandbySlotsConfigData::nslotnames, RecoveryInProgress(), ReplicationSlotPersistentData::restart_lsn, RS_INVAL_NONE, SearchNamedReplicationSlot(), SyncStandbySlotsConfigData::slot_names, SlotIsLogical, SpinLockAcquire, SpinLockRelease, ss_oldest_flush_lsn, synchronized_standby_slots_config, and XLogRecPtrIsInvalid.

Referenced by NeedToWaitForStandbys(), and WaitForStandbyConfirmation().

◆ StartupReplicationSlots()

void StartupReplicationSlots ( void  )

Definition at line 1924 of file slot.c.

1925 {
1926  DIR *replication_dir;
1927  struct dirent *replication_de;
1928 
1929  elog(DEBUG1, "starting up replication slots");
1930 
1931  /* restore all slots by iterating over all on-disk entries */
1932  replication_dir = AllocateDir(PG_REPLSLOT_DIR);
1933  while ((replication_de = ReadDir(replication_dir, PG_REPLSLOT_DIR)) != NULL)
1934  {
1935  char path[MAXPGPATH + sizeof(PG_REPLSLOT_DIR)];
1936  PGFileType de_type;
1937 
1938  if (strcmp(replication_de->d_name, ".") == 0 ||
1939  strcmp(replication_de->d_name, "..") == 0)
1940  continue;
1941 
1942  snprintf(path, sizeof(path), "%s/%s", PG_REPLSLOT_DIR, replication_de->d_name);
1943  de_type = get_dirent_type(path, replication_de, false, DEBUG1);
1944 
1945  /* we're only creating directories here, skip if it's not our's */
1946  if (de_type != PGFILETYPE_ERROR && de_type != PGFILETYPE_DIR)
1947  continue;
1948 
1949  /* we crashed while a slot was being setup or deleted, clean up */
1950  if (pg_str_endswith(replication_de->d_name, ".tmp"))
1951  {
1952  if (!rmtree(path, true))
1953  {
1954  ereport(WARNING,
1955  (errmsg("could not remove directory \"%s\"",
1956  path)));
1957  continue;
1958  }
1960  continue;
1961  }
1962 
1963  /* looks like a slot in a normal state, restore */
1964  RestoreSlotFromDisk(replication_de->d_name);
1965  }
1966  FreeDir(replication_dir);
1967 
1968  /* currently no slots exist, we're done. */
1969  if (max_replication_slots <= 0)
1970  return;
1971 
1972  /* Now that we have recovered all the data, compute replication xmin */
1975 }
#define WARNING
Definition: elog.h:36
struct dirent * ReadDir(DIR *dir, const char *dirname)
Definition: fd.c:2931
int FreeDir(DIR *dir)
Definition: fd.c:2983
void fsync_fname(const char *fname, bool isdir)
Definition: fd.c:755
DIR * AllocateDir(const char *dirname)
Definition: fd.c:2865
PGFileType get_dirent_type(const char *path, const struct dirent *de, bool look_through_symlinks, int elevel)
Definition: file_utils.c:526
PGFileType
Definition: file_utils.h:19
@ PGFILETYPE_DIR
Definition: file_utils.h:23
@ PGFILETYPE_ERROR
Definition: file_utils.h:20
bool rmtree(const char *path, bool rmtopdir)
Definition: rmtree.c:50
static void RestoreSlotFromDisk(const char *name)
Definition: slot.c:2201
bool pg_str_endswith(const char *str, const char *end)
Definition: string.c:31
Definition: dirent.c:26
Definition: dirent.h:10
char d_name[MAX_PATH]
Definition: dirent.h:15

References AllocateDir(), dirent::d_name, DEBUG1, elog, ereport, errmsg(), FreeDir(), fsync_fname(), get_dirent_type(), max_replication_slots, MAXPGPATH, PG_REPLSLOT_DIR, pg_str_endswith(), PGFILETYPE_DIR, PGFILETYPE_ERROR, ReadDir(), ReplicationSlotsComputeRequiredLSN(), ReplicationSlotsComputeRequiredXmin(), RestoreSlotFromDisk(), rmtree(), snprintf, and WARNING.

Referenced by StartupXLOG().

◆ WaitForStandbyConfirmation()

void WaitForStandbyConfirmation ( XLogRecPtr  wait_for_lsn)

Definition at line 2759 of file slot.c.

2760 {
2761  /*
2762  * Don't need to wait for the standby to catch up if the current acquired
2763  * slot is not a logical failover slot, or there is no value in
2764  * synchronized_standby_slots.
2765  */
2767  return;
2768 
2770 
2771  for (;;)
2772  {
2774 
2775  if (ConfigReloadPending)
2776  {
2777  ConfigReloadPending = false;
2779  }
2780 
2781  /* Exit if done waiting for every slot. */
2782  if (StandbySlotsHaveCaughtup(wait_for_lsn, WARNING))
2783  break;
2784 
2785  /*
2786  * Wait for the slots in the synchronized_standby_slots to catch up,
2787  * but use a timeout (1s) so we can also check if the
2788  * synchronized_standby_slots has been changed.
2789  */
2791  WAIT_EVENT_WAIT_FOR_STANDBY_CONFIRMATION);
2792  }
2793 
2795 }
bool ConditionVariableTimedSleep(ConditionVariable *cv, long timeout, uint32 wait_event_info)
@ PGC_SIGHUP
Definition: guc.h:71
void ProcessConfigFile(GucContext context)
volatile sig_atomic_t ConfigReloadPending
Definition: interrupt.c:27
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:122
bool StandbySlotsHaveCaughtup(XLogRecPtr wait_for_lsn, int elevel)
Definition: slot.c:2605
ConditionVariable wal_confirm_rcv_cv
WalSndCtlData * WalSndCtl
Definition: walsender.c:109

References CHECK_FOR_INTERRUPTS, ConditionVariableCancelSleep(), ConditionVariablePrepareToSleep(), ConditionVariableTimedSleep(), ConfigReloadPending, ReplicationSlot::data, ReplicationSlotPersistentData::failover, MyReplicationSlot, PGC_SIGHUP, ProcessConfigFile(), StandbySlotsHaveCaughtup(), synchronized_standby_slots_config, WalSndCtlData::wal_confirm_rcv_cv, WalSndCtl, and WARNING.

Referenced by LogicalSlotAdvanceAndCheckSnapState(), and pg_logical_slot_get_changes_guts().

Variable Documentation

◆ max_replication_slots

◆ MyReplicationSlot

PGDLLIMPORT ReplicationSlot* MyReplicationSlot
extern

Definition at line 138 of file slot.c.

Referenced by binary_upgrade_logical_slot_has_caught_up(), copy_replication_slot(), create_logical_replication_slot(), create_physical_replication_slot(), CreateDecodingContext(), CreateInitDecodingContext(), CreateReplicationSlot(), InvalidatePossiblyObsoleteSlot(), LogicalConfirmReceivedLocation(), LogicalIncreaseRestartDecodingForSlot(), LogicalIncreaseXminForSlot(), LogicalReplicationSlotHasPendingWal(), LogicalSlotAdvanceAndCheckSnapState(), NeedToWaitForStandbys(), pg_create_logical_replication_slot(), pg_create_physical_replication_slot(), pg_logical_slot_get_changes_guts(), pg_physical_replication_slot_advance(), pg_replication_slot_advance(), PhysicalConfirmReceivedLocation(), PhysicalReplicationSlotNewXmin(), PhysicalWakeupLogicalWalSnd(), PostgresMain(), ProcessStandbyHSFeedbackMessage(), ProcessStandbyReplyMessage(), ReorderBufferAllocate(), ReorderBufferFree(), ReorderBufferRestoreChanges(), ReorderBufferRestoreCleanup(), ReorderBufferSerializedPath(), ReorderBufferSerializeTXN(), ReplicationSlotAcquire(), ReplicationSlotAlter(), ReplicationSlotCleanup(), ReplicationSlotCreate(), ReplicationSlotDrop(), ReplicationSlotDropAcquired(), ReplicationSlotMarkDirty(), ReplicationSlotPersist(), ReplicationSlotRelease(), ReplicationSlotReserveWal(), ReplicationSlotSave(), ReplicationSlotsDropDBSlots(), ReplicationSlotShmemExit(), reserve_wal_for_local_slot(), slotsync_failure_callback(), slotsync_worker_onexit(), StartLogicalReplication(), StartReplication(), StartupDecodingContext(), synchronize_one_slot(), update_and_persist_local_synced_slot(), update_local_synced_slot(), WaitForStandbyConfirmation(), and WalSndErrorCleanup().

◆ ReplicationSlotCtl

◆ SlotInvalidationCauses

PGDLLIMPORT const char* const SlotInvalidationCauses[]
extern

◆ synchronized_standby_slots

PGDLLIMPORT char* synchronized_standby_slots
extern

Definition at line 148 of file slot.c.