详解I/O多路转接之poll&epoll

I/O多路转接之poll

操作流程：

1. 定义监控的描述符事件结构体数组，将需要监控的描述符以及时间标识信息，添加到数组的各个节点中
2. 发起调用开始监控，将描述符事件结构体数组，拷贝到内核中进行轮询遍历判断，若有就绪/等待超时则调用返回，并且在每个描述符对应的事件结构体中，标识当前就绪的事件。
3. 进程轮询遍历，判断数组中的每个节点中的就绪事件是哪个事件，决定是否就绪了以及如何对描述符进行操作

接口认识：

int poll(struct pollfd *arr_fds, nfds_t nfds,int timeout)

• arr_fds：事件结构体数组，填充要监控的描述符以及事件信息
• nfds：数组中的有效节点个数（数组有可能会很大，但是需要监控的节点只有前nfds个）
• timeout：监控的超市等待时间—单位：毫秒

返回值：>1表示就绪的描述符事件个数； 返回值==0–等待超时； <0–表示监控出错

poll监控采用事件结构体的形式：

struct poolfd{
	int fd;//要监控的描述符
	short events;//要监控的事件
	short revents;//调用返回时填充的就绪事件
}

events和revents的取值：

在这里插入图片描述

poll优缺点

poll的优点

• poll使用一个pollfd的指针实现.
• pollfd结构包含了要监视的event和发生的event，不再使用select“参数-值”传递的方式. 接口使用比select更方便.
• poll并没有最大数量限制 (但是数量过大后性能也是会下降).

poll的缺点

• poll中监听的文件描述符数目增多时和select函数一样，poll返回后，需要轮询pollfd来获取就绪的描述符.
• 每次调用poll都需要把大量的pollfd结构从用户态拷贝到内核中.
• 同时连接的大量客户端在一时刻可能只有很少的处于就绪状态, 因此随着监视的描述符数量的增长, 其效率也会线性下降.

使用poll监控标准输入

#include <poll.h>
#include <unistd.h>
#include <stdio.h>
int main(){
	struct pollfd poll_fd;
	poll_fd.fd = 0;  poll_fd.events = POLLIN;
	for (;;) {
		int ret = poll(&poll_fd, 1, 1000);
		if (ret < 0) {
			perror("poll");
			continue;
		}
		if (ret == 0) {
			printf("poll timeout\n");
			continue;
		}
		if (poll_fd.revents == POLLIN) {
			char buf[1024] = {0};
			read(0, buf, sizeof(buf) - 1);
			printf("stdin:%s", buf);
		}
	}
}

在这里插入图片描述

I/O多路转接之epoll

操作流程：

1. 在内核中创建epool句柄epollevent结构体（这个结构体包含很多信息，红黑树+双向链表）
2. 发起调用对内核中的epollevent结构添加/删除/修改所监控的描述符监控信息
3. 发起调用开始监控，在内核中采用异步阻塞操作实现监控，等待超时/有描述符就绪了事件调用返回，返回给用户就绪描述符的事件结构体信息
4. 进程直接对就绪的事件结构体中的描述符成员进行操作即可

接口信息

1.创建epoll句柄

