Linux多路复用

  IO多路复用是一种操作系统的技术，用于在单个线程或进程中管理多个输入输出操作。它的主要目的是通过将多个IO操作合并到一个系统调用中来提高系统的性能和资源利用率，避免了传统的多线程或多进程模型中因为阻塞IO而导致的资源浪费和低效率问题。

  在IO多路复用中，通常使用的系统调用有 select()、poll()、epoll() 等，它们允许程序等待多个文件描述符（sockets、文件句柄等）中的任何一个变为可读或可写，然后再进行实际的IO操作。这种模型相比于传统的多线程或多进程模型，具有更高的并发处理能力和更低的系统开销。

  

1. select

1.1 select的概念

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

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

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

  

1.2 select的函数使用

 int select(int nfds, fd_set *readfds, fd_set *writefds,
 			fd_set *exceptfds, struct timeval *timeout);

函数参数：

  nfds：是需要监视的最大的文件描述符值+1。

  readfds：需要检测的可读文件描述符的集合。

  writefds：需要检测的可写文件描述符的集合。

  exceptfds：需要检测的异常文件描述符的集合。

  timeout：为结构体timeval，用来设置select()的等待时间；

  当timeout等于NULL：则表示select（）没有timeout，select将一直被阻塞，直到某个文件描述符上发生了事件;

  当timeout为0：仅检测描述符集合的状态，然后立即返回，并不等待外部事件的发生。

  当timeout为特定的时间值：如果在指定的时间段里没有事件发生，select将超时返回。

  其中的可读，可写，异常文件描述符的集合是一个fd_set类型，fd_set是系统提供的位图类型，位图的位置是否是1，表示是否关系该事件。

  例如：

    输入时：假如我们要关心 0 1 2 3 文件描述符

    0000 0000->0000 1111 比特位的位置，表示文件描述符的编号
         比特位的内容 0or1 表示是否需要内核关心

    输出时：

    0000 0100->此时表示文件描述符的编号
         比特位的内容 0or1哪些用户关心的fd 上面的读事件已经就绪了，这里表示2描述符就绪了

  

  系统提供了关于fd_set的接口，便于我们使用位图：

 void FD_CLR(int fd, fd_set *set); // 用来清除描述词组set中相关fd 的位
 int FD_ISSET(int fd, fd_set *set); // 用来测试描述词组set中相关fd 的位是否为真
 void FD_SET(int fd, fd_set *set); // 用来设置描述词组set中相关fd的位
 void FD_ZERO(fd_set *set); // 用来清除描述词组set的全部位

  

函数返回值：

  执行成功则返回文件描述词状态已改变的个数。

  如果返回0代表在描述词状态改变前已超过timeout时间，没有返回。

  当有错误发生时则返回-1，错误原因存于errno，此时参数readfds，writefds, exceptfds和timeout的值变成不可预测。

  错误值可能为：

  EBADF 文件描述词为无效的或该文件已关闭
  EINTR 此调用被信号所中断
  EINVAL 参数n 为负值。
  ENOMEM 核心内存不足

  

select的执行过程：

  （1）执行fd_set set; FD_ZERO(&set);则set用位表示是0000,0000。

  （2）若fd＝5,执行FD_SET(fd,&set);后set变为0001,0000(第5位置为1) 。

  （3）若再加入fd＝2，fd=1,则set变为0001,0011 。

  （4）执行select(6,&set,0,0,0)阻塞等待，表示最大文件描述符+1是6，监控可读事件，立即返回。

  （5）若fd=1,fd=2上都发生可读事件，则select返回，此时set变为0000,0011。注意：没有事件发生的fd=5被清空。

  

1.3 select的优缺点

select的特点：

  （1）可监控的文件描述符个数取决与sizeof(fd_set)的值。一般大小是1024，但是fd_set的大小可以调整。

  （2）将fd加入select监控集的同时，还要再使用一个数据结构array保存放到select监控集中的fd。

    1. 是用于再select 返回后，array作为源数据和fd_set进行FD_ISSET判断。

    2. 是select返回后会把以前加入的但并无事件发生的fd清空，则每次开始select前都要重新从array取得fd逐一加入(FD_ZERO最先)，扫描array的同时取得fd最大值maxfd，用于select的第一个参数。

  

