PostgreSQL Source Code  git master
 All Data Structures Namespaces Files Functions Variables Typedefs Enumerations Enumerator Macros
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)
 
void ReplicationSlotAcquire (const char *name)
 
void ReplicationSlotRelease (void)
 
void ReplicationSlotCleanup ()
 
void ReplicationSlotDrop (const char *name)
 
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 CheckPointReplicationSlots (void)
 
void StartupReplicationSlots (void)
 

Variables

ReplicationSlotCtlDataReplicationSlotCtl = NULL
 
ReplicationSlotMyReplicationSlot = NULL
 
int max_replication_slots = 0
 

Macro Definition Documentation

#define ReplicationSlotOnDiskConstantSize   offsetof(ReplicationSlotOnDisk, slotdata)

Definition at line 77 of file slot.c.

Referenced by RestoreSlotFromDisk().

#define ReplicationSlotOnDiskV2Size   sizeof(ReplicationSlotOnDisk) - ReplicationSlotOnDiskConstantSize

Definition at line 86 of file slot.c.

Referenced by RestoreSlotFromDisk(), and SaveSlotToPath().

#define SLOT_MAGIC   0x1051CA1 /* format identifier */

Definition at line 89 of file slot.c.

Referenced by RestoreSlotFromDisk(), and SaveSlotToPath().

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

Definition at line 90 of file slot.c.

Referenced by RestoreSlotFromDisk(), and SaveSlotToPath().

#define SnapBuildOnDiskChecksummedSize   sizeof(ReplicationSlotOnDisk) - SnapBuildOnDiskNotChecksummedSize

Definition at line 83 of file slot.c.

Referenced by RestoreSlotFromDisk(), and SaveSlotToPath().

#define SnapBuildOnDiskNotChecksummedSize   offsetof(ReplicationSlotOnDisk, version)

Definition at line 80 of file slot.c.

Referenced by RestoreSlotFromDisk(), and SaveSlotToPath().

Typedef Documentation

Function Documentation

void CheckPointReplicationSlots ( void  )

Definition at line 983 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, and SaveSlotToPath().

Referenced by CheckPointGuts().

984 {
985  int i;
986 
987  elog(DEBUG1, "performing replication slot checkpoint");
988 
989  /*
990  * Prevent any slot from being created/dropped while we're active. As we
991  * explicitly do *not* want to block iterating over replication_slots or
992  * acquiring a slot we cannot take the control lock - but that's OK,
993  * because holding ReplicationSlotAllocationLock is strictly stronger, and
994  * enough to guarantee that nobody can change the in_use bits on us.
995  */
996  LWLockAcquire(ReplicationSlotAllocationLock, LW_SHARED);
997 
998  for (i = 0; i < max_replication_slots; i++)
999  {
1001  char path[MAXPGPATH];
1002 
1003  if (!s->in_use)
1004  continue;
1005 
1006  /* save the slot to disk, locking is handled in SaveSlotToPath() */
1007  sprintf(path, "pg_replslot/%s", NameStr(s->data.name));
1008  SaveSlotToPath(s, path, LOG);
1009  }
1010  LWLockRelease(ReplicationSlotAllocationLock);
1011 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
#define DEBUG1
Definition: elog.h:25
ReplicationSlotPersistentData data
Definition: slot.h:115
#define LOG
Definition: elog.h:26
static void SaveSlotToPath(ReplicationSlot *slot, const char *path, int elevel)
Definition: slot.c:1138
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1715
#define MAXPGPATH
bool in_use
Definition: slot.h:91
int max_replication_slots
Definition: slot.c:99
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1111
int i
#define NameStr(name)
Definition: c.h:499
ReplicationSlot replication_slots[1]
Definition: slot.h:145
#define elog
Definition: elog.h:219
void CheckSlotRequirements ( void  )

Definition at line 893 of file slot.c.

References ereport, errcode(), errmsg(), ERROR, max_replication_slots, wal_level, and WAL_LEVEL_REPLICA.

Referenced by CheckLogicalDecodingRequirements(), pg_create_physical_replication_slot(), and pg_drop_replication_slot().

894 {
895  if (max_replication_slots == 0)
896  ereport(ERROR,
897  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
898  (errmsg("replication slots can only be used if max_replication_slots > 0"))));
899 
901  ereport(ERROR,
902  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
903  errmsg("replication slots can only be used if wal_level >= replica")));
904 }
int wal_level
Definition: xlog.c:104
int errcode(int sqlerrcode)
Definition: elog.c:575
#define ERROR
Definition: elog.h:43
#define ereport(elevel, rest)
Definition: elog.h:122
int max_replication_slots
Definition: slot.c:99
int errmsg(const char *fmt,...)
Definition: elog.c:797
static void CreateSlotOnDisk ( ReplicationSlot slot)
static

Definition at line 1078 of file slot.c.

References ReplicationSlot::data, ReplicationSlot::dirty, END_CRIT_SECTION, ereport, errcode_for_file_access(), errmsg(), ERROR, fsync_fname(), MAXPGPATH, mkdir, ReplicationSlotPersistentData::name, NameStr, rmtree(), SaveSlotToPath(), and START_CRIT_SECTION.

Referenced by ReplicationSlotCreate().

1079 {
1080  char tmppath[MAXPGPATH];
1081  char path[MAXPGPATH];
1082  struct stat st;
1083 
1084  /*
1085  * No need to take out the io_in_progress_lock, nobody else can see this
1086  * slot yet, so nobody else will write. We're reusing SaveSlotToPath which
1087  * takes out the lock, if we'd take the lock here, we'd deadlock.
1088  */
1089 
1090  sprintf(path, "pg_replslot/%s", NameStr(slot->data.name));
1091  sprintf(tmppath, "pg_replslot/%s.tmp", NameStr(slot->data.name));
1092 
1093  /*
1094  * It's just barely possible that some previous effort to create or drop a
1095  * slot with this name left a temp directory lying around. If that seems
1096  * to be the case, try to remove it. If the rmtree() fails, we'll error
1097  * out at the mkdir() below, so we don't bother checking success.
1098  */
1099  if (stat(tmppath, &st) == 0 && S_ISDIR(st.st_mode))
1100  rmtree(tmppath, true);
1101 
1102  /* Create and fsync the temporary slot directory. */
1103  if (mkdir(tmppath, S_IRWXU) < 0)
1104  ereport(ERROR,
1106  errmsg("could not create directory \"%s\": %m",
1107  tmppath)));
1108  fsync_fname(tmppath, true);
1109 
1110  /* Write the actual state file. */
1111  slot->dirty = true; /* signal that we really need to write */
1112  SaveSlotToPath(slot, tmppath, ERROR);
1113 
1114  /* Rename the directory into place. */
1115  if (rename(tmppath, path) != 0)
1116  ereport(ERROR,
1118  errmsg("could not rename file \"%s\" to \"%s\": %m",
1119  tmppath, path)));
1120 
1121  /*
1122  * If we'd now fail - really unlikely - we wouldn't know whether this slot
1123  * would persist after an OS crash or not - so, force a restart. The
1124  * restart would try to fsync this again till it works.
1125  */
1127 
1128  fsync_fname(path, true);
1129  fsync_fname("pg_replslot", true);
1130 
1131  END_CRIT_SECTION();
1132 }
#define mkdir(a, b)
Definition: win32.h:65
#define END_CRIT_SECTION()
Definition: miscadmin.h:132
void fsync_fname(const char *fname, bool isdir)
Definition: fd.c:567
#define START_CRIT_SECTION()
Definition: miscadmin.h:130
ReplicationSlotPersistentData data
Definition: slot.h:115
static void SaveSlotToPath(ReplicationSlot *slot, const char *path, int elevel)
Definition: slot.c:1138
#define ERROR
Definition: elog.h:43
#define MAXPGPATH
int errcode_for_file_access(void)
Definition: elog.c:598
#define ereport(elevel, rest)
Definition: elog.h:122
bool rmtree(const char *path, bool rmtopdir)
Definition: rmtree.c:36
int errmsg(const char *fmt,...)
Definition: elog.c:797
#define NameStr(name)
Definition: c.h:499
bool dirty
Definition: slot.h:98
void ReplicationSlotAcquire ( const char *  name)

