网络篇|重温IO多路复用Select/poll/epoll

AKoalas2023-12-202024-03-17

前言

之所以重温IO多路复用，是RTSP服务器实现想放在Linux下实现，而那么网络编程不可避免的要接触。而且Windows下只有select可以直接使用，想使用poll和epoll只能用第三方库或者一些功能相似的I/O多路复用，但是性能相较于Linux下有着差距。

在回忆I/O多路复用之前，先看看一般场景下如何实现服务端与客户端的连接。

服务端与客户端的一般连接方式

这里我们用幽默的说法来对连接方式做一个区分。用斗破苍穹的等级来划分(都学习网络编程了，咋说C++也是有点水平了，起码也是斗王阶级了)：

斗王：能建立一个普通的一对一连接方式。

斗皇：在借用线程的情况下，能实现一对多的的连接方式。

斗宗：熟练使用select建立连接，极大的提高CPU利用率。

斗尊：熟练使用select和poll建立连接。因为二者没有什么太大的区别，只是在不同的情况下选择不同。

斗圣：select/poll/epoll各种技巧与原理不在话下，写个Demo信手拈来。

斗帝：恐怖如斯，桀桀桀桀桀桀桀~

服务端建立连接无非就是socket、bind、listen、accept、send、recv一般连接方式就是网络编程中最常见的单线程与多线程两种方式。

一般单线程连接方式

这里的单线程指的是普通的一对一连接方式。

accept是阻塞接口，只有接收到新的连接之后才能进行发送send和接收recv操作。

而accept一个请求之后，在recv或send调用阻塞时，也无法继续accept其他请求，此时又不能满足RTSP服务并发情况，那么这个连接方式不考虑。

一般多线程连接方式

上面一对一不满足并发？那就使用线程，每个线程调用一个accept。并发功能是满足了，那么问题又来了，连接数过万，性能能抗住吗？

无论是Linux还是Windows，进程的可用内存空间都有上限。32位Windows内存空间上限一般是2G（操作系统划分用户态和内核态导致的），并且默认情况下，一个线程的栈要预留1M的内存空间，所以理论一个进程最多可以开2048个线程。所以这个连接方式也不考虑。

IO多路复用相关概念

了解IO多路复用，得先回忆几个相关概念。链接参考点击子标题。

进程切换

为了控制进程的执行，内核有能力挂起正在CPU上运行的进程，并恢复以前挂起的某个进程的执行。这种行为被叫做进程切换。但是进程切换是非常消耗资源的。

一个进程切换往往经过如下几步：

保存处理机上下文，包括程序计数器和寄存器
更新PCB(Process Control Block)消息
把进程的PCB移入相应的队列，如就绪、在某事件阻塞等队列
选择另外一个进程执行，并更新PCB
更新内存管理的数据结构
恢复处理机的上下文

文件描述符

内核(kernel)利用文件描述符(file descriptor)来访问文件。文件描述符是非负整数。打开现存文件或新建文件时，内核会返回一个文件描述符。实际上，它是一个索引值，指向内核为每一个进程所维护的该进程打开文件的记录表。当程序打开一个现有文件或者创建一个新文件时，内核向进程返回一个文件描述符。在程序设计中，一些涉及底层的程序编写往往会围绕着文件描述符展开。但是文件描述符这一概念往往只适用于UNIX、Linux这样的操作系统。

缓存I/O

缓存I/O又称为标准I/O，大多数文件系统的默认I/O操作都是缓存I/O。在Linux的缓存I/O机制中，操作系统会将I/O的数据缓存在文件系统的页缓存中，即数据会先被拷贝到操作系统内核的缓冲区中，然后才会从操作系统内核的缓冲区拷贝到应用程序的地址空间。

I/O多路复用的三种实现方式

I/O多路复用的定义

I/O多路复用是一种同步IO模型，实现一个线程可以监视多个文件句柄。只要在监视范围内的文件句柄处于就绪状态，就能够通知应用程序进行相应的操作。如果没有监视任何一个文件句柄，那么将阻塞应用程序，交出CPU。

