epoll源码剖析

1.前言

好久好久没有更新博客了，最近一直在实习，刷算法找工作，忙里偷闲简单研究了一下epoll的源码。也是由于面试的时候经常被问到，我只会说那一套，什么epoll_create创建红黑树，以O(1)的方式去读取数据，它和poll与select的区别等等。本篇将从epoll的源码层面重新学习epoll。

2.应用层的体现

多路转接(epoll)实现我在这篇文章中实现了一个简单的epoll网络server。感兴趣的同学可以简单阅读一下，我只挑其中关键的代码来讲一下应用层是如何使用epoll的：

#include"Sock.hpp"
using namespace ns_Sock;
#define NUM 128
#include<sys/epoll.h>
#include <cstdlib>
void Usage(char* proc)
{cout<<"Usage \n\t"<<proc<<" port"<<endl;
}
int main(int argc,char* argv[])
{if(argc!=2){Usage(argv[0]);exit(-1);}uint16_t port=(uint16_t)atoi(argv[1]);int listen_sock=Sock::Socket();Sock::Bind(listen_sock,port);Sock::Listen(listen_sock);//建立epoll模型，获得epfdint epfd=epoll_create(128);//先添加listen_sock和它所关心的事件到内核中struct epoll_event ev;ev.events=EPOLLIN;ev.data.fd=listen_sock;//虽然epoll_ctl有文件描述符，但是revs数组中的元素是epoll_event没有fd，因此需要将fd添加都epoll_event的data字段中epoll_ctl(epfd,EPOLL_CTL_ADD,listen_sock,&ev);//事件循环volatile bool quit=false;struct epoll_event revs[NUM];//由于epoll_wait的数组是输出型参数，因此需要接收while(!quit){int timeout=1000;int n=epoll_wait(epfd,revs,NUM,-1);//epoll_wait会将epfd中就绪事件的epoll_event结构体放在revs数组中，返回值表示数组大小switch(n){case 0:cout<<"timeout....."<<endl;break;case -1:cerr<<"epoll error"<<endl;break;default:cout<<"有事件就绪了"<<endl;//处理就绪事件for(int i=0;i<n;i++){int sock=revs[i].data.fd;//暂时方案cout<<"文件描述符"<<sock<<"有数据就绪了"<<endl;if(revs[i].events&EPOLLIN)//读事件就绪{cout<<"文件描述符"<<sock<<"读事件就绪了"<<endl;if(sock==listen_sock){int fd=Sock::Accept(listen_sock);if(fd>=0){cout<<"获取新链接成功了"<<endl;//此时还不能读需要添加到epfd的空间中struct epoll_event _ev;_ev.events=EPOLLIN;_ev.data.fd=fd;epoll_ctl(epfd,EPOLL_CTL_ADD,fd,&_ev);cout<<"已经把"<<fd<<"添加到epfd空间中了"<<endl;}}//正常的读处理else{cout<<"文件描述符"<<sock<<"正常数据准备就绪"<<endl;char buffer[1024];ssize_t s=read(sock,buffer,sizeof(buffer)-1);if(s>0){buffer[s]=0;cout<<"client["<<sock<<"]#"<<buffer<<endl;}else if(s==0){cout<<"client quit "<<sock<<"将被关闭"<<endl;close(sock);epoll_ctl(epfd,EPOLL_CTL_DEL,sock,nullptr);//将该套接字从epoll空间关注的位置删除cout<<"Sock:"<<sock<<"delete from epoll success"<<endl;}else{cout<<"recv error"<<endl;close(sock);epoll_ctl(epfd,EPOLL_CTL_DEL,sock,nullptr);cout<<"delete sock"<<sock<<endl;}}}}}}
}

这是我那篇博客的服务器端的代码，使用telnet是可以直接访问的，通过这段代码我们可以发现调用epoll的过程以及一些细节。首先就是众所周知的：

• epoll_create创建一个epoll空间。
• 接着调用epoll_ctl将一个文件描述符以及对该文件描述符需要关心的事件放进epoll空间中。
• 然后调用epoll_wait进行等待就好了。事件就绪会使用epoll_wait这个函数来通知我们。