select缺点

  （1）每次调用select, 都需要手动设置fd集合, 从接口使用角度来说也非常不便。

  （2）每次调用select，都需要把fd集合从用户态拷贝到内核态，这个开销在fd很多时会很大。

  （3）同时每次调用select都需要在内核遍历传递进来的所有fd，这个开销在fd很多时也很大。

  （4）select支持的文件描述符数量太小。

  

select使用代码：

#pragma once

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

using namespace std;

static const uint16_t defaultport = 888;
static const int fd_num_max = (sizeof(fd_set) * 8);
int defaultfd = -1;

class SelectServer
{
public:
    SelectServer(uint16_t port = defaultport) : _port(port)
    {
        for (int i = 0; i < fd_num_max; i++)
        {
            fd_array[i] = defaultfd;
            // std::cout << "fd_array[" << i << "]" << " : " << fd_array[i] << std::endl;
        }
    }
    bool Init()
    {
        _listensock.Socket();
        _listensock.Bind(_port);
        _listensock.Listen();

        return true;
    }
    void Accepter()
    {
        // 我们的连接事件就绪了
        std::string clientip;
        uint16_t clientport = 0;
        int sock = _listensock.Accept(&clientip, &clientport); // 会不会阻塞在这里？不会
        if (sock < 0) return;
        lg(Info, "accept success, %s: %d, sock fd: %d", clientip.c_str(), clientport, sock);

        // sock -> fd_array[]
        int pos = 1;
        for (; pos < fd_num_max; pos++) // 第二个循环
        {
            if (fd_array[pos] != defaultfd)
                continue;
            else
                break;
        }
        if (pos == fd_num_max)
        {
            lg(Warning, "server is full, close %d now!", sock);
            close(sock);
        }
        else
        {
            fd_array[pos] = sock;
            PrintFd();
            // TODO
        }
    }
    void Recver(int fd, int pos)
    {
        // demo
        char buffer[1024];
        ssize_t n = read(fd, buffer, sizeof(buffer) - 1); // bug?
        if (n > 0)
        {
            buffer[n] = 0;
            cout << "get a messge: " << buffer << endl;
        }
        else if (n == 0)
        {
            lg(Info, "client quit, me too, close fd is : %d", fd);
            close(fd);
            fd_array[pos] = defaultfd; // 这里本质是从select中移除
        }
        else
        {
            lg(Warning, "recv error: fd is : %d", fd);
            close(fd);
            fd_array[pos] = defaultfd; // 这里本质是从select中移除
        }
    }
    void Dispatcher(fd_set &rfds)
    {
        for (int i = 0; i < fd_num_max; i++) // 这是第三个循环
        {
            int fd = fd_array[i];
            if (fd == defaultfd)
                continue;

            if (FD_ISSET(fd, &rfds))
            {
                if (fd == _listensock.Fd())
                {
                    Accepter(); // 连接管理器
                }
                else // non listenfd
                {
                    Recver(fd, i);
                }
            }
        }
    }
    void Start()
    {
        int listensock = _listensock.Fd();
        fd_array[0] = listensock;
        for (;;)
        {
            fd_set rfds;
            FD_ZERO(&rfds);

            int maxfd = fd_array[0];
            for (int i = 0; i < fd_num_max; i++) // 第一次循环
            {
                if (fd_array[i] == defaultfd)
                    continue;
                FD_SET(fd_array[i], &rfds);
                if (maxfd < fd_array[i])
                {
                    maxfd = fd_array[i];
                    lg(Info, "max fd update, max fd is: %d", maxfd);
                }
            }

            // accept?不能直接accept！检测并获取listensock上面的事件，新连接到来，等价于读事件就绪

            // struct timeval timeout = {1, 0}; // 输入输出，可能要进行周期的重复设置
            struct timeval timeout = {0, 0}; // 输入输出，可能要进行周期的重复设置
            // 如果事件就绪，上层不处理，select会一直通知你！
            // select告诉你就绪了，接下来的一次读取，我们读取fd的时候，不会被阻塞
            // rfds: 输入输出型参数。 1111 1111 -> 0000 0000
            int n = select(maxfd + 1, &rfds, nullptr, nullptr, /*&timeout*/ nullptr);
            switch (n)
            {
            case 0:
                cout << "time out, timeout: " << timeout.tv_sec << "." << timeout.tv_usec << endl;
                break;
            case -1:
                cerr << "select error" << endl;
                break;
            default:
                // 有事件就绪了，TODO
                cout << "get a new link!!!!!" << endl;
                Dispatcher(rfds); // 就绪的事件和fd你怎么知道只有一个呢？？？
                break;
            }
        }
    }
    void PrintFd()
    {
        cout << "online fd list: ";
        for (int i = 0; i < fd_num_max; i++)
        {
            if (fd_array[i] == defaultfd)
                continue;
            cout << fd_array[i] << " ";
        }
        cout << endl;
    }
    ~SelectServer()
    {
        _listensock.Close();
    }

private:
    Sock _listensock;
    uint16_t _port;
    int fd_array[fd_num_max];   // 数组, 用户维护的！
    // int wfd_array[fd_num_max];
};

  

