计算机网络（十二） —

#1024程序员节 | 征文#

一，预备

1.1 重新理解IO

1.2 五种IO模型

1.3 非阻塞IO

二，select

2.1 关于select

2.2 select接口参数解释

2.3 timeval结构体和fd_set类型

2.4 socket就绪条件

2.5 select基本工作流程

2.6 简单select的服务器代码

2.7 select优缺点

三，poll

3.1 关于poll

3.2 poll接口介绍

3.3 poll服务器

3.4 poll优缺点

四，epoll

4.1 关于epoll

4.2 epoll的系列系统调用

4.2.1 epoll_create

4.2.2 epoll_wait

4.2.3 epoll_ctl

4.3 epoll模型

4.4 epoll服务器编写

4.5 epoll优点

五，epoll两种工作模式

5.1 LT，ET工作模式

5.2 ET工作模式的epoll_server服务器（较难，了解即可）

一，预备

1.1 重新理解IO

我们以read && write这两个系统接口为例：

调用read时，底层缓冲区没数据，read就阻塞着；当write写的时候，写入缓冲区满了或者滑动窗口为0了，那么write也阻塞等着
应用层对套接字读和写，不是直接从网络里读写，而是从OS提供的缓冲区中读写 --> read和write本质就是拷贝函数
当进行IO的时候，大部分时间尤其是在网络的情况下是在等待的，所以 IO = 拷贝 + 等待，这个等待就是在等缓冲区的条件是否满足，所以要进行拷贝，必须要先判断条件成立（条件：读写事件，就是等待读写事件就绪）
网络通信本质就是在IO，那么，什么叫做高效的IO呢？（高效：单位时间内能拷贝的数据量越大，等的比重就越小，IO的效率就越高；而现在几乎所有提高IO效率的策略，本质就是“使单位时间内IO等的比重越小”）

1.2 五种IO模型

我们以钓鱼佬去钓鱼为例，宏观的钓鱼分为两个阶段：1，“等”鱼上钩 2，鱼上钩了把鱼拉上来（一个叫做“等”，一个叫做“钓”，鱼竿相当于文件描述符或者套接字）

场景①：张三是新手，抛竿后，张三就两只手握着鱼竿，哪也不去，啥也不做，就死死盯着水面，突然被咬钩了，就把鱼拉了上来

在这个过程中，张三钓鱼过程中，他什么也不做，什么也打扰不了他，我们叫做“阻塞式IO”，我们学到的大部分接口，都是阻塞式的

场景②：李四来了，是个有两三年钓鱼的人，李四开始钓鱼，但是不和张三一样死死盯着鱼竿了，开始刷视频，每隔几分钟去查看有没有咬钩，没有咬钩就继续去刷视频，咬钩了就把鱼拉上来

李四并不会因为鱼没咬钩卡在那里，李四会定期检测鱼竿又没有鱼，条件不满足就直接返回了做其他事情去了，满足就开始钓鱼，这个过程中，李四即把鱼钓了，也做了其它事情，我们叫做“非阻塞式IO”

场景③：王五来了，五年钓龄，他看着前面两人，一个一动不动，一个一直在动，他觉得这俩方式很喽。所以王五在鱼竿上放了一个铃铛，然后就躺下不管了，当有鱼来的时候，听到铃铛响了，就把鱼上来了

前面两个是自己检测有没有鱼上钩，而王五的钓鱼方法，是让鱼来通知我，这种方式叫做：“信号驱动式IO”，底层有数据时，主动通知上层让上层来读取数据

场景④：赵六来了，但是赵六是开着卡车过来的，拉了一车的鱼竿，然后每根鱼竿都插在岸边，然后赵六就去周期性地区轮询检测有没有鱼上钩

4，赵六这样周期性的去检测有没有鱼上钩的这种方式，我们叫做“多路复用，多路转接”

问题：前四个人谁的钓鱼效率最高？

解答：赵六，因为他鱼竿多，那为什么鱼竿多效率就高呢？
假设鱼咬一个钩的概率是1/100，那么鱼咬前面三个人的钩的概率是一样的，都是1/100，但是要赵六的概率就翻了十倍，就是1/10了；所以多个鱼竿可以让我们IO时等待的时间是重叠的，并行的

场景⑤：田七来了，是方园五百里的首富，是公司老板，很忙，田七看着岸边是四个人钓鱼，一个一动不动，一个一直在动，一个躺着玩手机，一个左右在动；田七不喜欢钓鱼，但是喜欢吃鱼，突然田七要回去开会，就喊同行的小王拿上桶，电话等装备，让小王去帮田七钓鱼，当桶满了的时候，就打电话给田七，到时候田七就开车来接小王

小王怎么钓不关心，但是在这个过程中，田七这个人他没有参与钓鱼的过程，真正钓鱼的过程是小王，所以田七不算真正意义上的钓鱼者，只是钓鱼的发起者，因为田七不是要钓鱼，他想要的只是鱼，田七的这种钓鱼方式，我们叫做“异步IO”，小王指的就是操作系统，田七就是应用层程序

我们把前四种IO我们叫做“同步IO”，第五种叫做“异步IO”

问题：阻塞式IO与非阻塞式IO有什么区别？

解答：在IO效率方面其实没有任何区别，因为IO = 等待 + 拷贝，从这个角度看，阻塞式和非阻塞式IO效率没有区别，区别就在于它们“等”的方式不一样

问题：同于IO与异步IO有什么区别？

解答：

上面前四种IO方式，其实都有在等（假设鱼就是数据，所以其实都在等，王五也在等），并且，它们四个都参与了钓鱼的两个阶段，所以同步IO的本质就是当前特定的进程或者人“有没有参与IO”（参与：参与了“等”或“拷贝”或者都参与了），只要参与了IO，那就都是同步IO
异步IO，就是不参与IO，只是发起IO，最后拿结果就行

问题：那么同步IO和我们前面讲的线程同步有什么关系？

解答：就是老婆和老婆饼的关系，只是名字像，没有其它关系

1.3 非阻塞IO

我们以前使用recv的时候，第四个参数一般是设置成0的，表示阻塞等待；如果我们想非阻塞等待，就把第四个参数设置成MSG_DONTWAIT:

但是这个用起来不方便，所以下面是一种更通用的方法：

先来看下这个接口：

这个函数作用就是设置一个文件描述符的属性，就是设置文件描述符在底层的struct_file结构体的标志位，从而告诉内阁这个指定的文件描述符，以其它的方式来操作：

复制一个现有的描述符（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)

下面我们简单使用下fcntl，将标准输入设置成非阻塞，这样之后通告read读取标准输入时就是非阻塞了：

//非阻塞轮询
#include <iostream>
#include <unistd.h>
#include <fcntl.h>
#include <cerrno>
#include<string.h>

using namespace std;

void SetNonBlock(int fd)
{
    int fl = fcntl(fd, F_GETFL); //获取struct_file的标记位
    if(fl < 0) //获取失败
    {
        perror("fcntl");
        return;
    }
    fcntl(fd, F_SETFL, fl | O_NONBLOCK); //设置非阻塞标志位
}

