寻求建议，关于异步编程

windoze

回复 30# codechurch

哦，我明白你在说什么了，你是说针对一个session多次创建/销毁context的时候stack位置会变化，所以不能保存指向桟上的指针。

我觉得如果程序没写错，那一个context在销毁的时候stack应该已经清空了，如果此时还有指向桟上的指针，那一定是程序写错了。

codechurch

回复 31# windoze


在那位（id太复杂）的设计里，session没有完成，栈也会销毁。

如：session在等待socket send完成，在socket send之后，将此session的栈回收，当send完成后，再恢复执行。

   

windoze

回复 32# codechurch

新创建的stack应该只包含“分岔点”之后的内容，之前的内容不应该在里面，这样就不会发生你说的问题了。
不过看 @yulihua49 的代码，他真的好像把整个stack都复制了一遍，这样真有可能出问题。

yulihua49

codechurch 发表于 2014-12-18 10:32
回复 22# yulihua49

我突然想到一个严重的问题。

我先前的那些程序都是错的，错的一塌糊涂，现在在调整，以后再说。

yulihua49

windoze 发表于 2014-12-18 11:42
回复 25# codechurch

不会崩溃的，尽管你切换到其它stack去了，但4001那块stack不是还在那儿吗。

会崩溃的，原来的地址被别人用了。
XXX（）
{
char a[400];
char *p=a;
.....
}

这种东西肯定完蛋。
栈迁移后，a改变了地址，p还指向原来的地方。
我这里这样用的挺多，真是个问题。

我这个方案，整体来说可能不行。

windoze

回复 35# yulihua49

我觉得直接给每个session创建一个context应该就行了，不需要搞的这么复杂。
如果你担心stack太大浪费内存，你可以用boost.context（这个东西说是C++写的，但其实就是一段几乎是纯C的代码加上些汇编，想避免C++自己拿来改成纯C花不了一个小时），stack的空间可以自己分配，在X86/X64上，最小可以是一个page，也就是4K，这样创建1M个fiber也不过只需要4G内存，对现有的64位系统来说真不是太大的开销，在Linux+GCC下，你还可以用split stack按需分配stack，这对现有的程序在源代码一级是透明的，不需要改动，代价是性能略有损失。

yulihua49

windoze 发表于 2014-12-18 17:44
回复 35# yulihua49

我觉得直接给每个session创建一个context应该就行了，不需要搞的这么复杂。

苦修了几天，终成正果。
“协程池”技术诞生了。
方案如下：
基于线程池，系统初始化，一堆空TCB，就是没有栈。
一个TCB，激活时，从线程获取栈，线程没有就分配。
完成任务后（一个service），归还栈给线程，归还不了就free。
其间，如果进入AIO未完成状态，就不归还栈。
这样，在do_work（）期间，就不会发生跳栈。

系统的栈需求为：线程数*2 + AIO未完成的fiber数。
这样，我可以根据需求为应用定义更大的栈空间。
在空间方面，任何时候fiber空间不足，就给他线程栈用，并不影响fiber的运行，只是不能AIO，变成同步阻塞的而已。

另外，4K的应用栈肯定不够，将够系统异常事件处理用。（就是栈下的空间）。
我现在是2M，将来根据用户需求配置。

