【网络编程】多路转接之select、poll、epoll

一、如何进行高效的IO

以read/recv为例

二、五种IO模型

三、常见的高级IO

四、非阻塞IO

1、recv和send自带的非阻塞IO

2、可以将文件描述符设为非阻塞

2.1open自带的非阻塞IO

2.2通过fcntl函数将一个文件描述符设置为非阻塞

2.3非阻塞IO的返回值判定

三、多路转接——select

1、select的原型

2、select所等待fd的三种就绪状态

2.1读就绪

2.2写就绪

2.3异常就绪

3、写一个select服务器

4、select的特点

四、多路转接——poll

1、针对select的问题，poll的解决方案

2、poll的原型

3、poll服务器部分代码（读事件）

4、poll的优缺点

4.1优点

4.2缺点

五、多路转接——epoll

1、epoll的三个系统调用

1.1epoll_create（创建epoll模型）

1.2epoll_ctl（对红黑树进行增删改）

1.3epoll_wait（从就绪队列取事件）

2、图解epoll事件就绪的过程

3、epoll底层的通知模式

3.1LT模式和ET模式

3.2为什么ET模式比LT模式高效

4、epoll服务器代码（读事件）

一、如何进行高效的IO

以read/recv为例

1、没有数据，进程/线程将会被阻塞在IO函数中（等待资源就绪）

2、有数据，read和recv会在拷贝完成之后返回

IO=等待+拷贝，所以提高IO的速度的方法就是减少等待时间。而恰巧网络中出现最频繁的事件，就是数据IO，想要提高IO效率，就必须减少资源等待的时间。

二、五种IO模型

以钓鱼为例：

张三：一人一杆，眼睛时刻死死的盯着鱼漂，观察是否有鱼上钩；

李四：每隔一段时间来看一下鱼漂的情况，有鱼就提竿，没鱼就继续低头玩手机等下一波轮询；

王五：在鱼漂上挂了个铃铛，铃铛响了就知道鱼上钩了；

赵六：一个人带了上百根杆子，哪根竿上钩了提哪根；

小王：是个老板，雇佣田七钓鱼，田七钓满一桶打电话通知老板来拿鱼。

钓鱼的例子：钓鱼人是进程/线程；鱼漂是数据就绪的事件；鱼竿是文件描述符；钓鱼的动作是read/recv的调用；被雇佣钓鱼的人是操作系统；鱼是数据；河是内核空间。

1、阻塞IO: （钓鱼人死死的盯着鱼漂，直到鱼咬钩）在内核将数据准备好之前, 系统调用会一直等待，直到被等待的事件准备就绪。所有的套接字, 默认都是阻塞方式。（常用）

2、非阻塞IO: （钓鱼人定期查看鱼漂的情况）如果内核还未将数据准备好, 系统调用仍然会直接返回, 并且返回EWOULDBLOCK错误码。非阻塞IO往往需要程序员循环的方式反复尝试读写文件描述符, 这个过程称为轮询（轮询，对CPU消耗大，特定场景使用）

3、信号驱动IO: （钓鱼挂铃铛，由铃铛声音进行上鱼通知）内核将数据准备好的时候, 使用SIGIO信号（需要进行捕捉并注册SIGIO的回调方法）通知应用程序进行IO操作。（多个信号同时就绪可能会出现信号丢失问题）

4、IO多路转接: （一人多竿）看起来和阻塞IO类似。实际上最核心在于IO多路转接能够同时等待多个文件描述符的就绪状态。（等的比重低，高效）

5、异步IO: （雇佣工具人钓鱼） 异步IO不参与等+拷贝，由内核在数据拷贝至缓冲区完成时, 通知应用程序(而信号驱动是告诉应用程序何时可以开始拷贝数据)。

对于第1~3种IO模型，在效率上没有差别，但是第二种和第三种可以在数据未就绪时做其他事情。

第1~4种IO模型都参与了“钓鱼”的过程，称之为同步IO。对于第五种IO，并没有参与IO的两个过程（等+拷贝），被称为异步IO。

三、常见的高级IO

非阻塞IO，纪录锁，系统V流机制，I/O多路转接（也叫I/O多路复用）,readv和writev函数以及存储映射IO（mmap），这些统称为高级IO。

四、非阻塞IO

1、recv和send自带的非阻塞IO

2、可以将文件描述符设为非阻塞

2.1open自带的非阻塞IO

2.2通过fcntl函数将一个文件描述符设置为非阻塞

#include <unistd.h>
#include <fcntl.h>
int fcntl(int fd, int cmd, ... /* arg */ );

用途：fcntl用于在运行时更改文件描述符的属性或执行一些与文件描述符相关的操作。这包括启用非阻塞模式、获取/设置文件状态标志、获取/设置文件锁等。

