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 "replication/slot.h"
#include "storage/fd.h"
#include "storage/ipc.h"
#include "storage/proc.h"
#include "storage/procarray.h"
#include "utils/builtins.h"
Include dependency graph for slot.c:

Go to the source code of this file.

Data Structures

struct  ReplicationSlotOnDisk
 

Macros

#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   3 /* version for new files */
 

Typedefs

typedef struct ReplicationSlotOnDisk ReplicationSlotOnDisk
 

Functions

static void ReplicationSlotShmemExit (int code, Datum arg)
 
static void ReplicationSlotDropAcquired (void)
 
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)
 
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 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)
 

Variables

ReplicationSlotCtlDataReplicationSlotCtl = NULL
 
ReplicationSlotMyReplicationSlot = NULL
 
int max_replication_slots = 10
 

Macro Definition Documentation

◆ ReplicationSlotOnDiskChecksummedSize

#define ReplicationSlotOnDiskChecksummedSize    sizeof(ReplicationSlotOnDisk) - ReplicationSlotOnDiskNotChecksummedSize

Definition at line 86 of file slot.c.

◆ ReplicationSlotOnDiskConstantSize

#define ReplicationSlotOnDiskConstantSize    offsetof(ReplicationSlotOnDisk, slotdata)

Definition at line 80 of file slot.c.

◆ ReplicationSlotOnDiskNotChecksummedSize

#define ReplicationSlotOnDiskNotChecksummedSize    offsetof(ReplicationSlotOnDisk, version)

Definition at line 83 of file slot.c.

◆ ReplicationSlotOnDiskV2Size

#define ReplicationSlotOnDiskV2Size    sizeof(ReplicationSlotOnDisk) - ReplicationSlotOnDiskConstantSize

Definition at line 89 of file slot.c.

◆ SLOT_MAGIC

#define SLOT_MAGIC   0x1051CA1 /* format identifier */

Definition at line 92 of file slot.c.

◆ SLOT_VERSION

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

Definition at line 93 of file slot.c.

Typedef Documentation

◆ ReplicationSlotOnDisk

Function Documentation

◆ CheckPointReplicationSlots()

void CheckPointReplicationSlots ( bool  is_shutdown)

Definition at line 1620 of file slot.c.

1621 {
1622  int i;
1623 
1624  elog(DEBUG1, "performing replication slot checkpoint");
1625 
1626  /*
1627  * Prevent any slot from being created/dropped while we're active. As we
1628  * explicitly do *not* want to block iterating over replication_slots or
1629  * acquiring a slot we cannot take the control lock - but that's OK,
1630  * because holding ReplicationSlotAllocationLock is strictly stronger, and
1631  * enough to guarantee that nobody can change the in_use bits on us.
1632  */
1633  LWLockAcquire(ReplicationSlotAllocationLock, LW_SHARED);
1634 
1635  for (i = 0; i < max_replication_slots; i++)
1636  {
1638  char path[MAXPGPATH];
1639 
1640  if (!s->in_use)
1641  continue;
1642 
1643  /* save the slot to disk, locking is handled in SaveSlotToPath() */
1644  sprintf(path, "pg_replslot/%s", NameStr(s->data.name));
1645 
1646  /*
1647  * Slot's data is not flushed each time the confirmed_flush LSN is
1648  * updated as that could lead to frequent writes. However, we decide
1649  * to force a flush of all logical slot's data at the time of shutdown
1650  * if the confirmed_flush LSN is changed since we last flushed it to
1651  * disk. This helps in avoiding an unnecessary retreat of the
1652  * confirmed_flush LSN after restart.
1653  */
1654  if (is_shutdown && SlotIsLogical(s))
1655  {
1656  SpinLockAcquire(&s->mutex);
1657 
1659 
1660  if (s->data.invalidated == RS_INVAL_NONE &&
1662  {
1663  s->just_dirtied = true;
1664  s->dirty = true;
1665  }
1666  SpinLockRelease(&s->mutex);
1667  }
1668 
1669  SaveSlotToPath(s, path, LOG);
1670  }
1671  LWLockRelease(ReplicationSlotAllocationLock);
1672 }
#define NameStr(name)
Definition: c.h:735
#define LOG
Definition: elog.h:31
#define DEBUG1
Definition: elog.h:30
int i
Definition: isn.c:73
Assert(fmt[strlen(fmt) - 1] !='\n')
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1195
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1808
@ LW_SHARED
Definition: lwlock.h:117
#define MAXPGPATH
#define sprintf
Definition: port.h:240
static void SaveSlotToPath(ReplicationSlot *slot, const char *dir, int elevel)
Definition: slot.c:1801
int max_replication_slots
Definition: slot.c:102
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:96
@ RS_INVAL_NONE
Definition: slot.h:46
#define SlotIsLogical(slot)
Definition: slot.h:191
#define SpinLockRelease(lock)
Definition: spin.h:64
#define SpinLockAcquire(lock)
Definition: spin.h:62
ReplicationSlot replication_slots[1]
Definition: slot.h:202
XLogRecPtr confirmed_flush
Definition: slot.h:99
ReplicationSlotInvalidationCause invalidated
Definition: slot.h:91
slock_t mutex
Definition: slot.h:135
XLogRecPtr last_saved_confirmed_flush
Definition: slot.h:187
bool in_use
Definition: slot.h:138
bool just_dirtied
Definition: slot.h:144
bool dirty
Definition: slot.h:145
ReplicationSlotPersistentData data
Definition: slot.h:162

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 1188 of file slot.c.

1189 {
1190  if (!has_rolreplication(GetUserId()))
1191  ereport(ERROR,
1192  (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
1193  errmsg("permission denied to use replication slots"),
1194  errdetail("Only roles with the %s attribute may use replication slots.",
1195  "REPLICATION")));
1196 }
int errdetail(const char *fmt,...)
Definition: elog.c:1202
int errcode(int sqlerrcode)
Definition: elog.c:858
int errmsg(const char *fmt,...)
Definition: elog.c:1069
#define ERROR
Definition: elog.h:39
#define ereport(elevel,...)
Definition: elog.h:149
Oid GetUserId(void)
Definition: miscinit.c:508
bool has_rolreplication(Oid roleid)
Definition: miscinit.c:705

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(), and pg_replication_slot_advance().

◆ CheckSlotRequirements()

void CheckSlotRequirements ( void  )

Definition at line 1166 of file slot.c.

1167 {
1168  /*
1169  * NB: Adding a new requirement likely means that RestoreSlotFromDisk()
1170  * needs the same check.
1171  */
1172 
1173  if (max_replication_slots == 0)
1174  ereport(ERROR,
1175  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1176  errmsg("replication slots can only be used if max_replication_slots > 0")));
1177 
1179  ereport(ERROR,
1180  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1181  errmsg("replication slots can only be used if wal_level >= replica")));
1182 }
int wal_level
Definition: xlog.c:134
@ 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 1740 of file slot.c.

1741 {
1742  char tmppath[MAXPGPATH];
1743  char path[MAXPGPATH];
1744  struct stat st;
1745 
1746  /*
1747  * No need to take out the io_in_progress_lock, nobody else can see this
1748  * slot yet, so nobody else will write. We're reusing SaveSlotToPath which
1749  * takes out the lock, if we'd take the lock here, we'd deadlock.
1750  */
1751 
1752  sprintf(path, "pg_replslot/%s", NameStr(slot->data.name));
1753  sprintf(tmppath, "pg_replslot/%s.tmp", NameStr(slot->data.name));
1754 
1755  /*
1756  * It's just barely possible that some previous effort to create or drop a
1757  * slot with this name left a temp directory lying around. If that seems
1758  * to be the case, try to remove it. If the rmtree() fails, we'll error
1759  * out at the MakePGDirectory() below, so we don't bother checking
1760  * success.
1761  */
1762  if (stat(tmppath, &st) == 0 && S_ISDIR(st.st_mode))
1763  rmtree(tmppath, true);
1764 
1765  /* Create and fsync the temporary slot directory. */
1766  if (MakePGDirectory(tmppath) < 0)
1767  ereport(ERROR,
1769  errmsg("could not create directory \"%s\": %m",
1770  tmppath)));
1771  fsync_fname(tmppath, true);
1772 
1773  /* Write the actual state file. */
1774  slot->dirty = true; /* signal that we really need to write */
1775  SaveSlotToPath(slot, tmppath, ERROR);
1776 
1777  /* Rename the directory into place. */
1778  if (rename(tmppath, path) != 0)
1779  ereport(ERROR,
1781  errmsg("could not rename file \"%s\" to \"%s\": %m",
1782  tmppath, path)));
1783 
1784  /*
1785  * If we'd now fail - really unlikely - we wouldn't know whether this slot
1786  * would persist after an OS crash or not - so, force a restart. The
1787  * restart would try to fsync this again till it works.
1788  */
1790 
1791  fsync_fname(path, true);
1792  fsync_fname("pg_replslot", true);
1793 
1794  END_CRIT_SECTION();
1795 }
int errcode_for_file_access(void)
Definition: elog.c:881
int MakePGDirectory(const char *directoryName)
Definition: fd.c:3883
void fsync_fname(const char *fname, bool isdir)
Definition: fd.c:733
#define START_CRIT_SECTION()
Definition: miscadmin.h:148
#define END_CRIT_SECTION()
Definition: miscadmin.h:150
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().

