解析 ‘Epoll’ 的红黑树与就绪链表：为什么它能比 `select` 支撑多出两个数量级的并发？

在构建高性能网络服务时，如何有效地管理并发I/O是核心挑战。从传统的select、poll到现代的epoll，Linux提供了多种I/O多路复用机制，而epoll无疑是其中的翘楚，其在支撑大规模并发连接方面展现出的卓越性能，常常能够比select或poll高出两个数量级。今天，我们将深入剖析epoll的内部工作机制，特别是其基于红黑树和就绪链表的设计，来理解它为何能达成这样的壮举。

传统I/O多路复用：select的局限性

在epoll诞生之前，select是Linux下最常见的I/O多路复用机制。它的基本思想是：允许程序监听一组文件描述符（file descriptors, FDs），并在其中任何一个FD就绪（例如，有数据可读、可写或发生错误）时通知程序。

select的工作原理

select系统调用接收三个fd_set类型的参数：readfds、writefds和exceptfds，分别用于监听可读、可写和异常事件。此外，它还接收一个nfds参数，表示所有待监听FD中的最大值加一，以及一个timeout参数。

#include <sys/select.h>
#include <sys/time.h>
#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <netinet/in.h>
#include <sys/socket.h>

#define MAX_CLIENTS 1024
#define PORT 8080

void handle_client(int client_fd) {
    char buffer[1024] = {0};
    ssize_t bytes_read = read(client_fd, buffer, 1023);
    if (bytes_read <= 0) {
        if (bytes_read == 0) {
            printf("Client disconnected: %dn", client_fd);
        } else {
            perror("read error");
        }
        close(client_fd);
        return;
    }
    printf("Received from client %d: %s", client_fd, buffer);
    send(client_fd, buffer, bytes_read, 0); // Echo back
}

int main() {
    int master_socket, addrlen, new_socket;
    struct sockaddr_in address;
    fd_set readfds; // Set of file descriptors for reading

    // Create a master socket
    if ((master_socket = socket(AF_INET, SOCK_STREAM, 0)) == 0) {
        perror("socket failed");
        exit(EXIT_FAILURE);
    }

    // Set master socket to allow multiple connections
    int opt = 1;
    if (setsockopt(master_socket, SOL_SOCKET, SO_REUSEADDR, (char *)&opt, sizeof(opt)) < 0) {
        perror("setsockopt failed");
        exit(EXIT_FAILURE);
    }

    // Type of socket created
    address.sin_family = AF_INET;
    address.sin_addr.s_addr = INADDR_ANY;
    address.sin_port = htons(PORT);

    // Bind the socket to localhost port 8080
    if (bind(master_socket, (struct sockaddr *)&address, sizeof(address)) < 0) {
        perror("bind failed");
        exit(EXIT_FAILURE);
    }
    printf("Listener on port %d n", PORT);

    // Try to specify backlog of 3 pending connections
    if (listen(master_socket, 3) < 0) {
        perror("listen");
        exit(EXIT_FAILURE);
    }

    addrlen = sizeof(address);
    printf("Waiting for connections ...n");

    int client_sockets[MAX_CLIENTS];
    for (int i = 0; i < MAX_CLIENTS; i++) {
        client_sockets[i] = 0; // Initialize all client sockets to 0
    }

    int max_sd = master_socket; // Initialize max_sd to master socket
    while (1) {
        FD_ZERO(&readfds); // Clear the socket set
        FD_SET(master_socket, &readfds); // Add master socket to set

        // Add child sockets to set
        for (int i = 0; i < MAX_CLIENTS; i++) {
            if (client_sockets[i] > 0) {
                FD_SET(client_sockets[i], &readfds);
            }
            if (client_sockets[i] > max_sd) {
                max_sd = client_sockets[i]; // Update max_sd
            }
        }

        // Wait for an activity on one of the sockets, timeout is NULL, so wait indefinitely
        int activity = select(max_sd + 1, &readfds, NULL, NULL, NULL);

        if ((activity < 0) && (errno != EINTR)) {
            perror("select error");
        }

        // If something happened on the master socket, then it's an incoming connection
        if (FD_ISSET(master_socket, &readfds)) {
            if ((new_socket = accept(master_socket, (struct sockaddr *)&address, (socklen_t *)&addrlen)) < 0) {
                perror("accept");
                exit(EXIT_FAILURE);
            }
            printf("New connection, socket fd is %d, ip is : %s, port : %d n", new_socket,
                   inet_ntoa(address.sin_addr), ntohs(address.sin_port));

            // Add new socket to array of sockets
            for (int i = 0; i < MAX_CLIENTS; i++) {
                if (client_sockets[i] == 0) {
                    client_sockets[i] = new_socket;
                    printf("Adding to list of sockets as %dn", i);
                    break;
                }
            }
        }

        // Else it's some other I/O operation on a client socket
        for (int i = 0; i < MAX_CLIENTS; i++) {
            int sd = client_sockets[i];
            if (sd > 0 && FD_ISSET(sd, &readfds)) {
                handle_client(sd);
                // If client disconnected, remove from array
                if (sd <= 0) { // handle_client closes the socket and sets sd to 0 implicitly (or we can explicitly set client_sockets[i]=0)
                    client_sockets[i] = 0;
                }
            }
        }
    }
    return 0;
}

