xlogprefetcher.c File Reference

#include "postgres.h"
#include "access/xlog.h"
#include "access/xlogprefetcher.h"
#include "access/xlogreader.h"
#include "access/xlogutils.h"
#include "catalog/pg_class.h"
#include "catalog/pg_control.h"
#include "catalog/storage_xlog.h"
#include "commands/dbcommands_xlog.h"
#include "utils/fmgrprotos.h"
#include "utils/timestamp.h"
#include "funcapi.h"
#include "pgstat.h"
#include "miscadmin.h"
#include "port/atomics.h"
#include "storage/bufmgr.h"
#include "storage/shmem.h"
#include "storage/smgr.h"
#include "utils/guc_hooks.h"
#include "utils/hsearch.h"

Include dependency graph for xlogprefetcher.c:

Go to the source code of this file.

Data Structures
struct	LsnReadQueue
struct	XLogPrefetcher
struct	XLogPrefetcherFilter
struct	XLogPrefetchStats

Macros
#define	XLOGPREFETCHER_STATS_DISTANCE   BLCKSZ
#define	XLOGPREFETCHER_SEQ_WINDOW_SIZE   4
#define	XLOGPREFETCHER_DISTANCE_MULTIPLIER   4
#define	RecoveryPrefetchEnabled()   false
#define	PG_STAT_GET_RECOVERY_PREFETCH_COLS   10

Typedefs
typedef LsnReadQueueNextStatus(*	LsnReadQueueNextFun) (uintptr_t lrq_private, XLogRecPtr *lsn)
typedef struct LsnReadQueue	LsnReadQueue
typedef struct XLogPrefetcherFilter	XLogPrefetcherFilter
typedef struct XLogPrefetchStats	XLogPrefetchStats

Enumerations
enum	LsnReadQueueNextStatus { LRQ_NEXT_NO_IO , LRQ_NEXT_IO , LRQ_NEXT_AGAIN }

Functions
static void	XLogPrefetcherAddFilter (XLogPrefetcher *prefetcher, RelFileLocator rlocator, BlockNumber blockno, XLogRecPtr lsn)
static bool	XLogPrefetcherIsFiltered (XLogPrefetcher *prefetcher, RelFileLocator rlocator, BlockNumber blockno)
static void	XLogPrefetcherCompleteFilters (XLogPrefetcher *prefetcher, XLogRecPtr replaying_lsn)
static LsnReadQueueNextStatus	XLogPrefetcherNextBlock (uintptr_t pgsr_private, XLogRecPtr *lsn)
static LsnReadQueue *	lrq_alloc (uint32 max_distance, uint32 max_inflight, uintptr_t lrq_private, LsnReadQueueNextFun next)
static void	lrq_free (LsnReadQueue *lrq)
static uint32	lrq_inflight (LsnReadQueue *lrq)
static uint32	lrq_completed (LsnReadQueue *lrq)
static void	lrq_prefetch (LsnReadQueue *lrq)
static void	lrq_complete_lsn (LsnReadQueue *lrq, XLogRecPtr lsn)
size_t	XLogPrefetchShmemSize (void)
void	XLogPrefetchResetStats (void)
void	XLogPrefetchShmemInit (void)
void	XLogPrefetchReconfigure (void)
static void	XLogPrefetchIncrement (pg_atomic_uint64 *counter)
XLogPrefetcher *	XLogPrefetcherAllocate (XLogReaderState *reader)
void	XLogPrefetcherFree (XLogPrefetcher *prefetcher)
XLogReaderState *	XLogPrefetcherGetReader (XLogPrefetcher *prefetcher)
void	XLogPrefetcherComputeStats (XLogPrefetcher *prefetcher)
Datum	pg_stat_get_recovery_prefetch (PG_FUNCTION_ARGS)
void	XLogPrefetcherBeginRead (XLogPrefetcher *prefetcher, XLogRecPtr recPtr)
XLogRecord *	XLogPrefetcherReadRecord (XLogPrefetcher *prefetcher, char **errmsg)
bool	check_recovery_prefetch (int *new_value, void **extra, GucSource source)
void	assign_recovery_prefetch (int new_value, void *extra)

Variables
int	recovery_prefetch = RECOVERY_PREFETCH_TRY
static int	XLogPrefetchReconfigureCount = 0
static XLogPrefetchStats *	SharedStats

Macro Definition Documentation

PG_STAT_GET_RECOVERY_PREFETCH_COLS

#define PG_STAT_GET_RECOVERY_PREFETCH_COLS   10

RecoveryPrefetchEnabled

#define RecoveryPrefetchEnabled

(

)

false

Definition at line 79 of file xlogprefetcher.c.

XLOGPREFETCHER_DISTANCE_MULTIPLIER

#define XLOGPREFETCHER_DISTANCE_MULTIPLIER   4

Definition at line 66 of file xlogprefetcher.c.

