tbox新增stackless协程支持

yulihua49

zylthinking 发表于 2017-02-08 19:50
回复 56# yulihua49

你并发的是 10000 个连接, 不是 10000 个协程;  你的每个协程都是在获得任务时创建 ...

又给你发了一个与之配套的ASIO－同步异步两用IO程序。50楼。

1. cat AIO_tcp.c
   
2. /**********************************************
   
3. * @(#) SDBC 7.1 TCP RECV/SEND  Tools                      *
   
4. * to suport coroutine
   
5. **********************************************/
   
6. #include <sys/socket.h>
   
7. #include <sys/time.h>
   
8. #include <unistd.h>
   
9. #include <fcntl.h>
   
10. #include <sc.h>
    
11. 
    
12. #ifndef MIN
    
13. #define MIN(a,b) ((a)<(b))?(a):(b)
    
14. #endif
    
15. 
    
16. static T_YIELD yield=NULL;
    
17. 
    
18. T_YIELD get_yield()
    
19. {
    
20.         return yield;
    
21. }
    
22. 
    
23. T_YIELD set_yield(T_YIELD new_yield)
    
24. {
    
25. T_YIELD oldyield=yield;
    
26.         yield=new_yield;
    
27.         return oldyield;
    
28. }
    
29. 
    
30. //timeout for second
    
31. int RecvNet(int socket,char *buf,int n,int timeout)
    
32. {
    
33. int bcount=0,br,ret;
    
34. int i,repeat=0;
    
35. int fflag=-1;
    
36. 
    
37.         if(socket<0) return SYSERR;
    
38.         if(!buf && n<0) return 0;
    
39.         if(yield) {
    
40.                 fflag=fcntl(socket,F_GETFL,0);
    
41.                 if(fflag!=-1) fcntl(socket,F_SETFL,fflag|O_ASYNC|O_NONBLOCK); //异步操作
    
42.         } else if(timeout>0) {
    
43.                 struct timeval tmout;
    
44.                 tmout.tv_sec=timeout;
    
45.                        tmout.tv_usec=0;
    
46.                        ret=setsockopt(socket,SOL_SOCKET,SO_RCVTIMEO,(char *)&tmout,sizeof(tmout));
    
47.                 if(ret) ShowLog(1,"%s:setsockopt errno=%d,%s",__FUNCTION__,errno,strerror(errno));
    
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.                         repeat=0;
    
58.                         continue;
    
59.                 }
    
60.                 if(fflag==-1 && errno==EAGAIN) return TIMEOUTERR;
    
61.                 if(br<=0 && errno && errno != EAGAIN){
    
62.                     if(errno!=ECONNRESET)
    
63.                         ShowLog(1,"%s:br=%d,err=%d,%s",__FUNCTION__,br,errno,strerror(errno));
    
64.                     break;
    
65.                 }
    
66.                 if(bcount < n && fflag!=-1) { //切换任务
    
67.                         if(repeat++>3) return -errno;
    
68. /*
    
69. yield:
    
70. 1.找到当前线程的context。找不到返回-1；(后续任务以同步阻塞完成）
    
71. 2.把事件提交给epoll，作为resume的条件。
    
72. 3.swapcontext，挂起这个任务，线程回到epoll_wait，可以为别人服务了。
    
73. 4.一旦这个任务的事件发生了，立即由epoll_wait激活一个线程，抓取相应的context，使用setcontext，恢复任务现场，返回0.
    
74. 后续的动作就是继续NONBLOCK的IO。直至完成返回。
    
75. */
    
76. ShowLog(5,"%s:tid=%lx,socket=%d,yield to schedle bcount=%d/%d",__FUNCTION__,pthread_self(),socket,bcount,n);
    
77.                         i=yield(socket,0,timeout);
    
78.                         if(i<0) {
    
79.                           if(timeout>0) {
    
80.                                 struct timeval tmout;
    
81.                                 tmout.tv_sec=timeout;
    
82.                                 tmout.tv_usec=0;
    
83.                                 ret=setsockopt(socket,SOL_SOCKET,SO_RCVTIMEO,(char *)&tmout,sizeof(tmout));
    
84.                           }
    
85.                                 fcntl(socket,F_SETFL,fflag);
    
86.                                 fflag=-1;
    
87.                                 if(i==TIMEOUTERR) return i;
    
88.                         }
    
89.                 }
    
90.         }
    
91.         if(fflag!=-1) fcntl(socket,F_SETFL,fflag);
    
92.         return bcount==0?-1:bcount;
    
93. }
    
94. 
    
95. int SendNet(int socket,char *buf,int n,int MTU)
    
