事件響應(yīng)方式

yulihua49

不能發(fā)代碼，發(fā)了就沒了。
/*******************************************
* 需要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++實現(xiàn)的協(xié)程我是無愛的。。。

yulihua49

發(fā)個代碼還等一天審核，還不告知，以為丟了呢，沒興趣了。

yulihua49

cokeboL 發(fā)表于 2016-08-16 14:50
原來這幾個是新API，我*nix c 還停留在apue unp范圍。

c/c++實現(xiàn)的協(xié)程我是無愛的。。。

c的協(xié)程，實現(xiàn)機(jī)制和resume條件都是明確的。epoll有響應(yīng)了就resume。別的語言，暈。就一個yield，啥時候能resume不知道。一般是輪到你了就resume。這不符合要求。
一般協(xié)程是配合AIO／NIO的，如果干的好好的為什么要yield呢？一定是等什么事件，干不下去了才需要yield。這個需求，其他語言是怎么實現(xiàn)的？

cokeboL

回復(fù) 15# yulihua49


我最初看lua的協(xié)程就不喜歡，因為要手動yield resume，而且lua又難利用多核，所以連深入研究的興趣都沒有
我不喜歡c的協(xié)程，也是因為手動檔，這種跟golang erlang比，寫法太麻煩太弱雞了，erlang golang的可以看
成是線程那樣，調(diào)度是應(yīng)用層實現(xiàn)也不需要自己操心，清爽干凈得多。。。

cokeboL

golang一個簡單的轉(zhuǎn)發(fā)網(wǎng)關(guān)

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. }
    

復(fù)制代碼

yulihua49

cokeboL 發(fā)表于 2016-08-18 10:45
golang一個簡單的轉(zhuǎn)發(fā)網(wǎng)關(guān)

不懂golang。
沒看見多線程。
err = clientConn.Read(buf)－－－－－－－－這個需要yield，怎樣實現(xiàn)的？沒有yield就成了1:1的事件等待。

cokeboL

go c2sCor()
go s2cCor()
這兩句就是啟動兩個協(xié)程，然后
err = clientConn.Read(buf)
nwrite, err = serverConn.Write(buf[:nread])
這兩句就都是阻塞了，不需要異步
神清氣爽的趕腳

yulihua49

cokeboL 發(fā)表于 2016-08-18 11:38
go c2sCor()
go s2cCor()
這兩句就是啟動兩個協(xié)程，然后

啟動兩個協(xié)程，而不是線程？那么沒有并行功能，無法充分利用多核。
我希望看到多線程協(xié)程，能夠并行，也能把任務(wù)在多個線程間切換。要看看你是怎么管理協(xié)程與線程的關(guān)系。

少線程服務(wù)器  －－－ 其實還是多線程服務(wù)器，就是線程數(shù)不多，大約＝核數(shù)。要充分利用所有的線程，一旦發(fā)生等待就yield，放棄這個任務(wù)趕緊接別的任務(wù)。

我那個例子只是應(yīng)用側(cè)代碼，很簡單，而且是同步異步自適應(yīng)的。就是：同步框架可以調(diào)用它，異步框架也可以調(diào)用它，只是需要事先提供yield函數(shù)即可。

多線程與多協(xié)程的關(guān)系在于yield函數(shù)內(nèi)的代碼，作為C，當(dāng)然是非常的龐雜，但是完全可以實現(xiàn)上述需求。

你是每個連接產(chǎn)生3個協(xié)程？如果10000個連接來了，就是30000個協(xié)程？30000個棧？每個棧就算1M，所需內(nèi)存？
在C里也是這個問題，我用棧池來解決，就是棧數(shù)＝線程數(shù)，激活的協(xié)程才擁有棧。