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 StartupReplicationSlots (void)
 
void CheckPointReplicationSlots (void)
 
void CheckSlotRequirements (void)
 

Variables

ReplicationSlotCtlDataReplicationSlotCtl
 
PGDLLIMPORT ReplicationSlotMyReplicationSlot
 
PGDLLIMPORT int max_replication_slots
 

Macro Definition Documentation

#define SlotIsLogical (   slot)    (slot->data.database != InvalidOid)
#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 894 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().

895 {
896  int i;
897 
898  elog(DEBUG1, "performing replication slot checkpoint");
899 
900  /*
901  * Prevent any slot from being created/dropped while we're active. As we
902  * explicitly do *not* want to block iterating over replication_slots or
903  * acquiring a slot we cannot take the control lock - but that's OK,
904  * because holding ReplicationSlotAllocationLock is strictly stronger, and
905  * enough to guarantee that nobody can change the in_use bits on us.
906  */
907  LWLockAcquire(ReplicationSlotAllocationLock, LW_SHARED);
908 
909  for (i = 0; i < max_replication_slots; i++)
910  {
912  char path[MAXPGPATH];
913 
914  if (!s->in_use)
915  continue;
916 
917  /* save the slot to disk, locking is handled in SaveSlotToPath() */
918  sprintf(path, "pg_replslot/%s", NameStr(s->data.name));
919  SaveSlotToPath(s, path, LOG);
920  }
921  LWLockRelease(ReplicationSlotAllocationLock);
922 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:92
#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:1049
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1714
#define MAXPGPATH
bool in_use
Definition: slot.h:91
int max_replication_slots
Definition: slot.c:98
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1110
int i
#define NameStr(name)
Definition: c.h:495
ReplicationSlot replication_slots[1]
Definition: slot.h:145
#define elog
Definition: elog.h:219
void CheckSlotRequirements ( void  )

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

805 {
806  if (max_replication_slots == 0)
807  ereport(ERROR,
808  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
809  (errmsg("replication slots can only be used if max_replication_slots > 0"))));
810 
812  ereport(ERROR,
813  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
814  errmsg("replication slots can only be used if wal_level >= replica")));
815 }
int wal_level
Definition: xlog.c:103
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:98
int errmsg(const char *fmt,...)
Definition: elog.c:797
void ReplicationSlotAcquire ( const char *  name)

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

