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, bool nowait)
 
void ReplicationSlotRelease (void)
 
void ReplicationSlotCleanup (void)
 
void ReplicationSlotDrop (const char *name, bool nowait)
 
void ReplicationSlotSave (void)
 
void ReplicationSlotMarkDirty (void)
 
void ReplicationSlotPersist (void)
 
void ReplicationSlotsComputeRequiredXmin (bool already_locked)
 
void ReplicationSlotsComputeRequiredLSN (void)
 
XLogRecPtr ReplicationSlotsComputeLogicalRestartLSN (void)
 
bool ReplicationSlotsCountDBSlots (Oid dboid, int *nslots, int *nactive)
 
void ReplicationSlotsDropDBSlots (Oid dboid)
 
void CheckSlotRequirements (void)
 
void 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 1055 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().

1056 {
1057  int i;
1058 
1059  elog(DEBUG1, "performing replication slot checkpoint");
1060 
1061  /*
1062  * Prevent any slot from being created/dropped while we're active. As we
1063  * explicitly do *not* want to block iterating over replication_slots or
1064  * acquiring a slot we cannot take the control lock - but that's OK,
1065  * because holding ReplicationSlotAllocationLock is strictly stronger, and
1066  * enough to guarantee that nobody can change the in_use bits on us.
1067  */
1068  LWLockAcquire(ReplicationSlotAllocationLock, LW_SHARED);
1069 
1070  for (i = 0; i < max_replication_slots; i++)
1071  {
1073  char path[MAXPGPATH];
1074 
1075  if (!s->in_use)
1076  continue;
1077 
1078  /* save the slot to disk, locking is handled in SaveSlotToPath() */
1079  sprintf(path, "pg_replslot/%s", NameStr(s->data.name));
1080  SaveSlotToPath(s, path, LOG);
1081  }
1082  LWLockRelease(ReplicationSlotAllocationLock);
1083 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
#define DEBUG1
Definition: elog.h:25
ReplicationSlotPersistentData data
Definition: slot.h:116
#define LOG
Definition: elog.h:26
static void SaveSlotToPath(ReplicationSlot *slot, const char *path, int elevel)
Definition: slot.c:1210
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1715
#define MAXPGPATH
bool in_use
Definition: slot.h:92
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:149
#define elog
Definition: elog.h:219
void CheckSlotRequirements ( void  )

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

966 {
967  if (max_replication_slots == 0)
968  ereport(ERROR,
969  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
970  (errmsg("replication slots can only be used if max_replication_slots > 0"))));
971 
973  ereport(ERROR,
974  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
975  errmsg("replication slots can only be used if wal_level >= replica")));
976 }
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 1150 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().

1151 {
1152  char tmppath[MAXPGPATH];
1153  char path[MAXPGPATH];
1154  struct stat st;
1155 
1156  /*
1157  * No need to take out the io_in_progress_lock, nobody else can see this
1158  * slot yet, so nobody else will write. We're reusing SaveSlotToPath which
1159  * takes out the lock, if we'd take the lock here, we'd deadlock.
1160  */
1161 
1162  sprintf(path, "pg_replslot/%s", NameStr(slot->data.name));
1163  sprintf(tmppath, "pg_replslot/%s.tmp", NameStr(slot->data.name));
1164 
1165  /*
1166  * It's just barely possible that some previous effort to create or drop a
1167  * slot with this name left a temp directory lying around. If that seems
1168  * to be the case, try to remove it. If the rmtree() fails, we'll error
1169  * out at the mkdir() below, so we don't bother checking success.
1170  */
1171  if (stat(tmppath, &st) == 0 && S_ISDIR(st.st_mode))
1172  rmtree(tmppath, true);
1173 
1174  /* Create and fsync the temporary slot directory. */
1175  if (mkdir(tmppath, S_IRWXU) < 0)
1176  ereport(ERROR,
1178  errmsg("could not create directory \"%s\": %m",
1179  tmppath)));
1180  fsync_fname(tmppath, true);
1181 
1182  /* Write the actual state file. */
1183  slot->dirty = true; /* signal that we really need to write */
1184  SaveSlotToPath(slot, tmppath, ERROR);
1185 
1186  /* Rename the directory into place. */
1187  if (rename(tmppath, path) != 0)
1188  ereport(ERROR,
1190  errmsg("could not rename file \"%s\" to \"%s\": %m",
1191  tmppath, path)));
1192 
1193  /*
1194  * If we'd now fail - really unlikely - we wouldn't know whether this slot
1195  * would persist after an OS crash or not - so, force a restart. The
1196  * restart would try to fsync this again till it works.
1197  */
1199 
1200  fsync_fname(path, true);
1201  fsync_fname("pg_replslot", true);
1202 
1203  END_CRIT_SECTION();
1204 }
#define mkdir(a, b)
Definition: win32.h:57
#define END_CRIT_SECTION()
Definition: miscadmin.h:133
void fsync_fname(const char *fname, bool isdir)
Definition: fd.c:567
#define START_CRIT_SECTION()
Definition: miscadmin.h:131
ReplicationSlotPersistentData data
Definition: slot.h:116
static void SaveSlotToPath(ReplicationSlot *slot, const char *path, int elevel)
Definition: slot.c:1210
#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:99
void ReplicationSlotAcquire ( const char *  name,
bool  nowait 
)

Definition at line 330 of file slot.c.

References ReplicationSlot::active_cv, ReplicationSlot::active_pid, Assert, ConditionVariableBroadcast(), ConditionVariableCancelSleep(), ConditionVariablePrepareToSleep(), ConditionVariableSleep(), 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, SpinLockAcquire, SpinLockRelease, and WAIT_EVENT_REPLICATION_SLOT_DROP.

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

