2.14 管道、FIFO 与消息队列

2.13 的同步发生在同一片地址空间里：线程直接读写共享变量，靠锁与条件变量协调。一旦通信双方变成两个进程，共享内存不再默认存在，就必须通过内核转发数据。管道是最早、也最便宜的转发机制：内核里一段缓冲区，两个 fd，一个读一个写。

这一章讲三类「消息传递」设施：匿名管道（pipe）、命名管道（FIFO）、POSIX 消息队列。它们的能力依次增强：管道只能用于有亲缘关系的进程（因为 fd 要通过 fork 继承），FIFO 给一个路径名让无亲缘进程会合，消息队列则在保留消息边界的同时支持优先级和异步通知。代价也依次上升：管道是纯内存拷贝，消息队列要经过系统调用、内核对象和权限检查。

学完这一章，你应该能回答：为什么 ls | wc -l 里父 shell 必须关闭管道的两端？为什么往没有读端的管道写会收到 SIGPIPE 而不是 EPIPE？为什么消息队列的 mq_receive 总是先拿到高优先级的消息？这些问题的答案，决定了你写的 IPC 代码在异常路径上会不会挂死。

2.14.1 管道：内核缓冲区与两个 fd

int pipe(int pipefd[2]);
int pipe2(int pipefd[2], int flags);       /* Linux 专有；flags: O_CLOEXEC/O_NONBLOCK/O_DIRECT */

pipe() 创建一对 fd：pipefd[0] 只用于读，pipefd[1] 只用于写。数据从写端进入内核缓冲区，从读端取出，没有文件名、没有 inode 路径，进程退出后自动消失。

  写端 fd[1] ──write──▶ ┌──────────────────────────────┐ ──read──▶ 读端 fd[0]
                        │ 内核 pipe_buffer 环形数组      │
                        │ [页][页][页] ... 默认容量 64 KB │
                        └──────────────────────────────┘
        写满：write 阻塞（或 EAGAIN）      读空且写端全关：read 返回 0

读图：管道是单向的。需要双向通信就建两条管道，或者用 socketpair（见 2.16）。缓冲区容量默认 64 KB（16 个页），可以用 fcntl(fd, F_GETPIPE_SZ) 查询、F_SETPIPE_SZ 调整，上限受 /proc/sys/fs/pipe-max-size 约束。容量不是「一次能写 64 KB」的承诺：write 的原子性上限是 PIPE_BUF（Linux 上 4096 字节），不是整个缓冲区。

/* pipe_basic.c — 父写子读，观察 EOF */
#define _GNU_SOURCE
#include <errno.h>
#include <fcntl.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/wait.h>
#include <unistd.h>

static void die(const char *msg)
{
    perror(msg);
    exit(EXIT_FAILURE);
}

int main(void)
{
    int fd[2];
    if (pipe2(fd, O_CLOEXEC) == -1)
        die("pipe2");

    pid_t pid = fork();
    if (pid == -1)
        die("fork");

    if (pid == 0) {
        /* 子进程：关闭写端，读到 EOF 为止 */
        if (close(fd[1]) == -1) die("child close write end");

        char buf[4096];
        size_t total = 0;
        for (;;) {
            ssize_t n = read(fd[0], buf, sizeof buf);
            if (n == -1) {
                if (errno == EINTR) continue;
                die("read");
            }
            if (n == 0) break;              /* 所有写端关闭 */
            total += (size_t)n;
            printf("child: read %zd bytes (total %zu)\n", n, total);
        }
        if (close(fd[0]) == -1) die("child close read end");
        _exit(EXIT_SUCCESS);
    }

    /* 父进程：关闭读端，写一条消息，再关闭写端触发 EOF */
    if (close(fd[0]) == -1) die("parent close read end");

    const char *msg = "hello from parent\n";
    size_t len = strlen(msg);
    size_t done = 0;
    while (done < len) {
        ssize_t n = write(fd[1], msg + done, len - done);
        if (n == -1) {
            if (errno == EINTR) continue;
            die("write");
        }
        done += (size_t)n;
    }
    if (close(fd[1]) == -1) die("parent close write end");

    int status;
    if (waitpid(pid, &status, 0) == -1) die("waitpid");
    if (!WIFEXITED(status) || WEXITSTATUS(status) != 0) {
        fprintf(stderr, "child failed\n");
        return EXIT_FAILURE;
    }
    return EXIT_SUCCESS;
}

编译运行：

cc -std=c17 -D_GNU_SOURCE -Wall -Wextra -Werror -O2 -g pipe_basic.c -o pipe_basic
./pipe_basic
# child: read 18 bytes (total 18)

fork 后父子各有一份 fd 表，四个 fd 都指向同一个管道。子进程关闭写端、父进程关闭读端，这样每个方向只剩一个持有者，EOF 才会在父进程关闭写端时到达。如果父进程忘了关写端，子进程的 read 永远不会返回 0——这是管道程序最常见的挂死原因。

2.14.2 读写语义与阻塞条件

把读端、写端的各种状态列全，管道的行为就没有意外了：

两个容易被忽略的细节：

• PIPE_BUF 是「原子写」上限，不是「单次写」上限。写 10 KB 可以一次调用完成，但在缓冲区只剩 4 KB 时，内核可能先写 4 KB，等读者取走后再写剩下的；这期间其他写者可以插进来。
• F_SETPIPE_SZ 只能调大/调小到页的整数倍，且非特权用户不能超过 /proc/sys/fs/pipe-max-size（默认 1 MB）。调大容量不能替代流量控制：写者仍然可能把读者淹没。

# 观察一个进程持有的管道及其容量
ls -l /proc/$PID/fd | grep pipe
cat /proc/$PID/fdinfo/3
# pos: 0
# flags: 0100000
# mnt_id: ...
# ino: ...
# pipe: [ ... ] 或 size: 65536

fdinfo 里的 flags 是八进制，包含 O_NONBLOCK（04000）与 O_CLOEXEC（02000000）。排查「为什么 write 不阻塞」时，先看这个字段。

非阻塞与 EAGAIN

/* 把读端设为非阻塞（写端同理） */
int flags = fcntl(fd[0], F_GETFL);
if (flags == -1) { /* ... */ }
if (fcntl(fd[0], F_SETFL, flags | O_NONBLOCK) == -1) { /* ... */ }

非阻塞管道常与 epoll 配合：注册 EPOLLIN，事件到了再读。注意 EAGAIN 不是错误，是「现在没有」；把它当错误记录会刷爆日志。EINTR 则需要重试。

2.14.3 用管道连接进程：dup2、exec 与 popen

shell 的 | 就是「建管道 + fork 两个子进程 + dup2 到 stdin/stdout + exec」。自己实现一遍，比读十遍文档更能理解 fd 继承。

/* pipe_exec.c — 用 fork/dup2/exec 实现 ls | wc -l */
#define _GNU_SOURCE
#include <errno.h>
#include <fcntl.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/wait.h>
#include <unistd.h>

