From 50bf1ce5dc97508ad74f18011ffc6391e9b3e290 Mon Sep 17 00:00:00 2001 From: Alvaro Herrera Date: Wed, 3 Apr 2024 12:20:01 +0200 Subject: [PATCH v15 3/3] Add Copy pointer to track data copied to WAL buffers --- src/backend/access/transam/xlog.c | 37 +++++++++++++++++++++++++++++-- 1 file changed, 35 insertions(+), 2 deletions(-) diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index e2cd0e8f52..52651be48b 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -469,6 +469,7 @@ typedef struct XLogCtlData XLogRecPtr lastSegSwitchLSN; /* These are accessed using atomics -- info_lck not needed */ + pg_atomic_uint64 logCopyResult; /* last byte + 1 copied to WAL buffers */ pg_atomic_uint64 logWriteResult; /* last byte + 1 written out */ pg_atomic_uint64 logFlushResult; /* last byte + 1 flushed */ @@ -1496,6 +1497,7 @@ static XLogRecPtr WaitXLogInsertionsToFinish(XLogRecPtr upto) { uint64 bytepos; + XLogRecPtr copyptr; XLogRecPtr reservedUpto; XLogRecPtr finishedUpto; XLogCtlInsert *Insert = &XLogCtl->Insert; @@ -1504,6 +1506,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->logCopyResult); + if (upto <= copyptr) + return copyptr; + /* Read the current insert position */ SpinLockAcquire(&Insert->insertpos_lck); bytepos = Insert->CurrBytePos; @@ -1583,6 +1590,9 @@ WaitXLogInsertionsToFinish(XLogRecPtr upto) if (insertingat != InvalidXLogRecPtr && insertingat < finishedUpto) finishedUpto = insertingat; } + + pg_atomic_monotonic_advance_u64(&XLogCtl->logCopyResult, finishedUpto); + return finishedUpto; } @@ -1724,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. + */ + copyptr = pg_atomic_read_membarrier_u64(&XLogCtl->logCopyResult); + if (startptr + count > copyptr) + ereport(ERROR, + errmsg("request to read past end of generated WAL; requested %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 @@ -2563,10 +2584,20 @@ XLogWrite(XLogwrtRqst WriteRqst, TimeLineID tli, bool flexible) #ifdef USE_ASSERT_CHECKING { + XLogRecPtr Copy = pg_atomic_read_membarrier_u64(&XLogCtl->logCopyResult); XLogRecPtr Write = pg_atomic_read_membarrier_u64(&XLogCtl->logWriteResult); XLogRecPtr Flush = pg_atomic_read_membarrier_u64(&XLogCtl->logFlushResult); - /* WAL flushed to disk is always ahead of WAL written */ + /* + * WaitXLogInsertionsToFinish must have been called before us, so + * Copy cannot be invalid. + */ + Assert(!XLogRecPtrIsInvalid(Copy)); + + /* WAL copied to buffers is always ahead of WAL written */ + Assert(Copy >= Write); + + /* WAL written to disk is always ahead of WAL flushed */ Assert(Write >= Flush); } #endif @@ -4941,6 +4972,7 @@ XLOGShmemInit(void) XLogCtl->InstallXLogFileSegmentActive = false; XLogCtl->WalWriterSleeping = false; + pg_atomic_init_u64(&XLogCtl->logCopyResult, InvalidXLogRecPtr); pg_atomic_init_u64(&XLogCtl->logWriteResult, InvalidXLogRecPtr); pg_atomic_init_u64(&XLogCtl->logFlushResult, InvalidXLogRecPtr); @@ -5973,6 +6005,7 @@ StartupXLOG(void) */ LogwrtResult.Write = LogwrtResult.Flush = EndOfLog; + pg_atomic_write_u64(&XLogCtl->logCopyResult, EndOfLog); pg_atomic_write_u64(&XLogCtl->logWriteResult, EndOfLog); pg_atomic_write_u64(&XLogCtl->logFlushResult, EndOfLog); -- 2.39.2