Linux异步IO的历史

异步IO一直是 Linux 系统的痛。Linux 很早就有 POSIX AIO 这套异步IO实现，但它是在用户空间自己开用户线程模拟的，效率极其低下。后来在 Linux 2.6 引入了真正的内核级别支持的异步IO实现（Linux aio），但是它只支持 Direct IO，只支持磁盘文件读写，而且对文件大小还有限制，总之各种麻烦。到目前为止（2019年5月），libuv 还是在用pthread+preadv的形式实现异步IO。

随着 Linux 5.1 的发布，Linux 终于有了自己好用的异步IO实现，并且支持大多数文件类型（磁盘文件、socket，管道等），这个就是本文的主角：io_uring

IOCP

于IO多路复用模型 epoll 不同，io_uring 的思想更类似于 Windows 上的 IOCP。用快递来举例：同步模型就是你从在电商平台下单前，就在你家楼下一直等，直到快递公司把货送到楼下，你再把东西带上楼。epoll 类似于你下单，快递公司送到楼下，通知你可以去楼下取货了，这时你下楼把东西带上来。虽然还是需要用户下楼取货（有一段同步读写的时间），但是由于不需要等快递在路上的时间，效率已经有非常大的提升。但是，epoll不适用于磁盘IO，因为磁盘文件总是可读的。

而 IOCP 就是一步到位，直接送货上门，连下楼取的动作都不需要。整个过程完全是非阻塞的。

io_uring 的简单使用

io_uring 是一套系统调用接口，虽然总共就3个系统调用，但实际使用却非常复杂。这里直接介绍封装过便于用户使用的 liburing。

在尝试前请首先确认自己的 Linux 内核版本在 5.1 以上（uname -r）。liburing 需要自己编译（之后可能会被各大Linux发行版以软件包的形式收录），git clone 后直接 ./configure && sudo make install 就好了。

io_uring 结构初始化

liburing 提供了自己的核心结构 io_uring，它内部封装了 io_uring 自己的文件描述符（fd）以及其他与内核通信所需变量。

struct io_uring {
    struct io_uring_sq sq;
    struct io_uring_cq cq;
    int ring_fd;
};

使用之前需要先初始化，使用 io_uring_queue_init 初始化此结构。

extern int io_uring_queue_init(unsigned entries, struct io_uring *ring,
    unsigned flags);

如函数名称所示， io_uring 是一个循环队列（ring_buffer）。第一个参数 entries 表示队列大小（实际空间可能比用户指定的大）；第二个参数 ring 就是需要初始化的 io_uring 结构指针；第三个参数 flags 是标志参数，无特殊需要传 0 即可。例如

#include <liburing.h>
struct io_uring ring;
io_uring_queue_init(32, &ring, 0);

提交读、写请求

首先使用 io_uring_get_sqe 获取 sqe 结构。

extern struct io_uring_sqe *io_uring_get_sqe(struct io_uring *ring);

一个 sqe（submission queue entry）代表一次 IO 请求，占用循环队列一个空位。io_uring 队列满时 io_uring_get_sqe 会返回 NULL，注意错误处理。注意这里的队列是指未提交的请求，已提交的（但未完成）请求不占位置。

struct io_uring_sqe *sqe = io_uring_get_sqe(&ring);

然后使用 io_uring_prep_readv 或 io_uring_prep_writev 初始化 sqe 结构。

static inline void io_uring_prep_readv(struct io_uring_sqe *sqe, int fd,
                       const struct iovec *iovecs,
                       unsigned nr_vecs, off_t offset);
static inline void io_uring_prep_writev(struct io_uring_sqe *sqe, int fd,
                    const struct iovec *iovecs,
                    unsigned nr_vecs, off_t offset);

第一个参数 sqe 即前面获取的 sqe 结构指针；fd 为需要读写的文件描述符，可以是磁盘文件也可以是socket；iovecs 为 iovec 数组，具体使用请参照 readv 和 writev，nr_vecs 为 iovecs 数组元素个数，offset 为文件操作的偏移量。

可以看到这两个函数完全按照 preadv 和 pwritev 设计，语义也相同，所以很好上手。需要注意的是，如果需要顺序读写文件，偏移量 offset 需要程序自己维护。

struct iovec iov = {
    .iov_base = "Hello world",
    .iov_len = strlen("Hello world"),
};
io_uring_prep_writev(sqe, fd, &iov, 1, 0);

