高级IO epoll模型

一.本质和作用

网络通信（本质是进程间的IO），文件读写等等本质都是IO,IO在操作系统中具有重要地位，那么我们如何进行高效的IO呢？

首先我们要理解IO的本质：

IO = 等 + 数据拷贝。

发送信息和接受信息这个动作相当于将一边的数据缓冲区拷到另外一边。当我们发送信息，我们需要等发送缓存区为空才能拷贝进去，当我们接收数据时，我们要等数据拷贝到接收缓存区才能读取。因此我们可以得出IO=等+数据拷贝，而数据拷贝主要和硬件相关，我们在软件方面无法提升。 因此我们要让IO高效，就要缩短平均IO的等待时间。

epoll就是利用这个原理。我们可以给epoll传入多个文件描述符，让epoll同时关注多个文件描述符下的io事件，当有任何一个或多个io事件就绪后，再通知上层来读取，这样我们就可以在相同的时间内等待多个io事件，缩短了平均等待时间从而提高了IO效率。

打个比方，等待io事件就像钓鱼，普通的io就是一根杠在钓鱼，而epoll就厉害了，他是用几十个杆子在钓鱼，效率肯定更高。

二.epoll 的相关系统调用

epoll_create

#include <sys/epoll.h>int epoll_create(int size);

创建一个 epoll 的句柄.

自从 linux2.6.8 之后，size 参数是被忽略的.不用管参数。
返回值是一个文件描述符
用完之后, 必须调用 close()关闭.

epoll_ctl

#include <sys/epoll.h>int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);

epoll 的事件注册函数.

第一个参数是 epoll_create()的返回值(epoll 的句柄).
第二个参数表示动作，用三个宏来表示.
第三个参数是需要监听的 fd.
第四个参数是告诉内核需要监听什么事，读或写之类的。

第二个参数的取值:

EPOLL_CTL_ADD：注册新的 fd 到 epfd 中；
EPOLL_CTL_MOD：修改已经注册的 fd 的监听事件；
EPOLL_CTL_DEL：从 epfd 中删除一个 fd；

第四个参数 struct epoll的结构

events （相当于位图）可以是以下几个宏的集合：

EPOLLIN : 表示对应的文件描述符可以读 (包括对端 SOCKET 正常关闭);
EPOLLOUT : 表示对应的文件描述符可以写;
EPOLLPRI : 表示对应的文件描述符有紧急的数据可读 (这里应该表示有带外数据到来);
EPOLLERR : 表示对应的文件描述符发生错误; • EPOLLHUP : 表示对应的文件描述符被挂断;
EPOLLET : 将 EPOLL 设为边缘触发(Edge Triggered)模式, 这是相对于水平触发(LevelTriggered)来说的
EPOLLONESHOT：只监听一次事件, 当监听完这次事件之后, 如果还需要继续监听这个 socket 的话, 需要再次把这个 socket 加入到 EPOLL 队列里.

epoll_wait

  #include <sys/epoll.h>int epoll_wait(int epfd, struct epoll_event *events,int maxevents, int timeout);

收集在 epoll 监控的事件中已经发送的事件.epoll_wait()阻塞式等待，直到有一个或多个事件就绪。

