From: Kartyshov Ivan <i.kartyshov@postgrespro.ru>
To: 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: Fri, 15 Mar 2024 23:32:23 +0300
Message-ID: <59cf2cbdc6f9bf81602e3515acca22b0@postgrespro.ru> (raw)
In-Reply-To: <9e0ad3332c0bf7c4f00ff10677245e46@postgrespro.ru>
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>
<9e0ad3332c0bf7c4f00ff10677245e46@postgrespro.ru>
On 2024-03-15 22:59, Kartyshov Ivan wrote:
> On 2024-03-11 13:44, Alexander Korotkov wrote:
>> I picked the second option and left only the AFTER clause for the
>> BEGIN statement. I think this should be enough for the beginning.
>
> Thank you for your rework on your patch, here I made some fixes:
> 0) autocomplete
> 1) less jumps
> 2) more description and add cases in doc
>
> I think, it will be useful to have stand-alone statement.
> Why you would like to see only AFTER clause for the BEGIN statement?
Rebase and update patch.
--
Ivan Kartyshov
Postgres Professional: www.postgrespro.com
Attachments:
[text/x-diff] v11-0001-Implement-AFTER-clause-for-BEGIN-command.patch (23.4K, ../59cf2cbdc6f9bf81602e3515acca22b0@postgrespro.ru/2-v11-0001-Implement-AFTER-clause-for-BEGIN-command.patch)
download | inline diff:diff --git a/doc/src/sgml/ref/begin.sgml b/doc/src/sgml/ref/begin.sgmlindex 016b021487..759e46ec24 100644--- a/doc/src/sgml/ref/begin.sgml+++ b/doc/src/sgml/ref/begin.sgml@@ -21,13 +21,16 @@ PostgreSQL documentation
<refsynopsisdiv>
<synopsis>
-BEGIN [ WORK | TRANSACTION ] [ <replaceable class="parameter">transaction_mode</replaceable> [, ...] ]+BEGIN [ WORK | TRANSACTION ] [ <replaceable class="parameter">transaction_mode</replaceable> [, ...] ] <replaceable class="parameter">wait_event</replaceable>
<phrase>where <replaceable class="parameter">transaction_mode</replaceable> is one of:</phrase>
ISOLATION LEVEL { SERIALIZABLE | REPEATABLE READ | READ COMMITTED | READ UNCOMMITTED }
READ WRITE | READ ONLY
[ NOT ] DEFERRABLE
++<phrase>where <replaceable class="parameter">wait_event</replaceable> is:</phrase>+ AFTER <replaceable class="parameter">lsn_value</replaceable> [ WITHIN <replaceable class="parameter">number_of_milliseconds</replaceable> ]
</synopsis>
</refsynopsisdiv>
@@ -78,6 +81,50 @@ BEGIN [ WORK | TRANSACTION ] [ <replaceable class="parameter">transaction_mode</
</para>
</listitem>
</varlistentry>
++ <varlistentry>+ <term><literal>AFTER</literal> <replaceable class="parameter">lsn_value</replaceable></term>+ <listitem>+ <para>+ <command>AFTER</command> clause is used on standby in+ <link linkend="streaming-replication">physical streaming replication</link>+ and specifies waiting for the specific WAL location (<acronym>LSN</acronym>)+ to be replayed before starting the transaction.+ </para>+ <para>+ This option is useful when the user makes some data changes on primary+ and needs a guarantee to see these changes on standby. The LSN to wait+ could be obtained on the primary using+ <link linkend="functions-admin-backup"><function>pg_current_wal_insert_lsn</function></link>+ function after committing the relevant changes.+ </para>+ </listitem>+ </varlistentry>++ <varlistentry>+ <term><literal>WITHIN</literal> <replaceable class="parameter">number_of_milliseconds</replaceable></term>+ <listitem>+ <para>+ Provides the timeout for the <command>AFTER</command> clause.+ Especially helpful to prevent freezing on streaming replication+ connection failures.+ </para>+ <para>+ On practice we have three variants of waiting current LSN:+ waiting forever - (finish on PM DEATH or SIGINT)+ BEGIN AFTER lsn;+ wait for changes to be replayed on standby and then start transaction+ waiting on timeout -+ BEGIN AFTER lsn WITHIN 60000;+ wait changes for 60 seconds and then cancel to start transaction+ to perevent freezing on streaming replication connection failures.+ no wait, just check -+ BEGIN AFTER lsn WITHIN 1;+ some time it is useful just check if changes was replayed+ <phrase>where <replaceable class="parameter">number_of_milliseconds</replaceable> can:</phrase>+ </para>+ </listitem>+ </varlistentry>
</variablelist>
<para>
@@ -123,6 +170,33 @@ BEGIN [ WORK | TRANSACTION ] [ <replaceable class="parameter">transaction_mode</
<programlisting>
BEGIN;
</programlisting></para>
++ <para>+ To begin a transaction block after replaying the given <acronym>LSN</acronym>.+ The command will be canceled if the given <acronym>LSN</acronym> is not+ reached within the timeout of one second.+<programlisting>+BEGIN AFTER '0/3F05F791' WITHIN 1000;+BEGIN+</programlisting></para>++ <para>+ This way we just check (without waiting) if <acronym>LSN</acronym> was replayed,+ and if it was then start transaction.+<programlisting>+BEGIN AFTER '0/3F0FF791' WITHIN 1;+ERROR: canceling waiting for LSN due to timeout+</programlisting></para>++ <para>+ To begin a transaction block after replaying the given <acronym>LSN</acronym>.+ The command will be canceled only on user request or postmaster death.+<programlisting>+BEGIN AFTER '0/3F0FF791';+^CCancel request sent+ERROR: canceling statement due to user request+</programlisting></para>+
</refsect1>
<refsect1>
diff --git a/doc/src/sgml/ref/start_transaction.sgml b/doc/src/sgml/ref/start_transaction.sgmlindex 74ccd7e345..46a3bcf1a8 100644--- a/doc/src/sgml/ref/start_transaction.sgml+++ b/doc/src/sgml/ref/start_transaction.sgml@@ -21,13 +21,16 @@ PostgreSQL documentation
<refsynopsisdiv>
<synopsis>
-START TRANSACTION [ <replaceable class="parameter">transaction_mode</replaceable> [, ...] ]+START TRANSACTION [ <replaceable class="parameter">transaction_mode</replaceable> [, ...] ] <replaceable class="parameter">wait_event</replaceable>
<phrase>where <replaceable class="parameter">transaction_mode</replaceable> is one of:</phrase>
ISOLATION LEVEL { SERIALIZABLE | REPEATABLE READ | READ COMMITTED | READ UNCOMMITTED }
READ WRITE | READ ONLY
[ NOT ] DEFERRABLE
++<phrase>where <replaceable class="parameter">wait_event</replaceable> is:</phrase>+ AFTER <replaceable class="parameter">lsn_value</replaceable> [ WITHIN number_of_milliseconds ]
</synopsis>
</refsynopsisdiv>
diff --git a/src/backend/access/transam/xlogrecovery.c b/src/backend/access/transam/xlogrecovery.cindex 29c5bec084..89b997ad47 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/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..ca15afa468--- /dev/null+++ b/src/backend/commands/waitlsn.c@@ -0,0 +1,278 @@+/*-------------------------------------------------------------------------+ *+ * waitlsn.c+ * Implements waiting for the given LSN, which is used in+ * BEGIN AFTER ... [ WITHIN ... ] clause.+ *+ * Copyright (c) 2024, PostgreSQL Global Development Group+ *+ * IDENTIFICATION+ * src/backend/commands/waitlsn.c+ *+ *-------------------------------------------------------------------------+ */++#include <float.h>+#include <math.h>+#include "postgres.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 "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/timestamp.h"+#include "executor/spi.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, int millisecs)+{+ 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() + millisecs * 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 (millisecs > 0)+ 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_CLIENT_READ);++ 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;+ }+}diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.yindex c6e2f679fd..be97f5a870 100644--- a/src/backend/parser/gram.y+++ b/src/backend/parser/gram.y@@ -647,6 +647,8 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query);
%type <list> hash_partbound
%type <defelt> hash_partbound_elem
+%type <ival> after_lsn_timeout+
%type <node> json_format_clause
json_format_clause_opt
json_value_expr
@@ -3193,6 +3195,14 @@ hash_partbound:
}
;
+/*+ * WITHIN timeout optional clause for BEGIN AFTER lsn+ */+after_lsn_timeout:+ WITHIN Iconst { $$ = $2; }+ | /* EMPTY */ { $$ = 0; }+ ;+
/*****************************************************************************
*
* ALTER TYPE
@@ -10949,6 +10959,17 @@ TransactionStmt:
n->location = -1;
$$ = (Node *) n;
}
+ | START TRANSACTION transaction_mode_list_or_empty AFTER Sconst after_lsn_timeout+ {+ TransactionStmt *n = makeNode(TransactionStmt);++ n->kind = TRANS_STMT_START;+ n->options = $3;+ n->afterLsn = $5;+ n->afterLsnTimeout = $6;+ n->location = -1;+ $$ = (Node *) n;+ }
| COMMIT opt_transaction opt_transaction_chain
{
TransactionStmt *n = makeNode(TransactionStmt);
@@ -11053,6 +11074,17 @@ TransactionStmtLegacy:
n->location = -1;
$$ = (Node *) n;
}
+ | BEGIN_P opt_transaction transaction_mode_list_or_empty AFTER Sconst after_lsn_timeout+ {+ TransactionStmt *n = makeNode(TransactionStmt);++ n->kind = TRANS_STMT_BEGIN;+ n->options = $3;+ n->afterLsn = $5;+ n->afterLsnTimeout = $6;+ n->location = -1;+ $$ = (Node *) n;+ }
| END_P opt_transaction opt_transaction_chain
{
TransactionStmt *n = makeNode(TransactionStmt);
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/backend/tcop/utility.c b/src/backend/tcop/utility.cindex 83f86a42f7..bf9a685a9a 100644--- a/src/backend/tcop/utility.c+++ b/src/backend/tcop/utility.c@@ -56,6 +56,7 @@
#include "commands/user.h"
#include "commands/vacuum.h"
#include "commands/view.h"
+#include "commands/waitlsn.h"
#include "miscadmin.h"
#include "parser/parse_utilcmd.h"
#include "postmaster/bgwriter.h"
@@ -63,8 +64,10 @@
#include "storage/fd.h"
#include "tcop/utility.h"
#include "utils/acl.h"
+#include "utils/fmgrprotos.h"
#include "utils/guc.h"
#include "utils/lsyscache.h"
+#include "utils/pg_lsn.h"
/* Hook for plugins to get control in ProcessUtility() */
ProcessUtility_hook_type ProcessUtility_hook = NULL;
@@ -606,6 +609,15 @@ standard_ProcessUtility(PlannedStmt *pstmt,
{
ListCell *lc;
+ if (stmt->afterLsn)+ {+ XLogRecPtr lsn = InvalidXLogRecPtr;++ lsn = DatumGetLSN(DirectFunctionCall1(pg_lsn_in,+ CStringGetDatum(stmt->afterLsn)));+ WaitForLSN(lsn, stmt->afterLsnTimeout);+ }+
BeginTransactionBlock();
foreach(lc, stmt->options)
{
diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.cindex 19db069ee9..28e22cb9d0 100644--- a/src/bin/psql/tab-complete.c+++ b/src/bin/psql/tab-complete.c@@ -2717,7 +2717,7 @@ psql_completion(const char *text, int start, int end)
/* BEGIN */
else if (Matches("BEGIN"))
- COMPLETE_WITH("WORK", "TRANSACTION", "ISOLATION LEVEL", "READ", "DEFERRABLE", "NOT DEFERRABLE");+ COMPLETE_WITH("WORK", "TRANSACTION", "ISOLATION LEVEL", "READ", "DEFERRABLE", "NOT DEFERRABLE", "AFTER");
/* END, ABORT */
else if (Matches("END|ABORT"))
COMPLETE_WITH("AND", "WORK", "TRANSACTION");
@@ -4610,7 +4610,7 @@ psql_completion(const char *text, int start, int end)
/* START TRANSACTION */
else if (Matches("START"))
- COMPLETE_WITH("TRANSACTION");+ COMPLETE_WITH("TRANSACTION", "AFTER");
/* TABLE, but not TABLE embedded in other commands */
else if (Matches("TABLE"))
diff --git a/src/include/commands/waitlsn.h b/src/include/commands/waitlsn.hnew file mode 100644index 0000000000..ff4d63d5b0--- /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"++extern void WaitForLSN(XLogRecPtr lsn, int millisecs);+extern Size WaitLSNShmemSize(void);+extern void WaitLSNShmemInit(void);+extern void WaitLSNSetLatches(XLogRecPtr curLSN);+extern void WaitLSNCleanup(void);++/* 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;++#endif /* WAIT_LSN_H */diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.hindex aadaf67f57..c2077b5252 100644--- a/src/include/nodes/parsenodes.h+++ b/src/include/nodes/parsenodes.h@@ -3527,6 +3527,8 @@ typedef struct TransactionStmt
/* for two-phase-commit related commands */
char *gid pg_node_attr(query_jumble_ignore);
bool chain; /* AND CHAIN option */
+ char *afterLsn; /* target LSN to wait */+ int afterLsnTimeout; /* LSN waiting timeout */
/* token location, or -1 if unknown */
int location pg_node_attr(query_jumble_location);
} TransactionStmt;
diff --git a/src/test/recovery/meson.build b/src/test/recovery/meson.buildindex b1eb77b1ec..a1c2a0b13d 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_begin_after.pl',
],
},
}
diff --git a/src/test/recovery/t/043_begin_after.pl b/src/test/recovery/t/043_begin_after.plnew file mode 100644index 0000000000..e7859623b5--- /dev/null+++ b/src/test/recovery/t/043_begin_after.pl@@ -0,0 +1,61 @@+# Checks waiting for lsn on standby AFTER+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 = 1;+$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 AFTER works: add new content to primary and memorize+# primary's new LSN, then wait for primary's LSN on standby. Prove that AFTER 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[+ BEGIN AFTER '${lsn1}';+ 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 AFTER");++my $lsn2 =+ $node_primary->safe_psql('postgres', "SELECT pg_current_wal_lsn() + 1");+my $stderr;+$node_standby->safe_psql('postgres', "BEGIN AFTER '$lsn1' WITHIN 1");+$node_standby->psql(+ 'postgres',+ "BEGIN AFTER '$lsn2' WITHIN 1",+ stderr => \$stderr);+ok( $stderr =~ /canceling waiting for LSN due to timeout/,+ "get timeout on waiting for unreachable LSN");++$node_standby->stop;+$node_primary->stop;+done_testing();diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.listindex aa7a25b8f8..7a2ac178bc 100644--- a/src/tools/pgindent/typedefs.list+++ b/src/tools/pgindent/typedefs.list@@ -3035,6 +3035,8 @@ WaitEventIO
WaitEventIPC
WaitEventSet
WaitEventTimeout
+WaitLSNProcInfo+WaitLSNState
WaitPMResult
WalCloseMethod
WalCompression
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, 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: <59cf2cbdc6f9bf81602e3515acca22b0@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