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 "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   2 /* 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 *path, 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)
 
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 bool InvalidatePossiblyObsoleteSlot (ReplicationSlot *s, XLogRecPtr oldestLSN, bool *invalidated)
 
bool InvalidateObsoleteReplicationSlots (XLogSegNo oldestSegno)
 
void CheckPointReplicationSlots (void)
 
void StartupReplicationSlots (void)
 

Variables

ReplicationSlotCtlDataReplicationSlotCtl = NULL
 
ReplicationSlotMyReplicationSlot = NULL
 
int max_replication_slots = 0
 

Macro Definition Documentation

◆ ReplicationSlotOnDiskChecksummedSize

#define ReplicationSlotOnDiskChecksummedSize    sizeof(ReplicationSlotOnDisk) - ReplicationSlotOnDiskNotChecksummedSize

Definition at line 84 of file slot.c.

◆ ReplicationSlotOnDiskConstantSize

#define ReplicationSlotOnDiskConstantSize    offsetof(ReplicationSlotOnDisk, slotdata)

Definition at line 78 of file slot.c.

◆ ReplicationSlotOnDiskNotChecksummedSize

#define ReplicationSlotOnDiskNotChecksummedSize    offsetof(ReplicationSlotOnDisk, version)

Definition at line 81 of file slot.c.

◆ ReplicationSlotOnDiskV2Size

#define ReplicationSlotOnDiskV2Size    sizeof(ReplicationSlotOnDisk) - ReplicationSlotOnDiskConstantSize

Definition at line 87 of file slot.c.

◆ SLOT_MAGIC

#define SLOT_MAGIC   0x1051CA1 /* format identifier */

Definition at line 90 of file slot.c.

◆ SLOT_VERSION

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

Definition at line 91 of file slot.c.

Typedef Documentation

◆ ReplicationSlotOnDisk

Function Documentation

◆ CheckPointReplicationSlots()

void CheckPointReplicationSlots ( void  )

Definition at line 1399 of file slot.c.

1400 {
1401  int i;
1402 
1403  elog(DEBUG1, "performing replication slot checkpoint");
1404 
1405  /*
1406  * Prevent any slot from being created/dropped while we're active. As we
1407  * explicitly do *not* want to block iterating over replication_slots or
1408  * acquiring a slot we cannot take the control lock - but that's OK,
1409  * because holding ReplicationSlotAllocationLock is strictly stronger, and
1410  * enough to guarantee that nobody can change the in_use bits on us.
1411  */
1412  LWLockAcquire(ReplicationSlotAllocationLock, LW_SHARED);
1413 
1414  for (i = 0; i < max_replication_slots; i++)
1415  {
1417  char path[MAXPGPATH];
1418 
1419  if (!s->in_use)
1420  continue;
1421 
1422  /* save the slot to disk, locking is handled in SaveSlotToPath() */
1423  sprintf(path, "pg_replslot/%s", NameStr(s->data.name));
1424  SaveSlotToPath(s, path, LOG);
1425  }
1426  LWLockRelease(ReplicationSlotAllocationLock);
1427 }
#define NameStr(name)
Definition: c.h:692
#define LOG
Definition: elog.h:25
#define DEBUG1
Definition: elog.h:24
int i
Definition: isn.c:73
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1196
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1800
@ LW_SHARED
Definition: lwlock.h:105
#define MAXPGPATH
#define sprintf
Definition: port.h:227
int max_replication_slots
Definition: slot.c:100
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:94
static void SaveSlotToPath(ReplicationSlot *slot, const char *path, int elevel)
Definition: slot.c:1555
ReplicationSlot replication_slots[1]
Definition: slot.h:180
bool in_use
Definition: slot.h:123
ReplicationSlotPersistentData data
Definition: slot.h:147

References ReplicationSlot::data, DEBUG1, elog(), i, ReplicationSlot::in_use, LOG, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_replication_slots, MAXPGPATH, ReplicationSlotPersistentData::name, NameStr, ReplicationSlotCtlData::replication_slots, ReplicationSlotCtl, SaveSlotToPath(), and sprintf.

Referenced by CheckPointGuts().

◆ CheckSlotPermissions()

void CheckSlotPermissions ( void  )

Definition at line 1105 of file slot.c.

1106 {
1107  if (!superuser() && !has_rolreplication(GetUserId()))
1108  ereport(ERROR,
1109  (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
1110  errmsg("must be superuser or replication role to use replication slots")));
1111 }
int errcode(int sqlerrcode)
Definition: elog.c:693
int errmsg(const char *fmt,...)
Definition: elog.c:904
#define ERROR
Definition: elog.h:33
#define ereport(elevel,...)
Definition: elog.h:143
Oid GetUserId(void)
Definition: miscinit.c:491
bool has_rolreplication(Oid roleid)
Definition: miscinit.c:678
bool superuser(void)
Definition: superuser.c:46

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

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

1084 {
1085  /*
1086  * NB: Adding a new requirement likely means that RestoreSlotFromDisk()
1087  * needs the same check.
1088  */
1089 
1090  if (max_replication_slots == 0)
1091  ereport(ERROR,
1092  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1093  errmsg("replication slots can only be used if max_replication_slots > 0")));
1094 
1096  ereport(ERROR,
1097  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1098  errmsg("replication slots can only be used if wal_level >= replica")));
1099 }
int wal_level
Definition: xlog.c:132
@ WAL_LEVEL_REPLICA
Definition: xlog.h:70

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

1495 {
1496  char tmppath[MAXPGPATH];
1497  char path[MAXPGPATH];
1498  struct stat st;
1499 
1500  /*
1501  * No need to take out the io_in_progress_lock, nobody else can see this
1502  * slot yet, so nobody else will write. We're reusing SaveSlotToPath which
1503  * takes out the lock, if we'd take the lock here, we'd deadlock.
1504  */
1505 
1506  sprintf(path, "pg_replslot/%s", NameStr(slot->data.name));
1507  sprintf(tmppath, "pg_replslot/%s.tmp", NameStr(slot->data.name));
1508 
1509  /*
1510  * It's just barely possible that some previous effort to create or drop a
1511  * slot with this name left a temp directory lying around. If that seems
1512  * to be the case, try to remove it. If the rmtree() fails, we'll error
1513  * out at the MakePGDirectory() below, so we don't bother checking
1514  * success.
1515  */
1516  if (stat(tmppath, &st) == 0 && S_ISDIR(st.st_mode))
1517  rmtree(tmppath, true);
1518 
1519  /* Create and fsync the temporary slot directory. */
1520  if (MakePGDirectory(tmppath) < 0)
1521  ereport(ERROR,
1523  errmsg("could not create directory \"%s\": %m",
1524  tmppath)));
1525  fsync_fname(tmppath, true);
1526 
1527  /* Write the actual state file. */
1528  slot->dirty = true; /* signal that we really need to write */
1529  SaveSlotToPath(slot, tmppath, ERROR);
1530 
1531  /* Rename the directory into place. */
1532  if (rename(tmppath, path) != 0)
1533  ereport(ERROR,
1535  errmsg("could not rename file \"%s\" to \"%s\": %m",
1536  tmppath, path)));
1537 
1538  /*
1539  * If we'd now fail - really unlikely - we wouldn't know whether this slot
1540  * would persist after an OS crash or not - so, force a restart. The
1541  * restart would try to fsync this again till it works.
1542  */
1544 
1545  fsync_fname(path, true);
1546  fsync_fname("pg_replslot", true);
1547 
1548  END_CRIT_SECTION();
1549 }
int errcode_for_file_access(void)
Definition: elog.c:716
int MakePGDirectory(const char *directoryName)
Definition: fd.c:3740
void fsync_fname(const char *fname, bool isdir)
Definition: fd.c:673
#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:42
bool dirty
Definition: slot.h:130
#define stat
Definition: win32_port.h:283
#define S_ISDIR(m)
Definition: win32_port.h:324

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 ( XLogSegNo  oldestSegno)

