From: Kyotaro Horiguchi Date: Wed, 7 Apr 2021 15:32:10 +0900 Subject: [PATCH v19 4/5] Make XLogFindNextRecord not use callback function The last function that uses page-read callback is XLogFindNextRecord. Lets make it free from call-back. This also simplifies the interface of WALDumpReadPage. --- src/backend/access/transam/xlogreader.c | 73 ++++++++------- src/bin/pg_waldump/pg_waldump.c | 115 ++++++++++++------------ src/include/access/xlogreader.h | 12 ++- 3 files changed, 110 insertions(+), 90 deletions(-) diff --git a/src/backend/access/transam/xlogreader.c b/src/backend/access/transam/xlogreader.c index 661863e94b..89c59843b9 100644 --- a/src/backend/access/transam/xlogreader.c +++ b/src/backend/access/transam/xlogreader.c @@ -1107,6 +1107,22 @@ XLogReaderValidatePageHeader(XLogReaderState *state, XLogRecPtr recptr, * here. */ +XLogFindNextRecordState * +InitXLogFindNextRecord(XLogReaderState *reader_state, XLogRecPtr start_ptr) +{ + XLogFindNextRecordState *state = (XLogFindNextRecordState *) + palloc_extended(sizeof(XLogFindNextRecordState), + MCXT_ALLOC_NO_OOM | MCXT_ALLOC_ZERO); + if (!state) + return NULL; + + state->reader_state = reader_state; + state->targetRecPtr = start_ptr; + state->currRecPtr = start_ptr; + + return state; +} + /* * Find the first record with an lsn >= RecPtr. * @@ -1118,24 +1134,21 @@ XLogReaderValidatePageHeader(XLogReaderState *state, XLogRecPtr recptr, * This positions the reader, like XLogBeginRead(), so that the next call to * XLogReadRecord() will read the next valid record. */ -XLogRecPtr -XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr, - XLogFindNextRecordCB read_page, void *private) +bool +XLogFindNextRecord(XLogFindNextRecordState *state) { - XLogRecPtr tmpRecPtr; XLogRecPtr found = InvalidXLogRecPtr; XLogPageHeader header; XLogRecord *record; XLogReadRecordResult result; char *errormsg; - Assert(!XLogRecPtrIsInvalid(RecPtr)); + Assert(!XLogRecPtrIsInvalid(state->currRecPtr)); /* * skip over potential continuation data, keeping in mind that it may span * multiple pages */ - tmpRecPtr = RecPtr; while (true) { XLogRecPtr targetPagePtr; @@ -1151,27 +1164,24 @@ XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr, * XLogNeedData() is prepared to handle that and will read at least * short page-header worth of data */ - targetRecOff = tmpRecPtr % XLOG_BLCKSZ; + targetRecOff = state->currRecPtr % XLOG_BLCKSZ; /* scroll back to page boundary */ - targetPagePtr = tmpRecPtr - targetRecOff; + targetPagePtr = state->currRecPtr - targetRecOff; - while (XLogNeedData(state, targetPagePtr, targetRecOff, + if (XLogNeedData(state->reader_state, targetPagePtr, targetRecOff, targetRecOff != 0)) - { - if (!read_page(state, private)) - break; - } + return true; - if (!state->page_verified) + if (!state->reader_state->page_verified) goto err; - header = (XLogPageHeader) state->readBuf; + header = (XLogPageHeader) state->reader_state->readBuf; pageHeaderSize = XLogPageHeaderSize(header); /* we should have read the page header */ - Assert(state->readLen >= pageHeaderSize); + Assert(state->reader_state->readLen >= pageHeaderSize); /* skip over potential continuation data */ if (header->xlp_info & XLP_FIRST_IS_CONTRECORD) @@ -1186,21 +1196,21 @@ XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr, * Note that record headers are MAXALIGN'ed */ if (MAXALIGN(header->xlp_rem_len) >= (XLOG_BLCKSZ - pageHeaderSize)) - tmpRecPtr = targetPagePtr + XLOG_BLCKSZ; + state->currRecPtr = targetPagePtr + XLOG_BLCKSZ; else { /* * The previous continuation record ends in this page. Set - * tmpRecPtr to point to the first valid record + * state->currRecPtr to point to the first valid record */ - tmpRecPtr = targetPagePtr + pageHeaderSize + state->currRecPtr = targetPagePtr + pageHeaderSize + MAXALIGN(header->xlp_rem_len); break; } } else { - tmpRecPtr = targetPagePtr + pageHeaderSize; + state->currRecPtr = targetPagePtr + pageHeaderSize; break; } } @@ -1210,31 +1220,28 @@ XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr, * 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. */ - XLogBeginRead(state, tmpRecPtr); - while ((result = XLogReadRecord(state, &record, &errormsg)) != + XLogBeginRead(state->reader_state, state->currRecPtr); + while ((result = XLogReadRecord(state->reader_state, &record, &errormsg)) != XLREAD_FAIL) { if (result == XLREAD_NEED_DATA) - { - if (!read_page(state, private)) - goto err; - continue; - } + return true; /* past the record we've found, break out */ - if (RecPtr <= state->ReadRecPtr) + if (state->targetRecPtr <= state->reader_state->ReadRecPtr) { /* Rewind the reader to the beginning of the last record. */ - found = state->ReadRecPtr; - XLogBeginRead(state, found); - return found; + state->currRecPtr = state->reader_state->ReadRecPtr; + XLogBeginRead(state->reader_state, found); + return false; } } err: - XLogReaderInvalReadState(state); + XLogReaderInvalReadState(state->reader_state); - return InvalidXLogRecPtr; + state->currRecPtr = InvalidXLogRecPtr;; + return false; } #endif /* FRONTEND */ diff --git a/src/bin/pg_waldump/pg_waldump.c b/src/bin/pg_waldump/pg_waldump.c index c4047b92b5..ab2d079bdb 100644 --- a/src/bin/pg_waldump/pg_waldump.c +++ b/src/bin/pg_waldump/pg_waldump.c @@ -29,14 +29,6 @@ static const char *progname; static int WalSegSz; -typedef struct XLogDumpPrivate -{ - TimeLineID timeline; - XLogRecPtr startptr; - XLogRecPtr endptr; - bool endptr_reached; -} XLogDumpPrivate; - typedef struct XLogDumpConfig { /* display options */ @@ -331,35 +323,40 @@ WALDumpCloseSegment(XLogReaderState *state) } /* - * pg_waldump's WAL page rader, also used as page_read callback for - * XLogFindNextRecord + * pg_waldump's WAL page rader + * + * timeline and startptr specifies the LSN, and reads up to endptr. */ static bool -WALDumpReadPage(XLogReaderState *state, void *priv) +WALDumpReadPage(XLogReaderState *state, TimeLineID timeline, + XLogRecPtr startptr, XLogRecPtr endptr) { XLogRecPtr targetPagePtr = state->readPagePtr; int reqLen = state->readLen; char *readBuff = state->readBuf; - XLogDumpPrivate *private = (XLogDumpPrivate *) priv; int count = XLOG_BLCKSZ; WALReadError errinfo; - if (private->endptr != InvalidXLogRecPtr) + /* determine the number of bytes to read on the page */ + if (endptr != InvalidXLogRecPtr) { - if (targetPagePtr + XLOG_BLCKSZ <= private->endptr) + if (targetPagePtr + XLOG_BLCKSZ <= endptr) count = XLOG_BLCKSZ; - else if (targetPagePtr + reqLen <= private->endptr) - count = private->endptr - targetPagePtr; + else if (targetPagePtr + reqLen <= endptr) + count = endptr - targetPagePtr; else { - private->endptr_reached = true; + /* Notify xlogreader that we didn't read at all */ state->readLen = -1; return false; } } + /* We should read more than requested by xlogreader */ + Assert(count >= state->readLen); + if (!WALRead(state, WALDumpOpenSegment, WALDumpCloseSegment, - readBuff, targetPagePtr, count, private->timeline, &errinfo)) + readBuff, targetPagePtr, count, timeline, &errinfo)) { WALOpenSegment *seg = &errinfo.wre_seg; char fname[MAXPGPATH]; @@ -379,7 +376,7 @@ WALDumpReadPage(XLogReaderState *state, void *priv) (Size) errinfo.wre_req); } - Assert(count >= state->readLen); + /* Notify xlogreader of how many bytes we have read */ state->readLen = count; return true; } @@ -762,7 +759,10 @@ main(int argc, char **argv) uint32 xlogid; uint32 xrecoff; XLogReaderState *xlogreader_state; - XLogDumpPrivate private; + XLogFindNextRecordState *findnext_state; + TimeLineID timeline; + XLogRecPtr startptr; + XLogRecPtr endptr; XLogDumpConfig config; XLogDumpStats stats; XLogRecord *record; @@ -808,14 +808,9 @@ main(int argc, char **argv) } } - memset(&private, 0, sizeof(XLogDumpPrivate)); - memset(&config, 0, sizeof(XLogDumpConfig)); - memset(&stats, 0, sizeof(XLogDumpStats)); - - private.timeline = 1; - private.startptr = InvalidXLogRecPtr; - private.endptr = InvalidXLogRecPtr; - private.endptr_reached = false; + timeline = 1; + startptr = InvalidXLogRecPtr; + endptr = InvalidXLogRecPtr; config.quiet = false; config.bkp_details = false; @@ -849,7 +844,7 @@ main(int argc, char **argv) optarg); goto bad_argument; } - private.endptr = (uint64) xlogid << 32 | xrecoff; + endptr = (uint64) xlogid << 32 | xrecoff; break; case 'f': config.follow = true; @@ -902,10 +897,10 @@ main(int argc, char **argv) goto bad_argument; } else - private.startptr = (uint64) xlogid << 32 | xrecoff; + startptr = (uint64) xlogid << 32 | xrecoff; break; case 't': - if (sscanf(optarg, "%d", &private.timeline) != 1) + if (sscanf(optarg, "%d", &timeline) != 1) { pg_log_error("could not parse timeline \"%s\"", optarg); goto bad_argument; @@ -982,21 +977,21 @@ main(int argc, char **argv) close(fd); /* parse position from file */ - XLogFromFileName(fname, &private.timeline, &segno, WalSegSz); + XLogFromFileName(fname, &timeline, &segno, WalSegSz); - if (XLogRecPtrIsInvalid(private.startptr)) - XLogSegNoOffsetToRecPtr(segno, 0, WalSegSz, private.startptr); - else if (!XLByteInSeg(private.startptr, segno, WalSegSz)) + if (XLogRecPtrIsInvalid(startptr)) + XLogSegNoOffsetToRecPtr(segno, 0, WalSegSz, startptr); + else if (!XLByteInSeg(startptr, segno, WalSegSz)) { pg_log_error("start WAL location %X/%X is not inside file \"%s\"", - LSN_FORMAT_ARGS(private.startptr), + LSN_FORMAT_ARGS(startptr), fname); goto bad_argument; } /* no second file specified, set end position */ - if (!(optind + 1 < argc) && XLogRecPtrIsInvalid(private.endptr)) - XLogSegNoOffsetToRecPtr(segno + 1, 0, WalSegSz, private.endptr); + if (!(optind + 1 < argc) && XLogRecPtrIsInvalid(endptr)) + XLogSegNoOffsetToRecPtr(segno + 1, 0, WalSegSz, endptr); /* parse ENDSEG if passed */ if (optind + 1 < argc) @@ -1012,26 +1007,26 @@ main(int argc, char **argv) close(fd); /* parse position from file */ - XLogFromFileName(fname, &private.timeline, &endsegno, WalSegSz); + XLogFromFileName(fname, &timeline, &endsegno, WalSegSz); if (endsegno < segno) fatal_error("ENDSEG %s is before STARTSEG %s", argv[optind + 1], argv[optind]); - if (XLogRecPtrIsInvalid(private.endptr)) + if (XLogRecPtrIsInvalid(endptr)) XLogSegNoOffsetToRecPtr(endsegno + 1, 0, WalSegSz, - private.endptr); + endptr); /* set segno to endsegno for check of --end */ segno = endsegno; } - if (!XLByteInSeg(private.endptr, segno, WalSegSz) && - private.endptr != (segno + 1) * WalSegSz) + if (!XLByteInSeg(endptr, segno, WalSegSz) && + endptr != (segno + 1) * WalSegSz) { pg_log_error("end WAL location %X/%X is not inside file \"%s\"", - LSN_FORMAT_ARGS(private.endptr), + LSN_FORMAT_ARGS(endptr), argv[argc - 1]); goto bad_argument; } @@ -1040,7 +1035,7 @@ main(int argc, char **argv) waldir = identify_target_directory(waldir, NULL); /* we don't know what to print */ - if (XLogRecPtrIsInvalid(private.startptr)) + if (XLogRecPtrIsInvalid(startptr)) { pg_log_error("no start WAL location given"); goto bad_argument; @@ -1055,27 +1050,37 @@ main(int argc, char **argv) if (!xlogreader_state) fatal_error("out of memory"); + findnext_state = + InitXLogFindNextRecord(xlogreader_state, startptr); + + if (!findnext_state) + fatal_error("out of memory"); + /* first find a valid recptr to start from */ - first_record = XLogFindNextRecord(xlogreader_state, private.startptr, - &WALDumpReadPage, (void*) &private); + while (XLogFindNextRecord(findnext_state)) + { + if (!WALDumpReadPage(xlogreader_state, timeline, startptr, endptr)) + break; + } + first_record = findnext_state->currRecPtr; if (first_record == InvalidXLogRecPtr) fatal_error("could not find a valid record after %X/%X", - LSN_FORMAT_ARGS(private.startptr)); + LSN_FORMAT_ARGS(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) + if (first_record != startptr && + XLogSegmentOffset(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)), - LSN_FORMAT_ARGS(private.startptr), + (first_record - startptr)), + LSN_FORMAT_ARGS(startptr), LSN_FORMAT_ARGS(first_record), - (uint32) (first_record - private.startptr)); + (uint32) (first_record - startptr)); for (;;) { @@ -1083,13 +1088,13 @@ main(int argc, char **argv) while (XLogReadRecord(xlogreader_state, &record, &errormsg) == XLREAD_NEED_DATA) { - if (!WALDumpReadPage(xlogreader_state, (void *) &private)) + if (!WALDumpReadPage(xlogreader_state, timeline, startptr, endptr)) break; } if (!record) { - if (!config.follow || private.endptr_reached) + if (!config.follow) break; else { diff --git a/src/include/access/xlogreader.h b/src/include/access/xlogreader.h index 1492f1992d..d8cb488820 100644 --- a/src/include/access/xlogreader.h +++ b/src/include/access/xlogreader.h @@ -56,6 +56,7 @@ typedef struct WALSegmentContext } WALSegmentContext; typedef struct XLogReaderState XLogReaderState; +typedef struct XLogFindNextRecordState XLogFindNextRecordState; /* Function type definition for the segment cleanup callback */ typedef void (*WALSegmentCleanupCB) (XLogReaderState *xlogreader); @@ -240,6 +241,13 @@ struct XLogReaderState char *errormsg_buf; }; +struct XLogFindNextRecordState +{ + XLogReaderState *reader_state; + XLogRecPtr targetRecPtr; + XLogRecPtr currRecPtr; +}; + /* Get a new XLogReader */ extern XLogReaderState *XLogReaderAllocate(int wal_segment_size, const char *waldir, @@ -254,8 +262,8 @@ extern void XLogBeginRead(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); +extern XLogFindNextRecordState *InitXLogFindNextRecord(XLogReaderState *reader_state, XLogRecPtr start_ptr); +extern bool XLogFindNextRecord(XLogFindNextRecordState *state); #endif /* FRONTEND */ /* Read the next XLog record. Returns NULL on end-of-WAL or failure */ -- 2.27.0 ----Next_Part(Wed_Apr__7_17_50_25_2021_942)-- Content-Type: Text/X-Patch; charset=us-ascii Content-Transfer-Encoding: 7bit Content-Disposition: inline; filename="v19-0005-Split-readLen-and-reqLen-of-XLogReaderState.patch"