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 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)
 
void ReplicationSlotPersist (void)
 
void ReplicationSlotDrop (const char *name, bool nowait)
 
void ReplicationSlotAcquire (const char *name, bool nowait)
 
void ReplicationSlotRelease (void)
 
void ReplicationSlotCleanup (void)
 
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)
 

Variables

PGDLLIMPORT ReplicationSlotCtlDataReplicationSlotCtl
 
PGDLLIMPORT ReplicationSlotMyReplicationSlot
 
PGDLLIMPORT int max_replication_slots
 

Macro Definition Documentation

◆ SlotIsLogical

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

Definition at line 191 of file slot.h.

◆ SlotIsPhysical

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

Definition at line 190 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 44 of file slot.h.

45 {
47  /* required WAL has been removed */
49  /* required rows have been removed */
51  /* wal_level insufficient for slot */
ReplicationSlotInvalidationCause
Definition: slot.h:45
@ RS_INVAL_WAL_REMOVED
Definition: slot.h:48
@ RS_INVAL_HORIZON
Definition: slot.h:50
@ RS_INVAL_WAL_LEVEL
Definition: slot.h:52
@ RS_INVAL_NONE
Definition: slot.h:46

◆ ReplicationSlotPersistency

Enumerator
RS_PERSISTENT 
RS_EPHEMERAL 
RS_TEMPORARY 

Definition at line 33 of file slot.h.

34 {
ReplicationSlotPersistency
Definition: slot.h:34
@ RS_PERSISTENT
Definition: slot.h:35
@ RS_EPHEMERAL
Definition: slot.h:36
@ RS_TEMPORARY
Definition: slot.h:37

Function Documentation

◆ CheckPointReplicationSlots()

void CheckPointReplicationSlots ( bool  is_shutdown)

Definition at line 1620 of file slot.c.

1621 {
1622  int i;
1623 
1624  elog(DEBUG1, "performing replication slot checkpoint");
1625 
1626  /*
1627  * Prevent any slot from being created/dropped while we're active. As we
1628  * explicitly do *not* want to block iterating over replication_slots or
1629  * acquiring a slot we cannot take the control lock - but that's OK,
1630  * because holding ReplicationSlotAllocationLock is strictly stronger, and
1631  * enough to guarantee that nobody can change the in_use bits on us.
1632  */
1633  LWLockAcquire(ReplicationSlotAllocationLock, LW_SHARED);
1634 
1635  for (i = 0; i < max_replication_slots; i++)
1636  {
1638  char path[MAXPGPATH];
1639 
1640  if (!s->in_use)
1641  continue;
1642 
1643  /* save the slot to disk, locking is handled in SaveSlotToPath() */
1644  sprintf(path, "pg_replslot/%s", NameStr(s->data.name));
1645 
1646  /*
1647  * Slot's data is not flushed each time the confirmed_flush LSN is
1648  * updated as that could lead to frequent writes. However, we decide
1649  * to force a flush of all logical slot's data at the time of shutdown
1650  * if the confirmed_flush LSN is changed since we last flushed it to
1651  * disk. This helps in avoiding an unnecessary retreat of the
1652  * confirmed_flush LSN after restart.
1653  */
1654  if (is_shutdown && SlotIsLogical(s))
1655  {
1656  SpinLockAcquire(&s->mutex);
1657 
1659 
1660  if (s->data.invalidated == RS_INVAL_NONE &&
1662  {
1663  s->just_dirtied = true;
1664  s->dirty = true;
1665  }
1666  SpinLockRelease(&s->mutex);
1667  }
1668 
1669  SaveSlotToPath(s, path, LOG);
1670  }
1671  LWLockRelease(ReplicationSlotAllocationLock);
1672 }
#define NameStr(name)
Definition: c.h:735
#define LOG
Definition: elog.h:31
#define DEBUG1
Definition: elog.h:30
int i
Definition: isn.c:73
Assert(fmt[strlen(fmt) - 1] !='\n')
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1195
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1808
@ LW_SHARED
Definition: lwlock.h:117
#define MAXPGPATH
#define sprintf
Definition: port.h:240
static void SaveSlotToPath(ReplicationSlot *slot, const char *dir, int elevel)
Definition: slot.c:1801
int max_replication_slots
Definition: slot.c:102
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:96
#define SlotIsLogical(slot)
Definition: slot.h:191
#define SpinLockRelease(lock)
Definition: spin.h:64
#define SpinLockAcquire(lock)
Definition: spin.h:62
ReplicationSlot replication_slots[1]
Definition: slot.h:202
XLogRecPtr confirmed_flush
Definition: slot.h:99
ReplicationSlotInvalidationCause invalidated
Definition: slot.h:91
slock_t mutex
Definition: slot.h:135
XLogRecPtr last_saved_confirmed_flush
Definition: slot.h:187
bool in_use
Definition: slot.h:138
bool just_dirtied
Definition: slot.h:144
bool dirty
Definition: slot.h:145
ReplicationSlotPersistentData data
Definition: slot.h:162

References Assert(), 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, ReplicationSlotCtlData::replication_slots, ReplicationSlotCtl, RS_INVAL_NONE, SaveSlotToPath(), SlotIsLogical, SpinLockAcquire, SpinLockRelease, and sprintf.

Referenced by CheckPointGuts().

◆ CheckSlotPermissions()

void CheckSlotPermissions ( void  )

Definition at line 1188 of file slot.c.

1189 {
1190  if (!has_rolreplication(GetUserId()))
1191  ereport(ERROR,
1192  (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
1193  errmsg("permission denied to use replication slots"),
1194  errdetail("Only roles with the %s attribute may use replication slots.",
1195  "REPLICATION")));
1196 }
int errdetail(const char *fmt,...)
Definition: elog.c:1202
int errcode(int sqlerrcode)
Definition: elog.c:858
int errmsg(const char *fmt,...)
Definition: elog.c:1069
#define ERROR
Definition: elog.h:39
#define ereport(elevel,...)
Definition: elog.h:149
Oid GetUserId(void)
Definition: miscinit.c:508
bool has_rolreplication(Oid roleid)
Definition: miscinit.c:705

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(), and pg_replication_slot_advance().

◆ CheckSlotRequirements()

void CheckSlotRequirements ( void  )

Definition at line 1166 of file slot.c.

1167 {
1168  /*
1169  * NB: Adding a new requirement likely means that RestoreSlotFromDisk()
1170  * needs the same check.
1171  */
1172 
1173  if (max_replication_slots == 0)
1174  ereport(ERROR,
1175  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1176  errmsg("replication slots can only be used if max_replication_slots > 0")));
1177 
1179  ereport(ERROR,
1180  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1181  errmsg("replication slots can only be used if wal_level >= replica")));
1182 }
int wal_level
Definition: xlog.c:134
@ WAL_LEVEL_REPLICA
Definition: xlog.h:73

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

◆ InvalidateObsoleteReplicationSlots()

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

Definition at line 1564 of file slot.c.

1567 {
1568  XLogRecPtr oldestLSN;
1569  bool invalidated = false;
1570 
1571  Assert(cause != RS_INVAL_HORIZON || TransactionIdIsValid(snapshotConflictHorizon));
1572  Assert(cause != RS_INVAL_WAL_REMOVED || oldestSegno > 0);
1573  Assert(cause != RS_INVAL_NONE);
1574 
1575  if (max_replication_slots == 0)
1576  return invalidated;
1577 
1578  XLogSegNoOffsetToRecPtr(oldestSegno, 0, wal_segment_size, oldestLSN);
1579 
1580 restart:
1581  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
1582  for (int i = 0; i < max_replication_slots; i++)
1583  {
1585 
1586  if (!s->in_use)
1587  continue;
1588 
1589  if (InvalidatePossiblyObsoleteSlot(cause, s, oldestLSN, dboid,
1590  snapshotConflictHorizon,
1591  &invalidated))
1592  {
1593  /* if the lock was released, start from scratch */
1594  goto restart;
1595  }
1596  }
1597  LWLockRelease(ReplicationSlotControlLock);
1598 
1599  /*
1600  * If any slots have been invalidated, recalculate the resource limits.
1601  */
1602  if (invalidated)
1603  {
1606  }
1607 
1608  return invalidated;
1609 }
void ReplicationSlotsComputeRequiredXmin(bool already_locked)
Definition: slot.c:867
void ReplicationSlotsComputeRequiredLSN(void)
Definition: slot.c:923
static bool InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause, ReplicationSlot *s, XLogRecPtr oldestLSN, Oid dboid, TransactionId snapshotConflictHorizon, bool *invalidated)
Definition: slot.c:1347
#define TransactionIdIsValid(xid)
Definition: transam.h:41
int wal_segment_size
Definition: xlog.c:146
#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 452 of file slot.c.

