io复用之epoll核心源码剖析-阿里云开发者社区

绪论

linux epoll主要函数只有三个,分别为:

epoll_create:创建epollpoll对象并初始化
epoll_ctl:操作epollooll对象,增加,修改,删除
epoll_wait:在epollpoll对象中返回活跃的事件

操作系统内部会用到一个名叫epoll_event_callback()的回调函数来调度epoll对象中的事件，在网络中收到数据时就会调用

源码来源

由于epoll的实现内嵌在内核中，直接查看内核源码的话会有一些无关代码影响阅读。这里有一份王靖博老师写的ntytcp开源库,里面有实现了epoll逻辑。链接为：https://github.com/wangbojing/NtyTcp

存放着以上4个关键函数的文件和头文件是[src\nty_epoll_rb.c和include/nty_epoll_inner.h]

核心数据结构

1.epitem

在这里插入图片描述

如图所示，epitem是中包含了两个主要的成员变量，分别是rbn和rdlink，前者是红黑树的节点，而后者是双链表的节点，也就是说一个epitem对象即可作为红黑树中的一个节点又可作为双链表中的一个节点。并且每个epitem中存放着一个event，对event的查询也就转换成了对epitem的查询。

一个结构体设计成共用类型,设计很巧妙

//这是个节点相关的结构
//作为红黑树的一个节点
struct epitem {
   
   
    RB_ENTRY(epitem) rbn;
    /*  RB_ENTRY相当如定义了如下的一个结构成员变量
    struct {                                            
    struct type *rbe_left;        //指向左子树
    struct type *rbe_right;        //指向右子树
    struct type *rbe_parent;    //指向父节点
    int rbe_color;                //该红黑树节点颜色
    } rbn*/

    LIST_ENTRY(epitem) rdlink;
    /*
    struct {                                    
        struct type *le_next;    //指向下个元素
        struct type **le_prev;    //前一个元素的地址
    }*/

    int rdy; //exist in list 是否这个节点是同时在双向链表中【这个节点刚开始是在红黑树中】

    int sockfd;
    struct epoll_event event; 
};

2.epollpoll

ep_rb_tree rbr:这个对象就是epoll红黑树的根节点,存储每一个通过epoll_ctl->epoll_ctl_add添加进来的socket对象

LIST_HEAD( ,epitem) rdlist:双向链表的头节点,存储所有有事件发生的节点

在这里插入图片描述

//调用epoll_create()的时候我们会创建这个结构的对象
struct eventpoll {
   
   
    ep_rb_tree rbr;      //ep_rb_tree是个结构，所以rbr是结构变量，这里代表红黑树的根；
    int rbcnt;            //红黑树中节点的数量（也就是添加了多少个TCP连接事件）

    LIST_HEAD( ,epitem) rdlist;    //rdlist是结构变量，这里代表双向链表的根；
    /*    这个LIST_HEAD等价于下边这个 
        struct {
            struct epitem *lh_first;
        }rdlist;
    */
    int rdnum; //双向链表里边的节点数量（也就是有多少个TCP连接来事件了）

    int waiting;

    pthread_mutex_t mtx; //rbtree update
    pthread_spinlock_t lock; //rdlist update

    pthread_cond_t cond; //block for event
    pthread_mutex_t cdmtx; //mutex for cond

};

四个关键函数

1.epoll_create()

对以上代码的逻辑进行梳理，可以总结为以下6步：

创建eventpoll对象
让eventpoll中的rbr指向空
让eventpoll中的rdlist指向空
在并发环境下进行互斥
保存eventpoll对象
返回eventpoll对象的句柄(id)

