epoll + 非阻塞I/O 怎么改进才能变成 多线程 + epoll + 非阻塞I/O

windoze

回复 50# yulihua49

不都是框架回调么？为什么要调用框架？

yulihua49

windoze 发表于 2014-12-12 23:48
回复 50# yulihua49

不都是框架回调么？为什么要调用框架？

正常是框架调用应用插件，但插件内也调用框架。
了解TUXEDO的可以知道。
我想异步化的，就是socket这一块。这是个公用程序，客户端、服务器都用，服务器还有PPC，TPC，TPOOL模式。
不是所有模式都支持这个机制，我需要一个兼容的方案。所以没办法用其他框架工具。

yulihua49

windoze 发表于 2014-12-12 23:48
回复 50# yulihua49

不都是框架回调么？为什么要调用框架？

请问，每一个fiber都有自己的栈，是吧？
那么与TPC的区别，除了线程少一些，栈空间一点不少。

windoze

回复 53# yulihua49

没错，fiber本来就省不下这些内存。
就我的观点这个其实也不能算是缺点，不管你用什么模型，每个客户session总是要存在什么地方的，不放在stack里也要想办法搞个context之类的东西把它存起来，这样做也许能省点内存，因为fiber的stack至少有一个page，大多数系统上是4k，实际的context也许没有这么大，但负面影响也是有的，你访问context内容的时候免不了要经过一个指针间接一下，而放在stack里就不用了，速度会比较快。
省stack内存的方案也是有的，gcc/clang支持split stack，可以按需分配stack内存，这样可以省不少地方，代价是性能会有点损失。

windoze

回复 52# yulihua49

fiber可以随意创建thread（所谓的foreign thread），foreign thread也可以随意创建fibio::scheduler并在其中运行一组fiber；
foreign thread可以用fibio::condition_variable通知fiber，fiber也可以用std::condition_variable/pthread_cond通知foreign thread。
fiberized.io中的*_mutex和condition_variable都可以在任意context里直接创建，不需要在fiber中创建（但需要在fiber中操作，除了cv的notify_one/notify_all），你可以直接定义全局变量；
fibio::concurrent_queue可以用来从外界向fiber中传数据，basic_concurrent_queue<T, std::mutex, std::condition_variable>可以用来从fiber向外界传数据。

特意这么设计的，这样fiberized.io会比较容易和其它库/框架合作。

yulihua49

windoze 发表于 2014-12-15 15:53
回复 53# yulihua49

没错，fiber本来就省不下这些内存。

我的设想如下：
只有IO完不成的时候才产生fiber，完成后销毁，SendNet是透明的，异步不异步看环境，就是能找到ucontext就做异步。

1. int SendNet(int socket,char *buf,int n,int MTU,int TCB_no)
   
