src/backend/storage/aio/read_stream.c - master - githubmirror/postgres

src/backend/storage/aio/read_stream.c
1 492 строки50 KB
Melanie Plageman
Restore vacuum failsafe abandonment of buffer access strategy
07 авг 2026, 00:22
07 авг 2026, 00:22112c268
/*-------------------------------------------------------------------------
 *
 * read_stream.c
 *	  Mechanism for accessing buffered relation data with look-ahead
 *
 * Code that needs to access relation data typically pins blocks one at a
 * time, often in a predictable order that might be sequential or data-driven.
 * Calling the simple ReadBuffer() function for each block is inefficient,
 * because blocks that are not yet in the buffer pool require I/O operations
 * that are small and might stall waiting for storage.  This mechanism looks
 * into the future and calls StartReadBuffers() and WaitReadBuffers() to read
 * neighboring blocks together and ahead of time, with an adaptive look-ahead
 * distance.
 *
 * A user-provided callback generates a stream of block numbers that is used
 * to form reads of up to io_combine_limit, by attempting to merge them with a
 * pending read.  When that isn't possible, the existing pending read is sent
 * to StartReadBuffers() so that a new one can begin to form.
 *
 * The algorithm for controlling the look-ahead distance is based on recent
 * cache / miss history, as well as whether we need to wait for I/O completion
 * after a miss.  When no I/O is necessary, there is no benefit in looking
 * ahead more than one block.  This is the default initial assumption.  When
 * blocks needing I/O are streamed, the combine distance is increased to
 * benefit from I/O combining and the read-ahead distance is increased
 * whenever we need to wait for I/O to try to benefit from increased I/O
 * concurrency. Both are reduced gradually when cached blocks are streamed.
 *
 * The main data structure is a circular queue of buffers of size
 * max_pinned_buffers plus some extra space for technical reasons, ready to be
 * returned by read_stream_next_buffer().  Each buffer also has an optional
 * variable sized object that is passed from the callback to the consumer of
 * buffers.
 *
 * Parallel to the queue of buffers, there is a circular queue of in-progress
 * I/Os that have been started with StartReadBuffers(), and for which
 * WaitReadBuffers() must be called before returning the buffer.
 *
 * For example, if the callback returns block numbers 10, 42, 43, 44, 60 in
 * successive calls, then these data structures might appear as follows:
 *
 *                          buffers buf/data       ios
 *
 *                          +----+  +-----+       +--------+
 *                          |    |  |     |  +----+ 42..44 | <- oldest_io_index
 *                          +----+  +-----+  |    +--------+
 *   oldest_buffer_index -> | 10 |  |  ?  |  | +--+ 60..60 |
 *                          +----+  +-----+  | |  +--------+
 *                          | 42 |  |  ?  |<-+ |  |        | <- next_io_index
 *                          +----+  +-----+    |  +--------+
 *                          | 43 |  |  ?  |    |  |        |
 *                          +----+  +-----+    |  +--------+
 *                          | 44 |  |  ?  |    |  |        |
 *                          +----+  +-----+    |  +--------+
 *                          | 60 |  |  ?  |<---+
 *                          +----+  +-----+
 *     next_buffer_index -> |    |  |     |
 *                          +----+  +-----+
 *
 * In the example, 5 buffers are pinned, and the next buffer to be streamed to
 * the client is block 10.  Block 10 was a hit and has no associated I/O, but
 * the range 42..44 requires an I/O wait before its buffers are returned, as
 * does block 60.
 *
 *
 * Portions Copyright (c) 2024-2026, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * IDENTIFICATION
 *	  src/backend/storage/aio/read_stream.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "miscadmin.h"
#include "executor/instrument_node.h"
#include "storage/aio.h"
#include "storage/fd.h"
#include "storage/smgr.h"
#include "storage/read_stream.h"
#include "utils/memdebug.h"
#include "utils/rel.h"
#include "utils/spccache.h"

typedef struct InProgressIO
{
	int16		buffer_index;
	ReadBuffersOperation op;
} InProgressIO;

/*
 * State for managing a stream of reads.
 */
struct ReadStream
{
	int16		max_ios;
	int16		io_combine_limit;
	int16		ios_in_progress;
	int16		queue_size;
	int16		max_pinned_buffers;
	int16		forwarded_buffers;
	int16		pinned_buffers;

	/*
	 * Limit of how far, in blocks, to look-ahead for IO combining and for
	 * read-ahead.
	 *
	 * The limits for read-ahead and combining are handled separately to allow
	 * for IO combining even in cases where the I/O subsystem can keep up at a
	 * low read-ahead distance, as doing larger IOs is more efficient.
	 *
	 * Set to 0 when the end of the stream is reached.
	 */
	int16		combine_distance;
	int16		readahead_distance;
	uint16		distance_decay_holdoff;
	int16		initialized_buffers;
	int16		resume_readahead_distance;
	int16		resume_combine_distance;
	int			read_buffers_flags;
	bool		sync_mode;		/* using io_method=sync */
	bool		batch_mode;		/* READ_STREAM_USE_BATCHING */
	bool		advice_enabled;
	bool		temporary;

	/* scan stats counters */
	IOStats    *stats;

	/*
	 * One-block buffer to support 'ungetting' a block number, to resolve flow
	 * control problems when I/Os are split.
	 */
	BlockNumber buffered_blocknum;

	/*
	 * The callback that will tell us which block numbers to read, and an
	 * opaque pointer that will be pass to it for its own purposes.
	 */
	ReadStreamBlockNumberCB callback;
	void	   *callback_private_data;

	/* Next expected block, for detecting sequential access. */
	BlockNumber seq_blocknum;
	BlockNumber seq_until_processed;