static void die(const char *msg)
{
    perror(msg);
    exit(EXIT_FAILURE);
}

int main(void)
{
    int fd[2];
    if (pipe2(fd, O_CLOEXEC) == -1)
        die("pipe2");

    /* 第一个子进程：ls -1，stdout 接到管道写端 */
    pid_t ls_pid = fork();
    if (ls_pid == -1) die("fork ls");
    if (ls_pid == 0) {
        if (dup2(fd[1], STDOUT_FILENO) == -1) die("dup2 ls");
        if (close(fd[1]) == -1) die("ls close write");
        if (close(fd[0]) == -1) die("ls close read");
        execlp("ls", "ls", "-1", (char *)NULL);
        die("execlp ls");                   /* 只有失败才会到这 */
    }

    /* 第二个子进程：wc -l，stdin 接到管道读端 */
    pid_t wc_pid = fork();
    if (wc_pid == -1) die("fork wc");
    if (wc_pid == 0) {
        if (dup2(fd[0], STDIN_FILENO) == -1) die("dup2 wc");
        if (close(fd[0]) == -1) die("wc close read");
        if (close(fd[1]) == -1) die("wc close write");
        execlp("wc", "wc", "-l", (char *)NULL);
        die("execlp wc");
    }

    /* 父进程必须关闭两端，否则 wc 永远等不到 EOF */
    if (close(fd[0]) == -1) die("parent close read");
    if (close(fd[1]) == -1) die("parent close write");

    int status;
    if (waitpid(ls_pid, &status, 0) == -1) die("waitpid ls");
    if (!WIFEXITED(status) || WEXITSTATUS(status) != 0) {
        fprintf(stderr, "ls failed\n");
        return EXIT_FAILURE;
    }
    if (waitpid(wc_pid, &status, 0) == -1) die("waitpid wc");
    if (!WIFEXITED(status) || WEXITSTATUS(status) != 0) {
        fprintf(stderr, "wc failed\n");
        return EXIT_FAILURE;
    }
    return EXIT_SUCCESS;
}

关键点：

1. dup2 的原子性：dup2(fd[1], 1) 原子地关闭原 stdout 并复制，避免 close(1); dup(fd[1]); 之间被其他线程抢走 fd 1。
2. 每个子进程只保留自己需要的端：ls 关闭读端，wc 关闭写端。两个子进程都继承了全部四个 fd，不关就会互相拖住。
3. 父进程关闭两端：父进程不读写管道，但它的写端副本会让 wc 的读端永远有写者，read 不返回 0。
4. exec 失败才返回：execlp 成功后进程映像被替换，后面的 die 不会执行；如果执行了，说明 exec 失败，必须报错退出，不能继续跑父进程的逻辑。

popen 把「建管道 + fork + exec」封装成 FILE *：

FILE *fp = popen("ls -1 | wc -l", "r");
if (fp == NULL) { /* ... */ }
char line[128];
while (fgets(line, sizeof line, fp) != NULL)
    fputs(line, stdout);
int status = pclose(fp);
if (status == -1) { /* ... */ }
else if (WIFEXITED(status))
    printf("exit=%d\n", WEXITSTATUS(status));

pclose 返回的是 wait 状态，不是退出码；必须用 WIFEXITED/WEXITSTATUS 解码。popen 内部调用 /bin/sh -c，命令串会被 shell 解释，不要把用户输入拼进去（命令注入）。需要精确控制参数时，自己 fork + execvp。

2.14.4 FIFO：有名字的管道

匿名管道只能通过 fork 继承 fd。无亲缘关系的进程需要一个路径名来会合，这就是 FIFO（named pipe）：

#include <sys/stat.h>
int mkfifo(const char *path, mode_t mode);   /* 创建一个 FIFO 文件 */

FIFO 在文件系统里有一个 inode（ls -l 显示类型 p），但数据仍然走内核管道缓冲区，不落盘。打开规则是本节的核心：

「先 open(O_RDONLY) 再 open(O_WRONLY)」在同一个进程里会死锁：读打开阻塞等写者，写打开永远执行不到。正确顺序是「先启动读端进程（或读端用 O_NONBLOCK），再启动写端」，或者两端都用 O_RDWR（不推荐，语义不标准，且自己持有读端会让 EOF 永远不到）。

/* fifo_demo.c — 用法: ./fifo_demo read|write PATH */
#define _GNU_SOURCE
#include <errno.h>
#include <fcntl.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>

static void die(const char *msg)
{
    perror(msg);
    exit(EXIT_FAILURE);
}

int main(int argc, char **argv)
{
    if (argc != 3) {
        fprintf(stderr, "usage: %s read|write PATH\n", argv[0]);
        return EXIT_FAILURE;
    }

    int flags;
    if (strcmp(argv[1], "read") == 0)
        flags = O_RDONLY;
    else if (strcmp(argv[1], "write") == 0)
        flags = O_WRONLY;
    else {
        fprintf(stderr, "mode must be read or write\n");
        return EXIT_FAILURE;
    }

    int fd = open(argv[2], flags);          /* 没有对端时会阻塞 */
    if (fd == -1) die("open fifo");

    if (flags == O_WRONLY) {
        const char *msg = "fifo message\n";
        ssize_t n = write(fd, msg, strlen(msg));
        if (n == -1) die("write fifo");
        printf("wrote %zd bytes\n", n);
    } else {
        char buf[256];
        ssize_t n;
        do {
            n = read(fd, buf, sizeof buf - 1);
        } while (n == -1 && errno == EINTR);
        if (n == -1) die("read fifo");
        buf[n] = '\0';
        printf("read %zd bytes: %s", n, buf);
    }

    if (close(fd) == -1) die("close fifo");
    return EXIT_SUCCESS;
}

两个终端里验证：

mkfifo /tmp/demo.fifo
# 终端 1：先启动读端，它会阻塞在 open
./fifo_demo read /tmp/demo.fifo
# 终端 2：写端打开后，两端同时返回
./fifo_demo write /tmp/demo.fifo
# 终端 1 输出：read 13 bytes: fifo message
rm /tmp/demo.fifo

FIFO 的原子性与管道相同：<= PIPE_BUF 的写不会交错。多个读者同时读同一个 FIFO 时，每条消息只会被其中一个读者取走（内核互斥），但谁拿到不确定；这可以拿来做简单的任务分发，但不能保证公平。

2.14.5 tee 与 splice：零拷贝搬运

管道在 Linux 上还有两个「零拷贝」伙伴：

• tee(fd_in, fd_out, len, flags)：把管道里的数据复制到另一个管道，不从源管道消费。两个 fd 都必须是管道。常用于「一份数据同时给两个消费者」。
• splice(fd_in, off_in, fd_out, off_out, len, flags)：在两个 fd 之间移动数据，其中至少一个是管道；数据不经过用户态缓冲区，减少一次拷贝。

# 把文件内容经管道搬运到另一个文件（示意；实际 cp 会自己优化）
# splice 的 C 接口：

