【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)

NAME
       epoll_create, epoll_create1 - open an epoll file descriptor

SYNOPSIS
       #include<sys/epoll.h>intepoll_create(int size);intepoll_create1(int flags);

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

epoll_ctl

EPOLL_CTL(2)                                                              Linux Programmer's Manual                                                              EPOLL_CTL(2)

NAME
       epoll_ctl - control interface for an epoll file descriptor

SYNOPSIS
       #include<sys/epoll.h>intepoll_ctl(int epfd,int op,int fd,structepoll_event*event);

DESCRIPTION
       This  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 that
       the 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，但内部更加复杂：typedefunion epoll_data {void*ptr;int fd;uint32_t u32;uint64_t u64;} epoll_data_t;structepoll_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)

NAME
       epoll_wait, epoll_pwait - wait for an I/O event on an epoll file descriptor

SYNOPSIS
       #include<sys/epoll.h>intepoll_wait(int epfd,structepoll_event*events,int maxevents,int timeout);intepoll_pwait(int epfd,structepoll_event*events,int maxevents,int timeout,const sigset_t *sigmask);

DESCRIPTION
       The 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操作。

#pragmaonce#include<string>#include<iostream>#include<memory>#include<sys/epoll.h>#include"Log.hpp"#include"Socket.hpp"usingnamespace log_ns;usingnamespace socket_ns;classEpollServer{private:conststaticint gnum =1024;conststaticint 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);}}voidInitServer(){// 将监听套接字放入epoll模型structepoll_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);}}voidAccepter(){}voidHandlerIO(int fd){// 普通fdIO 就绪}voidHandlerEvent(int n){}voidLoop(){}~EpollServer(){// 关闭epoll模型if(_epollfd >0)close(_epollfd);// 关闭监听套接字
        _listensock->Close();}private:// 端口号uint16_t _port;// 套接字
    std::unique_ptr<Socket> _listensock;// epoll模型描述符int _epollfd;// 文件描述符structepoll_event revs[gnum];};

Loop 循环函数，设置

timeout

调用

epoll_wait

接口进行等待事件就绪 ，将就绪的事件放入到revs数组中。根据返回值进行判断结果：

voidLoop(){int timeout =2000;while(true){// 进行等待int n =::epoll_wait(_epollfd, revs, gnum, timeout);// 判断结果switch(n){case0: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;}voidAccepter(){// 获取_listensock的新fd
        InetAddr 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模型中structepoll_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);}}voidHandlerIO(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);// 临时方案}elseif(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);}}voidHandlerEvent(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的基础服务，来看效果：

非常好！！！