From c17180fc37865a58bc1742b95a2b472e9229320b Mon Sep 17 00:00:00 2001 From: alterego655 <824662526@qq.com> Date: Wed, 26 Aug 2026 12:24:05 +0800 Subject: [PATCH v1 2/3] Wake primary_flush waiters after implicit WAL flushes XLogWrite() can advance the flush position when it finishes a WAL segment, even if its caller requested only a write. When this happens while AdvanceXLInsertBuffer() recycles a buffer, primary_flush waiters are not notified and can sleep until an unrelated flush or checkpoint. Track such progress with a process-local pending flag and perform the wakeup after releasing WAL write and insertion locks, keeping WaitLSNLock acquisition outside the contended WAL path. Add a deterministic injection-point test that forces this path and inspects the registration directly in shared memory. Co-authored-by: ChatGPT 5.6 Sol --- src/backend/access/transam/xlog.c | 34 ++++++ src/backend/access/transam/xlogwait.c | 3 + src/backend/postmaster/walwriter.c | 3 + src/test/recovery/t/049_wait_for_lsn.pl | 146 ++++++++++++++++++++++++ 4 files changed, 186 insertions(+) diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index de4c96e135f..7e3380a2afc 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -639,6 +639,12 @@ static int UsableBytesInSegment; */ static XLogwrtResult LogwrtResult = {0, 0}; +/* + * True if this process has published primary-flush progress that has not yet + * been reported to primary-flush waiters. + */ +static bool primaryFlushWakeupPending = false; + /* * Update local copy of shared XLogCtl->log{Write,Flush}Result * @@ -652,6 +658,23 @@ static XLogwrtResult LogwrtResult = {0, 0}; _target.Write = pg_atomic_read_u64(&XLogCtl->logWriteResult); \ } while (0) +/* + * Process a primary-flush wakeup requested by XLogWrite(). The caller must + * not hold WALWriteLock or any WAL insertion lock. + */ +static inline void +PrimaryFlushWakeupProcessRequests(void) +{ + if (unlikely(primaryFlushWakeupPending)) + { + /* Clear the process-local request before satisfying it. */ + primaryFlushWakeupPending = false; + + /* XLogWrite() published this frontier before setting the request. */ + WaitLSNWakeup(WAIT_LSN_TYPE_PRIMARY_FLUSH, LogwrtResult.Flush); + } +} + /* * openLogFile is -1 or a kernel FD for an open log file segment. * openLogSegNo identifies the segment, and openLogTLI the corresponding TLI. @@ -1048,6 +1071,9 @@ XLogInsertRecord(XLogRecData *rdata, } } + /* Process any flush progress published while making room for the record. */ + PrimaryFlushWakeupProcessRequests(); + #ifdef WAL_DEBUG if (XLOG_DEBUG) { @@ -2335,6 +2361,7 @@ XLogWrite(XLogwrtRqst WriteRqst, TimeLineID tli, bool flexible) bool ispartialpage; bool last_iteration; bool finishing_seg; + XLogRecPtr oldFlush; int curridx; int npages; int startidx; @@ -2347,6 +2374,7 @@ XLogWrite(XLogwrtRqst WriteRqst, TimeLineID tli, bool flexible) * Update local LogwrtResult (caller probably did this already, but...) */ RefreshXLogWriteResult(LogwrtResult); + oldFlush = LogwrtResult.Flush; /* * Since successive pages in the xlog cache are consecutively allocated, @@ -2608,6 +2636,10 @@ XLogWrite(XLogwrtRqst WriteRqst, TimeLineID tli, bool flexible) pg_write_barrier(); pg_atomic_write_u64(&XLogCtl->logFlushResult, LogwrtResult.Flush); + /* Defer notification until the caller has released its WAL locks. */ + if (LogwrtResult.Flush > oldFlush) + primaryFlushWakeupPending = true; + #ifdef USE_ASSERT_CHECKING { XLogRecPtr Flush; @@ -2946,6 +2978,7 @@ XLogFlush(XLogRecPtr record) * Wake up processes waiting for primary flush LSN to reach current flush * position. */ + primaryFlushWakeupPending = false; WaitLSNWakeup(WAIT_LSN_TYPE_PRIMARY_FLUSH, LogwrtResult.Flush); /* @@ -3134,6 +3167,7 @@ XLogBackgroundFlush(void) * Wake up processes waiting for primary flush LSN to reach current flush * position. */ + primaryFlushWakeupPending = false; WaitLSNWakeup(WAIT_LSN_TYPE_PRIMARY_FLUSH, LogwrtResult.Flush); /* diff --git a/src/backend/access/transam/xlogwait.c b/src/backend/access/transam/xlogwait.c index eee90e7f626..c624f9163a1 100644 --- a/src/backend/access/transam/xlogwait.c +++ b/src/backend/access/transam/xlogwait.c @@ -60,6 +60,7 @@ #include "storage/shmem.h" #include "storage/subsystems.h" #include "utils/fmgrprotos.h" +#include "utils/injection_point.h" #include "utils/pg_lsn.h" #include "utils/snapmgr.h" #include "utils/wait_event.h" @@ -239,6 +240,8 @@ addLSNWaiter(XLogRecPtr lsn, WaitLSNType lsnType) updateMinWaitedLSN(lsnType); LWLockRelease(WaitLSNLock); + + INJECTION_POINT("wait-for-lsn-after-register", NULL); } /* diff --git a/src/backend/postmaster/walwriter.c b/src/backend/postmaster/walwriter.c index 68dd5047c20..5e4e28f4540 100644 --- a/src/backend/postmaster/walwriter.c +++ b/src/backend/postmaster/walwriter.c @@ -60,6 +60,7 @@ #include "storage/procsignal.h" #include "storage/smgr.h" #include "utils/hsearch.h" +#include "utils/injection_point.h" #include "utils/memutils.h" #include "utils/resowner.h" #include "utils/wait_event.h" @@ -234,6 +235,8 @@ WalWriterMain(const void *startup_data, size_t startup_data_len) /* Process any signals received recently */ ProcessMainLoopInterrupts(); + INJECTION_POINT("walwriter-before-background-flush", NULL); + /* * Do what we're here for; then, if XLogBackgroundFlush() found useful * work to do, reset hibernation counter. diff --git a/src/test/recovery/t/049_wait_for_lsn.pl b/src/test/recovery/t/049_wait_for_lsn.pl index e577ef74435..61468ca81e9 100644 --- a/src/test/recovery/t/049_wait_for_lsn.pl +++ b/src/test/recovery/t/049_wait_for_lsn.pl @@ -1247,4 +1247,150 @@ $tl_standby2->stop; $tl_standby1->stop; $tl_primary->stop; +# This final test needs an injection-point-enabled build. Earlier test nodes +# have already been stopped, so check the extensions on its dedicated node. +if ($ENV{enable_injection_points} ne 'yes') +{ + done_testing(); + exit; +} + +# 13. AdvanceXLInsertBuffer() asks XLogWrite() only to write WAL, but completing +# a segment also advances the flush position. Verify that this implicit flush +# wakes primary_flush waiters. +my $flush_wake_primary = PostgreSQL::Test::Cluster->new('flush_wake_primary'); +$flush_wake_primary->init(extra => ['--wal-segsize=1']); +$flush_wake_primary->append_conf( + 'postgresql.conf', qq[ +wal_buffers = 4 +autovacuum = off +checkpoint_timeout = '1h' +max_wal_size = '1GB' +bgwriter_lru_maxpages = 0 +]); +$flush_wake_primary->start; + +# The modules may be absent under installcheck. +if ( !$flush_wake_primary->check_extension('injection_points') + || !$flush_wake_primary->check_extension('test_wait_lsn')) +{ + $flush_wake_primary->stop; + done_testing(); + exit; +} + +is( $flush_wake_primary->safe_psql( + 'postgres', + "SELECT setting::int FROM pg_settings WHERE name = 'wal_segment_size'" + ), + 1024 * 1024, + 'dedicated node runs with 1MB WAL segments'); + +# Keep the WAL writer out of XLogBackgroundFlush(), whose ordinary flush +# wakeup would mask the path under test. +$flush_wake_primary->safe_psql( + 'postgres', q[ + CREATE EXTENSION injection_points; + CREATE EXTENSION test_wait_lsn; + SELECT injection_points_attach('walwriter-before-background-flush', 'wait'); +]); +$flush_wake_primary->wait_for_event('walwriter', + 'walwriter-before-background-flush'); + +my $writer = $flush_wake_primary->background_psql('postgres'); +$writer->set_query_timer_restart; + +# Begin in a fresh segment, then round the insert pointer up to its end. +$writer->query_safe('SELECT pg_switch_wal()'); +my $target_lsn = $writer->query_safe( + q[ + WITH p AS + ( + SELECT pg_current_wal_insert_lsn() AS lsn, + setting::numeric AS segsz + FROM pg_settings + WHERE name = 'wal_segment_size' + ) + SELECT lsn + (segsz - (lsn - '0/0'::pg_lsn) % segsz) + FROM p +]); + +is( $writer->query_safe( + "SELECT pg_current_wal_flush_lsn() < '$target_lsn'::pg_lsn"), + 't', + 'wakeup target is not flushed initially'); + +my $waiter = $flush_wake_primary->background_psql('postgres'); +$waiter->set_query_timer_restart; +my $waiter_pid = $waiter->query_safe('SELECT pg_backend_pid()'); + +$flush_wake_primary->safe_psql( + 'postgres', q[ + SELECT injection_points_attach('wait-for-lsn-after-register', 'wait'); +]); + +$waiter->query_until( + qr/^wait_started\r?$/m, qq[ + \\echo wait_started + WAIT FOR LSN '$target_lsn' + WITH (MODE 'primary_flush', no_throw); + \\echo wait_finished +]); +$flush_wake_primary->wait_for_event('client backend', + 'wait-for-lsn-after-register'); + +is( $flush_wake_primary->safe_psql( + 'postgres', qq[ + SELECT test_wait_lsn_waiter_is_registered( + $waiter_pid, 'primary_flush', '$target_lsn') + ]), + 't', + 'primary_flush waiter is registered before the implicit flush'); + +# A 2MB nontransactional, non-flushing logical message assigns no XID and +# dirties no relation buffers. With four WAL buffers, it must recycle through +# the target. Keep its transaction open so implicit transaction completion +# cannot call XLogSetAsyncXactLSN(); the xid-less rollback writes no WAL. +$writer->query_safe('BEGIN'); +$writer->query_safe( + q[ + SELECT pg_logical_emit_message( + false, 'wfl', repeat('x', 2 * 1024 * 1024), false) +]); + +is( $writer->query_safe( + "SELECT pg_current_wal_flush_lsn() >= '$target_lsn'::pg_lsn"), + 't', + 'finishing a segment advanced flush through the target'); + +# The waiter is still parked, so only a progress waker can remove it. +is( $flush_wake_primary->safe_psql( + 'postgres', qq[ + SELECT test_wait_lsn_waiter_is_registered( + $waiter_pid, 'primary_flush', '$target_lsn') + ]), + 'f', + 'implicit segment flush removes the primary_flush waiter'); + +$flush_wake_primary->safe_psql( + 'postgres', q[ + SELECT injection_points_detach('wait-for-lsn-after-register'); + SELECT injection_points_wakeup('wait-for-lsn-after-register'); +]); + +like($waiter->query_until(qr/^wait_finished\r?$/m, ''), + qr/^success\r?$/m, 'primary_flush waiter returns success after release'); + +# Detach first so that the WAL writer cannot immediately re-enter the point. +$flush_wake_primary->safe_psql( + 'postgres', q[ + SELECT injection_points_detach('walwriter-before-background-flush'); + SELECT injection_points_wakeup('walwriter-before-background-flush'); +]); + +$writer->query_safe('ROLLBACK'); +$writer->quit; +$waiter->quit; +$flush_wake_primary->stop; + done_testing(); -- 2.51.0