//创建epoll对象，创建一颗空红黑树，一个空双向链表
int epoll_create(int size) {
   
   

    //最初的epoll_create的size参数是指定最大socket数量,后来改成非负数就行
    if (size <= 0) return -1;

    nty_tcp_manager *tcp = nty_get_tcp_manager();//获取tcp对象
    if (!tcp) return -1;

    struct _nty_socket *epsocket = nty_socket_allocate(NTY_TCP_SOCK_EPOLL);
    if (epsocket == NULL) {
   
   
        nty_trace_epoll("malloc failed\n");
        return -1;
    }

    //(1)相当于new了一个eventpoll对象【开辟了一块内存】
    struct eventpoll *ep = (struct eventpoll*)calloc(1, sizeof(struct eventpoll)); //参数1：元素数量 ，参数2：每个元素大小
    if (!ep) {
   
   
        nty_free_socket(epsocket->id, 0);
        return -1;
    }

    ep->rbcnt = 0;

    //(2)让红黑树根节点指向一个空
    RB_INIT(&ep->rbr);       //等价于ep->rbr.rbh_root = NULL;

    //(3)让双向链表的根节点指向一个空
    LIST_INIT(&ep->rdlist);  //等价于ep->rdlist.lh_first = NULL;

    // 4° 并发环境下进行互斥
    if (pthread_mutex_init(&ep->mtx, NULL)) {
   
   
        free(ep);
        nty_free_socket(epsocket->id, 0);
        return -2;
    }

    if (pthread_mutex_init(&ep->cdmtx, NULL)) {
   
   
        pthread_mutex_destroy(&ep->mtx);
        free(ep);
        nty_free_socket(epsocket->id, 0);
        return -2;
    }

    if (pthread_cond_init(&ep->cond, NULL)) {
   
   
        pthread_mutex_destroy(&ep->cdmtx);
        pthread_mutex_destroy(&ep->mtx);
        free(ep);
        nty_free_socket(epsocket->id, 0);
        return -2;
    }

    if (pthread_spin_init(&ep->lock, PTHREAD_PROCESS_SHARED)) {
   
   
        pthread_cond_destroy(&ep->cond);
        pthread_mutex_destroy(&ep->cdmtx);
        pthread_mutex_destroy(&ep->mtx);
        free(ep);

        nty_free_socket(epsocket->id, 0);
        return -2;
    }

    //5° 保存epoll对象
    tcp->ep = (void*)ep;
    epsocket->ep = (void*)ep;

    return epsocket->id;
}

(2)epoll_ctl()

该函数的逻辑其实很简单，无非就是将用户传入的参数封装为一个epitem对象，然后根据传入的op是①EPOLL_CTL_ADD、②EPOLL_CTL_MOD还是③EPOLL_CTL_DEL，来决定是①将epitem对象插入红黑树中，②更新红黑树中的epitem对象，还是③移除红黑树中的epitem对象。