多路就是可以建立多个网络连接，复用就是通过一个线程或者多个线程就能满足我们的使用。

所以可以提前说，I/O比上面两种方式就是利用最小的CPU消耗管理最多的连接数，并且还不需要维护。

Select机制

Windows下API接口如下，Linux下接口参数没有区别。

int WSAAPI select(
  [in]      int           nfds,
  [in, out] fd_set        *readfds,
  [in, out] fd_set        *writefds,
  [in, out] fd_set        *exceptfds,
  [in]      const timeval *timeout
);

其中readfds、writefds、exceptfds表示可读性文件描述符、可写性文件描述符、检查错误文件描述符。

这里简单描述一下Select的使用过程：select 会阻塞住监视 3 类文件描述符，等有数据、可读、可写、出异常或超时就会返回。返回后通过遍历 fdset 整个数组来找到就绪的描述符 fd，然后进行对应的 IO 操作。

Select在整个过程中是帮助监听套接字。如果返回值为0，则表示超时，此时可以自己设置超时操作，比如Close。正常情况下返回已就绪并包含在fd_set结构中的套接字句柄总数。

select的优点

支持跨平台，Windows和Linux下都支持。
相比于上面的两种连接方式，select可以一个线程控制多个连接，性能有了很大的提升与更好的利用。
在IO多路复用中，使用非常简单。

select的缺点

每次监视有可读、可写、异常之后，由于是采用轮询方式全盘扫描，会随着文件描述符 FD 数量增多而性能下降。
每次调用 select()，都需要把 fd 集合从用户态拷贝到内核态，并进行遍历(消息传递都是从内核到用户空间)。
单个进程打开的 FD 是有限制(通过FD_SETSIZE设置)的，默认是 1024 个，可修改宏定义，但是效率仍然慢。

select的示例

#include <iostream>
#include <cstring>
#include <cstdlib>
#include <vector>
#include <set>
#include <cerrno>

#include <unistd.h>
#include <arpa/inet.h>

int main() {
    // Create a server socket
    int serverSocket = socket(AF_INET, SOCK_STREAM, 0);
    if (serverSocket == -1) {
        perror("Error creating server socket");
        exit(EXIT_FAILURE);
    }

    // Set up the server address structure
    sockaddr_in serverAddress;
    serverAddress.sin_family = AF_INET;
    serverAddress.sin_addr.s_addr = INADDR_ANY;
    serverAddress.sin_port = htons(12345);  // Port number

    // Bind the server socket
    if (bind(serverSocket, (struct sockaddr*)&serverAddress, sizeof(serverAddress)) == -1) {
        perror("Error binding server socket");
        close(serverSocket);
        exit(EXIT_FAILURE);
    }

    // Listen for incoming connections
    if (listen(serverSocket, 10) == -1) {
        perror("Error listening for connections");
        close(serverSocket);
        exit(EXIT_FAILURE);
    }

    std::cout << "Server listening on port 12345...\n";

    // Set of socket descriptors
    fd_set readfds;
    FD_ZERO(&readfds);

    // Add the server socket to the set
    FD_SET(serverSocket, &readfds);

    // Track connected client sockets
    std::set<int> clientSockets;

    while (true) {
        // Copy the set of file descriptors
        fd_set tmpReadfds = readfds;

        // Use select to monitor socket activity
        if (select(FD_SETSIZE, &tmpReadfds, nullptr, nullptr, nullptr) == -1) {
            perror("Error in select");
            break;
        }

        // Check each socket for activity
        for (int i = 0; i < FD_SETSIZE; ++i) {
            if (FD_ISSET(i, &tmpReadfds)) {
                // New connection
                if (i == serverSocket) {
                    // Accept the new connection
                    int clientSocket = accept(serverSocket, nullptr, nullptr);
                    if (clientSocket == -1) {
                        perror("Error accepting connection");
                    } else {
                        std::cout << "New connection established. Socket FD: " << clientSocket << "\n";
                        FD_SET(clientSocket, &readfds);
                        clientSockets.insert(clientSocket);
                    }
                } else {
                    // Data from a client
                    char buffer[1024];
                    int bytesRead = recv(i, buffer, sizeof(buffer), 0);

                    if (bytesRead <= 0) {
                        // Connection closed or error
                        if (bytesRead == 0) {
                            std::cout << "Client " << i << " disconnected.\n";
                        } else {
                            perror("Error receiving data");
                        }

                        // Remove the socket from the set and close it
                        close(i);
                        FD_CLR(i, &readfds);
                        clientSockets.erase(i);
                    } else {
                        // Display received data
                        buffer[bytesRead] = '\0';
                        std::cout << "Received data from client " << i << ": " << buffer << "\n";
                    }
                }
            }
        }
    }

    // Close all client sockets
    for (int clientSocket : clientSockets) {
        close(clientSocket);
    }

    // Close the server socket
    close(serverSocket);

    return 0;
}

