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 struct ReplicationSlotPersistentData ReplicationSlotPersistentData
 
typedef struct ReplicationSlot ReplicationSlot
 
typedef struct ReplicationSlotCtlData ReplicationSlotCtlData
 

Enumerations

enum  ReplicationSlotPersistency { RS_PERSISTENT , RS_EPHEMERAL , RS_TEMPORARY }
 

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 (XLogSegNo oldestSegno)
 
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 (void)
 
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 169 of file slot.h.

◆ SlotIsPhysical

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

Definition at line 168 of file slot.h.

Typedef Documentation

◆ ReplicationSlot

◆ ReplicationSlotCtlData

◆ ReplicationSlotPersistency

◆ ReplicationSlotPersistentData

Enumeration Type Documentation

◆ 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 ( void  )

Definition at line 1442 of file slot.c.

1443 {
1444  int i;
1445 
1446  elog(DEBUG1, "performing replication slot checkpoint");
1447 
1448  /*
1449  * Prevent any slot from being created/dropped while we're active. As we
1450  * explicitly do *not* want to block iterating over replication_slots or
1451  * acquiring a slot we cannot take the control lock - but that's OK,
1452  * because holding ReplicationSlotAllocationLock is strictly stronger, and
1453  * enough to guarantee that nobody can change the in_use bits on us.
1454  */
1455  LWLockAcquire(ReplicationSlotAllocationLock, LW_SHARED);
1456 
1457  for (i = 0; i < max_replication_slots; i++)
1458  {
1460  char path[MAXPGPATH];
1461 
1462  if (!s->in_use)
1463  continue;
1464 
1465  /* save the slot to disk, locking is handled in SaveSlotToPath() */
1466  sprintf(path, "pg_replslot/%s", NameStr(s->data.name));
1467  SaveSlotToPath(s, path, LOG);
1468  }
1469  LWLockRelease(ReplicationSlotAllocationLock);
1470 }
#define NameStr(name)
Definition: c.h:730
#define LOG
Definition: elog.h:31
#define DEBUG1
Definition: elog.h:30
int i
Definition: isn.c:73
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1195
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1803
@ LW_SHARED
Definition: lwlock.h:116
#define MAXPGPATH
#define sprintf
Definition: port.h:240
static void SaveSlotToPath(ReplicationSlot *slot, const char *dir, int elevel)
Definition: slot.c:1599
int max_replication_slots
Definition: slot.c:101
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:95
ReplicationSlot replication_slots[1]
Definition: slot.h:180
bool in_use
Definition: slot.h:123
ReplicationSlotPersistentData data
Definition: slot.h:147

References ReplicationSlot::data, DEBUG1, elog(), i, ReplicationSlot::in_use, LOG, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_replication_slots, MAXPGPATH, ReplicationSlotPersistentData::name, NameStr, ReplicationSlotCtlData::replication_slots, ReplicationSlotCtl, SaveSlotToPath(), and sprintf.

Referenced by CheckPointGuts().

◆ CheckSlotPermissions()

void CheckSlotPermissions ( void  )

Definition at line 1141 of file slot.c.

1142 {
1143  if (!superuser() && !has_rolreplication(GetUserId()))
1144  ereport(ERROR,
1145  (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
1146  errmsg("must be superuser or replication role to use replication slots")));
1147 }
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:502
bool has_rolreplication(Oid roleid)
Definition: miscinit.c:699
bool superuser(void)
Definition: superuser.c:46

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

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

1120 {
1121  /*
1122  * NB: Adding a new requirement likely means that RestoreSlotFromDisk()
1123  * needs the same check.
1124  */
1125 
1126  if (max_replication_slots == 0)
1127  ereport(ERROR,
1128  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1129  errmsg("replication slots can only be used if max_replication_slots > 0")));
1130 
1132  ereport(ERROR,
1133  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1134  errmsg("replication slots can only be used if wal_level >= replica")));
1135 }
int wal_level
Definition: xlog.c:134
@ WAL_LEVEL_REPLICA
Definition: xlog.h:70

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 ( XLogSegNo  oldestSegno)