	/* The read operation we are currently preparing. */
	BlockNumber pending_read_blocknum;
	int16		pending_read_nblocks;

	/* Space for buffers and optional per-buffer private data. */
	size_t		per_buffer_data_size;
	void	   *per_buffer_data;

	/* Read operations that have been started but not waited for yet. */
	InProgressIO *ios;
	int16		oldest_io_index;
	int16		next_io_index;

	bool		fast_path;

	/* Circular queue of buffers. */
	int16		oldest_buffer_index;	/* Next pinned buffer to return */
	int16		next_buffer_index;	/* Index of next buffer to pin */
	Buffer		buffers[FLEXIBLE_ARRAY_MEMBER];
};

/*
 * Return a pointer to the per-buffer data by index.
 */
static inline void *
get_per_buffer_data(ReadStream *stream, int16 buffer_index)
{
	return (char *) stream->per_buffer_data +
		stream->per_buffer_data_size * buffer_index;
}

/*
 * General-use ReadStreamBlockNumberCB for block range scans.  Loops over the
 * blocks [current_blocknum, last_exclusive).
 */
BlockNumber
block_range_read_stream_cb(ReadStream *stream,
						   void *callback_private_data,
						   void *per_buffer_data)
{
	BlockRangeReadStreamPrivate *p = callback_private_data;

	if (p->current_blocknum < p->last_exclusive)
		return p->current_blocknum++;

	return InvalidBlockNumber;
}

/*
 * Update stream stats with current pinned buffer depth.
 *
 * Called once per buffer returned to the consumer in read_stream_next_buffer().
 * Records the number of pinned buffers at that moment, so we can compute the
 * average look-ahead depth.
 */
static inline void
read_stream_count_prefetch(ReadStream *stream)
{
	IOStats    *stats = stream->stats;

	if (stats == NULL)
		return;

	stats->prefetch_count++;
	stats->distance_sum += stream->pinned_buffers;
	if (stream->pinned_buffers > stats->distance_max)
		stats->distance_max = stream->pinned_buffers;
}

/*
 * Update stream stats about size of I/O requests.
 *
 * We count the number of I/O requests, size of requests (counted in blocks)
 * and number of in-progress I/Os.
 */
static inline void
read_stream_count_io(ReadStream *stream, int nblocks, int in_progress)
{
	IOStats    *stats = stream->stats;

	if (stats == NULL)
		return;

	stats->io_count++;
	stats->io_nblocks += nblocks;
	stats->io_in_progress += in_progress;
}

/*
 * Update stream stats about waits for I/O when consuming buffers.
 *
 * We count the number of I/O waits while pulling buffers out of a stream.
 */
static inline void
read_stream_count_wait(ReadStream *stream)
{
	IOStats    *stats = stream->stats;

	if (stats == NULL)
		return;

	stats->wait_count++;
}

/*
 * Enable collection of stats into the provided IOStats.
 */
void
read_stream_enable_stats(ReadStream *stream, IOStats *stats)
{
	stream->stats = stats;
	if (stream->stats)
		stream->stats->distance_capacity = stream->max_pinned_buffers;
}

/*
 * Ask the callback which block it would like us to read next, with a one block
 * buffer in front to allow read_stream_unget_block() to work.
 */
static inline BlockNumber
read_stream_get_block(ReadStream *stream, void *per_buffer_data)
{
	BlockNumber blocknum;

	blocknum = stream->buffered_blocknum;
	if (blocknum != InvalidBlockNumber)
		stream->buffered_blocknum = InvalidBlockNumber;
	else
	{
		/*
		 * Tell Valgrind that the per-buffer data is undefined.  That replaces
		 * the "noaccess" state that was set when the consumer moved past this
		 * entry last time around the queue, and should also catch callbacks
		 * that fail to initialize data that the buffer consumer later
		 * accesses.  On the first go around, it is undefined already.
		 */
		VALGRIND_MAKE_MEM_UNDEFINED(per_buffer_data,
									stream->per_buffer_data_size);
		blocknum = stream->callback(stream,
									stream->callback_private_data,
									per_buffer_data);
	}

	return blocknum;
}

/*
 * In order to deal with buffer shortages and I/O limits after short reads, we
 * sometimes need to defer handling of a block we've already consumed from the
 * registered callback until later.
 */
static inline void
read_stream_unget_block(ReadStream *stream, BlockNumber blocknum)
{
	/* We shouldn't ever unget more than one block. */
	Assert(stream->buffered_blocknum == InvalidBlockNumber);
	Assert(blocknum != InvalidBlockNumber);
	stream->buffered_blocknum = blocknum;
}

/*
 * 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
 * split it, then the pending read is adjusted to hold the remaining portion.
 *
 * We can always start a read of at least size one if we have no progress yet.
 * Otherwise it's possible that we can't start a read at all because of a lack
 * of buffers, and then false is returned.  Buffer shortages also reduce the
 * distance to a level that prevents look-ahead until buffers are released.
 */
