[转]ASIO 与协程

蔡万钊

转贴自  http://microcai.org/2013/04/22/asio-statemachine.html

前段时间看了 ASIO 爸爸关于ASIO的一个演讲. ASIO 爸爸说, ASIO 的设计理念就是作为一个 toolkit 而不是一个框架. ASIO并不强迫你使用某种编程模型. 它只是提供一系列的函数和类帮你更容易的编程.

ASIO 的设计思想其实和 GLIB 的 g_main_loop 非常像. 但是 C++ 因为有模板, 所以能更好的实现. 使用 g_main_loop 的时候, 我不记得我写了多少 void 到 struct 的转化了. 这就是 C 的弊病. 因为缺少模板,所以回调必须严格符合函数签名. 但是库作者永远想不到用户还需要什么样的参数. 于是void就是最好的解决办法了.

但是因为 C++ 有模板,所以放松了对函数签名的要求. 不符合 ASIO 回调函数签名要求的,都可以使用 bind 绑定以符合签名要求. 这就为库的使用者带来了巨大的灵活性的同时仍然保持极佳的易用性.

C++的另一个强大优势即是函数对象(bind的结果就是一个函数对象). 通过函数对象, 使得一些状态可以保存到成员变量中, 这样就给了我们一个不需要栈就实现的一个协程. 协程本质上是一个拥有多重入口和多重出口的函数. 每次调用都将从上一次退出的地方继续执行. 如果没有成员变量, 所有的变量都将保持于栈上, 这就要求协程的实现必定带来栈切换. 而栈的要求就导致协程必须有额外的创建和撤销动作. 这都给协程的实际使用造成了限制.

C++的函数对象, 既可以像普通函数那样调用, 又可以像通常的对象那样携带成员变量. 因此通过将变量置于对象而不是栈上, C++就可以实现无栈协程. 无栈协程的开销不过就是一个函数对象, 而无需一次性分配一个巨大的内存用于栈. 通常一个栈实现的协程,需要分配一个 1MB~10MB 的内存用于栈. 而无栈协程所使用的对象,不过就只是一个函数对象,通常大小不过几个字节到几百字节. 因此无栈协程允许创建大量的协程,而无需当心内存开销.

ASIO 爸爸即创造性的提出了无栈协程的概念, 然后将它用于实践中.

使用无栈协程, 伴随的ASIO的另一个重要用法: 复合回调函数. 将多个回调函数复合为一个. 多个ASIO操作使用同一个回调函数进行处理. 然后使用 Duff's Device 的方法自动构建状态机. 每次调用自动的从上一次退出的地方继续执行.

由于 ASIO 对于回调函数是执行的拷贝操作,因此用于 ASIO 协程的函数对象必须是可拷贝的. 实现这个其实也非常轻松, 即对于不能拷贝的成员变量使用 shared_ptr 封装即可.

每次使用 yield 发起异步操作的时候, 记住, 当前的对象会在退出函数的时候撤销. 但是于此同时, ASIO 也获得了对象的一份拷贝. 用于下次回调. 这个特性有时候可以被利用, 用于实现非常复杂的逻辑. avbot 的 RPC 实现即利用了这个特性. 将复杂的逻辑使用寥寥数行代码轻松实现.

yulihua49

蔡万钊 发表于 2013-04-22 16:36
转贴自  http://microcai.org/2013/04/22/asio-statemachine.html

前段时间看了 ASIO 爸爸关于ASIO的一 ...

看着和continuation有点像。
http://en.wikipedia.org/wiki/Continuation
http://msdn.microsoft.com/en-us/library/ee372288.aspx

windoze

回复 2# yulihua49

functional language里的continuation和imperative language里的coroutine，本来干的事就差不多。

yulihua49

windoze 发表于 2013-04-23 10:29
回复 2# yulihua49

functional language里的continuation和imperative language里的coroutine，本来干的 ...

时过一年半了，我终于踏入这个领域。
实现了自己的AIO/coroutine.
过程很复杂，结果很简单，C的ASIO也很容易的实现了，并做成了框架，供任何的应用调用。
似乎也没有那么多viod *了。

任何人只要调用这两个函数（收，发）就可以了。

1. #include <sys/socket.h>
   
2. #include <sys/time.h>
   
3. #include <unistd.h>
   
4. #include <fcntl.h>
   
5. #include <scsrv.h>
   
6. 
   
7. #ifndef MIN
   
8. #define MIN(a,b) ((a)<(b))?(a):(b)
   
9. #endif
   
10. 
    
11. //timeout for second
    
12. int RecvNet(int socket,char *buf,int n,int timeout)
    
