事件响应方式

yulihua49

不能发代码，发了就没了。
/*******************************************
* 需要linux 2.6.22 以上版本
* 和libaio 3.107以上版本
* 如果不具备这个条件，在makefile里用SIO_fd.o
* 取代本模块
*******************************************/

#include <unistd.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <sys/eventfd.h>
#include <libaio.h>

#include <scsrv.h>

static int AIO_oper(int fd,char *buff,size_t iosize,int flg)
{
io_context_t myctx;
int rc,num;
uint64_t finished_aio;
struct iocb _iocb,*io=&_iocb;
struct io_event event;
int     efd = eventfd(0, 0);
T_YIELD yield=get_yield();

        if (efd == -1) {   
                return flg?write(fd,buff,iosize):read(fd,buff,iosize);
        }   

        memset(&myctx,0,sizeof(myctx));
        io_set_eventfd(io,efd);
        io_queue_init(1, &myctx);
        if(flg) io_prep_pread(io, fd, buff, iosize, 0);
        else    io_prep_pwrite(io, fd, buff, iosize, 0);
        rc = io_submit(myctx, 1, &io);
        if(rc<0) {
                close(efd);
                io_destroy(myctx);
                return flg?write(fd,buff,iosize):read(fd,buff,iosize);
        }
        if(yield) {
                rc = yield(efd,0,0);//efd提交给epoll，并yield and resume.
                if(rc==0) eventfd_read(efd, &finished_aio);
        }
        close(efd);
        num = io_getevents(myctx, 1, 1, &event, NULL);
        if(num>0) {
                if(event.res2==0) num=event.res;
                else num=-1;
        }
        io_destroy(myctx);
        return num;
}

int AIO_read(int fd,char *buff,size_t iosize)
{
        return AIO_oper(fd,buff,iosize,0);
}

int AIO_write(int fd,char *buff,size_t iosize)
{
        return AIO_oper(fd,buff,iosize,1);
}

cokeboL

eventfd看了下，类似pipe

cokeboL

原来这几个是新API，我*nix c 还停留在apue unp范围。

c/c++实现的协程我是无爱的。。。

yulihua49

发个代码还等一天审核，还不告知，以为丢了呢，没兴趣了。

yulihua49

cokeboL 发表于 2016-08-16 14:50
原来这几个是新API，我*nix c 还停留在apue unp范围。

c/c++实现的协程我是无爱的。。。

c的协程，实现机制和resume条件都是明确的。epoll有响应了就resume。别的语言，晕。就一个yield，啥时候能resume不知道。一般是轮到你了就resume。这不符合要求。
一般协程是配合AIO／NIO的，如果干的好好的为什么要yield呢？一定是等什么事件，干不下去了才需要yield。这个需求，其他语言是怎么实现的？

cokeboL

回复 15# yulihua49


我最初看lua的协程就不喜欢，因为要手动yield resume，而且lua又难利用多核，所以连深入研究的兴趣都没有
我不喜欢c的协程，也是因为手动档，这种跟golang erlang比，写法太麻烦太弱鸡了，erlang golang的可以看
成是线程那样，调度是应用层实现也不需要自己操心，清爽干净得多。。。

cokeboL

golang一个简单的转发网关

1. 
   
2. import (
   
3.         "fmt"
   
4.         "net"
   
5. )
   
6. 
   