331 {
332  ReplicationSlot *slot;
333  int active_pid;
334  int i;
335 
336 retry:
338 
339  /*
340  * Search for the named slot and mark it active if we find it. If the
341  * slot is already active, we exit the loop with active_pid set to the PID
342  * of the backend that owns it.
343  */
344  active_pid = 0;
345  slot = NULL;
346  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
347  for (i = 0; i < max_replication_slots; i++)
348  {
350 
351  if (s->in_use && strcmp(name, NameStr(s->data.name)) == 0)
352  {
353  /*
354  * This is the slot we want. We don't know yet if it's active, so
355  * get ready to sleep on it in case it is. (We may end up not
356  * sleeping, but we don't want to do this while holding the
357  * spinlock.)
358  */
360 
361  SpinLockAcquire(&s->mutex);
362 
363  active_pid = s->active_pid;
364  if (active_pid == 0)
365  active_pid = s->active_pid = MyProcPid;
366 
367  SpinLockRelease(&s->mutex);
368  slot = s;
369 
370  break;
371  }
372  }
373  LWLockRelease(ReplicationSlotControlLock);
374 
375  /* If we did not find the slot, error out. */
376  if (slot == NULL)
377  ereport(ERROR,
378  (errcode(ERRCODE_UNDEFINED_OBJECT),
379  errmsg("replication slot \"%s\" does not exist", name)));
380 
381  /*
382  * If we found the slot but it's already active in another backend, we
383  * either error out or retry after a short wait, as caller specified.
384  */
385  if (active_pid != MyProcPid)
386  {
387  if (nowait)
388  ereport(ERROR,
389  (errcode(ERRCODE_OBJECT_IN_USE),
390  errmsg("replication slot \"%s\" is active for PID %d",
391  name, active_pid)));
392 
393  /* Wait here until we get signaled, and then restart */
397  goto retry;
398  }
399  else
400  ConditionVariableCancelSleep(); /* no sleep needed after all */
401 
402  /* Let everybody know we've modified this slot */
404 
405  /* We made this slot active, so it's ours now. */
406  MyReplicationSlot = slot;
407 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
int MyProcPid
Definition: globals.c:39
int ConditionVariableBroadcast(ConditionVariable *cv)
int errcode(int sqlerrcode)
Definition: elog.c:575
ReplicationSlotPersistentData data
Definition: slot.h:116
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1715
void ConditionVariablePrepareToSleep(ConditionVariable *cv)
#define SpinLockAcquire(lock)
Definition: spin.h:62
void ConditionVariableCancelSleep(void)
#define ERROR
Definition: elog.h:43
#define ereport(elevel, rest)
Definition: elog.h:122
bool in_use
Definition: slot.h:92
#define SpinLockRelease(lock)
Definition: spin.h:64
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
int max_replication_slots
Definition: slot.c:99
void ConditionVariableSleep(ConditionVariable *cv, uint32 wait_event_info)
#define NULL
Definition: c.h:229
#define Assert(condition)
Definition: c.h:676
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1111
ConditionVariable active_cv
Definition: slot.h:122
const char * name
Definition: encode.c:521
int errmsg(const char *fmt,...)
Definition: elog.c:797
pid_t active_pid
Definition: slot.h:95
int i
#define NameStr(name)
Definition: c.h:499
ReplicationSlot replication_slots[1]
Definition: slot.h:149
slock_t mutex
Definition: slot.h:89
void ReplicationSlotCleanup ( void  )

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

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

472 {
473  int i;
474 
476 
477 restart:
478  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
479  for (i = 0; i < max_replication_slots; i++)
480  {
482 
483  if (!s->in_use)
484  continue;
485 
486  SpinLockAcquire(&s->mutex);
487  if (s->active_pid == MyProcPid)
488  {
490  SpinLockRelease(&s->mutex);
491  LWLockRelease(ReplicationSlotControlLock); /* avoid deadlock */
492 
494 
496  goto restart;
497  }
498  else
499  SpinLockRelease(&s->mutex);
500  }
501 
502  LWLockRelease(ReplicationSlotControlLock);
503 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
int MyProcPid
Definition: globals.c:39
int ConditionVariableBroadcast(ConditionVariable *cv)
ReplicationSlotPersistency persistency
Definition: slot.h:50
static void ReplicationSlotDropPtr(ReplicationSlot *slot)
Definition: slot.c:539
ReplicationSlotPersistentData data
Definition: slot.h:116
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1715
#define SpinLockAcquire(lock)
Definition: spin.h:62
bool in_use
Definition: slot.h:92
#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:676
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1111
ConditionVariable active_cv
Definition: slot.h:122
pid_t active_pid
Definition: slot.h:95
int i
ReplicationSlot replication_slots[1]
Definition: slot.h:149
slock_t mutex
Definition: slot.h:89
void ReplicationSlotCreate ( const char *  name,
bool  db_specific,
ReplicationSlotPersistency  persistency 
)

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

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

Definition at line 509 of file slot.c.

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

Referenced by DropReplicationSlot(), and pg_drop_replication_slot().

510 {
512 
513  ReplicationSlotAcquire(name, nowait);
514 
516 }
void ReplicationSlotAcquire(const char *name, bool nowait)
Definition: slot.c:330
static void ReplicationSlotDropAcquired(void)
Definition: slot.c:522
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
#define NULL
Definition: c.h:229
#define Assert(condition)
Definition: c.h:676
const char * name
Definition: encode.c:521
static void ReplicationSlotDropAcquired ( void  )
static

Definition at line 522 of file slot.c.

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

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

523 {
525 
527 
528  /* slot isn't acquired anymore */
530 
532 }
static void ReplicationSlotDropPtr(ReplicationSlot *slot)
Definition: slot.c:539
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
#define NULL
Definition: c.h:229
#define Assert(condition)
Definition: c.h:676
static void ReplicationSlotDropPtr ( ReplicationSlot slot)
static

Definition at line 539 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, ReplicationSlotsComputeRequiredLSN(), ReplicationSlotsComputeRequiredXmin(), rmtree(), RS_PERSISTENT, SpinLockAcquire, SpinLockRelease, START_CRIT_SECTION, and WARNING.

Referenced by ReplicationSlotCleanup(), and ReplicationSlotDropAcquired().

540 {
541  char path[MAXPGPATH];
542  char tmppath[MAXPGPATH];
543 
544  /*
545  * If some other backend ran this code concurrently with us, we might try
546  * to delete a slot with a certain name while someone else was trying to
547  * create a slot with the same name.
548  */
549  LWLockAcquire(ReplicationSlotAllocationLock, LW_EXCLUSIVE);
550 
551  /* Generate pathnames. */
552  sprintf(path, "pg_replslot/%s", NameStr(slot->data.name));
553  sprintf(tmppath, "pg_replslot/%s.tmp", NameStr(slot->data.name));
554 
555  /*
556  * Rename the slot directory on disk, so that we'll no longer recognize
557  * this as a valid slot. Note that if this fails, we've got to mark the
558  * slot inactive before bailing out. If we're dropping an ephemeral or a
559  * temporary slot, we better never fail hard as the caller won't expect
560  * the slot to survive and this might get called during error handling.
561  */
562  if (rename(path, tmppath) == 0)
563  {
564  /*
565  * We need to fsync() the directory we just renamed and its parent to
566  * make sure that our changes are on disk in a crash-safe fashion. If
567  * fsync() fails, we can't be sure whether the changes are on disk or
568  * not. For now, we handle that by panicking;
569  * StartupReplicationSlots() will try to straighten it out after
570  * restart.
571  */
573  fsync_fname(tmppath, true);
574  fsync_fname("pg_replslot", true);
576  }
577  else
578  {
579  bool fail_softly = slot->data.persistency != RS_PERSISTENT;
580 
581  SpinLockAcquire(&slot->mutex);
582  slot->active_pid = 0;
583  SpinLockRelease(&slot->mutex);
584 
585  /* wake up anyone waiting on this slot */
587 
588  ereport(fail_softly ? WARNING : ERROR,
590  errmsg("could not rename file \"%s\" to \"%s\": %m",
591  path, tmppath)));
592  }
593 
594  /*
595  * The slot is definitely gone. Lock out concurrent scans of the array
596  * long enough to kill it. It's OK to clear the active PID here without
597  * grabbing the mutex because nobody else can be scanning the array here,
598  * and nobody can be attached to this slot and thus access it without
599  * scanning the array.
600  *
601  * Also wake up processes waiting for it.
602  */
603  LWLockAcquire(ReplicationSlotControlLock, LW_EXCLUSIVE);
604  slot->active_pid = 0;
605  slot->in_use = false;
606  LWLockRelease(ReplicationSlotControlLock);
608 
609  /*
610  * Slot is dead and doesn't prevent resource removal anymore, recompute
611  * limits.
612  */
615 
616  /*
617  * If removing the directory fails, the worst thing that will happen is
618  * that the user won't be able to create a new slot with the same name
619  * until the next server restart. We warn about it, but that's all.
620  */
621  if (!rmtree(tmppath, true))
624  errmsg("could not remove directory \"%s\"", tmppath)));
625 
626  /*
627  * We release this at the very end, so that nobody starts trying to create
628  * a slot while we're still cleaning up the detritus of the old one.
629  */
630  LWLockRelease(ReplicationSlotAllocationLock);
631 }
int ConditionVariableBroadcast(ConditionVariable *cv)
#define END_CRIT_SECTION()
Definition: miscadmin.h:133
void fsync_fname(const char *fname, bool isdir)
Definition: fd.c:567
ReplicationSlotPersistency persistency
Definition: slot.h:50
#define START_CRIT_SECTION()
Definition: miscadmin.h:131
ReplicationSlotPersistentData data
Definition: slot.h:116
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1715
#define SpinLockAcquire(lock)
Definition: spin.h:62
void ReplicationSlotsComputeRequiredLSN(void)
Definition: slot.c:741
#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:92
#define SpinLockRelease(lock)
Definition: spin.h:64
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1111
ConditionVariable active_cv
Definition: slot.h:122
int errmsg(const char *fmt,...)
Definition: elog.c:797
pid_t active_pid
Definition: slot.h:95
#define NameStr(name)
Definition: c.h:499
slock_t mutex
Definition: slot.h:89
void ReplicationSlotsComputeRequiredXmin(bool already_locked)
Definition: slot.c:695
void ReplicationSlotMarkDirty ( void  )

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

