#include "lib/pairingheap.h"
#include "postgres.h"
#include "port/atomics.h"
#include "storage/latch.h"
#include "storage/spin.h"
#include "tcop/dest.h"

Include dependency graph for waitlsn.h:

This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Data Structures
struct	WaitLSNProcInfo

struct	WaitLSNState

Typedefs
typedef struct WaitLSNProcInfo	WaitLSNProcInfo

typedef struct WaitLSNState	WaitLSNState

Functions
Size	WaitLSNShmemSize (void)

void	WaitLSNShmemInit (void)

void	WaitLSNSetLatches (XLogRecPtr currentLSN)

void	WaitLSNCleanup (void)

void	WaitForLSNReplay (XLogRecPtr targetLSN, int64 timeout)

Variables
PGDLLIMPORT WaitLSNState *	waitLSNState

Typedef Documentation

◆ WaitLSNProcInfo

typedef struct WaitLSNProcInfo WaitLSNProcInfo

◆ WaitLSNState

typedef struct WaitLSNState WaitLSNState

Function Documentation

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

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