◆ InvalidateObsoleteReplicationSlots()

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

Definition at line 1564 of file slot.c.

1567 {
1568  XLogRecPtr oldestLSN;
1569  bool invalidated = false;
1570 
1571  Assert(cause != RS_INVAL_HORIZON || TransactionIdIsValid(snapshotConflictHorizon));
1572  Assert(cause != RS_INVAL_WAL_REMOVED || oldestSegno > 0);
1573  Assert(cause != RS_INVAL_NONE);
1574 
1575  if (max_replication_slots == 0)
1576  return invalidated;
1577 
1578  XLogSegNoOffsetToRecPtr(oldestSegno, 0, wal_segment_size, oldestLSN);
1579 
1580 restart:
1581  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
1582  for (int i = 0; i < max_replication_slots; i++)
1583  {
1585 
1586  if (!s->in_use)
1587  continue;
1588 
1589  if (InvalidatePossiblyObsoleteSlot(cause, s, oldestLSN, dboid,
1590  snapshotConflictHorizon,
1591  &invalidated))
1592  {
1593  /* if the lock was released, start from scratch */
1594  goto restart;
1595  }
1596  }
1597  LWLockRelease(ReplicationSlotControlLock);
1598 
1599  /*
1600  * If any slots have been invalidated, recalculate the resource limits.
1601  */
1602  if (invalidated)
1603  {
1606  }
1607 
1608  return invalidated;
1609 }
void ReplicationSlotsComputeRequiredXmin(bool already_locked)
Definition: slot.c:867
void ReplicationSlotsComputeRequiredLSN(void)
Definition: slot.c:923
static bool InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause, ReplicationSlot *s, XLogRecPtr oldestLSN, Oid dboid, TransactionId snapshotConflictHorizon, bool *invalidated)
Definition: slot.c:1347
@ RS_INVAL_WAL_REMOVED
Definition: slot.h:48
@ RS_INVAL_HORIZON
Definition: slot.h:50
#define TransactionIdIsValid(xid)
Definition: transam.h:41
int wal_segment_size
Definition: xlog.c:146
#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 1347 of file slot.c.

1352 {
1353  int last_signaled_pid = 0;
1354  bool released_lock = false;
1355 
1356  for (;;)
1357  {
1358  XLogRecPtr restart_lsn;
1359  NameData slotname;
1360  int active_pid = 0;
1362 
1363  Assert(LWLockHeldByMeInMode(ReplicationSlotControlLock, LW_SHARED));
1364 
1365  if (!s->in_use)
1366  {
1367  if (released_lock)
1368  LWLockRelease(ReplicationSlotControlLock);
1369  break;
1370  }
1371 
1372  /*
1373  * Check if the slot needs to be invalidated. If it needs to be
1374  * invalidated, and is not currently acquired, acquire it and mark it
1375  * as having been invalidated. We do this with the spinlock held to
1376  * avoid race conditions -- for example the restart_lsn could move
1377  * forward, or the slot could be dropped.
1378  */
1379  SpinLockAcquire(&s->mutex);
1380 
1381  restart_lsn = s->data.restart_lsn;
1382 
1383  /*
1384  * If the slot is already invalid or is a non conflicting slot, we
1385  * don't need to do anything.
1386  */
1387  if (s->data.invalidated == RS_INVAL_NONE)
1388  {
1389  switch (cause)
1390  {
1391  case RS_INVAL_WAL_REMOVED:
1392  if (s->data.restart_lsn != InvalidXLogRecPtr &&
1393  s->data.restart_lsn < oldestLSN)
1394  conflict = cause;
1395  break;
1396  case RS_INVAL_HORIZON:
1397  if (!SlotIsLogical(s))
1398  break;
1399  /* invalid DB oid signals a shared relation */
1400  if (dboid != InvalidOid && dboid != s->data.database)
1401  break;
1404  snapshotConflictHorizon))
1405  conflict = cause;
1408  snapshotConflictHorizon))
1409  conflict = cause;
1410  break;
1411  case RS_INVAL_WAL_LEVEL:
1412  if (SlotIsLogical(s))
1413  conflict = cause;
1414  break;
1415  case RS_INVAL_NONE:
1416  pg_unreachable();
1417  }
1418  }
1419 
1420  /* if there's no conflict, we're done */
1421  if (conflict == RS_INVAL_NONE)
1422  {
1423  SpinLockRelease(&s->mutex);
1424  if (released_lock)
1425  LWLockRelease(ReplicationSlotControlLock);
1426  break;
1427  }
1428 
1429  slotname = s->data.name;
1430  active_pid = s->active_pid;
1431 
1432  /*
1433  * If the slot can be acquired, do so and mark it invalidated
1434  * immediately. Otherwise we'll signal the owning process, below, and
1435  * retry.
1436  */
1437  if (active_pid == 0)
1438  {
1439  MyReplicationSlot = s;
1440  s->active_pid = MyProcPid;
1441  s->data.invalidated = conflict;
1442 
1443  /*
1444  * XXX: We should consider not overwriting restart_lsn and instead
1445  * just rely on .invalidated.
1446  */
1447  if (conflict == RS_INVAL_WAL_REMOVED)
1449 
1450  /* Let caller know */
1451  *invalidated = true;
1452  }
1453 
1454  SpinLockRelease(&s->mutex);
1455 
1456  /*
1457  * The logical replication slots shouldn't be invalidated as GUC
1458  * max_slot_wal_keep_size is set to -1 during the binary upgrade. See
1459  * check_old_cluster_for_valid_slots() where we ensure that no
1460  * invalidated before the upgrade.
1461  */
1462  Assert(!(*invalidated && SlotIsLogical(s) && IsBinaryUpgrade));
1463 
1464  if (active_pid != 0)
1465  {
1466  /*
1467  * Prepare the sleep on the slot's condition variable before
1468  * releasing the lock, to close a possible race condition if the
1469  * slot is released before the sleep below.
1470  */
1472 
1473  LWLockRelease(ReplicationSlotControlLock);
1474  released_lock = true;
1475 
1476  /*
1477  * Signal to terminate the process that owns the slot, if we
1478  * haven't already signalled it. (Avoidance of repeated
1479  * signalling is the only reason for there to be a loop in this
1480  * routine; otherwise we could rely on caller's restart loop.)
1481  *
1482  * There is the race condition that other process may own the slot
1483  * after its current owner process is terminated and before this
1484  * process owns it. To handle that, we signal only if the PID of
1485  * the owning process has changed from the previous time. (This
1486  * logic assumes that the same PID is not reused very quickly.)
1487  */
1488  if (last_signaled_pid != active_pid)
1489  {
1490  ReportSlotInvalidation(conflict, true, active_pid,
1491  slotname, restart_lsn,
1492  oldestLSN, snapshotConflictHorizon);
1493 
1494  if (MyBackendType == B_STARTUP)
1495  (void) SendProcSignal(active_pid,
1498  else
1499  (void) kill(active_pid, SIGTERM);
1500 
1501  last_signaled_pid = active_pid;
1502  }
1503 
1504  /* Wait until the slot is released. */
1506  WAIT_EVENT_REPLICATION_SLOT_DROP);
1507 
1508  /*
1509  * Re-acquire lock and start over; we expect to invalidate the
1510  * slot next time (unless another process acquires the slot in the
1511  * meantime).
1512  */
1513  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
1514  continue;
1515  }
1516  else
1517  {
1518  /*
1519  * We hold the slot now and have already invalidated it; flush it
1520  * to ensure that state persists.
1521  *
1522  * Don't want to hold ReplicationSlotControlLock across file
1523  * system operations, so release it now but be sure to tell caller
1524  * to restart from scratch.
1525  */
1526  LWLockRelease(ReplicationSlotControlLock);
1527  released_lock = true;
1528 
1529  /* Make sure the invalidated state persists across server restart */
1534 
1535  ReportSlotInvalidation(conflict, false, active_pid,
1536  slotname, restart_lsn,
1537  oldestLSN, snapshotConflictHorizon);
1538 
1539  /* done with this slot for now */
1540  break;
1541  }
1542  }
1543 
1544  Assert(released_lock == !LWLockHeldByMe(ReplicationSlotControlLock));
1545 
1546  return released_lock;
1547 }
#define InvalidBackendId
Definition: backendid.h:23
#define pg_unreachable()
Definition: c.h:285
void ConditionVariablePrepareToSleep(ConditionVariable *cv)
void ConditionVariableSleep(ConditionVariable *cv, uint32 wait_event_info)
bool IsBinaryUpgrade
Definition: globals.c:116
int MyProcPid
Definition: globals.c:44
bool LWLockHeldByMe(LWLock *lock)
Definition: lwlock.c:1920
bool LWLockHeldByMeInMode(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1964
@ B_STARTUP
Definition: miscadmin.h:340
BackendType MyBackendType
Definition: miscinit.c:63
void pgstat_drop_replslot(ReplicationSlot *slot)
#define InvalidOid
Definition: postgres_ext.h:36
int SendProcSignal(pid_t pid, ProcSignalReason reason, BackendId backendId)
Definition: procsignal.c:262
@ PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT
Definition: procsignal.h:46
void ReplicationSlotMarkDirty(void)
Definition: slot.c:828
ReplicationSlot * MyReplicationSlot
Definition: slot.c:99
void ReplicationSlotSave(void)
Definition: slot.c:810
void ReplicationSlotRelease(void)
Definition: slot.c:559
static void ReportSlotInvalidation(ReplicationSlotInvalidationCause cause, bool terminating, int pid, NameData slotname, XLogRecPtr restart_lsn, XLogRecPtr oldestLSN, TransactionId snapshotConflictHorizon)
Definition: slot.c:1281
ReplicationSlotInvalidationCause
Definition: slot.h:45
@ RS_INVAL_WAL_LEVEL
Definition: slot.h:52
XLogRecPtr restart_lsn
Definition: slot.h:88
TransactionId effective_catalog_xmin
Definition: slot.h:159
pid_t active_pid
Definition: slot.h:141
TransactionId effective_xmin
Definition: slot.h:158
ConditionVariable active_cv
Definition: slot.h:168
Definition: c.h:730
bool TransactionIdPrecedesOrEquals(TransactionId id1, TransactionId id2)
Definition: transam.c:299
#define kill(pid, sig)
Definition: win32_port.h:485
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28

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, ReplicationSlotPersistentData::invalidated, InvalidBackendId, InvalidOid, InvalidXLogRecPtr, IsBinaryUpgrade, kill, LW_SHARED, LWLockAcquire(), LWLockHeldByMe(), LWLockHeldByMeInMode(), LWLockRelease(), ReplicationSlot::mutex, MyBackendType, MyProcPid, MyReplicationSlot, ReplicationSlotPersistentData::name, pg_unreachable, pgstat_drop_replslot(), 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 452 of file slot.c.

453 {
454  ReplicationSlot *s;
455  int active_pid;
456 
457  Assert(name != NULL);
458 
459 retry:
460  Assert(MyReplicationSlot == NULL);
461 
462  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
463 
464  /*
465  * Search for the slot with the specified name if the slot to acquire is
466  * not given. If the slot is not found, we either return -1 or error out.
467  */
468  s = SearchNamedReplicationSlot(name, false);
469  if (s == NULL || !s->in_use)
470  {
471  LWLockRelease(ReplicationSlotControlLock);
472 
473  ereport(ERROR,
474  (errcode(ERRCODE_UNDEFINED_OBJECT),
475  errmsg("replication slot \"%s\" does not exist",
476  name)));
477  }
478 
479  /*
480  * This is the slot we want; check if it's active under some other
481  * process. In single user mode, we don't need this check.
482  */
483  if (IsUnderPostmaster)
484  {
485  /*
486  * Get ready to sleep on the slot in case it is active. (We may end
487  * up not sleeping, but we don't want to do this while holding the
488  * spinlock.)
489  */
490  if (!nowait)
492 
493  SpinLockAcquire(&s->mutex);
494  if (s->active_pid == 0)
495  s->active_pid = MyProcPid;
496  active_pid = s->active_pid;
497  SpinLockRelease(&s->mutex);
498  }
499  else
500  active_pid = MyProcPid;
501  LWLockRelease(ReplicationSlotControlLock);
502 
503  /*
504  * If we found the slot but it's already active in another process, we
505  * wait until the owning process signals us that it's been released, or
506  * error out.
507  */
508  if (active_pid != MyProcPid)
509  {
510  if (!nowait)
511  {
512  /* Wait here until we get signaled, and then restart */
514  WAIT_EVENT_REPLICATION_SLOT_DROP);
516  goto retry;
517  }
518 
519  ereport(ERROR,
520  (errcode(ERRCODE_OBJECT_IN_USE),
521  errmsg("replication slot \"%s\" is active for PID %d",
522  NameStr(s->data.name), active_pid)));
523  }
524  else if (!nowait)
525  ConditionVariableCancelSleep(); /* no sleep needed after all */
526 
527  /* Let everybody know we've modified this slot */
529 
530  /* We made this slot active, so it's ours now. */
531  MyReplicationSlot = s;
532 
533  /*
534  * The call to pgstat_acquire_replslot() protects against stats for a
535  * different slot, from before a restart or such, being present during
536  * pgstat_report_replslot().
537  */
538  if (SlotIsLogical(s))
540 
541  if (am_walsender)
542  {
544  SlotIsLogical(s)
545  ? errmsg("acquired logical replication slot \"%s\"",
546  NameStr(s->data.name))
547  : errmsg("acquired physical replication slot \"%s\"",
548  NameStr(s->data.name)));
549  }
550 }
bool ConditionVariableCancelSleep(void)
void ConditionVariableBroadcast(ConditionVariable *cv)
bool IsUnderPostmaster
Definition: globals.c:115
void pgstat_acquire_replslot(ReplicationSlot *slot)
ReplicationSlot * SearchNamedReplicationSlot(const char *name, bool need_lock)
Definition: slot.c:376
const char * name
bool am_walsender
Definition: walsender.c:116
bool log_replication_commands
Definition: walsender.c:126