Definition at line 1356 of file slot.c.

1357 {
1358  XLogRecPtr oldestLSN;
1359  bool invalidated = false;
1360 
1361  XLogSegNoOffsetToRecPtr(oldestSegno, 0, wal_segment_size, oldestLSN);
1362 
1363 restart:
1364  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
1365  for (int i = 0; i < max_replication_slots; i++)
1366  {
1368 
1369  if (!s->in_use)
1370  continue;
1371 
1372  if (InvalidatePossiblyObsoleteSlot(s, oldestLSN, &invalidated))
1373  {
1374  /* if the lock was released, start from scratch */
1375  goto restart;
1376  }
1377  }
1378  LWLockRelease(ReplicationSlotControlLock);
1379 
1380  /*
1381  * If any slots have been invalidated, recalculate the resource limits.
1382  */
1383  if (invalidated)
1384  {
1387  }
1388 
1389  return invalidated;
1390 }
void ReplicationSlotsComputeRequiredXmin(bool already_locked)
Definition: slot.c:806
void ReplicationSlotsComputeRequiredLSN(void)
Definition: slot.c:856
static bool InvalidatePossiblyObsoleteSlot(ReplicationSlot *s, XLogRecPtr oldestLSN, bool *invalidated)
Definition: slot.c:1203
int wal_segment_size
Definition: xlog.c:144
#define XLogSegNoOffsetToRecPtr(segno, offset, wal_segsz_bytes, dest)
uint64 XLogRecPtr
Definition: xlogdefs.h:21

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

Referenced by CreateCheckPoint(), and CreateRestartPoint().

◆ InvalidatePossiblyObsoleteSlot()

static bool InvalidatePossiblyObsoleteSlot ( ReplicationSlot s,
XLogRecPtr  oldestLSN,
bool invalidated 
)
static

Definition at line 1203 of file slot.c.

1205 {
1206  int last_signaled_pid = 0;
1207  bool released_lock = false;
1208 
1209  for (;;)
1210  {
1211  XLogRecPtr restart_lsn;
1212  NameData slotname;
1213  int active_pid = 0;
1214 
1215  Assert(LWLockHeldByMeInMode(ReplicationSlotControlLock, LW_SHARED));
1216 
1217  if (!s->in_use)
1218  {
1219  if (released_lock)
1220  LWLockRelease(ReplicationSlotControlLock);
1221  break;
1222  }
1223 
1224  /*
1225  * Check if the slot needs to be invalidated. If it needs to be
1226  * invalidated, and is not currently acquired, acquire it and mark it
1227  * as having been invalidated. We do this with the spinlock held to
1228  * avoid race conditions -- for example the restart_lsn could move
1229  * forward, or the slot could be dropped.
1230  */
1231  SpinLockAcquire(&s->mutex);
1232 
1233  restart_lsn = s->data.restart_lsn;
1234 
1235  /*
1236  * If the slot is already invalid or is fresh enough, we don't need to
1237  * do anything.
1238  */
1239  if (XLogRecPtrIsInvalid(restart_lsn) || restart_lsn >= oldestLSN)
1240  {
1241  SpinLockRelease(&s->mutex);
1242  if (released_lock)
1243  LWLockRelease(ReplicationSlotControlLock);
1244  break;
1245  }
1246 
1247  slotname = s->data.name;
1248  active_pid = s->active_pid;
1249 
1250  /*
1251  * If the slot can be acquired, do so and mark it invalidated
1252  * immediately. Otherwise we'll signal the owning process, below, and
1253  * retry.
1254  */
1255  if (active_pid == 0)
1256  {
1257  MyReplicationSlot = s;
1258  s->active_pid = MyProcPid;
1259  s->data.invalidated_at = restart_lsn;
1261 
1262  /* Let caller know */
1263  *invalidated = true;
1264  }
1265 
1266  SpinLockRelease(&s->mutex);
1267 
1268  if (active_pid != 0)
1269  {
1270  /*
1271  * Prepare the sleep on the slot's condition variable before
1272  * releasing the lock, to close a possible race condition if the
1273  * slot is released before the sleep below.
1274  */
1276 
1277  LWLockRelease(ReplicationSlotControlLock);
1278  released_lock = true;
1279 
1280  /*
1281  * Signal to terminate the process that owns the slot, if we
1282  * haven't already signalled it. (Avoidance of repeated
1283  * signalling is the only reason for there to be a loop in this
1284  * routine; otherwise we could rely on caller's restart loop.)
1285  *
1286  * There is the race condition that other process may own the slot
1287  * after its current owner process is terminated and before this
1288  * process owns it. To handle that, we signal only if the PID of
1289  * the owning process has changed from the previous time. (This
1290  * logic assumes that the same PID is not reused very quickly.)
1291  */
1292  if (last_signaled_pid != active_pid)
1293  {
1294  ereport(LOG,
1295  (errmsg("terminating process %d to release replication slot \"%s\"",
1296  active_pid, NameStr(slotname))));
1297 
1298  (void) kill(active_pid, SIGTERM);
1299  last_signaled_pid = active_pid;
1300  }
1301 
1302  /* Wait until the slot is released. */
1305 
1306  /*
1307  * Re-acquire lock and start over; we expect to invalidate the
1308  * slot next time (unless another process acquires the slot in the
1309  * meantime).
1310  */
1311  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
1312  continue;
1313  }
1314  else
1315  {
1316  /*
1317  * We hold the slot now and have already invalidated it; flush it
1318  * to ensure that state persists.
1319  *
1320  * Don't want to hold ReplicationSlotControlLock across file
1321  * system operations, so release it now but be sure to tell caller
1322  * to restart from scratch.
1323  */
1324  LWLockRelease(ReplicationSlotControlLock);
1325  released_lock = true;
1326 
1327  /* Make sure the invalidated state persists across server restart */
1331 
1332  ereport(LOG,
1333  (errmsg("invalidating slot \"%s\" because its restart_lsn %X/%X exceeds max_slot_wal_keep_size",
1334  NameStr(slotname),
1335  LSN_FORMAT_ARGS(restart_lsn))));
1336 
1337  /* done with this slot for now */
1338  break;
1339  }
1340  }
1341 
1342  Assert(released_lock == !LWLockHeldByMe(ReplicationSlotControlLock));
1343 
1344  return released_lock;
1345 }
void ConditionVariablePrepareToSleep(ConditionVariable *cv)
void ConditionVariableSleep(ConditionVariable *cv, uint32 wait_event_info)
int MyProcPid
Definition: globals.c:44
Assert(fmt[strlen(fmt) - 1] !='\n')
bool LWLockHeldByMeInMode(LWLock *l, LWLockMode mode)
Definition: lwlock.c:1934
bool LWLockHeldByMe(LWLock *l)
Definition: lwlock.c:1916
void ReplicationSlotMarkDirty(void)
Definition: slot.c:767
ReplicationSlot * MyReplicationSlot
Definition: slot.c:97
void ReplicationSlotSave(void)
Definition: slot.c:749
void ReplicationSlotRelease(void)
Definition: slot.c:518
#define SpinLockRelease(lock)
Definition: spin.h:64
#define SpinLockAcquire(lock)
Definition: spin.h:62
XLogRecPtr restart_lsn
Definition: slot.h:73
XLogRecPtr invalidated_at
Definition: slot.h:76
slock_t mutex
Definition: slot.h:120
pid_t active_pid
Definition: slot.h:126
ConditionVariable active_cv
Definition: slot.h:153
Definition: c.h:687
@ WAIT_EVENT_REPLICATION_SLOT_DROP
Definition: wait_event.h:125
#define kill(pid, sig)
Definition: win32_port.h:464
#define LSN_FORMAT_ARGS(lsn)
Definition: xlogdefs.h:43
#define XLogRecPtrIsInvalid(r)
Definition: xlogdefs.h:29
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28

