leader/followers模式怎样高效地避免竞争状态的出现?

osmanthusgfy

leader/followers模式怎样高效地避免竞争状态的出现?

比如一个follower线程在某个sockfd上接收数据,需要很常时间(如传输文件),
这时leader线程调用reactor,得到可读的sockfd正是follower线程正在操作的sockfd,
这时就出现了竞争状态.

所以,一般使用leader/followers模式时要避免竞争状态的出现.
我想到一种方式:就是一个老外在多线程情况下使用libevent提出的一种思路:
使用socketpair创建一个unix域套接字.然后将这个套接字注册到reactor(比如epoll,select).


我仿照这种思路,可以在reactor中封装一个当前活动的sockfd的队列,这个队列中装的是当前被follower线程
操作的sockfd,每当leader线程调用reactor分离事件时,先在这个活动队列中查找这个sockfd,如果能找到,
则说明有某个follower线程正在操作这个sockfd,我们应该重新调用reactor,重新得到一个就绪的sockfd.
如果follower线程处理完毕,然后通过先前创建好的unix域套接字发送消息,请求将操作完毕sockfd从当前活动队列中
删除,这样就避免竞争状态的出现.

客户端连接达到很大数量后,我这种思路就出现性能低效的问题:
首先是每次leader->follower转变前,都必须在当前活动队列中查找一边;
其次,follower->leader转变前,又得将sockfd从当前活动队列中删除.
连接很多时,以上两步还是挺费时的,特别是查找.

有没有更好的解决方案了,希望大家赐教,谢谢!!!

yulihua49

leader/followers模式怎样高效地避免竞争状态的出现?

比如一个follower线程在某个sockfd上接收数据,需要 ...
osmanthusgfy 发表于 2011-03-07 16:27

完全没必要这么做。
在加入epoll时用EPOLLONESHOT标志。当正在进行操作的socket是不会被激活的。

一个socket具备了IO条件只被激活一次，然后被一个线程拿去它直接变成work线程。它可以任意处理这个socket，在重新EPOLL_CTL_MOD之前，是不会再一次激活的。
即，不会有另外的线程同时处理这个socket。

osmanthusgfy

我试试,之后报告测试结果
先感谢yulihua49大哥了!

bittertea

leader只accept ，
然后将accept的fd依次round 投递到各个followers线程里，通知followers，
各follows线程负责各自的fd 读写事件

不行么？

yovnchine

leader只accept ，
然后将accept的fd依次round 投递到各个followers线程里，通知followers，
各follows线 ...
bittertea 发表于 2011-03-07 18:16

这样就不是leader-follow模式了，每个follow都有可能成为leader的，你的模式follow永远是follow。

osmanthusgfy

报告测试结果:
yulihua49大哥的说法完全正确.如果是服务器的话,得注意区分服务器监听的sockfd,服务器与客户端连接的sockfd,使用epoll的时候,服务器与客户端连接的sockfd,可以使用EPOLLONESHOT标志(事件激活之后,处理完毕,再次注册即可);服务器监听的sockfd不要使用EPOLLONESHOT标志,因为使用这个标志,则导致服务器只能响应一个客户端的连接,这恐怕非我们所愿.
那服务器框架底层怎样分别对待这两种情况了?我是借鉴libevent的做法,服务器监听的sockfd在注册事件时,可以加上EV_PERSIST类似的标志,其他的情况,reactor默认只通知一次.

关于4楼提出的:leader/follower模式最重要的一点是:每个线程都应有公平的
leader->follower,follower->leader角色的转换,yovnchine大哥说的很正确.

再次yulihua49大哥!!!

yulihua49

leader只accept ，
然后将accept的fd依次round 投递到各个followers线程里，通知followers，
各follows线 ...
bittertea 发表于 2011-03-07 18:16

    LF的要点是L一旦取得资源，立即变成W，不进行任何线程切换，从而提高效率。你那样做导致一次任务在线程间的切换，违背了LF的初衷。
这个我有体会，我原来的交易中间件就是和你差不多，有一个调度线程，接到事件后丢到就绪队列，交给线程池处理。后来取消了调度线程，得到事件自己直接处理，效率提高了好多。

LF的缺点是一旦L离开，新的L到来之前，客户的请求被延迟。

后来我采用多个L，所有线程同时等待epoll，激活谁谁干活。没事的都去epoll_wait.实测性能很高。似乎没有发现无序竞争现象。

这样是否比LF好？

另外，监听口不在epoll处理，还是由select处理（select只处理这一个fd），激活后建立任务提交epoll。

结果，所有的F都成了L。
但是，在交易中间件里，不是所有的事件都是IO事件。有些非IO事件，例如取得连接事件，需要一个就绪队列，这个就绪队列需要一个线程守护，在这里采用了LF模式。
所有工作线程，留一个守候就绪队列，称为L，其他都去epoll，称为F。L可以直接变成W，F也可以。W们完成任务回来后，看一看有没有L，如果没有，留下来做L，其余都变成F。
代码片段：