References ReplicationSlot::active_cv, ReplicationSlot::active_pid, am_walsender, Assert(), ConditionVariableBroadcast(), ConditionVariableCancelSleep(), ConditionVariablePrepareToSleep(), ConditionVariableSleep(), ReplicationSlot::data, DEBUG1, ereport, errcode(), errmsg(), ERROR, ReplicationSlot::in_use, 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(), pg_logical_slot_get_changes_guts(), pg_replication_slot_advance(), ReplicationSlotDrop(), StartLogicalReplication(), and StartReplication().

◆ ReplicationSlotCleanup()

void ReplicationSlotCleanup ( void  )

Definition at line 635 of file slot.c.

636 {
637  int i;
638 
639  Assert(MyReplicationSlot == NULL);
640 
641 restart:
642  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
643  for (i = 0; i < max_replication_slots; i++)
644  {
646 
647  if (!s->in_use)
648  continue;
649 
650  SpinLockAcquire(&s->mutex);
651  if (s->active_pid == MyProcPid)
652  {
654  SpinLockRelease(&s->mutex);
655  LWLockRelease(ReplicationSlotControlLock); /* avoid deadlock */
656 
658 
660  goto restart;
661  }
662  else
663  SpinLockRelease(&s->mutex);
664  }
665 
666  LWLockRelease(ReplicationSlotControlLock);
667 }
static void ReplicationSlotDropPtr(ReplicationSlot *slot)
Definition: slot.c:703
@ RS_TEMPORARY
Definition: slot.h:37
ReplicationSlotPersistency persistency
Definition: slot.h:69

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 
)

Definition at line 253 of file slot.c.