XLOGPREFETCHER_SEQ_WINDOW_SIZE

#define XLOGPREFETCHER_SEQ_WINDOW_SIZE   4

Definition at line 60 of file xlogprefetcher.c.

XLOGPREFETCHER_STATS_DISTANCE

#define XLOGPREFETCHER_STATS_DISTANCE   BLCKSZ

Definition at line 54 of file xlogprefetcher.c.

Typedef Documentation

LsnReadQueue

typedef struct LsnReadQueue LsnReadQueue

LsnReadQueueNextFun

typedef LsnReadQueueNextStatus(* LsnReadQueueNextFun) (uintptr_t lrq_private, XLogRecPtr *lsn)

Definition at line 98 of file xlogprefetcher.c.

XLogPrefetcherFilter

typedef struct XLogPrefetcherFilter XLogPrefetcherFilter

XLogPrefetchStats

typedef struct XLogPrefetchStats XLogPrefetchStats

Enumeration Type Documentation

LsnReadQueueNextStatus

enum LsnReadQueueNextStatus

Enumerator
LRQ_NEXT_NO_IO
LRQ_NEXT_IO
LRQ_NEXT_AGAIN

Definition at line 87 of file xlogprefetcher.c.

{
    LRQ_NEXT_NO_IO,
    LRQ_NEXT_IO,
    LRQ_NEXT_AGAIN
} LsnReadQueueNextStatus;

Function Documentation

assign_recovery_prefetch()

void assign_recovery_prefetch	(	int	new_value,
		void *	extra
	)

Definition at line 1101 of file xlogprefetcher.c.

{
    /* Reconfigure prefetching, because a setting it depends on changed. */
    recovery_prefetch = new_value;
    if (AmStartupProcess())
        XLogPrefetchReconfigure();
}

References AmStartupProcess, recovery_prefetch, and XLogPrefetchReconfigure().

check_recovery_prefetch()

bool check_recovery_prefetch	(	int *	new_value,
		void **	extra,
		GucSource	source
	)

Definition at line 1087 of file xlogprefetcher.c.

{
#ifndef USE_PREFETCH
    if (*new_value == RECOVERY_PREFETCH_ON)
    {
        GUC_check_errdetail("recovery_prefetch is not supported on platforms that lack posix_fadvise().");
        return false;
    }
#endif
 
    return true;
}

References GUC_check_errdetail, and RECOVERY_PREFETCH_ON.

lrq_alloc()

static LsnReadQueue* lrq_alloc ( uint32  max_distance, uint32  max_inflight, uintptr_t  lrq_private, LsnReadQueueNextFun  next  )

inlinestatic

Definition at line 206 of file xlogprefetcher.c.

{
    LsnReadQueue *lrq;
    uint32      size;
 
    Assert(max_distance >= max_inflight);
 
    size = max_distance + 1;    /* full ring buffer has a gap */
    lrq = palloc(offsetof(LsnReadQueue, queue) + sizeof(lrq->queue[0]) * size);
    lrq->lrq_private = lrq_private;
    lrq->max_inflight = max_inflight;
    lrq->size = size;
    lrq->next = next;
    lrq->head = 0;
    lrq->tail = 0;
    lrq->inflight = 0;
    lrq->completed = 0;
 
    return lrq;
}

References Assert(), LsnReadQueue::completed, LsnReadQueue::head, LsnReadQueue::inflight, LsnReadQueue::lrq_private, LsnReadQueue::max_inflight, next, LsnReadQueue::next, palloc(), LsnReadQueue::queue, LsnReadQueue::size, and LsnReadQueue::tail.

Referenced by XLogPrefetcherReadRecord().

lrq_complete_lsn()

static void lrq_complete_lsn ( LsnReadQueue *  lrq, XLogRecPtr  lsn  )

inlinestatic

Definition at line 276 of file xlogprefetcher.c.

{
    /*
     * We know that LSNs before 'lsn' have been replayed, so we can now assume
     * that any IOs that were started before then have finished.
     */
    while (lrq->tail != lrq->head &&
           lrq->queue[lrq->tail].lsn < lsn)
    {
        if (lrq->queue[lrq->tail].io)
            lrq->inflight--;
        else
            lrq->completed--;
        lrq->tail++;
        if (lrq->tail == lrq->size)
            lrq->tail = 0;
    }
    if (RecoveryPrefetchEnabled())
        lrq_prefetch(lrq);
}

References LsnReadQueue::completed, LsnReadQueue::head, LsnReadQueue::inflight, LsnReadQueue::io, lrq_prefetch(), LsnReadQueue::lsn, LsnReadQueue::queue, RecoveryPrefetchEnabled, LsnReadQueue::size, and LsnReadQueue::tail.

Referenced by XLogPrefetcherReadRecord().

lrq_completed()

static uint32 lrq_completed ( LsnReadQueue *  lrq )

