From: Kyotaro Horiguchi Date: Tue, 10 Sep 2019 12:58:27 +0900 Subject: [PATCH v8 2/4] Move page-reader out of XLogReadRecord This is the second step of removing callbacks from WAL record reader. Since it is essential to take in additional data while reading a record, the function have to ask caller for new data while keeping working state. Thus the function is turned into a state machine. --- src/backend/access/transam/twophase.c | 11 +- src/backend/access/transam/xlog.c | 54 +- src/backend/access/transam/xlogreader.c | 681 +++++++++++------- src/backend/access/transam/xlogutils.c | 12 +- src/backend/replication/logical/logical.c | 17 +- .../replication/logical/logicalfuncs.c | 8 +- src/backend/replication/slotfuncs.c | 8 +- src/backend/replication/walsender.c | 13 +- src/bin/pg_rewind/parsexlog.c | 80 +- src/bin/pg_waldump/pg_waldump.c | 24 +- src/include/access/xlogreader.h | 90 +-- src/include/access/xlogutils.h | 4 +- src/include/replication/logical.h | 9 +- 13 files changed, 601 insertions(+), 410 deletions(-) diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c index 2f7d4ed59a..24b08810ac 100644 --- a/src/backend/access/transam/twophase.c +++ b/src/backend/access/transam/twophase.c @@ -1330,8 +1330,7 @@ XlogReadTwoPhaseData(XLogRecPtr lsn, char **buf, int *len) XLogReaderState *xlogreader; char *errormsg; - xlogreader = XLogReaderAllocate(wal_segment_size, NULL, - &read_local_xlog_page, NULL); + xlogreader = XLogReaderAllocate(wal_segment_size, NULL); if (!xlogreader) ereport(ERROR, (errcode(ERRCODE_OUT_OF_MEMORY), @@ -1339,7 +1338,13 @@ XlogReadTwoPhaseData(XLogRecPtr lsn, char **buf, int *len) errdetail("Failed while allocating a WAL reading processor."))); XLogBeginRead(xlogreader, lsn); - record = XLogReadRecord(xlogreader, &errormsg); + while (XLogReadRecord(xlogreader, &record, &errormsg) == + XLREAD_NEED_DATA) + { + if (!read_local_xlog_page(xlogreader)) + break; + } + if (record == NULL) ereport(ERROR, (errcode_for_file_access(), diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index b0ad9376d6..e380eaa186 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -827,13 +827,6 @@ static XLogSource currentSource = XLOG_FROM_ANY; static bool lastSourceFailed = false; static bool pendingWalRcvRestart = 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 @@ -908,8 +901,8 @@ static bool InstallXLogFileSegment(XLogSegNo *segno, char *tmppath, static int XLogFileRead(XLogSegNo segno, int emode, TimeLineID tli, XLogSource source, bool notfoundOk); static int XLogFileReadAnyTLI(XLogSegNo segno, int emode, XLogSource source); -static bool XLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr, - int reqLen, XLogRecPtr targetRecPtr, char *readBuf); +static bool 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); @@ -1221,8 +1214,7 @@ XLogInsertRecord(XLogRecData *rdata, appendBinaryStringInfo(&recordBuf, rdata->data, rdata->len); if (!debug_reader) - debug_reader = XLogReaderAllocate(wal_segment_size, NULL, - NULL, NULL); + debug_reader = XLogReaderAllocate(wal_segment_size, NULL); if (!debug_reader) { @@ -4322,15 +4314,10 @@ CleanupBackupHistory(void) * record is available. */ static XLogRecord * -ReadRecord(XLogReaderState *xlogreader, int emode, - bool fetching_ckpt) +ReadRecord(XLogReaderState *xlogreader, int emode, bool fetching_ckpt) { XLogRecord *record; - XLogPageReadPrivate *private = (XLogPageReadPrivate *) xlogreader->private_data; - - private->fetching_ckpt = fetching_ckpt; - private->emode = emode; - private->randAccess = (xlogreader->ReadRecPtr == InvalidXLogRecPtr); + bool randAccess = (xlogreader->ReadRecPtr == InvalidXLogRecPtr); /* This is the first attempt to read this page. */ lastSourceFailed = false; @@ -4338,8 +4325,16 @@ ReadRecord(XLogReaderState *xlogreader, int emode, for (;;) { char *errormsg; + XLogReadRecordResult result; + + while ((result = XLogReadRecord(xlogreader, &record, &errormsg)) + == XLREAD_NEED_DATA) + { + if (!XLogPageRead(xlogreader, fetching_ckpt, emode, randAccess)) + break; + + } - record = XLogReadRecord(xlogreader, &errormsg); ReadRecPtr = xlogreader->ReadRecPtr; EndRecPtr = xlogreader->EndRecPtr; if (record == NULL) @@ -6309,7 +6304,6 @@ StartupXLOG(void) bool backupFromStandby = false; DBState dbstate_at_startup; XLogReaderState *xlogreader; - XLogPageReadPrivate private; bool fast_promoted = false; struct stat st; @@ -6465,9 +6459,7 @@ StartupXLOG(void) OwnLatch(&XLogCtl->recoveryWakeupLatch); /* Set up XLOG reader facility */ - MemSet(&private, 0, sizeof(XLogPageReadPrivate)); - xlogreader = XLogReaderAllocate(wal_segment_size, NULL, - &XLogPageRead, &private); + xlogreader = XLogReaderAllocate(wal_segment_size, NULL); if (!xlogreader) ereport(ERROR, (errcode(ERRCODE_OUT_OF_MEMORY), @@ -11810,12 +11802,13 @@ CancelBackup(void) * sleep and retry. */ static bool -XLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr, int reqLen, - XLogRecPtr targetRecPtr, char *readBuf) +XLogPageRead(XLogReaderState *xlogreader, + bool fetching_ckpt, int emode, bool randAccess) { - XLogPageReadPrivate *private = - (XLogPageReadPrivate *) xlogreader->private_data; - int emode = private->emode; + char *readBuf = xlogreader->readBuf; + XLogRecPtr targetPagePtr = xlogreader->readPagePtr; + int reqLen = xlogreader->readLen; + XLogRecPtr targetRecPtr = xlogreader->ReadRecPtr; uint32 targetPageOff; XLogSegNo targetSegNo PG_USED_FOR_ASSERTS_ONLY; int r; @@ -11858,8 +11851,8 @@ retry: flushedUpto < targetPagePtr + reqLen)) { if (!WaitForWALToBecomeAvailable(targetPagePtr + reqLen, - private->randAccess, - private->fetching_ckpt, + randAccess, + fetching_ckpt, targetRecPtr)) { if (readFile >= 0) @@ -11964,6 +11957,7 @@ retry: goto next_record_is_invalid; } + Assert(xlogreader->readPagePtr == targetPagePtr); xlogreader->readLen = readLen; return true; diff --git a/src/backend/access/transam/xlogreader.c b/src/backend/access/transam/xlogreader.c index 6250093dd9..7c8c6fc314 100644 --- a/src/backend/access/transam/xlogreader.c +++ b/src/backend/access/transam/xlogreader.c @@ -40,7 +40,7 @@ static bool XLogNeedData(XLogReaderState *state, XLogRecPtr pageptr, int reqLen, bool header_inclusive); static void XLogReaderInvalReadState(XLogReaderState *state); 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 void ResetDecoder(XLogReaderState *state); @@ -70,8 +70,7 @@ report_invalid_record(XLogReaderState *state, const char *fmt,...) * Returns NULL if the xlogreader couldn't be allocated. */ XLogReaderState * -XLogReaderAllocate(int wal_segment_size, const char *waldir, - XLogPageReadCB pagereadfunc, void *private_data) +XLogReaderAllocate(int wal_segment_size, const char *waldir) { XLogReaderState *state; @@ -102,9 +101,6 @@ XLogReaderAllocate(int wal_segment_size, const char *waldir, WALOpenSegmentInit(&state->seg, &state->segcxt, wal_segment_size, waldir); - state->read_page = pagereadfunc; - /* system_identifier initialized to zeroes above */ - state->private_data = private_data; /* ReadRecPtr, EndRecPtr and readLen initialized to zeroes above */ state->errormsg_buf = palloc_extended(MAX_ERRORMSG_LEN + 1, MCXT_ALLOC_NO_OOM); @@ -242,6 +238,7 @@ XLogBeginRead(XLogReaderState *state, XLogRecPtr RecPtr) /* Begin at the passed-in record pointer. */ state->EndRecPtr = RecPtr; state->ReadRecPtr = InvalidXLogRecPtr; + state->readRecordState = XLREAD_NEXT_RECORD; } /* @@ -250,314 +247,452 @@ XLogBeginRead(XLogReaderState *state, XLogRecPtr RecPtr) * XLogBeginRead() or XLogFindNextRecord() must be called before the first call * to XLogReadRecord(). * - * 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. + * This function may return XLREAD_NEED_DATA several times before returning a + * result record. The caller shall read in some new data then call this + * function again with the same parameters. * - * If the reading fails for some other reason, NULL is also returned, and - * *errormsg is set to a string with details of the failure. + * When a record is successfully read, returns XLREAD_SUCCESS with result + * record being stored in *record. Otherwise *record is NULL. * - * The returned pointer (or *errormsg) points to an internal buffer that's - * valid until the next call to XLogReadRecord. + * Returns XLREAD_NEED_DATA if more data is needed to finish reading the + * current record. In that case, state->readPagePtr and state->readLen inform + * the desired 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->readPageTLI and + * state->readLen to indicate the timeline that it was read from, and the + * length of data that is now available (which must be >= given readLen), + * respectively. + * + * If invalid data is encountered, 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. + * + * + * This function runs a state machine consists of the following states. + * + * XLREAD_NEXT_RECORD : + * The initial state, if called with valid RecPtr, try to read a record at + * that position. If invalid RecPtr is given try to read a record just after + * the last one previously read. + * This state ens after setting ReadRecPtr. Then goes to XLREAD_TOT_LEN. + * + * XLREAD_TOT_LEN: + * Examining record header. Ends after reading record total + * length. recordRemainLen and recordGotLen are initialized. + * + * XLREAD_FIRST_FRAGMENT: + * Reading the first fragment. Ends with finishing reading a single + * record. Goes to XLREAD_NEXT_RECORD if that's all or + * XLREAD_CONTINUATION if we have continuation. + + * XLREAD_CONTINUATION: + * Reading continuation of record. Ends with finishing the whole record then + * goes to XLREAD_NEXT_RECORD. During this state, recordRemainLen indicates + * how much is left and readRecordBuf holds the partially assert + * record.recordContRecPtr points to the beginning of the next page where to + * continue. + * + * If wrong data found in any state, the state machine stays at the current + * state. This behavior allows to continue reading a reacord switching among + * different souces, while streaming replication. */ -XLogRecord * -XLogReadRecord(XLogReaderState *state, char **errormsg) +XLogReadRecordResult +XLogReadRecord(XLogReaderState *state, XLogRecord **record, char **errormsg) { - XLogRecPtr RecPtr; - XLogRecord *record; - XLogRecPtr targetPagePtr; - bool randAccess; - uint32 len, - total_len; - uint32 targetRecOff; - uint32 pageHeaderSize; - bool gotheader; + XLogRecord *prec; - /* - * 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. - */ - randAccess = false; + *record = NULL; /* reset error state */ *errormsg = NULL; state->errormsg_buf[0] = '\0'; - ResetDecoder(state); - - RecPtr = state->EndRecPtr; - - if (state->ReadRecPtr != InvalidXLogRecPtr) - { - /* read the record after the one we just read */ - - /* - * 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, 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, EndRecPtr should already be pointing to a valid - * record starting position. - */ - Assert(XRecOffIsValid(RecPtr)); - randAccess = true; - } - - state->currRecPtr = RecPtr; - - targetPagePtr = RecPtr - (RecPtr % XLOG_BLCKSZ); - targetRecOff = RecPtr % XLOG_BLCKSZ; - - /* - * 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. - */ - while (XLogNeedData(state, targetPagePtr, - Min(targetRecOff + SizeOfXLogRecord, XLOG_BLCKSZ), - targetRecOff != 0)) - { - if (!state->read_page(state, state->readPagePtr, state->readLen, - RecPtr, state->readBuf)) - break; - } - - if (!state->page_verified) - goto err; - - /* - * We have at least the page header, so we can examine it now. - */ - 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; - } - - 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; - } - - /* XLogNeedData has verified the page header */ - Assert(pageHeaderSize <= state->readLen); - - /* - * 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(state, RecPtr, state->ReadRecPtr, record, - randAccess)) - goto err; - gotheader = true; - } - 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) (RecPtr >> 32), (uint32) RecPtr, - (uint32) SizeOfXLogRecord, total_len); - goto err; - } - gotheader = false; - } - - len = XLOG_BLCKSZ - RecPtr % XLOG_BLCKSZ; - if (total_len > len) + switch (state->readRecordState) { - /* Need to reassemble record */ - char *contdata; - XLogPageHeader pageHeader; - char *buffer; - uint32 gotlen; - - /* - * 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) (RecPtr >> 32), (uint32) RecPtr); - goto err; - } + case XLREAD_NEXT_RECORD: + ResetDecoder(state); - /* 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 - { - int rest_len = total_len - gotlen; - - /* Calculate pointer to beginning of next page */ - targetPagePtr += XLOG_BLCKSZ; - - /* Wait for the next page to become available */ - while (XLogNeedData(state, targetPagePtr, - Min(rest_len, XLOG_BLCKSZ), - false)) + if (state->ReadRecPtr != InvalidXLogRecPtr) { - if (!state->read_page(state, state->readPagePtr, state->readLen, - state->ReadRecPtr, state->readBuf)) - break; + /* read the record after the one we just read */ + + /* + * 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, 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; + } + else + { + /* + * Caller supplied a position to start at. + * + * In this case, EndRecPtr should already be pointing to a + * valid record starting position. + */ + Assert(XRecOffIsValid(state->EndRecPtr)); + state->ReadRecPtr = state->EndRecPtr; + + /* + * 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. + */ + state->PrevRecPtr = InvalidXLogRecPtr; + state->EndRecPtr = InvalidXLogRecPtr; /* to be tidy */ } + state->record_verified = false; + state->readRecordState = XLREAD_TOT_LEN; + /* fall through */ + + case XLREAD_TOT_LEN: + { + uint32 total_len; + uint32 pageHeaderSize; + XLogRecPtr targetPagePtr; + uint32 targetRecOff; + XLogPageHeader pageHeader; + + targetPagePtr = + state->ReadRecPtr - (state->ReadRecPtr % XLOG_BLCKSZ); + targetRecOff = state->ReadRecPtr % XLOG_BLCKSZ; + + /* + * Check if we have enough data. For the first record in the page, + * the requesting length doesn't contain page header. + */ + if (XLogNeedData(state, targetPagePtr, + Min(targetRecOff + SizeOfXLogRecord, XLOG_BLCKSZ), + targetRecOff != 0)) + return XLREAD_NEED_DATA; + + /* error out if caller supplied bogus page */ if (!state->page_verified) goto err; - Assert(SizeOfXLogShortPHD <= state->readLen); + /* examine page header now. */ + pageHeaderSize = + XLogPageHeaderSize((XLogPageHeader) state->readBuf); + if (targetRecOff == 0) + { + /* At page start, so skip over page header. */ + state->ReadRecPtr += pageHeaderSize; + targetRecOff = pageHeaderSize; + } + else if (targetRecOff < pageHeaderSize) + { + report_invalid_record(state, "invalid record offset at %X/%X", + (uint32) (state->ReadRecPtr >> 32), + (uint32) state->ReadRecPtr); + goto err; + } - /* Check that the continuation on next page looks valid */ pageHeader = (XLogPageHeader) state->readBuf; - if (!(pageHeader->xlp_info & XLP_FIRST_IS_CONTRECORD)) + if ((pageHeader->xlp_info & XLP_FIRST_IS_CONTRECORD) && + targetRecOff == pageHeaderSize) { - report_invalid_record(state, - "there is no contrecord flag at %X/%X", - (uint32) (RecPtr >> 32), (uint32) RecPtr); + report_invalid_record(state, "contrecord is requested by %X/%X", + (uint32) (state->ReadRecPtr >> 32), + (uint32) state->ReadRecPtr); goto err; } - /* - * 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)) - { - report_invalid_record(state, - "invalid contrecord length %u at %X/%X", - pageHeader->xlp_rem_len, - (uint32) (RecPtr >> 32), (uint32) RecPtr); - goto err; - } - - /* Append the continuation from this page to the buffer */ - pageHeaderSize = XLogPageHeaderSize(pageHeader); - + /* XLogNeedData has verified the page header */ Assert(pageHeaderSize <= state->readLen); - contdata = (char *) state->readBuf + pageHeaderSize; - len = XLOG_BLCKSZ - pageHeaderSize; - if (pageHeader->xlp_rem_len < len) - len = pageHeader->xlp_rem_len; + /* + * 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. + */ + prec = (XLogRecord *) (state->readBuf + + state->ReadRecPtr % XLOG_BLCKSZ); + total_len = prec->xl_tot_len; - Assert (pageHeaderSize + len <= state->readLen); - memcpy(buffer, (char *) contdata, len); - buffer += len; - gotlen += 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 XLREAD_CONTINUATION state below; + * otherwise we might fail to apply ValidXLogRecordHeader at all. + */ + if (targetRecOff <= XLOG_BLCKSZ - SizeOfXLogRecord) + { + if (!ValidXLogRecordHeader(state, state->ReadRecPtr, + state->PrevRecPtr, prec)) + goto err; - /* If we just reassembled the record header, validate it. */ - if (!gotheader) + state->record_verified = true; + } + else { - record = (XLogRecord *) state->readRecordBuf; - if (!ValidXLogRecordHeader(state, RecPtr, state->ReadRecPtr, - record, randAccess)) + /* 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; - gotheader = true; + } } - } while (gotlen < total_len); - - Assert(gotheader); - record = (XLogRecord *) state->readRecordBuf; - if (!ValidXLogRecord(state, record, RecPtr)) - 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; + state->readRecordState = XLREAD_FIRST_FRAGMENT; + } + /* fall through */ - 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 */ - while (XLogNeedData(state, targetPagePtr, - Min(targetRecOff + total_len, XLOG_BLCKSZ), true)) + case XLREAD_FIRST_FRAGMENT: { - if (!state->read_page(state, state->readPagePtr, state->readLen, - state->ReadRecPtr, state->readBuf)) + uint32 total_len = state->recordRemainLen; + uint32 request_len; + uint32 record_len; + XLogRecPtr targetPagePtr; + uint32 targetRecOff; + + /* + * Wait for the rest of the record on the first page to become + * available + */ + targetPagePtr = + state->ReadRecPtr - (state->ReadRecPtr % XLOG_BLCKSZ); + targetRecOff = state->ReadRecPtr % XLOG_BLCKSZ; + + request_len = Min(targetRecOff + total_len, XLOG_BLCKSZ); + record_len = request_len - targetRecOff; + + /* ReadRecPtr contains page header */ + Assert (targetRecOff != 0); + if (XLogNeedData(state, targetPagePtr, request_len, true)) + return XLREAD_NEED_DATA; + + /* error out if caller supplied bogus page */ + if (!state->page_verified) + goto err; + + prec = (XLogRecord *) (state->readBuf + targetRecOff); + + /* validate record header if not yet */ + if (!state->record_verified && record_len >= SizeOfXLogRecord) + { + if (!ValidXLogRecordHeader(state, state->ReadRecPtr, + state->PrevRecPtr, prec)) + goto err; + + state->record_verified = true; + } + + + if (total_len == record_len) + { + /* Record does not cross a page boundary */ + Assert(state->record_verified); + + if (!ValidXLogRecord(state, prec, state->ReadRecPtr)) + goto err; + + state->record_verified = true; /* to be tidy */ + + /* We already checked the header earlier */ + state->EndRecPtr = state->ReadRecPtr + MAXALIGN(record_len); + + *record = prec; + state->readRecordState = XLREAD_NEXT_RECORD; break; + } + + /* + * The record continues on the next page. Need to reassemble + * record + */ + Assert(total_len > record_len); + + /* 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 + targetRecOff, + record_len); + state->recordGotLen += record_len; + state->recordRemainLen -= record_len; + + /* Calculate pointer to beginning of next page */ + state->recordContRecPtr = state->ReadRecPtr + record_len; + Assert(state->recordContRecPtr % XLOG_BLCKSZ == 0); + + state->readRecordState = XLREAD_CONTINUATION; } + /* fall through */ + + case XLREAD_CONTINUATION: + { + XLogPageHeader pageHeader; + uint32 pageHeaderSize; + XLogRecPtr targetPagePtr; + + /* we enter this state only if we haven't read the whole record. */ + Assert (state->recordRemainLen > 0); + + while(state->recordRemainLen > 0) + { + char *contdata; + uint32 request_len; + uint32 record_len; + + /* Wait for the next page to become available */ + targetPagePtr = state->recordContRecPtr; + + /* this request contains page header */ + Assert (targetPagePtr != 0); + if (XLogNeedData(state, targetPagePtr, + Min(state->recordRemainLen, XLOG_BLCKSZ), + false)) + return XLREAD_NEED_DATA; + + if (!state->page_verified) + goto err; + + Assert(SizeOfXLogShortPHD <= state->readLen); - if (!state->page_verified) - goto err; + /* Check that the continuation on next page looks valid */ + pageHeader = (XLogPageHeader) state->readBuf; + if (!(pageHeader->xlp_info & XLP_FIRST_IS_CONTRECORD)) + { + report_invalid_record( + state, + "there is no contrecord flag at %X/%X reading %X/%X", + (uint32) (state->recordContRecPtr >> 32), + (uint32) state->recordContRecPtr, + (uint32) (state->ReadRecPtr >> 32), + (uint32) state->ReadRecPtr); + goto err; + } - /* Record does not cross a page boundary */ - if (!ValidXLogRecord(state, record, RecPtr)) - goto err; + /* + * 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) + { + report_invalid_record( + state, + "invalid contrecord length %u at %X/%X reading %X/%X, expected %u", + pageHeader->xlp_rem_len, + (uint32) (state->recordContRecPtr >> 32), + (uint32) state->recordContRecPtr, + (uint32) (state->ReadRecPtr >> 32), + (uint32) state->ReadRecPtr, + state->recordRemainLen); + goto err; + } - state->EndRecPtr = RecPtr + MAXALIGN(total_len); + /* Append the continuation from this page to the buffer */ + pageHeaderSize = XLogPageHeaderSize(pageHeader); - state->ReadRecPtr = RecPtr; + /* + * XLogNeedData should have ensured that the whole page header + * was read + */ + Assert(state->readLen >= pageHeaderSize); + + contdata = (char *) state->readBuf + pageHeaderSize; + record_len = XLOG_BLCKSZ - pageHeaderSize; + if (pageHeader->xlp_rem_len < record_len) + record_len = pageHeader->xlp_rem_len; + + request_len = record_len + pageHeaderSize; + + /* XLogNeedData should have ensured all needed data was read */ + Assert (state->readLen >= request_len); + + memcpy(state->readRecordBuf + state->recordGotLen, + (char *) contdata, record_len); + state->recordGotLen += record_len; + state->recordRemainLen -= record_len; + + /* If we just reassembled the record header, validate it. */ + if (!state->record_verified) + { + Assert(state->recordGotLen >= SizeOfXLogRecord); + if (!ValidXLogRecordHeader(state, state->ReadRecPtr, + state->PrevRecPtr, + (XLogRecord *) state->readRecordBuf)) + goto err; + + state->record_verified = true; + } + + /* Calculate pointer to beginning of next page, and continue */ + state->recordContRecPtr += XLOG_BLCKSZ; + } + + /* targetPagePtr is pointing the last-read page here */ + prec = (XLogRecord *) state->readRecordBuf; + if (!ValidXLogRecord(state, prec, state->ReadRecPtr)) + goto err; + + pageHeaderSize = + XLogPageHeaderSize((XLogPageHeader) state->readBuf); + state->EndRecPtr = targetPagePtr + pageHeaderSize + + MAXALIGN(pageHeader->xlp_rem_len); + + *record = prec; + 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->segcxt.ws_segsize - 1; state->EndRecPtr -= XLogSegmentOffset(state->EndRecPtr, state->segcxt.ws_segsize); } - if (DecodeXLogRecord(state, record, errormsg)) - return record; - else - return NULL; + Assert (!*record || state->readLen >= 0); + if (DecodeXLogRecord(state, *record, errormsg)) + return XLREAD_SUCCESS; + + *record = NULL; + return XLREAD_FAIL; err: /* - * Invalidate the read state. We might read from a different source after + * Invalidate the read page. We might read from a different source after * failure. */ XLogReaderInvalReadState(state); @@ -565,7 +700,8 @@ err: if (state->errormsg_buf[0] != '\0') *errormsg = state->errormsg_buf; - return NULL; + *record = NULL; + return XLREAD_FAIL; } /* @@ -718,11 +854,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) { @@ -740,7 +877,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 @@ -972,11 +1109,14 @@ XLogReaderValidatePageHeader(XLogReaderState *state, XLogRecPtr recptr, * XLogReadRecord() will read the next valid record. */ XLogRecPtr -XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr) +XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr, + XLogFindNextRecordCB read_page, void *private) { XLogRecPtr tmpRecPtr; XLogRecPtr found = InvalidXLogRecPtr; XLogPageHeader header; + XLogRecord *record; + XLogReadRecordResult result; char *errormsg; Assert(!XLogRecPtrIsInvalid(RecPtr)); @@ -1009,8 +1149,7 @@ XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr) while(XLogNeedData(state, targetPagePtr, targetRecOff, targetRecOff != 0)) { - if (!state->read_page(state, state->readPagePtr, state->readLen, - state->ReadRecPtr, state->readBuf)) + if (!read_page(state, private)) break; } @@ -1062,8 +1201,16 @@ XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr) * or we just jumped over the remaining data of a continuation. */ XLogBeginRead(state, tmpRecPtr); - while (XLogReadRecord(state, &errormsg) != NULL) + while ((result = XLogReadRecord(state, &record, &errormsg)) != + XLREAD_FAIL) { + if (result == XLREAD_NEED_DATA) + { + if (!read_page(state, private)) + goto err; + continue; + } + /* past the record we've found, break out */ if (RecPtr <= state->ReadRecPtr) { diff --git a/src/backend/access/transam/xlogutils.c b/src/backend/access/transam/xlogutils.c index 2464a0d092..82f1ed2d1a 100644 --- a/src/backend/access/transam/xlogutils.c +++ b/src/backend/access/transam/xlogutils.c @@ -689,8 +689,7 @@ XLogTruncateRelation(RelFileNode rnode, ForkNumber forkNum, void XLogReadDetermineTimeline(XLogReaderState *state, XLogRecPtr wantPage, uint32 wantLength) { - const XLogRecPtr lastReadPage = state->seg.ws_segno * - state->segcxt.ws_segsize + state->readLen; + const XLogRecPtr lastReadPage = state->readPagePtr; Assert(wantPage != InvalidXLogRecPtr && wantPage % XLOG_BLCKSZ == 0); Assert(wantLength <= XLOG_BLCKSZ); @@ -705,7 +704,7 @@ XLogReadDetermineTimeline(XLogReaderState *state, XLogRecPtr wantPage, uint32 wa * current TLI has since become historical. */ if (lastReadPage == wantPage && - state->readLen != 0 && + state->page_verified && lastReadPage + state->readLen >= wantPage + Min(wantLength, XLOG_BLCKSZ - 1)) return; @@ -823,9 +822,11 @@ wal_segment_open(XLogSegNo nextSegNo, WALSegmentContext * segcxt, * loop for now. */ bool -read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, - int reqLen, XLogRecPtr targetRecPtr, char *cur_page) +read_local_xlog_page(XLogReaderState *state) { + XLogRecPtr targetPagePtr = state->readPagePtr; + int reqLen = state->readLen; + char *cur_page = state->readBuf; XLogRecPtr read_upto, loc; TimeLineID tli; @@ -943,6 +944,7 @@ read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, WALReadRaiseError(&errinfo); /* number of valid bytes in the buffer */ + 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 5adf253583..7782e3ad2f 100644 --- a/src/backend/replication/logical/logical.c +++ b/src/backend/replication/logical/logical.c @@ -120,7 +120,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) @@ -169,11 +169,12 @@ StartupDecodingContext(List *output_plugin_options, ctx->slot = slot; - ctx->reader = XLogReaderAllocate(wal_segment_size, NULL, read_page, ctx); + ctx->reader = XLogReaderAllocate(wal_segment_size, NULL); if (!ctx->reader) ereport(ERROR, (errcode(ERRCODE_OUT_OF_MEMORY), errmsg("out of memory"))); + ctx->read_page = read_page; ctx->reorder = ReorderBufferAllocate(); ctx->snapshot_builder = @@ -230,7 +231,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) @@ -372,7 +373,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) @@ -479,7 +480,13 @@ DecodingContextFindStartpoint(LogicalDecodingContext *ctx) char *err = NULL; /* the read_page callback waits for new WAL */ - record = XLogReadRecord(ctx->reader, &err); + while (XLogReadRecord(ctx->reader, &record, &err) == + XLREAD_NEED_DATA) + { + if (!ctx->read_page(ctx->reader)) + break; + } + if (err) elog(ERROR, "%s", err); if (!record) diff --git a/src/backend/replication/logical/logicalfuncs.c b/src/backend/replication/logical/logicalfuncs.c index f5384f1df8..8f2b2dd1fe 100644 --- a/src/backend/replication/logical/logicalfuncs.c +++ b/src/backend/replication/logical/logicalfuncs.c @@ -269,7 +269,13 @@ pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool bin XLogRecord *record; char *errm = NULL; - record = XLogReadRecord(ctx->reader, &errm); + while (XLogReadRecord(ctx->reader, &record, &errm) == + XLREAD_NEED_DATA) + { + if (!ctx->read_page(ctx->reader)) + break; + } + if (errm) elog(ERROR, "%s", errm); diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c index f776de3df7..bd34042458 100644 --- a/src/backend/replication/slotfuncs.c +++ b/src/backend/replication/slotfuncs.c @@ -489,7 +489,13 @@ 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, &errm); + while (XLogReadRecord(ctx->reader, &record, &errm) == + XLREAD_NEED_DATA) + { + if (!ctx->read_page(ctx->reader)) + break; + } + if (errm) elog(ERROR, "%s", errm); diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index e1c1de22c5..0dab43545d 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -799,9 +799,11 @@ StartReplication(StartReplicationCmd *cmd) * set every time WAL is flushed. */ static bool -logical_read_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen, - XLogRecPtr targetRecPtr, char *cur_page) +logical_read_xlog_page(XLogReaderState *state) { + XLogRecPtr targetPagePtr = state->readPagePtr; + int reqLen = state->readLen; + char *cur_page = state->readBuf; XLogRecPtr flushptr; int count; WALReadError errinfo; @@ -2834,7 +2836,12 @@ XLogSendLogical(void) */ WalSndCaughtUp = false; - record = XLogReadRecord(logical_decoding_ctx->reader, &errm); + while (XLogReadRecord(logical_decoding_ctx->reader, &record, &errm) == + XLREAD_NEED_DATA) + { + if (!logical_decoding_ctx->read_page(logical_decoding_ctx->reader)) + 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 1b0199f52f..5b16681a37 100644 --- a/src/bin/pg_rewind/parsexlog.c +++ b/src/bin/pg_rewind/parsexlog.c @@ -40,15 +40,9 @@ static int xlogreadfd = -1; static XLogSegNo xlogreadsegno = -1; static char xlogfpath[MAXPGPATH]; -typedef struct XLogPageReadPrivate -{ - const char *restoreCommand; - int tliIndex; -} XLogPageReadPrivate; - -static bool SimpleXLogPageRead(XLogReaderState *xlogreader, - XLogRecPtr targetPagePtr, - int reqLen, XLogRecPtr targetRecPtr, char *readBuf); +static bool SimpleXLogPageRead(XLogReaderState *xlogreader, + const char *datadir, int *tliIndex, + const char *restoreCommand); /* * Read WAL from the datadir/pg_wal, starting from 'startpoint' on timeline @@ -62,19 +56,22 @@ extractPageMap(const char *datadir, XLogRecPtr startpoint, int tliIndex, XLogRecord *record; XLogReaderState *xlogreader; char *errormsg; - XLogPageReadPrivate private; - private.tliIndex = tliIndex; - private.restoreCommand = restoreCommand; - xlogreader = XLogReaderAllocate(WalSegSz, datadir, &SimpleXLogPageRead, - &private); + xlogreader = XLogReaderAllocate(WalSegSz, datadir); + if (xlogreader == NULL) pg_fatal("out of memory"); XLogBeginRead(xlogreader, startpoint); do { - record = XLogReadRecord(xlogreader, &errormsg); + while (XLogReadRecord(xlogreader, &record, &errormsg) == + XLREAD_NEED_DATA) + { + if (!SimpleXLogPageRead(xlogreader, datadir, + &tliIndex, restoreCommand)) + break; + } if (record == NULL) { @@ -111,17 +108,19 @@ readOneRecord(const char *datadir, XLogRecPtr ptr, int tliIndex) XLogRecord *record; XLogReaderState *xlogreader; char *errormsg; - XLogPageReadPrivate private; XLogRecPtr endptr; - private.tliIndex = tliIndex; - xlogreader = XLogReaderAllocate(WalSegSz, datadir, &SimpleXLogPageRead, - &private); + xlogreader = XLogReaderAllocate(WalSegSz, datadir); if (xlogreader == NULL) pg_fatal("out of memory"); XLogBeginRead(xlogreader, ptr); - record = XLogReadRecord(xlogreader, &errormsg); + while (XLogReadRecord(xlogreader, &record, &errormsg) == + XLREAD_NEED_DATA) + { + if (!SimpleXLogPageRead(xlogreader, datadir, &tliIndex, NULL)) + break; + } if (record == NULL) { if (errormsg) @@ -156,7 +155,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, @@ -172,10 +170,7 @@ findLastCheckpoint(const char *datadir, XLogRecPtr forkptr, int tliIndex, forkptr += SizeOfXLogShortPHD; } - private.tliIndex = tliIndex; - private.restoreCommand = restoreCommand; - xlogreader = XLogReaderAllocate(WalSegSz, datadir, &SimpleXLogPageRead, - &private); + xlogreader = XLogReaderAllocate(WalSegSz, datadir); if (xlogreader == NULL) pg_fatal("out of memory"); @@ -185,7 +180,13 @@ findLastCheckpoint(const char *datadir, XLogRecPtr forkptr, int tliIndex, uint8 info; XLogBeginRead(xlogreader, searchptr); - record = XLogReadRecord(xlogreader, &errormsg); + while (XLogReadRecord(xlogreader, &record, &errormsg) == + XLREAD_NEED_DATA) + { + if (!SimpleXLogPageRead(xlogreader, datadir, + &tliIndex, restoreCommand)) + break; + } if (record == NULL) { @@ -232,10 +233,11 @@ findLastCheckpoint(const char *datadir, XLogRecPtr forkptr, int tliIndex, /* XLogReader callback function, to read a WAL page */ static bool -SimpleXLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr, - int reqLen, XLogRecPtr targetRecPtr, char *readBuf) +SimpleXLogPageRead(XLogReaderState *xlogreader, const char *datadir, + int *tliIndex, const char *restoreCommand) { - XLogPageReadPrivate *private = (XLogPageReadPrivate *) xlogreader->private_data; + XLogRecPtr targetPagePtr = xlogreader->readPagePtr; + char *readBuf = xlogreader->readBuf; uint32 targetPageOff; XLogRecPtr targetSegEnd; XLogSegNo targetSegNo; @@ -268,14 +270,14 @@ 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--; + while (*tliIndex < targetNentries - 1 && + targetHistory[*tliIndex].end < targetSegEnd) + (*tliIndex)++; + while (*tliIndex > 0 && + targetHistory[*tliIndex].begin >= targetSegEnd) + (*tliIndex)--; - XLogFileName(xlogfname, targetHistory[private->tliIndex].tli, + XLogFileName(xlogfname, targetHistory[*tliIndex].tli, xlogreadsegno, WalSegSz); snprintf(xlogfpath, MAXPGPATH, "%s/" XLOGDIR "/%s", @@ -288,7 +290,7 @@ SimpleXLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr, /* * If we have no restore_command to execute, then exit. */ - if (private->restoreCommand == NULL) + if (restoreCommand == NULL) { pg_log_error("could not open file \"%s\": %m", xlogfpath); xlogreader->readLen = -1; @@ -302,7 +304,7 @@ SimpleXLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr, xlogreadfd = RestoreArchivedFile(xlogreader->segcxt.ws_dir, xlogfname, WalSegSz, - private->restoreCommand); + restoreCommand); if (xlogreadfd < 0) { @@ -344,7 +346,7 @@ SimpleXLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr, Assert(targetSegNo == xlogreadsegno); - xlogreader->seg.ws_tli = targetHistory[private->tliIndex].tli; + xlogreader->seg.ws_tli = targetHistory[*tliIndex].tli; xlogreader->readLen = XLOG_BLCKSZ; return true; } diff --git a/src/bin/pg_waldump/pg_waldump.c b/src/bin/pg_waldump/pg_waldump.c index 33fde23e7b..1c42050277 100644 --- a/src/bin/pg_waldump/pg_waldump.c +++ b/src/bin/pg_waldump/pg_waldump.c @@ -324,10 +324,12 @@ WALDumpOpenSegment(XLogSegNo nextSegNo, WALSegmentContext *segcxt, * XLogReader read_page callback */ static bool -WALDumpReadPage(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen, - XLogRecPtr targetPtr, char *readBuff) +WALDumpReadPage(XLogReaderState *state, void *priv) { - XLogDumpPrivate *private = state->private_data; + XLogRecPtr targetPagePtr = state->readPagePtr; + int reqLen = state->readLen; + char *readBuff = state->readBuf; + XLogDumpPrivate *private = (XLogDumpPrivate *) priv; int count = XLOG_BLCKSZ; WALReadError errinfo; @@ -366,6 +368,7 @@ WALDumpReadPage(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen, (Size) errinfo.wre_req); } + Assert(count >= state->readLen); state->readLen = count; return true; } @@ -1033,13 +1036,14 @@ main(int argc, char **argv) /* done with argument parsing, do the actual work */ /* we have everything we need, start reading */ - xlogreader_state = XLogReaderAllocate(WalSegSz, waldir, WALDumpReadPage, - &private); + xlogreader_state = XLogReaderAllocate(WalSegSz, waldir); + if (!xlogreader_state) fatal_error("out of memory"); /* first find a valid recptr to start from */ - first_record = XLogFindNextRecord(xlogreader_state, private.startptr); + first_record = XLogFindNextRecord(xlogreader_state, private.startptr, + &WALDumpReadPage, (void*) &private); if (first_record == InvalidXLogRecPtr) fatal_error("could not find a valid record after %X/%X", @@ -1063,7 +1067,13 @@ main(int argc, char **argv) for (;;) { /* try to read the next record */ - record = XLogReadRecord(xlogreader_state, &errormsg); + while (XLogReadRecord(xlogreader_state, &record, &errormsg) == + XLREAD_NEED_DATA) + { + if (!WALDumpReadPage(xlogreader_state, (void *) &private)) + break; + } + if (!record) { if (!config.follow || private.endptr_reached) diff --git a/src/include/access/xlogreader.h b/src/include/access/xlogreader.h index 6ad953eea3..e59e42bee3 100644 --- a/src/include/access/xlogreader.h +++ b/src/include/access/xlogreader.h @@ -50,13 +50,6 @@ typedef struct WALSegmentContext typedef struct XLogReaderState XLogReaderState; -/* Function type definition for the read_page callback */ -typedef bool (*XLogPageReadCB) (XLogReaderState *xlogreader, - XLogRecPtr targetPagePtr, - int reqLen, - XLogRecPtr targetRecPtr, - char *readBuf); - typedef struct { /* Is this block ref in use? */ @@ -86,6 +79,29 @@ typedef struct uint16 data_bufsz; } DecodedBkpBlock; +/* 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; + +/* + * internal state of XLogReadRecord + * + * XLogReadState runs a state machine while reading a record. Theses states + * are not seen outside the function. Each state may repeat several times + * exiting requesting caller for new data. See the comment of XLogReadRecrod + * for details. + */ +typedef enum XLogReadRecordState { + XLREAD_NEXT_RECORD, + XLREAD_TOT_LEN, + XLREAD_FIRST_FRAGMENT, + XLREAD_CONTINUATION +} XLogReadRecordState; + struct XLogReaderState { /* ---------------------------------------- @@ -93,46 +109,20 @@ struct XLogReaderState * ---------------------------------------- */ - /* - * 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 ->seg.ws_tli to the TLI of the file the page was - * read from. - */ - XLogPageReadCB read_page; - /* * System identifier of the xlog files we're about to read. Set to zero * (the default value) if unknown or unimportant. */ uint64 system_identifier; - /* - * Opaque data for callbacks to use. Not used by XLogReader. - */ - void *private_data; - /* * Start and end point of last record read. EndRecPtr is also used as the * position to read next. Calling XLogBeginRead() sets EndRecPtr to the * starting position and ReadRecPtr to invalid. */ - XLogRecPtr ReadRecPtr; /* start of last record read */ + 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 */ /* ---------------------------------------- * Communication with page reader @@ -146,7 +136,9 @@ struct XLogReaderState * the request, or -1 on error */ TimeLineID readPageTLI; /* TLI for data currently in readBuf */ char *readBuf; /* buffer to store data */ - bool page_verified; /* is the page on the buffer verified? */ + bool page_verified; /* is the page header on the buffer verified? */ + bool record_verified;/* is the current record header verified? */ + /* ---------------------------------------- @@ -186,8 +178,6 @@ struct XLogReaderState XLogRecPtr latestPagePtr; TimeLineID latestPageTLI; - /* beginning of the WAL record being read. */ - XLogRecPtr currRecPtr; /* timeline to read it from, 0 if a lookup is required */ TimeLineID currTLI; @@ -214,15 +204,22 @@ struct XLogReaderState char *readRecordBuf; uint32 readRecordBufSize; + /* + * XLogReadRecord() state + */ + XLogReadRecordState readRecordState;/* state machine state */ + 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; }; /* Get a new XLogReader */ extern XLogReaderState *XLogReaderAllocate(int wal_segment_size, - const char *waldir, - XLogPageReadCB pagereadfunc, - void *private_data); + const char *waldir); /* Free an XLogReader */ extern void XLogReaderFree(XLogReaderState *state); @@ -252,12 +249,17 @@ extern void WALOpenSegmentInit(WALOpenSegment *seg, WALSegmentContext *segcxt, /* Position the XLogReader to given record */ extern void XLogBeginRead(XLogReaderState *state, XLogRecPtr RecPtr); #ifdef FRONTEND -extern XLogRecPtr XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr); +/* Function type definition for the read_page callback */ +typedef bool (*XLogFindNextRecordCB) (XLogReaderState *xlogreader, + void *private); +extern XLogRecPtr XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr, + XLogFindNextRecordCB read_page, void *private); #endif /* FRONTEND */ /* Read the next XLog record. Returns NULL on end-of-WAL or failure */ -extern struct XLogRecord *XLogReadRecord(XLogReaderState *state, - char **errormsg); +extern XLogReadRecordResult XLogReadRecord(XLogReaderState *state, + XLogRecord **record, + char **errormsg); /* Validate a page */ extern bool XLogReaderValidatePageHeader(XLogReaderState *state, diff --git a/src/include/access/xlogutils.h b/src/include/access/xlogutils.h index dc7d894e5d..312acb4fa3 100644 --- a/src/include/access/xlogutils.h +++ b/src/include/access/xlogutils.h @@ -47,9 +47,7 @@ extern Buffer XLogReadBufferExtended(RelFileNode rnode, ForkNumber forknum, extern Relation CreateFakeRelcacheEntry(RelFileNode rnode); extern void FreeFakeRelcacheEntry(Relation fakerel); -extern bool read_local_xlog_page(XLogReaderState *state, - XLogRecPtr targetPagePtr, int reqLen, - XLogRecPtr targetRecPtr, char *cur_page); +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 3b7ca7f1da..cbe9ed751c 100644 --- a/src/include/replication/logical.h +++ b/src/include/replication/logical.h @@ -29,6 +29,10 @@ typedef void (*LogicalOutputPluginWriterUpdateProgress) (struct LogicalDecodingC TransactionId xid ); +typedef struct LogicalDecodingContext LogicalDecodingContext; + +typedef bool (*LogicalDecodingXLogReadPageCB)(XLogReaderState *ctx); + typedef struct LogicalDecodingContext { /* memory context this is all allocated in */ @@ -39,6 +43,7 @@ typedef struct LogicalDecodingContext /* infrastructure pieces for decoding */ XLogReaderState *reader; + LogicalDecodingXLogReadPageCB read_page; struct ReorderBuffer *reorder; struct SnapBuild *snapshot_builder; @@ -95,14 +100,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); -- 2.18.2 ----Next_Part(Tue_Apr_21_17_04_27_2020_275)-- Content-Type: Text/X-Patch; charset=us-ascii Content-Transfer-Encoding: 7bit Content-Disposition: inline; filename="v8-0003-Remove-globals-readOff-readLen-and-readSegNo.patch"