【Linux后端服务器开发】poll/epoll多路转接IO服务器

目录

一、poll原理

二、poll实现多路转接IO服务器

三、epoll函数接口

四、epoll的工作原理

五、epoll实现多路转接IO服务器

---

一、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函数的超时时间，单位是毫秒：①timeout > 0 ，timeout时间以内阻塞等待，否则非阻塞返回一次；②timeout == 0，非阻塞等待；③timeout < 0，阻塞等待

events和revents的取值

事件	描述	是否可作为输入	是否可作为输出
POLLIN	数据（普通数据、优先数据）可读	是	是
POLLRDNORM	普通数据可读	是	是
POLLRDBAND	优先级带数据可读（Linux不支持）	是	是
POLLPRI	高优先级数据可读，如TCP带外数据	是	是
POLLOUT	数据（普通数据、优先数据）可写	是	是
POLLWRNORM	普通数据可写	是	是
POLLWRBAND	优先级带数据可写	是	是
POLLRDHUP	TCP连续被对方关闭或对方关闭了写操作，它由GNU引入	否	是
POLLERR	错误	否	是
POLLHUP	挂起。比如通道的写端被关闭后，读端描述符上将收到POLLHUP事件	否	是
POLLVNAL	文件描述符没有打开	否	是

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

poll的优点

poll多路转接的提出是为了解决select存在的部分缺陷，不同于select使用三个位图结构来表示三个fd_set的方式，poll使用一个pollfd的指针实现：

• pollfd结构包含了要监视的event和发生的event，不再使用select“参数-值”传递的方式，接口使用比select方便，不需要每次调用都重新设置需要关心的fd
• events参数：用户告诉内核你需要帮我监视哪些fd
• revents参数：内核告诉用户哪些正在监视的fd已经就绪了
• poll并没有最大数量限制（但是最大数量限制是由系统能接受的最大数量决定，监视的fd数量过大时性能也会下降）

poll的缺点

poll中监听的文件描述符增多时：

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

二、poll实现多路转接IO服务器

Log.hpp、Sock.hpp、main.cc与select中的多路转接是一样的：【Linux后端服务器开发】select多路转接IO服务器_命运on-9的博客-CSDN博客

PollServer.hpp

#pragma once#include <iostream>
#include <string>
#include <functional>
#include <poll.h>#include "Sock.hpp"using namespace std;static const int g_defaultport = 8080;
static const int g_num = 2024;
static const int g_defaultfd = -1;using func_t = function<string (const string&)>;class PollServer
{
public:PollServer(func_t f, int port = g_defaultport): _func(f), _port(port), _listensock(-1), _rfds(nullptr){}void Init(){_listensock = Sock::Socket();Sock::Bind(_listensock, _port);Sock::Listen(_listensock);_rfds = new struct pollfd[g_num];for (int i = 0; i < g_num; ++i)Reset_Item(i);_rfds[0].fd = _listensock;_rfds[0].events = POLLIN;}void Reset_Item(int i){_rfds[i].fd = g_defaultfd;_rfds[i].events = 0;_rfds[i].revents = 0;}void Print_Rfds(){cout << "rfd list: ";for (int i = 0; i < g_num; ++i)if (_rfds[i].fd != g_defaultfd)cout << _rfds[i].fd << " ";cout << endl;}void Accepter(int listensock){Log_Message(DEBUG, "Accepter in");string clientip;uint16_t clientport = 0;int sock = Sock::Accept(listensock, &clientip, &clientport);if (sock < 0)return;Log_Message(NORMAL, "accept success [%s: %d]", clientip.c_str(), clientport);int i = 0;for (; i < g_num; ++i){if (_rfds[i].fd != g_defaultfd)continue;elsebreak;}if (i == g_num){Log_Message(WARNING, "server is full, please wait");close(sock);}else{_rfds[i].fd = sock;_rfds[i].events = POLLIN;_rfds[i].revents = 0;}Print_Rfds();Log_Message(DEBUG, "Accepter out");}void Recver(int pos){Log_Message(DEBUG, "in Recver");// 1. 读取requestchar buffer[1024];ssize_t s = recv(_rfds[pos].fd, buffer, sizeof(buffer) - 1, 0);if (s > 0){buffer[s] = 0;Log_Message(NORMAL, "client# %s", buffer);}else if (s == 0){close(_rfds[pos].fd);Reset_Item(pos);Log_Message(NORMAL, "client quit");return;}else{close(_rfds[pos].fd);Reset_Item(pos);Log_Message(ERROR, "client quit: %s", strerror(errno));return;}// 2. 处理requeststring response = _func(buffer);// 3. 返回responsewrite(_rfds[pos].fd, response.c_str(), response.size());Log_Message(DEBUG, "out Recver");}void Handler_Read_Event(){for (int i = 0; i < g_num; ++i){// 过滤掉非法的fdif (_rfds[i].fd == g_defaultfd)continue;if (!(_rfds[i].events & POLLIN))continue;if (_rfds[i].fd == _listensock && (_rfds[i].revents & POLLIN))Accepter(_listensock);else if (_rfds[i].revents & POLLIN)Recver(i);}}void Start(){int timeout = -1;while (1){int n = poll(_rfds, g_num, timeout);switch (n){case 0:Log_Message(NORMAL, "timeout ...");break;case -1:Log_Message(WARNING, "poll error, code:%d, err string: %s", errno, strerror(errno));break;default:Log_Message(NORMAL, "have event ready!");Handler_Read_Event();break;}}}~PollServer(){if (_listensock < 0)close(_listensock);if (_rfds)delete[] _rfds;}private:int _port;int _listensock;struct pollfd* _rfds;func_t _func;
};

三、epoll函数接口

epoll的官方说法是为了处理大批量句柄而做了改进的poll，但是在底层原理的设计上，epoll与poll有着天差地别的区别（老婆与老婆饼的关系），epoll的设计远远优于poll。

它是在2.5.44内核中被引进的(epoll(4) is a new API introduced in Linux kernel 2.5.44)，几乎弥补了前两种多路转接接口(select、poll)的所有缺陷，被公认为Linux2.6之后性能最好的多路I/O就绪通知方法。

epoll_create

int epoll_create(int size);

创建一个epoll模型，自从2.6.8之后size参数是被忽略的，用完之后必须调用close关闭。

epoll_ctl

int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);

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

