Tokio 源码深度解析

第10章 TcpStream / UdpSocket 源码剖析

作者杨艺韬 · 13,719 字 · 发布于 2026-04-19 · 更新于 2026-09-24

本章要点

TcpStream 的 struct 定义极简：TcpStream { io: PollEvented<mio::net::TcpStream> } —— 一个字段、一层包装
PollEvented 是 Tokio 对任何 I/O 资源的通用包装层：PollEvented<T> 把任意 mio 事件源 T 和一个 Registration 绑定
Registration 是运行时侧的对接——持有 scheduler Handle + Arc<ScheduledIo>。poll_read_ready / poll_write_ready 是它的核心方法
四层包装：TcpStream → PollEvented<mio::net::TcpStream> → Registration → ScheduledIo，再往下才是 mio 和 epoll
TcpStream::connect 是一个 async fn，内部 DNS 解析后用 connect_addr 建立非阻塞连接、等 writable
AsyncRead::poll_read 是 try + register 模式：先查 readiness，就绪才 mio_io.read()；未就绪（或读到 WouldBlock、清掉 readiness 之后）就注册 Waker、返回 Pending
TcpListener::accept 用了漂亮的 async_io helper —— 写任何自定义 async I/O 都可以复用这个 helper
UdpSocket 的设计和 TcpStream 高度同构——学会一个就懂了所有 mio-based I/O 类型

10.1 读本章前的状态检查

第 10 章是前 9 章所有知识的收官章——它把所有底层机制串成一条用户看得到的路径。如果前几章有不扎实的地方、本章会让你感受到明显的吃力。反过来如果前几章读得扎实、本章会让你有"恍然大悟"的感觉。

本章是本专栏上半场的收束。在进入之前，做一次自我检查——如果你对下面任一条还不自信，建议先回翻那一章：

📖 Future trait 和 Poll 语义：不熟 → 读第 2 章
📖 Waker 的 vtable 机制 + wake 路径：不熟 → 读第 3 章
📖 Runtime 和 Handle 的关系：不熟 → 读第 4 章
📖 multi_thread scheduler 的三条腿：不熟 → 读第 5 章
📖 Task 的 Cell / Header / State bit layout：不熟 → 读第 6 章
📖 ScheduledIo 和 poll_readiness：不熟 → 读第 8 章
📖 mio 三层架构和 epoll：不熟 → 读第 9 章

这 7 章是本章的必备基础。本章把它们全部拼起来——每一段代码都依赖前面多个概念的理解。如果你带着某个薄弱环节读本章，会感觉"怎么这么多东西串起来"——那不是章节的问题，是基础没建好。花 1 小时回去补，比硬读本章快。

建立好基础后，本章会出奇地顺畅——因为你会一直看到"哦，这就是我之前学的那个机制在这里的用法"。

10.2 TcpStream 的四层包装

Tokio 的 TcpStream 表面就是一个类型——但它内部包着四层：最外层 是 Tokio 的 API 表面（read / write / peer_addr 等）、第二层 是 PollEvented（通用的"响应式 I/O 对象"抽象，里面装着 mio 的 TcpStream）、第三层 是 Registration（和运行时的连接）、最内层 是 ScheduledIo（per-fd 的 readiness 状态）；mio 和 epoll 在这四层之下，负责 OS 兼容性。这种分层让每一层职责清晰：上层关心"用户 API"、中间两层关心"通用响应式模式"和注册的生命周期、最内层关心"就绪状态"。读懂这四层就读懂了 Tokio 所有 I/O 类型的设计骨架——UdpSocket、UnixStream、NamedPipe 全都是同构的。

打开 tokio/src/net/tcp/stream.rs，Tokio 1.40 里 TcpStream 的定义极简：

// 来源：tokio-rs/tokio · tokio/src/net/tcp/stream.rs:69 (tokio-1.40.0)
pub struct TcpStream {
    io: PollEvented<mio::net::TcpStream>,
}

一个字段、一层包装。从用户视角看 TcpStream 是一个能 await 的 I/O 类型；从内部看，它只是把这几层依赖组装起来：

TcpStream                                  ← 用户看到的 API 层
  │
  └─ PollEvented<mio::net::TcpStream>      ← Tokio 的 I/O 包装层
       │
       ├─ inner: mio::net::TcpStream        ← mio 的同步 TcpStream（非阻塞）
       └─ registration: Registration        ← 和 Tokio runtime 的连接
            │
            ├─ handle: scheduler::Handle    ← Runtime Handle
            └─ shared: Arc<ScheduledIo>     ← 第 8 章讲的 per-fd 状态
                 │
                 └─ ... 底层 mio 的 Token 指向这个 ScheduledIo 地址

4 个分层，每一层都在做一件明确的事：

TcpStream：用户 API（connect / AsyncRead / AsyncWrite / peer_addr 等），不含任何异步 / 调度逻辑
PollEvented<T>：通用 I/O 包装，接受任何 mio 事件源 T，管理它的 registration
Registration：ScheduledIo 的生命周期拥有者
ScheduledIo（第 8 章已拆）：readiness 状态机

这 4 层的分工是 Tokio 对 I/O 类型"模板化"的关键。

"为什么不合并层"的几个诱惑与回答

你可能想：既然每层都这么薄，为什么不合并？少一层不就更简单？

诱惑 1：把 PollEvented 去掉，TcpStream 直接持有 mio::TcpStream + Registration

回答：这样每个 I/O 类型都要写自己的"注册 + deregister + 生命周期管理"——每个类型重复一遍样板。PollEvented 存在就是为了消除这些重复

诱惑 2：把 Registration 和 ScheduledIo 合并

回答：Registration 是用户侧 + 可能被用户代码直接持有（通过 PollEvented），ScheduledIo 是运行时侧 + 被 I/O Driver 内部持有。两者生命周期不一致——ScheduledIo 可能被多个引用持有、Registration 是单一所有者

诱惑 3：直接让 TcpStream 继承 mio::TcpStream

回答：Rust 没有继承。即使用 Deref 也会让 mio 方法直接暴露给用户——破坏抽象（用户可能直接调 mio 的非阻塞 read，拿到 WouldBlock 却没有登记 Waker，任务就再也没人唤醒）

每一层都有存在的必要性。看起来冗余其实是在用分层换扩展性、换正确性、换代码复用。这种"一眼觉得多余、仔细看发现恰到好处"的设计，是成熟软件的标志。

只要你实现一个类型 T 并让它实现 mio 的 event::Source trait，你就能把它包装成 PollEvented<T>——自动获得完整的 async 能力。后面 TcpListener、UdpSocket、UnixStream、ChildStdout 全部走这一条路。

10.3 PollEvented：Tokio 的 I/O 通用包装

PollEvented<E> 是一个泛型——它可以包装 mio 的任何"有事件"类型：mio::TcpStream、mio::UdpSocket、mio::UnixStream 都能往里塞。这种泛型设计让 Tokio 避免了"每种 I/O 类型写一份响应式代码"的重复——所有 I/O 类型共享同一个 PollEvented 包装器、每个类型只需要提供"syscall 怎么调"的业务逻辑。这是 Rust 泛型的标志性应用——一次泛型定义、无数具体用例。

PollEvented 的签名（简化）（tokio/src/io/poll_evented.rs:66）：

// 来源：tokio/src/io/poll_evented.rs:66
pub(crate) struct PollEvented<E: Source> {
    io: Option<E>,
    registration: Registration,
}

两个字段：

io: Option<E> —— 包装的 mio 事件源。Option 的作用是支持从 PollEvented 里把 E 取出来（比如 TcpStream::into_std）。取出后 io = None，这个 PollEvented 不再能用
registration: Registration —— 运行时侧的注册句柄

PollEvented 的 new 方法（tokio/src/io/poll_evented.rs:89）做两件事：

调用 mio 的 register 把 io 注册到 Tokio 的 I/O Driver（第 8 章讲的 add_source 流程）
拿到返回的 Registration、和 io 一起打包成 PollEvented

典型用法：

// 伪代码：构造一个自定义的 async I/O 类型
impl MyAsyncType {
    fn from_mio(mio_io: MyMioType) -> io::Result<Self> {
        let io = PollEvented::new(mio_io)?;   // 一行完成注册
        Ok(MyAsyncType { io })
    }
    
    fn poll_read(&self, cx: &mut Context<'_>, buf: &mut [u8]) -> Poll<io::Result<usize>> {
        // 委托给内部 registration + mio_io 的 read
    }
}

一行 PollEvented::new 就做好了：

分配 ScheduledIo
把 mio_io 通过 mio::Registry::register 注册到 Tokio 的 Poll，token 设为 *const ScheduledIo
返回 PollEvented，其中 Registration 持有 Arc<ScheduledIo>

