v17-0004-Make-XLogFindNextRecord-not-use-callback-functio.patch

From 14fee3e4a2256b5d47ca7b1b9ff562bc5f487e53 Mon Sep 17 00:00:00 2001
From: Kyotaro Horiguchi <horikyota.ntt@gmail.com>
Date: Wed, 7 Apr 2021 15:32:10 +0900
Subject: [PATCH v17 04/10] 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.30.1