参数：

fd：要操作的文件描述符。
cmd：复制一个现有的描述符（cmd=F_DUPFD）.

获得/设置文件描述符标记(cmd=F_GETFD或F_SETFD).

获得/设置文件状态标记(cmd=F_GETFL或F_SETFL).

获得/设置异步I/O所有权(cmd=F_GETOWN或F_SETOWN).

获得/设置记录锁(cmd=F_GETLK,F_SETLK或F_SETLKW)

返回值：

对于获取操作（如F_GETFD、F_GETFL、F_GETLK等），返回相应的信息或标志。
对于设置操作（如F_SETFD、F_SETFL、F_SETLK等），通常返回0表示成功，-1表示失败，并设置 errno 以指示错误的原因。

使用fcntl将一个文件描述符设置为非阻塞：

void SetNoBlock(int fd) 
{ 
    int fl = fcntl(fd, F_GETFL); 
    if (fl < 0) 
    { 
        //perror("fcntl");
        std::cerr << "fcntl:" << strerror(errno) << std::endl;
        return; 
    }
    fcntl(fd, F_SETFL, fl | O_NONBLOCK); 
}

2.3非阻塞IO的返回值判定

由于fcntl可以让文件描述符由阻塞变为非阻塞，这就会出现一种情况：IO接口，例如read函数通过非阻塞文件描述符进行数据的读取，如果没有读到数据，返回值是-1（这是一种正常的情况），如果读取出错，返回值也是-1。Linux系统中通过设置errno以区分这种情况，程序员通过判断errno的值即可分辨出IO接口的返回情况。

EAGAIN or EWOULDBLOCK：表示数据未就绪，并不代表IO出错
EINTR：读取时被信号所中断，并不代表IO出错
出现其他的宏可以认定IO接口是真的出错了

三、多路转接——select

select系统调用是让程序监视多个文件描述符的状态变化的;

程序会停在select这里等待，直到被监视的文件描述符有一个或多个发生了状态改变。

select只负责等。

1、select的原型

#include <sys/select.h>
int select(int nfds, fd_set *readfds, fd_set *writefds, fd_set *exceptfds, 
			struct timeval *timeout);

部分参数类型的意义：

fd_set：是一种位图结构，表示文件描述符的集合，
其中比特位的位置，表示fd的值，比特位的内容，表示哪些fd需要被关心（输入）或已经就绪（输出）
使用Linux操作系统自带的位图操作对这几个参数进行操作
void FD_CLR(int fd, fd_set *set);//把一个fd从集合当中清除
int  FD_ISSET(int fd, fd_set *set);//判断一个fd是否在集合里
void FD_SET(int fd, fd_set *set);//把一个fd设置到集合里
void FD_ZERO(fd_set *set);//把一个文件描述符集合全部清零

struct timeval {
   time_t      tv_sec;     /* seconds */
   suseconds_t tv_usec;    /* microseconds */
};

用途：系统提供select函数来实现多路复用输入/输出模型。

select系统调用能够让程序监视多个文件描述符状态是否发生变化;

程序会停在select这里等待，直到被监视的文件描述符有一个或多个发生了状态改变。

参数：

nfds：select待监视的文件描述符中最大的文件描述符值加1。
readfds：输入输出型参数。检查可读性。输入：表示用户输入一个位图结构，想让内核关心一下其中标志位被设为1的fd的读事件。输出：内核将告诉用户，用户所关心的fd中有哪些就绪了。
writefds：输入输出型参数。检查可写性。同上。
exceptfds：输入输出型参数。检查异常条。同上。
timeout：输入输出型参数。用于设置超时时间，如果设置为nullptr，select函数将会一直阻塞直到有事件发生。如果传入的结构体是struct timeval = {0,0}；表示select非阻塞，不管有没有等到变化的fd，均立刻超时返回。如果传入的结构体是struct timeval = {5,0}；表示阻塞5秒，期间等到了就立刻返回，5秒到了还没等到的话就不等了，超时返回，需要注意的是，timeout单次计时时间减到了0，需要手动更新timeout，否则timeout将一直是0。

返回值：

返回值大于零，返回值代表满足条件（可读、可写、异常）的文件描述符总数。
返回值等于零，代表超时返回
出错时返回-1，同时设置errno来指示具体的错误原因。

2、select所等待fd的三种就绪状态

2.1读就绪

1、socket内核中, 接收缓冲区中的字节数, 大于等于低水位标记SO_RCVLOWAT，此时可以无阻塞读取该文件描述符, 且返回值大于0;

2、socket TCP通信中, 对端关闭连接, 此时对该socket读, 则返回0;

3、监听的socket上有新的连接请求;

4、socket上有未处理的错误;

2.2写就绪