inlinestatic

Definition at line 243 of file xlogprefetcher.c.

{
    return lrq->completed;
}

References LsnReadQueue::completed.

Referenced by XLogPrefetcherComputeStats(), and XLogPrefetcherReadRecord().

lrq_free()

static void lrq_free ( LsnReadQueue *  lrq )

inlinestatic

Definition at line 231 of file xlogprefetcher.c.

{
    pfree(lrq);
}

References pfree().

Referenced by XLogPrefetcherFree(), and XLogPrefetcherReadRecord().

lrq_inflight()

static uint32 lrq_inflight ( LsnReadQueue *  lrq )

inlinestatic

Definition at line 237 of file xlogprefetcher.c.

{
    return lrq->inflight;
}

References LsnReadQueue::inflight.

Referenced by XLogPrefetcherComputeStats(), and XLogPrefetcherReadRecord().

lrq_prefetch()

static void lrq_prefetch ( LsnReadQueue *  lrq )

inlinestatic

Definition at line 249 of file xlogprefetcher.c.

{
    /* Try to start as many IOs as we can within our limits. */
    while (lrq->inflight < lrq->max_inflight &&
           lrq->inflight + lrq->completed < lrq->size - 1)
    {
        Assert(((lrq->head + 1) % lrq->size) != lrq->tail);
        switch (lrq->next(lrq->lrq_private, &lrq->queue[lrq->head].lsn))
        {
            case LRQ_NEXT_AGAIN:
                return;
            case LRQ_NEXT_IO:
                lrq->queue[lrq->head].io = true;
                lrq->inflight++;
                break;
            case LRQ_NEXT_NO_IO:
                lrq->queue[lrq->head].io = false;
                lrq->completed++;
                break;
        }
        lrq->head++;
        if (lrq->head == lrq->size)
            lrq->head = 0;
    }
}

References Assert(), LsnReadQueue::completed, LsnReadQueue::head, LsnReadQueue::inflight, LsnReadQueue::io, LRQ_NEXT_AGAIN, LRQ_NEXT_IO, LRQ_NEXT_NO_IO, LsnReadQueue::lrq_private, LsnReadQueue::lsn, LsnReadQueue::max_inflight, LsnReadQueue::next, LsnReadQueue::queue, LsnReadQueue::size, and LsnReadQueue::tail.

Referenced by lrq_complete_lsn(), and XLogPrefetcherReadRecord().

pg_stat_get_recovery_prefetch()

Datum pg_stat_get_recovery_prefetch

(

PG_FUNCTION_ARGS

)

Definition at line 830 of file xlogprefetcher.c.

{
#define PG_STAT_GET_RECOVERY_PREFETCH_COLS 10
    ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
    Datum       values[PG_STAT_GET_RECOVERY_PREFETCH_COLS];
    bool        nulls[PG_STAT_GET_RECOVERY_PREFETCH_COLS];
 
    InitMaterializedSRF(fcinfo, 0);
 
    for (int i = 0; i < PG_STAT_GET_RECOVERY_PREFETCH_COLS; ++i)
        nulls[i] = false;
 
    values[0] = TimestampTzGetDatum(pg_atomic_read_u64(&SharedStats->reset_time));
    values[1] = Int64GetDatum(pg_atomic_read_u64(&SharedStats->prefetch));
    values[2] = Int64GetDatum(pg_atomic_read_u64(&SharedStats->hit));
    values[3] = Int64GetDatum(pg_atomic_read_u64(&SharedStats->skip_init));
    values[4] = Int64GetDatum(pg_atomic_read_u64(&SharedStats->skip_new));
    values[5] = Int64GetDatum(pg_atomic_read_u64(&SharedStats->skip_fpw));
    values[6] = Int64GetDatum(pg_atomic_read_u64(&SharedStats->skip_rep));
    values[7] = Int32GetDatum(SharedStats->wal_distance);
    values[8] = Int32GetDatum(SharedStats->block_distance);
    values[9] = Int32GetDatum(SharedStats->io_depth);
    tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc, values, nulls);
 
    return (Datum) 0;
}

References XLogPrefetchStats::block_distance, XLogPrefetchStats::hit, i, InitMaterializedSRF(), Int32GetDatum(), Int64GetDatum(), XLogPrefetchStats::io_depth, pg_atomic_read_u64(), PG_STAT_GET_RECOVERY_PREFETCH_COLS, XLogPrefetchStats::prefetch, XLogPrefetchStats::reset_time, ReturnSetInfo::setDesc, ReturnSetInfo::setResult, SharedStats, XLogPrefetchStats::skip_fpw, XLogPrefetchStats::skip_init, XLogPrefetchStats::skip_new, XLogPrefetchStats::skip_rep, TimestampTzGetDatum(), tuplestore_putvalues(), values, and XLogPrefetchStats::wal_distance.

