PostgreSQL 源码解读（211）- 后台进程#10（checkpointer-BufferSync）

本节介绍了checkpoint中用于刷盘的函数:BufferSync,该函数Write out all dirty buffers in the pool(把缓冲池中所有脏页持久化到物理存储中).
值得一提的是:checkpoint只会处理在checkpoint开始时的脏页(标记为BM_CHECKPOINT_NEEDED),而不会处理在checkpoint变脏的page.

一、数据结构

宏定义
checkpoints request flag bits,检查点请求标记位定义.

#define CHECKPOINT_IS_SHUTDOWN  0x0001  
#define CHECKPOINT_END_OF_RECOVERY  0x0002  
#define CHECKPOINT_IMMEDIATE  0x0004  
#define CHECKPOINT_FORCE    0x0008  
#define CHECKPOINT_FLUSH_ALL  0x0010  

#define CHECKPOINT_WAIT     0x0020  
#define CHECKPOINT_REQUESTED  0x0040  

#define CHECKPOINT_CAUSE_XLOG 0x0080  
#define CHECKPOINT_CAUSE_TIME 0x0100  

二、源码解读

BufferSync : 把缓冲池中所有脏页持久化到物理存储中.其主要逻辑如下:
1.执行相关校验,如确保调用SyncOneBuffer函数的正确性等;
2.根据checkpoint标记设置mask标记(如为XX,则unlogged buffer也会flush);
3.遍历缓存,使用BM_CHECKPOINT_NEEDED标记需要刷盘的缓存page;如无需要处理的page,则返回;
4.排序需刷盘的脏页,避免随机IO,提升性能;
5.为每一个需要刷脏页的表空间分配进度状态;
6.在单个标记的写进度上构建最小堆,并计算单个处理缓冲区占比多少;
7.如ts_heap不为空,循环处理
7.1获取buf_id
7.2调用SyncOneBuffer刷盘
7.3调用CheckpointWriteDelay,休眠以控制I/O频率
7.4释放资源,更新统计信息

