#include "postgres.h"
#include <unistd.h>
#include <sys/stat.h>
#include "access/transam.h"
#include "access/xlog_internal.h"
#include "common/file_utils.h"
#include "common/string.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "replication/slot.h"
#include "storage/fd.h"
#include "storage/ipc.h"
#include "storage/proc.h"
#include "storage/procarray.h"
#include "utils/builtins.h"

Include dependency graph for slot.c:

Data Structures
struct	ReplicationSlotOnDisk

Macros
#define	ReplicationSlotOnDiskConstantSize offsetof(ReplicationSlotOnDisk, slotdata)

#define	ReplicationSlotOnDiskNotChecksummedSize offsetof(ReplicationSlotOnDisk, version)

#define	ReplicationSlotOnDiskChecksummedSize sizeof(ReplicationSlotOnDisk) - ReplicationSlotOnDiskNotChecksummedSize

#define	ReplicationSlotOnDiskV2Size sizeof(ReplicationSlotOnDisk) - ReplicationSlotOnDiskConstantSize

#define	SLOT_MAGIC 0x1051CA1 /* format identifier */

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

Typedefs
typedef struct ReplicationSlotOnDisk	ReplicationSlotOnDisk

Functions
static void	ReplicationSlotShmemExit (int code, Datum arg)

static void	ReplicationSlotDropAcquired (void)

static void	ReplicationSlotDropPtr (ReplicationSlot *slot)

static void	RestoreSlotFromDisk (const char *name)

static void	CreateSlotOnDisk (ReplicationSlot *slot)

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

Size	ReplicationSlotsShmemSize (void)

void	ReplicationSlotsShmemInit (void)

void	ReplicationSlotInitialize (void)

bool	ReplicationSlotValidateName (const char *name, int elevel)

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

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

int	ReplicationSlotIndex (ReplicationSlot *slot)

bool	ReplicationSlotName (int index, Name name)

void	ReplicationSlotAcquire (const char *name, bool nowait)

void	ReplicationSlotRelease (void)

void	ReplicationSlotCleanup (void)

void	ReplicationSlotDrop (const char *name, bool nowait)

void	ReplicationSlotSave (void)

void	ReplicationSlotMarkDirty (void)

void	ReplicationSlotPersist (void)

void	ReplicationSlotsComputeRequiredXmin (bool already_locked)

void	ReplicationSlotsComputeRequiredLSN (void)

XLogRecPtr	ReplicationSlotsComputeLogicalRestartLSN (void)

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

void	ReplicationSlotsDropDBSlots (Oid dboid)

void	CheckSlotRequirements (void)

void	CheckSlotPermissions (void)

void	ReplicationSlotReserveWal (void)

static bool	InvalidatePossiblyObsoleteSlot (ReplicationSlot s, XLogRecPtr oldestLSN, bool invalidated)

bool	InvalidateObsoleteReplicationSlots (XLogSegNo oldestSegno)

void	CheckPointReplicationSlots (void)

void	StartupReplicationSlots (void)

Variables
ReplicationSlotCtlData *	ReplicationSlotCtl = NULL

ReplicationSlot *	MyReplicationSlot = NULL

int	max_replication_slots = 10

Macro Definition Documentation

◆ ReplicationSlotOnDiskChecksummedSize

#define ReplicationSlotOnDiskChecksummedSize sizeof(ReplicationSlotOnDisk) - ReplicationSlotOnDiskNotChecksummedSize

Definition at line 85 of file slot.c.

◆ ReplicationSlotOnDiskConstantSize

#define ReplicationSlotOnDiskConstantSize offsetof(ReplicationSlotOnDisk, slotdata)

Definition at line 79 of file slot.c.

◆ ReplicationSlotOnDiskNotChecksummedSize

#define ReplicationSlotOnDiskNotChecksummedSize offsetof(ReplicationSlotOnDisk, version)

Definition at line 82 of file slot.c.

◆ ReplicationSlotOnDiskV2Size

#define ReplicationSlotOnDiskV2Size sizeof(ReplicationSlotOnDisk) - ReplicationSlotOnDiskConstantSize

Definition at line 88 of file slot.c.

◆ SLOT_MAGIC

#define SLOT_MAGIC 0x1051CA1 /* format identifier */

Definition at line 91 of file slot.c.

◆ SLOT_VERSION

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

Definition at line 92 of file slot.c.

Typedef Documentation

◆ ReplicationSlotOnDisk

typedef struct ReplicationSlotOnDisk ReplicationSlotOnDisk

Function Documentation

◆ CheckPointReplicationSlots()

void CheckPointReplicationSlots ( void )

Definition at line 1442 of file slot.c.

 {
     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);
 }

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

Referenced by CheckPointGuts().

◆ CheckSlotPermissions()

void CheckSlotPermissions ( void )

Definition at line 1141 of file slot.c.

 {
     if (!superuser() && !has_rolreplication(GetUserId()))
         ereport(ERROR,
                 (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
                  errmsg("must be superuser or replication role to use replication slots")));
 }

References ereport, errcode(), errmsg(), ERROR, GetUserId(), has_rolreplication(), and superuser().

Referenced by copy_replication_slot(), pg_create_logical_replication_slot(), pg_create_physical_replication_slot(), pg_drop_replication_slot(), pg_logical_slot_get_changes_guts(), and pg_replication_slot_advance().

◆ CheckSlotRequirements()

void CheckSlotRequirements ( void )

Definition at line 1119 of file slot.c.

 {
     /*
      * NB: Adding a new requirement likely means that RestoreSlotFromDisk()
      * needs the same check.
      */
  
     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")));
 }

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

◆ CreateSlotOnDisk()

static void CreateSlotOnDisk ( ReplicationSlot * slot )

static

Definition at line 1538 of file slot.c.

 {
     char        tmppath[MAXPGPATH];
     char        path[MAXPGPATH];
     struct stat st;
  
     /*
      * No need to take out the io_in_progress_lock, nobody else can see this
      * slot yet, so nobody else will write. We're reusing SaveSlotToPath which
      * takes out the lock, if we'd take the lock here, we'd deadlock.
      */
  
     sprintf(path, "pg_replslot/%s", NameStr(slot->data.name));
     sprintf(tmppath, "pg_replslot/%s.tmp", NameStr(slot->data.name));
  
     /*
      * It's just barely possible that some previous effort to create or drop a
      * slot with this name left a temp directory lying around. If that seems
      * to be the case, try to remove it.  If the rmtree() fails, we'll error
      * out at the MakePGDirectory() below, so we don't bother checking
      * success.
      */
     if (stat(tmppath, &st) == 0 && S_ISDIR(st.st_mode))
         rmtree(tmppath, true);
  
     /* Create and fsync the temporary slot directory. */
     if (MakePGDirectory(tmppath) < 0)
         ereport(ERROR,
                 (errcode_for_file_access(),
                  errmsg("could not create directory \"%s\": %m",
                         tmppath)));
     fsync_fname(tmppath, true);
  
     /* Write the actual state file. */
     slot->dirty = true;         /* signal that we really need to write */
     SaveSlotToPath(slot, tmppath, ERROR);
  
     /* Rename the directory into place. */
     if (rename(tmppath, path) != 0)
         ereport(ERROR,
                 (errcode_for_file_access(),
                  errmsg("could not rename file \"%s\" to \"%s\": %m",
                         tmppath, path)));
  
     /*
      * If we'd now fail - really unlikely - we wouldn't know whether this slot
      * would persist after an OS crash or not - so, force a restart. The
      * restart would try to fsync this again till it works.
      */
     START_CRIT_SECTION();
  
     fsync_fname(path, true);
     fsync_fname("pg_replslot", true);
  
     END_CRIT_SECTION();
 }