Definition at line 1399 of file slot.c.

1400 {
1401  XLogRecPtr oldestLSN;
1402  bool invalidated = false;
1403 
1404  XLogSegNoOffsetToRecPtr(oldestSegno, 0, wal_segment_size, oldestLSN);
1405 
1406 restart:
1407  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
1408  for (int i = 0; i < max_replication_slots; i++)
1409  {
1411 
1412  if (!s->in_use)
1413  continue;
1414 
1415  if (InvalidatePossiblyObsoleteSlot(s, oldestLSN, &invalidated))
1416  {
1417  /* if the lock was released, start from scratch */
1418  goto restart;
1419  }
1420  }
1421  LWLockRelease(ReplicationSlotControlLock);
1422 
1423  /*
1424  * If any slots have been invalidated, recalculate the resource limits.
1425  */
1426  if (invalidated)
1427  {
1430  }
1431 
1432  return invalidated;
1433 }
void ReplicationSlotsComputeRequiredXmin(bool already_locked)
Definition: slot.c:835
void ReplicationSlotsComputeRequiredLSN(void)
Definition: slot.c:892
static bool InvalidatePossiblyObsoleteSlot(ReplicationSlot *s, XLogRecPtr oldestLSN, bool *invalidated)
Definition: slot.c:1239
int wal_segment_size
Definition: xlog.c:146
#define XLogSegNoOffsetToRecPtr(segno, offset, wal_segsz_bytes, dest)
uint64 XLogRecPtr
Definition: xlogdefs.h:21

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

Referenced by CreateCheckPoint(), and CreateRestartPoint().

◆ ReplicationSlotAcquire()

void ReplicationSlotAcquire ( const char *  name,
bool  nowait 
)

Definition at line 450 of file slot.c.

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

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

Referenced by pg_logical_slot_get_changes_guts(), pg_replication_slot_advance(), ReplicationSlotDrop(), StartLogicalReplication(), and StartReplication().

◆ ReplicationSlotCleanup()

void ReplicationSlotCleanup ( void  )

Definition at line 603 of file slot.c.

604 {
605  int i;
606 
607  Assert(MyReplicationSlot == NULL);
608 
609 restart:
610  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
611  for (i = 0; i < max_replication_slots; i++)
612  {
614 
615  if (!s->in_use)
616  continue;
617 
618  SpinLockAcquire(&s->mutex);
619  if (s->active_pid == MyProcPid)
620  {
622  SpinLockRelease(&s->mutex);
623  LWLockRelease(ReplicationSlotControlLock); /* avoid deadlock */
624 
626 
628  goto restart;
629  }
630  else
631  SpinLockRelease(&s->mutex);
632  }
633 
634  LWLockRelease(ReplicationSlotControlLock);
635 }
static void ReplicationSlotDropPtr(ReplicationSlot *slot)
Definition: slot.c:671
ReplicationSlotPersistency persistency
Definition: slot.h:54

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

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

642 {
643  Assert(MyReplicationSlot == NULL);
644 
645  ReplicationSlotAcquire(name, nowait);
646 
648 }
void ReplicationSlotAcquire(const char *name, bool nowait)
Definition: slot.c:450
static void ReplicationSlotDropAcquired(void)
Definition: slot.c:654

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 1662 of file subscriptioncmds.c.