但仔细看代码还是会发现一些细节，在epoll空间建立完成后，添加的第一个文件描述符就是listen_sock，并且关心它的读事件。
在epoll_wait成功的时候，会返回一个就绪事件的数组，通过遍历这个数组去对数据进行操作，但当该数组元素表示的是listen_sock事件就绪的时候，需要在listen_sock中将新监听到的链接accept上来。这是对监听套接字独有的一种处理方式，也说明epoll的回调函数不止一个触发条件(数据就绪orl链接就绪)。

其中这里还涉及到了一个重要的结构体epoll_event，这个结构体在内核源码中也经常使用到，下面给出它在linux系统中的结构：

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_PACKED;

其中它的events表示的是事件，data又是一个结构体，它其中的一个重要的东西就是fd。它表示该epoll_event是哪个套接字的。

3.两个重要结构

(1)eventpoll

其中一个结构名为eventpoll，当调用epoll_create的时候内核会自动创建它，所以其实所谓的epoll空间仅仅是一个结构体而已。

struct eventpoll {/*struct ep_rb_tree {struct epitem *rbh_root; 			}*/ep_rb_tree rbr;      //rbr指向红黑树的根节点int rbcnt; //红黑树中节点的数量（也就是添加了多少个TCP连接事件）LIST_HEAD( ,epitem) rdlist;    //rdlist指向双向链表的头节点；/*	这个LIST_HEAD等价于 struct {struct epitem *lh_first;}rdlist;*/int rdnum; //双向链表中节点的数量（也就是有多少个TCP连接来事件了）// ...略...
};

它包含了几个内容，有我们熟知的红黑树根节点，还有所谓的就绪队列(是一个双链表)，以及红黑树与就绪队列的节点数量。

(2)epitem

这个是比较关键的一个结构体，epoll_ctl将文件描述符以及所关心的事件插入到epoll空间中，插入的就是这么个玩意。

struct epitem {RB_ENTRY(epitem) rbn;/*  RB_ENTRY(epitem) rbn等价于struct {											struct type *rbe_left;		//指向左子树struct type *rbe_right;		//指向右子树struct type *rbe_parent;	//指向父节点int rbe_color;			    //该节点的颜色} rbn*/LIST_ENTRY(epitem) rdlink;/* LIST_ENTRY(epitem) rdlink等价于struct {									struct type *le_next;	//指向下个元素struct type **le_prev;	//前一个元素的地址}*/int rdy; //判断该节点是否同时存在与红黑树和双向链表中int sockfd; //socket句柄struct epoll_event event;  //存放用户填充的事件
};

首先它包含了一个红黑树节点的结构，其次他又包含了双向链表的结构，到这里我们就可以发现，红黑树和双向链表中插入的是同一个结构体epitem。也很容易想到epoll_ctl的本质其实就是使用传入的参数来构造一个epitem。
因此它包含的参数显而易见：sockfd表示该epitem对应的套接字，epoll_event表示需要关心的该套接字的事件。

4.四个函数

(1)epoll_create源码

int epoll_create(int size) {//size没有什么卵用if (size <= 0) return -1;//tcp服务，我也不懂，目前不是重点nty_tcp_manager *tcp = nty_get_tcp_manager();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.建立一个eventpollstruct eventpoll *ep = (struct eventpoll*)calloc(1, sizeof(struct eventpoll));if (!ep) {nty_free_socket(epsocket->id, 0);return -1;}//2.初始化红黑树指针为空，节点数为0ep->rbcnt = 0;RB_INIT(&ep->rbr);//3.双向链表头指向空LIST_INIT(&ep->rdlist);//线程操作，暂时不用关心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;}//4.保存ep对象，通过epid可以在系统中找到eventpoll结构tcp->ep = (void*)ep;epsocket->ep = (void*)ep;return epsocket->id;
}

我们发现在epoll_create的时候会传入一个参数，但是这个参数似乎没有意义，从这里就可以很简单的看出了，这个size除了判断一下是不是大于0之外，之后什么都没做。因此这个参数是没有意义的。
它的主要功能我在代码中标注了，大致如下：

