hyt15810656168
/
openGauss-server
forked from huawei/openGauss-server
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
openGauss-server/src/gausskernel/storage/remote/remote_adapter.cpp

973 lines
41 KiB
C++
Executable File
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

/*
* Copyright (c) 2020 Huawei Technologies Co.,Ltd.
*
* openGauss is licensed under Mulan PSL v2.
* You can use this software according to the terms and conditions of the Mulan PSL v2.
* You may obtain a copy of Mulan PSL v2 at:
*
* http://license.coscl.org.cn/MulanPSL2
*
* THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
* EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
* MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
* See the Mulan PSL v2 for more details.
* ---------------------------------------------------------------------------------------
*
* remote_adapter.cpp
* using simple C API interface for rpc call PG founction
* Don't include any of RPC header file.
*
* IDENTIFICATION
* src/gausskernel/storage/remote/remote_adapter.cpp
*
* ---------------------------------------------------------------------------------------
*/
#include "postgres.h"
#include "knl/knl_variable.h"
#include "catalog/pg_authid.h"
#include "catalog/pg_type.h"
#include "access/xlog.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "funcapi.h"
#include "postmaster/postmaster.h"
#include "storage/smgr/segment_internal.h"
#include "storage/smgr/segment.h"
#include "storage/custorage.h"
#include "storage/ipc.h"
#include "storage/remote_adapter.h"
#include "libpq/pqformat.h"
#include "utils/guc.h"
#include "utils/memutils.h"
#include "utils/builtins.h"
#include "utils/aiomem.h"
const int DEFAULT_WAIT_TIMES = 60;
int ReadFileForRemote(RemoteReadFileKey *key, XLogRecPtr lsn, bytea** fileData, int timeout);
int ReadCOrCsnFileForRemote(RelFileNode rnode, bytea** fileData);
/*
* @Description: wait lsn to replay
* @IN primary_insert_lsn: remote request lsn
* @Return: remote read error code
* @See also:
*/
//用于等待指定LSN（日志序列号）被回放到本地
// 常见的应用场景是在流复制（Streaming Replication）中，等待从节点接收并成功应用指定的事务日志
// primary_insert_lsn：远程请求的LSN（日志序列号）
int XLogWaitForReplay(uint64 primary_insert_lsn, int timeout = DEFAULT_WAIT_TIMES)
{
int wait_times = 0;
/* local replayed lsn */
//获取本地正在回放的LSN
XLogRecPtr standby_replay_lsn = GetXLogReplayRecPtr(NULL, NULL);
/* if primary_insert_lsn > standby_replay_lsn then need wait */
//循环判断如果primary_insert_lsn大于standby_replay_lsn，则需要继续等待
while (!XLByteLE(primary_insert_lsn, standby_replay_lsn)) {
/* if sleep to much times */
//判断等待次数是否超过超时时间
if (wait_times >= timeout) {
//如果超过则返回错误并打印相应日志信息
ereport(LOG,
(errmodule(MOD_REMOTE),
errmsg("replay slow. requre lsn %X/%X, replayed lsn %X/%X",
(uint32)(primary_insert_lsn >> 32),
(uint32)primary_insert_lsn,
(uint32)(standby_replay_lsn >> 32),
(uint32)standby_replay_lsn)));
//等待超时，则返回REMOTE_READ_NEED_WAIT（远程读取需要等待）
return REMOTE_READ_NEED_WAIT;
}
/* sleep 1s */
//进行1秒的延迟
pg_usleep(1000000L);
//递增等待次数
++wait_times;
/* get current replay lsn again */
//再次获取当前的回放LSN，更新
(void)GetXLogReplayRecPtr(NULL, &standby_replay_lsn);
}
//直到primary_insert_lsn小于等于standby_replay_lsn，表示LSN已经回放完成
//成功回放，则返回REMOTE_READ_OK（远程读取完成）
return REMOTE_READ_OK;
}
/*
* Read block from buffer from primary, returning it as bytea
*/
//实现了从主节点读取缓冲区中的数据块，并将其作为bytea类型返回
Datum gs_read_block_from_remote(PG_FUNCTION_ARGS)
{
uint32 spcNode; //文件空间节点号
uint32 dbNode; //数据库节点号
uint32 relNode; //关系节点号
int16 bucketNode; //哈希桶节点号
int32 forkNum; //分叉号
uint64 blockNum; //块号
uint32 blockSize; //块大小
uint64 lsn; //日志序列号
bool isForCU = false; //是否是针对CU（压缩单元）块的请求
bytea* result = NULL;
int timeout = 0;
//检查当前用户是否为超级用户
if (GetUserId() != BOOTSTRAP_SUPERUSERID) {
//如果不是则抛出异常
ereport(ERROR, (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE), (errmsg("must be initial account to read files"))));
}
/* handle optional arguments */
//解析传入的参数
spcNode = PG_GETARG_UINT32(0);
dbNode = PG_GETARG_UINT32(1);
relNode = PG_GETARG_UINT32(2);
bucketNode = PG_GETARG_INT16(3);
forkNum = PG_GETARG_INT32(4);
blockNum = (uint64)PG_GETARG_TRANSACTIONID(5);
blockSize = PG_GETARG_UINT32(6);
lsn = (uint64)PG_GETARG_TRANSACTIONID(7);
isForCU = PG_GETARG_BOOL(8);
timeout = PG_GETARG_INT32(9);
//创建修复块的键（RepairBlockKey）对象，并填充相关属性
RepairBlockKey key;
key.relfilenode.spcNode = spcNode;
key.relfilenode.dbNode = dbNode;
key.relfilenode.relNode = relNode;
key.relfilenode.bucketNode = bucketNode;
key.relfilenode.opt = 0;
key.forknum = forkNum;
key.blocknum = blockNum;
/* get block from local buffer */
// 从本地缓冲区读取块
if (isForCU) {
// 从本地缓冲区读取 CU 块
/* if request to read CU block, we use forkNum column to replace colid. */
(void)StandbyReadCUforPrimary(key, blockNum, blockSize, lsn, timeout, &result);
} else {
//从本地缓冲区读取普通页面块
(void)StandbyReadPageforPrimary(key, blockSize, lsn, &result, timeout, NULL);
}
if (NULL != result) {
//成功获取到数据块，则将其转换为bytea类型并返回
PG_RETURN_BYTEA_P(result);
} else {
//未获取到数据块，则返回NULL
PG_RETURN_NULL();
}
}
/*
* Read block from buffer from primary, returning it as bytea
*/
//用于从主节点的压缩块缓冲区中读取数据，并将其以bytea类型返回
Datum gs_read_block_from_remote_compress(PG_FUNCTION_ARGS)
{
RepairBlockKey key; // 修复块的关键信息
uint32 blockSize; // 块大小
uint64 lsn; // 日志序列号
int timeout = 0; // 超时时间
bool isForCU = false; // 是否请求读取CU块
bytea* result = NULL; // 返回的块数据
// 检查当前用户是否为超级用户
if (GetUserId() != BOOTSTRAP_SUPERUSERID) {
ereport(ERROR, (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE), (errmsg("must be initial account to read files"))));
}
/* handle optional arguments */
// 获取传入的参数值
key.relfilenode.spcNode = PG_GETARG_UINT32(0); // 表空间OID
key.relfilenode.dbNode = PG_GETARG_UINT32(1); // 数据库OID
key.relfilenode.relNode = PG_GETARG_UINT32(2); // 关系OID
key.relfilenode.bucketNode = PG_GETARG_INT16(3); // 桶OID
key.relfilenode.opt = PG_GETARG_UINT16(4); // 扩展选项
key.forknum = PG_GETARG_INT32(5); // 分支号
key.blocknum = (uint64)PG_GETARG_TRANSACTIONID(6); // 块号
blockSize = PG_GETARG_UINT32(7); // 块大小
lsn = (uint64)PG_GETARG_TRANSACTIONID(8); // 日志序列号
isForCU = PG_GETARG_BOOL(9); // 是否请求读取CU块
timeout = PG_GETARG_INT32(10); // 超时时间
/* get block from local buffer */
if (isForCU) {
// 从本地缓冲区读取 CU 块
/* if request to read CU block, we use forkNum column to replace colid. */
(void)StandbyReadCUforPrimary(key, key.blocknum, blockSize, lsn, timeout, &result);
} else {
// 从本地缓冲区读取页块
(void)StandbyReadPageforPrimary(key, blockSize, lsn, &result, timeout, NULL);
}
// 检查块是否成功获取
if (NULL != result) {
// 返回块数据
PG_RETURN_BYTEA_P(result);
} else {
// 返回空值
PG_RETURN_NULL();
}
}
/*
* @Description: read cu for primary
* @IN spcnode: tablespace id
* @IN dbnode: database id
* @IN relnode: relfilenode
* @IN colid: column id
* @IN offset: cu offset
* @IN size: cu size
* @IN lsn: lsn wait for replay
* @IN/OUT context: read context
* @OUT cudata: output cu data
* @Return: remote read error code
* @See also:
*/
//用于从主节点读取某个文件中指定块号的CU数据块
int StandbyReadCUforPrimary(RepairBlockKey key, uint64 offset, int32 size, uint64 lsn, int32 timeout,
bytea** cudata)
{
// 确保输出参数cudata存在
Assert(cudata);
// 初始化返回值
int ret_code = REMOTE_READ_OK;
/* wait request lsn for replay */
//等待请求的LSN回放完成
// 检查是否处于恢复进程中
if (RecoveryInProgress()) {
// 等待指定LSN的日志重放，超时时间为timeout
ret_code = XLogWaitForReplay(lsn, timeout);
if (ret_code != REMOTE_READ_OK) {
// 如果错误码不等于REMOTE_READ_OK，则发生了错误，记录错误日志并返回错误码
ereport(ERROR, (errmodule(MOD_REMOTE), errmsg("could not redo request lsn.")));
return ret_code;
}
}
// 使用传入的参数构造关系文件节点RelFileNode对象
RelFileNode relfilenode {key.relfilenode.spcNode, key.relfilenode.dbNode, key.relfilenode.relNode, InvalidBktId};
// 在匿名块中执行以下语句
{
/* read from disk */
//创建CFileNode对象，用于标识具体的文件和文件块
CFileNode cfilenode(relfilenode, key.forknum, MAIN_FORKNUM);
// 创建CUStorage对象，用于管理和访问CU数据
CUStorage* custorage = New(CurrentMemoryContext) CUStorage(cfilenode);
// 创建CU对象，用于加载和处理CU数据
CU* cu = New(CurrentMemoryContext) CU();
cu->m_inCUCache = false;
// 从磁盘加载指定偏移量和大小的CU块到内存中的CU对象中
custorage->LoadCU(cu, offset, size, false, false);
/* check crc */
// 检查CU数据的CRC校验码是否正确
if (ret_code == REMOTE_READ_OK) {
if (!cu->CheckCrc()) {
// 如果CRC校验失败，则记录错误日志并更新返回值为REMOTE_READ_CRC_ERROR
ereport(ERROR, (errmodule(MOD_REMOTE), errmsg("check CU crc error.")));
ret_code = REMOTE_READ_CRC_ERROR;
} else {
// 复制CU数据到返回值缓冲区
bytea* buf = (bytea*)palloc0(VARHDRSZ + size);
SET_VARSIZE(buf, size + VARHDRSZ);
errno_t rc = memcpy_s(VARDATA(buf), size, cu->m_compressedLoadBuf, size);
if (rc != EOK) {
// 如果复制过程中发生错误，则记录错误日志并更新返回值为REMOTE_READ_MEMCPY_ERROR
ereport(ERROR, (errmodule(MOD_REMOTE), errmsg("memcpy_s error, retcode=%d", rc)));
ret_code = REMOTE_READ_MEMCPY_ERROR;
}
// 将buf对象赋值给输出参数cudata
*cudata = buf;
// 释放内存，删除custorage和cu对象
DELETE_EX(custorage);
DELETE_EX(cu);
}
}
}
// 返回错误代码
return ret_code;
}
/*
* @Description: read page for primary
* @IN spcnode: tablespace id
* @IN dbnode: database id
* @IN relnode: relfilenode
* @IN forknum: forknum
* @IN blocknum: block number
* @IN blocksize: block size
* @IN lsn: lsn wait for replay
* @IN/OUT context: read context
* @OUT pagedata: output page data
* @Return: remote read error code
* @See also:
*/
//读取主库的数据页，并将数据存储在pagedata中
int StandbyReadPageforPrimary(RepairBlockKey key, uint32 blocksize, uint64 lsn, bytea** pagedata,
int timeout, const XLogPhyBlock *pblk)
{
//通过断言语句，检查pagedata是否为空指针
Assert(pagedata);
//传入的数据块大小是否与定义的常量BLCKSZ相同
if (unlikely(blocksize != BLCKSZ))
//不同则返回错误码
return REMOTE_READ_BLCKSZ_NOT_SAME;
int ret_code = REMOTE_READ_OK;
/* wait request lsn for replay */
//如果系统正在恢复中
if (RecoveryInProgress()) {
//等待请求的LSN（日志序列号）进行重放
ret_code = XLogWaitForReplay(lsn, timeout);
if (ret_code != REMOTE_READ_OK) {
ereport(ERROR, (errmodule(MOD_REMOTE), errmsg("could not redo request lsn.")));
return ret_code;
}
}
//该对象用于标识要读取数据页的关系文件在数据库中的位置
RelFileNode relfilenode{key.relfilenode.spcNode, key.relfilenode.dbNode, key.relfilenode.relNode,
key.relfilenode.bucketNode, key.relfilenode.opt};
//如果指针 pblk 不为空，则表示当前正在读取的是备库上的数据块
// 这时需要验证备库中的数据块位置与主库是否一致
if (NULL != pblk) {
//获取分段物理位置信息
SegPageLocation loc = seg_get_physical_location(relfilenode, key.forknum, key.blocknum);
//将loc.extent_size（扩展大小）转换为一个无符号8位整数
uint8 standby_relNode = (uint8) EXTENT_SIZE_TO_TYPE(loc.extent_size);
BlockNumber standby_block = loc.blocknum;
//验证备库中的关系文件和数据块的位置与主库是否一致
if (standby_relNode != pblk->relNode || standby_block != pblk->block) {
//如果不一致，将报告错误并返回相应的错误码
ereport(ERROR, (errmodule(MOD_REMOTE),
errmsg("Standby page file is invalid! Standby relnode is %u, "
"master is %u; Standby block is %u, master is %u.",
standby_relNode, pblk->relNode, standby_block, pblk->block)));
return REMOTE_READ_BLCKSZ_NOT_SAME;
}
}
//如果 pblk 为空，则表示当前正在读取主库上的数据块，不需要进行验证
//分配了一个大小为 BLCKSZ + VARHDRSZ 的内存空间，并设置 pageData 的头部为该空间的有效大小
bytea* pageData = (bytea*)palloc(BLCKSZ + VARHDRSZ);
SET_VARSIZE(pageData, BLCKSZ + VARHDRSZ);
//判断关系文件节点是否属于分段物理模型
if (IsSegmentPhysicalRelNode(relfilenode)) {
Buffer buffer = InvalidBuffer;
//如果是，则使用 spc_open 函数打开对应的分段空间，并获取关系文件的数据块总数
SegSpace *spc = spc_open(relfilenode.spcNode, relfilenode.dbNode, false, false);
BlockNumber spc_nblocks = spc_size(spc, relfilenode.relNode, key.forknum);
//查给定的数据块号 key.blocknum 是否小于数据块总数 spc_nblocks
if (key.blocknum < spc_nblocks) {
//读取磁盘上对应数据块的内容
buffer = ReadBufferFast(spc, relfilenode, key.forknum, key.blocknum, RBM_FOR_REMOTE);
}
//检查 Buffer 是否有效（即是否成功读取到了数据块）
if (BufferIsValid(buffer)) {
//如果有效，则通过 LockBuffer 将该 Buffer 锁定，并获取其对应的 Block
LockBuffer(buffer, BUFFER_LOCK_SHARE);
Block block = BufferGetBlock(buffer);
//将 Block 的内容拷贝到 pageData 中
errno_t rc = memcpy_s(VARDATA(pageData), BLCKSZ, block, BLCKSZ);
//如果 memcpy_s 函数执行失败
if (rc != EOK) {
//释放之前分配的内存空间 pageData，通过 ereport 报告错误
pfree(pageData);
ereport(ERROR, (errmodule(MOD_REMOTE), errmsg("memcpy_s error, retcode=%d", rc)));
//将返回值 ret_code 设置为 REMOTE_READ_MEMCPY_ERROR
ret_code = REMOTE_READ_MEMCPY_ERROR;
}
//解锁并释放 Buffer
LockBuffer(buffer, BUFFER_LOCK_UNLOCK);
ReleaseBuffer(buffer);
} else {
//没有成功读取到数据块，则将 ret_code 设置为 REMOTE_READ_SIZE_ERROR
ret_code = REMOTE_READ_SIZE_ERROR;
}
} else { //关系文件节点不是分段物理关系
bool hit = false;
//读取指定的数据块
/* read page, if PageIsVerified failed will long jump to PG_CATCH() */
Buffer buf = ReadBufferForRemote(relfilenode, key.forknum, key.blocknum, RBM_FOR_REMOTE, NULL, &hit, pblk);
//如果读取成功，则获取对应的 Buffer
if (BufferIsInvalid(buf)) {
//则通过 ereport 报告错误并设置返回值 ret_code 为 REMOTE_READ_BLCKSZ_NOT_SAME
ereport(ERROR, (errmodule(MOD_REMOTE), errmsg("standby page buffer is invalid!")));
return REMOTE_READ_BLCKSZ_NOT_SAME;
}
//通过 LockBuffer 锁定该 Buffer，并获取其对应的 Block
LockBuffer(buf, BUFFER_LOCK_SHARE);
Block block = BufferGetBlock(buf);
//将 Block 的内容拷贝到 pageData 中
errno_t rc = memcpy_s(VARDATA(pageData), BLCKSZ, block, BLCKSZ);
//如果 memcpy_s 函数执行失败
if (rc != EOK) {
//释放之前分配的内存空间 pageData，通过 ereport 报告错误
pfree(pageData);
ereport(ERROR, (errmodule(MOD_REMOTE), errmsg("memcpy_s error, retcode=%d", rc)));
//将返回值 ret_code 设置为 REMOTE_READ_MEMCPY_ERROR
ret_code = REMOTE_READ_MEMCPY_ERROR;
}
//解锁并释放 Buffer
LockBuffer(buf, BUFFER_LOCK_UNLOCK);
ReleaseBuffer(buf);
}
//读取数据块成功的情况下执行
if (ret_code == REMOTE_READ_OK) {
//外部调用者就可以获取到读取到的数据块的内容
*pagedata = pageData;
//计算并设置该页的校验和
PageSetChecksumInplace((Page) VARDATA(*pagedata), key.blocknum);
}
//如果读取操作成功，返回的结果是 REMOTE_READ_OK，否则返回的是之前判断出的其他错误码
return ret_code;
}
const int RES_COL_NUM = 2;
//从远程读取文件，并返回结果
Datum gs_read_file_from_remote(PG_FUNCTION_ARGS)
{
RelFileNode rnode; //文件节点信息
RemoteReadFileKey key;
int32 forknum; //分片号
uint32 blockstart; //起始块号
uint64 lsn;
bytea* result = NULL;
Datum values[RES_COL_NUM];
bool nulls[RES_COL_NUM] = {false};
HeapTuple tuple = NULL;
TupleDesc tupdesc;
int ret_code = 0;
int32 timeout;
int parano = 0;
XLogRecPtr current_lsn = InvalidXLogRecPtr;
//检查当前用户是否具有足够的权限执行该函数
if (GetUserId() != BOOTSTRAP_SUPERUSERID) {
//如果不是超级用户，则会抛出错误并中断执行
ereport(ERROR, (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE), (errmsg("must be initial account to read files"))));
}
/* handle optional arguments */
//使用 PG_GETARG_* 宏来获取每个参数并将它们存储在对应的变量中
//这些参数是通过调用该函数的 SQL 语句传递给函数的
rnode.spcNode = PG_GETARG_UINT32(parano++);
rnode.dbNode = PG_GETARG_UINT32(parano++);
rnode.relNode = PG_GETARG_UINT32(parano++);
rnode.bucketNode = PG_GETARG_INT32(parano++);
rnode.opt = 0;
forknum = PG_GETARG_INT32(parano++);
blockstart = PG_GETARG_INT32(parano++);
lsn = (uint64)PG_GETARG_TRANSACTIONID(parano++);
timeout = PG_GETARG_INT32(parano++);
if (rnode.spcNode != 1 && rnode.spcNode != 2) {
//存储文件的表空间节点号不是 1 或 2，需要获取数据文件
/* get tale data file */
//初始化了一个 key 变量，并将传入的参数赋值给 key 的相应字段
key.relfilenode = rnode;
key.forknum = forknum;
key.blockstart = blockstart;
//检查 forknum 的值是否等于 MAIN_FORKNUM
if (forknum != MAIN_FORKNUM) {
//如果不是，会发出一个警告并返回 NULL
ereport(WARNING, (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
(errmsg("Forknum should be 0. Now is %d. \n", forknum))));
PG_RETURN_NULL();
}
//forknum 等于 MAIN_FORKNUM，远程读取文件
//读取指定位置的文件内容
ret_code = ReadFileForRemote(&key, lsn, &result, timeout);
} else {//rnode.spcNode 的值为 1 或 2
//读取与 rnode 相关的文件内容
ret_code = ReadCOrCsnFileForRemote(rnode, &result);
}
if (ret_code != REMOTE_READ_OK) {
//如果文件读取失败，返回 NULL
PG_RETURN_NULL();
}
if (!RecoveryInProgress()) {
//如果当前没有进行恢复操作，函数会获取当前的逻辑日志序列号（LSN）
current_lsn = GetXLogInsertRecPtr();
}
//创建一个元组描述符（TupleDesc）对象，并设置两个字段：file 和 lsn
tupdesc = CreateTemplateTupleDesc(RES_COL_NUM, false, TAM_HEAP);
parano = 1;
TupleDescInitEntry(tupdesc, (AttrNumber)parano++, "file", BYTEAOID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)parano++, "lsn", XIDOID, -1, 0);
//values 数组存储了对应字段的值，nulls 数组表示对应字段是否为 NULL
values[0] = PointerGetDatum(result);
nulls[0] = false;
values[1] = UInt64GetDatum(current_lsn);
nulls[1] = false;
//将函数的返回值转换为合适的格式，并返回给调用者
tupdesc = BlessTupleDesc(tupdesc);
tuple = heap_form_tuple(tupdesc, values, nulls);
PG_RETURN_DATUM(HeapTupleGetDatum(tuple));
}
//获取指定文件在远程节点上的大小
Datum gs_read_file_size_from_remote(PG_FUNCTION_ARGS)
{
RelFileNode rnode;
int32 forknum;
uint64 lsn;
int64 size = 0;
int32 timeout;
int parano = 0;
int ret_code = REMOTE_READ_OK;
//检查当前用户 id 是否是 Bootstrap 超级用户
if (GetUserId() != BOOTSTRAP_SUPERUSERID) {
//如果不是，则抛出一个权限不足的错误
ereport(ERROR, (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE), (errmsg("must be initial account to read files"))));
}
/* handle optional arguments */
//从输入参数中获取文件所在的 RelFileNode、forknum、lsn 和 timeout 等信息
rnode.spcNode = PG_GETARG_UINT32(parano++);
rnode.dbNode = PG_GETARG_UINT32(parano++);
rnode.relNode = PG_GETARG_UINT32(parano++);
rnode.bucketNode = PG_GETARG_INT32(parano++);
rnode.opt = 0;
forknum = PG_GETARG_INT32(parano++);
lsn = (uint64)PG_GETARG_TRANSACTIONID(parano++);
timeout = PG_GETARG_INT32(parano++);
/* get file size */
//尝试在远程节点上获取指定文件的大小
ret_code = ReadFileSizeForRemote(rnode, forknum, lsn, &size, timeout);
if (ret_code == REMOTE_READ_OK) {
//如果成功获取文件大小，则通过 PG_RETURN_INT64 宏将 size 的值作为函数的返回值返回给调用者
PG_RETURN_INT64(size);
} else {
//如果获取文件大小失败，则返回 null 值
PG_RETURN_NULL();
}
}
//在远程节点上获取指定文件的大小，并将结果通过 res 参数返回
int ReadFileSizeForRemote(RelFileNode rnode, int32 forknum, XLogRecPtr lsn, int64* res, int timeout)
{
SMgrRelation smgr = NULL;
int64 nblock = 0;
int ret_code = REMOTE_READ_OK;
/* wait request lsn for replay */
//检查是否处于恢复状态
if (RecoveryInProgress()) {
//如果正在进行恢复，等待请求的 lsn 进行重放
ret_code = XLogWaitForReplay(lsn, timeout);
//如果等待超时或发生错误，函数会抛出一个错误并返回相应的错误码
if (ret_code != REMOTE_READ_OK) {
ereport(ERROR, (errmodule(MOD_REMOTE), errmsg("could not redo request lsn.")));
return ret_code;
}
}
//发送请求进行立即强制的检查点，并等待完成
RequestCheckpoint(CHECKPOINT_IMMEDIATE | CHECKPOINT_FORCE | CHECKPOINT_WAIT);
/* check whether the file exists. not exist, return size -1 */
//检查文件是否存在
struct stat statBuf;
//根据 rnode 和 forknum 构建文件路径
char* path = relpathperm(rnode, forknum);
//调用 stat 函数检查文件的状态信息
if (stat(path, &statBuf) < 0 && errno == ENOENT) {
//如果文件不存在，则将 res 设置为 -1，并直接返回
*res = -1;
pfree(path);
return ret_code;
}
pfree(path);
//如果文件存在，且不属于分段文件
if (!IsSegmentFileNode(rnode)) {
//打开相应的 SMgrRelation
smgr = smgropen(rnode, InvalidBackendId);
//获取文件的块数量
nblock = smgrnblocks(smgr, forknum);
//关闭所有 SMgrRelation
smgrcloseall();
} else { //如果文件属于分段文件
//打开相应的 SegSpace
SegSpace *spc = spc_open(rnode.spcNode, rnode.dbNode, true, true);
spc_datafile_create(spc, rnode.relNode, forknum);
//获取分段文件的大小
nblock = spc_size(spc, rnode.relNode, forknum);
}
//将 nblock 乘以 BLCKSZ，得到文件的大小
*res = nblock * BLCKSZ;
return ret_code;
}
//从远程文件中读取指定块的内容，并将读取到的数据复制到 pageData 中的相应位置
int ReadFileByReadBufferComom(RemoteReadFileKey *key, bytea* pageData, uint32 nblock)
{
int ret_code = REMOTE_READ_OK;
uint32 i = 0;
uint32 j = 0;
uint32 blk_start;
uint32 blk_end;
bool hit = false; //标志是否命中
char* bufBlock = NULL;
errno_t rc;
/* get the segno file block start and block end */
//计算要读取的段号文件块的起始块和终止块
blk_start = key->blockstart;
blk_end = (nblock >= blk_start + MAX_BATCH_READ_BLOCKNUM ? blk_start + MAX_BATCH_READ_BLOCKNUM : nblock);
//使用循环从起始块到终止块，依次读取每个块的内容
for (i = blk_start, j = 0; i < blk_end; i++, j++) {
/* read page, if PageIsVerified failed will long jump */
//来获取访问策略，将策略指针赋给 bstrategy 变量
BufferAccessStrategy bstrategy = GetAccessStrategy(BAS_REPAIR);
//获取指定块的缓冲区，并通过 Buffer 类型的变量 buf 进行引用
Buffer buf = ReadBufferForRemote(key->relfilenode, key->forknum, i, RBM_FOR_REMOTE, bstrategy, &hit, NULL);
//如果 buf 无效，即无法获取有效的缓冲区
if (BufferIsInvalid(buf)) {
//返回错误码 REMOTE_READ_BLCKSZ_NOT_SAME
ereport(WARNING, (errmodule(MOD_REMOTE), errmsg("repair file failed!")));
return REMOTE_READ_BLCKSZ_NOT_SAME;
}
//对缓冲区进行共享锁定
LockBuffer(buf, BUFFER_LOCK_SHARE);
bufBlock = (char*)BufferGetBlock(buf);
//将缓冲区的内容复制到 pageData 的相应位置
rc = memcpy_s(VARDATA(pageData) + j * BLCKSZ, BLCKSZ, bufBlock, BLCKSZ);
//memcpy_s 函数调用失败
if (rc != EOK) {
ereport(WARNING, (errmodule(MOD_REMOTE), errmsg("repair file failed, memcpy_s error, retcode=%d", rc)));
//设置 ret_code 为 REMOTE_READ_MEMCPY_ERROR
ret_code = REMOTE_READ_MEMCPY_ERROR;
//返回之前解锁并释放缓冲区
LockBuffer(buf, BUFFER_LOCK_UNLOCK);
ReleaseBuffer(buf);
return ret_code;
}
//解锁并释放缓冲区
LockBuffer(buf, BUFFER_LOCK_UNLOCK);
ReleaseBuffer(buf);
//每个块的内容复制完成后，为复制的页面计算并设置校验和
PageSetChecksumInplace((Page) (VARDATA(pageData) + j * BLCKSZ), i);
}
//返回 ret_code 表示操作的结果
return ret_code;
}
const int MAX_RETRY_TIMES = 60;
//根据指定的参数从磁盘或者远程位置读取指定块的数据
int ReadFileByReadDisk(SegSpace* spc, RemoteReadFileKey *key, char* bufBlock, BlockNumber blocknum)
{
int ret_code = REMOTE_READ_OK;
int pageStatus; //页面状态
int retryTimes = 0; //重试次数
//如果传入的文件节点是段文件节点
if (IsSegmentFileNode(key->relfilenode)) {
//创建一个伪造的文件节点 fakenode，将其属性设置为传入的文件节点属性
// 但将 bucketNode 属性设置为 SegmentBktId
RelFileNode fakenode = {
.spcNode = key->relfilenode.spcNode,
.dbNode = key->relfilenode.dbNode,
.relNode = key->relfilenode.relNode,
.bucketNode = SegmentBktId,
.opt = 0
};
SEG_RETRY:
//从远程位置读取指定块的数据，并将数据存储到 bufBlock 中
seg_physical_read(spc, fakenode, key->forknum, blocknum, (char *)bufBlock);
retryTimes++;
//检查页面的校验和是否匹配
if (PageIsVerified((Page)bufBlock, blocknum)) {
//如果匹配，则将页面状态设置为 SMGR_RD_OK
pageStatus = SMGR_RD_OK;
} else {
//如果校验和不匹配，将页面状态设置为 SMGR_RD_CRC_ERROR
pageStatus = SMGR_RD_CRC_ERROR;
//如果重试次数小于最大重试次数
if (retryTimes < MAX_RETRY_TIMES) {
/* sleep 10ms */
pg_usleep(10000L);
//等待 10 毫秒，并跳转到标签 SEG_RETRY 处再次尝试读取数据
goto SEG_RETRY;
} else { //如果重试次数达到最大重试次数
//释放 bufBlock 的内存，并输出警告信息
pfree(bufBlock);
ereport(WARNING, (errmodule(MOD_REMOTE),
errmsg("repair file failed, read page crc check error, page: %u/%u/%u/%d, "
"forknum is %d, block num is %u", key->relfilenode.spcNode, key->relfilenode.dbNode,
key->relfilenode.relNode, key->relfilenode.bucketNode, key->forknum, blocknum)));
//将返回码 ret_code 设置为 REMOTE_READ_CRC_ERROR，并返回该返回码
ret_code = REMOTE_READ_CRC_ERROR;
return ret_code;
}
}
} else { //传入的文件节点不是段文件节点
//打开指定文件节点对应的存储管理器关系对象 smgr
SMgrRelation smgr = smgropen(key->relfilenode, InvalidBackendId);
/* standby read page, replay finish, there will be no synchronous changes. */
//从磁盘读取指定块的数据，并将数据存储到 bufBlock 中
pageStatus = smgrread(smgr, key->forknum, blocknum, (char *)bufBlock);
retryTimes++;
//检查页面状态是否为 SMGR_RD_OK
if (pageStatus != SMGR_RD_OK) {
//如果不是，则释放 bufBlock 的内存，并输出警告信息
pfree(bufBlock);
ereport(WARNING, (errmodule(MOD_REMOTE),
errmsg("repair file failed, read page crc check error, page: %u/%u/%u/%d, "
"forknum is %d, block num is %u", key->relfilenode.spcNode, key->relfilenode.dbNode,
key->relfilenode.relNode, key->relfilenode.bucketNode, key->forknum, blocknum)));
//然后将返回码 ret_code 设置为 REMOTE_READ_CRC_ERROR
ret_code = REMOTE_READ_CRC_ERROR;
//关闭存储管理器关系对象 smgr
smgrclose(smgr);
return ret_code;
}
//关闭存储管理器关系对象 smgr
smgrclose(smgr);
}
return ret_code;
}
//远程读取指定文件的数据块，并将读取的数据存储到bytea类型的变量中
int ReadFileForRemote(RemoteReadFileKey *key, XLogRecPtr lsn, bytea** fileData, int timeout)
{
int ret_code = REMOTE_READ_OK;
SMgrRelation smgr = NULL;
SegSpace* spc = NULL;
bytea* pageData = NULL;
char* bufBlock = NULL;
uint32 nblock = 0;
uint32 blk_start = 0;
uint32 blk_end = 0;
uint32 i = 0;
uint32 j = 0;
errno_t rc;
/* wait request lsn for replay */
//检查是否在恢复模式中
if (RecoveryInProgress()) {
//如果是，则等待请求的日志序列号（lsn）进行重放
ret_code = XLogWaitForReplay(lsn, timeout);
if (ret_code != REMOTE_READ_OK) {
//如果超时或出错，返回错误代码
ereport(ERROR, (errmodule(MOD_REMOTE), errmsg("could not redo request lsn.")));
return ret_code;
}
}
//如果在恢复模式中或者目标文件是段文件节点
if (RecoveryInProgress() || IsSegmentFileNode(key->relfilenode)) {
//发送请求进行检查点
RequestCheckpoint(CHECKPOINT_WAIT | CHECKPOINT_FORCE | CHECKPOINT_IMMEDIATE);
}
/* get block num */
//如果不是段文件节点
if (!IsSegmentFileNode(key->relfilenode)) {
//打开对应的SMgrRelation
smgr = smgropen(key->relfilenode, InvalidBackendId);
//获取块数
nblock = smgrnblocks(smgr, key->forknum);
//关闭SMgrRelation
smgrclose(smgr);
} else { //如果是段文件节点
//打开对应的SegSpace
spc = spc_open(key->relfilenode.spcNode, key->relfilenode.dbNode, false, false);
if (!spc) {
ereport(WARNING, (errmodule(MOD_REMOTE),
errmsg("Spc open failed. spcNode is: %u, dbNode is %u",
key->relfilenode.spcNode, key->relfilenode.dbNode)));
return REMOTE_READ_IO_ERROR;
}
//获取块数
nblock = spc_size(spc, key->relfilenode.relNode, key->forknum);
}
//检查请求的起始块（blockstart）是否超出文件块数
if (nblock <= key->blockstart) {
ret_code = REMOTE_READ_SIZE_ERROR;
return ret_code;
}
/* get the segno file block start and block end */
//根据起始块和最大批量读取块数计算出要读取的段（segno）的起始块（blk_start）和结束块（blk_end）
blk_start = key->blockstart;
blk_end = (nblock >= blk_start + MAX_BATCH_READ_BLOCKNUM ? blk_start + MAX_BATCH_READ_BLOCKNUM : nblock);
//分配用于存储读取的数据的内存空间
pageData = (bytea*)palloc((blk_end - blk_start) * BLCKSZ + VARHDRSZ);
SET_VARSIZE(pageData, ((blk_end - blk_start) * BLCKSZ + VARHDRSZ));
/* primary read page, need read page by ReadBuffer_common */
//如果不是段文件节点并且不在恢复模式中
if (!IsSegmentFileNode(key->relfilenode) && !RecoveryInProgress()) {
//读取主数据节点的页数据
ret_code = ReadFileByReadBufferComom(key, pageData, nblock);
if (ret_code != REMOTE_READ_OK) {
//如果读取失败，则释放分配的内存空间，返回错误代码
pfree(pageData);
ereport(ERROR, (errmodule(MOD_REMOTE), errmsg("read file failed!")));
return ret_code;
}
} else { //如果是段文件节点或者在恢复模式中
// 判断是否使用ADIO（Asynchronous Direct I/O）读取数据
ADIO_RUN()
{
//如果是，则使用adio_align_alloc函数分配对齐的内存块bufBlock，大小为BLCKSZ
bufBlock = (Page)adio_align_alloc(BLCKSZ);
}
ADIO_ELSE()
{
//如果不是，则使用palloc函数分配内存块bufBlock，大小为BLCKSZ
bufBlock = (Page)palloc(BLCKSZ);
}
ADIO_END();
//用循环从起始块（blk_start）到结束块（blk_end）依次读取数据
for (i = blk_start, j = 0; i < blk_end; i++, j++) {
//并存储到bufBlock中
ret_code = ReadFileByReadDisk(spc, key, bufBlock, i);
if (ret_code != REMOTE_READ_OK) {
//读取失败，则释放分配的内存块bufBlock和pageData
pfree(bufBlock);
pfree(pageData);
ereport(ERROR, (errmodule(MOD_REMOTE), errmsg("repair file failed, read block %u error, retcode=%d",
i, rc)));
//返回错误代码
return ret_code;
}
//使用memcpy_s函数将bufBlock中的数据复制到pageData的合适位置上
rc = memcpy_s(VARDATA(pageData) + j * BLCKSZ, BLCKSZ, bufBlock, BLCKSZ);
if (rc != EOK) {
//复制失败，则释放分配的内存块bufBlock和pageData，返回错误代码
pfree(bufBlock);
pfree(pageData);
ret_code = REMOTE_READ_MEMCPY_ERROR;
ereport(ERROR, (errmodule(MOD_REMOTE), errmsg("repair file failed, memcpy_s error, retcode=%d", rc)));
return ret_code;
}
}
//释放分配的内存块bufBlock
pfree(bufBlock);
//如果读取失败，则释放分配的内存空间pageData，返回错误代码
if (ret_code != REMOTE_READ_OK) {
pfree(pageData);
ereport(ERROR, (errmodule(MOD_REMOTE), errmsg("read file failed!")));
return ret_code;
}
}
//将读取的数据赋值给fileData指针
*fileData = pageData;
//返回读取结果代码
return ret_code;
}
const int REGR_MCR_SIZE_1MB = 1048576;
const int CLOG_NODE = 1;
const int CSN_NODE = 2;
//读取指定文件（根据RelFileNode确定）的数据，并将数据存储到bytea类型的变量中
int ReadCOrCsnFileForRemote(RelFileNode rnode, bytea** fileData)
{
uint32 flags = O_RDWR | PG_BINARY;
int fd = -1;
char* logType = NULL;
char* path = (char*)palloc0(MAX_PATH);
errno_t rc;
uint32 logSize = 16 * REGR_MCR_SIZE_1MB;
char *buffer = (char*)palloc(logSize);
int result = -1;
//根据给定的文件类型（rnode.spcNode），确定文件所在的路径（logType）
if (rnode.spcNode == CLOG_NODE) {
logType = "pg_clog";
} else if (rnode.spcNode == CSN_NODE) {
logType = "pg_csnlog";
} else {
ereport(LOG, (errmodule(MOD_SEGMENT_PAGE), errmsg("File type\"%u\" does not exist, stop read here.",
rnode.spcNode)));
}
//构建完整的文件路径（path），格式为"logType/relNode"，其中relNode是文件的节点号
rc = snprintf_s(path, MAX_PATH, MAX_PATH - 1, "%s/%012u", logType, rnode.relNode);
securec_check_ss(rc, "\0", "\0");
//调用BasicOpenFile函数以读写模式打开文件，并获取文件描述符（fd）
fd = BasicOpenFile(path, flags, S_IWUSR | S_IRUSR);
if (fd < 0) {
//如果打开失败
pfree(buffer);
if (errno != ENOENT) { //且错误码不是ENOENT
ereport(ERROR, (errcode_for_file_access(), errmsg("could not open file \"%s\": %m", path)));
}
// The file does not exist, break.
ereport(LOG,
(errmodule(MOD_SEGMENT_PAGE), errmsg("File \"%s\" does not exist, stop read here.", path)));
pfree(path);
//则报错并返回-1
return -1;
}
//为了记录这次读取文件数据的等待事件
pgstat_report_waitevent(WAIT_EVENT_DATA_FILE_READ);
//分配用于存储文件数据的内存空间（buffer），大小为logSize，其中logSize默认为16 MB
//使用pread函数从文件中读取数据，将读取的数据存储到buffer中。读取的字节数由nbytes返回
uint32 nbytes = pread(fd, buffer, logSize, 0);
//结束等待事件的记录
pgstat_report_waitevent(WAIT_EVENT_END);
if (close(fd)) { //关闭文件，并检查关闭是否成功
pfree(path);
pfree(buffer);
ereport(ERROR, (errcode_for_file_access(), errmsg("could not close file \"%s\": %m", path)));
}
//检查实际读取的字节数（nbytes）是否大于logSize
if (nbytes > logSize) {
//如果是，则表示读取的数据超过了预设的大小
pfree(buffer);
ereport(ERROR,
(errcode(MOD_SEGMENT_PAGE),
errcode_for_file_access(),
errmsg("could not read file %s. nbytes:%u, logSize:%u", path, nbytes, logSize)));
pfree(path);
//报错并返回-1
return -1;
} else {
//分配用于存储读取的数据的内存空间（pageData），大小为nbytes + VARHDRSZ，其中VARHDRSZ是页头变长字段的大小
bytea* pageData = (bytea*)palloc(nbytes + VARHDRSZ);
SET_VARSIZE(pageData, (nbytes + VARHDRSZ));
//将读取的数据从buffer复制到pageData中
rc = memcpy_s(VARDATA(pageData), nbytes, buffer, nbytes);
if (rc != EOK) {
pfree(path);
pfree(pageData);
pfree(buffer);
ereport(ERROR, (errmodule(MOD_REMOTE), errmsg("repair file failed, memcpy_s error, retcode=%d", rc)));
return -1;
} else { //将pageData赋值给fileData指针
*fileData = pageData;
result = 0;
}
}
//释放不再使用的内存空间
pfree(path);
pfree(buffer);
//返回读取结果代码
return result;
}
Reference in New Issue View Git Blame Copy Permalink