试验日志：
5 midsc:19720 12/22 13:54'29 thread_work:reuse fiber
5 DB2FIRST:127.0.0.1:3 12/22 13:54'29 do_work:call_back tid=7F216DA7A700,USEC=397637667
3 DB2FIRST:127.0.0.1:3 12/22 13:54'29 do_epoll:tid=7F216DA7A700 epoll_ctl fd[3]=-1,deleted,op=2
3 DB2FIRST:127.0.0.1:3 12/22 13:54'29 from_server:TCB:3,tid=139781550352128,Status Recv from server localhost PROTO_NUM=0X0009,ERRNO1=0,ERRNO2=6,PKG_LEN=4589,T_LEN=4589,t_cont=00000006,USEC=3628302869992435
5 DB2FIRST:127.0.0.1:3 12/22 13:54'29 thread_work:fiber yield
5 DB2FIRST:127.0.0.1:3 12/22 13:54'29 thread_work:send to fiber
5 DB2FIRST:127.0.0.1:3 12/22 13:54'29 thread_work:fiber yield
5 DB2FIRST:127.0.0.1:3 12/22 13:54'29 thread_work:send to fiber
5 DB2FIRST:127.0.0.1:3 12/22 13:54'29 thread_work:fiber yield
5 DB2FIRST:127.0.0.1:3 12/22 13:54'29 thread_work:send to fiber
5 DB2FIRST:127.0.0.1:3 12/22 13:54'29 thread_work:fiber yield
5 DB2FIRST:127.0.0.1:3 12/22 13:54'29 thread_work:send to fiber
5 midsc:19720 12/22 13:54'29 thread_work:回收 fiber

yulihua49

yulihua49 发表于 2014-12-22 14:05
苦修了几天，终成正果。
“协程池”技术诞生了。
方案如下：

任务调度器：

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. //从就绪队列取一个任务
    
11.                 pthread_mutex_lock(&rpool.mut);
    
12.                 while(!(task=rdy_get())) {
    
13.                         if(rpool.flg >= tpool.rdy_num) break;
    
14.                         rpool.flg++;
    
15.                         ret=pthread_cond_wait(&rpool.cond,&rpool.mut); //没有任务，等待
    
16.                         rpool.flg--;
    
17.                 }
    
18.                 pthread_mutex_unlock(&rpool.mut);
    
19.                 if(!task) {
    
20.                         fds = epoll_wait(g_epoll_fd, &event, 1 , -1);
    
21.                         if(fds < 0){
    
22.                                 ShowLog(1,"%s:epoll_wait err=%d,%s",__FUNCTION__,errno,strerror(errno));
    
23.                                 sleep(30);
    
24.                                 continue;
    
25.                         }
    
26.                         task = (TCB *)event.data.ptr;
    
27.                         if(task->events == 0X10||task->fd==-1) {
    
28.                                 ShowLog(1,"%s:task already timeout",__FUNCTION__);
    
29.                                 continue;//已经进入超时状态
    
30.                         }
    
31.                         task->events=event.events;
    
32.                 }
    
33.                 rs->timestamp=now_usec();
    
34. 
    
35.                 task->uc.uc_link=&rs->tc;
    
36.                 if(task->AIO_flg) {//fiber task
    
37.                         set_showid(task->ctx);//Showid 应该在会话上下文结构里
    
38.                         pthread_mutex_lock(&task->lock);//防止其他线程提前闯入
    
39. ShowLog(5,"%s:send to fiber",__FUNCTION__);
    
40.                         setcontext(&task->uc);  //== longjmp()
    
41.                         continue;//no action,logic only
    
42.                 }
    
43.                 while(task->uc.uc_stack.ss_size>0)
    
44.                         ShowLog(2,"%s:wait pre fiber exit!",__FUNCTION__);
    
45.                 if(!rs->tc.uc_stack.ss_size) {
    
46. ShowLog(5,"%s:create fiber",__FUNCTION__);
    
47.                         task->uc.uc_stack.ss_sp=malloc(use_stack_size);
    
48.                         if(!task->uc.uc_stack.ss_sp) {//create fiber fault
    
49.                                 do_work(task->sv.TCB_no); //进行你的服务,不使用AIO
    
50.                                 if(task->status == -2) {
    
51.                                         client_del(task);
    
52.                                 }
    
53.                                 continue;
    
54.                         }
    
55.                 } else {
    
56. ShowLog(5,"%s:reuse fiber",__FUNCTION__);
    
57.                         task->uc.uc_stack.ss_sp=rs->tc.uc_stack.ss_sp;
    
58.                         rs->tc.uc_stack.ss_sp=NULL;
    
59.                         rs->tc.uc_stack.ss_size=0;
    
60.                 }
    
61.                 task->uc.uc_stack.ss_size=use_stack_size;
    
