PostgreSQL Source Code  git master
slot.c File Reference
#include "postgres.h"
#include <unistd.h>
#include <sys/stat.h>
#include "access/transam.h"
#include "access/xlog_internal.h"
#include "access/xlogrecovery.h"
#include "common/file_utils.h"
#include "common/string.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "postmaster/interrupt.h"
#include "replication/slotsync.h"
#include "replication/slot.h"
#include "replication/walsender_private.h"
#include "storage/fd.h"
#include "storage/ipc.h"
#include "storage/proc.h"
#include "storage/procarray.h"
#include "utils/builtins.h"
#include "utils/guc_hooks.h"
#include "utils/varlena.h"
Include dependency graph for slot.c:

Go to the source code of this file.

Data Structures

struct  ReplicationSlotOnDisk
 
struct  StandbySlotNamesConfigData
 

Macros

#define RS_INVAL_MAX_CAUSES   RS_INVAL_WAL_LEVEL
 
#define ReplicationSlotOnDiskConstantSize    offsetof(ReplicationSlotOnDisk, slotdata)
 
#define ReplicationSlotOnDiskNotChecksummedSize    offsetof(ReplicationSlotOnDisk, version)
 
#define ReplicationSlotOnDiskChecksummedSize    sizeof(ReplicationSlotOnDisk) - ReplicationSlotOnDiskNotChecksummedSize
 
#define ReplicationSlotOnDiskV2Size    sizeof(ReplicationSlotOnDisk) - ReplicationSlotOnDiskConstantSize
 
#define SLOT_MAGIC   0x1051CA1 /* format identifier */
 
#define SLOT_VERSION   5 /* version for new files */
 

Typedefs

typedef struct ReplicationSlotOnDisk ReplicationSlotOnDisk
 

Functions

 StaticAssertDecl (lengthof(SlotInvalidationCauses)==(RS_INVAL_MAX_CAUSES+1), "array length mismatch")
 
static void ReplicationSlotShmemExit (int code, Datum arg)
 
static void ReplicationSlotDropPtr (ReplicationSlot *slot)
 
static void RestoreSlotFromDisk (const char *name)
 
static void CreateSlotOnDisk (ReplicationSlot *slot)
 
static void SaveSlotToPath (ReplicationSlot *slot, const char *dir, int elevel)
 
Size ReplicationSlotsShmemSize (void)
 
void ReplicationSlotsShmemInit (void)
 
void ReplicationSlotInitialize (void)
 
bool ReplicationSlotValidateName (const char *name, int elevel)
 
void ReplicationSlotCreate (const char *name, bool db_specific, ReplicationSlotPersistency persistency, bool two_phase, bool failover, bool synced)
 
ReplicationSlotSearchNamedReplicationSlot (const char *name, bool need_lock)
 
int ReplicationSlotIndex (ReplicationSlot *slot)
 
bool ReplicationSlotName (int index, Name name)
 
void ReplicationSlotAcquire (const char *name, bool nowait)
 
void ReplicationSlotRelease (void)
 
void ReplicationSlotCleanup (void)
 
void ReplicationSlotDrop (const char *name, bool nowait)
 
void ReplicationSlotAlter (const char *name, bool failover)
 
void ReplicationSlotDropAcquired (void)
 
void ReplicationSlotSave (void)
 
void ReplicationSlotMarkDirty (void)
 
void ReplicationSlotPersist (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 CheckSlotRequirements (void)
 
void CheckSlotPermissions (void)
 
void ReplicationSlotReserveWal (void)
 
static void ReportSlotInvalidation (ReplicationSlotInvalidationCause cause, bool terminating, int pid, NameData slotname, XLogRecPtr restart_lsn, XLogRecPtr oldestLSN, TransactionId snapshotConflictHorizon)
 
static bool InvalidatePossiblyObsoleteSlot (ReplicationSlotInvalidationCause cause, ReplicationSlot *s, XLogRecPtr oldestLSN, Oid dboid, TransactionId snapshotConflictHorizon, bool *invalidated)
 
bool InvalidateObsoleteReplicationSlots (ReplicationSlotInvalidationCause cause, XLogSegNo oldestSegno, Oid dboid, TransactionId snapshotConflictHorizon)
 
void CheckPointReplicationSlots (bool is_shutdown)
 
void StartupReplicationSlots (void)
 
ReplicationSlotInvalidationCause GetSlotInvalidationCause (const char *invalidation_reason)
 
static bool validate_standby_slots (char *rawname, List **elemlist)
 
bool check_standby_slot_names (char **newval, void **extra, GucSource source)
 
void assign_standby_slot_names (const char *newval, void *extra)
 
bool SlotExistsInStandbySlotNames (const char *slot_name)
 
bool StandbySlotsHaveCaughtup (XLogRecPtr wait_for_lsn, int elevel)
 
void WaitForStandbyConfirmation (XLogRecPtr wait_for_lsn)
 

Variables

const char *const SlotInvalidationCauses []
 
ReplicationSlotCtlDataReplicationSlotCtl = NULL
 
ReplicationSlotMyReplicationSlot = NULL
 
int max_replication_slots = 10
 
char * standby_slot_names
 
static StandbySlotNamesConfigDatastandby_slot_names_config
 
static XLogRecPtr ss_oldest_flush_lsn = InvalidXLogRecPtr
 

Macro Definition Documentation

◆ ReplicationSlotOnDiskChecksummedSize

#define ReplicationSlotOnDiskChecksummedSize    sizeof(ReplicationSlotOnDisk) - ReplicationSlotOnDiskNotChecksummedSize

Definition at line 125 of file slot.c.

◆ ReplicationSlotOnDiskConstantSize

#define ReplicationSlotOnDiskConstantSize    offsetof(ReplicationSlotOnDisk, slotdata)

Definition at line 119 of file slot.c.

◆ ReplicationSlotOnDiskNotChecksummedSize

#define ReplicationSlotOnDiskNotChecksummedSize    offsetof(ReplicationSlotOnDisk, version)

Definition at line 122 of file slot.c.

◆ ReplicationSlotOnDiskV2Size

#define ReplicationSlotOnDiskV2Size    sizeof(ReplicationSlotOnDisk) - ReplicationSlotOnDiskConstantSize

Definition at line 128 of file slot.c.

◆ RS_INVAL_MAX_CAUSES

#define RS_INVAL_MAX_CAUSES   RS_INVAL_WAL_LEVEL

Definition at line 113 of file slot.c.

◆ SLOT_MAGIC

#define SLOT_MAGIC   0x1051CA1 /* format identifier */

Definition at line 131 of file slot.c.

◆ SLOT_VERSION

#define SLOT_VERSION   5 /* version for new files */

Definition at line 132 of file slot.c.

Typedef Documentation

◆ ReplicationSlotOnDisk

Function Documentation

◆ assign_standby_slot_names()

void assign_standby_slot_names ( const char *  newval,
void *  extra 
)

Definition at line 2538 of file slot.c.

2539 {
2540  /*
2541  * The standby slots may have changed, so we must recompute the oldest
2542  * LSN.
2543  */
2545 
2547 }
static XLogRecPtr ss_oldest_flush_lsn
Definition: slot.c:157
static StandbySlotNamesConfigData * standby_slot_names_config
Definition: slot.c:151
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28

References InvalidXLogRecPtr, ss_oldest_flush_lsn, and standby_slot_names_config.

◆ check_standby_slot_names()

bool check_standby_slot_names ( char **  newval,
void **  extra,
GucSource  source 
)

Definition at line 2484 of file slot.c.

2485 {
2486  char *rawname;
2487  char *ptr;
2488  List *elemlist;
2489  int size;
2490  bool ok;
2492 
2493  if ((*newval)[0] == '\0')
2494  return true;
2495 
2496  /* Need a modifiable copy of the GUC string */
2497  rawname = pstrdup(*newval);
2498 
2499  /* Now verify if the specified slots exist and have correct type */
2500  ok = validate_standby_slots(rawname, &elemlist);
2501 
2502  if (!ok || elemlist == NIL)
2503  {
2504  pfree(rawname);
2505  list_free(elemlist);
2506  return ok;
2507  }
2508 
2509  /* Compute the size required for the StandbySlotNamesConfigData struct */
2510  size = offsetof(StandbySlotNamesConfigData, slot_names);
2511  foreach_ptr(char, slot_name, elemlist)
2512  size += strlen(slot_name) + 1;
2513 
2514  /* GUC extra value must be guc_malloc'd, not palloc'd */
2516 
2517  /* Transform the data into StandbySlotNamesConfigData */
2518  config->nslotnames = list_length(elemlist);
2519 
2520  ptr = config->slot_names;
2521  foreach_ptr(char, slot_name, elemlist)
2522  {
2523  strcpy(ptr, slot_name);
2524  ptr += strlen(slot_name) + 1;
2525  }
2526 
2527  *extra = (void *) config;
2528 
2529  pfree(rawname);
2530  list_free(elemlist);
2531  return true;
2532 }
#define LOG
Definition: elog.h:31
void * guc_malloc(int elevel, size_t size)
Definition: guc.c:640
#define newval
void list_free(List *list)
Definition: list.c:1546
char * pstrdup(const char *in)
Definition: mcxt.c:1683
void pfree(void *pointer)
Definition: mcxt.c:1508
static int list_length(const List *l)
Definition: pg_list.h:152
#define NIL
Definition: pg_list.h:68
#define foreach_ptr(type, var, lst)
Definition: pg_list.h:469
static pg_noinline void Size size
Definition: slab.c:607
static bool validate_standby_slots(char *rawname, List **elemlist)
Definition: slot.c:2426
Definition: pg_list.h:54
char slot_names[FLEXIBLE_ARRAY_MEMBER]
Definition: slot.c:99

References foreach_ptr, guc_malloc(), list_free(), list_length(), LOG, newval, NIL, StandbySlotNamesConfigData::nslotnames, pfree(), pstrdup(), size, StandbySlotNamesConfigData::slot_names, and validate_standby_slots().

◆ CheckPointReplicationSlots()

void CheckPointReplicationSlots ( bool  is_shutdown)

Definition at line 1838 of file slot.c.

1839 {
1840  int i;
1841 
1842  elog(DEBUG1, "performing replication slot checkpoint");
1843 
1844  /*
1845  * Prevent any slot from being created/dropped while we're active. As we
1846  * explicitly do *not* want to block iterating over replication_slots or
1847  * acquiring a slot we cannot take the control lock - but that's OK,
1848  * because holding ReplicationSlotAllocationLock is strictly stronger, and
1849  * enough to guarantee that nobody can change the in_use bits on us.
1850  */
1851  LWLockAcquire(ReplicationSlotAllocationLock, LW_SHARED);
1852 
1853  for (i = 0; i < max_replication_slots; i++)
1854  {
1856  char path[MAXPGPATH];
1857 
1858  if (!s->in_use)
1859  continue;
1860 
1861  /* save the slot to disk, locking is handled in SaveSlotToPath() */
1862  sprintf(path, "pg_replslot/%s", NameStr(s->data.name));
1863 
1864  /*
1865  * Slot's data is not flushed each time the confirmed_flush LSN is
1866  * updated as that could lead to frequent writes. However, we decide
1867  * to force a flush of all logical slot's data at the time of shutdown
1868  * if the confirmed_flush LSN is changed since we last flushed it to
1869  * disk. This helps in avoiding an unnecessary retreat of the
1870  * confirmed_flush LSN after restart.
1871  */
1872  if (is_shutdown && SlotIsLogical(s))
1873  {
1874  SpinLockAcquire(&s->mutex);
1875 
1877 
1878  if (s->data.invalidated == RS_INVAL_NONE &&
1880  {
1881  s->just_dirtied = true;
1882  s->dirty = true;
1883  }
1884  SpinLockRelease(&s->mutex);
1885  }
1886 
1887  SaveSlotToPath(s, path, LOG);
1888  }
1889  LWLockRelease(ReplicationSlotAllocationLock);
1890 }
#define NameStr(name)
Definition: c.h:733
#define DEBUG1
Definition: elog.h:30
#define elog(elevel,...)
Definition: elog.h:224
int i
Definition: isn.c:73
Assert(fmt[strlen(fmt) - 1] !='\n')
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1169
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1782
@ LW_SHARED
Definition: lwlock.h:115
#define MAXPGPATH
#define sprintf
Definition: port.h:240
static void SaveSlotToPath(ReplicationSlot *slot, const char *dir, int elevel)
Definition: slot.c:2019
int max_replication_slots
Definition: slot.c:141
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:135
@ RS_INVAL_NONE
Definition: slot.h:49
#define SlotIsLogical(slot)
Definition: slot.h:210
#define SpinLockRelease(lock)
Definition: spin.h:64
#define SpinLockAcquire(lock)
Definition: spin.h:62
ReplicationSlot replication_slots[1]
Definition: slot.h:221
XLogRecPtr confirmed_flush
Definition: slot.h:104
ReplicationSlotInvalidationCause invalidated
Definition: slot.h:96
slock_t mutex
Definition: slot.h:151
XLogRecPtr last_saved_confirmed_flush
Definition: slot.h:203
bool in_use
Definition: slot.h:154
bool just_dirtied
Definition: slot.h:160
bool dirty
Definition: slot.h:161
ReplicationSlotPersistentData data
Definition: slot.h:178

References Assert(), ReplicationSlotPersistentData::confirmed_flush, ReplicationSlot::data, DEBUG1, ReplicationSlot::dirty, elog, i, ReplicationSlot::in_use, ReplicationSlotPersistentData::invalidated, ReplicationSlot::just_dirtied, ReplicationSlot::last_saved_confirmed_flush, LOG, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_replication_slots, MAXPGPATH, ReplicationSlot::mutex, ReplicationSlotPersistentData::name, NameStr, ReplicationSlotCtlData::replication_slots, ReplicationSlotCtl, RS_INVAL_NONE, SaveSlotToPath(), SlotIsLogical, SpinLockAcquire, SpinLockRelease, and sprintf.

Referenced by CheckPointGuts().

◆ CheckSlotPermissions()

void CheckSlotPermissions ( void  )

Definition at line 1383 of file slot.c.

1384 {
1385  if (!has_rolreplication(GetUserId()))
1386  ereport(ERROR,
1387  (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
1388  errmsg("permission denied to use replication slots"),
1389  errdetail("Only roles with the %s attribute may use replication slots.",
1390  "REPLICATION")));
1391 }
int errdetail(const char *fmt,...)
Definition: elog.c:1205
int errcode(int sqlerrcode)
Definition: elog.c:859
int errmsg(const char *fmt,...)
Definition: elog.c:1072
#define ERROR
Definition: elog.h:39
#define ereport(elevel,...)
Definition: elog.h:149
Oid GetUserId(void)
Definition: miscinit.c:514
bool has_rolreplication(Oid roleid)
Definition: miscinit.c:711

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

Referenced by copy_replication_slot(), pg_create_logical_replication_slot(), pg_create_physical_replication_slot(), pg_drop_replication_slot(), pg_logical_slot_get_changes_guts(), pg_replication_slot_advance(), and pg_sync_replication_slots().

◆ CheckSlotRequirements()

void CheckSlotRequirements ( void  )

Definition at line 1361 of file slot.c.

1362 {
1363  /*
1364  * NB: Adding a new requirement likely means that RestoreSlotFromDisk()
1365  * needs the same check.
1366  */
1367 
1368  if (max_replication_slots == 0)
1369  ereport(ERROR,
1370  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1371  errmsg("replication slots can only be used if max_replication_slots > 0")));
1372 
1374  ereport(ERROR,
1375  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1376  errmsg("replication slots can only be used if wal_level >= replica")));
1377 }
int wal_level
Definition: xlog.c:131
@ WAL_LEVEL_REPLICA
Definition: xlog.h:73

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