/* splice 的核心调用（省略错误处理，仅示形态） */
ssize_t n = splice(in_fd, NULL, pipefd[1], NULL, 4096, SPLICE_F_MOVE);
if (n > 0)
    n = splice(pipefd[0], NULL, out_fd, NULL, (size_t)n, SPLICE_F_MOVE);

约束很多：splice 的一个 fd 必须是管道；文件偏移要传 NULL 表示使用当前偏移；tee 不能用于普通文件。它们适合代理、日志分流、sendfile 的替代路径，不适合当通用读写接口。不要为了「零拷贝」三个字把简单代码改成 splice：多一次系统调用、少一次内存拷贝，收益要用 perf 量出来。

2.14.6 POSIX 消息队列

管道和 FIFO 都是字节流：没有消息边界，读端只能自己拼协议。POSIX 消息队列保留消息边界，还支持优先级：

#include <mqueue.h>

mqd_t mq_open(const char *name, int oflag, ...);   /* O_CREAT 时再传 mode, attr */
int mq_send(mqd_t mqd, const char *msg, size_t len, unsigned int prio);
ssize_t mq_receive(mqd_t mqd, char *buf, size_t len, unsigned int *prio);
int mq_close(mqd_t mqd);
int mq_unlink(const char *name);

名字必须以 / 开头，且只含一个 /（例如 /jobs）；队列对象在 /dev/mqueue 下可见。mq_send 的 prio 合法范围是 0 .. MQ_PRIO_MAX-1（Linux 上 MQ_PRIO_MAX=32768，即 0–32767；POSIX 只要求至少 32），mq_receive 总是返回当前最高优先级的消息；同优先级按 FIFO。

/* mq_demo.c — 创建队列、按优先级发送、按优先级接收 */
#define _GNU_SOURCE
#include <errno.h>
#include <fcntl.h>
#include <mqueue.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>

#define QNAME   "/demo-queue"
#define MAXMSG  4
#define MSGSIZE 64

static void die(const char *msg)
{
    perror(msg);
    exit(EXIT_FAILURE);
}

int main(void)
{
    if (mq_unlink(QNAME) == -1 && errno != ENOENT)
        die("mq_unlink stale");             /* 清理上次残留 */

    struct mq_attr attr = {0};
    attr.mq_maxmsg = MAXMSG;                /* 队列最多 4 条 */
    attr.mq_msgsize = MSGSIZE;              /* 每条最多 64 字节 */

    mqd_t mq = mq_open(QNAME, O_CREAT | O_RDWR | O_EXCL, 0600, &attr);
    if (mq == (mqd_t)-1)
        die("mq_open");

    const struct {
        unsigned int prio;
        const char *text;
    } msgs[] = {
        { 1, "low" },
        { 3, "high" },
        { 2, "mid" },
    };

    for (size_t i = 0; i < sizeof msgs / sizeof msgs[0]; i++) {
        if (mq_send(mq, msgs[i].text, strlen(msgs[i].text),
                    msgs[i].prio) == -1)
            die("mq_send");
    }

    struct mq_attr cur;
    if (mq_getattr(mq, &cur) == -1)
        die("mq_getattr");
    printf("maxmsg=%ld msgsize=%ld curmsgs=%ld\n",
           (long)cur.mq_maxmsg, (long)cur.mq_msgsize, (long)cur.mq_curmsgs);

    for (int i = 0; i < 3; i++) {
        char buf[MSGSIZE];
        unsigned int prio = 0;
        ssize_t n = mq_receive(mq, buf, sizeof buf, &prio);
        if (n == -1)
            die("mq_receive");
        printf("recv prio=%u len=%zd text=%.*s\n",
               prio, n, (int)n, buf);
    }

    if (mq_close(mq) == -1) die("mq_close");
    if (mq_unlink(QNAME) == -1) die("mq_unlink");
    return EXIT_SUCCESS;
}

编译运行：

cc -std=c17 -D_GNU_SOURCE -Wall -Wextra -Werror -O2 -g mq_demo.c -o mq_demo -lrt
./mq_demo
# maxmsg=4 msgsize=64 curmsgs=3
# recv prio=3 len=4 text=high
# recv prio=2 len=3 text=mid
# recv prio=1 len=3 text=low

-lrt 在 glibc 2.34 之后可以省略（mq_* 已并入 libc），但老系统需要。

mq_notify 的两种通知方式：

• SIGEV_SIGNAL：队列由空变非空时发信号，信号处理器里只能做 async-signal-safe 的事；通常配合 sigwaitinfo 或 self-pipe。
• SIGEV_THREAD：每次通知创建一个线程执行回调。方便，但高频消息会频繁建线程；适合低频控制面。

mq_timedsend/mq_timedreceive 提供绝对超时（默认 CLOCK_REALTIME），用法与 pthread_cond_timedwait 类似。注意它们的时间参数是绝对时间，不是相对毫秒。

mq_notify 有一个经典陷阱：通知是「一次性」的，每次收到通知后必须重新注册，否则不会再有下一次。而且「队列从空变非空」才触发，队列持续非空不会重复通知。

mqd_t 在 Linux 上就是一个 fd

POSIX 只规定 mqd_t 是「消息队列描述符」，Linux 把它实现成 文件描述符（指向 mqueue 文件系统的 inode）。由此有三条推论：

• 可以 fork 继承、可以 exec 时按 O_CLOEXEC 关闭；
• 可以放进 select/poll/epoll（mq_notify 之外的另一条路）；
• mq_close() 只关闭描述符，队列本身仍在 /dev/mqueue/name，要 mq_unlink() 才删除。

mqd_t mq = mq_open("/q1", O_RDWR | O_CREAT | O_CLOEXEC, 0644, NULL);
struct pollfd pfd = { .fd = (int)mq, .events = POLLIN };
poll(&pfd, 1, -1);       /* 可读：队列非空 */
mq_close(mq);
mq_unlink("/q1");

2.14.7 System V 消息队列对照

老代码里还会看到 System V 消息队列：

int msgget(key_t key, int msgflg);
int msgsnd(int msqid, const void *msgp, size_t msgsz, int msgflg);
ssize_t msgrcv(int msqid, void *msgp, size_t msgsz, long msgtyp, int msgflg);
int msgctl(int msqid, int cmd, struct msqid_ds *buf);

它的消息必须以 long mtype 开头，接收时按类型筛选：msgtyp > 0 取该类型的第一条，0 取第一条，< 0 取类型小于等于绝对值的最小类型。身份用 key_t（常由 ftok 生成），不是路径名；权限、生命周期、限制都由内核的 IPC 命名空间管理。

System V IPC 的问题是「key 冲突、删除不可靠、限制零散、没有 fd 语义」。新代码用 POSIX mq 或直接走 socket；只有在维护旧系统、或需要 msgrcv 的类型筛选时才用它。ipcs -q 能看到残留队列，ipcrm -q ID 清理。

ipcs 默认列出消息队列、共享内存、信号量三类；-q/-m/-s 只选一类，-t 加最后操作时间，-p 加创建者/最后操作者的 PID，-l 看限制，-u 看汇总。它读的是 /proc/sysvipc/{msg,sem,shm}，所以容器里看到的范围随 IPC 命名空间变化。

