本文档的Copyleft归yfydz所有，使用GPL发布，可以自由拷贝，转载，转载时请保持文档的完整性，严禁用于任何商业用途。
msn: yfydz_no1@hotmail.com
来源：http://yfydz.cublog.cn
10. IPVS的同步
IPVS支持对连接的同步，两台IPVS设备可分别以MASTER或BACKUP运行，MASTER进程可将连接信息备份到BACKUP设备上,这样主设备死机时从设备可以无缝切换。
可以在IPVS设备上同时启动MASTER和BACKUP进程，使设备之间互为备份，实现IPVS设备的均衡。
IPVS同步实现在net/ipv4/ipvs/ip_vs_sync.c中
10.0 数据结构
同步信息块的格式如下,开始是4字节的信息头,后面是多个IPVS连接同步信息,每个块大小不固定,连接同步信息个数从0到多个:
0 1 2 3
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Count Conns | SyncID | Size |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| |
| IPVS Sync Connection (1) |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| . |
| . |
| . |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| |
| IPVS Sync Connection (n) |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Count Conns | SyncID | Size |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| |
| IPVS Sync Connection (1) |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| . |
| . |
| . |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| |
| IPVS Sync Connection (n) |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
信息头结构:
#define SYNC_MESG_HEADER_LEN 4
struct ip_vs_sync_mesg {
// 连接数
__u8 nr_conns;
// 同步ID
__u8 syncid;
// 数据总长
__u16 size;
struct ip_vs_sync_mesg {
// 连接数
__u8 nr_conns;
// 同步ID
__u8 syncid;
// 数据总长
__u16 size;
/* ip_vs_sync_conn entries start here */
};
};
IPVS连接同步信息结构:
struct ip_vs_sync_conn {
__u8 reserved;
// 连接基本信息
/* Protocol, addresses and port numbers */
__u8 protocol; /* Which protocol (TCP/UDP) */
__u16 cport;
__u16 vport;
__u16 dport;
__u32 caddr; /* client address */
__u32 vaddr; /* virtual address */
__u32 daddr; /* destination address */
/* Protocol, addresses and port numbers */
__u8 protocol; /* Which protocol (TCP/UDP) */
__u16 cport;
__u16 vport;
__u16 dport;
__u32 caddr; /* client address */
__u32 vaddr; /* virtual address */
__u32 daddr; /* destination address */
// 连接的状态和标志
/* Flags and state transition */
__u16 flags; /* status flags */
__u16 state; /* state info */
/* Flags and state transition */
__u16 flags; /* status flags */
__u16 state; /* state info */
// 后续可能有连接选项参数,就是TCP的序列号和确认号信息
/* The sequence options start here */
};
/* The sequence options start here */
};
IPVS连接同步选项结构,,就是进入和发出发现TCP的序列号信息
struct ip_vs_sync_conn_options {
struct ip_vs_seq in_seq; /* incoming seq. struct */
struct ip_vs_seq out_seq; /* outgoing seq. struct */
};
struct ip_vs_sync_conn_options {
struct ip_vs_seq in_seq; /* incoming seq. struct */
struct ip_vs_seq out_seq; /* outgoing seq. struct */
};
连接数据控制块结构
struct ip_vs_sync_buff {
// 形成队列
struct list_head list;
unsigned long firstuse;
/* pointers for the message data */
// 实际的同步信息
struct ip_vs_sync_mesg *mesg;
// 数据空闲区头指针
unsigned char *head;
// 数据尾指针
unsigned char *end;
};
// 实际的同步信息
struct ip_vs_sync_mesg *mesg;
// 数据空闲区头指针
unsigned char *head;
// 数据尾指针
unsigned char *end;
};
10.1 进程启动
IPVS同步进程是一个内核进程，是由IPVSADM通过命令启动的.
int start_sync_thread(int state, char *mcast_ifn, __u8 syncid)
{
DECLARE_COMPLETION(startup);
pid_t pid;
// 检查进程是否已经存在
if ((state == IP_VS_STATE_MASTER && sync_master_pid) ||
(state == IP_VS_STATE_BACKUP && sync_backup_pid))
return -EEXIST;
if ((state == IP_VS_STATE_MASTER && sync_master_pid) ||
(state == IP_VS_STATE_BACKUP && sync_backup_pid))
return -EEXIST;
IP_VS_DBG(7, "%s: pid %d\n", __FUNCTION__, current->pid);
IP_VS_DBG(7, "Each ip_vs_sync_conn entry need %Zd bytes\n",
sizeof(struct ip_vs_sync_conn));
IP_VS_DBG(7, "Each ip_vs_sync_conn entry need %Zd bytes\n",
sizeof(struct ip_vs_sync_conn));
ip_vs_sync_state |= state;
if (state == IP_VS_STATE_MASTER) {
// MASTER进程
// 通信的网卡
strlcpy(ip_vs_master_mcast_ifn, mcast_ifn, sizeof(ip_vs_master_mcast_ifn));
// 同步ID
ip_vs_master_syncid = syncid;
} else {
// SLAVE进程
// 通信的网卡
strlcpy(ip_vs_backup_mcast_ifn, mcast_ifn, sizeof(ip_vs_backup_mcast_ifn));
// 同步ID
ip_vs_backup_syncid = syncid;
}
if (state == IP_VS_STATE_MASTER) {
// MASTER进程
// 通信的网卡
strlcpy(ip_vs_master_mcast_ifn, mcast_ifn, sizeof(ip_vs_master_mcast_ifn));
// 同步ID
ip_vs_master_syncid = syncid;
} else {
// SLAVE进程
// 通信的网卡
strlcpy(ip_vs_backup_mcast_ifn, mcast_ifn, sizeof(ip_vs_backup_mcast_ifn));
// 同步ID
ip_vs_backup_syncid = syncid;
}
repeat:
// 启动内核进程
if ((pid = kernel_thread(fork_sync_thread, &startup, 0)) < 0) {
IP_VS_ERR("could not create fork_sync_thread due to %d... "
"retrying.\n", pid);
ssleep(1);
goto repeat;
}
// 启动内核进程
if ((pid = kernel_thread(fork_sync_thread, &startup, 0)) < 0) {
IP_VS_ERR("could not create fork_sync_thread due to %d... "
"retrying.\n", pid);
ssleep(1);
goto repeat;
}
wait_for_completion(&startup);
return 0;
}
}
fork_sync_thread()函数也继续fork出一个进程,形成守护进程
static int fork_sync_thread(void *startup)
{
pid_t pid;
{
pid_t pid;
/* fork the sync thread here, then the parent process of the
sync thread is the init process after this thread exits. */
repeat:
if ((pid = kernel_thread(sync_thread, startup, 0)) < 0) {
IP_VS_ERR("could not create sync_thread due to %d... "
"retrying.\n", pid);
ssleep(1);
goto repeat;
}
sync thread is the init process after this thread exits. */
repeat:
if ((pid = kernel_thread(sync_thread, startup, 0)) < 0) {
IP_VS_ERR("could not create sync_thread due to %d... "
"retrying.\n", pid);
ssleep(1);
goto repeat;
}
return 0;
}
}
static int sync_thread(void *startup)
{
DECLARE_WAITQUEUE(wait, current);
mm_segment_t oldmm;
int state;
const char *name;
/* increase the module use count */
// 增加IPVS模块引用计数
ip_vs_use_count_inc();
// 增加IPVS模块引用计数
ip_vs_use_count_inc();
// 设置进程状态的名称
if (ip_vs_sync_state & IP_VS_STATE_MASTER && !sync_master_pid) {
state = IP_VS_STATE_MASTER;
name = "ipvs_syncmaster";
} else if (ip_vs_sync_state & IP_VS_STATE_BACKUP && !sync_backup_pid) {
state = IP_VS_STATE_BACKUP;
name = "ipvs_syncbackup";
} else {
IP_VS_BUG();
ip_vs_use_count_dec();
return -EINVAL;
}
// daemon化进程
daemonize(name);
if (ip_vs_sync_state & IP_VS_STATE_MASTER && !sync_master_pid) {
state = IP_VS_STATE_MASTER;
name = "ipvs_syncmaster";
} else if (ip_vs_sync_state & IP_VS_STATE_BACKUP && !sync_backup_pid) {
state = IP_VS_STATE_BACKUP;
name = "ipvs_syncbackup";
} else {
IP_VS_BUG();
ip_vs_use_count_dec();
return -EINVAL;
}
// daemon化进程
daemonize(name);
oldmm = get_fs();
set_fs(KERNEL_DS);
set_fs(KERNEL_DS);
/* Block all signals */
// 本进程不接收所有信号
spin_lock_irq(¤t->sighand->siglock);
siginitsetinv(¤t->blocked, 0);
recalc_sigpending();
spin_unlock_irq(¤t->sighand->siglock);
// 本进程不接收所有信号
spin_lock_irq(¤t->sighand->siglock);
siginitsetinv(¤t->blocked, 0);
recalc_sigpending();
spin_unlock_irq(¤t->sighand->siglock);
/* set the maximum length of sync message */
// 设置最大同步信息长度
set_sync_mesg_maxlen(state);
// 设置最大同步信息长度
set_sync_mesg_maxlen(state);
/* set up multicast address */
// 设置UDP多播sock的参数
mcast_addr.sin_family = AF_INET;
// 端口为8848
mcast_addr.sin_port = htons(IP_VS_SYNC_PORT);
// 多播地址为224.0.0.81
mcast_addr.sin_addr.s_addr = htonl(IP_VS_SYNC_GROUP);
// 设置UDP多播sock的参数
mcast_addr.sin_family = AF_INET;
// 端口为8848
mcast_addr.sin_port = htons(IP_VS_SYNC_PORT);
// 多播地址为224.0.0.81
mcast_addr.sin_addr.s_addr = htonl(IP_VS_SYNC_GROUP);
// 增加等待队列,由于接收发送数据必须在top half中,在bottom half中
// 只是发个WQ信号告诉top half可以进行接收或发送数据了
add_wait_queue(&sync_wait, &wait);
// 只是发个WQ信号告诉top half可以进行接收或发送数据了
add_wait_queue(&sync_wait, &wait);
// 设置当前进程的pid
set_sync_pid(state, current->pid);
complete((struct completion *)startup);
set_sync_pid(state, current->pid);
complete((struct completion *)startup);
/* processing master/backup loop here */
// 分别进行MASTER或SLAVE循环
if (state == IP_VS_STATE_MASTER)
sync_master_loop();
else if (state == IP_VS_STATE_BACKUP)
sync_backup_loop();
else IP_VS_BUG();
// 分别进行MASTER或SLAVE循环
if (state == IP_VS_STATE_MASTER)
sync_master_loop();
else if (state == IP_VS_STATE_BACKUP)
sync_backup_loop();
else IP_VS_BUG();
// 循环退出,删除等待队列
remove_wait_queue(&sync_wait, &wait);
remove_wait_queue(&sync_wait, &wait);
/* thread exits */
set_sync_pid(state, 0);
IP_VS_INFO("sync thread stopped!\n");
set_sync_pid(state, 0);
IP_VS_INFO("sync thread stopped!\n");
set_fs(oldmm);
/* decrease the module use count */
// 减少IPVS模块引用
ip_vs_use_count_dec();
// 减少IPVS模块引用
ip_vs_use_count_dec();
// 设置同步状态标志为0
set_stop_sync(state, 0);
wake_up(&stop_sync_wait);
set_stop_sync(state, 0);
wake_up(&stop_sync_wait);
return 0;
}
}
10.2 进程停止
同步进程也是由ipvsadm命令发出的.
int stop_sync_thread(int state)
{
DECLARE_WAITQUEUE(wait, current);
// 检查进程是否在运行
if ((state == IP_VS_STATE_MASTER && !sync_master_pid) ||
(state == IP_VS_STATE_BACKUP && !sync_backup_pid))
return -ESRCH;
if ((state == IP_VS_STATE_MASTER && !sync_master_pid) ||
(state == IP_VS_STATE_BACKUP && !sync_backup_pid))
return -ESRCH;
IP_VS_DBG(7, "%s: pid %d\n", __FUNCTION__, current->pid);
IP_VS_INFO("stopping sync thread %d ...\n",
(state == IP_VS_STATE_MASTER) ? sync_master_pid : sync_backup_pid);
IP_VS_INFO("stopping sync thread %d ...\n",
(state == IP_VS_STATE_MASTER) ? sync_master_pid : sync_backup_pid);
__set_current_state(TASK_UNINTERRUPTIBLE);
// 增加停止等待队列
add_wait_queue(&stop_sync_wait, &wait);
// 设置同步停止标志为1,让所有同步进程停止
set_stop_sync(state, 1);
ip_vs_sync_state -= state;
wake_up(&sync_wait);
// 重新调度,等待同步进程结束
schedule();
// 重新调度会来执行本进程, 同步进程应该已经结束
__set_current_state(TASK_RUNNING);
// 删除等待队列
remove_wait_queue(&stop_sync_wait, &wait);
// 增加停止等待队列
add_wait_queue(&stop_sync_wait, &wait);
// 设置同步停止标志为1,让所有同步进程停止
set_stop_sync(state, 1);
ip_vs_sync_state -= state;
wake_up(&sync_wait);
// 重新调度,等待同步进程结束
schedule();
// 重新调度会来执行本进程, 同步进程应该已经结束
__set_current_state(TASK_RUNNING);
// 删除等待队列
remove_wait_queue(&stop_sync_wait, &wait);
/* Note: no need to reap the sync thread, because its parent
process is the init process */
// 检查一下进程状态和标志是否支持
if ((state == IP_VS_STATE_MASTER && stop_master_sync) ||
(state == IP_VS_STATE_BACKUP && stop_backup_sync))
IP_VS_BUG();
process is the init process */
// 检查一下进程状态和标志是否支持
if ((state == IP_VS_STATE_MASTER && stop_master_sync) ||
(state == IP_VS_STATE_BACKUP && stop_backup_sync))
IP_VS_BUG();
return 0;
}
}
10.3 MASTER循环
static void sync_master_loop(void)
{
struct socket *sock;
struct ip_vs_sync_buff *sb;
// 建立多播SOCK,同步信息由此SOCK发出
/* create the sending multicast socket */
sock = make_send_sock();
if (!sock)
return;
/* create the sending multicast socket */
sock = make_send_sock();
if (!sock)
return;
IP_VS_INFO("sync thread started: state = MASTER, mcast_ifn = %s, "
"syncid = %d\n",
ip_vs_master_mcast_ifn, ip_vs_master_syncid);
// 进入死循环
for (;;) {
// 从队列中取发送数据块
while ((sb=sb_dequeue())) {
// 发出同步数据块
ip_vs_send_sync_msg(sock, sb->mesg);
// 释放数据块缓冲
ip_vs_sync_buff_release(sb);
}
"syncid = %d\n",
ip_vs_master_mcast_ifn, ip_vs_master_syncid);
// 进入死循环
for (;;) {
// 从队列中取发送数据块
while ((sb=sb_dequeue())) {
// 发出同步数据块
ip_vs_send_sync_msg(sock, sb->mesg);
// 释放数据块缓冲
ip_vs_sync_buff_release(sb);
}
/* check if entries stay in curr_sb for 2 seconds */
// 如果2秒内数据块没准备好,直接将未完成的数据块发出去
// 最差情况下数据块里没有IPVS连接信息,只有一个数据头,
// 相当于同步信号,表明MASTER还没死
if ((sb = get_curr_sync_buff(2*HZ))) {
ip_vs_send_sync_msg(sock, sb->mesg);
ip_vs_sync_buff_release(sb);
}
// 如果2秒内数据块没准备好,直接将未完成的数据块发出去
// 最差情况下数据块里没有IPVS连接信息,只有一个数据头,
// 相当于同步信号,表明MASTER还没死
if ((sb = get_curr_sync_buff(2*HZ))) {
ip_vs_send_sync_msg(sock, sb->mesg);
ip_vs_sync_buff_release(sb);
}
// 发现停止MASTER进程标志,中断循环
if (stop_master_sync)
break;
// 休眠1秒
ssleep(1);
}
if (stop_master_sync)
break;
// 休眠1秒
ssleep(1);
}
// 循环退出,将当前发送队列中的数据块都发送完
/* clean up the sync_buff queue */
while ((sb=sb_dequeue())) {
ip_vs_sync_buff_release(sb);
}
/* clean up the sync_buff queue */
while ((sb=sb_dequeue())) {
ip_vs_sync_buff_release(sb);
}
/* clean up the current sync_buff */
// 立即返回当前块,当前块是构造中还没放到发送队列中的数据块
if ((sb = get_curr_sync_buff(0))) {
ip_vs_sync_buff_release(sb);
}
// 立即返回当前块,当前块是构造中还没放到发送队列中的数据块
if ((sb = get_curr_sync_buff(0))) {
ip_vs_sync_buff_release(sb);
}
// 释放UDP多播SOCK
/* release the sending multicast socket */
sock_release(sock);
}
/* release the sending multicast socket */
sock_release(sock);
}
同步数据块出队列
static inline struct ip_vs_sync_buff * sb_dequeue(void)
{
struct ip_vs_sync_buff *sb;
spin_lock_bh(&ip_vs_sync_lock);
// 检查队列是否为空
if (list_empty(&ip_vs_sync_queue)) {
sb = NULL;
} else {
// 获取队列第一个节点中的数据
sb = list_entry(ip_vs_sync_queue.next,
struct ip_vs_sync_buff,
list);
// 将节点从链表中拆除
list_del(&sb->list);
}
spin_unlock_bh(&ip_vs_sync_lock);
// 检查队列是否为空
if (list_empty(&ip_vs_sync_queue)) {
sb = NULL;
} else {
// 获取队列第一个节点中的数据
sb = list_entry(ip_vs_sync_queue.next,
struct ip_vs_sync_buff,
list);
// 将节点从链表中拆除
list_del(&sb->list);
}
spin_unlock_bh(&ip_vs_sync_lock);
return sb;
}
}
发现同步信息块
static void
ip_vs_send_sync_msg(struct socket *sock, struct ip_vs_sync_mesg *msg)
{
int msize;
msize = msg->size;
/* Put size in network byte order */
// 网络传输的长度必须是网络序
msg->size = htons(msg->size);
// 将数据块作为普通缓冲数据发送
if (ip_vs_send_async(sock, (char *)msg, msize) != msize)
IP_VS_ERR("ip_vs_send_async error\n");
}
// 网络传输的长度必须是网络序
msg->size = htons(msg->size);
// 将数据块作为普通缓冲数据发送
if (ip_vs_send_async(sock, (char *)msg, msize) != msize)
IP_VS_ERR("ip_vs_send_async error\n");
}
发送缓冲数据,就是调用kernel_sendmsg()函数发送
static int
ip_vs_send_async(struct socket *sock, const char *buffer, const size_t length)
{
struct msghdr msg = {.msg_flags = MSG_DONTWAIT|MSG_NOSIGNAL};
struct kvec iov;
int len;
static int
ip_vs_send_async(struct socket *sock, const char *buffer, const size_t length)
{
struct msghdr msg = {.msg_flags = MSG_DONTWAIT|MSG_NOSIGNAL};
struct kvec iov;
int len;
EnterFunction(7);
iov.iov_base = (void *)buffer;
iov.iov_len = length;
iov.iov_base = (void *)buffer;
iov.iov_len = length;
len = kernel_sendmsg(sock, &msg, &iov, 1, (size_t)(length));
LeaveFunction(7);
return len;
}
return len;
}
获取当前同步数据块
static inline struct ip_vs_sync_buff *
get_curr_sync_buff(unsigned long time)
{
struct ip_vs_sync_buff *sb;
static inline struct ip_vs_sync_buff *
get_curr_sync_buff(unsigned long time)
{
struct ip_vs_sync_buff *sb;
spin_lock_bh(&curr_sb_lock);
if (curr_sb && (time == 0 ||
// 当前同步数据块存在,而且立即返回(time == 0)或
// 数据块已经存在超过time个jiffies时返回
time_before(jiffies - curr_sb->firstuse, time))) {
sb = curr_sb;
curr_sb = NULL;
} else
sb = NULL;
spin_unlock_bh(&curr_sb_lock);
return sb;
}
if (curr_sb && (time == 0 ||
// 当前同步数据块存在,而且立即返回(time == 0)或
// 数据块已经存在超过time个jiffies时返回
time_before(jiffies - curr_sb->firstuse, time))) {
sb = curr_sb;
curr_sb = NULL;
} else
sb = NULL;
spin_unlock_bh(&curr_sb_lock);
return sb;
}
10.4 BACKUP循环
static void sync_backup_loop(void)
{
struct socket *sock;
char *buf;
int len;
{
struct socket *sock;
char *buf;
int len;
// 分配数据接收空间
if (!(buf = kmalloc(sync_recv_mesg_maxlen, GFP_ATOMIC))) {
IP_VS_ERR("sync_backup_loop: kmalloc error\n");
return;
}
// 创建UDP接收SOCK
/* create the receiving multicast socket */
sock = make_receive_sock();
if (!sock)
goto out;
if (!(buf = kmalloc(sync_recv_mesg_maxlen, GFP_ATOMIC))) {
IP_VS_ERR("sync_backup_loop: kmalloc error\n");
return;
}
// 创建UDP接收SOCK
/* create the receiving multicast socket */
sock = make_receive_sock();
if (!sock)
goto out;
IP_VS_INFO("sync thread started: state = BACKUP, mcast_ifn = %s, "
"syncid = %d\n",
ip_vs_backup_mcast_ifn, ip_vs_backup_syncid);
// 进入接收循环
for (;;) {
/* do you have data now? */
// 接收队列非空
while (!skb_queue_empty(&(sock->sk->sk_receive_queue))) {
// 接收数据到缓冲区
if ((len =
ip_vs_receive(sock, buf,
sync_recv_mesg_maxlen)) <= 0) {
IP_VS_ERR("receiving message error\n");
break;
}
/* disable bottom half, because it accessed the data
shared by softirq while getting/creating conns */
// 处理数据时不能再进入bottom half
local_bh_disable();
// 处理接收数据
ip_vs_process_message(buf, len);
local_bh_enable();
}
"syncid = %d\n",
ip_vs_backup_mcast_ifn, ip_vs_backup_syncid);
// 进入接收循环
for (;;) {
/* do you have data now? */
// 接收队列非空
while (!skb_queue_empty(&(sock->sk->sk_receive_queue))) {
// 接收数据到缓冲区
if ((len =
ip_vs_receive(sock, buf,
sync_recv_mesg_maxlen)) <= 0) {
IP_VS_ERR("receiving message error\n");
break;
}
/* disable bottom half, because it accessed the data
shared by softirq while getting/creating conns */
// 处理数据时不能再进入bottom half
local_bh_disable();
// 处理接收数据
ip_vs_process_message(buf, len);
local_bh_enable();
}
// 检查是否设置进程停止标志
if (stop_backup_sync)
break;
// 睡眠1秒
ssleep(1);
}
// 释放UDP SOCK
/* release the sending multicast socket */
sock_release(sock);
if (stop_backup_sync)
break;
// 睡眠1秒
ssleep(1);
}
// 释放UDP SOCK
/* release the sending multicast socket */
sock_release(sock);
out:
// 释放接收缓冲
kfree(buf);
}
// 释放接收缓冲
kfree(buf);
}
接收数据函数,比较简单,直接调用内核的kernel_recvmsg函数
static int
ip_vs_receive(struct socket *sock, char *buffer, const size_t buflen)
{
struct msghdr msg = {NULL,};
struct kvec iov;
int len;
static int
ip_vs_receive(struct socket *sock, char *buffer, const size_t buflen)
{
struct msghdr msg = {NULL,};
struct kvec iov;
int len;
EnterFunction(7);
/* Receive a packet */
iov.iov_base = buffer;
iov.iov_len = (size_t)buflen;
iov.iov_base = buffer;
iov.iov_len = (size_t)buflen;
len = kernel_recvmsg(sock, &msg, &iov, 1, buflen, 0);
if (len < 0)
return -1;
return -1;
LeaveFunction(7);
return len;
}
return len;
}
处理接收数据函数
/*
* Process received multicast message and create the corresponding
* ip_vs_conn entries.
*/
static void ip_vs_process_message(const char *buffer, const size_t buflen)
{
struct ip_vs_sync_mesg *m = (struct ip_vs_sync_mesg *)buffer;
struct ip_vs_sync_conn *s;
struct ip_vs_sync_conn_options *opt;
struct ip_vs_conn *cp;
char *p;
int i;
/* Convert size back to host byte order */
m->size = ntohs(m->size);
// 检查接收的数据长度是否正确
if (buflen != m->size) {
IP_VS_ERR("bogus message\n");
return;
}
m->size = ntohs(m->size);
// 检查接收的数据长度是否正确
if (buflen != m->size) {
IP_VS_ERR("bogus message\n");
return;
}
// 检查同步ID是否匹配
/* SyncID sanity check */
if (ip_vs_backup_syncid != 0 && m->syncid != ip_vs_backup_syncid) {
IP_VS_DBG(7, "Ignoring incoming msg with syncid = %d\n",
m->syncid);
return;
}
/* SyncID sanity check */
if (ip_vs_backup_syncid != 0 && m->syncid != ip_vs_backup_syncid) {
IP_VS_DBG(7, "Ignoring incoming msg with syncid = %d\n",
m->syncid);
return;
}
// 同步信息块头后面是真正的IPVS连接同步信息
// p现在是第一个同步连接结构指针
p = (char *)buffer + sizeof(struct ip_vs_sync_mesg);
for (i=0; i<m->nr_conns; i++) {
// 循环读取缓冲区中的同步连接信息
unsigned flags;
// p现在是第一个同步连接结构指针
p = (char *)buffer + sizeof(struct ip_vs_sync_mesg);
for (i=0; i<m->nr_conns; i++) {
// 循环读取缓冲区中的同步连接信息
unsigned flags;
s = (struct ip_vs_sync_conn *)p;
flags = ntohs(s->flags);
// 根据同步连接信息查找连接
if (!(flags & IP_VS_CONN_F_TEMPLATE))
cp = ip_vs_conn_in_get(s->protocol,
s->caddr, s->cport,
s->vaddr, s->vport);
else
cp = ip_vs_ct_in_get(s->protocol,
s->caddr, s->cport,
s->vaddr, s->vport);
if (!cp) {
// 找不到连接,说明是MASTER新建的连接同步过来了
// 新建连接,连接的dest参数为NULL,表明是同步产生的连接,而不是BACKUP自己生成的连接
cp = ip_vs_conn_new(s->protocol,
s->caddr, s->cport,
s->vaddr, s->vport,
s->daddr, s->dport,
flags, NULL);
if (!cp) {
IP_VS_ERR("ip_vs_conn_new failed\n");
return;
}
// 设置连接状态
cp->state = ntohs(s->state);
} else if (!cp->dest) {
// 找到了连接但没有dest指针,说明该连接是同步产生的连接,而不是BACKUP主动产生的连接
/* it is an entry created by the synchronization */
cp->state = ntohs(s->state);
cp->flags = flags | IP_VS_CONN_F_HASHED;
} /* Note that we don't touch its state and flags
if it is a normal entry. */
flags = ntohs(s->flags);
// 根据同步连接信息查找连接
if (!(flags & IP_VS_CONN_F_TEMPLATE))
cp = ip_vs_conn_in_get(s->protocol,
s->caddr, s->cport,
s->vaddr, s->vport);
else
cp = ip_vs_ct_in_get(s->protocol,
s->caddr, s->cport,
s->vaddr, s->vport);
if (!cp) {
// 找不到连接,说明是MASTER新建的连接同步过来了
// 新建连接,连接的dest参数为NULL,表明是同步产生的连接,而不是BACKUP自己生成的连接
cp = ip_vs_conn_new(s->protocol,
s->caddr, s->cport,
s->vaddr, s->vport,
s->daddr, s->dport,
flags, NULL);
if (!cp) {
IP_VS_ERR("ip_vs_conn_new failed\n");
return;
}
// 设置连接状态
cp->state = ntohs(s->state);
} else if (!cp->dest) {
// 找到了连接但没有dest指针,说明该连接是同步产生的连接,而不是BACKUP主动产生的连接
/* it is an entry created by the synchronization */
cp->state = ntohs(s->state);
cp->flags = flags | IP_VS_CONN_F_HASHED;
} /* Note that we don't touch its state and flags
if it is a normal entry. */
if (flags & IP_VS_CONN_F_SEQ_MASK) {
// 拷贝连接选项
opt = (struct ip_vs_sync_conn_options *)&s[1];
memcpy(&cp->in_seq, opt, sizeof(*opt));
// 缓冲数据后移连接加选项长度
p += FULL_CONN_SIZE;
} else
// 缓冲数据后移连接长度
p += SIMPLE_CONN_SIZE;
// 拷贝连接选项
opt = (struct ip_vs_sync_conn_options *)&s[1];
memcpy(&cp->in_seq, opt, sizeof(*opt));
// 缓冲数据后移连接加选项长度
p += FULL_CONN_SIZE;
} else
// 缓冲数据后移连接长度
p += SIMPLE_CONN_SIZE;
// 设置连接计数,是个固定值
atomic_set(&cp->in_pkts, sysctl_ip_vs_sync_threshold[0]);
// 超时
cp->timeout = IP_VS_SYNC_CONN_TIMEOUT;
// 减少连接引用计数
ip_vs_conn_put(cp);
atomic_set(&cp->in_pkts, sysctl_ip_vs_sync_threshold[0]);
// 超时
cp->timeout = IP_VS_SYNC_CONN_TIMEOUT;
// 减少连接引用计数
ip_vs_conn_put(cp);
// 检查当前缓冲区指针是否越界了
if (p > buffer+buflen) {
IP_VS_ERR("bogus message\n");
return;
}
}
}
if (p > buffer+buflen) {
IP_VS_ERR("bogus message\n");
return;
}
}
}
10.5 连接同步
连接同步函数ip_vs_sync_conn()是由ip_vs_in()函数调用的:
......
// MASTER状态
if ((ip_vs_sync_state & IP_VS_STATE_MASTER) &&
// 非TCP协议或是TCP的连接建立好状态才同步
(cp->protocol != IPPROTO_TCP ||
cp->state == IP_VS_TCP_S_ESTABLISHED) &&
// 积攒了一定包数后才同步,不是每个包都同步
(atomic_read(&cp->in_pkts) % sysctl_ip_vs_sync_threshold[1]
== sysctl_ip_vs_sync_threshold[0]))
ip_vs_sync_conn(cp);
......
/*
* Add an ip_vs_conn information into the current sync_buff.
* Called by ip_vs_in.
*/
void ip_vs_sync_conn(struct ip_vs_conn *cp)
{
struct ip_vs_sync_mesg *m;
struct ip_vs_sync_conn *s;
int len;
* Add an ip_vs_conn information into the current sync_buff.
* Called by ip_vs_in.
*/
void ip_vs_sync_conn(struct ip_vs_conn *cp)
{
struct ip_vs_sync_mesg *m;
struct ip_vs_sync_conn *s;
int len;
spin_lock(&curr_sb_lock);
if (!curr_sb) {
// 当前连接数据块为空,分配新块
if (!(curr_sb=ip_vs_sync_buff_create())) {
spin_unlock(&curr_sb_lock);
IP_VS_ERR("ip_vs_sync_buff_create failed.\n");
return;
}
}
if (!curr_sb) {
// 当前连接数据块为空,分配新块
if (!(curr_sb=ip_vs_sync_buff_create())) {
spin_unlock(&curr_sb_lock);
IP_VS_ERR("ip_vs_sync_buff_create failed.\n");
return;
}
}
// 检查是否包括选项长度
len = (cp->flags & IP_VS_CONN_F_SEQ_MASK) ? FULL_CONN_SIZE :
SIMPLE_CONN_SIZE;
len = (cp->flags & IP_VS_CONN_F_SEQ_MASK) ? FULL_CONN_SIZE :
SIMPLE_CONN_SIZE;
m = curr_sb->mesg;
// 空闲缓冲区头,作为一个连接同步单元头
s = (struct ip_vs_sync_conn *)curr_sb->head;
// 空闲缓冲区头,作为一个连接同步单元头
s = (struct ip_vs_sync_conn *)curr_sb->head;
// 基本同步信息
/* copy members */
s->protocol = cp->protocol;
s->cport = cp->cport;
s->vport = cp->vport;
s->dport = cp->dport;
s->caddr = cp->caddr;
s->vaddr = cp->vaddr;
s->daddr = cp->daddr;
// 连接标志和状态
s->flags = htons(cp->flags & ~IP_VS_CONN_F_HASHED);
s->state = htons(cp->state);
if (cp->flags & IP_VS_CONN_F_SEQ_MASK) {
// 增加选项信息,即TCP序列号
struct ip_vs_sync_conn_options *opt =
(struct ip_vs_sync_conn_options *)&s[1];
memcpy(opt, &cp->in_seq, sizeof(*opt));
}
// 同步块中连接数增加
m->nr_conns++;
// 有效数据长度增加
m->size += len;
// 空闲指针后移
curr_sb->head += len;
/* copy members */
s->protocol = cp->protocol;
s->cport = cp->cport;
s->vport = cp->vport;
s->dport = cp->dport;
s->caddr = cp->caddr;
s->vaddr = cp->vaddr;
s->daddr = cp->daddr;
// 连接标志和状态
s->flags = htons(cp->flags & ~IP_VS_CONN_F_HASHED);
s->state = htons(cp->state);
if (cp->flags & IP_VS_CONN_F_SEQ_MASK) {
// 增加选项信息,即TCP序列号
struct ip_vs_sync_conn_options *opt =
(struct ip_vs_sync_conn_options *)&s[1];
memcpy(opt, &cp->in_seq, sizeof(*opt));
}
// 同步块中连接数增加
m->nr_conns++;
// 有效数据长度增加
m->size += len;
// 空闲指针后移
curr_sb->head += len;
// 检查剩下的空间是否还能容纳一个同步连接结构
/* check if there is a space for next one */
if (curr_sb->head+FULL_CONN_SIZE > curr_sb->end) {
// 空间不够的话将当前同步数据块添加到发送链表中
// 饺子包好了
sb_queue_tail(curr_sb);
// 当前同步块指针置空
// 要重新给张饺皮了
curr_sb = NULL;
}
spin_unlock(&curr_sb_lock);
/* check if there is a space for next one */
if (curr_sb->head+FULL_CONN_SIZE > curr_sb->end) {
// 空间不够的话将当前同步数据块添加到发送链表中
// 饺子包好了
sb_queue_tail(curr_sb);
// 当前同步块指针置空
// 要重新给张饺皮了
curr_sb = NULL;
}
spin_unlock(&curr_sb_lock);
/* synchronize its controller if it has */
// 如果有主连接,递归调用本函数同步主连接信息
if (cp->control)
ip_vs_sync_conn(cp->control);
}
// 如果有主连接,递归调用本函数同步主连接信息
if (cp->control)
ip_vs_sync_conn(cp->control);
}
分配新连接同步数据块
static inline struct ip_vs_sync_buff * ip_vs_sync_buff_create(void)
{
struct ip_vs_sync_buff *sb;
// 分配控制信息块头
if (!(sb=kmalloc(sizeof(struct ip_vs_sync_buff), GFP_ATOMIC)))
return NULL;
// 分配具体的信息缓冲区
if (!(sb->mesg=kmalloc(sync_send_mesg_maxlen, GFP_ATOMIC))) {
kfree(sb);
return NULL;
}
// 当前同步连接数为0
sb->mesg->nr_conns = 0;
// 同步ID
sb->mesg->syncid = ip_vs_master_syncid;
// 目前有效数据包长就是同步信息头长
sb->mesg->size = 4;
// 空闲数据头指针
sb->head = (unsigned char *)sb->mesg + 4;
// 数据缓冲结尾
sb->end = (unsigned char *)sb->mesg + sync_send_mesg_maxlen;
// 同步数据块创建时间
sb->firstuse = jiffies;
return sb;
}
if (!(sb=kmalloc(sizeof(struct ip_vs_sync_buff), GFP_ATOMIC)))
return NULL;
// 分配具体的信息缓冲区
if (!(sb->mesg=kmalloc(sync_send_mesg_maxlen, GFP_ATOMIC))) {
kfree(sb);
return NULL;
}
// 当前同步连接数为0
sb->mesg->nr_conns = 0;
// 同步ID
sb->mesg->syncid = ip_vs_master_syncid;
// 目前有效数据包长就是同步信息头长
sb->mesg->size = 4;
// 空闲数据头指针
sb->head = (unsigned char *)sb->mesg + 4;
// 数据缓冲结尾
sb->end = (unsigned char *)sb->mesg + sync_send_mesg_maxlen;
// 同步数据块创建时间
sb->firstuse = jiffies;
return sb;
}