epoll的兩種工做模式

epoll有兩種模式,Edge Triggered(簡稱ET) 和 Level Triggered(簡稱LT).在採用這兩種模式時要注意的是,假設採用ET模式,那麼僅當狀態發生變化時纔會通知,而採用LT模式相似於原來的select/poll操做,僅僅要還有沒有處理的事件就會一直通知. 

以代碼來講明問題: 
首先給出server的代碼,需要說明的是每次accept的鏈接,增長可讀集的時候採用的都是ET模式,而且接收緩衝區是5字節的,也就是每次僅僅接收5字節的數據: 

 #include <iostream>
 #include <sys/socket.h>
 #include <sys/epoll.h>
 #include <netinet/in.h>
 #include <arpa/inet.h>
 #include <fcntl.h>
 #include <unistd.h>
 #include <stdio.h>
 #include <errno.h>

 using namespace std;

 #define MAXLINE 5
 #define OPEN_MAX 100
 #define LISTENQ 20
 #define SERV_PORT 5000
 #define INFTIM 1000

 void setnonblocking(int sock)
 {
     int opts;
     opts=fcntl(sock,F_GETFL);
     if(opts<0)
     {
         perror("fcntl(sock,GETFL)");
         exit(1);
     }
     opts = opts|O_NONBLOCK;
     if(fcntl(sock,F_SETFL,opts)<0)
     {
         perror("fcntl(sock,SETFL,opts)");
         exit(1);
     }
 }

 int main()
 {
     int i, maxi, listenfd, connfd, sockfd,epfd,nfds;
     ssize_t n;
     char line[MAXLINE];
     socklen_t clilen;
     //聲明epoll_event結構體的變量,ev用於註冊事件,數組用於回傳要處理的事件
     struct epoll_event ev,events[20];
     //生成用於處理accept的epoll專用的文件描寫敘述符
     epfd=epoll_create(256);
     struct sockaddr_in clientaddr;
     struct sockaddr_in serveraddr;
     listenfd = socket(AF_INET, SOCK_STREAM, 0);
     //把socket設置爲非堵塞方式
     //setnonblocking(listenfd);
     //設置與要處理的事件相關的文件描寫敘述符
     ev.data.fd=listenfd;
     //設置要處理的事件類型
     ev.events=EPOLLIN|EPOLLET;
     //ev.events=EPOLLIN;
     //註冊epoll事件
     epoll_ctl(epfd,EPOLL_CTL_ADD,listenfd,&ev);
     bzero(&serveraddr, sizeof(serveraddr));
     serveraddr.sin_family = AF_INET;
     char *local_addr="127.0.0.1";
     inet_aton(local_addr,&(serveraddr.sin_addr));//htons(SERV_PORT);
     serveraddr.sin_port=htons(SERV_PORT);
     bind(listenfd,(sockaddr *)&serveraddr, sizeof(serveraddr));
     listen(listenfd, LISTENQ);
     maxi = 0;
     for ( ; ; ) {
         //等待epoll事件的發生
         nfds=epoll_wait(epfd,events,20,500);
         //處理所發生的所有事件
         for(i=0;i<nfds;++i)
         {
             if(events[i].data.fd==listenfd)
             {
                 connfd = accept(listenfd,(sockaddr *)&clientaddr, &clilen);
                 if(connfd<0){
                     perror("connfd<0");
                     exit(1);
                 }
                 //setnonblocking(connfd);
                 char *str = inet_ntoa(clientaddr.sin_addr);
                 cout << "accapt a connection from " << str << endl;
                 //設置用於讀操做的文件描寫敘述符
                 ev.data.fd=connfd;
                 //設置用於注測的讀操做事件
                 ev.events=EPOLLIN|EPOLLET;
                 //ev.events=EPOLLIN;
                 //註冊ev
                 epoll_ctl(epfd,EPOLL_CTL_ADD,connfd,&ev);
             }
             else if(events[i].events&EPOLLIN)
             {
                 cout << "EPOLLIN" << endl;
                 if ( (sockfd = events[i].data.fd) < 0)
                     continue;
                 if ( (n = read(sockfd, line, MAXLINE)) < 0) {
                     if (errno == ECONNRESET) {
                         close(sockfd);
                         events[i].data.fd = -1;
                     } else
                         std::cout<<"readline error"<<std::endl;
                 } else if (n == 0) {
                     close(sockfd);
                     events[i].data.fd = -1;
                 }
                 line[n] = '\0';
                 cout << "read " << line << endl;
                 //設置用於寫操做的文件描寫敘述符
                 ev.data.fd=sockfd;
                 //設置用於注測的寫操做事件
                 ev.events=EPOLLOUT|EPOLLET;
                 //改動sockfd上要處理的事件爲EPOLLOUT
                 //epoll_ctl(epfd,EPOLL_CTL_MOD,sockfd,&ev);
             }
             else if(events[i].events&EPOLLOUT)
             {
                 sockfd = events[i].data.fd;
                 write(sockfd, line, n);
                 //設置用於讀操做的文件描寫敘述符
                 ev.data.fd=sockfd;
                 //設置用於注測的讀操做事件
                 ev.events=EPOLLIN|EPOLLET;
                 //改動sockfd上要處理的事件爲EPOLIN
                 epoll_ctl(epfd,EPOLL_CTL_MOD,sockfd,&ev);
             }
         }
     }
     return 0;
 }

