轉載地址： http://blog.csdn.net/segments/article/details/6894189

要想完全理解I/O多路複用，需先要了解I/O模型：

一、五種I/O模型

1、阻塞I/O模型

     最流行的I/O模型是阻塞I/O模型，預設情形下，所有套介面都是阻塞的。我們以資料報套介面為例來講解此模型（我們使用UDP而不是TCP作為例子的原因在於就UDP而言，資料準備好讀取的概念比較簡單：要麼整個資料報已經收到，要麼還沒有。然而對於TCP來說，諸如套介面低潮標記等額外變數開始活動，導致這個概念變得複雜）。

     程序呼叫recvfrom，其系統呼叫直到資料報到達且被拷貝到應用程序的緩衝區中或者發生錯誤才返回，期間一直在等待。我們就說程序在從呼叫recvfrom開始到它返回的整段時間內是被阻塞的。

2、非阻塞I/O模型

      程序把一個套介面設定成非阻塞是在通知核心：當所請求的I/O操作非得把本程序投入睡眠才能完成時，不要把本程序投入睡眠，而是返回一個錯誤。也就是說當資料沒有到達時並不等待，而是以一個錯誤返回。

3、I/O複用模型

     呼叫select或poll，在這兩個系統呼叫中的某一個上阻塞，而不是阻塞於真正I/O系統呼叫。 阻塞於select呼叫，等待資料報套介面可讀。當select返回套介面可讀條件時，呼叫recevfrom將資料報拷貝到應用緩衝區中。

4、訊號驅動I/O模型

     首先開啟套介面訊號驅動I/O功能,並通過系統呼叫sigaction安裝一個訊號處理函式（此係統呼叫立即返回，程序繼續工作，它是非阻塞的）。當資料報準備好被讀時，就為該程序生成一個SIGIO訊號。隨即可以在訊號處理程式中呼叫recvfrom來讀資料報，井通知主迴圈資料已準備好被處理中。也可以通知主迴圈，讓它來讀資料報。

5、非同步I/O模型

     告知核心啟動某個操作，並讓核心在整個操作完成後(包括將資料從核心拷貝到使用者自己的緩衝區)通知我們。這種模型與訊號驅動模型的主要區別是：
           訊號驅動I/O：由核心通知我們何時可以啟動一個I/O操作，
           非同步I/O模型：由核心通知我們I/O操作何時完成。

二、I/O複用的典型應用場合：

 1、當客戶處理多個描述字（通常是互動式輸入和網路套介面）時，必須使用I/O複用。

 2、如果一個伺服器要處理多個服務或者多個協議（例如既要處理TCP，又要處理UDP），一般就要使用I/O複用。

三、支援I/O複用的系統呼叫

     目前支援I/O複用的系統呼叫有select、pselect、poll、epoll：

1、select函式

select系統呼叫是用來讓我們的程式監視多個檔案控制代碼(file descrīptor)的狀態變化的。程式會停在select這裡等待，直到被監視的檔案控制代碼有某一個或多個發生了狀態改變。

檔案在控制代碼在Linux裡很多，如果你man某個函式，在函式返回值部分說到成功後有一個檔案控制代碼被建立的都是的，如man socket可以看到“On success, afile descrīptor for the new socket is returned.”而man 2 open可以看到“open()and creat() return the new filedescrīptor”，其實檔案控制代碼就是一個整數，看socket函式的宣告就明白了：
int socket(int domain, int type, int protocol);
當然，我們最熟悉的控制代碼是0、1、2三個，0是標準輸入，1是標準輸出，2是標準錯誤輸出。0、1、2是整數表示的，對應的FILE *結構的表示就是stdin、stdout、stderr，0就是stdin，1就是stdout，2就是stderr。
比如下面這兩段程式碼都是從標準輸入讀入9個位元組字元：

#include  #include  #include  int main(int argc, char ** argv) {         char buf[10] = "";         read(0, buf, 9); /* 從標準輸入 0 讀入字元 */         fprintf(stdout, "%s\n", buf); /* 向標準輸出 stdout 寫字元 */         return 0; } /* **上面和下面的程式碼都可以用來從標準輸入讀使用者輸入的9個字元** */ #include  #include  #include  int main(int argc, char ** argv) {         char buf[10] = "";         fread(buf, 9, 1, stdin); /* 從標準輸入 stdin 讀入字元 */         write(1, buf, strlen(buf));         return 0; }

