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

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

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)

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.

Typedef Documentation

◆ BlockRangeReadStreamPrivate

typedef struct BlockRangeReadStreamPrivate BlockRangeReadStreamPrivate

◆ ReadStream

typedef struct ReadStream ReadStream

Definition at line 1 of file read_stream.h.

◆ ReadStreamBlockNumberCB

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

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

Definition at line 172 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 collect_visibility_data(), pg_prewarm(), and RelationCopyStorageUsingBuffer().

◆ 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 552 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(), collect_corrupt_items(), collect_visibility_data(), heap_beginscan(), and pg_prewarm().

◆ 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 576 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(), and read_stream_begin_impl().

Referenced by RelationCopyStorageUsingBuffer().

◆ read_stream_end()

void read_stream_end ( ReadStream * stream )

Definition at line 847 of file read_stream.c.

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

References pfree(), and read_stream_reset().

Referenced by acquire_sample_rows(), collect_corrupt_items(), collect_visibility_data(), heap_endscan(), pg_prewarm(), and RelationCopyStorageUsingBuffer().

◆ read_stream_next_block()

BlockNumber read_stream_next_block	(	ReadStream *	stream,
		BufferAccessStrategy *	strategy
	)

Definition at line 808 of file read_stream.c.

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

References 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 606 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. */
         next_blocknum = read_stream_get_block(stream, NULL);
  
         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, 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_get_block(), read_stream_look_ahead(), ReadStream::seq_blocknum, StartReadBuffer(), unlikely, and WaitReadBuffers().

Referenced by collect_corrupt_items(), collect_visibility_data(), heap_fetch_next_buffer(), heapam_scan_analyze_next_block(), pg_prewarm(), read_stream_reset(), and RelationCopyStorageUsingBuffer().

◆ read_stream_reset()

void read_stream_reset ( ReadStream * stream )

Definition at line 821 of file read_stream.c.

 {
     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);
  
     Assert(stream->pinned_buffers == 0);
     Assert(stream->ios_in_progress == 0);
  
     /* Start off assuming data is cached. */
     stream->distance = 1;
 }

References Assert, ReadStream::buffered_blocknum, ReadStream::distance, ReadStream::fast_path, InvalidBlockNumber, 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().

Data Structures

Macros

Typedefs

Functions