ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

嵌入式Linux文件IO多路复用:select、poll、epoll原理与实战

嵌入式Linux文件IO多路复用:select、poll、epoll原理与实战

1. 从阻塞到并发:为什么文件IO最终都会走到I/O多路复用这一步

刚接触嵌入式Linux文件IO编程的人,几乎都会经历同一条路径:先学会open、read、write、close这几个最基本的系统调用,然后写一个读写文件或者串口的小程序,跑通了,觉得不过如此。接着开始碰网络编程,写一个TCP服务端,发现accept之后如果只用一个线程处理一个连接,第二个客户端连上来就没人理了。于是要么上多线程,要么上多进程,问题看似解决了,但资源开销和上下文切换的代价在嵌入式设备上很快就暴露出来——内存紧张、CPU主频不高、线程栈本身就要吃掉不少RAM。

这时候I/O多路复用就登场了。它要解决的核心问题非常明确:单个线程如何同时监控多个文件描述符,并在其中任意一个或几个变为“就绪”状态时得到通知,从而只处理真正有数据的那些描述符。注意这里的关键词是“就绪”,不是“读取”。多路复用本身不搬运数据,它只负责告诉你“现在可以读了”或者“现在可以写了”,真正的read/write还是你自己来。

在嵌入式场景下,这个机制的价值被进一步放大。嵌入式设备往往需要同时处理串口、网络、GPIO中断上报、传感器数据采集等多个IO源,如果每个IO源开一个线程,线程间的同步和资源竞争会让代码迅速变得难以维护。而select、poll、epoll这三套接口,恰好提供了一种“事件驱动”的编程范式,让一个主循环就能把多个IO源管得井井有条。

这篇文章我会把这三套接口从原理到实操完整拆一遍,重点讲清楚它们各自的数据结构、内核行为、性能差异,以及在嵌入式Linux下怎么选、怎么写、怎么避坑。如果你正在做嵌入式网关、数据采集器、串口服务器这类需要多路IO并发的项目,这篇内容应该能帮你少走不少弯路。

2. 三套接口的底层逻辑与选型考量

2.1 select:最古老也最通用的方案

select的接口长这样:

int select(int nfds, fd_set *readfds, fd_set *writefds, fd_set *exceptfds, struct timeval *timeout);

它的核心是fd_set这个位图结构。在大多数实现里,fd_set是一个固定大小的数组,每一位对应一个文件描述符。比如在32位系统上,fd_set通常是1024位,也就是最多能监控1024个描述符,这个限制由FD_SETSIZE宏定义。

使用select的典型流程是:先把关心的描述符用FD_SET加进集合,调用select阻塞等待,返回后遍历集合用FD_ISSET判断哪个就绪。这里有个非常容易踩的坑:select返回后,传入的fd_set会被内核修改,只保留就绪的描述符。也就是说,如果你要在循环里反复调用select,每次调用前都必须重新用FD_SET把集合构建一遍。我见过不少初学者写的代码,第一次循环正常,第二次就再也检测不到事件了,原因就是忘了重新填充集合。

另一个坑是nfds参数。它不是你监控的描述符个数,而是最大描述符值加一。比如你监控的描述符是3、5、8,那nfds应该传9。这个参数如果传错,内核可能扫描不到你关心的某些描述符。

select的优点是跨平台性极好,几乎所有的Unix-like系统和Windows都支持,代码可移植性强。缺点是三个:描述符数量受限、每次调用都要把整个集合从用户态拷贝到内核态、返回后需要线性遍历所有描述符来判断就绪状态。在描述符数量少、调用频率不高的场景下,这些缺点可以忽略;但在嵌入式设备上如果监控几十个描述符且事件频繁,select的O(n)遍历开销就会变得明显。

2.2 poll:去掉数量限制的改进版

poll的接口是这样的:

int poll(struct pollfd *fds, nfds_t nfds, int timeout); struct pollfd { int fd; short events; short revents; };