453 {
454  ReplicationSlot *s;
455  int active_pid;
456 
457  Assert(name != NULL);
458 
459 retry:
460  Assert(MyReplicationSlot == NULL);
461 
462  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
463 
464  /*
465  * Search for the slot with the specified name if the slot to acquire is
466  * not given. If the slot is not found, we either return -1 or error out.
467  */
468  s = SearchNamedReplicationSlot(name, false);
469  if (s == NULL || !s->in_use)
470  {
471  LWLockRelease(ReplicationSlotControlLock);
472 
473  ereport(ERROR,
474  (errcode(ERRCODE_UNDEFINED_OBJECT),
475  errmsg("replication slot \"%s\" does not exist",
476  name)));
477  }
478 
479  /*
480  * This is the slot we want; check if it's active under some other
481  * process. In single user mode, we don't need this check.
482  */
483  if (IsUnderPostmaster)
484  {
485  /*
486  * Get ready to sleep on the slot in case it is active. (We may end
487  * up not sleeping, but we don't want to do this while holding the
488  * spinlock.)
489  */
490  if (!nowait)
492 
493  SpinLockAcquire(&s->mutex);
494  if (s->active_pid == 0)
495  s->active_pid = MyProcPid;
496  active_pid = s->active_pid;
497  SpinLockRelease(&s->mutex);
498  }
499  else
500  active_pid = MyProcPid;
501  LWLockRelease(ReplicationSlotControlLock);
502 
503  /*
504  * If we found the slot but it's already active in another process, we
505  * wait until the owning process signals us that it's been released, or
506  * error out.
507  */
508  if (active_pid != MyProcPid)
509  {
510  if (!nowait)
511  {
512  /* Wait here until we get signaled, and then restart */
514  WAIT_EVENT_REPLICATION_SLOT_DROP);
516  goto retry;
517  }
518 
519  ereport(ERROR,
520  (errcode(ERRCODE_OBJECT_IN_USE),
521  errmsg("replication slot \"%s\" is active for PID %d",
522  NameStr(s->data.name), active_pid)));
523  }
524  else if (!nowait)
525  ConditionVariableCancelSleep(); /* no sleep needed after all */
526 
527  /* Let everybody know we've modified this slot */
529 
530  /* We made this slot active, so it's ours now. */
531  MyReplicationSlot = s;
532 
533  /*
534  * The call to pgstat_acquire_replslot() protects against stats for a
535  * different slot, from before a restart or such, being present during
536  * pgstat_report_replslot().
537  */
538  if (SlotIsLogical(s))
540 
541  if (am_walsender)
542  {
544  SlotIsLogical(s)
545  ? errmsg("acquired logical replication slot \"%s\"",
546  NameStr(s->data.name))
547  : errmsg("acquired physical replication slot \"%s\"",
548  NameStr(s->data.name)));
549  }
550 }
bool ConditionVariableCancelSleep(void)
void ConditionVariableBroadcast(ConditionVariable *cv)
void ConditionVariablePrepareToSleep(ConditionVariable *cv)
void ConditionVariableSleep(ConditionVariable *cv, uint32 wait_event_info)
int MyProcPid
Definition: globals.c:44
bool IsUnderPostmaster
Definition: globals.c:115
void pgstat_acquire_replslot(ReplicationSlot *slot)
ReplicationSlot * SearchNamedReplicationSlot(const char *name, bool need_lock)
Definition: slot.c:376
ReplicationSlot * MyReplicationSlot
Definition: slot.c:99
pid_t active_pid
Definition: slot.h:141
ConditionVariable active_cv
Definition: slot.h:168
const char * name
bool am_walsender
Definition: walsender.c:116
bool log_replication_commands
Definition: walsender.c:126