References ReplicationSlot::data, ReplicationSlot::dirty, END_CRIT_SECTION, ereport, errcode_for_file_access(), errmsg(), ERROR, fsync_fname(), MakePGDirectory(), MAXPGPATH, ReplicationSlotPersistentData::name, NameStr, rmtree(), S_ISDIR, SaveSlotToPath(), sprintf, stat::st_mode, START_CRIT_SECTION, and stat.

Referenced by ReplicationSlotCreate().

◆ InvalidateObsoleteReplicationSlots()

bool InvalidateObsoleteReplicationSlots ( XLogSegNo oldestSegno )

Definition at line 1399 of file slot.c.

 {
     XLogRecPtr  oldestLSN;
     bool        invalidated = false;
  
     XLogSegNoOffsetToRecPtr(oldestSegno, 0, wal_segment_size, oldestLSN);
  
 restart:
     LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
     for (int i = 0; i < max_replication_slots; i++)
     {
         ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
  
         if (!s->in_use)
             continue;
  
         if (InvalidatePossiblyObsoleteSlot(s, oldestLSN, &invalidated))
         {
             /* if the lock was released, start from scratch */
             goto restart;
         }
     }
     LWLockRelease(ReplicationSlotControlLock);
  
     /*
      * If any slots have been invalidated, recalculate the resource limits.
      */
     if (invalidated)
     {
         ReplicationSlotsComputeRequiredXmin(false);
         ReplicationSlotsComputeRequiredLSN();
     }
  
     return invalidated;
 }

References i, ReplicationSlot::in_use, InvalidatePossiblyObsoleteSlot(), LW_SHARED, LWLockAcquire(), LWLockRelease(), max_replication_slots, ReplicationSlotCtlData::replication_slots, ReplicationSlotCtl, ReplicationSlotsComputeRequiredLSN(), ReplicationSlotsComputeRequiredXmin(), wal_segment_size, and XLogSegNoOffsetToRecPtr.

Referenced by CreateCheckPoint(), and CreateRestartPoint().

◆ InvalidatePossiblyObsoleteSlot()

static bool InvalidatePossiblyObsoleteSlot	(	ReplicationSlot *	s,
		XLogRecPtr	oldestLSN,
		bool *	invalidated
	)

static

Definition at line 1239 of file slot.c.

 {
     int         last_signaled_pid = 0;
     bool        released_lock = false;
  
     for (;;)
     {
         XLogRecPtr  restart_lsn;
         NameData    slotname;
         int         active_pid = 0;
  
         Assert(LWLockHeldByMeInMode(ReplicationSlotControlLock, LW_SHARED));
  
         if (!s->in_use)
         {
             if (released_lock)
                 LWLockRelease(ReplicationSlotControlLock);
             break;
         }
  
         /*
          * Check if the slot needs to be invalidated. If it needs to be
          * invalidated, and is not currently acquired, acquire it and mark it
          * as having been invalidated.  We do this with the spinlock held to
          * avoid race conditions -- for example the restart_lsn could move
          * forward, or the slot could be dropped.
          */
         SpinLockAcquire(&s->mutex);
  
         restart_lsn = s->data.restart_lsn;
  
         /*
          * If the slot is already invalid or is fresh enough, we don't need to
          * do anything.
          */
         if (XLogRecPtrIsInvalid(restart_lsn) || restart_lsn >= oldestLSN)
         {
             SpinLockRelease(&s->mutex);
             if (released_lock)
                 LWLockRelease(ReplicationSlotControlLock);
             break;
         }
  
         slotname = s->data.name;
         active_pid = s->active_pid;
  
         /*
          * If the slot can be acquired, do so and mark it invalidated
          * immediately.  Otherwise we'll signal the owning process, below, and
          * retry.
          */
         if (active_pid == 0)
         {
             MyReplicationSlot = s;
             s->active_pid = MyProcPid;
             s->data.invalidated_at = restart_lsn;
             s->data.restart_lsn = InvalidXLogRecPtr;
  
             /* Let caller know */
             *invalidated = true;
         }
  
         SpinLockRelease(&s->mutex);
  
         if (active_pid != 0)
         {
             /*
              * Prepare the sleep on the slot's condition variable before
              * releasing the lock, to close a possible race condition if the
              * slot is released before the sleep below.
              */
             ConditionVariablePrepareToSleep(&s->active_cv);
  
             LWLockRelease(ReplicationSlotControlLock);
             released_lock = true;
  
             /*
              * Signal to terminate the process that owns the slot, if we
              * haven't already signalled it.  (Avoidance of repeated
              * signalling is the only reason for there to be a loop in this
              * routine; otherwise we could rely on caller's restart loop.)
              *
              * There is the race condition that other process may own the slot
              * after its current owner process is terminated and before this
              * process owns it. To handle that, we signal only if the PID of
              * the owning process has changed from the previous time. (This
              * logic assumes that the same PID is not reused very quickly.)
              */
             if (last_signaled_pid != active_pid)
             {
                 ereport(LOG,
                         errmsg("terminating process %d to release replication slot \"%s\"",
                                active_pid, NameStr(slotname)),
                         errdetail("The slot's restart_lsn %X/%X exceeds the limit by %llu bytes.",
                                   LSN_FORMAT_ARGS(restart_lsn),
                                   (unsigned long long) (oldestLSN - restart_lsn)),
                         errhint("You might need to increase max_slot_wal_keep_size."));
  
                 (void) kill(active_pid, SIGTERM);
                 last_signaled_pid = active_pid;
             }
  
             /* Wait until the slot is released. */
             ConditionVariableSleep(&s->active_cv,
                                    WAIT_EVENT_REPLICATION_SLOT_DROP);
  
             /*
              * Re-acquire lock and start over; we expect to invalidate the
              * slot next time (unless another process acquires the slot in the
              * meantime).
              */
             LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
             continue;
         }
         else
         {
             /*
              * We hold the slot now and have already invalidated it; flush it
              * to ensure that state persists.
              *
              * Don't want to hold ReplicationSlotControlLock across file
              * system operations, so release it now but be sure to tell caller
              * to restart from scratch.
              */
             LWLockRelease(ReplicationSlotControlLock);
             released_lock = true;
  
             /* Make sure the invalidated state persists across server restart */
             ReplicationSlotMarkDirty();
             ReplicationSlotSave();
             ReplicationSlotRelease();
  
             ereport(LOG,
                     errmsg("invalidating obsolete replication slot \"%s\"",
                            NameStr(slotname)),
                     errdetail("The slot's restart_lsn %X/%X exceeds the limit by %llu bytes.",
                               LSN_FORMAT_ARGS(restart_lsn),
                               (unsigned long long) (oldestLSN - restart_lsn)),
                     errhint("You might need to increase max_slot_wal_keep_size."));
  
             /* done with this slot for now */
             break;
         }
     }
  
     Assert(released_lock == !LWLockHeldByMe(ReplicationSlotControlLock));
  
     return released_lock;
 }

