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/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 SnapBuildOnDiskNotChecksummedSize   offsetof(ReplicationSlotOnDisk, version)
 
#define SnapBuildOnDiskChecksummedSize   sizeof(ReplicationSlotOnDisk) - SnapBuildOnDiskNotChecksummedSize
 
#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 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)
 
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)
 
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

◆ ReplicationSlotOnDiskConstantSize

#define ReplicationSlotOnDiskConstantSize   offsetof(ReplicationSlotOnDisk, slotdata)

Definition at line 77 of file slot.c.

Referenced by RestoreSlotFromDisk().

◆ ReplicationSlotOnDiskV2Size

#define ReplicationSlotOnDiskV2Size   sizeof(ReplicationSlotOnDisk) - ReplicationSlotOnDiskConstantSize

Definition at line 86 of file slot.c.

Referenced by RestoreSlotFromDisk(), and SaveSlotToPath().

◆ SLOT_MAGIC

#define SLOT_MAGIC   0x1051CA1 /* format identifier */

Definition at line 89 of file slot.c.

Referenced by RestoreSlotFromDisk(), and SaveSlotToPath().

◆ SLOT_VERSION

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

Definition at line 90 of file slot.c.

Referenced by RestoreSlotFromDisk(), and SaveSlotToPath().

◆ SnapBuildOnDiskChecksummedSize

#define SnapBuildOnDiskChecksummedSize   sizeof(ReplicationSlotOnDisk) - SnapBuildOnDiskNotChecksummedSize

Definition at line 83 of file slot.c.

Referenced by RestoreSlotFromDisk(), and SaveSlotToPath().

◆ SnapBuildOnDiskNotChecksummedSize

#define SnapBuildOnDiskNotChecksummedSize   offsetof(ReplicationSlotOnDisk, version)

Definition at line 80 of file slot.c.

Referenced by RestoreSlotFromDisk(), and SaveSlotToPath().

Typedef Documentation

◆ ReplicationSlotOnDisk

Function Documentation

◆ CheckPointReplicationSlots()

void CheckPointReplicationSlots ( void  )

Definition at line 1360 of file slot.c.

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

Referenced by CheckPointGuts().

1361 {
1362  int i;
1363 
1364  elog(DEBUG1, "performing replication slot checkpoint");
1365 
1366  /*
1367  * Prevent any slot from being created/dropped while we're active. As we
1368  * explicitly do *not* want to block iterating over replication_slots or
1369  * acquiring a slot we cannot take the control lock - but that's OK,
1370  * because holding ReplicationSlotAllocationLock is strictly stronger, and
1371  * enough to guarantee that nobody can change the in_use bits on us.
1372  */
1373  LWLockAcquire(ReplicationSlotAllocationLock, LW_SHARED);
1374 
1375  for (i = 0; i < max_replication_slots; i++)
1376  {
1378  char path[MAXPGPATH];
1379 
1380  if (!s->in_use)
1381  continue;
1382 
1383  /* save the slot to disk, locking is handled in SaveSlotToPath() */
1384  sprintf(path, "pg_replslot/%s", NameStr(s->data.name));
1385  SaveSlotToPath(s, path, LOG);
1386  }
1387  LWLockRelease(ReplicationSlotAllocationLock);
1388 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
#define DEBUG1
Definition: elog.h:25
ReplicationSlotPersistentData data
Definition: slot.h:147
#define LOG
Definition: elog.h:26
static void SaveSlotToPath(ReplicationSlot *slot, const char *path, int elevel)
Definition: slot.c:1516
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1803
#define sprintf
Definition: port.h:219
#define MAXPGPATH
bool in_use
Definition: slot.h:123
int max_replication_slots
Definition: slot.c:99
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1199
#define elog(elevel,...)
Definition: elog.h:232
int i
#define NameStr(name)
Definition: c.h:681
ReplicationSlot replication_slots[1]
Definition: slot.h:180

◆ CheckSlotPermissions()

void CheckSlotPermissions ( void  )

Definition at line 1066 of file slot.c.

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

1067 {
1068  if (!superuser() && !has_rolreplication(GetUserId()))
1069  ereport(ERROR,
1070  (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
1071  errmsg("must be superuser or replication role to use replication slots")));
1072 }
Oid GetUserId(void)
Definition: miscinit.c:495
int errcode(int sqlerrcode)
Definition: elog.c:698
bool superuser(void)
Definition: superuser.c:46
#define ERROR
Definition: elog.h:46
#define ereport(elevel,...)
Definition: elog.h:157
bool has_rolreplication(Oid roleid)
Definition: miscinit.c:676
int errmsg(const char *fmt,...)
Definition: elog.c:909

◆ CheckSlotRequirements()

void CheckSlotRequirements ( void  )

Definition at line 1044 of file slot.c.

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

1045 {
1046  /*
1047  * NB: Adding a new requirement likely means that RestoreSlotFromDisk()
1048  * needs the same check.
1049  */
1050 
1051  if (max_replication_slots == 0)
1052  ereport(ERROR,
1053  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1054  errmsg("replication slots can only be used if max_replication_slots > 0")));
1055 
1057  ereport(ERROR,
1058  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1059  errmsg("replication slots can only be used if wal_level >= replica")));
1060 }
int wal_level
Definition: xlog.c:108
int errcode(int sqlerrcode)
Definition: elog.c:698
#define ERROR
Definition: elog.h:46
#define ereport(elevel,...)
Definition: elog.h:157
int max_replication_slots
Definition: slot.c:99
int errmsg(const char *fmt,...)
Definition: elog.c:909

◆ CreateSlotOnDisk()

static void CreateSlotOnDisk ( ReplicationSlot slot)
static

Definition at line 1455 of file slot.c.

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

1456 {
1457  char tmppath[MAXPGPATH];
1458  char path[MAXPGPATH];
1459  struct stat st;
1460 
1461  /*
1462  * No need to take out the io_in_progress_lock, nobody else can see this
1463  * slot yet, so nobody else will write. We're reusing SaveSlotToPath which
1464  * takes out the lock, if we'd take the lock here, we'd deadlock.
1465  */
1466 
1467  sprintf(path, "pg_replslot/%s", NameStr(slot->data.name));
1468  sprintf(tmppath, "pg_replslot/%s.tmp", NameStr(slot->data.name));
1469 
1470  /*
1471  * It's just barely possible that some previous effort to create or drop a
1472  * slot with this name left a temp directory lying around. If that seems
1473  * to be the case, try to remove it. If the rmtree() fails, we'll error
1474  * out at the MakePGDirectory() below, so we don't bother checking
1475  * success.
1476  */
1477  if (stat(tmppath, &st) == 0 && S_ISDIR(st.st_mode))
1478  rmtree(tmppath, true);
1479 
1480  /* Create and fsync the temporary slot directory. */
1481  if (MakePGDirectory(tmppath) < 0)
1482  ereport(ERROR,
1484  errmsg("could not create directory \"%s\": %m",
1485  tmppath)));
1486  fsync_fname(tmppath, true);
1487 
1488  /* Write the actual state file. */
1489  slot->dirty = true; /* signal that we really need to write */
1490  SaveSlotToPath(slot, tmppath, ERROR);
1491 
1492  /* Rename the directory into place. */
1493  if (rename(tmppath, path) != 0)
1494  ereport(ERROR,
1496  errmsg("could not rename file \"%s\" to \"%s\": %m",
1497  tmppath, path)));
1498 
1499  /*
1500  * If we'd now fail - really unlikely - we wouldn't know whether this slot
1501  * would persist after an OS crash or not - so, force a restart. The
1502  * restart would try to fsync this again till it works.
1503  */
1505 
1506  fsync_fname(path, true);
1507  fsync_fname("pg_replslot", true);
1508 
1509  END_CRIT_SECTION();
1510 }
#define END_CRIT_SECTION()
Definition: miscadmin.h:149
void fsync_fname(const char *fname, bool isdir)
Definition: fd.c:671
#define START_CRIT_SECTION()
Definition: miscadmin.h:147
ReplicationSlotPersistentData data
Definition: slot.h:147
static void SaveSlotToPath(ReplicationSlot *slot, const char *path, int elevel)
Definition: slot.c:1516
#define sprintf
Definition: port.h:219
#define ERROR
Definition: elog.h:46
#define MAXPGPATH
int errcode_for_file_access(void)
Definition: elog.c:721
bool rmtree(const char *path, bool rmtopdir)
Definition: rmtree.c:42
#define ereport(elevel,...)
Definition: elog.h:157
int MakePGDirectory(const char *directoryName)
Definition: fd.c:3782
#define S_ISDIR(m)
Definition: win32_port.h:324
int errmsg(const char *fmt,...)
Definition: elog.c:909
#define NameStr(name)
Definition: c.h:681
bool dirty
Definition: slot.h:130
#define stat
Definition: win32_port.h:283

◆ InvalidateObsoleteReplicationSlots()

bool InvalidateObsoleteReplicationSlots ( XLogSegNo  oldestSegno)

Definition at line 1317 of file slot.c.

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

Referenced by CreateCheckPoint(), and CreateRestartPoint().

