select & epoll

同步、異步、堵塞和非堵塞差異

　　同步：發出一個功能調用時。在沒有獲得結果以前，該調用就不返回
　　異步：當一個異步過程調用發出後。調用者不能立馬獲得結果。實際處理這個調用的部件在完畢後。經過狀態、通知和回調來通知調用者

　　堵塞：堵塞調用是指調用結果返回以前，當前線程會被掛起。

函數僅僅有在獲得結果以後纔會返回
　　
　　非堵塞：不能立馬獲得結果以前，該函數不會堵塞當前線程，而會立馬返回
　　

　　同步與堵塞的差異：

　　 - 同步不必定堵塞，對於同步調用來講，許多時候當前線程仍是激活的，僅僅是從邏輯上當前函數沒有返回而已，如調用recv函數，假設緩衝區中沒有數據，這個函數就會一直等待，直到有數據才返回。而此時，當前線程還會繼續處理各類各樣的消息
　　 - 當socket工做在堵塞模式的時候。 假設沒有數據的狀況下調用該函數，則當前線程就會被掛起，直到有數據爲止

　　同步堵塞：效率最低
　　同步非堵塞：將fd設爲O_NONBLOCK，則可立刻返回
　　異步堵塞：如selectIO複用。不是在處理消息時堵塞。而是在等待消息被觸發時被堵塞

IO多路複用

　　select、poll、epoll都是IO複用機制，IO複用是指在一個線程中同一時候監聽多個描寫敘述符，一旦有一個描寫敘述符就緒。就可以通知程序對該描寫敘述符進行讀寫操做。

　　IO複用的基本方法都是：先構建一張有關描寫敘述符的列表。而後使用統一的堵塞函數，在等待消息被觸發時被堵塞。當不論什麼socket有事件通知時跳出堵塞狀態。

　　IO複用本質都是同步IO。只是經過提早堵塞地監聽描寫敘述符，保證有描寫敘述符需要進行讀寫時才調用相應的同步讀寫函數。跳過了對每個描寫敘述符的等待過程。提升了併發時的效率。
　　

select

相關接口

• 函數原型

  #include <sys/select.h> //select; 
   #include <sys/time.h> //struct timeval;
   int select(
        int nfds,
        fd_set *readset,
        fd_set *writeset,
        fd_set* exceptset,
        struct timeval *timeout
     );
     //返回已就緒的描寫敘述符個數
     //當某些調用出錯時，返回-1。可經過errno查看錯誤代碼
     //當超時仍沒有事件就緒時，返回0

• 參數

  • nfds：第一個參數是：最大的文件描寫敘述符值+1；
  • readset：可讀描寫敘述符集合；
  • writeset：可寫描寫敘述符集合；
  • exceptset：異常描寫敘述符；
  • timeout：select 的監聽時長。timeout爲NULL時，堵塞等待。timeout爲0時，立刻返回

FD_ZERO(fd_set *)  //清空描寫敘述符集合
FD_SET(int, fd_set *)  //向描寫敘述符集合增長指定描寫敘述符
FD_CLR(int, fd_set *) //從描寫敘述符集合刪除指定描寫敘述符
FD_ISSET(int, fd_set *) //檢測指定描寫敘述符是否在描寫敘述符集合中
#define FD_SETSIZE 1024  //描寫敘述符最大數量

select調用的基本框架（以read操做爲例）：

fd_set rdset;
listNode *p=head;
int max_fd;
int ret=0;
struct timeval timeout={3,0};  //設置超時時間
while(1){ 
    //每次調用select以前都要清空fdset。並又一次增長所有描寫敘述符
    FD_ZERO(&rdset); 
    p=head; 
    while(p){        //依次將每個待監聽的描寫敘述符增長列表
        FD_SET(p->fd，&rdset);
        if(p->fd>max_fd)
            max_fd=p->fd;
        p=p->next;
    }
    ret=select(max_fd+1,&rdset,NULL,NULL,&timeout);
    if(ret<0)  //返回-1。則出錯
        error;
    if(ret==0)  //timeout超時時，仍沒有不論什麼描寫敘述符就緒
        continue;
    for(int i=0;i<=max_fd;++i){
    //依次檢測每個描寫敘述符。若就緒，則調用相應的回調函數進行處理
        if(FD_ISSET(i,&rdset){  
            callback(i);
        }
    }
}

select的原理

　　select是基於整型數組的，。每個整型數字有32bit，每一位可表示一個fd。好比假設最大描寫敘述符數爲1024，則需要1024/32 = 32，即fd_set set爲int a[32]。
　　1）每次在調用select以前，都需要經過FD_SET增長待監聽的描寫敘述符。
　　FD_SET(int fd, fd_set *pSet)；則是將pSet的第fd位置爲1（從0開始計數）；如fd = 3，則通過FD_SET後，最後一字節爲 0000 1000
　　2）當調用select後，就將各fd_set從用戶態copy到內核。當在超時時間內相應的描寫敘述符沒有事件就緒，則將該位置0.
　　如在調用select以前，調用FD_SET設置了二、 三、 4這三個描寫敘述符，則fd_set最低字節爲 0001 1100。若在超時時間內，二、 3描寫敘述符就緒，4未就緒。

則在超時返回時，fd_set最低字節變爲 0000 1100。將改變後的fd_set從內核copy到用戶態。
　　3）遍歷每個描寫敘述符，經過FD_ISSET推斷是否就緒，若就緒。則進行相應處理。
　　儘管可以改變描寫敘述符數量的最大值。但是這樣會致使數組很大。效率變得很低，所以意義不大。

　　

select的缺陷

　　經過上述對select原理的分析可知，select主要有下面幾個缺陷：
　　1）描寫敘述符數量的限制。對於高併發的場景1024個描寫敘述符遠遠不夠
　　2）每次調用select以前都需要依次向fd_set中增長每個待監聽的描寫敘述符
　　3）調用select時，都需要將fd_set從用戶態copy到內核，超時時。又要將返回的fd_set從內核copy到用戶態
　　4）調用select時，底層是經過輪詢每個描寫敘述符來推斷是否有描寫敘述符就緒的
　　5）select返回的結果僅僅是就緒的描寫敘述符的個數，而不是詳細事件。所以需要遍歷描寫敘述符，依次推斷每個描寫敘述符是否就緒

epoll

epoll是對select的改進：

　　1）epoll的底層是基於紅黑樹的，每次增長一個監聽的描寫敘述符。便是向紅黑樹中增長一個節點。
　　所以沒有最大描寫敘述符的限制
　　2）同一時候監聽的描寫敘述符列表沒有發生變化時，底層的紅黑樹不會發生不論什麼變化，所以再也不需要每次都又一次增長描寫敘述符。

當要新增長或刪除某個描寫敘述符時，僅僅是在紅黑樹中增長或刪除一個節點
　　3）epoll還相應一個雙向鏈表。每當有一個描寫敘述符就緒時，就將該事件增長到雙向鏈表中。

　　當調用epoll_wait等待就緒事件時，內核是經過檢查雙向鏈表是否爲空來推斷是否有事件就緒的，而不用輪詢每個描寫敘述符 　　
　　4）epoll採用了共享內存，所以用戶空間和內核傳遞數據再也不需要copy，使得效率大大提升
　　5）epoll_wait返回時，還會返回就緒描寫敘述符，所以再也不需要依次遍歷每個描寫敘述符進行推斷
　　

epoll的相關接口

　　epoll頭文件爲　　

#include <sys/epoll.h>

　　epoll**僅僅有epoll_create,epoll_ctl,epoll_wait** 3個系統調用
　　1. int epoll_create(int size);
　　建立一個epoll的句柄。自從linux2.6.8以後，size參數是被忽略的。需要注意的是。當建立好epoll句柄後，它就是會佔用一個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。
• event：告訴內核需要監聽什麼事件

struct 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 */  
};