static bool
read_stream_start_pending_read(ReadStream *stream)
{
	bool		need_wait;
	int			requested_nblocks;
	int			nblocks;
	int			flags;
	int			forwarded;
	int16		io_index;
	int16		overflow;
	int16		buffer_index;
	int			buffer_limit;

	/* This should only be called with a pending read. */
	Assert(stream->pending_read_nblocks > 0);
	Assert(stream->pending_read_nblocks <= stream->io_combine_limit);

	/* 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);

#ifdef USE_ASSERT_CHECKING
	/* We had better not be overwriting an existing pinned buffer. */
	if (stream->pinned_buffers > 0)
		Assert(stream->next_buffer_index != stream->oldest_buffer_index);
	else
		Assert(stream->next_buffer_index == stream->oldest_buffer_index);

	/*
	 * Pinned buffers forwarded by a preceding StartReadBuffers() call that
	 * had to split the operation should match the leading blocks of this
	 * following StartReadBuffers() call.
	 */
	Assert(stream->forwarded_buffers <= stream->pending_read_nblocks);
	for (int i = 0; i < stream->forwarded_buffers; ++i)
		Assert(BufferGetBlockNumber(stream->buffers[stream->next_buffer_index + i]) ==
			   stream->pending_read_blocknum + i);

	/*
	 * Check that we've cleared the queue/overflow entries corresponding to
	 * the rest of the blocks covered by this read, unless it's the first go
	 * around and we haven't even initialized them yet.
	 */
	for (int i = stream->forwarded_buffers; i < stream->pending_read_nblocks; ++i)
		Assert(stream->next_buffer_index + i >= stream->initialized_buffers ||
			   stream->buffers[stream->next_buffer_index + i] == InvalidBuffer);
#endif

	/* Do we need to issue read-ahead advice? */
	flags = stream->read_buffers_flags;
	if (stream->advice_enabled)
	{
		if (stream->pending_read_blocknum == stream->seq_blocknum)
		{
			/*
			 * Sequential:  Issue advice until the preadv() calls have caught
			 * up with the first advice issued for this sequential region, and
			 * then stay out of the way of the kernel's own read-ahead.
			 */
			if (stream->seq_until_processed != InvalidBlockNumber)
				flags |= READ_BUFFERS_ISSUE_ADVICE;
		}
		else
		{
			/*
			 * Random jump:  Note the starting location of a new potential
			 * sequential region and start issuing advice.  Skip it this time
			 * if the preadv() follows immediately, eg first block in stream.
			 */
			stream->seq_until_processed = stream->pending_read_blocknum;
			if (stream->pinned_buffers > 0)
				flags |= READ_BUFFERS_ISSUE_ADVICE;
		}
	}

	/*
	 * How many more buffers is this backend allowed?
	 *
	 * Forwarded buffers are already pinned and map to the leading blocks of
	 * the pending read (the remaining portion of an earlier short read that
	 * we're about to continue).  They are not counted in pinned_buffers, but
	 * they are counted as pins already held by this backend according to the
	 * buffer manager, so they must be added to the limit it grants us.
	 */
	if (stream->temporary)
		buffer_limit = Min(GetAdditionalLocalPinLimit(), PG_INT16_MAX);
	else
		buffer_limit = Min(GetAdditionalPinLimit(), PG_INT16_MAX);
	Assert(stream->forwarded_buffers <= stream->pending_read_nblocks);

	buffer_limit += stream->forwarded_buffers;
	buffer_limit = Min(buffer_limit, PG_INT16_MAX);

	if (buffer_limit == 0 && stream->pinned_buffers == 0)
		buffer_limit = 1;		/* guarantee progress */

	/* Does the per-backend limit affect this read? */
	nblocks = stream->pending_read_nblocks;
	if (buffer_limit < nblocks)
	{
		int16		new_distance;

		/* Shrink distance: no more look-ahead until buffers are released. */
		new_distance = stream->pinned_buffers + buffer_limit;
		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)
			return false;

		/* A short read is required to make progress. */
		nblocks = buffer_limit;
	}

	/*
	 * 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
	 * to InvalidBuffer (= not a forwarded buffer) as input on first use only,
	 * and keep the original nblocks number so we can check for forwarded
	 * buffers as output, below.
	 */
	buffer_index = stream->next_buffer_index;
	io_index = stream->next_io_index;
	while (stream->initialized_buffers < buffer_index + nblocks)
		stream->buffers[stream->initialized_buffers++] = InvalidBuffer;
	requested_nblocks = nblocks;
	need_wait = StartReadBuffers(&stream->ios[io_index].op,
								 &stream->buffers[buffer_index],
								 stream->pending_read_blocknum,
								 &nblocks,
								 flags);
	stream->pinned_buffers += nblocks;

	/* Remember whether we need to wait before returning this buffer. */
	if (!need_wait)
	{
		/*
		 * If there currently is no IO in progress, and we have not needed to
		 * issue IO recently, decay the look-ahead distance.  We detect if we
		 * had to issue IO recently by having a decay holdoff that's set to
		 * the max look-ahead distance whenever we need to do IO.  This is
		 * important to ensure we eventually reach a high enough distance to
		 * perform IO asynchronously when starting out with a small look-ahead
		 * distance.
		 */
		if (stream->ios_in_progress == 0)
		{
			if (stream->distance_decay_holdoff > 0)
				stream->distance_decay_holdoff--;
			else
			{
				if (stream->readahead_distance > 1)
					stream->readahead_distance--;

				/*
				 * For now we reduce the IO combine distance after
				 * sufficiently many buffer hits. There is no clear
				 * performance argument for doing so, but at the moment we
				 * need to do so to make the entrance into fast_path work
				 * correctly: We require combine_distance == 1 to enter
				 * fast-path, as without that condition we would wrongly
				 * re-enter fast-path when readahead_distance == 1 and
				 * pinned_buffers == 1, as we would not yet have prepared
				 * another IO in that situation.
				 */
				if (stream->combine_distance > 1)
					stream->combine_distance--;
			}
		}
	}
	else
	{
		/*
		 * Remember to call WaitReadBuffers() before returning head buffer.
		 * Look-ahead distance will be adjusted after waiting.
		 */
		stream->ios[io_index].buffer_index = buffer_index;
		if (++stream->next_io_index == stream->max_ios)
			stream->next_io_index = 0;
		Assert(stream->ios_in_progress < stream->max_ios);
		stream->ios_in_progress++;
		stream->seq_blocknum = stream->pending_read_blocknum + nblocks;

		/* update I/O stats */
		read_stream_count_io(stream, nblocks, stream->ios_in_progress);
	}

	/*
	 * How many pins were acquired but forwarded to the next call?  These need
	 * to be passed to the next StartReadBuffers() call by leaving them
	 * exactly where they are in the queue, or released if the stream ends
	 * early.  We need the number for accounting purposes, since they are not
	 * counted in stream->pinned_buffers but we already hold them.
	 */
	forwarded = 0;
	while (nblocks + forwarded < requested_nblocks &&
		   stream->buffers[buffer_index + nblocks + forwarded] != InvalidBuffer)
		forwarded++;
	stream->forwarded_buffers = forwarded;

	/*
	 * We gave a contiguous range of buffer space to StartReadBuffers(), but
	 * we want it to wrap around at queue_size.  Copy overflowing buffers to
	 * the front of the array where they'll be consumed, but also leave a copy
	 * in the overflow zone which the I/O operation has a pointer to (it needs
	 * a contiguous array).  Both copies will be cleared when the buffers are
	 * handed to the consumer.
	 */
	overflow = (buffer_index + nblocks + forwarded) - stream->queue_size;
	if (overflow > 0)
	{
		Assert(overflow < stream->queue_size);	/* can't overlap */
		memcpy(&stream->buffers[0],
			   &stream->buffers[stream->queue_size],
			   sizeof(stream->buffers[0]) * overflow);
	}

	/* Compute location of start of next read, without using % operator. */
	buffer_index += nblocks;
	if (buffer_index >= stream->queue_size)
		buffer_index -= stream->queue_size;
	Assert(buffer_index >= 0 && buffer_index < stream->queue_size);
	stream->next_buffer_index = buffer_index;

	/* Adjust the pending read to cover the remaining portion, if any. */
	stream->pending_read_blocknum += nblocks;
	stream->pending_read_nblocks -= nblocks;

	return true;
}