select的性能瓶颈

尽管select可以处理并发，但其在处理大量并发连接时效率低下，主要有以下几个原因：

1. 线性扫描（O(N)）:

   • 用户空间到内核空间的数据拷贝: 每次调用select时，用户程序都需要将完整的fd_set（一个位图，每个位代表一个FD）从用户空间拷贝到内核空间。随着FD数量N的增加，这个拷贝操作的开销也线性增加。
   • 内核对FD的遍历: 内核接收到fd_set后，必须遍历所有N个可能的文件描述符，检查它们是否就绪。这个遍历操作的复杂度是O(N)。
   • 内核到用户空间的数据拷贝: select返回时，内核会修改fd_set，只保留就绪的FD位，然后将其拷贝回用户空间。用户程序同样需要遍历整个fd_set来找出哪些FD是就绪的，这又是一个O(N)的操作。
   • 因此，select的总时间复杂度是O(N)，其中N是所有被监听的FD数量。当N达到数千甚至上万时，这种线性开销会变得非常显著。

2. 文件描述符数量限制:

   • fd_set通常是一个固定大小的位图，由FD_SETSIZE宏定义，在大多数系统上默认为1024。这意味着一个进程使用select最多只能同时监听1024个文件描述符。虽然可以通过重新编译内核来修改这个限制，但这并不灵活，也无法解决根本的性能问题。

3. 每次调用都需要重新设置:

   • 每次调用select前，都需要清空fd_set并重新添加所有需要监听的FD，这增加了用户空间的开销。

poll：改进但未根本解决问题

poll系统调用是select的一个改进版本。它使用一个struct pollfd数组替代了fd_set，从而解决了FD_SETSIZE的限制，理论上可以监听无限数量的FD。

struct pollfd {
    int fd;        // 文件描述符
    short events;  // 关注的事件
    short revents; // 实际发生的事件
};

poll的调用方式如下：

int poll(struct pollfd *fds, nfds_t nfds, int timeout);

fds是一个pollfd结构体数组，nfds是数组中元素的数量。

尽管poll解决了FD数量限制，但其核心的性能瓶颈与select相同：内核仍然需要遍历整个pollfd数组来检查每个FD的状态，并将其就绪状态填充到revents字段中，然后将整个数组拷贝回用户空间。因此，poll的时间复杂度仍然是O(N)。对于大规模并发，它仍然无法提供令人满意的性能。

epoll：革新与高效的秘密

epoll是Linux特有的I/O多路复用机制，自Linux 2.5.44引入。它彻底改变了事件通知的模型，从“我问你哪些FD准备好了”转变为“你告诉我哪些FD准备好了”。这种模式的转变是epoll能支撑海量并发的关键。

epoll提供了三个核心系统调用：

