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 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 p)
 
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)
 
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 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 153 of file slot.h.

Referenced by CreateDecodingContext(), and CreateInitDecodingContext().

Typedef Documentation

◆ ReplicationSlot

◆ ReplicationSlotCtlData

◆ ReplicationSlotPersistency

◆ ReplicationSlotPersistentData

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

Function Documentation

◆ CheckPointReplicationSlots()

void CheckPointReplicationSlots ( void  )

Definition at line 1074 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().

1075 {
1076  int i;
1077 
1078  elog(DEBUG1, "performing replication slot checkpoint");
1079 
1080  /*
1081  * Prevent any slot from being created/dropped while we're active. As we
1082  * explicitly do *not* want to block iterating over replication_slots or
1083  * acquiring a slot we cannot take the control lock - but that's OK,
1084  * because holding ReplicationSlotAllocationLock is strictly stronger, and
1085  * enough to guarantee that nobody can change the in_use bits on us.
1086  */
1087  LWLockAcquire(ReplicationSlotAllocationLock, LW_SHARED);
1088 
1089  for (i = 0; i < max_replication_slots; i++)
1090  {
1092  char path[MAXPGPATH];
1093 
1094  if (!s->in_use)
1095  continue;
1096 
1097  /* save the slot to disk, locking is handled in SaveSlotToPath() */
1098  sprintf(path, "pg_replslot/%s", NameStr(s->data.name));
1099  SaveSlotToPath(s, path, LOG);
1100  }
1101  LWLockRelease(ReplicationSlotAllocationLock);
1102 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
#define DEBUG1
Definition: elog.h:25
ReplicationSlotPersistentData data
Definition: slot.h:132
#define LOG
Definition: elog.h:26
static void SaveSlotToPath(ReplicationSlot *slot, const char *path, int elevel)
Definition: slot.c:1230
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1726
#define sprintf
Definition: port.h:194
#define MAXPGPATH
bool in_use
Definition: slot.h:108
int max_replication_slots
Definition: slot.c:99
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1122
#define elog(elevel,...)
Definition: elog.h:226
int i
#define NameStr(name)
Definition: c.h:609
ReplicationSlot replication_slots[1]
Definition: slot.h:165

◆ CheckSlotRequirements()

void CheckSlotRequirements ( void  )

Definition at line 972 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().

973 {
974  /*
975  * NB: Adding a new requirement likely means that RestoreSlotFromDisk()
976  * needs the same check.
977  */
978 
979  if (max_replication_slots == 0)
980  ereport(ERROR,
981  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
982  (errmsg("replication slots can only be used if max_replication_slots > 0"))));
983 
985  ereport(ERROR,
986  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
987  errmsg("replication slots can only be used if wal_level >= replica")));
988 }
int wal_level
Definition: xlog.c:103
int errcode(int sqlerrcode)
Definition: elog.c:570
#define ERROR
Definition: elog.h:43
#define ereport(elevel, rest)
Definition: elog.h:141
int max_replication_slots
Definition: slot.c:99
int errmsg(const char *fmt,...)
Definition: elog.c:784

◆ ReplicationSlotAcquire()

void ReplicationSlotAcquire ( const char *  name,
bool  nowait 
)

Definition at line 330 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, SpinLockAcquire, SpinLockRelease, and WAIT_EVENT_REPLICATION_SLOT_DROP.

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