int main()
{
    char buffer[1024];
    SetNonBlock(0); //将标准输入流设置为非阻塞形式
    sleep(1);
    while(true)
    {    
        ssize_t n = read(0, buffer, sizeof(buffer)-1); //阻塞式等待标准输入
        if(n > 0)
        {
            buffer[n - 1] = 0; //-1是去掉换行
            cout << "echo: " << buffer << endl;        
        }
        else if(n == 0) //一般是读到结尾，网络中读到0是对方把连接关了，来Linux中也可以Ctrl + D结束读取
        {
            cout << "read done" << endl;
            break;
        }
        else //读取出错
        {
            //1，fd设置成非阻塞，如果底层fd没有数据，recv/read/write/send，返回值会以错误的形式返回
            //2，出错也有两种情况：a，真的出错了    b，底层没有就绪
            //3，我怎么区分？通过errno错误码去区分
            if(errno == EWOULDBLOCK)
            {
                cout << "0 fd data not ready, try again!" << endl;
                //这里就可以做其它事情
            }
            else //真正意义上的读取出错
            {
                cerr << "read error, n:" << n << " errno code: " << errno << " error str: " << strerror(errno) << endl;
            }
            
            sleep(1);
        }
    }
    
    return 0;
}

二，select

2.1 关于select

前面我们说过，IO = 等 + 拷贝，我们现在要处理的就是等，所以select的作用就是负责“等”，它用的是我们前面五种IO模型中的第四个模型，也就是select可以一次同时等待多个fd

2.2 select接口参数解释

select在man手册中的说明如下：

下面解释一下参数：

nfds：表示未来要等待的多个文件描述符最大的哪个+1，架设有12345五个描述符需要关心，那么这个参数就设置为6
readfds，writefds，exceptfds：中间三个参数都是作为输出型参数，用户通过这三个参数告诉内核需要监视的文件描述符的读/写/异常事件，而用户也通过这三个参数告诉用户对应的读/写/异常事件是否就绪（有点绕，但是没关系，后面会展开讲）
timeout：也是输入输出型参数，调用时表示用户设置的select的等待事件，当有事件就绪后返回时，内核也通过该参数告诉用户本次等待的剩余事件

参数timeout的取值：

nullptr / NULL：表示让select进行阻塞等待，就比如把模型四变成了模型一，直到被监视的某个文件描述符上的某个事件就绪才返回
0：表示让select非阻塞等待，无论被监视的文件是否有事件就绪，只要检测后就直接返回
特定的时间：让select调用后在指定的事件内进行阻塞等待，如果超过设置的事件，select将进行超时返回

返回值说明：

如果函数调用成功，则返回值代表有事件就绪的文件描述符个数
如果是超时返回，则返回0，表示没有事件就绪
调用失败返回-1，错误码被设置

select调用失败后，错误码设置：

EBADF：文件描述符已失效或文件已关闭
EINTR：被信号中断
EINVAL：参数nfds设置为负数
ENOMEM：内核内存不足

2.3 timeval结构体和fd_set类型

我们先来看下timeval结构体类型：

在select接口介绍中，timeval作为第四个参数的结构体指针被传参，这个timeval也是一个结构体，我们可以在man 2 gettimeofday获取时间的接口介绍看到：

这个就是系统提供的时间结构体，time_t就是无符号整数，以秒为单位，代表时间戳；第二个代表微秒
select最后一个参数表示给select设置等待方式，假设我们struct timeval timeout = {5, 0}; 表示让select每隔5秒去timeout（超时返回）一次，就是在5秒内没有任何一个文件描述符就绪，就返回，重新开始；如果有文件描述符就绪了，就立即返回，如果我们设为 {0 ,0}，那么就是立马返回，非阻塞的一种；也可以设为NULL，也就是阻塞等待
如果这个函数设置了，就是一个输入输出型参数，比如我设置5秒timeout一次，如果过了2秒，就有文件描述符就绪了，就会返回，timeout就会变成[3, 0]返回，表示还剩 3 秒就超时了

下面我们再来看下fd_set类型：

看到set，就要想到“位图”，所以fd_set是操作系统给我们提供的一种数据类型，如下图：

我们目前关心的fd事件：读，写，异常
如果我想关心读事件是否就绪，就把文件描述符设置进readfds，如果我想关心写事件，就设置进writefds，异常事件同理
如果想即关心读又关心写，就设置两个；如果我想先关心读再关心写，我们可以把描述符先设置进readfds，等读完了，再添加进writefds
在后面我们只学习readfds读事件，只要懂了这一个，就能举一反三，其它的就都会了

问题：为什么是位图呢？

解答：已readfds为例

在输入时，表示用户告诉内核，我给你的一个或者多个fd，你要帮我关心fd上面的读事件，如果读事件就绪了，你要告诉我！
在输出时（返回时），内核告诉用户，你让我关心的多个fd中有哪些已经就绪了，用户赶紧来读
假设这个位图是8位的0000 0000，我想添加0 1 2 3这四个fd，就需要把位图改为0000 1111，所以比特位的位置表示文件描述符编号，比特位的内容，0和1就是表示内核是否需要关心
假设当2号fd就绪了，所以操作系统在返回的时候，把没有就绪的比特位清零，返回的就是0000 0010，这样用户就知道2号fd就绪了，可以去读了
所以返回时，比特位的位置还是文件描述符编号，但是比特位的内容，0和1就是表示哪些用户关心的fd上面的读事件已经就绪了

所以fd_set是一张位图，本质是为了让用户与内核相互传递fd是否就绪的信息的，所以这也注定了，使用select的时候，一定会有大量的位图操作，所以系统也给我们提供了位图操作的接口：

2.4 socket就绪条件

读就绪前置条件：

socket内核中, 接收缓冲区中的字节数，大于等于低水位标记SO_RCVLOWAT，此时可以无阻塞的读该文件描述符，并且返回值大于0
socket TCP通信中，对端关闭连接，此时对该socket读，则返回0
监听的socket上有新的连接请求
socket上有未处理的错误

写就绪前置条件：

socket内核中，发送缓冲区中的可用字节数(发送缓冲区的空闲位置大小)，大于等于低水位标记 SO_SNDLOWAT，此时可以无阻塞的写，并且返回值大于0
socket的写操作被关闭(close或者shutdown)，对一个写操作被关闭的socket进行写操作，会触发SIGPIPE信号
socket使用非阻塞connect连接成功或失败之后
socket上有未读取的错误

异常就绪前置条件：

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

2.5 select基本工作流程

我们实现一个简单的基于select服务器，服务器要做的就是读取客户端发来的数据进行处理，这个select服务器的工作流程如下：

初始化服务器，创建套接字，并完成绑定端口，开启监听
定义一个fd_array数组用于保存监听套接字和已经与客户端建立连接的套接字（没错，我们把两种套接字放一个数组里面）
然后服务器循环调用select函数，轮询检测select关心的fd中的读事件是否就绪，如果就绪则执行对应的操作
每次调用select函数之前，都需要定义一个fd_set readfds读文件描述符位图，然后每次循环都将fd_array中的fd依次设置进readfds，然后select就可以去检测根据readfds中fd的读事件是否就绪了
当select检测到数据就绪后，就可以再次设置readfds，然后返回给我们，这样我们就知道了哪些文件描述符就绪了，而这步操作后就绪的文件描述符有两种：
如果是监听套接字就绪了，说明有新连接到来，于是调用accept获取新连接，然后再添加到fd_array中
如果是读事件就绪了，就调用read函数从套接字从套接字中读取数据并打印输出
当客户端把连接关掉后，服务器这边也应该close掉对应的fd，并且将该fd从fd_array中去掉

