#include "access/xlogdefs.h"
#include "access/xlogreader.h"
#include "access/xlogrecord.h"

Include dependency graph for xlogprefetcher.h:

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

Typedefs
typedef struct XLogPrefetcher	XLogPrefetcher

Enumerations
enum	RecoveryPrefetchValue { RECOVERY_PREFETCH_OFF , RECOVERY_PREFETCH_ON , RECOVERY_PREFETCH_TRY }

Functions
void	XLogPrefetchReconfigure (void)

size_t	XLogPrefetchShmemSize (void)

void	XLogPrefetchShmemInit (void)

void	XLogPrefetchResetStats (void)

XLogPrefetcher *	XLogPrefetcherAllocate (XLogReaderState *reader)

void	XLogPrefetcherFree (XLogPrefetcher *prefetcher)

XLogReaderState *	XLogPrefetcherGetReader (XLogPrefetcher *prefetcher)

void	XLogPrefetcherBeginRead (XLogPrefetcher *prefetcher, XLogRecPtr recPtr)

XLogRecord *	XLogPrefetcherReadRecord (XLogPrefetcher prefetcher, char *errmsg)

void	XLogPrefetcherComputeStats (XLogPrefetcher *prefetcher)

Variables
PGDLLIMPORT int	recovery_prefetch

Typedef Documentation

◆ XLogPrefetcher

typedef struct XLogPrefetcher XLogPrefetcher

Definition at line 21 of file xlogprefetcher.h.

Enumeration Type Documentation

◆ RecoveryPrefetchValue

enum RecoveryPrefetchValue

Enumerator
RECOVERY_PREFETCH_OFF
RECOVERY_PREFETCH_ON
RECOVERY_PREFETCH_TRY

Definition at line 24 of file xlogprefetcher.h.

 {
     RECOVERY_PREFETCH_OFF,
     RECOVERY_PREFETCH_ON,
     RECOVERY_PREFETCH_TRY,
 }           RecoveryPrefetchValue;

Function Documentation

◆ XLogPrefetcherAllocate()

XLogPrefetcher* XLogPrefetcherAllocate ( XLogReaderState * reader )

Definition at line 362 of file xlogprefetcher.c.

 {
     XLogPrefetcher *prefetcher;
     static HASHCTL hash_table_ctl = {
         .keysize = sizeof(RelFileLocator),
         .entrysize = sizeof(XLogPrefetcherFilter)
     };
  
     prefetcher = palloc0(sizeof(XLogPrefetcher));
  
     prefetcher->reader = reader;
     prefetcher->filter_table = hash_create("XLogPrefetcherFilterTable", 1024,
                                            &hash_table_ctl,
                                            HASH_ELEM | HASH_BLOBS);
     dlist_init(&prefetcher->filter_queue);
  
     SharedStats->wal_distance = 0;
     SharedStats->block_distance = 0;
     SharedStats->io_depth = 0;
  
     /* First usage will cause streaming_read to be allocated. */
     prefetcher->reconfigure_count = XLogPrefetchReconfigureCount - 1;
  
     return prefetcher;
 }

References XLogPrefetchStats::block_distance, dlist_init(), XLogPrefetcher::filter_queue, XLogPrefetcher::filter_table, HASH_BLOBS, hash_create(), HASH_ELEM, XLogPrefetchStats::io_depth, HASHCTL::keysize, palloc0(), XLogPrefetcher::reader, XLogPrefetcher::reconfigure_count, SharedStats, XLogPrefetchStats::wal_distance, and XLogPrefetchReconfigureCount.

Referenced by InitWalRecovery().

◆ XLogPrefetcherBeginRead()

void XLogPrefetcherBeginRead	(	XLogPrefetcher *	prefetcher,
		XLogRecPtr	recPtr
	)

Definition at line 964 of file xlogprefetcher.c.

 {
     /* This will forget about any in-flight IO. */
     prefetcher->reconfigure_count--;
  
     /* Book-keeping to avoid readahead on first read. */
     prefetcher->begin_ptr = recPtr;
  
     prefetcher->no_readahead_until = 0;
  
     /* This will forget about any queued up records in the decoder. */
     XLogBeginRead(prefetcher->reader, recPtr);
 }

References XLogPrefetcher::begin_ptr, XLogPrefetcher::no_readahead_until, XLogPrefetcher::reader, XLogPrefetcher::reconfigure_count, and XLogBeginRead().

Referenced by FinishWalRecovery(), InitWalRecovery(), PerformWalRecovery(), and ReadCheckpointRecord().

◆ XLogPrefetcherComputeStats()

void XLogPrefetcherComputeStats ( XLogPrefetcher * prefetcher )