References ReplicationSlot::active_cv, ReplicationSlot::active_pid, Assert(), ConditionVariablePrepareToSleep(), ConditionVariableSleep(), ReplicationSlot::data, ereport, errdetail(), errhint(), errmsg(), ReplicationSlot::in_use, ReplicationSlotPersistentData::invalidated_at, InvalidXLogRecPtr, kill, LOG, LSN_FORMAT_ARGS, LW_SHARED, LWLockAcquire(), LWLockHeldByMe(), LWLockHeldByMeInMode(), LWLockRelease(), ReplicationSlot::mutex, MyProcPid, MyReplicationSlot, ReplicationSlotPersistentData::name, NameStr, ReplicationSlotMarkDirty(), ReplicationSlotRelease(), ReplicationSlotSave(), ReplicationSlotPersistentData::restart_lsn, SpinLockAcquire, SpinLockRelease, WAIT_EVENT_REPLICATION_SLOT_DROP, and XLogRecPtrIsInvalid.

Referenced by InvalidateObsoleteReplicationSlots().

◆ ReplicationSlotAcquire()

void ReplicationSlotAcquire	(	const char *	name,
		bool	nowait
	)

Definition at line 450 of file slot.c.

 {
     ReplicationSlot *s;
     int         active_pid;
  
     Assert(name != NULL);
  
 retry:
     Assert(MyReplicationSlot == NULL);
  
     LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
  
     /*
      * Search for the slot with the specified name if the slot to acquire is
      * not given. If the slot is not found, we either return -1 or error out.
      */
     s = SearchNamedReplicationSlot(name, false);
     if (s == NULL || !s->in_use)
     {
         LWLockRelease(ReplicationSlotControlLock);
  
         ereport(ERROR,
                 (errcode(ERRCODE_UNDEFINED_OBJECT),
                  errmsg("replication slot \"%s\" does not exist",
                         name)));
     }
  
     /*
      * This is the slot we want; check if it's active under some other
      * process.  In single user mode, we don't need this check.
      */
     if (IsUnderPostmaster)
     {
         /*
          * Get ready to sleep on the slot in case it is active.  (We may end
          * up not sleeping, but we don't want to do this while holding the
          * spinlock.)
          */
         if (!nowait)
             ConditionVariablePrepareToSleep(&s->active_cv);
  
         SpinLockAcquire(&s->mutex);
         if (s->active_pid == 0)
             s->active_pid = MyProcPid;
         active_pid = s->active_pid;
         SpinLockRelease(&s->mutex);
     }
     else
         active_pid = MyProcPid;
     LWLockRelease(ReplicationSlotControlLock);
  
     /*
      * If we found the slot but it's already active in another process, we
      * wait until the owning process signals us that it's been released, or
      * error out.
      */
     if (active_pid != MyProcPid)
     {
         if (!nowait)
         {
             /* Wait here until we get signaled, and then restart */
             ConditionVariableSleep(&s->active_cv,
                                    WAIT_EVENT_REPLICATION_SLOT_DROP);
             ConditionVariableCancelSleep();
             goto retry;
         }
  
         ereport(ERROR,
                 (errcode(ERRCODE_OBJECT_IN_USE),
                  errmsg("replication slot \"%s\" is active for PID %d",
                         NameStr(s->data.name), active_pid)));
     }
     else if (!nowait)
         ConditionVariableCancelSleep(); /* no sleep needed after all */
  
     /* Let everybody know we've modified this slot */
     ConditionVariableBroadcast(&s->active_cv);
  
     /* We made this slot active, so it's ours now. */
     MyReplicationSlot = s;
  
     /*
      * The call to pgstat_acquire_replslot() protects against stats for a
      * different slot, from before a restart or such, being present during
      * pgstat_report_replslot().
      */
     if (SlotIsLogical(s))
         pgstat_acquire_replslot(s);
 }

References ReplicationSlot::active_cv, ReplicationSlot::active_pid, Assert(), ConditionVariableBroadcast(), ConditionVariableCancelSleep(), ConditionVariablePrepareToSleep(), ConditionVariableSleep(), ReplicationSlot::data, ereport, errcode(), errmsg(), ERROR, ReplicationSlot::in_use, IsUnderPostmaster, LW_SHARED, LWLockAcquire(), LWLockRelease(), ReplicationSlot::mutex, MyProcPid, MyReplicationSlot, name, ReplicationSlotPersistentData::name, NameStr, pgstat_acquire_replslot(), SearchNamedReplicationSlot(), SlotIsLogical, SpinLockAcquire, SpinLockRelease, and WAIT_EVENT_REPLICATION_SLOT_DROP.

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

◆ ReplicationSlotCleanup()

void ReplicationSlotCleanup ( void )

Definition at line 603 of file slot.c.

 {
     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);
 }

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

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

◆ ReplicationSlotCreate()

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

