database/postgres
1
0
Fork 0
mirror of https://github.com/postgres/postgres.git synced 2025-07-02 09:02:37 +03:00
Code Activity

Revert: Implement pg_wal_replay_wait() stored procedure

Browse Source 
This commit reverts 06c418e163, e37662f221, bf1e650806, 25f42429e2,
ee79928441, and 74eaf66f98 per review by Heikki Linnakangas.

Discussion: https://postgr.es/m/b155606b-e744-4218-bda5-29379779da1a%40iki.fi
This commit is contained in:
Alexander Korotkov
2024-04-11 16:30:32 +03:00
parent 922c4c461d
commit 772faafca1
20 changed files with 4 additions and 694 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
src/backend/access/transam/xact.c
View File

@ -38,7 +38,6 @@
#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"
@ -2772,11 +2771,6 @@ 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();

7
src/backend/access/transam/xlog.c
View File

@ -66,7 +66,6 @@
#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"
@ -6130,12 +6129,6 @@ 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())

11
src/backend/access/transam/xlogrecovery.c
View File

@ -43,7 +43,6 @@
#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"
@ -1829,16 +1828,6 @@ 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 (waitLSN &&
(XLogRecoveryCtl->lastReplayedEndRecPtr >=
pg_atomic_read_u64(&waitLSN->minWaitedLSN)))
WaitLSNSetLatches(XLogRecoveryCtl->lastReplayedEndRecPtr);
/* Else, try to fetch the next WAL record */
record = ReadRecord(xlogprefetcher, LOG, false, replayTLI);
} while (record != NULL);

3
src/backend/catalog/system_functions.sql
View File

@ -414,9 +414,6 @@ 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)

3
src/backend/commands/Makefile
View File

@ -61,7 +61,6 @@ OBJS = \
vacuum.o \
vacuumparallel.o \
variable.o \
view.o \
waitlsn.o
view.o
include $(top_srcdir)/src/backend/common.mk

1
src/backend/commands/meson.build
View File

@ -50,5 +50,4 @@ backend_sources += files(
'vacuumparallel.c',
'variable.c',
'view.c',
'waitlsn.c',
)

337
src/backend/commands/waitlsn.c
View File

