探索使用 io_uring 的自动缓冲区管理

# 使用 io_uring 探索自动缓冲区管理 来源：https://noteflakes.com/articles/2026-06-07-automatic-buffer-management ### 2026年6月7日 在过去大约一年里，我一直在开发 [UringMachine](https://github.com/digital-fabric/uringmachine)，这是一个用于通过 io\_uring 进行 I/O 操作的 Ruby gem，并作为我接受 [Ruby 协会](https://www.ruby.or.jp/en/) 资助工作的一部分，在我的网站上定期汇报进展。 ## 快速回顾 简要回顾一下 UringMachine 的功能：UringMachine 提供了一个底层 API，用于通过 io\_uring 执行 I/O 操作——io\_uring 是近期 Linux 内核上用于异步执行 I/O 操作的接口。 UringMachine 还提供了一个 [Fiber Scheduler](https://docs.ruby-lang.org/en/master/Fiber/Scheduler.html) 实现，使其能够与 Ruby 生态系统的其余部分很好地集成，并可用于任何支持纤程并发的 Ruby 应用。 在这个项目的工作中，我一直在寻找恰到好处的抽象层次：一方面能够充分利用 io\_uring 的强大能力，为 Ruby 带来高性能 I/O；另一方面则提供便捷实用的 Ruby API，以及与整个 Ruby 生态系统的良好整合。 以下是我在资助工作开始以来所从事的一些事项： - 完整的 `FiberScheduler` 接口实现。 - 对 Ruby 运行时中 `FiberScheduler` 集成代码的一些小贡献。 - 全面的测试。 - 对不同 I/O 方法中 `IO::Buffer` 的支持。 - 对向量化 `writev` / `sendv` 的支持。 - 全面的度量指标。 - 对 SQPOLL 模式的支持。 - 对 Sidecar 模式的支持。 - 大量的 [基准测试](https://noteflakes.com/articles/2025-12-19-friday-update)。 ## 自动缓冲区管理 在最近几个月里，我一直在为 UringMachine 实现自动缓冲区管理。按照我的习惯，我思考了这个特性的设计，并尝试了不同的想法。在去年圣诞节假期之后，我认为设计已经足够成熟，可以开始编写代码了。但让我先退一步，解释一下我试图实现什么。 io\_uring 接口较新的一项功能是设置缓冲区环（buffer rings）的设施。其思路是：应用向内核提供缓冲区，内核可以在给定的文件或套接字上重复使用这些缓冲区进行读取或接收数据，并通过每个 CQE 告知应用使用了哪个缓冲区以及其中放入了多少数据。 应用在每个连接上发起多投读取（multishot read）/ recv 操作，内核拥有一组应用提供的缓冲池，每当读取/接收到数据块时即可使用。内核根据需要使用这些缓冲区，并将从套接字读取到的数据块填入其中。这些数据块将在应用准备好处理 CQEs 时稍后进行处理。最终，处理完数据后，应用将已消耗的缓冲区重新添加回缓冲区环，使其再次可供内核使用。 应用可以注册多个缓冲区环，每个环具有设置的最大缓冲区数量和一个缓冲区组 ID（`bgid`）。添加到缓冲区环的缓冲区可以是任意大小。缓冲区环中的每个缓冲区还有一个 ID（`bid`）。因此，缓冲区通过元组 `[bgid, bid]` 来标识。当提交多投读取/recv 操作时，我们指明缓冲区组 ID（`bgid`），告知内核使用哪个缓冲区环。然后内核生成 CQEs（完成队列条目），其中包含包含数据的缓冲区的 ID（`bid`）。关键是，单个缓冲区环可以同时用于不同文件描述符上的多个并发多投读取/recv 操作。 此外，在较新的内核上，io\_uring 能够部分消耗缓冲区，从而防止缓冲区空间浪费。当缓冲区环设置为 [部分消耗](https://www.man7.org/linux/man-pages/man3/io_uring_setup_buf_ring.3.html) 时，每个与多投读取/recv 操作相关的 CQE 还会带有一个标志，告知应用该缓冲区 [是否会被进一步使用](https://www.man7.org/linux/man-pages/man3/io_uring_prep_recv.3.html)（超出当前可用数据量）。每次具有相同缓冲区 ID 的读取/recv 完成会从上次离开之处继续。这意味着缓冲区空间被充分利用，但“缺点”是应用需要为每个缓冲区跟踪一个“游标”。 所以我设计了一个自动管理缓冲区的子系统：注册缓冲区组，分配缓冲区并将其添加到各个缓冲区环，跟踪每个缓冲区的使用情况。但我还希望从应用的角度提出一种使用这些缓冲区的好方法。 ## 应用如何使用缓冲区 Ruby 应用中通常如何使用 I/O 缓冲区？标准的 `IO` 类方便地包含了从文件/套接字读取和写入的缓冲功能。这使我们能够实现像 `IO#gets` 这样的 API，它执行带缓冲的读取并在读取缓冲区中查找行分隔符。 根据协议的不同，我们可能需要逐行读取数据，或读取单个字节，或任意长度的字符串，或者这些方式的组合。因此，一个想要解析（比如）HTTP/1.1 请求的应用，需要首先读取请求头部，每个头部以 `\r\n` 分隔符结束，然后根据给定的头部读取任意长度的请求体。这使得需要将数据读入缓冲区，而随着更多数据的读取，缓冲区可能需要调整大小和/或截断。 所以，我们可以想象一种抽象，让我们从某个字节流的源进行读取。我们可能想读取一行： 或者想精确读取 42 个字节： 为此，我们需要缓冲从流中读取的数据，因为我们要么需要读取直到遇到分隔符，要么需要读取精确的数量（但可能得到较短的读取），并且我们希望缓冲能够自动工作，就像在标准的 `IO` 类中那样——你不需要费心，只需调用 `IO#read` 即可。 因此，我们的目标是： - 提供简单的 API，同时适用于二进制和基于行的协议。 - 使用 io\_uring 的提供缓冲区（provided buffers）特性。 - 重用缓冲区。 - 根据读取压力调整总缓冲区空间。 - 最小化缓冲区分配。 - 最小化读取数据的拷贝。 现在让我们看看 UringMachine 如何实现这些目标。 ## 自动为读取操作提供缓冲区 如上所述，io\_uring 将提供的缓冲区组织成缓冲区组（或缓冲区环）。同一个缓冲区组可以用于任意数量的并发多投读取——这意味着 io\_uring 可以使用相同的缓冲区空间来存放来自应用当前服务的任意数量套接字的数据。应用只需与内核一起跟踪缓冲区使用情况，以了解每个套接字的数据位于何处。 因此，我们首先设置一个有 1024 个条目的缓冲区环，它将用于任何多投读取/recv。我们用 16 个缓冲区填充该缓冲区环/组，每个缓冲区大小为 16KB，总共 256KB。我们的目标是随时保持可用缓冲区空间在 128KB 到 256KB 之间。 当执行多投读取时，io\_uring 会逐步消耗这些缓冲区中的数据，因此对于每个缓冲区我们还有一个游标，跟踪已消耗多少数据。从 io\_uring 收到的每个 CQE 中，内核告诉我们使用了哪个缓冲区以及读入了多少数据，我们可以据此增加游标。 随着多投 CQEs 的到达，我们还可以跟踪内核可用的总缓冲区空间量。我们设置了一个 *自动补充* 机制来跟踪总缓冲区空间，如果它低于 128KB，则向缓冲区组中添加更多缓冲区，以恢复至少 256KB 可供内核使用的空间。 如果有大量数据同时到达，可能会出现缓冲区空间耗尽的情况。内核会通过停止多投读取并返回 `ENOBUFS` 错误码来告知我们（这也意味着我们需要重新启动多投读取）。在这种情况下，自动补充机制会将总缓冲区空间水平以及最低阈值加倍，因此可用缓冲区空间将始终保持 256KB 到 512KB。 ## 最小化拷贝 尽管我们的目标之一是尽可能避免拷贝数据，但我们面临一个问题：由于相同的缓冲区可能用于多个并发的多投读取，我们不能保证某个特定套接字的读取数据在缓冲区中是连续的。换句话说，我们需要能够处理 *分段* 缓冲区，它由一个或多个 *段* 组成。每个段基本上是对特定缓冲区中一块数据的引用。然后我们可以将这些段组织成链表，从而能够重新构成整个接收到的消息： 这意味着我们只需要在将读取数据转换为 Ruby 字符串时拷贝一次数据。在上面的例子中，当调用 `#read_line` 时，我们从第一个段开始搜索 `\n` 的首次出现。一旦在第三个段中找到分隔符，我们就可以分配一个具有所需容量的 Ruby 字符串，并将每个段中的数据拷贝到该字符串中。 这样我们只将数据从那些缓冲区拷贝一次。一旦某个缓冲区已被内核完全消耗，并且所有引用该缓冲区的段都被应用消耗完毕，该缓冲区就可以安全地回收并最终再次提供给内核。 ## 最小化分配 我们有一些为内核提供通用读取空间的缓冲区，我们将这些缓冲区提供给内核，内核消耗它们，然后我们从这些缓冲区读取数据；一旦某个缓冲区使用完毕，我们希望能够重用它。我们还有那些需要分配和管理的小型段结构体（struct）。如何做到这一点？当某个段被消耗后，我们将其放入一个空闲列表，这样下次需要段结构体时，只需从空闲列表中取出一个即可。这样，我们最小化了分配次数。实际上，UringMachine 对保存 I/O 操作元数据的 `um_op` 结构体以及其他各种结构体类型也采用了相同的做法。 当然，由于我们可以使用相同的缓冲区来服务任意数量的正在进行的多投读取操作，这意味着我们不再需要为要读取的每个文件描述符分配（并在之后释放）缓冲区空间。 ## 整合在一起 我喜欢这个设计的一点是，它利用了一个高级的 io\_uring 特性，并且以一种对开发者完全透明的方式实现，开发者能够受益于简单实用的 API。在 UringMachine 中，我选择将这一 API 作为 `UringMachine::IO` 类的一部分提供，该类提供了一小组用于缓冲 I/O 的方法，以及一些用于写入/发送数据和查询缓冲区状态的其他方法： ```ruby # 实例化一个 IO 对象 io = UM::IO.new(machine, fd) # 或者： io = machine.io(fd) io.read(count) #=> 从流中读取 count 个字节 io.read_line(maxlen) #=> 读取直到遇到 \n io.read_to_delim(delim, maxlen) # 读取直到遇到分隔符 io.read_each { |segment| } # 遍历各个段 io.skip(count) # 在缓冲区中跳过 count 个字节 io.write(*strings) # 写入给定的字符串 io.clear # 清空缓冲区 ``` 以下是一个基本的 HTTP/1.1 解析器如何构建在它之上： ```ruby # UM::IO 的 HTTP 协议扩展 class UM::IO def http_read_request_headers line = read_line(MAX_REQUEST_LINE_LEN) headers = parse_request_line(line) return nil if !headers loop do line = read_line(MAX_HEADER_LINE_LEN) k, v = parse_header_line(line) break if !k headers[k] = m[v] end headers end def http_read_body(headers) content_length = headers['content-length'] if content_length content_length = content_length.to_i return nil if content_length == 0 chunk = read(content_length) return chunk end nil end # ... end ``` 然后我们可以轻易地在这些 HTTP 协议原语之上构建一个 Web 服务器： ```ruby def handle_http_client(fd) io = @machine.io(fd) while true headers = @io.http_read_request_headers break if !headers body = @io.http_read_body(headers) handle_request(io, headers, body) end ensure @machine.close(fd) end ``` 我认为这个设计的伟大之处在于，一方面它隐藏了 UringMachine 所做的所有缓冲工作，另一方面它让你能够继续以顺序风格编写代码，保持控制权，避免使用回调。 另外，这个设计的抽象层次与协议的设计相匹配，也是我所喜欢的。在 HTTP/1 中，一个 HTTP 请求的序列总是相同的：头部（包括请求行），然后是主体。因此，我们有两个对应于消息结构的方法是合适的：`http_read_request_headers`，然后是 `http_read_body`。 ## 实现其他协议 因此，就像 HTTP 一样，我们也可以在 `UM::IO` 类之上实现其他协议。事实上，UringMachine 包含了一个 Redis 服务器使用的 [RESP 协议](https://redis.io/docs/latest/develop/reference/protocol-spec/) 的实现。 由于 RESP 协议建立在交换简单、可嵌套的数据类型（包括数组和哈希表）之上，我们可以围绕这一点设计协议： ```ruby io.resp_read #=> 读取一个 String, Integer, Array, Hash 等 io.resp_write(obj) # 发送一个对象 ``` 这样，要与 Redis 服务器通信，我们需要做的如下： ```ruby fd = machine.tcp_connect('127.0.0.1', 6379) io = machine.io(fd) # 客户端握手 io.write("HELLO 3\r\n") res = io.resp_read # 发出命令 io.resp_write(['get', 'foo']) value = io.resp_read ``` ## UringMachine 的下一步计划 总而言之，我对 UringMachine 非常满意。设计感觉稳健，性能良好，所包含的纤程调度器实现使其能够与整个 Ruby 生态系统集成。 那么，UringMachine 的下一步是什么？以下是我打算继续从事的一些事项： - 支持 IPv6 地址。 - 支持 `sendto` / `recvfrom`。 - 在 `UM::IO` 之上实现更多协议：HTTP/1、HTTP/2、PostgreSQL 有线协议。 - 允许与 Rails（基本可用！）、Hanami 和 Sidekiq 等项目一同使用。 在接下来的几周里，我将开始撰写关于目前重点工作的项目，该项目基于我在 UringMachine 上的工作成果。保重！