2.18 备选 I/O：select/poll/epoll/io_uring

2.17 的服务端一次只服务一条连接：accept 之后就在 read/write 上阻塞，直到这条连接结束。要同时服务一万条连接，就得一万个线程/进程——内存与调度开销都不可接受。I/O 多路复用换了个思路：让一个线程同时等多个 fd，谁就绪就处理谁。

四种接口是同一需求的不同代际：select（1983）用位图，受 FD_SETSIZE 限制；poll（1997）用数组，去掉数量限制但仍要线性扫描；epoll（2002）把「注册」与「等待」分开，内核维护兴趣集合与就绪链表；io_uring（2019）进一步把「提交」与「收割」做成两个环形队列，系统调用可以批量、可以异步完成。

这一章的重点不是 API 列表，而是两种触发语义与事件循环的状态机。LT/ET 写错，症状是「压测时连接卡住」或「CPU 100% 空转」；事件循环里少处理一个 EPOLLERR，症状是「fd 泄漏」。这些都不是靠背函数签名能避免的。

2.18.1 五种 I/O 模型与「就绪」的含义

 阻塞 I/O          非阻塞 I/O          I/O 多路复用       信号驱动 I/O        异步 I/O
┌─────────┐       ┌─────────┐        ┌─────────┐       ┌─────────┐       ┌─────────┐
│ read()  │       │ read()  │        │ select  │       │ sigaction│       │ io_uring│
│  阻塞   │       │  EAGAIN │        │ poll    │       │ SIGIO    │       │ submit  │
│  等待   │       │  轮询   │        │ epoll   │       │ 等待信号 │       │ 立即返回│
└────┬────┘       └────┬────┘        └────┬────┘       └────┬────┘       └────┬────┘
     │                 │                  │                  │                  │
     ▼                 ▼                  ▼                  ▼                  ▼
 数据拷贝          数据拷贝           通知「就绪」        通知「就绪」        通知「完成」
 后返回            后返回             再 read            再 read            数据已在缓冲区

读图：前四种模型都是同步的——「就绪」只表示「现在读不会阻塞」，真正的数据拷贝仍由 read 完成，仍可能阻塞一小段时间。只有异步 I/O 的「完成」表示数据已经被内核放进你提供的缓冲区。io_uring 同时支持「就绪」与「完成」两种用法，但它的目标是后者。

区分两个词：

• 就绪（readiness）：fd 上有数据可读/可写，调用 read/write 不会阻塞（至少能读到一个字节）；
• 完成（completion）：内核已经把数据拷贝到用户缓冲区，或已经把用户缓冲区写完。

select/poll/epoll 都是就绪通知。它们不减少数据拷贝次数，只减少「等待」的次数。io_uring 的 IORING_OP_READ 才是完成通知：提交后不用再调 read。

2.18.2 select：位图、1024 与「入参被修改」

#include <sys/select.h>

fd_set rfds;
FD_ZERO(&rfds);
FD_SET(fd, &rfds);

struct timeval tv = { .tv_sec = 5, .tv_usec = 0 };
int n = select(fd + 1, &rfds, NULL, NULL, &tv);
if (n == -1) {
    if (errno == EINTR) { /* 被信号打断 */ }
} else if (n == 0) {
    /* 超时 */
} else if (FD_ISSET(fd, &rfds)) {
    /* fd 可读 */
}

四个必须记住的细节：

1. nfds 是「最大 fd + 1」，不是 fd 个数。内核要遍历 0..nfds-1。
2. fd_set 是入参兼出参。select 返回后它只保留「就绪的 fd」，下一轮必须重新 FD_ZERO/FD_SET。这是最常见的 bug：循环里忘了重建，第二轮起就永远等不到。
3. FD_SETSIZE 默认 1024。fd 号大于等于 1024 时 FD_SET 越界写栈，行为未定义。改 FD_SETSIZE 要重新编译整个程序（包括 libc 内部使用它的地方），不现实。
4. timeout 在 Linux 上会被修改。返回后 tv 是「剩余时间」，循环里复用必须重新赋值。

select 的优点是可移植：POSIX、Windows、所有 Unix 都有。缺点是 fd 多时每次调用都要把整个位图从用户态拷到内核、再线性扫描，复杂度 O(n)；1024 的上限让它不适合现代服务端。

2.18.3 poll：数组、无上限、仍要线性扫描

#include <poll.h>

struct pollfd fds[2] = {
    { .fd = fd0, .events = POLLIN },
    { .fd = fd1, .events = POLLIN },
};

int n = poll(fds, 2, 1000);          /* 超时 1000 ms；-1 表示永久 */
if (n == -1) {
    /* EINTR 或错误 */
} else {
    for (int i = 0; i < 2; i++) {
        if (fds[i].revents & POLLIN)  { /* 可读 */ }
        if (fds[i].revents & POLLOUT) { /* 可写 */ }
        if (fds[i].revents & (POLLERR | POLLHUP | POLLNVAL)) {
            /* 错误、挂断、fd 无效：无论 events 是否请求都会返回 */
        }
    }
}

与 select 相比：

• 没有 FD_SETSIZE 限制，fd 数量只受内存与 RLIMIT_NOFILE 限制；
• events 是入参，revents 是出参，不需要每轮重建（但每次调用前仍要把 revents 清零，或依赖内核覆盖）；
• 仍然是 O(n)：内核要遍历整个数组，用户态也要遍历找就绪项；
• 超时单位是毫秒（int），比 timeval 简单。

POLLERR、POLLHUP、POLLNVAL 是总是报告的，不受 events 过滤。很多代码只检查 POLLIN，结果对端关闭（POLLHUP）时死循环——poll 一直返回，read 一直返回 0，CPU 打满。

poll 适合「fd 数量几百、跨平台要求不高」的场景。上万连接时，每次 poll 的线性扫描成为瓶颈。

2.18.4 epoll：兴趣集合与就绪链表

epoll 把「注册」与「等待」拆成两个系统调用：

int epfd = epoll_create1(EPOLL_CLOEXEC);
if (epfd == -1) { perror("epoll_create1"); }

struct epoll_event ev = {
    .events = EPOLLIN | EPOLLRDHUP,   /* 关心可读与对端半关闭 */
    .data.fd = fd,                    /* 或 data.ptr，随你 */
};
if (epoll_ctl(epfd, EPOLL_CTL_ADD, fd, &ev) == -1) { perror("epoll_ctl ADD"); }

struct epoll_event events[64];
int n = epoll_wait(epfd, events, 64, -1);   /* -1：永久等待 */
for (int i = 0; i < n; i++) {
    if (events[i].events & EPOLLIN) { /* 可读 */ }
}

读图：epoll_ctl(ADD) 把一个 epitem 挂进内核的红黑树（兴趣集合），同时向目标 socket 的等待队列注册一个回调 ep_poll_callback。数据到达时，回调把 epitem 挂到就绪链表 rdllist，并唤醒 epoll_wait。epoll_wait 只取就绪链表，不扫描全部 fd——这是它相对 poll 的复杂度优势：注册 O(log n)，等待 O(就绪数)。

