基于Linux的集群系统（十一）--Linux集群系统的实现（下）

 

级别： 初级

许广斌 (), 工程师

2004 年 6 月 01 日

本篇是《基于linux的集群系统》实现篇的下部分。将向我们讲述各个模块的具体设计和实现。

6.1 设计思想

ip伪装模块的主要工作是：

1．接收内部网发向外部网的所有请求。

2．内部网中的连接请求通过平衡器转发到外部网。

3．将内部网发向外部网中的所有请求的源地址隐藏，使所有请求看上去都是由平衡器发送的。

4． 建立HASH表来记录已经建立的所有连接。

5. 接收外部网对请求的回应并将其转发到内部网中的发出请求的机器上。

6.2 模块流程

（1） 内部网中的机器向外部网中的机器发送连接请求的流程。

（2） 外部网中的机器向内部网中的机器发送连接请求的流程。

6.3 结构设计

程序名	标识符	功能	源程序
建立HASH队列	ip_masq_hash	根据m{addr， port}和s{addr， port}建立两个HASH队列	ip_masq.c
从HASH队列删除表项	ip_masq_unhash	将ip_masq表项从HASH队列中删除	ip_masq.c
处理从外到内的请求	__ip_masq_in_get	处理从外到内的请求，查询HASH表，找到内部机器的ip地址及端口	ip_masq.c
处理从内到外的请求	__ip_masq_out_get	处理从内到外的请求，查询HASH表，找出源地址、端口及目标地址、端口都匹配的表项。	ip_masq.c
减少某个连接的访问计数	__ip_masq_put	将ip_masq表项的访问计数减一。	ip_masq.c
取下一伪装端口	get_next_mport	取得下一个mport。	ip_masq.c
创建新的ip_masq结构	ip_masq_new	创建一个新ip_masq结构，并分配一个新的mport	ip_masq.c
从内到外请求处理的最上层函数	ip_fw_masquerade	负责从内向外请求的处理全过程。	ip_masq.c
从外到内请求处理的最上层函数	ip_fw_demasquerade	负责从外向内请求的处理全过程。	ip_masq.c

6.4 主要数据结构

struct  ip_masq{
struct  ip_masq  *m_link，  *s_link；
atomic_t   refcnt；
struct   timer_list  timer；
__u16  protocol；
__u16  sport，  dport，  mport；
__u32  saddr，  daddr，  maddr；
struct  ip_masq_seq  out_seq， in_seq；
void  *app_data；
struct  ip_masq  *control；
atomic_t  n_control；
unsigned  flags；
unsigned  timeout；
unsigned  state；
struct  ip_masq_timeout_table  *timeout_table；
}

6.5 算法及流程

（1） ip_masq_hash

格式：static ip_masq_hash(struct ip_masq *ms)

返回值：1：成功；0：失败。

处理流程：

1. 如果ms->flags 已设IP_MASQ_F_HASHED标志位，则返回0。

2．根据ms->protocol、 ms->maddr和 ms->mport产生hash->key。

3．将ms链接到hash[hash_key] 链的头位置。

4．增加ms的访问计数。

5．根据ms->protocol 、ms->sadd和ms->sport产生hash_key。

6．将ms链接到hash[hash_key] 链的头位置。

7．增加ms的访问计数。

8．ms->flags 设IP_MASQ_F_HASHED位。

9. 返回1。

（2）ip_masq_unhash

格式：static int ip_masq_unhash(struct ip_masq *ms)

返回值： 1：成功；0：失败。

处理流程：

1． 如果ms->flags未设IP_MASQ_F_HASHED位，则返回0。

2． 不然，做以下几步。

3． 根据ms->protocol、 ms->maddr和 ms->mport产生hash->key。

4．在hash[hash_key]链中找匹配的表项，将匹配项的访问计数减一，并从链表中删除此项。

5． 根据ms->protocol 、ms->sadd和 ms->sport产生hash_key。

6．在hash[hash_key]链中找匹配的表项，将匹配项的访问计数减一，并从链表中删除此项。

