From dd9f2db260c8e95f434cf23ab508db55a54c8acb Mon Sep 17 00:00:00 2001 From: Nathan Bossart 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)