1、socket内核中, 发送缓冲区中的可用字节数(发送缓冲区的空闲位置大小), 大于等于低水位标记SO_SNDLOWAT, 此时可以无阻塞的写, 并且返回值大于0;

2、socket的写操作被关闭(close或者shutdown)。对一个写操作被关闭的socket进行写操作, 会触发SIGPIPE信号;

3、socket使用非阻塞connect连接成功或失败之后;

4、socket上有未读取的错误;

2.3异常就绪

socket上收到带外数据。关于带外数据, 和TCP紧急模式相关(TCP协议头中, 有一个紧急指针的字段)

3、写一个select服务器

对于服务器中的listenSocket（监听套接字），它是能交给select来管理的，listenSocket的连接就绪事件等价于读事件就绪。如果没有及时取走就绪的套接字，select的返回值将一直是就绪的。

//select阻塞式等待，若监听到了，但没有取走就绪的fd，select的返回值将一直是就绪的
int n = select(maxFd + 1, &rfds, nullptr, nullptr, nullptr);

所以当select获取到新连接后，及时处理这个fd对应的事件，此处select发现_listenSocket就绪，就去事件处理函数中处理_listenSocket的逻辑。

在处理完_listenSocket事件后，我们获得了一个新的sock，但是我们不能write/read，因为整个代码只有select有资格检测事件是否就绪，所以要把这个fd交给select来管理。程序员需要自己维护一个合法的fd数组，这个fd数组的容量上限取决于fd_set类型的大小【sizeof(fd_set)*8 = 128字节 = 1024比特位】：

对于服务器的accept来说，也是阻塞式的等待客户端的连接，这里也可以让select进行等，当select等到了新连接，再让accept只用来获取新连接。

#pragma once
#include <functional>
#include "Sock.hpp"
//读到什么返回时什么
static const uint16_t defaultPort = 8080;
static const int fdNum = sizeof(fd_set) * 8;//fd_set是位图结构，*8表示共有几位
static const int defaultFd = -1;
static const int bufferNum = 1024;
using func_t = std::function<std::string(const std::string)>;//传入读到的数据，返回处理后的数据
class SelectServer
{
public:
    SelectServer(func_t func, uint16_t port = defaultPort)
        :_port(port)
        ,_listenSocket(-1)
        ,_fdArray(nullptr)
        ,_func(func)
    {}
    void InitServer()
    {
        _listenSocket = Sock::Socket();
        Sock::Bind(_listenSocket, _port);
        Sock::Listen(_listenSocket);
        _fdArray = new int[fdNum];
        for (int i = 0; i < fdNum; ++i)//清空fd数组
        {
            _fdArray[i] = defaultFd;
        }
        _fdArray[0] = _listenSocket;
        LogMessage(NORMAL,"create sock success %d", _listenSocket);
    }
    void Start()
    {
        while(1) 
        {
            //定义读文件描述符集
            fd_set rfds;
            //清空读文件描述符集，每次循环都需要重新清空
            FD_ZERO(&rfds);
            //最大的文件描述符，给select传参用
            int maxFd = defaultFd;
            for (int i = 0; i < fdNum; ++i)
            {
                if(_fdArray[i] == defaultFd) continue;
                //将合法fd全部添加到读文件描述符集中
                FD_SET(_fdArray[i], &rfds);
                if(maxFd < _fdArray[i]) maxFd = _fdArray[i];//更新所有fd中的最大fd
            }
            //select多路转接
            //struct timeval timeout = {3, 0};//最多阻塞3秒就让select返回
            //int n = select(_listenSocket + 1, &rfds, nullptr, nullptr, &timeout);//timeout会从三秒减到0，需要手动更新，否则一直是0
            int n = select(maxFd + 1, &rfds, nullptr, nullptr, nullptr);//select阻塞式等待
            switch(n)
            {
            case 0://超时返回
                LogMessage(NORMAL, "timeOut...");
                break;
            case -1://select失败
                LogMessage(WARNING, "select error, code: %d, err string:%s", errno, strerror(errno));
                break;
            default://事件就绪
                LogMessage(NORMAL, "have event ready");//如果没有把底层的连接取走，那么_listenSocket将一直就绪
                HandlerEvent(rfds);
                //HandlerWriteEvent(wfds)//写事件回调
                //HandlerErrorEvent(efds)//读事件回调
                break;
            }
        }
    }
    void ListenEvent(int listenSocket)//处理_listenSocket并将新获得的fd加入合法数组中
    {
        //select告诉我，_listenSocket读事件就绪
        std::string clientIp;
        uint16_t clientPort = -1;
        int sock = Sock::Accept(listenSocket, &clientIp, &clientPort);//accept ==等 + 获取  
        if(sock < 0) 
        {
            return;
        }
        //此处不能write/read，因为整个代码只有select有资格检测事件是否就绪.此处应该将新的sock托管给select
        //在数组中寻找新的fd坑位
        int i = 0;
        for(; i < fdNum; ++i)
        {
            if(_fdArray[i] != defaultFd) continue;
            else break;
        }
        if(i == fdNum)//说明文件描述符被写满了
        {
            LogMessage(WARNING, "Server is full, please wait");
            close(sock);//满载关闭消息
        }
        else
        {
            _fdArray[i] = sock;
        }
    }
    void Read(int sock, int pos)//pos是文件描述符在数组中的位置
    {
        //1、读取request（这样读取存在问题，无法保证读到一个完整报文）
        //并且同时可能有多个文件描述符就绪，同时调用这段代码
        char buffer[bufferNum];
        ssize_t s = recv(sock, buffer, sizeof(buffer)-1, 0);
        if(s > 0)
        {
            buffer[s] = 0;
            LogMessage(NORMAL, "client# %s", buffer);
        }
        else if(0 == s)//连接关闭
        {
            close(sock);//close之后，需要在所维护的数组中去掉对应的fd
            _fdArray[pos] = defaultFd;
            LogMessage(NORMAL, "client quit");
            return ;
        }
        else//读取失败
        {
            close(sock);//close之后，需要在所维护的数组中去掉对应的fd
            _fdArray[pos] = defaultFd;
            LogMessage(ERROR, "read fail%s", strerror(errno));
            return ;
        }
        //2、处理request
        std::string response = _func(buffer);
        //3、返回response
        //写事件，有问题的，需要重新维护一个写就绪的文件描述符数组
        write(sock, response.c_str(), response.size());
    }
    void HandlerEvent(fd_set& rfds)
    {
        for(int i = 0; i < fdNum; ++i)
        {
            //过滤掉空的fd
            if(_fdArray[i] == defaultFd) continue;
            //但是正常的fd不一定就绪
            //处理_listenSocket
            if(FD_ISSET(_fdArray[i], &rfds) && _fdArray[i] == _listenSocket)//如果_listenSocket在rfds集合里
            {
                ListenEvent(_listenSocket);
            }
            else if(FD_ISSET(_fdArray[i], &rfds))//判定该fd有没有就绪
            {
                Read(_fdArray[i], i);
            }
            else
            {}
        }
    }
    ~SelectServer()
    {
        if(_listenSocket >= 0)
        {
            close(_listenSocket); 
        }
        if(_fdArray)
        {
            for(int i = 0; i < fdNum; ++i)
            {
                if(_fdArray[i] != defaultFd)
                {
                    close(_fdArray[i]);
                }
            }
            delete[] _fdArray;
        }
    }
private:
    uint16_t _port;
    int _listenSocket;
    int* _fdArray;//由程序员维护的合法fd数组
    func_t _func;   
};

