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

Include dependency graph for slot.c:

Data Structures
struct	ReplicationSlotOnDisk

Macros
#define	ReplicationSlotOnDiskConstantSize offsetof(ReplicationSlotOnDisk, slotdata)

#define	SnapBuildOnDiskNotChecksummedSize offsetof(ReplicationSlotOnDisk, version)

#define	SnapBuildOnDiskChecksummedSize sizeof(ReplicationSlotOnDisk) - SnapBuildOnDiskNotChecksummedSize

#define	ReplicationSlotOnDiskV2Size sizeof(ReplicationSlotOnDisk) - ReplicationSlotOnDiskConstantSize

#define	SLOT_MAGIC 0x1051CA1 /* format identifier */

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

Typedefs
typedef struct ReplicationSlotOnDisk	ReplicationSlotOnDisk

Functions
static void	ReplicationSlotDropAcquired (void)

static void	ReplicationSlotDropPtr (ReplicationSlot *slot)

static void	RestoreSlotFromDisk (const char *name)

static void	CreateSlotOnDisk (ReplicationSlot *slot)

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

Size	ReplicationSlotsShmemSize (void)

void	ReplicationSlotsShmemInit (void)

bool	ReplicationSlotValidateName (const char *name, int elevel)

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

void	ReplicationSlotAcquire (const char *name, bool nowait)

void	ReplicationSlotRelease (void)

void	ReplicationSlotCleanup (void)

void	ReplicationSlotDrop (const char *name, bool nowait)

void	ReplicationSlotSave (void)

void	ReplicationSlotMarkDirty (void)

void	ReplicationSlotPersist (void)

void	ReplicationSlotsComputeRequiredXmin (bool already_locked)

void	ReplicationSlotsComputeRequiredLSN (void)

XLogRecPtr	ReplicationSlotsComputeLogicalRestartLSN (void)

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

void	ReplicationSlotsDropDBSlots (Oid dboid)

void	CheckSlotRequirements (void)

void	ReplicationSlotReserveWal (void)

void	CheckPointReplicationSlots (void)

void	StartupReplicationSlots (void)

Variables
ReplicationSlotCtlData *	ReplicationSlotCtl = NULL

ReplicationSlot *	MyReplicationSlot = NULL

int	max_replication_slots = 0

Macro Definition Documentation

#define ReplicationSlotOnDiskConstantSize offsetof(ReplicationSlotOnDisk, slotdata)

Definition at line 77 of file slot.c.

Referenced by RestoreSlotFromDisk().

#define ReplicationSlotOnDiskV2Size sizeof(ReplicationSlotOnDisk) - ReplicationSlotOnDiskConstantSize

Definition at line 86 of file slot.c.

Referenced by RestoreSlotFromDisk(), and SaveSlotToPath().

#define SLOT_MAGIC 0x1051CA1 /* format identifier */

Definition at line 89 of file slot.c.

Referenced by RestoreSlotFromDisk(), and SaveSlotToPath().

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

Definition at line 90 of file slot.c.

Referenced by RestoreSlotFromDisk(), and SaveSlotToPath().

#define SnapBuildOnDiskChecksummedSize sizeof(ReplicationSlotOnDisk) - SnapBuildOnDiskNotChecksummedSize

Definition at line 83 of file slot.c.

Referenced by RestoreSlotFromDisk(), and SaveSlotToPath().

#define SnapBuildOnDiskNotChecksummedSize offsetof(ReplicationSlotOnDisk, version)

Definition at line 80 of file slot.c.

Referenced by RestoreSlotFromDisk(), and SaveSlotToPath().

Typedef Documentation

typedef struct ReplicationSlotOnDisk ReplicationSlotOnDisk

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

static void CreateSlotOnDisk ( ReplicationSlot * slot )

static

Definition at line 1150 of file slot.c.

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

