xlogwait.c File Reference

#include "postgres.h"
#include <float.h>
#include <math.h>
#include "access/xlog.h"
#include "access/xlogrecovery.h"
#include "access/xlogwait.h"
#include "miscadmin.h"
#include "pgstat.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 dependency graph for xlogwait.c:

Go to the source code of this file.

Macros
#define	WAKEUP_PROC_STATIC_ARRAY_SIZE   (16)

Functions
static int	waitlsn_cmp (const pairingheap_node *a, const pairingheap_node *b, void *arg)
Size	WaitLSNShmemSize (void)
void	WaitLSNShmemInit (void)
static void	updateMinWaitedLSN (WaitLSNType lsnType)
static void	addLSNWaiter (XLogRecPtr lsn, WaitLSNType lsnType)
static void	deleteLSNWaiter (WaitLSNType lsnType)
static void	wakeupWaiters (WaitLSNType lsnType, XLogRecPtr currentLSN)
void	WaitLSNWakeup (WaitLSNType lsnType, XLogRecPtr currentLSN)
void	WaitLSNCleanup (void)
WaitLSNResult	WaitForLSN (WaitLSNType lsnType, XLogRecPtr targetLSN, int64 timeout)

Variables
struct WaitLSNState *	waitLSNState = NULL

Macro Definition Documentation

WAKEUP_PROC_STATIC_ARRAY_SIZE

#define WAKEUP_PROC_STATIC_ARRAY_SIZE   (16)

Definition at line 195 of file xlogwait.c.

Function Documentation

addLSNWaiter()

static void addLSNWaiter ( XLogRecPtr  lsn, WaitLSNType  lsnType  )

static

Definition at line 145 of file xlogwait.c.

{
    WaitLSNProcInfo *procInfo = &waitLSNState->procInfos[MyProcNumber];
    int         i = (int) lsnType;
 
    Assert(i >= 0 && i < (int) WAIT_LSN_TYPE_COUNT);
 
    LWLockAcquire(WaitLSNLock, LW_EXCLUSIVE);
 
    procInfo->procno = MyProcNumber;
    procInfo->waitLSN = lsn;
    procInfo->lsnType = lsnType;
 
    Assert(!procInfo->inHeap);
    pairingheap_add(&waitLSNState->waitersHeap[i], &procInfo->heapNode);
    procInfo->inHeap = true;
    updateMinWaitedLSN(lsnType);
 
    LWLockRelease(WaitLSNLock);
}

References Assert(), WaitLSNProcInfo::heapNode, i, WaitLSNProcInfo::inHeap, WaitLSNProcInfo::lsnType, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MyProcNumber, pairingheap_add(), WaitLSNState::procInfos, WaitLSNProcInfo::procno, updateMinWaitedLSN(), WAIT_LSN_TYPE_COUNT, WaitLSNState::waitersHeap, WaitLSNProcInfo::waitLSN, and waitLSNState.

Referenced by WaitForLSN().

deleteLSNWaiter()

static void deleteLSNWaiter ( WaitLSNType  lsnType )

static

Definition at line 170 of file xlogwait.c.

{
    WaitLSNProcInfo *procInfo = &waitLSNState->procInfos[MyProcNumber];
    int         i = (int) lsnType;
 
    Assert(i >= 0 && i < (int) WAIT_LSN_TYPE_COUNT);
 
    LWLockAcquire(WaitLSNLock, LW_EXCLUSIVE);
 
    Assert(procInfo->lsnType == lsnType);
 
    if (procInfo->inHeap)
    {
        pairingheap_remove(&waitLSNState->waitersHeap[i], &procInfo->heapNode);
        procInfo->inHeap = false;
        updateMinWaitedLSN(lsnType);
    }
 
    LWLockRelease(WaitLSNLock);
}

References Assert(), WaitLSNProcInfo::heapNode, i, WaitLSNProcInfo::inHeap, WaitLSNProcInfo::lsnType, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MyProcNumber, pairingheap_remove(), WaitLSNState::procInfos, updateMinWaitedLSN(), WAIT_LSN_TYPE_COUNT, WaitLSNState::waitersHeap, and waitLSNState.