References ReplicationSlot::active_cv, ReplicationSlot::active_pid, am_walsender, Assert(), ConditionVariableBroadcast(), ConditionVariableCancelSleep(), ConditionVariablePrepareToSleep(), ConditionVariableSleep(), ReplicationSlot::data, DEBUG1, ereport, errcode(), errmsg(), ERROR, ReplicationSlot::in_use, 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(), pg_logical_slot_get_changes_guts(), pg_replication_slot_advance(), ReplicationSlotDrop(), StartLogicalReplication(), and StartReplication().

◆ ReplicationSlotCleanup()

void ReplicationSlotCleanup ( void  )

Definition at line 635 of file slot.c.

636 {
637  int i;
638 
639  Assert(MyReplicationSlot == NULL);
640 
641 restart:
642  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
643  for (i = 0; i < max_replication_slots; i++)
644  {
646 
647  if (!s->in_use)
648  continue;
649 
650  SpinLockAcquire(&s->mutex);
651  if (s->active_pid == MyProcPid)
652  {
654  SpinLockRelease(&s->mutex);
655  LWLockRelease(ReplicationSlotControlLock); /* avoid deadlock */
656 
658 
660  goto restart;
661  }
662  else
663  SpinLockRelease(&s->mutex);
664  }
665 
666  LWLockRelease(ReplicationSlotControlLock);
667 }
static void ReplicationSlotDropPtr(ReplicationSlot *slot)
Definition: slot.c:703
ReplicationSlotPersistency persistency
Definition: slot.h:69

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, and SpinLockRelease.

Referenced by PostgresMain(), ReplicationSlotShmemExit(), and WalSndErrorCleanup().

◆ ReplicationSlotCreate()

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

Definition at line 253 of file slot.c.

255 {
256  ReplicationSlot *slot = NULL;
257  int i;
258 
259  Assert(MyReplicationSlot == NULL);
260 
262 
263  /*
264  * If some other backend ran this code concurrently with us, we'd likely
265  * both allocate the same slot, and that would be bad. We'd also be at
266  * risk of missing a name collision. Also, we don't want to try to create
267  * a new slot while somebody's busy cleaning up an old one, because we
268  * might both be monkeying with the same directory.
269  */
270  LWLockAcquire(ReplicationSlotAllocationLock, LW_EXCLUSIVE);
271 
272  /*
273  * Check for name collision, and identify an allocatable slot. We need to
274  * hold ReplicationSlotControlLock in shared mode for this, so that nobody
275  * else can change the in_use flags while we're looking at them.
276  */
277  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
278  for (i = 0; i < max_replication_slots; i++)
279  {
281 
282  if (s->in_use && strcmp(name, NameStr(s->data.name)) == 0)
283  ereport(ERROR,
285  errmsg("replication slot \"%s\" already exists", name)));
286  if (!s->in_use && slot == NULL)
287  slot = s;
288  }
289  LWLockRelease(ReplicationSlotControlLock);
290 
291  /* If all slots are in use, we're out of luck. */
292  if (slot == NULL)
293  ereport(ERROR,
294  (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
295  errmsg("all replication slots are in use"),
296  errhint("Free one or increase max_replication_slots.")));
297 
298  /*
299  * Since this slot is not in use, nobody should be looking at any part of
300  * it other than the in_use field unless they're trying to allocate it.
301  * And since we hold ReplicationSlotAllocationLock, nobody except us can
302  * be doing that. So it's safe to initialize the slot.
303  */
304  Assert(!slot->in_use);
305  Assert(slot->active_pid == 0);
306 
307  /* first initialize persistent data */
308  memset(&slot->data, 0, sizeof(ReplicationSlotPersistentData));
309  namestrcpy(&slot->data.name, name);
310  slot->data.database = db_specific ? MyDatabaseId : InvalidOid;
311  slot->data.persistency = persistency;
312  slot->data.two_phase = two_phase;
314 
315  /* and then data only present in shared memory */
316  slot->just_dirtied = false;
317  slot->dirty = false;
325 
326  /*
327  * Create the slot on disk. We haven't actually marked the slot allocated
328  * yet, so no special cleanup is required if this errors out.
329  */
330  CreateSlotOnDisk(slot);
331 
332  /*
333  * We need to briefly prevent any other backend from iterating over the
334  * slots while we flip the in_use flag. We also need to set the active
335  * flag while holding the ControlLock as otherwise a concurrent
336  * ReplicationSlotAcquire() could acquire the slot as well.
337  */
338  LWLockAcquire(ReplicationSlotControlLock, LW_EXCLUSIVE);
339 
340  slot->in_use = true;
341 
342  /* We can now mark the slot active, and that makes it our slot. */
343  SpinLockAcquire(&slot->mutex);
344  Assert(slot->active_pid == 0);
345  slot->active_pid = MyProcPid;
346  SpinLockRelease(&slot->mutex);
347  MyReplicationSlot = slot;
348 
349  LWLockRelease(ReplicationSlotControlLock);
350 
351  /*
352  * Create statistics entry for the new logical slot. We don't collect any
353  * stats for physical slots, so no need to create an entry for the same.
354  * See ReplicationSlotDropPtr for why we need to do this before releasing
355  * ReplicationSlotAllocationLock.
356  */
357  if (SlotIsLogical(slot))
359 
360  /*
361  * Now that the slot has been marked as in_use and active, it's safe to
362  * let somebody else try to allocate a slot.
363  */
364  LWLockRelease(ReplicationSlotAllocationLock);
365 
366  /* Let everybody know we've modified this slot */
368 }
int errhint(const char *fmt,...)
Definition: elog.c:1316
Oid MyDatabaseId
Definition: globals.c:89
@ LW_EXCLUSIVE
Definition: lwlock.h:116
void namestrcpy(Name name, const char *str)
Definition: name.c:233
static bool two_phase
void pgstat_create_replslot(ReplicationSlot *slot)
#define InvalidOid
Definition: postgres_ext.h:36
static void CreateSlotOnDisk(ReplicationSlot *slot)
Definition: slot.c:1740
bool ReplicationSlotValidateName(const char *name, int elevel)
Definition: slot.c:199
#define ERRCODE_DUPLICATE_OBJECT
Definition: streamutil.c:32
XLogRecPtr candidate_xmin_lsn
Definition: slot.h:178
TransactionId effective_catalog_xmin
Definition: slot.h:159
XLogRecPtr candidate_restart_valid
Definition: slot.h:179
TransactionId effective_xmin
Definition: slot.h:158
XLogRecPtr candidate_restart_lsn
Definition: slot.h:180
TransactionId candidate_catalog_xmin
Definition: slot.h:177
#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, i, ReplicationSlot::in_use, InvalidOid, InvalidTransactionId, InvalidXLogRecPtr, 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(), ReplicationSlotCtlData::replication_slots, ReplicationSlotCtl, ReplicationSlotValidateName(), SlotIsLogical, SpinLockAcquire, SpinLockRelease, two_phase, ReplicationSlotPersistentData::two_phase, and ReplicationSlotPersistentData::two_phase_at.

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

