Linux 异步 IO 之 epoll

epoll 是 Linux 内核在 2.5 引入的一个 I/O 事件通知机制,用来取代旧的 select(2) 和 poll(2)。

select() 的工作模式

从通知机制上说,select() 的方式和 epoll 的方式类似,都是阻塞在某个函数上,如果有指定的 I/O 事件发生的话函数就会返回。这样的通知机制的好处在于,在没有事件发生的时候系统可以去干别的事情。不过 select() 只告诉你有事件发生了,但是没有说清楚是哪些 fd 触发了事件通知,你需要去遍历所有的 fd 来找出哪些真正需要处理。select() 的函数原型如下:

int select(int nfds, fd_set *readfds, fd_set *writefds,
           fd_set *exceptfds, struct timeval *timeout);

fd_set 是一个 fd 的集合,参数列表中的 readfds,writefds,exceptfds 分别对应需要监控读事件,写事件和异常事件的 fd 集合,如果不需要监控某种类型的事件只需传入 NULL 即可;第一个参数 nfds 是这三个 fd 集合中的最大的 fd 加 1;最后一个参数是超时设置。

对于 fd_set 的操作由以下几个宏完成:

void FD_CLR(int fd, fd_set *set); // 将 fd 从 set 中清除
int  FD_ISSET(int fd, fd_set *set); // 判断 fd 是否在 set 中
void FD_SET(int fd, fd_set *set); // 将 fd 加入 set
void FD_ZERO(fd_set *set); // 清空 set

使用 select() 来监听网络事件的大概处理流程如下:

struct timeval ts;
fd_set readfds, writefds, exceptfds;
int server_fd, clientlist[];
int nfds = MAX_FD + 1;

while (1) {
    FD_ZERO(&readfds);
    FD_ZERO(&writefds);
    FD_ZERO(&exceptfds);

    FD_SET(server_fd, &readfds); // add server socket fd to readfds
    foreach (client in clientlist[]) {
        FD_SET(client, &readfds);
    }

    res = select(nfds, &readfds, &writefds, &exceptfds, &ts);
    if (res == -1)
        continue;

    if (FD_ISSET(server_fd, &readfds)) {
        client = accept(server_fd, NULL, NULL);
        clientlist[] = client; // add client to clientlist
    }

    foreach (client in clientlist[]) {
        if (FD_ISSET(client, &readfds))
            // do something
    }
}

从伪代码中可以看到,每次循环开始的时候都需要清空 fd_set,然后把需要监听的 fd 重新加入到对应的 fd_set 中;select() 返回后,需要遍历要监听的 client,判断每个 client 是否在对应事件的 fd_set 中。如果连接很多但是每次活动的 client 很少,这样其实大部分的比较都是不需要的。

epoll 模型

与 select() 不同的地方是,epoll 除了告诉用户有事件发生外,还告诉用户到底是哪些 fd 发生了哪些事件,这样就避免了对整个监控的 fd 集合进行遍历。在一些需要监控大量 fd,但是每次只有少量 fd 触发事件的情形下能大大提高效率。

epoll 相关的函数原型都在 <sys/epoll.h> 中定义。要使用 epoll,首先需要调用

int epoll_create(int size);

创建一个 epoll。参数 size 指定要监控事件的最大值,但是从内核 2.6.8 开始只要求 size 大于 0 就可以了。如果创建成功则会返回一个与 epoll 相关的 fd,后续对 epoll 的操作都通过这个 fd 完成。

要对某个 fd 的事件(例如读或写,或者网络连接是否被关闭)进行监听,使用函数

int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);

第一个参数就是 epoll_create() 的返回值;第三个参数是要进行操作的 fd;第二个参数是操作的类型,可选的参数有

  • EPOLL_CTL_ADD:将 fd 加入到 epfd 的监听列表中,要监听的事件类型由第四个参数指定;
  • EPOLL_CTL_MOD:将 fd 的相关事件修改为第四个参数指定的事件;
  • EPOLL_CTL_DEL:将 fd 从 epfd 中移除,这时第四个参数可以为 NULL (2.6.9 之前的内核要求不能为空)。

第四个参数是一个结构体 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 */
};

epoll_data_t 是一个 union,允许用户保存一些与该 fd 相关的内容,以便在事件被触发的时候能够获得相关的上下文信息,里面的字段取值没有要求;events 是一个 bitmap,用来保存要对 fd 进行监控的事件类型:

  • EPOLLIN:可以使用 read() 从 fd 读取数据;
  • EPOLLOUT:可以使用 write() 往 fd 写入数据;
  • EPOLLRDHUP(从 2.6.17 开始可用):对方(socket fd)关闭连接;
  • EPOLLPRI:fd 有紧急的数据需要读取(?不是很明白);
  • EPOLLERR:fd 出错(默认会带上这个事件,不需要显式指定);
  • EPOLLHUP:fd 连接中断(有可能是网络出错或其它异常事件,默认会带上这个事件,不需要显式指定);
  • EPOLLET:将 fd 的模式设为边缘触发(见后面的解释),如果不指定这个选项,默认的模式是水平触发;
  • EPOLLONESHOT(2.6.2 版本后可用):对指定的事件类型只触发一次,后面就不会再触发了,用户需要使用 epoll_ctl() 来修改对应 fd 的 event mask。