◆ CreateSlotOnDisk()

static void CreateSlotOnDisk ( ReplicationSlot slot)
static

Definition at line 1958 of file slot.c.

1959 {
1960  char tmppath[MAXPGPATH];
1961  char path[MAXPGPATH];
1962  struct stat st;
1963 
1964  /*
1965  * No need to take out the io_in_progress_lock, nobody else can see this
1966  * slot yet, so nobody else will write. We're reusing SaveSlotToPath which
1967  * takes out the lock, if we'd take the lock here, we'd deadlock.
1968  */
1969 
1970  sprintf(path, "pg_replslot/%s", NameStr(slot->data.name));
1971  sprintf(tmppath, "pg_replslot/%s.tmp", NameStr(slot->data.name));
1972 
1973  /*
1974  * It's just barely possible that some previous effort to create or drop a
1975  * slot with this name left a temp directory lying around. If that seems
1976  * to be the case, try to remove it. If the rmtree() fails, we'll error
1977  * out at the MakePGDirectory() below, so we don't bother checking
1978  * success.
1979  */
1980  if (stat(tmppath, &st) == 0 && S_ISDIR(st.st_mode))
1981  rmtree(tmppath, true);
1982 
1983  /* Create and fsync the temporary slot directory. */
1984  if (MakePGDirectory(tmppath) < 0)
1985  ereport(ERROR,
1987  errmsg("could not create directory \"%s\": %m",
1988  tmppath)));
1989  fsync_fname(tmppath, true);
1990 
1991  /* Write the actual state file. */
1992  slot->dirty = true; /* signal that we really need to write */
1993  SaveSlotToPath(slot, tmppath, ERROR);
1994 
1995  /* Rename the directory into place. */
1996  if (rename(tmppath, path) != 0)
1997  ereport(ERROR,
1999  errmsg("could not rename file \"%s\" to \"%s\": %m",
2000  tmppath, path)));
2001 
2002  /*
2003  * If we'd now fail - really unlikely - we wouldn't know whether this slot
2004  * would persist after an OS crash or not - so, force a restart. The
2005  * restart would try to fsync this again till it works.
2006  */
2008 
2009  fsync_fname(path, true);
2010  fsync_fname("pg_replslot", true);
2011 
2012  END_CRIT_SECTION();
2013 }
int errcode_for_file_access(void)
Definition: elog.c:882
int MakePGDirectory(const char *directoryName)
Definition: fd.c:3913
void fsync_fname(const char *fname, bool isdir)
Definition: fd.c:756
#define START_CRIT_SECTION()
Definition: miscadmin.h:149
#define END_CRIT_SECTION()
Definition: miscadmin.h:151
bool rmtree(const char *path, bool rmtopdir)
Definition: rmtree.c:50
#define stat
Definition: win32_port.h:284
#define S_ISDIR(m)
Definition: win32_port.h:325

References ReplicationSlot::data, ReplicationSlot::dirty, END_CRIT_SECTION, ereport, errcode_for_file_access(), errmsg(), ERROR, fsync_fname(), MakePGDirectory(), MAXPGPATH, ReplicationSlotPersistentData::name, NameStr, rmtree(), S_ISDIR, SaveSlotToPath(), sprintf, stat::st_mode, START_CRIT_SECTION, and stat.

Referenced by ReplicationSlotCreate().

◆ GetSlotInvalidationCause()

ReplicationSlotInvalidationCause GetSlotInvalidationCause ( const char *  invalidation_reason)

Definition at line 2398 of file slot.c.

2399 {
2402  bool found PG_USED_FOR_ASSERTS_ONLY = false;
2403 
2404  Assert(invalidation_reason);
2405 
2406  for (cause = RS_INVAL_NONE; cause <= RS_INVAL_MAX_CAUSES; cause++)
2407  {
2408  if (strcmp(SlotInvalidationCauses[cause], invalidation_reason) == 0)
2409  {
2410  found = true;
2411  result = cause;
2412  break;
2413  }
2414  }
2415 
2416  Assert(found);
2417  return result;
2418 }
#define PG_USED_FOR_ASSERTS_ONLY
Definition: c.h:169
#define RS_INVAL_MAX_CAUSES
Definition: slot.c:113
const char *const SlotInvalidationCauses[]
Definition: slot.c:105
ReplicationSlotInvalidationCause
Definition: slot.h:48

References Assert(), PG_USED_FOR_ASSERTS_ONLY, RS_INVAL_MAX_CAUSES, RS_INVAL_NONE, and SlotInvalidationCauses.

Referenced by synchronize_slots().

◆ InvalidateObsoleteReplicationSlots()

bool InvalidateObsoleteReplicationSlots ( ReplicationSlotInvalidationCause  cause,
XLogSegNo  oldestSegno,
Oid  dboid,
TransactionId  snapshotConflictHorizon 
)

Definition at line 1782 of file slot.c.

1785 {
1786  XLogRecPtr oldestLSN;
1787  bool invalidated = false;
1788 
1789  Assert(cause != RS_INVAL_HORIZON || TransactionIdIsValid(snapshotConflictHorizon));
1790  Assert(cause != RS_INVAL_WAL_REMOVED || oldestSegno > 0);
1791  Assert(cause != RS_INVAL_NONE);
1792 
1793  if (max_replication_slots == 0)
1794  return invalidated;
1795 
1796  XLogSegNoOffsetToRecPtr(oldestSegno, 0, wal_segment_size, oldestLSN);
1797 
1798 restart:
1799  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
1800  for (int i = 0; i < max_replication_slots; i++)
1801  {
1803 
1804  if (!s->in_use)
1805  continue;
1806 
1807  if (InvalidatePossiblyObsoleteSlot(cause, s, oldestLSN, dboid,
1808  snapshotConflictHorizon,
1809  &invalidated))
1810  {
1811  /* if the lock was released, start from scratch */
1812  goto restart;
1813  }
1814  }
1815  LWLockRelease(ReplicationSlotControlLock);
1816 
1817  /*
1818  * If any slots have been invalidated, recalculate the resource limits.
1819  */
1820  if (invalidated)
1821  {
1824  }
1825 
1826  return invalidated;
1827 }
void ReplicationSlotsComputeRequiredXmin(bool already_locked)
Definition: slot.c:1048
void ReplicationSlotsComputeRequiredLSN(void)
Definition: slot.c:1104
static bool InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause, ReplicationSlot *s, XLogRecPtr oldestLSN, Oid dboid, TransactionId snapshotConflictHorizon, bool *invalidated)
Definition: slot.c:1542
@ RS_INVAL_WAL_REMOVED
Definition: slot.h:51
@ RS_INVAL_HORIZON
Definition: slot.h:53
#define TransactionIdIsValid(xid)
Definition: transam.h:41
int wal_segment_size
Definition: xlog.c:143
#define XLogSegNoOffsetToRecPtr(segno, offset, wal_segsz_bytes, dest)
uint64 XLogRecPtr
Definition: xlogdefs.h:21

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

Referenced by CreateCheckPoint(), CreateRestartPoint(), ResolveRecoveryConflictWithSnapshot(), and xlog_redo().

◆ InvalidatePossiblyObsoleteSlot()

static bool InvalidatePossiblyObsoleteSlot ( ReplicationSlotInvalidationCause  cause,
ReplicationSlot s,
XLogRecPtr  oldestLSN,
Oid  dboid,
TransactionId  snapshotConflictHorizon,
bool invalidated 
)
static

Definition at line 1542 of file slot.c.