657 {
659 
661 
662  SpinLockAcquire(&slot->mutex);
664  MyReplicationSlot->dirty = true;
665  SpinLockRelease(&slot->mutex);
666 }
#define SpinLockAcquire(lock)
Definition: spin.h:62
#define SpinLockRelease(lock)
Definition: spin.h:64
bool just_dirtied
Definition: slot.h:98
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
#define NULL
Definition: c.h:229
#define Assert(condition)
Definition: c.h:676
slock_t mutex
Definition: slot.h:89
bool dirty
Definition: slot.h:99
void ReplicationSlotPersist ( void  )

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

674 {
676 
677  Assert(slot != NULL);
679 
680  SpinLockAcquire(&slot->mutex);
682  SpinLockRelease(&slot->mutex);
683 
686 }
ReplicationSlotPersistency persistency
Definition: slot.h:50
void ReplicationSlotSave(void)
Definition: slot.c:638
ReplicationSlotPersistentData data
Definition: slot.h:116
#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:676
slock_t mutex
Definition: slot.h:89
void ReplicationSlotMarkDirty(void)
Definition: slot.c:656
void ReplicationSlotRelease ( void  )

Definition at line 416 of file slot.c.

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

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

417 {
419 
420  Assert(slot != NULL && slot->active_pid != 0);
421 
422  if (slot->data.persistency == RS_EPHEMERAL)
423  {
424  /*
425  * Delete the slot. There is no !PANIC case where this is allowed to
426  * fail, all that may happen is an incomplete cleanup of the on-disk
427  * data.
428  */
430  }
431 
432  /*
433  * If slot needed to temporarily restrain both data and catalog xmin to
434  * create the catalog snapshot, remove that temporary constraint.
435  * Snapshots can only be exported while the initial snapshot is still
436  * acquired.
437  */
438  if (!TransactionIdIsValid(slot->data.xmin) &&
440  {
441  SpinLockAcquire(&slot->mutex);
443  SpinLockRelease(&slot->mutex);
445  }
446 
447  if (slot->data.persistency == RS_PERSISTENT)
448  {
449  /*
450  * Mark persistent slot inactive. We're not freeing it, just
451  * disconnecting, but wake up others that may be waiting for it.
452  */
453  SpinLockAcquire(&slot->mutex);
454  slot->active_pid = 0;
455  SpinLockRelease(&slot->mutex);
457  }
458 
460 
461  /* might not have been set when we've been a plain slot */
462  LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
464  LWLockRelease(ProcArrayLock);
465 }
#define PROC_IN_LOGICAL_DECODING
Definition: proc.h:56
int ConditionVariableBroadcast(ConditionVariable *cv)
ReplicationSlotPersistency persistency
Definition: slot.h:50
ReplicationSlotPersistentData data
Definition: slot.h:116
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:522
TransactionId effective_xmin
Definition: slot.h:112
#define InvalidTransactionId
Definition: transam.h:31
TransactionId xmin
Definition: slot.h:58
#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:676
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1111
ConditionVariable active_cv
Definition: slot.h:122
pid_t active_pid
Definition: slot.h:95
slock_t mutex
Definition: slot.h:89
#define TransactionIdIsValid(xid)
Definition: transam.h:41
void ReplicationSlotsComputeRequiredXmin(bool already_locked)
Definition: slot.c:695
void ReplicationSlotReserveWal ( void  )

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

986 {
988 
989  Assert(slot != NULL);
991 
992  /*
993  * The replication slot mechanism is used to prevent removal of required
994  * WAL. As there is no interlock between this routine and checkpoints, WAL
995  * segments could concurrently be removed when a now stale return value of
996  * ReplicationSlotsComputeRequiredLSN() is used. In the unlikely case that
997  * this happens we'll just retry.
998  */
999  while (true)
1000  {
1001  XLogSegNo segno;
1002 
1003  /*
1004  * For logical slots log a standby snapshot and start logical decoding
1005  * at exactly that position. That allows the slot to start up more
1006  * quickly.
1007  *
1008  * That's not needed (or indeed helpful) for physical slots as they'll
1009  * start replay at the last logged checkpoint anyway. Instead return
1010  * the location of the last redo LSN. While that slightly increases
1011  * the chance that we have to retry, it's where a base backup has to
1012  * start replay at.
1013  */
1014  if (!RecoveryInProgress() && SlotIsLogical(slot))
1015  {
1016  XLogRecPtr flushptr;
1017 
1018  /* start at current insert position */
1020 
1021  /* make sure we have enough information to start */
1022  flushptr = LogStandbySnapshot();
1023 
1024  /* and make sure it's fsynced to disk */
1025  XLogFlush(flushptr);
1026  }
1027  else
1028  {
1029  slot->data.restart_lsn = GetRedoRecPtr();
1030  }
1031 
1032  /* prevent WAL removal as fast as possible */
1034 
1035  /*
1036  * If all required WAL is still there, great, otherwise retry. The
1037  * slot should prevent further removal of WAL, unless there's a
1038  * concurrent ReplicationSlotsComputeRequiredLSN() after we've written
1039  * the new restart_lsn above, so normally we should never need to loop
1040  * more than twice.
1041  */
1042  XLByteToSeg(slot->data.restart_lsn, segno);
1043  if (XLogGetLastRemovedSegno() < segno)
1044  break;
1045  }
1046 }
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
ReplicationSlotPersistentData data
Definition: slot.h:116
bool RecoveryInProgress(void)
Definition: xlog.c:7878
void XLogFlush(XLogRecPtr record)
Definition: xlog.c:2757
XLogSegNo XLogGetLastRemovedSegno(void)
Definition: xlog.c:3793
void ReplicationSlotsComputeRequiredLSN(void)
Definition: slot.c:741
XLogRecPtr LogStandbySnapshot(void)
Definition: standby.c:909
XLogRecPtr GetXLogInsertRecPtr(void)
Definition: xlog.c:11117
uint64 XLogSegNo
Definition: xlogdefs.h:34
#define SlotIsLogical(slot)
Definition: slot.h:138
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:676
XLogRecPtr restart_lsn
Definition: slot.h:69
XLogRecPtr GetRedoRecPtr(void)
Definition: xlog.c:8168
void ReplicationSlotSave ( void  )

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

639 {
640  char path[MAXPGPATH];
641 
643 
644  sprintf(path, "pg_replslot/%s", NameStr(MyReplicationSlot->data.name));
646 }
ReplicationSlotPersistentData data
Definition: slot.h:116
static void SaveSlotToPath(ReplicationSlot *slot, const char *path, int elevel)
Definition: slot.c:1210
#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:676
#define NameStr(name)
Definition: c.h:499
XLogRecPtr ReplicationSlotsComputeLogicalRestartLSN ( void  )

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

785 {
787  int i;
788 
789  if (max_replication_slots <= 0)
790  return InvalidXLogRecPtr;
791 
792  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
793 
794  for (i = 0; i < max_replication_slots; i++)
795  {
796  ReplicationSlot *s;
797  XLogRecPtr restart_lsn;
798 
800 
801  /* cannot change while ReplicationSlotCtlLock is held */
802  if (!s->in_use)
803  continue;
804 
805  /* we're only interested in logical slots */
806  if (!SlotIsLogical(s))
807  continue;
808 
809  /* read once, it's ok if it increases while we're checking */
810  SpinLockAcquire(&s->mutex);
811  restart_lsn = s->data.restart_lsn;
812  SpinLockRelease(&s->mutex);
813 
814  if (result == InvalidXLogRecPtr ||
815  restart_lsn < result)
816  result = restart_lsn;
817  }
818 
819  LWLockRelease(ReplicationSlotControlLock);
820 
821  return result;
822 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
return result
Definition: formatting.c:1633
ReplicationSlotPersistentData data
Definition: slot.h:116
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1715
#define SpinLockAcquire(lock)
Definition: spin.h:62
#define SlotIsLogical(slot)
Definition: slot.h:138
bool in_use
Definition: slot.h:92
#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:69
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1111
int i
ReplicationSlot replication_slots[1]
Definition: slot.h:149
slock_t mutex
Definition: slot.h:89
void ReplicationSlotsComputeRequiredLSN ( void  )

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

742 {
743  int i;
744  XLogRecPtr min_required = InvalidXLogRecPtr;
745 
747 
748  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
749  for (i = 0; i < max_replication_slots; i++)
750  {
752  XLogRecPtr restart_lsn;
753 
754  if (!s->in_use)
755  continue;
756 
757  SpinLockAcquire(&s->mutex);
758  restart_lsn = s->data.restart_lsn;
759  SpinLockRelease(&s->mutex);
760 
761  if (restart_lsn != InvalidXLogRecPtr &&
762  (min_required == InvalidXLogRecPtr ||
763  restart_lsn < min_required))
764  min_required = restart_lsn;
765  }
766  LWLockRelease(ReplicationSlotControlLock);
767 
768  XLogSetReplicationSlotMinimumLSN(min_required);
769 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
ReplicationSlotPersistentData data
Definition: slot.h:116
void XLogSetReplicationSlotMinimumLSN(XLogRecPtr lsn)
Definition: xlog.c:2648
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1715
#define SpinLockAcquire(lock)
Definition: spin.h:62
bool in_use
Definition: slot.h:92
#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:676
XLogRecPtr restart_lsn
Definition: slot.h:69
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1111
int i
ReplicationSlot replication_slots[1]
Definition: slot.h:149
slock_t mutex
Definition: slot.h:89
void ReplicationSlotsComputeRequiredXmin ( bool  already_locked)

Definition at line 695 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(), ReplicationSlotRelease(), and StartupReplicationSlots().

696 {
697  int i;
699  TransactionId agg_catalog_xmin = InvalidTransactionId;
700 
702 
703  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
704 
705  for (i = 0; i < max_replication_slots; i++)
706  {
708  TransactionId effective_xmin;
709  TransactionId effective_catalog_xmin;
710 
711  if (!s->in_use)
712  continue;
713 
714  SpinLockAcquire(&s->mutex);
715  effective_xmin = s->effective_xmin;
716  effective_catalog_xmin = s->effective_catalog_xmin;
717  SpinLockRelease(&s->mutex);
718 
719  /* check the data xmin */
720  if (TransactionIdIsValid(effective_xmin) &&
721  (!TransactionIdIsValid(agg_xmin) ||
722  TransactionIdPrecedes(effective_xmin, agg_xmin)))
723  agg_xmin = effective_xmin;
724 
725  /* check the catalog xmin */
726  if (TransactionIdIsValid(effective_catalog_xmin) &&
727  (!TransactionIdIsValid(agg_catalog_xmin) ||
728  TransactionIdPrecedes(effective_catalog_xmin, agg_catalog_xmin)))
729  agg_catalog_xmin = effective_catalog_xmin;
730  }
731 
732  LWLockRelease(ReplicationSlotControlLock);
733 
734  ProcArraySetReplicationSlotXmin(agg_xmin, agg_catalog_xmin, already_locked);
735 }
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:112
#define InvalidTransactionId
Definition: transam.h:31
bool TransactionIdPrecedes(TransactionId id1, TransactionId id2)
Definition: transam.c:300
bool in_use
Definition: slot.h:92
#define SpinLockRelease(lock)
Definition: spin.h:64
TransactionId effective_catalog_xmin
Definition: slot.h:113
int max_replication_slots
Definition: slot.c:99
#define NULL
Definition: c.h:229
#define Assert(condition)
Definition: c.h:676
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:2964
ReplicationSlot replication_slots[1]
Definition: slot.h:149
slock_t mutex
Definition: slot.h:89
#define TransactionIdIsValid(xid)
Definition: transam.h:41
bool ReplicationSlotsCountDBSlots ( Oid  dboid,
int *  nslots,
int *  nactive 
)

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

834 {
835  int i;
836 
837  *nslots = *nactive = 0;
838 
839  if (max_replication_slots <= 0)
840  return false;
841 
842  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
843  for (i = 0; i < max_replication_slots; i++)
844  {
845  ReplicationSlot *s;
846 
848 
849  /* cannot change while ReplicationSlotCtlLock is held */
850  if (!s->in_use)
851  continue;
852 
853  /* only logical slots are database specific, skip */
854  if (!SlotIsLogical(s))
855  continue;
856 
857  /* not our database, skip */
858  if (s->data.database != dboid)
859  continue;
860 
861  /* count slots with spinlock held */
862  SpinLockAcquire(&s->mutex);
863  (*nslots)++;
864  if (s->active_pid != 0)
865  (*nactive)++;
866  SpinLockRelease(&s->mutex);
867  }
868  LWLockRelease(ReplicationSlotControlLock);
869 
870  if (*nslots > 0)
871  return true;
872  return false;
873 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
ReplicationSlotPersistentData data
Definition: slot.h:116
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1715
#define SpinLockAcquire(lock)
Definition: spin.h:62
#define SlotIsLogical(slot)
Definition: slot.h:138
bool in_use
Definition: slot.h:92
#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:95
int i
ReplicationSlot replication_slots[1]
Definition: slot.h:149
slock_t mutex
Definition: slot.h:89
void ReplicationSlotsDropDBSlots ( Oid  dboid)

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

890 {
891  int i;
892 
893  if (max_replication_slots <= 0)
894  return;
895 
896 restart:
897  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
898  for (i = 0; i < max_replication_slots; i++)
899  {
900  ReplicationSlot *s;
901  char *slotname;
902  int active_pid;
903 
905 
906  /* cannot change while ReplicationSlotCtlLock is held */
907  if (!s->in_use)
908  continue;
909 
910  /* only logical slots are database specific, skip */
911  if (!SlotIsLogical(s))
912  continue;
913 
914  /* not our database, skip */
915  if (s->data.database != dboid)
916  continue;
917 
918  /* acquire slot, so ReplicationSlotDropAcquired can be reused */
919  SpinLockAcquire(&s->mutex);
920  /* can't change while ReplicationSlotControlLock is held */
921  slotname = NameStr(s->data.name);
922  active_pid = s->active_pid;
923  if (active_pid == 0)
924  {
925  MyReplicationSlot = s;
926  s->active_pid = MyProcPid;
927  }
928  SpinLockRelease(&s->mutex);
929 
930  /*
931  * Even though we hold an exclusive lock on the database object a
932  * logical slot for that DB can still be active, e.g. if it's
933  * concurrently being dropped by a backend connected to another DB.
934  *
935  * That's fairly unlikely in practice, so we'll just bail out.
936  */
937  if (active_pid)
938  ereport(ERROR,
939  (errcode(ERRCODE_OBJECT_IN_USE),
940  errmsg("replication slot \"%s\" is active for PID %d",
941  slotname, active_pid)));
942 
943  /*
944  * To avoid duplicating ReplicationSlotDropAcquired() and to avoid
945  * holding ReplicationSlotControlLock over filesystem operations,
946  * release ReplicationSlotControlLock and use
947  * ReplicationSlotDropAcquired.
948  *
949  * As that means the set of slots could change, restart scan from the
950  * beginning each time we release the lock.
951  */
952  LWLockRelease(ReplicationSlotControlLock);
954  goto restart;
955  }
956  LWLockRelease(ReplicationSlotControlLock);
957 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
int MyProcPid
Definition: globals.c:39
int errcode(int sqlerrcode)
Definition: elog.c:575
ReplicationSlotPersistentData data
Definition: slot.h:116
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1715
#define SpinLockAcquire(lock)
Definition: spin.h:62
static void ReplicationSlotDropAcquired(void)
Definition: slot.c:522
#define ERROR
Definition: elog.h:43
#define SlotIsLogical(slot)
Definition: slot.h:138
#define ereport(elevel, rest)
Definition: elog.h:122
bool in_use
Definition: slot.h:92
#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
int errmsg(const char *fmt,...)
Definition: elog.c:797
pid_t active_pid
Definition: slot.h:95
int i
#define NameStr(name)
Definition: c.h:499
ReplicationSlot replication_slots[1]
Definition: slot.h:149
slock_t mutex
Definition: slot.h:89
void ReplicationSlotsShmemInit ( void  )

Definition at line 132 of file slot.c.

References ReplicationSlot::active_cv, ConditionVariableInit(), 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);
161  }
162  }
163 }
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:858
void ConditionVariableInit(ConditionVariable *cv)
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
ConditionVariable active_cv
Definition: slot.h:122
int i
ReplicationSlot replication_slots[1]
Definition: slot.h:149
slock_t mutex
Definition: slot.h:89
LWLock io_in_progress_lock
Definition: slot.h:119
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 174 of file slot.c.

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

Referenced by readRecoveryCommandFile(), 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: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:136
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 1334 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().

1335 {
1337  int i;
1338  char path[MAXPGPATH + 22];
1339  int fd;
1340  bool restored = false;
1341  int readBytes;
1342  pg_crc32c checksum;
1343 
1344  /* no need to lock here, no concurrent access allowed yet */
1345 
1346  /* delete temp file if it exists */
1347  sprintf(path, "pg_replslot/%s/state.tmp", name);
1348  if (unlink(path) < 0 && errno != ENOENT)
1349  ereport(PANIC,
1351  errmsg("could not remove file \"%s\": %m", path)));
1352 
1353  sprintf(path, "pg_replslot/%s/state", name);
1354 
1355  elog(DEBUG1, "restoring replication slot from \"%s\"", path);
1356 
1357  fd = OpenTransientFile(path, O_RDWR | PG_BINARY, 0);
1358 
1359  /*
1360  * We do not need to handle this as we are rename()ing the directory into
1361  * place only after we fsync()ed the state file.
1362  */
1363  if (fd < 0)
1364  ereport(PANIC,
1366  errmsg("could not open file \"%s\": %m", path)));
1367 
1368  /*
1369  * Sync state file before we're reading from it. We might have crashed
1370  * while it wasn't synced yet and we shouldn't continue on that basis.
1371  */
1373  if (pg_fsync(fd) != 0)
1374  {
1375  CloseTransientFile(fd);
1376  ereport(PANIC,
1378  errmsg("could not fsync file \"%s\": %m",
1379  path)));
1380  }
1382 
1383  /* Also sync the parent directory */
1385  fsync_fname(path, true);
1386  END_CRIT_SECTION();
1387 
1388  /* read part of statefile that's guaranteed to be version independent */
1390  readBytes = read(fd, &cp, ReplicationSlotOnDiskConstantSize);
1392  if (readBytes != ReplicationSlotOnDiskConstantSize)
1393  {
1394  int saved_errno = errno;
1395 
1396  CloseTransientFile(fd);
1397  errno = saved_errno;
1398  ereport(PANIC,
1400  errmsg("could not read file \"%s\", read %d of %u: %m",
1401  path, readBytes,
1403  }
1404 
1405  /* verify magic */
1406  if (cp.magic != SLOT_MAGIC)
1407  ereport(PANIC,
1409  errmsg("replication slot file \"%s\" has wrong magic number: %u instead of %u",
1410  path, cp.magic, SLOT_MAGIC)));
1411 
1412  /* verify version */
1413  if (cp.version != SLOT_VERSION)
1414  ereport(PANIC,
1416  errmsg("replication slot file \"%s\" has unsupported version %u",
1417  path, cp.version)));
1418 
1419  /* boundary check on length */
1421  ereport(PANIC,
1423  errmsg("replication slot file \"%s\" has corrupted length %u",
1424  path, cp.length)));
1425 
1426  /* Now that we know the size, read the entire file */
1428  readBytes = read(fd,
1429  (char *) &cp + ReplicationSlotOnDiskConstantSize,
1430  cp.length);
1432  if (readBytes != cp.length)
1433  {
1434  int saved_errno = errno;
1435 
1436  CloseTransientFile(fd);
1437  errno = saved_errno;
1438  ereport(PANIC,
1440  errmsg("could not read file \"%s\", read %d of %u: %m",
1441  path, readBytes, cp.length)));
1442  }
1443 
1444  CloseTransientFile(fd);
1445 
1446  /* now verify the CRC */
1447  INIT_CRC32C(checksum);
1448  COMP_CRC32C(checksum,
1449  (char *) &cp + SnapBuildOnDiskNotChecksummedSize,
1451  FIN_CRC32C(checksum);
1452 
1453  if (!EQ_CRC32C(checksum, cp.checksum))
1454  ereport(PANIC,
1455  (errmsg("checksum mismatch for replication slot file \"%s\": is %u, should be %u",
1456  path, checksum, cp.checksum)));
1457 
1458  /*
1459  * If we crashed with an ephemeral slot active, don't restore but delete
1460  * it.
1461  */
1463  {
1464  sprintf(path, "pg_replslot/%s", name);
1465 
1466  if (!rmtree(path, true))
1467  {
1468  ereport(WARNING,
1470  errmsg("could not remove directory \"%s\"", path)));
1471  }
1472  fsync_fname("pg_replslot", true);
1473  return;
1474  }
1475 
1476  /* nothing can be active yet, don't lock anything */
1477  for (i = 0; i < max_replication_slots; i++)
1478  {
1479  ReplicationSlot *slot;
1480 
1482 
1483  if (slot->in_use)
1484  continue;
1485 
1486  /* restore the entire set of persistent data */
1487  memcpy(&slot->data, &cp.slotdata,
1489 
1490  /* initialize in memory state */
1491  slot->effective_xmin = cp.slotdata.xmin;
1493 
1498 
1499  slot->in_use = true;
1500  slot->active_pid = 0;
1501 
1502  restored = true;
1503  break;
1504  }
1505 
1506  if (!restored)
1507  ereport(PANIC,
1508  (errmsg("too many replication slots active before shutdown"),
1509  errhint("Increase max_replication_slots and try again.")));
1510 }
#define INIT_CRC32C(crc)
Definition: pg_crc32c.h:41
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
TransactionId candidate_catalog_xmin
Definition: slot.h:131
#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:133
void fsync_fname(const char *fname, bool isdir)
Definition: fd.c:567
ReplicationSlotPersistency persistency
Definition: slot.h:50
#define START_CRIT_SECTION()
Definition: miscadmin.h:131
#define SnapBuildOnDiskNotChecksummedSize
Definition: slot.c:80
ReplicationSlotPersistentData data
Definition: slot.h:116
#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:1039
ReplicationSlotPersistentData slotdata
Definition: slot.c:73
#define MAXPGPATH
TransactionId effective_xmin
Definition: slot.h:112
XLogRecPtr candidate_restart_valid
Definition: slot.h:133
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:66
#define InvalidTransactionId
Definition: transam.h:31
unsigned int uint32
Definition: c.h:268
static void pgstat_report_wait_end(void)
Definition: pgstat.h:1235
TransactionId xmin
Definition: slot.h:58
#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:92
TransactionId effective_catalog_xmin
Definition: slot.h:113
#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:1211
#define SnapBuildOnDiskChecksummedSize
Definition: slot.c:83
const char * name
Definition: encode.c:521
XLogRecPtr candidate_xmin_lsn
Definition: slot.h:132
int errmsg(const char *fmt,...)
Definition: elog.c:797
pid_t active_pid
Definition: slot.h:95
int i
int pg_fsync(int fd)
Definition: fd.c:333
ReplicationSlot replication_slots[1]
Definition: slot.h:149
#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:13
XLogRecPtr candidate_restart_lsn
Definition: slot.h:134
static void SaveSlotToPath ( ReplicationSlot slot,
const char *  path,
int  elevel 
)
static

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