epoll 的三个关键语义：

1. epoll_wait 返回的是「当前就绪」的 fd，不是「自上次以来变化」的 fd。LT 下只要还有数据，下一次 epoll_wait 仍会返回它。
2. close(fd) 会自动从 epoll 移除，但如果 fd 被 dup 过，要等所有副本都关闭。epoll_ctl(DEL) 不是必须的，但显式删除能避免「fd 号被复用后仍挂在旧事件上」的困惑。
3. data 是唯一能带回用户态的东西。用 data.fd 最简单；用 data.ptr 可以直接带回连接对象指针，省一次查找。注意对象生命周期：epoll_wait 返回的事件数组里可能包含已经关闭的 fd，处理前要确认对象还有效。

LT 与 ET

ET 的纪律：

1. fd 必须非阻塞。ET 通知后你要一直读到 EAGAIN；如果 fd 阻塞，最后一次 read 会卡住整个事件循环。
2. 必须读到 EAGAIN。只读一次会留下未读数据，而 ET 不会再通知，连接就「饿死」了。
3. 写也要处理 EAGAIN。ET 下写缓冲区满时 write 返回 EAGAIN，必须把剩余数据存进用户缓冲区，等 EPOLLOUT 再写。
4. accept 也要循环到 EAGAIN。监听 fd 设为 ET 后，一次事件可能对应多个已完成连接。

EPOLLONESHOT 让 fd 在一次事件后自动从兴趣集合里禁用，处理完必须 epoll_ctl(MOD) 重新 armed。它适合「同一个连接可能被多个线程处理」的场景，避免两个线程同时读同一条连接。

EPOLLEXCLUSIVE（Linux 4.5+）用于 accept：多个进程/线程监听同一个 fd 时，内核只唤醒一个，避免惊群。它只能用于 EPOLL_CTL_ADD，不能与 EPOLLONESHOT 同用。

2.18.5 LT/ET 两版 echo 服务端

下面的程序用一个 mode 参数切换 LT/ET：ET 模式加 EPOLLET，读循环一直读到 EAGAIN；LT 模式每次事件只读一轮。两种模式都使用非阻塞 fd 与用户态输出缓冲区，因为 ET 下写也可能 EAGAIN。

/* epoll_echo.c：epoll 版 echo 服务端，支持 lt/et 两种模式 */
#include <errno.h>
#include <netinet/in.h>
#include <signal.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/epoll.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <unistd.h>

#define PORT 12348
#define MAX_EVENTS 64
#define IN_BUF 4096
#define OUT_BUF 16384

struct conn {
    int fd;
    size_t out_len;
    char out[OUT_BUF];
};

static void conn_destroy(struct conn *c)
{
    if (c->fd >= 0)
        close(c->fd);
    free(c);
}

/* 把输出缓冲区尽量写出去；返回 -1 表示连接应关闭 */
static int conn_flush(struct conn *c)
{
    size_t off = 0;
    while (off < c->out_len) {
        ssize_t n = write(c->fd, c->out + off, c->out_len - off);
        if (n == -1) {
            if (errno == EINTR)
                continue;
            if (errno == EAGAIN || errno == EWOULDBLOCK)
                break;                 /* 内核缓冲区满，等 EPOLLOUT */
            return -1;
        }
        off += (size_t)n;
    }
    if (off > 0) {
        memmove(c->out, c->out + off, c->out_len - off);
        c->out_len -= off;
    }
    return 0;
}

static int conn_append(struct conn *c, const char *data, size_t len)
{
    if (c->out_len + len > sizeof c->out)
        return -1;                     /* 输出积压过多：断开，避免内存无限增长 */
    memcpy(c->out + c->out_len, data, len);
    c->out_len += len;
    return 0;
}

int main(int argc, char **argv)
{
    int et = (argc > 1 && strcmp(argv[1], "et") == 0);
    uint32_t base_events = EPOLLIN | EPOLLRDHUP | (et ? EPOLLET : 0);

    int lfd = socket(AF_INET, SOCK_STREAM | SOCK_CLOEXEC | SOCK_NONBLOCK, 0);
    if (lfd == -1) {
        perror("socket");
        return EXIT_FAILURE;
    }

    int one = 1;
    if (setsockopt(lfd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof one) == -1) {
        perror("setsockopt");
        return EXIT_FAILURE;
    }

    struct sockaddr_in addr = {0};
    addr.sin_family = AF_INET;
    addr.sin_port = htons(PORT);
    addr.sin_addr.s_addr = htonl(INADDR_ANY);
    if (bind(lfd, (struct sockaddr *)&addr, sizeof addr) == -1) {
        perror("bind");
        return EXIT_FAILURE;
    }
    if (listen(lfd, 128) == -1) {
        perror("listen");
        return EXIT_FAILURE;
    }

    int epfd = epoll_create1(EPOLL_CLOEXEC);
    if (epfd == -1) {
        perror("epoll_create1");
        return EXIT_FAILURE;
    }

    struct epoll_event ev = { .events = base_events, .data.ptr = NULL };
    if (epoll_ctl(epfd, EPOLL_CTL_ADD, lfd, &ev) == -1) {
        perror("epoll_ctl listen");
        return EXIT_FAILURE;
    }

    printf("epoll echo (%s) on %d\n", et ? "ET" : "LT", PORT);
    fflush(stdout);

    struct epoll_event events[MAX_EVENTS];
    for (;;) {
        int n = epoll_wait(epfd, events, MAX_EVENTS, -1);
        if (n == -1) {
            if (errno == EINTR)
                continue;
            perror("epoll_wait");
            break;
        }

        for (int i = 0; i < n; i++) {
            struct conn *c = events[i].data.ptr;

            if (c == NULL) {           /* 监听 fd */
                for (;;) {
                    int cfd = accept4(lfd, NULL, NULL,
                                      SOCK_NONBLOCK | SOCK_CLOEXEC);
                    if (cfd == -1) {
                        if (errno == EAGAIN || errno == EWOULDBLOCK)
                            break;     /* ET 下必须取空 accept 队列 */
                        if (errno == EINTR)
                            continue;
                        perror("accept4");
                        break;
                    }

                    c = calloc(1, sizeof *c);
                    if (c == NULL) {
                        close(cfd);
                        continue;
                    }
                    c->fd = cfd;

                    struct epoll_event cev = {
                        .events = base_events,
                        .data.ptr = c,
                    };
                    if (epoll_ctl(epfd, EPOLL_CTL_ADD, cfd, &cev) == -1) {
                        perror("epoll_ctl conn");
                        conn_destroy(c);
                    }
                }
                continue;
            }

            if (events[i].events & (EPOLLERR | EPOLLHUP)) {
                conn_destroy(c);
                continue;
            }

            if (events[i].events & EPOLLIN) {
                char buf[IN_BUF];
                for (;;) {
                    ssize_t r = read(c->fd, buf, sizeof buf);
                    if (r > 0) {
                        if (conn_append(c, buf, (size_t)r) == -1) {
                            conn_destroy(c);
                            c = NULL;
                            break;
                        }
                    } else if (r == 0) {
                        conn_destroy(c);
                        c = NULL;
                        break;
                    } else {
                        if (errno == EINTR)
                            continue;
                        if (errno == EAGAIN || errno == EWOULDBLOCK)
                            break;        /* 读空了 */
                        conn_destroy(c);
                        c = NULL;
                        break;
                    }
                    if (!et)
                        break;            /* LT：留一点给别的连接，下次还会通知 */
                }
                if (c == NULL)
                    continue;
            }

            if (c->out_len > 0) {
                if (conn_flush(c) == -1) {
                    conn_destroy(c);
                    continue;
                }
            }

            /* 有积压就订阅 EPOLLOUT，清空后取消订阅 */
            struct epoll_event cev = {
                .events = base_events | (c->out_len > 0 ? EPOLLOUT : 0),
                .data.ptr = c,
            };
            if (epoll_ctl(epfd, EPOLL_CTL_MOD, c->fd, &cev) == -1) {
                perror("epoll_ctl MOD");
                conn_destroy(c);
            }

            if (events[i].events & EPOLLOUT) {
                /* 下一轮 EPOLLOUT 会再次进入上面的 flush 逻辑 */
            }
        }
    }

    close(epfd);
    close(lfd);
    return EXIT_SUCCESS;
}

