Rust in Everything
基于 hyper 1.12.0

IO trait

主线位置：两条主线的最底层。服务端主线的「字节」从这里读进来、响应字节从这里写出去；客户端主线把请求字节写进同一类对象。hyper 只认 rt::Read + rt::Write + Unpin，TCP、TLS、Unix socket 都要先变成这个样子。

flowchart TB
  buf["BytesMut 空闲容量"] --> rb["ReadBuf::uninit"]
  rb --> cur["unfilled() 游标"]
  cur --> tio["TokioIo::poll_read"]
  tio --> tbuf["tokio ReadBuf"]
  tbuf --> tcp["TcpStream 读"]
  tcp -->|"读到 n 字节"| adv["advance(n)"]
  adv --> got["read_buf 记下 n"]

这张图是服务端读请求时，一次 poll_read 从 hyper 的读缓冲一直走到 tokio 的 TcpStream 再回来的路径。整条路上同一块内存只是换了几层「包装」，没有清零，也没有拷贝。

为什么要自己定义 IO trait

hyper 0.14 直接要求 IO 类型实现 tokio::io::{AsyncRead, AsyncWrite}。1.0 换成了 hyper::rt::{Read, Write}，rt/io.rs 开头的注释说了为什么：

// New IO traits? What?! Why, are you bonkers?
//
// I mean, yes, probably. But, here's the goals:
//
// 1. Supports poll-based IO operations.
// 2. Opt-in vectored IO.
// 3. Can use an optional buffer pool.
// 4. Able to add completion-based (uring) IO eventually.
//
// Frankly, the last point is the entire reason we're doing this. We want to
// have forwards-compatibility with an eventually stable io-uring runtime. We
// don't need that to work right away. But it must be possible to add in here
// without breaking hyper 1.0.

第 4 条是关键。io-uring 这类完成式 IO 是「先把缓冲区交给内核，做完再通知」，和 poll_read 的「就绪了再来读」是两种模型（tokio 那边的情况见 io-uring）。trait 归 tokio 所有，hyper 就没法在不破坏 1.0 API 的前提下往里加东西；归 hyper 自己所有，以后可以加带默认实现的新方法。

📝NOTE
冷知识：1.0 的最后一个 rc 才换掉 tokio 的 IO trait

#3110（issue，标题「hyper should use its own IO traits」）里写着当时的状况：「hyper 1.0 (as of of rc.2) still depends on tokio with all features turned off, to ask that IO transports implement tokio::io::{AsyncRead, AsyncWrite}.」给出的理由有两条：接其他运行时的人「don't want to compile multiple runtimes」；以及「If we own the traits, we can decide to try different things that perhaps Tokio itself wouldn't want to do」，包括向量化读写和完成式 IO。issue 里还提议由 hyper-util 提供一个包装 tokio 类型的适配器。

#3230 落实了这件事：「This replaces the usage of tokio::io::{AsyncRead, AsyncWrite} in hyper's public API with new traits in the hyper::rt module.」合并前的清单里有一条「Prove that it is forwards-compatible with completion-based IO」。提交信息里的 BREAKING CHANGE 让用户「grab a helper type from hyper-util to wrap Tokio types」，那就是下文的 TokioIo。第一个包含它的发布是 v1.0.0-rc.4，之后的 v1.0.0 正式版沿用。

注意说的是「公开 API」。hyper 的 Cargo.toml 里仍然有 tokio = { version = "1", features = ["sync"] }：upgrade.rs 和客户端的 dispatch.rs 用到 tokio 的 oneshot / mpsc。HTTP/2 也绕不开 tokio 的 IO trait，因为 h2 0.4.20 要求 IO 实现 tokio::io::{AsyncRead, AsyncWrite}，hyper 内部用一个 Compat 把 rt::Read + rt::Write 转回去再交给 h2。只开 sync 不会带进 tokio 的运行时、网络或计时器，「不依赖运行时」指的是这一层。

