Linux学习记录——사십삼 高级IO（4）--- Epoll型服务器

---

1、理解Epoll和对应接口

poll依然需要OS去遍历所有fd。一个进程去多个特定的文件中等待，只要有一个就绪，就使用select/poll系统调用，让操作系统把所有文件遍历一遍，哪些就绪就加上哪些fd，再返回。一旦文件太多了，遍历效率就显而易见地低。epoll是为处理大批量句柄而作了改进的poll，句柄就是访问某种资源时标识这个资源的东西，比如C语言中的FILE结构体，文件描述符等。不过select/poll并不是没有用处，一些老型操作系统并不支持epoll，就得使用poll或者select。epoll是在Linux内核2.5.44时引入的，到现在为止都是Linux中最高效的多路转接IO方案。

epoll有3个接口。

size是一个被忽略的参数，只要大于0就行。如果成功，返回一个epoll文件描述符，在系统内部创建一些数据结构，帮助进行已就绪的fd的管理，暂且叫做epoll模型，失败返回-1。不用这个epoll文件描述符后要close(epollfd)。

创建后，用户要告诉内核，应当关心哪个文件描述符上的哪个事件是否就绪，select通过一个位图结构fd_set来实现，poll通过poll_fd来实现的。另外，内核要告诉用户，关心的哪些fd上的哪些事件event已经就绪了。epoll还有两个接口去做这两个事。

epfd就是创建函数的返回值；op表示想做什么，有3个值，EPOLL_ADD，EPOLL_MOD，EPOLL_DEL，分别是添加、修改、删除；fd表示哪一个fd，event表示这个fd上的哪个事件要被关心。

进行等待的接口。返回值和select，poll接口一样，就绪的fd数量；timeout的作用和poll一样，输入型参数，单位是毫秒ms，为0表示非阻塞，小于0表示阻塞，大于0poll在这段时间内阻塞等待，如果一直没有事件就绪，那么超过时间就返回0；中间两个参数是输出型参数，操作系统通过这两个告知用户就绪的fd上就绪的事件event。

events是一个32位整数，用户输入的是关心的事件，返回时操作系统通过这个整数来告诉用户哪些fd的events事件就绪了；data的类型是一个联合体，通常会使用prt或者fd。events有几种取值：

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

上面的就是宏。这里只关心EPOLLIN和EPOLLOUT。

TCP报头中6个标记位中有一个代表PSH，用来提示对方应用层立刻从接收缓冲区读取数据。但PSH并不一定能让应用层读取数据，它的催促是让套接字观察的fd对应的文件里的数据处于就绪状态。

操作系统可以把数据从应用层拷贝到缓冲区，然后将数据交给网卡。当网卡收到数据后，网卡会发送硬件中断，操作系统通过查看中断向量表，知道发来的中断号是网卡的，所以就知道网卡有了数据。select/poll都是在软件层面去检测是否有数据的。

CPU有对应的寄存器，寄存器是二进制序列，是一种存储单元，由硬件电路构成。数据拷贝到CPU的硬件本质是利用高低电频对CPU内的寄存器进行充放电，让CPU的寄存器变成和内存一样的值。CPU和所有外设之间都有针脚间接相连。发送中断就像是某个外设产生电流，从和它间接相连的针脚向寄存器充电，把数据放到寄存器中。之后网卡就可以发送中断号让CPU拷贝数据到内存了。所以数据是可以从外设拷贝到内存的。

用户层往下是系统调用层，再往下是操作系统，再往下就是传输层及以下了。当用户层创建epoll时，OS会维护一个红黑树，开始时只有一个根节点，并且epoll还会创建一个就绪队列，为空。红黑树的节点是结构体，里面有fd，有事件event，整个红黑树就是用户告诉OS，要关心哪些fd，以及fd上的哪些事件。所以可以看出epoll_ctl本质是对这个红黑树进行增删改，比如要删，就传对应的fd，事件设为nullptr/NULL，那就是对红黑树某个节点的删除。fd决定节点是红还是黑，左节点还是右节点，插入到哪里。内核中，一个数据结构对象，既可以属于红黑树，也可以属于另一个结构。

红黑树上只有某个fd上有对应的事件发生了，那么就把这个fd的节点接入到就绪队列中，队列只保存已经准备好的fd && 对应的event。队列每一个元素也可以是一个结构体，只取红黑树中已就绪节点里面的值来填充。epoll_wait接口中间两个参数就是从就绪队列中拿取节点，这个接口只看就绪队列，可以以时间复杂度为O(1)的方式来检测事件就绪，也就是队列是否为空。

节点放入队列实际不是将一个节点内容拷贝到队列节点里，而是红黑树节点也是队列节点，节点就是一个结构体，结构体里可以放入表示已经就绪的事件，放入红黑树相关指针信息，放入队列相关指针信息，建立起队列就是用这个队列相关的指针去指向下一个节点。