326 {
327  ReplicationSlot *slot = NULL;
328  int i;
329  int active_pid = 0; /* Keep compiler quiet */
330 
332 
334 
335  /* Search for the named slot and mark it active if we find it. */
336  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
337  for (i = 0; i < max_replication_slots; i++)
338  {
340 
341  if (s->in_use && strcmp(name, NameStr(s->data.name)) == 0)
342  {
343  SpinLockAcquire(&s->mutex);
344  active_pid = s->active_pid;
345  if (active_pid == 0)
346  active_pid = s->active_pid = MyProcPid;
347  SpinLockRelease(&s->mutex);
348  slot = s;
349  break;
350  }
351  }
352  LWLockRelease(ReplicationSlotControlLock);
353 
354  /* If we did not find the slot or it was already active, error out. */
355  if (slot == NULL)
356  ereport(ERROR,
357  (errcode(ERRCODE_UNDEFINED_OBJECT),
358  errmsg("replication slot \"%s\" does not exist", name)));
359  if (active_pid != MyProcPid)
360  ereport(ERROR,
361  (errcode(ERRCODE_OBJECT_IN_USE),
362  errmsg("replication slot \"%s\" is active for PID %d",
363  name, active_pid)));
364 
365  /* We made this slot active, so it's ours now. */
366  MyReplicationSlot = slot;
367 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:92
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:1714
bool ReplicationSlotValidateName(const char *name, int elevel)
Definition: slot.c:172
#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:95
int max_replication_slots
Definition: slot.c:98
#define NULL
Definition: c.h:226
#define Assert(condition)
Definition: c.h:671
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1110
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:495
ReplicationSlot replication_slots[1]
Definition: slot.h:145
slock_t mutex
Definition: slot.h:88
void ReplicationSlotCleanup ( void  )

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

413 {
414  int i;
415 
417 
418  /*
419  * No need for locking as we are only interested in slots active in
420  * current process and those are not touched by other processes.
421  */
422  for (i = 0; i < max_replication_slots; i++)
423  {
425 
426  if (s->active_pid == MyProcPid)
427  {
429 
431  }
432  }
433 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:92
int MyProcPid
Definition: globals.c:38
ReplicationSlotPersistency persistency
Definition: slot.h:49
static void ReplicationSlotDropPtr(ReplicationSlot *slot)
Definition: slot.c:469
ReplicationSlotPersistentData data
Definition: slot.h:115
bool in_use
Definition: slot.h:91
ReplicationSlot * MyReplicationSlot
Definition: slot.c:95
int max_replication_slots
Definition: slot.c:98
#define NULL
Definition: c.h:226
#define Assert(condition)
Definition: c.h:671
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 219 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().

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

Definition at line 439 of file slot.c.

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

Referenced by DropReplicationSlot(), and pg_drop_replication_slot().

440 {
442 
444 
446 }
static void ReplicationSlotDropAcquired(void)
Definition: slot.c:452
ReplicationSlot * MyReplicationSlot
Definition: slot.c:95
#define NULL
Definition: c.h:226
#define Assert(condition)
Definition: c.h:671
void ReplicationSlotAcquire(const char *name)
Definition: slot.c:325
const char * name
Definition: encode.c:521
void ReplicationSlotMarkDirty ( void  )

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

581 {
583 
585 
586  SpinLockAcquire(&slot->mutex);
588  MyReplicationSlot->dirty = true;
589  SpinLockRelease(&slot->mutex);
590 }
#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:95
#define NULL
Definition: c.h:226
#define Assert(condition)
Definition: c.h:671
slock_t mutex
Definition: slot.h:88
bool dirty
Definition: slot.h:98
void ReplicationSlotPersist ( void  )

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

598 {
600 
601  Assert(slot != NULL);
603 
604  SpinLockAcquire(&slot->mutex);
606  SpinLockRelease(&slot->mutex);
607 
610 }
ReplicationSlotPersistency persistency
Definition: slot.h:49
void ReplicationSlotSave(void)
Definition: slot.c:562
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:95
#define NULL
Definition: c.h:226
#define Assert(condition)
Definition: c.h:671
slock_t mutex
Definition: slot.h:88
void ReplicationSlotMarkDirty(void)
Definition: slot.c:580
void ReplicationSlotRelease ( void  )

Definition at line 374 of file slot.c.

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

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

375 {
377 
378  Assert(slot != NULL && slot->active_pid != 0);
379 
380  if (slot->data.persistency == RS_EPHEMERAL)
381  {
382  /*
383  * Delete the slot. There is no !PANIC case where this is allowed to
384  * fail, all that may happen is an incomplete cleanup of the on-disk
385  * data.
386  */
388  }
389  else if (slot->data.persistency == RS_PERSISTENT)
390  {
391  /*
392  * Mark persistent slot inactive. We're not freeing it, just
393  * disconnecting.
394  */
395  SpinLockAcquire(&slot->mutex);
396  slot->active_pid = 0;
397  SpinLockRelease(&slot->mutex);
398  }
399 
401 
402  /* might not have been set when we've been a plain slot */
403  LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
405  LWLockRelease(ProcArrayLock);
406 }
#define PROC_IN_LOGICAL_DECODING
Definition: proc.h:47
ReplicationSlotPersistency persistency
Definition: slot.h:49
ReplicationSlotPersistentData data
Definition: slot.h:115
PGXACT * MyPgXact
Definition: proc.c:68
uint8 vacuumFlags
Definition: proc.h:208
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1714
#define SpinLockAcquire(lock)
Definition: spin.h:62
static void ReplicationSlotDropAcquired(void)
Definition: slot.c:452
#define SpinLockRelease(lock)
Definition: spin.h:64
ReplicationSlot * MyReplicationSlot
Definition: slot.c:95
#define NULL
Definition: c.h:226
#define Assert(condition)
Definition: c.h:671
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1110
pid_t active_pid
Definition: slot.h:94
slock_t mutex
Definition: slot.h:88
void ReplicationSlotReserveWal ( void  )

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

825 {
827 
828  Assert(slot != NULL);
830 
831  /*
832  * The replication slot mechanism is used to prevent removal of required
833  * WAL. As there is no interlock between this routine and checkpoints, WAL
834  * segments could concurrently be removed when a now stale return value of
835  * ReplicationSlotsComputeRequiredLSN() is used. In the unlikely case that
836  * this happens we'll just retry.
837  */
838  while (true)
839  {
840  XLogSegNo segno;
841 
842  /*
843  * For logical slots log a standby snapshot and start logical decoding
844  * at exactly that position. That allows the slot to start up more
845  * quickly.
846  *
847  * That's not needed (or indeed helpful) for physical slots as they'll
848  * start replay at the last logged checkpoint anyway. Instead return
849  * the location of the last redo LSN. While that slightly increases
850  * the chance that we have to retry, it's where a base backup has to
851  * start replay at.
852  */
853  if (!RecoveryInProgress() && SlotIsLogical(slot))
854  {
855  XLogRecPtr flushptr;
856 
857  /* start at current insert position */
859 
860  /* make sure we have enough information to start */
861  flushptr = LogStandbySnapshot();
862 
863  /* and make sure it's fsynced to disk */
864  XLogFlush(flushptr);
865  }
866  else
867  {
868  slot->data.restart_lsn = GetRedoRecPtr();
869  }
870 
871  /* prevent WAL removal as fast as possible */
873 
874  /*
875  * If all required WAL is still there, great, otherwise retry. The
876  * slot should prevent further removal of WAL, unless there's a
877  * concurrent ReplicationSlotsComputeRequiredLSN() after we've written
878  * the new restart_lsn above, so normally we should never need to loop
879  * more than twice.
880  */
881  XLByteToSeg(slot->data.restart_lsn, segno);
882  if (XLogGetLastRemovedSegno() < segno)
883  break;
884  }
885 }
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
ReplicationSlotPersistentData data
Definition: slot.h:115
bool RecoveryInProgress(void)
Definition: xlog.c:7805
void XLogFlush(XLogRecPtr record)
Definition: xlog.c:2745
XLogSegNo XLogGetLastRemovedSegno(void)
Definition: xlog.c:3771
void ReplicationSlotsComputeRequiredLSN(void)
Definition: slot.c:664
XLogRecPtr LogStandbySnapshot(void)
Definition: standby.c:913
XLogRecPtr GetXLogInsertRecPtr(void)
Definition: xlog.c:11003
uint64 XLogSegNo
Definition: xlogdefs.h:34
#define SlotIsLogical(slot)
Definition: slot.h:134
ReplicationSlot * MyReplicationSlot
Definition: slot.c:95
#define XLByteToSeg(xlrp, logSegNo)
#define NULL
Definition: c.h:226
uint64 XLogRecPtr
Definition: xlogdefs.h:21
#define Assert(condition)
Definition: c.h:671
XLogRecPtr restart_lsn
Definition: slot.h:68
XLogRecPtr GetRedoRecPtr(void)
Definition: xlog.c:8095
void ReplicationSlotSave ( void  )

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

563 {
564  char path[MAXPGPATH];
565 
567 
568  sprintf(path, "pg_replslot/%s", NameStr(MyReplicationSlot->data.name));
570 }
ReplicationSlotPersistentData data
Definition: slot.h:115
static void SaveSlotToPath(ReplicationSlot *slot, const char *path, int elevel)
Definition: slot.c:1049
#define ERROR
Definition: elog.h:43
#define MAXPGPATH
ReplicationSlot * MyReplicationSlot
Definition: slot.c:95
#define NULL
Definition: c.h:226
#define Assert(condition)
Definition: c.h:671
#define NameStr(name)
Definition: c.h:495
XLogRecPtr ReplicationSlotsComputeLogicalRestartLSN ( void  )

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

708 {
709  XLogRecPtr result = InvalidXLogRecPtr;
710  int i;
711 
712  if (max_replication_slots <= 0)
713  return InvalidXLogRecPtr;
714 
715  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
716 
717  for (i = 0; i < max_replication_slots; i++)
718  {
719  ReplicationSlot *s;
720  XLogRecPtr restart_lsn;
721 
723 
724  /* cannot change while ReplicationSlotCtlLock is held */
725  if (!s->in_use)
726  continue;
727 
728  /* we're only interested in logical slots */
729  if (!SlotIsLogical(s))
730  continue;
731 
732  /* read once, it's ok if it increases while we're checking */
733  SpinLockAcquire(&s->mutex);
734  restart_lsn = s->data.restart_lsn;
735  SpinLockRelease(&s->mutex);
736 
737  if (result == InvalidXLogRecPtr ||
738  restart_lsn < result)
739  result = restart_lsn;
740  }
741 
742  LWLockRelease(ReplicationSlotControlLock);
743 
744  return result;
745 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:92
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
ReplicationSlotPersistentData data
Definition: slot.h:115
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1714
#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:98
uint64 XLogRecPtr
Definition: xlogdefs.h:21
XLogRecPtr restart_lsn
Definition: slot.h:68
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1110
int i
ReplicationSlot replication_slots[1]
Definition: slot.h:145
slock_t mutex
Definition: slot.h:88
void ReplicationSlotsComputeRequiredLSN ( void  )

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

665 {
666  int i;
667  XLogRecPtr min_required = InvalidXLogRecPtr;
668 
670 
671  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
672  for (i = 0; i < max_replication_slots; i++)
673  {
675  XLogRecPtr restart_lsn;
676 
677  if (!s->in_use)
678  continue;
679 
680  SpinLockAcquire(&s->mutex);
681  restart_lsn = s->data.restart_lsn;
682  SpinLockRelease(&s->mutex);
683 
684  if (restart_lsn != InvalidXLogRecPtr &&
685  (min_required == InvalidXLogRecPtr ||
686  restart_lsn < min_required))
687  min_required = restart_lsn;
688  }
689  LWLockRelease(ReplicationSlotControlLock);
690 
691  XLogSetReplicationSlotMinimumLSN(min_required);
692 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:92
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
ReplicationSlotPersistentData data
Definition: slot.h:115
void XLogSetReplicationSlotMinimumLSN(XLogRecPtr lsn)
Definition: xlog.c:2636
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1714
#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:98
#define NULL
Definition: c.h:226
uint64 XLogRecPtr
Definition: xlogdefs.h:21
#define Assert(condition)
Definition: c.h:671
XLogRecPtr restart_lsn
Definition: slot.h:68
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1110
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 616 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(), and StartupReplicationSlots().

617 {
618  int i;
620  TransactionId agg_catalog_xmin = InvalidTransactionId;
621 
623 
624  if (!already_locked)
625  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
626 
627  for (i = 0; i < max_replication_slots; i++)
628  {
630  TransactionId effective_xmin;
631  TransactionId effective_catalog_xmin;
632 
633  if (!s->in_use)
634  continue;
635 
636  SpinLockAcquire(&s->mutex);
637  effective_xmin = s->effective_xmin;
638  effective_catalog_xmin = s->effective_catalog_xmin;
639  SpinLockRelease(&s->mutex);
640 
641  /* check the data xmin */
642  if (TransactionIdIsValid(effective_xmin) &&
643  (!TransactionIdIsValid(agg_xmin) ||
644  TransactionIdPrecedes(effective_xmin, agg_xmin)))
645  agg_xmin = effective_xmin;
646 
647  /* check the catalog xmin */
648  if (TransactionIdIsValid(effective_catalog_xmin) &&
649  (!TransactionIdIsValid(agg_catalog_xmin) ||
650  TransactionIdPrecedes(effective_catalog_xmin, agg_catalog_xmin)))
651  agg_catalog_xmin = effective_catalog_xmin;
652  }
653 
654  if (!already_locked)
655  LWLockRelease(ReplicationSlotControlLock);
656 
657  ProcArraySetReplicationSlotXmin(agg_xmin, agg_catalog_xmin, already_locked);
658 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:92
uint32 TransactionId
Definition: c.h:394
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1714
#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:98
#define NULL
Definition: c.h:226
#define Assert(condition)
Definition: c.h:671
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1110
int i
void ProcArraySetReplicationSlotXmin(TransactionId xmin, TransactionId catalog_xmin, bool already_locked)
Definition: procarray.c:2947
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 756 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().

757 {
758  int i;
759 
760  *nslots = *nactive = 0;
761 
762  if (max_replication_slots <= 0)
763  return false;
764 
765  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
766  for (i = 0; i < max_replication_slots; i++)
767  {
768  ReplicationSlot *s;
769 
771 
772  /* cannot change while ReplicationSlotCtlLock is held */
773  if (!s->in_use)
774  continue;
775 
776  /* only logical slots are database specific, skip */
777  if (!SlotIsLogical(s))
778  continue;
779 
780  /* not our database, skip */
781  if (s->data.database != dboid)
782  continue;
783 
784  /* count slots with spinlock held */
785  SpinLockAcquire(&s->mutex);
786  (*nslots)++;
787  if (s->active_pid != 0)
788  (*nactive)++;
789  SpinLockRelease(&s->mutex);
790  }
791  LWLockRelease(ReplicationSlotControlLock);
792 
793  if (*nslots > 0)
794  return true;
795  return false;
796 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:92
ReplicationSlotPersistentData data
Definition: slot.h:115
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1714
#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:98
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1110
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 ReplicationSlotsShmemInit ( void  )

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

132 {
133  bool found;
134 
135  if (max_replication_slots == 0)
136  return;
137 
139  ShmemInitStruct("ReplicationSlot Ctl", ReplicationSlotsShmemSize(),
140  &found);
141 
143  "replication_slot_io");
144 
145  if (!found)
146  {
147  int i;
148 
149  /* First time through, so initialize */
151 
152  for (i = 0; i < max_replication_slots; i++)
153  {
155 
156  /* everything else is zeroed by the memset above */
157  SpinLockInit(&slot->mutex);
159  }
160  }
161 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:92
Size ReplicationSlotsShmemSize(void)
Definition: slot.c:113
#define SpinLockInit(lock)
Definition: spin.h:60
#define MemSet(start, val, len)
Definition: c.h:853
void * ShmemInitStruct(const char *name, Size size, bool *foundPtr)
Definition: shmem.c:372
void LWLockInitialize(LWLock *lock, int tranche_id)
Definition: lwlock.c:666
int max_replication_slots
Definition: slot.c:98
void LWLockRegisterTranche(int tranche_id, char *tranche_name)
Definition: lwlock.c:591
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 113 of file slot.c.

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

Referenced by CreateSharedMemoryAndSemaphores(), and ReplicationSlotsShmemInit().

114 {
115  Size size = 0;
116 
117  if (max_replication_slots == 0)
118  return size;
119 
120  size = offsetof(ReplicationSlotCtlData, replication_slots);
121  size = add_size(size,
123 
124  return size;
125 }
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:98
size_t Size
Definition: c.h:353
#define offsetof(type, field)
Definition: c.h:551
bool ReplicationSlotValidateName ( const char *  name,
int  elevel 
)

Definition at line 172 of file slot.c.

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

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

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

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

930 {
931  DIR *replication_dir;
932  struct dirent *replication_de;
933 
934  elog(DEBUG1, "starting up replication slots");
935 
936  /* restore all slots by iterating over all on-disk entries */
937  replication_dir = AllocateDir("pg_replslot");
938  while ((replication_de = ReadDir(replication_dir, "pg_replslot")) != NULL)
939  {
940  struct stat statbuf;
941  char path[MAXPGPATH];
942 
943  if (strcmp(replication_de->d_name, ".") == 0 ||
944  strcmp(replication_de->d_name, "..") == 0)
945  continue;
946 
947  snprintf(path, MAXPGPATH, "pg_replslot/%s", replication_de->d_name);
948 
949  /* we're only creating directories here, skip if it's not our's */
950  if (lstat(path, &statbuf) == 0 && !S_ISDIR(statbuf.st_mode))
951  continue;
952 
953  /* we crashed while a slot was being setup or deleted, clean up */
954  if (pg_str_endswith(replication_de->d_name, ".tmp"))
955  {
956  if (!rmtree(path, true))
957  {
960  errmsg("could not remove directory \"%s\"", path)));
961  continue;
962  }
963  fsync_fname("pg_replslot", true);
964  continue;
965  }
966 
967  /* looks like a slot in a normal state, restore */
968  RestoreSlotFromDisk(replication_de->d_name);
969  }
970  FreeDir(replication_dir);
971 
972  /* currently no slots exist, we're done. */
973  if (max_replication_slots <= 0)
974  return;
975 
976  /* Now that we have recovered all the data, compute replication xmin */
979 }
static void RestoreSlotFromDisk(const char *name)
Definition: slot.c:1167
#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:664
Definition: dirent.c:25
#define MAXPGPATH
int errcode_for_file_access(void)
Definition: elog.c:598
DIR * AllocateDir(const char *dirname)
Definition: fd.c:2284
#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:98
#define NULL
Definition: c.h:226
struct dirent * ReadDir(DIR *dir, const char *dirname)
Definition: fd.c:2350
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:272
void ReplicationSlotsComputeRequiredXmin(bool already_locked)
Definition: slot.c:616
int FreeDir(DIR *dir)
Definition: fd.c:2393

Variable Documentation

ReplicationSlotCtlData* ReplicationSlotCtl

Definition at line 92 of file slot.c.

Referenced by pg_get_replication_slots().