如下給出測試所用的Perl寫的client端,在client中發送10字節的數據,同一時候讓client在發送完數據以後進入死循環, 也就是在發送完以後鏈接的狀態不發生改變--既再也不發送數據, 也不關閉鏈接,這樣才幹觀察出server的狀態: 

 #!/usr/bin/perl

 use IO::Socket;

 my $host = "127.0.0.1";
 my $port = 5000;

 my $socket = IO::Socket::INET->new("$host:$port") or die "create socket error $@";
 my $msg_out = "1234567890";
 print $socket $msg_out;
 print "now send over, go to sleep \n";

 while (1)
 {
     sleep(1);
 }

執行server和client發現,server只讀取了5字節的數據,而client事實上發送了10字節的數據,也就是說,server僅當第一次監聽到了EPOLLIN事件,由於沒有讀取完數據,而且採用的是ET模式,狀態在此以後不發生變化,所以server再也接收不到EPOLLIN事件了. 
(友情提示:上面的這個測試client,當你關閉它的時候會再次出發IO可讀事件給server,此時server就會去讀取剩下的5字節數據了,但是這一事件與前面描寫敘述的ET性質並不矛盾.) 

假設咱們把client改成這樣: 

 #!/usr/bin/perl

 use IO::Socket;

 my $host = "127.0.0.1";
 my $port = 5000;

 my $socket = IO::Socket::INET->new("$host:$port") or die "create socket error $@";
 my $msg_out = "1234567890";
 print $socket $msg_out;
 print "now send over, go to sleep \n";
 sleep(5);
 print "5 second gone send another line\n";
 print $socket $msg_out;

 while (1)
 {
     sleep(1);
 }

可以發現,在server接收完5字節的數據以後一直監聽不到client的事件,而當client休眠5秒以後又一次發送數據,server再次監聽到了變化,僅僅只是因爲僅僅是讀取了5個字節,仍然有10個字節的數據(client第二次發送的數據)沒有接收完. 

假設上面的實驗中,對accept的socket都採用的是LT模式,那麼僅僅要還有數據留在buffer中,server就會繼續獲得通知,讀者可以自行修改代碼進行實驗. 

基於這兩個實驗,可以得出這種結論:ET模式僅當狀態發生變化的時候纔得到通知,這裏所謂的狀態的變化並不包含緩衝區中還有未處理的數據,也就是說,假設要採用ET模式,需要一直read/write直到出錯爲止,很是多人反映爲何採用ET模式僅僅接收了一部分數據就再也得不到通知了,大多因爲這樣;而LT模式是僅僅要有數據沒有處理就會一直通知下去的. 
補充說明一下這裏一直強調的"狀態變化"是什麼: 

1)對於監聽可讀事件時,假設是socket是監聽socket,那麼當有新的主動鏈接到來爲狀態發生變化;對通常的socket而言,協議棧中相應的緩衝區有新的數據爲狀態發生變化.但是,假設在一個時間同一時候接收了N個鏈接(N>1),但是監聽socket僅僅accept了一個鏈接,那麼其餘未 accept的鏈接將不會在ET模式下給監聽socket發出通知,此時狀態不發生變化;對於通常的socket,就如樣例中而言,假設相應的緩衝區自己已經有了N字節的數據,而僅僅取出了小於N字節的數據,那麼殘存的數據不會形成狀態發生變化. 

2)對於監聽可寫事件時,同理可推,再也不詳述. 

而不管是監聽可讀仍是可寫,對方關閉socket鏈接都將形成狀態發生變化,比方在樣例中,假設強行中斷client腳本,也就是主動中斷了socket鏈接,那麼都將形成server端發生狀態的變化,從而server獲得通知,將已經在本方緩衝區中的數據讀出. 

把前面的描寫敘述可以總結例如如下:僅當對方的動做(發出數據,關閉鏈接等)形成的事件才幹致使狀態發生變化,而本方協議棧中已經處理的事件(包含接收了對方的數據,接收了對方的主動鏈接請求)並不是形成狀態發生變化的必要條件,狀態變化必定是對方形成的.因此在ET模式下的,必須一直處理到出錯或者全然處理完成,才幹進行下一個動做,不然可能會錯誤發生. 