Definition at line 252 of file slot.c.

 {
     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));
     namestrcpy(&slot->data.name, name);
     slot->data.database = db_specific ? MyDatabaseId : InvalidOid;
     slot->data.persistency = persistency;
     slot->data.two_phase = two_phase;
     slot->data.two_phase_at = InvalidXLogRecPtr;
  
     /* 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
      * ReplicationSlotAcquire() 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);
  
     /*
      * Create statistics entry for the new logical slot. We don't collect any
      * stats for physical slots, so no need to create an entry for the same.
      * See ReplicationSlotDropPtr for why we need to do this before releasing
      * ReplicationSlotAllocationLock.
      */
     if (SlotIsLogical(slot))
         pgstat_create_replslot(slot);
  
     /*
      * 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);
 }

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, MyReplicationSlot, name, ReplicationSlotPersistentData::name, NameStr, namestrcpy(), ReplicationSlotPersistentData::persistency, pgstat_create_replslot(), ReplicationSlotCtlData::replication_slots, ReplicationSlotCtl, ReplicationSlotValidateName(), SlotIsLogical, SpinLockAcquire, SpinLockRelease, two_phase, ReplicationSlotPersistentData::two_phase, and ReplicationSlotPersistentData::two_phase_at.

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

◆ ReplicationSlotDrop()

void ReplicationSlotDrop	(	const char *	name,
		bool	nowait
	)

Definition at line 641 of file slot.c.

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

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

Referenced by DropReplicationSlot(), and pg_drop_replication_slot().

◆ ReplicationSlotDropAcquired()

static void ReplicationSlotDropAcquired ( void )

static

Definition at line 654 of file slot.c.

 {
     ReplicationSlot *slot = MyReplicationSlot;
  
     Assert(MyReplicationSlot != NULL);
  
     /* slot isn't acquired anymore */
     MyReplicationSlot = NULL;
  
     ReplicationSlotDropPtr(slot);
 }

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

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

◆ ReplicationSlotDropPtr()

static void ReplicationSlotDropPtr ( ReplicationSlot * slot )

static

Definition at line 671 of file slot.c.

 {
     char        path[MAXPGPATH];
     char        tmppath[MAXPGPATH];
  
     /*
      * If some other backend ran this code concurrently with us, we might try
      * to delete a slot with a certain name while someone else was trying to
      * create a slot with the same name.
      */
     LWLockAcquire(ReplicationSlotAllocationLock, LW_EXCLUSIVE);
  
     /* Generate pathnames. */
     sprintf(path, "pg_replslot/%s", NameStr(slot->data.name));
     sprintf(tmppath, "pg_replslot/%s.tmp", NameStr(slot->data.name));
  
     /*
      * Rename the slot directory on disk, so that we'll no longer recognize
      * this as a valid slot.  Note that if this fails, we've got to mark the
      * slot inactive before bailing out.  If we're dropping an ephemeral or a
      * temporary slot, we better never fail hard as the caller won't expect
      * the slot to survive and this might get called during error handling.
      */
     if (rename(path, tmppath) == 0)
     {
         /*
          * We need to fsync() the directory we just renamed and its parent to
          * make sure that our changes are on disk in a crash-safe fashion.  If
          * fsync() fails, we can't be sure whether the changes are on disk or
          * not.  For now, we handle that by panicking;
          * StartupReplicationSlots() will try to straighten it out after
          * restart.
          */
         START_CRIT_SECTION();
         fsync_fname(tmppath, true);
         fsync_fname("pg_replslot", true);
         END_CRIT_SECTION();
     }
     else
     {
         bool        fail_softly = slot->data.persistency != RS_PERSISTENT;
  
         SpinLockAcquire(&slot->mutex);
         slot->active_pid = 0;
         SpinLockRelease(&slot->mutex);
  
         /* wake up anyone waiting on this slot */
         ConditionVariableBroadcast(&slot->active_cv);
  
         ereport(fail_softly ? WARNING : ERROR,
                 (errcode_for_file_access(),
                  errmsg("could not rename file \"%s\" to \"%s\": %m",
                         path, tmppath)));
     }
  
     /*
      * The slot is definitely gone.  Lock out concurrent scans of the array
      * long enough to kill it.  It's OK to clear the active PID here without
      * grabbing the mutex because nobody else can be scanning the array here,
      * and nobody can be attached to this slot and thus access it without
      * scanning the array.
      *
      * Also wake up processes waiting for it.
      */
     LWLockAcquire(ReplicationSlotControlLock, LW_EXCLUSIVE);
     slot->active_pid = 0;
     slot->in_use = false;
     LWLockRelease(ReplicationSlotControlLock);
     ConditionVariableBroadcast(&slot->active_cv);
  
     /*
      * Slot is dead and doesn't prevent resource removal anymore, recompute
      * limits.
      */
     ReplicationSlotsComputeRequiredXmin(false);
     ReplicationSlotsComputeRequiredLSN();
  
     /*
      * If removing the directory fails, the worst thing that will happen is
      * that the user won't be able to create a new slot with the same name
      * until the next server restart.  We warn about it, but that's all.
      */
     if (!rmtree(tmppath, true))
         ereport(WARNING,
                 (errmsg("could not remove directory \"%s\"", tmppath)));
  
     /*
      * Drop the statistics entry for the replication slot.  Do this while
      * holding ReplicationSlotAllocationLock so that we don't drop a
      * statistics entry for another slot with the same name just created in
      * another session.
      */
     if (SlotIsLogical(slot))
         pgstat_drop_replslot(slot);
  
     /*
      * We release this at the very end, so that nobody starts trying to create
      * a slot while we're still cleaning up the detritus of the old one.
      */
     LWLockRelease(ReplicationSlotAllocationLock);
 }

References ReplicationSlot::active_cv, ReplicationSlot::active_pid, ConditionVariableBroadcast(), ReplicationSlot::data, END_CRIT_SECTION, ereport, errcode_for_file_access(), errmsg(), ERROR, fsync_fname(), ReplicationSlot::in_use, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MAXPGPATH, ReplicationSlot::mutex, ReplicationSlotPersistentData::name, NameStr, ReplicationSlotPersistentData::persistency, pgstat_drop_replslot(), ReplicationSlotsComputeRequiredLSN(), ReplicationSlotsComputeRequiredXmin(), rmtree(), RS_PERSISTENT, SlotIsLogical, SpinLockAcquire, SpinLockRelease, sprintf, START_CRIT_SECTION, and WARNING.

Referenced by ReplicationSlotCleanup(), and ReplicationSlotDropAcquired().

◆ ReplicationSlotIndex()

int ReplicationSlotIndex ( ReplicationSlot * slot )

Definition at line 407 of file slot.c.

 {
     Assert(slot >= ReplicationSlotCtl->replication_slots &&
            slot < ReplicationSlotCtl->replication_slots + max_replication_slots);
  
     return slot - ReplicationSlotCtl->replication_slots;
 }

References Assert(), max_replication_slots, ReplicationSlotCtlData::replication_slots, and ReplicationSlotCtl.

Referenced by get_replslot_index(), pgstat_acquire_replslot(), pgstat_create_replslot(), pgstat_drop_replslot(), pgstat_report_replslot(), and pgstat_reset_replslot().

◆ ReplicationSlotInitialize()

void ReplicationSlotInitialize ( void )

Definition at line 170 of file slot.c.

 {
     before_shmem_exit(ReplicationSlotShmemExit, 0);
 }

References before_shmem_exit(), and ReplicationSlotShmemExit().

Referenced by BaseInit().

