PostgreSQL Source Code  git master
slot.h File Reference
#include "access/xlog.h"
#include "access/xlogreader.h"
#include "storage/condition_variable.h"
#include "storage/lwlock.h"
#include "storage/shmem.h"
#include "storage/spin.h"
Include dependency graph for slot.h:
This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Data Structures

struct  ReplicationSlotPersistentData
 
struct  ReplicationSlot
 
struct  ReplicationSlotCtlData
 

Macros

#define SlotIsPhysical(slot)   ((slot)->data.database == InvalidOid)
 
#define SlotIsLogical(slot)   ((slot)->data.database != InvalidOid)
 

Typedefs

typedef enum ReplicationSlotPersistency ReplicationSlotPersistency
 
typedef enum SlotAcquireBehavior SlotAcquireBehavior
 
typedef struct ReplicationSlotPersistentData ReplicationSlotPersistentData
 
typedef struct ReplicationSlot ReplicationSlot
 
typedef struct ReplicationSlotCtlData ReplicationSlotCtlData
 

Enumerations

enum  ReplicationSlotPersistency { RS_PERSISTENT, RS_EPHEMERAL, RS_TEMPORARY }
 
enum  SlotAcquireBehavior { SAB_Error, SAB_Block, SAB_Inquire }
 

Functions

Size ReplicationSlotsShmemSize (void)
 
void ReplicationSlotsShmemInit (void)
 
void ReplicationSlotCreate (const char *name, bool db_specific, ReplicationSlotPersistency p)
 
void ReplicationSlotPersist (void)
 
void ReplicationSlotDrop (const char *name, bool nowait)
 
int ReplicationSlotAcquire (const char *name, SlotAcquireBehavior behavior)
 
void ReplicationSlotRelease (void)
 
void ReplicationSlotCleanup (void)
 
void ReplicationSlotSave (void)
 
void ReplicationSlotMarkDirty (void)
 
bool ReplicationSlotValidateName (const char *name, int elevel)
 
