海山数据库(He3DB)源码详解：主备复制start - qidao123.com技术社区-IT企服评测·应用市场

Datum
pg_backup_start(PG_FUNCTION_ARGS)
{
// 获取第一个参数：备份ID，类型为text的指针的指针
text *backupid = PG_GETARG_TEXT_PP(0);
// 获取第二个参数：是否进行快速备份，类型为bool
bool fast = PG_GETARG_BOOL(1);
char *backupidstr;
XLogRecPtr startpoint; // 用于存储备份开始的日志序列号（LSN）
SessionBackupState status = get_backup_status(); // 获取当前会话的备份状态
MemoryContext oldcontext; // 用于保存原始的内存上下文
// 将text类型的备份ID转换为C字符串
backupidstr = text_to_cstring(backupid);
// 检查当前会话是否已经有备份正在进行
if (status == SESSION_BACKUP_RUNNING)
// 如果有，则报错
ereport(ERROR,
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
errmsg("a backup is already in progress in this session")));

复制代码

// 切换到顶级内存上下文，因为标签文件和表空间映射文件需要在整个会话中保持有效
oldcontext = MemoryContextSwitchTo(TopMemoryContext);
// 初始化用于存储标签文件内容的StringInfo
label_file = makeStringInfo();
// 初始化用于存储表空间映射文件内容的StringInfo
tblspc_map_file = makeStringInfo();
// 切换回原始的内存上下文
MemoryContextSwitchTo(oldcontext);

复制代码

register_persistent_abort_backup_handler();

复制代码

startpoint = do_pg_backup_start(backupidstr, fast, NULL, label_file,
NULL, tblspc_map_file);

复制代码

PG_RETURN_LSN(startpoint);

复制代码

XLogRecPtr
do_pg_backup_start(const char *backupidstr, bool fast, TimeLineID *starttli_p,
StringInfo labelfile, List **tablespaces,
StringInfo tblspcmapfile)
{
bool backup_started_in_recovery = false;
XLogRecPtr checkpointloc;
XLogRecPtr startpoint;
TimeLineID starttli;
pg_time_t stamp_time;
char strfbuf[128];
char xlogfilename[MAXFNAMELEN];
XLogSegNo _logSegNo;
// 检查当前是否处于恢复模式
backup_started_in_recovery = RecoveryInProgress();
/*
* 在恢复模式下，我们不需要检查WAL级别。因为，如果WAL级别不足，在恢复期间是不可能到达这里的。
* 这意味着，如果系统正在恢复，它已经有了足够的WAL级别来支持恢复操作。
*/
if (!backup_started_in_recovery && !XLogIsNeeded())
// 如果不在恢复模式下，并且当前不需要WAL（即WAL级别可能不足），则报告错误
ereport(ERROR,
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
errmsg("WAL level not sufficient for making an online backup"),
errhint("wal_level must be set to "replica" or "logical" at server start.")));
// 检查提供的备份标签字符串长度是否过长
if (strlen(backupidstr) > MAXPGPATH)
ereport(ERROR,
(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
errmsg("backup label too long (max %d bytes)",
MAXPGPATH)));
WALInsertLockAcquireExclusive();// 以独占模式获取WAL插入锁
XLogCtl->Insert.runningBackups++;// 增加正在运行的备份计数
XLogCtl->Insert.forcePageWrites = true;// 启用强制全页写入
WALInsertLockRelease();
// 释放WAL插入锁
/* Ensure we release forcePageWrites if fail below */
PG_ENSURE_ERROR_CLEANUP(pg_backup_start_callback, (Datum) 0);
{
bool gotUniqueStartpoint = false; // 标记是否已获取到唯一的起始点
DIR *tblspcdir; // 指向表空间目录的DIR指针
struct dirent *de; // 用于读取目录条目的指针
tablespaceinfo *ti; // 指向表空间信息结构体的指针
int datadirpathlen; // 数据目录路径的长度
if (!backup_started_in_recovery)
RequestXLogSwitch(false);

复制代码

do
{
// 用于存储检查点时的fullPageWrites状态
bool checkpointfpw;
/*
* 强制进行一次CHECKPOINT。这不仅是防止页面撕裂问题所必需的，而且确保两次连续的备份运行将有不同的检查点位置，
* 从而有不同的历史文件名，即使两次备份之间没有任何操作发生。
* 如果用户请求了快速模式（通过传递fast = true），则使用CHECKPOINT_IMMEDIATE；否则，可能会花费一些时间。
*/
RequestCheckpoint(CHECKPOINT_FORCE | CHECKPOINT_WAIT |
(fast ? CHECKPOINT_IMMEDIATE : 0));
/*
* 现在我们需要获取检查点记录的位置，以及它的REDO指针。从检查点开始恢复所需的最早的WAL位置正是REDO指针。
*/
LWLockAcquire(ControlFileLock, LW_SHARED); // 获取对控制文件的共享锁
checkpointloc = ControlFile->checkPoint; // 读取检查点的位置
startpoint = ControlFile->checkPointCopy.redo; // 读取REDO指针
starttli = ControlFile->checkPointCopy.ThisTimeLineID; // 读取时间线ID
checkpointfpw = ControlFile->checkPointCopy.fullPageWrites; // 读取检查点时的fullPageWrites状态
LWLockRelease(ControlFileLock); // 释放对控制文件的锁
if (backup_started_in_recovery) // 如果备份是在恢复模式下启动的
{
XLogRecPtr recptr; // 用于存储最后一次禁用fullPageWrites的WAL记录指针
/*
* 检查自上次用作备份起始检查点的重启点以来，在线备份期间重放的所有WAL是否都包含全页写入。
*/
SpinLockAcquire(&XLogCtl->info_lck); // 获取XLogCtl的自旋锁
recptr = XLogCtl->lastFpwDisableRecPtr;// 读取最后一次禁用fullPageWrites的WAL记录指针
SpinLockRelease(&XLogCtl->info_lck); // 释放XLogCtl的自旋锁
/*
* 如果检查点时的fullPageWrites未启用，或者REDO指针小于或等于最后一次禁用fullPageWrites的WAL记录指针，
* 则报错，因为这意味着在备库上进行的在线备份可能已损坏。
*/
if (!checkpointfpw || startpoint <= recptr)
ereport(ERROR,
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
errmsg("WAL generated with full_page_writes=off was replayed "
"since last restartpoint"),
errhint("This means that the backup being taken on the standby "
"is corrupt and should not be used. "
"Enable full_page_writes and run CHECKPOINT on the primary, "
"and then try an online backup again.")));
/*
* 在恢复模式下，由于我们不使用备份结束WAL记录，也不写入备份历史文件，
* 因此起始WAL位置不需要唯一。这意味着同时开始的两个基础备份可能会使用相同的检查点作为起始位置。
*/
gotUniqueStartpoint = true;// 标记为已获得唯一起始点（在恢复模式下总是为真）
}
WALInsertLockAcquireExclusive();// 以独占模式获取WAL插入锁
if (XLogCtl->Insert.lastBackupStart < startpoint)// 如果当前检查点位置大于之前的最后一个备份起始点
{
XLogCtl->Insert.lastBackupStart = startpoint;// 更新最后一个备份起始点
gotUniqueStartpoint = true;// 标记为已获得唯一起始点
}
WALInsertLockRelease();
// 释放WAL插入锁
} while (!gotUniqueStartpoint); // 如果未获得唯一起始点，则继续循环