References ReplicationSlot::active_cv, ReplicationSlot::active_pid, Assert(), ConditionVariablePrepareToSleep(), ConditionVariableSleep(), ReplicationSlot::data, ereport, errmsg(), ReplicationSlot::in_use, ReplicationSlotPersistentData::invalidated_at, InvalidXLogRecPtr, kill, LOG, LSN_FORMAT_ARGS, LW_SHARED, LWLockAcquire(), LWLockHeldByMe(), LWLockHeldByMeInMode(), LWLockRelease(), ReplicationSlot::mutex, MyProcPid, MyReplicationSlot, ReplicationSlotPersistentData::name, NameStr, ReplicationSlotMarkDirty(), ReplicationSlotRelease(), ReplicationSlotSave(), ReplicationSlotPersistentData::restart_lsn, SpinLockAcquire, SpinLockRelease, WAIT_EVENT_REPLICATION_SLOT_DROP, and XLogRecPtrIsInvalid.

Referenced by InvalidateObsoleteReplicationSlots().

◆ ReplicationSlotAcquire()

void ReplicationSlotAcquire ( const char *  name,
bool  nowait 
)

Definition at line 421 of file slot.c.

422 {
423  ReplicationSlot *s;
424  int active_pid;
425 
426  AssertArg(name != NULL);
427 
428 retry:
429  Assert(MyReplicationSlot == NULL);
430 
431  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
432 
433  /*
434  * Search for the slot with the specified name if the slot to acquire is
435  * not given. If the slot is not found, we either return -1 or error out.
436  */
437  s = SearchNamedReplicationSlot(name, false);
438  if (s == NULL || !s->in_use)
439  {
440  LWLockRelease(ReplicationSlotControlLock);
441 
442  ereport(ERROR,
443  (errcode(ERRCODE_UNDEFINED_OBJECT),
444  errmsg("replication slot \"%s\" does not exist",
445  name)));
446  }
447 
448  /*
449  * This is the slot we want; check if it's active under some other
450  * process. In single user mode, we don't need this check.
451  */
452  if (IsUnderPostmaster)
453  {
454  /*
455  * Get ready to sleep on the slot in case it is active. (We may end
456  * up not sleeping, but we don't want to do this while holding the
457  * spinlock.)
458  */
459  if (!nowait)
461 
462  SpinLockAcquire(&s->mutex);
463  if (s->active_pid == 0)
464  s->active_pid = MyProcPid;
465  active_pid = s->active_pid;
466  SpinLockRelease(&s->mutex);
467  }
468  else
469  active_pid = MyProcPid;
470  LWLockRelease(ReplicationSlotControlLock);
471 
472  /*
473  * If we found the slot but it's already active in another process, we
474  * wait until the owning process signals us that it's been released, or
475  * error out.
476  */
477  if (active_pid != MyProcPid)
478  {
479  if (!nowait)
480  {
481  /* Wait here until we get signaled, and then restart */
485  goto retry;
486  }
487 
488  ereport(ERROR,
489  (errcode(ERRCODE_OBJECT_IN_USE),
490  errmsg("replication slot \"%s\" is active for PID %d",
491  NameStr(s->data.name), active_pid)));
492  }
493  else if (!nowait)
494  ConditionVariableCancelSleep(); /* no sleep needed after all */
495 
496  /* Let everybody know we've modified this slot */
498 
499  /* We made this slot active, so it's ours now. */
500  MyReplicationSlot = s;
501 
502  /*
503  * The call to pgstat_acquire_replslot() protects against stats for a
504  * different slot, from before a restart or such, being present during
505  * pgstat_report_replslot().
506  */
507  if (SlotIsLogical(s))
509 }
#define AssertArg(condition)
Definition: c.h:817
void ConditionVariableBroadcast(ConditionVariable *cv)
void ConditionVariableCancelSleep(void)
const char * name
Definition: encode.c:561
bool IsUnderPostmaster
Definition: globals.c:113
void pgstat_acquire_replslot(ReplicationSlot *slot)
ReplicationSlot * SearchNamedReplicationSlot(const char *name, bool need_lock)
Definition: slot.c:373
#define SlotIsLogical(slot)
Definition: slot.h:169

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

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

◆ ReplicationSlotCleanup()

void ReplicationSlotCleanup ( void  )

Definition at line 574 of file slot.c.

575 {
576  int i;
577 
578  Assert(MyReplicationSlot == NULL);
579 
580 restart:
581  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
582  for (i = 0; i < max_replication_slots; i++)
583  {
585 
586  if (!s->in_use)
587  continue;
588 
589  SpinLockAcquire(&s->mutex);
590  if (s->active_pid == MyProcPid)
591  {
593  SpinLockRelease(&s->mutex);
594  LWLockRelease(ReplicationSlotControlLock); /* avoid deadlock */
595 
597 
599  goto restart;
600  }
601  else
602  SpinLockRelease(&s->mutex);
603  }
604 
605  LWLockRelease(ReplicationSlotControlLock);
606 }
static void ReplicationSlotDropPtr(ReplicationSlot *slot)
Definition: slot.c:642
@ RS_TEMPORARY
Definition: slot.h:37
ReplicationSlotPersistency persistency
Definition: slot.h:54

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

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

613 {
614  Assert(MyReplicationSlot == NULL);
615 
616  ReplicationSlotAcquire(name, nowait);
617 
619 }
void ReplicationSlotAcquire(const char *name, bool nowait)
Definition: slot.c:421
static void ReplicationSlotDropAcquired(void)
Definition: slot.c:625

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

Referenced by DropReplicationSlot(), and pg_drop_replication_slot().

