Heartbeat 通信层结构分析

发布时间：2022-10-19 12:46:31 所属栏目：Unix 来源：

导读：　　Heartbeat 通信层结构分析

　　前言

　　Heartbeat是Linux-HA开源项目发布的用于关键应用环境的HA软件名称。从1999以来到现在，历经1.2.x, 2.0.x等多个版本，在全球开源HA领域具有举足轻重的知名度, 应

　　Heartbeat 通信层结构分析

　　前言

　　Heartbeat是Linux-HA开源项目发布的用于关键应用环境的HA软件名称。从1999以来到现在，历经1.2.x, 2.0.x等多个版本，在全球开源HA领域具有举足轻重的知名度, 应用日益广泛，并且得到了一些主流Linux操作系统厂商的支持。

　　而通信层实现无疑是集群软件运行的最基本底层支撑。本文对通过分析Heartbeat源码，对其通信层基本结构和机制进行了分析和阐述。给出了基本数据结构和实现流程。

　　所有分析基于Heartbeat 2.0.4版本。

　　相关源码：

　　Heartbeat通信结构概述

　　主要分2种:

　　1.HBcomm 通信层PLUGIN (节点之间的进程通信)

　　实现主要是在各个媒介的Plugin里，通过PILS动态连接库加载。比如支持多播，单播，串口等通信方式。所有节点间通信PLUGIN模块放在lib/plugins/hbcomm/路径下。

　　2.Unix Domain Socket (节点内的进程通信)

　　/include/clplumbing/Ipc.h, IPC抽象层数据结构定义

　　/lib/clplumbing/ocf_ipc.c, IPC底层抽象实现

　　/lib/clplumbing/ipcsocket.c, IPC的unix域套接字具体实现

　　交叉点：

　　节点间和节点内2种通信方式的交接点在heartbeat.c的read_child(), write_child()等函数中，在这里实现消息的转发。

　　Heartbeat API

　　是基于ipc抽象层的Unix域实现基础上，用于满足heartbeat和client子模块之间的应用层通信需求。:

　　client_lib.c实现了Heartbeat API 客户端部分。

　　hb_api.c实现了heartbeat API服务器端部分。

　　图1：Heartbeat通信结构概图

　　上图描述了一个client子模块把消息通过Heartbeat通信机制发送到另一个节点相同模块的过程。

　　1. client子模块通过FIFO管道把消息发送到FIFO子进程fifo_child。为什么使用FIFO来进行通信呢，应该是有些进程不能很方便的和Heartbeat主进程建立Unix域IPC通道的关系，比如执行的脚本和集群管理程序, 集群状态查询程序。

　　2. FIFO子进程通过msgfromstream()从fifo管道收到消息后，利用事先建立好的和Heartbeat之间的IPC通道转发给Heartbeat主进程

　　3. 主进程判断消息是发给自己的则调用process_msg()进行处理，否则调用send_to_all_media()通过各个媒介的wchan通道发送给write_child子进程。

　　4. write_child子进程通过ipcmsgfromIPC()从主进程收到消息，调用各个媒介结构hb_media的write函数把消息发送到集群其他节点。

　　5. 其他节点的read_child子进程通过各个媒介结构hb_media的read函数读到消息后，使用事先和Heartbeat主进程建立的IPC通道发送消息到Heartbeat主进程

　　6. Heartbeat主进程通过msgfromIPC()收到消息后，调用process_clustermsg()函数进行处理。具体为，如果是主进程处理的消息调用HBDoMsgCallback进行处理，否则通过newstartha_monitor发送到各个client子进程

　　节点间通信Plugin

　　代码在lib/plugins/hbcomm/目录中

　　bcast.c /* 广播 */

　　mcast.c /* 多播 */

　　ucast.c /* 单播 */

　　openais.c /* openais */

　　serial.c /* 串口 */

　　ping.c /* icmp */

　　ping_group.c /* ping一组主机 */

　　hbaping.c /* 光纤总线适配器ping */

　　/* 这个结构的每个函数对应Plugin里的具体函数。*/

　　struct hb_media_fns {

　　struct hb_media*(*new) (const char * token); /* 建立媒介 */

　　int (*parse) (const char * options); /* 读取配置文件参数 */

　　int (*open) (struct hb_media *mp); /* 打开 */

　　int (*close) (struct hb_media *mp); /* 关闭 */

　　void* (*read) (struct hb_media *mp, int *len ); /* 读 */

　　int (*write) (struct hb_media *mp , void *msg, int len); /* 写 */

　　int (*mtype) (char **buffer); /* 获取媒介类型 */

　　int (*descr) (char **buffer); /* 获取媒介描述 */

　　int (*isping) (void); /* 是否ping类型媒介 */

　　};

