PostgreSQL Source Code  git master
 All Data Structures Namespaces Files Functions Variables Typedefs Enumerations Enumerator Macros
slot.h File Reference
#include "fmgr.h"
#include "access/xlog.h"
#include "access/xlogreader.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)
 
void ReplicationSlotAcquire (const char *name)
 
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

ReplicationSlotCtlDataReplicationSlotCtl
 
PGDLLIMPORT ReplicationSlotMyReplicationSlot
 
PGDLLIMPORT int max_replication_slots
 

Macro Definition Documentation

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

Definition at line 133 of file slot.h.

Referenced by CreateDecodingContext(), and CreateInitDecodingContext().

Typedef Documentation

Enumeration Type Documentation

Enumerator
RS_PERSISTENT 
RS_EPHEMERAL 
RS_TEMPORARY 

Definition at line 28 of file slot.h.

29 {
ReplicationSlotPersistency
Definition: slot.h:28

Function Documentation

void CheckPointReplicationSlots ( void  )

Definition at line 996 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, and SaveSlotToPath().

Referenced by CheckPointGuts().

997 {
998  int i;
999 
1000  elog(DEBUG1, "performing replication slot checkpoint");
1001 
1002  /*
1003  * Prevent any slot from being created/dropped while we're active. As we
1004  * explicitly do *not* want to block iterating over replication_slots or
1005  * acquiring a slot we cannot take the control lock - but that's OK,
1006  * because holding ReplicationSlotAllocationLock is strictly stronger, and
1007  * enough to guarantee that nobody can change the in_use bits on us.
1008  */
1009  LWLockAcquire(ReplicationSlotAllocationLock, LW_SHARED);
1010 
1011  for (i = 0; i < max_replication_slots; i++)
1012  {
1014  char path[MAXPGPATH];
1015 
1016  if (!s->in_use)
1017  continue;
1018 
1019  /* save the slot to disk, locking is handled in SaveSlotToPath() */
1020  sprintf(path, "pg_replslot/%s", NameStr(s->data.name));
1021  SaveSlotToPath(s, path, LOG);
1022  }
1023  LWLockRelease(ReplicationSlotAllocationLock);
1024 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
#define DEBUG1
Definition: elog.h:25
ReplicationSlotPersistentData data
Definition: slot.h:115
#define LOG
Definition: elog.h:26
static void SaveSlotToPath(ReplicationSlot *slot, const char *path, int elevel)
Definition: slot.c:1151
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1715
#define MAXPGPATH
bool in_use
Definition: slot.h:91
int max_replication_slots
Definition: slot.c:99
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1111
int i
#define NameStr(name)
Definition: c.h:499
ReplicationSlot replication_slots[1]
Definition: slot.h:145
#define elog
Definition: elog.h:219
void CheckSlotRequirements ( void  )

Definition at line 906 of file slot.c.

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

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

907 {
908  if (max_replication_slots == 0)
909  ereport(ERROR,
910  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
911  (errmsg("replication slots can only be used if max_replication_slots > 0"))));
912 
914  ereport(ERROR,
915  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
916  errmsg("replication slots can only be used if wal_level >= replica")));
917 }
int wal_level
Definition: xlog.c:104
int errcode(int sqlerrcode)
Definition: elog.c:575
#define ERROR
Definition: elog.h:43
#define ereport(elevel, rest)
Definition: elog.h:122
int max_replication_slots
Definition: slot.c:99
int errmsg(const char *fmt,...)
Definition: elog.c:797
void ReplicationSlotAcquire ( const char *  name)

Definition at line 326 of file slot.c.

References ReplicationSlot::active_pid, Assert, ReplicationSlot::data, ereport, errcode(), errmsg(), ERROR, i, ReplicationSlot::in_use, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_replication_slots, ReplicationSlot::mutex, MyProcPid, ReplicationSlotPersistentData::name, NameStr, NULL, ReplicationSlotCtlData::replication_slots, ReplicationSlotValidateName(), SpinLockAcquire, and SpinLockRelease.

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

