The attached patch set converts various variables to atomics, thereby allowing us to remove a handful of spinlocks and volatile qualifiers. We've been slowly moving in this direction for a while already. I think all of these are pretty straightforward and easy to reason about.
-- nathan
>From 69b00b2d955e4098a7c17f82e9f7f4c1d6e9e0f4 Mon Sep 17 00:00:00 2001 From: Nathan Bossart <[email protected]> Date: Thu, 9 Jul 2026 14:20:37 -0500 Subject: [PATCH v1 1/8] convert SISeg->maxMsgNum to an atomic --- 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 37a21ffaf1a..17e98c9efdc 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 an int 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 */ int minMsgNum; /* oldest message still needed */ - int maxMsgNum; /* next message number to be assigned */ + pg_atomic_uint32 maxMsgNum; /* next message number to be assigned */ int 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) { @@ -598,7 +581,7 @@ SICleanupQueue(bool callerHasWriteLock, int minFree) * backends here it is possible for them to keep sending messages without * a problem even when they are the only active backend. */ - min = segP->maxMsgNum; + min = pg_atomic_read_u32(&segP->maxMsgNum); minsig = min - SIG_THRESHOLD; lowbound = min - MAXNUMMESSAGES + minFree; @@ -644,7 +627,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; } @@ -653,7 +636,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.50.1 (Apple Git-155)
>From 9ae1ed9879d4a0242ba084ae41d741d1c7aa5097 Mon Sep 17 00:00:00 2001 From: Nathan Bossart <[email protected]> Date: Thu, 9 Jul 2026 13:38:33 -0500 Subject: [PATCH v1 2/8] convert ParallelBitmapHeapState->state to an atomic --- src/backend/executor/nodeBitmapHeapscan.c | 21 ++++++--------------- 1 file changed, 6 insertions(+), 15 deletions(-) diff --git a/src/backend/executor/nodeBitmapHeapscan.c b/src/backend/executor/nodeBitmapHeapscan.c index 83d6478bc2b..f2bb487bf10 100644 --- a/src/backend/executor/nodeBitmapHeapscan.c +++ b/src/backend/executor/nodeBitmapHeapscan.c @@ -79,7 +79,6 @@ typedef enum /* ---------------- * ParallelBitmapHeapState information * tbmiterator iterator for scanning current pages - * mutex mutual exclusion for state * state current state of the TIDBitmap * cv conditional wait variable * ---------------- @@ -87,8 +86,7 @@ typedef enum typedef struct ParallelBitmapHeapState { dsa_pointer tbmiterator; - slock_t mutex; - SharedBitmapState state; + pg_atomic_uint32 state; ConditionVariable cv; } ParallelBitmapHeapState; @@ -228,9 +226,7 @@ BitmapHeapNext(BitmapHeapScanState *node) static inline void BitmapDoneInitializingSharedState(ParallelBitmapHeapState *pstate) { - SpinLockAcquire(&pstate->mutex); - pstate->state = BM_FINISHED; - SpinLockRelease(&pstate->mutex); + pg_atomic_write_membarrier_u32(&pstate->state, BM_FINISHED); ConditionVariableBroadcast(&pstate->cv); } @@ -480,11 +476,8 @@ BitmapShouldInitializeSharedState(ParallelBitmapHeapState *pstate) while (1) { - SpinLockAcquire(&pstate->mutex); - state = pstate->state; - if (pstate->state == BM_INITIAL) - pstate->state = BM_INPROGRESS; - SpinLockRelease(&pstate->mutex); + state = BM_INITIAL; + pg_atomic_compare_exchange_u32(&pstate->state, &state, BM_INPROGRESS); /* Exit if bitmap is done, or if we're the leader. */ if (state != BM_INPROGRESS) @@ -538,9 +531,7 @@ ExecBitmapHeapInitializeDSM(BitmapHeapScanState *node, pstate->tbmiterator = 0; - /* Initialize the mutex */ - SpinLockInit(&pstate->mutex); - pstate->state = BM_INITIAL; + pg_atomic_init_u32(&pstate->state, BM_INITIAL); ConditionVariableInit(&pstate->cv); @@ -565,7 +556,7 @@ ExecBitmapHeapReInitializeDSM(BitmapHeapScanState *node, if (dsa == NULL) return; - pstate->state = BM_INITIAL; + pg_atomic_write_u32(&pstate->state, BM_INITIAL); if (DsaPointerIsValid(pstate->tbmiterator)) tbm_free_shared_area(dsa, pstate->tbmiterator); -- 2.50.1 (Apple Git-155)
>From 52c99655fe9025d47b001593ac45fe0de1fdd54a Mon Sep 17 00:00:00 2001 From: Nathan Bossart <[email protected]> Date: Thu, 9 Jul 2026 13:57:14 -0500 Subject: [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic --- src/backend/access/transam/parallel.c | 22 ++++++++-------------- 1 file changed, 8 insertions(+), 14 deletions(-) diff --git a/src/backend/access/transam/parallel.c b/src/backend/access/transam/parallel.c index 89e9d224eec..57f5f26bbec 100644 --- a/src/backend/access/transam/parallel.c +++ b/src/backend/access/transam/parallel.c @@ -37,7 +37,6 @@ #include "storage/ipc.h" #include "storage/predicate.h" #include "storage/proc.h" -#include "storage/spin.h" #include "tcop/tcopprot.h" #include "utils/combocid.h" #include "utils/guc.h" @@ -101,11 +100,8 @@ typedef struct FixedParallelState TimestampTz stmt_ts; SerializableXactHandle serializable_xact_handle; - /* Mutex protects remaining fields. */ - slock_t mutex; - /* Maximum XactLastRecEnd of any worker. */ - XLogRecPtr last_xlog_end; + pg_atomic_uint64 last_xlog_end; } FixedParallelState; /* @@ -358,8 +354,7 @@ InitializeParallelDSM(ParallelContext *pcxt) fps->xact_ts = GetCurrentTransactionStartTimestamp(); fps->stmt_ts = GetCurrentStatementStartTimestamp(); fps->serializable_xact_handle = ShareSerializableXact(); - SpinLockInit(&fps->mutex); - fps->last_xlog_end = InvalidXLogRecPtr; + pg_atomic_init_u64(&fps->last_xlog_end, InvalidXLogRecPtr); shm_toc_insert(pcxt->toc, PARALLEL_KEY_FIXED, fps); /* We can skip the rest of this if we're not budgeting for any workers. */ @@ -532,7 +527,7 @@ ReinitializeParallelDSM(ParallelContext *pcxt) /* Reset a few bits of fixed parallel state to a clean state. */ fps = shm_toc_lookup(pcxt->toc, PARALLEL_KEY_FIXED, false); - fps->last_xlog_end = InvalidXLogRecPtr; + pg_atomic_write_u64(&fps->last_xlog_end, InvalidXLogRecPtr); /* Recreate error queues (if they exist). */ if (pcxt->nworkers > 0) @@ -900,10 +895,12 @@ WaitForParallelWorkersToFinish(ParallelContext *pcxt) if (pcxt->toc != NULL) { FixedParallelState *fps; + XLogRecPtr last_xlog_end; fps = shm_toc_lookup(pcxt->toc, PARALLEL_KEY_FIXED, false); - if (fps->last_xlog_end > XactLastRecEnd) - XactLastRecEnd = fps->last_xlog_end; + last_xlog_end = pg_atomic_read_u64(&fps->last_xlog_end); + if (last_xlog_end > XactLastRecEnd) + XactLastRecEnd = last_xlog_end; } } @@ -1596,10 +1593,7 @@ ParallelWorkerReportLastRecEnd(XLogRecPtr last_xlog_end) FixedParallelState *fps = MyFixedParallelState; Assert(fps != NULL); - SpinLockAcquire(&fps->mutex); - if (fps->last_xlog_end < last_xlog_end) - fps->last_xlog_end = last_xlog_end; - SpinLockRelease(&fps->mutex); + pg_atomic_monotonic_advance_u64(&fps->last_xlog_end, last_xlog_end); } /* -- 2.50.1 (Apple Git-155)
>From 08b71cf2c1b71191ed1ab4c9b12aedc63e98e9e2 Mon Sep 17 00:00:00 2001 From: Nathan Bossart <[email protected]> Date: Thu, 9 Jul 2026 14:26:49 -0500 Subject: [PATCH v1 4/8] convert PROC_HDR->startupBufferPinWaitBufId to an atomic --- src/backend/storage/lmgr/proc.c | 12 +++--------- src/include/storage/proc.h | 2 +- 2 files changed, 4 insertions(+), 10 deletions(-) diff --git a/src/backend/storage/lmgr/proc.c b/src/backend/storage/lmgr/proc.c index 9d6e69175a5..5f314efb289 100644 --- a/src/backend/storage/lmgr/proc.c +++ b/src/backend/storage/lmgr/proc.c @@ -239,7 +239,7 @@ ProcGlobalShmemInit(void *arg) dlist_init(&ProcGlobal->autovacFreeProcs); dlist_init(&ProcGlobal->bgworkerFreeProcs); dlist_init(&ProcGlobal->walsenderFreeProcs); - ProcGlobal->startupBufferPinWaitBufId = -1; + pg_atomic_init_u32(&ProcGlobal->startupBufferPinWaitBufId, -1); pg_atomic_init_u32(&ProcGlobal->avLauncherProc, INVALID_PROC_NUMBER); pg_atomic_init_u32(&ProcGlobal->walwriterProc, INVALID_PROC_NUMBER); pg_atomic_init_u32(&ProcGlobal->checkpointerProc, INVALID_PROC_NUMBER); @@ -768,10 +768,7 @@ InitAuxiliaryProcess(void) void SetStartupBufferPinWaitBufId(int bufid) { - /* use volatile pointer to prevent code rearrangement */ - volatile PROC_HDR *procglobal = ProcGlobal; - - procglobal->startupBufferPinWaitBufId = bufid; + pg_atomic_write_u32(&ProcGlobal->startupBufferPinWaitBufId, bufid); } /* @@ -780,10 +777,7 @@ SetStartupBufferPinWaitBufId(int bufid) int GetStartupBufferPinWaitBufId(void) { - /* use volatile pointer to prevent code rearrangement */ - volatile PROC_HDR *procglobal = ProcGlobal; - - return procglobal->startupBufferPinWaitBufId; + return pg_atomic_read_u32(&ProcGlobal->startupBufferPinWaitBufId); } /* diff --git a/src/include/storage/proc.h b/src/include/storage/proc.h index 03a1a466fa8..e2034255124 100644 --- a/src/include/storage/proc.h +++ b/src/include/storage/proc.h @@ -499,7 +499,7 @@ typedef struct PROC_HDR /* Current shared estimate of appropriate spins_per_delay value */ int spins_per_delay; /* Buffer id of the buffer that Startup process waits for pin on, or -1 */ - int startupBufferPinWaitBufId; + pg_atomic_uint32 startupBufferPinWaitBufId; } PROC_HDR; extern PGDLLIMPORT PROC_HDR *ProcGlobal; -- 2.50.1 (Apple Git-155)
>From dd9f2db260c8e95f434cf23ab508db55a54c8acb Mon Sep 17 00:00:00 2001 From: Nathan Bossart <[email protected]> Date: Thu, 9 Jul 2026 15:04:43 -0500 Subject: [PATCH v1 5/8] convert Sharedsort->{currentWorker,workersFinished} to atomics --- src/backend/utils/sort/tuplesort.c | 30 ++++++++---------------------- 1 file changed, 8 insertions(+), 22 deletions(-) diff --git a/src/backend/utils/sort/tuplesort.c b/src/backend/utils/sort/tuplesort.c index c0e7527b9ca..81e0b2816d6 100644 --- a/src/backend/utils/sort/tuplesort.c +++ b/src/backend/utils/sort/tuplesort.c @@ -104,6 +104,7 @@ #include "commands/tablespace.h" #include "miscadmin.h" #include "pg_trace.h" +#include "port/atomics.h" #include "port/pg_bitutils.h" #include "storage/shmem.h" #include "utils/guc.h" @@ -340,9 +341,6 @@ struct Tuplesortstate */ struct Sharedsort { - /* mutex protects all fields prior to tapes */ - slock_t mutex; - /* * currentWorker generates ordinal identifier numbers for parallel sort * workers. These start from 0, and are always gapless. @@ -351,8 +349,8 @@ struct Sharedsort * is equal to state.nParticipants within the leader, leader is ready to * merge worker runs. */ - int currentWorker; - int workersFinished; + pg_atomic_uint32 currentWorker; + pg_atomic_uint32 workersFinished; /* Temporary file space */ SharedFileSet fileset; @@ -3252,9 +3250,8 @@ tuplesort_initialize_shared(Sharedsort *shared, int nWorkers, dsm_segment *seg) Assert(nWorkers > 0); - SpinLockInit(&shared->mutex); - shared->currentWorker = 0; - shared->workersFinished = 0; + pg_atomic_init_u32(&shared->currentWorker, 0); + pg_atomic_init_u32(&shared->workersFinished, 0); SharedFileSetInit(&shared->fileset, seg); shared->nTapes = nWorkers; for (i = 0; i < nWorkers; i++) @@ -3291,16 +3288,9 @@ tuplesort_attach_shared(Sharedsort *shared, dsm_segment *seg) static int worker_get_identifier(Tuplesortstate *state) { - Sharedsort *shared = state->shared; - int worker; - Assert(WORKER(state)); - SpinLockAcquire(&shared->mutex); - worker = shared->currentWorker++; - SpinLockRelease(&shared->mutex); - - return worker; + return pg_atomic_fetch_add_u32(&state->shared->currentWorker, 1); } /* @@ -3342,10 +3332,8 @@ worker_freeze_result_tape(Tuplesortstate *state) LogicalTapeFreeze(state->result_tape, &output); /* Store properties of output tape, and update finished worker count */ - SpinLockAcquire(&shared->mutex); shared->tapes[state->worker] = output; - shared->workersFinished++; - SpinLockRelease(&shared->mutex); + pg_atomic_fetch_add_u32(&shared->workersFinished, 1); } /* @@ -3387,9 +3375,7 @@ leader_takeover_tapes(Tuplesortstate *state) Assert(LEADER(state)); Assert(nParticipants >= 1); - SpinLockAcquire(&shared->mutex); - workersFinished = shared->workersFinished; - SpinLockRelease(&shared->mutex); + workersFinished = pg_atomic_read_membarrier_u32(&shared->workersFinished); if (nParticipants != workersFinished) elog(ERROR, "cannot take over tapes before all workers finish"); -- 2.50.1 (Apple Git-155)
>From d273a3f06068dd9e63478f6c126d2702d318510a Mon Sep 17 00:00:00 2001 From: Nathan Bossart <[email protected]> Date: Thu, 9 Jul 2026 14:53:49 -0500 Subject: [PATCH v1 6/8] convert SharedFileSet->refcnt to an atomic --- src/backend/storage/file/sharedfileset.c | 27 +++++++++--------------- src/include/storage/sharedfileset.h | 5 ++--- 2 files changed, 12 insertions(+), 20 deletions(-) diff --git a/src/backend/storage/file/sharedfileset.c b/src/backend/storage/file/sharedfileset.c index d76bd72dc63..4f12f92beae 100644 --- a/src/backend/storage/file/sharedfileset.c +++ b/src/backend/storage/file/sharedfileset.c @@ -38,8 +38,7 @@ void SharedFileSetInit(SharedFileSet *fileset, dsm_segment *seg) { /* Initialize the shared fileset specific members. */ - SpinLockInit(&fileset->mutex); - fileset->refcnt = 1; + pg_atomic_init_u32(&fileset->refcnt, 1); /* Initialize the fileset. */ FileSetInit(&fileset->fs); @@ -55,19 +54,15 @@ SharedFileSetInit(SharedFileSet *fileset, dsm_segment *seg) void SharedFileSetAttach(SharedFileSet *fileset, dsm_segment *seg) { - bool success; + uint32 refcnt; - SpinLockAcquire(&fileset->mutex); - if (fileset->refcnt == 0) - success = false; - else - { - ++fileset->refcnt; - success = true; - } - SpinLockRelease(&fileset->mutex); + refcnt = pg_atomic_read_u32(&fileset->refcnt); + while (refcnt != 0 && + !pg_atomic_compare_exchange_u32(&fileset->refcnt, &refcnt, + refcnt + 1)) + ; - if (!success) + if (refcnt == 0) ereport(ERROR, (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), errmsg("could not attach to a SharedFileSet that is already destroyed"))); @@ -98,11 +93,9 @@ SharedFileSetOnDetach(dsm_segment *segment, Datum datum) bool unlink_all = false; SharedFileSet *fileset = (SharedFileSet *) DatumGetPointer(datum); - SpinLockAcquire(&fileset->mutex); - Assert(fileset->refcnt > 0); - if (--fileset->refcnt == 0) + Assert(pg_atomic_read_u32(&fileset->refcnt) > 0); + if (pg_atomic_sub_fetch_u32(&fileset->refcnt, 1) == 0) unlink_all = true; - SpinLockRelease(&fileset->mutex); /* * If we are the last to detach, we delete the directory in all diff --git a/src/include/storage/sharedfileset.h b/src/include/storage/sharedfileset.h index 904396e7173..d89626ae64b 100644 --- a/src/include/storage/sharedfileset.h +++ b/src/include/storage/sharedfileset.h @@ -15,10 +15,10 @@ #ifndef SHAREDFILESET_H #define SHAREDFILESET_H +#include "port/atomics.h" #include "storage/dsm.h" #include "storage/fd.h" #include "storage/fileset.h" -#include "storage/spin.h" /* * A set of temporary files that can be shared by multiple backends. @@ -26,8 +26,7 @@ typedef struct SharedFileSet { FileSet fs; - slock_t mutex; /* mutex protecting the reference count */ - int refcnt; /* number of attached backends */ + pg_atomic_uint32 refcnt; /* number of attached backends */ } SharedFileSet; extern void SharedFileSetInit(SharedFileSet *fileset, dsm_segment *seg); -- 2.50.1 (Apple Git-155)
>From ff987247ff84460f5aa23deb41a9fd0a3149bd85 Mon Sep 17 00:00:00 2001 From: Nathan Bossart <[email protected]> Date: Thu, 9 Jul 2026 15:21:13 -0500 Subject: [PATCH v1 7/8] convert ParallelBlockTableScanDescData->phs_{start,num}block to atomics --- src/backend/access/heap/heapam_handler.c | 2 +- src/backend/access/table/tableam.c | 59 +++++++++++------------- src/include/access/relscan.h | 8 ++-- 3 files changed, 31 insertions(+), 38 deletions(-) diff --git a/src/backend/access/heap/heapam_handler.c b/src/backend/access/heap/heapam_handler.c index bf87430cf01..0f24a132564 100644 --- a/src/backend/access/heap/heapam_handler.c +++ b/src/backend/access/heap/heapam_handler.c @@ -1965,7 +1965,7 @@ heapam_scan_get_blocks_done(HeapScanDesc hscan) if (hscan->rs_base.rs_parallel != NULL) { bpscan = (ParallelBlockTableScanDesc) hscan->rs_base.rs_parallel; - startblock = bpscan->phs_startblock; + startblock = pg_atomic_read_u32(&bpscan->phs_startblock); } else startblock = hscan->rs_startblock; diff --git a/src/backend/access/table/tableam.c b/src/backend/access/table/tableam.c index 68ff0966f1c..f2038ea9205 100644 --- a/src/backend/access/table/tableam.c +++ b/src/backend/access/table/tableam.c @@ -421,9 +421,8 @@ table_block_parallelscan_initialize(Relation rel, ParallelTableScanDesc pscan) bpscan->base.phs_syncscan = synchronize_seqscans && !RelationUsesLocalBuffers(rel) && bpscan->phs_nblocks > NBuffers / 4; - SpinLockInit(&bpscan->phs_mutex); - bpscan->phs_startblock = InvalidBlockNumber; - bpscan->phs_numblock = InvalidBlockNumber; + pg_atomic_init_u32(&bpscan->phs_startblock, InvalidBlockNumber); + pg_atomic_init_u32(&bpscan->phs_numblock, InvalidBlockNumber); pg_atomic_init_u64(&bpscan->phs_nallocated, 0); return sizeof(ParallelBlockTableScanDescData); @@ -459,25 +458,22 @@ table_block_parallelscan_startblock_init(Relation rel, StaticAssertDecl(MaxBlockNumber <= 0xFFFFFFFE, "pg_nextpower2_32 may be too small for non-standard BlockNumber width"); - BlockNumber sync_startpage = InvalidBlockNumber; BlockNumber scan_nblocks; /* Reset the state we use for controlling allocation size. */ memset(pbscanwork, 0, sizeof(*pbscanwork)); -retry: - /* Grab the spinlock. */ - SpinLockAcquire(&pbscan->phs_mutex); - /* * When the caller specified a limit on the number of blocks to scan, set * that in the ParallelBlockTableScanDesc, if it's not been done by * another worker already. */ - if (numblocks != InvalidBlockNumber && - pbscan->phs_numblock == InvalidBlockNumber) + if (numblocks != InvalidBlockNumber) { - pbscan->phs_numblock = numblocks; + uint32 expected = InvalidBlockNumber; + + pg_atomic_compare_exchange_u32(&pbscan->phs_numblock, &expected, + numblocks); } /* @@ -485,36 +481,35 @@ retry: * so now. If a startblock was specified, start there, otherwise if this * is not a synchronized scan, we just start at block 0, but if it is a * synchronized scan, we must get the starting position from the - * synchronized scan machinery. We can't hold the spinlock while doing - * that, though, so release the spinlock, get the information we need, and - * retry. If nobody else has initialized the scan in the meantime, we'll - * fill in the value we fetched on the second time through. + * synchronized scan machinery. + * + * If another worker initializes phs_startblock concurrently, just use + * their value. */ - if (pbscan->phs_startblock == InvalidBlockNumber) + if (pg_atomic_read_u32(&pbscan->phs_startblock) == InvalidBlockNumber) { + BlockNumber newstartblock; + uint32 expected = InvalidBlockNumber; + if (startblock != InvalidBlockNumber) - pbscan->phs_startblock = startblock; + newstartblock = startblock; else if (!pbscan->base.phs_syncscan) - pbscan->phs_startblock = 0; - else if (sync_startpage != InvalidBlockNumber) - pbscan->phs_startblock = sync_startpage; + newstartblock = 0; else - { - SpinLockRelease(&pbscan->phs_mutex); - sync_startpage = ss_get_location(rel, pbscan->phs_nblocks); - goto retry; - } + newstartblock = ss_get_location(rel, pbscan->phs_nblocks); + + pg_atomic_compare_exchange_u32(&pbscan->phs_startblock, &expected, + newstartblock); } - SpinLockRelease(&pbscan->phs_mutex); /* * Figure out how many blocks we're going to scan; either all of them, or * just phs_numblock's worth, if a limit has been imposed. */ - if (pbscan->phs_numblock == InvalidBlockNumber) + if (pg_atomic_read_u32(&pbscan->phs_numblock) == InvalidBlockNumber) scan_nblocks = pbscan->phs_nblocks; else - scan_nblocks = pbscan->phs_numblock; + scan_nblocks = pg_atomic_read_u32(&pbscan->phs_numblock); /* * We determine the chunk size based on scan_nblocks. First we split @@ -595,10 +590,10 @@ table_block_parallelscan_nextpage(Relation rel, */ /* First, figure out how many blocks we're planning on scanning */ - if (pbscan->phs_numblock == InvalidBlockNumber) + if (pg_atomic_read_u32(&pbscan->phs_numblock) == InvalidBlockNumber) scan_nblocks = pbscan->phs_nblocks; else - scan_nblocks = pbscan->phs_numblock; + scan_nblocks = pg_atomic_read_u32(&pbscan->phs_numblock); /* * Now check if we have any remaining blocks in a previous chunk for this @@ -644,7 +639,7 @@ table_block_parallelscan_nextpage(Relation rel, if (nallocated >= scan_nblocks) page = InvalidBlockNumber; /* all blocks have been allocated */ else - page = (nallocated + pbscan->phs_startblock) % pbscan->phs_nblocks; + page = (nallocated + pg_atomic_read_u32(&pbscan->phs_startblock)) % pbscan->phs_nblocks; /* * Report scan location. Normally, we report the current page number. @@ -658,7 +653,7 @@ table_block_parallelscan_nextpage(Relation rel, if (page != InvalidBlockNumber) ss_report_location(rel, page); else if (nallocated == pbscan->phs_nblocks) - ss_report_location(rel, pbscan->phs_startblock); + ss_report_location(rel, pg_atomic_read_u32(&pbscan->phs_startblock)); } return page; diff --git a/src/include/access/relscan.h b/src/include/access/relscan.h index 2ea06a67a63..2305d0159f3 100644 --- a/src/include/access/relscan.h +++ b/src/include/access/relscan.h @@ -19,7 +19,6 @@ #include "nodes/tidbitmap.h" #include "port/atomics.h" #include "storage/relfilelocator.h" -#include "storage/spin.h" #include "utils/relcache.h" @@ -99,10 +98,9 @@ typedef struct ParallelBlockTableScanDescData ParallelTableScanDescData base; BlockNumber phs_nblocks; /* # blocks in relation at start of scan */ - slock_t phs_mutex; /* mutual exclusion for setting startblock */ - BlockNumber phs_startblock; /* starting block number */ - BlockNumber phs_numblock; /* # blocks to scan, or InvalidBlockNumber if - * no limit */ + pg_atomic_uint32 phs_startblock; /* starting block number */ + pg_atomic_uint32 phs_numblock; /* # blocks to scan, or InvalidBlockNumber + * if no limit */ pg_atomic_uint64 phs_nallocated; /* number of blocks allocated to * workers so far. */ } ParallelBlockTableScanDescData; -- 2.50.1 (Apple Git-155)
>From 42fcfca3062352582c12250482e7fe93b14f45e2 Mon Sep 17 00:00:00 2001 From: Nathan Bossart <[email protected]> Date: Thu, 9 Jul 2026 15:38:08 -0500 Subject: [PATCH v1 8/8] convert FastPathStrongRelationLocks to atomics --- src/backend/storage/lmgr/lock.c | 57 ++++++++++----------------------- 1 file changed, 17 insertions(+), 40 deletions(-) diff --git a/src/backend/storage/lmgr/lock.c b/src/backend/storage/lmgr/lock.c index 5ee80c7632e..035932197a4 100644 --- a/src/backend/storage/lmgr/lock.c +++ b/src/backend/storage/lmgr/lock.c @@ -40,11 +40,11 @@ #include "miscadmin.h" #include "pg_trace.h" #include "pgstat.h" +#include "port/atomics.h" #include "storage/lmgr.h" #include "storage/proc.h" #include "storage/procarray.h" #include "storage/shmem.h" -#include "storage/spin.h" #include "storage/standby.h" #include "storage/subsystems.h" #include "utils/memutils.h" @@ -306,13 +306,7 @@ static PROCLOCK *FastPathGetRelationLockEntry(LOCALLOCK *locallock); #define FastPathStrongLockHashPartition(hashcode) \ ((hashcode) % FAST_PATH_STRONG_LOCK_HASH_PARTITIONS) -typedef struct -{ - slock_t mutex; - uint32 count[FAST_PATH_STRONG_LOCK_HASH_PARTITIONS]; -} FastPathStrongRelationLockData; - -static FastPathStrongRelationLockData *FastPathStrongRelationLocks; +static pg_atomic_uint32 *FastPathStrongRelationLocks; static void LockManagerShmemRequest(void *arg); static void LockManagerShmemInit(void *arg); @@ -484,7 +478,8 @@ LockManagerShmemRequest(void *arg) ); ShmemRequestStruct(.name = "Fast Path Strong Relation Lock Data", - .size = sizeof(FastPathStrongRelationLockData), + .size = mul_size(sizeof(pg_atomic_uint32), + FAST_PATH_STRONG_LOCK_HASH_PARTITIONS), .ptr = (void **) (void *) &FastPathStrongRelationLocks, ); } @@ -492,7 +487,8 @@ LockManagerShmemRequest(void *arg) static void LockManagerShmemInit(void *arg) { - SpinLockInit(&FastPathStrongRelationLocks->mutex); + for (int i = 0; i < FAST_PATH_STRONG_LOCK_HASH_PARTITIONS; i++) + pg_atomic_init_u32(&FastPathStrongRelationLocks[i], 0); } /* @@ -992,11 +988,11 @@ LockAcquireExtended(const LOCKTAG *locktag, /* * LWLockAcquire acts as a memory sequencing point, so it's safe * to assume that any strong locker whose increment to - * FastPathStrongRelationLocks->counts becomes visible after we - * test it has yet to begin to transfer fast-path locks. + * FastPathStrongRelationLocks becomes visible after we test it + * has yet to begin to transfer fast-path locks. */ LWLockAcquire(&MyProc->fpInfoLock, LW_EXCLUSIVE); - if (FastPathStrongRelationLocks->count[fasthashcode] != 0) + if (pg_atomic_read_u32(&FastPathStrongRelationLocks[fasthashcode]) != 0) acquired = false; else acquired = FastPathGrantRelationLock(locktag->locktag_field2, @@ -1501,11 +1497,9 @@ RemoveLocalLock(LOCALLOCK *locallock) fasthashcode = FastPathStrongLockHashPartition(locallock->hashcode); - SpinLockAcquire(&FastPathStrongRelationLocks->mutex); - Assert(FastPathStrongRelationLocks->count[fasthashcode] > 0); - FastPathStrongRelationLocks->count[fasthashcode]--; + Assert(pg_atomic_read_u32(&FastPathStrongRelationLocks[fasthashcode]) > 0); + pg_atomic_fetch_sub_u32(&FastPathStrongRelationLocks[fasthashcode], 1); locallock->holdsStrongLockCount = false; - SpinLockRelease(&FastPathStrongRelationLocks->mutex); } if (!hash_search(LockMethodLocalHash, @@ -1834,20 +1828,9 @@ BeginStrongLockAcquire(LOCALLOCK *locallock, uint32 fasthashcode) Assert(StrongLockInProgress == NULL); Assert(locallock->holdsStrongLockCount == false); - /* - * Adding to a memory location is not atomic, so we take a spinlock to - * ensure we don't collide with someone else trying to bump the count at - * the same time. - * - * XXX: It might be worth considering using an atomic fetch-and-add - * instruction here, on architectures where that is supported. - */ - - SpinLockAcquire(&FastPathStrongRelationLocks->mutex); - FastPathStrongRelationLocks->count[fasthashcode]++; + pg_atomic_fetch_add_u32(&FastPathStrongRelationLocks[fasthashcode], 1); locallock->holdsStrongLockCount = true; StrongLockInProgress = locallock; - SpinLockRelease(&FastPathStrongRelationLocks->mutex); } /* @@ -1875,12 +1858,10 @@ AbortStrongLockAcquire(void) fasthashcode = FastPathStrongLockHashPartition(locallock->hashcode); Assert(locallock->holdsStrongLockCount == true); - SpinLockAcquire(&FastPathStrongRelationLocks->mutex); - Assert(FastPathStrongRelationLocks->count[fasthashcode] > 0); - FastPathStrongRelationLocks->count[fasthashcode]--; + Assert(pg_atomic_read_u32(&FastPathStrongRelationLocks[fasthashcode]) > 0); + pg_atomic_fetch_sub_u32(&FastPathStrongRelationLocks[fasthashcode], 1); locallock->holdsStrongLockCount = false; StrongLockInProgress = NULL; - SpinLockRelease(&FastPathStrongRelationLocks->mutex); } /* @@ -3365,10 +3346,8 @@ LockRefindAndRelease(LockMethod lockMethodTable, PGPROC *proc, { uint32 fasthashcode = FastPathStrongLockHashPartition(hashcode); - SpinLockAcquire(&FastPathStrongRelationLocks->mutex); - Assert(FastPathStrongRelationLocks->count[fasthashcode] > 0); - FastPathStrongRelationLocks->count[fasthashcode]--; - SpinLockRelease(&FastPathStrongRelationLocks->mutex); + Assert(pg_atomic_read_u32(&FastPathStrongRelationLocks[fasthashcode]) > 0); + pg_atomic_fetch_sub_u32(&FastPathStrongRelationLocks[fasthashcode], 1); } } @@ -4504,9 +4483,7 @@ lock_twophase_recover(FullTransactionId fxid, uint16 info, { uint32 fasthashcode = FastPathStrongLockHashPartition(hashcode); - SpinLockAcquire(&FastPathStrongRelationLocks->mutex); - FastPathStrongRelationLocks->count[fasthashcode]++; - SpinLockRelease(&FastPathStrongRelationLocks->mutex); + pg_atomic_fetch_add_u32(&FastPathStrongRelationLocks[fasthashcode], 1); } LWLockRelease(partitionLock); -- 2.50.1 (Apple Git-155)