Definition at line 326 of file slot.c.

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

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

327 {
328  ReplicationSlot *slot = NULL;
329  int i;
330  int active_pid = 0; /* Keep compiler quiet */
331 
333 
335 
336  /* Search for the named slot and mark it active if we find it. */
337  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
338  for (i = 0; i < max_replication_slots; i++)
339  {
341 
342  if (s->in_use && strcmp(name, NameStr(s->data.name)) == 0)
343  {
344  SpinLockAcquire(&s->mutex);
345  active_pid = s->active_pid;
346  if (active_pid == 0)
347  active_pid = s->active_pid = MyProcPid;
348  SpinLockRelease(&s->mutex);
349  slot = s;
350  break;
351  }
352  }
353  LWLockRelease(ReplicationSlotControlLock);
354 
355  /* If we did not find the slot or it was already active, error out. */
356  if (slot == NULL)
357  ereport(ERROR,
358  (errcode(ERRCODE_UNDEFINED_OBJECT),
359  errmsg("replication slot \"%s\" does not exist", name)));
360  if (active_pid != MyProcPid)
361  ereport(ERROR,
362  (errcode(ERRCODE_OBJECT_IN_USE),
363  errmsg("replication slot \"%s\" is active for PID %d",
364  name, active_pid)));
365 
366  /* We made this slot active, so it's ours now. */
367  MyReplicationSlot = slot;
368 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
int MyProcPid
Definition: globals.c:38
int errcode(int sqlerrcode)
Definition: elog.c:575
ReplicationSlotPersistentData data
Definition: slot.h:115
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1715
bool ReplicationSlotValidateName(const char *name, int elevel)
Definition: slot.c:173
#define SpinLockAcquire(lock)
Definition: spin.h:62
#define ERROR
Definition: elog.h:43
#define ereport(elevel, rest)
Definition: elog.h:122
bool in_use
Definition: slot.h:91
#define SpinLockRelease(lock)
Definition: spin.h:64
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
int max_replication_slots
Definition: slot.c:99
#define NULL
Definition: c.h:229
#define Assert(condition)
Definition: c.h:675
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1111
const char * name
Definition: encode.c:521
int errmsg(const char *fmt,...)
Definition: elog.c:797
pid_t active_pid
Definition: slot.h:94
int i
#define NameStr(name)
Definition: c.h:499
ReplicationSlot replication_slots[1]
Definition: slot.h:145
slock_t mutex
Definition: slot.h:88
void ReplicationSlotCleanup ( void  )

Definition at line 413 of file slot.c.

References ReplicationSlot::active_pid, Assert, ReplicationSlot::data, i, ReplicationSlot::in_use, max_replication_slots, MyProcPid, NULL, ReplicationSlotPersistentData::persistency, ReplicationSlotCtlData::replication_slots, ReplicationSlotDropPtr(), and RS_TEMPORARY.

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

414 {
415  int i;
416 
418 
419  /*
420  * No need for locking as we are only interested in slots active in
421  * current process and those are not touched by other processes.
422  */
423  for (i = 0; i < max_replication_slots; i++)
424  {
426 
427  if (s->active_pid == MyProcPid)
428  {
430 
432  }
433  }
434 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
int MyProcPid
Definition: globals.c:38
ReplicationSlotPersistency persistency
Definition: slot.h:49
static void ReplicationSlotDropPtr(ReplicationSlot *slot)
Definition: slot.c:470
ReplicationSlotPersistentData data
Definition: slot.h:115
bool in_use
Definition: slot.h:91
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
int max_replication_slots
Definition: slot.c:99
#define NULL
Definition: c.h:229
#define Assert(condition)
Definition: c.h:675
pid_t active_pid
Definition: slot.h:94
int i
ReplicationSlot replication_slots[1]
Definition: slot.h:145
void ReplicationSlotCreate ( const char *  name,
bool  db_specific,
ReplicationSlotPersistency  persistency 
)

Definition at line 220 of file slot.c.

References ReplicationSlot::active_pid, Assert, ReplicationSlot::candidate_catalog_xmin, ReplicationSlot::candidate_restart_lsn, ReplicationSlot::candidate_restart_valid, ReplicationSlot::candidate_xmin_lsn, 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, NAMEDATALEN, NameStr, NULL, ReplicationSlotPersistentData::persistency, ReplicationSlotCtlData::replication_slots, ReplicationSlotValidateName(), SpinLockAcquire, SpinLockRelease, and StrNCpy.

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

222 {
223  ReplicationSlot *slot = NULL;
224  int i;
225 
227 
229 
230  /*
231  * If some other backend ran this code concurrently with us, we'd likely
232  * both allocate the same slot, and that would be bad. We'd also be at
233  * risk of missing a name collision. Also, we don't want to try to create
234  * a new slot while somebody's busy cleaning up an old one, because we
235  * might both be monkeying with the same directory.
236  */
237  LWLockAcquire(ReplicationSlotAllocationLock, LW_EXCLUSIVE);
238 
239  /*
240  * Check for name collision, and identify an allocatable slot. We need to
241  * hold ReplicationSlotControlLock in shared mode for this, so that nobody
242  * else can change the in_use flags while we're looking at them.
243  */
244  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
245  for (i = 0; i < max_replication_slots; i++)
246  {
248 
249  if (s->in_use && strcmp(name, NameStr(s->data.name)) == 0)
250  ereport(ERROR,
252  errmsg("replication slot \"%s\" already exists", name)));
253  if (!s->in_use && slot == NULL)
254  slot = s;
255  }
256  LWLockRelease(ReplicationSlotControlLock);
257 
258  /* If all slots are in use, we're out of luck. */
259  if (slot == NULL)
260  ereport(ERROR,
261  (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
262  errmsg("all replication slots are in use"),
263  errhint("Free one or increase max_replication_slots.")));
264 
265  /*
266  * Since this slot is not in use, nobody should be looking at any part of
267  * it other than the in_use field unless they're trying to allocate it.
268  * And since we hold ReplicationSlotAllocationLock, nobody except us can
269  * be doing that. So it's safe to initialize the slot.
270  */
271  Assert(!slot->in_use);
272  Assert(slot->active_pid == 0);
273 
274  /* first initialize persistent data */
275  memset(&slot->data, 0, sizeof(ReplicationSlotPersistentData));
277  slot->data.database = db_specific ? MyDatabaseId : InvalidOid;
278  slot->data.persistency = persistency;
279 
280  /* and then data only present in shared memory */
281  slot->just_dirtied = false;
282  slot->dirty = false;
289 
290  /*
291  * Create the slot on disk. We haven't actually marked the slot allocated
292  * yet, so no special cleanup is required if this errors out.
293  */
294  CreateSlotOnDisk(slot);
295 
296  /*
297  * We need to briefly prevent any other backend from iterating over the
298  * slots while we flip the in_use flag. We also need to set the active
299  * flag while holding the ControlLock as otherwise a concurrent
300  * SlotAcquire() could acquire the slot as well.
301  */
302  LWLockAcquire(ReplicationSlotControlLock, LW_EXCLUSIVE);
303 
304  slot->in_use = true;
305 
306  /* We can now mark the slot active, and that makes it our slot. */
307  SpinLockAcquire(&slot->mutex);
308  Assert(slot->active_pid == 0);
309  slot->active_pid = MyProcPid;
310  SpinLockRelease(&slot->mutex);
311  MyReplicationSlot = slot;
312 
313  LWLockRelease(ReplicationSlotControlLock);
314 
315  /*
316  * Now that the slot has been marked as in_use and in_active, it's safe to
317  * let somebody else try to allocate a slot.
318  */
319  LWLockRelease(ReplicationSlotAllocationLock);
320 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
TransactionId candidate_catalog_xmin
Definition: slot.h:127
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
int MyProcPid
Definition: globals.c:38
int errhint(const char *fmt,...)
Definition: elog.c:987
ReplicationSlotPersistency persistency
Definition: slot.h:49
int errcode(int sqlerrcode)
Definition: elog.c:575
ReplicationSlotPersistentData data
Definition: slot.h:115
static void CreateSlotOnDisk(ReplicationSlot *slot)
Definition: slot.c:1078
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1715
#define NAMEDATALEN
bool ReplicationSlotValidateName(const char *name, int elevel)
Definition: slot.c:173
#define SpinLockAcquire(lock)
Definition: spin.h:62
#define ERROR
Definition: elog.h:43
TransactionId effective_xmin
Definition: slot.h:111
XLogRecPtr candidate_restart_valid
Definition: slot.h:129
#define InvalidTransactionId
Definition: transam.h:31
#define ereport(elevel, rest)
Definition: elog.h:122
bool in_use
Definition: slot.h:91
#define SpinLockRelease(lock)
Definition: spin.h:64
bool just_dirtied
Definition: slot.h:97
TransactionId effective_catalog_xmin
Definition: slot.h:112
Oid MyDatabaseId
Definition: globals.c:76
#define InvalidOid
Definition: postgres_ext.h:36
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
int max_replication_slots
Definition: slot.c:99
#define NULL
Definition: c.h:229
#define Assert(condition)
Definition: c.h:675
#define StrNCpy(dst, src, len)
Definition: c.h:830
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1111
const char * name
Definition: encode.c:521
XLogRecPtr candidate_xmin_lsn
Definition: slot.h:128
int errmsg(const char *fmt,...)
Definition: elog.c:797
pid_t active_pid
Definition: slot.h:94
int i
#define NameStr(name)
Definition: c.h:499
ReplicationSlot replication_slots[1]
Definition: slot.h:145
slock_t mutex
Definition: slot.h:88
#define ERRCODE_DUPLICATE_OBJECT
Definition: streamutil.c:31
bool dirty
Definition: slot.h:98
XLogRecPtr candidate_restart_lsn
Definition: slot.h:130
void ReplicationSlotDrop ( const char *  name)

Definition at line 440 of file slot.c.

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

Referenced by DropReplicationSlot(), and pg_drop_replication_slot().

441 {
443 
445 
447 }
static void ReplicationSlotDropAcquired(void)
Definition: slot.c:453
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
#define NULL
Definition: c.h:229
#define Assert(condition)
Definition: c.h:675
void ReplicationSlotAcquire(const char *name)
Definition: slot.c:326
const char * name
Definition: encode.c:521
static void ReplicationSlotDropAcquired ( void  )
static

Definition at line 453 of file slot.c.

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

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

454 {
456 
458 
459  /* slot isn't acquired anymore */
461 
463 }
static void ReplicationSlotDropPtr(ReplicationSlot *slot)
Definition: slot.c:470
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
#define NULL
Definition: c.h:229
#define Assert(condition)
Definition: c.h:675
static void ReplicationSlotDropPtr ( ReplicationSlot slot)
static

Definition at line 470 of file slot.c.

References ReplicationSlot::active_pid, 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, ReplicationSlotsComputeRequiredLSN(), ReplicationSlotsComputeRequiredXmin(), rmtree(), RS_PERSISTENT, SpinLockAcquire, SpinLockRelease, START_CRIT_SECTION, and WARNING.

Referenced by ReplicationSlotCleanup(), and ReplicationSlotDropAcquired().

471 {
472  char path[MAXPGPATH];
473  char tmppath[MAXPGPATH];
474 
475  /*
476  * If some other backend ran this code concurrently with us, we might try
477  * to delete a slot with a certain name while someone else was trying to
478  * create a slot with the same name.
479  */
480  LWLockAcquire(ReplicationSlotAllocationLock, LW_EXCLUSIVE);
481 
482  /* Generate pathnames. */
483  sprintf(path, "pg_replslot/%s", NameStr(slot->data.name));
484  sprintf(tmppath, "pg_replslot/%s.tmp", NameStr(slot->data.name));
485 
486  /*
487  * Rename the slot directory on disk, so that we'll no longer recognize
488  * this as a valid slot. Note that if this fails, we've got to mark the
489  * slot inactive before bailing out. If we're dropping an ephemeral or
490  * a temporary slot, we better never fail hard as the caller won't expect
491  * the slot to survive and this might get called during error handling.
492  */
493  if (rename(path, tmppath) == 0)
494  {
495  /*
496  * We need to fsync() the directory we just renamed and its parent to
497  * make sure that our changes are on disk in a crash-safe fashion. If
498  * fsync() fails, we can't be sure whether the changes are on disk or
499  * not. For now, we handle that by panicking;
500  * StartupReplicationSlots() will try to straighten it out after
501  * restart.
502  */
504  fsync_fname(tmppath, true);
505  fsync_fname("pg_replslot", true);
507  }
508  else
509  {
510  bool fail_softly = slot->data.persistency != RS_PERSISTENT;
511 
512  SpinLockAcquire(&slot->mutex);
513  slot->active_pid = 0;
514  SpinLockRelease(&slot->mutex);
515 
516  ereport(fail_softly ? WARNING : ERROR,
518  errmsg("could not rename file \"%s\" to \"%s\": %m",
519  path, tmppath)));
520  }
521 
522  /*
523  * The slot is definitely gone. Lock out concurrent scans of the array
524  * long enough to kill it. It's OK to clear the active flag here without
525  * grabbing the mutex because nobody else can be scanning the array here,
526  * and nobody can be attached to this slot and thus access it without
527  * scanning the array.
528  */
529  LWLockAcquire(ReplicationSlotControlLock, LW_EXCLUSIVE);
530  slot->active_pid = 0;
531  slot->in_use = false;
532  LWLockRelease(ReplicationSlotControlLock);
533 
534  /*
535  * Slot is dead and doesn't prevent resource removal anymore, recompute
536  * limits.
537  */
540 
541  /*
542  * If removing the directory fails, the worst thing that will happen is
543  * that the user won't be able to create a new slot with the same name
544  * until the next server restart. We warn about it, but that's all.
545  */
546  if (!rmtree(tmppath, true))
549  errmsg("could not remove directory \"%s\"", tmppath)));
550 
551  /*
552  * We release this at the very end, so that nobody starts trying to create
553  * a slot while we're still cleaning up the detritus of the old one.
554  */
555  LWLockRelease(ReplicationSlotAllocationLock);
556 }
#define END_CRIT_SECTION()
Definition: miscadmin.h:132
void fsync_fname(const char *fname, bool isdir)
Definition: fd.c:567
ReplicationSlotPersistency persistency
Definition: slot.h:49
#define START_CRIT_SECTION()
Definition: miscadmin.h:130
ReplicationSlotPersistentData data
Definition: slot.h:115
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1715
#define SpinLockAcquire(lock)
Definition: spin.h:62
void ReplicationSlotsComputeRequiredLSN(void)
Definition: slot.c:665
#define ERROR
Definition: elog.h:43
#define MAXPGPATH
int errcode_for_file_access(void)
Definition: elog.c:598
#define ereport(elevel, rest)
Definition: elog.h:122
#define WARNING
Definition: elog.h:40
bool rmtree(const char *path, bool rmtopdir)
Definition: rmtree.c:36
bool in_use
Definition: slot.h:91
#define SpinLockRelease(lock)
Definition: spin.h:64
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1111
int errmsg(const char *fmt,...)
Definition: elog.c:797
pid_t active_pid
Definition: slot.h:94
#define NameStr(name)
Definition: c.h:499
slock_t mutex
Definition: slot.h:88
void ReplicationSlotsComputeRequiredXmin(bool already_locked)
Definition: slot.c:617
void ReplicationSlotMarkDirty ( void  )