XLogPrefetcherAddFilter()

static void XLogPrefetcherAddFilter ( XLogPrefetcher *  prefetcher, RelFileLocator  rlocator, BlockNumber  blockno, XLogRecPtr  lsn  )

inlinestatic

Definition at line 862 of file xlogprefetcher.c.

{
    XLogPrefetcherFilter *filter;
    bool        found;
 
    filter = hash_search(prefetcher->filter_table, &rlocator, HASH_ENTER, &found);
    if (!found)
    {
        /*
         * Don't allow any prefetching of this block or higher until replayed.
         */
        filter->filter_until_replayed = lsn;
        filter->filter_from_block = blockno;
        dlist_push_head(&prefetcher->filter_queue, &filter->link);
    }
    else
    {
        /*
         * We were already filtering this rlocator.  Extend the filter's
         * lifetime to cover this WAL record, but leave the lower of the block
         * numbers there because we don't want to have to track individual
         * blocks.
         */
        filter->filter_until_replayed = lsn;
        dlist_delete(&filter->link);
        dlist_push_head(&prefetcher->filter_queue, &filter->link);
        filter->filter_from_block = Min(filter->filter_from_block, blockno);
    }
}

References dlist_delete(), dlist_push_head(), XLogPrefetcherFilter::filter_from_block, XLogPrefetcher::filter_queue, XLogPrefetcher::filter_table, XLogPrefetcherFilter::filter_until_replayed, HASH_ENTER, hash_search(), XLogPrefetcherFilter::link, and Min.

Referenced by XLogPrefetcherNextBlock().

XLogPrefetcherAllocate()

XLogPrefetcher* XLogPrefetcherAllocate

(

XLogReaderState *

reader

)

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

XLogPrefetcherCompleteFilters()

static void XLogPrefetcherCompleteFilters ( XLogPrefetcher *  prefetcher, XLogRecPtr  replaying_lsn  )

inlinestatic

Definition at line 900 of file xlogprefetcher.c.

{
    while (unlikely(!dlist_is_empty(&prefetcher->filter_queue)))
    {
        XLogPrefetcherFilter *filter = dlist_tail_element(XLogPrefetcherFilter,
                                                          link,
                                                          &prefetcher->filter_queue);
 
        if (filter->filter_until_replayed >= replaying_lsn)
            break;
 
        dlist_delete(&filter->link);
        hash_search(prefetcher->filter_table, filter, HASH_REMOVE, NULL);
    }
}

References dlist_delete(), dlist_is_empty(), dlist_tail_element, XLogPrefetcher::filter_queue, XLogPrefetcher::filter_table, XLogPrefetcherFilter::filter_until_replayed, HASH_REMOVE, hash_search(), XLogPrefetcherFilter::link, and unlikely.

Referenced by XLogPrefetcherReadRecord().

XLogPrefetcherComputeStats()

void XLogPrefetcherComputeStats

(

XLogPrefetcher *

prefetcher

)

Definition at line 416 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 396 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 407 of file xlogprefetcher.c.

{
    return prefetcher->reader;
}

References XLogPrefetcher::reader.

Referenced by ReadRecord().

XLogPrefetcherIsFiltered()

static bool XLogPrefetcherIsFiltered ( XLogPrefetcher *  prefetcher, RelFileLocator  rlocator, BlockNumber  blockno  )

inlinestatic

Definition at line 920 of file xlogprefetcher.c.

{
    /*
     * Test for empty queue first, because we expect it to be empty most of
     * the time and we can avoid the hash table lookup in that case.
     */
    if (unlikely(!dlist_is_empty(&prefetcher->filter_queue)))
    {
        XLogPrefetcherFilter *filter;
 
        /* See if the block range is filtered. */
        filter = hash_search(prefetcher->filter_table, &rlocator, HASH_FIND, NULL);
        if (filter && filter->filter_from_block <= blockno)
        {
#ifdef XLOGPREFETCHER_DEBUG_LEVEL
            elog(XLOGPREFETCHER_DEBUG_LEVEL,
                 "prefetch of %u/%u/%u block %u suppressed; filtering until LSN %X/%X is replayed (blocks >= %u filtered)",
                 rlocator.spcOid, rlocator.dbOid, rlocator.relNumber, blockno,
                 LSN_FORMAT_ARGS(filter->filter_until_replayed),
                 filter->filter_from_block);
#endif
            return true;
        }
 
        /* See if the whole database is filtered. */
        rlocator.relNumber = InvalidRelFileNumber;
        rlocator.spcOid = InvalidOid;
        filter = hash_search(prefetcher->filter_table, &rlocator, HASH_FIND, NULL);
        if (filter)
        {
#ifdef XLOGPREFETCHER_DEBUG_LEVEL
            elog(XLOGPREFETCHER_DEBUG_LEVEL,
                 "prefetch of %u/%u/%u block %u suppressed; filtering until LSN %X/%X is replayed (whole database)",
                 rlocator.spcOid, rlocator.dbOid, rlocator.relNumber, blockno,
                 LSN_FORMAT_ARGS(filter->filter_until_replayed));
#endif
            return true;
        }
    }
 
    return false;
}