1. epoll_create / epoll_create1: 创建一个epoll实例（或称epoll句柄），返回一个文件描述符。这个FD代表了内核中一个epoll事件的上下文。
2. epoll_ctl: 用于向epoll实例中添加、修改或删除要监听的文件描述符及其事件。
3. epoll_wait: 阻塞等待，直到被监听的FD中有事件发生，并返回就绪事件的列表。

epoll的内部机制：红黑树与就绪链表

epoll之所以高效，在于它在内核中维护了两个关键的数据结构：

1. 管理所有被监听FD的红黑树（Red-Black Tree）
2. 存储所有已就绪FD的就绪链表（Ready List）

1. 红黑树：高效管理被监听的FD集合

当调用epoll_create时，内核会创建一个epoll实例。这个实例在内核中关联了一个红黑树。红黑树是一种自平衡的二叉查找树，它能保证在最坏情况下，插入、删除和查找操作的时间复杂度都是O(log N)，其中N是被监听的文件描述符总数。

• epoll_ctl(EPOLL_CTL_ADD, fd, event):

  • 当应用程序调用epoll_ctl并传入EPOLL_CTL_ADD操作时，内核会将新的文件描述符fd及其关注的事件event（包括用户自定义数据data）封装成一个内部结构（通常是struct epitem），然后将其插入到该epoll实例对应的红黑树中。
  • 这个插入操作的复杂度是O(log N)。
  • 同时，内核还会对这个fd进行回调注册。这意味着，当fd对应的设备（例如网卡）上发生I/O事件时，设备驱动程序能够直接通知epoll模块。

• epoll_ctl(EPOLL_CTL_MOD, fd, event):

  • 修改某个FD的关注事件。内核会在红黑树中查找对应的epitem，然后更新其事件信息。查找和更新的复杂度也是O(log N)。

• epoll_ctl(EPOLL_CTL_DEL, fd):

  • 从epoll实例中删除一个FD。内核会在红黑树中查找并删除对应的epitem，并取消回调注册。删除操作的复杂度是O(log N)。

为什么是红黑树？
相较于select或poll每次都要从头遍历所有FD，epoll使用红黑树来高效地管理FD集合。这意味着无论需要添加、删除还是查找一个FD，其开销都只与log N相关，而不是N。这对于处理成千上万个连接的服务器来说，是一个巨大的性能提升。

2. 就绪链表：只返回真正就绪的FD

除了红黑树，epoll实例在内核中还维护了一个双向链表，被称为“就绪链表”（或活跃链表）。这个链表专门用于存放那些真正已经就绪的文件描述符。

• 事件发生时:

  • 当一个被epoll监听的文件描述符fd（在红黑树中）发生I/O事件时，例如数据到达网卡，对应的设备驱动程序会检测到这个事件。
  • 由于在EPOLL_CTL_ADD时已经注册了回调，驱动程序会直接调用epoll的回调函数。
  • epoll的回调函数会将这个fd对应的epitem结构体添加到该epoll实例的就绪链表中。这个添加操作是O(1)的。
  • 如果此时有线程正在调用epoll_wait并处于阻塞状态，它会被唤醒。

• epoll_wait(epoll_fd, events, maxevents, timeout):

  • 当应用程序调用epoll_wait时，内核会检查该epoll实例的就绪链表。
  • 如果就绪链表不为空，epoll_wait会立即将链表中的所有（或最多maxevents个）epitem的数据拷贝到用户提供的events数组中，然后返回。
  • 这个拷贝操作的复杂度是O(K)，其中K是就绪事件的数量。
  • 如果就绪链表为空，epoll_wait会阻塞，直到有新的事件发生并被添加到就绪链表，或者超时。

为什么是就绪链表？
这是epoll最核心的优势所在。select和poll每次都需要遍历所有N个FD来找出K个就绪的FD，其开销是O(N)。而epoll_wait则直接从就绪链表中取出已就绪的K个事件，其开销是O(K)。当N非常大（例如几十万），而K相对较小（例如同时只有几百个连接活跃）时，O(K)的开销远小于O(N)。这正是epoll能够支撑两个数量级以上并发的根本原因。

epoll的触发模式：LT与ET

epoll支持两种事件触发模式：