1547 {
1548  int last_signaled_pid = 0;
1549  bool released_lock = false;
1550  bool terminated = false;
1551  XLogRecPtr initial_effective_xmin = InvalidXLogRecPtr;
1552  XLogRecPtr initial_catalog_effective_xmin = InvalidXLogRecPtr;
1553  XLogRecPtr initial_restart_lsn = InvalidXLogRecPtr;
1555 
1556  for (;;)
1557  {
1558  XLogRecPtr restart_lsn;
1559  NameData slotname;
1560  int active_pid = 0;
1561  ReplicationSlotInvalidationCause invalidation_cause = RS_INVAL_NONE;
1562 
1563  Assert(LWLockHeldByMeInMode(ReplicationSlotControlLock, LW_SHARED));
1564 
1565  if (!s->in_use)
1566  {
1567  if (released_lock)
1568  LWLockRelease(ReplicationSlotControlLock);
1569  break;
1570  }
1571 
1572  /*
1573  * Check if the slot needs to be invalidated. If it needs to be
1574  * invalidated, and is not currently acquired, acquire it and mark it
1575  * as having been invalidated. We do this with the spinlock held to
1576  * avoid race conditions -- for example the restart_lsn could move
1577  * forward, or the slot could be dropped.
1578  */
1579  SpinLockAcquire(&s->mutex);
1580 
1581  restart_lsn = s->data.restart_lsn;
1582 
1583  /* we do nothing if the slot is already invalid */
1584  if (s->data.invalidated == RS_INVAL_NONE)
1585  {
1586  /*
1587  * The slot's mutex will be released soon, and it is possible that
1588  * those values change since the process holding the slot has been
1589  * terminated (if any), so record them here to ensure that we
1590  * would report the correct invalidation cause.
1591  */
1592  if (!terminated)
1593  {
1594  initial_restart_lsn = s->data.restart_lsn;
1595  initial_effective_xmin = s->effective_xmin;
1596  initial_catalog_effective_xmin = s->effective_catalog_xmin;
1597  }
1598 
1599  switch (cause)
1600  {
1601  case RS_INVAL_WAL_REMOVED:
1602  if (initial_restart_lsn != InvalidXLogRecPtr &&
1603  initial_restart_lsn < oldestLSN)
1604  invalidation_cause = cause;
1605  break;
1606  case RS_INVAL_HORIZON:
1607  if (!SlotIsLogical(s))
1608  break;
1609  /* invalid DB oid signals a shared relation */
1610  if (dboid != InvalidOid && dboid != s->data.database)
1611  break;
1612  if (TransactionIdIsValid(initial_effective_xmin) &&
1613  TransactionIdPrecedesOrEquals(initial_effective_xmin,
1614  snapshotConflictHorizon))
1615  invalidation_cause = cause;
1616  else if (TransactionIdIsValid(initial_catalog_effective_xmin) &&
1617  TransactionIdPrecedesOrEquals(initial_catalog_effective_xmin,
1618  snapshotConflictHorizon))
1619  invalidation_cause = cause;
1620  break;
1621  case RS_INVAL_WAL_LEVEL:
1622  if (SlotIsLogical(s))
1623  invalidation_cause = cause;
1624  break;
1625  case RS_INVAL_NONE:
1626  pg_unreachable();
1627  }
1628  }
1629 
1630  /*
1631  * The invalidation cause recorded previously should not change while
1632  * the process owning the slot (if any) has been terminated.
1633  */
1634  Assert(!(invalidation_cause_prev != RS_INVAL_NONE && terminated &&
1635  invalidation_cause_prev != invalidation_cause));
1636 
1637  /* if there's no invalidation, we're done */
1638  if (invalidation_cause == RS_INVAL_NONE)
1639  {
1640  SpinLockRelease(&s->mutex);
1641  if (released_lock)
1642  LWLockRelease(ReplicationSlotControlLock);
1643  break;
1644  }
1645 
1646  slotname = s->data.name;
1647  active_pid = s->active_pid;
1648 
1649  /*
1650  * If the slot can be acquired, do so and mark it invalidated
1651  * immediately. Otherwise we'll signal the owning process, below, and
1652  * retry.
1653  */
1654  if (active_pid == 0)
1655  {
1656  MyReplicationSlot = s;
1657  s->active_pid = MyProcPid;
1658  s->data.invalidated = invalidation_cause;
1659 
1660  /*
1661  * XXX: We should consider not overwriting restart_lsn and instead
1662  * just rely on .invalidated.
1663  */
1664  if (invalidation_cause == RS_INVAL_WAL_REMOVED)
1666 
1667  /* Let caller know */
1668  *invalidated = true;
1669  }
1670 
1671  SpinLockRelease(&s->mutex);
1672 
1673  /*
1674  * The logical replication slots shouldn't be invalidated as GUC
1675  * max_slot_wal_keep_size is set to -1 during the binary upgrade. See
1676  * check_old_cluster_for_valid_slots() where we ensure that no
1677  * invalidated before the upgrade.
1678  */
1679  Assert(!(*invalidated && SlotIsLogical(s) && IsBinaryUpgrade));
1680 
1681  if (active_pid != 0)
1682  {
1683  /*
1684  * Prepare the sleep on the slot's condition variable before
1685  * releasing the lock, to close a possible race condition if the
1686  * slot is released before the sleep below.
1687  */
1689 
1690  LWLockRelease(ReplicationSlotControlLock);
1691  released_lock = true;
1692 
1693  /*
1694  * Signal to terminate the process that owns the slot, if we
1695  * haven't already signalled it. (Avoidance of repeated
1696  * signalling is the only reason for there to be a loop in this
1697  * routine; otherwise we could rely on caller's restart loop.)
1698  *
1699  * There is the race condition that other process may own the slot
1700  * after its current owner process is terminated and before this
1701  * process owns it. To handle that, we signal only if the PID of
1702  * the owning process has changed from the previous time. (This
1703  * logic assumes that the same PID is not reused very quickly.)
1704  */
1705  if (last_signaled_pid != active_pid)
1706  {
1707  ReportSlotInvalidation(invalidation_cause, true, active_pid,
1708  slotname, restart_lsn,
1709  oldestLSN, snapshotConflictHorizon);
1710 
1711  if (MyBackendType == B_STARTUP)
1712  (void) SendProcSignal(active_pid,
1715  else
1716  (void) kill(active_pid, SIGTERM);
1717 
1718  last_signaled_pid = active_pid;
1719  terminated = true;
1720  invalidation_cause_prev = invalidation_cause;
1721  }
1722 
1723  /* Wait until the slot is released. */
1725  WAIT_EVENT_REPLICATION_SLOT_DROP);
1726 
1727  /*
1728  * Re-acquire lock and start over; we expect to invalidate the
1729  * slot next time (unless another process acquires the slot in the
1730  * meantime).
1731  */
1732  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
1733  continue;
1734  }
1735  else
1736  {
1737  /*
1738  * We hold the slot now and have already invalidated it; flush it
1739  * to ensure that state persists.
1740  *
1741  * Don't want to hold ReplicationSlotControlLock across file
1742  * system operations, so release it now but be sure to tell caller
1743  * to restart from scratch.
1744  */
1745  LWLockRelease(ReplicationSlotControlLock);
1746  released_lock = true;
1747 
1748  /* Make sure the invalidated state persists across server restart */
1752 
1753  ReportSlotInvalidation(invalidation_cause, false, active_pid,
1754  slotname, restart_lsn,
1755  oldestLSN, snapshotConflictHorizon);
1756 
1757  /* done with this slot for now */
1758  break;
1759  }
1760  }
1761 
1762  Assert(released_lock == !LWLockHeldByMe(ReplicationSlotControlLock));
1763 
1764  return released_lock;
1765 }
#define pg_unreachable()
Definition: c.h:283
void ConditionVariablePrepareToSleep(ConditionVariable *cv)
void ConditionVariableSleep(ConditionVariable *cv, uint32 wait_event_info)
bool IsBinaryUpgrade
Definition: globals.c:118
int MyProcPid
Definition: globals.c:45
bool LWLockHeldByMe(LWLock *lock)
Definition: lwlock.c:1894
bool LWLockHeldByMeInMode(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1938
@ B_STARTUP
Definition: miscadmin.h:358
BackendType MyBackendType
Definition: miscinit.c:63
#define InvalidOid
Definition: postgres_ext.h:36
#define INVALID_PROC_NUMBER
Definition: procnumber.h:26
int SendProcSignal(pid_t pid, ProcSignalReason reason, ProcNumber procNumber)
Definition: procsignal.c:257
@ PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT
Definition: procsignal.h:46
void ReplicationSlotMarkDirty(void)
Definition: slot.c:1009
ReplicationSlot * MyReplicationSlot
Definition: slot.c:138
void ReplicationSlotSave(void)
Definition: slot.c:991
void ReplicationSlotRelease(void)
Definition: slot.c:652
static void ReportSlotInvalidation(ReplicationSlotInvalidationCause cause, bool terminating, int pid, NameData slotname, XLogRecPtr restart_lsn, XLogRecPtr oldestLSN, TransactionId snapshotConflictHorizon)
Definition: slot.c:1476
@ RS_INVAL_WAL_LEVEL
Definition: slot.h:55
XLogRecPtr restart_lsn
Definition: slot.h:93
TransactionId effective_catalog_xmin
Definition: slot.h:175
pid_t active_pid
Definition: slot.h:157
TransactionId effective_xmin
Definition: slot.h:174
ConditionVariable active_cv
Definition: slot.h:184
Definition: c.h:728
bool TransactionIdPrecedesOrEquals(TransactionId id1, TransactionId id2)
Definition: transam.c:299
#define kill(pid, sig)
Definition: win32_port.h:485

References ReplicationSlot::active_cv, ReplicationSlot::active_pid, Assert(), B_STARTUP, ConditionVariablePrepareToSleep(), ConditionVariableSleep(), ReplicationSlot::data, ReplicationSlotPersistentData::database, ReplicationSlot::effective_catalog_xmin, ReplicationSlot::effective_xmin, ReplicationSlot::in_use, INVALID_PROC_NUMBER, ReplicationSlotPersistentData::invalidated, InvalidOid, InvalidXLogRecPtr, IsBinaryUpgrade, kill, LW_SHARED, LWLockAcquire(), LWLockHeldByMe(), LWLockHeldByMeInMode(), LWLockRelease(), ReplicationSlot::mutex, MyBackendType, MyProcPid, MyReplicationSlot, ReplicationSlotPersistentData::name, pg_unreachable, PG_USED_FOR_ASSERTS_ONLY, PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT, ReplicationSlotMarkDirty(), ReplicationSlotRelease(), ReplicationSlotSave(), ReportSlotInvalidation(), ReplicationSlotPersistentData::restart_lsn, RS_INVAL_HORIZON, RS_INVAL_NONE, RS_INVAL_WAL_LEVEL, RS_INVAL_WAL_REMOVED, SendProcSignal(), SlotIsLogical, SpinLockAcquire, SpinLockRelease, TransactionIdIsValid, and TransactionIdPrecedesOrEquals().

Referenced by InvalidateObsoleteReplicationSlots().

◆ ReplicationSlotAcquire()

void ReplicationSlotAcquire ( const char *  name,
bool  nowait 
)

Definition at line 540 of file slot.c.

541 {
542  ReplicationSlot *s;
543  int active_pid;
544 
545  Assert(name != NULL);
546 
547 retry:
548  Assert(MyReplicationSlot == NULL);
549 
550  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
551 
552  /* Check if the slot exits with the given name. */
553  s = SearchNamedReplicationSlot(name, false);
554  if (s == NULL || !s->in_use)
555  {
556  LWLockRelease(ReplicationSlotControlLock);
557 
558  ereport(ERROR,
559  (errcode(ERRCODE_UNDEFINED_OBJECT),
560  errmsg("replication slot \"%s\" does not exist",
561  name)));
562  }
563 
564  /*
565  * This is the slot we want; check if it's active under some other
566  * process. In single user mode, we don't need this check.
567  */
568  if (IsUnderPostmaster)
569  {
570  /*
571  * Get ready to sleep on the slot in case it is active. (We may end
572  * up not sleeping, but we don't want to do this while holding the
573  * spinlock.)
574  */
575  if (!nowait)
577 
578  SpinLockAcquire(&s->mutex);
579  if (s->active_pid == 0)
580  s->active_pid = MyProcPid;
581  active_pid = s->active_pid;
582  SpinLockRelease(&s->mutex);
583  }
584  else
585  active_pid = MyProcPid;
586  LWLockRelease(ReplicationSlotControlLock);
587 
588  /*
589  * If we found the slot but it's already active in another process, we
590  * wait until the owning process signals us that it's been released, or
591  * error out.
592  */
593  if (active_pid != MyProcPid)
594  {
595  if (!nowait)
596  {
597  /* Wait here until we get signaled, and then restart */
599  WAIT_EVENT_REPLICATION_SLOT_DROP);
601  goto retry;
602  }
603 
604  ereport(ERROR,
605  (errcode(ERRCODE_OBJECT_IN_USE),
606  errmsg("replication slot \"%s\" is active for PID %d",
607  NameStr(s->data.name), active_pid)));
608  }
609  else if (!nowait)
610  ConditionVariableCancelSleep(); /* no sleep needed after all */
611 
612  /* Let everybody know we've modified this slot */
614 
615  /* We made this slot active, so it's ours now. */
616  MyReplicationSlot = s;
617 
618  /*
619  * The call to pgstat_acquire_replslot() protects against stats for a
620  * different slot, from before a restart or such, being present during
621  * pgstat_report_replslot().
622  */
623  if (SlotIsLogical(s))
625 
626  /*
627  * Reset the time since the slot has become inactive as the slot is active
628  * now.
629  */
630  SpinLockAcquire(&s->mutex);
631  s->inactive_since = 0;
632  SpinLockRelease(&s->mutex);
633 
634  if (am_walsender)
635  {
637  SlotIsLogical(s)
638  ? errmsg("acquired logical replication slot \"%s\"",
639  NameStr(s->data.name))
640  : errmsg("acquired physical replication slot \"%s\"",
641  NameStr(s->data.name)));
642  }
643 }
bool ConditionVariableCancelSleep(void)
void ConditionVariableBroadcast(ConditionVariable *cv)
bool IsUnderPostmaster
Definition: globals.c:117
void pgstat_acquire_replslot(ReplicationSlot *slot)
ReplicationSlot * SearchNamedReplicationSlot(const char *name, bool need_lock)
Definition: slot.c:464
TimestampTz inactive_since
Definition: slot.h:206
const char * name
bool am_walsender
Definition: walsender.c:115
bool log_replication_commands
Definition: walsender.c:125

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

Referenced by binary_upgrade_logical_slot_has_caught_up(), drop_local_obsolete_slots(), pg_logical_slot_get_changes_guts(), pg_replication_slot_advance(), ReplicationSlotAlter(), ReplicationSlotDrop(), StartLogicalReplication(), StartReplication(), and synchronize_one_slot().

◆ ReplicationSlotAlter()

void ReplicationSlotAlter ( const char *  name,
bool  failover 
)

Definition at line 806 of file slot.c.

807 {
808  Assert(MyReplicationSlot == NULL);
809 
811 
813  ereport(ERROR,
814  errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
815  errmsg("cannot use %s with a physical replication slot",
816  "ALTER_REPLICATION_SLOT"));
817 
818  if (RecoveryInProgress())
819  {
820  /*
821  * Do not allow users to alter the slots which are currently being
822  * synced from the primary to the standby.
823  */
825  ereport(ERROR,
826  errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
827  errmsg("cannot alter replication slot \"%s\"", name),
828  errdetail("This slot is being synced from the primary server."));
829 
830  /*
831  * Do not allow users to enable failover on the standby as we do not
832  * support sync to the cascading standby.
833  */
834  if (failover)
835  ereport(ERROR,
836  errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
837  errmsg("cannot enable failover for a replication slot"
838  " on the standby"));
839  }
840 
841  /*
842  * Do not allow users to enable failover for temporary slots as we do not
843  * support syncing temporary slots to the standby.
844  */
845  if (failover && MyReplicationSlot->data.persistency == RS_TEMPORARY)
846  ereport(ERROR,
847  errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
848  errmsg("cannot enable failover for a temporary replication slot"));
849 
850  if (MyReplicationSlot->data.failover != failover)
851  {
853  MyReplicationSlot->data.failover = failover;
855 
858  }
859 
861 }
void ReplicationSlotAcquire(const char *name, bool nowait)
Definition: slot.c:540
@ RS_TEMPORARY
Definition: slot.h:37
#define SlotIsPhysical(slot)
Definition: slot.h:209
ReplicationSlotPersistency persistency
Definition: slot.h:74
bool RecoveryInProgress(void)
Definition: xlog.c:6201

References Assert(), ReplicationSlot::data, ereport, errcode(), errdetail(), errmsg(), ERROR, ReplicationSlotPersistentData::failover, ReplicationSlot::mutex, MyReplicationSlot, name, ReplicationSlotPersistentData::persistency, RecoveryInProgress(), ReplicationSlotAcquire(), ReplicationSlotMarkDirty(), ReplicationSlotRelease(), ReplicationSlotSave(), RS_TEMPORARY, SlotIsPhysical, SpinLockAcquire, SpinLockRelease, and ReplicationSlotPersistentData::synced.

Referenced by AlterReplicationSlot().

◆ ReplicationSlotCleanup()

void ReplicationSlotCleanup ( void  )

Definition at line 745 of file slot.c.

746 {
747  int i;
748 
749  Assert(MyReplicationSlot == NULL);
750 
751 restart:
752  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
753  for (i = 0; i < max_replication_slots; i++)
754  {
756 
757  if (!s->in_use)
758  continue;
759 
760  SpinLockAcquire(&s->mutex);
761  if (s->active_pid == MyProcPid)
762  {
764  SpinLockRelease(&s->mutex);
765  LWLockRelease(ReplicationSlotControlLock); /* avoid deadlock */
766 
768 
770  goto restart;
771  }
772  else
773  SpinLockRelease(&s->mutex);
774  }
775 
776  LWLockRelease(ReplicationSlotControlLock);
777 }
static void ReplicationSlotDropPtr(ReplicationSlot *slot)
Definition: slot.c:884

References ReplicationSlot::active_cv, ReplicationSlot::active_pid, Assert(), ConditionVariableBroadcast(), ReplicationSlot::data, i, ReplicationSlot::in_use, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_replication_slots, ReplicationSlot::mutex, MyProcPid, MyReplicationSlot, ReplicationSlotPersistentData::persistency, ReplicationSlotCtlData::replication_slots, ReplicationSlotCtl, ReplicationSlotDropPtr(), RS_TEMPORARY, SpinLockAcquire, and SpinLockRelease.

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

◆ ReplicationSlotCreate()

void ReplicationSlotCreate ( const char *  name,
bool  db_specific,
ReplicationSlotPersistency  persistency,
bool  two_phase,
bool  failover,
bool  synced 
)

Definition at line 309 of file slot.c.

