database/postgres
1
0
Fork 0
mirror of https://github.com/postgres/postgres.git synced 2026-10-23 13:29:09 +03:00
Code Activity

Convert Sharedsort's worker counters to atomic variables.

Browse Source 
Currently, currentWorker and workersFinished are ints protected by
a spinlock.  By converting them to atomic variables, we can remove
the spinlock.  Note that the spinlock also covered the write of
each worker's tape metadata.  Removing it is still okay: each
worker writes only its own slot, and the atomic increment that
follows is a full barrier.  Furthermore, the leader doesn't read
the slot values until all workers have finished.

Reviewed-by: Zsolt Parragi <zsolt.parragi@percona.com>
Tested-by: solai v <solai.cdac@gmail.com>
Discussion: https://postgr.es/m/alAJeRRzehDjLaF1%40nathan
This commit is contained in:
Nathan Bossart
2026-09-22 13:50:58 -05:00
parent cb4a03db9e
commit ee97eef695
1 changed files with 8 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
src/backend/utils/sort/tuplesort.c
View File

@@ -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;
@@ -3254,9 +3252,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++)
@@ -3293,16 +3290,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);
}
/*
@@ -3344,10 +3334,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);
}
/*
@@ -3389,9 +3377,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");