Heartbeat 通信层结构分析

深碍√TFBOYSˉ_ 2022-09-30 03:58 272阅读 0赞

Heartbeat 通信层结构分析  
前言  
　　Heartbeat是Linux-HA开源项目发布的用于关键应用环境的HA软件名称。从1999以来到现在， 历经1.2.x, 2.0.x等多个版本，在全球开源HA领域具有举足轻重的知名度, 应用日益广泛， 并且得到了一些主流Linux操作系统厂商的支持。  
而通信层实现无疑是集群软件运行的最基本底层支撑。本文对通过分析Heartbeat源码，对其通信层基本结构和机制进行了分析和阐述。给出了基本数据结构和实现流程。  
　　所有分析基于Heartbeat 2.0.4版本。  
相关源码： http://www.linux-ha.org/download/heartbeat-2.0.4.tar.gz  
  
  
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服务器端部分。  
http://www.gd-linux.org/liu\_attach/4.11forum.jpg  
　图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()建立等待连接管道，成功则返回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端查询来得到