寻求建议，关于异步编程

yulihua49

action08 发表于 2014-12-17 15:18
你好，这些接口都是自己手动包装实现的吗？？？

不是。
用ucontext，自己补了两个asm小函数（4楼），不知对不对。
调度程序和异步IO程序见4楼给你的链接。
重发异步IO的程序：

1. /**********************************************
   
2. * @(#) TCP SERVER Tools                      *
   
3. * to suport Fiberized.IO
   
4. **********************************************/
   
5. #include <sys/socket.h>
   
6. #include <sys/time.h>
   
7. #include <unistd.h>
   
8. #include <fcntl.h>
   
9. 
   
10. #include "reg.h"
    
11. 
    
12. #if __WORDSIZE == 64
    
13. #define MAX_STACK 0X200000
    
14. #else
    
15. #define MAX_STACK 0X100000
    
16. #endif
    
17. 
    
18. #ifndef MIN
    
19. #define MIN(a,b) ((a)<(b))?(a):(b)
    
20. #endif
    
21. 
    
22. extern char *set_sp(char *new_stack);
    
23. extern char *restore_sp(char *to,char *from,char *bp,unsigned long size);
    
24. //timeout for second
    
25. int RecvNet(int socket,char *buf,int n,int timeout,int TCB_no)
    
26. {
    
27. int bcount=0,br,ret;
    
28. int i;
    
29. int fflag;
    
30. ucontext_t *tc=NULL,*uc=get_uc(TCB_no);
    
31. unsigned long begin_stack,save_stack_size=0;
    
32. 
    
33.         if(socket<0) return SYSERR;
    
34.         if(!buf && n<0) return 0;
    
35.         fflag=fcntl(socket,F_GETFL,0);
    
36.         if(uc) {
    
37.                 tc=uc->uc_link;
    
38.                 fcntl(socket,F_SETFL,fflag|O_NONBLOCK); //异步操作
    
39.         } else {
    
40.                 struct timeval tmout;
    
41.                 tmout.tv_sec=timeout;
    
42.                 tmout.tv_usec=0;
    
43.                 ret=setsockopt(socket,SOL_SOCKET,SO_RCVTIMEO,(char *)&tmout,sizeof(tmout));
    
44.                 if(ret) {
    
45.                         ShowLog(5,"%s:setsockopt err=%d,%s",__FUNCTION__,
    
46.                                 errno,strerror(errno));
    
47.                 }
    
48.         }
    
49. 
    
50.         *buf=0;
    
51.         br=0;
    
52. 
    
53.         while(bcount<n){
    
54.                 if((br=read(socket,buf,n-bcount))>0){
    
55.                         bcount+=br;
    
56.                         buf+=br;
    
57.                         continue;
    
58.                 }
    
59.                 if(errno==EAGAIN) return TIMEOUTERR;
    
60.                 if(br<0){
    
61.                     if(errno!=ECONNRESET)
    
62.                         ShowLog(1,"%s:br=%d,err=%d,%s",__FUNCTION__,br,errno,strerror(errno));
    
63.                     break;
    
64.                 }
    
65. //ShowLog(5,"RecvNet:read br=0,errno=%d,%s",errno,strerror(errno));
    
66.                 if(bcount < n && uc) { //切换任务
    
67.                       if(!uc->uc_stack.ss_size) {
    
68. //计算所需的栈帧
    
69.                         swapcontext(uc,uc);
    
70. #if __WORDSIZE == 64
    
71.                         begin_stack=tc->uc_mcontext.gregs[REG_RSP];
    
72.                         save_stack_size=uc->uc_mcontext.gregs[REG_RSP]-begin_stack;
    
73. #else
    
74.                         if(uc->uc_mcontext.gregs[REG_SS] == uc->uc_link->uc_mcontext.gregs[REG_SS]) {
    
75.                                 begin_stack=tc->uc_mcontext.gregs[REG_ESP];
    
76.                                 save_stack_size=uc->uc_mcontext.gregs[REG_ESP]-begin_stack;
    
77.                         }
    
78. #endif
    
79.                         if(save_stack_size > MAX_STACK) save_stack_size=0;
    
80.                         if(save_stack_size==0) {        //如果需要保存的栈帧太大，就不要异步了
    
81.                                 uc=NULL;
    
82.                                 fcntl(socket,F_SETFL,fflag);
    
83.                                 continue;
    
84.                         }
    
85. //创建fiber
    
86.                         uc->uc_stack.ss_size=save_stack_size+2048;
    
87.                         uc->uc_stack.ss_sp=malloc(uc->uc_stack.ss_size);
    
88.                         if(!uc->uc_stack.ss_sp) {
    
89.                                 uc->uc_stack.ss_size=0;
    
90.                                 uc=NULL;
    
91.                                 fcntl(socket,F_SETFL,fflag);
    
92.                                 continue;
    
93.                         }
    
94. //保存线程栈帧          memcpy(uc->uc_stack.ss_sp-save_stack_size,begin_stack-save_stack_size,save_stack_size);
    
95. //将实际的rsp,rbp也调过来  这需要一段asm
    
96.                         set_sp(uc->uc_stack.ss_sp-save_stack_size);
    
97.                       }
    
98.                         swapcontext(uc,uc);//在do_event之前，设定好栈
    
99.                         if(tc != uc->uc_link) //do_event后被别线程抢入了
    
100.                             continue;
     
101.                         i=do_event(TCB_no,socket,0);//EPOOLIN
     
102.                         swapcontext(uc,uc->uc_link); // thread escape
     
103.                 }
     
104.         }
     
105.         if(uc) {
     
106.                 if(uc->uc_stack.ss_sp) {// 销毁fiber
     
107. //恢复线程栈帧 需要一点ASM
     
108. //到新的uc->uc_link.uc_mcontext.gregs[REG_RSP];
     
109. #if __WORDSIZE == 64
     
110.                         memcpy(restore_sp(uc->uc_link->uc_mcontext.gregs[REG_RSP],
     
111.                                         uc->uc_stack.ss_sp,uc->uc_link->uc_mcontext.gregs[REG_RBP],
     
112.                                         save_stack_size),
     
113.                                 uc->uc_stack.ss_sp-save_stack_size,save_stack_size);
     
114. #else
     
115.                         memcpy(restore_sp(uc->uc_link->uc_mcontext.gregs[REG_ESP],
     
116.                                         uc->uc_stack.ss_sp,uc->uc_link->uc_mcontext.gregs[REG_EBP],
     
117.                                         save_stack_size),
     
118.                                 uc->uc_stack.ss_sp-save_stack_size,save_stack_size);
     
119. #endif
     
120. //需要恢复其他寄存器吗？
     
121.                         free(uc->uc_stack.ss_sp);
     
122.                         uc->uc_stack.ss_size=0;
     
123.                 }
     
124.                 fcntl(socket,F_SETFL,fflag);
     
125.         }
     
126.         return bcount==0?-1:bcount;
     
127. }
     
