当前位置：首页 > news >正文

epoll的ET与LT模式深度对比：边缘触发的非阻塞I/O要求、饥饿问题与高吞吐场景的正确实现

news 2026/7/7 0:59:23

epoll的ET与LT模式深度对比：边缘触发的非阻塞I/O要求、饥饿问题与高吞吐场景的正确实现

一、两种触发模式的本质差异

epoll是Linux高性能I/O多路复用的核心机制。它有两种触发模式：

LT（Level Triggered，水平触发）：只要fd上有数据可读/可写，epoll_wait每次都会返回该fd。直到数据被完全处理，才不再通知
ET（Edge Triggered，边缘触发）：只在fd状态发生变化时通知一次。从无数据到有数据是一次边缘，从不可写到可写是一次边缘。处理完毕后不再重复通知

核心差异：LT是"有数据就通知"，ET是"数据来了通知一次"。ET模式要求程序在一次通知后必须将fd上的所有数据读完/写完，否则下一次数据到达前不会有新的通知。

模式对比时序图

sequenceDiagram participant App as 应用程序 participant Kernel as 内核epoll Note over App,Kernel: LT模式时序 Kernel->>App: fd_A可读（数据100字节） App->>Kernel: read(fd_A, 50字节) Kernel->>App: fd_A仍可读（剩余50字节） 再次通知 App->>Kernel: read(fd_A, 50字节) Kernel->>App: fd_A无数据 不再通知 Note over App,Kernel: ET模式时序 Kernel->>App: fd_A可读（数据100字节） 状态变化通知一次 App->>Kernel: read(fd_A, 50字节) Note over App: 必须继续读取！不再有通知 App->>Kernel: read(fd_A, 50字节) Note over App: 返回EAGAIN 数据已读完 App->>App: 注册fd_A等待下一次边缘 Note over App,Kernel: ET模式的错误使用 Kernel->>App: fd_A可读（数据100字节） App->>Kernel: read(fd_A, 50字节) Note over App: 未读完剩余数据 不再收到通知 Note over Kernel: 50字节滞留 直到新数据到达才触发下一次边缘

二、边缘触发的非阻塞I/O要求

ET模式必须使用非阻塞fd。原因：当程序循环读取fd直到返回EAGAIN时，如果fd是阻塞模式，最后一次read（缓冲区已空）会阻塞线程，整个事件循环卡死。

非阻塞fd的设置：

int flags = fcntl(fd, F_GETFL, 0); fcntl(fd, F_SETFL, flags | O_NONBLOCK);

ET模式的正确处理流程：

收到可读通知后，循环调用read直到返回EAGAIN
收到可写通知后，循环调用write直到返回EAGAIN或数据写完
EAGAIN表示缓冲区已空/满，当前轮次处理完毕
下一次边缘到来时才会再次通知

关键错误：在ET模式下只read一次就返回事件循环。这会导致数据滞留在内核缓冲区，直到fd上有新的数据到来触发下一次边缘，滞留数据才会被读取。对于低频数据源，滞留可能持续数秒甚至数分钟。

三、ET模式的饥饿问题

ET模式有一个容易被忽视的饥饿问题：当某个fd持续产生大量数据时，程序在一次通知中循环读取该fd，可能长时间占用事件循环，导致其他fd得不到处理。

LT模式天然避免饥饿：每次epoll_wait最多返回一个fd的可读状态，程序处理一部分后返回事件循环，其他fd在下一次epoll_wait中得到通知。

ET模式避免饥饿的策略：

限时读取：为每个fd的单次处理设置时间上限（如5ms），超时后注册fd等待下一次处理
限量读取：为每个fd的单次处理设置字节上限（如64KB），达到上限后返回事件循环
优先级队列：将fd按优先级分组，高优先级fd先处理，低优先级fd在空闲时处理

四、高吞吐场景的C服务器实现

以下代码实现一个支持ET模式的高吞吐TCP服务器，包含限时读取防饥饿和非阻塞I/O的正确处理：

