PostgreSQL Source Code  git master
slot.h File Reference
#include "access/xlog.h"
#include "access/xlogreader.h"
#include "storage/condition_variable.h"
#include "storage/lwlock.h"
#include "storage/shmem.h"
#include "storage/spin.h"
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 SlotAcquireBehavior SlotAcquireBehavior
 
typedef struct ReplicationSlotPersistentData ReplicationSlotPersistentData
 
typedef struct ReplicationSlot ReplicationSlot
 
typedef struct ReplicationSlotCtlData ReplicationSlotCtlData
 

Enumerations

enum  ReplicationSlotPersistency { RS_PERSISTENT, RS_EPHEMERAL, RS_TEMPORARY }
 
enum  SlotAcquireBehavior { SAB_Error, SAB_Block, SAB_Inquire }
 

Functions

Size ReplicationSlotsShmemSize (void)
 
void ReplicationSlotsShmemInit (void)
 
void ReplicationSlotCreate (const char *name, bool db_specific, ReplicationSlotPersistency p)
 
void ReplicationSlotPersist (void)
 
void ReplicationSlotDrop (const char *name, bool nowait)
 
int ReplicationSlotAcquire (const char *name, SlotAcquireBehavior behavior)
 
void ReplicationSlotRelease (void)
 
void ReplicationSlotCleanup (void)
 
void ReplicationSlotSave (void)
 
void ReplicationSlotMarkDirty (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)
 
void InvalidateObsoleteReplicationSlots (XLogSegNo oldestSegno)
 
void StartupReplicationSlots (void)
 
void CheckPointReplicationSlots (void)
 
void CheckSlotRequirements (void)
 

Variables

PGDLLIMPORT ReplicationSlotCtlDataReplicationSlotCtl
 
PGDLLIMPORT ReplicationSlotMyReplicationSlot
 
PGDLLIMPORT int max_replication_slots
 

Macro Definition Documentation

◆ SlotIsLogical

◆ SlotIsPhysical

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

Definition at line 161 of file slot.h.

Referenced by CreateDecodingContext(), and CreateInitDecodingContext().

Typedef Documentation

◆ ReplicationSlot

◆ ReplicationSlotCtlData

◆ ReplicationSlotPersistency

◆ ReplicationSlotPersistentData

◆ SlotAcquireBehavior

Enumeration Type Documentation

◆ ReplicationSlotPersistency

Enumerator
RS_PERSISTENT 
RS_EPHEMERAL 
RS_TEMPORARY 

Definition at line 32 of file slot.h.