poll机制

poll本质上和select没有区别。下面是对二者的对比：

内核对应文件描述符的检测也是按照线性的方式轮询，根据描述符的状态进行处理。

poll和select检测的文件描述符集合会在检测过程中频繁的进行用户态和内核态的拷贝，开销随着文件描述的增加也开始线性的增加，这样导致效率也越来越低。

select有最大连接限制，但是poll没有。因为poll实现原理是用链表来存储的

select可以支持跨平台，poll只能在Linux下。

poll的接口

#include <poll.h>
// 每个委托poll检测的fd都对应这样一个结构体
struct pollfd {
    int   fd;         /* 委托内核检测的文件描述符 */
    short events;     /* 委托内核检测文件描述符的什么事件 */
    short revents;    /* 文件描述符实际发生的事件 -> 传出 */
};

struct pollfd myfd[100];
int poll(struct pollfd *fds, nfds_t nfds, int timeout);

文件描述符（FD）是一个整数，用于标识一个打开的文件、套接字或其他系统资源。在 Unix 系统中，每个进程都可以有一个文件描述符集合，用于标识该进程打开的所有文件、套接字和其他系统资源。
事件标志（event mask）是一个整数，用于表示当文件描述符的状态发生变化时应该触发的事件。例如，如果套接字接收到数据，则可以使用 POLLIN 标志表示。其他常用的标志包括 POLLOUT（套接字准备好写入数据）、POLLERR（套接字发生错误）和 POLLHUP（套接字断开连接）。
超时值（timeout）是一个整数，表示在等待套接字事件时不会超时。如果超时发生，poll 函数将返回一个非零值，表示套接字事件尚未发生。如果没有超时，poll 函数将等待指定的时间，然后返回一个零值，表示套接字事件已经发生。

poll的示例

一个简单的例子如下：

#include <iostream>
#include <cstring>
#include <cstdlib>
#include <vector>
#include <algorithm>

#include <unistd.h>
#include <arpa/inet.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <netinet/in.h>
#include <poll.h>

