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,检查点请求标记位定义.
/*
* OR-able request flag bits for checkpoints. The "cause" bits are used only
* for logging purposes. Note: the flags must be defined so that it's
* sensible to OR together request flags arising from different requestors.
*/
/* These directly affect the behavior of CreateCheckPoint and subsidiaries */
#define CHECKPOINT_IS_SHUTDOWN 0x0001 /* Checkpoint is for shutdown */
#define CHECKPOINT_END_OF_RECOVERY 0x0002 /* Like shutdown checkpoint, but
* issued at end of WAL recovery */
#define CHECKPOINT_IMMEDIATE 0x0004 /* Do it without delays */
#define CHECKPOINT_FORCE 0x0008 /* Force even if no activity */
#define CHECKPOINT_FLUSH_ALL 0x0010 /* Flush all pages, including those
* belonging to unlogged tables */
/* These are important to RequestCheckpoint */
#define CHECKPOINT_WAIT 0x0020 /* Wait for completion */
#define CHECKPOINT_REQUESTED 0x0040 /* Checkpoint request has been made */
/* These indicate the cause of a checkpoint request */
#define CHECKPOINT_CAUSE_XLOG 0x0080 /* XLOG consumption */
#define CHECKPOINT_CAUSE_TIME 0x0100 /* Elapsed time */
二、源码解读
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释放资源,更新统计信息
/*
* BufferSync -- Write out all dirty buffers in the pool.
* 把缓冲池中所有脏页持久化到物理存储中.
*
* This is called at checkpoint time to write out all dirty shared buffers.
* The checkpoint request flags should be passed in. If CHECKPOINT_IMMEDIATE
* is set, we disable delays between writes; if CHECKPOINT_IS_SHUTDOWN,
* CHECKPOINT_END_OF_RECOVERY or CHECKPOINT_FLUSH_ALL is set, we write even
* unlogged buffers, which are otherwise skipped. The remaining flags
* currently have no effect here.
* 该函数在checkpoint时把缓冲池中所有脏页刷到磁盘上.
* 输入参数为checkpoint请求标记.
* 如请求标记为CHECKPOINT_IMMEDIATE,在写入期间禁用延迟;
* 如为CHECKPOINT_IS_SHUTDOWN/CHECKPOINT_END_OF_RECOVERY/CHECKPOINT_FLUSH_ALL,
* 就算正常情况下会忽略的unlogged缓存,也会写入到磁盘上.
* 其他标记在这里没有影响.
*/
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;
/* Make sure we can handle the pin inside SyncOneBuffer */
//确保可以处理在SyncOneBuffer函数中的pin page
ResourceOwnerEnlargeBuffers(CurrentResourceOwner);
/*
* Unless this is a shutdown checkpoint or we have been explicitly told,
* we write only permanent, dirty buffers. But at shutdown or end of
* recovery, we write all dirty buffers.
*/
//如为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;
/*
* Loop over all buffers, and mark the ones that need to be written with
* BM_CHECKPOINT_NEEDED. Count them as we go (num_to_scan), so that we
* can estimate how much work needs to be done.
* 遍历缓存,使用BM_CHECKPOINT_NEEDED标记需要写入的page.
* 对这些pages计数以便估算有多少工作需要完成.
*
* This allows us to write only those pages that were dirty when the
* checkpoint began, and not those that get dirtied while it proceeds.
* Whenever a page with BM_CHECKPOINT_NEEDED is written out, either by us
* later in this function, or by normal backends or the bgwriter cleaning
* scan, the flag is cleared. Any buffer dirtied after this point won't
* have the flag set.
* 只需要写在checkpoint开始时的脏页,不需要包括在checkpoint期间变脏的page.
* 一旦标记为BM_CHECKPOINT_NEEDED的脏页完成刷盘,
* 在这个函数后续处理逻辑或者普通的后台进程/bgwriter进程会重置该标记.
* 所有在该时点的脏页不会设置为BM_CHECKPOINT_NEEDED.
*
* Note that if we fail to write some buffer, we may leave buffers with
* BM_CHECKPOINT_NEEDED still set. This is OK since any such buffer would
* certainly need to be written for the next checkpoint attempt, too.
* 要注意的是脏页刷盘出错,脏页的标记仍为BM_CHECKPOINT_NEEDED,在下次checkpoint是尝试再次刷盘.
*/
num_to_scan = 0;
for (buf_id = 0; buf_id < NBuffers; buf_id++)
{
BufferDesc *bufHdr = GetBufferDescriptor(buf_id);
/*
* Header spinlock is enough to examine BM_DIRTY, see comment in
* SyncOneBuffer.
*/
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; /* nothing to do */
WritebackContextInit(&wb_context, &checkpoint_flush_after);
TRACE_POSTGRESQL_BUFFER_SYNC_START(NBuffers, num_to_scan);
/*
* Sort buffers that need to be written to reduce the likelihood of random
* IO. The sorting is also important for the implementation of balancing
* writes between tablespaces. Without balancing writes we'd potentially
* end up writing to the tablespaces one-by-one; possibly overloading the
* underlying system.
* 排序需刷盘的脏页,用于避免随机IO.
*/
qsort(CkptBufferIds, num_to_scan, sizeof(CkptSortItem),
ckpt_buforder_comparator);
num_spaces = 0;
/*
* Allocate progress status for each tablespace with buffers that need to
* be flushed. This requires the to-be-flushed array to be sorted.
* 为每一个需要刷脏页的表空间分配进度状态.
*/
last_tsid = InvalidOid;
for (i = 0; i < num_to_scan; i++)
{
CkptTsStatus *s;
Oid cur_tsid;
cur_tsid = CkptBufferIds[i].tsId;
/*
* Grow array of per-tablespace status structs, every time a new
* tablespace is found.
*/
if (last_tsid == InvalidOid || last_tsid != cur_tsid)
{
Size sz;
num_spaces++;
/*
* Not worth adding grow-by-power-of-2 logic here - even with a
* few hundred tablespaces this should be fine.
*/
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;
/*
* The first buffer in this tablespace. As CkptBufferIds is sorted
* by tablespace all (s->num_to_scan) buffers in this tablespace
* will follow afterwards.
*/
s->index = i;
/*
* progress_slice will be determined once we know how many buffers
* are in each tablespace, i.e. after this loop.
*/
last_tsid = cur_tsid;
}
else
{
s = &per_ts_stat[num_spaces - 1];
}
s->num_to_scan++;
}
Assert(num_spaces > 0);
/*
* Build a min-heap over the write-progress in the individual tablespaces,
* and compute how large a portion of the total progress a single
* processed buffer is.
* 在单个标记的写进度上构建最小堆,并计算单个处理缓冲区占比多少.
*/
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);
/*
* Iterate through to-be-checkpointed buffers and write the ones (still)
* marked with BM_CHECKPOINT_NEEDED. The writes are balanced between
* tablespaces; otherwise the sorting would lead to only one tablespace
* receiving writes at a time, making inefficient use of the hardware.
* 迭代处理to-be-checkpointed buffers,刷脏页.
* 在表空间之间写入是平衡的.
*/
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++;
/*
* We don't need to acquire the lock here, because we're only looking
* at a single bit. It's possible that someone else writes the buffer
* and clears the flag right after we check, but that doesn't matter
* since SyncOneBuffer will then do nothing. However, there is a
* further race condition: it's conceivable that between the time we
* examine the bit here and the time SyncOneBuffer acquires the lock,
* someone else not only wrote the buffer but replaced it with another
* page and dirtied it. In that improbable case, SyncOneBuffer will
* write the buffer though we didn't need to. It doesn't seem worth
* guarding against this, though.
*/
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++;
}
}
/*
* Measure progress independent of actually having to flush the buffer
* - otherwise writing become unbalanced.
*/
ts_stat->progress += ts_stat->progress_slice;
ts_stat->num_scanned++;
ts_stat->index++;
/* Have all the buffers from the tablespace been processed? */
if (ts_stat->num_scanned == ts_stat->num_to_scan)
{
binaryheap_remove_first(ts_heap);
}
else
{
/* update heap with the new progress */
binaryheap_replace_first(ts_heap, PointerGetDatum(ts_stat));
}
/*
* Sleep to throttle our I/O rate.
* 休眠 : 控制I/O频率
*/
CheckpointWriteDelay(flags, (double) num_processed / num_to_scan);
}
/* issue all pending flushes */
IssuePendingWritebacks(&wb_context);
pfree(per_ts_stat);
per_ts_stat = NULL;
binaryheap_free(ts_heap);
/*
* Update checkpoint statistics. As noted above, this doesn't include
* buffers written by other backends or bgwriter scan.
*/
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); /* performs all required fsyncs */
(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 ,
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的调度
网站题目：PostgreSQL源码解读（211）-后台进程#10（checkpointer-BufferSync）
本文URL：http://shouzuofang.com/article/ijoghg.html