PostgreSQL Source Code  git master
xlogprefetcher.h File Reference
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)
 
XLogPrefetcherXLogPrefetcherAllocate (XLogReaderState *reader)
 
void XLogPrefetcherFree (XLogPrefetcher *prefetcher)
 
XLogReaderStateXLogPrefetcherGetReader (XLogPrefetcher *prefetcher)
 
void XLogPrefetcherBeginRead (XLogPrefetcher *prefetcher, XLogRecPtr recPtr)
 
XLogRecordXLogPrefetcherReadRecord (XLogPrefetcher *prefetcher, char **errmsg)
 
void XLogPrefetcherComputeStats (XLogPrefetcher *prefetcher)
 

Variables

PGDLLIMPORT int recovery_prefetch
 

Typedef Documentation

◆ XLogPrefetcher

Definition at line 21 of file xlogprefetcher.h.

Enumeration Type Documentation

◆ RecoveryPrefetchValue

Enumerator
RECOVERY_PREFETCH_OFF 
RECOVERY_PREFETCH_ON 
RECOVERY_PREFETCH_TRY 

Definition at line 24 of file xlogprefetcher.h.

25 {
RecoveryPrefetchValue
@ RECOVERY_PREFETCH_OFF
@ RECOVERY_PREFETCH_ON
@ RECOVERY_PREFETCH_TRY

Function Documentation

◆ XLogPrefetcherAllocate()

XLogPrefetcher* XLogPrefetcherAllocate ( XLogReaderState reader)

Definition at line 366 of file xlogprefetcher.c.

367 {
368  XLogPrefetcher *prefetcher;
369  static HASHCTL hash_table_ctl = {
370  .keysize = sizeof(RelFileLocator),
371  .entrysize = sizeof(XLogPrefetcherFilter)
372  };
373 
374  prefetcher = palloc0(sizeof(XLogPrefetcher));
375 
376  prefetcher->reader = reader;
377  prefetcher->filter_table = hash_create("XLogPrefetcherFilterTable", 1024,
378  &hash_table_ctl,
380  dlist_init(&prefetcher->filter_queue);
381 
384  SharedStats->io_depth = 0;
385 
386  /* First usage will cause streaming_read to be allocated. */
388 
389  return prefetcher;
390 }
HTAB * hash_create(const char *tabname, long nelem, const HASHCTL *info, int flags)
Definition: dynahash.c:350
#define HASH_ELEM
Definition: hsearch.h:95
#define HASH_BLOBS
Definition: hsearch.h:97
static void dlist_init(dlist_head *head)
Definition: ilist.h:314
void * palloc0(Size size)
Definition: mcxt.c:1257
struct RelFileLocator RelFileLocator
Size keysize
Definition: hsearch.h:75
dlist_head filter_queue
XLogReaderState * reader
static int XLogPrefetchReconfigureCount
static XLogPrefetchStats * SharedStats

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.

969 {
970  /* This will forget about any in-flight IO. */
971  prefetcher->reconfigure_count--;
972 
973  /* Book-keeping to avoid readahead on first read. */
974  prefetcher->begin_ptr = recPtr;
975 
976  prefetcher->no_readahead_until = 0;
977 
978  /* This will forget about any queued up records in the decoder. */
979  XLogBeginRead(prefetcher->reader, recPtr);
980 }
XLogRecPtr no_readahead_until
XLogRecPtr begin_ptr
void XLogBeginRead(XLogReaderState *state, XLogRecPtr RecPtr)
Definition: xlogreader.c:233

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 416 of file xlogprefetcher.c.

417 {
418  uint32 io_depth;
419  uint32 completed;
420  int64 wal_distance;
421 
422 
423  /* How far ahead of replay are we now? */
424  if (prefetcher->reader->decode_queue_tail)
425  {
426  wal_distance =
427  prefetcher->reader->decode_queue_tail->lsn -
428  prefetcher->reader->decode_queue_head->lsn;
429  }
430  else
431  {
432  wal_distance = 0;
433  }
434 
435  /* How many IOs are currently in flight and completed? */
436  io_depth = lrq_inflight(prefetcher->streaming_read);
437  completed = lrq_completed(prefetcher->streaming_read);
438 
439  /* Update the instantaneous stats visible in pg_stat_recovery_prefetch. */
440  SharedStats->io_depth = io_depth;
441  SharedStats->block_distance = io_depth + completed;
442  SharedStats->wal_distance = wal_distance;
443 
444  prefetcher->next_stats_shm_lsn =
446 }
unsigned int uint32
Definition: c.h:495
XLogRecPtr lsn
Definition: xlogreader.h:164
LsnReadQueue * streaming_read
XLogRecPtr next_stats_shm_lsn
DecodedXLogRecord * decode_queue_head
Definition: xlogreader.h:260
XLogRecPtr ReadRecPtr
Definition: xlogreader.h:206
DecodedXLogRecord * decode_queue_tail
Definition: xlogreader.h:261
static uint32 lrq_completed(LsnReadQueue *lrq)
static uint32 lrq_inflight(LsnReadQueue *lrq)
#define 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.

397 {
398  lrq_free(prefetcher->streaming_read);
399  hash_destroy(prefetcher->filter_table);
400  pfree(prefetcher);
401 }
void hash_destroy(HTAB *hashp)
Definition: dynahash.c:863
void pfree(void *pointer)
Definition: mcxt.c:1456
static void lrq_free(LsnReadQueue *lrq)

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.

408 {
409  return prefetcher->reader;
410 }

References XLogPrefetcher::reader.

Referenced by ReadRecord().

◆ XLogPrefetcherReadRecord()

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

Definition at line 987 of file xlogprefetcher.c.

988 {
989  DecodedXLogRecord *record;
990  XLogRecPtr replayed_up_to;
991 
992  /*
993  * See if it's time to reset the prefetching machinery, because a relevant
994  * GUC was changed.
995  */
997  {
998  uint32 max_distance;
999  uint32 max_inflight;
1000 
1001  if (prefetcher->streaming_read)
1002  lrq_free(prefetcher->streaming_read);
1003 
1005  {
1007  max_inflight = maintenance_io_concurrency;
1008  max_distance = max_inflight * XLOGPREFETCHER_DISTANCE_MULTIPLIER;
1009  }
1010  else
1011  {
1012  max_inflight = 1;
1013  max_distance = 1;
1014  }
1015 
1016  prefetcher->streaming_read = lrq_alloc(max_distance,
1017  max_inflight,
1018  (uintptr_t) prefetcher,
1020 
1022  }
1023 
1024  /*
1025  * Release last returned record, if there is one, as it's now been
1026  * replayed.
1027  */
1028  replayed_up_to = XLogReleasePreviousRecord(prefetcher->reader);
1029 
1030  /*
1031  * Can we drop any filters yet? If we were waiting for a relation to be
1032  * created or extended, it is now OK to access blocks in the covered
1033  * range.
1034  */
1035  XLogPrefetcherCompleteFilters(prefetcher, replayed_up_to);
1036 
1037  /*
1038  * All IO initiated by earlier WAL is now completed. This might trigger
1039  * further prefetching.
1040  */
1041  lrq_complete_lsn(prefetcher->streaming_read, replayed_up_to);
1042 
1043  /*
1044  * If there's nothing queued yet, then start prefetching to cause at least
1045  * one record to be queued.
1046  */
1047  if (!XLogReaderHasQueuedRecordOrError(prefetcher->reader))
1048  {
1049  Assert(lrq_inflight(prefetcher->streaming_read) == 0);
1050  Assert(lrq_completed(prefetcher->streaming_read) == 0);
1051  lrq_prefetch(prefetcher->streaming_read);
1052  }
1053 
1054  /* Read the next record. */
1055  record = XLogNextRecord(prefetcher->reader, errmsg);
1056  if (!record)
1057  return NULL;
1058 
1059  /*
1060  * The record we just got is the "current" one, for the benefit of the
1061  * XLogRecXXX() macros.
1062  */
1063  Assert(record == prefetcher->reader->record);
1064 
1065  /*
1066  * If maintenance_io_concurrency is set very low, we might have started
1067  * prefetching some but not all of the blocks referenced in the record
1068  * we're about to return. Forget about the rest of the blocks in this
1069  * record by dropping the prefetcher's reference to it.
1070  */
1071  if (record == prefetcher->record)
1072  prefetcher->record = NULL;
1073 
1074  /*
1075  * See if it's time to compute some statistics, because enough WAL has
1076  * been processed.
1077  */
1078  if (unlikely(record->lsn >= prefetcher->next_stats_shm_lsn))
1079  XLogPrefetcherComputeStats(prefetcher);
1080 
1081  Assert(record == prefetcher->reader->record);
1082 
1083  return &record->header;
1084 }
int maintenance_io_concurrency
Definition: bufmgr.c:154
#define unlikely(x)
Definition: c.h:300
int errmsg(const char *fmt,...)
Definition: elog.c:1069
Assert(fmt[strlen(fmt) - 1] !='\n')
XLogRecord header
Definition: xlogreader.h:166
DecodedXLogRecord * record
DecodedXLogRecord * record
Definition: xlogreader.h:236
uint64 XLogRecPtr
Definition: xlogdefs.h:21
void XLogPrefetcherComputeStats(XLogPrefetcher *prefetcher)
#define RecoveryPrefetchEnabled()
static LsnReadQueue * lrq_alloc(uint32 max_distance, uint32 max_inflight, uintptr_t lrq_private, LsnReadQueueNextFun next)
static void XLogPrefetcherCompleteFilters(XLogPrefetcher *prefetcher, XLogRecPtr replaying_lsn)
static void lrq_prefetch(LsnReadQueue *lrq)
static LsnReadQueueNextStatus XLogPrefetcherNextBlock(uintptr_t pgsr_private, XLogRecPtr *lsn)
static void lrq_complete_lsn(LsnReadQueue *lrq, XLogRecPtr lsn)
#define XLOGPREFETCHER_DISTANCE_MULTIPLIER
DecodedXLogRecord * XLogNextRecord(XLogReaderState *state, char **errormsg)
Definition: xlogreader.c:327
XLogRecPtr XLogReleasePreviousRecord(XLogReaderState *state)
Definition: xlogreader.c:251
static bool XLogReaderHasQueuedRecordOrError(XLogReaderState *state)
Definition: xlogreader.h:325

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  )

◆ XLogPrefetchResetStats()

void XLogPrefetchResetStats ( void  )

Definition at line 307 of file xlogprefetcher.c.

308 {
316 }
static void pg_atomic_write_u64(volatile pg_atomic_uint64 *ptr, uint64 val)
Definition: atomics.h:433
TimestampTz GetCurrentTimestamp(void)
Definition: timestamp.c:1649
pg_atomic_uint64 skip_fpw
pg_atomic_uint64 skip_init
pg_atomic_uint64 reset_time
pg_atomic_uint64 hit
pg_atomic_uint64 prefetch
pg_atomic_uint64 skip_rep
pg_atomic_uint64 skip_new

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.

320 {
321  bool found;
322 
324  ShmemInitStruct("XLogPrefetchStats",
325  sizeof(XLogPrefetchStats),
326  &found);
327 
328  if (!found)
329  {
337  }
338 }
static void pg_atomic_init_u64(volatile pg_atomic_uint64 *ptr, uint64 val)
Definition: atomics.h:410
void * ShmemInitStruct(const char *name, Size size, bool *foundPtr)
Definition: shmem.c:396

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 298 of file xlogprefetcher.c.

299 {
300  return sizeof(XLogPrefetchStats);
301 }
struct XLogPrefetchStats XLogPrefetchStats

Referenced by CalculateShmemSize().

Variable Documentation

◆ recovery_prefetch

PGDLLIMPORT int recovery_prefetch
extern

Definition at line 72 of file xlogprefetcher.c.

Referenced by assign_recovery_prefetch().