◆ ReplicationSlotDrop()

void ReplicationSlotDrop ( const char *  name,
bool  nowait 
)

Definition at line 673 of file slot.c.

674 {
675  Assert(MyReplicationSlot == NULL);
676 
677  ReplicationSlotAcquire(name, nowait);
678 
680 }
void ReplicationSlotAcquire(const char *name, bool nowait)
Definition: slot.c:452
static void ReplicationSlotDropAcquired(void)
Definition: slot.c:686

References Assert(), MyReplicationSlot, name, ReplicationSlotAcquire(), and ReplicationSlotDropAcquired().

Referenced by DropReplicationSlot(), and pg_drop_replication_slot().

◆ ReplicationSlotDropAtPubNode()

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

Definition at line 1761 of file subscriptioncmds.c.

1762 {
1763  StringInfoData cmd;
1764 
1765  Assert(wrconn);
1766 
1767  load_file("libpqwalreceiver", false);
1768 
1769  initStringInfo(&cmd);
1770  appendStringInfo(&cmd, "DROP_REPLICATION_SLOT %s WAIT", quote_identifier(slotname));
1771 
1772  PG_TRY();
1773  {
1775 
1776  res = walrcv_exec(wrconn, cmd.data, 0, NULL);
1777 
1778  if (res->status == WALRCV_OK_COMMAND)
1779  {
1780  /* NOTICE. Success. */
1781  ereport(NOTICE,
1782  (errmsg("dropped replication slot \"%s\" on publisher",
1783  slotname)));
1784  }
1785  else if (res->status == WALRCV_ERROR &&
1786  missing_ok &&
1787  res->sqlstate == ERRCODE_UNDEFINED_OBJECT)
1788  {
1789  /* LOG. Error, but missing_ok = true. */
1790  ereport(LOG,
1791  (errmsg("could not drop replication slot \"%s\" on publisher: %s",
1792  slotname, res->err)));
1793  }
1794  else
1795  {
1796  /* ERROR. */
1797  ereport(ERROR,
1798  (errcode(ERRCODE_CONNECTION_FAILURE),
1799  errmsg("could not drop replication slot \"%s\" on publisher: %s",
1800  slotname, res->err)));
1801  }
1802 
1804  }
1805  PG_FINALLY();
1806  {
1807  pfree(cmd.data);
1808  }
1809  PG_END_TRY();
1810 }
void load_file(const char *filename, bool restricted)
Definition: dfmgr.c:144
#define PG_TRY(...)
Definition: elog.h:370
#define PG_END_TRY(...)
Definition: elog.h:395
#define NOTICE
Definition: elog.h:35
#define PG_FINALLY(...)
Definition: elog.h:387
void pfree(void *pointer)
Definition: mcxt.c:1456
const char * quote_identifier(const char *ident)
Definition: ruleutils.c:11975
void appendStringInfo(StringInfo str, const char *fmt,...)
Definition: stringinfo.c:97
void initStringInfo(StringInfo str)
Definition: stringinfo.c:59
static WalReceiverConn * wrconn
Definition: walreceiver.c:95
@ WALRCV_OK_COMMAND
Definition: walreceiver.h:204
@ WALRCV_ERROR
Definition: walreceiver.h:203
static void walrcv_clear_result(WalRcvExecResult *walres)
Definition: walreceiver.h:441
#define walrcv_exec(conn, exec, nRetTypes, retTypes)
Definition: walreceiver.h:435

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 171 of file slot.c.

