【Linux】多路转接 -- epoll-Toy模板网

这篇具有很好参考价值的文章主要介绍了【Linux】多路转接 -- epoll。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

1. 认识epoll

epoll系统调用和select以及poll是一样的，都是可以让我们的程序同时监视多个文件描述符上的事件是否就绪。

epoll在命名上比poll多了一个poll，这个e可以理解为extend，epoll就是为了同时处理大量文件描述符而改进的poll。

epoll在2.5.44内核中被引进，它几乎具备了select和poll的所有优点，被公认为Linux2.6下性能最好的多路IO就绪通知方法。

2. epoll相关系统调用接口

epoll有三个相关系统调用接口，分别是epoll_create，epoll_ctl 和 epoll_wait。

epoll_create

epoll_create函数的作用就是创建一个epoll的文件描述符。

【Linux】多路转接 -- epoll,Linux,计算机网络,linux
参数说明：

size：自从Linux2.6.8之后，size参数是被忽略的，但size的值必须设置为大于0的值。

返回值说明：

epoll模型创建成功返回其对应的文件描述符，否则返回-1，同时错误码会被设置。

注意：当不再使用时，必须调用close函数关闭epoll模型对应的文件描述符，当所有引用epoll实例的文件描述符都已经关闭时，内核将销毁该实例并释放相关资源。

epoll_ctl

epoll_ctl 函数用于向指定的epoll模型中注册事件，它不同于seletct()的一点就是，select在监听事件时告诉内核要监听什么类型的事件，而它是先注册要监听的事件类型。

【Linux】多路转接 -- epoll,Linux,计算机网络,linux

参数说明：

epfd：epoll_create的返回值
op：表示具体的动作，用三个宏来表示
fd：需要监视的文件描述符
event：告诉内核需要监听什么事件

第二个参数op的取值有以下三种：

EPOLL_CTL_ADD：注册新的fd到epfd中。
EPOLL_CTL_MOD：修改已经注册的fd的监听事件。
EPOLL_CTL_DEL：从epfd中删除指定的文件描述符。

返回值说明：

函数调用成功返回0，调用失败返回-1，同时错误码会被设置。

第四个参数struct epoll_event 结构如下：
【Linux】多路转接 -- epoll,Linux,计算机网络,linux
struct epoll_event结构当中有两个成员，第一个成员events表示的是需要监听的事件，第二个成员data是一个联合体结构，一般选择使用该结构当中的fd，表示需要监听的文件描述符。

events常用取值如下：

EPOLLIN：表示对应的文件描述符可以读（包括对端SOCKET正常关闭）
EPOLLOUT：表示对应的文件描述符可以写
EPOLLPRI：表示对应的文件描述符有紧急的数据可读（这里应该表示有带外数据到来）
EPOLLERR：表示对应的文件描述符发送错误
EPOLLHUP：表示对应的文件描述符被挂断，即对端文件描述符关闭
EPOLLET：将epoll的工作方式设置为边缘触发模式。
EPOLLONESHOT：只监听一次事件，当监听完这次事件之后，如果还需要继续监听该文件描述符的话，需要重新将该文件描述符添加到EPOLL队列中。

这些取值也是以宏的方式进行定义的，它们的二进制序列当中有且只有一个比特位是1，且为1的比特位是各不相同的。

epoll_wait

epoll_wait 函数用于收集监视的事件中已经就绪的事件
【Linux】多路转接 -- epoll,Linux,计算机网络,linux
参数说明：

epfd：指定的epoll模型，epoll_create的返回值
events：epoll会把发送的事件赋值到events数组中（events不可以是空指针，内核只负责把数组复制到这个events数组中，不会帮助我们在用户态中分配内存）。
maxevents：events数组中的元素个数，该值不能大于创建epoll模型使传入的size值。
timeout：表示epoll_wait函数的超时时间，单位是毫秒（ms）。

参数timeout的取值：

-1：epoll_wait调用后进行阻塞等待，直到被监视的某个文件描述符的某个事件就绪。
0：epoll_wait调用后进行非阻塞等待，无论被监视的文件描述符上的事件是否就绪，epoll_wait检测后都会立刻返回。
特定的时间值：epoll_wait 调用后在直到的时间内进行阻塞等待，如果监视的文件描述符上一直没有事件就绪，则在该时间后epoll_wait进行超时返回。

