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 int ReplicationSlotAcquireInternal (ReplicationSlot *slot, const char *name, SlotAcquireBehavior behavior)
 
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)
 
int ReplicationSlotAcquire (const char *name, SlotAcquireBehavior behavior)
 
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 ReplicationSlotReserveWal (void)
 
void 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 1285 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().

1286 {
1287  int i;
1288 
1289  elog(DEBUG1, "performing replication slot checkpoint");
1290 
1291  /*
1292  * Prevent any slot from being created/dropped while we're active. As we
1293  * explicitly do *not* want to block iterating over replication_slots or
1294  * acquiring a slot we cannot take the control lock - but that's OK,
1295  * because holding ReplicationSlotAllocationLock is strictly stronger, and
1296  * enough to guarantee that nobody can change the in_use bits on us.
1297  */
1298  LWLockAcquire(ReplicationSlotAllocationLock, LW_SHARED);
1299 
1300  for (i = 0; i < max_replication_slots; i++)
1301  {
1303  char path[MAXPGPATH];
1304 
1305  if (!s->in_use)
1306  continue;
1307 
1308  /* save the slot to disk, locking is handled in SaveSlotToPath() */
1309  sprintf(path, "pg_replslot/%s", NameStr(s->data.name));
1310  SaveSlotToPath(s, path, LOG);
1311  }
1312  LWLockRelease(ReplicationSlotAllocationLock);
1313 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
#define DEBUG1
Definition: elog.h:25
ReplicationSlotPersistentData data
Definition: slot.h:156
#define LOG
Definition: elog.h:26
static void SaveSlotToPath(ReplicationSlot *slot, const char *path, int elevel)
Definition: slot.c:1441
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1816
#define sprintf
Definition: port.h:218
#define MAXPGPATH
bool in_use
Definition: slot.h:132
int max_replication_slots
Definition: slot.c:99
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1203
#define elog(elevel,...)
Definition: elog.h:232
int i
#define NameStr(name)
Definition: c.h:681
ReplicationSlot replication_slots[1]
Definition: slot.h:189

◆ CheckSlotRequirements()

void CheckSlotRequirements ( void  )

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

1069 {
1070  /*
1071  * NB: Adding a new requirement likely means that RestoreSlotFromDisk()
1072  * needs the same check.
1073  */
1074 
1075  if (max_replication_slots == 0)
1076  ereport(ERROR,
1077  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1078  errmsg("replication slots can only be used if max_replication_slots > 0")));
1079 
1081  ereport(ERROR,
1082  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1083  errmsg("replication slots can only be used if wal_level >= replica")));
1084 }
int wal_level
Definition: xlog.c:109
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 1380 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().

1381 {
1382  char tmppath[MAXPGPATH];
1383  char path[MAXPGPATH];
1384  struct stat st;
1385 
1386  /*
1387  * No need to take out the io_in_progress_lock, nobody else can see this
1388  * slot yet, so nobody else will write. We're reusing SaveSlotToPath which
1389  * takes out the lock, if we'd take the lock here, we'd deadlock.
1390  */
1391 
1392  sprintf(path, "pg_replslot/%s", NameStr(slot->data.name));
1393  sprintf(tmppath, "pg_replslot/%s.tmp", NameStr(slot->data.name));
1394 
1395  /*
1396  * It's just barely possible that some previous effort to create or drop a
1397  * slot with this name left a temp directory lying around. If that seems
1398  * to be the case, try to remove it. If the rmtree() fails, we'll error
1399  * out at the MakePGDirectory() below, so we don't bother checking
1400  * success.
1401  */
1402  if (stat(tmppath, &st) == 0 && S_ISDIR(st.st_mode))
1403  rmtree(tmppath, true);
1404 
1405  /* Create and fsync the temporary slot directory. */
1406  if (MakePGDirectory(tmppath) < 0)
1407  ereport(ERROR,
1409  errmsg("could not create directory \"%s\": %m",
1410  tmppath)));
1411  fsync_fname(tmppath, true);
1412 
1413  /* Write the actual state file. */
1414  slot->dirty = true; /* signal that we really need to write */
1415  SaveSlotToPath(slot, tmppath, ERROR);
1416 
1417  /* Rename the directory into place. */
1418  if (rename(tmppath, path) != 0)
1419  ereport(ERROR,
1421  errmsg("could not rename file \"%s\" to \"%s\": %m",
1422  tmppath, path)));
1423 
1424  /*
1425  * If we'd now fail - really unlikely - we wouldn't know whether this slot
1426  * would persist after an OS crash or not - so, force a restart. The
1427  * restart would try to fsync this again till it works.
1428  */
1430 
1431  fsync_fname(path, true);
1432  fsync_fname("pg_replslot", true);
1433 
1434  END_CRIT_SECTION();
1435 }
#define END_CRIT_SECTION()
Definition: miscadmin.h:137
void fsync_fname(const char *fname, bool isdir)
Definition: fd.c:666
#define START_CRIT_SECTION()
Definition: miscadmin.h:135
ReplicationSlotPersistentData data
Definition: slot.h:156
static void SaveSlotToPath(ReplicationSlot *slot, const char *path, int elevel)
Definition: slot.c:1441
#define sprintf
Definition: port.h:218
#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:3691
#define S_ISDIR(m)
Definition: win32_port.h:316
int errmsg(const char *fmt,...)
Definition: elog.c:909
#define NameStr(name)
Definition: c.h:681
bool dirty
Definition: slot.h:139
#define stat
Definition: win32_port.h:275

◆ InvalidateObsoleteReplicationSlots()

void InvalidateObsoleteReplicationSlots ( XLogSegNo  oldestSegno)

Definition at line 1170 of file slot.c.

References ReplicationSlot::active_cv, CHECK_FOR_INTERRUPTS, ConditionVariableCancelSleep(), ConditionVariablePrepareToSleep(), ConditionVariableTimedSleep(), ReplicationSlot::data, ereport, errmsg(), i, ReplicationSlot::in_use, ReplicationSlotPersistentData::invalidated_at, InvalidXLogRecPtr, kill, LOG, LSN_FORMAT_ARGS, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_replication_slots, ReplicationSlot::mutex, ReplicationSlotPersistentData::name, NameStr, ReplicationSlotCtlData::replication_slots, ReplicationSlotAcquireInternal(), ReplicationSlotMarkDirty(), ReplicationSlotRelease(), ReplicationSlotSave(), ReplicationSlotPersistentData::restart_lsn, SAB_Inquire, SpinLockAcquire, SpinLockRelease, WAIT_EVENT_REPLICATION_SLOT_DROP, wal_segment_size, XLogRecPtrIsInvalid, and XLogSegNoOffsetToRecPtr.

Referenced by CreateCheckPoint(), and CreateRestartPoint().