1663 {
1664  StringInfoData cmd;
1665 
1666  Assert(wrconn);
1667 
1668  load_file("libpqwalreceiver", false);
1669 
1670  initStringInfo(&cmd);
1671  appendStringInfo(&cmd, "DROP_REPLICATION_SLOT %s WAIT", quote_identifier(slotname));
1672 
1673  PG_TRY();
1674  {
1676 
1677  res = walrcv_exec(wrconn, cmd.data, 0, NULL);
1678 
1679  if (res->status == WALRCV_OK_COMMAND)
1680  {
1681  /* NOTICE. Success. */
1682  ereport(NOTICE,
1683  (errmsg("dropped replication slot \"%s\" on publisher",
1684  slotname)));
1685  }
1686  else if (res->status == WALRCV_ERROR &&
1687  missing_ok &&
1688  res->sqlstate == ERRCODE_UNDEFINED_OBJECT)
1689  {
1690  /* LOG. Error, but missing_ok = true. */
1691  ereport(LOG,
1692  (errmsg("could not drop replication slot \"%s\" on publisher: %s",
1693  slotname, res->err)));
1694  }
1695  else
1696  {
1697  /* ERROR. */
1698  ereport(ERROR,
1699  (errcode(ERRCODE_CONNECTION_FAILURE),
1700  errmsg("could not drop replication slot \"%s\" on publisher: %s",
1701  slotname, res->err)));
1702  }
1703 
1705  }
1706  PG_FINALLY();
1707  {
1708  pfree(cmd.data);
1709  }
1710  PG_END_TRY();
1711 }
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:1436
const char * quote_identifier(const char *ident)
Definition: ruleutils.c:11529
void appendStringInfo(StringInfo str, const char *fmt,...)
Definition: stringinfo.c:91
void initStringInfo(StringInfo str)
Definition: stringinfo.c:59
static WalReceiverConn * wrconn
Definition: walreceiver.c:95
@ WALRCV_OK_COMMAND
Definition: walreceiver.h:205
@ WALRCV_ERROR
Definition: walreceiver.h:204
static void walrcv_clear_result(WalRcvExecResult *walres)
Definition: walreceiver.h:440
#define walrcv_exec(conn, exec, nRetTypes, retTypes)
Definition: walreceiver.h:434

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

171 {
173 }
void before_shmem_exit(pg_on_exit_callback function, Datum arg)
Definition: ipc.c:333
static void ReplicationSlotShmemExit(int code, Datum arg)
Definition: slot.c:179

References before_shmem_exit(), and ReplicationSlotShmemExit().

Referenced by BaseInit().

◆ ReplicationSlotMarkDirty()

◆ ReplicationSlotName()

bool ReplicationSlotName ( int  index,
Name  name 
)

Definition at line 423 of file slot.c.

424 {
425  ReplicationSlot *slot;
426  bool found;
427 
429 
430  /*
431  * Ensure that the slot cannot be dropped while we copy the name. Don't
432  * need the spinlock as the name of an existing slot cannot change.
433  */
434  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
435  found = slot->in_use;
436  if (slot->in_use)
437  namestrcpy(name, NameStr(slot->data.name));
438  LWLockRelease(ReplicationSlotControlLock);
439 
440  return found;
441 }
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 1213 of file tablesync.c.

1215 {
1216  snprintf(syncslotname, szslot, "pg_%u_sync_%u_" UINT64_FORMAT, suboid,
1217  relid, GetSystemIdentifier());
1218 }
#define UINT64_FORMAT
Definition: c.h:533
#define snprintf
Definition: port.h:238
uint64 GetSystemIdentifier(void)
Definition: xlog.c:4174

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