返回值说明：

如果函数调用成功，则返回有事件就绪的文件描述符个数。
如果timeout时间耗尽，则返回0。
如果函数调用失败，则返回-1，同时错误码会被设置。

epoll_wait 调用失败时，错误码可能被设置为：

EBADF：传入的epoll模型对应的文件描述符无效。
EFAULT：events指向的数组空间无法通过写入权限访问。
EINTR：此调用被信号所中断。
EINVAL：epfd不是一个epoll模型对应的文件描述符，或传入的maxevents小于等于0。

3. epoll工作原理

红黑树和就绪队列

【Linux】多路转接 -- epoll,Linux,计算机网络,linux
当某一进程调用epoll_create函数时，Linux内核会创建一个eventpoll结构体，也就是我们所说的epoll模型，eventpoll结构体当中的成员rbr和成员rdlist与epoll的使用方式密切相关。

epoll模型当中的红黑树本质就是告诉内核，需要监视哪些文件描述符上的哪些事件，调用epoll_ctl 函数实际就是在对这颗红黑树进行对应的增删查改操作。
epoll模型当中的就绪队列本质就是告诉内核，哪些文件描述符上的哪些事件已经就绪了，调用epoll_wait 函数实际就是在从就绪队列当中获取已经就绪的事件。

注意：

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

【Linux】多路转接 -- epoll,Linux,计算机网络,linux

对于epitem结构当中的rbn成员来说，ffd与event的含义是，需要监视的ffd上的event事件是否就绪。
对于epitem结构当中的rdllink成员来说，ffd与event的含义是，ffd上的event事件已经就绪了。
当调用epoll_wait检查是否有事件发生时，只需要检查eventpoll对象中的rdlist双链表中是否含有epitem元素即可。
如果rdlist不为空，则把发送的事件复制到用户态，同时将事件数量返回给用户，这个操作的时间复杂度为O(1)。

说明一下：

红黑树是一种二叉搜索树，因此必须有键值key，而这里的文件描述符就可以天然地作为红黑树的key值。
调用epoll_ctl向红黑树当中新增节点时，如果设置了EPOLLONESHEOT选项，当监听完这次事件之后，如果还要继续监听该文件描述符则需要重新将其添加到epoll模型中，本质就是当设置了EPOLLONESHOT的事件就绪时，操作系统会自动将其从红黑树当中删除。
而如果调用epoll_ctl向红黑树当中新增节点时没有设置EPOLLONSHOT，那么该节点插入红黑树之后就会一直存在，除非用户调用epoll_ctl将该节点从红黑树当中删除。

回调机制

所有添加到红黑树当中的事件，都会与设备（网卡）驱动程序建立回调方法，这个回调方法在内核中叫做ep_poll_callback。

对于select和poll来说，操作系统在监视多个文件描述符上的事件是否就绪时，需要让操作系统主动对这多个文件描述符进行轮询检测，这一定会增加操作系统的负担。
而对于epoll来说，操作系统不需要主动进行事件的检测，当红黑树中监视的事件就绪时，会自动调用对应的回调方法，将就绪的事件添加到就绪队列当中。
当用户调用epoll_wait函数获取就绪事件时，只需要关注底层就绪队列是否为空，如果不为空则将就绪队列当中的就绪事件拷贝给用户即可。

采用回调机制最大的好处，就是不再需要操作系统主动对就绪事件进行检测了，当时间就绪时会自动调用对应的回调函数进行处理。

说明一下：

只有添加到红黑树当中的事件才会与底层建立回调方法，因此只有当红黑树当中对应的事件就绪时，才会执行对应的回调方法将其添加到就绪队列当中。
当不断有监视的事件就绪时，会不断有回调方法向就绪队列当中插入节点，而上层也会不断调用epoll_wait函数从就绪队列中获取节点，这也是典型的生产者消费者模型。
由于就绪队列可能会被多个执行流同时访问，因此必须要使用互斥锁对其进行保护，eventpoll结构当中的lock和mtx就是保护临界资源的，因此epoll本身是线程安全的。
eventpoll结构当中的wa（wait queue）就是等待队列，当多个执行流想要同时访问同一个epoll模型时，就需要在该等待队列下进行等待。

