From ad66312a6d30b45423ffae0633cdfe687707370e Mon Sep 17 00:00:00 2001 From: Nazir Bilal Yavuz Date: Thu, 16 Jul 2026 15:20:11 +0300 Subject: [PATCH v1 2/3] Add read-stream backwards I/O combining Until now the read stream only combined consecutive blocks into one larger I/O when block numbers ascended. Backward scans hand the callback descending block numbers, so they issued one I/O per block. This teaches the stream to combine a run of descending blocks into a single read. The knowledge of backward combining lives entirely in the read stream: the lower layers have no notion of direction. StartReadBuffers() still reads a contiguous range forward, from the lowest block upward, and the read stream simply reverses the order in which it hands those buffers back to the caller. This comes with some limitations. The main one is that such a reversed range must be fully pinned and waited for before any of its buffers can be returned, so it can't overlap consumption the way forward reads do. See the comments in read_stream.c for the details. --- src/backend/storage/aio/read_stream.c | 629 ++++++++++++++---- src/test/modules/test_aio/meson.build | 1 + .../test_aio/t/005_read_stream_backward.pl | 226 +++++++ src/test/modules/test_aio/test_aio--1.0.sql | 2 +- src/test/modules/test_aio/test_aio.c | 37 +- src/tools/pgindent/typedefs.list | 1 + 6 files changed, 763 insertions(+), 133 deletions(-) create mode 100644 src/test/modules/test_aio/t/005_read_stream_backward.pl diff --git a/src/backend/storage/aio/read_stream.c b/src/backend/storage/aio/read_stream.c index 125d8babf41..96c5f495bd6 100644 --- a/src/backend/storage/aio/read_stream.c +++ b/src/backend/storage/aio/read_stream.c @@ -89,6 +89,16 @@ typedef struct InProgressIO ReadBuffersOperation op; } InProgressIO; +/* + * A range of buffers in the queue that must be returned to the caller in + * reverse order, to support backward scans. + */ +typedef struct ReadStreamReverseRange +{ + int16 buffer_index; + int16 nblocks; +} ReadStreamReverseRange; + /* * State for managing a stream of reads. */ @@ -148,11 +158,26 @@ struct ReadStream /* The read operation we are currently preparing. */ BlockNumber pending_read_blocknum; int16 pending_read_nblocks; + bool pending_read_combine; + bool pending_read_reverse; /* Space for buffers and optional per-buffer private data. */ size_t per_buffer_data_size; void *per_buffer_data; + /* Queue of block ranges that need to be reversed before returning. */ + ReadStreamReverseRange *reverse_ranges; + int16 reverse_ranges_size; + int16 reverse_ranges_count; + int16 oldest_reverse_range_index; + int16 next_reverse_range_index; + + /* Cursor for blocks being returned from the queue in reverse order. */ + int16 reverse_buffer_nblocks; + int16 reverse_buffer_count; + int16 reverse_buffer_index; + int16 reverse_pbd_index; + /* Read operations that have been started but not waited for yet. */ InProgressIO *ios; int16 oldest_io_index; @@ -305,6 +330,51 @@ read_stream_unget_block(ReadStream *stream, BlockNumber blocknum) stream->buffered_blocknum = blocknum; } +/* + * Wait for the oldest in-progress I/O to complete in order to free one I/O + * slot, without advancing the consumer's oldest_buffer_index. + * + * This exists only to guarantee forward progress while completing a reverse + * range: such a range must be fully pinned before the consumer reaches it, so + * we cannot stop part way through even if pinning the remaining blocks needs + * more concurrent I/Os than max_ios allows. The buffers belonging to the + * waited-for I/O stay pinned in the queue and are returned to the consumer + * later; because we advance oldest_io_index past this I/O, the consumer will + * not wait for it again. + */ +static void +read_stream_wait_oldest_io(ReadStream *stream) +{ + int16 io_index = stream->oldest_io_index; + + Assert(stream->ios_in_progress > 0); + + /* + * If the stream has been reset, don't even wait for the I/O, just abandon + * it. + */ + if (stream->readahead_distance < 0) + AbandonReadBuffers(&stream->ios[io_index].op); + else if (WaitReadBuffers(&stream->ios[io_index].op) || + (stream->ios[io_index].op.flags & READ_BUFFERS_SYNCHRONOUSLY)) + read_stream_count_wait(stream); + + stream->ios_in_progress--; + if (++stream->oldest_io_index == stream->max_ios) + stream->oldest_io_index = 0; + + /* We did I/O, so hold off distance decay, like the normal wait path. */ + stream->distance_decay_holdoff = stream->max_pinned_buffers; + + /* + * If we've reached the first block of a sequential region we're issuing + * advice for, cancel that until the next jump. + */ + if (stream->advice_enabled && + stream->ios[io_index].op.blocknum == stream->seq_until_processed) + stream->seq_until_processed = InvalidBlockNumber; +} + /* * Start as much of the current pending read as we can. If we have to split it * because of the per-backend buffer limit, or the buffer manager decides to @@ -319,6 +389,7 @@ static bool read_stream_start_pending_read(ReadStream *stream) { bool need_wait; + bool completing_reverse; int requested_nblocks; int nblocks; int flags; @@ -332,6 +403,16 @@ read_stream_start_pending_read(ReadStream *stream) Assert(stream->pending_read_nblocks > 0); Assert(stream->pending_read_nblocks <= stream->io_combine_limit); + /* + * Are we completing the still-pending tail of a reverse range? Such a + * range has pending_read_combine cleared until the whole range has been + * started. Its leading blocks are already pinned but cannot be consumed + * until the entire range is pinned, so the usual "give the consumer what + * we have first" logic does not apply, and we must force progress even + * when other backend pins have exhausted the per-backend limit. + */ + completing_reverse = !stream->pending_read_combine; + /* We had better not exceed the per-stream buffer limit with this read. */ Assert(stream->pinned_buffers + stream->pending_read_nblocks <= stream->max_pinned_buffers); @@ -390,6 +471,16 @@ read_stream_start_pending_read(ReadStream *stream) } } + /* + * When completing the tail of a reverse range we must be able to start + * this read even if all I/O slots are in use, because the range has to be + * fully pinned before the consumer reaches it. Free one slot by waiting + * for the oldest in-progress I/O; the loop that drives reverse completion + * is guaranteed to terminate because each such wait retires one I/O. + */ + if (completing_reverse && stream->ios_in_progress == stream->max_ios) + read_stream_wait_oldest_io(stream); + /* * How many more buffers is this backend allowed? * @@ -408,7 +499,8 @@ read_stream_start_pending_read(ReadStream *stream) buffer_limit += stream->forwarded_buffers; buffer_limit = Min(buffer_limit, PG_INT16_MAX); - if (buffer_limit == 0 && stream->pinned_buffers == 0) + if (buffer_limit == 0 && + (stream->pinned_buffers == 0 || completing_reverse)) buffer_limit = 1; /* guarantee progress */ /* Does the per-backend limit affect this read? */ @@ -422,14 +514,51 @@ read_stream_start_pending_read(ReadStream *stream) if (stream->readahead_distance > new_distance) stream->readahead_distance = new_distance; - /* Unless we have nothing to give the consumer, stop here. */ - if (stream->pinned_buffers > 0) + /* + * Unless we have nothing consumable to give the consumer, stop here. + * When completing the tail of a reverse range the already-pinned + * buffers cannot be consumed yet, so we must continue with a short + * read to guarantee the range eventually becomes fully pinned. + */ + if (stream->pinned_buffers > 0 && !completing_reverse) return false; /* A short read is required to make progress. */ nblocks = buffer_limit; } + /* Do we need to remember to reverse these blocks later? */ + if (stream->pending_read_nblocks > 1 && stream->pending_read_reverse) + { + /* + * All blocks in this range must be waited for and reversed before + * returning any of them, no matter how many IOs it takes. + * + * We record the length of the *entire* pending run, not just the + * blocks this particular StartReadBuffers() call will start. The + * per-backend pin limit (buffer_limit) may have shortened this read + * to fewer blocks than the run, in which case the remaining blocks + * are pinned by the follow-up "completing_reverse" reads driven from + * read_stream_look_ahead(). Those follow-up reads deliberately do + * not record a range of their own (pending_read_reverse is cleared + * just below), so this single range must cover the whole contiguous + * run or the tail would be returned to the consumer in forward order. + */ + stream->reverse_ranges[stream->next_reverse_range_index].buffer_index = + stream->next_buffer_index; + stream->reverse_ranges[stream->next_reverse_range_index].nblocks = + stream->pending_read_nblocks; + + if (++stream->next_reverse_range_index == stream->reverse_ranges_size) + stream->next_reverse_range_index = 0; + stream->reverse_ranges_count++; + Assert(stream->reverse_ranges_count <= stream->reverse_ranges_size); + + stream->pending_read_reverse = false; + /* No more I/O combining until this whole pending range is started. */ + stream->pending_read_combine = false; + } + /* * We say how many blocks we want to read, but it may be smaller on return * if the buffer manager decides to shorten the read. Initialize buffers @@ -544,6 +673,13 @@ read_stream_start_pending_read(ReadStream *stream) stream->pending_read_blocknum += nblocks; stream->pending_read_nblocks -= nblocks; + /* + * If I/O combining was disabled while dealing with a reversed block + * range, re-enable it as soon as the pending read is fully cleared. + */ + if (stream->pending_read_nblocks == 0) + stream->pending_read_combine = true; + return true; } @@ -698,19 +834,66 @@ read_stream_look_ahead(ReadStream *stream) break; } - /* Can we merge it with the pending read? */ - if (stream->pending_read_nblocks > 0 && - stream->pending_read_blocknum + stream->pending_read_nblocks == blocknum) + /* + * Can we combine this block with the pending read? + * + * pending_read_blocknum always tracks the lowest block number in the + * pending read. For reverse-direction reads we decrement it when + * prepending, so that StartReadBuffers() can always start from the + * lowest block number and read forward to disk; the resulting buffers + * are then returned to the caller in reverse order. + */ + if (stream->pending_read_nblocks > 0 && stream->pending_read_combine) { - stream->pending_read_nblocks++; - continue; + if (blocknum == stream->pending_read_blocknum + stream->pending_read_nblocks) + { + /* Append block number. */ + if (stream->pending_read_nblocks == 1) + stream->pending_read_reverse = false; + if (!stream->pending_read_reverse) + { + stream->pending_read_nblocks++; + continue; + } + } + else if (blocknum + 1 == stream->pending_read_blocknum && + stream->forwarded_buffers == 0) + { + /* + * Prepend block number. + * + * We must not prepend while there are forwarded buffers: + * those buffers are already pinned in the queue starting at + * next_buffer_index and correspond to pending_read_blocknum + * and the blocks above it. Decrementing + * pending_read_blocknum here would break that correspondence, + * so we instead fall through and start the pending read + * first. + */ + if (stream->pending_read_nblocks == 1) + stream->pending_read_reverse = true; + if (stream->pending_read_reverse) + { + stream->pending_read_nblocks++; + stream->pending_read_blocknum--; + continue; + } + } } /* We have to start the pending read before we can build another. */ while (stream->pending_read_nblocks > 0) { + /* + * A reverse range must be pinned in full before we can stop, so + * keep going even at the I/O limit; + * read_stream_start_pending_read() frees an I/O slot for us when + * necessary and always makes progress in that state. For a + * normal read we stop once we've hit the buffer or I/O limit. + */ if (!read_stream_start_pending_read(stream) || - stream->ios_in_progress == stream->max_ios) + (stream->pending_read_combine && + stream->ios_in_progress == stream->max_ios)) { /* We've hit the buffer or I/O limit. Rewind and stop here. */ read_stream_unget_block(stream, blocknum); @@ -723,6 +906,8 @@ read_stream_look_ahead(ReadStream *stream) /* This is the start of a new pending read. */ stream->pending_read_blocknum = blocknum; stream->pending_read_nblocks = 1; + stream->pending_read_reverse = false; + stream->pending_read_combine = true; } /* @@ -735,6 +920,28 @@ read_stream_look_ahead(ReadStream *stream) if (read_stream_should_issue_now(stream)) read_stream_start_pending_read(stream); + /* + * A reverse range must be fully pinned before the consumer reaches it, + * because read_stream_next_buffer() expects every buffer of the range to + * be valid in the queue. Starting a reverse read (here or earlier in + * this pass, e.g. from the issue-now shortcut at the top of the + * look-ahead loop) may have been shortened by StartReadBuffers() or the + * per-backend buffer limit, leaving the tail of the range still pending. + * pending_read_combine is cleared exactly while a reverse range is + * mid-flight and re-enabled once it is fully started, so it tells us when + * we still owe blocks to such a range. Finish pinning it now, regardless + * of how we got here. + * + * read_stream_start_pending_read() guarantees at least one block of + * forward progress per call in this state (see its handling of + * completing_reverse), even under buffer pressure or when max_ios has + * been reached, so this loop always terminates with the range fully + * pinned. + */ + while (stream->pending_read_nblocks && + !stream->pending_read_combine) + read_stream_start_pending_read(stream); + /* * There should always be something pinned when we leave this function, * whether started by this call or not, unless we've hit the end of the @@ -876,12 +1083,16 @@ read_stream_begin_impl(int flags, size = offsetof(ReadStream, buffers); size += sizeof(Buffer) * (queue_size + queue_overflow); size += sizeof(InProgressIO) * Max(1, max_ios); + size += sizeof(ReadStreamReverseRange) * (queue_size / 2); size += per_buffer_data_size * queue_size; - size += MAXIMUM_ALIGNOF * 2; + size += MAXIMUM_ALIGNOF * 3; stream = (ReadStream *) palloc(size); memset(stream, 0, offsetof(ReadStream, buffers)); - stream->ios = (InProgressIO *) + stream->reverse_ranges_size = queue_size / 2; + stream->reverse_ranges = (ReadStreamReverseRange *) MAXALIGN(&stream->buffers[queue_size + queue_overflow]); + stream->ios = (InProgressIO *) + MAXALIGN(&stream->reverse_ranges[stream->reverse_ranges_size]); if (per_buffer_data_size > 0) stream->per_buffer_data = (void *) MAXALIGN(&stream->ios[Max(1, max_ios)]); @@ -933,6 +1144,8 @@ read_stream_begin_impl(int flags, stream->seq_until_processed = InvalidBlockNumber; stream->temporary = SmgrIsTemp(smgr); stream->distance_decay_holdoff = 0; + stream->pending_read_reverse = false; + stream->pending_read_combine = true; /* * Skip the initial ramp-up phase if the caller says we're going to be @@ -1018,6 +1231,212 @@ read_stream_begin_smgr_relation(int flags, per_buffer_data_size); } +/* + * Wait for the I/O associated with the buffer at oldest_buffer_index (if + * any), update look-ahead bookkeeping, and advance oldest_buffer_index by one + * with wrap-around. Used both for the normal forward path and for waiting on + * each buffer of a reversed range. The caller is responsible for adjusting + * stream->pinned_buffers and for zapping queue slots. + */ +static void +read_stream_wait_advance_oldest(ReadStream *stream) +{ + int16 oldest_buffer_index = stream->oldest_buffer_index; + + /* Do we have to wait for an associated I/O first? */ + if (stream->ios_in_progress > 0 && + stream->ios[stream->oldest_io_index].buffer_index == oldest_buffer_index) + { + int16 io_index = stream->oldest_io_index; + bool needed_wait; + + /* Sanity check that we still agree on the buffers. */ + Assert(stream->ios[io_index].op.buffers == + &stream->buffers[oldest_buffer_index]); + + /* + * If the stream has been reset, don't even wait for the IO, just + * abandon it. + */ + if (stream->readahead_distance < 0) + { + AbandonReadBuffers(&stream->ios[io_index].op); + needed_wait = false; + } + else + needed_wait = WaitReadBuffers(&stream->ios[io_index].op); + + Assert(stream->ios_in_progress > 0); + stream->ios_in_progress--; + if (++stream->oldest_io_index == stream->max_ios) + stream->oldest_io_index = 0; + + /* + * If the IO was executed synchronously, we will never see + * WaitReadBuffers() block. Treat it as if it did block. This is + * particularly crucial when effective_io_concurrency=0 is used, as + * all IO will be synchronous. Without treating synchronous IO as + * having waited, we'd never allow the distance to get large enough to + * allow for IO combining, resulting in bad performance. + */ + if (stream->ios[io_index].op.flags & READ_BUFFERS_SYNCHRONOUSLY) + needed_wait = true; + + /* Count it as a wait if we need to wait for IO */ + if (needed_wait) + read_stream_count_wait(stream); + + /* + * Have the read-ahead distance ramp up rapidly after we needed to + * wait for IO. We only increase the read-ahead-distance when we + * needed to wait, to avoid increasing the distance further than + * necessary, as looking ahead too far can be costly, both due to the + * cost of unnecessarily pinning many buffers and due to doing IOs + * that may never be consumed if the stream is ended/reset before + * completion. + * + * If we did not need to wait, the current distance was evidently + * sufficient. + * + * NB: Must not increase the distance if we already reached the end of + * the stream, as stream->readahead_distance == 0 is used to keep + * track of having reached the end. + */ + if (stream->readahead_distance > 0 && needed_wait) + { + /* wider temporary value, due to overflow risk */ + int32 readahead_distance; + + readahead_distance = stream->readahead_distance * 2; + readahead_distance = Min(readahead_distance, stream->max_pinned_buffers); + stream->readahead_distance = readahead_distance; + } + + /* + * As we needed IO, prevent distances from being reduced within our + * maximum look-ahead window. This avoids collapsing distances too + * quickly in workloads where most of the required blocks are cached, + * but where the remaining IOs are a sufficient enough factor to cause + * a substantial slowdown if executed synchronously. + * + * There are valid arguments for preventing decay for max_ios or for + * max_pinned_buffers. But the argument for max_pinned_buffers seems + * clearer - if we can't see any misses within the maximum look-ahead + * distance, we can't do any useful read-ahead. + */ + stream->distance_decay_holdoff = stream->max_pinned_buffers; + + /* + * Whether we needed to wait or not, allow for more IO combining if we + * needed to do IO. The reason to do so independent of needing to wait + * is that when the data is resident in the kernel page cache, IO + * combining reduces the syscall / dispatch overhead, making it + * worthwhile regardless of needing to wait. + * + * It is also important with io_uring as it will never signal the need + * to wait for reads if all the data is in the page cache. There are + * heuristics to deal with that in method_io_uring.c, but they only + * work when the IO gets large enough. + */ + if (stream->combine_distance > 0 && + stream->combine_distance < stream->io_combine_limit) + { + /* wider temporary value, due to overflow risk */ + int32 combine_distance; + + combine_distance = stream->combine_distance * 2; + combine_distance = Min(combine_distance, stream->io_combine_limit); + combine_distance = Min(combine_distance, stream->max_pinned_buffers); + stream->combine_distance = combine_distance; + } + + /* + * If we've reached the first block of a sequential region we're + * issuing advice for, cancel that until the next jump. The kernel + * will see the sequential preadv() pattern starting here. + */ + if (stream->advice_enabled && + stream->ios[io_index].op.blocknum == stream->seq_until_processed) + stream->seq_until_processed = InvalidBlockNumber; + } + + /* Advance oldest buffer, with wrap-around. */ + stream->oldest_buffer_index++; + if (stream->oldest_buffer_index == stream->queue_size) + stream->oldest_buffer_index = 0; +} + +/* + * Return the next buffer of an in-progress reverse range. The range was set + * up by read_stream_next_buffer() after waiting for all of its I/Os to + * complete; this helper walks the queue backwards over the range's buffer + * slots. + * + * The buffers were read forward from the lowest block, so they sit in the + * queue in ascending order and this cursor walks them back down. The + * per-buffer data, however, was populated by the callback in the order the + * blocks were emitted (descending), i.e. in ascending queue-slot order + * starting at the range's first slot. We therefore hand out per-buffer data + * with a separate cursor that advances forward while the buffer cursor moves + * backward, so each buffer is paired with the data the callback stored for it. + * + * Note that stream->pinned_buffers is not decremented per buffer here. The + * full range's worth of pins is released only once the final reversed buffer + * has been returned to the caller, which prevents look-ahead from trying to + * re-use the queue slots while their buffers are still being streamed back + * out. + */ +static Buffer +read_stream_next_buffer_reverse(ReadStream *stream, void **per_buffer_data) +{ + Buffer buffer; + int16 idx; + + Assert(stream->reverse_buffer_count > 0); + + idx = stream->reverse_buffer_index; + buffer = stream->buffers[idx]; + Assert(BufferIsValid(buffer)); + if (per_buffer_data) + { + *per_buffer_data = get_per_buffer_data(stream, stream->reverse_pbd_index); + + /* Walk the per-buffer data cursor forwards, with wrap-around. */ + if (++stream->reverse_pbd_index == stream->queue_size) + stream->reverse_pbd_index = 0; + } + + /* + * Zap the queue entry, or else it would later look like a forwarded + * buffer. Also zap the overflow copy if applicable. + */ + stream->buffers[idx] = InvalidBuffer; + if (idx < stream->io_combine_limit - 1) + stream->buffers[stream->queue_size + idx] = InvalidBuffer; + + /* Walk reverse cursor backwards, with wrap-around. */ + if (idx == 0) + stream->reverse_buffer_index = stream->queue_size - 1; + else + stream->reverse_buffer_index = idx - 1; + + read_stream_count_prefetch(stream); + + /* + * Once the whole reversed range has been streamed back to the caller, + * release the pins all at once and resume look-ahead. + */ + if (--stream->reverse_buffer_count == 0) + { + Assert(stream->pinned_buffers >= stream->reverse_buffer_nblocks); + stream->pinned_buffers -= stream->reverse_buffer_nblocks; + stream->reverse_buffer_nblocks = 0; + read_stream_look_ahead(stream); + } + + return buffer; +} + /* * Pull one pinned buffer out of a stream. Each call returns successive * blocks in the order specified by the callback. If per_buffer_data_size was @@ -1033,6 +1452,13 @@ read_stream_next_buffer(ReadStream *stream, void **per_buffer_data) Buffer buffer; int16 oldest_buffer_index; + /* + * If we are in the middle of streaming a reversed range back to the + * caller, just return the next buffer from the reverse cursor. + */ + if (unlikely(stream->reverse_buffer_count > 0)) + return read_stream_next_buffer_reverse(stream, per_buffer_data); + #ifndef READ_STREAM_DISABLE_FAST_PATH /* @@ -1172,128 +1598,65 @@ read_stream_next_buffer(ReadStream *stream, void **per_buffer_data) oldest_buffer_index = stream->oldest_buffer_index; Assert(oldest_buffer_index >= 0 && oldest_buffer_index < stream->queue_size); - buffer = stream->buffers[oldest_buffer_index]; - if (per_buffer_data) - *per_buffer_data = get_per_buffer_data(stream, oldest_buffer_index); - - Assert(BufferIsValid(buffer)); - /* Do we have to wait for an associated I/O first? */ - if (stream->ios_in_progress > 0 && - stream->ios[stream->oldest_io_index].buffer_index == oldest_buffer_index) + /* + * Is this the start of a range that needs to be returned in reverse + * order? If so, wait for the I/O for every buffer in the range to + * complete, then set up the reverse cursor and stream the buffers back + * out in reverse order. + * + * We send block numbers to the lower layers in forward order even for + * reverse scans, so the underlying I/O is always a simple forward read. + * The reversal happens only when we hand the buffers back to the caller. + */ + if (unlikely(stream->reverse_ranges_count > 0 && + stream->reverse_ranges[stream->oldest_reverse_range_index].buffer_index == + oldest_buffer_index)) { - int16 io_index = stream->oldest_io_index; - bool needed_wait; - - /* Sanity check that we still agree on the buffers. */ - Assert(stream->ios[io_index].op.buffers == - &stream->buffers[oldest_buffer_index]); - - /* - * If the stream has been reset, don't even wait for the IO, just - * abandon it. - */ - if (stream->readahead_distance < 0) - { - AbandonReadBuffers(&stream->ios[io_index].op); - needed_wait = false; - } - else - needed_wait = WaitReadBuffers(&stream->ios[io_index].op); - - Assert(stream->ios_in_progress > 0); - stream->ios_in_progress--; - if (++stream->oldest_io_index == stream->max_ios) - stream->oldest_io_index = 0; - - /* - * If the IO was executed synchronously, we will never see - * WaitReadBuffers() block. Treat it as if it did block. This is - * particularly crucial when effective_io_concurrency=0 is used, as - * all IO will be synchronous. Without treating synchronous IO as - * having waited, we'd never allow the distance to get large enough to - * allow for IO combining, resulting in bad performance. - */ - if (stream->ios[io_index].op.flags & READ_BUFFERS_SYNCHRONOUSLY) - needed_wait = true; + int16 nblocks; + int32 last_idx; - /* Count it as a wait if we need to wait for IO */ - if (needed_wait) - read_stream_count_wait(stream); + nblocks = stream->reverse_ranges[stream->oldest_reverse_range_index].nblocks; + Assert(nblocks > 1); - /* - * Have the read-ahead distance ramp up rapidly after we needed to - * wait for IO. We only increase the read-ahead-distance when we - * needed to wait, to avoid increasing the distance further than - * necessary, as looking ahead too far can be costly, both due to the - * cost of unnecessarily pinning many buffers and due to doing IOs - * that may never be consumed if the stream is ended/reset before - * completion. - * - * If we did not need to wait, the current distance was evidently - * sufficient. - * - * NB: Must not increase the distance if we already reached the end of - * the stream, as stream->readahead_distance == 0 is used to keep - * track of having reached the end. - */ - if (stream->readahead_distance > 0 && needed_wait) - { - /* wider temporary value, due to overflow risk */ - int32 readahead_distance; - - readahead_distance = stream->readahead_distance * 2; - readahead_distance = Min(readahead_distance, stream->max_pinned_buffers); - stream->readahead_distance = readahead_distance; - } + /* Wait for I/O for every buffer in the range to be finished. */ + for (int i = 0; i < nblocks; i++) + read_stream_wait_advance_oldest(stream); /* - * As we needed IO, prevent distances from being reduced within our - * maximum look-ahead window. This avoids collapsing distances too - * quickly in workloads where most of the required blocks are cached, - * but where the remaining IOs are a sufficient enough factor to cause - * a substantial slowdown if executed synchronously. - * - * There are valid arguments for preventing decay for max_ios or for - * max_pinned_buffers. But the argument for max_pinned_buffers seems - * clearer - if we can't see any misses within the maximum look-ahead - * distance, we can't do any useful read-ahead. + * The last buffer in the range will be streamed first, walking the + * buffer cursor backwards. The per-buffer data was populated in + * emission (descending) order starting at the range's first slot, so + * its cursor starts there and walks forwards; see + * read_stream_next_buffer_reverse(). */ - stream->distance_decay_holdoff = stream->max_pinned_buffers; + last_idx = (int32) oldest_buffer_index + nblocks - 1; + if (last_idx >= stream->queue_size) + last_idx -= stream->queue_size; + Assert(last_idx >= 0 && last_idx < stream->queue_size); + + stream->reverse_buffer_nblocks = nblocks; + stream->reverse_buffer_count = nblocks; + stream->reverse_buffer_index = (int16) last_idx; + stream->reverse_pbd_index = oldest_buffer_index; + + /* Discard this range from the queue. */ + if (++stream->oldest_reverse_range_index == stream->reverse_ranges_size) + stream->oldest_reverse_range_index = 0; + stream->reverse_ranges_count--; + + return read_stream_next_buffer_reverse(stream, per_buffer_data); + } - /* - * Whether we needed to wait or not, allow for more IO combining if we - * needed to do IO. The reason to do so independent of needing to wait - * is that when the data is resident in the kernel page cache, IO - * combining reduces the syscall / dispatch overhead, making it - * worthwhile regardless of needing to wait. - * - * It is also important with io_uring as it will never signal the need - * to wait for reads if all the data is in the page cache. There are - * heuristics to deal with that in method_io_uring.c, but they only - * work when the IO gets large enough. - */ - if (stream->combine_distance > 0 && - stream->combine_distance < stream->io_combine_limit) - { - /* wider temporary value, due to overflow risk */ - int32 combine_distance; + /* Otherwise, return the oldest buffer in forward order. */ + buffer = stream->buffers[oldest_buffer_index]; + if (per_buffer_data) + *per_buffer_data = get_per_buffer_data(stream, oldest_buffer_index); - combine_distance = stream->combine_distance * 2; - combine_distance = Min(combine_distance, stream->io_combine_limit); - combine_distance = Min(combine_distance, stream->max_pinned_buffers); - stream->combine_distance = combine_distance; - } + Assert(BufferIsValid(buffer)); - /* - * If we've reached the first block of a sequential region we're - * issuing advice for, cancel that until the next jump. The kernel - * will see the sequential preadv() pattern starting here. - */ - if (stream->advice_enabled && - stream->ios[io_index].op.blocknum == stream->seq_until_processed) - stream->seq_until_processed = InvalidBlockNumber; - } + /* Wait for any associated I/O and advance oldest_buffer_index. */ + read_stream_wait_advance_oldest(stream); /* * We must zap this queue entry, or else it would appear as a forwarded @@ -1339,11 +1702,6 @@ read_stream_next_buffer(ReadStream *stream, void **per_buffer_data) Assert(stream->pinned_buffers > 0); stream->pinned_buffers--; - /* Advance oldest buffer, with wrap-around. */ - stream->oldest_buffer_index++; - if (stream->oldest_buffer_index == stream->queue_size) - stream->oldest_buffer_index = 0; - /* Prepare for the next call. */ read_stream_look_ahead(stream); @@ -1355,7 +1713,9 @@ read_stream_next_buffer(ReadStream *stream, void **per_buffer_data) stream->readahead_distance == 1 && stream->combine_distance == 1 && stream->pending_read_nblocks == 0 && - stream->per_buffer_data_size == 0) + stream->per_buffer_data_size == 0 && + stream->reverse_ranges_count == 0 && + stream->reverse_buffer_count == 0) { /* * The fast path spins on one buffer entry repeatedly instead of @@ -1462,6 +1822,17 @@ read_stream_reset(ReadStream *stream) Assert(stream->forwarded_buffers == 0); Assert(stream->pinned_buffers == 0); Assert(stream->ios_in_progress == 0); + Assert(stream->reverse_ranges_count == 0); + Assert(stream->reverse_buffer_count == 0); + + /* Reset reverse-range bookkeeping for the next round. */ + stream->oldest_reverse_range_index = 0; + stream->next_reverse_range_index = 0; + stream->reverse_buffer_nblocks = 0; + stream->reverse_buffer_index = 0; + stream->reverse_pbd_index = 0; + stream->pending_read_reverse = false; + stream->pending_read_combine = true; /* Start off assuming data is cached. */ stream->readahead_distance = 1; diff --git a/src/test/modules/test_aio/meson.build b/src/test/modules/test_aio/meson.build index 909f81d96c1..73d7be49a70 100644 --- a/src/test/modules/test_aio/meson.build +++ b/src/test/modules/test_aio/meson.build @@ -34,6 +34,7 @@ tests += { 't/002_io_workers.pl', 't/003_initdb.pl', 't/004_read_stream.pl', + 't/005_read_stream_backward.pl', ], }, } diff --git a/src/test/modules/test_aio/t/005_read_stream_backward.pl b/src/test/modules/test_aio/t/005_read_stream_backward.pl new file mode 100644 index 00000000000..d88b333c5ee --- /dev/null +++ b/src/test/modules/test_aio/t/005_read_stream_backward.pl @@ -0,0 +1,226 @@ +# Copyright (c) 2025-2026, PostgreSQL Global Development Group +# +# Correctness tests for read-stream backward I/O combining +# +# The read stream combines a run of descending block numbers into one forward +# read, then hands the buffers back in reverse order. +# +# read_stream_for_blocks(rel, blocks[]) hands back one buffer per requested +# block, in stream order. We check that the stream returns exactly the +# requested blocks in the reversed order. With check_per_buffer_data => true +# the helper additionally verifies, inside the backend, that every buffer is +# the block that was requested and is paired with the per-buffer data the +# callback populated for it. Every pattern runs after evict_rel, so real I/O +# and backward combining happen. +# +# The reversal logic under test lives in read_stream.c and is independent of +# the io_method, so we run the tests under 'worker' io_method only. + +use strict; +use warnings FATAL => 'all'; + +use PostgreSQL::Test::Cluster; +use PostgreSQL::Test::Utils; +use Test::More; + +use FindBin; +use lib $FindBin::RealBin; + +use TestAio; + + +my $node = PostgreSQL::Test::Cluster->new('test'); +$node->init(); + +TestAio::configure($node); + +# Large shared_buffers and high effective_io_concurrency let the stream look +# ahead far enough to build large reverse ranges. +$node->append_conf( + 'postgresql.conf', qq( +max_connections=8 +io_method=worker +io_max_combine_limit=16 +)); + +$node->start(); + +my $nblocks = test_setup($node); +my $last = $nblocks - 1; + +$node->stop(); + +# Run all of the patterns +test_backward_main($node, $nblocks); + +# Buffer-starved cluster: exercises the "guarantee progress" path in +# read_stream_start_pending_read(), where the per-backend pin limit can drop to +# zero mid-range and the code must keep making one block of forward progress. +test_backward_low_buffers($node, $nblocks); + +# Backward I/O combining under per-backend pin pressure +test_backward_pin_pressure($node, $nblocks); + +done_testing(); + + +# Create the test relation and return its size in blocks +sub test_setup +{ + my $node = shift; + + $node->safe_psql( + 'postgres', qq( +CREATE EXTENSION test_aio; + +CREATE TABLE largeish(k int not null) WITH (FILLFACTOR=10); +INSERT INTO largeish(k) SELECT generate_series(1, 40000); +)); + + my $blocks = $node->safe_psql('postgres', + q{SELECT pg_relation_size('largeish') / current_setting('block_size')::int8} + ); + note "largeish is $blocks blocks"; + + # Reference blocks well inside the relation; ensure it is big enough that + # the "spans more than the read-stream queue" pattern is meaningful. + ok($blocks >= 512, "setup: relation large enough ($blocks blocks)"); + + return $blocks; +} + + +# Build the [ label => \@blocks ] access patterns, sized to the relation. +# Covers ascending, descending, mixed, gapped, repeated, boundary and +# queue-wrapping cases). +sub access_patterns +{ + my ($want) = @_; + + my @patterns = ( + [ 'single' => [5] ], + [ 'asc_pair' => [ 0, 1 ] ], + [ 'desc_pair' => [ 1, 0 ] ], + # full-region backward scan + [ 'full_backward' => [ reverse(0 .. $last) ] ], + # Direction changes: ascending, descending, then ascending + [ 'mixed_dir' => [ 0 .. 6, reverse(0 .. 5), 0 .. 6 ] ], + # Zig-zag flipping direction every block + [ 'zigzag' => [ 5, 4, 5, 4, 5, 4 ] ], + # Descending run near block 0 then another at the relation's end + [ 'desc_two_regions' => [ 3, 2, 1, 0, $last, $last - 1, $last - 2 ] ], + ); + + return @patterns unless defined $want; + + my ($p) = grep { $_->[0] eq $want } @patterns; + die "unknown access pattern '$want'" unless $p; + return $p; +} + + +# Run one access pattern and return the block numbers the stream handed back, +# in order, as a "{...}" array literal. The caller compares this against the +# requested blocks, confirming the stream yields exactly those blocks in the +# expected order. +sub stream_and_verify +{ + my ($node, $rel, $blocks, $icl, $check_pbd) = @_; + + my $arr = 'ARRAY[' . join(',', @$blocks) . ']::int4[]'; + $check_pbd = $check_pbd ? 'true' : 'false'; + + # Evict first so real I/O and (backward) combining happen. + return $node->safe_psql( + 'postgres', qq{ +SET io_combine_limit = $icl; +DO \$\$ BEGIN PERFORM evict_rel('$rel'); END \$\$; +SELECT array_agg(blocknum ORDER BY blockoff)::text +FROM read_stream_for_blocks('$rel', $arr, $check_pbd); +}); +} + + +sub test_backward_main +{ + my ($node, $nblocks) = @_; + + my @patterns = access_patterns(); + + $node->start(); + + foreach my $icl (1, 8, 16) + { + foreach my $pattern (@patterns) + { + my ($label, $blocks) = @$pattern; + my $want = '{' . join(',', @$blocks) . '}'; + + my $res = stream_and_verify($node, 'largeish', $blocks, $icl, 1); + is($res, $want, "icl=$icl $label"); + } + } + + $node->stop(); +} + + +# Buffer-starved cluster: verify a long backward combine still terminates and +# returns the right blocks, in order, under a tight per-backend pin budget. +sub test_backward_low_buffers +{ + my ($node, $nblocks) = @_; + + # 128kB == 16 shared buffers: enough to start, but too few to hold a long + # reverse range, so the per-backend pin limit binds and the + # guarantee-progress path is exercised repeatedly. + $node->append_conf( + 'postgresql.conf', qq( +max_connections=8 +shared_buffers=128kB +)); + $node->start(); + + my ($label, $blocks) = @{ access_patterns('full_backward') }; + my $want = '{' . join(',', @$blocks) . '}'; + + foreach my $icl (8, 16) + { + my $res = stream_and_verify($node, 'largeish', $blocks, $icl); + is($res, $want, + "low shared_buffers icl=$icl long backward scan completes"); + } + + $node->stop(); +} + +# Verify backward I/O combining under a small per-backend pin limit. +# +# The pin limit is GetAdditionalPinLimit(), derived from MaxProportionalPins = +# NBuffers / (MaxBackends + NUM_AUXILIARY_PROCS). A high max_connections makes +# it tiny even with plenty of free buffers. +sub test_backward_pin_pressure +{ + my ($node, $nblocks) = @_; + + # 32MB == 4096 shared buffers keeps a few hundred blocks resident, but + # max_connections=600 drives MaxProportionalPins to a single-digit value, so + # GetAdditionalPinLimit() stays far below io_combine_limit. + $node->append_conf( + 'postgresql.conf', qq( +max_connections=600 +shared_buffers=32MB +)); + $node->start(); + + my ($label, $blocks) = @{ access_patterns('full_backward') }; + my $want = '{' . join(',', @$blocks) . '}'; + + foreach my $icl (8, 16) + { + my $res = stream_and_verify($node, 'largeish', $blocks, $icl); + is($res, $want, "pin pressure icl=$icl $label"); + } + + $node->stop(); +} diff --git a/src/test/modules/test_aio/test_aio--1.0.sql b/src/test/modules/test_aio/test_aio--1.0.sql index ff3b200cff3..0faddf93d02 100644 --- a/src/test/modules/test_aio/test_aio--1.0.sql +++ b/src/test/modules/test_aio/test_aio--1.0.sql @@ -60,7 +60,7 @@ AS 'MODULE_PATHNAME' LANGUAGE C; /* * Read stream related functions */ -CREATE FUNCTION read_stream_for_blocks(rel regclass, blocks int4[], OUT blockoff int4, OUT blocknum int4, OUT buf int4) +CREATE FUNCTION read_stream_for_blocks(rel regclass, blocks int4[], check_per_buffer_data bool DEFAULT false, OUT blockoff int4, OUT blocknum int4, OUT buf int4) RETURNS SETOF record STRICT AS 'MODULE_PATHNAME' LANGUAGE C; diff --git a/src/test/modules/test_aio/test_aio.c b/src/test/modules/test_aio/test_aio.c index a1075e59a51..39637f68661 100644 --- a/src/test/modules/test_aio/test_aio.c +++ b/src/test/modules/test_aio/test_aio.c @@ -860,10 +860,22 @@ read_stream_for_blocks_cb(ReadStream *stream, void *per_buffer_data) { BlocksReadStreamData *stream_data = callback_private_data; + BlockNumber blocknum; if (stream_data->curblock >= stream_data->nblocks) return InvalidBlockNumber; - return stream_data->blocks[stream_data->curblock++]; + + blocknum = stream_data->blocks[stream_data->curblock++]; + + /* + * Stash the block number in the per-buffer data so that + * read_stream_for_blocks() can verify that each buffer is handed back + * paired with the per-buffer data the callback populated for it. + */ + if (per_buffer_data) + *((BlockNumber *) per_buffer_data) = blocknum; + + return blocknum; } PG_FUNCTION_INFO_V1(read_stream_for_blocks); @@ -872,6 +884,7 @@ read_stream_for_blocks(PG_FUNCTION_ARGS) { Oid relid = PG_GETARG_OID(0); ArrayType *blocksarray = PG_GETARG_ARRAYTYPE_P(1); + bool check_per_buffer_data = PG_GETARG_BOOL(2); ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo; Relation rel; BlocksReadStreamData stream_data; @@ -895,23 +908,41 @@ read_stream_for_blocks(PG_FUNCTION_ARGS) rel = relation_open(relid, AccessShareLock); + /* + * When asked to check per-buffer data, request room for one block number + * per buffer. The callback stores each emitted block number there so we + * can verify below that the stream hands every buffer back paired with + * the per-buffer data the callback populated for it. + */ stream = read_stream_begin_relation(READ_STREAM_FULL, NULL, rel, MAIN_FORKNUM, read_stream_for_blocks_cb, &stream_data, - 0); + check_per_buffer_data ? + sizeof(BlockNumber) : 0); for (int i = 0; i < stream_data.nblocks; i++) { - Buffer buf = read_stream_next_buffer(stream, NULL); + void *per_buffer_data = NULL; + Buffer buf = read_stream_next_buffer(stream, + check_per_buffer_data ? + &per_buffer_data : NULL); Datum values[3] = {0}; bool nulls[3] = {0}; if (!BufferIsValid(buf)) elog(ERROR, "read_stream_next_buffer() call %d is unexpectedly invalid", i); + if (check_per_buffer_data) + { + if (*((BlockNumber *) per_buffer_data) != BufferGetBlockNumber(buf)) + elog(ERROR, "read_stream_next_buffer() call %d paired block %u with per-buffer data for block %u", + i, BufferGetBlockNumber(buf), + *((BlockNumber *) per_buffer_data)); + } + values[0] = Int32GetDatum(i); values[1] = UInt32GetDatum(stream_data.blocks[i]); values[2] = UInt32GetDatum(buf); diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index 92c84570788..9a043a52a74 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -2573,6 +2573,7 @@ ReadLocalXLogPageNoWaitPrivate ReadReplicationSlotCmd ReadStream ReadStreamBlockNumberCB +ReadStreamReverseRange ReassignOwnedStmt RecheckForeignScan_function RecordCacheArrayEntry -- 2.47.3