read_stream.c File Reference

#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"

Include dependency graph for read_stream.c:

Go to the source code of this file.

Data Structures
struct	InProgressIO
struct	ReadStream

Typedefs
typedef struct InProgressIO	InProgressIO

Functions
static void *	get_per_buffer_data (ReadStream *stream, int16 buffer_index)
BlockNumber	block_range_read_stream_cb (ReadStream *stream, void *callback_private_data, void *per_buffer_data)
static void	read_stream_count_prefetch (ReadStream *stream)
static void	read_stream_count_io (ReadStream *stream, int nblocks, int in_progress)
static void	read_stream_count_wait (ReadStream *stream)
void	read_stream_enable_stats (ReadStream *stream, IOStats *stats)
static BlockNumber	read_stream_get_block (ReadStream *stream, void *per_buffer_data)
static void	read_stream_unget_block (ReadStream *stream, BlockNumber blocknum)
static bool	read_stream_start_pending_read (ReadStream *stream)
static bool	read_stream_should_look_ahead (ReadStream *stream)
static bool	read_stream_should_issue_now (ReadStream *stream)
static void	read_stream_look_ahead (ReadStream *stream)
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 *	read_stream_begin_relation (int flags, BufferAccessStrategy strategy, Relation rel, ForkNumber forknum, ReadStreamBlockNumberCB callback, void *callback_private_data, size_t per_buffer_data_size)
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)
Buffer	read_stream_next_buffer (ReadStream *stream, void **per_buffer_data)
BlockNumber	read_stream_next_block (ReadStream *stream, BufferAccessStrategy *strategy)
BlockNumber	read_stream_pause (ReadStream *stream)
void	read_stream_resume (ReadStream *stream)
void	read_stream_reset (ReadStream *stream)
void	read_stream_end (ReadStream *stream)

Typedef Documentation

InProgressIO

typedef struct InProgressIO InProgressIO

Function Documentation

block_range_read_stream_cb()

BlockNumber block_range_read_stream_cb	(	ReadStream *	stream,
		void *	callback_private_data,
		void *	per_buffer_data
	)

Definition at line 183 of file read_stream.c.

{
    BlockRangeReadStreamPrivate *p = callback_private_data;
 
    if (p->current_blocknum < p->last_exclusive)
        return p->current_blocknum++;
 
    return InvalidBlockNumber;
}

References BlockRangeReadStreamPrivate::current_blocknum, InvalidBlockNumber, and BlockRangeReadStreamPrivate::last_exclusive.

Referenced by blbulkdelete(), blgetbitmap(), blvacuumcleanup(), brin_vacuum_scan(), btvacuumscan(), collect_visibility_data(), ginvacuumcleanup(), gistvacuumscan(), pg_prewarm(), pgstathashindex(), pgstatindex_impl(), RelationCopyStorageUsingBuffer(), spgvacuumscan(), and verify_heapam().

get_per_buffer_data()

static void * get_per_buffer_data ( ReadStream *  stream, int16  buffer_index  )

inlinestatic

Definition at line 172 of file read_stream.c.

{
    return (char *) stream->per_buffer_data +
        stream->per_buffer_data_size * buffer_index;
}

References ReadStream::per_buffer_data, and ReadStream::per_buffer_data_size.

Referenced by read_stream_look_ahead(), and read_stream_next_buffer().

read_stream_begin_impl()

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  )

static

Definition at line 759 of file read_stream.c.