1318 {
1319  XLogRecPtr oldestLSN;
1320  bool invalidated = false;
1321 
1322  XLogSegNoOffsetToRecPtr(oldestSegno, 0, wal_segment_size, oldestLSN);
1323 
1324 restart:
1325  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
1326  for (int i = 0; i < max_replication_slots; i++)
1327  {
1329 
1330  if (!s->in_use)
1331  continue;
1332 
1333  if (InvalidatePossiblyObsoleteSlot(s, oldestLSN, &invalidated))
1334  {
1335  /* if the lock was released, start from scratch */
1336  goto restart;
1337  }
1338  }
1339  LWLockRelease(ReplicationSlotControlLock);
1340 
1341  /*
1342  * If any slots have been invalidated, recalculate the resource limits.
1343  */
1344  if (invalidated)
1345  {
1348  }
1349 
1350  return invalidated;
1351 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
int wal_segment_size
Definition: xlog.c:119
static bool InvalidatePossiblyObsoleteSlot(ReplicationSlot *s, XLogRecPtr oldestLSN, bool *invalidated)
Definition: slot.c:1164
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1803
void ReplicationSlotsComputeRequiredLSN(void)
Definition: slot.c:817
bool in_use
Definition: slot.h:123
int max_replication_slots
Definition: slot.c:99
#define XLogSegNoOffsetToRecPtr(segno, offset, wal_segsz_bytes, dest)
uint64 XLogRecPtr
Definition: xlogdefs.h:21
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1199
int i
ReplicationSlot replication_slots[1]
Definition: slot.h:180
void ReplicationSlotsComputeRequiredXmin(bool already_locked)
Definition: slot.c:767

◆ InvalidatePossiblyObsoleteSlot()

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

Definition at line 1164 of file slot.c.

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, ReplicationSlotPersistentData::name, NameStr, ReplicationSlotMarkDirty(), ReplicationSlotRelease(), ReplicationSlotSave(), ReplicationSlotPersistentData::restart_lsn, SpinLockAcquire, SpinLockRelease, WAIT_EVENT_REPLICATION_SLOT_DROP, and XLogRecPtrIsInvalid.

Referenced by InvalidateObsoleteReplicationSlots().

1166 {
1167  int last_signaled_pid = 0;
1168  bool released_lock = false;
1169 
1170  for (;;)
1171  {
1172  XLogRecPtr restart_lsn;
1173  NameData slotname;
1174  int active_pid = 0;
1175 
1176  Assert(LWLockHeldByMeInMode(ReplicationSlotControlLock, LW_SHARED));
1177 
1178  if (!s->in_use)
1179  {
1180  if (released_lock)
1181  LWLockRelease(ReplicationSlotControlLock);
1182  break;
1183  }
1184 
1185  /*
1186  * Check if the slot needs to be invalidated. If it needs to be
1187  * invalidated, and is not currently acquired, acquire it and mark it
1188  * as having been invalidated. We do this with the spinlock held to
1189  * avoid race conditions -- for example the restart_lsn could move
1190  * forward, or the slot could be dropped.
1191  */
1192  SpinLockAcquire(&s->mutex);
1193 
1194  restart_lsn = s->data.restart_lsn;
1195 
1196  /*
1197  * If the slot is already invalid or is fresh enough, we don't need to
1198  * do anything.
1199  */
1200  if (XLogRecPtrIsInvalid(restart_lsn) || restart_lsn >= oldestLSN)
1201  {
1202  SpinLockRelease(&s->mutex);
1203  if (released_lock)
1204  LWLockRelease(ReplicationSlotControlLock);
1205  break;
1206  }
1207 
1208  slotname = s->data.name;
1209  active_pid = s->active_pid;
1210 
1211  /*
1212  * If the slot can be acquired, do so and mark it invalidated
1213  * immediately. Otherwise we'll signal the owning process, below, and
1214  * retry.
1215  */
1216  if (active_pid == 0)
1217  {
1218  MyReplicationSlot = s;
1219  s->active_pid = MyProcPid;
1220  s->data.invalidated_at = restart_lsn;
1222 
1223  /* Let caller know */
1224  *invalidated = true;
1225  }
1226 
1227  SpinLockRelease(&s->mutex);
1228 
1229  if (active_pid != 0)
1230  {
1231  /*
1232  * Prepare the sleep on the slot's condition variable before
1233  * releasing the lock, to close a possible race condition if the
1234  * slot is released before the sleep below.
1235  */
1237 
1238  LWLockRelease(ReplicationSlotControlLock);
1239  released_lock = true;
1240 
1241  /*
1242  * Signal to terminate the process that owns the slot, if we
1243  * haven't already signalled it. (Avoidance of repeated
1244  * signalling is the only reason for there to be a loop in this
1245  * routine; otherwise we could rely on caller's restart loop.)
1246  *
1247  * There is the race condition that other process may own the slot
1248  * after its current owner process is terminated and before this
1249  * process owns it. To handle that, we signal only if the PID of
1250  * the owning process has changed from the previous time. (This
1251  * logic assumes that the same PID is not reused very quickly.)
1252  */
1253  if (last_signaled_pid != active_pid)
1254  {
1255  ereport(LOG,
1256  (errmsg("terminating process %d to release replication slot \"%s\"",
1257  active_pid, NameStr(slotname))));
1258 
1259  (void) kill(active_pid, SIGTERM);
1260  last_signaled_pid = active_pid;
1261  }
1262 
1263  /* Wait until the slot is released. */
1266 
1267  /*
1268  * Re-acquire lock and start over; we expect to invalidate the
1269  * slot next time (unless another process acquires the slot in the
1270  * meantime).
1271  */
1272  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
1273  continue;
1274  }
1275  else
1276  {
1277  /*
1278  * We hold the slot now and have already invalidated it; flush it
1279  * to ensure that state persists.
1280  *
1281  * Don't want to hold ReplicationSlotControlLock across file
1282  * system operations, so release it now but be sure to tell caller
1283  * to restart from scratch.
1284  */
1285  LWLockRelease(ReplicationSlotControlLock);
1286  released_lock = true;
1287 
1288  /* Make sure the invalidated state persists across server restart */
1292 
1293  ereport(LOG,
1294  (errmsg("invalidating slot \"%s\" because its restart_lsn %X/%X exceeds max_slot_wal_keep_size",
1295  NameStr(slotname),
1296  LSN_FORMAT_ARGS(restart_lsn))));
1297 
1298  /* done with this slot for now */
1299  break;
1300  }
1301  }
1302 
1303  Assert(released_lock == !LWLockHeldByMe(ReplicationSlotControlLock));
1304 
1305  return released_lock;
1306 }
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
bool LWLockHeldByMeInMode(LWLock *l, LWLockMode mode)
Definition: lwlock.c:1937
int MyProcPid
Definition: globals.c:43
bool LWLockHeldByMe(LWLock *l)
Definition: lwlock.c:1919
#define kill(pid, sig)
Definition: win32_port.h:464
void ReplicationSlotSave(void)
Definition: slot.c:710
ReplicationSlotPersistentData data
Definition: slot.h:147
#define LOG
Definition: elog.h:26
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1803
void ConditionVariablePrepareToSleep(ConditionVariable *cv)
#define SpinLockAcquire(lock)
Definition: spin.h:62
#define LSN_FORMAT_ARGS(lsn)
Definition: xlogdefs.h:43
Definition: c.h:675
void ReplicationSlotRelease(void)
Definition: slot.c:469
#define XLogRecPtrIsInvalid(r)
Definition: xlogdefs.h:29
bool in_use
Definition: slot.h:123
#define SpinLockRelease(lock)
Definition: spin.h:64
#define ereport(elevel,...)
Definition: elog.h:157
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
void ConditionVariableSleep(ConditionVariable *cv, uint32 wait_event_info)
uint64 XLogRecPtr
Definition: xlogdefs.h:21
#define Assert(condition)
Definition: c.h:804
XLogRecPtr restart_lsn
Definition: slot.h:73
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1199
ConditionVariable active_cv
Definition: slot.h:153
int errmsg(const char *fmt,...)
Definition: elog.c:909
pid_t active_pid
Definition: slot.h:126
#define NameStr(name)
Definition: c.h:681
XLogRecPtr invalidated_at
Definition: slot.h:76
slock_t mutex
Definition: slot.h:120
void ReplicationSlotMarkDirty(void)
Definition: slot.c:728

◆ ReplicationSlotAcquire()

void ReplicationSlotAcquire ( const char *  name,
bool  nowait 
)

Definition at line 380 of file slot.c.

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, ReplicationSlotPersistentData::name, NameStr, SearchNamedReplicationSlot(), SpinLockAcquire, SpinLockRelease, and WAIT_EVENT_REPLICATION_SLOT_DROP.

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

381 {
382  ReplicationSlot *s;
383  int active_pid;
384 
385  AssertArg(name != NULL);
386 
387 retry:
388  Assert(MyReplicationSlot == NULL);
389 
390  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
391 
392  /*
393  * Search for the slot with the specified name if the slot to acquire is
394  * not given. If the slot is not found, we either return -1 or error out.
395  */
396  s = SearchNamedReplicationSlot(name, false);
397  if (s == NULL || !s->in_use)
398  {
399  LWLockRelease(ReplicationSlotControlLock);
400 
401  ereport(ERROR,
402  (errcode(ERRCODE_UNDEFINED_OBJECT),
403  errmsg("replication slot \"%s\" does not exist",
404  name)));
405  }
406 
407  /*
408  * This is the slot we want; check if it's active under some other
409  * process. In single user mode, we don't need this check.
410  */
411  if (IsUnderPostmaster)
412  {
413  /*
414  * Get ready to sleep on the slot in case it is active. (We may end
415  * up not sleeping, but we don't want to do this while holding the
416  * spinlock.)
417  */
418  if (!nowait)
420 
421  SpinLockAcquire(&s->mutex);
422  if (s->active_pid == 0)
423  s->active_pid = MyProcPid;
424  active_pid = s->active_pid;
425  SpinLockRelease(&s->mutex);
426  }
427  else
428  active_pid = MyProcPid;
429  LWLockRelease(ReplicationSlotControlLock);
430 
431  /*
432  * If we found the slot but it's already active in another process, we
433  * wait until the owning process signals us that it's been released, or
434  * error out.
435  */
436  if (active_pid != MyProcPid)
437  {
438  if (!nowait)
439  {
440  /* Wait here until we get signaled, and then restart */
444  goto retry;
445  }
446 
447  ereport(ERROR,
448  (errcode(ERRCODE_OBJECT_IN_USE),
449  errmsg("replication slot \"%s\" is active for PID %d",
450  NameStr(s->data.name), active_pid)));
451  }
452  else if (!nowait)
453  ConditionVariableCancelSleep(); /* no sleep needed after all */
454 
455  /* Let everybody know we've modified this slot */
457 
458  /* We made this slot active, so it's ours now. */
459  MyReplicationSlot = s;
460 }
ReplicationSlot * SearchNamedReplicationSlot(const char *name, bool need_lock)
Definition: slot.c:348
int MyProcPid
Definition: globals.c:43
void ConditionVariableBroadcast(ConditionVariable *cv)
int errcode(int sqlerrcode)
Definition: elog.c:698
ReplicationSlotPersistentData data
Definition: slot.h:147
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1803
void ConditionVariablePrepareToSleep(ConditionVariable *cv)
#define SpinLockAcquire(lock)
Definition: spin.h:62
void ConditionVariableCancelSleep(void)
#define ERROR
Definition: elog.h:46
bool IsUnderPostmaster
Definition: globals.c:112
#define AssertArg(condition)
Definition: c.h:806
bool in_use
Definition: slot.h:123
#define SpinLockRelease(lock)
Definition: spin.h:64
#define ereport(elevel,...)
Definition: elog.h:157
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
void ConditionVariableSleep(ConditionVariable *cv, uint32 wait_event_info)
#define Assert(condition)
Definition: c.h:804
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1199
ConditionVariable active_cv
Definition: slot.h:153
const char * name
Definition: encode.c:561
int errmsg(const char *fmt,...)
Definition: elog.c:909
pid_t active_pid
Definition: slot.h:126
#define NameStr(name)
Definition: c.h:681
slock_t mutex
Definition: slot.h:120

◆ ReplicationSlotCleanup()

void ReplicationSlotCleanup ( void  )

Definition at line 525 of file slot.c.

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, ReplicationSlotPersistentData::persistency, ReplicationSlotCtlData::replication_slots, ReplicationSlotDropPtr(), RS_TEMPORARY, SpinLockAcquire, and SpinLockRelease.

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

526 {
527  int i;
528 
529  Assert(MyReplicationSlot == NULL);
530 
531 restart:
532  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
533  for (i = 0; i < max_replication_slots; i++)
534  {
536 
537  if (!s->in_use)
538  continue;
539 
540  SpinLockAcquire(&s->mutex);
541  if (s->active_pid == MyProcPid)
542  {
544  SpinLockRelease(&s->mutex);
545  LWLockRelease(ReplicationSlotControlLock); /* avoid deadlock */
546 
548 
550  goto restart;
551  }
552  else
553  SpinLockRelease(&s->mutex);
554  }
555 
556  LWLockRelease(ReplicationSlotControlLock);
557 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
int MyProcPid
Definition: globals.c:43
ReplicationSlotPersistency persistency
Definition: slot.h:54
void ConditionVariableBroadcast(ConditionVariable *cv)
static void ReplicationSlotDropPtr(ReplicationSlot *slot)
Definition: slot.c:593
ReplicationSlotPersistentData data
Definition: slot.h:147
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1803
#define SpinLockAcquire(lock)
Definition: spin.h:62
bool in_use
Definition: slot.h:123
#define SpinLockRelease(lock)
Definition: spin.h:64
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
int max_replication_slots
Definition: slot.c:99
#define Assert(condition)
Definition: c.h:804
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1199
ConditionVariable active_cv
Definition: slot.h:153
pid_t active_pid
Definition: slot.h:126
int i
ReplicationSlot replication_slots[1]
Definition: slot.h:180
slock_t mutex
Definition: slot.h:120

◆ ReplicationSlotCreate()

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

Definition at line 226 of file slot.c.

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, ReplicationSlotPersistentData::name, NameStr, namestrcpy(), ReplicationSlotPersistentData::persistency, pgstat_report_replslot_create(), ReplicationSlotCtlData::replication_slots, 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().

228 {
229  ReplicationSlot *slot = NULL;
230  int i;
231 
232  Assert(MyReplicationSlot == NULL);
233 
235 
236  /*
237  * If some other backend ran this code concurrently with us, we'd likely
238  * both allocate the same slot, and that would be bad. We'd also be at
239  * risk of missing a name collision. Also, we don't want to try to create
240  * a new slot while somebody's busy cleaning up an old one, because we
241  * might both be monkeying with the same directory.
242  */
243  LWLockAcquire(ReplicationSlotAllocationLock, LW_EXCLUSIVE);
244 
245  /*
246  * Check for name collision, and identify an allocatable slot. We need to
247  * hold ReplicationSlotControlLock in shared mode for this, so that nobody
248  * else can change the in_use flags while we're looking at them.
249  */
250  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
251  for (i = 0; i < max_replication_slots; i++)
252  {
254 
255  if (s->in_use && strcmp(name, NameStr(s->data.name)) == 0)
256  ereport(ERROR,
258  errmsg("replication slot \"%s\" already exists", name)));
259  if (!s->in_use && slot == NULL)
260  slot = s;
261  }
262  LWLockRelease(ReplicationSlotControlLock);
263 
264  /* If all slots are in use, we're out of luck. */
265  if (slot == NULL)
266  ereport(ERROR,
267  (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
268  errmsg("all replication slots are in use"),
269  errhint("Free one or increase max_replication_slots.")));
270 
271  /*
272  * Since this slot is not in use, nobody should be looking at any part of
273  * it other than the in_use field unless they're trying to allocate it.
274  * And since we hold ReplicationSlotAllocationLock, nobody except us can
275  * be doing that. So it's safe to initialize the slot.
276  */
277  Assert(!slot->in_use);
278  Assert(slot->active_pid == 0);
279 
280  /* first initialize persistent data */
281  memset(&slot->data, 0, sizeof(ReplicationSlotPersistentData));
282  namestrcpy(&slot->data.name, name);
283  slot->data.database = db_specific ? MyDatabaseId : InvalidOid;
284  slot->data.persistency = persistency;
285  slot->data.two_phase = two_phase;
287 
288  /* and then data only present in shared memory */
289  slot->just_dirtied = false;
290  slot->dirty = false;
297 
298  /*
299  * Create the slot on disk. We haven't actually marked the slot allocated
300  * yet, so no special cleanup is required if this errors out.
301  */
302  CreateSlotOnDisk(slot);
303 
304  /*
305  * We need to briefly prevent any other backend from iterating over the
306  * slots while we flip the in_use flag. We also need to set the active
307  * flag while holding the ControlLock as otherwise a concurrent
308  * ReplicationSlotAcquire() could acquire the slot as well.
309  */
310  LWLockAcquire(ReplicationSlotControlLock, LW_EXCLUSIVE);
311 
312  slot->in_use = true;
313 
314  /* We can now mark the slot active, and that makes it our slot. */
315  SpinLockAcquire(&slot->mutex);
316  Assert(slot->active_pid == 0);
317  slot->active_pid = MyProcPid;
318  SpinLockRelease(&slot->mutex);
319  MyReplicationSlot = slot;
320 
321  LWLockRelease(ReplicationSlotControlLock);
322 
323  /*
324  * Create statistics entry for the new logical slot. We don't collect any
325  * stats for physical slots, so no need to create an entry for the same.
326  * See ReplicationSlotDropPtr for why we need to do this before releasing
327  * ReplicationSlotAllocationLock.
328  */
329  if (SlotIsLogical(slot))
331 
332  /*
333  * Now that the slot has been marked as in_use and active, it's safe to
334  * let somebody else try to allocate a slot.
335  */
336  LWLockRelease(ReplicationSlotAllocationLock);
337 
338  /* Let everybody know we've modified this slot */
340 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
TransactionId candidate_catalog_xmin
Definition: slot.h:162
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
int MyProcPid
Definition: globals.c:43
int errhint(const char *fmt,...)
Definition: elog.c:1156
void namestrcpy(Name name, const char *str)
Definition: name.c:233
ReplicationSlotPersistency persistency
Definition: slot.h:54
void ConditionVariableBroadcast(ConditionVariable *cv)
int errcode(int sqlerrcode)
Definition: elog.c:698
ReplicationSlotPersistentData data
Definition: slot.h:147
static void CreateSlotOnDisk(ReplicationSlot *slot)
Definition: slot.c:1455
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1803
bool ReplicationSlotValidateName(const char *name, int elevel)
Definition: slot.c:172
#define SpinLockAcquire(lock)
Definition: spin.h:62
#define ERROR
Definition: elog.h:46
TransactionId effective_xmin
Definition: slot.h:143
XLogRecPtr candidate_restart_valid
Definition: slot.h:164
#define InvalidTransactionId
Definition: transam.h:31
#define SlotIsLogical(slot)
Definition: slot.h:169
bool in_use
Definition: slot.h:123
#define SpinLockRelease(lock)
Definition: spin.h:64
bool just_dirtied
Definition: slot.h:129
TransactionId effective_catalog_xmin
Definition: slot.h:144
Oid MyDatabaseId
Definition: globals.c:88
static bool two_phase
#define InvalidOid
Definition: postgres_ext.h:36
#define ereport(elevel,...)
Definition: elog.h:157
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
int max_replication_slots
Definition: slot.c:99
#define Assert(condition)
Definition: c.h:804
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1199
ConditionVariable active_cv
Definition: slot.h:153
const char * name
Definition: encode.c:561
XLogRecPtr candidate_xmin_lsn
Definition: slot.h:163
void pgstat_report_replslot_create(const char *slotname)
Definition: pgstat.c:1843
int errmsg(const char *fmt,...)
Definition: elog.c:909
pid_t active_pid
Definition: slot.h:126
int i
#define NameStr(name)
Definition: c.h:681
ReplicationSlot replication_slots[1]
Definition: slot.h:180
slock_t mutex
Definition: slot.h:120
#define ERRCODE_DUPLICATE_OBJECT
Definition: streamutil.c:32
bool dirty
Definition: slot.h:130
XLogRecPtr candidate_restart_lsn
Definition: slot.h:165
XLogRecPtr two_phase_at
Definition: slot.h:90

◆ ReplicationSlotDrop()

void ReplicationSlotDrop ( const char *  name,
bool  nowait 
)

Definition at line 563 of file slot.c.

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

Referenced by DropReplicationSlot(), and pg_drop_replication_slot().

564 {
565  Assert(MyReplicationSlot == NULL);
566 
567  ReplicationSlotAcquire(name, nowait);
568 
570 }
void ReplicationSlotAcquire(const char *name, bool nowait)
Definition: slot.c:380
static void ReplicationSlotDropAcquired(void)
Definition: slot.c:576
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
#define Assert(condition)
Definition: c.h:804
const char * name
Definition: encode.c:561

◆ ReplicationSlotDropAcquired()

static void ReplicationSlotDropAcquired ( void  )
static

Definition at line 576 of file slot.c.

References Assert, MyReplicationSlot, and ReplicationSlotDropPtr().

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

577 {
579 
580  Assert(MyReplicationSlot != NULL);
581 
582  /* slot isn't acquired anymore */
583  MyReplicationSlot = NULL;
584 
586 }
static void ReplicationSlotDropPtr(ReplicationSlot *slot)
Definition: slot.c:593
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
#define Assert(condition)
Definition: c.h:804

◆ ReplicationSlotDropPtr()

static void ReplicationSlotDropPtr ( ReplicationSlot slot)
static

Definition at line 593 of file slot.c.

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_report_replslot_drop(), ReplicationSlotsComputeRequiredLSN(), ReplicationSlotsComputeRequiredXmin(), rmtree(), RS_PERSISTENT, SlotIsLogical, SpinLockAcquire, SpinLockRelease, sprintf, START_CRIT_SECTION, and WARNING.

Referenced by ReplicationSlotCleanup(), and ReplicationSlotDropAcquired().

594 {
595  char path[MAXPGPATH];
596  char tmppath[MAXPGPATH];
597 
598  /*
599  * If some other backend ran this code concurrently with us, we might try
600  * to delete a slot with a certain name while someone else was trying to
601  * create a slot with the same name.
602  */
603  LWLockAcquire(ReplicationSlotAllocationLock, LW_EXCLUSIVE);
604 
605  /* Generate pathnames. */
606  sprintf(path, "pg_replslot/%s", NameStr(slot->data.name));
607  sprintf(tmppath, "pg_replslot/%s.tmp", NameStr(slot->data.name));
608 
609  /*
610  * Rename the slot directory on disk, so that we'll no longer recognize
611  * this as a valid slot. Note that if this fails, we've got to mark the
612  * slot inactive before bailing out. If we're dropping an ephemeral or a
613  * temporary slot, we better never fail hard as the caller won't expect
614  * the slot to survive and this might get called during error handling.
615  */
616  if (rename(path, tmppath) == 0)
617  {
618  /*
619  * We need to fsync() the directory we just renamed and its parent to
620  * make sure that our changes are on disk in a crash-safe fashion. If
621  * fsync() fails, we can't be sure whether the changes are on disk or
622  * not. For now, we handle that by panicking;
623  * StartupReplicationSlots() will try to straighten it out after
624  * restart.
625  */
627  fsync_fname(tmppath, true);
628  fsync_fname("pg_replslot", true);
630  }
631  else
632  {
633  bool fail_softly = slot->data.persistency != RS_PERSISTENT;
634 
635  SpinLockAcquire(&slot->mutex);
636  slot->active_pid = 0;
637  SpinLockRelease(&slot->mutex);
638 
639  /* wake up anyone waiting on this slot */
641 
642  ereport(fail_softly ? WARNING : ERROR,
644  errmsg("could not rename file \"%s\" to \"%s\": %m",
645  path, tmppath)));
646  }
647 
648  /*
649  * The slot is definitely gone. Lock out concurrent scans of the array
650  * long enough to kill it. It's OK to clear the active PID here without
651  * grabbing the mutex because nobody else can be scanning the array here,
652  * and nobody can be attached to this slot and thus access it without
653  * scanning the array.
654  *
655  * Also wake up processes waiting for it.
656  */
657  LWLockAcquire(ReplicationSlotControlLock, LW_EXCLUSIVE);
658  slot->active_pid = 0;
659  slot->in_use = false;
660  LWLockRelease(ReplicationSlotControlLock);
662 
663  /*
664  * Slot is dead and doesn't prevent resource removal anymore, recompute
665  * limits.
666  */
669 
670  /*
671  * If removing the directory fails, the worst thing that will happen is
672  * that the user won't be able to create a new slot with the same name
673  * until the next server restart. We warn about it, but that's all.
674  */
675  if (!rmtree(tmppath, true))
677  (errmsg("could not remove directory \"%s\"", tmppath)));
678 
679  /*
680  * Send a message to drop the replication slot to the stats collector.
681  * Since there is no guarantee of the order of message transfer on a UDP
682  * connection, it's possible that a message for creating a new slot
683  * reaches before a message for removing the old slot. We send the drop
684  * and create messages while holding ReplicationSlotAllocationLock to
685  * reduce that possibility. If the messages reached in reverse, we would
686  * lose one statistics update message. But the next update message will
687  * create the statistics for the replication slot.
688  *
689  * XXX In case, the messages for creation and drop slot of the same name
690  * get lost and create happens before (auto)vacuum cleans up the dead
691  * slot, the stats will be accumulated into the old slot. One can imagine
692  * having OIDs for each slot to avoid the accumulation of stats but that
693  * doesn't seem worth doing as in practice this won't happen frequently.
694  */
695  if (SlotIsLogical(slot))
697 
698  /*
699  * We release this at the very end, so that nobody starts trying to create
700  * a slot while we're still cleaning up the detritus of the old one.
701  */
702  LWLockRelease(ReplicationSlotAllocationLock);
703 }
#define END_CRIT_SECTION()
Definition: miscadmin.h:149
void fsync_fname(const char *fname, bool isdir)
Definition: fd.c:671
ReplicationSlotPersistency persistency
Definition: slot.h:54
#define START_CRIT_SECTION()
Definition: miscadmin.h:147
void ConditionVariableBroadcast(ConditionVariable *cv)
ReplicationSlotPersistentData data
Definition: slot.h:147
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1803
#define sprintf
Definition: port.h:219
#define SpinLockAcquire(lock)
Definition: spin.h:62
void ReplicationSlotsComputeRequiredLSN(void)
Definition: slot.c:817
#define ERROR
Definition: elog.h:46
#define MAXPGPATH
int errcode_for_file_access(void)
Definition: elog.c:721
#define SlotIsLogical(slot)
Definition: slot.h:169
#define WARNING
Definition: elog.h:40
bool rmtree(const char *path, bool rmtopdir)
Definition: rmtree.c:42
bool in_use
Definition: slot.h:123
#define SpinLockRelease(lock)
Definition: spin.h:64
#define ereport(elevel,...)
Definition: elog.h:157
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1199
ConditionVariable active_cv
Definition: slot.h:153
void pgstat_report_replslot_drop(const char *slotname)
Definition: pgstat.c:1861
int errmsg(const char *fmt,...)
Definition: elog.c:909
pid_t active_pid
Definition: slot.h:126
#define NameStr(name)
Definition: c.h:681
slock_t mutex
Definition: slot.h:120
void ReplicationSlotsComputeRequiredXmin(bool already_locked)
Definition: slot.c:767

