#include "fmgr.h"
#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 struct ReplicationSlotPersistentData	ReplicationSlotPersistentData

typedef struct ReplicationSlot	ReplicationSlot

typedef struct ReplicationSlotCtlData	ReplicationSlotCtlData

Enumerations
enum	ReplicationSlotPersistency { RS_PERSISTENT, RS_EPHEMERAL, RS_TEMPORARY }

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)

void	ReplicationSlotAcquire (const char *name, bool nowait)

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	StartupReplicationSlots (void)

void	CheckPointReplicationSlots (void)

void	CheckSlotRequirements (void)

Variables
ReplicationSlotCtlData *	ReplicationSlotCtl

PGDLLIMPORT ReplicationSlot *	MyReplicationSlot

PGDLLIMPORT int	max_replication_slots

Macro Definition Documentation

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

Definition at line 142 of file slot.h.

Referenced by ProcessStandbyReplyMessage(), ReplicationSlotReserveWal(), ReplicationSlotsComputeLogicalRestartLSN(), ReplicationSlotsCountDBSlots(), ReplicationSlotsDropDBSlots(), and StartReplication().

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

Definition at line 141 of file slot.h.

Referenced by CreateDecodingContext(), and CreateInitDecodingContext().

Typedef Documentation

typedef struct ReplicationSlot ReplicationSlot

typedef struct ReplicationSlotCtlData ReplicationSlotCtlData

typedef enum ReplicationSlotPersistency ReplicationSlotPersistency

typedef struct ReplicationSlotPersistentData ReplicationSlotPersistentData

Enumeration Type Documentation

enum ReplicationSlotPersistency

Enumerator
RS_PERSISTENT
RS_EPHEMERAL
RS_TEMPORARY