注意：

每次循环都要设置readfds，是因为select返回时readfds里面的参数已经被修改过了，所以我们才会用fd_array保存套接字，并在每次循环时重新设置readfds，writefds和exceptfds同理
我们目前是只让服务器关心读事件，如果要关心写事件和异常事件，需要再定义writefds和exceptfds，并且再定义两个fd_array数组来做和readfds相同的操作
由于select还需要传入被监视的多个文件描述符的最大值+1，所以每次循环调用select前，还需要获取fd_array中最大的文件描述符值

2.6 简单select的服务器代码

这个服务器用到的文件如下：

源码：计算机网络/高级IO/Select · 小堃学编程/Linux学习 - 码云 - 开源中国

Socket.hpp和Log.hpp我们在之前就已经写过了：计算机网络（五） —— 自定义协议简单网络程序-CSDN博客

Socket.hpp：

#pragma once
#include <iostream>
#include <string>
#include <unistd.h>
#include <cstring>
#include <sys/types.h>
#include <sys/stat.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include "Log.hpp"

#include <cstring>

enum
{
    SocketErr = 2,
    BindErr,
    ListenErr,
};

const int backlog = 10;

class Sock
{
public:
    Sock()
    {
    }
    ~Sock()
    {
    }

public:
    void Socket() // 创建套接字
    {
        _sockfd = socket(AF_INET, SOCK_STREAM, 0);
        if (_sockfd < 0)
        {
            log(Fatal, "socket error, %s: %d", strerror(errno), errno);
            exit(SocketErr);
        }
        int opt = 1;
        setsockopt(_sockfd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
    }
    void Bind(uint16_t port) // 绑定套接字
    {
        struct sockaddr_in local;
        memset(&local, 0, sizeof(local));
        local.sin_family = AF_INET;
        local.sin_port = htons(port);
        local.sin_addr.s_addr = INADDR_ANY;

        if (bind(_sockfd, (struct sockaddr *)&local, sizeof(local)) < 0) // 如果小于0就绑定失败
        {
            log(Fatal, "bind error, %s: %d", strerror(errno), errno);
            exit(BindErr);
        }
    }

    void Listen() // 监听套接字
    {
        if (listen(_sockfd, backlog) < 0) // 如果小于0就代表监听失败
        {
            log(Fatal, "listen error, %s: %d", strerror(errno), errno);
            exit(ListenErr);
        }
    }

    int Accept(std::string *clientip, uint16_t *clientport) // 获取连接
    {
        struct sockaddr_in peer;
        socklen_t len = sizeof(peer);
        int newfd = accept(_sockfd, (struct sockaddr *)&peer, &len);
        if (newfd < 0) // 获取失败
        {
            log(Warning, "accept error, %s: %d", strerror(errno), errno);
            return -1;
        }
        char ipstr[64];
        inet_ntop(AF_INET, &peer.sin_addr, ipstr, sizeof(ipstr)); // 把网络字节序列转化为字符串保存在ipstr数组里供用户读取
        *clientip = ipstr;
        *clientport = ntohs(peer.sin_port);

        return newfd;
    }

    bool Connect(const std::string &ip, const uint16_t port)
    {
        struct sockaddr_in peer;
        memset(&peer, 0, sizeof(peer));
        peer.sin_family = AF_INET; // 粗心忘记加了
        peer.sin_port = htons(port);
        inet_pton(AF_INET, ip.c_str(), &(peer.sin_addr));

        int n = connect(_sockfd, (struct sockaddr *)&peer, sizeof(peer));
        if (n == -1)
        {
            std::cerr << "connect to " << ip << ":" << port << "error" << std::endl;
            return false;
        }
        return true;
    }

    void Close()
    {
        close(_sockfd);
    }

    int Fd()
    {
        return _sockfd;
    }

private:
    int _sockfd;
};

Log.hpp：

#pragma once

#include <iostream>
#include <time.h>
#include <stdarg.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <unistd.h>
#include <stdlib.h>

#define SIZE 1024

#define Info 0
#define Debug 1
#define Warning 2
#define Error 3
#define Fatal 4

#define Screen 1
#define Onefile 2
#define Classfile 3

#define LogFile "log.txt"

class Log
{
public:
    Log()
    {
        printMethod = Screen;
        path = "./log/";
    }
    void Enable(int method)
    {
        printMethod = method;
    }
    std::string levelToString(int level)
    {
        switch (level)
        {
        case Info:
            return "Info";
        case Debug:
            return "Debug";
        case Warning:
            return "Warning";
        case Error:
            return "Error";
        case Fatal:
            return "Fatal";
        default:
            return "None";
        }
    }
    void printLog(int level, const std::string &logtxt)
    {
        switch (printMethod)
        {
        case Screen:
            std::cout << logtxt << std::endl;
            break;
        case Onefile:
            printOneFile(LogFile, logtxt);
            break;
        case Classfile:
            printClassFile(level, logtxt);
            break;
        default:
            break;
        }
    }
    void printOneFile(const std::string &logname, const std::string &logtxt)
    {
        std::string _logname = path + logname;
        int fd = open(_logname.c_str(), O_WRONLY | O_CREAT | O_APPEND, 0666); // "log.txt"
        if (fd < 0)
            return;
        write(fd, logtxt.c_str(), logtxt.size());
        close(fd);
    }
    void printClassFile(int level, const std::string &logtxt)
    {
        std::string filename = LogFile;
        filename += ".";
        filename += levelToString(level); // "log.txt.Debug/Warning/Fatal"
        printOneFile(filename, logtxt);
    }