◆ ReplicationSlotMarkDirty()

void ReplicationSlotMarkDirty ( void )

Definition at line 796 of file slot.c.

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

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

Referenced by copy_replication_slot(), create_physical_replication_slot(), CreateDecodingContext(), CreateInitDecodingContext(), CreateReplicationSlot(), InvalidatePossiblyObsoleteSlot(), LogicalConfirmReceivedLocation(), pg_logical_replication_slot_advance(), pg_logical_slot_get_changes_guts(), pg_physical_replication_slot_advance(), PhysicalConfirmReceivedLocation(), PhysicalReplicationSlotNewXmin(), and ReplicationSlotPersist().

◆ ReplicationSlotName()

bool ReplicationSlotName	(	int	index,
		Name	name
	)

Definition at line 423 of file slot.c.

 {
     ReplicationSlot *slot;
     bool        found;
  
     slot = &ReplicationSlotCtl->replication_slots[index];
  
     /*
      * Ensure that the slot cannot be dropped while we copy the name. Don't
      * need the spinlock as the name of an existing slot cannot change.
      */
     LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
     found = slot->in_use;
     if (slot->in_use)
         namestrcpy(name, NameStr(slot->data.name));
     LWLockRelease(ReplicationSlotControlLock);
  
     return found;
 }

References ReplicationSlot::data, ReplicationSlot::in_use, LW_SHARED, LWLockAcquire(), LWLockRelease(), name, ReplicationSlotPersistentData::name, NameStr, namestrcpy(), ReplicationSlotCtlData::replication_slots, and ReplicationSlotCtl.

Referenced by pgstat_replslot_to_serialized_name_cb().

◆ ReplicationSlotPersist()

void ReplicationSlotPersist ( void )

Definition at line 813 of file slot.c.

 {
     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();
 }

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

◆ ReplicationSlotRelease()

void ReplicationSlotRelease ( void )

Definition at line 547 of file slot.c.

 {
     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);
     MyProc->statusFlags &= ~PROC_IN_LOGICAL_DECODING;
     ProcGlobal->statusFlags[MyProc->pgxactoff] = MyProc->statusFlags;
     LWLockRelease(ProcArrayLock);
 }

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, PGPROC::statusFlags, PROC_HDR::statusFlags, TransactionIdIsValid, and ReplicationSlotPersistentData::xmin.

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

◆ ReplicationSlotReserveWal()

void ReplicationSlotReserveWal ( void )

Definition at line 1156 of file slot.c.

 {
     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;
         XLogRecPtr  restart_lsn;
  
         /*
          * 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 */
             restart_lsn = GetXLogInsertRecPtr();
             SpinLockAcquire(&slot->mutex);
             slot->data.restart_lsn = restart_lsn;
             SpinLockRelease(&slot->mutex);
  
             /* make sure we have enough information to start */
             flushptr = LogStandbySnapshot();
  
             /* and make sure it's fsynced to disk */
             XLogFlush(flushptr);
         }
         else
         {
             restart_lsn = GetRedoRecPtr();
             SpinLockAcquire(&slot->mutex);
             slot->data.restart_lsn = restart_lsn;
             SpinLockRelease(&slot->mutex);
         }
  
         /* 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, wal_segment_size);
         if (XLogGetLastRemovedSegno() < segno)
             break;
     }
 }

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

◆ ReplicationSlotSave()

void ReplicationSlotSave ( void )

Definition at line 778 of file slot.c.

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

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

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

◆ ReplicationSlotsComputeLogicalRestartLSN()

XLogRecPtr ReplicationSlotsComputeLogicalRestartLSN ( void )

Definition at line 935 of file slot.c.

 {
     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 (restart_lsn == InvalidXLogRecPtr)
             continue;
  
         if (result == InvalidXLogRecPtr ||
             restart_lsn < result)
             result = restart_lsn;
     }
  
     LWLockRelease(ReplicationSlotControlLock);
  
     return result;
 }

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

Referenced by CheckPointLogicalRewriteHeap(), and CheckPointSnapBuild().

◆ ReplicationSlotsComputeRequiredLSN()

void ReplicationSlotsComputeRequiredLSN ( void )

Definition at line 892 of file slot.c.

 {
     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);
 }

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

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

◆ ReplicationSlotsComputeRequiredXmin()

void ReplicationSlotsComputeRequiredXmin ( bool already_locked )

Definition at line 835 of file slot.c.

 {
     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;
         bool        invalidated;
  
         if (!s->in_use)
             continue;
  
         SpinLockAcquire(&s->mutex);
         effective_xmin = s->effective_xmin;
         effective_catalog_xmin = s->effective_catalog_xmin;
         invalidated = (!XLogRecPtrIsInvalid(s->data.invalidated_at) &&
                        XLogRecPtrIsInvalid(s->data.restart_lsn));
         SpinLockRelease(&s->mutex);
  
         /* invalidated slots need not apply */
         if (invalidated)
             continue;
  
         /* 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);
 }

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

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

◆ ReplicationSlotsCountDBSlots()

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

Definition at line 987 of file slot.c.

 {
     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;
 }

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

Referenced by dropdb().

◆ ReplicationSlotsDropDBSlots()

void ReplicationSlotsDropDBSlots ( Oid dboid )

Definition at line 1043 of file slot.c.

 {
     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);
 }

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, MyReplicationSlot, ReplicationSlotPersistentData::name, NameStr, ReplicationSlotCtlData::replication_slots, ReplicationSlotCtl, ReplicationSlotDropAcquired(), SlotIsLogical, SpinLockAcquire, and SpinLockRelease.

Referenced by dbase_redo(), and dropdb().

◆ ReplicationSlotShmemExit()

static void ReplicationSlotShmemExit	(	int	code,
		Datum	arg
	)

static

Definition at line 179 of file slot.c.

 {
     /* Make sure active replication slots are released */
     if (MyReplicationSlot != NULL)
         ReplicationSlotRelease();
  
     /* Also cleanup all the temporary slots. */
     ReplicationSlotCleanup();
 }

References MyReplicationSlot, ReplicationSlotCleanup(), and ReplicationSlotRelease().

Referenced by ReplicationSlotInitialize().

◆ ReplicationSlotsShmemInit()

void ReplicationSlotsShmemInit ( void )

Definition at line 135 of file slot.c.

 {
     bool        found;
  
     if (max_replication_slots == 0)
         return;
  
     ReplicationSlotCtl = (ReplicationSlotCtlData *)
         ShmemInitStruct("ReplicationSlot Ctl", ReplicationSlotsShmemSize(),
                         &found);
  
     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);
             ConditionVariableInit(&slot->active_cv);
         }
     }
 }

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, ReplicationSlotCtl, ReplicationSlotsShmemSize(), ShmemInitStruct(), and SpinLockInit.