复制代码

// 分配并初始化一个目录对象tblspcdir，用于读取"pg_tblspc"目录（存储表空间链接的目录）。
// 调用XLByteToSeg函数，将startpoint（一个表示WAL位置的字节偏移量）和日志段大小（wal_segment_size）
// 转换为日志段编号（_logSegNo）。这通常用于找到WAL日志文件中特定位置所在的段。
XLByteToSeg(startpoint, _logSegNo, wal_segment_size);
// 调用XLogFileName函数，根据事务日志标识符（starttli），日志段编号（_logSegNo），
// 和日志段大小（wal_segment_size）来生成WAL日志文件的名称，并将该名称存储在xlogfilename中。
XLogFileName(xlogfilename, starttli, _logSegNo, wal_segment_size);
// 计算DataDir（数据库数据目录的路径）的长度，并将其存储在datadirpathlen中
datadirpathlen = strlen(DataDir);
tblspcdir = AllocateDir("pg_tblspc");
// 使用ReadDir函数遍历"pg_tblspc"目录下的所有条目
while ((de = ReadDir(tblspcdir, "pg_tblspc")) != NULL)
{
// 定义一个足够大的字符数组fullpath来存储表空间目录的完整路径
char fullpath[MAXPGPATH + 10];
// 定义一个字符数组linkpath来存储表空间链接的目标路径（如果表空间是符号链接的话）。
char linkpath[MAXPGPATH];
// 定义一个字符指针relpath，用于存储相对于DataDir的表空间路径（初始化为NULL）
char *relpath = NULL;
// 定义一个整数rllen来存储relpath的长度（稍后可能使用）
int rllen;
// 定义一个StringInfoData类型的escapedpath，用于存储转义后的路径（这里未直接使用，但可能用于后续处理）
StringInfoData escapedpath;
// 定义一个字符指针s，用于遍历字符串（未直接使用）
char *s;
/* Skip anything that doesn't look like a tablespace */
// 跳过那些名字不包含纯数字的条目，因为表空间目录名通常只包含数字。
if (strspn(de->d_name, "0123456789") != strlen(de->d_name))
continue;
// 构造表空间目录的完整路径
snprintf(fullpath, sizeof(fullpath), "pg_tblspc/%s", de->d_name);
// 跳过那些不是符号链接或junction的条目。这里检查条目类型，确保它确实是一个指向表空间的链接。
// 注意：在测试中，如果启用了allow_in_place_tablespaces，可能会直接在pg_tblspc下创建目录，这将不满足条件。
if (get_dirent_type(fullpath, de, false, ERROR) != PGFILETYPE_LNK)
continue;
#if defined(HAVE_READLINK) || defined(WIN32)
rllen = readlink(fullpath, linkpath, sizeof(linkpath));
if (rllen < 0)
{
ereport(WARNING,
(errmsg("could not read symbolic link "%s": %m",
fullpath)));
continue;
}
else if (rllen >= sizeof(linkpath))
{
ereport(WARNING,
(errmsg("symbolic link "%s" target is too long",
fullpath)));
continue;
}
linkpath[rllen] = '\0';
initStringInfo(&escapedpath);
for (s = linkpath; *s; s++)
{
if (*s == '\n' || *s == '\r' || *s == '\\')
appendStringInfoChar(&escapedpath, '\\');
appendStringInfoChar(&escapedpath, *s);
}
// 检查linkpath是否以PGDATA目录的路径开头，并且紧接着是目录分隔符
// 如果是，那么计算相对路径，否则relpath为NULL
if (rllen > datadirpathlen &&
strncmp(linkpath, DataDir, datadirpathlen) == 0 &&
IS_DIR_SEP(linkpath[datadirpathlen]))
relpath = linkpath + datadirpathlen + 1; //计算相对路径的起始位置
// 为tablespaceinfo结构体分配内存
ti = palloc(sizeof(tablespaceinfo));
// 设置tablespaceinfo结构体的字段
// 注意：这里将de->d_name作为OID可能是一个错误，因为OID通常是整数类型
ti->oid = pstrdup(de->d_name);
ti->path = pstrdup(linkpath);//设置原始路径
ti->rpath = relpath ? pstrdup(relpath) : NULL;// 如果计算了相对路径，则设置，否则为NULL
ti->size = -1;//初始化大小为-1，可能表示未知或未计算
// 如果tablespaces列表不为空，则将ti添加到列表中
if (tablespaces)
*tablespaces = lappend(*tablespaces, ti);
// 将tablespace的OID和转义后的路径写入到tblspcmapfile中
// 假设tblspcmapfile是一个用于存储映射信息的文件
appendStringInfo(tblspcmapfile, "%s %s\n",
ti->oid, escapedpath.data);
// 释放escapedpath.data占用的内存
pfree(escapedpath.data);
#else
/*
* 如果平台不支持符号链接，那么理论上不应该存在表空间，因为表空间通常是通过符号链接实现的。
* 如果检测到表空间，那么可能是由其他机制或外部工具创建的。这里发出警告并忽略它们。
*/
ereport(WARNING,
(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
errmsg("tablespaces are not supported on this platform")));
#endif
}
//释放之前打开的目录句柄
FreeDir(tblspcdir);