2. {
   
3. int bcount,br;
   
4. int sz,i=0;
   
5. int fflag;
   
6. ucontext_t *uc=get_uc(TCB_no);
   
7. unsigned long begin_stack,save_stack_size=0;
   
8. 
   
9.         if(socket<0) return SYSERR;
   
10.         fflag=fcntl(socket,F_GETFL,0);
    
11.         if(uc) {
    
12. //计算所需的栈帧
    
13.                 swapcontext(uc,uc);
    
14. #if __WORDSIZE == 64
    
15.                 begin_stack=uc->uc_link->uc_mcontext.gregs[REG_RSP];
    
16.                 save_stack_size=(uc->uc_mcontext.gregs[REG_RSP]-begin_stack);
    
17. #else
    
18.                 if(uc->uc_mcontext.gregs[REG_SS] == uc->uc_link->uc_mcontext.gregs[REG_SS]) {
    
19.                   begin_stack=uc->uc_link->uc_mcontext.gregs[REG_ESP];
    
20.                   save_stack_size=(uc->uc_mcontext.gregs[REG_ESP]-begin_stack);
    
21.                 }
    
22. #endif
    
23.                 if(save_stack_size > 0X200000) save_stack_size=0; //>2M,
    
24.                 if(save_stack_size==0) {        //如果需要保存的栈帧太大，就不要异步了
    
25.                         uc=NULL;
    
26.                 } else {
    
27.                         fcntl(socket,F_SETFL,fflag|O_NONBLOCK); //异步操作
    
28.                 }
    
29.         }
    
30.         bcount=0;
    
31.         br=0;
    
32.         if(MTU>500) SendSize=MTU;
    
33.         else SendSize=n;
    
34.         while(bcount<n){
    
35.                 sz=MIN(n-bcount,SendSize);
    
36.                 if((br=write(socket,buf,sz))>0){
    
37.                         bcount+=br;
    
38.                         buf+=br;
    
39.                         continue;
    
40.                 }
    
41.                 if(br<0) break;
    
42.                 if(bcount < n && uc) { //切换任务
    
43.                     if(!uc->uc_stack.ss_size) {
    
44. //创建fiber
    
45.                         uc->uc_stack.ss_size=save_stack_size+256;
    
46.                         uc->uc_stack.ss_sp=malloc(uc->uc_stack.ss_size);
    
47. //保存线程栈帧   memcpy(uc->uc_stack.ss_sp-save_stack_size,begin_stack-save_stack_size,save_stack_size);
    
48.                         swapcontext(uc,uc);
    
49. //将REG_RSP，REG_RBP调整到新位置
    
50. //将实际的rsp,rbp也调过来  这需要一段asm
    
51. //  ......
    
52.                     }
    
53.                     i=do_event(socket,1); //do_epoll EPOLLOUT
    
54.                     swapcontext(uc,uc->uc_link); // thread escape
    
55.                 }
    
56.         }
    
57.         if(uc) {
    
58.                 if(uc->uc_stack.ss_sp) {// 销毁fiber
    
59. //恢复线程栈帧 需要一点ASM
    
60. //到新的uc->uc_link.uc_mcontext.gregs[REG_RSP];
    
61. //  ......
    
62.                         free(uc->uc_stack.ss_sp);
    
63.                         uc->uc_stack.ss_size=0;
    
64.                 }
    
65.                 fcntl(socket,F_SETFL,fflag);
    
66.         }
    
67.         return bcount==0?-1:bcount;
    
68. }
    

复制代码

48,49行可能是不需要的。uc->uc_link是本线程返回点的context。
54行本线程完成了退栈，转其他任务。再回来时使用的fiber栈。
你提些看法。

windoze

回复 56# yulihua49

大致看了下，我还是觉得这么搞太麻烦了，一个SendNet随着调用点不同，语义也会发生变化，调用这个函数的程序怎么适应这种变化呢？

另外对你这段程序有个小问题，你根据条件动态创建了一个新context，这个新context要执行什么东西呢？新context什么时候切回当前的context？看起来你这段程序只是在做AIO，没法加入其它功能啊？

实际情况下绝大多数程序里stack都不会太大的，一般context里占内存的东西都是在heap里，stack上只有个小结构体而已，fiberized.io里default stack size只有几k，如果你打开了segmented stack就只有一个page（确切说是2个，但有一个是guard page，不占物理内存），实在犯不上折腾。
当然，如果你是在搞VxWorks之类的嵌入式，就当我没说…………

yulihua49

windoze 发表于 2014-12-15 17:11
回复 56# yulihua49

大致看了下，我还是觉得这么搞太麻烦了，一个SendNet随着调用点不同，语义也会发生 ...

新的conext就是为了这个fd能够回来，在epoll_wait之后。那时可能是另一个线程了。
这只是一个底层的功能。其他功能都在主调程序处理。
调用时，如果是线程池，TCB为正整数，可以取出uc。否则，TCB <0,取uc＝＝NULL，就不进行异步IO。

53行把fd加到epoll。
54行回到主调程序。回去后线程退栈，忘掉先前的do_work那些事，守护epoll_wait。许多线程没事都在此。epoll有事返回，取出TCB，如果该TCB没有fiber，就调用do_work。否则setcontext()该TCB的context，它就回到55行。
实际的应用逻辑是do_work干的，do_work可能会调用SendNet来发信息，它视SedNet为同步阻塞调用。
IO完成后恢复原线程栈到新的线程，然后返回原do_work。

windoze

回复 58# yulihua49

这么做的话是不是需要一个全局的TCB表，每个连接/session/task一个TCB？如果你有多个连接/session/task，这个表是不是需要动态增长？是不是在里面还要加个读写锁？
PS. 没找到do_work，你是说do_event吗？

yulihua49

windoze 发表于 2014-12-16 10:46
回复 58# yulihua49

这么做的话是不是需要一个全局的TCB表，每个连接/session/task一个TCB？如果你有多 ...