int epoll_create(int size//创建epoll句柄
	//size：在linux2.6.2之后被忽略，只要大于0即可
	//返回值：文件描述符---epoll的操作句柄

2.epoll事件注册

int epoll_ctl(int epfd, int cmd, int fd, 
				struct epoll_event* ev);
//epfd：epoll_create返回的操作句柄
//cmd：针对fd描述符的监控信息要进行的操作-添加/删除/修改  EPOLL_CTL_ADD/EPOLL_CTL_DEL/EPOLL_CTL_MOD
//fd：要监控操作的描述符
//ev：fd描述符对应的事件结构体信息

struct epoll_event{
uint32_t events;//对fd描述符要监控的事件--EPOLLIN/EPOLLOUT
union{int fd; void *ptr;}data;//要填充的描述符信息
}

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

• EPOLLIN : 表示对应的文件描述符可以读 (包括对端SOCKET正常关闭);
• EPOLLOUT : 表示对应的文件描述符可以写;
• EPOLLPRI : 表示对应的文件描述符有紧急的数据可读 (这里应该表示有带外数据到来);
• EPOLLERR : 表示对应的文件描述符发生错误;
• EPOLLHUP : 表示对应的文件描述符被挂断;
• EPOLLET : 将EPOLL设为边缘触发(Edge Triggered)模式, 这是相对于水平触发(Level Triggered)来说的.
• EPOLLONESHOT：只监听一次事件, 当监听完这次事件之后, 如果还需要继续监听这个socket的话, 需要 再次把这个socket加入到EPOLL队列里.

一旦epoll开始监控，描述符若就绪了进行关心的事件，则就会给用户返回我们所添加的对应时间结构体信息，通过时间结构体信息中所包含的描述符进行操作—因此第三个参数与结构体中的fd描述符通常是同一个描述符

3.开始监控

int epoll_wait(int epfd, struct epoll_event* evs,
				int max_event, int timeout);
//epfd：epoll操作句柄
//evs：struct epoll_event结构体数组的首地址，用于接收就绪描述符对应的时间结构体信息
//max_event：本次监控想要获取的就绪事件的最大数量，不大于evs数组的最大节点个数，防止访问越界
//timeout：超市等待时间单位：毫秒

返回值：>0–就绪的事件个数 ==0–等待超时 <0–监控出错

epoll的监控原理：异步阻塞操作 监控由系统完成，用户添加的描述符以及对应事件结构体会被添加到内核的eventpoll结构体中的红黑树中 一旦发起调用开始监控，则操作系统为每个操作符的事件做了一个回调函数，功能室当描述符就绪了关心的事件，则将描述符对应的事件结构体添加到双向链表中 进程自身，只是每隔一段时间，判断双向链表是否为NULL，决定是否有就绪 4.进程遍历获取evs中就绪的事件结构体信息，针对其中的events就绪时间对data.fd进行相应操作

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工作方式

水平触发Level Triggered 工作模式

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

边缘触发Edge Triggered工作模式

如果我们在第1步将socket添加到epoll描述符的时候使用了EPOLLET标志, epoll进入ET工作模式.

• 当epoll检测到socket上事件就绪时, 必须立刻处理.
• 如上面的例子, 虽然只读了1K的数据, 缓冲区还剩1K的数据, 在第二次调用 epoll_wait 的时候,epoll_wait 不会再返回了.
• 也就是说, ET模式下, 文件描述符上的事件就绪后, 只有一次处理机会.
• ET的性能比LT性能更高( epoll_wait 返回的次数少了很多). Nginx默认采用ET模式使用epoll.
• 只支持非阻塞的读写

epoll的使用场景

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

对于多连接, 且多连接中只有一部分连接比较活跃时, 比较适合使用epoll.

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

封装一个epoll类

#include<cstdio>
#include"tcpsock.hpp"
#include<vector>
#include<sys/epoll.h>
#define MAX_TIMEOUT 3000
class Epoll
{
public:
    Epoll():_epfd(-1){
        _epfd=epoll_create(1);
        if(_epfd<0){
            perror("create epoll error\n");
            exit(-1);
        }
    }
    bool Add(TcpSocket &sock){
        //2.添加描述符监控事件信息
        //获取描述符
        int fd=sock.GetFd();
        //定义描述符对应的事件结构体
        struct epoll_event ev;
        ev.events=EPOLLIN;
        ev.data.fd=fd;
        //添加到内核中
        int ret=epoll_ctl(_epfd,EPOLL_CTL_ADD,fd,&ev);
        if(ret<0){
            perror("epoll ctl error\n");
            return false;
        }
        return true;
    }
    bool Del(TcpSocket &sock){
        int fd=sock.GetFd();
        int ret=epoll_ctl(_epfd,EPOLL_CTL_DEL,fd,NULL);
        if(ret<0){
            perror("epoll ctl error\n");
            return false;
        }
        return true;
    }
    bool Wait(std::vector<TcpSocket> *list,int timeout=MAX_TIMEOUT){
        struct epoll_event evs[10];
        int nfds=epoll_wait(_epfd,evs,10,timeout);
        if(nfds<0){
            perror("epoll wait error\n");
            return false;
        }else if(nfds==0){
            printf("epoll wait timeout\n");
            list->clear();
            return true;
        }
        for(int i=0;i<nfds;i++){
            if(evs[i].events&EPOLLIN){
                TcpSocket sock;
                sock.SetFd(evs[i].data.fd);
                list->push_back(sock);
            }
        }
        return true;
    }
private:
    int _epfd;
};