1171 {
1172  XLogRecPtr oldestLSN;
1173 
1174  XLogSegNoOffsetToRecPtr(oldestSegno, 0, wal_segment_size, oldestLSN);
1175 
1176 restart:
1177  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
1178  for (int i = 0; i < max_replication_slots; i++)
1179  {
1181  XLogRecPtr restart_lsn = InvalidXLogRecPtr;
1182  NameData slotname;
1183  int wspid;
1184  int last_signaled_pid = 0;
1185 
1186  if (!s->in_use)
1187  continue;
1188 
1189  SpinLockAcquire(&s->mutex);
1190  slotname = s->data.name;
1191  restart_lsn = s->data.restart_lsn;
1192  SpinLockRelease(&s->mutex);
1193 
1194  if (XLogRecPtrIsInvalid(restart_lsn) || restart_lsn >= oldestLSN)
1195  continue;
1196  LWLockRelease(ReplicationSlotControlLock);
1198 
1199  /* Get ready to sleep on the slot in case it is active */
1201 
1202  for (;;)
1203  {
1204  /*
1205  * Try to mark this slot as used by this process.
1206  *
1207  * Note that ReplicationSlotAcquireInternal(SAB_Inquire)
1208  * should not cancel the prepared condition variable
1209  * if this slot is active in other process. Because in this case
1210  * we have to wait on that CV for the process owning
1211  * the slot to be terminated, later.
1212  */
1213  wspid = ReplicationSlotAcquireInternal(s, NULL, SAB_Inquire);
1214 
1215  /*
1216  * Exit the loop if we successfully acquired the slot or
1217  * the slot was dropped during waiting for the owning process
1218  * to be terminated. For example, the latter case is likely to
1219  * happen when the slot is temporary because it's automatically
1220  * dropped by the termination of the owning process.
1221  */
1222  if (wspid <= 0)
1223  break;
1224 
1225  /*
1226  * Signal to terminate the process that owns the slot.
1227  *
1228  * There is the race condition where other process may own
1229  * the slot after the process using it was terminated and before
1230  * this process owns it. To handle this case, we signal again
1231  * if the PID of the owning process is changed than the last.
1232  *
1233  * XXX This logic assumes that the same PID is not reused
1234  * very quickly.
1235  */
1236  if (last_signaled_pid != wspid)
1237  {
1238  ereport(LOG,
1239  (errmsg("terminating process %d because replication slot \"%s\" is too far behind",
1240  wspid, NameStr(slotname))));
1241  (void) kill(wspid, SIGTERM);
1242  last_signaled_pid = wspid;
1243  }
1244 
1247  }
1249 
1250  /*
1251  * Do nothing here and start from scratch if the slot has
1252  * already been dropped.
1253  */
1254  if (wspid == -1)
1255  goto restart;
1256 
1257  ereport(LOG,
1258  (errmsg("invalidating slot \"%s\" because its restart_lsn %X/%X exceeds max_slot_wal_keep_size",
1259  NameStr(slotname),
1260  LSN_FORMAT_ARGS(restart_lsn))));
1261 
1262  SpinLockAcquire(&s->mutex);
1265  SpinLockRelease(&s->mutex);
1266 
1267  /* Make sure the invalidated state persists across server restart */
1271 
1272  /* if we did anything, start from scratch */
1273  goto restart;
1274  }
1275  LWLockRelease(ReplicationSlotControlLock);
1276 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
int wal_segment_size
Definition: xlog.c:121
#define kill(pid, sig)
Definition: win32_port.h:454
void ReplicationSlotSave(void)
Definition: slot.c:734
ReplicationSlotPersistentData data
Definition: slot.h:156
#define LOG
Definition: elog.h:26
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1816
void ConditionVariablePrepareToSleep(ConditionVariable *cv)
#define SpinLockAcquire(lock)
Definition: spin.h:62
#define LSN_FORMAT_ARGS(lsn)
Definition: xlogdefs.h:43
void ConditionVariableCancelSleep(void)
bool ConditionVariableTimedSleep(ConditionVariable *cv, long timeout, uint32 wait_event_info)
Definition: c.h:675
void ReplicationSlotRelease(void)
Definition: slot.c:493
#define XLogRecPtrIsInvalid(r)
Definition: xlogdefs.h:29
bool in_use
Definition: slot.h:132
#define SpinLockRelease(lock)
Definition: spin.h:64
#define ereport(elevel,...)
Definition: elog.h:157
int max_replication_slots
Definition: slot.c:99
#define XLogSegNoOffsetToRecPtr(segno, offset, wal_segsz_bytes, dest)
uint64 XLogRecPtr
Definition: xlogdefs.h:21
XLogRecPtr restart_lsn
Definition: slot.h:81
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1203
static int ReplicationSlotAcquireInternal(ReplicationSlot *slot, const char *name, SlotAcquireBehavior behavior)
Definition: slot.c:398
ConditionVariable active_cv
Definition: slot.h:162
int errmsg(const char *fmt,...)
Definition: elog.c:909
int i
#define NameStr(name)
Definition: c.h:681
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:102
ReplicationSlot replication_slots[1]
Definition: slot.h:189
XLogRecPtr invalidated_at
Definition: slot.h:84
slock_t mutex
Definition: slot.h:129
void ReplicationSlotMarkDirty(void)
Definition: slot.c:752

◆ ReplicationSlotAcquire()

int ReplicationSlotAcquire ( const char *  name,
SlotAcquireBehavior  behavior 
)

Definition at line 384 of file slot.c.

References ReplicationSlotAcquireInternal().

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

385 {
386  return ReplicationSlotAcquireInternal(NULL, name, behavior);
387 }
static int ReplicationSlotAcquireInternal(ReplicationSlot *slot, const char *name, SlotAcquireBehavior behavior)
Definition: slot.c:398
const char * name
Definition: encode.c:515

◆ ReplicationSlotAcquireInternal()

static int ReplicationSlotAcquireInternal ( ReplicationSlot slot,
const char *  name,
SlotAcquireBehavior  behavior 
)
static

Definition at line 398 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, SAB_Block, SAB_Error, SAB_Inquire, SearchNamedReplicationSlot(), SpinLockAcquire, SpinLockRelease, and WAIT_EVENT_REPLICATION_SLOT_DROP.

Referenced by InvalidateObsoleteReplicationSlots(), and ReplicationSlotAcquire().

400 {
401  ReplicationSlot *s;
402  int active_pid;
403 
404  AssertArg((slot == NULL) ^ (name == NULL));
405 
406 retry:
407  Assert(MyReplicationSlot == NULL);
408 
409  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
410 
411  /*
412  * Search for the slot with the specified name if the slot to acquire is
413  * not given. If the slot is not found, we either return -1 or error out.
414  */
415  s = slot ? slot : SearchNamedReplicationSlot(name, false);
416  if (s == NULL || !s->in_use)
417  {
418  LWLockRelease(ReplicationSlotControlLock);
419 
420  if (behavior == SAB_Inquire)
421  return -1;
422  ereport(ERROR,
423  (errcode(ERRCODE_UNDEFINED_OBJECT),
424  errmsg("replication slot \"%s\" does not exist",
425  name ? name : NameStr(slot->data.name))));
426  }
427 
428  /*
429  * This is the slot we want; check if it's active under some other
430  * process. In single user mode, we don't need this check.
431  */
432  if (IsUnderPostmaster)
433  {
434  /*
435  * Get ready to sleep on the slot in case it is active if SAB_Block.
436  * (We may end up not sleeping, but we don't want to do this while
437  * holding the spinlock.)
438  */
439  if (behavior == SAB_Block)
441 
442  SpinLockAcquire(&s->mutex);
443  if (s->active_pid == 0)
444  s->active_pid = MyProcPid;
445  active_pid = s->active_pid;
446  SpinLockRelease(&s->mutex);
447  }
448  else
449  active_pid = MyProcPid;
450  LWLockRelease(ReplicationSlotControlLock);
451 
452  /*
453  * If we found the slot but it's already active in another process, we
454  * either error out, return the PID of the owning process, or retry
455  * after a short wait, as caller specified.
456  */
457  if (active_pid != MyProcPid)
458  {
459  if (behavior == SAB_Error)
460  ereport(ERROR,
461  (errcode(ERRCODE_OBJECT_IN_USE),
462  errmsg("replication slot \"%s\" is active for PID %d",
463  NameStr(s->data.name), active_pid)));
464  else if (behavior == SAB_Inquire)
465  return active_pid;
466 
467  /* Wait here until we get signaled, and then restart */
471  goto retry;
472  }
473  else if (behavior == SAB_Block)
474  ConditionVariableCancelSleep(); /* no sleep needed after all */
475 
476  /* Let everybody know we've modified this slot */
478 
479  /* We made this slot active, so it's ours now. */
480  MyReplicationSlot = s;
481 
482  /* success */
483  return 0;
484 }
ReplicationSlot * SearchNamedReplicationSlot(const char *name, bool need_lock)
Definition: slot.c:349
int MyProcPid
Definition: globals.c:43
void ConditionVariableBroadcast(ConditionVariable *cv)
int errcode(int sqlerrcode)
Definition: elog.c:698
ReplicationSlotPersistentData data
Definition: slot.h:156
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1816
void ConditionVariablePrepareToSleep(ConditionVariable *cv)
#define SpinLockAcquire(lock)
Definition: spin.h:62
void ConditionVariableCancelSleep(void)
#define ERROR
Definition: elog.h:46
Definition: slot.h:44
bool IsUnderPostmaster
Definition: globals.c:112
#define AssertArg(condition)
Definition: c.h:806
bool in_use
Definition: slot.h:132
#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:1203
ConditionVariable active_cv
Definition: slot.h:162
const char * name
Definition: encode.c:515
int errmsg(const char *fmt,...)
Definition: elog.c:909
pid_t active_pid
Definition: slot.h:135
Definition: slot.h:43
#define NameStr(name)
Definition: c.h:681
slock_t mutex
Definition: slot.h:129

◆ ReplicationSlotCleanup()

void ReplicationSlotCleanup ( void  )

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

550 {
551  int i;
552 
553  Assert(MyReplicationSlot == NULL);
554 
555 restart:
556  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
557  for (i = 0; i < max_replication_slots; i++)
558  {
560 
561  if (!s->in_use)
562  continue;
563 
564  SpinLockAcquire(&s->mutex);
565  if (s->active_pid == MyProcPid)
566  {
568  SpinLockRelease(&s->mutex);
569  LWLockRelease(ReplicationSlotControlLock); /* avoid deadlock */
570 
572 
574  goto restart;
575  }
576  else
577  SpinLockRelease(&s->mutex);
578  }
579 
580  LWLockRelease(ReplicationSlotControlLock);
581 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
int MyProcPid
Definition: globals.c:43
ReplicationSlotPersistency persistency
Definition: slot.h:62
void ConditionVariableBroadcast(ConditionVariable *cv)
static void ReplicationSlotDropPtr(ReplicationSlot *slot)
Definition: slot.c:617
ReplicationSlotPersistentData data
Definition: slot.h:156
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1816
#define SpinLockAcquire(lock)
Definition: spin.h:62
bool in_use
Definition: slot.h:132
#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:1203
ConditionVariable active_cv
Definition: slot.h:162
pid_t active_pid
Definition: slot.h:135
int i
ReplicationSlot replication_slots[1]
Definition: slot.h:189
slock_t mutex
Definition: slot.h:129

◆ ReplicationSlotCreate()

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

Definition at line 228 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, and ReplicationSlotPersistentData::two_phase.

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

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

◆ ReplicationSlotDrop()

void ReplicationSlotDrop ( const char *  name,
bool  nowait 
)

Definition at line 587 of file slot.c.

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

Referenced by DropReplicationSlot(), and pg_drop_replication_slot().

588 {
589  Assert(MyReplicationSlot == NULL);
590 
591  (void) ReplicationSlotAcquire(name, nowait ? SAB_Error : SAB_Block);
592 
594 }
static void ReplicationSlotDropAcquired(void)
Definition: slot.c:600
Definition: slot.h:44
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
#define Assert(condition)
Definition: c.h:804
const char * name
Definition: encode.c:515
int ReplicationSlotAcquire(const char *name, SlotAcquireBehavior behavior)
Definition: slot.c:384
Definition: slot.h:43

◆ ReplicationSlotDropAcquired()

static void ReplicationSlotDropAcquired ( void  )
static

Definition at line 600 of file slot.c.

References Assert, MyReplicationSlot, and ReplicationSlotDropPtr().

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

601 {
603 
604  Assert(MyReplicationSlot != NULL);
605 
606  /* slot isn't acquired anymore */
607  MyReplicationSlot = NULL;
608 
610 }
static void ReplicationSlotDropPtr(ReplicationSlot *slot)
Definition: slot.c:617
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
#define Assert(condition)
Definition: c.h:804

◆ ReplicationSlotDropPtr()

static void ReplicationSlotDropPtr ( ReplicationSlot slot)
static

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

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

◆ ReplicationSlotMarkDirty()

◆ ReplicationSlotPersist()

void ReplicationSlotPersist ( void  )

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

770 {
772 
773  Assert(slot != NULL);
775 
776  SpinLockAcquire(&slot->mutex);
778  SpinLockRelease(&slot->mutex);
779 
782 }
ReplicationSlotPersistency persistency
Definition: slot.h:62
void ReplicationSlotSave(void)
Definition: slot.c:734
ReplicationSlotPersistentData data
Definition: slot.h:156
#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:129
void ReplicationSlotMarkDirty(void)
Definition: slot.c:752

◆ ReplicationSlotRelease()

void ReplicationSlotRelease ( void  )

Definition at line 493 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(), InvalidateObsoleteReplicationSlots(), 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().

494 {
496 
497  Assert(slot != NULL && slot->active_pid != 0);
498 
499  if (slot->data.persistency == RS_EPHEMERAL)
500  {
501  /*
502  * Delete the slot. There is no !PANIC case where this is allowed to
503  * fail, all that may happen is an incomplete cleanup of the on-disk
504  * data.
505  */
507  }
508 
509  /*
510  * If slot needed to temporarily restrain both data and catalog xmin to
511  * create the catalog snapshot, remove that temporary constraint.
512  * Snapshots can only be exported while the initial snapshot is still
513  * acquired.
514  */
515  if (!TransactionIdIsValid(slot->data.xmin) &&
517  {
518  SpinLockAcquire(&slot->mutex);
520  SpinLockRelease(&slot->mutex);
522  }
523 
524  if (slot->data.persistency == RS_PERSISTENT)
525  {
526  /*
527  * Mark persistent slot inactive. We're not freeing it, just
528  * disconnecting, but wake up others that may be waiting for it.
529  */
530  SpinLockAcquire(&slot->mutex);
531  slot->active_pid = 0;
532  SpinLockRelease(&slot->mutex);
534  }
535 
536  MyReplicationSlot = NULL;
537 
538  /* might not have been set when we've been a plain slot */
539  LWLockAcquire(ProcArrayLock, LW_SHARED);
542  LWLockRelease(ProcArrayLock);
543 }
#define PROC_IN_LOGICAL_DECODING
Definition: proc.h:61
PGPROC * MyProc
Definition: proc.c:68
ReplicationSlotPersistency persistency
Definition: slot.h:62
void ConditionVariableBroadcast(ConditionVariable *cv)
PROC_HDR * ProcGlobal
Definition: proc.c:80
uint8 statusFlags
Definition: proc.h:189
ReplicationSlotPersistentData data
Definition: slot.h:156
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1816
#define SpinLockAcquire(lock)
Definition: spin.h:62
static void ReplicationSlotDropAcquired(void)
Definition: slot.c:600
TransactionId effective_xmin
Definition: slot.h:152
#define InvalidTransactionId
Definition: transam.h:31
TransactionId xmin
Definition: slot.h:70
#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:1203
ConditionVariable active_cv
Definition: slot.h:162
pid_t active_pid
Definition: slot.h:135
int pgxactoff
Definition: proc.h:148
uint8 * statusFlags
Definition: proc.h:333
slock_t mutex
Definition: slot.h:129
#define TransactionIdIsValid(xid)
Definition: transam.h:41
void ReplicationSlotsComputeRequiredXmin(bool already_locked)
Definition: slot.c:791

◆ ReplicationSlotReserveWal()

void ReplicationSlotReserveWal ( void  )

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

1094 {
1096 
1097  Assert(slot != NULL);
1099 
1100  /*
1101  * The replication slot mechanism is used to prevent removal of required
1102  * WAL. As there is no interlock between this routine and checkpoints, WAL
1103  * segments could concurrently be removed when a now stale return value of
1104  * ReplicationSlotsComputeRequiredLSN() is used. In the unlikely case that
1105  * this happens we'll just retry.
1106  */
1107  while (true)
1108  {
1109  XLogSegNo segno;
1110  XLogRecPtr restart_lsn;
1111 
1112  /*
1113  * For logical slots log a standby snapshot and start logical decoding
1114  * at exactly that position. That allows the slot to start up more
1115  * quickly.
1116  *
1117  * That's not needed (or indeed helpful) for physical slots as they'll
1118  * start replay at the last logged checkpoint anyway. Instead return
1119  * the location of the last redo LSN. While that slightly increases
1120  * the chance that we have to retry, it's where a base backup has to
1121  * start replay at.
1122  */
1123  if (!RecoveryInProgress() && SlotIsLogical(slot))
1124  {
1125  XLogRecPtr flushptr;
1126 
1127  /* start at current insert position */
1128  restart_lsn = GetXLogInsertRecPtr();
1129  SpinLockAcquire(&slot->mutex);
1130  slot->data.restart_lsn = restart_lsn;
1131  SpinLockRelease(&slot->mutex);
1132 
1133  /* make sure we have enough information to start */
1134  flushptr = LogStandbySnapshot();
1135 
1136  /* and make sure it's fsynced to disk */
1137  XLogFlush(flushptr);
1138  }
1139  else
1140  {
1141  restart_lsn = GetRedoRecPtr();
1142  SpinLockAcquire(&slot->mutex);
1143  slot->data.restart_lsn = restart_lsn;
1144  SpinLockRelease(&slot->mutex);
1145  }
1146 
1147  /* prevent WAL removal as fast as possible */
1149 
1150  /*
1151  * If all required WAL is still there, great, otherwise retry. The
1152  * slot should prevent further removal of WAL, unless there's a
1153  * concurrent ReplicationSlotsComputeRequiredLSN() after we've written
1154  * the new restart_lsn above, so normally we should never need to loop
1155  * more than twice.
1156  */
1158  if (XLogGetLastRemovedSegno() < segno)
1159  break;
1160  }
1161 }
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
int wal_segment_size
Definition: xlog.c:121
ReplicationSlotPersistentData data
Definition: slot.h:156
bool RecoveryInProgress(void)
Definition: xlog.c:8237
void XLogFlush(XLogRecPtr record)
Definition: xlog.c:2881
#define SpinLockAcquire(lock)
Definition: spin.h:62
XLogSegNo XLogGetLastRemovedSegno(void)
Definition: xlog.c:4007
void ReplicationSlotsComputeRequiredLSN(void)
Definition: slot.c:841
XLogRecPtr LogStandbySnapshot(void)
Definition: standby.c:1220
XLogRecPtr GetXLogInsertRecPtr(void)
Definition: xlog.c:11747
uint64 XLogSegNo
Definition: xlogdefs.h:48
#define SlotIsLogical(slot)
Definition: slot.h:178
#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:81
XLogRecPtr GetRedoRecPtr(void)
Definition: xlog.c:8529
slock_t mutex
Definition: slot.h:129
#define XLByteToSeg(xlrp, logSegNo, wal_segsz_bytes)

◆ ReplicationSlotSave()

void ReplicationSlotSave ( void  )

Definition at line 734 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(), CreateInitDecodingContext(), CreateReplicationSlot(), InvalidateObsoleteReplicationSlots(), LogicalConfirmReceivedLocation(), and ReplicationSlotPersist().

735 {
736  char path[MAXPGPATH];
737 
738  Assert(MyReplicationSlot != NULL);
739 
740  sprintf(path, "pg_replslot/%s", NameStr(MyReplicationSlot->data.name));
742 }
ReplicationSlotPersistentData data
Definition: slot.h:156
static void SaveSlotToPath(ReplicationSlot *slot, const char *path, int elevel)
Definition: slot.c:1441
#define sprintf
Definition: port.h:218
#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 884 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().

885 {
886  XLogRecPtr result = InvalidXLogRecPtr;
887  int i;
888 
889  if (max_replication_slots <= 0)
890  return InvalidXLogRecPtr;
891 
892  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
893 
894  for (i = 0; i < max_replication_slots; i++)
895  {
896  ReplicationSlot *s;
897  XLogRecPtr restart_lsn;
898 
900 
901  /* cannot change while ReplicationSlotCtlLock is held */
902  if (!s->in_use)
903  continue;
904 
905  /* we're only interested in logical slots */
906  if (!SlotIsLogical(s))
907  continue;
908 
909  /* read once, it's ok if it increases while we're checking */
910  SpinLockAcquire(&s->mutex);
911  restart_lsn = s->data.restart_lsn;
912  SpinLockRelease(&s->mutex);
913 
914  if (restart_lsn == InvalidXLogRecPtr)
915  continue;
916 
917  if (result == InvalidXLogRecPtr ||
918  restart_lsn < result)
919  result = restart_lsn;
920  }
921 
922  LWLockRelease(ReplicationSlotControlLock);
923 
924  return result;
925 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
ReplicationSlotPersistentData data
Definition: slot.h:156
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1816
#define SpinLockAcquire(lock)
Definition: spin.h:62
#define SlotIsLogical(slot)
Definition: slot.h:178
bool in_use
Definition: slot.h:132
#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:81
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1203
int i
ReplicationSlot replication_slots[1]
Definition: slot.h:189
slock_t mutex
Definition: slot.h:129

◆ ReplicationSlotsComputeRequiredLSN()

void ReplicationSlotsComputeRequiredLSN ( void  )

Definition at line 841 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(), LogicalConfirmReceivedLocation(), pg_replication_slot_advance(), PhysicalConfirmReceivedLocation(), ReplicationSlotDropPtr(), ReplicationSlotReserveWal(), and StartupReplicationSlots().

842 {
843  int i;
844  XLogRecPtr min_required = InvalidXLogRecPtr;
845 
846  Assert(ReplicationSlotCtl != NULL);
847 
848  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
849  for (i = 0; i < max_replication_slots; i++)
850  {
852  XLogRecPtr restart_lsn;
853 
854  if (!s->in_use)
855  continue;
856 
857  SpinLockAcquire(&s->mutex);
858  restart_lsn = s->data.restart_lsn;
859  SpinLockRelease(&s->mutex);
860 
861  if (restart_lsn != InvalidXLogRecPtr &&
862  (min_required == InvalidXLogRecPtr ||
863  restart_lsn < min_required))
864  min_required = restart_lsn;
865  }
866  LWLockRelease(ReplicationSlotControlLock);
867 
868  XLogSetReplicationSlotMinimumLSN(min_required);
869 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
ReplicationSlotPersistentData data
Definition: slot.h:156
void XLogSetReplicationSlotMinimumLSN(XLogRecPtr lsn)
Definition: xlog.c:2763
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1816
#define SpinLockAcquire(lock)
Definition: spin.h:62
bool in_use
Definition: slot.h:132
#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:81
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1203
int i
ReplicationSlot replication_slots[1]
Definition: slot.h:189
slock_t mutex
Definition: slot.h:129

◆ ReplicationSlotsComputeRequiredXmin()

void ReplicationSlotsComputeRequiredXmin ( bool  already_locked)

Definition at line 791 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(), LogicalConfirmReceivedLocation(), pg_replication_slot_advance(), PhysicalReplicationSlotNewXmin(), ReplicationSlotDropPtr(), ReplicationSlotRelease(), and StartupReplicationSlots().

792 {
793  int i;
795  TransactionId agg_catalog_xmin = InvalidTransactionId;
796 
797  Assert(ReplicationSlotCtl != NULL);
798 
799  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
800 
801  for (i = 0; i < max_replication_slots; i++)
802  {
804  TransactionId effective_xmin;
805  TransactionId effective_catalog_xmin;
806 
807  if (!s->in_use)
808  continue;
809 
810  SpinLockAcquire(&s->mutex);
811  effective_xmin = s->effective_xmin;
812  effective_catalog_xmin = s->effective_catalog_xmin;
813  SpinLockRelease(&s->mutex);
814 
815  /* check the data xmin */
816  if (TransactionIdIsValid(effective_xmin) &&
817  (!TransactionIdIsValid(agg_xmin) ||
818  TransactionIdPrecedes(effective_xmin, agg_xmin)))
819  agg_xmin = effective_xmin;
820 
821  /* check the catalog xmin */
822  if (TransactionIdIsValid(effective_catalog_xmin) &&
823  (!TransactionIdIsValid(agg_catalog_xmin) ||
824  TransactionIdPrecedes(effective_catalog_xmin, agg_catalog_xmin)))
825  agg_catalog_xmin = effective_catalog_xmin;
826  }
827 
828  LWLockRelease(ReplicationSlotControlLock);
829 
830  ProcArraySetReplicationSlotXmin(agg_xmin, agg_catalog_xmin, already_locked);
831 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
uint32 TransactionId
Definition: c.h:587
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1816
#define SpinLockAcquire(lock)
Definition: spin.h:62
TransactionId effective_xmin
Definition: slot.h:152
#define InvalidTransactionId
Definition: transam.h:31
bool TransactionIdPrecedes(TransactionId id1, TransactionId id2)
Definition: transam.c:300
bool in_use
Definition: slot.h:132
#define SpinLockRelease(lock)
Definition: spin.h:64
TransactionId effective_catalog_xmin
Definition: slot.h:153
int max_replication_slots
Definition: slot.c:99
#define Assert(condition)
Definition: c.h:804
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1203
int i
void ProcArraySetReplicationSlotXmin(TransactionId xmin, TransactionId catalog_xmin, bool already_locked)
Definition: procarray.c:3802
ReplicationSlot replication_slots[1]
Definition: slot.h:189
slock_t mutex
Definition: slot.h:129
#define TransactionIdIsValid(xid)
Definition: transam.h:41

◆ ReplicationSlotsCountDBSlots()

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

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

937 {
938  int i;
939 
940  *nslots = *nactive = 0;
941 
942  if (max_replication_slots <= 0)
943  return false;
944 
945  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
946  for (i = 0; i < max_replication_slots; i++)
947  {
948  ReplicationSlot *s;
949 
951 
952  /* cannot change while ReplicationSlotCtlLock is held */
953  if (!s->in_use)
954  continue;
955 
956  /* only logical slots are database specific, skip */
957  if (!SlotIsLogical(s))
958  continue;
959 
960  /* not our database, skip */
961  if (s->data.database != dboid)
962  continue;
963 
964  /* count slots with spinlock held */
965  SpinLockAcquire(&s->mutex);
966  (*nslots)++;
967  if (s->active_pid != 0)
968  (*nactive)++;
969  SpinLockRelease(&s->mutex);
970  }
971  LWLockRelease(ReplicationSlotControlLock);
972 
973  if (*nslots > 0)
974  return true;
975  return false;
976 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
ReplicationSlotPersistentData data
Definition: slot.h:156
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1816
#define SpinLockAcquire(lock)
Definition: spin.h:62
#define SlotIsLogical(slot)
Definition: slot.h:178
bool in_use
Definition: slot.h:132
#define SpinLockRelease(lock)
Definition: spin.h:64
int max_replication_slots
Definition: slot.c:99
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1203
pid_t active_pid
Definition: slot.h:135
int i
ReplicationSlot replication_slots[1]
Definition: slot.h:189
slock_t mutex
Definition: slot.h:129

◆ ReplicationSlotsDropDBSlots()

void ReplicationSlotsDropDBSlots ( Oid  dboid)

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

993 {
994  int i;
995 
996  if (max_replication_slots <= 0)
997  return;
998 
999 restart:
1000  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
1001  for (i = 0; i < max_replication_slots; i++)
1002  {
1003  ReplicationSlot *s;
1004  char *slotname;
1005  int active_pid;
1006 
1008 
1009  /* cannot change while ReplicationSlotCtlLock is held */
1010  if (!s->in_use)
1011  continue;
1012 
1013  /* only logical slots are database specific, skip */
1014  if (!SlotIsLogical(s))
1015  continue;
1016 
1017  /* not our database, skip */
1018  if (s->data.database != dboid)
1019  continue;
1020 
1021  /* acquire slot, so ReplicationSlotDropAcquired can be reused */
1022  SpinLockAcquire(&s->mutex);
1023  /* can't change while ReplicationSlotControlLock is held */
1024  slotname = NameStr(s->data.name);
1025  active_pid = s->active_pid;
1026  if (active_pid == 0)
1027  {
1028  MyReplicationSlot = s;
1029  s->active_pid = MyProcPid;
1030  }
1031  SpinLockRelease(&s->mutex);
1032 
1033  /*
1034  * Even though we hold an exclusive lock on the database object a
1035  * logical slot for that DB can still be active, e.g. if it's
1036  * concurrently being dropped by a backend connected to another DB.
1037  *
1038  * That's fairly unlikely in practice, so we'll just bail out.
1039  */
1040  if (active_pid)
1041  ereport(ERROR,
1042  (errcode(ERRCODE_OBJECT_IN_USE),
1043  errmsg("replication slot \"%s\" is active for PID %d",
1044  slotname, active_pid)));
1045 
1046  /*
1047  * To avoid duplicating ReplicationSlotDropAcquired() and to avoid
1048  * holding ReplicationSlotControlLock over filesystem operations,
1049  * release ReplicationSlotControlLock and use
1050  * ReplicationSlotDropAcquired.
1051  *
1052  * As that means the set of slots could change, restart scan from the
1053  * beginning each time we release the lock.
1054  */
1055  LWLockRelease(ReplicationSlotControlLock);
1057  goto restart;
1058  }
1059  LWLockRelease(ReplicationSlotControlLock);
1060 }
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:156
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1816
#define SpinLockAcquire(lock)
Definition: spin.h:62
static void ReplicationSlotDropAcquired(void)
Definition: slot.c:600
#define ERROR
Definition: elog.h:46
#define SlotIsLogical(slot)
Definition: slot.h:178
bool in_use
Definition: slot.h:132
#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:1203
int errmsg(const char *fmt,...)
Definition: elog.c:909
pid_t active_pid
Definition: slot.h:135
int i
#define NameStr(name)
Definition: c.h:681
ReplicationSlot replication_slots[1]
Definition: slot.h:189
slock_t mutex
Definition: slot.h:129

◆ ReplicationSlotsShmemInit()

void ReplicationSlotsShmemInit ( void  )

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

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 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
Size ReplicationSlotsShmemSize(void)
Definition: slot.c:116
#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:740
int max_replication_slots
Definition: slot.c:99
ConditionVariable active_cv
Definition: slot.h:162
int i
ReplicationSlot replication_slots[1]
Definition: slot.h:189
slock_t mutex
Definition: slot.h:129
LWLock io_in_progress_lock
Definition: slot.h:159

◆ ReplicationSlotsShmemSize()

Size ReplicationSlotsShmemSize ( void  )

Definition at line 116 of file slot.c.

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

Referenced by CreateSharedMemoryAndSemaphores(), and ReplicationSlotsShmemInit().

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

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

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

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

◆ RestoreSlotFromDisk()

static void RestoreSlotFromDisk ( const char *  name)
static

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

1596 {
1598  int i;
1599  char slotdir[MAXPGPATH + 12];
1600  char path[MAXPGPATH + 22];
1601  int fd;
1602  bool restored = false;
1603  int readBytes;
1604  pg_crc32c checksum;
1605 
1606  /* no need to lock here, no concurrent access allowed yet */
1607 
1608  /* delete temp file if it exists */
1609  sprintf(slotdir, "pg_replslot/%s", name);
1610  sprintf(path, "%s/state.tmp", slotdir);
1611  if (unlink(path) < 0 && errno != ENOENT)
1612  ereport(PANIC,
1614  errmsg("could not remove file \"%s\": %m", path)));
1615 
1616  sprintf(path, "%s/state", slotdir);
1617 
1618  elog(DEBUG1, "restoring replication slot from \"%s\"", path);
1619 
1620  /* on some operating systems fsyncing a file requires O_RDWR */
1621  fd = OpenTransientFile(path, O_RDWR | PG_BINARY);
1622 
1623  /*
1624  * We do not need to handle this as we are rename()ing the directory into
1625  * place only after we fsync()ed the state file.
1626  */
1627  if (fd < 0)
1628  ereport(PANIC,
1630  errmsg("could not open file \"%s\": %m", path)));
1631 
1632  /*
1633  * Sync state file before we're reading from it. We might have crashed
1634  * while it wasn't synced yet and we shouldn't continue on that basis.
1635  */
1637  if (pg_fsync(fd) != 0)
1638  ereport(PANIC,
1640  errmsg("could not fsync file \"%s\": %m",
1641  path)));
1643 
1644  /* Also sync the parent directory */
1646  fsync_fname(slotdir, true);
1647  END_CRIT_SECTION();
1648 
1649  /* read part of statefile that's guaranteed to be version independent */
1651  readBytes = read(fd, &cp, ReplicationSlotOnDiskConstantSize);
1653  if (readBytes != ReplicationSlotOnDiskConstantSize)
1654  {
1655  if (readBytes < 0)
1656  ereport(PANIC,
1658  errmsg("could not read file \"%s\": %m", path)));
1659  else
1660  ereport(PANIC,
1662  errmsg("could not read file \"%s\": read %d of %zu",
1663  path, readBytes,
1665  }
1666 
1667  /* verify magic */
1668  if (cp.magic != SLOT_MAGIC)
1669  ereport(PANIC,
1671  errmsg("replication slot file \"%s\" has wrong magic number: %u instead of %u",
1672  path, cp.magic, SLOT_MAGIC)));
1673 
1674  /* verify version */
1675  if (cp.version != SLOT_VERSION)
1676  ereport(PANIC,
1678  errmsg("replication slot file \"%s\" has unsupported version %u",
1679  path, cp.version)));
1680 
1681  /* boundary check on length */
1683  ereport(PANIC,
1685  errmsg("replication slot file \"%s\" has corrupted length %u",
1686  path, cp.length)));
1687 
1688  /* Now that we know the size, read the entire file */
1690  readBytes = read(fd,
1691  (char *) &cp + ReplicationSlotOnDiskConstantSize,
1692  cp.length);
1694  if (readBytes != cp.length)
1695  {
1696  if (readBytes < 0)
1697  ereport(PANIC,
1699  errmsg("could not read file \"%s\": %m", path)));
1700  else
1701  ereport(PANIC,
1703  errmsg("could not read file \"%s\": read %d of %zu",
1704  path, readBytes, (Size) cp.length)));
1705  }
1706 
1707  if (CloseTransientFile(fd) != 0)
1708  ereport(PANIC,
1710  errmsg("could not close file \"%s\": %m", path)));
1711 
1712  /* now verify the CRC */
1713  INIT_CRC32C(checksum);
1714  COMP_CRC32C(checksum,
1715  (char *) &cp + SnapBuildOnDiskNotChecksummedSize,
1717  FIN_CRC32C(checksum);
1718 
1719  if (!EQ_CRC32C(checksum, cp.checksum))
1720  ereport(PANIC,
1721  (errmsg("checksum mismatch for replication slot file \"%s\": is %u, should be %u",
1722  path, checksum, cp.checksum)));
1723 
1724  /*
1725  * If we crashed with an ephemeral slot active, don't restore but delete
1726  * it.
1727  */
1729  {
1730  if (!rmtree(slotdir, true))
1731  {
1732  ereport(WARNING,
1733  (errmsg("could not remove directory \"%s\"",
1734  slotdir)));
1735  }
1736  fsync_fname("pg_replslot", true);
1737  return;
1738  }
1739 
1740  /*
1741  * Verify that requirements for the specific slot type are met. That's
1742  * important because if these aren't met we're not guaranteed to retain
1743  * all the necessary resources for the slot.
1744  *
1745  * NB: We have to do so *after* the above checks for ephemeral slots,
1746  * because otherwise a slot that shouldn't exist anymore could prevent
1747  * restarts.
1748  *
1749  * NB: Changing the requirements here also requires adapting
1750  * CheckSlotRequirements() and CheckLogicalDecodingRequirements().
1751  */
1753  ereport(FATAL,
1754  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1755  errmsg("logical replication slot \"%s\" exists, but wal_level < logical",
1756  NameStr(cp.slotdata.name)),
1757  errhint("Change wal_level to be logical or higher.")));
1758  else if (wal_level < WAL_LEVEL_REPLICA)
1759  ereport(FATAL,
1760  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1761  errmsg("physical replication slot \"%s\" exists, but wal_level < replica",
1762  NameStr(cp.slotdata.name)),
1763  errhint("Change wal_level to be replica or higher.")));
1764 
1765  /* nothing can be active yet, don't lock anything */
1766  for (i = 0; i < max_replication_slots; i++)
1767  {
1768  ReplicationSlot *slot;
1769 
1771 
1772  if (slot->in_use)
1773  continue;
1774 
1775  /* restore the entire set of persistent data */
1776  memcpy(&slot->data, &cp.slotdata,
1778 
1779  /* initialize in memory state */
1780  slot->effective_xmin = cp.slotdata.xmin;
1782 
1787 
1788  slot->in_use = true;
1789  slot->active_pid = 0;
1790 
1791  restored = true;
1792  break;
1793  }
1794 
1795  if (!restored)
1796  ereport(FATAL,
1797  (errmsg("too many replication slots active before shutdown"),
1798  errhint("Increase max_replication_slots and try again.")));
1799 }
#define INIT_CRC32C(crc)
Definition: pg_crc32c.h:41
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
TransactionId candidate_catalog_xmin
Definition: slot.h:171
#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:277
#define SLOT_MAGIC
Definition: slot.c:89
uint32 pg_crc32c
Definition: pg_crc32c.h:38
int wal_level
Definition: xlog.c:109
#define END_CRIT_SECTION()
Definition: miscadmin.h:137
void fsync_fname(const char *fname, bool isdir)
Definition: fd.c:666
ReplicationSlotPersistency persistency
Definition: slot.h:62
#define START_CRIT_SECTION()
Definition: miscadmin.h:135
int errcode(int sqlerrcode)
Definition: elog.c:698
#define SnapBuildOnDiskNotChecksummedSize
Definition: slot.c:80
ReplicationSlotPersistentData data
Definition: slot.h:156
#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:218
ReplicationSlotPersistentData slotdata
Definition: slot.c:73
int OpenTransientFile(const char *fileName, int fileFlags)
Definition: fd.c:2423
#define FATAL
Definition: elog.h:49
#define MAXPGPATH
TransactionId effective_xmin
Definition: slot.h:152
XLogRecPtr candidate_restart_valid
Definition: slot.h:173
int errcode_for_file_access(void)
Definition: elog.c:721
TransactionId catalog_xmin
Definition: slot.h:78
#define InvalidTransactionId
Definition: transam.h:31
TransactionId xmin
Definition: slot.h:70
#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:261
#define ERRCODE_DATA_CORRUPTED
Definition: pg_basebackup.c:45
int CloseTransientFile(int fd)
Definition: fd.c:2600
#define WARNING
Definition: elog.h:40
bool rmtree(const char *path, bool rmtopdir)
Definition: rmtree.c:42
bool in_use
Definition: slot.h:132
TransactionId effective_catalog_xmin
Definition: slot.h:153
#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:515
XLogRecPtr candidate_xmin_lsn
Definition: slot.h:172
int errmsg(const char *fmt,...)
Definition: elog.c:909
pid_t active_pid
Definition: slot.h:135
#define elog(elevel,...)
Definition: elog.h:232
int i
#define NameStr(name)
Definition: c.h:681
int pg_fsync(int fd)
Definition: fd.c:352
ReplicationSlot replication_slots[1]
Definition: slot.h:189
#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:174