◆ ReplicationSlotMarkDirty()

◆ ReplicationSlotPersist()

void ReplicationSlotPersist ( void  )

Definition at line 745 of file slot.c.

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

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

746 {
748 
749  Assert(slot != NULL);
751 
752  SpinLockAcquire(&slot->mutex);
754  SpinLockRelease(&slot->mutex);
755 
758 }
ReplicationSlotPersistency persistency
Definition: slot.h:54
void ReplicationSlotSave(void)
Definition: slot.c:710
ReplicationSlotPersistentData data
Definition: slot.h:147
#define SpinLockAcquire(lock)
Definition: spin.h:62
#define SpinLockRelease(lock)
Definition: spin.h:64
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
#define Assert(condition)
Definition: c.h:804
slock_t mutex
Definition: slot.h:120
void ReplicationSlotMarkDirty(void)
Definition: slot.c:728

◆ ReplicationSlotRelease()

void ReplicationSlotRelease ( void  )

Definition at line 469 of file slot.c.

References ReplicationSlot::active_cv, ReplicationSlot::active_pid, Assert, ConditionVariableBroadcast(), ReplicationSlot::data, ReplicationSlot::effective_xmin, InvalidTransactionId, LW_SHARED, 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(), ProcKill(), StartLogicalReplication(), StartReplication(), and WalSndErrorCleanup().