events可以是下面幾個宏的集合：

• EPOLLIN ：讀事件（包含對端SOCKET正常關閉）；
• EPOLLOUT：寫事件；
• EPOLLPRI：表示相應的文件描寫敘述符有緊急的數據可讀（這裏應該表示有帶外數據到來）；
• EPOLLERR：錯誤；
• EPOLLHUP：表示相應的文件描寫敘述符被掛斷；
• EPOLLET： 將EPOLL設爲ET模式，這是相對於LT來講的。

• EPOLLONESHOT：僅僅監聽一次事件，當監聽完此次事件以後。假設還需要繼續監聽這個socket的話。需要再次把這個socket增長到EPOLL隊列裏

  3. int epoll_wait(int epfd, struct epoll_event * events, int maxevents, int timeout);
  　　得到在epoll監聽的描寫敘述符中已就緒的事件。
  　　events是分配好的epoll_event結構體數組。epoll將會把發生的事件賦值到events數組中（events不可以是空指針，內核僅僅負責把數據拷貝到這個events數組中，不會去幫助咱們在用戶態中分配內存）。
  　　maxevents告以內核這個events有多大。這個 maxevents的值不能大於建立epoll_create()時的size，
  　　timeout是超時時間
  　　假設函數調用成功。返回相應I/O上已準備好的文件描寫敘述符數目。如返回0表示已超時。

epoll兩種工做模式：

• LT（水平觸發）：若程序此次沒有處理，下次還會再次通知　　
• ET（邊沿觸發）：僅僅通知一次，若程序沒有處理。後面不會再通知
  系統默以爲LT模式

epoll的基本使用框架　　

int epfd;
struct epoll_event event;  
struct epoll_event *events;  
epfd = epoll_create1 (0);  
event.data.fd = listenfd;  //listenfd爲server端監聽fd
event.events = EPOLLIN | EPOLLET;//讀入,邊緣觸發方式 
s = epoll_ctl (efd, EPOLL_CTL_ADD, listenfd, &event);  
events = calloc (MAXEVENTS, sizeof event);  
while(1)  {
       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); //accept這個鏈接 
               ev.data.fd=connfd;  
               ev.events=EPOLLIN|EPOLLET;  
               epoll_ctl(epfd,EPOLL_CTL_ADD,connfd,&ev); //將新的fd增長到epoll的監聽隊列中 
           } else if( events[i].events&EPOLLIN ) { //接收到數據，讀socket 
               n = read(sockfd, line, MAXLINE)) < 0    //讀 
               ev.data.ptr = md;     //md爲本身定義類型，增長數據 
               ev.events=EPOLLOUT|EPOLLET;  
               epoll_ctl(epfd,EPOLL_CTL_MOD,sockfd,&ev);//改動標識符，等待下一個循環時發送數據，異步處理的精髓 
           } else if(events[i].events&EPOLLOUT){ //有數據待發送，寫socket 
               struct myepoll_data* md = (myepoll_data*)events[i].data.ptr;    //取數據 
               sockfd = md->fd;  
               send( sockfd, md->ptr, strlen((char*)md->ptr), 0 );        //發送數據 
               ev.data.fd=sockfd;  
               ev.events=EPOLLIN|EPOLLET;  
               epoll_ctl(epfd,EPOLL_CTL_MOD,sockfd,&ev); //改動標識符，等待下一個循環時接收數據 
           } else  {  
               //其它的處理 
           }  
       }  
   }