huawei
/
openGauss-server
0
0
Fork 77
Code Issues Pull Requests 24 Projects Releases Wiki Activity

opengauss项目代码注释 #26

Open
nuoya wants to merge 55 commits from nuoya/openGauss-server:master into master
pull from: nuoya/openGauss-server:master
merge into: huawei:master
Conversation 0 Commits 55 Files Changed 130 +212 -103
1 changed files with 212 additions and 103 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 1e0967d668 - Show all commits

315
src/gausskernel/storage/xlog_share_storage/xlog_share_storage.cpp
View File

@ -37,7 +37,7 @@
#include "miscadmin.h"
#include "pgstat.h"
#include "replication/syncrep_gramparse.h"
#include "replication/walsender_private.h"
#include "replication/ walsender_private.h"
#include "storage/ipc.h"
#include "storage/dorado_operation/dorado_fd.h"
#include "storage/xlog_share_storage/xlog_share_storage.h"
@ -53,33 +53,39 @@ const static uint32 CHECK_LOCK_INTERVAL = 0xFF;
const static uint32 MAX_SIZE_CAN_COPY_TO_SHARE = 128 * 1024 * 1024;
const static uint32 PTR_PRINT_SHIFT_SIZE = 32;
// 这个函数是用来处理 SIGHUP 信号的。
// 功能是在接收到 SIGHUP 信号时，设置一个标志来指示已经收到信号，并尝试唤醒进程。
// 以便进程可以在需要时执行相应的操作。
static void SharedStorageXlogCopyBackendSigHupHandler(SIGNAL_ARGS)
{
int save_errno = errno;
int save_errno = errno; // 保存当前的错误信号到 save_errno 变量。
// 改变标志表示已经收到了 SIGHUP 信号。
t_thrd.sharestoragexlogcopyer_cxt.got_SIGHUP = true;
/*
*检查全局变量 t_thrd.proc:
*如果不为空，就调用 SetLatch 函数。
*该函数会触发该进程的等待标志（Latch），以便在需要的时候唤醒进程执行一些操作。
*/
if (t_thrd.proc) {
SetLatch(&t_thrd.proc->procLatch);
}
errno = save_errno;
errno = save_errno; // 将错误信号恢复为之前保存的值。
}
// 用来处理关闭信号,和 SharedStorageXlogCopyBackendSigHupHandler()函数类似
static void SharedStorageXlogCopyBackendShutdownHandler(SIGNAL_ARGS)
{
int save_errno = errno;
t_thrd.sharestoragexlogcopyer_cxt.shutdown_requested = true;
if (t_thrd.proc)
SetLatch(&t_thrd.proc->procLatch);
errno = save_errno;
}
// 用于快速终止进程的信号处理函数。它的主要作用是退出进程，并在退出之前执行一些清理操作。
static void SharedStorageXlogCopyBackendQuickDie(SIGNAL_ARGS)
{
gs_signal_setmask(&t_thrd.libpq_cxt.BlockSig, NULL);
gs_signal_setmask(&t_thrd.libpq_cxt.BlockSig, NULL); // 用于设置信号掩码,为了防止在退出过程中受到其他信号的干扰。
/*
* We DO NOT want to run proc_exit() callbacks -- we're here because
@ -89,7 +95,7 @@ static void SharedStorageXlogCopyBackendQuickDie(SIGNAL_ARGS)
* things by calling exit() directly, we have to reset the callbacks
* explicitly to make this work as intended.
*/
on_exit_reset();
on_exit_reset(); // 为了重置在进程退出时应该调用的回调函数，以确保不执行任何已注册的退出回调函数。
/*
* Note we do exit(2) not exit(0). This is to force the postmaster into a
@ -99,14 +105,16 @@ static void SharedStorageXlogCopyBackendQuickDie(SIGNAL_ARGS)
* should ensure the postmaster sees this as a crash, too, but no harm in
* being doubly sure.)
*/
exit(2);
exit(2); // 调用 exit，使进程以状态码 2 终止。
}
// 用于处理 SigUsr1信号的信号处理函数。
static void SharedStorageXlogCopyBackendSigUsr1Handler(SIGNAL_ARGS)
{
// 保存当前的错误信号到 save_errno 变量。
// 目的是为了在函数执行期间不影响其他部分对错误号的使用。
int saveErrno = errno;
latch_sigusr1_handler();
latch_sigusr1_handler(); // SetLatch使用SigUsr1唤醒在latch上等待的进程。如果我们在等待，唤醒 WaitLatch。
errno = saveErrno;
}
@ -117,29 +125,32 @@ void LocalXLogRead(char *buf, XLogRecPtr startptr, Size count)
XLogRecPtr recptr;
Size nbytes;
p = buf;
recptr = startptr;
nbytes = count;
p = buf; // 指向缓冲区的指针
recptr = startptr; // 当前的WAL日志指针
nbytes = count; // 要读取的字节数
while (nbytes > 0) {
uint32 startoff;
int segbytes;
int readbytes;
startoff = recptr % XLogSegSize;
startoff = recptr % XLogSegSize; // 计算当前WAL日志指针在XLOG段内的偏移量
/* Do we need to switch to a different xlog segment? */
if (t_thrd.sharestoragexlogcopyer_cxt.readFile < 0 ||
!XLByteInSeg(recptr, t_thrd.sharestoragexlogcopyer_cxt.readSegNo)) {
// 检查是否需要切换到不同的XLOG段
if (t_thrd.sharestoragexlogcopyer_cxt.readFile <
0 || // 检查名为 readFile 的文件描述符是否为负值，从而判断文件是否已经关闭或无效。
!XLByteInSeg(recptr, t_thrd.sharestoragexlogcopyer_cxt
.readSegNo)) { // 条件判断表达式，用于检查给定的WAL日志位置是否在当前的XLOG段内。
char path[MAXPGPATH];
// 如果之前打开的文件描述符有效，则关闭它
if (t_thrd.sharestoragexlogcopyer_cxt.readFile >= 0) {
(void)close(t_thrd.sharestoragexlogcopyer_cxt.readFile);
}
// 计算新的XLOG段文件路径
XLByteToSeg(recptr, t_thrd.sharestoragexlogcopyer_cxt.readSegNo);
XLogFilePath(path, MAXPGPATH, t_thrd.xlog_cxt.ThisTimeLineID, t_thrd.sharestoragexlogcopyer_cxt.readSegNo);
// 打开新的XLOG段文件进行读取
t_thrd.sharestoragexlogcopyer_cxt.readFile = BasicOpenFile(path, O_RDONLY | PG_BINARY, 0);
if (t_thrd.sharestoragexlogcopyer_cxt.readFile < 0) {
/*
@ -147,24 +158,29 @@ void LocalXLogRead(char *buf, XLogRecPtr startptr, Size count)
* asked for a too old WAL segment that has already been
* removed or recycled.
*/
// 如果文件不存在，可能是因为要求的WAL段已被删除或回收
if (errno == ENOENT) {
ereport(ERROR, (errcode_for_file_access(),
errmsg("requested WAL segment %s has already been removed",
XLogFileNameP(t_thrd.xlog_cxt.ThisTimeLineID,
t_thrd.sharestoragexlogcopyer_cxt.readSegNo))));
} else {
// 打开文件失败，报告错误
ereport(ERROR, (errcode_for_file_access(),
errmsg("could not open file \"%s\" (log segment %s): %m", path,
XLogFileNameP(t_thrd.xlog_cxt.ThisTimeLineID,
t_thrd.sharestoragexlogcopyer_cxt.readSegNo))));
}
}
t_thrd.sharestoragexlogcopyer_cxt.readOff = 0;
t_thrd.sharestoragexlogcopyer_cxt.readOff = 0; // 重置读取偏移量
}
/* Need to seek in the file? */
// 重置读取偏移量
if (t_thrd.sharestoragexlogcopyer_cxt.readOff != startoff) {
// 定位到指定的偏移量
if (lseek(t_thrd.sharestoragexlogcopyer_cxt.readFile, (off_t)startoff, SEEK_SET) < 0) {
// 定位失败，关闭文件并报告错误
(void)close(t_thrd.sharestoragexlogcopyer_cxt.readFile);
t_thrd.sharestoragexlogcopyer_cxt.readFile = -1;
ereport(ERROR,
@ -173,22 +189,27 @@ void LocalXLogRead(char *buf, XLogRecPtr startptr, Size count)
t_thrd.sharestoragexlogcopyer_cxt.readSegNo),
startoff)));
}
t_thrd.sharestoragexlogcopyer_cxt.readOff = startoff;
t_thrd.sharestoragexlogcopyer_cxt.readOff = startoff; // 更新读取偏移量
}
/* How many bytes are within this segment? */
// 计算在当前XLOG段中要读取的字节数
if (nbytes > (XLogSegSize - startoff)) {
segbytes = XLogSegSize - startoff;
} else {
segbytes = nbytes;
}
// 报告WAL读取等待事件
pgstat_report_waitevent(WAIT_EVENT_WAL_READ);
// 从文件中读取数据
readbytes = read(t_thrd.sharestoragexlogcopyer_cxt.readFile, p, segbytes);
// 报告WAL读取完成事件
pgstat_report_waitevent(WAIT_EVENT_END);
// 检查读取是否失败
if (readbytes <= 0) {
// 读取失败，关闭文件并报告错误
(void)close(t_thrd.sharestoragexlogcopyer_cxt.readFile);
t_thrd.sharestoragexlogcopyer_cxt.readFile = -1;
t_thrd.sharestoragexlogcopyer_cxt.readFile = -1; // 标志读取失败
ereport(ERROR,
(errcode_for_file_access(),
errmsg("could not read from log segment %s, offset %u, length %lu: %m",
@ -201,226 +222,284 @@ void LocalXLogRead(char *buf, XLogRecPtr startptr, Size count)
t_thrd.sharestoragexlogcopyer_cxt.readOff += readbytes;
nbytes -= readbytes;
p += readbytes;
p += readbytes; // 更新缓冲区指针
}
}
// 用于通知等待同步的后端（如流复制的从节点）有新的日志位置需要同步。
void NotifySyncWaiters(XLogRecPtr newPos)
{
volatile WalSndCtlData *walsndctl = t_thrd.walsender_cxt.WalSndCtl;
volatile WalSndCtlData *walsndctl = t_thrd.walsender_cxt.WalSndCtl;
// 获取同步复制锁以确保原子性地更新同步等待状态
(void)LWLockAcquire(SyncRepLock, LW_EXCLUSIVE);
/*
* Set the lsn first so that when we wake backends they will release up to
* this location.
* 检查当前位置是否大于等待接收、等待写入和等待刷新的LSN，
* 如果是，则更新对应的LSN并唤醒相应的等待队列。
*/
// 如果新位置大于等待接收的LSN，则更新并唤醒等待接收的队列
if (XLByteLT(walsndctl->lsn[SYNC_REP_WAIT_RECEIVE], newPos)) {
walsndctl->lsn[SYNC_REP_WAIT_RECEIVE] = newPos;
(void)SyncRepWakeQueue(false, SYNC_REP_WAIT_RECEIVE);
}
// 如果新位置大于等待写入的LSN，则更新并唤醒等待写入的队列
if (XLByteLT(walsndctl->lsn[SYNC_REP_WAIT_WRITE], newPos)) {
walsndctl->lsn[SYNC_REP_WAIT_WRITE] = newPos;
(void)SyncRepWakeQueue(false, SYNC_REP_WAIT_WRITE);
}
// 如果新位置大于等待刷新的LSN，则更新并唤醒等待刷新的队列
if (XLByteLT(walsndctl->lsn[SYNC_REP_WAIT_FLUSH], newPos)) {
walsndctl->lsn[SYNC_REP_WAIT_FLUSH] = newPos;
(void)SyncRepWakeQueue(false, SYNC_REP_WAIT_FLUSH);
}
// 释放同步复制锁
LWLockRelease(SyncRepLock);
}
/*
* 该函数用于将新的LSN信息写入共享存储，确保相关控制信息和数据持久化，
* 并通过日志和通知机制，确保同步等待的后端能够得知新的LSN可用，从而进行相应的同步操作。
*/
void PushCtlLsn(XLogRecPtr flushPtr)
{
ShareStorageXLogCtl *sharestorageCtl = g_instance.xlog_cxt.shareStorageXLogCtl;
// 将flushPtr设置为共享存储XLog控制结构中的insertHead字段。
sharestorageCtl->insertHead = flushPtr;
// 如果insertHead和insertTail之间的距离大于xlogFileSize，调整insertTail的值
if ((sharestorageCtl->insertHead - sharestorageCtl->insertTail) > sharestorageCtl->xlogFileSize) {
sharestorageCtl->insertTail = sharestorageCtl->insertHead - sharestorageCtl->xlogFileSize;
}
// 计算共享存储XLog控制结构的CRC校验码
sharestorageCtl->crc = CalShareStorageCtlInfoCrc(sharestorageCtl);
// 更新共享存储XLog控制信息
UpdateShareStorageCtlInfo(sharestorageCtl);
// 将XLOG数据刷写到共享存储
FsyncXlogToShareStorage();
// 根据HaModuleDebug设置日志级别
int mode = u_sess->attr.attr_storage.HaModuleDebug ? LOG : DEBUG2;
// 以指定的日志级别输出消息，显示insertHead的LSN
const uint32 shftSz = 32;
ereport(mode, (errmsg("[PushCtlLsn]insertHead lsn %08X/%08X\n", (uint32)(flushPtr >> shftSz), (uint32)flushPtr)));
// 通知等待同步的后端，有新的LSN可用
NotifySyncWaiters(flushPtr);
}
// 该函数用于在给定的约束下计算要写入共享存储的实际数据长度，
// 确保数据写入的位置和边界符合块大小和共享存储缓冲区大小的要求。
static inline int CalcWriteLen(XLogRecPtr startWrite, XLogRecPtr endPtr)
{
// 确保startWrite的偏移量是XLOG_BLCKSZ的倍数，即起始位置为块的开始，如果断言失败，就会终止程序
Assert((startWrite % XLOG_BLCKSZ) == 0);
// 计算对齐的写入结束位置，使其与ShareStorageBufSize对齐
XLogRecPtr alignWriteEnd = startWrite - startWrite % ShareStorageBufSize + ShareStorageBufSize;
// 如果对齐后的写入结束位置大于endPtr，执行以下操作
if (alignWriteEnd > endPtr) {
// 使用TYPEALIGN计算对齐的实际写入结束位置
XLogRecPtr ActualCopyEnd = TYPEALIGN(XLOG_BLCKSZ, endPtr);
// 计算要写入的数据长度，并将其转换为int类型返回
return static_cast<int>(ActualCopyEnd - startWrite);
} else {
// 如果对齐后的写入结束位置不大于endPtr，计算要写入的数据长度，并将其转换为 int类型返回
return static_cast<int>(alignWriteEnd - startWrite);
}
}
// 该函数用于获取可以进行覆写的最大XLog位置。
XLogRecPtr GetMaxPosCanOverWrite()
{
// 如果不在Postmaster进程中，直接返回最大的XLog位置
if (!IsUnderPostmaster) {
return MAX_XLOG_REC_PTR;
}
// 初始化最大刷新位置为无效位置
XLogRecPtr maxFlush = InvalidXLogRecPtr;
// 遍历所有的WAL发送者，找到最大的刷新位置
for (int i = 0; i < g_instance.attr.attr_storage.max_wal_senders; i++) {
/* use volatile pointer to prevent code rearrangement */
volatile WalSnd *walsnd = &t_thrd.walsender_cxt.WalSndCtl->walsnds[i];
SpinLockAcquire(&walsnd->mutex);
// 如果WAL发送者的进程ID不为0且peer_role为STANDBY_CLUSTER_MODE
if (walsnd->pid != 0 && walsnd->peer_role == STANDBY_CLUSTER_MODE) {
if (XLByteLT(maxFlush, walsnd->flush)) {
maxFlush = walsnd->flush;
maxFlush = walsnd->flush; // 更新最大刷新位置
}
}
SpinLockRelease(&walsnd->mutex);
}
// 如果没有找到有效的刷新位置，返回最大的XLog位置
if (maxFlush == InvalidXLogRecPtr) {
return MAX_XLOG_REC_PTR;
}
// 返回可进行覆写的最大位置，即xlog_file_size加上最大刷新位置
return g_instance.attr.attr_storage.xlog_file_size + maxFlush;
}
void AddXLogPageHeader(char* buf, XLogRecPtr startWrite, int writeLen, XLogRecPtr endPtr)
// 该函数用于确保写入的每个数据块都有正确的XLog页面头部，以便在后续的操作中能够正确识别和处理写入的WAL数据。
void AddXLogPageHeader(char *buf, XLogRecPtr startWrite, int writeLen, XLogRecPtr endPtr)
{
int offset = 0;
while (offset < writeLen) {
// 使用循环向每个写入的块添加XLog页面头部
while (offset < writeLen) {
// 计算当前偏移处的XLog页面头部
XLogPageHeader xlogPageHeader = (XLogPageHeader)(buf + offset);
// 设置XLog页面头部的总长度为XLOG_BLCKSZ（即块的大小）
xlogPageHeader->xlp_total_len = XLOG_BLCKSZ;
// 移动偏移，准备处理下一个块
offset += XLOG_BLCKSZ;
}
// 如果endPtr小于（不包括等于）startWrite + writeLen，表示写入的数据跨越了一个块的边界
if (XLByteLT(endPtr, startWrite + writeLen)) {
// 计算跨越的部分在最后一个块中的长度
XLogPageHeader xlogPageHeader = (XLogPageHeader)(buf + writeLen - XLOG_BLCKSZ);
// 设置最后一个块中XLog页面头部的总长度，以确保不超过写入的数据的边界
xlogPageHeader->xlp_total_len = endPtr % XLOG_BLCKSZ;
}
}
// 该函数用于执行XLog数据的拷贝操作
void DoXlogCopy(XLogRecPtr targetPtr)
{
uint64 writeLength = 0;
ShareStorageXLogCtl *sharestorageCtl = g_instance.xlog_cxt.shareStorageXLogCtl;
// 断言目标位置有效
Assert(targetPtr != InvalidXLogRecPtr);
// 如果目标位置小于等于当前插入头位置，无需进行XLog拷贝
if (XLByteLE(targetPtr, sharestorageCtl->insertHead)) {
return;
}
// 计算起始写入位置（整块对齐）
XLogRecPtr startWrite = sharestorageCtl->insertHead - (sharestorageCtl->insertHead % XLOG_BLCKSZ);
// 循环进行XLog拷贝
while (XLByteLT(startWrite, targetPtr)) {
// 计算本次写入的长度
int writeLen = CalcWriteLen(startWrite, targetPtr);
// 从本地XLog中读取数据
LocalXLogRead(t_thrd.sharestoragexlogcopyer_cxt.buf, startWrite, static_cast<Size>(writeLen));
// 为写入数据块添加XLog页面头部
AddXLogPageHeader(t_thrd.sharestoragexlogcopyer_cxt.buf, startWrite, writeLen, targetPtr);
// 将数据块写入共享存储
(void)WriteXlogToShareStorage(startWrite, t_thrd.sharestoragexlogcopyer_cxt.buf, writeLen);
// 更新起始写入位置，并累计写入长度
startWrite += writeLen;
writeLength += writeLen;
// 如果累计写入长度达到了一个XLog段的大小，或者目标位置小于起始写入位置，推进控制LSN并重置累计写入长度
if ((writeLength >= XLogSegSize) || (XLByteLT(targetPtr, startWrite))) {
PushCtlLsn(XLByteLE(startWrite, targetPtr) ? startWrite : targetPtr);
writeLength = 0;
}
}
// 最后如果还有未推进的LSN，进行推进
if (writeLength > 0) {
PushCtlLsn(targetPtr);
}
}
// 获取锁
static bool GetLock()
{
// 如果XLog锁文件路径为NULL，表示无需获取锁，直接返回成功
if (g_instance.attr.attr_storage.xlog_lock_file_path == NULL) {
return true;
}
// 尝试获取NAS写锁，如果成功获取锁则返回成功
if (LockNasWriteFile(g_instance.xlog_cxt.shareStorageLockFd)) {
return true;
}
// 如果获取锁失败，报告致命错误，指明无法锁定XLog锁文件
ereport(FATAL, (errmsg("could not lock lock file(%d) %s", g_instance.xlog_cxt.shareStorageLockFd,
g_instance.attr.attr_storage.xlog_lock_file_path)));
g_instance.attr.attr_storage.xlog_lock_file_path)));
return false;
}
// 释放锁
static bool ReleaseLock()
{
// 如果XLog锁文件路径为NULL，表示无需释放锁，直接返回成功
if (g_instance.attr.attr_storage.xlog_lock_file_path == NULL) {
return true;
}
// 尝试释放NAS写锁，如果成功释放锁则返回成功
if (UnlockNasWriteFile(g_instance.xlog_cxt.shareStorageLockFd)) {
return true;
}
// 如果释放锁失败，报告致命错误，指明无法解锁XLog锁文件
ereport(FATAL, (errmsg("could not unlock lock file(%d) %s", g_instance.xlog_cxt.shareStorageLockFd,
g_instance.attr.attr_storage.xlog_lock_file_path)));
g_instance.attr.attr_storage.xlog_lock_file_path)));
return false;
}
// 检查并拷贝XLog数据
bool CheckAndCopyXLog(bool forceCopy)
{
ShareStorageXLogCtl *sharestorageCtl = g_instance.xlog_cxt.shareStorageXLogCtl;
do {
XLogRecPtr localFlush = InvalidXLogRecPtr;
// 如果当前为主备模式，获取主备同步位置，否则获取本地刷写位置
if (IS_SHARED_STORAGE_MAIN_STANDBY_MODE) {
localFlush = GetFlushMainStandby();
} else {
localFlush = GetFlushRecPtr();
}
// 如果本地刷写位置小于等于共享存储中插入位置，表示无需拷贝
if (XLByteLE(localFlush, sharestorageCtl->insertHead)) {
return true;
}
// 获取可以覆写的最大位置
XLogRecPtr maxPosCanWrite = GetMaxPosCanOverWrite();
// 如果共享存储插入位置小于最大可覆写位置，进行XLog拷贝
if (XLByteLT(sharestorageCtl->insertHead, maxPosCanWrite)) {
// 计算期望的拷贝位置，为本地刷写位置与最大可覆写位置中较小的值
XLogRecPtr expectPos = XLByteLT(localFlush, maxPosCanWrite) ? localFlush : maxPosCanWrite;
// 计算期望的拷贝位置，为本地刷写位置与最大可覆写位置中较小的值
if ((expectPos - sharestorageCtl->insertHead) > MAX_SIZE_CAN_COPY_TO_SHARE) {
expectPos = sharestorageCtl->insertHead + MAX_SIZE_CAN_COPY_TO_SHARE;
}
// 执行XLog数据拷贝
DoXlogCopy(expectPos);
}
} while (forceCopy && GetLock());
return false;
return false; // 返回false表示未成功执行拷贝
}
// 该函数用于关闭共享存储的XLog拷贝操作
void ShutdownShareStorageXLogCopy()
{
// 如果未配置XLog文件路径，直接返回
if (g_instance.attr.attr_storage.xlog_file_path == NULL) {
return;
}
// 分配和初始化拷贝缓冲区
if (t_thrd.sharestoragexlogcopyer_cxt.originBuf == NULL) {
t_thrd.sharestoragexlogcopyer_cxt.originBuf =
(char *)palloc(ShareStorageBufSize + g_instance.xlog_cxt.shareStorageopCtl.blkSize);
t_thrd.sharestoragexlogcopyer_cxt.buf = (char *)TYPEALIGN(g_instance.xlog_cxt.shareStorageopCtl.blkSize,
t_thrd.sharestoragexlogcopyer_cxt.originBuf);
}
// 读取共享存储控制信息并执行XLog拷贝
ReadShareStorageCtlInfo(g_instance.xlog_cxt.shareStorageXLogCtl);
CheckAndCopyXLog(true);
// 如果读取文件描述符有效，关闭文件
if (t_thrd.sharestoragexlogcopyer_cxt.readFile >= 0) {
(void)close(t_thrd.sharestoragexlogcopyer_cxt.readFile);
}
// 释放拷贝缓冲区的内存
if (t_thrd.sharestoragexlogcopyer_cxt.originBuf != NULL) {
pfree(t_thrd.sharestoragexlogcopyer_cxt.originBuf);
t_thrd.sharestoragexlogcopyer_cxt.originBuf = NULL;
t_thrd.sharestoragexlogcopyer_cxt.buf = NULL;
}
// 清除相关的Latch和标志
g_instance.proc_base->ShareStoragexlogCopyerLatch = NULL;
// 打印日志，表示XLog拷贝已停止
ereport(LOG, (errmsg("stopped xlog copy at %X/%X",
(uint32)(g_instance.xlog_cxt.shareStorageXLogCtl->insertHead >> PTR_PRINT_SHIFT_SIZE),
(uint32)g_instance.xlog_cxt.shareStorageXLogCtl->insertHead)));
(uint32)(g_instance.xlog_cxt.shareStorageXLogCtl->insertHead >> PTR_PRINT_SHIFT_SIZE),
(uint32)g_instance.xlog_cxt.shareStorageXLogCtl->insertHead)));
}
// 该函数用于检查文件大小是否可以更新
bool FileSizeCanUpdate()
{
// 获取结束检查点的位置
XLogRecPtr endCheckpointPtr = sizeof(CheckPointPlus) + t_thrd.shemem_ptr_cxt.ControlFile->checkPoint;
// 获取共享存储控制信息
ShareStorageXLogCtl *ctlInfo = g_instance.xlog_cxt.shareStorageXLogCtl;
// 计算检查点位置和插入位置的偏移量
uint64 checkpointPos = endCheckpointPtr % ctlInfo->xlogFileSize;
uint64 headPos = ctlInfo->insertHead % ctlInfo->xlogFileSize;
// 如果满足一定条件，返回true表示文件大小可以更新，否则返回false
if ((XLByteLE(endCheckpointPtr, ctlInfo->insertHead) && checkpointPos <= headPos) ||
XLByteLT(ctlInfo->insertHead, endCheckpointPtr)) {
return true;
@ -428,32 +507,37 @@ bool FileSizeCanUpdate()
return false;
}
// 该函数用于检查共享存储控制信息
void CheckShareStorageCtlInfo(XLogRecPtr localEnd)
{
// 如果不处于共享存储模式，直接返回
if (!IS_SHARED_STORAGE_MODE) {
return;
}
// 获取共享存储的控制信息
ShareStorageXLogCtl *ctlInfo = g_instance.xlog_cxt.shareStorageXLogCtl;
ReadShareStorageCtlInfo(ctlInfo);
// 检查系统标识是否一致，不一致时报错
if (ctlInfo->systemIdentifier != GetSystemIdentifier()) {
ereport(FATAL, (errmsg("database system version is different between shared storage %lu and local %lu",
ctlInfo->systemIdentifier, GetSystemIdentifier())));
}
// 初始化
uint32 shiftSize = 32;
XLogRecPtr shareStorageLatestRecordStart = InvalidXLogRecPtr;
XLogRecPtr shareStorageLatestRecordStart = InvalidXLogRecPtr; // 初始化为无效的位置
int shareStorageLatestRecordLen;
pg_crc32 shareStorageLatestRecordCrc;
// 在共享存储中查找最后一个记录的信息
FindLastRecordCheckInfoOnShareStorage(&shareStorageLatestRecordStart, &shareStorageLatestRecordCrc,
&shareStorageLatestRecordLen);
// 断言共享存储最新记录的起始位置加长度小于等于插入位置，确保共享存储数据正确性
Assert(XLByteLE(shareStorageLatestRecordStart + shareStorageLatestRecordLen, ctlInfo->insertHead));
// 如果本地的结束位置小于共享存储的插入位置
if (XLByteLT(localEnd, ctlInfo->insertHead)) {
// 如果共享存储最新记录的起始位置是无效的，或者本地的结束位置和最新记录的起始位置加对齐长度不一致
if (XLByteEQ(shareStorageLatestRecordStart, InvalidXLogRecPtr) ||
!XLByteEQ(localEnd, shareStorageLatestRecordStart + MAXALIGN(shareStorageLatestRecordLen))) {
// 报错，表示本地头部小于共享存储头部
ereport(FATAL, (errmsg("the local's head is smaller than The shared storage's head"),
errdetail("The shared storage's head %X/%X, the local's head is %X/%X. "
"lastrecord on share storage: startlsn:%X/%X crc %u, len %d",
@ -465,10 +549,12 @@ void CheckShareStorageCtlInfo(XLogRecPtr localEnd)
shareStorageLatestRecordLen)));
}
// 打印日志，表示修改共享存储头部
ereport(LOG,
(errmsg("modify share storage head from %X/%X to %X/%X",
static_cast<uint32>(ctlInfo->insertHead >> shiftSize), static_cast<uint32>(ctlInfo->insertHead),
static_cast<uint32>(localEnd >> shiftSize), static_cast<uint32>(localEnd))));
// 修改共享存储控制信息的插入位置和CRC，然后刷新到共享存储
ctlInfo->insertHead = localEnd;
ctlInfo->crc = CalShareStorageCtlInfoCrc(ctlInfo);
UpdateShareStorageCtlInfo(ctlInfo);
@ -476,10 +562,12 @@ void CheckShareStorageCtlInfo(XLogRecPtr localEnd)
} else {
char path[MAXPGPATH];
XLogSegNo sendSegNo;
// 将插入位置转换为XLog段号，并获取路径
XLByteToSeg(ctlInfo->insertHead, sendSegNo);
XLogFilePath(path, MAXPGPATH, t_thrd.xlog_cxt.ThisTimeLineID, sendSegNo);
struct stat stat_buf;
// 如果路径对应的文件不存在，报错，表示本地尾部大于共享存储头部
if (stat(path, &stat_buf) != 0) {
ereport(FATAL, (errmsg("the local's tail is bigger than The shared storage's head %X/%X",
static_cast<uint32>(ctlInfo->insertHead >> shiftSize),
@ -488,7 +576,9 @@ void CheckShareStorageCtlInfo(XLogRecPtr localEnd)
bool crcValid = false;
// 获取共享存储最新记录的CRC校验值，用于比较本地记录的校验值
pg_crc32 localCheckCrc = GetXlogRecordCrc(shareStorageLatestRecordStart, crcValid, XLogPageRead, 0);
// 如果校验值不匹配，报错，表示共享存储请求的LSN的
if (shareStorageLatestRecordCrc != localCheckCrc) {
ereport(FATAL, (errmsg("shared storage request lsn[%X/%X]'s crc mismatched (share, local):[%u,%u].",
static_cast<uint32>(shareStorageLatestRecordStart >> shiftSize),
@ -497,8 +587,8 @@ void CheckShareStorageCtlInfo(XLogRecPtr localEnd)
}
}
uint32 localTerm = Max(g_instance.comm_cxt.localinfo_cxt.term_from_file,
g_instance.comm_cxt.localinfo_cxt.term_from_xlog);
uint32 localTerm =
Max(g_instance.comm_cxt.localinfo_cxt.term_from_file, g_instance.comm_cxt.localinfo_cxt.term_from_xlog);
if (localTerm > ctlInfo->term) {
ctlInfo->term = localTerm;
ctlInfo->crc = CalShareStorageCtlInfoCrc(ctlInfo);
@ -506,48 +596,55 @@ void CheckShareStorageCtlInfo(XLogRecPtr localEnd)
FsyncXlogToShareStorage();
}
}
// 更新共享存储控制信息
void UpdateShareStorageCtlInfo()
{
bool changed = false;
ShareStorageXLogCtl *ctlInfo = g_instance.xlog_cxt.shareStorageXLogCtl;
ReadShareStorageCtlInfo(ctlInfo);
// 检查版本号是否一致，如果不一致则更新版本号，并设置changed标志为true
if (ctlInfo->version != CURRENT_SHARE_STORAGE_CTL_VERSION) {
ctlInfo->version = CURRENT_SHARE_STORAGE_CTL_VERSION;
changed = true;
}
// 检查长度是否一致，如果不一致则更新长度，并设置changed标志为true
if (ctlInfo->length != SizeOfShareStorageXLogCtl) {
ctlInfo->length = SizeOfShareStorageXLogCtl;
changed = true;
}
// 检查xlog文件大小是否一致，如果不一致则更新xlog文件大小，并设置changed标志为true
if (ctlInfo->xlogFileSize != (uint64)g_instance.attr.attr_storage.xlog_file_size) {
Assert(g_instance.xlog_cxt.shareStorageopCtl.xlogFileSize == ctlInfo->xlogFileSize);
// 如果文件大小无法更新，报错
if (!FileSizeCanUpdate()) {
ereport(FATAL, (errmsg("could not update share storage size."),
errdetail("current size:%lu, new size:%lu", ctlInfo->xlogFileSize,
g_instance.attr.attr_storage.xlog_file_size)));
}
// 更新共享存储操作控制中的xlog文件大小，并将控制信息中的xlog文件大小设置为新值
g_instance.xlog_cxt.shareStorageopCtl.xlogFileSize = g_instance.attr.attr_storage.xlog_file_size;
ctlInfo->xlogFileSize = g_instance.xlog_cxt.shareStorageopCtl.xlogFileSize;
changed = true;
}
// 如果有更新，则重新计算CRC校验值，然后更新控制信息
if (changed) {
ctlInfo->crc = CalShareStorageCtlInfoCrc(ctlInfo);
UpdateShareStorageCtlInfo(ctlInfo);
}
}
// 共享存储XLog复制后台退出及清理函数
static void SharedStorageXlogCopyBackendQuitAndClean(int code, Datum arg)
{
// 打印日志，表示解锁文件
ereport(LOG, (errmsg("SharedStorageXlogCopyBackendMain unlock file(%d) %s", g_instance.xlog_cxt.shareStorageLockFd,
g_instance.attr.attr_storage.xlog_lock_file_path)));
ReleaseLock();
g_instance.attr.attr_storage.xlog_lock_file_path)));
ReleaseLock(); // 释放锁
}
// 这段代码的目的是确保线程能够正确地响应各种系统和自定义的信号，以保证程序的稳定运行和正确退出。
static void InitThreadSignal()
{
(void)gspqsignal(SIGHUP, SharedStorageXlogCopyBackendSigHupHandler); /* reload config file */
@ -570,31 +667,37 @@ static void InitThreadSignal()
gs_signal_setmask(&t_thrd.libpq_cxt.UnBlockSig, NULL);
(void)gs_signal_unblock_sigusr2();
}
// 该函数主要用于在后台进程中执行共享存储的 XLog 复制操作
void SharedStorageXlogCopyBackendMain(void)
{
// 初始化线程信号处理
InitThreadSignal();
if (IS_SHARED_STORAGE_MAIN_STANDBY_MODE) {
// 检查共享存储的写锁是否已经获取，如果没有获取到写锁，则会尝试获取写锁。
CheckShareStorageWriteLock();
}
// 如果处于恢复状态且不是主-备模式，记录一条日志并退出
if (RecoveryInProgress() && !IS_SHARED_STORAGE_MAIN_STANDBY_MODE) {
ereport(LOG, (errmsg("stopped xlog copy in recovery")));
proc_exit(0); /* done */
}
// 注册进程退出时的清理函数
on_proc_exit(SharedStorageXlogCopyBackendQuitAndClean, 0);
// 设置进程 latch，用于标识进程活动
g_instance.proc_base->ShareStoragexlogCopyerLatch = &t_thrd.proc->procLatch;
// 分配用于复制 XLog 数据的缓冲区
t_thrd.sharestoragexlogcopyer_cxt.originBuf =
(char *)palloc(ShareStorageBufSize + g_instance.xlog_cxt.shareStorageopCtl.blkSize);
t_thrd.sharestoragexlogcopyer_cxt.buf = (char *)TYPEALIGN(g_instance.xlog_cxt.shareStorageopCtl.blkSize,
t_thrd.sharestoragexlogcopyer_cxt.originBuf);
t_thrd.sharestoragexlogcopyer_cxt.buf =
(char *)TYPEALIGN(g_instance.xlog_cxt.shareStorageopCtl.blkSize, t_thrd.sharestoragexlogcopyer_cxt.originBuf);
// 更新共享存储控制信息
UpdateShareStorageCtlInfo();
// 记录 XLog 复制操作的开始
ereport(LOG, (errmsg("start xlog copy at %X/%X",
(uint32)(g_instance.xlog_cxt.shareStorageXLogCtl->insertHead >> PTR_PRINT_SHIFT_SIZE),
(uint32)g_instance.xlog_cxt.shareStorageXLogCtl->insertHead)));
(uint32)(g_instance.xlog_cxt.shareStorageXLogCtl->insertHead >> PTR_PRINT_SHIFT_SIZE),
(uint32)g_instance.xlog_cxt.shareStorageXLogCtl->insertHead)));
// 根据模式确定当前时间线 ID
if (IS_SHARED_STORAGE_MAIN_STANDBY_MODE) {
t_thrd.xlog_cxt.ThisTimeLineID = GetRecoveryTargetTLI();
} else {
@ -604,47 +707,54 @@ void SharedStorageXlogCopyBackendMain(void)
uint64 checkLockCount = 0;
pgstat_report_activity(STATE_IDLE, NULL);
// 主循环，处理 XLog 复制和后台活动
for (;;) {
ResetLatch(&t_thrd.proc->procLatch);
pgstat_report_activity(STATE_RUNNING, NULL);
// 每次循环定期检查并获取锁
++checkLockCount;
if (checkLockCount & CHECK_LOCK_INTERVAL) {
GetLock();
}
// 检查是否有关闭请求，如果有，执行相关清理操作并退出
if (t_thrd.sharestoragexlogcopyer_cxt.shutdown_requested) {
t_thrd.sharestoragexlogcopyer_cxt.shutdown_requested = false;
g_instance.proc_base->ShareStoragexlogCopyerLatch = NULL;
ShutdownShareStorageXLogCopy();
proc_exit(0); /* done */
proc_exit(0); /*终止当前进程或线程*/
}
// 检查是否收到 SIGHUP 信号，如果是，重新加载配置文件
if (t_thrd.sharestoragexlogcopyer_cxt.got_SIGHUP) {
t_thrd.sharestoragexlogcopyer_cxt.got_SIGHUP = false;
ProcessConfigFile(PGC_SIGHUP);
}
// 检查并复制 XLog 数据，如果需要等待则继续下一次循环
bool waitCopy = CheckAndCopyXLog(false);
if (waitCopy) {
continue;
}
// 报告进程状态为空闲并等待活动
pgstat_report_activity(STATE_IDLE, NULL);
const long sleepTime = 1000L;
(void)WaitLatch(&t_thrd.proc->procLatch, WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH, sleepTime);
}
// 清理操作并退出进程
g_instance.proc_base->ShareStoragexlogCopyerLatch = NULL;
ereport(LOG, (errmsg("stopped xlog copy at %X/%X",
(uint32)(g_instance.xlog_cxt.shareStorageXLogCtl->insertHead >> PTR_PRINT_SHIFT_SIZE),
(uint32)g_instance.xlog_cxt.shareStorageXLogCtl->insertHead)));
(uint32)(g_instance.xlog_cxt.shareStorageXLogCtl->insertHead >> PTR_PRINT_SHIFT_SIZE),
(uint32)g_instance.xlog_cxt.shareStorageXLogCtl->insertHead)));
proc_exit(0);
}
// 此函数的作用是唤醒名为 XLogCopyerBackend 的后台线程。
void WakeUpXLogCopyerBackend()
{
if (g_instance.proc_base->ShareStoragexlogCopyerLatch != NULL) {
// 用于将闩锁设置为激活状态，从而唤醒等待中的 XLogCopyerBackend 线程继续执行。
SetLatch(g_instance.proc_base->ShareStoragexlogCopyerLatch);
}
}
@ -714,8 +824,8 @@ bool XLogOverwriteFromLocal(bool force)
ReadShareStorageCtlInfo(ctlInfo);
if (ctlInfo->systemIdentifier != GetSystemIdentifier()) {
ereport(WARNING, (errmsg("database system version is different between shared storage and local"),
errdetail("The shared storage's system version is %lu, the local's system version is %lu.",
ctlInfo->systemIdentifier, GetSystemIdentifier())));
errdetail("The shared storage's system version is %lu, the local's system version is %lu.",
ctlInfo->systemIdentifier, GetSystemIdentifier())));
if (!force) {
return false;
}
@ -826,8 +936,8 @@ static bool XLogCopyWrite(char *buf, int nbytes, XLogRecPtr recptr)
if (copyFile >= 0) {
if (close(copyFile) != 0) {
ereport(WARNING, (errcode_for_file_access(),
errmsg("could not close log file %s: %m", XLogFileNameP(copyFileTLI, copySegNo))));
ereport(WARNING, (errcode_for_file_access(), errmsg("could not close log file %s: %m",
XLogFileNameP(copyFileTLI, copySegNo))));
return false;
}
}
@ -852,7 +962,7 @@ static bool XLogCopyWrite(char *buf, int nbytes, XLogRecPtr recptr)
if (copyOff != (uint32)startoff) {
if (lseek(copyFile, (off_t)startoff, SEEK_SET) < 0) {
ereport(WARNING, (errcode_for_file_access(), errmsg("could not seek in log file %s to offset %lu: %m",
XLogFileNameP(copyFileTLI, copySegNo), startoff)));
XLogFileNameP(copyFileTLI, copySegNo), startoff)));
return false;
}
copyOff = startoff;
@ -868,16 +978,16 @@ static bool XLogCopyWrite(char *buf, int nbytes, XLogRecPtr recptr)
errno = ENOSPC;
}
ereport(WARNING, (errcode_for_file_access(),
errmsg("could not write to log file %s at offset %u, length %lu: %m",
XLogFileNameP(copyFileTLI, copySegNo), copyOff, INT2ULONG(segbytes))));
errmsg("could not write to log file %s at offset %u, length %lu: %m",
XLogFileNameP(copyFileTLI, copySegNo), copyOff, INT2ULONG(segbytes))));
return false;
}
if (copyOff == (uint32)0 && segbytes >= (int)sizeof(XLogPageHeaderData)) {
if (((XLogPageHeader)buf)->xlp_magic == XLOG_PAGE_MAGIC &&
(copySegNo * XLogSegSize) != ((XLogPageHeader)buf)->xlp_pageaddr) {
ereport(WARNING, (errcode_for_file_access(),
errmsg("unexpected page addr %lu of log file %s", ((XLogPageHeader)buf)->xlp_pageaddr,
XLogFileNameP(copyFileTLI, copySegNo))));
errmsg("unexpected page addr %lu of log file %s", ((XLogPageHeader)buf)->xlp_pageaddr,
XLogFileNameP(copyFileTLI, copySegNo))));
return false;
}
}
@ -912,7 +1022,7 @@ static bool DoXLogCopyFromShare(XLogRecPtr copyStart)
xlogreader = XLogReaderAllocate(SharedStorageXLogPageRead, 0, g_instance.xlog_cxt.shareStorageopCtl.blkSize);
if (xlogreader == NULL) {
ereport(WARNING, (errcode(ERRCODE_OUT_OF_MEMORY), errmsg("out of memory"),
errdetail("Failed while allocating an XLog reading processor")));
errdetail("Failed while allocating an XLog reading processor")));
return false;
}
xlogreader->system_identifier = GetSystemIdentifier();
@ -937,7 +1047,7 @@ static bool DoXLogCopyFromShare(XLogRecPtr copyStart)
if (copyFile >= 0) {
if (close(copyFile) != 0) {
ereport(WARNING, (errcode_for_file_access(),
errmsg("could not close log file %s: %m", XLogFileNameP(copyFileTLI, copySegNo))));
errmsg("could not close log file %s: %m", XLogFileNameP(copyFileTLI, copySegNo))));
return false;
}
}
@ -967,8 +1077,8 @@ bool XLogOverwriteFromShare()
if (ctlInfo->systemIdentifier != GetSystemIdentifier()) {
ereport(WARNING, (errmsg("database system version is different between shared storage and local"),
errdetail("The shared storage's system version is %lu, the local's system version is %lu.",
ctlInfo->systemIdentifier, GetSystemIdentifier())));
errdetail("The shared storage's system version is %lu, the local's system version is %lu.",
ctlInfo->systemIdentifier, GetSystemIdentifier())));
return false;
}
/* The local log is consistent with the local data and cannot be simply overwritten from the shared storage.
@ -1062,4 +1172,3 @@ bool UnlockNasWriteFile(int fd)
ereport(WARNING, (errcode_for_file_access(), errmsg("could not unlock lock file : %m")));
return false;
}