Referenced by WaitForLSN(), and WaitLSNCleanup().

updateMinWaitedLSN()

static void updateMinWaitedLSN ( WaitLSNType  lsnType )

static

Definition at line 124 of file xlogwait.c.

{
    XLogRecPtr  minWaitedLSN = PG_UINT64_MAX;
    int         i = (int) lsnType;
 
    Assert(i >= 0 && i < (int) WAIT_LSN_TYPE_COUNT);
 
    if (!pairingheap_is_empty(&waitLSNState->waitersHeap[i]))
    {
        pairingheap_node *node = pairingheap_first(&waitLSNState->waitersHeap[i]);
        WaitLSNProcInfo *procInfo = pairingheap_container(WaitLSNProcInfo, heapNode, node);
 
        minWaitedLSN = procInfo->waitLSN;
    }
    pg_atomic_write_u64(&waitLSNState->minWaitedLSN[i], minWaitedLSN);
}

References Assert(), i, WaitLSNState::minWaitedLSN, pairingheap_container, pairingheap_first(), pairingheap_is_empty, pg_atomic_write_u64(), PG_UINT64_MAX, WAIT_LSN_TYPE_COUNT, WaitLSNState::waitersHeap, WaitLSNProcInfo::waitLSN, and waitLSNState.

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

WaitForLSN()

WaitLSNResult WaitForLSN	(	WaitLSNType	lsnType,
		XLogRecPtr	targetLSN,
		int64	timeout
	)

Definition at line 314 of file xlogwait.c.

{
    XLogRecPtr  currentLSN;
    TimestampTz endtime = 0;
    int         wake_events = WL_LATCH_SET | WL_POSTMASTER_DEATH;
 
    /* Shouldn't be called when shmem isn't initialized */
    Assert(waitLSNState);
 
    /* Should have a valid proc number */
    Assert(MyProcNumber >= 0 && MyProcNumber < MaxBackends + NUM_AUXILIARY_PROCS);
 
    if (timeout > 0)
    {
        endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), timeout);
        wake_events |= WL_TIMEOUT;
    }
 
    /*
     * Add our process to the waiters heap.  It might happen that target LSN
     * gets reached before we do.  The check at the beginning of the loop
     * below prevents the race condition.
     */
    addLSNWaiter(targetLSN, lsnType);
 
    for (;;)
    {
        int         rc;
        long        delay_ms = -1;
 
        if (lsnType == WAIT_LSN_TYPE_REPLAY)
            currentLSN = GetXLogReplayRecPtr(NULL);
        else
            currentLSN = GetFlushRecPtr(NULL);
 
        /* Check that recovery is still in-progress */
        if (lsnType == WAIT_LSN_TYPE_REPLAY && !RecoveryInProgress())
        {
            /*
             * Recovery was ended, but check if target LSN was already
             * reached.
             */
            deleteLSNWaiter(lsnType);
 
            if (PromoteIsTriggered() && targetLSN <= currentLSN)
                return WAIT_LSN_RESULT_SUCCESS;
            return WAIT_LSN_RESULT_NOT_IN_RECOVERY;
        }
        else
        {
            /* Check if the waited LSN has been reached */
            if (targetLSN <= currentLSN)
                break;
        }
 
        if (timeout > 0)
        {
            delay_ms = TimestampDifferenceMilliseconds(GetCurrentTimestamp(), endtime);
            if (delay_ms <= 0)
                break;
        }
 
        CHECK_FOR_INTERRUPTS();
 
        rc = WaitLatch(MyLatch, wake_events, delay_ms,
                       (lsnType == WAIT_LSN_TYPE_REPLAY) ? WAIT_EVENT_WAIT_FOR_WAL_REPLAY : WAIT_EVENT_WAIT_FOR_WAL_FLUSH);
 
        /*
         * Emergency bailout if postmaster has died.  This is to avoid the
         * necessity for manual cleanup of all postmaster children.
         */
        if (rc & WL_POSTMASTER_DEATH)
            ereport(FATAL,
                    errcode(ERRCODE_ADMIN_SHUTDOWN),
                    errmsg("terminating connection due to unexpected postmaster exit"),
                    errcontext("while waiting for LSN"));
 
        if (rc & WL_LATCH_SET)
            ResetLatch(MyLatch);
    }
 
    /*
     * Delete our process from the shared memory heap.  We might already be
     * deleted by the startup process.  The 'inHeap' flags prevents us from
     * the double deletion.
     */
    deleteLSNWaiter(lsnType);
 
    /*
     * If we didn't reach the target LSN, we must be exited by timeout.
     */
    if (targetLSN > currentLSN)
        return WAIT_LSN_RESULT_TIMEOUT;
 
    return WAIT_LSN_RESULT_SUCCESS;
}