172 {
174 }
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:180

References before_shmem_exit(), and ReplicationSlotShmemExit().

Referenced by BaseInit().

◆ ReplicationSlotMarkDirty()

◆ ReplicationSlotName()

bool ReplicationSlotName ( int  index,
Name  name 
)

Definition at line 425 of file slot.c.

426 {
427  ReplicationSlot *slot;
428  bool found;
429 
431 
432  /*
433  * Ensure that the slot cannot be dropped while we copy the name. Don't
434  * need the spinlock as the name of an existing slot cannot change.
435  */
436  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
437  found = slot->in_use;
438  if (slot->in_use)
439  namestrcpy(name, NameStr(slot->data.name));
440  LWLockRelease(ReplicationSlotControlLock);
441 
442  return found;
443 }
Definition: type.h:95

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 1250 of file tablesync.c.

1252 {
1253  snprintf(syncslotname, szslot, "pg_%u_sync_%u_" UINT64_FORMAT, suboid,
1254  relid, GetSystemIdentifier());
1255 }
#define UINT64_FORMAT
Definition: c.h:538
#define snprintf
Definition: port.h:238
uint64 GetSystemIdentifier(void)
Definition: xlog.c:4294

References GetSystemIdentifier(), snprintf, and UINT64_FORMAT.

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

◆ ReplicationSlotPersist()

void ReplicationSlotPersist ( void  )

◆ ReplicationSlotRelease()

void ReplicationSlotRelease ( void  )

Definition at line 559 of file slot.c.

560 {
562  char *slotname = NULL; /* keep compiler quiet */
563  bool is_logical = false; /* keep compiler quiet */
564 
565  Assert(slot != NULL && slot->active_pid != 0);
566 
567  if (am_walsender)
568  {
569  slotname = pstrdup(NameStr(slot->data.name));
570  is_logical = SlotIsLogical(slot);
571  }
572 
573  if (slot->data.persistency == RS_EPHEMERAL)
574  {
575  /*
576  * Delete the slot. There is no !PANIC case where this is allowed to
577  * fail, all that may happen is an incomplete cleanup of the on-disk
578  * data.
579  */
581  }
582 
583  /*
584  * If slot needed to temporarily restrain both data and catalog xmin to
585  * create the catalog snapshot, remove that temporary constraint.
586  * Snapshots can only be exported while the initial snapshot is still
587  * acquired.
588  */
589  if (!TransactionIdIsValid(slot->data.xmin) &&
591  {
592  SpinLockAcquire(&slot->mutex);
594  SpinLockRelease(&slot->mutex);
596  }
597 
598  if (slot->data.persistency == RS_PERSISTENT)
599  {
600  /*
601  * Mark persistent slot inactive. We're not freeing it, just
602  * disconnecting, but wake up others that may be waiting for it.
603  */
604  SpinLockAcquire(&slot->mutex);
605  slot->active_pid = 0;
606  SpinLockRelease(&slot->mutex);
608  }
609 
610  MyReplicationSlot = NULL;
611 
612  /* might not have been set when we've been a plain slot */
613  LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
616  LWLockRelease(ProcArrayLock);
617 
618  if (am_walsender)
619  {
621  is_logical
622  ? errmsg("released logical replication slot \"%s\"",
623  slotname)
624  : errmsg("released physical replication slot \"%s\"",
625  slotname));
626 
627  pfree(slotname);
628  }
629 }
char * pstrdup(const char *in)
Definition: mcxt.c:1644
#define PROC_IN_LOGICAL_DECODING
Definition: proc.h:60
PGPROC * MyProc
Definition: proc.c:66
PROC_HDR * ProcGlobal
Definition: proc.c:78
uint8 statusFlags
Definition: proc.h:233
int pgxactoff
Definition: proc.h:188
uint8 * statusFlags
Definition: proc.h:377
TransactionId xmin
Definition: slot.h:77

References ReplicationSlot::active_cv, ReplicationSlot::active_pid, am_walsender, Assert(), ConditionVariableBroadcast(), ReplicationSlot::data, DEBUG1, ReplicationSlot::effective_xmin, ereport, errmsg(), InvalidTransactionId, LOG, log_replication_commands, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), ReplicationSlot::mutex, MyProc, MyReplicationSlot, ReplicationSlotPersistentData::name, NameStr, 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(), ReplicationSlotShmemExit(), StartLogicalReplication(), StartReplication(), and WalSndErrorCleanup().

◆ ReplicationSlotReserveWal()

void ReplicationSlotReserveWal ( void  )

Definition at line 1205 of file slot.c.

