一、EPOLL的優點

在Linux中，select/poll/epoll是I/O多路復用的三種方式，epoll是Linux系統上獨有的高效率I/O多路復用方式，區別於select/poll。先說select/poll的缺點，以體現epoll的優點。

select:

(1)可監聽的socket受到限制，在32位的系統中，默認最大值為1024.

(2)采用輪詢方式，當要監聽的sock數量很大時，效率低。

(3)隨著要監聽socket數據的增加，要維護一個存放大量fd的數據結構，系統開銷太大。

poll：

解決了可監聽socket數據受限的問題（采用鏈表存儲的方式），但是其他確定跟select一樣，在有大量並發時，效率並不高。

epoll：

相對於select/poll方式，epoll最大的優點是把哪個fd發生的I/O事件通知我們，而不是像select/poll那樣，只是知道有I/O事件發生，具體是哪些fd，並不知道，所以需要從頭到尾輪詢，而隨著要監聽的fd數量增加時，效率會變低，而且當只有幾個活躍的fd時，這個低效率的缺點會更加明顯。總結起來就是：

(1)沒有最大可監聽數量的限制

(2)效率並不會因為要監聽數量的增加而變得低效率

(3)使用mmap文件映射內存來加快消息傳遞

二、EPOLL的ET模式和LT模式

LT模式，也就是水平觸發(select/poll都是水平觸發的)。什麽意思呢？就是說，比如對於寫操作，只要系統緩沖區還有空間可以寫，就一直觸發可寫EPOLLOUT，而讀操作，只要系統緩沖區還有未讀的數據，就一直觸發可讀EPOLLIN。

而ET模式，就是邊沿觸發，邊沿，類似於電子電路中的邊沿概念。具體來說，有點復雜，請看下面：

對於讀操作：

(1) 當buffer由不可讀狀態變為可讀的時候，即由空變為不空的時候。

(2) 當有新數據到達時，即buffer中的待讀內容變多的時候。

(3) 當buffer中有數據可讀（即buffer不空）且用戶對相應fd進行epoll_mod IN事件時。

對於寫操作：

(1) 當buffer由不可寫變為可寫的時候，即由滿狀態變為不滿狀態的時候。

(2) 當有舊數據被發送走時，即buffer中待寫的內容變少得時候。

(3) 當buffer中有可寫空間（即buffer不滿）且用戶對相應fd進行epoll_mod OUT事件時（具體見下節內容）。

請看下面圖示：

(1)可讀:由空到非空

(2)可讀，可讀數據變多了

圖1 ET讀觸發的兩種情況