/*
 * Should we continue to perform look ahead?  Looking ahead may allow us to
 * make the pending IO larger via IO combining or to issue more read-ahead.
 */
static inline bool
read_stream_should_look_ahead(ReadStream *stream)
{
	/* If the callback has signaled end-of-stream, we're done */
	if (stream->readahead_distance == 0)
		return false;

	/* never start more IOs than our cap */
	if (stream->ios_in_progress >= stream->max_ios)
		return false;

	/*
	 * Allow looking further ahead if we are in the process of building a
	 * larger IO, the IO is not yet big enough, and we don't yet have IO in
	 * flight.
	 *
	 * We do so to allow building larger reads when readahead_distance is
	 * small (e.g. because the I/O subsystem is keeping up or
	 * effective_io_concurrency is small). That's a useful goal because larger
	 * reads are more CPU efficient than smaller reads, even if the system is
	 * not IO bound.
	 *
	 * The reason we do *not* do so when we already have a read prepared (i.e.
	 * why we check for pinned_buffers == 0) is once we are actually reading
	 * ahead, we don't need it:
	 *
	 * - We won't issue unnecessarily small reads as
	 * read_stream_should_issue_now() will return false until the IO is
	 * suitably sized. The issuance of the pending read will be delayed until
	 * enough buffers have been consumed.
	 *
	 * - If we are not reading ahead aggressively enough, future
	 * WaitReadBuffers() calls will return true, leading to readahead_distance
	 * being increased. After that more full-sized IOs can be issued.
	 *
	 * Furthermore, if we did not have the pinned_buffers == 0 condition, we
	 * might end up issuing I/O more aggressively than we need.
	 *
	 * Note that a return of true here can lead to exceeding the read-ahead
	 * limit, but we won't exceed the buffer pin limit (because pinned_buffers
	 * == 0 and combine_distance is capped by max_pinned_buffers).
	 */
	if (stream->pending_read_nblocks > 0 &&
		stream->pinned_buffers == 0 &&
		stream->pending_read_nblocks < stream->combine_distance)
		return true;

	/*
	 * Don't start more read-ahead if that'd put us over the distance limit
	 * for doing read-ahead. As stream->readahead_distance is capped by
	 * max_pinned_buffers, this prevents us from looking ahead so far that it
	 * would put us over the pin limit.
	 */
	if (stream->pinned_buffers + stream->pending_read_nblocks >= stream->readahead_distance)
		return false;

	return true;
}

/*
 * We don't start the pending read just because we've hit the distance limit,
 * preferring to give it another chance to grow to full io_combine_limit size
 * once more buffers have been consumed.  But this is not desirable in all
 * situations - see below.
 */
static inline bool
read_stream_should_issue_now(ReadStream *stream)
{
	int16		pending_read_nblocks = stream->pending_read_nblocks;

	/* there is no pending IO that could be issued */
	if (pending_read_nblocks == 0)
		return false;

	/* never start more IOs than our cap */
	if (stream->ios_in_progress >= stream->max_ios)
		return false;

	/*
	 * If the callback has signaled end-of-stream, start the pending read
	 * immediately. There is no further potential for IO combining.
	 */
	if (stream->readahead_distance == 0)
		return true;

	/*
	 * If we've already reached combine_distance, there's no chance of growing
	 * the read further.
	 */
	if (pending_read_nblocks >= stream->combine_distance)
		return true;

	/*
	 * If we currently have no reads in flight or prepared, issue the IO once
	 * we are not looking ahead further. This ensures there's always at least
	 * one IO prepared.
	 */
	if (stream->pinned_buffers == 0 &&
		!read_stream_should_look_ahead(stream))
		return true;

	return false;
}