Definition at line 581 of file slot.c.

References Assert, ReplicationSlot::dirty, ReplicationSlot::just_dirtied, ReplicationSlot::mutex, MyReplicationSlot, NULL, SpinLockAcquire, and SpinLockRelease.

Referenced by CreateInitDecodingContext(), CreateReplicationSlot(), LogicalConfirmReceivedLocation(), pg_create_physical_replication_slot(), pg_logical_slot_get_changes_guts(), PhysicalConfirmReceivedLocation(), PhysicalReplicationSlotNewXmin(), and ReplicationSlotPersist().

582 {
584 
586 
587  SpinLockAcquire(&slot->mutex);
589  MyReplicationSlot->dirty = true;
590  SpinLockRelease(&slot->mutex);
591 }
#define SpinLockAcquire(lock)
Definition: spin.h:62
#define SpinLockRelease(lock)
Definition: spin.h:64
bool just_dirtied
Definition: slot.h:97
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
#define NULL
Definition: c.h:229
#define Assert(condition)
Definition: c.h:675
slock_t mutex
Definition: slot.h:88
bool dirty
Definition: slot.h:98
void ReplicationSlotPersist ( void  )

Definition at line 598 of file slot.c.

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

Referenced by CreateReplicationSlot(), and pg_create_logical_replication_slot().

