diff --git a/src/backend/replication/syncrep.c b/src/backend/replication/syncrep.c index d870f09e0a0a0..b23e029661cc8 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, @@ -190,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); @@ -491,6 +505,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 @@ -509,18 +525,15 @@ SyncRepReleaseWaiters(void) } /* - * We're a potential sync standby. Release waiters if there are enough - * sync standbys and we are considered as sync. - */ - 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); @@ -548,11 +561,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. @@ -560,21 +576,44 @@ SyncRepReleaseWaiters(void) if (WalSndCtl->lsn[SYNC_REP_WAIT_WRITE] < writePtr) { WalSndCtl->lsn[SYNC_REP_WAIT_WRITE] = writePtr; - numwrite = SyncRepWakeQueue(false, SYNC_REP_WAIT_WRITE); + 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; - numflush = SyncRepWakeQueue(false, SYNC_REP_WAIT_FLUSH); + 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; - numapply = SyncRepWakeQueue(false, SYNC_REP_WAIT_APPLY); + pg_atomic_write_u64(&WalSndCtl->lsn_published[SYNC_REP_WAIT_APPLY], + (uint64) applyPtr); + 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 +941,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 +1009,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 +1034,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 +1049,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 +1063,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) { diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index c65dd324325c9..17f49aa9f8fe0 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. @@ -4019,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 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();