【网络】poll 与epoll（原理、工作模式LT、ET）

在前面用的select中，它的使用方式非常麻烦，而且能支持的文件描诉符太少了。
下面来介绍一下更加方便、高效的方式:

1. poll

poll函数接口：

include <poll.h>int poll(struct pollfd *fds, nfds_t nfds, int timeout);

// pollfd 结构
struct pollfd 
{int fd; /* file descriptor */short events; /* requested events */short revents; /* returned events */
};

• fds 是一个 poll 函数监听的结构列表，每一个元素中，包含了三部分内容：文件描述符、监听的事件集合、返回的事件集合.
• nfds 表示 fds 数组的长度.
• timeout 表示 poll 函数的超时时间，单位是毫秒(ms).

events 和revents的取值如下：

常用的是POLLIN、POLLOUT

返回结果

• 返回值小于 0, 表示出错;
• 返回值等于 0, 表示 poll 函数等待超时;
• 返回值大于 0, 表示 poll 由于监听的文件描述符就绪而返回

#pragma once
#include "Socket.hpp"
#include "Log.hpp"
#include <poll.h>
using namespace MyLogModule;
using namespace MySocketModule;#define NUM (sizeof(fd_set) * 8)
const int gdefaultfd = -1;
#define ARRAY_SIZE 64
class PollServer
{
private:std::unique_ptr<Socket> _listen_socket;int16_t _port;bool _isrunning;struct pollfd* _fd_array;	//fd数组public:PollServer(int16_t port): _port(port), _listen_socket(make_unique<TcpSocket>()),_fd_array(new pollfd[ARRAY_SIZE])	//给数组所开空间的大小，决定所能连接的fd的数量{}~PollServer(){}void Init(){_listen_socket->BuildTcpSocketMethod(_port);// 初始化辅助数组for (int i = 0; i < ARRAY_SIZE; i++){_fd_array[i].fd = gdefaultfd;_fd_array[i].events = 0;_fd_array[i].revents = 0;}// 将listen_socket添加至辅助数组_fd_array[0].fd = _listen_socket->GetSockfd();_fd_array[0].events |= POLLIN;}void Start(){_isrunning = true;while (_isrunning){int timeout = 1000;int n = poll(_fd_array, ARRAY_SIZE, timeout);switch (n){case 0:cout << "time out," << endl;break;case -1:cout << "select error" << endl;break;default:// 内核告诉用户，你关心的rfds中的fd，有哪些已经就绪了！！cout << "有事件就绪了..." << endl;Dispatcher();break;}}_isrunning = false;}void Accepter(){InetAddr client;// 此时accept就不会阻塞了int newfd = _listen_socket->AcceptOrDie(&client);if (newfd < 0)return;else{cout << "get a new link,fd:" << newfd << endl;//依旧需要遍历来添加新的fdint pos = -1;for (int j = 0; j < ARRAY_SIZE; j++){if (_fd_array[j].fd == gdefaultfd){pos = j;break;}}if (pos == -1){//可扩容cout << "服务器满载了..." << endl;close(newfd); // 关闭新连接}else{_fd_array[pos].fd = newfd; // 关心是否可读_fd_array[pos].events |= POLLIN;}}}void Recver(int who){char buf[1024];size_t n = recv(_fd_array[who].fd, buf, sizeof(buf) - 1, 0);if (n > 0){buf[n] = 0;cout << "Client# " << buf << endl;string str = "Echo#";str += buf;send(_fd_array[who].fd, str.c_str(), str.size(), 0); // bug}else if (n == 0){LOG(LogGrade::DEBUG) << "客户端退出了,fd:" << _fd_array[who].fd;close(_fd_array[who].fd);_fd_array[who].fd = gdefaultfd;_fd_array[who].events = _fd_array[who].revents = 0;}else{LOG(LogGrade::DEBUG) << "客户端读取出错了,fd:" << _fd_array[who].fd;close(_fd_array[who].fd);_fd_array[who].fd = gdefaultfd;_fd_array[who].events = _fd_array[who].revents = 0;}}void Dispatcher(){//依旧要轮巡检测fd数组for (int i = 0; i < ARRAY_SIZE; i++){if (_fd_array[i].fd == gdefaultfd)continue;// 判断是否是listen_socketif (_fd_array[i].fd == _listen_socket->GetSockfd()){// 判断listen_socket是否在rfds中if (_fd_array[i].revents & POLLIN){Accepter();}}else{// 普通fd就绪了// 可以读了，不用阻塞！if (_fd_array[i].revents & POLLIN){Recver(i);  //进行IO}}}}
};