◆ SaveSlotToPath()

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

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

1442 {
1443  char tmppath[MAXPGPATH];
1444  char path[MAXPGPATH];
1445  int fd;
1447  bool was_dirty;
1448 
1449  /* first check whether there's something to write out */
1450  SpinLockAcquire(&slot->mutex);
1451  was_dirty = slot->dirty;
1452  slot->just_dirtied = false;
1453  SpinLockRelease(&slot->mutex);
1454 
1455  /* and don't do anything if there's nothing to write */
1456  if (!was_dirty)
1457  return;
1458 
1460 
1461  /* silence valgrind :( */
1462  memset(&cp, 0, sizeof(ReplicationSlotOnDisk));
1463 
1464  sprintf(tmppath, "%s/state.tmp", dir);
1465  sprintf(path, "%s/state", dir);
1466 
1467  fd = OpenTransientFile(tmppath, O_CREAT | O_EXCL | O_WRONLY | PG_BINARY);
1468  if (fd < 0)
1469  {
1470  /*
1471  * If not an ERROR, then release the lock before returning. In case
1472  * of an ERROR, the error recovery path automatically releases the
1473  * lock, but no harm in explicitly releasing even in that case. Note
1474  * that LWLockRelease() could affect errno.
1475  */
1476  int save_errno = errno;
1477 
1479  errno = save_errno;
1480  ereport(elevel,
1482  errmsg("could not create file \"%s\": %m",
1483  tmppath)));
1484  return;
1485  }
1486 
1487  cp.magic = SLOT_MAGIC;
1488  INIT_CRC32C(cp.checksum);
1489  cp.version = SLOT_VERSION;
1491 
1492  SpinLockAcquire(&slot->mutex);
1493 
1494  memcpy(&cp.slotdata, &slot->data, sizeof(ReplicationSlotPersistentData));
1495 
1496  SpinLockRelease(&slot->mutex);
1497 
1498  COMP_CRC32C(cp.checksum,
1499  (char *) (&cp) + SnapBuildOnDiskNotChecksummedSize,
1501  FIN_CRC32C(cp.checksum);
1502 
1503  errno = 0;
1505  if ((write(fd, &cp, sizeof(cp))) != sizeof(cp))
1506  {
1507  int save_errno = errno;
1508 
1510  CloseTransientFile(fd);
1512 
1513  /* if write didn't set errno, assume problem is no disk space */
1514  errno = save_errno ? save_errno : ENOSPC;
1515  ereport(elevel,
1517  errmsg("could not write to file \"%s\": %m",
1518  tmppath)));
1519  return;
1520  }
1522 
1523  /* fsync the temporary file */
1525  if (pg_fsync(fd) != 0)
1526  {
1527  int save_errno = errno;
1528 
1530  CloseTransientFile(fd);
1532  errno = save_errno;
1533  ereport(elevel,
1535  errmsg("could not fsync file \"%s\": %m",
1536  tmppath)));
1537  return;
1538  }
1540 
1541  if (CloseTransientFile(fd) != 0)
1542  {
1543  int save_errno = errno;
1544 
1546  errno = save_errno;
1547  ereport(elevel,
1549  errmsg("could not close file \"%s\": %m",
1550  tmppath)));
1551  return;
1552  }
1553 
1554  /* rename to permanent file, fsync file and directory */
1555  if (rename(tmppath, path) != 0)
1556  {
1557  int save_errno = errno;
1558 
1560  errno = save_errno;
1561  ereport(elevel,
1563  errmsg("could not rename file \"%s\" to \"%s\": %m",
1564  tmppath, path)));
1565  return;
1566  }
1567 
1568  /*
1569  * Check CreateSlotOnDisk() for the reasoning of using a critical section.
1570  */
1572 
1573  fsync_fname(path, false);
1574  fsync_fname(dir, true);
1575  fsync_fname("pg_replslot", true);
1576 
1577  END_CRIT_SECTION();
1578 
1579  /*
1580  * Successfully wrote, unset dirty bit, unless somebody dirtied again
1581  * already.
1582  */
1583  SpinLockAcquire(&slot->mutex);
1584  if (!slot->just_dirtied)
1585  slot->dirty = false;
1586  SpinLockRelease(&slot->mutex);
1587 
1589 }
#define INIT_CRC32C(crc)
Definition: pg_crc32c.h:41
static void pgstat_report_wait_end(void)
Definition: wait_event.h:277
#define write(a, b, c)
Definition: win32.h:14
#define SLOT_MAGIC
Definition: slot.c:89
#define END_CRIT_SECTION()
Definition: miscadmin.h:137
void fsync_fname(const char *fname, bool isdir)
Definition: fd.c:666
#define START_CRIT_SECTION()
Definition: miscadmin.h:135
#define SnapBuildOnDiskNotChecksummedSize
Definition: slot.c:80
ReplicationSlotPersistentData data
Definition: slot.h:156
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:1816
#define sprintf
Definition: port.h:218
#define SpinLockAcquire(lock)
Definition: spin.h:62
ReplicationSlotPersistentData slotdata
Definition: slot.c:73
int OpenTransientFile(const char *fileName, int fileFlags)
Definition: fd.c:2423
#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:261
int CloseTransientFile(int fd)
Definition: fd.c:2600
static int elevel
Definition: vacuumlazy.c:400
#define SpinLockRelease(lock)
Definition: spin.h:64
bool just_dirtied
Definition: slot.h:138
#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:1203
int errmsg(const char *fmt,...)
Definition: elog.c:909
int pg_fsync(int fd)
Definition: fd.c:352
slock_t mutex
Definition: slot.h:129
#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:139
LWLock io_in_progress_lock
Definition: slot.h:159