References addLSNWaiter(), Assert(), CHECK_FOR_INTERRUPTS, deleteLSNWaiter(), ereport, errcode(), errcontext, errmsg(), FATAL, GetCurrentTimestamp(), GetFlushRecPtr(), GetXLogReplayRecPtr(), MaxBackends, MyLatch, MyProcNumber, NUM_AUXILIARY_PROCS, PromoteIsTriggered(), RecoveryInProgress(), ResetLatch(), TimestampDifferenceMilliseconds(), TimestampTzPlusMilliseconds, WAIT_LSN_RESULT_NOT_IN_RECOVERY, WAIT_LSN_RESULT_SUCCESS, WAIT_LSN_RESULT_TIMEOUT, WAIT_LSN_TYPE_REPLAY, WaitLatch(), waitLSNState, WL_LATCH_SET, WL_POSTMASTER_DEATH, and WL_TIMEOUT.

Referenced by ExecWaitStmt().

waitlsn_cmp()

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

static

Definition at line 107 of file xlogwait.c.

{
    const WaitLSNProcInfo *aproc = pairingheap_const_container(WaitLSNProcInfo, heapNode, a);
    const WaitLSNProcInfo *bproc = pairingheap_const_container(WaitLSNProcInfo, heapNode, 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 290 of file xlogwait.c.

{
    if (waitLSNState)
    {
        /*
         * 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(waitLSNState->procInfos[MyProcNumber].lsnType);
    }
}

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

Referenced by AbortTransaction(), and ProcKill().

WaitLSNShmemInit()

void WaitLSNShmemInit

(

void

)

Definition at line 78 of file xlogwait.c.

{
    bool        found;
 
    waitLSNState = (WaitLSNState *) ShmemInitStruct("WaitLSNState",
                                                    WaitLSNShmemSize(),
                                                    &found);
    if (!found)
    {
        int         i;
 
        /* Initialize heaps and tracking */
        for (i = 0; i < WAIT_LSN_TYPE_COUNT; i++)
        {
            pg_atomic_init_u64(&waitLSNState->minWaitedLSN[i], PG_UINT64_MAX);
            pairingheap_initialize(&waitLSNState->waitersHeap[i], waitlsn_cmp, NULL);
        }
 
        /* Initialize process info array */
        memset(&waitLSNState->procInfos, 0,
               (MaxBackends + NUM_AUXILIARY_PROCS) * sizeof(WaitLSNProcInfo));
    }
}

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

Referenced by CreateOrAttachShmemStructs().

WaitLSNShmemSize()

Size WaitLSNShmemSize

(

void

)

Definition at line 67 of file xlogwait.c.

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

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

Referenced by CalculateShmemSize(), and WaitLSNShmemInit().

WaitLSNWakeup()

void WaitLSNWakeup	(	WaitLSNType	lsnType,
		XLogRecPtr	currentLSN
	)

Definition at line 269 of file xlogwait.c.

{
    int         i = (int) lsnType;
 
    Assert(i >= 0 && i < (int) WAIT_LSN_TYPE_COUNT);
 
    /*
     * Fast path check.  Skip if currentLSN is InvalidXLogRecPtr, which means
     * "wake all waiters" (e.g., during promotion when recovery ends).
     */
    if (XLogRecPtrIsValid(currentLSN) &&
        pg_atomic_read_u64(&waitLSNState->minWaitedLSN[i]) > currentLSN)
        return;
 
    wakeupWaiters(lsnType, currentLSN);
}

References Assert(), i, WaitLSNState::minWaitedLSN, pg_atomic_read_u64(), WAIT_LSN_TYPE_COUNT, waitLSNState, wakeupWaiters(), and XLogRecPtrIsValid.

Referenced by PerformWalRecovery(), and StartupXLOG().

wakeupWaiters()

static void wakeupWaiters ( WaitLSNType  lsnType, XLogRecPtr  currentLSN  )

static

Definition at line 210 of file xlogwait.c.

{
    ProcNumber  wakeUpProcs[WAKEUP_PROC_STATIC_ARRAY_SIZE];
    int         numWakeUpProcs;
    int         i = (int) lsnType;
 
    Assert(i >= 0 && i < (int) WAIT_LSN_TYPE_COUNT);
 
    do
    {
        numWakeUpProcs = 0;
        LWLockAcquire(WaitLSNLock, LW_EXCLUSIVE);
 
        /*
         * Iterate the waiters heap until we find LSN not yet reached. Record
         * process numbers to wake up, but send wakeups after releasing lock.
         */
        while (!pairingheap_is_empty(&waitLSNState->waitersHeap[i]))
        {
            pairingheap_node *node = pairingheap_first(&waitLSNState->waitersHeap[i]);
            WaitLSNProcInfo *procInfo;
 
            /* Get procInfo using appropriate heap node */
            procInfo = pairingheap_container(WaitLSNProcInfo, heapNode, node);
 
            if (XLogRecPtrIsValid(currentLSN) && procInfo->waitLSN > currentLSN)
                break;
 
            Assert(numWakeUpProcs < WAKEUP_PROC_STATIC_ARRAY_SIZE);
            wakeUpProcs[numWakeUpProcs++] = procInfo->procno;
            (void) pairingheap_remove_first(&waitLSNState->waitersHeap[i]);
 
            /* Update appropriate flag */
            procInfo->inHeap = false;
 
            if (numWakeUpProcs == WAKEUP_PROC_STATIC_ARRAY_SIZE)
                break;
        }
 
        updateMinWaitedLSN(lsnType);
        LWLockRelease(WaitLSNLock);
 
        /*
         * Set latches for processes whose waited LSNs have been reached.
         * Since SetLatch() is a time-consuming operation, we do this outside
         * of WaitLSNLock. This is safe because procLatch is never freed, so
         * at worst we may set a latch for the wrong process or for no process
         * at all, which is harmless.
         */
        for (i = 0; i < numWakeUpProcs; i++)
            SetLatch(&GetPGProcByNumber(wakeUpProcs[i])->procLatch);
 
    } while (numWakeUpProcs == WAKEUP_PROC_STATIC_ARRAY_SIZE);
}

References Assert(), GetPGProcByNumber, i, WaitLSNProcInfo::inHeap, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), pairingheap_container, pairingheap_first(), pairingheap_is_empty, pairingheap_remove_first(), WaitLSNProcInfo::procno, SetLatch(), updateMinWaitedLSN(), WAIT_LSN_TYPE_COUNT, WaitLSNState::waitersHeap, WaitLSNProcInfo::waitLSN, waitLSNState, WAKEUP_PROC_STATIC_ARRAY_SIZE, and XLogRecPtrIsValid.

Referenced by WaitLSNWakeup().

Variable Documentation

waitLSNState

struct WaitLSNState* waitLSNState = NULL

Definition at line 63 of file xlogwait.c.

Referenced by addLSNWaiter(), deleteLSNWaiter(), PerformWalRecovery(), updateMinWaitedLSN(), WaitForLSN(), WaitLSNCleanup(), WaitLSNShmemInit(), WaitLSNWakeup(), and wakeupWaiters().