【转载声明】本文为转载文章，原作者「alchemystarlzy」，版权归原作者所有。　原文链接：https://mp.weixin.qq.com/s/A7a_r0MUVOiUAj5jpXckew

公众号：杨建荣的学习笔记 · 作者：alchemystarlzy · 发布：2021-07-17 23:58:17 · 原文链接

从linux源码看epoll

前言

在linux的高性能网络编程中，绕不开的就是epoll。和select、poll等系统调用相比,epoll在需要监视大量文件描述符并且其中只有少数活跃的时候，表现出无可比拟的优势。epoll能让内核记住所关注的描述符，并在对应的描述符事件就绪的时候,在epoll的就绪链表中添加这些就绪元素，并唤醒对应的epoll等待进程。
本文就是笔者在探究epoll源码过程中，对kernel将就绪描述符添加到epoll并唤醒对应进程的一次源码分析(基于linux-2.6.32内核版本)。由于篇幅所限，笔者聚焦于tcp协议下socket可读事件的源码分析。

简单的epoll例子

下面的例子，是从笔者本人用c语言写的dbproxy中的一段代码。由于细节过多，所以做了一些删减。

int init_reactor(int listen_fd,int worker_count){
    ......
    // 创建多个epoll fd，以充分利用多核
    for(i=0;i<worker_count;i++){
        reactor->worker_fd = epoll_create(EPOLL_MAX_EVENTS);
    }
    /* epoll add listen_fd and accept */
    // 将accept后的事件加入到对应的epoll fd中
    int client_fd = accept(listen_fd,(struct sockaddr *)&client_addr,&client_len)));
    // 将连接描述符注册到对应的worker里面
    epoll_ctl(reactor->client_fd,EPOLL_CTL_ADD,epifd,&event);
}
// reactor的worker线程
static void* rw_thread_func(void* arg){
    ......

    for(;;){
          // epoll_wait等待事件触发
        int retval = epoll_wait(epfd,events,EPOLL_MAX_EVENTS,500);
        if(retval > 0){
            for(j=0; j < retval; j++){
                // 处理读事件
               if(event & EPOLLIN){
                 handle_ready_read_connection(conn);
                 continue;
             }
             /* 处理其它事件 */
            }
        }
    }
    ......
}

上述代码事实上就是实现了一个reactor模式中的accept与read/write处理线程，如下图所示:

epoll_create

Unix的万物皆文件的思想在epoll里面也有体现，epoll_create调用返回一个文件描述符，此描述符挂载在anon_inode_fs(匿名inode文件系统)的根目录下面。让我们看下具体的epoll_create系统调用源码:

SYSCALL_DEFINE1(epoll_create, int, size)
{
    if (size <= 0)
        return -EINVAL;

    return sys_epoll_create1(0);
}

由上述源码可见，epoll_create的参数是基本没有意义的，kernel简单的判断是否为0，然后就直接就调用了sys_epoll_create1。由于linux的系统调用是通过(SYSCALL_DEFINE1,SYSCALL_DEFINE2……SYSCALL_DEFINE6)定义的，那么sys_epoll_create1对应的源码即是SYSCALL_DEFINE(epoll_create1)。
(注:受限于寄存器数量的限制，(80×86下的)kernel限制系统调用最多有6个参数。据ulk3所述，这是由于32位80×86寄存器的限制)
接下来，我们就看下epoll_create1的源码:

SYSCALL_DEFINE1(epoll_create1, int, flags)
{
    // kzalloc(sizeof(*ep), GFP_KERNEL),用的是内核空间
    error = ep_alloc(&ep);
    // 获取尚未被使用的文件描述符，即描述符数组的槽位
    fd = get_unused_fd_flags(O_RDWR | (flags & O_CLOEXEC));
    // 在匿名inode文件系统中分配一个inode,并得到其file结构体
    // 且file->f_op = &eventpoll_fops
    // 且file->private_data = ep;
    file = anon_inode_getfile("[eventpoll]", &eventpoll_fops, ep,
                 O_RDWR | (flags & O_CLOEXEC));
    // 将file填入到对应的文件描述符数组的槽里面
    fd_install(fd,file);
    ep->file = file;
    return fd;
}

最后epoll_create生成的文件描述符如下图所示:

struct eventpoll

所有的epoll系统调用都是围绕eventpoll结构体做操作,现简要描述下其中的成员:

/*
 * 此结构体存储在file->private_data中
 */