470 {
472 
473  Assert(slot != NULL && slot->active_pid != 0);
474 
475  if (slot->data.persistency == RS_EPHEMERAL)
476  {
477  /*
478  * Delete the slot. There is no !PANIC case where this is allowed to
479  * fail, all that may happen is an incomplete cleanup of the on-disk
480  * data.
481  */
483  }
484 
485  /*
486  * If slot needed to temporarily restrain both data and catalog xmin to
487  * create the catalog snapshot, remove that temporary constraint.
488  * Snapshots can only be exported while the initial snapshot is still
489  * acquired.
490  */
491  if (!TransactionIdIsValid(slot->data.xmin) &&
493  {
494  SpinLockAcquire(&slot->mutex);
496  SpinLockRelease(&slot->mutex);
498  }
499 
500  if (slot->data.persistency == RS_PERSISTENT)
501  {
502  /*
503  * Mark persistent slot inactive. We're not freeing it, just
504  * disconnecting, but wake up others that may be waiting for it.
505  */
506  SpinLockAcquire(&slot->mutex);
507  slot->active_pid = 0;
508  SpinLockRelease(&slot->mutex);
510  }
511 
512  MyReplicationSlot = NULL;
513 
514  /* might not have been set when we've been a plain slot */
515  LWLockAcquire(ProcArrayLock, LW_SHARED);
518  LWLockRelease(ProcArrayLock);
519 }
#define PROC_IN_LOGICAL_DECODING
Definition: proc.h:61
PGPROC * MyProc
Definition: proc.c:68
ReplicationSlotPersistency persistency
Definition: slot.h:54
void ConditionVariableBroadcast(ConditionVariable *cv)
PROC_HDR * ProcGlobal
Definition: proc.c:80
uint8 statusFlags
Definition: proc.h:189
ReplicationSlotPersistentData data
Definition: slot.h:147
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1803
#define SpinLockAcquire(lock)
Definition: spin.h:62
static void ReplicationSlotDropAcquired(void)
Definition: slot.c:576
TransactionId effective_xmin
Definition: slot.h:143
#define InvalidTransactionId
Definition: transam.h:31
TransactionId xmin
Definition: slot.h:62
#define SpinLockRelease(lock)
Definition: spin.h:64
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
#define Assert(condition)
Definition: c.h:804
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1199
ConditionVariable active_cv
Definition: slot.h:153
pid_t active_pid
Definition: slot.h:126
int pgxactoff
Definition: proc.h:148
uint8 * statusFlags
Definition: proc.h:333
slock_t mutex
Definition: slot.h:120
#define TransactionIdIsValid(xid)
Definition: transam.h:41
void ReplicationSlotsComputeRequiredXmin(bool already_locked)
Definition: slot.c:767

◆ ReplicationSlotReserveWal()

void ReplicationSlotReserveWal ( void  )

Definition at line 1081 of file slot.c.

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

1082 {
1084 
1085  Assert(slot != NULL);
1087 
1088  /*
1089  * The replication slot mechanism is used to prevent removal of required
1090  * WAL. As there is no interlock between this routine and checkpoints, WAL
1091  * segments could concurrently be removed when a now stale return value of
1092  * ReplicationSlotsComputeRequiredLSN() is used. In the unlikely case that
1093  * this happens we'll just retry.
1094  */
1095  while (true)
1096  {
1097  XLogSegNo segno;
1098  XLogRecPtr restart_lsn;
1099 
1100  /*
1101  * For logical slots log a standby snapshot and start logical decoding
1102  * at exactly that position. That allows the slot to start up more
1103  * quickly.
1104  *
1105  * That's not needed (or indeed helpful) for physical slots as they'll
1106  * start replay at the last logged checkpoint anyway. Instead return
1107  * the location of the last redo LSN. While that slightly increases
1108  * the chance that we have to retry, it's where a base backup has to
1109  * start replay at.
1110  */
1111  if (!RecoveryInProgress() && SlotIsLogical(slot))
1112  {
1113  XLogRecPtr flushptr;
1114 
1115  /* start at current insert position */
1116  restart_lsn = GetXLogInsertRecPtr();
1117  SpinLockAcquire(&slot->mutex);
1118  slot->data.restart_lsn = restart_lsn;
1119  SpinLockRelease(&slot->mutex);
1120 
1121  /* make sure we have enough information to start */
1122  flushptr = LogStandbySnapshot();
1123 
1124  /* and make sure it's fsynced to disk */
1125  XLogFlush(flushptr);
1126  }
1127  else
1128  {
1129  restart_lsn = GetRedoRecPtr();
1130  SpinLockAcquire(&slot->mutex);
1131  slot->data.restart_lsn = restart_lsn;
1132  SpinLockRelease(&slot->mutex);
1133  }
1134 
1135  /* prevent WAL removal as fast as possible */
1137 
1138  /*
1139  * If all required WAL is still there, great, otherwise retry. The
1140  * slot should prevent further removal of WAL, unless there's a
1141  * concurrent ReplicationSlotsComputeRequiredLSN() after we've written
1142  * the new restart_lsn above, so normally we should never need to loop
1143  * more than twice.
1144  */
1146  if (XLogGetLastRemovedSegno() < segno)
1147  break;
1148  }
1149 }
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
int wal_segment_size
Definition: xlog.c:119
ReplicationSlotPersistentData data
Definition: slot.h:147
bool RecoveryInProgress(void)
Definition: xlog.c:8328
void XLogFlush(XLogRecPtr record)
Definition: xlog.c:2887
#define SpinLockAcquire(lock)
Definition: spin.h:62
XLogSegNo XLogGetLastRemovedSegno(void)
Definition: xlog.c:4030
void ReplicationSlotsComputeRequiredLSN(void)
Definition: slot.c:817
XLogRecPtr LogStandbySnapshot(void)
Definition: standby.c:1220
XLogRecPtr GetXLogInsertRecPtr(void)
Definition: xlog.c:11946
uint64 XLogSegNo
Definition: xlogdefs.h:48
#define SlotIsLogical(slot)
Definition: slot.h:169
#define SpinLockRelease(lock)
Definition: spin.h:64
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
uint64 XLogRecPtr
Definition: xlogdefs.h:21
#define Assert(condition)
Definition: c.h:804
XLogRecPtr restart_lsn
Definition: slot.h:73
XLogRecPtr GetRedoRecPtr(void)
Definition: xlog.c:8620
slock_t mutex
Definition: slot.h:120
#define XLByteToSeg(xlrp, logSegNo, wal_segsz_bytes)

◆ ReplicationSlotSave()

void ReplicationSlotSave ( void  )

Definition at line 710 of file slot.c.

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

Referenced by copy_replication_slot(), create_physical_replication_slot(), CreateDecodingContext(), CreateInitDecodingContext(), CreateReplicationSlot(), InvalidatePossiblyObsoleteSlot(), LogicalConfirmReceivedLocation(), and ReplicationSlotPersist().

711 {
712  char path[MAXPGPATH];
713 
714  Assert(MyReplicationSlot != NULL);
715 
716  sprintf(path, "pg_replslot/%s", NameStr(MyReplicationSlot->data.name));
718 }
ReplicationSlotPersistentData data
Definition: slot.h:147
static void SaveSlotToPath(ReplicationSlot *slot, const char *path, int elevel)
Definition: slot.c:1516
#define sprintf
Definition: port.h:219
#define ERROR
Definition: elog.h:46
#define MAXPGPATH
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
#define Assert(condition)
Definition: c.h:804
#define NameStr(name)
Definition: c.h:681

◆ ReplicationSlotsComputeLogicalRestartLSN()

XLogRecPtr ReplicationSlotsComputeLogicalRestartLSN ( void  )

Definition at line 860 of file slot.c.

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

Referenced by CheckPointLogicalRewriteHeap(), and CheckPointSnapBuild().

861 {
862  XLogRecPtr result = InvalidXLogRecPtr;
863  int i;
864 
865  if (max_replication_slots <= 0)
866  return InvalidXLogRecPtr;
867 
868  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
869 
870  for (i = 0; i < max_replication_slots; i++)
871  {
872  ReplicationSlot *s;
873  XLogRecPtr restart_lsn;
874 
876 
877  /* cannot change while ReplicationSlotCtlLock is held */
878  if (!s->in_use)
879  continue;
880 
881  /* we're only interested in logical slots */
882  if (!SlotIsLogical(s))
883  continue;
884 
885  /* read once, it's ok if it increases while we're checking */
886  SpinLockAcquire(&s->mutex);
887  restart_lsn = s->data.restart_lsn;
888  SpinLockRelease(&s->mutex);
889 
890  if (restart_lsn == InvalidXLogRecPtr)
891  continue;
892 
893  if (result == InvalidXLogRecPtr ||
894  restart_lsn < result)
895  result = restart_lsn;
896  }
897 
898  LWLockRelease(ReplicationSlotControlLock);
899 
900  return result;
901 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
ReplicationSlotPersistentData data
Definition: slot.h:147
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1803
#define SpinLockAcquire(lock)
Definition: spin.h:62
#define SlotIsLogical(slot)
Definition: slot.h:169
bool in_use
Definition: slot.h:123
#define SpinLockRelease(lock)
Definition: spin.h:64
int max_replication_slots
Definition: slot.c:99
uint64 XLogRecPtr
Definition: xlogdefs.h:21
XLogRecPtr restart_lsn
Definition: slot.h:73
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1199
int i
ReplicationSlot replication_slots[1]
Definition: slot.h:180
slock_t mutex
Definition: slot.h:120

◆ ReplicationSlotsComputeRequiredLSN()

void ReplicationSlotsComputeRequiredLSN ( void  )

Definition at line 817 of file slot.c.

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

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

818 {
819  int i;
820  XLogRecPtr min_required = InvalidXLogRecPtr;
821 
822  Assert(ReplicationSlotCtl != NULL);
823 
824  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
825  for (i = 0; i < max_replication_slots; i++)
826  {
828  XLogRecPtr restart_lsn;
829 
830  if (!s->in_use)
831  continue;
832 
833  SpinLockAcquire(&s->mutex);
834  restart_lsn = s->data.restart_lsn;
835  SpinLockRelease(&s->mutex);
836 
837  if (restart_lsn != InvalidXLogRecPtr &&
838  (min_required == InvalidXLogRecPtr ||
839  restart_lsn < min_required))
840  min_required = restart_lsn;
841  }
842  LWLockRelease(ReplicationSlotControlLock);
843 
844  XLogSetReplicationSlotMinimumLSN(min_required);
845 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
ReplicationSlotPersistentData data
Definition: slot.h:147
void XLogSetReplicationSlotMinimumLSN(XLogRecPtr lsn)
Definition: xlog.c:2769
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1803
#define SpinLockAcquire(lock)
Definition: spin.h:62
bool in_use
Definition: slot.h:123
#define SpinLockRelease(lock)
Definition: spin.h:64
int max_replication_slots
Definition: slot.c:99
uint64 XLogRecPtr
Definition: xlogdefs.h:21
#define Assert(condition)
Definition: c.h:804
XLogRecPtr restart_lsn
Definition: slot.h:73
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1199
int i
ReplicationSlot replication_slots[1]
Definition: slot.h:180
slock_t mutex
Definition: slot.h:120

◆ ReplicationSlotsComputeRequiredXmin()

void ReplicationSlotsComputeRequiredXmin ( bool  already_locked)

Definition at line 767 of file slot.c.

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

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

768 {
769  int i;
771  TransactionId agg_catalog_xmin = InvalidTransactionId;
772 
773  Assert(ReplicationSlotCtl != NULL);
774 
775  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
776 
777  for (i = 0; i < max_replication_slots; i++)
778  {
780  TransactionId effective_xmin;
781  TransactionId effective_catalog_xmin;
782 
783  if (!s->in_use)
784  continue;
785 
786  SpinLockAcquire(&s->mutex);
787  effective_xmin = s->effective_xmin;
788  effective_catalog_xmin = s->effective_catalog_xmin;
789  SpinLockRelease(&s->mutex);
790 
791  /* check the data xmin */
792  if (TransactionIdIsValid(effective_xmin) &&
793  (!TransactionIdIsValid(agg_xmin) ||
794  TransactionIdPrecedes(effective_xmin, agg_xmin)))
795  agg_xmin = effective_xmin;
796 
797  /* check the catalog xmin */
798  if (TransactionIdIsValid(effective_catalog_xmin) &&
799  (!TransactionIdIsValid(agg_catalog_xmin) ||
800  TransactionIdPrecedes(effective_catalog_xmin, agg_catalog_xmin)))
801  agg_catalog_xmin = effective_catalog_xmin;
802  }
803 
804  LWLockRelease(ReplicationSlotControlLock);
805 
806  ProcArraySetReplicationSlotXmin(agg_xmin, agg_catalog_xmin, already_locked);
807 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
uint32 TransactionId
Definition: c.h:587
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1803
#define SpinLockAcquire(lock)
Definition: spin.h:62
TransactionId effective_xmin
Definition: slot.h:143
#define InvalidTransactionId
Definition: transam.h:31
bool TransactionIdPrecedes(TransactionId id1, TransactionId id2)
Definition: transam.c:300
bool in_use
Definition: slot.h:123
#define SpinLockRelease(lock)
Definition: spin.h:64
TransactionId effective_catalog_xmin
Definition: slot.h:144
int max_replication_slots
Definition: slot.c:99
#define Assert(condition)
Definition: c.h:804
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1199
int i
void ProcArraySetReplicationSlotXmin(TransactionId xmin, TransactionId catalog_xmin, bool already_locked)
Definition: procarray.c:3868
ReplicationSlot replication_slots[1]
Definition: slot.h:180
slock_t mutex
Definition: slot.h:120
#define TransactionIdIsValid(xid)
Definition: transam.h:41

◆ ReplicationSlotsCountDBSlots()

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

Definition at line 912 of file slot.c.

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

Referenced by dropdb().

913 {
914  int i;
915 
916  *nslots = *nactive = 0;
917 
918  if (max_replication_slots <= 0)
919  return false;
920 
921  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
922  for (i = 0; i < max_replication_slots; i++)
923  {
924  ReplicationSlot *s;
925 
927 
928  /* cannot change while ReplicationSlotCtlLock is held */
929  if (!s->in_use)
930  continue;
931 
932  /* only logical slots are database specific, skip */
933  if (!SlotIsLogical(s))
934  continue;
935 
936  /* not our database, skip */
937  if (s->data.database != dboid)
938  continue;
939 
940  /* count slots with spinlock held */
941  SpinLockAcquire(&s->mutex);
942  (*nslots)++;
943  if (s->active_pid != 0)
944  (*nactive)++;
945  SpinLockRelease(&s->mutex);
946  }
947  LWLockRelease(ReplicationSlotControlLock);
948 
949  if (*nslots > 0)
950  return true;
951  return false;
952 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
ReplicationSlotPersistentData data
Definition: slot.h:147
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1803
#define SpinLockAcquire(lock)
Definition: spin.h:62
#define SlotIsLogical(slot)
Definition: slot.h:169
bool in_use
Definition: slot.h:123
#define SpinLockRelease(lock)
Definition: spin.h:64
int max_replication_slots
Definition: slot.c:99
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1199
pid_t active_pid
Definition: slot.h:126
int i
ReplicationSlot replication_slots[1]
Definition: slot.h:180
slock_t mutex
Definition: slot.h:120

◆ ReplicationSlotsDropDBSlots()

void ReplicationSlotsDropDBSlots ( Oid  dboid)

Definition at line 968 of file slot.c.

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

Referenced by dbase_redo(), and dropdb().

969 {
970  int i;
971 
972  if (max_replication_slots <= 0)
973  return;
974 
975 restart:
976  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
977  for (i = 0; i < max_replication_slots; i++)
978  {
979  ReplicationSlot *s;
980  char *slotname;
981  int active_pid;
982 
984 
985  /* cannot change while ReplicationSlotCtlLock is held */
986  if (!s->in_use)
987  continue;
988 
989  /* only logical slots are database specific, skip */
990  if (!SlotIsLogical(s))
991  continue;
992 
993  /* not our database, skip */
994  if (s->data.database != dboid)
995  continue;
996 
997  /* acquire slot, so ReplicationSlotDropAcquired can be reused */
998  SpinLockAcquire(&s->mutex);
999  /* can't change while ReplicationSlotControlLock is held */
1000  slotname = NameStr(s->data.name);
1001  active_pid = s->active_pid;
1002  if (active_pid == 0)
1003  {
1004  MyReplicationSlot = s;
1005  s->active_pid = MyProcPid;
1006  }
1007  SpinLockRelease(&s->mutex);
1008 
1009  /*
1010  * Even though we hold an exclusive lock on the database object a
1011  * logical slot for that DB can still be active, e.g. if it's
1012  * concurrently being dropped by a backend connected to another DB.
1013  *
1014  * That's fairly unlikely in practice, so we'll just bail out.
1015  */
1016  if (active_pid)
1017  ereport(ERROR,
1018  (errcode(ERRCODE_OBJECT_IN_USE),
1019  errmsg("replication slot \"%s\" is active for PID %d",
1020  slotname, active_pid)));
1021 
1022  /*
1023  * To avoid duplicating ReplicationSlotDropAcquired() and to avoid
1024  * holding ReplicationSlotControlLock over filesystem operations,
1025  * release ReplicationSlotControlLock and use
1026  * ReplicationSlotDropAcquired.
1027  *
1028  * As that means the set of slots could change, restart scan from the
1029  * beginning each time we release the lock.
1030  */
1031  LWLockRelease(ReplicationSlotControlLock);
1033  goto restart;
1034  }
1035  LWLockRelease(ReplicationSlotControlLock);
1036 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
int MyProcPid
Definition: globals.c:43
int errcode(int sqlerrcode)
Definition: elog.c:698
ReplicationSlotPersistentData data
Definition: slot.h:147
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1803
#define SpinLockAcquire(lock)
Definition: spin.h:62
static void ReplicationSlotDropAcquired(void)
Definition: slot.c:576
#define ERROR
Definition: elog.h:46
#define SlotIsLogical(slot)
Definition: slot.h:169
bool in_use
Definition: slot.h:123
#define SpinLockRelease(lock)
Definition: spin.h:64
#define ereport(elevel,...)
Definition: elog.h:157
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
int max_replication_slots
Definition: slot.c:99
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1199
int errmsg(const char *fmt,...)
Definition: elog.c:909
pid_t active_pid
Definition: slot.h:126
int i
#define NameStr(name)
Definition: c.h:681
ReplicationSlot replication_slots[1]
Definition: slot.h:180
slock_t mutex
Definition: slot.h:120

◆ ReplicationSlotsShmemInit()

void ReplicationSlotsShmemInit ( void  )

Definition at line 132 of file slot.c.

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, ReplicationSlotsShmemSize(), ShmemInitStruct(), and SpinLockInit.

Referenced by CreateSharedMemoryAndSemaphores().

133 {
134  bool found;
135 
136  if (max_replication_slots == 0)
137  return;
138 
140  ShmemInitStruct("ReplicationSlot Ctl", ReplicationSlotsShmemSize(),
141  &found);
142 
143  if (!found)
144  {
145  int i;
146 
147  /* First time through, so initialize */
149 
150  for (i = 0; i < max_replication_slots; i++)
151  {
153 
154  /* everything else is zeroed by the memset above */
155  SpinLockInit(&slot->mutex);
159  }
160  }
161 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
Size ReplicationSlotsShmemSize(void)
Definition: slot.c:114
#define SpinLockInit(lock)
Definition: spin.h:60
#define MemSet(start, val, len)
Definition: c.h:1008
void ConditionVariableInit(ConditionVariable *cv)
void * ShmemInitStruct(const char *name, Size size, bool *foundPtr)
Definition: shmem.c:396
void LWLockInitialize(LWLock *lock, int tranche_id)
Definition: lwlock.c:736
int max_replication_slots
Definition: slot.c:99
ConditionVariable active_cv
Definition: slot.h:153
int i
ReplicationSlot replication_slots[1]
Definition: slot.h:180
slock_t mutex
Definition: slot.h:120
LWLock io_in_progress_lock
Definition: slot.h:150

◆ ReplicationSlotsShmemSize()

Size ReplicationSlotsShmemSize ( void  )

Definition at line 114 of file slot.c.

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

Referenced by CalculateShmemSize(), and ReplicationSlotsShmemInit().

115 {
116  Size size = 0;
117 
118  if (max_replication_slots == 0)
119  return size;
120 
121  size = offsetof(ReplicationSlotCtlData, replication_slots);
122  size = add_size(size,
124 
125  return size;
126 }
Size mul_size(Size s1, Size s2)
Definition: shmem.c:519
Size add_size(Size s1, Size s2)
Definition: shmem.c:502
int max_replication_slots
Definition: slot.c:99
size_t Size
Definition: c.h:540
#define offsetof(type, field)
Definition: c.h:727

◆ ReplicationSlotValidateName()

bool ReplicationSlotValidateName ( const char *  name,
int  elevel 
)

Definition at line 172 of file slot.c.

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

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

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

◆ RestoreSlotFromDisk()

static void RestoreSlotFromDisk ( const char *  name)
static

Definition at line 1670 of file slot.c.

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, ReplicationSlotPersistentData::name, NameStr, OpenTransientFile(), PANIC, ReplicationSlotPersistentData::persistency, PG_BINARY, pg_fsync(), pgstat_report_wait_end(), pgstat_report_wait_start(), read, ReplicationSlotCtlData::replication_slots, ReplicationSlotOnDiskConstantSize, ReplicationSlotOnDiskV2Size, rmtree(), RS_PERSISTENT, SLOT_MAGIC, SLOT_VERSION, ReplicationSlotOnDisk::slotdata, SnapBuildOnDiskChecksummedSize, SnapBuildOnDiskNotChecksummedSize, 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().

1671 {
1673  int i;
1674  char slotdir[MAXPGPATH + 12];
1675  char path[MAXPGPATH + 22];
1676  int fd;
1677  bool restored = false;
1678  int readBytes;
1679  pg_crc32c checksum;
1680 
1681  /* no need to lock here, no concurrent access allowed yet */
1682 
1683  /* delete temp file if it exists */
1684  sprintf(slotdir, "pg_replslot/%s", name);
1685  sprintf(path, "%s/state.tmp", slotdir);
1686  if (unlink(path) < 0 && errno != ENOENT)
1687  ereport(PANIC,
1689  errmsg("could not remove file \"%s\": %m", path)));
1690 
1691  sprintf(path, "%s/state", slotdir);
1692 
1693  elog(DEBUG1, "restoring replication slot from \"%s\"", path);
1694 
1695  /* on some operating systems fsyncing a file requires O_RDWR */
1696  fd = OpenTransientFile(path, O_RDWR | PG_BINARY);
1697 
1698  /*
1699  * We do not need to handle this as we are rename()ing the directory into
1700  * place only after we fsync()ed the state file.
1701  */
1702  if (fd < 0)
1703  ereport(PANIC,
1705  errmsg("could not open file \"%s\": %m", path)));
1706 
1707  /*
1708  * Sync state file before we're reading from it. We might have crashed
1709  * while it wasn't synced yet and we shouldn't continue on that basis.
1710  */
1712  if (pg_fsync(fd) != 0)
1713  ereport(PANIC,
1715  errmsg("could not fsync file \"%s\": %m",
1716  path)));
1718 
1719  /* Also sync the parent directory */
1721  fsync_fname(slotdir, true);
1722  END_CRIT_SECTION();
1723 
1724  /* read part of statefile that's guaranteed to be version independent */
1726  readBytes = read(fd, &cp, ReplicationSlotOnDiskConstantSize);
1728  if (readBytes != ReplicationSlotOnDiskConstantSize)
1729  {
1730  if (readBytes < 0)
1731  ereport(PANIC,
1733  errmsg("could not read file \"%s\": %m", path)));
1734  else
1735  ereport(PANIC,
1737  errmsg("could not read file \"%s\": read %d of %zu",
1738  path, readBytes,
1740  }
1741 
1742  /* verify magic */
1743  if (cp.magic != SLOT_MAGIC)
1744  ereport(PANIC,
1746  errmsg("replication slot file \"%s\" has wrong magic number: %u instead of %u",
1747  path, cp.magic, SLOT_MAGIC)));
1748 
1749  /* verify version */
1750  if (cp.version != SLOT_VERSION)
1751  ereport(PANIC,
1753  errmsg("replication slot file \"%s\" has unsupported version %u",
1754  path, cp.version)));
1755 
1756  /* boundary check on length */
1758  ereport(PANIC,
1760  errmsg("replication slot file \"%s\" has corrupted length %u",
1761  path, cp.length)));
1762 
1763  /* Now that we know the size, read the entire file */
1765  readBytes = read(fd,
1766  (char *) &cp + ReplicationSlotOnDiskConstantSize,
1767  cp.length);
1769  if (readBytes != cp.length)
1770  {
1771  if (readBytes < 0)
1772  ereport(PANIC,
1774  errmsg("could not read file \"%s\": %m", path)));
1775  else
1776  ereport(PANIC,
1778  errmsg("could not read file \"%s\": read %d of %zu",
1779  path, readBytes, (Size) cp.length)));
1780  }
1781 
1782  if (CloseTransientFile(fd) != 0)
1783  ereport(PANIC,
1785  errmsg("could not close file \"%s\": %m", path)));
1786 
1787  /* now verify the CRC */
1788  INIT_CRC32C(checksum);
1789  COMP_CRC32C(checksum,
1790  (char *) &cp + SnapBuildOnDiskNotChecksummedSize,
1792  FIN_CRC32C(checksum);
1793 
1794  if (!EQ_CRC32C(checksum, cp.checksum))
1795  ereport(PANIC,
1796  (errmsg("checksum mismatch for replication slot file \"%s\": is %u, should be %u",
1797  path, checksum, cp.checksum)));
1798 
1799  /*
1800  * If we crashed with an ephemeral slot active, don't restore but delete
1801  * it.
1802  */
1804  {
1805  if (!rmtree(slotdir, true))
1806  {
1807  ereport(WARNING,
1808  (errmsg("could not remove directory \"%s\"",
1809  slotdir)));
1810  }
1811  fsync_fname("pg_replslot", true);
1812  return;
1813  }
1814 
1815  /*
1816  * Verify that requirements for the specific slot type are met. That's
1817  * important because if these aren't met we're not guaranteed to retain
1818  * all the necessary resources for the slot.
1819  *
1820  * NB: We have to do so *after* the above checks for ephemeral slots,
1821  * because otherwise a slot that shouldn't exist anymore could prevent
1822  * restarts.
1823  *
1824  * NB: Changing the requirements here also requires adapting
1825  * CheckSlotRequirements() and CheckLogicalDecodingRequirements().
1826  */
1828  ereport(FATAL,
1829  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1830  errmsg("logical replication slot \"%s\" exists, but wal_level < logical",
1831  NameStr(cp.slotdata.name)),
1832  errhint("Change wal_level to be logical or higher.")));
1833  else if (wal_level < WAL_LEVEL_REPLICA)
1834  ereport(FATAL,
1835  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1836  errmsg("physical replication slot \"%s\" exists, but wal_level < replica",
1837  NameStr(cp.slotdata.name)),
1838  errhint("Change wal_level to be replica or higher.")));
1839 
1840  /* nothing can be active yet, don't lock anything */
1841  for (i = 0; i < max_replication_slots; i++)
1842  {
1843  ReplicationSlot *slot;
1844 
1846 
1847  if (slot->in_use)
1848  continue;
1849 
1850  /* restore the entire set of persistent data */
1851  memcpy(&slot->data, &cp.slotdata,
1853 
1854  /* initialize in memory state */
1855  slot->effective_xmin = cp.slotdata.xmin;
1857 
1862 
1863  slot->in_use = true;
1864  slot->active_pid = 0;
1865 
1866  restored = true;
1867  break;
1868  }
1869 
1870  if (!restored)
1871  ereport(FATAL,
1872  (errmsg("too many replication slots active before shutdown"),
1873  errhint("Increase max_replication_slots and try again.")));
1874 }
#define INIT_CRC32C(crc)
Definition: pg_crc32c.h:41
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
TransactionId candidate_catalog_xmin
Definition: slot.h:162
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
#define DEBUG1
Definition: elog.h:25
int errhint(const char *fmt,...)
Definition: elog.c:1156
static void pgstat_report_wait_end(void)
Definition: wait_event.h:278
#define SLOT_MAGIC
Definition: slot.c:89
uint32 pg_crc32c
Definition: pg_crc32c.h:38
int wal_level
Definition: xlog.c:108
#define END_CRIT_SECTION()
Definition: miscadmin.h:149
void fsync_fname(const char *fname, bool isdir)
Definition: fd.c:671
ReplicationSlotPersistency persistency
Definition: slot.h:54
#define START_CRIT_SECTION()
Definition: miscadmin.h:147
int errcode(int sqlerrcode)
Definition: elog.c:698
#define SnapBuildOnDiskNotChecksummedSize
Definition: slot.c:80
ReplicationSlotPersistentData data
Definition: slot.h:147
#define PANIC
Definition: elog.h:50
static int fd(const char *x, int i)
Definition: preproc-init.c:105
#define PG_BINARY
Definition: c.h:1271
#define sprintf
Definition: port.h:219
ReplicationSlotPersistentData slotdata
Definition: slot.c:73
int OpenTransientFile(const char *fileName, int fileFlags)
Definition: fd.c:2509
#define FATAL
Definition: elog.h:49
#define MAXPGPATH
TransactionId effective_xmin
Definition: slot.h:143
XLogRecPtr candidate_restart_valid
Definition: slot.h:164
int errcode_for_file_access(void)
Definition: elog.c:721
TransactionId catalog_xmin
Definition: slot.h:70
#define InvalidTransactionId
Definition: transam.h:31
TransactionId xmin
Definition: slot.h:62
#define EQ_CRC32C(c1, c2)
Definition: pg_crc32c.h:42
pg_crc32c checksum
Definition: slot.c:62
static void pgstat_report_wait_start(uint32 wait_event_info)
Definition: wait_event.h:262
#define ERRCODE_DATA_CORRUPTED
Definition: pg_basebackup.c:47
int CloseTransientFile(int fd)
Definition: fd.c:2686
#define WARNING
Definition: elog.h:40
bool rmtree(const char *path, bool rmtopdir)
Definition: rmtree.c:42
bool in_use
Definition: slot.h:123
TransactionId effective_catalog_xmin
Definition: slot.h:144
#define SLOT_VERSION
Definition: slot.c:90
#define InvalidOid
Definition: postgres_ext.h:36
#define ereport(elevel,...)
Definition: elog.h:157
int max_replication_slots
Definition: slot.c:99
#define ReplicationSlotOnDiskV2Size
Definition: slot.c:86
#define ReplicationSlotOnDiskConstantSize
Definition: slot.c:77
size_t Size
Definition: c.h:540
uint32 version
Definition: slot.c:65
#define SnapBuildOnDiskChecksummedSize
Definition: slot.c:83
const char * name
Definition: encode.c:561
XLogRecPtr candidate_xmin_lsn
Definition: slot.h:163
int errmsg(const char *fmt,...)
Definition: elog.c:909
pid_t active_pid
Definition: slot.h:126
#define elog(elevel,...)
Definition: elog.h:232
int i
#define NameStr(name)
Definition: c.h:681
int pg_fsync(int fd)
Definition: fd.c:357
ReplicationSlot replication_slots[1]
Definition: slot.h:180
#define COMP_CRC32C(crc, data, len)
Definition: pg_crc32c.h:89
#define FIN_CRC32C(crc)
Definition: pg_crc32c.h:94
#define read(a, b, c)
Definition: win32.h:13
XLogRecPtr candidate_restart_lsn
Definition: slot.h:165

