xlogwait.c File Reference

#include "postgres.h"
#include <float.h>
#include "access/xlog.h"
#include "access/xlogrecovery.h"
#include "access/xlogwait.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "replication/walreceiver.h"
#include "storage/latch.h"
#include "storage/proc.h"
#include "storage/shmem.h"
#include "storage/subsystems.h"
#include "utils/fmgrprotos.h"
#include "utils/pg_lsn.h"
#include "utils/snapmgr.h"
#include "utils/wait_event.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)
static void	WaitLSNShmemRequest (void *arg)
static void	WaitLSNShmemInit (void *arg)
	StaticAssertDecl (lengthof(WaitLSNWaitEvents)==WAIT_LSN_TYPE_COUNT, "WaitLSNWaitEvents must match WaitLSNType enum")
XLogRecPtr	GetCurrentLSNForWaitType (WaitLSNType lsnType)
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)
static bool	WaitLSNTypeRequiresRecovery (WaitLSNType t)
WaitLSNResult	WaitForLSN (WaitLSNType lsnType, XLogRecPtr targetLSN, int64 timeout)

Variables
struct WaitLSNState *	waitLSNState = NULL
const ShmemCallbacks	WaitLSNShmemCallbacks
static const uint32	WaitLSNWaitEvents []

Macro Definition Documentation

WAKEUP_PROC_STATIC_ARRAY_SIZE

#define WAKEUP_PROC_STATIC_ARRAY_SIZE   (16)

Definition at line 244 of file xlogwait.c.

Function Documentation

addLSNWaiter()

static void addLSNWaiter ( XLogRecPtr  lsn, WaitLSNType  lsnType  )

static

Definition at line 194 of file xlogwait.c.