Referenced by ReplicationSlotCreate().

 {
     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 mkdir() 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 (mkdir(tmppath, S_IRWXU) < 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();
 }

void ReplicationSlotAcquire	(	const char *	name,
		bool	nowait
	)

Definition at line 330 of file slot.c.

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

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

 {
     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, NULL, 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	persistency
	)

Definition at line 221 of file slot.c.

References ReplicationSlot::active_cv, ReplicationSlot::active_pid, Assert, ReplicationSlot::candidate_catalog_xmin, ReplicationSlot::candidate_restart_lsn, ReplicationSlot::candidate_restart_valid, ReplicationSlot::candidate_xmin_lsn, ConditionVariableBroadcast(), CreateSlotOnDisk(), ReplicationSlot::data, ReplicationSlotPersistentData::database, ReplicationSlot::dirty, ReplicationSlot::effective_catalog_xmin, ReplicationSlot::effective_xmin, ereport, errcode(), ERRCODE_DUPLICATE_OBJECT, errhint(), errmsg(), ERROR, i, ReplicationSlot::in_use, InvalidOid, InvalidTransactionId, InvalidXLogRecPtr, ReplicationSlot::just_dirtied, LW_EXCLUSIVE, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_replication_slots, ReplicationSlot::mutex, MyDatabaseId, MyProcPid, ReplicationSlotPersistentData::name, NAMEDATALEN, NameStr, NULL, ReplicationSlotPersistentData::persistency, ReplicationSlotCtlData::replication_slots, ReplicationSlotValidateName(), SpinLockAcquire, SpinLockRelease, and StrNCpy.

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

 {
     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, NULL, ReplicationSlotAcquire(), and ReplicationSlotDropAcquired().

Referenced by DropReplicationSlot(), and pg_drop_replication_slot().

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

static void ReplicationSlotDropAcquired ( void )

static

Definition at line 522 of file slot.c.

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

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

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

static void ReplicationSlotDropPtr ( ReplicationSlot * slot )

static

Definition at line 539 of file slot.c.

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

Referenced by ReplicationSlotCleanup(), and ReplicationSlotDropAcquired().

 {
     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,
                 (errcode_for_file_access(),
                  errmsg("could not remove directory \"%s\"", tmppath)));
 
     /*
      * 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);
 }

void ReplicationSlotMarkDirty ( void )

Definition at line 656 of file slot.c.

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

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

 {
     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, NULL, 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, NULL, ReplicationSlotPersistentData::persistency, PROC_IN_LOGICAL_DECODING, ReplicationSlotDropAcquired(), ReplicationSlotsComputeRequiredXmin(), RS_EPHEMERAL, RS_PERSISTENT, SpinLockAcquire, SpinLockRelease, TransactionIdIsValid, PGXACT::vacuumFlags, and ReplicationSlotPersistentData::xmin.

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

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

static void RestoreSlotFromDisk ( const char * name )

static

Definition at line 1334 of file slot.c.

References ReplicationSlot::active_pid, ReplicationSlot::candidate_catalog_xmin, ReplicationSlot::candidate_restart_lsn, ReplicationSlot::candidate_restart_valid, ReplicationSlot::candidate_xmin_lsn, ReplicationSlotPersistentData::catalog_xmin, ReplicationSlotOnDisk::checksum, CloseTransientFile(), COMP_CRC32C, ReplicationSlot::data, DEBUG1, ReplicationSlot::effective_catalog_xmin, ReplicationSlot::effective_xmin, elog, END_CRIT_SECTION, EQ_CRC32C, ereport, errcode_for_file_access(), errhint(), errmsg(), fd(), FIN_CRC32C, fsync_fname(), i, ReplicationSlot::in_use, INIT_CRC32C, InvalidTransactionId, InvalidXLogRecPtr, ReplicationSlotOnDisk::length, ReplicationSlotOnDisk::magic, max_replication_slots, MAXPGPATH, OpenTransientFile(), PANIC, ReplicationSlotPersistentData::persistency, PG_BINARY, pg_fsync(), pgstat_report_wait_end(), pgstat_report_wait_start(), read, ReplicationSlotCtlData::replication_slots, ReplicationSlotOnDiskConstantSize, ReplicationSlotOnDiskV2Size, rmtree(), RS_PERSISTENT, SLOT_MAGIC, SLOT_VERSION, ReplicationSlotOnDisk::slotdata, SnapBuildOnDiskChecksummedSize, SnapBuildOnDiskNotChecksummedSize, START_CRIT_SECTION, unlink(), ReplicationSlotOnDisk::version, WAIT_EVENT_REPLICATION_SLOT_READ, WAIT_EVENT_REPLICATION_SLOT_RESTORE_SYNC, WARNING, and ReplicationSlotPersistentData::xmin.

Referenced by StartupReplicationSlots().

 {
     ReplicationSlotOnDisk cp;
     int         i;
     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(path, "pg_replslot/%s/state.tmp", name);
     if (unlink(path) < 0 && errno != ENOENT)
         ereport(PANIC,
                 (errcode_for_file_access(),
                  errmsg("could not remove file \"%s\": %m", path)));
 
     sprintf(path, "pg_replslot/%s/state", name);
 
     elog(DEBUG1, "restoring replication slot from \"%s\"", path);
 
     fd = OpenTransientFile(path, O_RDWR | PG_BINARY, 0);
 
     /*
      * 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)
     {
         CloseTransientFile(fd);
         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(path, 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)
     {
         int         saved_errno = errno;
 
         CloseTransientFile(fd);
         errno = saved_errno;
         ereport(PANIC,
                 (errcode_for_file_access(),
                  errmsg("could not read file \"%s\", read %d of %u: %m",
                         path, readBytes,
                         (uint32) ReplicationSlotOnDiskConstantSize)));
     }
 
     /* verify magic */
     if (cp.magic != SLOT_MAGIC)
         ereport(PANIC,
                 (errcode_for_file_access(),
                  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_for_file_access(),
                  errmsg("replication slot file \"%s\" has unsupported version %u",
                         path, cp.version)));
 
     /* boundary check on length */
     if (cp.length != ReplicationSlotOnDiskV2Size)
         ereport(PANIC,
                 (errcode_for_file_access(),
                  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)
     {
         int         saved_errno = errno;
 
         CloseTransientFile(fd);
         errno = saved_errno;
         ereport(PANIC,
                 (errcode_for_file_access(),
                  errmsg("could not read file \"%s\", read %d of %u: %m",
                         path, readBytes, cp.length)));
     }
 
     CloseTransientFile(fd);
 
     /* now verify the CRC */
     INIT_CRC32C(checksum);
     COMP_CRC32C(checksum,
                 (char *) &cp + SnapBuildOnDiskNotChecksummedSize,
                 SnapBuildOnDiskChecksummedSize);
     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)
     {
         sprintf(path, "pg_replslot/%s", name);
 
         if (!rmtree(path, true))
         {
             ereport(WARNING,
                     (errcode_for_file_access(),
                      errmsg("could not remove directory \"%s\"", path)));
         }
         fsync_fname("pg_replslot", true);
         return;
     }
 
     /* 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(PANIC,
                 (errmsg("too many replication slots active before shutdown"),
                  errhint("Increase max_replication_slots and try again.")));
 }

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

static

Definition at line 1210 of file slot.c.

References ReplicationSlotOnDisk::checksum, CloseTransientFile(), COMP_CRC32C, ReplicationSlot::data, ReplicationSlot::dirty, END_CRIT_SECTION, ereport, errcode_for_file_access(), errmsg(), fd(), FIN_CRC32C, fsync_fname(), INIT_CRC32C, ReplicationSlot::io_in_progress_lock, ReplicationSlot::just_dirtied, ReplicationSlotOnDisk::length, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), ReplicationSlotOnDisk::magic, MAXPGPATH, ReplicationSlot::mutex, OpenTransientFile(), PG_BINARY, pg_fsync(), pgstat_report_wait_end(), pgstat_report_wait_start(), ReplicationSlotOnDiskV2Size, SLOT_MAGIC, SLOT_VERSION, ReplicationSlotOnDisk::slotdata, SnapBuildOnDiskChecksummedSize, SnapBuildOnDiskNotChecksummedSize, SpinLockAcquire, SpinLockRelease, START_CRIT_SECTION, ReplicationSlotOnDisk::version, WAIT_EVENT_REPLICATION_SLOT_SYNC, WAIT_EVENT_REPLICATION_SLOT_WRITE, and write.

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

 {
     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,
                            S_IRUSR | S_IWUSR);
     if (fd < 0)
     {
         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) + SnapBuildOnDiskNotChecksummedSize,
                 SnapBuildOnDiskChecksummedSize);
     FIN_CRC32C(cp.checksum);
 
     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);
         errno = save_errno;
         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);
         errno = save_errno;
         ereport(elevel,
                 (errcode_for_file_access(),
                  errmsg("could not fsync file \"%s\": %m",
                         tmppath)));
         return;
     }
     pgstat_report_wait_end();
 
     CloseTransientFile(fd);
 
     /* rename to permanent file, fsync file and directory */
     if (rename(tmppath, path) != 0)
     {
         ereport(elevel,
                 (errcode_for_file_access(),
                  errmsg("could not rename file \"%s\" to \"%s\": %m",
                         tmppath, path)));
         return;
     }
 
     /* Check CreateSlot() for the reasoning of using a crit. 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);
 }

void StartupReplicationSlots ( void )

Definition at line 1090 of file slot.c.

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

Referenced by StartupXLOG().

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