epoll简介

Posted 2020-09-01 Sawyer Ford

Linux下谈论I/O复用、高并发，一定会说到epoll。因为epoll是最有效的I/O复用方式。

epoll的使用非常简单，总共3个API：

// 创建epoll对象
int epoll_create(int size);

Linux2.6.8之后，size参数已被忽略，为了向前兼容，size大于0即可。

// 向epoll对象中添加、修改或删除事件
int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);

op有3种取值：EPOLL_CTL_ADD、EPOLL_CTL_MOD、EPOLL_CTL_DEL。

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_event中events取值意义如下：

关于epoll的两种触发方式：水平触发(Level-triggered, LT)、边缘触发(Edge-triggered, ET)

    LT模式下，只要一个fd上的事件一次没有处理完，下次调用epoll_wait还会返回这个fd，而在ET模式下，仅第一次返回这个fd。

 

假设epoll监听一个缓冲区的读事件，然后发生如下事件：

• 缓冲区被写入2KB数据
• epoll_wait返回
• 程序只读取了1KB数据

如果在LT模式下，下次epoll_wait还会返回读事件；如果在ET模式下，epoll_wait不会返回。

为了避免epoll_wait永久阻塞，在ET模式下一定要使用非阻塞套接字。

 

// 等待epoll对象上I/O事件的发生
int epoll_wait(int epfd, struct epoll_event *events,
                      int maxevents, int timeout);

就绪的事件会被保存在events数组中，最多返回maxevents个事件。

timeout对应等待时间，超时函数返回。如果设为-1，则永久等待。

具体事例：

#define MAX_EVENTS 10
struct epoll_event ev, events[MAX_EVENTS];
int listen_sock, conn_sock, nfds, epollfd;

/* Code to set up listening socket, \'listen_sock\',
 (socket(), bind(), listen()) omitted */

epollfd = epoll_create(MAX_EVENTS);
if (epollfd == -1) {
    perror("epoll_create1");
    exit(EXIT_FAILURE);
}

ev.events = EPOLLIN;
ev.data.fd = listen_sock;
if (epoll_ctl(epollfd, EPOLL_CTL_ADD, listen_sock, &ev) == -1) {
    perror("epoll_ctl: listen_sock");
    exit(EXIT_FAILURE);
}

for (;;) {
    nfds = epoll_wait(epollfd, events, MAX_EVENTS, -1);
    if (nfds == -1) {
        perror("epoll_wait");
        exit(EXIT_FAILURE);
    }
    
    for (n = 0; n < nfds; ++n) {
        if (events[n].data.fd == listen_sock) {
            conn_sock = accept(listen_sock,
                               (struct sockaddr *) &addr, &addrlen);
            if (conn_sock == -1) {
                perror("accept");
                exit(EXIT_FAILURE);
            }
            setnonblocking(conn_sock);
            ev.events = EPOLLIN | EPOLLET;
            ev.data.fd = conn_sock;
            if (epoll_ctl(epollfd, EPOLL_CTL_ADD, conn_sock,
                          &ev) == -1) {
                perror("epoll_ctl: conn_sock");
                exit(EXIT_FAILURE);
            }
        } else {
            do_use_fd(events[n].data.fd);
        }
    }
}

View Code

 

最后谈谈epoll的实现。

每个epoll对象对应一个eventpoll结构体：

struct eventpoll {
     /* 红黑树的根结点，这棵树中存储着所有添加到epoll中的事件，也就是这个epoll监控的事件 */
    struct rb_root rbr;
    /*双向链表rdllist保存着将要通过epoll_wait返回给用户的、满足条件的事件*/
    struct list_head rdllist;
};

通过epoll_ctl添加事件时，发生两件事：

1. 将事件添加到红黑树上

2. 为事件注册回调函数(一旦事件发生，就把事件添加到链表中)

epoll_wait通过检查链表是否为空确定是否有事件就绪。如果不为空，就返回就绪的事件。

 

epoll的实现带来了两个优点：

1. 避免了在用户态和内核态之间大量拷贝描述符

2. 使用事件驱动代替轮询确定已经就绪的描述符

 

参考资料:

我读过的最好的epoll讲解--转自”知乎“

linux下epoll如何实现高效处理百万句柄的

http://man7.org/linux/man-pages/man7/epoll.7.html