◆ SaveSlotToPath()

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

Definition at line 1516 of file slot.c.

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(), ReplicationSlotOnDiskV2Size, SLOT_MAGIC, SLOT_VERSION, ReplicationSlotOnDisk::slotdata, SnapBuildOnDiskChecksummedSize, SnapBuildOnDiskNotChecksummedSize, 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().

1517 {
1518  char tmppath[MAXPGPATH];
1519  char path[MAXPGPATH];
1520  int fd;
1522  bool was_dirty;
1523 
1524  /* first check whether there's something to write out */
1525  SpinLockAcquire(&slot->mutex);
1526  was_dirty = slot->dirty;
1527  slot->just_dirtied = false;
1528  SpinLockRelease(&slot->mutex);
1529 
1530  /* and don't do anything if there's nothing to write */
1531  if (!was_dirty)
1532  return;
1533 
1535 
1536  /* silence valgrind :( */
1537  memset(&cp, 0, sizeof(ReplicationSlotOnDisk));
1538 
1539  sprintf(tmppath, "%s/state.tmp", dir);
1540  sprintf(path, "%s/state", dir);
1541 
1542  fd = OpenTransientFile(tmppath, O_CREAT | O_EXCL | O_WRONLY | PG_BINARY);
1543  if (fd < 0)
1544  {
1545  /*
1546  * If not an ERROR, then release the lock before returning. In case
1547  * of an ERROR, the error recovery path automatically releases the
1548  * lock, but no harm in explicitly releasing even in that case. Note
1549  * that LWLockRelease() could affect errno.
1550  */
1551  int save_errno = errno;
1552 
1554  errno = save_errno;
1555  ereport(elevel,
1557  errmsg("could not create file \"%s\": %m",
1558  tmppath)));
1559  return;
1560  }
1561 
1562  cp.magic = SLOT_MAGIC;
1563  INIT_CRC32C(cp.checksum);
1564  cp.version = SLOT_VERSION;
1566 
1567  SpinLockAcquire(&slot->mutex);
1568 
1569  memcpy(&cp.slotdata, &slot->data, sizeof(ReplicationSlotPersistentData));
1570 
1571  SpinLockRelease(&slot->mutex);
1572 
1573  COMP_CRC32C(cp.checksum,
1574  (char *) (&cp) + SnapBuildOnDiskNotChecksummedSize,
1576  FIN_CRC32C(cp.checksum);
1577 
1578  errno = 0;
1580  if ((write(fd, &cp, sizeof(cp))) != sizeof(cp))
1581  {
1582  int save_errno = errno;
1583 
1585  CloseTransientFile(fd);
1587 
1588  /* if write didn't set errno, assume problem is no disk space */
1589  errno = save_errno ? save_errno : ENOSPC;
1590  ereport(elevel,
1592  errmsg("could not write to file \"%s\": %m",
1593  tmppath)));
1594  return;
1595  }
1597 
1598  /* fsync the temporary file */
1600  if (pg_fsync(fd) != 0)
1601  {
1602  int save_errno = errno;
1603 
1605  CloseTransientFile(fd);
1607  errno = save_errno;
1608  ereport(elevel,
1610  errmsg("could not fsync file \"%s\": %m",
1611  tmppath)));
1612  return;
1613  }
1615 
1616  if (CloseTransientFile(fd) != 0)
1617  {
1618  int save_errno = errno;
1619 
1621  errno = save_errno;
1622  ereport(elevel,
1624  errmsg("could not close file \"%s\": %m",
1625  tmppath)));
1626  return;
1627  }
1628 
1629  /* rename to permanent file, fsync file and directory */
1630  if (rename(tmppath, path) != 0)
1631  {
1632  int save_errno = errno;
1633 
1635  errno = save_errno;
1636  ereport(elevel,
1638  errmsg("could not rename file \"%s\" to \"%s\": %m",
1639  tmppath, path)));
1640  return;
1641  }
1642 
1643  /*
1644  * Check CreateSlotOnDisk() for the reasoning of using a critical section.
1645  */
1647 
1648  fsync_fname(path, false);
1649  fsync_fname(dir, true);
1650  fsync_fname("pg_replslot", true);
1651 
1652  END_CRIT_SECTION();
1653 
1654  /*
1655  * Successfully wrote, unset dirty bit, unless somebody dirtied again
1656  * already.
1657  */
1658  SpinLockAcquire(&slot->mutex);
1659  if (!slot->just_dirtied)
1660  slot->dirty = false;
1661  SpinLockRelease(&slot->mutex);
1662 
1664 }
#define INIT_CRC32C(crc)
Definition: pg_crc32c.h:41
static void pgstat_report_wait_end(void)
Definition: wait_event.h:278
#define write(a, b, c)
Definition: win32.h:14
#define SLOT_MAGIC
Definition: slot.c:89
#define END_CRIT_SECTION()
Definition: miscadmin.h:149
void fsync_fname(const char *fname, bool isdir)
Definition: fd.c:671
#define START_CRIT_SECTION()
Definition: miscadmin.h:147
#define SnapBuildOnDiskNotChecksummedSize
Definition: slot.c:80
ReplicationSlotPersistentData data
Definition: slot.h:147
static int fd(const char *x, int i)
Definition: preproc-init.c:105
#define PG_BINARY
Definition: c.h:1271
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1803
#define sprintf
Definition: port.h:219
#define SpinLockAcquire(lock)
Definition: spin.h:62
ReplicationSlotPersistentData slotdata
Definition: slot.c:73
int OpenTransientFile(const char *fileName, int fileFlags)
Definition: fd.c:2509
#define MAXPGPATH
int errcode_for_file_access(void)
Definition: elog.c:721
pg_crc32c checksum
Definition: slot.c:62
static void pgstat_report_wait_start(uint32 wait_event_info)
Definition: wait_event.h:262
int CloseTransientFile(int fd)
Definition: fd.c:2686
static int elevel
Definition: vacuumlazy.c:401
#define SpinLockRelease(lock)
Definition: spin.h:64
bool just_dirtied
Definition: slot.h:129
#define SLOT_VERSION
Definition: slot.c:90
#define ereport(elevel,...)
Definition: elog.h:157
#define ReplicationSlotOnDiskV2Size
Definition: slot.c:86
uint32 version
Definition: slot.c:65
#define SnapBuildOnDiskChecksummedSize
Definition: slot.c:83
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1199
int errmsg(const char *fmt,...)
Definition: elog.c:909
int pg_fsync(int fd)
Definition: fd.c:357
slock_t mutex
Definition: slot.h:120
#define COMP_CRC32C(crc, data, len)
Definition: pg_crc32c.h:89
#define FIN_CRC32C(crc)
Definition: pg_crc32c.h:94
bool dirty
Definition: slot.h:130
LWLock io_in_progress_lock
Definition: slot.h:150