初始化 sqe 后，可以用 io_uring_sqe_set_data，传入你自己的数据，一般是一个 malloc 得到的指针，C++ 里面可以直接传 this。

static inline void io_uring_sqe_set_data(struct io_uring_sqe *sqe, void *data);

注意 prep_* 中会 memset(0)，所以一定要先 prep_* 再 set_data。笔者这里纠结了两个小时。

准备好 sqe 后即可使用 io_uring_submit 提交请求。

extern int io_uring_submit(struct io_uring *ring);

你可以初始化多个 sqe 然后一次性 submit。

io_uring_submit(&ring);

完成 IO 请求

io_uring_submit 都是异步操作，不会阻塞当前线程。那么如何得知提交的操作何时完成呢？liburing 提供了函数 io_uring_peek_cqe 和 io_uring_wait_cqe 两个函数获取当前已完成的 IO 操作。

extern int io_uring_peek_cqe(struct io_uring *ring,
    struct io_uring_cqe **cqe_ptr);
extern int io_uring_wait_cqe(struct io_uring *ring,
    struct io_uring_cqe **cqe_ptr);

第一个参数是 io_uring 结构指针；第二个参数 cqe_ptr 是输出参数，是 cqe 指针变量的地址。

cqe（completion queue entry）标记一个已完成的 IO 操作，同时也记录的之前传入的用户数据。每个 cqe 都与前面的 sqe 对应。

这两个函数，io_uring_peek_cqe 如果没有已完成的 IO 操作时，也会立即返回，cqe_ptr 被置空；而
io_uring_wait_cqe 会阻塞线程，等待 IO 操作完成。

for (;;) {
    io_uring_peek_cqe(&ring, &cqe);
    if (!cqe) {
        puts("Waiting...");
        // accept 新连接，做其他事
    } else {
        puts("Finished.");
        break;
    }
}

上文简单起见用忙等待做示例，在实际应用场景中应该是一个事件循环，浏览器、nodejs 给我们内部隐藏了事件循环的实现，而写 C/C++ 语言只能我们自己做。

可通过 io_uring_cqe_get_data 获取前面给 sqe 设置的用户数据。

static inline void *io_uring_cqe_get_data(struct io_uring_cqe *cqe);

默认情况下 IO 完成事件不会从队列中清除，导致 io_uring_peek_cqe 会获取到相同事件，使用 io_uring_cqe_seen 标记该事件已被处理

static inline void io_uring_cqe_seen(struct io_uring *ring,
                     struct io_uring_cqe *cqe);

io_uring_cqe_seen(&ring, cqe);

清除 io_uring，释放资源

清除 io_uring 结构使用 io_uring_queue_exit

extern void io_uring_queue_exit(struct io_uring *ring);

io_uring_queue_exit(&ring);

完

完整代码列举如下：这段代码作用就是创建文件 /home/carter/test.txt 并写入字符串 Hello world

#include <liburing.h>
#include <unistd.h>
#include <fcntl.h>
#include <stdio.h>

int main()
{
    struct io_uring ring;
    io_uring_queue_init(32, &ring, 0);

    struct io_uring_sqe *sqe = io_uring_get_sqe(&ring);
    int fd = open("/home/carter/test.txt", O_WRONLY | O_CREAT);
    struct iovec iov = {
        .iov_base = "Hello world",
        .iov_len = strlen("Hello world"),
    };
    io_uring_prep_writev(sqe, fd, &iov, 1, 0);
    io_uring_submit(&ring);

    struct io_uring_cqe *cqe;

    for (;;) {
        io_uring_peek_cqe(&ring, &cqe);
        if (!cqe) {
            puts("Waiting...");
            // accept 新连接，做其他事
        } else {
            puts("Finished.");
            break;
        }
    }
    io_uring_cqe_seen(&ring, cqe);
    io_uring_queue_exit(&ring);
}

可以看到，C语言的异步操作还是比同步操作复杂不少，libuv（nodejs 的底层 IO 库）已经 表示会引入 io_uring。如果要自己用，一定要使用一个协程库简化异步操作。

这里 是我使用自己编写的协程库 Cxx-yield 实现的一个简单的文件服务器 demo。可以看到，经过简单封装后，异步文件读写可以简化到一行：https://github.com/CarterLi/C…。就是那种在 JavaScript 里写 async、await 的快感