epoll系列系统调用

以下内容均出自于Linux高性能服务器编程。
I/O复用使得程序能同时处理多个文件描述符。常用的I/O复用有select、poll、epoll三种。
在linux的网络编程中，很长的时间都在使用select来做事件触发。在linux新的内核中，有了一种替换它的机制，就是epoll。相比于select，epoll最大的好处在于它不会随着监听fd数目的增长而降低效率。因为在内核中的select实现中，它是采用轮询来处理的，轮询的fd数目越多，自然耗时越多。
并且，在linux/posix_types.h头文件有这样的声明：

#define __FD_SETSIZE    1024

表示select最多同时监听1024个fd，当然，可以通过修改头文件再重编译内核来扩大这个数目，但这似乎并不治本。
epoll是Linux特有的I/O复用函数，其使用一组函数来完成任务，而不是单个函数。epoll把用户关系的文件描述符上的事件放在内核里的一个事件表中，从而不需要像select和poll那样每次都重复传入文件描述符集或者事件集。
epoll的接口非常简单，一共就三个函数：

1. epoll_create函数:

#include <sys/epoll.h>
int epoll_create(int size)

epoll需要使用一个额外的文件描述符，来唯一标识内核中的这个事件表，这个文件描述符是通过epoll_create函数来创建。需要注意的是，当创建好epoll句柄后，它就是会占用一个fd值，在linux下如果查看/proc/进程id/fd/，是能够看到这个fd的，所以在使用完epoll后，必须调用close()关闭，否则可能导致fd被耗尽。
注意：size参数只是给内核一个提示，这个 epoll对象会处理事件的大概数目，而不是能够处理的事件的最大个数，在Linux最新的一些内核版本的实现中，这个 size参数没有任何意义。

2. epoll_ctl函数：

#include <sys/epoll.h>
int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event)

epoll_ctl函数是epoll的事件注册函数，通过epoll_ctl函数向 epoll对象中添加、修改或者删除感兴趣的事件，成功时返回0，失败时返回–1并设置errno，此时需要根据errno错误码判断错误类型。
第一个参数是epoll_create函数的返回值；
第二个参数op指定操作类型，用三个宏来表示：

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事件类型
    epoll_data_t data; //存储用户数据
 };
typedef union epoll_data 
{
     void *ptr;
     int fd;
     __uint32_t u32;
     __uint64_t u64;
} epoll_data_t;

events可以是以下几个宏的集合：
EPOLLIN ：表示对应的文件描述符可以读（包括对端SOCKET正常关闭）；
EPOLLOUT：表示对应的文件描述符可以写；
EPOLLPRI：表示对应的文件描述符有紧急的数据可读（这里应该表示有带外数据到来）；
EPOLLERR：表示对应的文件描述符发生错误；
EPOLLHUP：表示对应的文件描述符被挂断；
EPOLLET： 将EPOLL设为边缘触发(Edge Triggered)模式，这是相对于水平触发(Level Triggered)来说的；
EPOLLONESHOT：只监听一次事件，当监听完这次事件之后，如果还需要继续监听这个socket的话，需要再次把这个socket加入到EPOLL队列里。

3.epoll_wait函数：

#include <sys/epoll.h>
int epoll_wait(int epfd, struct epoll_event* events, int maxevents, int timeout);

epoll_wait函数是epoll系列系统调用的主要接口，它在一段超时时间内等待一组文件描述符上的事件。
第一个参数是epoll_create函数的返回值；
第二个参数events则是分配好的 epoll_event结构体数组，epoll将会把发生的事件复制到 events数组中（events不可以是空指针，内核只负责把数据复制到这个 events数组中，不会去帮助我们在用户态中分配内存）；
第三个参数 maxevents指定最多可以监听多少个事件，它必须大于0，maxevents的值不能大于创建epoll_create()时的size；
第四个参数 timeout表示在没有检测到事件发生时最多等待的时间（单位为毫秒），如果 timeout为0，则表示 epoll_wait在 rdllist链表中为空，立刻返回，不会等待。

epoll_wait函数成功时会返回就绪的文件描述符的个数，失败时返回-1并设置errno。如果函数检测到事件，就将所有就绪的事件从内核事件表（由epfd参数指定）中复制到它的第二个参数events指向的数组中。这个数组只用于传输epoll_wait检测到的就绪事件，而不像select和poll的数组参数那样既用于传入用户注册的事件，又用于传输内核检测到的就绪事件，这就极大地提高了应用程序索引就绪文件描述符的效率。

LT和ET模式