◆ ReplicationSlotDropAcquired()

static void ReplicationSlotDropAcquired ( void  )
static

Definition at line 625 of file slot.c.

626 {
628 
629  Assert(MyReplicationSlot != NULL);
630 
631  /* slot isn't acquired anymore */
632  MyReplicationSlot = NULL;
633 
635 }

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

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

◆ ReplicationSlotDropPtr()

static void ReplicationSlotDropPtr ( ReplicationSlot slot)
static

Definition at line 642 of file slot.c.

643 {
644  char path[MAXPGPATH];
645  char tmppath[MAXPGPATH];
646 
647  /*
648  * If some other backend ran this code concurrently with us, we might try
649  * to delete a slot with a certain name while someone else was trying to
650  * create a slot with the same name.
651  */
652  LWLockAcquire(ReplicationSlotAllocationLock, LW_EXCLUSIVE);
653 
654  /* Generate pathnames. */
655  sprintf(path, "pg_replslot/%s", NameStr(slot->data.name));
656  sprintf(tmppath, "pg_replslot/%s.tmp", NameStr(slot->data.name));
657 
658  /*
659  * Rename the slot directory on disk, so that we'll no longer recognize
660  * this as a valid slot. Note that if this fails, we've got to mark the
661  * slot inactive before bailing out. If we're dropping an ephemeral or a
662  * temporary slot, we better never fail hard as the caller won't expect
663  * the slot to survive and this might get called during error handling.
664  */
665  if (rename(path, tmppath) == 0)
666  {
667  /*
668  * We need to fsync() the directory we just renamed and its parent to
669  * make sure that our changes are on disk in a crash-safe fashion. If
670  * fsync() fails, we can't be sure whether the changes are on disk or
671  * not. For now, we handle that by panicking;
672  * StartupReplicationSlots() will try to straighten it out after
673  * restart.
674  */
676  fsync_fname(tmppath, true);
677  fsync_fname("pg_replslot", true);
679  }
680  else
681  {
682  bool fail_softly = slot->data.persistency != RS_PERSISTENT;
683 
684  SpinLockAcquire(&slot->mutex);
685  slot->active_pid = 0;
686  SpinLockRelease(&slot->mutex);
687 
688  /* wake up anyone waiting on this slot */
690 
691  ereport(fail_softly ? WARNING : ERROR,
693  errmsg("could not rename file \"%s\" to \"%s\": %m",
694  path, tmppath)));
695  }
696 
697  /*
698  * The slot is definitely gone. Lock out concurrent scans of the array
699  * long enough to kill it. It's OK to clear the active PID here without
700  * grabbing the mutex because nobody else can be scanning the array here,
701  * and nobody can be attached to this slot and thus access it without
702  * scanning the array.
703  *
704  * Also wake up processes waiting for it.
705  */
706  LWLockAcquire(ReplicationSlotControlLock, LW_EXCLUSIVE);
707  slot->active_pid = 0;
708  slot->in_use = false;
709  LWLockRelease(ReplicationSlotControlLock);
711 
712  /*
713  * Slot is dead and doesn't prevent resource removal anymore, recompute
714  * limits.
715  */
718 
719  /*
720  * If removing the directory fails, the worst thing that will happen is
721  * that the user won't be able to create a new slot with the same name
722  * until the next server restart. We warn about it, but that's all.
723  */
724  if (!rmtree(tmppath, true))
726  (errmsg("could not remove directory \"%s\"", tmppath)));
727 
728  /*
729  * Drop the statistics entry for the replication slot. Do this while
730  * holding ReplicationSlotAllocationLock so that we don't drop a
731  * statistics entry for another slot with the same name just created in
732  * another session.
733  */
734  if (SlotIsLogical(slot))
735  pgstat_drop_replslot(slot);
736 
737  /*
738  * We release this at the very end, so that nobody starts trying to create
739  * a slot while we're still cleaning up the detritus of the old one.
740  */
741  LWLockRelease(ReplicationSlotAllocationLock);
742 }
#define WARNING
Definition: elog.h:30
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 169 of file slot.c.

170 {
172 }
void before_shmem_exit(pg_on_exit_callback function, Datum arg)
Definition: ipc.c:333
static void ReplicationSlotShmemExit(int code, Datum arg)
Definition: slot.c:178

References before_shmem_exit(), and ReplicationSlotShmemExit().

Referenced by BaseInit().

◆ ReplicationSlotMarkDirty()

◆ ReplicationSlotPersist()

◆ ReplicationSlotRelease()

void ReplicationSlotRelease ( void  )

Definition at line 518 of file slot.c.

519 {
521 
522  Assert(slot != NULL && slot->active_pid != 0);
523 
524  if (slot->data.persistency == RS_EPHEMERAL)
525  {
526  /*
527  * Delete the slot. There is no !PANIC case where this is allowed to
528  * fail, all that may happen is an incomplete cleanup of the on-disk
529  * data.
530  */
532  }
533 
534  /*
535  * If slot needed to temporarily restrain both data and catalog xmin to
536  * create the catalog snapshot, remove that temporary constraint.
537  * Snapshots can only be exported while the initial snapshot is still
538  * acquired.
539  */
540  if (!TransactionIdIsValid(slot->data.xmin) &&
542  {
543  SpinLockAcquire(&slot->mutex);
545  SpinLockRelease(&slot->mutex);
547  }
548 
549  if (slot->data.persistency == RS_PERSISTENT)
550  {
551  /*
552  * Mark persistent slot inactive. We're not freeing it, just
553  * disconnecting, but wake up others that may be waiting for it.
554  */
555  SpinLockAcquire(&slot->mutex);
556  slot->active_pid = 0;
557  SpinLockRelease(&slot->mutex);
559  }
560 
561  MyReplicationSlot = NULL;
562 
563  /* might not have been set when we've been a plain slot */
564  LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
567  LWLockRelease(ProcArrayLock);
568 }
#define PROC_IN_LOGICAL_DECODING
Definition: proc.h:58
@ RS_EPHEMERAL
Definition: slot.h:36
PGPROC * MyProc
Definition: proc.c:68
PROC_HDR * ProcGlobal
Definition: proc.c:80
uint8 statusFlags
Definition: proc.h:227
int pgxactoff
Definition: proc.h:186
uint8 * statusFlags
Definition: proc.h:371
TransactionId xmin
Definition: slot.h:62
#define TransactionIdIsValid(xid)
Definition: transam.h:41

References ReplicationSlot::active_cv, ReplicationSlot::active_pid, Assert(), ConditionVariableBroadcast(), ReplicationSlot::data, ReplicationSlot::effective_xmin, InvalidTransactionId, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), ReplicationSlot::mutex, MyProc, MyReplicationSlot, ReplicationSlotPersistentData::persistency, PGPROC::pgxactoff, PROC_IN_LOGICAL_DECODING, ProcGlobal, ReplicationSlotDropAcquired(), ReplicationSlotsComputeRequiredXmin(), RS_EPHEMERAL, RS_PERSISTENT, SpinLockAcquire, SpinLockRelease, PGPROC::statusFlags, PROC_HDR::statusFlags, TransactionIdIsValid, and ReplicationSlotPersistentData::xmin.