这层包装的设计价值：任何 I/O 类型"异步化"的成本被压到几行代码。没有这层，每个 I/O 类型都要重复写注册、生命周期管理、deregister 逻辑——几十行样板代码。PollEvented 把这些压缩到一个泛型结构里，所有 I/O 类型直接复用。

PollEvented 的 Drop 行为

当 PollEvented drop 时（tokio/src/io/poll_evented.rs:299 的 impl Drop），它会：

先 self.io.take() 取出 mio_io，从 Tokio I/O Driver deregister（调 mio::Registry::deregister）
Drop mio_io（关闭 fd）
字段析构时 Drop Registration，释放它那份 Arc<ScheduledIo>

整个生命周期管理在 Drop 里一气呵成——用户代码不需要手动清理。这是 RAII 最漂亮的应用：资源获取和释放完全对称、编译期保证。

"接受任何 mio::Source" 的扩展性

PollEvented<E: Source> 的 E 约束是 mio::event::Source——任何实现了这个 trait 的类型都可以被包装。这意味着：

Tokio 自带的 mio::net::TcpStream 等当然可以
第三方 crate（比如 mio-serial 之类给串口做 Source 实现的库）也可以
你自己写的实现了 Source 的类型也可以

从来没有人规定 PollEvented 只能装"socket 相关的东西"——任何 fd-based 或 mio 支持的事件源都行。这种"基于 trait 的泛型约束"让 Tokio 的 I/O 生态极易扩展。

mio 换成 io-uring 会怎样

前面提过 tokio-uring 用 io-uring 代替 mio。它的 I/O 类型架构上很像：

tokio_uring::TcpStream 包装一个 io-uring-based TcpStream
但它的中间层不是 PollEvented——因为 io-uring 是 completion 模型，不能走 readiness-based 的 PollEvented
内部有一个 Op<T> 类型 + submission queue 管理

所以 PollEvented 只适用于 readiness 模型。如果未来 Tokio 主干加入 io-uring 支持，多半也只能和 PollEvented 并存（两种 I/O 类型，两套后端）。兼容性和灵活性比统一模型更重要。

10.4 Registration：运行时侧的对接点

Registration 是 PollEvented 和 I/O Driver 之间的契约——PollEvented 持有一个 Registration、Registration 持有对 ScheduledIo 的引用。这种"单向引用"让 PollEvented drop 时能经由 Registration 正确注销到 I/O Driver、不会留下悬垂的 ScheduledIo。这种RAII（Resource Acquisition Is Initialization）模式在 Rust 里是标配——所有需要清理的资源都通过 Drop trait 自动清理、用户不用操心。Tokio 的每个 I/O 类型都通过 Registration 享受这个自动清理能力。

打开 tokio/src/runtime/io/registration.rs。Registration 的定义（tokio/src/runtime/io/registration.rs:46）：

// 来源：tokio/src/runtime/io/registration.rs:46
pub(crate) struct Registration {
    handle: scheduler::Handle,
    shared: Arc<ScheduledIo>,
}

两个字段：

handle：runtime Handle（指向 scheduler 和 I/O Driver）。deregister 时需要
shared：指向 ScheduledIo 的共享引用

注意是 Arc<ScheduledIo>——多方共享：

Registration 持有一份（runtime 活着时）
Tokio I/O Driver 的 RegistrationSet 持有一份（作为链表节点）
deregister 之后、I/O Driver 真正释放之前，RegistrationSet 的 pending_release 里还有一份（见下文 deregister 一节）

`new_with_interest_and_handle` 的真实流程

// 来源：tokio/src/runtime/io/registration.rs:73
pub(crate) fn new_with_interest_and_handle(
    io: &mut impl Source,
    interest: Interest,
    handle: scheduler::Handle,
) -> io::Result<Registration> {
    let shared = handle.driver().io().add_source(io, interest)?;
    Ok(Registration { handle, shared })
}

两行代码：

handle.driver().io().add_source(io, interest) —— 调用 I/O Driver 的 add_source 方法（第 8 章讲过，内部由 RegistrationSet::allocate 做 Arc::new(ScheduledIo::default()) 并挂进链表，再 mio::Registry::register）
打包返回

所有"复杂性"都下沉到 add_source——Registration 自己几乎什么都没做。这种"薄对接 + 下层厚逻辑"的分层让每一层单独看都简单。

poll_read_ready / poll_write_ready

Registration 对外提供的核心方法：

// 来源：tokio/src/runtime/io/registration.rs:109 与 tokio/src/runtime/io/registration.rs:115
pub(crate) fn poll_read_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<ReadyEvent>> {
    self.poll_ready(cx, Direction::Read)
}

pub(crate) fn poll_write_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<ReadyEvent>> {
    self.poll_ready(cx, Direction::Write)
}

两个方法都委托给 poll_ready（tokio/src/runtime/io/registration.rs:144），后者再委托给 ScheduledIo::poll_readiness（tokio/src/runtime/io/scheduled_io.rs:326，第 8 章讲过）。这条调用链很深但每一层都很薄。

deregister

// 来源：tokio/src/runtime/io/registration.rs:99
pub(crate) fn deregister(&mut self, io: &mut impl Source) -> io::Result<()> {
    self.handle().deregister_source(&self.shared, io)
}

Drop PollEvented 时调这个，把 fd 从 mio unregister。

所有复杂性都在更深的层——handle.deregister_source 会先调 mio::Registry::deregister，再把这份 Arc<ScheduledIo> 放进 RegistrationSet 的 pending_release，等 I/O Driver 下一次 turn 时才从链表里移除（攒满 16 个会主动 unpark Driver）——这样保证 ScheduledIo 不会在它还可能出现在某个 I/O 事件里时被释放（tokio/src/runtime/io/driver.rs:242-260，pending_release 的入队在 tokio/src/runtime/io/registration_set.rs:75）。

`Arc<ScheduledIo>` 为什么用 Arc 而不是 Box

这是一个微妙的设计决策。ScheduledIo 是不是单一所有者？似乎是——每个 Registration 持有一个。那为什么用 Arc<ScheduledIo> 而不是 Box<ScheduledIo>？

因为至少有两方共同需要访问 ScheduledIo：

Registration（用户侧）：创建时持有 Arc，drop 时释放一份
Tokio I/O Driver 的 RegistrationSet（内部）：为了能遍历所有活跃资源、批量 shutdown，Driver 需要持有所有活跃 ScheduledIo 的引用

需要多份引用 → 必须用 Arc（或者引用计数的等价物）。如果用 Box，只有一方持有，另一方就拿不到。

有没有不用 Arc 的方案？理论上可以——比如用 &'static ScheduledIo（所有 ScheduledIo 永不 drop）。但这会导致内存泄漏（Tokio 常驻程序 spawn 百万次 fd 最终会 OOM）。或者用 Weak<ScheduledIo> 让 Driver 只持弱引用——但弱引用每次 upgrade 都是原子操作，没便宜多少。

Arc 是最简单可行的方案。Tokio 作者选择了简单，不追求"极致零成本"。这也是一个提醒：零成本抽象是个目标，不是铁律——有时候多一次原子操作换来代码清晰是值得的。

10.5 `TcpStream::connect`：一个 async fn 的完整拆解

TcpStream::connect(addr).await 看起来就是一行代码——但它背后要完成的事非常多：DNS 解析（如果 addr 是域名）、创建 socket、设置非阻塞、调 connect()、等 connect 完成（就是要等 writable 事件）、可能的重试。把这一行代码的内部拆成几十步看下来、你会对"async fn 能把多么复杂的流程隐藏成多么简洁的 API"产生直观感受——这就是 Rust 异步相对回调式 async（Node.js 早期、老式 C 库）的根本优势。

原样代码：

// 来源：tokio/src/net/tcp/stream.rs:114
pub async fn connect<A: ToSocketAddrs>(addr: A) -> io::Result<TcpStream> {
    let addrs = to_socket_addrs(addr).await?;

    let mut last_err = None;

    for addr in addrs {
        match TcpStream::connect_addr(addr).await {
            Ok(stream) => return Ok(stream),
            Err(e) => last_err = Some(e),
        }
    }

    Err(last_err.unwrap_or_else(|| {
        io::Error::new(
            io::ErrorKind::InvalidInput,
            "could not resolve to any address",
        )
    }))
}

逻辑直白：

DNS 解析：to_socket_addrs(addr).await? —— addr 可能是 "example.com:80" 这种字符串，DNS 解析成一组 SocketAddr（迭代器）
依次尝试每个地址：DNS 可能返回多个 IP（IPv4 + IPv6、或多个 A 记录），依次尝试
第一个成功就返回；全部失败 → 返回最后一个错误