599 {
601 
602  Assert(slot != NULL);
604 
605  SpinLockAcquire(&slot->mutex);
607  SpinLockRelease(&slot->mutex);
608 
611 }
ReplicationSlotPersistency persistency
Definition: slot.h:49
void ReplicationSlotSave(void)
Definition: slot.c:563
ReplicationSlotPersistentData data
Definition: slot.h:115
#define SpinLockAcquire(lock)
Definition: spin.h:62
#define SpinLockRelease(lock)
Definition: spin.h:64
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
#define NULL
Definition: c.h:229
#define Assert(condition)
Definition: c.h:675
slock_t mutex
Definition: slot.h:88
void ReplicationSlotMarkDirty(void)
Definition: slot.c:581
void ReplicationSlotRelease ( void  )

Definition at line 375 of file slot.c.

References ReplicationSlot::active_pid, Assert, ReplicationSlot::data, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), ReplicationSlot::mutex, MyPgXact, MyReplicationSlot, NULL, ReplicationSlotPersistentData::persistency, PROC_IN_LOGICAL_DECODING, ReplicationSlotDropAcquired(), RS_EPHEMERAL, RS_PERSISTENT, SpinLockAcquire, SpinLockRelease, and PGXACT::vacuumFlags.

Referenced by CreateReplicationSlot(), pg_create_logical_replication_slot(), pg_create_physical_replication_slot(), pg_logical_slot_get_changes_guts(), PostgresMain(), ProcKill(), StartLogicalReplication(), StartReplication(), and WalSndErrorCleanup().

376 {
378 
379  Assert(slot != NULL && slot->active_pid != 0);
380 
381  if (slot->data.persistency == RS_EPHEMERAL)
382  {
383  /*
384  * Delete the slot. There is no !PANIC case where this is allowed to
385  * fail, all that may happen is an incomplete cleanup of the on-disk
386  * data.
387  */
389  }
390  else if (slot->data.persistency == RS_PERSISTENT)
391  {
392  /*
393  * Mark persistent slot inactive. We're not freeing it, just
394  * disconnecting.
395  */
396  SpinLockAcquire(&slot->mutex);
397  slot->active_pid = 0;
398  SpinLockRelease(&slot->mutex);
399  }
400 
402 
403  /* might not have been set when we've been a plain slot */
404  LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
406  LWLockRelease(ProcArrayLock);
407 }
#define PROC_IN_LOGICAL_DECODING
Definition: proc.h:56
ReplicationSlotPersistency persistency
Definition: slot.h:49
ReplicationSlotPersistentData data
Definition: slot.h:115
PGXACT * MyPgXact
Definition: proc.c:68
uint8 vacuumFlags
Definition: proc.h:218
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1715
#define SpinLockAcquire(lock)
Definition: spin.h:62
static void ReplicationSlotDropAcquired(void)
Definition: slot.c:453
#define SpinLockRelease(lock)
Definition: spin.h:64
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
#define NULL
Definition: c.h:229
#define Assert(condition)
Definition: c.h:675
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1111
pid_t active_pid
Definition: slot.h:94
slock_t mutex
Definition: slot.h:88
void ReplicationSlotReserveWal ( void  )

Definition at line 913 of file slot.c.

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

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

914 {
916 
917  Assert(slot != NULL);
919 
920  /*
921  * The replication slot mechanism is used to prevent removal of required
922  * WAL. As there is no interlock between this routine and checkpoints, WAL
923  * segments could concurrently be removed when a now stale return value of
924  * ReplicationSlotsComputeRequiredLSN() is used. In the unlikely case that
925  * this happens we'll just retry.
926  */
927  while (true)
928  {
929  XLogSegNo segno;
930 
931  /*
932  * For logical slots log a standby snapshot and start logical decoding
933  * at exactly that position. That allows the slot to start up more
934  * quickly.
935  *
936  * That's not needed (or indeed helpful) for physical slots as they'll
937  * start replay at the last logged checkpoint anyway. Instead return
938  * the location of the last redo LSN. While that slightly increases
939  * the chance that we have to retry, it's where a base backup has to
940  * start replay at.
941  */
942  if (!RecoveryInProgress() && SlotIsLogical(slot))
943  {
944  XLogRecPtr flushptr;
945 
946  /* start at current insert position */
948 
949  /* make sure we have enough information to start */
950  flushptr = LogStandbySnapshot();
951 
952  /* and make sure it's fsynced to disk */
953  XLogFlush(flushptr);
954  }
955  else
956  {
957  slot->data.restart_lsn = GetRedoRecPtr();
958  }
959 
960  /* prevent WAL removal as fast as possible */
962 
963  /*
964  * If all required WAL is still there, great, otherwise retry. The
965  * slot should prevent further removal of WAL, unless there's a
966  * concurrent ReplicationSlotsComputeRequiredLSN() after we've written
967  * the new restart_lsn above, so normally we should never need to loop
968  * more than twice.
969  */
970  XLByteToSeg(slot->data.restart_lsn, segno);
971  if (XLogGetLastRemovedSegno() < segno)
972  break;
973  }
974 }
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
ReplicationSlotPersistentData data
Definition: slot.h:115
bool RecoveryInProgress(void)
Definition: xlog.c:7860
void XLogFlush(XLogRecPtr record)
Definition: xlog.c:2754
XLogSegNo XLogGetLastRemovedSegno(void)
Definition: xlog.c:3790
void ReplicationSlotsComputeRequiredLSN(void)
Definition: slot.c:665
XLogRecPtr LogStandbySnapshot(void)
Definition: standby.c:909
XLogRecPtr GetXLogInsertRecPtr(void)
Definition: xlog.c:11082
uint64 XLogSegNo
Definition: xlogdefs.h:34
#define SlotIsLogical(slot)
Definition: slot.h:134
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
#define XLByteToSeg(xlrp, logSegNo)
#define NULL
Definition: c.h:229
uint64 XLogRecPtr
Definition: xlogdefs.h:21
#define Assert(condition)
Definition: c.h:675
XLogRecPtr restart_lsn
Definition: slot.h:68
XLogRecPtr GetRedoRecPtr(void)
Definition: xlog.c:8150
void ReplicationSlotSave ( void  )

Definition at line 563 of file slot.c.

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

Referenced by CreateInitDecodingContext(), CreateReplicationSlot(), LogicalConfirmReceivedLocation(), pg_create_physical_replication_slot(), and ReplicationSlotPersist().

564 {
565  char path[MAXPGPATH];
566 
568 
569  sprintf(path, "pg_replslot/%s", NameStr(MyReplicationSlot->data.name));
571 }
ReplicationSlotPersistentData data
Definition: slot.h:115
static void SaveSlotToPath(ReplicationSlot *slot, const char *path, int elevel)
Definition: slot.c:1138
#define ERROR
Definition: elog.h:43
#define MAXPGPATH
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
#define NULL
Definition: c.h:229
#define Assert(condition)
Definition: c.h:675
#define NameStr(name)
Definition: c.h:499
XLogRecPtr ReplicationSlotsComputeLogicalRestartLSN ( void  )