{
    WaitLSNProcInfo *procInfo = &waitLSNState->procInfos[MyProcNumber];
    int         i = (int) lsnType;
 
    Assert(i >= 0 && i < 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, fb(), i, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MyProcNumber, pairingheap_add(), WaitLSNState::procInfos, updateMinWaitedLSN(), WAIT_LSN_TYPE_COUNT, WaitLSNState::waitersHeap, and waitLSNState.

Referenced by WaitForLSN().

deleteLSNWaiter()

static void deleteLSNWaiter ( WaitLSNType  lsnType )

static

Definition at line 219 of file xlogwait.c.

{
    WaitLSNProcInfo *procInfo = &waitLSNState->procInfos[MyProcNumber];
    int         i = (int) lsnType;
 
    Assert(i >= 0 && i < 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, fb(), i, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MyProcNumber, pairingheap_remove(), WaitLSNState::procInfos, updateMinWaitedLSN(), WAIT_LSN_TYPE_COUNT, WaitLSNState::waitersHeap, and waitLSNState.

Referenced by WaitForLSN(), and WaitLSNCleanup().

GetCurrentLSNForWaitType()

XLogRecPtr GetCurrentLSNForWaitType

(

WaitLSNType

lsnType

)

Definition at line 98 of file xlogwait.c.

{
    Assert(lsnType >= 0 && lsnType < WAIT_LSN_TYPE_COUNT);
 
    switch (lsnType)
    {
        case WAIT_LSN_TYPE_STANDBY_REPLAY:
            return GetXLogReplayRecPtr(NULL);
 
        case WAIT_LSN_TYPE_STANDBY_WRITE:
            return GetWalRcvWriteRecPtr();
 
        case WAIT_LSN_TYPE_STANDBY_FLUSH:
            return GetWalRcvFlushRecPtr(NULL, NULL);
 
        case WAIT_LSN_TYPE_PRIMARY_FLUSH:
            return GetFlushRecPtr(NULL);
    }
 
    elog(ERROR, "invalid LSN wait type: %d", lsnType);
    pg_unreachable();
}

References Assert, elog, ERROR, fb(), GetFlushRecPtr(), GetWalRcvFlushRecPtr(), GetWalRcvWriteRecPtr(), GetXLogReplayRecPtr(), pg_unreachable, WAIT_LSN_TYPE_COUNT, WAIT_LSN_TYPE_PRIMARY_FLUSH, WAIT_LSN_TYPE_STANDBY_FLUSH, WAIT_LSN_TYPE_STANDBY_REPLAY, and WAIT_LSN_TYPE_STANDBY_WRITE.

Referenced by ExecWaitStmt(), and WaitForLSN().

StaticAssertDecl()

StaticAssertDecl	(	lengthof(WaitLSNWaitEvents)	= =WAIT_LSN_TYPE_COUNT,
		"WaitLSNWaitEvents must match WaitLSNType enum"
	)

updateMinWaitedLSN()

static void updateMinWaitedLSN ( WaitLSNType  lsnType )

static

Definition at line 173 of file xlogwait.c.

{
    XLogRecPtr  minWaitedLSN = PG_UINT64_MAX;
    int         i = (int) lsnType;
 
    Assert(i >= 0 && i < 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, fb(), i, WaitLSNState::minWaitedLSN, pairingheap_container, pairingheap_first(), pairingheap_is_empty, pg_atomic_write_u64(), PG_UINT64_MAX, WAIT_LSN_TYPE_COUNT, WaitLSNState::waitersHeap, and waitLSNState.

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

WaitForLSN()

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

Definition at line 378 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;
 
        /* Get current LSN for the wait type */
        currentLSN = GetCurrentLSNForWaitType(lsnType);
 
        /* Check that recovery is still in-progress */
        if (WaitLSNTypeRequiresRecovery(lsnType) && !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,
                       WaitLSNWaitEvents[lsnType]);
 
        /*
         * 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, fb(), GetCurrentLSNForWaitType(), GetCurrentTimestamp(), 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, WaitLatch(), waitLSNState, WaitLSNTypeRequiresRecovery(), WaitLSNWaitEvents, WL_LATCH_SET, WL_POSTMASTER_DEATH, and WL_TIMEOUT.

Referenced by decode_concurrent_changes(), and ExecWaitStmt().

waitlsn_cmp()

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

static

Definition at line 156 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, fb(), and pairingheap_const_container.

Referenced by WaitLSNShmemInit().

WaitLSNCleanup()

void WaitLSNCleanup

(

void

)

Definition at line 341 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()

static void WaitLSNShmemInit ( void *  arg )

static

Definition at line 137 of file xlogwait.c.

{
    /* Initialize heaps and tracking */
    for (int 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 fb(), i, MaxBackends, WaitLSNState::minWaitedLSN, NUM_AUXILIARY_PROCS, pairingheap_initialize(), pg_atomic_init_u64(), PG_UINT64_MAX, WaitLSNState::procInfos, WAIT_LSN_TYPE_COUNT, WaitLSNState::waitersHeap, waitlsn_cmp(), and waitLSNState.

WaitLSNShmemRequest()

static void WaitLSNShmemRequest ( void *  arg )

static

Definition at line 123 of file xlogwait.c.

{
    Size        size;
 
    size = offsetof(WaitLSNState, procInfos);
    size = add_size(size, mul_size(MaxBackends + NUM_AUXILIARY_PROCS, sizeof(WaitLSNProcInfo)));
    ShmemRequestStruct(.name = "WaitLSNState",
                       .size = size,
                       .ptr = (void **) &waitLSNState,
        );
}

References add_size(), fb(), MaxBackends, mul_size(), name, NUM_AUXILIARY_PROCS, WaitLSNState::procInfos, ShmemRequestStruct, and waitLSNState.

WaitLSNTypeRequiresRecovery()

static bool WaitLSNTypeRequiresRecovery ( WaitLSNType  t )

inlinestatic

Definition at line 362 of file xlogwait.c.

{
    return t == WAIT_LSN_TYPE_STANDBY_REPLAY ||
        t == WAIT_LSN_TYPE_STANDBY_WRITE ||
        t == WAIT_LSN_TYPE_STANDBY_FLUSH;
}

References WAIT_LSN_TYPE_STANDBY_FLUSH, WAIT_LSN_TYPE_STANDBY_REPLAY, and WAIT_LSN_TYPE_STANDBY_WRITE.

Referenced by WaitForLSN().

WaitLSNWakeup()

void WaitLSNWakeup	(	WaitLSNType	lsnType,
		XLogRecPtr	currentLSN
	)

Definition at line 320 of file xlogwait.c.

{
    int         i = (int) lsnType;
 
    Assert(i >= 0 && i < 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, fb(), i, WaitLSNState::minWaitedLSN, pg_atomic_read_u64(), WAIT_LSN_TYPE_COUNT, waitLSNState, wakeupWaiters(), and XLogRecPtrIsValid.

Referenced by PerformWalRecovery(), StartupXLOG(), XLogBackgroundFlush(), XLogFlush(), XLogWalRcvFlush(), and XLogWalRcvWrite().

wakeupWaiters()

static void wakeupWaiters ( WaitLSNType  lsnType, XLogRecPtr  currentLSN  )

static

Definition at line 259 of file xlogwait.c.

{
    ProcNumber  wakeUpProcs[WAKEUP_PROC_STATIC_ARRAY_SIZE];
    int         numWakeUpProcs;
    int         i = (int) lsnType;
 
    Assert(i >= 0 && i < WAIT_LSN_TYPE_COUNT);
 
    do
    {
        int         j;
 
        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 (j = 0; j < numWakeUpProcs; j++)
            SetLatch(&GetPGProcByNumber(wakeUpProcs[j])->procLatch);
 
    } while (numWakeUpProcs == WAKEUP_PROC_STATIC_ARRAY_SIZE);
}

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

Referenced by WaitLSNWakeup().

Variable Documentation

WaitLSNShmemCallbacks

const ShmemCallbacks WaitLSNShmemCallbacks

Initial value:

= {
    .request_fn = WaitLSNShmemRequest,
    .init_fn = WaitLSNShmemInit,
}

Definition at line 75 of file xlogwait.c.

                                             {
    .request_fn = WaitLSNShmemRequest,
    .init_fn = WaitLSNShmemInit,
};

waitLSNState

struct WaitLSNState* waitLSNState = NULL

Definition at line 70 of file xlogwait.c.

Referenced by addLSNWaiter(), deleteLSNWaiter(), PerformWalRecovery(), updateMinWaitedLSN(), WaitForLSN(), WaitLSNCleanup(), WaitLSNShmemInit(), WaitLSNShmemRequest(), WaitLSNWakeup(), wakeupWaiters(), XLogBackgroundFlush(), XLogFlush(), XLogWalRcvFlush(), and XLogWalRcvWrite().

WaitLSNWaitEvents

const uint32 WaitLSNWaitEvents[]

static

Initial value:

= {
    [WAIT_LSN_TYPE_STANDBY_REPLAY] = WAIT_EVENT_WAIT_FOR_WAL_REPLAY,
    [WAIT_LSN_TYPE_STANDBY_WRITE] = WAIT_EVENT_WAIT_FOR_WAL_WRITE,
    [WAIT_LSN_TYPE_STANDBY_FLUSH] = WAIT_EVENT_WAIT_FOR_WAL_FLUSH,
    [WAIT_LSN_TYPE_PRIMARY_FLUSH] = WAIT_EVENT_WAIT_FOR_WAL_FLUSH,
}

Definition at line 84 of file xlogwait.c.

                                          {
    [WAIT_LSN_TYPE_STANDBY_REPLAY] = WAIT_EVENT_WAIT_FOR_WAL_REPLAY,
    [WAIT_LSN_TYPE_STANDBY_WRITE] = WAIT_EVENT_WAIT_FOR_WAL_WRITE,
    [WAIT_LSN_TYPE_STANDBY_FLUSH] = WAIT_EVENT_WAIT_FOR_WAL_FLUSH,
    [WAIT_LSN_TYPE_PRIMARY_FLUSH] = WAIT_EVENT_WAIT_FOR_WAL_FLUSH,
};

Referenced by WaitForLSN().