7． ms->flags设位~IP_MASQ_F_HASHED。

8． 返回1。

（3）__ip_masq_in_get

格式：static struct ip_masq *__ip_masq_in_get(int protocol， __u32 s_addr， __u16 s_port， __u32 d_addr， __u16 d_port)

返回值：返回一个ip_masq结构ms。

处理流程：

1． 根据 protocol 、 d_addr、 d_port产生hash_key。

2． 在hash[hash_key]链中找匹配的表项，满足： （protocol == ms->protocol && d_addr == ms->maddr && dport == ms->mport && (s_addr == ms->daddr || ms->flags & MASQ_DADDR_PASS) && (s_port == ms->dport || ms_flags & MASQ_DPORT_PASS)）。

3． ms的访问计数增1。

4． 返回ms。

（4）__ip_masq_out_get

格式：static struct ip_masq *__ip_masq_out_get(int protocol ， __u32 s_addr， __u16 s_port， __u32 d_addr， __u16 d_port)

返回值：返回一个ms。

处理流程：

1. 如果s_port不为0，做以下几步；不然，跳至第5步。

2． 根据 protocol 、 s_addr和s_port产生hash_key。

3. 在hash[hash_key]链中找匹配的表项，满足：（protocol == ms->protocol && s_addr == ms->saddr && sport == ms->sport && (d_addr == ms->daddr || ms->flags & MASQ_DADDR_PASS) && (d_port == ms->dport || ms_flags & MASQ_DPORT_PASS)）。

4. ms的访问计数增1。

5. 返回ms。

6. 根据 protocol 、 s_add、 0产生hash_key。

7. 在hash[hash_key]链中找匹配的表项，满足：（protocol == ms->protocol && s_addr == ms->saddr && sport == ms->sport && (d_addr == ms->daddr || ms->flags & MASQ_DADDR_PASS) && (d_port == ms->dport || ms_flags & MASQ_DPORT_PASS)）。

8. ms的访问计数增1。

9. 返回ms。

（5）__ip_masq_getbym

格式：static struct ip_masq *__ip_masq_getbym(int protocol， __u32 m_addr， __u16 m_port)

返回值：返回一个ms。

处理流程：

1． 根据 protocol 、m_add和 m_port产生hash_key。

2． 在hash[hash_key]链中找匹配的表项，满足：（protocol == ms->protocol && m_addr == ms->maddr && mport == ms->mport ）。

3． ms的访问计数加1。

4． 返回ms。

（6）ip_masq_out_get

格式：struct ip_masq *ip_masq_out_get(int protocol， __u32 s_addr， __u16 s_port， __u32 d_addr， __u16 d_port)

返回值：返回ms。

处理流程：

1. 为__ip_masq_lock加读锁。

2. 调__ip_masq_out_get(protocol， s_addr， s_port， d_addr， d_port)返回ms。

3. 解读锁__ip_masq_lock。

4. 为ms设置超时。

5. 返回ms。

（7）ip_masq_in_get

格式：struct ip_masq_in_get(int protocol， __u32 s_addr， __u16 s_port， __u32 d_addr， __u16 d_port)

返回值：返回ms。

处理流程：

1．为__ip_masq_lock加读锁。

2．调__ip_masq_in_get(protocol， s_addr， s_port， d_addr， d_port)返回ms。

3. 解读锁__ip_masq_lock。

4. 为ms设置超时。

6. 返回ms。

（8）__ip_masq_put

格式：static __inline __void __ip_masq_put(struct ip_masq *ms)

返回值：无

处理流程：

1. 将ms的访问计数减一。

（9）get_next_mport

格式：static __u16 get_next_mport(void)

返回值：返回一个16位的端口号。

处理流程：

1． 为masq_mport_lock加旋转锁。

2． Mport = htons(masq_port ++)。

3． 如果masq_port是最后一个锁，则masq_port = 开始锁。

4． 解旋转锁masq_port_lock。

5． 返回mport。

（10）ip_masq_new