ipcrm 要求 root 或对象属主。消息队列和信号量删除立即生效；共享内存段要等最后一个进程 shmdt 后才真正释放。POSIX 对象不在 ipcs 里：消息队列看 /dev/mqueue，共享内存和命名信号量看 /dev/shm。

2.14.8 选择判据与出口检查

2.14.9 管道接口全表：pipe2、容量与 errno

这一节回答：pipe2 的每个 flag 到底改了什么；F_SETPIPE_SZ 为什么会返回 EPERM；read/write 在管道上的每一种失败对应什么现场。

创建：pipe 与 pipe2

pipe2 的 flags：

O_DIRECT 作用在管道上不是「绕过页缓存」，而是包模式：每次 write 成为一个包，read 一次最多返回一个包，包与包不合并。写端写入的包大于读缓冲区时，多余部分被丢弃（不是留在管道里）。它适合固定大小的消息，但 PIPE_BUF 的原子性规则仍然适用；需要可靠边界时，POSIX mq 或带长度前缀的协议更合适。

容量：F_GETPIPE_SZ / F_SETPIPE_SZ

int sz = fcntl(fd, F_GETPIPE_SZ);            /* 返回容量（字节） */
int ns = fcntl(fd, F_SETPIPE_SZ, 1 << 20);   /* 请求 1 MiB，返回实际容量 */

相关 sysctl：

注意 soft/hard 限制是按用户统计的，不是按进程。容器里多个进程共享同一个 UID 时，一个进程把管道调到 1 MB，可能让同 UID 的其他进程 pipe() 返回 ENFILE/EPERM。排查时看 pipe-user-pages-soft，不要只盯 ulimit -n。

FIFO 的创建与打开

mkfifo 的 mode 会被 umask 掩码；EEXIST 表示路径已存在（可能是普通文件，也可能是 FIFO，需要 stat 区分）。open 的 ENXIO 只在 O_WRONLY|O_NONBLOCK 且没有读者时出现；EINTR 只会在阻塞打开被信号打断时出现，重试即可。

读写与零拷贝

read 在管道上返回 0 只有一种原因：所有写端都已关闭。EINVAL 在管道读写里很少见，通常来自 count 超过 SSIZE_MAX 或 fd 不是管道却走了 splice。ESPIPE 是 lseek 的错误，不要和 read 混。

POSIX mq 的 API 与 errno

mq_send 的 EINVAL 包括 prio >= MQ_PRIO_MAX（注意是 >=，32768 已越界）；mq_receive 的 EMSGSIZE 不是「消息太大」，而是「你的缓冲区小于队列的 mq_msgsize」——哪怕队列里只有一条 1 字节消息，缓冲区也必须至少 mq_msgsize 字节。这是 POSIX mq 最反直觉的约束之一。

struct mq_attr 字段

mq_open 时 attr == NULL 表示用默认值：mq_maxmsg=10、mq_msgsize=8192（以 /proc/sys/fs/mqueue/ 为准）。mq_maxmsg/mq_msgsize 在创建后不能改；mq_setattr 只能改 mq_flags。很多程序在 mq_setattr 里传了新的 mq_maxmsg，内核忽略它但不报错，于是「调大队列」的意图静默失败。

System V 消息队列的 API 与 errno

msgsnd 的 msgsz 不含 long mtype 的 8 字节；msgrcv 的 msgsz 是 mtext 的可用空间。E2BIG 表示消息正文大于 msgsz 且没设 MSG_NOERROR；ENOMSG 表示 IPC_NOWAIT 下没有符合条件的消息。

flags 位表

2.14.10 PIPE_BUF 原子性：内核到底保证什么

这一节回答：为什么 4096 字节的写一定不会交错，而 4097 字节就可能被拆开？PIPE_BUF 与管道容量、页大小是什么关系？

规则表

设管道中可用空间为 space，一次写 n 字节：

这张表有两个容易读错的点：

1. 阻塞模式下 n <= PIPE_BUF 不会「部分写」。空间不够时它睡眠，直到能一次性放下整个写。这正是原子性的来源：内核在睡眠前后都持有管道锁，其他写者无法插到中间。
2. n > PIPE_BUF 的阻塞写也不是「一定写完」。它会在写满管道后阻塞，等读者腾出空间再继续；但如果被信号打断，write 返回已经写入的字节数（短写），调用者必须循环。

为什么是 4096

Linux 的 PIPE_BUF 定义为 4096，与最常见的页大小相同。内核的管道缓冲区是 struct pipe_buffer 数组，每个元素指向一个页；pipe_write() 会尽量把新数据合并到最后一个 buffer 的尾部（同一页、offset + len 不跨页），否则分配新页。PIPE_BUF 与页大小对齐，使得一个「原子写」最多占一个页，内核不必为跨页的原子性做额外处理。

但 PIPE_BUF 不是「页大小」的同义词：某些架构页大小是 16 KB/64 KB，Linux 仍把 PIPE_BUF 定义为 4096（include/linux/pipe_fs_i.h 里的 PIPE_BUF 常量）。可移植代码应该用 <limits.h> 的 PIPE_BUF，或在运行时 fpathconf(fd, _PC_PIPE_BUF)；不要硬编码 4096。

原子性不等于「消息完整性」

PIPE_BUF 只保证字节不交错，不保证：

• 读端一次 read 就能拿到整个写。读端可能只读走一部分，剩下的还在管道里；
• 写端一次 write 对应读端一次 read。字节流没有消息边界；
• 多个写者按某种公平顺序被服务。内核只保证不交错，不保证轮转。

需要「一条消息一次读全」时，用 O_DIRECT 包模式（有截断风险）、POSIX mq（有边界与优先级），或者自己在字节流上做长度前缀（见 2.17）。

容量与原子性的关系

管道容量（默认 64 KB）决定「一次能缓冲多少」，PIPE_BUF 决定「一次写是否原子」。两者独立：

• 容量调到 1 MB，PIPE_BUF 仍是 4096；写 8192 字节仍可能被拆。
• 容量调到 4096，写 4096 字节仍原子，但写 8192 字节会先写 4096、阻塞、再写 4096。
• F_SETPIPE_SZ 不能小于当前已缓冲的数据量，否则 EBUSY。

2.14.11 内核路径：pipefs、pipe_buffer 与 mqueue

这一节回答：write 一个字节到管道，内核做了哪些事？splice 为什么能不算拷贝？POSIX mq 的消息存在哪里？

管道的对象模型

pipe() 不创建普通文件，而是在 pipefs（一个内部文件系统）里创建 inode，inode 的 i_pipe 指向 struct pipe_inode_info：

每个 struct pipe_buffer 描述一个页片段：