Referenced by 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 1120 of file slot.c.

1121 {
1123 
1124  Assert(slot != NULL);
1126 
1127  /*
1128  * The replication slot mechanism is used to prevent removal of required
1129  * WAL. As there is no interlock between this routine and checkpoints, WAL
1130  * segments could concurrently be removed when a now stale return value of
1131  * ReplicationSlotsComputeRequiredLSN() is used. In the unlikely case that
1132  * this happens we'll just retry.
1133  */
1134  while (true)
1135  {
1136  XLogSegNo segno;
1137  XLogRecPtr restart_lsn;
1138 
1139  /*
1140  * For logical slots log a standby snapshot and start logical decoding
1141  * at exactly that position. That allows the slot to start up more
1142  * quickly.
1143  *
1144  * That's not needed (or indeed helpful) for physical slots as they'll
1145  * start replay at the last logged checkpoint anyway. Instead return
1146  * the location of the last redo LSN. While that slightly increases
1147  * the chance that we have to retry, it's where a base backup has to
1148  * start replay at.
1149  */
1150  if (!RecoveryInProgress() && SlotIsLogical(slot))
1151  {
1152  XLogRecPtr flushptr;
1153 
1154  /* start at current insert position */
1155  restart_lsn = GetXLogInsertRecPtr();
1156  SpinLockAcquire(&slot->mutex);
1157  slot->data.restart_lsn = restart_lsn;
1158  SpinLockRelease(&slot->mutex);
1159 
1160  /* make sure we have enough information to start */
1161  flushptr = LogStandbySnapshot();
1162 
1163  /* and make sure it's fsynced to disk */
1164  XLogFlush(flushptr);
1165  }
1166  else
1167  {
1168  restart_lsn = GetRedoRecPtr();
1169  SpinLockAcquire(&slot->mutex);
1170  slot->data.restart_lsn = restart_lsn;
1171  SpinLockRelease(&slot->mutex);
1172  }
1173 
1174  /* prevent WAL removal as fast as possible */
1176 
1177  /*
1178  * If all required WAL is still there, great, otherwise retry. The
1179  * slot should prevent further removal of WAL, unless there's a
1180  * concurrent ReplicationSlotsComputeRequiredLSN() after we've written
1181  * the new restart_lsn above, so normally we should never need to loop
1182  * more than twice.
1183  */
1185  if (XLogGetLastRemovedSegno() < segno)
1186  break;
1187  }
1188 }
XLogRecPtr LogStandbySnapshot(void)
Definition: standby.c:1222
bool RecoveryInProgress(void)
Definition: xlog.c:5759
XLogSegNo XLogGetLastRemovedSegno(void)
Definition: xlog.c:3482
XLogRecPtr GetRedoRecPtr(void)
Definition: xlog.c:5862
XLogRecPtr GetXLogInsertRecPtr(void)
Definition: xlog.c:8809
void XLogFlush(XLogRecPtr record)
Definition: xlog.c:2509
#define XLByteToSeg(xlrp, logSegNo, wal_segsz_bytes)
uint64 XLogSegNo
Definition: xlogdefs.h:48

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

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

◆ ReplicationSlotSave()

◆ ReplicationSlotsComputeLogicalRestartLSN()

XLogRecPtr ReplicationSlotsComputeLogicalRestartLSN ( void  )

Definition at line 899 of file slot.c.

900 {
901  XLogRecPtr result = InvalidXLogRecPtr;
902  int i;
903 
904  if (max_replication_slots <= 0)
905  return InvalidXLogRecPtr;
906 
907  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
908 
909  for (i = 0; i < max_replication_slots; i++)
910  {
911  ReplicationSlot *s;
912  XLogRecPtr restart_lsn;
913 
915 
916  /* cannot change while ReplicationSlotCtlLock is held */
917  if (!s->in_use)
918  continue;
919 
920  /* we're only interested in logical slots */
921  if (!SlotIsLogical(s))
922  continue;
923 
924  /* read once, it's ok if it increases while we're checking */
925  SpinLockAcquire(&s->mutex);
926  restart_lsn = s->data.restart_lsn;
927  SpinLockRelease(&s->mutex);
928 
929  if (restart_lsn == InvalidXLogRecPtr)
930  continue;
931 
932  if (result == InvalidXLogRecPtr ||
933  restart_lsn < result)
934  result = restart_lsn;
935  }
936 
937  LWLockRelease(ReplicationSlotControlLock);
938 
939  return result;
940 }

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

Referenced by CheckPointLogicalRewriteHeap(), and CheckPointSnapBuild().

◆ ReplicationSlotsComputeRequiredLSN()

void ReplicationSlotsComputeRequiredLSN ( void  )

Definition at line 856 of file slot.c.

857 {
858  int i;
859  XLogRecPtr min_required = InvalidXLogRecPtr;
860 
861  Assert(ReplicationSlotCtl != NULL);
862 
863  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
864  for (i = 0; i < max_replication_slots; i++)
865  {
867  XLogRecPtr restart_lsn;
868 
869  if (!s->in_use)
870  continue;
871 
872  SpinLockAcquire(&s->mutex);
873  restart_lsn = s->data.restart_lsn;
874  SpinLockRelease(&s->mutex);
875 
876  if (restart_lsn != InvalidXLogRecPtr &&
877  (min_required == InvalidXLogRecPtr ||
878  restart_lsn < min_required))
879  min_required = restart_lsn;
880  }
881  LWLockRelease(ReplicationSlotControlLock);
882 
883  XLogSetReplicationSlotMinimumLSN(min_required);
884 }
void XLogSetReplicationSlotMinimumLSN(XLogRecPtr lsn)
Definition: xlog.c:2395

References Assert(), ReplicationSlot::data, i, ReplicationSlot::in_use, InvalidXLogRecPtr, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_replication_slots, ReplicationSlot::mutex, ReplicationSlotCtlData::replication_slots, ReplicationSlotCtl, ReplicationSlotPersistentData::restart_lsn, 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 806 of file slot.c.

807 {
808  int i;
810  TransactionId agg_catalog_xmin = InvalidTransactionId;
811 
812  Assert(ReplicationSlotCtl != NULL);
813 
814  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
815 
816  for (i = 0; i < max_replication_slots; i++)
817  {
819  TransactionId effective_xmin;
820  TransactionId effective_catalog_xmin;
821 
822  if (!s->in_use)
823  continue;
824 
825  SpinLockAcquire(&s->mutex);
826  effective_xmin = s->effective_xmin;
827  effective_catalog_xmin = s->effective_catalog_xmin;
828  SpinLockRelease(&s->mutex);
829 
830  /* check the data xmin */
831  if (TransactionIdIsValid(effective_xmin) &&
832  (!TransactionIdIsValid(agg_xmin) ||
833  TransactionIdPrecedes(effective_xmin, agg_xmin)))
834  agg_xmin = effective_xmin;
835 
836  /* check the catalog xmin */
837  if (TransactionIdIsValid(effective_catalog_xmin) &&
838  (!TransactionIdIsValid(agg_catalog_xmin) ||
839  TransactionIdPrecedes(effective_catalog_xmin, agg_catalog_xmin)))
840  agg_catalog_xmin = effective_catalog_xmin;
841  }
842 
843  LWLockRelease(ReplicationSlotControlLock);
844 
845  ProcArraySetReplicationSlotXmin(agg_xmin, agg_catalog_xmin, already_locked);
846 }
uint32 TransactionId
Definition: c.h:598
void ProcArraySetReplicationSlotXmin(TransactionId xmin, TransactionId catalog_xmin, bool already_locked)
Definition: procarray.c:3890
bool TransactionIdPrecedes(TransactionId id1, TransactionId id2)
Definition: transam.c:273