static void
read_stream_look_ahead(ReadStream *stream)
{
	/*
	 * Allow amortizing the cost of submitting IO over multiple IOs. This
	 * requires that we don't do any operations that could lead to a deadlock
	 * with staged-but-unsubmitted IO. The callback needs to opt-in to being
	 * careful.
	 */
	if (stream->batch_mode)
		pgaio_enter_batchmode();

	while (read_stream_should_look_ahead(stream))
	{
		BlockNumber blocknum;
		int16		buffer_index;
		void	   *per_buffer_data;

		if (read_stream_should_issue_now(stream))
		{
			read_stream_start_pending_read(stream);
			continue;
		}

		/*
		 * See which block the callback wants next in the stream.  We need to
		 * compute the index of the Nth block of the pending read including
		 * wrap-around, but we don't want to use the expensive % operator.
		 */
		buffer_index = stream->next_buffer_index + stream->pending_read_nblocks;
		if (buffer_index >= stream->queue_size)
			buffer_index -= stream->queue_size;
		Assert(buffer_index >= 0 && buffer_index < stream->queue_size);
		per_buffer_data = get_per_buffer_data(stream, buffer_index);
		blocknum = read_stream_get_block(stream, per_buffer_data);
		if (blocknum == InvalidBlockNumber)
		{
			/* End of stream. */
			stream->readahead_distance = 0;
			stream->combine_distance = 0;
			break;
		}

		/* Can we merge it with the pending read? */
		if (stream->pending_read_nblocks > 0 &&
			stream->pending_read_blocknum + stream->pending_read_nblocks == blocknum)
		{
			stream->pending_read_nblocks++;
			continue;
		}

		/* We have to start the pending read before we can build another. */
		while (stream->pending_read_nblocks > 0)
		{
			if (!read_stream_start_pending_read(stream) ||
				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);
				if (stream->batch_mode)
					pgaio_exit_batchmode();
				return;
			}
		}

		/* This is the start of a new pending read. */
		stream->pending_read_blocknum = blocknum;
		stream->pending_read_nblocks = 1;
	}

	/*
	 * Check if the pending read should be issued now, or if we should give it
	 * another chance to grow to the full size.
	 *
	 * Note that the pending read can exceed the distance goal, if the latter
	 * was reduced after hitting the per-backend buffer limit.
	 */
	if (read_stream_should_issue_now(stream))
		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
	 * stream.  In the worst case we can always make progress one buffer at a
	 * time.
	 */
	Assert(stream->pinned_buffers > 0 || stream->readahead_distance == 0);

	if (stream->batch_mode)
		pgaio_exit_batchmode();
}

/*
 * Create a new read stream object that can be used to perform the equivalent
 * of a series of ReadBuffer() calls for one fork of one relation.
 * Internally, it generates larger vectored reads where possible by looking
 * ahead.  The callback should return block numbers or InvalidBlockNumber to
 * signal end-of-stream, and if per_buffer_data_size is non-zero, it may also
 * write extra data for each block into the space provided to it.  It will
 * also receive callback_private_data for its own purposes.
 */
static ReadStream *
read_stream_begin_impl(int flags,
					   BufferAccessStrategy strategy,
					   Relation rel,
					   SMgrRelation smgr,
					   char persistence,
					   ForkNumber forknum,
					   ReadStreamBlockNumberCB callback,
					   void *callback_private_data,
					   size_t per_buffer_data_size)
{
	ReadStream *stream;
	size_t		size;
	int16		queue_size;
	int16		queue_overflow;
	int			max_ios;
	int			strategy_pin_limit;
	uint32		max_pinned_buffers;
	uint32		max_possible_buffer_limit;
	Oid			tablespace_id;

	/*
	 * Reject attempts to read non-local temporary relations; we would be
	 * likely to get wrong data since we have no visibility into the owning
	 * session's local buffers.
	 */
	if (rel && RELATION_IS_OTHER_TEMP(rel))
		ereport(ERROR,
				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
				 errmsg("cannot access temporary tables of other sessions")));

	/*
	 * Decide how many I/Os we will allow to run at the same time.  This
	 * number also affects how far we look ahead for opportunities to start
	 * more I/Os.
	 */
	tablespace_id = smgr->smgr_rlocator.locator.spcOid;
	if (!OidIsValid(MyDatabaseId) ||
		(rel && IsCatalogRelation(rel)) ||
		IsCatalogRelationOid(smgr->smgr_rlocator.locator.relNumber))
	{
		/*
		 * Avoid circularity while trying to look up tablespace settings or
		 * before spccache.c is ready.
		 */
		max_ios = effective_io_concurrency;
	}
	else if (flags & READ_STREAM_MAINTENANCE)
		max_ios = get_tablespace_maintenance_io_concurrency(tablespace_id);
	else
		max_ios = get_tablespace_io_concurrency(tablespace_id);

	/* 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
	 * even if max_ios is set to 0 (see also further down).  For max_ios > 0,
	 * this also allows an extra full I/O's worth of buffers: after an I/O
	 * finishes we don't want to have to wait for its buffers to be consumed
	 * before starting a new one.
	 *
	 * Be careful not to allow int16 to overflow.  That is possible with the
	 * current GUC range limits, so this is an artificial limit of ~32k
	 * buffers and we'd need to adjust the types to exceed that.  We also have
	 * to allow for the spare entry and the overflow space.
	 */
	max_pinned_buffers = (max_ios + 1) * io_combine_limit;
	max_pinned_buffers = Min(max_pinned_buffers,
							 PG_INT16_MAX - queue_overflow - 1);

	/* Give the strategy a chance to limit the number of buffers we pin. */
	strategy_pin_limit = GetAccessStrategyPinLimit(strategy);
	max_pinned_buffers = Min(strategy_pin_limit, max_pinned_buffers);

	/*
	 * Also limit our queue to the maximum number of pins we could ever be
	 * allowed to acquire according to the buffer manager.  We may not really
	 * be able to use them all due to other pins held by this backend, but
	 * we'll check that later in read_stream_start_pending_read().
	 */
	if (SmgrIsTemp(smgr))
		max_possible_buffer_limit = GetLocalPinLimit();
	else
		max_possible_buffer_limit = GetPinLimit();
	max_pinned_buffers = Min(max_pinned_buffers, max_possible_buffer_limit);

	/*
	 * The limit might be zero on a system configured with too few buffers for
	 * the number of connections.  We need at least one to make progress.
	 */
	max_pinned_buffers = Max(1, max_pinned_buffers);

	/*
	 * We need one extra entry for buffers and per-buffer data, because users
	 * of per-buffer data have access to the object until the next call to
	 * read_stream_next_buffer(), so we need a gap between the head and tail
	 * of the queue so that we don't clobber it.
	 */
	queue_size = max_pinned_buffers + 1;

	/*
	 * Allocate the object, the buffers, the ios and per_buffer_data space in
	 * 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 overflow size.
	 */
	size = offsetof(ReadStream, buffers);
	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 + queue_overflow]);
	if (per_buffer_data_size > 0)
		stream->per_buffer_data = (void *)
			MAXALIGN(&stream->ios[Max(1, max_ios)]);

	stream->sync_mode = io_method == IOMETHOD_SYNC;
	stream->batch_mode = flags & READ_STREAM_USE_BATCHING;

