一直以来，在Linux高性能网络编程中使用最多的网络I/O模型是epoll。epoll除了性能优越，稳定性也很高，似乎epoll成了高并发应用场景的唯一选择。然而随着io_uring的出现，这一局面慢慢被打破。 io_uring是异步I/O，同样性能优越，那么epoll和io_uring到底谁更牛呢？今天我们来深入探讨一下。 笔者将epoll重要的知识点都汇聚在一张图，如图1所示。epoll总共有三大系统调用：epoll_create、epoll_ctl、epoll_wait。
epoll_create用来在内核里创建一个eventpoll对象（epoll实例）并返回一个文件描述符（epfd），通过epfd我们能够快速访问eventpoll对象。eventpoll对象内核定义如下：
structeventpoll {wait_queue_head_t wq; // 等待队列structlist_headrdllist;// 就绪队列structrb_rootrbr;// 红黑树 ......};
eventpoll对象中有三个重要成员：
wq：等待队列，内核通过等待队列唤醒epoll线程。
rdllist：就绪队列，用于存储就绪事件。
rbr：红黑树，用于存储所有通过epoll_ctl注册的I/O事件。
epoll_ctl用于向epoll实例中添加、修改或删除需要监控的文件描述符及其事件，内核对于epoll事件的定义为struct epitem结构。
structepitem {union {structrb_noderbn;/* 红黑树节点*/ };structlist_headrdllink;/* 就绪队列节点*/structepoll_filefdffd;/* 文件描述符信息，包含fd和对应的file指针 */int nwait; structeventpoll *ep;/* 指向所属的eventpoll实例 */structepoll_eventevent;/* 通过epoll_ctl设置的epoll_event事件，包含文件描述符和感兴趣的事件类型 */ ......};
epitem对象（图1中简称epi）是内核epoll事件管理的基本单元。
用户程序调用epoll_ctl函数（操作类型指定为EPOLL_CTL_ADD）时，内核会创建epitem对象，然后将用户设置的epoll事件（包含文件描述符和事件类型）复制至epitem对象，再将epitem对象插入红黑树。
epitem对象插入红黑树后，内核需要激活事件驱动机制，即当事件就绪后，主动通知epoll。具体做法是创建一个socket等待队列项（struct eppoll_entry结构），eppoll_entry的base成员指向epitem对象，同时epoll_entry中注册一个回调函数ep_poll_callback（用于通知epoll事件就绪），最后将epoll_entry插入socket等待队列。
完成epoll事件注册后，用户程序调用epoll_wait函数获取epoll就绪事件，epoll_wait负责把就绪事件从内核搬到用户程序。内核会循环检测epoll就绪队列是否为空。如果就绪队列不为空，内核会将所有就绪事件拷贝至epoll_event数组（epoll_wait函数参数）；否则，epoll线程会创建一个epoll等待队列项并注册回调函数ep_autoremove_wake_function（用于唤醒epoll线程），等待队列项将插入epoll等待队列，随后内核切换epoll线程至阻塞状态。
数据到来时，数据经网卡->网络设备子系统->网络协议栈，最后存储至套接字接收缓冲区。一旦数据被套接字接收，则epoll读事件就绪。内核会轮询socket等待队列，并执行ep_poll_callback回调函数，该函数有两个作用：
将socket等待队列项中的epitem对象插入epoll就绪队列。
轮询epoll等待队列，执行ep_autoremove_wake_function回调函数，将epoll线程唤醒。
epoll线程被唤醒后，它会将epoll就绪队列中的就绪事件从内核拷贝至用户程序。用户程序检测到就绪事件后，就能够读写数据了。 下面，我们简单总结一下epoll的优缺点。
epoll的优点如下：
相对于epoll的优点，笔者更愿意来讨论epoll的缺点，如果我们能够准确地指出epoll的缺点，意味着我们对epoll有了非常深入的理解。
epoll的缺点为：epoll处理I/O事件时，伴随着大量的系统调用，并且epoll不能够批量地处理I/O事件。
io_uring是异步I/O模型，它的实现原理和同步I/O模型有很大的区别。我们同样通过一张图来学习io_uring，如图2所示。io_uring_setup：用于创建和初始化io_uring实例。
io_uring_enter：负责提交I/O请求和获取完成事件。
io_uring_register：用于将文件描述符、缓冲区等资源预先注册到io_uring实例中。
图2中并没有出现这三个系统调用，实际开发中，我们会使用liburing库，liburing库将io_uring中的三个系统调用进行了封装，提供了更加丰富和友好的编程接口。 用户程序调用io_uring_queue_init函数（内部封装了io_uring_setup系统调用），该函数将创建和初始化一个io_uring实例（struct io_ring_ctx），内核定义如下：structio_ring_ctx {structio_rings *rings;/* 指向完成队列CQ */structio_uring_sqe *sq_sqes;/* 指向提交队列SQ */ ......};
struct io_ring_ctx结构非常复杂，这里不展开讲解，我们需要关注两个重要成员：rings（CQ，完成队列）和sq_sqes（SQ，提交队列）。 CQ和SQ都是无锁环形队列。内核初始化io_uring实例时，会从内核直接映射区分配内存作为CQ和SQ。之所以从内核直接映射区分配内存，是为了方便用户程序执行mmap内存映射，确保用户程序和内核能够直接都能够访问这块内存区域。CQ和SQ是用户程序和内核数据传输的通道，借用这个通道，用户程序可以在不使用系统调用的情况下和内核实现数据交换。CQ和SQ采用无锁化编程，在高并发场景下，能够最大程度减少锁的开销。 内核维护了一张io_uring I/O操作表，这张表定义了io_uring所有的I/O操作，共50多种操作，定义如下：/* 内核io_uring I/O操作表 */conststructio_issue_def io_issue_defs[] = { ...... [IORING_OP_SENDMSG] = { .needs_file = 1, .unbound_nonreg_file = 1, .pollout = 1, .ioprio = 1, .async_size = sizeof(struct io_async_msghdr), .prep = io_sendmsg_prep, .issue = io_sendmsg,}, [IORING_OP_RECVMSG] = { .needs_file = 1, .unbound_nonreg_file = 1, .pollin = 1, .buffer_select = 1, .ioprio = 1, .async_size = sizeof(struct io_async_msghdr), .prep = io_recvmsg_prep, .issue = io_recvmsg,},......}
每个I/O操作都有一个操作码，内核根据操作码查找I/O操作表，获取I/O操作函数并执行。- 步骤1：用户程序从SQ中申请一个空闲的SQE，然后设置SQE中的I/O操作码和用户数据。
- 步骤2：用户程序调用io_uring_submit函数（内部封装了io_uring_enter系统调用）提交SQE，内核通过I/O操作码查询I/O操作表获取I/O操作函数并执行。I/O操作成功后，内核从CQ中申请一个CQE，将返回结果存储在CQE。
- 步骤3：用户程序检测CQ是否为空，如果CQ不为空，用户程序从CQ中获取CQE，并解析I/O操作返回结果。
最后，我们同样来看一下io_uring有哪些优缺点。- 用户程序和内核共享内存：SQ/CQ通过mmap映射，避免传统read/write的多次数据拷贝。
- 批处理：一次提交/收割多个I/O请求，减少系统调用次数。
- SQPOLL模式：内核线程（SQ线程）主动轮询SQ，进一步减少系统调用。
io_uring在设计上并没有很大的缺陷，由于io_uring是一种比较新的网络I/O模型，所以还存在一些bug，需要持续优化。 从表1可以看出，io_uring在数据拷贝和系统调用开销两个方面是优于epoll的，当然，这些只是理论上的分析。下面，我们通过一个测试来验证上述观点。- 客户端主机：AMD Ryzen 7 8845HS，8核16线程，主频3.8GHz，24G内存。
- 服务端主机：12th Gen Intel(R) Core(TM) i5-12500H，8核16线程，16G内存，主频4.1GHz。
TCP客户端程序创建50个线程，每个线程和TCP服务端建立20000个TCP连接，总共百万个TCP连接。每个TCP连接发送和接收100个数据包（数据包长度100字节）。/* TCP连接处理线程 */void *test_proc(void *arg){int pos = (int)(intptr_t)arg;char send_buffer[BUFFER_SIZE] = {0};char recv_buffer[BUFFER_SIZE] = {0};memset(send_buffer, 'a', sizeof(send_buffer));/* 每个线程建立20000个连接 */for (int i = 0; i < REQS_THREAD; i++) {int sockfd = do_connect(server_ip, server_port, pos);if (sockfd == -1) continue;/* 每个连接发送和接收100个数据包 */for (int j = 0; j < PACKETS_THREAD; j++) {int ret = send(sockfd, send_buffer, 100, 0);if (ret < 0) break; ret = recv(sockfd, recv_buffer, sizeof(recv_buffer), 0);if (ret < 0) break; } close(sockfd); }returnNULL;}intmain(int argc, char *argv[]){ server_ip = argv[1]; server_port = atoi(argv[2]);int start_pos = atoi(argv[3]);structtimevalstart, end; gettimeofday(&start, NULL);pthread_t th[THREADS];/* 创建50个线程 */for (int i = 0; i < THREADS; i++) {int ret = pthread_create(&th[i], NULL, test_proc, (void *)(intptr_t)(start_pos + i));if (ret != 0) { perror("pthread_create"); } }for (int i = 0; i < THREADS; i++) { pthread_join(th[i], NULL); } gettimeofday(&end, NULL);/* 统计百万连接处理时间 */double time_used = (end.tv_sec - start.tv_sec) * 1000.0 + (end.tv_usec - start.tv_usec) / 1000.0;int reqs = (REQS_THREAD * THREADS);double qps = reqs / (time_used / 1000.0);printf("sucess: %d, time_used: %.0f(ms), qps: %.2f\n", reqs, time_used, qps);return0;}
TCP服务端分为：epoll服务端和io_uring服务端。epoll服务端代码（只保留主线代码）如下：intmain(int argc, char *argv[]){char *local_ip = argv[1];unsignedshort local_port = atoi(argv[2]);/* 初始化TCP服务端 */int sockfd = init_server(local_ip, local_port);/* 创建和初始化epoll实例 */int epfd = epoll_create(100);/* epoll注册监听套接字事件 */ epoll_add_event(epfd, sockfd);while(1) {structepoll_eventevents[MAX_EVENTS] = {0};int timeout = 2000;int nfds = epoll_wait(epfd, events, MAX_EVENTS, timeout);if (nfds == 0) {continue; } elseif (nfds == -1) {break; }for (int i = 0; i < nfds; i++) {int fd = events[i].data.fd;if (events[i].events & EPOLLIN) {if (fd == sockfd) {/* 处理连接请求 */int new_sockfd = accept(sockfd, NULL, NULL);if (new_sockfd == -1) {continue; } set_nonblock(new_sockfd); epoll_add_event(epfd, new_sockfd); } else {char buf[BUF_LEN] = {0};int len = 0;int pos = 0;while (1) {/* 接收数据 */ len = recv(fd, buf+pos, ONCE_LEN, 0);if ((len == 0) || ((len == -1) && (errno != EAGAIN))) { close(fd); epoll_del_event(epfd, fd);break; } elseif ((len == -1) && (errno == EAGAIN)) {break; } else { pos += (len > 0 ? len : 0);break; } }if (pos > 0) {/* 发送数据 */int slen = send(fd, buf, pos, 0);if (slen < 0) { close(fd); epoll_del_event(epfd, fd);continue; } } } } else { } } } close(epfd); close(sockfd);return 0;}
io_uring服务端代码（只保留主线代码）如下：intmain(int argc, char *argv[]){char *server_ip = argv[1];unsignedshort server_port = atoi(argv[2]);int listen_fd = init_server(server_ip, server_port);structio_uringring;#if 1 /* 普通模式 */if (io_uring_queue_init(ENTRIES, &ring, 0) < 0) {return-1; }#else/* SQPOLL模式 */structio_uring_paramsparams = {0}; params.flags = IORING_SETUP_SQPOLL | IORING_SETUP_SQ_AFF; params.sq_thread_cpu = 1;if (io_uring_queue_init_params(ENTRIES, &ring, ¶ms) < 0) {return1; }#endif/* accept I/O请求插入SQ */ setup_accept(&ring, listen_fd);structio_uring_cqe *cqes[BATCH_SIZE];while (1) {/* 提交全部SQE */ io_uring_submit(&ring);/* 阻塞等待CQE */int ret = io_uring_wait_cqe(&ring, &cqes[0]);if (ret < 0) {break; }/* 批处理CQE */int ready = io_uring_peek_batch_cqe(&ring, cqes, BATCH_SIZE);if (ready == 0) {continue; }for (int i = 0; i < ready; i++) {structio_uring_cqe *cqe = cqes[i]; io_info* info = (io_info*)io_uring_cqe_get_data(cqe);switch (info->type) {case ACCEPT:if (cqe->res == -1) { close(info->client_fd);break; }/* 读数据I/O请求插入SQ */ setup_read(&ring, cqe->res, &info->client_addr);/* accept I/O请求插入SQ */ setup_accept(&ring, listen_fd);break;case READ:if (cqe->res > 0) { info->bytes_read = cqe->res; info->buffer[cqe->res] = '\0';/* 写数据I/O请求插入SQ */ setup_write(&ring, info->client_fd, info->buffer, info->bytes_read, &info->client_addr); } elseif (cqe->res <= 0) { close(info->client_fd); }break;case WRITE:if (cqe->res > 0) {/* 读数据I/O请求插入SQ */ setup_read(&ring, info->client_fd, &info->client_addr); } elseif (cqe->res <= 0) { close(info->client_fd); }break; } free_connection_info(info); }/* CQ推进 */ io_uring_cq_advance(&ring, ready); } close(listen_fd); io_uring_queue_exit(&ring);return0;}
TCP客户端分别和两个TCP服务端建立百万连接，并分别统计处理百万连接所花费的时间。同时通过strace -c命令统计epoll服务端和io_uring服务端系统调用情况。 每一列的解析如下：
% time：该系统调用耗时占总时间的百分比。
seconds：该系统调用总耗时（秒）。
usecs/call：每次调用平均耗时（微妙）。
calls：总调用次数。
errors：调用失败次数。
syscall：系统调用名称。
（1）epoll服务端测试结果
执行strace -c ./tcp_epoll 192.168.2.2 9999命令统计epoll服务端系统调用情况，统计结果如下：
epoll服务端总系统调用次数为211828279（超2亿次），除了发送和接收系统调用次数超百万次，其他系统调用（如epoll_wait、epoll_ctl）次数也有几十万次。完成百万连接测试用时632008毫秒。
（2）io_uring服务端测试结果
执行strace -c ./io_uring_server 192.168.2.2 9999命令统计io_uring服务端系统调用情况，统计结果如下：
开启批处理和SQ线程后，io_uring服务端总系统调用次数为28647744次（近3千万次），其中io_uring_enter系统调用次数占比为98%。百万连接测试时间为626473毫秒。
从测试结果可以看出，epoll服务端的系统调用次数远远大于io_uring服务端的系统调用次数。如果每次系统调用的处理时间约为1us，1千万次系统调用将耗时10秒，所以io_uring服务端完成百万连接测试用时相对较短。
文章中涉及到的epoll、io_uring以及百万并发相关的知识在我的新书《图解Linux网络编程》中有详细介绍，如果你想系统性地学习Linux网络编程，成为别人眼中的大佬，欢迎入手我的新书。