相比select,poll用了一个pollfd数组来代替位图,每个元素包含描述符、关心的事件和返回的事件。这样就没有了1024的数量限制,理论上能监控多少描述符只取决于内存。而且events和revents是分开的,每次调用不需要重新构建整个集合,内核只修改revents字段,events保持不变,这一点比select友好很多。

poll支持的事件类型也更丰富,除了POLLIN、POLLOUT,还有POLLPRI(紧急数据)、POLLERR、POLLHUP、POLLNVAL等。其中POLLERR、POLLHUP、POLLNVAL是内核自动设置的,不需要你在events里指定,但返回时一定要检查,否则可能漏掉错误状态导致死循环。

poll的缺点和select类似:每次调用仍然需要把整个pollfd数组从用户态拷贝到内核态,返回后仍然需要遍历数组判断revents。所以它的时间复杂度还是O(n),只是常数比select小一些,且没有数量上限。在嵌入式Linux下,如果描述符数量在几十到一两百这个量级,poll是一个非常稳妥的选择,代码比select清晰,兼容性也好。

2.3 epoll:为高并发而生的现代方案

epoll是Linux特有的机制,接口由三个函数组成:

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);

epoll的设计思路和select/poll完全不同。它把“注册关心的描述符”和“等待事件”这两件事拆开了。epoll_ctl负责增删改监控的描述符,内核会在内部维护一棵红黑树来管理这些描述符;epoll_wait只负责等待和返回就绪的事件,不需要每次都传入全部描述符。

这个设计带来的好处是:每次epoll_wait不需要把全部描述符从用户态拷贝到内核态,因为描述符已经在epoll_ctl阶段注册到内核了。返回时,内核只把就绪的事件填充到events数组里,用户程序直接遍历这个数组即可,数组长度就是就绪事件的数量,不需要遍历全部监控的描述符。所以epoll的时间复杂度是O(1)(准确说是O(就绪数量)),在描述符数量大、活跃连接比例低的场景下优势极其明显。

epoll支持两种触发模式:水平触发(LT)和边缘触发(ET)。水平触发是默认模式,只要描述符还有数据可读,每次epoll_wait都会返回它;边缘触发则只在状态发生变化时通知一次,如果你没把数据读完,后续不会再通知。ET模式配合非阻塞IO可以减少epoll_wait的调用次数,提高效率,但编程难度更高,稍不注意就会丢数据或卡死。嵌入式场景下我一般建议先用LT模式把逻辑跑通,确认稳定后再考虑是否切换到ET。

2.4 三套接口的对比与选型建议

特性selectpollepoll
描述符数量限制有(通常1024)无无
每次调用是否拷贝全部描述符是是否
返回后遍历方式遍历全部遍历全部只遍历就绪
时间复杂度O(n)O(n)O(就绪数)
触发模式LTLTLT/ET
跨平台性好较好Linux特有
嵌入式适用场景描述符少、需移植中等数量、代码清晰大量描述符、高并发

选型上我的经验是:如果项目只在Linux上跑,描述符数量可能超过几十个,直接上epoll;如果需要兼顾可移植性,或者描述符就几个,poll比select更值得用,因为代码更干净;select现在基本只在维护老代码或者需要跨平台到非Linux系统时才会选。

3. 核心细节解析与实操要点

3.1 文件描述符就绪到底意味着什么

很多人对“就绪”的理解是模糊的。对于读操作,描述符就绪意味着一次read调用不会被阻塞,但不代表read一定能读到你要的那么多字节。比如一个TCP连接,内核接收缓冲区里有10个字节,你read的时候传了100字节的缓冲区,read会立刻返回10,而不是等满100。对于写操作,描述符就绪意味着内核发送缓冲区有空间可写,但能写多少也不确定。

