SIO,AIO通用的socket收发接口

yulihua49

虽然AIO有某些好处，但是程序难写。
你可以用下面的程序进行接口。其中sc.h是我专用的，你换成你的即可，里边就是函数原型。那里边有个：
typedef int (*T_YIELD)(int socket,int rwflg,int timeout);
你的yield功能按这个接口。

要想发送就：SendNet(socket,buf,size,MTU);参数的含义你懂的。
要想接收就： RecvNet(socket,buf,size,timeout);
一般情况下他是同步阻塞的。
如果你能提供协程的yield功能，把yield设置好，他就变成非阻塞的了。
你可以去掉ShowLog或改成你的

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

复制代码

windoze

这个没有说清回调函数是在哪个context下跑呢？

yulihua49

windoze 发表于 2015-11-16 18:02
这个没有说清回调函数是在哪个context下跑呢？

它是通用的，不涉及特定的context。
context是在你的yield里，自己定义和使用。

我是在yield里，通过线程池记录的thread context和user context交换。
以当前线程号为索引，在线程池里查找。如果是foriegn thread就返回－5，
超时返回－11.正常resume返回0；
ret!=0 就改阻塞方式了。这个模块经过调试。是我现在用的。

明天发个yield程序。供参考，你们可以使用你们的yield，按这个接口即可。
对于初友，先用这两个程序完成普通的网络IO，实现主要功能，今后长进了，加入coroutine，自动成了NIO。其他代码不用大改。
对于我来说，我的中间件要支持PPC、TPC、TPOOL模式，前二种不需要NIO，只有3需要NIO和协程。三种模式可以共用底层模块。
在3下，set_yield，就成了NIO。
玩coroutine很久才想出这个全通用，全透明的方案。

yulihua49

yulihua49 发表于 2015-11-16 20:41
它是通用的，不涉及特定的context。
context是在你的yield里，自己定义和使用。

在线程池服务模块里：

1. 
   
2. static T_YIELD other_yield=NULL;//保存先前应用设置的yield。
   
3. 
   
4. TPOOL_srv()
   
5. {
   
6. ........
   
7.         other_yield=set_yield(do_yield);//返回通常是NULL。
   
8. .......
   
9.         服务循环
   
10. ......
    
11. }
    
12. 
    

复制代码

那么，do_yield是什么呢？

yulihua49

yulihua49 发表于 2015-11-17 14:08
在线程池服务模块里：那么，do_yield是什么呢？

1. 
   
2. static int do_yield(int socket,int rwflg,int timeout)
   
3. {
   
4. int ret=do_event(socket,rwflg,timeout);
   
5.         if(THREAD_ESCAPE == ret && other_yield)
   
6.                 ret=other_yield(socket,rwflg,timeout);
   
7.         return ret;
   
8. }
   

复制代码

可以看到，这个方案支持多个线程组分别进行yield，每个都可以不同的context和yield方法。

首先在服务器线程组进行，如果是foriegn thread，就转入应用提供的线程组，再不行就转入失败，进入同步模式。

那么，do_event是什么呢？

yulihua49

yulihua49 发表于 2015-11-17 14:15
那么，do_event是什么呢？ ...

它就是先在本服务器线程组里查找当前线程，然后。。。。

1. 
   
2. //yield to schedle
   
3. static int do_event(int sock,int flg,int timeout)
   
4. {
   
5. TCB *task;
   
6. int save_fd,save_timeo=-1;
   
7. int ret;
   
8. pthread_t tid=pthread_self();
   
9. resource *rs=tpool.pool;
   
10. 
    
11.         for(ret=0;ret<tpool.num;ret++,rs++) { //找原来的线程
    
12.                 if(tid == rs->tid) break;
    
13.         }
    
14.         if(ret>=tpool.num) return THREAD_ESCAPE;
    
15.         task=(TCB *)rs->tc.uc_link;//在这里拿到了context
    
16.         if(!task || task->uc.uc_stack.ss_size == 0) return MEMERR;
    
17.         save_fd=task->fd;
    
18.         if(timeout>0) {
    
19.                 save_timeo=task->timeout;
    
20.                 task->timeout=timeout;
    
21.         }
    
22.         task->AIO_flg=flg+1;
    
23.         task->fd=sock;
    
24.         task->timestamp=now_usec();
    
25.         pthread_mutex_lock(&task->lock);//防止其他线程提前闯入
    
26.         ret=do_epoll(task,0,flg);
    
27.         if(ret<0) {
    
28.                 pthread_mutex_unlock(&task->lock);
    
29.                 task->fd=save_fd;
    
30.                 task->AIO_flg=0;
    
31.                 if(save_timeo>-1) task->timeout=save_timeo;
    
32.                 return FORMATERR;
    
33.         }
    
34.         ret=swapcontext(&task->uc,&rs->tc);//这才是真正的yield
    
35.         pthread_mutex_unlock(&task->lock);
    
36.         task->fd=save_fd;
    
37.         task->AIO_flg=0;
    
38.         if(save_timeo>-1) task->timeout=save_timeo;
    
39.         if(flg==0 && task->events != EPOLLIN) {
    
40.                 ShowLog(1,"%s:tid=%lX resumed events=%X",__FUNCTION__,
    
41.                         pthread_self(),task->events);
    
42.                 return -11;
    
43.         }
    
44.         return ret;
    
45. }
    

复制代码

好了，所有的秘密都在这里了。你可以写自己的yield了吗？