    ~Log()
    {
    }
    void operator()(int level, const char *format, ...)
    {
        time_t t = time(nullptr);
        struct tm *ctime = localtime(&t);
        char leftbuffer[SIZE];
        snprintf(leftbuffer, sizeof(leftbuffer), "[%s][%d-%d-%d %d:%d:%d]", levelToString(level).c_str(),
                 ctime->tm_year + 1900, ctime->tm_mon + 1, ctime->tm_mday,
                 ctime->tm_hour, ctime->tm_min, ctime->tm_sec);

        va_list s;
        va_start(s, format);
        char rightbuffer[SIZE];
        vsnprintf(rightbuffer, sizeof(rightbuffer), format, s);
        va_end(s);

        // 格式：默认部分+自定义部分
        char logtxt[SIZE * 2];
        snprintf(logtxt, sizeof(logtxt), "%s %s", leftbuffer, rightbuffer);

        // printf("%s", logtxt); // 暂时打印
        printLog(level, logtxt);
    }

private:
    int printMethod;
    std::string path;
};

Log log;

Makefile：

select_server:Main.cc
	g++ -o $@ $^ -std=c++11
.PNONY:clean
clean:
	rm -rf select_server

然后下面就是select的主逻辑，SelectServer.hpp，主要的逻辑和上面select的基本工作流程是一样的：

#pragma once
#include<iostream>
#include<sys/select.h>
#include<sys/time.h>
#include "Socket.hpp"
#include "Log.hpp"

static const uint16_t defaultport = 8081;
static const int fd_max_num = (sizeof(fd_set)); //表示一个位图能存多少个比特位，也就是一个fd_set能存多少个文件描述符
int defaultfd = -1;

class SelectServer
{
public:
    SelectServer(uint16_t port = defaultport)
        : _port(port)
    {
        for(int i = 0; i < fd_max_num; i++) //初始化辅助数组
        {
            fd_array[i] = defaultfd;
        }
    }
    bool Init()
    {
        _listensock.Socket();
        _listensock.Bind(_port);
        _listensock.Listen();
    }
    void Accepter()
    {
        //走到这里，就是监听套接字就绪了，底层有新连接上来了
        std::string clientip;
        uint16_t clientport = 0;
        int sock = _listensock.Accept(&clientip, &clientport); // 不会阻塞在这里，因为走到着一步就是因为底层已经有连接了，已经可以直接拿上来
        if (sock < 0) return;
        log(Info, "accept success, %s: %d, sock fd: %d", clientip.c_str(), clientport, sock);

        // 在获取到新连接后也不能直接读，因为只是把连接读上来了，但是对方可能没有发数据，所以当对方没立即发数据的时候，文件里就是空的，这时候读取会出问题哦
        int pos = 1;
        for(; pos < fd_max_num; pos++)
        {
            if (fd_array[pos] != defaultfd)
                continue; // 说明这个位置是已经被占用的位置
            else
                break;
        }
        // 走到这里有两种结果
        if (pos == fd_max_num) // 1，说明辅助数组已经被合法文件描述符占满了
        {
            log(Warning, "server is full, close %d now ", sock);
            close(sock); // 满了直接关了
        }
        else // 2，说明当前pos的位置可以被使用
        {
            fd_array[pos] = sock; // 把新获取的连接搞到数组里去
            PrintFd();
        }
    }
    void Recver(int fd, int pos)
    {
        //走到这里代表套接字的读事件就绪了，客户端给我发数据过来了
        char buffer[1024];
        ssize_t n = read(fd, buffer, sizeof(buffer) - 1); // 有bug，因为目前不能保证，对方通过网络发过来的报文可能不是完整的
        if (n > 0) //读成功
        {
            buffer[n] = 0;
            std::cout << "get a messge: " << buffer << std::endl;
        }
        else if (n == 0) //对方把连接关了，那么我服务器也要把套解析关了
        {
            log(Info, "client quit, me too, close fd is : %d", fd);
            close(fd);
            fd_array[pos] = defaultfd; // 这里本质是从select中移除
        }
        else //读出错
        {
            log(Warning, "recv error: fd is : %d", fd);
            close(fd);
            fd_array[pos] = defaultfd; // 这里本质是从select中移除，因为start第一个循环会根据这个数组重新设置fd_set
        }
    }
    void Dispatcher(fd_set &rfds)
    {
        for (int i = 0; i < fd_max_num; i++)
        {
            int fd = fd_array[i];
            if (fd == defaultfd) continue; //判断文件描述符是否就绪
            //所有合法的文件描述符放在辅助数组里，select一返回，所有就绪的文件描述符就会在rfds里
            //判断我们数组里的合法的文件描述符是否也在rfds里，如果在，代表该文件描述符已经就绪：1，读事件就绪    2，连接事件就绪
            if (FD_ISSET(fd, &rfds))
            {
                //读事件就绪后也有两种情况
                if (fd == _listensock.Fd()) //1，如果等于listen套接字，就是连接事件就绪，就获取新连接，就把这个新的文件描述符继续添加进数组里
                {
                    Accepter(); // 连接管理器
                }
                else //2，如果不是linsten，那么就是别的文件描述符就绪了，也就是读事件就绪了
                {
                    Recver(fd, i);
                }
            }
        }
    }
    bool Start()
    {
        int fd = _listensock.Fd();
        //struct timeval timeout = {1, 0};

        fd_array[0] = fd;
        while(true)
        {
            fd_set rfds; //读文件描述符位图
            FD_ZERO(&rfds); //是一个宏，负责把位图清空

            //每次调用select前都对rfds进行重新设定
            int maxfd = fd_array[0];
            for(int i = 0; i < fd_max_num; i++) 
            {
                if(fd_array[i] == defaultfd) continue;
                FD_SET(fd_array[i], &rfds); //把指定的文件描述符添加到rfds中，并且此时没有设置进内核里
                if(maxfd < fd_array[i]);
                {
                    maxfd = fd_array[i];
                    log(Info, "max fd update, max fd is: %d", maxfd);
                }
            }

            //开始监听后，不能"直接"accept，因为accept就是在检测并获取listensock上面的事件，是阻塞，那么就不能一次等待多个文件描述符了，那么select就没作用了
            //是什么事件呢？就是新连接到来的事件，就是操作系统底层把三次握手完成，把新连接放到了全连接队列里，然后通过select从底层把文件拿上来
            //新连接到来，就相当于读事件就绪
            struct timeval timeout = {0, 0}; //输入输出，可能要进行周期性的重复设置
            //设为0的话，就是非阻塞了，如果不设置timeout，就是永久性阻塞，直到有事件就绪
            int n = select(maxfd + 1, &rfds, nullptr, nullptr, /*&timeout*/ nullptr);
            //如果select告诉你就绪了，那么接下来的一次读取fd的时候，不会被阻塞，就是不会再“等”了，会直接读
            switch(n)
            {
                case 0:
                    std::cout << "timeout: " << timeout.tv_sec << "." <<timeout.tv_usec << std::endl;
                    break;
                case -1:
                    std::cerr << "select error" << std::endl;
                    break;
                default:
                    //有事件就绪了,但是如果上层不处理，select就会一直通知你
                    std::cout << "get a new link " << std::endl;
                    Dispatcher(rfds); //处理事件
                    break;
            }
        }
    }
    void PrintFd()
    {
        std::cout << "online fd list: ";
        for (int i = 0; i < fd_max_num; i++)
        {
            if (fd_array[i] == defaultfd)
                continue;
            std::cout << fd_array[i] << " ";
        }
        std::cout << std::endl;
    }
    ~SelectServer()
    {
        _listensock.Close();
    }

private:
    Sock _listensock;
    uint16_t _port;
    int fd_array[fd_max_num]{defaultfd}; //辅助数组，用来保存监听套接字和已经和客户端建立好连接的套接字
};

然后就是 Main.cc：

#include "SelectServer.hpp"
#include<memory>

void Usage(std::string proc)
{
    std::cout << "\n\rUsage: " << proc << "port[1024+]\n"
              << std::endl;
}

int main(int argc, char *argv[])
{
    if (argc != 2)
    {
        Usage(argv[0]);
        exit(1);
    }
    //单进程演示同时接收多个连接的消息
    uint16_t port = std::stoi(argv[1]);
    std::unique_ptr<SelectServer> svr(new SelectServer(port));
    svr->Init();
    svr->Start(); 
    return 0;
}

主要逻辑如下图：

2.7 select优缺点

select优点：

可以同时等待多个文件描述符，并且只负责等待，实际的拷贝动作由read，write等接口完成，并且最大的好处就是select之后再调用read这些接口不会再被阻塞
select同时等待多个文件描述符，可以将“等”的时间重叠，提高IO效率

select缺点：

到的fd是有上限的，这个和操作系统有关系，不同的操作系统对位图的实现不同
select的输入输出型参数比较多，所有可能会有比较频繁的用户层与内核层的数据拷贝
输入输出型参数较多，每次都要对关心的fd进行重置，就会有很多次的循环遍历，容易出bug或者其它问题
用户层，使用第三方数组管理用户的fd，用户层需要很多次遍历，内核中检测fd事件就绪也要遍历，效率比较低

所以select有好有坏，简单来说就是“还不够完美”，所以针对select的缺点，推出了poll

三，poll

3.1 关于poll

poll也是操作系统提供的一个多路转接接口，poll可以让我们的程序同时监视多个文件描述符上的事件是否就绪，和select的定位是一样的，poll是为了解决select的几个缺点

3.2 poll接口介绍

参数说明：

fds：表示一个poll函数监视的结构列表，是一个结构体，有三个元素，分别为文件描述符，监视的事件集合，就绪的事件集合
nfds：表示fds数组的长度，也就是第一个参数的长度
timeout：这个和select的那个是一样的，这里不再赘述
返回值和错误码也是和select一样的

pollfd结构体

下面是结构体的三个字段

fd：就是一个特定的文件描述符，如果设为负值则忽略events字段并revents返回0
events：表示需要监视该文件描述符上的哪些事件
revents：内核通过revents告诉用户该文件描述符哪些事件已经就绪

事件宏	描述	是否可作为输入	是否可作为输出
POLLIN	数据可读（包括普通和优先数据）	是	是
POLLRDNORM	普通数据可读	是	是
POLLRDBAND	优先级数据可读（Linux不支持）	是	是
POLLPRI	高优先级数据可读，比如Tcp紧急指针	是	是
POLLOUT	数据可写（包括普通和优先数据）	是	是
POLLWRNORM	普通数据可写	是	是
POLLWRBAND	优先级数据可写	是	是
POLLRDHUP	对方三次挥手，或者对方变比了写操作，由GNU引入	是	是
POLLERR	错误	否	是
POLLHUP	挂起，比如管道写端关闭后，读端描述符会收到POLLHUP事件	否	是
POLLNVAL	文件描述符没有打开	否	是

这个结构体的最大特点是，将输入与输出进行了分离，用户通过这个结构体告诉内核“你要帮我关系fd文件描述符上的event事件”，然后内核就设置一下revent，告诉用户“fd文件描述符上的event已经就绪了”
如果我要同时关心多个文件描述符，就只要构建结构体数组，把数组指针传过去就OK了，然后只需要遍历这个数组即可
select有三个类似参数，把事件类型做了区分，那poll呢？结构体是short类型，Linux很喜欢用比特位来进行传参，所以就把事件设置成位图

3.3 poll服务器

源码：计算机网络/高级IO/poll · 小堃学编程/Linux学习 - 码云 - 开源中国

poll的工作流程和select是基本类似的，所以只需要改下SelectServer.hpp就可以了，重命名为PollServer.hpp：可以发现poll的使用比select简单很多，因为接口的进步就是越来越简单

#pragma once
#include<iostream>
#include<poll.h>
#include<sys/time.h>
#include "Socket.hpp"
#include "Log.hpp"

static const uint16_t defaultport = 8081;
static const int fd_max_num = 64; //表示一个evevt_fds数组能存多少个pollfd结构体
int defaultfd = -1;
int long non_event = 0;

class PollServer
{
public:
    PollServer(uint16_t port = defaultport)
        : _port(port)
    {
        for(int i = 0; i < fd_max_num; i++) //初始化辅助数组
        {
            event_fds[i].fd = defaultfd;
            event_fds[i].events = non_event;
            event_fds[i].revents = non_event;
        }
    }
    bool Init()
    {
        _listensock.Socket();
        _listensock.Bind(_port);
        _listensock.Listen();
    }
    void Accepter()
    {
        //走到这里，就是监听套接字就绪了，有新连接来了
        std::string clientip;
        uint16_t clientport = 0;
        int sock = _listensock.Accept(&clientip, &clientport); // 不会阻塞在这里，因为走到着一步就是因为底层已经有连接了，已经可以直接拿上来
        if (sock < 0) return;
        log(Info, "accept success, %s: %d, sock fd: %d", clientip.c_str(), clientport, sock);

        // 在获取到新连接后也不能直接读，因为只是把连接读上来了，但是对方可能没有发数据，所以当对方没立即发数据的时候，文件里就是空的，这时候读取会出问题哦
        int pos = 1;
        for(; pos < fd_max_num; pos++)
        {
            if (event_fds[pos].fd != defaultfd)
                continue; // 说明这个位置是已经被占用的位置
            else
                break;
        }
        // 走到这里有两种结果
        if (pos == fd_max_num) // 1，说明辅助数组已经被合法文件描述符占满了
        {
            log(Warning, "server is full, close %d now ", sock);
            close(sock); 

            //select满了就直接把新链接关了，但是现在满了的时候，我们可以直接进行扩充

        }
        else // 2，说明当前pos的位置可以被使用
        {
            event_fds[pos].fd = sock;
            event_fds[pos].events = POLLIN; //关心读事件，如果还想关心读和写，就| POLLOUT
            event_fds[pos].revents = non_event;
            PrintFd();
        }
    }
    void Recver(int fd, int pos)
    {
        //读事件就绪了
        char buffer[1024];
        ssize_t n = read(fd, buffer, sizeof(buffer) - 1); // 有bug，因为不能保证，对方通过网络发过来的报文可能不是完整的
        if (n > 0) //读成功
        {
            buffer[n] = 0;
            std::cout << "get a messge: " << buffer << std::endl;
        }
        else if (n == 0) //对方把连接关了，那么我服务器也要把套解析关了
        {
            log(Info, "client quit, me too, close fd is : %d", fd);
            close(fd);
            event_fds[pos].fd = defaultfd; // 这里本质是从select中移除
        }
        else //读出错，原因可能
        {
            log(Warning, "recv error: fd is : %d", fd);
            close(fd);
            event_fds[pos].fd = defaultfd; // 这里本质是从select中移除，因为start第一个循环会根据这个数组重新设置fd_set
        }
    }
    void Dispatcher()
    {
        for (int i = 0; i < fd_max_num; i++)
        {
            int fd = event_fds[i].fd;
            if (fd == defaultfd) continue; //判断文件描述符是否就绪
            //判断我们数组里的合法的文件描述符是否也在rfds里，如果在，代表该文件描述符已经就绪，接下来就是检测：1，读事件就绪    2，连接事件就绪
            if (event_fds[i].revents & POLLIN) //revent按位与上POLLIN，如果为真，就是读事件就绪了
            {
                //读事件就绪后也有两种情况
                if (fd == _listensock.Fd()) //1，如果等于listen套接字，就是连接事件就绪，就获取新连接，就把这个新的文件描述符继续添加进数组里
                {
                    Accepter(); // 连接管理器
                }
                else //2，如果不是linsten，那么就是别的文件描述符就绪了，也就是读事件就绪了
                {
                    Recver(fd, i);
                }
            }
        }
    }
    bool Start()
    {
        //首先要清楚文件描述符是代表文件的，一个文件有很多属性，我们现阶段以读写属性为例
        event_fds[0].fd = _listensock.Fd();
        event_fds[0].events = POLLIN; //用户告诉内核要等待的事件
        int timeout = 3000;

        while(true)
        {
            int n = poll(event_fds, fd_max_num, timeout); //可以发现确实比select简单很多，正常，因为接口的进步就是越来越简单
            switch(n)
            {
                case 0:
                    std::cout << "time out... " << std::endl;
                    break;
                case -1:
                    std::cerr << "poll error" << std::endl;
                    break;
                default:
                    //有事件就绪了,但是如果上层不处理，select就会一直通知你
                    std::cout << "get a new link " << std::endl;
                    Dispatcher(); //处理事件
                    break;
            }
        }
    }
    void PrintFd()
    {
        std::cout << "online fd list: ";
        for (int i = 0; i < fd_max_num; i++)
        {
            if (event_fds[i].fd == defaultfd)
                continue;
            std::cout << event_fds[i].fd << " ";
        }
        std::cout << std::endl;
    }
    ~PollServer()
    {
        _listensock.Close();
    }

private:
    Sock _listensock;
    uint16_t _port;
    struct pollfd event_fds[fd_max_num]; //表示事件对应的文件描述符集合
};

Main.cc：

#include "PollServer.hpp"
#include<memory>

void Usage(std::string proc)
{
    std::cout << "\n\rUsage: " << proc << "port[1024+]\n"
              << std::endl;
}


int main(int argc, char *argv[])
{
    if (argc != 2)
    {
        Usage(argv[0]);
        exit(1);
    }
    //单进程演示同时接收多个连接的消息
    uint16_t port = std::stoi(argv[1]);
    std::unique_ptr<PollServer> svr(new PollServer(port));
    svr->Init();
    svr->Start(); 
    return 0;
}

makefile:

poll_server:Main.cc
	g++ -o $@ $^ -std=c++11
.PNONY:clean
clean:
	rm -rf poll_server

3.4 poll优缺点

poll优点：

struct pollfd结构体可以将select的输入输出型参数分离，减少编程成本，使接口使用更简单
poll可监控的文件描述符没有限制，因为数组大小是用户定的，相当于没有限制
poll的优点，就是解决了select的缺点

poll缺点：

和select一样，poll也需要遍历fds数组来获取就绪的文件描述符
每次调用poll，都伴随着大量的struct pollfd在用户态和内核态之间的转换，并且当poll监视的文件描述符很多时，代价会很大
总结一下，就是poll需要“遍历”，当监视的套接字很多时，效率会下降，所以poll的缺点不再是功能上的了，而是效率上的了

四，epoll

4.1 关于epoll

epoll也是系统提供的一个多路转接接口：

epoll比poll多了一个e，这个e是extend也就是加强的意思，是为了解决处理大量句柄而做了改进的poll
epoll在2.5.44内核中被加入，几乎具备了select和poll的所有优点，所以epoll是被公认为的Linux 2.6 下性能最好的I/O就绪通知方法
但是高性能也是有代价的，epoll和poll虽然只差了一个字母，但是二者的实现方式大不相同，epoll的内部实现方式比较复杂

4.2 epoll的系列系统调用

4.2.1 epoll_create

该接口作用是创建epoll模型，其中参数size已被废弃，随便设置成大于0的数即可
创建模型成功后，打开epoll_file，返回它的文件描述符，所以可以初见端倪：epoll模型有文件描述符

4.2.2 epoll_wait

epfd：就是epoll_create返回值，表示epoll模型的文件描述符
events：是一个epoll_event结构体数组，内核会将已经就绪的事件拷贝到events数组中
maxevents：表示events数组中的元素个数，该值不能大于创建epoll模型时传入的size大小
timeout：和select和poll一样，单位是毫秒（ms）

epoll_events结构体

struct_epoll结构体有两个成员变量，events表示需要关心的事件，第二个成员data是一个联合体结构，一般使用结构中的fd，用来保存需要监视的文件描述符，关于联合体，我们之前已经介绍过：C语言进阶——自定义类型_c语言自义类型-CSDN博客

events的常用取值如下：

事件宏	描述
EPOLLIN	文件描述符可读
EPOLLOUT	文件描述符可写
EPOLLPRI	文件描述符有紧急数据可读
EPOLLERR	文件描述符发送错误
EPOLLHUP	对方把文件描述符close掉了
EPOLLET	将epoll的工作模式设置为ET（后面会讲）
EPOLLONESHOT	只监视一次事件，当这次事件监听完之后，如果还想继续监听该文件描述符，需要重新将文件描述符添加到epoll模型中

4.2.3 epoll_ctl

该接口作用是对epoll模型进行相关的管理工作，参数：

epfd：指定的epoll模型
op：表示具体的动作，有三个选项
fd：表示需要监视的文件描述符
event：表示需要监视该文件描述符上的哪些事件

op的三个选项如下：

EPOLL_CTL_ADD	表示添加新的文件描述符到epoll模型中
EPOLL_CTL_MOD	修改已经注册的文件描述符的监听事件
EPOLL_CTL_DEL	表示从epoll模型中删除指定的文件描述符

4.3 epoll模型

select和poll，都是用数组来管理文件描述符和事件的
网卡上面是网卡驱动，再上面就是OS，而OS不相信任何用户，所以就有了系统接口层system call，在上面就是用户了
select和poll只有一个接口，在操作系统上就是只有一个进程，遍历去文件描述符表，去遍历多个数组，当一个文件描述符都没有就绪的时候，进程就会被挂起等待，放到等待队列当中，在这之后，遍历啥的操作就是操作系统做的了，那么操作系统就被绑在这里了，不管有没有就绪，操作系统都得自己去查

问题：网卡是外设，那么操作系统在硬件层面，怎么知道网卡上有数据了呢？

解答：

所有的外设最大的特点，就是会以硬件中断的方式告诉我，就是外设一旦有事件就绪了，就给CPU发一个异步信号，称为硬件中断
每个设备都有自己的中断请求（IRQ），CPU根据IRQ将请求通过操作系统分发给相应的硬件驱动程序。硬件驱动通常是内核中的子程序，而不是独立的进程；例如，当网卡接收到数据包时，它会发出中断信号，CPU暂停当前任务，保存当前状态，并执行中断服务程序来处理这个事件。
硬件中断可以直接中断CPU的当前活动，并触发内核中相关代码的执行。对于需要时间处理的任务，中断代码本身也可能被其他硬件中断所中断。例如，时钟中断会导致内核调度代码挂起当前正在运行的代码，以便其他任务可以执行。

当某一个进程调用epoll_create接口时，Linux内核会为我们创建一个eventpoll结构体，也就是我们说的epoll模型：

struct eventpoll{  
    ....  
    /*红黑树的根节点，这颗树中存储着所有添加到epoll中的需要监控的事件*/  
    struct rb_root  rbr;  
    /*双链表中则存放着将要通过epoll_wait返回给用户的满足条件的事件*/  
    struct list_head rdlist;  
    ....  
};

操作系统为了支持epoll，为我们提供了三种机制：

①红黑树

操作系统在自身内部为我们维护了一颗红黑树，每个节点包含了一个结构体：

struct rb_node {
	int fd; //表示内核要关心的fd
	uint32_t event; //表示以位图形式呈现的要等待就绪的事件
	//还有其它的，比如链接事件
}

epoll模型中的红黑树本质就是在告诉内核，需要监视哪些文件描述符上的哪些事件，epoll_ctl函数本质就是在对这颗红黑树进行对应的增删改操作

②就绪队列

操作系统会为我们维护一个就绪队列ready_queue，一般是双链表实现的，一旦红黑树有一个节点里面的事件就绪了，操作系统就生成一个队列节点：

struct list_node{
	int fd; //表示已经就绪的fd
	uint32_t event; //已经就绪的事件
}

在epoll中，对于每一个事件都有一个对应的epitem结构体，红黑树和就绪队列当中的节点分别是基于epitem结构体中的rbn成员和rbllink成员，ffd记录的就是指定的文件描述符，event成员记录的就是对应的事件：

struct epitem{  
    struct rb_node  rbn;//红黑树节点  
    struct list_head    rdllink;//双向链表节点  
    struct epoll_filefd  ffd;  //事件句柄信息  
    struct eventpoll *ep;    //指向其所属的eventpoll对象  
    struct epoll_event event; //期待发生的事件类型  
}

③回调机制

网卡允许操作系统注册一些回调机制，操作系统在底层就会提供一个回调函数callback，比如网卡通过驱动把数据搬到了数据链路层，而一旦数据链路层有数据了，就会自动调用callback：

void callback()
{
	1，向上交付
	2，交付给Tcp的接收缓冲区
	3，查找红黑树，确认接收队列和哪一个是关联的，有没有关心比如EPOLLIN/RPOLLOUT
	4，如果查到了3，并且3号fd关心了EPOLLIN，并且这次到来的数据真的是我们3号fd对应的struct_file
	5，构建就绪节点，插入到就绪队列中
}

对于用户，只要从就绪队列里拿就好了，上面三套机制，我们就叫做epoll模型，在系统里，会把上面三个放到一块，这样后续要创建多个epoll模型时，方便先描述再组织

问题：进程如何找到epoll模型？

解答：

给epoll模型再创建一个struct_file，Linux下一切皆文件，所以把模型也当成文件
strucy_file中有指针指向这个epoll模型，所以未来把这个file对象添加到进程的文件描述符表里，这之后进程只要去找自己文件描述表，就能找到struct_file的文件描述符，然后再去找这个struct_file的指针，找到epoll模型的struct_file，再找到epoll模型，就可以访问所有东西了，
所以epoll_create返回值也是文件描述符，这时因为epoll模型也统一被接入到了struct_file里，所以poll和epoll是两个完全不同的东西

问题：epoll的三个接口的作用是什么？

解答：

epoll_create就是在操作系统内部创建struct_file，里面有指针指向epoll模型，然后把struct_file的文件描述符放到进程的文件描述符表里就OK了
epoll_ctl根据epfd找到epoll模型，找到红黑树，然后op增加修改或删除红黑树的节点
epoll_wait也是找到epoll模型，找到之后，第二个参数是一个输出型参数，是一个数组指针，它会把就绪队列中就绪的节点一个一个全放到数组里，这样用户就拿到了已经就绪的文件描述符

4.4 epoll服务器编写

源码：计算机网络/高级IO/epoll · 小堃学编程/Linux学习 - 码云 - 开源中国

为了方便，这里就使用cmake生成makefile文件了

CMakeLists.txt：

cmake_minimum_required(VERSION 3.10) 
project(EpollServer)
add_executable(epoll_server Main.cc)

由于epoll的接口有三个，所以我们先把这三个接口都给封装一下，也就是Epoller.hpp：

#pragma once
#include <cerrno>
#include<sys/epoll.h>
#include<cstring>
#include "Log.hpp"

class nocopy //让Epoller不要拷贝
{
public:
    nocopy(){}
    nocopy(const nocopy &) = delete;
    const nocopy& operator=(const nocopy &) = delete;
};

static const int size = 128;
class Epoller : public nocopy
{
public:
    Epoller()
    {
        _epfd = epoll_create(size); //创建epoll模型
        if(_epfd == -1)
        {
            log(Error, "epoll_create error: %s", strerror(errno));
        }
        else
        {
            log(Info, "epollcreate success: %d, _epfd");
        }
    }
    int EpollerWait(struct epoll_event revents[], /*timeout*/int num)
    {
        int n = epoll_wait(_epfd, revents, num, -1); //阻塞等待，
        return n; //n表示有多少个文件描述符就绪了
    }
    int EpollUpdate(int oper, int sock, uint32_t event)
    {
        int n = 0;
        if(oper == EPOLL_CTL_DEL)
        {
            n = epoll_ctl(_epfd, oper, sock, nullptr);
            if (n != 0)
            {
                log(Error, "epoll_ctl delete error!");
            }
        }
        else
        {
            struct epoll_event ev; //构建要添加的结构体
            ev.events = event;
            ev.data.fd = sock; //方便后面处理事件时知道是哪个文件描述符的哪个事件就绪了，因为这个data是一个联合体

            n = epoll_ctl(_epfd, oper, sock, &ev); //注册进去
            if (n != 0)
            {
                log(Error, "epoll_ctl add error!");
            }
        }
        return n; 
    }