这个细节直接影响到编程模型。如果你用阻塞IO配合多路复用,那么在一次事件通知后,你只能保证“读一次或写一次不会阻塞”,不能保证“读完所有数据”。所以正确的做法通常是:把描述符设置为非阻塞,然后在事件通知后循环读写直到返回EAGAIN或EWOULDBLOCK。这一点在ET模式下是必须的,在LT模式下虽然不强制,但为了效率也推荐这么做。

设置非阻塞的方法是在open时加O_NONBLOCK标志,或者用fcntl动态设置:

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

嵌入式Linux下,串口、socket、管道都支持非阻塞模式,但要注意某些字符设备驱动可能对非阻塞支持不完善,需要实测确认。

3.2 epoll的事件结构体与数据传递

epoll_event结构体里有一个data联合体,通常用来存放文件描述符或者指针:

struct epoll_event { uint32_t events; epoll_data_t data; }; typedef union epoll_data { void *ptr; int fd; uint32_t u32; uint64_t u64; } epoll_data_t;

这个data字段是epoll非常实用的一个设计。你可以把描述符存进去,epoll_wait返回后直接从events[i].data.fd拿到描述符,不需要额外维护映射关系。如果描述符和某个上下文结构体关联,也可以把结构体指针存进data.ptr,这样事件到来时直接就能拿到完整上下文,省去了一次查找。我在做串口服务器时,就是把每个串口的配置结构体指针存进data.ptr,事件一到直接操作对应结构体,代码非常简洁。

3.3 超时参数的设置技巧

三套接口都有超时参数,select用struct timeval,poll和epoll_wait用毫秒整数。超时设置有几个实践要点:

  • 传NULL或-1表示永久阻塞,适合纯事件驱动的场景,但嵌入式设备上如果没有任何事件,主循环就完全停住了,可能影响看门狗喂狗或其他周期性任务。
  • 传0表示立即返回,适合轮询模式,但会占满CPU,一般不推荐。
  • 传一个合理的小值,比如100毫秒或500毫秒,既能及时响应事件,又能让主循环有机会处理其他周期性任务。这是嵌入式场景下我最常用的方式。

需要注意的是,select的timeout在返回后会被修改为剩余时间,如果要在循环里复用同一个timeval结构,每次都得重新赋值。poll和epoll_wait的timeout是值传递,不会被修改,用起来更省心。

3.4 信号中断与EINTR处理

在嵌入式Linux下,如果程序注册了信号处理函数,select、poll、epoll_wait可能会被信号中断,返回-1并设置errno为EINTR。这不是错误,只是系统调用被打断了。正确的处理方式是判断errno == EINTR后重新调用,而不是当成致命错误退出。

int ret; do { ret = epoll_wait(epfd, events, MAX_EVENTS, timeout); } while (ret < 0 && errno == EINTR);

这个细节在嵌入式设备上尤其重要,因为很多设备会用SIGALRM做定时器,或者用SIGTERM做优雅退出,信号中断是常态。我见过有人的程序跑一段时间就莫名退出,排查半天发现就是没处理EINTR。

4. 实操过程与核心环节实现

4.1 用poll实现一个多路串口采集程序

假设我们要做一个嵌入式数据采集器,同时监控4个串口,把收到的数据转发到网络。用poll实现的主循环大概是这样:

