前言：io_uring是大神Jens Axboe开发的异步IO框架，在Linux内核5.1引入。本文介绍什么是异步框架和io_uring的一些基础内容，最后介绍Node.js(Libuv)中，之前有人提但至今还没有合并的一个关于io_uring的pr。
在io_uring之前，Linux没有成熟的异步IO能力，什么是异步IO呢?回想我们读取资源的过程，我们可以以阻塞或非阻塞的模式调用read、readv，也可以通过epoll监听文件描述符和事件的方式，在回调里调用read系列函数进行读取，这些API有个共同的地方是，不管是主动探还是被动探测资源是否可读，当可读的时候，都需要进程自己去执行读操作。而io_uring强大的地方是，进程不需要再自己主动执行读操作，而是内核读完后通知进程，相比epoll，io_uring又进了一步，类似的能力是windows的IOCP。
io_uring和epoll一样，API不多，但是io_uring比epoll复杂得多。我们首先需要调用io_uring_setup初始化io_uring，拿到一个fd。
复制
int ring_fd; unsigned *sring_tail, *sring_mask, *sring_array, *cring_head, *cring_tail, *cring_mask; struct io_uring_sqe *sqes; struct io_uring_cqe *cqes; char buff[BLOCK_SZ]; off_t offset; struct io_uring_params p; void *sq_ptr, *cq_ptr; memset(&p, 0, sizeof(p)); // 拿到io_uring对应的fd int ring_fd = io_uring_setup(QUEUE_DEPTH, &p); int sring_sz = p.sq_off.array + p.sq_entries * sizeof(unsigned); int cring_sz = p.cq_off.cqes + p.cq_entries * sizeof(struct io_uring_cqe); // 映射ring_fd到mmap返回的地址，我们可以以操作返回地址的方式操作ring_fd，达到用户和内核共享数据的目的 cq_ptr = sq_ptr = mmap(0, sring_sz, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_POPULATE, ring_fd, IORING_OFF_SQ_RING); sqes = mmap(0, p.sq_entries * sizeof(struct io_uring_sqe), PROT_READ | PROT_WRITE, MAP_SHARED | MAP_POPULATE, ring_fd, IORING_OFF_SQES); // 保存任务队列和完成队列的地址，后续提交任务和获取完成任务节点时需要用 sring_tail = sq_ptr + p.sq_off.tail; sring_mask = sq_ptr + p.sq_off.ring_mask; sring_array = sq_ptr + p.sq_off.array; cring_head = cq_ptr + p.cq_off.head; cring_tail = cq_ptr + p.cq_off.tail; cring_mask = cq_ptr + p.cq_off.ring_mask; cqes = cq_ptr + p.cq_off.cqes;
1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.
34.
35.
36.
37.
38.
39.
40.
41.
42.
43.
44.
45.
46.
47.
io_uring不仅实现非常复杂，就连使用也非常复杂，但是目前只需要大致了解原理就好了。上面的代码主要目的有以下几个。
1 出生后io_uring并拿到一个io_uring实例对应的fd。
2 通过mmap映射io_uring对应的fd到一个内存地址，后续我们就可以通过操作内存地址的方式和内核通信。
3 保存任务队列和完成队列的地址信息，后续需要用到。
我们看到io_uring底层维护了任务队列(sq)和完成队列两个队列(cq)。对应的节点叫sqe和cqe。当我们需要操作一个资源的时候，就可以获取一个seq，并且填充字段，然后提交给内核，我们看一下sqe的核心字段。
复制
![[[409009]]](https://s6.51cto.com/oss/202107/03/da3703cfdb61ff7d2f94db60d4c23fd0.png){kind=link}
struct io_uring_sqe { __u8 opcode; /* 操作类型，比如读、写 */ __s32 fd; /* 资源对应的fd */ __u64 off; /* 资源的偏移（操作的起点） */ __u64 addr; /* 保存数据的内存首地址 */ __u32 len; /* 数据长度 */ __u64 user_data; /* 用户定义的字段，通常用于关联请求和响应 */ __u8 flags; /* 标记 */ ... };
1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
io_uring_sqe的核心字段都比较好理解，构造了一个请求后，就插入到内核的请求任务队列。接着调用io_uring_enter通知内核，有需要处理的任务，我们可以在调用io_uring_enter的时候设置等待多少个请求完成后再返回。另外内核处理poll的模式，这时候内核会开启内核线程去检测任务是否完成，不需要进程调用io_uring_enter。下面是我们发送一个读取请求的逻辑。
复制
unsigned index, tail; tail = *sring_tail; // 拿到请求队列的一个空闲位置，是一个环，需要做回环处理 index = tail & *sring_mask; struct io_uring_sqe *sqe = &sqes[index]; // 初始化请求结构体 sqe->opcode = op; // 读取的fd sqe->fd = fd; // 读取的数据保存到buff // 可以通过关联buff，等到响应的时候能找到对应的请求上下文 sqe->addr = (unsigned long) buff; sqe->user_data = (unsigned long long) buff; memset(buff, 0, sizeof(buff)); sqe->len = BLOCK_SZ; sqe->off = offset; // 插入请求队列 sring_array[index] = index; // 更新索引 tail++; // 通知内核有任务需要处理，并等待有一个任务完成后再返回 io_uring_smp_store_release(sring_tail, tail); int ret = io_uring_enter(ring_fd, 1,1, IORING_ENTER_GETEVENTS);
1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.
34.
35.
36.
37.
38.
39.
40.
41.
42.
当任务完成的时候，io_uring_enter就会返回。但是这里有个问题，请求任务和响应不是对应的，内核不保证任务完成的顺序，内核只是告诉我们哪些任务完成了，我们可以通过user_data关联请求和响应，类似rpc通信里的seq一样。user_data字段在请求里设置，响应里会返回，从而请求方知道这个响应对应的是哪个请求。响应对应的结构体比较简单。
复制
struct io_uring_cqe { /* 用户定义字段，通常用于关联请求和响应 */ __u64 user_data; /* 系统调用的返回码，比如read */ __s32 res; // 暂时没用到 __u32 flags; };
1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
我们这里假设请求和响应是串行的，所以不需要用到user_data字段关联请求和响应。从前面代码我们可以看到，我们把数据读取到buff变量里。我们看看内核返回后我们的处理逻辑。
复制
struct io_uring_cqe *cqe; unsigned head, reaped = 0; // 拿到完成队列队头节点，可消费buff里面存储的数据 head = io_uring_smp_load_acquire(cring_head); cqe = &cqes[head & (*cring_mask)]; // 更新头索引 head++; io_uring_smp_store_release(cring_head, head);
1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
这就是io_uring一个读取操作的大致过程，我们看到用户层面的逻辑还是挺复杂的，作者也想到了，所以又封装了Liburing库简化使用。
那么我们到底怎么使用它呢，我们回想epoll的使用。
复制
// 创建epoll 实例 int epollfd = epoll_create(); // 封装fd和订阅事件 struct epoll_event event; event.events = EPOLLIN; event.data.fd = listenFd; // 注册到epoll epoll_ctl(epollfd, EPOLL_CTL_ADD, listenFd, &event); // 等待事件触发 int num = epoll_wait(epollfd, events, MAX_EVENTS, -1); for (i = 0; i < num; ++i) { // 处理事件，比如读写 }
1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
接着我们看看基于Liburing的o_uring的使用。
复制
// 拿到一个请求结构体 struct io_uring_sqe *sqe = io_uring_get_sqe(ring); // 设置fd和数据地址 io_uring_prep_recv(sqe, fd, data, len, 0); // 通知内核有任务处理 io_uring_submit(&ring); // 等待事件完成 io_uring_submit_and_wait(&ring, 1); // 获取完成的任务 int nums = io_uring_peek_batch_cqe(&ring, cqes, sizeof(cqes) / sizeof(cqes[0])); for (i = 0; i < nums; ++i) { // 处理完成的任务 struct io_uring_cqe *cqe = cqes[i]; }
1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
我们看到基于Liburing的使用简单了很多，有点epoll的风格了。io_uring就介绍到这里，io_uring的细节比较多，实现也比较复杂，代码量也达到了近1万行(epoll是2500左右)，关于io_uring网上有非常多讲解得非常好的文章，大家可以自行阅读。
最后介绍一下之前看到的一个Node.js的pr(https://github.com/libuv/libuv/pull/2322)，这个pr引入了io_uring。虽然不是取代epoll对Libuv的核心进行重构，但是依然值得探讨。该pr涉及了150+文件，不过大部分是Liburing的代码，我们只关注核心改动。首先Libuv初始化的时候做了一个处理。
复制
// loop里做了修改 struct loop { ... // int backend_fd; 改成下面的联合体 union { int fd; void* data; } backend; } // 定义一个使用io_uring时的结构体 struct uv__backend_data_io_uring { // io_uring的fd int fd; // 等待io_uring处理的任务个数 int32_t pending; // io_uring相关结构体 struct io_uring ring; // 用于epoll中监听io_uring是否有事件触发 uv_poll_t poll_handle; }; // 分配一个uv__backend_data_io_uring结构体 backend_data = uv__malloc(sizeof(*backend_data)); ring = &backend_data->ring; // 初始化io_uring rc = io_uring_queue_init(IOURING_SQ_SIZE, ring, 0); // epoll的fd backend_data->fd = fd; // 初始化 uv__handle_init(loop, &backend_data->poll_handle, UV_POLL); backend_data->poll_handle.flags |= UV_HANDLE_INTERNAL; // 初始化poll_handle的io观察者，fd是io_uring的fd，回调是uv__io_uring_done。 uv__io_init(&backend_data->poll_handle.io_watcher, uv__io_uring_done, ring->ring_fd); loop->flags |= UV_LOOP_USE_IOURING; loop->backend.data = backend_data;
1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.
34.
35.
36.
37.
38.
39.
40.
41.
42.
43.
44.
45.
46.
47.
48.
49.
50.
51.
52.
53.
54.
55.
56.
57.
58.
59.
我们看到初始化时对io_uring进行了初始化并且初始化了一个io观察者。接下来我们看在哪里使用。
复制
int uv_fs_read(uv_loop_t* loop, uv_fs_t* req, uv_file file, const uv_buf_t bufs[], unsigned int nbufs, int64_t off, uv_fs_cb cb) { int rc; INIT(READ); req->file = file; req->nbufs = nbufs; req->bufs = req->bufsml; memcpy(req->bufs, bufs, nbufs * sizeof(*bufs)); req->off = off; /* 优先调用uv__platform_fs_read，不支持则降级到原来线程池的方案 static int uv__fs_retry_with_threadpool(int rc) { return rc == UV_ENOSYS || rc == UV_ENOTSUP || rc == UV_ENOMEM; } */ rc = uv__platform_fs_read(loop, req, file, bufs, nbufs, off, cb); if (!uv__fs_retry_with_threadpool(rc)) return rc; // 走到这说明使用降级方案 POST; }
1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
uv_fs_read函数是读取文件内容时执行的函数，之前时候给线程池提交一个任务，修改后，加了个前置的逻辑uvplatform_fs_read。我们看看uvplatform_fs_read。
复制
int uv__platform_fs_read(uv_loop_t* loop, uv_fs_t* req, uv_os_fd_t file, const uv_buf_t bufs[], unsigned int nbufs, int64_t off, uv_fs_cb cb) { return uv__io_uring_fs_work(IORING_OP_READV, loop, req, file, bufs, nbufs, off, cb);}int uv__io_uring_fs_work(uint8_t opcode, uv_loop_t* loop, uv_fs_t* req, uv_os_fd_t file, const uv_buf_t bufs[], unsigned int nbufs, int64_t off, uv_fs_cb cb) { struct uv__backend_data_io_uring* backend_data; struct io_uring_sqe* sqe; int submitted; uint32_t incr_val; uv_poll_t* handle; backend_data = loop->backend.data; incr_val = (uint32_t)backend_data->pending + 1; // 获取一个请求结构体 sqe = io_uring_get_sqe(&backend_data->ring); // 初始化请求 sqe->opcode = opcode; sqe->fd = file; sqe->off = off; sqe->addr = (uint64_t)req->bufs; sqe->len = nbufs; // 管理req上下文，任务完成时会用到 sqe->user_data = (uint64_t)req; // 提交给内核，非阻塞式调用，返回提交任务的个数 submitted = io_uring_submit(&backend_data->ring); // 提交成功 if (submitted == 1) { req->priv.fs_req_engine |= UV__ENGINE_IOURING; // 提交的时是第一个任务，则注册io观察者的等待可读事件 if (backend_data->pending++ == 0) { handle = &backend_data->poll_handle; uv__io_start(loop, &handle->io_watcher, POLLIN); uv__handle_start(handle); } return 0; } return UV__ERR(errno); }
1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.
34.
35.
36.
37.
38.
39.
40.
41.
42.
43.
44.
45.
46.
47.
48.
49.
50.
51.
52.
53.
54.
55.
56.
57.
我们看到上面的代码会给内核提交一个任务，但是不会等待内核返回，并在提交第一个任务的时候给epoll注册一个等待可读事件。我们看看io_uring的poll接口的实现(epoll原理可参考之前的文章)。
复制
static __poll_t io_uring_poll(struct file *file, poll_table *wait){ struct io_ring_ctx *ctx = file->private_data; __poll_t mask = 0; poll_wait(file, &ctx->cq_wait, wait); smp_rmb(); // 提交队列没满则可写 if (READ_ONCE(ctx->rings->sq.tail) - ctx->cached_sq_head != ctx->rings->sq_ring_entries) mask |= EPOLLOUT | EPOLLWRNORM; // 完成队列非空则可读 if (io_cqring_events(ctx, false)) mask |= EPOLLIN | EPOLLRDNORM; return mask;}static unsigned io_cqring_events(struct io_ring_ctx *ctx, bool noflush){ struct io_rings *rings = ctx->rings; smp_rmb(); // 完成队列非空则可读 return ctx->cached_cq_tail - READ_ONCE(rings->cq.head); }
1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
所以当io_uring有任务完成，即完成队列非空的时候，就会在Libuv的poll io被检测到，从而执行回调。
复制
void uv__io_uring_done(uv_loop_t* loop, uv__io_t* w, unsigned int events) { uv_poll_t* handle; struct io_uring* ring; struct uv__backend_data_io_uring* backend_data; struct io_uring_cqe* cqe; uv_fs_t* req; int finished1; handle = container_of(w, uv_poll_t, io_watcher); backend_data = loop->backend.data; ring = &backend_data->ring; finished1 = 0; while (1) { // 获取完成节点 io_uring_peek_cqe(ring, &cqe); // 全部任务完成则注销事件 if (--backend_data->pending == 0) uv_poll_stop(handle); // 获取响应对应的请求上下文 req = (void*) (uintptr_t) cqe->user_data; if (req->result == 0) req->result = cqe->res; io_uring_cq_advance(ring, 1); // 执行回调 req->cb(req); } }
1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
至此我们看到了这个pr的逻辑，主要是为文件io引入了io_uring，文件io因为兼容性问题，在Libuv中使用线程池实现的，而io_uring支持普通文件，自然可以用于在Linux新版本上替换掉线程池方案。
后记：io_uring既强大又复杂。一切都交给内核来处理，完成后通知我们，我们不仅不需要再手动执行read，同时也减少了系统调用的成本，尤其需要多次read的时候。看起来是一个很棒的事情，io_uring---Linux上真正的异步IO。但其中所蕴含的知识远不止于此，有空再更。