255 {
256  ReplicationSlot *slot = NULL;
257  int i;
258 
259  Assert(MyReplicationSlot == NULL);
260 
262 
263  /*
264  * If some other backend ran this code concurrently with us, we'd likely
265  * both allocate the same slot, and that would be bad. We'd also be at
266  * risk of missing a name collision. Also, we don't want to try to create
267  * a new slot while somebody's busy cleaning up an old one, because we
268  * might both be monkeying with the same directory.
269  */
270  LWLockAcquire(ReplicationSlotAllocationLock, LW_EXCLUSIVE);
271 
272  /*
273  * Check for name collision, and identify an allocatable slot. We need to
274  * hold ReplicationSlotControlLock in shared mode for this, so that nobody
275  * else can change the in_use flags while we're looking at them.
276  */
277  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
278  for (i = 0; i < max_replication_slots; i++)
279  {
281 
282  if (s->in_use && strcmp(name, NameStr(s->data.name)) == 0)
283  ereport(ERROR,
285  errmsg("replication slot \"%s\" already exists", name)));
286  if (!s->in_use && slot == NULL)
287  slot = s;
288  }
289  LWLockRelease(ReplicationSlotControlLock);
290 
291  /* If all slots are in use, we're out of luck. */
292  if (slot == NULL)
293  ereport(ERROR,
294  (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
295  errmsg("all replication slots are in use"),
296  errhint("Free one or increase max_replication_slots.")));
297 
298  /*
299  * Since this slot is not in use, nobody should be looking at any part of
300  * it other than the in_use field unless they're trying to allocate it.
301  * And since we hold ReplicationSlotAllocationLock, nobody except us can
302  * be doing that. So it's safe to initialize the slot.
303  */
304  Assert(!slot->in_use);
305  Assert(slot->active_pid == 0);
306 
307  /* first initialize persistent data */
308  memset(&slot->data, 0, sizeof(ReplicationSlotPersistentData));
309  namestrcpy(&slot->data.name, name);
310  slot->data.database = db_specific ? MyDatabaseId : InvalidOid;
311  slot->data.persistency = persistency;
312  slot->data.two_phase = two_phase;
314 
315  /* and then data only present in shared memory */
316  slot->just_dirtied = false;
317  slot->dirty = false;
325 
326  /*
327  * Create the slot on disk. We haven't actually marked the slot allocated
328  * yet, so no special cleanup is required if this errors out.
329  */
330  CreateSlotOnDisk(slot);
331 
332  /*
333  * We need to briefly prevent any other backend from iterating over the
334  * slots while we flip the in_use flag. We also need to set the active
335  * flag while holding the ControlLock as otherwise a concurrent
336  * ReplicationSlotAcquire() could acquire the slot as well.
337  */
338  LWLockAcquire(ReplicationSlotControlLock, LW_EXCLUSIVE);
339 
340  slot->in_use = true;
341 
342  /* We can now mark the slot active, and that makes it our slot. */
343  SpinLockAcquire(&slot->mutex);
344  Assert(slot->active_pid == 0);
345  slot->active_pid = MyProcPid;
346  SpinLockRelease(&slot->mutex);
347  MyReplicationSlot = slot;
348 
349  LWLockRelease(ReplicationSlotControlLock);
350 
351  /*
352  * Create statistics entry for the new logical slot. We don't collect any
353  * stats for physical slots, so no need to create an entry for the same.
354  * See ReplicationSlotDropPtr for why we need to do this before releasing
355  * ReplicationSlotAllocationLock.
356  */
357  if (SlotIsLogical(slot))
359 
360  /*
361  * Now that the slot has been marked as in_use and active, it's safe to
362  * let somebody else try to allocate a slot.
363  */
364  LWLockRelease(ReplicationSlotAllocationLock);
365 
366  /* Let everybody know we've modified this slot */
368 }
int errhint(const char *fmt,...)
Definition: elog.c:1316
Oid MyDatabaseId
Definition: globals.c:89
@ LW_EXCLUSIVE
Definition: lwlock.h:116
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:1740
bool ReplicationSlotValidateName(const char *name, int elevel)
Definition: slot.c:199
#define ERRCODE_DUPLICATE_OBJECT
Definition: streamutil.c:32
XLogRecPtr candidate_xmin_lsn
Definition: slot.h:178
XLogRecPtr candidate_restart_valid
Definition: slot.h:179
XLogRecPtr candidate_restart_lsn
Definition: slot.h:180
TransactionId candidate_catalog_xmin
Definition: slot.h:177
#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, i, ReplicationSlot::in_use, InvalidOid, InvalidTransactionId, InvalidXLogRecPtr, 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(), ReplicationSlotCtlData::replication_slots, ReplicationSlotCtl, ReplicationSlotValidateName(), SlotIsLogical, SpinLockAcquire, SpinLockRelease, two_phase, ReplicationSlotPersistentData::two_phase, and ReplicationSlotPersistentData::two_phase_at.

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

◆ ReplicationSlotDrop()

void ReplicationSlotDrop ( const char *  name,
bool  nowait 
)

Definition at line 673 of file slot.c.

674 {
675  Assert(MyReplicationSlot == NULL);
676 
677  ReplicationSlotAcquire(name, nowait);
678 
680 }
void ReplicationSlotAcquire(const char *name, bool nowait)
Definition: slot.c:452
static void ReplicationSlotDropAcquired(void)
Definition: slot.c:686

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

Referenced by DropReplicationSlot(), and pg_drop_replication_slot().

◆ ReplicationSlotDropAcquired()

static void ReplicationSlotDropAcquired ( void  )
static

Definition at line 686 of file slot.c.

687 {
689 
690  Assert(MyReplicationSlot != NULL);
691 
692  /* slot isn't acquired anymore */
693  MyReplicationSlot = NULL;
694 
696 }

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

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

◆ ReplicationSlotDropPtr()

static void ReplicationSlotDropPtr ( ReplicationSlot slot)
static

Definition at line 703 of file slot.c.

704 {
705  char path[MAXPGPATH];
706  char tmppath[MAXPGPATH];
707 
708  /*
709  * If some other backend ran this code concurrently with us, we might try
710  * to delete a slot with a certain name while someone else was trying to
711  * create a slot with the same name.
712  */
713  LWLockAcquire(ReplicationSlotAllocationLock, LW_EXCLUSIVE);
714 
715  /* Generate pathnames. */
716  sprintf(path, "pg_replslot/%s", NameStr(slot->data.name));
717  sprintf(tmppath, "pg_replslot/%s.tmp", NameStr(slot->data.name));
718 
719  /*
720  * Rename the slot directory on disk, so that we'll no longer recognize
721  * this as a valid slot. Note that if this fails, we've got to mark the
722  * slot inactive before bailing out. If we're dropping an ephemeral or a
723  * temporary slot, we better never fail hard as the caller won't expect
724  * the slot to survive and this might get called during error handling.
725  */
726  if (rename(path, tmppath) == 0)
727  {
728  /*
729  * We need to fsync() the directory we just renamed and its parent to
730  * make sure that our changes are on disk in a crash-safe fashion. If
731  * fsync() fails, we can't be sure whether the changes are on disk or
732  * not. For now, we handle that by panicking;
733  * StartupReplicationSlots() will try to straighten it out after
734  * restart.
735  */
737  fsync_fname(tmppath, true);
738  fsync_fname("pg_replslot", true);
740  }
741  else
742  {
743  bool fail_softly = slot->data.persistency != RS_PERSISTENT;
744 
745  SpinLockAcquire(&slot->mutex);
746  slot->active_pid = 0;
747  SpinLockRelease(&slot->mutex);
748 
749  /* wake up anyone waiting on this slot */
751 
752  ereport(fail_softly ? WARNING : ERROR,
754  errmsg("could not rename file \"%s\" to \"%s\": %m",
755  path, tmppath)));
756  }
757 
758  /*
759  * The slot is definitely gone. Lock out concurrent scans of the array
760  * long enough to kill it. It's OK to clear the active PID here without
761  * grabbing the mutex because nobody else can be scanning the array here,
762  * and nobody can be attached to this slot and thus access it without
763  * scanning the array.
764  *
765  * Also wake up processes waiting for it.
766  */
767  LWLockAcquire(ReplicationSlotControlLock, LW_EXCLUSIVE);
768  slot->active_pid = 0;
769  slot->in_use = false;
770  LWLockRelease(ReplicationSlotControlLock);
772 
773  /*
774  * Slot is dead and doesn't prevent resource removal anymore, recompute
775  * limits.
776  */
779 
780  /*
781  * If removing the directory fails, the worst thing that will happen is
782  * that the user won't be able to create a new slot with the same name
783  * until the next server restart. We warn about it, but that's all.
784  */
785  if (!rmtree(tmppath, true))
787  (errmsg("could not remove directory \"%s\"", tmppath)));
788 
789  /*
790  * Drop the statistics entry for the replication slot. Do this while
791  * holding ReplicationSlotAllocationLock so that we don't drop a
792  * statistics entry for another slot with the same name just created in
793  * another session.
794  */
795  if (SlotIsLogical(slot))
796  pgstat_drop_replslot(slot);
797 
798  /*
799  * We release this at the very end, so that nobody starts trying to create
800  * a slot while we're still cleaning up the detritus of the old one.
801  */
802  LWLockRelease(ReplicationSlotAllocationLock);
803 }
#define WARNING
Definition: elog.h:36
@ 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 171 of file slot.c.

172 {
174 }
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:180

References before_shmem_exit(), and ReplicationSlotShmemExit().

Referenced by BaseInit().

◆ ReplicationSlotMarkDirty()

◆ ReplicationSlotName()

bool ReplicationSlotName ( int  index,
Name  name 
)

Definition at line 425 of file slot.c.

426 {
427  ReplicationSlot *slot;
428  bool found;
429 
431 
432  /*
433  * Ensure that the slot cannot be dropped while we copy the name. Don't
434  * need the spinlock as the name of an existing slot cannot change.
435  */
436  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
437  found = slot->in_use;
438  if (slot->in_use)
439  namestrcpy(name, NameStr(slot->data.name));
440  LWLockRelease(ReplicationSlotControlLock);
441 
442  return found;
443 }
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 559 of file slot.c.

