大云海山数据库(He3DB)源码详解：主备复制-流复制

大云海山数据库(He3DB)源码详解：主备复制-流复制

背景

He3DB 采用了先进的存储引擎和查询优化技术，可以或许快速处理大量数据和复杂查询。无论是 OLTP（在线事件处理）还是 OLAP（在线分析处理）场景，都能提供精彩的性能表现。He3DB 具备完善的数据备份和恢复机制，可以或许在系统故障或数据破坏时快速恢复数据，确保业务的连续性。He3DB 支持水平扩展和垂直扩展，可以轻松应对不断增长的数据需求。He3DB 提供了严格的访问控制和数据加密功能，确保数据的安全性和隐私性。
流复制——概述

在数据库中，是一种数据库复制技术，它答应一主多从的数据库集簇实实际时数据复制和同步。
流复制——XLogInsert

XLogInsert是用于插入预写日志（Write-Ahead Log，WAL）记录的关键函数。

• 查抄条件条件
  起首查抄是否已经调用了XLogBeginInsert()，如果没有调用，则通过elog(ERROR, "XLogBeginInsert was not called")记录错误信息
  查抄传入的info参数中的位掩码是否有效，只答应设置特定的位，否则通过elog(PANIC, "invalid xlog info mask %02X", info)记录错误信息
  

2. XLogRecPtr
3. XLogInsert(RmgrId rmid, uint8 info)
4. {
5.         XLogRecPtr        EndPos;// 用于存储返回的记录末尾位置（LSN）
7.         /* XLogBeginInsert() must have been called. */
8.         /* 确保已经调用了XLogBeginInsert()。 */
9.         if (!begininsert_called)
10.                 elog(ERROR, "XLogBeginInsert was not called");
12.         /*
13.          * The caller can set rmgr bits, XLR_SPECIAL_REL_UPDATE and
14.          * XLR_CHECK_CONSISTENCY; the rest are reserved for use by me.
15.          */
16.         /*  
17.          * 检查info参数中的位掩码是否有效。只允许设置RMID位、XLR_SPECIAL_REL_UPDATE和XLR_CHECK_CONSISTENCY。  
18.          */
19.         if ((info & ~(XLR_RMGR_INFO_MASK |
20.                                   XLR_SPECIAL_REL_UPDATE |
21.                                   XLR_CHECK_CONSISTENCY)) != 0)
22.                 elog(PANIC, "invalid xlog info mask %02X", info);// 如果设置了无效位，则记录恐慌
24.         TRACE_POSTGRESQL_WAL_INSERT(rmid, info);// 跟踪WAL插入事件（宏或函数）

复制代码

• 处理启动模式
  如果处于启动模式（IsBootstrapProcessingMode()为真）且资源管理器 ID（rmid）不是RM_XLOG_ID，则重置插入状态（XLogResetInsertion()），并返回一个模拟的记录指针（通常指向第一个查抄点记录的开头）
  

1. /*  
2.          * 在启动模式（bootstrap）下，除了RM_XLOG_ID外，不实际记录任何内容；  
3.          * 返回一个模拟的记录指针。  
4.          */  
5.         if (IsBootstrapProcessingMode() && rmid != RM_XLOG_ID)
6.         {
7.                 XLogResetInsertion();// 重置插入状态
8.                 EndPos = SizeOfXLogLongPHD; /* start of 1st chkpt record */// 返回一个假定的位置，通常指向第一个检查点记录的开头
9.                 return EndPos;
10.         }

复制代码

• 正常插入流程
  如果不是启动模式，则进入循环尝试插入记录。
  在循环内部：
  获取决定是否举行全页写的信息，这些值在获取锁之前可能会改变，但XLogInsertRecord会在获得锁后重新查抄它们。
  组装XLOG记录，包罗收集所有之前通过XLogRegister*函数注册的数据和缓冲区引用。
  尝试将组装好的记录插入到XLOG中，并处理全页写（如果必要）。如果插入失败（返回InvalidXLogRecPtr），则继续循环尝试插入
  插入成功后，重置插入状态，并返回记录末尾的LSN（EndPos）
  

1. do
2.         {
3.                 // 局部变量声明，用于后续操作
4.                 XLogRecPtr        RedoRecPtr;
5.                 bool                doPageWrites;
6.                 bool                topxid_included = false;
7.                 XLogRecPtr        fpw_lsn;
8.                 XLogRecData *rdt;
9.                 int                        num_fpi = 0;
10.                 /*  
11.                  * 获取决定是否需要全页写（full-page writes）的信息。  
12.                  * 注意：这些值在获取锁之前可能会改变，但XLogInsertRecord会在获得锁后重新检查它们。  
13.                  */  
14.                 GetFullPageWriteInfo(&RedoRecPtr, &doPageWrites);
15.         
16.                 /*  
17.                  * 组装XLOG记录。这包括收集所有之前通过XLogRegister*函数注册的数据和缓冲区引用。  
18.                  */  
19.                 rdt = XLogRecordAssemble(rmid, info, RedoRecPtr, doPageWrites,
20.                                                                  &fpw_lsn, &num_fpi, &topxid_included);
21.                 /*  
22.                  * 将组装好的记录插入到XLOG中，并处理全页写（如果需要）。  
23.                  * 如果因为某些原因（如锁竞争）无法插入，则返回InvalidXLogRecPtr，并尝试再次插入。  
24.                  */  
26.                 EndPos = XLogInsertRecord(rdt, fpw_lsn, curinsert_flags, num_fpi,
27.                                                                   topxid_included);
28.         } while (EndPos == InvalidXLogRecPtr);// 如果插入失败，则重试  
30.         XLogResetInsertion(); // 重置插入状态
32.         return EndPos;// 返回记录末尾的LSN  
33. }

