2.15 共享内存与信号量

2.14 的所有设施都要把数据从用户态拷进内核、再拷出来。管道、FIFO、消息队列，无一例外。共享内存走另一条路：内核只负责把同一块物理页映射进两个进程的地址空间，之后读写就像访问自己的变量，一次拷贝都没有。它是所有 IPC 里最快的，也是最危险的——没有边界、没有消息、没有同步，一切都要你自己建。

这一章的重点不是 mmap 的参数（2.3 已经讲过），而是「共享之后怎么办」。两个进程同时写一个结构体，谁先谁后？一个进程在写、另一个进程在读到一半时被调度走，读到的是一半旧值一半新值吗？进程崩溃时留下的锁谁来解？这些问题决定了共享内存能否用于生产。

信号量是本章的另一半。它既可以当「计数器」限制并发，也可以当「事件标志」唤醒等待者，还能跨进程使用。把共享内存与信号量组合起来，就得到一个跨进程的有界队列——这是数据库、消息中间件、高性能代理里最常见的 IPC 结构。本章最后会给一个完整实现。

2.15.1 共享内存为什么最快：零拷贝

一次管道传输的数据路径是：用户缓冲区 → 内核管道页 → 用户缓冲区，两次拷贝。一次 TCP 传输至少两次拷贝，外加协议栈处理。共享内存的路径是：进程 A 写自己的虚拟页 → 进程 B 读同一物理页，零次拷贝（页表映射除外）。

 进程 A 的虚拟地址                物理页                 进程 B 的虚拟地址
 0x7f8a_0000 ──┐              ┌────────────────┐              ┌── 0x7f9b_0000
               ├──── PTE ───▶ │  shared page   │ ◀─── PTE ────┤
 0x7f8a_1000 ──┘              └────────────────┘              └── 0x7f9b_1000
        │                                                        │
        └──────── 同一个 shm 对象 / 同一个 MAP_SHARED 文件 ────────┘

读图：两个进程的虚拟地址可以完全不同，页表项指向同一个物理页。写 A 的地址，B 立刻能在自己的地址看到（在同一缓存一致性域内）。这不是「复制过去」，而是「本来就只有一份」。也正因为如此，共享内存没有「发送」与「接收」的语义：没有队列、没有 EOF、没有消息边界，只有一个裸露的字节数组。

/* shm_basic.c — POSIX 共享内存对象：父写子读 */
#define _GNU_SOURCE
#include <errno.h>
#include <fcntl.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/mman.h>
#include <sys/stat.h>
#include <sys/wait.h>
#include <unistd.h>

struct shared {
    int magic;
    long value;
    char text[64];
};

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

int main(void)
{
    const char *name = "/shm-basic-demo";

    if (shm_unlink(name) == -1 && errno != ENOENT)
        die("shm_unlink stale");

    int fd = shm_open(name, O_CREAT | O_RDWR | O_EXCL, 0600);
    if (fd == -1)
        die("shm_open");

    if (ftruncate(fd, (off_t)sizeof(struct shared)) == -1)
        die("ftruncate");                   /* 必须先定大小，再 mmap */

    struct shared *p = mmap(NULL, sizeof *p, PROT_READ | PROT_WRITE,
                            MAP_SHARED, fd, 0);
    if (p == MAP_FAILED)
        die("mmap");

    if (close(fd) == -1)                    /* 映射仍然有效，fd 可以关 */
        die("close shm fd");

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

    if (pid == 0) {
        p->magic = 0x5a5a;
        p->value = 42;
        snprintf(p->text, sizeof p->text, "written by child");
        _exit(EXIT_SUCCESS);                /* 不刷新 stdio，直接退出 */
    }

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

    /* waitpid 返回后，子进程的写入对父进程可见 */
    printf("magic=0x%x value=%ld text=%s\n",
           p->magic, p->value, p->text);

    if (munmap(p, sizeof *p) == -1)
        die("munmap");
    if (shm_unlink(name) == -1)
        die("shm_unlink");

    return EXIT_SUCCESS;
}

编译运行：

cc -std=c17 -D_GNU_SOURCE -Wall -Wextra -Werror -O2 -g shm_basic.c -o shm_basic -lrt
./shm_basic
# magic=0x5a5a value=42 text=written by child
ls -l /dev/shm            # 运行期间能看到 /dev/shm/shm-basic-demo

shm_open 的名字规则与消息队列一样：以 / 开头、只含一个 /。对象创建在 tmpfs 上，ls -l /dev/shm 能看到对应文件；shm_unlink 相当于删除名字，已映射的进程不受影响，最后一个映射解除后内存才释放。

waitpid 在这里充当同步原语：它保证子进程已经退出，而进程退出隐含「之前的写已经完成」。但这是进程级的同步，不是数据级的；如果两个进程并发读写同一个结构体，waitpid 帮不了你。

2.15.2 POSIX 共享内存对象：创建、映射与生命周期

shm_open 返回的是一个 fd，和普通文件 fd 一样可以 ftruncate、fstat、mmap、close。几个容易踩的点：

1. 先 ftruncate 再 mmap。新创建的 shm 对象大小为 0，直接映射 0 字节会得到 EINVAL；映射后再访问超出文件大小的页会收到 SIGBUS（不是 SIGSEGV）。
2. mmap 的 length 可以大于文件大小，但访问超出部分会 SIGBUS。用 fstat 确认大小，或让创建者把大小写进共享结构体的头部。
3. close(fd) 不影响映射。映射建立后，fd 只是「打开的文件描述」，munmap 才解除映射。创建者可以 shm_open + ftruncate + mmap + close + shm_unlink，对象在映射期间仍然存在。
4. shm_unlink 之后，已经打开/映射的进程继续可用，新进程无法再按名字打开。这是「创建即删除」的常用技巧：避免名字残留，同时让对象随最后一个使用者消失。