    ~Epoller()
    {
       if (_epfd >= 0) close(_epfd);
    }

private:
    int _epfd; //创建的epoll模型
    int _timeout{3000};
};

下面就是EpollServer.hpp的代码：

#pragma once
#include <iostream>
#include <sys/epoll.h>
#include <memory>
#include "Log.hpp"
#include "Socket.hpp"
#include "Epoller.hpp"

static const int num = 64;

class EpollServer : public nocopy
{
public:
    EpollServer(uint16_t port)
        : _port(port)
        , _listensock(new Sock())
        , _epoller(new Epoller())
    {}
    void Init()
    {
        _listensock->Socket();
        _listensock->Bind(_port);
        _listensock->Listen();

        log(Info, "sock create success: %d\n", _listensock->Fd());
    }
    void Accepter()
    {
        std::string clientip;
        uint16_t clientport;
        int sock = _listensock->Accept(&clientip, &clientport); //不会再阻塞了
        if(sock > 0)
        {
            //不能直接读取，我们再添加到红黑树里，统一处理
            _epoller->EpollUpdate(EPOLL_CTL_ADD, sock, EPOLLIN);
            log(Info, "get a new link, client info@ %s:%d", clientip.c_str(), clientport);
        }
    }
    void Recver(int fd)
    {
        // 简单模拟事件处理
        char buffer[1024];
        // 从套接字里面读信息
        ssize_t n = read(fd, buffer, sizeof(buffer) - 1); // bug?
        if (n > 0)
        {
            buffer[n] = 0;
            std::cout << "get a messge: " << buffer << std::endl;

            //再把数据返回给客户端
            std::string echo_str = "server echo &  ";
            echo_str += buffer;
            write(fd, echo_str.c_str(), echo_str.size());
        }
        else if (n == 0)
        {
            log(Info, "client quit, close fd is : %d", fd);
            //链接一旦断开，直接把红黑树里对应的节点去掉，节省资源
            //移除时，要想保证该文件描述符是合法的，所以要想移除再close关闭
            _epoller->EpollUpdate(EPOLL_CTL_DEL, fd, 0);
            close(fd);
        }
        else
        {
            log(Warning, "recv error: fd is : %d", fd);
            _epoller->EpollUpdate(EPOLL_CTL_DEL, fd, 0);
            close(fd);
        }
    }
    void Dispatcher(struct epoll_event revs[], int num)
    {
        for(int i = 0;i < num; i++) //根据num，就可以一次性遍历获取并处理所有已经就绪的事件
        {
            uint32_t events = revs[i].events; //表示已经就绪的事件
            int fd = revs[i].data.fd; //获取已经就绪的文件描述符
            //现在就知道了，哪个文件描述符的哪个事件就绪了
            if (events & EPOLLIN) //读事件就绪了
            {
                if(fd == _listensock->Fd()) //说明来了一个新链接，需要accept获取上来
                {
                    Accepter(); //事件派发
                }   
                else //说明其它fd上面的读事件就绪
                {
                    Recver(fd); //事件处理
                }             
            }
            else if(events & EPOLLOUT) //写事件就绪了
            {
                //可以在这里扩充写的处理
            }
            else //其它事件就绪了
            {
            }
        }
    }

