From 6e5504562b52e88bf9d3f6cee1e5f43bb2471fa1 Mon Sep 17 00:00:00 2001 From: Vadim Ponomarev Date: Sat, 15 Aug 2026 12:08:13 +0300 Subject: [PATCH 1/4] Wake the released sync-rep waiters after the queue lock is down SyncRepWakeQueue() sets each released backend's latch while holding SyncRepLock exclusively. A latch is a kill() syscall whenever its proc is asleep, and at a high commit rate the walsender runs one of them per released commit inside the very section every committer lines up on. ProcArrayGroupClearXid() already wakes its batch only after ProcArrayLock is down, for the same reason. Collect the released procs into a list instead, and set their latches once the lock is released. The unlink, the write barrier and the state store stay under the lock: a waiter reads syncRepState without the lock and must never find itself completed while still on the queue. Nothing in the deferred loop can error out, so a released proc cannot be left completed but unlatched short of the process dying outright -- a window the in-lock SetLatch had as well. A proc that noticed its state on its own and moved on, even into a new wait, gets a spurious latch set, which every latch sleeper tolerates. The list is sized to MaxBackends and allocated once per releasing process. A proc waits in at most one queue, so one list of that size bounds a walk over all three. --- src/backend/replication/syncrep.c | 78 +++++++++++++++++++++++++------ 1 file changed, 65 insertions(+), 13 deletions(-) diff --git a/src/backend/replication/syncrep.c b/src/backend/replication/syncrep.c index d870f09e0a0a0..ccd51022a5f28 100644 --- a/src/backend/replication/syncrep.c +++ b/src/backend/replication/syncrep.c @@ -84,6 +84,7 @@ #include "storage/proc.h" #include "tcop/tcopprot.h" #include "utils/guc_hooks.h" +#include "utils/memutils.h" #include "utils/ps_status.h" #include "utils/wait_event.h" @@ -100,7 +101,9 @@ static int SyncRepWaitMode = SYNC_REP_NO_WAIT; static void SyncRepQueueInsert(int mode); static void SyncRepCancelWait(void); -static int SyncRepWakeQueue(bool all, int mode); +static PGPROC **SyncRepWakeList(void); +static int SyncRepWakeQueue(bool all, int mode, + PGPROC **wakelist, int *nwake); static bool SyncRepGetSyncRecPtr(XLogRecPtr *writePtr, XLogRecPtr *flushPtr, @@ -491,6 +494,8 @@ SyncRepReleaseWaiters(void) int numwrite = 0; int numflush = 0; int numapply = 0; + PGPROC **wakelist; + int nwake = 0; /* * If this WALSender is serving a standby that is not on the list of @@ -512,6 +517,8 @@ SyncRepReleaseWaiters(void) * We're a potential sync standby. Release waiters if there are enough * sync standbys and we are considered as sync. */ + wakelist = SyncRepWakeList(); + LWLockAcquire(SyncRepLock, LW_EXCLUSIVE); /* @@ -560,21 +567,38 @@ SyncRepReleaseWaiters(void) if (WalSndCtl->lsn[SYNC_REP_WAIT_WRITE] < writePtr) { WalSndCtl->lsn[SYNC_REP_WAIT_WRITE] = writePtr; - numwrite = SyncRepWakeQueue(false, SYNC_REP_WAIT_WRITE); + numwrite = SyncRepWakeQueue(false, SYNC_REP_WAIT_WRITE, + wakelist, &nwake); } if (WalSndCtl->lsn[SYNC_REP_WAIT_FLUSH] < flushPtr) { WalSndCtl->lsn[SYNC_REP_WAIT_FLUSH] = flushPtr; - numflush = SyncRepWakeQueue(false, SYNC_REP_WAIT_FLUSH); + numflush = SyncRepWakeQueue(false, SYNC_REP_WAIT_FLUSH, + wakelist, &nwake); } if (WalSndCtl->lsn[SYNC_REP_WAIT_APPLY] < applyPtr) { WalSndCtl->lsn[SYNC_REP_WAIT_APPLY] = applyPtr; - numapply = SyncRepWakeQueue(false, SYNC_REP_WAIT_APPLY); + numapply = SyncRepWakeQueue(false, SYNC_REP_WAIT_APPLY, + wakelist, &nwake); } LWLockRelease(SyncRepLock); + /* + * Wake the released backends now that the lock is down. Each latch is a + * kill() for a sleeping proc, and running one per released commit inside + * the exclusive section makes every committer wait for those syscalls. + * The procs below are off the queue with their state already complete, so + * nothing here needs the lock's protection. Nothing here can error out + * either: only this process dying outright could leave a released proc + * completed but unlatched, and the in-lock SetLatch had that same window. + * A proc that noticed its state on its own and moved on, even into a new + * wait, gets a spurious latch set, which every latch sleeper tolerates. + */ + for (int i = 0; i < nwake; i++) + SetLatch(&(wakelist[i]->procLatch)); + elog(DEBUG3, "released %d procs up to write %X/%08X, %d procs up to flush %X/%08X, %d procs up to apply %X/%08X", numwrite, LSN_FORMAT_ARGS(writePtr), numflush, LSN_FORMAT_ARGS(flushPtr), @@ -902,16 +926,38 @@ SyncRepGetStandbyPriority(void) return (SyncRepConfig->syncrep_method == SYNC_REP_PRIORITY) ? priority : 1; } +/* + * Return the list a release collects the procs to latch into, allocating it + * the first time this process releases anybody. It is sized for every + * backend to be waiting at once; a proc waits in at most one queue, so one + * list of that size is enough for a pass over all three. + */ +static PGPROC ** +SyncRepWakeList(void) +{ + static PGPROC **wakelist = NULL; + + if (wakelist == NULL) + wakelist = (PGPROC **) + MemoryContextAlloc(TopMemoryContext, + MaxBackends * sizeof(PGPROC *)); + return wakelist; +} + /* * Walk the specified queue from head. Set the state of any backends that - * need to be woken, remove them from the queue, and then wake them. - * Pass all = true to wake whole queue; otherwise, just wake up to + * need to be woken and remove them from the queue; the procs to wake are + * appended to wakelist for the caller to latch once the lock is down. + * Pass all = true to release the whole queue; otherwise, just release up to * the walsender's LSN. * - * The caller must hold SyncRepLock in exclusive mode. + * The caller must hold SyncRepLock in exclusive mode, and must set the + * latches itself after releasing it. Unlink, barrier and state stay together + * in here: a waiter reads syncRepState without the lock and must never find + * itself completed while still on the queue. */ static int -SyncRepWakeQueue(bool all, int mode) +SyncRepWakeQueue(bool all, int mode, PGPROC **wakelist, int *nwake) { int numprocs = 0; dlist_mutable_iter iter; @@ -948,10 +994,9 @@ SyncRepWakeQueue(bool all, int mode) */ proc->syncRepState = SYNC_REP_WAIT_COMPLETE; - /* - * Wake only when we have set state and removed from queue. - */ - SetLatch(&(proc->procLatch)); + /* the list is sized to every process that can ever queue here */ + Assert(*nwake < MaxBackends); + wakelist[(*nwake)++] = proc; numprocs++; } @@ -974,6 +1019,9 @@ SyncRepUpdateSyncStandbysDefined(void) if (sync_standbys_defined != ((WalSndCtl->sync_standbys_status & SYNC_STANDBY_DEFINED) != 0)) { + PGPROC **wakelist = SyncRepWakeList(); + int nwake = 0; + LWLockAcquire(SyncRepLock, LW_EXCLUSIVE); /* @@ -986,7 +1034,7 @@ SyncRepUpdateSyncStandbysDefined(void) int i; for (i = 0; i < NUM_SYNC_REP_WAIT_MODE; i++) - SyncRepWakeQueue(true, i); + SyncRepWakeQueue(true, i, wakelist, &nwake); } /* @@ -1000,6 +1048,10 @@ SyncRepUpdateSyncStandbysDefined(void) (sync_standbys_defined ? SYNC_STANDBY_DEFINED : 0); LWLockRelease(SyncRepLock); + + /* wake the released backends now that the lock is down */ + for (int i = 0; i < nwake; i++) + SetLatch(&(wakelist[i]->procLatch)); } else if ((WalSndCtl->sync_standbys_status & SYNC_STANDBY_INIT) == 0) { From a666381e25a998bee5e8f458756e590e335fa6c5 Mon Sep 17 00:00:00 2001 From: Vadim Ponomarev Date: Sat, 15 Aug 2026 12:08:53 +0300 Subject: [PATCH 2/4] Compute the synced positions before taking the sync-rep queue lock SyncRepReleaseWaiters() takes SyncRepLock and only then walks the walsender slots to work out the synced write, flush and apply positions. That walk takes a spinlock per slot, allocates, and for a quorum set sorts the result, and every cycle of it is spent in the section every committer lines up on. The comment there conceded the work does not need the lock and kept it inside anyway, to guarantee the positions are newer than any previous execution of the routine used. That guarantee is not needed. The three sites that consume the positions each move lsn[] forward only when the new reading is ahead of the stored one, so positions gone stale while the lock was being taken release nobody and change nothing; a concurrent walsender that got further has already stored its own. Compute them before taking the lock, and leave without taking it at all when this walsender turns out not to be a sync standby. --- src/backend/replication/syncrep.c | 28 +++++++++++++--------------- 1 file changed, 13 insertions(+), 15 deletions(-) diff --git a/src/backend/replication/syncrep.c b/src/backend/replication/syncrep.c index ccd51022a5f28..6406168721b5a 100644 --- a/src/backend/replication/syncrep.c +++ b/src/backend/replication/syncrep.c @@ -514,20 +514,15 @@ SyncRepReleaseWaiters(void) } /* - * We're a potential sync standby. Release waiters if there are enough - * sync standbys and we are considered as sync. - */ - wakelist = SyncRepWakeList(); - - LWLockAcquire(SyncRepLock, LW_EXCLUSIVE); - - /* - * Check whether we are a sync standby or not, and calculate the synced - * positions among all sync standbys. (Note: although this step does not - * of itself require holding SyncRepLock, it seems like a good idea to do - * it after acquiring the lock. This ensures that the WAL pointers we use - * to release waiters are newer than any previous execution of this - * routine used.) + * We're a potential sync standby. Check whether we are a sync standby + * and calculate the synced positions among all sync standbys before + * taking the lock: the walk over the walsender slots takes their + * spinlocks, allocates, and possibly sorts, and doing all of it under + * SyncRepLock delays every committer. + * + * Positions gone stale by the time the lock is held cost nothing. The + * guards further down only ever move lsn[] forward, so a reading older + * than a concurrent walsender's simply releases nobody. */ got_recptr = SyncRepGetSyncRecPtr(&writePtr, &flushPtr, &applyPtr, &am_sync); @@ -555,11 +550,14 @@ SyncRepReleaseWaiters(void) */ if (!got_recptr || !am_sync) { - LWLockRelease(SyncRepLock); announce_next_takeover = !am_sync; return; } + wakelist = SyncRepWakeList(); + + LWLockAcquire(SyncRepLock, LW_EXCLUSIVE); + /* * Set the lsn first so that when we wake backends they will release up to * this location. From 2e1ccd990114a393c4537b1e2e688778b3511e48 Mon Sep 17 00:00:00 2001 From: Vadim Ponomarev Date: Sat, 15 Aug 2026 12:12:19 +0300 Subject: [PATCH 3/4] Release the sync-rep waiters once per drained batch of standby replies ProcessStandbyReplyMessage() calls SyncRepReleaseWaiters() for every reply it processes, and several replies routinely sit in the walsender's socket together. Each of those calls takes SyncRepLock exclusively, so a batch of replies costs the committers one period of that lock apiece -- computed, for all but the last reply, from positions the next message in the same batch immediately makes stale. Have a reply only mark a release as pending, and run one release at the end of the drain. The positions in shared memory are the newest of the batch by then, so the single pass releases everything the individual passes would have. A deferred release must survive every way out of the drain, because the positions the reply already stored are valid whatever follows and the committers it acknowledged have no other process to wake them: - the standby's goodbye, an EOF, an invalid message type and an unexpected message type each run the pending release before leaving. A clean standby shutdown sends its final reply and the goodbye back to back, which makes that exit the routine one rather than the exotic one. - an error thrown while a later message in the same drain is parsed -- a torn message above all -- leaves through WalSndErrorCleanup(), which runs the pending release after the locks are dropped. The test makes both coincidences certain instead of likely. For the first, a paused standby holds a remote_apply committer in the queue, the walsender is held with SIGSTOP while the standby applies past the commit and shuts down, and the released walsender drains the final reply and the goodbye in one pass. For the second, an injection point right after a drained reply stands in for the torn message; it fires on every reply, so the walreceiver is held with SIGSTOP while replay proceeds from WAL already on standby disk, which makes the first reply after release the one carrying the apply position the committer waits for. Both halves fail without their fix. --- src/backend/replication/walsender.c | 59 +++++- src/test/recovery/meson.build | 1 + src/test/recovery/t/056_syncrep_release.pl | 206 +++++++++++++++++++++ 3 files changed, 264 insertions(+), 2 deletions(-) create mode 100644 src/test/recovery/t/056_syncrep_release.pl diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index c65dd324325c9..0db72ac85b38b 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -94,6 +94,7 @@ #include "utils/acl.h" #include "utils/builtins.h" #include "utils/guc.h" +#include "utils/injection_point.h" #include "utils/lsyscache.h" #include "utils/memutils.h" #include "utils/pg_lsn.h" @@ -206,6 +207,13 @@ static TimestampTz last_reply_timestamp = 0; /* Have we sent a heartbeat message asking for reply, since last reply? */ static bool waiting_for_ping_response = false; +/* + * Set when a standby reply has updated this walsender's positions and the + * waiters those positions release have not been released yet. Raised per + * reply, acted on once per drain of the socket. + */ +static bool syncrep_release_pending = false; + /* Timestamp when walsender received the shutdown request */ static TimestampTz shutdown_request_timestamp = 0; @@ -300,6 +308,7 @@ static void CreateReplicationSlot(CreateReplicationSlotCmd *cmd); static void DropReplicationSlot(DropReplicationSlotCmd *cmd); static void StartReplication(StartReplicationCmd *cmd); static void StartLogicalReplication(StartReplicationCmd *cmd); +static void SyncRepFlushPendingRelease(void); static void ProcessStandbyMessage(void); static void ProcessStandbyReplyMessage(void); static void ProcessStandbyHSFeedbackMessage(void); @@ -381,6 +390,15 @@ WalSndErrorCleanup(void) pgstat_report_wait_end(); pgaio_error_cleanup(); + /* + * A release deferred by the reply drain survives an error thrown while a + * later message in the same drain was being parsed. The positions the + * drained reply put in shared memory are valid whatever came after it, + * and the committers it acknowledged have no other process to wake them. + * The locks are released above, so the queue lock is free to take. + */ + SyncRepFlushPendingRelease(); + if (xlogreader != NULL && xlogreader->seg.ws_file >= 0) wal_segment_close(xlogreader); @@ -2353,6 +2371,22 @@ exec_replication_command(const char *cmd_string) return true; } +/* + * Run the release a drained reply deferred. A reply already processed has + * put its positions in shared memory, and the committers it acknowledged + * have nothing but this process to wake them, so every exit out of the reply + * drain runs through here before leaving. The standby's goodbye is the + * common one. + */ +static void +SyncRepFlushPendingRelease(void) +{ + if (!syncrep_release_pending) + return; + syncrep_release_pending = false; + SyncRepReleaseWaiters(); +} + /* * Process any incoming messages while streaming. Also checks if the remote * end has closed the connection. @@ -2379,6 +2413,7 @@ ProcessRepliesIfAny(void) if (r < 0) { /* unexpected error or EOF */ + SyncRepFlushPendingRelease(); ereport(COMMERROR, (errcode(ERRCODE_PROTOCOL_VIOLATION), errmsg("unexpected EOF on standby connection"))); @@ -2402,6 +2437,7 @@ ProcessRepliesIfAny(void) maxmsglen = PQ_SMALL_MESSAGE_LIMIT; break; default: + SyncRepFlushPendingRelease(); ereport(FATAL, (errcode(ERRCODE_PROTOCOL_VIOLATION), errmsg("invalid standby message type \"%c\"", @@ -2414,6 +2450,7 @@ ProcessRepliesIfAny(void) resetStringInfo(&reply_message); if (pq_getmessage(&reply_message, maxmsglen)) { + SyncRepFlushPendingRelease(); ereport(COMMERROR, (errcode(ERRCODE_PROTOCOL_VIOLATION), errmsg("unexpected EOF on standby connection"))); @@ -2429,6 +2466,7 @@ ProcessRepliesIfAny(void) */ case PqMsg_CopyData: ProcessStandbyMessage(); + INJECTION_POINT("walsender-reply-drained", NULL); received = true; break; @@ -2450,9 +2488,12 @@ ProcessRepliesIfAny(void) /* * PqMsg_Terminate means that the standby is closing down the - * socket. + * socket. The last reply it sent is drained already, and + * what that reply acknowledged must not leave with this + * process. */ case PqMsg_Terminate: + SyncRepFlushPendingRelease(); proc_exit(0); default: @@ -2468,6 +2509,13 @@ ProcessRepliesIfAny(void) last_reply_timestamp = last_processing; waiting_for_ping_response = false; } + + /* + * One release covers every reply drained above: the positions in shared + * memory are already the newest ones, and each release takes SyncRepLock + * exclusively. + */ + SyncRepFlushPendingRelease(); } /* @@ -2498,6 +2546,7 @@ ProcessStandbyMessage(void) break; default: + SyncRepFlushPendingRelease(); ereport(COMMERROR, (errcode(ERRCODE_PROTOCOL_VIOLATION), errmsg("unexpected message type \"%c\"", msgtype))); @@ -2633,8 +2682,14 @@ ProcessStandbyReplyMessage(void) SpinLockRelease(&walsnd->mutex); } + /* + * The release is left for ProcessRepliesIfAny() to run once per drain of + * the socket: several replies routinely sit in the buffer together, and + * every release takes SyncRepLock exclusively to compute positions this + * message has just made stale anyway. + */ if (!am_cascading_walsender) - SyncRepReleaseWaiters(); + syncrep_release_pending = true; /* * Advance our local xmin horizon when the client confirmed a flush. diff --git a/src/test/recovery/meson.build b/src/test/recovery/meson.build index 39ec8c4946ddf..dca522ed0ce12 100644 --- a/src/test/recovery/meson.build +++ b/src/test/recovery/meson.build @@ -64,6 +64,7 @@ tests += { 't/053_standby_login_event_trigger.pl', 't/054_unlogged_sequence_promotion.pl', 't/055_cascade_reconnect.pl', + 't/056_syncrep_release.pl', ], }, } diff --git a/src/test/recovery/t/056_syncrep_release.pl b/src/test/recovery/t/056_syncrep_release.pl new file mode 100644 index 0000000000000..bc205356fe091 --- /dev/null +++ b/src/test/recovery/t/056_syncrep_release.pl @@ -0,0 +1,206 @@ +# Copyright (c) 2026, PostgreSQL Global Development Group + +# A walsender coalesces the releases a drained batch of standby replies +# asks for into one pass at the end of the drain. The drain has early +# exits, the standby's goodbye being the common one, and a release owed +# by a reply processed in the same drain must survive them: the positions are +# in shared memory already, and a committer acknowledged by that reply has +# nothing else to wake it. What this file proves is that a commit whose +# ack arrives in the same drain as the standby's goodbye comes back. +# +# The choreography makes that coincidence certain instead of likely. A +# paused standby holds a remote_apply committer in the queue while the +# flush acks flow; the walsender is then held with SIGSTOP, the standby +# is resumed, allowed to apply past the commit, and shut down, so its +# final reply, the one carrying the apply position the committer waits +# for, lands in the walsender's socket right next to the goodbye. The +# walsender, released, drains both in one pass. + +use strict; +use warnings FATAL => 'all'; +use PostgreSQL::Test::Cluster; +use PostgreSQL::Test::Utils; +use Time::HiRes qw(usleep time); +use Test::More; + +my $primary = PostgreSQL::Test::Cluster->new('rel_primary'); +$primary->init(allows_streaming => 1); +$primary->append_conf( + 'postgresql.conf', q( +autovacuum = off +checkpoint_timeout = 1h +)); +$primary->start; +$primary->safe_psql('postgres', 'CREATE TABLE t (id int)'); +$primary->backup('bkp'); + +# A node streaming from a backup reports its own name as its +# application_name, which is what the synchronous set goes by. +my $standby = PostgreSQL::Test::Cluster->new('rel_standby'); +$standby->init_from_backup($primary, 'bkp', has_streaming => 1); +$standby->start; +$primary->wait_for_catchup($standby, 'replay'); + +$primary->safe_psql('postgres', + "ALTER SYSTEM SET synchronous_standby_names = 'rel_standby'"); +$primary->reload; +$primary->poll_query_until('postgres', + "SELECT sync_state = 'sync' FROM pg_stat_replication WHERE application_name = 'rel_standby'" +) or die "standby never became synchronous"; + +# Hold replay: the flush acks keep flowing, the apply position does not, +# so a remote_apply commit queues and stays queued. +$standby->safe_psql('postgres', 'SELECT pg_wal_replay_pause()'); + +my $committer = $primary->background_psql('postgres'); +$committer->query_until( + qr/inserting/, q( +\echo inserting +SET synchronous_commit = remote_apply; +INSERT INTO t VALUES (1); +)); + +$primary->poll_query_until('postgres', + "SELECT count(*) > 0 FROM pg_stat_activity WHERE wait_event = 'SyncRep'") + or die "committer never reached the sync-rep queue"; +my $commit_lsn = + $primary->safe_psql('postgres', 'SELECT pg_current_wal_lsn()'); + +# Hold the walsender, so everything the standby says from here on is +# drained in one pass. +my $walsender = $primary->safe_psql('postgres', + "SELECT pid FROM pg_stat_replication WHERE application_name = 'rel_standby'" +); +die "no walsender pid" unless $walsender =~ /^\d+$/; +kill 'STOP', $walsender or die "SIGSTOP walsender: $!"; + +# Let the standby apply past the commit, then say goodbye. +$standby->safe_psql('postgres', 'SELECT pg_wal_replay_resume()'); +my $deadline = time() + 30; +while (time() < $deadline) +{ + my $replayed = $standby->safe_psql('postgres', + "SELECT pg_last_wal_replay_lsn() >= '$commit_lsn'::pg_lsn"); + last if $replayed eq 't'; + usleep(100_000); +} +$standby->stop('fast'); + +# The final reply and the goodbye are now side by side in the held +# walsender's socket. Release it: the drain must not drop the release +# the reply asks for on its way out. +kill 'CONT', $walsender or die "SIGCONT walsender: $!"; + +$deadline = time() + 10; +my $released = 0; +while (time() < $deadline) +{ + my $waiting = $primary->safe_psql('postgres', + "SELECT count(*) FROM pg_stat_activity WHERE wait_event = 'SyncRep'"); + if ($waiting eq '0') + { + $released = 1; + last; + } + usleep(200_000); +} +ok($released, + 'a commit acknowledged in the drain the standby left in is released'); + +# Free the committer session whatever state it is in. +$primary->safe_psql('postgres', + "SELECT pg_cancel_backend(pid) FROM pg_stat_activity WHERE wait_event = 'SyncRep'" +) if !$released; +$committer->quit; + +# The other way out of the drain: an error thrown while parsing a later +# message in the same pass. A reply processed just before it has already +# asked for a release, and the walsender's error cleanup must run that +# release on the way out. The connection is lost either way; the +# committers the reply acknowledged are not. An injection point right +# after a drained reply stands in for the torn message. +# +# The reply the error lands on has to be the one that carries the apply +# position the committer waits for, and the injection point fires on +# every drained reply, so no reply may reach the walsender between arming +# it and the apply position passing the commit. Holding the +# walreceiver with SIGSTOP is what guarantees that: replay proceeds from +# WAL already on standby disk, and the receiver's first words on release +# are the positions as they stand then. +if (($ENV{enable_injection_points} // 'no') eq 'yes') +{ + $standby->start; + $primary->wait_for_catchup($standby, 'replay'); + $primary->safe_psql('postgres', 'CREATE EXTENSION injection_points'); + + $standby->safe_psql('postgres', 'SELECT pg_wal_replay_pause()'); + + my $committer2 = $primary->background_psql('postgres'); + $committer2->query_until( + qr/inserting2/, q( +\echo inserting2 +SET synchronous_commit = remote_apply; +INSERT INTO t VALUES (3); +)); + $primary->poll_query_until('postgres', + "SELECT count(*) > 0 FROM pg_stat_activity WHERE wait_event = 'SyncRep'" + ) or die "second committer never reached the sync-rep queue"; + my $lsn2 = $primary->safe_psql('postgres', 'SELECT pg_current_wal_lsn()'); + + # The commit's WAL must be on standby disk before the receiver is + # held, or replay below has nothing to apply. + $standby->poll_query_until('postgres', + "SELECT pg_last_wal_receive_lsn() >= '$lsn2'::pg_lsn") + or die "standby never flushed the commit's WAL"; + + my $walreceiver = + $standby->safe_psql('postgres', 'SELECT pid FROM pg_stat_wal_receiver'); + die "no walreceiver pid" unless $walreceiver =~ /^\d+$/; + kill 'STOP', $walreceiver or die "SIGSTOP walreceiver: $!"; + + # A reply already in flight when the receiver stopped is drained -- + # and released, long before the injection point is armed. + usleep(300_000); + $primary->safe_psql('postgres', + "SELECT injection_points_attach('walsender-reply-drained', 'error')"); + + $standby->safe_psql('postgres', 'SELECT pg_wal_replay_resume()'); + $deadline = time() + 30; + while (time() < $deadline) + { + my $replayed = $standby->safe_psql('postgres', + "SELECT pg_last_wal_replay_lsn() >= '$lsn2'::pg_lsn"); + last if $replayed eq 't'; + usleep(100_000); + } + + # The receiver's first reply now carries an apply position past the + # commit, and the injection point tears the drain right after it. + kill 'CONT', $walreceiver or die "SIGCONT walreceiver: $!"; + + $deadline = time() + 15; + my $released2 = 0; + while (time() < $deadline) + { + my $waiting = $primary->safe_psql('postgres', + "SELECT count(*) FROM pg_stat_activity WHERE wait_event = 'SyncRep'" + ); + if ($waiting eq '0') + { + $released2 = 1; + last; + } + usleep(200_000); + } + ok($released2, + 'a commit acknowledged right before a torn message is released'); + + $primary->safe_psql('postgres', + "SELECT injection_points_detach('walsender-reply-drained')"); + $primary->safe_psql('postgres', + "SELECT pg_cancel_backend(pid) FROM pg_stat_activity WHERE wait_event = 'SyncRep'" + ) if !$released2; + $committer2->quit; +} + +done_testing(); From f0643957eb29d57dab401ec20e746875d5000e0e Mon Sep 17 00:00:00 2001 From: Vadim Ponomarev Date: Sat, 15 Aug 2026 12:14:40 +0300 Subject: [PATCH 4/4] 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. Mirror lsn[] into an atomic watermark, written under SyncRepLock right after lsn[] itself, 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. Both are 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 | 17 +++++++++++++++++ src/backend/replication/walsender.c | 3 +++ src/include/replication/walsender_private.h | 11 +++++++++++ 3 files changed, 31 insertions(+) diff --git a/src/backend/replication/syncrep.c b/src/backend/replication/syncrep.c index 6406168721b5a..b23e029661cc8 100644 --- a/src/backend/replication/syncrep.c +++ b/src/backend/replication/syncrep.c @@ -193,6 +193,17 @@ SyncRepWaitForLSN(XLogRecPtr lsn, bool commit) Assert(dlist_node_is_detached(&MyProc->syncRepLinks)); Assert(WalSndCtl != NULL); + /* + * A published 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 <= (XLogRecPtr) pg_atomic_read_u64(&WalSndCtl->lsn_published[mode])) + return; + LWLockAcquire(SyncRepLock, LW_EXCLUSIVE); Assert(MyProc->syncRepState == SYNC_REP_NOT_WAITING); @@ -565,18 +576,24 @@ SyncRepReleaseWaiters(void) if (WalSndCtl->lsn[SYNC_REP_WAIT_WRITE] < writePtr) { WalSndCtl->lsn[SYNC_REP_WAIT_WRITE] = writePtr; + pg_atomic_write_u64(&WalSndCtl->lsn_published[SYNC_REP_WAIT_WRITE], + (uint64) writePtr); numwrite = SyncRepWakeQueue(false, SYNC_REP_WAIT_WRITE, wakelist, &nwake); } if (WalSndCtl->lsn[SYNC_REP_WAIT_FLUSH] < flushPtr) { WalSndCtl->lsn[SYNC_REP_WAIT_FLUSH] = flushPtr; + pg_atomic_write_u64(&WalSndCtl->lsn_published[SYNC_REP_WAIT_FLUSH], + (uint64) flushPtr); numflush = SyncRepWakeQueue(false, SYNC_REP_WAIT_FLUSH, wakelist, &nwake); } if (WalSndCtl->lsn[SYNC_REP_WAIT_APPLY] < applyPtr) { WalSndCtl->lsn[SYNC_REP_WAIT_APPLY] = applyPtr; + pg_atomic_write_u64(&WalSndCtl->lsn_published[SYNC_REP_WAIT_APPLY], + (uint64) applyPtr); numapply = SyncRepWakeQueue(false, SYNC_REP_WAIT_APPLY, wakelist, &nwake); } diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index 0db72ac85b38b..17f49aa9f8fe0 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -4074,7 +4074,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_published[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 b0c80deeb24de..4354bce759a3a 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" @@ -95,6 +96,16 @@ typedef struct */ XLogRecPtr lsn[NUM_SYNC_REP_WAIT_MODE]; + /* + * An atomic mirror of lsn[], written right after it under SyncRepLock and + * only ever forward. 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. On platforms where a + * 64-bit atomic read is not a plain load, this read is itself a + * compare-and-exchange or a spinlock acquisition. + */ + pg_atomic_uint64 lsn_published[NUM_SYNC_REP_WAIT_MODE]; + /* * Status of data related to the synchronous standbys. Waiting backends * can't reload the config file safely, so checkpointer updates this value