Definition at line 33 of file slot.h.

 {
     RS_PERSISTENT,
     RS_EPHEMERAL,
     RS_TEMPORARY
 } ReplicationSlotPersistency;

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

 {
     int         i;
 
     elog(DEBUG1, "performing replication slot checkpoint");
 
     /*
      * Prevent any slot from being created/dropped while we're active. As we
      * explicitly do *not* want to block iterating over replication_slots or
      * acquiring a slot we cannot take the control lock - but that's OK,
      * because holding ReplicationSlotAllocationLock is strictly stronger, and
      * enough to guarantee that nobody can change the in_use bits on us.
      */
     LWLockAcquire(ReplicationSlotAllocationLock, LW_SHARED);
 
     for (i = 0; i < max_replication_slots; i++)
     {
         ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
         char        path[MAXPGPATH];
 
         if (!s->in_use)
             continue;
 
         /* save the slot to disk, locking is handled in SaveSlotToPath() */
         sprintf(path, "pg_replslot/%s", NameStr(s->data.name));
         SaveSlotToPath(s, path, LOG);
     }
     LWLockRelease(ReplicationSlotAllocationLock);
 }

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

 {
     if (max_replication_slots == 0)
         ereport(ERROR,
                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                  (errmsg("replication slots can only be used if max_replication_slots > 0"))));
 
     if (wal_level < WAL_LEVEL_REPLICA)
         ereport(ERROR,
                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                  errmsg("replication slots can only be used if wal_level >= replica")));
 }

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, ReplicationSlotCtlData::replication_slots, SpinLockAcquire, SpinLockRelease, and WAIT_EVENT_REPLICATION_SLOT_DROP.

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

 {
     ReplicationSlot *slot;
     int         active_pid;
     int         i;
 
 retry:
     Assert(MyReplicationSlot == NULL);
 
     /*
      * Search for the named slot and mark it active if we find it.  If the
      * slot is already active, we exit the loop with active_pid set to the PID
      * of the backend that owns it.
      */
     active_pid = 0;
     slot = NULL;
     LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
     for (i = 0; i < max_replication_slots; i++)
     {
         ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
 
         if (s->in_use && strcmp(name, NameStr(s->data.name)) == 0)
         {
             /*
              * This is the slot we want.  We don't know yet if it's active, so
              * get ready to sleep on it in case it is.  (We may end up not
              * sleeping, but we don't want to do this while holding the
              * spinlock.)
              */
             ConditionVariablePrepareToSleep(&s->active_cv);
 
             SpinLockAcquire(&s->mutex);
 
             active_pid = s->active_pid;
             if (active_pid == 0)
                 active_pid = s->active_pid = MyProcPid;
 
             SpinLockRelease(&s->mutex);
             slot = s;
 
             break;
         }
     }
     LWLockRelease(ReplicationSlotControlLock);
 
     /* If we did not find the slot, error out. */
     if (slot == NULL)
         ereport(ERROR,
                 (errcode(ERRCODE_UNDEFINED_OBJECT),
                  errmsg("replication slot \"%s\" does not exist", name)));
 
     /*
      * If we found the slot but it's already active in another backend, we
      * either error out or retry after a short wait, as caller specified.
      */
     if (active_pid != MyProcPid)
     {
         if (nowait)
             ereport(ERROR,
                     (errcode(ERRCODE_OBJECT_IN_USE),
                      errmsg("replication slot \"%s\" is active for PID %d",
                             name, active_pid)));
 
         /* Wait here until we get signaled, and then restart */
         ConditionVariableSleep(&slot->active_cv,
                                WAIT_EVENT_REPLICATION_SLOT_DROP);
         ConditionVariableCancelSleep();
         goto retry;
     }
     else
         ConditionVariableCancelSleep(); /* no sleep needed after all */
 
     /* Let everybody know we've modified this slot */
     ConditionVariableBroadcast(&slot->active_cv);
 
     /* We made this slot active, so it's ours now. */
     MyReplicationSlot = slot;
 }

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

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

 {
     int         i;
 
     Assert(MyReplicationSlot == NULL);
 
 restart:
     LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
     for (i = 0; i < max_replication_slots; i++)
     {
         ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
 
         if (!s->in_use)
             continue;
 
         SpinLockAcquire(&s->mutex);
         if (s->active_pid == MyProcPid)
         {
             Assert(s->data.persistency == RS_TEMPORARY);
             SpinLockRelease(&s->mutex);
             LWLockRelease(ReplicationSlotControlLock);  /* avoid deadlock */
 
             ReplicationSlotDropPtr(s);
 
             ConditionVariableBroadcast(&s->active_cv);
             goto restart;
         }
         else
             SpinLockRelease(&s->mutex);
     }
 
     LWLockRelease(ReplicationSlotControlLock);
 }

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

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

 {
     ReplicationSlot *slot = NULL;
     int         i;
 
     Assert(MyReplicationSlot == NULL);
 
     ReplicationSlotValidateName(name, ERROR);
 
     /*
      * If some other backend ran this code concurrently with us, we'd likely
      * both allocate the same slot, and that would be bad.  We'd also be at
      * risk of missing a name collision.  Also, we don't want to try to create
      * a new slot while somebody's busy cleaning up an old one, because we
      * might both be monkeying with the same directory.
      */
     LWLockAcquire(ReplicationSlotAllocationLock, LW_EXCLUSIVE);
 
     /*
      * Check for name collision, and identify an allocatable slot.  We need to
      * hold ReplicationSlotControlLock in shared mode for this, so that nobody
      * else can change the in_use flags while we're looking at them.
      */
     LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
     for (i = 0; i < max_replication_slots; i++)
     {
         ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
 
         if (s->in_use && strcmp(name, NameStr(s->data.name)) == 0)
             ereport(ERROR,
                     (errcode(ERRCODE_DUPLICATE_OBJECT),
                      errmsg("replication slot \"%s\" already exists", name)));
         if (!s->in_use && slot == NULL)
             slot = s;
     }
     LWLockRelease(ReplicationSlotControlLock);
 
     /* If all slots are in use, we're out of luck. */
     if (slot == NULL)
         ereport(ERROR,
                 (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
                  errmsg("all replication slots are in use"),
                  errhint("Free one or increase max_replication_slots.")));
 
     /*
      * Since this slot is not in use, nobody should be looking at any part of
      * it other than the in_use field unless they're trying to allocate it.
      * And since we hold ReplicationSlotAllocationLock, nobody except us can
      * be doing that.  So it's safe to initialize the slot.
      */
     Assert(!slot->in_use);
     Assert(slot->active_pid == 0);
 
     /* first initialize persistent data */
     memset(&slot->data, 0, sizeof(ReplicationSlotPersistentData));
     StrNCpy(NameStr(slot->data.name), name, NAMEDATALEN);
     slot->data.database = db_specific ? MyDatabaseId : InvalidOid;
     slot->data.persistency = persistency;
 
     /* and then data only present in shared memory */
     slot->just_dirtied = false;
     slot->dirty = false;
     slot->effective_xmin = InvalidTransactionId;
     slot->effective_catalog_xmin = InvalidTransactionId;
     slot->candidate_catalog_xmin = InvalidTransactionId;
     slot->candidate_xmin_lsn = InvalidXLogRecPtr;
     slot->candidate_restart_valid = InvalidXLogRecPtr;
     slot->candidate_restart_lsn = InvalidXLogRecPtr;
 
     /*
      * Create the slot on disk.  We haven't actually marked the slot allocated
      * yet, so no special cleanup is required if this errors out.
      */
     CreateSlotOnDisk(slot);
 
     /*
      * We need to briefly prevent any other backend from iterating over the
      * slots while we flip the in_use flag. We also need to set the active
      * flag while holding the ControlLock as otherwise a concurrent
      * SlotAcquire() could acquire the slot as well.
      */
     LWLockAcquire(ReplicationSlotControlLock, LW_EXCLUSIVE);
 
     slot->in_use = true;
 
     /* We can now mark the slot active, and that makes it our slot. */
     SpinLockAcquire(&slot->mutex);
     Assert(slot->active_pid == 0);
     slot->active_pid = MyProcPid;
     SpinLockRelease(&slot->mutex);
     MyReplicationSlot = slot;
 
     LWLockRelease(ReplicationSlotControlLock);
 
     /*
      * Now that the slot has been marked as in_use and active, it's safe to
      * let somebody else try to allocate a slot.
      */
     LWLockRelease(ReplicationSlotAllocationLock);
 
     /* Let everybody know we've modified this slot */
     ConditionVariableBroadcast(&slot->active_cv);
 }