References RelFileLocator::dbOid, dlist_is_empty(), elog(), XLogPrefetcherFilter::filter_from_block, XLogPrefetcher::filter_queue, XLogPrefetcher::filter_table, XLogPrefetcherFilter::filter_until_replayed, HASH_FIND, hash_search(), InvalidOid, InvalidRelFileNumber, LSN_FORMAT_ARGS, RelFileLocator::relNumber, RelFileLocator::spcOid, and unlikely.

Referenced by XLogPrefetcherNextBlock().

XLogPrefetcherNextBlock()

static LsnReadQueueNextStatus XLogPrefetcherNextBlock ( uintptr_t  pgsr_private, XLogRecPtr *  lsn  )

static

Definition at line 465 of file xlogprefetcher.c.

{
    XLogPrefetcher *prefetcher = (XLogPrefetcher *) pgsr_private;
    XLogReaderState *reader = prefetcher->reader;
    XLogRecPtr  replaying_lsn = reader->ReadRecPtr;
 
    /*
     * We keep track of the record and block we're up to between calls with
     * prefetcher->record and prefetcher->next_block_id.
     */
    for (;;)
    {
        DecodedXLogRecord *record;
 
        /* Try to read a new future record, if we don't already have one. */
        if (prefetcher->record == NULL)
        {
            bool        nonblocking;
 
            /*
             * If there are already records or an error queued up that could
             * be replayed, we don't want to block here.  Otherwise, it's OK
             * to block waiting for more data: presumably the caller has
             * nothing else to do.
             */
            nonblocking = XLogReaderHasQueuedRecordOrError(reader);
 
            /* Readahead is disabled until we replay past a certain point. */
            if (nonblocking && replaying_lsn <= prefetcher->no_readahead_until)
                return LRQ_NEXT_AGAIN;
 
            record = XLogReadAhead(prefetcher->reader, nonblocking);
            if (record == NULL)
            {
                /*
                 * We can't read any more, due to an error or lack of data in
                 * nonblocking mode.  Don't try to read ahead again until
                 * we've replayed everything already decoded.
                 */
                if (nonblocking && prefetcher->reader->decode_queue_tail)
                    prefetcher->no_readahead_until =
                        prefetcher->reader->decode_queue_tail->lsn;
 
                return LRQ_NEXT_AGAIN;
            }
 
            /*
             * If prefetching is disabled, we don't need to analyze the record
             * or issue any prefetches.  We just need to cause one record to
             * be decoded.
             */
            if (!RecoveryPrefetchEnabled())
            {
                *lsn = InvalidXLogRecPtr;
                return LRQ_NEXT_NO_IO;
            }
 
            /* We have a new record to process. */
            prefetcher->record = record;
            prefetcher->next_block_id = 0;
        }
        else
        {
            /* Continue to process from last call, or last loop. */
            record = prefetcher->record;
        }
 
        /*
         * Check for operations that require us to filter out block ranges, or
         * pause readahead completely.
         */
        if (replaying_lsn < record->lsn)
        {
            uint8       rmid = record->header.xl_rmid;
            uint8       record_type = record->header.xl_info & ~XLR_INFO_MASK;
 
            if (rmid == RM_XLOG_ID)
            {
                if (record_type == XLOG_CHECKPOINT_SHUTDOWN ||
                    record_type == XLOG_END_OF_RECOVERY)
                {
                    /*
                     * These records might change the TLI.  Avoid potential
                     * bugs if we were to allow "read TLI" and "replay TLI" to
                     * differ without more analysis.
                     */
                    prefetcher->no_readahead_until = record->lsn;
 
#ifdef XLOGPREFETCHER_DEBUG_LEVEL
                    elog(XLOGPREFETCHER_DEBUG_LEVEL,
                         "suppressing all readahead until %X/%X is replayed due to possible TLI change",
                         LSN_FORMAT_ARGS(record->lsn));
#endif
 
                    /* Fall through so we move past this record. */
                }
            }
            else if (rmid == RM_DBASE_ID)
            {
                /*
                 * When databases are created with the file-copy strategy,
                 * there are no WAL records to tell us about the creation of
                 * individual relations.
                 */
                if (record_type == XLOG_DBASE_CREATE_FILE_COPY)
                {
                    xl_dbase_create_file_copy_rec *xlrec =
                        (xl_dbase_create_file_copy_rec *) record->main_data;
                    RelFileLocator rlocator =
                    {InvalidOid, xlrec->db_id, InvalidRelFileNumber};
 
                    /*
                     * Don't try to prefetch anything in this database until
                     * it has been created, or we might confuse the blocks of
                     * different generations, if a database OID or
                     * relfilenumber is reused.  It's also more efficient than
                     * discovering that relations don't exist on disk yet with
                     * ENOENT errors.
                     */
                    XLogPrefetcherAddFilter(prefetcher, rlocator, 0, record->lsn);
 
#ifdef XLOGPREFETCHER_DEBUG_LEVEL
                    elog(XLOGPREFETCHER_DEBUG_LEVEL,
                         "suppressing prefetch in database %u until %X/%X is replayed due to raw file copy",
                         rlocator.dbOid,
                         LSN_FORMAT_ARGS(record->lsn));
#endif
                }
            }
            else if (rmid == RM_SMGR_ID)
            {
                if (record_type == XLOG_SMGR_CREATE)
                {
                    xl_smgr_create *xlrec = (xl_smgr_create *)
                        record->main_data;
 
                    if (xlrec->forkNum == MAIN_FORKNUM)
                    {
                        /*
                         * Don't prefetch anything for this whole relation
                         * until it has been created.  Otherwise we might
                         * confuse the blocks of different generations, if a
                         * relfilenumber is reused.  This also avoids the need
                         * to discover the problem via extra syscalls that
                         * report ENOENT.
                         */
                        XLogPrefetcherAddFilter(prefetcher, xlrec->rlocator, 0,
                                                record->lsn);
 
#ifdef XLOGPREFETCHER_DEBUG_LEVEL
                        elog(XLOGPREFETCHER_DEBUG_LEVEL,
                             "suppressing prefetch in relation %u/%u/%u until %X/%X is replayed, which creates the relation",
                             xlrec->rlocator.spcOid,
                             xlrec->rlocator.dbOid,
                             xlrec->rlocator.relNumber,
                             LSN_FORMAT_ARGS(record->lsn));
#endif
                    }
                }
                else if (record_type == XLOG_SMGR_TRUNCATE)
                {
                    xl_smgr_truncate *xlrec = (xl_smgr_truncate *)
                        record->main_data;
 
                    /*
                     * Don't consider prefetching anything in the truncated
                     * range until the truncation has been performed.
                     */
                    XLogPrefetcherAddFilter(prefetcher, xlrec->rlocator,
                                            xlrec->blkno,
                                            record->lsn);
 
#ifdef XLOGPREFETCHER_DEBUG_LEVEL
                    elog(XLOGPREFETCHER_DEBUG_LEVEL,
                         "suppressing prefetch in relation %u/%u/%u from block %u until %X/%X is replayed, which truncates the relation",
                         xlrec->rlocator.spcOid,
                         xlrec->rlocator.dbOid,
                         xlrec->rlocator.relNumber,
                         xlrec->blkno,
                         LSN_FORMAT_ARGS(record->lsn));
#endif
                }
            }
        }
 
        /* Scan the block references, starting where we left off last time. */
        while (prefetcher->next_block_id <= record->max_block_id)
        {
            int         block_id = prefetcher->next_block_id++;
            DecodedBkpBlock *block = &record->blocks[block_id];
            SMgrRelation reln;
            PrefetchBufferResult result;
 
            if (!block->in_use)
                continue;
 
            Assert(!BufferIsValid(block->prefetch_buffer));
 
            /*
             * Record the LSN of this record.  When it's replayed,
             * LsnReadQueue will consider any IOs submitted for earlier LSNs
             * to be finished.
             */
            *lsn = record->lsn;
 
            /* We don't try to prefetch anything but the main fork for now. */
            if (block->forknum != MAIN_FORKNUM)
            {
                return LRQ_NEXT_NO_IO;
            }
 
            /*
             * If there is a full page image attached, we won't be reading the
             * page, so don't bother trying to prefetch.
             */
            if (block->has_image)
            {
                XLogPrefetchIncrement(&SharedStats->skip_fpw);
                return LRQ_NEXT_NO_IO;
            }
 
            /* There is no point in reading a page that will be zeroed. */
            if (block->flags & BKPBLOCK_WILL_INIT)
            {
                XLogPrefetchIncrement(&SharedStats->skip_init);
                return LRQ_NEXT_NO_IO;
            }
 
            /* Should we skip prefetching this block due to a filter? */
            if (XLogPrefetcherIsFiltered(prefetcher, block->rlocator, block->blkno))
            {
                XLogPrefetchIncrement(&SharedStats->skip_new);
                return LRQ_NEXT_NO_IO;
            }
 
            /* There is no point in repeatedly prefetching the same block. */
            for (int i = 0; i < XLOGPREFETCHER_SEQ_WINDOW_SIZE; ++i)
            {
                if (block->blkno == prefetcher->recent_block[i] &&
                    RelFileLocatorEquals(block->rlocator, prefetcher->recent_rlocator[i]))
                {
                    /*
                     * XXX If we also remembered where it was, we could set
                     * recent_buffer so that recovery could skip smgropen()
                     * and a buffer table lookup.
                     */
                    XLogPrefetchIncrement(&SharedStats->skip_rep);
                    return LRQ_NEXT_NO_IO;
                }
            }
            prefetcher->recent_rlocator[prefetcher->recent_idx] = block->rlocator;
            prefetcher->recent_block[prefetcher->recent_idx] = block->blkno;
            prefetcher->recent_idx =
                (prefetcher->recent_idx + 1) % XLOGPREFETCHER_SEQ_WINDOW_SIZE;
 
            /*
             * We could try to have a fast path for repeated references to the
             * same relation (with some scheme to handle invalidations
             * safely), but for now we'll call smgropen() every time.
             */
            reln = smgropen(block->rlocator, InvalidBackendId);
 
            /*
             * If the relation file doesn't exist on disk, for example because
             * we're replaying after a crash and the file will be created and
             * then unlinked by WAL that hasn't been replayed yet, suppress
             * further prefetching in the relation until this record is
             * replayed.
             */
            if (!smgrexists(reln, MAIN_FORKNUM))
            {
#ifdef XLOGPREFETCHER_DEBUG_LEVEL
                elog(XLOGPREFETCHER_DEBUG_LEVEL,
                     "suppressing all prefetch in relation %u/%u/%u until %X/%X is replayed, because the relation does not exist on disk",
                     reln->smgr_rlocator.locator.spcOid,
                     reln->smgr_rlocator.locator.dbOid,
                     reln->smgr_rlocator.locator.relNumber,
                     LSN_FORMAT_ARGS(record->lsn));
#endif
                XLogPrefetcherAddFilter(prefetcher, block->rlocator, 0,
                                        record->lsn);
                XLogPrefetchIncrement(&SharedStats->skip_new);
                return LRQ_NEXT_NO_IO;
            }
 
            /*
             * If the relation isn't big enough to contain the referenced
             * block yet, suppress prefetching of this block and higher until
             * this record is replayed.
             */
            if (block->blkno >= smgrnblocks(reln, block->forknum))
            {
#ifdef XLOGPREFETCHER_DEBUG_LEVEL
                elog(XLOGPREFETCHER_DEBUG_LEVEL,
                     "suppressing prefetch in relation %u/%u/%u from block %u until %X/%X is replayed, because the relation is too small",
                     reln->smgr_rlocator.locator.spcOid,
                     reln->smgr_rlocator.locator.dbOid,
                     reln->smgr_rlocator.locator.relNumber,
                     block->blkno,
                     LSN_FORMAT_ARGS(record->lsn));
#endif
                XLogPrefetcherAddFilter(prefetcher, block->rlocator, block->blkno,
                                        record->lsn);
                XLogPrefetchIncrement(&SharedStats->skip_new);
                return LRQ_NEXT_NO_IO;
            }
 
            /* Try to initiate prefetching. */
            result = PrefetchSharedBuffer(reln, block->forknum, block->blkno);
            if (BufferIsValid(result.recent_buffer))
            {
                /* Cache hit, nothing to do. */
                XLogPrefetchIncrement(&SharedStats->hit);
                block->prefetch_buffer = result.recent_buffer;
                return LRQ_NEXT_NO_IO;
            }
            else if (result.initiated_io)
            {
                /* Cache miss, I/O (presumably) started. */
                XLogPrefetchIncrement(&SharedStats->prefetch);
                block->prefetch_buffer = InvalidBuffer;
                return LRQ_NEXT_IO;
            }
            else if ((io_direct_flags & IO_DIRECT_DATA) == 0)
            {
                /*
                 * This shouldn't be possible, because we already determined
                 * that the relation exists on disk and is big enough.
                 * Something is wrong with the cache invalidation for
                 * smgrexists(), smgrnblocks(), or the file was unlinked or
                 * truncated beneath our feet?
                 */
                elog(ERROR,
                     "could not prefetch relation %u/%u/%u block %u",
                     reln->smgr_rlocator.locator.spcOid,
                     reln->smgr_rlocator.locator.dbOid,
                     reln->smgr_rlocator.locator.relNumber,
                     block->blkno);
            }
        }
 
        /*
         * Several callsites need to be able to read exactly one record
         * without any internal readahead.  Examples: xlog.c reading
         * checkpoint records with emode set to PANIC, which might otherwise
         * cause XLogPageRead() to panic on some future page, and xlog.c
         * determining where to start writing WAL next, which depends on the
         * contents of the reader's internal buffer after reading one record.
         * Therefore, don't even think about prefetching until the first
         * record after XLogPrefetcherBeginRead() has been consumed.
         */
        if (prefetcher->reader->decode_queue_tail &&
            prefetcher->reader->decode_queue_tail->lsn == prefetcher->begin_ptr)
            return LRQ_NEXT_AGAIN;
 
        /* Advance to the next record. */
        prefetcher->record = NULL;
    }
    pg_unreachable();
}