(註：這幾個圖來自：http://blog.chinaunix.net/uid-28541347-id-4285054.html)

三、EPOLL 觸發時機

（1）EPOLLIN

ET模式：每次EPOLL_CTL_ADD或EPOLL_CTL_MOD時，如果加入前，就是可讀狀態，那麽加入後會觸發1次 ，不管sock緩沖是否讀完，只要對方有send或connect或close或強退，就會觸發EPOLLIN（ET/LT是針對一次socket fd就緒的，即一次fd就緒後有數據沒讀完/沒寫完，是否還會通知，所以該連接新的數據到來，該事件會觸發）

LT模式：只要socket可讀，就會一直觸發EPOLLIN

（2）EPOLLOUT

ET模式： 每次EPOLL_CTL_ADD或EPOLL_CTL_MOD時，如果加入前，就是可寫狀態，那麽加入後會觸發1次 ，如果EPOLLOUT與EPOLLIN一起註冊，不管sock發送緩沖是否從滿變不滿，只要socket發送是不滿的，那麽每次EPOLLIN觸發時，都會觸發EPOLLOUT，即獲取到不被期望的寫事件，這也是為什麽要使用ATM模式的原因

LT模式：只要socket可寫，就會一直觸發EPOLLOUT

簡單來說，ET模式下，只要監聽了EPOLLIN和EPOLLOUT，socket的每次動作（包括close與不close直接強退），都會觸發1次， 與讀寫緩沖的狀態無關。

註意：EPOLLERR/EPOLLHUP，這兩個是默認已經加到epoll events裏面的，無需手動加入。

四、EPOLL ET模式讀寫以及accept方式

1. EPOLL ET模式的fd為什麽要設置為非阻塞模式？

答：因為ET模式下的讀寫需要一直讀或寫直到出錯（對於讀，當讀到的實際字節數小於請求字節數時就可以停止），而如果你的文件描述符如果不是非阻塞的，那這個一直讀或一直寫勢必會在最後一次阻塞。這樣就不能在阻塞在epoll_wait上了，造成其他文件描述符的任務餓死。下面是設置為非阻塞的代碼：

void set_nonblock(int fd)
{
    int fl = fcntl(fd, F_GETFL);
    assert(fl != -1);
    int rc = fcntl(fd, F_SETFL, fl | O_NONBLOCK);
    assert(rc != -1);
}

2. ET模式的讀寫

對於讀操作，就一直讀，直到遇到EAGAIN錯誤，或者讀到為0（對端關閉）或者小於buffer（一次讀取的信息）。

對於寫操作，就一直寫，直到數據發送完，或者 errno = EAGAIN（表示系統緩沖區已滿，這個時候，可以選擇返回或者等待）。下面是偽代碼:

/*
 * Return Value: data len that have read
 * Error: -1: read failed, -2: peer fd is closed, -3: no more space
 */
int sock_recv(int fd, char *ptr, int len)
{
    assert(len > 0 && fd > 0);
    assert(ptr != NULL);
    int nread = 0, n = 0;
    while(1) {
        nread = read(fd, ptr+n, len-1);
        if(nread < 0) {
            if(errno == EAGAIN || errno == EWOULDBLOCK) {
                return nread; //have read one
            } else if(errno == EINTR) {
                continue; //interrupt by signal, continue
            } else if(errno == ECONNRESET) {
                return -1; //client send RST
            } else {
                return -1; //faild
            }
        } else if(nread == 0) {
            return -2; //client is closed
        } else if(nread < len-1) {
            return nread; //no more data, read done
        } else {
            /*
             * Here, if nread == len-1, maybe have add done,
             * For simple, we just return here,
             * A better way is to MOD EPOLLIN into epoll events again
             */
            return -3; //no more space
        }
    }

    return nread;
}

/*
 * Return Value: data len that can not send out
 * Normal Value: 0, Error Value: -1
 */
int sock_send(int fd, char *ptr, int len)
{
    assert(fd > 0);
    assert(ptr != NULL);
    assert(len > 0);
    int nsend = 0, n = len;
    while(n > 0) {
        nsend = send(fd, ptr+len-n, n, 0);
        if(nsend < 0) {
            if(errno == EINTR) {
                nsend = 0; //interrupt by signal
            } else if(errno == EAGAIN) {
                //Here, write buffer is full, for simple, just sleep,
                //A better is add EPOLLOUT again?
                usleep(1); 
                continue;
            } else {
                return -1; //send failed!
            }
        }
        
        if(nsend == n) {
            return 0; //send all data
        }

        n -= nsend;
    }

    return n;
}

3. ET模式下accept問題

考慮這種情況：多個連接同時到達，服務器的TCP就緒隊列瞬間積累多個就緒連接，由於是邊緣觸發模式，epoll只會通知一次，accept只處理一個連接，導致TCP就緒隊列中剩下的連接都得不到處理。 解決辦法是用while循環抱住accept調用，處理完TCP就緒隊列中的所有連接後再退出循環。如何知道是否處理完就緒隊列中的所有連接呢？accept返回-1並且errno設置為EAGAIN就表示所有連接都處理完。

綜合以上兩種情況，服務器應該使用非阻塞地accept，accept在ET模式下的正確使用方式為：

while((fd = accept(listenfd, (struct sockaddr *)&addr, (size_t *)&addrlen)) > 0)
{
    handle_client(fd);
}

if(fd == -1)
{
    if(errno != EAGAIN && errno != ECONNABORTED && errno != EPROTO && errno != EINTR)
    {
         printf("accept failed!");
    }
}

所謂局域網聊天室，就是把客戶端發過來的信息，轉發給其他客戶端。具體實現如下：

1. 聊天室服務端

(1)在accept之後，把新的connect_fd EPOLL_CTL_ADD EPOLLIN 到epoll events中。同時，在accept之後，服務端就可以發送消息給客戶端了（這個時候，服務端不能在這裏接收客戶端的消息，請問為什麽？）。

(2)監聽EPOLLIN事件，接收來自客戶端的信息，並把它轉發給其他客戶端（請問為什麽在這裏可以直接轉發消息給客戶端，而不需要再MOD EPOLLOUT事件，然後再監聽EPOLLOUT事件？）

(3)使用雙鏈表存儲客戶端的信息（為什麽使用雙鏈表？因為首先你並不知道有多少個客戶端，其次使用雙鏈表便於動態增加或刪除客戶端信息（當客戶端退出的時候，要刪除對應的記錄））

下面是相關的偽代碼：

int nfds = epoll_wait(efd, p_events, MAX_EPOLL_NUM, -1);
int i, conn_fd;
for(i = 0; i < nfds; i++)
{
    if(p_events[i].data.fd == fd) //new connect is come in, accept it
    {
        while((conn_fd = accept(fd, (struct sockaddr *)&client_addr, &client_addr_len)) > 0)
        {
            ev.data.fd = conn_fd;
            ev.events = EPOLLIN;
            rc = epoll_ctl(efd, EPOLL_CTL_ADD, conn_fd, &ev);

            bzero(message, MAX_BUF_SIZE);
            sprintf(message, STR_WELCOME, conn_fd);
            rc = send(conn_fd, message, strlen(message), 0);

            //insert our double list to store client information
        }

        if(conn_fd == -1)
        {
            if(errno != EAGAIN && errno != ECONNABORTED 
               && errno != EPROTO && errno != EINTR)
            {
                perror("accept");
                return -1;
            }
            continue; //should not return here, since maybe we have handle all fd
        }
    }
    else if(p_events[i].events & EPOLLERR || p_events[i].events & EPOLLHUP)
    {
        //happen error, delete it 
        rc = epoll_ctl(efd, EPOLL_CTL_DEL, p_events[i].data.fd, &ev);
        assert(rc != -1);
        close(p_events[i].data.fd);
        p_events[i].data.fd = -1;
    }
    else
    {
        //After accept, we can receive msg from clinet and resend back to other clients.
        handle_message(&head, &tail, p_events[i].data.fd);
    }
}                    

int handle_message(struct double_list **head, struct double_list **tail, int fd)
{
    //receive msg from fd
    
    //send msg to other client except fd
}

2. 聊天室客戶端

客戶端實現的功能是，基於EPOLL模型，等待用戶輸入，把所輸入的信息發送給服務端，並從服務端接收信息，最後顯示出來。具體實現為父進程+子進程，使用PIPE的IPC方式。子進程等待用戶出入，然把消息通過PIPE發送給父進程，而父進程從子進程接收信息再發送給服務端，並從服務端接收信息再顯示出來。下面是偽代碼：

#define CHK(eval) if(eval < 0){perror("eval"); exit(-1);}
#define CHK2(res, eval) if((res = eval) < 0){perror("eval"); exit(-1);}

int pipe_fd[2]; //pipe_fd[0]: read, pipe_fd[1]: write
CHK(pipe(pipe_fd));
ev.data.fd = fd;
CHK2(rc, epoll_ctl(efd, EPOLL_CTL_ADD, fd, &ev));
ev.data.fd = pipe_fd[0];
CHK2(rc, epoll_ctl(efd, EPOLL_CTL_ADD, pipe_fd[0], &ev));
    
int exit_flag = 0;
CHK2(rc, fork());

if(rc < 0)
{
    perror("fork");
}
else if(rc == 0)
{
    //child recv message from input and pass it to parent to send to server
    close(pipe_fd[0]); //close read fd
    while(exit_flag == 0)
    {    
        printf("Enter ‘exit‘ to exit\n");
        fgets(message, sizeof(message), stdin);
        message[strlen(message)-1] = ‘\0‘;

        CHK(write(pipe_fd[1], message, strlen(message))); //pass it to parent
    }
}
else
{
    //parent recv message from server and print it
    close(pipe_fd[1]); //close write fd
    int i, n, has_data_flag, nread, nfds = 0;
    while(exit_flag == 0)
    {
        CHK2(nfds, epoll_wait(efd, events, MAX_EPOLL_NUM, -1));

        for(i=0; i<nfds; i++)
        {
            if(events[i].data.fd == fd)
                //msg from char server, receive it and print it to stdout
            else if(events[i].data.fd == pipe_fd[0])
                //msg from child process, recive it and resend it to char server
        }
    }
}

if(rc == 0)
{
    //child
    close(pipe_fd[1]); //close write fd
}
else
{
    //parent
    close(pipe_fd[0]); //close read fd
    close(fd);
}

六、基於EPOLL的echo服務器

所謂echo服務，就是實現回顯功能，具體就是，echo客戶端把用戶出入的信息發給echo服務端，echo服務端再把消息返回發送給客戶端，最後客戶端再把接收的消息顯出出來。

其實，跟上面局域網聊天室非常類似，只要稍微改改代碼就可以了，這裏就不具體分析了。

七、EPOLL並發測試

寫一個客戶端，並發1000個fd去並發連接上面的聊天室服務端，可以看到EPOLL對於並發的處理效率還是挺高的（TBD：用select /poll來做測試對比），下面是部分代碼：

clock_t start_time = clock();
    for(i=0; i<MAX_CLIENT_NUM; i++)
    {
        fd = socket(AF_INET, SOCK_STREAM, 0);
        assert(fd != -1);

        rc = connect(fd, (struct sockaddr *)&serv_addr, serv_addr_len);
        assert(rc != -1);
        fds[i] = fd;

        bzero(message, MAX_BUF_SIZE);
        rc = recv(fd, message, MAX_BUF_SIZE, 0);
        printf("%s\n", message);
    }
    
    for(i=0; i<MAX_CLIENT_NUM; i++)
    {
        close(fds[i]);
    }
    printf("Total connections: %d, Test passed at: %.2f seconds\n", MAX_CLIENT_NUM, (double)(clock()-start_time)/CLOCKS_PER_SEC);
    

八、寫在最後

上面所有的代碼都可以在我的GitHub上找到。我的GitHub地址：https://github.com/wolf623/chat_epoll

參考：http://blog.chinaunix.net/uid-28541347-id-4296180.html

<wiz_tmp_tag id="wiz-table-range-border" contenteditable="false" style="display: none;">

基於EPOLL模型的局域網聊天室和Echo服務器