Definition at line 412 of file xlogprefetcher.c.

 {
     uint32      io_depth;
     uint32      completed;
     int64       wal_distance;
  
  
     /* How far ahead of replay are we now? */
     if (prefetcher->reader->decode_queue_tail)
     {
         wal_distance =
             prefetcher->reader->decode_queue_tail->lsn -
             prefetcher->reader->decode_queue_head->lsn;
     }
     else
     {
         wal_distance = 0;
     }
  
     /* How many IOs are currently in flight and completed? */
     io_depth = lrq_inflight(prefetcher->streaming_read);
     completed = lrq_completed(prefetcher->streaming_read);
  
     /* Update the instantaneous stats visible in pg_stat_recovery_prefetch. */
     SharedStats->io_depth = io_depth;
     SharedStats->block_distance = io_depth + completed;
     SharedStats->wal_distance = wal_distance;
  
     prefetcher->next_stats_shm_lsn =
         prefetcher->reader->ReadRecPtr + XLOGPREFETCHER_STATS_DISTANCE;
 }

References XLogPrefetchStats::block_distance, XLogReaderState::decode_queue_head, XLogReaderState::decode_queue_tail, XLogPrefetchStats::io_depth, lrq_completed(), lrq_inflight(), DecodedXLogRecord::lsn, XLogPrefetcher::next_stats_shm_lsn, XLogPrefetcher::reader, XLogReaderState::ReadRecPtr, SharedStats, XLogPrefetcher::streaming_read, XLogPrefetchStats::wal_distance, and XLOGPREFETCHER_STATS_DISTANCE.

Referenced by ShutdownWalRecovery(), WaitForWALToBecomeAvailable(), and XLogPrefetcherReadRecord().

◆ XLogPrefetcherFree()

void XLogPrefetcherFree ( XLogPrefetcher * prefetcher )

Definition at line 392 of file xlogprefetcher.c.

 {
     lrq_free(prefetcher->streaming_read);
     hash_destroy(prefetcher->filter_table);
     pfree(prefetcher);
 }

References XLogPrefetcher::filter_table, hash_destroy(), lrq_free(), pfree(), and XLogPrefetcher::streaming_read.

Referenced by ShutdownWalRecovery().

◆ XLogPrefetcherGetReader()

XLogReaderState* XLogPrefetcherGetReader ( XLogPrefetcher * prefetcher )

Definition at line 403 of file xlogprefetcher.c.

 {
     return prefetcher->reader;
 }

References XLogPrefetcher::reader.

Referenced by ReadRecord().

◆ XLogPrefetcherReadRecord()

XLogRecord* XLogPrefetcherReadRecord	(	XLogPrefetcher *	prefetcher,
		char **	errmsg
	)

Definition at line 983 of file xlogprefetcher.c.

 {
     DecodedXLogRecord *record;
     XLogRecPtr  replayed_up_to;
  
     /*
      * See if it's time to reset the prefetching machinery, because a relevant
      * GUC was changed.
      */
     if (unlikely(XLogPrefetchReconfigureCount != prefetcher->reconfigure_count))
     {
         uint32      max_distance;
         uint32      max_inflight;
  
         if (prefetcher->streaming_read)
             lrq_free(prefetcher->streaming_read);
  
         if (RecoveryPrefetchEnabled())
         {
             Assert(maintenance_io_concurrency > 0);
             max_inflight = maintenance_io_concurrency;
             max_distance = max_inflight * XLOGPREFETCHER_DISTANCE_MULTIPLIER;
         }
         else
         {
             max_inflight = 1;
             max_distance = 1;
         }
  
         prefetcher->streaming_read = lrq_alloc(max_distance,
                                                max_inflight,
                                                (uintptr_t) prefetcher,
                                                XLogPrefetcherNextBlock);
  
         prefetcher->reconfigure_count = XLogPrefetchReconfigureCount;
     }
  
     /*
      * Release last returned record, if there is one, as it's now been
      * replayed.
      */
     replayed_up_to = XLogReleasePreviousRecord(prefetcher->reader);
  
     /*
      * Can we drop any filters yet?  If we were waiting for a relation to be
      * created or extended, it is now OK to access blocks in the covered
      * range.
      */
     XLogPrefetcherCompleteFilters(prefetcher, replayed_up_to);
  
     /*
      * All IO initiated by earlier WAL is now completed.  This might trigger
      * further prefetching.
      */
     lrq_complete_lsn(prefetcher->streaming_read, replayed_up_to);
  
     /*
      * If there's nothing queued yet, then start prefetching to cause at least
      * one record to be queued.
      */
     if (!XLogReaderHasQueuedRecordOrError(prefetcher->reader))
     {
         Assert(lrq_inflight(prefetcher->streaming_read) == 0);
         Assert(lrq_completed(prefetcher->streaming_read) == 0);
         lrq_prefetch(prefetcher->streaming_read);
     }
  
     /* Read the next record. */
     record = XLogNextRecord(prefetcher->reader, errmsg);
     if (!record)
         return NULL;
  
     /*
      * The record we just got is the "current" one, for the benefit of the
      * XLogRecXXX() macros.
      */
     Assert(record == prefetcher->reader->record);
  
     /*
      * If maintenance_io_concurrency is set very low, we might have started
      * prefetching some but not all of the blocks referenced in the record
      * we're about to return.  Forget about the rest of the blocks in this
      * record by dropping the prefetcher's reference to it.
      */
     if (record == prefetcher->record)
         prefetcher->record = NULL;
  
     /*
      * See if it's time to compute some statistics, because enough WAL has
      * been processed.
      */
     if (unlikely(record->lsn >= prefetcher->next_stats_shm_lsn))
         XLogPrefetcherComputeStats(prefetcher);
  
     Assert(record == prefetcher->reader->record);
  
     return &record->header;
 }

