From da650de97d5a4526470e7330976f204e61180f61 Mon Sep 17 00:00:00 2001 From: Bharath Rupireddy Date: Thu, 14 Mar 2024 07:11:43 +0000 Subject: [PATCH v14 3/4] Add Copy pointer to track data copied to WAL buffers --- src/backend/access/transam/xlog.c | 24 +++++++++++++++++++++++- 1 file changed, 23 insertions(+), 1 deletion(-) diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index 762f97b991..3fd71f6787 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -321,6 +321,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; @@ -1497,6 +1498,7 @@ static XLogRecPtr WaitXLogInsertionsToFinish(XLogRecPtr upto) { uint64 bytepos; + XLogRecPtr copyptr; XLogRecPtr reservedUpto; XLogRecPtr finishedUpto; XLogCtlInsert *Insert = &XLogCtl->Insert; @@ -1505,6 +1507,11 @@ WaitXLogInsertionsToFinish(XLogRecPtr upto) if (MyProc == NULL) elog(PANIC, "cannot wait without a PGPROC structure"); + /* check if there's any work to do */ + copyptr = pg_atomic_read_membarrier_u64(&XLogCtl->LogwrtResult.Copy); + if (upto <= copyptr) + return copyptr; + /* Read the current insert position */ SpinLockAcquire(&Insert->insertpos_lck); bytepos = Insert->CurrBytePos; @@ -1584,6 +1591,9 @@ WaitXLogInsertionsToFinish(XLogRecPtr upto) if (insertingat != InvalidXLogRecPtr && insertingat < finishedUpto) finishedUpto = insertingat; } + + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Copy, finishedUpto); + return finishedUpto; } @@ -1725,13 +1735,23 @@ 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. + */ + copyptr = pg_atomic_read_membarrier_u64(&XLogCtl->LogwrtResult.Copy); + if (startptr + count > copyptr) + ereport(ERROR, + (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 @@ -4914,6 +4934,7 @@ XLOGShmemInit(void) XLogCtl->InstallXLogFileSegmentActive = false; XLogCtl->WalWriterSleeping = false; + pg_atomic_init_u64(&XLogCtl->LogwrtResult.Copy, InvalidXLogRecPtr); pg_atomic_init_u64(&XLogCtl->LogwrtResult.Write, InvalidXLogRecPtr); pg_atomic_init_u64(&XLogCtl->LogwrtResult.Flush, InvalidXLogRecPtr); @@ -5940,6 +5961,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