diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml index 607dafcb2ed16..d7b4de4b6bc29 100644 --- a/doc/src/sgml/config.sgml +++ b/doc/src/sgml/config.sgml @@ -3591,6 +3591,29 @@ include_dir 'conf.d' + + wal_flush_backend_flushers (integer) + + wal_flush_backend_flushers configuration parameter + + + + + Sets the maximum number of client backends that can flush WAL + concurrently. When the limit is reached, further backends needing + their WAL flushed wait until one of the active flushers has flushed + it for them, or until they are allowed to flush themselves. This + can reduce contention on WAL flushing with very many concurrently + committing sessions. Background processes, such as autovacuum + workers and background workers, are not subject to this limit. + The value 0, which is the default, disables the + limit. This parameter can only be set in the + postgresql.conf file or on the server command + line. + + + + wal_skip_threshold (integer) diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index e07cb9103515f..dc96bf86d8a5d 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -83,6 +83,7 @@ #include "replication/walreceiver.h" #include "replication/walsender.h" #include "storage/bufmgr.h" +#include "storage/condition_variable.h" #include "storage/fd.h" #include "storage/ipc.h" #include "storage/large_object.h" @@ -132,6 +133,7 @@ int wal_sync_method = DEFAULT_WAL_SYNC_METHOD; int wal_level = WAL_LEVEL_REPLICA; int CommitDelay = 0; /* precommit delay in microseconds */ int CommitSiblings = 5; /* # concurrent xacts needed to sleep */ +int wal_flush_backend_flushers = 0; int wal_retrieve_retry_interval = 5000; int max_slot_wal_keep_size_mb = -1; int wal_decode_buffer_size = 512 * 1024; @@ -552,6 +554,36 @@ typedef struct XLogCtlData XLogRecPtr lastFpwDisableRecPtr; slock_t info_lck; /* locks shared variables shown above */ + + /* + * State for limiting the number of concurrent backend WAL flushers, see + * wal_flush_backend_flushers. + * + * backendFlushers counts the client backends currently allowed to perform + * the flush loop in XLogFlush(). backendFlushWaiters counts the backends + * waiting for a slot or for their WAL to be flushed by someone else; it + * gates the wakeup calls so that they cost nothing when the feature is + * idle. backendFlushRequest is a monotonically-advancing maximum of the + * flush LSNs the waiters need; slot holders extend their flush requests + * to cover it, so waiters piggyback on the holders' fsyncs. Waiters + * sleep on backendFlushCV; it is signaled when a slot is released and + * broadcast when the flushed position advances. + * + * All of these take atomic read-modify-write traffic from every + * committing backend, so they must not share cache lines with anything + * else that is hot: not with the log*Result fields that + * RefreshXLogWriteResult() reads on every insert/write/flush, and not + * with info_lck above, which every XLogFlush() entry and every + * GetRedoRecPtr() call takes. The condition variable's internal spinlock + * is pounded harder than the counters (two acquisitions per waiter plus + * every signal/broadcast), so it gets a cache line of its own as well. + */ + char backendFlushPad1[PG_CACHE_LINE_SIZE]; + pg_atomic_uint32 backendFlushers; /* active backend WAL flushers */ + pg_atomic_uint32 backendFlushWaiters; /* backends waiting in XLogFlush */ + pg_atomic_uint64 backendFlushRequest; /* max flush LSN of waiters */ + char backendFlushPad2[PG_CACHE_LINE_SIZE]; + ConditionVariable backendFlushCV; } XLogCtlData; /* @@ -612,6 +644,15 @@ static int UsableBytesInSegment; */ static XLogwrtResult LogwrtResult = {0, 0}; +/* + * True when an XLogWrite() performed by this process advanced the shared + * flushed position and backends waiting in WalBackendFlushAcquireOrWait() + * may need to be woken up. XLogWrite() runs with WALWriteLock held, so the + * wakeup is deferred until the caller has released the lock; see + * WalBackendFlushProcessWakeup(). + */ +static bool backendFlushWakeupPending = false; + /* * Update local copy of shared XLogCtl->log{Write,Flush}Result * @@ -705,6 +746,10 @@ static void ReserveXLogInsertLocation(int size, XLogRecPtr *StartPos, static bool ReserveXLogSwitch(XLogRecPtr *StartPos, XLogRecPtr *EndPos, XLogRecPtr *PrevPtr); static XLogRecPtr WaitXLogInsertionsToFinish(XLogRecPtr upto); +static bool WalBackendFlushTryAcquire(void); +static bool WalBackendFlushAcquireOrWait(XLogRecPtr record); +static void WalBackendFlushRelease(void); +static void WalBackendFlushProcessWakeup(void); static char *GetXLogBuffer(XLogRecPtr ptr, TimeLineID tli); static XLogRecPtr XLogBytePosToRecPtr(uint64 bytepos); static XLogRecPtr XLogBytePosToEndRecPtr(uint64 bytepos); @@ -2060,6 +2105,13 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, TimeLineID tli, bool opportunistic) WriteRqst.Flush = 0; XLogWrite(WriteRqst, tli, false); LWLockRelease(WALWriteLock); + + /* + * The flushed position advances here only when XLogWrite + * had to finish a segment, so waking the flush-limit + * waiters from this spot is rare. + */ + WalBackendFlushProcessWakeup(); pgWalUsage.wal_buffers_full++; TRACE_POSTGRESQL_WAL_BUFFER_WRITE_DIRTY_DONE(); @@ -2310,6 +2362,7 @@ XLogWrite(XLogwrtRqst WriteRqst, TimeLineID tli, bool flexible) int npages; int startidx; uint32 startoffset; + XLogRecPtr oldFlush; /* We should always be inside a critical section here */ Assert(CritSectionCount > 0); @@ -2318,6 +2371,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, @@ -2579,6 +2633,15 @@ XLogWrite(XLogwrtRqst WriteRqst, TimeLineID tli, bool flexible) pg_write_barrier(); pg_atomic_write_u64(&XLogCtl->logFlushResult, LogwrtResult.Flush); + /* + * If we advanced the flushed position, remember to wake up backends + * waiting in WalBackendFlushAcquireOrWait() once our caller has released + * WALWriteLock; waking them here would lengthen the hold time of the most + * contended WAL lock. + */ + if (oldFlush < LogwrtResult.Flush) + backendFlushWakeupPending = true; + #ifdef USE_ASSERT_CHECKING { XLogRecPtr Flush; @@ -2770,6 +2833,170 @@ UpdateMinRecoveryPoint(XLogRecPtr lsn, bool force) LWLockRelease(ControlFileLock); } +/* + * Try to acquire one of the wal_flush_backend_flushers slots. + * + * Returns true if the caller now holds a slot and must eventually call + * WalBackendFlushRelease(). Callers must have checked that the limit is + * enabled; it cannot change underneath us, see WalBackendFlushAcquireOrWait. + */ +static bool +WalBackendFlushTryAcquire(void) +{ + int limit = wal_flush_backend_flushers; + uint32 old_flushers; + + Assert(limit > 0); + + old_flushers = pg_atomic_read_u32(&XLogCtl->backendFlushers); + for (;;) + { + uint32 expected; + + if (old_flushers >= (uint32) limit) + return false; + + expected = old_flushers; + if (pg_atomic_compare_exchange_u32(&XLogCtl->backendFlushers, + &expected, old_flushers + 1)) + return true; + + old_flushers = expected; + } +} + +/* + * Release a slot acquired by WalBackendFlushTryAcquire() and hand it over + * to a waiter, if any. ConditionVariableSignal() removes the process it + * wakes from the wait queue, so the wakeup cannot be swallowed by a process + * that is no longer interested; a woken process that leaves without taking + * the slot passes the signal on (see WalBackendFlushAcquireOrWait). + */ +static void +WalBackendFlushRelease(void) +{ + pg_atomic_fetch_sub_u32(&XLogCtl->backendFlushers, 1); + + /* + * The fetch_sub above is a full barrier, so if a concurrent waiter missed + * us here (read backendFlushWaiters as zero), it is guaranteed to see the + * freed slot when it retries the acquire after registering. + */ + if (pg_atomic_read_u32(&XLogCtl->backendFlushWaiters) > 0) + ConditionVariableSignal(&XLogCtl->backendFlushCV); +} + +/* + * Wait until we acquire a WAL flusher slot, or until the wait becomes moot. + * + * Called by XLogFlush() when no slot was immediately available. Returns + * true if a slot was acquired, false if 'record' was flushed by someone + * else in the meantime. + * + * Note that wal_flush_backend_flushers cannot change while we are in here: + * config reload happens only between statements, so each XLogFlush() call + * works with a consistent value, keeping acquire and release balanced. + * + * This runs inside a critical section, which is fine: the condition + * variable sleep path performs no memory allocation, and + * CHECK_FOR_INTERRUPTS() is a no-op while in a critical section. + */ +static bool +WalBackendFlushAcquireOrWait(XLogRecPtr record) +{ + bool acquired = false; + + /* Fast path: slot free, no need to register as a waiter. */ + if (WalBackendFlushTryAcquire()) + return true; + + /* + * Publish the LSN we need flushed, so that the current slot holders + * extend their flush requests to cover it and we can piggyback on their + * fsyncs. Clamp the published value to the end of inserted WAL: an + * invalid 'record' (e.g. coming from a corrupted page LSN) must not cause + * every slot holder to request a flush past the end of generated WAL. The + * caller still detects and reports the bad LSN itself, just like in the + * unlimited case. + * + * The clamp must use the end-position conversion of the insert position, + * matching what WaitXLogInsertionsToFinish() compares its argument + * against. The start-position variant (GetXLogInsertRecPtr) points past + * the page header when the insert position sits exactly on a page + * boundary, and a request published from there would trip the + * flush-past-end-of-WAL complaint in the flushing backend. + * + * The maximum only ever advances; a stale-high value merely makes a + * holder flush WAL that has been inserted anyway, which is exactly the + * piggybacking XLogFlush() strives for. + */ + pg_atomic_monotonic_advance_u64(&XLogCtl->backendFlushRequest, + Min(record, GetXLogInsertEndRecPtr())); + + /* + * Register as a waiter before the final condition checks below. The + * fetch_add is a full barrier, pairing with the one in + * WalBackendFlushRelease() and with the flushed-position update in + * XLogWrite(): whoever misses our registration is guaranteed to be + * visible to our rechecks, and vice versa, so no wakeup can be lost. + */ + pg_atomic_fetch_add_u32(&XLogCtl->backendFlushWaiters, 1); + + INJECTION_POINT_CACHED("wal-backend-flush-wait", NULL); + + ConditionVariablePrepareToSleep(&XLogCtl->backendFlushCV); + for (;;) + { + /* Did someone else flush our record while we were waiting? */ + RefreshXLogWriteResult(LogwrtResult); + if (record <= LogwrtResult.Flush) + break; + + if (WalBackendFlushTryAcquire()) + { + acquired = true; + break; + } + + ConditionVariableSleep(&XLogCtl->backendFlushCV, + WAIT_EVENT_WAL_FLUSH_LIMIT); + } + + /* + * If we consumed a slot-release signal but are not taking the slot, pass + * the wakeup on so that the freed slot is not lost on a sleeping waiter. + */ + if (ConditionVariableCancelSleep() && !acquired) + ConditionVariableSignal(&XLogCtl->backendFlushCV); + + pg_atomic_fetch_sub_u32(&XLogCtl->backendFlushWaiters, 1); + + return acquired; +} + +/* + * Wake up backends waiting for their WAL to be flushed, if an XLogWrite() + * performed by this process advanced the flushed position. + * + * Call this after releasing WALWriteLock; the wakeups (one SetLatch per + * waiter) are too expensive to run under it. + */ +static void +WalBackendFlushProcessWakeup(void) +{ + if (!backendFlushWakeupPending) + return; + backendFlushWakeupPending = false; + + /* + * Waiters whose flush LSN is still not reached will re-check their + * condition and go back to sleep; that is the same wake-all behavior the + * WALWriteLock wait queue has always had. + */ + if (pg_atomic_read_u32(&XLogCtl->backendFlushWaiters) > 0) + ConditionVariableBroadcast(&XLogCtl->backendFlushCV); +} + /* * Ensure that all XLOG data through the given position is flushed to disk. * @@ -2782,6 +3009,7 @@ XLogFlush(XLogRecPtr record) XLogRecPtr WriteRqstPtr; XLogwrtRqst WriteRqst; TimeLineID insertTLI = XLogCtl->InsertTimeLineID; + bool backend_flusher_acquired = false; /* * During REDO, we are reading not writing WAL. Therefore, instead of @@ -2808,6 +3036,22 @@ XLogFlush(XLogRecPtr record) LSN_FORMAT_ARGS(LogwrtResult.Flush)); #endif + /* + * The injection points below run inside the critical section, where the + * first use of an injection point is not allowed to allocate; load them + * into the local cache beforehand. Loading is itself only safe when no + * critical section is active yet, and some callers (such as + * RecordTransactionCommit()) already hold one here, so a test process + * that must reach these points from such a caller has to pre-load them + * with injection_points_load(). + */ + if (wal_flush_backend_flushers > 0 && AmRegularBackendProcess() && + CritSectionCount == 0) + { + INJECTION_POINT_LOAD("wal-backend-flush-wait"); + INJECTION_POINT_LOAD("wal-backend-flush-after-acquire"); + } + START_CRIT_SECTION(); /* @@ -2842,6 +3086,27 @@ XLogFlush(XLogRecPtr record) if (WriteRqstPtr < XLogCtl->LogwrtRqst.Write) WriteRqstPtr = XLogCtl->LogwrtRqst.Write; SpinLockRelease(&XLogCtl->info_lck); + + /* + * If the number of concurrent backend WAL flushers is limited, + * acquire a flusher slot before proceeding, and keep it until we are + * done: a slot holder that is merely waiting for WALWriteLock is + * still one of the configured number of active flushers. While + * holding a slot, extend our request to also cover the WAL the queued + * waiters need flushed, so they piggyback on our flush. + */ + if (!backend_flusher_acquired && + wal_flush_backend_flushers > 0 && AmRegularBackendProcess()) + { + if (!WalBackendFlushAcquireOrWait(record)) + continue; + backend_flusher_acquired = true; + INJECTION_POINT_CACHED("wal-backend-flush-after-acquire", NULL); + } + if (backend_flusher_acquired) + WriteRqstPtr = Max(WriteRqstPtr, + pg_atomic_read_u64(&XLogCtl->backendFlushRequest)); + insertpos = WaitXLogInsertionsToFinish(WriteRqstPtr); /* @@ -2907,8 +3172,14 @@ XLogFlush(XLogRecPtr record) break; } + if (backend_flusher_acquired) + WalBackendFlushRelease(); + END_CRIT_SECTION(); + /* wake up waiters now that we've released heavily contended locks */ + WalBackendFlushProcessWakeup(); + /* wake up walsenders now that we've released heavily contended locks */ WalSndWakeupProcessRequests(true, !RecoveryInProgress()); @@ -3084,6 +3355,9 @@ XLogBackgroundFlush(void) END_CRIT_SECTION(); + /* wake up waiters now that we've released heavily contended locks */ + WalBackendFlushProcessWakeup(); + /* wake up walsenders now that we've released heavily contended locks */ WalSndWakeupProcessRequests(true, !RecoveryInProgress()); @@ -5060,6 +5334,10 @@ XLOGShmemInit(void) pg_atomic_init_u64(&XLogCtl->logInsertResult, InvalidXLogRecPtr); pg_atomic_init_u64(&XLogCtl->logWriteResult, InvalidXLogRecPtr); pg_atomic_init_u64(&XLogCtl->logFlushResult, InvalidXLogRecPtr); + pg_atomic_init_u32(&XLogCtl->backendFlushers, 0); + pg_atomic_init_u32(&XLogCtl->backendFlushWaiters, 0); + pg_atomic_init_u64(&XLogCtl->backendFlushRequest, InvalidXLogRecPtr); + ConditionVariableInit(&XLogCtl->backendFlushCV); pg_atomic_init_u64(&XLogCtl->unloggedLSN, InvalidXLogRecPtr); } diff --git a/src/backend/utils/activity/wait_event_names.txt b/src/backend/utils/activity/wait_event_names.txt index b9c1e6900ec1b..069135071714c 100644 --- a/src/backend/utils/activity/wait_event_names.txt +++ b/src/backend/utils/activity/wait_event_names.txt @@ -163,6 +163,8 @@ XACT_GROUP_UPDATE "Waiting for the group leader to update transaction status at ABI_compatibility: +WAL_FLUSH_LIMIT "Waiting to become one of the limited number of concurrent WAL flushers, or for another process to flush the needed WAL." + # # Wait Events - Timeout # diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c index 6b82a23435efe..1319a02b861b1 100644 --- a/src/backend/utils/misc/guc_tables.c +++ b/src/backend/utils/misc/guc_tables.c @@ -3048,6 +3048,16 @@ struct config_int ConfigureNamesInt[] = NULL, NULL, NULL }, + { + {"wal_flush_backend_flushers", PGC_SIGHUP, WAL_SETTINGS, + gettext_noop("Sets the maximum number of client backends that can flush WAL concurrently."), + gettext_noop("0 disables the limit.") + }, + &wal_flush_backend_flushers, + 0, 0, MAX_BACKENDS, + NULL, NULL, NULL + }, + { {"wal_skip_threshold", PGC_USERSET, WAL_SETTINGS, gettext_noop("Minimum size of new file to fsync instead of writing WAL."), diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample index d91133dbd7357..49bb342631ebb 100644 --- a/src/backend/utils/misc/postgresql.conf.sample +++ b/src/backend/utils/misc/postgresql.conf.sample @@ -257,6 +257,7 @@ # (change requires restart) #wal_writer_delay = 200ms # 1-10000 milliseconds #wal_writer_flush_after = 1MB # measured in pages, 0 disables +#wal_flush_backend_flushers = 0 # 0 disables the limit #wal_skip_threshold = 2MB #commit_delay = 0 # range 0-100000, in microseconds diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h index f20f5edb43876..47c9c80cac96e 100644 --- a/src/include/access/xlog.h +++ b/src/include/access/xlog.h @@ -54,6 +54,7 @@ extern PGDLLIMPORT char *wal_consistency_checking_string; extern PGDLLIMPORT bool log_checkpoints; extern PGDLLIMPORT int CommitDelay; extern PGDLLIMPORT int CommitSiblings; +extern PGDLLIMPORT int wal_flush_backend_flushers; extern PGDLLIMPORT bool track_wal_io_timing; extern PGDLLIMPORT int wal_decode_buffer_size; diff --git a/src/test/recovery/meson.build b/src/test/recovery/meson.build index 38e1e43e04154..df6d854f3a5ae 100644 --- a/src/test/recovery/meson.build +++ b/src/test/recovery/meson.build @@ -59,6 +59,7 @@ tests += { 't/048_vacuum_horizon_floor.pl', 't/053_standby_login_event_trigger.pl', 't/054_unlogged_sequence_promotion.pl', + 't/055_wal_flush_backend_flushers.pl', ], }, } diff --git a/src/test/recovery/t/055_wal_flush_backend_flushers.pl b/src/test/recovery/t/055_wal_flush_backend_flushers.pl new file mode 100644 index 0000000000000..91b5a26e720d6 --- /dev/null +++ b/src/test/recovery/t/055_wal_flush_backend_flushers.pl @@ -0,0 +1,243 @@ +# Copyright (c) 2026, PostgreSQL Global Development Group + +# Test concurrent commits while limiting backend processes that may flush WAL. +use strict; +use warnings FATAL => 'all'; + +use PostgreSQL::Test::Cluster; +use Test::More; + +my $use_injection_points = ($ENV{enable_injection_points} // '') eq 'yes'; + +my $node = PostgreSQL::Test::Cluster->new('wal_flush_backend_flushers'); +$node->init; +$node->append_conf( + 'postgresql.conf', q{ +wal_flush_backend_flushers = 1 +}); +$node->start; + +is($node->safe_psql('postgres', 'SHOW wal_flush_backend_flushers'), + '1', 'wal_flush_backend_flushers is set'); + +$node->safe_psql('postgres', + 'CREATE TABLE wal_flush_backend_flushers_probe (id bigserial)'); +$node->safe_psql('postgres', + 'CREATE TABLE wal_flush_backend_flushers_test (id bigserial)'); + +if ($use_injection_points && $node->check_extension('injection_points')) +{ + # The wait-mode injection points used below fire inside critical + # sections, where the injection_points module must not allocate its + # lazily-initialized local state. Preloading the module sets that + # state up at backend start instead (compare test_slru/t/001_multixact.pl, + # which waits inside a critical section the same way). The availability + # check above ran without the preload, so a build without the module + # still skips this section gracefully instead of failing to start. + $node->append_conf( + 'postgresql.conf', q{ +shared_preload_libraries = 'injection_points' +}); + $node->restart; + + $node->safe_psql('postgres', 'CREATE EXTENSION injection_points'); + $node->safe_psql( + 'postgres', q{ +SELECT injection_points_attach('wal-backend-flush-after-acquire', 'wait'); +SELECT injection_points_attach('wal-backend-flush-wait', 'wait'); +}); + + # The injection points fire during the commit's XLogFlush, which runs + # inside the caller's critical section where the points cannot be loaded + # into the local cache anymore; pre-load them in each session first. + my $preload_points = q{ +SELECT injection_points_load('wal-backend-flush-after-acquire'); +SELECT injection_points_load('wal-backend-flush-wait'); +}; + + # The flusher acquires the single flusher slot and then blocks at the + # injection point, holding the slot. + my $flusher = $node->background_psql('postgres', on_error_stop => 1); + $flusher->query_safe($preload_points); + $flusher->query_until( + qr/flusher_started/, + q{ +\echo flusher_started +INSERT INTO wal_flush_backend_flushers_probe DEFAULT VALUES; +\echo flusher_done +}); + $node->wait_for_event('client backend', + 'wal-backend-flush-after-acquire'); + + # With the only slot taken, a second backend fails to acquire one and + # reaches the wait path. Both injection points sit in code reachable + # only by client backends, so no background process can consume them. + my $waiter = $node->background_psql('postgres', on_error_stop => 1); + $waiter->query_safe($preload_points); + $waiter->query_until( + qr/waiter_started/, + q{ +\echo waiter_started +INSERT INTO wal_flush_backend_flushers_probe DEFAULT VALUES; +\echo waiter_done +}); + $node->wait_for_event('client backend', 'wal-backend-flush-wait'); + + pass('a backend waited in the limited WAL flush queue'); + + # Let the waiter proceed into its condition-variable sleep, then let the + # flusher finish: it flushes the waiter's WAL too, releases the slot and + # wakes the waiter. Each backend passes its injection point exactly + # once, so a single wakeup per point suffices. + $node->safe_psql('postgres', + q{SELECT injection_points_wakeup('wal-backend-flush-wait')}); + $node->safe_psql('postgres', + q{SELECT injection_points_wakeup('wal-backend-flush-after-acquire')}); + + $flusher->query_until(qr/flusher_done/, ''); + $waiter->query_until(qr/waiter_done/, ''); + pass('waiter completed after the flusher released the slot'); + + $flusher->quit; + $waiter->quit; + + # Second scenario: the slot is released and the flushed position + # advances while the waiter is registered but not yet sleeping on the + # condition variable (parked at the injection point placed before + # ConditionVariablePrepareToSleep), so both the handoff signal and the + # broadcast find an empty wait queue. The waiter must not rely on them: + # rechecking the flushed position after preparing to sleep has to let it + # complete. + my $flusher2 = $node->background_psql('postgres', on_error_stop => 1); + $flusher2->query_safe($preload_points); + $flusher2->query_until( + qr/flusher2_started/, + q{ +\echo flusher2_started +INSERT INTO wal_flush_backend_flushers_probe DEFAULT VALUES; +\echo flusher2_done +}); + $node->wait_for_event('client backend', + 'wal-backend-flush-after-acquire'); + + my $waiter2 = $node->background_psql('postgres', on_error_stop => 1); + $waiter2->query_safe($preload_points); + $waiter2->query_until( + qr/waiter2_started/, + q{ +\echo waiter2_started +INSERT INTO wal_flush_backend_flushers_probe DEFAULT VALUES; +\echo waiter2_done +}); + $node->wait_for_event('client backend', 'wal-backend-flush-wait'); + + # Let the flusher finish first: its group flush covers the waiter's + # already-published LSN, and its slot release signals an empty queue. + $node->safe_psql('postgres', + q{SELECT injection_points_wakeup('wal-backend-flush-after-acquire')}); + $flusher2->query_until(qr/flusher2_done/, ''); + + # Only now release the waiter; it must notice that its WAL has been + # flushed and complete without any wakeup arriving. + $node->safe_psql('postgres', + q{SELECT injection_points_wakeup('wal-backend-flush-wait')}); + $waiter2->query_until(qr/waiter2_done/, ''); + pass('waiter completed although wakeups preceded its sleep'); + + $flusher2->quit; + $waiter2->quit; + + $node->safe_psql( + 'postgres', q{ +SELECT injection_points_detach('wal-backend-flush-after-acquire'); +SELECT injection_points_detach('wal-backend-flush-wait'); +}); +} +elsif ($use_injection_points) +{ + note( + 'extension injection_points not installed; skipping wait/wakeup probe' + ); +} +else +{ + note( + 'injection points not supported by this build; skipping wait/wakeup probe' + ); +} + +$node->pgbench( + '--no-vacuum --client=8 --jobs=4 --transactions=200', + 0, + [qr/number of failed transactions: 0/], + [], + 'concurrent commits with one backend allowed to flush WAL', + { + 'wal_flush_backend_flushers.pgb' => + 'INSERT INTO wal_flush_backend_flushers_test DEFAULT VALUES' + }); + +is( $node->safe_psql( + 'postgres', 'SELECT count(*) FROM wal_flush_backend_flushers_test'), + '1600', + 'all concurrent WAL-writing transactions committed'); + +# The limit can be changed with a reload; make sure commits still work after +# disabling it on the fly. +$node->adjust_conf('postgresql.conf', 'wal_flush_backend_flushers', '0'); +$node->reload; +$node->safe_psql('postgres', + 'INSERT INTO wal_flush_backend_flushers_test DEFAULT VALUES'); +is( $node->safe_psql( + 'postgres', 'SELECT count(*) FROM wal_flush_backend_flushers_test'), + '1601', + 'commits work after disabling the limit via reload'); + +# Stress the reload path: keep sessions committing while the limit bounces +# between off, high and one. Any acquire/release imbalance across the +# transitions would permanently wedge the flusher slots (with the limit back +# at one, a leaked slot blocks every commit), which shows up here as a hang. +my @churners = + map { $node->background_psql('postgres', on_error_stop => 1) } (1 .. 3); +my @limits = ('3', '1', '0', '1'); +my $churn_rows = 0; +foreach my $limit (@limits) +{ + $node->adjust_conf('postgresql.conf', 'wal_flush_backend_flushers', + $limit); + $node->reload; + for my $i (1 .. 5) + { + foreach my $session (@churners) + { + $session->query( + 'INSERT INTO wal_flush_backend_flushers_test DEFAULT VALUES'); + $churn_rows++; + } + } +} +$_->quit foreach @churners; +is( $node->safe_psql( + 'postgres', 'SELECT count(*) FROM wal_flush_backend_flushers_test'), + 1601 + $churn_rows, + 'commits survive limit changes between off, high and one'); + +# The published flush request is clamped to the end position of inserted +# WAL; a wrong clamp (e.g. the start-position conversion, which lands past +# the page header on page boundaries) surfaces as this complaint from the +# flushing backend. The trigger needs an unlucky insert position, so a +# clean pass is only probabilistic insurance, but the message must never +# appear with a correct clamp. +ok( !$node->log_contains('request to flush past end of generated WAL'), + 'no flush-past-end complaints from limited backend flushing'); + +# All the commits above went through the limited flush path with +# synchronous_commit on, so they must survive a crash. +$node->stop('immediate'); +$node->start; +is( $node->safe_psql( + 'postgres', 'SELECT count(*) FROM wal_flush_backend_flushers_test'), + 1601 + $churn_rows, + 'limited-path commits are durable across a crash'); + +done_testing();