1. 水平触发（Level-Triggered, LT）:

   • 默认模式，也是select和poll的工作模式。
   • 只要文件描述符上还有I/O事件未处理，epoll_wait就会一直通知应用程序。例如，如果一个socket有100字节数据可读，而应用程序只读了50字节，那么在下一次调用epoll_wait时，仍然会再次收到该socket可读的通知。
   • 优点: 编程简单，不容易丢失事件。
   • 缺点: 如果应用程序处理不及时，可能会导致重复唤醒，增加不必要的开销。

2. 边缘触发（Edge-Triggered, ET）:

   • 只有在文件描述符的状态发生变化时，epoll_wait才会通知应用程序一次。例如，当一个socket从不可读变为可读时，epoll_wait会通知一次。即使应用程序只读了一部分数据，只要状态没有再次变化（例如又接收到新数据），就不会再次收到通知。
   • 优点: 效率更高，只通知一次，避免了重复唤醒，减少了epoll_wait的调用次数和内核态/用户态的切换。
   • 缺点: 编程复杂，应用程序必须一次性将所有可读数据读完，或将所有可写数据写完，否则剩余的数据将不会再次触发事件，可能导致数据丢失或死锁。因此，在使用ET模式时，文件描述符必须设置为非阻塞模式。

在高性能服务器中，ET模式通常与非阻塞I/O结合使用，以最大化效率。

epoll代码示例

下面是一个简化的epoll服务器示例：

#include <sys/epoll.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <fcntl.h>
#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>
#include <errno.h>
#include <string.h>

#define MAX_EVENTS 1024 // epoll_wait一次最多返回的事件数量
#define PORT 8080
#define BUFFER_SIZE 1024

// 设置文件描述符为非阻塞模式
int set_nonblocking(int fd) {
    int flags = fcntl(fd, F_GETFL, 0);
    if (flags == -1) {
        perror("fcntl F_GETFL");
        return -1;
    }
    flags |= O_NONBLOCK;
    if (fcntl(fd, F_SETFL, flags) == -1) {
        perror("fcntl F_SETFL");
        return -1;
    }
    return 0;
}