312 {
313  ReplicationSlot *slot = NULL;
314  int i;
315 
316  Assert(MyReplicationSlot == NULL);
317 
319 
320  if (failover)
321  {
322  /*
323  * Do not allow users to create the failover enabled slots on the
324  * standby as we do not support sync to the cascading standby.
325  *
326  * However, failover enabled slots can be created during slot
327  * synchronization because we need to retain the same values as the
328  * remote slot.
329  */
331  ereport(ERROR,
332  errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
333  errmsg("cannot enable failover for a replication slot created on the standby"));
334 
335  /*
336  * Do not allow users to create failover enabled temporary slots,
337  * because temporary slots will not be synced to the standby.
338  *
339  * However, failover enabled temporary slots can be created during
340  * slot synchronization. See the comments atop slotsync.c for details.
341  */
342  if (persistency == RS_TEMPORARY && !IsSyncingReplicationSlots())
343  ereport(ERROR,
344  errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
345  errmsg("cannot enable failover for a temporary replication slot"));
346  }
347 
348  /*
349  * If some other backend ran this code concurrently with us, we'd likely
350  * both allocate the same slot, and that would be bad. We'd also be at
351  * risk of missing a name collision. Also, we don't want to try to create
352  * a new slot while somebody's busy cleaning up an old one, because we
353  * might both be monkeying with the same directory.
354  */
355  LWLockAcquire(ReplicationSlotAllocationLock, LW_EXCLUSIVE);
356 
357  /*
358  * Check for name collision, and identify an allocatable slot. We need to
359  * hold ReplicationSlotControlLock in shared mode for this, so that nobody
360  * else can change the in_use flags while we're looking at them.
361  */
362  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
363  for (i = 0; i < max_replication_slots; i++)
364  {
366 
367  if (s->in_use && strcmp(name, NameStr(s->data.name)) == 0)
368  ereport(ERROR,
370  errmsg("replication slot \"%s\" already exists", name)));
371  if (!s->in_use && slot == NULL)
372  slot = s;
373  }
374  LWLockRelease(ReplicationSlotControlLock);
375 
376  /* If all slots are in use, we're out of luck. */
377  if (slot == NULL)
378  ereport(ERROR,
379  (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
380  errmsg("all replication slots are in use"),
381  errhint("Free one or increase max_replication_slots.")));
382 
383  /*
384  * Since this slot is not in use, nobody should be looking at any part of
385  * it other than the in_use field unless they're trying to allocate it.
386  * And since we hold ReplicationSlotAllocationLock, nobody except us can
387  * be doing that. So it's safe to initialize the slot.
388  */
389  Assert(!slot->in_use);
390  Assert(slot->active_pid == 0);
391 
392  /* first initialize persistent data */
393  memset(&slot->data, 0, sizeof(ReplicationSlotPersistentData));
394  namestrcpy(&slot->data.name, name);
395  slot->data.database = db_specific ? MyDatabaseId : InvalidOid;
396  slot->data.persistency = persistency;
397  slot->data.two_phase = two_phase;
399  slot->data.failover = failover;
400  slot->data.synced = synced;
401 
402  /* and then data only present in shared memory */
403  slot->just_dirtied = false;
404  slot->dirty = false;
412  slot->inactive_since = 0;
413 
414  /*
415  * Create the slot on disk. We haven't actually marked the slot allocated
416  * yet, so no special cleanup is required if this errors out.
417  */
418  CreateSlotOnDisk(slot);
419 
420  /*
421  * We need to briefly prevent any other backend from iterating over the
422  * slots while we flip the in_use flag. We also need to set the active
423  * flag while holding the ControlLock as otherwise a concurrent
424  * ReplicationSlotAcquire() could acquire the slot as well.
425  */
426  LWLockAcquire(ReplicationSlotControlLock, LW_EXCLUSIVE);
427 
428  slot->in_use = true;
429 
430  /* We can now mark the slot active, and that makes it our slot. */
431  SpinLockAcquire(&slot->mutex);
432  Assert(slot->active_pid == 0);
433  slot->active_pid = MyProcPid;
434  SpinLockRelease(&slot->mutex);
435  MyReplicationSlot = slot;
436 
437  LWLockRelease(ReplicationSlotControlLock);
438 
439  /*
440  * Create statistics entry for the new logical slot. We don't collect any
441  * stats for physical slots, so no need to create an entry for the same.
442  * See ReplicationSlotDropPtr for why we need to do this before releasing
443  * ReplicationSlotAllocationLock.
444  */
445  if (SlotIsLogical(slot))
447 
448  /*
449  * Now that the slot has been marked as in_use and active, it's safe to
450  * let somebody else try to allocate a slot.
451  */
452  LWLockRelease(ReplicationSlotAllocationLock);
453 
454  /* Let everybody know we've modified this slot */
456 }
int errhint(const char *fmt,...)
Definition: elog.c:1319
Oid MyDatabaseId
Definition: globals.c:91
@ LW_EXCLUSIVE
Definition: lwlock.h:114
void namestrcpy(Name name, const char *str)
Definition: name.c:233
static bool two_phase
void pgstat_create_replslot(ReplicationSlot *slot)
static void CreateSlotOnDisk(ReplicationSlot *slot)
Definition: slot.c:1958
bool ReplicationSlotValidateName(const char *name, int elevel)
Definition: slot.c:252
bool IsSyncingReplicationSlots(void)
Definition: slotsync.c:1378
#define ERRCODE_DUPLICATE_OBJECT
Definition: streamutil.c:32
XLogRecPtr candidate_xmin_lsn
Definition: slot.h:194
XLogRecPtr candidate_restart_valid
Definition: slot.h:195
XLogRecPtr candidate_restart_lsn
Definition: slot.h:196
TransactionId candidate_catalog_xmin
Definition: slot.h:193
#define InvalidTransactionId
Definition: transam.h:31

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, ReplicationSlotPersistentData::failover, i, ReplicationSlot::in_use, ReplicationSlot::inactive_since, InvalidOid, InvalidTransactionId, InvalidXLogRecPtr, IsSyncingReplicationSlots(), ReplicationSlot::just_dirtied, ReplicationSlot::last_saved_confirmed_flush, LW_EXCLUSIVE, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_replication_slots, ReplicationSlot::mutex, MyDatabaseId, MyProcPid, MyReplicationSlot, name, ReplicationSlotPersistentData::name, NameStr, namestrcpy(), ReplicationSlotPersistentData::persistency, pgstat_create_replslot(), RecoveryInProgress(), ReplicationSlotCtlData::replication_slots, ReplicationSlotCtl, ReplicationSlotValidateName(), RS_TEMPORARY, SlotIsLogical, SpinLockAcquire, SpinLockRelease, ReplicationSlotPersistentData::synced, two_phase, ReplicationSlotPersistentData::two_phase, and ReplicationSlotPersistentData::two_phase_at.

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

◆ ReplicationSlotDrop()

void ReplicationSlotDrop ( const char *  name,
bool  nowait 
)

Definition at line 783 of file slot.c.

784 {
785  Assert(MyReplicationSlot == NULL);
786 
787  ReplicationSlotAcquire(name, nowait);
788 
789  /*
790  * Do not allow users to drop the slots which are currently being synced
791  * from the primary to the standby.
792  */
794  ereport(ERROR,
795  errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
796  errmsg("cannot drop replication slot \"%s\"", name),
797  errdetail("This slot is being synced from the primary server."));
798 
800 }
void ReplicationSlotDropAcquired(void)
Definition: slot.c:867

References Assert(), ReplicationSlot::data, ereport, errcode(), errdetail(), errmsg(), ERROR, MyReplicationSlot, name, RecoveryInProgress(), ReplicationSlotAcquire(), ReplicationSlotDropAcquired(), and ReplicationSlotPersistentData::synced.

Referenced by DropReplicationSlot(), and pg_drop_replication_slot().

◆ ReplicationSlotDropAcquired()

void ReplicationSlotDropAcquired ( void  )

Definition at line 867 of file slot.c.

868 {
870 
871  Assert(MyReplicationSlot != NULL);
872 
873  /* slot isn't acquired anymore */
874  MyReplicationSlot = NULL;
875 
877 }

References Assert(), MyReplicationSlot, and ReplicationSlotDropPtr().

Referenced by drop_local_obsolete_slots(), ReplicationSlotDrop(), ReplicationSlotRelease(), and ReplicationSlotsDropDBSlots().

◆ ReplicationSlotDropPtr()

static void ReplicationSlotDropPtr ( ReplicationSlot slot)
static

Definition at line 884 of file slot.c.

885 {
886  char path[MAXPGPATH];
887  char tmppath[MAXPGPATH];
888 
889  /*
890  * If some other backend ran this code concurrently with us, we might try
891  * to delete a slot with a certain name while someone else was trying to
892  * create a slot with the same name.
893  */
894  LWLockAcquire(ReplicationSlotAllocationLock, LW_EXCLUSIVE);
895 
896  /* Generate pathnames. */
897  sprintf(path, "pg_replslot/%s", NameStr(slot->data.name));
898  sprintf(tmppath, "pg_replslot/%s.tmp", NameStr(slot->data.name));
899 
900  /*
901  * Rename the slot directory on disk, so that we'll no longer recognize
902  * this as a valid slot. Note that if this fails, we've got to mark the
903  * slot inactive before bailing out. If we're dropping an ephemeral or a
904  * temporary slot, we better never fail hard as the caller won't expect
905  * the slot to survive and this might get called during error handling.
906  */
907  if (rename(path, tmppath) == 0)
908  {
909  /*
910  * We need to fsync() the directory we just renamed and its parent to
911  * make sure that our changes are on disk in a crash-safe fashion. If
912  * fsync() fails, we can't be sure whether the changes are on disk or
913  * not. For now, we handle that by panicking;
914  * StartupReplicationSlots() will try to straighten it out after
915  * restart.
916  */
918  fsync_fname(tmppath, true);
919  fsync_fname("pg_replslot", true);
921  }
922  else
923  {
924  bool fail_softly = slot->data.persistency != RS_PERSISTENT;
925 
926  SpinLockAcquire(&slot->mutex);
927  slot->active_pid = 0;
928  SpinLockRelease(&slot->mutex);
929 
930  /* wake up anyone waiting on this slot */
932 
933  ereport(fail_softly ? WARNING : ERROR,
935  errmsg("could not rename file \"%s\" to \"%s\": %m",
936  path, tmppath)));
937  }
938 
939  /*
940  * The slot is definitely gone. Lock out concurrent scans of the array
941  * long enough to kill it. It's OK to clear the active PID here without
942  * grabbing the mutex because nobody else can be scanning the array here,
943  * and nobody can be attached to this slot and thus access it without
944  * scanning the array.
945  *
946  * Also wake up processes waiting for it.
947  */
948  LWLockAcquire(ReplicationSlotControlLock, LW_EXCLUSIVE);
949  slot->active_pid = 0;
950  slot->in_use = false;
951  LWLockRelease(ReplicationSlotControlLock);
953 
954  /*
955  * Slot is dead and doesn't prevent resource removal anymore, recompute
956  * limits.
957  */
960 
961  /*
962  * If removing the directory fails, the worst thing that will happen is
963  * that the user won't be able to create a new slot with the same name
964  * until the next server restart. We warn about it, but that's all.
965  */
966  if (!rmtree(tmppath, true))
968  (errmsg("could not remove directory \"%s\"", tmppath)));
969 
970  /*
971  * Drop the statistics entry for the replication slot. Do this while
972  * holding ReplicationSlotAllocationLock so that we don't drop a
973  * statistics entry for another slot with the same name just created in
974  * another session.
975  */
976  if (SlotIsLogical(slot))
977  pgstat_drop_replslot(slot);
978 
979  /*
980  * We release this at the very end, so that nobody starts trying to create
981  * a slot while we're still cleaning up the detritus of the old one.
982  */
983  LWLockRelease(ReplicationSlotAllocationLock);
984 }
#define WARNING
Definition: elog.h:36
void pgstat_drop_replslot(ReplicationSlot *slot)
@ RS_PERSISTENT
Definition: slot.h:35

References ReplicationSlot::active_cv, ReplicationSlot::active_pid, ConditionVariableBroadcast(), ReplicationSlot::data, END_CRIT_SECTION, ereport, errcode_for_file_access(), errmsg(), ERROR, fsync_fname(), ReplicationSlot::in_use, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MAXPGPATH, ReplicationSlot::mutex, ReplicationSlotPersistentData::name, NameStr, ReplicationSlotPersistentData::persistency, pgstat_drop_replslot(), ReplicationSlotsComputeRequiredLSN(), ReplicationSlotsComputeRequiredXmin(), rmtree(), RS_PERSISTENT, SlotIsLogical, SpinLockAcquire, SpinLockRelease, sprintf, START_CRIT_SECTION, and WARNING.

Referenced by ReplicationSlotCleanup(), and ReplicationSlotDropAcquired().

◆ ReplicationSlotIndex()

int ReplicationSlotIndex ( ReplicationSlot slot)

◆ ReplicationSlotInitialize()

void ReplicationSlotInitialize ( void  )

Definition at line 224 of file slot.c.

225 {
227 }
void before_shmem_exit(pg_on_exit_callback function, Datum arg)
Definition: ipc.c:337
static void ReplicationSlotShmemExit(int code, Datum arg)
Definition: slot.c:233

References before_shmem_exit(), and ReplicationSlotShmemExit().

Referenced by BaseInit().

◆ ReplicationSlotMarkDirty()

◆ ReplicationSlotName()

bool ReplicationSlotName ( int  index,
Name  name 
)

Definition at line 513 of file slot.c.

514 {
515  ReplicationSlot *slot;
516  bool found;
517 
519 
520  /*
521  * Ensure that the slot cannot be dropped while we copy the name. Don't
522  * need the spinlock as the name of an existing slot cannot change.
523  */
524  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
525  found = slot->in_use;
526  if (slot->in_use)
527  namestrcpy(name, NameStr(slot->data.name));
528  LWLockRelease(ReplicationSlotControlLock);
529 
530  return found;
531 }
Definition: type.h:95

References ReplicationSlot::data, ReplicationSlot::in_use, LW_SHARED, LWLockAcquire(), LWLockRelease(), name, ReplicationSlotPersistentData::name, NameStr, namestrcpy(), ReplicationSlotCtlData::replication_slots, and ReplicationSlotCtl.

Referenced by pgstat_replslot_to_serialized_name_cb().

◆ ReplicationSlotPersist()

◆ ReplicationSlotRelease()

void ReplicationSlotRelease ( void  )

Definition at line 652 of file slot.c.

653 {
655  char *slotname = NULL; /* keep compiler quiet */
656  bool is_logical = false; /* keep compiler quiet */
657  TimestampTz now = 0;
658 
659  Assert(slot != NULL && slot->active_pid != 0);
660 
661  if (am_walsender)
662  {
663  slotname = pstrdup(NameStr(slot->data.name));
664  is_logical = SlotIsLogical(slot);
665  }
666 
667  if (slot->data.persistency == RS_EPHEMERAL)
668  {
669  /*
670  * Delete the slot. There is no !PANIC case where this is allowed to
671  * fail, all that may happen is an incomplete cleanup of the on-disk
672  * data.
673  */
675  }
676 
677  /*
678  * If slot needed to temporarily restrain both data and catalog xmin to
679  * create the catalog snapshot, remove that temporary constraint.
680  * Snapshots can only be exported while the initial snapshot is still
681  * acquired.
682  */
683  if (!TransactionIdIsValid(slot->data.xmin) &&
685  {
686  SpinLockAcquire(&slot->mutex);
688  SpinLockRelease(&slot->mutex);
690  }
691 
692  /*
693  * Set the last inactive time after marking the slot inactive. We don't
694  * set it for the slots currently being synced from the primary to the
695  * standby because such slots are typically inactive as decoding is not
696  * allowed on those.
697  */
698  if (!(RecoveryInProgress() && slot->data.synced))
700 
701  if (slot->data.persistency == RS_PERSISTENT)
702  {
703  /*
704  * Mark persistent slot inactive. We're not freeing it, just
705  * disconnecting, but wake up others that may be waiting for it.
706  */
707  SpinLockAcquire(&slot->mutex);
708  slot->active_pid = 0;
709  slot->inactive_since = now;
710  SpinLockRelease(&slot->mutex);
712  }
713  else
714  {
715  SpinLockAcquire(&slot->mutex);
716  slot->inactive_since = now;
717  SpinLockRelease(&slot->mutex);
718  }
719 
720  MyReplicationSlot = NULL;
721 
722  /* might not have been set when we've been a plain slot */
723  LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
726  LWLockRelease(ProcArrayLock);
727 
728  if (am_walsender)
729  {
731  is_logical
732  ? errmsg("released logical replication slot \"%s\"",
733  slotname)
734  : errmsg("released physical replication slot \"%s\"",
735  slotname));
736 
737  pfree(slotname);
738  }
739 }
TimestampTz GetCurrentTimestamp(void)
Definition: timestamp.c:1654
Datum now(PG_FUNCTION_ARGS)
Definition: timestamp.c:1618
int64 TimestampTz
Definition: timestamp.h:39
#define PROC_IN_LOGICAL_DECODING
Definition: proc.h:61
@ RS_EPHEMERAL
Definition: slot.h:36
PGPROC * MyProc
Definition: proc.c:66
PROC_HDR * ProcGlobal
Definition: proc.c:78
uint8 statusFlags
Definition: proc.h:238
int pgxactoff
Definition: proc.h:180
uint8 * statusFlags
Definition: proc.h:395
TransactionId xmin
Definition: slot.h:82