格式：struct ip_masq *ip_masq_new(int proto， __u32 maddr， __u16 mport， __u32 saddr， __u16 sport， __u32 daddr， __u16 dport， unsigned mflags)

返回值：成功：返回ms；失败：返回null。

处理流程：

1． 如果没有可用端口，返回NULL。

2． 若mflags 设IP_MASQ_F_USER位，则 prio = GFP_KERNEL；不然， prio=GFP_ATOMIC。

3． 为ms分配空间，若失败，返回NULL。

4． MOD_INC_USE_CONUNT。

5． 清空ms。

6． 初始化ms->timer。

7． ms->timer.data = (unsigned long)ms；

ms->timer.function = masq_expire；

ms->protocol = proto；

ms->saddr = saddr；

ms->sport = sport；

ms->daddr = daddr；

ms->dport = dport；

ms->flags = NULL；

ms->app_data = NULL；

ms->control = NULL；

设ms->n_control 为0；

设ms->refcnt 为0；

ms->maddr = maddr；

ms->flags设IP_MASQ_F_NO_REPLY位。

8． 如果(（协议不是TCP 和UDP）|| mport)，ms的伪装端口设为mport。调用__ip_masq_in_get取得ms。

9． 如果没有匹配的ms，则ms->flags设 IP_MASQ_F_MPORT位；增加伪装端口计数；将ms加入HASH队列；调用ip_masq_bind_app(ms)；ms的访问计数加一；为ms的IP_MASQ_S_NONE状态设置超时；返回ms。

10. 如果有匹配项，则调用__ip_masq_put(mst)使访问计数减一；释放ms； MOD_DEC_USE_COUNT；返回NULL。

11. 调用mport = get_next_mport（）来找一个可用的端口。

12. 赋值 ms->mport = mport。

13. 调用ip_masq_bind_app(ms)。

14. 设n_fails为0。

15．ms的访问计数增一。

16．为ms的IP_MASQ_S_NONE设置超时。

17．返回ms。

(11)ip_fw_masquerade

格式：int ip_masquerade(struct sk_buff **skb_p， __u32 maddr)

返回值：成功：返回大于0的数；失败：返回-1。

处理流程：

1. 先做一些有关校验和的检查。

2. 调用ip_masq_out_get_iph看入口是否已经存在。

3. 如果入口存在的话，将老的ms从HASH表中删除，将ms->maddr替换为maddr，并将新的ms结构加入HASH队列。

4. 如果没有定义源端口，做以下几步：将老的ms从HASH表中删除，设置ms->sport =h.portp[0]，即设置ms的源端口为TCP头中的源端口，将新的ms加入HASH表。

5. 如果定义了IP_MASQ_F_DLOOSE，ms->dport = h .port[1]； ms->daddr = iph->daddr。

6. 如果入口不存在则创建一个新的ms。

7. 将IP头的源地址替换为平衡器的ip地址。

8. 将TCP头的源端口替换为ms的伪装端口。

9. 做一些有关校验和的操作。

10. ms的访问计数减一。

11. 返回0。

(12)ip_fw_demasquerade

格式： int ip_fw_demasquerade(struct sk_buff **skb_p)

返回值：成功：返回1；失败：返回-1。

处理流程：

1． 取得sk_buff中的数据取的起始地址。

2． 取得数据区的大小。

3． 取得针对协议的偏移量。

4． Maddr = iph->daddr。

5．察看协议类型，如果是ICMP，则调用ip_fw_demasq_icmp；如果是TCP，什么也不做；如果是UDP协议，则首先确认端口号在BEGIN 和END之间，然后对校验和进行一定的操作。

6． 看现有表中是否有匹配项，返回ms。

7．如果有匹配项，设置标识位~IP_MASQ_F_NO_REPLY；如果定义了非严格路由，则ms->dport = h.portp[0]， ms->daddr = iph_saddr。

8． 如果定义了IP_MASQ_F_NO_DPORT，则标识位设~IP_MASQ_F_DPORT， ms的目的端口设为TCP头中的源端口。