复制代码

//向标签文件中添加wal开始位置信息，检查点的位置信息，流式备份，指出是主库还是备库开始的，添加备份开始的时间信息，标签信息以及开始时的时间线
stamp_time = (pg_time_t) time(NULL);
pg_strftime(strfbuf, sizeof(strfbuf),
"%Y-%m-%d %H:%M:%S %Z",
pg_localtime(&stamp_time, log_timezone));
appendStringInfo(labelfile, "START WAL LOCATION: %X/%X (file %s)\n",
LSN_FORMAT_ARGS(startpoint), xlogfilename);
appendStringInfo(labelfile, "CHECKPOINT LOCATION: %X/%X\n",
LSN_FORMAT_ARGS(checkpointloc));
appendStringInfo(labelfile, "BACKUP METHOD: streamed\n");
appendStringInfo(labelfile, "BACKUP FROM: %s\n",
backup_started_in_recovery ? "standby" : "primary");
appendStringInfo(labelfile, "START TIME: %s\n", strfbuf);
appendStringInfo(labelfile, "LABEL: %s\n", backupidstr);
appendStringInfo(labelfile, "START TIMELINE: %u\n", starttli);
}
//结束错误处理保证块的定义
PG_END_ENSURE_ERROR_CLEANUP(pg_backup_start_callback, (Datum) 0);
//标记备份的开始阶段已成功完成
sessionBackupState = SESSION_BACKUP_RUNNING;
//返回wal的开始位置
if (starttli_p)
*starttli_p = starttli;
return startpoint;
}

复制代码

static void
pg_backup_start_callback(int code, Datum arg)
{
WALInsertLockAcquireExclusive();

复制代码

Assert(XLogCtl->Insert.runningBackups > 0);

复制代码

XLogCtl->Insert.runningBackups--;

复制代码

if (XLogCtl->Insert.runningBackups == 0)
{
XLogCtl->Insert.forcePageWrites = false;
}

复制代码

WALInsertLockRelease();

复制代码

if (XLogCtl->Insert.runningBackups == 0)
{
XLogCtl->Insert.forcePageWrites = false;
}

复制代码

WALInsertLockRelease();

复制代码