560 {
562  char *slotname = NULL; /* keep compiler quiet */
563  bool is_logical = false; /* keep compiler quiet */
564 
565  Assert(slot != NULL && slot->active_pid != 0);
566 
567  if (am_walsender)
568  {
569  slotname = pstrdup(NameStr(slot->data.name));
570  is_logical = SlotIsLogical(slot);
571  }
572 
573  if (slot->data.persistency == RS_EPHEMERAL)
574  {
575  /*
576  * Delete the slot. There is no !PANIC case where this is allowed to
577  * fail, all that may happen is an incomplete cleanup of the on-disk
578  * data.
579  */
581  }
582 
583  /*
584  * If slot needed to temporarily restrain both data and catalog xmin to
585  * create the catalog snapshot, remove that temporary constraint.
586  * Snapshots can only be exported while the initial snapshot is still
587  * acquired.
588  */
589  if (!TransactionIdIsValid(slot->data.xmin) &&
591  {
592  SpinLockAcquire(&slot->mutex);
594  SpinLockRelease(&slot->mutex);
596  }
597 
598  if (slot->data.persistency == RS_PERSISTENT)
599  {
600  /*
601  * Mark persistent slot inactive. We're not freeing it, just
602  * disconnecting, but wake up others that may be waiting for it.
603  */
604  SpinLockAcquire(&slot->mutex);
605  slot->active_pid = 0;
606  SpinLockRelease(&slot->mutex);
608  }
609 
610  MyReplicationSlot = NULL;
611 
612  /* might not have been set when we've been a plain slot */
613  LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
616  LWLockRelease(ProcArrayLock);
617 
618  if (am_walsender)
619  {
621  is_logical
622  ? errmsg("released logical replication slot \"%s\"",
623  slotname)
624  : errmsg("released physical replication slot \"%s\"",
625  slotname));
626 
627  pfree(slotname);
628  }
629 }
char * pstrdup(const char *in)
Definition: mcxt.c:1644
void pfree(void *pointer)
Definition: mcxt.c:1456
#define PROC_IN_LOGICAL_DECODING
Definition: proc.h:60
@ RS_EPHEMERAL
Definition: slot.h:36
PGPROC * MyProc
Definition: proc.c:66
PROC_HDR * ProcGlobal
Definition: proc.c:78
uint8 statusFlags
Definition: proc.h:233
int pgxactoff
Definition: proc.h:188
uint8 * statusFlags
Definition: proc.h:377
TransactionId xmin
Definition: slot.h:77

References ReplicationSlot::active_cv, ReplicationSlot::active_pid, am_walsender, Assert(), ConditionVariableBroadcast(), ReplicationSlot::data, DEBUG1, ReplicationSlot::effective_xmin, ereport, errmsg(), InvalidTransactionId, LOG, log_replication_commands, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), ReplicationSlot::mutex, MyProc, MyReplicationSlot, ReplicationSlotPersistentData::name, NameStr, ReplicationSlotPersistentData::persistency, pfree(), PGPROC::pgxactoff, PROC_IN_LOGICAL_DECODING, ProcGlobal, pstrdup(), ReplicationSlotDropAcquired(), ReplicationSlotsComputeRequiredXmin(), RS_EPHEMERAL, RS_PERSISTENT, SlotIsLogical, SpinLockAcquire, SpinLockRelease, PGPROC::statusFlags, PROC_HDR::statusFlags, 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(), ReplicationSlotShmemExit(), StartLogicalReplication(), StartReplication(), and WalSndErrorCleanup().

◆ ReplicationSlotReserveWal()

void ReplicationSlotReserveWal ( void  )

Definition at line 1205 of file slot.c.

1206 {
1208 
1209  Assert(slot != NULL);
1211 
1212  /*
1213  * The replication slot mechanism is used to prevent removal of required
1214  * WAL. As there is no interlock between this routine and checkpoints, WAL
1215  * segments could concurrently be removed when a now stale return value of
1216  * ReplicationSlotsComputeRequiredLSN() is used. In the unlikely case that
1217  * this happens we'll just retry.
1218  */
1219  while (true)
1220  {
1221  XLogSegNo segno;
1222  XLogRecPtr restart_lsn;
1223 
1224  /*
1225  * For logical slots log a standby snapshot and start logical decoding
1226  * at exactly that position. That allows the slot to start up more
1227  * quickly. But on a standby we cannot do WAL writes, so just use the
1228  * replay pointer; effectively, an attempt to create a logical slot on
1229  * standby will cause it to wait for an xl_running_xact record to be
1230  * logged independently on the primary, so that a snapshot can be
1231  * built using the record.
1232  *
1233  * None of this is needed (or indeed helpful) for physical slots as
1234  * they'll start replay at the last logged checkpoint anyway. Instead
1235  * return the location of the last redo LSN. While that slightly
1236  * increases the chance that we have to retry, it's where a base
1237  * backup has to start replay at.
1238  */
1239  if (SlotIsPhysical(slot))
1240  restart_lsn = GetRedoRecPtr();
1241  else if (RecoveryInProgress())
1242  restart_lsn = GetXLogReplayRecPtr(NULL);
1243  else
1244  restart_lsn = GetXLogInsertRecPtr();
1245 
1246  SpinLockAcquire(&slot->mutex);
1247  slot->data.restart_lsn = restart_lsn;
1248  SpinLockRelease(&slot->mutex);
1249 
1250  /* prevent WAL removal as fast as possible */
1252 
1253  /*
1254  * If all required WAL is still there, great, otherwise retry. The
1255  * slot should prevent further removal of WAL, unless there's a
1256  * concurrent ReplicationSlotsComputeRequiredLSN() after we've written
1257  * the new restart_lsn above, so normally we should never need to loop
1258  * more than twice.
1259  */
1261  if (XLogGetLastRemovedSegno() < segno)
1262  break;
1263  }
1264 
1265  if (!RecoveryInProgress() && SlotIsLogical(slot))
1266  {
1267  XLogRecPtr flushptr;
1268 
1269  /* make sure we have enough information to start */
1270  flushptr = LogStandbySnapshot();
1271 
1272  /* and make sure it's fsynced to disk */
1273  XLogFlush(flushptr);
1274  }
1275 }
#define SlotIsPhysical(slot)
Definition: slot.h:190
XLogRecPtr LogStandbySnapshot(void)
Definition: standby.c:1287
bool RecoveryInProgress(void)
Definition: xlog.c:6039
XLogSegNo XLogGetLastRemovedSegno(void)
Definition: xlog.c:3581
XLogRecPtr GetRedoRecPtr(void)
Definition: xlog.c:6142
XLogRecPtr GetXLogInsertRecPtr(void)
Definition: xlog.c:9067
void XLogFlush(XLogRecPtr record)
Definition: xlog.c:2621
#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 972 of file slot.c.

973 {
974  XLogRecPtr result = InvalidXLogRecPtr;
975  int i;
976 
977  if (max_replication_slots <= 0)
978  return InvalidXLogRecPtr;
979 
980  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
981 
982  for (i = 0; i < max_replication_slots; i++)
983  {
984  ReplicationSlot *s;
985  XLogRecPtr restart_lsn;
986  bool invalidated;
987 
989 
990  /* cannot change while ReplicationSlotCtlLock is held */
991  if (!s->in_use)
992  continue;
993 
994  /* we're only interested in logical slots */
995  if (!SlotIsLogical(s))
996  continue;
997 
998  /* read once, it's ok if it increases while we're checking */
999  SpinLockAcquire(&s->mutex);
1000  restart_lsn = s->data.restart_lsn;
1001  invalidated = s->data.invalidated != RS_INVAL_NONE;
1002  SpinLockRelease(&s->mutex);
1003 
1004  /* invalidated slots need not apply */
1005  if (invalidated)
1006  continue;
1007 
1008  if (restart_lsn == InvalidXLogRecPtr)
1009  continue;
1010 
1011  if (result == InvalidXLogRecPtr ||
1012  restart_lsn < result)
1013  result = restart_lsn;
1014  }
1015 
1016  LWLockRelease(ReplicationSlotControlLock);
1017 
1018  return result;
1019 }

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 923 of file slot.c.

924 {
925  int i;
926  XLogRecPtr min_required = InvalidXLogRecPtr;
927 
928  Assert(ReplicationSlotCtl != NULL);
929 
930  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
931  for (i = 0; i < max_replication_slots; i++)
932  {
934  XLogRecPtr restart_lsn;
935  bool invalidated;
936 
937  if (!s->in_use)
938  continue;
939 
940  SpinLockAcquire(&s->mutex);
941  restart_lsn = s->data.restart_lsn;
942  invalidated = s->data.invalidated != RS_INVAL_NONE;
943  SpinLockRelease(&s->mutex);
944 
945  /* invalidated slots need not apply */
946  if (invalidated)
947  continue;
948 
949  if (restart_lsn != InvalidXLogRecPtr &&
950  (min_required == InvalidXLogRecPtr ||
951  restart_lsn < min_required))
952  min_required = restart_lsn;
953  }
954  LWLockRelease(ReplicationSlotControlLock);
955 
956  XLogSetReplicationSlotMinimumLSN(min_required);
957 }
void XLogSetReplicationSlotMinimumLSN(XLogRecPtr lsn)
Definition: xlog.c:2507

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(), and StartupReplicationSlots().