/* 创建者的标准顺序 */
int fd = shm_open(name, O_CREAT | O_RDWR | O_EXCL, 0600);
ftruncate(fd, size);
void *p = mmap(NULL, size, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
close(fd);
shm_unlink(name);       /* 名字立即消失；映射还在 */

使用者则用 shm_open(name, O_RDWR, 0)（不带 O_CREAT），fstat 拿大小，再 mmap。要处理「创建者还没创建」的 ENOENT：重试、等待，或者约定由启动器先创建。

2.15.3 共享内存不提供同步

把上一节的 waitpid 去掉，让父子进程同时读写，问题立刻出现：

/* 错误示范：没有同步的共享计数器 */
if (pid == 0) {
    for (int i = 0; i < 1000000; i++)
        p->value++;                 /* 读-改-写，跨进程同样会丢更新 */
    _exit(0);
}
for (int i = 0; i < 1000000; i++)
    p->value++;

这和 2.13 的线程竞态是同一个问题，只是发生在两个地址空间里。C 语言的内存模型不会因为「跨进程」而变得宽松：没有同步的并发读写就是数据竞争，结果未定义。

跨进程同步有三条路：

进程共享互斥锁的初始化：

pthread_mutexattr_t mattr;
pthread_mutex_t *lock = /* 指向共享内存中的锁 */;

pthread_mutexattr_init(&mattr);
pthread_mutexattr_setpshared(&mattr, PTHREAD_PROCESS_SHARED);
pthread_mutexattr_setrobust(&mattr, PTHREAD_MUTEX_ROBUST);   /* 持有者崩溃可恢复 */
pthread_mutex_init(lock, &mattr);
pthread_mutexattr_destroy(&mattr);

PTHREAD_MUTEX_ROBUST 很重要：普通互斥锁的持有者进程崩溃后，锁永远处于锁定状态，其他进程 pthread_mutex_lock 会永久阻塞。robust 锁的等待者会得到 EOWNERDEAD，表示「上一个持有者死了，临界区可能处于不一致状态」；等待者必须修复数据，然后调用 pthread_mutex_consistent()，锁才恢复正常。如果它不修，直接 unlock，锁会进入 ENOTRECOVERABLE，以后再也无法使用。

条件变量也要 pthread_condattr_setpshared(&cattr, PTHREAD_PROCESS_SHARED)，否则跨进程 wait 行为未定义。而且条件变量的等待队列在内核/glibc 里与进程相关，跨进程使用必须严格按「共享内存 + 进程共享属性」配置。

2.15.4 POSIX 信号量：命名与未命名

#include <semaphore.h>

/* 未命名：放在共享内存里，pshared=1 表示跨进程 */
int sem_init(sem_t *sem, int pshared, unsigned int value);
int sem_destroy(sem_t *sem);

/* 命名：有名字，任何进程可 open */
sem_t *sem_open(const char *name, int oflag, ...);   /* O_CREAT 时再传 mode, value */
int sem_close(sem_t *sem);
int sem_unlink(const char *name);

/* 通用操作 */
int sem_wait(sem_t *sem);        /* P：减一，0 则阻塞；可被信号打断 EINTR */
int sem_trywait(sem_t *sem);     /* 非阻塞；0 则 EAGAIN */
int sem_timedwait(sem_t *sem, const struct timespec *abs);
int sem_post(sem_t *sem);        /* V：加一，唤醒等待者；async-signal-safe */
int sem_getvalue(sem_t *sem, int *sval);

信号量的值表示「可用资源的数量」：

• sem_init(&s, 1, 1)：二值信号量，初值 1，第一个 wait 减到 0，第二个阻塞；
• sem_init(&s, 1, N)：计数信号量，最多 N 个并发；
• sem_init(&s, 1, 0)：事件标志，post 之前所有 wait 都阻塞。

sem_wait 会因信号返回 -1/EINTR，必须重试：

while (sem_wait(&sem) == -1) {
    if (errno != EINTR) { /* 真错误 */ }
}

sem_getvalue 的返回值只适合监控：它反映某个瞬间的值，下一秒就可能变。Linux 的 futex 实现里，有等待者时内部计数可能为负，sem_getvalue 可能返回 0 或负数；POSIX 允许这种表示，不要拿它做控制流。

命名信号量的名字形如 /mysem，对象出现在 /dev/shm/sem.mysem。sem_open 的 O_CREAT 需要 mode 与初值；O_EXCL 用于「只允许创建一次」。sem_close 只关闭句柄，sem_unlink 删除名字；和 shm 一样，最后一个引用消失后才真正释放。

/* sem_named.c — 命名信号量做父子 ping-pong */
#define _GNU_SOURCE
#include <errno.h>
#include <fcntl.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <semaphore.h>
#include <sys/wait.h>
#include <unistd.h>

#define PING "/sem-ping-demo"
#define PONG "/sem-pong-demo"

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

int main(void)
{
    sem_unlink(PING);
    sem_unlink(PONG);

    sem_t *ping = sem_open(PING, O_CREAT | O_EXCL, 0600, 0);
    if (ping == SEM_FAILED) die("sem_open ping");
    sem_t *pong = sem_open(PONG, O_CREAT | O_EXCL, 0600, 0);
    if (pong == SEM_FAILED) die("sem_open pong");

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

    if (pid == 0) {
        for (int i = 0; i < 3; i++) {
            while (sem_wait(ping) == -1 && errno == EINTR)
                ;
            printf("child: ping %d\n", i);
            if (sem_post(pong) == -1) die("sem_post pong");
        }
        _exit(EXIT_SUCCESS);
    }

    for (int i = 0; i < 3; i++) {
        printf("parent: pong %d\n", i);
        if (sem_post(ping) == -1) die("sem_post ping");
        while (sem_wait(pong) == -1 && errno == EINTR)
            ;
    }

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

    if (sem_close(ping) == -1) die("sem_close ping");
    if (sem_close(pong) == -1) die("sem_close pong");
    if (sem_unlink(PING) == -1) die("sem_unlink ping");
    if (sem_unlink(PONG) == -1) die("sem_unlink pong");
    return EXIT_SUCCESS;
}

sem_post 是 async-signal-safe 的，可以在信号处理器里调用；sem_wait 不是，不能在处理器里等。这个不对称性决定了「信号处理器只 post，工作线程 wait」的模式。

2.15.5 共享内存 + 信号量：跨进程有界队列

下面把三者组合起来：POSIX 共享内存对象放队列，进程共享互斥锁保护头尾指针，两个未命名信号量分别计数「空位」与「数据」。

/* shm_queue.c — 父进程生产、子进程消费的跨进程有界队列 */
#define _GNU_SOURCE
#include <errno.h>
#include <fcntl.h>
#include <pthread.h>
#include <semaphore.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/mman.h>
#include <sys/wait.h>
#include <unistd.h>

#define QUEUE_CAP 8
#define ITEMS     100

struct shm_queue {
    pthread_mutex_t lock;
    sem_t empty;                    /* 空位数 */
    sem_t full;                     /* 数据条数 */
    unsigned int head;
    unsigned int tail;
    int buf[QUEUE_CAP];
};

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

/* Pthreads 函数返回错误码而不设置 errno，必须用 strerror(rc) 报告 */
static void die_pthread(const char *what, int rc)
{
    fprintf(stderr, "%s: %s\n", what, strerror(rc));
    exit(EXIT_FAILURE);
}
#define PCHECK(expr) do { int rc_ = (expr); if (rc_ != 0) die_pthread(#expr, rc_); } while (0)

static void wait_sem(sem_t *sem)
{
    while (sem_wait(sem) == -1) {
        if (errno != EINTR)
            die("sem_wait");
    }
}

int main(void)
{
    const char *name = "/shm-queue-demo";
    if (shm_unlink(name) == -1 && errno != ENOENT)
        die("shm_unlink stale");

    int fd = shm_open(name, O_CREAT | O_RDWR | O_EXCL, 0600);
    if (fd == -1) die("shm_open");
    if (ftruncate(fd, (off_t)sizeof(struct shm_queue)) == -1)
        die("ftruncate");

    struct shm_queue *q = mmap(NULL, sizeof *q, PROT_READ | PROT_WRITE,
                               MAP_SHARED, fd, 0);
    if (q == MAP_FAILED) die("mmap");
    if (close(fd) == -1) die("close shm fd");

    /* 进程共享互斥锁：允许另一个进程加锁 */
    pthread_mutexattr_t mattr;
    PCHECK(pthread_mutexattr_init(&mattr));
    PCHECK(pthread_mutexattr_setpshared(&mattr, PTHREAD_PROCESS_SHARED));
    PCHECK(pthread_mutex_init(&q->lock, &mattr));
    pthread_mutexattr_destroy(&mattr);

    if (sem_init(&q->empty, 1, QUEUE_CAP) == -1) die("sem_init empty");
    if (sem_init(&q->full, 1, 0) == -1) die("sem_init full");
    q->head = q->tail = 0;

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

    if (pid == 0) {
        /* 消费者：取 ITEMS 条，校验顺序 */
        int expect = 0;
        for (int i = 0; i < ITEMS; i++) {
            wait_sem(&q->full);
            PCHECK(pthread_mutex_lock(&q->lock));
            int v = q->buf[q->head];
            q->head = (q->head + 1) % QUEUE_CAP;
            PCHECK(pthread_mutex_unlock(&q->lock));
            if (sem_post(&q->empty) == -1) die("post empty");
            if (v != expect) {
                fprintf(stderr, "order error: got %d want %d\n", v, expect);
                _exit(EXIT_FAILURE);
            }
            expect++;
        }
        _exit(EXIT_SUCCESS);
    }

    /* 生产者：放入 0..ITEMS-1 */
    for (int i = 0; i < ITEMS; i++) {
        wait_sem(&q->empty);
        PCHECK(pthread_mutex_lock(&q->lock));
        q->buf[q->tail] = i;
        q->tail = (q->tail + 1) % QUEUE_CAP;
        PCHECK(pthread_mutex_unlock(&q->lock));
        if (sem_post(&q->full) == -1) die("post full");
    }

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

    if (sem_destroy(&q->empty) == -1) die("sem_destroy empty");
    if (sem_destroy(&q->full) == -1) die("sem_destroy full");
    PCHECK(pthread_mutex_destroy(&q->lock));
    if (munmap(q, sizeof *q) == -1) die("munmap");
    if (shm_unlink(name) == -1) die("shm_unlink");
    return EXIT_SUCCESS;
}

运行与验证：

cc -std=c17 -D_GNU_SOURCE -Wall -Wextra -Werror -O2 -g -pthread shm_queue.c -o shm_queue -lrt
./shm_queue && echo "queue ok"

设计要点：

1. 信号量计数与互斥锁分工：empty/full 负责「能不能进」，互斥锁负责「进去后改头尾指针」。只用一个信号量或只用锁都不够。
2. sem_wait 在锁外：先等空位，再拿锁。反过来的话，一个生产者拿到锁后发现队列满，会持有锁睡眠，其他生产者连队列都进不去。
3. sem_post 在锁外：数据已经写好，唤醒等待者即可；放在锁内会多一次「唤醒后抢锁」的竞争。
4. 顺序保证：单生产者单消费者时，队列本身是 FIFO；多生产者时，入队顺序由锁的获取顺序决定，不保证与调用顺序一致。
5. 崩溃处理：消费者若在持锁时崩溃，生产者会卡在 pthread_mutex_lock。生产环境应使用 PTHREAD_MUTEX_ROBUST，并处理 EOWNERDEAD。

2.15.6 可见性、内存屏障与崩溃一致性

可见性不是「自动的」

共享内存的写最终会被其他进程看到，但顺序没有保证：

/* 生产者 */
q->buf[q->tail] = value;        /* ① 写数据 */
q->tail++;                      /* ② 发布位置 */

编译器和 CPU 都可能把 ② 重排到 ① 之前；另一个进程看到 tail 更新后去读 buf，可能读到旧值。x86-64 的 TSO 模型不允许「写-写」重排，所以这段在 x86 上通常没问题；ARM/POWER 允许，移植时会炸。

正确做法不是 volatile，而是用带屏障语义的原子操作发布：

#include <stdatomic.h>

struct queue {
    int buf[QUEUE_CAP];
    _Atomic unsigned int tail;          /* 必须是原子类型 */
};

/* 数据写完后，用 release 发布索引 */
atomic_store_explicit(&q->tail, new_tail, memory_order_release);

/* 消费者用 acquire 读取索引，再读数据 */
unsigned int tail = atomic_load_explicit(&q->tail, memory_order_acquire);

release 保证「之前的普通写」不会被重排到这条 store 之后；acquire 保证「之后的普通读」不会被重排到这条 load 之前。两者配对，建立 happens-before。信号量的 post/wait 本身带完整的屏障语义，所以上节的队列在 sem_post 之前写的数据，消费者 sem_wait 之后一定看得到；这就是为什么把数据写进 buf 再 post 是安全的。

崩溃一致性

共享内存 + 锁的组合有一个单进程程序没有的问题：持锁进程可能死掉。

• 普通 pthread_mutex_t：持有者死亡后锁永远不释放，其他进程永久阻塞。
• robust mutex：等待者得到 EOWNERDEAD，可以检查/修复共享数据，然后 pthread_mutex_consistent() 让锁恢复；不修复就 unlock 会让锁变成 ENOTRECOVERABLE。
• 信号量：没有所有者概念，进程死了不会「自动归还」它已经 wait 掉的计数。设计时要考虑「令牌丢失」：例如用超时、租约、或定期重建。

int rc = pthread_mutex_lock(&q->lock);
if (rc == EOWNERDEAD) {
    /* 上一个持有者死了：共享数据可能只更新了一半 */
    repair_queue(q);
    if (pthread_mutex_consistent(&q->lock) != 0) {
        /* 无法修复：让锁进入不可恢复状态，其他进程会得到 ENOTRECOVERABLE */
    }
} else if (rc == ENOTRECOVERABLE) {
    /* 共享区已不可信，只能重建 */
} else if (rc != 0) {
    /* 其他错误 */
}

/dev/shm 是 tmpfs，重启后清空；但进程崩溃不会清理 shm 对象和命名信号量。程序启动时应该用 O_EXCL 或 shm_unlink 处理残留，或者在名字里带 PID/启动时间，避免「上一次崩溃留下的队列」被误用。

2.15.7 System V 共享内存与信号量对照

老系统里常见 System V 版本：

/* 共享内存 */
int shmid = shmget(key, size, IPC_CREAT | 0600);
void *p = shmat(shmid, NULL, 0);
shmdt(p);
shmctl(shmid, IPC_RMID, NULL);

/* 信号量集 */
int semid = semget(key, nsems, IPC_CREAT | 0600);
struct sembuf op = { .sem_num = 0, .sem_op = -1, .sem_flg = 0 };
semop(semid, &op, 1);
semctl(semid, 0, SETVAL, 1);
semctl(semid, 0, IPC_RMID);

System V 信号量的 semop 支持「一次原子地操作多个信号量」，这是它相对 POSIX 单信号量的唯一明显优势；但它的 API 更难用、限制更分散、删除语义更绕。新代码优先 POSIX；维护旧系统时，用 ipcs -m/ipcs -s 查看残留，ipcrm -m ID/ipcrm -s ID 清理。

semop 数组与 SEM_UNDO：SysV 信号量的两个特有语义

semop(semid, sops, nsops) 可以一次操作多个信号量，内核保证这组操作要么全部完成，要么一个都不做（并可能阻塞）：

struct sembuf ops[2] = {
    { .sem_num = 0, .sem_op = -1, .sem_flg = SEM_UNDO },  /* P(s0) */
    { .sem_num = 1, .sem_op = +1, .sem_flg = 0 },         /* V(s1) */
};
if (semop(semid, ops, 2) == -1) { /* EINTR/EAGAIN/... */ }

SEM_UNDO 让内核记录「这个进程对每个信号量的净操作」，进程异常退出时自动反向补偿，防止持锁进程崩溃导致永久死锁。代价是内核要为每个 (semid, semnum, pid) 维护 undo 结构，进程多时内存/性能开销明显。

2.15.8 共享内存接口全表：shm_open、mmap 与 memfd

这一节回答：shm_open 到底做了什么系统调用？ftruncate 之后 mmap 为什么还会 SIGBUS？memfd_create 与 /dev/shm 相比多了什么？

创建与映射

shm_open 的 oflag 只认这几个：

mmap 的 flags 组合决定了「共不共享、是否匿名」：

ftruncate 与 SIGBUS

新创建的 shm 对象大小是 0。mmap 的 length 可以大于文件大小，映射本身会成功；但访问超出文件大小的页会触发 SIGBUS，而不是 SIGSEGV。这是共享内存程序最常见的崩溃：

/* 错误顺序：先 mmap 再 ftruncate，或忘了 ftruncate */
int fd = shm_open(name, O_CREAT | O_RDWR, 0600);
void *p = mmap(NULL, 4096, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
/* p[0] = 1;  -> SIGBUS，因为对象大小还是 0 */

正确顺序是先 ftruncate(fd, size)，再 mmap。如果多个进程协作，创建者应该在 ftruncate 之后、发布名字之前完成初始化；使用者打开后先 fstat 校验 st_size >= sizeof(struct shared)，不满足就报错退出，而不是硬访问。

ftruncate 缩小对象同样危险：如果还有进程映射了被截掉的页，访问会 SIGBUS。缩小之前必须确保没有映射者，或者用 mremap/重新 mmap 的协议。

memfd_create：没有名字的共享文件

memfd_create 创建一个匿名内存文件，行为像 tmpfs 文件，但不出现在 /dev/shm，只能通过 fd 传递（fork 继承、SCM_RIGHTS 传 fd）。它适合：

• 需要 mmap 共享、但不想管理名字与残留；
• 需要密封（seal）防止内容被改写，例如把配置/密钥交给子进程；
• 需要大页（MFD_HUGETLB）。

flags：

密封（seal）位：

F_SEAL_WRITE 的一个坑：如果当前存在 PROT_WRITE 的共享映射，加 seal 会返回 EBUSY；必须先 munmap 所有可写映射。F_SEAL_FUTURE_WRITE 则是为「已有可写映射不能撤销」的场景准备的。

2.15.9 信号量接口全表：sem_* 与 errno

这一节回答：sem_wait 被信号打断后要不要重试？sem_post 什么时候返回 EOVERFLOW？sem_timedwait 用的是哪条时钟？

创建、销毁与打开

sem_open 的 oflag 与 shm_open 类似：O_CREAT、O_EXCL，但没有 O_RDONLY/O_WRONLY/O_RDWR 的可移植语义；Linux 默认按读写打开。value 只在创建时生效，已存在时被忽略——这意味着「用 sem_open 重置一个已有信号量的初值」是做不到的，必须 sem_unlink 后重建。

等待与发布

sem_wait 的 EINTR 有一个反直觉的点：即使 sigaction 设了 SA_RESTART，sem_wait 仍可能返回 EINTR。POSIX 明确允许这一行为，Linux 就是这样。所以等待循环必须写成：

while (sem_wait(&sem) == -1) {
    if (errno != EINTR) { /* 真错误 */ }
}

sem_post 的 EOVERFLOW 只在信号量值已经等于 SEM_VALUE_MAX 时出现。Linux 的 SEM_VALUE_MAX 是 INT_MAX（2147483647），正常计数很难碰到；但如果把信号量当「无界队列长度」用，计数泄漏最终会溢出。溢出的 sem_post 不会增加计数，等待者也不会被唤醒——这是一个很难排查的挂死。

sem_timedwait 的绝对时间是 CLOCK_REALTIME。系统时间被向前调时，等待可能提前超时；被向后调时，可能多等。glibc 2.30 起提供 sem_clockwait，可以传 CLOCK_MONOTONIC，超时逻辑应该优先用它。

sem_t 的布局与复制禁令

sem_t 在 glibc 里是一个 32 字节的结构（不同架构可能不同），内部有一个 data 字段和一个等待队列相关的字段。复制 sem_t 是未定义行为：

sem_t a;
sem_init(&a, 1, 1);
sem_t b = a;          /* 禁止：b 与 a 不是同一个信号量 */
sem_wait(&b);         /* 未定义 */

同样，sem_open 返回的 sem_t * 不能解引用复制；所有操作都必须用原指针。把 sem_t 放进共享内存时，sem_init 只能由一个进程调用一次；多个进程同时 sem_init 是未定义行为。

2.15.10 内核路径：tmpfs、缺页异常与 futex

这一节回答：shm_open 创建的对象在内核里是什么？第一次写共享页发生了什么？sem_wait 阻塞时进程睡在哪里？

POSIX 共享内存 = tmpfs 文件 + 缺页异常

shm_open("/foo", ...) 在 glibc 里等价于 open("/dev/shm/foo", O_NOFOLLOW|O_CLOEXEC|...)。/dev/shm 是一个 tmpfs，内核里由 mm/shmem.c 实现。对象是 tmpfs 的 inode + page cache 页，不是特殊的「共享内存段」：

缺页异常是关键：mmap 只建立 VMA，不分配物理页；第一次读/写触发 shmem_fault，内核在 page cache 里查找对应偏移的页，没有就分配一页并清零，然后把它映射进进程页表。两个进程映射同一偏移，会命中同一个 page cache 页，所以看到同一份数据。

/* 按 Linux 6.8 mm/shmem.c 的结构简化，省略锁与错误分支 */
static vm_fault_t shmem_fault(struct vm_fault *vmf)
{
    struct inode *inode = file_inode(vmf->vma->vm_file);
    struct folio *folio;

    folio = shmem_get_folio(inode, vmf->pgoff, 0, SGP_CACHE, NULL);
    if (IS_ERR(folio))
        return vmf_error(PTR_ERR(folio));

    vmf->page = folio_file_page(folio, vmf->pgoff);
    return 0;
}

SIGBUS 就发生在这里：如果 vmf->pgoff 超出 i_size，shmem_get_folio 返回 -EINVAL，缺页处理失败，内核向进程投递 SIGBUS。这解释了为什么 ftruncate 是必须的，也解释了为什么「映射长度大于文件大小」不会在 mmap 时失败，而是在访问时崩溃。

MAP_SHARED 与 MAP_PRIVATE 的区别在 VMA 标志：VM_SHARED 置位时，缺页映射的是 page cache 页本身，写直接改页；MAP_PRIVATE 时，写触发 CoW，复制一份新页给该进程，其他进程看不到。

命名信号量 = tmpfs 文件 + sem_t

sem_open("/s", O_CREAT, 0600, 1) 在 /dev/shm/sem.s 创建一个文件，大小是 sizeof(sem_t)，内容是初始化好的 sem_t；sem_close 关闭 fd，sem_unlink 删名字。fork 后子进程继承 sem_t *，因为底层映射/文件描述符被继承。

sem_wait / sem_post 与 futex

glibc 的 sem_wait 走「快速路径 + futex 慢路径」：

/* 按 glibc nptl/sem_wait.c 的结构简化 */
int __new_sem_wait(sem_t *sem)
{
    int cur;

    do {
        cur = atomic_load_relaxed(&sem->data);
        if (cur == 0)
            return __new_sem_wait_slow(sem);   /* futex(FUTEX_WAIT) */
    } while (!atomic_compare_exchange_weak_relaxed(&sem->data, &cur, cur - 1));

    return 0;                                   /* 无竞争：纯用户态 */
}

sem_post 对称：先 atomic_fetch_add_release(&sem->data, 1)，如果之前是 0（可能有等待者），再 futex(FUTEX_WAKE, 1)。所以：

• 无竞争时，sem_wait/sem_post 不进入内核，只是原子指令；
• 有竞争时，才通过 futex 睡眠/唤醒；
• pshared=1 时，futex 操作不带 FUTEX_PRIVATE_FLAG，等待队列是全局的，能跨进程；pshared=0 时带 private flag，只能同进程线程间用。

futex 的等待队列 key 是共享内存的物理页 + 偏移，所以两个进程必须映射同一页、同一偏移，才能互相唤醒。把 sem_t 放在 MAP_PRIVATE 里，两个进程的虚拟地址不同、物理页也不同，sem_post 唤醒不了对方——这是「信号量不生效」的最常见原因。

robust mutex 的内核配合

pthread_mutex_lock 在竞争时也走 futex。普通互斥锁的持有者死亡后，内核不会做任何事；robust mutex 则依赖 set_robust_list 注册的链表：线程退出时，内核遍历 robust_list，把每个 robust mutex 的 owner 字段标成 FUTEX_OWNER_DIED，并唤醒一个等待者。等待者看到这个标志，pthread_mutex_lock 返回 EOWNERDEAD。

这解释了 robust mutex 的三个限制：

1. 只有线程正常/异常退出时才会触发；如果进程被 SIGKILL，内核仍会走 exit_robust_list，所以也能触发。
2. 只在「锁已经被持有」时有效；如果崩溃发生在 lock 之前，什么也不会发生。
3. 修复责任在等待者；内核只标记「owner 死了」，不知道你的数据结构是否一致。

sem_wait:  atomic dec 成功 ──▶ 返回（无系统调用）
              │ 失败（值为 0）
              ▼
         futex(FUTEX_WAIT, &sem->data, 0) ──▶ 睡眠
                                              │
sem_post:  atomic inc ──▶ 之前为 0? ──▶ futex(FUTEX_WAKE, 1)
                                              │
                                        唤醒等待者，重试 atomic dec

读图：信号量的「值」在用户态，内核只提供「在某个地址上睡眠/唤醒」的机制。这就是为什么 sem_post 是 async-signal-safe 的（只是一个原子加和可选系统调用），而 sem_wait 不是（可能睡眠）。

2.15.11 程序一：共享内存里的原子计数

这个程序对比 _Atomic long 与普通 long 在多个进程同时自增时的结果，并检查原子类型是否 lock-free——如果不是 lock-free，C11 原子操作可能用进程私有的锁，跨进程共享就是错的。

/* shm_atomic_counter.c — 多进程自增：原子 vs 普通 */
#define _GNU_SOURCE
#include <errno.h>
#include <fcntl.h>
#include <stdatomic.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/mman.h>
#include <sys/wait.h>
#include <unistd.h>

#define WORKERS 4
#define ITERS   100000

struct shared {
    _Atomic long counter;
    volatile long plain;
};

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

int main(void)
{
    const char *name = "/shm-atomic-demo";
    if (shm_unlink(name) == -1 && errno != ENOENT)
        die("shm_unlink stale");

    int fd = shm_open(name, O_CREAT | O_RDWR | O_EXCL, 0600);
    if (fd == -1) die("shm_open");
    if (ftruncate(fd, (off_t)sizeof(struct shared)) == -1) die("ftruncate");

    struct shared *p = mmap(NULL, sizeof *p, PROT_READ | PROT_WRITE,
                            MAP_SHARED, fd, 0);
    if (p == MAP_FAILED) die("mmap");
    if (close(fd) == -1) die("close");
    if (shm_unlink(name) == -1) die("shm_unlink");

    if (!atomic_is_lock_free(&p->counter)) {
        fprintf(stderr, "atomic long is not lock-free on this target; "
                        "cross-process atomics would use process-private locks\n");
        return EXIT_FAILURE;
    }

    atomic_store_explicit(&p->counter, 0, memory_order_relaxed);
    p->plain = 0;

    for (int w = 0; w < WORKERS; w++) {
        pid_t pid = fork();
        if (pid == -1) die("fork");
        if (pid == 0) {
            for (int i = 0; i < ITERS; i++) {
                atomic_fetch_add_explicit(&p->counter, 1,
                                          memory_order_relaxed);
                p->plain++;             /* 故意制造数据竞争 */
            }
            _exit(EXIT_SUCCESS);
        }
    }

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

    long atomic_v = atomic_load_explicit(&p->counter, memory_order_relaxed);
    printf("expected=%d atomic=%ld plain=%ld\n",
           WORKERS * ITERS, atomic_v, p->plain);

    if (munmap(p, sizeof *p) == -1) die("munmap");
    return atomic_v == WORKERS * ITERS ? EXIT_SUCCESS : EXIT_FAILURE;
}

编译运行：

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

预期输出：

expected=400000 atomic=400000 plain=213877

atomic 精确等于 400000；plain 几乎一定小于 400000（具体值每次不同），因为 p->plain++ 是「读-改-写」，多进程会丢更新。volatile 只阻止编译器把值缓存到寄存器，不提供原子性，也不提供内存屏障。

这个程序还暴露了一个跨平台陷阱：atomic_is_lock_free 为假时，C11 原子操作会退化为 libatomic 的锁实现，而锁在进程私有内存里，两个进程各锁各的，等于没锁。共享内存里的原子类型必须是 lock-free 的；不是的话，改用 __atomic_* 内建 + 平台保证，或者干脆用信号量。

2.15.12 程序二：robust mutex 与持有者崩溃

这个程序让子进程在持锁状态下 _exit，父进程随后加锁，验证 EOWNERDEAD、修复数据、pthread_mutex_consistent() 的完整流程。

/* shm_robust.c — 进程共享 robust mutex：持有者崩溃后的恢复 */
#define _GNU_SOURCE
#include <errno.h>
#include <fcntl.h>
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/mman.h>
#include <sys/wait.h>
#include <unistd.h>

struct shared {
    pthread_mutex_t lock;
    int value;
    int phase;          /* 0=初始 1=写了一半 2=已修复 */
};

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

#define PCHECK(expr) do { \
    int rc_ = (expr); \
    if (rc_ != 0) { \
        fprintf(stderr, "%s: %s\n", #expr, strerror(rc_)); \
        exit(EXIT_FAILURE); \
    } \
} while (0)

int main(void)
{
    const char *name = "/shm-robust-demo";
    if (shm_unlink(name) == -1 && errno != ENOENT)
        die("shm_unlink stale");

    int fd = shm_open(name, O_CREAT | O_RDWR | O_EXCL, 0600);
    if (fd == -1) die("shm_open");
    if (ftruncate(fd, (off_t)sizeof(struct shared)) == -1) die("ftruncate");

    struct shared *p = mmap(NULL, sizeof *p, PROT_READ | PROT_WRITE,
                            MAP_SHARED, fd, 0);
    if (p == MAP_FAILED) die("mmap");
    close(fd);
    shm_unlink(name);

    pthread_mutexattr_t mattr;
    PCHECK(pthread_mutexattr_init(&mattr));
    PCHECK(pthread_mutexattr_setpshared(&mattr, PTHREAD_PROCESS_SHARED));
    PCHECK(pthread_mutexattr_setrobust(&mattr, PTHREAD_MUTEX_ROBUST));
    PCHECK(pthread_mutex_init(&p->lock, &mattr));
    pthread_mutexattr_destroy(&mattr);

    p->value = 0;
    p->phase = 0;

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

    if (pid == 0) {
        int rc = pthread_mutex_lock(&p->lock);
        if (rc != 0) _exit(2);
        p->value = 123;
        p->phase = 1;               /* 临界区只完成一半 */
        _exit(0);                   /* 不 unlock，模拟崩溃 */
    }

    int status;
    if (waitpid(pid, &status, 0) == -1) die("waitpid");

    int rc = pthread_mutex_lock(&p->lock);
    if (rc == EOWNERDEAD) {
        printf("lock: EOWNERDEAD (phase=%d value=%d)\n", p->phase, p->value);
        if (p->phase == 1) {
            p->value = 456;         /* 修复不一致状态 */
            p->phase = 2;
        }
        rc = pthread_mutex_consistent(&p->lock);
        if (rc != 0) {
            fprintf(stderr, "pthread_mutex_consistent: %s\n", strerror(rc));
            return EXIT_FAILURE;
        }
    } else if (rc != 0) {
        fprintf(stderr, "pthread_mutex_lock: %s\n", strerror(rc));
        return EXIT_FAILURE;
    }
    PCHECK(pthread_mutex_unlock(&p->lock));

    /* 锁应恢复可用 */
    PCHECK(pthread_mutex_lock(&p->lock));
    printf("after recovery: phase=%d value=%d\n", p->phase, p->value);
    PCHECK(pthread_mutex_unlock(&p->lock));

    PCHECK(pthread_mutex_destroy(&p->lock));
    if (munmap(p, sizeof *p) == -1) die("munmap");
    return EXIT_SUCCESS;
}

编译运行：

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

预期输出：

lock: EOWNERDEAD (phase=1 value=123)
after recovery: phase=2 value=456

关键点：

• 子进程 _exit 时不执行 pthread_mutex_unlock，但内核在进程退出时走 exit_robust_list，把锁标成 owner died；
• 父进程 pthread_mutex_lock 返回 EOWNERDEAD，而不是永久阻塞；
• 修复后必须 pthread_mutex_consistent，否则 unlock 会把锁变成 ENOTRECOVERABLE，后续所有加锁都失败；
• 修复逻辑必须能判断「临界区做到哪一步」。本程序用 phase 字段；真实系统要有日志、校验和或可重建的数据结构。

把 PTHREAD_MUTEX_ROBUST 去掉再跑，父进程会永久阻塞在 pthread_mutex_lock——这就是普通进程共享互斥锁的崩溃后果。

2.15.13 程序三：无锁 SPSC 环形队列

这个程序只用 _Atomic 的 acquire/release 实现单生产者单消费者队列，不用锁、不用信号量，验证内存屏障足以建立跨进程的 happens-before。

/* shm_spsc.c — 无锁单生产者/单消费者环形队列 */
#define _GNU_SOURCE
#include <errno.h>
#include <sched.h>
#include <stdatomic.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/mman.h>
#include <sys/wait.h>
#include <unistd.h>

#define CAP   1024
#define ITEMS 200000

struct ring {
    _Atomic size_t head;    /* 生产者发布 */
    _Atomic size_t tail;    /* 消费者发布 */
    int buf[CAP];
};

int main(void)
{
    struct ring *r = mmap(NULL, sizeof *r, PROT_READ | PROT_WRITE,
                          MAP_SHARED | MAP_ANONYMOUS, -1, 0);
    if (r == MAP_FAILED) {
        perror("mmap");
        return EXIT_FAILURE;
    }

    if (!atomic_is_lock_free(&r->head) || !atomic_is_lock_free(&r->tail)) {
        fprintf(stderr, "atomic size_t is not lock-free\n");
        return EXIT_FAILURE;
    }

    atomic_store_explicit(&r->head, 0, memory_order_relaxed);
    atomic_store_explicit(&r->tail, 0, memory_order_relaxed);

    pid_t pid = fork();
    if (pid == -1) { perror("fork"); return EXIT_FAILURE; }

    if (pid == 0) {
        /* 生产者 */
        for (int i = 0; i < ITEMS; i++) {
            size_t h, t;
            for (;;) {
                h = atomic_load_explicit(&r->head, memory_order_relaxed);
                t = atomic_load_explicit(&r->tail, memory_order_acquire);
                if (h - t < CAP)
                    break;
                sched_yield();          /* 队列满 */
            }
            r->buf[h % CAP] = i;
            atomic_store_explicit(&r->head, h + 1, memory_order_release);
        }
        _exit(0);
    }

    /* 消费者 */
    int bad = 0;
    for (int i = 0; i < ITEMS; i++) {
        size_t h, t;
        for (;;) {
            t = atomic_load_explicit(&r->tail, memory_order_relaxed);
            h = atomic_load_explicit(&r->head, memory_order_acquire);
            if (h - t > 0)
                break;
            sched_yield();              /* 队列空 */
        }
        int v = r->buf[t % CAP];
        atomic_store_explicit(&r->tail, t + 1, memory_order_release);
        if (v != i) {
            fprintf(stderr, "order error at %d: got %d\n", i, v);
            bad = 1;
            break;
        }
    }

    int status;
    if (waitpid(pid, &status, 0) == -1) { perror("waitpid"); return EXIT_FAILURE; }
    printf("spsc: %s\n", bad ? "FAILED" : "ok");
    if (munmap(r, sizeof *r) == -1) { perror("munmap"); return EXIT_FAILURE; }
    return bad ? EXIT_FAILURE : EXIT_SUCCESS;
}

编译运行：

cc -std=c17 -D_GNU_SOURCE -Wall -Wextra -Werror -O2 -g \
   shm_spsc.c -o shm_spsc
./shm_spsc
# spsc: ok

关键点：

• head/tail 是单调递增的 size_t，用 % CAP 取模；这样「满」与「空」的判定不需要浪费一个槽位；
• 生产者写 buf 后用 memory_order_release 发布 head，消费者用 memory_order_acquire 读 head，保证「看到新 head 就能看到 buf 里的数据」；
• 反向的 tail 也一样：消费者释放槽位用 release，生产者 acquire 读取；
• sched_yield 只是让出 CPU，不保证公平；真实系统会用 eventfd/futex 在空/满时睡眠，避免忙等。

这个队列只支持单生产者单消费者。多生产者时 head 的 CAS 竞争、多消费者时 tail 的 CAS 竞争都需要额外的原子操作；更复杂的是「多个生产者同时判断队列满」，需要 fetch_add 预留槽位或 CAS 循环。不要把一个 SPSC 队列直接改成 MPSC 就上线。

2.15.14 实验：/dev/shm、SIGBUS 与 futex

实验一：共享内存的生命周期与 SIGBUS

# 1. 看 tmpfs 配额与现有对象
df -h /dev/shm
ls -l /dev/shm

# 2. 运行程序，观察对象出现/消失
./shm_atomic_counter & pid=$!
ls -l /dev/shm | grep shm-atomic
wait $pid
ls -l /dev/shm | grep shm-atomic || echo "gone"

# 3. 看系统调用：shm_open 实际是 openat("/dev/shm/...")
strace -e trace=openat,ftruncate,mmap,munmap,unlink \
       ./shm_atomic_counter 2>&1 | grep -E "shm|ftruncate|mmap" | head

# 4. 看映射
./shm_spsc & pid=$!
cat /proc/$pid/maps | grep -E "shm|SYSV|/dev/shm"
pmap -x $pid | head
wait $pid

SIGBUS 复现（单独编译一个小程序）：

/* sigbus_demo.c — 映射超过对象大小，访问越界页 */
#define _GNU_SOURCE
#include <fcntl.h>
#include <stdio.h>
#include <stdlib.h>
#include <sys/mman.h>
#include <unistd.h>

int main(void)
{
    int fd = shm_open("/sigbus-demo", O_CREAT | O_RDWR, 0600);
    if (fd == -1) { perror("shm_open"); return 1; }
    if (ftruncate(fd, 4096) == -1) { perror("ftruncate"); return 1; }

    char *p = mmap(NULL, 8192, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
    if (p == MAP_FAILED) { perror("mmap"); return 1; }

    p[0] = 'A';                 /* 第一页：正常 */
    printf("first page ok\n");
    fflush(stdout);
    p[4096] = 'B';              /* 第二页超出对象大小：SIGBUS */
    printf("second page ok (unexpected)\n");
    return 0;
}

cc -std=c17 -D_GNU_SOURCE -Wall -Wextra -Werror sigbus_demo.c -o sigbus_demo -lrt
./sigbus_demo; echo "exit=$?"
# 预期：打印 first page ok，然后 Bus error，exit=135 (128+SIGBUS)

怎么读：mmap 成功不代表能访问。ftruncate 决定对象大小，mmap 的 length 只是「我想映射多大」；超出对象大小的页在缺页时失败，内核发 SIGBUS。SIGBUS 的退出码是 128+7=135（shell 显示 Bus error）。

实验二：futex、robust 与信号量开销

# 1. 看 sem_wait/sem_post 是否进入内核
strace -f -e trace=futex ./shm_spsc 2>&1 | head
# 预期：无竞争时几乎看不到 futex；有竞争时出现 FUTEX_WAIT/FUTEX_WAKE

# 2. 用 bpftrace 统计 futex 操作（需要 root）
sudo bpftrace -e '
tracepoint:syscalls:sys_enter_futex /comm=="shm_spsc"/ {
    @op[args->op] = count();
}
interval:s:3 { exit(); }'

# 3. 观察 robust mutex 的退出清理
strace -f -e trace=set_robust_list,futex ./shm_robust 2>&1 | tail -30
# 预期：子进程退出时内核处理 robust list；父进程 futex 返回 EOWNERDEAD

# 4. 对比有锁/无锁的上下文切换
perf stat -e context-switches,page-faults,cycles ./shm_spsc
perf stat -e context-switches,page-faults,cycles ./shm_queue

怎么读：

• shm_spsc 在队列未满/未空时完全在用户态跑，strace 看不到 futex；shm_queue 每次 sem_wait/sem_post 无竞争时也只是原子操作，但队列满/空时会 futex 睡眠。
• perf stat 的 context-switches 高，说明睡眠/唤醒多；page-faults 在第一次触碰共享页时出现，之后稳定。
• set_robust_list 只在启用 robust mutex 的线程上出现；没有它，持有者死亡后锁不会自动标记。

2.15.15 陷阱清单：共享内存与信号量的十五个现场

2.15.16 本章出口检查

母本对应

下一站：2.16 套接字基础把通信从「同一台机器」扩展到「两台机器」：socket/bind/listen/accept 每一步在内核里做了什么，TCP 与 UDP 的编程差异，以及为什么 accept 返回的新 fd 不继承 O_NONBLOCK。