cc -std=c17 -D_GNU_SOURCE -Wall -Wextra -Werror -O2 -g epoll_echo.c -o epoll_echo
./epoll_echo lt        # 水平触发
./epoll_echo et        # 边缘触发
# 另一个终端：
printf 'hello\n' | nc 127.0.0.1 12348

hello

两种模式都能 echo，但内部路径不同：

• LT：每次 epoll_wait 都告诉你「可读」，读一次就返回，剩下的数据下一轮再读。实现简单，但数据多时通知次数多。
• ET：只在「有新数据」时通知一次，必须循环读到 EAGAIN；写方向也要在 EPOLLOUT 时继续 flush。

程序里有两个容易写错的点：

1. accept 循环到 EAGAIN。即使 LT，accept 一次也可能取不完 accept 队列；ET 下则是必须。
2. conn_destroy 后立刻 continue。epoll_wait 返回的事件数组里，同一个 fd 只会出现一次，但后续事件可能指向已释放的对象——所以关闭连接后不能继续访问 c。

2.18.6 eventfd、timerfd、signalfd：把一切变成 fd

事件循环最舒服的形态是「所有事件源都是 fd」。Linux 提供了三个把非 fd 事件变成 fd 的接口：

eventfd 的典型用途是跨线程唤醒：一个线程往 eventfd 写 1，epoll_wait 立刻返回，事件循环处理队列里的任务。

int efd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
uint64_t one = 1;
if (write(efd, &one, sizeof one) != sizeof one) { /* 错误 */ }
/* 事件循环里： */
uint64_t val;
if (read(efd, &val, sizeof val) == sizeof val) {
    /* val 是累计计数 */
}

timerfd 在 2.11 讲过，signalfd 在 2.10 讲过。三者与 epoll 组合，就不需要 self-pipe、不需要信号处理器、不需要单独的定时器线程。

2.18.7 io_uring：提交队列与完成队列

io_uring 的核心是两个环形缓冲区，位于用户态与内核共享的内存里：

        用户态                                  内核态
┌─────────────────────────┐          ┌─────────────────────────┐
│ SQ ring（提交队列）      │  mmap    │ 内核读取 SQ entry        │
│  tail ──► [SQE][SQE]... │ ───────► │ 执行 read/write/accept  │
└─────────────────────────┘          └────────────┬────────────┘
                                                   │ 完成
┌─────────────────────────┐          ┌────────────▼────────────┐
│ CQ ring（完成队列）      │  mmap    │ 内核写入 CQE（res）      │
│ [CQE][CQE]... ◄── head  │ ◄─────── │                         │
└─────────────────────────┘          └─────────────────────────┘

一次 io_uring_enter 可以提交多个 SQE、收割多个 CQE，系统调用次数从「每个 I/O 一次」降到「一批一次」。配合 IORING_SETUP_SQPOLL（内核线程轮询 SQ），甚至可以不进内核就提交。

最小使用分四步：

1. io_uring_setup(entries, &params) 创建 ring，返回 fd；
2. mmap SQ ring、CQ ring、SQE 数组；
3. 填 SQE、更新 SQ tail、io_uring_enter 提交；
4. 从 CQ head 收割 CQE，更新 head。

直接用系统调用需要处理内存屏障与环形队列索引，容易出错。生产代码用 liburing，它把这些封装成 io_uring_get_sqe/io_uring_submit/io_uring_wait_cqe。

下面的程序用 liburing 批量读一个文件：一次提交 4 个读请求，再按顺序把结果写到 stdout。

/* uring_cat.c：用 io_uring 批量读文件（需要 liburing） */
#include <fcntl.h>
#include <liburing.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <unistd.h>

#define QD 4
#define BUF_SIZE 4096

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

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

    int fd = open(argv[1], O_RDONLY);
    if (fd == -1)
        die("open");

    struct io_uring ring;
    int rc = io_uring_queue_init(QD, &ring, 0);
    if (rc < 0) {
        fprintf(stderr, "io_uring_queue_init: %s\n", strerror(-rc));
        return EXIT_FAILURE;
    }

    char *bufs[QD];
    for (int i = 0; i < QD; i++) {
        bufs[i] = malloc(BUF_SIZE);
        if (bufs[i] == NULL)
            die("malloc");
    }

    off_t offset = 0;
    int eof = 0;

    while (!eof) {
        int submitted = 0;

        for (int i = 0; i < QD; i++) {
            struct io_uring_sqe *sqe = io_uring_get_sqe(&ring);
            if (sqe == NULL)
                die("io_uring_get_sqe");

            io_uring_prep_read(sqe, fd, bufs[i], BUF_SIZE, offset + (off_t)i * BUF_SIZE);
            sqe->user_data = (unsigned long)i;
            submitted++;
        }

        rc = io_uring_submit(&ring);
        if (rc < 0) {
            fprintf(stderr, "io_uring_submit: %s\n", strerror(-rc));
            break;
        }

        /* 按 user_data 收集结果，再按偏移顺序输出，保证输出顺序 */
        ssize_t got[QD];
        memset(got, 0, sizeof got);

        for (int i = 0; i < submitted; i++) {
            struct io_uring_cqe *cqe;
            rc = io_uring_wait_cqe(&ring, &cqe);
            if (rc < 0) {
                fprintf(stderr, "io_uring_wait_cqe: %s\n", strerror(-rc));
                goto out;
            }
            int idx = (int)cqe->user_data;
            got[idx] = cqe->res;       /* 负数是 -errno */
            io_uring_cqe_seen(&ring, cqe);
        }

        for (int i = 0; i < submitted; i++) {
            if (got[i] < 0) {
                fprintf(stderr, "read: %s\n", strerror((int)-got[i]));
                goto out;
            }
            if (got[i] == 0) {
                eof = 1;
                break;
            }
            if (write(STDOUT_FILENO, bufs[i], (size_t)got[i]) != got[i]) {
                perror("write");
                goto out;
            }
            if (got[i] < BUF_SIZE) {
                eof = 1;               /* 短读：文件结束 */
                break;
            }
        }

        offset += (off_t)QD * BUF_SIZE;
    }