4、select的特点

1、select所能等待的文件描述符是由上限的，除非改内核。可监控的文件描述符个数取决与sizeof(fd_set)的值。本服务器上sizeof(fd_set)＝128，每bit表示一个文件描述符，则我服务器上支持的最大文件描述符是128*8=1024。

2、将fd加入select监控集的同时，还要再使用一个数组array保存放到select监控集中的fd。调用select前，需要重新设置所有的fd，事件发生之后，还要从数组中遍历找到就绪的fd【O(N)级别的遍历】

4、select采用位图结构，线性遍历时，会从用户态<==>内核态来回切换并发生位图的拷贝。

5、select第一个参数是最大fd+1,是为了在陷入内核时确定遍历文件描述符表的范围

6、云服务器的文件描述符一般是65535个，（ulimit -a查询，其中的open file栏就是可打开的最大文件描述符数量）但是一个select最多管理1024个文件描述符（虽然可以采用多进程的方式解决）

7、timeout超时时间需要在循环中进行更新，否则超时一次后，后续的timeout都是0了。

四、多路转接——poll

poll的作用和select一模一样，但是：

1、针对select的问题，poll的解决方案

1、poll解决select的fd有上限的问题；

2、poll解决select每次调用都要重新设置关心的fd

2、poll的原型

#include <poll.h>
int poll(struct pollfd *fds, nfds_t nfds, int timeout);//fds想象成new出来的数组

部分参数类型的意义：

由于此处fd所对应的events的输入输出分离了，所以不用像select那样每次重新设定参数
struct pollfd
{
   int   fd;//用户告诉内核要关心哪个fd
   short events;//用户告诉内核要关心的fd上的哪种事件（读、写、异常）
   short revents;//内核返回给用户，告诉用户这个fd对应的读、写、异常事件是否就绪
};
对于events和revents中的事件，如下图所示：