7. func newAgentConn(clientConn *net.TCPConn, serverAddr string) {
   
8.         serverConn, dailErr := net.Dial("tcp", serverAddr)
   
9. 
   
10.         if dailErr == nil {
    
11.                 c2sCor := func() {
    
12.                         defer func() {
    
13.                                 _ = recover()
    
14.                         }()
    
15. 
    
16.                         var nread int
    
17.                         var nwrite int
    
18.                         var err error
    
19.                         var buf = make([]byte, 1024)
    
20.                         for {
    
21.                                 nread, err = clientConn.Read(buf)
    
22.                                 if err != nil {
    
23.                                         clientConn.Close()
    
24.                                         serverConn.Close()
    
25.                                         break
    
26.                                 }
    
27. 
    
28.                                 nwrite, err = serverConn.Write(buf[:nread])
    
29.                                 if nwrite != nread || err != nil {
    
30.                                         clientConn.Close()
    
31.                                         serverConn.Close()
    
32.                                         break
    
33.                                 }
    
34.                         }
    
35.                 }
    
36. 
    
37.                 s2cCor := func() {
    
38.                         defer func() {
    
39.                                 _ = recover()
    
40.                         }()
    
41. 
    
42.                         var nread int
    
43.                         var nwrite int
    
44.                         var err error
    
45.                         var buf = make([]byte, 1024)
    
46.                         for {
    
47.                                 nread, err = serverConn.Read(buf)
    
48.                                 if err != nil {
    
49.                                         clientConn.Close()
    
50.                                         serverConn.Close()
    
51.                                         break
    
52.                                 }
    
53. 
    
54.                                 nwrite, err = clientConn.Write(buf[:nread])
    
55.                                 if nwrite != nread || err != nil {
    
56.                                         clientConn.Close()
    
57.                                         serverConn.Close()
    
58.                                         break
    
59.                                 }
    
60.                         }
    
61.                 }
    
62. 
    
63.                 go c2sCor()
    
64.                 go s2cCor()
    
65.         } else {
    
66.                 clientConn.Close()
    
67.         }
    
68. }
    
69. 
    
70. func RunAgent(agentAddr string, serverAddr string) {
    
71.         tcpAddr, err := net.ResolveTCPAddr("tcp4", agentAddr)
    
72.         if err != nil {
    
73.                 fmt.Println("ResolveTCPAddr Error: ", err)
    
74.                 return
    
75.         }
    
76. 
    
77.         listener, err2 := net.ListenTCP("tcp", tcpAddr)
    
78.         if err2 != nil {
    
79.                 fmt.Println("ListenTCP Error: ", err2)
    
80.                 return
    
81.         }
    
82. 
    
83.         defer listener.Close()
    
84. 
    
85.         fmt.Println(fmt.Sprintf("Agent Start Running on: Agent(%s) -> Server(%s)!", agentAddr, serverAddr))
    
86.         for {
    
87.                 conn, err := listener.AcceptTCP()
    
88. 
    
89.                 if err != nil {
    
90.                         fmt.Println("AcceptTCP Error: ", err2)
    
91.                 } else {
    
92.                         go newAgentConn(conn, serverAddr)
    
93.                 }
    
94.         }
    
95. }
    
96. 
    
97. func TestAgent() {
    
98.         Run("127.0.0.1:8888", "127.0.0.1:9999")
    
99. }
    

复制代码

yulihua49

cokeboL 发表于 2016-08-18 10:45
golang一个简单的转发网关

不懂golang。
没看见多线程。
err = clientConn.Read(buf)－－－－－－－－这个需要yield，怎样实现的？没有yield就成了1:1的事件等待。

cokeboL

go c2sCor()
go s2cCor()
这两句就是启动两个协程，然后
err = clientConn.Read(buf)
nwrite, err = serverConn.Write(buf[:nread])
这两句就都是阻塞了，不需要异步
神清气爽的赶脚

yulihua49

cokeboL 发表于 2016-08-18 11:38
go c2sCor()
go s2cCor()
这两句就是启动两个协程，然后

启动两个协程，而不是线程？那么没有并行功能，无法充分利用多核。
我希望看到多线程协程，能够并行，也能把任务在多个线程间切换。要看看你是怎么管理协程与线程的关系。

少线程服务器  －－－ 其实还是多线程服务器，就是线程数不多，大约＝核数。要充分利用所有的线程，一旦发生等待就yield，放弃这个任务赶紧接别的任务。

我那个例子只是应用侧代码，很简单，而且是同步异步自适应的。就是：同步框架可以调用它，异步框架也可以调用它，只是需要事先提供yield函数即可。

多线程与多协程的关系在于yield函数内的代码，作为C，当然是非常的庞杂，但是完全可以实现上述需求。

你是每个连接产生3个协程？如果10000个连接来了，就是30000个协程？30000个棧？每个棧就算1M，所需内存？
在C里也是这个问题，我用棧池来解决，就是栈数＝线程数，激活的协程才拥有棧。