Linux應用——epoll多路復用模型

前言

后端開發的應該都知道Nginx服務器，Nginx是一個高性能的 HTTP 和反向代理服務器，也是一個 IMAP/POP3/SMTP 代理服務器。后端部署中一般使用的就是Nginx反向代理技術。

Nginx 相較於 Apache 具有占有內存少，穩定性高等優勢，並發能力強的優點。它所使用的網絡通信模型就是epoll。

(*注：epoll模型編程實例需要先了解紅黑樹、tcp/ip、socket、文件描述符fd、阻塞、回調等概念)

epoll介紹

一、epoll模型概念整理

傳統的並發服務器Apache，使用的是多進程/線程模型，每一個客戶端請求都要開啟一個進程去處理，占用的資源大。

epoll是一個I/O多路復用模型，可以用一個進程去處理多個客戶端。

epoll是在2.6內核中提出的，是之前的select和poll的增強版本。關於select和poll是更早的多路復用IO模型，這里不做介紹。

相對於select和poll來說，epoll更加靈活，沒有描述符限制。

epoll使用一個文件描述符管理多個描述符，將用戶關心的文件描述符的事件存放到內核的一個事件表中，這樣在用戶空間和內核空間的copy只需一次。

epoll是基於事件驅動模型。展開應該叫event poll，事件輪詢（猜測），所以程序圍繞着event運行。

二、epoll模型詳細運行過程

在Linux中，epoll模型相關的有3個系統API，通過man 2查看手冊。

int epoll_create(int size);
int epoll_ctl(int  epfd, int op, int fd, struct epoll_event *event);
int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout);

*此外還有一個close()，create返回的是一個文件描述符int epoll_fd，結束時和普通fd一樣要關閉，保證邏輯的完整性。

1.第一步，創建eventepoll結構體

當某一進程調用epoll_create函數時（參數size是事件最大數量，實際上這只是給內核的一個參考值，Linux2.6.8以后這個參數被忽略，但是api文檔仍然建議填寫），

Linux內核會創建一個eventpoll結構體，並返回一個int epoll_fd，這就是epoll通過一個文件描述符操作多個文件描述符的方法。

這個結構體中有兩個成員與epoll的使用方式密切相關。

eventpoll結構體如下所示：

struct eventpoll{
　　struct rb_root rbr;
　　struct list_head rdlist;
};

其中rbr是一個紅黑樹，它的每個結點用來存儲用戶關心的事件（用戶關心的事件，比如服務端server_fd的accept連接請求就是一個事件）。

rblist是一個雙向鏈表用來存儲已發生的事件。

事件的結構體如下所示：

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

一個事件應該對應至少着一個文件描述符加I/O操作，代表這個事件對應的文件描述符和它是讀事件還是寫事件。

對應的I/O操作再epoll_event結構體中是同一個變量events，通過"按位或"操作可以同時添加關心讀和寫事件，"按位與"操作把它讀取。

而它對應的文件描述符不直接在epoll_event結構體下，而是epoll_eventd的data union體下的fd。

2. 第二步，epoll_ctl操作紅黑樹

當用戶調用epoll_ctl向結構體加入event時，會把事件掛在到紅黑樹rbr中。

而所有添加到epoll中的事件都會與低層接口（設備、網卡驅動程序）建立回調關系，也就是說，當相應的事件發生時會調用這個回調方法。

這個回調方法在內核中叫ep_poll_callback，這個過程中，因為用戶關心的事件掛載在紅黑樹上，所以查找效率高只有O(ln(n))的事件復雜度。

然后它會將發生的事件添加到rdlist雙鏈表中。紅黑樹加上函數回調的機制造就了它的高效。

 epoll_event示意圖

3. 第三步，epoll_wait檢查雙向鏈表

當調用epoll_wait檢查是否有事件發生時，只需要檢查eventpoll對象中的rdlist雙鏈表中是否有epitem元素即可。

如果rdlist不為空，則把發生的事件復制到用戶態（把內核的雙向鏈表拷貝成一個struct epoll_event數組），同時返回文件描述符的數量。

用戶只需要用這個數組去接收就可以。

為什么這里要用雙向鏈表而不是單鏈表？

就緒列表引用着就緒的Socket，所以它應能夠快速的插入數據。程序可能隨時調用 epoll_ctl 添加監視Socket，也可能隨時刪除。

當刪除時，若該Socket 已經存放在就緒列表中，它也應該被移除。所以就緒列表應是一種能夠快速插入和刪除的數據結構。

雙向鏈表就是這樣一種數據結構，Epoll 使用雙向鏈表來實現就緒隊列。

三、編程實例

具體實現可以用一個進程去處理多客戶端請求，而不用

1. 使用的頭文件

#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include <pthread.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <netinet/ip.h>
#include <errno.h>
#include <sys/time.h>
#include <sys/epoll.h>　　/* epoll模型api */

2. 函數聲明

#define SERV_PORT 5001
#define SERV_IP_ADDR "192.168.1.7"
#define QUIT "quit"　　/*用戶退出指令*/
#define BACKLOG 5　　/*監聽的最大等待連接隊列*/
#define EPOSIZE 100　　/*接收已發生事件的最大數量*/

/* 套接字初始化的封裝 */
int sock_init(int fd, struct sockaddr_in *sin); 

/* epoll_wait獲得已發生的事件集合之后，具體的業務邏輯 */
void handle_events(int epoll_fd, struct epoll_event *events, int num, int accept_fd); 

