read_stream.h File Reference

#include "storage/bufmgr.h"
#include "storage/smgr.h"

Include dependency graph for read_stream.h:

This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Data Structures
struct	BlockRangeReadStreamPrivate

Macros
#define	READ_STREAM_DEFAULT   0x00
#define	READ_STREAM_MAINTENANCE   0x01
#define	READ_STREAM_SEQUENTIAL   0x02
#define	READ_STREAM_FULL   0x04
#define	READ_STREAM_USE_BATCHING   0x08

Typedefs
typedef struct ReadStream	ReadStream
typedef struct BlockRangeReadStreamPrivate	BlockRangeReadStreamPrivate
typedef BlockNumber(*	ReadStreamBlockNumberCB) (ReadStream *stream, void *callback_private_data, void *per_buffer_data)

Functions
BlockNumber	block_range_read_stream_cb (ReadStream *stream, void *callback_private_data, void *per_buffer_data)
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)
BlockNumber	read_stream_next_block (ReadStream *stream, BufferAccessStrategy *strategy)
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)
BlockNumber	read_stream_pause (ReadStream *stream)
void	read_stream_resume (ReadStream *stream)
void	read_stream_reset (ReadStream *stream)
void	read_stream_end (ReadStream *stream)

Macro Definition Documentation

READ_STREAM_DEFAULT

#define READ_STREAM_DEFAULT   0x00

Definition at line 21 of file read_stream.h.

READ_STREAM_FULL

#define READ_STREAM_FULL   0x04

Definition at line 43 of file read_stream.h.

READ_STREAM_MAINTENANCE

#define READ_STREAM_MAINTENANCE   0x01

Definition at line 28 of file read_stream.h.

READ_STREAM_SEQUENTIAL

#define READ_STREAM_SEQUENTIAL   0x02

Definition at line 36 of file read_stream.h.

READ_STREAM_USE_BATCHING

#define READ_STREAM_USE_BATCHING   0x08

Definition at line 64 of file read_stream.h.

Typedef Documentation

BlockRangeReadStreamPrivate

typedef struct BlockRangeReadStreamPrivate BlockRangeReadStreamPrivate

ReadStream

typedef struct ReadStream ReadStream

Definition at line 67 of file read_stream.h.

ReadStreamBlockNumberCB

typedef BlockNumber(* ReadStreamBlockNumberCB) (ReadStream *stream, void *callback_private_data, void *per_buffer_data)

Definition at line 77 of file read_stream.h.

Function Documentation

block_range_read_stream_cb()

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

extern

Definition at line 163 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().

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  )

extern

Definition at line 739 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(), 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  )

extern

Definition at line 763 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_end()

void read_stream_end ( ReadStream *  stream )

extern

Definition at line 1116 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(), RelationCopyStorageUsingBuffer(), spgvacuumscan(), statapprox_heap(), and verify_heapam().

read_stream_next_block()

BlockNumber read_stream_next_block ( ReadStream *  stream, BufferAccessStrategy *  strategy  )

extern

Definition at line 1033 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  )

extern

Definition at line 793 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->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;
 
            /*
             * 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. */
                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->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->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->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;
        int32       distance;   /* wider temporary value, clamped below */
 
        /* 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;
 
        /* Look-ahead distance ramps up rapidly after we do I/O. */
        distance = stream->distance * 2;
        distance = Min(distance, stream->max_pinned_buffers);
        stream->distance = 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
 
    /* 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->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::distance, ReadStream::fast_path, fb(), 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_stream_get_block(), read_stream_look_ahead(), 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_reset(), RelationCopyStorageUsingBuffer(), spgvacuumscan(), statapprox_heap(), and verify_heapam().

read_stream_pause()

BlockNumber read_stream_pause ( ReadStream *  stream )

extern

Definition at line 1045 of file read_stream.c.

{
    stream->resume_distance = stream->distance;
    stream->distance = 0;
    return InvalidBlockNumber;
}

References ReadStream::distance, InvalidBlockNumber, and ReadStream::resume_distance.

read_stream_reset()

void read_stream_reset ( ReadStream *  stream )

extern

Definition at line 1070 of file read_stream.c.

{
    int16       index;
    Buffer      buffer;
 
    /* Stop looking ahead. */
    stream->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->distance = 1;
    stream->resume_distance = stream->distance;
}

References Assert, ReadStream::buffered_blocknum, ReadStream::buffers, ReadStream::distance, 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(), ReleaseBuffer(), and ReadStream::resume_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 )

extern

Definition at line 1058 of file read_stream.c.

{
    stream->distance = stream->resume_distance;
}

References ReadStream::distance, and ReadStream::resume_distance.