#ifdef USE_PREFETCH

	/*
	 * Read-ahead advice simulating asynchronous I/O with synchronous calls.
	 * Issue advice only if AIO is not used, direct I/O isn't enabled, the
	 * caller hasn't promised sequential access (overriding our detection
	 * heuristics), and max_ios hasn't been set to zero.
	 */
	if (stream->sync_mode &&
		(io_direct_flags & IO_DIRECT_DATA) == 0 &&
		(flags & READ_STREAM_SEQUENTIAL) == 0 &&
		max_ios > 0)
		stream->advice_enabled = true;
#endif

	/*
	 * Setting max_ios to zero disables AIO and advice-based pseudo AIO, but
	 * we still need to allocate space to combine and run one I/O.  Bump it up
	 * to one, and remember to ask for synchronous I/O only.
	 */
	if (max_ios == 0)
	{
		max_ios = 1;
		stream->read_buffers_flags = READ_BUFFERS_SYNCHRONOUSLY;
	}

	/*
	 * 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;
	stream->callback = callback;
	stream->callback_private_data = callback_private_data;
	stream->buffered_blocknum = InvalidBlockNumber;
	stream->seq_blocknum = InvalidBlockNumber;
	stream->seq_until_processed = InvalidBlockNumber;
	stream->temporary = SmgrIsTemp(smgr);
	stream->distance_decay_holdoff = 0;

	/*
	 * Skip the initial ramp-up phase if the caller says we're going to be
	 * reading the whole relation.  This way we start out assuming we'll be
	 * doing full io_combine_limit sized reads.
	 */
	if (flags & READ_STREAM_FULL)
	{
		stream->readahead_distance = Min(max_pinned_buffers, stream->io_combine_limit);
		stream->combine_distance = Min(max_pinned_buffers, stream->io_combine_limit);
	}
	else
	{
		stream->readahead_distance = 1;
		stream->combine_distance = 1;
	}
	stream->resume_readahead_distance = stream->readahead_distance;
	stream->resume_combine_distance = stream->combine_distance;

	/*
	 * Since we always access the same relation, we can initialize parts of
	 * the ReadBuffersOperation objects and leave them that way, to avoid
	 * wasting CPU cycles writing to them for each read.
	 */
	for (int i = 0; i < max_ios; ++i)
	{
		stream->ios[i].op.rel = rel;
		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;
}

/*
 * Create a new read stream for reading a relation.
 * See read_stream_begin_impl() for the detailed explanation.
 */
ReadStream *
read_stream_begin_relation(int flags,
						   BufferAccessStrategy strategy,
						   Relation rel,
						   ForkNumber forknum,
						   ReadStreamBlockNumberCB callback,
						   void *callback_private_data,
						   size_t per_buffer_data_size)
{
	return read_stream_begin_impl(flags,
								  strategy,
								  rel,
								  RelationGetSmgr(rel),
								  rel->rd_rel->relpersistence,
								  forknum,
								  callback,
								  callback_private_data,
								  per_buffer_data_size);
}

/*
 * Create a new read stream for reading a SMgr relation.
 * See read_stream_begin_impl() for the detailed explanation.
 */
ReadStream *
read_stream_begin_smgr_relation(int flags,
								BufferAccessStrategy strategy,
								SMgrRelation smgr,
								char smgr_persistence,
								ForkNumber forknum,
								ReadStreamBlockNumberCB callback,
								void *callback_private_data,
								size_t per_buffer_data_size)
{
	return read_stream_begin_impl(flags,
								  strategy,
								  NULL,
								  smgr,
								  smgr_persistence,
								  forknum,
								  callback,
								  callback_private_data,
								  per_buffer_data_size);
}

/*
 * 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
 * set to a non-zero size, *per_buffer_data receives a pointer to the extra
 * per-buffer data that the callback had a chance to populate, which remains
 * valid until the next call to read_stream_next_buffer().  When the stream
 * runs out of data, InvalidBuffer is returned.  The caller may decide to end
 * the stream early at any time by calling read_stream_end().
 */