/* 具體操作1，接收客戶端連接 */
void do_accpet(int epoll_fd, int accept_fd); 

/* 具體操作2，讀操作 */
void do_read(int epoll_fd, int fd, char *buff); 

/* 具體操作3，寫操作 */
void do_write(int epoll_fd, int fd, char *buff); 

/*把epoll_ctl函數的操作再封裝*/
void event_ctl(int epoll_fd, int fd, int flag, int state);
/* 
argument: flag
EPOLL_CTL_ADD    添加事件
EPOLL_CTL_DEL    刪除事件
EPOLL_CTL_MOD    修改事件

argument: state
EPOLLIN    input事件
EPOLLOUT    output事件
*/ 

3. demo實現

#include "server.h"
int main()
{
    int ret = -1;
    int accept_fd = socket(AF_INET, SOCK_STREAM, 0);
    if(accept_fd < 0)
    {
        perror("socket");
        return 1;
    }

    struct sockaddr_in sin;
    ret = sock_init(accept_fd, &sin);
    if(ret < 0)
    {
        perror("sock_init");
        return 2;
    }

    int epoll_fd = epoll_create(EPOSIZE);
    struct epoll_event events[EPOSIZE];/*用戶空間數組去接收內核的雙向鏈表*/
    event_ctl(epoll_fd, accept_fd, EPOLL_CTL_ADD, EPOLLIN);//先把server_fd accept input事件加入紅黑樹

    while(1)
    {
        ret = epoll_wait(epoll_fd, events, EPOSIZE, -1);/*參數4，超時時間，特別的-1為阻塞等待，詳見linux api: man 2 epoll_wait*/
        handle_events(epoll_fd, events, ret, accept_fd);
    }

    close(epoll_fd);
    close(accept_fd);
}

int sock_init(int fd, struct sockaddr_in *sin)
{
    bzero(sin,sizeof(*sin));
    sin->sin_family = AF_INET;
    sin->sin_port = htons(SERV_PORT);
    sin->sin_addr.s_addr = INADDR_ANY;
    if(bind(fd, (struct sockaddr*)sin, sizeof(*sin)) < 0)
    {
        perror("bind");
        return -1;
    }

    if(listen(fd, BACKLOG) < 0)
    {
        perror("listen");
        return -2;
    }
    return 0;
}


void handle_events(int epoll_fd, struct epoll_event *events, int num, int accept_fd)
{
    int i,fd;
    char buff[BUFSIZ];
    for(i=0; i<num; i++)
    {
        fd = events[i].data.fd;
        if((fd == accept_fd) && events[i].events & EPOLLIN)/* 對events和EPOLLIN“與”操作，判斷這個文件描述符是否有input事件*/
            do_accpet(epoll_fd, fd);
        else if(events[i].events & EPOLLIN)
            do_read(epoll_fd, fd, buff);
        else if(events[i].events & EPOLLOUT)
            do_write(epoll_fd, fd, buff);
    }

}

void do_accpet(int epoll_fd, int accept_fd)
{
    int new_fd;
    struct sockaddr_in cin;
    socklen_t len;
    new_fd = accept(accept_fd, (struct sockaddr*)&cin, &len);
    if(new_fd < 0)
    {
        perror("accpet");
        return;
    }
    printf("a new client connected!\n");

　　/*add_event input，
　　client連接之后，把它的文件描述符的input事件加入到紅黑樹中*/
　　*/

    event_ctl(epoll_fd, new_fd, EPOLL_CTL_ADD, EPOLLIN);
}

void do_read(int epoll_fd, int fd, char *buff)
{
    int ret = -1;
    bzero(buff, BUFSIZ);
    ret = read(fd, buff, BUFSIZ-1);
    if(ret == 0 || !strncmp(buff, QUIT, strlen(QUIT)))
    {
        printf("a client quit.\n");
        close(fd);
        event_ctl(epoll_fd, fd, EPOLL_CTL_DEL, EPOLLIN);//delete_event in
        return;
    }
    
    if(ret < 0)
    {
        perror("read");
        return;
    }
    printf("%s\n", buff);
    event_ctl(epoll_fd, fd, EPOLL_CTL_MOD, EPOLLOUT);//modif_event out
}

void do_write(int epoll_fd, int fd, char *buff)
{
    int ret = -1;
    ret = write(fd, buff, strlen(buff));
    if(ret <= 0)
        perror("write");
    event_ctl(epoll_fd, fd, EPOLL_CTL_MOD, EPOLLIN);//modif_event in
}

/* 
參考的資料中把他分成幾個操作，更直觀
這里為了方便多加一個參數，增加事件/刪除事件/修改事件/，把增刪改集成到一個函數。
*/
void event_ctl(int epoll_fd, int fd, int flag, int state)
{
    struct epoll_event ev;
    ev.events = state;
    ev.data.fd = fd;
    epoll_ctl(epoll_fd, flag, fd, &ev);
}

四、其它

觸發模式

關於epoll的水平觸發LT和邊緣觸發ET還沒研究清楚，

應該是類似驅動程序中檢測硬件信號中，高/低電平觸發，上升沿/下降沿觸發。

* !!FIXME

 

參考資料：

https://www.cnblogs.com/Anker/archive/2013/08/17/3263780.html

https://blog.csdn.net/u011063112/article/details/81771440

https://blog.csdn.net/armlinuxww/article/details/92803381