//往红黑树中加每个tcp连接以及相关的事件
/*该函数的逻辑其实很简单，无非就是将用户传入的参数封装为一个epitem对象，
然后根据传入的op是①EPOLL_CTL_ADD、②EPOLL_CTL_MOD还是③EPOLL_CTL_DEL，
来决定是①将epitem对象插入红黑树中，②更新红黑树中的epitem对象，还是③移除红黑树中的epitem对象。
*/
int epoll_ctl(int epid, int op, int sockid, struct epoll_event *event) {
   
   

    nty_tcp_manager *tcp = nty_get_tcp_manager();
    if (!tcp) return -1;

    nty_trace_epoll(" epoll_ctl --> 1111111:%d, sockid:%d\n", epid, sockid);
    struct _nty_socket *epsocket = tcp->fdtable->sockfds[epid];
    //struct _nty_socket *socket = tcp->fdtable->sockfds[sockid];

    //nty_trace_epoll(" epoll_ctl --> 1111111:%d, sockid:%d\n", epsocket->id, sockid);
    if (epsocket->socktype == NTY_TCP_SOCK_UNUSED) {
   
   
        errno = -EBADF;
        return -1;
    }

    if (epsocket->socktype != NTY_TCP_SOCK_EPOLL) {
   
   
        errno = -EINVAL;
        return -1;
    }

    nty_trace_epoll(" epoll_ctl --> eventpoll\n");

    struct eventpoll *ep = (struct eventpoll*)epsocket->ep;
    if (!ep || (!event && op != EPOLL_CTL_DEL)) {
   
   
        errno = -EINVAL;
        return -1;
    }

    if (op == EPOLL_CTL_ADD) {
   
   
        //添加sockfd上关联的事件
        pthread_mutex_lock(&ep->mtx);

        struct epitem tmp;
        tmp.sockfd = sockid;
        struct epitem *epi = RB_FIND(_epoll_rb_socket, &ep->rbr, &tmp); //先在红黑树上找，根据key来找，也就是这个sockid，找的速度会非常快
        if (epi) {
   
   
            //原来有这个节点，不能再次插入
            nty_trace_epoll("rbtree is exist\n");
            pthread_mutex_unlock(&ep->mtx);
            return -1;
        }

        //只有红黑树上没有该节点【没有用过EPOLL_CTL_ADD的tcp连接才能走到这里】；

        //(1)生成了一个epitem对象，大家注意这个结构epitem，这个结构对象，其实就是红黑的一个节点，也就是说，红黑树的每个节点都是 一个epitem对象；
        epi = (struct epitem*)calloc(1, sizeof(struct epitem));
        if (!epi) {
   
   
            pthread_mutex_unlock(&ep->mtx);
            errno = -ENOMEM;
            return -1;
        }

        //(2)把socket(TCP连接)保存到节点中；
        epi->sockfd = sockid;  //作为红黑树节点的key，保存在红黑树中

        //(3)我们要增加的事件也保存到节点中；
        memcpy(&epi->event, event, sizeof(struct epoll_event));

        //(4)把这个节点插入到红黑树中去
        epi = RB_INSERT(_epoll_rb_socket, &ep->rbr, epi); //实际上这个时候epi的rbn成员就会发挥作用，如果这个红黑树中有多个节点，那么RB_INSERT就会epi->rbi相应的值：可以参考图来理解
        assert(epi == NULL);
        ep->rbcnt ++;//红黑树节点数量+1

        pthread_mutex_unlock(&ep->mtx);

    } else if (op == EPOLL_CTL_DEL) {
   
   
        //把红黑树节点从红黑树上删除
        pthread_mutex_lock(&ep->mtx);

        struct epitem tmp;
        tmp.sockfd = sockid;

        struct epitem *epi = RB_FIND(_epoll_rb_socket, &ep->rbr, &tmp);//先在红黑树上找，根据key来找，也就是这个sockid，找的速度会非常快
        if (!epi) {
   
   //如果没找到就退出吧
            nty_trace_epoll("rbtree no exist\n");
            pthread_mutex_unlock(&ep->mtx);
            return -1;
        }

        //只有在红黑树上找到该节点【用过EPOLL_CTL_ADD的tcp连接才能走到这里】；

        //(1)从红黑树上把这个节点干掉
        epi = RB_REMOVE(_epoll_rb_socket, &ep->rbr, epi);
        if (!epi) {
   
   
            nty_trace_epoll("rbtree is no exist\n");
            pthread_mutex_unlock(&ep->mtx);
            return -1;
        }

        ep->rbcnt --;//红黑树节点自减
        free(epi);//释放节点内存

        pthread_mutex_unlock(&ep->mtx);

    } else if (op == EPOLL_CTL_MOD) {
   
   
        //修改红黑树某个节点的内容
        struct epitem tmp;
        tmp.sockfd = sockid;
        struct epitem *epi = RB_FIND(_epoll_rb_socket, &ep->rbr, &tmp); //先在红黑树上找，根据key来找，也就是这个sockid，找的速度会非常快
        if (epi) {
   
   
            //(1)红黑树上有该节点，则修改对应的事件
            epi->event.events = event->events;
            epi->event.events |= EPOLLERR | EPOLLHUP;
        } else {
   
   
            errno = -ENOENT;
            return -1;
        }

    } else {
   
   
        nty_trace_epoll("op is no exist\n");
        assert(0);
    }

    return 0;
}

(3)epoll_wait()

该函数的逻辑也十分简单，就是让先看一下eventpoll对象的双链表中是否有节点。如果有节点的话则取出节点中的事件填充到用户传入的指针所指向的内存中。如果没有节点的话，则在while循环中等待一定时间，直到有事件被触发后操作系统会将epitem插入到双向链表上使得rdnum>0时(这个过程是由操作系统调用epoll_event_callback函数完成的)，程序才会跳出while循环，去双向链表中取数据。