为什么要 DNS async？因为 DNS 本身是网络 I/O，不能阻塞。但 Tokio 并没有自带异步 DNS 解析器——tokio/src/net/addr.rs:182、tokio/src/net/addr.rs:219 里对 &str / (&str, u16) 这类需要真解析的输入，一律是 spawn_blocking(move || std::net::ToSocketAddrs::to_socket_addrs(&s))，即把 libc 的同步 getaddrinfo 丢进阻塞线程池；而 SocketAddr / (IpAddr, u16) 这类无需解析的输入直接 Ready 返回——字符串本身就是 IP 字面量（如 "127.0.0.1:8080"）时，也会先 parse 成功、同样直接 Ready（tokio/src/net/addr.rs:172-177）。要真正的异步 DNS 得换 hickory-resolver 之类的第三方 crate。

`connect_addr` 的内部

connect_addr 是真正做 TCP 三次握手的地方。它大致长这样（简化）：

// 来源：tokio/src/net/tcp/stream.rs:135 与 tokio/src/net/tcp/stream.rs:140（注释精简）
async fn connect_addr(addr: SocketAddr) -> io::Result<TcpStream> {
    let sys = mio::net::TcpStream::connect(addr)?;   // 非阻塞 connect
    TcpStream::connect_mio(sys).await
}

pub(crate) async fn connect_mio(sys: mio::net::TcpStream) -> io::Result<TcpStream> {
    let stream = TcpStream::new(sys)?;

    // 连接是否真的建立，要等 fd 变 writable 才有结论
    poll_fn(|cx| stream.io.registration().poll_write_ready(cx)).await?;

    // writable 之后用 SO_ERROR 判断 connect 到底成没成
    if let Some(e) = stream.io.take_error()? {
        return Err(e);
    }

    Ok(stream)
}

注意这里的 .await：它不是一次 await，是两次：

第一次：DNS 解析（走 spawn_blocking 里的 getaddrinfo；SocketAddr 或 IP 字面量输入则立刻 Ready）
第二次：等 TCP 三次握手完成（fd 变 writable）

每次 await 都可能让出当前 Task、让出给 runtime 跑其他事。一个看似简单的 TcpStream::connect("example.com:80").await 内部是一个异步状态机，经过几次 yield。

`async_io` helper：Tokio 最重要的 I/O helper

connect_mio 用的是最朴素的 poll_write_ready —— connect 只需要等一次 writable、没有重试的必要。但 Tokio I/O 里更常见的形态是下面这个 helper：

// 来源：tokio/src/net/tcp/listener.rs:164
self.io.registration()
    .async_io(Interest::READABLE, || self.io.accept())
    .await?

async_io(interest, op) 的语义：

尝试同步执行 op。如果 op 返回 WouldBlock，注册 Waker 等 interest 就绪、就绪后再试。循环直到 op 返回非 WouldBlock 结果。

它是一个通用模式（tokio/src/runtime/io/registration.rs:214），封装了"try → poll_ready → retry" 这个三步循环。TcpListener::accept 也用它、UdpSocket::recv 也用它。下一节详细讲。

DNS 解析的边角：`ToSocketAddrs` 的 async 设计

to_socket_addrs 的signature 设计值得看一下：

// 来源：tokio/src/net/addr.rs:19
pub trait ToSocketAddrs: sealed::ToSocketAddrsPriv {}

Tokio 用了一个 sealed trait pattern——ToSocketAddrs 是公开的空 trait，但实际解析逻辑在 ToSocketAddrsPriv（tokio/src/net/addr.rs:262）里，它住在 private 的 sealed 模块（tokio/src/net/addr.rs:252）里。

为什么这么设计：

让 impl ToSocketAddrs for String / &str / SocketAddr / (IpAddr, u16) 等公开的
但不允许用户自定义实现（因为 sealed）——所有解析逻辑都走 Tokio 控制的代码路径
未来 Tokio 可以改 ToSocketAddrsPriv 的签名（比如加参数）——不破坏公开 API

sealed trait pattern 是 Rust 库 API 设计的常见技巧。它让你获得 trait 的好处（开放多态）同时保持内部改造自由。当你写 Rust 库时值得借鉴——尤其是"看起来像 enum 但需要扩展性"的场景。

`connect_addr` 内部的"两次 await"

重新看 connect_addr 的伪代码：

// tokio/src/net/tcp/stream.rs:135
async fn connect_addr(addr: SocketAddr) -> io::Result<TcpStream> {
    let sys = mio::net::TcpStream::connect(addr)?;   // 非阻塞 connect 系统调用
    TcpStream::connect_mio(sys).await
}

pub(crate) async fn connect_mio(sys: mio::net::TcpStream) -> io::Result<TcpStream> {
    let stream = TcpStream::new(sys)?;
    poll_fn(|cx| stream.io.registration().poll_write_ready(cx)).await?;
    if let Some(e) = stream.io.take_error()? {
        return Err(e);
    }
    Ok(stream)
}

注意这里：

mio::net::TcpStream::connect 是同步调用——但是它发的是非阻塞 connect（Linux 上底层是 socket(SOCK_NONBLOCK) + connect()）
connect() 系统调用通常立刻返回 EINPROGRESS（连接进行中）——不等三次握手完成
然后 Tokio 等 fd 变 writable（POSIX connect(2) 的约定——non-blocking connect 完成的信号是 fd 可写，不是 Linux 独有）
writable 后调 take_error()（底层是 getsockopt(SO_ERROR)）——connect 失败的真实原因就藏在这里；成功则返回 None

这是 Unix 系统编程的经典"异步 connect"模式。Tokio 忠实地遵循了它。

10.6 `AsyncRead` / `AsyncWrite` impl：标准的 try + delegate

这一节的代码有高度模板化的特征——poll_read / poll_write / poll_flush / poll_shutdown 几乎都遵循同一个模式：检查 readiness → 如果就绪调 syscall → syscall 可能返回 WouldBlock → 清掉 readiness 标记返 Pending。这个模式在所有 Tokio I/O 类型里都一样——模板化的好处是写新 I/O 类型有现成套路可抄、代价是读代码容易觉得"都长得像"。一旦你掌握了这个模式、新 I/O 类型读起来会非常快。

原样：

// 来源：tokio/src/net/tcp/stream.rs:1325
impl AsyncRead for TcpStream {
    fn poll_read(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<io::Result<()>> {
        self.poll_read_priv(cx, buf)
    }
}

impl AsyncWrite for TcpStream {
    fn poll_write(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &[u8],
    ) -> Poll<io::Result<usize>> {
        self.poll_write_priv(cx, buf)
    }