Referenced by CreateSharedMemoryAndSemaphores().

◆ ReplicationSlotsShmemSize()

Size ReplicationSlotsShmemSize ( void )

Definition at line 117 of file slot.c.

 {
     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;
 }

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

Referenced by CalculateShmemSize(), and ReplicationSlotsShmemInit().

◆ ReplicationSlotValidateName()

bool ReplicationSlotValidateName	(	const char *	name,
		int	elevel
	)

Definition at line 198 of file slot.c.

 {
     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;
 }

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

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

◆ RestoreSlotFromDisk()

static void RestoreSlotFromDisk ( const char * name )

static

Definition at line 1753 of file slot.c.

 {
     ReplicationSlotOnDisk cp;
     int         i;
     char        slotdir[MAXPGPATH + 12];
     char        path[MAXPGPATH + 22];
     int         fd;
     bool        restored = false;
     int         readBytes;
     pg_crc32c   checksum;
  
     /* no need to lock here, no concurrent access allowed yet */
  
     /* delete temp file if it exists */
     sprintf(slotdir, "pg_replslot/%s", name);
     sprintf(path, "%s/state.tmp", slotdir);
     if (unlink(path) < 0 && errno != ENOENT)
         ereport(PANIC,
                 (errcode_for_file_access(),
                  errmsg("could not remove file \"%s\": %m", path)));
  
     sprintf(path, "%s/state", slotdir);
  
     elog(DEBUG1, "restoring replication slot from \"%s\"", path);
  
     /* on some operating systems fsyncing a file requires O_RDWR */
     fd = OpenTransientFile(path, O_RDWR | PG_BINARY);
  
     /*
      * We do not need to handle this as we are rename()ing the directory into
      * place only after we fsync()ed the state file.
      */
     if (fd < 0)
         ereport(PANIC,
                 (errcode_for_file_access(),
                  errmsg("could not open file \"%s\": %m", path)));
  
     /*
      * Sync state file before we're reading from it. We might have crashed
      * while it wasn't synced yet and we shouldn't continue on that basis.
      */
     pgstat_report_wait_start(WAIT_EVENT_REPLICATION_SLOT_RESTORE_SYNC);
     if (pg_fsync(fd) != 0)
         ereport(PANIC,
                 (errcode_for_file_access(),
                  errmsg("could not fsync file \"%s\": %m",
                         path)));
     pgstat_report_wait_end();
  
     /* Also sync the parent directory */
     START_CRIT_SECTION();
     fsync_fname(slotdir, true);
     END_CRIT_SECTION();
  
     /* read part of statefile that's guaranteed to be version independent */
     pgstat_report_wait_start(WAIT_EVENT_REPLICATION_SLOT_READ);
     readBytes = read(fd, &cp, ReplicationSlotOnDiskConstantSize);
     pgstat_report_wait_end();
     if (readBytes != ReplicationSlotOnDiskConstantSize)
     {
         if (readBytes < 0)
             ereport(PANIC,
                     (errcode_for_file_access(),
                      errmsg("could not read file \"%s\": %m", path)));
         else
             ereport(PANIC,
                     (errcode(ERRCODE_DATA_CORRUPTED),
                      errmsg("could not read file \"%s\": read %d of %zu",
                             path, readBytes,
                             (Size) ReplicationSlotOnDiskConstantSize)));
     }
  
     /* verify magic */
     if (cp.magic != SLOT_MAGIC)
         ereport(PANIC,
                 (errcode(ERRCODE_DATA_CORRUPTED),
                  errmsg("replication slot file \"%s\" has wrong magic number: %u instead of %u",
                         path, cp.magic, SLOT_MAGIC)));
  
     /* verify version */
     if (cp.version != SLOT_VERSION)
         ereport(PANIC,
                 (errcode(ERRCODE_DATA_CORRUPTED),
                  errmsg("replication slot file \"%s\" has unsupported version %u",
                         path, cp.version)));
  
     /* boundary check on length */
     if (cp.length != ReplicationSlotOnDiskV2Size)
         ereport(PANIC,
                 (errcode(ERRCODE_DATA_CORRUPTED),
                  errmsg("replication slot file \"%s\" has corrupted length %u",
                         path, cp.length)));
  
     /* Now that we know the size, read the entire file */
     pgstat_report_wait_start(WAIT_EVENT_REPLICATION_SLOT_READ);
     readBytes = read(fd,
                      (char *) &cp + ReplicationSlotOnDiskConstantSize,
                      cp.length);
     pgstat_report_wait_end();
     if (readBytes != cp.length)
     {
         if (readBytes < 0)
             ereport(PANIC,
                     (errcode_for_file_access(),
                      errmsg("could not read file \"%s\": %m", path)));
         else
             ereport(PANIC,
                     (errcode(ERRCODE_DATA_CORRUPTED),
                      errmsg("could not read file \"%s\": read %d of %zu",
                             path, readBytes, (Size) cp.length)));
     }
  
     if (CloseTransientFile(fd) != 0)
         ereport(PANIC,
                 (errcode_for_file_access(),
                  errmsg("could not close file \"%s\": %m", path)));
  
     /* now verify the CRC */
     INIT_CRC32C(checksum);
     COMP_CRC32C(checksum,
                 (char *) &cp + ReplicationSlotOnDiskNotChecksummedSize,
                 ReplicationSlotOnDiskChecksummedSize);
     FIN_CRC32C(checksum);
  
     if (!EQ_CRC32C(checksum, cp.checksum))
         ereport(PANIC,
                 (errmsg("checksum mismatch for replication slot file \"%s\": is %u, should be %u",
                         path, checksum, cp.checksum)));
  
     /*
      * If we crashed with an ephemeral slot active, don't restore but delete
      * it.
      */
     if (cp.slotdata.persistency != RS_PERSISTENT)
     {
         if (!rmtree(slotdir, true))
         {
             ereport(WARNING,
                     (errmsg("could not remove directory \"%s\"",
                             slotdir)));
         }
         fsync_fname("pg_replslot", true);
         return;
     }
  
     /*
      * Verify that requirements for the specific slot type are met. That's
      * important because if these aren't met we're not guaranteed to retain
      * all the necessary resources for the slot.
      *
      * NB: We have to do so *after* the above checks for ephemeral slots,
      * because otherwise a slot that shouldn't exist anymore could prevent
      * restarts.
      *
      * NB: Changing the requirements here also requires adapting
      * CheckSlotRequirements() and CheckLogicalDecodingRequirements().
      */
     if (cp.slotdata.database != InvalidOid && wal_level < WAL_LEVEL_LOGICAL)
         ereport(FATAL,
                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                  errmsg("logical replication slot \"%s\" exists, but wal_level < logical",
                         NameStr(cp.slotdata.name)),
                  errhint("Change wal_level to be logical or higher.")));
     else if (wal_level < WAL_LEVEL_REPLICA)
         ereport(FATAL,
                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                  errmsg("physical replication slot \"%s\" exists, but wal_level < replica",
                         NameStr(cp.slotdata.name)),
                  errhint("Change wal_level to be replica or higher.")));
  
     /* nothing can be active yet, don't lock anything */
     for (i = 0; i < max_replication_slots; i++)
     {
         ReplicationSlot *slot;
  
         slot = &ReplicationSlotCtl->replication_slots[i];
  
         if (slot->in_use)
             continue;
  
         /* restore the entire set of persistent data */
         memcpy(&slot->data, &cp.slotdata,
                sizeof(ReplicationSlotPersistentData));
  
         /* initialize in memory state */
         slot->effective_xmin = cp.slotdata.xmin;
         slot->effective_catalog_xmin = cp.slotdata.catalog_xmin;
  
         slot->candidate_catalog_xmin = InvalidTransactionId;
         slot->candidate_xmin_lsn = InvalidXLogRecPtr;
         slot->candidate_restart_lsn = InvalidXLogRecPtr;
         slot->candidate_restart_valid = InvalidXLogRecPtr;
  
         slot->in_use = true;
         slot->active_pid = 0;
  
         restored = true;
         break;
     }
  
     if (!restored)
         ereport(FATAL,
                 (errmsg("too many replication slots active before shutdown"),
                  errhint("Increase max_replication_slots and try again.")));
 }

