On Mon, Jul 15, 2024 at 2:02 PM Alexander Korotkov <aekorot...@gmail.com> wrote: > On Mon, Jul 15, 2024 at 4:24 AM Alexander Korotkov <aekorot...@gmail.com> > wrote: > > Thanks to Kyotaro for the review. And thanks to Ivan for the patch > > revision. I made another revision of the patch. > > I've noticed failures on cfbot. The attached revision addressed docs > build failure. Also it adds some "quits" for background psql sessions > for tests. Probably this will address test hangs on windows.
I made the following changes to the patch. 1) I've changed the check for snapshot in pg_wal_replay_wait(). Now it checks that GetOldestSnapshot() returns NULL. It happens when both ActiveSnapshot is NULL and RegisteredSnapshots pairing heap is empty. This is the same condition when SnapshotResetXmin() sets out xmin to invalid. Thus, we are not preventing WAL from replay. This should be satisfied when pg_wal_replay_wait() isn't called within a transaction with an isolation level higher than READ COMMITTED, another procedure, or a function. Documented it this way. 2) Explicitly document in the PortalRunUtility() comment that pg_wal_replay_wait() is another case when active snapshot gets released. 3) I've removed tests with cascading replication. It's rather unclear what problem these tests could potentially spot. 4) Did some improvements to docs, comments and commit message to make them consistent with the patch contents. The commit to pg17 was inconsiderate. But I feel this patch got much better shape. Especially, now it's clear when the pg_wal_replay_wait() procedure can be used. So, I dare commit this to pg18 if nobody objects. ------ Regards, Alexander Korotkov Supabase
From f6713100b6231f2d124a98021d9fbc353caad0db Mon Sep 17 00:00:00 2001 From: Alexander Korotkov <akorotkov@postgresql.org> Date: Wed, 31 Jul 2024 19:13:55 +0300 Subject: [PATCH v23] Implement pg_wal_replay_wait() stored procedure pg_wal_replay_wait() is to be used on standby and specifies waiting for the specific WAL location to be replayed. This option is useful when the user makes some data changes on primary and needs a guarantee to see these changes are on standby. The queue of waiters is stored in the shared memory as an LSN-ordered pairing heap, where the waiter with the nearest LSN stays on the top. During the replay of WAL, waiters whose LSNs have already been replayed are deleted from the shared memory pairing heap and woken up by setting their latches. pg_wal_replay_wait() needs to wait without any snapshot held. Otherwise, the snapshot could prevent the replay of WAL records, implying a kind of self-deadlock. This is why it is only possible to implement pg_wal_replay_wait() as a procedure working without an active snapshot, not a function. Catversion is bumped. Discussion: https://postgr.es/m/eb12f9b03851bb2583adab5df9579b4b%40postgrespro.ru Author: Kartyshov Ivan, Alexander Korotkov Reviewed-by: Michael Paquier, Peter Eisentraut, Dilip Kumar, Amit Kapila Reviewed-by: Alexander Lakhin, Bharath Rupireddy, Euler Taveira Reviewed-by: Heikki Linnakangas, Kyotaro Horiguchi --- doc/src/sgml/func.sgml | 117 ++++++ src/backend/access/transam/xact.c | 6 + src/backend/access/transam/xlog.c | 7 + src/backend/access/transam/xlogrecovery.c | 11 + src/backend/catalog/system_functions.sql | 3 + src/backend/commands/Makefile | 3 +- src/backend/commands/meson.build | 1 + src/backend/commands/waitlsn.c | 363 ++++++++++++++++++ src/backend/lib/pairingheap.c | 18 +- src/backend/storage/ipc/ipci.c | 3 + src/backend/storage/lmgr/proc.c | 6 + src/backend/tcop/pquery.c | 9 +- .../utils/activity/wait_event_names.txt | 2 + src/include/catalog/pg_proc.dat | 6 + src/include/commands/waitlsn.h | 80 ++++ src/include/lib/pairingheap.h | 3 + src/include/storage/lwlocklist.h | 1 + src/test/recovery/meson.build | 1 + src/test/recovery/t/043_wal_replay_wait.pl | 150 ++++++++ src/tools/pgindent/typedefs.list | 2 + 20 files changed, 785 insertions(+), 7 deletions(-) create mode 100644 src/backend/commands/waitlsn.c create mode 100644 src/include/commands/waitlsn.h create mode 100644 src/test/recovery/t/043_wal_replay_wait.pl diff --git a/doc/src/sgml/func.sgml b/doc/src/sgml/func.sgml index b39f97dc8de..3cf896b22fa 100644 --- a/doc/src/sgml/func.sgml +++ b/doc/src/sgml/func.sgml @@ -28911,6 +28911,123 @@ postgres=# SELECT '0/0'::pg_lsn + pd.segment_number * ps.setting::int + :offset the pause, the rate of WAL generation and available disk space. </para> + <para> + The procedure shown in <xref linkend="recovery-synchronization-procedure-table"/> + can be executed only during recovery. + </para> + + <table id="recovery-synchronization-procedure-table"> + <title>Recovery Synchronization Procedure</title> + <tgroup cols="1"> + <thead> + <row> + <entry role="func_table_entry"><para role="func_signature"> + Procedure + </para> + <para> + Description + </para></entry> + </row> + </thead> + + <tbody> + <row> + <entry role="func_table_entry"><para role="func_signature"> + <indexterm> + <primary>pg_wal_replay_wait</primary> + </indexterm> + <function>pg_wal_replay_wait</function> ( + <parameter>target_lsn</parameter> <type>pg_lsn</type>, + <parameter>timeout</parameter> <type>bigint</type> <literal>DEFAULT</literal> <literal>0</literal>) + <returnvalue>void</returnvalue> + </para> + <para> + Waits until recovery replays <literal>target_lsn</literal>. + If no <parameter>timeout</parameter> is specified or it is set to + zero, this procedure waits indefinitely for the + <literal>target_lsn</literal>. If the <parameter>timeout</parameter> + is specified (in milliseconds) and is greater than zero, the + procedure waits until <literal>target_lsn</literal> is reached or + the specified <parameter>timeout</parameter> has elapsed. + On timeout, or if the server is promoted before + <literal>target_lsn</literal> is reached, an error is emitted. + </para></entry> + </row> + </tbody> + </tgroup> + </table> + + <para> + <function>pg_wal_replay_wait</function> waits till + <parameter>target_lsn</parameter> to be replayed on standby. + That is, after this function execution, the value returned by + <function>pg_last_wal_replay_lsn</function> should be greater or equal + to the <parameter>target_lsn</parameter> value. This is useful to achieve + read-your-writes-consistency, while using async replica for reads and + primary for writes. In that case <acronym>lsn</acronym> of the last + modification should be stored on the client application side or the + connection pooler side. + </para> + + <para> + You can use <function>pg_wal_replay_wait</function> to wait for + the <type>pg_lsn</type> value. For example, an application could update + the <literal>movie</literal> table and get the <acronym>lsn</acronym> after + changes just made. This example uses <function>pg_current_wal_insert_lsn</function> + on primary server to get the <acronym>lsn</acronym> given that + <varname>synchronous_commit</varname> could be set to + <literal>off</literal>. + + <programlisting> +postgres=# UPDATE movie SET genre = 'Dramatic' WHERE genre = 'Drama'; +UPDATE 100 +postgres=# SELECT pg_current_wal_insert_lsn(); +pg_current_wal_insert_lsn +-------------------- +0/306EE20 +(1 row) + </programlisting> + + Then an application could run <function>pg_wal_replay_wait</function> + with the <acronym>lsn</acronym> obtained from primary. After that the + changes made of primary should be guaranteed to be visible on replica. + + <programlisting> +postgres=# CALL pg_wal_replay_wait('0/306EE20'); +CALL +postgres=# SELECT * FROM movie WHERE genre = 'Drama'; + genre +------- +(0 rows) + </programlisting> + + It may also happen that target <acronym>lsn</acronym> is not achieved + within the timeout. In that case the error is thrown. + + <programlisting> +postgres=# CALL pg_wal_replay_wait('0/306EE20', 100); +ERROR: timed out while waiting for target LSN 0/306EE20 to be replayed; current replay LSN 0/306EA60 + </programlisting> + + </para> + + <para> + <function>pg_wal_replay_wait</function> can't be used within + a transaction with an isolation level higher than + <literal>READ COMMITTED</literal>, another procedure, or a function. + All the cases above imply holding a snapshot, which could prevent + WAL records from replaying (see <xref linkend="hot-standby-conflict"/>) + and cause an indirect deadlock. + + <programlisting> +postgres=# BEGIN; +BEGIN +postgres=*# CALL pg_wal_replay_wait('0/306EE20'); +ERROR: pg_wal_replay_wait() must be only called without an active or registered snapshot +DETAIL: Make sure pg_wal_replay_wait() isn't called within a transaction with an isolation level higher than READ COMMITTED, another procedure, or a function. + </programlisting> + + </para> </sect2> <sect2 id="functions-snapshot-synchronization"> diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c index d119ab909dc..dfc8cf2dcf2 100644 --- a/src/backend/access/transam/xact.c +++ b/src/backend/access/transam/xact.c @@ -38,6 +38,7 @@ #include "commands/async.h" #include "commands/tablecmds.h" #include "commands/trigger.h" +#include "commands/waitlsn.h" #include "common/pg_prng.h" #include "executor/spi.h" #include "libpq/be-fsstubs.h" @@ -2809,6 +2810,11 @@ AbortTransaction(void) */ LWLockReleaseAll(); + /* + * Cleanup waiting for LSN if any. + */ + WaitLSNCleanup(); + /* Clear wait information and command progress indicator */ pgstat_report_wait_end(); pgstat_progress_end_command(); diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index f86f4b5c4b7..d7fd1c9823f 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -66,6 +66,7 @@ #include "catalog/catversion.h" #include "catalog/pg_control.h" #include "catalog/pg_database.h" +#include "commands/waitlsn.h" #include "common/controldata_utils.h" #include "common/file_utils.h" #include "executor/instrument.h" @@ -6143,6 +6144,12 @@ StartupXLOG(void) UpdateControlFile(); LWLockRelease(ControlFileLock); + /* + * Wake up all waiters for replay LSN. They need to report an error that + * recovery was ended before achieving the target LSN. + */ + WaitLSNSetLatches(InvalidXLogRecPtr); + /* * Shutdown the recovery environment. This must occur after * RecoverPreparedTransactions() (see notes in lock_twophase_recover()) diff --git a/src/backend/access/transam/xlogrecovery.c b/src/backend/access/transam/xlogrecovery.c index 2ed3ea2b45b..ad817fbca67 100644 --- a/src/backend/access/transam/xlogrecovery.c +++ b/src/backend/access/transam/xlogrecovery.c @@ -43,6 +43,7 @@ #include "backup/basebackup.h" #include "catalog/pg_control.h" #include "commands/tablespace.h" +#include "commands/waitlsn.h" #include "common/file_utils.h" #include "miscadmin.h" #include "pgstat.h" @@ -1828,6 +1829,16 @@ PerformWalRecovery(void) break; } + /* + * If we replayed an LSN that someone was waiting for then walk + * over the shared memory array and set latches to notify the + * waiters. + */ + if (waitLSNState && + (XLogRecoveryCtl->lastReplayedEndRecPtr >= + pg_atomic_read_u64(&waitLSNState->minWaitedLSN))) + WaitLSNSetLatches(XLogRecoveryCtl->lastReplayedEndRecPtr); + /* Else, try to fetch the next WAL record */ record = ReadRecord(xlogprefetcher, LOG, false, replayTLI); } while (record != NULL); diff --git a/src/backend/catalog/system_functions.sql b/src/backend/catalog/system_functions.sql index ae099e328c2..623b9539b15 100644 --- a/src/backend/catalog/system_functions.sql +++ b/src/backend/catalog/system_functions.sql @@ -414,6 +414,9 @@ CREATE OR REPLACE FUNCTION json_populate_recordset(base anyelement, from_json json, use_json_as_text boolean DEFAULT false) RETURNS SETOF anyelement LANGUAGE internal STABLE ROWS 100 AS 'json_populate_recordset' PARALLEL SAFE; +CREATE OR REPLACE PROCEDURE pg_wal_replay_wait(target_lsn pg_lsn, timeout int8 DEFAULT 0) + LANGUAGE internal AS 'pg_wal_replay_wait'; + CREATE OR REPLACE FUNCTION pg_logical_slot_get_changes( IN slot_name name, IN upto_lsn pg_lsn, IN upto_nchanges int, VARIADIC options text[] DEFAULT '{}', OUT lsn pg_lsn, OUT xid xid, OUT data text) diff --git a/src/backend/commands/Makefile b/src/backend/commands/Makefile index 48f7348f91c..cede90c3b98 100644 --- a/src/backend/commands/Makefile +++ b/src/backend/commands/Makefile @@ -61,6 +61,7 @@ OBJS = \ vacuum.o \ vacuumparallel.o \ variable.o \ - view.o + view.o \ + waitlsn.o include $(top_srcdir)/src/backend/common.mk diff --git a/src/backend/commands/meson.build b/src/backend/commands/meson.build index 6dd00a4abde..7549be5dc3b 100644 --- a/src/backend/commands/meson.build +++ b/src/backend/commands/meson.build @@ -50,4 +50,5 @@ backend_sources += files( 'vacuumparallel.c', 'variable.c', 'view.c', + 'waitlsn.c', ) diff --git a/src/backend/commands/waitlsn.c b/src/backend/commands/waitlsn.c new file mode 100644 index 00000000000..3170f0792a5 --- /dev/null +++ b/src/backend/commands/waitlsn.c @@ -0,0 +1,363 @@ +/*------------------------------------------------------------------------- + * + * waitlsn.c + * Implements waiting for the given replay LSN, which is used in + * CALL pg_wal_replay_wait(target_lsn pg_lsn, timeout float8). + * + * Copyright (c) 2024, PostgreSQL Global Development Group + * + * IDENTIFICATION + * src/backend/commands/waitlsn.c + * + *------------------------------------------------------------------------- + */ + +#include "postgres.h" + +#include <float.h> +#include <math.h> + +#include "pgstat.h" +#include "access/xlog.h" +#include "access/xlogrecovery.h" +#include "commands/waitlsn.h" +#include "funcapi.h" +#include "miscadmin.h" +#include "storage/latch.h" +#include "storage/proc.h" +#include "storage/shmem.h" +#include "utils/fmgrprotos.h" +#include "utils/pg_lsn.h" +#include "utils/snapmgr.h" +#include "utils/wait_event_types.h" + +static int waitlsn_cmp(const pairingheap_node *a, const pairingheap_node *b, + void *arg); + +struct WaitLSNState *waitLSNState = NULL; + +/* Report the amount of shared memory space needed for WaitLSNState. */ +Size +WaitLSNShmemSize(void) +{ + Size size; + + size = offsetof(WaitLSNState, procInfos); + size = add_size(size, mul_size(MaxBackends, sizeof(WaitLSNProcInfo))); + return size; +} + +/* Initialize the WaitLSNState in the shared memory. */ +void +WaitLSNShmemInit(void) +{ + bool found; + + waitLSNState = (WaitLSNState *) ShmemInitStruct("WaitLSNState", + WaitLSNShmemSize(), + &found); + if (!found) + { + pg_atomic_init_u64(&waitLSNState->minWaitedLSN, PG_UINT64_MAX); + pairingheap_initialize(&waitLSNState->waitersHeap, waitlsn_cmp, NULL); + memset(&waitLSNState->procInfos, 0, MaxBackends * sizeof(WaitLSNProcInfo)); + } +} + +/* + * Comparison function for waitLSN->waitersHeap heap. Waiting processes are + * ordered by lsn, so that the waiter with smallest lsn is at the top. + */ +static int +waitlsn_cmp(const pairingheap_node *a, const pairingheap_node *b, void *arg) +{ + const WaitLSNProcInfo *aproc = pairingheap_const_container(WaitLSNProcInfo, phNode, a); + const WaitLSNProcInfo *bproc = pairingheap_const_container(WaitLSNProcInfo, phNode, b); + + if (aproc->waitLSN < bproc->waitLSN) + return 1; + else if (aproc->waitLSN > bproc->waitLSN) + return -1; + else + return 0; +} + +/* + * Update waitLSN->minWaitedLSN according to the current state of + * waitLSN->waitersHeap. + */ +static void +updateMinWaitedLSN(void) +{ + XLogRecPtr minWaitedLSN = PG_UINT64_MAX; + + if (!pairingheap_is_empty(&waitLSNState->waitersHeap)) + { + pairingheap_node *node = pairingheap_first(&waitLSNState->waitersHeap); + + minWaitedLSN = pairingheap_container(WaitLSNProcInfo, phNode, node)->waitLSN; + } + + pg_atomic_write_u64(&waitLSNState->minWaitedLSN, minWaitedLSN); +} + +/* + * Put the current process into the heap of LSN waiters. + */ +static void +addLSNWaiter(XLogRecPtr lsn) +{ + WaitLSNProcInfo *procInfo = &waitLSNState->procInfos[MyProcNumber]; + + LWLockAcquire(WaitLSNLock, LW_EXCLUSIVE); + + Assert(!procInfo->inHeap); + + procInfo->latch = MyLatch; + procInfo->waitLSN = lsn; + + pairingheap_add(&waitLSNState->waitersHeap, &procInfo->phNode); + procInfo->inHeap = true; + updateMinWaitedLSN(); + + LWLockRelease(WaitLSNLock); +} + +/* + * Remove the current process from the heap of LSN waiters if it's there. + */ +static void +deleteLSNWaiter(void) +{ + WaitLSNProcInfo *procInfo = &waitLSNState->procInfos[MyProcNumber]; + + LWLockAcquire(WaitLSNLock, LW_EXCLUSIVE); + + if (!procInfo->inHeap) + { + LWLockRelease(WaitLSNLock); + return; + } + + pairingheap_remove(&waitLSNState->waitersHeap, &procInfo->phNode); + procInfo->inHeap = false; + updateMinWaitedLSN(); + + LWLockRelease(WaitLSNLock); +} + +/* + * Set latches of LSN waiters whose LSN has been replayed. Set latches of all + * LSN waiters when InvalidXLogRecPtr is given. + */ +void +WaitLSNSetLatches(XLogRecPtr currentLSN) +{ + int i; + Latch **wakeUpProcLatches; + int numWakeUpProcs = 0; + + wakeUpProcLatches = palloc(sizeof(Latch *) * MaxBackends); + + LWLockAcquire(WaitLSNLock, LW_EXCLUSIVE); + + /* + * Iterate the pairing heap of waiting processes till we find LSN not yet + * replayed. Record the process latches to set them later. + */ + while (!pairingheap_is_empty(&waitLSNState->waitersHeap)) + { + pairingheap_node *node = pairingheap_first(&waitLSNState->waitersHeap); + WaitLSNProcInfo *procInfo = pairingheap_container(WaitLSNProcInfo, phNode, node); + + if (!XLogRecPtrIsInvalid(currentLSN) && + procInfo->waitLSN > currentLSN) + break; + + wakeUpProcLatches[numWakeUpProcs++] = procInfo->latch; + (void) pairingheap_remove_first(&waitLSNState->waitersHeap); + procInfo->inHeap = false; + } + + updateMinWaitedLSN(); + + LWLockRelease(WaitLSNLock); + + /* + * Set latches for processes, whose waited LSNs are already replayed. As + * the time consuming operations, we do it this outside of WaitLSNLock. + * This is actually fine because procLatch isn't ever freed, so we just + * can potentially set the wrong process' (or no process') latch. + */ + for (i = 0; i < numWakeUpProcs; i++) + { + SetLatch(wakeUpProcLatches[i]); + } + pfree(wakeUpProcLatches); +} + +/* + * Delete our item from shmem array if any. + */ +void +WaitLSNCleanup(void) +{ + /* + * We do a fast-path check of the 'inHeap' flag without the lock. This + * flag is set to true only by the process itself. So, it's only possible + * to get a false positive. But that will be eliminated by a recheck + * inside deleteLSNWaiter(). + */ + if (waitLSNState->procInfos[MyProcNumber].inHeap) + deleteLSNWaiter(); +} + +/* + * Wait using MyLatch till the given LSN is replayed, the postmaster dies or + * timeout happens. + */ +static void +WaitForLSNReplay(XLogRecPtr targetLSN, int64 timeout) +{ + XLogRecPtr currentLSN; + TimestampTz endtime = 0; + int wake_events = WL_LATCH_SET | WL_EXIT_ON_PM_DEATH; + + /* Shouldn't be called when shmem isn't initialized */ + Assert(waitLSNState); + + /* Should have a valid proc number */ + Assert(MyProcNumber >= 0 && MyProcNumber < MaxBackends); + + if (!RecoveryInProgress()) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("recovery is not in progress"), + errhint("Waiting for LSN can only be executed during recovery."))); + + /* If target LSN is already replayed, exit immediately */ + if (targetLSN <= GetXLogReplayRecPtr(NULL)) + return; + + if (timeout > 0) + { + endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), timeout); + wake_events |= WL_TIMEOUT; + } + + /* + * Add our process to the pairing heap of waiters. It might happen that + * target LSN gets replayed before we do. Another check at the beginning + * of the loop below prevents the race condition. + */ + addLSNWaiter(targetLSN); + + for (;;) + { + int rc; + long delay_ms = 0; + + /* Check if the waited LSN has been replayed */ + currentLSN = GetXLogReplayRecPtr(NULL); + if (targetLSN <= currentLSN) + break; + + /* Recheck that recovery is still in-progress */ + if (!RecoveryInProgress()) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("recovery is not in progress"), + errdetail("Recovery ended before replaying target LSN %X/%X; last replay LSN %X/%X.", + LSN_FORMAT_ARGS(targetLSN), + LSN_FORMAT_ARGS(currentLSN)))); + + /* + * If the timeout value is specified, calculate the number of + * milliseconds before the timeout. Exit if the timeout is already + * achieved. + */ + if (timeout > 0) + { + delay_ms = TimestampDifferenceMilliseconds(GetCurrentTimestamp(), endtime); + if (delay_ms <= 0) + break; + } + + CHECK_FOR_INTERRUPTS(); + + rc = WaitLatch(MyLatch, wake_events, delay_ms, + WAIT_EVENT_WAIT_FOR_WAL_REPLAY); + + if (rc & WL_LATCH_SET) + ResetLatch(MyLatch); + } + + /* + * Delete our process from the shared memory pairing heap. We might + * already be deleted by the startup process. The 'inHeap' flag prevents + * us from the double deletion. + */ + deleteLSNWaiter(); + + /* + * If we didn't achieve the target LSN, we must be exited by timeout. + */ + if (targetLSN > currentLSN) + { + ereport(ERROR, + (errcode(ERRCODE_QUERY_CANCELED), + errmsg("timed out while waiting for target LSN %X/%X to be replayed; current replay LSN %X/%X", + LSN_FORMAT_ARGS(targetLSN), + LSN_FORMAT_ARGS(currentLSN)))); + } +} + +Datum +pg_wal_replay_wait(PG_FUNCTION_ARGS) +{ + XLogRecPtr target_lsn = PG_GETARG_LSN(0); + int64 timeout = PG_GETARG_INT64(1); + + if (timeout < 0) + ereport(ERROR, + (errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE), + errmsg("\"timeout\" must not be negative"))); + + /* + * We are going to wait for the LSN replay. We should first care that we + * don't hold a snapshot and correspondingly our MyProc->xmin is invalid. + * Otherwise, our snapshot could prevent the replay of WAL records + * implying a kind of self-deadlock. This is the reason why + * pg_wal_replay_wait() is a procedure, not a function. + * + * At first, we should check there is no active snapshot. According to + * PlannedStmtRequiresSnapshot(), even in an atomic context, CallStmt is + * processed with a snapshot. Thankfully, we can pop this snapshot, + * because PortalRunUtility() can tolerate this. + */ + if (ActiveSnapshotSet()) + PopActiveSnapshot(); + + /* + * At second, invalidate a catalog snapshot if any. And we should be done + * with the preparation. + */ + InvalidateCatalogSnapshot(); + + /* Give up if there is still an active or registered sanpshot. */ + if (GetOldestSnapshot()) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("pg_wal_replay_wait() must be only called without an active or registered snapshot"), + errdetail("Make sure pg_wal_replay_wait() isn't called within a transaction with an isolation level higher than READ COMMITTED, another procedure, or a function."))); + + /* + * As the result we should hold no snapshot, and correspondingly our xmin + * should be unset. + */ + Assert(MyProc->xmin == InvalidTransactionId); + + (void) WaitForLSNReplay(target_lsn, timeout); + + PG_RETURN_VOID(); +} diff --git a/src/backend/lib/pairingheap.c b/src/backend/lib/pairingheap.c index fe1deba13ec..7858e5e076b 100644 --- a/src/backend/lib/pairingheap.c +++ b/src/backend/lib/pairingheap.c @@ -44,12 +44,26 @@ pairingheap_allocate(pairingheap_comparator compare, void *arg) pairingheap *heap; heap = (pairingheap *) palloc(sizeof(pairingheap)); + pairingheap_initialize(heap, compare, arg); + + return heap; +} + +/* + * pairingheap_initialize + * + * Same as pairingheap_allocate(), but initializes the pairing heap in-place + * rather than allocating a new chunk of memory. Useful to store the pairing + * heap in a shared memory. + */ +void +pairingheap_initialize(pairingheap *heap, pairingheap_comparator compare, + void *arg) +{ heap->ph_compare = compare; heap->ph_arg = arg; heap->ph_root = NULL; - - return heap; } /* diff --git a/src/backend/storage/ipc/ipci.c b/src/backend/storage/ipc/ipci.c index ca930af08f2..e3e7651d7b7 100644 --- a/src/backend/storage/ipc/ipci.c +++ b/src/backend/storage/ipc/ipci.c @@ -25,6 +25,7 @@ #include "access/xlogprefetcher.h" #include "access/xlogrecovery.h" #include "commands/async.h" +#include "commands/waitlsn.h" #include "miscadmin.h" #include "pgstat.h" #include "postmaster/autovacuum.h" @@ -152,6 +153,7 @@ CalculateShmemSize(int *num_semaphores) size = add_size(size, WaitEventCustomShmemSize()); size = add_size(size, InjectionPointShmemSize()); size = add_size(size, SlotSyncShmemSize()); + size = add_size(size, WaitLSNShmemSize()); #ifdef EXEC_BACKEND size = add_size(size, ShmemBackendArraySize()); #endif @@ -357,6 +359,7 @@ CreateOrAttachShmemStructs(void) StatsShmemInit(); WaitEventCustomShmemInit(); InjectionPointShmemInit(); + WaitLSNShmemInit(); } /* diff --git a/src/backend/storage/lmgr/proc.c b/src/backend/storage/lmgr/proc.c index 1b23efb26f3..ac66da8638f 100644 --- a/src/backend/storage/lmgr/proc.c +++ b/src/backend/storage/lmgr/proc.c @@ -36,6 +36,7 @@ #include "access/transam.h" #include "access/twophase.h" #include "access/xlogutils.h" +#include "commands/waitlsn.h" #include "miscadmin.h" #include "pgstat.h" #include "postmaster/autovacuum.h" @@ -862,6 +863,11 @@ ProcKill(int code, Datum arg) */ LWLockReleaseAll(); + /* + * Cleanup waiting for LSN if any. + */ + WaitLSNCleanup(); + /* Cancel any pending condition variable sleep, too */ ConditionVariableCancelSleep(); diff --git a/src/backend/tcop/pquery.c b/src/backend/tcop/pquery.c index 0c45fcf318f..a1f8d03db1e 100644 --- a/src/backend/tcop/pquery.c +++ b/src/backend/tcop/pquery.c @@ -1168,10 +1168,11 @@ PortalRunUtility(Portal portal, PlannedStmt *pstmt, MemoryContextSwitchTo(portal->portalContext); /* - * Some utility commands (e.g., VACUUM) pop the ActiveSnapshot stack from - * under us, so don't complain if it's now empty. Otherwise, our snapshot - * should be the top one; pop it. Note that this could be a different - * snapshot from the one we made above; see EnsurePortalSnapshotExists. + * Some utility commands (e.g., VACUUM, CALL pg_wal_replay_wait()) pop the + * ActiveSnapshot stack from under us, so don't complain if it's now + * empty. Otherwise, our snapshot should be the top one; pop it. Note + * that this could be a different snapshot from the one we made above; see + * EnsurePortalSnapshotExists. */ if (portal->portalSnapshot != NULL && ActiveSnapshotSet()) { diff --git a/src/backend/utils/activity/wait_event_names.txt b/src/backend/utils/activity/wait_event_names.txt index db37beeaae6..d10ca723dc8 100644 --- a/src/backend/utils/activity/wait_event_names.txt +++ b/src/backend/utils/activity/wait_event_names.txt @@ -87,6 +87,7 @@ LIBPQWALRECEIVER_CONNECT "Waiting in WAL receiver to establish connection to rem LIBPQWALRECEIVER_RECEIVE "Waiting in WAL receiver to receive data from remote server." SSL_OPEN_SERVER "Waiting for SSL while attempting connection." WAIT_FOR_STANDBY_CONFIRMATION "Waiting for WAL to be received and flushed by the physical standby." +WAIT_FOR_WAL_REPLAY "Waiting for a replay of the particular WAL position on the physical standby." WAL_SENDER_WAIT_FOR_WAL "Waiting for WAL to be flushed in WAL sender process." WAL_SENDER_WRITE_DATA "Waiting for any activity when processing replies from WAL receiver in WAL sender process." @@ -345,6 +346,7 @@ WALSummarizer "Waiting to read or update WAL summarization state." DSMRegistry "Waiting to read or update the dynamic shared memory registry." InjectionPoint "Waiting to read or update information related to injection points." SerialControl "Waiting to read or update shared <filename>pg_serial</filename> state." +WaitLSN "Waiting to read or update shared Wait-for-LSN state." # # END OF PREDEFINED LWLOCKS (DO NOT CHANGE THIS LINE) diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat index 06b2f4ba66c..2dbf0fd6aaa 100644 --- a/src/include/catalog/pg_proc.dat +++ b/src/include/catalog/pg_proc.dat @@ -6581,6 +6581,12 @@ prorettype => 'text', proargtypes => '', prosrc => 'pg_get_wal_replay_pause_state' }, +{ oid => '16387', + descr => 'wait for the target LSN to be replayed on standby with an optional timeout', + proname => 'pg_wal_replay_wait', prokind => 'p', prorettype => 'void', + proargtypes => 'pg_lsn int8', proargnames => '{target_lsn,timeout}', + prosrc => 'pg_wal_replay_wait' }, + { oid => '6224', descr => 'get resource managers loaded in system', proname => 'pg_get_wal_resource_managers', prorows => '50', proretset => 't', provolatile => 'v', prorettype => 'record', proargtypes => '', diff --git a/src/include/commands/waitlsn.h b/src/include/commands/waitlsn.h new file mode 100644 index 00000000000..f719feadb05 --- /dev/null +++ b/src/include/commands/waitlsn.h @@ -0,0 +1,80 @@ +/*------------------------------------------------------------------------- + * + * waitlsn.h + * Declarations for LSN replay waiting routines. + * + * Copyright (c) 2024, PostgreSQL Global Development Group + * + * src/include/commands/waitlsn.h + * + *------------------------------------------------------------------------- + */ +#ifndef WAIT_LSN_H +#define WAIT_LSN_H + +#include "lib/pairingheap.h" +#include "postgres.h" +#include "port/atomics.h" +#include "storage/latch.h" +#include "storage/spin.h" +#include "tcop/dest.h" + +/* + * WaitLSNProcInfo - the shared memory structure representing information + * about the single process, which may wait for LSN replay. An item of + * waitLSN->procInfos array. + */ +typedef struct WaitLSNProcInfo +{ + /* LSN, which this process is waiting for */ + XLogRecPtr waitLSN; + + /* + * A pointer to the latch, which should be set once the waitLSN is + * replayed. + */ + Latch *latch; + + /* A pairing heap node for participation in waitLSNState->waitersHeap */ + pairingheap_node phNode; + + /* + * A flag indicating that this item is present in + * waitLSNState->waitersHeap + */ + bool inHeap; +} WaitLSNProcInfo; + +/* + * WaitLSNState - the shared memory state for the replay LSN waiting facility. + */ +typedef struct WaitLSNState +{ + /* + * The minimum LSN value some process is waiting for. Used for the + * fast-path checking if we need to wake up any waiters after replaying a + * WAL record. Could be read lock-less. Update protected by WaitLSNLock. + */ + pg_atomic_uint64 minWaitedLSN; + + /* + * A pairing heap of waiting processes order by LSN values (least LSN is + * on top). Protected by WaitLSNLock. + */ + pairingheap waitersHeap; + + /* + * An array with per-process information, indexed by the process number. + * Protected by WaitLSNLock. + */ + WaitLSNProcInfo procInfos[FLEXIBLE_ARRAY_MEMBER]; +} WaitLSNState; + +extern PGDLLIMPORT WaitLSNState *waitLSNState; + +extern Size WaitLSNShmemSize(void); +extern void WaitLSNShmemInit(void); +extern void WaitLSNSetLatches(XLogRecPtr currentLSN); +extern void WaitLSNCleanup(void); + +#endif /* WAIT_LSN_H */ diff --git a/src/include/lib/pairingheap.h b/src/include/lib/pairingheap.h index 7eade81535a..9e1c26033a1 100644 --- a/src/include/lib/pairingheap.h +++ b/src/include/lib/pairingheap.h @@ -77,6 +77,9 @@ typedef struct pairingheap extern pairingheap *pairingheap_allocate(pairingheap_comparator compare, void *arg); +extern void pairingheap_initialize(pairingheap *heap, + pairingheap_comparator compare, + void *arg); extern void pairingheap_free(pairingheap *heap); extern void pairingheap_add(pairingheap *heap, pairingheap_node *node); extern pairingheap_node *pairingheap_first(pairingheap *heap); diff --git a/src/include/storage/lwlocklist.h b/src/include/storage/lwlocklist.h index 6a2f64c54fb..88dc79b2bd6 100644 --- a/src/include/storage/lwlocklist.h +++ b/src/include/storage/lwlocklist.h @@ -83,3 +83,4 @@ PG_LWLOCK(49, WALSummarizer) PG_LWLOCK(50, DSMRegistry) PG_LWLOCK(51, InjectionPoint) PG_LWLOCK(52, SerialControl) +PG_LWLOCK(53, WaitLSN) diff --git a/src/test/recovery/meson.build b/src/test/recovery/meson.build index b1eb77b1ec1..712924c2fad 100644 --- a/src/test/recovery/meson.build +++ b/src/test/recovery/meson.build @@ -51,6 +51,7 @@ tests += { 't/040_standby_failover_slots_sync.pl', 't/041_checkpoint_at_promote.pl', 't/042_low_level_backup.pl', + 't/043_wal_replay_wait.pl', ], }, } diff --git a/src/test/recovery/t/043_wal_replay_wait.pl b/src/test/recovery/t/043_wal_replay_wait.pl new file mode 100644 index 00000000000..e4842730b05 --- /dev/null +++ b/src/test/recovery/t/043_wal_replay_wait.pl @@ -0,0 +1,150 @@ +# Checks waiting for the lsn replay on standby using +# pg_wal_replay_wait() procedure. +use strict; +use warnings; + +use PostgreSQL::Test::Cluster; +use PostgreSQL::Test::Utils; +use Test::More; + +# Initialize primary node +my $node_primary = PostgreSQL::Test::Cluster->new('primary'); +$node_primary->init(allows_streaming => 1); +$node_primary->start; + +# And some content and take a backup +$node_primary->safe_psql('postgres', + "CREATE TABLE wait_test AS SELECT generate_series(1,10) AS a"); +my $backup_name = 'my_backup'; +$node_primary->backup($backup_name); + +# Create a streaming standby with a 1 second delay from the backup +my $node_standby1 = PostgreSQL::Test::Cluster->new('standby'); +my $delay = 1; +$node_standby1->init_from_backup($node_primary, $backup_name, + has_streaming => 1); +$node_standby1->append_conf( + 'postgresql.conf', qq[ + recovery_min_apply_delay = '${delay}s' +]); +$node_standby1->start; + +# 1. Make sure that pg_wal_replay_wait() works: add new content to +# primary and memorize primary's insert LSN, then wait for that LSN to be +# replayed on standby. +$node_primary->safe_psql('postgres', + "INSERT INTO wait_test VALUES (generate_series(11, 20))"); +my $lsn1 = + $node_primary->safe_psql('postgres', "SELECT pg_current_wal_insert_lsn()"); +my $output = $node_standby1->safe_psql( + 'postgres', qq[ + CALL pg_wal_replay_wait('${lsn1}', 1000000); + SELECT pg_lsn_cmp(pg_last_wal_replay_lsn(), '${lsn1}'::pg_lsn); +]); + +# Make sure the current LSN on standby is at least as big as the LSN we +# observed on primary's before. +ok($output >= 0, + "standby reached the same LSN as primary after pg_wal_replay_wait()"); + +# 2. Check that new data is visible after calling pg_wal_replay_wait() +$node_primary->safe_psql('postgres', + "INSERT INTO wait_test VALUES (generate_series(21, 30))"); +my $lsn2 = + $node_primary->safe_psql('postgres', "SELECT pg_current_wal_insert_lsn()"); +$output = $node_standby1->safe_psql( + 'postgres', qq[ + CALL pg_wal_replay_wait('${lsn2}'); + SELECT count(*) FROM wait_test; +]); + +# Make sure the count(*) on standby reflects the recent changes on primary +ok($output eq 30, "standby reached the same LSN as primary"); + +# 3. Check that waiting for unreachable LSN triggers the timeout. The +# unreachable LSN must be well in advance. So WAL records issued by +# the concurrent autovacuum could not affect that. +my $lsn3 = + $node_primary->safe_psql('postgres', + "SELECT pg_current_wal_insert_lsn() + 10000000000"); +my $stderr; +$node_standby1->safe_psql('postgres', + "CALL pg_wal_replay_wait('${lsn2}', 10);"); +$node_standby1->psql( + 'postgres', + "CALL pg_wal_replay_wait('${lsn3}', 1000);", + stderr => \$stderr); +ok( $stderr =~ /timed out while waiting for target LSN/, + "get timeout on waiting for unreachable LSN"); + + +# 4. Also, check the scenario of multiple LSN waiters. We make 5 background +# psql sessions each waiting for a corresponding insertion. When waiting is +# finished, stored procedures logs if there are visible as many rows as +# should be. +$node_primary->safe_psql( + 'postgres', qq[ +CREATE FUNCTION log_count(i int) RETURNS void AS \$\$ + DECLARE + count int; + BEGIN + SELECT count(*) FROM wait_test INTO count; + IF count >= 31 + i THEN + RAISE LOG 'count %', i; + END IF; + END +\$\$ +LANGUAGE plpgsql; +]); +$node_standby1->safe_psql('postgres', "SELECT pg_wal_replay_pause();"); +my @psql_sessions; +for (my $i = 0; $i < 5; $i++) +{ + print($i); + $node_primary->safe_psql('postgres', + "INSERT INTO wait_test VALUES (${i});"); + my $lsn = + $node_primary->safe_psql('postgres', + "SELECT pg_current_wal_insert_lsn()"); + $psql_sessions[$i] = $node_standby1->background_psql('postgres'); + $psql_sessions[$i]->query_until( + qr/start/, qq[ + \\echo start + CALL pg_wal_replay_wait('${lsn}'); + SELECT log_count(${i}); + ]); +} +my $log_offset = -s $node_standby1->logfile; +$node_standby1->safe_psql('postgres', "SELECT pg_wal_replay_resume();"); +for (my $i = 0; $i < 5; $i++) +{ + $node_standby1->wait_for_log("count ${i}", $log_offset); + $psql_sessions[$i]->quit; +} + +ok(1, 'multiple LSN waiters reported consistent data'); + +# 5. Check that the standby promotion terminates the wait on LSN. Start +# waiting for an unreachable LSN then promote. Check the log for the relevant +# error message. +my $psql_session = $node_standby1->background_psql('postgres'); +$psql_session->query_until( + qr/start/, qq[ + \\echo start + CALL pg_wal_replay_wait('${lsn3}'); +]); + +$log_offset = -s $node_standby1->logfile; +$node_standby1->promote; +$node_standby1->wait_for_log('recovery is not in progress', $log_offset); + +ok(1, 'got error after standby promote'); + +$node_standby1->stop; +$node_primary->stop; + +# If we send \q with $psql_session->quit the command can be sent to the session +# already closed. So \q is in initial script, here we only finish IPC::Run. +$psql_session->{run}->finish; + +done_testing(); diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index 3deb6113b80..aaea2468536 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -3110,6 +3110,8 @@ WaitEventIO WaitEventIPC WaitEventSet WaitEventTimeout +WaitLSNProcInfo +WaitLSNState WaitPMResult WalCloseMethod WalCompression -- 2.39.3 (Apple Git-145)