1206 {
1208 
1209  Assert(slot != NULL);
1211 
1212  /*
1213  * The replication slot mechanism is used to prevent removal of required
1214  * WAL. As there is no interlock between this routine and checkpoints, WAL
1215  * segments could concurrently be removed when a now stale return value of
1216  * ReplicationSlotsComputeRequiredLSN() is used. In the unlikely case that
1217  * this happens we'll just retry.
1218  */
1219  while (true)
1220  {
1221  XLogSegNo segno;
1222  XLogRecPtr restart_lsn;
1223 
1224  /*
1225  * For logical slots log a standby snapshot and start logical decoding
1226  * at exactly that position. That allows the slot to start up more
1227  * quickly. But on a standby we cannot do WAL writes, so just use the
1228  * replay pointer; effectively, an attempt to create a logical slot on
1229  * standby will cause it to wait for an xl_running_xact record to be
1230  * logged independently on the primary, so that a snapshot can be
1231  * built using the record.
1232  *
1233  * None of this is needed (or indeed helpful) for physical slots as
1234  * they'll start replay at the last logged checkpoint anyway. Instead
1235  * return the location of the last redo LSN. While that slightly
1236  * increases the chance that we have to retry, it's where a base
1237  * backup has to start replay at.
1238  */
1239  if (SlotIsPhysical(slot))
1240  restart_lsn = GetRedoRecPtr();
1241  else if (RecoveryInProgress())
1242  restart_lsn = GetXLogReplayRecPtr(NULL);
1243  else
1244  restart_lsn = GetXLogInsertRecPtr();
1245 
1246  SpinLockAcquire(&slot->mutex);
1247  slot->data.restart_lsn = restart_lsn;
1248  SpinLockRelease(&slot->mutex);
1249 
1250  /* prevent WAL removal as fast as possible */
1252 
1253  /*
1254  * If all required WAL is still there, great, otherwise retry. The
1255  * slot should prevent further removal of WAL, unless there's a
1256  * concurrent ReplicationSlotsComputeRequiredLSN() after we've written
1257  * the new restart_lsn above, so normally we should never need to loop
1258  * more than twice.
1259  */
1261  if (XLogGetLastRemovedSegno() < segno)
1262  break;
1263  }
1264 
1265  if (!RecoveryInProgress() && SlotIsLogical(slot))
1266  {
1267  XLogRecPtr flushptr;
1268 
1269  /* make sure we have enough information to start */
1270  flushptr = LogStandbySnapshot();
1271 
1272  /* and make sure it's fsynced to disk */
1273  XLogFlush(flushptr);
1274  }
1275 }
#define SlotIsPhysical(slot)
Definition: slot.h:190
XLogRecPtr LogStandbySnapshot(void)
Definition: standby.c:1287
XLogRecPtr restart_lsn
Definition: slot.h:88
bool RecoveryInProgress(void)
Definition: xlog.c:6039
XLogSegNo XLogGetLastRemovedSegno(void)
Definition: xlog.c:3581
XLogRecPtr GetRedoRecPtr(void)
Definition: xlog.c:6142
XLogRecPtr GetXLogInsertRecPtr(void)
Definition: xlog.c:9067
void XLogFlush(XLogRecPtr record)
Definition: xlog.c:2621
#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 972 of file slot.c.

973 {
974  XLogRecPtr result = InvalidXLogRecPtr;
975  int i;
976 
977  if (max_replication_slots <= 0)
978  return InvalidXLogRecPtr;
979 
980  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
981 
982  for (i = 0; i < max_replication_slots; i++)
983  {
984  ReplicationSlot *s;
985  XLogRecPtr restart_lsn;
986  bool invalidated;
987 
989 
990  /* cannot change while ReplicationSlotCtlLock is held */
991  if (!s->in_use)
992  continue;
993 
994  /* we're only interested in logical slots */
995  if (!SlotIsLogical(s))
996  continue;
997 
998  /* read once, it's ok if it increases while we're checking */
999  SpinLockAcquire(&s->mutex);
1000  restart_lsn = s->data.restart_lsn;
1001  invalidated = s->data.invalidated != RS_INVAL_NONE;
1002  SpinLockRelease(&s->mutex);
1003 
1004  /* invalidated slots need not apply */
1005  if (invalidated)
1006  continue;
1007 
1008  if (restart_lsn == InvalidXLogRecPtr)
1009  continue;
1010 
1011  if (result == InvalidXLogRecPtr ||
1012  restart_lsn < result)
1013  result = restart_lsn;
1014  }
1015 
1016  LWLockRelease(ReplicationSlotControlLock);
1017 
1018  return result;
1019 }

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 923 of file slot.c.

924 {
925  int i;
926  XLogRecPtr min_required = InvalidXLogRecPtr;
927 
928  Assert(ReplicationSlotCtl != NULL);
929 
930  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
931  for (i = 0; i < max_replication_slots; i++)
932  {
934  XLogRecPtr restart_lsn;
935  bool invalidated;
936 
937  if (!s->in_use)
938  continue;
939 
940  SpinLockAcquire(&s->mutex);
941  restart_lsn = s->data.restart_lsn;
942  invalidated = s->data.invalidated != RS_INVAL_NONE;
943  SpinLockRelease(&s->mutex);
944 
945  /* invalidated slots need not apply */
946  if (invalidated)
947  continue;
948 
949  if (restart_lsn != InvalidXLogRecPtr &&
950  (min_required == InvalidXLogRecPtr ||
951  restart_lsn < min_required))
952  min_required = restart_lsn;
953  }
954  LWLockRelease(ReplicationSlotControlLock);
955 
956  XLogSetReplicationSlotMinimumLSN(min_required);
957 }
void XLogSetReplicationSlotMinimumLSN(XLogRecPtr lsn)
Definition: xlog.c:2507

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(), and StartupReplicationSlots().

◆ ReplicationSlotsComputeRequiredXmin()

void ReplicationSlotsComputeRequiredXmin ( bool  already_locked)

Definition at line 867 of file slot.c.

