agora inbox for [email protected]help / color / mirror / Atom feed
[PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic 35+ messages / 1 participants [nested] [flat]
* [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic @ 2026-07-09 18:57 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 35+ messages in thread From: Nathan Bossart @ 2026-07-09 18:57 UTC (permalink / raw) --- 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) --Ot5x3ffkWEuxilWO Content-Type: text/plain; charset=us-ascii Content-Disposition: attachment; filename=v1-0004-convert-PROC_HDR-startupBufferPinWaitBufId-to-an-.patch ^ permalink raw reply [nested|flat] 35+ messages in thread
* [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic @ 2026-07-09 18:57 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 35+ messages in thread From: Nathan Bossart @ 2026-07-09 18:57 UTC (permalink / raw) --- 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) --Ot5x3ffkWEuxilWO Content-Type: text/plain; charset=us-ascii Content-Disposition: attachment; filename=v1-0004-convert-PROC_HDR-startupBufferPinWaitBufId-to-an-.patch ^ permalink raw reply [nested|flat] 35+ messages in thread
* [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic @ 2026-07-09 18:57 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 35+ messages in thread From: Nathan Bossart @ 2026-07-09 18:57 UTC (permalink / raw) --- 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) --Ot5x3ffkWEuxilWO Content-Type: text/plain; charset=us-ascii Content-Disposition: attachment; filename=v1-0004-convert-PROC_HDR-startupBufferPinWaitBufId-to-an-.patch ^ permalink raw reply [nested|flat] 35+ messages in thread
* [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic @ 2026-07-09 18:57 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 35+ messages in thread From: Nathan Bossart @ 2026-07-09 18:57 UTC (permalink / raw) --- 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) --Ot5x3ffkWEuxilWO Content-Type: text/plain; charset=us-ascii Content-Disposition: attachment; filename=v1-0004-convert-PROC_HDR-startupBufferPinWaitBufId-to-an-.patch ^ permalink raw reply [nested|flat] 35+ messages in thread
* [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic @ 2026-07-09 18:57 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 35+ messages in thread From: Nathan Bossart @ 2026-07-09 18:57 UTC (permalink / raw) --- 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) --Ot5x3ffkWEuxilWO Content-Type: text/plain; charset=us-ascii Content-Disposition: attachment; filename=v1-0004-convert-PROC_HDR-startupBufferPinWaitBufId-to-an-.patch ^ permalink raw reply [nested|flat] 35+ messages in thread
* [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic @ 2026-07-09 18:57 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 35+ messages in thread From: Nathan Bossart @ 2026-07-09 18:57 UTC (permalink / raw) --- 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) --Ot5x3ffkWEuxilWO Content-Type: text/plain; charset=us-ascii Content-Disposition: attachment; filename=v1-0004-convert-PROC_HDR-startupBufferPinWaitBufId-to-an-.patch ^ permalink raw reply [nested|flat] 35+ messages in thread
* [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic @ 2026-07-09 18:57 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 35+ messages in thread From: Nathan Bossart @ 2026-07-09 18:57 UTC (permalink / raw) --- 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) --Ot5x3ffkWEuxilWO Content-Type: text/plain; charset=us-ascii Content-Disposition: attachment; filename=v1-0004-convert-PROC_HDR-startupBufferPinWaitBufId-to-an-.patch ^ permalink raw reply [nested|flat] 35+ messages in thread
* [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic @ 2026-07-09 18:57 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 35+ messages in thread From: Nathan Bossart @ 2026-07-09 18:57 UTC (permalink / raw) --- 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) --Ot5x3ffkWEuxilWO Content-Type: text/plain; charset=us-ascii Content-Disposition: attachment; filename=v1-0004-convert-PROC_HDR-startupBufferPinWaitBufId-to-an-.patch ^ permalink raw reply [nested|flat] 35+ messages in thread
* [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic @ 2026-07-09 18:57 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 35+ messages in thread From: Nathan Bossart @ 2026-07-09 18:57 UTC (permalink / raw) --- 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) --Ot5x3ffkWEuxilWO Content-Type: text/plain; charset=us-ascii Content-Disposition: attachment; filename=v1-0004-convert-PROC_HDR-startupBufferPinWaitBufId-to-an-.patch ^ permalink raw reply [nested|flat] 35+ messages in thread
* [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic @ 2026-07-09 18:57 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 35+ messages in thread From: Nathan Bossart @ 2026-07-09 18:57 UTC (permalink / raw) --- 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) --Ot5x3ffkWEuxilWO Content-Type: text/plain; charset=us-ascii Content-Disposition: attachment; filename=v1-0004-convert-PROC_HDR-startupBufferPinWaitBufId-to-an-.patch ^ permalink raw reply [nested|flat] 35+ messages in thread
* [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic @ 2026-07-09 18:57 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 35+ messages in thread From: Nathan Bossart @ 2026-07-09 18:57 UTC (permalink / raw) --- 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) --Ot5x3ffkWEuxilWO Content-Type: text/plain; charset=us-ascii Content-Disposition: attachment; filename=v1-0004-convert-PROC_HDR-startupBufferPinWaitBufId-to-an-.patch ^ permalink raw reply [nested|flat] 35+ messages in thread
* [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic @ 2026-07-09 18:57 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 35+ messages in thread From: Nathan Bossart @ 2026-07-09 18:57 UTC (permalink / raw) --- 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) --Ot5x3ffkWEuxilWO Content-Type: text/plain; charset=us-ascii Content-Disposition: attachment; filename=v1-0004-convert-PROC_HDR-startupBufferPinWaitBufId-to-an-.patch ^ permalink raw reply [nested|flat] 35+ messages in thread
* [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic @ 2026-07-09 18:57 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 35+ messages in thread From: Nathan Bossart @ 2026-07-09 18:57 UTC (permalink / raw) --- 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) --Ot5x3ffkWEuxilWO Content-Type: text/plain; charset=us-ascii Content-Disposition: attachment; filename=v1-0004-convert-PROC_HDR-startupBufferPinWaitBufId-to-an-.patch ^ permalink raw reply [nested|flat] 35+ messages in thread
* [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic @ 2026-07-09 18:57 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 35+ messages in thread From: Nathan Bossart @ 2026-07-09 18:57 UTC (permalink / raw) --- 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) --Ot5x3ffkWEuxilWO Content-Type: text/plain; charset=us-ascii Content-Disposition: attachment; filename=v1-0004-convert-PROC_HDR-startupBufferPinWaitBufId-to-an-.patch ^ permalink raw reply [nested|flat] 35+ messages in thread
* [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic @ 2026-07-09 18:57 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 35+ messages in thread From: Nathan Bossart @ 2026-07-09 18:57 UTC (permalink / raw) --- 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) --Ot5x3ffkWEuxilWO Content-Type: text/plain; charset=us-ascii Content-Disposition: attachment; filename=v1-0004-convert-PROC_HDR-startupBufferPinWaitBufId-to-an-.patch ^ permalink raw reply [nested|flat] 35+ messages in thread
* [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic @ 2026-07-09 18:57 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 35+ messages in thread From: Nathan Bossart @ 2026-07-09 18:57 UTC (permalink / raw) --- 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) --Ot5x3ffkWEuxilWO Content-Type: text/plain; charset=us-ascii Content-Disposition: attachment; filename=v1-0004-convert-PROC_HDR-startupBufferPinWaitBufId-to-an-.patch ^ permalink raw reply [nested|flat] 35+ messages in thread
* [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic @ 2026-07-09 18:57 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 35+ messages in thread From: Nathan Bossart @ 2026-07-09 18:57 UTC (permalink / raw) --- 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) --Ot5x3ffkWEuxilWO Content-Type: text/plain; charset=us-ascii Content-Disposition: attachment; filename=v1-0004-convert-PROC_HDR-startupBufferPinWaitBufId-to-an-.patch ^ permalink raw reply [nested|flat] 35+ messages in thread
* [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic @ 2026-07-09 18:57 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 35+ messages in thread From: Nathan Bossart @ 2026-07-09 18:57 UTC (permalink / raw) --- 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) --Ot5x3ffkWEuxilWO Content-Type: text/plain; charset=us-ascii Content-Disposition: attachment; filename=v1-0004-convert-PROC_HDR-startupBufferPinWaitBufId-to-an-.patch ^ permalink raw reply [nested|flat] 35+ messages in thread
* [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic @ 2026-07-09 18:57 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 35+ messages in thread From: Nathan Bossart @ 2026-07-09 18:57 UTC (permalink / raw) --- 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) --Ot5x3ffkWEuxilWO Content-Type: text/plain; charset=us-ascii Content-Disposition: attachment; filename=v1-0004-convert-PROC_HDR-startupBufferPinWaitBufId-to-an-.patch ^ permalink raw reply [nested|flat] 35+ messages in thread
* [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic @ 2026-07-09 18:57 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 35+ messages in thread From: Nathan Bossart @ 2026-07-09 18:57 UTC (permalink / raw) --- 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) --Ot5x3ffkWEuxilWO Content-Type: text/plain; charset=us-ascii Content-Disposition: attachment; filename=v1-0004-convert-PROC_HDR-startupBufferPinWaitBufId-to-an-.patch ^ permalink raw reply [nested|flat] 35+ messages in thread
* [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic @ 2026-07-09 18:57 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 35+ messages in thread From: Nathan Bossart @ 2026-07-09 18:57 UTC (permalink / raw) --- 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) --Ot5x3ffkWEuxilWO Content-Type: text/plain; charset=us-ascii Content-Disposition: attachment; filename=v1-0004-convert-PROC_HDR-startupBufferPinWaitBufId-to-an-.patch ^ permalink raw reply [nested|flat] 35+ messages in thread
* [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic @ 2026-07-09 18:57 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 35+ messages in thread From: Nathan Bossart @ 2026-07-09 18:57 UTC (permalink / raw) --- 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) --Ot5x3ffkWEuxilWO Content-Type: text/plain; charset=us-ascii Content-Disposition: attachment; filename=v1-0004-convert-PROC_HDR-startupBufferPinWaitBufId-to-an-.patch ^ permalink raw reply [nested|flat] 35+ messages in thread
* [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic @ 2026-07-09 18:57 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 35+ messages in thread From: Nathan Bossart @ 2026-07-09 18:57 UTC (permalink / raw) --- 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) --Ot5x3ffkWEuxilWO Content-Type: text/plain; charset=us-ascii Content-Disposition: attachment; filename=v1-0004-convert-PROC_HDR-startupBufferPinWaitBufId-to-an-.patch ^ permalink raw reply [nested|flat] 35+ messages in thread
* [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic @ 2026-07-09 18:57 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 35+ messages in thread From: Nathan Bossart @ 2026-07-09 18:57 UTC (permalink / raw) --- 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) --Ot5x3ffkWEuxilWO Content-Type: text/plain; charset=us-ascii Content-Disposition: attachment; filename=v1-0004-convert-PROC_HDR-startupBufferPinWaitBufId-to-an-.patch ^ permalink raw reply [nested|flat] 35+ messages in thread
* [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic @ 2026-07-09 18:57 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 35+ messages in thread From: Nathan Bossart @ 2026-07-09 18:57 UTC (permalink / raw) --- 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) --Ot5x3ffkWEuxilWO Content-Type: text/plain; charset=us-ascii Content-Disposition: attachment; filename=v1-0004-convert-PROC_HDR-startupBufferPinWaitBufId-to-an-.patch ^ permalink raw reply [nested|flat] 35+ messages in thread
* [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic @ 2026-07-09 18:57 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 35+ messages in thread From: Nathan Bossart @ 2026-07-09 18:57 UTC (permalink / raw) --- 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) --Ot5x3ffkWEuxilWO Content-Type: text/plain; charset=us-ascii Content-Disposition: attachment; filename=v1-0004-convert-PROC_HDR-startupBufferPinWaitBufId-to-an-.patch ^ permalink raw reply [nested|flat] 35+ messages in thread
* [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic @ 2026-07-09 18:57 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 35+ messages in thread From: Nathan Bossart @ 2026-07-09 18:57 UTC (permalink / raw) --- 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) --Ot5x3ffkWEuxilWO Content-Type: text/plain; charset=us-ascii Content-Disposition: attachment; filename=v1-0004-convert-PROC_HDR-startupBufferPinWaitBufId-to-an-.patch ^ permalink raw reply [nested|flat] 35+ messages in thread
* [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic @ 2026-07-09 18:57 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 35+ messages in thread From: Nathan Bossart @ 2026-07-09 18:57 UTC (permalink / raw) --- 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) --Ot5x3ffkWEuxilWO Content-Type: text/plain; charset=us-ascii Content-Disposition: attachment; filename=v1-0004-convert-PROC_HDR-startupBufferPinWaitBufId-to-an-.patch ^ permalink raw reply [nested|flat] 35+ messages in thread
* [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic @ 2026-07-09 18:57 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 35+ messages in thread From: Nathan Bossart @ 2026-07-09 18:57 UTC (permalink / raw) --- 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) --Ot5x3ffkWEuxilWO Content-Type: text/plain; charset=us-ascii Content-Disposition: attachment; filename=v1-0004-convert-PROC_HDR-startupBufferPinWaitBufId-to-an-.patch ^ permalink raw reply [nested|flat] 35+ messages in thread
* [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic @ 2026-07-09 18:57 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 35+ messages in thread From: Nathan Bossart @ 2026-07-09 18:57 UTC (permalink / raw) --- 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) --Ot5x3ffkWEuxilWO Content-Type: text/plain; charset=us-ascii Content-Disposition: attachment; filename=v1-0004-convert-PROC_HDR-startupBufferPinWaitBufId-to-an-.patch ^ permalink raw reply [nested|flat] 35+ messages in thread
* [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic @ 2026-07-09 18:57 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 35+ messages in thread From: Nathan Bossart @ 2026-07-09 18:57 UTC (permalink / raw) --- 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) --Ot5x3ffkWEuxilWO Content-Type: text/plain; charset=us-ascii Content-Disposition: attachment; filename=v1-0004-convert-PROC_HDR-startupBufferPinWaitBufId-to-an-.patch ^ permalink raw reply [nested|flat] 35+ messages in thread
* [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic @ 2026-07-09 18:57 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 35+ messages in thread From: Nathan Bossart @ 2026-07-09 18:57 UTC (permalink / raw) --- 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) --Ot5x3ffkWEuxilWO Content-Type: text/plain; charset=us-ascii Content-Disposition: attachment; filename=v1-0004-convert-PROC_HDR-startupBufferPinWaitBufId-to-an-.patch ^ permalink raw reply [nested|flat] 35+ messages in thread
* [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic @ 2026-07-09 18:57 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 35+ messages in thread From: Nathan Bossart @ 2026-07-09 18:57 UTC (permalink / raw) --- 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) --Ot5x3ffkWEuxilWO Content-Type: text/plain; charset=us-ascii Content-Disposition: attachment; filename=v1-0004-convert-PROC_HDR-startupBufferPinWaitBufId-to-an-.patch ^ permalink raw reply [nested|flat] 35+ messages in thread
* [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic @ 2026-07-09 18:57 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 35+ messages in thread From: Nathan Bossart @ 2026-07-09 18:57 UTC (permalink / raw) --- 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) --Ot5x3ffkWEuxilWO Content-Type: text/plain; charset=us-ascii Content-Disposition: attachment; filename=v1-0004-convert-PROC_HDR-startupBufferPinWaitBufId-to-an-.patch ^ permalink raw reply [nested|flat] 35+ messages in thread
* [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic @ 2026-07-09 18:57 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 35+ messages in thread From: Nathan Bossart @ 2026-07-09 18:57 UTC (permalink / raw) --- 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) --Ot5x3ffkWEuxilWO Content-Type: text/plain; charset=us-ascii Content-Disposition: attachment; filename=v1-0004-convert-PROC_HDR-startupBufferPinWaitBufId-to-an-.patch ^ permalink raw reply [nested|flat] 35+ messages in thread
end of thread, other threads:[~2026-07-09 18:57 UTC | newest] Thread overview: 35+ messages (download: mbox mbox.gz follow: Atom feed) -- links below jump to the message on this page -- 2026-07-09 18:57 [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic Nathan Bossart <[email protected]> 2026-07-09 18:57 [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic Nathan Bossart <[email protected]> 2026-07-09 18:57 [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic Nathan Bossart <[email protected]> 2026-07-09 18:57 [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic Nathan Bossart <[email protected]> 2026-07-09 18:57 [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic Nathan Bossart <[email protected]> 2026-07-09 18:57 [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic Nathan Bossart <[email protected]> 2026-07-09 18:57 [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic Nathan Bossart <[email protected]> 2026-07-09 18:57 [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic Nathan Bossart <[email protected]> 2026-07-09 18:57 [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic Nathan Bossart <[email protected]> 2026-07-09 18:57 [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic Nathan Bossart <[email protected]> 2026-07-09 18:57 [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic Nathan Bossart <[email protected]> 2026-07-09 18:57 [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic Nathan Bossart <[email protected]> 2026-07-09 18:57 [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic Nathan Bossart <[email protected]> 2026-07-09 18:57 [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic Nathan Bossart <[email protected]> 2026-07-09 18:57 [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic Nathan Bossart <[email protected]> 2026-07-09 18:57 [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic Nathan Bossart <[email protected]> 2026-07-09 18:57 [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic Nathan Bossart <[email protected]> 2026-07-09 18:57 [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic Nathan Bossart <[email protected]> 2026-07-09 18:57 [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic Nathan Bossart <[email protected]> 2026-07-09 18:57 [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic Nathan Bossart <[email protected]> 2026-07-09 18:57 [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic Nathan Bossart <[email protected]> 2026-07-09 18:57 [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic Nathan Bossart <[email protected]> 2026-07-09 18:57 [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic Nathan Bossart <[email protected]> 2026-07-09 18:57 [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic Nathan Bossart <[email protected]> 2026-07-09 18:57 [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic Nathan Bossart <[email protected]> 2026-07-09 18:57 [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic Nathan Bossart <[email protected]> 2026-07-09 18:57 [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic Nathan Bossart <[email protected]> 2026-07-09 18:57 [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic Nathan Bossart <[email protected]> 2026-07-09 18:57 [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic Nathan Bossart <[email protected]> 2026-07-09 18:57 [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic Nathan Bossart <[email protected]> 2026-07-09 18:57 [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic Nathan Bossart <[email protected]> 2026-07-09 18:57 [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic Nathan Bossart <[email protected]> 2026-07-09 18:57 [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic Nathan Bossart <[email protected]> 2026-07-09 18:57 [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic Nathan Bossart <[email protected]> 2026-07-09 18:57 [PATCH v1 3/8] convert FixedParallelState->last_xlog_end to an atomic Nathan Bossart <[email protected]>
This inbox is served by agora; see mirroring instructions for how to clone and mirror all data and code used for this inbox