read_stream.c File Reference

#include "postgres.h"
#include "catalog/pg_tablespace.h"
#include "miscadmin.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)
static BlockNumber	read_stream_get_block (ReadStream *stream, void *per_buffer_data)
static void	read_stream_unget_block (ReadStream *stream, BlockNumber blocknum)
static void	read_stream_fill_blocknums (ReadStream *stream)
static void	read_stream_start_pending_read (ReadStream *stream, bool suppress_advice)
static void	read_stream_look_ahead (ReadStream *stream, bool suppress_advice)
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)
Buffer	read_stream_next_buffer (ReadStream *stream, void **per_buffer_data)
void	read_stream_reset (ReadStream *stream)
void	read_stream_end (ReadStream *stream)

Typedef Documentation

InProgressIO

typedef struct InProgressIO InProgressIO

Function Documentation

get_per_buffer_data()

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

inlinestatic

Definition at line 163 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_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 410 of file read_stream.c.

{
    ReadStream *stream;
    size_t      size;
    int16       queue_size;
    int16       max_ios;
    int         strategy_pin_limit;
    uint32      max_pinned_buffers;
    Oid         tablespace_id;
    SMgrRelation smgr;
 
    smgr = RelationGetSmgr(rel);
 
    /*
     * Decide how many I/Os we will allow to run at the same time.  That
     * currently means advice to the kernel to tell it that we will soon read.
     * 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) ||
        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);
    max_ios = Min(max_ios, PG_INT16_MAX);
 
    /*
     * Choose the maximum number of buffers we're prepared to pin.  We try to
     * pin fewer if we can, though.  We clamp it to at least io_combine_limit
     * so that we can have a chance to build up a full io_combine_limit sized
     * read, even when max_ios is zero.  Be careful not to allow int16 to
     * overflow (even though that's not possible with the current GUC range
     * limits), allowing also for the spare entry and the overflow space.
     */
    max_pinned_buffers = Max(max_ios * 4, io_combine_limit);
    max_pinned_buffers = Min(max_pinned_buffers,
                             PG_INT16_MAX - io_combine_limit - 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);
 
    /* Don't allow this backend to pin more than its share of buffers. */
    if (SmgrIsTemp(smgr))
        LimitAdditionalLocalPins(&max_pinned_buffers);
    else
        LimitAdditionalPins(&max_pinned_buffers);
    Assert(max_pinned_buffers > 0);
 
    /*
     * 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_data_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 read size by allocating an extra
     * io_combine_limit - 1 elements.
     */
    size = offsetof(ReadStream, buffers);
    size += sizeof(Buffer) * (queue_size + io_combine_limit - 1);
    size += sizeof(InProgressIO) * Max(1, max_ios);
    size += per_buffer_data_size * queue_size;
    size += MAXIMUM_ALIGNOF * 2;
    stream = (ReadStream *) palloc(size);
    memset(stream, 0, offsetof(ReadStream, buffers));
    stream->ios = (InProgressIO *)
        MAXALIGN(&stream->buffers[queue_size + io_combine_limit - 1]);
    if (per_buffer_data_size > 0)
        stream->per_buffer_data = (void *)
            MAXALIGN(&stream->ios[Max(1, max_ios)]);
 
#ifdef USE_PREFETCH
 
    /*
     * This system supports prefetching advice.  We can use it as long as
     * 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 ((io_direct_flags & IO_DIRECT_DATA) == 0 &&
        (flags & READ_STREAM_SEQUENTIAL) == 0 &&
        max_ios > 0)
        stream->advice_enabled = true;
#endif
 
    /*
     * For now, max_ios = 0 is interpreted as max_ios = 1 with advice disabled
     * above.  If we had real asynchronous I/O we might need a slightly
     * different definition.
     */
    if (max_ios == 0)
        max_ios = 1;
 
    stream->max_ios = max_ios;
    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;
 
    /*
     * 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 (behavior B).
     */
    if (flags & READ_STREAM_FULL)
        stream->distance = Min(max_pinned_buffers, io_combine_limit);
    else
        stream->distance = 1;
 
    /*
     * 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 = RelationGetSmgr(rel);
        stream->ios[i].op.smgr_persistence = 0;
        stream->ios[i].op.forknum = forknum;
        stream->ios[i].op.strategy = strategy;
    }
 
    return stream;
}

References ReadStream::advice_enabled, Assert, ReadStream::buffers, ReadStream::callback, callback(), ReadStream::callback_private_data, ReadStream::distance, effective_io_concurrency, ReadBuffersOperation::forknum, get_tablespace_io_concurrency(), get_tablespace_maintenance_io_concurrency(), GetAccessStrategyPinLimit(), i, io_combine_limit, IO_DIRECT_DATA, io_direct_flags, ReadStream::ios, IsCatalogRelation(), IsCatalogRelationOid(), LimitAdditionalLocalPins(), LimitAdditionalPins(), 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, PG_INT16_MAX, ReadStream::queue_size, READ_STREAM_FULL, READ_STREAM_MAINTENANCE, READ_STREAM_SEQUENTIAL, ReadBuffersOperation::rel, RelationGetSmgr(), RelFileLocator::relNumber, size, ReadBuffersOperation::smgr, ReadBuffersOperation::smgr_persistence, SMgrRelationData::smgr_rlocator, SmgrIsTemp, RelFileLocator::spcOid, and ReadBuffersOperation::strategy.

Referenced by acquire_sample_rows(), heap_beginscan(), and pg_prewarm().

read_stream_end()

void read_stream_end

(

ReadStream *

stream

)

Definition at line 800 of file read_stream.c.

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

References pfree(), and read_stream_reset().

Referenced by acquire_sample_rows(), heap_endscan(), and pg_prewarm().

read_stream_fill_blocknums()

static void read_stream_fill_blocknums ( ReadStream *  stream )

static

Definition at line 214 of file read_stream.c.

{
    BlockNumber blocknum;
    int         i = 0;
 
    do
    {
        blocknum = stream->callback(stream,
                                    stream->callback_private_data,
                                    NULL);
        stream->blocknums[i++] = blocknum;
    } while (i < lengthof(stream->blocknums) &&
             blocknum != InvalidBlockNumber);
    stream->blocknums_count = i;
    stream->blocknums_next = 0;
}

References ReadStream::blocknums, ReadStream::blocknums_count, ReadStream::blocknums_next, ReadStream::callback, ReadStream::callback_private_data, i, InvalidBlockNumber, and lengthof.

Referenced by read_stream_next_buffer().

read_stream_get_block()

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

inlinestatic

Definition at line 175 of file read_stream.c.

{
    if (stream->blocknums_next < stream->blocknums_count)
        return stream->blocknums[stream->blocknums_next++];
 
    /*
     * We only bother to fetch one at a time here (but see the fast path which
     * uses more).
     */
    return stream->callback(stream,
                            stream->callback_private_data,
                            per_buffer_data);
}