{
    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;
 
    /*
     * 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;
}

References ReadStream::advice_enabled, ReadStream::batch_mode, ReadStream::buffered_blocknum, ReadStream::buffers, ReadStream::callback, callback(), ReadStream::callback_private_data, ReadStream::combine_distance, ReadStream::distance_decay_holdoff, effective_io_concurrency, fb(), ReadBuffersOperation::forknum, get_tablespace_io_concurrency(), get_tablespace_maintenance_io_concurrency(), GetAccessStrategyPinLimit(), GetLocalPinLimit(), GetPinLimit(), i, InvalidBlockNumber, ReadStream::io_combine_limit, io_combine_limit, IO_DIRECT_DATA, io_direct_flags, io_method, IOMETHOD_SYNC, ReadStream::ios, IsCatalogRelation(), IsCatalogRelationOid(), RelFileLocatorBackend::locator, Max, ReadStream::max_ios, ReadStream::max_pinned_buffers, MAXALIGN, Min, MyDatabaseId, OidIsValid, InProgressIO::op, palloc(), ReadStream::per_buffer_data, ReadStream::per_buffer_data_size, ReadBuffersOperation::persistence, PG_INT16_MAX, ReadStream::queue_size, ReadStream::read_buffers_flags, READ_BUFFERS_SYNCHRONOUSLY, READ_STREAM_FULL, READ_STREAM_MAINTENANCE, READ_STREAM_SEQUENTIAL, READ_STREAM_USE_BATCHING, ReadStream::readahead_distance, ReadBuffersOperation::rel, RelFileLocator::relNumber, ReadStream::resume_combine_distance, ReadStream::resume_readahead_distance, ReadStream::seq_blocknum, ReadStream::seq_until_processed, ReadBuffersOperation::smgr, SMgrRelationData::smgr_rlocator, SmgrIsTemp, RelFileLocator::spcOid, ReadBuffersOperation::strategy, ReadStream::sync_mode, and ReadStream::temporary.

Referenced by read_stream_begin_relation(), and read_stream_begin_smgr_relation().

read_stream_begin_relation()

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
	)

Definition at line 966 of file read_stream.c.

{
    return read_stream_begin_impl(flags,
                                  strategy,
                                  rel,
                                  RelationGetSmgr(rel),
                                  rel->rd_rel->relpersistence,
                                  forknum,
                                  callback,
                                  callback_private_data,
                                  per_buffer_data_size);
}

References callback(), RelationData::rd_rel, read_stream_begin_impl(), and RelationGetSmgr().

Referenced by acquire_sample_rows(), autoprewarm_database_main(), blbulkdelete(), blgetbitmap(), blvacuumcleanup(), brin_vacuum_scan(), btvacuumscan(), collect_corrupt_items(), collect_visibility_data(), ginvacuumcleanup(), gistvacuumscan(), hashbulkdelete(), heap_beginscan(), lazy_scan_heap(), lazy_vacuum_heap_rel(), pg_prewarm(), pgstathashindex(), pgstatindex_impl(), read_stream_for_blocks(), spgvacuumscan(), statapprox_heap(), and verify_heapam().

read_stream_begin_smgr_relation()

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
	)

Definition at line 990 of file read_stream.c.

{
    return read_stream_begin_impl(flags,
                                  strategy,
                                  NULL,
                                  smgr,
                                  smgr_persistence,
                                  forknum,
                                  callback,
                                  callback_private_data,
                                  per_buffer_data_size);
}

References callback(), fb(), and read_stream_begin_impl().

Referenced by RelationCopyStorageUsingBuffer().

read_stream_count_io()

static void read_stream_count_io ( ReadStream *  stream, int  nblocks, int  in_progress  )

inlinestatic

Definition at line 223 of file read_stream.c.

{
    IOStats    *stats = stream->stats;
 
    if (stats == NULL)
        return;
 
    stats->io_count++;
    stats->io_nblocks += nblocks;
    stats->io_in_progress += in_progress;
}

References fb(), IOStats::io_count, IOStats::io_in_progress, IOStats::io_nblocks, and ReadStream::stats.

Referenced by read_stream_next_buffer(), and read_stream_start_pending_read().

read_stream_count_prefetch()

static void read_stream_count_prefetch ( ReadStream *  stream )

inlinestatic

Definition at line 203 of file read_stream.c.

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

References IOStats::distance_max, IOStats::distance_sum, fb(), ReadStream::pinned_buffers, IOStats::prefetch_count, and ReadStream::stats.

Referenced by read_stream_next_buffer().

read_stream_count_wait()

static void read_stream_count_wait ( ReadStream *  stream )

inlinestatic

Definition at line 241 of file read_stream.c.

{
    IOStats    *stats = stream->stats;
 
    if (stats == NULL)
        return;
 
    stats->wait_count++;
}

References fb(), ReadStream::stats, and IOStats::wait_count.

Referenced by read_stream_next_buffer().

read_stream_enable_stats()

void read_stream_enable_stats	(	ReadStream *	stream,
		IOStats *	stats
	)

Definition at line 255 of file read_stream.c.

{
    stream->stats = stats;
    if (stream->stats)
        stream->stats->distance_capacity = stream->max_pinned_buffers;
}

References IOStats::distance_capacity, ReadStream::max_pinned_buffers, and ReadStream::stats.

Referenced by heap_beginscan().

read_stream_end()

void read_stream_end

(

ReadStream *

stream

)

Definition at line 1457 of file read_stream.c.

{
    read_stream_reset(stream);
    pfree(stream);
}

References pfree(), and read_stream_reset().

Referenced by acquire_sample_rows(), autoprewarm_database_main(), blbulkdelete(), blgetbitmap(), blvacuumcleanup(), brin_vacuum_scan(), btvacuumscan(), collect_corrupt_items(), collect_visibility_data(), ginvacuumcleanup(), gistvacuumscan(), hashbulkdelete(), heap_endscan(), lazy_scan_heap(), lazy_vacuum_heap_rel(), pg_prewarm(), pgstathashindex(), pgstatindex_impl(), read_stream_for_blocks(), RelationCopyStorageUsingBuffer(), spgvacuumscan(), statapprox_heap(), and verify_heapam().

read_stream_get_block()

static BlockNumber read_stream_get_block ( ReadStream *  stream, void *  per_buffer_data  )

inlinestatic

Definition at line 267 of file read_stream.c.

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

References ReadStream::buffered_blocknum, ReadStream::callback, ReadStream::callback_private_data, InvalidBlockNumber, ReadStream::per_buffer_data_size, and VALGRIND_MAKE_MEM_UNDEFINED.

Referenced by read_stream_look_ahead(), read_stream_next_block(), and read_stream_next_buffer().

read_stream_look_ahead()

static void read_stream_look_ahead ( ReadStream *  stream )

static

Definition at line 658 of file read_stream.c.

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

References Assert, ReadStream::batch_mode, ReadStream::combine_distance, fb(), get_per_buffer_data(), InvalidBlockNumber, ReadStream::ios_in_progress, ReadStream::max_ios, ReadStream::next_buffer_index, ReadStream::pending_read_blocknum, ReadStream::pending_read_nblocks, pgaio_enter_batchmode(), pgaio_exit_batchmode(), ReadStream::pinned_buffers, ReadStream::queue_size, read_stream_get_block(), read_stream_should_issue_now(), read_stream_should_look_ahead(), read_stream_start_pending_read(), read_stream_unget_block(), and ReadStream::readahead_distance.

Referenced by read_stream_next_buffer().

read_stream_next_block()

BlockNumber read_stream_next_block	(	ReadStream *	stream,
		BufferAccessStrategy *	strategy
	)

Definition at line 1367 of file read_stream.c.

{
    *strategy = stream->ios[0].op.strategy;
    return read_stream_get_block(stream, NULL);
}

References fb(), ReadStream::ios, InProgressIO::op, read_stream_get_block(), and ReadBuffersOperation::strategy.

read_stream_next_buffer()

Buffer read_stream_next_buffer	(	ReadStream *	stream,
		void **	per_buffer_data
	)

Definition at line 1020 of file read_stream.c.

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

References ReadStream::advice_enabled, Assert, ReadBuffersOperation::blocknum, InProgressIO::buffer_index, BufferIsValid(), ReadStream::buffers, ReadBuffersOperation::buffers, ReadStream::combine_distance, ReadStream::distance_decay_holdoff, ReadStream::fast_path, fb(), ReadBuffersOperation::flags, ReadStream::forwarded_buffers, get_per_buffer_data(), ReadStream::initialized_buffers, InvalidBlockNumber, InvalidBuffer, ReadStream::io_combine_limit, io_combine_limit, ReadStream::ios, ReadStream::ios_in_progress, likely, ReadStream::max_ios, ReadStream::max_pinned_buffers, Min, ReadStream::next_buffer_index, ReadStream::next_io_index, ReadStream::oldest_buffer_index, ReadStream::oldest_io_index, InProgressIO::op, ReadStream::pending_read_nblocks, ReadStream::per_buffer_data, ReadStream::per_buffer_data_size, ReadStream::pinned_buffers, ReadStream::queue_size, ReadStream::read_buffers_flags, READ_BUFFERS_ISSUE_ADVICE, READ_BUFFERS_SYNCHRONOUSLY, read_stream_count_io(), read_stream_count_prefetch(), read_stream_count_wait(), read_stream_get_block(), read_stream_look_ahead(), ReadStream::readahead_distance, ReadStream::seq_blocknum, ReadStream::seq_until_processed, StartReadBuffer(), unlikely, VALGRIND_MAKE_MEM_NOACCESS, and WaitReadBuffers().

Referenced by autoprewarm_database_main(), BitmapHeapScanNextBlock(), blbulkdelete(), blgetbitmap(), blvacuumcleanup(), brin_vacuum_scan(), btvacuumscan(), collect_corrupt_items(), collect_visibility_data(), ginvacuumcleanup(), gistvacuumscan(), hashbulkdelete(), heap_fetch_next_buffer(), heapam_scan_analyze_next_block(), lazy_scan_heap(), lazy_vacuum_heap_rel(), pg_prewarm(), pgstathashindex(), pgstatindex_impl(), read_stream_for_blocks(), read_stream_reset(), RelationCopyStorageUsingBuffer(), spgvacuumscan(), statapprox_heap(), and verify_heapam().

read_stream_pause()

BlockNumber read_stream_pause

(

ReadStream *

stream

)

Definition at line 1379 of file read_stream.c.

{
    stream->resume_readahead_distance = stream->readahead_distance;
    stream->resume_combine_distance = stream->combine_distance;
    stream->readahead_distance = 0;
    stream->combine_distance = 0;
    return InvalidBlockNumber;
}

References ReadStream::combine_distance, InvalidBlockNumber, ReadStream::readahead_distance, ReadStream::resume_combine_distance, and ReadStream::resume_readahead_distance.

read_stream_reset()

void read_stream_reset

(

ReadStream *

stream

)

Definition at line 1407 of file read_stream.c.

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

References Assert, ReadStream::buffered_blocknum, ReadStream::buffers, ReadStream::combine_distance, ReadStream::distance_decay_holdoff, ReadStream::fast_path, fb(), ReadStream::forwarded_buffers, InvalidBlockNumber, InvalidBuffer, io_combine_limit, ReadStream::ios_in_progress, ReadStream::next_buffer_index, ReadStream::pinned_buffers, ReadStream::queue_size, read_stream_next_buffer(), ReadStream::readahead_distance, ReleaseBuffer(), ReadStream::resume_combine_distance, and ReadStream::resume_readahead_distance.

Referenced by btvacuumscan(), gistvacuumscan(), hashbulkdelete(), heap_fetch_next_buffer(), heap_rescan(), read_stream_end(), and spgvacuumscan().

read_stream_resume()

void read_stream_resume

(

ReadStream *

stream

)

Definition at line 1394 of file read_stream.c.

{
    stream->readahead_distance = stream->resume_readahead_distance;
    stream->combine_distance = stream->resume_combine_distance;
}

References ReadStream::combine_distance, ReadStream::readahead_distance, ReadStream::resume_combine_distance, and ReadStream::resume_readahead_distance.

read_stream_should_issue_now()

static bool read_stream_should_issue_now ( ReadStream *  stream )

inlinestatic

Definition at line 619 of file read_stream.c.

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

References ReadStream::combine_distance, ReadStream::ios_in_progress, ReadStream::max_ios, ReadStream::pending_read_nblocks, ReadStream::pinned_buffers, read_stream_should_look_ahead(), and ReadStream::readahead_distance.

Referenced by read_stream_look_ahead().

read_stream_should_look_ahead()

static bool read_stream_should_look_ahead ( ReadStream *  stream )

inlinestatic

Definition at line 554 of file read_stream.c.

{
    /* 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;
}

References ReadStream::combine_distance, ReadStream::ios_in_progress, ReadStream::max_ios, ReadStream::pending_read_nblocks, ReadStream::pinned_buffers, and ReadStream::readahead_distance.

Referenced by read_stream_look_ahead(), and read_stream_should_issue_now().

read_stream_start_pending_read()

static bool read_stream_start_pending_read ( ReadStream *  stream )

static

Definition at line 318 of file read_stream.c.

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

References ReadStream::advice_enabled, Assert, InProgressIO::buffer_index, BufferGetBlockNumber(), ReadStream::buffers, ReadStream::combine_distance, ReadStream::distance_decay_holdoff, fb(), ReadStream::forwarded_buffers, GetAdditionalLocalPinLimit(), GetAdditionalPinLimit(), i, ReadStream::initialized_buffers, InvalidBlockNumber, InvalidBuffer, ReadStream::io_combine_limit, ReadStream::ios, ReadStream::ios_in_progress, ReadStream::max_ios, ReadStream::max_pinned_buffers, memcpy(), Min, ReadStream::next_buffer_index, ReadStream::next_io_index, ReadStream::oldest_buffer_index, InProgressIO::op, ReadStream::pending_read_blocknum, ReadStream::pending_read_nblocks, PG_INT16_MAX, ReadStream::pinned_buffers, ReadStream::queue_size, ReadStream::read_buffers_flags, READ_BUFFERS_ISSUE_ADVICE, read_stream_count_io(), ReadStream::readahead_distance, ReadStream::seq_blocknum, ReadStream::seq_until_processed, StartReadBuffers(), and ReadStream::temporary.

Referenced by read_stream_look_ahead().

read_stream_unget_block()

static void read_stream_unget_block ( ReadStream *  stream, BlockNumber  blocknum  )

inlinestatic

Definition at line 299 of file read_stream.c.

{
    /* We shouldn't ever unget more than one block. */
    Assert(stream->buffered_blocknum == InvalidBlockNumber);
    Assert(blocknum != InvalidBlockNumber);
    stream->buffered_blocknum = blocknum;
}

References Assert, ReadStream::buffered_blocknum, and InvalidBlockNumber.

Referenced by read_stream_look_ahead().