waitlsn.c File Reference

#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"

Include dependency graph for waitlsn.c:

Go to the source code of this file.

Functions
static int	waitlsn_cmp (const pairingheap_node *a, const pairingheap_node *b, void *arg)
Size	WaitLSNShmemSize (void)
void	WaitLSNShmemInit (void)
static void	updateMinWaitedLSN (void)
static void	addLSNWaiter (XLogRecPtr lsn)
static void	deleteLSNWaiter (void)
void	WaitLSNSetLatches (XLogRecPtr currentLSN)
void	WaitLSNCleanup (void)
static void	WaitForLSNReplay (XLogRecPtr targetLSN, int64 timeout)
Datum	pg_wal_replay_wait (PG_FUNCTION_ARGS)

Variables
struct WaitLSNState *	waitLSNState = NULL

Function Documentation

addLSNWaiter()

static void addLSNWaiter ( XLogRecPtr  lsn )

static

Definition at line 108 of file waitlsn.c.

{
    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);
}

References Assert, WaitLSNProcInfo::inHeap, WaitLSNProcInfo::latch, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MyLatch, MyProcNumber, pairingheap_add(), WaitLSNProcInfo::phNode, WaitLSNState::procInfos, updateMinWaitedLSN(), WaitLSNState::waitersHeap, WaitLSNProcInfo::waitLSN, and waitLSNState.

Referenced by WaitForLSNReplay().

deleteLSNWaiter()

static void deleteLSNWaiter ( void  )

static

Definition at line 130 of file waitlsn.c.

{
    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);
}

References WaitLSNProcInfo::inHeap, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MyProcNumber, pairingheap_remove(), WaitLSNProcInfo::phNode, WaitLSNState::procInfos, updateMinWaitedLSN(), WaitLSNState::waitersHeap, and waitLSNState.

Referenced by WaitForLSNReplay(), and WaitLSNCleanup().

pg_wal_replay_wait()

Datum pg_wal_replay_wait

(

PG_FUNCTION_ARGS

)

Definition at line 341 of file waitlsn.c.

{
    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 snapshot. */
    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();
}

References ActiveSnapshotSet(), Assert, ereport, errcode(), errdetail(), errmsg(), ERROR, GetOldestSnapshot(), InvalidateCatalogSnapshot(), InvalidTransactionId, MyProc, PG_GETARG_INT64, PG_GETARG_LSN, PG_RETURN_VOID, PopActiveSnapshot(), WaitForLSNReplay(), and PGPROC::xmin.

updateMinWaitedLSN()

static void updateMinWaitedLSN ( void  )

static

Definition at line 90 of file waitlsn.c.

{
    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);
}

References WaitLSNState::minWaitedLSN, pairingheap_container, pairingheap_first(), pairingheap_is_empty, pg_atomic_write_u64(), PG_UINT64_MAX, WaitLSNState::waitersHeap, and waitLSNState.

Referenced by addLSNWaiter(), deleteLSNWaiter(), and WaitLSNSetLatches().

WaitForLSNReplay()

static void WaitForLSNReplay ( XLogRecPtr  targetLSN, int64  timeout  )

static

Definition at line 221 of file waitlsn.c.

{
    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())
    {
        /*
         * Recovery is not in progress.  Given that we detected this in the
         * very first check, this procedure was mistakenly called on primary.
         * However, it's possible that standby was promoted concurrently to
         * the procedure call, while target LSN is replayed.  So, we still
         * check the last replay LSN before reporting an error.
         */
        if (targetLSN <= GetXLogReplayRecPtr(NULL))
            return;
        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.")));
    }
    else
    {
        /* 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;
 
        /* Recheck that recovery is still in-progress */
        if (!RecoveryInProgress())
        {
            /*
             * Recovery was ended, but recheck if target LSN was already
             * replayed.
             */
            currentLSN = GetXLogReplayRecPtr(NULL);
            if (targetLSN <= currentLSN)
                return;
            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))));
        }
        else
        {
            /* Check if the waited LSN has been replayed */
            currentLSN = GetXLogReplayRecPtr(NULL);
            if (targetLSN <= currentLSN)
                break;
        }
 
        /*
         * 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))));
    }
}

References addLSNWaiter(), Assert, CHECK_FOR_INTERRUPTS, deleteLSNWaiter(), ereport, errcode(), errdetail(), errhint(), errmsg(), ERROR, GetCurrentTimestamp(), GetXLogReplayRecPtr(), LSN_FORMAT_ARGS, MaxBackends, MyLatch, MyProcNumber, RecoveryInProgress(), ResetLatch(), TimestampDifferenceMilliseconds(), TimestampTzPlusMilliseconds, WaitLatch(), waitLSNState, WL_EXIT_ON_PM_DEATH, WL_LATCH_SET, and WL_TIMEOUT.

Referenced by pg_wal_replay_wait().

waitlsn_cmp()

static int waitlsn_cmp ( const pairingheap_node *  a, const pairingheap_node *  b, void *  arg  )

static

Definition at line 72 of file waitlsn.c.

{
    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;
}

References a, b, pairingheap_const_container, and WaitLSNProcInfo::waitLSN.

Referenced by WaitLSNShmemInit().

WaitLSNCleanup()

void WaitLSNCleanup

(

void

)

Definition at line 204 of file waitlsn.c.

{
    /*
     * 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();
}

References deleteLSNWaiter(), WaitLSNProcInfo::inHeap, MyProcNumber, WaitLSNState::procInfos, and waitLSNState.

Referenced by AbortTransaction(), and ProcKill().

WaitLSNSetLatches()

void WaitLSNSetLatches

(

XLogRecPtr

currentLSN

)

Definition at line 155 of file waitlsn.c.

{
    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);
}

References i, WaitLSNProcInfo::inHeap, WaitLSNProcInfo::latch, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MaxBackends, pairingheap_container, pairingheap_first(), pairingheap_is_empty, pairingheap_remove_first(), palloc(), pfree(), SetLatch(), updateMinWaitedLSN(), WaitLSNState::waitersHeap, WaitLSNProcInfo::waitLSN, waitLSNState, and XLogRecPtrIsInvalid.

Referenced by PerformWalRecovery(), and StartupXLOG().

WaitLSNShmemInit()

void WaitLSNShmemInit

(

void

)

Definition at line 52 of file waitlsn.c.

{
    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));
    }
}

References MaxBackends, WaitLSNState::minWaitedLSN, pairingheap_initialize(), pg_atomic_init_u64(), PG_UINT64_MAX, WaitLSNState::procInfos, ShmemInitStruct(), WaitLSNState::waitersHeap, waitlsn_cmp(), WaitLSNShmemSize(), and waitLSNState.

Referenced by CreateOrAttachShmemStructs().

WaitLSNShmemSize()

Size WaitLSNShmemSize

(

void

)

Definition at line 41 of file waitlsn.c.

{
    Size        size;
 
    size = offsetof(WaitLSNState, procInfos);
    size = add_size(size, mul_size(MaxBackends, sizeof(WaitLSNProcInfo)));
    return size;
}

References add_size(), MaxBackends, mul_size(), WaitLSNState::procInfos, and size.

Referenced by CalculateShmemSize(), and WaitLSNShmemInit().

Variable Documentation

waitLSNState

struct WaitLSNState* waitLSNState = NULL

Definition at line 37 of file waitlsn.c.

Referenced by addLSNWaiter(), deleteLSNWaiter(), PerformWalRecovery(), updateMinWaitedLSN(), WaitForLSNReplay(), WaitLSNCleanup(), WaitLSNSetLatches(), and WaitLSNShmemInit().