96. {
    
97. int bcount,bw;
    
98. int sz,i=0;
    
99. int fflag=-1;
    
100. size_t SendSize;
     
101. 
     
102.         if(socket<0) return SYSERR;
     
103.         if(yield) {
     
104.                 fflag=fcntl(socket,F_GETFL,0);
     
105.                 if(fflag != -1) fcntl(socket,F_SETFL,fflag|O_NONBLOCK); //异步操作
     
106.         }
     
107.         bcount=0;
     
108.         bw=0;
     
109.         if(MTU>500) SendSize=MTU;
     
110.         else SendSize=n;
     
111.         while(bcount<n){
     
112.                 sz=MIN(n-bcount,SendSize);
     
113.                 if((bw=write(socket,buf,sz))>0){
     
114.                         bcount+=bw;
     
115.                         buf+=bw;
     
116.                 }
     
117.                 if(bw<0&&errno!=EAGAIN) {
     
118.                         ShowLog(1,"%s:err=%d,%s",__FUNCTION__,errno,strerror(errno));
     
119.                         break;
     
120.                 }
     
121.                 if(bw < sz && fflag != -1) { //切换任务
     
122. //ShowLog(5,"%s:tid=%lx,socket=%d,yield bw=%d/%d",__FUNCTION__,pthread_self(),socket,bw,sz);
     
123.                     i=yield(socket,1,0);
     
124.                     if(i<0) {
     
125.                         fcntl(socket,F_SETFL,fflag);
     
126.                         fflag = -1;
     
127.                     }
     
128.                 }
     
129.         }
     
130.         if(fflag != -1)  fcntl(socket,F_SETFL,fflag);
     
131.         return bcount==0?-1:bcount;
     
132. }
     

复制代码

yulihua49

yulihua49 发表于 2017-02-08 20:27
又给你发了一个与之配套的ASIO－同步异步两用IO程序。50楼。

RecvNet:注意看54行,是直接进行read。多数情况直接就读出来了，不会进入后边的yield。
这个程序可以直接用于同步或异步场合，一般情况，yield是空的，就不会进入异步。在线程池服务器里会给它设一个yield函数，它就可以调用yield了，这时AIO_flag为真，task进入AIO模式，栈保持，线程逃逸。这个线程再要接受别的任务就会再分配栈。如果不调用yield，线程就不会逃逸，不会接受新任务，也就不会再分配栈。

一般的，异步调用会比同步慢。所以，这个程序尽量直接完成，多数不会进入异步状态，所以效率很高。

yulihua49

zylthinking 发表于 2017-02-08 19:50
回复 56# yulihua49

你并发的是 10000 个连接, 不是 10000 个协程;  你的每个协程都是在获得任务时创建 ...

所以，回到楼主的专题，stackless，使用起来限制太大，没什么实用价值。
自适应的协程池，是一个不错的解决方案。

yulihua49

zylthinking 发表于 2017-02-08 19:16
没说你的代码不好; 我说的是你之前的帖子有误导
你看看你之前说的:

你注意18楼，72行，如果需要的内存太多，不够用了，这个任务自动进入同步状态，不会耽误事的。这个系统设计就是这个原则，异步失败就改同步，不允许耽误事。
静态分配的协程，一旦资源不够，程序就无法运行。

实际的top显示，服务器的内存始终保持很小。

zylthinking

yulihua49 发表于 2017-02-09 09:42
你注意18楼，72行，如果需要的内存太多，不够用了，这个任务自动进入同步状态，不会耽误事的。这个系统设 ...

好吧.
我本来也是蒙圈了感觉一个协程一个栈似乎很浪费; 看着你似乎能做到多协程共享栈才跑来问问;
昨天仔细想了想, 貌似64bit的机器根本不用考虑堆栈使用多少内存问题了; 反正至少几百G的可用虚拟空间不是只有最多3G可用的时代了;

我不打算再纠结这玩意了, 就这样吧

yulihua49

zylthinking 发表于 2017-02-09 11:41
好吧.
我本来也是蒙圈了感觉一个协程一个栈似乎很浪费; 看着你似乎能做到多协程共享栈才跑来问问;
...

好的，就更没必要纠结stackless了。

windoze

回复 66# yulihua49

stackless coroutine的确没什么大用，但是单说stackless还是有用的，因为做了CPS变换之后，一个函数就可以无限递归，不用担心stack爆炸了。比如说tail recursion优化就可以认为是一种简化版的CPS变换，不光是节省了stack，而且性能也有提高。

yulihua49

windoze 发表于 2017-02-09 13:54
回复 66# yulihua49

stackless coroutine的确没什么大用，但是单说stackless还是有用的，因为做了CPS变 ...

同意。你可以在有栈的系统上自己做tail redcurion。

yulihua49

zylthinking 发表于 2017-02-09 11:41
好吧.
我本来也是蒙圈了感觉一个协程一个栈似乎很浪费; 看着你似乎能做到多协程共享栈才跑来问问;
...

一般的协程栈还是有问题。
如果支持10000个连接，每个连接一个协程，所需的栈是6M。那么需要60G的空间。这个仅仅是一个服务器的需求。
如果空间不够，连接时分配，可能连接到一定数量就崩溃了。可以采用动态commit的分配方式，更惨，不定什么时候就崩溃了。
采用我这个自适应栈池的方式，即使内存不够了，系统也不会崩溃，只不过后续的任务变成同步方式，不采用AIO罢了。稍等片刻系统就恢复如常。

分配不成就不分配，异步不成就同步，反正不会耽误事。系统具有超级鲁棒性。尤其是在遭受DDOS攻击时这个优点很明显。

wlmqgzm

不错， 讨论的很激烈啊，很久没有来， 这个帖子不错，感觉技术含量很高的样子，一些细节的东西我也不懂，

我这边做代码遇到各种异步的类似的需求，总体上是靠call_back回调解决， 就是编程代码比较碎片化，但是性能还是不错的，限制少。