void ReplicationSlotDrop	(	const char *	name,
		bool	nowait
	)

Definition at line 509 of file slot.c.

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

Referenced by DropReplicationSlot(), and pg_drop_replication_slot().

 {
     Assert(MyReplicationSlot == NULL);
 
     ReplicationSlotAcquire(name, nowait);
 
     ReplicationSlotDropAcquired();
 }

void ReplicationSlotMarkDirty ( void )

Definition at line 656 of file slot.c.

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

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

 {
     ReplicationSlot *slot = MyReplicationSlot;
 
     Assert(MyReplicationSlot != NULL);
 
     SpinLockAcquire(&slot->mutex);
     MyReplicationSlot->just_dirtied = true;
     MyReplicationSlot->dirty = true;
     SpinLockRelease(&slot->mutex);
 }

void ReplicationSlotPersist ( void )

Definition at line 673 of file slot.c.

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

Referenced by CreateReplicationSlot(), and pg_create_logical_replication_slot().

 {
     ReplicationSlot *slot = MyReplicationSlot;
 
     Assert(slot != NULL);
     Assert(slot->data.persistency != RS_PERSISTENT);
 
     SpinLockAcquire(&slot->mutex);
     slot->data.persistency = RS_PERSISTENT;
     SpinLockRelease(&slot->mutex);
 
     ReplicationSlotMarkDirty();
     ReplicationSlotSave();
 }

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

 {
     ReplicationSlot *slot = MyReplicationSlot;
 
     Assert(slot != NULL && slot->active_pid != 0);
 
     if (slot->data.persistency == RS_EPHEMERAL)
     {
         /*
          * Delete the slot. There is no !PANIC case where this is allowed to
          * fail, all that may happen is an incomplete cleanup of the on-disk
          * data.
          */
         ReplicationSlotDropAcquired();
     }
 
     /*
      * If slot needed to temporarily restrain both data and catalog xmin to
      * create the catalog snapshot, remove that temporary constraint.
      * Snapshots can only be exported while the initial snapshot is still
      * acquired.
      */
     if (!TransactionIdIsValid(slot->data.xmin) &&
         TransactionIdIsValid(slot->effective_xmin))
     {
         SpinLockAcquire(&slot->mutex);
         slot->effective_xmin = InvalidTransactionId;
         SpinLockRelease(&slot->mutex);
         ReplicationSlotsComputeRequiredXmin(false);
     }
 
     if (slot->data.persistency == RS_PERSISTENT)
     {
         /*
          * Mark persistent slot inactive.  We're not freeing it, just
          * disconnecting, but wake up others that may be waiting for it.
          */
         SpinLockAcquire(&slot->mutex);
         slot->active_pid = 0;
         SpinLockRelease(&slot->mutex);
         ConditionVariableBroadcast(&slot->active_cv);
     }
 
     MyReplicationSlot = NULL;
 
     /* might not have been set when we've been a plain slot */
     LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
     MyPgXact->vacuumFlags &= ~PROC_IN_LOGICAL_DECODING;
     LWLockRelease(ProcArrayLock);
 }