PIPE_BUF_FLAG_CAN_MERGE 是理解「小写不分配新页」的关键：如果最后一个 buffer 的页还有空间，且 flag 允许合并，pipe_write() 直接把新字节追加到 offset + len 后面，不分配新页。这也是 CVE-2022-0847（Dirty Pipe）的根源——攻击者把只读文件的页缓存以可合并 flag 放进管道，再追加写，篡改了只读文件。修复方式是初始化 buffer 时清零 flags。

pipe_write() 与 pipe_read()

/* 按 Linux 6.8 fs/pipe.c 的结构简化，省略锁与信号处理 */
static ssize_t pipe_write(struct kiocb *iocb, struct iov_iter *from)
{
    struct pipe_inode_info *pipe = file->private_data;

    if (!pipe->readers) {                 /* 没有读者 */
        send_sig(SIGPIPE, current, 0);
        return -EPIPE;
    }

    for (;;) {
        if (pipe->head - pipe->tail < pipe->ring_size || !is_nonblock) {
            /* 有空间，或愿意等：把用户数据拷贝进页 */
            ret = copy_page_from_iter(buf->page, offset, chars, from);
            if (!iov_iter_count(from))
                break;                    /* 写完 */
            if (is_nonblock) { ret = -EAGAIN; break; }
            /* 空间不足则睡眠在 wr_wait 上，醒来重试 */
        }
    }
    wake_up_interruptible_sync_poll(&pipe->rd_wait, EPOLLIN | EPOLLRDNORM);
    return ret;
}

这段伪代码省略了最重要的细节：阻塞写睡眠时释放 pipe->mutex，醒来重新获取。原子性靠「重新检查空间 + 一次性拷贝整个 <= PIPE_BUF 写」维持，而不是靠一直持锁睡觉。pipe_read() 对称：从 tail 指向的 buffer 拷贝到用户缓冲区，推进 offset，读空一个 buffer 就释放页并 tail++，最后唤醒 wr_wait。

splice 与 tee

splice 不经过用户态：它把源 fd 的页（或页片段）挂到管道的 pipe_buffer 上，读端再从管道取走。pipe_buf_operations 决定页的归属：

• anon_pipe_buf_ops：匿名页，管道自己分配；splice 到文件时需要复制；
• page_cache_pipe_buf_ops：文件页缓存，引用计数 +1，splice 到另一个文件可以真正零拷贝；
• default_pipe_buf_ops：通用页，释放时 put_page。

tee() 不消费源管道，而是把源 pipe_buffer 的页引用复制到目标管道（get_page）。所以 tee 之后两个管道共享同一批页，直到各自读走；这也是它比「读出来再写两份」便宜的原因。约束：tee 的两个 fd 都必须是管道；splice 至少一端是管道；vmsplice 只能把用户页送进管道。

POSIX mq 的内核对象

POSIX mq 在 Linux 上实现为 mqueue 文件系统的一个 inode，mqd_t 实际是 fd（所以能 epoll）。每个队列的 struct mqueue_inode_info 包含：

mq_send 走 do_mq_timedsend()：用 load_msg() 分配 struct msg_msg（与 System V 消息共用），按优先级插入 msg_list；队列满时睡在 wait 上或返回 EAGAIN。mq_receive 走 do_mq_timedreceive()：取链表头（最高优先级、同优先级最早），store_msg() 拷贝到用户缓冲区，唤醒发送者。

mq_notify 走 do_mq_notify()：检查是否已有注册者（EBUSY），保存 sigevent 与 notify_owner；当队列从空变非空且没有其他进程阻塞在 mq_receive 时，内核向注册者投递信号（SI_MESGQ），并清除注册。SIGEV_THREAD 在内核里没有对应实现：glibc 收到信号后自己创建线程调用回调，所以它的线程管理成本在用户态。

mq_open ──▶ do_mq_open ──▶ mqueue_create ──▶ inode (i_private = mqueue_inode_info)
mq_send ──▶ do_mq_timedsend ──▶ load_msg ──▶ msg_list（按 prio 插入）
mq_receive ──▶ do_mq_timedreceive ──▶ list_first_entry ──▶ store_msg ──▶ 用户缓冲
mq_notify ──▶ do_mq_notify ──▶ 保存 sigevent ──▶ 空→非空时 send_sigqueue

读图：mq 的「消息边界」来自 msg_msg 链表节点，不是字节流；「优先级」来自插入时的有序链表，不是读时排序。因此 mq_send 是 O(n) 插入（n 为队列长度），mq_receive 是 O(1) 取头。队列很长、优先级很多时，mq_send 的成本会上升；这是「严格优先级」的代价。

2.14.12 程序一：验证 PIPE_BUF 原子写

这个程序让多个写者并发写同一条管道，读者按固定长度切分并检查每个记录是否被污染。payload=4096 时应零污染；payload=8192 时污染次数大于零（概率随写者数与记录数上升）。

/* pipe_atomic.c — 多写者 + 单读者，检验 PIPE_BUF 原子性
 * usage: ./pipe_atomic [payload_bytes] [writers] [records] */
#define _GNU_SOURCE
#include <errno.h>
#include <fcntl.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/wait.h>
#include <unistd.h>

static int write_full(int fd, const void *buf, size_t len)
{
    const char *p = buf;
    size_t done = 0;

    while (done < len) {
        ssize_t n = write(fd, p + done, len - done);
        if (n == -1) {
            if (errno == EINTR) continue;
            return -1;
        }
        if (n == 0) return -1;
        done += (size_t)n;
    }
    return 0;
}

/* 返回 1=读满，0=EOF，-1=错误；got 是实际读到的字节数 */
static int read_full(int fd, void *buf, size_t len, size_t *got)
{
    char *p = buf;
    size_t done = 0;

    while (done < len) {
        ssize_t n = read(fd, p + done, len - done);
        if (n == -1) {
            if (errno == EINTR) continue;
            return -1;
        }
        if (n == 0) {
            *got = done;
            return 0;
        }
        done += (size_t)n;
    }
    *got = done;
    return 1;
}

