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)
void	WaitForLSNReplay (XLogRecPtr targetLSN, int64 timeout)

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().

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()

void WaitForLSNReplay	(	XLogRecPtr	targetLSN,
		int64	timeout
	)

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
         * reached.
         */
        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 reach 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().