• 1.建立一个eventpoll对象，并为其分配空间。
• 2.将该对象中红黑树根节点指向空，节点个数设为0。
• 3.将该对象中双向链表头节点指向空。
• 4.将eventpoll对象保存起来。以后可以通过epid找到它，并返回这个epid。

(2)epoll_ctl源码

下面来看看epoll_ctl都干了些什么，在讲解这个函数之前，我们还是先回头看一下我们是怎么向它传参的：

epoll_ctl(epfd,EPOLL_CTL_ADD,listen_sock,&ev);

这是我们将listen_sock加入到epoll空间中的代码，可以看到epfd用于寻找eventpoll结构体，EPOLL_CTL_ADD表示我是要加入一个epoll节点，listen_sock表示我加入的这个节点的文件描述符是listen_sock，&ev表示我关心的事件(使用地址传参为了节省空间)，这一切是不是很明确了呢?下面我们来阅读这部分的源码。

int epoll_ctl(int epid, int op, int sockid, struct epoll_event *event) {//tcp部分，暂时不管nty_tcp_manager *tcp = nty_get_tcp_manager();if (!tcp) return -1;nty_trace_epoll(" epoll_ctl --> 1111111:%d, sockid:%d\n", epid, sockid);//1.通过传入进来的epid找到对应的eventpoll结构体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;}//2.判断如果是增加节点if (op == EPOLL_CTL_ADD) {pthread_mutex_lock(&ep->mtx);//在栈区先创建一个对象，接收sockidstruct epitem tmp;tmp.sockfd = sockid;//查找这个节点是否在红黑树中，其实也就是根据epollfd查找的，这查找传入tmp，可能表示的是红黑树节点类型，这是一个小细节struct epitem *epi = RB_FIND(_epoll_rb_socket, &ep->rbr, &tmp);if (epi) {nty_trace_epoll("rbtree is exist\n");pthread_mutex_unlock(&ep->mtx);return -1;}//如果不存在才为结构体分配完整空间epi = (struct epitem*)calloc(1, sizeof(struct epitem));if (!epi) {pthread_mutex_unlock(&ep->mtx);errno = -ENOMEM;return -1;}//使用参数构建这个epitem节点epi->sockfd = sockid;memcpy(&epi->event, event, sizeof(struct epoll_event));epi = RB_INSERT(_epoll_rb_socket, &ep->rbr, epi);assert(epi == NULL);ep->rbcnt ++;pthread_mutex_unlock(&ep->mtx);} else if (op == EPOLL_CTL_DEL) {pthread_mutex_lock(&ep->mtx);struct epitem tmp;//先把fd分配给sockfdtmp.sockfd = sockid;//查找该节点是否已经存在，这也说明是根据sockfd查找的struct epitem *epi = RB_FIND(_epoll_rb_socket, &ep->rbr, &tmp);if (!epi) {nty_trace_epoll("rbtree no exist\n");pthread_mutex_unlock(&ep->mtx);return -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);//找到了就将该节点的事件进行修改，这里还有一个细节，修改之后会默认加入EPOLLERR | EPOLLHUP这两个事件监听，一个表示套接字发生错误，一个表示套接字被挂断。if (epi) {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;
}

每根据一个不同的选择就创建一个结构，并查找它是不是在红黑树中可能有点冗余，感觉直接在上面创建一个即可。
这段代码看起来不难，其实细节有很多的，首先先来说一下大体的步骤：

• 根据传入的epfd找到对应的eventpoll结构。
• 根据不同的选择类型进行if-else判断
  1.首先所有操作，都要先在栈区建立一个epitem的结构，然后将sock传入其中。并使用红黑树查找函数传入该结构查找。
  2.根据有没有来进行构造删除，或者修改。
  以插入为例，没有则在堆区构建结构体对象，然后向红黑树中插入，使用memcpy来加入事件。
  其中的两个细节包括：

都是先建立一个栈区对象，然后红黑树中进行查找的，这里可以优化一下。
在修改的时候，会加入监听事件：文件描述符出错，套接字被挂断。

(3)epoll_wait的源码

这个函数就是用于等待事件就绪，然后将他插入就绪队列中的，其中这里的epoll_event是一个输出型参数，它通常表示一个数组的首地址。这里可以再回顾一下它是怎么进行传参的：