327 {
328  ReplicationSlot *slot = NULL;
329  int i;
330  int active_pid = 0; /* Keep compiler quiet */
331 
333 
335 
336  /* Search for the named slot and mark it active if we find it. */
337  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
338  for (i = 0; i < max_replication_slots; i++)
339  {
341 
342  if (s->in_use && strcmp(name, NameStr(s->data.name)) == 0)
343  {
344  SpinLockAcquire(&s->mutex);
345  active_pid = s->active_pid;
346  if (active_pid == 0)
347  active_pid = s->active_pid = MyProcPid;
348  SpinLockRelease(&s->mutex);
349  slot = s;
350  break;
351  }
352  }
353  LWLockRelease(ReplicationSlotControlLock);
354 
355  /* If we did not find the slot or it was already active, error out. */
356  if (slot == NULL)
357  ereport(ERROR,
358  (errcode(ERRCODE_UNDEFINED_OBJECT),
359  errmsg("replication slot \"%s\" does not exist", name)));
360  if (active_pid != MyProcPid)
361  ereport(ERROR,
362  (errcode(ERRCODE_OBJECT_IN_USE),
363  errmsg("replication slot \"%s\" is active for PID %d",
364  name, active_pid)));
365 
366  /* We made this slot active, so it's ours now. */
367  MyReplicationSlot = slot;
368 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
int MyProcPid
Definition: globals.c:38
int errcode(int sqlerrcode)
Definition: elog.c:575
ReplicationSlotPersistentData data
Definition: slot.h:115
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1715
bool ReplicationSlotValidateName(const char *name, int elevel)
Definition: slot.c:173
#define SpinLockAcquire(lock)
Definition: spin.h:62
#define ERROR
Definition: elog.h:43
#define ereport(elevel, rest)
Definition: elog.h:122
bool in_use
Definition: slot.h:91
#define SpinLockRelease(lock)
Definition: spin.h:64
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
int max_replication_slots
Definition: slot.c:99
#define NULL
Definition: c.h:229
#define Assert(condition)
Definition: c.h:675
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1111
const char * name
Definition: encode.c:521
int errmsg(const char *fmt,...)
Definition: elog.c:797
pid_t active_pid
Definition: slot.h:94
int i
#define NameStr(name)
Definition: c.h:499
ReplicationSlot replication_slots[1]
Definition: slot.h:145
slock_t mutex
Definition: slot.h:88
void ReplicationSlotCleanup ( void  )

Definition at line 429 of file slot.c.

References ReplicationSlot::active_pid, Assert, ReplicationSlot::data, i, ReplicationSlot::in_use, max_replication_slots, MyProcPid, NULL, ReplicationSlotPersistentData::persistency, ReplicationSlotCtlData::replication_slots, ReplicationSlotDropPtr(), and RS_TEMPORARY.

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

430 {
431  int i;
432 
434 
435  /*
436  * No need for locking as we are only interested in slots active in
437  * current process and those are not touched by other processes.
438  */
439  for (i = 0; i < max_replication_slots; i++)
440  {
442 
443  if (s->active_pid == MyProcPid)
444  {
446 
448  }
449  }
450 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
int MyProcPid
Definition: globals.c:38
ReplicationSlotPersistency persistency
Definition: slot.h:49
static void ReplicationSlotDropPtr(ReplicationSlot *slot)
Definition: slot.c:486
ReplicationSlotPersistentData data
Definition: slot.h:115
bool in_use
Definition: slot.h:91
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
int max_replication_slots
Definition: slot.c:99
#define NULL
Definition: c.h:229
#define Assert(condition)
Definition: c.h:675
pid_t active_pid
Definition: slot.h:94
int i
ReplicationSlot replication_slots[1]
Definition: slot.h:145
void ReplicationSlotCreate ( const char *  name,
bool  db_specific,
ReplicationSlotPersistency  p 
)

Definition at line 220 of file slot.c.

References ReplicationSlot::active_pid, Assert, ReplicationSlot::candidate_catalog_xmin, ReplicationSlot::candidate_restart_lsn, ReplicationSlot::candidate_restart_valid, ReplicationSlot::candidate_xmin_lsn, 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, NULL, ReplicationSlotPersistentData::persistency, ReplicationSlotCtlData::replication_slots, ReplicationSlotValidateName(), SpinLockAcquire, SpinLockRelease, and StrNCpy.

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

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

Definition at line 456 of file slot.c.

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

Referenced by DropReplicationSlot(), and pg_drop_replication_slot().

457 {
459 
461 
463 }
static void ReplicationSlotDropAcquired(void)
Definition: slot.c:469
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
#define NULL
Definition: c.h:229
#define Assert(condition)
Definition: c.h:675
void ReplicationSlotAcquire(const char *name)
Definition: slot.c:326
const char * name
Definition: encode.c:521
void ReplicationSlotMarkDirty ( void  )

Definition at line 597 of file slot.c.

References Assert, ReplicationSlot::dirty, ReplicationSlot::just_dirtied, ReplicationSlot::mutex, MyReplicationSlot, NULL, SpinLockAcquire, and SpinLockRelease.

Referenced by CreateInitDecodingContext(), CreateReplicationSlot(), LogicalConfirmReceivedLocation(), pg_create_physical_replication_slot(), pg_logical_slot_get_changes_guts(), PhysicalConfirmReceivedLocation(), PhysicalReplicationSlotNewXmin(), and ReplicationSlotPersist().

598 {
600 
602 
603  SpinLockAcquire(&slot->mutex);
605  MyReplicationSlot->dirty = true;
606  SpinLockRelease(&slot->mutex);
607 }
#define SpinLockAcquire(lock)
Definition: spin.h:62
#define SpinLockRelease(lock)
Definition: spin.h:64
bool just_dirtied
Definition: slot.h:97
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
#define NULL
Definition: c.h:229
#define Assert(condition)
Definition: c.h:675
slock_t mutex
Definition: slot.h:88
bool dirty
Definition: slot.h:98
void ReplicationSlotPersist ( void  )

Definition at line 614 of file slot.c.

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

Referenced by CreateReplicationSlot(), and pg_create_logical_replication_slot().

615 {
617 
618  Assert(slot != NULL);
620 
621  SpinLockAcquire(&slot->mutex);
623  SpinLockRelease(&slot->mutex);
624 
627 }
ReplicationSlotPersistency persistency
Definition: slot.h:49
void ReplicationSlotSave(void)
Definition: slot.c:579
ReplicationSlotPersistentData data
Definition: slot.h:115
#define SpinLockAcquire(lock)
Definition: spin.h:62
#define SpinLockRelease(lock)
Definition: spin.h:64
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
#define NULL
Definition: c.h:229
#define Assert(condition)
Definition: c.h:675
slock_t mutex
Definition: slot.h:88
void ReplicationSlotMarkDirty(void)
Definition: slot.c:597
void ReplicationSlotRelease ( void  )

Definition at line 375 of file slot.c.

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

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

376 {
378 
379  Assert(slot != NULL && slot->active_pid != 0);
380 
381  if (slot->data.persistency == RS_EPHEMERAL)
382  {
383  /*
384  * Delete the slot. There is no !PANIC case where this is allowed to
385  * fail, all that may happen is an incomplete cleanup of the on-disk
386  * data.
387  */
389  }
390  else if (slot->data.persistency == RS_PERSISTENT)
391  {
392  /*
393  * Mark persistent slot inactive. We're not freeing it, just
394  * disconnecting.
395  */
396  SpinLockAcquire(&slot->mutex);
397  slot->active_pid = 0;
398  SpinLockRelease(&slot->mutex);
399  }
400 
401 
402  /*
403  * If slot needed to temporarily restrain both data and catalog xmin to
404  * create the catalog snapshot, remove that temporary constraint.
405  * Snapshots can only be exported while the initial snapshot is still
406  * acquired.
407  */
408  if (!TransactionIdIsValid(slot->data.xmin) &&
410  {
411  SpinLockAcquire(&slot->mutex);
413  SpinLockRelease(&slot->mutex);
415  }
416 
418 
419  /* might not have been set when we've been a plain slot */
420  LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
422  LWLockRelease(ProcArrayLock);
423 }
#define PROC_IN_LOGICAL_DECODING
Definition: proc.h:56
ReplicationSlotPersistency persistency
Definition: slot.h:49
ReplicationSlotPersistentData data
Definition: slot.h:115
PGXACT * MyPgXact
Definition: proc.c:68
uint8 vacuumFlags
Definition: proc.h:218
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1715
#define SpinLockAcquire(lock)
Definition: spin.h:62
static void ReplicationSlotDropAcquired(void)
Definition: slot.c:469
TransactionId effective_xmin
Definition: slot.h:111
#define InvalidTransactionId
Definition: transam.h:31
TransactionId xmin
Definition: slot.h:57
#define SpinLockRelease(lock)
Definition: spin.h:64
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
#define NULL
Definition: c.h:229
#define Assert(condition)
Definition: c.h:675
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1111
pid_t active_pid
Definition: slot.h:94
slock_t mutex
Definition: slot.h:88
#define TransactionIdIsValid(xid)
Definition: transam.h:41
void ReplicationSlotsComputeRequiredXmin(bool already_locked)
Definition: slot.c:636
void ReplicationSlotReserveWal ( void  )

Definition at line 926 of file slot.c.

References Assert, ReplicationSlot::data, GetRedoRecPtr(), GetXLogInsertRecPtr(), InvalidXLogRecPtr, LogStandbySnapshot(), MyReplicationSlot, NULL, RecoveryInProgress(), ReplicationSlotsComputeRequiredLSN(), ReplicationSlotPersistentData::restart_lsn, SlotIsLogical, XLByteToSeg, XLogFlush(), and XLogGetLastRemovedSegno().

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

927 {
929 
930  Assert(slot != NULL);
932 
933  /*
934  * The replication slot mechanism is used to prevent removal of required
935  * WAL. As there is no interlock between this routine and checkpoints, WAL
936  * segments could concurrently be removed when a now stale return value of
937  * ReplicationSlotsComputeRequiredLSN() is used. In the unlikely case that
938  * this happens we'll just retry.
939  */
940  while (true)
941  {
942  XLogSegNo segno;
943 
944  /*
945  * For logical slots log a standby snapshot and start logical decoding
946  * at exactly that position. That allows the slot to start up more
947  * quickly.
948  *
949  * That's not needed (or indeed helpful) for physical slots as they'll
950  * start replay at the last logged checkpoint anyway. Instead return
951  * the location of the last redo LSN. While that slightly increases
952  * the chance that we have to retry, it's where a base backup has to
953  * start replay at.
954  */
955  if (!RecoveryInProgress() && SlotIsLogical(slot))
956  {
957  XLogRecPtr flushptr;
958 
959  /* start at current insert position */
961 
962  /* make sure we have enough information to start */
963  flushptr = LogStandbySnapshot();
964 
965  /* and make sure it's fsynced to disk */
966  XLogFlush(flushptr);
967  }
968  else
969  {
970  slot->data.restart_lsn = GetRedoRecPtr();
971  }
972 
973  /* prevent WAL removal as fast as possible */
975 
976  /*
977  * If all required WAL is still there, great, otherwise retry. The
978  * slot should prevent further removal of WAL, unless there's a
979  * concurrent ReplicationSlotsComputeRequiredLSN() after we've written
980  * the new restart_lsn above, so normally we should never need to loop
981  * more than twice.
982  */
983  XLByteToSeg(slot->data.restart_lsn, segno);
984  if (XLogGetLastRemovedSegno() < segno)
985  break;
986  }
987 }
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
ReplicationSlotPersistentData data
Definition: slot.h:115
bool RecoveryInProgress(void)
Definition: xlog.c:7873
void XLogFlush(XLogRecPtr record)
Definition: xlog.c:2758
XLogSegNo XLogGetLastRemovedSegno(void)
Definition: xlog.c:3794
void ReplicationSlotsComputeRequiredLSN(void)
Definition: slot.c:682
XLogRecPtr LogStandbySnapshot(void)
Definition: standby.c:909
XLogRecPtr GetXLogInsertRecPtr(void)
Definition: xlog.c:11103
uint64 XLogSegNo
Definition: xlogdefs.h:34
#define SlotIsLogical(slot)
Definition: slot.h:134
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
#define XLByteToSeg(xlrp, logSegNo)
#define NULL
Definition: c.h:229
uint64 XLogRecPtr
Definition: xlogdefs.h:21
#define Assert(condition)
Definition: c.h:675
XLogRecPtr restart_lsn
Definition: slot.h:68
XLogRecPtr GetRedoRecPtr(void)
Definition: xlog.c:8163
void ReplicationSlotSave ( void  )

Definition at line 579 of file slot.c.

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

Referenced by CreateInitDecodingContext(), CreateReplicationSlot(), LogicalConfirmReceivedLocation(), pg_create_physical_replication_slot(), and ReplicationSlotPersist().

580 {
581  char path[MAXPGPATH];
582 
584 
585  sprintf(path, "pg_replslot/%s", NameStr(MyReplicationSlot->data.name));
587 }
ReplicationSlotPersistentData data
Definition: slot.h:115
static void SaveSlotToPath(ReplicationSlot *slot, const char *path, int elevel)
Definition: slot.c:1151
#define ERROR
Definition: elog.h:43
#define MAXPGPATH
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
#define NULL
Definition: c.h:229
#define Assert(condition)
Definition: c.h:675
#define NameStr(name)
Definition: c.h:499
XLogRecPtr ReplicationSlotsComputeLogicalRestartLSN ( void  )

Definition at line 725 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, result, SlotIsLogical, SpinLockAcquire, and SpinLockRelease.

Referenced by CheckPointLogicalRewriteHeap(), and CheckPointSnapBuild().

726 {
728  int i;
729 
730  if (max_replication_slots <= 0)
731  return InvalidXLogRecPtr;
732 
733  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
734 
735  for (i = 0; i < max_replication_slots; i++)
736  {
737  ReplicationSlot *s;
738  XLogRecPtr restart_lsn;
739 
741 
742  /* cannot change while ReplicationSlotCtlLock is held */
743  if (!s->in_use)
744  continue;
745 
746  /* we're only interested in logical slots */
747  if (!SlotIsLogical(s))
748  continue;
749 
750  /* read once, it's ok if it increases while we're checking */
751  SpinLockAcquire(&s->mutex);
752  restart_lsn = s->data.restart_lsn;
753  SpinLockRelease(&s->mutex);
754 
755  if (result == InvalidXLogRecPtr ||
756  restart_lsn < result)
757  result = restart_lsn;
758  }
759 
760  LWLockRelease(ReplicationSlotControlLock);
761 
762  return result;
763 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
return result
Definition: formatting.c:1632
ReplicationSlotPersistentData data
Definition: slot.h:115
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1715
#define SpinLockAcquire(lock)
Definition: spin.h:62
#define SlotIsLogical(slot)
Definition: slot.h:134
bool in_use
Definition: slot.h:91
#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:68
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1111
int i
ReplicationSlot replication_slots[1]
Definition: slot.h:145
slock_t mutex
Definition: slot.h:88
void ReplicationSlotsComputeRequiredLSN ( void  )

Definition at line 682 of file slot.c.

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

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

683 {
684  int i;
685  XLogRecPtr min_required = InvalidXLogRecPtr;
686 
688 
689  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
690  for (i = 0; i < max_replication_slots; i++)
691  {
693  XLogRecPtr restart_lsn;
694 
695  if (!s->in_use)
696  continue;
697 
698  SpinLockAcquire(&s->mutex);
699  restart_lsn = s->data.restart_lsn;
700  SpinLockRelease(&s->mutex);
701 
702  if (restart_lsn != InvalidXLogRecPtr &&
703  (min_required == InvalidXLogRecPtr ||
704  restart_lsn < min_required))
705  min_required = restart_lsn;
706  }
707  LWLockRelease(ReplicationSlotControlLock);
708 
709  XLogSetReplicationSlotMinimumLSN(min_required);
710 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
ReplicationSlotPersistentData data
Definition: slot.h:115
void XLogSetReplicationSlotMinimumLSN(XLogRecPtr lsn)
Definition: xlog.c:2649
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1715
#define SpinLockAcquire(lock)
Definition: spin.h:62
bool in_use
Definition: slot.h:91
#define SpinLockRelease(lock)
Definition: spin.h:64
int max_replication_slots
Definition: slot.c:99
#define NULL
Definition: c.h:229
uint64 XLogRecPtr
Definition: xlogdefs.h:21
#define Assert(condition)
Definition: c.h:675
XLogRecPtr restart_lsn
Definition: slot.h:68
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1111
int i
ReplicationSlot replication_slots[1]
Definition: slot.h:145
slock_t mutex
Definition: slot.h:88
void ReplicationSlotsComputeRequiredXmin ( bool  already_locked)

Definition at line 636 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, NULL, ProcArraySetReplicationSlotXmin(), ReplicationSlotCtlData::replication_slots, SpinLockAcquire, SpinLockRelease, TransactionIdIsValid, and TransactionIdPrecedes().

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

637 {
638  int i;
640  TransactionId agg_catalog_xmin = InvalidTransactionId;
641 
643 
644  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
645 
646  for (i = 0; i < max_replication_slots; i++)
647  {
649  TransactionId effective_xmin;
650  TransactionId effective_catalog_xmin;
651 
652  if (!s->in_use)
653  continue;
654 
655  SpinLockAcquire(&s->mutex);
656  effective_xmin = s->effective_xmin;
657  effective_catalog_xmin = s->effective_catalog_xmin;
658  SpinLockRelease(&s->mutex);
659 
660  /* check the data xmin */
661  if (TransactionIdIsValid(effective_xmin) &&
662  (!TransactionIdIsValid(agg_xmin) ||
663  TransactionIdPrecedes(effective_xmin, agg_xmin)))
664  agg_xmin = effective_xmin;
665 
666  /* check the catalog xmin */
667  if (TransactionIdIsValid(effective_catalog_xmin) &&
668  (!TransactionIdIsValid(agg_catalog_xmin) ||
669  TransactionIdPrecedes(effective_catalog_xmin, agg_catalog_xmin)))
670  agg_catalog_xmin = effective_catalog_xmin;
671  }
672 
673  LWLockRelease(ReplicationSlotControlLock);
674 
675  ProcArraySetReplicationSlotXmin(agg_xmin, agg_catalog_xmin, already_locked);
676 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
uint32 TransactionId
Definition: c.h:397
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1715
#define SpinLockAcquire(lock)
Definition: spin.h:62
TransactionId effective_xmin
Definition: slot.h:111
#define InvalidTransactionId
Definition: transam.h:31
bool TransactionIdPrecedes(TransactionId id1, TransactionId id2)
Definition: transam.c:300
bool in_use
Definition: slot.h:91
#define SpinLockRelease(lock)
Definition: spin.h:64
TransactionId effective_catalog_xmin
Definition: slot.h:112
int max_replication_slots
Definition: slot.c:99
#define NULL
Definition: c.h:229
#define Assert(condition)
Definition: c.h:675
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1111
int i
void ProcArraySetReplicationSlotXmin(TransactionId xmin, TransactionId catalog_xmin, bool already_locked)
Definition: procarray.c:2960
ReplicationSlot replication_slots[1]
Definition: slot.h:145
slock_t mutex
Definition: slot.h:88
#define TransactionIdIsValid(xid)
Definition: transam.h:41
bool ReplicationSlotsCountDBSlots ( Oid  dboid,
int *  nslots,
int *  nactive 
)

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

775 {
776  int i;
777 
778  *nslots = *nactive = 0;
779 
780  if (max_replication_slots <= 0)
781  return false;
782 
783  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
784  for (i = 0; i < max_replication_slots; i++)
785  {
786  ReplicationSlot *s;
787 
789 
790  /* cannot change while ReplicationSlotCtlLock is held */
791  if (!s->in_use)
792  continue;
793 
794  /* only logical slots are database specific, skip */
795  if (!SlotIsLogical(s))
796  continue;
797 
798  /* not our database, skip */
799  if (s->data.database != dboid)
800  continue;
801 
802  /* count slots with spinlock held */
803  SpinLockAcquire(&s->mutex);
804  (*nslots)++;
805  if (s->active_pid != 0)
806  (*nactive)++;
807  SpinLockRelease(&s->mutex);
808  }
809  LWLockRelease(ReplicationSlotControlLock);
810 
811  if (*nslots > 0)
812  return true;
813  return false;
814 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
ReplicationSlotPersistentData data
Definition: slot.h:115
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1715
#define SpinLockAcquire(lock)
Definition: spin.h:62
#define SlotIsLogical(slot)
Definition: slot.h:134
bool in_use
Definition: slot.h:91
#define SpinLockRelease(lock)
Definition: spin.h:64
int max_replication_slots
Definition: slot.c:99
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1111
pid_t active_pid
Definition: slot.h:94
int i
ReplicationSlot replication_slots[1]
Definition: slot.h:145
slock_t mutex
Definition: slot.h:88
void ReplicationSlotsDropDBSlots ( Oid  dboid)

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

831 {
832  int i;
833 
834  if (max_replication_slots <= 0)
835  return;
836 
837 restart:
838  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
839  for (i = 0; i < max_replication_slots; i++)
840  {
841  ReplicationSlot *s;
842  char *slotname;
843  int active_pid;
844 
846 
847  /* cannot change while ReplicationSlotCtlLock is held */
848  if (!s->in_use)
849  continue;
850 
851  /* only logical slots are database specific, skip */
852  if (!SlotIsLogical(s))
853  continue;
854 
855  /* not our database, skip */
856  if (s->data.database != dboid)
857  continue;
858 
859  /* acquire slot, so ReplicationSlotDropAcquired can be reused */
860  SpinLockAcquire(&s->mutex);
861  /* can't change while ReplicationSlotControlLock is held */
862  slotname = NameStr(s->data.name);
863  active_pid = s->active_pid;
864  if (active_pid == 0)
865  {
866  MyReplicationSlot = s;
867  s->active_pid = MyProcPid;
868  }
869  SpinLockRelease(&s->mutex);
870 
871  /*
872  * Even though we hold an exclusive lock on the database object a
873  * logical slot for that DB can still be active, e.g. if it's
874  * concurrently being dropped by a backend connected to another DB.
875  *
876  * That's fairly unlikely in practice, so we'll just bail out.
877  */
878  if (active_pid)
879  ereport(ERROR,
880  (errcode(ERRCODE_OBJECT_IN_USE),
881  errmsg("replication slot \"%s\" is active for PID %d",
882  slotname, active_pid)));
883 
884  /*
885  * To avoid duplicating ReplicationSlotDropAcquired() and to avoid
886  * holding ReplicationSlotControlLock over filesystem operations,
887  * release ReplicationSlotControlLock and use
888  * ReplicationSlotDropAcquired.
889  *
890  * As that means the set of slots could change, restart scan from the
891  * beginning each time we release the lock.
892  */
893  LWLockRelease(ReplicationSlotControlLock);
895  goto restart;
896  }
897  LWLockRelease(ReplicationSlotControlLock);
898 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
int MyProcPid
Definition: globals.c:38
int errcode(int sqlerrcode)
Definition: elog.c:575
ReplicationSlotPersistentData data
Definition: slot.h:115
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1715
#define SpinLockAcquire(lock)
Definition: spin.h:62
static void ReplicationSlotDropAcquired(void)
Definition: slot.c:469
#define ERROR
Definition: elog.h:43
#define SlotIsLogical(slot)
Definition: slot.h:134
#define ereport(elevel, rest)
Definition: elog.h:122
bool in_use
Definition: slot.h:91
#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:1111
int errmsg(const char *fmt,...)
Definition: elog.c:797
pid_t active_pid
Definition: slot.h:94
int i
#define NameStr(name)
Definition: c.h:499
ReplicationSlot replication_slots[1]
Definition: slot.h:145
slock_t mutex
Definition: slot.h:88
void ReplicationSlotsShmemInit ( void  )

Definition at line 132 of file slot.c.

References 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);
160  }
161  }
162 }
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:857
void * ShmemInitStruct(const char *name, Size size, bool *foundPtr)
Definition: shmem.c:372
void LWLockInitialize(LWLock *lock, int tranche_id)
Definition: lwlock.c:667
int max_replication_slots
Definition: slot.c:99
void LWLockRegisterTranche(int tranche_id, char *tranche_name)
Definition: lwlock.c:592
int i
ReplicationSlot replication_slots[1]
Definition: slot.h:145
slock_t mutex
Definition: slot.h:88
LWLock io_in_progress_lock
Definition: slot.h:118
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:356
#define offsetof(type, field)
Definition: c.h:555
bool ReplicationSlotValidateName ( const char *  name,
int  elevel 
)

Definition at line 173 of file slot.c.

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

Referenced by readRecoveryCommandFile(), ReplicationSlotAcquire(), ReplicationSlotCreate(), and StartupReorderBuffer().

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

Definition at line 1031 of file slot.c.

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

Referenced by StartupXLOG().

1032 {
1033  DIR *replication_dir;
1034  struct dirent *replication_de;
1035 
1036  elog(DEBUG1, "starting up replication slots");
1037 
1038  /* restore all slots by iterating over all on-disk entries */
1039  replication_dir = AllocateDir("pg_replslot");
1040  while ((replication_de = ReadDir(replication_dir, "pg_replslot")) != NULL)
1041  {
1042  struct stat statbuf;
1043  char path[MAXPGPATH + 12];
1044 
1045  if (strcmp(replication_de->d_name, ".") == 0 ||
1046  strcmp(replication_de->d_name, "..") == 0)
1047  continue;
1048 
1049  snprintf(path, sizeof(path), "pg_replslot/%s", replication_de->d_name);
1050 
1051  /* we're only creating directories here, skip if it's not our's */
1052  if (lstat(path, &statbuf) == 0 && !S_ISDIR(statbuf.st_mode))
1053  continue;
1054 
1055  /* we crashed while a slot was being setup or deleted, clean up */
1056  if (pg_str_endswith(replication_de->d_name, ".tmp"))
1057  {
1058  if (!rmtree(path, true))
1059  {
1060  ereport(WARNING,
1062  errmsg("could not remove directory \"%s\"", path)));
1063  continue;
1064  }
1065  fsync_fname("pg_replslot", true);
1066  continue;
1067  }
1068 
1069  /* looks like a slot in a normal state, restore */
1070  RestoreSlotFromDisk(replication_de->d_name);
1071  }
1072  FreeDir(replication_dir);
1073 
1074  /* currently no slots exist, we're done. */
1075  if (max_replication_slots <= 0)
1076  return;
1077 
1078  /* Now that we have recovered all the data, compute replication xmin */
1081 }
static void RestoreSlotFromDisk(const char *name)
Definition: slot.c:1275
#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:567
int snprintf(char *str, size_t count, const char *fmt,...) pg_attribute_printf(3
Definition: dirent.h:9
void ReplicationSlotsComputeRequiredLSN(void)
Definition: slot.c:682
Definition: dirent.c:25
#define MAXPGPATH
int errcode_for_file_access(void)
Definition: elog.c:598
DIR * AllocateDir(const char *dirname)
Definition: fd.c:2335
#define ereport(elevel, rest)
Definition: elog.h:122
#define WARNING
Definition: elog.h:40
bool rmtree(const char *path, bool rmtopdir)
Definition: rmtree.c:36
int max_replication_slots
Definition: slot.c:99
#define NULL
Definition: c.h:229
struct dirent * ReadDir(DIR *dir, const char *dirname)
Definition: fd.c:2401
int errmsg(const char *fmt,...)
Definition: elog.c:797
char d_name[MAX_PATH]
Definition: dirent.h:14
#define elog
Definition: elog.h:219
#define lstat(path, sb)
Definition: win32.h:262
void ReplicationSlotsComputeRequiredXmin(bool already_locked)
Definition: slot.c:636
int FreeDir(DIR *dir)
Definition: fd.c:2444

Variable Documentation

ReplicationSlotCtlData* ReplicationSlotCtl

Definition at line 93 of file slot.c.

Referenced by pg_get_replication_slots().