Fix read_stream.c for changing io_combine_limit.

In a couple of places, read_stream.c assumed that io_combine_limit would
be stable during the lifetime of a stream.  That is not true in at least
one unusual case: streams held by CURSORs where you could change the GUC
between FETCH commands, with unpredictable results.

Fix, by storing stream->io_combine_limit and referring only to that
after construction.  This mirrors the treatment of the other important
setting {effective,maintenance}_io_concurrency, which is stored in
stream->max_ios.

One of the cases was the queue overflow space, which was sized for
io_combine_limit and could be overrun if the GUC was increased.  Since
that coding was a little hard to follow, also introduce a variable for
better readability instead of open-coding the arithmetic.  Doing so
revealed an off-by-one thinko while clamping max_pinned_buffers to
INT16_MAX, though that wasn't a live bug due to the current limits on
GUC values.

Back-patch to 17.

Discussion: https://postgr.es/m/CA%2BhUKG%2B2T9p-%2BzM6Eeou-RAJjTML6eit1qn26f9twznX59qtCA%40mail.gmail.com

diff --git a/src/backend/storage/aio/read_stream.c b/src/backend/storage/aio/read_stream.c
index 04bdb5e6d4b8b8c295611ca62393e604c85c5dd0..36fb9fe152cf4355aa91a0c712d010d3293cf20c 100644 (file)
--- a/src/backend/storage/aio/read_stream.c
+++ b/src/backend/storage/aio/read_stream.c
@@ -109,6 +109,7 @@ typedef struct InProgressIO
 struct ReadStream
 {
        int16           max_ios;
+       int16           io_combine_limit;
        int16           ios_in_progress;
        int16           queue_size;
        int16           max_pinned_buffers;
@@ -236,7 +237,7 @@ read_stream_start_pending_read(ReadStream *stream, bool suppress_advice)
 
        /* This should only be called with a pending read. */
        Assert(stream->pending_read_nblocks > 0);
-       Assert(stream->pending_read_nblocks <= io_combine_limit);
+       Assert(stream->pending_read_nblocks <= stream->io_combine_limit);
 
        /* We had better not exceed the pin limit by starting this read. */
        Assert(stream->pinned_buffers + stream->pending_read_nblocks <=
@@ -324,7 +325,7 @@ read_stream_look_ahead(ReadStream *stream, bool suppress_advice)
                int16           buffer_index;
                void       *per_buffer_data;
 
-               if (stream->pending_read_nblocks == io_combine_limit)
+               if (stream->pending_read_nblocks == stream->io_combine_limit)
                {
                        read_stream_start_pending_read(stream, suppress_advice);
                        suppress_advice = false;
@@ -384,7 +385,7 @@ read_stream_look_ahead(ReadStream *stream, bool suppress_advice)
         * signaled end-of-stream, we start the read immediately.
         */
        if (stream->pending_read_nblocks > 0 &&
-               (stream->pending_read_nblocks == io_combine_limit ||
+               (stream->pending_read_nblocks == stream->io_combine_limit ||
                 (stream->pending_read_nblocks == stream->distance &&
                  stream->pinned_buffers == 0) ||
                 stream->distance == 0) &&
@@ -415,6 +416,7 @@ read_stream_begin_impl(int flags,
        ReadStream *stream;
        size_t          size;
        int16           queue_size;
+       int16           queue_overflow;
        int                     max_ios;
        int                     strategy_pin_limit;
        uint32          max_pinned_buffers;
@@ -445,6 +447,14 @@ read_stream_begin_impl(int flags,
        /* Cap to INT16_MAX to avoid overflowing below */
        max_ios = Min(max_ios, PG_INT16_MAX);
 
+       /*
+        * If starting a multi-block I/O near the end of the queue, we might
+        * temporarily need extra space for overflowing buffers before they are
+        * moved to regular circular position.  This is the maximum extra space we
+        * could need.
+        */
+       queue_overflow = io_combine_limit - 1;
+
        /*
         * Choose the maximum number of buffers we're prepared to pin.  We try to
         * pin fewer if we can, though.  We add one so that we can make progress
@@ -459,7 +469,7 @@ read_stream_begin_impl(int flags,
         */
        max_pinned_buffers = (max_ios + 1) * io_combine_limit;
        max_pinned_buffers = Min(max_pinned_buffers,
-                                                        PG_INT16_MAX - io_combine_limit - 1);
+                                                        PG_INT16_MAX - queue_overflow - 1);
 
        /* Give the strategy a chance to limit the number of buffers we pin. */
        strategy_pin_limit = GetAccessStrategyPinLimit(strategy);
@@ -485,18 +495,17 @@ read_stream_begin_impl(int flags,
         * one big chunk.  Though we have queue_size buffers, we want to be able
         * to assume that all the buffers for a single read are contiguous (i.e.
         * don't wrap around halfway through), so we allow temporary overflows of
-        * up to the maximum possible read size by allocating an extra
-        * io_combine_limit - 1 elements.
+        * up to the maximum possible overflow size.
         */
        size = offsetof(ReadStream, buffers);
-       size += sizeof(Buffer) * (queue_size + io_combine_limit - 1);
+       size += sizeof(Buffer) * (queue_size + queue_overflow);
        size += sizeof(InProgressIO) * Max(1, max_ios);
        size += per_buffer_data_size * queue_size;
        size += MAXIMUM_ALIGNOF * 2;
        stream = (ReadStream *) palloc(size);
        memset(stream, 0, offsetof(ReadStream, buffers));
        stream->ios = (InProgressIO *)
-               MAXALIGN(&stream->buffers[queue_size + io_combine_limit - 1]);
+               MAXALIGN(&stream->buffers[queue_size + queue_overflow]);
        if (per_buffer_data_size > 0)
                stream->per_buffer_data = (void *)
                        MAXALIGN(&stream->ios[Max(1, max_ios)]);
@@ -523,7 +532,14 @@ read_stream_begin_impl(int flags,
        if (max_ios == 0)
                max_ios = 1;
 
+       /*
+        * Capture stable values for these two GUC-derived numbers for the
+        * lifetime of this stream, so we don't have to worry about the GUCs
+        * changing underneath us beyond this point.
+        */
        stream->max_ios = max_ios;
+       stream->io_combine_limit = io_combine_limit;
+
        stream->per_buffer_data_size = per_buffer_data_size;
        stream->max_pinned_buffers = max_pinned_buffers;
        stream->queue_size = queue_size;
@@ -537,7 +553,7 @@ read_stream_begin_impl(int flags,
         * doing full io_combine_limit sized reads (behavior B).
         */
        if (flags & READ_STREAM_FULL)
-               stream->distance = Min(max_pinned_buffers, io_combine_limit);
+               stream->distance = Min(max_pinned_buffers, stream->io_combine_limit);
        else
                stream->distance = 1;
 
@@ -752,14 +768,14 @@ read_stream_next_buffer(ReadStream *stream, void **per_buffer_data)
                else
                {
                        /* No advice; move towards io_combine_limit (behavior B). */
-                       if (stream->distance > io_combine_limit)
+                       if (stream->distance > stream->io_combine_limit)
                        {
                                stream->distance--;
                        }
                        else
                        {
                                distance = stream->distance * 2;
-                               distance = Min(distance, io_combine_limit);
+                               distance = Min(distance, stream->io_combine_limit);
                                distance = Min(distance, stream->max_pinned_buffers);
                                stream->distance = distance;
                        }