References ReplicationSlot::active_cv, ReplicationSlot::active_pid, am_walsender, Assert(), ConditionVariableBroadcast(), ReplicationSlot::data, DEBUG1, ReplicationSlot::effective_xmin, ereport, errmsg(), GetCurrentTimestamp(), ReplicationSlot::inactive_since, InvalidTransactionId, LOG, log_replication_commands, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), ReplicationSlot::mutex, MyProc, MyReplicationSlot, ReplicationSlotPersistentData::name, NameStr, now(), ReplicationSlotPersistentData::persistency, pfree(), PGPROC::pgxactoff, PROC_IN_LOGICAL_DECODING, ProcGlobal, pstrdup(), RecoveryInProgress(), ReplicationSlotDropAcquired(), ReplicationSlotsComputeRequiredXmin(), RS_EPHEMERAL, RS_PERSISTENT, SlotIsLogical, SpinLockAcquire, SpinLockRelease, PGPROC::statusFlags, PROC_HDR::statusFlags, ReplicationSlotPersistentData::synced, TransactionIdIsValid, and ReplicationSlotPersistentData::xmin.

Referenced by binary_upgrade_logical_slot_has_caught_up(), copy_replication_slot(), CreateReplicationSlot(), InvalidatePossiblyObsoleteSlot(), pg_create_logical_replication_slot(), pg_create_physical_replication_slot(), pg_logical_slot_get_changes_guts(), pg_replication_slot_advance(), PostgresMain(), ReplicationSlotAlter(), ReplicationSlotShmemExit(), StartLogicalReplication(), StartReplication(), synchronize_one_slot(), and WalSndErrorCleanup().

◆ ReplicationSlotReserveWal()

void ReplicationSlotReserveWal ( void  )

Definition at line 1400 of file slot.c.

1401 {
1403 
1404  Assert(slot != NULL);
1406 
1407  /*
1408  * The replication slot mechanism is used to prevent removal of required
1409  * WAL. As there is no interlock between this routine and checkpoints, WAL
1410  * segments could concurrently be removed when a now stale return value of
1411  * ReplicationSlotsComputeRequiredLSN() is used. In the unlikely case that
1412  * this happens we'll just retry.
1413  */
1414  while (true)
1415  {
1416  XLogSegNo segno;
1417  XLogRecPtr restart_lsn;
1418 
1419  /*
1420  * For logical slots log a standby snapshot and start logical decoding
1421  * at exactly that position. That allows the slot to start up more
1422  * quickly. But on a standby we cannot do WAL writes, so just use the
1423  * replay pointer; effectively, an attempt to create a logical slot on
1424  * standby will cause it to wait for an xl_running_xact record to be
1425  * logged independently on the primary, so that a snapshot can be
1426  * built using the record.
1427  *
1428  * None of this is needed (or indeed helpful) for physical slots as
1429  * they'll start replay at the last logged checkpoint anyway. Instead
1430  * return the location of the last redo LSN. While that slightly
1431  * increases the chance that we have to retry, it's where a base
1432  * backup has to start replay at.
1433  */
1434  if (SlotIsPhysical(slot))
1435  restart_lsn = GetRedoRecPtr();
1436  else if (RecoveryInProgress())
1437  restart_lsn = GetXLogReplayRecPtr(NULL);
1438  else
1439  restart_lsn = GetXLogInsertRecPtr();
1440 
1441  SpinLockAcquire(&slot->mutex);
1442  slot->data.restart_lsn = restart_lsn;
1443  SpinLockRelease(&slot->mutex);
1444 
1445  /* prevent WAL removal as fast as possible */
1447 
1448  /*
1449  * If all required WAL is still there, great, otherwise retry. The
1450  * slot should prevent further removal of WAL, unless there's a
1451  * concurrent ReplicationSlotsComputeRequiredLSN() after we've written
1452  * the new restart_lsn above, so normally we should never need to loop
1453  * more than twice.
1454  */
1456  if (XLogGetLastRemovedSegno() < segno)
1457  break;
1458  }
1459 
1460  if (!RecoveryInProgress() && SlotIsLogical(slot))
1461  {
1462  XLogRecPtr flushptr;
1463 
1464  /* make sure we have enough information to start */
1465  flushptr = LogStandbySnapshot();
1466 
1467  /* and make sure it's fsynced to disk */
1468  XLogFlush(flushptr);
1469  }
1470 }
XLogRecPtr LogStandbySnapshot(void)
Definition: standby.c:1285
XLogSegNo XLogGetLastRemovedSegno(void)
Definition: xlog.c:3688
XLogRecPtr GetRedoRecPtr(void)
Definition: xlog.c:6304
XLogRecPtr GetXLogInsertRecPtr(void)
Definition: xlog.c:9266
void XLogFlush(XLogRecPtr record)
Definition: xlog.c:2728
#define XLByteToSeg(xlrp, logSegNo, wal_segsz_bytes)
uint64 XLogSegNo
Definition: xlogdefs.h:48
XLogRecPtr GetXLogReplayRecPtr(TimeLineID *replayTLI)

References Assert(), ReplicationSlot::data, GetRedoRecPtr(), GetXLogInsertRecPtr(), GetXLogReplayRecPtr(), InvalidXLogRecPtr, LogStandbySnapshot(), ReplicationSlot::mutex, MyReplicationSlot, RecoveryInProgress(), ReplicationSlotsComputeRequiredLSN(), ReplicationSlotPersistentData::restart_lsn, SlotIsLogical, SlotIsPhysical, SpinLockAcquire, SpinLockRelease, wal_segment_size, XLByteToSeg, XLogFlush(), and XLogGetLastRemovedSegno().

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

◆ ReplicationSlotSave()

◆ ReplicationSlotsComputeLogicalRestartLSN()

XLogRecPtr ReplicationSlotsComputeLogicalRestartLSN ( void  )

Definition at line 1153 of file slot.c.

1154 {
1155  XLogRecPtr result = InvalidXLogRecPtr;
1156  int i;
1157 
1158  if (max_replication_slots <= 0)
1159  return InvalidXLogRecPtr;
1160 
1161  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
1162 
1163  for (i = 0; i < max_replication_slots; i++)
1164  {
1165  ReplicationSlot *s;
1166  XLogRecPtr restart_lsn;
1167  bool invalidated;
1168 
1170 
1171  /* cannot change while ReplicationSlotCtlLock is held */
1172  if (!s->in_use)
1173  continue;
1174 
1175  /* we're only interested in logical slots */
1176  if (!SlotIsLogical(s))
1177  continue;
1178 
1179  /* read once, it's ok if it increases while we're checking */
1180  SpinLockAcquire(&s->mutex);
1181  restart_lsn = s->data.restart_lsn;
1182  invalidated = s->data.invalidated != RS_INVAL_NONE;
1183  SpinLockRelease(&s->mutex);
1184 
1185  /* invalidated slots need not apply */
1186  if (invalidated)
1187  continue;
1188 
1189  if (restart_lsn == InvalidXLogRecPtr)
1190  continue;
1191 
1192  if (result == InvalidXLogRecPtr ||
1193  restart_lsn < result)
1194  result = restart_lsn;
1195  }
1196 
1197  LWLockRelease(ReplicationSlotControlLock);
1198 
1199  return result;
1200 }

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

Referenced by CheckPointLogicalRewriteHeap(), and CheckPointSnapBuild().

◆ ReplicationSlotsComputeRequiredLSN()

void ReplicationSlotsComputeRequiredLSN ( void  )

Definition at line 1104 of file slot.c.

1105 {
1106  int i;
1107  XLogRecPtr min_required = InvalidXLogRecPtr;
1108 
1109  Assert(ReplicationSlotCtl != NULL);
1110 
1111  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
1112  for (i = 0; i < max_replication_slots; i++)
1113  {
1115  XLogRecPtr restart_lsn;
1116  bool invalidated;
1117 
1118  if (!s->in_use)
1119  continue;
1120 
1121  SpinLockAcquire(&s->mutex);
1122  restart_lsn = s->data.restart_lsn;
1123  invalidated = s->data.invalidated != RS_INVAL_NONE;
1124  SpinLockRelease(&s->mutex);
1125 
1126  /* invalidated slots need not apply */
1127  if (invalidated)
1128  continue;
1129 
1130  if (restart_lsn != InvalidXLogRecPtr &&
1131  (min_required == InvalidXLogRecPtr ||
1132  restart_lsn < min_required))
1133  min_required = restart_lsn;
1134  }
1135  LWLockRelease(ReplicationSlotControlLock);
1136 
1137  XLogSetReplicationSlotMinimumLSN(min_required);
1138 }
void XLogSetReplicationSlotMinimumLSN(XLogRecPtr lsn)
Definition: xlog.c:2614

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

Referenced by copy_replication_slot(), InvalidateObsoleteReplicationSlots(), LogicalConfirmReceivedLocation(), pg_replication_slot_advance(), PhysicalConfirmReceivedLocation(), ReplicationSlotDropPtr(), ReplicationSlotReserveWal(), reserve_wal_for_local_slot(), StartupReplicationSlots(), and update_local_synced_slot().

◆ ReplicationSlotsComputeRequiredXmin()

void ReplicationSlotsComputeRequiredXmin ( bool  already_locked)

Definition at line 1048 of file slot.c.

1049 {
1050  int i;
1052  TransactionId agg_catalog_xmin = InvalidTransactionId;
1053 
1054  Assert(ReplicationSlotCtl != NULL);
1055 
1056  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
1057 
1058  for (i = 0; i < max_replication_slots; i++)
1059  {
1061  TransactionId effective_xmin;
1062  TransactionId effective_catalog_xmin;
1063  bool invalidated;
1064 
1065  if (!s->in_use)
1066  continue;
1067 
1068  SpinLockAcquire(&s->mutex);
1069  effective_xmin = s->effective_xmin;
1070  effective_catalog_xmin = s->effective_catalog_xmin;
1071  invalidated = s->data.invalidated != RS_INVAL_NONE;
1072  SpinLockRelease(&s->mutex);
1073 
1074  /* invalidated slots need not apply */
1075  if (invalidated)
1076  continue;
1077 
1078  /* check the data xmin */
1079  if (TransactionIdIsValid(effective_xmin) &&
1080  (!TransactionIdIsValid(agg_xmin) ||
1081  TransactionIdPrecedes(effective_xmin, agg_xmin)))
1082  agg_xmin = effective_xmin;
1083 
1084  /* check the catalog xmin */
1085  if (TransactionIdIsValid(effective_catalog_xmin) &&
1086  (!TransactionIdIsValid(agg_catalog_xmin) ||
1087  TransactionIdPrecedes(effective_catalog_xmin, agg_catalog_xmin)))
1088  agg_catalog_xmin = effective_catalog_xmin;
1089  }
1090 
1091  LWLockRelease(ReplicationSlotControlLock);
1092 
1093  ProcArraySetReplicationSlotXmin(agg_xmin, agg_catalog_xmin, already_locked);
1094 }
uint32 TransactionId
Definition: c.h:639
void ProcArraySetReplicationSlotXmin(TransactionId xmin, TransactionId catalog_xmin, bool already_locked)
Definition: procarray.c:3927
bool TransactionIdPrecedes(TransactionId id1, TransactionId id2)
Definition: transam.c:280

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

Referenced by copy_replication_slot(), CreateInitDecodingContext(), InvalidateObsoleteReplicationSlots(), LogicalConfirmReceivedLocation(), pg_replication_slot_advance(), PhysicalReplicationSlotNewXmin(), ReplicationSlotDropPtr(), ReplicationSlotRelease(), StartupReplicationSlots(), synchronize_one_slot(), and update_local_synced_slot().

◆ ReplicationSlotsCountDBSlots()

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

Definition at line 1211 of file slot.c.

1212 {
1213  int i;
1214 
1215  *nslots = *nactive = 0;
1216 
1217  if (max_replication_slots <= 0)
1218  return false;
1219 
1220  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
1221  for (i = 0; i < max_replication_slots; i++)
1222  {
1223  ReplicationSlot *s;
1224 
1226 
1227  /* cannot change while ReplicationSlotCtlLock is held */
1228  if (!s->in_use)
1229  continue;
1230 
1231  /* only logical slots are database specific, skip */
1232  if (!SlotIsLogical(s))
1233  continue;
1234 
1235  /* not our database, skip */
1236  if (s->data.database != dboid)
1237  continue;
1238 
1239  /* NB: intentionally counting invalidated slots */
1240 
1241  /* count slots with spinlock held */
1242  SpinLockAcquire(&s->mutex);
1243  (*nslots)++;
1244  if (s->active_pid != 0)
1245  (*nactive)++;
1246  SpinLockRelease(&s->mutex);
1247  }
1248  LWLockRelease(ReplicationSlotControlLock);
1249 
1250  if (*nslots > 0)
1251  return true;
1252  return false;
1253 }

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

Referenced by dropdb().

◆ ReplicationSlotsDropDBSlots()

void ReplicationSlotsDropDBSlots ( Oid  dboid)

Definition at line 1269 of file slot.c.

1270 {
1271  int i;
1272 
1273  if (max_replication_slots <= 0)
1274  return;
1275 
1276 restart:
1277  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
1278  for (i = 0; i < max_replication_slots; i++)
1279  {
1280  ReplicationSlot *s;
1281  char *slotname;
1282  int active_pid;
1283 
1285 
1286  /* cannot change while ReplicationSlotCtlLock is held */
1287  if (!s->in_use)
1288  continue;
1289 
1290  /* only logical slots are database specific, skip */
1291  if (!SlotIsLogical(s))
1292  continue;
1293 
1294  /* not our database, skip */
1295  if (s->data.database != dboid)
1296  continue;
1297 
1298  /* NB: intentionally including invalidated slots */
1299 
1300  /* acquire slot, so ReplicationSlotDropAcquired can be reused */
1301  SpinLockAcquire(&s->mutex);
1302  /* can't change while ReplicationSlotControlLock is held */
1303  slotname = NameStr(s->data.name);
1304  active_pid = s->active_pid;
1305  if (active_pid == 0)
1306  {
1307  MyReplicationSlot = s;
1308  s->active_pid = MyProcPid;
1309  }
1310  SpinLockRelease(&s->mutex);
1311 
1312  /*
1313  * Even though we hold an exclusive lock on the database object a
1314  * logical slot for that DB can still be active, e.g. if it's
1315  * concurrently being dropped by a backend connected to another DB.
1316  *
1317  * That's fairly unlikely in practice, so we'll just bail out.
1318  *
1319  * The slot sync worker holds a shared lock on the database before
1320  * operating on synced logical slots to avoid conflict with the drop
1321  * happening here. The persistent synced slots are thus safe but there
1322  * is a possibility that the slot sync worker has created a temporary
1323  * slot (which stays active even on release) and we are trying to drop
1324  * that here. In practice, the chances of hitting this scenario are
1325  * less as during slot synchronization, the temporary slot is
1326  * immediately converted to persistent and thus is safe due to the
1327  * shared lock taken on the database. So, we'll just bail out in such
1328  * a case.
1329  *
1330  * XXX: We can consider shutting down the slot sync worker before
1331  * trying to drop synced temporary slots here.
1332  */
1333  if (active_pid)
1334  ereport(ERROR,
1335  (errcode(ERRCODE_OBJECT_IN_USE),
1336  errmsg("replication slot \"%s\" is active for PID %d",
1337  slotname, active_pid)));
1338 
1339  /*
1340  * To avoid duplicating ReplicationSlotDropAcquired() and to avoid
1341  * holding ReplicationSlotControlLock over filesystem operations,
1342  * release ReplicationSlotControlLock and use
1343  * ReplicationSlotDropAcquired.
1344  *
1345  * As that means the set of slots could change, restart scan from the
1346  * beginning each time we release the lock.
1347  */
1348  LWLockRelease(ReplicationSlotControlLock);
1350  goto restart;
1351  }
1352  LWLockRelease(ReplicationSlotControlLock);
1353 }