References Assert(), XLogPrefetcher::begin_ptr, BKPBLOCK_WILL_INIT, DecodedBkpBlock::blkno, xl_smgr_truncate::blkno, BufferIsValid(), xl_dbase_create_file_copy_rec::db_id, RelFileLocator::dbOid, XLogReaderState::decode_queue_tail, elog(), ERROR, DecodedBkpBlock::flags, DecodedBkpBlock::forknum, xl_smgr_create::forkNum, DecodedBkpBlock::has_image, DecodedXLogRecord::header, XLogPrefetchStats::hit, i, if(), DecodedBkpBlock::in_use, PrefetchBufferResult::initiated_io, InvalidBackendId, InvalidBuffer, InvalidOid, InvalidRelFileNumber, InvalidXLogRecPtr, IO_DIRECT_DATA, io_direct_flags, RelFileLocatorBackend::locator, LRQ_NEXT_AGAIN, LRQ_NEXT_IO, LRQ_NEXT_NO_IO, DecodedXLogRecord::lsn, LSN_FORMAT_ARGS, DecodedXLogRecord::main_data, MAIN_FORKNUM, XLogPrefetcher::next_block_id, XLogPrefetcher::no_readahead_until, XLogPrefetchStats::prefetch, DecodedBkpBlock::prefetch_buffer, PrefetchSharedBuffer(), XLogPrefetcher::reader, XLogReaderState::ReadRecPtr, XLogPrefetcher::recent_block, PrefetchBufferResult::recent_buffer, XLogPrefetcher::recent_idx, XLogPrefetcher::recent_rlocator, XLogPrefetcher::record, RecoveryPrefetchEnabled, RelFileLocatorEquals, RelFileLocator::relNumber, DecodedBkpBlock::rlocator, xl_smgr_create::rlocator, xl_smgr_truncate::rlocator, SharedStats, XLogPrefetchStats::skip_fpw, XLogPrefetchStats::skip_init, XLogPrefetchStats::skip_new, XLogPrefetchStats::skip_rep, SMgrRelationData::smgr_rlocator, smgrexists(), smgrnblocks(), smgropen(), RelFileLocator::spcOid, XLogRecord::xl_info, XLogRecord::xl_rmid, XLOG_CHECKPOINT_SHUTDOWN, XLOG_DBASE_CREATE_FILE_COPY, XLOG_END_OF_RECOVERY, XLOG_SMGR_CREATE, XLOG_SMGR_TRUNCATE, XLOGPREFETCHER_SEQ_WINDOW_SIZE, XLogPrefetcherAddFilter(), XLogPrefetcherIsFiltered(), XLogPrefetchIncrement(), XLogReadAhead(), XLogReaderHasQueuedRecordOrError(), and XLR_INFO_MASK.