1. 
   
2. while(1) {
   
3. //从就绪队列取一个任务
   
4. 
   
5. //              if(rpool.flg==0) {
   
6.                 pthread_mutex_lock(&rpool.mut);
   
7.                 do {
   
8.                         task=rdy_get();
   
9.                         if(!task) {
   
10.                              if(rpool.flg) break;
    
11.                              rpool.flg=1;
    
12.                              ret=pthread_cond_wait(&rpool.cond,&rpool.mut); //没有任务，等待
    
13.                    rpool.flg=0;
    
14.                         }
    
15.                 } while(!task);
    
16.                 pthread_mutex_unlock(&rpool.mut);
    
17. //              } else task=NULL;
    
18.                 if(!task) {
    
19.                         fds = epoll_wait(g_epoll_fd, &event, 1 , -1);
    
20.                         if(fds < 0){
    
21.                                 ShowLog(1,"%s:epoll_wait err=%d,%s",__FUNCTION__,errno,strerror(errno));
    
22.                                 sleep(30);
    
23.                                 continue;
    
24.                         }
    
25.                         task = (TCB *)event.data.ptr;
    
26.                         task->events=event.events;
    
27.                         if(!task->call_back && !(event.events&EPOLLIN)) {
    
28.                                 do_epoll(task,EPOLL_CTL_DEL);
    
29.                                 client_del(task);
    
30.                                 continue;
    
31.                         }
    
32.                 }
    
33. //              rs->timestamp=
    
34.                 task->timestamp=now_usec();
    
35.                 ret=do_work(task);
    
36.         }

复制代码

osmanthusgfy

yulihua49大哥这种思路真的有优化作用吗?
在这种模式至少有需要两个线程,一个线程select,用于accept新客户端的连接,另外的线程epoll或处理具体的事务.不知道select的线程,accept新客户端的连接之后是怎么处理的?是将这个新的clienfd注册到epoll,还是通过某种方式传递给另外那部分等待epoll的线程?
如果是第二种方式,那也会出现延时的显现.

yulihua49

yulihua49大哥这种思路真的有优化作用吗?
在这种模式至少有需要两个线程,一个线程select,用于accept新客户 ...
osmanthusgfy 发表于 2011-03-09 14:09

   select - accept是主线程干的。
select起超时作用，主线程除了负责客户的初始接入，还要在超时时处理服务器内的资源维护工作。
因此，本中间件是免维护的。

如果不是需要自动维护资源，主线程也可以不用select，直接accept即可。
accept之后，生成task，将task推给epoll。

代码：

1.         while(1) {
   
2.                 do {
   
3.                         FD_ZERO(&efds);
   
4.                         FD_SET(sock, &efds);
   
5. //健康检查周期
   
6.                         tm.tv_sec=30;
   
7.                         tm.tv_usec=0;
   
8.                         ret=select(sock+1,&efds,NULL,&efds,&tm);
   
9.                         if(ret==-1) {
   
10.                                 ShowLog(1,"select error %s",strerror(errno));
    
11.                                 close(sock);
    
12.                                 quit(3);
    
13.                         }
    
14.                         if(ret==0) {
    
15.                                 if(poolchk) poolchk();
    
16.                                 check_TCB_timeout();
    
17.                         }
    
18.                 } while(ret<=0);
    
19.                 i=event_no();
    
20.                 s=accept(sock,(struct sockaddr *)&cin,&leng);
    
21.                 if(s<0) {
    
22.                         ShowLog(1,"%s:accept err=%d,%s",__FUNCTION__,errno,strerror(errno));
    
23.                         switch(errno) {
    
24.                         case EMFILE:    //fd用完了,其他线程还要继续工作，主线程休息一下。  
    
25.                         case ENFILE:
    
26.                                 sleep(30);
    
27.                                 continue;
    
28.                         default:break;
    
29.                         }
    
30.                         sleep(15);
    
31.                         if(++repeat < 20) continue;
    
32.                         ShowLog(1,"%s:network fail! err=%s",__FUNCTION__,strerror(errno));
    
33.                         close(sock);
    
34.                         quit(5);
    
35.                 }
    
36.                 repeat=0;
    
37.                 client_q.pool[i].fd=s;
    
38.                 client_q.pool[i].conn.Socket=s;
    
39.                 client_q.pool[i].conn.timeout=120;
    
40.                 client_q.pool[i].status=-1;
    
41.                 client_q.pool[i].conn.only_do=(int (*)())conn_init;
    
42.                 ret=do_epoll(&client_q.pool[i],EPOLL_CTL_ADD);
    
43.         }
    

复制代码

这个方法已经在16core的SUSE11上实现了14万服务/秒的吞吐量。

osmanthusgfy

有时间试一下,实践出真理.

再次感谢yulihua49大哥!!!