13. {
    
14. int bcount=0,br,ret;
    
15. int i,repeat=0;
    
16. int fflag=-1;
    
17. 
    
18.         if(socket<0) return SYSERR;
    
19.         if(!buf && n<0) return 0;
    
20.         fflag=fcntl(socket,F_GETFL,0);
    
21.         if(fflag!=-1) fcntl(socket,F_SETFL,fflag|O_ASYNC|O_NONBLOCK); //异步操作
    
22. 
    
23.         *buf=0;
    
24.         br=0;
    
25. 
    
26.         while(bcount<n){
    
27.                 if((br=read(socket,buf,n-bcount))>0){
    
28.                         bcount+=br;
    
29.                         buf+=br;
    
30.                         repeat=0;
    
31.                         continue;
    
32.                 }
    
33.                 if(fflag==-1 && errno==EAGAIN) return TIMEOUTERR;
    
34.                 if(br<=0 && errno && errno != EAGAIN){
    
35.                     if(errno!=ECONNRESET)
    
36.                         ShowLog(1,"%s:br=%d,err=%d,%s",__FUNCTION__,br,errno,strerror(errno));
    
37.                     break;
    
38.                 }
    
39.                 if(bcount < n && fflag!=-1) { //切换任务
    
40.                         if(repeat++>3) return -errno;
    
41. ShowLog(5,"%s:tid=%lX,socket=%d,yield to schedle bcount=%d/%d",__FUNCTION__,pthread_self(),socket,bcount,n);
    
42.                         i=do_event(socket,0,timeout);//yield by EPOOLIN
    
43.                         if(i<0) {
    
44.                           if(timeout>0) {
    
45.                                 struct timeval tmout;
    
46.                                 tmout.tv_sec=timeout;
    
47.                                 tmout.tv_usec=0;
    
48.                                 ret=setsockopt(socket,SOL_SOCKET,SO_RCVTIMEO,(char *)&tmout,sizeof(tmout));
    
49.                           }
    
50.                                 fcntl(socket,F_SETFL,fflag);
    
51.                                 fflag=-1;
    
52.                                 if(i==-11) return -11;
    
53.                         }
    
54.                 }
    
55.         }
    
56.         if(fflag!=-1) fcntl(socket,F_SETFL,fflag);
    
57.         return bcount==0?-1:bcount;
    
58. }
    
59. 
    
60. int SendNet(int socket,char *buf,int n,int MTU)
    
61. {
    
62. int bcount,bw;
    
63. int sz,i=0;
    
64. int fflag;
    
65. size_t SendSize;
    
66. 
    
67.         if(socket<0) return SYSERR;
    
68.         fflag=fcntl(socket,F_GETFL,0);
    
69.         if(fflag != -1) fcntl(socket,F_SETFL,fflag|O_NONBLOCK); //异步操作
    
70.         bcount=0;
    
71.         bw=0;
    
72.         if(MTU>500) SendSize=MTU;
    
73.         else SendSize=n;
    
74.         while(bcount<n){
    
75.                 sz=MIN(n-bcount,SendSize);
    
76.                 if((bw=write(socket,buf,sz))>0){
    
77.                         bcount+=bw;
    
78.                         buf+=bw;
    
79.                 }
    
80.                 if(bw<0&&errno!=EAGAIN) {
    
81.                         ShowLog(1,"%s:err=%d,%s",__FUNCTION__,errno,strerror(errno));
    
82.                         break;
    
83.                 }
    
84.                 if(bw < sz && fflag != -1) { //切换任务
    
85. ShowLog(5,"%s:tid=%lX,bw=%d,sz=%d",__FUNCTION__,pthread_self(),bw,sz);
    
86.                     i=do_event(socket,1,0); //yield by EPOLLOUT
    
87.                     if(i<0) {
    
88.                         fcntl(socket,F_SETFL,fflag);
    
89.                         fflag = -1;
    
90.                     }
    
91.                 }
    
92.         }
    
93.         if(fflag != -1)  fcntl(socket,F_SETFL,fflag);
    
94.         return bcount==0?-1:bcount;
    
95. }
    

复制代码

秘密在于：do_event(); 它把socket加入epoll，然后yield到schedle（线程池）。返回时可能就不是原来的线程了。
它身后有几百行的代码支持。
原先的代码，基本无需修改（还是改了一点点），就自动变成AIO了。

这样实现了一个封装，用户什么也不知道，悄悄把事办了。
什么线程池啊，epoll啊，ucontext啊，coroutine啊，统统与应用没有关系。
应用只需注意：1.线程安全；2.线程锁不得跨越AIO。

这框架是可靠的，栈、锁、超时都管理到位了，死锁和死循环都被消灭了。

windoze

回复 4# yulihua49

你还可以更进一步，直接hook库函数，这样就二进制兼容了

yulihua49

windoze 发表于 2015-01-20 15:50
回复 4# yulihua49

你还可以更进一步，直接hook库函数，这样就二进制兼容了

怎么整？

windoze

回复 6# yulihua49

把你的send/recv函数搞得和标准函数调用方式一样，打包在一个so里，然后用LD_PRELOAD

yulihua49

windoze 发表于 2015-01-20 16:03
回复 6# yulihua49

把你的send/recv函数搞得和标准函数调用方式一样，打包在一个so里，然后用LD_PRELOA ...

现在难办的就是数据库，拿不到它的socket，也不知道他们用的是哪种IO函数，有没有跨越IO的锁；
而且，我也不敢重载read，write。毕竟我只能处理socket，文件和其他的也处理不了。
我这个已经跟read、write差不多了，read多了一个timeout，write多了一个MTU。

windoze

回复 8# yulihua49

嗯，所以说光做coroutine+Async I/O是不够的，你总归要想办法和foreign thread通信，早晚而已。
或者你干脆把数据库操作放到另外一个进程里，用Unix domain socket和主程序通信……

yulihua49

windoze 发表于 2015-01-20 17:56
回复 8# yulihua49

嗯，所以说光做coroutine+Async I/O是不够的，你总归要想办法和foreign thread通信， ...

这个建议值得考虑。我想可以放到另一些线程处理，通信机制我这里有。