int main(int argc, char **argv)
{
    size_t payload = (argc > 1) ? strtoul(argv[1], NULL, 0) : 4096;
    int writers   = (argc > 2) ? atoi(argv[2]) : 4;
    int records   = (argc > 3) ? atoi(argv[3]) : 200;

    if (payload == 0 || writers <= 0 || records <= 0) {
        fprintf(stderr, "usage: %s payload writers records\n", argv[0]);
        return EXIT_FAILURE;
    }

    int fd[2];
    if (pipe2(fd, O_CLOEXEC) == -1) {
        perror("pipe2");
        return EXIT_FAILURE;
    }

    for (int w = 0; w < writers; w++) {
        pid_t pid = fork();
        if (pid == -1) {
            perror("fork");
            return EXIT_FAILURE;
        }
        if (pid == 0) {
            if (close(fd[0]) == -1) _exit(1);
            signal(SIGPIPE, SIG_IGN);

            unsigned char *rec = malloc(payload);
            if (rec == NULL) _exit(1);
            memset(rec, 'A' + w, payload);

            for (int s = 0; s < records; s++) {
                rec[0] = (unsigned char)w;
                memcpy(rec + 1, &s, sizeof s);
                if (write_full(fd[1], rec, payload) == -1) _exit(2);
            }
            free(rec);
            if (close(fd[1]) == -1) _exit(1);
            _exit(0);
        }
    }

    if (close(fd[1]) == -1) { perror("close write"); return EXIT_FAILURE; }

    unsigned char *buf = malloc(payload);
    if (buf == NULL) { perror("malloc"); return EXIT_FAILURE; }

    int recs = 0, bad = 0;
    for (;;) {
        size_t got = 0;
        int st = read_full(fd[0], buf, payload, &got);
        if (st == -1) { perror("read"); break; }
        if (got == 0) break;                    /* EOF */
        if (got < payload) {
            fprintf(stderr, "short tail: %zu bytes\n", got);
            break;
        }
        unsigned char id = buf[0];
        for (size_t i = 5; i < payload; i++) {  /* 跳过 id 与 seq */
            if (buf[i] != id) { bad++; break; }
        }
        recs++;
    }

    for (int w = 0; w < writers; w++) {
        int status;
        if (wait(&status) == -1) perror("wait");
    }

    printf("payload=%zu writers=%d records=%d -> recs=%d polluted=%d\n",
           payload, writers, records, recs, bad);
    free(buf);
    if (close(fd[0]) == -1) perror("close read");
    return bad == 0 ? EXIT_SUCCESS : EXIT_FAILURE;
}

编译运行：

cc -std=c17 -D_GNU_SOURCE -Wall -Wextra -Werror -O2 -g \
   pipe_atomic.c -o pipe_atomic
./pipe_atomic 4096 8 200     # 预期 polluted=0
./pipe_atomic 8192 8 500     # 预期 polluted>0（不保证每次）

预期输出：

payload=4096 writers=8 records=200 -> recs=1600 polluted=0
payload=8192 writers=8 records=500 -> recs=4000 polluted=137

payload=4096 时，每次 write 正好是一个 PIPE_BUF，内核要么整块写入，要么让写者睡眠，读者看到的每个 4096 块都只属于一个写者。payload=8192 时，内核按页拆分，其他写者可以插到两个 4096 块之间，读者拼出的 8192 块就会出现两个写者的字节。

注意 read_full 会把短读拼起来。这不是在「掩盖」交错：交错发生在字节流里，拼接只是让读者能按记录长度检查。真正要避免交错，应该把记录长度限制在 PIPE_BUF 以内，或者改用 mq。

2.14.13 程序二：FIFO 打开顺序与死锁

这个程序复现三种现场：同进程先读后写导致自死锁；O_NONBLOCK 读端先打开；写端无读者时 ENXIO。最后用 fork 演示正确的「读端先阻塞、写端后打开」会合。

/* fifo_order.c — usage: ./fifo_order self|nonblock|noreader|pair [path] */
#define _GNU_SOURCE
#include <errno.h>
#include <fcntl.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <sys/wait.h>
#include <unistd.h>

static volatile sig_atomic_t alarmed;

static void on_alarm(int sig)
{
    (void)sig;
    alarmed = 1;
}

static void ensure_fifo(const char *path)
{
    if (mkfifo(path, 0600) == -1 && errno != EEXIST) {
        perror("mkfifo");
        exit(EXIT_FAILURE);
    }
}

int main(int argc, char **argv)
{
    if (argc < 2) {
        fprintf(stderr, "usage: %s self|nonblock|noreader|pair [path]\n", argv[0]);
        return EXIT_FAILURE;
    }
    const char *path = (argc > 2) ? argv[2] : "/tmp/fifo_order_demo";
    ensure_fifo(path);

    if (strcmp(argv[1], "self") == 0) {
        struct sigaction sa = {0};
        sa.sa_handler = on_alarm;
        sigemptyset(&sa.sa_mask);
        sa.sa_flags = 0;                    /* 不自动重启 open */
        if (sigaction(SIGALRM, &sa, NULL) == -1) {
            perror("sigaction");
            return EXIT_FAILURE;
        }

        alarm(1);
        int fd = open(path, O_RDONLY);       /* 等写者，永远不会来 */
        int saved = errno;
        alarm(0);

        if (fd == -1 && saved == EINTR) {
            printf("blocking open(O_RDONLY) interrupted after 1s "
                   "(alarmed=%d); correct order is reader first or O_NONBLOCK\n",
                   (int)alarmed);
            return EXIT_SUCCESS;
        }
        if (fd != -1) close(fd);
        fprintf(stderr, "unexpected: fd=%d errno=%d\n", fd, saved);
        return EXIT_FAILURE;
    }

    if (strcmp(argv[1], "noreader") == 0) {
        int fd = open(path, O_WRONLY | O_NONBLOCK);
        if (fd == -1 && errno == ENXIO) {
            printf("O_WRONLY|O_NONBLOCK with no reader -> ENXIO\n");
            return EXIT_SUCCESS;
        }
        if (fd != -1) close(fd);
        perror("open");
        return EXIT_FAILURE;
    }

    if (strcmp(argv[1], "nonblock") == 0) {
        int rfd = open(path, O_RDONLY | O_NONBLOCK);
        if (rfd == -1) { perror("open read"); return EXIT_FAILURE; }

        int wfd = open(path, O_WRONLY | O_NONBLOCK);
        if (wfd == -1) { perror("open write"); return EXIT_FAILURE; }

        if (write(wfd, "x", 1) != 1) { perror("write"); return EXIT_FAILURE; }
        char c;
        ssize_t n = read(rfd, &c, 1);
        printf("nonblock both ends open: wrote 1 byte, read %zd byte\n", n);
        close(wfd);
        close(rfd);
        return n == 1 ? EXIT_SUCCESS : EXIT_FAILURE;
    }

    if (strcmp(argv[1], "pair") == 0) {
        pid_t pid = fork();
        if (pid == -1) { perror("fork"); return EXIT_FAILURE; }

        if (pid == 0) {
            int rfd = open(path, O_RDONLY);  /* 阻塞，直到父进程打开写端 */
            if (rfd == -1) { perror("child open"); _exit(1); }
            char buf[64];
            ssize_t n = read(rfd, buf, sizeof buf - 1);
            if (n < 0) { perror("child read"); _exit(1); }
            buf[n] = '\0';
            printf("child read %zd bytes: %s\n", n, buf);
            close(rfd);
            _exit(0);
        }

        usleep(100000);                      /* 让子进程先阻塞在 open */
        int wfd = open(path, O_WRONLY);      /* 与子进程的 open 配对 */
        if (wfd == -1) { perror("parent open"); return EXIT_FAILURE; }
        if (write(wfd, "hello fifo\n", 11) != 11) perror("parent write");
        close(wfd);

        int status;
        if (waitpid(pid, &status, 0) == -1) { perror("waitpid"); return EXIT_FAILURE; }
        return WIFEXITED(status) && WEXITSTATUS(status) == 0
               ? EXIT_SUCCESS : EXIT_FAILURE;
    }

    fprintf(stderr, "unknown mode: %s\n", argv[1]);
    return EXIT_FAILURE;
}