◆ ReplicationSlotsComputeRequiredXmin()

void ReplicationSlotsComputeRequiredXmin ( bool  already_locked)

Definition at line 867 of file slot.c.

868 {
869  int i;
871  TransactionId agg_catalog_xmin = InvalidTransactionId;
872 
873  Assert(ReplicationSlotCtl != NULL);
874 
875  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
876 
877  for (i = 0; i < max_replication_slots; i++)
878  {
880  TransactionId effective_xmin;
881  TransactionId effective_catalog_xmin;
882  bool invalidated;
883 
884  if (!s->in_use)
885  continue;
886 
887  SpinLockAcquire(&s->mutex);
888  effective_xmin = s->effective_xmin;
889  effective_catalog_xmin = s->effective_catalog_xmin;
890  invalidated = s->data.invalidated != RS_INVAL_NONE;
891  SpinLockRelease(&s->mutex);
892 
893  /* invalidated slots need not apply */
894  if (invalidated)
895  continue;
896 
897  /* check the data xmin */
898  if (TransactionIdIsValid(effective_xmin) &&
899  (!TransactionIdIsValid(agg_xmin) ||
900  TransactionIdPrecedes(effective_xmin, agg_xmin)))
901  agg_xmin = effective_xmin;
902 
903  /* check the catalog xmin */
904  if (TransactionIdIsValid(effective_catalog_xmin) &&
905  (!TransactionIdIsValid(agg_catalog_xmin) ||
906  TransactionIdPrecedes(effective_catalog_xmin, agg_catalog_xmin)))
907  agg_catalog_xmin = effective_catalog_xmin;
908  }
909 
910  LWLockRelease(ReplicationSlotControlLock);
911 
912  ProcArraySetReplicationSlotXmin(agg_xmin, agg_catalog_xmin, already_locked);
913 }
uint32 TransactionId
Definition: c.h:641
void ProcArraySetReplicationSlotXmin(TransactionId xmin, TransactionId catalog_xmin, bool already_locked)
Definition: procarray.c:3846
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(), and StartupReplicationSlots().

◆ ReplicationSlotsCountDBSlots()

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

Definition at line 1030 of file slot.c.

1031 {
1032  int i;
1033 
1034  *nslots = *nactive = 0;
1035 
1036  if (max_replication_slots <= 0)
1037  return false;
1038 
1039  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
1040  for (i = 0; i < max_replication_slots; i++)
1041  {
1042  ReplicationSlot *s;
1043 
1045 
1046  /* cannot change while ReplicationSlotCtlLock is held */
1047  if (!s->in_use)
1048  continue;
1049 
1050  /* only logical slots are database specific, skip */
1051  if (!SlotIsLogical(s))
1052  continue;
1053 
1054  /* not our database, skip */
1055  if (s->data.database != dboid)
1056  continue;
1057 
1058  /* NB: intentionally counting invalidated slots */
1059 
1060  /* count slots with spinlock held */
1061  SpinLockAcquire(&s->mutex);
1062  (*nslots)++;
1063  if (s->active_pid != 0)
1064  (*nactive)++;
1065  SpinLockRelease(&s->mutex);
1066  }
1067  LWLockRelease(ReplicationSlotControlLock);
1068 
1069  if (*nslots > 0)
1070  return true;
1071  return false;
1072 }

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 1088 of file slot.c.

1089 {
1090  int i;
1091 
1092  if (max_replication_slots <= 0)
1093  return;
1094 
1095 restart:
1096  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
1097  for (i = 0; i < max_replication_slots; i++)
1098  {
1099  ReplicationSlot *s;
1100  char *slotname;
1101  int active_pid;
1102 
1104 
1105  /* cannot change while ReplicationSlotCtlLock is held */
1106  if (!s->in_use)
1107  continue;
1108 
1109  /* only logical slots are database specific, skip */
1110  if (!SlotIsLogical(s))
1111  continue;
1112 
1113  /* not our database, skip */
1114  if (s->data.database != dboid)
1115  continue;
1116 
1117  /* NB: intentionally including invalidated slots */
1118 
1119  /* acquire slot, so ReplicationSlotDropAcquired can be reused */
1120  SpinLockAcquire(&s->mutex);
1121  /* can't change while ReplicationSlotControlLock is held */
1122  slotname = NameStr(s->data.name);
1123  active_pid = s->active_pid;
1124  if (active_pid == 0)
1125  {
1126  MyReplicationSlot = s;
1127  s->active_pid = MyProcPid;
1128  }
1129  SpinLockRelease(&s->mutex);
1130 
1131  /*
1132  * Even though we hold an exclusive lock on the database object a
1133  * logical slot for that DB can still be active, e.g. if it's
1134  * concurrently being dropped by a backend connected to another DB.
1135  *
1136  * That's fairly unlikely in practice, so we'll just bail out.
1137  */
1138  if (active_pid)
1139  ereport(ERROR,
1140  (errcode(ERRCODE_OBJECT_IN_USE),
1141  errmsg("replication slot \"%s\" is active for PID %d",
1142  slotname, active_pid)));
1143 
1144  /*
1145  * To avoid duplicating ReplicationSlotDropAcquired() and to avoid
1146  * holding ReplicationSlotControlLock over filesystem operations,
1147  * release ReplicationSlotControlLock and use
1148  * ReplicationSlotDropAcquired.
1149  *
1150  * As that means the set of slots could change, restart scan from the
1151  * beginning each time we release the lock.
1152  */
1153  LWLockRelease(ReplicationSlotControlLock);
1155  goto restart;
1156  }
1157  LWLockRelease(ReplicationSlotControlLock);
1158 }

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 180 of file slot.c.

181 {
182  /* Make sure active replication slots are released */
183  if (MyReplicationSlot != NULL)
185 
186  /* Also cleanup all the temporary slots. */
188 }
void ReplicationSlotCleanup(void)
Definition: slot.c:635

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

Referenced by ReplicationSlotInitialize().

◆ ReplicationSlotsShmemInit()

void ReplicationSlotsShmemInit ( void  )

Definition at line 136 of file slot.c.

137 {
138  bool found;
139 
140  if (max_replication_slots == 0)
141  return;
142 
144  ShmemInitStruct("ReplicationSlot Ctl", ReplicationSlotsShmemSize(),
145  &found);
146 
147  if (!found)
148  {
149  int i;
150 
151  /* First time through, so initialize */
153 
154  for (i = 0; i < max_replication_slots; i++)
155  {
157 
158  /* everything else is zeroed by the memset above */
159  SpinLockInit(&slot->mutex);
163  }
164  }
165 }
#define MemSet(start, val, len)
Definition: c.h:1009
void ConditionVariableInit(ConditionVariable *cv)
void LWLockInitialize(LWLock *lock, int tranche_id)
Definition: lwlock.c:730
@ LWTRANCHE_REPLICATION_SLOT_IO
Definition: lwlock.h:191
void * ShmemInitStruct(const char *name, Size size, bool *foundPtr)
Definition: shmem.c:388
Size ReplicationSlotsShmemSize(void)
Definition: slot.c:118
#define SpinLockInit(lock)
Definition: spin.h:60
LWLock io_in_progress_lock
Definition: slot.h:165

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 118 of file slot.c.