548 {
550 
551  Assert(slot != NULL && slot->active_pid != 0);
552 
553  if (slot->data.persistency == RS_EPHEMERAL)
554  {
555  /*
556  * Delete the slot. There is no !PANIC case where this is allowed to
557  * fail, all that may happen is an incomplete cleanup of the on-disk
558  * data.
559  */
561  }
562 
563  /*
564  * If slot needed to temporarily restrain both data and catalog xmin to
565  * create the catalog snapshot, remove that temporary constraint.
566  * Snapshots can only be exported while the initial snapshot is still
567  * acquired.
568  */
569  if (!TransactionIdIsValid(slot->data.xmin) &&
571  {
572  SpinLockAcquire(&slot->mutex);
574  SpinLockRelease(&slot->mutex);
576  }
577 
578  if (slot->data.persistency == RS_PERSISTENT)
579  {
580  /*
581  * Mark persistent slot inactive. We're not freeing it, just
582  * disconnecting, but wake up others that may be waiting for it.
583  */
584  SpinLockAcquire(&slot->mutex);
585  slot->active_pid = 0;
586  SpinLockRelease(&slot->mutex);
588  }
589 
590  MyReplicationSlot = NULL;
591 
592  /* might not have been set when we've been a plain slot */
593  LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
596  LWLockRelease(ProcArrayLock);
597 }
#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:62
#define TransactionIdIsValid(xid)
Definition: transam.h:41

References ReplicationSlot::active_cv, ReplicationSlot::active_pid, Assert(), ConditionVariableBroadcast(), ReplicationSlot::data, ReplicationSlot::effective_xmin, InvalidTransactionId, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), ReplicationSlot::mutex, MyProc, MyReplicationSlot, ReplicationSlotPersistentData::persistency, PGPROC::pgxactoff, PROC_IN_LOGICAL_DECODING, ProcGlobal, ReplicationSlotDropAcquired(), ReplicationSlotsComputeRequiredXmin(), RS_EPHEMERAL, RS_PERSISTENT, SpinLockAcquire, SpinLockRelease, PGPROC::statusFlags, PROC_HDR::statusFlags, TransactionIdIsValid, and ReplicationSlotPersistentData::xmin.

Referenced by 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 1156 of file slot.c.

1157 {
1159 
1160  Assert(slot != NULL);
1162 
1163  /*
1164  * The replication slot mechanism is used to prevent removal of required
1165  * WAL. As there is no interlock between this routine and checkpoints, WAL
1166  * segments could concurrently be removed when a now stale return value of
1167  * ReplicationSlotsComputeRequiredLSN() is used. In the unlikely case that
1168  * this happens we'll just retry.
1169  */
1170  while (true)
1171  {
1172  XLogSegNo segno;
1173  XLogRecPtr restart_lsn;
1174 
1175  /*
1176  * For logical slots log a standby snapshot and start logical decoding
1177  * at exactly that position. That allows the slot to start up more
1178  * quickly.
1179  *
1180  * That's not needed (or indeed helpful) for physical slots as they'll
1181  * start replay at the last logged checkpoint anyway. Instead return
1182  * the location of the last redo LSN. While that slightly increases
1183  * the chance that we have to retry, it's where a base backup has to
1184  * start replay at.
1185  */
1186  if (!RecoveryInProgress() && SlotIsLogical(slot))
1187  {
1188  XLogRecPtr flushptr;
1189 
1190  /* start at current insert position */
1191  restart_lsn = GetXLogInsertRecPtr();
1192  SpinLockAcquire(&slot->mutex);
1193  slot->data.restart_lsn = restart_lsn;
1194  SpinLockRelease(&slot->mutex);
1195 
1196  /* make sure we have enough information to start */
1197  flushptr = LogStandbySnapshot();
1198 
1199  /* and make sure it's fsynced to disk */
1200  XLogFlush(flushptr);
1201  }
1202  else
1203  {
1204  restart_lsn = GetRedoRecPtr();
1205  SpinLockAcquire(&slot->mutex);
1206  slot->data.restart_lsn = restart_lsn;
1207  SpinLockRelease(&slot->mutex);
1208  }
1209 
1210  /* prevent WAL removal as fast as possible */
1212 
1213  /*
1214  * If all required WAL is still there, great, otherwise retry. The
1215  * slot should prevent further removal of WAL, unless there's a
1216  * concurrent ReplicationSlotsComputeRequiredLSN() after we've written
1217  * the new restart_lsn above, so normally we should never need to loop
1218  * more than twice.
1219  */
1221  if (XLogGetLastRemovedSegno() < segno)
1222  break;
1223  }
1224 }
XLogRecPtr LogStandbySnapshot(void)
Definition: standby.c:1273
XLogRecPtr restart_lsn
Definition: slot.h:73
bool RecoveryInProgress(void)
Definition: xlog.c:5905
XLogSegNo XLogGetLastRemovedSegno(void)
Definition: xlog.c:3463
XLogRecPtr GetRedoRecPtr(void)
Definition: xlog.c:6008
XLogRecPtr GetXLogInsertRecPtr(void)
Definition: xlog.c:8854
void XLogFlush(XLogRecPtr record)
Definition: xlog.c:2514
#define XLByteToSeg(xlrp, logSegNo, wal_segsz_bytes)
uint64 XLogSegNo
Definition: xlogdefs.h:48