poll 的优点：

• 不同于 select 使用三个位图来表示三个 fdset 的方式， poll 使用一个 pollfd 的指针实现
• pollfd 结构中包含了要监视的 event 和发生的 event，使输入参数与输出参数分离，无需重新设置，使用比 select 更方便.
• poll 并没有最大数量限制 (但是数量过大后性能也是会下降)

poll 的缺点

• poll 中监听的文件描述符数目增多时和 select 函数一样，poll 返回后，需要轮询 pollfd 来获取就绪的描述符
• 每次调用 poll 都需要把大量的 pollfd 结构从用户态拷贝到内核中
• 同时连接的大量客户端在一时刻可能只有很少的处于就绪状态，因此随着监视的描述符数量的增长，其效率也会线性下降

2. epoll

按照 man 手册的说法: 是为处理大批量句柄而作了改进的 poll。
它是在 2.5.44 内核中被引进的，几乎具备了之前所说的一切优点，被公认为 Linux2.6 下性能最好的多路 I/O 就绪通知方法

所需函数接口

1. 创建epoll模型

#include <sys/epoll.h>
int epoll_create(int size);	//自从 linux2.6.8 之后，size 参数是被忽略的.

• size：自从 linux2.6.8 之后，size 参数是被忽略的，随便填一个正整数
• 用完之后, 必须调用 close()关闭.

2. 事件注册

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

它不同于 select()是在监听事件时告诉内核要监听什么类型的事件，而是在这里先注册要监听的事件类型.

• 第一个参数是 epoll_create()的返回值(epoll 的句柄).
• 第二个参数表示动作，用三个宏来表示
  • EPOLL_CTL_ADD： 注册新的 fd 到 epfd 中；
  • EPOLL_CTL_MOD：修改已经注册的 fd 的监听事件；
  • EPOLL_CTL_DEL：从 epfd 中删除一个 fd；
• 第三个参数是需要监听的 fd.
• 第四个参数是告诉内核需要监听什么事

其中epoll_event的结构如下：

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 */
};

其中，events 可以是以下几个宏的集合：

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

3. 收集就绪事件

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

收集在 epoll 监控的事件中已经发送的事件.

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

使用案例：