119 {
120  Size size = 0;
121 
122  if (max_replication_slots == 0)
123  return size;
124 
125  size = offsetof(ReplicationSlotCtlData, replication_slots);
126  size = add_size(size,
128 
129  return size;
130 }
size_t Size
Definition: c.h:594
Size add_size(Size s1, Size s2)
Definition: shmem.c:494
Size mul_size(Size s1, Size s2)
Definition: shmem.c:511

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

Referenced by CalculateShmemSize(), and ReplicationSlotsShmemInit().

◆ ReplicationSlotValidateName()

bool ReplicationSlotValidateName ( const char *  name,
int  elevel 
)

Definition at line 199 of file slot.c.

200 {
201  const char *cp;
202 
203  if (strlen(name) == 0)
204  {
205  ereport(elevel,
206  (errcode(ERRCODE_INVALID_NAME),
207  errmsg("replication slot name \"%s\" is too short",
208  name)));
209  return false;
210  }
211 
212  if (strlen(name) >= NAMEDATALEN)
213  {
214  ereport(elevel,
215  (errcode(ERRCODE_NAME_TOO_LONG),
216  errmsg("replication slot name \"%s\" is too long",
217  name)));
218  return false;
219  }
220 
221  for (cp = name; *cp; cp++)
222  {
223  if (!((*cp >= 'a' && *cp <= 'z')
224  || (*cp >= '0' && *cp <= '9')
225  || (*cp == '_')))
226  {
227  ereport(elevel,
228  (errcode(ERRCODE_INVALID_NAME),
229  errmsg("replication slot name \"%s\" contains invalid character",
230  name),
231  errhint("Replication slot names may only contain lower case letters, numbers, and the underscore character.")));
232  return false;
233  }
234  }
235  return true;
236 }
#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 1281 of file slot.c.

1288 {
1289  StringInfoData err_detail;
1290  bool hint = false;
1291 
1292  initStringInfo(&err_detail);
1293 
1294  switch (cause)
1295  {
1296  case RS_INVAL_WAL_REMOVED:
1297  {
1298  unsigned long long ex = oldestLSN - restart_lsn;
1299 
1300  hint = true;
1301  appendStringInfo(&err_detail,
1302  ngettext("The slot's restart_lsn %X/%X exceeds the limit by %llu byte.",
1303  "The slot's restart_lsn %X/%X exceeds the limit by %llu bytes.",
1304  ex),
1305  LSN_FORMAT_ARGS(restart_lsn),
1306  ex);
1307  break;
1308  }
1309  case RS_INVAL_HORIZON:
1310  appendStringInfo(&err_detail, _("The slot conflicted with xid horizon %u."),
1311  snapshotConflictHorizon);
1312  break;
1313 
1314  case RS_INVAL_WAL_LEVEL:
1315  appendStringInfoString(&err_detail, _("Logical decoding on standby requires wal_level >= logical on the primary server."));
1316  break;
1317  case RS_INVAL_NONE:
1318  pg_unreachable();
1319  }
1320 
1321  ereport(LOG,
1322  terminating ?
1323  errmsg("terminating process %d to release replication slot \"%s\"",
1324  pid, NameStr(slotname)) :
1325  errmsg("invalidating obsolete replication slot \"%s\"",
1326  NameStr(slotname)),
1327  errdetail_internal("%s", err_detail.data),
1328  hint ? errhint("You might need to increase %s.", "max_slot_wal_keep_size") : 0);
1329 
1330  pfree(err_detail.data);
1331 }
#define ngettext(s, p, n)
Definition: c.h:1194
int errdetail_internal(const char *fmt,...)
Definition: elog.c:1229
#define _(x)
Definition: elog.c:91
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 1956 of file slot.c.

1957 {
1959  int i;
1960  char slotdir[MAXPGPATH + 12];
1961  char path[MAXPGPATH + 22];
1962  int fd;
1963  bool restored = false;
1964  int readBytes;
1965  pg_crc32c checksum;
1966 
1967  /* no need to lock here, no concurrent access allowed yet */
1968 
1969  /* delete temp file if it exists */
1970  sprintf(slotdir, "pg_replslot/%s", name);
1971  sprintf(path, "%s/state.tmp", slotdir);
1972  if (unlink(path) < 0 && errno != ENOENT)
1973  ereport(PANIC,
1975  errmsg("could not remove file \"%s\": %m", path)));
1976 
1977  sprintf(path, "%s/state", slotdir);
1978 
1979  elog(DEBUG1, "restoring replication slot from \"%s\"", path);
1980 
1981  /* on some operating systems fsyncing a file requires O_RDWR */
1982  fd = OpenTransientFile(path, O_RDWR | PG_BINARY);
1983 
1984  /*
1985  * We do not need to handle this as we are rename()ing the directory into
1986  * place only after we fsync()ed the state file.
1987  */
1988  if (fd < 0)
1989  ereport(PANIC,
1991  errmsg("could not open file \"%s\": %m", path)));
1992 
1993  /*
1994  * Sync state file before we're reading from it. We might have crashed
1995  * while it wasn't synced yet and we shouldn't continue on that basis.
1996  */
1997  pgstat_report_wait_start(WAIT_EVENT_REPLICATION_SLOT_RESTORE_SYNC);
1998  if (pg_fsync(fd) != 0)
1999  ereport(PANIC,
2001  errmsg("could not fsync file \"%s\": %m",
2002  path)));
2004 
2005  /* Also sync the parent directory */
2007  fsync_fname(slotdir, true);
2008  END_CRIT_SECTION();
2009 
2010  /* read part of statefile that's guaranteed to be version independent */
2011  pgstat_report_wait_start(WAIT_EVENT_REPLICATION_SLOT_READ);
2012  readBytes = read(fd, &cp, ReplicationSlotOnDiskConstantSize);
2014  if (readBytes != ReplicationSlotOnDiskConstantSize)
2015  {
2016  if (readBytes < 0)
2017  ereport(PANIC,
2019  errmsg("could not read file \"%s\": %m", path)));
2020  else
2021  ereport(PANIC,
2023  errmsg("could not read file \"%s\": read %d of %zu",
2024  path, readBytes,
2026  }
2027 
2028  /* verify magic */
2029  if (cp.magic != SLOT_MAGIC)
2030  ereport(PANIC,
2032  errmsg("replication slot file \"%s\" has wrong magic number: %u instead of %u",
2033  path, cp.magic, SLOT_MAGIC)));
2034 
2035  /* verify version */
2036  if (cp.version != SLOT_VERSION)
2037  ereport(PANIC,
2039  errmsg("replication slot file \"%s\" has unsupported version %u",
2040  path, cp.version)));
2041 
2042  /* boundary check on length */
2044  ereport(PANIC,
2046  errmsg("replication slot file \"%s\" has corrupted length %u",
2047  path, cp.length)));
2048 
2049  /* Now that we know the size, read the entire file */
2050  pgstat_report_wait_start(WAIT_EVENT_REPLICATION_SLOT_READ);
2051  readBytes = read(fd,
2052  (char *) &cp + ReplicationSlotOnDiskConstantSize,
2053  cp.length);
2055  if (readBytes != cp.length)
2056  {
2057  if (readBytes < 0)
2058  ereport(PANIC,
2060  errmsg("could not read file \"%s\": %m", path)));
2061  else
2062  ereport(PANIC,
2064  errmsg("could not read file \"%s\": read %d of %zu",
2065  path, readBytes, (Size) cp.length)));
2066  }
2067 
2068  if (CloseTransientFile(fd) != 0)
2069  ereport(PANIC,
2071  errmsg("could not close file \"%s\": %m", path)));
2072 
2073  /* now verify the CRC */
2074  INIT_CRC32C(checksum);
2075  COMP_CRC32C(checksum,
2078  FIN_CRC32C(checksum);
2079 
2080  if (!EQ_CRC32C(checksum, cp.checksum))
2081  ereport(PANIC,
2082  (errmsg("checksum mismatch for replication slot file \"%s\": is %u, should be %u",
2083  path, checksum, cp.checksum)));
2084 
2085  /*
2086  * If we crashed with an ephemeral slot active, don't restore but delete
2087  * it.
2088  */
2090  {
2091  if (!rmtree(slotdir, true))
2092  {
2093  ereport(WARNING,
2094  (errmsg("could not remove directory \"%s\"",
2095  slotdir)));
2096  }
2097  fsync_fname("pg_replslot", true);
2098  return;
2099  }
2100 
2101  /*
2102  * Verify that requirements for the specific slot type are met. That's
2103  * important because if these aren't met we're not guaranteed to retain
2104  * all the necessary resources for the slot.
2105  *
2106  * NB: We have to do so *after* the above checks for ephemeral slots,
2107  * because otherwise a slot that shouldn't exist anymore could prevent
2108  * restarts.
2109  *
2110  * NB: Changing the requirements here also requires adapting
2111  * CheckSlotRequirements() and CheckLogicalDecodingRequirements().
2112  */
2114  ereport(FATAL,
2115  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
2116  errmsg("logical replication slot \"%s\" exists, but wal_level < logical",
2117  NameStr(cp.slotdata.name)),
2118  errhint("Change wal_level to be logical or higher.")));
2119  else if (wal_level < WAL_LEVEL_REPLICA)
2120  ereport(FATAL,
2121  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
2122  errmsg("physical replication slot \"%s\" exists, but wal_level < replica",
2123  NameStr(cp.slotdata.name)),
2124  errhint("Change wal_level to be replica or higher.")));
2125 
2126  /* nothing can be active yet, don't lock anything */
2127  for (i = 0; i < max_replication_slots; i++)
2128  {
2129  ReplicationSlot *slot;
2130 
2132 
2133  if (slot->in_use)
2134  continue;
2135 
2136  /* restore the entire set of persistent data */
2137  memcpy(&slot->data, &cp.slotdata,
2139 
2140  /* initialize in memory state */
2141  slot->effective_xmin = cp.slotdata.xmin;
2144 
2149 
2150  slot->in_use = true;
2151  slot->active_pid = 0;
2152 
2153  restored = true;
2154  break;
2155  }
2156 
2157  if (!restored)
2158  ereport(FATAL,
2159  (errmsg("too many replication slots active before shutdown"),
2160  errhint("Increase max_replication_slots and try again.")));
2161 }
#define PG_BINARY
Definition: c.h:1283
#define FATAL
Definition: elog.h:41
#define PANIC
Definition: elog.h:42
int CloseTransientFile(int fd)
Definition: fd.c:2779
int pg_fsync(int fd)
Definition: fd.c:386
int OpenTransientFile(const char *fileName, int fileFlags)
Definition: fd.c:2603
#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:86
#define ReplicationSlotOnDiskNotChecksummedSize
Definition: slot.c:83
#define ReplicationSlotOnDiskV2Size
Definition: slot.c:89
#define SLOT_VERSION
Definition: slot.c:93
#define SLOT_MAGIC
Definition: slot.c:92
#define ReplicationSlotOnDiskConstantSize
Definition: slot.c:80
uint32 version
Definition: slot.c:68
ReplicationSlotPersistentData slotdata
Definition: slot.c:76
pg_crc32c checksum
Definition: slot.c:65
TransactionId catalog_xmin
Definition: slot.h:85
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(), i, ReplicationSlot::in_use, 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, ReplicationSlotCtlData::replication_slots, ReplicationSlotCtl, ReplicationSlotOnDiskChecksummedSize, ReplicationSlotOnDiskConstantSize, ReplicationSlotOnDiskNotChecksummedSize, ReplicationSlotOnDiskV2Size, rmtree(), RS_PERSISTENT, SLOT_MAGIC, SLOT_VERSION, ReplicationSlotOnDisk::slotdata, sprintf, START_CRIT_SECTION, 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 1801 of file slot.c.