　　hb_media_fns各功能函数调用之处:

　　new(): config.c的add_option函数

　　parse(): config.c的parse_config函数

　　open(): heartbeat.c的initialize_heartbeat函数

　　close(): heartbeat.c的initialize_heartbeat函数

　　read(): heartbeat.c的read_child函数

　　write(): heartbeat.c的write_child函数

　　mtype(): config.c的parse_config函数

　　descr(): config.c的parse_config函数

　　isping(): 在config.c和hb_api.c被调用

　　节点内IPC通信

　　IPC通信抽象层（include/clplumbing/ipc.h）

　　IPC抽象层数据结构概述：

　　(注：缩进的为该数据结构所属元素)

　　IPC_AUTH /* 安全认证数据结构 */

　　IPC_WAIT_CONNECTION /* 等待连接数据结构 */

　　IPC_WAIT_OPS /* 等待连接函数集 */

　　IPC_CHANNEL /* 通信管道数据结构 */

　　IPC_OPS /* 通信管道函数集 */

　　IPC_QUEUE /* 信息队列 */

　　ipc_bufpool /* 接收缓冲池，经处理转化为接收队列 */

　　IPC_MESSAGE /* IPC通信信息数据结构 */

　　IPC_CHANNEL /* 信息所属通信管道 */

　　SOCKET_MSG_HEAD /* 信息头数据结构 */

　　其中2种主要抽象数据结构：

　　/* server端等待客户端的连接 */

　　struct IPC_WAIT_CONNECTION{

　　int ch_status; /* wait conn. status.*/

　　void * ch_private; /* wait conn. private data. */

　　IPC_WaitOps *ops; /* wait conn. function table .*/

　　};

　　/* 活动的通信管道结构 */

　　struct IPC_CHANNEL{

　　int ch_status; /* 通道状态 */

　　pid_t farside_pid; /* 远端 pid */

　　void* ch_private; /* channel private data. (may contain conn. info.) */

　　IPC_Ops* ops; /* 通道函数集 */

　　unsigned int msgpad; /* 信息前缀字节数 */

　　unsigned int bytes_remaining; /* 剩余未发送的字节数 */

　　gboolean should_send_block; /* */

　　/* private: */

　　IPC_Queue* send_queue; /* 发送缓冲 */

　　IPC_Queue* recv_queue; /* 接收缓冲 */

　　/* 接收缓冲池, 经处理后转化为接收信息队列recv_queue */

　　struct ipc_bufpool* pool; /* buffer pool */

　　/* 发送的流量控制 */

　　int high_flow_mark;

　　int low_flow_mark;

　　void* high_flow_userdata;

　　void* low_flow_userdata;

　　flow_callback_t high_flow_callback;

　　flow_callback_t low_flow_callback;

　　int conntype;

　　char failreason[MAXFAILREASON];

　　};

　　IPC抽象层通信

　　server端:

　　1. 调用ipc_wait_conn_constructor()建立等待连接管道unix结构分析，成功则返回IPC_WaitConnection.

　　2. 通过poll/select来轮询客户请求。使用accept_connection接受连接，返回IPC_Channel。

　　client端：

　　调用ipc_channel_constructor()连接server，返回IPC_Channel。

　　IPC抽象层的UNIX Domain Socket实现