References ReadStream::blocknums, ReadStream::blocknums_count, ReadStream::blocknums_next, ReadStream::callback, and ReadStream::callback_private_data.

Referenced by read_stream_look_ahead().

read_stream_look_ahead()

static void read_stream_look_ahead ( ReadStream *  stream, bool  suppress_advice  )

static

Definition at line 323 of file read_stream.c.

{
    while (stream->ios_in_progress < stream->max_ios &&
           stream->pinned_buffers + stream->pending_read_nblocks < stream->distance)
    {
        BlockNumber blocknum;
        int16       buffer_index;
        void       *per_buffer_data;
 
        if (stream->pending_read_nblocks == io_combine_limit)
        {
            read_stream_start_pending_read(stream, suppress_advice);
            suppress_advice = false;
            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->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)
        {
            read_stream_start_pending_read(stream, suppress_advice);
            suppress_advice = false;
            if (stream->ios_in_progress == stream->max_ios)
            {
                /* And we've hit the limit.  Rewind, and stop here. */
                read_stream_unget_block(stream, blocknum);
                return;
            }
        }
 
        /* This is the start of a new pending read. */
        stream->pending_read_blocknum = blocknum;
        stream->pending_read_nblocks = 1;
    }
 
    /*
     * 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.  However,
     * if we've already reached io_combine_limit, or we've reached the
     * distance limit and there isn't anything pinned yet, or the callback has
     * signaled end-of-stream, we start the read immediately.
     */
    if (stream->pending_read_nblocks > 0 &&
        (stream->pending_read_nblocks == io_combine_limit ||
         (stream->pending_read_nblocks == stream->distance &&
          stream->pinned_buffers == 0) ||
         stream->distance == 0) &&
        stream->ios_in_progress < stream->max_ios)
        read_stream_start_pending_read(stream, suppress_advice);
}