2. poll

2.1 poll的概念

  poll和select实现原理基本类似，

  poll只为了解决select的两个硬伤：

  1.等待的fd是有上限的，（底层类似链表储存实现，而不是位图）

  2.每次要对关心的fd进行事件重置，（pollfd结构包含了要监视的event和发生的event，使用前后不用初始化fd_set）

  

2.2 poll的函数使用

int poll(struct pollfd *fds, nfds_t nfds, int timeout);

// pollfd结构
struct pollfd {
	int fd; /* file descriptor */
	short events; /* requested events */
	short revents; /* returned events */
};

函数参数解释：

  fds：是一个poll函数监听的结构列表. 每一个元素中, 包含了三部分内容: 文件描述符, 监听的事件集合, 返回的事件集合。

  nfds：表示fds数组的长度。

  timeout：表示poll函数的超时时间, 单位是毫秒(ms)。

返回结果：

  返回值小于0, 表示出错。

  返回值等于0, 表示poll函数等待超时。

  返回值大于0, 表示poll由于监听的文件描述符就绪而返回。

  

2.3 poll的优缺点

poll的优点

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

  （2）poll并没有最大数量限制 (但是数量过大后性能也是会下降)。

poll的缺点

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

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

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

  

poll使用代码：

#pragma once

#include <iostream>
#include <poll.h>
#include <sys/time.h>
#include "../select/Socket.hpp"

using namespace std;

static const uint16_t defaultport = 8888;
static const int fd_num_max = 64;
int defaultfd = -1;
int non_event = 0;

class PollServer
{
public:
    PollServer(uint16_t port = defaultport) : _port(port)
    {
        for (int i = 0; i < fd_num_max; i++)
        {
            _event_fds[i].fd = defaultfd;
            _event_fds[i].events = non_event;
            _event_fds[i].revents = non_event;

            // std::cout << "fd_array[" << i << "]" << " : " << fd_array[i] << std::endl;
        }
    }
    bool Init()
    {
        _listensock.Socket();
        _listensock.Bind(_port);
        _listensock.Listen();

        return true;
    }
    void Accepter()
    {
        // 我们的连接事件就绪了
        std::string clientip;
        uint16_t clientport = 0;
        int sock = _listensock.Accept(&clientip, &clientport); // 会不会阻塞在这里？不会
        if (sock < 0) return;
        lg(Info, "accept success, %s: %d, sock fd: %d", clientip.c_str(), clientport, sock);

        // sock -> fd_array[]
        int pos = 1;
        for (; pos < fd_num_max; pos++) // 第二个循环
        {
            if (_event_fds[pos].fd != defaultfd)
                continue;
            else
                break;
        }
        if (pos == fd_num_max)
        {
            lg(Warning, "server is full, close %d now!", sock);
            close(sock);
            // 扩容
        }
        else
        {
            // fd_array[pos] = sock;
            _event_fds[pos].fd = sock;
            _event_fds[pos].events = POLLIN;
            _event_fds[pos].revents = non_event;
            PrintFd();
            // TODO
        }
    }
    void Recver(int fd, int pos)
    {
        // demo
        char buffer[1024];
        ssize_t n = read(fd, buffer, sizeof(buffer) - 1); // bug?
        if (n > 0)
        {
            buffer[n] = 0;
            cout << "get a messge: " << buffer << endl;
        }
        else if (n == 0)
        {
            lg(Info, "client quit, me too, close fd is : %d", fd);
            close(fd);
            _event_fds[pos].fd = defaultfd; // 这里本质是从select中移除
        }
        else
        {
            lg(Warning, "recv error: fd is : %d", fd);
            close(fd);
            _event_fds[pos].fd = defaultfd; // 这里本质是从select中移除
        }
    }
    void Dispatcher()
    {
        for (int i = 0; i < fd_num_max; i++) // 这是第三个循环
        {
            int fd = _event_fds[i].fd;
            if (fd == defaultfd)
                continue;

            if (_event_fds[i].revents & POLLIN)
            {
                if (fd == _listensock.Fd())
                {
                    Accepter(); // 连接管理器
                }
                else // non listenfd
                {
                    Recver(fd, i);
                }
            }
        }
    }
    void Start()
    {
        _event_fds[0].fd = _listensock.Fd();
        _event_fds[0].events = POLLIN;
        int timeout = 3000; // 3s
        for (;;)
        {
            int n = poll(_event_fds, fd_num_max, timeout);
            switch (n)
            {
            case 0:
                cout << "time out... " << endl;
                break;
            case -1:
                cerr << "poll error" << endl;
                break;
            default:
                // 有事件就绪了，TODO
                cout << "get a new link!!!!!" << endl;
                Dispatcher(); // 就绪的事件和fd你怎么知道只有一个呢？？？
                break;
            }
        }
    }
    void PrintFd()
    {
        cout << "online fd list: ";
        for (int i = 0; i < fd_num_max; i++)
        {
            if (_event_fds[i].fd == defaultfd)
                continue;
            cout << _event_fds[i].fd << " ";
        }
        cout << endl;
    }
    ~PollServer()
    {
        _listensock.Close();
    }

private:
    Sock _listensock;
    uint16_t _port;
    struct pollfd _event_fds[fd_num_max]; // 数组, 用户维护的！
    // struct pollfd *_event_fds;

