PostgreSQL 源码解读（211）- 后台进程#10（checkpointer-BufferSync）
短信预约 -IT技能 免费直播动态提醒
本节介绍了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的调度
免责声明：
① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的，并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据，供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。
② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341
PostgreSQL 源码解读（211）- 后台进程#10（checkpointer-BufferSync）
下载Word文档到电脑，方便收藏和打印～
下载Word文档
![[mysql]mysql8修改root密码](https://static.528045.com/imgs/11.jpg?imageMogr2/format/webp/blur/1x0/quality/35)