9．如果定义了IP_MASQ_F_NO_DADDR，ms的标识位设~IP_MASQ_F_NO_DADDR，ms的目标地址设为IP头的源地址。

10. 调用masq_skb_cow复制一个skb。

11. 将IP头的目标地址设为ms的源地址。

12. 将TCP头中的目标端口设为ms的源端口。

13. 调用ip_masq_app_pkt_in。

14. 对校验和进行操作。

15. 调用ip_send_check(iph)。

16. 调用masq_set_state(ms， 0 ， iph， h.portp )。

17. ms的访问计数减一。

18. 返回1。

回页首

7.1 设计思想

ip端口转发模块的主要工作：

1. 接受外部网的连接请求。

2. 对外只呈现平衡器，使所有请求看起来都是由平衡器处理的。

3. 建立一个端口转发链表。

4. 接收外部网发向内部网的请求。根据连接请求的源地址、源端口、目标地址和目标端口的信息察看链表中是否有对应表项，如果有，则调用ip伪装模块将请求转发到实际服务器上；如果没有对应表项，则创建一个新的端口转发表项，并且在ip伪装HASH表中增加相应的表项。将实际服务器的处理结果回传给平衡器，再由平衡器发给外部网中的客户。

5. 在ip端口转发模块中通过系统调用的实现函数对用户的系统调用进行处理，首先清空ip端口转发链表，然后建立对应于最轻负载机器ip地址的ip端口转发表项。

7.2 模块流程

7.3 结构设计

程序名	标识符	功能	源程序
删除表项	ip_portfw_del	在双向链表中删除一个ip_portfw结构的表项	ip_masq_portfw.c
清空双向链表	ip_portfw_flush	清空双向链表，释放空间	ip_masq_portfw.c
寻找匹配表项	ip_portfw_lookup	寻找符合要求的表项	ip_masq_portfw.c
设置权重	ip_portfw_edit	为匹配的表项设置权重	ip_masq_portfw.c
添加表项	ip_portfw_add	添加一个表项	ip_masq_portfw.c
控制双向链表	ip_portfw_ctl	对双向链表进行控制，进行添加、删除和清空操作	ip_masq_portfw.c
匹配表项	portfw_in_rule	判断是否存在匹配表项	ip_masq_portfw.c
创建新入口	portfw_in_creat	如果没有匹配表项，则创建一个新的ip_masq结构的表项。	ip_masq_portfw.c
系统调用处理函数	getip	接受来自用户层的系统调用，根据负载最轻的机器的IP值建立新的端口转发链表	ip_masq_portfw.c

7.4 主要数据结构

struct  ip_portfw{
    struct  list_head  list；
    __u32  laddr，  raddr；
    __u16  lport，  rport；
    atomic_t  pref_cnt；
    int  pref；
    }
struct  list_head{
struct  list_head  *next， *prev；
  }
 

7.5 算法及流程

（1） ip_portfw_del

格式： static __inline __ int ip_portfw_del(__u16 protocol， __u16 lport， __u32 laddr， __u16 rport， __u32 raddr)

返回值：如果模块的访问计数〉0，则返回一个大于0的值，否则返回0；

处理流程：

1. 在双向链表中找到这样的ip_portfw *n 结构，满足：

（n->lport == lport && (!laddr || n->laddr == laddr) && (!raddr || n->raddr == raddr)　&& (!rport || n->rport == rport )），从链表中删除此入口，释放n结构。

1． 若(&mmod_self -> mmod_nent)为真，则返回ESRCH，否则返回0。

（2） ip_portfw_flush

格式： static __inline__ void ip_portfw_flush(void)

返回值：无。

处理流程：

1. 对于TCP 和UDP链，循环做2. 3步

2. 从链表上删除一个入口，释放结构n。

3. 取下一入口。

（3） ip_portfw_lookup

格式： static __inline__ struct ip_portfw *ip_portfw_lookup(__u16 protocol， __u16 lport， __u32 laddr， __u32 *daddr_p， __u16 *dport_p)