1211 {
1212  char tmppath[MAXPGPATH];
1213  char path[MAXPGPATH];
1214  int fd;
1216  bool was_dirty;
1217 
1218  /* first check whether there's something to write out */
1219  SpinLockAcquire(&slot->mutex);
1220  was_dirty = slot->dirty;
1221  slot->just_dirtied = false;
1222  SpinLockRelease(&slot->mutex);
1223 
1224  /* and don't do anything if there's nothing to write */
1225  if (!was_dirty)
1226  return;
1227 
1229 
1230  /* silence valgrind :( */
1231  memset(&cp, 0, sizeof(ReplicationSlotOnDisk));
1232 
1233  sprintf(tmppath, "%s/state.tmp", dir);
1234  sprintf(path, "%s/state", dir);
1235 
1236  fd = OpenTransientFile(tmppath,
1237  O_CREAT | O_EXCL | O_WRONLY | PG_BINARY,
1238  S_IRUSR | S_IWUSR);
1239  if (fd < 0)
1240  {
1241  ereport(elevel,
1243  errmsg("could not create file \"%s\": %m",
1244  tmppath)));
1245  return;
1246  }
1247 
1248  cp.magic = SLOT_MAGIC;
1249  INIT_CRC32C(cp.checksum);
1250  cp.version = SLOT_VERSION;
1252 
1253  SpinLockAcquire(&slot->mutex);
1254 
1255  memcpy(&cp.slotdata, &slot->data, sizeof(ReplicationSlotPersistentData));
1256 
1257  SpinLockRelease(&slot->mutex);
1258 
1259  COMP_CRC32C(cp.checksum,
1260  (char *) (&cp) + SnapBuildOnDiskNotChecksummedSize,
1262  FIN_CRC32C(cp.checksum);
1263 
1265  if ((write(fd, &cp, sizeof(cp))) != sizeof(cp))
1266  {
1267  int save_errno = errno;
1268 
1270  CloseTransientFile(fd);
1271  errno = save_errno;
1272  ereport(elevel,
1274  errmsg("could not write to file \"%s\": %m",
1275  tmppath)));
1276  return;
1277  }
1279 
1280  /* fsync the temporary file */
1282  if (pg_fsync(fd) != 0)
1283  {
1284  int save_errno = errno;
1285 
1287  CloseTransientFile(fd);
1288  errno = save_errno;
1289  ereport(elevel,
1291  errmsg("could not fsync file \"%s\": %m",
1292  tmppath)));
1293  return;
1294  }
1296 
1297  CloseTransientFile(fd);
1298 
1299  /* rename to permanent file, fsync file and directory */
1300  if (rename(tmppath, path) != 0)
1301  {
1302  ereport(elevel,
1304  errmsg("could not rename file \"%s\" to \"%s\": %m",
1305  tmppath, path)));
1306  return;
1307  }
1308 
1309  /* Check CreateSlot() for the reasoning of using a crit. section. */
1311 
1312  fsync_fname(path, false);
1313  fsync_fname(dir, true);
1314  fsync_fname("pg_replslot", true);
1315 
1316  END_CRIT_SECTION();
1317 
1318  /*
1319  * Successfully wrote, unset dirty bit, unless somebody dirtied again
1320  * already.
1321  */
1322  SpinLockAcquire(&slot->mutex);
1323  if (!slot->just_dirtied)
1324  slot->dirty = false;
1325  SpinLockRelease(&slot->mutex);
1326 
1328 }
#define INIT_CRC32C(crc)
Definition: pg_crc32c.h:41
#define write(a, b, c)
Definition: win32.h:14
#define SLOT_MAGIC
Definition: slot.c:89
#define END_CRIT_SECTION()
Definition: miscadmin.h:133
void fsync_fname(const char *fname, bool isdir)
Definition: fd.c:567
#define START_CRIT_SECTION()
Definition: miscadmin.h:131
#define SnapBuildOnDiskNotChecksummedSize
Definition: slot.c:80
ReplicationSlotPersistentData data
Definition: slot.h:116
static int fd(const char *x, int i)
Definition: preproc-init.c:105
#define PG_BINARY
Definition: c.h:1039
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:1235
#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:136
#define SpinLockRelease(lock)
Definition: spin.h:64
bool just_dirtied
Definition: slot.h:98
#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:1211
#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:89
#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:99
LWLock io_in_progress_lock
Definition: slot.h:119
void StartupReplicationSlots ( void  )

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

