论epoll的实现

上一篇博客 论select的实现 里面已经说了为什么 select 比较慢。poll 的实现和 select 类似,只是少了最大 fd 限制,如果有兴趣可以自己去看代码。我这里来简单来过一下 epoll 的实现。

1) 一次添加

select / poll 为了实现简单,不对已有的 fd 进行管理。每次需要传入最大的轮询 fd, 然后每个监听 fd 挂到设备一次导致性能不佳。epoll 对于监听的 fd,通过 epoll_ctl 来把对应的 fd 添加到红黑树,实现快速的查询和添加,删除。

1.1) epoll 实例创建

使用 epoll 之前会使用 epoll_create 创建一个 epoll 实例,它实际上是一个文件, 只是存在于内存中的文件。下面实现来自 linux-2.6.24 的 fs/eventpoll.c:

asmlinkage long sys_epoll_create(int size) {
    ...
    struct eventpoll *ep;
    ...
    // 创建 eventpoll 实例
    if (size <= 0 || (error = ep_alloc(&ep)) != 0)
        goto error_return;
    ...
    // 为 epoll 文件添加文件操作函数
    error = anon_inode_getfd(&fd, &inode, &file, "[eventpoll]",
                 &eventpoll_fops, ep);
}

epoll_create 的参数 size 是老版本的实现,使用的是 hash 表, size 应该是用来算 bucket 数目,后面因为使用红黑树,这个参数不再使用, 可以忽略。

1.2) 添加 fd

asmlinkage long sys_epoll_ctl(int epfd, int op, int fd,
                  struct epoll_event __user *event) {
    ...
    // 先查找 fd 是否已经存在
    epi = ep_find(ep, tfile, fd);

    error = -EINVAL;
    switch (op) {
    // 如果是添加,就插入到 eventpoll 实例的红黑树
    case EPOLL_CTL_ADD:
        if (!epi) {
            epds.events |= POLLERR | POLLHUP;

            // 添加监听的 fd 到epoll
            error = ep_insert(ep, &epds, tfile, fd);
        } else
            error = -EEXIST;
        break;
        ...
    }
}

接着调用 ep_insert 是添加 fd 到红黑树以及把进程的回调函数添加文件句柄的监听队列,当有事件到来时,会唤醒进程。

static int ep_insert(struct eventpoll *ep, struct epoll_event *event,
             struct file *tfile, int fd)
{
    ...
    // 创建 epitem 并设置回调函数 ep_ptable_queue_proc
    init_poll_funcptr(&epq.pt, ep_ptable_queue_proc);
    ...
    // 这里会回调 ep_ptable_queue_proc, 并查询 fd 可读写状态
    revents = tfile->f_op->poll(tfile, &epq.pt);
    ...
    // epitem 添加到 eventpoll 的红黑树
    ep_rbtree_insert(ep, epi);
    ...
}

ep_ptable_queue_proc 这个回调函数, 除了把进程添加文件句柄的监听列表,并注册回调函数为 ep_poll_callback。 这个函数会查询 fd 的读写状态, 如果当前文件句柄可以读写,就把当前的 fd 添加到就绪队列。后续查询是否有 fd 可以读写,只要只要拷贝这个就绪列表,不用查询。我们下面会来看看 epoll_wait 的实现。

2) 快速查询

epoll 之所以快,除了没有多次重复挂载事件之外,在有读写事件到来的实现,也是很高效。没有像 select/poll 那样, 需要轮询所有的fd, 才能知道哪些 fd 有事件需要处理。

epoll 有一个专门的链表用来存放哪些 fd 有事件到来,用户空间需要查询哪些 fd 有读写等待处理,只需要拷贝这个链表即可。

我们使用系统调用 epoll_wait, 会到内核调用 sys_epoll_wait, 这个函数的主要实现就是调用了 ep_poll:

static int ep_poll(struct eventpoll *ep, struct epoll_event __user *events,
           int maxevents, long timeout)
{
    ...
    res = 0;
    if (list_empty(&ep->rdllist)) {
        // 如果没有事件, 不断等待读写事件到来
        // 直到超时,或者有读写事件
        for (;;) {
            /*
             * We don't want to sleep if the ep_poll_callback() sends us
             * a wakeup in between. That's why we set the task state
             * to TASK_INTERRUPTIBLE before doing the checks.
             */
            // 当前进程设置为可中断
            set_current_state(TASK_INTERRUPTIBLE);
            // 有连接就绪
            if (!list_empty(&ep->rdllist) || !jtimeout)
                break;
            if (signal_pending(current)) {
                res = -EINTR;
                break;
            }
        }
    }

    //如果 rdllist 不为空, 说明有事件到来。
    eavail = !list_empty(&ep->rdllist);

    spin_unlock_irqrestore(&ep->lock, flags);

    // 拷贝到用户空间
    if (!res && eavail &&
        !(res = ep_send_events(ep, events, maxevents)) && jtimeout)
        goto retry;

     return res;
}

我们可以看到,epoll 在实现 epoll_wait的时候,并不会去查询 fd 的可读写状态。 而是等待 fd 有读写到来时, 通过回调函数把有事件到来的 fd 主动拷贝到 rdllist。

另外一个有一个细节点就是, 当用户在拷贝事件到用户空间时,刚好有事件到来,那么这些读写事件会不会正好丢了。答案是当然不会,epoll 准备了另外一个链表,叫 overflow list, 当检查正在拷贝时,会把这些 fd 临时放到这个链表,下次再拷贝到 rdllist.

3) 总结

select/poll 还有比较坑的是,每次查询到 fd 读写事件结果之后,需要把所有 fd 对应的结果的 bitmap 拷贝到用户空间。 比如监听 100w 的 fd, 只有一个 fd 有读写事件, 却要拷贝 100w fd的结果 bitmap。

对比来看,select/poll 实现极为简单,但并不适合用来维护大量的连接。