From: a.pervushina@postgrespro.ru
To: Anna Akenteva <a.akenteva@postgrespro.ru>
Cc: Daniel Gustafsson <daniel@yesql.se>
Cc: Andres Freund <andres@anarazel.de>
Cc: Tom Lane <tgl@sss.pgh.pa.us>
Cc: Fujii Masao <masao.fujii@oss.nttdata.com>
Cc: Alexey Kondratov <a.kondratov@postgrespro.ru>
Cc: Kyotaro Horiguchi <horikyota.ntt@gmail.com>
Cc: a.korotkov@postgrespro.ru
Cc: i.kartyshov@postgrespro.ru
Cc: amit.kapila16@gmail.com
Cc: pgsql-hackers@lists.postgresql.org
Subject: Re: [HACKERS] make async slave to wait for lsn to be replayed
Date: Mon, 16 Nov 2020 13:09:30 +0300
Message-ID: <f6522632dc01fb4a3d3659a4c4917242@postgrespro.ru> (raw)
Hello,
I've changed the BEGIN WAIT FOR LSN statement to core functions
pg_waitlsn, pg_waitlsn_infinite and pg_waitlsn_no_wait.
Currently the functions work inside repeatable read transactions, but
waitlsn creates a snapshot if called first in a transaction block, which
can possibly lead the transaction to working incorrectly, so the
function gives a warning.
Usage examples
==========
select pg_waitlsn(‘LSN’, timeout);
select pg_waitlsn_infinite(‘LSN’);
select pg_waitlsn_no_wait(‘LSN’);
Attachments:
[text/x-diff] pg_waitlsn_v10.patch (16.7K, ../f6522632dc01fb4a3d3659a4c4917242@postgrespro.ru/2-pg_waitlsn_v10.patch)
download | inline diff:diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.cindex 9d3f1c12fc5..27e964c8111 100644--- a/src/backend/access/transam/xlog.c+++ b/src/backend/access/transam/xlog.c@@ -42,6 +42,7 @@
#include "catalog/pg_database.h"
#include "commands/progress.h"
#include "commands/tablespace.h"
+#include "commands/wait.h"
#include "common/controldata_utils.h"
#include "executor/instrument.h"
#include "miscadmin.h"
@@ -7387,6 +7388,15 @@ StartupXLOG(void)
break;
}
+ /*+ * If we replayed an LSN that someone was waiting for,+ * set latches in shared memory array to notify the waiter.+ */+ if (XLogCtl->lastReplayedEndRecPtr >= GetMinWaitedLSN())+ {+ WaitSetLatch(XLogCtl->lastReplayedEndRecPtr);+ }+
/* Else, try to fetch the next WAL record */
record = ReadRecord(xlogreader, LOG, false);
} while (record != NULL);
diff --git a/src/backend/commands/Makefile b/src/backend/commands/Makefileindex d4815d3ce65..9b310926c12 100644--- a/src/backend/commands/Makefile+++ b/src/backend/commands/Makefile@@ -57,6 +57,7 @@ OBJS = \
user.o \
vacuum.o \
variable.o \
- view.o+ view.o \+ wait.o
include $(top_srcdir)/src/backend/common.mk
diff --git a/src/backend/commands/wait.c b/src/backend/commands/wait.cnew file mode 100644index 00000000000..ac18040c362--- /dev/null+++ b/src/backend/commands/wait.c@@ -0,0 +1,297 @@+/*-------------------------------------------------------------------------+ *+ * wait.c+ * Implements waitlsn, which allows waiting for events such as+ * LSN having been replayed on replica.+ *+ * Portions Copyright (c) 1996-2020, PostgreSQL Global Development Group+ * Portions Copyright (c) 2020, Regents of PostgresPro+ *+ * IDENTIFICATION+ * src/backend/commands/wait.c+ *+ *-------------------------------------------------------------------------+ */+#include "postgres.h"++#include <math.h>++#include "access/xact.h"+#include "access/xlog.h"+#include "access/xlogdefs.h"+#include "commands/wait.h"+#include "funcapi.h"+#include "miscadmin.h"+#include "pgstat.h"+#include "storage/backendid.h"+#include "storage/pmsignal.h"+#include "storage/proc.h"+#include "storage/shmem.h"+#include "storage/sinvaladt.h"+#include "storage/spin.h"+#include "utils/builtins.h"+#include "utils/pg_lsn.h"+#include "utils/timestamp.h"++/* Add to shared memory array */+static void AddWaitedLSN(XLogRecPtr lsn_to_wait);++/* Shared memory structure */+typedef struct+{+ int backend_maxid;+ pg_atomic_uint64 min_lsn; /* XLogRecPtr of minimal waited for LSN */+ slock_t mutex;+ /* LSNs that different backends are waiting */+ XLogRecPtr lsn[FLEXIBLE_ARRAY_MEMBER];+} WaitState;++static WaitState *state;++/*+ * Add the wait event of the current backend to shared memory array+ */+static void+AddWaitedLSN(XLogRecPtr lsn_to_wait)+{+ SpinLockAcquire(&state->mutex);+ if (state->backend_maxid < MyBackendId)+ state->backend_maxid = MyBackendId;++ state->lsn[MyBackendId] = lsn_to_wait;++ if (lsn_to_wait < state->min_lsn.value)+ state->min_lsn.value = lsn_to_wait;+ SpinLockRelease(&state->mutex);+}++/*+ * Delete wait event of the current backend from the shared memory array.+ */+void+DeleteWaitedLSN(void)+{+ int i;+ XLogRecPtr lsn_to_delete;++ SpinLockAcquire(&state->mutex);++ lsn_to_delete = state->lsn[MyBackendId];+ state->lsn[MyBackendId] = InvalidXLogRecPtr;++ /* If we are deleting the minimal LSN, then choose the next min_lsn */+ if (lsn_to_delete != InvalidXLogRecPtr &&+ lsn_to_delete == state->min_lsn.value)+ {+ state->min_lsn.value = PG_UINT64_MAX;+ for (i = 2; i <= state->backend_maxid; i++)+ if (state->lsn[i] != InvalidXLogRecPtr &&+ state->lsn[i] < state->min_lsn.value)+ state->min_lsn.value = state->lsn[i];+ }++ /* If deleting from the end of the array, shorten the array's used part */+ if (state->backend_maxid == MyBackendId)+ for (i = (MyBackendId); i >= 2; i--)+ if (state->lsn[i] != InvalidXLogRecPtr)+ {+ state->backend_maxid = i;+ break;+ }++ SpinLockRelease(&state->mutex);+}++/*+ * Report amount of shared memory space needed for WaitState+ */+Size+WaitShmemSize(void)+{+ Size size;++ size = offsetof(WaitState, lsn);+ size = add_size(size, mul_size(MaxBackends + 1, sizeof(XLogRecPtr)));+ return size;+}++/*+ * Initialize an array of events to wait for in shared memory+ */+void+WaitShmemInit(void)+{+ bool found;+ uint32 i;++ state = (WaitState *) ShmemInitStruct("pg_wait_lsn",+ WaitShmemSize(),+ &found);+ if (!found)+ {+ SpinLockInit(&state->mutex);++ for (i = 0; i < (MaxBackends + 1); i++)+ state->lsn[i] = InvalidXLogRecPtr;++ state->backend_maxid = 0;+ state->min_lsn.value = PG_UINT64_MAX;+ }+}++/*+ * Set latches in shared memory to signal that new LSN has been replayed+ */+void+WaitSetLatch(XLogRecPtr cur_lsn)+{+ uint32 i;+ int backend_maxid;+ PGPROC *backend;++ SpinLockAcquire(&state->mutex);+ backend_maxid = state->backend_maxid;++ for (i = 2; i <= backend_maxid; i++)+ {+ backend = BackendIdGetProc(i);++ if (backend && state->lsn[i] != 0 &&+ state->lsn[i] <= cur_lsn)+ {+ SetLatch(&backend->procLatch);+ }+ }+ SpinLockRelease(&state->mutex);+}++/*+ * Get minimal LSN that someone waits for+ */+XLogRecPtr+GetMinWaitedLSN(void)+{+ return state->min_lsn.value;+}++/*+ * On WAIT use a latch to wait till LSN is replayed,+ * postmaster dies or timeout happens. Timeout is specified in milliseconds.+ * Returns true if LSN was reached and false otherwise.+ */+bool+WaitUtility(XLogRecPtr target_lsn, const int timeout_ms)+{+ XLogRecPtr cur_lsn = GetXLogReplayRecPtr(NULL);+ int latch_events;+ float8 endtime;+ bool res = false;+ bool wait_forever = (timeout_ms <= 0);++ if (!RecoveryInProgress()) {+ ereport(ERROR,+ errmsg("Cannot use waitlsn on primary"));+ return false;+ }++ /*+ * In transactions, that have isolation level repeatable read or higher+ * waitlsn creates a snapshot if called first in a block, which can+ * lead the transaction to working incorrectly+ */++ if (IsTransactionBlock() && XactIsoLevel != XACT_READ_COMMITTED) {+ ereport(WARNING,+ errmsg("Waitlsn may work incorrectly in this isolation level"),+ errhint("Call waitlsn before starting the transaction"));+ }++ endtime = GetNowFloat() + timeout_ms / 1000.0;++ latch_events = WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH;++ /* Check if we already reached the needed LSN */+ if (cur_lsn >= target_lsn)+ return true;++ AddWaitedLSN(target_lsn);++ for (;;)+ {+ int rc;+ float8 time_left = 0;+ long time_left_ms = 0;++ time_left = endtime - GetNowFloat();++ /* Use 100 ms as the default timeout to check for interrupts */+ if (wait_forever || time_left < 0 || time_left > 0.1)+ time_left_ms = 100;+ else+ time_left_ms = (long) ceil(time_left * 1000.0);++ /* If interrupt, LockErrorCleanup() will do DeleteWaitedLSN() for us */+ CHECK_FOR_INTERRUPTS();++ /* If postmaster dies, finish immediately */+ if (!PostmasterIsAlive())+ break;++ rc = WaitLatch(MyLatch, latch_events, time_left_ms,+ WAIT_EVENT_CLIENT_READ);++ ResetLatch(MyLatch);++ if (rc & WL_LATCH_SET)+ cur_lsn = GetXLogReplayRecPtr(NULL);++ if (rc & WL_TIMEOUT)+ {+ cur_lsn = GetXLogReplayRecPtr(NULL);+ /* If the time specified by user has passed, stop waiting */+ time_left = endtime - GetNowFloat();+ if (!wait_forever && time_left <= 0.0)+ break;+ }++ /* If LSN has been replayed */+ if (target_lsn <= cur_lsn)+ break;+ }++ DeleteWaitedLSN();++ if (cur_lsn < target_lsn)+ ereport(WARNING,+ errmsg("LSN was not reached"),+ errhint("Try to increase wait time."));+ else+ res = true;++ return res;+}++Datum+pg_waitlsn(PG_FUNCTION_ARGS)+{+ XLogRecPtr trg_lsn = PG_GETARG_LSN(0);+ uint64_t delay = PG_GETARG_INT32(1);++ PG_RETURN_BOOL(WaitUtility(trg_lsn, delay));+}++Datum+pg_waitlsn_infinite(PG_FUNCTION_ARGS)+{+ XLogRecPtr trg_lsn = PG_GETARG_LSN(0);++ PG_RETURN_BOOL(WaitUtility(trg_lsn, 0));+}++Datum+pg_waitlsn_no_wait(PG_FUNCTION_ARGS)+{+ XLogRecPtr trg_lsn = PG_GETARG_LSN(0);++ PG_RETURN_BOOL(WaitUtility(trg_lsn, 1));+}
\ No newline at end of file
diff --git a/src/backend/storage/ipc/ipci.c b/src/backend/storage/ipc/ipci.cindex 427b0d59cde..2dcfdde5f3f 100644--- a/src/backend/storage/ipc/ipci.c+++ b/src/backend/storage/ipc/ipci.c@@ -22,6 +22,7 @@
#include "access/syncscan.h"
#include "access/twophase.h"
#include "commands/async.h"
+#include "commands/wait.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "postmaster/autovacuum.h"
@@ -147,6 +148,7 @@ CreateSharedMemoryAndSemaphores(void)
size = add_size(size, BTreeShmemSize());
size = add_size(size, SyncScanShmemSize());
size = add_size(size, AsyncShmemSize());
+ size = add_size(size, WaitShmemSize());
#ifdef EXEC_BACKEND
size = add_size(size, ShmemBackendArraySize());
#endif
@@ -264,6 +266,11 @@ CreateSharedMemoryAndSemaphores(void)
SyncScanShmemInit();
AsyncShmemInit();
+ /*+ * Init array of events for the wait clause in shared memory+ */+ WaitShmemInit();+
#ifdef EXEC_BACKEND
/*
diff --git a/src/backend/storage/lmgr/proc.c b/src/backend/storage/lmgr/proc.cindex 5844540ccca..42f8b33c2f4 100644--- a/src/backend/storage/lmgr/proc.c+++ b/src/backend/storage/lmgr/proc.c@@ -38,6 +38,7 @@
#include "access/transam.h"
#include "access/twophase.h"
#include "access/xact.h"
+#include "commands/wait.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "postmaster/autovacuum.h"
@@ -717,6 +718,9 @@ LockErrorCleanup(void)
AbortStrongLockAcquire();
+ /* If waitlsn was interrupted, then stop waiting for that LSN */+ DeleteWaitedLSN();+
/* Nothing to do if we weren't waiting for a lock */
if (lockAwaited == NULL)
{
diff --git a/src/backend/utils/adt/misc.c b/src/backend/utils/adt/misc.cindex 37c23c9155a..2181191d99e 100644--- a/src/backend/utils/adt/misc.c+++ b/src/backend/utils/adt/misc.c@@ -373,8 +373,6 @@ pg_sleep(PG_FUNCTION_ARGS)
* less than the specified time when WaitLatch is terminated early by a
* non-query-canceling signal such as SIGHUP.
*/
-#define GetNowFloat() ((float8) GetCurrentTimestamp() / 1000000.0)-
endtime = GetNowFloat() + secs;
for (;;)
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.datindex c2287273a93..16177049f6f 100644--- a/src/include/catalog/pg_proc.dat+++ b/src/include/catalog/pg_proc.dat@@ -10937,4 +10937,19 @@
proname => 'is_normalized', prorettype => 'bool', proargtypes => 'text text',
prosrc => 'unicode_is_normalized' },
+{ oid => '16387', descr => 'wait for LSN until timeout',+ proname => 'pg_waitlsn', prorettype => 'bool', proargtypes => 'pg_lsn int8',+ proargnames => '{trg_lsn,delay}',+ prosrc => 'pg_waitlsn' },++{ oid => '16388', descr => 'wait for LSN for an infinite time',+ proname => 'pg_waitlsn_infinite', prorettype => 'bool', proargtypes => 'pg_lsn',+ proargnames => '{trg_lsn}',+ prosrc => 'pg_waitlsn_infinite' },++{ oid => '16389', descr => 'wait for LSN with no timeout',+ proname => 'pg_waitlsn_no_wait', prorettype => 'bool', proargtypes => 'pg_lsn',+ proargnames => '{trg_lsn}',+ prosrc => 'pg_waitlsn_no_wait' },+
]
diff --git a/src/include/commands/wait.h b/src/include/commands/wait.hnew file mode 100644index 00000000000..fd21e434167--- /dev/null+++ b/src/include/commands/wait.h@@ -0,0 +1,26 @@+/*-------------------------------------------------------------------------+ *+ * wait.h+ * prototypes for commands/wait.c+ *+ * Portions Copyright (c) 1996-2020, PostgreSQL Global Development Group+ * Portions Copyright (c) 2020, Regents of PostgresPro+ *+ * src/include/commands/wait.h+ *+ *-------------------------------------------------------------------------+ */+#ifndef WAIT_H+#define WAIT_H+#include "postgres.h"+#include "tcop/dest.h"+#include "nodes/parsenodes.h"++extern bool WaitUtility(XLogRecPtr lsn, const int timeout_ms);+extern Size WaitShmemSize(void);+extern void WaitShmemInit(void);+extern void WaitSetLatch(XLogRecPtr cur_lsn);+extern XLogRecPtr GetMinWaitedLSN(void);+extern void DeleteWaitedLSN(void);++#endif /* WAIT_H */diff --git a/src/include/utils/timestamp.h b/src/include/utils/timestamp.hindex 16c3fd8ec97..4e85b6f9db1 100644--- a/src/include/utils/timestamp.h+++ b/src/include/utils/timestamp.h@@ -111,4 +111,6 @@ extern int date2isoyearday(int year, int mon, int mday);
extern bool TimestampTimestampTzRequiresRewrite(void);
+#define GetNowFloat() ((float8) GetCurrentTimestamp() / 1000000.0)+
#endif /* TIMESTAMP_H */
diff --git a/src/test/recovery/t/021_waitlsn.pl b/src/test/recovery/t/021_waitlsn.plnew file mode 100644index 00000000000..81dd70ef963--- /dev/null+++ b/src/test/recovery/t/021_waitlsn.pl@@ -0,0 +1,91 @@+# Checks waitlsn+use strict;+use warnings;++use PostgresNode;+use TestLib;+use Test::More tests => 11;++# Initialize primary node+my $node_primary = get_new_node('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);++# Using the backup, create a streaming standby with a 1 second delay+my $node_standby = get_new_node('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;++# Check that timeouts make us wait for the specified time (1s here)+my $current_time = $node_standby->safe_psql('postgres', "SELECT now()");+my $two_seconds = 2000; # in milliseconds+my $start_time = time();+$node_standby->safe_psql('postgres',+ "SELECT pg_waitlsn('0/FFFFFFFF', $two_seconds)");+my $time_waited = (time() - $start_time) * 1000; # convert to milliseconds+ok($time_waited >= $two_seconds, "waitlsn waits for enough time");++# Check that timeouts let us stop waiting right away, before reaching target LSN+$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 ($ret, $out, $err) = $node_standby->psql('postgres',+ "SELECT pg_waitlsn('$lsn1', 1)");++ok($ret == 0, "zero return value when failed to waitlsn on standby");+ok($err =~ /WARNING: LSN was not reached/,+ "correct error message when failed to waitlsn on standby");+ok($out eq "f", "if given too little wait time, WAIT doesn't reach target LSN");+++# Check that waitlsn works fine and reaches target LSN if given no timeout++# Add data on primary, memorize primary's last LSN+$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_lsn()");++# Wait for it to appear on replica, memorize replica's last LSN+$node_standby->safe_psql('postgres',+ "SELECT pg_waitlsn_infinite('$lsn2')");+my $reached_lsn = $node_standby->safe_psql('postgres',+ "SELECT pg_last_wal_replay_lsn()");++# Make sure that primary's and replica's LSNs are the same after WAIT+my $compare_lsns = $node_standby->safe_psql('postgres',+ "SELECT pg_lsn_cmp('$reached_lsn'::pg_lsn, '$lsn2'::pg_lsn)");+ok($compare_lsns eq 0,+ "standby reached the same LSN as primary before starting transaction");+++# Make sure that it's not allowed to use waitlsn on primary+($ret, $out, $err) = $node_primary->psql('postgres',+ "SELECT pg_waitlsn_infinite('0/FFFFFFFF')");++ok($ret != 0, "non-zero return value when trying to waitlsn on primary");+ok($err =~ /ERROR: Cannot use waitlsn on primary/,+ "correct error message when trying to waitlsn on primary");+ok($out eq '', "empty output when trying to waitlsn on primary");++# Make sure that waitlsn gives a warning inside a read commited transaction++($ret, $out, $err) = $node_standby->psql('postgres',+ "BEGIN ISOLATION LEVEL REPEATABLE READ; SELECT pg_waitlsn_no_wait('0/FFFFFFFF')");+ok($ret == 0, "zero return value when trying to waitlsn in transaction");+ok($err =~ /WARNING: Waitlsn may work incorrectly in this isolation level/,+ "correct warning message when trying to waitlsn in transaction");+ok($out eq "f", "non empty output when trying to waitlsn in transaction");++$node_standby->stop;+$node_primary->stop;
\ No newline at end of file
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.listindex 7eaaad1e140..204a9bc67e1 100644--- a/src/tools/pgindent/typedefs.list+++ b/src/tools/pgindent/typedefs.list@@ -2686,6 +2686,7 @@ WaitEventIPC
WaitEventSet
WaitEventTimeout
WaitPMResult
+WaitState
WalCloseMethod
WalLevel
WalRcvData
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: a.pervushina@postgrespro.ru, a.akenteva@postgrespro.ru, daniel@yesql.se, andres@anarazel.de, tgl@sss.pgh.pa.us, masao.fujii@oss.nttdata.com, a.kondratov@postgrespro.ru, horikyota.ntt@gmail.com, a.korotkov@postgrespro.ru, i.kartyshov@postgrespro.ru, amit.kapila16@gmail.com, pgsql-hackers@lists.postgresql.org
Subject: Re: [HACKERS] make async slave to wait for lsn to be replayed
In-Reply-To: <f6522632dc01fb4a3d3659a4c4917242@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