static void
BufferSync(int flags)
{
  uint32    buf_state;
  int     buf_id;
  int     num_to_scan;
  int     num_spaces;
  int     num_processed;
  int     num_written;
  CkptTsStatus *per_ts_stat = NULL;
  Oid     last_tsid;
  binaryheap *ts_heap;
  int     i;
  int     mask = BM_DIRTY;
  WritebackContext wb_context;
  
  //确保可以处理在SyncOneBuffer函数中的pin page
  ResourceOwnerEnlargeBuffers(CurrentResourceOwner);
  
  //如为CHECKPOINT_IS_SHUTDOWN/CHECKPOINT_END_OF_RECOVERY/CHECKPOINT_FLUSH_ALL,
  //就算正常情况下会忽略的unlogged缓存,也会写入到磁盘上.
  if (!((flags & (CHECKPOINT_IS_SHUTDOWN | CHECKPOINT_END_OF_RECOVERY |
          CHECKPOINT_FLUSH_ALL))))
    mask |= BM_PERMANENT;
  
  num_to_scan = 0;
  for (buf_id = 0; buf_id < NBuffers; buf_id++)
  {
    BufferDesc *bufHdr = GetBufferDescriptor(buf_id);
    
    buf_state = LockBufHdr(bufHdr);
    if ((buf_state & mask) == mask)
    {
      CkptSortItem *item;
      buf_state |= BM_CHECKPOINT_NEEDED;
      item = &CkptBufferIds[num_to_scan++];
      item->buf_id = buf_id;
      item->tsId = bufHdr->tag.rnode.spcNode;
      item->relNode = bufHdr->tag.rnode.relNode;
      item->forkNum = bufHdr->tag.forkNum;
      item->blockNum = bufHdr->tag.blockNum;
    }
    UnlockBufHdr(bufHdr, buf_state);
  }
  if (num_to_scan == 0)
    return;         
  WritebackContextInit(&wb_context, &checkpoint_flush_after);
  TRACE_POSTGRESQL_BUFFER_SYNC_START(NBuffers, num_to_scan);
  
  qsort(CkptBufferIds, num_to_scan, sizeof(CkptSortItem),
      ckpt_buforder_comparator);
  num_spaces = 0;
  
  last_tsid = InvalidOid;
  for (i = 0; i < num_to_scan; i++)
  {
    CkptTsStatus *s;
    Oid     cur_tsid;
    cur_tsid = CkptBufferIds[i].tsId;
    
    if (last_tsid == InvalidOid || last_tsid != cur_tsid)
    {
      Size    sz;
      num_spaces++;
      
      sz = sizeof(CkptTsStatus) * num_spaces;
      if (per_ts_stat == NULL)
        per_ts_stat = (CkptTsStatus *) palloc(sz);
      else
        per_ts_stat = (CkptTsStatus *) repalloc(per_ts_stat, sz);
      s = &per_ts_stat[num_spaces - 1];
      memset(s, 0, sizeof(*s));
      s->tsId = cur_tsid;
      
      s->index = i;
      
      last_tsid = cur_tsid;
    }
    else
    {
      s = &per_ts_stat[num_spaces - 1];
    }
    s->num_to_scan++;
  }
  Assert(num_spaces > 0);
  
  ts_heap = binaryheap_allocate(num_spaces,
                  ts_ckpt_progress_comparator,
                  NULL);
  for (i = 0; i < num_spaces; i++)
  {
    CkptTsStatus *ts_stat = &per_ts_stat[i];
    ts_stat->progress_slice = (float8) num_to_scan / ts_stat->num_to_scan;
    binaryheap_add_unordered(ts_heap, PointerGetDatum(ts_stat));
  }
  binaryheap_build(ts_heap);
  
  num_processed = 0;
  num_written = 0;
  while (!binaryheap_empty(ts_heap))
  {
    BufferDesc *bufHdr = NULL;
    CkptTsStatus *ts_stat = (CkptTsStatus *)
    DatumGetPointer(binaryheap_first(ts_heap));
    buf_id = CkptBufferIds[ts_stat->index].buf_id;
    Assert(buf_id != -1);
    bufHdr = GetBufferDescriptor(buf_id);
    num_processed++;
    
    if (pg_atomic_read_u32(&bufHdr->state) & BM_CHECKPOINT_NEEDED)
    {
      //只处理标记为BM_CHECKPOINT_NEEDED的page
      //调用SyncOneBuffer刷盘(一次一个page)
      if (SyncOneBuffer(buf_id, false, &wb_context) & BUF_WRITTEN)
      {
        TRACE_POSTGRESQL_BUFFER_SYNC_WRITTEN(buf_id);
        BgWriterStats.m_buf_written_checkpoints++;
        num_written++;
      }
    }
    
    ts_stat->progress += ts_stat->progress_slice;
    ts_stat->num_scanned++;
    ts_stat->index++;
    
    if (ts_stat->num_scanned == ts_stat->num_to_scan)
    {
      binaryheap_remove_first(ts_heap);
    }
    else
    {
      
      binaryheap_replace_first(ts_heap, PointerGetDatum(ts_stat));
    }
    
    CheckpointWriteDelay(flags, (double) num_processed / num_to_scan);
  }
  
  IssuePendingWritebacks(&wb_context);
  pfree(per_ts_stat);
  per_ts_stat = NULL;
  binaryheap_free(ts_heap);
  
  CheckpointStats.ckpt_bufs_written += num_written;
  TRACE_POSTGRESQL_BUFFER_SYNC_DONE(NBuffers, num_written, num_to_scan);
}

三、跟踪分析

测试脚本

testdb=# update t_wal_ckpt set c2 = 'C4#'||substr(c2,4,40);
UPDATE 1
testdb=# checkpoint;

跟踪分析