返回值：返回一个ip_portfw结构的变量n。

处理流程：

1．在双向链表中找到这样的ip_portfw *n 结构，满足：

（lport == n->lport && laddr == n->laddr） ，若没有找到，则n = NULL。

2.赋值：*daddr_p = n->raddr；*dport_p = n->rport。

3.返回n。

（4） ip_portfw_edit

格式： static __inline__ int ip_portfw_edit(__u16 protocol， __u16 lport， __u32 laddr， __u16 rport， __u32 raddr， int pref)

返回值：返回满足匹配条件的表项的数目。

处理流程：

1. 设count为0。

2. 在双向链表中找这样的n，满足：（ lport == n->lport && (!laddr || laddr == n->laddr) && (!raddr || raddr == n->raddr)　&& (!rport || rport == n->rport )） 。

3. 为n设置权重。

4. 设ｎ->pref_cnt = pref。

5. 每找到一个匹配项， count都加1。

6. 解读锁portfw_lock。

7. 返回count。

（5） ip_portfw_add

格式： static __inline__ int ip_portfw_add(__u16 protocol， __u16 lport，__u32 laddr， __u16 rport，__u32 raddr， int pref)

返回值：成功：返回0；失败：返回EINVAL。

处理流程：

1. 调用ip_portfw_edit， 若有匹配项，则返回0。

2. 为npf分配一个 struct ip_portfw的空间，若分配失败，则返回"无可用空间" 。

3. MOD_INC_USE_COUNT。

4. npf清0。

5. npf->laddr = laddr；

npf->lport = lport；

npf->rport = rport；

npf->raddr = raddr；

npf->pref = pref；

设npf的访问计数为npf的权重；

6． 调用INIT_LIST_HEAD初始化npf->list。

7． 加写锁&portfw_lock。

8． 调用list_add增加一项到链头。

9． 解写锁。

10. 增加mmod_self计数。

11. 返回0。

（6） portfw_ctl

格式： static __inline__ int portfw_ctl(int optname，struct ip_masq_ctl *mctl， int optlen) 返回值：成功：返回0；失败：返回EINVAL。

处理流程：

1． 如果cmd为IP_MASQ_CMD_NONE，返回0。

2． 如果cmd为IP_MASQ_CMD_ADD，调用 ip_portfw_add。

3． 如果cmd为IP_MASQ_CMD_DEL，调用 ip_portfw_del。

4． 如果cmd为IP_MASQ_CMD_FLUSH，调用ip_portfw_flush。

5． 返回EINVAL。

（7）portfw_in_creat

格式： static struct ip_masq *portfw_in_creat(const struct sk_buff *skb， const struct iphdr *iph， __u32 maddr)

返回值： 成功：返回ms。

处理流程：

1. 加写锁&portfw_lock。

2. 调用ip_portfw_lookup，若找到匹配项，则调ip_masq_new创建一个新入口表项ms。

3. 调用ip_masq_listen。

4. pf的访问计数减一，若访问计数为0，则设pf的访问计数为pf的权重，将pf 从双向链表中删除，并重新加到链表尾。

5. 返回ms。

(8) 函数getip（）

格式：asmlinkage getip(__u32 laddr， __u16 lport， __u32 raddr， __u16 rport) 返回值：无。

处理流程：

1.调用ip_portfw_flush清空ip端口转发链表。

2.调用ip_portfw_add（laddr， lport， raddr， rport， 1）向链表中加入新表项。

回页首

8.1 设计思想

调度模块的主要任务是：

1． 平衡器向各台实际服务器发送收集负载信息的命令。

2． 各台实际服务器分别运行取cpu运行队列长度的程序。

3． 各台机器将各自的cpu运行队列长度信息回传给平衡器。

4．平衡器对各台机器的cpu运行队列长度进行比较并选出cpu运行队列长度最短的机器，认为此机器就是负载最轻的机器。

5． 通过系统调用将负载最轻的机器的IP地址传入ip端口转发模块。

8.2 模块流程