    // int fd_array[fd_num_max];
    // int wfd_array[fd_num_max];
};

  

3. epoll

3.1 epoll的概念

  epoll: 是为处理大批量句柄而作了改进的poll（真的是大改进）。

  epoll是IO多路复用技术，在实现上维护了一个用于返回触发事件的Socket的链表和一个记录监听事件的红黑树，epoll的高效体现在：

  （1）对监听事件的修改是 log N(红黑树)。

  （2）用户程序无需遍历所有的Socket(发生事件的Socket被放到链表中直接返回)。

  （3）内核无需遍历所有的套接字，内核使用回调函数在事件发生时直接转到对应的处理函数。

  

3.2 epoll的函数使用

  epoll 有3个相关的系统调用：

epoll_create

int epoll_create(int size);

  创建一个epoll的句柄，自从linux2.6.8之后，size参数是被忽略的，用完之后, 必须调用close()关闭。

  

epoll_ctl

int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);

epoll的事件注册函数：

  它不同于select()是在监听事件时告诉内核要监听什么类型的事件, 而是在这里先注册要监听的事件类型。

  第一个参数是epoll_create()的返回值(epoll的句柄)。

  第二个参数表示动作，用三个宏来表示。

  第三个参数是需要监听的fd。

  第四个参数是告诉内核需要监听什么事。

第二个参数的取值:

  EPOLL_CTL_ADD ：注册新的fd到epfd中。

  EPOLL_CTL_MOD ：修改已经注册的fd的监听事件。

  EPOLL_CTL_DEL ：从epfd中删除一个fd。

struct epoll_event结构如下：

  

events可以是以下几个宏的集合：

  EPOLLIN : 表示对应的文件描述符可以读 (包括对端SOCKET正常关闭)。

  EPOLLOUT : 表示对应的文件描述符可以写。

  EPOLLPRI : 表示对应的文件描述符有紧急的数据可读 (这里应该表示有带外数据到来)。

  EPOLLERR : 表示对应的文件描述符发生错误。

  EPOLLHUP : 表示对应的文件描述符被挂断。

  EPOLLET : 将EPOLL设为边缘触发(Edge Triggered)模式, 这是相对于水平触发(Level Triggered)来说的。

  EPOLLONESHOT：只监听一次事件, 当监听完这次事件之后, 如果还需要继续监听这个socket的话, 需要再次把这个socket加入到EPOLL队列里。

  

epoll_wait

int epoll_wait(int epfd, struct epoll_event * events, int maxevents, int timeout);