    fn poll_write_vectored(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        bufs: &[io::IoSlice<'_>],
    ) -> Poll<io::Result<usize>> {
        self.poll_write_vectored_priv(cx, bufs)
    }

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

    #[inline]
    fn poll_flush(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<()>> {
        // tcp flush is a no-op
        Poll::Ready(Ok(()))
    }

    fn poll_shutdown(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<()>> {
        self.shutdown_std(std::net::Shutdown::Write)?;
        Poll::Ready(Ok(()))
    }
}

几个有趣细节：

AsyncRead / AsyncWrite 的 trait 本身就是公开 API 的形状——用户看到的"TcpStream 可以 .await read/write"就是 impl 这两个 trait 的结果
poll_flush 是 no-op（tokio/src/net/tcp/stream.rs:1357）——TCP 不需要用户层 flush（内核已经在 kernel buffer 里管理写缓冲），直接返回 Ready(Ok)
poll_shutdown（tokio/src/net/tcp/stream.rs:1362）调 shutdown_std(Shutdown::Write)——对应 POSIX shutdown(2) 的写方向关闭。这个是同步操作（不需要 await），因为只是告诉内核"我不再写了"、不涉及真实 I/O
poll_write_vectored 支持（impl AsyncWrite 在 tokio/src/net/tcp/stream.rs:1335）——TcpStream 支持用多个 IoSlice 一次写（对应 Linux 的 writev(2)），HTTP/2 这类"帧头和帧体分处两块 buffer"的协议因此可以少一次系统调用、也少一次拷贝拼接。注意配套的 is_write_vectored 返回 true——上层（如 hyper）就是靠这个标志决定要不要走 vectored 路径的

`poll_read_priv` 的真实实现

// 来源：tokio/src/net/tcp/stream.rs:1285
pub(crate) fn poll_read_priv(
    &self,
    cx: &mut Context<'_>,
    buf: &mut ReadBuf<'_>,
) -> Poll<io::Result<()>> {
    unsafe { self.io.poll_read(cx, buf) }
}

一行！委托给 PollEvented::poll_read。这个方法的实现（概念上）：

// 简化自 PollEvented 的 poll_read 逻辑
pub fn poll_read(&self, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<io::Result<()>> {
    loop {
        // 1. 先 poll readiness（第 8 章讲过的 poll_read_ready）
        let event = ready!(self.registration.poll_read_ready(cx))?;
        
        // 2. 读出 ready 状态后尝试真正 read
        match self.io.as_ref().unwrap().read(buf.initialize_unfilled()) {
            Ok(n) => {
                buf.advance(n);
                return Poll::Ready(Ok(()));
            }
            Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
                // 3. ready 显示可读但真读 WouldBlock：这是 epoll ET 的常见情况
                self.registration.clear_readiness(event);
                continue;   // 清除 readiness 标记，循环回去等下一次 wake
            }
            Err(e) => return Poll::Ready(Err(e)),
        }
    }
}

核心逻辑（9 章讲过 ET 模式的必要处理）：

先 poll_read_ready——如果 readiness 没到，注册 Waker、返回 Pending
readiness 到了，真 read——可能成功、可能 WouldBlock、可能真错误
WouldBlock → clear_readiness + loop——因为 epoll ET 模式下 readiness 位可能"过时"（内核已经知道 fd 不可读了，但 Tokio 的 ScheduledIo 还标着 READABLE）。clear 后再 loop 等下一次真正的 wake

但"只有读到 WouldBlock 才清 readiness"并不是全部——真实实现里还有第二个清除点：短读也清。tokio/src/io/poll_evented.rs:218-220 在 Ok(n) 分支里有 if 0 < n && n < len { self.registration.clear_readiness(evt); }——读到的字节数少于 buffer 容量，就说明内核的 socket 缓冲已经被抽干了，不必再多绕一次 syscall 去撞 WouldBlock。这个优化只在 epoll / kqueue 这类边缘触发选择器上成立，所以源码用 cfg 把它限制在 Linux / *BSD / macOS 等平台（Windows 和强制走 poll 的配置是水平触发，短读并不代表已抽干）。poll_write 有对称的一条：tokio/src/io/poll_evented.rs:251-252，部分写入同样说明发送缓冲已满，直接清 readiness——不过它的条件写得更宽，是 !cfg!(windows) && !cfg!(mio_unsupported_force_poll_poll)，只排除 Windows 和强制走 poll 的配置。所以准确的表述是：WouldBlock 一定清；短读在 epoll / kqueue 平台上也清，短写在非 Windows、未强制 poll 的平台上也清。

这个循环是 ET 模式下的正确姿势——不循环会导致偶发的"误以为可读"bug。Tokio 的 poll_read 内置这个循环，用户代码不需要知道。

`unsafe { self.io.poll_read(cx, buf) }` 里的 unsafe

注意 poll_read_priv 里那个 unsafe —— 为什么需要？

pub(crate) fn poll_read_priv(
    &self,
    cx: &mut Context<'_>,
    buf: &mut ReadBuf<'_>,
) -> Poll<io::Result<()>> {
    unsafe { self.io.poll_read(cx, buf) }
}

原因在 PollEvented::poll_read 的签名——它是 unsafe fn，要求调用者保证 read 实现正确处理未初始化内存（ReadBuf 里 uninit 的部分不能被读作 u8）。

TcpStream 的 read 保证满足这个条件——它调底层的 recv 系统调用，内核只写入实际收到的字节数。所以 unsafe block 是安全的。注释在源码里明确写着：

// Safety: `TcpStream::read` correctly handles reads into uninitialized memory
unsafe { self.io.poll_read(cx, buf) }

这是 Rust unsafe 的正确使用姿势——有文档 + 有不变式 + 有局部化。不是到处撒 unsafe 图省事，而是在必要的边界上带注释地使用。

10.7 `async_io` helper：自定义 I/O 的万能钥匙

async_io 是 Tokio 里一个极其有用但鲜为人知的 helper——它把"就绪检查 + syscall + Pending / Ready 处理"封装成一个复用函数。用它写自定义 I/O 类型只需要提供"syscall 本身"的代码、其他全部交给 async_io 处理。这是 Rust 写自定义 async I/O 的"万能钥匙"——不用它你就要重复写前面 10.6 讲的那套模板代码、用了它大部分重复工作都省掉。值得把它作为"写自定义 async I/O 库"的起点工具。

async_io 在 Registration 上是 pub(crate) 的，同时 TcpStream / UdpSocket / UnixStream / UnixDatagram 以及 Windows 的 NamedPipe 都把它转发成了公开方法（TcpListener 没有）（TcpStream::async_io 在 tokio/src/net/tcp/stream.rs:1047；由 PR #5512 加入，随 tokio 1.27.0 发布）。签名一模一样：

// 来源：tokio/src/runtime/io/registration.rs:214
pub(crate) async fn async_io<R>(
    &self,
    interest: Interest,
    mut f: impl FnMut() -> io::Result<R>,
) -> io::Result<R>

语义：反复尝试 f()，WouldBlock 时等 interest 就绪，直到 f() 成功或返回非 WouldBlock 错误。

伪代码实现：

// 示意，非逐字：tokio/src/runtime/io/registration.rs:214（省略 coop 记账）
pub(crate) async fn async_io<R>(
    &self,
    interest: Interest,
    mut f: impl FnMut() -> io::Result<R>,
) -> io::Result<R> {
    loop {
        let event = self.readiness(interest).await?;

        match f() {
            Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
                self.clear_readiness(event);
            }
            x => return x,
        }
    }
}

真实实现里在 f() 前后还夹了一次 coop::poll_proceed / coop.made_progress()——这是第 5 章讲过的协作式预算记账，防止一个一直有数据可读的 socket 把 worker 饿住。

`async_io` 用 FnMut 而不是 Fn

细心的读者会注意到 async_io 接受的 f 是 FnMut——可变闭包。为什么？

因为 f 在循环里可能被多次调用（每次 ready 就尝试一次），调用者可能需要在 f 之间保留状态：

// 用公开的 TcpStream::async_io；stream: TcpStream，buf: [u8; N]
let mut retry_count = 0;
let n = stream.async_io(Interest::READABLE, || {
    retry_count += 1;  // 这里需要 FnMut 才能可变借用
    stream.try_read(&mut buf)
}).await?;

用 Fn 的话这种 retry counter 就无法实现。Tokio API 设计非常精细——这类 FnMut vs Fn 的选择看似细节，但决定了用户代码的灵活性。

async_io 不是零成本：为什么？

async_io 是一个 async 函数，它会被 Rust 编译器展开成一个 Future state machine。这个状态机有大小——它至少要装：

自身的 f（闭包）
内部的 ready() Future
当前循环的状态

对于 FnMut 闭包，它可能捕获很多变量——状态机也跟着大。tokio::spawn 的 BOX_FUTURE_THRESHOLD（第 4 章讲过）就是为这种情况准备的——debug 构建下超过 2048 字节的 Future 自动 Box::pin（tokio/src/task/spawn.rs:172）。

实践影响：如果你在 hot path 上用 async_io + 大闭包，可能偶尔看到"状态机 > 2 KB、debug 下 boxed"的现象。不常见但存在——第 19 章（性能调优）会更详细讨论。

async_io 的 Cancel Safety

一个少被讨论但重要的特性：async_io 是 cancel-safe 的。意思是：

tokio::select! {
    r = stream.readable() => { /* ... */ }   // 内部走 Registration::readiness，与 async_io 同一套等待机制
    _ = timeout_future => { /* ... */ }      // 超时分支
}

如果 timeout 先 Ready，stream.readable() 会被 drop——但 drop 是安全的：

async_io 内部等的 Waker 会被从 ScheduledIo 的 waiters 里自动移除（Readiness 的 Drop，tokio/src/runtime/io/scheduled_io.rs:584-595）
已经注册但没触发的 readiness 不会丢失
下次你再调 readable() 能看到该看到的 readiness

这是 select! 能安全工作的前提——accept、readable、recv_from 这类方法的文档里都逐一标注了"This method is cancel safe"；但并非所有 I/O 方法都如此，read_exact、write_all 这类组合方法就不是。第 14 章会详细讲 cancel safety——在"drop 即取消"的 Rust 模型里，这是每个会被 select! 丢弃的 Future 都要回答的问题。

自己写 async I/O 方法时：如果用 async_io 包装，自动继承 cancel safety；如果你自己写 state machine，得自己保证。

用 `async_io` 写 TcpListener::accept

TcpListener::accept 的真实代码（原样）：

// 来源：tokio/src/net/tcp/listener.rs:160
pub async fn accept(&self) -> io::Result<(TcpStream, SocketAddr)> {
    let (mio, addr) = self
        .io
        .registration()
        .async_io(Interest::READABLE, || self.io.accept())
        .await?;

    let stream = TcpStream::new(mio)?;
    Ok((stream, addr))
}

漂亮——一行 async_io 封装了全部"等 readable + 非阻塞 accept + WouldBlock 重试"的循环。

为什么 async_io 是如此重要的 helper

async_io 这个 helper 最早由 2020 年 9 月的 PR #2828（随 tokio 0.3.0 发布）以 PollEvented::async_io 的形式引入，2020 年 11 月的 #3124（tokio 0.3.4）在 I/O driver 内部清理时把它挪到了 Registration 上；在 #2828 之前每个 I/O 类型的 accept / recv / connect 都手写 try + register + loop——十几行样板、容易出错（忘记 clear_readiness 就是 bug）。把它开放成公开方法（TcpStream::async_io 等）则要晚得多：PR #5512 在 tokio 1.27.0 才给各 socket 类型加上，AsyncFd::async_io 又在 #5542（tokio 1.28.0）跟进。另外 #6221 后来给 async_io 补上了协作式预算记账。

async_io 把这个模式封装成一个 helper——现在 Tokio I/O 类型里 async 版本的 accept / recv 等大多是一行调 async_io。源码简洁、bug 大减、性能一致。

这种"提炼共同模式为 helper"是库维护者的日常艺术。你自己写 Rust async 库时，提前想"这个模式会出现多少次"——出现 3 次以上就值得抽成 helper。

`poll_flush` 为什么是 no-op

这行值得单独讲：

fn poll_flush(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<()>> {
    // tcp flush is a no-op
    Poll::Ready(Ok(()))
}

为什么 TCP 的 flush 是 no-op？

因为 TCP 不保证"立刻发送"。poll_write 成功意味着数据已经交给内核的 socket buffer——内核决定什么时候真的发出去（Nagle 算法、TSO、ACK 合并等）。用户层的 flush 对 TCP 毫无意义——你的数据已经被内核接管了，内核不会因为你 flush 就改变发送策略。

这和 BufWriter / File 不同——BufWriter 有用户态的缓冲需要 flush 到 File，File 也可能有内核缓冲需要 fsync。TCP 没有用户态缓冲、内核缓冲也不由你控制。

如果真想"立即发"，用 TCP_NODELAY（禁用 Nagle）+ TcpStream::set_nodelay(true)。这改变内核发送策略，效果比 flush 直接多了。

`poll_shutdown` 关闭的是写方向

fn poll_shutdown(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<()>> {
    self.shutdown_std(std::net::Shutdown::Write)?;
    Poll::Ready(Ok(()))
}

注意是 Shutdown::Write——只关闭写方向。为什么不关双向？

因为 AsyncWrite trait 的语义是"writer 关闭"——它不管 reader。如果双向关闭，正在 read 的代码会意外得到 EOF。Tokio 遵循了这个约定：半关闭写方向 = 告诉对端"我不再写了"；但 read 方向仍然能继续。

对应 TCP 语义：半关闭会让对端 read 拿到 0 字节（EOF），但你仍可以 read 对端还没发完的数据。

为什么 poll_write_vectored 支持但 poll_read_vectored 没有

细节细到这里：AsyncWrite 的 impl 里有 poll_write_vectored——接受 &[IoSlice] 一次写多个 buffer。对应 Linux writev(2)，一次系统调用写多段。

但 AsyncRead 的 impl 里没有 poll_read_vectored（Tokio 的 TcpStream 实现里）。为什么？

原因：Tokio 的 AsyncRead trait 接受的是 ReadBuf<'_>——一个连续 buffer，不是 IoSlice 数组。这是 Tokio 故意的 API 简化：

写入场景真实需要 vectored——HTTP 帧的 header + body 分开、WebSocket 掩码 + payload 分开
读取场景绝大多数时候是"读到一个 buffer"——应用层分析之后再切分。Vectored read 的实际需求不高

trade-off：API 更简单（一个 ReadBuf vs 多个 IoSlice），代价是少数场景需要多次 read。这是 Tokio 的取舍——对绝大多数用户更友好。

如果你真需要 vectored read：TcpStream 提供了 try_read_vectored（tokio/src/net/tcp/stream.rs:691），配合 readable().await 使用，绕过 AsyncRead 抽象。但你知道你在做什么才这么干。

TcpStream 的 `set_nodelay` 及其他 setter

TcpStream 还暴露了一些同步的配置方法：

set_nodelay(true) —— 禁用 Nagle 算法（低延迟优先）
set_ttl(u32) —— 设置 IP TTL
linger() / set_linger(Option<Duration>) —— SO_LINGER 控制 close 行为
peer_addr() / local_addr() —— 查询本端 / 对端地址

这些都是同步方法，不带 async——因为底层 setsockopt(2) 是非阻塞的、立即返回。不是每个方法都需要 async——只有涉及 I/O 等待的才需要。

Tokio 在标记 async 这件事上很有纪律——有 I/O 可能性的标 async、纯查询 / 配置的不标。这让用户代码清晰知道"哪里可能让步、哪里不会"。对比一些库给所有方法都加 async（哪怕内部是同步操作）——那让用户无法判断"这一行是不是阻塞点"。

10.8 UdpSocket：和 TcpStream 的同构对比

把 UdpSocket 和 TcpStream 放一起对比读非常有教育意义——两者结构几乎一样，只在 syscall 层面有差异（TCP 用 read/write、UDP 用 recvfrom/sendto）。这种"高度同构"证明 Tokio 的 I/O 抽象层设计得相当成功——核心复杂度在 PollEvented / Registration 里处理好了、新增一种 socket 类型只需要写薄薄的一层业务代码。

打开 tokio/src/net/udp.rs，UdpSocket 的定义（tokio/src/net/udp.rs:121）：

// 来源：tokio/src/net/udp.rs:121
pub struct UdpSocket {
    io: PollEvented<mio::net::UdpSocket>,
}

和 TcpStream 一模一样的结构——只是泛型参数从 mio::net::TcpStream 换成 mio::net::UdpSocket。

主要差异在 API 层：

不实现 AsyncRead / AsyncWrite——因为 UDP 是 datagram-based，AsyncRead 的 stream 语义不适用
提供 send_to(buf, addr) / recv_from(buf) 方法——each call 发 / 收一个完整数据包
内部仍然用 async_io 实现异步：

// 来源：tokio/src/net/udp.rs:1306
pub async fn recv_from(&self, buf: &mut [u8]) -> io::Result<(usize, SocketAddr)> {
    self.io
        .registration()
        .async_io(Interest::READABLE, || self.io.recv_from(buf))
        .await
}

又是一行 async_io。

TcpStream vs UdpSocket：架构层的相同，应用层的不同

维度	TcpStream	UdpSocket
底层 mio 类型	`mio::net::TcpStream`	`mio::net::UdpSocket`
Tokio 包装	`PollEvented`	`PollEvented`
Registration 层	完全相同	完全相同
async_io helper	相同	相同
API 风格	AsyncRead + AsyncWrite	send_to / recv_from
读 vs 写的 Waker 注册	分开（readable / writable）	分开

Tokio 所有基于 mio 的 I/O 类型（TcpStream / TcpListener / UdpSocket / UnixStream / UnixListener / ChildStdin/Out/Err 等）都是这个套路——四层包装 + async_io。学会其中一个，其他都秒懂。

`send_to` 一次成功 vs `recv_from` 需要 retry

UDP 和 TCP 的一个语义差异：

send_to：UDP datagram 是独立的包——要么整个成功要么整个失败。send_to 要么 Ready 要么 WouldBlock，不存在"发了一半"
recv_from：UDP 也是整个 packet 读。如果你传的 buffer 比 packet 小，多余字节会被丢弃（而不是等下次 read）

这些 UDP 特有语义 Tokio 忠实地暴露——send_to / recv_from 返回 Result<usize, Error> + SocketAddr，对应 packet size 和来源地址。不像 TCP 有 stream 语义需要你拼接 buffer、UDP 是离散 message。

一个常见的 UDP 陷阱：buffer 太小截断

let mut buf = [0u8; 512];   // 只有 512 字节
let (n, addr) = socket.recv_from(&mut buf).await?;
// 如果对端发了 1500 字节，n = 512，**剩余 988 字节被悄悄丢弃**

没有任何报错——你只是拿到一个"截断的"packet。对应 Linux recvfrom(2) 的默认行为。

修复：用足够大的 buffer（IPv4 上 UDP 载荷最多 65507 字节，常见以太网 MTU 是 1500——确定对端报文不超过 MTU 时取 2048 即可，否则取 65536）。注意 peek_from 只是带 MSG_PEEK 的 recvfrom，buffer 小了一样会截断，拿不到报文真实长度（Linux 上要 MSG_PEEK | MSG_TRUNC 才行，Tokio 没有直接封装）。

这类 UDP 陷阱在 Tokio 层面不特殊——UDP 本身就这样、Tokio 保持透明。你写 UDP 代码时得按 UDP 语义来，不能用 TCP 直觉套。

其他 I/O 类型快览

除了 TcpStream / TcpListener / UdpSocket，Tokio 还包装了：

tokio::net::UnixStream / UnixListener：Unix domain socket

相同架构：UnixStream { io: PollEvented<mio::net::UnixStream> }（tokio/src/net/unix/stream.rs:42）
AsyncRead / AsyncWrite 实现和 TcpStream 几乎一样
差别只在 mio 底层（Unix socket 不需要 IP 地址）

tokio::net::UnixDatagram：UDP-like Unix socket

对应 UdpSocket，只是底层是 Unix socket
方法同名（send_to / recv_from），但地址参数换成了路径，返回的是 Unix 域的 SocketAddr

tokio::process::ChildStdin / ChildStdout / ChildStderr：子进程的 stdio 管道异步化

底层是 pipe fd
Tokio 给这些 fd 做了 PollEvented 包装
AsyncRead / AsyncWrite 可以直接用

tokio::signal：异步信号处理

底层不是 signalfd——tokio/src/signal/unix.rs:276 用 signal_hook_registry::register 装一个真正的信号处理函数（action，tokio/src/signal/unix.rs:242），处理函数里往一对 mio::net::UnixStream::pair() 的发送端写一个字节（自管道技巧，只是用 socketpair 代替了 pipe）
接收端才是被 Tokio 注册进 I/O Driver 的 fd（tokio/src/runtime/io/driver.rs:81 的 TOKEN_SIGNAL），信号因此被翻译成一次普通的可读事件

net 和 process 下这些类型的 I/O 核心都只是薄薄一层——复用 PollEvented，async 方法大多再配 async_io；signal 则不走 PollEvented，而是靠 I/O Driver 里专门保留的 TOKEN_SIGNAL。

PollEvented 的泛型约束

PollEvented<E: Source> 里的 Source trait 是 mio 提供的（就是第 9 章讲的 event::Source）。任何实现它的类型都能被 PollEvented 包装。

这个泛型约束让 Tokio 的抽象对外开放——你自己实现 Source 的类型也能享受 PollEvented 的全部功能。Tokio 没有把 I/O 类型硬编码成 "只有 TcpStream / UdpSocket 能异步"——任何符合接口的都行。

这种"关键抽象对外开放"是好库的标志。对比一些闭门造车的库——API 只支持库作者预设的几种类型，扩展要 fork 重写。Tokio 走的是开放路径。

10.9 一次 `TcpStream::read.await` 的全局回看

这是前 10 章所有知识的最终整合章节——一次 TcpStream::read.await 从用户代码开始、经过 Tokio API、PollEvented、I/O Driver、mio、epoll、OS 内核、硬件网卡、再原路返回——跨越 7 个抽象层、几十个源码位置。把这整条路径画到纸上、你会看到前面学过的所有概念（Future / Waker / Task / Scheduler / I/O Driver / mio / epoll）各自在哪一步发挥作用。读完这一节你对 Tokio 的理解就从"一堆孤立概念"升级到"完整心智模型"。

把第 1-10 章的所有组件串起来，看一次 let n = stream.read(&mut buf).await? 的完整路径：

Step 1（用户代码）：read 返回 Read<'a> Future，用户 .await 它

Step 2（Future trait）：Read 的 poll 调 stream.poll_read(cx, buf)

Step 3（TcpStream）：<TcpStream as AsyncRead>::poll_read（tokio-1.40.0/tokio/src/net/tcp/stream.rs:1326）→ poll_read_priv（tokio-1.40.0/tokio/src/net/tcp/stream.rs:1285）→ unsafe { self.io.poll_read(cx, buf) }，这里的 self.io 是 PollEvented<mio::net::TcpStream>

Step 4（PollEvented）：PollEvented::poll_read 是个 unsafe fn（tokio-1.40.0/tokio/src/io/poll_evented.rs:169），循环体第一句就是：

4.1 self.registration.poll_read_ready(cx)（tokio-1.40.0/tokio/src/runtime/io/registration.rs:109）→ poll_ready(cx, Direction::Read) → self.shared.poll_readiness(cx, direction)，读 ScheduledIo.readiness 这个 AtomicUsize（tokio-1.40.0/tokio/src/runtime/io/scheduled_io.rs:326-333）
4.2 没就绪 → 拿 waiters Mutex，把 Waker 放进 waiters.reader（槽里已有 Waker 且 will_wake 为真就什么都不做，否则用 clone_from 覆盖；槽为空才 clone 一份）→ 再读一次 readiness 做双重检查 → 返回 Pending（tokio-1.40.0/tokio/src/runtime/io/scheduled_io.rs:336-370）

Step 5（Task 挂起）：

5.1 Read Future 返回 Pending 向上冒泡
5.2 外层 async fn 也返回 Pending
5.3 Task 的 state 从 RUNNING 变 Idle（第 6 章）

Step 6（Scheduler park）：

6.1 worker 的本地队列没其他任务 → 准备 park
6.2 worker 调 Driver::park(&mut self, rt_handle: &driver::Handle) → 内部 self.turn(handle, None)（tokio-1.40.0/tokio/src/runtime/io/driver.rs:120-123；turn 的真实签名是 fn turn(&mut self, handle: &Handle, max_wait: Option<Duration>)，见 tokio-1.40.0/tokio/src/runtime/io/driver.rs:140，带超时的那条路走 park_timeout → turn(handle, Some(duration))）

Step 7（mio → 内核）：

7.1 turn 调 self.poll.poll(events, max_wait)，即 mio::Poll::poll(events, None)（tokio-1.40.0/tokio/src/runtime/io/driver.rs:149）
7.2 Linux 下就是 epoll_wait(epfd, events, 1024, -1) 系统调用（1024 是 Builder 的 nevents 默认值，见 tokio-1.40.0/tokio/src/runtime/builder.rs:277）
7.3 线程阻塞在内核——CPU 让出给其他进程

Step 8（内核侧数据到达）：

8.1 远端 TCP 包到达网卡 → 内核 TCP/IP 栈处理
8.2 数据放进 socket receive buffer
8.3 内核把这个 fd 标记为 READY 在 epoll 的就绪列表里
8.4 epoll_wait 返回

Step 9（mio → Tokio I/O Driver）：

9.1 epoll_wait 返回 N 个事件
9.2 Tokio turn 的 event loop：for event in events.iter()（tokio-1.40.0/tokio/src/runtime/io/driver.rs:161）；TOKEN_WAKEUP / TOKEN_SIGNAL 两个保留 token 先被筛掉
9.3 对 matching event：let ptr: *const ScheduledIo = token.0 as *const _（tokio-1.40.0/tokio/src/runtime/io/driver.rs:172）
9.4 io.set_readiness(Tick::Set, |curr| curr | ready)（tokio-1.40.0/tokio/src/runtime/io/driver.rs:180，闭包参数是这次事件算出来的 ready，不是写死的 READABLE）
9.5 io.wake(ready) → 进入 ScheduledIo::wake（tokio-1.40.0/tokio/src/runtime/io/driver.rs:181，实现在 tokio-1.40.0/tokio/src/runtime/io/scheduled_io.rs:259）

Step 10（Tokio 内部 wake）：

10.1 io.wake 先 let mut wakers = WakeList::new()，再拿 waiters Mutex 锁 → waiters.reader.take() 取走 Waker（tokio-1.40.0/tokio/src/runtime/io/scheduled_io.rs:259-280）
10.2 drop 锁 → wakers.wake_all()（tokio-1.40.0/tokio/src/runtime/io/scheduled_io.rs:297-308；waiter 多时会分批：放锁 → 唤醒一批 → 重新上锁）
10.3 进入第 3 章讲的 6 层 waker 路径
10.4 RawTask::wake_by_val（定义在 harness.rs 的 impl RawTask 里）→ transition_to_notified_by_val → RawTask::schedule → scheduler.schedule

Step 11（Scheduler 重调度）：

11.1 Task 被推回本地队列或 LIFO slot
11.2 某个 worker pop 到 Task → Harness::poll
11.3 transition_to_running → state 变 RUNNING

Step 12（Task 重新 poll）：

12.1 Task 的 poll 重新进入 async fn 状态机
12.2 恢复到 Read Future 的 .await 点
12.3 再次 stream.poll_read(cx, buf)
12.4 registration.poll_read_ready → readiness 有 READABLE → 返回 Ready
12.5 执行 self.io.as_ref().unwrap().read(b) → 真正的 recv 系统调用（tokio-1.40.0/tokio/src/io/poll_evented.rs:188）
12.6 返回读到的字节数

Step 13（用户代码继续）：

13.1 Read Future 返回 Ready(Ok(n))
13.2 用户代码拿到 n、继续处理

13 步、每一步都有具体代码位置、每一步你在前 10 章都见过。

13 步里每一步的开销量级

本机没有可复现的基准环境，下面不给具体数字，只给一个各步骤开销的相对量级排序——它才是有工程意义的部分：

Step	操作	开销性质
1-3	用户代码 → TcpStream::poll_read	纯函数调用，通常会被内联
4	poll_read_ready → 原子 load + 掩码	一次无竞争原子读，fast path
5	Task 状态 RUNNING → Idle	一次 state CAS
6-7	Driver park → epoll_wait	一次系统调用进内核
8	内核等数据到来 → 唤醒 epoll_wait	外部延迟，0 到无限——完全不由 Tokio 决定
9	mio 事件分发 + Token 转 ScheduledIo 指针	指针转换 + 原子 RMW，无查表
10	ScheduledIo::wake + Waker vtable	一次 Mutex + 一次间接调用
11	scheduler.schedule + worker 取任务	队列 push/pop，可能带一次 unpark
12	Task re-poll → recv 系统调用	一次系统调用，是这条路径上除 Step 8 外最重的一步
13	用户代码拿结果	纯函数返回

关键结论不是"多少纳秒"，而是这个结构：除 Step 8 的外部延迟外，整条 Tokio 路径上必经的系统调用只有两次（epoll_wait 和 recv；Step 11 若要唤醒别的 worker 还会多一次 unpark），其余全是用户态的原子操作、指针运算和函数调用——没有一次堆分配、没有一次哈希查找、没有一次线程创建。

对比你写的业务逻辑（JSON 解析、DB 查询、序列化），业务时间通常远大于这条路径的调度开销。这就是为什么 Tokio 在实际服务里很少成为瓶颈——想知道自己的服务里 Tokio 到底占多少，第 17 章的 tokio-console 和第 19 章的火焰图方法比任何"标称数字"都可靠。

这就是 Tokio——没有任何魔法，只有 13 步精心设计的交互。读完本章你应该真正理解"async I/O"——不是魔法、不是黑盒、不是"就这样子"，而是 13 步确定的、可观察的、可测量的交互。

10.10 本专栏上半场的终结：一次心理复盘

到这里本专栏上半场全部结束。从第 1 章讲 Rust 为什么要把运行时踢出语言、到第 10 章讲 TcpStream 的完整实现路径，你走过了 Tokio 的主体架构：

第 1 章：Rust 异步的宇宙观
第 2-3 章：Future / Waker —— 语言层的基础
第 4 章：Runtime —— 运行时入口
第 5 章：multi_thread Scheduler —— 主调度器
第 6 章：Task —— 调度对象
第 7 章：current_thread + LocalSet —— 另一种调度器
第 8 章：I/O Driver —— 运行时和事件的桥梁
第 9 章：mio —— 事件和系统调用的桥梁
第 10 章：TcpStream / UdpSocket —— 前面所有在一个类型里爆发

心理复盘一下：

前 3 章让你有了"Rust async 是什么"的直觉——不是 Go 的 goroutine、不是 JS 的 Promise，是一种"状态机 + Waker 协议"的新模型
4-7 章让你有了"Tokio 如何执行 async"的机械图像——Runtime 装 Scheduler、Scheduler 分派 Task、Task 是状态机 + 引用计数
8-10 章让你有了"Tokio 如何接外部世界"的图像——epoll_wait 是 heartbeat、ScheduledIo 是 per-fd 状态、每种 I/O 类型都是一个薄包装

这 10 章的关系就像一本建筑图纸的大概念图 + 主体结构 + 关键节点——够你独立阅读任何 Tokio 源码文件、独立诊断生产 Tokio 问题、独立设计一个小规模的自定义 async runtime。

下半场（第 11-20 章）会带你进入时间管理、同步原语、channel、select、blocking、observability、工程实践——每一章都是某个具体能力的深入，但不会再引入根本性的新概念。你在前 10 章建立的那个 mental model 会一直够用。

10.11 和这个系列的其他专栏的关联

本章是前 9 章的大汇总——所有前面的概念在 TcpStream 里一次呈现。和其他专栏关联最紧的是 《Rust 编译器与运行时揭秘》第 13 章（FFI：与 C 世界的桥梁） 和 第 10 章（Pin、Waker 与 Future：异步运行时的三大支柱）——本章的 ScheduledIo 作为 mio Token 背后的指针、unsafe { self.io.poll_read(cx, buf) } 的安全性、Pin<&mut Self> 的所有使用——都是那两章理论在 Tokio 里的具体应用。

和 《vLLM 推理内核深度解析》第 17 章（API 服务器与生产部署） 对比也很有启发——vLLM 的 API server 基于 FastAPI + uvloop，但它的 I/O 模型（epoll + callback）和 Tokio 的 I/O 模型在架构上相似。Python / Rust 在这层做出了几乎相同的选择——说明 readiness + 事件循环是这类服务端的主流解法。

10.12 写自定义 async I/O 类型的完整步骤

这一节提供实战模板——很多人读完 Tokio 源码后想"自己也写个类似 I/O 类型试试"（比如封装一个自定义协议的 socket、封装 io_uring 的 fd、封装一个 mmap 文件）——但不知道从哪开始。这里给你一个三步模板——从事件源到注册、再到读写方法——照着做就能产出一个合格的 Tokio 风格 async I/O 类型。

现在把本章变成可操作的——假设你有一个 Rust 库提供 MyIoDevice（一个 fd-based 的设备，比如串口、PTY、FIFO），你想把它异步化。完整步骤：

Step 1（仅当你要走 mio 那条路）：实现 mio 的 `event::Source` trait

// Cargo.toml：mio = { version = "1", features = ["os-ext"] }（SourceFd 需要 os-ext）
use std::io;
use std::os::fd::RawFd;
use mio::event::Source;
use mio::unix::SourceFd;
use mio::{Interest, Registry, Token};

pub struct MyMioDevice {
    fd: RawFd,
}

impl Source for MyMioDevice {
    fn register(&mut self, registry: &Registry, token: Token, interests: Interest)
        -> io::Result<()>
    {
        SourceFd(&self.fd).register(registry, token, interests)
    }
    fn reregister(&mut self, registry: &Registry, token: Token, interests: Interest)
        -> io::Result<()>
    {
        SourceFd(&self.fd).reregister(registry, token, interests)
    }
    fn deregister(&mut self, registry: &Registry) -> io::Result<()> {
        SourceFd(&self.fd).deregister(registry)
    }
}

SourceFd 是 mio 提供的"给裸 fd 做 Source"的 wrapper——大多数 Unix fd 直接用它就行。复杂设备（比如需要额外 ioctl 的）自己实现。

Step 2：用 `AsyncFd` 包装（不是 PollEvented）

这里有个所有人第一次都会踩的坑：PollEvented 是 pub(crate) 的（tokio/src/io/mod.rs 里写着 pub(crate) use poll_evented::PollEvented;），crate 外面根本 use 不到。Tokio 给外部用户准备的等价物是 tokio::io::unix::AsyncFd<T: AsRawFd>——它内部同样持有一个 Registration，是 PollEvented 的公开对应物（tokio/src/io/async_fd.rs:178-183）：

// 来源：tokio/src/io/async_fd.rs:178-183
pub struct AsyncFd<T: AsRawFd> {
    registration: Registration,
    // The inner value is always present. the Option is required for `drop` and `into_inner`.
    // In all other methods `unwrap` is valid, and will never panic.
    inner: Option<T>,
}

用法：

use tokio::io::unix::AsyncFd;

pub struct MyAsyncDevice {
    io: AsyncFd<MyDevice>,   // MyDevice: AsRawFd
}

impl MyAsyncDevice {
    pub fn new(device: MyDevice) -> io::Result<Self> {
        Ok(MyAsyncDevice { io: AsyncFd::new(device)? })
    }
}

一行完成 Tokio 侧注册——不需要自己操心 Registration / ScheduledIo / mio::Registry。用 AsyncFd 时 Step 1 的 event::Source 也可以省掉：AsyncFd::new 内部会用 SourceFd 把裸 fd 包起来，只要你的类型实现了 AsRawFd。

Step 3：用 ready guard 实现 AsyncRead / 自定义方法

AsyncFd 的核心是 readable() / writable() 返回的 ready guard——拿到 guard 后调同步 syscall，WouldBlock 就 guard.clear_ready() 再循环：

impl MyAsyncDevice {
    pub async fn read(&self, buf: &mut [u8]) -> io::Result<usize> {
        loop {
            let mut guard = self.io.readable().await?;
            // read_sync 是 MyDevice 自己的非阻塞读
            match guard.try_io(|inner| inner.get_ref().read_sync(buf)) {
                Ok(result) => return result,
                Err(_would_block) => continue,   // try_io 已经帮你 clear 了 readiness
            }
        }
    }
}

try_io 就是外部版的 async_io 内层——它替你做"调 syscall、WouldBlock 就清 readiness"这件事。

几十行代码，你的自定义设备就拥有了完整 async 能力——无阻塞、集成 Tokio 调度、支持 .await。这就是 Tokio 抽象层次的威力。

真实案例：mio-serial / tokio-vsock / tokio-tun

Rust 生态里已经有很多这样的库用这个模式包装特殊 fd：

mio-serial + tokio-serial：异步串口
tokio-vsock：VSOCK 虚拟机间通信
tokio-tun：TUN/TAP 网络设备

它们的 Tokio 集成层都很薄，主体是具体设备的 mio::event::Source 实现或 AsyncFd 包装。生态繁荣依赖这种低成本扩展的可能性。

10.13 一个实战小项目：写一个最简 echo server

合起前面所有知识，看一个最小但完整的 echo server——不到 40 行代码，跑在 Tokio 上：

use tokio::net::TcpListener;
use tokio::io::{AsyncReadExt, AsyncWriteExt};

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let listener = TcpListener::bind("127.0.0.1:8080").await?;
    println!("Listening on 127.0.0.1:8080");

    loop {
        let (mut socket, peer) = listener.accept().await?;
        println!("New connection: {}", peer);

        tokio::spawn(async move {
            let mut buf = [0u8; 1024];

            loop {
                match socket.read(&mut buf).await {
                    Ok(0) => {
                        println!("Connection closed: {}", peer);
                        break;
                    }
                    Ok(n) => {
                        if let Err(e) = socket.write_all(&buf[..n]).await {
                            eprintln!("Write error: {}", e);
                            break;
                        }
                    }
                    Err(e) => {
                        eprintln!("Read error: {}", e);
                        break;
                    }
                }
            }
        });
    }
}

不到 40 行，一个进程里就能同时挂住大量并发连接（上限由 fd 数和内存决定，而不是线程数）。所有异步能力都来自本章讲的架构：

TcpListener::bind → PollEvented + Registration + mio::Registry
listener.accept().await → async_io + 等 READABLE + 非阻塞 accept
tokio::spawn → 第 6 章的 Task 结构 + 第 5 章的 worker 队列
socket.read(&mut buf).await → poll_read → poll_read_ready → epoll_wait
socket.write_all(...) → poll_write → poll_write_ready

你现在读得懂这段代码每一个 .await 背后做了什么——这就是前 10 章的收获。

如果你想验证自己的理解，对这个 echo server 做一次"心理执行"：假设有一个客户端连接进来、发了 "hello"——从 listener.accept().await 返回到 socket.write_all 完成，你能一步一步讲清楚每一个内核系统调用、每一个 Tokio 函数、每一个 Future state transition 吗？如果能，你已经完全掌握了本专栏上半场的全部内容。

10.14 下半场预告：每一章的甜点

给你一个下半场（11-20 章）的预告——每一章会给你一个具体的"甜点"（一个让你觉得"原来是这样"的惊喜）：

第 11 章（Time Driver）：分层时间轮——6 层 × 64 槽，插入和删除都是 O(1)，与定时器时长无关（对比最小堆的 O(log n)）
第 12 章（同步原语）：tokio::sync::Mutex 是严格 FIFO 的公平锁（源码注释原话："Tokio's Mutex operates on a guaranteed FIFO basis"）——为什么这让它比 std::sync::Mutex 更可预测
第 13 章（channel）：mpsc 是分块链表（Block<T> 里放一段连续数组、块之间用 AtomicPtr 串起来），broadcast 是 Box<[RwLock<Slot<T>>]> 环形缓冲
第 14 章（select!）：宏展开—— select! 不是运行时魔法，是一段确定的 Future state machine
第 15 章（JoinHandle）：JoinSet 和 FuturesUnordered 的区别——JoinSet 内部是 IdleNotifiedSet<JoinHandle<T>>，把"已被唤醒"和"空闲"的句柄分成两条链表；FuturesUnordered 则要自己承担被轮询 future 的执行
第 16 章（blocking）：spawn_blocking 的线程池启动策略、为什么 block_in_place 不能在 current_thread 运行时里用（没有别的 worker 线程可以接手剩下的任务）
第 17 章（metrics + tracing）：tokio-console 的 protocol——它如何通过 trace event 重建 runtime 状态
第 18 章（多 runtime）：一个 process 里多个 runtime 的合理场景、IO runtime + CPU runtime 的分工架构
第 19 章（性能调优）：典型性能陷阱清单——每个都附诊断步骤
第 20 章（设计模式）：架构模式总结——把全专栏的所有模式总结成可迁移的工程范式

每一章都是独立自洽的——你可以按需跳读，也可以完整读完。下半场比上半场更贴近生产实践——上半场是"理解原理"，下半场是"成为能用好 Tokio 的工程师"。

10.15 本章小结

本章把 Tokio 的 TCP/UDP 网络编程从 API 表面一路拆到 syscall 级别。关键 take-home：Tokio 所有 I/O 类型共享同一套"PollEvented + Registration + AsyncRead/AsyncWrite"基础设施、具体类型只提供"syscall 语义"这一小层差异化代码。这种"共享最大化、差异化最小化"的设计让 Tokio 能在保持一致性的同时覆盖广泛的 I/O 场景——TCP、UDP、Unix Sockets、Windows Named Pipes 全都走同一套骨架。最典型的例外是 tokio::fs：普通文件在 epoll 里"永远可读可写"，readiness 模型对它毫无意义，所以 tokio::fs 完全不碰 PollEvented，而是把同步文件操作丢进 spawn_blocking 线程池（第 16 章会讲）——这是 Tokio 文档里明确写下的实现选择。

带走三件事：

TcpStream 是极简的四层包装：TcpStream → PollEvented<T> → Registration → ScheduledIo，底下是 mio。每一层只有一两个字段、几个方法——简单的分层让整个系统可理解
async_io(interest, f) 是 Tokio I/O 的万能 helper——把"try + register + retry + cancel-safe" 模式压成一行调用。TcpListener::accept、UdpSocket::recv_from、任何自定义 I/O 都用它
TcpStream / TcpListener / UdpSocket / UnixStream 架构完全同构——学会一个就学会了所有。差异只在 API 层（AsyncRead vs send_to）和 mio 底层类型

总结：上半场的一个核心洞察

上半场（第 1-10 章）最核心的一个洞察：Tokio 的所有复杂性都在"为并发和正确性服务"——而不是"为了玩花样"。

每一处看似复杂的设计都有明确目的：

state bit packing：保证状态转移原子
侵入式链表：避免堆分配
全局 VTABLE：让 will_wake 高效
ScheduledIo 指针做 Token：避开 HashMap 查找
分批 wake：避开锁内副作用
async_io helper：消除重复 + cancel safety
PollEvented 泛型：让扩展零成本

每一个都是问题 + 解法的精确对应。没有一段代码是"看起来酷"——每一段都在解决一个具体的工程问题。

读完上半场，你已经看见了工业级基础设施软件是怎么写出来的。下半场的工程实践会让你把这些知识用起来——但"怎么想"的功夫已经在这里建立了。

读完本章值得做的 3 件事：

打开你最近写的一个使用 TcpStream / TcpListener 的 Rust 项目、对着本章重新读一遍——你会看到新的层次
打开 GitHub 上 tokio/src/net/ 浏览 10 分钟，挑一个你没见过的模块（比如 unix/datagram/ 或 tokio/src/net/windows/named_pipe.rs）——你会发现它们和 TcpStream 惊人地相似
思考一个题目：如果让你给一个新的内核协议（比如 SCTP）做 Tokio 异步化，你的第一步会是什么？ 答案：拿到一个非阻塞 fd，交给 AsyncFd（或实现 mio::event::Source），然后一切自然展开。

下一章离开 I/O，进入时间——Tokio 的 Time Driver。你会看到一个分层时间轮（hierarchical timing wheel）如何用 6 层 × 64 槽（NUM_LEVELS = 6、LEVEL_MULT = 64）覆盖长达 (1 << 36) - 1 毫秒的时间跨度、tokio::time::sleep 的精度由什么决定、以及为什么很长的 sleep 和很短的 sleep 都可以在 O(1) 时间注册。

延伸阅读

Tokio 源码：tokio/src/net/tcp/stream.rs
Tokio 源码：tokio/src/net/tcp/listener.rs
Tokio 源码：tokio/src/net/udp.rs
Tokio 源码：tokio/src/runtime/io/registration.rs
《Rust 编译器与运行时揭秘》第 10 章：Pin、Waker 与 Future：异步运行时的三大支柱
《vLLM 推理内核深度解析》第 17 章：API 服务器与生产部署