参数 events 是分配好的 epoll_event
epoll 将会把发生的事件赋值到 events 数组中 (events 不可以是空指针，内核只负责把数据复制到这个 events 数组中，不会去帮助我们在用户态中分配内存).
maxevents 告之内核这个 events 有多大，这个 maxevents 的值不能大于创建 epoll_create()时的 size
参数 timeout 是超时时间 (毫秒，0 会立即返回，-1 是永久阻塞
如果函数调用成功，返回对应 I/O 上已准备好的文件描述符数目，如返回0表示已超时, 返回小于 0 表示函数失败

三.工作原理

要想快速搞懂上述接口，最好的办法就是理解epoll的工作原理。

首先，epoll在内核中是有一个结构体来描述的：

当我们调用epoll_create（）函数就会初始这个结构体，同时epoll_create()为什么会返回一个文件描述符呢？Linux下一切皆文件。 epoll_create（）调用时，Linux初始化一个struct file结构体，并分配一个文件描述符，在通过这个struct file的一个指针指向eventpoll(和socket原理类似).

这样之后的epoll_ctl和epoll_wait通过对应的文件描述符就能找到这个结构体。

在上面的eventpoll结构体中有一个成员变量rbr ，这个是又是什么呢？

在linux内核中还有一颗红黑树，这个rbr就是指向这颗树，当我们调用epoll_ctl（）函数时，通过传入文件的描述符，可以找到eventpoll结构体，在通过其他参数传入，就会在这个树中插入一个含有关心的事件和文件描述符等信息的节点。

这样通过这颗红黑树，我们就可以将对应文件描述符和关心的事件关联起来。

在上面的eventpoll结构体中还有一个成员变量rdllist,这个指向一个链表。当我们关心的事件到来时，我们就可以通过底层对应的操作，将这个事件放到链表中。

当我们调用epoll_wait()时就可以从就绪队列里将对应的事件读取出来。

因此，我们可以得出epoll总体模型如下

四.简易代码实现

通过代码我们可以快速掌握用法。我们首先要了解——读事件关心数据，刚开始没有数据就会阻塞，所以要先添加到epoll中，而写事件关心缓冲区，缓存区不满时才能写，当我们缓存区满时才需要添加到epoll中。

这里我们要还要明白，send()当缓存区满时，代码会阻塞住，我们此时无法将他添加到epoll中，这时我们我们在创建套接字时，可以先将对应的文件描述符设置为非阻塞，这样send()就不会阻塞，可以添加到epoll中。将文件设置为非阻塞可以用fcntl（）函数，这里请自行查阅。

当我们将send（）设置成非阻塞后，当输入缓存区满了，send()无法进行后，会设置EWOULDBLOCK 或 EAGAIN错误码。

示例代码如下

#include <sys/epoll.h>
#include <sys/types.h> /* See NOTES */
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <cstring>
#include <pthread.h>
#include <sys/types.h>
#include <memory>
#include<iostream>
#include <fcntl.h>
void SetNonBlock(int fd)//用来对文件描述符设置成非阻塞
{int fl = ::fcntl(fd, F_GETFL);if(fl < 0) {return;}fcntl(fd, F_SETFL, fl | O_NONBLOCK);
}int main()
{int epfd=epoll_create(128);int sockfd = socket(AF_INET, SOCK_STREAM, 0);SetNonBlock(sockfd);std::cout<<"创建套接字事件成功\n";struct sockaddr_in local;memset(&local, 0, sizeof(local));local.sin_family = AF_INET;local.sin_port = htons(8889);local.sin_addr.s_addr =  INADDR_ANY;int n = ::bind(sockfd, (struct sockaddr *)&local, sizeof(local));if(n<0){std::cout<<"绑定套接字事件失败\n";}std::cout<<"绑定套接字事件成功\n";int m=listen(sockfd,16);if(m<0){std::cout<<"监听套接字事件失败\n";}struct sockaddr_in client;struct epoll_event ev;ev.data.fd=sockfd;ev.events=EPOLLIN; epoll_ctl(epfd,EPOLL_CTL_ADD,sockfd,&ev);std::cout<<"添加监听文件描述符的读事件成功\n";struct epoll_event evs[100];char buffer[1024];while(1){int n= epoll_wait(epfd,evs,100,5000);std::cout<<"监听到事件\n";for(int i=0;i<n;i++){if( evs[i].data.fd==sockfd){std::cout<<"监听到连接\n";int new_fd=accept(sockfd,nullptr,0);SetNonBlock(new_fd);struct epoll_event ev;ev.data.fd=new_fd;ev.events=EPOLLIN; epoll_ctl(epfd,EPOLL_CTL_ADD,new_fd,&ev);std::cout<<"添加事件成功\n";continue;}else if(evs[i].events&EPOLLIN&&evs[i].data.fd!=sockfd){int fd=evs[i].data.fd;int m= recv(fd,buffer,sizeof(buffer),0);buffer[m]=0;std::string sendstr="server said:";sendstr+=buffer;std::cout<<"服务器收到了客户端的消息为"<<buffer<<std::endl;int n=send(fd,sendstr.c_str(),sendstr.size(),0);if(n>0){std::cout<<"发送成功\n";//解除对写事件的关心。struct epoll_event ev;ev.data.fd=fd;ev.events=EPOLLIN; epoll_ctl(epfd,EPOLL_CTL_ADD,fd,&ev);memset(buffer,0,sizeof(buffer));}else{if(errno == EWOULDBLOCK || errno == EAGAIN)//阻塞时会将错误码设置为这两个中的一个{struct epoll_event ev;ev.data.fd=fd;ev.events=EPOLLIN&EPOLLOUT; //关心读写epoll_ctl(epfd,EPOLL_CTL_ADD,fd,&ev);}else{std::cout<<"send 错误了\n";return 0;}}}else if((evs[i].events&EPOLLOUT)&&evs[i].data.fd!=sockfd){int fd=evs[i].data.fd;std::string sendstr="server said:";sendstr+=buffer;int n=send(fd,sendstr.c_str(),sendstr.size(),0);if(n>0){std::cout<<"发送成功\n";//解除对写事件的关心。struct epoll_event ev;ev.data.fd=fd;ev.events=EPOLLIN; epoll_ctl(epfd,EPOLL_CTL_ADD,fd,&ev);memset(buffer,0,sizeof(buffer));}else{if(errno == EWOULDBLOCK || errno == EAGAIN)//阻塞时会将错误码设置为这两个中的一个{struct epoll_event ev;ev.data.fd=fd;ev.events=EPOLLIN&EPOLLOUT; //关心读写epoll_ctl(epfd,EPOLL_CTL_ADD,fd,&ev);}else{std::cout<<"send 错误了\n";return 0;}}}}if(n==0){std::cout<<"时间截止wait返回\n";continue;}if(n<0){std::cout<<"epoll_wait 出错\n";return 0;}}}

运行结果：

服务端：

客户端（用telnet）：

在上面的代码中，我们写的许多地方都有缺陷，如我们无法确保读到完整的报文，代码结构复杂。如果想跟进一步理解运用epoll，我们需要了解Reactor模型。这里就不作讲解。