Read 和 Write

Read 去掉文档注释后只有一个方法：

pub trait Read {
    fn poll_read(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: ReadBufCursor<'_>,
    ) -> Poll<Result<(), std::io::Error>>;
}

Write 有五个，后两个带默认实现（同样去掉了文档注释）：

pub trait Write {
    fn poll_write(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &[u8],
    ) -> Poll<Result<usize, std::io::Error>>;

    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), std::io::Error>>;

    fn poll_shutdown(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Result<(), std::io::Error>>;

    fn is_write_vectored(&self) -> bool {
        false
    }

    fn poll_write_vectored(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        bufs: &[std::io::IoSlice<'_>],
    ) -> Poll<Result<usize, std::io::Error>> {
        let buf = bufs
            .iter()
            .find(|b| !b.is_empty())
            .map_or(&[][..], |b| &**b);
        self.poll_write(cx, buf)
    }
}

和 tokio 的 AsyncRead / AsyncWrite 对照：

hyper rttokio io
读的参数ReadBufCursor<'_>，按值传，只能碰未填充的部分&mut ReadBuf<'_>，能看到整个缓冲区
EOF返回 Ready(Ok(())) 且一个字节都没写同
Pending登记 cx.waker()，可读 / 可写或关闭时唤醒同，tokio 的实现见 从 TcpStream::read 到挂起
向量化写is_write_vectored 默认 false，poll_write_vectored 默认只写第一个非空切片同
转发实现Box<T>、&mut T、Pin<P>这三个之外还有 &[u8] 等

方法几乎一一对应，所以两套之间的适配器很薄。真正的区别在读的参数：tokio 给实现方一个 &mut ReadBuf，hyper 只给一个指向未填充部分的游标。

「向量化写」那一行对 hyper 有实际影响：HTTP/1 的写缓冲在创建时就看 io.is_write_vectored()，true 用排队策略（头和 body 分开，用 writev 一次写出），false 用拼平策略（拷进一块连续内存）。tokio 的 TcpStream 返回 true。细节见 读写缓冲，http1::Builder::writev 可以强制指定，见 服务端 Builder 与 Connection。

ReadBuf：两条边界

ReadBuf 是一段 MaybeUninit<u8> 加两个下标：

pub struct ReadBuf<'data> {
    raw: &'data mut [MaybeUninit<u8>],
    filled: usize,
    init: usize,
}

文档注释把它画成：

[             capacity              ]
[ filled |         unfilled         ]
[    initialized    | uninitialized ]

不变量是 filled <= init <= capacity：已填充的一定已初始化，已初始化的不一定已填充。为什么要单独记 init？因为 Rust 里读未初始化内存是未定义行为，&mut [u8] 必须指向已初始化的字节。如果缓冲区只有 filled 一条边界，每次交给 read 之前都要把剩下的空间清零；记下 init 以后，清过一次的部分下次就不用再清。

两个构造函数对应两种起点：

构造init 初值用在哪
ReadBuf::new(&mut [u8])raw.len()，整块都已初始化调用方手里本来就是 [u8]
ReadBuf::uninit(&mut [MaybeUninit<u8>])0hyper 自己的读路径

