使用 Go 动手实践 io_uring。
5.00/5 (2投票s)
现在，我们有了一个全新的接口用于与内核执行 I/O：io_uring。
在 Linux 中，系统调用（syscall）是一切的核心。它们是应用程序与内核交互的主要接口。因此，它们的速度至关重要。尤其是在 Spectre/Meltdown 之后的世界，这一点更加重要。
大部分系统调用都涉及 I/O，因为这是大多数应用程序所做的。对于网络 I/O，我们已经有了 `epoll` 系列系统调用，它们提供了相当快的性能。但在文件系统 I/O 方面，事情有些不足。我们已经有了 `async_io` 一段时间了，但除了少数特定应用程序外，它并没有多大益处。主要原因是它仅在文件以 `O_DIRECT` 标志打开时才有效。这会使内核绕过任何操作系统缓存，直接尝试从设备读取和写入。当我们试图加快速度时，这不是一种很好的 I/O 方式。而在缓冲模式下，它的行为会是同步的。
所有这些都在缓慢改变，因为现在我们有一个全新的接口用于与内核执行 I/O：`io_uring`。
围绕它有很多讨论。而且是很有道理的，因为它提供了一个全新的与内核交互的模型。让我们深入探讨一下，试着理解它是什么以及它如何解决问题。然后，我们将用 Go 构建一个小型演示应用程序来玩一下。
背景
让我们回顾一下，想想通常的系统调用是如何工作的。我们发起一个系统调用，用户层中的应用程序调用内核，并将数据复制到内核空间。内核执行完毕后，会将结果复制回用户空间缓冲区。然后返回。所有这些都发生在系统调用阻塞的同时。
从一开始，我们就能看到很多瓶颈。有很多数据复制，而且是阻塞的。Go 通过在应用程序和内核之间引入另一个层：运行时，来解决这个问题。它使用一个虚拟实体（通常称为 *P*），其中包含一个待执行的 goroutine 队列，然后将其映射到 OS 线程。
这种间接层允许它进行一些有趣的优化。每当进行阻塞系统调用时，运行时都会意识到这一点，并会从执行 goroutine 的 *P* 中分离出线程，并获取一个新线程来执行其他 goroutine。这被称为“交接”。当系统调用返回时，运行时会尝试将其重新附加到 *P*。如果无法获得空闲的 *P*，它就会将 goroutine 推送到一个稍后执行的队列中，并将线程存储在一个池中。这就是 Go 在代码进入系统调用时如何呈现“非阻塞”外观。
这很棒，但它仍然没有解决主要问题，即数据复制仍然发生，并且实际的系统调用仍然阻塞。
让我们来考虑第一个问题：数据复制。我们如何防止数据从用户空间复制到内核空间？嗯，显然我们需要某种共享内存。好的，可以使用 `mmap` 系统调用来完成，它可以映射用户和内核之间共享的一块内存。
这解决了复制问题。但同步呢？即使我们不复制，我们也需要某种方式来同步我们与内核之间的数据访问。否则，我们会遇到同样的问题，因为应用程序需要再次调用系统调用来执行锁定。
如果我们把用户和内核作为两个相互通信的独立组件来思考，这本质上是一个生产者-消费者问题。用户创建系统调用请求，内核接受它们。一旦完成，它就会通知用户它已准备就绪，用户则接受它们。
幸运的是，这个问题有一个古老的解决方案：环形缓冲区。环形缓冲区允许生产者和消费者之间高效地同步，而无需任何锁定。正如您可能已经猜到的，我们需要两个环形缓冲区：一个提交队列 (SQ)，用户充当生产者并推入系统调用请求，内核进行消费；以及一个完成队列 (CQ)，内核充当生产者并推入完成结果，用户进行消费。
通过这种模型，我们完全消除了内存复制和锁。所有从用户到内核的通信都可以非常高效地进行。这本质上是 `io_uring` 实现的核心思想。让我们简要了解一下它的内部机制，看看它是如何实现的。
io_uring 简介
要将请求推送到 SQ，我们需要创建一个提交队列条目 (SQE)。假设我们要读取一个文件。忽略许多细节，一个 SQE 基本包含：
- 操作码：描述要进行的系统调用的操作码。由于我们对读取文件感兴趣，我们将使用 `readv` 系统调用，它映射到操作码 `IORING_OP_READV`。
- 标志：这些是任何请求都可以使用的修饰符。稍后我们将讨论它们。
- Fd：我们要读取的文件的文件描述符。
- 地址：对于我们的 `readv` 调用，它会创建一个缓冲区（或向量）数组来读取数据。因此，地址字段包含该数组的地址。
- 长度：我们的向量数组的长度。
- 用户数据：一个标识符，用于在请求从完成队列中出来时与其关联。请记住，完成结果不一定与提交队列中的顺序相同。那样的话，就失去了拥有异步 API 的意义。因此，我们需要某种东西来标识我们发出的请求。这起到了这个作用。通常是指向某个包含请求元数据结构的指针。
在完成端，我们从 CQ 获取一个完成队列事件 (CQE)。这是一个非常简单的结构，包含：
- 结果：`readv` 系统调用的返回值。如果成功，它将包含读取的字节数；否则，它将包含一个错误代码。
- 用户数据：我们在 SQE 中传递的标识符。
这里有一个重要的细节需要注意：SQ 和 CQ 都在用户和内核之间共享。但 CQ 实际上包含 CQE，而 SQ 有点不同。它本质上是一个间接层，其中 SQ 数组中索引的值实际上包含指向保存 SQE 项的实际数组的索引。这对于某些应用程序很有用，因为这些应用程序的提交请求位于内部结构中，因此允许它们在一个操作中提交多个请求，从而简化了 `io_uring` API 的采用。
这意味着我们实际上有三样东西映射在内存中：提交队列、完成队列和提交队列数组。下图应该能清楚地说明这一点。
现在让我们回顾一下之前跳过的 `flags` 字段。正如我们所讨论的，CQE 条目可能与它们在队列中的提交顺序完全不同。这就带来了一个有趣的问题。如果我们想依次执行一系列 I/O 操作怎么办？例如，文件复制。我们会想从一个文件描述符读取，然后写入另一个。在当前情况下，在看到读取事件出现在 CQ 中之前，我们甚至无法开始提交写入操作。这时 `flags` 就派上用场了。
我们可以设置 `flags` 字段中的 `IOSQE_IO_LINK` 来实现这一点。如果设置了此标志，下一个 SQE 将自动链接到当前 SQE，并且在当前 SQE 完成之前不会开始执行。这允许我们按照需要强制执行 I/O 事件的顺序。文件复制只是一个例子。理论上，我们可以将*任何*系统调用一个接一个地链接起来，直到我们推送一个字段未设置的 SQE，此时，链条就被认为中断了。
系统调用
有了对 `io_uring` 如何工作的简要概述，让我们来看看实际实现这一目标的系统调用。只有两个。
int io_uring_setup(unsigned entries, struct io_uring_params *params);
`entries` 表示此环的 SQE 数量。`params` 是一个结构，包含有关应用程序将使用的 CQ 和 SQ 的各种详细信息。它返回一个指向此 `io_uring` 实例的文件描述符。
int io_uring_enter(unsigned int fd, unsigned int to_submit, unsigned int min_complete, unsigned int flags, sigset_t sig);
此调用用于向内核提交请求。让我们快速浏览一下重要的部分：
- `fd` 是上一个调用返回的环的文件描述符。
- `to_submit` 告诉内核要从环中消耗多少个条目。请记住，环是在共享内存中。因此，我们可以在要求内核处理它们之前随意推送任意数量的条目。
- `min_complete` 指示调用应该等待多少个条目完成才能返回。
细心的读者会注意到，将 `to_submit` 和 `min_complete` 放在同一个调用中意味着我们可以使用它来仅执行提交，或仅执行完成，或两者兼而有之！这为 API 提供了各种有趣的用法，具体取决于应用程序的工作负载。
轮询模式
对于延迟敏感型应用程序或 IOPS 极高的应用程序，让设备驱动程序每次在有数据可读时都中断内核效率不高。如果我们有大量数据要读取，高频率的中断实际上会降低内核处理事件的吞吐量。在这些情况下，我们实际上会回退到轮询设备驱动程序。要使用 `io_uring` 进行轮询，我们可以在 `io_uring_setup` 调用中设置 `IORING_SETUP_IOPOLL` 标志，并在 `io_uring_enter` 调用中设置 `IORING_ENTER_GETEVENTS` 来持续轮询事件。
但这仍然需要我们用户来发起调用。为了更进一步，`io_uring` 还具有“内核端轮询”功能，如果我们在此 `io_uring_params` 中设置 `IORING_SETUP_SQPOLL` 标志，内核将自动轮询 SQ 以检查新条目并消耗它们。这本质上意味着我们可以执行所有所需的 I/O，而无需进行*任何*一次*系统*调用*。这改变了一切。
但是所有这些灵活性和原始功能都要付出代价。直接使用此 API 并不简单，而且容易出错。由于我们的数据结构在用户和内核之间共享，我们需要设置内存屏障（强制内存操作排序的神奇编译器咒语）和其他细节才能正确完成工作。
幸运的是，`io_uring` 的创建者 Jens Axboe 创建了一个名为 `liburing` 的包装库来帮助简化这一切。使用 `liburing`，我们大致需要执行以下步骤：
- `io_uring_queue_(init|exit)` 用于设置和拆除环。
- `io_uring_get_sqe` 用于获取 SQE。
- `io_uring_prep_(readv|writev|other)` 用于标记要使用的系统调用。
- `io_uring_sqe_set_data` 用于标记用户数据字段。
- `io_uring_(wait|peek)_cqe` 用于等待 CQE 或在不等待的情况下查看它。
- `io_uring_cqe_get_data` 用于获取用户数据字段。
- `io_uring_cqe_seen` 用于将 CQE 标记为已完成。
在 Go 中封装 io_uring
这是很多理论需要消化。还有更多内容我为了简洁起见故意省略了。现在，让我们回到用 Go 编写一些代码，并尝试一下。
为了简单和安全，我们将使用 `liburing` 库，这意味着我们需要使用 CGo。没关系，因为这只是一个玩具，正确的方法是在 Go 运行时中拥有原生支持。因此，我们将不得不使用回调。在原生 Go 中，正在运行的 goroutine 会被运行时暂停，然后在完成队列中有数据可用时唤醒。
我们将包命名为 `frodo`（就这样，我解决了计算机科学中最难的两个问题之一）。我们将只有一个非常简单的 API 来读写文件。以及另外两个函数，用于在完成后设置和清理环。
我们的主力将是一个单独的 goroutine，它接收提交请求并将它们推送到 SQ。然后从 C 发起回调到 Go，并附带 CQE 条目。我们将使用文件的 `fd` 来知道在收到数据时要执行哪个回调。但是，我们也需要决定何时实际将队列提交给内核。我们维护一个队列阈值，如果超过挂起请求的阈值，我们就提交。此外，我们还公开另一个函数供用户自己进行提交，以便他们能够更好地控制应用程序行为。
请注意，再次强调，这是一种低效的做法。由于 CQ 和 SQ 完全分离，它们不需要任何类型的锁定，因此提交和完成可以从不同线程自由进行。理想情况下，我们只会将一个条目推送到 SQ，然后有一个单独的 goroutine 监听完成等待，每当看到一个条目时，我们就发起一个回调并返回等待。还记得我们如何使用 `io_uring_enter` 来仅执行完成吗？这只是一个例子！这仍然是每个 CQE 条目进行一次系统调用，我们甚至可以通过指定要等待的 CQE 条目数量来进一步优化它。
回到我们简单的模型，以下是它的伪代码：
// ReadFile reads a file from the given path and returns the result as a byte slice
// in the passed callback function.
func ReadFile(path string, cb func(buf []byte)) error {
f, err := os.Open(path)
// handle error
fi, err := f.Stat()
// handle error
submitChan <- &request{
code: opCodeRead, // a constant to identify which syscall we are going to make
f: f, // the file descriptor
size: fi.Size(), // size of the file
readCb: cb, // the callback to call when the read is done
}
return nil
}
// WriteFile writes data to a file at the given path. After the file is written,
// it then calls the callback with the number of bytes written.
func WriteFile(path string, data []byte, perm os.FileMode, cb func(written int)) error {
f, err := os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, perm)
// handle error
submitChan <- &request{
code: opCodeWrite, // same as above. This is for the writev syscall
buf: data, // the byte slice of data to be written
f: f, // the file descriptor
writeCb: cb, // the callback to call when the write is done
}
return nil
}
`submitChan` 将请求发送到我们的主工作 goroutine，该 goroutine 负责提交它们。这是它的伪代码：
queueSize := 0
for {
select {
case sqe := <-submitChan:
switch sqe.code {
case opCodeRead:
// We store the fd in our cbMap to be called later from the callback from C.
cbMap[sqe.f.Fd()] = cbInfo{
readCb: sqe.readCb,
close: sqe.f.Close,
}
C.push_read_request(C.int(sqe.f.Fd()), C.long(sqe.size))
case opCodeWrite:
cbMap[sqe.f.Fd()] = cbInfo{
writeCb: sqe.writeCb,
close: sqe.f.Close,
}
C.push_write_request(C.int(sqe.f.Fd()), ptr, C.long(len(sqe.buf)))
}
queueSize++
if queueSize > queueThreshold { // if queue_size > threshold, then pop all.
submitAndPop(queueSize)
queueSize = 0
}
case <-pollChan:
if queueSize > 0 {
submitAndPop(queueSize)
queueSize = 0
}
case <-quitChan:
// possibly drain channel.
// pop_request till everything is done.
return
}
}
`cbMap` 将文件描述符映射到要调用的实际回调函数。当 CGo 代码调用 Go 代码以指示事件完成时，会使用它。`submitAndPop` 使用 `queueSize` 调用 `io_uring_submit_and_wait`，然后从 CQ 中弹出条目。
让我们看看 `C.push_read_request` 和 `C.push_write_request` 在做什么。它们本质上是将读/写请求推送到 SQ。
它们看起来像这样：
int push_read_request(int file_fd, off_t file_sz) {
// Create a file_info struct
struct file_info *fi;
// Populate the struct with the vectors and some metadata
// like the file size, fd and the opcode IORING_OP_READV.
// Get an SQE.
struct io_uring_sqe *sqe = io_uring_get_sqe(&ring);
// Mark the operation to be readv.
io_uring_prep_readv(sqe, file_fd, fi->iovecs, total_blocks, 0);
// Set the user data section.
io_uring_sqe_set_data(sqe, fi);
return 0;
}
int push_write_request(int file_fd, void *data, off_t file_sz) {
// Create a file_info struct
struct file_info *fi;
// Populate the struct with the vectors and some metadata
// like the file size, fd and the opcode IORING_OP_WRITEV.
// Get an SQE.
struct io_uring_sqe *sqe = io_uring_get_sqe(&ring);
// Mark the operation to be writev.
io_uring_prep_writev(sqe, file_fd, fi->iovecs, 1, 0);
// Set the user data section.
io_uring_sqe_set_data(sqe, fi);
return 0;
}
当 `submitAndPop` 尝试从 CQ 中弹出条目时，会执行此操作：
int pop_request() {
struct io_uring_cqe *cqe;
// Get an element from CQ without waiting.
int ret = io_uring_peek_cqe(&ring, &cqe);
// some error handling
// Get the user data set in the set_data call.
struct file_info *fi = io_uring_cqe_get_data(cqe);
if (fi->opcode == IORING_OP_READV) {
// Calculate the number of blocks read.
// Call read_callback to Go.
read_callback(fi->iovecs, total_blocks, fi->file_fd);
} else if (fi->opcode == IORING_OP_WRITEV) {
// Call write_callback to Go.
write_callback(cqe->res, fi->file_fd);
}
// Mark the queue item as seen.
io_uring_cqe_seen(&ring, cqe);
return 0;
}
`read_callback` 和 `write_callback` 仅使用传递的 `fd` 从 `cbMap` 获取条目，然后调用原始 `ReadFile`/`WriteFile` 调用函数的相应回调。
//export read_callback
func read_callback(iovecs *C.struct_iovec, length C.int, fd C.int) {
var buf bytes.Buffer
// Populate the buffer with the data passed.
cbMut.Lock()
cbMap[uintptr(fd)].close()
cbMap[uintptr(fd)].readCb(buf.Bytes())
cbMut.Unlock()
}
//export write_callback
func write_callback(written C.int, fd C.int) {
cbMut.Lock()
cbMap[uintptr(fd)].close()
cbMap[uintptr(fd)].writeCb(int(written))
cbMut.Unlock()
}
基本上就是这样！一个使用该库的示例将如下所示：
err := frodo.ReadFile("shire.html", func(buf []byte) {
// handle buf
})
if err != nil {
// handle err
}
欢迎查看源代码以深入了解实现细节。
性能
没有性能数据就没有完整的博客文章。然而，对 I/O 引擎进行适当的基准测试比较可能需要另一篇博客文章。为了完整起见，我将仅发布我在笔记本电脑上进行的简短非科学测试的结果。不要太在意这些结果，因为任何基准测试都高度依赖于工作负载、队列参数、硬件、一天中的时间以及您的衬衫颜色。
我们将使用由 Jens 本人编写的 `fio`，这是一个很棒的工具，用于使用不同的工作负载对多个 I/O 引擎进行基准测试，它同时支持 `io_uring` 和 `libaio`。可更改的选项太多了。但我们将执行一个非常简单的实验，使用 75/25 的随机读/写比例工作负载，使用 1GiB 文件，并改变 16KiB、32KiB 和 1MiB 的块大小。然后，我们将使用 8、16 和 32 的队列大小重复整个实验。
请注意，这是 `io_uring` 的基本模式，没有轮询，在这种情况下，结果可能会更高。
结论
这是一篇很长的帖子，感谢您阅读到这里！
`io_uring` 仍处于初级阶段，但正在迅速获得大量关注。许多大公司（例如 libuv 和 RocksDB）已经支持它。甚至还有一个针对 `nginx` 的补丁增加了 `io_uring` 支持。希望 Go 原生支持能够尽快实现。
内核的每个新版本都会带来 API 的新功能，并且越来越多的系统调用开始得到支持。这是 Linux 性能领域令人兴奋的新前沿！
以下是我为准备此帖子使用的一些资源。如果您想了解更多信息，请务必查看。
最后，我想感谢我的同事Ibrahim 和Claudio 审阅并纠正了我糟糕的 C 代码。
有任何问题或评论吗？请加入我们的 community.mattermost.com，并在那里给我发消息 @agnivade。
资源
- https://kernel.dk/io_uring.pdf
- https://kernel.dk/io_uring-whatsnew.pdf
- https://www.youtube.com/watch?v=-5T4Cjw46ys
- https://lwn.net/Articles/776703/
- https://github.com/axboe/liburing