128. 
     
129. int SendNet(int socket,char *buf,int n,int MTU,int TCB_no)
     
130. {
     
131. int bcount,br;
     
132. int sz,i=0;
     
133. int fflag;
     
134. size_t SendSize;
     
135. ucontext_t *tc=NULL,*uc=get_uc(TCB_no);
     
136. unsigned long begin_stack,save_stack_size=0;
     
137. 
     
138.         if(socket<0) return SYSERR;
     
139.         fflag=fcntl(socket,F_GETFL,0);
     
140.         if(uc) {
     
141.                 tc=uc->uc_link;
     
142.                 fcntl(socket,F_SETFL,fflag|O_NONBLOCK); //异步操作
     
143.         }
     
144.         bcount=0;
     
145.         br=0;
     
146.         if(MTU>500) SendSize=MTU;
     
147.         else SendSize=n;
     
148.         while(bcount<n){
     
149.                 sz=MIN(n-bcount,SendSize);
     
150.                 if((br=write(socket,buf,sz))>0){
     
151.                         bcount+=br;
     
152.                         buf+=br;
     
153.                         continue;
     
154.                 }
     
155.                 if(br<0) break;
     
156.                 if(bcount < n && uc) { //切换任务
     
157.                     if(!uc->uc_stack.ss_size) {
     
158. //计算所需的栈帧
     
159.                         swapcontext(uc,uc);
     
160. #if __WORDSIZE == 64
     
161.                         begin_stack=tc->uc_mcontext.gregs[REG_RSP];
     
162.                         save_stack_size=uc->uc_mcontext.gregs[REG_RSP]-begin_stack;
     
163. #else
     
164.                         if(uc->uc_mcontext.gregs[REG_SS] == uc->uc_link->uc_mcontext.gregs[REG_SS]) {
     
165.                                 begin_stack=tc->uc_mcontext.gregs[REG_ESP];
     
166.                                 save_stack_size=uc->uc_mcontext.gregs[REG_ESP]-begin_stack;
     
167.                         }
     
168. #endif
     
169.                         if(save_stack_size > MAX_STACK) save_stack_size=0;
     
170.                         if(save_stack_size==0) {        //如果需要保存的栈帧太大，就不要异步了
     
171.                                 uc=NULL;
     
172.                                 fcntl(socket,F_SETFL,fflag);
     
173.                                 continue;
     
174.                         }
     
175. //创建fiber
     
176.                         uc->uc_stack.ss_size=save_stack_size+2048;
     
177.                         uc->uc_stack.ss_sp=malloc(uc->uc_stack.ss_size);
     
178. //保存线程栈帧          memcpy(uc->uc_stack.ss_sp-save_stack_size,begin_stack-save_stack_size,save_stack_size);
     
179. //将实际的rsp,rbp也调过来  这需要一段asm
     
180.                         set_sp(uc->uc_stack.ss_sp-save_stack_size);
     
181.                     }
     
182.                     swapcontext(uc,uc);//在do_event之前，设定好栈
     
183.                     if(tc != uc->uc_link) //do_event后被别线程抢入了
     
184.                         continue;
     
185.                     i=do_event(TCB_no,socket,1); //do_epoll EPOLLOUT
     
186.                     swapcontext(uc,tc); // thread escape
     
187.                 }
     
188.         }
     
189.         if(uc) {
     
190.                 if(uc->uc_stack.ss_sp) {// 销毁fiber
     
191. //恢复线程栈帧 需要一点ASM
     
192. //到新的uc->uc_link.uc_mcontext.gregs[REG_RSP];
     
193. #if __WORDSIZE == 64
     
194.                         memcpy(restore_sp(uc->uc_link->uc_mcontext.gregs[REG_RSP],
     
195.                                         uc->uc_stack.ss_sp,uc->uc_link->uc_mcontext.gregs[REG_RBP],
     
196.                                         save_stack_size),
     
197.                                 uc->uc_stack.ss_sp-save_stack_size,save_stack_size);
     
198. #else
     
199.                         memcpy(restore_sp(uc->uc_link->uc_mcontext.gregs[REG_ESP],
     
200.                                         uc->uc_stack.ss_sp,uc->uc_link->uc_mcontext.gregs[REG_EBP],
     
201.                                         save_stack_size),
     
202.                                 uc->uc_stack.ss_sp-save_stack_size,save_stack_size);
     
203. #endif
     
204. //需要恢复其他寄存器吗？
     
205.                         free(uc->uc_stack.ss_sp);
     
206.                         uc->uc_stack.ss_size=0;
     
207.                 }
     
208.                 fcntl(socket,F_SETFL,fflag);
     
209.         }
     
210.         return bcount==0?-1:bcount;
     
211. }
     