out:
    for (int i = 0; i < QD; i++)
        free(bufs[i]);
    io_uring_queue_exit(&ring);
    close(fd);
    return EXIT_SUCCESS;
}

# Debian/Ubuntu：sudo apt install liburing-dev
cc -std=c17 -D_GNU_SOURCE -Wall -Wextra -Werror -O2 -g uring_cat.c -o uring_cat -luring
./uring_cat /etc/hosts

这个例子故意保持「一次提交 QD 个、收齐再写」，所以比同步 read 快不了多少；它的意义是展示批量提交与乱序完成：CQE 可能以任意顺序到达，程序用 user_data 把结果对回缓冲区，再按文件偏移输出。

生产中的 io_uring 用法：

• 网络服务：IORING_OP_ACCEPT、IORING_OP_RECV、IORING_OP_SEND 全部异步，一个线程可以管理上万连接；
• IORING_OP_READV/WRITEV：散列/聚集 I/O；
• IOSQE_IO_LINK：把多个操作串成链，前一个成功才执行下一个；
• IORING_REGISTER_FILES/BUFFERS：预注册 fd 与缓冲区，减少每次提交的开销。

io_uring 的限制：内核 5.1+（网络操作在后续版本才补齐），部分发行版默认禁用（担心安全面），容器里的 seccomp 可能拦截 io_uring_setup。用之前先检查 io_uring_setup 是否返回 ENOSYS/EPERM。

2.18.8 选型判据

三个常见误区：

1. 「epoll 一定比 poll 快」——连接少、活跃度高时，poll 的常数更小；epoll 的优势在「连接多、活跃少」。
2. 「ET 一定比 LT 快」——ET 减少通知次数，但要求应用自己维护状态机；写不好（漏读、漏 flush）反而更慢、更容易卡死。
3. 「io_uring 能替代 epoll」——它可以，但生态、调试工具、内核版本支持都还不统一；多数服务端仍以 epoll 为主。

2.18.9 多路复用 API 全表：事件位、flags 与 errno

这一节回答：epoll_event.events 的每一位是什么？EPOLLERR/EPOLLHUP 为什么总会出现？epoll_ctl 的 EEXIST/ENOENT/ENOSPC 分别对应什么现场？

select / poll / epoll 的系统调用

select 的 EINVAL 包括 nfds < 0 或 nfds > FD_SETSIZE；poll 的 EINVAL 是 nfds > RLIMIT_NOFILE；epoll_wait 的 EINVAL 是 maxevents <= 0。epoll_ctl 的 ENOSPC 不是「内存不足」，而是「注册的 fd 数超过 /proc/sys/fs/epoll/max_user_watches」，容器里经常遇到。

epoll_event.events 位表

EPOLLERR/EPOLLHUP 是输出位：即使 events 没请求，只要 fd 出错/挂断，epoll_wait 也会返回它们。只检查 EPOLLIN 的代码会在对端关闭后死循环：epoll_wait 一直返回 EPOLLHUP，read 一直返回 0。

struct epoll_event 在 x86-64 上是 __attribute__((packed)) 的 12 字节（events 4 字节 + data 8 字节）。不要自己定义结构体或按 16 字节传，内核按 ABI 校验大小。

struct pollfd 与 fd_set

poll 的 revents 每次调用都会被内核覆盖，不需要手动清零；但如果你在循环里跳过某些 fd，最好显式清零，避免读到上一轮的残留。

eventfd / signalfd / timerfd

eventfd 的 read 长度必须是 8，否则 EINVAL；signalfd 的 read 长度小于 sizeof(struct signalfd_siginfo) 也返回 EINVAL。这两个约束和 timerfd 一样，是内核强制的。

signalfd_siginfo 字段

signalfd 只对被阻塞的信号有效。如果信号没有 sigprocmask(SIG_BLOCK)，它会走默认处置（可能终止进程），signalfd 读不到。正确顺序：先阻塞信号，再创建 signalfd，再进事件循环。

io_uring 系统调用

io_uring_setup 在容器里可能返回 EPERM（seccomp 拦截）或 ENOSYS（内核未开 CONFIG_IO_URING）。生产代码必须把 io_uring 当作「可选加速」，失败时回退到 epoll + 线程池。

2.18.10 内核路径：eventpoll、ep_poll_callback 与惊群

这一节回答：epoll_ctl(ADD) 在内核里存了什么？数据到达时谁把 fd 挂到就绪链表？EPOLLEXCLUSIVE 如何避免惊群？

eventpoll 与 epitem

epoll_create1 创建一个 struct eventpoll，通过匿名 inode 暴露成 fd：

每个注册的 fd 对应一个 struct epitem：

epoll_ctl(ADD) 调 ep_insert：分配 epitem，插入红黑树，然后调用 ep_item_poll → vfs_poll → 目标文件的 ->poll()，把 ep_ptable_queue_proc 注册到目标 fd 的等待队列。之后数据到达，目标 fd 的等待队列回调 ep_poll_callback 被调用：

/* 按 Linux 6.8 fs/eventpoll.c 的结构简化 */
static int ep_poll_callback(wait_queue_entry_t *wait, unsigned mode,
                            int sync, void *key)
{
    struct epitem *epi = ep_item_from_wait(wait);
    struct eventpoll *ep = epi->ep;
    unsigned long flags;

    if (!(epi->event.events & ~EP_PRIVATE_BITS))
        return 1;                         /* 不关心这个事件 */

    spin_lock_irqsave(&ep->lock, flags);
    if (!(epi->event.events & EPOLLEXCLUSIVE))
        ;                                  /* 普通模式：所有等待者都唤醒 */
    if (!ep_is_linked(epi)) {
        list_add_tail(&epi->rdllink, &ep->rdllist);
        ep_pm_stay_awake(epi);
    }
    spin_unlock_irqrestore(&ep->lock, flags);

    wake_up(&ep->wq);                      /* 唤醒 epoll_wait */
    return 1;
}

这段伪代码省略了 EPOLLEXCLUSIVE 的关键分支：独占模式下内核只唤醒一个等待者（wake_up 的 exclusive 语义），而不是 wake_up_all。具体实现会检查 epi->event.events & EPOLLEXCLUSIVE，决定唤醒一个还是全部。效果是：

• 多个进程/线程在同一个监听 fd 上 epoll_wait，没有 EPOLLEXCLUSIVE：每个新连接唤醒所有等待者，只有一个能 accept 成功，其余白跑（惊群）；
• 加了 EPOLLEXCLUSIVE：内核只唤醒一个，失败再唤醒下一个。

