Branch data Line data Source code
1 : : /*-------------------------------------------------------------------------
2 : : *
3 : : * xlogwait.c
4 : : * Implements waiting for WAL operations to reach specific LSNs.
5 : : *
6 : : * Copyright (c) 2025-2026, PostgreSQL Global Development Group
7 : : *
8 : : * IDENTIFICATION
9 : : * src/backend/access/transam/xlogwait.c
10 : : *
11 : : * NOTES
12 : : * This file implements waiting for WAL operations to reach specific LSNs
13 : : * on both physical standby and primary servers. The core idea is simple:
14 : : * every process that wants to wait publishes the LSN it needs to the
15 : : * shared memory, and the appropriate process (startup on standby,
16 : : * walreceiver on standby, or WAL writer/backend on primary) wakes it
17 : : * once that LSN has been reached.
18 : : *
19 : : * The shared memory used by this module comprises a procInfos
20 : : * per-backend array with the information of the awaited LSN for each
21 : : * of the backend processes. The elements of that array are organized
22 : : * into pairing heaps (waitersHeap), one for each WaitLSNType, which
23 : : * allows for very fast finding of the least awaited LSN for each type.
24 : : *
25 : : * In addition, the least-awaited LSN for each type is cached in the
26 : : * minWaitedLSN array. The waiter process publishes information about
27 : : * itself to the shared memory and waits on the latch until it is woken
28 : : * up by the appropriate process, standby is promoted, or the postmaster
29 : : * dies. Then, it cleans information about itself in the shared memory.
30 : : *
31 : : * On standby servers:
32 : : * - After replaying a WAL record, the startup process performs a fast
33 : : * path check minWaitedLSN[REPLAY] > replayLSN. If this check is
34 : : * negative, it checks waitersHeap[REPLAY] and wakes up the backends
35 : : * whose awaited LSNs are reached.
36 : : * - After receiving WAL, the walreceiver process performs similar checks
37 : : * against the flush and write LSNs, waking up waiters in the FLUSH
38 : : * and WRITE heaps, respectively.
39 : : *
40 : : * On primary servers: After flushing WAL, the WAL writer or backend
41 : : * process performs a similar check against the flush LSN and wakes up
42 : : * waiters whose target flush LSNs have been reached.
43 : : *
44 : : *-------------------------------------------------------------------------
45 : : */
46 : :
47 : : #include "postgres.h"
48 : :
49 : : #include <float.h>
50 : :
51 : : #include "access/xlog.h"
52 : : #include "access/xlogrecovery.h"
53 : : #include "access/xlogwait.h"
54 : : #include "miscadmin.h"
55 : : #include "pgstat.h"
56 : : #include "replication/walreceiver.h"
57 : : #include "storage/ipc.h"
58 : : #include "storage/latch.h"
59 : : #include "storage/proc.h"
60 : : #include "storage/shmem.h"
61 : : #include "storage/subsystems.h"
62 : : #include "utils/fmgrprotos.h"
63 : : #include "utils/pg_lsn.h"
64 : : #include "utils/snapmgr.h"
65 : : #include "utils/wait_event.h"
66 : :
67 : :
68 : : static int waitlsn_cmp(const pairingheap_node *a, const pairingheap_node *b,
69 : : void *arg);
70 : :
71 : : struct WaitLSNState *waitLSNState = NULL;
72 : :
73 : : static bool waitLSNShmemExitRegistered = false;
74 : :
75 : : static void WaitLSNShmemRequest(void *arg);
76 : : static void WaitLSNShmemInit(void *arg);
77 : : static void WaitLSNShmemExit(int code, Datum arg);
78 : : static void RegisterWaitLSNShmemExit(void);
79 : :
80 : : const ShmemCallbacks WaitLSNShmemCallbacks = {
81 : : .request_fn = WaitLSNShmemRequest,
82 : : .init_fn = WaitLSNShmemInit,
83 : : };
84 : :
85 : : /*
86 : : * Wait event for each WaitLSNType, used with WaitLatch() to report
87 : : * the wait in pg_stat_activity.
88 : : */
89 : : static const uint32 WaitLSNWaitEvents[] = {
90 : : [WAIT_LSN_TYPE_STANDBY_REPLAY] = WAIT_EVENT_WAIT_FOR_WAL_REPLAY,
91 : : [WAIT_LSN_TYPE_STANDBY_WRITE] = WAIT_EVENT_WAIT_FOR_WAL_WRITE,
92 : : [WAIT_LSN_TYPE_STANDBY_FLUSH] = WAIT_EVENT_WAIT_FOR_WAL_FLUSH,
93 : : [WAIT_LSN_TYPE_PRIMARY_FLUSH] = WAIT_EVENT_WAIT_FOR_WAL_FLUSH,
94 : : };
95 : :
96 : : StaticAssertDecl(lengthof(WaitLSNWaitEvents) == WAIT_LSN_TYPE_COUNT,
97 : : "WaitLSNWaitEvents must match WaitLSNType enum");
98 : :
99 : : /*
100 : : * Get the current LSN for the specified wait type. Provide memory
101 : : * barrier semantics before getting the value.
102 : : */
103 : : XLogRecPtr
104 : 3586 : GetCurrentLSNForWaitType(WaitLSNType lsnType)
105 : : {
106 : : Assert(lsnType >= 0 && lsnType < WAIT_LSN_TYPE_COUNT);
107 : :
108 : : /*
109 : : * All of the cases below provide memory barrier semantics:
110 : : * GetWalRcvWriteRecPtr() and GetFlushRecPtr() have explicit barriers,
111 : : * while GetXLogReplayRecPtr() and GetWalRcvFlushRecPtr() use spinlocks.
112 : : */
113 [ + + + + : 3586 : switch (lsnType)
- ]
114 : : {
115 : 230 : case WAIT_LSN_TYPE_STANDBY_REPLAY:
116 : 230 : return GetXLogReplayRecPtr(NULL);
117 : :
118 : 52 : case WAIT_LSN_TYPE_STANDBY_WRITE:
119 : : {
120 : 52 : XLogRecPtr recptr = GetWalRcvWriteRecPtr();
121 : 52 : XLogRecPtr replay = GetXLogReplayRecPtr(NULL);
122 : :
123 : : /*
124 : : * Use the replay position as a floor. WAL up to the replay
125 : : * point is already on disk from a base backup, archive
126 : : * restore, or prior streaming, so there is no reason to wait
127 : : * for the walreceiver to re-receive it.
128 : : */
129 : 52 : return Max(recptr, replay);
130 : : }
131 : :
132 : 36 : case WAIT_LSN_TYPE_STANDBY_FLUSH:
133 : : {
134 : 36 : XLogRecPtr recptr = GetWalRcvFlushRecPtr(NULL, NULL);
135 : 36 : XLogRecPtr replay = GetXLogReplayRecPtr(NULL);
136 : :
137 : : /* Same floor as standby_write; see comment above. */
138 : 36 : return Max(recptr, replay);
139 : : }
140 : :
141 : 3268 : case WAIT_LSN_TYPE_PRIMARY_FLUSH:
142 : 3268 : return GetFlushRecPtr(NULL);
143 : : }
144 : :
145 [ # # ]: 0 : elog(ERROR, "invalid LSN wait type: %d", lsnType);
146 : : pg_unreachable();
147 : : }
148 : :
149 : : /* Register the shared memory space needed for WaitLSNState. */
150 : : static void
151 : 1283 : WaitLSNShmemRequest(void *arg)
152 : : {
153 : : Size size;
154 : :
155 : 1283 : size = offsetof(WaitLSNState, procInfos);
156 : 1283 : size = add_size(size, mul_size(MaxBackends + NUM_AUXILIARY_PROCS, sizeof(WaitLSNProcInfo)));
157 : 1283 : ShmemRequestStruct(.name = "WaitLSNState",
158 : : .size = size,
159 : : .ptr = (void **) &waitLSNState,
160 : : );
161 : 1283 : }
162 : :
163 : : /* Initialize the WaitLSNState in the shared memory. */
164 : : static void
165 : 1280 : WaitLSNShmemInit(void *arg)
166 : : {
167 : : /* Initialize heaps and tracking */
168 [ + + ]: 6400 : for (int i = 0; i < WAIT_LSN_TYPE_COUNT; i++)
169 : : {
170 : 5120 : pg_atomic_init_u64(&waitLSNState->minWaitedLSN[i], PG_UINT64_MAX);
171 : 5120 : pairingheap_initialize(&waitLSNState->waitersHeap[i], waitlsn_cmp, NULL);
172 : : }
173 : :
174 : : /* Initialize process info array */
175 : 1280 : memset(&waitLSNState->procInfos, 0,
176 : 1280 : (MaxBackends + NUM_AUXILIARY_PROCS) * sizeof(WaitLSNProcInfo));
177 : 1280 : }
178 : :
179 : : /*
180 : : * Comparison function for LSN waiters heaps. Waiting processes are ordered by
181 : : * LSN, so that the waiter with smallest LSN is at the top.
182 : : */
183 : : static int
184 : 28 : waitlsn_cmp(const pairingheap_node *a, const pairingheap_node *b, void *arg)
185 : : {
186 : 28 : const WaitLSNProcInfo *aproc = pairingheap_const_container(WaitLSNProcInfo, heapNode, a);
187 : 28 : const WaitLSNProcInfo *bproc = pairingheap_const_container(WaitLSNProcInfo, heapNode, b);
188 : :
189 [ + + ]: 28 : if (aproc->waitLSN < bproc->waitLSN)
190 : 15 : return 1;
191 [ + + ]: 13 : else if (aproc->waitLSN > bproc->waitLSN)
192 : 10 : return -1;
193 : : else
194 : 3 : return 0;
195 : : }
196 : :
197 : : /*
198 : : * Update minimum waited LSN for the specified LSN type
199 : : */
200 : : static void
201 : 10082 : updateMinWaitedLSN(WaitLSNType lsnType)
202 : : {
203 : 10082 : XLogRecPtr minWaitedLSN = PG_UINT64_MAX;
204 : 10082 : int i = (int) lsnType;
205 : :
206 : : Assert(i >= 0 && i < WAIT_LSN_TYPE_COUNT);
207 : :
208 [ + + ]: 10082 : if (!pairingheap_is_empty(&waitLSNState->waitersHeap[i]))
209 : : {
210 : 3491 : pairingheap_node *node = pairingheap_first(&waitLSNState->waitersHeap[i]);
211 : 3491 : WaitLSNProcInfo *procInfo = pairingheap_container(WaitLSNProcInfo, heapNode, node);
212 : :
213 : 3491 : minWaitedLSN = procInfo->waitLSN;
214 : : }
215 : : /* Pairs with pg_atomic_read_membarrier_u64() in WaitLSNWakeup(). */
216 : 10082 : pg_atomic_write_membarrier_u64(&waitLSNState->minWaitedLSN[i], minWaitedLSN);
217 : 10082 : }
218 : :
219 : : /*
220 : : * Add current process to appropriate waiters heap based on LSN type
221 : : */
222 : : static void
223 : 3486 : addLSNWaiter(XLogRecPtr lsn, WaitLSNType lsnType)
224 : : {
225 : 3486 : WaitLSNProcInfo *procInfo = &waitLSNState->procInfos[MyProcNumber];
226 : 3486 : int i = (int) lsnType;
227 : :
228 : : Assert(i >= 0 && i < WAIT_LSN_TYPE_COUNT);
229 : :
230 : 3486 : LWLockAcquire(WaitLSNLock, LW_EXCLUSIVE);
231 : :
232 : 3486 : procInfo->procno = MyProcNumber;
233 : 3486 : procInfo->waitLSN = lsn;
234 : 3486 : procInfo->lsnType = lsnType;
235 : :
236 : : Assert(!procInfo->inHeap);
237 : 3486 : pairingheap_add(&waitLSNState->waitersHeap[i], &procInfo->heapNode);
238 : 3486 : procInfo->inHeap = true;
239 : 3486 : updateMinWaitedLSN(lsnType);
240 : :
241 : 3486 : LWLockRelease(WaitLSNLock);
242 : 3486 : }
243 : :
244 : : /*
245 : : * Remove current process from appropriate waiters heap based on LSN type
246 : : */
247 : : static void
248 : 45412 : deleteLSNWaiter(WaitLSNType lsnType)
249 : : {
250 : 45412 : WaitLSNProcInfo *procInfo = &waitLSNState->procInfos[MyProcNumber];
251 : 45412 : int i = (int) lsnType;
252 : :
253 : : Assert(i >= 0 && i < WAIT_LSN_TYPE_COUNT);
254 : :
255 : : /*
256 : : * Avoid taking WaitLSNLock if a waker has already removed us. Only this
257 : : * backend can set inHeap; other processes can only clear it. Therefore
258 : : * false is conclusive, while a stale true is harmless because it is
259 : : * rechecked under WaitLSNLock below.
260 : : */
261 [ + + ]: 45412 : if (!procInfo->inHeap)
262 : 41984 : return;
263 : :
264 : 3428 : LWLockAcquire(WaitLSNLock, LW_EXCLUSIVE);
265 : :
266 : : Assert(procInfo->lsnType == lsnType);
267 : :
268 [ + - ]: 3428 : if (procInfo->inHeap)
269 : : {
270 : 3428 : pairingheap_remove(&waitLSNState->waitersHeap[i], &procInfo->heapNode);
271 : 3428 : procInfo->inHeap = false;
272 : 3428 : updateMinWaitedLSN(lsnType);
273 : : }
274 : :
275 : 3428 : LWLockRelease(WaitLSNLock);
276 : : }
277 : :
278 : : /*
279 : : * Size of a static array of procs to wakeup by WaitLSNWakeup() allocated
280 : : * on the stack. It should be enough to take single iteration for most cases.
281 : : */
282 : : #define WAKEUP_PROC_STATIC_ARRAY_SIZE (16)
283 : :
284 : : /*
285 : : * Remove waiters whose LSN has been reached from the heap and set their
286 : : * latches. If InvalidXLogRecPtr is given, remove all waiters from the heap
287 : : * and set latches for all waiters.
288 : : *
289 : : * This function first accumulates waiters to wake up into an array, then
290 : : * wakes them up without holding a WaitLSNLock. The array size is static and
291 : : * equal to WAKEUP_PROC_STATIC_ARRAY_SIZE. That should be more than enough
292 : : * to wake up all the waiters at once in the vast majority of cases. However,
293 : : * if there are more waiters, this function will loop to process them in
294 : : * multiple chunks.
295 : : */
296 : : static void
297 : 3168 : wakeupWaiters(WaitLSNType lsnType, XLogRecPtr currentLSN)
298 : : {
299 : : ProcNumber wakeUpProcs[WAKEUP_PROC_STATIC_ARRAY_SIZE];
300 : : int numWakeUpProcs;
301 : 3168 : int i = (int) lsnType;
302 : :
303 : : Assert(i >= 0 && i < WAIT_LSN_TYPE_COUNT);
304 : :
305 : : do
306 : : {
307 : : int j;
308 : :
309 : 3168 : numWakeUpProcs = 0;
310 : 3168 : LWLockAcquire(WaitLSNLock, LW_EXCLUSIVE);
311 : :
312 : : /*
313 : : * Iterate the waiters heap until we find LSN not yet reached. Record
314 : : * process numbers to wake up, but send wakeups after releasing lock.
315 : : */
316 [ + + ]: 3224 : while (!pairingheap_is_empty(&waitLSNState->waitersHeap[i]))
317 : : {
318 : 60 : pairingheap_node *node = pairingheap_first(&waitLSNState->waitersHeap[i]);
319 : : WaitLSNProcInfo *procInfo;
320 : :
321 : : /* Get procInfo using appropriate heap node */
322 : 60 : procInfo = pairingheap_container(WaitLSNProcInfo, heapNode, node);
323 : :
324 [ + + + + ]: 60 : if (XLogRecPtrIsValid(currentLSN) && procInfo->waitLSN > currentLSN)
325 : 4 : break;
326 : :
327 : : Assert(numWakeUpProcs < WAKEUP_PROC_STATIC_ARRAY_SIZE);
328 : 56 : wakeUpProcs[numWakeUpProcs++] = procInfo->procno;
329 : 56 : (void) pairingheap_remove_first(&waitLSNState->waitersHeap[i]);
330 : :
331 : : /* Update appropriate flag */
332 : 56 : procInfo->inHeap = false;
333 : :
334 [ - + ]: 56 : if (numWakeUpProcs == WAKEUP_PROC_STATIC_ARRAY_SIZE)
335 : 0 : break;
336 : : }
337 : :
338 : 3168 : updateMinWaitedLSN(lsnType);
339 : 3168 : LWLockRelease(WaitLSNLock);
340 : :
341 : : /*
342 : : * Set latches for processes whose waited LSNs have been reached.
343 : : * Since SetLatch() is a time-consuming operation, we do this outside
344 : : * of WaitLSNLock. This is safe because procLatch is never freed, so
345 : : * at worst we may set a latch for the wrong process or for no process
346 : : * at all, which is harmless.
347 : : */
348 [ + + ]: 3224 : for (j = 0; j < numWakeUpProcs; j++)
349 : 56 : SetLatch(&GetPGProcByNumber(wakeUpProcs[j])->procLatch);
350 : :
351 [ - + ]: 3168 : } while (numWakeUpProcs == WAKEUP_PROC_STATIC_ARRAY_SIZE);
352 : 3168 : }
353 : :
354 : : /*
355 : : * Wake up processes waiting for LSN to reach currentLSN
356 : : */
357 : : void
358 : 9162026 : WaitLSNWakeup(WaitLSNType lsnType, XLogRecPtr currentLSN)
359 : : {
360 : 9162026 : int i = (int) lsnType;
361 : :
362 : : Assert(i >= 0 && i < WAIT_LSN_TYPE_COUNT);
363 : :
364 : : /*
365 : : * Fast path check. Skip if currentLSN is InvalidXLogRecPtr, which means
366 : : * "wake all waiters" (e.g., during promotion when recovery ends). Pairs
367 : : * with pg_atomic_write_membarrier_u64() in updateMinWaitedLSN().
368 : : */
369 [ + + + + ]: 18320926 : if (XLogRecPtrIsValid(currentLSN) &&
370 : 9158900 : pg_atomic_read_membarrier_u64(&waitLSNState->minWaitedLSN[i]) > currentLSN)
371 : 9158858 : return;
372 : :
373 : 3168 : wakeupWaiters(lsnType, currentLSN);
374 : : }
375 : :
376 : : /*
377 : : * Clean up any LSN wait state for the current process.
378 : : */
379 : : void
380 : 41928 : WaitLSNCleanup(void)
381 : : {
382 : : /*
383 : : * deleteLSNWaiter() starts with the same lockless inHeap check, so
384 : : * calling it unconditionally costs nothing when this process isn't
385 : : * waiting. Its lsnType is then unused, and reading it is harmless in any
386 : : * case: an entry that was never used is zeroed, which is a valid
387 : : * WaitLSNType.
388 : : */
389 [ + - ]: 41928 : if (waitLSNState)
390 : 41928 : deleteLSNWaiter(waitLSNState->procInfos[MyProcNumber].lsnType);
391 : 41928 : }
392 : :
393 : : /*
394 : : * Exit callback to clean up any LSN wait state left behind if this process
395 : : * exits while waiting. Transaction abort paths call WaitLSNCleanup()
396 : : * directly.
397 : : */
398 : : static void
399 : 263 : WaitLSNShmemExit(int code, Datum arg)
400 : : {
401 : 263 : WaitLSNCleanup();
402 : 263 : }
403 : :
404 : : /*
405 : : * Register shared-memory exit cleanup once per process. A backend may
406 : : * execute WAIT FOR LSN more than once.
407 : : */
408 : : static void
409 : 3485 : RegisterWaitLSNShmemExit(void)
410 : : {
411 [ + + ]: 3485 : if (!waitLSNShmemExitRegistered)
412 : : {
413 : 263 : on_shmem_exit(WaitLSNShmemExit, 0);
414 : 263 : waitLSNShmemExitRegistered = true;
415 : : }
416 : 3485 : }
417 : :
418 : : /*
419 : : * Check if the given LSN type requires recovery to be in progress.
420 : : * Standby wait types (replay, write, flush) require recovery;
421 : : * primary wait types (flush) do not.
422 : : */
423 : : static inline bool
424 : 3582 : WaitLSNTypeRequiresRecovery(WaitLSNType t)
425 : : {
426 [ + + ]: 3354 : return t == WAIT_LSN_TYPE_STANDBY_REPLAY ||
427 [ + + + + ]: 6936 : t == WAIT_LSN_TYPE_STANDBY_WRITE ||
428 : : t == WAIT_LSN_TYPE_STANDBY_FLUSH;
429 : : }
430 : :
431 : : /*
432 : : * Wait using MyLatch till the given LSN is reached, the replica gets
433 : : * promoted, or the postmaster dies.
434 : : *
435 : : * Returns WAIT_LSN_RESULT_SUCCESS if target LSN was reached.
436 : : * Returns WAIT_LSN_RESULT_NOT_IN_RECOVERY if run not in recovery,
437 : : * or replica got promoted before the target LSN reached.
438 : : */
439 : : WaitLSNResult
440 : 3485 : WaitForLSN(WaitLSNType lsnType, XLogRecPtr targetLSN, int64 timeout)
441 : : {
442 : : XLogRecPtr currentLSN;
443 : : WaitLSNProcInfo *procInfo;
444 : 3485 : TimestampTz endtime = 0;
445 : 3485 : int wake_events = WL_LATCH_SET | WL_POSTMASTER_DEATH;
446 : :
447 : : /* Shouldn't be called when shmem isn't initialized */
448 : : Assert(waitLSNState);
449 : :
450 : : /* Should have a valid proc number */
451 : : Assert(MyProcNumber >= 0 && MyProcNumber < MaxBackends + NUM_AUXILIARY_PROCS);
452 : :
453 : 3485 : procInfo = &waitLSNState->procInfos[MyProcNumber];
454 : :
455 : : /*
456 : : * Ensure cleanup is registered before publishing our waiter entry.
457 : : * on_shmem_exit callbacks run in reverse registration order, so this
458 : : * callback runs before the earlier-registered ProcKill() and removes the
459 : : * entry before our PGPROC slot can be reused.
460 : : */
461 : 3485 : RegisterWaitLSNShmemExit();
462 : :
463 [ + + ]: 3485 : if (timeout > 0)
464 : : {
465 : 3469 : endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), timeout);
466 : 3469 : wake_events |= WL_TIMEOUT;
467 : : }
468 : :
469 : : /*
470 : : * Add our process to the waiters heap. It might happen that target LSN
471 : : * gets reached before we do. The check at the beginning of the loop
472 : : * below prevents the race condition.
473 : : */
474 : 3485 : addLSNWaiter(targetLSN, lsnType);
475 : :
476 : : for (;;)
477 : 97 : {
478 : : int rc;
479 : 3582 : long delay_ms = -1;
480 : :
481 : : /* Get current LSN for the wait type */
482 : 3582 : currentLSN = GetCurrentLSNForWaitType(lsnType);
483 : :
484 : : /* Check that recovery is still in-progress */
485 [ + + + + ]: 3582 : if (WaitLSNTypeRequiresRecovery(lsnType) && !RecoveryInProgress())
486 : : {
487 : : /*
488 : : * Recovery was ended, but check if target LSN was already
489 : : * reached.
490 : : */
491 : 6 : deleteLSNWaiter(lsnType);
492 : :
493 [ + + + + ]: 6 : if (PromoteIsTriggered() && targetLSN <= currentLSN)
494 : 1 : return WAIT_LSN_RESULT_SUCCESS;
495 : 5 : return WAIT_LSN_RESULT_NOT_IN_RECOVERY;
496 : : }
497 : : else
498 : : {
499 : : /* Check if the waited LSN has been reached */
500 [ + + ]: 3576 : if (targetLSN <= currentLSN)
501 : 3466 : break;
502 : : }
503 : :
504 [ + + ]: 110 : CHECK_FOR_INTERRUPTS();
505 : :
506 : : /*
507 : : * The target is not reached. Normally we remain in the waiters heap
508 : : * and can sleep again. A wakeup can become stale, however, if the
509 : : * position moves backwards after the waker removed us. That happens
510 : : * with the walreceiver-tracked positions: when streaming starts on a
511 : : * new timeline, or after receiveStart was reset,
512 : : * RequestXLogStreaming() re-seeds writtenUpto and flushedUpto with
513 : : * the requested start position, which can be below what was published
514 : : * before. Re-register in that case and reread the position, since an
515 : : * advance between the previous read and the re-add could not have
516 : : * woken us.
517 : : *
518 : : * A wakeup that goes stale again sends us around the loop once more,
519 : : * so interrupts are processed before we re-register: however often
520 : : * that repeats, the wait stays cancellable. Repeating requires a
521 : : * fresh wakeup, hence the position reaching the target and falling
522 : : * back below it, so it follows streaming restarts rather than burning
523 : : * CPU. The deadline is checked on every iteration that goes on to
524 : : * sleep, which is the only place it matters.
525 : : *
526 : : * It is safe to read inHeap without the lock because only this
527 : : * process sets it true. If a waker clears it concurrently, it also
528 : : * sets our latch, so we will recheck and re-register if necessary.
529 : : */
530 [ + + ]: 109 : if (!procInfo->inHeap)
531 : : {
532 : 1 : addLSNWaiter(targetLSN, lsnType);
533 : 1 : continue;
534 : : }
535 : :
536 [ + + ]: 108 : if (timeout > 0)
537 : : {
538 : 92 : delay_ms = TimestampDifferenceMilliseconds(GetCurrentTimestamp(), endtime);
539 [ + + ]: 92 : if (delay_ms <= 0)
540 : 12 : break;
541 : : }
542 : :
543 : 96 : rc = WaitLatch(MyLatch, wake_events, delay_ms,
544 : 96 : WaitLSNWaitEvents[lsnType]);
545 : :
546 : : /*
547 : : * Emergency bailout if postmaster has died. This is to avoid the
548 : : * necessity for manual cleanup of all postmaster children.
549 : : */
550 [ - + ]: 96 : if (rc & WL_POSTMASTER_DEATH)
551 [ # # ]: 0 : ereport(FATAL,
552 : : errcode(ERRCODE_ADMIN_SHUTDOWN),
553 : : errmsg("terminating connection due to unexpected postmaster exit"),
554 : : errcontext("while waiting for LSN"));
555 : :
556 : 96 : ResetLatch(MyLatch);
557 : : }
558 : :
559 : : /*
560 : : * A progress waker, such as the startup process during WAL replay, may
561 : : * already have removed this waiter through WaitLSNWakeup() before setting
562 : : * its latch. The inHeap flag makes this cleanup safe whether or not the
563 : : * entry remains in the heap.
564 : : */
565 : 3478 : deleteLSNWaiter(lsnType);
566 : :
567 : : /*
568 : : * If we didn't reach the target LSN, we must be exited by timeout.
569 : : */
570 [ + + ]: 3478 : if (targetLSN > currentLSN)
571 : 12 : return WAIT_LSN_RESULT_TIMEOUT;
572 : :
573 : 3466 : return WAIT_LSN_RESULT_SUCCESS;
574 : : }
|
