I've now committed everything except for these last two patches, which I'm planning to commit tomorrow.
-- nathan
>From 8438305333b38a31c101f0051cc8b86aaec1e087 Mon Sep 17 00:00:00 2001 From: Nathan Bossart <[email protected]> Date: Tue, 22 Sep 2026 14:31:14 -0500 Subject: [PATCH v4 1/2] Use unsigned integers for sinval message numbers. Currently, the message numbers in sinvaladt.c are ints, but they are never negative, and the code already takes pains to keep them from overflowing. This commit changes them to uint32. The only wrinkle is that SICleanupQueue() computes two thresholds by subtracting from maxMsgNum, and those could previously go negative. They are now clamped at zero, which disables the corresponding checks just as a negative threshold did. This is preparatory work for a follow-up commit that will convert maxMsgNum to an unsigned atomic variable. Author: Yura Sokolov <[email protected]> Reviewed-by: Heikki Linnakangas <[email protected]> Reviewed-by: Peter Eisentraut <[email protected]> Reviewed-by: Andres Freund <[email protected]> Discussion: https://postgr.es/m/30aa0030-f694-44ef-a19d-6ef7ddb69374%40postgrespro.ru Discussion: https://postgr.es/m/alAJeRRzehDjLaF1%40nathan --- src/backend/storage/ipc/sinvaladt.c | 32 ++++++++++++++++------------- 1 file changed, 18 insertions(+), 14 deletions(-) diff --git a/src/backend/storage/ipc/sinvaladt.c b/src/backend/storage/ipc/sinvaladt.c index 37a21ffaf1a..b29b4bcc5be 100644 --- a/src/backend/storage/ipc/sinvaladt.c +++ b/src/backend/storage/ipc/sinvaladt.c @@ -93,7 +93,7 @@ * read maxMsgNum if you are not holding SInvalWriteLock, and you need the * spinlock to write maxMsgNum unless you are holding both locks.) * - * Note: since maxMsgNum is an int and hence presumably atomically readable/ + * Note: since maxMsgNum is a uint32 and hence presumably atomically readable/ * writable, the spinlock might seem unnecessary. The reason it is needed * is to provide a memory barrier: we need to be sure that messages written * to the array are actually there before maxMsgNum is increased, and that @@ -140,7 +140,7 @@ typedef struct ProcState /* procPid is zero in an inactive ProcState array entry. */ pid_t procPid; /* PID of backend, for signaling */ /* nextMsgNum is meaningless if procPid == 0 or resetState is true. */ - int nextMsgNum; /* next message number to read */ + uint32 nextMsgNum; /* next message number to read */ bool resetState; /* backend needs to reset its state */ bool signaled; /* backend has been sent catchup signal */ bool hasMessages; /* backend has unread messages */ @@ -168,9 +168,9 @@ typedef struct SISeg /* * General state information */ - int minMsgNum; /* oldest message still needed */ - int maxMsgNum; /* next message number to be assigned */ - int nextThreshold; /* # of messages to call SICleanupQueue */ + uint32 minMsgNum; /* oldest message still needed */ + uint32 maxMsgNum; /* next message number to be assigned */ + uint32 nextThreshold; /* # of messages to call SICleanupQueue */ slock_t msgnumLock; /* spinlock protecting maxMsgNum */ @@ -385,8 +385,8 @@ SIInsertDataEntries(const SharedInvalidationMessage *data, int n) while (n > 0) { int nthistime = Min(n, WRITE_QUANTUM); - int numMsgs; - int max; + uint32 numMsgs; + uint32 max; int i; n -= nthistime; @@ -476,7 +476,7 @@ SIGetDataEntries(SharedInvalidationMessage *data, int datasize) { SISeg *segP; ProcState *stateP; - int max; + uint32 max; int n; segP = shmInvalBuffer; @@ -579,11 +579,11 @@ void SICleanupQueue(bool callerHasWriteLock, int minFree) { SISeg *segP = shmInvalBuffer; - int min, + uint32 min, minsig, lowbound, - numMsgs, - i; + numMsgs; + int i; ProcState *needSig = NULL; /* Lock out all writers and readers */ @@ -597,15 +597,19 @@ SICleanupQueue(bool callerHasWriteLock, int minFree) * backends that are too far back. Note that because we ignore sendOnly * backends here it is possible for them to keep sending messages without * a problem even when they are the only active backend. + * + * Note that the thresholds are clamped at zero rather than allowed to + * wrap around. */ min = segP->maxMsgNum; - minsig = min - SIG_THRESHOLD; - lowbound = min - MAXNUMMESSAGES + minFree; + minsig = (min > SIG_THRESHOLD) ? min - SIG_THRESHOLD : 0; + lowbound = (min + minFree > MAXNUMMESSAGES) ? + min + minFree - MAXNUMMESSAGES : 0; for (i = 0; i < segP->numProcs; i++) { ProcState *stateP = &segP->procState[segP->pgprocnos[i]]; - int n = stateP->nextMsgNum; + uint32 n = stateP->nextMsgNum; /* Ignore if already in reset state */ Assert(stateP->procPid != 0); -- 2.55.0
>From 9903d1cdc01f12acb510ffa8f22ddb1048ec1f48 Mon Sep 17 00:00:00 2001 From: Nathan Bossart <[email protected]> Date: Tue, 22 Sep 2026 14:36:42 -0500 Subject: [PATCH v4 2/2] Convert SISeg->maxMsgNum to an atomic variable. Currently, this variable is a uint32 protected by a spinlock. The spinlock exists only to provide memory barriers, so by converting the variable to an atomic and using the barrier-providing accessors in the spinlock's place, we can remove the spinlock. Author: Yura Sokolov <[email protected]> Reviewed-by: Heikki Linnakangas <[email protected]> Reviewed-by: Peter Eisentraut <[email protected]> Reviewed-by: Andres Freund <[email protected]> Reviewed-by: Zsolt Parragi <[email protected]> Tested-by: solai v <[email protected]> Discussion: https://postgr.es/m/30aa0030-f694-44ef-a19d-6ef7ddb69374%40postgrespro.ru Discussion: https://postgr.es/m/alAJeRRzehDjLaF1%40nathan --- src/backend/storage/ipc/sinvaladt.c | 51 ++++++++++------------------- 1 file changed, 17 insertions(+), 34 deletions(-) diff --git a/src/backend/storage/ipc/sinvaladt.c b/src/backend/storage/ipc/sinvaladt.c index b29b4bcc5be..bc5f9537710 100644 --- a/src/backend/storage/ipc/sinvaladt.c +++ b/src/backend/storage/ipc/sinvaladt.c @@ -24,7 +24,6 @@ #include "storage/procsignal.h" #include "storage/shmem.h" #include "storage/sinvaladt.h" -#include "storage/spin.h" #include "storage/subsystems.h" /* @@ -87,19 +86,10 @@ * has no need to touch anyone's ProcState, except in the infrequent cases * when SICleanupQueue is needed. The only point of overlap is that * the writer wants to change maxMsgNum while readers need to read it. - * We deal with that by having a spinlock that readers must take for just - * long enough to read maxMsgNum, while writers take it for just long enough - * to write maxMsgNum. (The exact rule is that you need the spinlock to - * read maxMsgNum if you are not holding SInvalWriteLock, and you need the - * spinlock to write maxMsgNum unless you are holding both locks.) - * - * Note: since maxMsgNum is a uint32 and hence presumably atomically readable/ - * writable, the spinlock might seem unnecessary. The reason it is needed - * is to provide a memory barrier: we need to be sure that messages written - * to the array are actually there before maxMsgNum is increased, and that - * readers will see that data after fetching maxMsgNum. Multiprocessors - * that have weak memory-ordering guarantees can fail without the memory - * barrier instructions that are included in the spinlock sequences. + * We deal with that by making maxMsgNum an atomic variable. (The exact rule + * is that you need to use a barrier-providing accessor to read maxMsgNum if + * you are not holding SInvalWriteLock, and you need a barrier-providing + * accessor to write maxMsgNum unless you are holding both locks.) */ @@ -169,11 +159,9 @@ typedef struct SISeg * General state information */ uint32 minMsgNum; /* oldest message still needed */ - uint32 maxMsgNum; /* next message number to be assigned */ + pg_atomic_uint32 maxMsgNum; /* next message number to be assigned */ uint32 nextThreshold; /* # of messages to call SICleanupQueue */ - slock_t msgnumLock; /* spinlock protecting maxMsgNum */ - /* * Circular buffer holding shared-inval messages */ @@ -244,11 +232,10 @@ SharedInvalShmemInit(void *arg) { int i; - /* Clear message counters, init spinlock */ + /* Clear message counters */ shmInvalBuffer->minMsgNum = 0; - shmInvalBuffer->maxMsgNum = 0; + pg_atomic_init_u32(&shmInvalBuffer->maxMsgNum, 0); shmInvalBuffer->nextThreshold = CLEANUP_MIN; - SpinLockInit(&shmInvalBuffer->msgnumLock); /* The buffer[] array is initially all unused, so we need not fill it */ @@ -306,7 +293,7 @@ SharedInvalBackendInit(bool sendOnly) /* mark myself active, with all extant messages already read */ stateP->procPid = MyProcPid; - stateP->nextMsgNum = segP->maxMsgNum; + stateP->nextMsgNum = pg_atomic_read_u32(&segP->maxMsgNum); stateP->resetState = false; stateP->signaled = false; stateP->hasMessages = false; @@ -402,7 +389,7 @@ SIInsertDataEntries(const SharedInvalidationMessage *data, int n) */ for (;;) { - numMsgs = segP->maxMsgNum - segP->minMsgNum; + numMsgs = pg_atomic_read_u32(&segP->maxMsgNum) - segP->minMsgNum; if (numMsgs + nthistime > MAXNUMMESSAGES || numMsgs >= segP->nextThreshold) SICleanupQueue(true, nthistime); @@ -413,17 +400,15 @@ SIInsertDataEntries(const SharedInvalidationMessage *data, int n) /* * Insert new message(s) into proper slot of circular buffer */ - max = segP->maxMsgNum; + max = pg_atomic_read_u32(&segP->maxMsgNum); while (nthistime-- > 0) { segP->buffer[max % MAXNUMMESSAGES] = *data++; max++; } - /* Update current value of maxMsgNum using spinlock */ - SpinLockAcquire(&segP->msgnumLock); - segP->maxMsgNum = max; - SpinLockRelease(&segP->msgnumLock); + /* Update current value of maxMsgNum using barrier */ + pg_atomic_write_membarrier_u32(&segP->maxMsgNum, max); /* * Now that the maxMsgNum change is globally visible, we give everyone @@ -509,10 +494,8 @@ SIGetDataEntries(SharedInvalidationMessage *data, int datasize) */ stateP->hasMessages = false; - /* Fetch current value of maxMsgNum using spinlock */ - SpinLockAcquire(&segP->msgnumLock); - max = segP->maxMsgNum; - SpinLockRelease(&segP->msgnumLock); + /* Fetch current value of maxMsgNum using barrier */ + max = pg_atomic_read_membarrier_u32(&segP->maxMsgNum); if (stateP->resetState) { @@ -601,7 +584,7 @@ SICleanupQueue(bool callerHasWriteLock, int minFree) * Note that the thresholds are clamped at zero rather than allowed to * wrap around. */ - min = segP->maxMsgNum; + min = pg_atomic_read_u32(&segP->maxMsgNum); minsig = (min > SIG_THRESHOLD) ? min - SIG_THRESHOLD : 0; lowbound = (min + minFree > MAXNUMMESSAGES) ? min + minFree - MAXNUMMESSAGES : 0; @@ -648,7 +631,7 @@ SICleanupQueue(bool callerHasWriteLock, int minFree) if (min >= MSGNUMWRAPAROUND) { segP->minMsgNum -= MSGNUMWRAPAROUND; - segP->maxMsgNum -= MSGNUMWRAPAROUND; + pg_atomic_fetch_sub_u32(&segP->maxMsgNum, MSGNUMWRAPAROUND); for (i = 0; i < segP->numProcs; i++) segP->procState[segP->pgprocnos[i]].nextMsgNum -= MSGNUMWRAPAROUND; } @@ -657,7 +640,7 @@ SICleanupQueue(bool callerHasWriteLock, int minFree) * Determine how many messages are still in the queue, and set the * threshold at which we should repeat SICleanupQueue(). */ - numMsgs = segP->maxMsgNum - segP->minMsgNum; + numMsgs = pg_atomic_read_u32(&segP->maxMsgNum) - segP->minMsgNum; if (numMsgs < CLEANUP_MIN) segP->nextThreshold = CLEANUP_MIN; else -- 2.55.0
