From: Kartyshov Ivan <i.kartyshov@postgrespro.ru>
To: Bharath Rupireddy <bharath.rupireddyforpostgres@gmail.com>
Cc: Peter Eisentraut <peter@eisentraut.org>
Cc: Alexander Korotkov <aekorotkov@gmail.com>
Cc: Amit Kapila <amit.kapila16@gmail.com>
Cc: dilipbalaut@gmail.com
Cc: smithpb2250@gmail.com
Cc: pgsql-hackers@lists.postgresql.org
Subject: Re: [HACKERS] make async slave to wait for lsn to be replayed
Date: Tue, 26 Mar 2024 17:06:51 +0300
Message-ID: <5358d911a08ef2952b1e9f74062426ee@postgrespro.ru> (raw)
In-Reply-To: <CALj2ACV5+A2fRkMdB1YLH8tHMy8xsUr1HMu5RffJpmnfH-_42A@mail.gmail.com>
References: <CAPpHfdtiGgn0iS1KbW2HTam-1+oK+vhXZDAcnX9hKaA7Oe=F-A@mail.gmail.com>
<22f122-655b3e80-3-688b3700@97587957>
<CAPpHfdt=Edtk9B3owg4of6v5Y=vi6=5R9ZHM84KPOdR6mER0wg@mail.gmail.com>
<b515af28c21b11305d71062f1590a2aa@postgrespro.ru>
<4953563546cb8c8851f84c7debf723ef@postgrespro.ru>
<31547e4e8c263392a768ea89a3f57833@postgrespro.ru>
<CAA4eK1KBqcH74uJ-PP7J=ghEdvigazWoWgct-af+LLwnbT8SXQ@mail.gmail.com>
<CAPpHfdsZ8hJiYaRxTbnKqVee_Nmbb+PVd+QtWifoBYkX71f0dA@mail.gmail.com>
<CAPpHfduV3=v=0TmR-95To=7ZRQMimM5uqB1rtW520fcz-J5Dew@mail.gmail.com>
<CAA4eK1+FsVRV5bH9838XzSRdiB9eKUFcPpGCknhZ9tHePWbuNw@mail.gmail.com>
<CALj2ACUfS7LH1PaWmSZ5KwH4BpQxO9izeMw4qC3a1DAwi6nfbQ@mail.gmail.com>
<CAPpHfdu+HAF=XPX2v+Qq+faV2bVhj=qLz1TbtAZX5shg8Zviow@mail.gmail.com>
<a7601ec6-01d8-4561-b81d-a6605e9b4d55@eisentraut.org>
<CALj2ACV5+A2fRkMdB1YLH8tHMy8xsUr1HMu5RffJpmnfH-_42A@mail.gmail.com>
Thank you for your interest to the patch.
I understand you questions, but I fully support Alexander Korotkov idea
to commit the minimal required functionality. And then keep working on
other improvements.
On 2024-03-24 05:39, Bharath Rupireddy wrote:
> On Fri, Mar 22, 2024 at 4:28 AM Peter Eisentraut <peter@eisentraut.org>
> wrote:
>>
>> I had written in [0] about my questions related to using this with
>> connection poolers. I don't think this was addressed at all. I
>> haven't
>> seen any discussion about how to make this kind of facility usable in
>> a
>> full system. You have to manually query and send LSNs; that seems
>> pretty cumbersome. Sure, this is part of something that could be
>> useful, but how would an actual user with actual application code get
>> to
>> use this?
>>
>> [0]:
>> https://www.postgresql.org/message-id/8b5b172f-0ae7-d644-8358-e2851dded43b%40enterprisedb.com>>> But I wonder how a client is going to get the LSN. How would all of
>>> this be used by a client? I can think of a scenarios where you have
>>> an application that issues a bunch of SQL commands and you have some
>>> kind of pooler in the middle that redirects those commands to
>>> different hosts, and what you really want is to have it transparently
>>> behave as if it's just a single host. Do we want to inject a bunch
>>> of "SELECT pg_get_lsn()", "SELECT pg_wait_lsn()" calls into that?
As I understand your question, application make dml on the primary
server, get LSN of changes and send bunch SQL read-only commands to
pooler. Transparent behave we can get using #synchronous_commit, but
it is very slow.
>>> I'm tempted to think this could be a protocol-layer facility. Every
>>> query automatically returns the current LSN, and every query can also
>>> send along an LSN to wait for, and the client library would just keep
>>> track of the LSN for (what it thinks of as) the connection. So you
>>> get some automatic serialization without having to modify your client
>>> code.
Thank you, it is a good question for future versions.
You say about a protocol-layer facility, what you meen. May be we can
use signals, like hot_standby_feedback.
> I share the same concern as yours and had proposed something upthread
> [1]. The idea is something like how each query takes a snapshot at the
> beginning of txn/query (depending on isolation level), the same way
> the standby can wait for the primary's current LSN as of the moment
> (at the time of taking snapshot). And, primary keeps sending its
> current LSN as part of regular WAL to standbys so that the standbys
> doesn't have to make connections to the primary to know its current
> LSN every time. Perhps, this may not even fully guarantee (considered
> to be achieving) the read-after-write consistency on standbys unless
> there's a way for the application to tell the wait LSN.
>
> Thoughts?
>
> [1]
> https://www.postgresql.org/message-id/CALj2ACUfS7LH1PaWmSZ5KwH4BpQxO9izeMw4qC3a1DAwi6nfbQ%40mail.gma...> +1 to have support for implicit txns. A strawman solution I can think
> of is to let primary send its current insert LSN to the standby every
> time it sends a bunch of WAL, and the standby waits for that LSN to be
> replayed on it at the start of every implicit txn automatically.
And how standby will get lsn to wait for? All solutions I can think of
are very invasive and poorly scalable.
For example, every dml can send back LSN if dml is success. And
application could use it to wait actual changes.
> The new BEGIN syntax requires application code changes. This led me to
> think how one can achieve read-after-write consistency today in a
> primary - standby set up. All the logic of this patch, that is, waiting
> for the standby to pass a given primary LSN needs to be done in the
> application code (or in proxy or in load balancer?). I believe there
> might be someone doing this already, it's good to hear from them.
You may use #synchronous_commit mode but it slow. So my implementation
don`t make primary to wait all standby to sent its feedbacks.
--
Ivan Kartyshov
Postgres Professional: www.postgrespro.com
Attachments:
[text/x-diff] v14_0002-Procedure-wait-lsn.patch (18.4K, ../5358d911a08ef2952b1e9f74062426ee@postgrespro.ru/2-v14_0002-Procedure-wait-lsn.patch)
download | inline diff:diff --git a/doc/src/sgml/func.sgml b/doc/src/sgml/func.sgmlindex 8ecc02f2b9..8042201bab 100644--- a/doc/src/sgml/func.sgml+++ b/doc/src/sgml/func.sgml@@ -28086,10 +28086,55 @@ postgres=# SELECT '0/0'::pg_lsn + pd.segment_number * ps.setting::int + :offset
extension.
</para></entry>
</row>
++ <row>+ <entry role="func_table_entry"><para role="func_signature">+ <indexterm>+ <primary>pg_wait_lsn</primary>+ </indexterm>+ <function>pg_wait_lsn</function> (trg_lsn pg_lsn, delay int8 DEFAULT 0)+ </para>+ <para>+ Returns ERROR if target LSN was not replayed on standby.+ Parameter <parameter>delay</parameter> sets seconds, the time to wait to the LSN.+ </para></entry>+ </row>
</tbody>
</tgroup>
</table>
+ <para>+ <function>pg_wait_lsn</function> waits till <parameter>wait_lsn</parameter>+ to be replayed on standby to achieve read-your-writes-consistency, while+ using async replica for reads and primary for writes. In that case lsn of+ last modification is stored inside application.+ Note: pg_wait_lsn(trg_lsn pg_lsn, delay int8 DEFAULT 0)+ </para>++ <para>+ You can use <function>pg_wait_lsn</function> to wait the <type>pg_lsn</type>+ value. For example:+ <programlisting>+ Replica:+ postgresql.conf+ recovery_min_apply_delay = 10s;++ Primary: Application update table and get lsn of changes+ postgres=# UPDATE films SET kind = 'Dramatic' WHERE kind = 'Drama';+ postgres=# SELECT pg_current_wal_lsn();+ pg_current_wal_lsn+ --------------------+ 0/306EE20+ (1 row)++ Replica: Wait and read updated data+ postgres=# CALL pg_wait_lsn('0/306EE20', 100); // Timeout 100ms is insufficent+ ERROR: canceling waiting for LSN due to timeout+ postgres=# CALL pg_wait_lsn('0/306EE20');+ postgres=# SELECT * FROM films WHERE kind = 'Drama';+ </programlisting>+ </para>+
<para>
The functions shown in <xref
linkend="functions-recovery-control-table"/> control the progress of recovery.
diff --git a/src/backend/access/transam/xlogrecovery.c b/src/backend/access/transam/xlogrecovery.cindex 29c5bec084..0b783dc733 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,14 @@ PerformWalRecovery(void)
break;
}
+ /*+ * If we replayed an LSN that someone was waiting for, set latches+ * in shared memory array to notify the waiter.+ */+ if (waitLSN &&+ (XLogRecoveryCtl->lastReplayedEndRecPtr >= pg_atomic_read_u64(&waitLSN->minLSN)))+ 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.sqlindex fe2bb50f46..5c4f9c78a1 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_wait_lsn(trg_lsn pg_lsn, delay float8 DEFAULT 0)+ LANGUAGE internal AS 'pg_wait_lsn';+
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/Makefileindex 48f7348f91..cede90c3b9 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.buildindex 6dd00a4abd..7549be5dc3 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.cnew file mode 100644index 0000000000..bd9ea6b7db--- /dev/null+++ b/src/backend/commands/waitlsn.c@@ -0,0 +1,303 @@+/*-------------------------------------------------------------------------+ *+ * waitlsn.c+ * Implements waiting for the given LSN, which is used in+ * CALL pg_wait_lsn(wait_lsn pg_lsn, timeout int).+ *+ * 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 "fmgr.h"+#include "access/transam.h"+#include "access/xact.h"+#include "access/xlog.h"+#include "access/xlogdefs.h"+#include "access/xlogrecovery.h"+#include "catalog/pg_type.h"+#include "commands/waitlsn.h"+#include "executor/spi.h"+#include "funcapi.h"+#include "miscadmin.h"+#include "storage/ipc.h"+#include "storage/latch.h"+#include "storage/pmsignal.h"+#include "storage/proc.h"+#include "storage/shmem.h"+#include "storage/sinvaladt.h"+#include "utils/builtins.h"+#include "utils/pg_lsn.h"+#include "utils/snapmgr.h"+#include "utils/timestamp.h"+#include "utils/fmgrprotos.h"++/* Add to / delete from shared memory array */+static void addLSNWaiter(XLogRecPtr lsn);+static void deleteLSNWaiter(void);++struct WaitLSNState *waitLSN = NULL;+static volatile sig_atomic_t haveShmemItem = false;++/*+ * 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;++ waitLSN = (WaitLSNState *) ShmemInitStruct("WaitLSNState",+ WaitLSNShmemSize(),+ &found);+ if (!found)+ {+ SpinLockInit(&waitLSN->mutex);+ waitLSN->numWaitedProcs = 0;+ pg_atomic_init_u64(&waitLSN->minLSN, PG_UINT64_MAX);+ }+}++/*+ * Add the information about the LSN waiter backend to the shared memory+ * array.+ */+static void+addLSNWaiter(XLogRecPtr lsn)+{+ WaitLSNProcInfo cur;+ int i;++ SpinLockAcquire(&waitLSN->mutex);++ cur.procnum = MyProcNumber;+ cur.waitLSN = lsn;++ for (i = 0; i < waitLSN->numWaitedProcs; i++)+ {+ if (waitLSN->procInfos[i].waitLSN >= cur.waitLSN)+ {+ WaitLSNProcInfo tmp;++ tmp = waitLSN->procInfos[i];+ waitLSN->procInfos[i] = cur;+ cur = tmp;+ }+ }+ waitLSN->procInfos[i] = cur;+ waitLSN->numWaitedProcs++;++ pg_atomic_write_u64(&waitLSN->minLSN, waitLSN->procInfos[i].waitLSN);+ SpinLockRelease(&waitLSN->mutex);+}++/*+ * Delete the information about the LSN waiter backend from the shared memory+ * array.+ */+static void+deleteLSNWaiter(void)+{+ int i;+ bool found = false;++ SpinLockAcquire(&waitLSN->mutex);++ for (i = 0; i < waitLSN->numWaitedProcs; i++)+ {+ if (waitLSN->procInfos[i].procnum == MyProcNumber)+ found = true;++ if (found && i < waitLSN->numWaitedProcs - 1)+ {+ waitLSN->procInfos[i] = waitLSN->procInfos[i + 1];+ }+ }++ if (!found)+ {+ SpinLockRelease(&waitLSN->mutex);+ return;+ }+ waitLSN->numWaitedProcs--;++ if (waitLSN->numWaitedProcs != 0)+ pg_atomic_write_u64(&waitLSN->minLSN, waitLSN->procInfos[i].waitLSN);+ else+ pg_atomic_write_u64(&waitLSN->minLSN, PG_UINT64_MAX);++ SpinLockRelease(&waitLSN->mutex);+}++/*+ * Set all latches in shared memory to signal that new LSN has been replayed+*/+void+WaitLSNSetLatches(XLogRecPtr curLSN)+{+ uint32 i,+ numWakeUpProcs;++ SpinLockAcquire(&waitLSN->mutex);++ /*+ * Set latches for process, whose waited LSNs are already replayed.+ */+ for (i = 0; i < waitLSN->numWaitedProcs; i++)+ {+ PGPROC *backend;++ if (waitLSN->procInfos[i].waitLSN > curLSN)+ break;++ backend = GetPGProcByNumber(waitLSN->procInfos[i].procnum);+ SetLatch(&backend->procLatch);+ }++ /*+ * Immediately remove those processes from the shmem array. Otherwise,+ * shmem array items will be here till corresponding processes wake up and+ * delete themselves.+ */+ numWakeUpProcs = i;+ for (i = 0; i < waitLSN->numWaitedProcs - numWakeUpProcs; i++)+ waitLSN->procInfos[i] = waitLSN->procInfos[i + numWakeUpProcs];+ waitLSN->numWaitedProcs -= numWakeUpProcs;++ if (waitLSN->numWaitedProcs != 0)+ pg_atomic_write_u64(&waitLSN->minLSN, waitLSN->procInfos[i].waitLSN);+ else+ pg_atomic_write_u64(&waitLSN->minLSN, PG_UINT64_MAX);++ SpinLockRelease(&waitLSN->mutex);+}++/*+ * Delete our item from shmem array if any.+ */+void+WaitLSNCleanup(void)+{+ if (haveShmemItem)+ deleteLSNWaiter();+}++/*+ * Wait using MyLatch to wait till the given LSN is replayed, the postmaster dies or+ * timeout happens.+ */+void+WaitForLSN(XLogRecPtr lsn, float8 sec)+{+ XLogRecPtr curLSN;+ int latch_events;+ TimestampTz endtime;++ /* Shouldn't be called when shmem isn't initialized */+ Assert(waitLSN);++ /* Should be only called by a backend */+ Assert(MyBackendType == B_BACKEND);++ 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.")));++ endtime = GetCurrentTimestamp() + (int64)(sec * 1000 * 1000);++ latch_events = WL_TIMEOUT | WL_LATCH_SET | WL_EXIT_ON_PM_DEATH;+ addLSNWaiter(lsn);+ haveShmemItem = true;++ for (;;)+ {+ int rc;+ long delay_ms;++ /* Check if the waited LSN has been replayed */+ curLSN = GetXLogReplayRecPtr(NULL);+ if (lsn <= curLSN)+ break;++ if (sec >= 1)+ delay_ms = (endtime - GetCurrentTimestamp()) / 1000;+ else+ /* If no timeout is set then wake up in 1 minute for interrupts */+ delay_ms = 60000;++ if (delay_ms <= 0)+ break;++ /*+ * If received an interruption from CHECK_FOR_INTERRUPTS, then delete+ * the current event from array.+ */+ CHECK_FOR_INTERRUPTS();++ /* If postmaster dies, finish immediately */+ if (!PostmasterIsAlive())+ break;++ rc = WaitLatch(MyLatch, latch_events, delay_ms,+ WAIT_EVENT_WAIT_FOR_STANDBY_CONFIRMATION);++ if (rc & WL_LATCH_SET)+ ResetLatch(MyLatch);+ }++ if (lsn > curLSN)+ {+ deleteLSNWaiter();+ haveShmemItem = false;+ ereport(ERROR,+ (errcode(ERRCODE_QUERY_CANCELED),+ errmsg("canceling waiting for LSN due to timeout")));+ }+ else+ {+ haveShmemItem = false;+ }+}++Datum+pg_wait_lsn(PG_FUNCTION_ARGS)+{+ XLogRecPtr trg_lsn = PG_GETARG_LSN(0);+ float8 delay = PG_GETARG_FLOAT8(1);+ CallContext *context = (CallContext *) fcinfo->context;++ if (context->atomic)+ ereport(ERROR,+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),+ errmsg("pg_wait_lsn() must be only called in non-atomic context")));++ if (ActiveSnapshotSet())+ PopActiveSnapshot();+ Assert(!ActiveSnapshotSet());++ (void) WaitForLSN(trg_lsn, delay);++ PG_RETURN_VOID();+}diff --git a/src/backend/storage/ipc/ipci.c b/src/backend/storage/ipc/ipci.cindex 521ed5418c..5aed90c935 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, WaitEventExtensionShmemSize());
size = add_size(size, InjectionPointShmemSize());
size = add_size(size, SlotSyncShmemSize());
+ size = add_size(size, WaitLSNShmemSize());
#ifdef EXEC_BACKEND
size = add_size(size, ShmemBackendArraySize());
#endif
@@ -244,6 +246,11 @@ CreateSharedMemoryAndSemaphores(void)
/* Initialize subsystems */
CreateOrAttachShmemStructs();
+ /*+ * Init array of Latches in shared memory for wait lsn+ */+ WaitLSNShmemInit();+
#ifdef EXEC_BACKEND
/*
diff --git a/src/backend/storage/lmgr/proc.c b/src/backend/storage/lmgr/proc.cindex 162b1f919d..4b830dc3c8 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/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.datindex 0d26e5b422..cb325b3be0 100644--- a/src/include/catalog/pg_proc.dat+++ b/src/include/catalog/pg_proc.dat@@ -12138,6 +12138,11 @@
prorettype => 'bytea', proargtypes => 'pg_brin_minmax_multi_summary',
prosrc => 'brin_minmax_multi_summary_send' },
+{ oid => '16387', descr => 'wait for LSN until timeout',+ proname => 'pg_wait_lsn', prokind => 'p', prorettype => 'void',+ proargtypes => 'pg_lsn float8', proargnames => '{trg_lsn,delay}',+ prosrc => 'pg_wait_lsn' },+
{ oid => '6291', descr => 'arbitrary value from among input values',
proname => 'any_value', prokind => 'a', proisstrict => 'f',
prorettype => 'anyelement', proargtypes => 'anyelement',
diff --git a/src/include/commands/waitlsn.h b/src/include/commands/waitlsn.hnew file mode 100644index 0000000000..53ab9b7b6b--- /dev/null+++ b/src/include/commands/waitlsn.h@@ -0,0 +1,43 @@+/*-------------------------------------------------------------------------+ *+ * waitlsn.h+ * Declarations for LSN waiting routines.+ *+ * Copyright (c) 2024, PostgreSQL Global Development Group+ *+ * src/include/commands/waitlsn.h+ *+ *-------------------------------------------------------------------------+ */+#ifndef WAIT_LSN_H+#define WAIT_LSN_H++#include "postgres.h"+#include "port/atomics.h"+#include "storage/spin.h"+#include "tcop/dest.h"++/* Shared memory structure */+typedef struct WaitLSNProcInfo+{+ int procnum;+ XLogRecPtr waitLSN;+} WaitLSNProcInfo;++typedef struct WaitLSNState+{+ pg_atomic_uint64 minLSN;+ slock_t mutex;+ int numWaitedProcs;+ WaitLSNProcInfo procInfos[FLEXIBLE_ARRAY_MEMBER];+} WaitLSNState;++extern PGDLLIMPORT struct WaitLSNState *waitLSN;++extern void WaitForLSN(XLogRecPtr lsn, float8 sec);+extern Size WaitLSNShmemSize(void);+extern void WaitLSNShmemInit(void);+extern void WaitLSNSetLatches(XLogRecPtr curLSN);+extern void WaitLSNCleanup(void);++#endif /* WAIT_LSN_H */diff --git a/src/test/recovery/meson.build b/src/test/recovery/meson.buildindex b1eb77b1ec..bc47c93902 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_wait_lsn.pl',
],
},
}
diff --git a/src/test/recovery/t/043_wait_lsn.pl b/src/test/recovery/t/043_wait_lsn.plnew file mode 100644index 0000000000..1875702c93--- /dev/null+++ b/src/test/recovery/t/043_wait_lsn.pl@@ -0,0 +1,77 @@+# Checks waiting for lsn on standby pg_wait_lsn()+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_standby = PostgreSQL::Test::Cluster->new('standby');+my $delay = 3;+$node_standby->init_from_backup($node_primary, $backup_name,+ has_streaming => 1);+$node_standby->append_conf(+ 'postgresql.conf', qq[+ recovery_min_apply_delay = '${delay}s'+]);+$node_standby->start;+++# Make sure that pg_wait_lsn() works: add new content to primary and memorize+# primary's new LSN, then wait for primary's LSN on standby. Prove that pg_wait_lsn() is+# able to setup an infinite waiting loop and exit it if given no wait timeout.+$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_lsn()");+my $output = $node_standby->safe_psql(+ 'postgres', qq[+ CALL pg_wait_lsn('${lsn1}', 1000000);+ SELECT pg_lsn_cmp(pg_last_wal_replay_lsn(), '${lsn1}'::pg_lsn);+]);++# Get the current LSN on standby and make sure it's the same as primary's LSN+ok($output eq 0, "standby reached the same LSN as primary pg_wait_lsn()");++my $lsn2 =+ $node_primary->safe_psql('postgres', "SELECT pg_current_wal_lsn() + 1");+my $stderr;+$node_standby->safe_psql('postgres', "CALL pg_wait_lsn('${lsn1}', 1);");+$node_standby->psql(+ 'postgres',+ "CALL pg_wait_lsn('${lsn2}', 1);",+ stderr => \$stderr);+ok( $stderr =~ /canceling waiting for LSN due to timeout/,+ "get timeout on waiting for unreachable LSN");++# Make sure that pg_wait_lsn() works: add new content to primary and memorize+# primary's new LSN, then wait for primary's LSN on standby. Prove that pg_wait_lsn() is+# able to setup an infinite waiting loop and exit it if current LSN replayed.+$node_primary->safe_psql('postgres',+ "INSERT INTO wait_test VALUES (generate_series(21, 30))");+my $lsn3 =+ $node_primary->safe_psql('postgres', "SELECT pg_current_wal_lsn()");+$output = $node_standby->safe_psql(+ 'postgres', qq[+ CALL pg_wait_lsn('${lsn3}');+ SELECT pg_lsn_cmp(pg_last_wal_replay_lsn(), '${lsn3}'::pg_lsn);+]);++# Get the current LSN on standby and make sure it's the same as primary's LSN+ok($output eq 0, "standby reached the same LSN as primary");++$node_standby->stop;+$node_primary->stop;+done_testing();
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: pgsql-hackers@postgresql.org
Cc: i.kartyshov@postgrespro.ru, bharath.rupireddyforpostgres@gmail.com, peter@eisentraut.org, aekorotkov@gmail.com, amit.kapila16@gmail.com, dilipbalaut@gmail.com, smithpb2250@gmail.com, pgsql-hackers@lists.postgresql.org
Subject: Re: [HACKERS] make async slave to wait for lsn to be replayed
In-Reply-To: <5358d911a08ef2952b1e9f74062426ee@postgrespro.ru>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by DDX for PostgreSQL; see mirroring instructions
for how to clone and mirror all data and code used for this inbox