Subject: Re: Improve WALRead() to suck data directly from WAL buffers when possible

From: Jeff Davis <[email-masked]>
Date: Tue, 13 Feb 2024 17:29:47 -0800

Attached 2 patches.

Per Andres's suggestion, 0001 adds an:
  Assert(startptr + count <= LogwrtResult.Write)

Though if we want to allow the caller (e.g. in an extension) to
determine the valid range, perhaps using WaitXLogInsertionsToFinish(),
then the check is wrong. Maybe we should just get rid of that code
entirely and trust the caller to request a reasonable range?

On Mon, 2024-02-12 at 17:33 -0800, Jeff Davis wrote:
> That makes me wonder whether my previous idea[1] might matter: when
> some buffers have been evicted, should WALReadFromBuffers() keep
> going
> through the loop and return the end portion of the requested data
> rather than the beginning?
> [1] 
> https://www.postgresql.org/message-id/[email-masked]

0002 is to illustrate the above idea. It's a strange API so I don't
intend to commit it in this form, but I think we will ultimately need
to do something like it when we want to replicate unflushed data.

The idea is that data past the Write pointer is always (and only)
available in the WAL buffers, so WALReadFromBuffers() should always
return it. That way we can always safely fall through to ordinary
WALRead(), which can only see before the Write pointer. There's also
data before the Write pointer that could be in the WAL buffers, and we
might as well copy that, too, if it's not evicted.

If some buffers are evicted, it will fill in the *end* of the buffer,
leaving a gap at the beginning. The nice thing is that if there is any
gap, it will be before the Write pointer, so we can always fall back to
WALRead() to fill the gap and it should always succeed.

Regards,
	Jeff Davis

From f890362e9f5cefd04ee3f9406c7fcafb6a277e45 Mon Sep 17 00:00:00 2001
From: Jeff Davis <[email-masked]>
Date: Tue, 13 Feb 2024 11:17:08 -0800
Subject: [PATCH 1/2] Add assert to WALReadFromBuffers().

Per suggestion from Andres.
---
 src/backend/access/transam/xlog.c | 24 ++++++------------------
 1 file changed, 6 insertions(+), 18 deletions(-)

diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 4e14c242b1..50c347a679 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -1710,12 +1710,13 @@ GetXLogBuffer(XLogRecPtr ptr, TimeLineID tli)
  * of bytes read successfully.
  *
  * Fewer than 'count' bytes may be read if some of the requested WAL data has
- * already been evicted from the WAL buffers, or if the caller requests data
- * that is not yet available.
+ * already been evicted.
  *
  * No locks are taken.
  *
- * The 'tli' argument is only used as a convenient safety check so that
+ * Caller should ensure that it reads no further than LogwrtResult.Write
+ * (which should have been updated by the caller when determining how far to
+ * read). The 'tli' argument is only used as a convenient safety check so that
  * callers do not read from WAL buffers on a historical timeline.
  */
 Size
