【Linux】从零开始使用多路转接IO --- epoll

当你偶尔发现语言变得无力时，

不妨安静下来，

让沉默替你发声。

--- 里则林 ---

---

1 epoll的作用和定位

之前提过的多路转接方案select和poll 都有致命缺点：底层都是暴力的遍历，效率不高！
对此，诞生出了epoll这个更好的方案！

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

2 epoll 的接口

epoll的相关接口有三个：

epoll_create

EPOLL_CREATE(2)                Linux Programmer's Manual                                                           EPOLL_CREATE(2)NAMEepoll_create, epoll_create1 - open an epoll file descriptorSYNOPSIS#include <sys/epoll.h>int epoll_create(int size);int epoll_create1(int flags);

epoll_create接口只有一个参数，其功能是在内核创建一个epoll模型！这个模型我们后面详细谈。这个size我们只有设置为一个大于零的数即可。创建成功之后会给我们返回一个文件描述符，现在我们还理解不了，后续讲解。

epoll_ctl

EPOLL_CTL(2)                                                              Linux Programmer's Manual                                                              EPOLL_CTL(2)NAMEepoll_ctl - control interface for an epoll file descriptorSYNOPSIS#include <sys/epoll.h>int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);DESCRIPTIONThis  system  call  is used to add, modify, or remove entries in the interest list of the epoll(7) instance referred to by the file descriptor epfd.  It requests thatthe operation op be performed for the target file descriptor, fd.

epoll_ctl有四个参数：

1. int epfd：这个就是通过epoll_create获得的文件描述符
2. int op：这个是操作选项，我们这个函数共用三种选项:EPOLL_CTL_ADD增加 EPOLL_CTL_MOD 修改EPOLL_CTL_DEL删除。
3. int fd：对这个文件描述符进行操作。
4. struct epoll_event * event：这时一个结构体，类似struct pollfd，但内部更加复杂：

   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位图就可以设置读事件，写事件…注意这里没有返回事件！

epoll_wait

EPOLL_WAIT(2)                                                             Linux Programmer's Manual                                                             EPOLL_WAIT(2)NAMEepoll_wait, epoll_pwait - wait for an I/O event on an epoll file descriptorSYNOPSIS#include <sys/epoll.h>int epoll_wait(int epfd, struct epoll_event *events,int maxevents, int timeout);int epoll_pwait(int epfd, struct epoll_event *events,int maxevents, int timeout,const sigset_t *sigmask);DESCRIPTIONThe epoll_wait() system call waits for events on the epoll(7) instance referred to by the file descriptor epfd.  The buffer pointed to by events is used to return in‐formation from the ready list about file descriptors in the interest list that have some events available.  Up to maxevents are returned by  epoll_wait().   The  max‐events argument must be greater than zero.

epoll_wait有四个参数：

1. int epfd：这个就是通过epoll_create获得的文件描述符。
2. *struct epoll_event events ：这是一个数组，向内核输入一个缓冲区，想让内核提供这个数组将就绪事件返回来！
3. ** int maxevents**：数组的元素个数。
4. int timeout：等价于poll接口的timeout，以毫秒为单位！
5. 返回值等价于poll！

总而言之：epoll将传入与传出分成了两个接口来进行！

3 epoll工作原理

对于epoll更深入的理解我们需要从底层进行讲解：

数据到达主机时，数据首先会到达物理层，那么操作系统如何知道网卡里有数据呢？通过硬件中断！通过针脚中断，就可以通知操作系统！从而数据链路层从网络层读取数据！

当我们使用epoll时，系统内部会建立一个红黑树，这个红黑树创建时是空树。红黑树的节点字段主要存在：文件描述符fd , 事件位图 events ，左右指针，节点颜色...，这个树标识了用户想让OS关心的文件操作符fd以及其对应事件！epoll_ctl接口中的op就是对应的增添修改删除红黑树节点！注意：这个红黑树的键值是fd!

其中还有一个就绪队列，这是一个双向链表，每个节点与红黑树中的节点类似。当网卡中有数据了，网卡通过硬件中断把数据交给网络协议栈。OS可以知道每个文件描述符对应的输入输出缓冲区状态，当回红黑树节点对应fd的EPOLLIN事件等就绪，那么OS就把这个fd的事件放入就绪队列。这个就绪队列就是储存就绪事件的数据结构，当用户调用epoll_wait时，就通过就绪队列进行检测哪个fd对应事件就绪了！将事件依次严格按照顺序放入struct epoll_event *events数组中！

