I/O多路转接之epoll

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

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

当调用epoll_wait检查是否有事件发生时，只需要检查eventpoll对象中的rdlist双链表中是否有epitem 元素即可.

如果rdlist不为空，则把发生的事件复制到用户态，同时将事件数量返回给用户. 这个操作的时间复杂度 是O(1).

总结一下, epoll的使用过程就是三部曲:

调用epoll_create创建一个epoll句柄;

调用epoll_ctl, 将要监控的文件描述符进行注册;

调用epoll_wait, 等待文件描述符就绪

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

接口使用方便: 虽然拆分成了三个函数, 但是反而使用起来更方便高效. 不需要每次循环都设置关注的文 件描述符, 也做到了输入输出参数分离开

数据拷贝轻量: 只在合适的时候调用 EPOLL_CTL_ADD 将文件描述符结构拷贝到内核中, 这个操作并不频 繁(而select/poll都是每次循环都要进行拷贝)

事件回调机制: 避免使用遍历, 而是使用回调函数的方式, 将就绪的文件描述符结构加入到就绪队列中, epoll_wait 返回直接访问就绪队列就知道哪些文件描述符就绪. 这个操作时间复杂度O(1). 即使文件描述 符数目很多, 效率也不会受到影响.

没有数量限制: 文件描述符数目无上限

注意!!

网上有些博客说, epoll中使用了内存映射机制

内存映射机制: 内核直接将就绪队列通过mmap的方式映射到用户态. 避免了拷贝内存这样的额外性能开销

这种说法是不准确的. 我们定义的struct epoll_event是我们在用户空间中分配好的内存. 势必还是需要将内核的数据拷贝到这个用户空间的内存中的.

epoll工作方式

你妈喊你吃饭的例子

你正在吃鸡, 眼看进入了决赛圈, 你妈饭做好了, 喊你吃饭的时候有两种方式:

1. 如果你妈喊你一次, 你没动, 那么你妈会继续喊你第二次, 第三次...(亲妈, 水平触发)

2. 如果你妈喊你一次, 你没动, 你妈就不管你了(后妈, 边缘触发)

epoll有2种工作方式-水平触发(LT)和边缘触发(ET)

假如有这样一个例子

水平触发Level Triggered 工作模式

epoll默认状态下就是LT工作模式.

当epoll检测到socket上事件就绪的时候, 可以不立刻进行处理. 或者只处理一部分.

如上面的例子, 由于只读了1K数据, 缓冲区中还剩1K数据, 在第二次调用 epoll_wait 时, epoll_wait

仍然会立刻返回并通知socket读事件就绪.

直到缓冲区上所有的数据都被处理完, epoll_wait 才不会立刻返回.

支持阻塞读写和非阻塞读写

边缘触发Edge Triggered工作模式

如果我们在第1步将socket添加到epoll描述符的时候使用了EPOLLET标志, epoll进入ET工作模式.

当epoll检测到socket上事件就绪时, 必须立刻处理.

如上面的例子, 虽然只读了1K的数据, 缓冲区还剩1K的数据, 在第二次调用 epoll_wait 的时候,

epoll_wait 不会再返回了.

也就是说, ET模式下, 文件描述符上的事件就绪后, 只有一次处理机会.

ET的性能比LT性能更高( epoll_wait 返回的次数少了很多). Nginx默认采用ET模式使用epoll.

只支持非阻塞的读写

select和poll其实也是工作在LT模式下. epoll既可以支持LT, 也可以支持ET.

LT是 epoll 的默认行为. 使用 ET 能够减少 epoll 触发的次数. 但是代价就是强逼着程序猿一次响应就绪过程中就把 所有的数据都处理完.

相当于一个文件描述符就绪之后, 不会反复被提示就绪, 看起来就比 LT 更高效一些. 但是在 LT 情况下如果也能做到 每次就绪的文件描述符都立刻处理, 不让这个就绪被重复提示的话, 其实性能也是一样的.

另一方面, ET 的代码复杂程度更高了

使用 ET 模式的 epoll, 需要将文件描述设置为非阻塞. 这个不是接口上的要求, 而是 "工程实践" 上的要求.

假设这样的场景: 服务器接受到一个10k的请求, 会向客户端返回一个应答数据. 如果客户端收不到应答, 不会发送第二个10k请求

如果服务端写的代码是阻塞式的read, 并且一次只 read 1k 数据的话(read不能保证一次就把所有的数据都读出来,参考 man 手册的说明, 可能被信号打断), 剩下的9k数据就会待在缓冲区中

此时由于 epoll 是ET模式, 并不会认为文件描述符读就绪. epoll_wait 就不会再次返回. 剩下的 9k 数据会一直在缓冲区中. 直到下一次客户端再给服务器写数据. epoll_wait 才能返回