收集在epoll监控的事件中已经发送的事件：

  参数events是分配好的epoll_event结构体数组。

  epoll将会把发生的事件赋值到events数组中 (events不可以是空指针，内核只负责把数据复制到这个events数组中，不会去帮助我们在用户态中分配内存)。

  maxevents告之内核这个events有多大，这个 maxevents的值不能大于创建epoll_create()时的size。

  参数timeout是超时时间 (毫秒，0会立即返回，-1是永久阻塞)。

  如果函数调用成功，返回对应I/O上已准备好的文件描述符数目，如返回0表示已超时, 返回小于0表示函数失败。

  

epoll原理：

  （1）当某一进程调用epoll_create方法时，Linux内核会创建一个eventpoll结构体，这个结构体中有两个成员与epoll的使用方式密切相关。

  （2）每一个epoll对象都有一个独立的eventpoll结构体，用于存放通过epoll_ctl方法向epoll对象中添加进来的事件。

  （3）这些事件都会挂载在红黑树中，如此，重复添加的事件就可以通过红黑树而高效的识别出来(红黑树的插入时间效率是lgn，其中n为树的高度)。

  （4）而所有添加到epoll中的事件都会与设备(网卡)驱动程序建立回调关系，也就是说，当响应的事件发生时会调用这个回调方法。

  （5）这个回调方法在内核中叫ep_poll_callback,它会将发生的事件添加到rdlist双链表中。

  （6）在epoll中，对于每一个事件，都会建立一个epitem结构体。

  （7）当调用epoll_wait检查是否有事件发生时，只需要检查eventpoll对象中的rdlist双链表中是否有epitem元素即可。

  （8）如果rdlist不为空，则把发生的事件复制到用户态，同时将事件数量返回给用户. 这个操作的时间复杂度是O(1)。
  

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

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

总结一下, epoll的使用过程简单看就三步：

  （1）调用epoll_create创建一个epoll句柄。

  （2）调用epoll_ctl, 将要监控的文件描述符进行注册。

  （3）调用epoll_wait, 等待文件描述符就绪。

  

3.3 epoll的优点

  （1）接口使用方便: 虽然拆分成了三个函数，但是反而使用起来更方便高效，不需要每次循环都设置关注的文件描述符，也做到了输入输出参数分离开。

  （2）数据拷贝轻量: 只在合适的时候调用 EPOLL_CTL_ADD 将文件描述符结构拷贝到内核中，这个操作并不频繁(而select/poll都是每次循环都要进行拷贝)。

  （3）事件回调机制: 避免使用遍历，而是使用回调函数的方式，将就绪的文件描述符结构加入到就绪队列中，epoll_wait 返回直接访问就绪队列就知道哪些文件描述符就绪，这个操作时间复杂度O(1)，即使文件描述符数目很多，效率也不会受到影响。

  （4）没有数量限制: 文件描述符数目无上限。

  

3.4 epoll工作模式

  epoll默认：LT模式，事件到来但是上层不处理，高电平，一直有效。
          ET模式，数据或者连接，从无到有，从有到多，变化的时候才通知我们一次。

  ET的通知效率更高：倒逼程序员，每次通知都必须把本轮的数据取走 -> 循环读取，读取错误 -> fd默认是阻塞的 -> ET，所有的fd必须是非阻塞的。

  ET的IO效率也更高 -> tcp会向对方通告一个更大的窗口，从而概率上让对方一次给我发生更多数据，如果LT每次也可以就绪，那效率差不多。

  本质就是向就绪队列，添加一次或者是多次就绪节点。

  

Epoller.hpp Epoller对epoll进行封装

#pragma once

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

//封装我们的epoll，epoll公有继承于我们的nocopy类，不能被拷贝
class Epoller: public nocopy 
{
    static const int size=128;

public:
    Epoller()
    {
        _epfd=epoll_create(size);
        if(_epfd<0)
        {
            lg(Error,"epoll_create error: %s",strerror(errno));
        }
        else
        {
            lg(Info,"epoll_create success: %d",_epfd);
        }
    }

    //进行epoll事件等待
    //返回的是就绪事件的数量
    int EpollerWait(struct epoll_event revents[], int num)
    {
        int n=epoll_wait(_epfd,revents,num,-1/*_timeout*/);
        return n;
    }

