I/O多路转接之epoll（实现epoll版本的TCP服务器）

epoll的相关系统调用

1. epoll_create

int epoll_create(int size);

创建一个epoll的句柄.

• 自从linux2.6.8之后，size参数是被忽略的（>0即可）.
• 用完之后, 必须调用close()关闭

2. 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队列里.

3. 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工作原理

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

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

• 每一个epoll对象都有一个独立的eventpoll结构体，用于存放通过epoll_ctl方法向epoll对象中添加进来的事件.
• 这些事件都会挂载在红黑树中，如此，重复添加的事件就可以通过红黑树而高效的识别出来(红黑树的插入时间效率是lgn，其中n为树的高度).
• 而所有添加到epoll中的事件都会与设备(网卡)驱动程序建立回调关系，也就是说，当响应的事件发生时会调用这个回调方法.
• 这个回调方法在内核中叫ep_poll_callback,它会将发生的事件添加到rdlist双链表中.
• 在epoll中，对于每一个事件，都会建立一个epitem结构体.

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

• 当调用epoll_wait检查是否有事件发生时，只需要检查eventpoll对象中的rdlist双链表中是否有epitem元素即可.
• 如果rdlist不为空，则把发生的事件复制到用户态，同时将事件数量返回给用户. 这个操作的时间复杂度是O(1).

总结一下, epoll的使用过程就是三部曲:

• 调用epoll_create创建一个epoll句柄;
• 调用epoll_ctl, 将要监控的文件描述符进行注册;
• 调用epoll_wait, 等待文件描述符就绪;

epoll的优点(对比select)

• 接口使用方便: 虽然拆分成了三个函数, 但是反而使用起来更方便高效. 不需要每次循环都设置关注的文件描述符, 也做到了输入输出参数分离开
• 数据拷贝轻量: 只在合适的时候调用EPOLL_CTL_ADD 将文件描述符结构拷贝到内核中, 这个操作并不频繁(而select/poll都是每次循环都要进行拷贝)
• 事件回调机制: 避免使用遍历, 而是使用回调函数的方式, 将就绪的文件描述符结构加入到就绪队列中, epoll_wait 返回直接访问就绪队列就知道哪些文件描述符就绪. 这个操作时间复杂度O(1). 即使文件描述符数目很多, 效率也不会受到影响.
• 没有数量限制: 文件描述符数目无上限.

epoll工作方式

epoll有2种工作方式-水平触发(LT)和边缘触发(ET)

水平触发Level Triggered 工作模式

• epoll默认状态下就是LT工作模式.
• 当epoll检测到socket上事件就绪的时候, 可以不立刻进行处理. 或者只处理一部分.
• 如果recv每次只读了一部分数据, 缓冲区中还剩余有数据, 在第二次调用 epoll_wait 时, epoll_wait仍然会立刻返回并通知socket读事件就绪.
• 直到缓冲区上所有的数据都被处理完, epoll_wait 才不会立刻返回.
• 支持阻塞读写和非阻塞读写

边缘触发Edge Triggered工作模式

• 如果我们将socket添加到epoll描述符的时候使用了EPOLLET标志, epoll进入ET工作模式.
• 当epoll检测到socket上事件就绪时, 必须立刻处理.
• 虽然每次recv只读了一部分的数据, 缓冲区还剩有数据, 在第二次调用 epoll_wait 的时候, epoll_wait 不会再返回了.
• 也就是说, ET模式下, 文件描述符上的事件就绪后, 只有一次处理机会.
• ET的性能比LT性能更高( epoll_wait 返回的次数少了很多).
• 只支持非阻塞的读写
• select和poll其实也是工作在LT模式下. epoll既可以支持LT, 也可以支持ET.

对比LT和ET

• LT是 epoll 的默认行为. 使用 ET 能够减少 epoll 触发的次数. 但是代价就是强逼着程序猿一次响应就绪过程中就把所有的数据都处理完.
• 相当于一个文件描述符就绪之后, 不会反复被提示就绪, 看起来就比 LT 更高效一些. 但是在 LT 情况下如果也能做到每次就绪的文件描述符都立刻处理, 不让这个就绪被重复提示的话, 其实性能也是一样的.
• 另一方面, ET 的代码复杂程度更高了.

理解ET模式和非阻塞文件描述符