用途：poll 函数是一个在 POSIX 系统中用于多路复用的系统调用。它用于监视一组文件描述符（包括套接字、管道、设备等），以确定哪些文件描述符处于可读、可写或出现错误的状态

参数：

fds：指向一个 pollfd 结构体数组的指针，用于指定需要监视的文件描述符及相关事件。
nfds：fds 指向的数组的长度
timeout：表示超时时间（以毫秒为单位）。

如果设置的值 >0 ，则表示在指定的超时时间内等待就绪事件，等不到则超时返回。

如果设置的值 =0 ，则表示立即超时返回，不等待任何事件；(非阻塞)

如果设置的值 -1，则表示无限期阻塞，直到有就绪事件发生；

返回值：(同select)

如果成功，poll 函数返回就绪文件描述符的数量。
如果超时时间结束而没有任何文件描述符就绪，返回值为 0。
如果发生错误，返回值为 -1，并设置相应的错误码（可以使用 errno 获取）。

3、poll服务器部分代码（读事件）

#pragma once
#include <functional>
#include <poll.h>
#include "Sock.hpp"
//读到什么返回时什么
namespace poll_ns
{
    static const uint16_t defaultPort = 8080;
    static const int fdNum = 2048;
    static const int defaultFd = -1;
    static const int bufferNum = 1024;
    static const int timeOut = 1000;//1000ms
    using func_t = std::function<std::string(const std::string)>;//传入读到的数据，返回处理后的数据
    class PollServer
    {
    public:
        PollServer(func_t func, uint16_t port = defaultPort)
            :_port(port)
            ,_listenSocket(-1)
            ,_readFds(nullptr)
            ,_func(func)
        {}
        void InitServer()
        {
            _listenSocket = Sock::Socket();
            Sock::Bind(_listenSocket, _port);
            Sock::Listen(_listenSocket);
            _readFds = new struct pollfd[fdNum];
            for (int i = 0; i < fdNum; ++i)//清空fd数组
            {
                ResetItem(i);
            }
            //将listen套接字添加到数组里
            _readFds[0].fd = _listenSocket;
            _readFds[0].events = POLLIN;//关心读事件
            LogMessage(NORMAL,"create sock success %d", _listenSocket);
        }
        void Start()
        {
            while(1) 
            {
                int n = poll(_readFds, fdNum, timeOut);//阻塞，直到某个事件就绪
                switch(n)
                {
                case 0://超时返回
                    LogMessage(NORMAL, "timeOut...");
                    break;
                case -1://select失败
                    LogMessage(WARNING, "select error, code: %d, err string:%s", errno, strerror(errno));
                    break;
                default://事件就绪
                    //LogMessage(NORMAL, "have event ready");//如果没有把底层的连接取走，那么_listenSocket将一直就绪
                    HandlerReadEvent();
                    //HandlerWriteEvent()//写事件回调
                    //HandlerErrorEvent()//读事件回调
                    break;
                }
            }
        }
        void ListenEvent(int listenSocket)//监听到来事件
        {
            std::string clientIp;
            uint16_t clientPort = -1;
            int sock = Sock::Accept(listenSocket, &clientIp, &clientPort);//accept ==等 + 获取  
            if(sock < 0) 
            {
                return;
            }
            //在数组中寻找新的fd坑位
            int i = 0;
            for(; i < fdNum; ++i)
            {
                if(_readFds[i].fd != defaultFd) continue;
                else break;
            }
            if(i == fdNum)//说明文件描述符被写满了
            {
                LogMessage(WARNING, "Server is full, please wait");
                close(sock);//满载关闭消息
            }
            else
            {
                _readFds[i].fd = sock;
                _readFds[i].events = POLLIN;
                _readFds[i].revents = 0; 
            }
        }
        void Read(int pos)//pos是文件描述符在数组中的位置,代表哪个文件描述符就绪了
        {
            //1、读取request（这样读取存在问题）
            char buffer[bufferNum];
            ssize_t s = recv(_readFds[pos].fd, buffer, sizeof(buffer)-1, 0);
            if(s > 0)
            {
                buffer[s] = 0;
                LogMessage(NORMAL, "client# %s", buffer);
            }
            else if(0 == s)//连接关闭
            {
                close(_readFds[pos].fd);//close之后，需要在所维护的数组中去掉对应的fd
                ResetItem(pos);
                LogMessage(NORMAL, "client quit");
                return ;
            }
            else//读取失败
            {
                close(_readFds[pos].fd);//close之后，需要在所维护的数组中去掉对应的fd
                ResetItem(pos);
                LogMessage(ERROR, "read fail%s", strerror(errno));
                return ;
            }
            //2、处理request
            std::string response = _func(buffer);
            //3、返回response
            //写事件，有问题的，需要重新维护一个写就绪的文件描述符数组
            write(_readFds[pos].fd, response.c_str(), response.size());
        }
        void ResetItem(int resetIndex)
        {
            _readFds[resetIndex].fd =defaultFd;
            _readFds[resetIndex].events = 0;
            _readFds[resetIndex].revents = 0;
        }
        void HandlerReadEvent()
        {
            for(int i = 0; i < fdNum; ++i)
            {
                //过滤掉空的fd
                if(_readFds[i].fd == defaultFd) continue;
                //如果事件的读未被关心，跳过本轮循环
                if(!(_readFds[i].events & POLLIN)) continue;
                //但是正常的fd不一定就绪
                if(_readFds[i].fd == _listenSocket && (_readFds[i].revents & POLLIN))//如果_listenSocket在rfds集合里
                {
                    ListenEvent(_listenSocket);//监听事件
                }
                else if(_readFds[i].revents & POLLIN)//其他读事件就绪
                {
                    Read(i);//读取事件
                } 
                else
                {}
            }
        }
        ~PollServer()
        {
            if(_readFds)
            {
                for(int i = 0; i < fdNum; ++i)
                {
                    if(_readFds[i].fd != defaultFd)
                    {
                        close(_readFds[i].fd);
                    }
                }
                delete[] _readFds;
            }
        }
    private:
        uint16_t _port;
        int _listenSocket;
        struct pollfd* _readFds;//读文件事件poll的结构体
        func_t _func;
    };
}