1802 {
1803  char tmppath[MAXPGPATH];
1804  char path[MAXPGPATH];
1805  int fd;
1807  bool was_dirty;
1808 
1809  /* first check whether there's something to write out */
1810  SpinLockAcquire(&slot->mutex);
1811  was_dirty = slot->dirty;
1812  slot->just_dirtied = false;
1813  SpinLockRelease(&slot->mutex);
1814 
1815  /* and don't do anything if there's nothing to write */
1816  if (!was_dirty)
1817  return;
1818 
1820 
1821  /* silence valgrind :( */
1822  memset(&cp, 0, sizeof(ReplicationSlotOnDisk));
1823 
1824  sprintf(tmppath, "%s/state.tmp", dir);
1825  sprintf(path, "%s/state", dir);
1826 
1827  fd = OpenTransientFile(tmppath, O_CREAT | O_EXCL | O_WRONLY | PG_BINARY);
1828  if (fd < 0)
1829  {
1830  /*
1831  * If not an ERROR, then release the lock before returning. In case
1832  * of an ERROR, the error recovery path automatically releases the
1833  * lock, but no harm in explicitly releasing even in that case. Note
1834  * that LWLockRelease() could affect errno.
1835  */
1836  int save_errno = errno;
1837 
1839  errno = save_errno;
1840  ereport(elevel,
1842  errmsg("could not create file \"%s\": %m",
1843  tmppath)));
1844  return;
1845  }
1846 
1847  cp.magic = SLOT_MAGIC;
1848  INIT_CRC32C(cp.checksum);
1849  cp.version = SLOT_VERSION;
1851 
1852  SpinLockAcquire(&slot->mutex);
1853 
1854  memcpy(&cp.slotdata, &slot->data, sizeof(ReplicationSlotPersistentData));
1855 
1856  SpinLockRelease(&slot->mutex);
1857 
1858  COMP_CRC32C(cp.checksum,
1859  (char *) (&cp) + ReplicationSlotOnDiskNotChecksummedSize,
1861  FIN_CRC32C(cp.checksum);
1862 
1863  errno = 0;
1864  pgstat_report_wait_start(WAIT_EVENT_REPLICATION_SLOT_WRITE);
1865  if ((write(fd, &cp, sizeof(cp))) != sizeof(cp))
1866  {
1867  int save_errno = errno;
1868 
1872 
1873  /* if write didn't set errno, assume problem is no disk space */
1874  errno = save_errno ? save_errno : ENOSPC;
1875  ereport(elevel,
1877  errmsg("could not write to file \"%s\": %m",
1878  tmppath)));
1879  return;
1880  }
1882 
1883  /* fsync the temporary file */
1884  pgstat_report_wait_start(WAIT_EVENT_REPLICATION_SLOT_SYNC);
1885  if (pg_fsync(fd) != 0)
1886  {
1887  int save_errno = errno;
1888 
1892  errno = save_errno;
1893  ereport(elevel,
1895  errmsg("could not fsync file \"%s\": %m",
1896  tmppath)));
1897  return;
1898  }
1900 
1901  if (CloseTransientFile(fd) != 0)
1902  {
1903  int save_errno = errno;
1904 
1906  errno = save_errno;
1907  ereport(elevel,
1909  errmsg("could not close file \"%s\": %m",
1910  tmppath)));
1911  return;
1912  }
1913 
1914  /* rename to permanent file, fsync file and directory */
1915  if (rename(tmppath, path) != 0)
1916  {
1917  int save_errno = errno;
1918 
1920  errno = save_errno;
1921  ereport(elevel,
1923  errmsg("could not rename file \"%s\" to \"%s\": %m",
1924  tmppath, path)));
1925  return;
1926  }
1927 
1928  /*
1929  * Check CreateSlotOnDisk() for the reasoning of using a critical section.
1930  */
1932 
1933  fsync_fname(path, false);
1934  fsync_fname(dir, true);
1935  fsync_fname("pg_replslot", true);
1936 
1937  END_CRIT_SECTION();
1938 
1939  /*
1940  * Successfully wrote, unset dirty bit, unless somebody dirtied again
1941  * already and remember the confirmed_flush LSN value.
1942  */
1943  SpinLockAcquire(&slot->mutex);
1944  if (!slot->just_dirtied)
1945  slot->dirty = false;
1947  SpinLockRelease(&slot->mutex);
1948 
1950 }
#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 376 of file slot.c.

377 {
378  int i;
379  ReplicationSlot *slot = NULL;
380 
381  if (need_lock)
382  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
383 
384  for (i = 0; i < max_replication_slots; i++)
385  {
387 
388  if (s->in_use && strcmp(name, NameStr(s->data.name)) == 0)
389  {
390  slot = s;
391  break;
392  }
393  }
394 
395  if (need_lock)
396  LWLockRelease(ReplicationSlotControlLock);
397 
398  return slot;
399 }

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(), and ReplicationSlotAcquire().

◆ StartupReplicationSlots()

void StartupReplicationSlots ( void  )

Definition at line 1679 of file slot.c.

1680 {
1681  DIR *replication_dir;
1682  struct dirent *replication_de;
1683 
1684  elog(DEBUG1, "starting up replication slots");
1685 
1686  /* restore all slots by iterating over all on-disk entries */
1687  replication_dir = AllocateDir("pg_replslot");
1688  while ((replication_de = ReadDir(replication_dir, "pg_replslot")) != NULL)
1689  {
1690  char path[MAXPGPATH + 12];
1691  PGFileType de_type;
1692 
1693  if (strcmp(replication_de->d_name, ".") == 0 ||
1694  strcmp(replication_de->d_name, "..") == 0)
1695  continue;
1696 
1697  snprintf(path, sizeof(path), "pg_replslot/%s", replication_de->d_name);
1698  de_type = get_dirent_type(path, replication_de, false, DEBUG1);
1699 
1700  /* we're only creating directories here, skip if it's not our's */
1701  if (de_type != PGFILETYPE_ERROR && de_type != PGFILETYPE_DIR)
1702  continue;
1703 
1704  /* we crashed while a slot was being setup or deleted, clean up */
1705  if (pg_str_endswith(replication_de->d_name, ".tmp"))
1706  {
1707  if (!rmtree(path, true))
1708  {
1709  ereport(WARNING,
1710  (errmsg("could not remove directory \"%s\"",
1711  path)));
1712  continue;
1713  }
1714  fsync_fname("pg_replslot", true);
1715  continue;
1716  }
1717 
1718  /* looks like a slot in a normal state, restore */
1719  RestoreSlotFromDisk(replication_de->d_name);
1720  }
1721  FreeDir(replication_dir);
1722 
1723  /* currently no slots exist, we're done. */
1724  if (max_replication_slots <= 0)
1725  return;
1726 
1727  /* Now that we have recovered all the data, compute replication xmin */
1730 }
struct dirent * ReadDir(DIR *dir, const char *dirname)
Definition: fd.c:2879
int FreeDir(DIR *dir)
Definition: fd.c:2931
DIR * AllocateDir(const char *dirname)
Definition: fd.c:2813
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:1956
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().

Variable Documentation

◆ max_replication_slots

◆ MyReplicationSlot

◆ ReplicationSlotCtl