From adbb710f122c08a0594a3082880402dd16e1ec6d Mon Sep 17 00:00:00 2001 From: Melanie Plageman Date: Mon, 3 Aug 2026 17:06:12 -0400 Subject: [PATCH 1/2] Record the IOContext on a ReadBuffersOperation instead of the strategy A ReadBuffersOperation only needs the buffer access strategy while a read is being started to choose the buffer to read into. When the IO is completed, it needs to know what IOContext to count it under. Instead of saving the strategy in every operation, pass it to StartReadBuffers() and save the IOContext in the operation. We need to save the BufferAccessStrategy in the ReadStream because there are users that may abandon use of it during an ongoing stream, and we need to be able to clear it for future IOs without affecting in-progress IOs. If we have it in the ReadStream itself, it is cleaner not to have it duplicated in each operation as well. --- src/backend/storage/aio/read_stream.c | 11 ++++--- src/backend/storage/buffer/bufmgr.c | 47 ++++++++++++--------------- src/include/storage/bufmgr.h | 16 +++++++-- src/test/modules/test_aio/test_aio.c | 4 +-- 4 files changed, 42 insertions(+), 36 deletions(-) diff --git a/src/backend/storage/aio/read_stream.c b/src/backend/storage/aio/read_stream.c index a318539e56c..03dd7fee0ae 100644 --- a/src/backend/storage/aio/read_stream.c +++ b/src/backend/storage/aio/read_stream.c @@ -119,6 +119,7 @@ struct ReadStream int16 resume_readahead_distance; int16 resume_combine_distance; int read_buffers_flags; + BufferAccessStrategy strategy; bool sync_mode; /* using io_method=sync */ bool batch_mode; /* READ_STREAM_USE_BATCHING */ bool advice_enabled; @@ -445,7 +446,8 @@ read_stream_start_pending_read(ReadStream *stream) &stream->buffers[buffer_index], stream->pending_read_blocknum, &nblocks, - flags); + flags, + stream->strategy); stream->pinned_buffers += nblocks; /* Remember whether we need to wait before returning this buffer. */ @@ -932,6 +934,7 @@ read_stream_begin_impl(int flags, stream->seq_until_processed = InvalidBlockNumber; stream->temporary = SmgrIsTemp(smgr); stream->distance_decay_holdoff = 0; + stream->strategy = strategy; /* * Skip the initial ramp-up phase if the caller says we're going to be @@ -962,7 +965,6 @@ read_stream_begin_impl(int flags, stream->ios[i].op.smgr = smgr; stream->ios[i].op.persistence = persistence; stream->ios[i].op.forknum = forknum; - stream->ios[i].op.strategy = strategy; } return stream; @@ -1101,7 +1103,8 @@ read_stream_next_buffer(ReadStream *stream, void **per_buffer_data) if (likely(!StartReadBuffer(&stream->ios[0].op, &stream->buffers[oldest_buffer_index], next_blocknum, - flags))) + flags, + stream->strategy))) { /* Fast return. */ read_stream_count_prefetch(stream); @@ -1376,7 +1379,7 @@ read_stream_next_buffer(ReadStream *stream, void **per_buffer_data) BlockNumber read_stream_next_block(ReadStream *stream, BufferAccessStrategy *strategy) { - *strategy = stream->ios[0].op.strategy; + *strategy = stream->strategy; return read_stream_get_block(stream, NULL); } diff --git a/src/backend/storage/buffer/bufmgr.c b/src/backend/storage/buffer/bufmgr.c index 169829eb020..8d412ee544d 100644 --- a/src/backend/storage/buffer/bufmgr.c +++ b/src/backend/storage/buffer/bufmgr.c @@ -1357,11 +1357,11 @@ ReadBuffer_common(Relation rel, SMgrRelation smgr, char smgr_persistence, operation.rel = rel; operation.persistence = persistence; operation.forknum = forkNum; - operation.strategy = strategy; if (StartReadBuffer(&operation, &buffer, blockNum, - flags)) + flags, + strategy)) WaitReadBuffers(&operation); return buffer; @@ -1373,7 +1373,8 @@ StartReadBuffersImpl(ReadBuffersOperation *operation, BlockNumber blockNum, int *nblocks, int flags, - bool allow_forwarding) + bool allow_forwarding, + BufferAccessStrategy strategy) { int actual_nblocks = *nblocks; int maxcombine = 0; @@ -1398,10 +1399,12 @@ StartReadBuffersImpl(ReadBuffersOperation *operation, } else { - io_context = IOContextForStrategy(operation->strategy); + io_context = IOContextForStrategy(strategy); io_object = IOOBJECT_RELATION; } + operation->io_context = io_context; + for (int i = 0; i < actual_nblocks; ++i) { bool found; @@ -1449,7 +1452,7 @@ StartReadBuffersImpl(ReadBuffersOperation *operation, operation->persistence, operation->forknum, blockNum + i, - operation->strategy, + strategy, io_object, io_context, &found); } @@ -1619,10 +1622,12 @@ StartReadBuffers(ReadBuffersOperation *operation, Buffer *buffers, BlockNumber blockNum, int *nblocks, - int flags) + int flags, + BufferAccessStrategy strategy) { return StartReadBuffersImpl(operation, buffers, blockNum, nblocks, flags, - true /* expect forwarded buffers */ ); + true /* expect forwarded buffers */ , + strategy); } /* @@ -1637,13 +1642,15 @@ bool StartReadBuffer(ReadBuffersOperation *operation, Buffer *buffer, BlockNumber blocknum, - int flags) + int flags, + BufferAccessStrategy strategy) { int nblocks = 1; bool result; result = StartReadBuffersImpl(operation, buffer, blocknum, &nblocks, flags, - false /* single block, no forwarding */ ); + false /* single block, no forwarding */ , + strategy); Assert(nblocks == 1); /* single block can't be short */ return result; @@ -1759,20 +1766,13 @@ bool WaitReadBuffers(ReadBuffersOperation *operation) { PgAioReturn *aio_ret = &operation->io_return; - IOContext io_context; IOObject io_object; bool needed_wait = false; if (operation->persistence == RELPERSISTENCE_TEMP) - { - io_context = IOCONTEXT_NORMAL; io_object = IOOBJECT_TEMP_RELATION; - } else - { - io_context = IOContextForStrategy(operation->strategy); io_object = IOOBJECT_RELATION; - } /* * If we get here without an IO operation having been issued, the @@ -1833,7 +1833,7 @@ WaitReadBuffers(ReadBuffersOperation *operation) * itself was already counted earlier in AsyncReadBuffers() -- * either by us or by another backend if this is a foreign IO. */ - pgstat_count_io_op_time(io_object, io_context, IOOP_READ, + pgstat_count_io_op_time(io_object, operation->io_context, IOOP_READ, io_start, 0, 0); } else @@ -1860,7 +1860,7 @@ WaitReadBuffers(ReadBuffersOperation *operation) * Track this as a 'hit' for this backend. The backend * performing the IO will track it as a 'read'. */ - TrackBufferHit(io_object, io_context, + TrackBufferHit(io_object, operation->io_context, operation->rel, operation->persistence, operation->smgr, operation->forknum, blocknum); @@ -1949,21 +1949,14 @@ AsyncReadBuffers(ReadBuffersOperation *operation, int *nblocks_progress) PgAioHandle *ioh; uint32 ioh_flags = 0; void *io_pages[MAX_IO_COMBINE_LIMIT]; - IOContext io_context; IOObject io_object; instr_time io_start; StartBufferIOResult status; if (persistence == RELPERSISTENCE_TEMP) - { - io_context = IOCONTEXT_NORMAL; io_object = IOOBJECT_TEMP_RELATION; - } else - { - io_context = IOContextForStrategy(operation->strategy); io_object = IOOBJECT_RELATION; - } /* * When this IO is executed synchronously, either because the caller will @@ -2080,7 +2073,7 @@ AsyncReadBuffers(ReadBuffersOperation *operation, int *nblocks_progress) * it must have started out as a miss in PinBufferForBlock(). The * other backend will track this as a 'read'. */ - TrackBufferHit(io_object, io_context, + TrackBufferHit(io_object, operation->io_context, operation->rel, operation->persistence, operation->smgr, operation->forknum, blocknum); @@ -2153,7 +2146,7 @@ AsyncReadBuffers(ReadBuffersOperation *operation, int *nblocks_progress) smgrstartreadv(ioh, operation->smgr, forknum, blocknum, io_pages, io_buffers_len); - pgstat_count_io_op_time(io_object, io_context, IOOP_READ, + pgstat_count_io_op_time(io_object, operation->io_context, IOOP_READ, io_start, 1, io_buffers_len * BLCKSZ); if (persistence == RELPERSISTENCE_TEMP) diff --git a/src/include/storage/bufmgr.h b/src/include/storage/bufmgr.h index 6837b35fc6d..d5a31ad428c 100644 --- a/src/include/storage/bufmgr.h +++ b/src/include/storage/bufmgr.h @@ -20,6 +20,7 @@ #include "storage/buf.h" #include "storage/bufpage.h" #include "storage/relfilelocator.h" +#include "pgstat.h" #include "utils/relcache.h" #include "utils/snapmgr.h" @@ -135,7 +136,14 @@ struct ReadBuffersOperation SMgrRelation smgr; char persistence; ForkNumber forknum; - BufferAccessStrategy strategy; + + /* + * The IO context this read is counted under. This is saved in the + * operation because it is possible for the IO context used for read + * operations to change while IO is ongoing, and this read should be + * counted under the IO context in which it was started. + */ + IOContext io_context; /* * The following private members are private state for communication @@ -245,12 +253,14 @@ extern Buffer ReadBufferWithoutRelcache(RelFileLocator rlocator, extern bool StartReadBuffer(ReadBuffersOperation *operation, Buffer *buffer, BlockNumber blocknum, - int flags); + int flags, + BufferAccessStrategy strategy); extern bool StartReadBuffers(ReadBuffersOperation *operation, Buffer *buffers, BlockNumber blockNum, int *nblocks, - int flags); + int flags, + BufferAccessStrategy strategy); extern bool WaitReadBuffers(ReadBuffersOperation *operation); extern void ReleaseBuffer(Buffer buffer); diff --git a/src/test/modules/test_aio/test_aio.c b/src/test/modules/test_aio/test_aio.c index 6270775af7c..ffe9258e1e8 100644 --- a/src/test/modules/test_aio/test_aio.c +++ b/src/test/modules/test_aio/test_aio.c @@ -737,14 +737,14 @@ read_buffers(PG_FUNCTION_ARGS) operation->rel = rel; operation->smgr = smgr; operation->persistence = rel->rd_rel->relpersistence; - operation->strategy = NULL; operation->forknum = MAIN_FORKNUM; io_reqds[nios] = StartReadBuffers(operation, &buffers[nblocks_done], startblock + nblocks_done, &nblocks_this_io, - 0); + 0, + NULL); nblocks_per_io[nios] = nblocks_this_io; nios++; nblocks_done += nblocks_this_io; -- 2.47.3