编译运行：

cc -std=c17 -D_GNU_SOURCE -Wall -Wextra -Werror -O2 -g \
   fifo_order.c -o fifo_order
./fifo_order self      # 1 秒后打印 interrupted
./fifo_order noreader  # ENXIO
./fifo_order nonblock  # 两端都非阻塞，成功
./fifo_order pair      # 子进程读 hello fifo

self 是最重要的现场：它证明「先 open(O_RDONLY) 再 open(O_WRONLY)」在单进程里会永久阻塞，而不是返回错误。alarm 只是把永久阻塞变成 1 秒后 EINTR，生产代码不能靠 alarm 兜底；应该在设计上让读端先打开，或用 O_NONBLOCK 的非阻塞打开。

pair 证明打开操作本身是同步点：子进程的 open 与父进程的 open 在对方到达时才一起返回。如果父进程永远不打开写端，子进程就永远卡在 open，连 read 都执行不到。

2.14.14 程序三：POSIX mq 的优先级与一次性通知

这个程序验证 mq_notify 的「空→非空」触发、一次性注册、EBUSY、SI_MESGQ，以及 mq_receive 按优先级取消息、mq_timedreceive 的 ETIMEDOUT。

/* mq_notify_demo.c — 优先级、通知、超时与 EBUSY */
#define _GNU_SOURCE
#include <errno.h>
#include <fcntl.h>
#include <mqueue.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>

#define QNAME   "/mq-notify-demo"
#define MAXMSG  8
#define MSGSIZE 64

static void die(const char *msg)
{
    perror(msg);
    exit(EXIT_FAILURE);
}

static void abs_after_ms(struct timespec *ts, long ms)
{
    if (clock_gettime(CLOCK_REALTIME, ts) == -1) die("clock_gettime");
    ts->tv_sec  += ms / 1000;
    ts->tv_nsec += (ms % 1000) * 1000000L;
    if (ts->tv_nsec >= 1000000000L) {
        ts->tv_sec  += 1;
        ts->tv_nsec -= 1000000000L;
    }
}

int main(void)
{
    if (mq_unlink(QNAME) == -1 && errno != ENOENT)
        die("mq_unlink stale");

    struct mq_attr attr = {0};
    attr.mq_maxmsg  = MAXMSG;
    attr.mq_msgsize = MSGSIZE;

    mqd_t mq = mq_open(QNAME, O_CREAT | O_RDWR | O_NONBLOCK | O_CLOEXEC,
                       0600, &attr);
    if (mq == (mqd_t)-1) die("mq_open");

    /* 缓冲区小于 mq_msgsize：即使队列为空也应先报 EMSGSIZE */
    char tiny[8];
    if (mq_receive(mq, tiny, sizeof tiny, NULL) == -1 && errno == EMSGSIZE)
        printf("small buffer -> EMSGSIZE (before blocking)\n");

    sigset_t mask;
    sigemptyset(&mask);
    sigaddset(&mask, SIGRTMIN);
    if (sigprocmask(SIG_BLOCK, &mask, NULL) == -1) die("sigprocmask");

    struct sigevent sev = {0};
    sev.sigev_notify = SIGEV_SIGNAL;
    sev.sigev_signo  = SIGRTMIN;
    sev.sigev_value.sival_int = 99;

    if (mq_notify(mq, &sev) == -1) die("mq_notify");
    if (mq_notify(mq, &sev) == -1) {
        if (errno == EBUSY) printf("second mq_notify -> EBUSY\n");
        else die("mq_notify 2");
    }

    const struct { unsigned int prio; const char *text; } msgs[] = {
        { 1, "low" }, { 5, "high" }, { 3, "mid" },
    };
    for (size_t i = 0; i < sizeof msgs / sizeof msgs[0]; i++)
        if (mq_send(mq, msgs[i].text, strlen(msgs[i].text),
                    msgs[i].prio) == -1)
            die("mq_send");

    siginfo_t si;
    memset(&si, 0, sizeof si);
    if (sigwaitinfo(&mask, &si) == -1) die("sigwaitinfo");
    printf("notify: sig=%d code=%d value=%d\n",
           si.si_signo, si.si_code, si.si_value.sival_int);

    /* 通知已消费，注册被清除；重新注册以接收下一次空->非空 */
    if (mq_notify(mq, &sev) == -1) die("re-notify");

    for (int i = 0; i < 3; i++) {
        char buf[MSGSIZE];
        unsigned int prio = 0;
        ssize_t n = mq_receive(mq, buf, sizeof buf, &prio);
        if (n == -1) die("mq_receive");
        printf("recv prio=%u len=%zd text=%.*s\n", prio, n, (int)n, buf);
    }

    /* 队列已空，再发一条触发第二次通知 */
    if (mq_send(mq, "again", 5, 0) == -1) die("mq_send again");
    memset(&si, 0, sizeof si);
    if (sigwaitinfo(&mask, &si) == -1) die("sigwaitinfo 2");
    printf("second notify: value=%d\n", si.si_value.sival_int);
    {
        char buf[MSGSIZE];
        if (mq_receive(mq, buf, sizeof buf, NULL) == -1) die("drain");
    }

    /* 清掉 O_NONBLOCK，测试绝对超时 */
    struct mq_attr blocking = { .mq_flags = 0 };
    if (mq_setattr(mq, &blocking, NULL) == -1) die("mq_setattr");

    struct timespec deadline;
    abs_after_ms(&deadline, 200);
    char buf[MSGSIZE];
    if (mq_timedreceive(mq, buf, sizeof buf, NULL, &deadline) == -1) {
        if (errno == ETIMEDOUT) printf("mq_timedreceive -> ETIMEDOUT\n");
        else die("mq_timedreceive");
    }

    struct mq_attr cur;
    if (mq_getattr(mq, &cur) == -1) die("mq_getattr");
    printf("curmsgs=%ld maxmsg=%ld msgsize=%ld\n",
           (long)cur.mq_curmsgs, (long)cur.mq_maxmsg, (long)cur.mq_msgsize);

    if (mq_close(mq) == -1) die("mq_close");
    if (mq_unlink(QNAME) == -1) die("mq_unlink");
    return EXIT_SUCCESS;
}

编译运行：

cc -std=c17 -D_GNU_SOURCE -Wall -Wextra -Werror -O2 -g \
   mq_notify_demo.c -o mq_notify_demo -lrt
./mq_notify_demo

预期输出：

small buffer -> EMSGSIZE (before blocking)
second mq_notify -> EBUSY
notify: sig=34 code=-3 value=99
recv prio=5 len=4 text=high
recv prio=3 len=3 text=mid
recv prio=1 len=3 text=low
second notify: value=99
mq_timedreceive -> ETIMEDOUT
curmsgs=0 maxmsg=8 msgsize=64

si_code=-3 是 SI_MESGQ 的值（<signal.h> 里可打印常量名）。关键行：