int main() {
    // Create a server socket
    int serverSocket = socket(AF_INET, SOCK_STREAM, 0);
    if (serverSocket == -1) {
        perror("Error creating server socket");
        exit(EXIT_FAILURE);
    }

    // Set up the server address structure
    sockaddr_in serverAddress;
    serverAddress.sin_family = AF_INET;
    serverAddress.sin_addr.s_addr = INADDR_ANY;
    serverAddress.sin_port = htons(12345);  // Port number

    // Bind the server socket
    if (bind(serverSocket, (struct sockaddr*)&serverAddress, sizeof(serverAddress)) == -1) {
        perror("Error binding server socket");
        close(serverSocket);
        exit(EXIT_FAILURE);
    }

    // Listen for incoming connections
    if (listen(serverSocket, 10) == -1) {
        perror("Error listening for connections");
        close(serverSocket);
        exit(EXIT_FAILURE);
    }

    std::cout << "Server listening on port 12345...\n";

    // Prepare the pollfd structures
    std::vector<pollfd> fds;
    fds.push_back({serverSocket, POLLIN, 0}); // Add server socket to the vector

    while (true) {
        // Call poll to wait for events on the sockets
        int activity = poll(fds.data(), fds.size(), -1);
        if (activity == -1) {
            perror("Error in poll");
            break;
        }

        // Check if the server socket has an incoming connection
        if (fds[0].revents & POLLIN) {
            int clientSocket = accept(serverSocket, nullptr, nullptr);
            if (clientSocket == -1) {
                perror("Error accepting connection");
            } else {
                std::cout << "New connection established. Socket FD: " << clientSocket << "\n";
                fds.push_back({clientSocket, POLLIN, 0}); // Add new client socket to the vector
            }
        }

        // Check for input from clients
        for (size_t i = 1; i < fds.size(); ++i) {
            if (fds[i].revents & POLLIN) {
                char buffer[1024];
                int bytesRead = recv(fds[i].fd, buffer, sizeof(buffer), 0);

                if (bytesRead <= 0) {
                    if (bytesRead == 0) {
                        std::cout << "Client " << fds[i].fd << " disconnected.\n";
                    } else {
                        perror("Error receiving data");
                    }

                    close(fds[i].fd);
                    fds.erase(fds.begin() + i); // Remove the client socket from the vector
                    --i;
                } else {
                    buffer[bytesRead] = '\0';
                    std::cout << "Received data from client " << fds[i].fd << ": " << buffer << "\n";

                    // Echo received data back to the client
                    send(fds[i].fd, buffer, bytesRead, 0);
                }
            }
        }
    }

    // Close all client sockets
    for (size_t i = 0; i < fds.size(); ++i) {
        close(fds[i].fd);
    }

    // Close the server socket
    close(serverSocket);

    return 0;
}

与上面的select例子相对比，代码实现有很多相同地方。

epoll的机制

epoll全称eventpoll，在IO多路转接/复用中，一个操作同时监听多个输入输出源，在其中一个或多个输入输出源可用的时候返回，然后对其进行读写操作。从接口开始，一步一步看epoll的高效是如何体现的。

epoll的接口

#include <sys/epoll.h>

// 数据结构
// 每一个epoll对象都有一个独立的eventpoll结构体
// 用于存放通过epoll_ctl方法向epoll对象中添加进来的事件
// epoll_wait检查是否有事件发生时，只需要检查eventpoll对象中的rdlist双链表中是否有epitem元素即可
struct eventpoll {
    /*红黑树的根节点，这颗树中存储着所有添加到epoll中的需要监控的事件*/
    struct rb_root  rbr;
    /*双链表中则存放着将要通过epoll_wait返回给用户的满足条件的事件*/
    struct list_head rdlist;
};

// API
int epoll_create(int size); 
int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event); 
int epoll_wait(int epfd, struct epoll_event * events, int maxevents, int timeout);

在下面实例中，使用了一个API变体epoll_create1()，引入了一个 flags 参数，允许在创建 epoll 实例时设置一些标志。

1	int epollFd = epoll_create1(0);

当使用上面这个创建epoll实例之后，epollFd 将是一个新的 epoll 文件描述符。

epoll_ctl 是 Linux 下用于控制 epoll 实例的函数之一。它用于注册或删除文件描述符（sockets 或其他 I/O 对象）以监视的事件，并将其添加到或从 epoll 实例中移除。

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

epfd是epoll的实例描述符
op是操作类型，有3个枚举值可以填入：
- EPOLL_CTL_ADD：添加一个文件描述符到epoll实例中。
- EPOLL*_CTL_MOD*：修改已经注册的文件描述符的事件。
- EPOLL_CTL_DEL：从epoll实例中删除一个文件描述符。
fd是待操作的文件描述符，这里填入socket返回值。
event是一个结构体，指定需要监视的事件类型(见上文结构体)。定义如下：

struct epoll_event {
    uint32_t events;   // Bitmask specifying the events to monitor
    epoll_data_t data; // User data variable
};