212. 
     
213.      

复制代码

codechurch

回复 9# action08

是啊。

   

codechurch

回复 6# windoze

如果是用 windows 的iocp 或者 linux aio 那种回调机制，我就在回调函数里直接执行了。此时，应该比用线程池快一点，虽然系统回调的线程，可能是在系统的线程池里。


   

yulihua49

codechurch 发表于 2014-12-17 15:39
回复 9# action08

是啊。

我的思路是：
基本中间件框架还是线程池模式，只在socket IO时异步，就是前边的两个函数。
只有在IO不能完成时生成fiber，释放线程。
完成后销毁fiber。
这样就不需要很多的fiber，很多的应用栈（栈的大小，你不知道留多少，尽可能大一些，很浪费的）。
我的方案，一是只有IO完不成才需要fiber。二是所需的栈大小是已知的，就是线程栈底，就是应用栈顶到当前rsp这一段内存。

windoze

回复 10# yulihua49

按我的理解大致描述一下你的程序流程，你看看对不对：

一个线程池做epoll_wait
有事件发生时，获得事件的thread得到fd，查找对应的TCB，不存在则创建一个
如果TCB里有一个挂起的context，则切换到这个context，
否则按照标准流程处理TCB，比如建立会话处理请求之类
在处理过程中，如果遇到阻塞操作，则为当前TCB保存一个现场context并挂起，然后该thread切换出去继续epoll_wait