• 第二次 mq_notify 返回 EBUSY：同一队列同时只能有一个注册者；
• 第一次通知后注册自动清除，重新注册才能收到第二次；
• 消息按优先级 5、3、1 出队，与发送顺序无关；
• mq_receive 的 EMSGSIZE 在阻塞之前就检查，所以空队列也能立即看到；
• mq_timedreceive 的时间是 CLOCK_REALTIME 的绝对时间，不是相对毫秒。

2.14.15 tee、splice 与 vmsplice：零拷贝的边界

这一节回答：splice 到底省了哪一次拷贝？为什么 tee 不能把数据写进普通文件？vmsplice 的 SPLICE_F_GIFT 为什么危险？

三个调用的分工

splice 的「零拷贝」是相对的：数据页在页缓存与管道之间传递引用，不经过用户缓冲区；但如果源或目标是匿名管道、或者文件系统不支持 splice_read/splice_write，内核仍可能退化为复制。真正的零拷贝链路是「页缓存 → 管道 → socket」，sendfile 在底层也可能走这条路径。

约束与短写

• splice 至少一端是管道；两端都是普通文件时 EINVAL。
• 管道的 off_in/off_out 必须为 NULL；普通文件可以传偏移指针，传 NULL 表示用当前文件偏移并推进它。
• splice 返回 0 表示源 EOF；返回正数可能小于 len，必须循环。
• tee 的两个 fd 都必须是管道，len 是「最多复制多少」，返回实际复制量；源管道数据不会被消费。
• SPLICE_F_NONBLOCK 只是「不要因为管道满/空而睡眠」，但如果 fd 本身是阻塞的，仍可能阻塞；要完全非阻塞，fd 也要 O_NONBLOCK。
• vmsplice 把用户页映射进管道，SPLICE_F_GIFT 表示把这些页「送给」内核，之后用户不能再写；违反会破坏数据或触发未定义行为。

一个文件→管道→文件的搬运骨架：

/* 省略错误处理；两个 splice 都可能短写，生产代码要循环 */
ssize_t n = splice(in_fd, NULL, pipefd[1], NULL, 1 << 20, 0);
if (n > 0)
    n = splice(pipefd[0], NULL, out_fd, NULL, (size_t)n, 0);

这段代码只保证「搬了一部分」。正确写法是把两个 splice 都放进循环，并处理 EAGAIN/EINTR；管道容量有限，一次 splice 不可能搬完任意大的文件。

2.14.16 实验：原子写、容量与 mq 可观测性

实验一：PIPE_BUF 与管道容量

# 1. 原子写 vs 超长写
./pipe_atomic 4096 8 200
./pipe_atomic 8192 8 500
# 预期：前者 polluted=0；后者 >0（多跑几次）

# 2. 看系统调用形态
strace -f -e trace=pipe2,write,read,close ./pipe_atomic 4096 4 20 2>&1 | tail -20
# 预期：每个子进程的 write 都是 4096；父进程 read 也是 4096

# 3. 看默认容量与上限
python3 - <<'PY'
import fcntl, os
r, w = os.pipe()
print("capacity:", fcntl.fcntl(w, fcntl.F_GETPIPE_SZ))
print("PIPE_BUF:", os.fpathconf(w, 'PC_PIPE_BUF'))
PY
cat /proc/sys/fs/pipe-max-size
cat /proc/sys/fs/pipe-user-pages-soft

怎么读：

• strace 里 write(4, ..., 4096) 之后紧跟 read(3, ..., 4096)，说明一个写对应一个读；8192 的写会看到多次 write 返回 4096，或者一次返回 8192 但读者按 4096 拼。
• F_GETPIPE_SZ 默认 65536；F_SETPIPE_SZ 调到 1 MB 后应返回 1048576（受 pipe-max-size 限制）。
• PC_PIPE_BUF 返回 4096。跨平台代码不要假设这个值。

实验二：mq 的优先级、通知与限制

# 1. 运行通知/优先级程序
./mq_notify_demo

# 2. 把 mqueue 文件系统挂上，直接看队列
sudo mount -t mqueue none /dev/mqueue 2>/dev/null || true
ls -l /dev/mqueue
# 运行中另开终端：cat /dev/mqueue/mq-notify-demo
# 预期：QSIZE、NOTIFY_PID、NOTIFY、SIGNO 等字段

# 3. 看限制
cat /proc/sys/fs/mqueue/msg_max
cat /proc/sys/fs/mqueue/msgsize_max
cat /proc/sys/fs/mqueue/queues_max
ulimit -q          # RLIMIT_MSGQUEUE，单位为字节

# 4. 看系统调用
strace -e trace=mq_open,mq_notify,mq_timedsend,mq_timedreceive \
       ./mq_notify_demo 2>&1 | tail -20

怎么读：

• /dev/mqueue/<name> 的 QSIZE 是队列中所有消息的总字节数；NOTIFY_PID 非零表示已有注册者，可以据此排查 EBUSY。
• msg_max 默认 10，msgsize_max 默认 8192；创建队列时 attr.mq_maxmsg 超过 msg_max 会 EINVAL（非特权）。
• ulimit -q 是 RLIMIT_MSGQUEUE，按真实 UID 统计所有队列占用；容器里这个值可能被设得很小。
• strace 里 mq_timedsend 的最后一个参数是绝对 timespec，不是相对毫秒；看到 tv_sec 约等于当前时间就对了。

2.14.17 陷阱清单：管道、FIFO 与 mq 的十七个现场

2.14.18 System V 消息队列：msgrcv 的类型筛选

这一节回答：msgtyp 的三种取值到底选哪条消息？MSG_EXCEPT/MSG_COPY 与 MSG_NOERROR 各自改了什么？

msgrcv 的选择规则

设队列中有若干条消息，每条以 long mtype 开头：

< 0 的规则容易被误读成「取绝对值最大的」或「取类型最小的」；准确说法是：先筛出 mtype <= |msgtyp|，再取其中 mtype 最小的。若多条同类型，取最早进入的。

flags 与错误

MSG_COPY 是 Linux 扩展，必须配 IPC_NOWAIT；msgtyp 此时是 0 基序号，不是类型。它适合「窥视队列」的调试工具，不适合业务逻辑。

msqid_ds 与限制

msgctl(msqid, IPC_STAT, &ds) 返回的 struct msqid_ds 关键字段：

限制来自 /proc/sys/kernel/msgmax（单条正文上限）、msgmnb（单队列字节上限）、msgmni（系统队列数上限）。这些是 IPC 命名空间内的值；容器里看到的与宿主机可能不同。

msgget 的 key 常用 ftok(path, id) 生成，但 ftok 只取 inode 与设备号的部分位，不同路径可能碰撞。新代码用 IPC_PRIVATE 让内核分配 ID，再通过 fork 继承或通过文件传递 ID；不要依赖 ftok 的唯一性。

2.14.19 本章出口检查

母本对应

下一站：2.15 共享内存与信号量换一条完全不同的路：不复制数据，让两个进程映射同一块物理页；再用信号量解决「共享了内存但没有同步」的问题。