From df6e5249d7a1f325a02ea8ef9a3b690e1fe0bef6 Mon Sep 17 00:00:00 2001 From: Vadim Ponomarev Date: Sat, 15 Aug 2026 12:14:40 +0300 Subject: [PATCH v2 4/5] Let a committer whose acknowledgement already arrived skip the queue lock SyncRepWaitForLSN() runs on every commit that wrote WAL, and it takes SyncRepLock exclusively before it can find out whether there is anything to wait for. On a busy primary a large share of those commits find their LSN already acknowledged and queue for nothing, so the answer costs them a period of the lock every other committer is lining up on. Turn lsn[] into an atomic watermark, and read it before taking the lock. A watermark that already covers the commit's LSN says a valid quorum acknowledged it, which is exactly the answer the check under the lock gives. It is only ever moved forward, so a read gone stale can send a committer to the slow path that would have exited, but never past a wait it owes. On platforms where pg_atomic_read_u64() is not a plain load the read is itself a compare-and-exchange, or a spinlock acquisition where 64-bit atomics are emulated. Whether the exit still pays for itself there is untested. --- src/backend/replication/syncrep.c | 69 ++++++++++++++------- src/backend/replication/walsender.c | 3 + src/include/replication/walsender_private.h | 11 +++- 3 files changed, 59 insertions(+), 24 deletions(-) diff --git a/src/backend/replication/syncrep.c b/src/backend/replication/syncrep.c index d67d7bd16ad..1b78375e2c9 100644 --- a/src/backend/replication/syncrep.c +++ b/src/backend/replication/syncrep.c @@ -134,6 +134,20 @@ static int cmp_lsn(const void *a, const void *b); static bool SyncRepQueueIsOrderedByLSN(int mode); #endif +static inline +XLogRecPtr +WalSndCtl_getlsn(int mode) +{ + return pg_atomic_read_u64(&WalSndCtl->lsn[mode]); +} + +static inline +void +WalSndCtl_setlsn(int mode, XLogRecPtr lsn) +{ + pg_atomic_write_u64(&WalSndCtl->lsn[mode], lsn); +} + /* * =========================================================== * Synchronous Replication functions for normal user backends @@ -199,9 +213,32 @@ SyncRepWaitForLSN(XLogRecPtr lsn, bool commit) Assert(dlist_node_is_detached(&MyProc->syncRepLinks)); Assert(WalSndCtl != NULL); + /* + * A watermark that already covers this LSN says a valid quorum + * acknowledged it, which is the same answer the check below the lock + * would give. The watermark only ever moves forward, so a stale read can + * only send us to take the lock for nothing, never past a wait we owe. + * How often this exit fires depends on the wait mode: it needs the + * acknowledgement to have arrived before the committer got here. + */ + if (lsn <= WalSndCtl_getlsn(mode)) + return; + LWLockAcquire(SyncRepLock, LW_EXCLUSIVE); Assert(MyProc->syncRepState == SYNC_REP_NOT_WAITING); + if (lsn <= WalSndCtl_getlsn(mode)) + { + /* + * The LSN is older than what we need to wait for. Even if the sync + * standby data has not been initialized yet, we are OK to not wait + * because we know that there is no point in doing so based on the + * LSN. + */ + LWLockRelease(SyncRepLock); + return; + } + /* * We don't wait for sync rep if SYNC_STANDBY_DEFINED is not set. See * SyncRepUpdateSyncStandbysDefined(). @@ -211,29 +248,16 @@ SyncRepWaitForLSN(XLogRecPtr lsn, bool commit) * to be a low cost check. * * If the sync standby data has not been initialized yet - * (SYNC_STANDBY_INIT is not set), fall back to a check based on the LSN, - * then do a direct GUC check. + * (SYNC_STANDBY_INIT is not set), fall back to direct GUC check. */ if (WalSndCtl->sync_standbys_status & SYNC_STANDBY_INIT) { - if ((WalSndCtl->sync_standbys_status & SYNC_STANDBY_DEFINED) == 0 || - lsn <= WalSndCtl->lsn[mode]) + if ((WalSndCtl->sync_standbys_status & SYNC_STANDBY_DEFINED) == 0) { LWLockRelease(SyncRepLock); return; } } - else if (lsn <= WalSndCtl->lsn[mode]) - { - /* - * The LSN is older than what we need to wait for. The sync standby - * data has not been initialized yet, but we are OK to not wait - * because we know that there is no point in doing so based on the - * LSN. - */ - LWLockRelease(SyncRepLock); - return; - } else if (!SyncStandbysDefined()) { /* @@ -566,19 +590,19 @@ SyncRepReleaseWaiters(void) * Set the lsn first so that when we wake backends they will release up to * this location. */ - if (WalSndCtl->lsn[SYNC_REP_WAIT_WRITE] < writePtr) + if (WalSndCtl_getlsn(SYNC_REP_WAIT_WRITE) < writePtr) { - WalSndCtl->lsn[SYNC_REP_WAIT_WRITE] = writePtr; + WalSndCtl_setlsn(SYNC_REP_WAIT_WRITE, writePtr); numwrite = SyncRepWakeQueue(false, SYNC_REP_WAIT_WRITE); } - if (WalSndCtl->lsn[SYNC_REP_WAIT_FLUSH] < flushPtr) + if (WalSndCtl_getlsn(SYNC_REP_WAIT_FLUSH) < flushPtr) { - WalSndCtl->lsn[SYNC_REP_WAIT_FLUSH] = flushPtr; + WalSndCtl_setlsn(SYNC_REP_WAIT_FLUSH, flushPtr); numflush = SyncRepWakeQueue(false, SYNC_REP_WAIT_FLUSH); } - if (WalSndCtl->lsn[SYNC_REP_WAIT_APPLY] < applyPtr) + if (WalSndCtl_getlsn(SYNC_REP_WAIT_APPLY) < applyPtr) { - WalSndCtl->lsn[SYNC_REP_WAIT_APPLY] = applyPtr; + WalSndCtl_setlsn(SYNC_REP_WAIT_APPLY, applyPtr); numapply = SyncRepWakeQueue(false, SYNC_REP_WAIT_APPLY); } @@ -967,6 +991,7 @@ SyncRepWakeQueue(bool all, int mode) { int numprocs = 0; dlist_mutable_iter iter; + XLogRecPtr lsn = WalSndCtl_getlsn(mode); Assert(mode >= 0 && mode < NUM_SYNC_REP_WAIT_MODE); Assert(LWLockHeldByMeInMode(SyncRepLock, LW_EXCLUSIVE)); @@ -979,7 +1004,7 @@ SyncRepWakeQueue(bool all, int mode) /* * Assume the queue is ordered by LSN */ - if (!all && WalSndCtl->lsn[mode] < proc->waitLSN) + if (!all && lsn < proc->waitLSN) return numprocs; /* diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index 89fed06f851..5afcb7d6153 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -4097,7 +4097,10 @@ static void WalSndShmemInit(void *arg) { for (int i = 0; i < NUM_SYNC_REP_WAIT_MODE; i++) + { dlist_init(&(WalSndCtl->SyncRepQueue[i])); + pg_atomic_init_u64(&(WalSndCtl->lsn[i]), 0); + } for (int i = 0; i < max_wal_senders; i++) { diff --git a/src/include/replication/walsender_private.h b/src/include/replication/walsender_private.h index b0c80deeb24..964dfcf2b5f 100644 --- a/src/include/replication/walsender_private.h +++ b/src/include/replication/walsender_private.h @@ -16,6 +16,7 @@ #include "lib/ilist.h" #include "nodes/nodes.h" #include "nodes/replnodes.h" +#include "port/atomics.h" #include "replication/syncrep.h" #include "storage/condition_variable.h" #include "storage/shmem.h" @@ -91,9 +92,15 @@ typedef struct /* * Current location of the head of the queue. All waiters should have a - * waitLSN that follows this value. Protected by SyncRepLock. + * waitLSN that follows this value. It is atomic, but protected by + * SyncRepLock for concurrent writting since it's change triggers wake of + * waiters. + * + * A committer whose LSN this mirror already covers was acknowledged by a + * valid quorum and has nothing to wait for, so it reads this before + * taking SyncRepLock at all. */ - XLogRecPtr lsn[NUM_SYNC_REP_WAIT_MODE]; + pg_atomic_uint64 lsn[NUM_SYNC_REP_WAIT_MODE]; /* * Status of data related to the synchronous standbys. Waiting backends -- 2.43.0