References Assert(), ReplicationSlot::effective_catalog_xmin, ReplicationSlot::effective_xmin, i, ReplicationSlot::in_use, InvalidTransactionId, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_replication_slots, ReplicationSlot::mutex, ProcArraySetReplicationSlotXmin(), ReplicationSlotCtlData::replication_slots, ReplicationSlotCtl, 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 951 of file slot.c.

952 {
953  int i;
954 
955  *nslots = *nactive = 0;
956 
957  if (max_replication_slots <= 0)
958  return false;
959 
960  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
961  for (i = 0; i < max_replication_slots; i++)
962  {
963  ReplicationSlot *s;
964 
966 
967  /* cannot change while ReplicationSlotCtlLock is held */
968  if (!s->in_use)
969  continue;
970 
971  /* only logical slots are database specific, skip */
972  if (!SlotIsLogical(s))
973  continue;
974 
975  /* not our database, skip */
976  if (s->data.database != dboid)
977  continue;
978 
979  /* count slots with spinlock held */
980  SpinLockAcquire(&s->mutex);
981  (*nslots)++;
982  if (s->active_pid != 0)
983  (*nactive)++;
984  SpinLockRelease(&s->mutex);
985  }
986  LWLockRelease(ReplicationSlotControlLock);
987 
988  if (*nslots > 0)
989  return true;
990  return false;
991 }

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

1008 {
1009  int i;
1010 
1011  if (max_replication_slots <= 0)
1012  return;
1013 
1014 restart:
1015  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
1016  for (i = 0; i < max_replication_slots; i++)
1017  {
1018  ReplicationSlot *s;
1019  char *slotname;
1020  int active_pid;
1021 
1023 
1024  /* cannot change while ReplicationSlotCtlLock is held */
1025  if (!s->in_use)
1026  continue;
1027 
1028  /* only logical slots are database specific, skip */
1029  if (!SlotIsLogical(s))
1030  continue;
1031 
1032  /* not our database, skip */
1033  if (s->data.database != dboid)
1034  continue;
1035 
1036  /* acquire slot, so ReplicationSlotDropAcquired can be reused */
1037  SpinLockAcquire(&s->mutex);
1038  /* can't change while ReplicationSlotControlLock is held */
1039  slotname = NameStr(s->data.name);
1040  active_pid = s->active_pid;
1041  if (active_pid == 0)
1042  {
1043  MyReplicationSlot = s;
1044  s->active_pid = MyProcPid;
1045  }
1046  SpinLockRelease(&s->mutex);
1047 
1048  /*
1049  * Even though we hold an exclusive lock on the database object a
1050  * logical slot for that DB can still be active, e.g. if it's
1051  * concurrently being dropped by a backend connected to another DB.
1052  *
1053  * That's fairly unlikely in practice, so we'll just bail out.
1054  */
1055  if (active_pid)
1056  ereport(ERROR,
1057  (errcode(ERRCODE_OBJECT_IN_USE),
1058  errmsg("replication slot \"%s\" is active for PID %d",
1059  slotname, active_pid)));
1060 
1061  /*
1062  * To avoid duplicating ReplicationSlotDropAcquired() and to avoid
1063  * holding ReplicationSlotControlLock over filesystem operations,
1064  * release ReplicationSlotControlLock and use
1065  * ReplicationSlotDropAcquired.
1066  *
1067  * As that means the set of slots could change, restart scan from the
1068  * beginning each time we release the lock.
1069  */
1070  LWLockRelease(ReplicationSlotControlLock);
1072  goto restart;
1073  }
1074  LWLockRelease(ReplicationSlotControlLock);
1075 }

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

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

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

Referenced by ReplicationSlotInitialize().

◆ ReplicationSlotsShmemInit()

void ReplicationSlotsShmemInit ( void  )

Definition at line 134 of file slot.c.

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

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

◆ ReplicationSlotsShmemSize()

Size ReplicationSlotsShmemSize ( void  )

Definition at line 116 of file slot.c.