当数据就绪时，操作系统通过网卡，经过网络协议栈，拷贝到每个文件的文件缓冲区中。每个节点都有回调机制，假设每个文件结构体都有一个变量，如果没设置回调，就置为空，每次操作系统拷贝数据到缓冲区后就去判断一下这个变量，为空就退出，不为空就调用回调函数，回调函数做的工作就是把红黑树上已就绪的节点放到就绪队列中。

红黑树，就绪队列，回调机制这三个整体就是epoll模型，所以epoll_create使用时就是创建了这些，从操作系统内部到系统调用形成了一个体系。红黑树就像select/poll中的数组，但epoll这里核心的维护交由系统来做，不让用户去做。

为什么epoll_create要返回就绪fd的个数，以及另外两个接口还需要用这个数字？整个机制是由系统做的，接口是由进程调用的，进程在运行时，会创建task_struct指向文件描述符表files_struct，表里有一个数组，类型是struct file，012默认被占用，当创建epoll模型，操作系统也创建了一个struct file，里面有个指针指向epoll模型，这个struct file就在调用epoll接口的进程的文件描述符表中。用户，进程，task_struct，files_struct，struct file，这是一整个路线。通过epoll_create的返回值，也就是另外两个接口的参数epfd，两个接口就可以找到进程维护的文件描述符表，进而找到struct file，然后找到epoll模型，就可以对红黑树，就绪队列进行操作了。

epoll的红黑树比数组更有效率；也不需要底层在线性遍历所有节点；上层也不需要遍历节点只需要查看就绪队列；用户只需要调用接口就可以操作整个体系。

2、实现

Main.cc

#include "EpollServer.hpp"
#include <memory>int main()
{std::unique_ptr<EpollServer> svr(new EpollServer());svr->InitServer();svr->Start();return 0;
}

Makefile

epollserver:Main.ccg++ -o $@ $^ -std=c++11
.PHONY:clean
clean:rm -f epollserver

EpollServer.hpp中先写基础的

#pragma once#include <iostream>
#include <string>
#include "Sock.hpp"
#include "log.hpp"const static int gport = 8888;class EpollServer
{
public:EpollServer(uint16_t port = gport) : port_(port){}void InitServer(){listensock_.Socket();listensock_.Bind(port_);listensock_.Listen();}void Start(){while(true){sleep(3);}}~EpollServer(){}
private:uint16_t port_;Sock listensock_;
};

现在还不能Accept，因为还不知道底层是否有文件就绪，如果没有，整个服务器就得阻塞了。epoll这里的思路就是把自己的权利交给epoll。要将listensock添加到epoll中，不过得先有epoll模型。

创建一个Epoll.hpp

#pragma once#include <iostream>
#include <string>
#include <sys/epoll.h>static const  int defaultepfd = -1;class Epoller
{
public:Epoller():epfd_(defaultepfd){}~Epoller(){}
private:int epfd_;
};

完善一下Epoll模型，并初始化和析构

Epoll.hpp