第一个参数是epoll_create()的返回值（epoll模型），第二个参数表示动作（用三个宏表示），第三个参数是需要监听的fd，第四个参数是告诉内核需要监听什么事。

第二个参数的取值：①EPOLL_CTL_ADD，注册新的fd到epfd中；②EPOLL_CTL_MOD，修改已经注册的fd监听事件；③EPOLL_CTL_DEL，从epfd中删除一个fd。

epoll_wait

int epoll_wait(int epfd, struct epoll_event * events, int maxevents, int timeout);

收集在epoll监听的事件中已经发生的事件：

• 参数events是分配好的epoll_event结构体
• epoll将会把发生的事件赋值到events数组中（events不可以是空指针，内核只负责把数据复制到这个events数组中，不会去帮助我们在用户态中分配内存）
• maxevents参数是告知内核这个events多大，这个maxevents的值不能大于创建的epoll_create()时的size
• 参数timeout是超时时间（ms，0会立即返回，-1是阻塞等待）
• 如果函数调用成功，返回I/O已经准备好的文件描述符数目，如返回0表示已超时，返回值小于0表示报错

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

events可以是以下几个宏的集合:

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

四、epoll的工作原理

1. 通过epoll_create()创建epoll模型
2. 通过epoll_ctl()将需要监视的文件描述符告诉OS，由OS维护红黑树进行监视
3. 当数据从硬件层传入网卡驱动再写入操作系统中监视的文件缓冲区，会通过每个文件的回调机制将受监视的fd节点传入就绪队列中
4. 上层直接对就绪队列进行事件处理，而不用对所有监视进行遍历

什么是事件就绪？底层的IO条件满足了，可以进行某种IO行为了，即事件就绪。

select / poll / epoll  ---->   IO事件的通知机制（等）

事件的通知有什么策略呢？LT工作模式和ET工作模式

关于事件的通知，这里有一个取快递的示例：

假设你网购了一些东西，快递到了，这时快递员张三就把快递送到你家楼下，打电话通知你让你下去取快递，你不下去他就一直给你打，直到你去将你的所有快递取完了他才离开。

之后你又网购了一些东西，快递到了，这时换了一个快递员变成李四了，他也将快递送到了你家楼下，但是他不会一直给你打电话，只要你接了电话他就离开了，也不管你有没有下来取快递。

这两种不同的通知策略，对应的就是LT水平触发模式和ET边缘触发模式，张三是LT，李四是ET。

epoll如何进行事件通知呢？通过文件的回调机制将红黑树中的就绪节点添加到就绪队列中

• LT水平触发：只要就绪队列中的数据没有处理完，epoll就会一直通知用户
• ET边缘触发：只要epoll将数据放入就绪队列里了，无论用户有没有将就绪队列里的数据读完，epoll就不再通知用户，除非底层的数据变化（数据增多），才会再次进行通知