868 {
869  int i;
871  TransactionId agg_catalog_xmin = InvalidTransactionId;
872 
873  Assert(ReplicationSlotCtl != NULL);
874 
875  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
876 
877  for (i = 0; i < max_replication_slots; i++)
878  {
880  TransactionId effective_xmin;
881  TransactionId effective_catalog_xmin;
882  bool invalidated;
883 
884  if (!s->in_use)
885  continue;
886 
887  SpinLockAcquire(&s->mutex);
888  effective_xmin = s->effective_xmin;
889  effective_catalog_xmin = s->effective_catalog_xmin;
890  invalidated = s->data.invalidated != RS_INVAL_NONE;
891  SpinLockRelease(&s->mutex);
892 
893  /* invalidated slots need not apply */
894  if (invalidated)
895  continue;
896 
897  /* check the data xmin */
898  if (TransactionIdIsValid(effective_xmin) &&
899  (!TransactionIdIsValid(agg_xmin) ||
900  TransactionIdPrecedes(effective_xmin, agg_xmin)))
901  agg_xmin = effective_xmin;
902 
903  /* check the catalog xmin */
904  if (TransactionIdIsValid(effective_catalog_xmin) &&
905  (!TransactionIdIsValid(agg_catalog_xmin) ||
906  TransactionIdPrecedes(effective_catalog_xmin, agg_catalog_xmin)))
907  agg_catalog_xmin = effective_catalog_xmin;
908  }
909 
910  LWLockRelease(ReplicationSlotControlLock);
911 
912  ProcArraySetReplicationSlotXmin(agg_xmin, agg_catalog_xmin, already_locked);
913 }
uint32 TransactionId
Definition: c.h:641
void ProcArraySetReplicationSlotXmin(TransactionId xmin, TransactionId catalog_xmin, bool already_locked)
Definition: procarray.c:3846
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(), and StartupReplicationSlots().

◆ ReplicationSlotsCountDBSlots()

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

Definition at line 1030 of file slot.c.

1031 {
1032  int i;
1033 
1034  *nslots = *nactive = 0;
1035 
1036  if (max_replication_slots <= 0)
1037  return false;
1038 
1039  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
1040  for (i = 0; i < max_replication_slots; i++)
1041  {
1042  ReplicationSlot *s;
1043 
1045 
1046  /* cannot change while ReplicationSlotCtlLock is held */
1047  if (!s->in_use)
1048  continue;
1049 
1050  /* only logical slots are database specific, skip */
1051  if (!SlotIsLogical(s))
1052  continue;
1053 
1054  /* not our database, skip */
1055  if (s->data.database != dboid)
1056  continue;
1057 
1058  /* NB: intentionally counting invalidated slots */
1059 
1060  /* count slots with spinlock held */
1061  SpinLockAcquire(&s->mutex);
1062  (*nslots)++;
1063  if (s->active_pid != 0)
1064  (*nactive)++;
1065  SpinLockRelease(&s->mutex);
1066  }
1067  LWLockRelease(ReplicationSlotControlLock);
1068 
1069  if (*nslots > 0)
1070  return true;
1071  return false;
1072 }

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 1088 of file slot.c.

1089 {
1090  int i;
1091 
1092  if (max_replication_slots <= 0)
1093  return;
1094 
1095 restart:
1096  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
1097  for (i = 0; i < max_replication_slots; i++)
1098  {
1099  ReplicationSlot *s;
1100  char *slotname;
1101  int active_pid;
1102 
1104 
1105  /* cannot change while ReplicationSlotCtlLock is held */
1106  if (!s->in_use)
1107  continue;
1108 
1109  /* only logical slots are database specific, skip */
1110  if (!SlotIsLogical(s))
1111  continue;
1112 
1113  /* not our database, skip */
1114  if (s->data.database != dboid)
1115  continue;
1116 
1117  /* NB: intentionally including invalidated slots */
1118 
1119  /* acquire slot, so ReplicationSlotDropAcquired can be reused */
1120  SpinLockAcquire(&s->mutex);
1121  /* can't change while ReplicationSlotControlLock is held */
1122  slotname = NameStr(s->data.name);
1123  active_pid = s->active_pid;
1124  if (active_pid == 0)
1125  {
1126  MyReplicationSlot = s;
1127  s->active_pid = MyProcPid;
1128  }
1129  SpinLockRelease(&s->mutex);
1130 
1131  /*
1132  * Even though we hold an exclusive lock on the database object a
1133  * logical slot for that DB can still be active, e.g. if it's
1134  * concurrently being dropped by a backend connected to another DB.
1135  *
1136  * That's fairly unlikely in practice, so we'll just bail out.
1137  */
1138  if (active_pid)
1139  ereport(ERROR,
1140  (errcode(ERRCODE_OBJECT_IN_USE),
1141  errmsg("replication slot \"%s\" is active for PID %d",
1142  slotname, active_pid)));
1143 
1144  /*
1145  * To avoid duplicating ReplicationSlotDropAcquired() and to avoid
1146  * holding ReplicationSlotControlLock over filesystem operations,
1147  * release ReplicationSlotControlLock and use
1148  * ReplicationSlotDropAcquired.
1149  *
1150  * As that means the set of slots could change, restart scan from the
1151  * beginning each time we release the lock.
1152  */
1153  LWLockRelease(ReplicationSlotControlLock);
1155  goto restart;
1156  }
1157  LWLockRelease(ReplicationSlotControlLock);
1158 }

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 136 of file slot.c.

137 {
138  bool found;
139 
140  if (max_replication_slots == 0)
141  return;
142 
144  ShmemInitStruct("ReplicationSlot Ctl", ReplicationSlotsShmemSize(),
145  &found);
146 
147  if (!found)
148  {
149  int i;
150 
151  /* First time through, so initialize */
153 
154  for (i = 0; i < max_replication_slots; i++)
155  {
157 
158  /* everything else is zeroed by the memset above */
159  SpinLockInit(&slot->mutex);
163  }
164  }
165 }
#define MemSet(start, val, len)
Definition: c.h:1009
void ConditionVariableInit(ConditionVariable *cv)
void LWLockInitialize(LWLock *lock, int tranche_id)
Definition: lwlock.c:730
@ LWTRANCHE_REPLICATION_SLOT_IO
Definition: lwlock.h:191
void * ShmemInitStruct(const char *name, Size size, bool *foundPtr)
Definition: shmem.c:396
Size ReplicationSlotsShmemSize(void)
Definition: slot.c:118
#define SpinLockInit(lock)
Definition: spin.h:60
LWLock io_in_progress_lock
Definition: slot.h:165

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 118 of file slot.c.