struct eventpoll {
    // 自旋锁，在kernel内部用自旋锁加锁，就可以同时多线(进)程对此结构体进行操作
    // 主要是保护ready_list
    spinlock_t lock;
    // 这个互斥锁是为了保证在eventloop使用对应的文件描述符的时候，文件描述符不会被移除掉
    struct mutex mtx;
    // epoll_wait使用的等待队列，和进程唤醒有关
    wait_queue_head_t wq;
    // file->poll使用的等待队列，和进程唤醒有关
    wait_queue_head_t poll_wait;
    // 就绪的描述符队列
    struct list_head rdllist;
    // 通过红黑树来组织当前epoll关注的文件描述符
    struct rb_root rbr;
    // 在向用户空间传输就绪事件的时候，将同时发生事件的文件描述符链入到这个链表里面
    struct epitem *ovflist;
    // 对应的user
    struct user_struct *user;
    // 对应的文件描述符
    struct file *file;
    // 下面两个是用于环路检测的优化
    int visited;
    struct list_head visited_list_link;
};

本文讲述的是kernel是如何将就绪事件传递给epoll并唤醒对应进程上，因此在这里主要聚焦于(wait_queue_head_t wq)等成员。

epoll_ctl(add)

我们看下epoll_ctl(EPOLL_CTL_ADD)是如何将对应的文件描述符插入到eventpoll中的。
借助于spin_lock(自旋锁)和mutex(互斥锁),epoll_ctl调用可以在多个KSE(内核调度实体,即进程/线程)中并发执行。

SYSCALL_DEFINE4(epoll_ctl, int, epfd, int, op, int, fd,
        struct epoll_event __user *, event)
{
    /* 校验epfd是否是epoll的描述符 */
    // 此处的互斥锁是为了防止并发调用epoll_ctl,即保护内部数据结构
    // 不会被并发的添加修改删除破坏
    mutex_lock_nested(&ep->mtx, 0);
    switch (op) {
        case EPOLL_CTL_ADD:
            ...
            // 插入到红黑树中
            error = ep_insert(ep, &epds, tfile, fd);
            ...
            break;
        ......
    }
    mutex_unlock(&ep->mtx);
}

上述过程如下图所示：

ep_insert

在ep_insert中初始化了epitem，然后初始化了本文关注的焦点,即事件就绪时候的回调函数，代码如下所示:

static int ep_insert(struct eventpoll *ep, struct epoll_event *event,
             struct file *tfile, int fd)
{
    /* 初始化epitem */
    // &epq.pt->qproc = ep_ptable_queue_proc
    init_poll_funcptr(&epq.pt, ep_ptable_queue_proc);
    // 在这里将回调函数注入
    revents = tfile->f_op->poll(tfile, &epq.pt);
    // 如果当前有事件已经就绪，那么一开始就会被加入到ready list
    // 例如可写事件
    // 另外，在tcp内部ack之后调用tcp_check_space,最终调用sock_def_write_space来唤醒对应的epoll_wait下的进程
    if ((revents & event->events) && !ep_is_linked(&epi->rdllink)) {
        list_add_tail(&epi->rdllink, &ep->rdllist);
        // wake_up ep对应在epoll_wait下的进程
        if (waitqueue_active(&ep->wq)){
            wake_up_locked(&ep->wq);
        }
        ......
    }
    // 将epitem插入红黑树
    ep_rbtree_insert(ep, epi);
    ......
}

tfile->f_op->poll的实现

向kernel更底层注册回调函数的是tfile->f_op->poll(tfile, &epq.pt)这一句，我们来看一下对于对应的socket文件描述符，其fd=>file->f_op->poll的初始化过程：

    // 将accept后的事件加入到对应的epoll fd中
    int client_fd = accept(listen_fd,(struct sockaddr *)&client_addr,&client_len)));
    // 将连接描述符注册到对应的worker里面
    epoll_ctl(reactor->client_fd,EPOLL_CTL_ADD,epifd,&event);

回顾一下上述user space代码，fd即client_fd是由tcp的listen_fd通过accept调用而来，那么我们看下accept调用链的关键路径:

accept
      |->accept4
            |->sock_attach_fd(newsock, newfile, flags & O_NONBLOCK);
                  |->init_file(file,...,&socket_file_ops);
                        |->file->f_op = fop;
                              /* file->f_op = &socket_file_ops */
            |->fd_install(newfd, newfile); // 安装fd

那么，由accept获得的client_fd的结构如下图所示:

(注:由于是tcp socket,所以这边sock->ops=inet_stream_ops,这个初始化的过程在我的另一篇博客<<从linux源码看socket的阻塞和非阻塞>>中，博客地址如下:
https://my.oschina.net/alchemystar/blog/1791017)
既然知道了tfile->f_op->poll的实现，我们就可以看下此poll是如何将安装回调函数的。