Definition at line 708 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, result, SlotIsLogical, SpinLockAcquire, and SpinLockRelease.

Referenced by CheckPointLogicalRewriteHeap(), and CheckPointSnapBuild().

709 {
711  int i;
712 
713  if (max_replication_slots <= 0)
714  return InvalidXLogRecPtr;
715 
716  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
717 
718  for (i = 0; i < max_replication_slots; i++)
719  {
720  ReplicationSlot *s;
721  XLogRecPtr restart_lsn;
722 
724 
725  /* cannot change while ReplicationSlotCtlLock is held */
726  if (!s->in_use)
727  continue;
728 
729  /* we're only interested in logical slots */
730  if (!SlotIsLogical(s))
731  continue;
732 
733  /* read once, it's ok if it increases while we're checking */
734  SpinLockAcquire(&s->mutex);
735  restart_lsn = s->data.restart_lsn;
736  SpinLockRelease(&s->mutex);
737 
738  if (result == InvalidXLogRecPtr ||
739  restart_lsn < result)
740  result = restart_lsn;
741  }
742 
743  LWLockRelease(ReplicationSlotControlLock);
744 
745  return result;
746 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
return result
Definition: formatting.c:1618
ReplicationSlotPersistentData data
Definition: slot.h:115
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1715
#define SpinLockAcquire(lock)
Definition: spin.h:62
#define SlotIsLogical(slot)
Definition: slot.h:134
bool in_use
Definition: slot.h:91
#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:68
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1111
int i
ReplicationSlot replication_slots[1]
Definition: slot.h:145
slock_t mutex
Definition: slot.h:88
void ReplicationSlotsComputeRequiredLSN ( void  )

Definition at line 665 of file slot.c.

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

Referenced by LogicalConfirmReceivedLocation(), PhysicalConfirmReceivedLocation(), ReplicationSlotDropPtr(), ReplicationSlotReserveWal(), ReplicationSlotsDropDBSlots(), and StartupReplicationSlots().

666 {
667  int i;
668  XLogRecPtr min_required = InvalidXLogRecPtr;
669 
671 
672  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
673  for (i = 0; i < max_replication_slots; i++)
674  {
676  XLogRecPtr restart_lsn;
677 
678  if (!s->in_use)
679  continue;
680 
681  SpinLockAcquire(&s->mutex);
682  restart_lsn = s->data.restart_lsn;
683  SpinLockRelease(&s->mutex);
684 
685  if (restart_lsn != InvalidXLogRecPtr &&
686  (min_required == InvalidXLogRecPtr ||
687  restart_lsn < min_required))
688  min_required = restart_lsn;
689  }
690  LWLockRelease(ReplicationSlotControlLock);
691 
692  XLogSetReplicationSlotMinimumLSN(min_required);
693 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
ReplicationSlotPersistentData data
Definition: slot.h:115
void XLogSetReplicationSlotMinimumLSN(XLogRecPtr lsn)
Definition: xlog.c:2645
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1715
#define SpinLockAcquire(lock)
Definition: spin.h:62
bool in_use
Definition: slot.h:91
#define SpinLockRelease(lock)
Definition: spin.h:64
int max_replication_slots
Definition: slot.c:99
#define NULL
Definition: c.h:229
uint64 XLogRecPtr
Definition: xlogdefs.h:21
#define Assert(condition)
Definition: c.h:675
XLogRecPtr restart_lsn
Definition: slot.h:68
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1111
int i
ReplicationSlot replication_slots[1]
Definition: slot.h:145
slock_t mutex
Definition: slot.h:88
void ReplicationSlotsComputeRequiredXmin ( bool  already_locked)

Definition at line 617 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, NULL, ProcArraySetReplicationSlotXmin(), ReplicationSlotCtlData::replication_slots, SpinLockAcquire, SpinLockRelease, TransactionIdIsValid, and TransactionIdPrecedes().

Referenced by CreateInitDecodingContext(), LogicalConfirmReceivedLocation(), PhysicalReplicationSlotNewXmin(), ReplicationSlotDropPtr(), ReplicationSlotsDropDBSlots(), and StartupReplicationSlots().

618 {
619  int i;
621  TransactionId agg_catalog_xmin = InvalidTransactionId;
622 
624 
625  if (!already_locked)
626  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
627 
628  for (i = 0; i < max_replication_slots; i++)
629  {
631  TransactionId effective_xmin;
632  TransactionId effective_catalog_xmin;
633 
634  if (!s->in_use)
635  continue;
636 
637  SpinLockAcquire(&s->mutex);
638  effective_xmin = s->effective_xmin;
639  effective_catalog_xmin = s->effective_catalog_xmin;
640  SpinLockRelease(&s->mutex);
641 
642  /* check the data xmin */
643  if (TransactionIdIsValid(effective_xmin) &&
644  (!TransactionIdIsValid(agg_xmin) ||
645  TransactionIdPrecedes(effective_xmin, agg_xmin)))
646  agg_xmin = effective_xmin;
647 
648  /* check the catalog xmin */
649  if (TransactionIdIsValid(effective_catalog_xmin) &&
650  (!TransactionIdIsValid(agg_catalog_xmin) ||
651  TransactionIdPrecedes(effective_catalog_xmin, agg_catalog_xmin)))
652  agg_catalog_xmin = effective_catalog_xmin;
653  }
654 
655  if (!already_locked)
656  LWLockRelease(ReplicationSlotControlLock);
657 
658  ProcArraySetReplicationSlotXmin(agg_xmin, agg_catalog_xmin, already_locked);
659 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
uint32 TransactionId
Definition: c.h:397
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1715
#define SpinLockAcquire(lock)
Definition: spin.h:62
TransactionId effective_xmin
Definition: slot.h:111
#define InvalidTransactionId
Definition: transam.h:31
bool TransactionIdPrecedes(TransactionId id1, TransactionId id2)
Definition: transam.c:300
bool in_use
Definition: slot.h:91
#define SpinLockRelease(lock)
Definition: spin.h:64
TransactionId effective_catalog_xmin
Definition: slot.h:112
int max_replication_slots
Definition: slot.c:99
#define NULL
Definition: c.h:229
#define Assert(condition)
Definition: c.h:675
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1111
int i
void ProcArraySetReplicationSlotXmin(TransactionId xmin, TransactionId catalog_xmin, bool already_locked)
Definition: procarray.c:2948
ReplicationSlot replication_slots[1]
Definition: slot.h:145
slock_t mutex
Definition: slot.h:88
#define TransactionIdIsValid(xid)
Definition: transam.h:41
bool ReplicationSlotsCountDBSlots ( Oid  dboid,
int *  nslots,
int *  nactive 
)

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

758 {
759  int i;
760 
761  *nslots = *nactive = 0;
762 
763  if (max_replication_slots <= 0)
764  return false;
765 
766  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
767  for (i = 0; i < max_replication_slots; i++)
768  {
769  ReplicationSlot *s;
770 
772 
773  /* cannot change while ReplicationSlotCtlLock is held */
774  if (!s->in_use)
775  continue;
776 
777  /* only logical slots are database specific, skip */
778  if (!SlotIsLogical(s))
779  continue;
780 
781  /* not our database, skip */
782  if (s->data.database != dboid)
783  continue;
784 
785  /* count slots with spinlock held */
786  SpinLockAcquire(&s->mutex);
787  (*nslots)++;
788  if (s->active_pid != 0)
789  (*nactive)++;
790  SpinLockRelease(&s->mutex);
791  }
792  LWLockRelease(ReplicationSlotControlLock);
793 
794  if (*nslots > 0)
795  return true;
796  return false;
797 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
ReplicationSlotPersistentData data
Definition: slot.h:115
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1715
#define SpinLockAcquire(lock)
Definition: spin.h:62
#define SlotIsLogical(slot)
Definition: slot.h:134
bool in_use
Definition: slot.h:91
#define SpinLockRelease(lock)
Definition: spin.h:64
int max_replication_slots
Definition: slot.c:99
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1111
pid_t active_pid
Definition: slot.h:94
int i
ReplicationSlot replication_slots[1]
Definition: slot.h:145
slock_t mutex
Definition: slot.h:88
void ReplicationSlotsDropDBSlots ( Oid  dboid)

Definition at line 813 of file slot.c.

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

Referenced by dbase_redo(), and dropdb().

814 {
815  int i;
816 
817  if (max_replication_slots <= 0)
818  return;
819 
820 restart:
821  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
822  for (i = 0; i < max_replication_slots; i++)
823  {
824  ReplicationSlot *s;
825  NameData slotname;
826  int active_pid;
827 
829 
830  /* cannot change while ReplicationSlotCtlLock is held */
831  if (!s->in_use)
832  continue;
833 
834  /* only logical slots are database specific, skip */
835  if (!SlotIsLogical(s))
836  continue;
837 
838  /* not our database, skip */
839  if (s->data.database != dboid)
840  continue;
841 
842  /* Claim the slot, as if ReplicationSlotAcquire()ing. */
843  SpinLockAcquire(&s->mutex);
844  strncpy(NameStr(slotname), NameStr(s->data.name), NAMEDATALEN);
845  NameStr(slotname)[NAMEDATALEN-1] = '\0';
846  active_pid = s->active_pid;
847  if (active_pid == 0)
848  {
849  MyReplicationSlot = s;
850  s->active_pid = MyProcPid;
851  }
852  SpinLockRelease(&s->mutex);
853 
854  /*
855  * We might fail here if the slot was active. Even though we hold an
856  * exclusive lock on the database object a logical slot for that DB can
857  * still be active if it's being dropped by a backend connected to
858  * another DB or is otherwise acquired.
859  *
860  * It's an unlikely race that'll only arise from concurrent user action,
861  * so we'll just bail out.
862  */
863  if (active_pid)
864  elog(ERROR, "replication slot %s is in use by pid %d",
865  NameStr(slotname), active_pid);
866 
867  /*
868  * To avoid largely duplicating ReplicationSlotDropAcquired() or
869  * complicating it with already_locked flags for ProcArrayLock,
870  * ReplicationSlotControlLock and ReplicationSlotAllocationLock, we
871  * just release our ReplicationSlotControlLock to drop the slot.
872  *
873  * For safety we'll restart our scan from the beginning each
874  * time we release the lock.
875  */
876  LWLockRelease(ReplicationSlotControlLock);
878  goto restart;
879  }
880  LWLockRelease(ReplicationSlotControlLock);
881 
882  /* recompute limits once after all slots are dropped */
885 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
int MyProcPid
Definition: globals.c:38
ReplicationSlotPersistentData data
Definition: slot.h:115
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1715
#define NAMEDATALEN
#define SpinLockAcquire(lock)
Definition: spin.h:62
static void ReplicationSlotDropAcquired(void)
Definition: slot.c:453
void ReplicationSlotsComputeRequiredLSN(void)
Definition: slot.c:665
#define ERROR
Definition: elog.h:43
Definition: c.h:493
#define SlotIsLogical(slot)
Definition: slot.h:134
bool in_use
Definition: slot.h:91
#define SpinLockRelease(lock)
Definition: spin.h:64
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
int max_replication_slots
Definition: slot.c:99
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1111
pid_t active_pid
Definition: slot.h:94
int i
#define NameStr(name)
Definition: c.h:499
ReplicationSlot replication_slots[1]
Definition: slot.h:145
#define elog
Definition: elog.h:219
slock_t mutex
Definition: slot.h:88
void ReplicationSlotsComputeRequiredXmin(bool already_locked)
Definition: slot.c:617
void ReplicationSlotsShmemInit ( void  )

Definition at line 132 of file slot.c.

References i, ReplicationSlot::io_in_progress_lock, LWLockInitialize(), LWLockRegisterTranche(), LWTRANCHE_REPLICATION_SLOT_IO_IN_PROGRESS, 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 
144  "replication_slot_io");
145 
146  if (!found)
147  {
148  int i;
149 
150  /* First time through, so initialize */
152 
153  for (i = 0; i < max_replication_slots; i++)
154  {
156 
157  /* everything else is zeroed by the memset above */
158  SpinLockInit(&slot->mutex);
160  }
161  }
162 }
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:857
void * ShmemInitStruct(const char *name, Size size, bool *foundPtr)
Definition: shmem.c:372
void LWLockInitialize(LWLock *lock, int tranche_id)
Definition: lwlock.c:667
int max_replication_slots
Definition: slot.c:99
void LWLockRegisterTranche(int tranche_id, char *tranche_name)
Definition: lwlock.c:592
int i
ReplicationSlot replication_slots[1]
Definition: slot.h:145
slock_t mutex
Definition: slot.h:88
LWLock io_in_progress_lock
Definition: slot.h:118
Size ReplicationSlotsShmemSize ( void  )

Definition at line 114 of file slot.c.

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

Referenced by CreateSharedMemoryAndSemaphores(), 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:492
Size add_size(Size s1, Size s2)
Definition: shmem.c:475
int max_replication_slots
Definition: slot.c:99
size_t Size
Definition: c.h:356
#define offsetof(type, field)
Definition: c.h:555
bool ReplicationSlotValidateName ( const char *  name,
int  elevel 
)

Definition at line 173 of file slot.c.

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

Referenced by readRecoveryCommandFile(), ReplicationSlotAcquire(), ReplicationSlotCreate(), and StartupReorderBuffer().

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

Definition at line 1262 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, DEBUG1, ReplicationSlot::effective_catalog_xmin, ReplicationSlot::effective_xmin, elog, END_CRIT_SECTION, EQ_CRC32C, ereport, errcode_for_file_access(), errhint(), errmsg(), fd(), FIN_CRC32C, fsync_fname(), i, ReplicationSlot::in_use, INIT_CRC32C, InvalidTransactionId, InvalidXLogRecPtr, ReplicationSlotOnDisk::length, ReplicationSlotOnDisk::magic, max_replication_slots, MAXPGPATH, 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, START_CRIT_SECTION, unlink(), ReplicationSlotOnDisk::version, WAIT_EVENT_REPLICATION_SLOT_READ, WAIT_EVENT_REPLICATION_SLOT_RESTORE_SYNC, WARNING, and ReplicationSlotPersistentData::xmin.

Referenced by StartupReplicationSlots().

1263 {
1265  int i;
1266  char path[MAXPGPATH];
1267  int fd;
1268  bool restored = false;
1269  int readBytes;
1270  pg_crc32c checksum;
1271 
1272  /* no need to lock here, no concurrent access allowed yet */
1273 
1274  /* delete temp file if it exists */
1275  sprintf(path, "pg_replslot/%s/state.tmp", name);
1276  if (unlink(path) < 0 && errno != ENOENT)
1277  ereport(PANIC,
1279  errmsg("could not remove file \"%s\": %m", path)));
1280 
1281  sprintf(path, "pg_replslot/%s/state", name);
1282 
1283  elog(DEBUG1, "restoring replication slot from \"%s\"", path);
1284 
1285  fd = OpenTransientFile(path, O_RDWR | PG_BINARY, 0);
1286 
1287  /*
1288  * We do not need to handle this as we are rename()ing the directory into
1289  * place only after we fsync()ed the state file.
1290  */
1291  if (fd < 0)
1292  ereport(PANIC,
1294  errmsg("could not open file \"%s\": %m", path)));
1295 
1296  /*
1297  * Sync state file before we're reading from it. We might have crashed
1298  * while it wasn't synced yet and we shouldn't continue on that basis.
1299  */
1301  if (pg_fsync(fd) != 0)
1302  {
1303  CloseTransientFile(fd);
1304  ereport(PANIC,
1306  errmsg("could not fsync file \"%s\": %m",
1307  path)));
1308  }
1310 
1311  /* Also sync the parent directory */
1313  fsync_fname(path, true);
1314  END_CRIT_SECTION();
1315 
1316  /* read part of statefile that's guaranteed to be version independent */
1318  readBytes = read(fd, &cp, ReplicationSlotOnDiskConstantSize);
1320  if (readBytes != ReplicationSlotOnDiskConstantSize)
1321  {
1322  int saved_errno = errno;
1323 
1324  CloseTransientFile(fd);
1325  errno = saved_errno;
1326  ereport(PANIC,
1328  errmsg("could not read file \"%s\", read %d of %u: %m",
1329  path, readBytes,
1331  }
1332 
1333  /* verify magic */
1334  if (cp.magic != SLOT_MAGIC)
1335  ereport(PANIC,
1337  errmsg("replication slot file \"%s\" has wrong magic number: %u instead of %u",
1338  path, cp.magic, SLOT_MAGIC)));
1339 
1340  /* verify version */
1341  if (cp.version != SLOT_VERSION)
1342  ereport(PANIC,
1344  errmsg("replication slot file \"%s\" has unsupported version %u",
1345  path, cp.version)));
1346 
1347  /* boundary check on length */
1349  ereport(PANIC,
1351  errmsg("replication slot file \"%s\" has corrupted length %u",
1352  path, cp.length)));
1353 
1354  /* Now that we know the size, read the entire file */
1356  readBytes = read(fd,
1357  (char *) &cp + ReplicationSlotOnDiskConstantSize,
1358  cp.length);
1360  if (readBytes != cp.length)
1361  {
1362  int saved_errno = errno;
1363 
1364  CloseTransientFile(fd);
1365  errno = saved_errno;
1366  ereport(PANIC,
1368  errmsg("could not read file \"%s\", read %d of %u: %m",
1369  path, readBytes, cp.length)));
1370  }
1371 
1372  CloseTransientFile(fd);
1373 
1374  /* now verify the CRC */
1375  INIT_CRC32C(checksum);
1376  COMP_CRC32C(checksum,
1377  (char *) &cp + SnapBuildOnDiskNotChecksummedSize,
1379  FIN_CRC32C(checksum);
1380 
1381  if (!EQ_CRC32C(checksum, cp.checksum))
1382  ereport(PANIC,
1383  (errmsg("checksum mismatch for replication slot file \"%s\": is %u, should be %u",
1384  path, checksum, cp.checksum)));
1385 
1386  /*
1387  * If we crashed with an ephemeral slot active, don't restore but delete
1388  * it.
1389  */
1391  {
1392  sprintf(path, "pg_replslot/%s", name);
1393 
1394  if (!rmtree(path, true))
1395  {
1396  ereport(WARNING,
1398  errmsg("could not remove directory \"%s\"", path)));
1399  }
1400  fsync_fname("pg_replslot", true);
1401  return;
1402  }
1403 
1404  /* nothing can be active yet, don't lock anything */
1405  for (i = 0; i < max_replication_slots; i++)
1406  {
1407  ReplicationSlot *slot;
1408 
1410 
1411  if (slot->in_use)
1412  continue;
1413 
1414  /* restore the entire set of persistent data */
1415  memcpy(&slot->data, &cp.slotdata,
1417 
1418  /* initialize in memory state */
1419  slot->effective_xmin = cp.slotdata.xmin;
1421 
1426 
1427  slot->in_use = true;
1428  slot->active_pid = 0;
1429 
1430  restored = true;
1431  break;
1432  }
1433 
1434  if (!restored)
1435  ereport(PANIC,
1436  (errmsg("too many replication slots active before shutdown"),
1437  errhint("Increase max_replication_slots and try again.")));
1438 }
#define INIT_CRC32C(crc)
Definition: pg_crc32c.h:41
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
TransactionId candidate_catalog_xmin
Definition: slot.h:127
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
#define DEBUG1
Definition: elog.h:25
int errhint(const char *fmt,...)
Definition: elog.c:987
#define SLOT_MAGIC
Definition: slot.c:89
uint32 pg_crc32c
Definition: pg_crc32c.h:38
#define END_CRIT_SECTION()
Definition: miscadmin.h:132
void fsync_fname(const char *fname, bool isdir)
Definition: fd.c:567
ReplicationSlotPersistency persistency
Definition: slot.h:49
#define START_CRIT_SECTION()
Definition: miscadmin.h:130
#define SnapBuildOnDiskNotChecksummedSize
Definition: slot.c:80
ReplicationSlotPersistentData data
Definition: slot.h:115
#define PANIC
Definition: elog.h:53
static int fd(const char *x, int i)
Definition: preproc-init.c:105
#define PG_BINARY
Definition: c.h:1038
ReplicationSlotPersistentData slotdata
Definition: slot.c:73
#define MAXPGPATH
TransactionId effective_xmin
Definition: slot.h:111
XLogRecPtr candidate_restart_valid
Definition: slot.h:129
int OpenTransientFile(FileName fileName, int fileFlags, int fileMode)
Definition: fd.c:2144
int errcode_for_file_access(void)
Definition: elog.c:598
TransactionId catalog_xmin
Definition: slot.h:65
#define InvalidTransactionId
Definition: transam.h:31
unsigned int uint32
Definition: c.h:268
static void pgstat_report_wait_end(void)
Definition: pgstat.h:1231
TransactionId xmin
Definition: slot.h:57
#define EQ_CRC32C(c1, c2)
Definition: pg_crc32c.h:42
int unlink(const char *filename)
#define ereport(elevel, rest)
Definition: elog.h:122
pg_crc32c checksum
Definition: slot.c:62
int CloseTransientFile(int fd)
Definition: fd.c:2305
#define WARNING
Definition: elog.h:40
bool rmtree(const char *path, bool rmtopdir)
Definition: rmtree.c:36
bool in_use
Definition: slot.h:91
TransactionId effective_catalog_xmin
Definition: slot.h:112
#define SLOT_VERSION
Definition: slot.c:90
int max_replication_slots
Definition: slot.c:99
#define ReplicationSlotOnDiskV2Size
Definition: slot.c:86
#define ReplicationSlotOnDiskConstantSize
Definition: slot.c:77
uint32 version
Definition: slot.c:65
static void pgstat_report_wait_start(uint32 wait_event_info)
Definition: pgstat.h:1207
#define SnapBuildOnDiskChecksummedSize
Definition: slot.c:83
const char * name
Definition: encode.c:521
XLogRecPtr candidate_xmin_lsn
Definition: slot.h:128
int errmsg(const char *fmt,...)
Definition: elog.c:797
pid_t active_pid
Definition: slot.h:94
int i
int pg_fsync(int fd)
Definition: fd.c:333
ReplicationSlot replication_slots[1]
Definition: slot.h:145
#define elog
Definition: elog.h:219
#define COMP_CRC32C(crc, data, len)
Definition: pg_crc32c.h:73
#define FIN_CRC32C(crc)
Definition: pg_crc32c.h:78
#define read(a, b, c)
Definition: win32.h:18
XLogRecPtr candidate_restart_lsn
Definition: slot.h:130
static void SaveSlotToPath ( ReplicationSlot slot,
const char *  path,
int  elevel 
)
static