References ReplicationSlot::active_pid, ReplicationSlot::data, ReplicationSlotPersistentData::database, ereport, errcode(), errmsg(), ERROR, i, ReplicationSlot::in_use, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_replication_slots, ReplicationSlot::mutex, MyProcPid, MyReplicationSlot, ReplicationSlotPersistentData::name, NameStr, ReplicationSlotCtlData::replication_slots, ReplicationSlotCtl, ReplicationSlotDropAcquired(), SlotIsLogical, SpinLockAcquire, and SpinLockRelease.

Referenced by dbase_redo(), and dropdb().

◆ ReplicationSlotShmemExit()

static void ReplicationSlotShmemExit ( int  code,
Datum  arg 
)
static

Definition at line 233 of file slot.c.

234 {
235  /* Make sure active replication slots are released */
236  if (MyReplicationSlot != NULL)
238 
239  /* Also cleanup all the temporary slots. */
241 }
void ReplicationSlotCleanup(void)
Definition: slot.c:745

References MyReplicationSlot, ReplicationSlotCleanup(), and ReplicationSlotRelease().

Referenced by ReplicationSlotInitialize().

◆ ReplicationSlotsShmemInit()

void ReplicationSlotsShmemInit ( void  )

Definition at line 189 of file slot.c.

190 {
191  bool found;
192 
193  if (max_replication_slots == 0)
194  return;
195 
197  ShmemInitStruct("ReplicationSlot Ctl", ReplicationSlotsShmemSize(),
198  &found);
199 
200  if (!found)
201  {
202  int i;
203 
204  /* First time through, so initialize */
206 
207  for (i = 0; i < max_replication_slots; i++)
208  {
210 
211  /* everything else is zeroed by the memset above */
212  SpinLockInit(&slot->mutex);
216  }
217  }
218 }
#define MemSet(start, val, len)
Definition: c.h:1007
void ConditionVariableInit(ConditionVariable *cv)
void LWLockInitialize(LWLock *lock, int tranche_id)
Definition: lwlock.c:708
@ LWTRANCHE_REPLICATION_SLOT_IO
Definition: lwlock.h:189
void * ShmemInitStruct(const char *name, Size size, bool *foundPtr)
Definition: shmem.c:387
Size ReplicationSlotsShmemSize(void)
Definition: slot.c:171
#define SpinLockInit(lock)
Definition: spin.h:60
LWLock io_in_progress_lock
Definition: slot.h:181

References ReplicationSlot::active_cv, ConditionVariableInit(), i, ReplicationSlot::io_in_progress_lock, LWLockInitialize(), LWTRANCHE_REPLICATION_SLOT_IO, max_replication_slots, MemSet, ReplicationSlot::mutex, ReplicationSlotCtlData::replication_slots, ReplicationSlotCtl, ReplicationSlotsShmemSize(), ShmemInitStruct(), and SpinLockInit.

Referenced by CreateOrAttachShmemStructs().

◆ ReplicationSlotsShmemSize()

Size ReplicationSlotsShmemSize ( void  )

Definition at line 171 of file slot.c.

172 {
173  Size size = 0;
174 
175  if (max_replication_slots == 0)
176  return size;
177 
178  size = offsetof(ReplicationSlotCtlData, replication_slots);
179  size = add_size(size,
181 
182  return size;
183 }
size_t Size
Definition: c.h:592
Size add_size(Size s1, Size s2)
Definition: shmem.c:493
Size mul_size(Size s1, Size s2)
Definition: shmem.c:510

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

Referenced by CalculateShmemSize(), and ReplicationSlotsShmemInit().

◆ ReplicationSlotValidateName()

bool ReplicationSlotValidateName ( const char *  name,
int  elevel 
)

Definition at line 252 of file slot.c.

253 {
254  const char *cp;
255 
256  if (strlen(name) == 0)
257  {
258  ereport(elevel,
259  (errcode(ERRCODE_INVALID_NAME),
260  errmsg("replication slot name \"%s\" is too short",
261  name)));
262  return false;
263  }
264 
265  if (strlen(name) >= NAMEDATALEN)
266  {
267  ereport(elevel,
268  (errcode(ERRCODE_NAME_TOO_LONG),
269  errmsg("replication slot name \"%s\" is too long",
270  name)));
271  return false;
272  }
273 
274  for (cp = name; *cp; cp++)
275  {
276  if (!((*cp >= 'a' && *cp <= 'z')
277  || (*cp >= '0' && *cp <= '9')
278  || (*cp == '_')))
279  {
280  ereport(elevel,
281  (errcode(ERRCODE_INVALID_NAME),
282  errmsg("replication slot name \"%s\" contains invalid character",
283  name),
284  errhint("Replication slot names may only contain lower case letters, numbers, and the underscore character.")));
285  return false;
286  }
287  }
288  return true;
289 }
#define NAMEDATALEN

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

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

◆ ReportSlotInvalidation()

static void ReportSlotInvalidation ( ReplicationSlotInvalidationCause  cause,
bool  terminating,
int  pid,
NameData  slotname,
XLogRecPtr  restart_lsn,
XLogRecPtr  oldestLSN,
TransactionId  snapshotConflictHorizon 
)
static

Definition at line 1476 of file slot.c.

1483 {
1484  StringInfoData err_detail;
1485  bool hint = false;
1486 
1487  initStringInfo(&err_detail);
1488 
1489  switch (cause)
1490  {
1491  case RS_INVAL_WAL_REMOVED:
1492  {
1493  unsigned long long ex = oldestLSN - restart_lsn;
1494 
1495  hint = true;
1496  appendStringInfo(&err_detail,
1497  ngettext("The slot's restart_lsn %X/%X exceeds the limit by %llu byte.",
1498  "The slot's restart_lsn %X/%X exceeds the limit by %llu bytes.",
1499  ex),
1500  LSN_FORMAT_ARGS(restart_lsn),
1501  ex);
1502  break;
1503  }
1504  case RS_INVAL_HORIZON:
1505  appendStringInfo(&err_detail, _("The slot conflicted with xid horizon %u."),
1506  snapshotConflictHorizon);
1507  break;
1508 
1509  case RS_INVAL_WAL_LEVEL:
1510  appendStringInfoString(&err_detail, _("Logical decoding on standby requires wal_level >= logical on the primary server."));
1511  break;
1512  case RS_INVAL_NONE:
1513  pg_unreachable();
1514  }
1515 
1516  ereport(LOG,
1517  terminating ?
1518  errmsg("terminating process %d to release replication slot \"%s\"",
1519  pid, NameStr(slotname)) :
1520  errmsg("invalidating obsolete replication slot \"%s\"",
1521  NameStr(slotname)),
1522  errdetail_internal("%s", err_detail.data),
1523  hint ? errhint("You might need to increase %s.", "max_slot_wal_keep_size") : 0);
1524 
1525  pfree(err_detail.data);
1526 }
#define ngettext(s, p, n)
Definition: c.h:1168
int errdetail_internal(const char *fmt,...)
Definition: elog.c:1232
#define _(x)
Definition: elog.c:90
void appendStringInfo(StringInfo str, const char *fmt,...)
Definition: stringinfo.c:97
void appendStringInfoString(StringInfo str, const char *s)
Definition: stringinfo.c:182
void initStringInfo(StringInfo str)
Definition: stringinfo.c:59
#define LSN_FORMAT_ARGS(lsn)
Definition: xlogdefs.h:43

References _, appendStringInfo(), appendStringInfoString(), StringInfoData::data, ereport, errdetail_internal(), errhint(), errmsg(), initStringInfo(), LOG, LSN_FORMAT_ARGS, NameStr, ngettext, pfree(), pg_unreachable, RS_INVAL_HORIZON, RS_INVAL_NONE, RS_INVAL_WAL_LEVEL, and RS_INVAL_WAL_REMOVED.

Referenced by InvalidatePossiblyObsoleteSlot().

◆ RestoreSlotFromDisk()

static void RestoreSlotFromDisk ( const char *  name)
static

Definition at line 2174 of file slot.c.

2175 {
2177  int i;
2178  char slotdir[MAXPGPATH + 12];
2179  char path[MAXPGPATH + 22];
2180  int fd;
2181  bool restored = false;
2182  int readBytes;
2183  pg_crc32c checksum;
2184 
2185  /* no need to lock here, no concurrent access allowed yet */
2186 
2187  /* delete temp file if it exists */
2188  sprintf(slotdir, "pg_replslot/%s", name);
2189  sprintf(path, "%s/state.tmp", slotdir);
2190  if (unlink(path) < 0 && errno != ENOENT)
2191  ereport(PANIC,
2193  errmsg("could not remove file \"%s\": %m", path)));
2194 
2195  sprintf(path, "%s/state", slotdir);
2196 
2197  elog(DEBUG1, "restoring replication slot from \"%s\"", path);
2198 
2199  /* on some operating systems fsyncing a file requires O_RDWR */
2200  fd = OpenTransientFile(path, O_RDWR | PG_BINARY);
2201 
2202  /*
2203  * We do not need to handle this as we are rename()ing the directory into
2204  * place only after we fsync()ed the state file.
2205  */
2206  if (fd < 0)
2207  ereport(PANIC,
2209  errmsg("could not open file \"%s\": %m", path)));
2210 
2211  /*
2212  * Sync state file before we're reading from it. We might have crashed
2213  * while it wasn't synced yet and we shouldn't continue on that basis.
2214  */
2215  pgstat_report_wait_start(WAIT_EVENT_REPLICATION_SLOT_RESTORE_SYNC);
2216  if (pg_fsync(fd) != 0)
2217  ereport(PANIC,
2219  errmsg("could not fsync file \"%s\": %m",
2220  path)));
2222 
2223  /* Also sync the parent directory */
2225  fsync_fname(slotdir, true);
2226  END_CRIT_SECTION();
2227 
2228  /* read part of statefile that's guaranteed to be version independent */
2229  pgstat_report_wait_start(WAIT_EVENT_REPLICATION_SLOT_READ);
2230  readBytes = read(fd, &cp, ReplicationSlotOnDiskConstantSize);
2232  if (readBytes != ReplicationSlotOnDiskConstantSize)
2233  {
2234  if (readBytes < 0)
2235  ereport(PANIC,
2237  errmsg("could not read file \"%s\": %m", path)));
2238  else
2239  ereport(PANIC,
2241  errmsg("could not read file \"%s\": read %d of %zu",
2242  path, readBytes,
2244  }
2245 
2246  /* verify magic */
2247  if (cp.magic != SLOT_MAGIC)
2248  ereport(PANIC,
2250  errmsg("replication slot file \"%s\" has wrong magic number: %u instead of %u",
2251  path, cp.magic, SLOT_MAGIC)));
2252 
2253  /* verify version */
2254  if (cp.version != SLOT_VERSION)
2255  ereport(PANIC,
2257  errmsg("replication slot file \"%s\" has unsupported version %u",
2258  path, cp.version)));
2259 
2260  /* boundary check on length */
2262  ereport(PANIC,
2264  errmsg("replication slot file \"%s\" has corrupted length %u",
2265  path, cp.length)));
2266 
2267  /* Now that we know the size, read the entire file */
2268  pgstat_report_wait_start(WAIT_EVENT_REPLICATION_SLOT_READ);
2269  readBytes = read(fd,
2270  (char *) &cp + ReplicationSlotOnDiskConstantSize,
2271  cp.length);
2273  if (readBytes != cp.length)
2274  {
2275  if (readBytes < 0)
2276  ereport(PANIC,
2278  errmsg("could not read file \"%s\": %m", path)));
2279  else
2280  ereport(PANIC,
2282  errmsg("could not read file \"%s\": read %d of %zu",
2283  path, readBytes, (Size) cp.length)));
2284  }
2285 
2286  if (CloseTransientFile(fd) != 0)
2287  ereport(PANIC,
2289  errmsg("could not close file \"%s\": %m", path)));
2290 
2291  /* now verify the CRC */
2292  INIT_CRC32C(checksum);
2293  COMP_CRC32C(checksum,
2296  FIN_CRC32C(checksum);
2297 
2298  if (!EQ_CRC32C(checksum, cp.checksum))
2299  ereport(PANIC,
2300  (errmsg("checksum mismatch for replication slot file \"%s\": is %u, should be %u",
2301  path, checksum, cp.checksum)));
2302 
2303  /*
2304  * If we crashed with an ephemeral slot active, don't restore but delete
2305  * it.
2306  */
2308  {
2309  if (!rmtree(slotdir, true))
2310  {
2311  ereport(WARNING,
2312  (errmsg("could not remove directory \"%s\"",
2313  slotdir)));
2314  }
2315  fsync_fname("pg_replslot", true);
2316  return;
2317  }
2318 
2319  /*
2320  * Verify that requirements for the specific slot type are met. That's
2321  * important because if these aren't met we're not guaranteed to retain
2322  * all the necessary resources for the slot.
2323  *
2324  * NB: We have to do so *after* the above checks for ephemeral slots,
2325  * because otherwise a slot that shouldn't exist anymore could prevent
2326  * restarts.
2327  *
2328  * NB: Changing the requirements here also requires adapting
2329  * CheckSlotRequirements() and CheckLogicalDecodingRequirements().
2330  */
2332  ereport(FATAL,
2333  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
2334  errmsg("logical replication slot \"%s\" exists, but wal_level < logical",
2335  NameStr(cp.slotdata.name)),
2336  errhint("Change wal_level to be logical or higher.")));
2337  else if (wal_level < WAL_LEVEL_REPLICA)
2338  ereport(FATAL,
2339  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
2340  errmsg("physical replication slot \"%s\" exists, but wal_level < replica",
2341  NameStr(cp.slotdata.name)),
2342  errhint("Change wal_level to be replica or higher.")));
2343 
2344  /* nothing can be active yet, don't lock anything */
2345  for (i = 0; i < max_replication_slots; i++)
2346  {
2347  ReplicationSlot *slot;
2348 
2350 
2351  if (slot->in_use)
2352  continue;
2353 
2354  /* restore the entire set of persistent data */
2355  memcpy(&slot->data, &cp.slotdata,
2357 
2358  /* initialize in memory state */
2359  slot->effective_xmin = cp.slotdata.xmin;
2362 
2367 
2368  slot->in_use = true;
2369  slot->active_pid = 0;
2370 
2371  /*
2372  * We set the last inactive time after loading the slot from the disk
2373  * into memory. Whoever acquires the slot i.e. makes the slot active
2374  * will reset it. We don't set it for the slots currently being synced
2375  * from the primary to the standby because such slots are typically
2376  * inactive as decoding is not allowed on those.
2377  */
2378  if (!(RecoveryInProgress() && slot->data.synced))
2380  else
2381  slot->inactive_since = 0;
2382 
2383  restored = true;
2384  break;
2385  }
2386 
2387  if (!restored)
2388  ereport(FATAL,
2389  (errmsg("too many replication slots active before shutdown"),
2390  errhint("Increase max_replication_slots and try again.")));
2391 }
#define PG_BINARY
Definition: c.h:1260
#define FATAL
Definition: elog.h:41
#define PANIC
Definition: elog.h:42
int CloseTransientFile(int fd)
Definition: fd.c:2809
int pg_fsync(int fd)
Definition: fd.c:386
int OpenTransientFile(const char *fileName, int fileFlags)
Definition: fd.c:2633
#define read(a, b, c)
Definition: win32.h:13
#define ERRCODE_DATA_CORRUPTED
Definition: pg_basebackup.c:41
uint32 pg_crc32c
Definition: pg_crc32c.h:38
#define COMP_CRC32C(crc, data, len)
Definition: pg_crc32c.h:98
#define EQ_CRC32C(c1, c2)
Definition: pg_crc32c.h:42
#define INIT_CRC32C(crc)
Definition: pg_crc32c.h:41
#define FIN_CRC32C(crc)
Definition: pg_crc32c.h:103
static int fd(const char *x, int i)
Definition: preproc-init.c:105
#define ReplicationSlotOnDiskChecksummedSize
Definition: slot.c:125
#define ReplicationSlotOnDiskNotChecksummedSize
Definition: slot.c:122
#define ReplicationSlotOnDiskV2Size
Definition: slot.c:128
#define SLOT_VERSION
Definition: slot.c:132
#define SLOT_MAGIC
Definition: slot.c:131
#define ReplicationSlotOnDiskConstantSize
Definition: slot.c:119
uint32 version
Definition: slot.c:73
ReplicationSlotPersistentData slotdata
Definition: slot.c:81
pg_crc32c checksum
Definition: slot.c:70
TransactionId catalog_xmin
Definition: slot.h:90
static void pgstat_report_wait_start(uint32 wait_event_info)
Definition: wait_event.h:88
static void pgstat_report_wait_end(void)
Definition: wait_event.h:104
@ WAL_LEVEL_LOGICAL
Definition: xlog.h:74