331 {
332  ReplicationSlot *slot;
333  int active_pid;
334  int i;
335 
336 retry:
337  Assert(MyReplicationSlot == NULL);
338 
339  /*
340  * Search for the named slot and mark it active if we find it. If the
341  * slot is already active, we exit the loop with active_pid set to the PID
342  * of the backend that owns it.
343  */
344  active_pid = 0;
345  slot = NULL;
346  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
347  for (i = 0; i < max_replication_slots; i++)
348  {
350 
351  if (s->in_use && strcmp(name, NameStr(s->data.name)) == 0)
352  {
353  /*
354  * This is the slot we want; check if it's active under some other
355  * process. In single user mode, we don't need this check.
356  */
357  if (IsUnderPostmaster)
358  {
359  /*
360  * Get ready to sleep on it in case it is active. (We may end
361  * up not sleeping, but we don't want to do this while holding
362  * the spinlock.)
363  */
365 
366  SpinLockAcquire(&s->mutex);
367 
368  active_pid = s->active_pid;
369  if (active_pid == 0)
370  active_pid = s->active_pid = MyProcPid;
371 
372  SpinLockRelease(&s->mutex);
373  }
374  else
375  active_pid = MyProcPid;
376  slot = s;
377 
378  break;
379  }
380  }
381  LWLockRelease(ReplicationSlotControlLock);
382 
383  /* If we did not find the slot, error out. */
384  if (slot == NULL)
385  ereport(ERROR,
386  (errcode(ERRCODE_UNDEFINED_OBJECT),
387  errmsg("replication slot \"%s\" does not exist", name)));
388 
389  /*
390  * If we found the slot but it's already active in another backend, we
391  * either error out or retry after a short wait, as caller specified.
392  */
393  if (active_pid != MyProcPid)
394  {
395  if (nowait)
396  ereport(ERROR,
397  (errcode(ERRCODE_OBJECT_IN_USE),
398  errmsg("replication slot \"%s\" is active for PID %d",
399  name, active_pid)));
400 
401  /* Wait here until we get signaled, and then restart */
405  goto retry;
406  }
407  else
408  ConditionVariableCancelSleep(); /* no sleep needed after all */
409 
410  /* Let everybody know we've modified this slot */
412 
413  /* We made this slot active, so it's ours now. */
414  MyReplicationSlot = slot;
415 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
int MyProcPid
Definition: globals.c:40
void ConditionVariableBroadcast(ConditionVariable *cv)
int errcode(int sqlerrcode)
Definition: elog.c:570
ReplicationSlotPersistentData data
Definition: slot.h:132
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1726
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
#define ereport(elevel, rest)
Definition: elog.h:141
bool in_use
Definition: slot.h:108
#define SpinLockRelease(lock)
Definition: spin.h:64
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:732
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1122
ConditionVariable active_cv
Definition: slot.h:138
const char * name
Definition: encode.c:521
int errmsg(const char *fmt,...)
Definition: elog.c:784
pid_t active_pid
Definition: slot.h:111
int i
#define NameStr(name)
Definition: c.h:609
ReplicationSlot replication_slots[1]
Definition: slot.h:165
slock_t mutex
Definition: slot.h:105

◆ ReplicationSlotCleanup()

void ReplicationSlotCleanup ( void  )

Definition at line 479 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().

480 {
481  int i;
482 
483  Assert(MyReplicationSlot == NULL);
484 
485 restart:
486  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
487  for (i = 0; i < max_replication_slots; i++)
488  {
490 
491  if (!s->in_use)
492  continue;
493 
494  SpinLockAcquire(&s->mutex);
495  if (s->active_pid == MyProcPid)
496  {
498  SpinLockRelease(&s->mutex);
499  LWLockRelease(ReplicationSlotControlLock); /* avoid deadlock */
500 
502 
504  goto restart;
505  }
506  else
507  SpinLockRelease(&s->mutex);
508  }
509 
510  LWLockRelease(ReplicationSlotControlLock);
511 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
int MyProcPid
Definition: globals.c:40
ReplicationSlotPersistency persistency
Definition: slot.h:53
void ConditionVariableBroadcast(ConditionVariable *cv)
static void ReplicationSlotDropPtr(ReplicationSlot *slot)
Definition: slot.c:547
ReplicationSlotPersistentData data
Definition: slot.h:132
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1726
#define SpinLockAcquire(lock)
Definition: spin.h:62
bool in_use
Definition: slot.h:108
#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:732
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1122
ConditionVariable active_cv
Definition: slot.h:138
pid_t active_pid
Definition: slot.h:111
int i
ReplicationSlot replication_slots[1]
Definition: slot.h:165
slock_t mutex
Definition: slot.h:105

◆ ReplicationSlotCreate()

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

Definition at line 221 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().

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

◆ ReplicationSlotDrop()

void ReplicationSlotDrop ( const char *  name,
bool  nowait 
)

Definition at line 517 of file slot.c.

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

Referenced by DropReplicationSlot(), and pg_drop_replication_slot().

518 {
519  Assert(MyReplicationSlot == NULL);
520 
521  ReplicationSlotAcquire(name, nowait);
522 
524 }
void ReplicationSlotAcquire(const char *name, bool nowait)
Definition: slot.c:330
static void ReplicationSlotDropAcquired(void)
Definition: slot.c:530
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
#define Assert(condition)
Definition: c.h:732
const char * name
Definition: encode.c:521

◆ ReplicationSlotMarkDirty()

◆ ReplicationSlotPersist()

void ReplicationSlotPersist ( void  )

Definition at line 680 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().

681 {
683 
684  Assert(slot != NULL);
686 
687  SpinLockAcquire(&slot->mutex);
689  SpinLockRelease(&slot->mutex);
690 
693 }
ReplicationSlotPersistency persistency
Definition: slot.h:53
void ReplicationSlotSave(void)
Definition: slot.c:645
ReplicationSlotPersistentData data
Definition: slot.h:132
#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:732
slock_t mutex
Definition: slot.h:105
void ReplicationSlotMarkDirty(void)
Definition: slot.c:663

◆ ReplicationSlotRelease()

void ReplicationSlotRelease ( void  )

Definition at line 424 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(), 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().

425 {
427 
428  Assert(slot != NULL && slot->active_pid != 0);
429 
430  if (slot->data.persistency == RS_EPHEMERAL)
431  {
432  /*
433  * Delete the slot. There is no !PANIC case where this is allowed to
434  * fail, all that may happen is an incomplete cleanup of the on-disk
435  * data.
436  */
438  }
439 
440  /*
441  * If slot needed to temporarily restrain both data and catalog xmin to
442  * create the catalog snapshot, remove that temporary constraint.
443  * Snapshots can only be exported while the initial snapshot is still
444  * acquired.
445  */
446  if (!TransactionIdIsValid(slot->data.xmin) &&
448  {
449  SpinLockAcquire(&slot->mutex);
451  SpinLockRelease(&slot->mutex);
453  }
454 
455  if (slot->data.persistency == RS_PERSISTENT)
456  {
457  /*
458  * Mark persistent slot inactive. We're not freeing it, just
459  * disconnecting, but wake up others that may be waiting for it.
460  */
461  SpinLockAcquire(&slot->mutex);
462  slot->active_pid = 0;
463  SpinLockRelease(&slot->mutex);
465  }
466 
467  MyReplicationSlot = NULL;
468 
469  /* might not have been set when we've been a plain slot */
470  LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
472  LWLockRelease(ProcArrayLock);
473 }
#define PROC_IN_LOGICAL_DECODING
Definition: proc.h:57
ReplicationSlotPersistency persistency
Definition: slot.h:53
void ConditionVariableBroadcast(ConditionVariable *cv)
ReplicationSlotPersistentData data
Definition: slot.h:132
PGXACT * MyPgXact
Definition: proc.c:69
uint8 vacuumFlags
Definition: proc.h:233
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1726
#define SpinLockAcquire(lock)
Definition: spin.h:62
static void ReplicationSlotDropAcquired(void)
Definition: slot.c:530
TransactionId effective_xmin
Definition: slot.h:128
#define InvalidTransactionId
Definition: transam.h:31
TransactionId xmin
Definition: slot.h:61
#define SpinLockRelease(lock)
Definition: spin.h:64
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
#define Assert(condition)
Definition: c.h:732
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1122
ConditionVariable active_cv
Definition: slot.h:138
pid_t active_pid
Definition: slot.h:111
slock_t mutex
Definition: slot.h:105
#define TransactionIdIsValid(xid)
Definition: transam.h:41
void ReplicationSlotsComputeRequiredXmin(bool already_locked)
Definition: slot.c:702

◆ ReplicationSlotReserveWal()

void ReplicationSlotReserveWal ( void  )

Definition at line 997 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().

998 {
1000 
1001  Assert(slot != NULL);
1003 
1004  /*
1005  * The replication slot mechanism is used to prevent removal of required
1006  * WAL. As there is no interlock between this routine and checkpoints, WAL
1007  * segments could concurrently be removed when a now stale return value of
1008  * ReplicationSlotsComputeRequiredLSN() is used. In the unlikely case that
1009  * this happens we'll just retry.
1010  */
1011  while (true)
1012  {
1013  XLogSegNo segno;
1014  XLogRecPtr restart_lsn;
1015 
1016  /*
1017  * For logical slots log a standby snapshot and start logical decoding
1018  * at exactly that position. That allows the slot to start up more
1019  * quickly.
1020  *
1021  * That's not needed (or indeed helpful) for physical slots as they'll
1022  * start replay at the last logged checkpoint anyway. Instead return
1023  * the location of the last redo LSN. While that slightly increases
1024  * the chance that we have to retry, it's where a base backup has to
1025  * start replay at.
1026  */
1027  if (!RecoveryInProgress() && SlotIsLogical(slot))
1028  {
1029  XLogRecPtr flushptr;
1030 
1031  /* start at current insert position */
1032  restart_lsn = GetXLogInsertRecPtr();
1033  SpinLockAcquire(&slot->mutex);
1034  slot->data.restart_lsn = restart_lsn;
1035  SpinLockRelease(&slot->mutex);
1036 
1037  /* make sure we have enough information to start */
1038  flushptr = LogStandbySnapshot();
1039 
1040  /* and make sure it's fsynced to disk */
1041  XLogFlush(flushptr);
1042  }
1043  else
1044  {
1045  restart_lsn = GetRedoRecPtr();
1046  SpinLockAcquire(&slot->mutex);
1047  slot->data.restart_lsn = restart_lsn;
1048  SpinLockRelease(&slot->mutex);
1049  }
1050 
1051  /* prevent WAL removal as fast as possible */
1053 
1054  /*
1055  * If all required WAL is still there, great, otherwise retry. The
1056  * slot should prevent further removal of WAL, unless there's a
1057  * concurrent ReplicationSlotsComputeRequiredLSN() after we've written
1058  * the new restart_lsn above, so normally we should never need to loop
1059  * more than twice.
1060  */
1062  if (XLogGetLastRemovedSegno() < segno)
1063  break;
1064  }
1065 }
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
int wal_segment_size
Definition: xlog.c:112
ReplicationSlotPersistentData data
Definition: slot.h:132
bool RecoveryInProgress(void)
Definition: xlog.c:7898
void XLogFlush(XLogRecPtr record)
Definition: xlog.c:2798
#define SpinLockAcquire(lock)
Definition: spin.h:62
XLogSegNo XLogGetLastRemovedSegno(void)
Definition: xlog.c:3877
void ReplicationSlotsComputeRequiredLSN(void)
Definition: slot.c:748
XLogRecPtr LogStandbySnapshot(void)
Definition: standby.c:901
XLogRecPtr GetXLogInsertRecPtr(void)
Definition: xlog.c:11149
uint64 XLogSegNo
Definition: xlogdefs.h:41
#define SlotIsLogical(slot)
Definition: slot.h:154
#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:732
XLogRecPtr restart_lsn
Definition: slot.h:72
XLogRecPtr GetRedoRecPtr(void)
Definition: xlog.c:8171
slock_t mutex
Definition: slot.h:105
#define XLByteToSeg(xlrp, logSegNo, wal_segsz_bytes)

◆ ReplicationSlotSave()

void ReplicationSlotSave ( void  )

Definition at line 645 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(), pg_replication_slot_advance(), and ReplicationSlotPersist().

646 {
647  char path[MAXPGPATH];
648 
649  Assert(MyReplicationSlot != NULL);
650 
651  sprintf(path, "pg_replslot/%s", NameStr(MyReplicationSlot->data.name));
653 }
ReplicationSlotPersistentData data
Definition: slot.h:132
static void SaveSlotToPath(ReplicationSlot *slot, const char *path, int elevel)
Definition: slot.c:1230
#define sprintf
Definition: port.h:194
#define ERROR
Definition: elog.h:43
#define MAXPGPATH
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
#define Assert(condition)
Definition: c.h:732
#define NameStr(name)
Definition: c.h:609

◆ ReplicationSlotsComputeLogicalRestartLSN()

XLogRecPtr ReplicationSlotsComputeLogicalRestartLSN ( void  )

Definition at line 791 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().

792 {
793  XLogRecPtr result = InvalidXLogRecPtr;
794  int i;
795 
796  if (max_replication_slots <= 0)
797  return InvalidXLogRecPtr;
798 
799  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
800 
801  for (i = 0; i < max_replication_slots; i++)
802  {
803  ReplicationSlot *s;
804  XLogRecPtr restart_lsn;
805 
807 
808  /* cannot change while ReplicationSlotCtlLock is held */
809  if (!s->in_use)
810  continue;
811 
812  /* we're only interested in logical slots */
813  if (!SlotIsLogical(s))
814  continue;
815 
816  /* read once, it's ok if it increases while we're checking */
817  SpinLockAcquire(&s->mutex);
818  restart_lsn = s->data.restart_lsn;
819  SpinLockRelease(&s->mutex);
820 
821  if (result == InvalidXLogRecPtr ||
822  restart_lsn < result)
823  result = restart_lsn;
824  }
825 
826  LWLockRelease(ReplicationSlotControlLock);
827 
828  return result;
829 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
ReplicationSlotPersistentData data
Definition: slot.h:132
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1726
#define SpinLockAcquire(lock)
Definition: spin.h:62
#define SlotIsLogical(slot)
Definition: slot.h:154
bool in_use
Definition: slot.h:108
#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:72
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1122
int i
ReplicationSlot replication_slots[1]
Definition: slot.h:165
slock_t mutex
Definition: slot.h:105

◆ ReplicationSlotsComputeRequiredLSN()

void ReplicationSlotsComputeRequiredLSN ( void  )

Definition at line 748 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(), pg_replication_slot_advance(), PhysicalConfirmReceivedLocation(), ReplicationSlotDropPtr(), ReplicationSlotReserveWal(), and StartupReplicationSlots().

749 {
750  int i;
751  XLogRecPtr min_required = InvalidXLogRecPtr;
752 
753  Assert(ReplicationSlotCtl != NULL);
754 
755  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
756  for (i = 0; i < max_replication_slots; i++)
757  {
759  XLogRecPtr restart_lsn;
760 
761  if (!s->in_use)
762  continue;
763 
764  SpinLockAcquire(&s->mutex);
765  restart_lsn = s->data.restart_lsn;
766  SpinLockRelease(&s->mutex);
767 
768  if (restart_lsn != InvalidXLogRecPtr &&
769  (min_required == InvalidXLogRecPtr ||
770  restart_lsn < min_required))
771  min_required = restart_lsn;
772  }
773  LWLockRelease(ReplicationSlotControlLock);
774 
775  XLogSetReplicationSlotMinimumLSN(min_required);
776 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
ReplicationSlotPersistentData data
Definition: slot.h:132
void XLogSetReplicationSlotMinimumLSN(XLogRecPtr lsn)
Definition: xlog.c:2677
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1726
#define SpinLockAcquire(lock)
Definition: spin.h:62
bool in_use
Definition: slot.h:108
#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:732
XLogRecPtr restart_lsn
Definition: slot.h:72
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1122
int i
ReplicationSlot replication_slots[1]
Definition: slot.h:165
slock_t mutex
Definition: slot.h:105

◆ ReplicationSlotsComputeRequiredXmin()

void ReplicationSlotsComputeRequiredXmin ( bool  already_locked)

Definition at line 702 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(), pg_replication_slot_advance(), PhysicalReplicationSlotNewXmin(), ReplicationSlotDropPtr(), ReplicationSlotRelease(), and StartupReplicationSlots().

703 {
704  int i;
706  TransactionId agg_catalog_xmin = InvalidTransactionId;
707 
708  Assert(ReplicationSlotCtl != NULL);
709 
710  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
711 
712  for (i = 0; i < max_replication_slots; i++)
713  {
715  TransactionId effective_xmin;
716  TransactionId effective_catalog_xmin;
717 
718  if (!s->in_use)
719  continue;
720 
721  SpinLockAcquire(&s->mutex);
722  effective_xmin = s->effective_xmin;
723  effective_catalog_xmin = s->effective_catalog_xmin;
724  SpinLockRelease(&s->mutex);
725 
726  /* check the data xmin */
727  if (TransactionIdIsValid(effective_xmin) &&
728  (!TransactionIdIsValid(agg_xmin) ||
729  TransactionIdPrecedes(effective_xmin, agg_xmin)))
730  agg_xmin = effective_xmin;
731 
732  /* check the catalog xmin */
733  if (TransactionIdIsValid(effective_catalog_xmin) &&
734  (!TransactionIdIsValid(agg_catalog_xmin) ||
735  TransactionIdPrecedes(effective_catalog_xmin, agg_catalog_xmin)))
736  agg_catalog_xmin = effective_catalog_xmin;
737  }
738 
739  LWLockRelease(ReplicationSlotControlLock);
740 
741  ProcArraySetReplicationSlotXmin(agg_xmin, agg_catalog_xmin, already_locked);
742 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
uint32 TransactionId
Definition: c.h:507
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1726
#define SpinLockAcquire(lock)
Definition: spin.h:62
TransactionId effective_xmin
Definition: slot.h:128
#define InvalidTransactionId
Definition: transam.h:31
bool TransactionIdPrecedes(TransactionId id1, TransactionId id2)
Definition: transam.c:300
bool in_use
Definition: slot.h:108
#define SpinLockRelease(lock)
Definition: spin.h:64
TransactionId effective_catalog_xmin
Definition: slot.h:129
int max_replication_slots
Definition: slot.c:99
#define Assert(condition)
Definition: c.h:732
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1122
int i
void ProcArraySetReplicationSlotXmin(TransactionId xmin, TransactionId catalog_xmin, bool already_locked)
Definition: procarray.c:2981
ReplicationSlot replication_slots[1]
Definition: slot.h:165
slock_t mutex
Definition: slot.h:105
#define TransactionIdIsValid(xid)
Definition: transam.h:41

◆ ReplicationSlotsCountDBSlots()

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

Definition at line 840 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().

841 {
842  int i;
843 
844  *nslots = *nactive = 0;
845 
846  if (max_replication_slots <= 0)
847  return false;
848 
849  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
850  for (i = 0; i < max_replication_slots; i++)
851  {
852  ReplicationSlot *s;
853 
855 
856  /* cannot change while ReplicationSlotCtlLock is held */
857  if (!s->in_use)
858  continue;
859 
860  /* only logical slots are database specific, skip */
861  if (!SlotIsLogical(s))
862  continue;
863 
864  /* not our database, skip */
865  if (s->data.database != dboid)
866  continue;
867 
868  /* count slots with spinlock held */
869  SpinLockAcquire(&s->mutex);
870  (*nslots)++;
871  if (s->active_pid != 0)
872  (*nactive)++;
873  SpinLockRelease(&s->mutex);
874  }
875  LWLockRelease(ReplicationSlotControlLock);
876 
877  if (*nslots > 0)
878  return true;
879  return false;
880 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
ReplicationSlotPersistentData data
Definition: slot.h:132
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1726
#define SpinLockAcquire(lock)
Definition: spin.h:62
#define SlotIsLogical(slot)
Definition: slot.h:154
bool in_use
Definition: slot.h:108
#define SpinLockRelease(lock)
Definition: spin.h:64
int max_replication_slots
Definition: slot.c:99
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1122
pid_t active_pid
Definition: slot.h:111
int i
ReplicationSlot replication_slots[1]
Definition: slot.h:165
slock_t mutex
Definition: slot.h:105

◆ ReplicationSlotsDropDBSlots()

void ReplicationSlotsDropDBSlots ( Oid  dboid)

Definition at line 896 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().

897 {
898  int i;
899 
900  if (max_replication_slots <= 0)
901  return;
902 
903 restart:
904  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
905  for (i = 0; i < max_replication_slots; i++)
906  {
907  ReplicationSlot *s;
908  char *slotname;
909  int active_pid;
910 
912 
913  /* cannot change while ReplicationSlotCtlLock is held */
914  if (!s->in_use)
915  continue;
916 
917  /* only logical slots are database specific, skip */
918  if (!SlotIsLogical(s))
919  continue;
920 
921  /* not our database, skip */
922  if (s->data.database != dboid)
923  continue;
924 
925  /* acquire slot, so ReplicationSlotDropAcquired can be reused */
926  SpinLockAcquire(&s->mutex);
927  /* can't change while ReplicationSlotControlLock is held */
928  slotname = NameStr(s->data.name);
929  active_pid = s->active_pid;
930  if (active_pid == 0)
931  {
932  MyReplicationSlot = s;
933  s->active_pid = MyProcPid;
934  }
935  SpinLockRelease(&s->mutex);
936 
937  /*
938  * Even though we hold an exclusive lock on the database object a
939  * logical slot for that DB can still be active, e.g. if it's
940  * concurrently being dropped by a backend connected to another DB.
941  *
942  * That's fairly unlikely in practice, so we'll just bail out.
943  */
944  if (active_pid)
945  ereport(ERROR,
946  (errcode(ERRCODE_OBJECT_IN_USE),
947  errmsg("replication slot \"%s\" is active for PID %d",
948  slotname, active_pid)));
949 
950  /*
951  * To avoid duplicating ReplicationSlotDropAcquired() and to avoid
952  * holding ReplicationSlotControlLock over filesystem operations,
953  * release ReplicationSlotControlLock and use
954  * ReplicationSlotDropAcquired.
955  *
956  * As that means the set of slots could change, restart scan from the
957  * beginning each time we release the lock.
958  */
959  LWLockRelease(ReplicationSlotControlLock);
961  goto restart;
962  }
963  LWLockRelease(ReplicationSlotControlLock);
964 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
int MyProcPid
Definition: globals.c:40
int errcode(int sqlerrcode)
Definition: elog.c:570
ReplicationSlotPersistentData data
Definition: slot.h:132
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1726
#define SpinLockAcquire(lock)
Definition: spin.h:62
static void ReplicationSlotDropAcquired(void)
Definition: slot.c:530
#define ERROR
Definition: elog.h:43
#define SlotIsLogical(slot)
Definition: slot.h:154
#define ereport(elevel, rest)
Definition: elog.h:141
bool in_use
Definition: slot.h:108
#define SpinLockRelease(lock)
Definition: spin.h:64
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
int max_replication_slots
Definition: slot.c:99
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1122
int errmsg(const char *fmt,...)
Definition: elog.c:784
pid_t active_pid
Definition: slot.h:111
int i
#define NameStr(name)
Definition: c.h:609
ReplicationSlot replication_slots[1]
Definition: slot.h:165
slock_t mutex
Definition: slot.h:105

◆ ReplicationSlotsShmemInit()

void ReplicationSlotsShmemInit ( void  )

Definition at line 132 of file slot.c.

References ReplicationSlot::active_cv, ConditionVariableInit(), i, ReplicationSlot::io_in_progress_lock, LWLockInitialize(), LWLockRegisterTranche(), LWTRANCHE_REPLICATION_SLOT_IO_IN_PROGRESS, 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 
144  "replication_slot_io");
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);
161  }
162  }
163 }
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:955
void LWLockRegisterTranche(int tranche_id, const char *tranche_name)
Definition: lwlock.c:603
void ConditionVariableInit(ConditionVariable *cv)
void * ShmemInitStruct(const char *name, Size size, bool *foundPtr)
Definition: shmem.c:372
void LWLockInitialize(LWLock *lock, int tranche_id)
Definition: lwlock.c:678
int max_replication_slots
Definition: slot.c:99
ConditionVariable active_cv
Definition: slot.h:138
int i
ReplicationSlot replication_slots[1]
Definition: slot.h:165
slock_t mutex
Definition: slot.h:105
LWLock io_in_progress_lock
Definition: slot.h:135

◆ 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:492
Size add_size(Size s1, Size s2)
Definition: shmem.c:475
int max_replication_slots
Definition: slot.c:99
size_t Size
Definition: c.h:466
#define offsetof(type, field)
Definition: c.h:655

◆ ReplicationSlotValidateName()

bool ReplicationSlotValidateName ( const char *  name,
int  elevel 
)

Definition at line 174 of file slot.c.

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

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

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

◆ StartupReplicationSlots()

void StartupReplicationSlots ( void  )

Definition at line 1109 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().

1110 {
1111  DIR *replication_dir;
1112  struct dirent *replication_de;
1113 
1114  elog(DEBUG1, "starting up replication slots");
1115 
1116  /* restore all slots by iterating over all on-disk entries */
1117  replication_dir = AllocateDir("pg_replslot");
1118  while ((replication_de = ReadDir(replication_dir, "pg_replslot")) != NULL)
1119  {
1120  struct stat statbuf;
1121  char path[MAXPGPATH + 12];
1122 
1123  if (strcmp(replication_de->d_name, ".") == 0 ||
1124  strcmp(replication_de->d_name, "..") == 0)
1125  continue;
1126 
1127  snprintf(path, sizeof(path), "pg_replslot/%s", replication_de->d_name);
1128 
1129  /* we're only creating directories here, skip if it's not our's */
1130  if (lstat(path, &statbuf) == 0 && !S_ISDIR(statbuf.st_mode))
1131  continue;
1132 
1133  /* we crashed while a slot was being setup or deleted, clean up */
1134  if (pg_str_endswith(replication_de->d_name, ".tmp"))
1135  {
1136  if (!rmtree(path, true))
1137  {
1138  ereport(WARNING,
1139  (errmsg("could not remove directory \"%s\"",
1140  path)));
1141  continue;
1142  }
1143  fsync_fname("pg_replslot", true);
1144  continue;
1145  }
1146 
1147  /* looks like a slot in a normal state, restore */
1148  RestoreSlotFromDisk(replication_de->d_name);
1149  }
1150  FreeDir(replication_dir);
1151 
1152  /* currently no slots exist, we're done. */
1153  if (max_replication_slots <= 0)
1154  return;
1155 
1156  /* Now that we have recovered all the data, compute replication xmin */
1159 }
static void RestoreSlotFromDisk(const char *name)
Definition: slot.c:1364
#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:580
Definition: dirent.h:9
void ReplicationSlotsComputeRequiredLSN(void)
Definition: slot.c:748
Definition: dirent.c:25
#define MAXPGPATH
DIR * AllocateDir(const char *dirname)
Definition: fd.c:2466
#define ereport(elevel, rest)
Definition: elog.h:141
#define WARNING
Definition: elog.h:40
#define stat(a, b)
Definition: win32_port.h:264
bool rmtree(const char *path, bool rmtopdir)
Definition: rmtree.c:42
int max_replication_slots
Definition: slot.c:99
struct dirent * ReadDir(DIR *dir, const char *dirname)
Definition: fd.c:2532
#define S_ISDIR(m)
Definition: win32_port.h:305
#define lstat(path, sb)
Definition: win32_port.h:253
int errmsg(const char *fmt,...)
Definition: elog.c:784
#define elog(elevel,...)
Definition: elog.h:226
char d_name[MAX_PATH]
Definition: dirent.h:14
#define snprintf
Definition: port.h:192
void ReplicationSlotsComputeRequiredXmin(bool already_locked)
Definition: slot.c:702
int FreeDir(DIR *dir)
Definition: fd.c:2584

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