Definition at line 1138 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, START_CRIT_SECTION, ReplicationSlotOnDisk::version, WAIT_EVENT_REPLICATION_SLOT_SYNC, WAIT_EVENT_REPLICATION_SLOT_WRITE, and write.

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

1139 {
1140  char tmppath[MAXPGPATH];
1141  char path[MAXPGPATH];
1142  int fd;
1144  bool was_dirty;
1145 
1146  /* first check whether there's something to write out */
1147  SpinLockAcquire(&slot->mutex);
1148  was_dirty = slot->dirty;
1149  slot->just_dirtied = false;
1150  SpinLockRelease(&slot->mutex);
1151 
1152  /* and don't do anything if there's nothing to write */
1153  if (!was_dirty)
1154  return;
1155 
1157 
1158  /* silence valgrind :( */
1159  memset(&cp, 0, sizeof(ReplicationSlotOnDisk));
1160 
1161  sprintf(tmppath, "%s/state.tmp", dir);
1162  sprintf(path, "%s/state", dir);
1163 
1164  fd = OpenTransientFile(tmppath,
1165  O_CREAT | O_EXCL | O_WRONLY | PG_BINARY,
1166  S_IRUSR | S_IWUSR);
1167  if (fd < 0)
1168  {
1169  ereport(elevel,
1171  errmsg("could not create file \"%s\": %m",
1172  tmppath)));
1173  return;
1174  }
1175 
1176  cp.magic = SLOT_MAGIC;
1177  INIT_CRC32C(cp.checksum);
1178  cp.version = SLOT_VERSION;
1180 
1181  SpinLockAcquire(&slot->mutex);
1182 
1183  memcpy(&cp.slotdata, &slot->data, sizeof(ReplicationSlotPersistentData));
1184 
1185  SpinLockRelease(&slot->mutex);
1186 
1187  COMP_CRC32C(cp.checksum,
1188  (char *) (&cp) + SnapBuildOnDiskNotChecksummedSize,
1190  FIN_CRC32C(cp.checksum);
1191 
1193  if ((write(fd, &cp, sizeof(cp))) != sizeof(cp))
1194  {
1195  int save_errno = errno;
1196 
1198  CloseTransientFile(fd);
1199  errno = save_errno;
1200  ereport(elevel,
1202  errmsg("could not write to file \"%s\": %m",
1203  tmppath)));
1204  return;
1205  }
1207 
1208  /* fsync the temporary file */
1210  if (pg_fsync(fd) != 0)
1211  {
1212  int save_errno = errno;
1213 
1215  CloseTransientFile(fd);
1216  errno = save_errno;
1217  ereport(elevel,
1219  errmsg("could not fsync file \"%s\": %m",
1220  tmppath)));
1221  return;
1222  }
1224 
1225  CloseTransientFile(fd);
1226 
1227  /* rename to permanent file, fsync file and directory */
1228  if (rename(tmppath, path) != 0)
1229  {
1230  ereport(elevel,
1232  errmsg("could not rename file \"%s\" to \"%s\": %m",
1233  tmppath, path)));
1234  return;
1235  }
1236 
1237  /* Check CreateSlot() for the reasoning of using a crit. section. */
1239 
1240  fsync_fname(path, false);
1241  fsync_fname(dir, true);
1242  fsync_fname("pg_replslot", true);
1243 
1244  END_CRIT_SECTION();
1245 
1246  /*
1247  * Successfully wrote, unset dirty bit, unless somebody dirtied again
1248  * already.
1249  */
1250  SpinLockAcquire(&slot->mutex);
1251  if (!slot->just_dirtied)
1252  slot->dirty = false;
1253  SpinLockRelease(&slot->mutex);
1254 
1256 }
#define INIT_CRC32C(crc)
Definition: pg_crc32c.h:41
#define write(a, b, c)
Definition: win32.h:19
#define SLOT_MAGIC
Definition: slot.c:89
#define END_CRIT_SECTION()
Definition: miscadmin.h:132
void fsync_fname(const char *fname, bool isdir)
Definition: fd.c:567
#define START_CRIT_SECTION()
Definition: miscadmin.h:130
#define SnapBuildOnDiskNotChecksummedSize
Definition: slot.c:80
ReplicationSlotPersistentData data
Definition: slot.h:115
static int fd(const char *x, int i)
Definition: preproc-init.c:105
#define PG_BINARY
Definition: c.h:1038
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1715
#define SpinLockAcquire(lock)
Definition: spin.h:62
ReplicationSlotPersistentData slotdata
Definition: slot.c:73
#define MAXPGPATH
int OpenTransientFile(FileName fileName, int fileFlags, int fileMode)
Definition: fd.c:2144
int errcode_for_file_access(void)
Definition: elog.c:598
static void pgstat_report_wait_end(void)
Definition: pgstat.h:1231
#define ereport(elevel, rest)
Definition: elog.h:122
pg_crc32c checksum
Definition: slot.c:62
int CloseTransientFile(int fd)
Definition: fd.c:2305
static int elevel
Definition: vacuumlazy.c:137
#define SpinLockRelease(lock)
Definition: spin.h:64
bool just_dirtied
Definition: slot.h:97
#define SLOT_VERSION
Definition: slot.c:90
#define ReplicationSlotOnDiskV2Size
Definition: slot.c:86
uint32 version
Definition: slot.c:65
static void pgstat_report_wait_start(uint32 wait_event_info)
Definition: pgstat.h:1207
#define SnapBuildOnDiskChecksummedSize
Definition: slot.c:83
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1111
int errmsg(const char *fmt,...)
Definition: elog.c:797
int pg_fsync(int fd)
Definition: fd.c:333
slock_t mutex
Definition: slot.h:88
#define COMP_CRC32C(crc, data, len)
Definition: pg_crc32c.h:73
#define FIN_CRC32C(crc)
Definition: pg_crc32c.h:78
bool dirty
Definition: slot.h:98
LWLock io_in_progress_lock
Definition: slot.h:118
void StartupReplicationSlots ( void  )