8.3 结构设计

程序名	功能	源程序
client程序	收集server方的负载信息并进行处理	client.c
server程序	收集本机负载信息并传向client方	server.c
负载信息收集程序	取cpu运行队列长度	cpu.c

8.4 主要数据结构

struct info{
     char  host[15]；
     char  cpu[20]；
          }
         

8.5 算法及流程

(1) cpu.c

处理流程：

1. 调用readprottab2取得进程表p_table。

2. 调用do_stat计算 cpu运行队列长度的信息。

(2) 函数do_stat

格式：void do_stat(proc_t **p， float elapsed_time， int pass， int ctl)

处理流程：

1． 设进程状态变量sleeping =0，stopped = 0， zombie = 0， running = 0。

2． 做以下循环：3，4步。

3． 根据进程表取得一个进程，判断它的状态是sleeping、 stopped 、 zombie还是 running，并且给相应的变量sleeping、 stopped 、 zombie、 running加1。

4． 取下一个进程，返回3。

5． 打印正在运行的进程的个数。

(3)client.c

处理流程：

1. 循环做以下步骤：2-11。

2. 设置控制变量i，循环做3-10步。

3. 如果i < NUM（内部网中机器的总数），做以下步骤。

4. 创建套接口sockfd。

5. 调用connect请求建立一个连接。

6. 调用recv将取得的字节流放入缓冲区buf中。

7. 将buf转换为struct info的格式并将转换后的结果赋给load[I]。

8. 比较各机器的cpu运行队列的长度并且确定负载最轻的机器的IP值，以该值为参数通过自定义的系统调用getip进入内核。

9. 关闭套接口sockfd。

8. 睡眠30秒钟。

(4)server.c

流程：

1. 创建套接口sockfd用来侦听。

2. 调用bind进行绑定。

3. 调用listen进行侦听。

4. 循环做以下步骤：5-11。

5. 调用accept接受连接请求。

6. 调用gethostname取得主机名，调用gethostbyname取得主机IP。

7. 将主机IP赋给load->host。

8. 将/loadbalan/cpu.o的运行结果写入./cpuinfo。

9. 打开./cpuinfo文件，将读取到的值写入 load->cpu。

10. 将结构load以字节流的形式发送出去。

11. 关闭套接口。

回页首

该负载平衡系统具有以下功能：

（1） 能够对基于TCP/IP协议的多种服务如telnet、ftp、 http等进行转发。在应用程序中，用户可以通过加入特定服务对应的端口号（如ftp对应的端口号是21）来增加对特定的服务的支持。

（2）实现了动态负载平衡。负载平衡器不断收集各台实际服务器正在运行的进程的个数，通过比较找出具有最少运行进程个数的机器，并且将请求发向这台机器。这种负载平衡是随各台机器现有的负载情况的变化而变化的，因此是动态的负载平衡。

（3）保证了持续连接。当来一个新的请求时，负载平衡器查找已建立连接的表项来看是否有匹配表项，如果发现已有匹配项，则将请求发往对应的实际服务器，这样就保证了来自同一个用户的同一个服务的多次请求能发送到同一台实际服务器上。

该负载平衡器具有以下的性能特点：

（1）具有较短的响应时间。从负载平衡器向集群中实际服务器发出收集负载的请求到决定出负载最轻的机器所用的总时间约为30ms,该负载平衡系统能够保证毫秒级的响应时间。

（2）具有良好的可伸缩性。用户可以在应用程序中对集群中的机器个数进行控制，程序中有一个数组来记录机器的IP地址，因此增加一台机器只需要向数组中增加一个新的元素，删除一台机器只需要从数组中删除对应的元素。

（3）具有良好的容错性。当集群中的某台机器突然崩溃时，应用程序能够立即发现，并且不会再将请求分发给失效的机器；当机器修复了错误重新进入集群中时，应用程序能够探测到并开始进行正常的负载收集工作。

• 
  
  
• 
  

		许广斌，男，工程师。您可以通过电子邮件 与他联系。