EPOLLEXCLUSIVE 只能用于 EPOLL_CTL_ADD，且不能与 EPOLLONESHOT 同用；它解决的是「多个消费者等同一个 fd」的唤醒放大，不解决「一个连接被多个线程读」的问题（那需要 EPOLLONESHOT 或应用层分片）。

LT 与 ET 的内核差异

epoll_wait 走 ep_poll → ep_scan_ready_list → ep_send_events。对每个就绪的 epitem：

• LT：ep_send_events_proc 调用 ep_item_poll 重新检查；如果仍然就绪，把 epitem 放回 rdllist。所以下一次 epoll_wait 还会返回它。
• ET：只在 ep_poll_callback 被调用时挂 rdllist 一次；ep_send_events 取走后不再重查。如果应用没读空，内核不会再通知，直到下一次状态变化。

EPOLLONESHOT 在 ep_send_events 后把 epitem 从兴趣集合里「禁用」（清 EPOLLIN 等），需要 EPOLL_CTL_MOD 重新设置。它常用于多线程：一个连接同一时刻只允许一个线程处理，处理完再重新 arm，避免两个线程同时读同一条连接。

惊群与 SO_REUSEPORT

SO_REUSEPORT 的分流是按四元组哈希，不是轮询；同一客户端的重连可能落到同一个 worker。它适合无状态服务，不适合需要连接粘性的场景（如 WebSocket 会话）。

2.18.11 eventfd、signalfd、timerfd：把事件源统一成 fd

这一节回答：eventfd 的计数器语义是什么？signalfd 为什么必须先阻塞信号？三者如何与 epoll 组合成一个事件循环？

eventfd：8 字节的计数器

eventfd(initval, flags) 创建一个内核计数器，read 取走当前值，write 累加。默认模式下 read 返回累计值并清零；EFD_SEMAPHORE 模式下每次 read 只减 1。

eventfd 的典型用途：

• 线程唤醒：工作线程往 eventfd 写 1，事件循环 epoll_wait 立刻返回，处理任务队列；
• 信号处理器：write(eventfd, &one, 8) 是 async-signal-safe 的，可以在信号处理器里唤醒事件循环（比 self-pipe 少一个 fd）；
• 计数器：多个生产者 write，消费者一次 read 拿到总量，避免多次唤醒。

eventfd 的 read 必须正好 8 字节；缓冲区小于 8 返回 EINVAL。write 的值不能为 ULLONG_MAX（保留为错误），否则 EINVAL。

signalfd：把信号变成可读事件

signalfd 把「被阻塞的信号」变成 fd 上的可读事件。使用顺序固定：

sigset_t mask;
sigemptyset(&mask);
sigaddset(&mask, SIGINT);
sigaddset(&mask, SIGTERM);
sigprocmask(SIG_BLOCK, &mask, NULL);      /* 必须先阻塞 */

int sfd = signalfd(-1, &mask, SFD_CLOEXEC | SFD_NONBLOCK);
/* epoll_ctl(ADD, sfd, EPOLLIN) */
/* 事件循环里 read(sfd, &si, sizeof si) */

读出的 struct signalfd_siginfo 有 128 字节，包含 ssi_signo、ssi_pid、ssi_uid、ssi_code 等。read 一次返回一个信号；多个相同实时信号会排队，标准信号会合并。

signalfd 的常见错误：

• 忘了 sigprocmask：信号走默认处置，进程可能直接退出，signalfd 永远读不到；
• read 缓冲区太小：返回 EINVAL，不是短读；
• 多线程：信号可能被任意一个未阻塞该信号的线程处理；要统一到事件循环，所有线程都要阻塞，只有事件循环线程读 signalfd。

timerfd：到期变成可读

timerfd 在 2.11.6 已详述。它在事件循环里的用法是：timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK|TFD_CLOEXEC)，timerfd_settime 设置周期，epoll_ctl(ADD, EPOLLIN)，read 8 字节计数。它比 epoll_wait 的毫秒超时更精确，也能同时管理多个不同周期的定时器。

三者组合的事件循环

        ┌──────────────┐
        │  epoll_wait  │
        └──────┬───────┘
     ┌─────────┼──────────┬─────────────┐
     ▼         ▼          ▼             ▼
  listen fd  signalfd   timerfd      eventfd
  accept     SIGINT     tick         wakeup
     │         │          │             │
     └─────────┴──────────┴─────────────┘
                    统一处理

读图：所有事件源都是 fd，事件循环不需要信号处理器、不需要 self-pipe、不需要定时线程。signalfd 负责「退出/重载」，timerfd 负责「周期任务」，eventfd 负责「其他线程投递任务」，listen fd 负责「新连接」。

2.18.12 io_uring：SQE/CQE、注册与 liburing

这一节回答：SQE 和 CQE 各有哪些字段？io_uring_enter 的 min_complete 是什么意思？注册缓冲区/fd 省了什么？

三个系统调用与两个环

io_uring 的共享内存布局：

io_uring_setup(entries, &params) 返回 ring fd，并回填 params.sq_off/params.cq_off（各字段在 ring 内的字节偏移）。如果 params.features & IORING_FEAT_SINGLE_MMAP，SQ 与 CQ 可以映射同一块内存，只需一次 mmap。

io_uring_enter(fd, to_submit, min_complete, flags, sig, sz)：

• to_submit：本次提交多少个 SQE；
• min_complete：至少等多少个 CQE 完成再返回；0 表示不等；
• flags：IORING_ENTER_GETEVENTS 表示「提交后等完成」；
• sig/sz：信号掩码（可 NULL）。

一次调用可以「提交 N 个 + 等 M 个完成」，系统调用次数从「每个 I/O 一次」降到「一批一次」。

SQE 字段

IOSQE_* 常用位：

CQE 字段

CQE 按完成顺序返回，不保证与提交顺序一致；用 user_data 关联。res < 0 时是 -errno，不是 -1；要打印错误用 strerror(-res)。

注册：省掉每次提交的开销

注册缓冲区后，SQE 用 buf_index 指定索引，addr 可以是偏移；注册 fd 后，SQE 设 IOSQE_FIXED_FILE，fd 是索引。注意注册缓冲区必须保持有效且对齐，io_uring_register 会 pin 住这些页。

liburing 的封装

生产代码通常用 liburing：

struct io_uring ring;
io_uring_queue_init(256, &ring, 0);

struct io_uring_sqe *sqe = io_uring_get_sqe(&ring);
io_uring_prep_read(sqe, fd, buf, len, offset);
sqe->user_data = id;

io_uring_submit(&ring);

struct io_uring_cqe *cqe;
io_uring_wait_cqe(&ring, &cqe);
int res = cqe->res;
io_uring_cqe_seen(&ring, cqe);

io_uring_queue_exit(&ring);