1091 {
1092  DIR *replication_dir;
1093  struct dirent *replication_de;
1094 
1095  elog(DEBUG1, "starting up replication slots");
1096 
1097  /* restore all slots by iterating over all on-disk entries */
1098  replication_dir = AllocateDir("pg_replslot");
1099  while ((replication_de = ReadDir(replication_dir, "pg_replslot")) != NULL)
1100  {
1101  struct stat statbuf;
1102  char path[MAXPGPATH + 12];
1103 
1104  if (strcmp(replication_de->d_name, ".") == 0 ||
1105  strcmp(replication_de->d_name, "..") == 0)
1106  continue;
1107 
1108  snprintf(path, sizeof(path), "pg_replslot/%s", replication_de->d_name);
1109 
1110  /* we're only creating directories here, skip if it's not our's */
1111  if (lstat(path, &statbuf) == 0 && !S_ISDIR(statbuf.st_mode))
1112  continue;
1113 
1114  /* we crashed while a slot was being setup or deleted, clean up */
1115  if (pg_str_endswith(replication_de->d_name, ".tmp"))
1116  {
1117  if (!rmtree(path, true))
1118  {
1119  ereport(WARNING,
1121  errmsg("could not remove directory \"%s\"", path)));
1122  continue;
1123  }
1124  fsync_fname("pg_replslot", true);
1125  continue;
1126  }
1127 
1128  /* looks like a slot in a normal state, restore */
1129  RestoreSlotFromDisk(replication_de->d_name);
1130  }
1131  FreeDir(replication_dir);
1132 
1133  /* currently no slots exist, we're done. */
1134  if (max_replication_slots <= 0)
1135  return;
1136 
1137  /* Now that we have recovered all the data, compute replication xmin */
1140 }
static void RestoreSlotFromDisk(const char *name)
Definition: slot.c:1334
#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:741
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:262
void ReplicationSlotsComputeRequiredXmin(bool already_locked)
Definition: slot.c:695
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().