PostgreSQL Source Code git master
xlog.c
Go to the documentation of this file.
1/*-------------------------------------------------------------------------
2 *
3 * xlog.c
4 * PostgreSQL write-ahead log manager
5 *
6 * The Write-Ahead Log (WAL) functionality is split into several source
7 * files, in addition to this one:
8 *
9 * xloginsert.c - Functions for constructing WAL records
10 * xlogrecovery.c - WAL recovery and standby code
11 * xlogreader.c - Facility for reading WAL files and parsing WAL records
12 * xlogutils.c - Helper functions for WAL redo routines
13 *
14 * This file contains functions for coordinating database startup and
15 * checkpointing, and managing the write-ahead log buffers when the
16 * system is running.
17 *
18 * StartupXLOG() is the main entry point of the startup process. It
19 * coordinates database startup, performing WAL recovery, and the
20 * transition from WAL recovery into normal operations.
21 *
22 * XLogInsertRecord() inserts a WAL record into the WAL buffers. Most
23 * callers should not call this directly, but use the functions in
24 * xloginsert.c to construct the WAL record. XLogFlush() can be used
25 * to force the WAL to disk.
26 *
27 * In addition to those, there are many other functions for interrogating
28 * the current system state, and for starting/stopping backups.
29 *
30 *
31 * Portions Copyright (c) 1996-2025, PostgreSQL Global Development Group
32 * Portions Copyright (c) 1994, Regents of the University of California
33 *
34 * src/backend/access/transam/xlog.c
35 *
36 *-------------------------------------------------------------------------
37 */
38
39#include "postgres.h"
40
41#include <ctype.h>
42#include <math.h>
43#include <time.h>
44#include <fcntl.h>
45#include <sys/stat.h>
46#include <sys/time.h>
47#include <unistd.h>
48
49#include "access/clog.h"
50#include "access/commit_ts.h"
51#include "access/heaptoast.h"
52#include "access/multixact.h"
53#include "access/rewriteheap.h"
54#include "access/subtrans.h"
55#include "access/timeline.h"
56#include "access/transam.h"
57#include "access/twophase.h"
58#include "access/xact.h"
60#include "access/xlogarchive.h"
61#include "access/xloginsert.h"
62#include "access/xlogreader.h"
63#include "access/xlogrecovery.h"
64#include "access/xlogutils.h"
65#include "backup/basebackup.h"
66#include "catalog/catversion.h"
67#include "catalog/pg_control.h"
68#include "catalog/pg_database.h"
70#include "common/file_utils.h"
71#include "executor/instrument.h"
72#include "miscadmin.h"
73#include "pg_trace.h"
74#include "pgstat.h"
75#include "port/atomics.h"
76#include "postmaster/bgwriter.h"
77#include "postmaster/startup.h"
80#include "replication/origin.h"
81#include "replication/slot.h"
85#include "storage/bufmgr.h"
86#include "storage/fd.h"
87#include "storage/ipc.h"
89#include "storage/latch.h"
90#include "storage/predicate.h"
91#include "storage/proc.h"
92#include "storage/procarray.h"
93#include "storage/reinit.h"
94#include "storage/spin.h"
95#include "storage/sync.h"
96#include "utils/guc_hooks.h"
97#include "utils/guc_tables.h"
99#include "utils/ps_status.h"
100#include "utils/relmapper.h"
101#include "utils/snapmgr.h"
102#include "utils/timeout.h"
103#include "utils/timestamp.h"
104#include "utils/varlena.h"
105
106#ifdef WAL_DEBUG
107#include "utils/memutils.h"
108#endif
109
110/* timeline ID to be used when bootstrapping */
111#define BootstrapTimeLineID 1
112
113/* User-settable parameters */
114int max_wal_size_mb = 1024; /* 1 GB */
115int min_wal_size_mb = 80; /* 80 MB */
117int XLOGbuffers = -1;
121bool EnableHotStandby = false;
122bool fullPageWrites = true;
123bool wal_log_hints = false;
127bool wal_init_zero = true;
128bool wal_recycle = true;
129bool log_checkpoints = true;
132int CommitDelay = 0; /* precommit delay in microseconds */
133int CommitSiblings = 5; /* # concurrent xacts needed to sleep */
136int wal_decode_buffer_size = 512 * 1024;
138
139#ifdef WAL_DEBUG
140bool XLOG_DEBUG = false;
141#endif
142
144
145/*
146 * Number of WAL insertion locks to use. A higher value allows more insertions
147 * to happen concurrently, but adds some CPU overhead to flushing the WAL,
148 * which needs to iterate all the locks.
149 */
150#define NUM_XLOGINSERT_LOCKS 8
151
152/*
153 * Max distance from last checkpoint, before triggering a new xlog-based
154 * checkpoint.
155 */
157
158/* Estimated distance between checkpoints, in bytes */
160static double PrevCheckPointDistance = 0;
161
162/*
163 * Track whether there were any deferred checks for custom resource managers
164 * specified in wal_consistency_checking.
165 */
167
168/*
169 * GUC support
170 */
172 {"fsync", WAL_SYNC_METHOD_FSYNC, false},
173#ifdef HAVE_FSYNC_WRITETHROUGH
174 {"fsync_writethrough", WAL_SYNC_METHOD_FSYNC_WRITETHROUGH, false},
175#endif
176 {"fdatasync", WAL_SYNC_METHOD_FDATASYNC, false},
177#ifdef O_SYNC
178 {"open_sync", WAL_SYNC_METHOD_OPEN, false},
179#endif
180#ifdef O_DSYNC
181 {"open_datasync", WAL_SYNC_METHOD_OPEN_DSYNC, false},
182#endif
183 {NULL, 0, false}
184};
185
186
187/*
188 * Although only "on", "off", and "always" are documented,
189 * we accept all the likely variants of "on" and "off".
190 */
192 {"always", ARCHIVE_MODE_ALWAYS, false},
193 {"on", ARCHIVE_MODE_ON, false},
194 {"off", ARCHIVE_MODE_OFF, false},
195 {"true", ARCHIVE_MODE_ON, true},
196 {"false", ARCHIVE_MODE_OFF, true},
197 {"yes", ARCHIVE_MODE_ON, true},
198 {"no", ARCHIVE_MODE_OFF, true},
199 {"1", ARCHIVE_MODE_ON, true},
200 {"0", ARCHIVE_MODE_OFF, true},
201 {NULL, 0, false}
202};
203
204/*
205 * Statistics for current checkpoint are collected in this global struct.
206 * Because only the checkpointer or a stand-alone backend can perform
207 * checkpoints, this will be unused in normal backends.
208 */
210
211/*
212 * During recovery, lastFullPageWrites keeps track of full_page_writes that
213 * the replayed WAL records indicate. It's initialized with full_page_writes
214 * that the recovery starting checkpoint record indicates, and then updated
215 * each time XLOG_FPW_CHANGE record is replayed.
216 */
218
219/*
220 * Local copy of the state tracked by SharedRecoveryState in shared memory,
221 * It is false if SharedRecoveryState is RECOVERY_STATE_DONE. True actually
222 * means "not known, need to check the shared state".
223 */
224static bool LocalRecoveryInProgress = true;
225
226/*
227 * Local state for XLogInsertAllowed():
228 * 1: unconditionally allowed to insert XLOG
229 * 0: unconditionally not allowed to insert XLOG
230 * -1: must check RecoveryInProgress(); disallow until it is false
231 * Most processes start with -1 and transition to 1 after seeing that recovery
232 * is not in progress. But we can also force the value for special cases.
233 * The coding in XLogInsertAllowed() depends on the first two of these states
234 * being numerically the same as bool true and false.
235 */
237
238/*
239 * ProcLastRecPtr points to the start of the last XLOG record inserted by the
240 * current backend. It is updated for all inserts. XactLastRecEnd points to
241 * end+1 of the last record, and is reset when we end a top-level transaction,
242 * or start a new one; so it can be used to tell if the current transaction has
243 * created any XLOG records.
244 *
245 * While in parallel mode, this may not be fully up to date. When committing,
246 * a transaction can assume this covers all xlog records written either by the
247 * user backend or by any parallel worker which was present at any point during
248 * the transaction. But when aborting, or when still in parallel mode, other
249 * parallel backends may have written WAL records at later LSNs than the value
250 * stored here. The parallel leader advances its own copy, when necessary,
251 * in WaitForParallelWorkersToFinish.
252 */
256
257/*
258 * RedoRecPtr is this backend's local copy of the REDO record pointer
259 * (which is almost but not quite the same as a pointer to the most recent
260 * CHECKPOINT record). We update this from the shared-memory copy,
261 * XLogCtl->Insert.RedoRecPtr, whenever we can safely do so (ie, when we
262 * hold an insertion lock). See XLogInsertRecord for details. We are also
263 * allowed to update from XLogCtl->RedoRecPtr if we hold the info_lck;
264 * see GetRedoRecPtr.
265 *
266 * NB: Code that uses this variable must be prepared not only for the
267 * possibility that it may be arbitrarily out of date, but also for the
268 * possibility that it might be set to InvalidXLogRecPtr. We used to
269 * initialize it as a side effect of the first call to RecoveryInProgress(),
270 * which meant that most code that might use it could assume that it had a
271 * real if perhaps stale value. That's no longer the case.
272 */
274
275/*
276 * doPageWrites is this backend's local copy of (fullPageWrites ||
277 * runningBackups > 0). It is used together with RedoRecPtr to decide whether
278 * a full-page image of a page need to be taken.
279 *
280 * NB: Initially this is false, and there's no guarantee that it will be
281 * initialized to any other value before it is first used. Any code that
282 * makes use of it must recheck the value after obtaining a WALInsertLock,
283 * and respond appropriately if it turns out that the previous value wasn't
284 * accurate.
285 */
286static bool doPageWrites;
287
288/*----------
289 * Shared-memory data structures for XLOG control
290 *
291 * LogwrtRqst indicates a byte position that we need to write and/or fsync
292 * the log up to (all records before that point must be written or fsynced).
293 * The positions already written/fsynced are maintained in logWriteResult
294 * and logFlushResult using atomic access.
295 * In addition to the shared variable, each backend has a private copy of
296 * both in LogwrtResult, which is updated when convenient.
297 *
298 * The request bookkeeping is simpler: there is a shared XLogCtl->LogwrtRqst
299 * (protected by info_lck), but we don't need to cache any copies of it.
300 *
301 * info_lck is only held long enough to read/update the protected variables,
302 * so it's a plain spinlock. The other locks are held longer (potentially
303 * over I/O operations), so we use LWLocks for them. These locks are:
304 *
305 * WALBufMappingLock: must be held to replace a page in the WAL buffer cache.
306 * It is only held while initializing and changing the mapping. If the
307 * contents of the buffer being replaced haven't been written yet, the mapping
308 * lock is released while the write is done, and reacquired afterwards.
309 *
310 * WALWriteLock: must be held to write WAL buffers to disk (XLogWrite or
311 * XLogFlush).
312 *
313 * ControlFileLock: must be held to read/update control file or create
314 * new log file.
315 *
316 *----------
317 */
318
319typedef struct XLogwrtRqst
320{
321 XLogRecPtr Write; /* last byte + 1 to write out */
322 XLogRecPtr Flush; /* last byte + 1 to flush */
324
325typedef struct XLogwrtResult
326{
327 XLogRecPtr Write; /* last byte + 1 written out */
328 XLogRecPtr Flush; /* last byte + 1 flushed */
330
331/*
332 * Inserting to WAL is protected by a small fixed number of WAL insertion
333 * locks. To insert to the WAL, you must hold one of the locks - it doesn't
334 * matter which one. To lock out other concurrent insertions, you must hold
335 * of them. Each WAL insertion lock consists of a lightweight lock, plus an
336 * indicator of how far the insertion has progressed (insertingAt).
337 *
338 * The insertingAt values are read when a process wants to flush WAL from
339 * the in-memory buffers to disk, to check that all the insertions to the
340 * region the process is about to write out have finished. You could simply
341 * wait for all currently in-progress insertions to finish, but the
342 * insertingAt indicator allows you to ignore insertions to later in the WAL,
343 * so that you only wait for the insertions that are modifying the buffers
344 * you're about to write out.
345 *
346 * This isn't just an optimization. If all the WAL buffers are dirty, an
347 * inserter that's holding a WAL insert lock might need to evict an old WAL
348 * buffer, which requires flushing the WAL. If it's possible for an inserter
349 * to block on another inserter unnecessarily, deadlock can arise when two
350 * inserters holding a WAL insert lock wait for each other to finish their
351 * insertion.
352 *
353 * Small WAL records that don't cross a page boundary never update the value,
354 * the WAL record is just copied to the page and the lock is released. But
355 * to avoid the deadlock-scenario explained above, the indicator is always
356 * updated before sleeping while holding an insertion lock.
357 *
358 * lastImportantAt contains the LSN of the last important WAL record inserted
359 * using a given lock. This value is used to detect if there has been
360 * important WAL activity since the last time some action, like a checkpoint,
361 * was performed - allowing to not repeat the action if not. The LSN is
362 * updated for all insertions, unless the XLOG_MARK_UNIMPORTANT flag was
363 * set. lastImportantAt is never cleared, only overwritten by the LSN of newer
364 * records. Tracking the WAL activity directly in WALInsertLock has the
365 * advantage of not needing any additional locks to update the value.
366 */
367typedef struct
368{
373
374/*
375 * All the WAL insertion locks are allocated as an array in shared memory. We
376 * force the array stride to be a power of 2, which saves a few cycles in
377 * indexing, but more importantly also ensures that individual slots don't
378 * cross cache line boundaries. (Of course, we have to also ensure that the
379 * array start address is suitably aligned.)
380 */
382{
386
387/*
388 * Session status of running backup, used for sanity checks in SQL-callable
389 * functions to start and stop backups.
390 */
392
393/*
394 * Shared state data for WAL insertion.
395 */
396typedef struct XLogCtlInsert
397{
398 slock_t insertpos_lck; /* protects CurrBytePos and PrevBytePos */
399
400 /*
401 * CurrBytePos is the end of reserved WAL. The next record will be
402 * inserted at that position. PrevBytePos is the start position of the
403 * previously inserted (or rather, reserved) record - it is copied to the
404 * prev-link of the next record. These are stored as "usable byte
405 * positions" rather than XLogRecPtrs (see XLogBytePosToRecPtr()).
406 */
409
410 /*
411 * Make sure the above heavily-contended spinlock and byte positions are
412 * on their own cache line. In particular, the RedoRecPtr and full page
413 * write variables below should be on a different cache line. They are
414 * read on every WAL insertion, but updated rarely, and we don't want
415 * those reads to steal the cache line containing Curr/PrevBytePos.
416 */
418
419 /*
420 * fullPageWrites is the authoritative value used by all backends to
421 * determine whether to write full-page image to WAL. This shared value,
422 * instead of the process-local fullPageWrites, is required because, when
423 * full_page_writes is changed by SIGHUP, we must WAL-log it before it
424 * actually affects WAL-logging by backends. Checkpointer sets at startup
425 * or after SIGHUP.
426 *
427 * To read these fields, you must hold an insertion lock. To modify them,
428 * you must hold ALL the locks.
429 */
430 XLogRecPtr RedoRecPtr; /* current redo point for insertions */
432
433 /*
434 * runningBackups is a counter indicating the number of backups currently
435 * in progress. lastBackupStart is the latest checkpoint redo location
436 * used as a starting point for an online backup.
437 */
440
441 /*
442 * WAL insertion locks.
443 */
446
447/*
448 * Total shared-memory state for XLOG.
449 */
450typedef struct XLogCtlData
451{
453
454 /* Protected by info_lck: */
456 XLogRecPtr RedoRecPtr; /* a recent copy of Insert->RedoRecPtr */
457 FullTransactionId ckptFullXid; /* nextXid of latest checkpoint */
458 XLogRecPtr asyncXactLSN; /* LSN of newest async commit/abort */
459 XLogRecPtr replicationSlotMinLSN; /* oldest LSN needed by any slot */
460
461 XLogSegNo lastRemovedSegNo; /* latest removed/recycled XLOG segment */
462
463 /* Fake LSN counter, for unlogged relations. */
465
466 /* Time and LSN of last xlog segment switch. Protected by WALWriteLock. */
469
470 /* These are accessed using atomics -- info_lck not needed */
471 pg_atomic_uint64 logInsertResult; /* last byte + 1 inserted to buffers */
472 pg_atomic_uint64 logWriteResult; /* last byte + 1 written out */
473 pg_atomic_uint64 logFlushResult; /* last byte + 1 flushed */
474
475 /*
476 * Latest initialized page in the cache (last byte position + 1).
477 *
478 * To change the identity of a buffer (and InitializedUpTo), you need to
479 * hold WALBufMappingLock. To change the identity of a buffer that's
480 * still dirty, the old page needs to be written out first, and for that
481 * you need WALWriteLock, and you need to ensure that there are no
482 * in-progress insertions to the page by calling
483 * WaitXLogInsertionsToFinish().
484 */
486
487 /*
488 * These values do not change after startup, although the pointed-to pages
489 * and xlblocks values certainly do. xlblocks values are protected by
490 * WALBufMappingLock.
491 */
492 char *pages; /* buffers for unwritten XLOG pages */
493 pg_atomic_uint64 *xlblocks; /* 1st byte ptr-s + XLOG_BLCKSZ */
494 int XLogCacheBlck; /* highest allocated xlog buffer index */
495
496 /*
497 * InsertTimeLineID is the timeline into which new WAL is being inserted
498 * and flushed. It is zero during recovery, and does not change once set.
499 *
500 * If we create a new timeline when the system was started up,
501 * PrevTimeLineID is the old timeline's ID that we forked off from.
502 * Otherwise it's equal to InsertTimeLineID.
503 *
504 * We set these fields while holding info_lck. Most that reads these
505 * values knows that recovery is no longer in progress and so can safely
506 * read the value without a lock, but code that could be run either during
507 * or after recovery can take info_lck while reading these values.
508 */
511
512 /*
513 * SharedRecoveryState indicates if we're still in crash or archive
514 * recovery. Protected by info_lck.
515 */
517
518 /*
519 * InstallXLogFileSegmentActive indicates whether the checkpointer should
520 * arrange for future segments by recycling and/or PreallocXlogFiles().
521 * Protected by ControlFileLock. Only the startup process changes it. If
522 * true, anyone can use InstallXLogFileSegment(). If false, the startup
523 * process owns the exclusive right to install segments, by reading from
524 * the archive and possibly replacing existing files.
525 */
527
528 /*
529 * WalWriterSleeping indicates whether the WAL writer is currently in
530 * low-power mode (and hence should be nudged if an async commit occurs).
531 * Protected by info_lck.
532 */
534
535 /*
536 * During recovery, we keep a copy of the latest checkpoint record here.
537 * lastCheckPointRecPtr points to start of checkpoint record and
538 * lastCheckPointEndPtr points to end+1 of checkpoint record. Used by the
539 * checkpointer when it wants to create a restartpoint.
540 *
541 * Protected by info_lck.
542 */
546
547 /*
548 * lastFpwDisableRecPtr points to the start of the last replayed
549 * XLOG_FPW_CHANGE record that instructs full_page_writes is disabled.
550 */
552
553 slock_t info_lck; /* locks shared variables shown above */
555
556/*
557 * Classification of XLogInsertRecord operations.
558 */
559typedef enum
560{
565
566static XLogCtlData *XLogCtl = NULL;
567
568/* a private copy of XLogCtl->Insert.WALInsertLocks, for convenience */
570
571/*
572 * We maintain an image of pg_control in shared memory.
573 */
575
576/*
577 * Calculate the amount of space left on the page after 'endptr'. Beware
578 * multiple evaluation!
579 */
580#define INSERT_FREESPACE(endptr) \
581 (((endptr) % XLOG_BLCKSZ == 0) ? 0 : (XLOG_BLCKSZ - (endptr) % XLOG_BLCKSZ))
582
583/* Macro to advance to next buffer index. */
584#define NextBufIdx(idx) \
585 (((idx) == XLogCtl->XLogCacheBlck) ? 0 : ((idx) + 1))
586
587/*
588 * XLogRecPtrToBufIdx returns the index of the WAL buffer that holds, or
589 * would hold if it was in cache, the page containing 'recptr'.
590 */
591#define XLogRecPtrToBufIdx(recptr) \
592 (((recptr) / XLOG_BLCKSZ) % (XLogCtl->XLogCacheBlck + 1))
593
594/*
595 * These are the number of bytes in a WAL page usable for WAL data.
596 */
597#define UsableBytesInPage (XLOG_BLCKSZ - SizeOfXLogShortPHD)
598
599/*
600 * Convert values of GUCs measured in megabytes to equiv. segment count.
601 * Rounds down.
602 */
603#define ConvertToXSegs(x, segsize) XLogMBVarToSegs((x), (segsize))
604
605/* The number of bytes in a WAL segment usable for WAL data. */
607
608/*
609 * Private, possibly out-of-date copy of shared LogwrtResult.
610 * See discussion above.
611 */
613
614/*
615 * Update local copy of shared XLogCtl->log{Write,Flush}Result
616 *
617 * It's critical that Flush always trails Write, so the order of the reads is
618 * important, as is the barrier. See also XLogWrite.
619 */
620#define RefreshXLogWriteResult(_target) \
621 do { \
622 _target.Flush = pg_atomic_read_u64(&XLogCtl->logFlushResult); \
623 pg_read_barrier(); \
624 _target.Write = pg_atomic_read_u64(&XLogCtl->logWriteResult); \
625 } while (0)
626
627/*
628 * openLogFile is -1 or a kernel FD for an open log file segment.
629 * openLogSegNo identifies the segment, and openLogTLI the corresponding TLI.
630 * These variables are only used to write the XLOG, and so will normally refer
631 * to the active segment.
632 *
633 * Note: call Reserve/ReleaseExternalFD to track consumption of this FD.
634 */
635static int openLogFile = -1;
638
639/*
640 * Local copies of equivalent fields in the control file. When running
641 * crash recovery, LocalMinRecoveryPoint is set to InvalidXLogRecPtr as we
642 * expect to replay all the WAL available, and updateMinRecoveryPoint is
643 * switched to false to prevent any updates while replaying records.
644 * Those values are kept consistent as long as crash recovery runs.
645 */
648static bool updateMinRecoveryPoint = true;
649
650/* For WALInsertLockAcquire/Release functions */
651static int MyLockNo = 0;
652static bool holdingAllLocks = false;
653
654#ifdef WAL_DEBUG
655static MemoryContext walDebugCxt = NULL;
656#endif
657
658static void CleanupAfterArchiveRecovery(TimeLineID EndOfLogTLI,
659 XLogRecPtr EndOfLog,
660 TimeLineID newTLI);
661static void CheckRequiredParameterValues(void);
662static void XLogReportParameters(void);
663static int LocalSetXLogInsertAllowed(void);
664static void CreateEndOfRecoveryRecord(void);
666 XLogRecPtr pagePtr,
667 TimeLineID newTLI);
668static void CheckPointGuts(XLogRecPtr checkPointRedo, int flags);
669static void KeepLogSeg(XLogRecPtr recptr, XLogSegNo *logSegNo);
671
672static void AdvanceXLInsertBuffer(XLogRecPtr upto, TimeLineID tli,
673 bool opportunistic);
674static void XLogWrite(XLogwrtRqst WriteRqst, TimeLineID tli, bool flexible);
675static bool InstallXLogFileSegment(XLogSegNo *segno, char *tmppath,
676 bool find_free, XLogSegNo max_segno,
677 TimeLineID tli);
678static void XLogFileClose(void);
679static void PreallocXlogFiles(XLogRecPtr endptr, TimeLineID tli);
680static void RemoveTempXlogFiles(void);
681static void RemoveOldXlogFiles(XLogSegNo segno, XLogRecPtr lastredoptr,
682 XLogRecPtr endptr, TimeLineID insertTLI);
683static void RemoveXlogFile(const struct dirent *segment_de,
684 XLogSegNo recycleSegNo, XLogSegNo *endlogSegNo,
685 TimeLineID insertTLI);
686static void UpdateLastRemovedPtr(char *filename);
687static void ValidateXLOGDirectoryStructure(void);
688static void CleanupBackupHistory(void);
689static void UpdateMinRecoveryPoint(XLogRecPtr lsn, bool force);
690static bool PerformRecoveryXLogAction(void);
691static void InitControlFile(uint64 sysidentifier, uint32 data_checksum_version);
692static void WriteControlFile(void);
693static void ReadControlFile(void);
694static void UpdateControlFile(void);
695static char *str_time(pg_time_t tnow);
696
697static int get_sync_bit(int method);
698
699static void CopyXLogRecordToWAL(int write_len, bool isLogSwitch,
700 XLogRecData *rdata,
701 XLogRecPtr StartPos, XLogRecPtr EndPos,
702 TimeLineID tli);
703static void ReserveXLogInsertLocation(int size, XLogRecPtr *StartPos,
704 XLogRecPtr *EndPos, XLogRecPtr *PrevPtr);
705static bool ReserveXLogSwitch(XLogRecPtr *StartPos, XLogRecPtr *EndPos,
706 XLogRecPtr *PrevPtr);
708static char *GetXLogBuffer(XLogRecPtr ptr, TimeLineID tli);
712
713static void WALInsertLockAcquire(void);
714static void WALInsertLockAcquireExclusive(void);
715static void WALInsertLockRelease(void);
716static void WALInsertLockUpdateInsertingAt(XLogRecPtr insertingAt);
717
718/*
719 * Insert an XLOG record represented by an already-constructed chain of data
720 * chunks. This is a low-level routine; to construct the WAL record header
721 * and data, use the higher-level routines in xloginsert.c.
722 *
723 * If 'fpw_lsn' is valid, it is the oldest LSN among the pages that this
724 * WAL record applies to, that were not included in the record as full page
725 * images. If fpw_lsn <= RedoRecPtr, the function does not perform the
726 * insertion and returns InvalidXLogRecPtr. The caller can then recalculate
727 * which pages need a full-page image, and retry. If fpw_lsn is invalid, the
728 * record is always inserted.
729 *
730 * 'flags' gives more in-depth control on the record being inserted. See
731 * XLogSetRecordFlags() for details.
732 *
733 * 'topxid_included' tells whether the top-transaction id is logged along with
734 * current subtransaction. See XLogRecordAssemble().
735 *
736 * The first XLogRecData in the chain must be for the record header, and its
737 * data must be MAXALIGNed. XLogInsertRecord fills in the xl_prev and
738 * xl_crc fields in the header, the rest of the header must already be filled
739 * by the caller.
740 *
741 * Returns XLOG pointer to end of record (beginning of next record).
742 * This can be used as LSN for data pages affected by the logged action.
743 * (LSN is the XLOG point up to which the XLOG must be flushed to disk
744 * before the data page can be written out. This implements the basic
745 * WAL rule "write the log before the data".)
746 */
749 XLogRecPtr fpw_lsn,
750 uint8 flags,
751 int num_fpi,
752 bool topxid_included)
753{
755 pg_crc32c rdata_crc;
756 bool inserted;
757 XLogRecord *rechdr = (XLogRecord *) rdata->data;
758 uint8 info = rechdr->xl_info & ~XLR_INFO_MASK;
760 XLogRecPtr StartPos;
761 XLogRecPtr EndPos;
762 bool prevDoPageWrites = doPageWrites;
763 TimeLineID insertTLI;
764
765 /* Does this record type require special handling? */
766 if (unlikely(rechdr->xl_rmid == RM_XLOG_ID))
767 {
768 if (info == XLOG_SWITCH)
770 else if (info == XLOG_CHECKPOINT_REDO)
772 }
773
774 /* we assume that all of the record header is in the first chunk */
775 Assert(rdata->len >= SizeOfXLogRecord);
776
777 /* cross-check on whether we should be here or not */
778 if (!XLogInsertAllowed())
779 elog(ERROR, "cannot make new WAL entries during recovery");
780
781 /*
782 * Given that we're not in recovery, InsertTimeLineID is set and can't
783 * change, so we can read it without a lock.
784 */
785 insertTLI = XLogCtl->InsertTimeLineID;
786
787 /*----------
788 *
789 * We have now done all the preparatory work we can without holding a
790 * lock or modifying shared state. From here on, inserting the new WAL
791 * record to the shared WAL buffer cache is a two-step process:
792 *
793 * 1. Reserve the right amount of space from the WAL. The current head of
794 * reserved space is kept in Insert->CurrBytePos, and is protected by
795 * insertpos_lck.
796 *
797 * 2. Copy the record to the reserved WAL space. This involves finding the
798 * correct WAL buffer containing the reserved space, and copying the
799 * record in place. This can be done concurrently in multiple processes.
800 *
801 * To keep track of which insertions are still in-progress, each concurrent
802 * inserter acquires an insertion lock. In addition to just indicating that
803 * an insertion is in progress, the lock tells others how far the inserter
804 * has progressed. There is a small fixed number of insertion locks,
805 * determined by NUM_XLOGINSERT_LOCKS. When an inserter crosses a page
806 * boundary, it updates the value stored in the lock to the how far it has
807 * inserted, to allow the previous buffer to be flushed.
808 *
809 * Holding onto an insertion lock also protects RedoRecPtr and
810 * fullPageWrites from changing until the insertion is finished.
811 *
812 * Step 2 can usually be done completely in parallel. If the required WAL
813 * page is not initialized yet, you have to grab WALBufMappingLock to
814 * initialize it, but the WAL writer tries to do that ahead of insertions
815 * to avoid that from happening in the critical path.
816 *
817 *----------
818 */
820
821 if (likely(class == WALINSERT_NORMAL))
822 {
824
825 /*
826 * Check to see if my copy of RedoRecPtr is out of date. If so, may
827 * have to go back and have the caller recompute everything. This can
828 * only happen just after a checkpoint, so it's better to be slow in
829 * this case and fast otherwise.
830 *
831 * Also check to see if fullPageWrites was just turned on or there's a
832 * running backup (which forces full-page writes); if we weren't
833 * already doing full-page writes then go back and recompute.
834 *
835 * If we aren't doing full-page writes then RedoRecPtr doesn't
836 * actually affect the contents of the XLOG record, so we'll update
837 * our local copy but not force a recomputation. (If doPageWrites was
838 * just turned off, we could recompute the record without full pages,
839 * but we choose not to bother.)
840 */
841 if (RedoRecPtr != Insert->RedoRecPtr)
842 {
843 Assert(RedoRecPtr < Insert->RedoRecPtr);
844 RedoRecPtr = Insert->RedoRecPtr;
845 }
846 doPageWrites = (Insert->fullPageWrites || Insert->runningBackups > 0);
847
848 if (doPageWrites &&
849 (!prevDoPageWrites ||
850 (fpw_lsn != InvalidXLogRecPtr && fpw_lsn <= RedoRecPtr)))
851 {
852 /*
853 * Oops, some buffer now needs to be backed up that the caller
854 * didn't back up. Start over.
855 */
858 return InvalidXLogRecPtr;
859 }
860
861 /*
862 * Reserve space for the record in the WAL. This also sets the xl_prev
863 * pointer.
864 */
865 ReserveXLogInsertLocation(rechdr->xl_tot_len, &StartPos, &EndPos,
866 &rechdr->xl_prev);
867
868 /* Normal records are always inserted. */
869 inserted = true;
870 }
871 else if (class == WALINSERT_SPECIAL_SWITCH)
872 {
873 /*
874 * In order to insert an XLOG_SWITCH record, we need to hold all of
875 * the WAL insertion locks, not just one, so that no one else can
876 * begin inserting a record until we've figured out how much space
877 * remains in the current WAL segment and claimed all of it.
878 *
879 * Nonetheless, this case is simpler than the normal cases handled
880 * below, which must check for changes in doPageWrites and RedoRecPtr.
881 * Those checks are only needed for records that can contain buffer
882 * references, and an XLOG_SWITCH record never does.
883 */
884 Assert(fpw_lsn == InvalidXLogRecPtr);
886 inserted = ReserveXLogSwitch(&StartPos, &EndPos, &rechdr->xl_prev);
887 }
888 else
889 {
891
892 /*
893 * We need to update both the local and shared copies of RedoRecPtr,
894 * which means that we need to hold all the WAL insertion locks.
895 * However, there can't be any buffer references, so as above, we need
896 * not check RedoRecPtr before inserting the record; we just need to
897 * update it afterwards.
898 */
899 Assert(fpw_lsn == InvalidXLogRecPtr);
901 ReserveXLogInsertLocation(rechdr->xl_tot_len, &StartPos, &EndPos,
902 &rechdr->xl_prev);
903 RedoRecPtr = Insert->RedoRecPtr = StartPos;
904 inserted = true;
905 }
906
907 if (inserted)
908 {
909 /*
910 * Now that xl_prev has been filled in, calculate CRC of the record
911 * header.
912 */
913 rdata_crc = rechdr->xl_crc;
914 COMP_CRC32C(rdata_crc, rechdr, offsetof(XLogRecord, xl_crc));
915 FIN_CRC32C(rdata_crc);
916 rechdr->xl_crc = rdata_crc;
917
918 /*
919 * All the record data, including the header, is now ready to be
920 * inserted. Copy the record in the space reserved.
921 */
923 class == WALINSERT_SPECIAL_SWITCH, rdata,
924 StartPos, EndPos, insertTLI);
925
926 /*
927 * Unless record is flagged as not important, update LSN of last
928 * important record in the current slot. When holding all locks, just
929 * update the first one.
930 */
931 if ((flags & XLOG_MARK_UNIMPORTANT) == 0)
932 {
933 int lockno = holdingAllLocks ? 0 : MyLockNo;
934
935 WALInsertLocks[lockno].l.lastImportantAt = StartPos;
936 }
937 }
938 else
939 {
940 /*
941 * This was an xlog-switch record, but the current insert location was
942 * already exactly at the beginning of a segment, so there was no need
943 * to do anything.
944 */
945 }
946
947 /*
948 * Done! Let others know that we're finished.
949 */
951
953
955
956 /*
957 * Mark top transaction id is logged (if needed) so that we should not try
958 * to log it again with the next WAL record in the current subtransaction.
959 */
960 if (topxid_included)
962
963 /*
964 * Update shared LogwrtRqst.Write, if we crossed page boundary.
965 */
966 if (StartPos / XLOG_BLCKSZ != EndPos / XLOG_BLCKSZ)
967 {
969 /* advance global request to include new block(s) */
970 if (XLogCtl->LogwrtRqst.Write < EndPos)
971 XLogCtl->LogwrtRqst.Write = EndPos;
974 }
975
976 /*
977 * If this was an XLOG_SWITCH record, flush the record and the empty
978 * padding space that fills the rest of the segment, and perform
979 * end-of-segment actions (eg, notifying archiver).
980 */
981 if (class == WALINSERT_SPECIAL_SWITCH)
982 {
983 TRACE_POSTGRESQL_WAL_SWITCH();
984 XLogFlush(EndPos);
985
986 /*
987 * Even though we reserved the rest of the segment for us, which is
988 * reflected in EndPos, we return a pointer to just the end of the
989 * xlog-switch record.
990 */
991 if (inserted)
992 {
993 EndPos = StartPos + SizeOfXLogRecord;
994 if (StartPos / XLOG_BLCKSZ != EndPos / XLOG_BLCKSZ)
995 {
996 uint64 offset = XLogSegmentOffset(EndPos, wal_segment_size);
997
998 if (offset == EndPos % XLOG_BLCKSZ)
999 EndPos += SizeOfXLogLongPHD;
1000 else
1001 EndPos += SizeOfXLogShortPHD;
1002 }
1003 }
1004 }
1005
1006#ifdef WAL_DEBUG
1007 if (XLOG_DEBUG)
1008 {
1009 static XLogReaderState *debug_reader = NULL;
1010 XLogRecord *record;
1011 DecodedXLogRecord *decoded;
1013 StringInfoData recordBuf;
1014 char *errormsg = NULL;
1015 MemoryContext oldCxt;
1016
1017 oldCxt = MemoryContextSwitchTo(walDebugCxt);
1018
1020 appendStringInfo(&buf, "INSERT @ %X/%X: ", LSN_FORMAT_ARGS(EndPos));
1021
1022 /*
1023 * We have to piece together the WAL record data from the XLogRecData
1024 * entries, so that we can pass it to the rm_desc function as one
1025 * contiguous chunk.
1026 */
1027 initStringInfo(&recordBuf);
1028 for (; rdata != NULL; rdata = rdata->next)
1029 appendBinaryStringInfo(&recordBuf, rdata->data, rdata->len);
1030
1031 /* We also need temporary space to decode the record. */
1032 record = (XLogRecord *) recordBuf.data;
1033 decoded = (DecodedXLogRecord *)
1035
1036 if (!debug_reader)
1037 debug_reader = XLogReaderAllocate(wal_segment_size, NULL,
1038 XL_ROUTINE(.page_read = NULL,
1039 .segment_open = NULL,
1040 .segment_close = NULL),
1041 NULL);
1042 if (!debug_reader)
1043 {
1044 appendStringInfoString(&buf, "error decoding record: out of memory while allocating a WAL reading processor");
1045 }
1046 else if (!DecodeXLogRecord(debug_reader,
1047 decoded,
1048 record,
1049 EndPos,
1050 &errormsg))
1051 {
1052 appendStringInfo(&buf, "error decoding record: %s",
1053 errormsg ? errormsg : "no error message");
1054 }
1055 else
1056 {
1057 appendStringInfoString(&buf, " - ");
1058
1059 debug_reader->record = decoded;
1060 xlog_outdesc(&buf, debug_reader);
1061 debug_reader->record = NULL;
1062 }
1063 elog(LOG, "%s", buf.data);
1064
1065 pfree(decoded);
1066 pfree(buf.data);
1067 pfree(recordBuf.data);
1068 MemoryContextSwitchTo(oldCxt);
1069 }
1070#endif
1071
1072 /*
1073 * Update our global variables
1074 */
1075 ProcLastRecPtr = StartPos;
1076 XactLastRecEnd = EndPos;
1077
1078 /* Report WAL traffic to the instrumentation. */
1079 if (inserted)
1080 {
1081 pgWalUsage.wal_bytes += rechdr->xl_tot_len;
1083 pgWalUsage.wal_fpi += num_fpi;
1084 }
1085
1086 return EndPos;
1087}
1088
1089/*
1090 * Reserves the right amount of space for a record of given size from the WAL.
1091 * *StartPos is set to the beginning of the reserved section, *EndPos to
1092 * its end+1. *PrevPtr is set to the beginning of the previous record; it is
1093 * used to set the xl_prev of this record.
1094 *
1095 * This is the performance critical part of XLogInsert that must be serialized
1096 * across backends. The rest can happen mostly in parallel. Try to keep this
1097 * section as short as possible, insertpos_lck can be heavily contended on a
1098 * busy system.
1099 *
1100 * NB: The space calculation here must match the code in CopyXLogRecordToWAL,
1101 * where we actually copy the record to the reserved space.
1102 *
1103 * NB: Testing shows that XLogInsertRecord runs faster if this code is inlined;
1104 * however, because there are two call sites, the compiler is reluctant to
1105 * inline. We use pg_attribute_always_inline here to try to convince it.
1106 */
1109 XLogRecPtr *PrevPtr)
1110{
1112 uint64 startbytepos;
1113 uint64 endbytepos;
1114 uint64 prevbytepos;
1115
1116 size = MAXALIGN(size);
1117
1118 /* All (non xlog-switch) records should contain data. */
1120
1121 /*
1122 * The duration the spinlock needs to be held is minimized by minimizing
1123 * the calculations that have to be done while holding the lock. The
1124 * current tip of reserved WAL is kept in CurrBytePos, as a byte position
1125 * that only counts "usable" bytes in WAL, that is, it excludes all WAL
1126 * page headers. The mapping between "usable" byte positions and physical
1127 * positions (XLogRecPtrs) can be done outside the locked region, and
1128 * because the usable byte position doesn't include any headers, reserving
1129 * X bytes from WAL is almost as simple as "CurrBytePos += X".
1130 */
1131 SpinLockAcquire(&Insert->insertpos_lck);
1132
1133 startbytepos = Insert->CurrBytePos;
1134 endbytepos = startbytepos + size;
1135 prevbytepos = Insert->PrevBytePos;
1136 Insert->CurrBytePos = endbytepos;
1137 Insert->PrevBytePos = startbytepos;
1138
1139 SpinLockRelease(&Insert->insertpos_lck);
1140
1141 *StartPos = XLogBytePosToRecPtr(startbytepos);
1142 *EndPos = XLogBytePosToEndRecPtr(endbytepos);
1143 *PrevPtr = XLogBytePosToRecPtr(prevbytepos);
1144
1145 /*
1146 * Check that the conversions between "usable byte positions" and
1147 * XLogRecPtrs work consistently in both directions.
1148 */
1149 Assert(XLogRecPtrToBytePos(*StartPos) == startbytepos);
1150 Assert(XLogRecPtrToBytePos(*EndPos) == endbytepos);
1151 Assert(XLogRecPtrToBytePos(*PrevPtr) == prevbytepos);
1152}
1153
1154/*
1155 * Like ReserveXLogInsertLocation(), but for an xlog-switch record.
1156 *
1157 * A log-switch record is handled slightly differently. The rest of the
1158 * segment will be reserved for this insertion, as indicated by the returned
1159 * *EndPos value. However, if we are already at the beginning of the current
1160 * segment, *StartPos and *EndPos are set to the current location without
1161 * reserving any space, and the function returns false.
1162*/
1163static bool
1165{
1167 uint64 startbytepos;
1168 uint64 endbytepos;
1169 uint64 prevbytepos;
1171 XLogRecPtr ptr;
1172 uint32 segleft;
1173
1174 /*
1175 * These calculations are a bit heavy-weight to be done while holding a
1176 * spinlock, but since we're holding all the WAL insertion locks, there
1177 * are no other inserters competing for it. GetXLogInsertRecPtr() does
1178 * compete for it, but that's not called very frequently.
1179 */
1180 SpinLockAcquire(&Insert->insertpos_lck);
1181
1182 startbytepos = Insert->CurrBytePos;
1183
1184 ptr = XLogBytePosToEndRecPtr(startbytepos);
1185 if (XLogSegmentOffset(ptr, wal_segment_size) == 0)
1186 {
1187 SpinLockRelease(&Insert->insertpos_lck);
1188 *EndPos = *StartPos = ptr;
1189 return false;
1190 }
1191
1192 endbytepos = startbytepos + size;
1193 prevbytepos = Insert->PrevBytePos;
1194
1195 *StartPos = XLogBytePosToRecPtr(startbytepos);
1196 *EndPos = XLogBytePosToEndRecPtr(endbytepos);
1197
1199 if (segleft != wal_segment_size)
1200 {
1201 /* consume the rest of the segment */
1202 *EndPos += segleft;
1203 endbytepos = XLogRecPtrToBytePos(*EndPos);
1204 }
1205 Insert->CurrBytePos = endbytepos;
1206 Insert->PrevBytePos = startbytepos;
1207
1208 SpinLockRelease(&Insert->insertpos_lck);
1209
1210 *PrevPtr = XLogBytePosToRecPtr(prevbytepos);
1211
1213 Assert(XLogRecPtrToBytePos(*EndPos) == endbytepos);
1214 Assert(XLogRecPtrToBytePos(*StartPos) == startbytepos);
1215 Assert(XLogRecPtrToBytePos(*PrevPtr) == prevbytepos);
1216
1217 return true;
1218}
1219
1220/*
1221 * Subroutine of XLogInsertRecord. Copies a WAL record to an already-reserved
1222 * area in the WAL.
1223 */
1224static void
1225CopyXLogRecordToWAL(int write_len, bool isLogSwitch, XLogRecData *rdata,
1226 XLogRecPtr StartPos, XLogRecPtr EndPos, TimeLineID tli)
1227{
1228 char *currpos;
1229 int freespace;
1230 int written;
1231 XLogRecPtr CurrPos;
1232 XLogPageHeader pagehdr;
1233
1234 /*
1235 * Get a pointer to the right place in the right WAL buffer to start
1236 * inserting to.
1237 */
1238 CurrPos = StartPos;
1239 currpos = GetXLogBuffer(CurrPos, tli);
1240 freespace = INSERT_FREESPACE(CurrPos);
1241
1242 /*
1243 * there should be enough space for at least the first field (xl_tot_len)
1244 * on this page.
1245 */
1246 Assert(freespace >= sizeof(uint32));
1247
1248 /* Copy record data */
1249 written = 0;
1250 while (rdata != NULL)
1251 {
1252 const char *rdata_data = rdata->data;
1253 int rdata_len = rdata->len;
1254
1255 while (rdata_len > freespace)
1256 {
1257 /*
1258 * Write what fits on this page, and continue on the next page.
1259 */
1260 Assert(CurrPos % XLOG_BLCKSZ >= SizeOfXLogShortPHD || freespace == 0);
1261 memcpy(currpos, rdata_data, freespace);
1262 rdata_data += freespace;
1263 rdata_len -= freespace;
1264 written += freespace;
1265 CurrPos += freespace;
1266
1267 /*
1268 * Get pointer to beginning of next page, and set the xlp_rem_len
1269 * in the page header. Set XLP_FIRST_IS_CONTRECORD.
1270 *
1271 * It's safe to set the contrecord flag and xlp_rem_len without a
1272 * lock on the page. All the other flags were already set when the
1273 * page was initialized, in AdvanceXLInsertBuffer, and we're the
1274 * only backend that needs to set the contrecord flag.
1275 */
1276 currpos = GetXLogBuffer(CurrPos, tli);
1277 pagehdr = (XLogPageHeader) currpos;
1278 pagehdr->xlp_rem_len = write_len - written;
1280
1281 /* skip over the page header */
1282 if (XLogSegmentOffset(CurrPos, wal_segment_size) == 0)
1283 {
1284 CurrPos += SizeOfXLogLongPHD;
1285 currpos += SizeOfXLogLongPHD;
1286 }
1287 else
1288 {
1289 CurrPos += SizeOfXLogShortPHD;
1290 currpos += SizeOfXLogShortPHD;
1291 }
1292 freespace = INSERT_FREESPACE(CurrPos);
1293 }
1294
1295 Assert(CurrPos % XLOG_BLCKSZ >= SizeOfXLogShortPHD || rdata_len == 0);
1296 memcpy(currpos, rdata_data, rdata_len);
1297 currpos += rdata_len;
1298 CurrPos += rdata_len;
1299 freespace -= rdata_len;
1300 written += rdata_len;
1301
1302 rdata = rdata->next;
1303 }
1304 Assert(written == write_len);
1305
1306 /*
1307 * If this was an xlog-switch, it's not enough to write the switch record,
1308 * we also have to consume all the remaining space in the WAL segment. We
1309 * have already reserved that space, but we need to actually fill it.
1310 */
1311 if (isLogSwitch && XLogSegmentOffset(CurrPos, wal_segment_size) != 0)
1312 {
1313 /* An xlog-switch record doesn't contain any data besides the header */
1314 Assert(write_len == SizeOfXLogRecord);
1315
1316 /* Assert that we did reserve the right amount of space */
1318
1319 /* Use up all the remaining space on the current page */
1320 CurrPos += freespace;
1321
1322 /*
1323 * Cause all remaining pages in the segment to be flushed, leaving the
1324 * XLog position where it should be, at the start of the next segment.
1325 * We do this one page at a time, to make sure we don't deadlock
1326 * against ourselves if wal_buffers < wal_segment_size.
1327 */
1328 while (CurrPos < EndPos)
1329 {
1330 /*
1331 * The minimal action to flush the page would be to call
1332 * WALInsertLockUpdateInsertingAt(CurrPos) followed by
1333 * AdvanceXLInsertBuffer(...). The page would be left initialized
1334 * mostly to zeros, except for the page header (always the short
1335 * variant, as this is never a segment's first page).
1336 *
1337 * The large vistas of zeros are good for compressibility, but the
1338 * headers interrupting them every XLOG_BLCKSZ (with values that
1339 * differ from page to page) are not. The effect varies with
1340 * compression tool, but bzip2 for instance compresses about an
1341 * order of magnitude worse if those headers are left in place.
1342 *
1343 * Rather than complicating AdvanceXLInsertBuffer itself (which is
1344 * called in heavily-loaded circumstances as well as this lightly-
1345 * loaded one) with variant behavior, we just use GetXLogBuffer
1346 * (which itself calls the two methods we need) to get the pointer
1347 * and zero most of the page. Then we just zero the page header.
1348 */
1349 currpos = GetXLogBuffer(CurrPos, tli);
1350 MemSet(currpos, 0, SizeOfXLogShortPHD);
1351
1352 CurrPos += XLOG_BLCKSZ;
1353 }
1354 }
1355 else
1356 {
1357 /* Align the end position, so that the next record starts aligned */
1358 CurrPos = MAXALIGN64(CurrPos);
1359 }
1360
1361 if (CurrPos != EndPos)
1362 ereport(PANIC,
1364 errmsg_internal("space reserved for WAL record does not match what was written"));
1365}
1366
1367/*
1368 * Acquire a WAL insertion lock, for inserting to WAL.
1369 */
1370static void
1372{
1373 bool immed;
1374
1375 /*
1376 * It doesn't matter which of the WAL insertion locks we acquire, so try
1377 * the one we used last time. If the system isn't particularly busy, it's
1378 * a good bet that it's still available, and it's good to have some
1379 * affinity to a particular lock so that you don't unnecessarily bounce
1380 * cache lines between processes when there's no contention.
1381 *
1382 * If this is the first time through in this backend, pick a lock
1383 * (semi-)randomly. This allows the locks to be used evenly if you have a
1384 * lot of very short connections.
1385 */
1386 static int lockToTry = -1;
1387
1388 if (lockToTry == -1)
1389 lockToTry = MyProcNumber % NUM_XLOGINSERT_LOCKS;
1390 MyLockNo = lockToTry;
1391
1392 /*
1393 * The insertingAt value is initially set to 0, as we don't know our
1394 * insert location yet.
1395 */
1397 if (!immed)
1398 {
1399 /*
1400 * If we couldn't get the lock immediately, try another lock next
1401 * time. On a system with more insertion locks than concurrent
1402 * inserters, this causes all the inserters to eventually migrate to a
1403 * lock that no-one else is using. On a system with more inserters
1404 * than locks, it still helps to distribute the inserters evenly
1405 * across the locks.
1406 */
1407 lockToTry = (lockToTry + 1) % NUM_XLOGINSERT_LOCKS;
1408 }
1409}
1410
1411/*
1412 * Acquire all WAL insertion locks, to prevent other backends from inserting
1413 * to WAL.
1414 */
1415static void
1417{
1418 int i;
1419
1420 /*
1421 * When holding all the locks, all but the last lock's insertingAt
1422 * indicator is set to 0xFFFFFFFFFFFFFFFF, which is higher than any real
1423 * XLogRecPtr value, to make sure that no-one blocks waiting on those.
1424 */
1425 for (i = 0; i < NUM_XLOGINSERT_LOCKS - 1; i++)
1426 {
1431 }
1432 /* Variable value reset to 0 at release */
1434
1435 holdingAllLocks = true;
1436}
1437
1438/*
1439 * Release our insertion lock (or locks, if we're holding them all).
1440 *
1441 * NB: Reset all variables to 0, so they cause LWLockWaitForVar to block the
1442 * next time the lock is acquired.
1443 */
1444static void
1446{
1447 if (holdingAllLocks)
1448 {
1449 int i;
1450
1451 for (i = 0; i < NUM_XLOGINSERT_LOCKS; i++)
1454 0);
1455
1456 holdingAllLocks = false;
1457 }
1458 else
1459 {
1462 0);
1463 }
1464}
1465
1466/*
1467 * Update our insertingAt value, to let others know that we've finished
1468 * inserting up to that point.
1469 */
1470static void
1472{
1473 if (holdingAllLocks)
1474 {
1475 /*
1476 * We use the last lock to mark our actual position, see comments in
1477 * WALInsertLockAcquireExclusive.
1478 */
1481 insertingAt);
1482 }
1483 else
1486 insertingAt);
1487}
1488
1489/*
1490 * Wait for any WAL insertions < upto to finish.
1491 *
1492 * Returns the location of the oldest insertion that is still in-progress.
1493 * Any WAL prior to that point has been fully copied into WAL buffers, and
1494 * can be flushed out to disk. Because this waits for any insertions older
1495 * than 'upto' to finish, the return value is always >= 'upto'.
1496 *
1497 * Note: When you are about to write out WAL, you must call this function
1498 * *before* acquiring WALWriteLock, to avoid deadlocks. This function might
1499 * need to wait for an insertion to finish (or at least advance to next
1500 * uninitialized page), and the inserter might need to evict an old WAL buffer
1501 * to make room for a new one, which in turn requires WALWriteLock.
1502 */
1503static XLogRecPtr
1505{
1506 uint64 bytepos;
1507 XLogRecPtr inserted;
1508 XLogRecPtr reservedUpto;
1509 XLogRecPtr finishedUpto;
1511 int i;
1512
1513 if (MyProc == NULL)
1514 elog(PANIC, "cannot wait without a PGPROC structure");
1515
1516 /*
1517 * Check if there's any work to do. Use a barrier to ensure we get the
1518 * freshest value.
1519 */
1521 if (upto <= inserted)
1522 return inserted;
1523
1524 /* Read the current insert position */
1525 SpinLockAcquire(&Insert->insertpos_lck);
1526 bytepos = Insert->CurrBytePos;
1527 SpinLockRelease(&Insert->insertpos_lck);
1528 reservedUpto = XLogBytePosToEndRecPtr(bytepos);
1529
1530 /*
1531 * No-one should request to flush a piece of WAL that hasn't even been
1532 * reserved yet. However, it can happen if there is a block with a bogus
1533 * LSN on disk, for example. XLogFlush checks for that situation and
1534 * complains, but only after the flush. Here we just assume that to mean
1535 * that all WAL that has been reserved needs to be finished. In this
1536 * corner-case, the return value can be smaller than 'upto' argument.
1537 */
1538 if (upto > reservedUpto)
1539 {
1540 ereport(LOG,
1541 (errmsg("request to flush past end of generated WAL; request %X/%X, current position %X/%X",
1542 LSN_FORMAT_ARGS(upto), LSN_FORMAT_ARGS(reservedUpto))));
1543 upto = reservedUpto;
1544 }
1545
1546 /*
1547 * Loop through all the locks, sleeping on any in-progress insert older
1548 * than 'upto'.
1549 *
1550 * finishedUpto is our return value, indicating the point upto which all
1551 * the WAL insertions have been finished. Initialize it to the head of
1552 * reserved WAL, and as we iterate through the insertion locks, back it
1553 * out for any insertion that's still in progress.
1554 */
1555 finishedUpto = reservedUpto;
1556 for (i = 0; i < NUM_XLOGINSERT_LOCKS; i++)
1557 {
1558 XLogRecPtr insertingat = InvalidXLogRecPtr;
1559
1560 do
1561 {
1562 /*
1563 * See if this insertion is in progress. LWLockWaitForVar will
1564 * wait for the lock to be released, or for the 'value' to be set
1565 * by a LWLockUpdateVar call. When a lock is initially acquired,
1566 * its value is 0 (InvalidXLogRecPtr), which means that we don't
1567 * know where it's inserting yet. We will have to wait for it. If
1568 * it's a small insertion, the record will most likely fit on the
1569 * same page and the inserter will release the lock without ever
1570 * calling LWLockUpdateVar. But if it has to sleep, it will
1571 * advertise the insertion point with LWLockUpdateVar before
1572 * sleeping.
1573 *
1574 * In this loop we are only waiting for insertions that started
1575 * before WaitXLogInsertionsToFinish was called. The lack of
1576 * memory barriers in the loop means that we might see locks as
1577 * "unused" that have since become used. This is fine because
1578 * they only can be used for later insertions that we would not
1579 * want to wait on anyway. Not taking a lock to acquire the
1580 * current insertingAt value means that we might see older
1581 * insertingAt values. This is also fine, because if we read a
1582 * value too old, we will add ourselves to the wait queue, which
1583 * contains atomic operations.
1584 */
1585 if (LWLockWaitForVar(&WALInsertLocks[i].l.lock,
1587 insertingat, &insertingat))
1588 {
1589 /* the lock was free, so no insertion in progress */
1590 insertingat = InvalidXLogRecPtr;
1591 break;
1592 }
1593
1594 /*
1595 * This insertion is still in progress. Have to wait, unless the
1596 * inserter has proceeded past 'upto'.
1597 */
1598 } while (insertingat < upto);
1599
1600 if (insertingat != InvalidXLogRecPtr && insertingat < finishedUpto)
1601 finishedUpto = insertingat;
1602 }
1603
1604 /*
1605 * Advance the limit we know to have been inserted and return the freshest
1606 * value we know of, which might be beyond what we requested if somebody
1607 * is concurrently doing this with an 'upto' pointer ahead of us.
1608 */
1610 finishedUpto);
1611
1612 return finishedUpto;
1613}
1614
1615/*
1616 * Get a pointer to the right location in the WAL buffer containing the
1617 * given XLogRecPtr.
1618 *
1619 * If the page is not initialized yet, it is initialized. That might require
1620 * evicting an old dirty buffer from the buffer cache, which means I/O.
1621 *
1622 * The caller must ensure that the page containing the requested location
1623 * isn't evicted yet, and won't be evicted. The way to ensure that is to
1624 * hold onto a WAL insertion lock with the insertingAt position set to
1625 * something <= ptr. GetXLogBuffer() will update insertingAt if it needs
1626 * to evict an old page from the buffer. (This means that once you call
1627 * GetXLogBuffer() with a given 'ptr', you must not access anything before
1628 * that point anymore, and must not call GetXLogBuffer() with an older 'ptr'
1629 * later, because older buffers might be recycled already)
1630 */
1631static char *
1633{
1634 int idx;
1635 XLogRecPtr endptr;
1636 static uint64 cachedPage = 0;
1637 static char *cachedPos = NULL;
1638 XLogRecPtr expectedEndPtr;
1639
1640 /*
1641 * Fast path for the common case that we need to access again the same
1642 * page as last time.
1643 */
1644 if (ptr / XLOG_BLCKSZ == cachedPage)
1645 {
1646 Assert(((XLogPageHeader) cachedPos)->xlp_magic == XLOG_PAGE_MAGIC);
1647 Assert(((XLogPageHeader) cachedPos)->xlp_pageaddr == ptr - (ptr % XLOG_BLCKSZ));
1648 return cachedPos + ptr % XLOG_BLCKSZ;
1649 }
1650
1651 /*
1652 * The XLog buffer cache is organized so that a page is always loaded to a
1653 * particular buffer. That way we can easily calculate the buffer a given
1654 * page must be loaded into, from the XLogRecPtr alone.
1655 */
1656 idx = XLogRecPtrToBufIdx(ptr);
1657
1658 /*
1659 * See what page is loaded in the buffer at the moment. It could be the
1660 * page we're looking for, or something older. It can't be anything newer
1661 * - that would imply the page we're looking for has already been written
1662 * out to disk and evicted, and the caller is responsible for making sure
1663 * that doesn't happen.
1664 *
1665 * We don't hold a lock while we read the value. If someone is just about
1666 * to initialize or has just initialized the page, it's possible that we
1667 * get InvalidXLogRecPtr. That's ok, we'll grab the mapping lock (in
1668 * AdvanceXLInsertBuffer) and retry if we see anything other than the page
1669 * we're looking for.
1670 */
1671 expectedEndPtr = ptr;
1672 expectedEndPtr += XLOG_BLCKSZ - ptr % XLOG_BLCKSZ;
1673
1675 if (expectedEndPtr != endptr)
1676 {
1677 XLogRecPtr initializedUpto;
1678
1679 /*
1680 * Before calling AdvanceXLInsertBuffer(), which can block, let others
1681 * know how far we're finished with inserting the record.
1682 *
1683 * NB: If 'ptr' points to just after the page header, advertise a
1684 * position at the beginning of the page rather than 'ptr' itself. If
1685 * there are no other insertions running, someone might try to flush
1686 * up to our advertised location. If we advertised a position after
1687 * the page header, someone might try to flush the page header, even
1688 * though page might actually not be initialized yet. As the first
1689 * inserter on the page, we are effectively responsible for making
1690 * sure that it's initialized, before we let insertingAt to move past
1691 * the page header.
1692 */
1693 if (ptr % XLOG_BLCKSZ == SizeOfXLogShortPHD &&
1694 XLogSegmentOffset(ptr, wal_segment_size) > XLOG_BLCKSZ)
1695 initializedUpto = ptr - SizeOfXLogShortPHD;
1696 else if (ptr % XLOG_BLCKSZ == SizeOfXLogLongPHD &&
1697 XLogSegmentOffset(ptr, wal_segment_size) < XLOG_BLCKSZ)
1698 initializedUpto = ptr - SizeOfXLogLongPHD;
1699 else
1700 initializedUpto = ptr;
1701
1702 WALInsertLockUpdateInsertingAt(initializedUpto);
1703
1704 AdvanceXLInsertBuffer(ptr, tli, false);
1706
1707 if (expectedEndPtr != endptr)
1708 elog(PANIC, "could not find WAL buffer for %X/%X",
1709 LSN_FORMAT_ARGS(ptr));
1710 }
1711 else
1712 {
1713 /*
1714 * Make sure the initialization of the page is visible to us, and
1715 * won't arrive later to overwrite the WAL data we write on the page.
1716 */
1718 }
1719
1720 /*
1721 * Found the buffer holding this page. Return a pointer to the right
1722 * offset within the page.
1723 */
1724 cachedPage = ptr / XLOG_BLCKSZ;
1725 cachedPos = XLogCtl->pages + idx * (Size) XLOG_BLCKSZ;
1726
1727 Assert(((XLogPageHeader) cachedPos)->xlp_magic == XLOG_PAGE_MAGIC);
1728 Assert(((XLogPageHeader) cachedPos)->xlp_pageaddr == ptr - (ptr % XLOG_BLCKSZ));
1729
1730 return cachedPos + ptr % XLOG_BLCKSZ;
1731}
1732
1733/*
1734 * Read WAL data directly from WAL buffers, if available. Returns the number
1735 * of bytes read successfully.
1736 *
1737 * Fewer than 'count' bytes may be read if some of the requested WAL data has
1738 * already been evicted.
1739 *
1740 * No locks are taken.
1741 *
1742 * Caller should ensure that it reads no further than LogwrtResult.Write
1743 * (which should have been updated by the caller when determining how far to
1744 * read). The 'tli' argument is only used as a convenient safety check so that
1745 * callers do not read from WAL buffers on a historical timeline.
1746 */
1747Size
1748WALReadFromBuffers(char *dstbuf, XLogRecPtr startptr, Size count,
1749 TimeLineID tli)
1750{
1751 char *pdst = dstbuf;
1752 XLogRecPtr recptr = startptr;
1753 XLogRecPtr inserted;
1754 Size nbytes = count;
1755
1757 return 0;
1758
1759 Assert(!XLogRecPtrIsInvalid(startptr));
1760
1761 /*
1762 * Caller should ensure that the requested data has been inserted into WAL
1763 * buffers before we try to read it.
1764 */
1766 if (startptr + count > inserted)
1767 ereport(ERROR,
1768 errmsg("cannot read past end of generated WAL: requested %X/%X, current position %X/%X",
1769 LSN_FORMAT_ARGS(startptr + count),
1770 LSN_FORMAT_ARGS(inserted)));
1771
1772 /*
1773 * Loop through the buffers without a lock. For each buffer, atomically
1774 * read and verify the end pointer, then copy the data out, and finally
1775 * re-read and re-verify the end pointer.
1776 *
1777 * Once a page is evicted, it never returns to the WAL buffers, so if the
1778 * end pointer matches the expected end pointer before and after we copy
1779 * the data, then the right page must have been present during the data
1780 * copy. Read barriers are necessary to ensure that the data copy actually
1781 * happens between the two verification steps.
1782 *
1783 * If either verification fails, we simply terminate the loop and return
1784 * with the data that had been already copied out successfully.
1785 */
1786 while (nbytes > 0)
1787 {
1788 uint32 offset = recptr % XLOG_BLCKSZ;
1789 int idx = XLogRecPtrToBufIdx(recptr);
1790 XLogRecPtr expectedEndPtr;
1791 XLogRecPtr endptr;
1792 const char *page;
1793 const char *psrc;
1794 Size npagebytes;
1795
1796 /*
1797 * Calculate the end pointer we expect in the xlblocks array if the
1798 * correct page is present.
1799 */
1800 expectedEndPtr = recptr + (XLOG_BLCKSZ - offset);
1801
1802 /*
1803 * First verification step: check that the correct page is present in
1804 * the WAL buffers.
1805 */
1807 if (expectedEndPtr != endptr)
1808 break;
1809
1810 /*
1811 * The correct page is present (or was at the time the endptr was
1812 * read; must re-verify later). Calculate pointer to source data and
1813 * determine how much data to read from this page.
1814 */
1815 page = XLogCtl->pages + idx * (Size) XLOG_BLCKSZ;
1816 psrc = page + offset;
1817 npagebytes = Min(nbytes, XLOG_BLCKSZ - offset);
1818
1819 /*
1820 * Ensure that the data copy and the first verification step are not
1821 * reordered.
1822 */
1824
1825 /* data copy */
1826 memcpy(pdst, psrc, npagebytes);
1827
1828 /*
1829 * Ensure that the data copy and the second verification step are not
1830 * reordered.
1831 */
1833
1834 /*
1835 * Second verification step: check that the page we read from wasn't
1836 * evicted while we were copying the data.
1837 */
1839 if (expectedEndPtr != endptr)
1840 break;
1841
1842 pdst += npagebytes;
1843 recptr += npagebytes;
1844 nbytes -= npagebytes;
1845 }
1846
1847 Assert(pdst - dstbuf <= count);
1848
1849 return pdst - dstbuf;
1850}
1851
1852/*
1853 * Converts a "usable byte position" to XLogRecPtr. A usable byte position
1854 * is the position starting from the beginning of WAL, excluding all WAL
1855 * page headers.
1856 */
1857static XLogRecPtr
1859{
1860 uint64 fullsegs;
1861 uint64 fullpages;
1862 uint64 bytesleft;
1863 uint32 seg_offset;
1864 XLogRecPtr result;
1865
1866 fullsegs = bytepos / UsableBytesInSegment;
1867 bytesleft = bytepos % UsableBytesInSegment;
1868
1869 if (bytesleft < XLOG_BLCKSZ - SizeOfXLogLongPHD)
1870 {
1871 /* fits on first page of segment */
1872 seg_offset = bytesleft + SizeOfXLogLongPHD;
1873 }
1874 else
1875 {
1876 /* account for the first page on segment with long header */
1877 seg_offset = XLOG_BLCKSZ;
1878 bytesleft -= XLOG_BLCKSZ - SizeOfXLogLongPHD;
1879
1880 fullpages = bytesleft / UsableBytesInPage;
1881 bytesleft = bytesleft % UsableBytesInPage;
1882
1883 seg_offset += fullpages * XLOG_BLCKSZ + bytesleft + SizeOfXLogShortPHD;
1884 }
1885
1886 XLogSegNoOffsetToRecPtr(fullsegs, seg_offset, wal_segment_size, result);
1887
1888 return result;
1889}
1890
1891/*
1892 * Like XLogBytePosToRecPtr, but if the position is at a page boundary,
1893 * returns a pointer to the beginning of the page (ie. before page header),
1894 * not to where the first xlog record on that page would go to. This is used
1895 * when converting a pointer to the end of a record.
1896 */
1897static XLogRecPtr
1899{
1900 uint64 fullsegs;
1901 uint64 fullpages;
1902 uint64 bytesleft;
1903 uint32 seg_offset;
1904 XLogRecPtr result;
1905
1906 fullsegs = bytepos / UsableBytesInSegment;
1907 bytesleft = bytepos % UsableBytesInSegment;
1908
1909 if (bytesleft < XLOG_BLCKSZ - SizeOfXLogLongPHD)
1910 {
1911 /* fits on first page of segment */
1912 if (bytesleft == 0)
1913 seg_offset = 0;
1914 else
1915 seg_offset = bytesleft + SizeOfXLogLongPHD;
1916 }
1917 else
1918 {
1919 /* account for the first page on segment with long header */
1920 seg_offset = XLOG_BLCKSZ;
1921 bytesleft -= XLOG_BLCKSZ - SizeOfXLogLongPHD;
1922
1923 fullpages = bytesleft / UsableBytesInPage;
1924 bytesleft = bytesleft % UsableBytesInPage;
1925
1926 if (bytesleft == 0)
1927 seg_offset += fullpages * XLOG_BLCKSZ + bytesleft;
1928 else
1929 seg_offset += fullpages * XLOG_BLCKSZ + bytesleft + SizeOfXLogShortPHD;
1930 }
1931
1932 XLogSegNoOffsetToRecPtr(fullsegs, seg_offset, wal_segment_size, result);
1933
1934 return result;
1935}
1936
1937/*
1938 * Convert an XLogRecPtr to a "usable byte position".
1939 */
1940static uint64
1942{
1943 uint64 fullsegs;
1944 uint32 fullpages;
1945 uint32 offset;
1946 uint64 result;
1947
1948 XLByteToSeg(ptr, fullsegs, wal_segment_size);
1949
1950 fullpages = (XLogSegmentOffset(ptr, wal_segment_size)) / XLOG_BLCKSZ;
1951 offset = ptr % XLOG_BLCKSZ;
1952
1953 if (fullpages == 0)
1954 {
1955 result = fullsegs * UsableBytesInSegment;
1956 if (offset > 0)
1957 {
1958 Assert(offset >= SizeOfXLogLongPHD);
1959 result += offset - SizeOfXLogLongPHD;
1960 }
1961 }
1962 else
1963 {
1964 result = fullsegs * UsableBytesInSegment +
1965 (XLOG_BLCKSZ - SizeOfXLogLongPHD) + /* account for first page */
1966 (fullpages - 1) * UsableBytesInPage; /* full pages */
1967 if (offset > 0)
1968 {
1969 Assert(offset >= SizeOfXLogShortPHD);
1970 result += offset - SizeOfXLogShortPHD;
1971 }
1972 }
1973
1974 return result;
1975}
1976
1977/*
1978 * Initialize XLOG buffers, writing out old buffers if they still contain
1979 * unwritten data, upto the page containing 'upto'. Or if 'opportunistic' is
1980 * true, initialize as many pages as we can without having to write out
1981 * unwritten data. Any new pages are initialized to zeros, with pages headers
1982 * initialized properly.
1983 */
1984static void
1985AdvanceXLInsertBuffer(XLogRecPtr upto, TimeLineID tli, bool opportunistic)
1986{
1988 int nextidx;
1989 XLogRecPtr OldPageRqstPtr;
1990 XLogwrtRqst WriteRqst;
1991 XLogRecPtr NewPageEndPtr = InvalidXLogRecPtr;
1992 XLogRecPtr NewPageBeginPtr;
1993 XLogPageHeader NewPage;
1994 int npages pg_attribute_unused() = 0;
1995
1996 LWLockAcquire(WALBufMappingLock, LW_EXCLUSIVE);
1997
1998 /*
1999 * Now that we have the lock, check if someone initialized the page
2000 * already.
2001 */
2002 while (upto >= XLogCtl->InitializedUpTo || opportunistic)
2003 {
2005
2006 /*
2007 * Get ending-offset of the buffer page we need to replace (this may
2008 * be zero if the buffer hasn't been used yet). Fall through if it's
2009 * already written out.
2010 */
2011 OldPageRqstPtr = pg_atomic_read_u64(&XLogCtl->xlblocks[nextidx]);
2012 if (LogwrtResult.Write < OldPageRqstPtr)
2013 {
2014 /*
2015 * Nope, got work to do. If we just want to pre-initialize as much
2016 * as we can without flushing, give up now.
2017 */
2018 if (opportunistic)
2019 break;
2020
2021 /* Advance shared memory write request position */
2023 if (XLogCtl->LogwrtRqst.Write < OldPageRqstPtr)
2024 XLogCtl->LogwrtRqst.Write = OldPageRqstPtr;
2026
2027 /*
2028 * Acquire an up-to-date LogwrtResult value and see if we still
2029 * need to write it or if someone else already did.
2030 */
2032 if (LogwrtResult.Write < OldPageRqstPtr)
2033 {
2034 /*
2035 * Must acquire write lock. Release WALBufMappingLock first,
2036 * to make sure that all insertions that we need to wait for
2037 * can finish (up to this same position). Otherwise we risk
2038 * deadlock.
2039 */
2040 LWLockRelease(WALBufMappingLock);
2041
2042 WaitXLogInsertionsToFinish(OldPageRqstPtr);
2043
2044 LWLockAcquire(WALWriteLock, LW_EXCLUSIVE);
2045
2047 if (LogwrtResult.Write >= OldPageRqstPtr)
2048 {
2049 /* OK, someone wrote it already */
2050 LWLockRelease(WALWriteLock);
2051 }
2052 else
2053 {
2054 /* Have to write it ourselves */
2055 TRACE_POSTGRESQL_WAL_BUFFER_WRITE_DIRTY_START();
2056 WriteRqst.Write = OldPageRqstPtr;
2057 WriteRqst.Flush = 0;
2058 XLogWrite(WriteRqst, tli, false);
2059 LWLockRelease(WALWriteLock);
2061 TRACE_POSTGRESQL_WAL_BUFFER_WRITE_DIRTY_DONE();
2062 }
2063 /* Re-acquire WALBufMappingLock and retry */
2064 LWLockAcquire(WALBufMappingLock, LW_EXCLUSIVE);
2065 continue;
2066 }
2067 }
2068
2069 /*
2070 * Now the next buffer slot is free and we can set it up to be the
2071 * next output page.
2072 */
2073 NewPageBeginPtr = XLogCtl->InitializedUpTo;
2074 NewPageEndPtr = NewPageBeginPtr + XLOG_BLCKSZ;
2075
2076 Assert(XLogRecPtrToBufIdx(NewPageBeginPtr) == nextidx);
2077
2078 NewPage = (XLogPageHeader) (XLogCtl->pages + nextidx * (Size) XLOG_BLCKSZ);
2079
2080 /*
2081 * Mark the xlblock with InvalidXLogRecPtr and issue a write barrier
2082 * before initializing. Otherwise, the old page may be partially
2083 * zeroed but look valid.
2084 */
2087
2088 /*
2089 * Be sure to re-zero the buffer so that bytes beyond what we've
2090 * written will look like zeroes and not valid XLOG records...
2091 */
2092 MemSet((char *) NewPage, 0, XLOG_BLCKSZ);
2093
2094 /*
2095 * Fill the new page's header
2096 */
2097 NewPage->xlp_magic = XLOG_PAGE_MAGIC;
2098
2099 /* NewPage->xlp_info = 0; */ /* done by memset */
2100 NewPage->xlp_tli = tli;
2101 NewPage->xlp_pageaddr = NewPageBeginPtr;
2102
2103 /* NewPage->xlp_rem_len = 0; */ /* done by memset */
2104
2105 /*
2106 * If online backup is not in progress, mark the header to indicate
2107 * that WAL records beginning in this page have removable backup
2108 * blocks. This allows the WAL archiver to know whether it is safe to
2109 * compress archived WAL data by transforming full-block records into
2110 * the non-full-block format. It is sufficient to record this at the
2111 * page level because we force a page switch (in fact a segment
2112 * switch) when starting a backup, so the flag will be off before any
2113 * records can be written during the backup. At the end of a backup,
2114 * the last page will be marked as all unsafe when perhaps only part
2115 * is unsafe, but at worst the archiver would miss the opportunity to
2116 * compress a few records.
2117 */
2118 if (Insert->runningBackups == 0)
2119 NewPage->xlp_info |= XLP_BKP_REMOVABLE;
2120
2121 /*
2122 * If first page of an XLOG segment file, make it a long header.
2123 */
2124 if ((XLogSegmentOffset(NewPage->xlp_pageaddr, wal_segment_size)) == 0)
2125 {
2126 XLogLongPageHeader NewLongPage = (XLogLongPageHeader) NewPage;
2127
2128 NewLongPage->xlp_sysid = ControlFile->system_identifier;
2129 NewLongPage->xlp_seg_size = wal_segment_size;
2130 NewLongPage->xlp_xlog_blcksz = XLOG_BLCKSZ;
2131 NewPage->xlp_info |= XLP_LONG_HEADER;
2132 }
2133
2134 /*
2135 * Make sure the initialization of the page becomes visible to others
2136 * before the xlblocks update. GetXLogBuffer() reads xlblocks without
2137 * holding a lock.
2138 */
2140
2141 pg_atomic_write_u64(&XLogCtl->xlblocks[nextidx], NewPageEndPtr);
2142 XLogCtl->InitializedUpTo = NewPageEndPtr;
2143
2144 npages++;
2145 }
2146 LWLockRelease(WALBufMappingLock);
2147
2148#ifdef WAL_DEBUG
2149 if (XLOG_DEBUG && npages > 0)
2150 {
2151 elog(DEBUG1, "initialized %d pages, up to %X/%X",
2152 npages, LSN_FORMAT_ARGS(NewPageEndPtr));
2153 }
2154#endif
2155}
2156
2157/*
2158 * Calculate CheckPointSegments based on max_wal_size_mb and
2159 * checkpoint_completion_target.
2160 */
2161static void
2163{
2164 double target;
2165
2166 /*-------
2167 * Calculate the distance at which to trigger a checkpoint, to avoid
2168 * exceeding max_wal_size_mb. This is based on two assumptions:
2169 *
2170 * a) we keep WAL for only one checkpoint cycle (prior to PG11 we kept
2171 * WAL for two checkpoint cycles to allow us to recover from the
2172 * secondary checkpoint if the first checkpoint failed, though we
2173 * only did this on the primary anyway, not on standby. Keeping just
2174 * one checkpoint simplifies processing and reduces disk space in
2175 * many smaller databases.)
2176 * b) during checkpoint, we consume checkpoint_completion_target *
2177 * number of segments consumed between checkpoints.
2178 *-------
2179 */
2180 target = (double) ConvertToXSegs(max_wal_size_mb, wal_segment_size) /
2182
2183 /* round down */
2184 CheckPointSegments = (int) target;
2185
2186 if (CheckPointSegments < 1)
2188}
2189
2190void
2192{
2195}
2196
2197void
2199{
2202}
2203
2204bool
2206{
2208 {
2209 GUC_check_errdetail("The WAL segment size must be a power of two between 1 MB and 1 GB.");
2210 return false;
2211 }
2212
2213 return true;
2214}
2215
2216/*
2217 * GUC check_hook for max_slot_wal_keep_size
2218 *
2219 * We don't allow the value of max_slot_wal_keep_size other than -1 during the
2220 * binary upgrade. See start_postmaster() in pg_upgrade for more details.
2221 */
2222bool
2224{
2225 if (IsBinaryUpgrade && *newval != -1)
2226 {
2227 GUC_check_errdetail("\"%s\" must be set to -1 during binary upgrade mode.",
2228 "max_slot_wal_keep_size");
2229 return false;
2230 }
2231
2232 return true;
2233}
2234
2235/*
2236 * At a checkpoint, how many WAL segments to recycle as preallocated future
2237 * XLOG segments? Returns the highest segment that should be preallocated.
2238 */
2239static XLogSegNo
2241{
2242 XLogSegNo minSegNo;
2243 XLogSegNo maxSegNo;
2244 double distance;
2245 XLogSegNo recycleSegNo;
2246
2247 /*
2248 * Calculate the segment numbers that min_wal_size_mb and max_wal_size_mb
2249 * correspond to. Always recycle enough segments to meet the minimum, and
2250 * remove enough segments to stay below the maximum.
2251 */
2252 minSegNo = lastredoptr / wal_segment_size +
2254 maxSegNo = lastredoptr / wal_segment_size +
2256
2257 /*
2258 * Between those limits, recycle enough segments to get us through to the
2259 * estimated end of next checkpoint.
2260 *
2261 * To estimate where the next checkpoint will finish, assume that the
2262 * system runs steadily consuming CheckPointDistanceEstimate bytes between
2263 * every checkpoint.
2264 */
2266 /* add 10% for good measure. */
2267 distance *= 1.10;
2268
2269 recycleSegNo = (XLogSegNo) ceil(((double) lastredoptr + distance) /
2271
2272 if (recycleSegNo < minSegNo)
2273 recycleSegNo = minSegNo;
2274 if (recycleSegNo > maxSegNo)
2275 recycleSegNo = maxSegNo;
2276
2277 return recycleSegNo;
2278}
2279
2280/*
2281 * Check whether we've consumed enough xlog space that a checkpoint is needed.
2282 *
2283 * new_segno indicates a log file that has just been filled up (or read
2284 * during recovery). We measure the distance from RedoRecPtr to new_segno
2285 * and see if that exceeds CheckPointSegments.
2286 *
2287 * Note: it is caller's responsibility that RedoRecPtr is up-to-date.
2288 */
2289bool
2291{
2292 XLogSegNo old_segno;
2293
2295
2296 if (new_segno >= old_segno + (uint64) (CheckPointSegments - 1))
2297 return true;
2298 return false;
2299}
2300
2301/*
2302 * Write and/or fsync the log at least as far as WriteRqst indicates.
2303 *
2304 * If flexible == true, we don't have to write as far as WriteRqst, but
2305 * may stop at any convenient boundary (such as a cache or logfile boundary).
2306 * This option allows us to avoid uselessly issuing multiple writes when a
2307 * single one would do.
2308 *
2309 * Must be called with WALWriteLock held. WaitXLogInsertionsToFinish(WriteRqst)
2310 * must be called before grabbing the lock, to make sure the data is ready to
2311 * write.
2312 */
2313static void
2314XLogWrite(XLogwrtRqst WriteRqst, TimeLineID tli, bool flexible)
2315{
2316 bool ispartialpage;
2317 bool last_iteration;
2318 bool finishing_seg;
2319 int curridx;
2320 int npages;
2321 int startidx;
2322 uint32 startoffset;
2323
2324 /* We should always be inside a critical section here */
2326
2327 /*
2328 * Update local LogwrtResult (caller probably did this already, but...)
2329 */
2331
2332 /*
2333 * Since successive pages in the xlog cache are consecutively allocated,
2334 * we can usually gather multiple pages together and issue just one
2335 * write() call. npages is the number of pages we have determined can be
2336 * written together; startidx is the cache block index of the first one,
2337 * and startoffset is the file offset at which it should go. The latter
2338 * two variables are only valid when npages > 0, but we must initialize
2339 * all of them to keep the compiler quiet.
2340 */
2341 npages = 0;
2342 startidx = 0;
2343 startoffset = 0;
2344
2345 /*
2346 * Within the loop, curridx is the cache block index of the page to
2347 * consider writing. Begin at the buffer containing the next unwritten
2348 * page, or last partially written page.
2349 */
2351
2352 while (LogwrtResult.Write < WriteRqst.Write)
2353 {
2354 /*
2355 * Make sure we're not ahead of the insert process. This could happen
2356 * if we're passed a bogus WriteRqst.Write that is past the end of the
2357 * last page that's been initialized by AdvanceXLInsertBuffer.
2358 */
2359 XLogRecPtr EndPtr = pg_atomic_read_u64(&XLogCtl->xlblocks[curridx]);
2360
2361 if (LogwrtResult.Write >= EndPtr)
2362 elog(PANIC, "xlog write request %X/%X is past end of log %X/%X",
2364 LSN_FORMAT_ARGS(EndPtr));
2365
2366 /* Advance LogwrtResult.Write to end of current buffer page */
2367 LogwrtResult.Write = EndPtr;
2368 ispartialpage = WriteRqst.Write < LogwrtResult.Write;
2369
2372 {
2373 /*
2374 * Switch to new logfile segment. We cannot have any pending
2375 * pages here (since we dump what we have at segment end).
2376 */
2377 Assert(npages == 0);
2378 if (openLogFile >= 0)
2379 XLogFileClose();
2382 openLogTLI = tli;
2383
2384 /* create/use new log file */
2387 }
2388
2389 /* Make sure we have the current logfile open */
2390 if (openLogFile < 0)
2391 {
2394 openLogTLI = tli;
2397 }
2398
2399 /* Add current page to the set of pending pages-to-dump */
2400 if (npages == 0)
2401 {
2402 /* first of group */
2403 startidx = curridx;
2404 startoffset = XLogSegmentOffset(LogwrtResult.Write - XLOG_BLCKSZ,
2406 }
2407 npages++;
2408
2409 /*
2410 * Dump the set if this will be the last loop iteration, or if we are
2411 * at the last page of the cache area (since the next page won't be
2412 * contiguous in memory), or if we are at the end of the logfile
2413 * segment.
2414 */
2415 last_iteration = WriteRqst.Write <= LogwrtResult.Write;
2416
2417 finishing_seg = !ispartialpage &&
2418 (startoffset + npages * XLOG_BLCKSZ) >= wal_segment_size;
2419
2420 if (last_iteration ||
2421 curridx == XLogCtl->XLogCacheBlck ||
2422 finishing_seg)
2423 {
2424 char *from;
2425 Size nbytes;
2426 Size nleft;
2427 ssize_t written;
2429
2430 /* OK to write the page(s) */
2431 from = XLogCtl->pages + startidx * (Size) XLOG_BLCKSZ;
2432 nbytes = npages * (Size) XLOG_BLCKSZ;
2433 nleft = nbytes;
2434 do
2435 {
2436 errno = 0;
2437
2438 /*
2439 * Measure I/O timing to write WAL data, for pg_stat_io and/or
2440 * pg_stat_wal.
2441 */
2443
2444 pgstat_report_wait_start(WAIT_EVENT_WAL_WRITE);
2445 written = pg_pwrite(openLogFile, from, nleft, startoffset);
2447
2449 IOOP_WRITE, start, 1, written);
2450
2451 /*
2452 * Increment the I/O timing and the number of times WAL data
2453 * were written out to disk.
2454 */
2456 {
2457 instr_time end;
2458
2461 }
2462
2464
2465 if (written <= 0)
2466 {
2467 char xlogfname[MAXFNAMELEN];
2468 int save_errno;
2469
2470 if (errno == EINTR)
2471 continue;
2472
2473 save_errno = errno;
2474 XLogFileName(xlogfname, tli, openLogSegNo,
2476 errno = save_errno;
2477 ereport(PANIC,
2479 errmsg("could not write to log file \"%s\" at offset %u, length %zu: %m",
2480 xlogfname, startoffset, nleft)));
2481 }
2482 nleft -= written;
2483 from += written;
2484 startoffset += written;
2485 } while (nleft > 0);
2486
2487 npages = 0;
2488
2489 /*
2490 * If we just wrote the whole last page of a logfile segment,
2491 * fsync the segment immediately. This avoids having to go back
2492 * and re-open prior segments when an fsync request comes along
2493 * later. Doing it here ensures that one and only one backend will
2494 * perform this fsync.
2495 *
2496 * This is also the right place to notify the Archiver that the
2497 * segment is ready to copy to archival storage, and to update the
2498 * timer for archive_timeout, and to signal for a checkpoint if
2499 * too many logfile segments have been used since the last
2500 * checkpoint.
2501 */
2502 if (finishing_seg)
2503 {
2505
2506 /* signal that we need to wakeup walsenders later */
2508
2509 LogwrtResult.Flush = LogwrtResult.Write; /* end of page */
2510
2511 if (XLogArchivingActive())
2513
2514 XLogCtl->lastSegSwitchTime = (pg_time_t) time(NULL);
2516
2517 /*
2518 * Request a checkpoint if we've consumed too much xlog since
2519 * the last one. For speed, we first check using the local
2520 * copy of RedoRecPtr, which might be out of date; if it looks
2521 * like a checkpoint is needed, forcibly update RedoRecPtr and
2522 * recheck.
2523 */
2525 {
2526 (void) GetRedoRecPtr();
2529 }
2530 }
2531 }
2532
2533 if (ispartialpage)
2534 {
2535 /* Only asked to write a partial page */
2536 LogwrtResult.Write = WriteRqst.Write;
2537 break;
2538 }
2539 curridx = NextBufIdx(curridx);
2540
2541 /* If flexible, break out of loop as soon as we wrote something */
2542 if (flexible && npages == 0)
2543 break;
2544 }
2545
2546 Assert(npages == 0);
2547
2548 /*
2549 * If asked to flush, do so
2550 */
2551 if (LogwrtResult.Flush < WriteRqst.Flush &&
2553 {
2554 /*
2555 * Could get here without iterating above loop, in which case we might
2556 * have no open file or the wrong one. However, we do not need to
2557 * fsync more than one file.
2558 */
2561 {
2562 if (openLogFile >= 0 &&
2565 XLogFileClose();
2566 if (openLogFile < 0)
2567 {
2570 openLogTLI = tli;
2573 }
2574
2576 }
2577
2578 /* signal that we need to wakeup walsenders later */
2580
2582 }
2583
2584 /*
2585 * Update shared-memory status
2586 *
2587 * We make sure that the shared 'request' values do not fall behind the
2588 * 'result' values. This is not absolutely essential, but it saves some
2589 * code in a couple of places.
2590 */
2597
2598 /*
2599 * We write Write first, bar, then Flush. When reading, the opposite must
2600 * be done (with a matching barrier in between), so that we always see a
2601 * Flush value that trails behind the Write value seen.
2602 */
2606
2607#ifdef USE_ASSERT_CHECKING
2608 {
2612
2618
2619 /* WAL written to disk is always ahead of WAL flushed */
2620 Assert(Write >= Flush);
2621
2622 /* WAL inserted to buffers is always ahead of WAL written */
2623 Assert(Insert >= Write);
2624 }
2625#endif
2626}
2627
2628/*
2629 * Record the LSN for an asynchronous transaction commit/abort
2630 * and nudge the WALWriter if there is work for it to do.
2631 * (This should not be called for synchronous commits.)
2632 */
2633void
2635{
2636 XLogRecPtr WriteRqstPtr = asyncXactLSN;
2637 bool sleeping;
2638 bool wakeup = false;
2639 XLogRecPtr prevAsyncXactLSN;
2640
2642 sleeping = XLogCtl->WalWriterSleeping;
2643 prevAsyncXactLSN = XLogCtl->asyncXactLSN;
2644 if (XLogCtl->asyncXactLSN < asyncXactLSN)
2645 XLogCtl->asyncXactLSN = asyncXactLSN;
2647
2648 /*
2649 * If somebody else already called this function with a more aggressive
2650 * LSN, they will have done what we needed (and perhaps more).
2651 */
2652 if (asyncXactLSN <= prevAsyncXactLSN)
2653 return;
2654
2655 /*
2656 * If the WALWriter is sleeping, kick it to make it come out of low-power
2657 * mode, so that this async commit will reach disk within the expected
2658 * amount of time. Otherwise, determine whether it has enough WAL
2659 * available to flush, the same way that XLogBackgroundFlush() does.
2660 */
2661 if (sleeping)
2662 wakeup = true;
2663 else
2664 {
2665 int flushblocks;
2666
2668
2669 flushblocks =
2670 WriteRqstPtr / XLOG_BLCKSZ - LogwrtResult.Flush / XLOG_BLCKSZ;
2671
2672 if (WalWriterFlushAfter == 0 || flushblocks >= WalWriterFlushAfter)
2673 wakeup = true;
2674 }
2675
2676 if (wakeup)
2677 {
2678 volatile PROC_HDR *procglobal = ProcGlobal;
2679 ProcNumber walwriterProc = procglobal->walwriterProc;
2680
2681 if (walwriterProc != INVALID_PROC_NUMBER)
2682 SetLatch(&GetPGProcByNumber(walwriterProc)->procLatch);
2683 }
2684}
2685
2686/*
2687 * Record the LSN up to which we can remove WAL because it's not required by
2688 * any replication slot.
2689 */
2690void
2692{
2696}
2697
2698
2699/*
2700 * Return the oldest LSN we must retain to satisfy the needs of some
2701 * replication slot.
2702 */
2703static XLogRecPtr
2705{
2706 XLogRecPtr retval;
2707
2711
2712 return retval;
2713}
2714
2715/*
2716 * Advance minRecoveryPoint in control file.
2717 *
2718 * If we crash during recovery, we must reach this point again before the
2719 * database is consistent.
2720 *
2721 * If 'force' is true, 'lsn' argument is ignored. Otherwise, minRecoveryPoint
2722 * is only updated if it's not already greater than or equal to 'lsn'.
2723 */
2724static void
2726{
2727 /* Quick check using our local copy of the variable */
2728 if (!updateMinRecoveryPoint || (!force && lsn <= LocalMinRecoveryPoint))
2729 return;
2730
2731 /*
2732 * An invalid minRecoveryPoint means that we need to recover all the WAL,
2733 * i.e., we're doing crash recovery. We never modify the control file's
2734 * value in that case, so we can short-circuit future checks here too. The
2735 * local values of minRecoveryPoint and minRecoveryPointTLI should not be
2736 * updated until crash recovery finishes. We only do this for the startup
2737 * process as it should not update its own reference of minRecoveryPoint
2738 * until it has finished crash recovery to make sure that all WAL
2739 * available is replayed in this case. This also saves from extra locks
2740 * taken on the control file from the startup process.
2741 */
2743 {
2744 updateMinRecoveryPoint = false;
2745 return;
2746 }
2747
2748 LWLockAcquire(ControlFileLock, LW_EXCLUSIVE);
2749
2750 /* update local copy */
2753
2755 updateMinRecoveryPoint = false;
2756 else if (force || LocalMinRecoveryPoint < lsn)
2757 {
2758 XLogRecPtr newMinRecoveryPoint;
2759 TimeLineID newMinRecoveryPointTLI;
2760
2761 /*
2762 * To avoid having to update the control file too often, we update it
2763 * all the way to the last record being replayed, even though 'lsn'
2764 * would suffice for correctness. This also allows the 'force' case
2765 * to not need a valid 'lsn' value.
2766 *
2767 * Another important reason for doing it this way is that the passed
2768 * 'lsn' value could be bogus, i.e., past the end of available WAL, if
2769 * the caller got it from a corrupted heap page. Accepting such a
2770 * value as the min recovery point would prevent us from coming up at
2771 * all. Instead, we just log a warning and continue with recovery.
2772 * (See also the comments about corrupt LSNs in XLogFlush.)
2773 */
2774 newMinRecoveryPoint = GetCurrentReplayRecPtr(&newMinRecoveryPointTLI);
2775 if (!force && newMinRecoveryPoint < lsn)
2776 elog(WARNING,
2777 "xlog min recovery request %X/%X is past current point %X/%X",
2778 LSN_FORMAT_ARGS(lsn), LSN_FORMAT_ARGS(newMinRecoveryPoint));
2779
2780 /* update control file */
2781 if (ControlFile->minRecoveryPoint < newMinRecoveryPoint)
2782 {
2783 ControlFile->minRecoveryPoint = newMinRecoveryPoint;
2784 ControlFile->minRecoveryPointTLI = newMinRecoveryPointTLI;
2786 LocalMinRecoveryPoint = newMinRecoveryPoint;
2787 LocalMinRecoveryPointTLI = newMinRecoveryPointTLI;
2788
2790 (errmsg_internal("updated min recovery point to %X/%X on timeline %u",
2791 LSN_FORMAT_ARGS(newMinRecoveryPoint),
2792 newMinRecoveryPointTLI)));
2793 }
2794 }
2795 LWLockRelease(ControlFileLock);
2796}
2797
2798/*
2799 * Ensure that all XLOG data through the given position is flushed to disk.
2800 *
2801 * NOTE: this differs from XLogWrite mainly in that the WALWriteLock is not
2802 * already held, and we try to avoid acquiring it if possible.
2803 */
2804void
2806{
2807 XLogRecPtr WriteRqstPtr;
2808 XLogwrtRqst WriteRqst;
2809 TimeLineID insertTLI = XLogCtl->InsertTimeLineID;
2810
2811 /*
2812 * During REDO, we are reading not writing WAL. Therefore, instead of
2813 * trying to flush the WAL, we should update minRecoveryPoint instead. We
2814 * test XLogInsertAllowed(), not InRecovery, because we need checkpointer
2815 * to act this way too, and because when it tries to write the
2816 * end-of-recovery checkpoint, it should indeed flush.
2817 */
2818 if (!XLogInsertAllowed())
2819 {
2820 UpdateMinRecoveryPoint(record, false);
2821 return;
2822 }
2823
2824 /* Quick exit if already known flushed */
2825 if (record <= LogwrtResult.Flush)
2826 return;
2827
2828#ifdef WAL_DEBUG
2829 if (XLOG_DEBUG)
2830 elog(LOG, "xlog flush request %X/%X; write %X/%X; flush %X/%X",
2831 LSN_FORMAT_ARGS(record),
2834#endif
2835
2837
2838 /*
2839 * Since fsync is usually a horribly expensive operation, we try to
2840 * piggyback as much data as we can on each fsync: if we see any more data
2841 * entered into the xlog buffer, we'll write and fsync that too, so that
2842 * the final value of LogwrtResult.Flush is as large as possible. This
2843 * gives us some chance of avoiding another fsync immediately after.
2844 */
2845
2846 /* initialize to given target; may increase below */
2847 WriteRqstPtr = record;
2848
2849 /*
2850 * Now wait until we get the write lock, or someone else does the flush
2851 * for us.
2852 */
2853 for (;;)
2854 {
2855 XLogRecPtr insertpos;
2856
2857 /* done already? */
2859 if (record <= LogwrtResult.Flush)
2860 break;
2861
2862 /*
2863 * Before actually performing the write, wait for all in-flight
2864 * insertions to the pages we're about to write to finish.
2865 */
2867 if (WriteRqstPtr < XLogCtl->LogwrtRqst.Write)
2868 WriteRqstPtr = XLogCtl->LogwrtRqst.Write;
2870 insertpos = WaitXLogInsertionsToFinish(WriteRqstPtr);
2871
2872 /*
2873 * Try to get the write lock. If we can't get it immediately, wait
2874 * until it's released, and recheck if we still need to do the flush
2875 * or if the backend that held the lock did it for us already. This
2876 * helps to maintain a good rate of group committing when the system
2877 * is bottlenecked by the speed of fsyncing.
2878 */
2879 if (!LWLockAcquireOrWait(WALWriteLock, LW_EXCLUSIVE))
2880 {
2881 /*
2882 * The lock is now free, but we didn't acquire it yet. Before we
2883 * do, loop back to check if someone else flushed the record for
2884 * us already.
2885 */
2886 continue;
2887 }
2888
2889 /* Got the lock; recheck whether request is satisfied */
2891 if (record <= LogwrtResult.Flush)
2892 {
2893 LWLockRelease(WALWriteLock);
2894 break;
2895 }
2896
2897 /*
2898 * Sleep before flush! By adding a delay here, we may give further
2899 * backends the opportunity to join the backlog of group commit
2900 * followers; this can significantly improve transaction throughput,
2901 * at the risk of increasing transaction latency.
2902 *
2903 * We do not sleep if enableFsync is not turned on, nor if there are
2904 * fewer than CommitSiblings other backends with active transactions.
2905 */
2906 if (CommitDelay > 0 && enableFsync &&
2908 {
2910
2911 /*
2912 * Re-check how far we can now flush the WAL. It's generally not
2913 * safe to call WaitXLogInsertionsToFinish while holding
2914 * WALWriteLock, because an in-progress insertion might need to
2915 * also grab WALWriteLock to make progress. But we know that all
2916 * the insertions up to insertpos have already finished, because
2917 * that's what the earlier WaitXLogInsertionsToFinish() returned.
2918 * We're only calling it again to allow insertpos to be moved
2919 * further forward, not to actually wait for anyone.
2920 */
2921 insertpos = WaitXLogInsertionsToFinish(insertpos);
2922 }
2923
2924 /* try to write/flush later additions to XLOG as well */
2925 WriteRqst.Write = insertpos;
2926 WriteRqst.Flush = insertpos;
2927
2928 XLogWrite(WriteRqst, insertTLI, false);
2929
2930 LWLockRelease(WALWriteLock);
2931 /* done */
2932 break;
2933 }
2934
2936
2937 /* wake up walsenders now that we've released heavily contended locks */
2939
2940 /*
2941 * If we still haven't flushed to the request point then we have a
2942 * problem; most likely, the requested flush point is past end of XLOG.
2943 * This has been seen to occur when a disk page has a corrupted LSN.
2944 *
2945 * Formerly we treated this as a PANIC condition, but that hurts the
2946 * system's robustness rather than helping it: we do not want to take down
2947 * the whole system due to corruption on one data page. In particular, if
2948 * the bad page is encountered again during recovery then we would be
2949 * unable to restart the database at all! (This scenario actually
2950 * happened in the field several times with 7.1 releases.) As of 8.4, bad
2951 * LSNs encountered during recovery are UpdateMinRecoveryPoint's problem;
2952 * the only time we can reach here during recovery is while flushing the
2953 * end-of-recovery checkpoint record, and we don't expect that to have a
2954 * bad LSN.
2955 *
2956 * Note that for calls from xact.c, the ERROR will be promoted to PANIC
2957 * since xact.c calls this routine inside a critical section. However,
2958 * calls from bufmgr.c are not within critical sections and so we will not
2959 * force a restart for a bad LSN on a data page.
2960 */
2961 if (LogwrtResult.Flush < record)
2962 elog(ERROR,
2963 "xlog flush request %X/%X is not satisfied --- flushed only to %X/%X",
2964 LSN_FORMAT_ARGS(record),
2966}
2967
2968/*
2969 * Write & flush xlog, but without specifying exactly where to.
2970 *
2971 * We normally write only completed blocks; but if there is nothing to do on
2972 * that basis, we check for unwritten async commits in the current incomplete
2973 * block, and write through the latest one of those. Thus, if async commits
2974 * are not being used, we will write complete blocks only.
2975 *
2976 * If, based on the above, there's anything to write we do so immediately. But
2977 * to avoid calling fsync, fdatasync et. al. at a rate that'd impact
2978 * concurrent IO, we only flush WAL every wal_writer_delay ms, or if there's
2979 * more than wal_writer_flush_after unflushed blocks.
2980 *
2981 * We can guarantee that async commits reach disk after at most three
2982 * wal_writer_delay cycles. (When flushing complete blocks, we allow XLogWrite
2983 * to write "flexibly", meaning it can stop at the end of the buffer ring;
2984 * this makes a difference only with very high load or long wal_writer_delay,
2985 * but imposes one extra cycle for the worst case for async commits.)
2986 *
2987 * This routine is invoked periodically by the background walwriter process.
2988 *
2989 * Returns true if there was any work to do, even if we skipped flushing due
2990 * to wal_writer_delay/wal_writer_flush_after.
2991 */
2992bool
2994{
2995 XLogwrtRqst WriteRqst;
2996 bool flexible = true;
2997 static TimestampTz lastflush;
2999 int flushblocks;
3000 TimeLineID insertTLI;
3001
3002 /* XLOG doesn't need flushing during recovery */
3003 if (RecoveryInProgress())
3004 return false;
3005
3006 /*
3007 * Since we're not in recovery, InsertTimeLineID is set and can't change,
3008 * so we can read it without a lock.
3009 */
3010 insertTLI = XLogCtl->InsertTimeLineID;
3011
3012 /* read updated LogwrtRqst */
3014 WriteRqst = XLogCtl->LogwrtRqst;
3016
3017 /* back off to last completed page boundary */
3018 WriteRqst.Write -= WriteRqst.Write % XLOG_BLCKSZ;
3019
3020 /* if we have already flushed that far, consider async commit records */
3022 if (WriteRqst.Write <= LogwrtResult.Flush)
3023 {
3025 WriteRqst.Write = XLogCtl->asyncXactLSN;
3027 flexible = false; /* ensure it all gets written */
3028 }
3029
3030 /*
3031 * If already known flushed, we're done. Just need to check if we are
3032 * holding an open file handle to a logfile that's no longer in use,
3033 * preventing the file from being deleted.
3034 */
3035 if (WriteRqst.Write <= LogwrtResult.Flush)
3036 {
3037 if (openLogFile >= 0)
3038 {
3041 {
3042 XLogFileClose();
3043 }
3044 }
3045 return false;
3046 }
3047
3048 /*
3049 * Determine how far to flush WAL, based on the wal_writer_delay and
3050 * wal_writer_flush_after GUCs.
3051 *
3052 * Note that XLogSetAsyncXactLSN() performs similar calculation based on
3053 * wal_writer_flush_after, to decide when to wake us up. Make sure the
3054 * logic is the same in both places if you change this.
3055 */
3057 flushblocks =
3058 WriteRqst.Write / XLOG_BLCKSZ - LogwrtResult.Flush / XLOG_BLCKSZ;
3059
3060 if (WalWriterFlushAfter == 0 || lastflush == 0)
3061 {
3062 /* first call, or block based limits disabled */
3063 WriteRqst.Flush = WriteRqst.Write;
3064 lastflush = now;
3065 }
3066 else if (TimestampDifferenceExceeds(lastflush, now, WalWriterDelay))
3067 {
3068 /*
3069 * Flush the writes at least every WalWriterDelay ms. This is
3070 * important to bound the amount of time it takes for an asynchronous
3071 * commit to hit disk.
3072 */
3073 WriteRqst.Flush = WriteRqst.Write;
3074 lastflush = now;
3075 }
3076 else if (flushblocks >= WalWriterFlushAfter)
3077 {
3078 /* exceeded wal_writer_flush_after blocks, flush */
3079 WriteRqst.Flush = WriteRqst.Write;
3080 lastflush = now;
3081 }
3082 else
3083 {
3084 /* no flushing, this time round */
3085 WriteRqst.Flush = 0;
3086 }
3087
3088#ifdef WAL_DEBUG
3089 if (XLOG_DEBUG)
3090 elog(LOG, "xlog bg flush request write %X/%X; flush: %X/%X, current is write %X/%X; flush %X/%X",
3091 LSN_FORMAT_ARGS(WriteRqst.Write),
3092 LSN_FORMAT_ARGS(WriteRqst.Flush),
3095#endif
3096
3098
3099 /* now wait for any in-progress insertions to finish and get write lock */
3101 LWLockAcquire(WALWriteLock, LW_EXCLUSIVE);
3103 if (WriteRqst.Write > LogwrtResult.Write ||
3104 WriteRqst.Flush > LogwrtResult.Flush)
3105 {
3106 XLogWrite(WriteRqst, insertTLI, flexible);
3107 }
3108 LWLockRelease(WALWriteLock);
3109
3111
3112 /* wake up walsenders now that we've released heavily contended locks */
3114
3115 /*
3116 * Great, done. To take some work off the critical path, try to initialize
3117 * as many of the no-longer-needed WAL buffers for future use as we can.
3118 */
3119 AdvanceXLInsertBuffer(InvalidXLogRecPtr, insertTLI, true);
3120
3121 /*
3122 * If we determined that we need to write data, but somebody else
3123 * wrote/flushed already, it should be considered as being active, to
3124 * avoid hibernating too early.
3125 */
3126 return true;
3127}
3128
3129/*
3130 * Test whether XLOG data has been flushed up to (at least) the given position.
3131 *
3132 * Returns true if a flush is still needed. (It may be that someone else
3133 * is already in process of flushing that far, however.)
3134 */
3135bool
3137{
3138 /*
3139 * During recovery, we don't flush WAL but update minRecoveryPoint
3140 * instead. So "needs flush" is taken to mean whether minRecoveryPoint
3141 * would need to be updated.
3142 */
3143 if (RecoveryInProgress())
3144 {
3145 /*
3146 * An invalid minRecoveryPoint means that we need to recover all the
3147 * WAL, i.e., we're doing crash recovery. We never modify the control
3148 * file's value in that case, so we can short-circuit future checks
3149 * here too. This triggers a quick exit path for the startup process,
3150 * which cannot update its local copy of minRecoveryPoint as long as
3151 * it has not replayed all WAL available when doing crash recovery.
3152 */
3154 updateMinRecoveryPoint = false;
3155
3156 /* Quick exit if already known to be updated or cannot be updated */
3158 return false;
3159
3160 /*
3161 * Update local copy of minRecoveryPoint. But if the lock is busy,
3162 * just return a conservative guess.
3163 */
3164 if (!LWLockConditionalAcquire(ControlFileLock, LW_SHARED))
3165 return true;
3168 LWLockRelease(ControlFileLock);
3169
3170 /*
3171 * Check minRecoveryPoint for any other process than the startup
3172 * process doing crash recovery, which should not update the control
3173 * file value if crash recovery is still running.
3174 */
3176 updateMinRecoveryPoint = false;
3177
3178 /* check again */
3180 return false;
3181 else
3182 return true;
3183 }
3184
3185 /* Quick exit if already known flushed */
3186 if (record <= LogwrtResult.Flush)
3187 return false;
3188
3189 /* read LogwrtResult and update local state */
3191
3192 /* check again */
3193 if (record <= LogwrtResult.Flush)
3194 return false;
3195
3196 return true;
3197}
3198
3199/*
3200 * Try to make a given XLOG file segment exist.
3201 *
3202 * logsegno: identify segment.
3203 *
3204 * *added: on return, true if this call raised the number of extant segments.
3205 *
3206 * path: on return, this char[MAXPGPATH] has the path to the logsegno file.
3207 *
3208 * Returns -1 or FD of opened file. A -1 here is not an error; a caller
3209 * wanting an open segment should attempt to open "path", which usually will
3210 * succeed. (This is weird, but it's efficient for the callers.)
3211 */
3212static int
3214 bool *added, char *path)
3215{
3216 char tmppath[MAXPGPATH];
3217 XLogSegNo installed_segno;
3218 XLogSegNo max_segno;
3219 int fd;
3220 int save_errno;
3221 int open_flags = O_RDWR | O_CREAT | O_EXCL | PG_BINARY;
3222 instr_time io_start;
3223
3224 Assert(logtli != 0);
3225
3226 XLogFilePath(path, logtli, logsegno, wal_segment_size);
3227
3228 /*
3229 * Try to use existent file (checkpoint maker may have created it already)
3230 */
3231 *added = false;
3232 fd = BasicOpenFile(path, O_RDWR | PG_BINARY | O_CLOEXEC |
3234 if (fd < 0)
3235 {
3236 if (errno != ENOENT)
3237 ereport(ERROR,
3239 errmsg("could not open file \"%s\": %m", path)));
3240 }
3241 else
3242 return fd;
3243
3244 /*
3245 * Initialize an empty (all zeroes) segment. NOTE: it is possible that
3246 * another process is doing the same thing. If so, we will end up
3247 * pre-creating an extra log segment. That seems OK, and better than
3248 * holding the lock throughout this lengthy process.
3249 */
3250 elog(DEBUG2, "creating and filling new WAL file");
3251
3252 snprintf(tmppath, MAXPGPATH, XLOGDIR "/xlogtemp.%d", (int) getpid());
3253
3254 unlink(tmppath);
3255
3257 open_flags |= PG_O_DIRECT;
3258
3259 /* do not use get_sync_bit() here --- want to fsync only at end of fill */
3260 fd = BasicOpenFile(tmppath, open_flags);
3261 if (fd < 0)
3262 ereport(ERROR,
3264 errmsg("could not create file \"%s\": %m", tmppath)));
3265
3266 /* Measure I/O timing when initializing segment */
3268
3269 pgstat_report_wait_start(WAIT_EVENT_WAL_INIT_WRITE);
3270 save_errno = 0;
3271 if (wal_init_zero)
3272 {
3273 ssize_t rc;
3274
3275 /*
3276 * Zero-fill the file. With this setting, we do this the hard way to
3277 * ensure that all the file space has really been allocated. On
3278 * platforms that allow "holes" in files, just seeking to the end
3279 * doesn't allocate intermediate space. This way, we know that we
3280 * have all the space and (after the fsync below) that all the
3281 * indirect blocks are down on disk. Therefore, fdatasync(2) or
3282 * O_DSYNC will be sufficient to sync future writes to the log file.
3283 */
3285
3286 if (rc < 0)
3287 save_errno = errno;
3288 }
3289 else
3290 {
3291 /*
3292 * Otherwise, seeking to the end and writing a solitary byte is
3293 * enough.
3294 */
3295 errno = 0;
3296 if (pg_pwrite(fd, "\0", 1, wal_segment_size - 1) != 1)
3297 {
3298 /* if write didn't set errno, assume no disk space */
3299 save_errno = errno ? errno : ENOSPC;
3300 }
3301 }
3303
3304 /*
3305 * A full segment worth of data is written when using wal_init_zero. One
3306 * byte is written when not using it.
3307 */
3309 io_start, 1,
3311
3312 if (save_errno)
3313 {
3314 /*
3315 * If we fail to make the file, delete it to release disk space
3316 */
3317 unlink(tmppath);
3318
3319 close(fd);
3320
3321 errno = save_errno;
3322
3323 ereport(ERROR,
3325 errmsg("could not write to file \"%s\": %m", tmppath)));
3326 }
3327
3328 /* Measure I/O timing when flushing segment */
3330
3331 pgstat_report_wait_start(WAIT_EVENT_WAL_INIT_SYNC);
3332 if (pg_fsync(fd) != 0)
3333 {
3334 save_errno = errno;
3335 close(fd);
3336 errno = save_errno;
3337 ereport(ERROR,
3339 errmsg("could not fsync file \"%s\": %m", tmppath)));
3340 }
3342
3344 IOOP_FSYNC, io_start, 1, 0);
3345
3346 if (close(fd) != 0)
3347 ereport(ERROR,
3349 errmsg("could not close file \"%s\": %m", tmppath)));
3350
3351 /*
3352 * Now move the segment into place with its final name. Cope with
3353 * possibility that someone else has created the file while we were
3354 * filling ours: if so, use ours to pre-create a future log segment.
3355 */
3356 installed_segno = logsegno;
3357
3358 /*
3359 * XXX: What should we use as max_segno? We used to use XLOGfileslop when
3360 * that was a constant, but that was always a bit dubious: normally, at a
3361 * checkpoint, XLOGfileslop was the offset from the checkpoint record, but
3362 * here, it was the offset from the insert location. We can't do the
3363 * normal XLOGfileslop calculation here because we don't have access to
3364 * the prior checkpoint's redo location. So somewhat arbitrarily, just use
3365 * CheckPointSegments.
3366 */
3367 max_segno = logsegno + CheckPointSegments;
3368 if (InstallXLogFileSegment(&installed_segno, tmppath, true, max_segno,
3369 logtli))
3370 {
3371 *added = true;
3372 elog(DEBUG2, "done creating and filling new WAL file");
3373 }
3374 else
3375 {
3376 /*
3377 * No need for any more future segments, or InstallXLogFileSegment()
3378 * failed to rename the file into place. If the rename failed, a
3379 * caller opening the file may fail.
3380 */
3381 unlink(tmppath);
3382 elog(DEBUG2, "abandoned new WAL file");
3383 }
3384
3385 return -1;
3386}
3387
3388/*
3389 * Create a new XLOG file segment, or open a pre-existing one.
3390 *
3391 * logsegno: identify segment to be created/opened.
3392 *
3393 * Returns FD of opened file.
3394 *
3395 * Note: errors here are ERROR not PANIC because we might or might not be
3396 * inside a critical section (eg, during checkpoint there is no reason to
3397 * take down the system on failure). They will promote to PANIC if we are
3398 * in a critical section.
3399 */
3400int
3402{
3403 bool ignore_added;
3404 char path[MAXPGPATH];
3405 int fd;
3406
3407 Assert(logtli != 0);
3408
3409 fd = XLogFileInitInternal(logsegno, logtli, &ignore_added, path);
3410 if (fd >= 0)
3411 return fd;
3412
3413 /* Now open original target segment (might not be file I just made) */
3414 fd = BasicOpenFile(path, O_RDWR | PG_BINARY | O_CLOEXEC |
3416 if (fd < 0)
3417 ereport(ERROR,
3419 errmsg("could not open file \"%s\": %m", path)));
3420 return fd;
3421}
3422
3423/*
3424 * Create a new XLOG file segment by copying a pre-existing one.
3425 *
3426 * destsegno: identify segment to be created.
3427 *
3428 * srcTLI, srcsegno: identify segment to be copied (could be from
3429 * a different timeline)
3430 *
3431 * upto: how much of the source file to copy (the rest is filled with
3432 * zeros)
3433 *
3434 * Currently this is only used during recovery, and so there are no locking
3435 * considerations. But we should be just as tense as XLogFileInit to avoid
3436 * emplacing a bogus file.
3437 */
3438static void
3440 TimeLineID srcTLI, XLogSegNo srcsegno,
3441 int upto)
3442{
3443 char path[MAXPGPATH];
3444 char tmppath[MAXPGPATH];
3445 PGAlignedXLogBlock buffer;
3446 int srcfd;
3447 int fd;
3448 int nbytes;
3449
3450 /*
3451 * Open the source file
3452 */
3453 XLogFilePath(path, srcTLI, srcsegno, wal_segment_size);
3454 srcfd = OpenTransientFile(path, O_RDONLY | PG_BINARY);
3455 if (srcfd < 0)
3456 ereport(ERROR,
3458 errmsg("could not open file \"%s\": %m", path)));
3459
3460 /*
3461 * Copy into a temp file name.
3462 */
3463 snprintf(tmppath, MAXPGPATH, XLOGDIR "/xlogtemp.%d", (int) getpid());
3464
3465 unlink(tmppath);
3466
3467 /* do not use get_sync_bit() here --- want to fsync only at end of fill */
3468 fd = OpenTransientFile(tmppath, O_RDWR | O_CREAT | O_EXCL | PG_BINARY);
3469 if (fd < 0)
3470 ereport(ERROR,
3472 errmsg("could not create file \"%s\": %m", tmppath)));
3473
3474 /*
3475 * Do the data copying.
3476 */
3477 for (nbytes = 0; nbytes < wal_segment_size; nbytes += sizeof(buffer))
3478 {
3479 int nread;
3480
3481 nread = upto - nbytes;
3482
3483 /*
3484 * The part that is not read from the source file is filled with
3485 * zeros.
3486 */
3487 if (nread < sizeof(buffer))
3488 memset(buffer.data, 0, sizeof(buffer));
3489
3490 if (nread > 0)
3491 {
3492 int r;
3493
3494 if (nread > sizeof(buffer))
3495 nread = sizeof(buffer);
3496 pgstat_report_wait_start(WAIT_EVENT_WAL_COPY_READ);
3497 r = read(srcfd, buffer.data, nread);
3498 if (r != nread)
3499 {
3500 if (r < 0)
3501 ereport(ERROR,
3503 errmsg("could not read file \"%s\": %m",
3504 path)));
3505 else
3506 ereport(ERROR,
3508 errmsg("could not read file \"%s\": read %d of %zu",
3509 path, r, (Size) nread)));
3510 }
3512 }
3513 errno = 0;
3514 pgstat_report_wait_start(WAIT_EVENT_WAL_COPY_WRITE);
3515 if ((int) write(fd, buffer.data, sizeof(buffer)) != (int) sizeof(buffer))
3516 {
3517 int save_errno = errno;
3518
3519 /*
3520 * If we fail to make the file, delete it to release disk space
3521 */
3522 unlink(tmppath);
3523 /* if write didn't set errno, assume problem is no disk space */
3524 errno = save_errno ? save_errno : ENOSPC;
3525
3526 ereport(ERROR,
3528 errmsg("could not write to file \"%s\": %m", tmppath)));
3529 }
3531 }
3532
3533 pgstat_report_wait_start(WAIT_EVENT_WAL_COPY_SYNC);
3534 if (pg_fsync(fd) != 0)
3537 errmsg("could not fsync file \"%s\": %m", tmppath)));
3539
3540 if (CloseTransientFile(fd) != 0)
3541 ereport(ERROR,
3543 errmsg("could not close file \"%s\": %m", tmppath)));
3544
3545 if (CloseTransientFile(srcfd) != 0)
3546 ereport(ERROR,
3548 errmsg("could not close file \"%s\": %m", path)));
3549
3550 /*
3551 * Now move the segment into place with its final name.
3552 */
3553 if (!InstallXLogFileSegment(&destsegno, tmppath, false, 0, destTLI))
3554 elog(ERROR, "InstallXLogFileSegment should not have failed");
3555}
3556
3557/*
3558 * Install a new XLOG segment file as a current or future log segment.
3559 *
3560 * This is used both to install a newly-created segment (which has a temp
3561 * filename while it's being created) and to recycle an old segment.
3562 *
3563 * *segno: identify segment to install as (or first possible target).
3564 * When find_free is true, this is modified on return to indicate the
3565 * actual installation location or last segment searched.
3566 *
3567 * tmppath: initial name of file to install. It will be renamed into place.
3568 *
3569 * find_free: if true, install the new segment at the first empty segno
3570 * number at or after the passed numbers. If false, install the new segment
3571 * exactly where specified, deleting any existing segment file there.
3572 *
3573 * max_segno: maximum segment number to install the new file as. Fail if no
3574 * free slot is found between *segno and max_segno. (Ignored when find_free
3575 * is false.)
3576 *
3577 * tli: The timeline on which the new segment should be installed.
3578 *
3579 * Returns true if the file was installed successfully. false indicates that
3580 * max_segno limit was exceeded, the startup process has disabled this
3581 * function for now, or an error occurred while renaming the file into place.
3582 */
3583static bool
3584InstallXLogFileSegment(XLogSegNo *segno, char *tmppath,
3585 bool find_free, XLogSegNo max_segno, TimeLineID tli)
3586{
3587 char path[MAXPGPATH];
3588 struct stat stat_buf;
3589
3590 Assert(tli != 0);
3591
3592 XLogFilePath(path, tli, *segno, wal_segment_size);
3593
3594 LWLockAcquire(ControlFileLock, LW_EXCLUSIVE);
3596 {
3597 LWLockRelease(ControlFileLock);
3598 return false;
3599 }
3600
3601 if (!find_free)
3602 {
3603 /* Force installation: get rid of any pre-existing segment file */
3604 durable_unlink(path, DEBUG1);
3605 }
3606 else
3607 {
3608 /* Find a free slot to put it in */
3609 while (stat(path, &stat_buf) == 0)
3610 {
3611 if ((*segno) >= max_segno)
3612 {
3613 /* Failed to find a free slot within specified range */
3614 LWLockRelease(ControlFileLock);
3615 return false;
3616 }
3617 (*segno)++;
3618 XLogFilePath(path, tli, *segno, wal_segment_size);
3619 }
3620 }
3621
3622 Assert(access(path, F_OK) != 0 && errno == ENOENT);
3623 if (durable_rename(tmppath, path, LOG) != 0)
3624 {
3625 LWLockRelease(ControlFileLock);
3626 /* durable_rename already emitted log message */
3627 return false;
3628 }
3629
3630 LWLockRelease(ControlFileLock);
3631
3632 return true;
3633}
3634
3635/*
3636 * Open a pre-existing logfile segment for writing.
3637 */
3638int
3640{
3641 char path[MAXPGPATH];
3642 int fd;
3643
3644 XLogFilePath(path, tli, segno, wal_segment_size);
3645
3646 fd = BasicOpenFile(path, O_RDWR | PG_BINARY | O_CLOEXEC |
3648 if (fd < 0)
3649 ereport(PANIC,
3651 errmsg("could not open file \"%s\": %m", path)));
3652
3653 return fd;
3654}
3655
3656/*
3657 * Close the current logfile segment for writing.
3658 */
3659static void
3661{
3662 Assert(openLogFile >= 0);
3663
3664 /*
3665 * WAL segment files will not be re-read in normal operation, so we advise
3666 * the OS to release any cached pages. But do not do so if WAL archiving
3667 * or streaming is active, because archiver and walsender process could
3668 * use the cache to read the WAL segment.
3669 */
3670#if defined(USE_POSIX_FADVISE) && defined(POSIX_FADV_DONTNEED)
3671 if (!XLogIsNeeded() && (io_direct_flags & IO_DIRECT_WAL) == 0)
3672 (void) posix_fadvise(openLogFile, 0, 0, POSIX_FADV_DONTNEED);
3673#endif
3674
3675 if (close(openLogFile) != 0)
3676 {
3677 char xlogfname[MAXFNAMELEN];
3678 int save_errno = errno;
3679
3681 errno = save_errno;
3682 ereport(PANIC,
3684 errmsg("could not close file \"%s\": %m", xlogfname)));
3685 }
3686
3687 openLogFile = -1;
3689}
3690
3691/*
3692 * Preallocate log files beyond the specified log endpoint.
3693 *
3694 * XXX this is currently extremely conservative, since it forces only one
3695 * future log segment to exist, and even that only if we are 75% done with
3696 * the current one. This is only appropriate for very low-WAL-volume systems.
3697 * High-volume systems will be OK once they've built up a sufficient set of
3698 * recycled log segments, but the startup transient is likely to include
3699 * a lot of segment creations by foreground processes, which is not so good.
3700 *
3701 * XLogFileInitInternal() can ereport(ERROR). All known causes indicate big
3702 * trouble; for example, a full filesystem is one cause. The checkpoint WAL
3703 * and/or ControlFile updates already completed. If a RequestCheckpoint()
3704 * initiated the present checkpoint and an ERROR ends this function, the
3705 * command that called RequestCheckpoint() fails. That's not ideal, but it's
3706 * not worth contorting more functions to use caller-specified elevel values.
3707 * (With or without RequestCheckpoint(), an ERROR forestalls some inessential
3708 * reporting and resource reclamation.)
3709 */
3710static void
3712{
3713 XLogSegNo _logSegNo;
3714 int lf;
3715 bool added;
3716 char path[MAXPGPATH];
3717 uint64 offset;
3718
3720 return; /* unlocked check says no */
3721
3722 XLByteToPrevSeg(endptr, _logSegNo, wal_segment_size);
3723 offset = XLogSegmentOffset(endptr - 1, wal_segment_size);
3724 if (offset >= (uint32) (0.75 * wal_segment_size))
3725 {
3726 _logSegNo++;
3727 lf = XLogFileInitInternal(_logSegNo, tli, &added, path);
3728 if (lf >= 0)
3729 close(lf);
3730 if (added)
3732 }
3733}
3734
3735/*
3736 * Throws an error if the given log segment has already been removed or
3737 * recycled. The caller should only pass a segment that it knows to have
3738 * existed while the server has been running, as this function always
3739 * succeeds if no WAL segments have been removed since startup.
3740 * 'tli' is only used in the error message.
3741 *
3742 * Note: this function guarantees to keep errno unchanged on return.
3743 * This supports callers that use this to possibly deliver a better
3744 * error message about a missing file, while still being able to throw
3745 * a normal file-access error afterwards, if this does return.
3746 */
3747void
3749{
3750 int save_errno = errno;
3751 XLogSegNo lastRemovedSegNo;
3752
3754 lastRemovedSegNo = XLogCtl->lastRemovedSegNo;
3756
3757 if (segno <= lastRemovedSegNo)
3758 {
3759 char filename[MAXFNAMELEN];
3760
3762 errno = save_errno;
3763 ereport(ERROR,
3765 errmsg("requested WAL segment %s has already been removed",
3766 filename)));
3767 }
3768 errno = save_errno;
3769}
3770
3771/*
3772 * Return the last WAL segment removed, or 0 if no segment has been removed
3773 * since startup.
3774 *
3775 * NB: the result can be out of date arbitrarily fast, the caller has to deal
3776 * with that.
3777 */
3780{
3781 XLogSegNo lastRemovedSegNo;
3782
3784 lastRemovedSegNo = XLogCtl->lastRemovedSegNo;
3786
3787 return lastRemovedSegNo;
3788}
3789
3790/*
3791 * Return the oldest WAL segment on the given TLI that still exists in
3792 * XLOGDIR, or 0 if none.
3793 */
3796{
3797 DIR *xldir;
3798 struct dirent *xlde;
3799 XLogSegNo oldest_segno = 0;
3800
3801 xldir = AllocateDir(XLOGDIR);
3802 while ((xlde = ReadDir(xldir, XLOGDIR)) != NULL)
3803 {
3804 TimeLineID file_tli;
3805 XLogSegNo file_segno;
3806
3807 /* Ignore files that are not XLOG segments. */
3808 if (!IsXLogFileName(xlde->d_name))
3809 continue;
3810
3811 /* Parse filename to get TLI and segno. */
3812 XLogFromFileName(xlde->d_name, &file_tli, &file_segno,
3814
3815 /* Ignore anything that's not from the TLI of interest. */
3816 if (tli != file_tli)
3817 continue;
3818
3819 /* If it's the oldest so far, update oldest_segno. */
3820 if (oldest_segno == 0 || file_segno < oldest_segno)
3821 oldest_segno = file_segno;
3822 }
3823
3824 FreeDir(xldir);
3825 return oldest_segno;
3826}
3827
3828/*
3829 * Update the last removed segno pointer in shared memory, to reflect that the
3830 * given XLOG file has been removed.
3831 */
3832static void
3834{
3835 uint32 tli;
3836 XLogSegNo segno;
3837
3839
3841 if (segno > XLogCtl->lastRemovedSegNo)
3842 XLogCtl->lastRemovedSegNo = segno;
3844}
3845
3846/*
3847 * Remove all temporary log files in pg_wal
3848 *
3849 * This is called at the beginning of recovery after a previous crash,
3850 * at a point where no other processes write fresh WAL data.
3851 */
3852static void
3854{
3855 DIR *xldir;
3856 struct dirent *xlde;
3857
3858 elog(DEBUG2, "removing all temporary WAL segments");
3859
3860 xldir = AllocateDir(XLOGDIR);
3861 while ((xlde = ReadDir(xldir, XLOGDIR)) != NULL)
3862 {
3863 char path[MAXPGPATH];
3864
3865 if (strncmp(xlde->d_name, "xlogtemp.", 9) != 0)
3866 continue;
3867
3868 snprintf(path, MAXPGPATH, XLOGDIR "/%s", xlde->d_name);
3869 unlink(path);
3870 elog(DEBUG2, "removed temporary WAL segment \"%s\"", path);
3871 }
3872 FreeDir(xldir);
3873}
3874
3875/*
3876 * Recycle or remove all log files older or equal to passed segno.
3877 *
3878 * endptr is current (or recent) end of xlog, and lastredoptr is the
3879 * redo pointer of the last checkpoint. These are used to determine
3880 * whether we want to recycle rather than delete no-longer-wanted log files.
3881 *
3882 * insertTLI is the current timeline for XLOG insertion. Any recycled
3883 * segments should be reused for this timeline.
3884 */
3885static void
3887 TimeLineID insertTLI)
3888{
3889 DIR *xldir;
3890 struct dirent *xlde;
3891 char lastoff[MAXFNAMELEN];
3892 XLogSegNo endlogSegNo;
3893 XLogSegNo recycleSegNo;
3894
3895 /* Initialize info about where to try to recycle to */
3896 XLByteToSeg(endptr, endlogSegNo, wal_segment_size);
3897 recycleSegNo = XLOGfileslop(lastredoptr);
3898
3899 /*
3900 * Construct a filename of the last segment to be kept. The timeline ID
3901 * doesn't matter, we ignore that in the comparison. (During recovery,
3902 * InsertTimeLineID isn't set, so we can't use that.)
3903 */
3904 XLogFileName(lastoff, 0, segno, wal_segment_size);
3905
3906 elog(DEBUG2, "attempting to remove WAL segments older than log file %s",
3907 lastoff);
3908
3909 xldir = AllocateDir(XLOGDIR);
3910
3911 while ((xlde = ReadDir(xldir, XLOGDIR)) != NULL)
3912 {
3913 /* Ignore files that are not XLOG segments */
3914 if (!IsXLogFileName(xlde->d_name) &&
3916 continue;
3917
3918 /*
3919 * We ignore the timeline part of the XLOG segment identifiers in
3920 * deciding whether a segment is still needed. This ensures that we
3921 * won't prematurely remove a segment from a parent timeline. We could
3922 * probably be a little more proactive about removing segments of
3923 * non-parent timelines, but that would be a whole lot more
3924 * complicated.
3925 *
3926 * We use the alphanumeric sorting property of the filenames to decide
3927 * which ones are earlier than the lastoff segment.
3928 */
3929 if (strcmp(xlde->d_name + 8, lastoff + 8) <= 0)
3930 {
3931 if (XLogArchiveCheckDone(xlde->d_name))
3932 {
3933 /* Update the last removed location in shared memory first */
3935
3936 RemoveXlogFile(xlde, recycleSegNo, &endlogSegNo, insertTLI);
3937 }
3938 }
3939 }
3940
3941 FreeDir(xldir);
3942}
3943
3944/*
3945 * Recycle or remove WAL files that are not part of the given timeline's
3946 * history.
3947 *
3948 * This is called during recovery, whenever we switch to follow a new
3949 * timeline, and at the end of recovery when we create a new timeline. We
3950 * wouldn't otherwise care about extra WAL files lying in pg_wal, but they
3951 * might be leftover pre-allocated or recycled WAL segments on the old timeline
3952 * that we haven't used yet, and contain garbage. If we just leave them in
3953 * pg_wal, they will eventually be archived, and we can't let that happen.
3954 * Files that belong to our timeline history are valid, because we have
3955 * successfully replayed them, but from others we can't be sure.
3956 *
3957 * 'switchpoint' is the current point in WAL where we switch to new timeline,
3958 * and 'newTLI' is the new timeline we switch to.
3959 */
3960void
3962{
3963 DIR *xldir;
3964 struct dirent *xlde;
3965 char switchseg[MAXFNAMELEN];
3966 XLogSegNo endLogSegNo;
3967 XLogSegNo switchLogSegNo;
3968 XLogSegNo recycleSegNo;
3969
3970 /*
3971 * Initialize info about where to begin the work. This will recycle,
3972 * somewhat arbitrarily, 10 future segments.
3973 */
3974 XLByteToPrevSeg(switchpoint, switchLogSegNo, wal_segment_size);
3975 XLByteToSeg(switchpoint, endLogSegNo, wal_segment_size);
3976 recycleSegNo = endLogSegNo + 10;
3977
3978 /*
3979 * Construct a filename of the last segment to be kept.
3980 */
3981 XLogFileName(switchseg, newTLI, switchLogSegNo, wal_segment_size);
3982
3983 elog(DEBUG2, "attempting to remove WAL segments newer than log file %s",
3984 switchseg);
3985
3986 xldir = AllocateDir(XLOGDIR);
3987
3988 while ((xlde = ReadDir(xldir, XLOGDIR)) != NULL)
3989 {
3990 /* Ignore files that are not XLOG segments */
3991 if (!IsXLogFileName(xlde->d_name))
3992 continue;
3993
3994 /*
3995 * Remove files that are on a timeline older than the new one we're
3996 * switching to, but with a segment number >= the first segment on the
3997 * new timeline.
3998 */
3999 if (strncmp(xlde->d_name, switchseg, 8) < 0 &&
4000 strcmp(xlde->d_name + 8, switchseg + 8) > 0)
4001 {
4002 /*
4003 * If the file has already been marked as .ready, however, don't
4004 * remove it yet. It should be OK to remove it - files that are
4005 * not part of our timeline history are not required for recovery
4006 * - but seems safer to let them be archived and removed later.
4007 */
4008 if (!XLogArchiveIsReady(xlde->d_name))
4009 RemoveXlogFile(xlde, recycleSegNo, &endLogSegNo, newTLI);
4010 }
4011 }
4012
4013 FreeDir(xldir);
4014}
4015
4016/*
4017 * Recycle or remove a log file that's no longer needed.
4018 *
4019 * segment_de is the dirent structure of the segment to recycle or remove.
4020 * recycleSegNo is the segment number to recycle up to. endlogSegNo is
4021 * the segment number of the current (or recent) end of WAL.
4022 *
4023 * endlogSegNo gets incremented if the segment is recycled so as it is not
4024 * checked again with future callers of this function.
4025 *
4026 * insertTLI is the current timeline for XLOG insertion. Any recycled segments
4027 * should be used for this timeline.
4028 */
4029static void
4030RemoveXlogFile(const struct dirent *segment_de,
4031 XLogSegNo recycleSegNo, XLogSegNo *endlogSegNo,
4032 TimeLineID insertTLI)
4033{
4034 char path[MAXPGPATH];
4035#ifdef WIN32
4036 char newpath[MAXPGPATH];
4037#endif
4038 const char *segname = segment_de->d_name;
4039
4040 snprintf(path, MAXPGPATH, XLOGDIR "/%s", segname);
4041
4042 /*
4043 * Before deleting the file, see if it can be recycled as a future log
4044 * segment. Only recycle normal files, because we don't want to recycle
4045 * symbolic links pointing to a separate archive directory.
4046 */
4047 if (wal_recycle &&
4048 *endlogSegNo <= recycleSegNo &&
4049 XLogCtl->InstallXLogFileSegmentActive && /* callee rechecks this */
4050 get_dirent_type(path, segment_de, false, DEBUG2) == PGFILETYPE_REG &&
4051 InstallXLogFileSegment(endlogSegNo, path,
4052 true, recycleSegNo, insertTLI))
4053 {
4055 (errmsg_internal("recycled write-ahead log file \"%s\"",
4056 segname)));
4058 /* Needn't recheck that slot on future iterations */
4059 (*endlogSegNo)++;
4060 }
4061 else
4062 {
4063 /* No need for any more future segments, or recycling failed ... */
4064 int rc;
4065
4067 (errmsg_internal("removing write-ahead log file \"%s\"",
4068 segname)));
4069
4070#ifdef WIN32
4071
4072 /*
4073 * On Windows, if another process (e.g another backend) holds the file
4074 * open in FILE_SHARE_DELETE mode, unlink will succeed, but the file
4075 * will still show up in directory listing until the last handle is
4076 * closed. To avoid confusing the lingering deleted file for a live
4077 * WAL file that needs to be archived, rename it before deleting it.
4078 *
4079 * If another process holds the file open without FILE_SHARE_DELETE
4080 * flag, rename will fail. We'll try again at the next checkpoint.
4081 */
4082 snprintf(newpath, MAXPGPATH, "%s.deleted", path);
4083 if (rename(path, newpath) != 0)
4084 {
4085 ereport(LOG,
4087 errmsg("could not rename file \"%s\": %m",
4088 path)));
4089 return;
4090 }
4091 rc = durable_unlink(newpath, LOG);
4092#else
4093 rc = durable_unlink(path, LOG);
4094#endif
4095 if (rc != 0)
4096 {
4097 /* Message already logged by durable_unlink() */
4098 return;
4099 }
4101 }
4102
4103 XLogArchiveCleanup(segname);
4104}
4105
4106/*
4107 * Verify whether pg_wal, pg_wal/archive_status, and pg_wal/summaries exist.
4108 * If the latter do not exist, recreate them.
4109 *
4110 * It is not the goal of this function to verify the contents of these
4111 * directories, but to help in cases where someone has performed a cluster
4112 * copy for PITR purposes but omitted pg_wal from the copy.
4113 *
4114 * We could also recreate pg_wal if it doesn't exist, but a deliberate
4115 * policy decision was made not to. It is fairly common for pg_wal to be
4116 * a symlink, and if that was the DBA's intent then automatically making a
4117 * plain directory would result in degraded performance with no notice.
4118 */
4119static void
4121{
4122 char path[MAXPGPATH];
4123 struct stat stat_buf;
4124
4125 /* Check for pg_wal; if it doesn't exist, error out */
4126 if (stat(XLOGDIR, &stat_buf) != 0 ||
4127 !S_ISDIR(stat_buf.st_mode))
4128 ereport(FATAL,
4130 errmsg("required WAL directory \"%s\" does not exist",
4131 XLOGDIR)));
4132
4133 /* Check for archive_status */
4134 snprintf(path, MAXPGPATH, XLOGDIR "/archive_status");
4135 if (stat(path, &stat_buf) == 0)
4136 {
4137 /* Check for weird cases where it exists but isn't a directory */
4138 if (!S_ISDIR(stat_buf.st_mode))
4139 ereport(FATAL,
4141 errmsg("required WAL directory \"%s\" does not exist",
4142 path)));
4143 }
4144 else
4145 {
4146 ereport(LOG,
4147 (errmsg("creating missing WAL directory \"%s\"", path)));
4148 if (MakePGDirectory(path) < 0)
4149 ereport(FATAL,
4151 errmsg("could not create missing directory \"%s\": %m",
4152 path)));
4153 }
4154
4155 /* Check for summaries */
4156 snprintf(path, MAXPGPATH, XLOGDIR "/summaries");
4157 if (stat(path, &stat_buf) == 0)
4158 {
4159 /* Check for weird cases where it exists but isn't a directory */
4160 if (!S_ISDIR(stat_buf.st_mode))
4161 ereport(FATAL,
4162 (errmsg("required WAL directory \"%s\" does not exist",
4163 path)));
4164 }
4165 else
4166 {
4167 ereport(LOG,
4168 (errmsg("creating missing WAL directory \"%s\"", path)));
4169 if (MakePGDirectory(path) < 0)
4170 ereport(FATAL,
4171 (errmsg("could not create missing directory \"%s\": %m",
4172 path)));
4173 }
4174}
4175
4176/*
4177 * Remove previous backup history files. This also retries creation of
4178 * .ready files for any backup history files for which XLogArchiveNotify
4179 * failed earlier.
4180 */
4181static void
4183{
4184 DIR *xldir;
4185 struct dirent *xlde;
4186 char path[MAXPGPATH + sizeof(XLOGDIR)];
4187
4188 xldir = AllocateDir(XLOGDIR);
4189
4190 while ((xlde = ReadDir(xldir, XLOGDIR)) != NULL)
4191 {
4193 {
4194 if (XLogArchiveCheckDone(xlde->d_name))
4195 {
4196 elog(DEBUG2, "removing WAL backup history file \"%s\"",
4197 xlde->d_name);
4198 snprintf(path, sizeof(path), XLOGDIR "/%s", xlde->d_name);
4199 unlink(path);
4201 }
4202 }
4203 }
4204
4205 FreeDir(xldir);
4206}
4207
4208/*
4209 * I/O routines for pg_control
4210 *
4211 * *ControlFile is a buffer in shared memory that holds an image of the
4212 * contents of pg_control. WriteControlFile() initializes pg_control
4213 * given a preloaded buffer, ReadControlFile() loads the buffer from
4214 * the pg_control file (during postmaster or standalone-backend startup),
4215 * and UpdateControlFile() rewrites pg_control after we modify xlog state.
4216 * InitControlFile() fills the buffer with initial values.
4217 *
4218 * For simplicity, WriteControlFile() initializes the fields of pg_control
4219 * that are related to checking backend/database compatibility, and
4220 * ReadControlFile() verifies they are correct. We could split out the
4221 * I/O and compatibility-check functions, but there seems no need currently.
4222 */
4223
4224static void
4225InitControlFile(uint64 sysidentifier, uint32 data_checksum_version)
4226{
4227 char mock_auth_nonce[MOCK_AUTH_NONCE_LEN];
4228
4229 /*
4230 * Generate a random nonce. This is used for authentication requests that
4231 * will fail because the user does not exist. The nonce is used to create
4232 * a genuine-looking password challenge for the non-existent user, in lieu
4233 * of an actual stored password.
4234 */
4235 if (!pg_strong_random(mock_auth_nonce, MOCK_AUTH_NONCE_LEN))
4236 ereport(PANIC,
4237 (errcode(ERRCODE_INTERNAL_ERROR),
4238 errmsg("could not generate secret authorization token")));
4239
4240 memset(ControlFile, 0, sizeof(ControlFileData));
4241 /* Initialize pg_control status fields */
4242 ControlFile->system_identifier = sysidentifier;
4246
4247 /* Set important parameter values for use when replaying WAL */
4256 ControlFile->data_checksum_version = data_checksum_version;
4257}
4258
4259static void
4261{
4262 int fd;
4263 char buffer[PG_CONTROL_FILE_SIZE]; /* need not be aligned */
4264
4265 /*
4266 * Initialize version and compatibility-check fields
4267 */
4270
4271 ControlFile->maxAlign = MAXIMUM_ALIGNOF;
4273
4274 ControlFile->blcksz = BLCKSZ;
4275 ControlFile->relseg_size = RELSEG_SIZE;
4276 ControlFile->xlog_blcksz = XLOG_BLCKSZ;
4278
4281
4284
4286
4287 /* Contents are protected with a CRC */
4290 (char *) ControlFile,
4291 offsetof(ControlFileData, crc));
4293
4294 /*
4295 * We write out PG_CONTROL_FILE_SIZE bytes into pg_control, zero-padding
4296 * the excess over sizeof(ControlFileData). This reduces the odds of
4297 * premature-EOF errors when reading pg_control. We'll still fail when we
4298 * check the contents of the file, but hopefully with a more specific
4299 * error than "couldn't read pg_control".
4300 */
4301 memset(buffer, 0, PG_CONTROL_FILE_SIZE);
4302 memcpy(buffer, ControlFile, sizeof(ControlFileData));
4303
4305 O_RDWR | O_CREAT | O_EXCL | PG_BINARY);
4306 if (fd < 0)
4307 ereport(PANIC,
4309 errmsg("could not create file \"%s\": %m",
4311
4312 errno = 0;
4313 pgstat_report_wait_start(WAIT_EVENT_CONTROL_FILE_WRITE);
4315 {
4316 /* if write didn't set errno, assume problem is no disk space */
4317 if (errno == 0)
4318 errno = ENOSPC;
4319 ereport(PANIC,
4321 errmsg("could not write to file \"%s\": %m",
4323 }
4325
4326 pgstat_report_wait_start(WAIT_EVENT_CONTROL_FILE_SYNC);
4327 if (pg_fsync(fd) != 0)
4328 ereport(PANIC,
4330 errmsg("could not fsync file \"%s\": %m",
4333
4334 if (close(fd) != 0)
4335 ereport(PANIC,
4337 errmsg("could not close file \"%s\": %m",
4339}
4340
4341static void
4343{
4344 pg_crc32c crc;
4345 int fd;
4346 char wal_segsz_str[20];
4347 int r;
4348
4349 /*
4350 * Read data...
4351 */
4353 O_RDWR | PG_BINARY);
4354 if (fd < 0)
4355 ereport(PANIC,
4357 errmsg("could not open file \"%s\": %m",
4359
4360 pgstat_report_wait_start(WAIT_EVENT_CONTROL_FILE_READ);
4361 r = read(fd, ControlFile, sizeof(ControlFileData));
4362 if (r != sizeof(ControlFileData))
4363 {
4364 if (r < 0)
4365 ereport(PANIC,
4367 errmsg("could not read file \"%s\": %m",
4369 else
4370 ereport(PANIC,
4372 errmsg("could not read file \"%s\": read %d of %zu",
4373 XLOG_CONTROL_FILE, r, sizeof(ControlFileData))));
4374 }
4376
4377 close(fd);
4378
4379 /*
4380 * Check for expected pg_control format version. If this is wrong, the
4381 * CRC check will likely fail because we'll be checking the wrong number
4382 * of bytes. Complaining about wrong version will probably be more
4383 * enlightening than complaining about wrong CRC.
4384 */
4385
4387 ereport(FATAL,
4388 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
4389 errmsg("database files are incompatible with server"),
4390 errdetail("The database cluster was initialized with PG_CONTROL_VERSION %d (0x%08x),"
4391 " but the server was compiled with PG_CONTROL_VERSION %d (0x%08x).",
4394 errhint("This could be a problem of mismatched byte ordering. It looks like you need to initdb.")));
4395
4397 ereport(FATAL,
4398 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
4399 errmsg("database files are incompatible with server"),
4400 errdetail("The database cluster was initialized with PG_CONTROL_VERSION %d,"
4401 " but the server was compiled with PG_CONTROL_VERSION %d.",
4403 errhint("It looks like you need to initdb.")));
4404
4405 /* Now check the CRC. */
4408 (char *) ControlFile,
4409 offsetof(ControlFileData, crc));
4410 FIN_CRC32C(crc);
4411
4412 if (!EQ_CRC32C(crc, ControlFile->crc))
4413 ereport(FATAL,
4414 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
4415 errmsg("incorrect checksum in control file")));
4416
4417 /*
4418 * Do compatibility checking immediately. If the database isn't
4419 * compatible with the backend executable, we want to abort before we can
4420 * possibly do any damage.
4421 */
4423 ereport(FATAL,
4424 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
4425 errmsg("database files are incompatible with server"),
4426 /* translator: %s is a variable name and %d is its value */
4427 errdetail("The database cluster was initialized with %s %d,"
4428 " but the server was compiled with %s %d.",
4429 "CATALOG_VERSION_NO", ControlFile->catalog_version_no,
4430 "CATALOG_VERSION_NO", CATALOG_VERSION_NO),
4431 errhint("It looks like you need to initdb.")));
4432 if (ControlFile->maxAlign != MAXIMUM_ALIGNOF)
4433 ereport(FATAL,
4434 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
4435 errmsg("database files are incompatible with server"),
4436 /* translator: %s is a variable name and %d is its value */
4437 errdetail("The database cluster was initialized with %s %d,"
4438 " but the server was compiled with %s %d.",
4439 "MAXALIGN", ControlFile->maxAlign,
4440 "MAXALIGN", MAXIMUM_ALIGNOF),
4441 errhint("It looks like you need to initdb.")));
4443 ereport(FATAL,
4444 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
4445 errmsg("database files are incompatible with server"),
4446 errdetail("The database cluster appears to use a different floating-point number format than the server executable."),
4447 errhint("It looks like you need to initdb.")));
4448 if (ControlFile->blcksz != BLCKSZ)
4449 ereport(FATAL,
4450 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
4451 errmsg("database files are incompatible with server"),
4452 /* translator: %s is a variable name and %d is its value */
4453 errdetail("The database cluster was initialized with %s %d,"
4454 " but the server was compiled with %s %d.",
4455 "BLCKSZ", ControlFile->blcksz,
4456 "BLCKSZ", BLCKSZ),
4457 errhint("It looks like you need to recompile or initdb.")));
4458 if (ControlFile->relseg_size != RELSEG_SIZE)
4459 ereport(FATAL,
4460 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
4461 errmsg("database files are incompatible with server"),
4462 /* translator: %s is a variable name and %d is its value */
4463 errdetail("The database cluster was initialized with %s %d,"
4464 " but the server was compiled with %s %d.",
4465 "RELSEG_SIZE", ControlFile->relseg_size,
4466 "RELSEG_SIZE", RELSEG_SIZE),
4467 errhint("It looks like you need to recompile or initdb.")));
4468 if (ControlFile->xlog_blcksz != XLOG_BLCKSZ)
4469 ereport(FATAL,
4470 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
4471 errmsg("database files are incompatible with server"),
4472 /* translator: %s is a variable name and %d is its value */
4473 errdetail("The database cluster was initialized with %s %d,"
4474 " but the server was compiled with %s %d.",
4475 "XLOG_BLCKSZ", ControlFile->xlog_blcksz,
4476 "XLOG_BLCKSZ", XLOG_BLCKSZ),
4477 errhint("It looks like you need to recompile or initdb.")));
4479 ereport(FATAL,
4480 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
4481 errmsg("database files are incompatible with server"),
4482 /* translator: %s is a variable name and %d is its value */
4483 errdetail("The database cluster was initialized with %s %d,"
4484 " but the server was compiled with %s %d.",
4485 "NAMEDATALEN", ControlFile->nameDataLen,
4486 "NAMEDATALEN", NAMEDATALEN),
4487 errhint("It looks like you need to recompile or initdb.")));
4489 ereport(FATAL,
4490 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
4491 errmsg("database files are incompatible with server"),
4492 /* translator: %s is a variable name and %d is its value */
4493 errdetail("The database cluster was initialized with %s %d,"
4494 " but the server was compiled with %s %d.",
4495 "INDEX_MAX_KEYS", ControlFile->indexMaxKeys,
4496 "INDEX_MAX_KEYS", INDEX_MAX_KEYS),
4497 errhint("It looks like you need to recompile or initdb.")));
4499 ereport(FATAL,
4500 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
4501 errmsg("database files are incompatible with server"),
4502 /* translator: %s is a variable name and %d is its value */
4503 errdetail("The database cluster was initialized with %s %d,"
4504 " but the server was compiled with %s %d.",
4505 "TOAST_MAX_CHUNK_SIZE", ControlFile->toast_max_chunk_size,
4506 "TOAST_MAX_CHUNK_SIZE", (int) TOAST_MAX_CHUNK_SIZE),
4507 errhint("It looks like you need to recompile or initdb.")));
4509 ereport(FATAL,
4510 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
4511 errmsg("database files are incompatible with server"),
4512 /* translator: %s is a variable name and %d is its value */
4513 errdetail("The database cluster was initialized with %s %d,"
4514 " but the server was compiled with %s %d.",
4515 "LOBLKSIZE", ControlFile->loblksize,
4516 "LOBLKSIZE", (int) LOBLKSIZE),
4517 errhint("It looks like you need to recompile or initdb.")));
4518
4519#ifdef USE_FLOAT8_BYVAL
4520 if (ControlFile->float8ByVal != true)
4521 ereport(FATAL,
4522 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
4523 errmsg("database files are incompatible with server"),
4524 errdetail("The database cluster was initialized without USE_FLOAT8_BYVAL"
4525 " but the server was compiled with USE_FLOAT8_BYVAL."),
4526 errhint("It looks like you need to recompile or initdb.")));
4527#else
4528 if (ControlFile->float8ByVal != false)
4529 ereport(FATAL,
4530 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
4531 errmsg("database files are incompatible with server"),
4532 errdetail("The database cluster was initialized with USE_FLOAT8_BYVAL"
4533 " but the server was compiled without USE_FLOAT8_BYVAL."),
4534 errhint("It looks like you need to recompile or initdb.")));
4535#endif
4536
4538
4540 ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
4541 errmsg_plural("invalid WAL segment size in control file (%d byte)",
4542 "invalid WAL segment size in control file (%d bytes)",
4545 errdetail("The WAL segment size must be a power of two between 1 MB and 1 GB.")));
4546
4547 snprintf(wal_segsz_str, sizeof(wal_segsz_str), "%d", wal_segment_size);
4548 SetConfigOption("wal_segment_size", wal_segsz_str, PGC_INTERNAL,
4550
4551 /* check and update variables dependent on wal_segment_size */
4553 ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
4554 /* translator: both %s are GUC names */
4555 errmsg("\"%s\" must be at least twice \"%s\"",
4556 "min_wal_size", "wal_segment_size")));
4557
4559 ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
4560 /* translator: both %s are GUC names */
4561 errmsg("\"%s\" must be at least twice \"%s\"",
4562 "max_wal_size", "wal_segment_size")));
4563
4565 (wal_segment_size / XLOG_BLCKSZ * UsableBytesInPage) -
4567
4569
4570 /* Make the initdb settings visible as GUC variables, too */
4571 SetConfigOption("data_checksums", DataChecksumsEnabled() ? "yes" : "no",
4573}
4574
4575/*
4576 * Utility wrapper to update the control file. Note that the control
4577 * file gets flushed.
4578 */
4579static void
4581{
4583}
4584
4585/*
4586 * Returns the unique system identifier from control file.
4587 */
4588uint64
4590{
4591 Assert(ControlFile != NULL);
4593}
4594
4595/*
4596 * Returns the random nonce from control file.
4597 */
4598char *
4600{
4601 Assert(ControlFile != NULL);
4603}
4604
4605/*
4606 * Are checksums enabled for data pages?
4607 */
4608bool
4610{
4611 Assert(ControlFile != NULL);
4612 return (ControlFile->data_checksum_version > 0);
4613}
4614
4615/*
4616 * Returns a fake LSN for unlogged relations.
4617 *
4618 * Each call generates an LSN that is greater than any previous value
4619 * returned. The current counter value is saved and restored across clean
4620 * shutdowns, but like unlogged relations, does not survive a crash. This can
4621 * be used in lieu of real LSN values returned by XLogInsert, if you need an
4622 * LSN-like increasing sequence of numbers without writing any WAL.
4623 */
4626{
4628}
4629
4630/*
4631 * Auto-tune the number of XLOG buffers.
4632 *
4633 * The preferred setting for wal_buffers is about 3% of shared_buffers, with
4634 * a maximum of one XLOG segment (there is little reason to think that more
4635 * is helpful, at least so long as we force an fsync when switching log files)
4636 * and a minimum of 8 blocks (which was the default value prior to PostgreSQL
4637 * 9.1, when auto-tuning was added).
4638 *
4639 * This should not be called until NBuffers has received its final value.
4640 */
4641static int
4643{
4644 int xbuffers;
4645
4646 xbuffers = NBuffers / 32;
4647 if (xbuffers > (wal_segment_size / XLOG_BLCKSZ))
4648 xbuffers = (wal_segment_size / XLOG_BLCKSZ);
4649 if (xbuffers < 8)
4650 xbuffers = 8;
4651 return xbuffers;
4652}
4653
4654/*
4655 * GUC check_hook for wal_buffers
4656 */
4657bool
4659{
4660 /*
4661 * -1 indicates a request for auto-tune.
4662 */
4663 if (*newval == -1)
4664 {
4665 /*
4666 * If we haven't yet changed the boot_val default of -1, just let it
4667 * be. We'll fix it when XLOGShmemSize is called.
4668 */
4669 if (XLOGbuffers == -1)
4670 return true;
4671
4672 /* Otherwise, substitute the auto-tune value */
4674 }
4675
4676 /*
4677 * We clamp manually-set values to at least 4 blocks. Prior to PostgreSQL
4678 * 9.1, a minimum of 4 was enforced by guc.c, but since that is no longer
4679 * the case, we just silently treat such values as a request for the
4680 * minimum. (We could throw an error instead, but that doesn't seem very
4681 * helpful.)
4682 */
4683 if (*newval < 4)
4684 *newval = 4;
4685
4686 return true;
4687}
4688
4689/*
4690 * GUC check_hook for wal_consistency_checking
4691 */
4692bool
4694{
4695 char *rawstring;
4696 List *elemlist;
4697 ListCell *l;
4698 bool newwalconsistency[RM_MAX_ID + 1];
4699
4700 /* Initialize the array */
4701 MemSet(newwalconsistency, 0, (RM_MAX_ID + 1) * sizeof(bool));
4702
4703 /* Need a modifiable copy of string */
4704 rawstring = pstrdup(*newval);
4705
4706 /* Parse string into list of identifiers */
4707 if (!SplitIdentifierString(rawstring, ',', &elemlist))
4708 {
4709 /* syntax error in list */
4710 GUC_check_errdetail("List syntax is invalid.");
4711 pfree(rawstring);
4712 list_free(elemlist);
4713 return false;
4714 }
4715
4716 foreach(l, elemlist)
4717 {
4718 char *tok = (char *) lfirst(l);
4719 int rmid;
4720
4721 /* Check for 'all'. */
4722 if (pg_strcasecmp(tok, "all") == 0)
4723 {
4724 for (rmid = 0; rmid <= RM_MAX_ID; rmid++)
4725 if (RmgrIdExists(rmid) && GetRmgr(rmid).rm_mask != NULL)
4726 newwalconsistency[rmid] = true;
4727 }
4728 else
4729 {
4730 /* Check if the token matches any known resource manager. */
4731 bool found = false;
4732
4733 for (rmid = 0; rmid <= RM_MAX_ID; rmid++)
4734 {
4735 if (RmgrIdExists(rmid) && GetRmgr(rmid).rm_mask != NULL &&
4736 pg_strcasecmp(tok, GetRmgr(rmid).rm_name) == 0)
4737 {
4738 newwalconsistency[rmid] = true;
4739 found = true;
4740 break;
4741 }
4742 }
4743 if (!found)
4744 {
4745 /*
4746 * During startup, it might be a not-yet-loaded custom
4747 * resource manager. Defer checking until
4748 * InitializeWalConsistencyChecking().
4749 */
4751 {
4753 }
4754 else
4755 {
4756 GUC_check_errdetail("Unrecognized key word: \"%s\".", tok);
4757 pfree(rawstring);
4758 list_free(elemlist);
4759 return false;
4760 }
4761 }
4762 }
4763 }
4764
4765 pfree(rawstring);
4766 list_free(elemlist);
4767
4768 /* assign new value */
4769 *extra = guc_malloc(ERROR, (RM_MAX_ID + 1) * sizeof(bool));
4770 memcpy(*extra, newwalconsistency, (RM_MAX_ID + 1) * sizeof(bool));
4771 return true;
4772}
4773
4774/*
4775 * GUC assign_hook for wal_consistency_checking
4776 */
4777void
4779{
4780 /*
4781 * If some checks were deferred, it's possible that the checks will fail
4782 * later during InitializeWalConsistencyChecking(). But in that case, the
4783 * postmaster will exit anyway, so it's safe to proceed with the
4784 * assignment.
4785 *
4786 * Any built-in resource managers specified are assigned immediately,
4787 * which affects WAL created before shared_preload_libraries are
4788 * processed. Any custom resource managers specified won't be assigned
4789 * until after shared_preload_libraries are processed, but that's OK
4790 * because WAL for a custom resource manager can't be written before the
4791 * module is loaded anyway.
4792 */
4794}
4795
4796/*
4797 * InitializeWalConsistencyChecking: run after loading custom resource managers
4798 *
4799 * If any unknown resource managers were specified in the
4800 * wal_consistency_checking GUC, processing was deferred. Now that
4801 * shared_preload_libraries have been loaded, process wal_consistency_checking
4802 * again.
4803 */
4804void
4806{
4808
4810 {
4811 struct config_generic *guc;
4812
4813 guc = find_option("wal_consistency_checking", false, false, ERROR);
4814
4816
4817 set_config_option_ext("wal_consistency_checking",
4819 guc->scontext, guc->source, guc->srole,
4820 GUC_ACTION_SET, true, ERROR, false);
4821
4822 /* checking should not be deferred again */
4824 }
4825}
4826
4827/*
4828 * GUC show_hook for archive_command
4829 */
4830const char *
4832{
4833 if (XLogArchivingActive())
4834 return XLogArchiveCommand;
4835 else
4836 return "(disabled)";
4837}
4838
4839/*
4840 * GUC show_hook for in_hot_standby
4841 */
4842const char *
4844{
4845 /*
4846 * We display the actual state based on shared memory, so that this GUC
4847 * reports up-to-date state if examined intra-query. The underlying
4848 * variable (in_hot_standby_guc) changes only when we transmit a new value
4849 * to the client.
4850 */
4851 return RecoveryInProgress() ? "on" : "off";
4852}
4853
4854/*
4855 * Read the control file, set respective GUCs.
4856 *
4857 * This is to be called during startup, including a crash recovery cycle,
4858 * unless in bootstrap mode, where no control file yet exists. As there's no
4859 * usable shared memory yet (its sizing can depend on the contents of the
4860 * control file!), first store the contents in local memory. XLOGShmemInit()
4861 * will then copy it to shared memory later.
4862 *
4863 * reset just controls whether previous contents are to be expected (in the
4864 * reset case, there's a dangling pointer into old shared memory), or not.
4865 */
4866void
4868{
4869 Assert(reset || ControlFile == NULL);
4872}
4873
4874/*
4875 * Get the wal_level from the control file. For a standby, this value should be
4876 * considered as its active wal_level, because it may be different from what
4877 * was originally configured on standby.
4878 */
4881{
4882 return ControlFile->wal_level;
4883}
4884
4885/*
4886 * Initialization of shared memory for XLOG
4887 */
4888Size
4890{
4891 Size size;
4892
4893 /*
4894 * If the value of wal_buffers is -1, use the preferred auto-tune value.
4895 * This isn't an amazingly clean place to do this, but we must wait till
4896 * NBuffers has received its final value, and must do it before using the
4897 * value of XLOGbuffers to do anything important.
4898 *
4899 * We prefer to report this value's source as PGC_S_DYNAMIC_DEFAULT.
4900 * However, if the DBA explicitly set wal_buffers = -1 in the config file,
4901 * then PGC_S_DYNAMIC_DEFAULT will fail to override that and we must force
4902 * the matter with PGC_S_OVERRIDE.
4903 */
4904 if (XLOGbuffers == -1)
4905 {
4906 char buf[32];
4907
4908 snprintf(buf, sizeof(buf), "%d", XLOGChooseNumBuffers());
4909 SetConfigOption("wal_buffers", buf, PGC_POSTMASTER,
4911 if (XLOGbuffers == -1) /* failed to apply it? */
4912 SetConfigOption("wal_buffers", buf, PGC_POSTMASTER,
4914 }
4915 Assert(XLOGbuffers > 0);
4916
4917 /* XLogCtl */
4918 size = sizeof(XLogCtlData);
4919
4920 /* WAL insertion locks, plus alignment */
4922 /* xlblocks array */
4924 /* extra alignment padding for XLOG I/O buffers */
4925 size = add_size(size, Max(XLOG_BLCKSZ, PG_IO_ALIGN_SIZE));
4926 /* and the buffers themselves */
4927 size = add_size(size, mul_size(XLOG_BLCKSZ, XLOGbuffers));
4928
4929 /*
4930 * Note: we don't count ControlFileData, it comes out of the "slop factor"
4931 * added by CreateSharedMemoryAndSemaphores. This lets us use this
4932 * routine again below to compute the actual allocation size.
4933 */
4934
4935 return size;
4936}
4937
4938void
4940{
4941 bool foundCFile,
4942 foundXLog;
4943 char *allocptr;
4944 int i;
4945 ControlFileData *localControlFile;
4946
4947#ifdef WAL_DEBUG
4948
4949 /*
4950 * Create a memory context for WAL debugging that's exempt from the normal
4951 * "no pallocs in critical section" rule. Yes, that can lead to a PANIC if
4952 * an allocation fails, but wal_debug is not for production use anyway.
4953 */
4954 if (walDebugCxt == NULL)
4955 {
4957 "WAL Debug",
4959 MemoryContextAllowInCriticalSection(walDebugCxt, true);
4960 }
4961#endif
4962
4963
4964 XLogCtl = (XLogCtlData *)
4965 ShmemInitStruct("XLOG Ctl", XLOGShmemSize(), &foundXLog);
4966
4967 localControlFile = ControlFile;
4969 ShmemInitStruct("Control File", sizeof(ControlFileData), &foundCFile);
4970
4971 if (foundCFile || foundXLog)
4972 {
4973 /* both should be present or neither */
4974 Assert(foundCFile && foundXLog);
4975
4976 /* Initialize local copy of WALInsertLocks */
4978
4979 if (localControlFile)
4980 pfree(localControlFile);
4981 return;
4982 }
4983 memset(XLogCtl, 0, sizeof(XLogCtlData));
4984
4985 /*
4986 * Already have read control file locally, unless in bootstrap mode. Move
4987 * contents into shared memory.
4988 */
4989 if (localControlFile)
4990 {
4991 memcpy(ControlFile, localControlFile, sizeof(ControlFileData));
4992 pfree(localControlFile);
4993 }
4994
4995 /*
4996 * Since XLogCtlData contains XLogRecPtr fields, its sizeof should be a
4997 * multiple of the alignment for same, so no extra alignment padding is
4998 * needed here.
4999 */
5000 allocptr = ((char *) XLogCtl) + sizeof(XLogCtlData);
5001 XLogCtl->xlblocks = (pg_atomic_uint64 *) allocptr;
5002 allocptr += sizeof(pg_atomic_uint64) * XLOGbuffers;
5003
5004 for (i = 0; i < XLOGbuffers; i++)
5005 {
5007 }
5008
5009 /* WAL insertion locks. Ensure they're aligned to the full padded size */
5010 allocptr += sizeof(WALInsertLockPadded) -
5011 ((uintptr_t) allocptr) % sizeof(WALInsertLockPadded);
5013 (WALInsertLockPadded *) allocptr;
5014 allocptr += sizeof(WALInsertLockPadded) * NUM_XLOGINSERT_LOCKS;
5015
5016 for (i = 0; i < NUM_XLOGINSERT_LOCKS; i++)
5017 {
5021 }
5022
5023 /*
5024 * Align the start of the page buffers to a full xlog block size boundary.
5025 * This simplifies some calculations in XLOG insertion. It is also
5026 * required for O_DIRECT.
5027 */
5028 allocptr = (char *) TYPEALIGN(XLOG_BLCKSZ, allocptr);
5029 XLogCtl->pages = allocptr;
5030 memset(XLogCtl->pages, 0, (Size) XLOG_BLCKSZ * XLOGbuffers);
5031
5032 /*
5033 * Do basic initialization of XLogCtl shared data. (StartupXLOG will fill
5034 * in additional info.)
5035 */
5039 XLogCtl->WalWriterSleeping = false;
5040
5047}
5048
5049/*
5050 * This func must be called ONCE on system install. It creates pg_control
5051 * and the initial XLOG segment.
5052 */
5053void
5054BootStrapXLOG(uint32 data_checksum_version)
5055{
5056 CheckPoint checkPoint;
5057 char *buffer;
5058 XLogPageHeader page;
5059 XLogLongPageHeader longpage;
5060 XLogRecord *record;
5061 char *recptr;
5062 uint64 sysidentifier;
5063 struct timeval tv;
5064 pg_crc32c crc;
5065
5066 /* allow ordinary WAL segment creation, like StartupXLOG() would */
5068
5069 /*
5070 * Select a hopefully-unique system identifier code for this installation.
5071 * We use the result of gettimeofday(), including the fractional seconds
5072 * field, as being about as unique as we can easily get. (Think not to
5073 * use random(), since it hasn't been seeded and there's no portable way
5074 * to seed it other than the system clock value...) The upper half of the
5075 * uint64 value is just the tv_sec part, while the lower half contains the
5076 * tv_usec part (which must fit in 20 bits), plus 12 bits from our current
5077 * PID for a little extra uniqueness. A person knowing this encoding can
5078 * determine the initialization time of the installation, which could
5079 * perhaps be useful sometimes.
5080 */
5081 gettimeofday(&tv, NULL);
5082 sysidentifier = ((uint64) tv.tv_sec) << 32;
5083 sysidentifier |= ((uint64) tv.tv_usec) << 12;
5084 sysidentifier |= getpid() & 0xFFF;
5085
5086 /* page buffer must be aligned suitably for O_DIRECT */
5087 buffer = (char *) palloc(XLOG_BLCKSZ + XLOG_BLCKSZ);
5088 page = (XLogPageHeader) TYPEALIGN(XLOG_BLCKSZ, buffer);
5089 memset(page, 0, XLOG_BLCKSZ);
5090
5091 /*
5092 * Set up information for the initial checkpoint record
5093 *
5094 * The initial checkpoint record is written to the beginning of the WAL
5095 * segment with logid=0 logseg=1. The very first WAL segment, 0/0, is not
5096 * used, so that we can use 0/0 to mean "before any valid WAL segment".
5097 */
5101 checkPoint.fullPageWrites = fullPageWrites;
5102 checkPoint.wal_level = wal_level;
5103 checkPoint.nextXid =
5105 checkPoint.nextOid = FirstGenbkiObjectId;
5106 checkPoint.nextMulti = FirstMultiXactId;
5107 checkPoint.nextMultiOffset = 0;
5109 checkPoint.oldestXidDB = Template1DbOid;
5110 checkPoint.oldestMulti = FirstMultiXactId;
5111 checkPoint.oldestMultiDB = Template1DbOid;
5114 checkPoint.time = (pg_time_t) time(NULL);
5116
5117 TransamVariables->nextXid = checkPoint.nextXid;
5118 TransamVariables->nextOid = checkPoint.nextOid;
5120 MultiXactSetNextMXact(checkPoint.nextMulti, checkPoint.nextMultiOffset);
5121 AdvanceOldestClogXid(checkPoint.oldestXid);
5122 SetTransactionIdLimit(checkPoint.oldestXid, checkPoint.oldestXidDB);
5123 SetMultiXactIdLimit(checkPoint.oldestMulti, checkPoint.oldestMultiDB, true);
5125
5126 /* Set up the XLOG page header */
5127 page->xlp_magic = XLOG_PAGE_MAGIC;
5128 page->xlp_info = XLP_LONG_HEADER;
5131 longpage = (XLogLongPageHeader) page;
5132 longpage->xlp_sysid = sysidentifier;
5133 longpage->xlp_seg_size = wal_segment_size;
5134 longpage->xlp_xlog_blcksz = XLOG_BLCKSZ;
5135
5136 /* Insert the initial checkpoint record */
5137 recptr = ((char *) page + SizeOfXLogLongPHD);
5138 record = (XLogRecord *) recptr;
5139 record->xl_prev = 0;
5140 record->xl_xid = InvalidTransactionId;
5141 record->xl_tot_len = SizeOfXLogRecord + SizeOfXLogRecordDataHeaderShort + sizeof(checkPoint);
5143 record->xl_rmid = RM_XLOG_ID;
5144 recptr += SizeOfXLogRecord;
5145 /* fill the XLogRecordDataHeaderShort struct */
5146 *(recptr++) = (char) XLR_BLOCK_ID_DATA_SHORT;
5147 *(recptr++) = sizeof(checkPoint);
5148 memcpy(recptr, &checkPoint, sizeof(checkPoint));
5149 recptr += sizeof(checkPoint);
5150 Assert(recptr - (char *) record == record->xl_tot_len);
5151
5153 COMP_CRC32C(crc, ((char *) record) + SizeOfXLogRecord, record->xl_tot_len - SizeOfXLogRecord);
5154 COMP_CRC32C(crc, (char *) record, offsetof(XLogRecord, xl_crc));
5155 FIN_CRC32C(crc);
5156 record->xl_crc = crc;
5157
5158 /* Create first XLOG segment file */
5161
5162 /*
5163 * We needn't bother with Reserve/ReleaseExternalFD here, since we'll
5164 * close the file again in a moment.
5165 */
5166
5167 /* Write the first page with the initial record */
5168 errno = 0;
5169 pgstat_report_wait_start(WAIT_EVENT_WAL_BOOTSTRAP_WRITE);
5170 if (write(openLogFile, page, XLOG_BLCKSZ) != XLOG_BLCKSZ)
5171 {
5172 /* if write didn't set errno, assume problem is no disk space */
5173 if (errno == 0)
5174 errno = ENOSPC;
5175 ereport(PANIC,
5177 errmsg("could not write bootstrap write-ahead log file: %m")));
5178 }
5180
5181 pgstat_report_wait_start(WAIT_EVENT_WAL_BOOTSTRAP_SYNC);
5182 if (pg_fsync(openLogFile) != 0)
5183 ereport(PANIC,
5185 errmsg("could not fsync bootstrap write-ahead log file: %m")));
5187
5188 if (close(openLogFile) != 0)
5189 ereport(PANIC,
5191 errmsg("could not close bootstrap write-ahead log file: %m")));
5192
5193 openLogFile = -1;
5194
5195 /* Now create pg_control */
5196 InitControlFile(sysidentifier, data_checksum_version);
5197 ControlFile->time = checkPoint.time;
5198 ControlFile->checkPoint = checkPoint.redo;
5199 ControlFile->checkPointCopy = checkPoint;
5200
5201 /* some additional ControlFile fields are set in WriteControlFile() */
5203
5204 /* Bootstrap the commit log, too */
5205 BootStrapCLOG();
5209
5210 pfree(buffer);
5211
5212 /*
5213 * Force control file to be read - in contrast to normal processing we'd
5214 * otherwise never run the checks and GUC related initializations therein.
5215 */
5217}
5218
5219static char *
5221{
5222 char *buf = palloc(128);
5223
5224 pg_strftime(buf, 128,
5225 "%Y-%m-%d %H:%M:%S %Z",
5226 pg_localtime(&tnow, log_timezone));
5227
5228 return buf;
5229}
5230
5231/*
5232 * Initialize the first WAL segment on new timeline.
5233 */
5234static void
5236{
5237 char xlogfname[MAXFNAMELEN];
5238 XLogSegNo endLogSegNo;
5239 XLogSegNo startLogSegNo;
5240
5241 /* we always switch to a new timeline after archive recovery */
5242 Assert(endTLI != newTLI);
5243
5244 /*
5245 * Update min recovery point one last time.
5246 */
5248
5249 /*
5250 * Calculate the last segment on the old timeline, and the first segment
5251 * on the new timeline. If the switch happens in the middle of a segment,
5252 * they are the same, but if the switch happens exactly at a segment
5253 * boundary, startLogSegNo will be endLogSegNo + 1.
5254 */
5255 XLByteToPrevSeg(endOfLog, endLogSegNo, wal_segment_size);
5256 XLByteToSeg(endOfLog, startLogSegNo, wal_segment_size);
5257
5258 /*
5259