#include <poll.h> #include <fcntl.h> #include <unistd.h> #include <stdio.h> #include <string.h> #include <errno.h> #define MAX_PORTS 4 #define BUF_SIZE 256 int main(void) { int fds[MAX_PORTS]; struct pollfd pfd[MAX_PORTS]; char buf[BUF_SIZE]; int i, ret; /* 打开4个串口,设置为非阻塞 */ for (i = 0; i < MAX_PORTS; i++) { char path[32]; snprintf(path, sizeof(path), "/dev/ttyS%d", i); fds[i] = open(path, O_RDWR | O_NOCTTY | O_NONBLOCK); if (fds[i] < 0) { perror("open serial"); return -1; } pfd[i].fd = fds[i]; pfd[i].events = POLLIN; pfd[i].revents = 0; } while (1) { ret = poll(pfd, MAX_PORTS, 500); if (ret < 0) { if (errno == EINTR) continue; perror("poll"); break; } if (ret == 0) continue; /* 超时,处理其他周期任务 */ for (i = 0; i < MAX_PORTS; i++) { if (pfd[i].revents & POLLIN) { ssize_t n; while ((n = read(pfd[i].fd, buf, sizeof(buf))) > 0) { /* 处理数据,比如转发到网络 */ process_data(i, buf, n); } if (n < 0 && errno != EAGAIN && errno != EWOULDBLOCK) { perror("read"); } } if (pfd[i].revents & (POLLERR | POLLHUP | POLLNVAL)) { /* 处理异常,比如重新打开串口 */ handle_error(i); } } } for (i = 0; i < MAX_PORTS; i++) close(fds[i]); return 0; }

这段代码有几个关键点值得展开。第一,串口打开时加了O_NONBLOCK,这样read在没数据时返回EAGAIN而不是阻塞。第二,poll的超时设了500毫秒,这样即使没有串口数据,主循环也能定期醒来处理其他任务。第三,read用了while循环,一次事件通知后把缓冲区里的数据全部读干净,避免数据积压。第四,检查了POLLERR、POLLHUP、POLLNVAL,这些错误状态如果不处理,可能导致描述符一直返回就绪但读不出数据,形成忙循环。

4.2 用epoll重构为高并发网络服务

如果采集器还要同时接受多个客户端连接,描述符数量可能上升到几十上百个,这时候epoll更合适。下面是一个简化的epoll服务端框架:

#include <sys/epoll.h> #include <sys/socket.h> #include <netinet/in.h> #include <fcntl.h> #include <unistd.h> #include <stdio.h> #include <string.h> #include <errno.h> #define MAX_EVENTS 64 #define BUF_SIZE 1024 static int set_nonblock(int fd) { int flags = fcntl(fd, F_GETFL, 0); if (flags < 0) return -1; return fcntl(fd, F_SETFL, flags | O_NONBLOCK); } int main(void) { int listen_fd, epfd, nfds, i; struct epoll_event ev, events[MAX_EVENTS]; struct sockaddr_in addr; listen_fd = socket(AF_INET, SOCK_STREAM, 0); if (listen_fd < 0) { perror("socket"); return -1; } int opt = 1; setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)); memset(&addr, 0, sizeof(addr)); addr.sin_family = AF_INET; addr.sin_addr.s_addr = htonl(INADDR_ANY); addr.sin_port = htons(8080); if (bind(listen_fd, (struct sockaddr *)&addr, sizeof(addr)) < 0) { perror("bind"); return -1; } if (listen(listen_fd, 16) < 0) { perror("listen"); return -1; } set_nonblock(listen_fd); epfd = epoll_create1(0); if (epfd < 0) { perror("epoll_create1"); return -1; } ev.events = EPOLLIN; ev.data.fd = listen_fd; if (epoll_ctl(epfd, EPOLL_CTL_ADD, listen_fd, &ev) < 0) { perror("epoll_ctl add listen"); return -1; } while (1) { nfds = epoll_wait(epfd, events, MAX_EVENTS, 500); if (nfds < 0) { if (errno == EINTR) continue; perror("epoll_wait"); break; } for (i = 0; i < nfds; i++) { int fd = events[i].data.fd; if (fd == listen_fd) { /* 接受新连接 */ while (1) { int conn_fd = accept(listen_fd, NULL, NULL); if (conn_fd < 0) { if (errno == EAGAIN || errno == EWOULDBLOCK) break; perror("accept"); break; } set_nonblock(conn_fd); ev.events = EPOLLIN | EPOLLET; ev.data.fd = conn_fd; if (epoll_ctl(epfd, EPOLL_CTL_ADD, conn_fd, &ev) < 0) { perror("epoll_ctl add conn"); close(conn_fd); } } } else { /* 处理客户端数据 */ char buf[BUF_SIZE]; ssize_t n; while ((n = read(fd, buf, sizeof(buf))) > 0) { process_client_data(fd, buf, n); } if (n == 0) { /* 对端关闭 */ epoll_ctl(epfd, EPOLL_CTL_DEL, fd, NULL); close(fd); } else if (n < 0 && errno != EAGAIN && errno != EWOULDBLOCK) { perror("read"); epoll_ctl(epfd, EPOLL_CTL_DEL, fd, NULL); close(fd); } } } } close(listen_fd); close(epfd); return 0; }