hyper 读请求时走的是第二种。Buffered::poll_read_from_io 把 BytesMut 的空闲容量直接当成未初始化内存交出去：

        // SAFETY: ReadBuf and poll_read promise not to set any uninitialized
        // bytes onto `dst`.
        let dst = unsafe { self.read_buf.chunk_mut().as_uninit_slice_mut() };
        let mut buf = ReadBuf::uninit(dst);
        match Pin::new(&mut self.io).poll_read(cx, buf.unfilled()) {
            Poll::Ready(Ok(())) => {
                let n = buf.filled().len();
                trace!("received {} bytes", n);
                // Safety: we just read that many bytes into the
                // uninitialized part of the buffer, so this is okay.
                // @tokio pls give me back `poll_read_buf` thanks
                unsafe {
                    self.read_buf.advance_mut(n);
                }
                self.read_buf_strategy.record(n);
                Poll::Ready(Ok(n))
            }

BytesMut 刚 reserve 出来的空间没有初始化，这样读就省掉了每次清零。读缓冲怎么增长、上限多少，见 读写缓冲。

ReadBufCursor：安全的和不安全的写法

Read::poll_read 拿到的是 ReadBufCursor，由 ReadBuf::unfilled 造出来。它只暴露未填充的那部分，实现方碰不到已经读到的数据。方法分两组：

方法安全做什么维护边界
put_slice(&[u8])安全把切片拷进来；放不下就 panic（src.len() must fit in remaining()）filled 和 init 一起前移
initialize_unfilled_to(n) / initialize_unfilled()安全把未填充部分的前 n 字节清零（已初始化的跳过），返回 &mut [u8]只前移 init
remaining()安全剩余容量—
as_mut()unsafe拿到未填充部分的 &mut [MaybeUninit<u8>]调用方不得把已初始化的字节「反初始化」
advance(n)unsafefilled += n，init = max(init, filled)调用方保证这 n 字节确实写过

initialize_unfilled_to 不前移 filled。它把一块内存变成可以当 &mut [u8] 用的样子，适合交给只接受 &mut [u8] 的同步 API 去写；写完之后还要另外前移 filled，而这一步只有 unsafe 的 advance。只用安全方法的实现基本就是 put_slice。

unsafe 那一组是给 TokioIo 这类适配器用的：它们手里也是一块未初始化内存，要原样转交给下一层。

❗IMPORTANT
动脑筋：init 走到了哪里？
hyper = { version = "=1.12.0" }

use std::mem::MaybeUninit;

use hyper::rt::ReadBuf;

fn main() {
  let mut backing = [MaybeUninit::<u8>::uninit(); 64];
  let mut rb = ReadBuf::uninit(&mut backing);
  println!("new:       {rb:?}");

  rb.unfilled().put_slice(b"hello");
  println!("put_slice: {rb:?}");

  let mut cursor = rb.unfilled();
  let zeroed = cursor.initialize_unfilled_to(10);
  zeroed[..3].copy_from_slice(b"abc");
  println!("init_to:   {rb:?} filled={:?}", String::from_utf8_lossy(rb.filled()));
}

ReadBuf 的 Debug 打印 filled、init、capacity。第三行的 filled 和 init 各是多少？filled() 里有没有 abc？

TokioIo：两个方向都能转

hyper-util 的 TokioIo<T> 只有一个字段：

pin_project! {
    #[derive(Debug)]
    pub struct TokioIo<T> {
        #[pin]
        inner: T,
    }
}

它实现了四个 trait，靠的是约束不同：T: AsyncRead 时 TokioIo<T> 实现 hyper 的 Read，T: hyper::rt::Read 时实现 tokio 的 AsyncRead，写的方向同理。所以服务端 accept 到的 TcpStream 包一层就能交给 hyper，而 hyper 升级后交还的 Upgraded（只实现 hyper 的 trait）包一层就能当 tokio 的 IO 用。模块文档里的表（省掉了全为「否」的 WithHyperIo<H>、WithTokioIo<T> 两行）：

AsyncReadAsyncWriteReadWrite
tokio 类型 T是是否否
hyper 类型 H否否是是
TokioIo<T>否否是是
TokioIo<H>是是否否
WithHyperIo<T>是是是是
WithTokioIo<H>是是是是

TokioIo 是「换」，包进去之后原来那一套 trait 就看不到了；要「加」一套、两套都保留，用 WithHyperIo 或 WithTokioIo。

tokio → hyper 方向的 Read 实现：

impl<T> hyper::rt::Read for TokioIo<T>
where
    T: tokio::io::AsyncRead,
{
    fn poll_read(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        mut buf: hyper::rt::ReadBufCursor<'_>,
    ) -> Poll<Result<(), std::io::Error>> {
        let n = unsafe {
            let mut tbuf = tokio::io::ReadBuf::uninit(buf.as_mut());
            match tokio::io::AsyncRead::poll_read(self.project().inner, cx, &mut tbuf) {
                Poll::Ready(Ok(())) => tbuf.filled().len(),
                other => return other,
            }
        };

        unsafe {
            buf.advance(n);
        }
        Poll::Ready(Ok(()))
    }
}

三步：用 as_mut() 拿到 hyper 游标下面那块未初始化内存，原样包成一个 tokio 的 ReadBuf::uninit；让 tokio 的 IO 往里读；读到几个字节，就 advance 几个。Pending 和错误原样返回。整个过程里数据只被内核写了一次，就写在 hyper 的 BytesMut 里。

反方向的 AsyncRead 实现 做法对称：把 tokio ReadBuf 的未填充部分包成 hyper::rt::ReadBuf::uninit，读完后 assume_init(sub_filled) 和 set_filled(filled + sub_filled)。它只告诉 tokio「这次读到的字节已初始化」，不管 hyper 一侧有没有多清零，是保守但正确的做法。写的方向四个方法都是一行转发。

📝NOTE
没有蠢问题

问： 既然 TcpStream 只是要多实现一个 trait，为什么还要 TokioIo::new(stream) 包一层，不直接 impl hyper::rt::Read for TcpStream？

答： 孤儿规则不允许。hyper::rt::Read 是 hyper 的 trait，TcpStream 是 tokio 的类型，只有这两个 crate 之一能写这个实现。hyper-util 两个都不拥有，所以只能定义一个自己的包装类型 TokioIo。hyper 倒是能写，但那样 hyper 的公开 API 就又和 tokio 的类型绑在一起了，正是 1.0 想避开的事。

小结

  • hyper 1.0 用自己的 rt::Read / Write 代替 tokio 的 AsyncRead / AsyncWrite，主要为了以后能加完成式 IO 而不破坏 API（#3230，v1.0.0-rc.4）。hyper 仍依赖 tokio 的 sync 特性，HTTP/2 内部还要把 IO 转回 tokio 的 trait 交给 h2。
  • Read 只有 poll_read，参数是只能看到未填充部分的 ReadBufCursor；Write 有向量化写的开关，HTTP/1 写缓冲按它选排队还是拼平。
  • ReadBuf 记 filled 和 init 两条边界，filled <= init。hyper 的读路径用 ReadBuf::uninit 直接读进 BytesMut 的未初始化容量，不清零。
  • 安全的写法是 put_slice；as_mut + advance 是 unsafe 的，给适配器转交内存用；initialize_unfilled_to 只清零、不前移 filled。
  • TokioIo<T> 按 T 实现的是哪一套 trait，就实现另一套；转换不拷贝数据。要两套都保留用 WithHyperIo / WithTokioIo。

动脑筋答案

M2 Max、rustc 1.99.0、--release，连跑 3 次输出相同：

new:       ReadBuf { filled: 0, init: 0, capacity: 64 }
put_slice: ReadBuf { filled: 5, init: 5, capacity: 64 }
init_to:   ReadBuf { filled: 5, init: 15, capacity: 64 } filled="hello"
  • new：ReadBuf::uninit 从 init = 0 开始，整块都算未初始化。
  • put_slice：拷进 5 个字节，filled 和 init 一起走到 5。
  • init_to：initialize_unfilled_to(10) 把 filled 之后的 10 个字节清零，init 走到 5 + 10 = 15；filled 不动，所以 filled() 还是 hello。写进去的 abc 在内存里，但在 filled 边界之外，不算读到的数据。要让它算数，只能调 unsafe 的 advance(3)。

💬 关联讨论

发起讨论
加载中...