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

文章目录

1、理解Epoll和对应接口
2、简单实现

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.cc
	g++ -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"
#include "log.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");
            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 ...");
                        else
                            logMessage(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 ...");
                        else
                            logMessage(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;
}