开源中文网

您的位置: 首页 > 编程开发 > C语言 > 正文

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服务器端部分。


 

  图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() */