void ReplicationSlotReserveWal (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 InvalidateObsoleteReplicationSlots (XLogSegNo oldestSegno)
 
ReplicationSlotSearchNamedReplicationSlot (const char *name)
 
void StartupReplicationSlots (void)
 
void CheckPointReplicationSlots (void)
 
void CheckSlotRequirements (void)
 

Variables

PGDLLIMPORT ReplicationSlotCtlDataReplicationSlotCtl
 
PGDLLIMPORT ReplicationSlotMyReplicationSlot
 
PGDLLIMPORT int max_replication_slots
 

Macro Definition Documentation

◆ SlotIsLogical

◆ SlotIsPhysical

#define SlotIsPhysical (   slot)    ((slot)->data.database == InvalidOid)

Typedef Documentation

◆ ReplicationSlot

◆ ReplicationSlotCtlData

◆ ReplicationSlotPersistency

◆ ReplicationSlotPersistentData

◆ SlotAcquireBehavior

Enumeration Type Documentation

◆ ReplicationSlotPersistency

Enumerator
RS_PERSISTENT 
RS_EPHEMERAL 
RS_TEMPORARY 

Definition at line 32 of file slot.h.

33 {
ReplicationSlotPersistency
Definition: slot.h:32

◆ SlotAcquireBehavior

Enumerator
SAB_Error 
SAB_Block 
SAB_Inquire 

Definition at line 40 of file slot.h.

41 {
42  SAB_Error,
43  SAB_Block,
Definition: slot.h:43
SlotAcquireBehavior
Definition: slot.h:40
Definition: slot.h:42

Function Documentation

◆ CheckPointReplicationSlots()

void CheckPointReplicationSlots ( void  )

Definition at line 1271 of file slot.c.

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

Referenced by CheckPointGuts().

1272 {
1273  int i;
1274 
1275  elog(DEBUG1, "performing replication slot checkpoint");
1276 
1277  /*
1278  * Prevent any slot from being created/dropped while we're active. As we
1279  * explicitly do *not* want to block iterating over replication_slots or
1280  * acquiring a slot we cannot take the control lock - but that's OK,
1281  * because holding ReplicationSlotAllocationLock is strictly stronger, and
1282  * enough to guarantee that nobody can change the in_use bits on us.
1283  */
1284  LWLockAcquire(ReplicationSlotAllocationLock, LW_SHARED);
1285 
1286  for (i = 0; i < max_replication_slots; i++)
1287  {
1289  char path[MAXPGPATH];
1290 
1291  if (!s->in_use)
1292  continue;
1293 
1294  /* save the slot to disk, locking is handled in SaveSlotToPath() */
1295  sprintf(path, "pg_replslot/%s", NameStr(s->data.name));
1296  SaveSlotToPath(s, path, LOG);
1297  }
1298  LWLockRelease(ReplicationSlotAllocationLock);
1299 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
#define DEBUG1
Definition: elog.h:25
ReplicationSlotPersistentData data
Definition: slot.h:143
#define LOG
Definition: elog.h:26
static void SaveSlotToPath(ReplicationSlot *slot, const char *path, int elevel)
Definition: slot.c:1427
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1812
#define sprintf
Definition: port.h:217
#define MAXPGPATH
bool in_use
Definition: slot.h:119
int max_replication_slots
Definition: slot.c:99
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1208
#define elog(elevel,...)
Definition: elog.h:214
int i
#define NameStr(name)
Definition: c.h:623
ReplicationSlot replication_slots[1]
Definition: slot.h:176

◆ CheckSlotRequirements()

void CheckSlotRequirements ( void  )

Definition at line 1053 of file slot.c.

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

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

1054 {
1055  /*
1056  * NB: Adding a new requirement likely means that RestoreSlotFromDisk()
1057  * needs the same check.
1058  */
1059 
1060  if (max_replication_slots == 0)
1061  ereport(ERROR,
1062  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1063  errmsg("replication slots can only be used if max_replication_slots > 0")));
1064 
1066  ereport(ERROR,
1067  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1068  errmsg("replication slots can only be used if wal_level >= replica")));
1069 }
int wal_level
Definition: xlog.c:107
int errcode(int sqlerrcode)
Definition: elog.c:610
#define ERROR
Definition: elog.h:43
#define ereport(elevel,...)
Definition: elog.h:144
int max_replication_slots
Definition: slot.c:99
int errmsg(const char *fmt,...)
Definition: elog.c:821

◆ InvalidateObsoleteReplicationSlots()

void InvalidateObsoleteReplicationSlots ( XLogSegNo  oldestSegno)

Definition at line 1155 of file slot.c.

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

Referenced by CreateCheckPoint(), and CreateRestartPoint().

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

◆ ReplicationSlotAcquire()

int ReplicationSlotAcquire ( const char *  name,
SlotAcquireBehavior  behavior 
)

Definition at line 375 of file slot.c.

References ReplicationSlotAcquireInternal().

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

376 {
377  return ReplicationSlotAcquireInternal(NULL, name, behavior);
378 }
static int ReplicationSlotAcquireInternal(ReplicationSlot *slot, const char *name, SlotAcquireBehavior behavior)
Definition: slot.c:389
const char * name
Definition: encode.c:561

◆ ReplicationSlotCleanup()

void ReplicationSlotCleanup ( void  )

Definition at line 540 of file slot.c.

References ReplicationSlot::active_cv, ReplicationSlot::active_pid, Assert, ConditionVariableBroadcast(), ReplicationSlot::data, i, ReplicationSlot::in_use, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_replication_slots, ReplicationSlot::mutex, MyProcPid, ReplicationSlotPersistentData::persistency, ReplicationSlotCtlData::replication_slots, ReplicationSlotDropPtr(), RS_TEMPORARY, SpinLockAcquire, and SpinLockRelease.

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

541 {
542  int i;
543 
544  Assert(MyReplicationSlot == NULL);
545 
546 restart:
547  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
548  for (i = 0; i < max_replication_slots; i++)
549  {
551 
552  if (!s->in_use)
553  continue;
554 
555  SpinLockAcquire(&s->mutex);
556  if (s->active_pid == MyProcPid)
557  {
559  SpinLockRelease(&s->mutex);
560  LWLockRelease(ReplicationSlotControlLock); /* avoid deadlock */
561 
563 
565  goto restart;
566  }
567  else
568  SpinLockRelease(&s->mutex);
569  }
570 
571  LWLockRelease(ReplicationSlotControlLock);
572 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
int MyProcPid
Definition: globals.c:40
ReplicationSlotPersistency persistency
Definition: slot.h:61
void ConditionVariableBroadcast(ConditionVariable *cv)
static void ReplicationSlotDropPtr(ReplicationSlot *slot)
Definition: slot.c:608
ReplicationSlotPersistentData data
Definition: slot.h:143
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1812
#define SpinLockAcquire(lock)
Definition: spin.h:62
bool in_use
Definition: slot.h:119
#define SpinLockRelease(lock)
Definition: spin.h:64
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
int max_replication_slots
Definition: slot.c:99
#define Assert(condition)
Definition: c.h:746
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1208
ConditionVariable active_cv
Definition: slot.h:149
pid_t active_pid
Definition: slot.h:122
int i
ReplicationSlot replication_slots[1]
Definition: slot.h:176
slock_t mutex
Definition: slot.h:116

◆ ReplicationSlotCreate()

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

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, NameStr, namestrcpy(), ReplicationSlotPersistentData::persistency, pgstat_report_replslot(), ReplicationSlotCtlData::replication_slots, ReplicationSlotValidateName(), SlotIsLogical, SpinLockAcquire, and SpinLockRelease.

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

223 {
224  ReplicationSlot *slot = NULL;
225  int i;
226 
227  Assert(MyReplicationSlot == NULL);
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));
277  namestrcpy(&slot->data.name, name);
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  * ReplicationSlotAcquire() 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  * Create statistics entry for the new logical slot. We don't collect any
318  * stats for physical slots, so no need to create an entry for the same.
319  * See ReplicationSlotDropPtr for why we need to do this before releasing
320  * ReplicationSlotAllocationLock.
321  */
322  if (SlotIsLogical(slot))
323  pgstat_report_replslot(NameStr(slot->data.name), 0, 0, 0, 0, 0, 0);
324 
325  /*
326  * Now that the slot has been marked as in_use and active, it's safe to
327  * let somebody else try to allocate a slot.
328  */
329  LWLockRelease(ReplicationSlotAllocationLock);
330 
331  /* Let everybody know we've modified this slot */
333 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
TransactionId candidate_catalog_xmin
Definition: slot.h:158
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
int MyProcPid
Definition: globals.c:40
int errhint(const char *fmt,...)
Definition: elog.c:1068
void namestrcpy(Name name, const char *str)
Definition: name.c:233
ReplicationSlotPersistency persistency
Definition: slot.h:61
void ConditionVariableBroadcast(ConditionVariable *cv)
int errcode(int sqlerrcode)
Definition: elog.c:610
ReplicationSlotPersistentData data
Definition: slot.h:143
static void CreateSlotOnDisk(ReplicationSlot *slot)
Definition: slot.c:1366
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1812
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:139
XLogRecPtr candidate_restart_valid
Definition: slot.h:160
void pgstat_report_replslot(const char *slotname, int spilltxns, int spillcount, int spillbytes, int streamtxns, int streamcount, int streambytes)
Definition: pgstat.c:1710
#define InvalidTransactionId
Definition: transam.h:31
#define SlotIsLogical(slot)
Definition: slot.h:165
bool in_use
Definition: slot.h:119
#define SpinLockRelease(lock)
Definition: spin.h:64
bool just_dirtied
Definition: slot.h:125
TransactionId effective_catalog_xmin
Definition: slot.h:140
Oid MyDatabaseId
Definition: globals.c:85
#define InvalidOid
Definition: postgres_ext.h:36
#define ereport(elevel,...)
Definition: elog.h:144
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
int max_replication_slots
Definition: slot.c:99
#define Assert(condition)
Definition: c.h:746
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1208
ConditionVariable active_cv
Definition: slot.h:149
const char * name
Definition: encode.c:561
XLogRecPtr candidate_xmin_lsn
Definition: slot.h:159
int errmsg(const char *fmt,...)
Definition: elog.c:821
pid_t active_pid
Definition: slot.h:122
int i
#define NameStr(name)
Definition: c.h:623
ReplicationSlot replication_slots[1]
Definition: slot.h:176
slock_t mutex
Definition: slot.h:116
#define ERRCODE_DUPLICATE_OBJECT
Definition: streamutil.c:32
bool dirty
Definition: slot.h:126
XLogRecPtr candidate_restart_lsn
Definition: slot.h:161

◆ ReplicationSlotDrop()

void ReplicationSlotDrop ( const char *  name,
bool  nowait 
)

Definition at line 578 of file slot.c.

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

Referenced by DropReplicationSlot(), and pg_drop_replication_slot().

579 {
580  Assert(MyReplicationSlot == NULL);
581 
582  (void) ReplicationSlotAcquire(name, nowait ? SAB_Error : SAB_Block);
583 
585 }
static void ReplicationSlotDropAcquired(void)
Definition: slot.c:591
Definition: slot.h:43
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
#define Assert(condition)
Definition: c.h:746
const char * name
Definition: encode.c:561
int ReplicationSlotAcquire(const char *name, SlotAcquireBehavior behavior)
Definition: slot.c:375
Definition: slot.h:42

◆ ReplicationSlotMarkDirty()

◆ ReplicationSlotPersist()

void ReplicationSlotPersist ( void  )

Definition at line 754 of file slot.c.

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

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

755 {
757 
758  Assert(slot != NULL);
760 
761  SpinLockAcquire(&slot->mutex);
763  SpinLockRelease(&slot->mutex);
764 
767 }
ReplicationSlotPersistency persistency
Definition: slot.h:61
void ReplicationSlotSave(void)
Definition: slot.c:719
ReplicationSlotPersistentData data
Definition: slot.h:143
#define SpinLockAcquire(lock)
Definition: spin.h:62
#define SpinLockRelease(lock)
Definition: spin.h:64
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
#define Assert(condition)
Definition: c.h:746
slock_t mutex
Definition: slot.h:116
void ReplicationSlotMarkDirty(void)
Definition: slot.c:737

◆ ReplicationSlotRelease()

void ReplicationSlotRelease ( void  )

Definition at line 484 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, MyProc, MyReplicationSlot, ReplicationSlotPersistentData::persistency, PGPROC::pgxactoff, PROC_IN_LOGICAL_DECODING, ProcGlobal, ReplicationSlotDropAcquired(), ReplicationSlotsComputeRequiredXmin(), RS_EPHEMERAL, RS_PERSISTENT, SpinLockAcquire, SpinLockRelease, TransactionIdIsValid, PGPROC::vacuumFlags, PROC_HDR::vacuumFlags, and ReplicationSlotPersistentData::xmin.

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

485 {
487 
488  Assert(slot != NULL && slot->active_pid != 0);
489 
490  if (slot->data.persistency == RS_EPHEMERAL)
491  {
492  /*
493  * Delete the slot. There is no !PANIC case where this is allowed to
494  * fail, all that may happen is an incomplete cleanup of the on-disk
495  * data.
496  */
498  }
499 
500  /*
501  * If slot needed to temporarily restrain both data and catalog xmin to
502  * create the catalog snapshot, remove that temporary constraint.
503  * Snapshots can only be exported while the initial snapshot is still
504  * acquired.
505  */
506  if (!TransactionIdIsValid(slot->data.xmin) &&
508  {
509  SpinLockAcquire(&slot->mutex);
511  SpinLockRelease(&slot->mutex);
513  }
514 
515  if (slot->data.persistency == RS_PERSISTENT)
516  {
517  /*
518  * Mark persistent slot inactive. We're not freeing it, just
519  * disconnecting, but wake up others that may be waiting for it.
520  */
521  SpinLockAcquire(&slot->mutex);
522  slot->active_pid = 0;
523  SpinLockRelease(&slot->mutex);
525  }
526 
527  MyReplicationSlot = NULL;
528 
529  /* might not have been set when we've been a plain slot */
530  LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
533  LWLockRelease(ProcArrayLock);
534 }
#define PROC_IN_LOGICAL_DECODING
Definition: proc.h:57
PGPROC * MyProc
Definition: proc.c:67
ReplicationSlotPersistency persistency
Definition: slot.h:61
void ConditionVariableBroadcast(ConditionVariable *cv)
PROC_HDR * ProcGlobal
Definition: proc.c:79
ReplicationSlotPersistentData data
Definition: slot.h:143
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1812
#define SpinLockAcquire(lock)
Definition: spin.h:62
static void ReplicationSlotDropAcquired(void)
Definition: slot.c:591
TransactionId effective_xmin
Definition: slot.h:139
#define InvalidTransactionId
Definition: transam.h:31
TransactionId xmin
Definition: slot.h:69
#define SpinLockRelease(lock)
Definition: spin.h:64
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
#define Assert(condition)
Definition: c.h:746
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1208
ConditionVariable active_cv
Definition: slot.h:149
pid_t active_pid
Definition: slot.h:122
int pgxactoff
Definition: proc.h:139
uint8 * vacuumFlags
Definition: proc.h:321
uint8 vacuumFlags
Definition: proc.h:178
slock_t mutex
Definition: slot.h:116
#define TransactionIdIsValid(xid)
Definition: transam.h:41
void ReplicationSlotsComputeRequiredXmin(bool already_locked)
Definition: slot.c:776

◆ ReplicationSlotReserveWal()

void ReplicationSlotReserveWal ( void  )

Definition at line 1078 of file slot.c.

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

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

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

◆ ReplicationSlotSave()

void ReplicationSlotSave ( void  )

Definition at line 719 of file slot.c.

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

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

720 {
721  char path[MAXPGPATH];
722 
723  Assert(MyReplicationSlot != NULL);
724 
725  sprintf(path, "pg_replslot/%s", NameStr(MyReplicationSlot->data.name));
727 }
ReplicationSlotPersistentData data
Definition: slot.h:143
static void SaveSlotToPath(ReplicationSlot *slot, const char *path, int elevel)
Definition: slot.c:1427
#define sprintf
Definition: port.h:217
#define ERROR
Definition: elog.h:43
#define MAXPGPATH
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
#define Assert(condition)
Definition: c.h:746
#define NameStr(name)
Definition: c.h:623

◆ ReplicationSlotsComputeLogicalRestartLSN()

XLogRecPtr ReplicationSlotsComputeLogicalRestartLSN ( void  )

Definition at line 869 of file slot.c.

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

Referenced by CheckPointLogicalRewriteHeap(), and CheckPointSnapBuild().

870 {
871  XLogRecPtr result = InvalidXLogRecPtr;
872  int i;
873 
874  if (max_replication_slots <= 0)
875  return InvalidXLogRecPtr;
876 
877  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
878 
879  for (i = 0; i < max_replication_slots; i++)
880  {
881  ReplicationSlot *s;
882  XLogRecPtr restart_lsn;
883 
885 
886  /* cannot change while ReplicationSlotCtlLock is held */
887  if (!s->in_use)
888  continue;
889 
890  /* we're only interested in logical slots */
891  if (!SlotIsLogical(s))
892  continue;
893 
894  /* read once, it's ok if it increases while we're checking */
895  SpinLockAcquire(&s->mutex);
896  restart_lsn = s->data.restart_lsn;
897  SpinLockRelease(&s->mutex);
898 
899  if (restart_lsn == InvalidXLogRecPtr)
900  continue;
901 
902  if (result == InvalidXLogRecPtr ||
903  restart_lsn < result)
904  result = restart_lsn;
905  }
906 
907  LWLockRelease(ReplicationSlotControlLock);
908 
909  return result;
910 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
ReplicationSlotPersistentData data
Definition: slot.h:143
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1812
#define SpinLockAcquire(lock)
Definition: spin.h:62
#define SlotIsLogical(slot)
Definition: slot.h:165
bool in_use
Definition: slot.h:119
#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:80
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1208
int i
ReplicationSlot replication_slots[1]
Definition: slot.h:176
slock_t mutex
Definition: slot.h:116

◆ ReplicationSlotsComputeRequiredLSN()

void ReplicationSlotsComputeRequiredLSN ( void  )

Definition at line 826 of file slot.c.

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

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

827 {
828  int i;
829  XLogRecPtr min_required = InvalidXLogRecPtr;
830 
831  Assert(ReplicationSlotCtl != NULL);
832 
833  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
834  for (i = 0; i < max_replication_slots; i++)
835  {
837  XLogRecPtr restart_lsn;
838 
839  if (!s->in_use)
840  continue;
841 
842  SpinLockAcquire(&s->mutex);
843  restart_lsn = s->data.restart_lsn;
844  SpinLockRelease(&s->mutex);
845 
846  if (restart_lsn != InvalidXLogRecPtr &&
847  (min_required == InvalidXLogRecPtr ||
848  restart_lsn < min_required))
849  min_required = restart_lsn;
850  }
851  LWLockRelease(ReplicationSlotControlLock);
852 
853  XLogSetReplicationSlotMinimumLSN(min_required);
854 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
ReplicationSlotPersistentData data
Definition: slot.h:143
void XLogSetReplicationSlotMinimumLSN(XLogRecPtr lsn)
Definition: xlog.c:2726
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1812
#define SpinLockAcquire(lock)
Definition: spin.h:62
bool in_use
Definition: slot.h:119
#define SpinLockRelease(lock)
Definition: spin.h:64
int max_replication_slots
Definition: slot.c:99
uint64 XLogRecPtr
Definition: xlogdefs.h:21
#define Assert(condition)
Definition: c.h:746
XLogRecPtr restart_lsn
Definition: slot.h:80
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1208
int i
ReplicationSlot replication_slots[1]
Definition: slot.h:176
slock_t mutex
Definition: slot.h:116

◆ ReplicationSlotsComputeRequiredXmin()

void ReplicationSlotsComputeRequiredXmin ( bool  already_locked)

Definition at line 776 of file slot.c.

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

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

777 {
778  int i;
780  TransactionId agg_catalog_xmin = InvalidTransactionId;
781 
782  Assert(ReplicationSlotCtl != NULL);
783 
784  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
785 
786  for (i = 0; i < max_replication_slots; i++)
787  {
789  TransactionId effective_xmin;
790  TransactionId effective_catalog_xmin;
791 
792  if (!s->in_use)
793  continue;
794 
795  SpinLockAcquire(&s->mutex);
796  effective_xmin = s->effective_xmin;
797  effective_catalog_xmin = s->effective_catalog_xmin;
798  SpinLockRelease(&s->mutex);
799 
800  /* check the data xmin */
801  if (TransactionIdIsValid(effective_xmin) &&
802  (!TransactionIdIsValid(agg_xmin) ||
803  TransactionIdPrecedes(effective_xmin, agg_xmin)))
804  agg_xmin = effective_xmin;
805 
806  /* check the catalog xmin */
807  if (TransactionIdIsValid(effective_catalog_xmin) &&
808  (!TransactionIdIsValid(agg_catalog_xmin) ||
809  TransactionIdPrecedes(effective_catalog_xmin, agg_catalog_xmin)))
810  agg_catalog_xmin = effective_catalog_xmin;
811  }
812 
813  LWLockRelease(ReplicationSlotControlLock);
814 
815  ProcArraySetReplicationSlotXmin(agg_xmin, agg_catalog_xmin, already_locked);
816 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
uint32 TransactionId
Definition: c.h:521
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1812
#define SpinLockAcquire(lock)
Definition: spin.h:62
TransactionId effective_xmin
Definition: slot.h:139
#define InvalidTransactionId
Definition: transam.h:31
bool TransactionIdPrecedes(TransactionId id1, TransactionId id2)
Definition: transam.c:300
bool in_use
Definition: slot.h:119
#define SpinLockRelease(lock)
Definition: spin.h:64
TransactionId effective_catalog_xmin
Definition: slot.h:140
int max_replication_slots
Definition: slot.c:99
#define Assert(condition)
Definition: c.h:746
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1208
int i
void ProcArraySetReplicationSlotXmin(TransactionId xmin, TransactionId catalog_xmin, bool already_locked)
Definition: procarray.c:3760
ReplicationSlot replication_slots[1]
Definition: slot.h:176
slock_t mutex
Definition: slot.h:116
#define TransactionIdIsValid(xid)
Definition: transam.h:41

◆ ReplicationSlotsCountDBSlots()

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

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

922 {
923  int i;
924 
925  *nslots = *nactive = 0;
926 
927  if (max_replication_slots <= 0)
928  return false;
929 
930  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
931  for (i = 0; i < max_replication_slots; i++)
932  {
933  ReplicationSlot *s;
934 
936 
937  /* cannot change while ReplicationSlotCtlLock is held */
938  if (!s->in_use)
939  continue;
940 
941  /* only logical slots are database specific, skip */
942  if (!SlotIsLogical(s))
943  continue;
944 
945  /* not our database, skip */
946  if (s->data.database != dboid)
947  continue;
948 
949  /* count slots with spinlock held */
950  SpinLockAcquire(&s->mutex);
951  (*nslots)++;
952  if (s->active_pid != 0)
953  (*nactive)++;
954  SpinLockRelease(&s->mutex);
955  }
956  LWLockRelease(ReplicationSlotControlLock);
957 
958  if (*nslots > 0)
959  return true;
960  return false;
961 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
ReplicationSlotPersistentData data
Definition: slot.h:143
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1812
#define SpinLockAcquire(lock)
Definition: spin.h:62
#define SlotIsLogical(slot)
Definition: slot.h:165
bool in_use
Definition: slot.h:119
#define SpinLockRelease(lock)
Definition: spin.h:64
int max_replication_slots
Definition: slot.c:99
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1208
pid_t active_pid
Definition: slot.h:122
int i
ReplicationSlot replication_slots[1]
Definition: slot.h:176
slock_t mutex
Definition: slot.h:116

◆ ReplicationSlotsDropDBSlots()

void ReplicationSlotsDropDBSlots ( Oid  dboid)

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

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

◆ ReplicationSlotsShmemInit()

void ReplicationSlotsShmemInit ( void  )

Definition at line 134 of file slot.c.

References ReplicationSlot::active_cv, ConditionVariableInit(), i, ReplicationSlot::io_in_progress_lock, LWLockInitialize(), LWTRANCHE_REPLICATION_SLOT_IO, max_replication_slots, MemSet, ReplicationSlot::mutex, ReplicationSlotCtlData::replication_slots, ReplicationSlotsShmemSize(), ShmemInitStruct(), and SpinLockInit.

Referenced by CreateSharedMemoryAndSemaphores().

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

◆ ReplicationSlotsShmemSize()

Size ReplicationSlotsShmemSize ( void  )

Definition at line 116 of file slot.c.

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

Referenced by CreateSharedMemoryAndSemaphores(), and ReplicationSlotsShmemInit().

117 {
118  Size size = 0;
119 
120  if (max_replication_slots == 0)
121  return size;
122 
123  size = offsetof(ReplicationSlotCtlData, replication_slots);
124  size = add_size(size,
126 
127  return size;
128 }
Size mul_size(Size s1, Size s2)
Definition: shmem.c:515
Size add_size(Size s1, Size s2)
Definition: shmem.c:498
int max_replication_slots
Definition: slot.c:99
size_t Size
Definition: c.h:474
#define offsetof(type, field)
Definition: c.h:669

◆ ReplicationSlotValidateName()

bool ReplicationSlotValidateName ( const char *  name,
int  elevel 
)

Definition at line 174 of file slot.c.

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

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

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

◆ SearchNamedReplicationSlot()

ReplicationSlot* SearchNamedReplicationSlot ( const char *  name)

Definition at line 343 of file slot.c.

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

Referenced by pgstat_reset_replslot_counter(), and ReplicationSlotAcquireInternal().

344 {
345  int i;
346  ReplicationSlot *slot = NULL;
347 
348  Assert(LWLockHeldByMeInMode(ReplicationSlotControlLock,
349  LW_SHARED));
350 
351  for (i = 0; i < max_replication_slots; i++)
352  {
354 
355  if (s->in_use && strcmp(name, NameStr(s->data.name)) == 0)
356  {
357  slot = s;
358  break;
359  }
360  }
361 
362  return slot;
363 }
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
bool LWLockHeldByMeInMode(LWLock *l, LWLockMode mode)
Definition: lwlock.c:1946
ReplicationSlotPersistentData data
Definition: slot.h:143
bool in_use
Definition: slot.h:119
int max_replication_slots
Definition: slot.c:99
#define Assert(condition)
Definition: c.h:746
const char * name
Definition: encode.c:561
int i
#define NameStr(name)
Definition: c.h:623
ReplicationSlot replication_slots[1]
Definition: slot.h:176

◆ StartupReplicationSlots()

void StartupReplicationSlots ( void  )

Definition at line 1306 of file slot.c.

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

Referenced by StartupXLOG().

1307 {
1308  DIR *replication_dir;
1309  struct dirent *replication_de;
1310 
1311  elog(DEBUG1, "starting up replication slots");
1312 
1313  /* restore all slots by iterating over all on-disk entries */
1314  replication_dir = AllocateDir("pg_replslot");
1315  while ((replication_de = ReadDir(replication_dir, "pg_replslot")) != NULL)
1316  {
1317  struct stat statbuf;
1318  char path[MAXPGPATH + 12];
1319 
1320  if (strcmp(replication_de->d_name, ".") == 0 ||
1321  strcmp(replication_de->d_name, "..") == 0)
1322  continue;
1323 
1324  snprintf(path, sizeof(path), "pg_replslot/%s", replication_de->d_name);
1325 
1326  /* we're only creating directories here, skip if it's not our's */
1327  if (lstat(path, &statbuf) == 0 && !S_ISDIR(statbuf.st_mode))
1328  continue;
1329 
1330  /* we crashed while a slot was being setup or deleted, clean up */
1331  if (pg_str_endswith(replication_de->d_name, ".tmp"))
1332  {
1333  if (!rmtree(path, true))
1334  {
1335  ereport(WARNING,
1336  (errmsg("could not remove directory \"%s\"",
1337  path)));
1338  continue;
1339  }
1340  fsync_fname("pg_replslot", true);
1341  continue;
1342  }
1343 
1344  /* looks like a slot in a normal state, restore */
1345  RestoreSlotFromDisk(replication_de->d_name);
1346  }
1347  FreeDir(replication_dir);
1348 
1349  /* currently no slots exist, we're done. */
1350  if (max_replication_slots <= 0)
1351  return;
1352 
1353  /* Now that we have recovered all the data, compute replication xmin */
1356 }
static void RestoreSlotFromDisk(const char *name)
Definition: slot.c:1581
#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:633
Definition: dirent.h:9
void ReplicationSlotsComputeRequiredLSN(void)
Definition: slot.c:826
Definition: dirent.c:25
#define MAXPGPATH
DIR * AllocateDir(const char *dirname)
Definition: fd.c:2583
#define WARNING
Definition: elog.h:40
bool rmtree(const char *path, bool rmtopdir)
Definition: rmtree.c:42
#define ereport(elevel,...)
Definition: elog.h:144
int max_replication_slots
Definition: slot.c:99
struct dirent * ReadDir(DIR *dir, const char *dirname)
Definition: fd.c:2649
#define S_ISDIR(m)
Definition: win32_port.h:316
#define lstat(path, sb)
Definition: win32_port.h:276
int errmsg(const char *fmt,...)
Definition: elog.c:821
#define elog(elevel,...)
Definition: elog.h:214
char d_name[MAX_PATH]
Definition: dirent.h:15
#define snprintf
Definition: port.h:215
void ReplicationSlotsComputeRequiredXmin(bool already_locked)
Definition: slot.c:776
int FreeDir(DIR *dir)
Definition: fd.c:2701

Variable Documentation

◆ max_replication_slots

◆ MyReplicationSlot

◆ ReplicationSlotCtl

PGDLLIMPORT ReplicationSlotCtlData* ReplicationSlotCtl

Definition at line 93 of file slot.c.

Referenced by copy_replication_slot(), and pg_get_replication_slots().