当向 epoll 中添加了需要监控的 fd 及对应的事件后,就可以使用

int epoll_wait(int epfd, struct epoll_event *events,
               int maxevents, int timeout);

来等待被监控的 fd 是否有指定的事件发生。如果有事件发生,和 fd 关联的 epoll_event 结构体会被放到第二个参数指向的数组中;第三个参数指定数组中最多可容纳多少个事件;第四个参数指定超时间隔,如果值为 -1 表示 epoll 使用阻塞模式(即一直等待,直到有事件发生才返回),值为 0 表示使用非阻塞模式,即无论有没有事件发生都立即返回。

从上面的描述,我们可以得到使用 epoll 的一般步骤为:

epfd = epoll_create(MAX_EVENTS);

event.data.fd = fd;
event.events = EPOLLIN; // and other flag(s)
epoll_ctl(epfd, EPOLL_CTL_ADD, fd, &event);

while (1) {
    nr_fds = epoll_wait(epfd, eventlist, MAX_EVENTS, -1);
    for (i = 0; i < nr_fds; ++i)
        // do something with eventlist[i]
}

当不需要 epoll 的时候使用 close() 来销毁,这个和一般的文件操作是一样的。

epoll 的工作模式

系统调用 read() 有两种模式,分别是阻塞模式和非阻塞模式。默认使用的是阻塞模式,即调用 read() 的时候会阻塞,直到读到数据或出错才会返回,如果没有数据可读函数就会一直等待;另一种模式是非阻塞模式,即读取数据的时候无论有没有数据都立即返回,调用者通过 errno 的值判断读取是否成功。epoll 也有两种工作模式:水平触发(Level-Triggered,LT)和边缘触发(Edge-Triggered,ET),前者可用于阻塞和非阻塞模式,而后者只能用于非阻塞模式。参考资料 [2] 中举了个例子说明这两种工作模式的区别。

假设有一个管道,在一端写入数据,另一端读取数据,下面的事情依次发生:

  1. 向 epoll 添加了一个新的 fd;
  2. 其中一个 fd 的一端写入了 2 kB 的数据;
  3. epoll_wait() 监控到了这个事件并返回对应的 epoll_event;
  4. 从 fd 的读取端读了 1 kB 数据;
  5. 再次调用 epoll_wait()。

如果使用了边缘触发,在第 5 步可能会被阻塞(状态并没有发生变化,仍然处于有数据的状态),这样尽管 fd 中还有数据没有读完,但是 epoll_wait() 并不会再次触发 fd;如果使用水平触发,则在第 5 步调用 epoll_wait() 还会继续返回 fd 对应的 epoll_event(因为满足“可读”这个条件,所以事件还会被触发)。或者可以这么理解:边缘触发是根据边界条件来决定是否触发事件(即只有从无数据变为有数据这样一个跃迁状态的变化才会触发事件),而水平触发则是根据数据量的变化(假设数据的水平线是 0,如果未读数据量大于 0 则超出水平线,这时就要触发事件)来决定。

由于边缘触发只需要发送一次通知,不需要多次触发 epoll_wait(),效率较高。要使用边缘触发,建议的做法是把要监控的 fd 设置为非阻塞模式,这样可以避免在读数据的时候因为无数据可读而阻塞,从而影响了 epoll_wait() 返回的其它事件的处理;另外在调用 read()/write() 时要判断 errno 为 EAGAIN(表示无数据可读或不能写入数据)时才停止读写。

其它

如果加入 epoll 的 fd 不是通过 dup(2), dup2(2), fcntl(2) F_DUPFD, or fork(2) 复制的,当这个 fd 被关闭后会从 epoll 中删除。epoll 不是根据 fd 而是根据底层的文件描述符来标识打开的文件(参考资料 [2] 中最后的 FAQ 有说明)。只有当这个文件描述符的引用计数为 0 后才可能从 epoll 中删除。

最后是一个使用 epoll 实现的服务端例子,使用了非阻塞模式,将收到的数据打印出来。

#include <stdio.h>
#include <errno.h>
#include <string.h>
#include <unistd.h>

#include <fcntl.h>
#include <netdb.h>
#include <sys/epoll.h>

#define MAX_EVENTS 10