io_uring_prep_* 负责填 SQE 字段，io_uring_submit 负责更新 tail 并调 io_uring_enter，io_uring_wait_cqe 负责等 CQE。自己写裸系统调用的价值在于理解 ring 的内存布局；生产代码不要重复造轮子。

限制与安全

• 内核 5.1+；网络 op（ACCEPT/RECV/SEND）在 5.5+ 才逐步补齐；
• IORING_SETUP_SQPOLL 需要 CAP_SYS_ADMIN 或 IORING_SETUP_SQ_AFF + 特权；
• 容器 seccomp 可能拦截 io_uring_setup，返回 EPERM；
• 部分发行版通过 /proc/sys/kernel/io_uring_disabled 禁用（0=允许，1=仅特权，2=禁止）；
• io_uring 的漏洞历史较多，安全敏感环境要评估内核版本。

2.18.13 程序一：eventfd + signalfd + timerfd 的统一事件循环

这个程序把三类事件源挂到同一个 epoll：signalfd 处理 SIGINT/SIGTERM，timerfd 每 100 ms 打一次 tick，eventfd 由后台线程在 250 ms 后唤醒。

/* event_loop.c — signalfd + timerfd + eventfd + epoll */
#define _GNU_SOURCE
#include <errno.h>
#include <pthread.h>
#include <signal.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/epoll.h>
#include <sys/eventfd.h>
#include <sys/signalfd.h>
#include <sys/timerfd.h>
#include <time.h>
#include <unistd.h>

#define MAX_EVENTS 8

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

struct wake_arg {
    int efd;
    useconds_t delay_us;
};

static void *waker(void *p)
{
    struct wake_arg *a = p;
    usleep(a->delay_us);
    uint64_t one = 1;
    ssize_t n = write(a->efd, &one, sizeof one);
    if (n != (ssize_t)sizeof one)
        perror("write eventfd");
    return NULL;
}

static void add_fd(int epfd, int fd, uint32_t events)
{
    struct epoll_event ev = { .events = events, .data.fd = fd };
    if (epoll_ctl(epfd, EPOLL_CTL_ADD, fd, &ev) == -1)
        die("epoll_ctl");
}

int main(void)
{
    sigset_t mask;
    sigemptyset(&mask);
    sigaddset(&mask, SIGINT);
    sigaddset(&mask, SIGTERM);
    if (sigprocmask(SIG_BLOCK, &mask, NULL) == -1)
        die("sigprocmask");

    int sfd = signalfd(-1, &mask, SFD_CLOEXEC | SFD_NONBLOCK);
    if (sfd == -1) die("signalfd");

    int tfd = timerfd_create(CLOCK_MONOTONIC, TFD_CLOEXEC | TFD_NONBLOCK);
    if (tfd == -1) die("timerfd_create");
    struct itimerspec its = {0};
    its.it_value.tv_nsec    = 100 * 1000000L;
    its.it_interval.tv_nsec = 100 * 1000000L;
    if (timerfd_settime(tfd, 0, &its, NULL) == -1)
        die("timerfd_settime");

    int efd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
    if (efd == -1) die("eventfd");

    int epfd = epoll_create1(EPOLL_CLOEXEC);
    if (epfd == -1) die("epoll_create1");
    add_fd(epfd, sfd, EPOLLIN);
    add_fd(epfd, tfd, EPOLLIN);
    add_fd(epfd, efd, EPOLLIN);

    struct wake_arg arg = { .efd = efd, .delay_us = 250000 };
    pthread_t th;
    int rc = pthread_create(&th, NULL, waker, &arg);
    if (rc != 0) {
        fprintf(stderr, "pthread_create: %s\n", strerror(rc));
        return EXIT_FAILURE;
    }

    int ticks = 0;
    int running = 1;
    while (running) {
        struct epoll_event evs[MAX_EVENTS];
        int n = epoll_wait(epfd, evs, MAX_EVENTS, -1);
        if (n == -1) {
            if (errno == EINTR) continue;
            perror("epoll_wait");
            break;
        }

        for (int i = 0; i < n; i++) {
            int fd = evs[i].data.fd;

            if (fd == sfd) {
                struct signalfd_siginfo si;
                ssize_t r = read(sfd, &si, sizeof si);
                if (r == (ssize_t)sizeof si) {
                    printf("signal %u from pid %u -> exit\n",
                           si.ssi_signo, si.ssi_pid);
                    running = 0;
                }
            } else if (fd == tfd) {
                uint64_t exp;
                ssize_t r = read(tfd, &exp, sizeof exp);
                if (r == (ssize_t)sizeof exp) {
                    ticks += (int)exp;
                    printf("tick exp=%llu total=%d\n",
                           (unsigned long long)exp, ticks);
                }
            } else if (fd == efd) {
                uint64_t val;
                ssize_t r = read(efd, &val, sizeof val);
                if (r == (ssize_t)sizeof val)
                    printf("eventfd wakeup value=%llu\n",
                           (unsigned long long)val);
            }
        }
    }

    pthread_join(th, NULL);
    close(epfd);
    close(efd);
    close(tfd);
    close(sfd);
    return EXIT_SUCCESS;
}

编译运行：

cc -std=c17 -D_GNU_SOURCE -Wall -Wextra -Werror -O2 -g -pthread \
   event_loop.c -o event_loop
./event_loop
# tick exp=1 total=1
# tick exp=1 total=2
# eventfd wakeup value=1
# ...
# 按 Ctrl-C：
# signal 2 from pid ... -> exit

关键点：

• signalfd 之前必须 sigprocmask(SIG_BLOCK)；
• timerfd 的 read 返回累计到期次数，可能大于 1；
• eventfd 的 read 返回累计写值并清零；
• 所有 fd 都是非阻塞的，epoll_wait 是唯一阻塞点；
• 后台线程只负责 write(eventfd)，不直接操作事件循环的状态。

strace -f -e trace=epoll_wait,read 会看到一次 epoll_wait 返回多个 fd，read 分别消费。把 signalfd 换成 signal() 处理器，程序会失去「统一事件源」的优势：信号处理器里不能安全地 printf，也不能调用 epoll_ctl。

2.18.14 程序二：EPOLLEXCLUSIVE 与惊群

这个程序让两个子进程各自在同一个监听 fd 上 epoll_wait，父进程发起 20 个连接。shared 模式下两个子进程都被唤醒；exclusive 模式下内核只唤醒一个。子进程统计 epoll_wait 返回监听事件的次数，用来对比。

/* epoll_exclusive.c — 对比 EPOLLEXCLUSIVE 的唤醒次数 */
#define _GNU_SOURCE
#include <arpa/inet.h>
#include <errno.h>
#include <netinet/in.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/epoll.h>
#include <sys/socket.h>
#include <sys/wait.h>
#include <unistd.h>

#ifndef EPOLLEXCLUSIVE
#define EPOLLEXCLUSIVE (1u << 28)
#endif

static volatile sig_atomic_t stop;

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

