From 6fc92fa74a033c881624177365afbbffc37ed873 Mon Sep 17 00:00:00 2001 From: Jeff Davis 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