epoll对文件描述符的操作有两种模式：LT（Level Trigger，电平触发）模式和ET（Edge Trigger，边沿触发）模式。
对于采用LT工作模式的文件描述符，当epoll_wait检测到其上面有事件发生并将此事件通知应用程序后，应用程序可以不立即处理该事件，那么应用程序下一次调用epoll_wait时，epoll_wait还会向应用程序通知该事件，直至该事件被处理。而对于ET工作模式的文件描述符，当epoll_wait检测到其上面有事件发生并将此事件通知给应用程序后，应用程序必须立即处理该事件，因为后续的epoll_wait调用将不再向应用程序通知这一事件。因此，ET模式很大程度上降低了同一个epoll事件被重复触发的次数，相对效率较高。

ET模式与LT模式的区别在于：
当一个新的事件到来时，在ET模式下可以通过epoll_wait函数获取到这个事件，可是如果没有一次处理完事件对应的套接字缓冲区内的数据，那么在这个套接字没有新的事件再次到来之前，在ET模式下是无法再次从epoll_wait调用中获取到这个事件的；而LT模式则相反，只要事件对应的套接字缓冲区内还有数据，就能从epoll_wait调用中获取到这个事件。因此，在LT模式下开发基于epoll的应用相对容易一点，而在ET模式下发生事件时，如果没有彻底地将缓冲区内数据处理完，就会导致缓冲区中的用户请求得不到响应，也就是说，如果要采用ET模式，需要一直read/write直到出错为止。

下面给出一个完整的服务器端例子：

#include <stdio.h>
#include <stdlib.h>
#include <stdbool.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <assert.h>
#include <unistd.h>
#include <errno.h>
#include <string.h>
#include <fcntl.h>
#include <sys/epoll.h>
#include <pthread.h>

#define MAX_EVENT_NUMBER 1024
#define BUFFER_SIZE 10

