upstream/postgresql
1
0
Fork 0
mirror of https://github.com/postgres/postgres.git synced 2026-07-16 05:02:57 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

Use ProcNumber rather than pid in ReplicationSlot

Browse source 
This helps the next commit.

Reviewed-by: Chao Li <li.evan.chao@gmail.com>
Discussion: https://www.postgresql.org/message-id/4cc13ba1-4248-4884-b6ba-4805349e7f39@iki.fi
This commit is contained in:
Heikki Linnakangas 2026-02-10 16:23:05 +02:00
parent f33c585774
commit ddc3250208
4 changed files with 49 additions and 36 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
src/backend/replication/logical/slotsync.c
View file

@ -1757,7 +1757,7 @@ update_synced_slots_inactive_since(void)
Assert(SlotIsLogical(s)); Assert(SlotIsLogical(s));
/* The slot must not be acquired by any process */ /* The slot must not be acquired by any process */
Assert(s->active_pid == 0); Assert(s->active_proc == INVALID_PROC_NUMBER);
/* Use the same inactive_since time for all the slots. */ /* Use the same inactive_since time for all the slots. */
if (now == 0) if (now == 0)

63
src/backend/replication/slot.c
View file

@ -226,6 +226,7 @@ ReplicationSlotsShmemInit(void)
ReplicationSlot *slot = &ReplicationSlotCtl->replication_slots[i]; ReplicationSlot *slot = &ReplicationSlotCtl->replication_slots[i];
/* everything else is zeroed by the memset above */ /* everything else is zeroed by the memset above */
slot->active_proc = INVALID_PROC_NUMBER;
SpinLockInit(&slot->mutex); SpinLockInit(&slot->mutex);
LWLockInitialize(&slot->io_in_progress_lock, LWLockInitialize(&slot->io_in_progress_lock,
LWTRANCHE_REPLICATION_SLOT_IO); LWTRANCHE_REPLICATION_SLOT_IO);
@ -461,7 +462,7 @@ ReplicationSlotCreate(const char *name, bool db_specific,
* be doing that. So it's safe to initialize the slot. * be doing that. So it's safe to initialize the slot.
*/ */
Assert(!slot->in_use); Assert(!slot->in_use);
Assert(slot->active_pid == 0); Assert(slot->active_proc == INVALID_PROC_NUMBER);
/* first initialize persistent data */ /* first initialize persistent data */
memset(&slot->data, 0, sizeof(ReplicationSlotPersistentData)); memset(&slot->data, 0, sizeof(ReplicationSlotPersistentData));
@ -505,8 +506,8 @@ ReplicationSlotCreate(const char *name, bool db_specific,
/* We can now mark the slot active, and that makes it our slot. */ /* We can now mark the slot active, and that makes it our slot. */
SpinLockAcquire(&slot->mutex); SpinLockAcquire(&slot->mutex);
Assert(slot->active_pid == 0); Assert(slot->active_proc == INVALID_PROC_NUMBER);
slot->active_pid = MyProcPid; slot->active_proc = MyProcNumber;
SpinLockRelease(&slot->mutex); SpinLockRelease(&slot->mutex);
MyReplicationSlot = slot; MyReplicationSlot = slot;
@ -620,6 +621,7 @@ void
ReplicationSlotAcquire(const char *name, bool nowait, bool error_if_invalid) ReplicationSlotAcquire(const char *name, bool nowait, bool error_if_invalid)
{ {
ReplicationSlot *s; ReplicationSlot *s;
ProcNumber active_proc;
int active_pid; int active_pid;
Assert(name != NULL); Assert(name != NULL);
@ -672,17 +674,18 @@ retry:
* to inactive_since in InvalidatePossiblyObsoleteSlot. * to inactive_since in InvalidatePossiblyObsoleteSlot.
*/ */
SpinLockAcquire(&s->mutex); SpinLockAcquire(&s->mutex);
if (s->active_pid == 0) if (s->active_proc == INVALID_PROC_NUMBER)
s->active_pid = MyProcPid; s->active_proc = MyProcNumber;
active_pid = s->active_pid; active_proc = s->active_proc;
ReplicationSlotSetInactiveSince(s, 0, false); ReplicationSlotSetInactiveSince(s, 0, false);
SpinLockRelease(&s->mutex); SpinLockRelease(&s->mutex);
} }
else else
{ {
s->active_pid = active_pid = MyProcPid; s->active_proc = active_proc = MyProcNumber;
ReplicationSlotSetInactiveSince(s, 0, true); ReplicationSlotSetInactiveSince(s, 0, true);
} }
active_pid = GetPGProcByNumber(active_proc)->pid;
LWLockRelease(ReplicationSlotControlLock); LWLockRelease(ReplicationSlotControlLock);
/* /*
@ -690,7 +693,7 @@ retry:
* wait until the owning process signals us that it's been released, or * wait until the owning process signals us that it's been released, or
* error out. * error out.
*/ */
if (active_pid != MyProcPid) if (active_proc != MyProcNumber)
{ {
if (!nowait) if (!nowait)
{ {
@ -762,7 +765,7 @@ ReplicationSlotRelease(void)
bool is_logical; bool is_logical;
TimestampTz now = 0; TimestampTz now = 0;
Assert(slot != NULL && slot->active_pid != 0); Assert(slot != NULL && slot->active_proc != INVALID_PROC_NUMBER);
is_logical = SlotIsLogical(slot); is_logical = SlotIsLogical(slot);
@ -815,7 +818,7 @@ ReplicationSlotRelease(void)
* disconnecting, but wake up others that may be waiting for it. * disconnecting, but wake up others that may be waiting for it.
*/ */
SpinLockAcquire(&slot->mutex); SpinLockAcquire(&slot->mutex);
slot->active_pid = 0; slot->active_proc = INVALID_PROC_NUMBER;
ReplicationSlotSetInactiveSince(slot, now, false); ReplicationSlotSetInactiveSince(slot, now, false);
SpinLockRelease(&slot->mutex); SpinLockRelease(&slot->mutex);
ConditionVariableBroadcast(&slot->active_cv); ConditionVariableBroadcast(&slot->active_cv);
@ -877,7 +880,7 @@ restart:
found_valid_logicalslot |= found_valid_logicalslot |=
(SlotIsLogical(s) && s->data.invalidated == RS_INVAL_NONE); (SlotIsLogical(s) && s->data.invalidated == RS_INVAL_NONE);
if ((s->active_pid == MyProcPid && if ((s->active_proc == MyProcNumber &&
(!synced_only || s->data.synced))) (!synced_only || s->data.synced)))
{ {
Assert(s->data.persistency == RS_TEMPORARY); Assert(s->data.persistency == RS_TEMPORARY);
@ -1088,7 +1091,7 @@ ReplicationSlotDropPtr(ReplicationSlot *slot)
bool fail_softly = slot->data.persistency != RS_PERSISTENT; bool fail_softly = slot->data.persistency != RS_PERSISTENT;
SpinLockAcquire(&slot->mutex); SpinLockAcquire(&slot->mutex);
slot->active_pid = 0; slot->active_proc = INVALID_PROC_NUMBER;
SpinLockRelease(&slot->mutex); SpinLockRelease(&slot->mutex);
/* wake up anyone waiting on this slot */ /* wake up anyone waiting on this slot */
@ -1110,7 +1113,7 @@ ReplicationSlotDropPtr(ReplicationSlot *slot)
* Also wake up processes waiting for it. * Also wake up processes waiting for it.
*/ */
LWLockAcquire(ReplicationSlotControlLock, LW_EXCLUSIVE); LWLockAcquire(ReplicationSlotControlLock, LW_EXCLUSIVE);
slot->active_pid = 0; slot->active_proc = INVALID_PROC_NUMBER;
slot->in_use = false; slot->in_use = false;
LWLockRelease(ReplicationSlotControlLock); LWLockRelease(ReplicationSlotControlLock);
ConditionVariableBroadcast(&slot->active_cv); ConditionVariableBroadcast(&slot->active_cv);
@ -1476,7 +1479,7 @@ ReplicationSlotsCountDBSlots(Oid dboid, int *nslots, int *nactive)
/* count slots with spinlock held */ /* count slots with spinlock held */
SpinLockAcquire(&s->mutex); SpinLockAcquire(&s->mutex);
(*nslots)++; (*nslots)++;
if (s->active_pid != 0) if (s->active_proc != INVALID_PROC_NUMBER)
(*nactive)++; (*nactive)++;
SpinLockRelease(&s->mutex); SpinLockRelease(&s->mutex);
} }
@ -1520,7 +1523,7 @@ restart:
{ {
ReplicationSlot *s; ReplicationSlot *s;
char *slotname; char *slotname;
int active_pid; ProcNumber active_proc;
s = &ReplicationSlotCtl->replication_slots[i]; s = &ReplicationSlotCtl->replication_slots[i];
@ -1550,11 +1553,11 @@ restart:
SpinLockAcquire(&s->mutex); SpinLockAcquire(&s->mutex);
/* can't change while ReplicationSlotControlLock is held */ /* can't change while ReplicationSlotControlLock is held */
slotname = NameStr(s->data.name); slotname = NameStr(s->data.name);
active_pid = s->active_pid; active_proc = s->active_proc;
if (active_pid == 0) if (active_proc == INVALID_PROC_NUMBER)
{ {
MyReplicationSlot = s; MyReplicationSlot = s;
s->active_pid = MyProcPid; s->active_proc = MyProcNumber;
} }
SpinLockRelease(&s->mutex); SpinLockRelease(&s->mutex);
@ -1579,11 +1582,11 @@ restart:
* XXX: We can consider shutting down the slot sync worker before * XXX: We can consider shutting down the slot sync worker before
* trying to drop synced temporary slots here. * trying to drop synced temporary slots here.
*/ */
if (active_pid) if (active_proc != INVALID_PROC_NUMBER)
ereport(ERROR, ereport(ERROR,
(errcode(ERRCODE_OBJECT_IN_USE), (errcode(ERRCODE_OBJECT_IN_USE),
errmsg("replication slot \"%s\" is active for PID %d", errmsg("replication slot \"%s\" is active for PID %d",
slotname, active_pid))); slotname, GetPGProcByNumber(active_proc)->pid)));
/* /*
* To avoid duplicating ReplicationSlotDropAcquired() and to avoid * To avoid duplicating ReplicationSlotDropAcquired() and to avoid
@ -1974,6 +1977,7 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes,
{ {
XLogRecPtr restart_lsn; XLogRecPtr restart_lsn;
NameData slotname; NameData slotname;
ProcNumber active_proc;
int active_pid = 0; int active_pid = 0;
ReplicationSlotInvalidationCause invalidation_cause = RS_INVAL_NONE; ReplicationSlotInvalidationCause invalidation_cause = RS_INVAL_NONE;
TimestampTz now = 0; TimestampTz now = 0;
@ -2027,7 +2031,7 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes,
} }
slotname = s->data.name; slotname = s->data.name;
active_pid = s->active_pid; active_proc = s->active_proc;
/* /*
* If the slot can be acquired, do so and mark it invalidated * If the slot can be acquired, do so and mark it invalidated
@ -2039,10 +2043,10 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes,
* is terminated. So, the inactive slot can only be invalidated * is terminated. So, the inactive slot can only be invalidated
* immediately without being terminated. * immediately without being terminated.
*/ */
if (active_pid == 0) if (active_proc == INVALID_PROC_NUMBER)
{ {
MyReplicationSlot = s; MyReplicationSlot = s;
s->active_pid = MyProcPid; s->active_proc = MyProcNumber;
s->data.invalidated = invalidation_cause; s->data.invalidated = invalidation_cause;
/* /*
@ -2058,6 +2062,11 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes,
/* Let caller know */ /* Let caller know */
invalidated = true; invalidated = true;
} }
else
{
active_pid = GetPGProcByNumber(active_proc)->pid;
Assert(active_pid != 0);
}
SpinLockRelease(&s->mutex); SpinLockRelease(&s->mutex);
@ -2073,7 +2082,7 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes,
&slot_idle_usecs); &slot_idle_usecs);
} }
if (active_pid != 0) if (active_proc != INVALID_PROC_NUMBER)
{ {
/* /*
* Prepare the sleep on the slot's condition variable before * Prepare the sleep on the slot's condition variable before
@ -2107,7 +2116,7 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes,
if (MyBackendType == B_STARTUP) if (MyBackendType == B_STARTUP)
(void) SendProcSignal(active_pid, (void) SendProcSignal(active_pid,
PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT, PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT,
INVALID_PROC_NUMBER); active_proc);
else else
(void) kill(active_pid, SIGTERM); (void) kill(active_pid, SIGTERM);
@ -2875,7 +2884,7 @@ RestoreSlotFromDisk(const char *name)
slot->candidate_restart_valid = InvalidXLogRecPtr; slot->candidate_restart_valid = InvalidXLogRecPtr;
slot->in_use = true; slot->in_use = true;
slot->active_pid = 0; slot->active_proc = INVALID_PROC_NUMBER;
/* /*
* Set the time since the slot has become inactive after loading the * Set the time since the slot has become inactive after loading the
@ -3158,7 +3167,7 @@ StandbySlotsHaveCaughtup(XLogRecPtr wait_for_lsn, int elevel)
SpinLockAcquire(&slot->mutex); SpinLockAcquire(&slot->mutex);
restart_lsn = slot->data.restart_lsn; restart_lsn = slot->data.restart_lsn;
invalidated = slot->data.invalidated != RS_INVAL_NONE; invalidated = slot->data.invalidated != RS_INVAL_NONE;
inactive = slot->active_pid == 0; inactive = slot->active_proc == INVALID_PROC_NUMBER;
SpinLockRelease(&slot->mutex); SpinLockRelease(&slot->mutex);
if (invalidated) if (invalidated)

13
src/backend/replication/slotfuncs.c
View file

@ -20,6 +20,7 @@
#include "replication/logical.h" #include "replication/logical.h"
#include "replication/slot.h" #include "replication/slot.h"
#include "replication/slotsync.h" #include "replication/slotsync.h"
#include "storage/proc.h"
#include "utils/builtins.h" #include "utils/builtins.h"
#include "utils/guc.h" #include "utils/guc.h"
#include "utils/pg_lsn.h" #include "utils/pg_lsn.h"
@ -309,10 +310,10 @@ pg_get_replication_slots(PG_FUNCTION_ARGS)
values[i++] = ObjectIdGetDatum(slot_contents.data.database); values[i++] = ObjectIdGetDatum(slot_contents.data.database);
values[i++] = BoolGetDatum(slot_contents.data.persistency == RS_TEMPORARY); values[i++] = BoolGetDatum(slot_contents.data.persistency == RS_TEMPORARY);
values[i++] = BoolGetDatum(slot_contents.active_pid != 0); values[i++] = BoolGetDatum(slot_contents.active_proc != INVALID_PROC_NUMBER);
if (slot_contents.active_pid != 0) if (slot_contents.active_proc != INVALID_PROC_NUMBER)
values[i++] = Int32GetDatum(slot_contents.active_pid); values[i++] = Int32GetDatum(GetPGProcByNumber(slot_contents.active_proc)->pid);
else else
nulls[i++] = true; nulls[i++] = true;
@ -377,13 +378,13 @@ pg_get_replication_slots(PG_FUNCTION_ARGS)
*/ */
if (XLogRecPtrIsValid(slot_contents.data.restart_lsn)) if (XLogRecPtrIsValid(slot_contents.data.restart_lsn))
{ {
int pid; ProcNumber procno;
SpinLockAcquire(&slot->mutex); SpinLockAcquire(&slot->mutex);
pid = slot->active_pid; procno = slot->active_proc;
slot_contents.data.restart_lsn = slot->data.restart_lsn; slot_contents.data.restart_lsn = slot->data.restart_lsn;
SpinLockRelease(&slot->mutex); SpinLockRelease(&slot->mutex);
if (pid != 0) if (procno != INVALID_PROC_NUMBER)
{ {
values[i++] = CStringGetTextDatum("unreserved"); values[i++] = CStringGetTextDatum("unreserved");
walstate = WALAVAIL_UNRESERVED; walstate = WALAVAIL_UNRESERVED;

7
src/include/replication/slot.h
View file

@ -185,8 +185,11 @@ typedef struct ReplicationSlot
/* is this slot defined */ /* is this slot defined */
bool in_use; bool in_use;
/* Who is streaming out changes for this slot? 0 in unused slots. */ /*
pid_t active_pid; * Who is streaming out changes for this slot? INVALID_PROC_NUMBER in
* unused slots.
*/
ProcNumber active_proc;
/* any outstanding modifications? */ /* any outstanding modifications? */
bool just_dirtied; bool just_dirtied;