static void child_loop(int lfd, int exclusive, int idx)
{
    struct sigaction sa = {0};
    sa.sa_handler = on_alarm;
    sigemptyset(&sa.sa_mask);
    sa.sa_flags = 0;
    sigaction(SIGALRM, &sa, NULL);
    alarm(2);                             /* 2 秒后退出 */

    int epfd = epoll_create1(EPOLL_CLOEXEC);
    if (epfd == -1) _exit(1);

    struct epoll_event ev = {
        .events = EPOLLIN | (exclusive ? EPOLLEXCLUSIVE : 0),
        .data.fd = lfd,
    };
    if (epoll_ctl(epfd, EPOLL_CTL_ADD, lfd, &ev) == -1) _exit(1);

    int wakeups = 0, accepted = 0;
    while (!stop) {
        struct epoll_event out;
        int n = epoll_wait(epfd, &out, 1, -1);
        if (n == -1) {
            if (errno == EINTR) continue;
            break;
        }
        if (n == 1 && (out.events & EPOLLIN)) {
            wakeups++;
            for (;;) {
                int cfd = accept4(lfd, NULL, NULL, SOCK_NONBLOCK | SOCK_CLOEXEC);
                if (cfd == -1) {
                    if (errno == EAGAIN || errno == EWOULDBLOCK) break;
                    if (errno == EINTR) continue;
                    break;
                }
                accepted++;
                close(cfd);
            }
        }
    }
    printf("child %d exclusive=%d wakeups=%d accepted=%d\n",
           idx, exclusive, wakeups, accepted);
    fflush(stdout);
    close(epfd);
    _exit(0);
}

int main(int argc, char **argv)
{
    int exclusive = (argc > 1 && strcmp(argv[1], "exclusive") == 0);

    int lfd = socket(AF_INET, SOCK_STREAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0);
    if (lfd == -1) { perror("socket"); return EXIT_FAILURE; }
    int one = 1;
    setsockopt(lfd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof one);

    struct sockaddr_in addr = {
        .sin_family = AF_INET,
        .sin_addr.s_addr = htonl(INADDR_LOOPBACK),
        .sin_port = 0,
    };
    if (bind(lfd, (struct sockaddr *)&addr, sizeof addr) == -1) {
        perror("bind");
        return EXIT_FAILURE;
    }
    if (listen(lfd, 64) == -1) { perror("listen"); return EXIT_FAILURE; }
    socklen_t alen = sizeof addr;
    if (getsockname(lfd, (struct sockaddr *)&addr, &alen) == -1) {
        perror("getsockname");
        return EXIT_FAILURE;
    }

    for (int i = 0; i < 2; i++) {
        pid_t pid = fork();
        if (pid == -1) { perror("fork"); return EXIT_FAILURE; }
        if (pid == 0) {
            child_loop(lfd, exclusive, i);
        }
    }

    /* 20 个客户端连接；每个连接只发一个字节就关 */
    for (int i = 0; i < 20; i++) {
        int fd = socket(AF_INET, SOCK_STREAM | SOCK_CLOEXEC, 0);
        if (fd == -1) { perror("socket client"); break; }
        if (connect(fd, (struct sockaddr *)&addr, sizeof addr) == -1) {
            perror("connect");
            close(fd);
            continue;
        }
        if (write(fd, "x", 1) != 1) perror("write");
        close(fd);
        usleep(20000);                     /* 20ms，让子进程有时间 accept */
    }

    for (int i = 0; i < 2; i++) {
        int status;
        if (wait(&status) == -1) perror("wait");
    }
    close(lfd);
    return EXIT_SUCCESS;
}

编译运行：

cc -std=c17 -D_GNU_SOURCE -Wall -Wextra -Werror -O2 -g \
   epoll_exclusive.c -o epoll_exclusive
./epoll_exclusive shared      # 两个子进程都被唤醒
./epoll_exclusive exclusive   # 内核只唤醒一个

预期（数值会抖动）：

# shared
child 0 exclusive=0 wakeups=20 accepted=10
child 1 exclusive=0 wakeups=20 accepted=10

# exclusive
child 0 exclusive=1 wakeups=10 accepted=10
child 1 exclusive=1 wakeups=10 accepted=10

shared 模式下每个连接唤醒两个子进程，总 wakeups 接近 40；exclusive 模式下总 wakeups 接近 20。两个子进程的 accepted 之和应等于 20（可能有连接在 2 秒后才被 accept，取决于调度）。

注意：EPOLLEXCLUSIVE 只减少唤醒次数，不保证负载均衡；某个子进程可能抢到更多连接。要按连接分流，用 SO_REUSEPORT 让每个 worker 各自 bind。

2.18.15 程序三：裸 io_uring 读取文件

这个程序不依赖 liburing，直接用 io_uring_setup/io_uring_enter 与 mmap 完成一次 IORING_OP_READ，用来验证 SQ/CQ 的内存布局与 user_data 关联。