繼續上面說的select，就是用來監視某個或某些控制代碼的狀態變化的。select函式原型如下：
int select(int nfds, fd_set *readfds, fd_set *writefds, fd_set *exceptfds, struct timeval *timeout);
函式的最後一個引數timeout顯然是一個超時時間值，其型別是struct timeval *，即一個structtimeval結構的變數的指標，所以我們在程式裡要申明一個struct timevaltv;然後把變數tv的地址&tv傳遞給select函式。struct timeval結構如下：

struct timeval {              long    tv_sec;         /* seconds */              long    tv_usec;        /* microseconds */          };

第2、3、4三個引數是一樣的型別： fd_set *，即我們在程式裡要申明幾個fd_set型別的變數，比如rdfds, wtfds,exfds，然後把這個變數的地址&rdfds, &wtfds, &exfds傳遞給select函式。這三個引數都是一個控制代碼的集合，第一個rdfds是用來儲存這樣的控制代碼的：當控制代碼的狀態變成可讀的時系統就會告訴select函式返回，同理第二個wtfds是指有控制代碼狀態變成可寫的時系統就會告訴select函式返回，同理第三個引數exfds是特殊情況，即控制代碼上有特殊情況發生時系統會告訴select函式返回。特殊情況比如對方通過一個socket控制代碼發來了緊急資料。如果我們程式裡只想檢測某個socket是否有資料可讀，我們可以這樣：

fd_set rdfds; /* 先申明一個 fd_set 集合來儲存我們要檢測的 socket控制代碼 */ struct timeval tv; /* 申明一個時間變數來儲存時間 */ int ret; /* 儲存返回值 */ FD_ZERO(&rdfds); /* 用select函式之前先把集合清零 */ FD_SET(socket, &rdfds); /* 把要檢測的控制代碼socket加入到集合裡 */ tv.tv_sec = 1; tv.tv_usec = 500; /* 設定select等待的最大時間為1秒加500毫秒 */ ret = select(socket + 1, &rdfds, NULL, NULL, &tv); /* 檢測我們上面設定到集合rdfds裡的控制代碼是否有可讀資訊 */ if(ret < 0) perror("select");/* 這說明select函數出錯 */ else if(ret == 0) printf("超時\n"); /* 說明在我們設定的時間值1秒加500毫秒的時間內，socket的狀態沒有發生變化 */ else { /* 說明等待時間還未到1秒加500毫秒，socket的狀態發生了變化 */     printf("ret=%d\n", ret); /*ret這個返回值記錄了發生狀態變化的控制代碼的數目，由於我們只監視了socket這一個控制代碼，所以這裡一定ret=1，如果同時有多個控制代碼發生變化返回的就是控制代碼的總和了 */     /* 這裡我們就應該從socket這個控制代碼裡讀取資料了，因為select函式已經告訴我們這個控制代碼裡有資料可讀 */     if(FD_ISSET(socket, &rdfds)) { /* 先判斷一下socket這外被監視的控制代碼是否真的變成可讀的了 */         /* 讀取socket控制代碼裡的資料 */         recv(...);     } }

注意select函式的第一個引數，是所有加入集合的控制代碼值的最大那個值還要加1。比如我們建立了3個控制代碼：

int sa, sb, sc; sa = socket(...); /* 分別建立3個控制代碼並連線到伺服器上 */ connect(sa,...); sb = socket(...); connect(sb,...); sc = socket(...); connect(sc,...); FD_SET(sa, &rdfds);/* 分別把3個控制代碼加入讀監視集合裡去 */ FD_SET(sb, &rdfds); FD_SET(sc, &rdfds);

在使用select函式之前，一定要找到3個控制代碼中的最大值是哪個，我們一般定義一個變數來儲存最大值，取得最大socket值如下：

int maxfd = 0; if(sa > maxfd) maxfd = sa; if(sb > maxfd) maxfd = sb; if(sc > maxfd) maxfd = sc;

然後呼叫select函式：

ret = select(maxfd + 1, &rdfds, NULL, NULL, &tv); /* 注意是最大值還要加1 */

同樣的道理，如果我們要檢測使用者是否按了鍵盤進行輸入，我們就應該把標準輸入0這個控制代碼放到select裡來檢測，如下：
FD_ZERO(&rdfds);
FD_SET(0, &rdfds);
tv.tv_sec = 1;
tv.tv_usec = 0;
ret = select(1, &rdfds, NULL, NULL, &tv);           /* 注意是最大值還要加1 */
if(ret < 0) 
      perror("select");                      /* 出錯 */