这段代码用了EPOLLET边缘触发模式,所以accept和read都必须循环到EAGAIN为止,否则会丢事件。listen_fd本身也设了非阻塞,这样accept在连接队列空的时候返回EAGAIN,循环才能正常退出。如果listen_fd是阻塞的,accept在队列空时会阻塞住,整个事件循环就卡死了,这是ET模式下非常经典的坑。

4.3 参数计算与缓冲区大小选择

缓冲区大小不是随便定的。以串口为例,假设波特率是115200,8N1格式,那么每秒最多传输115200/10 = 11520字节。如果主循环500毫秒跑一次,两次之间最多积累约5760字节。所以串口读缓冲区至少要能容纳这个量级,否则一次读不完,数据会留在内核缓冲区里,虽然不会丢,但会增加事件通知次数。实际项目中我一般给串口读缓冲区设1024到4096字节,配合循环读取,基本不会出问题。

网络缓冲区则要考虑MTU。以太网MTU通常是1500字节,TCP会把数据切成MSS大小的段,所以读缓冲区设2048或4096比较合适,能一次读完一个完整的数据段。如果设得太小,比如64字节,一个1500字节的包要读二十几次,系统调用开销会明显上升。

epoll_wait的maxevents参数也要注意。它表示一次最多返回多少个就绪事件。如果设得太小,比如1,那么即使有10个描述符就绪,一次也只返回1个,需要多次调用才能处理完,效率低。一般设成64或128比较合理,既能一次处理一批事件,又不会占用太多栈空间。events数组是在栈上分配的,maxevents乘以sizeof(struct epoll_event)就是栈占用,64个事件大约占1KB左右,嵌入式设备完全能承受。

5. 常见问题与排查技巧实录

5.1 常见问题速查表

现象可能原因排查方法解决方案
select第二次循环检测不到事件fd_set被内核修改后未重新填充检查循环内是否重新FD_SET每次调用select前重建集合
poll一直返回就绪但read无数据描述符有错误状态未处理检查revents是否含POLLERR/POLLHUP处理错误状态并关闭或重开描述符
epoll ET模式丢数据read未循环到EAGAIN检查read是否只调用一次循环read直到EAGAIN
epoll_wait返回EINTR导致退出未处理信号中断检查errno是否为EINTR循环重试而非退出
CPU占用100%超时设为0或描述符一直就绪用top观察,检查超时参数设合理超时,处理错误状态
accept阻塞导致事件循环卡死listen_fd未设非阻塞检查listen_fd的O_NONBLOCK设置非阻塞并循环accept
描述符泄漏关闭连接时未epoll_ctl DEL用lsof查看描述符数量关闭前先DEL再close

5.2 一个真实的排查案例

之前有个项目,设备跑几个小时后就出现CPU占用飙升,但功能还正常。用top看,进程CPU占用从平时的5%涨到90%多。加了日志后发现,epoll_wait返回的事件里,有一个描述符反复返回EPOLLIN,但read返回0。read返回0在TCP里表示对端关闭了连接,但代码里只处理了read返回-1的情况,返回0时没有关闭描述符,导致这个描述符一直留在epoll集合里,每次epoll_wait都返回它,形成忙循环。

修复方法很简单,在read返回0时执行epoll_ctl(EPOLL_CTL_DEL)并close。这个案例说明,多路复用编程里,描述符的生命周期管理必须和事件处理严格对应,任何一条路径漏掉清理,都可能造成资源泄漏或忙循环。