@ -1,337 +0,0 @@
/*-------------------------------------------------------------------------
*
* 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 lsn_cmp(const pairingheap_node *a, const pairingheap_node *b,
void *arg);
struct WaitLSNState *waitLSN = 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;
waitLSN = (WaitLSNState *) ShmemInitStruct("WaitLSNState",
WaitLSNShmemSize(),
&found);
if (!found)
{
pg_atomic_init_u64(&waitLSN->minWaitedLSN, PG_UINT64_MAX);
pairingheap_initialize(&waitLSN->waitersHeap, lsn_cmp, NULL);
memset(&waitLSN->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
lsn_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(&waitLSN->waitersHeap))
{
pairingheap_node *node = pairingheap_first(&waitLSN->waitersHeap);
minWaitedLSN = pairingheap_container(WaitLSNProcInfo, phNode, node)->waitLSN;
}
pg_atomic_write_u64(&waitLSN->minWaitedLSN, minWaitedLSN);
}
/*
* Put the current process into the heap of LSN waiters.
*/
static void
addLSNWaiter(XLogRecPtr lsn)
{
WaitLSNProcInfo *procInfo = &waitLSN->procInfos[MyProcNumber];
LWLockAcquire(WaitLSNLock, LW_EXCLUSIVE);
Assert(!procInfo->inHeap);
procInfo->procnum = MyProcNumber;
procInfo->waitLSN = lsn;
pairingheap_add(&waitLSN->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 = &waitLSN->procInfos[MyProcNumber];
LWLockAcquire(WaitLSNLock, LW_EXCLUSIVE);
if (!procInfo->inHeap)
{
LWLockRelease(WaitLSNLock);
return;
}
pairingheap_remove(&waitLSN->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;
int *wakeUpProcNums;
int numWakeUpProcs = 0;
wakeUpProcNums = palloc(sizeof(int) * MaxBackends);
LWLockAcquire(WaitLSNLock, LW_EXCLUSIVE);
/*
* Iterate the pairing heap of waiting processes till we find LSN not yet
* replayed. Record the process numbers to set their latches later.
*/
while (!pairingheap_is_empty(&waitLSN->waitersHeap))
{
pairingheap_node *node = pairingheap_first(&waitLSN->waitersHeap);
WaitLSNProcInfo *procInfo = pairingheap_container(WaitLSNProcInfo, phNode, node);
if (!XLogRecPtrIsInvalid(currentLSN) &&
procInfo->waitLSN > currentLSN)
break;
wakeUpProcNums[numWakeUpProcs++] = procInfo->procnum;
(void) pairingheap_remove_first(&waitLSN->waitersHeap);
procInfo->inHeap = false;
}
updateMinWaitedLSN();
LWLockRelease(WaitLSNLock);
/*
* Set latches for processes, whose waited LSNs are already replayed. This
* involves spinlocks. So, we shouldn't do this under a spinlock.
*/
for (i = 0; i < numWakeUpProcs; i++)
{
PGPROC *backend;
backend = GetPGProcByNumber(wakeUpProcNums[i]);
SetLatch(&backend->procLatch);
}
pfree(wakeUpProcNums);
}
/*
* 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 (waitLSN->procInfos[MyProcNumber].inHeap)
deleteLSNWaiter();
}
/*
* Wait using MyLatch till the given LSN is replayed, the postmaster dies or
* timeout happens.
*/
void
WaitForLSN(XLogRecPtr targetLSN, int64 timeout)
{
XLogRecPtr currentLSN;
TimestampTz endtime = 0;
/* Shouldn't be called when shmem isn't initialized */
Assert(waitLSN);
/* Should be only called by a backend */
Assert(MyBackendType == B_BACKEND && 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);
addLSNWaiter(targetLSN);
for (;;)
{
int rc;
int latch_events = WL_LATCH_SET | WL_EXIT_ON_PM_DEATH;
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 the target LSN %X/%X; last replay LSN %X/%X.",
LSN_FORMAT_ARGS(targetLSN),
LSN_FORMAT_ARGS(currentLSN))));
if (timeout > 0)
{
delay_ms = (endtime - GetCurrentTimestamp()) / 1000;
latch_events |= WL_TIMEOUT;
if (delay_ms <= 0)
break;
}
CHECK_FOR_INTERRUPTS();
rc = WaitLatch(MyLatch, latch_events, delay_ms,
WAIT_EVENT_WAIT_FOR_WAL_REPLAY);
if (rc & WL_LATCH_SET)
ResetLatch(MyLatch);
}
if (targetLSN > currentLSN)
{
deleteLSNWaiter();
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);
CallContext *context = (CallContext *) fcinfo->context;
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 check that pg_wal_replay_wait() is called in a non-atomic
* context. That is, a procedure call isn't wrapped into a transaction,
* another procedure call, or a function call.
*
* Secondly, 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 (context->atomic)
ereport(ERROR,
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
errmsg("pg_wal_replay_wait() must be only called in non-atomic context"),
errdetail("Make sure pg_wal_replay_wait() isn't called within a transaction, another procedure, or a function.")));
if (ActiveSnapshotSet())
PopActiveSnapshot();
Assert(!ActiveSnapshotSet());
InvalidateCatalogSnapshot();
Assert(MyProc->xmin == InvalidTransactionId);
(void) WaitForLSN(target_lsn, timeout);
PG_RETURN_VOID();
}

18
src/backend/lib/pairingheap.c
View File

@ -44,26 +44,12 @@ 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;
}
/*

3
src/backend/storage/ipc/ipci.c
View File

@ -25,7 +25,6 @@
#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"
@ -153,7 +152,6 @@ 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
@ -359,7 +357,6 @@ CreateOrAttachShmemStructs(void)
StatsShmemInit();
WaitEventExtensionShmemInit();
InjectionPointShmemInit();
WaitLSNShmemInit();
}
/*

6
src/backend/storage/lmgr/proc.c
View File

@ -36,7 +36,6 @@
#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"
@ -863,11 +862,6 @@ ProcKill(int code, Datum arg)
*/
LWLockReleaseAll();
/*
* Cleanup waiting for LSN if any.
*/
WaitLSNCleanup();
/* Cancel any pending condition variable sleep, too */
ConditionVariableCancelSleep();

2
src/backend/utils/activity/wait_event_names.txt
View File

@ -87,7 +87,6 @@ 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."
@ -346,7 +345,6 @@ 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)