这个检测就绪事件的算法的时间复杂度就是O(1)！只需要判断就绪队列是否为空就可以！而将就绪事件获取的时间复杂度是O(n)!

这就是epoll模型！！！

而这个epoll模型是可以打开多个的，就和打开多个文件一样。当我们打开多个epoll模型时，那么操作系统如何管理这些epoll模型呢？

在内核中有一个eventpoll，这个是描述epoll模型的结构体，其中就有rbr红黑树与rdllist就绪队列。那为什么创建epoll模型之后会返回一个文件描述符呢？

在内核中有无数个task_struct进程结构体，每个进程都有一张文件描述符表struct files_struct，这个表的元素就指向文件结构体struct file。文件结构体中就有一个指针指向epoll模型。那么在进程中想要找到epoll模型就可以通过文件描述符表找到epoll模型！

我们来谈一个十分巧妙的设计。在epoll模型中，存在红黑树和就绪队列。每个节点都有对应的文件描述符。在之前所学的数据结构中，我们每个数据结构的节点都是独属于自身的，比如二叉树的节点不可能是链表的节点。
但是在epoll模型中，一个节点是可以属于多个数据结构的！我们来看是如何实现的：

1. 首先，有这样一个链表节点listnode，其中只包含左右指针。
2. 然后在task_struct中，就可以存在listnode link，那么每一个task_struct就可以通过这个link进行连接起来的。
3. 但是，这个指向的只是下一个task_struct结构体中的link，那么怎么才能访问task_struct全部的数据呢？
4. 可以先计算这个link在task_struct的偏移量，通过将0地址强制类型转换，得到里面link的地址，就知道了偏移量！然后通过task_struct中link里的指针减去偏移量，我们就得到了task_struct的起始地址，再进行类型转换我们就得到了task_struct！
5. 同样的，task_struct还可以存在二叉树节点link2 , 队列节点link3，就都可以通过这种方式进行链接，并且是一个节点属于了多个数据结构中！！！

这是十分巧妙的设计！！！而epoll模型中的epitem结构体就是这样设计的！一个节点既属于红黑树，也属于就绪队列！

其中epitem还有一个status变量，表示其是否被激活。可以判断是否在红黑树或者就绪队列中！
下面我们开始编写v1版本的epollserver

4 实现epollserverV1

下面我们来实现epollserver：

成员变量需要以下：

1. 端口号_port ：用于创建listen套接字
2. 套接字socket ：_listensock监听套接字，使用TCP进行通信。
3. 文件描述符_epfd ：epoll模型的文件操作符，是使用epoll系列接口的必要参数。
4. epoll_event revs[] 数组：从epoll模型中获取就绪事件的结构体数组。

根据成员变量，进行构造，创建套接字，创建epoll模型。
初始化函数中，建立struct epoll_event ev设置其中的 fd 与events位图；先将_listensock套接字fd添加到epoll中 通过epoll_ctl进行ADD操作。