#pragma once
#include "Socket.hpp"
#include "Log.hpp"
#include <sys/epoll.h>
using namespace MyLogModule;
using namespace MySocketModule;const int gdefaultfd = -1;
#define ARRAY_SIZE 64class EpollServer
{
private:std::unique_ptr<Socket> _listen_socket;int16_t _port;bool _isrunning;int _epollfd;       //epoll模型所对应的fdstruct epoll_event *_ready_events;  //接收就绪事件的数组
public:EpollServer(int16_t port): _port(port), _listen_socket(make_unique<TcpSocket>()),_ready_events(new epoll_event[ARRAY_SIZE]){}~EpollServer(){}void Init(){_listen_socket->BuildTcpSocketMethod(_port);//1. epoll模型创建成功_epollfd = epoll_create(128);//ince Linux 2.6.8, the size argument is ignored,参数随便填一个 正整数！if(_epollfd < 0){LOG(LogGrade::ERROR) << "epoll_create fail";exit(EPOLL_CREATE_ERR);}//2. 将listen_socket添加到模型中struct epoll_event ev;ev.data.fd = _listen_socket->GetSockfd();ev.events = EPOLLIN;int n = epoll_ctl(_epollfd,EPOLL_CTL_ADD,_listen_socket->GetSockfd(),&ev);if(n < 0){LOG(LogGrade::ERROR) << "epoll_ctl fail";exit(EPOLL_CTRL_ERR);}}void Start(){_isrunning = true;while (_isrunning){int timeout = 1000;//3. 等待int n = epoll_wait(_epollfd, _ready_events, ARRAY_SIZE, timeout);switch (n){case 0:cout << "time out," << endl;break;case -1:cout << "select error" << endl;break;default:cout << "有事件就绪了....." << endl;//将就绪事件的个数传递，不做无效的遍历Dispatcher(n);break;}}_isrunning = false;}void Accepter(){InetAddr client;// 此时accept就不会阻塞了int newfd = _listen_socket->AcceptOrDie(&client);if (newfd < 0)return;else{//将fd添加到epoll模型中struct epoll_event ev;ev.data.fd = newfd;ev.events = EPOLLIN;int n = epoll_ctl(_epollfd, EPOLL_CTL_ADD, newfd, &ev);if(n < 0){LOG(LogGrade::ERROR) << "epoll_ctl fail";close(newfd);}LOG(LogGrade::ERROR) << "epoll_ctl success";}}void Recver(int fd){char buf[1024];size_t n = recv(fd, buf, sizeof(buf) - 1, 0);   //bugif (n > 0){buf[n] = 0;cout << "Client# " << buf << endl;string str = "Echo#";str += buf;send(fd,str.c_str(),str.size(), 0);   //bug}else if (n == 0){LOG(LogGrade::DEBUG) << "客户端退出了,fd:" << fd;//epoll_ctl规定，传入的fd必须是非法的，因此需要先移除，后关闭int n = epoll_ctl(_epollfd, EPOLL_CTL_DEL, fd, nullptr);if(n < 0){LOG(LogGrade::ERROR) << "epoll_ctl fail";exit(EPOLL_CTRL_ERR);}close(fd);  //后关闭}else{LOG(LogGrade::DEBUG) << "客户端读取出错了,fd:" << fd;int n = epoll_ctl(_epollfd, EPOLL_CTL_DEL, fd, nullptr);if(n < 0){LOG(LogGrade::ERROR) << "epoll_ctl fail";exit(EPOLL_CTRL_ERR);}//epoll_ctl规定，传入的fd必须是非法的，因此需要先移除，后关闭close(fd);}}void Dispatcher(int readynum){//数组中全是就绪事件，没有无效的遍历for (int i = 0; i < readynum;i++){// 判断是否是listen_socketif (_ready_events[i].data.fd == _listen_socket->GetSockfd()){// 判断listen_socket是否在返回的事件中if (_ready_events[i].events & EPOLLIN){Accepter();}}else{// 普通fd就绪了if (_ready_events[i].events & EPOLLIN){//读事件就绪Recver(_ready_events[i].data.fd);  //进行IO}}}}
};

3. epoll原理

当某一进程调用 epoll_create 方法时， Linux 内核会创建一个 eventpoll 结构体， 这个结构体中有两个成员与 epoll 的使用方式密切相关.

struct eventpoll{//..../*红黑树的根节点， 这颗树中存储着所有添加到 epoll 中的需要监控的事件*/struct rb_root rbr;/*双链表中则存放着将要通过 epoll_wait 返回给用户的满足条件的事件*/struct list_head rdlist;//....
};

红黑树上的一个节点，代表内核要为用户关心哪个fd上的哪些事件（epoll_ctl本质在修改红黑树），该红黑树就相当于select 与poll 中的辅助数组，现在这个数据结构不需要调用者维护了，由内核维护。

就绪链表：内核告诉用户，哪个fd上的哪些事件已经就绪了(epoll_wait本质是从该链表中拿取的fd)

每一个 epoll 对象都有一个独立的 eventpoll 结构体， 用于存放通过 epoll_ctl 方法向 epoll对象中添加进来的事件.

• 这些事件都会挂载在红黑树中， 如此， 重复添加的事件就可以通过红黑树而高效的识别出来
• 而所有添加到 epoll 中的事件都会与设备(网卡)驱动程序建立回调关系，也就是说，当响应的事件发生时会调用这个回调方法.
• 这个回调方法在内核中叫 ep_poll_callback，它会将发生的事件添加到 rdlist 双链表中.
• 在 epoll 中， 对于每一个事件， 都会建立一个 epitem 结构体
• 调用 epoll_wait 检查是否有事件发生时，只需要检查 eventpoll 对象中的rdlist 双链表中是否有 epitem 元素即可O(1).
• 如果 rdlist 不为空，则把发生的事件复制到用户态，同时将事件数量返回给用户，这个操作的时间复杂度O(n)

struct epitem{struct rb_node rbn;//红黑树节点struct list_head rdllink;//双向链表节点struct epoll_filefd ffd; //事件句柄信息struct eventpoll *ep; //指向其所属的 eventpoll 对象struct epoll_event event; //期待发生的事件类型
}

如何理解epoll_create返回的是一个文件描诉符呢？

eventepoll 被struct file中的void *private_data指针指向，通过文件描述符，就可以找到epoll模型，进而找到红黑树、就绪队列

epoll为什么高效？

1. 原理层面：用户将要关心的fd和事件注册进去后，OS自己调用回调函数驱动；事件就绪后，将其结构加入到就绪队列中（不需要轮询检测了，OS自动将其加入到就绪队列中，只需要看就绪队列即可）
2. 接口层面：epoll返回给用户就绪事件时，直接将就绪的事件返回，不做无效的拷贝；

epoll 的优点(和 select 的缺点对应)

• 接口使用方便：虽然拆分成了三个函数，但是反而使用起来更方便高效，不需要每次循环都设置关注的文件描述符，也做到了输入输出参数分离开
• 数据拷贝轻量：只在合适的时候调用 EPOLL_CTL_ADD 将文件描述符结构拷贝到内核中， 这个操作并不频繁，用户到内核的拷贝少（select/poll 都是每次循环都要进行拷贝，涉及太多用户到内核的拷贝)
• 事件回调机制: 避免使用遍历，而是使用回调函数的方式，将就绪的文件描述符结构加入到就绪队列中，epoll_wait 返回直接访问就绪队列就知道哪些文件描述符就绪，这个操作时间复杂度 O(1)，即使文件描述符数目很多，效率也不会受到影响.
• 没有数量限制：文件描述符数目无上限