117 {
118  Size size = 0;
119 
120  if (max_replication_slots == 0)
121  return size;
122 
123  size = offsetof(ReplicationSlotCtlData, replication_slots);
124  size = add_size(size,
126 
127  return size;
128 }
#define offsetof(type, field)
Definition: c.h:738
size_t Size
Definition: c.h:551
Size add_size(Size s1, Size s2)
Definition: shmem.c:502
Size mul_size(Size s1, Size s2)
Definition: shmem.c:519

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

Referenced by CalculateShmemSize(), and ReplicationSlotsShmemInit().

◆ ReplicationSlotValidateName()

bool ReplicationSlotValidateName ( const char *  name,
int  elevel 
)

Definition at line 197 of file slot.c.

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

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

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

◆ RestoreSlotFromDisk()

static void RestoreSlotFromDisk ( const char *  name)
static

Definition at line 1709 of file slot.c.

1710 {
1712  int i;
1713  char slotdir[MAXPGPATH + 12];
1714  char path[MAXPGPATH + 22];
1715  int fd;
1716  bool restored = false;
1717  int readBytes;
1718  pg_crc32c checksum;
1719 
1720  /* no need to lock here, no concurrent access allowed yet */
1721 
1722  /* delete temp file if it exists */
1723  sprintf(slotdir, "pg_replslot/%s", name);
1724  sprintf(path, "%s/state.tmp", slotdir);
1725  if (unlink(path) < 0 && errno != ENOENT)
1726  ereport(PANIC,
1728  errmsg("could not remove file \"%s\": %m", path)));
1729 
1730  sprintf(path, "%s/state", slotdir);
1731 
1732  elog(DEBUG1, "restoring replication slot from \"%s\"", path);
1733 
1734  /* on some operating systems fsyncing a file requires O_RDWR */
1735  fd = OpenTransientFile(path, O_RDWR | PG_BINARY);
1736 
1737  /*
1738  * We do not need to handle this as we are rename()ing the directory into
1739  * place only after we fsync()ed the state file.
1740  */
1741  if (fd < 0)
1742  ereport(PANIC,
1744  errmsg("could not open file \"%s\": %m", path)));
1745 
1746  /*
1747  * Sync state file before we're reading from it. We might have crashed
1748  * while it wasn't synced yet and we shouldn't continue on that basis.
1749  */
1751  if (pg_fsync(fd) != 0)
1752  ereport(PANIC,
1754  errmsg("could not fsync file \"%s\": %m",
1755  path)));
1757 
1758  /* Also sync the parent directory */
1760  fsync_fname(slotdir, true);
1761  END_CRIT_SECTION();
1762 
1763  /* read part of statefile that's guaranteed to be version independent */
1765  readBytes = read(fd, &cp, ReplicationSlotOnDiskConstantSize);
1767  if (readBytes != ReplicationSlotOnDiskConstantSize)
1768  {
1769  if (readBytes < 0)
1770  ereport(PANIC,
1772  errmsg("could not read file \"%s\": %m", path)));
1773  else
1774  ereport(PANIC,
1776  errmsg("could not read file \"%s\": read %d of %zu",
1777  path, readBytes,
1779  }
1780 
1781  /* verify magic */
1782  if (cp.magic != SLOT_MAGIC)
1783  ereport(PANIC,
1785  errmsg("replication slot file \"%s\" has wrong magic number: %u instead of %u",
1786  path, cp.magic, SLOT_MAGIC)));
1787 
1788  /* verify version */
1789  if (cp.version != SLOT_VERSION)
1790  ereport(PANIC,
1792  errmsg("replication slot file \"%s\" has unsupported version %u",
1793  path, cp.version)));
1794 
1795  /* boundary check on length */
1797  ereport(PANIC,
1799  errmsg("replication slot file \"%s\" has corrupted length %u",
1800  path, cp.length)));
1801 
1802  /* Now that we know the size, read the entire file */
1804  readBytes = read(fd,
1805  (char *) &cp + ReplicationSlotOnDiskConstantSize,
1806  cp.length);
1808  if (readBytes != cp.length)
1809  {
1810  if (readBytes < 0)
1811  ereport(PANIC,
1813  errmsg("could not read file \"%s\": %m", path)));
1814  else
1815  ereport(PANIC,
1817  errmsg("could not read file \"%s\": read %d of %zu",
1818  path, readBytes, (Size) cp.length)));
1819  }
1820 
1821  if (CloseTransientFile(fd) != 0)
1822  ereport(PANIC,
1824  errmsg("could not close file \"%s\": %m", path)));
1825 
1826  /* now verify the CRC */
1827  INIT_CRC32C(checksum);
1828  COMP_CRC32C(checksum,
1831  FIN_CRC32C(checksum);
1832 
1833  if (!EQ_CRC32C(checksum, cp.checksum))
1834  ereport(PANIC,
1835  (errmsg("checksum mismatch for replication slot file \"%s\": is %u, should be %u",
1836  path, checksum, cp.checksum)));
1837 
1838  /*
1839  * If we crashed with an ephemeral slot active, don't restore but delete
1840  * it.
1841  */
1843  {
1844  if (!rmtree(slotdir, true))
1845  {
1846  ereport(WARNING,
1847  (errmsg("could not remove directory \"%s\"",
1848  slotdir)));
1849  }
1850  fsync_fname("pg_replslot", true);
1851  return;
1852  }
1853 
1854  /*
1855  * Verify that requirements for the specific slot type are met. That's
1856  * important because if these aren't met we're not guaranteed to retain
1857  * all the necessary resources for the slot.
1858  *
1859  * NB: We have to do so *after* the above checks for ephemeral slots,
1860  * because otherwise a slot that shouldn't exist anymore could prevent
1861  * restarts.
1862  *
1863  * NB: Changing the requirements here also requires adapting
1864  * CheckSlotRequirements() and CheckLogicalDecodingRequirements().
1865  */
1867  ereport(FATAL,
1868  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1869  errmsg("logical replication slot \"%s\" exists, but wal_level < logical",
1870  NameStr(cp.slotdata.name)),
1871  errhint("Change wal_level to be logical or higher.")));
1872  else if (wal_level < WAL_LEVEL_REPLICA)
1873  ereport(FATAL,
1874  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1875  errmsg("physical replication slot \"%s\" exists, but wal_level < replica",
1876  NameStr(cp.slotdata.name)),
1877  errhint("Change wal_level to be replica or higher.")));
1878 
1879  /* nothing can be active yet, don't lock anything */
1880  for (i = 0; i < max_replication_slots; i++)
1881  {
1882  ReplicationSlot *slot;
1883 
1885 
1886  if (slot->in_use)
1887  continue;
1888 
1889  /* restore the entire set of persistent data */
1890  memcpy(&slot->data, &cp.slotdata,
1892 
1893  /* initialize in memory state */
1894  slot->effective_xmin = cp.slotdata.xmin;
1896 
1901 
1902  slot->in_use = true;
1903  slot->active_pid = 0;
1904 
1905  restored = true;
1906  break;
1907  }
1908 
1909  if (!restored)
1910  ereport(FATAL,
1911  (errmsg("too many replication slots active before shutdown"),
1912  errhint("Increase max_replication_slots and try again.")));
1913 }
#define PG_BINARY
Definition: c.h:1279
#define FATAL
Definition: elog.h:35
#define PANIC
Definition: elog.h:36
int CloseTransientFile(int fd)
Definition: fd.c:2625
int pg_fsync(int fd)
Definition: fd.c:359
int OpenTransientFile(const char *fileName, int fileFlags)
Definition: fd.c:2448
#define read(a, b, c)
Definition: win32.h:13
#define ERRCODE_DATA_CORRUPTED
Definition: pg_basebackup.c:43
uint32 pg_crc32c
Definition: pg_crc32c.h:38
#define COMP_CRC32C(crc, data, len)
Definition: pg_crc32c.h:89
#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:94
static int fd(const char *x, int i)
Definition: preproc-init.c:105
#define ReplicationSlotOnDiskChecksummedSize
Definition: slot.c:84
#define ReplicationSlotOnDiskNotChecksummedSize
Definition: slot.c:81
#define ReplicationSlotOnDiskV2Size
Definition: slot.c:87
#define SLOT_VERSION
Definition: slot.c:91
#define SLOT_MAGIC
Definition: slot.c:90
#define ReplicationSlotOnDiskConstantSize
Definition: slot.c:78
uint32 version
Definition: slot.c:66
ReplicationSlotPersistentData slotdata
Definition: slot.c:74
pg_crc32c checksum
Definition: slot.c:63
TransactionId catalog_xmin
Definition: slot.h:70
@ WAIT_EVENT_REPLICATION_SLOT_READ
Definition: wait_event.h:201
@ WAIT_EVENT_REPLICATION_SLOT_RESTORE_SYNC
Definition: wait_event.h:202
static void pgstat_report_wait_start(uint32 wait_event_info)
Definition: wait_event.h:266
static void pgstat_report_wait_end(void)
Definition: wait_event.h:282
@ WAL_LEVEL_LOGICAL
Definition: xlog.h:71

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, 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, 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, WAIT_EVENT_REPLICATION_SLOT_READ, WAIT_EVENT_REPLICATION_SLOT_RESTORE_SYNC, wal_level, WAL_LEVEL_LOGICAL, WAL_LEVEL_REPLICA, WARNING, and ReplicationSlotPersistentData::xmin.

Referenced by StartupReplicationSlots().

◆ SaveSlotToPath()

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

Definition at line 1555 of file slot.c.