References Assert, ReadStream::distance, get_per_buffer_data(), InvalidBlockNumber, io_combine_limit, ReadStream::ios_in_progress, ReadStream::max_ios, ReadStream::next_buffer_index, ReadStream::pending_read_blocknum, ReadStream::pending_read_nblocks, ReadStream::pinned_buffers, ReadStream::queue_size, read_stream_get_block(), read_stream_start_pending_read(), and read_stream_unget_block().

Referenced by read_stream_next_buffer().

read_stream_next_buffer()

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

Definition at line 570 of file read_stream.c.

{
    Buffer      buffer;
    int16       oldest_buffer_index;
 
#ifndef READ_STREAM_DISABLE_FAST_PATH
 
    /*
     * A fast path for all-cached scans (behavior A).  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->pinned_buffers == 1);
        Assert(stream->distance == 1);
        Assert(stream->pending_read_nblocks == 0);
        Assert(stream->per_buffer_data_size == 0);
 
        /* 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. */
        if (unlikely(stream->blocknums_next == stream->blocknums_count))
            read_stream_fill_blocknums(stream);
        next_blocknum = stream->blocknums[stream->blocknums_next++];
 
        if (likely(next_blocknum != InvalidBlockNumber))
        {
            /*
             * 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.
             */
            if (likely(!StartReadBuffer(&stream->ios[0].op,
                                        &stream->buffers[oldest_buffer_index],
                                        next_blocknum,
                                        stream->advice_enabled ?
                                        READ_BUFFERS_ISSUE_ADVICE : 0)))
            {
                /* Fast return. */
                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;
        }
        else
        {
            /* No more blocks, end of stream. */
            stream->distance = 0;
            stream->oldest_buffer_index = stream->next_buffer_index;
            stream->pinned_buffers = 0;
        }
 
        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->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, true);
 
        /* End of stream reached? */
        if (stream->pinned_buffers == 0)
        {
            Assert(stream->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;
        int16       distance;
 
        /* Sanity check that we still agree on the buffers. */
        Assert(stream->ios[io_index].op.buffers ==
               &stream->buffers[oldest_buffer_index]);
 
        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 (stream->ios[io_index].op.flags & READ_BUFFERS_ISSUE_ADVICE)
        {
            /* Distance ramps up fast (behavior C). */
            distance = stream->distance * 2;
            distance = Min(distance, stream->max_pinned_buffers);
            stream->distance = distance;
        }
        else
        {
            /* No advice; move towards io_combine_limit (behavior B). */
            if (stream->distance > io_combine_limit)
            {
                stream->distance--;
            }
            else
            {
                distance = stream->distance * 2;
                distance = Min(distance, io_combine_limit);
                distance = Min(distance, stream->max_pinned_buffers);
                stream->distance = distance;
            }
        }
    }
 
#ifdef CLOBBER_FREED_MEMORY
    /* Clobber old buffer and per-buffer data for debugging purposes. */
    stream->buffers[oldest_buffer_index] = InvalidBuffer;
 
    /*
     * 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)
        wipe_mem(get_per_buffer_data(stream,
                                     oldest_buffer_index == 0 ?
                                     stream->queue_size - 1 :
                                     oldest_buffer_index - 1),
                 stream->per_buffer_data_size);
#endif
 
    /* 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, false);
 
#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->pinned_buffers == 1 &&
        stream->distance == 1 &&
        stream->pending_read_nblocks == 0 &&
        stream->per_buffer_data_size == 0)
    {
        stream->fast_path = true;
    }
#endif
 
    return buffer;
}

References ReadStream::advice_enabled, Assert, ReadStream::blocknums, ReadStream::blocknums_count, ReadStream::blocknums_next, InProgressIO::buffer_index, BufferIsValid(), ReadStream::buffers, ReadBuffersOperation::buffers, ReadStream::distance, ReadStream::fast_path, ReadBuffersOperation::flags, get_per_buffer_data(), InvalidBlockNumber, InvalidBuffer, 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, READ_BUFFERS_ISSUE_ADVICE, read_stream_fill_blocknums(), read_stream_look_ahead(), ReadStream::seq_blocknum, StartReadBuffer(), unlikely, and WaitReadBuffers().

Referenced by heap_fetch_next_buffer(), heapam_scan_analyze_next_block(), pg_prewarm(), and read_stream_reset().

read_stream_reset()

void read_stream_reset

(

ReadStream *

stream

)

Definition at line 773 of file read_stream.c.

{
    Buffer      buffer;
 
    /* Stop looking ahead. */
    stream->distance = 0;
 
    /* Forget buffered block numbers and fast path state. */
    stream->blocknums_next = 0;
    stream->blocknums_count = 0;
    stream->fast_path = false;
 
    /* Unpin anything that wasn't consumed. */
    while ((buffer = read_stream_next_buffer(stream, NULL)) != InvalidBuffer)
        ReleaseBuffer(buffer);
 
    Assert(stream->pinned_buffers == 0);
    Assert(stream->ios_in_progress == 0);
 
    /* Start off assuming data is cached. */
    stream->distance = 1;
}

References Assert, ReadStream::blocknums_count, ReadStream::blocknums_next, ReadStream::distance, ReadStream::fast_path, InvalidBuffer, ReadStream::ios_in_progress, ReadStream::pinned_buffers, read_stream_next_buffer(), and ReleaseBuffer().

Referenced by heap_fetch_next_buffer(), heap_rescan(), and read_stream_end().

read_stream_start_pending_read()

static void read_stream_start_pending_read ( ReadStream *  stream, bool  suppress_advice  )

static

Definition at line 233 of file read_stream.c.

{
    bool        need_wait;
    int         nblocks;
    int         flags;
    int16       io_index;
    int16       overflow;
    int16       buffer_index;
 
    /* This should only be called with a pending read. */
    Assert(stream->pending_read_nblocks > 0);
    Assert(stream->pending_read_nblocks <= io_combine_limit);
 
    /* We had better not exceed the pin limit by starting this read. */
    Assert(stream->pinned_buffers + stream->pending_read_nblocks <=
           stream->max_pinned_buffers);
 
    /* 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);
 
    /*
     * If advice hasn't been suppressed, this system supports it, and this
     * isn't a strictly sequential pattern, then we'll issue advice.
     */
    if (!suppress_advice &&
        stream->advice_enabled &&
        stream->pending_read_blocknum != stream->seq_blocknum)
        flags = READ_BUFFERS_ISSUE_ADVICE;
    else
        flags = 0;
 
    /* We say how many blocks we want to read, but may be smaller on return. */
    buffer_index = stream->next_buffer_index;
    io_index = stream->next_io_index;
    nblocks = stream->pending_read_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)
    {
        /* Look-ahead distance decays, no I/O necessary (behavior A). */
        if (stream->distance > 1)
            stream->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;
    }
 
    /*
     * We gave a contiguous range of buffer space to StartReadBuffers(), but
     * we want it to wrap around at queue_size.  Slide overflowing buffers to
     * the front of the array.
     */
    overflow = (buffer_index + nblocks) - stream->queue_size;
    if (overflow > 0)
        memmove(&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;
}

References ReadStream::advice_enabled, Assert, InProgressIO::buffer_index, ReadStream::buffers, ReadStream::distance, if(), io_combine_limit, ReadStream::ios, ReadStream::ios_in_progress, ReadStream::max_ios, ReadStream::max_pinned_buffers, ReadStream::next_buffer_index, ReadStream::next_io_index, ReadStream::oldest_buffer_index, InProgressIO::op, ReadStream::pending_read_blocknum, ReadStream::pending_read_nblocks, ReadStream::pinned_buffers, ReadStream::queue_size, READ_BUFFERS_ISSUE_ADVICE, ReadStream::seq_blocknum, and StartReadBuffers().

Referenced by read_stream_look_ahead().

read_stream_unget_block()

static void read_stream_unget_block ( ReadStream *  stream, BlockNumber  blocknum  )

inlinestatic

Definition at line 194 of file read_stream.c.

{
    if (stream->blocknums_next == stream->blocknums_count)
    {
        /* Never initialized or entirely consumed.  Re-initialize. */
        stream->blocknums[0] = blocknum;
        stream->blocknums_count = 1;
        stream->blocknums_next = 0;
    }
    else
    {
        /* Must be the last value return from blocknums array. */
        Assert(stream->blocknums_next > 0);
        stream->blocknums_next--;
        Assert(stream->blocknums[stream->blocknums_next] == blocknum);
    }
}

References Assert, ReadStream::blocknums, ReadStream::blocknums_count, and ReadStream::blocknums_next.

Referenced by read_stream_look_ahead().