有一个全局的TCB表，根据配置文件生成一个数组，数千到数万的规模吧。
所有TCB活动时使用线程栈。
主调函数如下：

1. static void *thread_work(void *param)
   
2. {
   
3. resource *rs=(resource *)param;
   
4. int ret,fds;
   
5. TCB *task;
   
6. struct epoll_event event;
   
7. 
   
8.         rs->tid=pthread_self();
   
9.         while(1) {
   
10.         int timeout=0;
    
11.                 if(!rs->tc.uc_stack.ss_size) {
    
12.                         rs->tc.uc_stack.ss_sp=malloc(4096);
    
13.                         rs->tc.uc_stack.ss_size=4096;
    
14.                         swapcontext(&rs->tc,&rs->tc); //==setjmp();
    
15.                 } else mthr_showid_del(rs->tid);
    
16. //从就绪队列取一个任务
    
17.                 pthread_mutex_lock(&rpool.mut);
    
18.                 while(!(task=rdy_get())) {
    
19.                         if(rpool.flg >= tpool.rdy_num) break;
    
20.                         rpool.flg++;
    
21.                         ret=pthread_cond_wait(&rpool.cond,&rpool.mut); //没有任务，等待
    
22.                         rpool.flg--;
    
23.                 }
    
24.                 pthread_mutex_unlock(&rpool.mut);
    
25.                 if(!task) {
    
26.                         fds = epoll_wait(g_epoll_fd, &event, 1 , -1);
    
27.                         if(fds < 0){
    
28.                                 ShowLog(1,"%s:epoll_wait err=%d,%s",__FUNCTION__,errno,strerror(errno));
    
29.                                 sleep(30);
    
30.                                 continue;
    
31.                         }
    
32.                         task = (TCB *)event.data.ptr;
    
33.                         timeout=task->timeout;
    
34.                         task->timeout=0;
    
35.                         if(task->events == 0X10||task->fd==-1) {
    
36.                                 ShowLog(1,"%s:task already timeout",__FUNCTION__);
    
37.                                 continue;//已经进入超时状态
    
38.                         }
    
39.                         task->events=event.events;
    
40.                 }
    
41.                 rs->timestamp=now_usec();
    
42. 
    
43.                 task->uc.uc_link=&rs->tc;
    
44.                 if(task->uc.uc_stack.ss_size) {//fiber task
    
45.                         set_showid(task->ctx);//Showid 应该在会话上下文结构里
    
46.                         setcontext(&task->uc);  //== longjmp()
    
47.                         continue;//no action,logic only
    
48.                 }
    
49.                 ret=do_work(task); //进行你的服务
    
50.                 task->timestamp=now_usec();//task与rs的差就是应用任务执行时间
    
51.                 switch(ret) {
    
52.                 case -1:
    
53.                         do_epoll(task,EPOLL_CTL_DEL,0);
    
54.                         client_del(task);
    
55.                 case THREAD_ESCAPE:
    
56.                         break;
    
57.                 default:
    
58.                         if(!task->timeout) task->timeout=timeout; //timeout 没有被应用设置过
    
59.                         if(do_epoll(task,EPOLL_CTL_MOD,0) && errno != EEXIST) {
    
60.                                 ShowLog(1,"%s:cancel by server",__FUNCTION__);
    
61.                                 client_del(task);
    
62.                         }
    
63.                         break;
    
64.                 }
    
65.                 mthr_showid_del(rs->tid);
    
66.         }
    
67.         ShowLog(1,"%s:tid=%lX canceled",__FUNCTION__,pthread_self());
    
68.         mthr_showid_del(rs->tid);
    
69.         rs->timestamp=now_usec();
    
70.         rs->status=0;
    
71.         rs->tid=0;
    
72.         if(rs->tc.uc_stack.ss_sp) {
    
73.                 free(rs->tc.uc_stack.ss_sp);
    
74.                 rs->tc.uc_stack.ss_sp=NULL;
    
75.                 rs->tc.uc_stack.ss_size=0;
    
76.         }
    
77.         return NULL;
    
78. }
    

复制代码

15行定义了返回点，46行发送到fiber。
49行do_work，貌似就是业务逻辑了。实际上还是框架逻辑，由它根据客户端请求或事件状态再调用应用函数或应用rollback。
在do_work和应用中，都可能调用SendNet，RecvNet，这两个函数需要fiberized。