4、poll的优缺点

4.1优点

不同与select使用三个位图来表示三个fdset的方式，poll使用一个pollfd的指针实现.

1、pollfd结构包含了要监视的event和发生的event，不再使用select“参数-值”传递的方式. 接口使用比select更方便。

2、poll并没有最大数量限制 (但是数量过大后性能也是会下降，原因是用户态和内核态的频繁切换和文件描述符的遍历)

4.2缺点

poll中监听的文件描述符数目增多时

1、和select函数一样，poll返回后，需要轮询pollfd来获取就绪的描述符。

2、每次调用poll都需要把大量的pollfd结构从用户态拷贝到内核中。

3、同时连接的大量客户端在一时刻可能只有很少的处于就绪状态, 因此随着监视的描述符数量的增长, 其效率也会线性下降。

五、多路转接——epoll

1、epoll的三个系统调用

1.1epoll_create（创建epoll模型）

#include <sys/epoll.h>
int epoll_create(int size);

用途：epoll_create用于创建一个 epoll模型 的系统调用

参数：

size：从Linux2.6.8开始，size参数被忽略，但设置时必须大于0.

返回值：

如果成功，epoll_create 返回一个非负整数的 epoll模型的文件描述符。
如果失败，返回值为 -1，并设置相应的错误码（可以使用 errno 获取）。

1.2epoll_ctl（对红黑树进行增删改）

#include <sys/epoll.h>
int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);

部分参数类型的意义：

typedef union epoll_data {
   void        *ptr;
   int          fd;
   uint32_t     u32;
   uint64_t     u64;
} epoll_data_t;

struct epoll_event {
   uint32_t     events;//用户让内核关心这个fd的哪种事件（事件如下）
   epoll_data_t data;//用户可自定义的变量
};
events可以是以下几个宏的集合：
EPOLLIN : 表示对应的文件描述符可以读 (包括对端SOCKET正常关闭); 
EPOLLOUT : 表示对应的文件描述符可以写;
EPOLLPRI : 表示对应的文件描述符有紧急的数据可读 (这里应该表示有带外数据到来);
EPOLLERR : 表示对应的文件描述符发生错误;
EPOLLHUP : 表示对应的文件描述符被挂断;
EPOLLET : 将EPOLL设为边缘触发(Edge Triggered)模式, 这是相对于水平触发(Level Triggered)来说的.
EPOLLONESHOT：只监听一次事件, 当监听完这次事件之后, 如果还需要继续监听这个socket的话, 需要
再次把这个socket加入到EPOLL队列里

用途：它用于向 epoll模型 注册、修改或删除要监听的文件描述符及其对应的事件。对于epoll模型的红黑树，使用fd充当红黑树的key值。所以在红黑树中进行增加需要提供fd和事件类型；进行删除提供fd即可。

参数：

epfd：表示 epoll模型的文件描述符，即通过 epoll_create 创建的返回值。
op：表示要进行的增加、修改或删除操作类型。可以是以下值之一：

EPOLL_CTL_ADD：向 epoll模型添加一个文件描述符并指定要监听的事件。
EPOLL_CTL_MOD：修改已注册的文件描述符的监听事件。
EPOLL_CTL_DEL：从 epoll模型中删除一个已注册的文件描述符。