else if(ret == 0) 
       printf("超時\n");                   /* 在我們設定的時間tv內，使用者沒有按鍵盤 */
else 
{ /* 使用者有按鍵盤，要讀取使用者的輸入 */
    scanf("%s", buf);
}

 2、pselect函式

1、pselect使用timespec結構，而不使用timeval結構。timespec結構是POSIX的又一個發明。

     struct timespec{

            time_t tv_sec;     //seconds

            long    tv_nsec;    //nanoseconds

     };

     這兩個結構的區別在於第二個成員：新結構的該成員tv_nsec指定納秒數，而舊結構的該成員tv_usec指定微秒數。

2、pselect函式增加了第六個引數：一個指向訊號掩碼的指標。該引數允許程式先禁止遞交某些訊號，再測試由這些當前被禁止的訊號處理函式設定的全域性變數，然後呼叫pselect，告訴它重新設定訊號掩碼。

     關於第二點，考慮下面的例子，這個程式的SIGINT訊號處理函式僅僅設定全域性變數intr_flag並返回。如果我們的程序阻塞於select呼叫，那麼從訊號處理函式的返回將導致select返回EINTR錯誤。然而呼叫select時，程式碼看起來大體如下：
 

if( intr_flag)    handle_intr(); if( (nready = select(...))< 0 ) {    if( errno == EINTR)    {       if( intr_flag)          handle_intr();    }    ... }

問題是在測試intr_flag和呼叫select之間如果有訊號發生，那麼要是select永遠阻塞，該訊號就會丟失。有了pselect後，我們可以如下可靠地編寫這個例子的程式碼：

sigset_t newmask, oldmask, zeromask; sigemptyset(&zeromask); sigemptyset(&newmask); sigaddset(&newmask,SIGINT); sigprocmask(SIG_BLOCK,&newmask,&oldmask);//block SIGINT if(intr_flag)    handle_intr(); if( (nready = pselect(...,&zeromask))< 0 ) {     if(errno== EINTR)     {        if(intr_flag)            handle_intr();     }     ... }

在測試intr_flag變數之前，我們阻塞SIGINT。當pselect被呼叫時，它先以空集（zeromask）取代程序的訊號掩碼，再檢查描述字，並可能進入睡眠。然而當pselect函式返回時，程序的訊號掩碼又被重置為呼叫pselect之前的值（即SIGINT被阻塞）。

 3、poll函式

      poll函式起源於SVR3，最初侷限於流裝置。SVR4取消了這種限制，允許poll工作在任何描述字上。poll提供的功能與select類似，不過在處理流裝置時，它能夠提供額外的資訊。

#include<poll.h> int poll(struct pollfd*fdarray,unsignedlong nfds,int timeout);                返回：就緒描述字的個數，0－超時，-1－出錯

第一個引數是指向一個結構陣列第一個元素的指標。每個陣列元素都是一個pollfd結構，用於指定測試某個給定描述字fd的條件。

           struct pollfd{

                    int fd;              //descriptor to check

                    short events;    //events of interest on fd

                    short revents;   //events that occurred on fd 

            };

       要測試的條件由events成員指定，而返回的結果則在revents中儲存。常用條件及含意說明如下：

poll函式可用的測試值

常量	說明
POLLIN	普通或優先順序帶資料可讀
POLLRDNORM	普通資料可讀
POLLRDBAND	優先順序帶資料可讀
POLLPRI	高優先順序資料可讀
POLLOUT	普通資料可寫
POLLWRNORM	普通資料可寫
POLLWRBAND	優先順序帶資料可寫
POLLERR	發生錯誤
POLLHUP	發生掛起
POLLNVAL	描述字不是一個開啟的檔案

 注意：後三個只能作為描述字的返回結果儲存在revents中，而不能作為測試條件用於events中。

poll()接受一個指向結構''''''''structpollfd''''''''列表的指標，其中包括了你想測試的檔案描述符和事件。事件由一個在結構中事件域的位元掩碼確定。當前的結構在呼叫後將被填寫並在事件發生後返回。在SVR4(可能更早的一些版本)中的 "poll.h"檔案中包含了用於確定事件的一些巨集定義。事件的等待時間精確到毫秒(但令人困惑的是等待時間的型別卻是int)，當等待時間為0時，poll()函式立即返回，-1則使poll()一直掛起直到一個指定事件發生。下面是pollfd的結構。

     struct pollfd {
         int fd;        /* 檔案描述符 */
         short events;  /* 等待的事件 */
         short revents; /* 實際發生了的事件 */
     };
      

