【Linux 36】多路转接 - epoll

🌈 一、epoll 初步认识

• epoll 系统调用可以让程序同时监视多个文件描述符上的事件是否就绪（多路转接接口都能），与 select 和 poll 的使用场景相同。
• epoll 在命名上比 poll 多了一个 e（extend），它是为了同时处理大量的文件描述符而产生的 poll 的 plus 版本。
• epoll 在 Linux 2.5.44 被引入，几乎具备了 select 和 poll 的所有优点，被公认为 Linux 2.6 下性能最好的多路 IO 就绪通知方法。

举个例子

• epoll 不仅能够告知用户，被监视的文件描述符上有事件就绪。还能够告诉用户是哪些文件描述符上的事件就绪了。
• 这样就避免了像 select 和 poll 那样的需要遍历文件描述符集，才能知道是哪些文件描述符上的事件就绪的情况。

🌈 二、epoll 相关接口

• epoll 有 3 个相关的系统调用接口，分别是 epoll_create、``epoll_ctl和epoll_wait`。

⭐ 1. 创建 epoll – epoll_create

#include <sys/epoll.h>

int epoll_create(int size);

• 函数功能：在内核中创建一个 epoll 模型，也能理解为创建一个 epoll 句柄。
• 函数参数：自 Linux 2.6.8 后，size 参数通常会被忽略，但 size 的值必须得 > 0
• 函数返回值：创建 epoll 模型成功时，返回对应的文件描述符；否则返回 -1，并设置错误码。
• 注意事项：当不再使用时，必须调用 close 函数关闭 epoll 模型所对应的文件描述符。

⭐ 2. 控制 epoll – epoll_ctr

• 该函数用于向指定的 epoll 模型中注册事件，就是用户告诉内核该干什么。

#include <sys/epoll.h>

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

1. 参数说明

• epfd：想要操纵的 epoll 模型（epoll_create 的返回值）。
• op：对 epoll 模型具体要执行的操作，用 3 个宏来表示，分别如下：
  • EPOLL_CTL_ADD：注册新的文件描述符 fd 到指定的 epoll 模型中。
  • EPOLL_CTL_MOD：修改已经注册的文件描述符 fd 的监听事件。
  • EPOLL_CTL_DEL：从 epoll 模型中删除指定的文件描述符 fd（想从 epoll 中移除 fd 时，这个 fd 必须是健康 & 合法的，否则会移除出错。如果想关闭 fd，则必须先将该 fd 从 epoll 模型中移除，再执行 close）。
• fd：需要监视的文件描述符。
• event：需要监视 fd 这个文件描述符上的哪些事件。

2. 返回值说明

• 调用成功时：返回 0
• 调用失败时：返回 -1，并设置错误码

3. struct epoll_event 的结构

• 在 struct epoll_event 中有两个成员：
  1. events：表示需要监视的事件
  2. data：联合体结构，一般使用该结构中的 fd 字段（四选一），表示需要监听的文件描述符。
• events 字段的常用取值如下：

⭐ 3. 等待 epoll – epoll_wait

• epoll_wait 函数用于收集所监视的事件中所有已经就绪的事件，就是内核告知用户，哪些事件就绪了。

#include <sys/epoll.h>

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

1. 参数说明

• epfd：指定的 epoll 模型。
• events：内核会将已经就绪的事件拷贝到 events 数组当中。events 不能是空指针，内核只负责将就绪事件拷贝到该数组中，不会帮我们在用户态中分配内存。
• maxevents：events 数组中的元素个数，该值不能大于在创建 epoll 模型时所传入的 size 参数的值。
• timeout：表示 epoll_wait 函数的超时时间（单位 ms）。

2. 参数 timeout 的取值

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

3. 返回值说明

• 调用函数成功：返回已经就绪了事件的文件描述符的个数。
• 调用函数失败：返回 -1，并设置错误码；错误码的取值如下：
  • EBADF：传入的 epoll 模型对应的文件描述符无效。
  • EFAULT：events 指向的数组空间无法通过写入权限访问。
  • EINTR：此调用被信号所中断。
  • EINVAL：epfd 不是一个 epoll 模型对应的文件描述符，或传入的maxevents 值 <= 0。
• timeout 时间耗尽：返回 0

🌈 三、epoll 工作原理

⭐ 1. 红黑树和就绪队列

• 当某个进程调用了 epoll_create 函数时，Linux 内核会创建一个 eventpoll 结构体，该结构体中有两个成员和 epoll 的使用方式相关。

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

• 每一个 epoll 对象都有一个独立的 eventpoll 结构体，用于存放通过 epoll_ctl 函数 epoll 对象中添加进来的事件。
• 这些事件都会挂载在红黑树中，如此，重复添加的事件就可以通过红黑树而高效的识别出来(红黑树的插入时间效率是 lgn，其中 n 为树的高度)。
• 而所有添加到 epoll 中的事件都会与设备(网卡)驱动程序建立回调关系，也就是说，当响应的事件发生时会调用这个回调方法。
• 这个回调方法在内核中叫 ep_poll_callback,它会将发生的事件添加到 rdlist 双链表中。
• 在 epoll 中，对于每一个事件，都会建立一个 epitem 结构体。红黑树和就绪队列当中的节点分别是基于 epitem 结构中的 rbn 成员和 rdllink 成员的，epitem 结构当中的成员 ffd 记录的是指定的文件描述符值，event 成员记录的就是该文件描述符对应的事件。

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)。

⭐ 2. 回调机制

• 所有添加到红黑树当中的事件，都会与设备（网卡）驱动程序建立回调方法，这个回调方法在内核中叫 ep_poll_callback。
• 对于 select 和 poll 来说，操作系统在监视多个文件描述符上的事件是否就绪时，需要让操作系统主动对这多个文件描述符进行轮询检测，这一定会增加操作系统的负担。
• 而对于 epoll 来说，操作系统不需要主动进行事件的检测，当红黑树中监视的事件就绪时，会自动调用对应的回调方法，将就绪的事件添加到就绪队列当中。
• 当用户调用 epoll_wait 函数获取就绪事件时，只需要关注底层就绪队列是否为空，如果不为空则将就绪队列当中的就绪事件拷贝给用户即可。
• 采用回调机制最大的好处，就是不再需要操作系统主动对就绪事件进行检测了，当事件就绪时会自动调用对应的回调函数进行处理。

⭐ 3. epoll 的使用过程

1. 调用 epoll_create 函数创建一个 epoll 模型。
2. 调用 epoll_ctl 函数告知内核要监视哪些文件描述符。
3. 调用 epoll_wait 获取已经有事件就绪的文件描述符。

🌈 四、epoll 工作模式

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

⭐ 1. 水平触发 LT

• 水平触发（LT，Level Triggered）：只要底层有事件就绪，就会一直通知用户。
  • 例：就像数电中的高电平触发，只要一直处于高电平，就会一直触发。

LT 是 epoll 的默认工作模式

• select、poll、epoll 默认采用的都是 LT 工作模式。
• 在 LT 工作模式中，只要底层有事件就绪就会一直通知用户。因此，当 epoll 检测到底层有读事件就绪时，可以不用立刻进行处理 / 只处理一部分。
• LT 工作模式支持阻塞读写和非阻塞读写。

⭐ 2. 边缘触发 ET

• 边缘触发（ET，Edge Triggered）：只有底层就绪事件的数量 【从无到有】 或 【从有到多】时，epoll 才会通知用户。
  • 例：就像数电中的上升沿触发一样，只有当电平由低到高的那一瞬间才会触发。

1. 如何将 epoll 设置成 ET 工作模式

• 如果想将默认为 LT 工作模式的 epoll 修改为 ET 工作模式，需要在调用 epoll_ctl 函数时，为 struct epoll_event 参数中的 events 字段设置 EPOLLET 选项。

event.data.fd = fd;
event.events |= EPOLLET;
epoll_ctl(epoll_fd, EPOLL_CTL_ADD, fd, &event);

2. ET 模式下必须立即处理就绪事件

• 在 ET 模式下，当 epoll 检测到地城有事件就绪时，必须立即进行处理，且必须全部处理完毕。
• 如果不一次性处理完所有事件，如果之后没有新的事件就绪，epoll 就不会再通知用户去处理，此时这些还没处理完的数据就丢失了。
• ET 模型下，epoll 通知用户的次数一般要比 LT 少，因此 ET 的性能一般是比 LT 高的。

3. ET 模式只支持非阻塞的读写

• 由于用户在 ET 工作模式下必须一次性处理完所有的数据，因此，在读 / 写数据时必须循环调用 recv / send 函数。具体的读写方式如下：
  • 当底层读事件就绪时，循环调用 recv 函数进行读取，直到某次调用 recv 读取时，实际读取到的字节数 < 期望读取到的字节数，说明本次底层数据已经读取完毕了。
  • 但有可能在最后一次调用 recv 读取时，实际读取的字节数刚好 = 期望读取的字节数，但此时底层数据也恰好读取完毕了，如果再调用 recv 函数进行读取，recv 就会因为底层没有数据而阻塞住。
  • 如果写的是单进程的服务器，如果 recv 被阻塞住，并且此后该数据再也不就绪，就相当于服务器挂掉了。因此，在 ET 工作模式下循环调用 recv 函数进行读取时，必须将对应的文件描述符设置为非阻塞状态。
  • 调用 send 函数写数据时也是同样的道理，需要循环调用 send 函数进行数据的写入，并且必须将对应的文件描述符设置为非阻塞状态。
• 由于 ET 工作模式只支持非阻塞的读写，因此 recv 和 send 所操作的文件描述符必须被设置为非阻塞状态。

⭐ 3. 对比 LT 和 ET

1. 举例说明 LT 和 ET 的区别

• LT 模式比较 “宽容”，会持续提醒你文件描述符上的情况，直到处理完成。
• ET 模式更像是 “一次性通知”，只会在状态变化时提醒你，你需要确保自己处理好这个状态变化，否则可能会错过某些数据，因为它不会持续提醒。

2. 对比 LT 和 ET

• 在 ET 模式下，一个文件描述符就绪之后，用户不会反复收到通知，看起来比 LT 更高效，但如果在 LT 模式下能够做到每次都将就绪的文件描述符立即全部处理，不让操作系统反复通知用户的话，其实 LT 和 ET 的性能也是一样的。
• ET 模式下可以避免某些情况下的无效通知，因此在实际使用中，ET 模式更常用，尤其是在高并发场景下。
• ET 的代码复杂度（编程难度）比 LT 更高。
• 由于 ET 模式强制用户层必须尽快 & 一次性将数据读取完，在 TCP 网络层面看，就能够给对端主机通告一个更大的接收窗口，从而提升 IO 效率。
• ET 和 LT 的根本区别：ET 和 LT 都可以设置成非阻塞的循环读写数据，但是 ET 有强制性要求，倒逼着程序员必须非阻塞的循环读取数据，从而保证能够提高 ET 的通知效率，以及 IO 效率。

3. 如何选择 LT 还是 ET

• 选择 ET 的场景：在某些需要更高的通知效率或 IO 效率的场景可以选择 ET。
• 选择 LT 的场景：在报文过长时，可能会被拆分成多个小报文发送。LT 可以更好的保证在这种情况下收到报文的有序性。

🌈 五、epoll 的特点

⭐ 1. epoll 的优点

• 接口使用方便：虽然拆分成了三个函数，但是反而使用起来更方便高效。不需要每次循环都设置要关注的文件描述符，也做到了将输入输出参数分离。
• 数据拷贝轻量：只在新增监视事件的时候调用 epoll_ctl 将数据从用户拷贝到内核，而 select 和 poll 每次都需要重新将需要监视的事件从用户拷贝到内核。此外，调用 epoll_wait 获取就绪事件时，只会拷贝就绪的事件，不会进行不必要的拷贝操作。
• 事件回调机制：避免操作系统主动轮询检测事件就绪，而是采用回调函数的方式，将就绪的文件描述符结构加入到就绪队列中。调用 epoll_wait 时直接访问就绪队列就知道哪些文件描述符已经就绪，检测是否有文件描述符就绪的时间复杂度是 O(1)，因为本质只需要判断就绪队列是否为空即可。
• 没有数量限制：可监视的文件描述符的数量没有上限。只要内存足够，就能一直往红黑树中添加结点。

⭐ 2. 与 select 和 poll 的区别

• 在使用 select 和 poll 时，都需要借助数组来维护文件描述符以及需要监视的事件.这个数组是由用户自己维护的，对该数组的增删改操作都需要用户自己来进行。
• 而使用 epoll 时，不需要用户自己维护这个数组，epoll 底层的红黑树就充当了这个第三方数组的功能。并且该红黑树的增删改操作都是由内核维护的，用户只需要调用 epoll_ctl 让内核对该红黑树进行对应的操作即可。
• 在使用多路转接接口时，数据流都有两个方向，分别是用户告知内核和内核告知用户。select 和 poll 将这两件事情都交给了同一个函数来完成，而 epoll 在接口层面上就将这两件事进行了分离。epoll 通过调用 epoll_ctl 完成用户告知内核，通过调用 epoll_wait 完成内核告知用户。

🌈 六、epoll 使用示例

#pragma once

#include <iostream>
#include <sys/epoll.h>

#include "socket.h"

using std::make_unique;
using std::unique_ptr;

using namespace socket_ns;

class epoll_server
{
    const static int size = 1024;
    const static int maxevents = 1024; // 就绪事件的个数

private:
    uint16_t _port;
    unique_ptr<Socket> _listen_socket;
    int _epfd;
    struct epoll_event ready_events[maxevents]; // 存储所有已经就绪的事件

public:
    epoll_server(uint16_t port)
        : _port(port), _listen_socket(make_unique<tcp_socket>())
    {
        _listen_socket->build_listen_socket(_port);

        // 创建 epoll 实例，由于整个服务器都是基于 epoll 运行的，如果 epoll 创建失败那就不要继续
        _epfd = ::epoll_create(size);
        if (_epfd < 0)
        {
            LOG(ERROR, "epoll_create error");
            exit(1);
        }
        LOG(INFO, "epoll_create success, epfd: %d", _epfd);
    }

    void init_server()
    {
        struct epoll_event ev;
        ev.events = EPOLLIN;                   // 让 epoll 监听读事件
        ev.data.fd = _listen_socket->sockfd(); // 关心 listen 套接字上的事件

        // 将 listen 套接字添加到 epoll 模型中
        int n = ::epoll_ctl(_epfd, EPOLL_CTL_ADD, _listen_socket->sockfd(), &ev);
        if (n < 0)
        {
            LOG(ERROR, "epoll_ctl error");
            exit(2);
        }
        LOG(INFO, "epoll_ctl success, epfd: %d, sockfd: %d", _epfd, _listen_socket->sockfd());
    }

    void loop()
    {
        int timeout = 1000;

        while (true)
        {
            // 等待事件就绪（获取就绪事件）
            int n = ::epoll_wait(_epfd, ready_events, maxevents, timeout);
            switch (n)
            {
            case 0:
                LOG(INFO, "timeout");
                break;
            case -1:
                LOG(ERROR, "epoll_wait error");
                break;
            default:
                LOG(INFO, "epoll_wait success, haved event happend, ready event num: %d", n);
                handle_event(n);
                break;
            }
        }
    }

    // 将事件转换为字符串
    string events_to_string(uint16_t events)
    {
        string events_str;
        if (events & EPOLLIN)
            events_str += "EPOLLIN ";
        if (events & EPOLLOUT)
            events_str += "EPOLLOUT ";
        if (events & EPOLLPRI)
            events_str += "EPOLLPRI ";
        if (events & EPOLLERR)
            events_str += "EPOLLERR ";
        if (events & EPOLLHUP)
            events_str += "EPOLLHUP ";
        return events_str;
    }

    // 处理连接事件
    void accepter_connection()
    {
        Inet_addr addr;
        int sockfd = _listen_socket->accepter(&addr);
        if (sockfd > 0)
        {
            LOG(INFO, "get a new link success, client info: %s:%d", addr.ip().c_str(), addr.port());

            // 将新的 fd 交给 epoll 管理
            struct epoll_event ev;
            ev.events = EPOLLIN;
            ev.data.fd = sockfd;
            int n = ::epoll_ctl(_epfd, EPOLL_CTL_ADD, sockfd, &ev);
            if (n < 0)
            {
                LOG(ERROR, "epoll_ctl error");
                ::close(sockfd);
            }
            else
            {
                LOG(INFO, "epoll_ctl success, epfd: %d, sockfd: %d", _epfd, sockfd);
            }
        }
        else if (sockfd < 0)
        {
            LOG(ERROR, "accepter error");
            return;
        }
    }

    // 处理普通的读事件
    void handle_normal_read_event(int i)
    {
        int fd = ready_events[i].data.fd;
        char buffer[1024];
        ssize_t n = ::recv(fd, buffer, sizeof(buffer) - 1, 0);

        if (n > 0)
        {
            buffer[n] = '\0';
            cout << "client say: " << buffer << endl;
            // 服务器给客户端回复
            string content = "<html><body><h1>hello world</h1></body></html>";
            string echo_str = "HTTP/1.0 200 OK\r\n";
            echo_str += "Content-Type: text/html\r\n";
            echo_str += "Content-Length: " + to_string(content.size()) + "\r\n\r\n";
            echo_str += content;
            // 发送数据
            ::send(fd, echo_str.c_str(), echo_str.size(), 0);
        }
        else if (0 == n)
        {
            // 对端关闭连接
            LOG(INFO, "client quit, fd: %d", fd);
            // 将 fd 从 epoll 模型中删除
            //  想从 epoll 中移除 fd 时，这个 fd 必须是健康 & 合法的，否则会移除出错
            ::epoll_ctl(_epfd, EPOLL_CTL_DEL, fd, nullptr);
            ::close(fd);
        }
        else
        {
            // 读取失败
            LOG(ERROR, "recv error, fd: %d", fd);
            // 将 fd 从 epoll 模型中删除
            ::epoll_ctl(_epfd, EPOLL_CTL_DEL, fd, nullptr);
            ::close(fd);
        }
    }

    // 处理已经就绪的 n 个事件
    void handle_event(int n)
    {
        for (size_t i = 0; i < n; i++)
        {
            int fd = ready_events[i].data.fd;
            uint16_t revents = ready_events[i].events;
            LOG(INFO, "已经有事件就绪了，具体事件是:  %s", events_to_string(revents).c_str());

            // 处理就绪的读事件
            if (revents & EPOLLIN)
            {
                if (fd == _listen_socket->sockfd())
                {    
                    // 如果是 listen 套接字就绪，说明有新的连接到来
                    accepter_connection();
                }
                else
                {
                    // 普通的读事件就绪
                    handle_normal_read_event(i);
                }
            }
        }
    }

    ~epoll_server()
    {
        if (_epfd >= 0)
            close(_epfd);
        if (_listen_socket)
            close(_listen_socket->sockfd());
    }
};
                 // 如果是 listen 套接字就绪，说明有新的连接到来
                    accepter_connection();
                }
                else
                {
                    // 普通的读事件就绪
                    handle_normal_read_event(i);
                }
            }
        }
    }

    ~epoll_server()
    {
        if (_epfd >= 0)
            close(_epfd);
        if (_listen_socket)
            close(_listen_socket->sockfd());
    }
};