fd：表示要关心文件描述符。(这个描述符epoll会帮忙关心它的状态)
event：指向 epoll_event 结构体的指针，用于指定要监听的事件类型。

返回值：

如果成功，epoll_ctl 返回 0。
如果出现错误，返回值为 -1，并设置相应的错误码（可以使用 errno 获取）。

1.3epoll_wait（从就绪队列取事件）

#include <sys/epoll.h>
int epoll_wait(int epfd, struct epoll_event *events,int maxevents, int timeout);

用途：用于等待 epoll模型上就绪事件的系统调用。它会阻塞当前线程，直到有文件描述符上的就绪事件发生或达到指定的超时时间

参数：

epfd：表示 epoll模型的文件描述符，即通过 epoll_create 创建的返回值。
events：输出型参数，指向 epoll_event 结构体数组的指针，用于接收就绪事件的信息。
maxevents：表示创建的 events 数组的大小，即最多可以存储多少个就绪事件。
timeout：表示等待的超时时间（以毫秒为单位）。如果设置为 -1，则表示无限期阻塞，直到有就绪事件发生；如果设置为 0，则表示立即返回，不等待任何事件；如果设置为大于 0 的值，则表示在指定的超时时间内等待就绪事件。

返回值：

如果成功，epoll_wait 返回就绪的事件数量。
如果超时时间结束而没有任何事件就绪，返回值为 0。
如果出现错误，返回值为 -1，并设置相应的错误码（可以使用 errno 获取）。

2、图解epoll事件就绪的过程

为什么epoll在确认哪些事件就绪的时候不用遍历，也不用频繁的切换状态呢？当读、写、异常事件就绪时，对端会传递电信号，CPU的引脚捕捉到电信号的变化时，触发中断，中断向量表中保存了各种驱动的函数指针，每一种事件中断对应一种事件的处理方法，这也是epoll为什么不用遍历就绪fd数组也能知道某种事件就绪的原因。

epoll模型中存在一颗红黑树和一个就绪队列，但他们的节点其实是共用的，图中已区分画出：

注：红黑树中的key中填充的是fd。所以，删除红黑树中的一个节点只需要在epoll_ctl中传入fd即可，关心的事件传入nullptr。

事件就绪的过程，就是节点的链接关系从红黑树转移至就绪队列的过程。

epoll_wait的细节：

1、select和poll所需要的辅助数组相当于epoll模型中的红黑树。这颗红黑树os会维护。

2、 epoll_wait将所有就绪的事件，按照顺序放入用户传入的数组中。例如5个fd就绪，events 数组下标0,1,2,3,4即为就绪事件的信息。使用时从0开始遍历即可，不用像select和poll一样遍历整个数组。epoll_wait对于检测事件是否就绪，时间复杂度为O(1)。

3、如果就绪队列有很多就绪的fd，一次拿不完，那么留到下次出队。

3、epoll底层的通知模式

3.1LT模式和ET模式

通过epoll_event中的event成员变量，将其设置为EPOLLET，即可是epoll的工作模式由水平触发转变为边缘触发。

LT模式（Level Triggered）：水平触发。以读为例，LT是epoll的默认行为，底层数据没读或者没读完，调用epoll_wait就会一直通知用户读走数据。

ET模式（Edge Triggered）：边缘触发。以读为例，只要上层从就绪队列读过数据了，不管这次有没有读完，epoll直接将该fd出队删除。直到下一次该fd的读事件到来，上层才有机会继续读取剩余的数据。使用 ET 能够减少 epoll 触发的次数。但是代价就是强逼着程序员一次响应就绪过程中就把所有的数据都处理完。那么就需要采用循环读取的方法，直到读取不到数据才算结束，所以在ET模式下，所有的fd必须都是非阻塞的。

3.2为什么ET模式比LT模式高效

1、通知机制

2、ET模式能让上层更快的把数据取走（程序员被倒逼了），那么TCP就可以给发送方提供一个更大的窗口大小，对端接收到这个窗口大小后就会更新出更大的滑动窗口，提高底层的数据发送效率。（TCP协议中有一个PSH标志位，这个标志位可以催促上层尽快取走数据，那么上层收到这个TCP报文的时候就知道读事件就绪了）

4、epoll服务器代码（读事件）