event是需要监视的事件类型，可以是EPOLLIN、EPOLLOUT、EPOLLERR
data是一个联合体，用于关联用户定义的数据

epoll_wait 是 Linux 下用于等待文件描述符上事件的函数，特别是与 epoll 实例一起使用。它会阻塞程序执行，直到指定的文件描述符上发生了感兴趣的事件。

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

events：传出参数，里面存储了已就绪的文件描述符的信息。
maxevents：修饰第二个参数，结构体数组的容量(元素个数)。
timeout：如果检测的epoll实例中没有已就绪的文件描述符，该函数阻塞的时长，单位ms毫秒。
- 0：函数不阻塞，不管epoll实例中有没有就绪的文件描述符，函数被调用后都直接返回。
- >0：如果epoll实例中没有就绪的文件描述符，函数阻塞对应的毫秒数再返回。
- -1：函数一直阻塞，知道epoll实例中有已就绪的文件描述符才解除阻塞。
返回值
- 等于0：函数阻塞被强制解除了，没有检测到满足条件的文件描述符。
- 大于0：检测到的已就绪的文件描述符的总个数。
- 返回-1：调用失败。

eventpoll原理

每一个epoll都有一个独立的eventpoll结构体，用于存放通过epoll_ctl方法向epoll对象中添加进来的事件。这些事件都会挂载在红黑树上。

在内存索引的场景下我们一般使用红黑树来作为首选的数据结构，首先红黑树的查找速度很快O(log(N))。其次在调用epoll_create()的时候，只需要创建一个红黑树树根即可，无需浪费额外的空间。

所有添加到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双链表不为空，则把发生的事件复制到用户态，同时将事件数量返回给用户。

下面是引用的两个图，链接分别为图1、图2

从图中也可以看出，红黑树的结点和rdlist双链表的结点的同一个节点，所谓的加入rdlist双链表，就是将结点的前后指针联系到一起。所以就绪了不是将红黑树结点delete掉然后加入队列。他们是同一个结点，不需要delete。

epoll示例

这是epoll的一个简单示例

#include <iostream>
#include <sys/epoll.h>
#include <unistd.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <sys/socket.h>
#include <cstring>
#include <vector>

const int MAX_EVENTS = 10;
const int PORT = 12345;

int main() {
    int serverSocket = socket(AF_INET, SOCK_STREAM, 0);
    if (serverSocket == -1) {
        std::cerr << "Error: Could not create server socket\n";
        return 1;
    }

    sockaddr_in serverAddress;
    serverAddress.sin_family = AF_INET;
    serverAddress.sin_addr.s_addr = INADDR_ANY;
    serverAddress.sin_port = htons(PORT);

    if (bind(serverSocket, reinterpret_cast<sockaddr*>(&serverAddress), sizeof(serverAddress)) == -1) {
        std::cerr << "Error: Could not bind to port " << PORT << "\n";
        close(serverSocket);
        return 1;
    }

    if (listen(serverSocket, SOMAXCONN) == -1) {
        std::cerr << "Error: Could not listen on socket\n";
        close(serverSocket);
        return 1;
    }

    int epollFd = epoll_create1(0);
    if (epollFd == -1) {
        std::cerr << "Error: Failed to create epoll file descriptor\n";
        close(serverSocket);
        return 1;
    }

    epoll_event event;
    event.events = EPOLLIN;
    event.data.fd = serverSocket;

    if (epoll_ctl(epollFd, EPOLL_CTL_ADD, serverSocket, &event) == -1) {
        std::cerr << "Error: Failed to add server socket to epoll\n";
        close(serverSocket);
        close(epollFd);
        return 1;
    }

    std::vector<epoll_event> events(MAX_EVENTS);

    std::cout << "Server listening on port " << PORT << "...\n";

    while (true) {
        int eventCount = epoll_wait(epollFd, events.data(), MAX_EVENTS, -1);
        if (eventCount == -1) {
            std::cerr << "Error: epoll_wait failed\n";
            break;
        }

        for (int i = 0; i < eventCount; ++i) {
            if (events[i].data.fd == serverSocket) {
                int clientSocket = accept(serverSocket, nullptr, nullptr);
                if (clientSocket == -1) {
                    std::cerr << "Error: Failed to accept client connection\n";
                    continue;
                }

                std::cout << "Client connected\n";

                event.events = EPOLLIN | EPOLLET;
                event.data.fd = clientSocket;
                if (epoll_ctl(epollFd, EPOLL_CTL_ADD, clientSocket, &event) == -1) {
                    std::cerr << "Error: Failed to add client socket to epoll\n";
                    close(clientSocket);
                    continue;
                }
            } else {
                char buffer[1024];
                memset(buffer, 0, sizeof(buffer));

                int bytesRead = read(events[i].data.fd, buffer, sizeof(buffer));
                if (bytesRead <= 0) {
                    if (bytesRead == 0) {
                        std::cout << "Client disconnected\n";
                    } else {
                        std::cerr << "Error: Failed to read from client\n";
                    }
                    close(events[i].data.fd);
                    continue;
                }

                std::cout << "Received: " << buffer << "\n";

                // Echo back to the client
                if (write(events[i].data.fd, buffer, bytesRead) == -1) {
                    std::cerr << "Error: Failed to write to client\n";
                }
            }
        }
    }

    close(serverSocket);
    close(epollFd);

    return 0;
}