Referenced by StartupReplicationSlots().

◆ SaveSlotToPath()

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

static

Definition at line 1599 of file slot.c.

 {
     char        tmppath[MAXPGPATH];
     char        path[MAXPGPATH];
     int         fd;
     ReplicationSlotOnDisk cp;
     bool        was_dirty;
  
     /* first check whether there's something to write out */
     SpinLockAcquire(&slot->mutex);
     was_dirty = slot->dirty;
     slot->just_dirtied = false;
     SpinLockRelease(&slot->mutex);
  
     /* and don't do anything if there's nothing to write */
     if (!was_dirty)
         return;
  
     LWLockAcquire(&slot->io_in_progress_lock, LW_EXCLUSIVE);
  
     /* silence valgrind :( */
     memset(&cp, 0, sizeof(ReplicationSlotOnDisk));
  
     sprintf(tmppath, "%s/state.tmp", dir);
     sprintf(path, "%s/state", dir);
  
     fd = OpenTransientFile(tmppath, O_CREAT | O_EXCL | O_WRONLY | PG_BINARY);
     if (fd < 0)
     {
         /*
          * If not an ERROR, then release the lock before returning.  In case
          * of an ERROR, the error recovery path automatically releases the
          * lock, but no harm in explicitly releasing even in that case.  Note
          * that LWLockRelease() could affect errno.
          */
         int         save_errno = errno;
  
         LWLockRelease(&slot->io_in_progress_lock);
         errno = save_errno;
         ereport(elevel,
                 (errcode_for_file_access(),
                  errmsg("could not create file \"%s\": %m",
                         tmppath)));
         return;
     }
  
     cp.magic = SLOT_MAGIC;
     INIT_CRC32C(cp.checksum);
     cp.version = SLOT_VERSION;
     cp.length = ReplicationSlotOnDiskV2Size;
  
     SpinLockAcquire(&slot->mutex);
  
     memcpy(&cp.slotdata, &slot->data, sizeof(ReplicationSlotPersistentData));
  
     SpinLockRelease(&slot->mutex);
  
     COMP_CRC32C(cp.checksum,
                 (char *) (&cp) + ReplicationSlotOnDiskNotChecksummedSize,
                 ReplicationSlotOnDiskChecksummedSize);
     FIN_CRC32C(cp.checksum);
  
     errno = 0;
     pgstat_report_wait_start(WAIT_EVENT_REPLICATION_SLOT_WRITE);
     if ((write(fd, &cp, sizeof(cp))) != sizeof(cp))
     {
         int         save_errno = errno;
  
         pgstat_report_wait_end();
         CloseTransientFile(fd);
         LWLockRelease(&slot->io_in_progress_lock);
  
         /* if write didn't set errno, assume problem is no disk space */
         errno = save_errno ? save_errno : ENOSPC;
         ereport(elevel,
                 (errcode_for_file_access(),
                  errmsg("could not write to file \"%s\": %m",
                         tmppath)));
         return;
     }
     pgstat_report_wait_end();
  
     /* fsync the temporary file */
     pgstat_report_wait_start(WAIT_EVENT_REPLICATION_SLOT_SYNC);
     if (pg_fsync(fd) != 0)
     {
         int         save_errno = errno;
  
         pgstat_report_wait_end();
         CloseTransientFile(fd);
         LWLockRelease(&slot->io_in_progress_lock);
         errno = save_errno;
         ereport(elevel,
                 (errcode_for_file_access(),
                  errmsg("could not fsync file \"%s\": %m",
                         tmppath)));
         return;
     }
     pgstat_report_wait_end();
  
     if (CloseTransientFile(fd) != 0)
     {
         int         save_errno = errno;
  
         LWLockRelease(&slot->io_in_progress_lock);
         errno = save_errno;
         ereport(elevel,
                 (errcode_for_file_access(),
                  errmsg("could not close file \"%s\": %m",
                         tmppath)));
         return;
     }
  
     /* rename to permanent file, fsync file and directory */
     if (rename(tmppath, path) != 0)
     {
         int         save_errno = errno;
  
         LWLockRelease(&slot->io_in_progress_lock);
         errno = save_errno;
         ereport(elevel,
                 (errcode_for_file_access(),
                  errmsg("could not rename file \"%s\" to \"%s\": %m",
                         tmppath, path)));
         return;
     }
  
     /*
      * Check CreateSlotOnDisk() for the reasoning of using a critical section.
      */
     START_CRIT_SECTION();
  
     fsync_fname(path, false);
     fsync_fname(dir, true);
     fsync_fname("pg_replslot", true);
  
     END_CRIT_SECTION();
  
     /*
      * Successfully wrote, unset dirty bit, unless somebody dirtied again
      * already.
      */
     SpinLockAcquire(&slot->mutex);
     if (!slot->just_dirtied)
         slot->dirty = false;
     SpinLockRelease(&slot->mutex);
  
     LWLockRelease(&slot->io_in_progress_lock);
 }

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

◆ SearchNamedReplicationSlot()

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

Definition at line 374 of file slot.c.

 {
     int         i;
     ReplicationSlot *slot = NULL;
  
     if (need_lock)
         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)
         {
             slot = s;
             break;
         }
     }
  
     if (need_lock)
         LWLockRelease(ReplicationSlotControlLock);
  
     return slot;
 }

