面试题：C语言中Linux异步I/O性能优化之事件驱动模型

Question

在Linux C语言异步I/O编程中，事件驱动模型是常用的性能提升手段。请阐述epoll模型相较于select和poll模型在异步I/O性能提升方面的优势，并且详细说明如何在C语言代码中使用epoll实现高效的异步I/O事件监听。

Answer 1

epoll 相较于 select 和 poll 在异步 I/O 性能提升方面的优势

1. 文件描述符数量限制：
   • select：其所能处理的文件描述符数量是有限制的，通常受限于系统定义的常量（如 FD_SETSIZE，一般为 1024）。这在需要处理大量并发连接时会成为瓶颈。
   • poll：理论上没有文件描述符数量的限制，它使用链表来管理文件描述符集合。但在实际应用中，随着文件描述符数量的增加，性能会显著下降。
   • epoll：能轻松处理大量的文件描述符，它在内核中使用红黑树来管理注册的文件描述符，这使得添加、删除和查找操作的时间复杂度都是 O(log n)，n 为文件描述符数量。所以在处理大量并发连接时，epoll 性能优势明显。
2. 事件通知机制：
   • select：采用轮询的方式检查文件描述符上的事件。每次调用 select 时，都需要将整个文件描述符集合从用户空间拷贝到内核空间，返回时又将检查结果从内核空间拷贝回用户空间，随着文件描述符数量增多，这种拷贝开销会变得很大。并且它需要用户自己遍历整个文件描述符集合来找出哪些文件描述符发生了事件。
   • poll：同样采用轮询方式，与 select 类似，每次调用 poll 也需要将文件描述符集合从用户空间拷贝到内核空间，返回时再拷贝回来。在处理大量文件描述符时，轮询的开销巨大。
   • epoll：使用事件驱动的方式。当某个文件描述符上有事件发生时，内核会将该事件添加到一个就绪链表中。应用程序通过调用 epoll_wait 时，只需要从这个就绪链表中获取发生事件的文件描述符，不需要轮询整个文件描述符集合，大大减少了系统开销。同时，epoll 在内核和用户空间之间传递数据时，采用了内存映射（mmap）的方式，减少了数据拷贝的次数。
3. 内存使用效率：
   • select：需要维护一个固定大小的文件描述符集合，即使实际使用的文件描述符数量很少，也需要占用固定大小的内存。
   • poll：虽然理论上可以处理无限个文件描述符，但随着文件描述符数量增加，链表的维护和遍历开销会占用大量内存和 CPU 资源。
   • epoll：使用红黑树和就绪链表，内存使用效率更高，在处理大量文件描述符时，内存占用相对稳定且合理。

在 C 语言代码中使用 epoll 实现高效的异步 I/O 事件监听

以下是一个简单的示例代码，展示如何使用 epoll 实现对多个 socket 连接的异步 I/O 事件监听：

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

#define MAX_EVENTS 10
#define PORT 8080