另外,從這個樣例中,也可以闡述一些主要的網絡編程概念.首先,鏈接的兩端中,一端發送成功並不表明着對方上層應用程序接收成功, 就拿上面的client測試程序來講,10字節的數據已經發送成功,但是上層的server並無調用read讀取數據,所以發送成功只說明瞭數據被對方的協議棧接收存放在了相應的buffer中,而上層的應用程序是否接收了這部分數據不得而知;相同的,讀取數據時也只表明着本方協議棧的相應buffer中有數據可讀,而此時時候在對端是否在發送數據也不得而知. 




epoll精髓 
在linux的網絡編程中，很是長的時間都在使用select來作事件觸發。在linux新的內核中，有了一種替換它的機制，就是epoll。

 
相比於select。epoll最大的優勢在於它不會隨着監聽fd數目的增加而減小效率。因爲在內核中的select實現中，它是採用輪詢來處理的，輪詢的fd數目越多，天然耗時越多。並且。在linux/posix_types.h頭文件有這種聲明： 
#define __FD_SETSIZE    1024 
表示select最多同一時候監聽1024個fd，固然，可以經過改動頭文件再重編譯內核來擴大這個數目，但這彷佛並不治本。 

epoll的接口很easy。一共就三個函數： 
1. int epoll_create(int size); 
建立一個epoll的句柄。size用來告訴內核這個監聽的數目一共同擁有多大。這個參數不一樣於select()中的第一個參數，給出最大監聽的fd+1的值。

需要注意的是，當建立好epoll句柄後。它就是會佔用一個fd值，在linux下假設查看/proc/進程id/fd/，是能夠看到這個fd的。因此在使用完epoll後，必須調用close()關閉，不然可能致使fd被耗盡。 


2. int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event); 
epoll的事件註冊函數，它不一樣與select()是在監聽事件時告訴內核要監聽什麼類型的事件，而是在這裏先註冊要監聽的事件類型。

第一個參數是epoll_create()的返回值。第二個參數表示動做，用三個宏來表示： 
EPOLL_CTL_ADD：註冊新的fd到epfd中； 
EPOLL_CTL_MOD：改動已經註冊的fd的監聽事件； 
EPOLL_CTL_DEL：從epfd中刪除一個fd； 
第三個參數是需要監聽的fd。第四個參數是告訴內核需要監聽什麼事，struct epoll_event結構例如如下： 

 struct epoll_event {
   __uint32_t events;  /* Epoll events */
   epoll_data_t data;  /* User data variable */
 };

events可以是下面幾個宏的集合： 
EPOLLIN ：表示相應的文件描寫敘述符可以讀（包含對端SOCKET正常關閉）； 
EPOLLOUT：表示相應的文件描寫敘述符可以寫； 
EPOLLPRI：表示相應的文件描寫敘述符有緊急的數據可讀（這裏應該表示有帶外數據到來）； 
EPOLLERR：表示相應的文件描寫敘述符錯誤發生； 
EPOLLHUP：表示相應的文件描寫敘述符被掛斷； 
EPOLLET： 將EPOLL設爲邊緣觸發(Edge Triggered)模式。這是相對於水平觸發(Level Triggered)來講的。 
EPOLLONESHOT：僅僅監聽一次事件。當監聽完此次事件以後，假設還需要繼續監聽這個socket的話，需要再次把這個socket增長到EPOLL隊列裏 


3. int epoll_wait(int epfd, struct epoll_event * events, int maxevents, int timeout); 
等待事件的產生，相似於select()調用。

參數events用來從內核獲得事件的集合，maxevents告以內核這個events有多大。這個maxevents的值不能大於建立epoll_create()時的size，參數timeout是超時時間（毫秒，0會立刻返回，-1將不肯定，也有說法說是永久堵塞）。該函數返回需要處理的事件數目。如返回0表示已超時。 

-------------------------------------------------------------------------------------------- 

從man手冊中。獲得ET和LT的詳細描寫敘述例如如下 

EPOLL事件有兩種模型： 
Edge Triggered (ET) 
Level Triggered (LT) 

假若有這樣一個樣例： 
1. 咱們已經把一個用來從管道中讀取數據的文件句柄(RFD)加入到epoll描寫敘述符 
2. 這個時候從管道的還有一端被寫入了2KB的數據 
3. 調用epoll_wait(2)，並且它會返回RFD，說明它已經準備好讀取操做 
4. 而後咱們讀取了1KB的數據 
5. 調用epoll_wait(2)...... 