◆ SearchNamedReplicationSlot()

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

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

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

◆ StartupReplicationSlots()

void StartupReplicationSlots ( void  )

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

1321 {
1322  DIR *replication_dir;
1323  struct dirent *replication_de;
1324 
1325  elog(DEBUG1, "starting up replication slots");
1326 
1327  /* restore all slots by iterating over all on-disk entries */
1328  replication_dir = AllocateDir("pg_replslot");
1329  while ((replication_de = ReadDir(replication_dir, "pg_replslot")) != NULL)
1330  {
1331  struct stat statbuf;
1332  char path[MAXPGPATH + 12];
1333 
1334  if (strcmp(replication_de->d_name, ".") == 0 ||
1335  strcmp(replication_de->d_name, "..") == 0)
1336  continue;
1337 
1338  snprintf(path, sizeof(path), "pg_replslot/%s", replication_de->d_name);
1339 
1340  /* we're only creating directories here, skip if it's not our's */
1341  if (lstat(path, &statbuf) == 0 && !S_ISDIR(statbuf.st_mode))
1342  continue;
1343 
1344  /* we crashed while a slot was being setup or deleted, clean up */
1345  if (pg_str_endswith(replication_de->d_name, ".tmp"))
1346  {
1347  if (!rmtree(path, true))
1348  {
1349  ereport(WARNING,
1350  (errmsg("could not remove directory \"%s\"",
1351  path)));
1352  continue;
1353  }
1354  fsync_fname("pg_replslot", true);
1355  continue;
1356  }
1357 
1358  /* looks like a slot in a normal state, restore */
1359  RestoreSlotFromDisk(replication_de->d_name);
1360  }
1361  FreeDir(replication_dir);
1362 
1363  /* currently no slots exist, we're done. */
1364  if (max_replication_slots <= 0)
1365  return;
1366 
1367  /* Now that we have recovered all the data, compute replication xmin */
1370 }
static void RestoreSlotFromDisk(const char *name)
Definition: slot.c:1595
#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:666
Definition: dirent.h:9
void ReplicationSlotsComputeRequiredLSN(void)
Definition: slot.c:841
Definition: dirent.c:25
#define MAXPGPATH
DIR * AllocateDir(const char *dirname)
Definition: fd.c:2634
#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:2700
#define S_ISDIR(m)
Definition: win32_port.h:316
#define lstat(path, sb)
Definition: win32_port.h:276
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:216
void ReplicationSlotsComputeRequiredXmin(bool already_locked)
Definition: slot.c:791
int FreeDir(DIR *dir)
Definition: fd.c:2752

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