资讯详情

资讯详情

Linux驱动开发中无名管道的原理、应用与避坑指南

聊到 Linux 驱动开发很多人第一反应是寄存器、中断、file_operations但真正在项目里跑起来你会发现“无名管道”这四个字出现的频率高得吓人。这玩意儿既不算驱动代码里必须出现的结构又总在测试、数据分发、日志转发时跳出来。今天我就以一次常见的驱动验证场景为例把无名管道在驱动开发中的作用、原理和坑一次讲明白。适合正在学驱动、或者被进程间通信折腾过的朋友。1. 为什么驱动开发绕不开无名管道1.1 无名管道是什么从系统调用说起无名管道在 Linux 里的标准说法是 pipe由pipe()系统调用创建。调用一下就返回两个文件描述符一个专门用来读一个专门用来写#include unistd.h int fd[2]; if (pipe(fd) ! 0) { perror(pipe); exit(1); } // fd[0] 是读端fd[1] 是写端之后write(fd[1], ...)把数据送进管道read(fd[0], ...)把数据取出来。这个交互过程可以理解为两个人共用一根吸管一个人往管子里塞另一个人从另一头吸数据只能往一个方向跑。“无名”这两个字很关键。它不像普通文件那样有一个 /tmp/xxx 之类的路径名你拿不到它的文件路径只能通过pipe()返回的文件描述符去操作。也正因为没有名字它不能像 FIFO 那样被两个互不相关的进程通过路径打开。无名管道的生命周期一般只存在于创建它的进程以及 fork 出来的子进程之间。在实际的驱动开发里我用到无名管道最多的地方就是测试程序。比如我写了一个字符设备驱动用户态程序打开/dev/xxx读数据然后需要把这些数据同时交给两个不同的处理模块。最省事的方式就是创建无名管道、fork 子进程让父子进程各管一头。管道本身不涉及驱动代码但它把驱动读出来的数据流转了起来能非常快地验证驱动的读写逻辑对不对。1.2 驱动开发里到底在哪用到无名管道很多人会把“驱动开发”想得很封闭觉得只要写内核模块、处理硬件中断就行了。但实际的驱动开发流程里用户态测试程序占的时间一点不比内核代码少。而无名管道就是用户态测试程序里最常见的 IPC 手段之一。驱动开发中常见的无名管道场景有三个第一是设备数据分发。驱动从硬件读回来一批数据父进程作为“总控”把数据写入管道若干个子进程从管道里拿数据去做各自的处理。比如一个传感器驱动需要同时做实时波形绘制、日志记录、阈值告警三个功能拆成三个子进程父进程只要往管道里扔数据就行。第二是驱动并发压力测试。驱动最难调的就是并发访问多个进程同时 read、write、ioctl。这时候无名管道可以作为任务分发通道父进程创建多个 worker 子进程每个 worker 去并发 open 设备、执行不同操作再通过管道把结果回传给父进程。比用线程模拟更接近真实场景。第三是日志和事件通知。驱动测试程序经常需要把底层上报的事件转发给另一个进程处理管道是最轻量的选择。不需要 IP、不需要端口、不需要 socket两个 fd 搞定。一个容易误导新人的点是有人会觉得“既然是驱动开发是不是应该在内核里创建管道”。答案是否定的。无名管道是用户态 IPC 工具驱动代码里不需要也不应该去直接操作别人进程的管道 fd。正确的分层是驱动只负责采集和传输设备数据用户态程序负责用管道把数据分发到各个消费者手里。这样数据结构清晰也避免在内核态做一堆本没必要的事。1.3 顺着实现看一眼内核里的管道虽然管道是用户态接口但它底层仍然是一套文件系统叫 pipefs。每个管道在创建时会在 pipefs 里分配一个 inode这个 inode 没有路径名只有一个编号。内核里对应结构是struct pipe_inode_info里面维护了一个环形缓冲区由若干struct pipe_buffer组成。pipe_buffer并不是简单地存一段连续内存而是一页一页的页面引用。每个pipe_buffer指向一个内存页存储一段数据。写入方把数据塞到某个页面里读端把数据取走之后这块缓冲区又可以被复用。这样做的好处是拷贝次数少而且在某些场景下还能用零拷贝特性比如splice()直接发送页面数据。这个环形缓冲区默认容量多大Linux 下默认一般是 64KB也就是 65536 字节。你可能看到网上有文章说“管道缓冲区只有 4096 字节”那说的其实是 POSIX 标准里write()原子写的上限PIPE_BUF。具体能一次写多少、缓冲区大小怎么改后面第 4 节我会单独讲。理解这层结构对驱动开发者最大的价值在于管道的读写本质上就是文件操作它有自己的file_operations内核在调用read()、write()时会走pipe_read()、pipe_write()。这和你写的字符设备驱动的回调函数是同一个框架体系。顺着管道实现去读内核代码比空看“字符设备文件操作”要直观得多。2. 动手前搭一个“驱动 管道”实验环境2.1 环境准备和内核模块编译光讲原理不过瘾我习惯直接跑一个可复现的实验。先说明环境我的测试机是 x86_64 架构的 Linux内核版本在 6.x 左右。你不需要完全一致只要内核能够编译模块就行。需要准备的东西有三个Linux 内核头文件必须和当前运行内核版本对应。编译器 gcc 和 make。root 权限因为加载内核模块需要insmod。以 Ubuntu/Debian 系为例sudo apt update sudo apt install build-essential linux-headers-$(uname -r)安装完成后用uname -r确认版本然后看一眼目录是否存在uname -r ls /lib/modules/$(uname -r)/build如果这个目录能进去说明内核头文件没问题。我建议在虚拟机里做实验因为加载一个不成熟的内核模块有概率把系统搞得不稳定。虚拟机出问题直接回滚快照比在主力开发机上反复重启舒服太多。2.2 准备一个最小字符设备驱动为了让无名管道有用武之地我先写一个最简单的模拟数据采集驱动。它不操作真实硬件每次 read 就返回一个递增的 8 字节整数模拟设备产生数据。我用 misc 设备框架注册这样不用手动分配主设备号内核会自动在 /dev 下创建节点。#include linux/module.h #include linux/miscdevice.h #include linux/fs.h #include linux/uaccess.h #define DEMO_DEV_NAME demo_pipe_dev static u64 fake_value; static DEFINE_MUTEX(demo_lock); static ssize_t demo_read(struct file *file, char __user *buf, size_t count, loff_t *ppos) { size_t n min(count, sizeof(fake_value)); int ret; mutex_lock(demo_lock); ret copy_to_user(buf, fake_value, n); if (ret 0) fake_value; mutex_unlock(demo_lock); return ret ? -EFAULT : (ssize_t)n; } static const struct file_operations demo_fops { .owner THIS_MODULE, .read demo_read, }; static struct miscdevice demo_misc { .minor MISC_DYNAMIC_MINOR, .name DEMO_DEV_NAME, .fops demo_fops, }; static int __init demo_init(void) { return misc_register(demo_misc); } static void __exit demo_exit(void) { misc_deregister(demo_misc); } module_init(demo_init); module_exit(demo_exit); MODULE_LICENSE(GPL);这个驱动只实现了 read没有实现 write 和 ioctl。demo_read()里用copy_to_user()把内核里的fake_value拷贝到用户态缓冲区每次成功后让值递增。这样用户态每次读到的都是不同的数据能够直观看到数据在流动。有一个细节要提醒copy_to_user()返回的是“还有多少字节没拷贝成功”不是“拷贝了多少字节”。所以ret 0才表示完整拷贝。如果返回非 0直接返回-EFAULT给用户态用户态 read 就会得到 -1errno 是 EFAULT。2.3 Makefile 与加载、卸载内核模块编译靠 Makefile这是一个典型模板obj-m demo_pipe_dev.o KDIR : /lib/modules/$(shell uname -r)/build PWD : $(shell pwd) all: $(MAKE) -C $(KDIR) M$(PWD) modules clean: $(MAKE) -C $(KDIR) M$(PWD) clean把驱动源码保存为 demo_pipe_dev.c然后在同一目录下执行make顺利的话会生成 demo_pipe_dev.ko。加载前最好看一眼 dmesg 是否干净避免一些无关报错干扰判断sudo insmod demo_pipe_dev.ko ls /dev/demo_pipe_dev dmesg | tail如果 /dev/demo_pipe_dev 存在说明设备节点创建成功。验证一下读取sudo cat /dev/demo_pipe_dev | xxd会看到每次输出 8 字节而且数值不断递增。这说明驱动的 read 通路是通的后面就可以放心地让用户态程序用管道去分发这些数据。用完记得卸载sudo rmmod demo_pipe_dev3. 用无名管道把设备数据分发给子进程3.1 管道布局一次 fork两个方向现在进入正题用无名管道把驱动数据分发给一个子进程。因为管道是单向的如果要父子进程互相通信一个管道不够必须创建两个。我习惯管它们叫 pipe1 和 pipe2数据流如下驱动设备 fd - 父进程 - pipe1 - 子进程 ^ | |____________________________| pipe2pipe1 用来让父进程把设备数据发给子进程。pipe2 用来让子进程把处理结果回传给父进程。fork 之后两个进程手里其实各有 pipe1 和 pipe2 的全部 fd。这里有一个新手必踩的坑如果不关闭自己用不到的那一端数据流永远不会按预期结束。父进程写完 pipe1 后关闭写端子进程 read 才会返回 0如果父进程还攥着 pipe1 的写端不放子进程就会一直阻塞等待。3.2 用户态程序完整代码下面这个程序是完整的实验代码。父进程打开驱动设备循环读 20 次每次 8 字节写入 pipe1。子进程不断从 pipe1 读数据累加求和每读 5 条就往 pipe2 回传一次汇总结果。父进程最后把 pipe2 里的结果打印出来。#include stdio.h #include stdlib.h #include string.h #include unistd.h #include fcntl.h #include sys/types.h #include sys/wait.h #include errno.h #define DEV_PATH /dev/demo_pipe_dev int main(void) { int devfd; int pipe1[2], pipe2[2]; pid_t pid; unsigned char buf[8]; int i; devfd open(DEV_PATH, O_RDONLY); if (devfd 0) { perror(open DEV_PATH); return 1; } if (pipe(pipe1) ! 0 || pipe(pipe2) ! 0) { perror(pipe); return 1; } pid fork(); if (pid 0) { perror(fork); return 1; } if (pid 0) { /* 子进程 */ close(pipe1[1]); /* 子进程不写 pipe1 */ close(pipe2[0]); /* 子进程不读 pipe2 */ long long sum 0; int cnt 0; unsigned char rbuf[8]; ssize_t n; while ((n read(pipe1[0], rbuf, sizeof(rbuf))) 0) { long long v; if (n ! (ssize_t)sizeof(v)) continue; memcpy(v, rbuf, sizeof(v)); sum v; cnt; if (cnt % 5 0) dprintf(pipe2[1], sum%lld after %d reads\n, sum, cnt); } close(pipe1[0]); close(pipe2[1]); _exit(0); } /* 父进程 */ close(pipe1[0]); /* 父进程不读 pipe1 */ close(pipe2[1]); /* 父进程不写 pipe2 */ for (i 0; i 20; i) { ssize_t len read(devfd, buf, sizeof(buf)); if (len ! (ssize_t)sizeof(buf)) { fprintf(stderr, device read failed: %zd\n, len); break; } if (write(pipe1[1], buf, sizeof(buf)) ! (ssize_t)sizeof(buf)) { perror(write pipe1); break; } } /* 父进程不再写 pipe1子进程才能读到 EOF */ close(pipe1[1]); close(devfd); /* 回读子进程的汇总结果 */ { char out[128]; ssize_t r; while ((r read(pipe2[0], out, sizeof(out) - 1)) 0) { out[r] \0; fputs(out, stdout); } close(pipe2[0]); } wait(NULL); return 0; }代码里的注释已经标出了每个 close 的作用。我特别强调父进程在写完所有数据后close(pipe1[1])这一步只有当 pipe1 的所有写端都被关闭子进程的 read 才会返回 0从而退出循环。如果漏掉这一步子进程会一直卡在 read 上程序看起来就像“死掉了”。3.3 编译运行看效果编译运行gcc -o pipe_demo pipe_demo.c sudo insmod demo_pipe_dev.ko ./pipe_demo预期输出如下sum10 after 5 reads sum45 after 10 reads sum105 after 15 reads sum190 after 20 reads为什么是这个数驱动里 fake_value 从 0 开始每读一次加 1。父进程连续读到的 20 个数是 0 到 19子进程前 5 个数求和就是 0123410前 10 个数求和是 45前 15 个是 105全部 20 个输出完正好是 190。这个实验虽然小但已经完整模拟了“设备数据生产 - 管道分发 - 子进程计算 - 结果回传”的闭环。实际项目里子进程可以是任何东西数据落盘、画面绘制、网络转发管道这一层不需要改动。4. 驱动与管道配合时的高频坑4.1 close 语义EOF 是怎么来的新手调试管道最大的困惑就是“程序明明没报错却卡住不动”。十次有八次是 fd 没有按预期关闭。管道的 EOF 规则很简单当且仅当某个管道的所有写端都被关闭读端 read 才会返回 0。读端所有 fd 都关闭时写端 write 会触发 SIGPIPE 信号默认直接终止进程。回到上面的代码fork 之后父进程必须关闭 pipe1[0]否则父进程也持有 pipe1 读端虽然不影响 EOF但属于不必要的 fd 泄漏。父进程必须关闭 pipe2[1]否则 pipe2 永远有写端存在父进程回读 pipe2 时永远不会读到 EOF。子进程必须关闭 pipe1[1]否则父进程写完数据后即使 close(pipe1[1])pipe1 还有一个写端在子进程手里子进程自己读自己手里的写端不释放永远读不到 EOF。子进程必须关闭 pipe2[0]否则子进程回写 pipe2 时系统认为还有读端存在不会触发 SIGPIPE。我调试时最常用的方法是用一个临时工具函数把 fork 后仍然留在进程里的不相关 fd 全部打印出来。只要发现读端或写端多了一个立刻就能定位到卡死原因。4.2 阻塞、非阻塞与 PIPE_BUF无名管道默认是阻塞模式。read 发现管道空就等着write 发现管道满也等着。这在很多场景是好事但驱动测试程序里如果某一路数据消费者处理慢生产者就可能被 write 卡住导致整个采集链路阻塞。解决办法是用fcntl()把写端或读端设置成非阻塞int flags fcntl(pipe1[1], F_GETFL); fcntl(pipe1[1], F_SETFL, flags | O_NONBLOCK);非阻塞模式下写管道如果缓冲区满write()直接返回 -1errno 为 EAGAIN。read 如果管道空同样返回 -1errno 也为 EAGAIN。这时候正确的姿势是用 poll/epoll 等待可写或可读而不是写一个白忙循环不断刷 EAGAIN白忙循环会让 CPU 飙到 100%。还有一件事必须分清PIPE_BUF和管道缓冲区容量是两码事。PIPE_BUF是 POSIX 保证原子写的最大长度Linux 上一般是 4096 字节。单次 write 长度不超过 PIPE_BUF 时内核保证写操作原子完成不会和多进程的其他写交错。超过这个长度多个进程同时写一个管道时数据可能交错。管道容量是内核缓冲区的总大小默认 64KB。可以用fcntl(fd, F_GETPIPE_SZ)查看用fcntl(fd, F_SETPIPE_SZ, size)调整。调整后可以通过 /proc/sys/fs/pipe-max-size 看到系统上限一般是 1MB。驱动数据量大的时候我通常会把它调大减少写端阻塞的概率。下面是这两个参数的对比参数含义Linux 常见默认值PIPE_BUF单次原子写上限4096 字节管道容量内核缓冲区总大小65536 字节系统管道最大容量F_SETPIPE_SZ 上限1048576 字节4.3 驱动 read 返回值和用户态读错位在实验里驱动每次固定返回 8 字节用户态也按 8 字节读问题不大。但真实驱动往往不是这样设备可能一次只来 4 字节也可能一次来 512 字节甚至一条消息被拆成两次中断上报。这时候如果用户态程序“想当然”地按固定长度 read就会出现数据半截、错位、解析失败。一个比较稳的做法是设备数据用固定帧格式比如[2字节长度][2字节类型][N字节负载]用户态程序先把头部读完整再根据头部长度字段读后续数据。如果 read 返回不足说明数据还没到齐要继续读、拼包绝不能把半截数据直接丢进管道做解析。另外驱动 read 返回负的错误码和用户态 errno 有一定对应关系。比如返回 -EFAULT用户态 read 返回 -1errno 是 EFAULT。返回 -EINTR用户态 errno 是 EINTR通常说明系统调用被信号打断在驱动测试程序里可以直接重试。如果是用 select/poll 等待驱动数据别忘了驱动 file_operations 里要实现.poll。没有实现 poll 的驱动不一定不能读但 select/poll 会一直认为设备不可读。一个标准的 poll 实现至少要返回 POLLIN表示“有数据可读”static unsigned int demo_poll(struct file *file, poll_table *wait) { unsigned int mask 0; poll_wait(file, demo_wq, wait); if (有数据待读取) mask | POLLIN; return mask; }把这个.poll demo_poll填进 file_operations用户态的 select/epoll 才能正常工作。这也是驱动开发里和管道联调时最容易被忽略的地方。4.4 排查实录上次给一个采集板卡写验证程序遇到的现象是父进程开了一个设备子进程不管怎么收都收不完整偶尔还会 SIGPIPE 直接挂掉。我排查步骤大概是这样的先用 strace 看系统调用strace -f -e tracepipe,read,write,open,close ./pipe_demo-f表示跟踪子进程。很快发现子进程在 read 返回 0 之前pipe1 写端实际上已经被父进程关了但子进程自己手里还攥着一个 pipe1[1] 没关导致 EOF 永远到不了。找到问题后把 fork 分支里多余的 close 补上一次通过。第二个工具是 /proc 文件系统。在程序卡住时另一个终端执行ls -l /proc/pid/fd看到类似这样的输出l-wx------ 1 user user 64 ... 3 - pipe:[12345] lr-x------ 1 user user 64 ... 4 - pipe:[67890]l-wx是写端lr-x是读端。根据方括号里的管道编号能判断哪些 fd 属于同一条管道。如果某个进程同时持有同一管道的读端和写端多半就是 close 漏了。第三个技巧是打印 errno。驱动测试程序里不要只写 perror要把 errno 对应的数值和字符串一起打出来比如 EAGAIN(11)、EINTR(4)、EFAULT(14)。不同 errno 对应完全不同的排查方向只看“read failed”这种信息等于没看。5. 从无名管道延伸到驱动编程的几个认知5.1 管道的 file_operations 是驱动最好的教材很多人学驱动花了很多时间啃 file_operations 的结构体但始终没有真实体感。无名管道恰恰是一个极好的、能跑起来的“类驱动”例子。管道在用户态表现出文件行为它的 fd 可以用 read、write、select、epoll 操作底层靠的是内核里一套完整的文件操作回调。你在用户态调用read(pipefd[0], ...)时内核实际执行的是管道自己的 file_operations 里的 read 回调。你的字符设备驱动也一样用户态调用read(devfd, ...)时执行的是你注册的.read函数。两者处于同一套文件子系统框架下。顺着管道实现去读 VFS 代码你会看到 fd、file、inode、file_operations 这些概念在真实路径里是怎么串起来的。所以我在带新人学驱动时会先让他们去读一遍内核 fs/pipe.c 的读写路径再去写自己的字符设备驱动。读完之后再看字符设备很多模棱两可的问题都会消失。5.2 无名管道与 FIFO、socketpair 怎么选驱动测试里除了无名管道也经常遇到 FIFO 和 socketpair。它们之间有细微差别选错就会多写不少代码。通信方式创建方式适用进程特点无名管道pipe()fork 出来的父子进程轻量无路径名FIFO 有名管道mkfifo()任意相关或不相关进程有路径需先创建socketpairsocketpair(AF_UNIX)父子进程全双工支持 sendmsg/recvmsg无名管道最大的限制是没有路径名只能用于 fork 时继承 fd 的进程。如果两个进程完全没有亲缘关系那就得用 FIFO。FIFO 本质也是管道但它有文件名进程 A 可以 open 写端进程 B 可以 open 读端不需要是父子关系。socketpair 的好处是天然全双工一对 socket fd 两个方向都能用还支持带外数据、文件描述符传递。但代价是需要理解 socket 的读写语义代码比 pipe 稍微重一点。如果只是简单的字符串回传无名管道足够了。5.3 数据采集场景里的推荐做法基于我自己的经验设备驱动加无名管道的最佳组合是驱动只做两件事采集数据和唤醒用户态用户态主进程负责把数据写进管道或共享内存消费者进程从管道读走数据做解析。不要让内核模块碰用户态传入的管道 fd更不要在内核线程里调用 vfs_write 来“模拟管道发送”。内核模块的唯一职责是设备数据搬运管道是用户态的组织手段。如果是高频、大流量数据管道不一定是最优选择。驱动数据量大时可以考虑共享内存加 eventfd或者直接用内核的 relayfs 把驱动数据批量映射到用户态。管道适合的是低频率、小包、事件型的数据比如几十到几百字节一条的消息。把管道当成一根“轻量数据总线”来用而不是当作大水管是最舒服的姿势。最后再分享一个我自己的小习惯写完驱动和用户态管道程序后不要急着跑完整逻辑先写一个最小复现脚本只测“驱动能读到数据”和“父子进程能通过管道传一条消息”这两个点。两个点都通过了再拼起来。驱动开发里 90% 的问题其实是出在“打通”这个环节而不是底层代码本身。把打通环节拆得足够小踩坑成本会低很多。
觉得有用,分享给同行:

为您的企业打造数字门面

稳重轻奢商务风格,端正雅致视觉,长效耐看不易过时。

立即咨询 →