epoll的优点

epoll 相对于 poll 和 select 的性能优势主要体现在两个方面：事件通知机制和处理大量并发连接。

事件通知机制： epoll 使用了事件驱动的机制，通过回调函数通知应用程序关注的事件。相比之下，poll 和 select 通常需要遍历整个文件描述符集合，即使只有极少数的文件描述符有事件发生。这就导致了在 poll 和 select 中，即便只有一个文件描述符有事件发生，也需要检查整个集合，而 epoll 只会通知发生事件的文件描述符。
处理大量并发连接： 在高并发环境下，epoll 的性能通常更好。epoll 使用基于事件的模型，能够轻松处理大量并发连接，而不会因为连接数增加而引起性能下降。相对而言，poll 和 select 的性能在大规模并发连接时可能会显著下降。

关于从内核态到用户态的开销，epoll 也有一些设计上的优势：

内核事件表：epoll使用一个内核事件表来跟踪文件描述符的状态，而不是通过应用程序提供的数组。这样就减少了内核向用户空间传递数据的开销，因为内核只需要通知应用程序发生变化的事件，而不需要将整个数组传递给用户空间。
内存映射(mmap)：查阅了一些资料，epoll使用内存映射技术，可以直接将内核中的事件表映射到用户空间，减少了数据的复制。这样epoll在事件通知时的开销更小。

总的来说，epoll 通过采用更为高效的事件通知机制和更优雅的内核设计，降低了从内核态到用户态的开销，特别是在处理大量并发连接的场景中，相较于 poll 和 select 显得更为高效。

epoll LT 与 ET 模式的区别

epoll 有 EPOLLLT 和 EPOLLET 两种触发模式，LT 是默认的模式，ET 是 “高速” 模式。

LT 模式下，只要这个 fd 还有数据可读，每次 epoll_wait都会返回它的事件，提醒用户程序去操作；
ET 模式下，它只会提示一次，直到下次再有数据流入之前都不会再提示了，无论 fd 中是否还有数据可读。所以在 ET 模式下，read 一个 fd 的时候一定要把它的 buffer 读完，或者遇到 EAGIN错误。

epoll使用“事件”的就绪通知方式，通过epoll_ctl注册fd，一旦该fd就绪，内核就会采用类似callback的回调机制来激活该fd，epoll_wait便可以收到通知。

关于IO多路复用的重温到这基本上就完成了。从篇幅也可以看出，epoll的占比很多。关于原理和高效也查阅了很多博客和资料来比对求证。

下一篇继续研究音视频相关，接下来可能会忙一个月，有空会多平台继续更新。