• 使用 ET 模式的 epoll, 需要将文件描述设置为非阻塞. 这个不是接口上的要求, 而是 “工程实践” 上的要求.
• 假设这样的场景: 服务器接受到一个10k的请求, 会向客户端返回一个应答数据. 如果客户端收不到应答, 不会发送第二个10k请求。
• 如果服务端写的代码是阻塞式的read, 并且一次只 read 1k 数据的话(read不能保证一次就把所有的数据都读出来,参考 man 手册的说明, 可能被信号打断), 剩下的9k数据就会待在缓冲区中
• 此时由于 epoll 是ET模式, 并不会认为文件描述符读就绪. epoll_wait 就不会再次返回. 剩下的 9k 数据会一直在缓冲区中. 直到下一次客户端再给服务器写数据. epoll_wait 才能返回
• 服务器只读到1k个数据, 要10k读完才会给客户端返回响应数据.
• 客户端要读到服务器的响应, 才会发送下一个请求
• 客户端发送了下一个请求, epoll_wait 才会返回, 才能去读缓冲区中剩余的数据（有点类似死锁的感觉啦）
• 所以, 为了解决上述问题(阻塞read不一定能一下把完整的请求读完), 于是就可以使用非阻塞轮训的方式来读缓冲区,保证一定能把完整的请求都读出来.
• 而如果是LT没这个问题. 只要缓冲区中的数据没读完, 就能够让 epoll_wait 返回文件描述符读就绪.

epoll的使用场景

epoll的高性能, 是有一定的特定场景的. 如果场景选择的不适宜, epoll的性能可能适得其反.

• 对于多连接, 且多连接中只有一部分连接比较活跃时, 比较适合使用epoll.
• 例如, 典型的一个需要处理上万个客户端的服务器, 例如各种互联网APP的入口服务器, 这样的服务器就很适合epoll.
• 如果只是系统内部, 服务器和服务器之间进行通信, 只有少数的几个连接, 这种情况下用epoll就并不合适. 具体要根据需求和场景特点来决定使用哪种IO模型.

epoll版本的TCP服务器代码

完整代码请前往我的GitHub仓库—>>点我哇

#pragma once
#include"tcp_socket.hpp"
#include<sys/epoll.h>
#include<vector>
#include<iostream>
using namespace std;

#define MAXEVENTS 30

class Epoll{
  public:
    Epoll()
      :epfd_(-1)
  {}
    bool Create(){
      epfd_=epoll_create(1);
      if(epfd_<0)
        return false;
      return true;
    }
    bool Add(const TcpSocket& sock){
      int op=EPOLL_CTL_ADD;
      int fd=sock.GetFd();
      epoll_event ev;
      ev.data.fd=fd;
      ev.events=EPOLLIN|EPOLLET;
      int ret=epoll_ctl(epfd_,op,fd,&ev);
      if(ret<0)
        return false;
      return true;
    }
    bool Del(const TcpSocket& sock){
      int op=EPOLL_CTL_DEL;
      int fd=sock.GetFd();
      int ret=epoll_ctl(epfd_,op,fd,nullptr);
      if(ret<0)
        return false;
      return true;
    }
    bool Wait(vector<TcpSocket>&list,int timeout=-1){
      epoll_event evs[MAXEVENTS];
      int ret=epoll_wait(epfd_,evs,MAXEVENTS,timeout);
      if(ret<0){
        perror("epoll Wait Error ");
        return false;
      }else if(ret==0){
        cout<<"epoll wait timeout"<<endl;
        return false;
      }
      for(int i=0;i<ret;++i){
        TcpSocket sock(evs[i].data.fd);
        list.push_back(sock);
      }
      return true;
    }
    
  private:
    int epfd_;
};


typedef void (*Handler)(const string& buf,string& ret);

class TcpEpollServer{
  public:
    TcpEpollServer()
    {}
    ~TcpEpollServer(){
      _sock.Close();
    }
    bool Start(const string& ip,const uint16_t port,Handler handler){
      if(_sock.Socket()==false)
        return false;
      //_sock.SetNonBlock();//设置非阻塞
      if(_sock.Bind(ip,port)==false)
        return false;
      if(_sock.Listen()==false)
        return false;
      
      Epoll epoll;
      epoll.Create();
      epoll.Add(_sock);

      while(1){
        vector<TcpSocket> list;
        if(epoll.Wait(list)==false)
          continue;
        for(size_t i=0;i<list.size();++i){
          if(list[i].GetFd()==_sock.GetFd()){
            TcpSocket NewSock;
            string ClientIp;
            uint16_t ClientPort;
            if(_sock.Accept(&NewSock,ClientIp,ClientPort)==false)
              continue;
            printf("[%s:%d]客户端已连接！~\n",ClientIp.c_str(),ClientPort);
            //NewSock.SetNonBlock();//设置非阻塞
            epoll.Add(NewSock);
          }
          else{
            string msg;
            int n=list[i].Recv(msg);
            if(n==1){
              cout<<"客户端发送："<<msg<<endl;
              string ret;
              handler(msg,ret);
              list[i].Send(ret);
            }
            else{
              epoll.Del(list[i]);
              list[i].Close();
              continue;
            }
          }
        }
      }
      return true;
    }
  private:
    TcpSocket _sock;
};