於 select()十分相似，當返回正值時，代表滿足響應事件的檔案描述符的個數，如果返回0則代表在規定事件內沒有事件發生。如發現返回為負則應該立即檢視 errno，因為這代表有錯誤發生。

如果沒有事件發生，revents會被清空，所以你不必多此一舉。

這裡是一個例子

   /* 檢測兩個檔案描述符，分別為一般資料和高優先資料。如果事件發生
      則用相關描述符和優先度呼叫函式handler()，無時間限制等待，直到
      錯誤發生或描述符掛起。*/

   
   #include 
   #include 
  
   #include 
   #include 
   #include 
  
   #include 
   #include 
   #include 
  
   #define NORMAL_DATA 1
   #define HIPRI_DATA 2

   int poll_two_normal(int fd1,int fd2)
   {
       struct pollfd poll_list[2];
       int retval;
  
       poll_list[0].fd = fd1;
       poll_list[1].fd = fd2;
       poll_list[0].events = POLLIN|POLLPRI;
       poll_list[1].events = POLLIN|POLLPRI;
  
       while(1)
       {
           retval = poll(poll_list,(unsigned long)2,-1);
           /* retval 總是大於0或為-1，因為我們在阻塞中工作 */
  
           if(retval < 0)
           {
               fprintf(stderr,"poll錯誤: %s\n",strerror(errno));
               return -1;
           }
    
           if(((poll_list[0].revents&POLLHUP) == POLLHUP) ||
              ((poll_list[0].revents&POLLERR) == POLLERR) ||
              ((poll_list[0].revents&POLLNVAL) == POLLNVAL) ||
              ((poll_list[1].revents&POLLHUP) == POLLHUP) ||
              ((poll_list[1].revents&POLLERR) == POLLERR) ||
              ((poll_list[1].revents&POLLNVAL) == POLLNVAL))
             return 0;
  
           if((poll_list[0].revents&POLLIN) == POLLIN)
             handle(poll_list[0].fd,NORMAL_DATA);
           if((poll_list[0].revents&POLLPRI) == POLLPRI)
             handle(poll_list[0].fd,HIPRI_DATA);
           if((poll_list[1].revents&POLLIN) == POLLIN)
             handle(poll_list[1].fd,NORMAL_DATA);
           if((poll_list[1].revents&POLLPRI) == POLLPRI)
             handle(poll_list[1].fd,HIPRI_DATA);
       }
   }
      

在linux的網路程式設計中，很長的時間都在使用select來做事件觸發。在linux新的核心中，有了一種替換它的機制，就是epoll。
相比於select，epoll最大的好處在於它不會隨著監聽fd數目的增長而降低效率。因為在核心中的select實現中，它是採用輪詢來處理的，輪詢的fd數目越多，自然耗時越多。並且，在linux/posix_types.h標頭檔案有這樣的宣告：
#define __FD_SETSIZE    1024
表示select最多同時監聽1024個fd，當然，可以通過修改標頭檔案再重編譯核心來擴大這個數目，但這似乎並不治本。

epoll的介面非常簡單，一共就三個函式：
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(leveltriggered)是預設的工作方式，並且同時支援block和no-blocksocket.在這種做法中，核心告訴你一個檔案描述符是否就緒了，然後你可以對這個就緒的fd進行IO操作。如果你不作任何操作，核心還是會繼續通知你的，所以，這種模式程式設計出錯誤可能性要小一點。傳統的select/poll都是這種模型的代表．

ET(edge-triggered)是高速工作方式，只支援no-blocksocket。在這種模式下，當描述符從未就緒變為就緒時，核心通過epoll告訴你。然後它會假設你知道檔案描述符已經就緒，並且不會再為那個檔案描述符傳送更多的就緒通知，直到你做了某些操作導致那個檔案描述符不再為就緒狀態了(比如，你在傳送，接收或者接收請求，或者傳送接收的資料少於一定量時導致了一個EWOULDBLOCK 錯誤）。但是請注意，如果一直不對這個fd作IO操作(從而導致它再次變成未就緒)，核心不會發送更多的通知(onlyonce),不過在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錯誤(參考mansend),同時,不理會這次請求傳送的資料.所以,需要封裝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);

相關推薦