int main() {
    int listen_fd, conn_fd;
    struct sockaddr_in servaddr, cliaddr;
    struct epoll_event ev, events[MAX_EVENTS];
    int epoll_fd;

    // 创建监听 socket
    listen_fd = socket(AF_INET, SOCK_STREAM, 0);
    if (listen_fd < 0) {
        perror("socket creation failed");
        exit(EXIT_FAILURE);
    }

    memset(&servaddr, 0, sizeof(servaddr));
    memset(&cliaddr, 0, sizeof(cliaddr));

    // 填充服务器地址结构
    servaddr.sin_family = AF_INET;
    servaddr.sin_addr.s_addr = INADDR_ANY;
    servaddr.sin_port = htons(PORT);

    // 绑定 socket 到指定地址和端口
    if (bind(listen_fd, (const struct sockaddr *)&servaddr, sizeof(servaddr)) < 0) {
        perror("bind failed");
        close(listen_fd);
        exit(EXIT_FAILURE);
    }

    // 监听连接
    if (listen(listen_fd, 10) < 0) {
        perror("listen failed");
        close(listen_fd);
        exit(EXIT_FAILURE);
    }

    // 创建 epoll 实例
    epoll_fd = epoll_create1(0);
    if (epoll_fd == -1) {
        perror("epoll_create1");
        close(listen_fd);
        exit(EXIT_FAILURE);
    }

    // 将监听 socket 添加到 epoll 实例中
    ev.events = EPOLLIN;
    ev.data.fd = listen_fd;
    if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, listen_fd, &ev) == -1) {
        perror("epoll_ctl: listen_fd");
        close(listen_fd);
        close(epoll_fd);
        exit(EXIT_FAILURE);
    }

    while (1) {
        // 等待事件发生
        int nfds = epoll_wait(epoll_fd, events, MAX_EVENTS, -1);
        if (nfds == -1) {
            perror("epoll_wait");
            break;
        }

        for (int i = 0; i < nfds; ++i) {
            if (events[i].data.fd == listen_fd) {
                // 处理新的连接
                socklen_t len = sizeof(cliaddr);
                conn_fd = accept(listen_fd, (struct sockaddr *)&cliaddr, &len);
                if (conn_fd == -1) {
                    perror("accept");
                    continue;
                }

                // 将新连接的 socket 添加到 epoll 实例中
                ev.events = EPOLLIN | EPOLLET; // 使用边缘触发模式
                ev.data.fd = conn_fd;
                if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, conn_fd, &ev) == -1) {
                    perror("epoll_ctl: conn_fd");
                    close(conn_fd);
                }
            } else {
                // 处理已连接 socket 上的 I/O 事件
                int fd = events[i].data.fd;
                char buffer[1024];
                ssize_t nread = read(fd, buffer, sizeof(buffer));
                if (nread == -1) {
                    if (errno == EAGAIN || errno == EWOULDBLOCK) {
                        // 非阻塞 I/O 时常见错误，继续循环
                        continue;
                    } else {
                        perror("read");
                        epoll_ctl(epoll_fd, EPOLL_CTL_DEL, fd, NULL);
                        close(fd);
                    }
                } else if (nread == 0) {
                    // 对端关闭连接
                    printf("Connection closed by peer\n");
                    epoll_ctl(epoll_fd, EPOLL_CTL_DEL, fd, NULL);
                    close(fd);
                } else {
                    // 处理读取到的数据
                    buffer[nread] = '\0';
                    printf("Received: %s\n", buffer);
                    // 简单回显数据
                    if (write(fd, buffer, nread) != nread) {
                        perror("write");
                    }
                }
            }
        }
    }

    close(listen_fd);
    close(epoll_fd);
    return 0;
}

在上述代码中：

1. 创建 epoll 实例：使用 epoll_create1 函数创建一个 epoll 实例，并返回一个文件描述符 epoll_fd。
2. 添加文件描述符到 epoll：通过 epoll_ctl 函数将监听 socket listen_fd 添加到 epoll 实例中，并指定监听的事件类型（这里是 EPOLLIN，表示可读事件）。
3. 等待事件发生：调用 epoll_wait 函数等待事件发生，该函数会阻塞，直到有事件发生或者超时。当有事件发生时，它会将发生事件的文件描述符及其相关事件信息填充到 events 数组中。
4. 处理事件：遍历 events 数组，判断事件对应的文件描述符是监听 socket 还是已连接 socket。如果是监听 socket，则处理新的连接请求；如果是已连接 socket，则处理该 socket 上的 I/O 事件（如读取和写入数据）。在处理已连接 socket 的 I/O 事件时，使用了非阻塞 I/O 的方式，通过检查 errno 判断是否需要继续等待。同时，代码中还展示了如何使用边缘触发（EPOLLET）模式，这种模式可以减少不必要的事件通知，进一步提升性能。