//到双向链表中去取相关的事件通知
int epoll_wait(int epid, struct epoll_event *events, int maxevents, int timeout) {
   
   

    nty_tcp_manager *tcp = nty_get_tcp_manager();
    if (!tcp) return -1;

    //nty_socket_map *epsocket = &tcp->smap[epid];
    struct _nty_socket *epsocket = tcp->fdtable->sockfds[epid];
    if (epsocket == NULL) return -1;

    if (epsocket->socktype == NTY_TCP_SOCK_UNUSED) {
   
   
        errno = -EBADF;
        return -1;
    }

    if (epsocket->socktype != NTY_TCP_SOCK_EPOLL) {
   
   
        errno = -EINVAL;
        return -1;
    }

    struct eventpoll *ep = (struct eventpoll*)epsocket->ep;
    // ...此处主要是一些负责验证性工作的代码...
    if (!ep || !events || maxevents <= 0) {
   
   
        errno = -EINVAL;
        return -1;
    }

    if (pthread_mutex_lock(&ep->cdmtx)) {
   
   
        if (errno == EDEADLK) {
   
   
            nty_trace_epoll("epoll lock blocked\n");
        }
        assert(0);
    }

    //(1)这个while用来等待一定的时间【在这段时间内，发生事件的TCP连接，相关的节点，会被操作系统扔到双向链表去【当然这个节点同时也在红黑树中呢】】
    while (ep->rdnum == 0 && timeout != 0) {
   
   //双向链表里面的节点数量等于零并且尚未超时的时候就会while循环
        // ...此处主要是一些与等待时间相关的代码...
        ep->waiting = 1;
        if (timeout > 0) {
   
   

            struct timespec deadline;

            clock_gettime(CLOCK_REALTIME, &deadline);
            if (timeout >= 1000) {
   
   
                int sec;
                sec = timeout / 1000;
                deadline.tv_sec += sec;
                timeout -= sec * 1000;
            }

            deadline.tv_nsec += timeout * 1000000;

            if (deadline.tv_nsec >= 1000000000) {
   
   
                deadline.tv_sec++;
                deadline.tv_nsec -= 1000000000;
            }

            int ret = pthread_cond_timedwait(&ep->cond, &ep->cdmtx, &deadline);
            if (ret && ret != ETIMEDOUT) {
   
   
                nty_trace_epoll("pthread_cond_timewait\n");

                pthread_mutex_unlock(&ep->cdmtx);

                return -1;
            }
            timeout = 0;
        } else if (timeout < 0) {
   
   //如果超时了还没有事件发生就直接返回-1

            int ret = pthread_cond_wait(&ep->cond, &ep->cdmtx);
            if (ret) {
   
   
                nty_trace_epoll("pthread_cond_wait\n");
                pthread_mutex_unlock(&ep->cdmtx);

                return -1;
            }
        }
        ep->waiting = 0; 

    }

    pthread_mutex_unlock(&ep->cdmtx);

    //等一小段时间，等时间到达后，流程来到这里。。。。。。。。。。。。。。

    pthread_spin_lock(&ep->lock);

    int cnt = 0;

    //(1)取得事件的数量
    //ep->rdnum：代表双向链表里边的节点数量（也就是有多少个TCP连接来事件了）
    //maxevents：此次调用最多可以收集到maxevents个已经就绪【已经准备好】的读写事件
    int num = (ep->rdnum > maxevents ? maxevents : ep->rdnum); //哪个数量少，就取得少的数字作为要取的事件数量
    int i = 0;

    while (num != 0 && !LIST_EMPTY(&ep->rdlist)) {
   
    //EPOLLET,事件数量和双向链表节点不为零

        //(2)每次都从双向链表头取得 一个一个的节点
        struct epitem *epi = LIST_FIRST(&ep->rdlist);

        //(3)把这个节点从双向链表中删除【但这并不影响这个节点依旧在红黑树中】
        LIST_REMOVE(epi, rdlink); 

        //(4)这是个标记，标记这个节点【这个节点本身是已经在红黑树中】已经不在双向链表中；
        epi->rdy = 0;  //当这个节点被操作系统 加入到 双向链表中时，这个标记会设置为1。

        //(5)把事件标记信息拷贝出来；拷贝到提供的events参数中
        memcpy(&events[i++], &epi->event, sizeof(struct epoll_event));

        num --;
        cnt ++;       //拷贝 出来的 双向链表 中节点数目累加
        ep->rdnum --; //双向链表里边的节点数量减1
    }

    pthread_spin_unlock(&ep->lock);

    //(5)返回 实际 发生事件的 tcp连接的数目；
    return cnt; 
}