62.                 makecontext(&task->uc,do_work,1,task->sv.TCB_no);
    
63. 
    
64.                 ret=swapcontext(&rs->tc,&task->uc);
    
65.                 if(!task->AIO_flg) {//service complate
    
66.                         if(!rs->tc.uc_stack.ss_size) {//回收fiber stack
    
67. ShowLog(5,"%s:回收 fiber",__FUNCTION__);
    
68.                                 rs->tc.uc_stack.ss_size=use_stack_size;
    
69.                                 rs->tc.uc_stack.ss_sp=task->uc.uc_stack.ss_sp;
    
70.                         } else {
    
71. ShowLog(5,"%s:free fiber",__FUNCTION__);
    
72.                                 free(task->uc.uc_stack.ss_sp);
    
73.                         }
    
74.                         task->uc.uc_stack.ss_sp=NULL;
    
75.                         task->uc.uc_stack.ss_size=0;//mark fiber cpmplate
    
76.                         if(task->status == -2) {
    
77.                                 client_del(task);
    
78.                         }
    
79.                 } else {
    
80.                         pthread_mutex_unlock(&task->lock);
    
81. ShowLog(5,"%s:fiber yield",__FUNCTION__);
    
82.                         mthr_showid_del(rs->tid);
    
83.                 }
    
84.         }
    
85.         ShowLog(1,"%s:tid=%lX canceled",__FUNCTION__,pthread_self());
    
86.         mthr_showid_del(rs->tid);
    
87.         rs->timestamp=now_usec();
    
88.         rs->status=0;
    
89.         rs->tid=0;
    
90.         if(rs->tc.uc_stack.ss_sp) {
    
91.                 free(rs->tc.uc_stack.ss_sp);
    
92.                 rs->tc.uc_stack.ss_sp=NULL;
    
93.                 rs->tc.uc_stack.ss_size=0;
    
94.         }
    
95.         return NULL;
    
96. }
    
97. //yield to schedle
    
98. int do_event(int TCBno,int sock,int flg)
    
99. {
    
100. TCB *task;
     
101. int save_fd;
     
102. int ret;
     
103. 
     
104.         if(TCBno<0 ||TCBno>client_q.max_client) return -1;
     
105.         task=&client_q.pool[TCBno];
     
106.         if(task->uc.uc_stack.ss_size == 0) return MEMERR;
     
107.         save_fd=task->fd;
     
108.         task->AIO_flg=flg+1;
     
109.         task->fd=sock;
     
110.         pthread_mutex_lock(&task->lock);//防止其他线程提前闯入
     
111.         ret=do_epoll(task,0,flg);
     
112.         if(ret<0) {
     
113.                 pthread_mutex_unlock(&task->lock);
     
114.                 task->fd=save_fd;
     
115.                 task->AIO_flg=0;
     
116.                 return FORMATERR;
     
117.         }
     
118.         ret=swapcontext(&task->uc,task->uc.uc_link);
     
119.         pthread_mutex_unlock(&task->lock);
     
120.         task->fd=save_fd;
     
121.         task->AIO_flg=0;
     
122.         return ret;
     
123. }
     

复制代码

yulihua49

yulihua49 发表于 2014-12-22 14:32
任务调度器：

AIO_tcp.c:

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

复制代码

比先前的简单了许多。

yulihua49

codechurch 发表于 2014-12-17 12:01
本人用C语言在x86上实现了类似于C#的async await异步操作。

（windows使用setthreadcontext\getthreadco ...

这就是我最担心的。
每个fiber至少1M内存，10000个，就是10G.有这些资源，留给应用，用作内存数据库好不好？
况且这些内存，绝大多数是休眠或空闲的。

还有，你们的那个说法，fiber找个线程执行，我特不理解。如果一堆线程，没人找，那它在干啥？
我的说法是，线程，收到一个事件，得到TCB，为他生成一个运行栈，并运行它。
没事时都在epoll_wait。

线程是执行主体，fiber是客体，所以是线程运行fiber，而非fiber运行线程。