References Assert(), ReplicationSlot::data, GetRedoRecPtr(), GetXLogInsertRecPtr(), InvalidXLogRecPtr, LogStandbySnapshot(), ReplicationSlot::mutex, MyReplicationSlot, RecoveryInProgress(), ReplicationSlotsComputeRequiredLSN(), ReplicationSlotPersistentData::restart_lsn, SlotIsLogical, 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 935 of file slot.c.

936 {
937  XLogRecPtr result = InvalidXLogRecPtr;
938  int i;
939 
940  if (max_replication_slots <= 0)
941  return InvalidXLogRecPtr;
942 
943  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
944 
945  for (i = 0; i < max_replication_slots; i++)
946  {
947  ReplicationSlot *s;
948  XLogRecPtr restart_lsn;
949 
951 
952  /* cannot change while ReplicationSlotCtlLock is held */
953  if (!s->in_use)
954  continue;
955 
956  /* we're only interested in logical slots */
957  if (!SlotIsLogical(s))
958  continue;
959 
960  /* read once, it's ok if it increases while we're checking */
961  SpinLockAcquire(&s->mutex);
962  restart_lsn = s->data.restart_lsn;
963  SpinLockRelease(&s->mutex);
964 
965  if (restart_lsn == InvalidXLogRecPtr)
966  continue;
967 
968  if (result == InvalidXLogRecPtr ||
969  restart_lsn < result)
970  result = restart_lsn;
971  }
972 
973  LWLockRelease(ReplicationSlotControlLock);
974 
975  return result;
976 }

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

Referenced by CheckPointLogicalRewriteHeap(), and CheckPointSnapBuild().

◆ ReplicationSlotsComputeRequiredLSN()

void ReplicationSlotsComputeRequiredLSN ( void  )

Definition at line 892 of file slot.c.

893 {
894  int i;
895  XLogRecPtr min_required = InvalidXLogRecPtr;
896 
897  Assert(ReplicationSlotCtl != NULL);
898 
899  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
900  for (i = 0; i < max_replication_slots; i++)
901  {
903  XLogRecPtr restart_lsn;
904 
905  if (!s->in_use)
906  continue;
907 
908  SpinLockAcquire(&s->mutex);
909  restart_lsn = s->data.restart_lsn;
910  SpinLockRelease(&s->mutex);
911 
912  if (restart_lsn != InvalidXLogRecPtr &&
913  (min_required == InvalidXLogRecPtr ||
914  restart_lsn < min_required))
915  min_required = restart_lsn;
916  }
917  LWLockRelease(ReplicationSlotControlLock);
918 
919  XLogSetReplicationSlotMinimumLSN(min_required);
920 }
void XLogSetReplicationSlotMinimumLSN(XLogRecPtr lsn)
Definition: xlog.c:2400

References Assert(), ReplicationSlot::data, i, ReplicationSlot::in_use, InvalidXLogRecPtr, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_replication_slots, ReplicationSlot::mutex, ReplicationSlotCtlData::replication_slots, ReplicationSlotCtl, ReplicationSlotPersistentData::restart_lsn, 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 835 of file slot.c.