119 {
120  Size size = 0;
121 
122  if (max_replication_slots == 0)
123  return size;
124 
125  size = offsetof(ReplicationSlotCtlData, replication_slots);
126  size = add_size(size,
128 
129  return size;
130 }
size_t Size
Definition: c.h:594
Size add_size(Size s1, Size s2)
Definition: shmem.c:502
Size mul_size(Size s1, Size s2)
Definition: shmem.c:519

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

Referenced by CalculateShmemSize(), and ReplicationSlotsShmemInit().

◆ ReplicationSlotValidateName()

bool ReplicationSlotValidateName ( const char *  name,
int  elevel 
)

Definition at line 199 of file slot.c.

200 {
201  const char *cp;
202 
203  if (strlen(name) == 0)
204  {
205  ereport(elevel,
206  (errcode(ERRCODE_INVALID_NAME),
207  errmsg("replication slot name \"%s\" is too short",
208  name)));
209  return false;
210  }
211 
212  if (strlen(name) >= NAMEDATALEN)
213  {
214  ereport(elevel,
215  (errcode(ERRCODE_NAME_TOO_LONG),
216  errmsg("replication slot name \"%s\" is too long",
217  name)));
218  return false;
219  }
220 
221  for (cp = name; *cp; cp++)
222  {
223  if (!((*cp >= 'a' && *cp <= 'z')
224  || (*cp >= '0' && *cp <= '9')
225  || (*cp == '_')))
226  {
227  ereport(elevel,
228  (errcode(ERRCODE_INVALID_NAME),
229  errmsg("replication slot name \"%s\" contains invalid character",
230  name),
231  errhint("Replication slot names may only contain lower case letters, numbers, and the underscore character.")));
232  return false;
233  }
234  }
235  return true;
236 }
#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 376 of file slot.c.

377 {
378  int i;
379  ReplicationSlot *slot = NULL;
380 
381  if (need_lock)
382  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
383 
384  for (i = 0; i < max_replication_slots; i++)
385  {
387 
388  if (s->in_use && strcmp(name, NameStr(s->data.name)) == 0)
389  {
390  slot = s;
391  break;
392  }
393  }
394 
395  if (need_lock)
396  LWLockRelease(ReplicationSlotControlLock);
397 
398  return slot;
399 }

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(), and ReplicationSlotAcquire().

◆ StartupReplicationSlots()

void StartupReplicationSlots ( void  )

Definition at line 1679 of file slot.c.

1680 {
1681  DIR *replication_dir;
1682  struct dirent *replication_de;
1683 
1684  elog(DEBUG1, "starting up replication slots");
1685 
1686  /* restore all slots by iterating over all on-disk entries */
1687  replication_dir = AllocateDir("pg_replslot");
1688  while ((replication_de = ReadDir(replication_dir, "pg_replslot")) != NULL)
1689  {
1690  char path[MAXPGPATH + 12];
1691  PGFileType de_type;
1692 
1693  if (strcmp(replication_de->d_name, ".") == 0 ||
1694  strcmp(replication_de->d_name, "..") == 0)
1695  continue;
1696 
1697  snprintf(path, sizeof(path), "pg_replslot/%s", replication_de->d_name);
1698  de_type = get_dirent_type(path, replication_de, false, DEBUG1);
1699 
1700  /* we're only creating directories here, skip if it's not our's */
1701  if (de_type != PGFILETYPE_ERROR && de_type != PGFILETYPE_DIR)
1702  continue;
1703 
1704  /* we crashed while a slot was being setup or deleted, clean up */
1705  if (pg_str_endswith(replication_de->d_name, ".tmp"))
1706  {
1707  if (!rmtree(path, true))
1708  {
1709  ereport(WARNING,
1710  (errmsg("could not remove directory \"%s\"",
1711  path)));
1712  continue;
1713  }
1714  fsync_fname("pg_replslot", true);
1715  continue;
1716  }
1717 
1718  /* looks like a slot in a normal state, restore */
1719  RestoreSlotFromDisk(replication_de->d_name);
1720  }
1721  FreeDir(replication_dir);
1722 
1723  /* currently no slots exist, we're done. */
1724  if (max_replication_slots <= 0)
1725  return;
1726 
1727  /* Now that we have recovered all the data, compute replication xmin */
1730 }
#define WARNING
Definition: elog.h:36
struct dirent * ReadDir(DIR *dir, const char *dirname)
Definition: fd.c:2879
int FreeDir(DIR *dir)
Definition: fd.c:2931
void fsync_fname(const char *fname, bool isdir)
Definition: fd.c:733
DIR * AllocateDir(const char *dirname)
Definition: fd.c:2813
PGFileType get_dirent_type(const char *path, const struct dirent *de, bool look_through_symlinks, int elevel)
Definition: file_utils.c:525
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:1956
bool pg_str_endswith(const char *str, const char *end)
Definition: string.c:32
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_str_endswith(), PGFILETYPE_DIR, PGFILETYPE_ERROR, ReadDir(), ReplicationSlotsComputeRequiredLSN(), ReplicationSlotsComputeRequiredXmin(), RestoreSlotFromDisk(), rmtree(), snprintf, and WARNING.

Referenced by StartupXLOG().

Variable Documentation

◆ max_replication_slots

◆ MyReplicationSlot

◆ ReplicationSlotCtl