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/latch.h"
58 : : #include "storage/proc.h"
59 : : #include "storage/shmem.h"
60 : : #include "storage/subsystems.h"
61 : : #include "utils/fmgrprotos.h"
62 : : #include "utils/pg_lsn.h"
63 : : #include "utils/snapmgr.h"
64 : : #include "utils/wait_event.h"
65 : :
66 : :
67 : : static int waitlsn_cmp(const pairingheap_node *a, const pairingheap_node *b,
68 : : void *arg);
69 : :
70 : : struct WaitLSNState *waitLSNState = NULL;
71 : :
72 : : static void WaitLSNShmemRequest(void *arg);
73 : : static void WaitLSNShmemInit(void *arg);
74 : :
75 : : const ShmemCallbacks WaitLSNShmemCallbacks = {
76 : : .request_fn = WaitLSNShmemRequest,
77 : : .init_fn = WaitLSNShmemInit,
78 : : };
79 : :
80 : : /*
81 : : * Wait event for each WaitLSNType, used with WaitLatch() to report
82 : : * the wait in pg_stat_activity.
83 : : */
84 : : static const uint32 WaitLSNWaitEvents[] = {
85 : : [WAIT_LSN_TYPE_STANDBY_REPLAY] = WAIT_EVENT_WAIT_FOR_WAL_REPLAY,
86 : : [WAIT_LSN_TYPE_STANDBY_WRITE] = WAIT_EVENT_WAIT_FOR_WAL_WRITE,
87 : : [WAIT_LSN_TYPE_STANDBY_FLUSH] = WAIT_EVENT_WAIT_FOR_WAL_FLUSH,
88 : : [WAIT_LSN_TYPE_PRIMARY_FLUSH] = WAIT_EVENT_WAIT_FOR_WAL_FLUSH,
89 : : };
90 : :
91 : : StaticAssertDecl(lengthof(WaitLSNWaitEvents) == WAIT_LSN_TYPE_COUNT,
92 : : "WaitLSNWaitEvents must match WaitLSNType enum");
93 : :
94 : : /*
95 : : * Get the current LSN for the specified wait type. Provide memory
96 : : * barrier semantics before getting the value.
97 : : */
98 : : XLogRecPtr
99 : 374 : GetCurrentLSNForWaitType(WaitLSNType lsnType)
100 : : {
101 : : Assert(lsnType >= 0 && lsnType < WAIT_LSN_TYPE_COUNT);
102 : :
103 : : /*
104 : : * All of the cases below provide memory barrier semantics:
105 : : * GetWalRcvWriteRecPtr() and GetFlushRecPtr() have explicit barriers,
106 : : * while GetXLogReplayRecPtr() and GetWalRcvFlushRecPtr() use spinlocks.
107 : : */
108 [ + + + + : 374 : switch (lsnType)
- ]
109 : : {
110 : 225 : case WAIT_LSN_TYPE_STANDBY_REPLAY:
111 : 225 : return GetXLogReplayRecPtr(NULL);
112 : :
113 : 48 : case WAIT_LSN_TYPE_STANDBY_WRITE:
114 : : {
115 : 48 : XLogRecPtr recptr = GetWalRcvWriteRecPtr();
116 : 48 : XLogRecPtr replay = GetXLogReplayRecPtr(NULL);
117 : :
118 : : /*
119 : : * Use the replay position as a floor. WAL up to the replay
120 : : * point is already on disk from a base backup, archive
121 : : * restore, or prior streaming, so there is no reason to wait
122 : : * for the walreceiver to re-receive it.
123 : : */
124 : 48 : return Max(recptr, replay);
125 : : }
126 : :
127 : 38 : case WAIT_LSN_TYPE_STANDBY_FLUSH:
128 : : {
129 : 38 : XLogRecPtr recptr = GetWalRcvFlushRecPtr(NULL, NULL);
130 : 38 : XLogRecPtr replay = GetXLogReplayRecPtr(NULL);
131 : :
132 : : /* Same floor as standby_write; see comment above. */
133 : 38 : return Max(recptr, replay);
134 : : }
135 : :
136 : 63 : case WAIT_LSN_TYPE_PRIMARY_FLUSH:
137 : 63 : return GetFlushRecPtr(NULL);
138 : : }
139 : :
140 [ # # ]: 0 : elog(ERROR, "invalid LSN wait type: %d", lsnType);
141 : : pg_unreachable();
142 : : }
143 : :
144 : : /* Register the shared memory space needed for WaitLSNState. */
145 : : static void
146 : 1249 : WaitLSNShmemRequest(void *arg)
147 : : {
148 : : Size size;
149 : :
150 : 1249 : size = offsetof(WaitLSNState, procInfos);
151 : 1249 : size = add_size(size, mul_size(MaxBackends + NUM_AUXILIARY_PROCS, sizeof(WaitLSNProcInfo)));
152 : 1249 : ShmemRequestStruct(.name = "WaitLSNState",
153 : : .size = size,
154 : : .ptr = (void **) &waitLSNState,
155 : : );
156 : 1249 : }
157 : :
158 : : /* Initialize the WaitLSNState in the shared memory. */
159 : : static void
160 : 1246 : WaitLSNShmemInit(void *arg)
161 : : {
162 : : /* Initialize heaps and tracking */
163 [ + + ]: 6230 : for (int i = 0; i < WAIT_LSN_TYPE_COUNT; i++)
164 : : {
165 : 4984 : pg_atomic_init_u64(&waitLSNState->minWaitedLSN[i], PG_UINT64_MAX);
166 : 4984 : pairingheap_initialize(&waitLSNState->waitersHeap[i], waitlsn_cmp, NULL);
167 : : }
168 : :
169 : : /* Initialize process info array */
170 : 1246 : memset(&waitLSNState->procInfos, 0,
171 : 1246 : (MaxBackends + NUM_AUXILIARY_PROCS) * sizeof(WaitLSNProcInfo));
172 : 1246 : }
173 : :
174 : : /*
175 : : * Comparison function for LSN waiters heaps. Waiting processes are ordered by
176 : : * LSN, so that the waiter with smallest LSN is at the top.
177 : : */
178 : : static int
179 : 28 : waitlsn_cmp(const pairingheap_node *a, const pairingheap_node *b, void *arg)
180 : : {
181 : 28 : const WaitLSNProcInfo *aproc = pairingheap_const_container(WaitLSNProcInfo, heapNode, a);
182 : 28 : const WaitLSNProcInfo *bproc = pairingheap_const_container(WaitLSNProcInfo, heapNode, b);
183 : :
184 [ + + ]: 28 : if (aproc->waitLSN < bproc->waitLSN)
185 : 15 : return 1;
186 [ + + ]: 13 : else if (aproc->waitLSN > bproc->waitLSN)
187 : 10 : return -1;
188 : : else
189 : 3 : return 0;
190 : : }
191 : :
192 : : /*
193 : : * Update minimum waited LSN for the specified LSN type
194 : : */
195 : : static void
196 : 3570 : updateMinWaitedLSN(WaitLSNType lsnType)
197 : : {
198 : 3570 : XLogRecPtr minWaitedLSN = PG_UINT64_MAX;
199 : 3570 : int i = (int) lsnType;
200 : :
201 : : Assert(i >= 0 && i < WAIT_LSN_TYPE_COUNT);
202 : :
203 [ + + ]: 3570 : if (!pairingheap_is_empty(&waitLSNState->waitersHeap[i]))
204 : : {
205 : 269 : pairingheap_node *node = pairingheap_first(&waitLSNState->waitersHeap[i]);
206 : 269 : WaitLSNProcInfo *procInfo = pairingheap_container(WaitLSNProcInfo, heapNode, node);
207 : :
208 : 269 : minWaitedLSN = procInfo->waitLSN;
209 : : }
210 : : /* Pairs with pg_atomic_read_membarrier_u64() in WaitLSNWakeup(). */
211 : 3570 : pg_atomic_write_membarrier_u64(&waitLSNState->minWaitedLSN[i], minWaitedLSN);
212 : 3570 : }
213 : :
214 : : /*
215 : : * Add current process to appropriate waiters heap based on LSN type
216 : : */
217 : : static void
218 : 264 : addLSNWaiter(XLogRecPtr lsn, WaitLSNType lsnType)
219 : : {
220 : 264 : WaitLSNProcInfo *procInfo = &waitLSNState->procInfos[MyProcNumber];
221 : 264 : int i = (int) lsnType;
222 : :
223 : : Assert(i >= 0 && i < WAIT_LSN_TYPE_COUNT);
224 : :
225 : 264 : LWLockAcquire(WaitLSNLock, LW_EXCLUSIVE);
226 : :
227 : 264 : procInfo->procno = MyProcNumber;
228 : 264 : procInfo->waitLSN = lsn;
229 : 264 : procInfo->lsnType = lsnType;
230 : :
231 : : Assert(!procInfo->inHeap);
232 : 264 : pairingheap_add(&waitLSNState->waitersHeap[i], &procInfo->heapNode);
233 : 264 : procInfo->inHeap = true;
234 : 264 : updateMinWaitedLSN(lsnType);
235 : :
236 : 264 : LWLockRelease(WaitLSNLock);
237 : 264 : }
238 : :
239 : : /*
240 : : * Remove current process from appropriate waiters heap based on LSN type
241 : : */
242 : : static void
243 : 264 : deleteLSNWaiter(WaitLSNType lsnType)
244 : : {
245 : 264 : WaitLSNProcInfo *procInfo = &waitLSNState->procInfos[MyProcNumber];
246 : 264 : int i = (int) lsnType;
247 : :
248 : : Assert(i >= 0 && i < WAIT_LSN_TYPE_COUNT);
249 : :
250 : 264 : LWLockAcquire(WaitLSNLock, LW_EXCLUSIVE);
251 : :
252 : : Assert(procInfo->lsnType == lsnType);
253 : :
254 [ + + ]: 264 : if (procInfo->inHeap)
255 : : {
256 : 208 : pairingheap_remove(&waitLSNState->waitersHeap[i], &procInfo->heapNode);
257 : 208 : procInfo->inHeap = false;
258 : 208 : updateMinWaitedLSN(lsnType);
259 : : }
260 : :
261 : 264 : LWLockRelease(WaitLSNLock);
262 : 264 : }
263 : :
264 : : /*
265 : : * Size of a static array of procs to wakeup by WaitLSNWakeup() allocated
266 : : * on the stack. It should be enough to take single iteration for most cases.
267 : : */
268 : : #define WAKEUP_PROC_STATIC_ARRAY_SIZE (16)
269 : :
270 : : /*
271 : : * Remove waiters whose LSN has been reached from the heap and set their
272 : : * latches. If InvalidXLogRecPtr is given, remove all waiters from the heap
273 : : * and set latches for all waiters.
274 : : *
275 : : * This function first accumulates waiters to wake up into an array, then
276 : : * wakes them up without holding a WaitLSNLock. The array size is static and
277 : : * equal to WAKEUP_PROC_STATIC_ARRAY_SIZE. That should be more than enough
278 : : * to wake up all the waiters at once in the vast majority of cases. However,
279 : : * if there are more waiters, this function will loop to process them in
280 : : * multiple chunks.
281 : : */
282 : : static void
283 : 3098 : wakeupWaiters(WaitLSNType lsnType, XLogRecPtr currentLSN)
284 : : {
285 : : ProcNumber wakeUpProcs[WAKEUP_PROC_STATIC_ARRAY_SIZE];
286 : : int numWakeUpProcs;
287 : 3098 : int i = (int) lsnType;
288 : :
289 : : Assert(i >= 0 && i < WAIT_LSN_TYPE_COUNT);
290 : :
291 : : do
292 : : {
293 : : int j;
294 : :
295 : 3098 : numWakeUpProcs = 0;
296 : 3098 : LWLockAcquire(WaitLSNLock, LW_EXCLUSIVE);
297 : :
298 : : /*
299 : : * Iterate the waiters heap until we find LSN not yet reached. Record
300 : : * process numbers to wake up, but send wakeups after releasing lock.
301 : : */
302 [ + + ]: 3153 : while (!pairingheap_is_empty(&waitLSNState->waitersHeap[i]))
303 : : {
304 : 59 : pairingheap_node *node = pairingheap_first(&waitLSNState->waitersHeap[i]);
305 : : WaitLSNProcInfo *procInfo;
306 : :
307 : : /* Get procInfo using appropriate heap node */
308 : 59 : procInfo = pairingheap_container(WaitLSNProcInfo, heapNode, node);
309 : :
310 [ + + + + ]: 59 : if (XLogRecPtrIsValid(currentLSN) && procInfo->waitLSN > currentLSN)
311 : 4 : break;
312 : :
313 : : Assert(numWakeUpProcs < WAKEUP_PROC_STATIC_ARRAY_SIZE);
314 : 55 : wakeUpProcs[numWakeUpProcs++] = procInfo->procno;
315 : 55 : (void) pairingheap_remove_first(&waitLSNState->waitersHeap[i]);
316 : :
317 : : /* Update appropriate flag */
318 : 55 : procInfo->inHeap = false;
319 : :
320 [ - + ]: 55 : if (numWakeUpProcs == WAKEUP_PROC_STATIC_ARRAY_SIZE)
321 : 0 : break;
322 : : }
323 : :
324 : 3098 : updateMinWaitedLSN(lsnType);
325 : 3098 : LWLockRelease(WaitLSNLock);
326 : :
327 : : /*
328 : : * Set latches for processes whose waited LSNs have been reached.
329 : : * Since SetLatch() is a time-consuming operation, we do this outside
330 : : * of WaitLSNLock. This is safe because procLatch is never freed, so
331 : : * at worst we may set a latch for the wrong process or for no process
332 : : * at all, which is harmless.
333 : : */
334 [ + + ]: 3153 : for (j = 0; j < numWakeUpProcs; j++)
335 : 55 : SetLatch(&GetPGProcByNumber(wakeUpProcs[j])->procLatch);
336 : :
337 [ - + ]: 3098 : } while (numWakeUpProcs == WAKEUP_PROC_STATIC_ARRAY_SIZE);
338 : 3098 : }
339 : :
340 : : /*
341 : : * Wake up processes waiting for LSN to reach currentLSN
342 : : */
343 : : void
344 : 9236607 : WaitLSNWakeup(WaitLSNType lsnType, XLogRecPtr currentLSN)
345 : : {
346 : 9236607 : int i = (int) lsnType;
347 : :
348 : : Assert(i >= 0 && i < WAIT_LSN_TYPE_COUNT);
349 : :
350 : : /*
351 : : * Fast path check. Skip if currentLSN is InvalidXLogRecPtr, which means
352 : : * "wake all waiters" (e.g., during promotion when recovery ends). Pairs
353 : : * with pg_atomic_write_membarrier_u64() in updateMinWaitedLSN().
354 : : */
355 [ + + + + ]: 18470157 : if (XLogRecPtrIsValid(currentLSN) &&
356 : 9233550 : pg_atomic_read_membarrier_u64(&waitLSNState->minWaitedLSN[i]) > currentLSN)
357 : 9233509 : return;
358 : :
359 : 3098 : wakeupWaiters(lsnType, currentLSN);
360 : : }
361 : :
362 : : /*
363 : : * Clean up any LSN wait state for the current process.
364 : : */
365 : : void
366 : 61079 : WaitLSNCleanup(void)
367 : : {
368 [ + - ]: 61079 : if (waitLSNState)
369 : : {
370 : : /*
371 : : * We do a fast-path check of the inHeap flag without the lock. This
372 : : * flag is set to true only by the process itself. So, it's only
373 : : * possible to get a false positive. But that will be eliminated by a
374 : : * recheck inside deleteLSNWaiter().
375 : : */
376 [ + + ]: 61079 : if (waitLSNState->procInfos[MyProcNumber].inHeap)
377 : 1 : deleteLSNWaiter(waitLSNState->procInfos[MyProcNumber].lsnType);
378 : : }
379 : 61079 : }
380 : :
381 : : /*
382 : : * Check if the given LSN type requires recovery to be in progress.
383 : : * Standby wait types (replay, write, flush) require recovery;
384 : : * primary wait types (flush) do not.
385 : : */
386 : : static inline bool
387 : 370 : WaitLSNTypeRequiresRecovery(WaitLSNType t)
388 : : {
389 [ + + ]: 147 : return t == WAIT_LSN_TYPE_STANDBY_REPLAY ||
390 [ + + + + ]: 517 : t == WAIT_LSN_TYPE_STANDBY_WRITE ||
391 : : t == WAIT_LSN_TYPE_STANDBY_FLUSH;
392 : : }
393 : :
394 : : /*
395 : : * Wait using MyLatch till the given LSN is reached, the replica gets
396 : : * promoted, or the postmaster dies.
397 : : *
398 : : * Returns WAIT_LSN_RESULT_SUCCESS if target LSN was reached.
399 : : * Returns WAIT_LSN_RESULT_NOT_IN_RECOVERY if run not in recovery,
400 : : * or replica got promoted before the target LSN reached.
401 : : */
402 : : WaitLSNResult
403 : 264 : WaitForLSN(WaitLSNType lsnType, XLogRecPtr targetLSN, int64 timeout)
404 : : {
405 : : XLogRecPtr currentLSN;
406 : 264 : TimestampTz endtime = 0;
407 : 264 : int wake_events = WL_LATCH_SET | WL_POSTMASTER_DEATH;
408 : :
409 : : /* Shouldn't be called when shmem isn't initialized */
410 : : Assert(waitLSNState);
411 : :
412 : : /* Should have a valid proc number */
413 : : Assert(MyProcNumber >= 0 && MyProcNumber < MaxBackends + NUM_AUXILIARY_PROCS);
414 : :
415 [ + + ]: 264 : if (timeout > 0)
416 : : {
417 : 252 : endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), timeout);
418 : 252 : wake_events |= WL_TIMEOUT;
419 : : }
420 : :
421 : : /*
422 : : * Add our process to the waiters heap. It might happen that target LSN
423 : : * gets reached before we do. The check at the beginning of the loop
424 : : * below prevents the race condition.
425 : : */
426 : 264 : addLSNWaiter(targetLSN, lsnType);
427 : :
428 : : for (;;)
429 : 106 : {
430 : : int rc;
431 : 370 : long delay_ms = -1;
432 : :
433 : : /* Get current LSN for the wait type */
434 : 370 : currentLSN = GetCurrentLSNForWaitType(lsnType);
435 : :
436 : : /* Check that recovery is still in-progress */
437 [ + + + + ]: 370 : if (WaitLSNTypeRequiresRecovery(lsnType) && !RecoveryInProgress())
438 : : {
439 : : /*
440 : : * Recovery was ended, but check if target LSN was already
441 : : * reached.
442 : : */
443 : 6 : deleteLSNWaiter(lsnType);
444 : :
445 [ + + + + ]: 6 : if (PromoteIsTriggered() && targetLSN <= currentLSN)
446 : 1 : return WAIT_LSN_RESULT_SUCCESS;
447 : 5 : return WAIT_LSN_RESULT_NOT_IN_RECOVERY;
448 : : }
449 : : else
450 : : {
451 : : /* Check if the waited LSN has been reached */
452 [ + + ]: 364 : if (targetLSN <= currentLSN)
453 : 244 : break;
454 : : }
455 : :
456 [ + + ]: 120 : if (timeout > 0)
457 : : {
458 : 103 : delay_ms = TimestampDifferenceMilliseconds(GetCurrentTimestamp(), endtime);
459 [ + + ]: 103 : if (delay_ms <= 0)
460 : 13 : break;
461 : : }
462 : :
463 [ + + ]: 107 : CHECK_FOR_INTERRUPTS();
464 : :
465 : 106 : rc = WaitLatch(MyLatch, wake_events, delay_ms,
466 : 106 : WaitLSNWaitEvents[lsnType]);
467 : :
468 : : /*
469 : : * Emergency bailout if postmaster has died. This is to avoid the
470 : : * necessity for manual cleanup of all postmaster children.
471 : : */
472 [ - + ]: 106 : if (rc & WL_POSTMASTER_DEATH)
473 [ # # ]: 0 : ereport(FATAL,
474 : : errcode(ERRCODE_ADMIN_SHUTDOWN),
475 : : errmsg("terminating connection due to unexpected postmaster exit"),
476 : : errcontext("while waiting for LSN"));
477 : :
478 : 106 : ResetLatch(MyLatch);
479 : : }
480 : :
481 : : /*
482 : : * Delete our process from the shared memory heap. We might already be
483 : : * deleted by the startup process. The 'inHeap' flags prevents us from
484 : : * the double deletion.
485 : : */
486 : 257 : deleteLSNWaiter(lsnType);
487 : :
488 : : /*
489 : : * If we didn't reach the target LSN, we must be exited by timeout.
490 : : */
491 [ + + ]: 257 : if (targetLSN > currentLSN)
492 : 13 : return WAIT_LSN_RESULT_TIMEOUT;
493 : :
494 : 244 : return WAIT_LSN_RESULT_SUCCESS;
495 : : }
|