int n=epoll_wait(epfd,revs,NUM,-1);

其中epfd显然还是去找eventpoll的，revs是一个数组首元素地址(我们建立一个数组，传入数组名其实就可以了)，NUM是一个整数，表示多少个套接字就绪了就可以返回了，-1表示的是只要没有文件描述符就绪，就永久阻塞。
值得注意的是，这里我们没有回调函数的实现，也就是说暂时没有套接字就绪了将它插入等待队列中的操作，它的实现在后面讲。

int epoll_wait(int epid, struct epoll_event *events, int maxevents, int timeout) {//tcp相关，可恶我写完这篇文章一定看看这是个啥东西nty_tcp_manager *tcp = nty_get_tcp_manager();if (!tcp) return -1;//nty_socket_map *epsocket = &tcp->smap[epid];//找到eventpollstruct _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);}
//如果rdnum为空，并且等待时间不为0的时候会等待一段时间while (ep->rdnum == 0 && timeout != 0) {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) {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;//哪个少将哪个作为事件的数量int num = (ep->rdnum > maxevents ? maxevents : ep->rdnum);int i = 0;while (num != 0 && !LIST_EMPTY(&ep->rdlist)) { //EPOLLET//每次从双向链表的头节点取得一个一个的节点struct epitem *epi = LIST_FIRST(&ep->rdlist);//把这个节点从双向链表中删除LIST_REMOVE(epi, rdlink);//标记这个节点不在双向链表中，但是在红黑树中epi->rdy = 0;//只有当该节点再次被放入双向链表中的时候，才会置为1//把标记的信息拷贝出来，拷贝到提供的events参数中memcpy(&events[i++], &epi->event, sizeof(struct epoll_event));//--，++的操作num --;cnt ++;ep->rdnum --;}pthread_spin_unlock(&ep->lock);return cnt;
}

(4)epoll_event_callback()

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

它的作用是向双向链表中添加一个红黑树的节点。它的参数有一个eventpoll类型，注意这里没有传epid，有一个sockid，有一个event。

int epoll_event_callback(struct eventpoll *ep, int sockid, uint32_t event) {struct epitem tmp;tmp.sockfd = sockid;//首先根据sockid找到红黑树中的节点struct epitem *epi = RB_FIND(_epoll_rb_socket, &ep->rbr, &tmp);if (!epi) {nty_trace_epoll("rbtree not exist\n");assert(0);}//根据epi->rdy判断该节点是否在双向链表里if (epi->rdy) {//如果在就将事件填入events中epi->event.events |= event;return 1;} //如果不在，要做的就是将这个节点添加到双向链表中的表头位置nty_trace_epoll("epoll_event_callback --> %d\n", epi->sockfd);pthread_spin_lock(&ep->lock);epi->rdy = 1;LIST_INSERT_HEAD(&ep->rdlist, epi, rdlink);ep->rdnum ++;pthread_spin_unlock(&ep->lock);pthread_mutex_lock(&ep->cdmtx);pthread_cond_signal(&ep->cond);pthread_mutex_unlock(&ep->cdmtx);return 0;
}

5.水平触发和边缘触发

1.状态变化

要讲明白水平触发和边缘触发就需要知道都有哪些状态会触发，无非也就这四种：

• 可读：socket上有数据
• 不可读：socket上没有数据了
• 可写：socket上有空间可写
• 不可写：socket上无空间可写

对于水平触发模式，一个事件只要有，就会一直被触发。对于边缘触发模式，一个事件只要从无到有就会被触发。

2.LT模式

读：只要接收缓冲区上有未读完的数据，就会一直被触发。
写：只要发送缓冲区上还有空间，就会一直被触发。如果程序依赖于可写事件触发去发送数据，要移除可写事件，避免无意义的触发。

在LT模式下，读事件触发后，可以按需收取想要的字节数，不用把本次接收到的数据收取干净。

3.ET模式

读：只有接收缓冲区上的数据从无到有，就会被触发一次。
写：只有发送缓冲区上由不可写到可写，就会触发，(由满到不满)

在ET模式下，当读事件触发后，需要将数据一次性读干净。