1556 {
1557  char tmppath[MAXPGPATH];
1558  char path[MAXPGPATH];
1559  int fd;
1561  bool was_dirty;
1562 
1563  /* first check whether there's something to write out */
1564  SpinLockAcquire(&slot->mutex);
1565  was_dirty = slot->dirty;
1566  slot->just_dirtied = false;
1567  SpinLockRelease(&slot->mutex);
1568 
1569  /* and don't do anything if there's nothing to write */
1570  if (!was_dirty)
1571  return;
1572 
1574 
1575  /* silence valgrind :( */
1576  memset(&cp, 0, sizeof(ReplicationSlotOnDisk));
1577 
1578  sprintf(tmppath, "%s/state.tmp", dir);
1579  sprintf(path, "%s/state", dir);
1580 
1581  fd = OpenTransientFile(tmppath, O_CREAT | O_EXCL | O_WRONLY | PG_BINARY);
1582  if (fd < 0)
1583  {
1584  /*
1585  * If not an ERROR, then release the lock before returning. In case
1586  * of an ERROR, the error recovery path automatically releases the
1587  * lock, but no harm in explicitly releasing even in that case. Note
1588  * that LWLockRelease() could affect errno.
1589  */
1590  int save_errno = errno;
1591 
1593  errno = save_errno;
1594  ereport(elevel,
1596  errmsg("could not create file \"%s\": %m",
1597  tmppath)));
1598  return;
1599  }
1600 
1601  cp.magic = SLOT_MAGIC;
1602  INIT_CRC32C(cp.checksum);
1603  cp.version = SLOT_VERSION;
1605 
1606  SpinLockAcquire(&slot->mutex);
1607 
1608  memcpy(&cp.slotdata, &slot->data, sizeof(ReplicationSlotPersistentData));
1609 
1610  SpinLockRelease(&slot->mutex);
1611 
1612  COMP_CRC32C(cp.checksum,
1613  (char *) (&cp) + ReplicationSlotOnDiskNotChecksummedSize,
1615  FIN_CRC32C(cp.checksum);
1616 
1617  errno = 0;
1619  if ((write(fd, &cp, sizeof(cp))) != sizeof(cp))
1620  {
1621  int save_errno = errno;
1622 
1626 
1627  /* if write didn't set errno, assume problem is no disk space */
1628  errno = save_errno ? save_errno : ENOSPC;
1629  ereport(elevel,
1631  errmsg("could not write to file \"%s\": %m",
1632  tmppath)));
1633  return;
1634  }
1636 
1637  /* fsync the temporary file */
1639  if (pg_fsync(fd) != 0)
1640  {
1641  int save_errno = errno;
1642 
1646  errno = save_errno;
1647  ereport(elevel,
1649  errmsg("could not fsync file \"%s\": %m",
1650  tmppath)));
1651  return;
1652  }
1654 
1655  if (CloseTransientFile(fd) != 0)
1656  {
1657  int save_errno = errno;
1658 
1660  errno = save_errno;
1661  ereport(elevel,
1663  errmsg("could not close file \"%s\": %m",
1664  tmppath)));
1665  return;
1666  }
1667 
1668  /* rename to permanent file, fsync file and directory */
1669  if (rename(tmppath, path) != 0)
1670  {
1671  int save_errno = errno;
1672 
1674  errno = save_errno;
1675  ereport(elevel,
1677  errmsg("could not rename file \"%s\" to \"%s\": %m",
1678  tmppath, path)));
1679  return;
1680  }
1681 
1682  /*
1683  * Check CreateSlotOnDisk() for the reasoning of using a critical section.
1684  */
1686 
1687  fsync_fname(path, false);
1688  fsync_fname(dir, true);
1689  fsync_fname("pg_replslot", true);
1690 
1691  END_CRIT_SECTION();
1692 
1693  /*
1694  * Successfully wrote, unset dirty bit, unless somebody dirtied again
1695  * already.
1696  */
1697  SpinLockAcquire(&slot->mutex);
1698  if (!slot->just_dirtied)
1699  slot->dirty = false;
1700  SpinLockRelease(&slot->mutex);
1701 
1703 }
#define write(a, b, c)
Definition: win32.h:14
@ WAIT_EVENT_REPLICATION_SLOT_WRITE
Definition: wait_event.h:204
@ WAIT_EVENT_REPLICATION_SLOT_SYNC
Definition: wait_event.h:203

References ReplicationSlotOnDisk::checksum, CloseTransientFile(), COMP_CRC32C, 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, 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, WAIT_EVENT_REPLICATION_SLOT_SYNC, WAIT_EVENT_REPLICATION_SLOT_WRITE, and write.

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

◆ SearchNamedReplicationSlot()

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

Definition at line 373 of file slot.c.

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

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

1435 {
1436  DIR *replication_dir;
1437  struct dirent *replication_de;
1438 
1439  elog(DEBUG1, "starting up replication slots");
1440 
1441  /* restore all slots by iterating over all on-disk entries */
1442  replication_dir = AllocateDir("pg_replslot");
1443  while ((replication_de = ReadDir(replication_dir, "pg_replslot")) != NULL)
1444  {
1445  struct stat statbuf;
1446  char path[MAXPGPATH + 12];
1447 
1448  if (strcmp(replication_de->d_name, ".") == 0 ||
1449  strcmp(replication_de->d_name, "..") == 0)
1450  continue;
1451 
1452  snprintf(path, sizeof(path), "pg_replslot/%s", replication_de->d_name);
1453 
1454  /* we're only creating directories here, skip if it's not our's */
1455  if (lstat(path, &statbuf) == 0 && !S_ISDIR(statbuf.st_mode))
1456  continue;
1457 
1458  /* we crashed while a slot was being setup or deleted, clean up */
1459  if (pg_str_endswith(replication_de->d_name, ".tmp"))
1460  {
1461  if (!rmtree(path, true))
1462  {
1463  ereport(WARNING,
1464  (errmsg("could not remove directory \"%s\"",
1465  path)));
1466  continue;
1467  }
1468  fsync_fname("pg_replslot", true);
1469  continue;
1470  }
1471 
1472  /* looks like a slot in a normal state, restore */
1473  RestoreSlotFromDisk(replication_de->d_name);
1474  }
1475  FreeDir(replication_dir);
1476 
1477  /* currently no slots exist, we're done. */
1478  if (max_replication_slots <= 0)
1479  return;
1480 
1481  /* Now that we have recovered all the data, compute replication xmin */
1484 }
struct dirent * ReadDir(DIR *dir, const char *dirname)
Definition: fd.c:2725
int FreeDir(DIR *dir)
Definition: fd.c:2777
DIR * AllocateDir(const char *dirname)
Definition: fd.c:2659
#define snprintf
Definition: port.h:225
static void RestoreSlotFromDisk(const char *name)
Definition: slot.c:1709
bool pg_str_endswith(const char *str, const char *end)
Definition: string.c:31
Definition: dirent.c:26
Definition: dirent.h:10
char d_name[MAX_PATH]
Definition: dirent.h:15
#define lstat(path, sb)
Definition: win32_port.h:284

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

Referenced by StartupXLOG().

Variable Documentation

◆ max_replication_slots

◆ MyReplicationSlot

◆ ReplicationSlotCtl