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

openGauss项目TCP/IP代码注释 #24

Open
shuimuzhihua wants to merge 15 commits from shuimuzhihua/openGauss-server:master into master
pull from: shuimuzhihua/openGauss-server:master
merge into: huawei:master
Conversation 0 Commits 15 Files Changed 13 +2529 -1499
13 changed files with 2529 additions and 1499 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

43
src/gausskernel/cbb/communication/libcomm_core/mc_poller.h
View File

@ -30,44 +30,43 @@
#include "libcomm_utils/libcomm_err.h"
#include "mc_poller_epoll.h"
// Initialize poller.
//
// 初始化轮询器
int mc_poller_init(struct mc_poller* self);
// Terminate poller.
//
// 终止轮询器
void mc_poller_term(struct mc_poller* self);
// Add fd to poller.
//
// 向轮询器中添加socket
void mc_poller_add(struct mc_poller* self, int fd, int id);
// Remove item from poller
//
// 从轮询器中删除socket
int mc_poller_rm(struct mc_poller* self, int fd);
// Set poller event as POLLIN.
//
// 设置轮询事件
void mc_poller_set_in(struct mc_poller* self, struct mc_poller_hndl* hndl);
// Wait poller events happens.
//
// 等待轮询事件发生
int mc_poller_wait(struct mc_poller* self, int timeout);
// declaration of Poller handler list.
//
class mc_poller_hndl_list {
// 事件轮询处理器列表
class mc_poller_hndl_list
{
public:
mc_poller_hndl_list();
~mc_poller_hndl_list();
struct mc_poller* get_poller();
int get_socket_count();
int init();
int add_fd(struct sock_id* fd_id);
int del_fd(struct sock_id* fd_id);
// Disable copy construction and assignment.
int add_fd(struct sock_id* fd_id); // 添加socket
int del_fd(struct sock_id* fd_id); // 删除socket
// 删除拷贝构造函数和赋值运算符重载
mc_poller_hndl_list(const mc_poller_hndl_list&) = delete;
const mc_poller_hndl_list& operator=(const mc_poller_hndl_list&) = delete;
struct mc_poller* m_poller;
public:
struct mc_poller* m_poller; // 存储事件的容器(事件数组)
private:
pthread_mutex_t m_lock;
int m_socket_count;
pthread_mutex_t m_lock; // 互斥锁
int m_socket_count; // 套接字个数
};
#endif // _MC_POLLER_H_

153
src/gausskernel/cbb/communication/libcomm_core/mc_poller_epoll.cpp
View File

@ -25,6 +25,12 @@
#include "mc_poller.h"
#include "libcomm_common.h"
/*
* 功能：初始化一个事件轮询器
* self：指向mc_poller结构的指针，用于管理事件轮询
* 返回值：返回0表示成功初始化(但也可能抛出异常)
* 注：该函数封装了epoll_create函数
*/
int mc_poller_init(struct mc_poller* self)
{
#ifndef EPOLL_CLOEXEC
@ -32,80 +38,116 @@ int mc_poller_init(struct mc_poller* self)
#endif
#ifdef EPOLL_CLOEXEC
// 创建epoll实例，并使用EPOLL_CLOEXEC标志
self->ep = epoll_create1(EPOLL_CLOEXEC);
#else
// Size parameter is unused, we can safely set it to 1.
//
self->ep = epoll_create(1);
// 设置epoll实例的FD_CLOEXEC标志，避免子进程继承该文件描述符
rc = fcntl(self->ep, F_SETFD, FD_CLOEXEC);
errno_assert(rc != -1);
#endif
if (self->ep == -1) {
// 处理创建epoll失败的情况
if (errno == ENFILE || errno == EMFILE) {
return -EMFILE;
return -EMFILE; // 文件描述符不足
}
errno_assert(false);
errno_assert(false); // 发生其他错误，抛出异常
}
// 初始化轮询器的nevents和index
self->nevents = 0;
self->index = 0;
return 0;
}
/*
* 功能：关闭轮询器描述符，释放资源
* self：指向mc_poller结构的指针，用于管理事件轮询
*/
void mc_poller_term(struct mc_poller* self)
{
close(self->ep);
close(self->ep); // 关闭轮询器描述符，释放资源
}
/*
* 功能：向事件轮询器中添加一个文件描述符，并进行相关的初始化操作
* self：指向mc_poller结构的指针，用于管理事件轮询
* fd：待添加到事件轮询器中的文件描述符
* id：与文件描述符相关联的标识符或索引(标识特定连接或资源的唯一标识符)
* 注：该函数封装了epoll_ctl函数
*/
void mc_poller_add(struct mc_poller* self, int fd, int id)
{
int rc;
struct epoll_event ev;
errno_t ss_rc = 0;
// Initialise the handle and add the file descriptor to the pollset.
//
// 初始化事件结构
ss_rc = memset_s(&ev, sizeof(ev), 0, sizeof(struct epoll_event));
securec_check(ss_rc, "\0", "\0");
// 设置事件类型为EPOLLIN，表示可读事件
ev.events = EPOLLIN;
// 将文件描述符和id组合成一个64位的整数，存储在事件的data.u64中
ev.data.u64 = (((uint64)(unsigned)(fd)) << MC_POLLER_FD_ID_OFFSET) + id;
// 将文件描述符添加到事件轮询器中
rc = epoll_ctl(self->ep, EPOLL_CTL_ADD, fd, &ev);
errno_assert(rc == 0);
errno_assert(rc == 0); // 检查epoll_ctl是否成功
}
int mc_poller_rm(struct mc_poller* self, int fd)
/*
* 功能：从事件轮询器中移除文件描述符
* self：指向mc_poller结构的指针，用于管理事件轮询
* fd：待添加到事件轮询器中的文件描述符
* 返回值：直接返回epoll_ctl结果
* 注：该函数封装了epoll_ctl函数
*/
int mc_poller_rm(struct mc_poller *self, int fd)
{
// Remove the file descriptor from the pollset.
//
// 使用epoll_ctl函数执行删除操作
return epoll_ctl(self->ep, EPOLL_CTL_DEL, fd, NULL);
}
int mc_poller_wait(struct mc_poller* self, int timeout)
/*
* 功能：在事件轮询器上等待事件的发生
* self：指向mc_poller结构的指针，用于管理事件轮询
* timeout：等待事件的超时时间(单位：毫秒)
* 返回值：返回0表示成功初始化(但也可能抛出异常)
* 注：该函数封装了epoll_wait函数
*/
int mc_poller_wait(struct mc_poller *self, int timeout)
{
int nevents;
// Clear all existing events.
//
// 清除现有事件
self->nevents = 0;
self->index = 0;
// Wait for new events.
//
// 等待新事件的发生
for (;;) {
// 使用epoll_wait函数等待事件，并将结果保存在self->events数组中
nevents = epoll_wait(self->ep, self->events, MC_POLLER_MAX_EVENTS, timeout);
// 如果返回结果是-1且错误是EINTR，表示被中断，继续等待
if (mc_slow(nevents == -1 && errno == EINTR)) {
continue;
} else {
break;
}
}
// 检查self->nevents是否不等于-1，如果等于-1，抛出异常
errno_assert(self->nevents != -1);
// 将实际获取的事件数量赋值给self->nevents
self->nevents = nevents;
return 0;
}
// implementation of mc_poller_hndl_list
//
// mc_poller_hndl_list类的构造函数和析构函数
mc_poller_hndl_list::mc_poller_hndl_list()
{
m_poller = NULL;
@ -116,72 +158,97 @@ mc_poller_hndl_list::~mc_poller_hndl_list()
{
LIBCOMM_FREE(m_poller, sizeof(struct mc_poller));
}
/*
* 功能：初始化事件轮询处理器列表
* 返回值：返回0表示成功初始化，返回-1表示出错
*/
int mc_poller_hndl_list::init()
{
LIBCOMM_PTHREAD_MUTEX_INIT(&m_lock, 0);
LIBCOMM_PTHREAD_MUTEX_INIT(&m_lock, 0); // 初始化互斥锁
LIBCOMM_MALLOC(m_poller, (sizeof(struct mc_poller)), mc_poller);
LIBCOMM_MALLOC(m_poller, (sizeof(struct mc_poller)), mc_poller); // 分配内存存储轮询器
if (NULL == m_poller) {
return -1;
return -1; // 内存分配失败，返回错误码
}
if (mc_poller_init(m_poller)) {
return -1;
return -1; // 轮询器初始化失败，返回错误码
}
m_socket_count = 0;
m_socket_count = 0; // 初始化socket数量
return 0;
return 0; // 初始化成功，返回0
}
struct mc_poller* mc_poller_hndl_list::get_poller()
{
return m_poller;
}
/*
* 功能：获取事件轮询处理器列表中的socket数量
* 返回值：返回获取到的socket数量
* 注：该读取操作线程不安全，所以需要使用互斥锁
*/
int mc_poller_hndl_list::get_socket_count()
{
int cnt = 0;
LIBCOMM_PTHREAD_MUTEX_LOCK(&m_lock);
cnt = m_socket_count;
LIBCOMM_PTHREAD_MUTEX_UNLOCK(&m_lock);
return cnt;
LIBCOMM_PTHREAD_MUTEX_LOCK(&m_lock); // 加锁
cnt = m_socket_count; // 获取socket数量
LIBCOMM_PTHREAD_MUTEX_UNLOCK(&m_lock); // 解锁
return cnt; // 返回获取到的socket数量
}
int mc_poller_hndl_list::add_fd(struct sock_id* fd_id)
/*
* 功能：将文件描述符添加到事件轮询处理器列表中
* fd_id：文件描述符
* 返回值：成功返回0；失败返回-1
*/
int mc_poller_hndl_list::add_fd(struct sock_id *fd_id)
{
if (fd_id == NULL) {
return -1;
return -1; // 如果传入的文件描述符为空，返回错误码-1
}
#ifdef LIBCOMM_FAULT_INJECTION_ENABLE
if (is_comm_fault_injection(LIBCOMM_FI_POLLER_ADD_FD_FAILED)) {
LIBCOMM_ELOG(
WARNING, "(poller add fd)\t[FAULT INJECTION]Failed to save socket[%d] version[%d].", fd_id->fd, fd_id->id);
// 如果启用了错误注入并且出现故障注入，记录日志并返回错误码-1
LIBCOMM_ELOG(WARNING, "(poller add fd)\t[FAULT INJECTION]Failed to save socket[%d] version[%d].", fd_id->fd,
fd_id->id);
return -1;
}
#endif
LIBCOMM_PTHREAD_MUTEX_LOCK(&m_lock);
// add the the socket to do epoll
//
LIBCOMM_PTHREAD_MUTEX_LOCK(&m_lock); // 加锁
// 将文件描述符添加到事件轮询器中进行监听
mc_poller_add(m_poller, fd_id->fd, fd_id->id);
// 增加监听的socket数量
m_socket_count++;
LIBCOMM_PTHREAD_MUTEX_UNLOCK(&m_lock);
LIBCOMM_PTHREAD_MUTEX_UNLOCK(&m_lock); // 解锁
return 0;
}
int mc_poller_hndl_list::del_fd(struct sock_id* fd_id)
/*
* 功能：从事件轮询处理器列表中移除文件描述符
* fd_id：文件描述符
* 返回值：移除结果（mc_poller_rm函数的返回值）
*/
int mc_poller_hndl_list::del_fd(struct sock_id *fd_id)
{
if (fd_id == NULL) {
return -1;
return -1; // 如果传入的文件描述符标识符为空，返回错误码-1
}
LIBCOMM_PTHREAD_MUTEX_LOCK(&m_lock);
LIBCOMM_PTHREAD_MUTEX_LOCK(&m_lock); // 加锁
int rc = mc_poller_rm(m_poller, fd_id->fd);
int rc = mc_poller_rm(m_poller, fd_id->fd); // 移除文件描述符
LIBCOMM_PTHREAD_MUTEX_UNLOCK(&m_lock);
LIBCOMM_PTHREAD_MUTEX_UNLOCK(&m_lock); // 解锁
return rc;
return rc; // 返回移除结果
}

22
src/gausskernel/cbb/communication/libcomm_core/mc_poller_epoll.h
View File

@ -30,24 +30,24 @@
#define MC_POLLER_HAVE_ASYNC_ADD 1
#define MC_POLLER_MAX_EVENTS 512
#define MC_POLLER_MAX_EVENTS 512 // 目前正在处理事件（epoll_event）的最大数目
#define MC_POLLER_FD_ID_OFFSET 32
#define MC_POLLER_FD_ID_OFFSET 32 // fd的移位值（左移）
#define MC_POLLER_FD_ID_MASK 0xffff
// Item of epoller list, including poller handler and list item.
//
struct mc_poller_hndl_item {
// 定义一个结构体mc_poller_hndl_item，用于存储poller句柄和列表项
struct mc_poller_hndl_item
{
struct mc_poller_hndl hndl;
struct mc_list_item item;
};
struct mc_poller {
int ep; // Current pollset.
int nevents; // Number of events being processed at the moment.
int index; // Index of the event being processed at the moment.
struct epoll_event events[MC_POLLER_MAX_EVENTS]; // Events being processed at the moment.
struct mc_poller
{
int ep; // 当前poller值
int nevents; // 当前正在处理的事件数
int index; // 当前正在处理的事件的索引
struct epoll_event events[MC_POLLER_MAX_EVENTS]; // 目前正在处理事件
};
#endif

655
src/gausskernel/cbb/communication/libcomm_core/mc_tcp.cpp
View File

File diff suppressed because it is too large Load Diff

44
src/gausskernel/cbb/communication/libcomm_core/mc_tcp.h
View File

@ -26,47 +26,49 @@
#include <sys/stat.h>
#include <sys/uio.h>
#include <sys/un.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <errno.h>
#include <sys/types.h> // 有关系统数据类型定义的头文件
#include <sys/wait.h> // 有关进程控制定义的头文件
#include <sys/socket.h> // 套接字编程的头文件
#include <arpa/inet.h> // 网络操作的头文件
#include <errno.h> // 有关错误码定义的头文件
#include <fcntl.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <netinet/in.h>
#include <netinet/in.h> // 提供网络地址和端口相关定义的头文件
#include <netinet/tcp.h>
#include "../libcomm_common.h"
#define TCP_LISTENQ 1024
#define TCP_LISTENQ 1024 // 侦听队列的容量
extern void mc_tcp_set_keepalive_param(int idle, int intvl, int count);
extern void mc_tcp_set_keepalive_param(int idle, int intvl, int count); // 设置TCP连接的保活参数
extern void mc_tcp_set_timeout_param(int conn_timeout, int send_timeout);
extern void mc_tcp_set_timeout_param(int conn_timeout, int send_timeout); // 设置TCP连接超时参数
extern int mc_tcp_get_connect_timeout();
extern int mc_tcp_get_connect_timeout(); // 获取超时参数
extern int mc_tcp_listen(const char* host, int port, socklen_t* addrlenp);
extern int mc_tcp_listen(const char* host, int port, socklen_t* addrlenp); // 在指定的主机和端口上进行TCP监听
extern int mc_tcp_accept(int fd, struct sockaddr* sa, socklen_t* salenptr);
extern int mc_tcp_accept(int fd, struct sockaddr* sa, socklen_t* salenptr); // 服务端接受客户端的连接
extern int mc_tcp_connect(const char* host, int port);
extern int mc_tcp_connect(const char* host, int port); // 创建一个TCP连接到指定的主机和端口
extern int mc_tcp_get_peer_name(int fd, char* host, int* port);
extern int mc_tcp_get_peer_name(int fd, char* host, int* port); // 获取目标IP地址和端口号
extern int mc_tcp_write_block(int fd, const void* data, int size);
extern int mc_tcp_write_block(int fd, const void* data, int size); // 以阻塞方式向指定的套接字写入数据
extern int mc_tcp_write_noblock(int fd, const void* data, int size);
extern int mc_tcp_write_noblock(int fd, const void* data, int size); // 以非阻塞方式向指定的套接字写入数据
extern int mc_tcp_read_block(int fd, void* data, int size, int flags);
extern int mc_tcp_read_block(int fd, void* data, int size, int flags); // 以阻塞方式从给定的套接字中读取数据
extern int mc_tcp_read_nonblock(int fd, void* data, int size, int flags);
extern int mc_tcp_read_nonblock(int fd, void* data, int size, int flags); // 以非阻塞方式从给定的套接字中读取数据
extern int mc_tcp_check_socket(int sock);
extern int mc_tcp_check_socket(int sock); // 检查套接字的状态
extern void mc_tcp_close(int fd);
extern void mc_tcp_close(int fd); // 关闭一个套接字
extern int mc_tcp_addr_init(const char* host, int port, struct sockaddr_storage* ss, int* in_len);
extern int mc_tcp_addr_init(const char* host, int port, struct sockaddr_storage* ss, int* in_len); // 设置主机和端口信息
#endif //_CORE_MC_TCP_H_

1324
src/gausskernel/cbb/communication/libcomm_utils/libcomm_adapter.cpp
View File

File diff suppressed because it is too large Load Diff

29
src/gausskernel/cbb/communication/libcomm_utils/libcomm_err.cpp
View File

@ -31,7 +31,7 @@ void mc_err_abort(void)
static const char unknown_error[] = "";
const char* mc_strerror(int errnum)
{
if (errnum >= 1000) {
if (errnum >= 1000) {//对不同的错误类型进行分类处理记录
return mc_comlib_strerror(errnum); // communication layer defined errors
} else if (errnum == 0) {
return unknown_error;
@ -39,3 +39,30 @@ const char* mc_strerror(int errnum)
return strerror(errnum); // system defined errors
}
}
/*
libcomm_err.cpp文件是关于对于数据库中出现的错误进行归类分析，其中定义了两个函数，分别是
1，mc_err_abort函数，此函数多存在于libcomm_err.h之中，主要目的是对libcomm_err.h之中的宏定义进行
终止处理，起到了制止函数运行的作用
2，mc_strerror函数，此函数作用在libcomm_errno-comlib.h之中，对libcomm_errno-comlib.h传递数值，
让libcomm_errno-comlib.h对错误类型进行判断
libcomm_err.cpp是一个查找错误的函数，其主要目的为找出数据库之中的错误并将其记录，保证了数据库的准确性。
下面将分别对libcomm_err.h文件和libcomm_errno-comlib.h文件进行注释
1，libcomm_err.h文件是将错误文件记录在文件夹之中的文件，并且中断函数的运行，
其中对不同的错误类型进行了区分，其中包含函数：
1，mc_assert(x)宏定义：在win32之中仍存在一些不适配，并将其中的文件路径以及函数输出到屏幕之上
2，gai_assert(x)宏定义：对mc_unlikely(x)进行判定，若发现出现错误，将错误信息、错误号和源文件名以及行号打印到标准错误输出
3，posix_assert(x)宏定义：与2函数相同，将错误信息、错误号和源文件名以及行号打印到标准错误输出，但是输出错误号为strerror(x)错误号
4，mc_assert_state(obj, state_name)宏定义：对obj->name和state_name进行比较，将错误信息、错误号和源文件名以及行号打印到标准错误输出，同时对错误缓冲区进行刷新，确保下一次错误能被及时显示在屏幕上
5，alloc_assert(x)宏定义：对储存分配进行检查，如果发现储存未正确分配，将错误信息、错误号和源文件名以及行号打印到标准错误输出
6，errno_assert(x)宏定义：对通信的条件进行错误检查，如果发现出现错误，将错误信息、错误号和源文件名以及行号打印到标准错误输出
7，errnum_assert(cond, err)宏定义：对6宏定义进行检查，查看错误信息，错误号是否正确，如果发现错误，将错误信息、错误号和源文件名以及行号打印到标准错误输出
8、9，win_assert(x)，wsa_assert(x)宏定义：与6函数类似，均为对条件进行判断，如果条件中存在作物，进行错误信息输出，不同于6，其输出错误信息分别为最后接收错误以及WSA最后接受错误
10，mc_fsm_error(message, state, src, type)宏定义：分别将message, state, src, type四个点作为错误信息进行输出
2.libcomm_errno-comlib.h文件对错误类型进行判断，分别判断从错误类型1000到1063的错误类型，对错误类型进行记录
主函数不仅对于1000-1063的错误类型进行了判断，同样也对错误类型不在此范围的错误进行判断，其中有未知错误以及不同别的错误类型
*/

129
src/gausskernel/cbb/communication/libcomm_utils/libcomm_err.h
View File

@ -48,109 +48,138 @@
/* Same as system assert(). However, under Win32 assert has some deficiencies.
Thus this macro. */
#define mc_assert(x) \
#define mc_assert(x) /*总体来说，这段代码的目的是在某些条件不满足时输出调试日志并终止程序，以确保程序的正确性。*/ \
do { \
if (mc_slow(!(x))) { \
COMM_DEBUG_LOG("sctp check failed: %s (%s:%d)\n", #x, __FILE__, __LINE__); \
mc_err_abort(); \
} \
} while (0)
/*如果表达式 x 不满足，则输出一条调试日志，记录断言失败的表达式 x，
以及该日志消息所在的源文件名和行号。__FILE__ 和 __LINE__ 是预定义的宏，
分别表示当前源文件的文件名和行号。*/
mc_err_abort(); \
/*如果表达式 x 不满足，则调用 mc_err_abort 函数，进行错误处理并终止程序执行。*/ \
}
} while (0)//由于条件为0所以只会执行一次
// Provides convenient way to check for errors from getaddrinfo.
#define gai_assert(x) \
#define gai_assert(x) // 定义一个名为gai_assert的宏，参数为x ，如果x为假将会执行错误处理
do { \
if (mc_unlikely(x)) { \
const char* errstr = gai_strerror(x); \
fprintf(stderr, "%s (%s:%d)\n", errstr, __FILE__, __LINE__); \
mc_err_abort(); \
} \
const char* errstr = gai_strerror(x); // 调用gai_strerror函数，将错误号转化为错误信息字符串
fprintf(stderr, "%s (%s:%d)\n", errstr, __FILE__, __LINE__); // 将错误信息，文件名和行号打印到标准错误输出
mc_err_abort(); //调用函数终止程序运行 \
}
} while (false)
// Provides convenient way to check for POSIX errors.
#define posix_assert(x) \
do { \
#define posix_assert(x) //次宏定义和上一个宏定义有相似的效果，区别在于次宏定义打印的标准错误输出与之前存在不同 \
do {
if (mc_unlikely(x)) { \
fprintf(stderr, "%s (%s:%d)\n", strerror(x), __FILE__, __LINE__); \
mc_err_abort(); \
} \
fprintf(stderr, "%s (%s:%d)\n", strerror(x), __FILE__, __LINE__); //对strerror(x)进行打印输出
mc_err_abort();
}
} while (false)
#define mc_assert_state(obj, state_name) \
do { \
if (mc_slow((obj)->state != (state_name))) { \
fprintf(stderr, "Assertion failed: %d == %s (%s:%d)\n", (obj)->state, #state_name, __FILE__, __LINE__); \
(void)fflush(stderr); \
mc_err_abort(); \
} \
} while (0)
if (mc_slow((obj)->state != (state_name))) { // 使用mc_slow宏，用于优化可能性较小的条件判断，如果obj的状态不等于state_name
fprintf(stderr, "Assertion failed: %d == %s (%s:%d)\n", (obj)->state, #state_name, __FILE__, __LINE__);// 将错误信息，文件名和行号打印到标准错误输出
(void)fflush(stderr); // 刷新标准错误输出缓冲区，确保错误信息立即显示
mc_err_abort(); //调用函数终止程序运行 \
}
} while (0)//由于条件为0所以只会执行一次
/* Checks whether memory allocation was successful. */
#define alloc_assert(x) \
do { \
if (mc_slow(!(x))) { \
fprintf(stderr, "Out of memory (%s:%d)\n", __FILE__, __LINE__); \
(void)fflush(stderr); \
mc_err_abort(); \
} \
#define alloc_assert(x) //宏定义，检查存储分配是否成功
do {
if (mc_slow(!(x))) { //如果存储分配未成功， 将错误信息、错误号和源文件名以及行号打印到标准错误输出
fprintf(stderr, "Out of memory (%s:%d)\n", __FILE__, __LINE__);
(void)fflush(stderr); // 刷新标准错误输出缓冲区，确保错误信息立即显示
mc_err_abort(); //调用函数终止程序运行
}
} while (0)
/* Check the condition. If false prints out the errno. */
#define errno_assert(x) \
// 定义一个名为errno_assert的宏，参数为x
#define errno_assert(x)
// 查看条件x，如果出现错误输出错误
do { \
// 使用mc_slow宏来优化可能性较小的条件判断。如果x不为真，则执行下面的代码块
if (mc_slow(!(x))) { \
// 将错误信息、错误号和源文件名以及行号打印到标准错误输出
fprintf(stderr, "%s [%d] (%s:%d)\n", mc_strerror(errno), (int)errno, __FILE__, __LINE__); \
// 刷新标准错误输出缓冲区，确保错误信息立即显示
(void)fflush(stderr); \
// 调用mc_err_abort函数终止程序运行
mc_err_abort(); \
} \
} while (0)
/* Checks whether supplied errno number is an error. */
#define errnum_assert(cond, err) \
// 定义一个名为errnum_assert的宏，参数为cond和err
#define errnum_assert(cond, err) //检查检查是否出错的出错码是否出错 \
// 使用do-while循环来保证宏的使用像一个语句
do { \
// 使用mc_slow宏来优化可能性较小的条件判断。如果cond不为真，则执行下面的代码块
if (mc_slow(!(cond))) { \
// 将错误信息、错误号和源文件名以及行号打印到标准错误输出
fprintf(stderr, "%s [%d] (%s:%d)\n", mc_strerror(err), (int)(err), __FILE__, __LINE__); \
// 刷新标准错误输出缓冲区，确保错误信息立即显示
(void)fflush(stderr); \
// 调用mc_err_abort函数终止程序运行
mc_err_abort(); \
} \
} while (0)
/* Checks the condition. If false prints out the GetLastError info. */
// 定义一个名为win_assert的宏，参数为x
#define win_assert(x) \
// 使用do-while循环来保证宏的使用像一个语句
do { \
// 使用mc_slow宏来优化可能性较小的条件判断。如果x不为真，则执行下面的代码块
if (mc_slow(!(x))) { \
// 定义一个长度为256的字符数组errstr，用于存储错误信息
char errstr[256]; \
// 调用mc_win_error函数，将错误信息存储在errstr中
mc_win_error((int)GetLastError(), errstr, 256); \
// 将错误信息、错误号和源文件名以及行号打印到标准错误输出
fprintf(stderr, "%s [%d] (%s:%d)\n", errstr, (int)GetLastError(), __FILE__, __LINE__); \
// 刷新标准错误输出缓冲区，确保错误信息立即显示
(void)fflush(stderr); \
// 调用mc_err_abort函数终止程序运行
mc_err_abort(); \
} \
} while (0)
/* Checks the condition. If false prints out the WSAGetLastError info. */
#define wsa_assert(x) \
do { \
if (mc_slow(!(x))) { \
char errstr[256]; \
mc_win_error(WSAGetLastError(), errstr, 256); \
fprintf(stderr, "%s [%d] (%s:%d)\n", errstr, (int)WSAGetLastError(), __FILE__, __LINE__); \
(void)fflush(stderr); \
mc_err_abort(); \
} \
} while (0)
#define wsa_assert(x) //与之前宏定义类似，区别一上一个宏定义，该宏定义对于错误存储为存储WSAGetLastError()错误，存在差别 \
do { // 开始一个do-while循环，这个循环至少会执行一次，即使条件不满足
if (mc_slow(!(x))) { // 如果函数mc_slow的参数为假（也就是说，x不满足），那么执行下面的代码块
char errstr[256]; // 定义一个长度为256的字符数组errstr，用于存储错误信息
mc_win_error(WSAGetLastError(), errstr, 256); // 调用函数mc_win_error，获取Windows的最后一个错误，并将其存储在errstr中
fprintf(stderr, "%s [%d] (%s:%d)\n", errstr, (int)WSAGetLastError(), __FILE__, __LINE__); // 将错误信息、错误号、当前文件名和行号打印到标准错误输出
(void)fflush(stderr); // 刷新标准错误输出缓冲区，确保上面的错误信息立即显示
mc_err_abort(); // 调用函数mc_err_abort，这个函数通常会导致程序终止
} // 结束if语句
} while (0); // 结束do-while循环，因为条件为0（即假），所以这个循环只会执行一次
/* Assertion-like macros for easier fsm debugging. */
#define mc_fsm_error(message, state, src, type) \
do { \
fprintf(stderr, "%s: state=%d source=%d action=%d (%s:%d)\n", message, state, src, type, __FILE__, __LINE__); \
(void)fflush(stderr); \
mc_err_abort(); \
} while (0)
#define mc_fsm_error(message, state, src, type) // 定义一个名为mc_fsm_error的宏，该宏接受四个参数：message（错误消息），state（当前状态），src（源），type（类型）
do { // 开始一个do-while循环，因为后面有一个分号，所以这个循环只会执行一次
fprintf(stderr, "%s: state=%d source=%d action=%d (%s:%d)\n", message, state, src, type, __FILE__, __LINE__); // 将错误消息，当前状态，源，类型以及当前的文件名和行号输出到标准错误流
(void)fflush(stderr); // 刷新标准错误流，确保上面的输出立即被显示
mc_err_abort(); // 调用mc_err_abort函数，终止函数进程
} while (0) // 结束do-while循环
//皆为调用mc_fsm_error（）宏并将错误信息传入的宏定义
#define mc_fsm_bad_action(state, src, type) mc_fsm_error("Unexpected action", state, src, type) // 如果在有限状态机中出现了预期之外的动作，调用mc_fsm_error宏，并传入错误消息"Unexpected action"以及当前的状态、源、类型
#define mc_fsm_bad_action(state, src, type) mc_fsm_error("Unexpected action", state, src, type)
#define mc_fsm_bad_state(state, src, type) mc_fsm_error("Unexpected state", state, src, type)
#define mc_fsm_bad_source(state, src, type) mc_fsm_error("Unexpected source", state, src, type)
#define mc_fsm_bad_state(state, src, type) mc_fsm_error("Unexpected state", state, src, type) // 如果在有限状态机中出现了预期之外的状态，调用mc_fsm_error宏，并传入错误消息"Unexpected state"以及当前的状态、源、类型
#define mc_fsm_bad_source(state, src, type) mc_fsm_error("Unexpected source", state, src, type) // 如果在有限状态机中出现了预期之外的源，调用mc_fsm_error宏，并传入错误消息"Unexpected source"以及当前的状态、源、类型
void mc_err_abort(void);
const char* mc_err_strerror(int errnum);
#endif //_UTILS_ERR_H_
/*
*/

2
src/gausskernel/cbb/communication/libcomm_utils/libcomm_errno-comlib.h
View File

@ -23,7 +23,7 @@
#ifndef _MC_ERRNO_COMLIB_H_
#define _MC_ERRNO_COMLIB_H_
static const char* comlib_error[] = {
static const char* comlib_error[] = {//进行了对错误类型的判断，从1000开始到1063分别为不同的错误类型
"1000 Reserved",
"1001 Invalid argument",
"1002 Memeory allocate error",

510
src/gausskernel/cbb/communication/libcomm_utils/libcomm_memory.cpp
View File

@ -65,7 +65,6 @@
#define static
#endif
#define STREAM_SCAN_FINISH 'F'
#define STREAM_SCAN_WAIT 'W'
#define STREAM_SCAN_DATA 'D'
@ -73,9 +72,12 @@
extern bool executorEarlyStop();
/* release memory of communication layer, just for LLT */
// 此函数的功能是释放通信层的内存
int gs_release_comm_memory()
{
// 构造一个通信上下文的对象，切换到全局通信内存上下文，用于管理通信内存。
AutoContextSwitch commContext(g_instance.comm_cxt.comm_global_mem_cxt);
// 调用函数gs_r_release_comm_memory来释放通信内存。
gs_r_release_comm_memory();
return 0;
}
@ -85,27 +87,41 @@ int gs_release_comm_memory()
*
* @param[IN] key_s: stream key
*/
void gs_memory_init_entry(StreamSharedContext* sharedContext, int consumerNum, int producerNum)
// 此函数用于初始化一个关于流和内存使用情况的哈希表条目
void gs_memory_init_entry(StreamSharedContext *sharedContext, int consumerNum, int producerNum)
{
struct hash_entry* entry = NULL;
struct hash_entry** poll_entrys = NULL;
struct hash_entry*** quota_entrys = NULL;
// 定义一个指向哈希表条目的指针
struct hash_entry *entry = NULL;
// 定义一个指向哈希表条目的指针的指针，用于存储与消费者相关的条目
struct hash_entry **poll_entrys = NULL;
// 定义一个指向哈希表条目的指针的指针的指针，用于存储与生产者相关的条目
struct hash_entry ***quota_entrys = NULL;
poll_entrys = (struct hash_entry**)palloc(sizeof(struct hash_entry*) * consumerNum);
quota_entrys = (struct hash_entry***)palloc(sizeof(struct hash_entry**) * consumerNum);
for (int i = 0; i < consumerNum; i++) {
entry = (struct hash_entry*)palloc(sizeof(struct hash_entry));
// 为每个消费者分配内存以存储哈希表条目指针
poll_entrys = (struct hash_entry **)palloc(sizeof(struct hash_entry *) * consumerNum);
// 为每个消费者分配内存以存储生产者相关的哈希表条目指针
quota_entrys = (struct hash_entry ***)palloc(sizeof(struct hash_entry **) * consumerNum);
// 遍历所有消费者
for (int i = 0; i < consumerNum; i++)
{
// 为每个消费者分配一个哈希表条目，并初始化它
entry = (struct hash_entry *)palloc(sizeof(struct hash_entry));
(void)entry->_init();
// 将当前消费者的哈希表条目存储在poll_entrys中
poll_entrys[i] = entry;
quota_entrys[i] = (struct hash_entry**)palloc(sizeof(struct hash_entry*) * producerNum);
for (int j = 0; j < producerNum; j++) {
entry = (struct hash_entry*)palloc(sizeof(struct hash_entry));
// 为每个生产者分配内存以存储哈希表条目指针
quota_entrys[i] = (struct hash_entry **)palloc(sizeof(struct hash_entry *) * producerNum);
// 遍历所有的生产者
for (int j = 0; j < producerNum; j++)
{
// 为每个生产者分配一个哈希表条目，并初始化它
entry = (struct hash_entry *)palloc(sizeof(struct hash_entry));
(void)entry->_init();
// 将当前生产者的哈希表条目存储在quota_entrys中
quota_entrys[i][j] = entry;
}
}
// 将poll_entrys和quota_entrys存储在共享上下文中，以便后续使用
sharedContext->poll_entrys = poll_entrys;
sharedContext->quota_entrys = quota_entrys;
}
@ -117,38 +133,51 @@ void gs_memory_init_entry(StreamSharedContext* sharedContext, int consumerNum, i
* @param[IN] sharedContext: context for shared memory stream
* @param[IN] nthChannel: destination consumer
*/
void gs_message_by_memory(StringInfo buf, StreamSharedContext* sharedContext, int nthChannel)
// 此函数的作用是通过内存发送错误/通知消息，其中参数buf为错误或者通知的字符串，sharedContext为共享内存流上下文，nthChannel为目标消费者
void gs_message_by_memory(StringInfo buf, StreamSharedContext *sharedContext, int nthChannel)
{
// 目标缓冲区，用于存储要发送的消息
StringInfo buf_dst = NULL;
struct hash_entry* entry = NULL;
// 哈希表条目，用于管理共享内存流
struct hash_entry *entry = NULL;
/* Copy Error/Notice messages to shared context. */
// 将错误/通知消息复制到共享上下文中
buf_dst = sharedContext->messages[nthChannel][u_sess->stream_cxt.smp_id];
/*
* If producer is waked up and shared buffer has been consumed while waiting,
* it can continue to append data to its messages of sharedContext.
*/
// 如果生产者在等待期间被唤醒，并且共享缓冲区已被消耗，它可以继续将其数据追加到sharedContext的消息中
entry = sharedContext->quota_entrys[nthChannel][u_sess->stream_cxt.smp_id];
while (buf_dst->len > 0) {
// 当目标缓冲区中还有未处理的数据时
while (buf_dst->len > 0)
{
// 等待一段时间，直到可以继续处理数据
(void)entry->_timewait(SINGLE_WAITQUOTA);
}
// 将源缓冲区的数据追加到目标缓冲区中
appendBinaryStringInfo(buf_dst, buf->data, buf->len);
// 更新目标缓冲区的游标位置
buf_dst->cursor = buf->cursor;
/* Send signal to dest consumer. */
// 向目标消费者发送信号。
entry = sharedContext->poll_entrys[nthChannel];
// 发送信号通知目标消费者有新消息到达
entry->_signal();
// 释放源缓冲区的数据内存
pfree(buf->data);
// 将源缓冲区的数据指针置为NULL，避免悬挂指针
buf->data = NULL;
}
void gs_memory_disconnect(StreamSharedContext* sharedContext, int nthChannel)
// 此函数的作用是断开内存连接
void gs_memory_disconnect(StreamSharedContext *sharedContext, int nthChannel)
{
struct hash_entry* entry = NULL;
// 定义一个指向哈希表条目的指针
struct hash_entry *entry = NULL;
// 将指定通道的数据状态设置为连接错误
sharedContext->dataStatus[nthChannel][u_sess->stream_cxt.smp_id] = CONN_ERR;
// 获取指定通道的轮询条目
entry = sharedContext->poll_entrys[nthChannel];
// 向轮询条目发送信号，通常用于通知其他进程或线程发生了某种事件或状态变化
entry->_signal();
}
@ -159,16 +188,27 @@ void gs_memory_disconnect(StreamSharedContext* sharedContext, int nthChannel)
* @param[IN] sharedContext: context for shared memory stream
* @param[IN] nthChannel: destination consumer
*/
bool gs_is_databuff_empty(StreamSharedContext* sharedContext, int nthChannel)
// 此函数的作用是判断数据缓冲区是否为空，其中参数sharedContext为共享内存流上下文，nthChannel为目标消费者
bool gs_is_databuff_empty(StreamSharedContext *sharedContext, int nthChannel)
{
if (sharedContext->vectorized) {
VectorBatch* batch = sharedContext->sharedBatches[nthChannel][u_sess->stream_cxt.smp_id];
if (batch->m_rows == 0) {
// 判断是否启用了向量化处理
if (sharedContext->vectorized)
{
// 获取指定通道和会话的共享批处理对象
VectorBatch *batch = sharedContext->sharedBatches[nthChannel][u_sess->stream_cxt.smp_id];
// 如果批处理的行数为0，则缓冲区为空
if (batch->m_rows == 0)
{
return true;
}
} else {
TupleVector* tupleVec = sharedContext->sharedTuples[nthChannel][u_sess->stream_cxt.smp_id];
if (tupleVec->tuplePointer == 0) {
}
else
{
// 获取指定通道和会话的共享元组向量对象
TupleVector *tupleVec = sharedContext->sharedTuples[nthChannel][u_sess->stream_cxt.smp_id];
// 如果元组指针为0，则缓冲区为空
if (tupleVec->tuplePointer == 0)
{
return true;
}
}
@ -185,94 +225,134 @@ bool gs_is_databuff_empty(StreamSharedContext* sharedContext, int nthChannel)
* @param[IN] nthChannel: destination consumer
* @param[IN] nthRow: the Nth row to be sent in batch
*/
// 此函数的作用是通过共享内存向本地消费者发送消息，参数tuple为要发送的元组，batchsrc为要发送的批次，sharedContext为共享内存流上下文，nthChannel为目标消费者，nthRow为批次中要发送的第n行
void gs_memory_send(
TupleTableSlot* tuple, VectorBatch* batchsrc, StreamSharedContext* sharedContext, int nthChannel, int nthRow)
TupleTableSlot *tuple, VectorBatch *batchsrc, StreamSharedContext *sharedContext, int nthChannel, int nthRow)
{
VectorBatch* batch = NULL;
TupleVector* tupleVec = NULL;
// 定义一个批处理指针，用于存储批处理对象
VectorBatch *batch = NULL;
// 定义一个元组向量指针，用于存储元组向量对象
TupleVector *tupleVec = NULL;
// 定义一个布尔型变量，用于存储是否可以发送的状态
bool ready_to_send = false;
// 定义一个数据状态变量，用于存储数据的状态
DataStatus dataStatus;
struct hash_entry* entry = NULL;
// 定义一个哈希表条目指针，用于存储共享内存流上下文的哈希表条目
struct hash_entry *entry = NULL;
// 报告等待状态，将当前状态设置为等待刷新数据状态
WaitState oldStatus = pgstat_report_waitstatus_comm(STATE_WAIT_FLUSH_DATA,
u_sess->pgxc_cxt.PGXCNodeId,
-1,
u_sess->stream_cxt.producer_obj->getParentPlanNodeId(),
global_node_definition ? global_node_definition->num_nodes : -1);
u_sess->pgxc_cxt.PGXCNodeId,
-1,
u_sess->stream_cxt.producer_obj->getParentPlanNodeId(),
global_node_definition ? global_node_definition->num_nodes : -1);
// 记录时间，开始发送数据
StreamTimeSendStart(t_thrd.pgxc_cxt.GlobalNetInstr);
// 获取指定通道和会话的共享内存流上下文的哈希表条目
entry = sharedContext->quota_entrys[nthChannel][u_sess->stream_cxt.smp_id];
for (;;) {
/* Check for interrupt at the beginning of the loop. */
// 进入无限循环，直到发送完成或发生中断等条件退出循环
for (;;)
{
// 在循环开始处检查中断。如果发生中断，则立即退出循环
CHECK_FOR_INTERRUPTS();
/* Check if we should early stop. */
/* Quit if the connection close, especially in a early close case. */
if (executorEarlyStop() || sharedContext->is_connect_end[nthChannel][u_sess->stream_cxt.smp_id]) {
// 检查是否需要提前停止。如果连接关闭，特别是在提前关闭的情况下，则退出循环
if (executorEarlyStop() || sharedContext->is_connect_end[nthChannel][u_sess->stream_cxt.smp_id])
{
// 恢复等待状态为原始状态
(void)pgstat_report_waitstatus(oldStatus);
return;
}
// 获取指定通道和会话的数据状态
dataStatus = sharedContext->dataStatus[nthChannel][u_sess->stream_cxt.smp_id];
/* Break the loop if we find quota. */
// 如果数据状态为DATA_EMPTY且（在__aarch64__架构下，数据缓冲区为空），或者数据状态为DATA_PREPARE，则跳出循环
if ((dataStatus == DATA_EMPTY
#ifdef __aarch64__
&& gs_is_databuff_empty(sharedContext, nthChannel)
#endif
) ||
dataStatus == DATA_PREPARE) {
) ||
dataStatus == DATA_PREPARE)
{
break;
}
// 记录时间，开始等待配额
StreamTimeWaitQuotaStart(t_thrd.pgxc_cxt.GlobalNetInstr);
// 调用entry的_timewait方法，传入SINGLE_WAITQUOTA作为参数，等待配额
(void)entry->_timewait(SINGLE_WAITQUOTA);
// 记录时间，结束等待配额
StreamTimeWaitQuotaEnd(t_thrd.pgxc_cxt.GlobalNetInstr);
}
// 记录时间，开始复制数据
StreamTimeCopyStart(t_thrd.pgxc_cxt.GlobalNetInstr);
/* Copy data to shared context. */
if (sharedContext->vectorized) {
// 将数据复制到共享上下文
if (sharedContext->vectorized)
{
// 如果启用了向量化处理，则断言共享批处理对象不为空
Assert(sharedContext->sharedBatches != NULL);
// 获取指定通道和会话的共享批处理对象
batch = sharedContext->sharedBatches[nthChannel][u_sess->stream_cxt.smp_id];
/* data copy */
if (-1 == nthRow) {
/* Do deep copy of all rows, for local roundrobin & local broadcast. */
// 如果nthRow为-1，则对所有行进行深度复制，用于本地循环和本地广播
if (-1 == nthRow)
{
// Assert批处理的行数为0，因为要进行所有行的深度复制
Assert(batch->m_rows == 0);
// 进行深度复制
batch->Copy<true, false>(batchsrc);
// 设置可以发送的状态为true
ready_to_send = true;
} else {
}
else
{
// 复制指定行的数据
batch->CopyNth(batchsrc, nthRow);
if (BatchMaxSize == batch->m_rows) {
// 如果批处理的行数等于BatchMaxSize，则设置可以发送的状态为true
if (BatchMaxSize == batch->m_rows)
{
ready_to_send = true;
}
}
} else {
}
else
{
// 如果未启用向量化处理，则断言共享元组向量对象不为空
Assert(sharedContext->sharedTuples != NULL);
// 获取指定通道和会话的共享元组向量对象
tupleVec = sharedContext->sharedTuples[nthChannel][u_sess->stream_cxt.smp_id];
// 获取元组指针
int n = tupleVec->tuplePointer;
// 复制元组到元组向量
ExecCopySlot(tupleVec->tupleVector[n], tuple);
// 元组指针加1
tupleVec->tuplePointer++;
if (TupleVectorMaxSize == tupleVec->tuplePointer) {
// 如果元组指针等于TupleVectorMaxSize，则设置可以发送的状态为true
if (TupleVectorMaxSize == tupleVec->tuplePointer)
{
ready_to_send = true;
}
}
// 记录时间，结束复制数据
StreamTimeCopyEnd(t_thrd.pgxc_cxt.GlobalNetInstr);
/* send the signal if copy finished */
if (ready_to_send) {
// 如果数据已经准备好发送，则执行以下代码块
if (ready_to_send)
{
#ifdef __aarch64__
// 在__aarch64__架构下，执行内存屏障操作，确保内存操作的正确顺序
pg_memory_barrier();
#endif
/* set flag */
// 设置数据状态为DATA_READY，表示数据已经准备好
sharedContext->dataStatus[nthChannel][u_sess->stream_cxt.smp_id] = DATA_READY;
/* send signal */
entry = sharedContext->poll_entrys[nthChannel];
entry->_signal();
} else {
}
else
{
// 如果数据还没有准备好，则将数据状态设置为DATA_PREPARE，表示数据正在准备中
sharedContext->dataStatus[nthChannel][u_sess->stream_cxt.smp_id] = DATA_PREPARE;
}
// 记录时间，结束发送数据
StreamTimeSendEnd(t_thrd.pgxc_cxt.GlobalNetInstr);
// 恢复等待状态为原始状态
(void)pgstat_report_waitstatus(oldStatus);
}
@ -282,17 +362,22 @@ void gs_memory_send(
* @param[IN] node: stream state
* @return bool: true -- found data
*/
// 此函数用于从流状态的缓冲区中获取一个元组，参数node为流状态
FORCE_INLINE
bool gs_return_tuple(StreamState* node)
bool gs_return_tuple(StreamState *node)
{
TupleVector* tupleVec = node->tempTupleVec;
if (tupleVec->tuplePointer == 0) {
// 获取流状态的临时元组向量
TupleVector *tupleVec = node->tempTupleVec;
// 如果元组指针为0，表示没有数据可返回
if (tupleVec->tuplePointer == 0)
{
return false;
}
// 元组指针减1，因为我们要返回的是当前指针指向的元组
tupleVec->tuplePointer--;
// 获取当前元组指针的索引
int n = tupleVec->tuplePointer;
// 将结果元组槽设置为当前元组指针指向的元组
node->ss.ps.ps_ResultTupleSlot = tupleVec->tupleVector[n];
return true;
@ -305,53 +390,67 @@ bool gs_return_tuple(StreamState* node)
* @param[IN] loc: data location
* @return bool: true -- found data
*/
bool gs_consume_memory_data(StreamState* node, int loc)
// 此函数的作用是从共享内存中消费本地生产者的数据，参数node为流状态，loc为数据位置
bool gs_consume_memory_data(StreamState *node, int loc)
{
StreamSharedContext* sharedContext = node->sharedContext;
// 获取流状态的共享上下文
StreamSharedContext *sharedContext = node->sharedContext;
// 记录时间，开始网络工作时间拷贝
NetWorkTimeCopyStart(t_thrd.pgxc_cxt.GlobalNetInstr);
/* Take data from the shared context. */
if (sharedContext->vectorized) {
VectorBatch* batchsrc = sharedContext->sharedBatches[u_sess->stream_cxt.smp_id][loc];
VectorBatch* batchdst = ((VecStreamState*)node)->m_CurrentBatch;
if (batchsrc->m_rows == 0) {
// 如果共享上下文已经向量化,从共享上下文中获取数据
if (sharedContext->vectorized)
{
// 获取源批处理对象和目标批处理对象
VectorBatch *batchsrc = sharedContext->sharedBatches[u_sess->stream_cxt.smp_id][loc];
VectorBatch *batchdst = ((VecStreamState *)node)->m_CurrentBatch;
// 如果源批处理的行数为0，表示没有数据可消费，返回false
if (batchsrc->m_rows == 0)
{
return false;
}
// 将源批处理的数据复制到目标批处理，进行深复制，不重置源批处理
batchdst->Copy<true, false>(batchsrc);
// 重置源批处理
batchsrc->Reset();
} else {
TupleVector* tuplesrc = sharedContext->sharedTuples[u_sess->stream_cxt.smp_id][loc];
TupleVector* tupledst = node->tempTupleVec;
}
else
{
// 如果共享上下文未向量化,获取源元组向量对象和目标元组向量对象
TupleVector *tuplesrc = sharedContext->sharedTuples[u_sess->stream_cxt.smp_id][loc];
TupleVector *tupledst = node->tempTupleVec;
if (tuplesrc->tuplePointer == 0) {
// 如果源元组指针为0，表示没有数据可消费，返回false
if (tuplesrc->tuplePointer == 0)
{
return false;
}
for (int i = 0; i < tuplesrc->tuplePointer; i++) {
// 将源元组向量的数据复制到目标元组向量
for (int i = 0; i < tuplesrc->tuplePointer; i++)
{
(void)ExecCopySlot(tupledst->tupleVector[i], tuplesrc->tupleVector[i]);
}
// 设置目标元组指针为源元组指针，重置源元组指针
tupledst->tuplePointer = tuplesrc->tuplePointer;
tuplesrc->tuplePointer = 0;
// 返回元组
(void)gs_return_tuple(node);
}
// 记录时间，结束网络工作时间拷贝
NetWorkTimeCopyEnd(t_thrd.pgxc_cxt.GlobalNetInstr);
struct hash_entry* entry = NULL;
struct hash_entry *entry = NULL;
// 获取配额条目
entry = sharedContext->quota_entrys[u_sess->stream_cxt.smp_id][loc];
// 如果编译在aarch64架构下，执行内存屏障操作，确保内存操作的正确顺序
#ifdef __aarch64__
pg_memory_barrier();
#endif
/* Reset flag */
// 重置标志位
sharedContext->dataStatus[u_sess->stream_cxt.smp_id][loc] = DATA_EMPTY;
/* send signal */
// 发送信号
entry->_signal();
// 更新扫描位置
node->sharedContext->scanLoc[u_sess->stream_cxt.smp_id] = loc;
return true;
}
@ -364,97 +463,139 @@ bool gs_consume_memory_data(StreamState* node, int loc)
* STREAM_SCAN_WAIT -- still need to poll to wait for data.
* STREAM_SCAN_FINISH -- stream scan finished.
*/
char gs_find_memory_data(StreamState* node, int* waitnode_count)
// 此函数的作用是从生产者状态扫描数据，参数node为流状态
char gs_find_memory_data(StreamState *node, int *waitnode_count)
{
// 定义一个数据状态变量
DataStatus dataStatus;
// 定义一个字符串信息变量，初始值为NULL
StringInfo buf = NULL;
// 获取上次扫描的位置
int scanLoc = node->sharedContext->scanLoc[u_sess->stream_cxt.smp_id];
// 定义一个计数器变量，初始值为上次扫描的位置
int i = scanLoc;
// 定义一个标志位，表示扫描是否完成，初始值为true
bool finished = true;
// 定义一个标志位，表示连接是否结束，初始值为false
bool is_conn_end = false;
// 定义一个计数器，用于统计需要等待的节点数量，初始值为0
int waitnodeCount = 0;
struct hash_entry* entry = NULL;
// 定义一个哈希表条目指针，初始值为NULL
struct hash_entry *entry = NULL;
/* Check if there is available data, and scan from last time location. */
do {
// 检查是否存在目标数据，并且从最后位置开始寻找
do
{
i++;
if (i == node->conn_count) {
// 如果计数器等于连接数，则重置为0
if (i == node->conn_count)
{
i = 0;
}
/* Update scan location. */
// 更新扫描位置
node->sharedContext->scanLoc[u_sess->stream_cxt.smp_id] = i;
// 获取当前位置的数据状态
dataStatus = node->sharedContext->dataStatus[u_sess->stream_cxt.smp_id][i];
// 获取当前位置的连接是否结束状态
is_conn_end = node->sharedContext->is_connect_end[u_sess->stream_cxt.smp_id][i];
if (!is_conn_end) {
if (!is_conn_end)
{
// 设置扫描完成标志位为false
finished = false;
// 需要等待的节点数量加1
waitnodeCount++;
}
/*
* Firstly, we handle error or notice messages.
* If an error occured, we should stop scan now.
* If an notice occured, we can still receive data.
*/
// 首先，我们处理错误或通知消息。如果错误，我们应该立即停止。如果发生了通知，我们仍然可以接收数据
// 从共享上下文中获取消息，这是一个字符串信息（StringInfo）结构，其中包含了消息的数据和长度等信息
buf = node->sharedContext->messages[u_sess->stream_cxt.smp_id][i];
if (buf->len > 0) {
if (buf->cursor == 'E') {
// 如果消息的长度大于0，即存在消息
if (buf->len > 0)
{
// 如果消息的游标为'E'，表示这是一个错误消息
if (buf->cursor == 'E')
{
// 调用函数处理流错误，参数为节点，错误消息的数据和长度
HandleStreamError(node, buf->data, buf->len);
// 返回一个标识，表示流扫描结束
return STREAM_SCAN_FINISH;
} else if (buf->cursor == 'N') {
}
// 如果消息的游标为'N'，表示这是一个通知消息
else if (buf->cursor == 'N')
{
// 调用函数处理流通知，参数为节点，通知消息的数据和长度
HandleStreamNotice(node, buf->data, buf->len);
// 重置字符串信息，清空游标和数据
resetStringInfo(buf);
/* After one notice message has handled, send signal and wake up the dest producer. */
// 在处理完一个通知消息后，发送信号并唤醒目标生产者
entry = node->sharedContext->quota_entrys[u_sess->stream_cxt.smp_id][i];
// 获取配额条目，可能是为了记录或控制生产者的行为
entry->_signal();
// 返回一个标识，表示流扫描需要等待
return STREAM_SCAN_WAIT;
}
}
switch (dataStatus) {
case DATA_EMPTY:
break;
case DATA_PREPARE:
/* Take the rest data away when the connection is end. */
if (is_conn_end) {
/* Return data if any. */
if (gs_consume_memory_data(node, i)) {
return STREAM_SCAN_DATA;
}
}
break;
case DATA_READY:
if (gs_consume_memory_data(node, i)) {
// 根据dataStatus的值选择执行的代码块
switch (dataStatus)
{
// 如果dataStatus的值为DATA_EMPTY,不执行任何操作
case DATA_EMPTY:
break;
// 如果dataStatus的值为DATA_PREPARE
case DATA_PREPARE:
// 当连接结束的时候带走其余的数据
if (is_conn_end)
{
/* Return data if any. */
// 如果有数据，通过调用gs_consume_memory_data函数来消耗数据
if (gs_consume_memory_data(node, i))
{
// 返回STREAM_SCAN_DATA，表示成功从生产者找到数据
return STREAM_SCAN_DATA;
} else {
break;
}
}
break;
case CONN_ERR:
ereport(ERROR,
case DATA_READY:
// 通过调用gs_consume_memory_data函数来消耗数据
if (gs_consume_memory_data(node, i))
{
// 返回STREAM_SCAN_DATA，表示成功从生产者找到数据
return STREAM_SCAN_DATA;
}
else
{
break;
}
// 如果dataStatus的值为CONN_ERR，生成一个错误报告
case CONN_ERR:
ereport(ERROR,
(errcode(ERRCODE_STREAM_REMOTE_CLOSE_SOCKET),
errmsg("Failed to read response from Local Stream Node,"
" Detail: Node %s, Plan Node ID %u, SMP ID %d",
errmsg("Failed to read response from Local Stream Node,"
" Detail: Node %s, Plan Node ID %u, SMP ID %d",
g_instance.attr.attr_common.PGXCNodeName,
node->sharedContext->key_s.planNodeId,
i)));
break;
// dataStatus is enum,
default:
break;
break;
// dataStatus is enum,
default:
break;
}
} while (i != scanLoc);
// 将waitnodeCount的值赋给指针waitnode_count所指向的变量，为了返回等待节点的数量
*waitnode_count = waitnodeCount;
if (finished) {
if (finished)
{
// 返回STREAM_SCAN_FINISH，表示流扫描完成
return STREAM_SCAN_FINISH;
} else {
}
else
{
// 返回STREAM_SCAN_WAIT，表示仍然需要轮询等待数据
return STREAM_SCAN_WAIT;
}
}
@ -466,51 +607,68 @@ char gs_find_memory_data(StreamState* node, int* waitnode_count)
* @return bool: true -- successed to find data and need more data.
* false -- all connection finished or recerive error.
*/
bool gs_memory_recv(StreamState* node)
// 此函数的作用是从共享内存中接收本地流的数据，返回是否成功找到数据并需要更多数据，或者所有连接已完成或接收错误
bool gs_memory_recv(StreamState *node)
{
// 存储操作结果的字符变量
char result;
struct hash_entry* entry = NULL;
// 哈希表条目指针，初始化为NULL
struct hash_entry *entry = NULL;
// 获取流状态对应的哈希表条目
entry = node->sharedContext->poll_entrys[u_sess->stream_cxt.smp_id];
bool re = true;
// 初始化等待节点数为0
int waitnode_count = 0;
/* If there is already tuple in buffer, return the data at once. */
if (!node->sharedContext->vectorized && gs_return_tuple(node)) {
// 如果缓冲区中已有元组，则立即返回数据
if (!node->sharedContext->vectorized && gs_return_tuple(node))
{
return true;
}
for (;;) {
/* Check for interrupt at the beginning of the loop. */
for (;;)
{
// 检查是否有中断请求
CHECK_FOR_INTERRUPTS();
/* Check if we can early stop now. */
if (executorEarlyStop()) {
// 检查是否可以提前结束循环
if (executorEarlyStop())
{
re = false;
break;
}
/* Search all producers to find data. */
// 搜索所有生产者以查找数据，同时更新等待节点数
result = gs_find_memory_data(node, &waitnode_count);
if (result == STREAM_SCAN_DATA) {
if (result == STREAM_SCAN_DATA)
{
re = true;
break;
} else if (result == STREAM_SCAN_FINISH) {
}
// 如果所有连接已完成或接收错误
else if (result == STREAM_SCAN_FINISH)
{
re = false;
break;
}
// 定义一个旧的等待状态阶段变量，初始化为PHASE_NONE，表示当前没有等待状态
WaitStatePhase oldPhase = pgstat_report_waitstatus_phase(PHASE_NONE, true);
// 定义一个旧的等待状态变量，通过调用pgstat_report_waitstatus_comm函数来初始化。
// 该函数将等待状态设置为STATE_WAIT_NODE，表示当前正在等待节点响应。
// 还将当前节点的ID、等待节点的数量、计划节点的ID以及全局节点定义的数量作为参数传递给该函数
WaitState oldStatus = pgstat_report_waitstatus_comm(STATE_WAIT_NODE,
u_sess->pgxc_cxt.PGXCNodeId,
waitnode_count,
node->sharedContext->key_s.planNodeId,
global_node_definition ? global_node_definition->num_nodes : -1);
u_sess->pgxc_cxt.PGXCNodeId,
waitnode_count,
node->sharedContext->key_s.planNodeId,
global_node_definition ? global_node_definition->num_nodes : -1);
/* Poll to wait data from producers. */
// 开始网络时间轮询，用于度量网络操作的耗时
NetWorkTimePollStart(t_thrd.pgxc_cxt.GlobalNetInstr);
// 调用entry的_timewait方法，传入SINGLE_WAITQUOTA作为参数，用于等待生产者提供数据
(void)entry->_timewait(SINGLE_WAITQUOTA);
// 结束网络时间轮询
NetWorkTimePollEnd(t_thrd.pgxc_cxt.GlobalNetInstr);
// 重置等待状态阶段和等待状态为旧的状态
pgstat_reset_waitStatePhase(oldStatus, oldPhase);
}
@ -523,16 +681,20 @@ bool gs_memory_recv(StreamState* node)
* @param[IN] sharedContext: context for shared memory stream
* @param[IN] connNum: producer connection number
*/
void gs_memory_send_finish(StreamSharedContext* sharedContext, int connNum)
// 此函数用于通知所有相关的消费者没有更多数据可发送
void gs_memory_send_finish(StreamSharedContext *sharedContext, int connNum)
{
struct hash_entry* entry = NULL;
// 定义一个哈希表条目指针，初始化为NULL
struct hash_entry *entry = NULL;
for (int i = 0; i < connNum; i++) {
/* Set flags. */
for (int i = 0; i < connNum; i++)
{
// 设置标志位，表示连接已经结束
sharedContext->is_connect_end[i][u_sess->stream_cxt.smp_id] = true;
/* send signal */
// 获取当前连接对应的哈希表条目
entry = sharedContext->poll_entrys[i];
// 调用哈希表条目的_signal方法，发送信号且通知等操作
entry->_signal();
}
}
@ -544,12 +706,15 @@ void gs_memory_send_finish(StreamSharedContext* sharedContext, int connNum)
* @param[IN] connNum: producer connection number
* @param[IN] smpId: producer smp id
*/
void gs_memory_close_conn(StreamSharedContext* sharedContext, int connNum, int consumerId)
// 此函数用于设置与特定生产者的所有连接关闭
void gs_memory_close_conn(StreamSharedContext *sharedContext, int connNum, int consumerId)
{
struct hash_entry* entry = NULL;
// 定义一个哈希表条目指针，初始化为NULL
struct hash_entry *entry = NULL;
for (int i = 0; i < connNum; i++) {
/* Set flags. */
for (int i = 0; i < connNum; i++)
{
// 设置标志位，表示与特定生产者的连接已结束
sharedContext->is_connect_end[consumerId][i] = true;
/*
@ -557,8 +722,7 @@ void gs_memory_close_conn(StreamSharedContext* sharedContext, int connNum, int c
* in a query like "limit XXX", when consumer don't need data anymore,
* but the producers haven't send all data yet.
*/
entry = sharedContext->quota_entrys[consumerId][i];
entry->_signal();
entry = sharedContext->quota_entrys[consumerId][i]; // 获取当前连接对应的哈希表条目
entry->_signal(); // 调用哈希表条目的_signal方法，发送信号且通知等操作
}
}
}

180
src/gausskernel/cbb/communication/libcomm_utils/libcomm_queue.cpp
View File

@ -26,121 +26,155 @@
#include "libcomm_queue.h"
#include "libcomm_common.h"
int mc_queue_pop(struct mc_queue* q, int* e)
//该代码用于在环形队列之中取出一个元素，使用POP表达弹出一个元素，按照本函数意思，将环形队列Q中的头部指针q取出，将q中的值赋给e,同时，该代码使用了互斥锁（mutex）以确保在多线程环境下的安全操作。
int mc_queue_pop(
struct mc_queue *q,
int *e) // 定义一个函数mc_queue_pop，输入参数为一个mc_queue结构体的指针q和一个整型指针e。函数返回一个整型值。
{
if (q == NULL) {
if (q == NULL) { // 如果q为NULL，表示队列没有初始化，返回错误代码-1。
return -1;
}
LIBCOMM_PTHREAD_MUTEX_LOCK(&(q->lock));
if (q->is_empty == 1) {
LIBCOMM_PTHREAD_MUTEX_LOCK(&(q->lock)); // 锁定队列的互斥锁，确保同一时间只有一个线程可以操作队列。
if (q->is_empty == 1) { // 如果队列为空，解锁互斥锁并返回0，表示队列为空。
LIBCOMM_PTHREAD_MUTEX_UNLOCK(&(q->lock));
return 0;
}
*e = q->data[q->head];
q->data[q->head] = -1; // unnormal value
++(q->head);
*e = q->data[q->head]; // 取出队列头部的元素，并赋值给e指向的变量。
q->data[q->head] = -1; // 将队列头部的元素设为-1，表示该元素已经被取出。
if (q->head == q->size) {
++(q->head); // 将队列头部指针向后移动一位。
if (q->head == q->size) { // 如果队列头部指针已经到达队列尾部，将其重置为0。
q->head = 0;
}
if (q->head == q->tail) {
if (q->head == q->tail) { // 如果队列头部指针和尾部指针重合，表示队列已经为空。
q->is_empty = 1;
}
q->is_full = 0;
q->count--;
q->is_full = 0; // 队列取出一个元素后，队列不可能再是满的，所以将is_full设为0。
q->count--; // 队列的元素数量减1。
LIBCOMM_PTHREAD_MUTEX_UNLOCK(&(q->lock));
return 1;
LIBCOMM_PTHREAD_MUTEX_UNLOCK(&(q->lock)); // 解锁互斥锁，允许其他线程访问队列。
return 1; // 返回1，表示成功从队列中取出一个元素。
}
int mc_queue_push(struct mc_queue* q, int e)
//该函数定义为向环形队列之中添加元素，按照函数之中定义为向环形队列Q的q处添加数值为e的元素，同样，该代码使用了互斥锁（mutex）以确保在多线程环境下的安全操作。
int mc_queue_push(struct mc_queue* q, int e) // 定义一个函数mc_queue_push，输入参数为一个mc_queue结构体的指针q和一个整型值e。函数返回一个整型值。
{
if (q == NULL) { // 如果q为NULL，表示队列没有初始化，返回错误代码-1。
return -1;
}
LIBCOMM_PTHREAD_MUTEX_LOCK(&(q->lock)); // 锁定队列的互斥锁，确保同一时间只有一个线程可以操作队列。
if (q->is_full) { // 如果队列已满，解锁互斥锁并返回0，表示队列已满。
LIBCOMM_PTHREAD_MUTEX_UNLOCK(&(q->lock));
return 0;
}
q->data[q->tail] = e; // 将元素e添加到队列的尾部。
++(q->tail); // 将队列尾部指针向后移动一位。
if (q->tail == q->size) { // 如果队列尾部指针已经到达队列的末尾，将其重置为0。
q->tail = 0;
}
if (q->tail == q->head) { // 如果队列尾部指针和头部指针重合，表示队列已满。
q->is_full = 1;
}
q->is_empty = 0; // 添加一个元素后，队列不可能再是空的，所以将is_empty设为0。
q->count++; // 队列的元素数量加1。
LIBCOMM_PTHREAD_MUTEX_UNLOCK(&(q->lock)); // 解锁互斥锁，允许其他线程访问队列。
return 1; // 返回1，表示成功将一个元素添加到队列中。
}
//此函数意思为创建一个数据流循环队列。其中q为队列的初始指针，size为队列大小。同样，该代码使用了互斥锁（mutex）以确保在多线程环境下的安全操作。
int mc_queue_init(
struct mc_queue *q,
int size) // 定义一个函数mc_queue_init，输入参数为一个mc_queue结构体的指针q和一个整型值size，表示队列的大小。函数返回一个整型值。
{
if (q == NULL) {
if (q == NULL || size <= 1) { // 如果q为NULL或size小于等于1，表示输入参数无效，返回错误代码-1。
return -1;
}
LIBCOMM_PTHREAD_MUTEX_LOCK(&(q->lock));
if (q->is_full) {
LIBCOMM_PTHREAD_MUTEX_UNLOCK(&(q->lock));
return 0;
}
q->data[q->tail] = e;
++(q->tail);
if (q->tail == q->size) {
q->tail = 0;
}
if (q->tail == q->head) {
q->is_full = 1;
}
q->is_empty = 0;
q->count++;
LIBCOMM_PTHREAD_MUTEX_UNLOCK(&(q->lock));
return 1;
}
int mc_queue_init(struct mc_queue* q, int size)
{
if (q == NULL || size <= 1) {
return -1;
}
q->size = size - 1;
q->data = NULL;
LIBCOMM_MALLOC(q->data, (q->size * sizeof(int)), int);
q->size = size - 1; // 队列的大小为size-1，因为环形队列的索引从0开始。
q->data = NULL; // 初始化队列的数据指针为NULL。
LIBCOMM_MALLOC(q->data, (q->size * sizeof(int)),
int); // 动态分配队列的数据内存空间，大小为(size-1)*sizeof(int)。如果分配失败，返回错误代码-1。
if (q->data == NULL) {
return -1;
}
LIBCOMM_PTHREAD_MUTEX_INIT(&(q->lock), 0);
LIBCOMM_PTHREAD_MUTEX_INIT(&(q->lock), 0); // 初始化队列的互斥锁。
// 这里是一段注释，解释了该队列的设计目的：为了获取streamid，我们不使用stream 0。
// we desgin this for getting streamid, we do not use stream 0
//
q->is_empty = 1;
q->is_full = 0;
q->head = 0;
q->tail = 0;
q->count = 0;
q->pop = mc_queue_pop;
q->push = mc_queue_push;
q->is_empty = 1; // 初始化队列的状态标志位，表示队列为空。
q->is_full = 0; // 初始化队列的状态标志位，表示队列不满。
q->head = 0; // 初始化队列的头部指针为0。
q->tail = 0; // 初始化队列的尾部指针为0。
q->count = 0; // 初始化队列的元素数量为0。
q->pop = mc_queue_pop; // 设置队列的出队函数为mc_queue_pop。
q->push = mc_queue_push; // 设置队列的入队函数为mc_queue_push。
return 0;
return 0; // 返回0，表示成功初始化队列。
}
struct mc_queue* mc_queue_clear(struct mc_queue* q)
//此为清楚函数，清楚位于队列中位置q的元素，，并且最后返回清除之后的函数
struct mc_queue *mc_queue_clear(
struct mc_queue
*q) // 定义一个函数mc_queue_clear，输入参数为一个mc_queue结构体的指针q，函数返回一个mc_queue结构体的指针。
{
if (q == NULL) {
if (q == NULL) { // 如果q为NULL，表示队列没有初始化，返回NULL。
return NULL;
}
int data = 0;
int data = 0; // 定义一个整型变量data，用于暂存队列中的元素。
while (!q->is_empty) {
(void)q->pop(q, &data);
while (!q->is_empty) { // 当队列不为空时，循环执行以下操作。
(void)q->pop(
q,
&data); // 调用队列的出队函数pop，将队列头部的元素取出并赋值给data。使用(void)是为了防止函数返回值被忽略导致的警告。
}
return q;
return q; // 返回清空后的队列指针。
}
struct mc_queue* mc_queue_destroy(struct mc_queue* q)
struct mc_queue *mc_queue_destroy(
struct mc_queue
*q) // 定义一个函数mc_queue_destroy，输入参数为一个mc_queue结构体的指针q，函数返回一个mc_queue结构体的指针。
{
if (q == NULL) {
if (q == NULL) { // 如果q为NULL，表示队列没有初始化，返回NULL。
return NULL;
}
if (q->data == NULL) {
if (q->data == NULL) { // 如果队列的数据指针为NULL，表示队列未初始化或已经被销毁，直接返回队列指针。
return q;
}
q = mc_queue_clear(q);
LIBCOMM_PTHREAD_MUTEX_LOCK(&(q->lock));
mc_free(q->data);
q->data = NULL;
LIBCOMM_PTHREAD_MUTEX_UNLOCK(&(q->lock));
LIBCOMM_PTHREAD_MUTEX_DESTORY(&(q->lock));
q = mc_queue_clear(q); // 调用mc_queue_clear函数清空队列中的所有元素。
LIBCOMM_PTHREAD_MUTEX_LOCK(&(q->lock)); // 锁定队列的互斥锁，确保同一时间只有一个线程可以访问队列。
mc_free(q->data); // 释放队列的数据内存空间。
q->data = NULL; // 将队列的数据指针设为NULL。
LIBCOMM_PTHREAD_MUTEX_UNLOCK(&(q->lock)); // 解锁互斥锁，允许其他线程访问队列。
LIBCOMM_PTHREAD_MUTEX_DESTORY(&(q->lock)); // 销毁互斥锁。
return q;
return q; // 返回销毁后的队列指针。
}
/*libcomm_queue代码较为简单容易理解，其表达了对数据流环形队列的各种操作方法，其中使用五个基础函数，分别为
* 1，弹出函数mc_queue_pop(struct mc_queue *q,int *e)：该函数所代表为将循环队列之中位于位置q的元素弹出循环队列数据流，
* 将弹出的数据的值赋值给e已达到弹出函数的效用
* 2，插入函数mc_queue_push(struct mc_queue* q, int e)：该函数所代表为向循环队列数据列之中位置q处插入e之中的数值，
* 并将循环队列向后指，达到循环队列插入的效果
* 3，创建函数mc_queue_init(struct mc_queue *q,int size)：该函数所代表创建循环队列数据流，在位置q处创建大小为size的数据流。
* 并且利用LIBCOMM_MALLOC(q->data, (q->size * sizeof(int)),int); 对size大小的内存空间进行合理的空间分配
* 4，清除函数struct mc_queue *mc_queue_clear(struct mc_queue *q)：利用调用弹出函数，并将值付给一块内存空间并删除该内存空间达到清除的效果，
* 该函数表明清楚循环队列数据流之中位于q的数据。
* 5，销毁函数struct mc_queue *mc_queue_destroy(struct mc_queue*q)：调用清除函数来清除队列之中所有的数据。
*/

652
src/gausskernel/cbb/communication/libcomm_utils/libcomm_shakehands.cpp
View File

File diff suppressed because it is too large Load Diff

285
src/gausskernel/cbb/communication/libcomm_utils/libcomm_util.cpp
View File

@ -36,22 +36,24 @@
extern long libpq_used_memory;
extern long libcomm_used_memory;
extern long comm_peak_used_memory;
//这个函数设置了全局实例的通信上下文中的可用内存量。
// 参数usable_memory表示可用内存量，单位是KB。函数内部将其乘以1024，转换为字节。
// 然后将其保存在全局实例的通信上下文的commutil_cxt成员中的g_total_usable_memory成员变量中。
void gs_set_usable_memory(long usable_memory)
{
g_instance.comm_cxt.commutil_cxt.g_total_usable_memory = usable_memory * 1024;
}
//为数据恢复函数设立存储空间，进行数据的内容存放
void gs_set_memory_pool_size(long mem_pool)
{
g_instance.comm_cxt.commutil_cxt.g_memory_pool_size = mem_pool * 1024;
}
//获得内存空间，该内存空间为进行通信传输所需要的内存空间，分别为通信进行过程中的内存空间和提供应用程序接口的内存空间
long gs_get_comm_used_memory(void)
{
return libcomm_used_memory + libpq_used_memory;
}
//获取内存空间，该内存空间为通信传递所需的最大的内存空间，也就是为libcomm_used_memory + libpq_used_memory的最大值
long gs_get_comm_peak_memory(void)
{
return comm_peak_used_memory;
@ -65,40 +67,51 @@ Size gs_get_comm_context_memory(void)
return 0;
}
}
int cmp_addr(sockaddr_storage_t* addr1, sockaddr_storage_t* addr2)
/*这个函数是用来比较两个套接字地址（sockaddr_storage_t类型）是否相等的。
首先如果这两个地址的协议族（IPv4和IPv6）不同，返回-1。如果相同开始进行判断，如果是IPv6的情况，
判断接口是否相等，如果相等返回IPv6的地址位置，地址大小。同理对IPv4进行判断，如果接口对应返回IPv6的地址位置，
地址大小。该函数为地址判断形函数，在满足地址相同的情况下对地址的接口类型进行判断，观察是否满足函数调用原则，如果满足则对函数的地址进行输出。*/
int cmp_addr(sockaddr_storage_t *addr1,
sockaddr_storage_t *addr2) // 定义函数cmp_addr，接收两个指向sockaddr_storage_t类型的指针作为参数
{
if (addr1->sa.sa_family != addr2->sa.sa_family) {
return -1;
if (addr1->sa.sa_family != addr2->sa.sa_family) { // 如果这两个地址的协议族（比如IPv4和IPv6）不同
return -1; // 返回-1表示这两个地址不相等
}
switch (addr1->sa.sa_family) {
case AF_INET6:
if (addr1->v6.sin6_port != addr2->v6.sin6_port) {
return -1;
switch (addr1->sa.sa_family) { // 根据协议族进行不同的比较
case AF_INET6: // 如果是IPv6
if (addr1->v6.sin6_port != addr2->v6.sin6_port) { // 如果这两个地址的端口号不同
return -1; // 返回-1表示这两个地址不相等
}
return memcmp(&addr1->v6.sin6_addr, &addr2->v6.sin6_addr, sizeof(addr1->v6.sin6_addr));
case AF_INET:
if (addr1->v4.sin_port != addr2->v4.sin_port) {
return -1;
return memcmp(
&addr1->v6.sin6_addr, &addr2->v6.sin6_addr,
sizeof(addr1->v6.sin6_addr)); // 使用memcmp函数比较IPv6地址部分是否相等，如果相等返回0，不相等返回非0值
case AF_INET: // 如果是IPv4
if (addr1->v4.sin_port != addr2->v4.sin_port) { // 如果这两个地址的端口号不同
return -1; // 返回-1表示这两个地址不相等
}
return memcmp(&addr1->v4.sin_addr, &addr2->v4.sin_addr, sizeof(addr1->v4.sin_addr));
default:
return -1;
return memcmp(
&addr1->v4.sin_addr, &addr2->v4.sin_addr,
sizeof(addr1->v4.sin_addr)); // 使用memcmp函数比较IPv4地址部分是否相等，如果相等返回0，不相等返回非0值
default: // 如果协议族不是IPv4也不是IPv6
return -1; // 返回-1表示这两个地址不相等
}
}
//查找程序错误的判断参数
void set_debug_mode(bool mod)
{
g_instance.comm_cxt.commutil_cxt.g_debug_mode = mod;
}
//时间设置函数的判断参数
void set_timer_mode(bool mod)
{
g_instance.comm_cxt.commutil_cxt.g_timer_mode = mod;
}
//生命函数的判断参数
void set_stat_mode(bool mod)
{
g_instance.comm_cxt.commutil_cxt.g_stat_mode = mod;
}
//延时函数的判断参数
void set_no_delay(bool mod)
{
g_instance.comm_cxt.commutil_cxt.g_no_delay = mod;
@ -107,54 +120,55 @@ void set_no_delay(bool mod)
// set FL or FD attribute of socket by parameter FL_OR_FD,
// FL_OR_FD is 0 for setting FL (O_NONBLOCK)
// FL_OR_FD is 1 for setting FD (FD_CLOEXEC)
//
int set_socketopt(int sock, int FL_OR_FD, long arg)
//这段文字是对设置socket的FL或FD属性的一种说明。其中，FL_OR_FD是一个参数，当它值为0时，表示设置FL属性（标记为O_NONBLOCK）。这个O_NONBLOCK标记表示该socket将处于非阻塞模式，意味着对socket的操作不会阻塞程序的执行。当FL_OR_FD的值为1时，表示设置FD属性（标记为FD_CLOEXEC）。这个FD_CLOEXEC标记表示在执行新的程序或命令时，这个socket将被关闭。
//函数在开头对FL以及FD进行了选择，并且根据不同情况分别返回对应的FL或者FD的值。后面对socket的fcntl函数进行了判定，排除两种小于零的错误情况，最后返回RC值表示成功。
// 该函数对socket的类型进行判定，表达了何时为非阻塞模式，何时为运行模式，该函数表明了进程的状态，防止进程出现错误。 int set_socketopt(
int sock, int FL_OR_FD,
long arg) // 定义一个函数set_socketopt，接收三个参数：socket文件描述符，标志位FL_OR_FD，以及长整型参数arg
{
int flg, rc;
int get_cmd = 0;
int set_cmd = 0;
switch (FL_OR_FD) {
case 0: // FL
get_cmd = F_GETFL;
set_cmd = F_SETFL;
int flg, rc; // 定义两个整型变量flg和rc，用于存储fcntl函数的返回值和设置选项的结果
int get_cmd = 0; // 定义一个整型变量get_cmd，并初始化为0，用于存储获取选项的命令
int set_cmd = 0; // 定义一个整型变量set_cmd，并初始化为0，用于存储设置选项的命令
switch (FL_OR_FD) { // 根据FL_OR_FD的值进行不同的操作
case 0: // FL
get_cmd = F_GETFL; // 如果FL_OR_FD为0，即要获取的是文件状态标志，因此设置get_cmd为F_GETFL
set_cmd = F_SETFL; // 如果FL_OR_FD为0，即要设置的是文件状态标志，因此设置set_cmd为F_SETFL
break;
case 1: // FD
get_cmd = F_GETFD;
set_cmd = F_SETFD;
case 1: // FD
get_cmd = F_GETFD; // 如果FL_OR_FD为1，即要获取的是文件描述符标志，因此设置get_cmd为F_GETFD
set_cmd = F_SETFD; // 如果FL_OR_FD为1，即要设置的是文件描述符标志，因此设置set_cmd为F_SETFD
break;
default:
get_cmd = -1;
get_cmd = -1; // 如果FL_OR_FD的值既不是0也不是1，即未知的情况，设置get_cmd和set_cmd为-1，表示错误
set_cmd = -1;
break;
}
if (get_cmd == -1 || set_cmd == -1) {
LIBCOMM_ELOG(WARNING, "(set socket opt)\tUnkown: command type[%d].", FL_OR_FD);
return -1;
if (get_cmd == -1 || set_cmd == -1) { // 如果get_cmd或set_cmd的值为-1，表示未知的命令类型
LIBCOMM_ELOG(WARNING, "(set socket opt)\tUnkown: command type[%d].",
FL_OR_FD); // 使用日志库输出警告信息，说明未知的命令类型
return -1; // 返回-1表示错误
}
if ((flg = fcntl(sock, get_cmd, 0)) < 0) { // F_GETFL, F_GETFD
LIBCOMM_ELOG(WARNING,
"(set socket opt)\tFail to get fcntl[%d:%ld] of socket[%d]:fcntl failed,return[%d].",
get_cmd,
arg,
sock,
flg);
return flg;
if ((flg = fcntl(sock, get_cmd, 0)) < 0) { // 调用fcntl函数获取socket的选项值，将返回值存储在flg中
// 如果fcntl函数调用失败，即返回值小于0
LIBCOMM_ELOG(WARNING, "(set socket opt)\tFail to get fcntl[%d:%ld] of socket[%d]:fcntl failed,return[%d].",
get_cmd, arg, sock, flg);
// 使用日志库输出警告信息，说明fcntl获取选项失败
return flg; // 返回flg的值，表示错误
}
flg |= arg; // O_NONBLOCK, FD_CLOEXEC
if ((rc = fcntl(sock, set_cmd, flg)) < 0) { // F_SETFL, F_SETFL
LIBCOMM_ELOG(WARNING,
"(set socket opt)\tFail to set fcntl[%d:%ld] of socket[%d]:fcntl failed,return[%d].",
set_cmd,
arg,
sock,
flg);
return rc;
flg |= arg; // 将arg的值与flg的值进行按位或操作，即将arg的值添加到flg的值中，用于设置新的选项值
// O_NONBLOCK为非阻塞标志，FD_CLOEXEC为执行时关闭文件描述符标志
if ((rc = fcntl(sock, set_cmd, flg)) < 0) { // 调用fcntl函数设置socket的选项值，将返回值存储在rc中
// 如果fcntl函数调用失败，即返回值小于0
LIBCOMM_ELOG(WARNING, "(set socket opt)\tFail to set fcntl[%d:%ld] of socket[%d]:fcntl failed,return[%d].",
set_cmd, arg, sock, flg);
// 使用日志库输出警告信息，说明fcntl设置选项失败
return rc; // 返回rc的值，表示错误
}
return rc;
return rc; // 返回rc的值，表示成功
}
//32位字符复制函数，此处采取不会出现复制错误的复制程序
uint32 comm_get_cpylen(const char* src, uint32 max_len)
{
uint32 cpylen = 0;
@ -177,7 +191,7 @@ uint32 comm_get_cpylen(const char* src, uint32 max_len)
return cpylen;
}
//本函数在数据库之中起到了字符流判定的作用，其中引入静态只读字符，并根据传入的type进行判定，当type< 0或者type> 0的情况则对函数报错，输出UNKNOWN值
static const char* MAILBOX_STAT[MAIL_MAX_TYPE] = {"UNKNOWN", "READY", "RUN", "HOLD", "CLOSED", "TO_CLOSED"};
const char* stream_stat_string(int type)
@ -188,7 +202,9 @@ const char* stream_stat_string(int type)
return MAILBOX_STAT[type];
}
//本函数起到了控制消息信号判定作用，当type在规定值以内，对函数进行判定，
// 返回CTRL_MSG_STAT之中的操作指令，如果函数并不在规定范围之内，则将type值赋0，
// 进行UNKNOWN判定
static const char* CTRL_MSG_STAT[CTRL_MAX_TYPE] = {"UNKNOWN",
"REGIST",
"REGIST_CN",
@ -212,7 +228,9 @@ const char* ctrl_msg_string(int type)
return CTRL_MSG_STAT[type];
}
/*该函数为消息控制台函数，对消息进行控制，的那个消息处于控制区间内，返回MSG_OPER_STAT之中静态定义值，请注意，此处返回为READ_DATA_FROM_LOGIC + 1数值，
READ_DATA_FROM_LOGIC + 1 的操作是因为数组 MSG_OPER_STAT 的索引是从 0 开始的。
数组的长度为 READ_DATA_FROM_LOGIC + 1，这样可以确保数组的最后一个元素的索引为 READ_DATA_FROM_LOGIC。*/
static const char *MSG_OPER_STAT[READ_DATA_FROM_LOGIC + 1] = {
"send_some",
"secure_read",
@ -240,7 +258,10 @@ void printfcmsg(const char* caller, struct FCMSG_T* pfcmsg)
pfcmsg->query_id,
pfcmsg->nodename);
}
/*请注意，该函数极为复杂，该函数所讲述为一种传输机制cmailbox传输机制，该传输机制已在前面进行阐明，
printf_cmailbox_statistic函数用于输出cmailbox的统计信息，帮助用户了解cmailbox的使用情况和性能表现。
通过调用该函数并传入cmailbox对象和节点名称作为参数，可以打印出相关的统计信息，例如已发送的消息数量、已接收的消息数量、消息发送的成功率等。
该传输机制因其稳定且高效的原因在华为数据库opengauss之中得到了广泛的应用。*/
void printf_cmailbox_statistic(c_mailbox* cmailbox, char* nodename)
{
if (NULL == cmailbox->statistic || MAIL_UNKNOWN == cmailbox->state) {
@ -295,7 +316,9 @@ void printf_cmailbox_statistic(c_mailbox* cmailbox, char* nodename)
cmailbox->statistic->recv_loop_time,
cmailbox->statistic->recv_loop_count);
}
/*与cmailbox极为相似，pmailbox同样也是一种消息传递机制，也在前面有具体的解释。
其中，printf_pmailbox_statistic函数用于输出pmailbox的统计信息，帮助用户了解pmailbox的使用情况和性能表现。
通过调用该函数并传入pmailbox对象和节点名称作为参数，可以打印出相关的统计信息，例如已发送的消息数量、已接收的消息数量、消息发送的成功率等。*/
void printf_pmailbox_statistic(p_mailbox* pmailbox, char* nodename)
{
if (NULL == pmailbox->statistic || MAIL_UNKNOWN == pmailbox->state) {
@ -340,7 +363,7 @@ void printf_pmailbox_statistic(p_mailbox* pmailbox, char* nodename)
pmailbox->statistic->os_send_overhead,
pmailbox->statistic->producer_elapsed_time);
}
/*输出接口函数*/
void print_socket_info(int sock, struct tcp_info* info, bool sender)
{
mc_elog(LOG,
@ -370,7 +393,8 @@ void print_socket_info(int sock, struct tcp_info* info, bool sender)
// error logging function
//
extern uint32 GetTopTransactionIdIfAny(void);
//mc_elog 函数在 OpenGauss 数据库中的 libcomm_util.cpp 文件中是一个用于错误日志记录的函数。mc_elog
// 是 "multi-copy elog" 的缩写，其中 "elog" 是 PostgreSQL 中的一个错误日志系统。
void mc_elog(int elevel, const char* format, ...)
{
#define MSLEN 4
@ -420,44 +444,71 @@ void mc_elog(int elevel, const char* format, ...)
ss_rc = strncpy_s(timebuf + MSOFFSET, TIMELEN - MSOFFSET - 1, msbuf, MSLEN);
securec_check(ss_rc, "\0", "\0");
// 检查u_sess是否存在，u_sess代表一个用户会话
if (u_sess) {
// 如果u_sess存在，那么将相关信息打印到标准输出
fprintf(stdout,
"%s %s " /* timestamp with milliseconds */
"%ld.%d " /* session ID */
"%s " /* database name */
"%lu " /* process ID */
"%s " /* application name */
"%u " /* transaction ID (0 if none) */
"%s " /* DataNode name */
"%s " /* SQL state */
"%lu " /* u_sess->debug_query_id */
"%s: " /* log level */
"%s\n", /* error message */
timebuf,
timebuf + MSOFFSET + MSLEN + 1,
(long)(t_thrd.proc_cxt.MyStartTime),
t_thrd.myLogicTid,
(u_sess->proc_cxt.MyProcPort && u_sess->proc_cxt.MyProcPort->database_name &&
u_sess->proc_cxt.MyProcPort->database_name[0] != '\0')
? u_sess->proc_cxt.MyProcPort->database_name
: "[unknown]",
t_thrd.proc_cxt.MyProcPid,
(u_sess->proc_cxt.MyProcPort && u_sess->attr.attr_common.application_name &&
u_sess->attr.attr_common.application_name[0] != '\0')
? u_sess->attr.attr_common.application_name
: "[unknown]",
GetTopTransactionIdIfAny(),
g_instance.attr.attr_common.PGXCNodeName ? g_instance.attr.attr_common.PGXCNodeName
: g_instance.comm_cxt.localinfo_cxt.g_self_nodename,
"00000",
u_sess->debug_query_id,
(elevel == LOG) ? "[LIBCOMM] LOG" : "[LIBCOMM] WARNING",
msg);
// 时间戳，包括毫秒部分
"%s %s " /* timestamp with milliseconds */
// 会话ID
"%ld.%d " /* session ID */
// 数据库名称，如果未知则为"[unknown]"
"%s " /* database name */
// 进程ID
"%lu " /* process ID */
// 应用程序名称，如果未知则为"[unknown]"
"%s " /* application name */
// 事务ID，如果没有则为0
"%u " /* transaction ID (0 if none) */
// DataNode名称，如果没有则为g_instance.comm_cxt.localinfo_cxt.g_self_nodename
"%s " /* DataNode name */
// SQL状态，这里固定为"00000"
"%s " /* SQL state */
// u_sess的debug_query_id
"%lu " /* u_sess->debug_query_id */
// 日志级别，如果是LOG则为"[LIBCOMM] LOG"，否则为"[LIBCOMM] WARNING"
"%s: " /* log level */
// 错误信息
"%s\n", /* error message */
// 时间戳
timebuf,
// 时间戳的毫秒部分
timebuf + MSOFFSET + MSLEN + 1,
// 会话开始时间的长整型表示
(long)(t_thrd.proc_cxt.MyStartTime),
// 逻辑线程ID
t_thrd.myLogicTid,
// 数据库名称，如果未知则为"[unknown]"
(u_sess->proc_cxt.MyProcPort && u_sess->proc_cxt.MyProcPort->database_name &&
u_sess->proc_cxt.MyProcPort->database_name[0] != '\0')
? u_sess->proc_cxt.MyProcPort->database_name
: "[unknown]",
// 进程ID
t_thrd.proc_cxt.MyProcPid,
// 应用程序名称，如果未知则为"[unknown]"
(u_sess->proc_cxt.MyProcPort && u_sess->attr.attr_common.application_name &&
u_sess->attr.attr_common.application_name[0] != '\0')
? u_sess->attr.attr_common.application_name
: "[unknown]",
// 获取顶级事务ID，如果没有则为0
GetTopTransactionIdIfAny(),
// DataNode名称，如果没有则为g_instance.comm_cxt.localinfo_cxt.g_self_nodename
g_instance.attr.attr_common.PGXCNodeName ? g_instance.attr.attr_common.PGXCNodeName
: g_instance.comm_cxt.localinfo_cxt.g_self_nodename,
// SQL状态，这里固定为"00000"
"00000",
// u_sess的debug_query_id
u_sess->debug_query_id,
// 日志级别，如果是LOG则为"[LIBCOMM] LOG"，否则为"[LIBCOMM] WARNING"
(elevel == LOG) ? "[LIBCOMM] LOG" : "[LIBCOMM] WARNING",
// 错误信息
msg);
}
// 刷新标准输出缓冲区，确保所有输出都被立即显示
(void)fflush(stdout);
}
//该函数记录程序运行时长
static inline int mc_clock_gettime(struct timespec* ts)
{
return clock_gettime(CLOCK_MONOTONIC, ts);
@ -465,54 +516,74 @@ static inline int mc_clock_gettime(struct timespec* ts)
#define MC_NS_IN_SEC 1000000000ULL
// 定义一个名为mc_timers_us的函数，该函数没有参数，并返回一个无符号的64位整数，表示以微秒为单位的时间
uint64_t mc_timers_us(void)
{
// 定义一个无符号的64位整数变量ns_monotonic，用于存储获取到的时间值
uint64_t ns_monotonic;
// 定义一个timespec结构体变量ts，并初始化其所有成员为0
struct timespec ts = {0};
// 调用mc_clock_gettime函数，获取当前的单调时间，并将其存储在ts变量中
(void)mc_clock_gettime(&ts);
// 将获取到的时间值（纳秒为单位）转换为微秒，然后存储在ns_monotonic变量中
// 这里的转换是先将秒数部分转换为纳秒（乘以MC_NS_IN_SEC，即1秒的纳秒数），再加上原有的纳秒部分，然后再除以1000得到微秒
ns_monotonic = (uint64)((ts.tv_sec * MC_NS_IN_SEC) + (uint64)ts.tv_nsec) / 1000;
// 返回转换后的时间值（微秒为单位）
return ns_monotonic;
}
//同样为单位换算函数，换算单位为纳秒的千分之一
uint64 mc_timers_ms(void)
{
return mc_timers_us() / 1000;
}
// 定义一个名为comm_ipc_log_get_time的函数，该函数接收一个字符指针now_date和一个整数time_len作为参数
void comm_ipc_log_get_time(char *now_date, int time_len)
{
// 定义常量MS_LEN为4，表示毫秒字符串的长度
const int MS_LEN = 4;
// 定义常量BUF_LEN为MS_LEN的两倍，表示毫秒缓冲区的长度
const int BUF_LEN = MS_LEN * 2;
// 定义常量MS_OFFSET为19，表示毫秒字符串在now_date中的偏移量
const int MS_OFFSET = 19;
struct timeval tv = {0}; /* make time string start, like setup_formatted_log_time */
struct pg_tm* localtime = NULL;
// 定义一个timeval结构体变量tv，并初始化其所有成员为0，用于获取当前时间
struct timeval tv = {0}; /* make time string start, like setup_formatted_log_time */
// 定义一个pg_tm结构体指针localtime，用于存储本地时间
struct pg_tm *localtime = NULL;
// 定义一个pg_time_t类型变量stamp_time，用于存储时间戳
pg_time_t stamp_time;
// 定义一个长度为BUF_LEN的字符数组msbuf，用于存储毫秒字符串
char msbuf[BUF_LEN];
// 定义一个errno_t类型变量ss_rc，用于存储函数调用返回值
errno_t ss_rc = 0;
// 调用gettimeofday函数获取当前时间，并将其存储在tv变量中
(void)gettimeofday(&tv, NULL);
// 将获取到的时间戳（秒为单位）转换为pg_time_t类型，并存储在stamp_time变量中
stamp_time = (pg_time_t)tv.tv_sec;
// 调用pg_localtime函数将时间戳转换为本地时间，并存储在localtime变量中
localtime = pg_localtime(&stamp_time, log_timezone);
// 如果localtime不为NULL，则调用pg_strftime函数将本地时间格式化为指定格式的字符串，并将其存储在now_date变量中
/* leave room for milliseconds. */
if (localtime != NULL) {
(void)pg_strftime(now_date, time_len,
"%Y-%m-%d %H:%M:%S",
localtime);
(void)pg_strftime(now_date, time_len, "%Y-%m-%d %H:%M:%S", localtime);
}
/* 'paste' milliseconds into place. */
// 将微秒转换为毫秒，并将其存储在msbuf变量中
/* 'paste' milliseconds into place. */
ss_rc = snprintf_s(msbuf, sizeof(msbuf), MS_LEN, ".%03d", (int)(tv.tv_usec / 1000));
securec_check_ss(ss_rc, "\0", "\0");
// 将毫秒字符串复制到now_date变量的指定位置中
ss_rc = strncpy_s(now_date + MS_OFFSET, time_len - MS_OFFSET - 1, msbuf, MS_LEN);
securec_check(ss_rc, "\0", "\0");
}
//唤醒通道类构造函数
WakeupPipe::WakeupPipe()
{
InitPipe(m_normal_wakeup_pipes);
}
//唤醒通道类唤醒函数
void WakeupPipe::DoWakeup()
{
int errno_tmp = errno;
@ -521,7 +592,7 @@ void WakeupPipe::DoWakeup()
}
errno = errno_tmp;
}
//唤醒通道类创建通道函数
void WakeupPipe::InitPipe(int *input_pipes)
{
if (input_pipes == NULL) {
@ -543,12 +614,12 @@ void WakeupPipe::InitPipe(int *input_pipes)
m_ev.events = EPOLLIN;
m_ev.data.fd = input_pipes[0];
}
//唤醒通道类移除通道开始函数
void WakeupPipe::RemoveWakeupFd()
{
(void)epoll_ctl(m_epfd, EPOLL_CTL_DEL, m_normal_wakeup_pipes[WAKEUP_PIPE_START], NULL);
}
//唤醒通道类关闭通道函数
void WakeupPipe::ClosePipe(int *input_pipes)
{
close(input_pipes[WAKEUP_PIPE_START]);
@ -556,7 +627,7 @@ void WakeupPipe::ClosePipe(int *input_pipes)
input_pipes[WAKEUP_PIPE_START] = INVALID_FD;
input_pipes[WAKEUP_PIPE_END] = INVALID_FD;
}
//唤醒通道类析构函数
WakeupPipe::~WakeupPipe()
{
ClosePipe(m_normal_wakeup_pipes);