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 268 of file xlogwait.c.

Function Documentation

addLSNWaiter()

static void addLSNWaiter ( XLogRecPtr  lsn, WaitLSNType  lsnType  )

static

Definition at line 218 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 243 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 99 of file xlogwait.c.

{
    Assert(lsnType >= 0 && lsnType < WAIT_LSN_TYPE_COUNT);
 
    /*
     * All of the cases below provide memory barrier semantics:
     * GetWalRcvWriteRecPtr() and GetFlushRecPtr() have explicit barriers,
     * while GetXLogReplayRecPtr() and GetWalRcvFlushRecPtr() use spinlocks.
     */
    switch (lsnType)
    {
        case WAIT_LSN_TYPE_STANDBY_REPLAY:
            return GetXLogReplayRecPtr(NULL);
 
        case WAIT_LSN_TYPE_STANDBY_WRITE:
            {
                XLogRecPtr  recptr = GetWalRcvWriteRecPtr();
                XLogRecPtr  replay = GetXLogReplayRecPtr(NULL);
 
                /*
                 * Use the replay position as a floor.  WAL up to the replay
                 * point is already on disk from a base backup, archive
                 * restore, or prior streaming, so there is no reason to wait
                 * for the walreceiver to re-receive it.
                 */
                return Max(recptr, replay);
            }
 
        case WAIT_LSN_TYPE_STANDBY_FLUSH:
            {
                XLogRecPtr  recptr = GetWalRcvFlushRecPtr(NULL, NULL);
                XLogRecPtr  replay = GetXLogReplayRecPtr(NULL);
 
                /* Same floor as standby_write; see comment above. */
                return Max(recptr, replay);
            }
 
        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(), Max, 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 196 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;
    }
    /* Pairs with pg_atomic_read_membarrier_u64() in WaitLSNWakeup(). */
    pg_atomic_write_membarrier_u64(&waitLSNState->minWaitedLSN[i], minWaitedLSN);
}

References Assert, fb(), i, WaitLSNState::minWaitedLSN, pairingheap_container, pairingheap_first(), pairingheap_is_empty, pg_atomic_write_membarrier_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 403 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"));
 
        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 179 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 366 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 AbortSubTransaction(), AbortTransaction(), and ProcKill().

WaitLSNShmemInit()

static void WaitLSNShmemInit ( void *  arg )

static

Definition at line 160 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 146 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 387 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 344 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). Pairs
     * with pg_atomic_write_membarrier_u64() in updateMinWaitedLSN().
     */
    if (XLogRecPtrIsValid(currentLSN) &&
        pg_atomic_read_membarrier_u64(&waitLSNState->minWaitedLSN[i]) > currentLSN)
        return;
 
    wakeupWaiters(lsnType, currentLSN);
}

References Assert, fb(), i, WaitLSNState::minWaitedLSN, pg_atomic_read_membarrier_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 283 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(), updateMinWaitedLSN(), WaitForLSN(), WaitLSNCleanup(), WaitLSNShmemInit(), WaitLSNShmemRequest(), WaitLSNWakeup(), and wakeupWaiters().

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