/* uring_read.c — 裸系统调用版 io_uring 读文件 */
#define _GNU_SOURCE
#include <errno.h>
#include <fcntl.h>
#include <linux/io_uring.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/mman.h>
#include <sys/syscall.h>
#include <unistd.h>

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

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

    int fd = open(argv[1], O_RDONLY | O_CLOEXEC);
    if (fd == -1) die("open");

    struct io_uring_params p;
    memset(&p, 0, sizeof p);
    int ring_fd = (int)syscall(__NR_io_uring_setup, 8, &p);
    if (ring_fd == -1) die("io_uring_setup");

    size_t sq_ring_sz = p.sq_off.array + p.sq_entries * sizeof(unsigned);
    size_t cq_ring_sz = p.cq_off.cqes + p.cq_entries * sizeof(struct io_uring_cqe);
    size_t ring_sz = sq_ring_sz > cq_ring_sz ? sq_ring_sz : cq_ring_sz;

    char *sq_ptr, *cq_ptr;
    if (p.features & IORING_FEAT_SINGLE_MMAP) {
        sq_ptr = mmap(NULL, ring_sz, PROT_READ | PROT_WRITE,
                      MAP_SHARED | MAP_POPULATE, ring_fd, IORING_OFF_SQ_RING);
        if (sq_ptr == MAP_FAILED) die("mmap sq");
        cq_ptr = sq_ptr;
    } else {
        sq_ptr = mmap(NULL, sq_ring_sz, PROT_READ | PROT_WRITE,
                      MAP_SHARED | MAP_POPULATE, ring_fd, IORING_OFF_SQ_RING);
        if (sq_ptr == MAP_FAILED) die("mmap sq");
        cq_ptr = mmap(NULL, cq_ring_sz, PROT_READ | PROT_WRITE,
                      MAP_SHARED | MAP_POPULATE, ring_fd, IORING_OFF_CQ_RING);
        if (cq_ptr == MAP_FAILED) die("mmap cq");
    }

    struct io_uring_sqe *sqes = mmap(NULL,
                                     p.sq_entries * sizeof(struct io_uring_sqe),
                                     PROT_READ | PROT_WRITE,
                                     MAP_SHARED | MAP_POPULATE, ring_fd,
                                     IORING_OFF_SQES);
    if (sqes == MAP_FAILED) die("mmap sqes");

    unsigned *sq_head = (unsigned *)(sq_ptr + p.sq_off.head);
    unsigned *sq_tail = (unsigned *)(sq_ptr + p.sq_off.tail);
    unsigned *sq_mask = (unsigned *)(sq_ptr + p.sq_off.ring_mask);
    unsigned *sq_array = (unsigned *)(sq_ptr + p.sq_off.array);

    if (*sq_tail - *sq_head >= p.sq_entries) {
        fprintf(stderr, "SQ ring is full\n");
        return EXIT_FAILURE;
    }

    unsigned *cq_head = (unsigned *)(cq_ptr + p.cq_off.head);
    unsigned *cq_tail = (unsigned *)(cq_ptr + p.cq_off.tail);
    unsigned *cq_mask = (unsigned *)(cq_ptr + p.cq_off.ring_mask);
    struct io_uring_cqe *cqes =
        (struct io_uring_cqe *)(cq_ptr + p.cq_off.cqes);

    char buf[4096];
    struct io_uring_sqe *sqe = &sqes[0];
    memset(sqe, 0, sizeof *sqe);
    sqe->opcode = IORING_OP_READ;
    sqe->fd = fd;
    sqe->addr = (uint64_t)(uintptr_t)buf;
    sqe->len = sizeof buf;
    sqe->off = 0;                         /* 从文件偏移 0 读 */
    sqe->user_data = 0x1234;

    unsigned tail = *sq_tail;
    sq_array[tail & *sq_mask] = 0;         /* SQE 下标 */
    __atomic_store_n(sq_tail, tail + 1, __ATOMIC_RELEASE);

    int rc = (int)syscall(__NR_io_uring_enter, ring_fd, 1, 1,
                          IORING_ENTER_GETEVENTS, NULL, 0);
    if (rc < 0) die("io_uring_enter");

    for (;;) {
        if (__atomic_load_n(cq_head, __ATOMIC_ACQUIRE) !=
            __atomic_load_n(cq_tail, __ATOMIC_ACQUIRE))
            break;
        /* 理论上 enter 已经等到，这里只是保险 */
    }

    unsigned head = *cq_head;
    struct io_uring_cqe *cqe = &cqes[head & *cq_mask];
    printf("cqe user_data=0x%llx res=%d\n",
           (unsigned long long)cqe->user_data, cqe->res);

    if (cqe->res < 0) {
        fprintf(stderr, "read failed: %s\n", strerror(-cqe->res));
    } else if (cqe->res > 0) {
        ssize_t w = write(STDOUT_FILENO, buf, (size_t)cqe->res);
        (void)w;
    }

    __atomic_store_n(cq_head, head + 1, __ATOMIC_RELEASE);

    munmap(sqes, p.sq_entries * sizeof(struct io_uring_sqe));
    if (p.features & IORING_FEAT_SINGLE_MMAP)
        munmap(sq_ptr, ring_sz);
    else {
        munmap(sq_ptr, sq_ring_sz);
        munmap(cq_ptr, cq_ring_sz);
    }
    close(ring_fd);
    close(fd);
    return 0;
}

编译运行：

cc -std=c17 -D_GNU_SOURCE -Wall -Wextra -Werror -O2 -g \
   uring_read.c -o uring_read
./uring_read /etc/hosts | head
# cqe user_data=0x1234 res=...

关键点：

• sq_tail 用 release 存储，确保 SQE 内容先可见；内核读 tail 后才会读 SQE。
• cq_head 用 acquire 读取，确保 CQE 内容可见后再访问。
• sq_array[tail & mask] 是「SQE 下标」，不是 SQE 地址；这是最容易写错的一步。
• cqe->res 是 ssize_t 语义：成功为字节数，失败为 -errno。
• IORING_FEAT_SINGLE_MMAP 时 SQ/CQ 共用一块映射，不能重复 munmap。

这个程序只做一次读，没有注册缓冲区/fd，也没有批量提交。生产代码应该用 liburing，并考虑 IORING_REGISTER_BUFFERS 与 IOSQE_FIXED_FILE 降低每次提交的开销。

2.18.16 实验：epoll 的 fdinfo、惊群与 io_uring 系统调用

实验一：epoll 的注册与唤醒

# 1. 看 epoll 系统调用
strace -f -e trace=epoll_create1,epoll_ctl,epoll_wait ./event_loop 2>&1 | head -20
# 预期：epoll_ctl ADD 三次；epoll_wait 返回 1..3

# 2. 看 epoll 实例的 fdinfo
./event_loop & pid=$!
ls -l /proc/$pid/fd | grep eventpoll
cat /proc/$pid/fdinfo/3 2>/dev/null
# 预期：tfd: 后面列出注册的 fd 与 events

# 3. 看 watch 上限
cat /proc/sys/fs/epoll/max_user_watches

# 4. 对比 LT/ET 的通知次数（用 2.18.5 的 epoll_echo）
./epoll_echo lt & pid=$!
# 另一个终端发 1 MB 数据，观察 strace 的 epoll_wait 次数
strace -p $pid -e trace=epoll_wait -c

怎么读：fdinfo 里的 tfd 行是「注册的 fd + 事件掩码」，可以直接看到 EPOLLIN/EPOLLOUT/EPOLLET 的数值。max_user_watches 是每用户上限，不是每进程；容器里多个服务共享同一个 UID 时会互相挤占。

实验二：惊群与 io_uring

# 1. 对比 shared/exclusive
./epoll_exclusive shared
./epoll_exclusive exclusive
# 预期：shared 的 wakeups 总数约为 exclusive 的两倍

# 2. 看 io_uring 的系统调用
strace -e trace=io_uring_setup,io_uring_enter,io_uring_register \
       ./uring_read /etc/hosts 2>&1 | head
# 预期：io_uring_setup(8, ...) = 3；io_uring_enter(3, 1, 1, ...) = 1

# 3. 看内核是否允许 io_uring
cat /proc/sys/kernel/io_uring_disabled 2>/dev/null || echo "sysctl not present"
grep -i io_uring /boot/config-$(uname -r) 2>/dev/null || true

# 4. 用 liburing 的示例做对照（如果已安装）
#    sudo apt install liburing-dev
#    cc uring_cat.c -o uring_cat -luring && ./uring_cat /etc/hosts

怎么读：io_uring_setup 返回的 fd 是 ring fd，不是普通文件；io_uring_enter 的第二个参数是提交数，第三个是等待完成数。/proc/sys/kernel/io_uring_disabled 不存在不代表启用（老内核没有这个 sysctl），要看 io_uring_setup 的返回值。

2.18.17 陷阱清单：多路复用的十六个现场

2.18.18 本章出口检查

母本对应

下一站：2.19 终端、pty、作业控制与守护进程 回到「进程与用户」的交界：Ctrl-C 如何变成信号、伪终端如何让 ssh 里跑交互程序、守护进程为什么要 setsid 与二次 fork。