#pragma once#include <iostream>
#include <string>
#include <cstdlib>
#include <sys/epoll.h>
#include "err.hpp"static const  int defaultepfd = -1;
static const int gsize = 128;class Epoller
{
public:Epoller():epfd_(defaultepfd){}void Create(){epfd_ = epoll_create(gsize);if(epfd_ < 0){logMessage(Fatal, "epoll_create error, code: %d, errstring: %s", errno, strerror(errno));exit(EPOLL_CREAT_ERR);//err.hpp里加上这个错误}}int Fd(){return epfd_;}void Close(){if(epfd_ != defaultepfd) close(epfd_);}~Epoller(){}
private:int epfd_;
};

EpollServer.hpp

#pragma once#include "Epoll.hpp"
#include "Sock.hpp"
#include "log.hpp"const static int gport = 8888;class EpollServer
{
public:EpollServer(uint16_t port = gport) : port_(port){}void InitServer(){listensock_.Socket();listensock_.Bind(port_);listensock_.Listen();epoller_.Create();logMessage(Debug, "init server success");}void Start(){//1、将listensock添加到epoll中，要先有epoll模型while(true){sleep(3);}}~EpollServer(){listensock_.Close();epoller_.Close();}
private:uint16_t port_;Sock listensock_;Epoller epoller_;
};

接下来关注事件。

Epoll.hpp

    //用户告诉内核要关心哪些事件bool AddEvent(int fd, uint32_t events){struct epoll_event ev;ev.events = events;ev.data.fd = fd;//fd就是就绪的文件描述符int n = epoll_ctl(epfd_, EPOLL_CTL_ADD, fd, &ev);if(n < 0){logMessage(Fatal, "epoll_ctl error, code: %d, errstring: %s", errno, strerror(errno));return false;}return true;}

EpollServer.hpp

    void Start(){//1、将listensock添加到epoll中，要先有epoll模型bool r = epoller_.AddEvent(listensock_.Fd(), EPOLLIN);//只关心读事件assert(r);//可以做别的判断(void)r;while(true){;}}

然后就可以在循环中获取事件了，使用wait。从队列里拿数据这个过程是线性拷贝的，因为系统不相信用户，所以要定义一个struct epoll_event类型的数组来接收。以及wait接口中的events参数里，由于拷贝的缘故，数据是从左到右连续有效的，而返回值 - 1就是当前最后一个有效的下标。

EpollServer.hpp

    void Start(){//1、将listensock添加到epoll中，要先有epoll模型bool r = epoller_.AddEvent(listensock_.Fd(), EPOLLIN);//只关心读事件assert(r);//可以做别的判断(void)r;struct epoll_event revs_[gnum];int timeout = 1000;while(true){int n = epoller_.Wait(revs_, gnum, timeout);switch (n){case 0:logMessage(Debug, "timeout...");break;case -1:logMessage(Warning, "epoll_wait failed");break;default:logMessage(Debug, "有%d个事件就绪了", n);HandlerEvents(n);//一定有数据就绪break;}}}void HandlerEvents(int num){for(int i = 0; i < num; i++){int fd = revs_[i].data.fd;uint32_t events = revs_[i].events;logMessage(Debug, "当前正在处理%d上的%s", fd, (events&EPOLLIN) ? "EPOLLIN" : "OTHER");int fd = revs_[i].data.fd;uint32_t events = revs_[i].events;if(events & EPOLLIN)//判断读事件就绪{if (fd == listensock_.Fd()){// 1、新连接到来std::string clientip;uint16_t clientport;int sock = listensock_.Accept(&clientip, &clientport);if (sock < 0)continue;logMessage(Debug, "%s:%d 已经连上服务器了", clientip.c_str(), clientport);// 还不能recv，即使有了连接但也不知道有没有数据// 只有epoll知道具体情况，所以将sock添加到epoll中bool r = epoller_.AddEvent(sock, EPOLLIN);assert(r);(void)r;}else // 2、读事件{char buffer[1024];ssize_t s = recv(fd, buffer, sizeof(buffer) - 1, 0);if (s > 0){buffer[s - 1] = 0;//对打印格式buffer[s - 2] = 0;//做一下调整std::string echo = buffer;echo += " [epoll server echo]\r\n";std::cout << "client# " << echo << std::endl;send(fd, echo.c_str(), echo.size(), 0);}else{if (s == 0)logMessage(Info, "client quit ...");elselogMessage(Warning, "recv error, client quit...");close(fd);//将文件描述符移除//在处理异常的时候，fd必须合法才能被处理epoller_.DelEvent(fd);}}}}}

Epoll.hpp

    //用户告诉内核要关心哪些事件bool AddEvent(int fd, uint32_t events){struct epoll_event ev;ev.events = events;ev.data.fd = fd;//属于用户的数据，epoll底层不对该数据做任何修改，为了给未来就绪返回int n = epoll_ctl(epfd_, EPOLL_CTL_ADD, fd, &ev);if(n < 0){logMessage(Fatal, "epoll_ctl error, code: %d, errstring: %s", errno, strerror(errno));return false;}return true;}bool DelEvent(int fd){return epoll_ctl(epfd_, EPOLL_CTL_DEL, fd, nullptr) == 0;}int Wait(struct epoll_event* revs, int num, int timeout){return epoll_wait(epfd_, revs, num, timeout);}

读事件处理中，我们目前无法读到一个完整的报文。因为完整报文由应用层协议规定，我们的代码没有应用层协议，所以得自定义一个。

先用回调函数来处理数据

#include <functional>
using func_t = std::function<std::string (std::string)>;public:EpollServer(func_t func, uint16_t port = gport) : func_(func), port_(port){}
private:uint16_t port_;Sock listensock_;Epoller epoller_;struct epoll_event revs_[gnum];func_t func_;

读事件处理时

                else // 2、读事件{char request[1024];ssize_t s = recv(fd, request, sizeof(request) - 1, 0);if (s > 0){request[s - 1] = 0;//对打印格式request[s - 2] = 0;//做一下调整std::string response = func_(request);send(fd, response.c_str(), response.size(), 0);}else{if (s == 0)logMessage(Info, "client quit ...");elselogMessage(Warning, "recv error, client quit...");close(fd);//将文件描述符移除//在处理异常的时候，fd必须合法才能被处理epoller_.DelEvent(fd);}}

在Main.cc中传入函数

#include "EpollServer.hpp"
#include <memory>std::string echoServer(std::string r)
{std::string resp = r;resp += "[echo]\r\n";return resp;
}int main()
{std::unique_ptr<EpollServer> svr(new EpollServer(echoServer));svr->InitServer();svr->Start();return 0;
}

下一篇仍然是Epoll代码，但基于本篇代码所写，体现出Reactor的思路。

基本版Epoll

结束。