void ReplicationSlotReserveWal ( void )

Definition at line 985 of file slot.c.

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

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

 {
     ReplicationSlot *slot = MyReplicationSlot;
 
     Assert(slot != NULL);
     Assert(slot->data.restart_lsn == InvalidXLogRecPtr);
 
     /*
      * The replication slot mechanism is used to prevent removal of required
      * WAL. As there is no interlock between this routine and checkpoints, WAL
      * segments could concurrently be removed when a now stale return value of
      * ReplicationSlotsComputeRequiredLSN() is used. In the unlikely case that
      * this happens we'll just retry.
      */
     while (true)
     {
         XLogSegNo   segno;
 
         /*
          * For logical slots log a standby snapshot and start logical decoding
          * at exactly that position. That allows the slot to start up more
          * quickly.
          *
          * That's not needed (or indeed helpful) for physical slots as they'll
          * start replay at the last logged checkpoint anyway. Instead return
          * the location of the last redo LSN. While that slightly increases
          * the chance that we have to retry, it's where a base backup has to
          * start replay at.
          */
         if (!RecoveryInProgress() && SlotIsLogical(slot))
         {
             XLogRecPtr  flushptr;
 
             /* start at current insert position */
             slot->data.restart_lsn = GetXLogInsertRecPtr();
 
             /* make sure we have enough information to start */
             flushptr = LogStandbySnapshot();
 
             /* and make sure it's fsynced to disk */
             XLogFlush(flushptr);
         }
         else
         {
             slot->data.restart_lsn = GetRedoRecPtr();
         }
 
         /* prevent WAL removal as fast as possible */
         ReplicationSlotsComputeRequiredLSN();
 
         /*
          * If all required WAL is still there, great, otherwise retry. The
          * slot should prevent further removal of WAL, unless there's a
          * concurrent ReplicationSlotsComputeRequiredLSN() after we've written
          * the new restart_lsn above, so normally we should never need to loop
          * more than twice.
          */
         XLByteToSeg(slot->data.restart_lsn, segno);
         if (XLogGetLastRemovedSegno() < segno)
             break;
     }
 }

void ReplicationSlotSave ( void )