epoll三部曲

调用epoll_create，创建一个epoll模型
调用epoll_ctl，将要监视的文件描述符进行注册
调用epoll_wait，等待文件描述符就绪

4. epoll服务器

为了简单演示一下epoll的使用方式，这里我们实现一个简单的epoll服务器，该服务器是获取客户端发来的数据并进行打印。

EpollServer类

EpollServer类中除了包含监听套接字和端口号两个成员变量之外，最好将epoll模型对应的文件描述符也作为一个成员变量。

在构造EpollServer对象时，需要指明epoll服务器的端口号，当然也可以在初始化epoll服务器的时候指明。
在初始化epoll服务器的时候调用Socket类中的函数（该Socket类中封装了进行TCP传输的方法），一次进行套接字的创建、绑定和监听、此外epoll模型的创建可以在服务器初始化的时候进行。
在析构函数中调用close函数，将监听套接字和epoll模型对应的文件描述符进行关闭。

#include "Socket.hpp"
#include <sys/epoll.h>

#define BACK_LOG 5
#define SIZE 256

class EpollServer
{
public:
    EpollServer(int port)
        : _port(port)
    {}

    void InitEpollServer()
    {
        _listen_sock = Socket::SocketCreate();
        Socket::SocketBind(_listen_sock, _port);
        Socket::SocketListen(_listen_sock, BACK_LOG);

        // 创建epoll模型
        _epfd = epoll_create(SIZE);
        if (_epfd < 0)
        {
            std::cerr << "epoll_create error" << std::endl;
            exit(5);
        }
    }

    ~EpollServer()
    {
        if (_listen_sock > 0) close(_listen_sock);
        if (_epfd) close(_epfd);
    }

private:
    int _listen_sock; // 监听套接字
    int _port; // 服务器端口号
    int _epfd; // epollfd
};

运行服务器

服务器初始化完毕之后就可以开始运行了，而epoll服务器要做的就是不断调用epoll_wait函数，从就绪队列中获取就绪事件进行处理即可。

首先，在epoll服务器开始死循环调用epoll_wait之前，需要先调用epoll_ctl将监听套接字添加到epoll模型中，表示服务器开始运行时只需要监视监听套接字的读事件。
此后，epoll服务器就不断调用epoll_wait函数监视读事件是否就绪。如果epoll_wait函数的返回值大于0，则说明已经有文件描述符的读事件就绪，并且此事的返回值代表的就是有事件就绪的文件描述符的个数，接下来就应该对就绪事件进行处理。
如果epoll_wait的函数返回值等于0，则说明timeout时间耗尽，此事直接准备下一次epoll_wait调用即可。
如果epoll_wait函数返回值为-1，此时也让服务器进行下一次epoll_wait调用，但是实际应该进一步判断错误码，根据错误码来判断是否应该继续调用epoll_wait函数。

    void HandlerEvent(struct epoll_event revs[], int num)
    {
        for (int i = 0; i < num; ++i)
        {
            int fd = revs[i].data.fd; // 就绪的文件描述符
            if (fd == _listen_sock && revs[i].events & EPOLLIN)
            {
                // 连接事件就绪
                struct sockaddr_in peer;
                memset(&peer, 0, sizeof(peer));
                socklen_t len = sizeof(peer);
                int sock = accept(_listen_sock, (struct sockaddr*)&peer, &len);
                if (sock < 0)
                {
                    std::cerr << "accept error" << std::endl;
                    continue;
                }
                std::string peer_ip = inet_ntoa(peer.sin_addr);
                int peer_port = ntohs(peer.sin_port);
                std::cout << "get a new link[" << peer_ip << ":" << peer_port << "]" << std::endl;
                
                // 将获取到的文件描述符添加到sock中，并关心其读事件
                AddEvent(sock, EPOLLIN);  
            }
            else if (revs[i].events & EPOLLIN)
            {
                char buffer[1024];
                ssize_t size = recv(fd, buffer, sizeof(buffer) - 1, 0);
                if (size > 0)
                {
                    buffer[size] = 0;
                    std::cout << "echo# " << buffer << std::endl;
                }
                else if (size == 0)
                {
                    std::cout << "client quit" << std::endl;
                    close(fd);
                    DelEvent(fd); // 将fd从epoll中删除
                }
                else
                {
                    std::cerr << "recv error" << std::endl;
                    close(fd);
                    DelEvent(fd);
                }
            }
        }
    }