836 {
837  int i;
839  TransactionId agg_catalog_xmin = InvalidTransactionId;
840 
841  Assert(ReplicationSlotCtl != NULL);
842 
843  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
844 
845  for (i = 0; i < max_replication_slots; i++)
846  {
848  TransactionId effective_xmin;
849  TransactionId effective_catalog_xmin;
850  bool invalidated;
851 
852  if (!s->in_use)
853  continue;
854 
855  SpinLockAcquire(&s->mutex);
856  effective_xmin = s->effective_xmin;
857  effective_catalog_xmin = s->effective_catalog_xmin;
858  invalidated = (!XLogRecPtrIsInvalid(s->data.invalidated_at) &&
860  SpinLockRelease(&s->mutex);
861 
862  /* invalidated slots need not apply */
863  if (invalidated)
864  continue;
865 
866  /* check the data xmin */
867  if (TransactionIdIsValid(effective_xmin) &&
868  (!TransactionIdIsValid(agg_xmin) ||
869  TransactionIdPrecedes(effective_xmin, agg_xmin)))
870  agg_xmin = effective_xmin;
871 
872  /* check the catalog xmin */
873  if (TransactionIdIsValid(effective_catalog_xmin) &&
874  (!TransactionIdIsValid(agg_catalog_xmin) ||
875  TransactionIdPrecedes(effective_catalog_xmin, agg_catalog_xmin)))
876  agg_catalog_xmin = effective_catalog_xmin;
877  }
878 
879  LWLockRelease(ReplicationSlotControlLock);
880 
881  ProcArraySetReplicationSlotXmin(agg_xmin, agg_catalog_xmin, already_locked);
882 }
uint32 TransactionId
Definition: c.h:636
void ProcArraySetReplicationSlotXmin(TransactionId xmin, TransactionId catalog_xmin, bool already_locked)
Definition: procarray.c:3902
XLogRecPtr invalidated_at
Definition: slot.h:76
bool TransactionIdPrecedes(TransactionId id1, TransactionId id2)
Definition: transam.c:280
#define XLogRecPtrIsInvalid(r)
Definition: xlogdefs.h:29

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

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

988 {
989  int i;
990 
991  *nslots = *nactive = 0;
992 
993  if (max_replication_slots <= 0)
994  return false;
995 
996  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
997  for (i = 0; i < max_replication_slots; i++)
998  {
999  ReplicationSlot *s;
1000 
1002 
1003  /* cannot change while ReplicationSlotCtlLock is held */
1004  if (!s->in_use)
1005  continue;
1006 
1007  /* only logical slots are database specific, skip */
1008  if (!SlotIsLogical(s))
1009  continue;
1010 
1011  /* not our database, skip */
1012  if (s->data.database != dboid)
1013  continue;
1014 
1015  /* count slots with spinlock held */
1016  SpinLockAcquire(&s->mutex);
1017  (*nslots)++;
1018  if (s->active_pid != 0)
1019  (*nactive)++;
1020  SpinLockRelease(&s->mutex);
1021  }
1022  LWLockRelease(ReplicationSlotControlLock);
1023 
1024  if (*nslots > 0)
1025  return true;
1026  return false;
1027 }

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

1044 {
1045  int i;
1046 
1047  if (max_replication_slots <= 0)
1048  return;
1049 
1050 restart:
1051  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
1052  for (i = 0; i < max_replication_slots; i++)
1053  {
1054  ReplicationSlot *s;
1055  char *slotname;
1056  int active_pid;
1057 
1059 
1060  /* cannot change while ReplicationSlotCtlLock is held */
1061  if (!s->in_use)
1062  continue;
1063 
1064  /* only logical slots are database specific, skip */
1065  if (!SlotIsLogical(s))
1066  continue;
1067 
1068  /* not our database, skip */
1069  if (s->data.database != dboid)
1070  continue;
1071 
1072  /* acquire slot, so ReplicationSlotDropAcquired can be reused */
1073  SpinLockAcquire(&s->mutex);
1074  /* can't change while ReplicationSlotControlLock is held */
1075  slotname = NameStr(s->data.name);
1076  active_pid = s->active_pid;
1077  if (active_pid == 0)
1078  {
1079  MyReplicationSlot = s;
1080  s->active_pid = MyProcPid;
1081  }
1082  SpinLockRelease(&s->mutex);
1083 
1084  /*
1085  * Even though we hold an exclusive lock on the database object a
1086  * logical slot for that DB can still be active, e.g. if it's
1087  * concurrently being dropped by a backend connected to another DB.
1088  *
1089  * That's fairly unlikely in practice, so we'll just bail out.
1090  */
1091  if (active_pid)
1092  ereport(ERROR,
1093  (errcode(ERRCODE_OBJECT_IN_USE),
1094  errmsg("replication slot \"%s\" is active for PID %d",
1095  slotname, active_pid)));
1096 
1097  /*
1098  * To avoid duplicating ReplicationSlotDropAcquired() and to avoid
1099  * holding ReplicationSlotControlLock over filesystem operations,
1100  * release ReplicationSlotControlLock and use
1101  * ReplicationSlotDropAcquired.
1102  *
1103  * As that means the set of slots could change, restart scan from the
1104  * beginning each time we release the lock.
1105  */
1106  LWLockRelease(ReplicationSlotControlLock);
1108  goto restart;
1109  }
1110  LWLockRelease(ReplicationSlotControlLock);
1111 }

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

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

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

◆ ReplicationSlotsShmemSize()

Size ReplicationSlotsShmemSize ( void  )

Definition at line 117 of file slot.c.

118 {
119  Size size = 0;
120 
121  if (max_replication_slots == 0)
122  return size;
123 
124  size = offsetof(ReplicationSlotCtlData, replication_slots);
125  size = add_size(size,
127 
128  return size;
129 }
size_t Size
Definition: c.h:589
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 198 of file slot.c.

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

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

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

1478 {
1479  DIR *replication_dir;
1480  struct dirent *replication_de;
1481 
1482  elog(DEBUG1, "starting up replication slots");
1483 
1484  /* restore all slots by iterating over all on-disk entries */
1485  replication_dir = AllocateDir("pg_replslot");
1486  while ((replication_de = ReadDir(replication_dir, "pg_replslot")) != NULL)
1487  {
1488  char path[MAXPGPATH + 12];
1489  PGFileType de_type;
1490 
1491  if (strcmp(replication_de->d_name, ".") == 0 ||
1492  strcmp(replication_de->d_name, "..") == 0)
1493  continue;
1494 
1495  snprintf(path, sizeof(path), "pg_replslot/%s", replication_de->d_name);
1496  de_type = get_dirent_type(path, replication_de, false, DEBUG1);
1497 
1498  /* we're only creating directories here, skip if it's not our's */
1499  if (de_type != PGFILETYPE_ERROR && de_type != PGFILETYPE_DIR)
1500  continue;
1501 
1502  /* we crashed while a slot was being setup or deleted, clean up */
1503  if (pg_str_endswith(replication_de->d_name, ".tmp"))
1504  {
1505  if (!rmtree(path, true))
1506  {
1507  ereport(WARNING,
1508  (errmsg("could not remove directory \"%s\"",
1509  path)));
1510  continue;
1511  }
1512  fsync_fname("pg_replslot", true);
1513  continue;
1514  }
1515 
1516  /* looks like a slot in a normal state, restore */
1517  RestoreSlotFromDisk(replication_de->d_name);
1518  }
1519  FreeDir(replication_dir);
1520 
1521  /* currently no slots exist, we're done. */
1522  if (max_replication_slots <= 0)
1523  return;
1524 
1525  /* Now that we have recovered all the data, compute replication xmin */
1528 }
#define WARNING
Definition: elog.h:36
struct dirent * ReadDir(DIR *dir, const char *dirname)
Definition: fd.c:2709
int FreeDir(DIR *dir)
Definition: fd.c:2761
void fsync_fname(const char *fname, bool isdir)
Definition: fd.c:662
DIR * AllocateDir(const char *dirname)
Definition: fd.c:2643
PGFileType get_dirent_type(const char *path, const struct dirent *de, bool look_through_symlinks, int elevel)
Definition: file_utils.c:406
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:1753
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