    void Start()
    {
        //将listensock添加到epoll中，这句话本质是：将listensock和它关心的事件，都给添加到内核epoll模型中的红黑树中
        _epoller->EpollUpdate(EPOLL_CTL_ADD, _listensock->Fd(), EPOLLIN); //我想把“读事件”，“添加”进去
        struct epoll_event revs[num];
        while(true)
        {
            int n = _epoller->EpollerWait(revs, num);
            if(n > 0) //有事件就绪了
            {
                log(Debug, "event happened, fd is : %d", revs[0].data.fd);
                Dispatcher(revs, n); //处理事件逻辑                
            }
            else if(n = 0) //超时了
            {
                log(Info, "time out ...");
            }
            else //错误了
            {
                log(Info, "epoll wait error");
            }
        }
    }
    ~EpollServer(){}
private:
    std::shared_ptr<Sock> _listensock;
    std::shared_ptr<Epoller> _epoller;
    uint16_t _port; //端口号
};

Main.cc：

#include "EpollServer.hpp"

void Usage(std::string proc)
{
    std::cout << "\n\rUsage: " << proc << "port[1024+]\n"
              << std::endl;
}

int main(int argc, char* argv[])
{
    if(argc != 2)
    {
        Usage(argv[0]);
        return 1;
    }
    auto port = std::stoi(argv[1]);
    std::unique_ptr<EpollServer> svr(new EpollServer(port));
    svr->Init();
    svr->Start();
    return 0;
}

下面是测试结果：

可以看到，我们是单进程，但是却可以同时接收多个客户端发来的请求，这就是多路转接的好处，相比多进程和多线程，多路转接所占用的资源更少

4.5 epoll优点

接口使用方便：虽然拆分成了三个函数，但是反而使用起来更方便高效不需要每次循环都设置关注的文件描述符，也做到了输入输出参数分离开
检测就绪时间复杂度为O(1)，因为检测步骤只有一个，就是看就绪队列是否为空，一个if就可以解决；获取就绪就是O(n)，就是把队列里的节点一个一个拷贝到用户层，一个while解决
fd和event没有上限，因为所有文件描述符和要关心的事件都是红黑树管理的，这颗红黑树有多大和epoll没关系
数据拷贝轻量：只在合适的时候调用 EPOLL_CTL_ADD 将文件描述符结构拷贝到内核中，这个操作并不频繁(而select/poll都是每次循环都要进行拷贝)
事件回调机制：避免使用遍历，而是使用回调函数的方式，将就绪的文件描述符结构加入到就绪队列中，epoll_wait返回直接访问就绪队列就知道哪些文件描述符就绪。这个操作时间复杂度O(1)，即使文件描述符数目很多，效率也不会受到影响
没有数量限制：文件描述符数目无上限
有select，poll的所有优势，也解决了它们的缺点，epoll_wait获取就绪队列数组后，返回值n，表示有几个fd就绪了，而且会把就绪队列的节点一个一个弹出，这也意味着，返回给用户的就绪事件是连续的，意味着上层用户处理所有已经就绪的事件，不用再一个个检测fd是否非法的，或者是否就绪，只要拿着n就可以按指定次数处理了，省去了用户层很多的多余动作

五，epoll两种工作模式

5.1 LT，ET工作模式

epoll有两种工作模式，分别是水平触发模式和边缘触发模式

水平触发（LT，Level Triggered）

只要底层有数据，epoll就会一直通知用户，就像数字电路的高电平触发一样，只要一直处于高电平，则会一直触发

epoll默认状态下就是LT模式

所哟当epoll检测到事件就绪时，可以不立即进行处理，因为只要底层数据没处理完，下一次epoll还会通知用户
select和poll就是LT模式的
支持阻塞读写和非阻塞读写

边缘触发（ET，Edge Triggered）

只有底层就绪事件数量发生变化时，epoll才会通知用户，就像数字电路中的上升沿触发一样，只有当电平由低到高那一瞬间才会触发

如果要将epoll改为工作模式，则需要在epoll_create设置EPOLLET选项

由于ET模式只通知一次，所以当epoll检测到底层有事件就绪了，必须立即进行处理，而且必须全部处理完毕，因为ET下的epoll只会通知一次，如果数据没有处理完相当于丢失了
ET模式下的epoll通知用户次数比LT少，因此ET的性能一般比LT高，Nginx就是默认采用ET模式的epoll
只支持非阻塞的读写

问题：ET模式下应该如何进行读写？

解答：倒逼用户读事件必须一次性读完，写事件必须一次性把缓冲区写满，所以读数据时必须循环调用recv，写事件时必须循环调用send

循环调用recv，直到某次recv实际读到的字节数小于期望独到的字节数，则说明本子底层数据已经读取完毕
但有可能最后一次调用recv时，刚好实际读取的字节数和期望值相同，此时底层没有数据了，再次调用recv就会阻塞住
这里的阻塞是很致命的，比如我们的服务器是单进程服务器，只要这里阻塞住，那么以后该数据再也不就绪，那么我们的服务器将会永远阻塞在这里，相当于挂掉了
所以我们在ET模式下调用recv函数进行读取，必须将对应的文件描述符设置为非阻塞，send函数同理

面试题：为什么ET的效率更高？

解答：

因为相同时间内，ET通知的数量比LT多
ET也会让我们IO的效率也变高，因为我每次通知上层有数据来了之后，倒逼程序员，每次通知都必须把本轮数据全部取走 --> 循环读取，直到读取出错 --> fd默认是阻塞的 --> ET下的所有fd必须是非阻塞的，因为如果是阻塞的，那么循环读取到最后没有数据时依然会阻塞，导致服务器挂起了
当把数据全取走之后，Tcp会向对方通告一个更大的窗口，从而从概率上让对方一次发送更多数据，能提升网络传输效率