@@ -1724,26 +1725,13 @@ WALReadFromBuffers(char *dstbuf, XLogRecPtr startptr, Size count,
 {
 	char	   *pdst = dstbuf;
 	XLogRecPtr	recptr = startptr;
-	XLogRecPtr	upto;
-	Size		nbytes;
+	Size		nbytes = count;
 
 	if (RecoveryInProgress() || tli != GetWALInsertionTimeLine())
 		return 0;
 
 	Assert(!XLogRecPtrIsInvalid(startptr));
-
-	/*
-	 * Don't read past the available WAL data.
-	 *
-	 * Check using local copy of LogwrtResult. Ordinarily it's been updated by
-	 * the caller when determining how far to read; but if not, it just means
-	 * we'll read less data.
-	 *
-	 * XXX: the available WAL could be extended to the WAL insert pointer by
-	 * calling WaitXLogInsertionsToFinish().
-	 */
-	upto = Min(startptr + count, LogwrtResult.Write);
-	nbytes = upto - startptr;
+	Assert(startptr + count <= LogwrtResult.Write);
 
 	/*
 	 * Loop through the buffers without a lock. For each buffer, atomically
-- 
2.34.1

From 6fc92fa74a033c881624177365afbbffc37ed873 Mon Sep 17 00:00:00 2001
From: Jeff Davis <[email-masked]>
Date: Tue, 13 Feb 2024 16:56:46 -0800
Subject: [PATCH 2/2] WALReadFromBuffers: read end of the requested range

---
 src/backend/access/transam/xlog.c   | 109 ++++++++++++++++------------
 src/backend/replication/walsender.c |  14 ++--
 src/include/access/xlog.h           |   2 +-
 3 files changed, 70 insertions(+), 55 deletions(-)

diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 50c347a679..ea55e1b77b 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -1706,11 +1706,21 @@ GetXLogBuffer(XLogRecPtr ptr, TimeLineID tli)
 }
 
 /*
- * Read WAL data directly from WAL buffers, if available. Returns the number
- * of bytes read successfully.
+ * Read WAL data directly from WAL buffers, if available.
  *
- * Fewer than 'count' bytes may be read if some of the requested WAL data has
- * already been evicted.
+ * Some pages in the requested range may already be evicted from the WAL
+ * buffers, in which case this function continues on and reads the *end* of
+ * the range requested (filling in the end of the buffer rather than the
+ * beginning).
+ *
+ * 'count' is an in/out parameter. On return, it's updated to represent the
+ * range of bytes that haven't been filled in. For example:
+ *   remaining = nbytes;
+ *   WALReadFromBuffers(buf, startptr, &remaining, ...);
+ *   WALRead(xlogreader, buf, startptr, remaining, ...);
+ *
+ * On return, startptr + *count <= LogwrtResult.Write, so it will always be
+ * safe to fall back to WALRead().
  *
  * No locks are taken.
  *
@@ -1719,19 +1729,20 @@ GetXLogBuffer(XLogRecPtr ptr, TimeLineID tli)
  * read). The 'tli' argument is only used as a convenient safety check so that
  * callers do not read from WAL buffers on a historical timeline.
  */
-Size
-WALReadFromBuffers(char *dstbuf, XLogRecPtr startptr, Size count,
+void
+WALReadFromBuffers(char *dstbuf, XLogRecPtr startptr, Size *count,
 				   TimeLineID tli)
 {
 	char	   *pdst = dstbuf;
 	XLogRecPtr	recptr = startptr;
-	Size		nbytes = count;
+	XLogRecPtr	valid = startptr;
+	Size		nbytes = *count;
 
 	if (RecoveryInProgress() || tli != GetWALInsertionTimeLine())
-		return 0;
+		return;
 
 	Assert(!XLogRecPtrIsInvalid(startptr));
-	Assert(startptr + count <= LogwrtResult.Write);
+	Assert(startptr + *count <= LogwrtResult.Write);
 
 	/*
 	 * Loop through the buffers without a lock. For each buffer, atomically
@@ -1744,8 +1755,8 @@ WALReadFromBuffers(char *dstbuf, XLogRecPtr startptr, Size count,
 	 * copy. Read barriers are necessary to ensure that the data copy actually
 	 * happens between the two verification steps.
 	 *
-	 * If either verification fails, we simply terminate the loop and return
-	 * with the data that had been already copied out successfully.
+	 * If either verification fails, we advance 'valid' to the next page
+	 * boundary and continue the loop.
 	 */
 	while (nbytes > 0)
 	{
@@ -1753,8 +1764,6 @@ WALReadFromBuffers(char *dstbuf, XLogRecPtr startptr, Size count,
 		int			idx = XLogRecPtrToBufIdx(recptr);
 		XLogRecPtr	expectedEndPtr;
 		XLogRecPtr	endptr;
-		const char *page;
-		const char *psrc;
 		Size		npagebytes;
 
 		/*
@@ -1763,54 +1772,62 @@ WALReadFromBuffers(char *dstbuf, XLogRecPtr startptr, Size count,
 		 */
 		expectedEndPtr = recptr + (XLOG_BLCKSZ - offset);
 
+		/* determine how much data we intend to read from this page */
+		npagebytes = Min(nbytes, XLOG_BLCKSZ - offset);
+
 		/*
 		 * First verification step: check that the correct page is present in
 		 * the WAL buffers.
 		 */
 		endptr = pg_atomic_read_u64(&XLogCtl->xlblocks[idx]);
-		if (expectedEndPtr != endptr)
-			break;
-
-		/*
-		 * The correct page is present (or was at the time the endptr was
-		 * read; must re-verify later). Calculate pointer to source data and
-		 * determine how much data to read from this page.
-		 */
-		page = XLogCtl->pages + idx * (Size) XLOG_BLCKSZ;
-		psrc = page + offset;
-		npagebytes = Min(nbytes, XLOG_BLCKSZ - offset);
+		if (expectedEndPtr == endptr)
+		{
+			/*
+			 * The correct page is present (or was at the time the endptr was
+			 * read; must re-verify later). Calculate pointer to source data.
+			 */
+			const char *page = XLogCtl->pages + idx * (Size) XLOG_BLCKSZ;
+			const char *psrc = page + offset;
 
-		/*
-		 * Ensure that the data copy and the first verification step are not
-		 * reordered.
-		 */
-		pg_read_barrier();
+			/*
+			 * Ensure that the data copy and the first verification step are
+			 * not reordered.
+			 */
+			pg_read_barrier();
 
-		/* data copy */
-		memcpy(pdst, psrc, npagebytes);
+			/* data copy */
+			memcpy(pdst, psrc, npagebytes);
 
-		/*
-		 * Ensure that the data copy and the second verification step are not
-		 * reordered.
-		 */
-		pg_read_barrier();
+			/*
+			 * Ensure that the data copy and the second verification step are
+			 * not reordered.
+			 */
+			pg_read_barrier();
 
-		/*
-		 * Second verification step: check that the page we read from wasn't
-		 * evicted while we were copying the data.
-		 */
-		endptr = pg_atomic_read_u64(&XLogCtl->xlblocks[idx]);
-		if (expectedEndPtr != endptr)
-			break;
+			/*
+			 * Second verification step: check that the page we read from wasn't
+			 * evicted while we were copying the data.
+			 */
+			endptr = pg_atomic_read_u64(&XLogCtl->xlblocks[idx]);
+			if (expectedEndPtr != endptr)
+			{
+				/* discard previously-copied data but keep going */
+				valid = recptr + npagebytes;
+			}
+		}
+		else
+		{
+			/* discard previously-copied data but keep going */
+			valid = recptr + npagebytes;
+		}
 
 		pdst += npagebytes;
 		recptr += npagebytes;
 		nbytes -= npagebytes;
 	}
 
-	Assert(pdst - dstbuf <= count);
-
-	return pdst - dstbuf;
+	*count = valid - startptr;
+	Assert(startptr + *count <= LogwrtResult.Write);
 }
 
 /*
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 146826d5db..f50328c01d 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -2966,7 +2966,7 @@ XLogSendPhysical(void)
 	Size		nbytes;
 	XLogSegNo	segno;
 	WALReadError errinfo;
-	Size		rbytes;
+	Size		bytes_remaining;
 
 	/* If requested switch the WAL sender to the stopping state. */
 	if (got_STOPPING)
@@ -3182,19 +3182,17 @@ XLogSendPhysical(void)
 	enlargeStringInfo(&output_message, nbytes);
 
 retry:
+	bytes_remaining = nbytes;
 	/* attempt to read WAL from WAL buffers first */
-	rbytes = WALReadFromBuffers(&output_message.data[output_message.len],
-								startptr, nbytes, xlogreader->seg.ws_tli);
-	output_message.len += rbytes;
-	startptr += rbytes;
-	nbytes -= rbytes;
+	WALReadFromBuffers(&output_message.data[output_message.len],
+					   startptr, &bytes_remaining, xlogreader->seg.ws_tli);
 
 	/* now read the remaining WAL from WAL file */
-	if (nbytes > 0 &&
+	if (bytes_remaining > 0 &&
 		!WALRead(xlogreader,
 				 &output_message.data[output_message.len],
 				 startptr,
-				 nbytes,
+				 bytes_remaining,
 				 xlogreader->seg.ws_tli,	/* Pass the current TLI because
 											 * only WalSndSegmentOpen controls
 											 * whether new TLI is needed. */
diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h
index 76787a8267..8709e97be0 100644
--- a/src/include/access/xlog.h
+++ b/src/include/access/xlog.h
@@ -252,7 +252,7 @@ extern XLogRecPtr GetLastImportantRecPtr(void);
 
 extern void SetWalWriterSleeping(bool sleeping);
 
-extern Size WALReadFromBuffers(char *dstbuf, XLogRecPtr startptr, Size count,
+extern void WALReadFromBuffers(char *dstbuf, XLogRecPtr startptr, Size *count,
 							   TimeLineID tli);
 
 /*
-- 
2.34.1

---

From: Bharath Rupireddy <[email-masked]>
Date: Fri, 16 Feb 2024 13:08:45 +0530

On Wed, Feb 14, 2024 at 6:59 AM Jeff Davis <[email-masked]> wrote:
> then the check is wrong.

Right.

> Maybe we should just get rid of that code

I'd suggest we strike a balance here - error out in assert builds if
startptr+count is past the current insert position and trust the
callers for production builds. It has a couple of advantages over
doing just Assert(startptr + count <= LogwrtResult.Write):
1) It allows the caller to read unflushed WAL directly from WAL
buffers, see the attached 0005 for an example.
2) All the existing callers where WALReadFromBuffers() is thought to
be used are ensuring WAL availability by reading upto the flush
position so no problem with it.

Also, a note before WALRead() stating the caller must request the WAL
at least that's written out (upto LogwrtResult.Write). I'm not so sure
about this, perhaps, we don't need this comment at all.

Here, I'm with v23 patch set:

0001 - Adds assertion in WALReadFromBuffers() to ensure the requested
WAL isn't beyond the current insert position.
0002 - Adds a new test module to demonstrate how one can use
WALReadFromBuffers() ensuring WaitXLogInsertionsToFinish() if need be.
0003 - Uses WALReadFromBuffers in more places like logical walsenders
and backends.
0004 - Removes zero-padding related stuff as discussed in
https://www.postgresql.org/message-id/CALj2ACWBRFac2TingD3PE3w2EBHXUHY3=[email-masked].
This is needed in this patch set otherwise the assertion added in 0001
fails after 0003.
0005 - Adds a page_read callback for reading from WAL buffers in the
new test module added in 0002. Also, adds tests.

Thoughts?

--
Bharath Rupireddy
PostgreSQL Contributors Team
RDS Open Source Databases
Amazon Web Services: https://aws.amazon.com