References ReplicationSlot::active_pid, ReplicationSlot::candidate_catalog_xmin, ReplicationSlot::candidate_restart_lsn, ReplicationSlot::candidate_restart_valid, ReplicationSlot::candidate_xmin_lsn, ReplicationSlotPersistentData::catalog_xmin, ReplicationSlotOnDisk::checksum, CloseTransientFile(), COMP_CRC32C, ReplicationSlotPersistentData::confirmed_flush, ReplicationSlot::data, ReplicationSlotPersistentData::database, DEBUG1, ReplicationSlot::effective_catalog_xmin, ReplicationSlot::effective_xmin, elog, END_CRIT_SECTION, EQ_CRC32C, ereport, errcode(), ERRCODE_DATA_CORRUPTED, errcode_for_file_access(), errhint(), errmsg(), FATAL, fd(), FIN_CRC32C, fsync_fname(), GetCurrentTimestamp(), i, ReplicationSlot::in_use, ReplicationSlot::inactive_since, INIT_CRC32C, InvalidOid, InvalidTransactionId, InvalidXLogRecPtr, ReplicationSlot::last_saved_confirmed_flush, ReplicationSlotOnDisk::length, ReplicationSlotOnDisk::magic, max_replication_slots, MAXPGPATH, name, ReplicationSlotPersistentData::name, NameStr, OpenTransientFile(), PANIC, ReplicationSlotPersistentData::persistency, PG_BINARY, pg_fsync(), pgstat_report_wait_end(), pgstat_report_wait_start(), read, RecoveryInProgress(), ReplicationSlotCtlData::replication_slots, ReplicationSlotCtl, ReplicationSlotOnDiskChecksummedSize, ReplicationSlotOnDiskConstantSize, ReplicationSlotOnDiskNotChecksummedSize, ReplicationSlotOnDiskV2Size, rmtree(), RS_PERSISTENT, SLOT_MAGIC, SLOT_VERSION, ReplicationSlotOnDisk::slotdata, sprintf, START_CRIT_SECTION, ReplicationSlotPersistentData::synced, ReplicationSlotOnDisk::version, wal_level, WAL_LEVEL_LOGICAL, WAL_LEVEL_REPLICA, WARNING, and ReplicationSlotPersistentData::xmin.

Referenced by StartupReplicationSlots().

◆ SaveSlotToPath()

static void SaveSlotToPath ( ReplicationSlot slot,
const char *  dir,
int  elevel 
)
static

Definition at line 2019 of file slot.c.

2020 {
2021  char tmppath[MAXPGPATH];
2022  char path[MAXPGPATH];
2023  int fd;
2025  bool was_dirty;
2026 
2027  /* first check whether there's something to write out */
2028  SpinLockAcquire(&slot->mutex);
2029  was_dirty = slot->dirty;
2030  slot->just_dirtied = false;
2031  SpinLockRelease(&slot->mutex);
2032 
2033  /* and don't do anything if there's nothing to write */
2034  if (!was_dirty)
2035  return;
2036 
2038 
2039  /* silence valgrind :( */
2040  memset(&cp, 0, sizeof(ReplicationSlotOnDisk));
2041 
2042  sprintf(tmppath, "%s/state.tmp", dir);
2043  sprintf(path, "%s/state", dir);
2044 
2045  fd = OpenTransientFile(tmppath, O_CREAT | O_EXCL | O_WRONLY | PG_BINARY);
2046  if (fd < 0)
2047  {
2048  /*
2049  * If not an ERROR, then release the lock before returning. In case
2050  * of an ERROR, the error recovery path automatically releases the
2051  * lock, but no harm in explicitly releasing even in that case. Note
2052  * that LWLockRelease() could affect errno.
2053  */
2054  int save_errno = errno;
2055 
2057  errno = save_errno;
2058  ereport(elevel,
2060  errmsg("could not create file \"%s\": %m",
2061  tmppath)));
2062  return;
2063  }
2064 
2065  cp.magic = SLOT_MAGIC;
2066  INIT_CRC32C(cp.checksum);
2067  cp.version = SLOT_VERSION;
2069 
2070  SpinLockAcquire(&slot->mutex);
2071 
2072  memcpy(&cp.slotdata, &slot->data, sizeof(ReplicationSlotPersistentData));
2073 
2074  SpinLockRelease(&slot->mutex);
2075 
2076  COMP_CRC32C(cp.checksum,
2077  (char *) (&cp) + ReplicationSlotOnDiskNotChecksummedSize,
2079  FIN_CRC32C(cp.checksum);
2080 
2081  errno = 0;
2082  pgstat_report_wait_start(WAIT_EVENT_REPLICATION_SLOT_WRITE);
2083  if ((write(fd, &cp, sizeof(cp))) != sizeof(cp))
2084  {
2085  int save_errno = errno;
2086 
2090 
2091  /* if write didn't set errno, assume problem is no disk space */
2092  errno = save_errno ? save_errno : ENOSPC;
2093  ereport(elevel,
2095  errmsg("could not write to file \"%s\": %m",
2096  tmppath)));
2097  return;
2098  }
2100 
2101  /* fsync the temporary file */
2102  pgstat_report_wait_start(WAIT_EVENT_REPLICATION_SLOT_SYNC);
2103  if (pg_fsync(fd) != 0)
2104  {
2105  int save_errno = errno;
2106 
2110  errno = save_errno;
2111  ereport(elevel,
2113  errmsg("could not fsync file \"%s\": %m",
2114  tmppath)));
2115  return;
2116  }
2118 
2119  if (CloseTransientFile(fd) != 0)
2120  {
2121  int save_errno = errno;
2122 
2124  errno = save_errno;
2125  ereport(elevel,
2127  errmsg("could not close file \"%s\": %m",
2128  tmppath)));
2129  return;
2130  }
2131 
2132  /* rename to permanent file, fsync file and directory */
2133  if (rename(tmppath, path) != 0)
2134  {
2135  int save_errno = errno;
2136 
2138  errno = save_errno;
2139  ereport(elevel,
2141  errmsg("could not rename file \"%s\" to \"%s\": %m",
2142  tmppath, path)));
2143  return;
2144  }
2145 
2146  /*
2147  * Check CreateSlotOnDisk() for the reasoning of using a critical section.
2148  */
2150 
2151  fsync_fname(path, false);
2152  fsync_fname(dir, true);
2153  fsync_fname("pg_replslot", true);
2154 
2155  END_CRIT_SECTION();
2156 
2157  /*
2158  * Successfully wrote, unset dirty bit, unless somebody dirtied again
2159  * already and remember the confirmed_flush LSN value.
2160  */
2161  SpinLockAcquire(&slot->mutex);
2162  if (!slot->just_dirtied)
2163  slot->dirty = false;
2165  SpinLockRelease(&slot->mutex);
2166 
2168 }
#define write(a, b, c)
Definition: win32.h:14

References ReplicationSlotOnDisk::checksum, CloseTransientFile(), COMP_CRC32C, ReplicationSlotPersistentData::confirmed_flush, ReplicationSlot::data, ReplicationSlot::dirty, END_CRIT_SECTION, ereport, errcode_for_file_access(), errmsg(), fd(), FIN_CRC32C, fsync_fname(), INIT_CRC32C, ReplicationSlot::io_in_progress_lock, ReplicationSlot::just_dirtied, ReplicationSlot::last_saved_confirmed_flush, ReplicationSlotOnDisk::length, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), ReplicationSlotOnDisk::magic, MAXPGPATH, ReplicationSlot::mutex, OpenTransientFile(), PG_BINARY, pg_fsync(), pgstat_report_wait_end(), pgstat_report_wait_start(), ReplicationSlotOnDiskChecksummedSize, ReplicationSlotOnDiskNotChecksummedSize, ReplicationSlotOnDiskV2Size, SLOT_MAGIC, SLOT_VERSION, ReplicationSlotOnDisk::slotdata, SpinLockAcquire, SpinLockRelease, sprintf, START_CRIT_SECTION, ReplicationSlotOnDisk::version, and write.

Referenced by CheckPointReplicationSlots(), CreateSlotOnDisk(), and ReplicationSlotSave().

◆ SearchNamedReplicationSlot()

ReplicationSlot* SearchNamedReplicationSlot ( const char *  name,
bool  need_lock 
)

Definition at line 464 of file slot.c.

465 {
466  int i;
467  ReplicationSlot *slot = NULL;
468 
469  if (need_lock)
470  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
471 
472  for (i = 0; i < max_replication_slots; i++)
473  {
475 
476  if (s->in_use && strcmp(name, NameStr(s->data.name)) == 0)
477  {
478  slot = s;
479  break;
480  }
481  }
482 
483  if (need_lock)
484  LWLockRelease(ReplicationSlotControlLock);
485 
486  return slot;
487 }

References ReplicationSlot::data, i, ReplicationSlot::in_use, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_replication_slots, name, ReplicationSlotPersistentData::name, NameStr, ReplicationSlotCtlData::replication_slots, and ReplicationSlotCtl.

Referenced by get_replslot_index(), pg_ls_replslotdir(), pgstat_reset_replslot(), ReadReplicationSlot(), ReplicationSlotAcquire(), StandbySlotsHaveCaughtup(), synchronize_one_slot(), and validate_standby_slots().

◆ SlotExistsInStandbySlotNames()

bool SlotExistsInStandbySlotNames ( const char *  slot_name)

Definition at line 2553 of file slot.c.

2554 {
2555  const char *standby_slot_name;
2556 
2557  /* Return false if there is no value in standby_slot_names */
2558  if (standby_slot_names_config == NULL)
2559  return false;
2560 
2561  /*
2562  * XXX: We are not expecting this list to be long so a linear search
2563  * shouldn't hurt but if that turns out not to be true then we can cache
2564  * this information for each WalSender as well.
2565  */
2566  standby_slot_name = standby_slot_names_config->slot_names;
2567  for (int i = 0; i < standby_slot_names_config->nslotnames; i++)
2568  {
2569  if (strcmp(standby_slot_name, slot_name) == 0)
2570  return true;
2571 
2572  standby_slot_name += strlen(standby_slot_name) + 1;
2573  }
2574 
2575  return false;
2576 }

References i, StandbySlotNamesConfigData::nslotnames, StandbySlotNamesConfigData::slot_names, and standby_slot_names_config.

Referenced by PhysicalWakeupLogicalWalSnd().

◆ StandbySlotsHaveCaughtup()

bool StandbySlotsHaveCaughtup ( XLogRecPtr  wait_for_lsn,
int  elevel 
)

Definition at line 2586 of file slot.c.