    //我们所要更新的时间操作和套接字监控的事件
    int EpollerUpdate(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)
            {
                lg(Error,"epoll_ctl delete error!");
            }
        }
        else //新增和修改
        {
            struct epoll_event ev;
            ev.events=event;
            ev.data.fd=sock; //传入sock，方便我们知道是哪一个fd就绪

            //完成了我们对于哪一个文件和那一个文件的描述符进行事件关心
            //接下来进行注册
            n=epoll_ctl(_epfd,oper,sock,&ev);
            if(n!=0)
            {
                lg(Error,"epoll_ctl error!");
            }
        }
        return n;
    }

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

private:
    int _epfd;
    int _timeout{3000};
};

  

EpollServer.hpp Epoll服务器

#pragma once

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

uint32_t EVENT_IN = (EPOLLIN); //表示更新读事件
uint32_t EVENT_OUT = (EPOLLOUT); //表示更新写事件

class EpollServer : public nocopy
{
    static const int num = 64;

public:
    EpollServer(uint16_t port)
        : _port(port),
          _listsocket_ptr(new Sock()),
          _epoll_ptr(new Epoller())
    {}

    void Init()
    {
        _listsocket_ptr->Socket();
        _listsocket_ptr->Bind(_port);
        _listsocket_ptr->Listen();

        lg(Info,"create listen socket success: %d\n",_listsocket_ptr->Fd());
    }

    void Accepter()
    {
        //获取了一个连接   
        std::string clientip;
        uint16_t clientport;
        int sock=_listsocket_ptr->Accept(&clientip,&clientport);
        if(sock>0)
        {
            //我们不能直接读取数据
            //ssize_t n=read(sock,...);
            _epoll_ptr->EpollerUpdate(EPOLL_CTL_ADD,sock,EVENT_IN);
            lg(Info,"get a new link, client info @ %s:%d",clientip.c_str(),clientport);
        }
    }

    void Recver(int fd)
    {
        // demo
        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;
            // wrirte
            std::string echo_str = "server echo $ ";
            echo_str += buffer;
            write(fd, echo_str.c_str(), echo_str.size());
        }
        else if (n == 0)
        {
            lg(Info, "client quit, me too, close fd is : %d", fd);
            //细节3
            _epoll_ptr->EpollerUpdate(EPOLL_CTL_DEL, fd, 0);
            close(fd);
        }
        else
        {
            lg(Warning, "recv error: fd is : %d", fd);
            _epoll_ptr->EpollerUpdate(EPOLL_CTL_DEL, fd, 0);
            close(fd);
        }
    }

    void Dispatcher(struct epoll_event revs[], int num)
    {
        //遍历获取文件描述符中已经就绪的事件
        for(int i=0;i<num;i++)
        {
            uint32_t events=revs[i].events;
            int fd=revs[i].data.fd;
            if(events & EVENT_IN) //判断事件类型，这是读事件就绪
            {
                if(fd==_listsocket_ptr->Fd())
                {
                    //获取了一个连接   
                    Accepter();
                }
                else 
                {
                    //其他fd上面的普通读取事件就绪
                    Recver(fd);
                }
            }
            else if(events & EVENT_OUT) //写事件就绪
            {

            }
            else 
            {

            }
        }
    }

    //开始我们的epoll事件监听
    void Start()
    {
        //将listensock添加到epoll中 -> listensock和他关心的事件，添加到内核epoll模型的rb_tree
        _epoll_ptr->EpollerUpdate(EPOLL_CTL_ADD,_listsocket_ptr->Fd(),EVENT_IN);
        //我们将我们的监听套接字listsocket给epoll进行读事件管理，接下来由红黑树自动关心我们的事件
        struct epoll_event revs[num];
        for(;;)
        {
            int n=_epoll_ptr->EpollerWait(revs,num);
            if(n>0)
            {
                //有事件就绪
                lg(Debug,"event happend, fd is %d",revs[0].data.fd);
                //处理就绪事件
                Dispatcher(revs,n);
            }
            else if(n==0)
            {
                lg(Info,"time out...");
            }
            else
            {
                lg(Error,"epoll wait error");
            }
        }
    }

    ~EpollServer()
    {
        _listsocket_ptr->Close();
    }

private:
    std::shared_ptr<Sock> _listsocket_ptr;
    std::shared_ptr<Epoller> _epoll_ptr;
    uint16_t _port;
};