(gdb) handle SIGINT print nostop pass
SIGINT is used by the debugger.
Are you sure you want to change it? (y or n) y
Signal        Stop  Print Pass to program Description
SIGINT        No  Yes Yes   Interrupt
(gdb) b CheckPointGuts
Breakpoint 1 at 0x56f0ca: file xlog.c, line 8968.
(gdb) c
Continuing.
Program received signal SIGINT, Interrupt.
Breakpoint 1, CheckPointGuts (checkPointRedo=16953420440, flags=108) at xlog.c:8968
8968    CheckPointCLOG();
(gdb) n
8969    CheckPointCommitTs();
(gdb) 
8970    CheckPointSUBTRANS();
(gdb) 
8971    CheckPointMultiXact();
(gdb) 
8972    CheckPointPredicate();
(gdb) 
8973    CheckPointRelationMap();
(gdb) 
8974    CheckPointReplicationSlots();
(gdb) 
8975    CheckPointSnapBuild();
(gdb) 
8976    CheckPointLogicalRewriteHeap();
(gdb) 
8977    CheckPointBuffers(flags); 
(gdb) step
CheckPointBuffers (flags=108) at bufmgr.c:2583
2583    TRACE_POSTGRESQL_BUFFER_CHECKPOINT_START(flags);
(gdb) n
2584    CheckpointStats.ckpt_write_t = GetCurrentTimestamp();
(gdb) 
2585    BufferSync(flags);
(gdb) step
BufferSync (flags=108) at bufmgr.c:1793
1793    CkptTsStatus *per_ts_stat = NULL;
(gdb) p flags
$1 = 108
(gdb) n
1797    int     mask = BM_DIRTY;
(gdb) 
1801    ResourceOwnerEnlargeBuffers(CurrentResourceOwner);
(gdb) 
1808    if (!((flags & (CHECKPOINT_IS_SHUTDOWN | CHECKPOINT_END_OF_RECOVERY |
(gdb) 
1810      mask |= BM_PERMANENT;
(gdb) 
1828    num_to_scan = 0;
(gdb) 
1829    for (buf_id = 0; buf_id < NBuffers; buf_id++)
(gdb) 
1831      BufferDesc *bufHdr = GetBufferDescriptor(buf_id);
(gdb) 
1837      buf_state = LockBufHdr(bufHdr);
(gdb) p buf_id
$2 = 0
(gdb) p NBuffers
$3 = 65536
(gdb) n
1839      if ((buf_state & mask) == mask)
(gdb) 
1853      UnlockBufHdr(bufHdr, buf_state);
(gdb) 
1829    for (buf_id = 0; buf_id < NBuffers; buf_id++)
(gdb) 
1831      BufferDesc *bufHdr = GetBufferDescriptor(buf_id);
(gdb) 
1837      buf_state = LockBufHdr(bufHdr);
(gdb) 
1839      if ((buf_state & mask) == mask)
(gdb) 
1853      UnlockBufHdr(bufHdr, buf_state);
(gdb) 
1829    for (buf_id = 0; buf_id < NBuffers; buf_id++)
(gdb) b bufmgr.c:1856
Breakpoint 2 at 0x8a68b3: file bufmgr.c, line 1856.
(gdb) c
Continuing.
Breakpoint 2, BufferSync (flags=108) at bufmgr.c:1856
1856    if (num_to_scan == 0)
(gdb) p num_to_scan
$4 = 1
(gdb) n
1859    WritebackContextInit(&wb_context, &checkpoint_flush_after);
(gdb) 
1861    TRACE_POSTGRESQL_BUFFER_SYNC_START(NBuffers, num_to_scan);
(gdb) 
1870    qsort(CkptBufferIds, num_to_scan, sizeof(CkptSortItem),
(gdb) 
1873    num_spaces = 0;
(gdb) 
1879    last_tsid = InvalidOid;
(gdb) 
1880    for (i = 0; i < num_to_scan; i++)
(gdb) 
1885      cur_tsid = CkptBufferIds[i].tsId;
(gdb) 
1891      if (last_tsid == InvalidOid || last_tsid != cur_tsid)
(gdb) p cur_tsid
$5 = 1663
(gdb) n
1895        num_spaces++;
(gdb) 
1901        sz = sizeof(CkptTsStatus) * num_spaces;
(gdb) 
1903        if (per_ts_stat == NULL)
(gdb) 
1904          per_ts_stat = (CkptTsStatus *) palloc(sz);
(gdb) 
1908        s = &per_ts_stat[num_spaces - 1];
(gdb) p sz
$6 = 40
(gdb) p num_spaces
$7 = 1
(gdb) n
1909        memset(s, 0, sizeof(*s));
(gdb) 
1910        s->tsId = cur_tsid;
(gdb) 
1917        s->index = i;
(gdb) 
1924        last_tsid = cur_tsid;
(gdb) 
1892      {
(gdb) 
1931      s->num_to_scan++;
(gdb) 
1880    for (i = 0; i < num_to_scan; i++)
(gdb) 
1934    Assert(num_spaces > 0);
(gdb) 
1941    ts_heap = binaryheap_allocate(num_spaces,
(gdb) 
1945    for (i = 0; i < num_spaces; i++)
(gdb) 
1947      CkptTsStatus *ts_stat = &per_ts_stat[i];
(gdb) 
1949      ts_stat->progress_slice = (float8) num_to_scan / ts_stat->num_to_scan;
(gdb) 
1951      binaryheap_add_unordered(ts_heap, PointerGetDatum(ts_stat));
(gdb) 
1945    for (i = 0; i < num_spaces; i++)
(gdb) 
1954    binaryheap_build(ts_heap);
(gdb) 
1962    num_processed = 0;
(gdb) p *ts_heap
$8 = {bh_size = 1, bh_space = 1, bh_has_heap_property = true, bh_compare = 0x8aa0d8 <ts_ckpt_progress_comparator>, 
  bh_arg = 0x0, bh_nodes = 0x2d666d8}
(gdb) n
1963    num_written = 0;
(gdb) 
1964    while (!binaryheap_empty(ts_heap))
(gdb) 
1966      BufferDesc *bufHdr = NULL;
(gdb) 
1968      DatumGetPointer(binaryheap_first(ts_heap));
(gdb) 
1967      CkptTsStatus *ts_stat = (CkptTsStatus *)
(gdb) 
1970      buf_id = CkptBufferIds[ts_stat->index].buf_id;
(gdb) 
1971      Assert(buf_id != -1);
(gdb) p buf_id
$9 = 160
(gdb) n
1973      bufHdr = GetBufferDescriptor(buf_id);
(gdb) 
1975      num_processed++;
(gdb) 
1989      if (pg_atomic_read_u32(&bufHdr->state) & BM_CHECKPOINT_NEEDED)
(gdb) p *bufHdr
$10 = {tag = {rnode = {spcNode = 1663, dbNode = 16384, relNode = 221290}, forkNum = MAIN_FORKNUM, blockNum = 0}, 
  buf_id = 160, state = {value = 3549691904}, wait_backend_pid = 0, freeNext = -2, content_lock = {tranche = 53, state = {
      value = 536870912}, waiters = {head = 2147483647, tail = 2147483647}}}
(gdb) n
1991        if (SyncOneBuffer(buf_id, false, &wb_context) & BUF_WRITTEN)
(gdb) 
1993          TRACE_POSTGRESQL_BUFFER_SYNC_WRITTEN(buf_id);
(gdb) 
1994          BgWriterStats.m_buf_written_checkpoints++;
(gdb) 
1995          num_written++;
(gdb) 
2003      ts_stat->progress += ts_stat->progress_slice;
(gdb) 
2004      ts_stat->num_scanned++;
(gdb) 
2005      ts_stat->index++;
(gdb) 
2008      if (ts_stat->num_scanned == ts_stat->num_to_scan)
(gdb) 
2010        binaryheap_remove_first(ts_heap);
(gdb) 
2021      CheckpointWriteDelay(flags, (double) num_processed / num_to_scan);
(gdb) 
1964    while (!binaryheap_empty(ts_heap))
(gdb) 
2025    IssuePendingWritebacks(&wb_context);
(gdb) 
2027    pfree(per_ts_stat);
(gdb) 
2028    per_ts_stat = NULL;
(gdb) 
2029    binaryheap_free(ts_heap);
(gdb) 
2035    CheckpointStats.ckpt_bufs_written += num_written;
(gdb) 
2037    TRACE_POSTGRESQL_BUFFER_SYNC_DONE(NBuffers, num_written, num_to_scan);
(gdb) 
2038  }
(gdb) 
CheckPointBuffers (flags=108) at bufmgr.c:2586
2586    CheckpointStats.ckpt_sync_t = GetCurrentTimestamp();
(gdb)

四、参考资料

PG Source Code
PgSQL · 特性分析 · 谈谈checkpoint的调度