#pragma once
#include <iostream>
#include <functional>
#include <string>
#include <sys/epoll.h>
#include <memory>
#include <cstring>
#include "err.hpp"
#include "log.hpp"
#include "sock.hpp"
namespace epoll_ns
{
    static const uint16_t defaultPort = 8080;
    static const int size = 128;//size不小于0就行 
    static const int defaultValue = -1;
    static const int defaultNum = 64;//申请就绪事件的空间
    static const int timeOut = 1000;//超时时间，单位毫秒
    static const int readBuffer = 1024*8;
    using func_t = std::function<std::string(const std::string&)>;
    class EpollServer
    {
    public:
        EpollServer(func_t func, uint16_t port = defaultPort)
            :_port(port)
            ,_listenSocket(defaultValue)
            ,_epFd(defaultValue)
            ,_rEvents(nullptr)
            ,_num(defaultNum)
            ,_func(func)
        {}
        ~EpollServer()
        {
            if(_listenSocket >= 0) close(_listenSocket);
            if(_epFd >= 0)         close(_epFd);
            if(_rEvents)           delete[] _rEvents;
        }
    public:
        void InitServer();
        void Start();
        void HandlerEvent(int readyNum);
    private:
        uint16_t _port;
        int _listenSocket;
        int _epFd;//epoll模型
        struct epoll_event* _rEvents;//已经就绪的事件数组指针
        int _num;//每一次从就绪队列中拿走的事件数量
        func_t _func;//业务处理函数
    };
    
    void EpollServer::InitServer()
    {
        //1、创建socket
        _listenSocket = Sock::Socket();
        Sock::Bind(_listenSocket, _port);
        Sock::Listen(_listenSocket);
        //2、创建epoll模型
        _epFd = epoll_create(size);
        if(_epFd < 0)//创建epoll模型失败
        {
            LogMessage(FATAL, "epoll create error: %s", strerror(errno));
            exit(EPOLL_CREATE_ERR);
        }
        //3、将_listenSocket添加到epoll模型中
        struct epoll_event ev;
        ev.events = EPOLLIN;//关心读事件     ev.events = EPOLLIN | EPOLLET;//边缘触发，fd需非阻塞
        ev.data.fd = _listenSocket;//当事件就绪，被重新捞取上来，就知道是哪一个fd就绪了
        epoll_ctl(_epFd, EPOLL_CTL_ADD, _listenSocket, &ev);
        //4、申请就绪事件的空间
        _rEvents = new struct epoll_event[_num];
        LogMessage(NORMAL, "Init server success");
    }
    void EpollServer::Start()
    {
        while(1)
        {
            int readyNum = epoll_wait(_epFd, _rEvents, _num, timeOut);
            switch(readyNum)
            {
            case 0://超时
                LogMessage(NORMAL, "Timeout");
                break;
            case -1:
                LogMessage(WARNING, "Epoll_wait failed, code: %d, errstring: %s", errno, strerror(errno));
                break;
            default:
                LogMessage(NORMAL, "Have event ready");
                HandlerEvent(readyNum);//n是就绪事件的数量
                break;
            }
        }
    }
    void EpollServer::HandlerEvent(int readyNum)
    {
        for(int i = 0; i<readyNum; ++i)
        {
            uint32_t events = _rEvents[i].events;//哪一个事件就绪了
            int sock = _rEvents[i].data.fd;//拿到这个事件的fd
            //处理这个fd的事件
            if(sock == _listenSocket && (events & EPOLLIN))//_listenSocket的读事件就绪,获取新链接
            {
                std::string clientIp;
                uint16_t clientPort;
                int fd = Sock::Accept(_listenSocket, &clientIp, &clientPort);
                if(fd < 0)
                {
                    LogMessage(WARNING, "Accept error");
                    continue;
                }
                //将读取的fd和事件加入epoll模型中
                struct epoll_event ev;
                ev.events = EPOLLIN;
                ev.data.fd = fd;
                epoll_ctl(_epFd, EPOLL_CTL_ADD, fd, &ev);
            }
            else if(events & EPOLLIN)//普通的读事件就绪
            {
                //这里读取依旧有问题
                char buffer[readBuffer];//因为没有自定义协议保证数据是否一次被读完，临时变量。下一次读取时上一次写入的数据已被释放
                int n = recv(sock, buffer, sizeof(buffer)-1, 0);
                if(n > 0)
                {
                    buffer[n] = 0;
                    LogMessage(DEBUG, "Client# %s", buffer);
                    //处理数据
                    std::string response = _func(buffer);
                    send(sock, response.c_str(), response.size(), 0);
                }
                else if(n == 0)//对端关闭连接
                {
                    //在epoll中对该fd的该事件取消关心,先在epoll中移除，再关闭fd
                    epoll_ctl(_epFd, EPOLL_CTL_DEL, sock, nullptr);
                    close(sock);
                    LogMessage(DEBUG, "Client quit");
                }
                else//读取出错
                {
                    epoll_ctl(_epFd, EPOLL_CTL_DEL, sock, nullptr);
                    close(sock);
                    LogMessage(ERROR, "recv error, code: %d, errstring: %s", errno, strerror(errno));
                }
            }
            else//写
            {}
        }
    }
}