　　static struct IPC_OPS socket_ops = {

　　destroy: socket_destroy_channel, /* 删除通信管道 */

　　initiate_connection: socket_initiate_connection, /* 从client端建立连接 */

　　verify_auth: socket_verify_auth, /* 客户端认证信息 */

　　assert_auth: socket_assert_auth, /* 断言认证, (未用)*/

　　send: socket_send, /* 向管道发送信息 */

　　recv: socket_recv, /* 从管道接收信息*/

　　waitin: socket_waitin, /* 等待输入信息, (然后读取) */

　　waitout: socket_waitout, /* 等待信息输出结束 */

　　is_message_pending: socket_is_message_pending, /* 有信息可读或挂断 */

　　is_sending_blocked: socket_is_output_pending, /* 输出是否阻塞 */

　　resume_io: socket_resume_io, /* 恢复所有可能的ipc操作 */

　　get_send_select_fd: socket_get_send_fd, /* 取得发送fd */

　　get_recv_select_fd: socket_get_recv_fd, /* 取得接收fd */

　　set_send_qlen: socket_set_send_qlen, /* 设置最大发送缓冲长度 */

　　set_recv_qlen: socket_set_recv_qlen, /* 设置最大接收缓冲长度*/

　　set_high_flow_callback: socket_set_high_flow_callback, /* 高流量callback函数 */

　　set_low_flow_callback: socket_set_low_flow_callback, /* 低流量callback函数 */

　　new_ipcmsg: socket_new_ipcmsg, /* 返回一个新建立的IPC信息 */

　　get_chan_status: socket_get_chan_status, /* 返回管道状态 */

　　is_sendq_full: socket_is_sendq_full, /* 发送缓冲是否已满 */

　　is_recvq_full: socket_is_recvq_full, /* 接收缓冲是否已满 */

　　get_conntype: socket_get_conntype, /* 返回管道类型 */

　　/* 可以是IPC_SERVER , IPC_CLIENT , IPC_PEER */

　　};

　　节点间通信Plugin / 节点内通信交叉点

　　主要实现代码在heartbeart.c中

　　Heartbeat通信媒介结构

　　struct hb_media {

　　void * pd; /* 自定义数据结构 */

　　const char * name; /* 媒介名 */

　　char* type; /* 媒介类型 */

　　char* description; /* 媒介描述 */

　　const struct hb_media_fns*vf; /* hbcomm媒介处理函数集 */

　　IPC_Channel* wchan[2]; /* Unix域写子进程通信管道 */

　　IPC_Channel* rchan[2]; /* Unix域读子进程通信管道 */

　　};

　　/* heartbeat发送信息集群 */

　　/* 1.发送消息到write_child子进程 */

　　send_cluster_msg{ /* 发送信息到集群 */

　　…

　　process_outbound_packet{ /* 带包重传控制 */

　　send_to_all_media{ /* 发送到所有媒介 */

　　for (j=0; j < nummedia; ++j) {

　　IPC_Channel* wch = sysmedia[j]->wchan[P_WRITEFD];

　　…

　　/* 发送到特定传送媒介的写子进程 */

　　wrc=wch->ops->send(wch, outmsg);

　　}

　　}

　　}

　　}

　　/* 2. write_child写子进程发送消息到集群 */

　　write_child(){

　　IPC_Channel* ourchan = mp->wchan[P_READFD];

　　for(;;){

　　/* write_child通过Unix Domain Socket 接收heartbeat信息 */

　　IPC_Message* ipcmsg = ipcmsgfromIPC(ourchan); /* 调用ops->recv() */

　　…

　　/* 发送到集群其他节点 */

　　if (mp->vf->write(mp, ipcmsg->msg_body, ipcmsg->msg_len) != HA_OK) {

　　……

　　}

　　}

　　}

　　/* 从集群接收信息 */

　　/* 1. read_child读子进程从集群接收消息 */

　　Read_child(){

　　IPC_Channel* ourchan = mp->rchan[P_READFD];

　　For(;;){

　　/* 从hbcomm PLUGIN接收 */

　　if ((pkt=mp->vf->read(mp, &pktlen)) == NULL) {

　　……

　　}

　　if (NULL != imsg){

　　/* read_child子进程通过UNIX Domain Socket, 发送到heartbeat */

　　rc = ourchan->ops->send(ourchan, imsg);

　　rc2 = ourchan->ops->waitout(ourchan);

　　…

　　}

　　}

　　}

　　/* 2. heartbeat从read_child子进程接收信息并进行处理 */

　　s = G_main_add_IPC_Channel(PRI_READPKT

　　, sysmedia[j]->rchan[P_WRITEFD], FALSE

　　, read_child_dispatch, sysmedia+j, NULL);

　　read_child_dispatch（）{

　　…

　　msg = msgfromIPC(source, MSG_NEEDAUTH); /*调用ops->recv()从read_child读 */

　　process_clustermsg(msg, lnk); /* 对读到信息进行处理 */

　　}

　　heartbeat API Server端