Buffer
read_stream_next_buffer(ReadStream *stream, void **per_buffer_data)
{
	Buffer		buffer;
	int16		oldest_buffer_index;

#ifndef READ_STREAM_DISABLE_FAST_PATH

	/*
	 * A fast path for all-cached scans.  This is the same as the usual
	 * algorithm, but it is specialized for no I/O and no per-buffer data, so
	 * we can skip the queue management code, stay in the same buffer slot and
	 * use singular StartReadBuffer().
	 */
	if (likely(stream->fast_path))
	{
		BlockNumber next_blocknum;

		/* Fast path assumptions. */
		Assert(stream->ios_in_progress == 0);
		Assert(stream->forwarded_buffers == 0);
		Assert(stream->pinned_buffers == 1);
		Assert(stream->readahead_distance == 1);
		Assert(stream->combine_distance == 1);
		Assert(stream->pending_read_nblocks == 0);
		Assert(stream->per_buffer_data_size == 0);
		Assert(stream->initialized_buffers > stream->oldest_buffer_index);

		/* We're going to return the buffer we pinned last time. */
		oldest_buffer_index = stream->oldest_buffer_index;
		Assert((oldest_buffer_index + 1) % stream->queue_size ==
			   stream->next_buffer_index);
		buffer = stream->buffers[oldest_buffer_index];
		Assert(buffer != InvalidBuffer);

		/* Choose the next block to pin. */
		next_blocknum = read_stream_get_block(stream, NULL);

		if (likely(next_blocknum != InvalidBlockNumber))
		{
			int			flags = stream->read_buffers_flags;

			if (stream->advice_enabled)
				flags |= READ_BUFFERS_ISSUE_ADVICE;

			/*
			 * While in fast-path, execute any IO that we might encounter
			 * synchronously. Because we are, right now, only looking one
			 * block ahead, dispatching any occasional IO to workers would
			 * have the overhead of dispatching to workers, without any
			 * realistic chance of the IO completing before we need it. We
			 * will switch to non-synchronous IO after this.
			 *
			 * Arguably we should do so only for worker, as there's far less
			 * dispatch overhead with io_uring. However, tests so far have not
			 * shown a clear downside and additional io_method awareness here
			 * seems not great from an abstraction POV.
			 */
			flags |= READ_BUFFERS_SYNCHRONOUSLY;

			/*
			 * Pin a buffer for the next call.  Same buffer entry, and
			 * arbitrary I/O entry (they're all free).  We don't have to
			 * adjust pinned_buffers because we're transferring one to caller
			 * but pinning one more.
			 *
			 * In the fast path we don't need to check the pin limit.  We're
			 * always allowed at least one pin so that progress can be made,
			 * and that's all we need here.  Although two pins are momentarily
			 * held at the same time, the model used here is that the stream
			 * holds only one, and the other now belongs to the caller.
			 */
			if (likely(!StartReadBuffer(&stream->ios[0].op,
										&stream->buffers[oldest_buffer_index],
										next_blocknum,
										flags)))
			{
				/* Fast return. */
				read_stream_count_prefetch(stream);
				return buffer;
			}

			/* Next call must wait for I/O for the newly pinned buffer. */
			stream->oldest_io_index = 0;
			stream->next_io_index = stream->max_ios > 1 ? 1 : 0;
			stream->ios_in_progress = 1;
			stream->ios[0].buffer_index = oldest_buffer_index;
			stream->seq_blocknum = next_blocknum + 1;

			/*
			 * XXX: It might be worth triggering additional read-ahead here,
			 * to avoid having to effectively do another synchronous IO for
			 * the next block (if it were also a miss).
			 */

			/* update I/O stats */
			read_stream_count_io(stream, 1, stream->ios_in_progress);

			/* update prefetch distance */
			read_stream_count_prefetch(stream);
		}
		else
		{
			/* No more blocks, end of stream. */
			stream->readahead_distance = 0;
			stream->combine_distance = 0;
			stream->oldest_buffer_index = stream->next_buffer_index;
			stream->pinned_buffers = 0;
			stream->buffers[oldest_buffer_index] = InvalidBuffer;
		}

		stream->fast_path = false;
		return buffer;
	}
#endif

	if (unlikely(stream->pinned_buffers == 0))
	{
		Assert(stream->oldest_buffer_index == stream->next_buffer_index);

		/* End of stream reached?  */
		if (stream->readahead_distance == 0)
			return InvalidBuffer;

		/*
		 * The usual order of operations is that we look ahead at the bottom
		 * of this function after potentially finishing an I/O and making
		 * space for more, but if we're just starting up we'll need to crank
		 * the handle to get started.
		 */
		read_stream_look_ahead(stream);

		/* End of stream reached? */
		if (stream->pinned_buffers == 0)
		{
			Assert(stream->readahead_distance == 0);
			return InvalidBuffer;
		}
	}

	/* Grab the oldest pinned buffer and associated per-buffer data. */
	Assert(stream->pinned_buffers > 0);
	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)
	{
		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]);

		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;
	}

	/*
	 * We must zap this queue entry, or else it would appear as a forwarded
	 * buffer.  If it's potentially in the overflow zone (ie from a
	 * multi-block I/O that wrapped around the queue), also zap the copy.
	 */
	stream->buffers[oldest_buffer_index] = InvalidBuffer;
	if (oldest_buffer_index < stream->io_combine_limit - 1)
		stream->buffers[stream->queue_size + oldest_buffer_index] =
			InvalidBuffer;

#if defined(CLOBBER_FREED_MEMORY) || defined(USE_VALGRIND)

	/*
	 * The caller will get access to the per-buffer data, until the next call.
	 * We wipe the one before, which is never occupied because queue_size
	 * allowed one extra element.  This will hopefully trip up client code
	 * that is holding a dangling pointer to it.
	 */
	if (stream->per_buffer_data)
	{
		void	   *per_buffer_data;

		per_buffer_data = get_per_buffer_data(stream,
											  oldest_buffer_index == 0 ?
											  stream->queue_size - 1 :
											  oldest_buffer_index - 1);

#if defined(CLOBBER_FREED_MEMORY)
		/* This also tells Valgrind the memory is "noaccess". */
		wipe_mem(per_buffer_data, stream->per_buffer_data_size);
#elif defined(USE_VALGRIND)
		/* Tell it ourselves. */
		VALGRIND_MAKE_MEM_NOACCESS(per_buffer_data,
								   stream->per_buffer_data_size);
#endif
	}