Referenced by XLogPrefetcherReadRecord().

XLogPrefetcherReadRecord()

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

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

XLogPrefetchIncrement()

static void XLogPrefetchIncrement ( pg_atomic_uint64 *  counter )

inlinestatic

Definition at line 355 of file xlogprefetcher.c.

{
    Assert(AmStartupProcess() || !IsUnderPostmaster);
    pg_atomic_write_u64(counter, pg_atomic_read_u64(counter) + 1);
}

References AmStartupProcess, Assert(), IsUnderPostmaster, pg_atomic_read_u64(), and pg_atomic_write_u64().

Referenced by XLogPrefetcherNextBlock().

XLogPrefetchReconfigure()

void XLogPrefetchReconfigure

(

void

)

Definition at line 344 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 307 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 319 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 CreateSharedMemoryAndSemaphores().

XLogPrefetchShmemSize()

size_t XLogPrefetchShmemSize

(

void

)

Definition at line 298 of file xlogprefetcher.c.

{
    return sizeof(XLogPrefetchStats);
}

Referenced by CalculateShmemSize().

Variable Documentation

recovery_prefetch

int recovery_prefetch = RECOVERY_PREFETCH_TRY

Definition at line 72 of file xlogprefetcher.c.

Referenced by assign_recovery_prefetch().

SharedStats

XLogPrefetchStats* SharedStats

static

Definition at line 203 of file xlogprefetcher.c.

Referenced by pg_stat_get_recovery_prefetch(), XLogPrefetcherAllocate(), XLogPrefetcherComputeStats(), XLogPrefetcherNextBlock(), XLogPrefetchResetStats(), and XLogPrefetchShmemInit().

XLogPrefetchReconfigureCount

int XLogPrefetchReconfigureCount = 0

static

Definition at line 82 of file xlogprefetcher.c.

Referenced by XLogPrefetcherAllocate(), XLogPrefetcherReadRecord(), and XLogPrefetchReconfigure().