2587 {
2588  const char *name;
2589  int caught_up_slot_num = 0;
2590  XLogRecPtr min_restart_lsn = InvalidXLogRecPtr;
2591 
2592  /*
2593  * Don't need to wait for the standbys to catch up if there is no value in
2594  * standby_slot_names.
2595  */
2596  if (standby_slot_names_config == NULL)
2597  return true;
2598 
2599  /*
2600  * Don't need to wait for the standbys to catch up if we are on a standby
2601  * server, since we do not support syncing slots to cascading standbys.
2602  */
2603  if (RecoveryInProgress())
2604  return true;
2605 
2606  /*
2607  * Don't need to wait for the standbys to catch up if they are already
2608  * beyond the specified WAL location.
2609  */
2611  ss_oldest_flush_lsn >= wait_for_lsn)
2612  return true;
2613 
2614  /*
2615  * To prevent concurrent slot dropping and creation while filtering the
2616  * slots, take the ReplicationSlotControlLock outside of the loop.
2617  */
2618  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
2619 
2621  for (int i = 0; i < standby_slot_names_config->nslotnames; i++)
2622  {
2623  XLogRecPtr restart_lsn;
2624  bool invalidated;
2625  bool inactive;
2626  ReplicationSlot *slot;
2627 
2628  slot = SearchNamedReplicationSlot(name, false);
2629 
2630  if (!slot)
2631  {
2632  /*
2633  * If a slot name provided in standby_slot_names does not exist,
2634  * report a message and exit the loop. A user can specify a slot
2635  * name that does not exist just before the server startup. The
2636  * GUC check_hook(validate_standby_slots) cannot validate such a
2637  * slot during startup as the ReplicationSlotCtl shared memory is
2638  * not initialized at that time. It is also possible for a user to
2639  * drop the slot in standby_slot_names afterwards.
2640  */
2641  ereport(elevel,
2642  errcode(ERRCODE_INVALID_PARAMETER_VALUE),
2643  errmsg("replication slot \"%s\" specified in parameter %s does not exist",
2644  name, "standby_slot_names"),
2645  errdetail("Logical replication is waiting on the standby associated with \"%s\".",
2646  name),
2647  errhint("Consider creating the slot \"%s\" or amend parameter %s.",
2648  name, "standby_slot_names"));
2649  break;
2650  }
2651 
2652  if (SlotIsLogical(slot))
2653  {
2654  /*
2655  * If a logical slot name is provided in standby_slot_names,
2656  * report a message and exit the loop. Similar to the non-existent
2657  * case, a user can specify a logical slot name in
2658  * standby_slot_names before the server startup, or drop an
2659  * existing physical slot and recreate a logical slot with the
2660  * same name.
2661  */
2662  ereport(elevel,
2663  errcode(ERRCODE_INVALID_PARAMETER_VALUE),
2664  errmsg("cannot have logical replication slot \"%s\" in parameter %s",
2665  name, "standby_slot_names"),
2666  errdetail("Logical replication is waiting for correction on \"%s\".",
2667  name),
2668  errhint("Consider removing logical slot \"%s\" from parameter %s.",
2669  name, "standby_slot_names"));
2670  break;
2671  }
2672 
2673  SpinLockAcquire(&slot->mutex);
2674  restart_lsn = slot->data.restart_lsn;
2675  invalidated = slot->data.invalidated != RS_INVAL_NONE;
2676  inactive = slot->active_pid == 0;
2677  SpinLockRelease(&slot->mutex);
2678 
2679  if (invalidated)
2680  {
2681  /* Specified physical slot has been invalidated */
2682  ereport(elevel,
2683  errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
2684  errmsg("physical slot \"%s\" specified in parameter %s has been invalidated",
2685  name, "standby_slot_names"),
2686  errdetail("Logical replication is waiting on the standby associated with \"%s\".",
2687  name),
2688  errhint("Consider dropping and recreating the slot \"%s\" or amend parameter %s.",
2689  name, "standby_slot_names"));
2690  break;
2691  }
2692 
2693  if (XLogRecPtrIsInvalid(restart_lsn) || restart_lsn < wait_for_lsn)
2694  {
2695  /* Log a message if no active_pid for this physical slot */
2696  if (inactive)
2697  ereport(elevel,
2698  errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
2699  errmsg("replication slot \"%s\" specified in parameter %s does not have active_pid",
2700  name, "standby_slot_names"),
2701  errdetail("Logical replication is waiting on the standby associated with \"%s\".",
2702  name),
2703  errhint("Consider starting standby associated with \"%s\" or amend parameter %s.",
2704  name, "standby_slot_names"));
2705 
2706  /* Continue if the current slot hasn't caught up. */
2707  break;
2708  }
2709 
2710  Assert(restart_lsn >= wait_for_lsn);
2711 
2712  if (XLogRecPtrIsInvalid(min_restart_lsn) ||
2713  min_restart_lsn > restart_lsn)
2714  min_restart_lsn = restart_lsn;
2715 
2716  caught_up_slot_num++;
2717 
2718  name += strlen(name) + 1;
2719  }
2720 
2721  LWLockRelease(ReplicationSlotControlLock);
2722 
2723  /*
2724  * Return false if not all the standbys have caught up to the specified
2725  * WAL location.
2726  */
2727  if (caught_up_slot_num != standby_slot_names_config->nslotnames)
2728  return false;
2729 
2730  /* The ss_oldest_flush_lsn must not retreat. */
2732  min_restart_lsn >= ss_oldest_flush_lsn);
2733 
2734  ss_oldest_flush_lsn = min_restart_lsn;
2735 
2736  return true;
2737 }
#define XLogRecPtrIsInvalid(r)
Definition: xlogdefs.h:29

References ReplicationSlot::active_pid, Assert(), ReplicationSlot::data, ereport, errcode(), errdetail(), errhint(), errmsg(), i, ReplicationSlotPersistentData::invalidated, InvalidXLogRecPtr, LW_SHARED, LWLockAcquire(), LWLockRelease(), ReplicationSlot::mutex, name, StandbySlotNamesConfigData::nslotnames, RecoveryInProgress(), ReplicationSlotPersistentData::restart_lsn, RS_INVAL_NONE, SearchNamedReplicationSlot(), StandbySlotNamesConfigData::slot_names, SlotIsLogical, SpinLockAcquire, SpinLockRelease, ss_oldest_flush_lsn, standby_slot_names_config, and XLogRecPtrIsInvalid.

Referenced by NeedToWaitForStandbys(), and WaitForStandbyConfirmation().

◆ StartupReplicationSlots()

void StartupReplicationSlots ( void  )

Definition at line 1897 of file slot.c.

1898 {
1899  DIR *replication_dir;
1900  struct dirent *replication_de;
1901 
1902  elog(DEBUG1, "starting up replication slots");
1903 
1904  /* restore all slots by iterating over all on-disk entries */
1905  replication_dir = AllocateDir("pg_replslot");
1906  while ((replication_de = ReadDir(replication_dir, "pg_replslot")) != NULL)
1907  {
1908  char path[MAXPGPATH + 12];
1909  PGFileType de_type;
1910 
1911  if (strcmp(replication_de->d_name, ".") == 0 ||
1912  strcmp(replication_de->d_name, "..") == 0)
1913  continue;
1914 
1915  snprintf(path, sizeof(path), "pg_replslot/%s", replication_de->d_name);
1916  de_type = get_dirent_type(path, replication_de, false, DEBUG1);
1917 
1918  /* we're only creating directories here, skip if it's not our's */
1919  if (de_type != PGFILETYPE_ERROR && de_type != PGFILETYPE_DIR)
1920  continue;
1921 
1922  /* we crashed while a slot was being setup or deleted, clean up */
1923  if (pg_str_endswith(replication_de->d_name, ".tmp"))
1924  {
1925  if (!rmtree(path, true))
1926  {
1927  ereport(WARNING,
1928  (errmsg("could not remove directory \"%s\"",
1929  path)));
1930  continue;
1931  }
1932  fsync_fname("pg_replslot", true);
1933  continue;
1934  }
1935 
1936  /* looks like a slot in a normal state, restore */
1937  RestoreSlotFromDisk(replication_de->d_name);
1938  }
1939  FreeDir(replication_dir);
1940 
1941  /* currently no slots exist, we're done. */
1942  if (max_replication_slots <= 0)
1943  return;
1944 
1945  /* Now that we have recovered all the data, compute replication xmin */
1948 }
struct dirent * ReadDir(DIR *dir, const char *dirname)
Definition: fd.c:2909
int FreeDir(DIR *dir)
Definition: fd.c:2961
DIR * AllocateDir(const char *dirname)
Definition: fd.c:2843
PGFileType get_dirent_type(const char *path, const struct dirent *de, bool look_through_symlinks, int elevel)
Definition: file_utils.c:525
PGFileType
Definition: file_utils.h:19
@ PGFILETYPE_DIR
Definition: file_utils.h:23
@ PGFILETYPE_ERROR
Definition: file_utils.h:20
#define snprintf
Definition: port.h:238
static void RestoreSlotFromDisk(const char *name)
Definition: slot.c:2174
bool pg_str_endswith(const char *str, const char *end)
Definition: string.c:32
Definition: dirent.c:26
Definition: dirent.h:10
char d_name[MAX_PATH]
Definition: dirent.h:15

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

Referenced by StartupXLOG().

◆ StaticAssertDecl()

StaticAssertDecl ( lengthof(SlotInvalidationCauses = =(RS_INVAL_MAX_CAUSES+1),
"array length mismatch"   
)

◆ validate_standby_slots()

static bool validate_standby_slots ( char *  rawname,
List **  elemlist 
)
static

Definition at line 2426 of file slot.c.

2427 {
2428  bool ok;
2429 
2430  /* Verify syntax and parse string into a list of identifiers */
2431  ok = SplitIdentifierString(rawname, ',', elemlist);
2432 
2433  if (!ok)
2434  {
2435  GUC_check_errdetail("List syntax is invalid.");
2436  }
2437  else if (!ReplicationSlotCtl)
2438  {
2439  /*
2440  * We cannot validate the replication slot if the replication slots'
2441  * data has not been initialized. This is ok as we will anyway
2442  * validate the specified slot when waiting for them to catch up. See
2443  * StandbySlotsHaveCaughtup() for details.
2444  */
2445  }
2446  else
2447  {
2448  /* Check that the specified slots exist and are logical slots */
2449  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
2450 
2451  foreach_ptr(char, name, *elemlist)
2452  {
2453  ReplicationSlot *slot;
2454 
2455  slot = SearchNamedReplicationSlot(name, false);
2456 
2457  if (!slot)
2458  {
2459  GUC_check_errdetail("replication slot \"%s\" does not exist",
2460  name);
2461  ok = false;
2462  break;
2463  }
2464 
2465  if (!SlotIsPhysical(slot))
2466  {
2467  GUC_check_errdetail("\"%s\" is not a physical replication slot",
2468  name);
2469  ok = false;
2470  break;
2471  }
2472  }
2473 
2474  LWLockRelease(ReplicationSlotControlLock);
2475  }
2476 
2477  return ok;
2478 }
#define GUC_check_errdetail
Definition: guc.h:447
bool SplitIdentifierString(char *rawstring, char separator, List **namelist)
Definition: varlena.c:3457

References foreach_ptr, GUC_check_errdetail, LW_SHARED, LWLockAcquire(), LWLockRelease(), name, ReplicationSlotCtl, SearchNamedReplicationSlot(), SlotIsPhysical, and SplitIdentifierString().

Referenced by check_standby_slot_names().

◆ WaitForStandbyConfirmation()

void WaitForStandbyConfirmation ( XLogRecPtr  wait_for_lsn)

Definition at line 2746 of file slot.c.

2747 {
2748  /*
2749  * Don't need to wait for the standby to catch up if the current acquired
2750  * slot is not a logical failover slot, or there is no value in
2751  * standby_slot_names.
2752  */
2754  return;
2755 
2757 
2758  for (;;)
2759  {
2761 
2762  if (ConfigReloadPending)
2763  {
2764  ConfigReloadPending = false;
2766  }
2767 
2768  /* Exit if done waiting for every slot. */
2769  if (StandbySlotsHaveCaughtup(wait_for_lsn, WARNING))
2770  break;
2771 
2772  /*
2773  * Wait for the slots in the standby_slot_names to catch up, but use a
2774  * timeout (1s) so we can also check if the standby_slot_names has
2775  * been changed.
2776  */
2778  WAIT_EVENT_WAIT_FOR_STANDBY_CONFIRMATION);
2779  }
2780 
2782 }
bool ConditionVariableTimedSleep(ConditionVariable *cv, long timeout, uint32 wait_event_info)
@ PGC_SIGHUP
Definition: guc.h:71
void ProcessConfigFile(GucContext context)
volatile sig_atomic_t ConfigReloadPending
Definition: interrupt.c:27
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:122
bool StandbySlotsHaveCaughtup(XLogRecPtr wait_for_lsn, int elevel)
Definition: slot.c:2586
ConditionVariable wal_confirm_rcv_cv
WalSndCtlData * WalSndCtl
Definition: walsender.c:109

References CHECK_FOR_INTERRUPTS, ConditionVariableCancelSleep(), ConditionVariablePrepareToSleep(), ConditionVariableTimedSleep(), ConfigReloadPending, ReplicationSlot::data, ReplicationSlotPersistentData::failover, MyReplicationSlot, PGC_SIGHUP, ProcessConfigFile(), standby_slot_names_config, StandbySlotsHaveCaughtup(), WalSndCtlData::wal_confirm_rcv_cv, WalSndCtl, and WARNING.

Referenced by pg_logical_replication_slot_advance(), and pg_logical_slot_get_changes_guts().

Variable Documentation

◆ max_replication_slots

◆ MyReplicationSlot

ReplicationSlot* MyReplicationSlot = NULL

Definition at line 138 of file slot.c.

Referenced by binary_upgrade_logical_slot_has_caught_up(), copy_replication_slot(), create_logical_replication_slot(), create_physical_replication_slot(), CreateDecodingContext(), CreateInitDecodingContext(), CreateReplicationSlot(), InvalidatePossiblyObsoleteSlot(), LogicalConfirmReceivedLocation(), LogicalIncreaseRestartDecodingForSlot(), LogicalIncreaseXminForSlot(), LogicalReplicationSlotHasPendingWal(), NeedToWaitForStandbys(), pg_create_logical_replication_slot(), pg_create_physical_replication_slot(), pg_logical_replication_slot_advance(), pg_logical_slot_get_changes_guts(), pg_physical_replication_slot_advance(), pg_replication_slot_advance(), PhysicalConfirmReceivedLocation(), PhysicalReplicationSlotNewXmin(), PhysicalWakeupLogicalWalSnd(), PostgresMain(), ProcessStandbyHSFeedbackMessage(), ProcessStandbyReplyMessage(), ReorderBufferAllocate(), ReorderBufferFree(), ReorderBufferRestoreChanges(), ReorderBufferRestoreCleanup(), ReorderBufferSerializedPath(), ReorderBufferSerializeTXN(), ReplicationSlotAcquire(), ReplicationSlotAlter(), ReplicationSlotCleanup(), ReplicationSlotCreate(), ReplicationSlotDrop(), ReplicationSlotDropAcquired(), ReplicationSlotMarkDirty(), ReplicationSlotPersist(), ReplicationSlotRelease(), ReplicationSlotReserveWal(), ReplicationSlotSave(), ReplicationSlotsDropDBSlots(), ReplicationSlotShmemExit(), reserve_wal_for_local_slot(), StartLogicalReplication(), StartReplication(), StartupDecodingContext(), synchronize_one_slot(), update_and_persist_local_synced_slot(), update_local_synced_slot(), WaitForStandbyConfirmation(), and WalSndErrorCleanup().

◆ ReplicationSlotCtl

◆ SlotInvalidationCauses

const char* const SlotInvalidationCauses[]
Initial value:
= {
[RS_INVAL_NONE] = "none",
[RS_INVAL_WAL_REMOVED] = "wal_removed",
[RS_INVAL_HORIZON] = "rows_removed",
[RS_INVAL_WAL_LEVEL] = "wal_level_insufficient",
}

Definition at line 105 of file slot.c.

Referenced by GetSlotInvalidationCause(), and pg_get_replication_slots().

◆ ss_oldest_flush_lsn

XLogRecPtr ss_oldest_flush_lsn = InvalidXLogRecPtr
static

Definition at line 157 of file slot.c.

Referenced by assign_standby_slot_names(), and StandbySlotsHaveCaughtup().

◆ standby_slot_names

char* standby_slot_names

Definition at line 148 of file slot.c.

◆ standby_slot_names_config