/*------------------------------------------------------------------------- * * xlogprefetcher.c * Prefetching support for recovery. * * Portions Copyright (c) 2022-2025, PostgreSQL Global Development Group * Portions Copyright (c) 1994, Regents of the University of California * * * IDENTIFICATION * src/backend/access/transam/xlogprefetcher.c * * This module provides a drop-in replacement for an XLogReader that tries to * minimize I/O stalls by looking ahead in the WAL. If blocks that will be * accessed in the near future are not already in the buffer pool, it initiates * I/Os that might complete before the caller eventually needs the data. When * referenced blocks are found in the buffer pool already, the buffer is * recorded in the decoded record so that XLogReadBufferForRedo() can try to * avoid a second buffer mapping table lookup. * * Currently, only the main fork is considered for prefetching. Currently, * prefetching is only effective on systems where PrefetchBuffer() does * something useful (mainly Linux). * *------------------------------------------------------------------------- */ #include "postgres.h" #include "access/xlogprefetcher.h" #include "access/xlogreader.h" #include "catalog/pg_control.h" #include "catalog/storage_xlog.h" #include "commands/dbcommands_xlog.h" #include "funcapi.h" #include "miscadmin.h" #include "port/atomics.h" #include "storage/bufmgr.h" #include "storage/shmem.h" #include "storage/smgr.h" #include "utils/fmgrprotos.h" #include "utils/guc_hooks.h" #include "utils/hsearch.h" #include "utils/timestamp.h" /* * Every time we process this much WAL, we'll update the values in * pg_stat_recovery_prefetch. */ #define XLOGPREFETCHER_STATS_DISTANCE BLCKSZ /* * To detect repeated access to the same block and skip useless extra system * calls, we remember a small window of recently prefetched blocks. */ #define XLOGPREFETCHER_SEQ_WINDOW_SIZE 4 /* * When maintenance_io_concurrency is not saturated, we're prepared to look * ahead up to N times that number of block references. */ #define XLOGPREFETCHER_DISTANCE_MULTIPLIER 4 /* Define to log internal debugging messages. */ /* #define XLOGPREFETCHER_DEBUG_LEVEL LOG */ /* GUCs */ int recovery_prefetch = RECOVERY_PREFETCH_TRY; #ifdef USE_PREFETCH #define RecoveryPrefetchEnabled() \ (recovery_prefetch != RECOVERY_PREFETCH_OFF && \ maintenance_io_concurrency > 0) #else #define RecoveryPrefetchEnabled() false #endif static int XLogPrefetchReconfigureCount = 0; /* * Enum used to report whether an IO should be started. */ typedef enum { LRQ_NEXT_NO_IO, LRQ_NEXT_IO, LRQ_NEXT_AGAIN, } LsnReadQueueNextStatus; /* * Type of callback that can decide which block to prefetch next. For now * there is only one. */ typedef LsnReadQueueNextStatus (*LsnReadQueueNextFun) (uintptr_t lrq_private, XLogRecPtr *lsn); /* * A simple circular queue of LSNs, using to control the number of * (potentially) inflight IOs. This stands in for a later more general IO * control mechanism, which is why it has the apparently unnecessary * indirection through a function pointer. */ typedef struct LsnReadQueue { LsnReadQueueNextFun next; uintptr_t lrq_private; uint32 max_inflight; uint32 inflight; uint32 completed; uint32 head; uint32 tail; uint32 size; struct { bool io; XLogRecPtr lsn; } queue[FLEXIBLE_ARRAY_MEMBER]; } LsnReadQueue; /* * A prefetcher. This is a mechanism that wraps an XLogReader, prefetching * blocks that will be soon be referenced, to try to avoid IO stalls. */ struct XLogPrefetcher { /* WAL reader and current reading state. */ XLogReaderState *reader; DecodedXLogRecord *record; int next_block_id; /* When to publish stats. */ XLogRecPtr next_stats_shm_lsn; /* Book-keeping to avoid accessing blocks that don't exist yet. */ HTAB *filter_table; dlist_head filter_queue; /* Book-keeping to avoid repeat prefetches. */ RelFileLocator recent_rlocator[XLOGPREFETCHER_SEQ_WINDOW_SIZE]; BlockNumber recent_block[XLOGPREFETCHER_SEQ_WINDOW_SIZE]; int recent_idx; /* Book-keeping to disable prefetching temporarily. */ XLogRecPtr no_readahead_until; /* IO depth manager. */ LsnReadQueue *streaming_read; XLogRecPtr begin_ptr; int reconfigure_count; }; /* * A temporary filter used to track block ranges that haven't been created * yet, whole relations that haven't been created yet, and whole relations * that (we assume) have already been dropped, or will be created by bulk WAL * operators. */ typedef struct XLogPrefetcherFilter { RelFileLocator rlocator; XLogRecPtr filter_until_replayed; BlockNumber filter_from_block; dlist_node link; } XLogPrefetcherFilter; /* * Counters exposed in shared memory for pg_stat_recovery_prefetch. */ typedef struct XLogPrefetchStats { pg_atomic_uint64 reset_time; /* Time of last reset. */ pg_atomic_uint64 prefetch; /* Prefetches initiated. */ pg_atomic_uint64 hit; /* Blocks already in cache. */ pg_atomic_uint64 skip_init; /* Zero-inited blocks skipped. */ pg_atomic_uint64 skip_new; /* New/missing blocks filtered. */ pg_atomic_uint64 skip_fpw; /* FPWs skipped. */ pg_atomic_uint64 skip_rep; /* Repeat accesses skipped. */ /* Dynamic values */ int wal_distance; /* Number of WAL bytes ahead. */ int block_distance; /* Number of block references ahead. */ int io_depth; /* Number of I/Os in progress. */ } XLogPrefetchStats; static inline void XLogPrefetcherAddFilter(XLogPrefetcher *prefetcher, RelFileLocator rlocator, BlockNumber blockno, XLogRecPtr lsn); static inline bool XLogPrefetcherIsFiltered(XLogPrefetcher *prefetcher, RelFileLocator rlocator, BlockNumber blockno); static inline void XLogPrefetcherCompleteFilters(XLogPrefetcher *prefetcher, XLogRecPtr replaying_lsn); static LsnReadQueueNextStatus XLogPrefetcherNextBlock(uintptr_t pgsr_private, XLogRecPtr *lsn); static XLogPrefetchStats *SharedStats; static inline LsnReadQueue * lrq_alloc(uint32 max_distance, uint32 max_inflight, uintptr_t lrq_private, LsnReadQueueNextFun next) { 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; } static inline void lrq_free(LsnReadQueue *lrq) { pfree(lrq); } static inline uint32 lrq_inflight(LsnReadQueue *lrq) { return lrq->inflight; } static inline uint32 lrq_completed(LsnReadQueue *lrq) { return lrq->completed; } static inline void lrq_prefetch(LsnReadQueue *lrq) { /* 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; } } static inline void lrq_complete_lsn(LsnReadQueue *lrq, XLogRecPtr lsn) { /* * 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); } size_t XLogPrefetchShmemSize(void) { return sizeof(XLogPrefetchStats); } /* * Reset all counters to zero. */ void XLogPrefetchResetStats(void) { 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); } void XLogPrefetchShmemInit(void) { 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); } } /* * Called when any GUC is changed that affects prefetching. */ void XLogPrefetchReconfigure(void) { XLogPrefetchReconfigureCount++; } /* * Increment a counter in shared memory. This is equivalent to *counter++ on a * plain uint64 without any memory barrier or locking, except on platforms * where readers can't read uint64 without possibly observing a torn value. */ static inline void XLogPrefetchIncrement(pg_atomic_uint64 *counter) { Assert(AmStartupProcess() || !IsUnderPostmaster); pg_atomic_write_u64(counter, pg_atomic_read_u64(counter) + 1); } /* * Create a prefetcher that is ready to begin prefetching blocks referenced by * WAL records. */ XLogPrefetcher * XLogPrefetcherAllocate(XLogReaderState *reader) { XLogPrefetcher *prefetcher; HASHCTL ctl; prefetcher = palloc0(sizeof(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; } /* * Destroy a prefetcher and release all resources. */ void XLogPrefetcherFree(XLogPrefetcher *prefetcher) { lrq_free(prefetcher->streaming_read); hash_destroy(prefetcher->filter_table); pfree(prefetcher); } /* * Provide access to the reader. */ XLogReaderState * XLogPrefetcherGetReader(XLogPrefetcher *prefetcher) { return prefetcher->reader; } /* * Update the statistics visible in the pg_stat_recovery_prefetch view. */ void XLogPrefetcherComputeStats(XLogPrefetcher *prefetcher) { 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; } /* * A callback that examines the next block reference in the WAL, and possibly * starts an IO so that a later read will be fast. * * Returns LRQ_NEXT_AGAIN if no more WAL data is available yet. * * Returns LRQ_NEXT_IO if the next block reference is for a main fork block * that isn't in the buffer pool, and the kernel has been asked to start * reading it to make a future read system call faster. An LSN is written to * *lsn, and the I/O will be considered to have completed once that LSN is * replayed. * * Returns LRQ_NEXT_NO_IO if we examined the next block reference and found * that it was already in the buffer pool, or we decided for various reasons * not to prefetch. */ static LsnReadQueueNextStatus XLogPrefetcherNextBlock(uintptr_t pgsr_private, XLogRecPtr *lsn) { 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, INVALID_PROC_NUMBER); /* * 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(); } /* * Expose statistics about recovery prefetching. */ Datum pg_stat_get_recovery_prefetch(PG_FUNCTION_ARGS) { #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; } /* * Don't prefetch any blocks >= 'blockno' from a given 'rlocator', until 'lsn' * has been replayed. */ static inline void XLogPrefetcherAddFilter(XLogPrefetcher *prefetcher, RelFileLocator rlocator, BlockNumber blockno, XLogRecPtr lsn) { 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); } } /* * Have we replayed any records that caused us to begin filtering a block * range? That means that relations should have been created, extended or * dropped as required, so we can stop filtering out accesses to a given * relfilenumber. */ static inline void XLogPrefetcherCompleteFilters(XLogPrefetcher *prefetcher, XLogRecPtr replaying_lsn) { 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); } } /* * Check if a given block should be skipped due to a filter. */ static inline bool XLogPrefetcherIsFiltered(XLogPrefetcher *prefetcher, RelFileLocator rlocator, BlockNumber blockno) { /* * 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; } /* * A wrapper for XLogBeginRead() that also resets the prefetcher. */ void XLogPrefetcherBeginRead(XLogPrefetcher *prefetcher, XLogRecPtr recPtr) { /* 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); } /* * A wrapper for XLogReadRecord() that provides the same interface, but also * tries to initiate I/O for blocks referenced in future WAL records. */ XLogRecord * XLogPrefetcherReadRecord(XLogPrefetcher *prefetcher, char **errmsg) { 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; } bool check_recovery_prefetch(int *new_value, void **extra, GucSource source) { #ifndef USE_PREFETCH if (*new_value == RECOVERY_PREFETCH_ON) { GUC_check_errdetail("\"recovery_prefetch\" is not supported on platforms that lack support for issuing read-ahead advice."); return false; } #endif return true; } void assign_recovery_prefetch(int new_value, void *extra) { /* Reconfigure prefetching, because a setting it depends on changed. */ recovery_prefetch = new_value; if (AmStartupProcess()) XLogPrefetchReconfigure(); }