　　struct api_query_handler query_handler_list[] = {

　　{API_SIGNOFF, api_signoff}, /* client登陆 */

　　{API_SETFILTER, api_setfilter}, /* 设置消息过滤 */

　　{API_SETSIGNAL, api_setsignal}, /* 设置消息到达信号通知 */

　　{API_NODELIST, api_nodelist}, /* 获取节点列表 */

　　{API_NODESTATUS, api_nodestatus}, /* 查询节点状态 */

　　{API_NODETYPE, api_nodetype}, /* 查询节点类型 */

　　{API_IFSTATUS, api_ifstatus}, /* 查询心跳状态 */

　　{API_IFLIST, api_iflist}, /* 查询心跳列表 */

　　{API_CLIENTSTATUS, api_clientstatus}, /* 查询client模块状态 */

　　{API_NUMNODES, api_num_nodes}, /* 返回集群普通节点数 */

　　{API_GETPARM, api_get_parameter}, /* 返回特定参数值 */

　　{API_GETRESOURCES, api_get_resources}, /* 返回资源状态(兼容1.2.x以前版本) */

　　{API_GETUUID, api_get_uuid}, /* 取得节点uuid值 */

　　{API_GETNAME, api_get_nodename}, /* 取得节点名 */

　　{API_SET_SENDQLEN, api_set_sendqlen} /* 设置发送队列长度 */

　　};

　　heartbeat API client端

　　static struct llc_ops heartbeat_ops = {

　　signon: hb_api_signon, /* 注册新的heartbeat client */

　　signoff: hb_api_signoff, /* 注销一个heartbeat client */

　　delete: hb_api_delete, /* 注销结构 */

　　set_msg_callback: set_msg_callback, /* 设置某信息类型callback */

　　set_nstatus_callback: set_nstatus_callback, /* 设置节点状态类型callback */

　　set_ifstatus_callback: set_ifstatus_callback, /* 设置心跳状态类型callback */

　　set_cstatus_callback: set_cstatus_callback, /* 设置client状态类型callback */

　　init_nodewalk: init_nodewalk, /* 初始化节点遍历 */

　　nextnode: nextnode, /* 下一个节点 */

　　end_nodewalk: end_nodewalk, /* 结束节点遍历 */

　　node_status: get_nodestatus, /* 节点当前状态 */

　　node_type: get_nodetype, /* 节点类型 */

　　init_ifwalk: init_ifwalk, /* 初始化心跳遍历 */

　　nextif: nextif, /* 下一个心跳接口 */

　　end_ifwalk: end_ifwalk, /* 结束心跳遍历 */

　　if_status: get_ifstatus, /* 心跳当前状态 */

　　client_status: get_clientstatus, /* client当前状态 */

　　get_uuid_by_name: get_uuid_by_name, /* 根据名字取得uuid */

　　get_name_by_uuid: get_name_by_uuid, /* 根据uuid取得名字 */

　　sendclustermsg: sendclustermsg, /* 发送消息到cluster中所有成员*/

　　sendnodemsg: sendnodemsg, /* 发送消息到特定节点 */

　　sendnodemsg_byuuid: sendnodemsg_byuuid, /* 发送消息到特定节点(by uuid)*/

　　send_ordered_clustermsg:send_ordered_clustermsg, /* 发送顺序集群信息 */

　　send_ordered_nodemsg: send_ordered_nodemsg, /* 发送顺序节点信息 */

　　inputfd: get_inputfd, /* 返回和检测信息到达*/

　　ipcchan: get_ipcchan, /* 返回IPC_Channel 类型ipc通道 */

　　msgready: msgready, /* 当有信息可读时返回true*/

　　setmsgsignal: hb_api_setsignal, /* setmsgsignal */

　　rcvmsg: rcvmsg, /* 接收msg, 交给callback处理 */

　　readmsg: read_msg_w_callbacks, /* 返回没有注册callback的msg */

　　setfmode: setfmode, /* setfmode */

　　get_parameter: get_parameter,

　　get_deadtime: get_deadtime,

　　get_keepalive: get_keepalive,

　　get_mynodeid: get_mynodeid, /* 取得本地节点名 */

　　get_logfacility: get_logfacility, /* suggested logging facility */

　　get_resources: get_resources, /* 取得资源当前分布状态 */

　　chan_is_connected: chan_is_connected,

　　set_sendq_len: set_sendq_len, /* 设置发送缓存区长度 */

　　set_send_block_mode: socket_set_send_block_mode,

　　errmsg: APIError,

　　};

　　注：

　　Client端API函数集明显比Server端查询处理函数集要多，是因为有些功能不需要通过Server端查询来得到

（编辑：我爱制作网_池州站长网）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!