References Assert, errmsg(), DecodedXLogRecord::header, lrq_alloc(), lrq_complete_lsn(), lrq_completed(), lrq_free(), lrq_inflight(), lrq_prefetch(), DecodedXLogRecord::lsn, maintenance_io_concurrency, XLogPrefetcher::next_stats_shm_lsn, XLogPrefetcher::reader, XLogPrefetcher::reconfigure_count, XLogPrefetcher::record, XLogReaderState::record, RecoveryPrefetchEnabled, XLogPrefetcher::streaming_read, unlikely, XLogNextRecord(), XLOGPREFETCHER_DISTANCE_MULTIPLIER, XLogPrefetcherCompleteFilters(), XLogPrefetcherComputeStats(), XLogPrefetcherNextBlock(), XLogPrefetchReconfigureCount, XLogReaderHasQueuedRecordOrError(), and XLogReleasePreviousRecord().

Referenced by ReadRecord().

◆ XLogPrefetchReconfigure()

void XLogPrefetchReconfigure ( void )

Definition at line 340 of file xlogprefetcher.c.

 {
     XLogPrefetchReconfigureCount++;
 }

References XLogPrefetchReconfigureCount.

Referenced by ApplyWalRecord(), assign_maintenance_io_concurrency(), and assign_recovery_prefetch().

◆ XLogPrefetchResetStats()

void XLogPrefetchResetStats ( void )

Definition at line 303 of file xlogprefetcher.c.

 {
     pg_atomic_write_u64(&SharedStats->reset_time, GetCurrentTimestamp());
     pg_atomic_write_u64(&SharedStats->prefetch, 0);
     pg_atomic_write_u64(&SharedStats->hit, 0);
     pg_atomic_write_u64(&SharedStats->skip_init, 0);
     pg_atomic_write_u64(&SharedStats->skip_new, 0);
     pg_atomic_write_u64(&SharedStats->skip_fpw, 0);
     pg_atomic_write_u64(&SharedStats->skip_rep, 0);
 }

References GetCurrentTimestamp(), XLogPrefetchStats::hit, pg_atomic_write_u64(), XLogPrefetchStats::prefetch, XLogPrefetchStats::reset_time, SharedStats, XLogPrefetchStats::skip_fpw, XLogPrefetchStats::skip_init, XLogPrefetchStats::skip_new, and XLogPrefetchStats::skip_rep.

Referenced by pg_stat_reset_shared().

◆ XLogPrefetchShmemInit()

void XLogPrefetchShmemInit ( void )

Definition at line 315 of file xlogprefetcher.c.

 {
     bool        found;
  
     SharedStats = (XLogPrefetchStats *)
         ShmemInitStruct("XLogPrefetchStats",
                         sizeof(XLogPrefetchStats),
                         &found);
  
     if (!found)
     {
         pg_atomic_init_u64(&SharedStats->reset_time, GetCurrentTimestamp());
         pg_atomic_init_u64(&SharedStats->prefetch, 0);
         pg_atomic_init_u64(&SharedStats->hit, 0);
         pg_atomic_init_u64(&SharedStats->skip_init, 0);
         pg_atomic_init_u64(&SharedStats->skip_new, 0);
         pg_atomic_init_u64(&SharedStats->skip_fpw, 0);
         pg_atomic_init_u64(&SharedStats->skip_rep, 0);
     }
 }

References GetCurrentTimestamp(), XLogPrefetchStats::hit, pg_atomic_init_u64(), XLogPrefetchStats::prefetch, XLogPrefetchStats::reset_time, SharedStats, ShmemInitStruct(), XLogPrefetchStats::skip_fpw, XLogPrefetchStats::skip_init, XLogPrefetchStats::skip_new, and XLogPrefetchStats::skip_rep.

Referenced by CreateOrAttachShmemStructs().

◆ XLogPrefetchShmemSize()

size_t XLogPrefetchShmemSize ( void )

Definition at line 294 of file xlogprefetcher.c.

 {
     return sizeof(XLogPrefetchStats);
 }

Referenced by CalculateShmemSize().

Variable Documentation

◆ recovery_prefetch

PGDLLIMPORT int recovery_prefetch

extern

Definition at line 68 of file xlogprefetcher.c.

Referenced by assign_recovery_prefetch().

Typedefs

Enumerations

Functions

Variables