diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c index 477709bbc23..8ecb5ea55c5 100644 --- a/src/backend/access/transam/twophase.c +++ b/src/backend/access/transam/twophase.c @@ -1386,15 +1386,21 @@ XlogReadTwoPhaseData(XLogRecPtr lsn, char **buf, int *len) XLogReaderState *xlogreader; char *errormsg; - xlogreader = XLogReaderAllocate(wal_segment_size, &read_local_xlog_page, - NULL); + xlogreader = XLogReaderAllocate(wal_segment_size); if (!xlogreader) ereport(ERROR, (errcode(ERRCODE_OUT_OF_MEMORY), errmsg("out of memory"), errdetail("Failed while allocating a WAL reading processor."))); + xlogreader->readBuf = palloc(XLOG_BLCKSZ); + + XLogBeginRead(xlogreader, lsn); + while (XLogReadRecord(xlogreader, &record, &errormsg) == + XLREAD_NEED_DATA) + { + read_local_xlog_page(xlogreader); + } - record = XLogReadRecord(xlogreader, lsn, &errormsg); if (record == NULL) ereport(ERROR, (errcode_for_file_access(), @@ -1416,6 +1422,7 @@ XlogReadTwoPhaseData(XLogRecPtr lsn, char **buf, int *len) *buf = palloc(sizeof(char) * XLogRecGetDataLen(xlogreader)); memcpy(*buf, XLogRecGetData(xlogreader), sizeof(char) * XLogRecGetDataLen(xlogreader)); + pfree(xlogreader->readBuf); XLogReaderFree(xlogreader); } diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index e651a841bbe..1bb303a90dc 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -803,13 +803,6 @@ static XLogSource readSource = 0; /* XLOG_FROM_* code */ static XLogSource currentSource = 0; /* XLOG_FROM_* code */ static bool lastSourceFailed = false; -typedef struct XLogPageReadPrivate -{ - int emode; - bool fetching_ckpt; /* are we fetching a checkpoint record? */ - bool randAccess; -} XLogPageReadPrivate; - /* * These variables track when we last obtained some WAL data to process, * and where we got it from. (XLogReceiptSource is initially the same as @@ -884,9 +877,8 @@ static bool InstallXLogFileSegment(XLogSegNo *segno, char *tmppath, static int XLogFileRead(XLogSegNo segno, int emode, TimeLineID tli, int source, bool notfoundOk); static int XLogFileReadAnyTLI(XLogSegNo segno, int emode, int source); -static int XLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr, - int reqLen, XLogRecPtr targetRecPtr, char *readBuf, - TimeLineID *readTLI); +static void XLogPageRead(XLogReaderState *xlogreader, + bool fetching_ckpt, int emode, bool randAccess); static bool WaitForWALToBecomeAvailable(XLogRecPtr RecPtr, bool randAccess, bool fetching_ckpt, XLogRecPtr tliRecPtr); static int emode_for_corrupt_record(int emode, XLogRecPtr RecPtr); @@ -1195,7 +1187,7 @@ XLogInsertRecord(XLogRecData *rdata, appendBinaryStringInfo(&recordBuf, rdata->data, rdata->len); if (!debug_reader) - debug_reader = XLogReaderAllocate(wal_segment_size, NULL, NULL); + debug_reader = XLogReaderAllocate(wal_segment_size); if (!debug_reader) { @@ -4247,21 +4239,29 @@ ReadRecord(XLogReaderState *xlogreader, XLogRecPtr RecPtr, int emode, bool fetching_ckpt) { XLogRecord *record; - XLogPageReadPrivate *private = (XLogPageReadPrivate *) xlogreader->private_data; - - /* Pass through parameters to XLogPageRead */ - private->fetching_ckpt = fetching_ckpt; - private->emode = emode; - private->randAccess = (RecPtr != InvalidXLogRecPtr); + bool randAccess = (RecPtr != InvalidXLogRecPtr); /* This is the first attempt to read this page. */ lastSourceFailed = false; + if (RecPtr != InvalidXLogRecPtr) + XLogBeginRead(xlogreader, RecPtr); + for (;;) { char *errormsg; - record = XLogReadRecord(xlogreader, RecPtr, &errormsg); + while (XLogReadRecord(xlogreader, &record, &errormsg) + == XLREAD_NEED_DATA) + { + XLogPageRead(xlogreader, fetching_ckpt, emode, randAccess); + if (xlogreader->readLen == -1) + { + record = NULL; + break; + } + } + ReadRecPtr = xlogreader->ReadRecPtr; EndRecPtr = xlogreader->EndRecPtr; if (record == NULL) @@ -4271,6 +4271,8 @@ ReadRecord(XLogReaderState *xlogreader, XLogRecPtr RecPtr, int emode, close(readFile); readFile = -1; } + /* Force the WAL page to be reloaded. */ + xlogreader->readLen = -1; /* * We only end up here without a message when XLogPageRead() @@ -6211,7 +6213,6 @@ StartupXLOG(void) bool backupFromStandby = false; DBState dbstate_at_startup; XLogReaderState *xlogreader; - XLogPageReadPrivate private; bool fast_promoted = false; struct stat st; @@ -6352,13 +6353,13 @@ StartupXLOG(void) OwnLatch(&XLogCtl->recoveryWakeupLatch); /* Set up XLOG reader facility */ - MemSet(&private, 0, sizeof(XLogPageReadPrivate)); - xlogreader = XLogReaderAllocate(wal_segment_size, &XLogPageRead, &private); + xlogreader = XLogReaderAllocate(wal_segment_size); if (!xlogreader) ereport(ERROR, (errcode(ERRCODE_OUT_OF_MEMORY), errmsg("out of memory"), errdetail("Failed while allocating a WAL reading processor."))); + xlogreader->readBuf = palloc(XLOG_BLCKSZ); xlogreader->system_identifier = ControlFile->system_identifier; /* @@ -7730,6 +7731,7 @@ StartupXLOG(void) close(readFile); readFile = -1; } + pfree(xlogreader->readBuf); XLogReaderFree(xlogreader); /* @@ -11521,13 +11523,15 @@ CancelBackup(void) * XLogPageRead() to try fetching the record from another source, or to * sleep and retry. */ -static int -XLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr, int reqLen, - XLogRecPtr targetRecPtr, char *readBuf, TimeLineID *readTLI) -{ - XLogPageReadPrivate *private = - (XLogPageReadPrivate *) xlogreader->private_data; - int emode = private->emode; +static void +XLogPageRead(XLogReaderState *xlogreader, + bool fetching_ckpt, int emode, bool randAccess) +{ + XLogRecPtr targetPagePtr = xlogreader->loadPagePtr; + int reqLen = xlogreader->loadLen; + XLogRecPtr targetRecPtr = xlogreader->ReadRecPtr; + char *readBuf = xlogreader->readBuf; + TimeLineID *readTLI = &xlogreader->readPageTLI; uint32 targetPageOff; XLogSegNo targetSegNo PG_USED_FOR_ASSERTS_ONLY; int r; @@ -11570,8 +11574,8 @@ retry: receivedUpto < targetPagePtr + reqLen)) { if (!WaitForWALToBecomeAvailable(targetPagePtr + reqLen, - private->randAccess, - private->fetching_ckpt, + randAccess, + fetching_ckpt, targetRecPtr)) { if (readFile >= 0) @@ -11580,7 +11584,9 @@ retry: readLen = 0; readSource = 0; - return -1; + xlogreader->readPagePtr = InvalidXLogRecPtr; + xlogreader->readLen = -1; + return; } } @@ -11675,7 +11681,9 @@ retry: goto next_record_is_invalid; } - return readLen; + xlogreader->readPagePtr = targetPagePtr; + xlogreader->readLen = readLen; + return; next_record_is_invalid: lastSourceFailed = true; @@ -11689,8 +11697,10 @@ next_record_is_invalid: /* In standby-mode, keep trying */ if (StandbyMode) goto retry; - else - return -1; + + xlogreader->readPagePtr = InvalidXLogRecPtr; + xlogreader->readLen = -1; + return; } /* diff --git a/src/backend/access/transam/xlogreader.c b/src/backend/access/transam/xlogreader.c index c6faf48d24f..a7a06c0c6aa 100644 --- a/src/backend/access/transam/xlogreader.c +++ b/src/backend/access/transam/xlogreader.c @@ -30,14 +30,46 @@ #include "utils/memutils.h" #endif +/* + * XLogBeginRead() puts the reader to XLREAD_NEED_SEGMENT_HEADER, so that + * the first call to XLogReadRecord() will read and validate the segment + * header. After that, we loop through NEED_TOT_LEN -> NEED_TOT_LEN -> + * NEED_CONTINUATION -> NEXT_RECORD. + */ +enum XLogReadRecordState +{ + XLREAD_NEED_SEGMENT_HEADER, + XLREAD_SKIP_CONTRECORDS, + + /* between records */ + XLREAD_NEXT_RECORD, + + /* + * ReadRecPtr has been set to point to beginning of record, nothing has + * been read yet. + */ + XLREAD_NEED_TOT_LEN, + + /* + * We have read a record partially, but need to read one or more continuation + * records to complete it. recordGotLen indicates how much we have read + * already, and recordRemainLen indicates how much is left. readRecordBuf holds + * the partially assembled record. recordContRecPtr points to the beginning + * of the next page where to continue. + */ + XLREAD_NEED_CONTINUATION +}; + static bool allocate_recordbuf(XLogReaderState *state, uint32 reclength); static bool ValidXLogRecordHeader(XLogReaderState *state, XLogRecPtr RecPtr, - XLogRecPtr PrevRecPtr, XLogRecord *record, bool randAccess); + XLogRecPtr PrevRecPtr, XLogRecord *record); static bool ValidXLogRecord(XLogReaderState *state, XLogRecord *record, XLogRecPtr recptr); -static int ReadPageInternal(XLogReaderState *state, XLogRecPtr pageptr, - int reqLen); +static bool XLogNeedPage(XLogReaderState *state, XLogRecPtr pageptr); +static bool XLogNeedData(XLogReaderState *state, XLogRecPtr pageptr, + int reqLen); +static void XLogReaderInvalReadState(XLogReaderState *state); static void report_invalid_record(XLogReaderState *state, const char *fmt,...) pg_attribute_printf(2, 3); static void ResetDecoder(XLogReaderState *state); @@ -67,8 +99,7 @@ report_invalid_record(XLogReaderState *state, const char *fmt,...) * Returns NULL if the xlogreader couldn't be allocated. */ XLogReaderState * -XLogReaderAllocate(int wal_segment_size, XLogPageReadCB pagereadfunc, - void *private_data) +XLogReaderAllocate(int wal_segment_size) { XLogReaderState *state; @@ -80,32 +111,12 @@ XLogReaderAllocate(int wal_segment_size, XLogPageReadCB pagereadfunc, state->max_block_id = -1; - /* - * 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 - * palloc_extended() will provide MAXALIGN'd storage. - */ - state->readBuf = (char *) palloc_extended(XLOG_BLCKSZ, - MCXT_ALLOC_NO_OOM); - if (!state->readBuf) - { - pfree(state); - return NULL; - } - state->wal_segment_size = wal_segment_size; - state->read_page = pagereadfunc; - /* system_identifier initialized to zeroes above */ - state->private_data = private_data; - /* ReadRecPtr and EndRecPtr initialized to zeroes above */ - /* readSegNo, readOff, readLen, readPageTLI initialized to zeroes above */ + /* All members are initialized to zeroes above */ state->errormsg_buf = palloc_extended(MAX_ERRORMSG_LEN + 1, MCXT_ALLOC_NO_OOM); if (!state->errormsg_buf) { - pfree(state->readBuf); pfree(state); return NULL; } @@ -118,11 +129,12 @@ XLogReaderAllocate(int wal_segment_size, XLogPageReadCB pagereadfunc, if (!allocate_recordbuf(state, 0)) { pfree(state->errormsg_buf); - pfree(state->readBuf); pfree(state); return NULL; } + XLogReaderInvalReadState(state); + return state; } @@ -142,7 +154,6 @@ XLogReaderFree(XLogReaderState *state) pfree(state->errormsg_buf); if (state->readRecordBuf) pfree(state->readRecordBuf); - pfree(state->readBuf); pfree(state); } @@ -197,306 +208,590 @@ allocate_recordbuf(XLogReaderState *state, uint32 reclength) return true; } + /* - * Attempt to read an XLOG record. - * - * If RecPtr is valid, try to read a record at that position. Otherwise - * try to read a record just after the last one previously read. - * - * If the read_page callback fails to read the requested data, NULL is - * returned. The callback is expected to have reported the error; errormsg - * is set to NULL. - * - * If the reading fails for some other reason, NULL is also returned, and - * *errormsg is set to a string with details of the failure. - * - * The returned pointer (or *errormsg) points to an internal buffer that's - * valid until the next call to XLogReadRecord. + * Begin reading WAL at 'recptr'. 'recptr' should point to the beginnning of + * a valid WAL record. (Pointing at the beginning of a page is also OK, if there is + * a new record right after the page header, i.e. not a continuation) */ -XLogRecord * -XLogReadRecord(XLogReaderState *state, XLogRecPtr RecPtr, char **errormsg) +void +XLogBeginRead(XLogReaderState *state, XLogRecPtr recptr) { - XLogRecord *record; - XLogRecPtr targetPagePtr; - bool randAccess; - uint32 len, - total_len; - uint32 targetRecOff; - uint32 pageHeaderSize; - bool gotheader; - int readOff; + Assert(!XLogRecPtrIsInvalid(recptr)); + + ResetDecoder(state); + + /* Begin at the passed-in record pointer. */ + state->ReadRecPtr = recptr; + state->SkipRecPtr = InvalidXLogRecPtr; /* - * randAccess indicates whether to verify the previous-record pointer of - * the record we're reading. We only do this if we're reading - * sequentially, which is what we initially assume. + * We cannot verify the previous-record pointer when we're seeking to a + * particular record. Reset PrevRecPtr so that we won't try doing that. */ - randAccess = false; + state->PrevRecPtr = InvalidXLogRecPtr; - /* reset error state */ - *errormsg = NULL; - state->errormsg_buf[0] = '\0'; + /* also reset EndRecPtr to be tidy */ + state->EndRecPtr = InvalidXLogRecPtr; - ResetDecoder(state); + /* + * Reset verifiedPagePtr to force reading the page again. (Not sure if this + * is needed, but seems better to play it safe.) + */ + XLogReaderInvalReadState(state); - if (RecPtr == InvalidXLogRecPtr) - { - /* No explicit start point; read the record after the one we just read */ - RecPtr = state->EndRecPtr; + state->readRecordState = XLREAD_NEED_SEGMENT_HEADER; +} - if (state->ReadRecPtr == InvalidXLogRecPtr) - randAccess = true; +#ifdef FRONTEND +/* + * Functions that are currently not needed in the backend, but are better + * implemented inside xlogreader.c because of the internal facilities available + * here. + */ - /* - * 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 - { - /* - * Caller supplied a position to start at. - * - * In this case, the passed-in record pointer should already be - * pointing to a valid record starting position. - */ - Assert(XRecOffIsValid(RecPtr)); - randAccess = true; - } +/* + * Begin reading at the first record with an lsn >= recptr. + * + * Useful for checking whether recptr is a valid xlog address for reading, and + * to find the first valid address after some address when dumping records for + * debugging purposes. + */ +void +XLogFindNextRecord(XLogReaderState *state, XLogRecPtr recptr) +{ + Assert(!XLogRecPtrIsInvalid(recptr)); - state->currRecPtr = RecPtr; + ResetDecoder(state); - targetPagePtr = RecPtr - (RecPtr % XLOG_BLCKSZ); - targetRecOff = RecPtr % XLOG_BLCKSZ; + /* Begin at the passed-in record pointer. */ + state->ReadRecPtr = recptr; + state->SkipRecPtr = recptr; /* - * Read the page containing the record into state->readBuf. Request enough - * byte to cover the whole record header, or at least the part of it that - * fits on the same page. + * We cannot verify the previous-record pointer when we're seeking to a + * particular record. Reset PrevRecPtr so that we won't try doing that. */ - readOff = ReadPageInternal(state, - targetPagePtr, - Min(targetRecOff + SizeOfXLogRecord, XLOG_BLCKSZ)); - if (readOff < 0) - goto err; + state->PrevRecPtr = InvalidXLogRecPtr; + + /* also reset EndRecPtr to be tidy */ + state->EndRecPtr = InvalidXLogRecPtr; /* - * ReadPageInternal always returns at least the page header, so we can - * examine it now. + * Reset verifiedPagePtr to force reading the page again. (Not sure if this + * is needed, but seems better to play it safe.) */ - pageHeaderSize = XLogPageHeaderSize((XLogPageHeader) state->readBuf); - if (targetRecOff == 0) - { - /* - * At page start, so skip over page header. - */ - RecPtr += pageHeaderSize; - targetRecOff = pageHeaderSize; - } - else if (targetRecOff < pageHeaderSize) - { - report_invalid_record(state, "invalid record offset at %X/%X", - (uint32) (RecPtr >> 32), (uint32) RecPtr); - goto err; - } + XLogReaderInvalReadState(state); - if ((((XLogPageHeader) state->readBuf)->xlp_info & XLP_FIRST_IS_CONTRECORD) && - targetRecOff == pageHeaderSize) - { - report_invalid_record(state, "contrecord is requested by %X/%X", - (uint32) (RecPtr >> 32), (uint32) RecPtr); - goto err; - } + state->readRecordState = XLREAD_NEED_SEGMENT_HEADER; +} - /* ReadPageInternal has verified the page header */ - Assert(pageHeaderSize <= readOff); +#endif /* FRONTEND */ - /* - * 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; +/* + * Attempt to read an XLOG record. + * + * This function runs a state machine and may need to call several times until + * a record is read. + * + * XLogBeginRead() or XLogFindNextRecord() must be called before the first call + * to XLogReadRecord(). + * + * When a record is successfully read, returns XLREAD_SUCCESS with result + * record being stored in *record. + * + * Returns XLREAD_NEED_DATA if more data is needed to finish reading the current + * record. In that case, state->loadPagePtr and state->loadLen are set to inform + * the caller the WAL position and minimum length of data needed. The caller + * shall read in the requested data and set state->readBuf to point to a buffer + * containing it. The caller must also set state->readPagePtr, state->readPageTLI, + * and state->readLen to indicate the starting position of the read data (which + * must equal loadPagePtr), the timeline that it was read from, and the length + * of data that is now available (which must be >= loadLen), respectively. + * + * If invalid data is encountered, XLogReadRecord returns XLREAD_FAIL with *record + * being set to NULL. *errormsg is set to a string with details of the failure. + * + * The returned pointer (or *errormsg) points to an internal buffer that's + * valid until the next call to XLogReadRecord. + * + * Note: DO NOT non-local exit (ereport) from inside of this function. + */ + +XLogReadRecordResult +XLogReadRecord(XLogReaderState *state, XLogRecord **record, char **errormsg) +{ + /* reset error state */ + *errormsg = NULL; + state->errormsg_buf[0] = '\0'; /* - * 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 the caller provided us with new data, after a XLREAD_NEED_DATA result + * from previous call, verify the page header on the data that was provided. */ - if (targetRecOff <= XLOG_BLCKSZ - SizeOfXLogRecord) + if (state->loadPagePtr != InvalidXLogRecPtr) { - if (!ValidXLogRecordHeader(state, RecPtr, state->ReadRecPtr, record, - randAccess)) - goto err; - gotheader = true; - } - else - { - /* XXX: more validation should be done here */ - if (total_len < SizeOfXLogRecord) + XLogPageHeader hdr; + + if (state->readPagePtr != state->loadPagePtr || state->readLen < state->loadLen) { - report_invalid_record(state, - "invalid record length at %X/%X: wanted %u, got %u", - (uint32) (RecPtr >> 32), (uint32) RecPtr, - (uint32) SizeOfXLogRecord, total_len); + /* + * We asked for data, but the caller didn't provide it. + * + * XXX: Would it be better to return XLREAD_NEED_DATA, and let the + * caller retry? There's a good chance that it would lead to an infinite + * loop, though. + */ + report_invalid_record(state, "invalid XLogReader state"); goto err; } - gotheader = false; - } - len = XLOG_BLCKSZ - RecPtr % XLOG_BLCKSZ; - if (total_len > len) - { - /* Need to reassemble record */ - char *contdata; - XLogPageHeader pageHeader; - char *buffer; - uint32 gotlen; + /* We have enough data to check the header length. */ + hdr = (XLogPageHeader) state->readBuf; + + /* still not enough */ + if (state->readLen < XLogPageHeaderSize(hdr)) + { + state->verifiedPagePtr = InvalidXLogRecPtr; + state->verifiedPageLen = 0; + state->loadLen = XLogPageHeaderSize(hdr); + return XLREAD_NEED_DATA; + } /* - * Enlarge readRecordBuf as needed. + * Now that we know we have the full header, validate it. */ - if (total_len > state->readRecordBufSize && - !allocate_recordbuf(state, total_len)) + if (!XLogReaderValidatePageHeader(state, state->readPagePtr, (char *) state->readBuf)) { - /* We treat this as a "bogus data" condition */ - report_invalid_record(state, "record length %u at %X/%X too long", - total_len, - (uint32) (RecPtr >> 32), (uint32) RecPtr); + XLogReaderInvalReadState(state); goto err; } - /* 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; + state->verifiedPagePtr = state->readPagePtr; + state->verifiedPageLen = state->readLen; + state->loadPagePtr = InvalidXLogRecPtr; + state->loadLen = 0; - do - { - /* Calculate pointer to beginning of next page */ - targetPagePtr += XLOG_BLCKSZ; + /* we have the requested page (or part of it) in the buffer now. */ + } - /* Wait for the next page to become available */ - readOff = ReadPageInternal(state, targetPagePtr, - Min(total_len - gotlen + SizeOfXLogShortPHD, - XLOG_BLCKSZ)); +again: + switch (state->readRecordState) + { + /* + * NEED_SEGMENT_HEADER is the first state after a call to XLogBeginRead(). + * In this state, we need to read the first page on the WAL segment containing + * the first record to read. This is not needed when we continue to read + * after a previous record, because we read the WAL sequentially and will + * encounter the first page of each WAL segment in due course, anyway. + */ + case XLREAD_NEED_SEGMENT_HEADER: + { + XLogRecPtr seg_begin_ptr; - if (readOff < 0) - goto err; + /* Request the first page in the segment. */ + seg_begin_ptr = state->ReadRecPtr - state->ReadRecPtr % state->wal_segment_size; + if (XLogNeedPage(state, seg_begin_ptr)) + return XLREAD_NEED_DATA; - Assert(SizeOfXLogShortPHD <= readOff); + if (state->SkipRecPtr != InvalidXLogRecPtr) + state->readRecordState = XLREAD_SKIP_CONTRECORDS; + else + state->readRecordState = XLREAD_NEED_TOT_LEN; + goto again; + } - /* Check that the continuation on next page looks valid */ - pageHeader = (XLogPageHeader) state->readBuf; - if (!(pageHeader->xlp_info & XLP_FIRST_IS_CONTRECORD)) + /* + * SKIP_CONTRECORDS state is only entered if the reader was initialized with + * XLogFindNextRecord(). In this state, we read the page containing ReadRecPtr, + * and skip over any continuation data at the beginning of the page, until + * we find the first valid record. + */ + case XLREAD_SKIP_CONTRECORDS: { - report_invalid_record(state, - "there is no contrecord flag at %X/%X", - (uint32) (RecPtr >> 32), (uint32) RecPtr); - goto err; + XLogRecPtr targetPagePtr; + uint32 pageHeaderSize; + XLogPageHeader hdr; + + /* + * Read the page containing the record into state->readBuf. + */ + targetPagePtr = state->ReadRecPtr - (state->ReadRecPtr % XLOG_BLCKSZ); + + if (XLogNeedPage(state, targetPagePtr)) + return XLREAD_NEED_DATA; + + /* + * We have loaded at least the page header, so we can examine it now. + * (XLogNeedPage() ensures that) + */ + hdr = (XLogPageHeader) state->readBuf; + pageHeaderSize = XLogPageHeaderSize(hdr); + /* XLogNeedPage has verified the page header */ + Assert(pageHeaderSize <= state->readLen); + + /* + * skip over potential continuation data. + */ + if (((XLogPageHeader) state->readBuf)->xlp_info & XLP_FIRST_IS_CONTRECORD) + { + /* + * If the length of the remaining continuation data is more than + * what can fit in this page, the continuation record crosses over + * this page. Read the next page and try again. xlp_rem_len in the + * next page header will contain the remaining length of the + * continuation data + * + * Note that record headers are MAXALIGN'ed + */ + if (MAXALIGN(hdr->xlp_rem_len) > (XLOG_BLCKSZ - pageHeaderSize)) + { + state->ReadRecPtr = targetPagePtr + XLOG_BLCKSZ; + /* stay in XLREAD_SKIP_CONTRECORDS state */ + } + else + { + /* + * The previous continuation record ends in this page. Set + * ReadRecPtr to point to the first valid record + */ + state->ReadRecPtr = targetPagePtr + pageHeaderSize + + MAXALIGN(hdr->xlp_rem_len); + state->readRecordState = XLREAD_NEED_TOT_LEN; + } + } + else + state->readRecordState = XLREAD_NEED_TOT_LEN; + goto again; } /* - * Cross-check that xlp_rem_len agrees with how much of the record - * we expect there to be left. + * Read next record, after the previous one. + * + * This state is entered when XLogReadReord() has returned a valid record. + * Initialize the state for reading the next record. + */ + case XLREAD_NEXT_RECORD: + ResetDecoder(state); + + /* + * EndRecPtr 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 on the next page, but we can't do that until we've + * read in the page, since the header size is variable. + */ + state->PrevRecPtr = state->ReadRecPtr; + state->ReadRecPtr = state->EndRecPtr; + + state->readRecordState = XLREAD_NEED_TOT_LEN; + /* fall through */ + + /* + * Start reading the record at ReadRecPtr. + * + * When entering this state, state->ReadRecPtr points to the record + * we want to read, and PrevRecPtr to the previous record, if + * known. + * + * We will read as much as we can from the first page containing + * ReadRecPtr, and continue in XLREAD_NEED_CONTINUATION state if + * needed. */ - if (pageHeader->xlp_rem_len == 0 || - total_len != (pageHeader->xlp_rem_len + gotlen)) + case XLREAD_NEED_TOT_LEN: { - report_invalid_record(state, - "invalid contrecord length %u at %X/%X", - pageHeader->xlp_rem_len, - (uint32) (RecPtr >> 32), (uint32) RecPtr); - goto err; - } + uint32 total_len; + uint32 pageHeaderSize; + XLogRecPtr targetPagePtr; + uint32 targetRecOff; + uint32 pageRemain; + uint32 len; + XLogRecord *rechdr; + + /* + * Read the page containing the record into state->readBuf. + */ + targetPagePtr = state->ReadRecPtr - (state->ReadRecPtr % XLOG_BLCKSZ); + targetRecOff = state->ReadRecPtr % XLOG_BLCKSZ; + + if (XLogNeedPage(state, targetPagePtr)) + return XLREAD_NEED_DATA; + + /* + * We have loaded at least the page header, so we can examine it now. + * (XLogNeedPage() ensures that) + */ + pageHeaderSize = XLogPageHeaderSize((XLogPageHeader) state->readBuf); + /* XLogNeedPage has verified the page header */ + Assert(pageHeaderSize <= state->readLen); + + /* + * If the previous record ended at page boundary, then we must skip over the + * page header to get to the beginning of the next record. + */ + if (targetRecOff == 0) + { + state->ReadRecPtr += pageHeaderSize; + targetRecOff = state->ReadRecPtr % XLOG_BLCKSZ; + } + + /* + * sanity check that the starting position doesn't point to + * the middle of the page header. + */ + if (targetRecOff < pageHeaderSize) + { + report_invalid_record(state, "invalid record offset at %X/%X", + (uint32) (state->ReadRecPtr >> 32), (uint32) state->ReadRecPtr); + goto err; + } - /* Append the continuation from this page to the buffer */ - pageHeaderSize = XLogPageHeaderSize(pageHeader); + /* + * If we're reading the first record on the page, then the CONTRECORD flag should not + * be set. + */ + if ((((XLogPageHeader) state->readBuf)->xlp_info & XLP_FIRST_IS_CONTRECORD) && + targetRecOff == pageHeaderSize) + { + report_invalid_record(state, "contrecord is requested by %X/%X", + (uint32) (state->ReadRecPtr >> 32), (uint32) state->ReadRecPtr); + goto err; + } - if (readOff < pageHeaderSize) - readOff = ReadPageInternal(state, targetPagePtr, - pageHeaderSize); + /* + * The page header looks OK, and targetRecOff points to the + * true beginning of the record now. Read the record length, + * which is the first field in the record header. The record + * must have at least the header, so request SizeOfXLogRecord + * or as much of it as fits on this page. + */ + if (XLogNeedData(state, targetPagePtr, + Min(targetRecOff + SizeOfXLogRecord, XLOG_BLCKSZ))) + return XLREAD_NEED_DATA; - Assert(pageHeaderSize <= readOff); + /* + * 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. + */ + rechdr = (XLogRecord *) (state->readBuf + targetRecOff); + total_len = rechdr->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(state, state->ReadRecPtr, state->PrevRecPtr, rechdr)) + goto err; + } + else + { + /* XXX: more validation should be done here */ + if (total_len < SizeOfXLogRecord) + { + report_invalid_record(state, + "invalid record length at %X/%X: wanted %u, got %u", + (uint32) (state->ReadRecPtr >> 32), (uint32) state->ReadRecPtr, + (uint32) SizeOfXLogRecord, total_len); + goto err; + } + } + + /* + * Wait for the rest of the record, or the part of the record that fit on the + * first page if crossed a page boundary, to become available. + */ + state->recordGotLen = 0; + state->recordRemainLen = total_len; - contdata = (char *) state->readBuf + pageHeaderSize; - len = XLOG_BLCKSZ - pageHeaderSize; - if (pageHeader->xlp_rem_len < len) - len = pageHeader->xlp_rem_len; + /* Wait for the rest of the record on the first page to become available */ + pageRemain = XLOG_BLCKSZ - state->ReadRecPtr % XLOG_BLCKSZ; + len = Min(pageRemain, total_len); - if (readOff < pageHeaderSize + len) - readOff = ReadPageInternal(state, targetPagePtr, - pageHeaderSize + len); + if (XLogNeedData(state, targetPagePtr, targetRecOff + len)) + return XLREAD_NEED_DATA; + rechdr = (XLogRecord *) (state->readBuf + targetRecOff); - memcpy(buffer, (char *) contdata, len); - buffer += len; - gotlen += len; + if (state->recordRemainLen <= len) + { + /* Record does not cross a page boundary */ + if (!ValidXLogRecord(state, rechdr, state->ReadRecPtr)) + goto err; + /* We already checked the record header earlier */ + state->EndRecPtr = state->ReadRecPtr + MAXALIGN(total_len); + state->readRecordState = XLREAD_NEXT_RECORD; + *record = rechdr; + break; + } + else + { + /* + * The record continues on the next page. Need to reassemble it. + */ + /* Enlarge readRecordBuf as needed. */ + if (total_len > state->readRecordBufSize && + !allocate_recordbuf(state, total_len)) + { + /* We treat this as a "bogus data" condition */ + report_invalid_record(state, "record length %u at %X/%X too long", + total_len, + (uint32) (state->ReadRecPtr >> 32), (uint32) state->ReadRecPtr); + goto err; + } + + /* Copy the first fragment of the record from the first page. */ + memcpy(state->readRecordBuf, + state->readBuf + state->ReadRecPtr % XLOG_BLCKSZ, len); + state->recordGotLen += len; + state->recordRemainLen -= len; + + /* Calculate pointer to beginning of next page */ + state->recordContRecPtr = state->ReadRecPtr + len; + Assert(state->recordContRecPtr % XLOG_BLCKSZ == 0); + + state->readRecordState = XLREAD_NEED_CONTINUATION; + } + } + /* fall through */ - /* If we just reassembled the record header, validate it. */ - if (!gotheader) + case XLREAD_NEED_CONTINUATION: { - record = (XLogRecord *) state->readRecordBuf; - if (!ValidXLogRecordHeader(state, RecPtr, state->ReadRecPtr, - record, randAccess)) + XLogRecPtr targetPagePtr; + char *contdata; + XLogPageHeader pageHeader; + uint32 pageHeaderSize; + uint32 len; + + /* Wait for the next page to become available */ + targetPagePtr = state->recordContRecPtr; + if (XLogNeedPage(state, targetPagePtr)) + return XLREAD_NEED_DATA; + + /* Check that the continuation on next page looks valid */ + pageHeader = (XLogPageHeader) state->readBuf; + if (!(pageHeader->xlp_info & XLP_FIRST_IS_CONTRECORD)) + { + /* XXX: should this report recordContRecPtr rather than beginning of the + * record? */ + report_invalid_record(state, + "there is no contrecord flag at %X/%X", + (uint32) (state->ReadRecPtr >> 32), (uint32) state->ReadRecPtr); goto err; - gotheader = true; - } - } while (gotlen < total_len); + } - Assert(gotheader); + /* + * 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 || + pageHeader->xlp_rem_len != state->recordRemainLen) + { + /* XXX: should this report recordContRecPtr rather than beginning of the + * record? */ + report_invalid_record(state, + "invalid contrecord length %u at %X/%X", + pageHeader->xlp_rem_len, + (uint32) (state->ReadRecPtr >> 32), (uint32) state->ReadRecPtr); + goto err; + } - record = (XLogRecord *) state->readRecordBuf; - if (!ValidXLogRecord(state, record, RecPtr)) - goto err; + /* Append the continuation from this page to the buffer */ + pageHeaderSize = XLogPageHeaderSize(pageHeader); + /* XLogNeedData should have ensured that the whole page header was read */ + Assert(pageHeaderSize <= state->readLen); + Assert(state->readLen >= SizeOfXLogShortPHD); - pageHeaderSize = XLogPageHeaderSize((XLogPageHeader) state->readBuf); - state->ReadRecPtr = RecPtr; - state->EndRecPtr = targetPagePtr + pageHeaderSize - + MAXALIGN(pageHeader->xlp_rem_len); - } - else - { - /* Wait for the record data to become available */ - readOff = ReadPageInternal(state, targetPagePtr, - Min(targetRecOff + total_len, XLOG_BLCKSZ)); - if (readOff < 0) - goto err; + /* Ok, read the rest of the record */ + /* + * Read the rest of the page containing the record, if we + * didn't get it already. (If we didn't get it yet, we'll + * read the page header again on next invocation. In + * practice, this should happen very rarely, assuming that + * the caller makes the whole page available to us even + * when we request just a part of it. XXX: This would also + * become less ugly, if we passed a flag to XLogNeedData() + * on the first call to tell it that the remaining length + * includes the page header) + */ + if (XLogNeedData(state, targetPagePtr, + Min(pageHeaderSize + state->recordRemainLen, + XLOG_BLCKSZ))) + return XLREAD_NEED_DATA; + + contdata = (char *) state->readBuf + pageHeaderSize; + len = XLOG_BLCKSZ - pageHeaderSize; + if (pageHeader->xlp_rem_len < len) + len = pageHeader->xlp_rem_len; + + memcpy(state->readRecordBuf + state->recordGotLen, + (char *) contdata, len); + state->recordGotLen += len; + state->recordRemainLen -= len; + + /* If we just reassembled the record header, validate it. */ + if (state->recordGotLen >= sizeof(XLogRecord) && + state->recordGotLen - len < sizeof(XLogRecord)) + { + *record = (XLogRecord *) state->readRecordBuf; + if (!ValidXLogRecordHeader(state, state->ReadRecPtr, state->PrevRecPtr, + *record)) + goto err; + } - /* Record does not cross a page boundary */ - if (!ValidXLogRecord(state, record, RecPtr)) - goto err; + if (state->recordRemainLen > 0) + { + /* Calculate pointer to beginning of next page, and continue */ + state->recordContRecPtr = targetPagePtr + XLOG_BLCKSZ; + goto again; + } + else + { + *record = (XLogRecord *) state->readRecordBuf; + if (!ValidXLogRecord(state, *record, state->ReadRecPtr)) + goto err; - state->EndRecPtr = RecPtr + MAXALIGN(total_len); + pageHeaderSize = XLogPageHeaderSize((XLogPageHeader) state->readBuf); + state->EndRecPtr = targetPagePtr + pageHeaderSize + + MAXALIGN(pageHeader->xlp_rem_len); - state->ReadRecPtr = RecPtr; + state->readRecordState = XLREAD_NEXT_RECORD; + break; + } + } } /* * Special processing if it's an XLOG SWITCH record */ - if (record->xl_rmid == RM_XLOG_ID && - (record->xl_info & ~XLR_INFO_MASK) == XLOG_SWITCH) + if ((*record)->xl_rmid == RM_XLOG_ID && + ((*record)->xl_info & ~XLR_INFO_MASK) == XLOG_SWITCH) { /* Pretend it extends to end of segment */ state->EndRecPtr += state->wal_segment_size - 1; state->EndRecPtr -= XLogSegmentOffset(state->EndRecPtr, state->wal_segment_size); } - if (DecodeXLogRecord(state, record, errormsg)) - return record; - else - return NULL; + state->loadPagePtr = InvalidXLogRecPtr; + state->loadLen = 0; + + /* all done. But before we claim victory, see if we were asked to skip over this record */ + if (state->SkipRecPtr > state->ReadRecPtr) + goto again; + + if (DecodeXLogRecord(state, *record, errormsg)) + return XLREAD_SUCCESS; + + *record = NULL; + return XLREAD_FAIL; err: @@ -509,124 +804,83 @@ err: if (state->errormsg_buf[0] != '\0') *errormsg = state->errormsg_buf; - return NULL; + *record = NULL; + return XLREAD_FAIL; } /* - * Read a single xlog page including at least [pageptr, reqLen] of valid data - * via the read_page() callback. + * Check that the current buffer (state->verifiedPagePtr) contains the page + * starting at 'pageptr'. * - * Returns -1 if the required page cannot be read for some reason; errormsg_buf - * is set in that case (unless the error occurs in the read_page callback). + * If the correct page is loaded, returns false. * - * We fetch the page from a reader-local cache if we know we have the required - * data and if there hasn't been any error since caching the data. + * If the required page is not yet loaded, returns false, and sets loadPagePtr + * and loadLen to request the caller to provide the data. The caller of + * XLogReadRecord() should load the region to state->readBuf and call it again. */ -static int -ReadPageInternal(XLogReaderState *state, XLogRecPtr pageptr, int reqLen) +static bool +XLogNeedPage(XLogReaderState *state, XLogRecPtr pageptr) { - int readLen; - uint32 targetPageOff; - XLogSegNo targetSegNo; - XLogPageHeader hdr; - - Assert((pageptr % XLOG_BLCKSZ) == 0); - - XLByteToSeg(pageptr, targetSegNo, state->wal_segment_size); - targetPageOff = XLogSegmentOffset(pageptr, state->wal_segment_size); - - /* check whether we have all the requested data already */ - if (targetSegNo == state->readSegNo && targetPageOff == state->readOff && - reqLen <= state->readLen) - return state->readLen; - - /* - * Data is not in our buffer. - * - * Every time we actually read the page, even if we looked at parts of it - * before, we need to do verification as the read_page callback might now - * be rereading data from a different source. - * - * Whenever switching to a new WAL segment, we read the first page of the - * file and validate its header, even if that's not where the target - * record is. This is so that we can check the additional identification - * info that is present in the first page's "long" header. - */ - if (targetSegNo != state->readSegNo && targetPageOff != 0) - { - XLogRecPtr targetSegmentPtr = pageptr - targetPageOff; - - readLen = state->read_page(state, targetSegmentPtr, XLOG_BLCKSZ, - state->currRecPtr, - state->readBuf, &state->readPageTLI); - if (readLen < 0) - goto err; - - /* we can be sure to have enough WAL available, we scrolled back */ - Assert(readLen == XLOG_BLCKSZ); - - if (!XLogReaderValidatePageHeader(state, targetSegmentPtr, - state->readBuf)) - goto err; - } - - /* - * First, read the requested data length, but at least a short page header - * so that we can validate it. - */ - readLen = state->read_page(state, pageptr, Max(reqLen, SizeOfXLogShortPHD), - state->currRecPtr, - state->readBuf, &state->readPageTLI); - if (readLen < 0) - goto err; - - Assert(readLen <= XLOG_BLCKSZ); - - /* Do we have enough data to check the header length? */ - if (readLen <= SizeOfXLogShortPHD) - goto err; - - Assert(readLen >= reqLen); - - hdr = (XLogPageHeader) state->readBuf; - - /* still not enough */ - if (readLen < XLogPageHeaderSize(hdr)) + /* check whether we have the requested page in the buffer already */ + if (state->verifiedPagePtr == pageptr && state->verifiedPageLen >= SizeOfXLogShortPHD) + return false; + else { - readLen = state->read_page(state, pageptr, XLogPageHeaderSize(hdr), - state->currRecPtr, - state->readBuf, &state->readPageTLI); - if (readLen < 0) - goto err; + /* request the page header */ + /* + * First, read the requested data length, but at least a short page header + * so that we can validate it. + */ + state->verifiedPagePtr = InvalidXLogRecPtr; + state->verifiedPageLen = 0; + state->loadPagePtr = pageptr; + state->loadLen = SizeOfXLogShortPHD; + return true; } +} - /* - * Now that we know we have the full header, validate it. - */ - if (!XLogReaderValidatePageHeader(state, pageptr, (char *) hdr)) - goto err; +/* + * Checks if the page current loaded into 'readBuf' contains 'reqLen' bytes. + * + * Returns false if there is enough data. Otherwise, returns true, and sets + * loadPagePtr and loadLen to request more. + * + * Note: This function assumes that you have already called XLogNeedPage() + * first, to ensure that the correct page has been loaded into 'readBuf'. + */ +static bool +XLogNeedData(XLogReaderState *state, XLogRecPtr pageptr, int reqLen) +{ + /* This should only be called after XLogNeedPage has verified the page header */ + Assert(state->readPagePtr == pageptr); + Assert(state->readLen == state->verifiedPageLen); + Assert(state->verifiedPagePtr == pageptr); - /* update read state information */ - state->readSegNo = targetSegNo; - state->readOff = targetPageOff; - state->readLen = readLen; + Assert(reqLen > XLogPageHeaderSize((XLogPageHeader) state->readBuf)); + Assert(reqLen <= XLOG_BLCKSZ); - return readLen; + /* check whether we have all the requested data already */ + if (reqLen <= state->verifiedPageLen) + return false; -err: - XLogReaderInvalReadState(state); - return -1; + /* force the page headers to be re-verified */ + state->verifiedPagePtr = InvalidXLogRecPtr; + state->verifiedPageLen = 0; + state->loadPagePtr = pageptr; + state->loadLen = reqLen; + return true; } /* * Invalidate the xlogreader's read state to force a re-read. */ -void +static void XLogReaderInvalReadState(XLogReaderState *state) { - state->readSegNo = 0; - state->readOff = 0; - state->readLen = 0; + state->verifiedPagePtr = InvalidXLogRecPtr; + state->verifiedPageLen = 0; + state->loadPagePtr = InvalidXLogRecPtr; + state->loadLen = 0; } /* @@ -634,11 +888,12 @@ XLogReaderInvalReadState(XLogReaderState *state) * * This is just a convenience subroutine to avoid duplicated code in * XLogReadRecord. It's not intended for use from anywhere else. + * + * If PrevRecPtr is valid, the xl_prev is is cross-checked with it. */ static bool ValidXLogRecordHeader(XLogReaderState *state, XLogRecPtr RecPtr, - XLogRecPtr PrevRecPtr, XLogRecord *record, - bool randAccess) + XLogRecPtr PrevRecPtr, XLogRecord *record) { if (record->xl_tot_len < SizeOfXLogRecord) { @@ -656,7 +911,7 @@ ValidXLogRecordHeader(XLogReaderState *state, XLogRecPtr RecPtr, (uint32) RecPtr); return false; } - if (randAccess) + if (PrevRecPtr == InvalidXLogRecPtr) { /* * We can't exactly verify the prev-link, but surely it should be less @@ -869,134 +1124,6 @@ XLogReaderValidatePageHeader(XLogReaderState *state, XLogRecPtr recptr, return true; } -#ifdef FRONTEND -/* - * Functions that are currently not needed in the backend, but are better - * implemented inside xlogreader.c because of the internal facilities available - * here. - */ - -/* - * Find the first record with an lsn >= RecPtr. - * - * Useful for checking whether RecPtr is a valid xlog address for reading, and - * to find the first valid address after some address when dumping records for - * debugging purposes. - */ -XLogRecPtr -XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr) -{ - XLogReaderState saved_state = *state; - XLogRecPtr tmpRecPtr; - XLogRecPtr found = InvalidXLogRecPtr; - XLogPageHeader header; - char *errormsg; - - Assert(!XLogRecPtrIsInvalid(RecPtr)); - - /* - * skip over potential continuation data, keeping in mind that it may span - * multiple pages - */ - tmpRecPtr = RecPtr; - while (true) - { - XLogRecPtr targetPagePtr; - int targetRecOff; - uint32 pageHeaderSize; - int readLen; - - /* - * Compute targetRecOff. It should typically be equal or greater than - * short page-header since a valid record can't start anywhere before - * that, except when caller has explicitly specified the offset that - * falls somewhere there or when we are skipping multi-page - * continuation record. It doesn't matter though because - * ReadPageInternal() is prepared to handle that and will read at - * least short page-header worth of data - */ - targetRecOff = tmpRecPtr % XLOG_BLCKSZ; - - /* scroll back to page boundary */ - targetPagePtr = tmpRecPtr - targetRecOff; - - /* Read the page containing the record */ - readLen = ReadPageInternal(state, targetPagePtr, targetRecOff); - if (readLen < 0) - goto err; - - header = (XLogPageHeader) state->readBuf; - - pageHeaderSize = XLogPageHeaderSize(header); - - /* make sure we have enough data for the page header */ - readLen = ReadPageInternal(state, targetPagePtr, pageHeaderSize); - if (readLen < 0) - goto err; - - /* skip over potential continuation data */ - if (header->xlp_info & XLP_FIRST_IS_CONTRECORD) - { - /* - * If the length of the remaining continuation data is more than - * what can fit in this page, the continuation record crosses over - * this page. Read the next page and try again. xlp_rem_len in the - * next page header will contain the remaining length of the - * continuation data - * - * Note that record headers are MAXALIGN'ed - */ - if (MAXALIGN(header->xlp_rem_len) > (XLOG_BLCKSZ - pageHeaderSize)) - tmpRecPtr = targetPagePtr + XLOG_BLCKSZ; - else - { - /* - * The previous continuation record ends in this page. Set - * tmpRecPtr to point to the first valid record - */ - tmpRecPtr = targetPagePtr + pageHeaderSize - + MAXALIGN(header->xlp_rem_len); - break; - } - } - else - { - tmpRecPtr = targetPagePtr + pageHeaderSize; - break; - } - } - - /* - * we know now that tmpRecPtr is an address pointing to a valid XLogRecord - * because either we're at the first record after the beginning of a page - * or we just jumped over the remaining data of a continuation. - */ - while (XLogReadRecord(state, tmpRecPtr, &errormsg) != NULL) - { - /* continue after the record */ - tmpRecPtr = InvalidXLogRecPtr; - - /* past the record we've found, break out */ - if (RecPtr <= state->ReadRecPtr) - { - found = state->ReadRecPtr; - goto out; - } - } - -err: -out: - /* Reset state to what we had before finding the record */ - state->ReadRecPtr = saved_state.ReadRecPtr; - state->EndRecPtr = saved_state.EndRecPtr; - XLogReaderInvalReadState(state); - - return found; -} - -#endif /* FRONTEND */ - - /* ---------------------------------------- * Functions for decoding the data and block references in a record. * ---------------------------------------- diff --git a/src/backend/access/transam/xlogutils.c b/src/backend/access/transam/xlogutils.c index 1fc39333f15..99d407f123b 100644 --- a/src/backend/access/transam/xlogutils.c +++ b/src/backend/access/transam/xlogutils.c @@ -802,8 +802,7 @@ XLogRead(char *buf, int segsize, TimeLineID tli, XLogRecPtr startptr, void XLogReadDetermineTimeline(XLogReaderState *state, XLogRecPtr wantPage, uint32 wantLength) { - const XLogRecPtr lastReadPage = state->readSegNo * - state->wal_segment_size + state->readOff; + const XLogRecPtr lastReadPage = state->readPagePtr; Assert(wantPage != InvalidXLogRecPtr && wantPage % XLOG_BLCKSZ == 0); Assert(wantLength <= XLOG_BLCKSZ); @@ -907,11 +906,13 @@ XLogReadDetermineTimeline(XLogReaderState *state, XLogRecPtr wantPage, uint32 wa * exists for normal backends, so we have to do a check/sleep/repeat style of * loop for now. */ -int -read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, - int reqLen, XLogRecPtr targetRecPtr, char *cur_page, - TimeLineID *pageTLI) +bool +read_local_xlog_page(XLogReaderState *state) { + XLogRecPtr targetPagePtr = state->loadPagePtr; + int reqLen = state->loadLen; + char *cur_page = state->readBuf; + TimeLineID *pageTLI = &state->readPageTLI; XLogRecPtr read_upto, loc; int count; @@ -1009,7 +1010,9 @@ read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, else if (targetPagePtr + reqLen > read_upto) { /* not enough data there */ - return -1; + state->readPagePtr = InvalidXLogRecPtr; + state->readLen = -1; + return false; } else { @@ -1026,5 +1029,7 @@ read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, XLOG_BLCKSZ); /* number of valid bytes in the buffer */ - return count; + state->readPagePtr = targetPagePtr; + state->readLen = count; + return true; } diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c index f8b9020081e..bff427395a1 100644 --- a/src/backend/replication/logical/logical.c +++ b/src/backend/replication/logical/logical.c @@ -124,7 +124,7 @@ StartupDecodingContext(List *output_plugin_options, TransactionId xmin_horizon, bool need_full_snapshot, bool fast_forward, - XLogPageReadCB read_page, + LogicalDecodingXLogReadPageCB read_page, LogicalOutputPluginWriterPrepareWrite prepare_write, LogicalOutputPluginWriterWrite do_write, LogicalOutputPluginWriterUpdateProgress update_progress) @@ -173,11 +173,13 @@ StartupDecodingContext(List *output_plugin_options, ctx->slot = slot; - ctx->reader = XLogReaderAllocate(wal_segment_size, read_page, ctx); + ctx->reader = XLogReaderAllocate(wal_segment_size); if (!ctx->reader) ereport(ERROR, (errcode(ERRCODE_OUT_OF_MEMORY), errmsg("out of memory"))); + ctx->reader->readBuf = palloc(XLOG_BLCKSZ); + ctx->read_page = read_page; ctx->reorder = ReorderBufferAllocate(); ctx->snapshot_builder = @@ -232,7 +234,7 @@ CreateInitDecodingContext(char *plugin, List *output_plugin_options, bool need_full_snapshot, XLogRecPtr restart_lsn, - XLogPageReadCB read_page, + LogicalDecodingXLogReadPageCB read_page, LogicalOutputPluginWriterPrepareWrite prepare_write, LogicalOutputPluginWriterWrite do_write, LogicalOutputPluginWriterUpdateProgress update_progress) @@ -374,7 +376,7 @@ LogicalDecodingContext * CreateDecodingContext(XLogRecPtr start_lsn, List *output_plugin_options, bool fast_forward, - XLogPageReadCB read_page, + LogicalDecodingXLogReadPageCB read_page, LogicalOutputPluginWriterPrepareWrite prepare_write, LogicalOutputPluginWriterWrite do_write, LogicalOutputPluginWriterUpdateProgress update_progress) @@ -465,11 +467,10 @@ DecodingContextReady(LogicalDecodingContext *ctx) void DecodingContextFindStartpoint(LogicalDecodingContext *ctx) { - XLogRecPtr startptr; ReplicationSlot *slot = ctx->slot; /* Initialize from where to start reading WAL. */ - startptr = slot->data.restart_lsn; + XLogBeginRead(ctx->reader, slot->data.restart_lsn); elog(DEBUG1, "searching for logical decoding starting point, starting at %X/%X", (uint32) (slot->data.restart_lsn >> 32), @@ -482,14 +483,21 @@ DecodingContextFindStartpoint(LogicalDecodingContext *ctx) char *err = NULL; /* the read_page callback waits for new WAL */ - record = XLogReadRecord(ctx->reader, startptr, &err); + while (XLogReadRecord(ctx->reader, &record, &err) == + XLREAD_NEED_DATA) + { + if (!ctx->read_page(ctx)) + { + record = NULL; + break; + } + } + if (err) elog(ERROR, "%s", err); if (!record) elog(ERROR, "no record found"); /* shouldn't happen */ - startptr = InvalidXLogRecPtr; - LogicalDecodingProcessRecord(ctx, ctx->reader); /* only continue till we found a consistent spot */ @@ -516,6 +524,7 @@ FreeDecodingContext(LogicalDecodingContext *ctx) ReorderBufferFree(ctx->reorder); FreeSnapshotBuilder(ctx->snapshot_builder); + pfree(ctx->reader->readBuf); XLogReaderFree(ctx->reader); MemoryContextDelete(ctx->context); } diff --git a/src/backend/replication/logical/logicalfuncs.c b/src/backend/replication/logical/logicalfuncs.c index d974400d6ef..fccae9a26ed 100644 --- a/src/backend/replication/logical/logicalfuncs.c +++ b/src/backend/replication/logical/logicalfuncs.c @@ -114,12 +114,10 @@ check_permissions(void) (errmsg("must be superuser or replication role to use replication slots")))); } -int -logical_read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, - int reqLen, XLogRecPtr targetRecPtr, char *cur_page, TimeLineID *pageTLI) +bool +logical_read_local_xlog_page(LogicalDecodingContext *ctx) { - return read_local_xlog_page(state, targetPagePtr, reqLen, - targetRecPtr, cur_page, pageTLI); + return read_local_xlog_page(ctx->reader); } /* @@ -135,7 +133,7 @@ pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool bin MemoryContext per_query_ctx; MemoryContext oldcontext; XLogRecPtr end_of_wal; - XLogRecPtr startptr; + XLogRecPtr nextptr; LogicalDecodingContext *ctx; ResourceOwner old_resowner = CurrentResourceOwner; ArrayType *arr; @@ -277,28 +275,31 @@ pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool bin * xacts that committed after the slot's confirmed_flush can be * accumulated into reorder buffers. */ - startptr = MyReplicationSlot->data.restart_lsn; + nextptr = MyReplicationSlot->data.restart_lsn; + XLogBeginRead(ctx->reader, nextptr); /* invalidate non-timetravel entries */ InvalidateSystemCaches(); /* Decode until we run out of records */ - while ((startptr != InvalidXLogRecPtr && startptr < end_of_wal) || - (ctx->reader->EndRecPtr != InvalidXLogRecPtr && ctx->reader->EndRecPtr < end_of_wal)) + while (nextptr < end_of_wal) { XLogRecord *record; char *errm = NULL; - record = XLogReadRecord(ctx->reader, startptr, &errm); + while (XLogReadRecord(ctx->reader, &record, &errm) == + XLREAD_NEED_DATA) + { + if (!ctx->read_page(ctx)) + { + record = NULL; + break; + } + } + if (errm) elog(ERROR, "%s", errm); - /* - * Now that we've set up the xlog reader state, subsequent calls - * pass InvalidXLogRecPtr to say "continue from last record" - */ - startptr = InvalidXLogRecPtr; - /* * The {begin_txn,change,commit_txn}_wrapper callbacks above will * store the description into our tuplestore. @@ -313,6 +314,9 @@ pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool bin if (upto_nchanges != 0 && upto_nchanges <= p->returned_rows) break; + + nextptr = ctx->reader->EndRecPtr; + CHECK_FOR_INTERRUPTS(); } diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c index 808a6f5b836..0f40215fc12 100644 --- a/src/backend/replication/slotfuncs.c +++ b/src/backend/replication/slotfuncs.c @@ -392,7 +392,7 @@ pg_logical_replication_slot_advance(XLogRecPtr moveto) { LogicalDecodingContext *ctx; ResourceOwner old_resowner = CurrentResourceOwner; - XLogRecPtr startlsn; + XLogRecPtr nextlsn; XLogRecPtr retlsn; PG_TRY(); @@ -412,7 +412,8 @@ pg_logical_replication_slot_advance(XLogRecPtr moveto) * Start reading at the slot's restart_lsn, which we know to point to * a valid record. */ - startlsn = MyReplicationSlot->data.restart_lsn; + nextlsn = MyReplicationSlot->data.restart_lsn; + XLogBeginRead(ctx->reader, nextlsn); /* Initialize our return value in case we don't do anything */ retlsn = MyReplicationSlot->data.confirmed_flush; @@ -421,10 +422,7 @@ pg_logical_replication_slot_advance(XLogRecPtr moveto) InvalidateSystemCaches(); /* Decode at least one record, until we run out of records */ - while ((!XLogRecPtrIsInvalid(startlsn) && - startlsn < moveto) || - (!XLogRecPtrIsInvalid(ctx->reader->EndRecPtr) && - ctx->reader->EndRecPtr < moveto)) + while (nextlsn < moveto) { char *errm = NULL; XLogRecord *record; @@ -433,13 +431,19 @@ pg_logical_replication_slot_advance(XLogRecPtr moveto) * Read records. No changes are generated in fast_forward mode, * but snapbuilder/slot statuses are updated properly. */ - record = XLogReadRecord(ctx->reader, startlsn, &errm); + while (XLogReadRecord(ctx->reader, &record, &errm) == + XLREAD_NEED_DATA) + { + if (!ctx->read_page(ctx)) + { + record = NULL; + break; + } + } + if (errm) elog(ERROR, "%s", errm); - /* Read sequentially from now on */ - startlsn = InvalidXLogRecPtr; - /* * Process the record. Storage-level changes are ignored in * fast_forward mode, but other modules (such as snapbuilder) @@ -452,6 +456,7 @@ pg_logical_replication_slot_advance(XLogRecPtr moveto) if (moveto <= ctx->reader->EndRecPtr) break; + nextlsn = ctx->reader->EndRecPtr; CHECK_FOR_INTERRUPTS(); } diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index 23870a25a56..8141d938ed2 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -761,10 +761,14 @@ StartReplication(StartReplicationCmd *cmd) * which has to do a plain sleep/busy loop, because the walsender's latch gets * set every time WAL is flushed. */ -static int -logical_read_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen, - XLogRecPtr targetRecPtr, char *cur_page, TimeLineID *pageTLI) +static bool +logical_read_xlog_page(LogicalDecodingContext *ctx) { + XLogReaderState *state = ctx->reader; + XLogRecPtr targetPagePtr = state->loadPagePtr; + int reqLen = state->loadLen; + char *cur_page = state->readBuf; + XLogRecPtr flushptr; int count; @@ -779,7 +783,11 @@ logical_read_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int req /* fail if not (implies we are going to shut down) */ if (flushptr < targetPagePtr + reqLen) - return -1; + { + state->readPagePtr = InvalidXLogRecPtr; + state->readLen = -1; + return false; + } if (targetPagePtr + XLOG_BLCKSZ <= flushptr) count = XLOG_BLCKSZ; /* more than one block available */ @@ -789,7 +797,9 @@ logical_read_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int req /* now actually read the data, we know it's there */ XLogRead(cur_page, targetPagePtr, XLOG_BLCKSZ); - return count; + state->readPagePtr = targetPagePtr; + state->readLen = count; + return true; } /* @@ -2823,8 +2833,21 @@ XLogSendLogical(void) */ WalSndCaughtUp = false; - record = XLogReadRecord(logical_decoding_ctx->reader, logical_startptr, &errm); - logical_startptr = InvalidXLogRecPtr; + if (logical_startptr != InvalidXLogRecPtr) + { + XLogBeginRead(logical_decoding_ctx->reader, logical_startptr); + logical_startptr = InvalidXLogRecPtr; + } + + while (XLogReadRecord(logical_decoding_ctx->reader, + &record, &errm) == XLREAD_NEED_DATA) + { + if (!logical_decoding_ctx->read_page(logical_decoding_ctx)) + { + record = NULL; + break; + } + } /* xlog record was invalid */ if (errm != NULL) diff --git a/src/bin/pg_rewind/parsexlog.c b/src/bin/pg_rewind/parsexlog.c index 63c3879ead8..41709cba7a5 100644 --- a/src/bin/pg_rewind/parsexlog.c +++ b/src/bin/pg_rewind/parsexlog.c @@ -41,16 +41,8 @@ static int xlogreadfd = -1; static XLogSegNo xlogreadsegno = -1; static char xlogfpath[MAXPGPATH]; -typedef struct XLogPageReadPrivate -{ - const char *datadir; - int tliIndex; -} XLogPageReadPrivate; - -static int SimpleXLogPageRead(XLogReaderState *xlogreader, - XLogRecPtr targetPagePtr, - int reqLen, XLogRecPtr targetRecPtr, char *readBuf, - TimeLineID *pageTLI); +static void SimpleXLogPageRead(XLogReaderState *xlogreader, + const char *datadir, int *tliIndex); /* * Read WAL from the datadir/pg_wal, starting from 'startpoint' on timeline @@ -64,24 +56,25 @@ extractPageMap(const char *datadir, XLogRecPtr startpoint, int tliIndex, XLogRecord *record; XLogReaderState *xlogreader; char *errormsg; - XLogPageReadPrivate private; - private.datadir = datadir; - private.tliIndex = tliIndex; - xlogreader = XLogReaderAllocate(WalSegSz, &SimpleXLogPageRead, - &private); + xlogreader = XLogReaderAllocate(WalSegSz); if (xlogreader == NULL) pg_fatal("out of memory"); + xlogreader->readBuf = pg_malloc(XLOG_BLCKSZ); + + XLogBeginRead(xlogreader, startpoint); do { - record = XLogReadRecord(xlogreader, startpoint, &errormsg); + while (XLogReadRecord(xlogreader, &record, &errormsg) == + XLREAD_NEED_DATA) + { + SimpleXLogPageRead(xlogreader, datadir, &tliIndex); + } if (record == NULL) { - XLogRecPtr errptr; - - errptr = startpoint ? startpoint : xlogreader->EndRecPtr; + XLogRecPtr errptr = xlogreader->EndRecPtr; if (errormsg) pg_fatal("could not read WAL record at %X/%X: %s", @@ -93,11 +86,9 @@ extractPageMap(const char *datadir, XLogRecPtr startpoint, int tliIndex, } extractPageInfo(xlogreader); - - startpoint = InvalidXLogRecPtr; /* continue reading at next record */ - } while (xlogreader->ReadRecPtr != endpoint); + pg_free(xlogreader->readBuf); XLogReaderFree(xlogreader); if (xlogreadfd != -1) { @@ -116,17 +107,21 @@ readOneRecord(const char *datadir, XLogRecPtr ptr, int tliIndex) XLogRecord *record; XLogReaderState *xlogreader; char *errormsg; - XLogPageReadPrivate private; XLogRecPtr endptr; - private.datadir = datadir; - private.tliIndex = tliIndex; - xlogreader = XLogReaderAllocate(WalSegSz, &SimpleXLogPageRead, - &private); + xlogreader = XLogReaderAllocate(WalSegSz); if (xlogreader == NULL) pg_fatal("out of memory"); + xlogreader->readBuf = pg_malloc(XLOG_BLCKSZ); + + XLogBeginRead(xlogreader, ptr); + + while (XLogReadRecord(xlogreader, &record, &errormsg) == + XLREAD_NEED_DATA) + { + SimpleXLogPageRead(xlogreader, datadir, &tliIndex); + } - record = XLogReadRecord(xlogreader, ptr, &errormsg); if (record == NULL) { if (errormsg) @@ -138,6 +133,7 @@ readOneRecord(const char *datadir, XLogRecPtr ptr, int tliIndex) } endptr = xlogreader->EndRecPtr; + pg_free(xlogreader->readBuf); XLogReaderFree(xlogreader); if (xlogreadfd != -1) { @@ -161,7 +157,6 @@ findLastCheckpoint(const char *datadir, XLogRecPtr forkptr, int tliIndex, XLogRecPtr searchptr; XLogReaderState *xlogreader; char *errormsg; - XLogPageReadPrivate private; /* * The given fork pointer points to the end of the last common record, @@ -177,19 +172,22 @@ findLastCheckpoint(const char *datadir, XLogRecPtr forkptr, int tliIndex, forkptr += SizeOfXLogShortPHD; } - private.datadir = datadir; - private.tliIndex = tliIndex; - xlogreader = XLogReaderAllocate(WalSegSz, &SimpleXLogPageRead, - &private); + xlogreader = XLogReaderAllocate(WalSegSz); if (xlogreader == NULL) pg_fatal("out of memory"); + xlogreader->readBuf = pg_malloc(XLOG_BLCKSZ); searchptr = forkptr; for (;;) { uint8 info; - record = XLogReadRecord(xlogreader, searchptr, &errormsg); + XLogBeginRead(xlogreader, searchptr); + while (XLogReadRecord(xlogreader, &record, &errormsg) == + XLREAD_NEED_DATA) + { + SimpleXLogPageRead(xlogreader, datadir, &tliIndex); + } if (record == NULL) { @@ -226,6 +224,7 @@ findLastCheckpoint(const char *datadir, XLogRecPtr forkptr, int tliIndex, searchptr = record->xl_prev; } + pg_free(xlogreader->readBuf); XLogReaderFree(xlogreader); if (xlogreadfd != -1) { @@ -234,13 +233,14 @@ findLastCheckpoint(const char *datadir, XLogRecPtr forkptr, int tliIndex, } } -/* XLogReader callback function, to read a WAL page */ -static int -SimpleXLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr, - int reqLen, XLogRecPtr targetRecPtr, char *readBuf, - TimeLineID *pageTLI) +/* XLogreader callback function, to read a WAL page */ +static void +SimpleXLogPageRead(XLogReaderState *xlogreader, + const char*datadir, int *tliIndex) { - XLogPageReadPrivate *private = (XLogPageReadPrivate *) xlogreader->private_data; + XLogRecPtr targetPagePtr = xlogreader->loadPagePtr; + char *readBuf = xlogreader->readBuf; + TimeLineID *pageTLI = &xlogreader->readPageTLI; uint32 targetPageOff; XLogRecPtr targetSegEnd; XLogSegNo targetSegNo; @@ -273,24 +273,26 @@ SimpleXLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr, * be done both forward and backward, consider also switching timeline * accordingly. */ - while (private->tliIndex < targetNentries - 1 && - targetHistory[private->tliIndex].end < targetSegEnd) - private->tliIndex++; - while (private->tliIndex > 0 && - targetHistory[private->tliIndex].begin >= targetSegEnd) - private->tliIndex--; - - XLogFileName(xlogfname, targetHistory[private->tliIndex].tli, + while (*tliIndex < targetNentries - 1 && + targetHistory[*tliIndex].end < targetSegEnd) + (*tliIndex)++; + while (*tliIndex > 0 && + targetHistory[*tliIndex].begin >= targetSegEnd) + (*tliIndex)--; + + XLogFileName(xlogfname, targetHistory[*tliIndex].tli, xlogreadsegno, WalSegSz); - snprintf(xlogfpath, MAXPGPATH, "%s/" XLOGDIR "/%s", private->datadir, xlogfname); + snprintf(xlogfpath, MAXPGPATH, "%s/" XLOGDIR "/%s", datadir, xlogfname); xlogreadfd = open(xlogfpath, O_RDONLY | PG_BINARY, 0); if (xlogreadfd < 0) { pg_log_error("could not open file \"%s\": %m", xlogfpath); - return -1; + xlogreader->readPagePtr = InvalidXLogRecPtr; + xlogreader->readLen = -1; + return; } } @@ -303,7 +305,9 @@ SimpleXLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr, if (lseek(xlogreadfd, (off_t) targetPageOff, SEEK_SET) < 0) { pg_log_error("could not seek in file \"%s\": %m", xlogfpath); - return -1; + xlogreader->readPagePtr = InvalidXLogRecPtr; + xlogreader->readLen = -1; + return; } @@ -316,13 +320,18 @@ SimpleXLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr, pg_log_error("could not read file \"%s\": read %d of %zu", xlogfpath, r, (Size) XLOG_BLCKSZ); - return -1; + xlogreader->readPagePtr = InvalidXLogRecPtr; + xlogreader->readLen = -1; + return; } Assert(targetSegNo == xlogreadsegno); - *pageTLI = targetHistory[private->tliIndex].tli; - return XLOG_BLCKSZ; + *pageTLI = targetHistory[*tliIndex].tli; + + xlogreader->readPagePtr = targetPagePtr; + xlogreader->readLen = XLOG_BLCKSZ; + return; } /* diff --git a/src/bin/pg_waldump/pg_waldump.c b/src/bin/pg_waldump/pg_waldump.c index b95d467805a..c82c39e7147 100644 --- a/src/bin/pg_waldump/pg_waldump.c +++ b/src/bin/pg_waldump/pg_waldump.c @@ -419,13 +419,14 @@ XLogDumpXLogRead(const char *directory, TimeLineID timeline_id, } /* - * XLogReader read_page callback + * Read the page requested by XLogReadReacord into state->readBuf. */ -static int -XLogDumpReadPage(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen, - XLogRecPtr targetPtr, char *readBuff, TimeLineID *curFileTLI) +static void +XLogDumpReadPage(XLogReaderState *state, XLogDumpPrivate *private) { - XLogDumpPrivate *private = state->private_data; + XLogRecPtr targetPagePtr = state->loadPagePtr; + int reqLen = state->loadLen; + char *readBuff = state->readBuf; int count = XLOG_BLCKSZ; if (private->endptr != InvalidXLogRecPtr) @@ -437,14 +438,18 @@ XLogDumpReadPage(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen, else { private->endptr_reached = true; - return -1; + state->readPagePtr = InvalidXLogRecPtr; + state->readLen = -1; + return; } } XLogDumpXLogRead(private->inpath, private->timeline, targetPagePtr, readBuff, count); - return count; + state->readPagePtr = targetPagePtr; + state->readLen = count; + return; } /* @@ -1100,37 +1105,23 @@ main(int argc, char **argv) /* done with argument parsing, do the actual work */ /* we have everything we need, start reading */ - xlogreader_state = XLogReaderAllocate(WalSegSz, XLogDumpReadPage, - &private); + xlogreader_state = XLogReaderAllocate(WalSegSz); if (!xlogreader_state) fatal_error("out of memory"); + xlogreader_state->readBuf = palloc(XLOG_BLCKSZ); /* first find a valid recptr to start from */ - first_record = XLogFindNextRecord(xlogreader_state, private.startptr); - - if (first_record == InvalidXLogRecPtr) - fatal_error("could not find a valid record after %X/%X", - (uint32) (private.startptr >> 32), - (uint32) private.startptr); - - /* - * Display a message that we're skipping data if `from` wasn't a pointer - * to the start of a record and also wasn't a pointer to the beginning of - * a segment (e.g. we were used in file mode). - */ - if (first_record != private.startptr && - XLogSegmentOffset(private.startptr, WalSegSz) != 0) - printf(ngettext("first record is after %X/%X, at %X/%X, skipping over %u byte\n", - "first record is after %X/%X, at %X/%X, skipping over %u bytes\n", - (first_record - private.startptr)), - (uint32) (private.startptr >> 32), (uint32) private.startptr, - (uint32) (first_record >> 32), (uint32) first_record, - (uint32) (first_record - private.startptr)); + XLogFindNextRecord(xlogreader_state, private.startptr); for (;;) { /* try to read the next record */ - record = XLogReadRecord(xlogreader_state, first_record, &errormsg); + while (XLogReadRecord(xlogreader_state, &record, &errormsg) == + XLREAD_NEED_DATA) + { + XLogDumpReadPage(xlogreader_state, &private); + } + if (!record) { if (!config.follow || private.endptr_reached) @@ -1142,8 +1133,23 @@ main(int argc, char **argv) } } - /* after reading the first record, continue at next one */ - first_record = InvalidXLogRecPtr; + if (first_record == InvalidXLogRecPtr) + { + /* + * Display a message that we're skipping data if `from` wasn't a pointer + * to the start of a record and also wasn't a pointer to the beginning of + * a segment (e.g. we were used in file mode). + */ + first_record = xlogreader_state->ReadRecPtr; + if (first_record != private.startptr && + XLogSegmentOffset(private.startptr, WalSegSz) != 0) + printf(ngettext("first record is after %X/%X, at %X/%X, skipping over %u byte\n", + "first record is after %X/%X, at %X/%X, skipping over %u bytes\n", + (first_record - private.startptr)), + (uint32) (private.startptr >> 32), (uint32) private.startptr, + (uint32) (first_record >> 32), (uint32) first_record, + (uint32) (first_record - private.startptr)); + } /* apply all specified filters */ if (config.filter_by_rmgr != -1 && @@ -1176,6 +1182,7 @@ main(int argc, char **argv) (uint32) xlogreader_state->ReadRecPtr, errormsg); + pfree(xlogreader_state->readBuf); XLogReaderFree(xlogreader_state); return EXIT_SUCCESS; diff --git a/src/include/access/xlogreader.h b/src/include/access/xlogreader.h index aa9bc637259..5e472893e6b 100644 --- a/src/include/access/xlogreader.h +++ b/src/include/access/xlogreader.h @@ -9,17 +9,60 @@ * src/include/access/xlogreader.h * * NOTES - * See 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. + * Usage * - * After reading a record with XLogReadRecord(), it's decomposed into - * the per-block and main data parts, and the parts can be accessed - * with the XLogRec* macros and functions. You can also decode a - * record that's already constructed in memory, without reading from - * disk, by calling the DecodeXLogRecord() function. + * 1. Allocate a new xlogreader with XLogReaderAllocate(). + * 2. Position the reader to the desired starting point with XLogBeginRead() or + * XLogFindNextRecord() + * 3. Call XLogReadRecord(). Whenever it returns XL_NEED_DATA, provide the + * requested data in readBuf, and call it again. + * + * xlogreader = XLogReaderAllocate(); + * if (!xlogreader) + * elog(ERROR, "out of memory"); + * + * XLogBeginRead(xlogreader, starting_point); + * for (;;) + * { + * XLogReadRecordResult rc; + * XLogRecord record; + * char *errormsg; + * + * rc = XLogReadRecord(xlogreader, &record, &errormsg); + * if (rc == XLREAD_NEED_DATA) + * { + * ... Load the data requested by xlogreader->loadPagePtr and loadLen + * ... into readBuf and readLen + * xlogreader->readBuf = + * xlogreader->readPagePtr = xxx; + * xlogreader->readLen = xxx; + * xlogreader->readTLI = xxx; + * continue; + * } + * + * if (rc == XLREAD_FAIL) + * ereport(ERROR, "read failed: %s", errormsg); + * + * Assert (rc == XLREAD_SUCCESS); + * + * ... process the record, using XLogRec* macros ... + * } + * + * + * It is up to the caller to manage the read buffer, at 'readBuf'. A common + * strategy is to allocate a buffer of size XLOG_BLCKSZ at the same time the + * xlogreader is allocated, and read the requested data into the same buffer. + * However, XLogReadRecord() assumes that it does not change between calls, + * unless XLogReadRecord() asks for a new page by returning XLREAD_NEED_DATA, + * or the caller resets the read position by calling XLogBeginRead(). + * + * + * When XLogReadRecord() finishes reading a record, it returns XLREAD_SUCCESS. + * The record decomposed into the per-block and main data parts, and the parts + * can be accessed with the XLogRec* macros and functions. You can also decode + * a record that's already constructed in memory, without reading from disk, + * by calling the DecodeXLogRecord() function. *------------------------------------------------------------------------- */ #ifndef XLOGREADER_H @@ -33,14 +76,6 @@ typedef struct XLogReaderState XLogReaderState; -/* Function type definition for the read_page callback */ -typedef int (*XLogPageReadCB) (XLogReaderState *xlogreader, - XLogRecPtr targetPagePtr, - int reqLen, - XLogRecPtr targetRecPtr, - char *readBuf, - TimeLineID *pageTLI); - typedef struct { /* Is this block ref in use? */ @@ -82,29 +117,6 @@ struct XLogReaderState */ int wal_segment_size; - /* - * Data input callback (mandatory). - * - * This callback shall read at least reqLen valid bytes of the xlog page - * starting at targetPagePtr, and store them in readBuf. The callback - * shall return the number of bytes read (never more than XLOG_BLCKSZ), or - * -1 on failure. The callback shall sleep, if necessary, to wait for the - * requested bytes to become available. The callback will not be invoked - * again for the same page unless more than the returned number of bytes - * are needed. - * - * targetRecPtr is the position of the WAL record we're reading. Usually - * it is equal to targetPagePtr + reqLen, but sometimes xlogreader needs - * to read and verify the page or segment header, before it reads the - * actual WAL record it's interested in. In that case, targetRecPtr can - * be used to determine which timeline to read the page from. - * - * The callback shall set *pageTLI to the TLI of the file the page was - * read from. It is currently used only for error reporting purposes, to - * reconstruct the name of the WAL file where an error occurred. - */ - XLogPageReadCB read_page; - /* * System identifier of the xlog files we're about to read. Set to zero * (the default value) if unknown or unimportant. @@ -112,23 +124,45 @@ struct XLogReaderState uint64 system_identifier; /* - * Opaque data for callbacks to use. Not used by XLogReader. + * Start and end point of last record read. (Use XLogBeginRead() to set these) */ - void *private_data; + XLogRecPtr ReadRecPtr; /* start of last record read or being read */ + XLogRecPtr EndRecPtr; /* end+1 of last record read */ + XLogRecPtr PrevRecPtr; /* start of previous record read */ /* - * Start and end point of last record read. EndRecPtr is also used as the - * position to read next, if XLogReadRecord receives an invalid recptr. + * If we are skipping records to find the starting point + * (because XLogFindNextRecord() was called), SkipRecPtr is the point to + * skip to. */ - XLogRecPtr ReadRecPtr; /* start of last record read */ - XLogRecPtr EndRecPtr; /* end+1 of last record read */ + XLogRecPtr SkipRecPtr; + + /* ---------------------------------------- + * Communication with page reader. + * + * When XLogReadRecord() returns XLREAD_NEED_DATA, the caller must provide + * the data indicated by loadPagePtr/loadLen, before calling XLogReadRecord() + * again. To provide the data, set the readPagePtr, readLen, readPageTLI and + * readBuf fields. + * ---------------------------------------- + */ + /* parameters to page reader */ + XLogRecPtr loadPagePtr; /* Pointer to the page */ + int loadLen; /* wanted length in bytes */ + /* return from page reader */ + XLogRecPtr readPagePtr; + int32 readLen; /* bytes actually read, must be at least + * loadLen. -1 on error. */ + TimeLineID readPageTLI; /* TLI for data currently in readBuf */ + char *readBuf; /* buffer to store data */ /* ---------------------------------------- * Decoded representation of current record * * Use XLogRecGet* functions to investigate the record; these fields - * should not be accessed directly. + * should not be accessed directly. The current record is valid after + * XLogReadRecord() returns XLREAD_SUCCESS, until the next call. * ---------------------------------------- */ XLogRecord *decoded_record; /* currently decoded record */ @@ -149,17 +183,12 @@ struct XLogReaderState * ---------------------------------------- */ - /* - * Buffer for currently read page (XLOG_BLCKSZ bytes, valid up to at least - * readLen bytes) - */ - char *readBuf; - uint32 readLen; + /* state of the XLogReadRecord state machine */ + int readRecordState; /* enum */ - /* last read segment, segment offset, TLI for data currently in readBuf */ - XLogSegNo readSegNo; - uint32 readOff; - TimeLineID readPageTLI; + /* last read segment and segment offset for data currently in readBuf */ + XLogRecPtr verifiedPagePtr; + uint32 verifiedPageLen; /* * beginning of prior page read, and its TLI. Doesn't necessarily @@ -168,8 +197,30 @@ struct XLogReaderState XLogRecPtr latestPagePtr; TimeLineID latestPageTLI; - /* beginning of the WAL record being read. */ - XLogRecPtr currRecPtr; + /* + * Buffer for current ReadRecord result (expandable), used when a record + * crosses a page boundary. + */ + char *readRecordBuf; + uint32 readRecordBufSize; + + int recordGotLen; /* amount of current record that has already been read */ + int recordRemainLen; /* length of current record that remains */ + XLogRecPtr recordContRecPtr; /* where the current record continues */ + + /* Buffer to hold error message */ + char *errormsg_buf; + + + + /* ---------------------------------------- + * private fields used by xlogutils.c + * + * XXX: These are not used by xlogreader itself. They don't really belong + * here.. + * ---------------------------------------- + */ + /* timeline to read it from, 0 if a lookup is required */ TimeLineID currTLI; @@ -188,41 +239,37 @@ struct XLogReaderState * read from when currTLIValidUntil is reached. */ TimeLineID nextTLI; - - /* - * Buffer for current ReadRecord result (expandable), used when a record - * crosses a page boundary. - */ - char *readRecordBuf; - uint32 readRecordBufSize; - - /* Buffer to hold error message */ - char *errormsg_buf; }; /* Get a new XLogReader */ -extern XLogReaderState *XLogReaderAllocate(int wal_segment_size, - XLogPageReadCB pagereadfunc, - void *private_data); +extern XLogReaderState *XLogReaderAllocate(int wal_segment_size); /* Free an XLogReader */ extern void XLogReaderFree(XLogReaderState *state); -/* Read the next XLog record. Returns NULL on end-of-WAL or failure */ -extern struct XLogRecord *XLogReadRecord(XLogReaderState *state, - XLogRecPtr recptr, char **errormsg); +/* Position the XLogReader to given record */ +extern void XLogBeginRead(XLogReaderState *state, XLogRecPtr recptr); +#ifdef FRONTEND +extern void XLogFindNextRecord(XLogReaderState *state, XLogRecPtr recptr); +#endif /* FRONTEND */ + +/* Return code from XLogReadRecord */ +typedef enum XLogReadRecordResult +{ + XLREAD_SUCCESS, /* record is successfully read */ + XLREAD_NEED_DATA, /* need more data. see XLogReadRecord. */ + XLREAD_FAIL /* failed during reading a record */ +} XLogReadRecordResult; + +/* Read the next XLog record. */ +extern XLogReadRecordResult XLogReadRecord(XLogReaderState *state, + XLogRecord **record, + char **errormsg); /* Validate a page */ extern bool XLogReaderValidatePageHeader(XLogReaderState *state, XLogRecPtr recptr, char *phdr); -/* Invalidate read state */ -extern void XLogReaderInvalReadState(XLogReaderState *state); - -#ifdef FRONTEND -extern XLogRecPtr XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr); -#endif /* FRONTEND */ - /* Functions for decoding an XLogRecord */ extern bool DecodeXLogRecord(XLogReaderState *state, XLogRecord *record, diff --git a/src/include/access/xlogutils.h b/src/include/access/xlogutils.h index 4105b59904b..55a9b6237ab 100644 --- a/src/include/access/xlogutils.h +++ b/src/include/access/xlogutils.h @@ -47,10 +47,7 @@ extern Buffer XLogReadBufferExtended(RelFileNode rnode, ForkNumber forknum, extern Relation CreateFakeRelcacheEntry(RelFileNode rnode); extern void FreeFakeRelcacheEntry(Relation fakerel); -extern int read_local_xlog_page(XLogReaderState *state, - XLogRecPtr targetPagePtr, int reqLen, - XLogRecPtr targetRecPtr, char *cur_page, - TimeLineID *pageTLI); +extern bool read_local_xlog_page(XLogReaderState *state); extern void XLogReadDetermineTimeline(XLogReaderState *state, XLogRecPtr wantPage, uint32 wantLength); diff --git a/src/include/replication/logical.h b/src/include/replication/logical.h index 31c796b7651..befe8ebbeb5 100644 --- a/src/include/replication/logical.h +++ b/src/include/replication/logical.h @@ -30,7 +30,11 @@ typedef void (*LogicalOutputPluginWriterUpdateProgress) (struct LogicalDecodingC TransactionId xid ); -typedef struct LogicalDecodingContext +typedef struct LogicalDecodingContext LogicalDecodingContext; + +typedef bool (*LogicalDecodingXLogReadPageCB)(LogicalDecodingContext *ctx); + +struct LogicalDecodingContext { /* memory context this is all allocated in */ MemoryContext context; @@ -40,6 +44,7 @@ typedef struct LogicalDecodingContext /* infrastructure pieces for decoding */ XLogReaderState *reader; + LogicalDecodingXLogReadPageCB read_page; struct ReorderBuffer *reorder; struct SnapBuild *snapshot_builder; @@ -87,7 +92,7 @@ typedef struct LogicalDecodingContext bool prepared_write; XLogRecPtr write_location; TransactionId write_xid; -} LogicalDecodingContext; +}; extern void CheckLogicalDecodingRequirements(void); @@ -96,14 +101,14 @@ extern LogicalDecodingContext *CreateInitDecodingContext(char *plugin, List *output_plugin_options, bool need_full_snapshot, XLogRecPtr restart_lsn, - XLogPageReadCB read_page, + LogicalDecodingXLogReadPageCB read_page, LogicalOutputPluginWriterPrepareWrite prepare_write, LogicalOutputPluginWriterWrite do_write, LogicalOutputPluginWriterUpdateProgress update_progress); extern LogicalDecodingContext *CreateDecodingContext(XLogRecPtr start_lsn, List *output_plugin_options, bool fast_forward, - XLogPageReadCB read_page, + LogicalDecodingXLogReadPageCB read_page, LogicalOutputPluginWriterPrepareWrite prepare_write, LogicalOutputPluginWriterWrite do_write, LogicalOutputPluginWriterUpdateProgress update_progress); diff --git a/src/include/replication/logicalfuncs.h b/src/include/replication/logicalfuncs.h index a9c178a9e68..25fa68d5b93 100644 --- a/src/include/replication/logicalfuncs.h +++ b/src/include/replication/logicalfuncs.h @@ -11,9 +11,6 @@ #include "replication/logical.h" -extern int logical_read_local_xlog_page(XLogReaderState *state, - XLogRecPtr targetPagePtr, - int reqLen, XLogRecPtr targetRecPtr, - char *cur_page, TimeLineID *pageTLI); +extern bool logical_read_local_xlog_page(LogicalDecodingContext *ctx); #endif