Edge Triggered 工做模式： 
假設咱們在第1步將RFD加入到epoll描寫敘述符的時候使用了EPOLLET標誌。那麼在第5步調用epoll_wait(2)以後將有可能會掛起，因爲剩餘的數據還存在於文件的輸入緩衝區內。而且數據發出端還在等待一個針對已經發出數據的反饋信息。

僅僅有在監視的文件句柄上發生了某個事件的時候 ET 工做模式纔會彙報事件。所以在第5步的時候，調用者可能會放棄等待仍在存在於文件輸入緩衝區內的剩餘數據。在上面的樣例中，會有一個事件產生在RFD句柄上。因爲在第2步運行了一個寫操做，而後，事件將會在第3步被銷燬。

因爲第4步的讀取操做沒有讀空文件輸入緩衝區內的數據，所以咱們在第5步調用 epoll_wait(2)完畢後。是否掛起是不肯定的。

epoll工做在ET模式的時候，必須使用非堵塞套接口。以免由於一個文件句柄的堵塞讀/堵塞寫操做把處理多個文件描寫敘述符的任務餓死。最好以如下的方式調用ET模式的epoll接口，在後面會介紹避免可能的缺陷。 
   i    基於非堵塞文件句柄 
   ii   僅僅有當read(2)或者write(2)返回EAGAIN時才需要掛起。等待。

但這並不是說每次read()時都需要循環讀。直到讀到產生一個EAGAIN才以爲這次事件處理完畢。當read()返回的讀到的數據長度小於請求的數據長度時，就可以肯定此時緩衝中已沒有數據了。也就可以以爲此事讀事件已處理完畢。 

Level Triggered 工做模式 
相反的，以LT方式調用epoll接口的時候，它就至關於一個速度比較快的poll(2)，並且無論後面的數據是否被使用，所以他們具備相同的職能。因爲即便使用ET模式的epoll，在收到多個chunk的數據的時候仍然會產生多個事件。調用者可以設定EPOLLONESHOT標誌，在 epoll_wait(2)收到事件後epoll會與事件關聯的文件句柄從epoll描寫敘述符中禁止掉。所以當EPOLLONESHOT設定後，使用帶有 EPOLL_CTL_MOD標誌的epoll_ctl(2)處理文件句柄就成爲調用者必須做的事情。 


而後詳解ET, LT: 

LT(level triggered)是缺省的工做方式，並且同一時候支持block和no-block socket.在這樣的作法中，內核告訴你一個文件描寫敘述符是否就緒了，而後你可以對這個就緒的fd進行IO操做。假設你不做不論什麼操做，內核仍是會繼續通知你的。因此。這樣的模式編程出錯誤可能性要小一點。

傳統的select/poll都是這樣的模型的表明． 