private:
    void DelEvent(int sock)
    {
        epoll_ctl(_epfd, EPOLL_CTL_DEL, sock, nullptr);
    }

epoll服务器测试

#include "EpollServer.hpp"
#include <string>

int main(int argc, char* argv[])
{
    if (argc != 2)
    {
        std::cout << "Usage: " << "./EpollServer port" << std::endl;
        exit(1);
    }

    int port = atoi(argv[1]);

    EpollServer* svr = new EpollServer(port);
    svr->InitEpollServer();
    svr->Run();

    return 0;
}

因为我们在调用epoll_wait函数时，将timeout的值设为了-1，因此服务器运行之后如果没有读事件就绪，那么就会阻塞等待。
【Linux】多路转接 -- epoll,Linux,计算机网络,linux

5. epoll的优点

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

注意：

有人说epoll中使用了内存映射机制，内核直接将底层就绪队列通过mmap的方式映射到用户态，此时用户就可以直接读取到内核中就绪队列当中的数据，避免了内存拷贝的额外性能开销。
这种说法是错误的，实际操作系统并没有做任何映射机制，因为操作系统是不相信任何人的，操作系统不会让用户进程直接访问到内核的数据的，用户只能通过系统调用来获取内核的数据。
因此用户要获取内核当中的数据，势必还是需要将内核的数据拷贝到用户空间。

6. epoll的工作方式

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

水平触发（LT， Level Triggered）

只要底层有事件就绪，epoll就会一直通知用户。

epoll默认状态下就是LT工作模式：

由于在LT工作模式下，只要底层有事件就绪就会一直通知用户，因此当epoll检测到底层读事件就绪时，可以不立即进行处理，或者只处理一部分，因为只要底层数据没有处理完，下一次epoll还会通知用户事件就绪。
select和poll的模式其实就是LT模型
支持阻塞读写和非阻塞读写。

边缘触发（ET， Edge Triggered）

只有底层就绪事件数量由无到有或者由有到多的时候，epoll才会通知用户。

如果要将epoll改为ET工作模式，则需要在添加时间时设置EPOLLET选项。

由于在ET工作模式下，只有底层就绪事件由无到有或者由有到多的时候才会通知用户，所以当epoll检测到底层读事件就绪的时候，必须立即进行处理，而且必须全部处理完毕，因为有可能此后底层再也没有事件就绪，那么epoll就再也不好通知用户进行事件处理。
ET工作模式下epoll通知用户的次数比LT少，因此ET的性能一般比LT性能跟高，Nginx就是默认采用ET模式使用epoll的。
只支持非阻塞的读写。

ET工作模式下应该如何进行读写？

因为在ET工作模式下，只有底层就绪事件由无到有或者由有到多时才会通知用户，这就倒逼用户当读事件就绪时必须一次性将数据全部读取完毕，当写事件就绪时就必须一次性将发送缓冲区写满，否则可能再也没有机会进行读写了。

因此读数据时必须循环调用recv函数进行读取，写数据时必须循环调用send函数进行写入。

当底层读事件就绪时，循环调用recv函数进行读取，直到某次调用recv函数时，实际读取到的字节数小于期望读取的字节数，则说明本次底层数据已经读取完毕了。
但有可能最后一次调用recv读取时，刚好实际读取的字节数和期望读取的字节数相等，但此时底层数据恰好读取完毕，如果我们再调用recv函数进行读取，那么recv就会因为没有数据而被阻塞住。
而这里的阻塞是非常严重的，就比如我们这里写的服务器都是单进程的服务器，如果recv被阻塞住，并且此后该数据再也不就绪，那么就相当于我们的服务器挂掉了，因此在ET模式下循环调用recv函数进行读取时，必须将文件描述符设置为非阻塞状态。
调用send函数写数据时也是同样的道理，需要循环调用send函数进行数据的写入，并且必须将对应的文件描述符设置为非阻塞状态。

注意：ET工作模式下，recv和send操作的文件描述符必须设置为非阻塞状态，这是必须的！