4. epoll工作模式

poll 有 2 种工作方式：水平触发(level triggered) 和 边缘触发(edge triggered)

假如你正在打游戏，进入关键时刻，你妈饭做好了，喊你吃饭的时候有两种方式:

• 如果你妈喊你一次，你没动，那么你妈会继续喊你第二次，第三次…，直到你过来吃饭(亲妈，水平触发)
• 如果你妈喊你一次，你没动，你妈就不管你了(后妈，边缘触发)

select、poll的工作模式就是 LT，epoll默认状态下就是 LT 工作模式

4.1 水平模式LT

水平触发工作模式

• 当 epoll 检测到 socket 上事件就绪的时候，可以不立刻进行处理，或者只处理一部分.
• 假如socket 的另一端被写入了 10KB 的数据，由于只读了 1K 数据，缓冲区中还剩 9K 数据；在第二次调用epoll_wait 时，epoll_wait 仍然会立刻返回并通知 socket读事件就绪（就绪数据，会一直在rdlist中，除非把数据取完，否则，该节点一直存在，epoll_wait就一直返回）
• 直到缓冲区上所有的数据都被处理完，epoll_wait 才不会立刻返回.
• 支持阻塞读写和非阻塞读写

4.2 边缘模式ET

边缘触发工作模式：如果我们在第1步将 socket 添加到 epoll 描述符的时候若使用了 EPOLLET 标志，epoll 进入 ET 工作模式

• 当 epoll 检测到 socket 上事件就绪时，必须立刻处理
• 假如socket 的另一端被写入了 10KB 的数据，虽然只读了 1K 的数据，缓冲区还剩 9K 的数据，在第二次调用epoll_wait 的时候，epoll_wait 不会再返回了
• 也就是说，ET 模式下，文件描述符上的事件就绪后，只有一次处理机会。 epol_wait取走了就绪节点，该节点在rdlist中立马移除，除非又有数据从网络传递到节点的缓冲区，否则，不再通知（即便缓冲区中数据没有取完）
• 只支持非阻塞的读写

ET模式更加高效的原因：

1. epoll_wait 返回的次数少了很多（内核到用户的拷贝变少了），因为不做重复通知，一旦通知，数据必须全部取完（倒逼程序猿取完）
2. 上层取完了，可以给通信的对端通告一个更大的接收窗口，对端滑动窗口也就更大，网络IO的吞吐量也就大了。

ET模式是有代价的：

• 一旦数据就绪，要读取，必须全部取完；
• 如果不取完，且该fd没有数据再来了，则会造成上次未取完的数据丢失

你怎么保证能取完呢？（循环读 + 非阻塞）

循环读取，一旦循环读取，就必须是非阻塞的 (O_NONBLOCK)；如果在读取时阻塞了，会导致服务器无法处理其他事件，服务器的性能会严重下降，甚至可能完全卡住。

epoll 的使用场景：
epoll 的高性能，是有一定的特定场景的，如果场景选择的不适宜， epoll 的性能可能适得其反

• 对于多连接，且多连接中只有一部分连接比较活跃时，比较适合使用 epoll；（若连接都是活跃的，则可以使用多进程/线程 + epoll的方式）

例如，典型的一个需要处理上万个客户端的服务器，例如各种互联网 APP 的入口服务器，这样的服务器就很适合 epoll.

• 对于少数连接，如果只是系统内部，服务器和服务器之间进行通信，只有少数的几个连接，这种情况下用epoll 就并不合适
• 具体要根据需求和场景特点来决定使用哪种 IO 模型