5.3 独家避坑技巧

第一个技巧:用epoll的data.ptr存上下文指针,而不是data.fd。当描述符关闭后,如果事件数组里还有它的事件,用data.fd去操作可能操作到已经被复用的新描述符,造成难以排查的bug。用指针的话,可以在关闭描述符时把上下文标记为无效,事件处理时先检查有效性,安全性更高。

第二个技巧:给每个描述符维护一个状态机。多路复用只告诉你“可读”或“可写”,但业务逻辑往往需要知道“当前处于哪个阶段”。比如一个协议解析,可能需要先读头部再读body,用一个状态机记录每个连接的解析进度,比用一堆全局变量清晰得多。

第三个技巧:在嵌入式设备上,把epoll_wait的超时和看门狗喂狗结合起来。超时设500毫秒,每次超时返回时喂一次狗,这样既保证了事件响应,又保证了看门狗不会误复位。如果超时设成-1永久阻塞,一旦没有事件,看门狗就可能超时复位,设备反复重启。

第四个技巧:测试时用strace观察系统调用。strace -e trace=epoll_wait,epoll_ctl,read,write ./your_program能清楚看到每次epoll_wait返回了哪些事件、read读了多少字节。排查丢数据、忙循环这类问题时,strace比加日志更直接。

5.4 水平触发与边缘触发的选择经验

LT和ET的选择没有绝对答案,但有几个判断依据。如果代码逻辑简单,对性能要求不高,用LT,不容易出错。如果描述符数量大、事件频繁,且你能保证每次事件都把数据读干净,用ET能减少epoll_wait的调用次数,降低CPU占用。我的一般做法是:先用LT把功能跑通,压测时如果发现epoll_wait调用过于频繁成为瓶颈,再切换到ET并仔细检查所有读写路径是否都循环到了EAGAIN。

需要注意的是,ET模式下accept也必须循环。因为多个连接可能在同一时刻到达,epoll_wait只通知一次,如果accept只调用一次,剩下的连接就会一直留在队列里,直到下一个新连接到来才会被处理,造成连接延迟。这个坑我在早期项目里踩过,现象是客户端连接偶尔要等好几秒才建立,排查了很久才定位到。

6. 从文件IO到事件驱动的思维转变

写完这几套接口的代码,你会发现一个有意思的事情:select、poll、epoll本质上都是对“文件描述符”这个抽象的统一操作。在Linux里,串口是文件、socket是文件、管道是文件、甚至某些设备也是文件,所以同一套多路复用机制能同时管理它们。这种“一切皆文件”的设计,让嵌入式开发者可以用统一的思维模型处理各种IO源,不用为每种设备学一套不同的并发模型。

从阻塞式编程转到事件驱动编程,最大的思维转变是:你不再主动去“读”,而是等系统告诉你“可以读了”。这个转变刚开始会有点别扭,但一旦适应,代码结构会变得非常清晰——一个主循环,一个事件分发,每个描述符的处理逻辑独立成函数,扩展新IO源只需要加一个描述符和对应的处理函数,不用改主循环结构。

我在实际项目中的体会是,嵌入式Linux下的IO并发,epoll配合非阻塞IO基本能覆盖90%以上的场景。剩下的10%里,有些是描述符极少、用poll更简单的,有些是需要跨平台、只能用select的。把这三套接口都掌握,遇到不同需求时就能快速做出选择,而不是硬套一种方案。

最后分享一个小技巧:如果你不确定该用LT还是ET,可以先写一个测试程序,用strace统计两种模式下epoll_wait的调用次数和read的调用次数。在同样的数据量下,ET模式的epoll_wait调用次数通常明显少于LT,但read次数可能更多(因为要循环读到EAGAIN)。根据你的实际数据特征,选择总系统调用次数更少的那种模式,通常就是更优解。

RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进