#include <stdio.h> #include <stdlib.h> #include <string.h> #include <unistd.h> #include <errno.h> #include <fcntl.h> #include <sys/epoll.h> #include <sys/socket.h> #include <netinet/in.h> #include <arpa/inet.h> #include <time.h> #define MAX_EVENTS 1024 #define READ_TIMEOUT_MS 5 /* 单fd限时5ms防饥饿 */ #define MAX_READ_PER_ROUND 65536 /* 单fd单轮最大读取64KB */ #define BUFFER_SIZE 4096 static int set_nonblocking(int fd) { int flags = fcntl(fd, F_GETFL, 0); if (flags == -1) return -1; return fcntl(fd, F_SETFL, flags | O_NONBLOCK); } static void handle_read_et( int epfd, int fd, struct epoll_event *ev) { char buf[BUFFER_SIZE]; ssize_t total_read = 0; struct timespec start, now; clock_gettime(CLOCK_MONOTONIC, &start); /* 循环读取直到EAGAIN或达到防饥饿限制 */ while (1) { ssize_t n = read(fd, buf, sizeof(buf)); if (n > 0) { total_read += n; /* 防饥饿：单轮读取量上限 */ if (total_read >= MAX_READ_PER_ROUND) { /* 重新注册fd等待下一轮处理 */ epoll_ctl(epfd, EPOLL_CTL_MOD, fd, ev); return; } /* 防饥饿：单轮时间上限 */ clock_gettime(CLOCK_MONOTONIC, &now); long elapsed_ms = (now.tv_sec - start.tv_sec) * 1000 + (now.tv_nsec - start.tv_nsec) / 1000000; if (elapsed_ms >= READ_TIMEOUT_MS) { epoll_ctl(epfd, EPOLL_CTL_MOD, fd, ev); return; } } else if (n == -1) { if (errno == EAGAIN || errno == EWOULDBLOCK) { /* 数据已读完，ET模式正常结束 */ break; } perror("read error"); close(fd); epoll_ctl(epfd, EPOLL_CTL_DEL, fd, NULL); return; } else { /* n == 0: 客户端关闭连接 */ close(fd); epoll_ctl(epfd, EPOLL_CTL_DEL, fd, NULL); return; } } } int main(int argc, char *argv[]) { int port = argc > 1 ? atoi(argv[1]) : 8080; int listen_fd = socket(AF_INET, SOCK_STREAM, 0); if (listen_fd < 0) { perror("socket"); exit(1); } set_nonblocking(listen_fd); struct sockaddr_in addr = { .sin_family = AF_INET, .sin_port = htons(port), .sin_addr.s_addr = INADDR_ANY, }; bind(listen_fd, (struct sockaddr *)&addr, sizeof(addr)); listen(listen_fd, 512); int epfd = epoll_create1(0); struct epoll_event ev = { .events = EPOLLIN | EPOLLET, /* ET模式 */ .data.fd = listen_fd, }; epoll_ctl(epfd, EPOLL_CTL_ADD, listen_fd, &ev); struct epoll_event events[MAX_EVENTS]; printf("ET模式服务器启动，端口: %d\n", port); while (1) { int nfds = epoll_wait(epfd, events, MAX_EVENTS, -1); for (int i = 0; i < nfds; i++) { if (events[i].data.fd == listen_fd) { /* ET模式下必须循环accept直到EAGAIN */ while (1) { int client_fd = accept4( listen_fd, NULL, NULL, SOCK_NONBLOCK ); if (client_fd == -1) { if (errno == EAGAIN) break; perror("accept"); break; } struct epoll_event cev = { .events = EPOLLIN | EPOLLET, .data.fd = client_fd, }; epoll_ctl(epfd, EPOLL_CTL_ADD, client_fd, &cev); } } else { handle_read_et( epfd, events[i].data.fd, &events[i]); } } } }

关键设计决策：