static int tcp_server_init(const char* addr, unsigned port)
{
    int fd, option;
    char buf[8];
    struct addrinfo hints, *res;

    if (!addr)
        addr = "0.0.0.0";

    memset(&hints, 0, sizeof(struct addrinfo));
    hints.ai_family = AF_UNSPEC;
    hints.ai_socktype = SOCK_STREAM;
    hints.ai_flags = AI_PASSIVE;
    sprintf(buf, "%d", port);

    fd = getaddrinfo(addr, buf, &hints, &res);
    if (fd != 0) {
        fprintf(stderr, "%s: getaddrinfo() error: %s\n",
                __func__, gai_strerror(fd));
        return -1;
    }

    fd = socket(res->ai_family, res->ai_socktype, res->ai_protocol);
    if (fd == -1) {
        fprintf(stderr, "%s: socket() error: %s\n",
                __func__, strerror(errno));
        goto err1;
    }

    option = 1;
    setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &option, sizeof(int));

    if (bind(fd, res->ai_addr, res->ai_addrlen) != 0) {
        fprintf(stderr, "%s: bind() error: %s\n",
                __func__, strerror(errno));
        goto err2;
    }

    if (listen(fd, 0) == -1) {
        fprintf(stderr, "%s: listen() error: %s\n",
                __func__, strerror(errno));
        goto err2;
    }

    freeaddrinfo(res);
    return fd;

err2:
    close(fd);
err1:
    freeaddrinfo(res);
    return -1;
}

static inline void setnonblocking(int sockfd)
{
    int opt;

    opt = fcntl(sockfd, F_GETFL);
    if (opt < 0) {
        fprintf(stderr, "fcntl(F_GETFL) failed.\n");
        return;
    }

    opt |= O_NONBLOCK;
    if (fcntl(sockfd, F_SETFL, opt) < 0)
        fprintf(stderr, "fcntl(F_SETFL) failed.\n");
}

int main(void)
{
    struct epoll_event ev, events[MAX_EVENTS];
    int sockfd, efd;

    sockfd = tcp_server_init(NULL, 12345);
    if (sockfd == -1) {
        perror("tcp_server_init");
        return 0;
    }
    fprintf(stderr, "server fd = %d.\n", sockfd);


    efd = epoll_create(MAX_EVENTS);
    if (efd == -1) {
        perror("epoll_create");
        goto end;
    }
    fprintf(stderr, "epoll_create ok, epollfd = %d.\n", efd);

    ev.events = EPOLLIN;
    ev.data.fd = sockfd;
    if (epoll_ctl(efd, EPOLL_CTL_ADD, sockfd, &ev) == -1) {
        perror("epoll_ctl");
        goto err;
    }
    fprintf(stderr, "epoll ctl ok.\n");

    while (1) {
        int nfds, i;

        nfds = epoll_wait(efd, events, MAX_EVENTS, -1);
        fprintf(stderr, "\nepoll_wait returns %d event(s)\n", nfds);

        for (i = 0; i < nfds; ++i) {
            if (events[i].data.fd == sockfd) {
                int fd = accept(sockfd, NULL, NULL);
                if (fd == -1) {
                    perror("accept");
                    goto err;
                }
                setnonblocking(fd);

                ev.events = EPOLLIN | EPOLLHUP | EPOLLRDHUP | EPOLLET;
                ev.data.fd = fd;
                if (epoll_ctl(efd, EPOLL_CTL_ADD, fd, &ev) == -1) {
                    perror("epoll_ctl");
                    goto err;
                }

                fprintf(stderr, "get client fd = %d.\n", fd);
            } else if ((events[i].events & EPOLLHUP) || (events[i].events & EPOLLRDHUP)) {
                fprintf(stderr, "client disconnected. fd %d removed from epoll\n", events[i].data.fd);
                close(events[i].data.fd);
            } else {
                int nbytes;
                int wanted = 5;
                char buf[BUFSIZ];

                nbytes = read(events[i].data.fd, buf, wanted);
                while (nbytes > 0) {
                    fprintf(stderr, "get data -> ");
                    buf[nbytes] = '\n';
                    write(2, buf, nbytes + 1);
                    nbytes = read(events[i].data.fd, buf, wanted);
                }

                if (nbytes == -1) {
                    if (errno == EAGAIN)
                        fprintf(stderr, "no more data\n");
                    else
                        fprintf(stderr, "errno = %d.\n", errno);
                } else if (nbytes == 0) {
                    fprintf(stderr, "client disconnected. fd %d removed from epoll\n", events[i].data.fd);
                    close(events[i].data.fd);
                } else {
                    fprintf(stderr, "nbytes = %d, errno = %d.\n", nbytes, errno);
                }
            }
        }
    }

err:
    close(efd);
end:
    close(sockfd);

    return 0;
}

参考资料

[1] select(2)
[2] epoll(7)

发表回复

您的电子邮箱地址不会被公开。 必填项已用 * 标注