看起来这个流程没什么问题，但有几点没搞清楚：
1、101行的do_event和下一行的swapcontext之间可能有race condition，因为你有一个thread pool在做epoll_wait，所以do_event把fd注册到epoll里之后，如果IO event立即/已经发生，这个fd可能立刻被其它thread获得，而这边的swapcontext可能还没来得及做，感觉可能会有问题。
2、如果这些TCB/session相互之间需要同步怎么办？如果你把一个获得某个锁的context挂起可能会导致其它thread阻塞甚至死锁
3、如果一个socket需要全双工操作怎么办？比如HTTP2的server，既要随时读client，又可能随时做server push，而这两个方向上的操作是完全异步的，按照你的设计TCB应该与连接一一对应，而好像又不可能给一个TCB建立两个context，所以你没办法在同一个连接上同时做两个阻塞操作。

yulihua49

windoze 发表于 2014-12-17 15:52
回复 10# yulihua49

按我的理解大致描述一下你的程序流程，你看看对不对：

TCB是一个预置的TCB池，其中之一被加入到epoll的event-》data中。
1.这就是11楼98-100行的作用。98行已经设置好context了，其他线程抢入时，直接进行io即可，尽管当前线程还未完全退出。
2.TCB之间，目前没有session，如果需要的话，可以用消息队列什么的，目前消息队列没有纳入异步。获取连接池的操作比较特殊，这就是主调程序里，那个rdy队列的作用。
3.目前我的交易中间件没有全双工操作。我还不知道epoll如何处理全双工操作。以SendNet为例，现状是，epoll使用ONESHOT，读的那个fd此时是不响应的，响应的这个fd一定是写的。

codechurch

回复 14# yulihua49

我的想法是，所有将所有客户业务，每个业务就是一个fiber。

因为无论如何，机器上总是有  N 个实际线程 和 M 个业务。

如果N个在实际线程的业务没有fiber，其实也不会优化多少，因为 N 是小值， M远大于N。而且，反而会造成 fiber create和destroy的性能损耗。


   

yulihua49

回复 17# codechurch
fiber栈不是在M里的吗？那会很多的啊。
我的方案，就是所有TCB（M个），不含栈，用线程栈（N个）。
创建、销毁fiber，就是malloc/free + 两个memcpy。


   

codechurch

回复 18# yulihua49

我计算过了：

    栈在windows上最小是16K。

    128K 个fiber （130000客户） 占内存 2G。

   


我认为在x64上，这不是问题。

而且在x86上可以多进程负载一下。



   

yulihua49

codechurch 发表于 2014-12-17 16:13
回复 18# yulihua49

我计算过了：

你不知道应用，它怎么用内存，递归什么的。所以线程栈，系统缺省值，32位是1M，64位是2M。
我要是做fiber，可能也要照这个数留，或者配置文件定一下。

当然现在大内存也没有什么太严重的经济问题。

我之所以用线程池模式，一个是避免大量线程，另一个是避免大量耗内存。这样，我可以通过调整线程栈尺寸，来适应大内存需求的应用。
我见过的，quicklz，我用的压缩器，栈需求是530K。
我自己的，创建hash表，栈需求与“桶”一样大。