复制代码

函数调用栈

部门调用栈如下图所示

流复制——pg_physical_replication_slot_advance

pg_physical_replication_slot_advance推进复制槽。

• 参数获取和初始化
  函数接受一个参数XLogRecPtr moveto，表示要推进到的WAL日志位置
  声明变量startlsn用于存储当前复制槽的起始WAL日志位置，初始化为MyReplicationSlot->data.restart_lsn
  声明变量retlsn用于存储返回的WAL日志位置，初始化为startlsn
  

1. static XLogRecPtr
2. pg_physical_replication_slot_advance(XLogRecPtr moveto)
3. {
4.         XLogRecPtr        startlsn = MyReplicationSlot->data.restart_lsn;
5.         XLogRecPtr        retlsn = startlsn;

复制代码

• 参数查抄
  通过Assert(moveto!= InvalidXLogRecPtr)断言传入的要推进到的位置不是无效的
  

1.         Assert(moveto != InvalidXLogRecPtr);

复制代码

• 查抄并推进复制槽位置
  如果startlsn小于moveto，表示可以推进复制槽位置
  起首获取MyReplicationSlot的互斥锁（SpinLockAcquire(&MyReplicationSlot->mutex)）
  将复制槽的restart_lsn更新为moveto，表示推进了复制槽的起始位置
  将retlsn更新为moveto，作为要返回的WAL日志位置
  开释互斥锁（SpinLockRelease(&MyReplicationSlot->mutex)）
  

1. if (startlsn < moveto)
2.         {
3.                 SpinLockAcquire(&MyReplicationSlot->mutex);
4.                 MyReplicationSlot->data.restart_lsn = moveto;
5.                 SpinLockRelease(&MyReplicationSlot->mutex);
6.                 retlsn = moveto;

复制代码

• 标记复制槽为脏
  调用ReplicationSlotMarkDirty函数将复制槽标记为脏，这样在下次查抄点时会将复制槽的信息写入磁盘，确保在干净关闭数据库后数据的一致性。即使在发生崩溃的情况下，推进的WAL日志位置可能会丢失，但这种方式可以保证在干净关闭后数据的一致性
  

1.         ReplicationSlotMarkDirty();

复制代码

返回效果
返回retlsn，即推进后的WAL日志位置或原始位置（如果没有推进）

1.         return retlsn;
2. }

复制代码

函数调用栈

He3DB其余文章参考链接

海山数据库(He3DB)源码详解：He3DB-CLOG日志管理器函数之TransactionIdSetTreeStatus
海山数据库(He3DB)+AI（五）：一种基于强化学习的数据库旋钮调优方法
海山数据库(He3DB)+AI（四）：一种基于迁移学习的启发式数据库旋钮调优方法
海山数据库(He3DB)源码解读：海山PG 词法、语法分析
海山数据库(He3DB)源码详解：海山PG 空闲空间映射表FSM
作者介绍

周雨慧 中移（苏州）软件技术有限公司 数据库内核开辟工程师
(https://www.modb.pro/db/1838032718672703488)
海山数据库(He3DB)源码详解：海山PG 空闲空间映射表FSM
作者介绍

周雨慧 中移（苏州）软件技术有限公司 数据库内核开辟工程师

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。