Definition at line 638 of file slot.c.

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

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

 {
     char        path[MAXPGPATH];
 
     Assert(MyReplicationSlot != NULL);
 
     sprintf(path, "pg_replslot/%s", NameStr(MyReplicationSlot->data.name));
     SaveSlotToPath(MyReplicationSlot, path, ERROR);
 }

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

 {
     XLogRecPtr  result = InvalidXLogRecPtr;
     int         i;
 
     if (max_replication_slots <= 0)
         return InvalidXLogRecPtr;
 
     LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
 
     for (i = 0; i < max_replication_slots; i++)
     {
         ReplicationSlot *s;
         XLogRecPtr  restart_lsn;
 
         s = &ReplicationSlotCtl->replication_slots[i];
 
         /* cannot change while ReplicationSlotCtlLock is held */
         if (!s->in_use)
             continue;
 
         /* we're only interested in logical slots */
         if (!SlotIsLogical(s))
             continue;
 
         /* read once, it's ok if it increases while we're checking */
         SpinLockAcquire(&s->mutex);
         restart_lsn = s->data.restart_lsn;
         SpinLockRelease(&s->mutex);
 
         if (result == InvalidXLogRecPtr ||
             restart_lsn < result)
             result = restart_lsn;
     }
 
     LWLockRelease(ReplicationSlotControlLock);
 
     return result;
 }

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, ReplicationSlotCtlData::replication_slots, ReplicationSlotPersistentData::restart_lsn, SpinLockAcquire, SpinLockRelease, and XLogSetReplicationSlotMinimumLSN().

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

 {
     int         i;
     XLogRecPtr  min_required = InvalidXLogRecPtr;
 
     Assert(ReplicationSlotCtl != NULL);
 
     LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
     for (i = 0; i < max_replication_slots; i++)
     {
         ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
         XLogRecPtr  restart_lsn;
 
         if (!s->in_use)
             continue;
 
         SpinLockAcquire(&s->mutex);
         restart_lsn = s->data.restart_lsn;
         SpinLockRelease(&s->mutex);
 
         if (restart_lsn != InvalidXLogRecPtr &&
             (min_required == InvalidXLogRecPtr ||
              restart_lsn < min_required))
             min_required = restart_lsn;
     }
     LWLockRelease(ReplicationSlotControlLock);
 
     XLogSetReplicationSlotMinimumLSN(min_required);
 }

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

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

 {
     int         i;
     TransactionId agg_xmin = InvalidTransactionId;
     TransactionId agg_catalog_xmin = InvalidTransactionId;
 
     Assert(ReplicationSlotCtl != NULL);
 
     LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
 
     for (i = 0; i < max_replication_slots; i++)
     {
         ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
         TransactionId effective_xmin;
         TransactionId effective_catalog_xmin;
 
         if (!s->in_use)
             continue;
 
         SpinLockAcquire(&s->mutex);
         effective_xmin = s->effective_xmin;
         effective_catalog_xmin = s->effective_catalog_xmin;
         SpinLockRelease(&s->mutex);
 
         /* check the data xmin */
         if (TransactionIdIsValid(effective_xmin) &&
             (!TransactionIdIsValid(agg_xmin) ||
              TransactionIdPrecedes(effective_xmin, agg_xmin)))
             agg_xmin = effective_xmin;
 
         /* check the catalog xmin */
         if (TransactionIdIsValid(effective_catalog_xmin) &&
             (!TransactionIdIsValid(agg_catalog_xmin) ||
              TransactionIdPrecedes(effective_catalog_xmin, agg_catalog_xmin)))
             agg_catalog_xmin = effective_catalog_xmin;
     }
 
     LWLockRelease(ReplicationSlotControlLock);
 
     ProcArraySetReplicationSlotXmin(agg_xmin, agg_catalog_xmin, already_locked);
 }

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

 {
     int         i;
 
     *nslots = *nactive = 0;
 
     if (max_replication_slots <= 0)
         return false;
 
     LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
     for (i = 0; i < max_replication_slots; i++)
     {
         ReplicationSlot *s;
 
         s = &ReplicationSlotCtl->replication_slots[i];
 
         /* cannot change while ReplicationSlotCtlLock is held */
         if (!s->in_use)
             continue;
 
         /* only logical slots are database specific, skip */
         if (!SlotIsLogical(s))
             continue;
 
         /* not our database, skip */
         if (s->data.database != dboid)
             continue;
 
         /* count slots with spinlock held */
         SpinLockAcquire(&s->mutex);
         (*nslots)++;
         if (s->active_pid != 0)
             (*nactive)++;
         SpinLockRelease(&s->mutex);
     }
     LWLockRelease(ReplicationSlotControlLock);
 
     if (*nslots > 0)
         return true;
     return false;
 }

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

 {
     int         i;
 
     if (max_replication_slots <= 0)
         return;
 
 restart:
     LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
     for (i = 0; i < max_replication_slots; i++)
     {
         ReplicationSlot *s;
         char       *slotname;
         int         active_pid;
 
         s = &ReplicationSlotCtl->replication_slots[i];
 
         /* cannot change while ReplicationSlotCtlLock is held */
         if (!s->in_use)
             continue;
 
         /* only logical slots are database specific, skip */
         if (!SlotIsLogical(s))
             continue;
 
         /* not our database, skip */
         if (s->data.database != dboid)
             continue;
 
         /* acquire slot, so ReplicationSlotDropAcquired can be reused  */
         SpinLockAcquire(&s->mutex);
         /* can't change while ReplicationSlotControlLock is held */
         slotname = NameStr(s->data.name);
         active_pid = s->active_pid;
         if (active_pid == 0)
         {
             MyReplicationSlot = s;
             s->active_pid = MyProcPid;
         }
         SpinLockRelease(&s->mutex);
 
         /*
          * Even though we hold an exclusive lock on the database object a
          * logical slot for that DB can still be active, e.g. if it's
          * concurrently being dropped by a backend connected to another DB.
          *
          * That's fairly unlikely in practice, so we'll just bail out.
          */
         if (active_pid)
             ereport(ERROR,
                     (errcode(ERRCODE_OBJECT_IN_USE),
                      errmsg("replication slot \"%s\" is active for PID %d",
                             slotname, active_pid)));
 
         /*
          * To avoid duplicating ReplicationSlotDropAcquired() and to avoid
          * holding ReplicationSlotControlLock over filesystem operations,
          * release ReplicationSlotControlLock and use
          * ReplicationSlotDropAcquired.
          *
          * As that means the set of slots could change, restart scan from the
          * beginning each time we release the lock.
          */
         LWLockRelease(ReplicationSlotControlLock);
         ReplicationSlotDropAcquired();
         goto restart;
     }
     LWLockRelease(ReplicationSlotControlLock);
 }

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

 {
     bool        found;
 
     if (max_replication_slots == 0)
         return;
 
     ReplicationSlotCtl = (ReplicationSlotCtlData *)
         ShmemInitStruct("ReplicationSlot Ctl", ReplicationSlotsShmemSize(),
                         &found);
 
     LWLockRegisterTranche(LWTRANCHE_REPLICATION_SLOT_IO_IN_PROGRESS,
                           "replication_slot_io");
 
     if (!found)
     {
         int         i;
 
         /* First time through, so initialize */
         MemSet(ReplicationSlotCtl, 0, ReplicationSlotsShmemSize());
 
         for (i = 0; i < max_replication_slots; i++)
         {
             ReplicationSlot *slot = &ReplicationSlotCtl->replication_slots[i];
 
             /* everything else is zeroed by the memset above */
             SpinLockInit(&slot->mutex);
             LWLockInitialize(&slot->io_in_progress_lock, LWTRANCHE_REPLICATION_SLOT_IO_IN_PROGRESS);
             ConditionVariableInit(&slot->active_cv);
         }
     }
 }

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

 {
     Size        size = 0;
 
     if (max_replication_slots == 0)
         return size;
 
     size = offsetof(ReplicationSlotCtlData, replication_slots);
     size = add_size(size,
                     mul_size(max_replication_slots, sizeof(ReplicationSlot)));
 
     return size;
 }

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

 {
     const char *cp;
 
     if (strlen(name) == 0)
     {
         ereport(elevel,
                 (errcode(ERRCODE_INVALID_NAME),
                  errmsg("replication slot name \"%s\" is too short",
                         name)));
         return false;
     }
 
     if (strlen(name) >= NAMEDATALEN)
     {
         ereport(elevel,
                 (errcode(ERRCODE_NAME_TOO_LONG),
                  errmsg("replication slot name \"%s\" is too long",
                         name)));
         return false;
     }
 
     for (cp = name; *cp; cp++)
     {
         if (!((*cp >= 'a' && *cp <= 'z')
               || (*cp >= '0' && *cp <= '9')
               || (*cp == '_')))
         {
             ereport(elevel,
                     (errcode(ERRCODE_INVALID_NAME),
                      errmsg("replication slot name \"%s\" contains invalid character",
                             name),
                      errhint("Replication slot names may only contain lower case letters, numbers, and the underscore character.")));
             return false;
         }
     }
     return true;
 }

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, pg_str_endswith(), ReadDir(), ReplicationSlotsComputeRequiredLSN(), ReplicationSlotsComputeRequiredXmin(), RestoreSlotFromDisk(), rmtree(), snprintf(), and WARNING.