//将文件描述符设置为非阻塞
int SetNonBlocking(int sockfd)
{
    int old_option = fcntl(sockfd, F_GETFL); //取得文件描述符旗帜，此旗帜为open函数的参数flag
    int new_option = old_option | O_NONBLOCK;
    fcntl(sockfd, F_SETFL, new_option);
    return old_option;
}
/*
将文件描述符fd上的EPOLLIN事件注册到epollfd指示到的epoll内核事件表中，
参数enable_et指定是否对fd启用ET模式。
*/
void AddFd(int epollfd, int fd, bool enable_et)
{
    struct epoll_event event;
    event.data.fd = fd;
    event.events = EPOLLIN; //EPOLLIN：连接到达，有数据来临
    if (enable_et)
    {
        event.events |= EPOLLET;
    }
    epoll_ctl(epollfd, EPOLL_CTL_ADD, fd, &event);
    SetNonBlocking(fd);
}
//LT模式的工作流程
void lt(struct epoll_event* events, int number, int epollfd, int listenfd)
{
    char buf[BUFFER_SIZE];
    for (int i = 0; i < number; ++i)
    {
        int sockfd = events[i].data.fd;
        //如果监测到一个新的socket用户连接到绑定的服务器端口，那么建立新的连接
        if (sockfd == listenfd)
        {
            struct sockaddr_in client_address;
            socklen_t client_addrlength = sizeof(client_address);
            int clientfd = accept(listenfd, (struct sockaddr*)&client_address,
                                  &client_addrlength);
            //将客户端套接字上的事件注册到内核事件表上
            AddFd(epollfd, clientfd, false);
        }
        else if (events[i].events & EPOLLIN)
        {
            //如果是已经连接的旧用户，并且接受到数据，那么进行数据的读取
//            printf("event trigger once
");
            memset(buf, '', BUFFER_SIZE);
            int ret = recv(sockfd, buf, BUFFER_SIZE - 1, 0);
            if (ret <= 0)
            {
                close(sockfd);
                continue;
            }
            printf("get %d bytes of content: %s
", ret, buf);
        }
        else
        {
            printf("something else happened 
");
        }
    }
}
//ET模式的工作流程
void et(struct epoll_event* events, int number, int epollfd, int listenfd)
{
    char buf[BUFFER_SIZE];
    for (int i = 0; i < number; ++i)
    {
        int sockfd = events[i].data.fd;
        if (sockfd == listenfd)
        {
            struct sockaddr_in client_address;
            socklen_t client_addrlength = sizeof(client_address);
            int clientfd = accept(listenfd, (struct sockaddr*)&client_address,
                                  &client_addrlength);
            AddFd(epollfd, clientfd, true);
        }
        else if (events[i].events & EPOLLIN)
        {
            //这段代码不会被重复触发，所以我们循环读取数据，确保socket读缓冲区中的所有数据被读出
            printf("event trigger once
");
            while (1)
            {
                memset(buf, '', BUFFER_SIZE);
                int ret = recv(sockfd, buf, BUFFER_SIZE - 1, 0);
                if (ret < 0)
                {
                    /*对于非阻塞IO，下面的条件成立表示数据已经全部读取完毕。
                      此后，epoll就能再次触发sockfd上的EPOLLIN事件，以驱动下一次读事件*/
                    if ((errno == EAGAIN) || (errno == EWOULDBLOCK))
                    {
                        printf("read later
");
                        break;
                    }
                    close(sockfd);
                    break;
                }
                else if (0 == ret)
                {
                    close(sockfd);
                }
                else
                {
                    printf("get %d bytes of content: %s
", ret, buf);
                }
            }
        }
        else
        {
            printf("something else happened 
");
        }
    }
}

int main(int argc, char* argv[])
{
    int ret = 0;
    int serverfd = 0;
    struct sockaddr_in server_address;

    if (argc != 2)
    {
        printf("Usage: %s <port>
", argv[0]);
        exit(1);
    }

    //创建套接字
    serverfd = socket(AF_INET, SOCK_STREAM, 0);
    assert(serverfd >= 0);
    //绑定IP地址
    memset(&server_address, 0, sizeof(server_address));
    server_address.sin_family = AF_INET;
    server_address.sin_addr.s_addr = htonl(INADDR_ANY);
    server_address.sin_port = htons(atoi(argv[1]));
    ret = bind(serverfd, (struct sockaddr*)&server_address, sizeof(server_address));
    assert(ret != -1);
    //监听请求
    ret = listen(serverfd, 5);
    assert(ret != -1);

    struct epoll_event events[MAX_EVENT_NUMBER];
    //
    int epollfd = epoll_create(5);
    assert(epollfd != -1);
    //将服务器套接字上的事件注册到epollfd所指向的内核事件表上
    AddFd(epollfd, serverfd, true);

    while (1)
    {
        int ret = epoll_wait(epollfd, events, MAX_EVENT_NUMBER, -1);
        if (ret < 10)
        {
            printf("epoll failure
");
        }
        lt(events, ret, epollfd, serverfd);  //LT模式
        //et(events, ret, epollfd, serverfd);
    }
    close(serverfd);
    return 0;
}

三组I/O复用函数的比较

select、poll和epoll三组I/O复用系统调用，这三组系统调用都能同时监听多个文件描述符，它们将等待由timeout参数指定的超时时间，直到一个或者多个文件描述符上有事件发生时的返回，返回值是就绪的文件描述符的数量。下面我们从事件集、最大支持文件描述符数量、工作模式和具体实现等四个方面来对比它们的异同。

这三组函数都通过某种结构体变量来告诉内核监听哪些文件描述符上的哪些事件，并用该结构体类型的参数来获取内核处理的结果。select的参数类型fd_set没有将文件描述符和事件绑定，它仅仅是一个文件描述符集合，因此select需要提供三个这种类型的参数来分别传入和输出可读、可写及异常等事件。select函数原型如下：

#include <sys/select.h>
#include <sys/time.h>
int select(int maxfd, fd_set* readset, fd_set* writeset, fd_set* exceptset, const struct timeval* timeout)

这种方式使得select不能处理更多的事件类型，同时由于内核对fd_set集合的在线修改，使得应用程序下次调用select前不得不重置这三个fd_set集合。

poll的参数类型pollfd相对要好一些，它把文件描述符和事件都定义在其中，任何事件统一被处理，从而使编程接口简洁很多，而且内核每次修改的都是pollfd结构体的revents成员，而events成员保持不变，因此下次调用poll时应用程序无需重置pollfd类的事件集参数。poll函数原型如下：

#include <poll.h>
int poll(struct pollfd fd[], nfds_t nfds, int timeout);

struct pollfd
{
　　int fd;          //文件描述符
　　short events;    //请求的事件
　　short revents;   //返回的事件
};

由于每次select和poll调用都返回整个用户注册的事件集合（其中包括就绪的和没就绪的），所以应用程序索引就绪文件描述符的时间复杂度为O(n)。

epoll则采用与select和poll完全不同的方式来管理用户注册的事件。它在内核中维护一个事件表，并提供了一个独立的系统调用epoll_ctl来控制往其中添加、删除、修改事件。epoll_wait函数采用回调函数的方式，当内核检测到就绪的文件描述符时，将触发回调函数，回调函数将文件描述符上对应的事件插入内核就绪事件队列，内核最后在适当的时机将就绪事件队列中的内容拷贝到用户空间，因此epoll_wait无须轮询整个文件描述符集合来检测哪些事件已经就绪，这使得应用程序索引就绪文件描述符的时间复杂度达到了O(1)。

poll和epoll_wait分别用nfds和maxevents参数指定最多监听多少个文件描述符和事件，这两个都能达到65535(cat/proc/sys/fs/filemax)，而select允许监听的最大文件描述符数量通常有限制，修改这个限制会产生不可预料的后果（官方说的）。select和poll都只能工作在相对低效的LT模式，而epoll则可以工作在ET高效模式，并且epoll还支持EPOLLONESHOT事件，该事件能进一步减少可读、可写和异常等事件被处触发的次数。

最后，来个表总结一下3组I/O复用系统调用的区别。