xlogreader-heikki-2.patch
text/x-diff
Filename: xlogreader-heikki-2.patch
Type: text/x-diff
Part: 0
Patch
Same data as JSON:
GET /api/v1/attachments/:id/patch
the parsed metadata as JSON — format, series position, per-file stats; never the diff bytes.
API reference →
Format: context
| File | + | − |
|---|---|---|
| src/backend/access/transam/Makefile | 1 | 0 |
| src/backend/access/transam/xlog.c | 290 | 0 |
| src/backend/access/transam/xlogreader.c | 532 | 0 |
| src/include/access/xlog_internal.h | 8 | 0 |
| src/include/access/xlogreader.h | 97 | 0 |
*** a/src/backend/access/transam/Makefile
--- b/src/backend/access/transam/Makefile
***************
*** 14,20 **** include $(top_builddir)/src/Makefile.global
OBJS = clog.o transam.o varsup.o xact.o rmgr.o slru.o subtrans.o multixact.o \
timeline.o twophase.o twophase_rmgr.o xlog.o xlogarchive.o xlogfuncs.o \
! xlogutils.o
include $(top_srcdir)/src/backend/common.mk
--- 14,20 ----
OBJS = clog.o transam.o varsup.o xact.o rmgr.o slru.o subtrans.o multixact.o \
timeline.o twophase.o twophase_rmgr.o xlog.o xlogarchive.o xlogfuncs.o \
! xlogreader.o xlogutils.o
include $(top_srcdir)/src/backend/common.mk
*** a/src/backend/access/transam/xlog.c
--- b/src/backend/access/transam/xlog.c
***************
*** 30,35 ****
--- 30,36 ----
#include "access/twophase.h"
#include "access/xact.h"
#include "access/xlog_internal.h"
+ #include "access/xlogreader.h"
#include "access/xlogutils.h"
#include "catalog/catversion.h"
#include "catalog/pg_control.h"
***************
*** 192,205 **** static bool LocalHotStandbyActive = false;
*/
static int LocalXLogInsertAllowed = -1;
! /* Are we recovering using offline XLOG archives? (only valid in the startup process) */
! bool InArchiveRecovery = false;
/* Was the last xlog file restored from archive, or local? */
static bool restoredFromArchive = false;
/* options taken from recovery.conf for archive recovery */
! char *recoveryRestoreCommand = NULL;
static char *recoveryEndCommand = NULL;
static char *archiveCleanupCommand = NULL;
static RecoveryTargetType recoveryTarget = RECOVERY_TARGET_UNSET;
--- 193,209 ----
*/
static int LocalXLogInsertAllowed = -1;
! /*
! * Are we recovering using offline XLOG archives? (only valid in the startup
! * process)
! */
! bool InArchiveRecovery = false;
/* Was the last xlog file restored from archive, or local? */
static bool restoredFromArchive = false;
/* options taken from recovery.conf for archive recovery */
! char *recoveryRestoreCommand = NULL;
static char *recoveryEndCommand = NULL;
static char *archiveCleanupCommand = NULL;
static RecoveryTargetType recoveryTarget = RECOVERY_TARGET_UNSET;
***************
*** 210,216 **** static TimestampTz recoveryTargetTime;
static char *recoveryTargetName;
/* options taken from recovery.conf for XLOG streaming */
! bool StandbyMode = false;
static char *PrimaryConnInfo = NULL;
static char *TriggerFile = NULL;
--- 214,220 ----
static char *recoveryTargetName;
/* options taken from recovery.conf for XLOG streaming */
! bool StandbyMode = false;
static char *PrimaryConnInfo = NULL;
static char *TriggerFile = NULL;
***************
*** 389,395 **** typedef struct XLogCtlData
uint32 ckptXidEpoch; /* nextXID & epoch of latest checkpoint */
TransactionId ckptXid;
XLogRecPtr asyncXactLSN; /* LSN of newest async commit/abort */
! XLogSegNo lastRemovedSegNo; /* latest removed/recycled XLOG segment */
/* Protected by WALWriteLock: */
XLogCtlWrite Write;
--- 393,400 ----
uint32 ckptXidEpoch; /* nextXID & epoch of latest checkpoint */
TransactionId ckptXid;
XLogRecPtr asyncXactLSN; /* LSN of newest async commit/abort */
! XLogSegNo lastRemovedSegNo; /* latest removed/recycled XLOG
! * segment */
/* Protected by WALWriteLock: */
XLogCtlWrite Write;
***************
*** 530,554 **** static XLogSegNo openLogSegNo = 0;
static uint32 openLogOff = 0;
/*
! * These variables are used similarly to the ones above, but for reading
* the XLOG. Note, however, that readOff generally represents the offset
* of the page just read, not the seek position of the FD itself, which
* will be just past that page. readLen indicates how much of the current
* page has been read into readBuf, and readSource indicates where we got
* the currently open file from.
*/
! static int readFile = -1;
! static XLogSegNo readSegNo = 0;
! static uint32 readOff = 0;
! static uint32 readLen = 0;
! static bool readFileHeaderValidated = false;
! static int readSource = 0; /* XLOG_FROM_* code */
!
! /*
! * Keeps track of which sources we've tried to read the current WAL
! * record from and failed.
! */
! static int failedSources = 0; /* OR of XLOG_FROM_* codes */
/*
* These variables track when we last obtained some WAL data to process,
--- 535,563 ----
static uint32 openLogOff = 0;
/*
! * Status data for XLogPageRead.
! *
! * The first three are used similarly to the ones above, but for reading
* the XLOG. Note, however, that readOff generally represents the offset
* of the page just read, not the seek position of the FD itself, which
* will be just past that page. readLen indicates how much of the current
* page has been read into readBuf, and readSource indicates where we got
* the currently open file from.
+ *
+ * failedSources keeps track of which sources we've tried to read the current
+ * WAL record from and failed.
*/
! typedef struct XLogPageReadPrivate
! {
! int readFile;
! XLogSegNo readSegNo;
! uint32 readOff;
! uint32 readLen;
! bool readFileHeaderValidated;
! bool fetching_ckpt; /* are we fetching a checkpoint record? */
! int readSource; /* XLOG_FROM_* code */
! int failedSources; /* OR of XLOG_FROM_* codes */
! } XLogPageReadPrivate;
/*
* These variables track when we last obtained some WAL data to process,
***************
*** 559,571 **** static int failedSources = 0; /* OR of XLOG_FROM_* codes */
static TimestampTz XLogReceiptTime = 0;
static int XLogReceiptSource = 0; /* XLOG_FROM_* code */
- /* Buffer for currently read page (XLOG_BLCKSZ bytes) */
- static char *readBuf = NULL;
-
- /* Buffer for current ReadRecord result (expandable) */
- static char *readRecordBuf = NULL;
- static uint32 readRecordBufSize = 0;
-
/* State information for XLOG reading */
static XLogRecPtr ReadRecPtr; /* start of last record read */
static XLogRecPtr EndRecPtr; /* end+1 of last record read */
--- 568,573 ----
***************
*** 608,614 **** typedef struct xl_restore_point
static void readRecoveryCommandFile(void);
! static void exitArchiveRecovery(TimeLineID endTLI, XLogSegNo endLogSegNo);
static bool recoveryStopsHere(XLogRecord *record, bool *includeThis);
static void recoveryPausesHere(void);
static void SetLatestXTime(TimestampTz xtime);
--- 610,617 ----
static void readRecoveryCommandFile(void);
! static void exitArchiveRecovery(XLogPageReadPrivate *private, TimeLineID endTLI,
! XLogSegNo endLogSegNo);
static bool recoveryStopsHere(XLogRecord *record, bool *includeThis);
static void recoveryPausesHere(void);
static void SetLatestXTime(TimestampTz xtime);
***************
*** 627,640 **** static void XLogWrite(XLogwrtRqst WriteRqst, bool flexible, bool xlog_switch);
static bool InstallXLogFileSegment(XLogSegNo *segno, char *tmppath,
bool find_free, int *max_advance,
bool use_lock);
! static int XLogFileRead(XLogSegNo segno, int emode, TimeLineID tli,
! int source, bool notexistOk);
! static int XLogFileReadAnyTLI(XLogSegNo segno, int emode, int sources);
! static bool XLogPageRead(XLogRecPtr *RecPtr, int emode, bool fetching_ckpt,
! bool randAccess);
! static bool WaitForWALToBecomeAvailable(XLogRecPtr RecPtr, bool randAccess,
! bool fetching_ckpt);
! static int emode_for_corrupt_record(int emode, XLogRecPtr RecPtr);
static void XLogFileClose(void);
static void PreallocXlogFiles(XLogRecPtr endptr);
static void RemoveOldXlogFiles(XLogSegNo segno, XLogRecPtr endptr);
--- 630,643 ----
static bool InstallXLogFileSegment(XLogSegNo *segno, char *tmppath,
bool find_free, int *max_advance,
bool use_lock);
! static int XLogFileRead(XLogPageReadPrivate *private, XLogSegNo segno,
! int emode, TimeLineID tli, int source, bool notexistOk);
! static int XLogFileReadAnyTLI(XLogPageReadPrivate *private, XLogSegNo segno,
! int emode, int sources);
! static bool XLogPageRead(XLogReaderState *xlogreader, XLogRecPtr RecPtr,
! int emode, bool randAccess, char *readBuf, void *private_data);
! static bool WaitForWALToBecomeAvailable(XLogPageReadPrivate *private,
! XLogRecPtr RecPtr, bool randAccess);
static void XLogFileClose(void);
static void PreallocXlogFiles(XLogRecPtr endptr);
static void RemoveOldXlogFiles(XLogSegNo segno, XLogRecPtr endptr);
***************
*** 642,653 **** static void UpdateLastRemovedPtr(char *filename);
static void ValidateXLOGDirectoryStructure(void);
static void CleanupBackupHistory(void);
static void UpdateMinRecoveryPoint(XLogRecPtr lsn, bool force);
! static XLogRecord *ReadRecord(XLogRecPtr *RecPtr, int emode, bool fetching_ckpt);
! static void CheckRecoveryConsistency(void);
! static bool ValidXLogPageHeader(XLogPageHeader hdr, int emode);
! static bool ValidXLogRecordHeader(XLogRecPtr *RecPtr, XLogRecord *record,
! int emode, bool randAccess);
! static XLogRecord *ReadCheckpointRecord(XLogRecPtr RecPtr, int whichChkpt);
static bool rescanLatestTimeLine(void);
static void WriteControlFile(void);
static void ReadControlFile(void);
--- 645,657 ----
static void ValidateXLOGDirectoryStructure(void);
static void CleanupBackupHistory(void);
static void UpdateMinRecoveryPoint(XLogRecPtr lsn, bool force);
! static XLogRecord *ReadRecord(XLogReaderState *xlogreader, XLogRecPtr RecPtr,
! int emode, bool fetching_ckpt);
! static void CheckRecoveryConsistency(XLogRecPtr EndRecPtr);
! static bool ValidXLogPageHeader(XLogSegNo segno, uint32 offset, int source,
! XLogPageHeader hdr, int emode);
! static XLogRecord *ReadCheckpointRecord(XLogReaderState *xlogreader,
! XLogRecPtr RecPtr, int whichChkpt);
static bool rescanLatestTimeLine(void);
static void WriteControlFile(void);
static void ReadControlFile(void);
***************
*** 1514,1520 **** XLogWrite(XLogwrtRqst WriteRqst, bool flexible, bool xlog_switch)
*/
if (!XLByteLT(LogwrtResult.Write, XLogCtl->xlblocks[curridx]))
elog(PANIC, "xlog write request %X/%X is past end of log %X/%X",
! (uint32) (LogwrtResult.Write >> 32), (uint32) LogwrtResult.Write,
(uint32) (XLogCtl->xlblocks[curridx] >> 32),
(uint32) XLogCtl->xlblocks[curridx]);
--- 1518,1524 ----
*/
if (!XLByteLT(LogwrtResult.Write, XLogCtl->xlblocks[curridx]))
elog(PANIC, "xlog write request %X/%X is past end of log %X/%X",
! (uint32) (LogwrtResult.Write >> 32), (uint32) LogwrtResult.Write,
(uint32) (XLogCtl->xlblocks[curridx] >> 32),
(uint32) XLogCtl->xlblocks[curridx]);
***************
*** 1580,1588 **** XLogWrite(XLogwrtRqst WriteRqst, bool flexible, bool xlog_switch)
if (lseek(openLogFile, (off_t) startoffset, SEEK_SET) < 0)
ereport(PANIC,
(errcode_for_file_access(),
! errmsg("could not seek in log file %s to offset %u: %m",
! XLogFileNameP(ThisTimeLineID, openLogSegNo),
! startoffset)));
openLogOff = startoffset;
}
--- 1584,1592 ----
if (lseek(openLogFile, (off_t) startoffset, SEEK_SET) < 0)
ereport(PANIC,
(errcode_for_file_access(),
! errmsg("could not seek in log file %s to offset %u: %m",
! XLogFileNameP(ThisTimeLineID, openLogSegNo),
! startoffset)));
openLogOff = startoffset;
}
***************
*** 1823,1829 **** UpdateMinRecoveryPoint(XLogRecPtr lsn, bool force)
if (!force && XLByteLT(newMinRecoveryPoint, lsn))
elog(WARNING,
"xlog min recovery request %X/%X is past current point %X/%X",
! (uint32) (lsn >> 32) , (uint32) lsn,
(uint32) (newMinRecoveryPoint >> 32),
(uint32) newMinRecoveryPoint);
--- 1827,1833 ----
if (!force && XLByteLT(newMinRecoveryPoint, lsn))
elog(WARNING,
"xlog min recovery request %X/%X is past current point %X/%X",
! (uint32) (lsn >> 32), (uint32) lsn,
(uint32) (newMinRecoveryPoint >> 32),
(uint32) newMinRecoveryPoint);
***************
*** 1877,1883 **** XLogFlush(XLogRecPtr record)
elog(LOG, "xlog flush request %X/%X; write %X/%X; flush %X/%X",
(uint32) (record >> 32), (uint32) record,
(uint32) (LogwrtResult.Write >> 32), (uint32) LogwrtResult.Write,
! (uint32) (LogwrtResult.Flush >> 32), (uint32) LogwrtResult.Flush);
#endif
START_CRIT_SECTION();
--- 1881,1887 ----
elog(LOG, "xlog flush request %X/%X; write %X/%X; flush %X/%X",
(uint32) (record >> 32), (uint32) record,
(uint32) (LogwrtResult.Write >> 32), (uint32) LogwrtResult.Write,
! (uint32) (LogwrtResult.Flush >> 32), (uint32) LogwrtResult.Flush);
#endif
START_CRIT_SECTION();
***************
*** 1941,1948 **** XLogFlush(XLogRecPtr record)
/*
* Sleep before flush! By adding a delay here, we may give further
* backends the opportunity to join the backlog of group commit
! * followers; this can significantly improve transaction throughput, at
! * the risk of increasing transaction latency.
*
* We do not sleep if enableFsync is not turned on, nor if there are
* fewer than CommitSiblings other backends with active transactions.
--- 1945,1952 ----
/*
* Sleep before flush! By adding a delay here, we may give further
* backends the opportunity to join the backlog of group commit
! * followers; this can significantly improve transaction throughput,
! * at the risk of increasing transaction latency.
*
* We do not sleep if enableFsync is not turned on, nor if there are
* fewer than CommitSiblings other backends with active transactions.
***************
*** 1957,1963 **** XLogFlush(XLogRecPtr record)
XLogCtlInsert *Insert = &XLogCtl->Insert;
uint32 freespace = INSERT_FREESPACE(Insert);
! if (freespace == 0) /* buffer is full */
WriteRqstPtr = XLogCtl->xlblocks[Insert->curridx];
else
{
--- 1961,1967 ----
XLogCtlInsert *Insert = &XLogCtl->Insert;
uint32 freespace = INSERT_FREESPACE(Insert);
! if (freespace == 0) /* buffer is full */
WriteRqstPtr = XLogCtl->xlblocks[Insert->curridx];
else
{
***************
*** 2010,2016 **** XLogFlush(XLogRecPtr record)
elog(ERROR,
"xlog flush request %X/%X is not satisfied --- flushed only to %X/%X",
(uint32) (record >> 32), (uint32) record,
! (uint32) (LogwrtResult.Flush >> 32), (uint32) LogwrtResult.Flush);
}
/*
--- 2014,2020 ----
elog(ERROR,
"xlog flush request %X/%X is not satisfied --- flushed only to %X/%X",
(uint32) (record >> 32), (uint32) record,
! (uint32) (LogwrtResult.Flush >> 32), (uint32) LogwrtResult.Flush);
}
/*
***************
*** 2089,2095 **** XLogBackgroundFlush(void)
elog(LOG, "xlog bg flush request %X/%X; write %X/%X; flush %X/%X",
(uint32) (WriteRqstPtr >> 32), (uint32) WriteRqstPtr,
(uint32) (LogwrtResult.Write >> 32), (uint32) LogwrtResult.Write,
! (uint32) (LogwrtResult.Flush >> 32), (uint32) LogwrtResult.Flush);
#endif
START_CRIT_SECTION();
--- 2093,2099 ----
elog(LOG, "xlog bg flush request %X/%X; write %X/%X; flush %X/%X",
(uint32) (WriteRqstPtr >> 32), (uint32) WriteRqstPtr,
(uint32) (LogwrtResult.Write >> 32), (uint32) LogwrtResult.Write,
! (uint32) (LogwrtResult.Flush >> 32), (uint32) LogwrtResult.Flush);
#endif
START_CRIT_SECTION();
***************
*** 2329,2335 **** XLogFileInit(XLogSegNo logsegno, bool *use_existent, bool use_lock)
if (fd < 0)
ereport(ERROR,
(errcode_for_file_access(),
! errmsg("could not open file \"%s\": %m", path)));
elog(DEBUG2, "done creating and filling new WAL file");
--- 2333,2339 ----
if (fd < 0)
ereport(ERROR,
(errcode_for_file_access(),
! errmsg("could not open file \"%s\": %m", path)));
elog(DEBUG2, "done creating and filling new WAL file");
***************
*** 2568,2575 **** XLogFileOpen(XLogSegNo segno)
* Otherwise, it's assumed to be already available in pg_xlog.
*/
static int
! XLogFileRead(XLogSegNo segno, int emode, TimeLineID tli,
! int source, bool notfoundOk)
{
char xlogfname[MAXFNAMELEN];
char activitymsg[MAXFNAMELEN + 16];
--- 2572,2579 ----
* Otherwise, it's assumed to be already available in pg_xlog.
*/
static int
! XLogFileRead(XLogPageReadPrivate *private, XLogSegNo segno, int emode,
! TimeLineID tli, int source, bool notfoundOk)
{
char xlogfname[MAXFNAMELEN];
char activitymsg[MAXFNAMELEN + 16];
***************
*** 2616,2624 **** XLogFileRead(XLogSegNo segno, int emode, TimeLineID tli,
XLogFilePath(xlogfpath, tli, segno);
if (stat(xlogfpath, &statbuf) == 0)
{
! char oldpath[MAXPGPATH];
#ifdef WIN32
static unsigned int deletedcounter = 1;
/*
* On Windows, if another process (e.g a walsender process) holds
* the file open in FILE_SHARE_DELETE mode, unlink will succeed,
--- 2620,2630 ----
XLogFilePath(xlogfpath, tli, segno);
if (stat(xlogfpath, &statbuf) == 0)
{
! char oldpath[MAXPGPATH];
!
#ifdef WIN32
static unsigned int deletedcounter = 1;
+
/*
* On Windows, if another process (e.g a walsender process) holds
* the file open in FILE_SHARE_DELETE mode, unlink will succeed,
***************
*** 2685,2698 **** XLogFileRead(XLogSegNo segno, int emode, TimeLineID tli,
set_ps_display(activitymsg, false);
/* Track source of data in assorted state variables */
! readSource = source;
XLogReceiptSource = source;
/* In FROM_STREAM case, caller tracks receipt time, not me */
if (source != XLOG_FROM_STREAM)
XLogReceiptTime = GetCurrentTimestamp();
/* The file header needs to be validated on first access */
! readFileHeaderValidated = false;
return fd;
}
--- 2691,2704 ----
set_ps_display(activitymsg, false);
/* Track source of data in assorted state variables */
! private->readSource = source;
XLogReceiptSource = source;
/* In FROM_STREAM case, caller tracks receipt time, not me */
if (source != XLOG_FROM_STREAM)
XLogReceiptTime = GetCurrentTimestamp();
/* The file header needs to be validated on first access */
! private->readFileHeaderValidated = false;
return fd;
}
***************
*** 2709,2715 **** XLogFileRead(XLogSegNo segno, int emode, TimeLineID tli,
* This version searches for the segment with any TLI listed in expectedTLIs.
*/
static int
! XLogFileReadAnyTLI(XLogSegNo segno, int emode, int sources)
{
char path[MAXPGPATH];
ListCell *cell;
--- 2715,2722 ----
* This version searches for the segment with any TLI listed in expectedTLIs.
*/
static int
! XLogFileReadAnyTLI(XLogPageReadPrivate *private, XLogSegNo segno, int emode,
! int sources)
{
char path[MAXPGPATH];
ListCell *cell;
***************
*** 2734,2740 **** XLogFileReadAnyTLI(XLogSegNo segno, int emode, int sources)
if (sources & XLOG_FROM_ARCHIVE)
{
! fd = XLogFileRead(segno, emode, tli, XLOG_FROM_ARCHIVE, true);
if (fd != -1)
{
elog(DEBUG1, "got WAL segment from archive");
--- 2741,2748 ----
if (sources & XLOG_FROM_ARCHIVE)
{
! fd = XLogFileRead(private, segno, emode, tli,
! XLOG_FROM_ARCHIVE, true);
if (fd != -1)
{
elog(DEBUG1, "got WAL segment from archive");
***************
*** 2744,2750 **** XLogFileReadAnyTLI(XLogSegNo segno, int emode, int sources)
if (sources & XLOG_FROM_PG_XLOG)
{
! fd = XLogFileRead(segno, emode, tli, XLOG_FROM_PG_XLOG, true);
if (fd != -1)
return fd;
}
--- 2752,2759 ----
if (sources & XLOG_FROM_PG_XLOG)
{
! fd = XLogFileRead(private, segno, emode, tli,
! XLOG_FROM_PG_XLOG, true);
if (fd != -1)
return fd;
}
***************
*** 3177,3278 **** RestoreBackupBlock(XLogRecPtr lsn, XLogRecord *record, int block_index,
}
/*
- * CRC-check an XLOG record. We do not believe the contents of an XLOG
- * record (other than to the minimal extent of computing the amount of
- * data to read in) until we've checked the CRCs.
- *
- * We assume all of the record (that is, xl_tot_len bytes) has been read
- * into memory at *record. Also, ValidXLogRecordHeader() has accepted the
- * record's header, which means in particular that xl_tot_len is at least
- * SizeOfXlogRecord, so it is safe to fetch xl_len.
- */
- static bool
- RecordIsValid(XLogRecord *record, XLogRecPtr recptr, int emode)
- {
- pg_crc32 crc;
- int i;
- uint32 len = record->xl_len;
- BkpBlock bkpb;
- char *blk;
- size_t remaining = record->xl_tot_len;
-
- /* First the rmgr data */
- if (remaining < SizeOfXLogRecord + len)
- {
- /* ValidXLogRecordHeader() should've caught this already... */
- ereport(emode_for_corrupt_record(emode, recptr),
- (errmsg("invalid record length at %X/%X",
- (uint32) (recptr >> 32), (uint32) recptr)));
- return false;
- }
- remaining -= SizeOfXLogRecord + len;
- INIT_CRC32(crc);
- COMP_CRC32(crc, XLogRecGetData(record), len);
-
- /* Add in the backup blocks, if any */
- blk = (char *) XLogRecGetData(record) + len;
- for (i = 0; i < XLR_MAX_BKP_BLOCKS; i++)
- {
- uint32 blen;
-
- if (!(record->xl_info & XLR_BKP_BLOCK(i)))
- continue;
-
- if (remaining < sizeof(BkpBlock))
- {
- ereport(emode_for_corrupt_record(emode, recptr),
- (errmsg("invalid backup block size in record at %X/%X",
- (uint32) (recptr >> 32), (uint32) recptr)));
- return false;
- }
- memcpy(&bkpb, blk, sizeof(BkpBlock));
-
- if (bkpb.hole_offset + bkpb.hole_length > BLCKSZ)
- {
- ereport(emode_for_corrupt_record(emode, recptr),
- (errmsg("incorrect hole size in record at %X/%X",
- (uint32) (recptr >> 32), (uint32) recptr)));
- return false;
- }
- blen = sizeof(BkpBlock) + BLCKSZ - bkpb.hole_length;
-
- if (remaining < blen)
- {
- ereport(emode_for_corrupt_record(emode, recptr),
- (errmsg("invalid backup block size in record at %X/%X",
- (uint32) (recptr >> 32), (uint32) recptr)));
- return false;
- }
- remaining -= blen;
- COMP_CRC32(crc, blk, blen);
- blk += blen;
- }
-
- /* Check that xl_tot_len agrees with our calculation */
- if (remaining != 0)
- {
- ereport(emode_for_corrupt_record(emode, recptr),
- (errmsg("incorrect total length in record at %X/%X",
- (uint32) (recptr >> 32), (uint32) recptr)));
- return false;
- }
-
- /* Finally include the record header */
- COMP_CRC32(crc, (char *) record, offsetof(XLogRecord, xl_crc));
- FIN_CRC32(crc);
-
- if (!EQ_CRC32(record->xl_crc, crc))
- {
- ereport(emode_for_corrupt_record(emode, recptr),
- (errmsg("incorrect resource manager data checksum in record at %X/%X",
- (uint32) (recptr >> 32), (uint32) recptr)));
- return false;
- }
-
- return true;
- }
-
- /*
* Attempt to read an XLOG record.
*
* If RecPtr is not NULL, try to read a record at that position. Otherwise
--- 3186,3191 ----
***************
*** 3285,3605 **** RecordIsValid(XLogRecord *record, XLogRecPtr recptr, int emode)
* the returned record pointer always points there.
*/
static XLogRecord *
! ReadRecord(XLogRecPtr *RecPtr, int emode, bool fetching_ckpt)
{
XLogRecord *record;
! XLogRecPtr tmpRecPtr = EndRecPtr;
! bool randAccess = false;
! uint32 len,
! total_len;
! uint32 targetRecOff;
! uint32 pageHeaderSize;
! bool gotheader;
!
! if (readBuf == NULL)
! {
! /*
! * First time through, permanently allocate readBuf. We do it this
! * way, rather than just making a static array, for two reasons: (1)
! * no need to waste the storage in most instantiations of the backend;
! * (2) a static char array isn't guaranteed to have any particular
! * alignment, whereas malloc() will provide MAXALIGN'd storage.
! */
! readBuf = (char *) malloc(XLOG_BLCKSZ);
! Assert(readBuf != NULL);
! }
! if (RecPtr == NULL)
! {
! RecPtr = &tmpRecPtr;
!
! /*
! * RecPtr is pointing to end+1 of the previous WAL record. If
! * we're at a page boundary, no more records can fit on the current
! * page. We must skip over the page header, but we can't do that
! * until we've read in the page, since the header size is variable.
! */
! }
! else
! {
! /*
! * In this case, the passed-in record pointer should already be
! * pointing to a valid record starting position.
! */
! if (!XRecOffIsValid(*RecPtr))
! ereport(PANIC,
! (errmsg("invalid record offset at %X/%X",
! (uint32) (*RecPtr >> 32), (uint32) *RecPtr)));
!
! /*
! * Since we are going to a random position in WAL, forget any prior
! * state about what timeline we were in, and allow it to be any
! * timeline in expectedTLIs. We also set a flag to allow curFileTLI
! * to go backwards (but we can't reset that variable right here, since
! * we might not change files at all).
! */
lastPageTLI = 0; /* see comment in ValidXLogPageHeader */
- randAccess = true; /* allow curFileTLI to go backwards too */
- }
! /* This is the first try to read this page. */
! failedSources = 0;
! retry:
! /* Read the page containing the record */
! if (!XLogPageRead(RecPtr, emode, fetching_ckpt, randAccess))
! return NULL;
!
! pageHeaderSize = XLogPageHeaderSize((XLogPageHeader) readBuf);
! targetRecOff = (*RecPtr) % XLOG_BLCKSZ;
! if (targetRecOff == 0)
! {
! /*
! * At page start, so skip over page header. The Assert checks that
! * we're not scribbling on caller's record pointer; it's OK because we
! * can only get here in the continuing-from-prev-record case, since
! * XRecOffIsValid rejected the zero-page-offset case otherwise.
! */
! Assert(RecPtr == &tmpRecPtr);
! (*RecPtr) += pageHeaderSize;
! targetRecOff = pageHeaderSize;
! }
! else if (targetRecOff < pageHeaderSize)
! {
! ereport(emode_for_corrupt_record(emode, *RecPtr),
! (errmsg("invalid record offset at %X/%X",
! (uint32) ((*RecPtr) >> 32), (uint32) *RecPtr)));
! goto next_record_is_invalid;
! }
! if ((((XLogPageHeader) readBuf)->xlp_info & XLP_FIRST_IS_CONTRECORD) &&
! targetRecOff == pageHeaderSize)
! {
! ereport(emode_for_corrupt_record(emode, *RecPtr),
! (errmsg("contrecord is requested by %X/%X",
! (uint32) ((*RecPtr) >> 32), (uint32) *RecPtr)));
! goto next_record_is_invalid;
! }
! /*
! * Read the record length.
! *
! * NB: Even though we use an XLogRecord pointer here, the whole record
! * header might not fit on this page. xl_tot_len is the first field of
! * the struct, so it must be on this page (the records are MAXALIGNed),
! * but we cannot access any other fields until we've verified that we
! * got the whole header.
! */
! record = (XLogRecord *) (readBuf + (*RecPtr) % XLOG_BLCKSZ);
! total_len = record->xl_tot_len;
!
! /*
! * If the whole record header is on this page, validate it immediately.
! * Otherwise do just a basic sanity check on xl_tot_len, and validate the
! * rest of the header after reading it from the next page. The xl_tot_len
! * check is necessary here to ensure that we enter the "Need to reassemble
! * record" code path below; otherwise we might fail to apply
! * ValidXLogRecordHeader at all.
! */
! if (targetRecOff <= XLOG_BLCKSZ - SizeOfXLogRecord)
! {
! if (!ValidXLogRecordHeader(RecPtr, record, emode, randAccess))
! goto next_record_is_invalid;
! gotheader = true;
! }
! else
! {
! if (total_len < SizeOfXLogRecord)
! {
! ereport(emode_for_corrupt_record(emode, *RecPtr),
! (errmsg("invalid record length at %X/%X",
! (uint32) ((*RecPtr) >> 32), (uint32) *RecPtr)));
! goto next_record_is_invalid;
! }
! gotheader = false;
! }
!
! /*
! * Allocate or enlarge readRecordBuf as needed. To avoid useless small
! * increases, round its size to a multiple of XLOG_BLCKSZ, and make sure
! * it's at least 4*Max(BLCKSZ, XLOG_BLCKSZ) to start with. (That is
! * enough for all "normal" records, but very large commit or abort records
! * might need more space.)
! */
! if (total_len > readRecordBufSize)
! {
! uint32 newSize = total_len;
!
! newSize += XLOG_BLCKSZ - (newSize % XLOG_BLCKSZ);
! newSize = Max(newSize, 4 * Max(BLCKSZ, XLOG_BLCKSZ));
! if (readRecordBuf)
! free(readRecordBuf);
! readRecordBuf = (char *) malloc(newSize);
! if (!readRecordBuf)
! {
! readRecordBufSize = 0;
! /* We treat this as a "bogus data" condition */
! ereport(emode_for_corrupt_record(emode, *RecPtr),
! (errmsg("record length %u at %X/%X too long",
! total_len, (uint32) ((*RecPtr) >> 32), (uint32) *RecPtr)));
! goto next_record_is_invalid;
! }
! readRecordBufSize = newSize;
! }
!
! len = XLOG_BLCKSZ - (*RecPtr) % XLOG_BLCKSZ;
! if (total_len > len)
{
! /* Need to reassemble record */
! char *contrecord;
! XLogPageHeader pageHeader;
! XLogRecPtr pagelsn;
! char *buffer;
! uint32 gotlen;
!
! /* Initialize pagelsn to the beginning of the page this record is on */
! pagelsn = ((*RecPtr) / XLOG_BLCKSZ) * XLOG_BLCKSZ;
!
! /* Copy the first fragment of the record from the first page. */
! memcpy(readRecordBuf, readBuf + (*RecPtr) % XLOG_BLCKSZ, len);
! buffer = readRecordBuf + len;
! gotlen = len;
!
! do
{
! /* Calculate pointer to beginning of next page */
! XLByteAdvance(pagelsn, XLOG_BLCKSZ);
! /* Wait for the next page to become available */
! if (!XLogPageRead(&pagelsn, emode, false, false))
! return NULL;
!
! /* Check that the continuation on next page looks valid */
! pageHeader = (XLogPageHeader) readBuf;
! if (!(pageHeader->xlp_info & XLP_FIRST_IS_CONTRECORD))
! {
! ereport(emode_for_corrupt_record(emode, *RecPtr),
! (errmsg("there is no contrecord flag in log segment %s, offset %u",
! XLogFileNameP(curFileTLI, readSegNo),
! readOff)));
! goto next_record_is_invalid;
! }
! /*
! * Cross-check that xlp_rem_len agrees with how much of the record
! * we expect there to be left.
! */
! if (pageHeader->xlp_rem_len == 0 ||
! total_len != (pageHeader->xlp_rem_len + gotlen))
! {
! ereport(emode_for_corrupt_record(emode, *RecPtr),
! (errmsg("invalid contrecord length %u in log segment %s, offset %u",
! pageHeader->xlp_rem_len,
! XLogFileNameP(curFileTLI, readSegNo),
! readOff)));
! goto next_record_is_invalid;
! }
! /* Append the continuation from this page to the buffer */
! pageHeaderSize = XLogPageHeaderSize(pageHeader);
! contrecord = (char *) readBuf + pageHeaderSize;
! len = XLOG_BLCKSZ - pageHeaderSize;
! if (pageHeader->xlp_rem_len < len)
! len = pageHeader->xlp_rem_len;
! memcpy(buffer, (char *) contrecord, len);
! buffer += len;
! gotlen += len;
!
! /* If we just reassembled the record header, validate it. */
! if (!gotheader)
{
! record = (XLogRecord *) readRecordBuf;
! if (!ValidXLogRecordHeader(RecPtr, record, emode, randAccess))
! goto next_record_is_invalid;
! gotheader = true;
}
! } while (pageHeader->xlp_rem_len > len);
!
! record = (XLogRecord *) readRecordBuf;
! if (!RecordIsValid(record, *RecPtr, emode))
! goto next_record_is_invalid;
! pageHeaderSize = XLogPageHeaderSize((XLogPageHeader) readBuf);
! XLogSegNoOffsetToRecPtr(
! readSegNo,
! readOff + pageHeaderSize + MAXALIGN(pageHeader->xlp_rem_len),
! EndRecPtr);
! ReadRecPtr = *RecPtr;
! }
! else
! {
! /* Record does not cross a page boundary */
! if (!RecordIsValid(record, *RecPtr, emode))
! goto next_record_is_invalid;
! EndRecPtr = *RecPtr + MAXALIGN(total_len);
!
! ReadRecPtr = *RecPtr;
! memcpy(readRecordBuf, record, total_len);
! }
!
! /*
! * Special processing if it's an XLOG SWITCH record
! */
! if (record->xl_rmid == RM_XLOG_ID && record->xl_info == XLOG_SWITCH)
! {
! /* Pretend it extends to end of segment */
! EndRecPtr += XLogSegSize - 1;
! EndRecPtr -= EndRecPtr % XLogSegSize;
- /*
- * Pretend that readBuf contains the last page of the segment. This is
- * just to avoid Assert failure in StartupXLOG if XLOG ends with this
- * segment.
- */
- readOff = XLogSegSize - XLOG_BLCKSZ;
- }
return record;
-
- next_record_is_invalid:
- failedSources |= readSource;
-
- if (readFile >= 0)
- {
- close(readFile);
- readFile = -1;
- }
-
- /* In standby-mode, keep trying */
- if (StandbyMode)
- goto retry;
- else
- return NULL;
}
/*
* Check whether the xlog header of a page just read in looks valid.
*
* This is just a convenience subroutine to avoid duplicated code in
! * ReadRecord. It's not intended for use from anywhere else.
*/
static bool
! ValidXLogPageHeader(XLogPageHeader hdr, int emode)
{
XLogRecPtr recaddr;
! XLogSegNoOffsetToRecPtr(readSegNo, readOff, recaddr);
if (hdr->xlp_magic != XLOG_PAGE_MAGIC)
{
! ereport(emode_for_corrupt_record(emode, recaddr),
! (errmsg("invalid magic number %04X in log segment %s, offset %u",
! hdr->xlp_magic,
! XLogFileNameP(curFileTLI, readSegNo),
! readOff)));
return false;
}
if ((hdr->xlp_info & ~XLP_ALL_FLAGS) != 0)
{
! ereport(emode_for_corrupt_record(emode, recaddr),
(errmsg("invalid info bits %04X in log segment %s, offset %u",
hdr->xlp_info,
! XLogFileNameP(curFileTLI, readSegNo),
! readOff)));
return false;
}
if (hdr->xlp_info & XLP_LONG_HEADER)
--- 3198,3268 ----
* the returned record pointer always points there.
*/
static XLogRecord *
! ReadRecord(XLogReaderState *xlogreader, XLogRecPtr RecPtr, int emode,
! bool fetching_ckpt)
{
XLogRecord *record;
! XLogPageReadPrivate *private =
! (XLogPageReadPrivate *) xlogreader->private_data;
! if (!XLogRecPtrIsInvalid(RecPtr))
lastPageTLI = 0; /* see comment in ValidXLogPageHeader */
! /* Set flag for XLogPageRead */
! private->fetching_ckpt = fetching_ckpt;
! /* This is the first try to read this page. */
! private->failedSources = 0;
! do
{
! record = XLogReadRecord(xlogreader, RecPtr, emode);
! ReadRecPtr = xlogreader->ReadRecPtr;
! EndRecPtr = xlogreader->EndRecPtr;
! if (record == NULL)
{
! private->failedSources |= private->readSource;
! if (private->readFile >= 0)
{
! close(private->readFile);
! private->readFile = -1;
}
! }
! } while (StandbyMode && record == NULL);
return record;
}
/*
* Check whether the xlog header of a page just read in looks valid.
*
* This is just a convenience subroutine to avoid duplicated code in
! * XLogPageRead. It's not intended for use from anywhere else.
*/
static bool
! ValidXLogPageHeader(XLogSegNo segno, uint32 offset, int source,
! XLogPageHeader hdr, int emode)
{
XLogRecPtr recaddr;
! XLogSegNoOffsetToRecPtr(segno, offset, recaddr);
if (hdr->xlp_magic != XLOG_PAGE_MAGIC)
{
! ereport(emode_for_corrupt_record(emode, source, recaddr),
! (errmsg("invalid magic number %04X in log segment %s, offset %u",
! hdr->xlp_magic,
! XLogFileNameP(curFileTLI, segno),
! offset)));
return false;
}
if ((hdr->xlp_info & ~XLP_ALL_FLAGS) != 0)
{
! ereport(emode_for_corrupt_record(emode, source, recaddr),
(errmsg("invalid info bits %04X in log segment %s, offset %u",
hdr->xlp_info,
! XLogFileNameP(curFileTLI, segno),
! offset)));
return false;
}
if (hdr->xlp_info & XLP_LONG_HEADER)
***************
*** 3619,3625 **** ValidXLogPageHeader(XLogPageHeader hdr, int emode)
longhdr->xlp_sysid);
snprintf(sysident_str, sizeof(sysident_str), UINT64_FORMAT,
ControlFile->system_identifier);
! ereport(emode_for_corrupt_record(emode, recaddr),
(errmsg("WAL file is from different database system"),
errdetail("WAL file database system identifier is %s, pg_control database system identifier is %s.",
fhdrident_str, sysident_str)));
--- 3282,3288 ----
longhdr->xlp_sysid);
snprintf(sysident_str, sizeof(sysident_str), UINT64_FORMAT,
ControlFile->system_identifier);
! ereport(emode_for_corrupt_record(emode, source, recaddr),
(errmsg("WAL file is from different database system"),
errdetail("WAL file database system identifier is %s, pg_control database system identifier is %s.",
fhdrident_str, sysident_str)));
***************
*** 3627,3663 **** ValidXLogPageHeader(XLogPageHeader hdr, int emode)
}
if (longhdr->xlp_seg_size != XLogSegSize)
{
! ereport(emode_for_corrupt_record(emode, recaddr),
(errmsg("WAL file is from different database system"),
errdetail("Incorrect XLOG_SEG_SIZE in page header.")));
return false;
}
if (longhdr->xlp_xlog_blcksz != XLOG_BLCKSZ)
{
! ereport(emode_for_corrupt_record(emode, recaddr),
(errmsg("WAL file is from different database system"),
errdetail("Incorrect XLOG_BLCKSZ in page header.")));
return false;
}
}
! else if (readOff == 0)
{
/* hmm, first page of file doesn't have a long header? */
! ereport(emode_for_corrupt_record(emode, recaddr),
(errmsg("invalid info bits %04X in log segment %s, offset %u",
hdr->xlp_info,
! XLogFileNameP(curFileTLI, readSegNo),
! readOff)));
return false;
}
if (!XLByteEQ(hdr->xlp_pageaddr, recaddr))
{
! ereport(emode_for_corrupt_record(emode, recaddr),
! (errmsg("unexpected pageaddr %X/%X in log segment %s, offset %u",
! (uint32) (hdr->xlp_pageaddr >> 32), (uint32) hdr->xlp_pageaddr,
! XLogFileNameP(curFileTLI, readSegNo),
! readOff)));
return false;
}
--- 3290,3326 ----
}
if (longhdr->xlp_seg_size != XLogSegSize)
{
! ereport(emode_for_corrupt_record(emode, source, recaddr),
(errmsg("WAL file is from different database system"),
errdetail("Incorrect XLOG_SEG_SIZE in page header.")));
return false;
}
if (longhdr->xlp_xlog_blcksz != XLOG_BLCKSZ)
{
! ereport(emode_for_corrupt_record(emode, source, recaddr),
(errmsg("WAL file is from different database system"),
errdetail("Incorrect XLOG_BLCKSZ in page header.")));
return false;
}
}
! else if (offset == 0)
{
/* hmm, first page of file doesn't have a long header? */
! ereport(emode_for_corrupt_record(emode, source, recaddr),
(errmsg("invalid info bits %04X in log segment %s, offset %u",
hdr->xlp_info,
! XLogFileNameP(curFileTLI, segno),
! offset)));
return false;
}
if (!XLByteEQ(hdr->xlp_pageaddr, recaddr))
{
! ereport(emode_for_corrupt_record(emode, source, recaddr),
! (errmsg("unexpected pageaddr %X/%X in log segment %s, offset %u",
! (uint32) (hdr->xlp_pageaddr >> 32), (uint32) hdr->xlp_pageaddr,
! XLogFileNameP(curFileTLI, segno),
! offset)));
return false;
}
***************
*** 3666,3676 **** ValidXLogPageHeader(XLogPageHeader hdr, int emode)
*/
if (!list_member_int(expectedTLIs, (int) hdr->xlp_tli))
{
! ereport(emode_for_corrupt_record(emode, recaddr),
! (errmsg("unexpected timeline ID %u in log segment %s, offset %u",
! hdr->xlp_tli,
! XLogFileNameP(curFileTLI, readSegNo),
! readOff)));
return false;
}
--- 3329,3339 ----
*/
if (!list_member_int(expectedTLIs, (int) hdr->xlp_tli))
{
! ereport(emode_for_corrupt_record(emode, source, recaddr),
! (errmsg("unexpected timeline ID %u in log segment %s, offset %u",
! hdr->xlp_tli,
! XLogFileNameP(curFileTLI, segno),
! offset)));
return false;
}
***************
*** 3685,3695 **** ValidXLogPageHeader(XLogPageHeader hdr, int emode)
*/
if (hdr->xlp_tli < lastPageTLI)
{
! ereport(emode_for_corrupt_record(emode, recaddr),
(errmsg("out-of-sequence timeline ID %u (after %u) in log segment %s, offset %u",
hdr->xlp_tli, lastPageTLI,
! XLogFileNameP(curFileTLI, readSegNo),
! readOff)));
return false;
}
lastPageTLI = hdr->xlp_tli;
--- 3348,3358 ----
*/
if (hdr->xlp_tli < lastPageTLI)
{
! ereport(emode_for_corrupt_record(emode, source, recaddr),
(errmsg("out-of-sequence timeline ID %u (after %u) in log segment %s, offset %u",
hdr->xlp_tli, lastPageTLI,
! XLogFileNameP(curFileTLI, segno),
! offset)));
return false;
}
lastPageTLI = hdr->xlp_tli;
***************
*** 3697,3784 **** ValidXLogPageHeader(XLogPageHeader hdr, int emode)
}
/*
- * Validate an XLOG record header.
- *
- * This is just a convenience subroutine to avoid duplicated code in
- * ReadRecord. It's not intended for use from anywhere else.
- */
- static bool
- ValidXLogRecordHeader(XLogRecPtr *RecPtr, XLogRecord *record, int emode,
- bool randAccess)
- {
- /*
- * xl_len == 0 is bad data for everything except XLOG SWITCH, where it is
- * required.
- */
- if (record->xl_rmid == RM_XLOG_ID && record->xl_info == XLOG_SWITCH)
- {
- if (record->xl_len != 0)
- {
- ereport(emode_for_corrupt_record(emode, *RecPtr),
- (errmsg("invalid xlog switch record at %X/%X",
- (uint32) ((*RecPtr) >> 32), (uint32) *RecPtr)));
- return false;
- }
- }
- else if (record->xl_len == 0)
- {
- ereport(emode_for_corrupt_record(emode, *RecPtr),
- (errmsg("record with zero length at %X/%X",
- (uint32) ((*RecPtr) >> 32), (uint32) *RecPtr)));
- return false;
- }
- if (record->xl_tot_len < SizeOfXLogRecord + record->xl_len ||
- record->xl_tot_len > SizeOfXLogRecord + record->xl_len +
- XLR_MAX_BKP_BLOCKS * (sizeof(BkpBlock) + BLCKSZ))
- {
- ereport(emode_for_corrupt_record(emode, *RecPtr),
- (errmsg("invalid record length at %X/%X",
- (uint32) ((*RecPtr) >> 32), (uint32) *RecPtr)));
- return false;
- }
- if (record->xl_rmid > RM_MAX_ID)
- {
- ereport(emode_for_corrupt_record(emode, *RecPtr),
- (errmsg("invalid resource manager ID %u at %X/%X",
- record->xl_rmid, (uint32) ((*RecPtr) >> 32), (uint32) *RecPtr)));
- return false;
- }
- if (randAccess)
- {
- /*
- * We can't exactly verify the prev-link, but surely it should be less
- * than the record's own address.
- */
- if (!XLByteLT(record->xl_prev, *RecPtr))
- {
- ereport(emode_for_corrupt_record(emode, *RecPtr),
- (errmsg("record with incorrect prev-link %X/%X at %X/%X",
- (uint32) (record->xl_prev >> 32), (uint32) record->xl_prev,
- (uint32) ((*RecPtr) >> 32), (uint32) *RecPtr)));
- return false;
- }
- }
- else
- {
- /*
- * Record's prev-link should exactly match our previous location. This
- * check guards against torn WAL pages where a stale but valid-looking
- * WAL record starts on a sector boundary.
- */
- if (!XLByteEQ(record->xl_prev, ReadRecPtr))
- {
- ereport(emode_for_corrupt_record(emode, *RecPtr),
- (errmsg("record with incorrect prev-link %X/%X at %X/%X",
- (uint32) (record->xl_prev >> 32), (uint32) record->xl_prev,
- (uint32) ((*RecPtr) >> 32), (uint32) *RecPtr)));
- return false;
- }
- }
-
- return true;
- }
-
- /*
* Scan for new timelines that might have appeared in the archive since we
* started recovery.
*
--- 3360,3365 ----
***************
*** 4739,4745 **** readRecoveryCommandFile(void)
* Exit archive-recovery state
*/
static void
! exitArchiveRecovery(TimeLineID endTLI, XLogSegNo endLogSegNo)
{
char recoveryPath[MAXPGPATH];
char xlogpath[MAXPGPATH];
--- 4320,4327 ----
* Exit archive-recovery state
*/
static void
! exitArchiveRecovery(XLogPageReadPrivate *private, TimeLineID endTLI,
! XLogSegNo endLogSegNo)
{
char recoveryPath[MAXPGPATH];
char xlogpath[MAXPGPATH];
***************
*** 4758,4767 **** exitArchiveRecovery(TimeLineID endTLI, XLogSegNo endLogSegNo)
* If the ending log segment is still open, close it (to avoid problems on
* Windows with trying to rename or delete an open file).
*/
! if (readFile >= 0)
{
! close(readFile);
! readFile = -1;
}
/*
--- 4340,4349 ----
* If the ending log segment is still open, close it (to avoid problems on
* Windows with trying to rename or delete an open file).
*/
! if (private->readFile >= 0)
{
! close(private->readFile);
! private->readFile = -1;
}
/*
***************
*** 5196,5201 **** StartupXLOG(void)
--- 4778,4785 ----
bool backupEndRequired = false;
bool backupFromStandby = false;
DBState dbstate_at_startup;
+ XLogReaderState *xlogreader;
+ XLogPageReadPrivate *private;
/*
* Read control file and check XLOG status looks valid.
***************
*** 5329,5334 **** StartupXLOG(void)
--- 4913,4923 ----
if (StandbyMode)
OwnLatch(&XLogCtl->recoveryWakeupLatch);
+ private = malloc(sizeof(XLogPageReadPrivate));
+ MemSet(private, 0, sizeof(XLogPageReadPrivate));
+ private->readFile = -1;
+ xlogreader = XLogReaderAllocate(InvalidXLogRecPtr, &XLogPageRead, private);
+
if (read_backup_label(&checkPointLoc, &backupEndRequired,
&backupFromStandby))
{
***************
*** 5336,5349 **** StartupXLOG(void)
* When a backup_label file is present, we want to roll forward from
* the checkpoint it identifies, rather than using pg_control.
*/
! record = ReadCheckpointRecord(checkPointLoc, 0);
if (record != NULL)
{
memcpy(&checkPoint, XLogRecGetData(record), sizeof(CheckPoint));
wasShutdown = (record->xl_info == XLOG_CHECKPOINT_SHUTDOWN);
ereport(DEBUG1,
(errmsg("checkpoint record is at %X/%X",
! (uint32) (checkPointLoc >> 32), (uint32) checkPointLoc)));
InRecovery = true; /* force recovery even if SHUTDOWNED */
/*
--- 4925,4938 ----
* When a backup_label file is present, we want to roll forward from
* the checkpoint it identifies, rather than using pg_control.
*/
! record = ReadCheckpointRecord(xlogreader, checkPointLoc, 0);
if (record != NULL)
{
memcpy(&checkPoint, XLogRecGetData(record), sizeof(CheckPoint));
wasShutdown = (record->xl_info == XLOG_CHECKPOINT_SHUTDOWN);
ereport(DEBUG1,
(errmsg("checkpoint record is at %X/%X",
! (uint32) (checkPointLoc >> 32), (uint32) checkPointLoc)));
InRecovery = true; /* force recovery even if SHUTDOWNED */
/*
***************
*** 5354,5360 **** StartupXLOG(void)
*/
if (XLByteLT(checkPoint.redo, checkPointLoc))
{
! if (!ReadRecord(&(checkPoint.redo), LOG, false))
ereport(FATAL,
(errmsg("could not find redo location referenced by checkpoint record"),
errhint("If you are not restoring from a backup, try removing the file \"%s/backup_label\".", DataDir)));
--- 4943,4949 ----
*/
if (XLByteLT(checkPoint.redo, checkPointLoc))
{
! if (!ReadRecord(xlogreader, checkPoint.redo, LOG, false))
ereport(FATAL,
(errmsg("could not find redo location referenced by checkpoint record"),
errhint("If you are not restoring from a backup, try removing the file \"%s/backup_label\".", DataDir)));
***************
*** 5378,5389 **** StartupXLOG(void)
*/
checkPointLoc = ControlFile->checkPoint;
RedoStartLSN = ControlFile->checkPointCopy.redo;
! record = ReadCheckpointRecord(checkPointLoc, 1);
if (record != NULL)
{
ereport(DEBUG1,
(errmsg("checkpoint record is at %X/%X",
! (uint32) (checkPointLoc >> 32), (uint32) checkPointLoc)));
}
else if (StandbyMode)
{
--- 4967,4978 ----
*/
checkPointLoc = ControlFile->checkPoint;
RedoStartLSN = ControlFile->checkPointCopy.redo;
! record = ReadCheckpointRecord(xlogreader, checkPointLoc, 1);
if (record != NULL)
{
ereport(DEBUG1,
(errmsg("checkpoint record is at %X/%X",
! (uint32) (checkPointLoc >> 32), (uint32) checkPointLoc)));
}
else if (StandbyMode)
{
***************
*** 5397,5408 **** StartupXLOG(void)
else
{
checkPointLoc = ControlFile->prevCheckPoint;
! record = ReadCheckpointRecord(checkPointLoc, 2);
if (record != NULL)
{
ereport(LOG,
(errmsg("using previous checkpoint record at %X/%X",
! (uint32) (checkPointLoc >> 32), (uint32) checkPointLoc)));
InRecovery = true; /* force recovery even if SHUTDOWNED */
}
else
--- 4986,4997 ----
else
{
checkPointLoc = ControlFile->prevCheckPoint;
! record = ReadCheckpointRecord(xlogreader, checkPointLoc, 2);
if (record != NULL)
{
ereport(LOG,
(errmsg("using previous checkpoint record at %X/%X",
! (uint32) (checkPointLoc >> 32), (uint32) checkPointLoc)));
InRecovery = true; /* force recovery even if SHUTDOWNED */
}
else
***************
*** 5417,5423 **** StartupXLOG(void)
ereport(DEBUG1,
(errmsg("redo record is at %X/%X; shutdown %s",
! (uint32) (checkPoint.redo >> 32), (uint32) checkPoint.redo,
wasShutdown ? "TRUE" : "FALSE")));
ereport(DEBUG1,
(errmsg("next transaction ID: %u/%u; next OID: %u",
--- 5006,5012 ----
ereport(DEBUG1,
(errmsg("redo record is at %X/%X; shutdown %s",
! (uint32) (checkPoint.redo >> 32), (uint32) checkPoint.redo,
wasShutdown ? "TRUE" : "FALSE")));
ereport(DEBUG1,
(errmsg("next transaction ID: %u/%u; next OID: %u",
***************
*** 5698,5704 **** StartupXLOG(void)
* Allow read-only connections immediately if we're consistent
* already.
*/
! CheckRecoveryConsistency();
/*
* Find the first record that logically follows the checkpoint --- it
--- 5287,5293 ----
* Allow read-only connections immediately if we're consistent
* already.
*/
! CheckRecoveryConsistency(EndRecPtr);
/*
* Find the first record that logically follows the checkpoint --- it
***************
*** 5707,5718 **** StartupXLOG(void)
if (XLByteLT(checkPoint.redo, RecPtr))
{
/* back up to find the record */
! record = ReadRecord(&(checkPoint.redo), PANIC, false);
}
else
{
/* just have to read next record after CheckPoint */
! record = ReadRecord(NULL, LOG, false);
}
if (record != NULL)
--- 5296,5307 ----
if (XLByteLT(checkPoint.redo, RecPtr))
{
/* back up to find the record */
! record = ReadRecord(xlogreader, checkPoint.redo, PANIC, false);
}
else
{
/* just have to read next record after CheckPoint */
! record = ReadRecord(xlogreader, InvalidXLogRecPtr, LOG, false);
}
if (record != NULL)
***************
*** 5727,5733 **** StartupXLOG(void)
ereport(LOG,
(errmsg("redo starts at %X/%X",
! (uint32) (ReadRecPtr >> 32), (uint32) ReadRecPtr)));
/*
* main redo apply loop
--- 5316,5322 ----
ereport(LOG,
(errmsg("redo starts at %X/%X",
! (uint32) (ReadRecPtr >> 32), (uint32) ReadRecPtr)));
/*
* main redo apply loop
***************
*** 5743,5750 **** StartupXLOG(void)
initStringInfo(&buf);
appendStringInfo(&buf, "REDO @ %X/%X; LSN %X/%X: ",
! (uint32) (ReadRecPtr >> 32), (uint32) ReadRecPtr,
! (uint32) (EndRecPtr >> 32), (uint32) EndRecPtr);
xlog_outrec(&buf, record);
appendStringInfo(&buf, " - ");
RmgrTable[record->xl_rmid].rm_desc(&buf,
--- 5332,5339 ----
initStringInfo(&buf);
appendStringInfo(&buf, "REDO @ %X/%X; LSN %X/%X: ",
! (uint32) (ReadRecPtr >> 32), (uint32) ReadRecPtr,
! (uint32) (EndRecPtr >> 32), (uint32) EndRecPtr);
xlog_outrec(&buf, record);
appendStringInfo(&buf, " - ");
RmgrTable[record->xl_rmid].rm_desc(&buf,
***************
*** 5759,5765 **** StartupXLOG(void)
HandleStartupProcInterrupts();
/* Allow read-only connections if we're consistent now */
! CheckRecoveryConsistency();
/*
* Have we reached our recovery target?
--- 5348,5354 ----
HandleStartupProcInterrupts();
/* Allow read-only connections if we're consistent now */
! CheckRecoveryConsistency(EndRecPtr);
/*
* Have we reached our recovery target?
***************
*** 5863,5869 **** StartupXLOG(void)
LastRec = ReadRecPtr;
! record = ReadRecord(NULL, LOG, false);
} while (record != NULL && recoveryContinue);
/*
--- 5452,5458 ----
LastRec = ReadRecPtr;
! record = ReadRecord(xlogreader, InvalidXLogRecPtr, LOG, false);
} while (record != NULL && recoveryContinue);
/*
***************
*** 5872,5878 **** StartupXLOG(void)
ereport(LOG,
(errmsg("redo done at %X/%X",
! (uint32) (ReadRecPtr >> 32), (uint32) ReadRecPtr)));
xtime = GetLatestXTime();
if (xtime)
ereport(LOG,
--- 5461,5467 ----
ereport(LOG,
(errmsg("redo done at %X/%X",
! (uint32) (ReadRecPtr >> 32), (uint32) ReadRecPtr)));
xtime = GetLatestXTime();
if (xtime)
ereport(LOG,
***************
*** 5913,5919 **** StartupXLOG(void)
* Re-fetch the last valid or last applied record, so we can identify the
* exact endpoint of what we consider the valid portion of WAL.
*/
! record = ReadRecord(&LastRec, PANIC, false);
EndOfLog = EndRecPtr;
XLByteToPrevSeg(EndOfLog, endLogSegNo);
--- 5502,5508 ----
* Re-fetch the last valid or last applied record, so we can identify the
* exact endpoint of what we consider the valid portion of WAL.
*/
! record = ReadRecord(xlogreader, LastRec, PANIC, false);
EndOfLog = EndRecPtr;
XLByteToPrevSeg(EndOfLog, endLogSegNo);
***************
*** 5976,5982 **** StartupXLOG(void)
*/
if (InArchiveRecovery)
{
! char reason[200];
ThisTimeLineID = findNewestTimeLine(recoveryTargetTLI) + 1;
ereport(LOG,
--- 5565,5571 ----
*/
if (InArchiveRecovery)
{
! char reason[200];
ThisTimeLineID = findNewestTimeLine(recoveryTargetTLI) + 1;
ereport(LOG,
***************
*** 6017,6023 **** StartupXLOG(void)
* we will use that below.)
*/
if (InArchiveRecovery)
! exitArchiveRecovery(curFileTLI, endLogSegNo);
/*
* Prepare to write WAL starting at EndOfLog position, and init xlog
--- 5606,5612 ----
* we will use that below.)
*/
if (InArchiveRecovery)
! exitArchiveRecovery(private, curFileTLI, endLogSegNo);
/*
* Prepare to write WAL starting at EndOfLog position, and init xlog
***************
*** 6036,6043 **** StartupXLOG(void)
* record spans, not the one it starts in. The last block is indeed the
* one we want to use.
*/
! Assert(readOff == (XLogCtl->xlblocks[0] - XLOG_BLCKSZ) % XLogSegSize);
! memcpy((char *) Insert->currpage, readBuf, XLOG_BLCKSZ);
Insert->currpos = (char *) Insert->currpage +
(EndOfLog + XLOG_BLCKSZ - XLogCtl->xlblocks[0]);
--- 5625,5639 ----
* record spans, not the one it starts in. The last block is indeed the
* one we want to use.
*/
! if (EndOfLog % XLOG_BLCKSZ == 0)
! {
! memset(Insert->currpage, 0, XLOG_BLCKSZ);
! }
! else
! {
! Assert(private->readOff == (XLogCtl->xlblocks[0] - XLOG_BLCKSZ) % XLogSegSize);
! memcpy((char *) Insert->currpage, xlogreader->readBuf, XLOG_BLCKSZ);
! }
Insert->currpos = (char *) Insert->currpage +
(EndOfLog + XLOG_BLCKSZ - XLogCtl->xlblocks[0]);
***************
*** 6189,6210 **** StartupXLOG(void)
ShutdownRecoveryTransactionEnvironment();
/* Shut down readFile facility, free space */
! if (readFile >= 0)
! {
! close(readFile);
! readFile = -1;
! }
! if (readBuf)
{
! free(readBuf);
! readBuf = NULL;
! }
! if (readRecordBuf)
! {
! free(readRecordBuf);
! readRecordBuf = NULL;
! readRecordBufSize = 0;
}
/*
* If any of the critical GUCs have changed, log them before we allow
--- 5785,5799 ----
ShutdownRecoveryTransactionEnvironment();
/* Shut down readFile facility, free space */
! private = (XLogPageReadPrivate *) xlogreader->private_data;
! if (private->readFile >= 0)
{
! close(private->readFile);
! private->readFile = -1;
}
+ if (xlogreader->private_data)
+ free(xlogreader->private_data);
+ XLogReaderFree(xlogreader);
/*
* If any of the critical GUCs have changed, log them before we allow
***************
*** 6235,6241 **** StartupXLOG(void)
* that it can start accepting read-only connections.
*/
static void
! CheckRecoveryConsistency(void)
{
/*
* During crash recovery, we don't reach a consistent state until we've
--- 5824,5830 ----
* that it can start accepting read-only connections.
*/
static void
! CheckRecoveryConsistency(XLogRecPtr EndRecPtr)
{
/*
* During crash recovery, we don't reach a consistent state until we've
***************
*** 6415,6421 **** LocalSetXLogInsertAllowed(void)
* 1 for "primary", 2 for "secondary", 0 for "other" (backup_label)
*/
static XLogRecord *
! ReadCheckpointRecord(XLogRecPtr RecPtr, int whichChkpt)
{
XLogRecord *record;
--- 6004,6010 ----
* 1 for "primary", 2 for "secondary", 0 for "other" (backup_label)
*/
static XLogRecord *
! ReadCheckpointRecord(XLogReaderState *xlogreader, XLogRecPtr RecPtr, int whichChkpt)
{
XLogRecord *record;
***************
*** 6439,6445 **** ReadCheckpointRecord(XLogRecPtr RecPtr, int whichChkpt)
return NULL;
}
! record = ReadRecord(&RecPtr, LOG, true);
if (record == NULL)
{
--- 6028,6034 ----
return NULL;
}
! record = ReadRecord(xlogreader, RecPtr, LOG, true);
if (record == NULL)
{
***************
*** 6667,6673 **** GetRecoveryTargetTLI(void)
{
/* use volatile pointer to prevent code rearrangement */
volatile XLogCtlData *xlogctl = XLogCtl;
! TimeLineID result;
SpinLockAcquire(&xlogctl->info_lck);
result = xlogctl->RecoveryTargetTLI;
--- 6256,6262 ----
{
/* use volatile pointer to prevent code rearrangement */
volatile XLogCtlData *xlogctl = XLogCtl;
! TimeLineID result;
SpinLockAcquire(&xlogctl->info_lck);
result = xlogctl->RecoveryTargetTLI;
***************
*** 6952,6958 **** CreateCheckPoint(int flags)
XLogRecPtr curInsert;
INSERT_RECPTR(curInsert, Insert, Insert->curridx);
! if (curInsert == ControlFile->checkPoint +
MAXALIGN(SizeOfXLogRecord + sizeof(CheckPoint)) &&
ControlFile->checkPoint == ControlFile->checkPointCopy.redo)
{
--- 6541,6547 ----
XLogRecPtr curInsert;
INSERT_RECPTR(curInsert, Insert, Insert->curridx);
! if (curInsert == ControlFile->checkPoint +
MAXALIGN(SizeOfXLogRecord + sizeof(CheckPoint)) &&
ControlFile->checkPoint == ControlFile->checkPointCopy.redo)
{
***************
*** 7382,7388 **** CreateRestartPoint(int flags)
{
ereport(DEBUG2,
(errmsg("skipping restartpoint, already performed at %X/%X",
! (uint32) (lastCheckPoint.redo >> 32), (uint32) lastCheckPoint.redo)));
UpdateMinRecoveryPoint(InvalidXLogRecPtr, true);
if (flags & CHECKPOINT_IS_SHUTDOWN)
--- 6971,6977 ----
{
ereport(DEBUG2,
(errmsg("skipping restartpoint, already performed at %X/%X",
! (uint32) (lastCheckPoint.redo >> 32), (uint32) lastCheckPoint.redo)));
UpdateMinRecoveryPoint(InvalidXLogRecPtr, true);
if (flags & CHECKPOINT_IS_SHUTDOWN)
***************
*** 7492,7498 **** CreateRestartPoint(int flags)
xtime = GetLatestXTime();
ereport((log_checkpoints ? LOG : DEBUG2),
(errmsg("recovery restart point at %X/%X",
! (uint32) (lastCheckPoint.redo >> 32), (uint32) lastCheckPoint.redo),
xtime ? errdetail("last completed transaction was at log time %s",
timestamptz_to_str(xtime)) : 0));
--- 7081,7087 ----
xtime = GetLatestXTime();
ereport((log_checkpoints ? LOG : DEBUG2),
(errmsg("recovery restart point at %X/%X",
! (uint32) (lastCheckPoint.redo >> 32), (uint32) lastCheckPoint.redo),
xtime ? errdetail("last completed transaction was at log time %s",
timestamptz_to_str(xtime)) : 0));
***************
*** 8017,8023 **** xlog_desc(StringInfo buf, uint8 xl_info, char *rec)
appendStringInfo(buf, "checkpoint: redo %X/%X; "
"tli %u; fpw %s; xid %u/%u; oid %u; multi %u; offset %u; "
"oldest xid %u in DB %u; oldest running xid %u; %s",
! (uint32) (checkpoint->redo >> 32), (uint32) checkpoint->redo,
checkpoint->ThisTimeLineID,
checkpoint->fullPageWrites ? "true" : "false",
checkpoint->nextXidEpoch, checkpoint->nextXid,
--- 7606,7612 ----
appendStringInfo(buf, "checkpoint: redo %X/%X; "
"tli %u; fpw %s; xid %u/%u; oid %u; multi %u; offset %u; "
"oldest xid %u in DB %u; oldest running xid %u; %s",
! (uint32) (checkpoint->redo >> 32), (uint32) checkpoint->redo,
checkpoint->ThisTimeLineID,
checkpoint->fullPageWrites ? "true" : "false",
checkpoint->nextXidEpoch, checkpoint->nextXid,
***************
*** 8198,8204 **** assign_xlog_sync_method(int new_sync_method, void *extra)
ereport(PANIC,
(errcode_for_file_access(),
errmsg("could not fsync log segment %s: %m",
! XLogFileNameP(ThisTimeLineID, openLogSegNo))));
if (get_sync_bit(sync_method) != get_sync_bit(new_sync_method))
XLogFileClose();
}
--- 7787,7793 ----
ereport(PANIC,
(errcode_for_file_access(),
errmsg("could not fsync log segment %s: %m",
! XLogFileNameP(ThisTimeLineID, openLogSegNo))));
if (get_sync_bit(sync_method) != get_sync_bit(new_sync_method))
XLogFileClose();
}
***************
*** 8229,8236 **** issue_xlog_fsync(int fd, XLogSegNo segno)
if (pg_fsync_writethrough(fd) != 0)
ereport(PANIC,
(errcode_for_file_access(),
! errmsg("could not fsync write-through log file %s: %m",
! XLogFileNameP(ThisTimeLineID, segno))));
break;
#endif
#ifdef HAVE_FDATASYNC
--- 7818,7825 ----
if (pg_fsync_writethrough(fd) != 0)
ereport(PANIC,
(errcode_for_file_access(),
! errmsg("could not fsync write-through log file %s: %m",
! XLogFileNameP(ThisTimeLineID, segno))));
break;
#endif
#ifdef HAVE_FDATASYNC
***************
*** 8259,8264 **** char *
--- 7848,7854 ----
XLogFileNameP(TimeLineID tli, XLogSegNo segno)
{
char *result = palloc(MAXFNAMELEN);
+
XLogFileName(result, tli, segno);
return result;
}
***************
*** 8504,8512 **** do_pg_start_backup(const char *backupidstr, bool fast, char **labelfile)
"%Y-%m-%d %H:%M:%S %Z",
pg_localtime(&stamp_time, log_timezone));
appendStringInfo(&labelfbuf, "START WAL LOCATION: %X/%X (file %s)\n",
! (uint32) (startpoint >> 32), (uint32) startpoint, xlogfilename);
appendStringInfo(&labelfbuf, "CHECKPOINT LOCATION: %X/%X\n",
! (uint32) (checkpointloc >> 32), (uint32) checkpointloc);
appendStringInfo(&labelfbuf, "BACKUP METHOD: %s\n",
exclusive ? "pg_start_backup" : "streamed");
appendStringInfo(&labelfbuf, "BACKUP FROM: %s\n",
--- 8094,8102 ----
"%Y-%m-%d %H:%M:%S %Z",
pg_localtime(&stamp_time, log_timezone));
appendStringInfo(&labelfbuf, "START WAL LOCATION: %X/%X (file %s)\n",
! (uint32) (startpoint >> 32), (uint32) startpoint, xlogfilename);
appendStringInfo(&labelfbuf, "CHECKPOINT LOCATION: %X/%X\n",
! (uint32) (checkpointloc >> 32), (uint32) checkpointloc);
appendStringInfo(&labelfbuf, "BACKUP METHOD: %s\n",
exclusive ? "pg_start_backup" : "streamed");
appendStringInfo(&labelfbuf, "BACKUP FROM: %s\n",
***************
*** 8854,8860 **** do_pg_stop_backup(char *labelfile, bool waitforarchive)
errmsg("could not create file \"%s\": %m",
histfilepath)));
fprintf(fp, "START WAL LOCATION: %X/%X (file %s)\n",
! (uint32) (startpoint >> 32), (uint32) startpoint, startxlogfilename);
fprintf(fp, "STOP WAL LOCATION: %X/%X (file %s)\n",
(uint32) (stoppoint >> 32), (uint32) stoppoint, stopxlogfilename);
/* transfer remaining lines from label to history file */
--- 8444,8450 ----
errmsg("could not create file \"%s\": %m",
histfilepath)));
fprintf(fp, "START WAL LOCATION: %X/%X (file %s)\n",
! (uint32) (startpoint >> 32), (uint32) startpoint, startxlogfilename);
fprintf(fp, "STOP WAL LOCATION: %X/%X (file %s)\n",
(uint32) (stoppoint >> 32), (uint32) stoppoint, stopxlogfilename);
/* transfer remaining lines from label to history file */
***************
*** 9262,9288 **** CancelBackup(void)
* sleep and retry.
*/
static bool
! XLogPageRead(XLogRecPtr *RecPtr, int emode, bool fetching_ckpt,
! bool randAccess)
{
uint32 targetPageOff;
uint32 targetRecOff;
XLogSegNo targetSegNo;
! XLByteToSeg(*RecPtr, targetSegNo);
! targetPageOff = (((*RecPtr) % XLogSegSize) / XLOG_BLCKSZ) * XLOG_BLCKSZ;
! targetRecOff = (*RecPtr) % XLOG_BLCKSZ;
/* Fast exit if we have read the record in the current buffer already */
! if (failedSources == 0 && targetSegNo == readSegNo &&
! targetPageOff == readOff && targetRecOff < readLen)
return true;
/*
* See if we need to switch to a new segment because the requested record
* is not in the currently open one.
*/
! if (readFile >= 0 && !XLByteInSeg(*RecPtr, readSegNo))
{
/*
* Request a restartpoint if we've replayed too much xlog since the
--- 8852,8879 ----
* sleep and retry.
*/
static bool
! XLogPageRead(XLogReaderState *xlogreader, XLogRecPtr RecPtr, int emode,
! bool randAccess, char *readBuf, void *private_data)
{
+ XLogPageReadPrivate *private = (XLogPageReadPrivate *) private_data;
uint32 targetPageOff;
uint32 targetRecOff;
XLogSegNo targetSegNo;
! XLByteToSeg(RecPtr, targetSegNo);
! targetPageOff = ((RecPtr % XLogSegSize) / XLOG_BLCKSZ) * XLOG_BLCKSZ;
! targetRecOff = RecPtr % XLOG_BLCKSZ;
/* Fast exit if we have read the record in the current buffer already */
! if (private->failedSources == 0 && targetSegNo == private->readSegNo &&
! targetPageOff == private->readOff && targetRecOff < private->readLen)
return true;
/*
* See if we need to switch to a new segment because the requested record
* is not in the currently open one.
*/
! if (private->readFile >= 0 && !XLByteInSeg(RecPtr, private->readSegNo))
{
/*
* Request a restartpoint if we've replayed too much xlog since the
***************
*** 9290,9325 **** XLogPageRead(XLogRecPtr *RecPtr, int emode, bool fetching_ckpt,
*/
if (StandbyMode && bgwriterLaunched)
{
! if (XLogCheckpointNeeded(readSegNo))
{
(void) GetRedoRecPtr();
! if (XLogCheckpointNeeded(readSegNo))
RequestCheckpoint(CHECKPOINT_CAUSE_XLOG);
}
}
! close(readFile);
! readFile = -1;
! readSource = 0;
}
! XLByteToSeg(*RecPtr, readSegNo);
retry:
/* See if we need to retrieve more data */
! if (readFile < 0 ||
! (readSource == XLOG_FROM_STREAM && !XLByteLT(*RecPtr, receivedUpto)))
{
if (StandbyMode)
{
! if (!WaitForWALToBecomeAvailable(*RecPtr, randAccess,
! fetching_ckpt))
goto triggered;
}
else
{
/* In archive or crash recovery. */
! if (readFile < 0)
{
int sources;
--- 8881,8916 ----
*/
if (StandbyMode && bgwriterLaunched)
{
! if (XLogCheckpointNeeded(private->readSegNo))
{
(void) GetRedoRecPtr();
! if (XLogCheckpointNeeded(private->readSegNo))
RequestCheckpoint(CHECKPOINT_CAUSE_XLOG);
}
}
! close(private->readFile);
! private->readFile = -1;
! private->readSource = 0;
}
! XLByteToSeg(RecPtr, private->readSegNo);
retry:
/* See if we need to retrieve more data */
! if (private->readFile < 0 ||
! (private->readSource == XLOG_FROM_STREAM &&
! !XLByteLT(RecPtr, receivedUpto)))
{
if (StandbyMode)
{
! if (!WaitForWALToBecomeAvailable(private, RecPtr, randAccess))
goto triggered;
}
else
{
/* In archive or crash recovery. */
! if (private->readFile < 0)
{
int sources;
***************
*** 9331,9338 **** retry:
if (InArchiveRecovery)
sources |= XLOG_FROM_ARCHIVE;
! readFile = XLogFileReadAnyTLI(readSegNo, emode, sources);
! if (readFile < 0)
return false;
}
}
--- 8922,8931 ----
if (InArchiveRecovery)
sources |= XLOG_FROM_ARCHIVE;
! private->readFile =
! XLogFileReadAnyTLI(private, private->readSegNo, emode,
! sources);
! if (private->readFile < 0)
return false;
}
}
***************
*** 9342,9348 **** retry:
* At this point, we have the right segment open and if we're streaming we
* know the requested record is in it.
*/
! Assert(readFile != -1);
/*
* If the current segment is being streamed from master, calculate how
--- 8935,8941 ----
* At this point, we have the right segment open and if we're streaming we
* know the requested record is in it.
*/
! Assert(private->readFile != -1);
/*
* If the current segment is being streamed from master, calculate how
***************
*** 9350,9368 **** retry:
* requested record has been received, but this is for the benefit of
* future calls, to allow quick exit at the top of this function.
*/
! if (readSource == XLOG_FROM_STREAM)
{
! if (((*RecPtr) / XLOG_BLCKSZ) != (receivedUpto / XLOG_BLCKSZ))
{
! readLen = XLOG_BLCKSZ;
}
else
! readLen = receivedUpto % XLogSegSize - targetPageOff;
}
else
! readLen = XLOG_BLCKSZ;
! if (!readFileHeaderValidated && targetPageOff != 0)
{
/*
* Whenever switching to a new WAL segment, we read the first page of
--- 8943,8961 ----
* requested record has been received, but this is for the benefit of
* future calls, to allow quick exit at the top of this function.
*/
! if (private->readSource == XLOG_FROM_STREAM)
{
! if (((RecPtr) / XLOG_BLCKSZ) != (receivedUpto / XLOG_BLCKSZ))
{
! private->readLen = XLOG_BLCKSZ;
}
else
! private->readLen = receivedUpto % XLogSegSize - targetPageOff;
}
else
! private->readLen = XLOG_BLCKSZ;
! if (!private->readFileHeaderValidated && targetPageOff != 0)
{
/*
* Whenever switching to a new WAL segment, we read the first page of
***************
*** 9371,9432 **** retry:
* identification info that is present in the first page's "long"
* header.
*/
! readOff = 0;
! if (read(readFile, readBuf, XLOG_BLCKSZ) != XLOG_BLCKSZ)
{
! char fname[MAXFNAMELEN];
! XLogFileName(fname, curFileTLI, readSegNo);
! ereport(emode_for_corrupt_record(emode, *RecPtr),
(errcode_for_file_access(),
! errmsg("could not read from log segment %s, offset %u: %m",
! fname, readOff)));
goto next_record_is_invalid;
}
! if (!ValidXLogPageHeader((XLogPageHeader) readBuf, emode))
goto next_record_is_invalid;
}
/* Read the requested page */
! readOff = targetPageOff;
! if (lseek(readFile, (off_t) readOff, SEEK_SET) < 0)
{
! char fname[MAXFNAMELEN];
! XLogFileName(fname, curFileTLI, readSegNo);
! ereport(emode_for_corrupt_record(emode, *RecPtr),
(errcode_for_file_access(),
! errmsg("could not seek in log segment %s to offset %u: %m",
! fname, readOff)));
goto next_record_is_invalid;
}
! if (read(readFile, readBuf, XLOG_BLCKSZ) != XLOG_BLCKSZ)
{
! char fname[MAXFNAMELEN];
! XLogFileName(fname, curFileTLI, readSegNo);
! ereport(emode_for_corrupt_record(emode, *RecPtr),
(errcode_for_file_access(),
! errmsg("could not read from log segment %s, offset %u: %m",
! fname, readOff)));
goto next_record_is_invalid;
}
! if (!ValidXLogPageHeader((XLogPageHeader) readBuf, emode))
goto next_record_is_invalid;
! readFileHeaderValidated = true;
! Assert(targetSegNo == readSegNo);
! Assert(targetPageOff == readOff);
! Assert(targetRecOff < readLen);
return true;
next_record_is_invalid:
! failedSources |= readSource;
! if (readFile >= 0)
! close(readFile);
! readFile = -1;
! readLen = 0;
! readSource = 0;
/* In standby-mode, keep trying */
if (StandbyMode)
--- 8964,9032 ----
* identification info that is present in the first page's "long"
* header.
*/
! private->readOff = 0;
! if (read(private->readFile, readBuf, XLOG_BLCKSZ) != XLOG_BLCKSZ)
{
! char fname[MAXFNAMELEN];
!
! XLogFileName(fname, curFileTLI, private->readSegNo);
! ereport(emode_for_corrupt_record(emode, private->readSource, RecPtr),
(errcode_for_file_access(),
! errmsg("could not read from log segment %s, offset %u: %m",
! fname, private->readOff)));
goto next_record_is_invalid;
}
! if (!ValidXLogPageHeader(private->readSegNo, private->readOff,
! private->readSource, (XLogPageHeader) readBuf,
! emode))
goto next_record_is_invalid;
}
/* Read the requested page */
! private->readOff = targetPageOff;
! if (lseek(private->readFile, (off_t) private->readOff, SEEK_SET) < 0)
{
! char fname[MAXFNAMELEN];
!
! XLogFileName(fname, curFileTLI, private->readSegNo);
! ereport(emode_for_corrupt_record(emode, private->readSource, RecPtr),
(errcode_for_file_access(),
! errmsg("could not seek in log segment %s to offset %u: %m",
! fname, private->readOff)));
goto next_record_is_invalid;
}
! if (read(private->readFile, readBuf, XLOG_BLCKSZ) != XLOG_BLCKSZ)
{
! char fname[MAXFNAMELEN];
!
! XLogFileName(fname, curFileTLI, private->readSegNo);
! ereport(emode_for_corrupt_record(emode, private->readSource, RecPtr),
(errcode_for_file_access(),
! errmsg("could not read from log segment %s, offset %u: %m",
! fname, private->readOff)));
goto next_record_is_invalid;
}
! if (!ValidXLogPageHeader(private->readSegNo, private->readOff,
! private->readSource, (XLogPageHeader) readBuf,
! emode))
goto next_record_is_invalid;
! private->readFileHeaderValidated = true;
! Assert(targetSegNo == private->readSegNo);
! Assert(targetPageOff == private->readOff);
! Assert(targetRecOff < private->readLen);
return true;
next_record_is_invalid:
! private->failedSources |= private->readSource;
! if (private->readFile >= 0)
! close(private->readFile);
! private->readFile = -1;
! private->readLen = 0;
! private->readSource = 0;
/* In standby-mode, keep trying */
if (StandbyMode)
***************
*** 9435,9445 **** next_record_is_invalid:
return false;
triggered:
! if (readFile >= 0)
! close(readFile);
! readFile = -1;
! readLen = 0;
! readSource = 0;
return false;
}
--- 9035,9045 ----
return false;
triggered:
! if (private->readFile >= 0)
! close(private->readFile);
! private->readFile = -1;
! private->readLen = 0;
! private->readSource = 0;
return false;
}
***************
*** 9456,9463 **** triggered:
* false.
*/
static bool
! WaitForWALToBecomeAvailable(XLogRecPtr RecPtr, bool randAccess,
! bool fetching_ckpt)
{
static pg_time_t last_fail_time = 0;
--- 9056,9063 ----
* false.
*/
static bool
! WaitForWALToBecomeAvailable(XLogPageReadPrivate *private, XLogRecPtr RecPtr,
! bool randAccess)
{
static pg_time_t last_fail_time = 0;
***************
*** 9476,9482 **** WaitForWALToBecomeAvailable(XLogRecPtr RecPtr, bool randAccess,
* the archive should be identical to what was streamed, so it's
* unlikely that it helps, but one can hope...
*/
! if (failedSources & XLOG_FROM_STREAM)
{
ShutdownWalRcv();
continue;
--- 9076,9082 ----
* the archive should be identical to what was streamed, so it's
* unlikely that it helps, but one can hope...
*/
! if (private->failedSources & XLOG_FROM_STREAM)
{
ShutdownWalRcv();
continue;
***************
*** 9515,9535 **** WaitForWALToBecomeAvailable(XLogRecPtr RecPtr, bool randAccess,
if (havedata)
{
/*
! * Great, streamed far enough. Open the file if it's not open
* already. Use XLOG_FROM_STREAM so that source info is set
* correctly and XLogReceiptTime isn't changed.
*/
! if (readFile < 0)
{
! readFile = XLogFileRead(readSegNo, PANIC,
! recoveryTargetTLI,
! XLOG_FROM_STREAM, false);
! Assert(readFile >= 0);
}
else
{
/* just make sure source info is correct... */
! readSource = XLOG_FROM_STREAM;
XLogReceiptSource = XLOG_FROM_STREAM;
}
break;
--- 9115,9136 ----
if (havedata)
{
/*
! * Great, streamed far enough. Open the file if it's not open
* already. Use XLOG_FROM_STREAM so that source info is set
* correctly and XLogReceiptTime isn't changed.
*/
! if (private->readFile < 0)
{
! private->readFile =
! XLogFileRead(private, private->readSegNo, PANIC,
! recoveryTargetTLI,
! XLOG_FROM_STREAM, false);
! Assert(private->readFile >= 0);
}
else
{
/* just make sure source info is correct... */
! private->readSource = XLOG_FROM_STREAM;
XLogReceiptSource = XLOG_FROM_STREAM;
}
break;
***************
*** 9558,9567 **** WaitForWALToBecomeAvailable(XLogRecPtr RecPtr, bool randAccess,
int sources;
pg_time_t now;
! if (readFile >= 0)
{
! close(readFile);
! readFile = -1;
}
/* Reset curFileTLI if random fetch. */
if (randAccess)
--- 9159,9168 ----
int sources;
pg_time_t now;
! if (private->readFile >= 0)
{
! close(private->readFile);
! private->readFile = -1;
}
/* Reset curFileTLI if random fetch. */
if (randAccess)
***************
*** 9572,9583 **** WaitForWALToBecomeAvailable(XLogRecPtr RecPtr, bool randAccess,
* from pg_xlog.
*/
sources = XLOG_FROM_ARCHIVE | XLOG_FROM_PG_XLOG;
! if (!(sources & ~failedSources))
{
/*
* We've exhausted all options for retrieving the file. Retry.
*/
! failedSources = 0;
/*
* Before we sleep, re-scan for possible new timelines if we
--- 9173,9184 ----
* from pg_xlog.
*/
sources = XLOG_FROM_ARCHIVE | XLOG_FROM_PG_XLOG;
! if (!(sources & ~private->failedSources))
{
/*
* We've exhausted all options for retrieving the file. Retry.
*/
! private->failedSources = 0;
/*
* Before we sleep, re-scan for possible new timelines if we
***************
*** 9606,9635 **** WaitForWALToBecomeAvailable(XLogRecPtr RecPtr, bool randAccess,
* stream the missing WAL, before retrying to restore from
* archive/pg_xlog.
*
! * If fetching_ckpt is TRUE, RecPtr points to the initial
! * checkpoint location. In that case, we use RedoStartLSN as
! * the streaming start position instead of RecPtr, so that
! * when we later jump backwards to start redo at RedoStartLSN,
! * we will have the logs streamed already.
*/
if (PrimaryConnInfo)
{
! XLogRecPtr ptr = fetching_ckpt ? RedoStartLSN : RecPtr;
RequestXLogStreaming(ptr, PrimaryConnInfo);
continue;
}
}
/* Don't try to read from a source that just failed */
! sources &= ~failedSources;
! readFile = XLogFileReadAnyTLI(readSegNo, DEBUG2, sources);
! if (readFile >= 0)
break;
/*
* Nope, not found in archive and/or pg_xlog.
*/
! failedSources |= sources;
/*
* Check to see if the trigger file exists. Note that we do this
--- 9207,9238 ----
* stream the missing WAL, before retrying to restore from
* archive/pg_xlog.
*
! * If we're fetching a checkpoint record, RecPtr points to the
! * initial checkpoint location. In that case, we use
! * RedoStartLSN as the streaming start position instead of
! * RecPtr, so that when we later jump backwards to start redo
! * at RedoStartLSN, we will have the logs streamed already.
*/
if (PrimaryConnInfo)
{
! XLogRecPtr ptr = private->fetching_ckpt ?
! RedoStartLSN : RecPtr;
RequestXLogStreaming(ptr, PrimaryConnInfo);
continue;
}
}
/* Don't try to read from a source that just failed */
! sources &= ~private->failedSources;
! private->readFile = XLogFileReadAnyTLI(private, private->readSegNo,
! DEBUG2, sources);
! if (private->readFile >= 0)
break;
/*
* Nope, not found in archive and/or pg_xlog.
*/
! private->failedSources |= sources;
/*
* Check to see if the trigger file exists. Note that we do this
***************
*** 9669,9680 **** WaitForWALToBecomeAvailable(XLogRecPtr RecPtr, bool randAccess,
* you are about to ereport(), or you might cause a later message to be
* erroneously suppressed.
*/
! static int
! emode_for_corrupt_record(int emode, XLogRecPtr RecPtr)
{
static XLogRecPtr lastComplaint = 0;
! if (readSource == XLOG_FROM_PG_XLOG && emode == LOG)
{
if (XLByteEQ(RecPtr, lastComplaint))
emode = DEBUG1;
--- 9272,9283 ----
* you are about to ereport(), or you might cause a later message to be
* erroneously suppressed.
*/
! int
! emode_for_corrupt_record(int emode, int source, XLogRecPtr RecPtr)
{
static XLogRecPtr lastComplaint = 0;
! if (source == XLOG_FROM_PG_XLOG && emode == LOG)
{
if (XLByteEQ(RecPtr, lastComplaint))
emode = DEBUG1;
*** /dev/null
--- b/src/backend/access/transam/xlogreader.c
***************
*** 0 ****
--- 1,532 ----
+ /*-------------------------------------------------------------------------
+ *
+ * xlogreader.c
+ * Generic xlog reading facility
+ *
+ * Portions Copyright (c) 2012, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ * src/backend/access/transam/xlogreader.c
+ *
+ * NOTES
+ * Documentation about how do use this interface can be found in
+ * xlogreader.h, more specifically in the definition of the
+ * XLogReaderState struct where all parameters are documented.
+ *
+ * TODO:
+ * * usable without backend code around
+ *-------------------------------------------------------------------------
+ */
+
+ #include "postgres.h"
+
+ #include "access/transam.h"
+ #include "access/xlog_internal.h"
+ #include "access/xlogreader.h"
+ #include "catalog/pg_control.h"
+
+ static bool allocate_recordbuf(XLogReaderState *state, uint32 reclength);
+ static bool ValidXLogRecordHeader(XLogRecPtr RecPtr, XLogRecPtr PrevRecPtr,
+ XLogRecord *record, int emode, bool randAccess);
+ static bool RecordIsValid(XLogRecord *record, XLogRecPtr recptr, int emode);
+
+ /*
+ * Initialize a new xlog reader
+ */
+ XLogReaderState *
+ XLogReaderAllocate(XLogRecPtr startpoint,
+ XLogPageReadCB pagereadfunc, void *private_data)
+ {
+ XLogReaderState *state;
+
+ state = (XLogReaderState *) palloc0(sizeof(XLogReaderState));
+
+ /*
+ * Permanently allocate readBuf. We do it this way, rather than just
+ * making a static array, for two reasons: (1) no need to waste the
+ * storage in most instantiations of the backend; (2) a static char array
+ * isn't guaranteed to have any particular alignment, whereas malloc()
+ * will provide MAXALIGN'd storage.
+ */
+ state->readBuf = (char *) malloc(XLOG_BLCKSZ);
+
+ state->read_page = pagereadfunc;
+ state->private_data = private_data;
+ state->EndRecPtr = startpoint;
+
+ /*
+ * Allocate an initial readRecordBuf of minimal size, which can later be
+ * enlarged if necessary.
+ */
+ if (!allocate_recordbuf(state, 0))
+ {
+ free(state->readBuf);
+ pfree(state);
+ return NULL;
+ }
+
+ return state;
+ }
+
+ void
+ XLogReaderFree(XLogReaderState *state)
+ {
+ if (state->readRecordBuf)
+ free(state->readRecordBuf);
+ free(state->readBuf);
+ pfree(state);
+ }
+
+ /*
+ * Allocate readRecordBuf to fit a record of at least the given length.
+ * Returns true if successful, false if out of memory.
+ *
+ * readRecordBufSize is set to the new buffer size.
+ *
+ * To avoid useless small increases, round its size to a multiple of
+ * XLOG_BLCKSZ, and make sure it's at least 4*Max(BLCKSZ, XLOG_BLCKSZ) to start
+ * with. (That is enough for all "normal" records, but very large commit or
+ * abort records might need more space.)
+ */
+ static bool
+ allocate_recordbuf(XLogReaderState *state, uint32 reclength)
+ {
+ uint32 newSize = reclength;
+
+ newSize += XLOG_BLCKSZ - (newSize % XLOG_BLCKSZ);
+ newSize = Max(newSize, 4 * Max(BLCKSZ, XLOG_BLCKSZ));
+
+ if (state->readRecordBuf)
+ free(state->readRecordBuf);
+ state->readRecordBuf = (char *) malloc(newSize);
+ if (!state->readRecordBuf)
+ {
+ state->readRecordBufSize = 0;
+ return false;
+ }
+
+ state->readRecordBufSize = newSize;
+ return true;
+ }
+
+ /*
+ * Attempt to read an XLOG record.
+ *
+ * If RecPtr is not NULL, try to read a record at that position. Otherwise
+ * try to read a record just after the last one previously read.
+ *
+ * If no valid record is available, returns NULL, or fails if emode is PANIC.
+ * (emode must be either PANIC, LOG)
+ *
+ * The record is copied into readRecordBuf, so that on successful return,
+ * the returned record pointer always points there.
+ */
+ XLogRecord *
+ XLogReadRecord(XLogReaderState *state, XLogRecPtr RecPtr, int emode)
+ {
+ XLogRecord *record;
+ XLogRecPtr tmpRecPtr = state->EndRecPtr;
+ bool randAccess = false;
+ uint32 len,
+ total_len;
+ uint32 targetRecOff;
+ uint32 pageHeaderSize;
+ bool gotheader;
+
+ if (RecPtr == InvalidXLogRecPtr)
+ {
+ RecPtr = tmpRecPtr;
+
+ /*
+ * RecPtr is pointing to end+1 of the previous WAL record. If we're
+ * at a page boundary, no more records can fit on the current page. We
+ * must skip over the page header, but we can't do that until we've
+ * read in the page, since the header size is variable.
+ */
+ }
+ else
+ {
+ /*
+ * In this case, the passed-in record pointer should already be
+ * pointing to a valid record starting position.
+ */
+ if (!XRecOffIsValid(RecPtr))
+ ereport(PANIC,
+ (errmsg("invalid record offset at %X/%X",
+ (uint32) (RecPtr >> 32), (uint32) RecPtr)));
+ randAccess = true; /* allow curFileTLI to go backwards too */
+ }
+
+ /* Read the page containing the record */
+ if (!state->read_page(state, RecPtr, emode, randAccess, state->readBuf,
+ state->private_data))
+ return NULL;
+
+ pageHeaderSize = XLogPageHeaderSize((XLogPageHeader) state->readBuf);
+ targetRecOff = RecPtr % XLOG_BLCKSZ;
+ if (targetRecOff == 0)
+ {
+ /*
+ * At page start, so skip over page header. The Assert checks that
+ * we're not scribbling on caller's record pointer; it's OK because we
+ * can only get here in the continuing-from-prev-record case, since
+ * XRecOffIsValid rejected the zero-page-offset case otherwise. XXX:
+ * does this assert make sense now that RecPtr is not a pointer?
+ */
+ Assert(RecPtr == tmpRecPtr);
+ RecPtr += pageHeaderSize;
+ targetRecOff = pageHeaderSize;
+ }
+ else if (targetRecOff < pageHeaderSize)
+ {
+ ereport(emode_for_corrupt_record(emode, 0, RecPtr),
+ (errmsg("invalid record offset at %X/%X",
+ (uint32) (RecPtr >> 32), (uint32) RecPtr)));
+ return NULL;
+ }
+ if ((((XLogPageHeader) state->readBuf)->xlp_info & XLP_FIRST_IS_CONTRECORD) &&
+ targetRecOff == pageHeaderSize)
+ {
+ ereport(emode_for_corrupt_record(emode, 0, RecPtr),
+ (errmsg("contrecord is requested by %X/%X",
+ (uint32) (RecPtr >> 32), (uint32) RecPtr)));
+ return NULL;
+ }
+
+ /*
+ * Read the record length.
+ *
+ * NB: Even though we use an XLogRecord pointer here, the whole record
+ * header might not fit on this page. xl_tot_len is the first field of the
+ * struct, so it must be on this page (the records are MAXALIGNed), but we
+ * cannot access any other fields until we've verified that we got the
+ * whole header.
+ */
+ record = (XLogRecord *) (state->readBuf + RecPtr % XLOG_BLCKSZ);
+ total_len = record->xl_tot_len;
+
+ /*
+ * If the whole record header is on this page, validate it immediately.
+ * Otherwise do just a basic sanity check on xl_tot_len, and validate the
+ * rest of the header after reading it from the next page. The xl_tot_len
+ * check is necessary here to ensure that we enter the "Need to reassemble
+ * record" code path below; otherwise we might fail to apply
+ * ValidXLogRecordHeader at all.
+ */
+ if (targetRecOff <= XLOG_BLCKSZ - SizeOfXLogRecord)
+ {
+ if (!ValidXLogRecordHeader(RecPtr, state->ReadRecPtr, record, emode,
+ randAccess))
+ return NULL;
+ gotheader = true;
+ }
+ else
+ {
+ if (total_len < SizeOfXLogRecord)
+ {
+ ereport(emode_for_corrupt_record(emode, 0, RecPtr),
+ (errmsg("invalid record length at %X/%X",
+ (uint32) (RecPtr >> 32), (uint32) RecPtr)));
+ return NULL;
+ }
+ gotheader = false;
+ }
+
+ /*
+ * Enlarge readRecordBuf as needed.
+ */
+ if (total_len > state->readRecordBufSize &&
+ !allocate_recordbuf(state, total_len))
+ {
+ /* We treat this as a "bogus data" condition */
+ ereport(emode_for_corrupt_record(emode, 0, RecPtr),
+ (errmsg("record length %u at %X/%X too long",
+ total_len, (uint32) (RecPtr >> 32), (uint32) RecPtr)));
+ return NULL;
+ }
+
+ len = XLOG_BLCKSZ - RecPtr % XLOG_BLCKSZ;
+ if (total_len > len)
+ {
+ /* Need to reassemble record */
+ char *contrecord;
+ XLogPageHeader pageHeader;
+ XLogRecPtr pagelsn;
+ char *buffer;
+ uint32 gotlen;
+
+ /* Initialize pagelsn to the beginning of the page this record is on */
+ pagelsn = (RecPtr / XLOG_BLCKSZ) * XLOG_BLCKSZ;
+
+ /* Copy the first fragment of the record from the first page. */
+ memcpy(state->readRecordBuf,
+ state->readBuf + RecPtr % XLOG_BLCKSZ, len);
+ buffer = state->readRecordBuf + len;
+ gotlen = len;
+
+ do
+ {
+ /* Calculate pointer to beginning of next page */
+ XLByteAdvance(pagelsn, XLOG_BLCKSZ);
+ /* Wait for the next page to become available */
+ if (!state->read_page(state, pagelsn, emode, false, state->readBuf,
+ state->private_data))
+ return NULL;
+
+ /* Check that the continuation on next page looks valid */
+ pageHeader = (XLogPageHeader) state->readBuf;
+ if (!(pageHeader->xlp_info & XLP_FIRST_IS_CONTRECORD))
+ {
+ ereport(emode_for_corrupt_record(emode, 0, RecPtr),
+ (errmsg("there is no contrecord flag at %X/%X",
+ (uint32) (RecPtr >> 32), (uint32) RecPtr)));
+ return NULL;
+ }
+
+ /*
+ * Cross-check that xlp_rem_len agrees with how much of the record
+ * we expect there to be left.
+ */
+ if (pageHeader->xlp_rem_len == 0 ||
+ total_len != (pageHeader->xlp_rem_len + gotlen))
+ {
+ ereport(emode_for_corrupt_record(emode, 0, RecPtr),
+ (errmsg("invalid contrecord length %u at %X/%X",
+ pageHeader->xlp_rem_len,
+ (uint32) (RecPtr >> 32), (uint32) RecPtr)));
+ return NULL;
+ }
+
+ /* Append the continuation from this page to the buffer */
+ pageHeaderSize = XLogPageHeaderSize(pageHeader);
+ contrecord = (char *) state->readBuf + pageHeaderSize;
+ len = XLOG_BLCKSZ - pageHeaderSize;
+ if (pageHeader->xlp_rem_len < len)
+ len = pageHeader->xlp_rem_len;
+ memcpy(buffer, (char *) contrecord, len);
+ buffer += len;
+ gotlen += len;
+
+ /* If we just reassembled the record header, validate it. */
+ if (!gotheader)
+ {
+ record = (XLogRecord *) state->readRecordBuf;
+ if (!ValidXLogRecordHeader(RecPtr, state->ReadRecPtr, record,
+ emode, randAccess))
+ return NULL;
+ gotheader = true;
+ }
+ } while (pageHeader->xlp_rem_len > len);
+
+ record = (XLogRecord *) state->readRecordBuf;
+ if (!RecordIsValid(record, RecPtr, emode))
+ return NULL;
+ pageHeaderSize = XLogPageHeaderSize((XLogPageHeader) state->readBuf);
+ state->ReadRecPtr = RecPtr;
+ state->EndRecPtr = pagelsn + pageHeaderSize + MAXALIGN(pageHeader->xlp_rem_len);
+ }
+ else
+ {
+ /* Record does not cross a page boundary */
+ if (!RecordIsValid(record, RecPtr, emode))
+ return NULL;
+ state->EndRecPtr = RecPtr + MAXALIGN(total_len);
+
+ state->ReadRecPtr = RecPtr;
+ memcpy(state->readRecordBuf, record, total_len);
+ }
+
+ /*
+ * Special processing if it's an XLOG SWITCH record
+ */
+ if (record->xl_rmid == RM_XLOG_ID && record->xl_info == XLOG_SWITCH)
+ {
+ /* Pretend it extends to end of segment */
+ state->EndRecPtr += XLogSegSize - 1;
+ state->EndRecPtr -= state->EndRecPtr % XLogSegSize;
+ }
+
+ return record;
+ }
+
+ /*
+ * Validate an XLOG record header.
+ *
+ * This is just a convenience subroutine to avoid duplicated code in
+ * XLogReadRecord. It's not intended for use from anywhere else.
+ */
+ static bool
+ ValidXLogRecordHeader(XLogRecPtr RecPtr, XLogRecPtr PrevRecPtr,
+ XLogRecord *record, int emode, bool randAccess)
+ {
+ /*
+ * xl_len == 0 is bad data for everything except XLOG SWITCH, where it is
+ * required.
+ */
+ if (record->xl_rmid == RM_XLOG_ID && record->xl_info == XLOG_SWITCH)
+ {
+ if (record->xl_len != 0)
+ {
+ ereport(emode_for_corrupt_record(emode, 0, RecPtr),
+ (errmsg("invalid xlog switch record at %X/%X",
+ (uint32) (RecPtr >> 32), (uint32) RecPtr)));
+ return false;
+ }
+ }
+ else if (record->xl_len == 0)
+ {
+ ereport(emode_for_corrupt_record(emode, 0, RecPtr),
+ (errmsg("record with zero length at %X/%X",
+ (uint32) (RecPtr >> 32), (uint32) RecPtr)));
+ return false;
+ }
+ if (record->xl_tot_len < SizeOfXLogRecord + record->xl_len ||
+ record->xl_tot_len > SizeOfXLogRecord + record->xl_len +
+ XLR_MAX_BKP_BLOCKS * (sizeof(BkpBlock) + BLCKSZ))
+ {
+ ereport(emode_for_corrupt_record(emode, 0, RecPtr),
+ (errmsg("invalid record length at %X/%X",
+ (uint32) (RecPtr >> 32), (uint32) RecPtr)));
+ return false;
+ }
+ if (record->xl_rmid > RM_MAX_ID)
+ {
+ ereport(emode_for_corrupt_record(emode, 0, RecPtr),
+ (errmsg("invalid resource manager ID %u at %X/%X",
+ record->xl_rmid, (uint32) (RecPtr >> 32),
+ (uint32) RecPtr)));
+ return false;
+ }
+ if (randAccess)
+ {
+ /*
+ * We can't exactly verify the prev-link, but surely it should be less
+ * than the record's own address.
+ */
+ if (!XLByteLT(record->xl_prev, RecPtr))
+ {
+ ereport(emode_for_corrupt_record(emode, 0, RecPtr),
+ (errmsg("record with incorrect prev-link %X/%X at %X/%X",
+ (uint32) (record->xl_prev >> 32),
+ (uint32) record->xl_prev,
+ (uint32) (RecPtr >> 32), (uint32) RecPtr)));
+ return false;
+ }
+ }
+ else
+ {
+ /*
+ * Record's prev-link should exactly match our previous location. This
+ * check guards against torn WAL pages where a stale but valid-looking
+ * WAL record starts on a sector boundary.
+ */
+ if (!XLByteEQ(record->xl_prev, PrevRecPtr))
+ {
+ ereport(emode_for_corrupt_record(emode, 0, RecPtr),
+ (errmsg("record with incorrect prev-link %X/%X at %X/%X",
+ (uint32) (record->xl_prev >> 32),
+ (uint32) record->xl_prev,
+ (uint32) (RecPtr >> 32), (uint32) RecPtr)));
+ return false;
+ }
+ }
+
+ return true;
+ }
+
+
+ /*
+ * CRC-check an XLOG record. We do not believe the contents of an XLOG
+ * record (other than to the minimal extent of computing the amount of
+ * data to read in) until we've checked the CRCs.
+ *
+ * We assume all of the record (that is, xl_tot_len bytes) has been read
+ * into memory at *record. Also, ValidXLogRecordHeader() has accepted the
+ * record's header, which means in particular that xl_tot_len is at least
+ * SizeOfXlogRecord, so it is safe to fetch xl_len.
+ */
+ static bool
+ RecordIsValid(XLogRecord *record, XLogRecPtr recptr, int emode)
+ {
+ pg_crc32 crc;
+ int i;
+ uint32 len = record->xl_len;
+ BkpBlock bkpb;
+ char *blk;
+ size_t remaining = record->xl_tot_len;
+
+ /* First the rmgr data */
+ if (remaining < SizeOfXLogRecord + len)
+ {
+ /* ValidXLogRecordHeader() should've caught this already... */
+ ereport(emode_for_corrupt_record(emode, 0, recptr),
+ (errmsg("invalid record length at %X/%X",
+ (uint32) (recptr >> 32), (uint32) recptr)));
+ return false;
+ }
+ remaining -= SizeOfXLogRecord + len;
+ INIT_CRC32(crc);
+ COMP_CRC32(crc, XLogRecGetData(record), len);
+
+ /* Add in the backup blocks, if any */
+ blk = (char *) XLogRecGetData(record) + len;
+ for (i = 0; i < XLR_MAX_BKP_BLOCKS; i++)
+ {
+ uint32 blen;
+
+ if (!(record->xl_info & XLR_BKP_BLOCK(i)))
+ continue;
+
+ if (remaining < sizeof(BkpBlock))
+ {
+ ereport(emode_for_corrupt_record(emode, 0, recptr),
+ (errmsg("invalid backup block size in record at %X/%X",
+ (uint32) (recptr >> 32), (uint32) recptr)));
+ return false;
+ }
+ memcpy(&bkpb, blk, sizeof(BkpBlock));
+
+ if (bkpb.hole_offset + bkpb.hole_length > BLCKSZ)
+ {
+ ereport(emode_for_corrupt_record(emode, 0, recptr),
+ (errmsg("incorrect hole size in record at %X/%X",
+ (uint32) (recptr >> 32), (uint32) recptr)));
+ return false;
+ }
+ blen = sizeof(BkpBlock) + BLCKSZ - bkpb.hole_length;
+
+ if (remaining < blen)
+ {
+ ereport(emode_for_corrupt_record(emode, 0, recptr),
+ (errmsg("invalid backup block size in record at %X/%X",
+ (uint32) (recptr >> 32), (uint32) recptr)));
+ return false;
+ }
+ remaining -= blen;
+ COMP_CRC32(crc, blk, blen);
+ blk += blen;
+ }
+
+ /* Check that xl_tot_len agrees with our calculation */
+ if (remaining != 0)
+ {
+ ereport(emode_for_corrupt_record(emode, 0, recptr),
+ (errmsg("incorrect total length in record at %X/%X",
+ (uint32) (recptr >> 32), (uint32) recptr)));
+ return false;
+ }
+
+ /* Finally include the record header */
+ COMP_CRC32(crc, (char *) record, offsetof(XLogRecord, xl_crc));
+ FIN_CRC32(crc);
+
+ if (!EQ_CRC32(record->xl_crc, crc))
+ {
+ ereport(emode_for_corrupt_record(emode, 0, recptr),
+ (errmsg("incorrect resource manager data checksum in record at %X/%X",
+ (uint32) (recptr >> 32), (uint32) recptr)));
+ return false;
+ }
+
+ return true;
+ }
*** a/src/include/access/xlog_internal.h
--- b/src/include/access/xlog_internal.h
***************
*** 231,236 **** extern XLogRecPtr RequestXLogSwitch(void);
--- 231,244 ----
extern void GetOldestRestartPoint(XLogRecPtr *oldrecptr, TimeLineID *oldtli);
+
+ /*
+ * Exported so that xlogreader.c can call this. TODO: Should be refactored
+ * into a callback, or just have xlogreader return the error string and have
+ * the caller of XLogReadRecord() do the ereport() call.
+ */
+ extern int emode_for_corrupt_record(int emode, int readSource, XLogRecPtr RecPtr);
+
/*
* Exported for the functions in timeline.c and xlogarchive.c. Only valid
* in the startup process.
*** /dev/null
--- b/src/include/access/xlogreader.h
***************
*** 0 ****
--- 1,97 ----
+ /*-------------------------------------------------------------------------
+ *
+ * readxlog.h
+ *
+ * Generic xlog reading facility.
+ *
+ * Portions Copyright (c) 2012, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ * src/include/access/xlogreader.h
+ *
+ * NOTES
+ * Check the definition of the XLogReaderState struct for instructions on
+ * how to use the XLogReader infrastructure.
+ *
+ * The basic idea is to allocate an XLogReaderState via
+ * XLogReaderAllocate, and call XLogReadRecord() until it returns NULL.
+ *-------------------------------------------------------------------------
+ */
+ #ifndef XLOGREADER_H
+ #define XLOGREADER_H
+
+ #include "access/xlog_internal.h"
+
+ struct XLogReaderState;
+
+ /*
+ * The callbacks are explained in more detail inside the XLogReaderState
+ * struct.
+ */
+ typedef bool (*XLogPageReadCB) (struct XLogReaderState *state,
+ XLogRecPtr RecPtr, int emode,
+ bool randAccess,
+ char *readBuf,
+ void *private_data);
+
+ typedef struct XLogReaderState
+ {
+ /* ----------------------------------------
+ * Public parameters
+ * ----------------------------------------
+ */
+
+ /*
+ * Data input callback (mandatory).
+ *
+ * This callback shall read XLOG_BLKSZ bytes, from the location 'RecPtr',
+ * into memory pointed at by 'readBuf' parameter. The callback shall
+ * return true on success, false if the page could not be read.
+ */
+ XLogPageReadCB read_page;
+
+ /*
+ * Opaque data for callbacks to use. Not used by XLogReader.
+ */
+ void *private_data;
+
+ /*
+ * From where to where are we reading
+ */
+ XLogRecPtr ReadRecPtr; /* start of last record read */
+ XLogRecPtr EndRecPtr; /* end+1 of last record read */
+
+ /* ----------------------------------------
+ * private/internal state
+ * ----------------------------------------
+ */
+
+ /* Buffer for currently read page (XLOG_BLCKSZ bytes) */
+ char *readBuf;
+
+ /* Buffer for current ReadRecord result (expandable) */
+ char *readRecordBuf;
+ uint32 readRecordBufSize;
+ } XLogReaderState;
+
+ /*
+ * Get a new XLogReader
+ *
+ * At least the read_page callback, startptr and endptr have to be set before
+ * the reader can be used.
+ */
+ extern XLogReaderState *XLogReaderAllocate(XLogRecPtr startpoint,
+ XLogPageReadCB pagereadfunc, void *private_data);
+
+ /*
+ * Free an XLogReader
+ */
+ extern void XLogReaderFree(XLogReaderState *state);
+
+ /*
+ * Read the next record from xlog. Returns NULL on end-of-WAL or on failure.
+ */
+ extern XLogRecord *XLogReadRecord(XLogReaderState *state, XLogRecPtr ptr,
+ int emode);
+
+ #endif /* XLOGREADER_H */