#endif

	read_stream_count_prefetch(stream);

	/* Pin transferred to caller. */
	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);

#ifndef READ_STREAM_DISABLE_FAST_PATH
	/* See if we can take the fast path for all-cached scans next time. */
	if (stream->ios_in_progress == 0 &&
		stream->forwarded_buffers == 0 &&
		stream->pinned_buffers == 1 &&
		stream->readahead_distance == 1 &&
		stream->combine_distance == 1 &&
		stream->pending_read_nblocks == 0 &&
		stream->per_buffer_data_size == 0)
	{
		/*
		 * The fast path spins on one buffer entry repeatedly instead of
		 * rotating through the whole queue and clearing the entries behind
		 * it.  If the buffer it starts with happened to be forwarded between
		 * StartReadBuffers() calls and also wrapped around the circular queue
		 * partway through, then a copy also exists in the overflow zone, and
		 * it won't clear it out as the regular path would.  Do that now, so
		 * it doesn't need code for that.
		 */
		if (stream->oldest_buffer_index < stream->io_combine_limit - 1)
			stream->buffers[stream->queue_size + stream->oldest_buffer_index] =
				InvalidBuffer;

		stream->fast_path = true;
	}
#endif

	return buffer;
}

/*
 * Transitional support for code that would like to perform or skip reads
 * itself, without using the stream.  Returns, and consumes, the next block
 * number that would be read by the stream's look-ahead algorithm, or
 * InvalidBlockNumber if the end of the stream is reached.  Also reports the
 * strategy that would be used to read it.
 */
BlockNumber
read_stream_next_block(ReadStream *stream, BufferAccessStrategy *strategy)
{
	*strategy = stream->ios[0].op.strategy;
	return read_stream_get_block(stream, NULL);
}

/*
 * Temporarily stop consuming block numbers from the block number callback.
 * If called inside the block number callback, its return value should be
 * returned by the callback.
 */
BlockNumber
read_stream_pause(ReadStream *stream)
{
	stream->resume_readahead_distance = stream->readahead_distance;
	stream->resume_combine_distance = stream->combine_distance;
	stream->readahead_distance = 0;
	stream->combine_distance = 0;
	return InvalidBlockNumber;
}

/*
 * Resume looking ahead after the block number callback reported
 * end-of-stream. This is useful for streams of self-referential blocks, after
 * a buffer needed to be consumed and examined to find more block numbers.
 */
void
read_stream_resume(ReadStream *stream)
{
	stream->readahead_distance = stream->resume_readahead_distance;
	stream->combine_distance = stream->resume_combine_distance;
}

/*
 * Stop using a buffer access strategy for reads from this stream.
 *
 * This clears the strategy for all of the stream's ReadBuffersOperations,
 * including those with in-progress IOs. The completion of an IO whose
 * strategy was cleared while it was in flight may have a small amount of its
 * read time attributed to IOCONTEXT_NORMAL instead of the strategy's
 * IOContext, because WaitReadBuffers() derives the IOContext from the (now
 * cleared) strategy. This is bounded by the stream's look-ahead window and
 * happens at most once, when the strategy is first cleared, so it is not worth
 * the complexity of preserving the original IOContext for those IOs.
 *
 * Note that the caller is responsible for freeing the strategy's memory.
 */
void
read_stream_clear_strategy(ReadStream *stream)
{
	for (int i = 0; i < stream->max_ios; ++i)
		stream->ios[i].op.strategy = NULL;
}

/*
 * Reset a read stream by releasing any queued up buffers, allowing the stream
 * to be used again for different blocks.  This can be used to clear an
 * end-of-stream condition and start again, or to throw away blocks that were
 * speculatively read and read some different blocks instead.
 */
void
read_stream_reset(ReadStream *stream)
{
	int16		index;
	Buffer		buffer;

	/* Stop looking ahead. */
	stream->readahead_distance = 0;
	stream->combine_distance = 0;

	/* Forget buffered block number and fast path state. */
	stream->buffered_blocknum = InvalidBlockNumber;
	stream->fast_path = false;

	/* Unpin anything that wasn't consumed. */
	while ((buffer = read_stream_next_buffer(stream, NULL)) != InvalidBuffer)
		ReleaseBuffer(buffer);

	/* Unpin any unused forwarded buffers. */
	index = stream->next_buffer_index;
	while (index < stream->initialized_buffers &&
		   (buffer = stream->buffers[index]) != InvalidBuffer)
	{
		Assert(stream->forwarded_buffers > 0);
		stream->forwarded_buffers--;
		ReleaseBuffer(buffer);

		stream->buffers[index] = InvalidBuffer;
		if (index < stream->io_combine_limit - 1)
			stream->buffers[stream->queue_size + index] = InvalidBuffer;

		if (++index == stream->queue_size)
			index = 0;
	}

	Assert(stream->forwarded_buffers == 0);
	Assert(stream->pinned_buffers == 0);
	Assert(stream->ios_in_progress == 0);

	/* Start off assuming data is cached. */
	stream->readahead_distance = 1;
	stream->combine_distance = 1;
	stream->resume_readahead_distance = stream->readahead_distance;
	stream->resume_combine_distance = stream->combine_distance;
	stream->distance_decay_holdoff = 0;
}

/*
 * Release and free a read stream.
 */
void
read_stream_end(ReadStream *stream)
{
	read_stream_reset(stream);
	pfree(stream);
}