(4)epoll_event_callback()

通过跟踪epoll_event_callback在内核中被调用的位置。可知，当服务器在以下5种情况会调用epoll_event_callback：

客户端connect()连入，服务器处于SYN_RCVD状态时
三路握手完成，服务器处于ESTABLISHED状态时
客户端close()断开连接，服务器处于FIN_WAIT_1和FIN_WAIT_2状态时
客户端send/write()数据，服务器可读时
服务器可以发送数据时

接下来，我们来看一下epoll_event_callback的源码：

//当发生客户端三路握手连入、可读、可写、客户端断开等情况时，操作系统会调用这个函数，用以往双向链表中增加一个节点【该节点同时 也在红黑树中】
int epoll_event_callback(struct eventpoll *ep, int sockid, uint32_t event) {
   
   
    struct epitem tmp;
    tmp.sockfd = sockid;

    //(1)根据给定的key【这个TCP连接的socket】从红黑树中找到这个节点
    struct epitem *epi = RB_FIND(_epoll_rb_socket, &ep->rbr, &tmp);
    if (!epi) {
   
   
        nty_trace_epoll("rbtree not exist\n");
        assert(0);
    }

    //(2)从红黑树中找到这个节点后，判断这个节点是否已经被连入到双向链表里【判断的是rdy标志】
    if (epi->rdy) {
   
   
        //这个节点已经在双向链表里，那无非是把新发生的事件标志增加到现有的事件标志中
        epi->event.events |= event;
        return 1;
    } 

    //走到这里，表示 双向链表中并没有这个节点，那要做的就是把这个节点连入到双向链表中

    nty_trace_epoll("epoll_event_callback --> %d\n", epi->sockfd);

    pthread_spin_lock(&ep->lock);

    //(3)标记这个节点已经被放入双向链表中，我们刚才研究epoll_wait()的时候，从双向链表中把这个节点取走的时候，这个标志被设置回了0
    epi->rdy = 1;  

    //(4)把这个节点链入到双向链表的表头位置
    LIST_INSERT_HEAD(&ep->rdlist, epi, rdlink);

    //(5)双向链表中的节点数量加1，刚才研究epoll_wait()的时候，从双向链表中把这个节点取走的时候，这个数量减了1
    ep->rdnum ++;

    pthread_spin_unlock(&ep->lock);
    pthread_mutex_lock(&ep->cdmtx);
    pthread_cond_signal(&ep->cond);
    pthread_mutex_unlock(&ep->cdmtx);

    return 0;
}

以上代码的逻辑也十分简单，就是将eventpoll所指向的红黑树的节点插入到双向链表中。

总结

epoll底层实现中有两个关键的数据结构，一个是eventpoll另一个是epitem，其中eventpoll中有两个成员变量分别是rbr和rdlist,前者指向一颗红黑树的根，后者指向双向链表的头。而epitem则是红黑树节点和双向链表节点的综合体，也就是说epitem即可作为树的节点，又可以作为链表的节点，并且epitem中包含着用户注册的事件。

当用户调用epoll_create()时，会创建eventpoll对象（包含一个红黑树和一个双链表）；
而用户调用epoll_ctl(ADD)时，会在红黑树上增加节点（epitem对象）；
接下来，操作系统会默默地在通过epoll_event_callback()来管理eventpoll对象。当有事件被触发时，操作系统则会调用epoll_event_callback函数，将含有该事件的epitem添加到双向链表中。
当用户需要管理连接时，只需通过epoll_wait()从eventpoll对象中的双链表下"摘取"epitem并取出其包含的事件即可。

io复用之epoll核心源码剖析

绪论