但是问题来了

服务器只读到1k个数据, 要10k读完才会给客户端返回响应数据.

客户端要读到服务器的响应, 才会发送下一个请求

客户端发送了下一个请求, epoll_wait 才会返回, 才能去读缓冲区中剩余的数据

 

#pragma once
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <string>
#include <cassert>
#include <unistd.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <fcntl.h>
typedef struct sockaddr sockaddr;
typedef struct sockaddr_in sockaddr_in;
#define CHECK_RET(exp) \\if (!(exp))        \\{                  \\return false;  \\}class TcpSocket
{
public:TcpSocket() : fd_(-1) {}TcpSocket(int fd) : fd_(fd) {}// 以下代码添加在 TcpSocket 类中// 非阻塞 IO 接口bool SetNoBlock(){int fl = fcntl(fd_, F_GETFL);if (fl < 0){perror("fcntl F_GETFL");return false;}int ret = fcntl(fd_, F_SETFL, fl | O_NONBLOCK);if (ret < 0){perror("fcntl F_SETFL");return false;}return true;}bool RecvNoBlock(std::string *buf) const{// 对于非阻塞 IO 读数据, 如果 TCP 接受缓冲区为空, 就会返回错误// 错误码为 EAGAIN 或者 EWOULDBLOCK, 这种情况也是意料之中, 需要重试// 如果当前读到的数据长度小于尝试读的缓冲区的长度, 就退出循环// 这种写法其实不算特别严谨(没有考虑粘包问题)buf->clear();char tmp[1024 * 10] = {0};for (;;){ssize_t read_size = recv(fd_, tmp, sizeof(tmp) - 1, 0);if (read_size < 0){if (errno == EWOULDBLOCK || errno == EAGAIN){continue;}perror("recv");return false;}if (read_size == 0){// 对端关闭, 返回 falsereturn false;}tmp[read_size] = '\\0';*buf += tmp;if (read_size < (ssize_t)sizeof(tmp) - 1){printf("数据太少\\n");break;}}return true;}bool SendNoBlock(const std::string &buf) const{// 对于非阻塞 IO 的写入, 如果 TCP 的发送缓冲区已经满了, 就会出现出错的情况// 此时的错误号是 EAGAIN 或者 EWOULDBLOCK. 这种情况下不应放弃治疗// 而要进行重试ssize_t cur_pos = 0; // 记录当前写到的位置ssize_t left_size = buf.size();for (;;){ssize_t write_size = send(fd_, buf.data() + cur_pos, left_size, 0);if (write_size < 0){if (errno == EAGAIN || errno == EWOULDBLOCK){// 重试写入continue;}return false;}cur_pos += write_size;left_size -= write_size;// 这个条件说明写完需要的数据了if (left_size <= 0){break;}}return true;}bool Socket(){fd_ = socket(AF_INET, SOCK_STREAM, 0);if (fd_ < 0){perror("socket");return false;}printf("open fd = %d\\n", fd_);return true;}bool Close() const{close(fd_);printf("close fd = %d\\n", fd_);return true;}bool Bind(const std::string &ip, uint16_t port) const{sockaddr_in addr;addr.sin_family = AF_INET;addr.sin_addr.s_addr = INADDR_ANY;addr.sin_port = htons(port);int ret = bind(fd_, (sockaddr *)&addr, sizeof(addr));if (ret < 0){perror("bind");return false;}return true;}bool Listen(int num) const{int ret = listen(fd_, num);if (ret < 0){perror("listen");return false;}return true;}bool Accept(TcpSocket *peer, std::string *ip = NULL, uint16_t *port = NULL) const{sockaddr_in peer_addr;socklen_t len = sizeof(peer_addr);int new_sock = accept(fd_, (sockaddr *)&peer_addr, &len);if (new_sock < 0){perror("accept");return false;}printf("accept fd = %d\\n", new_sock);peer->fd_ = new_sock;if (ip != NULL){*ip = inet_ntoa(peer_addr.sin_addr);}if (port != NULL){*port = ntohs(peer_addr.sin_port);}return true;}bool Recv(std::string *buf) const{buf->clear();char tmp[1024 * 10] = {0};// [注意!] 这里的读并不算很严谨, 因为一次 recv 并不能保证把所有的数据都全部读完// 参考 man 手册 MSG_WAITALL 节.ssize_t read_size = recv(fd_, tmp, sizeof(tmp), 0);if (read_size < 0){perror("recv");return false;}if (read_size == 0){return false;}buf->assign(tmp, read_size);return true;}bool Send(const std::string &buf) const{ssize_t write_size = send(fd_, buf.data(), buf.size(), 0);if (write_size < 0){perror("send");return false;}return true;}bool Connect(const std::string &ip, uint16_t port) const{sockaddr_in addr;addr.sin_family = AF_INET;addr.sin_addr.s_addr = inet_addr(ip.c_str());addr.sin_port = htons(port);int ret = connect(fd_, (sockaddr *)&addr, sizeof(addr));if (ret < 0){perror("connect");return false;}return true;}int GetFd() const{return fd_;}private:int fd_;
};

