On Wed, Sep 10, 2025 at 4:24 AM Chao Li <li.evan.c...@gmail.com> wrote:
>
Thanks for the review!
For any of your feedback that I simply implemented, I omitted an
inline comment about it. Those changes are included in the attached
v6. My inline replies below are only for feedback requiring more
discussion.
> On Sep 10, 2025, at 01:55, Melanie Plageman <melanieplage...@gmail.com> wrote:
>
> 2 - 0001
> ```
> --- a/src/backend/storage/buffer/freelist.c
> +++ b/src/backend/storage/buffer/freelist.c
>
> + if (XLogNeedsFlush(lsn))
> + {
> + /*
> + * Remove the dirty buffer from the ring; necessary to prevent an
> + * infinite loop if all ring members are dirty.
> + */
> + strategy->buffers[strategy->current] = InvalidBuffer;
> + return true;
> + }
>
> - return true;
> + return false;
> }
> ```
>
> We can do:
> ```
> If (!XLogNeedsFlush(lan))
> Return false
>
> /* Remove the dirty buffer ….
> */
> Return true;
> }
> ```
This would make the order of evaluation the same as master but I
actually prefer it this way because then we only take the buffer
header spinlock if there is a chance we will reject the buffer (e.g.
we don't need to examine it for strategies except BAS_BULKREAD)
> 4 - 0002
> ```
> - /* OK, do the I/O */
> - FlushBuffer(buf_hdr, NULL, IOOBJECT_RELATION, io_context);
> - LWLockRelease(content_lock);
> -
> - ScheduleBufferTagForWriteback(&BackendWritebackContext, io_context,
> - &buf_hdr->tag);
> + CleanVictimBuffer(buf_hdr, &buf_state, from_ring, io_context);
> ```
> I saw CleanVictimBuffer() will get content_lock from bufdesc and release it,
> but it makes the code hard to understand. Readers might be confused that why
> content_lock is not released after CleanVictimBuffer() without further
> reading CleanVictimBuffer().
>
> I’d suggest pass content_lock to CleanVictimBuffer() as a parameter, which
> gives a clear hint that CleanVictimBuffer() will release the lock.
I think for this specific patch in the set your idea makes sense.
However, in the later patch to do write combining, I release the
content locks for the batch in CompleteWriteBatchIO() and having the
start buffer's lock separate as a parameter would force me to have a
special case handling this.
I've added a comment to both CleanVictimBuffer() and its caller
specifying that the lock must be held and that it will be released
inside CleanVictimBuffer.
> 5 - 0002
> ```
> * disastrous system-wide consequences. To make sure that can't happen,
> * skip the flush if the buffer isn't permanent.
> */
> - if (buf_state & BM_PERMANENT)
> - XLogFlush(recptr);
> + if (!XLogRecPtrIsInvalid(buffer_lsn))
> + XLogFlush(buffer_lsn);
> ```
>
> Why this check is changed? Should the comment be updated accordingly as it
> says “if the buffer isn’t permanent”, which reflects to the old code.
It's changed because I split the logic for flushing to that LSN and
determining the LSN across the Prepare and Do functions. This is
needed because when we do batches, we want to flush to the max LSN
across all buffers in the batch.
I check if the buffer is BM_PERMANENT in PrepareFlushBuffer(). You
make a good point about my comment, though. I've moved it to
PrepareFlushBuffer() and updated it.
> 8 - 0003
> ```
> bool
> +strategy_supports_eager_flush(BufferAccessStrategy strategy)
> ```
>
> This function is only used in bufmgr.c, can we move it there and make it
> static?
BufferAccessStrategyData is opaque to bufmgr.c. Only freelist.c can
access it. I agree it is gross that I have these helpers and functions
that would otherwise be static, though.
> 10 - 0004
> ```
> --- a/src/backend/storage/buffer/bufmgr.c
> +++ b/src/backend/storage/buffer/bufmgr.c
>
> + limit = Min(max_batch_size, limit);
> ```
>
> Do we need to check max_batch_size should be less than
> (MAX_IO_COMBINE_LIMIT-1)? Because BufWriteBatch.bufdescs is defined with
> length of MAX_IO_COMBINE_LIMIT, and the first place has been used to store
> “start”.
I assert that in StrategyMaxWriteBatchSize(). io_combine_limit is not
allowed to exceed MAX_IO_COMBINE_LIMIT, so it shouldn't happen anyway,
since we are capping ourselves at io_combine_limit. Or is that your
point?
> 11 - 0004
> ```
> --- a/src/backend/storage/buffer/bufmgr.c
> +++ b/src/backend/storage/buffer/bufmgr.c
>
> + for (batch->n = 1; batch->n < limit; batch->n++)
> + {
> + Buffer bufnum;
> +
> + if ((bufnum = StrategySweepNextBuffer(strategy)) == InvalidBuffer)
> + break;
> ```
>
> Is sweep next buffer right next to start? If yes, can we assert that? But my
> guess is no, if my guess is true, then is it possible that bufnum meets
> start? If that’s true, then we should check next buffer doesn’t equal to
> start.
Ah, great point. I didn't think about this. Our sweep will always
start right after the start buffer, but then if it goes all the way
around, it will "lap" the start buffer. Because of this and because I
think it is weird to have the sweep variables in the
BufferAccessStrategy object, I've changed my approach in attached v6.
I set sweep_end to be the start block in the batch and then pass
around a sweep cursor variable. Hitting sweep_end is the termination
condition.
> 12 - 0004
> ```
> @@ -4306,19 +4370,22 @@ CleanVictimBuffer(BufferAccessStrategy strategy,
>
> if (from_ring && strategy_supports_eager_flush(strategy))
> {
> + uint32 max_batch_size = max_write_batch_size_for_strategy(strategy);
> ```
>
> I think max_batch_size can be attribute of strategy and set it when creating
> a strategy, so that we don’t need to calculate in every round of clean.
Actually, the max pin limit can change quite frequently. See
GetAdditionalPinLimit()'s usage in read stream code. If the query is
pinning other buffers in another part of the query, it can change our
limit.
I'm not sure if I should call GetAdditionalPinLImit() for each batch
or for each run of batches (like in StrategyMaxWriteBatchSize()).
Currently, I call it for each batch (in FindFlushAdjacents()). The
read stream calls it pretty frequently (each
read_stream_start_pending_read()). But, in the batch flush case,
nothing could change between batches in a run of batches. So maybe I
should move it up and out and make it per run of batches...
> 13 - 0004
> ```
> +void
> +CompleteWriteBatchIO(BufWriteBatch *batch, IOContext io_context,
> + WritebackContext *wb_context)
> +{
> + ErrorContextCallback errcallback =
> + {
> + .callback = shared_buffer_write_error_callback,
> + .previous = error_context_stack,
> + };
> +
> + error_context_stack = &errcallback;
> + pgBufferUsage.shared_blks_written += batch->n;
> ```
>
> Should we only increase shared_blks_written only after the loop of write-back
> is done?
On master, FlushBuffer() does it after smgrwrite() (before writeback).
I think pgBufferUsage is mainly used in EXPLAIN (also
pg_stat_statements) which won't be used until the end of the query and
won't be displayed if we error out.
> 14 - 0004
> ```
> --- a/src/backend/storage/buffer/freelist.c
> +++ b/src/backend/storage/buffer/freelist.c
>
> +uint32
> +max_write_batch_size_for_strategy(BufferAccessStrategy strategy)
> ```
>
> I think this function can be moved to bufmgr.c and make it static.
This technically could be moved, but it is a function giving you
information about a strategy which seemed to fit better in freelist.c.
> 18 - 0007
> ```
> --- a/src/backend/storage/buffer/bufmgr.c
> +++ b/src/backend/storage/buffer/bufmgr.c
>
> + max_batch_size = checkpointer_max_batch_size();
> ```
>
> Look like we don’t need to calculate max_batch_size in the for loop.
I don't think it's in the for loop.
- Melanie
From c782753a430c1c967125509c6390d4e710fd2a63 Mon Sep 17 00:00:00 2001
From: Melanie Plageman <melanieplage...@gmail.com>
Date: Tue, 2 Sep 2025 11:32:24 -0400
Subject: [PATCH v6 2/9] Split FlushBuffer() into two parts
Before adding write combining to write a batch of blocks when flushing
dirty buffers, refactor FlushBuffer() into the preparatory step and
actual buffer flushing step. This separation procides symmetry with
future code for batch flushing which necessarily separates these steps,
as it must prepare multiple buffers before flushing them together.
These steps are moved into a new FlushBuffer() helper function,
CleanVictimBuffer() which will contain both the batch flushing and
single flush code in future commits.
Author: Melanie Plageman <melanieplage...@gmail.com>
Reviewed-by: Chao Li <li.evan.c...@gmail.com>
Reviewed-by: Nazir Bilal Yavuz <byavu...@gmail.com>
Discussion: https://postgr.es/m/2FA0BAC7-5413-4ABD-94CA-4398FE77750D%40gmail.com
Discussion: https://postgr.es/m/flat/CAAKRu_Yjn4mvN9NBxtmsCQSGwup45CoA4e05nhR7ADP-v0WCig%40mail.gmail.com
---
src/backend/storage/buffer/bufmgr.c | 141 +++++++++++++++++++---------
1 file changed, 98 insertions(+), 43 deletions(-)
diff --git a/src/backend/storage/buffer/bufmgr.c b/src/backend/storage/buffer/bufmgr.c
index f3668051574..f40f57e5582 100644
--- a/src/backend/storage/buffer/bufmgr.c
+++ b/src/backend/storage/buffer/bufmgr.c
@@ -529,8 +529,13 @@ static inline BufferDesc *BufferAlloc(SMgrRelation smgr,
static bool AsyncReadBuffers(ReadBuffersOperation *operation, int *nblocks_progress);
static void CheckReadBuffersOperation(ReadBuffersOperation *operation, bool is_complete);
static Buffer GetVictimBuffer(BufferAccessStrategy strategy, IOContext io_context);
+static bool PrepareFlushBuffer(BufferDesc *bufdesc, uint32 *buf_state, XLogRecPtr *lsn);
+static void DoFlushBuffer(BufferDesc *buf, SMgrRelation reln, IOObject io_object,
+ IOContext io_context, XLogRecPtr buffer_lsn);
static void FlushBuffer(BufferDesc *buf, SMgrRelation reln,
IOObject io_object, IOContext io_context);
+static void CleanVictimBuffer(BufferDesc *bufdesc, uint32 *buf_state,
+ bool from_ring, IOContext io_context);
static void FindAndDropRelationBuffers(RelFileLocator rlocator,
ForkNumber forkNum,
BlockNumber nForkBlock,
@@ -2414,12 +2419,8 @@ GetVictimBuffer(BufferAccessStrategy strategy, IOContext io_context)
continue;
}
- /* OK, do the I/O */
- FlushBuffer(buf_hdr, NULL, IOOBJECT_RELATION, io_context);
- LWLockRelease(content_lock);
-
- ScheduleBufferTagForWriteback(&BackendWritebackContext, io_context,
- &buf_hdr->tag);
+ /* Content lock is released inside CleanVictimBuffer */
+ CleanVictimBuffer(buf_hdr, &buf_state, from_ring, io_context);
}
if (buf_state & BM_VALID)
@@ -4246,53 +4247,66 @@ static void
FlushBuffer(BufferDesc *buf, SMgrRelation reln, IOObject io_object,
IOContext io_context)
{
- XLogRecPtr recptr;
- ErrorContextCallback errcallback;
- instr_time io_start;
- Block bufBlock;
- char *bufToWrite;
uint32 buf_state;
+ XLogRecPtr lsn;
- /*
- * Try to start an I/O operation. If StartBufferIO returns false, then
- * someone else flushed the buffer before we could, so we need not do
- * anything.
- */
- if (!StartBufferIO(buf, false, false))
- return;
+ if (PrepareFlushBuffer(buf, &buf_state, &lsn))
+ DoFlushBuffer(buf, reln, io_object, io_context, lsn);
+}
- /* Setup error traceback support for ereport() */
- errcallback.callback = shared_buffer_write_error_callback;
- errcallback.arg = buf;
- errcallback.previous = error_context_stack;
- error_context_stack = &errcallback;
+/*
+ * Prepare and write out a dirty victim buffer.
+ *
+ * Buffer must be pinned, the content lock must be held exclusively, and the
+ * buffer header spinlock must not be held. The exclusive lock is released and
+ * the buffer is returned pinned but not locked.
+ *
+ * bufdesc and buf_state may be modified.
+ */
+static void
+CleanVictimBuffer(BufferDesc *bufdesc, uint32 *buf_state,
+ bool from_ring, IOContext io_context)
+{
- /* Find smgr relation for buffer */
- if (reln == NULL)
- reln = smgropen(BufTagGetRelFileLocator(&buf->tag), INVALID_PROC_NUMBER);
+ XLogRecPtr max_lsn = InvalidXLogRecPtr;
+ LWLock *content_lock;
- TRACE_POSTGRESQL_BUFFER_FLUSH_START(BufTagGetForkNum(&buf->tag),
- buf->tag.blockNum,
- reln->smgr_rlocator.locator.spcOid,
- reln->smgr_rlocator.locator.dbOid,
- reln->smgr_rlocator.locator.relNumber);
+ Assert(*buf_state & BM_DIRTY);
- buf_state = LockBufHdr(buf);
+ /* Set up this victim buffer to be flushed */
+ if (!PrepareFlushBuffer(bufdesc, buf_state, &max_lsn))
+ return;
+ DoFlushBuffer(bufdesc, NULL, IOOBJECT_RELATION, io_context, max_lsn);
+ content_lock = BufferDescriptorGetContentLock(bufdesc);
+ LWLockRelease(content_lock);
+ ScheduleBufferTagForWriteback(&BackendWritebackContext, io_context,
+ &bufdesc->tag);
+}
+
+/*
+ * Prepare the buffer with budesc for writing. buf_state and lsn are output
+ * parameters. Returns true if the buffer acutally needs writing and false
+ * otherwise. All three parameters may be modified.
+ */
+static bool
+PrepareFlushBuffer(BufferDesc *bufdesc, uint32 *buf_state, XLogRecPtr *lsn)
+{
/*
- * Run PageGetLSN while holding header lock, since we don't have the
- * buffer locked exclusively in all cases.
+ * Try to start an I/O operation. If StartBufferIO returns false, then
+ * someone else flushed the buffer before we could, so we need not do
+ * anything.
*/
- recptr = BufferGetLSN(buf);
+ if (!StartBufferIO(bufdesc, false, false))
+ return false;
- /* To check if block content changes while flushing. - vadim 01/17/97 */
- buf_state &= ~BM_JUST_DIRTIED;
- UnlockBufHdr(buf, buf_state);
+ *lsn = InvalidXLogRecPtr;
+ *buf_state = LockBufHdr(bufdesc);
/*
- * Force XLOG flush up to buffer's LSN. This implements the basic WAL
- * rule that log updates must hit disk before any of the data-file changes
- * they describe do.
+ * Record the buffer's LSN. We will force XLOG flush up to buffer's LSN.
+ * This implements the basic WAL rule that log updates must hit disk
+ * before any of the data-file changes they describe do.
*
* However, this rule does not apply to unlogged relations, which will be
* lost after a crash anyway. Most unlogged relation pages do not bear
@@ -4305,9 +4319,50 @@ FlushBuffer(BufferDesc *buf, SMgrRelation reln, IOObject io_object,
* happen, attempting to flush WAL through that location would fail, with
* disastrous system-wide consequences. To make sure that can't happen,
* skip the flush if the buffer isn't permanent.
+ *
+ * We must hold the buffer header lock when examining the page LSN since
+ * don't have buffer exclusively locked in all cases.
*/
- if (buf_state & BM_PERMANENT)
- XLogFlush(recptr);
+ if (*buf_state & BM_PERMANENT)
+ *lsn = BufferGetLSN(bufdesc);
+
+ /* To check if block content changes while flushing. - vadim 01/17/97 */
+ *buf_state &= ~BM_JUST_DIRTIED;
+ UnlockBufHdr(bufdesc, *buf_state);
+ return true;
+}
+
+/*
+ * Actually do the write I/O to clean a buffer. buf and reln may be modified.
+ */
+static void
+DoFlushBuffer(BufferDesc *buf, SMgrRelation reln, IOObject io_object,
+ IOContext io_context, XLogRecPtr buffer_lsn)
+{
+ ErrorContextCallback errcallback;
+ instr_time io_start;
+ Block bufBlock;
+ char *bufToWrite;
+
+ /* Setup error traceback support for ereport() */
+ errcallback.callback = shared_buffer_write_error_callback;
+ errcallback.arg = buf;
+ errcallback.previous = error_context_stack;
+ error_context_stack = &errcallback;
+
+ /* Find smgr relation for buffer */
+ if (reln == NULL)
+ reln = smgropen(BufTagGetRelFileLocator(&buf->tag), INVALID_PROC_NUMBER);
+
+ TRACE_POSTGRESQL_BUFFER_FLUSH_START(BufTagGetForkNum(&buf->tag),
+ buf->tag.blockNum,
+ reln->smgr_rlocator.locator.spcOid,
+ reln->smgr_rlocator.locator.dbOid,
+ reln->smgr_rlocator.locator.relNumber);
+
+ /* Force XLOG flush up to buffer's LSN */
+ if (!XLogRecPtrIsInvalid(buffer_lsn))
+ XLogFlush(buffer_lsn);
/*
* Now it's safe to write the buffer to disk. Note that no one else should
--
2.43.0
From dbcc430c4b92c2a69f84fe9ab3faa94f61eb3d99 Mon Sep 17 00:00:00 2001
From: Melanie Plageman <melanieplage...@gmail.com>
Date: Tue, 2 Sep 2025 12:43:24 -0400
Subject: [PATCH v6 3/9] Eagerly flush bulkwrite strategy ring
Operations using BAS_BULKWRITE (COPY FROM and createdb) will inevitably
need to flush buffers in the strategy ring in order to reuse them. By
eagerly flushing the buffers in a larger run, we encourage larger writes
at the kernel level and less interleaving of WAL flushes and data file
writes. The effect is mainly noticeable with multiple parallel COPY
FROMs. In this case, client backends achieve higher write throughput and
end up spending less time waiting on acquiring the lock to flush WAL.
Larger flush operations also mean less time waiting for flush operations
at the kernel level.
The heuristic for eager eviction is to only flush buffers in the
strategy ring which do not require a WAL flush.
This patch also is a step toward AIO writes.
Reviewed-by: Chao Li <li.evan.c...@gmail.com>
Reviewed-by: Nazir Bilal Yavuz <byavu...@gmail.com>
Earlier version Reviewed-by: Kirill Reshke <reshkekir...@gmail.com>
Discussion: https://postgr.es/m/2FA0BAC7-5413-4ABD-94CA-4398FE77750D%40gmail.com
Discussion: https://postgr.es/m/flat/CAAKRu_Yjn4mvN9NBxtmsCQSGwup45CoA4e05nhR7ADP-v0WCig%40mail.gmail.com
---
src/backend/storage/buffer/bufmgr.c | 189 +++++++++++++++++++++++++-
src/backend/storage/buffer/freelist.c | 48 +++++++
src/include/storage/buf_internals.h | 4 +
3 files changed, 235 insertions(+), 6 deletions(-)
diff --git a/src/backend/storage/buffer/bufmgr.c b/src/backend/storage/buffer/bufmgr.c
index f40f57e5582..c64268f31ae 100644
--- a/src/backend/storage/buffer/bufmgr.c
+++ b/src/backend/storage/buffer/bufmgr.c
@@ -534,7 +534,16 @@ static void DoFlushBuffer(BufferDesc *buf, SMgrRelation reln, IOObject io_object
IOContext io_context, XLogRecPtr buffer_lsn);
static void FlushBuffer(BufferDesc *buf, SMgrRelation reln,
IOObject io_object, IOContext io_context);
-static void CleanVictimBuffer(BufferDesc *bufdesc, uint32 *buf_state,
+static BufferDesc *NextStratBufToFlush(BufferAccessStrategy strategy,
+ Buffer sweep_end,
+ XLogRecPtr *lsn,
+ int *sweep_cursor);
+static BufferDesc *PrepareOrRejectEagerFlushBuffer(Buffer bufnum, BlockNumber require,
+ RelFileLocator *rlocator,
+ bool skip_pinned,
+ XLogRecPtr *max_lsn);
+static void CleanVictimBuffer(BufferAccessStrategy strategy,
+ BufferDesc *bufdesc, uint32 *buf_state,
bool from_ring, IOContext io_context);
static void FindAndDropRelationBuffers(RelFileLocator rlocator,
ForkNumber forkNum,
@@ -2420,7 +2429,7 @@ GetVictimBuffer(BufferAccessStrategy strategy, IOContext io_context)
}
/* Content lock is released inside CleanVictimBuffer */
- CleanVictimBuffer(buf_hdr, &buf_state, from_ring, io_context);
+ CleanVictimBuffer(strategy, buf_hdr, &buf_state, from_ring, io_context);
}
if (buf_state & BM_VALID)
@@ -4254,6 +4263,40 @@ FlushBuffer(BufferDesc *buf, SMgrRelation reln, IOObject io_object,
DoFlushBuffer(buf, reln, io_object, io_context, lsn);
}
+/*
+ * Returns the buffer descriptor of the buffer containing the next block we
+ * should eagerly flush or NULL when there are no further buffers to consider
+ * writing out.
+ */
+static BufferDesc *
+NextStratBufToFlush(BufferAccessStrategy strategy,
+ Buffer sweep_end,
+ XLogRecPtr *lsn, int *sweep_cursor)
+{
+ Buffer bufnum;
+ BufferDesc *bufdesc;
+
+ while ((bufnum =
+ StrategySweepNextBuffer(strategy, sweep_cursor)) != sweep_end)
+ {
+ /*
+ * For BAS_BULKWRITE, once you hit an InvalidBuffer, the remaining
+ * buffers in the ring will be invalid.
+ */
+ if (!BufferIsValid(bufnum))
+ break;
+
+ if ((bufdesc = PrepareOrRejectEagerFlushBuffer(bufnum,
+ InvalidBlockNumber,
+ NULL,
+ true,
+ lsn)) != NULL)
+ return bufdesc;
+ }
+
+ return NULL;
+}
+
/*
* Prepare and write out a dirty victim buffer.
*
@@ -4264,12 +4307,14 @@ FlushBuffer(BufferDesc *buf, SMgrRelation reln, IOObject io_object,
* bufdesc and buf_state may be modified.
*/
static void
-CleanVictimBuffer(BufferDesc *bufdesc, uint32 *buf_state,
+CleanVictimBuffer(BufferAccessStrategy strategy,
+ BufferDesc *bufdesc, uint32 *buf_state,
bool from_ring, IOContext io_context)
{
XLogRecPtr max_lsn = InvalidXLogRecPtr;
LWLock *content_lock;
+ bool first_buffer = true;
Assert(*buf_state & BM_DIRTY);
@@ -4277,11 +4322,143 @@ CleanVictimBuffer(BufferDesc *bufdesc, uint32 *buf_state,
if (!PrepareFlushBuffer(bufdesc, buf_state, &max_lsn))
return;
- DoFlushBuffer(bufdesc, NULL, IOOBJECT_RELATION, io_context, max_lsn);
+ if (from_ring && StrategySupportsEagerFlush(strategy))
+ {
+ Buffer sweep_end = BufferDescriptorGetBuffer(bufdesc);
+ int cursor = StrategySweepStart(strategy);
+
+ /* Clean victim buffer and find more to flush opportunistically */
+ do
+ {
+ DoFlushBuffer(bufdesc, NULL, IOOBJECT_RELATION, io_context, max_lsn);
+ content_lock = BufferDescriptorGetContentLock(bufdesc);
+ LWLockRelease(content_lock);
+ ScheduleBufferTagForWriteback(&BackendWritebackContext, io_context,
+ &bufdesc->tag);
+ /* We leave the first buffer pinned for the caller */
+ if (!first_buffer)
+ UnpinBuffer(bufdesc);
+ first_buffer = false;
+ } while ((bufdesc = NextStratBufToFlush(strategy, sweep_end,
+ &max_lsn, &cursor)) != NULL);
+ }
+ else
+ {
+ DoFlushBuffer(bufdesc, NULL, IOOBJECT_RELATION, io_context, max_lsn);
+ content_lock = BufferDescriptorGetContentLock(bufdesc);
+ LWLockRelease(content_lock);
+ ScheduleBufferTagForWriteback(&BackendWritebackContext, io_context,
+ &bufdesc->tag);
+ }
+}
+
+/*
+ * Prepare bufdesc for eager flushing.
+ *
+ * Given bufnum, return the block -- the pointer to the block data in memory
+ * -- which we will opportunistically flush or NULL if this buffer does not
+ * contain a block that should be flushed.
+ *
+ * require is the BlockNumber required by the caller. Some callers may require
+ * a specific BlockNumber to be in bufnum because they are assembling a
+ * contiguous run of blocks.
+ *
+ * If the caller needs the block to be from a specific relation, rlocator will
+ * be provided.
+ */
+BufferDesc *
+PrepareOrRejectEagerFlushBuffer(Buffer bufnum, BlockNumber require,
+ RelFileLocator *rlocator, bool skip_pinned,
+ XLogRecPtr *max_lsn)
+{
+ BufferDesc *bufdesc;
+ uint32 buf_state;
+ XLogRecPtr lsn;
+ BlockNumber blknum;
+ LWLock *content_lock;
+
+ if (!BufferIsValid(bufnum))
+ return NULL;
+
+ Assert(!BufferIsLocal(bufnum));
+
+ bufdesc = GetBufferDescriptor(bufnum - 1);
+
+ /* Block may need to be in a specific relation */
+ if (rlocator &&
+ !RelFileLocatorEquals(BufTagGetRelFileLocator(&bufdesc->tag),
+ *rlocator))
+ return NULL;
+
+ /* Must do this before taking the buffer header spinlock */
+ ResourceOwnerEnlarge(CurrentResourceOwner);
+ ReservePrivateRefCountEntry();
+
+ buf_state = LockBufHdr(bufdesc);
+
+ if (!(buf_state & BM_DIRTY) || !(buf_state & BM_VALID))
+ goto except_unlock_header;
+
+ /* We don't eagerly flush buffers used by others */
+ if (skip_pinned &&
+ (BUF_STATE_GET_REFCOUNT(buf_state) > 0 ||
+ BUF_STATE_GET_USAGECOUNT(buf_state) > 1))
+ goto except_unlock_header;
+
+ /* Get page LSN while holding header lock */
+ lsn = BufferGetLSN(bufdesc);
+
+ PinBuffer_Locked(bufdesc);
+ CheckBufferIsPinnedOnce(bufnum);
+
+ blknum = BufferGetBlockNumber(bufnum);
+ Assert(BlockNumberIsValid(blknum));
+
+ /* If we'll have to flush WAL to flush the block, we're done */
+ if (buf_state & BM_PERMANENT && XLogNeedsFlush(lsn))
+ goto except_unpin_buffer;
+
+ /* We only include contiguous blocks in the run */
+ if (BlockNumberIsValid(require) && blknum != require)
+ goto except_unpin_buffer;
+
content_lock = BufferDescriptorGetContentLock(bufdesc);
+ if (!LWLockConditionalAcquire(content_lock, LW_SHARED))
+ goto except_unpin_buffer;
+
+ /*
+ * Now that we have the content lock, we need to recheck if we need to
+ * flush WAL.
+ */
+ buf_state = LockBufHdr(bufdesc);
+ lsn = BufferGetLSN(bufdesc);
+ UnlockBufHdr(bufdesc, buf_state);
+
+ if (buf_state & BM_PERMANENT && XLogNeedsFlush(lsn))
+ goto except_unlock_content;
+
+ /* Try to start an I/O operation */
+ if (!StartBufferIO(bufdesc, false, true))
+ goto except_unlock_content;
+
+ if (lsn > *max_lsn)
+ *max_lsn = lsn;
+ buf_state = LockBufHdr(bufdesc);
+ buf_state &= ~BM_JUST_DIRTIED;
+ UnlockBufHdr(bufdesc, buf_state);
+
+ return bufdesc;
+
+except_unlock_content:
LWLockRelease(content_lock);
- ScheduleBufferTagForWriteback(&BackendWritebackContext, io_context,
- &bufdesc->tag);
+
+except_unpin_buffer:
+ UnpinBuffer(bufdesc);
+ return NULL;
+
+except_unlock_header:
+ UnlockBufHdr(bufdesc, buf_state);
+ return NULL;
}
/*
diff --git a/src/backend/storage/buffer/freelist.c b/src/backend/storage/buffer/freelist.c
index 12bb7e2312e..8716109221b 100644
--- a/src/backend/storage/buffer/freelist.c
+++ b/src/backend/storage/buffer/freelist.c
@@ -156,6 +156,31 @@ ClockSweepTick(void)
return victim;
}
+/*
+ * Some BufferAccessStrategies support eager flushing -- which is flushing
+ * buffers in the ring before they are needed. This can lead to better I/O
+ * patterns than lazily flushing buffers immediately before reusing them.
+ */
+bool
+StrategySupportsEagerFlush(BufferAccessStrategy strategy)
+{
+ Assert(strategy);
+
+ switch (strategy->btype)
+ {
+ case BAS_BULKWRITE:
+ return true;
+ case BAS_VACUUM:
+ case BAS_NORMAL:
+ case BAS_BULKREAD:
+ return false;
+ default:
+ elog(ERROR, "unrecognized buffer access strategy: %d",
+ (int) strategy->btype);
+ return false;
+ }
+}
+
/*
* StrategyGetBuffer
*
@@ -270,6 +295,29 @@ StrategyGetBuffer(BufferAccessStrategy strategy, uint32 *buf_state, bool *from_r
}
}
+/*
+ * Return the next buffer in the ring or InvalidBuffer if the current sweep is
+ * over.
+ */
+Buffer
+StrategySweepNextBuffer(BufferAccessStrategy strategy, int *sweep_cursor)
+{
+ if (++(*sweep_cursor) >= strategy->nbuffers)
+ *sweep_cursor = 0;
+
+ return strategy->buffers[*sweep_cursor];
+}
+
+/*
+ * Return the starting buffer of a sweep of the strategy ring
+ */
+int
+StrategySweepStart(BufferAccessStrategy strategy)
+{
+ return strategy->current;
+}
+
+
/*
* StrategySyncStart -- tell BgBufferSync where to start syncing
*
diff --git a/src/include/storage/buf_internals.h b/src/include/storage/buf_internals.h
index b1b81f31419..03faf80e441 100644
--- a/src/include/storage/buf_internals.h
+++ b/src/include/storage/buf_internals.h
@@ -437,6 +437,10 @@ extern void TerminateBufferIO(BufferDesc *buf, bool clear_dirty, uint32 set_flag
/* freelist.c */
+extern bool StrategySupportsEagerFlush(BufferAccessStrategy strategy);
+extern Buffer StrategySweepNextBuffer(BufferAccessStrategy strategy,
+ int *sweep_cursor);
+extern int StrategySweepStart(BufferAccessStrategy strategy);
extern IOContext IOContextForStrategy(BufferAccessStrategy strategy);
extern BufferDesc *StrategyGetBuffer(BufferAccessStrategy strategy,
uint32 *buf_state, bool *from_ring);
--
2.43.0
From 9a3f592e5dc7933decad22747d0a4335429d2170 Mon Sep 17 00:00:00 2001
From: Melanie Plageman <melanieplage...@gmail.com>
Date: Tue, 2 Sep 2025 10:01:17 -0400
Subject: [PATCH v6 5/9] Fix XLogNeedsFlush() for checkpointer
In normal operation, XLogNeedsFlush() returns true if the flush ptr has
not been advanced past the provided LSN. During normal recovery on a
standby (not crash recovery), it return true if the minimum recovery
point hasn't been advanced past the provided LSN.
However, during an end-of-recovery checkpoint, the checkpointer flushes
WAL, so XLogNeedsFlush() should compare the provided location with the
flush pointer.
Correct the logic in XLogNeedsFlush() to compare the LSN to the flush
pointer when WAL inserts are allowed and the minimum recovery point
otherwise.
This is not an active bug because no current users of XLogNeedsFlush()
temporarily allowed WAL inserts during recovery.
Author: Melanie Plageman <melanieplage...@gmail.com>
Reviewed-by: Chao Li <li.evan.c...@gmail.com>
Reviewed-by: Jeff Davis <pg...@j-davis.com>
Reviewed-by: Michael Paquier <mich...@paquier.xyz>
Discussion: https://postgr.es/m/flat/CAAKRu_a1vZRZRWO3_jv_X13RYoqLRVipGO0237g5PKzPa2YX6g%40mail.gmail.com
Discussion: https://postgr.es/m/flat/CAAKRu_bcWRvRwZUop_d9vzF9nHAiT%2B-uPzkJ%3DS3ShZ1GqeAYOw%40mail.gmail.com
---
src/backend/access/transam/xlog.c | 19 ++++++++++++-------
1 file changed, 12 insertions(+), 7 deletions(-)
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 0baf0ac6160..62923d33b79 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -3102,21 +3102,26 @@ XLogBackgroundFlush(void)
}
/*
- * Test whether XLOG data has been flushed up to (at least) the given position.
+ * Test whether XLOG data has been flushed up to (at least) the given position
+ * or whether the minimum recovery point is updated past the given position.
*
- * Returns true if a flush is still needed. (It may be that someone else
- * is already in process of flushing that far, however.)
+ * Returns true if a flush is still needed or if the minimum recovery point
+ * must be updated.
+ *
+ * It is possible that someone else is already in the process of flushing that
+ * far or updating the minimum recovery point that far.
*/
bool
XLogNeedsFlush(XLogRecPtr record)
{
/*
- * During recovery, we don't flush WAL but update minRecoveryPoint
- * instead. So "needs flush" is taken to mean whether minRecoveryPoint
- * would need to be updated.
+ * During recovery, when WAL inserts are forbidden, "needs flush" is taken
+ * to mean whether minRecoveryPoint would need to be updated.
*/
- if (RecoveryInProgress())
+ if (!XLogInsertAllowed())
{
+ Assert(RecoveryInProgress());
+
/*
* An invalid minRecoveryPoint means that we need to recover all the
* WAL, i.e., we're doing crash recovery. We never modify the control
--
2.43.0
From 6c46b33c7a51990f1d2df0fab7dfea2f88e0861e Mon Sep 17 00:00:00 2001
From: Melanie Plageman <melanieplage...@gmail.com>
Date: Tue, 2 Sep 2025 11:00:44 -0400
Subject: [PATCH v6 1/9] Refactor goto into for loop in GetVictimBuffer()
GetVictimBuffer() implemented a loop to optimistically lock a clean
victim buffer using a goto. Future commits will add batch flushing
functionality to GetVictimBuffer. The new logic works better with
standard for loop flow control.
This commit is only a refactor and does not introduce any new
functionality.
Author: Melanie Plageman <melanieplage...@gmail.com>
Reviewed-by: Nazir Bilal Yavuz <byavu...@gmail.com>
Reviewed-by: Chao Li <li.evan.c...@gmail.com>
Discussion: https://postgr.es/m/2FA0BAC7-5413-4ABD-94CA-4398FE77750D%40gmail.com
Discussion: https://postgr.es/m/flat/CAAKRu_Yjn4mvN9NBxtmsCQSGwup45CoA4e05nhR7ADP-v0WCig%40mail.gmail.com
---
src/backend/storage/buffer/bufmgr.c | 200 ++++++++++++--------------
src/backend/storage/buffer/freelist.c | 32 ++++-
src/include/storage/buf_internals.h | 5 +
3 files changed, 124 insertions(+), 113 deletions(-)
diff --git a/src/backend/storage/buffer/bufmgr.c b/src/backend/storage/buffer/bufmgr.c
index fe470de63f2..f3668051574 100644
--- a/src/backend/storage/buffer/bufmgr.c
+++ b/src/backend/storage/buffer/bufmgr.c
@@ -68,10 +68,6 @@
#include "utils/timestamp.h"
-/* Note: these two macros only work on shared buffers, not local ones! */
-#define BufHdrGetBlock(bufHdr) ((Block) (BufferBlocks + ((Size) (bufHdr)->buf_id) * BLCKSZ))
-#define BufferGetLSN(bufHdr) (PageGetLSN(BufHdrGetBlock(bufHdr)))
-
/* Note: this macro only works on local buffers, not shared ones! */
#define LocalBufHdrGetBlock(bufHdr) \
LocalBufferBlockPointers[-((bufHdr)->buf_id + 2)]
@@ -2344,130 +2340,122 @@ GetVictimBuffer(BufferAccessStrategy strategy, IOContext io_context)
ReservePrivateRefCountEntry();
ResourceOwnerEnlarge(CurrentResourceOwner);
- /* we return here if a prospective victim buffer gets used concurrently */
-again:
-
- /*
- * Select a victim buffer. The buffer is returned with its header
- * spinlock still held!
- */
- buf_hdr = StrategyGetBuffer(strategy, &buf_state, &from_ring);
- buf = BufferDescriptorGetBuffer(buf_hdr);
-
- Assert(BUF_STATE_GET_REFCOUNT(buf_state) == 0);
-
- /* Pin the buffer and then release the buffer spinlock */
- PinBuffer_Locked(buf_hdr);
-
- /*
- * We shouldn't have any other pins for this buffer.
- */
- CheckBufferIsPinnedOnce(buf);
-
- /*
- * If the buffer was dirty, try to write it out. There is a race
- * condition here, in that someone might dirty it after we released the
- * buffer header lock above, or even while we are writing it out (since
- * our share-lock won't prevent hint-bit updates). We will recheck the
- * dirty bit after re-locking the buffer header.
- */
- if (buf_state & BM_DIRTY)
+ /* Select a victim buffer using an optimistic locking scheme. */
+ for (;;)
{
- LWLock *content_lock;
+ /*
+ * Attempt to claim a victim buffer. The buffer is returned with its
+ * header spinlock still held!
+ */
+ buf_hdr = StrategyGetBuffer(strategy, &buf_state, &from_ring);
+ buf = BufferDescriptorGetBuffer(buf_hdr);
- Assert(buf_state & BM_TAG_VALID);
- Assert(buf_state & BM_VALID);
+ Assert(BUF_STATE_GET_REFCOUNT(buf_state) == 0);
+
+ /* Pin the buffer and then release the buffer spinlock */
+ PinBuffer_Locked(buf_hdr);
/*
- * We need a share-lock on the buffer contents to write it out (else
- * we might write invalid data, eg because someone else is compacting
- * the page contents while we write). We must use a conditional lock
- * acquisition here to avoid deadlock. Even though the buffer was not
- * pinned (and therefore surely not locked) when StrategyGetBuffer
- * returned it, someone else could have pinned and exclusive-locked it
- * by the time we get here. If we try to get the lock unconditionally,
- * we'd block waiting for them; if they later block waiting for us,
- * deadlock ensues. (This has been observed to happen when two
- * backends are both trying to split btree index pages, and the second
- * one just happens to be trying to split the page the first one got
- * from StrategyGetBuffer.)
+ * We shouldn't have any other pins for this buffer.
*/
- content_lock = BufferDescriptorGetContentLock(buf_hdr);
- if (!LWLockConditionalAcquire(content_lock, LW_SHARED))
- {
- /*
- * Someone else has locked the buffer, so give it up and loop back
- * to get another one.
- */
- UnpinBuffer(buf_hdr);
- goto again;
- }
+ CheckBufferIsPinnedOnce(buf);
/*
- * If using a nondefault strategy, and writing the buffer would
- * require a WAL flush, let the strategy decide whether to go ahead
- * and write/reuse the buffer or to choose another victim. We need a
- * lock to inspect the page LSN, so this can't be done inside
- * StrategyGetBuffer.
+ * If the buffer was dirty, try to write it out. There is a race
+ * condition here, in that someone might dirty it after we released
+ * the buffer header lock above, or even while we are writing it out
+ * (since our share-lock won't prevent hint-bit updates). We will
+ * recheck the dirty bit after re-locking the buffer header.
*/
- if (strategy != NULL)
+ if (buf_state & BM_DIRTY)
{
- XLogRecPtr lsn;
+ LWLock *content_lock;
- /* Read the LSN while holding buffer header lock */
- buf_state = LockBufHdr(buf_hdr);
- lsn = BufferGetLSN(buf_hdr);
- UnlockBufHdr(buf_hdr, buf_state);
+ Assert(buf_state & BM_TAG_VALID);
+ Assert(buf_state & BM_VALID);
- if (XLogNeedsFlush(lsn)
- && StrategyRejectBuffer(strategy, buf_hdr, from_ring))
+ /*
+ * We need a share-lock on the buffer contents to write it out
+ * (else we might write invalid data, eg because someone else is
+ * compacting the page contents while we write). We must use a
+ * conditional lock acquisition here to avoid deadlock. Even
+ * though the buffer was not pinned (and therefore surely not
+ * locked) when StrategyGetBuffer returned it, someone else could
+ * have pinned and exclusive-locked it by the time we get here. If
+ * we try to get the lock unconditionally, we'd block waiting for
+ * them; if they later block waiting for us, deadlock ensues.
+ * (This has been observed to happen when two backends are both
+ * trying to split btree index pages, and the second one just
+ * happens to be trying to split the page the first one got from
+ * StrategyGetBuffer.)
+ */
+ content_lock = BufferDescriptorGetContentLock(buf_hdr);
+ if (!LWLockConditionalAcquire(content_lock, LW_SHARED))
+ {
+ /*
+ * Someone else has locked the buffer, so give it up and loop
+ * back to get another one.
+ */
+ UnpinBuffer(buf_hdr);
+ continue;
+ }
+
+ /*
+ * If using a nondefault strategy, and writing the buffer would
+ * require a WAL flush, let the strategy decide whether to go
+ * ahead and write/reuse the buffer or to choose another victim.
+ * We need the content lock to inspect the page LSN, so this can't
+ * be done inside StrategyGetBuffer.
+ */
+ if (StrategyRejectBuffer(strategy, buf_hdr, from_ring))
{
LWLockRelease(content_lock);
UnpinBuffer(buf_hdr);
- goto again;
+ continue;
}
- }
- /* OK, do the I/O */
- FlushBuffer(buf_hdr, NULL, IOOBJECT_RELATION, io_context);
- LWLockRelease(content_lock);
+ /* OK, do the I/O */
+ FlushBuffer(buf_hdr, NULL, IOOBJECT_RELATION, io_context);
+ LWLockRelease(content_lock);
- ScheduleBufferTagForWriteback(&BackendWritebackContext, io_context,
- &buf_hdr->tag);
- }
+ ScheduleBufferTagForWriteback(&BackendWritebackContext, io_context,
+ &buf_hdr->tag);
+ }
+ if (buf_state & BM_VALID)
+ {
+ /*
+ * When a BufferAccessStrategy is in use, blocks evicted from
+ * shared buffers are counted as IOOP_EVICT in the corresponding
+ * context (e.g. IOCONTEXT_BULKWRITE). Shared buffers are evicted
+ * by a strategy in two cases: 1) while initially claiming buffers
+ * for the strategy ring 2) to replace an existing strategy ring
+ * buffer because it is pinned or in use and cannot be reused.
+ *
+ * Blocks evicted from buffers already in the strategy ring are
+ * counted as IOOP_REUSE in the corresponding strategy context.
+ *
+ * At this point, we can accurately count evictions and reuses,
+ * because we have successfully claimed the valid buffer.
+ * Previously, we may have been forced to release the buffer due
+ * to concurrent pinners or erroring out.
+ */
+ pgstat_count_io_op(IOOBJECT_RELATION, io_context,
+ from_ring ? IOOP_REUSE : IOOP_EVICT, 1, 0);
+ }
- if (buf_state & BM_VALID)
- {
/*
- * When a BufferAccessStrategy is in use, blocks evicted from shared
- * buffers are counted as IOOP_EVICT in the corresponding context
- * (e.g. IOCONTEXT_BULKWRITE). Shared buffers are evicted by a
- * strategy in two cases: 1) while initially claiming buffers for the
- * strategy ring 2) to replace an existing strategy ring buffer
- * because it is pinned or in use and cannot be reused.
- *
- * Blocks evicted from buffers already in the strategy ring are
- * counted as IOOP_REUSE in the corresponding strategy context.
- *
- * At this point, we can accurately count evictions and reuses,
- * because we have successfully claimed the valid buffer. Previously,
- * we may have been forced to release the buffer due to concurrent
- * pinners or erroring out.
+ * If the buffer has an entry in the buffer mapping table, delete it.
+ * This can fail because another backend could have pinned or dirtied
+ * the buffer. Then loop around and try again.
*/
- pgstat_count_io_op(IOOBJECT_RELATION, io_context,
- from_ring ? IOOP_REUSE : IOOP_EVICT, 1, 0);
- }
+ if ((buf_state & BM_TAG_VALID) && !InvalidateVictimBuffer(buf_hdr))
+ {
+ UnpinBuffer(buf_hdr);
+ continue;
+ }
- /*
- * If the buffer has an entry in the buffer mapping table, delete it. This
- * can fail because another backend could have pinned or dirtied the
- * buffer.
- */
- if ((buf_state & BM_TAG_VALID) && !InvalidateVictimBuffer(buf_hdr))
- {
- UnpinBuffer(buf_hdr);
- goto again;
+ break;
}
/* a final set of sanity checks */
diff --git a/src/backend/storage/buffer/freelist.c b/src/backend/storage/buffer/freelist.c
index 7d59a92bd1a..12bb7e2312e 100644
--- a/src/backend/storage/buffer/freelist.c
+++ b/src/backend/storage/buffer/freelist.c
@@ -15,6 +15,7 @@
*/
#include "postgres.h"
+#include "access/xlog.h"
#include "pgstat.h"
#include "port/atomics.h"
#include "storage/buf_internals.h"
@@ -716,12 +717,21 @@ IOContextForStrategy(BufferAccessStrategy strategy)
* be written out and doing so would require flushing WAL too. This gives us
* a chance to choose a different victim.
*
+ * The buffer must be pinned and content locked and the buffer header spinlock
+ * must not be held. We must hold the content lock to examine the LSN.
+ *
* Returns true if buffer manager should ask for a new victim, and false
- * if this buffer should be written and re-used.
+ * if this buffer should be written and reused.
*/
bool
StrategyRejectBuffer(BufferAccessStrategy strategy, BufferDesc *buf, bool from_ring)
{
+ uint32 buf_state;
+ XLogRecPtr lsn;
+
+ if (!strategy)
+ return false;
+
/* We only do this in bulkread mode */
if (strategy->btype != BAS_BULKREAD)
return false;
@@ -731,11 +741,19 @@ StrategyRejectBuffer(BufferAccessStrategy strategy, BufferDesc *buf, bool from_r
strategy->buffers[strategy->current] != BufferDescriptorGetBuffer(buf))
return false;
- /*
- * Remove the dirty buffer from the ring; necessary to prevent infinite
- * loop if all ring members are dirty.
- */
- strategy->buffers[strategy->current] = InvalidBuffer;
+ buf_state = LockBufHdr(buf);
+ lsn = BufferGetLSN(buf);
+ UnlockBufHdr(buf, buf_state);
+
+ if (XLogNeedsFlush(lsn))
+ {
+ /*
+ * Remove the dirty buffer from the ring; necessary to prevent an
+ * infinite loop if all ring members are dirty.
+ */
+ strategy->buffers[strategy->current] = InvalidBuffer;
+ return true;
+ }
- return true;
+ return false;
}
diff --git a/src/include/storage/buf_internals.h b/src/include/storage/buf_internals.h
index dfd614f7ca4..b1b81f31419 100644
--- a/src/include/storage/buf_internals.h
+++ b/src/include/storage/buf_internals.h
@@ -419,6 +419,11 @@ ResourceOwnerForgetBufferIO(ResourceOwner owner, Buffer buffer)
/*
* Internal buffer management routines
*/
+
+/* Note: these two macros only work on shared buffers, not local ones! */
+#define BufHdrGetBlock(bufHdr) ((Block) (BufferBlocks + ((Size) (bufHdr)->buf_id) * BLCKSZ))
+#define BufferGetLSN(bufHdr) (PageGetLSN(BufHdrGetBlock(bufHdr)))
+
/* bufmgr.c */
extern void WritebackContextInit(WritebackContext *context, int *max_pending);
extern void IssuePendingWritebacks(WritebackContext *wb_context, IOContext io_context);
--
2.43.0
From 78767ad0f17d2b74ad2aafa10ecd828d7e53bf0e Mon Sep 17 00:00:00 2001
From: Melanie Plageman <melanieplage...@gmail.com>
Date: Tue, 2 Sep 2025 12:56:38 -0400
Subject: [PATCH v6 4/9] Write combining for BAS_BULKWRITE
Implement write combining for users of the bulkwrite buffer access
strategy (e.g. COPY FROM). When the buffer access strategy needs to
clean a buffer for reuse, it already opportunistically flushes some
other buffers. Now, combine any contiguous blocks from the same relation
into larger writes and issue them with smgrwritev().
The performance benefit for COPY FROM is mostly noticeable for multiple
concurrent COPY FROMs because a single COPY FROM is either CPU bound or
bound by WAL writes.
The infrastructure for flushing larger batches of IOs will be reused by
checkpointer and other processes doing writes of dirty data.
XXX: Because this sets in-place checksums for batches, it is not
committable until additional infrastructure goes in place.
Author: Melanie Plageman <melanieplage...@gmail.com>
Reviewed-by: Chao Li <li.evan.c...@gmail.com>
Discussion: https://postgr.es/m/flat/CAAKRu_bcWRvRwZUop_d9vzF9nHAiT%2B-uPzkJ%3DS3ShZ1GqeAYOw%40mail.gmail.com
---
src/backend/storage/buffer/bufmgr.c | 218 ++++++++++++++++++++++++--
src/backend/storage/buffer/freelist.c | 26 +++
src/backend/storage/page/bufpage.c | 20 +++
src/backend/utils/probes.d | 2 +
src/include/storage/buf_internals.h | 32 ++++
src/include/storage/bufpage.h | 2 +
src/tools/pgindent/typedefs.list | 1 +
7 files changed, 290 insertions(+), 11 deletions(-)
diff --git a/src/backend/storage/buffer/bufmgr.c b/src/backend/storage/buffer/bufmgr.c
index c64268f31ae..4cc73bf4363 100644
--- a/src/backend/storage/buffer/bufmgr.c
+++ b/src/backend/storage/buffer/bufmgr.c
@@ -542,6 +542,10 @@ static BufferDesc *PrepareOrRejectEagerFlushBuffer(Buffer bufnum, BlockNumber re
RelFileLocator *rlocator,
bool skip_pinned,
XLogRecPtr *max_lsn);
+static void FindFlushAdjacents(BufferAccessStrategy strategy, Buffer sweep_end,
+ BufferDesc *batch_start,
+ uint32 max_batch_size, BufWriteBatch *batch,
+ int *sweep_cursor);
static void CleanVictimBuffer(BufferAccessStrategy strategy,
BufferDesc *bufdesc, uint32 *buf_state,
bool from_ring, IOContext io_context);
@@ -4263,10 +4267,91 @@ FlushBuffer(BufferDesc *buf, SMgrRelation reln, IOObject io_object,
DoFlushBuffer(buf, reln, io_object, io_context, lsn);
}
+
+/*
+ * Given a buffer descriptor, start, from a strategy ring, strategy, that
+ * supports eager flushing, find additional buffers from the ring that can be
+ * combined into a single write batch with this buffer.
+ *
+ * max_batch_size is the maximum number of blocks that can be combined into a
+ * single write in general. This function, based on the block number of start,
+ * will determine the maximum IO size for this particular write given how much
+ * of the file remains. max_batch_size is provided by the caller so it doesn't
+ * have to be recalculated for each write.
+ *
+ * batch is an output parameter that this function will fill with the needed
+ * information to write this IO.
+ *
+ * This function will pin and content lock all of the buffers that it
+ * assembles for the IO batch. The caller is responsible for issuing the IO.
+ */
+static void
+FindFlushAdjacents(BufferAccessStrategy strategy, Buffer sweep_end,
+ BufferDesc *batch_start,
+ uint32 max_batch_size,
+ BufWriteBatch *batch,
+ int *sweep_cursor)
+{
+ BlockNumber limit;
+ uint32 buf_state;
+
+ Assert(batch_start);
+ batch->bufdescs[0] = batch_start;
+
+ buf_state = LockBufHdr(batch_start);
+ batch->max_lsn = BufferGetLSN(batch_start);
+ UnlockBufHdr(batch_start, buf_state);
+
+ batch->start = batch->bufdescs[0]->tag.blockNum;
+ Assert(BlockNumberIsValid(batch->start));
+ batch->n = 1;
+ batch->forkno = BufTagGetForkNum(&batch->bufdescs[0]->tag);
+ batch->rlocator = BufTagGetRelFileLocator(&batch->bufdescs[0]->tag);
+ batch->reln = smgropen(batch->rlocator, INVALID_PROC_NUMBER);
+
+ limit = smgrmaxcombine(batch->reln, batch->forkno, batch->start);
+ limit = Min(max_batch_size, limit);
+ limit = Min(GetAdditionalPinLimit(), limit);
+
+ /*
+ * It's possible we're not allowed any more pins or there aren't more
+ * blocks in the target relation. In this case, just return. Our batch
+ * will have only one buffer.
+ */
+ if (limit <= 0)
+ return;
+
+ /* Now assemble a run of blocks to write out. */
+ for (; batch->n < limit; batch->n++)
+ {
+ Buffer bufnum;
+
+ if ((bufnum =
+ StrategySweepNextBuffer(strategy, sweep_cursor)) == sweep_end)
+ break;
+
+ /*
+ * For BAS_BULKWRITE, once you hit an InvalidBuffer, the remaining
+ * buffers in the ring will be invalid.
+ */
+ if (!BufferIsValid(bufnum))
+ break;
+
+ /* Stop when we encounter a buffer that will break the run */
+ if ((batch->bufdescs[batch->n] =
+ PrepareOrRejectEagerFlushBuffer(bufnum,
+ batch->start + batch->n,
+ &batch->rlocator,
+ true,
+ &batch->max_lsn)) == NULL)
+ break;
+ }
+}
+
/*
* Returns the buffer descriptor of the buffer containing the next block we
* should eagerly flush or NULL when there are no further buffers to consider
- * writing out.
+ * writing out. This will be the start of a new batch of buffers to write out.
*/
static BufferDesc *
NextStratBufToFlush(BufferAccessStrategy strategy,
@@ -4314,7 +4399,6 @@ CleanVictimBuffer(BufferAccessStrategy strategy,
XLogRecPtr max_lsn = InvalidXLogRecPtr;
LWLock *content_lock;
- bool first_buffer = true;
Assert(*buf_state & BM_DIRTY);
@@ -4326,19 +4410,22 @@ CleanVictimBuffer(BufferAccessStrategy strategy,
{
Buffer sweep_end = BufferDescriptorGetBuffer(bufdesc);
int cursor = StrategySweepStart(strategy);
+ uint32 max_batch_size = StrategyMaxWriteBatchSize(strategy);
+
+ /* Pin our victim again so it stays ours even after batch released */
+ ReservePrivateRefCountEntry();
+ ResourceOwnerEnlarge(CurrentResourceOwner);
+ IncrBufferRefCount(BufferDescriptorGetBuffer(bufdesc));
/* Clean victim buffer and find more to flush opportunistically */
do
{
- DoFlushBuffer(bufdesc, NULL, IOOBJECT_RELATION, io_context, max_lsn);
- content_lock = BufferDescriptorGetContentLock(bufdesc);
- LWLockRelease(content_lock);
- ScheduleBufferTagForWriteback(&BackendWritebackContext, io_context,
- &bufdesc->tag);
- /* We leave the first buffer pinned for the caller */
- if (!first_buffer)
- UnpinBuffer(bufdesc);
- first_buffer = false;
+ BufWriteBatch batch;
+
+ FindFlushAdjacents(strategy, sweep_end, bufdesc, max_batch_size,
+ &batch, &cursor);
+ FlushBufferBatch(&batch, io_context);
+ CompleteWriteBatchIO(&batch, io_context, &BackendWritebackContext);
} while ((bufdesc = NextStratBufToFlush(strategy, sweep_end,
&max_lsn, &cursor)) != NULL);
}
@@ -4461,6 +4548,73 @@ except_unlock_header:
return NULL;
}
+/*
+ * Given a prepared batch of buffers write them out as a vector.
+ */
+void
+FlushBufferBatch(BufWriteBatch *batch,
+ IOContext io_context)
+{
+ BlockNumber blknums[MAX_IO_COMBINE_LIMIT];
+ Block blocks[MAX_IO_COMBINE_LIMIT];
+ instr_time io_start;
+ ErrorContextCallback errcallback =
+ {
+ .callback = shared_buffer_write_error_callback,
+ .previous = error_context_stack,
+ };
+
+ error_context_stack = &errcallback;
+
+ if (!XLogRecPtrIsInvalid(batch->max_lsn))
+ XLogFlush(batch->max_lsn);
+
+ if (batch->reln == NULL)
+ batch->reln = smgropen(batch->rlocator, INVALID_PROC_NUMBER);
+
+#ifdef USE_ASSERT_CHECKING
+ for (uint32 i = 0; i < batch->n; i++)
+ {
+ BufferDesc *bufdesc = batch->bufdescs[i];
+ uint32 buf_state = LockBufHdr(bufdesc);
+ XLogRecPtr lsn = BufferGetLSN(bufdesc);
+
+ UnlockBufHdr(bufdesc, buf_state);
+ Assert(!(buf_state & BM_PERMANENT) || !XLogNeedsFlush(lsn));
+ }
+#endif
+
+ TRACE_POSTGRESQL_BUFFER_BATCH_FLUSH_START(batch->forkno,
+ batch->reln->smgr_rlocator.locator.spcOid,
+ batch->reln->smgr_rlocator.locator.dbOid,
+ batch->reln->smgr_rlocator.locator.relNumber,
+ batch->reln->smgr_rlocator.backend,
+ batch->n);
+
+ /*
+ * XXX: All blocks should be copied and then checksummed but doing so
+ * takes a lot of extra memory and a future patch will eliminate this
+ * requirement.
+ */
+ for (BlockNumber i = 0; i < batch->n; i++)
+ {
+ blknums[i] = batch->start + i;
+ blocks[i] = BufHdrGetBlock(batch->bufdescs[i]);
+ }
+
+ PageSetBatchChecksumInplace((Page *) blocks, blknums, batch->n);
+
+ io_start = pgstat_prepare_io_time(track_io_timing);
+
+ smgrwritev(batch->reln, batch->forkno,
+ batch->start, (const void **) blocks, batch->n, false);
+
+ pgstat_count_io_op_time(IOOBJECT_RELATION, io_context, IOOP_WRITE,
+ io_start, batch->n, BLCKSZ);
+
+ error_context_stack = errcallback.previous;
+}
+
/*
* Prepare the buffer with budesc for writing. buf_state and lsn are output
* parameters. Returns true if the buffer acutally needs writing and false
@@ -4606,6 +4760,48 @@ DoFlushBuffer(BufferDesc *buf, SMgrRelation reln, IOObject io_object,
error_context_stack = errcallback.previous;
}
+/*
+ * Given a previously initialized batch with buffers that have already been
+ * flushed, terminate the IO on each buffer and then unlock and unpin them.
+ * This assumes all the buffers were locked and pinned. wb_context will be
+ * modified.
+ */
+void
+CompleteWriteBatchIO(BufWriteBatch *batch, IOContext io_context,
+ WritebackContext *wb_context)
+{
+ ErrorContextCallback errcallback =
+ {
+ .callback = shared_buffer_write_error_callback,
+ .previous = error_context_stack,
+ };
+
+ error_context_stack = &errcallback;
+ pgBufferUsage.shared_blks_written += batch->n;
+
+ for (uint32 i = 0; i < batch->n; i++)
+ {
+ Buffer buffer = BufferDescriptorGetBuffer(batch->bufdescs[i]);
+
+ errcallback.arg = batch->bufdescs[i];
+
+ /* Mark the buffer as clean and end the BM_IO_IN_PROGRESS state. */
+ TerminateBufferIO(batch->bufdescs[i], true, 0, true, false);
+ LWLockRelease(BufferDescriptorGetContentLock(batch->bufdescs[i]));
+ ReleaseBuffer(buffer);
+ ScheduleBufferTagForWriteback(wb_context, io_context,
+ &batch->bufdescs[i]->tag);
+ }
+
+ TRACE_POSTGRESQL_BUFFER_BATCH_FLUSH_DONE(batch->forkno,
+ batch->reln->smgr_rlocator.locator.spcOid,
+ batch->reln->smgr_rlocator.locator.dbOid,
+ batch->reln->smgr_rlocator.locator.relNumber,
+ batch->reln->smgr_rlocator.backend,
+ batch->n, batch->start);
+ error_context_stack = errcallback.previous;
+}
+
/*
* RelationGetNumberOfBlocksInFork
* Determines the current number of pages in the specified relation fork.
diff --git a/src/backend/storage/buffer/freelist.c b/src/backend/storage/buffer/freelist.c
index 8716109221b..317f41cdfa2 100644
--- a/src/backend/storage/buffer/freelist.c
+++ b/src/backend/storage/buffer/freelist.c
@@ -712,6 +712,32 @@ GetBufferFromRing(BufferAccessStrategy strategy, uint32 *buf_state)
return NULL;
}
+
+/*
+ * Determine the largest IO we can assemble from the given strategy ring given
+ * strategy-specific as well as global constraints on the number of pinned
+ * buffers and max IO size.
+ */
+uint32
+StrategyMaxWriteBatchSize(BufferAccessStrategy strategy)
+{
+ uint32 max_possible_buffer_limit;
+ uint32 max_write_batch_size;
+ int strategy_pin_limit;
+
+ max_write_batch_size = io_combine_limit;
+
+ strategy_pin_limit = GetAccessStrategyPinLimit(strategy);
+ max_possible_buffer_limit = GetPinLimit();
+
+ max_write_batch_size = Min(strategy_pin_limit, max_write_batch_size);
+ max_write_batch_size = Min(max_possible_buffer_limit, max_write_batch_size);
+ max_write_batch_size = Max(1, max_write_batch_size);
+ max_write_batch_size = Min(max_write_batch_size, io_combine_limit);
+ Assert(max_write_batch_size < MAX_IO_COMBINE_LIMIT);
+ return max_write_batch_size;
+}
+
/*
* AddBufferToRing -- add a buffer to the buffer ring
*
diff --git a/src/backend/storage/page/bufpage.c b/src/backend/storage/page/bufpage.c
index dbb49ed9197..12503934502 100644
--- a/src/backend/storage/page/bufpage.c
+++ b/src/backend/storage/page/bufpage.c
@@ -1546,3 +1546,23 @@ PageSetChecksumInplace(Page page, BlockNumber blkno)
((PageHeader) page)->pd_checksum = pg_checksum_page(page, blkno);
}
+
+/*
+ * A helper to set multiple block's checksums
+ */
+void
+PageSetBatchChecksumInplace(Page *pages, const BlockNumber *blknos, uint32 length)
+{
+ /* If we don't need a checksum, just return */
+ if (!DataChecksumsEnabled())
+ return;
+
+ for (uint32 i = 0; i < length; i++)
+ {
+ Page page = pages[i];
+
+ if (PageIsNew(page))
+ continue;
+ ((PageHeader) page)->pd_checksum = pg_checksum_page(page, blknos[i]);
+ }
+}
diff --git a/src/backend/utils/probes.d b/src/backend/utils/probes.d
index e9e413477ba..36dd4f8375b 100644
--- a/src/backend/utils/probes.d
+++ b/src/backend/utils/probes.d
@@ -61,6 +61,8 @@ provider postgresql {
probe buffer__flush__done(ForkNumber, BlockNumber, Oid, Oid, Oid);
probe buffer__extend__start(ForkNumber, Oid, Oid, Oid, int, unsigned int);
probe buffer__extend__done(ForkNumber, Oid, Oid, Oid, int, unsigned int, BlockNumber);
+ probe buffer__batch__flush__start(ForkNumber, Oid, Oid, Oid, int, unsigned int);
+ probe buffer__batch__flush__done(ForkNumber, Oid, Oid, Oid, int, unsigned int, BlockNumber);
probe buffer__checkpoint__start(int);
probe buffer__checkpoint__sync__start();
diff --git a/src/include/storage/buf_internals.h b/src/include/storage/buf_internals.h
index 03faf80e441..7c60e5e6f54 100644
--- a/src/include/storage/buf_internals.h
+++ b/src/include/storage/buf_internals.h
@@ -416,6 +416,34 @@ ResourceOwnerForgetBufferIO(ResourceOwner owner, Buffer buffer)
ResourceOwnerForget(owner, Int32GetDatum(buffer), &buffer_io_resowner_desc);
}
+/*
+ * Used to write out multiple blocks at a time in a combined IO. bufdescs
+ * contains buffer descriptors for buffers containing adjacent blocks of the
+ * same fork of the same relation.
+ */
+typedef struct BufWriteBatch
+{
+ RelFileLocator rlocator;
+ ForkNumber forkno;
+ SMgrRelation reln;
+
+ /*
+ * The BlockNumber of the first block in the run of contiguous blocks to
+ * be written out as a single IO.
+ */
+ BlockNumber start;
+
+ /*
+ * While assembling the buffers, we keep track of the maximum LSN so that
+ * we can flush WAL through this LSN before flushing the buffers.
+ */
+ XLogRecPtr max_lsn;
+
+ /* The number of valid buffers in bufdescs */
+ uint32 n;
+ BufferDesc *bufdescs[MAX_IO_COMBINE_LIMIT];
+} BufWriteBatch;
+
/*
* Internal buffer management routines
*/
@@ -429,6 +457,7 @@ extern void WritebackContextInit(WritebackContext *context, int *max_pending);
extern void IssuePendingWritebacks(WritebackContext *wb_context, IOContext io_context);
extern void ScheduleBufferTagForWriteback(WritebackContext *wb_context,
IOContext io_context, BufferTag *tag);
+extern void FlushBufferBatch(BufWriteBatch *batch, IOContext io_context);
/* solely to make it easier to write tests */
extern bool StartBufferIO(BufferDesc *buf, bool forInput, bool nowait);
@@ -438,9 +467,12 @@ extern void TerminateBufferIO(BufferDesc *buf, bool clear_dirty, uint32 set_flag
/* freelist.c */
extern bool StrategySupportsEagerFlush(BufferAccessStrategy strategy);
+extern uint32 StrategyMaxWriteBatchSize(BufferAccessStrategy strategy);
extern Buffer StrategySweepNextBuffer(BufferAccessStrategy strategy,
int *sweep_cursor);
extern int StrategySweepStart(BufferAccessStrategy strategy);
+extern void CompleteWriteBatchIO(BufWriteBatch *batch, IOContext io_context,
+ WritebackContext *wb_context);
extern IOContext IOContextForStrategy(BufferAccessStrategy strategy);
extern BufferDesc *StrategyGetBuffer(BufferAccessStrategy strategy,
uint32 *buf_state, bool *from_ring);
diff --git a/src/include/storage/bufpage.h b/src/include/storage/bufpage.h
index aeb67c498c5..bb4e6af461a 100644
--- a/src/include/storage/bufpage.h
+++ b/src/include/storage/bufpage.h
@@ -507,5 +507,7 @@ extern bool PageIndexTupleOverwrite(Page page, OffsetNumber offnum,
Item newtup, Size newsize);
extern char *PageSetChecksumCopy(Page page, BlockNumber blkno);
extern void PageSetChecksumInplace(Page page, BlockNumber blkno);
+extern void PageSetBatchChecksumInplace(Page *pages, const BlockNumber *blknos,
+ uint32 length);
#endif /* BUFPAGE_H */
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index a13e8162890..9492adeee58 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -349,6 +349,7 @@ BufferManagerRelation
BufferStrategyControl
BufferTag
BufferUsage
+BufWriteBatch
BuildAccumulator
BuiltinScript
BulkInsertState
--
2.43.0
From 9fa36f0a1f489809d7d50798e07d491f4d806ecc Mon Sep 17 00:00:00 2001
From: Melanie Plageman <melanieplage...@gmail.com>
Date: Tue, 2 Sep 2025 15:22:11 -0400
Subject: [PATCH v6 6/9] Add database Oid to CkptSortItem
This is useful for checkpointer write combining -- which will be added
in a future commit.
Author: Melanie Plageman <melanieplage...@gmail.com>
Reviewed-by: Chao Li <li.evan.c...@gmail.com>
Discussion: https://postgr.es/m/2FA0BAC7-5413-4ABD-94CA-4398FE77750D%40gmail.com
---
src/backend/storage/buffer/bufmgr.c | 8 ++++++++
src/include/storage/buf_internals.h | 1 +
2 files changed, 9 insertions(+)
diff --git a/src/backend/storage/buffer/bufmgr.c b/src/backend/storage/buffer/bufmgr.c
index 4cc73bf4363..2ea91e777e2 100644
--- a/src/backend/storage/buffer/bufmgr.c
+++ b/src/backend/storage/buffer/bufmgr.c
@@ -3377,6 +3377,7 @@ BufferSync(int flags)
item = &CkptBufferIds[num_to_scan++];
item->buf_id = buf_id;
item->tsId = bufHdr->tag.spcOid;
+ item->dbId = bufHdr->tag.dbOid;
item->relNumber = BufTagGetRelNumber(&bufHdr->tag);
item->forkNum = BufTagGetForkNum(&bufHdr->tag);
item->blockNum = bufHdr->tag.blockNum;
@@ -6732,6 +6733,13 @@ ckpt_buforder_comparator(const CkptSortItem *a, const CkptSortItem *b)
return -1;
else if (a->tsId > b->tsId)
return 1;
+
+ /* compare database */
+ if (a->dbId < b->dbId)
+ return -1;
+ else if (a->dbId > b->dbId)
+ return 1;
+
/* compare relation */
if (a->relNumber < b->relNumber)
return -1;
diff --git a/src/include/storage/buf_internals.h b/src/include/storage/buf_internals.h
index 7c60e5e6f54..03c395903e5 100644
--- a/src/include/storage/buf_internals.h
+++ b/src/include/storage/buf_internals.h
@@ -382,6 +382,7 @@ UnlockBufHdr(BufferDesc *desc, uint32 buf_state)
typedef struct CkptSortItem
{
Oid tsId;
+ Oid dbId;
RelFileNumber relNumber;
ForkNumber forkNum;
BlockNumber blockNum;
--
2.43.0
From 898db59ca9a02f8eb5481caa66dca6cd02f30082 Mon Sep 17 00:00:00 2001
From: Melanie Plageman <melanieplage...@gmail.com>
Date: Tue, 2 Sep 2025 15:42:29 -0400
Subject: [PATCH v6 7/9] Implement checkpointer data write combining
When the checkpointer writes out dirty buffers, writing multiple
contiguous blocks as a single IO is a substantial performance
improvement. The checkpointer is usually bottlenecked on IO, so issuing
larger IOs leads to increased write throughput and faster checkpoints.
Author: Melanie Plageman <melanieplage...@gmail.com>
Reviewed-by: Chao Li <li.evan.c...@gmail.com>
Discussion: https://postgr.es/m/2FA0BAC7-5413-4ABD-94CA-4398FE77750D%40gmail.com
---
src/backend/storage/buffer/bufmgr.c | 236 +++++++++++++++++++++++++---
src/backend/utils/probes.d | 2 +-
2 files changed, 211 insertions(+), 27 deletions(-)
diff --git a/src/backend/storage/buffer/bufmgr.c b/src/backend/storage/buffer/bufmgr.c
index 2ea91e777e2..2475b1c85be 100644
--- a/src/backend/storage/buffer/bufmgr.c
+++ b/src/backend/storage/buffer/bufmgr.c
@@ -512,6 +512,7 @@ static bool PinBuffer(BufferDesc *buf, BufferAccessStrategy strategy);
static void PinBuffer_Locked(BufferDesc *buf);
static void UnpinBuffer(BufferDesc *buf);
static void UnpinBufferNoOwner(BufferDesc *buf);
+static uint32 CheckpointerMaxBatchSize(void);
static void BufferSync(int flags);
static uint32 WaitBufHdrUnlocked(BufferDesc *buf);
static int SyncOneBuffer(int buf_id, bool skip_recently_used,
@@ -3319,7 +3320,6 @@ UnpinBufferNoOwner(BufferDesc *buf)
static void
BufferSync(int flags)
{
- uint32 buf_state;
int buf_id;
int num_to_scan;
int num_spaces;
@@ -3331,6 +3331,8 @@ BufferSync(int flags)
int i;
int mask = BM_DIRTY;
WritebackContext wb_context;
+ uint32 max_batch_size;
+ BufWriteBatch batch;
/*
* Unless this is a shutdown checkpoint or we have been explicitly told,
@@ -3361,6 +3363,7 @@ BufferSync(int flags)
for (buf_id = 0; buf_id < NBuffers; buf_id++)
{
BufferDesc *bufHdr = GetBufferDescriptor(buf_id);
+ uint32 buf_state;
/*
* Header spinlock is enough to examine BM_DIRTY, see comment in
@@ -3501,48 +3504,212 @@ BufferSync(int flags)
*/
num_processed = 0;
num_written = 0;
+ max_batch_size = CheckpointerMaxBatchSize();
while (!binaryheap_empty(ts_heap))
{
+ BlockNumber limit = max_batch_size;
BufferDesc *bufHdr = NULL;
CkptTsStatus *ts_stat = (CkptTsStatus *)
DatumGetPointer(binaryheap_first(ts_heap));
+ int ts_end = ts_stat->index - ts_stat->num_scanned + ts_stat->num_to_scan;
+ int processed = 0;
- buf_id = CkptBufferIds[ts_stat->index].buf_id;
- Assert(buf_id != -1);
+ batch.start = InvalidBlockNumber;
+ batch.max_lsn = InvalidXLogRecPtr;
+ batch.n = 0;
- bufHdr = GetBufferDescriptor(buf_id);
+ while (batch.n < limit)
+ {
+ uint32 buf_state;
+ XLogRecPtr lsn = InvalidXLogRecPtr;
+ LWLock *content_lock;
+ CkptSortItem item;
- num_processed++;
+ if (ProcSignalBarrierPending)
+ ProcessProcSignalBarrier();
- /*
- * We don't need to acquire the lock here, because we're only looking
- * at a single bit. It's possible that someone else writes the buffer
- * and clears the flag right after we check, but that doesn't matter
- * since SyncOneBuffer will then do nothing. However, there is a
- * further race condition: it's conceivable that between the time we
- * examine the bit here and the time SyncOneBuffer acquires the lock,
- * someone else not only wrote the buffer but replaced it with another
- * page and dirtied it. In that improbable case, SyncOneBuffer will
- * write the buffer though we didn't need to. It doesn't seem worth
- * guarding against this, though.
- */
- if (pg_atomic_read_u32(&bufHdr->state) & BM_CHECKPOINT_NEEDED)
- {
- if (SyncOneBuffer(buf_id, false, &wb_context) & BUF_WRITTEN)
+ /* Check if we are done with this tablespace */
+ if (ts_stat->index + processed >= ts_end)
+ break;
+
+ item = CkptBufferIds[ts_stat->index + processed];
+
+ buf_id = item.buf_id;
+ Assert(buf_id != -1);
+
+ bufHdr = GetBufferDescriptor(buf_id);
+
+ /*
+ * If this is the first block of the batch, then check if we need
+ * to open a new relation. Open the relation now because we have
+ * to determine the maximum IO size based on how many blocks
+ * remain in the file.
+ */
+ if (!BlockNumberIsValid(batch.start))
+ {
+ Assert(batch.max_lsn == InvalidXLogRecPtr && batch.n == 0);
+ batch.rlocator.spcOid = item.tsId;
+ batch.rlocator.dbOid = item.dbId;
+ batch.rlocator.relNumber = item.relNumber;
+ batch.forkno = item.forkNum;
+ batch.start = item.blockNum;
+ batch.reln = smgropen(batch.rlocator, INVALID_PROC_NUMBER);
+ limit = smgrmaxcombine(batch.reln, batch.forkno, batch.start);
+ limit = Max(max_batch_size, limit);
+ limit = Min(GetAdditionalPinLimit(), limit);
+
+ /*
+ * If we aren't allowed any more pins or there are no more
+ * blocks in the relation, break out of the loop and issue the
+ * IO.
+ */
+ if (limit <= 0)
+ break;
+ }
+
+ /*
+ * Once we hit blocks from the next relation or fork of the
+ * relation, break out of the loop and issue the IO we've built up
+ * so far. It is important that we don't increment processed
+ * becasue we want to start the next IO with this item.
+ */
+ if (item.dbId != batch.rlocator.dbOid)
+ break;
+
+ if (item.relNumber != batch.rlocator.relNumber)
+ break;
+
+ if (item.forkNum != batch.forkno)
+ break;
+
+ /*
+ * It the next block is not contiguous, we can't include it in the
+ * IO we will issue. Break out of the loop and issue what we have
+ * so far. Do not count this item as processed -- otherwise we
+ * will end up skipping it.
+ */
+ if (item.blockNum != batch.start + batch.n)
+ break;
+
+ /*
+ * We don't need to acquire the lock here, because we're only
+ * looking at a single bit. It's possible that someone else writes
+ * the buffer and clears the flag right after we check, but that
+ * doesn't matter since StartBufferIO will then return false. If
+ * the buffer doesn't need checkpointing, don't include it in the
+ * batch we are building. We're done with the item, so count it as
+ * processed and break out of the loop to issue the IO we have
+ * built so far.
+ */
+ if (!(pg_atomic_read_u32(&bufHdr->state) & BM_CHECKPOINT_NEEDED))
+ {
+ processed++;
+ break;
+ }
+
+ ReservePrivateRefCountEntry();
+ ResourceOwnerEnlarge(CurrentResourceOwner);
+
+ buf_state = LockBufHdr(bufHdr);
+
+ /*
+ * If the buffer doesn't need eviction, we're done with the item,
+ * so count it as processed and break out of the loop to issue the
+ * IO so far.
+ */
+ if (!(buf_state & BM_VALID) || !(buf_state & BM_DIRTY))
+ {
+ processed++;
+ UnlockBufHdr(bufHdr, buf_state);
+ break;
+ }
+
+ PinBuffer_Locked(bufHdr);
+
+ /*
+ * There is a race condition here: it's conceivable that between
+ * the time we examine the buffer header for BM_CHECKPOINT_NEEDED
+ * above and when we are now acquiring the lock that, someone else
+ * not only wrote the buffer but replaced it with another page and
+ * dirtied it. In that improbable case, we will write the buffer
+ * though we didn't need to. It doesn't seem worth guarding
+ * against this, though.
+ */
+ content_lock = BufferDescriptorGetContentLock(bufHdr);
+
+ /*
+ * We are willing to wait for the content lock on the first IO in
+ * the batch. However, for subsequent IOs, waiting could lead to
+ * deadlock. We have to eventually flush all eligible buffers,
+ * though. So, if we fail to acquire the lock on a subsequent
+ * buffer, we break out and issue the IO we've built up so far.
+ * Then we come back and start a new IO with that buffer as the
+ * starting buffer. As such, we must not count the item as
+ * processed if we end up failing to acquire the content lock.
+ */
+ if (batch.n == 0)
+ LWLockAcquire(content_lock, LW_SHARED);
+ else if (!LWLockConditionalAcquire(content_lock, LW_SHARED))
{
- TRACE_POSTGRESQL_BUFFER_SYNC_WRITTEN(buf_id);
- PendingCheckpointerStats.buffers_written++;
- num_written++;
+ UnpinBuffer(bufHdr);
+ break;
+ }
+
+ /*
+ * If the buffer doesn't need IO, count the item as processed,
+ * release the buffer, and break out of the loop to issue the IO
+ * we have built up so far.
+ */
+ if (!StartBufferIO(bufHdr, false, true))
+ {
+ processed++;
+ LWLockRelease(content_lock);
+ UnpinBuffer(bufHdr);
+ break;
}
+
+ buf_state = LockBufHdr(bufHdr);
+ lsn = BufferGetLSN(bufHdr);
+ buf_state &= ~BM_JUST_DIRTIED;
+ UnlockBufHdr(bufHdr, buf_state);
+
+ /*
+ * Keep track of the max LSN so that we can be sure to flush
+ * enough WAL before flushing data from the buffers. See comment
+ * in DoFlushBuffer() for more on why we don't consider the LSNs
+ * of unlogged relations.
+ */
+ if (buf_state & BM_PERMANENT && lsn > batch.max_lsn)
+ batch.max_lsn = lsn;
+
+ batch.bufdescs[batch.n++] = bufHdr;
+ processed++;
}
/*
* Measure progress independent of actually having to flush the buffer
* - otherwise writing become unbalanced.
*/
- ts_stat->progress += ts_stat->progress_slice;
- ts_stat->num_scanned++;
- ts_stat->index++;
+ num_processed += processed;
+ ts_stat->progress += ts_stat->progress_slice * processed;
+ ts_stat->num_scanned += processed;
+ ts_stat->index += processed;
+
+ /*
+ * If we built up an IO, issue it. There's a chance we didn't find any
+ * items referencing buffers that needed flushing this time, but we
+ * still want to check if we should update the heap if we examined and
+ * processed the items.
+ */
+ if (batch.n > 0)
+ {
+ FlushBufferBatch(&batch, IOCONTEXT_NORMAL);
+ CompleteWriteBatchIO(&batch, IOCONTEXT_NORMAL, &wb_context);
+
+ TRACE_POSTGRESQL_BUFFER_BATCH_SYNC_WRITTEN(batch.n);
+ PendingCheckpointerStats.buffers_written += batch.n;
+ num_written += batch.n;
+ }
/* Have all the buffers from the tablespace been processed? */
if (ts_stat->num_scanned == ts_stat->num_to_scan)
@@ -4268,6 +4435,23 @@ FlushBuffer(BufferDesc *buf, SMgrRelation reln, IOObject io_object,
DoFlushBuffer(buf, reln, io_object, io_context, lsn);
}
+/*
+ * The maximum number of blocks that can be written out in a single batch by
+ * the checkpointer.
+ */
+static uint32
+CheckpointerMaxBatchSize(void)
+{
+ uint32 result;
+ uint32 pin_limit = GetPinLimit();
+
+ result = Max(pin_limit, 1);
+ result = Min(pin_limit, io_combine_limit);
+ result = Max(result, 1);
+ Assert(result < MAX_IO_COMBINE_LIMIT);
+ return result;
+}
+
/*
* Given a buffer descriptor, start, from a strategy ring, strategy, that
diff --git a/src/backend/utils/probes.d b/src/backend/utils/probes.d
index 36dd4f8375b..d6970731ba9 100644
--- a/src/backend/utils/probes.d
+++ b/src/backend/utils/probes.d
@@ -68,7 +68,7 @@ provider postgresql {
probe buffer__checkpoint__sync__start();
probe buffer__checkpoint__done();
probe buffer__sync__start(int, int);
- probe buffer__sync__written(int);
+ probe buffer__batch__sync__written(BlockNumber);
probe buffer__sync__done(int, int, int);
probe deadlock__found();
--
2.43.0