◆ SearchNamedReplicationSlot()

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

Definition at line 348 of file slot.c.

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

Referenced by pg_stat_reset_replication_slot(), pgstat_vacuum_stat(), and ReplicationSlotAcquire().

349 {
350  int i;
351  ReplicationSlot *slot = NULL;
352 
353  if (need_lock)
354  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
355 
356  for (i = 0; i < max_replication_slots; i++)
357  {
359 
360  if (s->in_use && strcmp(name, NameStr(s->data.name)) == 0)
361  {
362  slot = s;
363  break;
364  }
365  }
366 
367  if (need_lock)
368  LWLockRelease(ReplicationSlotControlLock);
369 
370  return slot;
371 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
ReplicationSlotPersistentData data
Definition: slot.h:147
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1803
bool in_use
Definition: slot.h:123
int max_replication_slots
Definition: slot.c:99
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1199
const char * name
Definition: encode.c:561
int i
#define NameStr(name)
Definition: c.h:681
ReplicationSlot replication_slots[1]
Definition: slot.h:180

◆ StartupReplicationSlots()

void StartupReplicationSlots ( void  )

Definition at line 1395 of file slot.c.

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

1396 {
1397  DIR *replication_dir;
1398  struct dirent *replication_de;
1399 
1400  elog(DEBUG1, "starting up replication slots");
1401 
1402  /* restore all slots by iterating over all on-disk entries */
1403  replication_dir = AllocateDir("pg_replslot");
1404  while ((replication_de = ReadDir(replication_dir, "pg_replslot")) != NULL)
1405  {
1406  struct stat statbuf;
1407  char path[MAXPGPATH + 12];
1408 
1409  if (strcmp(replication_de->d_name, ".") == 0 ||
1410  strcmp(replication_de->d_name, "..") == 0)
1411  continue;
1412 
1413  snprintf(path, sizeof(path), "pg_replslot/%s", replication_de->d_name);
1414 
1415  /* we're only creating directories here, skip if it's not our's */
1416  if (lstat(path, &statbuf) == 0 && !S_ISDIR(statbuf.st_mode))
1417  continue;
1418 
1419  /* we crashed while a slot was being setup or deleted, clean up */
1420  if (pg_str_endswith(replication_de->d_name, ".tmp"))
1421  {
1422  if (!rmtree(path, true))
1423  {
1424  ereport(WARNING,
1425  (errmsg("could not remove directory \"%s\"",
1426  path)));
1427  continue;
1428  }
1429  fsync_fname("pg_replslot", true);
1430  continue;
1431  }
1432 
1433  /* looks like a slot in a normal state, restore */
1434  RestoreSlotFromDisk(replication_de->d_name);
1435  }
1436  FreeDir(replication_dir);
1437 
1438  /* currently no slots exist, we're done. */
1439  if (max_replication_slots <= 0)
1440  return;
1441 
1442  /* Now that we have recovered all the data, compute replication xmin */
1445 }
static void RestoreSlotFromDisk(const char *name)
Definition: slot.c:1670
#define DEBUG1
Definition: elog.h:25
bool pg_str_endswith(const char *str, const char *end)
Definition: string.c:31
void fsync_fname(const char *fname, bool isdir)
Definition: fd.c:671
Definition: dirent.h:9
void ReplicationSlotsComputeRequiredLSN(void)
Definition: slot.c:817
Definition: dirent.c:25
#define MAXPGPATH
DIR * AllocateDir(const char *dirname)
Definition: fd.c:2720
#define WARNING
Definition: elog.h:40
bool rmtree(const char *path, bool rmtopdir)
Definition: rmtree.c:42
#define ereport(elevel,...)
Definition: elog.h:157
int max_replication_slots
Definition: slot.c:99
struct dirent * ReadDir(DIR *dir, const char *dirname)
Definition: fd.c:2786
#define S_ISDIR(m)
Definition: win32_port.h:324
#define lstat(path, sb)
Definition: win32_port.h:284
int errmsg(const char *fmt,...)
Definition: elog.c:909
#define elog(elevel,...)
Definition: elog.h:232
char d_name[MAX_PATH]
Definition: dirent.h:15
#define snprintf
Definition: port.h:217
void ReplicationSlotsComputeRequiredXmin(bool already_locked)
Definition: slot.c:767
int FreeDir(DIR *dir)
Definition: fd.c:2838

Variable Documentation

◆ max_replication_slots

◆ MyReplicationSlot

◆ ReplicationSlotCtl

ReplicationSlotCtlData* ReplicationSlotCtl = NULL

Definition at line 93 of file slot.c.

Referenced by copy_replication_slot(), and pg_get_replication_slots().