xlogprefetcher.h File Reference

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

Go to the source code of 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 32 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 )

extern

Definition at line 362 of file xlogprefetcher.c.

{
    XLogPrefetcher *prefetcher;
    HASHCTL     ctl;
 
    prefetcher = palloc0_object(XLogPrefetcher);
    prefetcher->reader = reader;
 
    ctl.keysize = sizeof(RelFileLocator);
    ctl.entrysize = sizeof(XLogPrefetcherFilter);
    prefetcher->filter_table = hash_create("XLogPrefetcherFilterTable", 1024,
                                           &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, ctl, dlist_init(), fb(), HASH_BLOBS, hash_create(), HASH_ELEM, XLogPrefetchStats::io_depth, palloc0_object, SharedStats, XLogPrefetchStats::wal_distance, and XLogPrefetchReconfigureCount.

Referenced by InitWalRecovery().

XLogPrefetcherBeginRead()

void XLogPrefetcherBeginRead ( XLogPrefetcher *  prefetcher, XLogRecPtr  recPtr  )

extern

Definition at line 962 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 = InvalidXLogRecPtr;
 
    /* This will forget about any queued up records in the decoder. */
    XLogBeginRead(prefetcher->reader, recPtr);
}

References fb(), InvalidXLogRecPtr, and XLogBeginRead().

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

XLogPrefetcherComputeStats()

void XLogPrefetcherComputeStats ( XLogPrefetcher *  prefetcher )

extern

Definition at line 410 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_tail, fb(), XLogPrefetchStats::io_depth, lrq_completed(), lrq_inflight(), DecodedXLogRecord::lsn, SharedStats, XLogPrefetchStats::wal_distance, and XLOGPREFETCHER_STATS_DISTANCE.

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

XLogPrefetcherFree()

void XLogPrefetcherFree ( XLogPrefetcher *  prefetcher )

extern

Definition at line 390 of file xlogprefetcher.c.

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

References fb(), hash_destroy(), lrq_free(), and pfree().

Referenced by ShutdownWalRecovery().

XLogPrefetcherGetReader()

XLogReaderState * XLogPrefetcherGetReader ( XLogPrefetcher *  prefetcher )

extern

Definition at line 401 of file xlogprefetcher.c.

{
    return prefetcher->reader;
}

References fb().

Referenced by ReadRecord().

XLogPrefetcherReadRecord()

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

extern

Definition at line 981 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(), fb(), DecodedXLogRecord::header, lrq_alloc(), lrq_complete_lsn(), lrq_completed(), lrq_free(), lrq_inflight(), lrq_prefetch(), DecodedXLogRecord::lsn, maintenance_io_concurrency, RecoveryPrefetchEnabled, unlikely, XLogNextRecord(), XLOGPREFETCHER_DISTANCE_MULTIPLIER, XLogPrefetcherCompleteFilters(), XLogPrefetcherComputeStats(), XLogPrefetcherNextBlock(), XLogPrefetchReconfigureCount, XLogReaderHasQueuedRecordOrError(), and XLogReleasePreviousRecord().

Referenced by ReadRecord().

XLogPrefetchReconfigure()

void XLogPrefetchReconfigure ( void  )

extern

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  )

extern

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  )

extern

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  )

extern

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().