References ReplicationSlot::data, i, ReplicationSlot::in_use, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_replication_slots, name, ReplicationSlotPersistentData::name, NameStr, ReplicationSlotCtlData::replication_slots, and ReplicationSlotCtl.

Referenced by get_replslot_index(), pg_ls_replslotdir(), pgstat_reset_replslot(), ReadReplicationSlot(), and ReplicationSlotAcquire().

◆ StartupReplicationSlots()

void StartupReplicationSlots ( void )

Definition at line 1477 of file slot.c.

 {
     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)
     {
         char        path[MAXPGPATH + 12];
         PGFileType  de_type;
  
         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);
         de_type = get_dirent_type(path, replication_de, false, DEBUG1);
  
         /* we're only creating directories here, skip if it's not our's */
         if (de_type != PGFILETYPE_ERROR && de_type != PGFILETYPE_DIR)
             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,
                         (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();
 }

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

Referenced by StartupXLOG().

Variable Documentation

◆ max_replication_slots

int max_replication_slots = 10

Definition at line 101 of file slot.c.

Referenced by CheckPointReplicationOrigin(), CheckPointReplicationSlots(), CheckSlotRequirements(), copy_replication_slot(), InvalidateObsoleteReplicationSlots(), logicalrep_worker_launch(), pg_get_replication_slots(), pg_show_replication_origin_status(), ReplicationOriginShmemInit(), ReplicationOriginShmemSize(), ReplicationSlotCleanup(), ReplicationSlotCreate(), ReplicationSlotIndex(), ReplicationSlotsComputeLogicalRestartLSN(), ReplicationSlotsComputeRequiredLSN(), ReplicationSlotsComputeRequiredXmin(), ReplicationSlotsCountDBSlots(), ReplicationSlotsDropDBSlots(), ReplicationSlotsShmemInit(), ReplicationSlotsShmemSize(), replorigin_advance(), replorigin_check_prerequisites(), replorigin_drop_guts(), replorigin_get_progress(), replorigin_redo(), replorigin_session_reset(), replorigin_session_setup(), RestoreSlotFromDisk(), SearchNamedReplicationSlot(), StartupReplicationOrigin(), and StartupReplicationSlots().

◆ MyReplicationSlot

ReplicationSlot* MyReplicationSlot = NULL

Definition at line 98 of file slot.c.

Referenced by copy_replication_slot(), create_logical_replication_slot(), create_physical_replication_slot(), CreateDecodingContext(), CreateInitDecodingContext(), CreateReplicationSlot(), InvalidatePossiblyObsoleteSlot(), LogicalConfirmReceivedLocation(), LogicalIncreaseRestartDecodingForSlot(), LogicalIncreaseXminForSlot(), pg_create_logical_replication_slot(), pg_create_physical_replication_slot(), pg_logical_replication_slot_advance(), pg_logical_slot_get_changes_guts(), pg_physical_replication_slot_advance(), pg_replication_slot_advance(), PhysicalConfirmReceivedLocation(), PhysicalReplicationSlotNewXmin(), PostgresMain(), ProcessStandbyHSFeedbackMessage(), ProcessStandbyReplyMessage(), ReorderBufferAllocate(), ReorderBufferFree(), ReorderBufferRestoreChanges(), ReorderBufferRestoreCleanup(), ReorderBufferSerializedPath(), ReorderBufferSerializeTXN(), ReplicationSlotAcquire(), ReplicationSlotCleanup(), ReplicationSlotCreate(), ReplicationSlotDrop(), ReplicationSlotDropAcquired(), ReplicationSlotMarkDirty(), ReplicationSlotPersist(), ReplicationSlotRelease(), ReplicationSlotReserveWal(), ReplicationSlotSave(), ReplicationSlotsDropDBSlots(), ReplicationSlotShmemExit(), StartLogicalReplication(), StartReplication(), StartupDecodingContext(), and WalSndErrorCleanup().

◆ ReplicationSlotCtl

ReplicationSlotCtlData* ReplicationSlotCtl = NULL

Definition at line 95 of file slot.c.

Referenced by CheckPointReplicationSlots(), copy_replication_slot(), InvalidateObsoleteReplicationSlots(), pg_get_replication_slots(), ReplicationSlotCleanup(), ReplicationSlotCreate(), ReplicationSlotIndex(), ReplicationSlotName(), ReplicationSlotsComputeLogicalRestartLSN(), ReplicationSlotsComputeRequiredLSN(), ReplicationSlotsComputeRequiredXmin(), ReplicationSlotsCountDBSlots(), ReplicationSlotsDropDBSlots(), ReplicationSlotsShmemInit(), RestoreSlotFromDisk(), and SearchNamedReplicationSlot().

Data Structures

Macros

Typedefs

Functions

Variables

Macro Definition Documentation

◆ ReplicationSlotOnDiskChecksummedSize

◆ ReplicationSlotOnDiskConstantSize

◆ ReplicationSlotOnDiskNotChecksummedSize

◆ ReplicationSlotOnDiskV2Size

◆ SLOT_MAGIC

◆ SLOT_VERSION

Typedef Documentation

◆ ReplicationSlotOnDisk

Function Documentation

◆ CheckPointReplicationSlots()

◆ CheckSlotPermissions()

◆ CheckSlotRequirements()

◆ CreateSlotOnDisk()

◆ InvalidateObsoleteReplicationSlots()

◆ InvalidatePossiblyObsoleteSlot()

◆ ReplicationSlotAcquire()

◆ ReplicationSlotCleanup()

◆ ReplicationSlotCreate()

◆ ReplicationSlotDrop()

◆ ReplicationSlotDropAcquired()

◆ ReplicationSlotDropPtr()

◆ ReplicationSlotIndex()

◆ ReplicationSlotInitialize()

◆ ReplicationSlotMarkDirty()

◆ ReplicationSlotName()

◆ ReplicationSlotPersist()

◆ ReplicationSlotRelease()

◆ ReplicationSlotReserveWal()

◆ ReplicationSlotSave()

◆ ReplicationSlotsComputeLogicalRestartLSN()

◆ ReplicationSlotsComputeRequiredLSN()

◆ ReplicationSlotsComputeRequiredXmin()

◆ ReplicationSlotsCountDBSlots()

◆ ReplicationSlotsDropDBSlots()

◆ ReplicationSlotShmemExit()

◆ ReplicationSlotsShmemInit()

◆ ReplicationSlotsShmemSize()

◆ ReplicationSlotValidateName()

◆ RestoreSlotFromDisk()

◆ SaveSlotToPath()

◆ SearchNamedReplicationSlot()

◆ StartupReplicationSlots()

Variable Documentation

◆ max_replication_slots

◆ MyReplicationSlot

◆ ReplicationSlotCtl