epoll默认的是LT工作模式，我们也可以设置events将其改为ET模式。

为什么ET模式必须是非阻塞读取呢？

ET模式   --->   底层只有数据从无到有、从有到多变化的时候，才会进行通知上层   --->   只会通知一次   --->   倒逼程序员将本轮就绪的数据全部读取完   --->   你怎么知道你把就绪的数据全部读取完毕了呢？循环读取，直到读取不到数据了   --->   一般的fd，是阻塞式读取，但是ET模式下，必须是非阻塞读取，防止最后一次因读取不到数据而阻塞

为什么ET模式更高效？因为ET高效不仅体现在通知机制上，还会倒逼程序一次把就绪数据读取完，TCP通信时让接收方可以给发送方提供一个更大的窗口大小，即让对端更新出一个更大的滑动窗口，提高网络通信的数据吞吐量。

所以TCP中PSH标志的作用？让底层就绪事件，再次通知给上层。

五、epoll实现多路转接IO服务器

Log.hpp、Sock.hpp、main.cc与poll版本的一样

EpollServer.hpp

#pragma once#include <iostream>
#include <string>
#include <cstring>
#include <functional>
#include <sys/epoll.h>#include "Sock.hpp"using namespace std;static const int g_defaultport = 8080;
static const int g_size = 2024;
static const int g_defaultvalue = -1;
static const int g_defaultnum = 64;using func_t = function<string (const string&)>;class EpollServer
{
public:EpollServer(func_t f, uint16_t port = g_defaultport, int num = g_defaultnum): _func(f), _num(num), _revs(nullptr), _port(port), _listensock(g_defaultvalue), _epfd(g_defaultvalue){}void Init(){// 1. 创建socket_listensock = Sock::Socket();Sock::Bind(_listensock, _port);Sock::Listen(_listensock);// 2. 创建epoll模型_epfd = epoll_create(g_size);if (_epfd < 0){Log_Message(FATAL, "epoll create error: %s", strerror(errno));exit(EPOLL_CREATE_ERR);}// 3. 添加listensock到epoll中struct epoll_event ev;ev.events = EPOLLIN;ev.data.fd = _listensock;epoll_ctl(_epfd, EPOLL_CTL_ADD, _listensock, &ev);// 4. 申请就绪事件的空间_revs = new struct epoll_event[_num];Log_Message(NORMAL, "init server success");}void Handler_Event(int ready_num){Log_Message(DEBUG, "Handler_Event in");// 遍历就绪队列for (int i = 0; i < ready_num; ++i){uint32_t events = _revs[i].events;int sock = _revs[i].data.fd;if (sock == _listensock && (events & EPOLLIN)){// _listensock事件就绪，建立新连接string clientip;uint16_t clientport;int fd = Sock::Accept(sock, &clientip, &clientport);if (fd < 0){Log_Message(WARNING, "accept error");continue;}// 获取fd成功，可以直接读取吗？不可以，放入epollstruct epoll_event ev;ev.events = EPOLLIN;ev.data.fd = fd;epoll_ctl(_epfd, EPOLL_CTL_ADD, fd, &ev);}else if (events & EPOLLIN){// 普通事件就绪char buffer[1024];int n = recv(sock, buffer, sizeof(buffer), 0);if (n > 0){buffer[n] = 0;Log_Message(NORMAL, "client say# %s", buffer);// TODOstring response = response = _func(buffer);send(sock, response.c_str(), response.size(), 0);}else if (n == 0){// 先从epoll移除，再close fdepoll_ctl(_epfd, EPOLL_CTL_DEL, sock, nullptr);close(sock);Log_Message(NORMAL, "client quit");}else{epoll_ctl(_epfd, EPOLL_CTL_DEL, sock, nullptr);close(sock);Log_Message(ERROR, "recv error, code: %d, errstring: %s", errno, strerror(errno));}}}Log_Message(DEBUG, "Handler_Event out");}void Start(){int timeout = -1;while (1){int n = epoll_wait(_epfd, _revs, _num, timeout);switch (n){case 0:Log_Message(NORMAL, "timeout ...");break;case -1:Log_Message(WARNING, "epoll_wait failed, code: %d, err string: %s", errno, strerror(errno));break;default:Log_Message(NORMAL, "have event ready");Handler_Event(n);break;}}}private:uint16_t _port;int _listensock;int _epfd;                      // epoll模型struct epoll_event* _revs;      // 就绪队列int _num;func_t _func;
};