int main() {
    int listen_sock, conn_sock, epoll_fd;
    struct sockaddr_in server_addr, client_addr;
    socklen_t client_len;
    struct epoll_event event; // 用于epoll_ctl
    struct epoll_event *events; // 用于epoll_wait返回事件

    // 1. 创建监听socket
    listen_sock = socket(AF_INET, SOCK_STREAM, 0);
    if (listen_sock == -1) {
        perror("socket");
        exit(EXIT_FAILURE);
    }

    // 设置SO_REUSEADDR选项，允许端口复用
    int opt = 1;
    if (setsockopt(listen_sock, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) == -1) {
        perror("setsockopt SO_REUSEADDR");
        close(listen_sock);
        exit(EXIT_FAILURE);
    }

    // 设置监听socket为非阻塞模式
    if (set_nonblocking(listen_sock) == -1) {
        close(listen_sock);
        exit(EXIT_FAILURE);
    }

    memset(&server_addr, 0, sizeof(server_addr));
    server_addr.sin_family = AF_INET;
    server_addr.sin_addr.s_addr = INADDR_ANY;
    server_addr.sin_port = htons(PORT);

    if (bind(listen_sock, (struct sockaddr *)&server_addr, sizeof(server_addr)) == -1) {
        perror("bind");
        close(listen_sock);
        exit(EXIT_FAILURE);
    }

    if (listen(listen_sock, SOMAXCONN) == -1) {
        perror("listen");
        close(listen_sock);
        exit(EXIT_FAILURE);
    }
    printf("Listening on port %dn", PORT);

    // 2. 创建epoll实例
    epoll_fd = epoll_create1(0); // 0表示默认flags, 等价于epoll_create(size)
    if (epoll_fd == -1) {
        perror("epoll_create1");
        close(listen_sock);
        exit(EXIT_FAILURE);
    }

    // 3. 将监听socket加入epoll实例
    event.data.fd = listen_sock;
    // 监听可读事件，并设置为边缘触发模式
    event.events = EPOLLIN | EPOLLET; // EPOLLIN for read, EPOLLET for edge-triggered
    if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, listen_sock, &event) == -1) {
        perror("epoll_ctl: listen_sock");
        close(listen_sock);
        close(epoll_fd);
        exit(EXIT_FAILURE);
    }

    // 分配内存用于存储epoll_wait返回的事件
    events = calloc(MAX_EVENTS, sizeof(event));
    if (events == NULL) {
        perror("calloc events");
        close(listen_sock);
        close(epoll_fd);
        exit(EXIT_FAILURE);
    }

    char buffer[BUFFER_SIZE];

    // 4. 事件循环
    while (1) {
        int nfds = epoll_wait(epoll_fd, events, MAX_EVENTS, -1); // -1表示无限等待
        if (nfds == -1) {
            if (errno == EINTR) continue; // 被信号中断，继续等待
            perror("epoll_wait");
            break;
        }

        for (int i = 0; i < nfds; ++i) {
            if (events[i].data.fd == listen_sock) {
                // 有新的连接请求
                while (1) { // 边缘触发模式下，需要循环接受所有新连接
                    client_len = sizeof(client_addr);
                    conn_sock = accept(listen_sock, (struct sockaddr *)&client_addr, &client_len);
                    if (conn_sock == -1) {
                        if (errno == EAGAIN || errno == EWOULDBLOCK) {
                            // 所有等待接受的连接都已处理完毕
                            break;
                        } else {
                            perror("accept");
                            break;
                        }
                    }

                    printf("Accepted connection from %s:%d on socket %dn",
                           inet_ntoa(client_addr.sin_addr), ntohs(client_addr.sin_port), conn_sock);

                    // 设置新连接socket为非阻塞模式
                    if (set_nonblocking(conn_sock) == -1) {
                        close(conn_sock);
                        continue;
                    }

                    // 将新连接socket加入epoll实例
                    event.data.fd = conn_sock;
                    event.events = EPOLLIN | EPOLLET; // 同样是边缘触发
                    if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, conn_sock, &event) == -1) {
                        perror("epoll_ctl: conn_sock");
                        close(conn_sock);
                    }
                }
            } else {
                // 客户端socket有数据可读或发生错误
                int current_fd = events[i].data.fd;

                if (events[i].events & (EPOLLRDHUP | EPOLLHUP | EPOLLERR)) {
                    // 客户端关闭连接或发生错误
                    printf("Client %d disconnected or error occurred.n", current_fd);
                    epoll_ctl(epoll_fd, EPOLL_CTL_DEL, current_fd, NULL); // 从epoll中移除
                    close(current_fd); // 关闭socket
                } else if (events[i].events & EPOLLIN) {
                    // 客户端有数据可读
                    while (1) { // 边缘触发模式下，需要循环读取所有数据
                        ssize_t bytes_read = read(current_fd, buffer, BUFFER_SIZE - 1);
                        if (bytes_read == -1) {
                            if (errno == EAGAIN || errno == EWOULDBLOCK) {
                                // 数据已读完
                                break;
                            } else {
                                perror("read");
                                epoll_ctl(epoll_fd, EPOLL_CTL_DEL, current_fd, NULL);
                                close(current_fd);
                                break;
                            }
                        } else if (bytes_read == 0) {
                            // 客户端优雅关闭
                            printf("Client %d closed connection.n", current_fd);
                            epoll_ctl(epoll_fd, EPOLL_CTL_DEL, current_fd, NULL);
                            close(current_fd);
                            break;
                        } else {
                            buffer[bytes_read] = '';
                            printf("Received from %d: %s", current_fd, buffer);
                            // 简单回显数据
                            ssize_t bytes_written = 0;
                            while (bytes_written < bytes_read) {
                                ssize_t nw = write(current_fd, buffer + bytes_written, bytes_read - bytes_written);
                                if (nw == -1) {
                                    if (errno == EAGAIN || errno == EWOULDBLOCK) {
                                        // 缓冲区满，需要等待下次EPOLLOUT事件
                                        // 这里为了简化，直接退出，实际生产环境需要更复杂的处理
                                        // 比如将剩余数据放入发送缓冲区，并修改epoll事件为EPOLLOUT
                                        fprintf(stderr, "Write buffer full for %d. Partial write.n", current_fd);
                                        break;
                                    } else {
                                        perror("write");
                                        epoll_ctl(epoll_fd, EPOLL_CTL_DEL, current_fd, NULL);
                                        close(current_fd);
                                        break;
                                    }
                                }
                                bytes_written += nw;
                            }
                        }
                    }
                }
                // 对于EPOLLOUT事件，如果需要的话，可以在此处理
                // 例如，当发送缓冲区从满变为可写时，触发EPOLLOUT，然后发送之前未发送完的数据
                // 这里为了简化，只处理了EPOLLIN
            }
        }
    }

    // 清理资源
    free(events);
    close(epoll_fd);
    close(listen_sock);

    return 0;
}