tcp_epoll_server.hpp

#pragma once
#include <vector>
#include <functional>
#include <sys/epoll.h>
#include "tcp_socket.hpp"typedef std::function<void(const std::string &, std::string *resp)> Handler;class Epoll
{
public:Epoll(){_epoll_fd = epoll_create(10);}~Epoll(){close(_epoll_fd);}bool Add(const TcpSocket &sock, bool epoll_et = false) const{int fd = sock.GetFd();printf("[Epoll Add] fd = %d\\n", fd);epoll_event ev;ev.data.fd = fd;if (epoll_et){ev.events = EPOLLIN | EPOLLET;}else{ev.events = EPOLLIN;}int ret = epoll_ctl(_epoll_fd, EPOLL_CTL_ADD, fd, &ev);if (ret < 0){perror("epoll_ctl ADD");return false;}return true;}bool Del(const TcpSocket &sock) const{int fd = sock.GetFd();printf("[Epoll Del] fd = %d\\n", fd);int ret = epoll_ctl(_epoll_fd, EPOLL_CTL_DEL, fd, NULL);if (ret < 0){perror("epoll_ctl DEL");return false;}return true;}bool Wait(std::vector<TcpSocket> *output) const{output->clear();epoll_event events[1000];int nfds = epoll_wait(_epoll_fd, events, sizeof(events) / sizeof(events[0]), -1);if (nfds < 0){perror("epool_wait");return false;}// [注意!] 此处必须是循环到 nfds, 不能多循环for (size_t i = 0; i < nfds; ++i){TcpSocket sock(events[i].data.fd);output->push_back(sock);}return true;}private:int _epoll_fd;
};class TcpEpollServer
{
public:TcpEpollServer(const std::string &ip, uint16_t port) : ip_(ip), port_(port){}TcpEpollServer(uint16_t port) : port_(port){}bool Start(Handler handler){// 1. 创建 socketTcpSocket listen_sock;CHECK_RET(listen_sock.Socket());// 2. 绑定CHECK_RET(listen_sock.Bind(ip_, port_));// 3. 监听CHECK_RET(listen_sock.Listen(5));// 4. 创建 Epoll 对象, 并将 listen_sock 加入进去Epoll epoll;epoll.Add(listen_sock);// 5. 进入事件循环for (;;){// 6. 进行 epoll_waitstd::vector<TcpSocket> output;if (!epoll.Wait(&output)){continue;}// 7. 根据就绪的文件描述符的种类决定如何处理for (size_t i = 0; i < output.size(); ++i){if (output[i].GetFd() == listen_sock.GetFd()){// 如果是 listen_sock, 就调用 acceptTcpSocket new_sock;listen_sock.Accept(&new_sock);epoll.Add(new_sock,true);}else{// 如果是 new_sock, 就进行一次读写std::string req, resp;bool ret = output[i].RecvNoBlock(&req);if (!ret){// [注意!!] 需要把不用的 socket 关闭// 先后顺序别搞反. 不过在 epoll 删除的时候其实就已经关闭 socket 了epoll.Del(output[i]);output[i].Close();continue;}handler(req, &resp);output[i].SendNoBlock(resp);printf("[client %d] req: %s, resp: %s\\n", output[i].GetFd(),req.c_str(), resp.c_str());} // end for}     // end for (;;)}return true;}private:std::string ip_;uint16_t port_;
};

server.cc

#include <unordered_map>
#include "tcp_epoll_server.hpp"
std::unordered_map<std::string, std::string> g_dict;
void Translate(const std::string &req, std::string *resp)
{auto it = g_dict.find(req);if (it == g_dict.end()){*resp = "未找到";return;}*resp = it->second;return;
}
int main(int argc, char *argv[])
{if (argc != 2){printf("Usage ./dict_server [port]\\n");return 1;}// 1. 初始化词典g_dict.insert(std::make_pair("hello", "你好"));g_dict.insert(std::make_pair("world", "世界"));g_dict.insert(std::make_pair("bit", "贼NB"));// 2. 启动服务器TcpEpollServer server(atoi(argv[1]));server.Start(Translate);return 0;
}