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