33 {
ReplicationSlotPersistency
Definition: slot.h:32

◆ SlotAcquireBehavior

Enumerator
SAB_Error 
SAB_Block 
SAB_Inquire 

Definition at line 40 of file slot.h.

41 {
42  SAB_Error,
43  SAB_Block,
Definition: slot.h:43
SlotAcquireBehavior
Definition: slot.h:40
Definition: slot.h:42

Function Documentation

◆ CheckPointReplicationSlots()

void CheckPointReplicationSlots ( void  )

Definition at line 1164 of file slot.c.

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

Referenced by CheckPointGuts().

1165 {
1166  int i;
1167 
1168  elog(DEBUG1, "performing replication slot checkpoint");
1169 
1170  /*
1171  * Prevent any slot from being created/dropped while we're active. As we
1172  * explicitly do *not* want to block iterating over replication_slots or
1173  * acquiring a slot we cannot take the control lock - but that's OK,
1174  * because holding ReplicationSlotAllocationLock is strictly stronger, and
1175  * enough to guarantee that nobody can change the in_use bits on us.
1176  */
1177  LWLockAcquire(ReplicationSlotAllocationLock, LW_SHARED);
1178 
1179  for (i = 0; i < max_replication_slots; i++)
1180  {
1182  char path[MAXPGPATH];
1183 
1184  if (!s->in_use)
1185  continue;
1186 
1187  /* save the slot to disk, locking is handled in SaveSlotToPath() */
1188  sprintf(path, "pg_replslot/%s", NameStr(s->data.name));
1189  SaveSlotToPath(s, path, LOG);
1190  }
1191  LWLockRelease(ReplicationSlotAllocationLock);
1192 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
#define DEBUG1
Definition: elog.h:25
ReplicationSlotPersistentData data
Definition: slot.h:140
#define LOG
Definition: elog.h:26
static void SaveSlotToPath(ReplicationSlot *slot, const char *path, int elevel)
Definition: slot.c:1320
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1812
#define sprintf
Definition: port.h:195
#define MAXPGPATH
bool in_use
Definition: slot.h:116
int max_replication_slots
Definition: slot.c:99
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1208
#define elog(elevel,...)
Definition: elog.h:214
int i
#define NameStr(name)
Definition: c.h:615
ReplicationSlot replication_slots[1]
Definition: slot.h:173

◆ CheckSlotRequirements()

void CheckSlotRequirements ( void  )

Definition at line 988 of file slot.c.

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

989 {
990  /*
991  * NB: Adding a new requirement likely means that RestoreSlotFromDisk()
992  * needs the same check.
993  */
994 
995  if (max_replication_slots == 0)
996  ereport(ERROR,
997  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
998  errmsg("replication slots can only be used if max_replication_slots > 0")));
999 
1001  ereport(ERROR,
1002  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1003  errmsg("replication slots can only be used if wal_level >= replica")));
1004 }
int wal_level
Definition: xlog.c:106
int errcode(int sqlerrcode)
Definition: elog.c:610
#define ERROR
Definition: elog.h:43
#define ereport(elevel,...)
Definition: elog.h:144
int max_replication_slots
Definition: slot.c:99
int errmsg(const char *fmt,...)
Definition: elog.c:824

◆ InvalidateObsoleteReplicationSlots()

void InvalidateObsoleteReplicationSlots ( XLogSegNo  oldestSegno)

Definition at line 1090 of file slot.c.

References ReplicationSlot::active_cv, CHECK_FOR_INTERRUPTS, ConditionVariableCancelSleep(), ConditionVariableTimedSleep(), ReplicationSlot::data, ereport, errmsg(), i, ReplicationSlot::in_use, InvalidXLogRecPtr, kill, LOG, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_replication_slots, ReplicationSlot::mutex, ReplicationSlotPersistentData::name, NameStr, pstrdup(), ReplicationSlotCtlData::replication_slots, ReplicationSlotAcquire(), ReplicationSlotRelease(), ReplicationSlotPersistentData::restart_lsn, SAB_Inquire, SpinLockAcquire, SpinLockRelease, WAIT_EVENT_REPLICATION_SLOT_DROP, wal_segment_size, and XLogSegNoOffsetToRecPtr.

Referenced by CreateCheckPoint(), and CreateRestartPoint().

1091 {
1092  XLogRecPtr oldestLSN;
1093 
1094  XLogSegNoOffsetToRecPtr(oldestSegno, 0, wal_segment_size, oldestLSN);
1095 
1096 restart:
1097  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
1098  for (int i = 0; i < max_replication_slots; i++)
1099  {
1101  XLogRecPtr restart_lsn = InvalidXLogRecPtr;
1102  char *slotname;
1103 
1104  if (!s->in_use)
1105  continue;
1106 
1107  SpinLockAcquire(&s->mutex);
1108  if (s->data.restart_lsn == InvalidXLogRecPtr ||
1109  s->data.restart_lsn >= oldestLSN)
1110  {
1111  SpinLockRelease(&s->mutex);
1112  continue;
1113  }
1114 
1115  slotname = pstrdup(NameStr(s->data.name));
1116  restart_lsn = s->data.restart_lsn;
1117 
1118  SpinLockRelease(&s->mutex);
1119  LWLockRelease(ReplicationSlotControlLock);
1120 
1121  for (;;)
1122  {
1123  int wspid = ReplicationSlotAcquire(slotname, SAB_Inquire);
1124 
1125  /* no walsender? success! */
1126  if (wspid == 0)
1127  break;
1128 
1129  ereport(LOG,
1130  (errmsg("terminating walsender %d because replication slot \"%s\" is too far behind",
1131  wspid, slotname)));
1132  (void) kill(wspid, SIGTERM);
1133 
1136  }
1138 
1139  ereport(LOG,
1140  (errmsg("invalidating slot \"%s\" because its restart_lsn %X/%X exceeds max_slot_wal_keep_size",
1141  slotname,
1142  (uint32) (restart_lsn >> 32),
1143  (uint32) restart_lsn)));
1144 
1145  SpinLockAcquire(&s->mutex);
1147  SpinLockRelease(&s->mutex);
1149 
1150  /* if we did anything, start from scratch */
1152  goto restart;
1153  }
1154  LWLockRelease(ReplicationSlotControlLock);
1155 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
int wal_segment_size
Definition: xlog.c:116
char * pstrdup(const char *in)
Definition: mcxt.c:1186
#define kill(pid, sig)
Definition: win32_port.h:426
ReplicationSlotPersistentData data
Definition: slot.h:140
#define LOG
Definition: elog.h:26
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1812
#define SpinLockAcquire(lock)
Definition: spin.h:62
void ConditionVariableCancelSleep(void)
bool ConditionVariableTimedSleep(ConditionVariable *cv, long timeout, uint32 wait_event_info)
unsigned int uint32
Definition: c.h:367
void ReplicationSlotRelease(void)
Definition: slot.c:433
bool in_use
Definition: slot.h:116
#define SpinLockRelease(lock)
Definition: spin.h:64
#define ereport(elevel,...)
Definition: elog.h:144
int max_replication_slots
Definition: slot.c:99
#define XLogSegNoOffsetToRecPtr(segno, offset, wal_segsz_bytes, dest)
uint64 XLogRecPtr
Definition: xlogdefs.h:21
XLogRecPtr restart_lsn
Definition: slot.h:80
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1208
ConditionVariable active_cv
Definition: slot.h:146
int errmsg(const char *fmt,...)
Definition: elog.c:824
int ReplicationSlotAcquire(const char *name, SlotAcquireBehavior behavior)
Definition: slot.c:334
int i
#define NameStr(name)
Definition: c.h:615
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:99
ReplicationSlot replication_slots[1]
Definition: slot.h:173
slock_t mutex
Definition: slot.h:113

◆ ReplicationSlotAcquire()

int ReplicationSlotAcquire ( const char *  name,
SlotAcquireBehavior  behavior 
)

Definition at line 334 of file slot.c.

References ReplicationSlot::active_cv, ReplicationSlot::active_pid, Assert, ConditionVariableBroadcast(), ConditionVariableCancelSleep(), ConditionVariablePrepareToSleep(), ConditionVariableSleep(), ReplicationSlot::data, ereport, errcode(), errmsg(), ERROR, i, ReplicationSlot::in_use, IsUnderPostmaster, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_replication_slots, ReplicationSlot::mutex, MyProcPid, ReplicationSlotPersistentData::name, NameStr, ReplicationSlotCtlData::replication_slots, SAB_Error, SAB_Inquire, SpinLockAcquire, SpinLockRelease, and WAIT_EVENT_REPLICATION_SLOT_DROP.

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

335 {
336  ReplicationSlot *slot;
337  int active_pid;
338  int i;
339 
340 retry:
341  Assert(MyReplicationSlot == NULL);
342 
343  /*
344  * Search for the named slot and mark it active if we find it. If the
345  * slot is already active, we exit the loop with active_pid set to the PID
346  * of the backend that owns it.
347  */
348  active_pid = 0;
349  slot = NULL;
350  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
351  for (i = 0; i < max_replication_slots; i++)
352  {
354 
355  if (s->in_use && strcmp(name, NameStr(s->data.name)) == 0)
356  {
357  /*
358  * This is the slot we want; check if it's active under some other
359  * process. In single user mode, we don't need this check.
360  */
361  if (IsUnderPostmaster)
362  {
363  /*
364  * Get ready to sleep on it in case it is active. (We may end
365  * up not sleeping, but we don't want to do this while holding
366  * the spinlock.)
367  */
369 
370  SpinLockAcquire(&s->mutex);
371 
372  active_pid = s->active_pid;
373  if (active_pid == 0)
374  active_pid = s->active_pid = MyProcPid;
375 
376  SpinLockRelease(&s->mutex);
377  }
378  else
379  active_pid = MyProcPid;
380  slot = s;
381 
382  break;
383  }
384  }
385  LWLockRelease(ReplicationSlotControlLock);
386 
387  /* If we did not find the slot, error out. */
388  if (slot == NULL)
389  ereport(ERROR,
390  (errcode(ERRCODE_UNDEFINED_OBJECT),
391  errmsg("replication slot \"%s\" does not exist", name)));
392 
393  /*
394  * If we found the slot but it's already active in another backend, we
395  * either error out or retry after a short wait, as caller specified.
396  */
397  if (active_pid != MyProcPid)
398  {
399  if (behavior == SAB_Error)
400  ereport(ERROR,
401  (errcode(ERRCODE_OBJECT_IN_USE),
402  errmsg("replication slot \"%s\" is active for PID %d",
403  name, active_pid)));
404  else if (behavior == SAB_Inquire)
405  return active_pid;
406 
407  /* Wait here until we get signaled, and then restart */
411  goto retry;
412  }
413  else
414  ConditionVariableCancelSleep(); /* no sleep needed after all */
415 
416  /* Let everybody know we've modified this slot */
418 
419  /* We made this slot active, so it's ours now. */
420  MyReplicationSlot = slot;
421 
422  /* success */
423  return 0;
424 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
int MyProcPid
Definition: globals.c:40
void ConditionVariableBroadcast(ConditionVariable *cv)
int errcode(int sqlerrcode)
Definition: elog.c:610
ReplicationSlotPersistentData data
Definition: slot.h:140
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1812
void ConditionVariablePrepareToSleep(ConditionVariable *cv)
#define SpinLockAcquire(lock)
Definition: spin.h:62
void ConditionVariableCancelSleep(void)
#define ERROR
Definition: elog.h:43
bool IsUnderPostmaster
Definition: globals.c:109
bool in_use
Definition: slot.h:116
#define SpinLockRelease(lock)
Definition: spin.h:64
#define ereport(elevel,...)
Definition: elog.h:144
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
int max_replication_slots
Definition: slot.c:99
void ConditionVariableSleep(ConditionVariable *cv, uint32 wait_event_info)
#define Assert(condition)
Definition: c.h:738
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1208
ConditionVariable active_cv
Definition: slot.h:146
const char * name
Definition: encode.c:555
int errmsg(const char *fmt,...)
Definition: elog.c:824
pid_t active_pid
Definition: slot.h:119
int i
Definition: slot.h:42
#define NameStr(name)
Definition: c.h:615
ReplicationSlot replication_slots[1]
Definition: slot.h:173
slock_t mutex
Definition: slot.h:113

◆ ReplicationSlotCleanup()

void ReplicationSlotCleanup ( void  )

Definition at line 488 of file slot.c.

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, ReplicationSlotPersistentData::persistency, ReplicationSlotCtlData::replication_slots, ReplicationSlotDropPtr(), RS_TEMPORARY, SpinLockAcquire, and SpinLockRelease.

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

489 {
490  int i;
491 
492  Assert(MyReplicationSlot == NULL);
493 
494 restart:
495  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
496  for (i = 0; i < max_replication_slots; i++)
497  {
499 
500  if (!s->in_use)
501  continue;
502 
503  SpinLockAcquire(&s->mutex);
504  if (s->active_pid == MyProcPid)
505  {
507  SpinLockRelease(&s->mutex);
508  LWLockRelease(ReplicationSlotControlLock); /* avoid deadlock */
509 
511 
513  goto restart;
514  }
515  else
516  SpinLockRelease(&s->mutex);
517  }
518 
519  LWLockRelease(ReplicationSlotControlLock);
520 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
int MyProcPid
Definition: globals.c:40
ReplicationSlotPersistency persistency
Definition: slot.h:61
void ConditionVariableBroadcast(ConditionVariable *cv)
static void ReplicationSlotDropPtr(ReplicationSlot *slot)
Definition: slot.c:556
ReplicationSlotPersistentData data
Definition: slot.h:140
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1812
#define SpinLockAcquire(lock)
Definition: spin.h:62
bool in_use
Definition: slot.h:116
#define SpinLockRelease(lock)
Definition: spin.h:64
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
int max_replication_slots
Definition: slot.c:99
#define Assert(condition)
Definition: c.h:738
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1208
ConditionVariable active_cv
Definition: slot.h:146
pid_t active_pid
Definition: slot.h:119
int i
ReplicationSlot replication_slots[1]
Definition: slot.h:173
slock_t mutex
Definition: slot.h:113

◆ ReplicationSlotCreate()

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

Definition at line 219 of file slot.c.

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, ReplicationSlotPersistentData::name, NAMEDATALEN, NameStr, ReplicationSlotPersistentData::persistency, ReplicationSlotCtlData::replication_slots, ReplicationSlotValidateName(), SpinLockAcquire, SpinLockRelease, and StrNCpy.

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

221 {
222  ReplicationSlot *slot = NULL;
223  int i;
224 
225  Assert(MyReplicationSlot == NULL);
226 
228 
229  /*
230  * If some other backend ran this code concurrently with us, we'd likely
231  * both allocate the same slot, and that would be bad. We'd also be at
232  * risk of missing a name collision. Also, we don't want to try to create
233  * a new slot while somebody's busy cleaning up an old one, because we
234  * might both be monkeying with the same directory.
235  */
236  LWLockAcquire(ReplicationSlotAllocationLock, LW_EXCLUSIVE);
237 
238  /*
239  * Check for name collision, and identify an allocatable slot. We need to
240  * hold ReplicationSlotControlLock in shared mode for this, so that nobody
241  * else can change the in_use flags while we're looking at them.
242  */
243  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
244  for (i = 0; i < max_replication_slots; i++)
245  {
247 
248  if (s->in_use && strcmp(name, NameStr(s->data.name)) == 0)
249  ereport(ERROR,
251  errmsg("replication slot \"%s\" already exists", name)));
252  if (!s->in_use && slot == NULL)
253  slot = s;
254  }
255  LWLockRelease(ReplicationSlotControlLock);
256 
257  /* If all slots are in use, we're out of luck. */
258  if (slot == NULL)
259  ereport(ERROR,
260  (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
261  errmsg("all replication slots are in use"),
262  errhint("Free one or increase max_replication_slots.")));
263 
264  /*
265  * Since this slot is not in use, nobody should be looking at any part of
266  * it other than the in_use field unless they're trying to allocate it.
267  * And since we hold ReplicationSlotAllocationLock, nobody except us can
268  * be doing that. So it's safe to initialize the slot.
269  */
270  Assert(!slot->in_use);
271  Assert(slot->active_pid == 0);
272 
273  /* first initialize persistent data */
274  memset(&slot->data, 0, sizeof(ReplicationSlotPersistentData));
276  slot->data.database = db_specific ? MyDatabaseId : InvalidOid;
277  slot->data.persistency = persistency;
278 
279  /* and then data only present in shared memory */
280  slot->just_dirtied = false;
281  slot->dirty = false;
288 
289  /*
290  * Create the slot on disk. We haven't actually marked the slot allocated
291  * yet, so no special cleanup is required if this errors out.
292  */
293  CreateSlotOnDisk(slot);
294 
295  /*
296  * We need to briefly prevent any other backend from iterating over the
297  * slots while we flip the in_use flag. We also need to set the active
298  * flag while holding the ControlLock as otherwise a concurrent
299  * ReplicationSlotAcquire() could acquire the slot as well.
300  */
301  LWLockAcquire(ReplicationSlotControlLock, LW_EXCLUSIVE);
302 
303  slot->in_use = true;
304 
305  /* We can now mark the slot active, and that makes it our slot. */
306  SpinLockAcquire(&slot->mutex);
307  Assert(slot->active_pid == 0);
308  slot->active_pid = MyProcPid;
309  SpinLockRelease(&slot->mutex);
310  MyReplicationSlot = slot;
311 
312  LWLockRelease(ReplicationSlotControlLock);
313 
314  /*
315  * Now that the slot has been marked as in_use and active, it's safe to
316  * let somebody else try to allocate a slot.
317  */
318  LWLockRelease(ReplicationSlotAllocationLock);
319 
320  /* Let everybody know we've modified this slot */
322 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
TransactionId candidate_catalog_xmin
Definition: slot.h:155
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
int MyProcPid
Definition: globals.c:40
int errhint(const char *fmt,...)
Definition: elog.c:1071
ReplicationSlotPersistency persistency
Definition: slot.h:61
void ConditionVariableBroadcast(ConditionVariable *cv)
int errcode(int sqlerrcode)
Definition: elog.c:610
ReplicationSlotPersistentData data
Definition: slot.h:140
static void CreateSlotOnDisk(ReplicationSlot *slot)
Definition: slot.c:1259
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1812
#define NAMEDATALEN
bool ReplicationSlotValidateName(const char *name, int elevel)
Definition: slot.c:172
#define SpinLockAcquire(lock)
Definition: spin.h:62
#define ERROR
Definition: elog.h:43
TransactionId effective_xmin
Definition: slot.h:136
XLogRecPtr candidate_restart_valid
Definition: slot.h:157
#define InvalidTransactionId
Definition: transam.h:31
bool in_use
Definition: slot.h:116
#define SpinLockRelease(lock)
Definition: spin.h:64
bool just_dirtied
Definition: slot.h:122
TransactionId effective_catalog_xmin
Definition: slot.h:137
Oid MyDatabaseId
Definition: globals.c:85
#define InvalidOid
Definition: postgres_ext.h:36
#define ereport(elevel,...)
Definition: elog.h:144
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
int max_replication_slots
Definition: slot.c:99
#define Assert(condition)
Definition: c.h:738
#define StrNCpy(dst, src, len)
Definition: c.h:944
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1208
ConditionVariable active_cv
Definition: slot.h:146
const char * name
Definition: encode.c:555
XLogRecPtr candidate_xmin_lsn
Definition: slot.h:156
int errmsg(const char *fmt,...)
Definition: elog.c:824
pid_t active_pid
Definition: slot.h:119
int i
#define NameStr(name)
Definition: c.h:615
ReplicationSlot replication_slots[1]
Definition: slot.h:173
slock_t mutex
Definition: slot.h:113
#define ERRCODE_DUPLICATE_OBJECT
Definition: streamutil.c:31
bool dirty
Definition: slot.h:123
XLogRecPtr candidate_restart_lsn
Definition: slot.h:158

◆ ReplicationSlotDrop()

void ReplicationSlotDrop ( const char *  name,
bool  nowait 
)

Definition at line 526 of file slot.c.

References Assert, ReplicationSlotAcquire(), ReplicationSlotDropAcquired(), SAB_Block, and SAB_Error.

Referenced by DropReplicationSlot(), and pg_drop_replication_slot().

527 {
528  Assert(MyReplicationSlot == NULL);
529 
530  (void) ReplicationSlotAcquire(name, nowait ? SAB_Error : SAB_Block);
531 
533 }
static void ReplicationSlotDropAcquired(void)
Definition: slot.c:539
Definition: slot.h:43
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
#define Assert(condition)
Definition: c.h:738
const char * name
Definition: encode.c:555
int ReplicationSlotAcquire(const char *name, SlotAcquireBehavior behavior)
Definition: slot.c:334
Definition: slot.h:42

◆ ReplicationSlotMarkDirty()

◆ ReplicationSlotPersist()

void ReplicationSlotPersist ( void  )

Definition at line 689 of file slot.c.

References Assert, ReplicationSlot::data, ReplicationSlot::mutex, MyReplicationSlot, ReplicationSlotPersistentData::persistency, ReplicationSlotMarkDirty(), ReplicationSlotSave(), RS_PERSISTENT, SpinLockAcquire, and SpinLockRelease.

Referenced by copy_replication_slot(), CreateReplicationSlot(), and pg_create_logical_replication_slot().

690 {
692 
693  Assert(slot != NULL);
695 
696  SpinLockAcquire(&slot->mutex);
698  SpinLockRelease(&slot->mutex);
699 
702 }
ReplicationSlotPersistency persistency
Definition: slot.h:61
void ReplicationSlotSave(void)
Definition: slot.c:654
ReplicationSlotPersistentData data
Definition: slot.h:140
#define SpinLockAcquire(lock)
Definition: spin.h:62
#define SpinLockRelease(lock)
Definition: spin.h:64
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
#define Assert(condition)
Definition: c.h:738
slock_t mutex
Definition: slot.h:113
void ReplicationSlotMarkDirty(void)
Definition: slot.c:672

◆ ReplicationSlotRelease()

void ReplicationSlotRelease ( void  )

Definition at line 433 of file slot.c.

References ReplicationSlot::active_cv, ReplicationSlot::active_pid, Assert, ConditionVariableBroadcast(), ReplicationSlot::data, ReplicationSlot::effective_xmin, InvalidTransactionId, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), ReplicationSlot::mutex, MyPgXact, MyReplicationSlot, ReplicationSlotPersistentData::persistency, PROC_IN_LOGICAL_DECODING, ReplicationSlotDropAcquired(), ReplicationSlotsComputeRequiredXmin(), RS_EPHEMERAL, RS_PERSISTENT, SpinLockAcquire, SpinLockRelease, TransactionIdIsValid, PGXACT::vacuumFlags, and ReplicationSlotPersistentData::xmin.

Referenced by copy_replication_slot(), CreateReplicationSlot(), InvalidateObsoleteReplicationSlots(), pg_create_logical_replication_slot(), pg_create_physical_replication_slot(), pg_logical_slot_get_changes_guts(), pg_replication_slot_advance(), PostgresMain(), ProcKill(), StartLogicalReplication(), StartReplication(), and WalSndErrorCleanup().

434 {
436 
437  Assert(slot != NULL && slot->active_pid != 0);
438 
439  if (slot->data.persistency == RS_EPHEMERAL)
440  {
441  /*
442  * Delete the slot. There is no !PANIC case where this is allowed to
443  * fail, all that may happen is an incomplete cleanup of the on-disk
444  * data.
445  */
447  }
448 
449  /*
450  * If slot needed to temporarily restrain both data and catalog xmin to
451  * create the catalog snapshot, remove that temporary constraint.
452  * Snapshots can only be exported while the initial snapshot is still
453  * acquired.
454  */
455  if (!TransactionIdIsValid(slot->data.xmin) &&
457  {
458  SpinLockAcquire(&slot->mutex);
460  SpinLockRelease(&slot->mutex);
462  }
463 
464  if (slot->data.persistency == RS_PERSISTENT)
465  {
466  /*
467  * Mark persistent slot inactive. We're not freeing it, just
468  * disconnecting, but wake up others that may be waiting for it.
469  */
470  SpinLockAcquire(&slot->mutex);
471  slot->active_pid = 0;
472  SpinLockRelease(&slot->mutex);
474  }
475 
476  MyReplicationSlot = NULL;
477 
478  /* might not have been set when we've been a plain slot */
479  LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
481  LWLockRelease(ProcArrayLock);
482 }
#define PROC_IN_LOGICAL_DECODING
Definition: proc.h:57
ReplicationSlotPersistency persistency
Definition: slot.h:61
void ConditionVariableBroadcast(ConditionVariable *cv)
ReplicationSlotPersistentData data
Definition: slot.h:140
PGXACT * MyPgXact
Definition: proc.c:68
uint8 vacuumFlags
Definition: proc.h:233
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1812
#define SpinLockAcquire(lock)
Definition: spin.h:62
static void ReplicationSlotDropAcquired(void)
Definition: slot.c:539
TransactionId effective_xmin
Definition: slot.h:136
#define InvalidTransactionId
Definition: transam.h:31
TransactionId xmin
Definition: slot.h:69
#define SpinLockRelease(lock)
Definition: spin.h:64
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
#define Assert(condition)
Definition: c.h:738
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1208
ConditionVariable active_cv
Definition: slot.h:146
pid_t active_pid
Definition: slot.h:119
slock_t mutex
Definition: slot.h:113
#define TransactionIdIsValid(xid)
Definition: transam.h:41
void ReplicationSlotsComputeRequiredXmin(bool already_locked)
Definition: slot.c:711

◆ ReplicationSlotReserveWal()

void ReplicationSlotReserveWal ( void  )

Definition at line 1013 of file slot.c.

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

1014 {
1016 
1017  Assert(slot != NULL);
1019 
1020  /*
1021  * The replication slot mechanism is used to prevent removal of required
1022  * WAL. As there is no interlock between this routine and checkpoints, WAL
1023  * segments could concurrently be removed when a now stale return value of
1024  * ReplicationSlotsComputeRequiredLSN() is used. In the unlikely case that
1025  * this happens we'll just retry.
1026  */
1027  while (true)
1028  {
1029  XLogSegNo segno;
1030  XLogRecPtr restart_lsn;
1031 
1032  /*
1033  * For logical slots log a standby snapshot and start logical decoding
1034  * at exactly that position. That allows the slot to start up more
1035  * quickly.
1036  *
1037  * That's not needed (or indeed helpful) for physical slots as they'll
1038  * start replay at the last logged checkpoint anyway. Instead return
1039  * the location of the last redo LSN. While that slightly increases
1040  * the chance that we have to retry, it's where a base backup has to
1041  * start replay at.
1042  */
1043  if (!RecoveryInProgress() && SlotIsLogical(slot))
1044  {
1045  XLogRecPtr flushptr;
1046 
1047  /* start at current insert position */
1048  restart_lsn = GetXLogInsertRecPtr();
1049  SpinLockAcquire(&slot->mutex);
1050  slot->data.restart_lsn = restart_lsn;
1051  SpinLockRelease(&slot->mutex);
1052 
1053  /* make sure we have enough information to start */
1054  flushptr = LogStandbySnapshot();
1055 
1056  /* and make sure it's fsynced to disk */
1057  XLogFlush(flushptr);
1058  }
1059  else
1060  {
1061  restart_lsn = GetRedoRecPtr();
1062  SpinLockAcquire(&slot->mutex);
1063  slot->data.restart_lsn = restart_lsn;
1064  SpinLockRelease(&slot->mutex);
1065  }
1066 
1067  /* prevent WAL removal as fast as possible */
1069 
1070  /*
1071  * If all required WAL is still there, great, otherwise retry. The
1072  * slot should prevent further removal of WAL, unless there's a
1073  * concurrent ReplicationSlotsComputeRequiredLSN() after we've written
1074  * the new restart_lsn above, so normally we should never need to loop
1075  * more than twice.
1076  */
1078  if (XLogGetLastRemovedSegno() < segno)
1079  break;
1080  }
1081 }
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
int wal_segment_size
Definition: xlog.c:116
ReplicationSlotPersistentData data
Definition: slot.h:140
bool RecoveryInProgress(void)
Definition: xlog.c:8069
void XLogFlush(XLogRecPtr record)
Definition: xlog.c:2844
#define SpinLockAcquire(lock)
Definition: spin.h:62
XLogSegNo XLogGetLastRemovedSegno(void)
Definition: xlog.c:3956
void ReplicationSlotsComputeRequiredLSN(void)
Definition: slot.c:761
XLogRecPtr LogStandbySnapshot(void)
Definition: standby.c:923
XLogRecPtr GetXLogInsertRecPtr(void)
Definition: xlog.c:11482
uint64 XLogSegNo
Definition: xlogdefs.h:41
#define SlotIsLogical(slot)
Definition: slot.h:162
#define SpinLockRelease(lock)
Definition: spin.h:64
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
uint64 XLogRecPtr
Definition: xlogdefs.h:21
#define Assert(condition)
Definition: c.h:738
XLogRecPtr restart_lsn
Definition: slot.h:80
XLogRecPtr GetRedoRecPtr(void)
Definition: xlog.c:8361
slock_t mutex
Definition: slot.h:113
#define XLByteToSeg(xlrp, logSegNo, wal_segsz_bytes)

◆ ReplicationSlotSave()

void ReplicationSlotSave ( void  )

Definition at line 654 of file slot.c.

References Assert, ReplicationSlot::data, ERROR, MAXPGPATH, ReplicationSlotPersistentData::name, NameStr, SaveSlotToPath(), and sprintf.

Referenced by copy_replication_slot(), create_physical_replication_slot(), CreateInitDecodingContext(), CreateReplicationSlot(), LogicalConfirmReceivedLocation(), and ReplicationSlotPersist().

655 {
656  char path[MAXPGPATH];
657 
658  Assert(MyReplicationSlot != NULL);
659 
660  sprintf(path, "pg_replslot/%s", NameStr(MyReplicationSlot->data.name));
662 }
ReplicationSlotPersistentData data
Definition: slot.h:140
static void SaveSlotToPath(ReplicationSlot *slot, const char *path, int elevel)
Definition: slot.c:1320
#define sprintf
Definition: port.h:195
#define ERROR
Definition: elog.h:43
#define MAXPGPATH
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
#define Assert(condition)
Definition: c.h:738
#define NameStr(name)
Definition: c.h:615

◆ ReplicationSlotsComputeLogicalRestartLSN()

XLogRecPtr ReplicationSlotsComputeLogicalRestartLSN ( void  )

Definition at line 804 of file slot.c.

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

Referenced by CheckPointLogicalRewriteHeap(), and CheckPointSnapBuild().

805 {
806  XLogRecPtr result = InvalidXLogRecPtr;
807  int i;
808 
809  if (max_replication_slots <= 0)
810  return InvalidXLogRecPtr;
811 
812  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
813 
814  for (i = 0; i < max_replication_slots; i++)
815  {
816  ReplicationSlot *s;
817  XLogRecPtr restart_lsn;
818 
820 
821  /* cannot change while ReplicationSlotCtlLock is held */
822  if (!s->in_use)
823  continue;
824 
825  /* we're only interested in logical slots */
826  if (!SlotIsLogical(s))
827  continue;
828 
829  /* read once, it's ok if it increases while we're checking */
830  SpinLockAcquire(&s->mutex);
831  restart_lsn = s->data.restart_lsn;
832  SpinLockRelease(&s->mutex);
833 
834  if (restart_lsn == InvalidXLogRecPtr)
835  continue;
836 
837  if (result == InvalidXLogRecPtr ||
838  restart_lsn < result)
839  result = restart_lsn;
840  }
841 
842  LWLockRelease(ReplicationSlotControlLock);
843 
844  return result;
845 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
ReplicationSlotPersistentData data
Definition: slot.h:140
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1812
#define SpinLockAcquire(lock)
Definition: spin.h:62
#define SlotIsLogical(slot)
Definition: slot.h:162
bool in_use
Definition: slot.h:116
#define SpinLockRelease(lock)
Definition: spin.h:64
int max_replication_slots
Definition: slot.c:99
uint64 XLogRecPtr
Definition: xlogdefs.h:21
XLogRecPtr restart_lsn
Definition: slot.h:80
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1208
int i
ReplicationSlot replication_slots[1]
Definition: slot.h:173
slock_t mutex
Definition: slot.h:113

◆ ReplicationSlotsComputeRequiredLSN()

void ReplicationSlotsComputeRequiredLSN ( void  )

Definition at line 761 of file slot.c.

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

Referenced by copy_replication_slot(), LogicalConfirmReceivedLocation(), PhysicalConfirmReceivedLocation(), ReplicationSlotDropPtr(), ReplicationSlotReserveWal(), and StartupReplicationSlots().

762 {
763  int i;
764  XLogRecPtr min_required = InvalidXLogRecPtr;
765 
766  Assert(ReplicationSlotCtl != NULL);
767 
768  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
769  for (i = 0; i < max_replication_slots; i++)
770  {
772  XLogRecPtr restart_lsn;
773 
774  if (!s->in_use)
775  continue;
776 
777  SpinLockAcquire(&s->mutex);
778  restart_lsn = s->data.restart_lsn;
779  SpinLockRelease(&s->mutex);
780 
781  if (restart_lsn != InvalidXLogRecPtr &&
782  (min_required == InvalidXLogRecPtr ||
783  restart_lsn < min_required))
784  min_required = restart_lsn;
785  }
786  LWLockRelease(ReplicationSlotControlLock);
787 
788  XLogSetReplicationSlotMinimumLSN(min_required);
789 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
ReplicationSlotPersistentData data
Definition: slot.h:140
void XLogSetReplicationSlotMinimumLSN(XLogRecPtr lsn)
Definition: xlog.c:2723
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1812
#define SpinLockAcquire(lock)
Definition: spin.h:62
bool in_use
Definition: slot.h:116
#define SpinLockRelease(lock)
Definition: spin.h:64
int max_replication_slots
Definition: slot.c:99
uint64 XLogRecPtr
Definition: xlogdefs.h:21
#define Assert(condition)
Definition: c.h:738
XLogRecPtr restart_lsn
Definition: slot.h:80
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1208
int i
ReplicationSlot replication_slots[1]
Definition: slot.h:173
slock_t mutex
Definition: slot.h:113

◆ ReplicationSlotsComputeRequiredXmin()

void ReplicationSlotsComputeRequiredXmin ( bool  already_locked)

Definition at line 711 of file slot.c.

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

Referenced by copy_replication_slot(), CreateInitDecodingContext(), LogicalConfirmReceivedLocation(), PhysicalReplicationSlotNewXmin(), ReplicationSlotDropPtr(), ReplicationSlotRelease(), and StartupReplicationSlots().

712 {
713  int i;
715  TransactionId agg_catalog_xmin = InvalidTransactionId;
716 
717  Assert(ReplicationSlotCtl != NULL);
718 
719  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
720 
721  for (i = 0; i < max_replication_slots; i++)
722  {
724  TransactionId effective_xmin;
725  TransactionId effective_catalog_xmin;
726 
727  if (!s->in_use)
728  continue;
729 
730  SpinLockAcquire(&s->mutex);
731  effective_xmin = s->effective_xmin;
732  effective_catalog_xmin = s->effective_catalog_xmin;
733  SpinLockRelease(&s->mutex);
734 
735  /* check the data xmin */
736  if (TransactionIdIsValid(effective_xmin) &&
737  (!TransactionIdIsValid(agg_xmin) ||
738  TransactionIdPrecedes(effective_xmin, agg_xmin)))
739  agg_xmin = effective_xmin;
740 
741  /* check the catalog xmin */
742  if (TransactionIdIsValid(effective_catalog_xmin) &&
743  (!TransactionIdIsValid(agg_catalog_xmin) ||
744  TransactionIdPrecedes(effective_catalog_xmin, agg_catalog_xmin)))
745  agg_catalog_xmin = effective_catalog_xmin;
746  }
747 
748  LWLockRelease(ReplicationSlotControlLock);
749 
750  ProcArraySetReplicationSlotXmin(agg_xmin, agg_catalog_xmin, already_locked);
751 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
uint32 TransactionId
Definition: c.h:513
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1812
#define SpinLockAcquire(lock)
Definition: spin.h:62
TransactionId effective_xmin
Definition: slot.h:136
#define InvalidTransactionId
Definition: transam.h:31
bool TransactionIdPrecedes(TransactionId id1, TransactionId id2)
Definition: transam.c:300
bool in_use
Definition: slot.h:116
#define SpinLockRelease(lock)
Definition: spin.h:64
TransactionId effective_catalog_xmin
Definition: slot.h:137
int max_replication_slots
Definition: slot.c:99
#define Assert(condition)
Definition: c.h:738
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1208
int i
void ProcArraySetReplicationSlotXmin(TransactionId xmin, TransactionId catalog_xmin, bool already_locked)
Definition: procarray.c:3092
ReplicationSlot replication_slots[1]
Definition: slot.h:173
slock_t mutex
Definition: slot.h:113
#define TransactionIdIsValid(xid)
Definition: transam.h:41

◆ ReplicationSlotsCountDBSlots()

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

Definition at line 856 of file slot.c.

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

Referenced by dropdb().

857 {
858  int i;
859 
860  *nslots = *nactive = 0;
861 
862  if (max_replication_slots <= 0)
863  return false;
864 
865  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
866  for (i = 0; i < max_replication_slots; i++)
867  {
868  ReplicationSlot *s;
869 
871 
872  /* cannot change while ReplicationSlotCtlLock is held */
873  if (!s->in_use)
874  continue;
875 
876  /* only logical slots are database specific, skip */
877  if (!SlotIsLogical(s))
878  continue;
879 
880  /* not our database, skip */
881  if (s->data.database != dboid)
882  continue;
883 
884  /* count slots with spinlock held */
885  SpinLockAcquire(&s->mutex);
886  (*nslots)++;
887  if (s->active_pid != 0)
888  (*nactive)++;
889  SpinLockRelease(&s->mutex);
890  }
891  LWLockRelease(ReplicationSlotControlLock);
892 
893  if (*nslots > 0)
894  return true;
895  return false;
896 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
ReplicationSlotPersistentData data
Definition: slot.h:140
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1812
#define SpinLockAcquire(lock)
Definition: spin.h:62
#define SlotIsLogical(slot)
Definition: slot.h:162
bool in_use
Definition: slot.h:116
#define SpinLockRelease(lock)
Definition: spin.h:64
int max_replication_slots
Definition: slot.c:99
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1208
pid_t active_pid
Definition: slot.h:119
int i
ReplicationSlot replication_slots[1]
Definition: slot.h:173
slock_t mutex
Definition: slot.h:113

◆ ReplicationSlotsDropDBSlots()

void ReplicationSlotsDropDBSlots ( Oid  dboid)

Definition at line 912 of file slot.c.

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, ReplicationSlotPersistentData::name, NameStr, ReplicationSlotCtlData::replication_slots, ReplicationSlotDropAcquired(), SlotIsLogical, SpinLockAcquire, and SpinLockRelease.

Referenced by dbase_redo(), and dropdb().

913 {
914  int i;
915 
916  if (max_replication_slots <= 0)
917  return;
918 
919 restart:
920  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
921  for (i = 0; i < max_replication_slots; i++)
922  {
923  ReplicationSlot *s;
924  char *slotname;
925  int active_pid;
926 
928 
929  /* cannot change while ReplicationSlotCtlLock is held */
930  if (!s->in_use)
931  continue;
932 
933  /* only logical slots are database specific, skip */
934  if (!SlotIsLogical(s))
935  continue;
936 
937  /* not our database, skip */
938  if (s->data.database != dboid)
939  continue;
940 
941  /* acquire slot, so ReplicationSlotDropAcquired can be reused */
942  SpinLockAcquire(&s->mutex);
943  /* can't change while ReplicationSlotControlLock is held */
944  slotname = NameStr(s->data.name);
945  active_pid = s->active_pid;
946  if (active_pid == 0)
947  {
948  MyReplicationSlot = s;
949  s->active_pid = MyProcPid;
950  }
951  SpinLockRelease(&s->mutex);
952 
953  /*
954  * Even though we hold an exclusive lock on the database object a
955  * logical slot for that DB can still be active, e.g. if it's
956  * concurrently being dropped by a backend connected to another DB.
957  *
958  * That's fairly unlikely in practice, so we'll just bail out.
959  */
960  if (active_pid)
961  ereport(ERROR,
962  (errcode(ERRCODE_OBJECT_IN_USE),
963  errmsg("replication slot \"%s\" is active for PID %d",
964  slotname, active_pid)));
965 
966  /*
967  * To avoid duplicating ReplicationSlotDropAcquired() and to avoid
968  * holding ReplicationSlotControlLock over filesystem operations,
969  * release ReplicationSlotControlLock and use
970  * ReplicationSlotDropAcquired.
971  *
972  * As that means the set of slots could change, restart scan from the
973  * beginning each time we release the lock.
974  */
975  LWLockRelease(ReplicationSlotControlLock);
977  goto restart;
978  }
979  LWLockRelease(ReplicationSlotControlLock);
980 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
int MyProcPid
Definition: globals.c:40
int errcode(int sqlerrcode)
Definition: elog.c:610
ReplicationSlotPersistentData data
Definition: slot.h:140
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1812
#define SpinLockAcquire(lock)
Definition: spin.h:62
static void ReplicationSlotDropAcquired(void)
Definition: slot.c:539
#define ERROR
Definition: elog.h:43
#define SlotIsLogical(slot)
Definition: slot.h:162
bool in_use
Definition: slot.h:116
#define SpinLockRelease(lock)
Definition: spin.h:64
#define ereport(elevel,...)
Definition: elog.h:144
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
int max_replication_slots
Definition: slot.c:99
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1208
int errmsg(const char *fmt,...)
Definition: elog.c:824
pid_t active_pid
Definition: slot.h:119
int i
#define NameStr(name)
Definition: c.h:615
ReplicationSlot replication_slots[1]
Definition: slot.h:173
slock_t mutex
Definition: slot.h:113

◆ ReplicationSlotsShmemInit()

void ReplicationSlotsShmemInit ( void  )

Definition at line 132 of file slot.c.

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, ReplicationSlotsShmemSize(), ShmemInitStruct(), and SpinLockInit.

Referenced by CreateSharedMemoryAndSemaphores().

133 {
134  bool found;
135 
136  if (max_replication_slots == 0)
137  return;
138 
140  ShmemInitStruct("ReplicationSlot Ctl", ReplicationSlotsShmemSize(),
141  &found);
142 
143  if (!found)
144  {
145  int i;
146 
147  /* First time through, so initialize */
149 
150  for (i = 0; i < max_replication_slots; i++)
151  {
153 
154  /* everything else is zeroed by the memset above */
155  SpinLockInit(&slot->mutex);
159  }
160  }
161 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
Size ReplicationSlotsShmemSize(void)
Definition: slot.c:114
#define SpinLockInit(lock)
Definition: spin.h:60
#define MemSet(start, val, len)
Definition: c.h:971
void ConditionVariableInit(ConditionVariable *cv)
void * ShmemInitStruct(const char *name, Size size, bool *foundPtr)
Definition: shmem.c:392
void LWLockInitialize(LWLock *lock, int tranche_id)
Definition: lwlock.c:745
int max_replication_slots
Definition: slot.c:99
ConditionVariable active_cv
Definition: slot.h:146
int i
ReplicationSlot replication_slots[1]
Definition: slot.h:173
slock_t mutex
Definition: slot.h:113
LWLock io_in_progress_lock
Definition: slot.h:143

◆ ReplicationSlotsShmemSize()

Size ReplicationSlotsShmemSize ( void  )

Definition at line 114 of file slot.c.

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

Referenced by CreateSharedMemoryAndSemaphores(), and ReplicationSlotsShmemInit().

115 {
116  Size size = 0;
117 
118  if (max_replication_slots == 0)
119  return size;
120 
121  size = offsetof(ReplicationSlotCtlData, replication_slots);
122  size = add_size(size,
124 
125  return size;
126 }
Size mul_size(Size s1, Size s2)
Definition: shmem.c:515
Size add_size(Size s1, Size s2)
Definition: shmem.c:498
int max_replication_slots
Definition: slot.c:99
size_t Size
Definition: c.h:466
#define offsetof(type, field)
Definition: c.h:661

◆ ReplicationSlotValidateName()

bool ReplicationSlotValidateName ( const char *  name,
int  elevel 
)

Definition at line 172 of file slot.c.

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

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

173 {
174  const char *cp;
175 
176  if (strlen(name) == 0)
177  {
178  ereport(elevel,
179  (errcode(ERRCODE_INVALID_NAME),
180  errmsg("replication slot name \"%s\" is too short",
181  name)));
182  return false;
183  }
184 
185  if (strlen(name) >= NAMEDATALEN)
186  {
187  ereport(elevel,
188  (errcode(ERRCODE_NAME_TOO_LONG),
189  errmsg("replication slot name \"%s\" is too long",
190  name)));
191  return false;
192  }
193 
194  for (cp = name; *cp; cp++)
195  {
196  if (!((*cp >= 'a' && *cp <= 'z')
197  || (*cp >= '0' && *cp <= '9')
198  || (*cp == '_')))
199  {
200  ereport(elevel,
201  (errcode(ERRCODE_INVALID_NAME),
202  errmsg("replication slot name \"%s\" contains invalid character",
203  name),
204  errhint("Replication slot names may only contain lower case letters, numbers, and the underscore character.")));
205  return false;
206  }
207  }
208  return true;
209 }
int errhint(const char *fmt,...)
Definition: elog.c:1071
int errcode(int sqlerrcode)
Definition: elog.c:610
#define NAMEDATALEN
static int elevel
Definition: vacuumlazy.c:323
#define ereport(elevel,...)
Definition: elog.h:144
const char * name
Definition: encode.c:555
int errmsg(const char *fmt,...)
Definition: elog.c:824

◆ StartupReplicationSlots()

void StartupReplicationSlots ( void  )

Definition at line 1199 of file slot.c.

References AllocateDir(), dirent::d_name, DEBUG1, elog, ereport, errmsg(), FreeDir(), fsync_fname(), lstat, max_replication_slots, MAXPGPATH, pg_str_endswith(), ReadDir(), ReplicationSlotsComputeRequiredLSN(), ReplicationSlotsComputeRequiredXmin(), RestoreSlotFromDisk(), rmtree(), S_ISDIR, snprintf, stat, and WARNING.

Referenced by StartupXLOG().

1200 {
1201  DIR *replication_dir;
1202  struct dirent *replication_de;
1203 
1204  elog(DEBUG1, "starting up replication slots");
1205 
1206  /* restore all slots by iterating over all on-disk entries */
1207  replication_dir = AllocateDir("pg_replslot");
1208  while ((replication_de = ReadDir(replication_dir, "pg_replslot")) != NULL)
1209  {
1210  struct stat statbuf;
1211  char path[MAXPGPATH + 12];
1212 
1213  if (strcmp(replication_de->d_name, ".") == 0 ||
1214  strcmp(replication_de->d_name, "..") == 0)
1215  continue;
1216 
1217  snprintf(path, sizeof(path), "pg_replslot/%s", replication_de->d_name);
1218 
1219  /* we're only creating directories here, skip if it's not our's */
1220  if (lstat(path, &statbuf) == 0 && !S_ISDIR(statbuf.st_mode))
1221  continue;
1222 
1223  /* we crashed while a slot was being setup or deleted, clean up */
1224  if (pg_str_endswith(replication_de->d_name, ".tmp"))
1225  {
1226  if (!rmtree(path, true))
1227  {
1228  ereport(WARNING,
1229  (errmsg("could not remove directory \"%s\"",
1230  path)));
1231  continue;
1232  }
1233  fsync_fname("pg_replslot", true);
1234  continue;
1235  }
1236 
1237  /* looks like a slot in a normal state, restore */
1238  RestoreSlotFromDisk(replication_de->d_name);
1239  }
1240  FreeDir(replication_dir);
1241 
1242  /* currently no slots exist, we're done. */
1243  if (max_replication_slots <= 0)
1244  return;
1245 
1246  /* Now that we have recovered all the data, compute replication xmin */
1249 }
static void RestoreSlotFromDisk(const char *name)
Definition: slot.c:1474
#define DEBUG1
Definition: elog.h:25
bool pg_str_endswith(const char *str, const char *end)
Definition: string.c:31
void fsync_fname(const char *fname, bool isdir)
Definition: fd.c:630
Definition: dirent.h:9
void ReplicationSlotsComputeRequiredLSN(void)
Definition: slot.c:761
Definition: dirent.c:25
#define MAXPGPATH
DIR * AllocateDir(const char *dirname)
Definition: fd.c:2581
#define WARNING
Definition: elog.h:40
#define stat(a, b)
Definition: win32_port.h:255
bool rmtree(const char *path, bool rmtopdir)
Definition: rmtree.c:42
#define ereport(elevel,...)
Definition: elog.h:144
int max_replication_slots
Definition: slot.c:99
struct dirent * ReadDir(DIR *dir, const char *dirname)
Definition: fd.c:2647
#define S_ISDIR(m)
Definition: win32_port.h:296
#define lstat(path, sb)
Definition: win32_port.h:244
int errmsg(const char *fmt,...)
Definition: elog.c:824
#define elog(elevel,...)
Definition: elog.h:214
char d_name[MAX_PATH]
Definition: dirent.h:14
#define snprintf
Definition: port.h:193
void ReplicationSlotsComputeRequiredXmin(bool already_locked)
Definition: slot.c:711
int FreeDir(DIR *dir)
Definition: fd.c:2699

Variable Documentation

◆ max_replication_slots

◆ MyReplicationSlot

◆ ReplicationSlotCtl

PGDLLIMPORT ReplicationSlotCtlData* ReplicationSlotCtl

Definition at line 93 of file slot.c.

Referenced by copy_replication_slot(), and pg_get_replication_slots().