代码解析要点：

• 非阻塞模式: listen_sock和所有conn_sock都被设置为非阻塞模式。这是使用epoll，特别是ET模式的关键。
• epoll_create1(0): 创建epoll实例。
• epoll_ctl(EPOLL_CTL_ADD, ..., listen_sock, EPOLLIN | EPOLLET): 将监听socket添加到epoll实例，并设置监听EPOLLIN（可读）事件，采用EPOLLET（边缘触发）模式。
• epoll_wait: 阻塞等待事件发生。
• 循环accept和read: 在边缘触发模式下，当listen_sock或conn_sock被通知就绪时，必须在一个循环中尽可能多地处理所有待处理的I/O操作（例如，accept所有挂起的连接，read所有可读数据），直到accept或read返回EAGAIN/EWOULDBLOCK错误，表示当前已无更多可处理的事件。
• 错误处理与连接关闭: 妥善处理read返回0（客户端关闭）和-1（错误），以及EPOLLRDHUP/EPOLLHUP/EPOLLERR等事件，及时从epoll实例中移除FD并关闭socket。

epoll与select/poll的性能对比总结

下表概括了select、poll和epoll在关键特性上的差异：

特性	select	poll	epoll
FD 数量限制	FD_SETSIZE (通常 1024)	仅受内存限制 (几乎无限制)	仅受内存限制 (几乎无限制)
内核数据结构	位图 (fd_set)	pollfd结构体数组	红黑树 (管理所有FD) + 双向链表 (就绪FD)
监听 FD 的开销	O(N) (每次调用都需重新设置位图)	O(N) (每次调用都需重新设置数组)	O(log N) (添加/修改/删除 FD)
等待/获取事件开销	O(N) (内核遍历所有 FD 查找就绪事件)	O(N) (内核遍历所有 FD 查找就绪事件)	O(K) (直接从就绪链表获取 K 个事件)
用户/内核数据拷贝	O(N) (每次拷贝整个 fd_set)	O(N) (每次拷贝整个 pollfd 数组)	O(K) (只拷贝 K 个就绪事件)
触发模式	水平触发 (LT)	水平触发 (LT)	水平触发 (LT) & 边缘触发 (ET)
适用场景	低并发 (<1000 连接)	中低并发 (数千连接)	高并发 (数万至数十万连接)
编程复杂性	中等 (位操作，需每次重置)	中等 (数组操作，需每次重置)	较高 (非阻塞I/O, ET模式下的循环处理等)

epoll之所以能够比select或poll支撑多出两个数量级的并发，根本原因在于其将“事件检测”的开销从O(N)（与总连接数线性相关）降低到了O(K)（与活跃连接数线性相关）和O(log N)（与总连接数对数相关）。在大量不活跃连接（N很大）中只有少量连接活跃（K很小）的场景下，epoll的优势被无限放大。此外，epoll通过避免不必要的用户-内核数据拷贝，以及支持效率更高的边缘触发模式，进一步巩固了其在高并发I/O领域的霸主地位。

epoll通过其精巧的内核数据结构（红黑树和就绪链表），以及事件驱动的通知机制，彻底革新了Linux下的高并发网络编程模型，成为现代高性能服务器（如Nginx、Redis、Node.js等）的基石。理解其内部机制，对于构建可伸缩、高性能的网络应用程序至关重要。