#pragma once#include <string>
#include <iostream>
#include <memory>
#include <sys/epoll.h>#include "Log.hpp"
#include "Socket.hpp"using namespace log_ns;
using namespace socket_ns;class EpollServer
{
private:const static int gnum = 1024;const static int size = 128;public:EpollServer(uint16_t port) : _port(port),_listensock(std::make_unique<TcpSocket>()){// 建立监听套接字_listensock->BuildListenSocket(port);// 建立epoll模型_epollfd = ::epoll_create(size);if (_epollfd < 0){// 创建失败LOG(FATAL, "epoll_create failed!\n");exit(1);}}void InitServer(){// 将监听套接字放入epoll模型struct epoll_event ev;ev.data.fd = _listensock->GetSockfd();ev.events = EPOLLIN;// 放入int n = ::epoll_ctl(_epollfd, EPOLL_CTL_ADD, _listensock->GetSockfd(), &ev);// 根据返回值判断if (n < 0){// 发生错误LOG(FATAL, "epoll_ctl failed ,errno :%d", errno);exit(1);}}void Accepter(){}void HandlerIO(int fd){// 普通fdIO 就绪}void HandlerEvent(int n){}void Loop(){}~EpollServer(){// 关闭epoll模型if (_epollfd > 0)close(_epollfd);// 关闭监听套接字_listensock->Close();}private:// 端口号uint16_t _port;// 套接字std::unique_ptr<Socket> _listensock;// epoll模型描述符int _epollfd;// 文件描述符struct epoll_event revs[gnum];
};

Loop 循环函数，设置timeout 调用epoll_wait接口进行等待事件就绪 ，将就绪的事件放入到revs数组中。根据返回值进行判断结果：

void Loop(){int timeout = 2000;while (true){// 进行等待int n = ::epoll_wait(_epollfd, revs, gnum, timeout);// 判断结果switch (n){case 0:LOG(INFO, "epoll timeout...\n");break;case -1:LOG(ERROR, "epoll error\n");break;default:LOG(INFO, "haved event happened! , n :%d\n", n);// 处理事件HandlerEvent(n);break;}}}

HandlerEvent处理事件，将数组中的n个事件全部处理遍历一遍， 根据就绪的文件描述符种类进行区分判断 （设计一个简单的接口可以通过事件级返回事件种类）；读事件就绪 我们进行处理

• _listensock套接字事件获取连接 Accepter 将新的fd加入到epoll模型 打印客户端信息
• 普通fd 事件HandlerIO 进行读取recv ;读取失败的话要从epoll删除后再close ，处理后Send回去。

    std::string PrintEvent(uint32_t revents){std::string ret;if (revents & EPOLLIN)ret += "EPOLLIN";if (revents & EPOLLOUT)ret += "| EPOLLOUT";return ret;}void Accepter(){// 获取_listensock的新fdInetAddr addr;int sockfd = _listensock->Accepter(&addr);if (sockfd < 0){LOG(ERROR, "Accepter error\n");exit(1);}// 成功获取连接LOG(INFO, "成功获取连接 ,客户端： %s\n", addr.AddrStr().c_str());// 将连接添加到epoll模型中struct epoll_event ev;ev.data.fd = sockfd;ev.events = EPOLLIN;int n = ::epoll_ctl(_epollfd, EPOLL_CTL_ADD, sockfd, &ev);// 根据返回值判断if (n < 0){// 发生错误LOG(FATAL, "epoll_ctl failed ,errno :%d", errno);exit(1);}}void HandlerIO(int fd){// 普通fdIO 就绪char buffer[4096];int n = ::recv(fd, buffer, sizeof(buffer), 0);if (n > 0){// 读取到了数据buffer[n] = 0;std::string echo_str = "[client say]#";echo_str += buffer;std::cout << echo_str << std::endl;// 返回一个报文std::string content = "<html><body><h1>hello bite</h1></body></html>";std::string ret_str = "HTTP/1.0 200 OK\r\n";ret_str += "Content-Type: text/html\r\n";ret_str += "Content-Length: " + std::to_string(content.size()) + "\r\n\r\n";ret_str += content;// echo_str += buffer;::send(fd, ret_str.c_str(), ret_str.size(), 0); // 临时方案}else if (n == 0){// 此时fd退出了LOG(INFO, "fd:%d quit!\n", fd);//先对epoll中的节点进行删除，因为epoll中的节点必须是合法fd ,不能进行close::epoll_ctl(_epollfd , EPOLL_CTL_DEL , fd , nullptr);::close(fd);}else{LOG(ERROR, "recv error! errno:%d\n", errno);::epoll_ctl(_epollfd , EPOLL_CTL_DEL , fd , nullptr);::close(fd);}}void HandlerEvent(int n){// 处理事件for (int i = 0; i < n; i++){int fd = revs[i].data.fd;uint32_t revents = revs[i].events;LOG(INFO, "fd:%d , %s事件就绪\n", fd, PrintEvent(revents).c_str());// 判断fd类型if (fd == _listensock->GetSockfd()){// 进行AccepterAccepter();}// 普通fdelse{HandlerIO(fd);}}}

这样我们就成功的完成了epollserver的基础服务，来看效果：

非常好！！！