ET(edge-triggered)是快速工做方式，僅僅支持no-block socket。在這樣的模式下，當描寫敘述符從未就緒變爲就緒時。內核經過epoll告訴你。而後它會若是你知道文件描寫敘述符已經就緒，並且不會再爲那個文件描寫敘述符發送不少其它的就緒通知，直到你作了某些操做致使那個文件描寫敘述符再也不爲就緒狀態了(比方，你在發送，接收或者接收請求，或者發送接收的數據少於必定量時致使了一個EWOULDBLOCK 錯誤）。

但是請注意。假設一直不正確這個fd做IO操做(從而致使它再次變成未就緒)，內核不會發送不少其它的通知(only once),只是在TCP協議中，ET模式的加速效用仍需要不少其它的benchmark確認（這句話不理解）。

 

在不少測試中咱們會看到假設沒有大量的idle -connection或者dead-connection。epoll的效率並不會比select/poll高很是多，但是當咱們遇到大量的idle- connection(好比WAN環境中存在大量的慢速鏈接)，就會發現epoll的效率大大高於select/poll。

（未測試） 



另外。當使用epoll的ET模型來工做時，當產生了一個EPOLLIN事件後， 
讀數據的時候需要考慮的是當recv()返回的大小假設等於請求的大小，那麼很是有多是緩衝區還有數據未讀完，也意味着該次事件尚未處理完，因此還需要再次讀取： 

 while(rs)
 {
   buflen = recv(activeevents[i].data.fd, buf, sizeof(buf), 0);
   if(buflen < 0)
   {
     // 由於是非堵塞的模式,因此當errno爲EAGAIN時,表示當前緩衝區已無數據可讀
     // 在這裏就看成是該次事件已處理處.
     if(errno == EAGAIN)
      break;
     else
      return;
    }
    else if(buflen == 0)
    {
      // 這裏表示對端的socket已正常關閉.
    }
    if(buflen == sizeof(buf)
      rs = 1;   // 需要再次讀取
    else
      rs = 0;
 }

還有，假如發送端流量大於接收端的流量(意思是epoll所在的程序讀比轉發的socket要快),由於是非堵塞的socket,那麼send()函數儘管返回,但實際緩衝區的數據並未真正發給接收端,這樣不斷的讀和發。當緩衝區滿後會產生EAGAIN錯誤(參考man send),同一時候,不理會此次請求發送的數據.因此,需要封裝socket_send()的函數用來處理這樣的狀況,該函數會盡可能將數據寫完再返回，返回-1表示出錯。在socket_send()內部,當寫緩衝已滿(send()返回-1,且errno爲EAGAIN),那麼會等待後再重試.這樣的方式並不很是完美,在理論上可能會長時間的堵塞在socket_send()內部,但暫沒有更好的辦法. 

 ssize_t socket_send(int sockfd, const char* buffer, size_t buflen)
 {
   ssize_t tmp;
   size_t total = buflen;
   const char *p = buffer;

   while(1)
   {
     tmp = send(sockfd, p, total, 0);
     if(tmp < 0)
     {
       // 當send收到信號時,可以繼續寫,但這裏返回-1.
       if(errno == EINTR)
         return -1;

       // 當socket是非堵塞時,如返回此錯誤,表示寫緩衝隊列已滿,
       // 在這裏作延時後再重試.
       if(errno == EAGAIN)
       {
         usleep(1000);
         continue;
       }

       return -1;
     }

     if((size_t)tmp == total)
       return buflen;

     total -= tmp;
     p += tmp;
   }

   return tmp;
 }

epoll爲何這麼快 

epoll是多路複用IO(I/O Multiplexing)中的一種方式,但是僅用於linux2.6以上內核,在開始討論這個問題以前,先來解釋一下爲何需要多路複用IO. 
以一個生活中的樣例來解釋. 
若是你在大學中讀書,要等待一個朋友來訪,而這個朋友僅僅知道你在A號樓,但是不知道你詳細住在哪裏,因而大家約好了在A號樓門口見面. 
假設你使用的堵塞IO模型來處理這個問題,那麼你就僅僅能一直守候在A號樓門口等待朋友的到來,在這段時間裏你不能作別的事情,不難知道,這樣的方式的效率是低下的. 
現在時代變化了,開始使用多路複用IO模型來處理這個問題.你告訴你的朋友來了A號樓找樓管大媽,讓她告訴你該怎麼走.這裏的樓管大媽扮演的就是多路複用IO的角色. 
進一步解釋select和epoll模型的差別. 
select版大媽作的是例如如下的事情:比方同窗甲的朋友來了,select版大媽比較笨,她帶着朋友挨個房間進行查詢誰是同窗甲,你等的朋友來了,因而在實際的代碼中,select版大媽作的是下面的事情: 

 int n = select(&readset,NULL,NULL,100);
 for (int i = 0; n > 0; ++i)
 {
    if (FD_ISSET(fdarray[i], &readset))
    {
       do_something(fdarray[i]);
       --n;
    }
 }

epoll版大媽就比較先進了,她記下了同窗甲的信息,比方說他的房間號,那麼等同窗甲的朋友到來時,僅僅需要告訴該朋友同窗甲在哪一個房間就能夠,不用本身親自帶着人滿大樓的找人了.因而epoll版大媽作的事情可以用例如如下的代碼表示: 

 n=epoll_wait(epfd,events,20,500);
 for(i=0;i<n;++i)
 {
     do_something(events[n]);
 }
 在epoll中,重要的做用結構epoll_event定義例如如下:
 typedef union epoll_data {
      void *ptr;
      int fd;
      __uint32_t u32;
      __uint64_t u64;
 } epoll_data_t;
 struct epoll_event {
                 __uint32_t events;      /* Epoll events */
                 epoll_data_t data;      /* User data variable */
 };

可以看到,epoll_data是一個union結構體,它就是epoll版大媽用於保存同窗信息的結構體,它可以保存很是多類型的信息:fd,指針,等等.有了這個結構體,epoll大媽可以不用吹灰之力就可以定位到同窗甲. 
別小看了這些效率的提升,在一個大規模併發的server中,輪詢IO是最耗時間的操做之中的一個.再回到那個樣例中,假設每到來一個朋友樓管大媽都要全樓的查詢同窗,那麼處理的效率一定就低下了,過不久樓底就有很多的人了. 
對照最先給出的堵塞IO的處理模型, 可以看到採用了多路複用IO以後, 程序可以自由的進行本身除了IO操做以外的工做, 僅僅有到IO狀態發生變化的時候由多路複用IO進行通知, 而後再採取對應的操做, 而不用一直堵塞等待IO狀態發生變化了. 
從上面的分析也可以看出,epoll比select的提升其實是一個用空間換時間思想的詳細應用. 

多進程server中,epoll的建立應該在建立子進程以後 

看個人測試代碼,彷佛應該是在建立子進程以後建立epoll的fd,不然程序將會有問題,試將代碼中兩個CreateWorker函數的調用位置分別調用,一個在建立epoll fd以前,一個在以後,在調用在建立以前的代碼會出問題,在個人機器上(linux內核2.6.26)表現的症狀就是所有進程的epoll_wait函數返回0, 而client彷佛被堵塞了: 
server端: 

 #include <iostream>
 #include <sys/socket.h>
 #include <sys/epoll.h>
 #include <netinet/in.h>
 #include <arpa/inet.h>
 #include <fcntl.h>
 #include <unistd.h>
 #include <stdio.h>
 #include <errno.h>
 #include <sys/types.h>
 #include <sys/wait.h>

 using namespace std;

 #define MAXLINE 5
 #define OPEN_MAX 100
 #define LISTENQ 20
 #define SERV_PORT 5000
 #define INFTIM 1000

 typedef struct task_t
 {
     int fd;
     char buffer[100];
     int n;
 }task_t;

 int CreateWorker(int nWorker)
 {
     if (0 < nWorker)
     {
         bool bIsChild;
         pid_t nPid;

         while (!bIsChild)
         {
             if (0 < nWorker)
             {
                 nPid = ::fork();
                 if (nPid > 0)
                 {
                     bIsChild = false;
                     --nWorker;
                 }
                 else if (0 == nPid)
                 {
                     bIsChild = true;
                     printf("create worker %d success!\n", ::getpid());
                 }
                 else
                 {
                     printf("fork error: %s\n", ::strerror(errno));
                     return -1;
                 }
             }
             else
             {
                 int nStatus;
                 if (-1 == ::wait(&nStatus))
                 {
                     ++nWorker;
                 }
             }
         }
     }

     return 0;
 }

 void setnonblocking(int sock)
 {
     int opts;
     opts=fcntl(sock,F_GETFL);
     if(opts<0)
     {
         perror("fcntl(sock,GETFL)");
         exit(1);
     }
     opts = opts|O_NONBLOCK;
     if(fcntl(sock,F_SETFL,opts)<0)
     {
         perror("fcntl(sock,SETFL,opts)");
         exit(1);
     }
 }

 int main()
 {
     int i, maxi, listenfd, connfd, sockfd,epfd,nfds;
     ssize_t n;
     char line[MAXLINE];
     socklen_t clilen;
     struct epoll_event ev,events[20];

     struct sockaddr_in clientaddr;
     struct sockaddr_in serveraddr;
     listenfd = socket(AF_INET, SOCK_STREAM, 0);
        bzero(&serveraddr, sizeof(serveraddr));
     serveraddr.sin_family = AF_INET;
     char *local_addr="127.0.0.1";
     inet_aton(local_addr,&(serveraddr.sin_addr));//htons(SERV_PORT);
     serveraddr.sin_port=htons(SERV_PORT);
       // 地址重用
     int nOptVal = 1;
     socklen_t nOptLen = sizeof(int);
     if (-1 == ::setsockopt(listenfd, SOL_SOCKET, SO_REUSEADDR, &nOptVal, nOptLen))
     {
         return -1;
     }
     setnonblocking(listenfd);
     bind(listenfd,(sockaddr *)&serveraddr, sizeof(serveraddr));
     listen(listenfd, LISTENQ);

     CreateWorker(5);

     //把socket設置爲非堵塞方式

     //生成用於處理accept的epoll專用的文件描寫敘述符
     epfd=epoll_create(256);
     //設置與要處理的事件相關的文件描寫敘述符
     ev.data.fd=listenfd;
     //設置要處理的事件類型
     ev.events=EPOLLIN|EPOLLET;
     //ev.events=EPOLLIN;
     //註冊epoll事件
     epoll_ctl(epfd,EPOLL_CTL_ADD,listenfd,&ev);

      //CreateWorker(5);

     maxi = 0;

     task_t task;
     task_t *ptask;
     while(true)
     {
         //等待epoll事件的發生
         nfds=epoll_wait(epfd,events,20,500);
         //處理所發生的所有事件
         for(i=0;i<nfds;++i)
         {
             if(events[i].data.fd==listenfd)
             {
                 connfd = accept(listenfd,NULL, NULL);
                 if(connfd<0){
                     printf("connfd<0, listenfd = %d\n", listenfd);
                     printf("error = %s\n", strerror(errno));
                     exit(1);
                 }
                 setnonblocking(connfd);

                 //設置用於讀操做的文件描寫敘述符
                 memset(&task, 0, sizeof(task));
                 task.fd = connfd;
                 ev.data.ptr = &task;
                 //設置用於註冊的讀操做事件
                 ev.events=EPOLLIN|EPOLLET;
                 //ev.events=EPOLLIN;
                 //註冊ev
                 epoll_ctl(epfd,EPOLL_CTL_ADD,connfd,&ev);
             }
             else if(events[i].events&EPOLLIN)
             {
                 cout << "EPOLLIN" << endl;
                 ptask = (task_t*)events[i].data.ptr;
                 sockfd = ptask->fd;

                 if ( (ptask->n = read(sockfd, ptask->buffer, 100)) < 0) {
                     if (errno == ECONNRESET) {
                         close(sockfd);
                         events[i].data.ptr = NULL;
                     } else
                         std::cout<<"readline error"<<std::endl;
                 } else if (ptask->n == 0) {
                     close(sockfd);
                     events[i].data.ptr = NULL;
                 }
                 ptask->buffer[ptask->n] = '\0';
                 cout << "read " << ptask->buffer << endl;

                 //設置用於寫操做的文件描寫敘述符
                 ev.data.ptr = ptask;
                 //設置用於注測的寫操做事件
                 ev.events=EPOLLOUT|EPOLLET;

                 //改動sockfd上要處理的事件爲EPOLLOUT
                 epoll_ctl(epfd,EPOLL_CTL_MOD,sockfd,&ev);
             }
             else if(events[i].events&EPOLLOUT)
             {
                 cout << "EPOLLOUT" << endl;
                 ptask = (task_t*)events[i].data.ptr;
                 sockfd = ptask->fd;

                 write(sockfd, ptask->buffer, ptask->n);

                 //設置用於讀操做的文件描寫敘述符
                 ev.data.ptr = ptask;

                 //改動sockfd上要處理的事件爲EPOLIN
                 epoll_ctl(epfd,EPOLL_CTL_DEL,sockfd,&ev);
                 cout << "write " << ptask->buffer;
                 memset(ptask, 0, sizeof(*ptask));
                 close(sockfd);
             }
         }
     }
     return 0;
 }

測試client: 
#!/usr/bin/perl 

use strict; 
use Socket; 
use IO::Handle; 

sub echoclient 
{ 
    my $host = "127.0.0.1"; 
    my $port = 5000; 

    my $protocol = getprotobyname("TCP"); 
    $host = inet_aton($host); 

    socket(SOCK, AF_INET, SOCK_STREAM, $protocol) or die "socket() failed: $!"; 

    my $dest_addr = sockaddr_in($port, $host); 
    connect(SOCK, $dest_addr) or die "connect() failed: $!"; 

    SOCK->autoflush(1); 

    my $msg_out = "hello world\n"; 
    print "out = ", $msg_out; 
    print SOCK $msg_out; 
    my $msg_in = <SOCK>; 
    print "in = ", $msg_in; 

    close SOCK; 
} 

#&echoclient; 
#exit(0); 

for (my $i = 0; $i < 9999; $i++) 
{ 
    echoclient; 
} 
我查看了lighttpd的實現,也是在建立完子進程以後才建立的epoll的fd. 
請問誰知道哪裏有解說這個的文檔? 
假如fd1是由A進程增長epfd的，而且用的是ET模式，那麼增長通知的是進程B，顯然B進程不會對fd1進行處理。因此之後fd1的事件再不會通知。因此 通過幾回循環以後，所有的fd都沒有事件通知了。因此epoll_wait在timeout以後就返回0了。

而在client的結果可想而知。僅僅能是被堵塞。 
也就是說, 這是一種發生在epoll fd上面的相似於"驚羣"的現象. 
對於linux socket與epoll配合相關的一些心得記錄 

沒有多少高深的東西。全當記錄，儘管簡單。但是沒有作過測試仍是挺easy讓人糊塗的 

     int nRecvBuf=32*1024;//設置爲32K 
setsockopt(s,SOL_SOCKET,SO_RCVBUF,(const char*)&nRecvBuf,sizeof(int)); 
一、經過上面語句可以簡單設置緩衝區大小,測試證實：跟epoll結合的時候僅僅有當單次發送的數據全被從緩衝區讀完成以後纔會再次被觸發。屢次發送數據假設沒有讀取完成當緩衝區未滿的時候數據不會丟失，會累加到後面。 
二、 假設緩衝區未滿。同一鏈接屢次發送數據會屢次收到EPOLLIN事件。 
單次發送數據>socket緩衝區大小的數據數據會被堵塞分次發送，因此循環接收可以用ENLIGE錯誤推斷。 
   三、假設緩衝區滿。新發送的數據不會觸發epoll事件（也無異常）。每次recv都會爲緩衝區騰出空間，僅僅有當緩衝區空暇大小能夠再次接收數據epollIN事件能夠再次被觸發 
接收時接收大小爲0表示client斷開（不可能有0數據包觸發EPOLLIN）,-1表示異常。針對errorno進行推斷可以肯定是合理異常仍是需要終止的異常，>0而不等於緩衝區大小表示單次發送結束。 
   四、 假設中途暫時調整接收緩存區大小。並且在上一次中數據沒有全然接收到用戶空間，數據不會丟失。會累加在一塊兒 

因此總結起來，系統對於數據的完整性仍是作了至關的保正，至於穩定性沒有做更深一步的測試 

   新添加： 
   五、假設主accept監聽的soctet fd也設置爲非堵塞，那麼單純靠epoll事件來驅動的服務器模型會存在問題，併發壓力下發現。每次accept僅僅從系統中取得第一個。因此假設恰馮多個鏈接同一時候觸發server fd的EPOLLIN事件，在返回的event數組中體現不出來，會出現丟失事件的現象，因此當用ab等工具簡單的壓載就會發現每次都會有最後幾條信息得不處處理，緣由就在於此，我現在的解決的方法是將server fd的監聽去掉。用一個線程堵塞監聽。accept成功就處理檢測client fd，而後在主線程循環監聽client事件。這樣epoll在邊緣模式下出錯的機率就小，測試代表效果明顯 
六、對於SIG部分信號仍是要作屏蔽處理，否則對方socket中斷等正常事件都會引發整個服務的退出 
七、sendfile(fd, f->SL->sendBuffer.inFd, (off_t *)&f->SL->sendBuffer.offset, size_need);注意sendfile函數的地三個變量是傳送地址，偏移量會本身主動添加。不需要手動再次添加。不然就會出現文件傳送丟失現象 
八、單線程epoll驅動模型誤解：曾經我一直以爲單線程是沒法處理webserver這種有嚴重網絡延遲的服務，但nginx等優秀server都是機遇事件驅動模型，開始我在些的時候也是操心這些問題，後來測試發現。當client socket設爲非堵塞模式的時候，從讀取數據到解析http協議，到發送數據均在epoll的驅動下速度很快，沒有必要採用多線程，個人單核cpu （奔三）就可以達到10000page／second，這在公網上是遠遠沒法達到的一個數字（網絡延遲更爲嚴重）。因此單線程的數據處理能力已經很是高了。就不需要多線程了，所不一樣的是你在架構server的時候需要將所有堵塞的部分拆分開來。當epoll通知你可以讀取的時候，實際上部分數據已經到了 socket緩衝區。你所讀取用的事件是將數據從內核空間複製到用戶空間，同理，寫也是同樣的，因此epoll重要的地方就是將這兩個延時的部分作了相似的異步處理，假設不需要處理更爲複雜的業務，那單線程足以知足1000M網卡的最高要求，這纔是單線程的意義。 
    我曾經構建的webserver就沒有理解epoll，採用epoll的邊緣觸發以後怕事件丟失，或者單線程處理堵塞，因此本身用多線程構建了一個任務調度器。所有收到的事件通通壓進任無調度器中，而後多任務處理。我還將read和write分別用兩個調度器處理。並打算假設中間需要特殊的耗時的處理就添加一套調度器，用少許線程＋epoll的方法來題高性能，後來發現read和write部分調度器是多餘的。epoll原本就是一個事件調度器，在後面再次緩存事件分部處理還不如將epoll設爲水平模式，因此畫蛇添足。但是這個調度起仍是實用處的 
   上面講到假設中間有耗時的工做。比方數據庫讀寫，外部資源請求（文件，socket）等這些操做就不能堵塞在主線程裏面。因此我設計的這個任務調度器就實用了，在epoll能處理的事件驅動部分就借用epoll的。中間部分採用模塊化的設計，用函數指針達到面相對象語言中的「託付」的做用，就可以知足不一樣的需要將任務（fd標識）增長調度器。讓多線程循環運行。假設中間再次遇到堵塞就會再次增長本身定義的堵塞器，檢測完畢就增長再次存入調度器，這樣就可以將多種複雜的任務劃分開來，至關於在處理的中間環節在本身購置一個相似於epoll的事件驅動器 
    九、多系統兼容：我現在卻是認爲與其構建一個多操做系統都支持的server不如構建特定系統的，假設想遷移再次修改，因爲一旦兼顧到多個系統的化會大大添加系統的複雜度，並且不能最優性能，每個系統都有本身的獨有的優化選項。因此我認爲遷移的工做量遠遠小於兼顧的工做量 
10模塊化編程，儘管用c仍是要講求一些模塊化的設計的。我現在才發現差點兒面相對想的語言所能實現的所有高級特性在c裏面差點兒都有相應的解決的方法（臨時發現除了操做符重載），所有學太高級面相對象的語言的朋友不放把模式用c來實現，也是一種樂趣，便於維護和本身閱讀 
   十一、養成凝視的好習慣