Definition at line 1018 of file slot.c.

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

Referenced by StartupXLOG().

1019 {
1020  DIR *replication_dir;
1021  struct dirent *replication_de;
1022 
1023  elog(DEBUG1, "starting up replication slots");
1024 
1025  /* restore all slots by iterating over all on-disk entries */
1026  replication_dir = AllocateDir("pg_replslot");
1027  while ((replication_de = ReadDir(replication_dir, "pg_replslot")) != NULL)
1028  {
1029  struct stat statbuf;
1030  char path[MAXPGPATH];
1031 
1032  if (strcmp(replication_de->d_name, ".") == 0 ||
1033  strcmp(replication_de->d_name, "..") == 0)
1034  continue;
1035 
1036  snprintf(path, MAXPGPATH, "pg_replslot/%s", replication_de->d_name);
1037 
1038  /* we're only creating directories here, skip if it's not our's */
1039  if (lstat(path, &statbuf) == 0 && !S_ISDIR(statbuf.st_mode))
1040  continue;
1041 
1042  /* we crashed while a slot was being setup or deleted, clean up */
1043  if (pg_str_endswith(replication_de->d_name, ".tmp"))
1044  {
1045  if (!rmtree(path, true))
1046  {
1047  ereport(WARNING,
1049  errmsg("could not remove directory \"%s\"", path)));
1050  continue;
1051  }
1052  fsync_fname("pg_replslot", true);
1053  continue;
1054  }
1055 
1056  /* looks like a slot in a normal state, restore */
1057  RestoreSlotFromDisk(replication_de->d_name);
1058  }
1059  FreeDir(replication_dir);
1060 
1061  /* currently no slots exist, we're done. */
1062  if (max_replication_slots <= 0)
1063  return;
1064 
1065  /* Now that we have recovered all the data, compute replication xmin */
1068 }
static void RestoreSlotFromDisk(const char *name)
Definition: slot.c:1262
#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:567
int snprintf(char *str, size_t count, const char *fmt,...) pg_attribute_printf(3
Definition: dirent.h:9
void ReplicationSlotsComputeRequiredLSN(void)
Definition: slot.c:665
Definition: dirent.c:25
#define MAXPGPATH
int errcode_for_file_access(void)
Definition: elog.c:598
DIR * AllocateDir(const char *dirname)
Definition: fd.c:2335
#define ereport(elevel, rest)
Definition: elog.h:122
#define WARNING
Definition: elog.h:40
bool rmtree(const char *path, bool rmtopdir)
Definition: rmtree.c:36
int max_replication_slots
Definition: slot.c:99
#define NULL
Definition: c.h:229
struct dirent * ReadDir(DIR *dir, const char *dirname)
Definition: fd.c:2401
int errmsg(const char *fmt,...)
Definition: elog.c:797
char d_name[MAX_PATH]
Definition: dirent.h:14
#define elog
Definition: elog.h:219
#define lstat(path, sb)
Definition: win32.h:272
void ReplicationSlotsComputeRequiredXmin(bool already_locked)
Definition: slot.c:617
int FreeDir(DIR *dir)
Definition: fd.c:2444

Variable Documentation

ReplicationSlotCtlData* ReplicationSlotCtl = NULL

Definition at line 93 of file slot.c.

Referenced by pg_get_replication_slots().