v11j-0003-Add-WAL-Copy-pointer-representing-data-copied-t.patch

From f4197c3413fd9c10e956509506b0b0c6354794e1 Mon Sep 17 00:00:00 2001
From: Jeff Davis <jeff@j-davis.com>
Date: Fri, 16 Feb 2024 12:24:13 -0800
Subject: [PATCH v11j 3/3] Add WAL Copy pointer, representing data copied to
 WAL buffers.

---
 src/backend/access/transam/xlog.c | 25 ++++++++++++++++++++++++-
 1 file changed, 24 insertions(+), 1 deletion(-)

diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 0a667a9dea..41a7cb2fa5 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -325,6 +325,7 @@ static bool doPageWrites;
 
 typedef struct XLogwrtAtomic
 {
+	pg_atomic_uint64 Copy;		/* last byte + 1 copied to WAL buffers */
 	pg_atomic_uint64 Write;		/* last byte + 1 written out */
 	pg_atomic_uint64 Flush;		/* last byte + 1 flushed */
 } XLogwrtAtomic;
@@ -1495,6 +1496,7 @@ static XLogRecPtr
 WaitXLogInsertionsToFinish(XLogRecPtr upto)
 {
 	uint64		bytepos;
+	XLogRecPtr	copyptr;
 	XLogRecPtr	reservedUpto;
 	XLogRecPtr	finishedUpto;
 	XLogCtlInsert *Insert = &XLogCtl->Insert;
@@ -1503,6 +1505,12 @@ WaitXLogInsertionsToFinish(XLogRecPtr upto)
 	if (MyProc == NULL)
 		elog(PANIC, "cannot wait without a PGPROC structure");
 
+	/* check if there's any work to do */
+	pg_read_barrier();
+	copyptr = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Copy);
+	if (upto <= copyptr)
+		return copyptr;
+
 	/* Read the current insert position */
 	SpinLockAcquire(&Insert->insertpos_lck);
 	bytepos = Insert->CurrBytePos;
@@ -1582,6 +1590,9 @@ WaitXLogInsertionsToFinish(XLogRecPtr upto)
 		if (insertingat != InvalidXLogRecPtr && insertingat < finishedUpto)
 			finishedUpto = insertingat;
 	}
+
+	pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Copy, finishedUpto);
+
 	return finishedUpto;
 }
 
@@ -1723,13 +1734,24 @@ WALReadFromBuffers(char *dstbuf, XLogRecPtr startptr, Size count,
 {
 	char	   *pdst = dstbuf;
 	XLogRecPtr	recptr = startptr;
+	XLogRecPtr	copyptr;
 	Size		nbytes = count;
 
 	if (RecoveryInProgress() || tli != GetWALInsertionTimeLine())
 		return 0;
 
 	Assert(!XLogRecPtrIsInvalid(startptr));
-	Assert(startptr + count <= LogwrtResult.Write);
+
+	/*
+	 * Caller should ensure that the requested data has been copied to WAL
+	 * buffers before we try to read it.
+	 */
+	pg_read_barrier();
+	copyptr = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Copy);
+	if (startptr + count > copyptr)
+		ereport(WARNING,
+				(errmsg("request to read past end of generated WAL; request %X/%X, current position %X/%X",
+						LSN_FORMAT_ARGS(startptr + count), LSN_FORMAT_ARGS(copyptr))));
 
 	/*
 	 * Loop through the buffers without a lock. For each buffer, atomically
@@ -5933,6 +5955,7 @@ StartupXLOG(void)
 	 * because no other process can be reading or writing WAL yet.
 	 */
 	LogwrtResult.Write = LogwrtResult.Flush = EndOfLog;
+	pg_atomic_write_u64(&XLogCtl->LogwrtResult.Copy, EndOfLog);
 	pg_atomic_write_u64(&XLogCtl->LogwrtResult.Write, EndOfLog);
 	pg_atomic_write_u64(&XLogCtl->LogwrtResult.Flush, EndOfLog);
 	XLogCtl->LogwrtRqst.Write = EndOfLog;
-- 
2.34.1