Referenced by StartupXLOG().

 {
     DIR        *replication_dir;
     struct dirent *replication_de;
 
     elog(DEBUG1, "starting up replication slots");
 
     /* restore all slots by iterating over all on-disk entries */
     replication_dir = AllocateDir("pg_replslot");
     while ((replication_de = ReadDir(replication_dir, "pg_replslot")) != NULL)
     {
         struct stat statbuf;
         char        path[MAXPGPATH + 12];
 
         if (strcmp(replication_de->d_name, ".") == 0 ||
             strcmp(replication_de->d_name, "..") == 0)
             continue;
 
         snprintf(path, sizeof(path), "pg_replslot/%s", replication_de->d_name);
 
         /* we're only creating directories here, skip if it's not our's */
         if (lstat(path, &statbuf) == 0 && !S_ISDIR(statbuf.st_mode))
             continue;
 
         /* we crashed while a slot was being setup or deleted, clean up */
         if (pg_str_endswith(replication_de->d_name, ".tmp"))
         {
             if (!rmtree(path, true))
             {
                 ereport(WARNING,
                         (errcode_for_file_access(),
                          errmsg("could not remove directory \"%s\"", path)));
                 continue;
             }
             fsync_fname("pg_replslot", true);
             continue;
         }
 
         /* looks like a slot in a normal state, restore */
         RestoreSlotFromDisk(replication_de->d_name);
     }
     FreeDir(replication_dir);
 
     /* currently no slots exist, we're done. */
     if (max_replication_slots <= 0)
         return;
 
     /* Now that we have recovered all the data, compute replication xmin */
     ReplicationSlotsComputeRequiredXmin(false);
     ReplicationSlotsComputeRequiredLSN();
 }