Hyper 与 Tower：工业级 HTTP 栈

第12章 Connection Dispatcher 与 Role：请求分发状态机

作者 杨艺韬 · 14,390 字 · 发布于 · 更新于

12.1 字节之上的编排

上一章我们读完了 decode.rs / encode.rs / role.rs——Hyper 如何把一个请求的字节变成 http::Request、又把一个 Response 变回字节。但这只是 "单条消息" 的处理。

真实的 HTTP/1.1 连接不只是一条消息。一个连接上：

  • 客户端可能发多个请求（keep-alive）。
  • 服务端正处理上一个请求时，客户端可能已经发来下一个请求（pipelining——虽然 RFC 允许但实际很少开启）。
  • 响应 body 可能在请求读完之前就开始写。
  • 发生 Upgrade 时，整条连接要切换到另一个协议（WebSocket、HTTP/2）。

把所有这些协议事件编排成一个可以被 tokio::spawn(connection.await) 驱动的 future——这就是 conn.rs 和 dispatch.rs 的职责。两个文件合计 2339 行（含测试），是整个 HTTP/1 协议栈的指挥中心。

这一章我们把这个指挥中心拆开——Conn<I, B, T> 的状态机、Dispatcher<D, Bs, I, T> 的 poll loop、它们如何把 Decoder / Encoder / user Service / body channel 编排成一个可正确取消、可 keep-alive、可 upgrade 的完整连接。

源码锁定：hyper 1.9.0 / hyper-1.9.0/src/proto/h1/conn.rs (1531 行) / hyper-1.9.0/src/proto/h1/dispatch.rs (808 行)。

12.2 Conn<I, B, T>：一条连接的所有状态

// hyper-1.9.0/src/proto/h1/conn.rs:38-42
pub(crate) struct Conn<I, B, T> {
    io: Buffered<I, EncodedBuf<B>>,
    state: State,
    _marker: PhantomData<fn(T)>,
}

三个字段：

  • io：buffered IO，把原始 IO（TCP stream）包上一层读写缓冲。第 14 章讲 max_buf_size 与 writev 时会再碰到它。
  • state：连接的所有状态——下面详述。
  • _marker：PhantomData<fn(T)>——T 是 Server 或 Client 的 tag，编译期分支用。

12.2.1 三套状态：Reading / Writing / KeepAlive

State struct 里有二十多个字段（源码 912-961 行），但最核心的是三个 enum：

// hyper-1.9.0/src/proto/h1/conn.rs:963-976, 1023-1028（KA 上的 derive 从略）
enum Reading {
    Init,
    Continue(Decoder),
    Body(Decoder),
    KeepAlive,
    Closed,
}

enum Writing {
    Init,
    Body(Encoder),
    KeepAlive,
    Closed,
}

enum KA {  // keep-alive status
    Idle,
    #[default]
    Busy,
    Disabled,
}

三个 enum 是正交的——连接同时有一个"当前在读什么"的状态 + "当前在写什么"的状态 + "还打算复用吗"的状态。三者组合起来决定连接下一步该做什么。

12.2.2 Reading 状态意义

  • Init：还没收到 request header——等待客户端发第一个字节。
  • Continue(Decoder)：收到了带 Expect: 100-continue 的请求头——发 100 Continue 响应，等 body 开始。
  • Body(Decoder)：正在流式读 body——Decoder 按之前决定的长度语义（Length / Chunked / Eof）切帧。
  • KeepAlive：body 读完了——如果连接允许复用，回到 Init 状态等下一个请求；否则转 Closed。
  • Closed：不能再读了——要么收到了 FIN、要么协议错误、要么用户说别读了。

12.2.3 Writing 状态意义

  • Init：没有响应正在写——等待 user 产出 response。
  • Body(Encoder)：正在写 response body——Encoder 负责 chunk 格式化 / length enforcement。
  • KeepAlive：响应写完了——等对应的 request 也读完就进入 Idle 阶段。
  • Closed：不能再写了。

12.2.4 KeepAlive 状态意义

KA 是最微妙的一个。它记录"这条连接能不能在当前 message 完成之后被复用"：

  • Busy：当前正在处理请求。默认状态。
  • Idle：当前没有请求在跑，可以接收新的。
  • Disabled：永远不复用——下一条 message 结束后必须关。原因可能是 HTTP/1.0、Connection: close header、或者协议错误。

源码有个有意思的小妙招：

// hyper-1.9.0/src/proto/h1/conn.rs:1013-1020
impl std::ops::BitAndAssign<bool> for KA {
    fn bitand_assign(&mut self, enabled: bool) {
        if !enabled {
            trace!("remote disabling keep-alive");
            *self = KA::Disabled;
        }
    }
}

给 KA 重载了 &= 运算符——state.keep_alive &= false; 等价于"如果 false 就设为 Disabled"。源码里实际只有一处这样写（hyper-1.9.0/src/proto/h1/conn.rs:294，poll_read_head 解析出消息头之后）：

self.state.keep_alive &= msg.keep_alive;

msg.keep_alive 是 role.rs 的解析器综合 HTTP 版本与 Connection 头算出来的"这条消息允不允许 keep-alive"。任何一条消息给出 false 就永久 disable——这是一个"单向锁"的语义，用 &= 表达得比 if !x { disable() } 更紧凑、更有代数感。这种小运算符重载在很多工业级 Rust 代码里能见到。

12.3 Reading × Writing × KA 的合法转移

把三个 enum 的 product 算出来——5 × 4 × 3 = 60 种状态组合。但实际上大部分组合不合法。典型的生命周期是：

时刻 1：空连接
  Reading=Init, Writing=Init, KA=Busy

时刻 2：收到 request header
  Reading=Body(decoder), Writing=Init, KA=Busy

时刻 3：body 读完 + 开始写 response
  Reading=KeepAlive, Writing=Body(encoder), KA=Busy

时刻 4：response 写完
  Reading=KeepAlive, Writing=KeepAlive, KA=Busy

时刻 5：状态清理完，回到空 ←─ 这是 try_keep_alive 完成的
  Reading=Init, Writing=Init, KA=Idle

时刻 6：收到下一个 request header
  Reading=Body, Writing=Init, KA=Busy
  ...

这套循环通过 State::try_keep_alive() 推动——当 Reading 和 Writing 同时是 KeepAlive 时，检查 KA 状态决定回到 Init 还是 Closed。源码在 hyper-1.9.0/src/proto/h1/conn.rs:1072-1091，其实很短：

// 精简版
fn try_keep_alive(&mut self) {
    match (&self.reading, &self.writing) {
        (Reading::KeepAlive, Writing::KeepAlive) => {
            if let KA::Busy = self.keep_alive.status() {
                self.keep_alive.idle();
                self.reading = Reading::Init;
                self.writing = Writing::Init;
            } else {
                self.close();
            }
        }
        ...
    }
}

读懂这段就掌握了 "keep-alive 复用" 的本质——不是给 socket 起别的名字，而是把两个状态枚举重置回 Init。socket 还是那个 socket、缓冲区还是那个缓冲区——只是这条连接愿意为下一个请求再跑一遍状态机。

12.4 Dispatcher<D, Bs, I, T>：顶层驱动

Conn 只负责状态，driver 在 Dispatcher 里：

// hyper-1.9.0/src/proto/h1/dispatch.rs:22-28
pub(crate) struct Dispatcher<D, Bs: Body, I, T> {
    conn: Conn<I, Bs::Data, T>,
    dispatch: D,
    body_tx: Option<crate::body::Sender>,
    body_rx: Pin<Box<Option<Bs>>>,
    is_closing: bool,
}

五个字段：

  • conn：上面讲的连接状态。
  • dispatch：D: Dispatch trait——Server / Client 的分支实现。这个后面细讲。
  • body_tx：入站 body 的发送端——当连接读入 body 数据时，塞给这个 sender，让 user handler 能从对应的 Incoming 拿到数据。
  • body_rx：出站 body 的接收端——user handler 产出的 Response body，由这里 poll frame 写回连接。
  • is_closing：是不是正在关闭。

12.4.1 Dispatch trait：Server/Client 的分界

// hyper-1.9.0/src/proto/h1/dispatch.rs:30-43
pub(crate) trait Dispatch {
    type PollItem;
    type PollBody;
    type PollError;
    type RecvItem;

    fn poll_msg(self: Pin<&mut Self>, cx: &mut Context<'_>)
        -> Poll<Option<Result<(Self::PollItem, Self::PollBody), Self::PollError>>>;

    fn recv_msg(&mut self, msg: crate::Result<(Self::RecvItem, IncomingBody)>)
        -> crate::Result<()>;

    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), ()>>;
    fn should_poll(&self) -> bool;
}

四个方法：

  • poll_msg：产出"下一个要发送"的消息（含 body）。
  • recv_msg：消费"刚收到"的消息（含 body）。
  • poll_ready：dispatch 现在能不能接一条新的入站消息？Server 侧看 in_flight 是否为空、Client 侧看是否还有 callback 在等响应——注意它不是 Service::poll_ready（hyper 的 Service 没有 poll_ready，见第 13 章）。
  • should_poll：有没有待做的工作？

这个抽象是 Server 和 Client 的对称点：

Server 侧 Client 侧
PollItem（要发的） StatusCode 响应行 RequestLine 请求行
RecvItem（收到的） RequestLine 请求行 StatusCode 响应行
poll_msg 的来源 poll service.call(req) 返回的 future 从 client::dispatch::Receiver 队列得到
recv_msg 的目的 交给 service.call(req) 交给等待响应的用户

两个 impl（Server 和 Client）在源码里分别是 hyper-1.9.0/src/proto/h1/dispatch.rs 的 521-580 行和 598-702 行。它们各自 handle user-facing 的 API 形状差异——Server 包一个 HttpService<B>、Client 包一个 client::dispatch::Receiver（内部是 tokio::sync::mpsc::UnboundedReceiver）。

12.4.2 Server 侧 poll_msg

// hyper-1.9.0/src/proto/h1/dispatch.rs:532-554（中文注释为本专栏所加）
fn poll_msg(
    mut self: Pin<&mut Self>,
    cx: &mut Context<'_>,
) -> Poll<Option<Result<(Self::PollItem, Self::PollBody), Self::PollError>>> {
    let mut this = self.as_mut();
    let ret = if let Some(ref mut fut) = this.in_flight.as_mut().as_pin_mut() {
        let resp = ready!(fut.as_mut().poll(cx)?);
        // 拿到 user 的 Response<B>
        let (parts, body) = resp.into_parts();
        let head = MessageHead {
            version: parts.version,
            subject: parts.status,
            headers: parts.headers,
            extensions: parts.extensions,
        };
        Poll::Ready(Some(Ok((head, body))))
    } else {
        unreachable!("poll_msg shouldn't be called if no inflight");
    };

    // Since in_flight finished, remove it
    this.in_flight.set(None);
    ret
}

逻辑：

  1. in_flight 是 Pin<Box<Option<S::Future>>>——这就是 user Service 产生的 future。
  2. poll 它，等 user handler 完成。
  3. 把 Response 拆成 head + body，head 交给 Conn 去写、body 留给 Dispatcher 的 body_rx 持续 poll。

12.4.3 Server 侧 recv_msg

// hyper-1.9.0/src/proto/h1/dispatch.rs:556-567
fn recv_msg(&mut self, msg: crate::Result<(Self::RecvItem, IncomingBody)>) -> crate::Result<()> {
    let (msg, body) = msg?;
    let mut req = Request::new(body);
    *req.method_mut() = msg.subject.0;
    *req.uri_mut() = msg.subject.1;
    *req.headers_mut() = msg.headers;
    *req.version_mut() = msg.version;
    *req.extensions_mut() = msg.extensions;
    let fut = self.service.call(req);
    self.in_flight.set(Some(fut));
    Ok(())
}

收到 request + body，构造 http::Request<Incoming>，调用 user Service 的 call——把返回的 future 存进 in_flight。

self.service.call(req) 就是 user handler 被实际调用的地方。整个 hyper 的所有工作——parse、decode、encode、state machine 管理——都是为了让这一行代码被调用。

12.5 poll_loop：一次 poll 做多少事

Dispatcher 的核心循环：

// hyper-1.9.0/src/proto/h1/dispatch.rs:165-193（注释有删节）
fn poll_loop(&mut self, cx: &mut Context<'_>) -> Poll<crate::Result<()>> {
    // Limit the looping on this connection, in case it is ready far too
    // often, so that other futures don't starve.
    //
    // 16 was chosen arbitrarily, as that is number of pipelined requests
    // benchmarks often use.
    for _ in 0..16 {
        let _ = self.poll_read(cx)?;
        let _ = self.poll_write(cx)?;
        let _ = self.poll_flush(cx)?;

        if !self.conn.wants_read_again() {
            return Poll::Ready(Ok(()));
        }
    }

    trace!("poll_loop yielding (self = {:p})", self);

    task::yield_now(cx).map(|never| match never {})
}

三件事，循环最多 16 次：

  1. poll_read：从 IO 读 header 或 body。如果读到了 message，传给 dispatch.recv_msg。
  2. poll_write：如果 user 产出了消息（从 dispatch.poll_msg），把 header 写到 buffered IO；如果正在写 body，poll user body 的 frame，通过 encoder 写。
  3. poll_flush：把 buffered IO 的字节真正提交到底层 socket。

12.5.1 为什么是 16

// 16 was chosen arbitrarily, as that is number of pipelined requests
// benchmarks often use. Perhaps it should be a config option instead.

注释里自己说"arbitrary"——随便选的。但 16 有一定道理：

  • 注释说 pipelining 的 benchmark 常用 16 个 pipelined 请求——上限取 16，一批请求有机会在一次 poll 里走完。
  • 大多数小请求（header-only）一次 poll 就能完成。
  • 如果真有超 16 个 message 堆在同一个 socket 上——最后一行 task::yield_now(cx) 让出 task，让 runtime scheduler 有机会先跑别的 future。

这是一个典型的 cooperative yielding 模式——循环到某个上限后主动让贤，避免"饿死其他 task"。《Tokio 源码深度解析》专栏第 19 章（coop budget）讨论过 Tokio runtime 的 budget 机制——task 每次 poll 有一份预算，tokio 的资源（socket、channel 等）在 poll 时消耗它，耗尽后即便就绪也返回 Pending，迫使 task 让出。Hyper 的 yield_now 是相同的思路——避免任何 future 独占 executor 太久。

12.5.2 wants_read_again 的退出条件

这是 poll_loop 的另一个智慧：循环会提前退出，除非 "读出新东西后状态又变得可读"：

if !self.conn.wants_read_again() {
    return Poll::Ready(Ok(()));
}

大多数时候一次 poll 只做一件事——读一次、写一次、flush 一次，然后返回 Pending 等 IO。只有 notify_read 被置位时才继续循环：典型是"上一条消息处理完、下一条的字节已经在 buffer 里"（pipelining 场景）；此外客户端回到 idle、服务端配置了 header 读超时的连接回到 idle 时也会置位（§12.8.4）。

这样做的好处：非 pipelining 场景（绝大多数）没有额外开销，pipelining 场景又能利用缓存中的字节提升吞吐。按 common case 优化是工业级代码的气质。

12.6 poll_read：身兼数职

这是 Dispatcher 最复杂的一个方法（hyper-1.9.0/src/proto/h1/dispatch.rs:195-269，75 行）。我们挑核心看：

fn poll_read(&mut self, cx: &mut Context<'_>) -> Poll<crate::Result<()>> {
    loop {
        if self.is_closing { return Poll::Ready(Ok(())); }

        if self.conn.can_read_head() {
            // 情况 1：还没读 header——读 header
            ready!(self.poll_read_head(cx))?;
        } else if let Some(mut body) = self.body_tx.take() {
            if self.conn.can_read_body() {
                // 情况 2：header 读过了、正在流式读 body
                match body.poll_ready(cx) { ... }
                match self.conn.poll_read_body(cx) {
                    Poll::Ready(Some(Ok(frame))) => {
                        if frame.is_data() {
                            body.try_send_data(frame.into_data().unwrap())
                        } else if frame.is_trailers() {
                            body.try_send_trailers(frame.into_trailers().unwrap())
                        }
                        ...
                    }
                    ...
                }
            }
        } else {
            // 情况 3：没有正在流的 body——idle 探测交给 Conn
            return self.conn.poll_read_keep_alive(cx);
        }
    }
}

读的三种情况：

  1. 还没读 header：调 poll_read_head —— 从 IO 读字节，调 role::parse_headers 解析。解析成功后构造 IncomingBody 的 channel（body_tx 这边存进 Dispatcher，另一边通过 dispatch.recv_msg 包进 http::Request 交给 Service）。
  2. 正在读 body：body.poll_ready(cx)（hyper-1.9.0/src/body/incoming.rs:353-359）先看 want 信号、再看 data_tx 这条容量为 0 的 mpsc（hyper-1.9.0/src/body/incoming.rs:115）有没有空位。want 的初值只有请求带 Expect: 100-continue 时才是 WANT_PENDING（hyper-1.9.0/src/proto/h1/dispatch.rs:288-289、hyper-1.9.0/src/body/incoming.rs:120）；其余情况下背压靠的是这条零缓冲通道——handler 不取走上一块，dispatcher 就不读下一块（与第 10 章 §10.8.2 同一口径）。检查通过后 poll decoder 拉一帧，送到 body_tx。
  3. 读侧暂时没活儿干（服务端：请求已读完、正在等 handler 或写响应；客户端：连接空闲）：交给 Conn::poll_read_keep_alive——探测对端是否提前断开，客户端空闲时还要检查有没有多余字节（§12.9.1）。

注意嵌套的三层 Poll 对话：

  • dispatcher poll_read 问 Conn：能读 body 吗？
  • Conn 问 decoder：decode 一帧吧。
  • Decoder 问 buffered IO：要读字节。
  • Buffered IO 问 underlying socket：有字节来吗？

任何一层 Pending，整条链挂起。任何一层有东西，信号向上反馈——一路到 dispatcher 把 frame 塞到 body_tx。

12.6.1 源码核对：poll_read 的"用户取消 body"主动检测

§12.6 给了 poll_read 的简化版。真实的 poll_read（hyper-1.9.0/src/proto/h1/dispatch.rs:195-269）有几条为"用户主动放弃 body"准备的容错路径：

fn poll_read(&mut self, cx: &mut Context<'_>) -> Poll<crate::Result<()>> {
    loop {
        if self.is_closing { return Poll::Ready(Ok(())); }
        else if self.conn.can_read_head() { ready!(self.poll_read_head(cx))?; }
        else if let Some(mut body) = self.body_tx.take() {
            if self.conn.can_read_body() {
                match body.poll_ready(cx) {
                    Poll::Ready(Ok(())) => (),
                    Poll::Pending => {
                        self.body_tx = Some(body);
                        return Poll::Pending;
                    }
                    Poll::Ready(Err(_canceled)) => {
                        // user doesn't care about the body
                        // so we should stop reading
                        trace!("body receiver dropped before eof, draining or closing");
                        self.conn.poll_drain_or_close_read(cx);
                        continue;
                    }
                }
                match self.conn.poll_read_body(cx) {
                    Poll::Ready(Some(Ok(frame))) => {
                        if frame.is_data() {
                            let chunk = frame.into_data()...;
                            match body.try_send_data(chunk) {
                                Ok(()) => { self.body_tx = Some(body); }
                                Err(_canceled) => {
                                    if self.conn.can_read_body() {
                                        trace!("body receiver dropped before eof, closing");
                                        self.conn.close_read();
                                    }
                                }
                            }
                        }
                        ...
                    }
                    ...
                }
            }
        }
        else { return self.conn.poll_read_keep_alive(cx); }
    }
}

两处 _canceled 错误的区别处理值得拆开：

1、body.poll_ready(cx) → Err(_canceled)：每读一块 body 之前都会先过 poll_ready，只要此刻 IncomingBody 已被 drop，就走这条分支——不管之前有没有收过 chunk。最典型的是 user handler 根本没读 body（async fn handler(_req: Request) -> Response { ... }），读了一半就 break 再 drop 的，下一轮循环也落在这里。

应对：self.conn.poll_drain_or_close_read(cx); continue;——尝试 drain（继续从 socket 读但丢弃数据）或者直接 close。drain 的好处是让 keep-alive 还能用——客户端发完 body 后这条连接还能复用；如果直接 close，必须断 connection。drain 是性能优化、close 是 fallback（drain 只试一次，见 §12.11.1）。

2、body.try_send_data(chunk) → Err(_canceled)：poll_ready 刚确认过接收端还在、这一块已经从 socket 解码出来，发送时接收端却不在了——handler 跑在另一个线程上、恰好在这两步之间 drop 了 IncomingBody 才会出现。这时手里这块数据已无处可送，hyper 直接 self.conn.close_read() 关掉 read 侧，不再尝试 drain。

两条分支的区别在于丢弃是在读下一块之前被发现，还是在读出一块之后才被发现——前者还有机会 drain 保住 keep-alive，后者直接关读侧。

注释里的 "trace!" 行让运维能追踪到这种异常——日志说 "body receiver dropped before eof"——根因就在用户的 handler 里。

12.6.2 源码核对：poll_read_head 的 dispatch.poll_ready 三态分支

§12.6 顺带提了 poll_read_head。真实代码（hyper-1.9.0/src/proto/h1/dispatch.rs:271-280）开头是这样的：

fn poll_read_head(&mut self, cx: &mut Context<'_>) -> Poll<crate::Result<()>> {
    // can dispatch receive, or does it still care about other incoming message?
    match ready!(self.dispatch.poll_ready(cx)) {
        Ok(()) => (),
        Err(()) => {
            trace!("dispatch no longer receiving messages");
            self.close();
            return Poll::Ready(Ok(()));
        }
    }

    // dispatch is ready for a message, try to read one
    match ready!(self.conn.poll_read_head(cx)) {
        Some(Ok((mut head, body_len, wants))) => { ... }
        ...
    }
}

先问 dispatch 是否 ready 接受新 message——再才向 socket 读 head。这是标准的 backpressure pattern：

  • Server 侧 dispatch.poll_ready 看 in_flight 是否为 None——上一个请求的 Service future 还没完成就返回 Pending——dispatcher 就不读下一个 request header（这也是 hyper HTTP/1 对 pipelined 请求逐个处理、不并发的原因）
  • Client 侧 dispatch.poll_ready 检查 callback 是否还在（用户是否还在等这个响应）——callback 已被 drop 或根本没有在途请求时回 Err(())——dispatch 关闭连接

Err(()) 走 close 路径——很重要：避免连接在没人要消息时还继续读字节、白白缓冲。注意这里执行的是 Dispatch 自己的 poll_ready，hyper 的 Service trait 并没有 poll_ready（第 13 章）。

这条 backpressure 链贯穿 hyper 整个 IO 路径——从 socket → buffered IO → conn → dispatch → user handler——下游没准备好，上游就不再往里读。（hyper 的 Service 没有 poll_ready，service 本身不参与这条链。）没有这条链、HTTP server 在过载时就会不断把字节读进内存缓冲。

12.7 poll_write 与 body_rx 的互动

与 poll_read 对称的 poll_write（略）：

  1. 如果 Writing::Init 且 dispatch 有 message——poll_msg 拿 head + body，encode_headers 写 header，body_rx 记住；writing 切到 Body。
  2. 如果 Writing::Body——poll body_rx 下一帧，用 Encoder 格式化（chunked / length）写 buffered IO。
  3. body 写完——Writing::KeepAlive；try_keep_alive 检查是否可以复用。

核心循环模式相同：循环往 IO 推数据，直到 Pending。

12.7.1 body_rx 的 Pin<Box<Option<Bs>>>

这个类型很有意思：

body_rx: Pin<Box<Option<Bs>>>,   // hyper-1.9.0/src/proto/h1/dispatch.rs:26

三层包装——Box 堆分配、Pin 固定地址、Option 允许清空。为什么要这么复杂？

  • Pin：user 的 Body 可能是 !Unpin（典型例子是 async_stream! 宏生成的 Stream），需要 pin 才能 poll。
  • Box：Pin<&mut Bs> 需要 Bs 在内存中地址稳定——堆分配保证这一点，而且 Pin<Box<_>> 本身是 Unpin 的，Dispatcher 不会因为装着一个 !Unpin 的 body 而失去 Unpin（它的 Future::poll 直接经 &mut self 调 poll_catch，要求自身 Unpin）。
  • Option：body 写完之后要被 drop——self.body_rx.set(None) 清空 Box 的内容（hyper-1.9.0/src/proto/h1/dispatch.rs:340；反向的装填在 348 行 set(Some(body))，初始值是 Box::pin(None)，85 行）。

Pin<Box<Option<T>>> 这种三层包装是 Rust 异步代码里处理"可替换、可 pin、可 unpin 的 future/stream"的标准模板。理解它比去死记它来得实用——下次你遇到类似需求，知道怎么拼就行了。

12.7.2 源码核对：poll_write 的 body chunk 处理三分支 + 空 chunk 优化

§12.7 给的 poll_write 是高层视角。真实代码（hyper-1.9.0/src/proto/h1/dispatch.rs:328-421）有几条专门优化 chunk 流的细节：

let item = ready!(body.as_mut().poll_frame(cx));
if let Some(item) = item {
    let frame = item.map_err(|e| {
        *clear_body = true;
        crate::Error::new_user_body(e)
    })?;

    if frame.is_data() {
        let chunk = frame.into_data().unwrap_or_else(|_| unreachable!());
        let eos = body.is_end_stream();
        if eos {
            *clear_body = true;
            if chunk.remaining() == 0 {
                trace!("discarding empty chunk");
                self.conn.end_body()?;
            } else {
                self.conn.write_body_and_end(chunk);
            }
        } else {
            if chunk.remaining() == 0 {
                trace!("discarding empty chunk");
                continue;
            }
            self.conn.write_body(chunk);
        }
    } else if frame.is_trailers() {
        *clear_body = true;
        self.conn.write_trailers(...);
    } else {
        trace!("discarding unknown frame");
        continue;
    }
} else {
    *clear_body = true;
    self.conn.end_body()?;
}

四条工程精细：

1、空 chunk 的双路径丢弃：用户 body 实现可能 yield Ok(empty_chunk)——比如 stream 里中间状态。chunked encoding 下传一个空 chunk 字面意味"end-of-message"——错误地写出去会让客户端以为 body 结束、提前关流。所以 hyper 检测 chunk.remaining() == 0 就discard——非 EOS 时 continue 等下一帧；EOS 时直接 end_body 而不写 chunk。

2、body.is_end_stream() 优化合并最后一帧 + 终止：当 user body 的当前 frame 是最后一帧——write_body_and_end(chunk) 一次性把 chunk + 结束标记（chunked 的 0\r\n\r\n terminator）编码进写缓冲、同时把 Writing 切到 KeepAlive/Closed（hyper-1.9.0/src/proto/h1/conn.rs:754-772），省掉一轮 poll_frame 和一次单独的 end_body。两种写法都只是写进缓冲，真正的系统调用发生在之后的 poll_flush。

3、*clear_body = true 配合 OptGuard 模式：OptGuard::new(self.body_rx.as_mut()).guard_mut()（hyper-1.9.0/src/proto/h1/dispatch.rs:361）拿到的 clear_body 是 RAII guard 里的那个 bool——struct OptGuard<'a, T>(Pin<&'a mut Option<T>>, bool)（hyper-1.9.0/src/proto/h1/dispatch.rs:479），它的 Drop（hyper-1.9.0/src/proto/h1/dispatch.rs:491）在 bool 为 true 时把 body_rx 置 None。好处是提前退出也不漏清理：先 *clear_body = true 再经 ? 返回错误、或者 continue 进下一轮，guard 在离开作用域时都会把 body_rx 清掉，不用在每条出口手写 set(None)。（bool 默认是 false，poll_frame 中途 panic 时 guard 并不会清空 body_rx。）

4、unknown frame 丢弃：http_body::Frame 是个 struct，里面包着一个私有的 Kind 枚举（http-body-1.0.1/http-body/src/frame.rs:5-10，目前是 Data / Trailers，未来可能加新变体）。hyper 用 is_data() / is_trailers() 判明确支持的两种、其他静默丢弃（continue）。Forward compatibility——http_body crate 未来加新 frame type 不会让 hyper 编译失败、只是不处理新类型。

这些细节合起来让 hyper 的 body 写入路径对各种 abnormal user body 都鲁棒——空 chunk、未知 frame、body 报错、提前 EOS——都有应对。生产级 HTTP server 必须 handle 这些 case、否则一个 buggy user body 就可能写坏整条连接。

12.8 错误、Upgrade、Shutdown 的正交处理

Dispatcher 还要处理三类"不走正常流程"的事件。

12.8.1 错误：poll_catch

// hyper-1.9.0/src/proto/h1/dispatch.rs:123-141
fn poll_catch(&mut self, cx: &mut Context<'_>, should_shutdown: bool)
    -> Poll<crate::Result<Dispatched>>
{
    Poll::Ready(ready!(self.poll_inner(cx, should_shutdown)).or_else(|e| {
        // Be sure to alert a streaming body of the failure.
        if let Some(mut body) = self.body_tx.take() {
            body.send_error(crate::Error::new_body("connection error"));
        }
        self.dispatch.recv_msg(Err(e))?;
        Ok(Dispatched::Shutdown)
    }))
}

poll_inner 的错误都汇到 poll_catch 处理：

  1. 如果有正在接收的 body——把错误发到 body_tx，让 handler 侧的 body.poll_frame 看到 Err。
  2. 把错误交给 dispatch 的 recv_msg(Err(e))——Client 侧若有在等响应的 callback（或队列里还没发出的请求），就把错误递给那个用户，recv_msg 返回 Ok；Server 侧的 recv_msg 第一行就是 msg?，错误原样返回。
  3. recv_msg 返回 Ok 时返回 Dispatched::Shutdown；返回 Err 时经 ? 把错误作为整个连接 future 的结果抛出去。

所以错误不会静默消失：Client 侧顺着 body channel 和 callback 传给等待中的用户，Server 侧则成为 serve_connection 返回的 Err。

一个直接后果：user Service 返回 Err 时，poll_write 里的 msg.map_err(crate::Error::new_user_service)?（hyper-1.9.0/src/proto/h1/dispatch.rs:337）把它变成连接错误，走的正是这条路——hyper 不会替你写一个 500，而是直接结束这条连接。用 hyper 1.9.0 + tokio 1.46.1 实测：handler 返回 Err 后客户端读到 0 字节就 EOF，serve_connection 返回 Err("error from user's Service")。想给客户端回错误响应，得在 handler 里自己把错误映射成 Response。

12.8.2 Upgrade：特殊分支

当客户端发 Upgrade: websocket + 服务端响应 101 Switching Protocols，连接就脱离 HTTP/1 协议了——TCP stream 被交给 user 继续用。

// hyper-1.9.0/src/proto/h1/dispatch.rs:150-157（摘自 poll_inner，142-162）
if self.is_done() {
    if let Some(pending) = self.conn.pending_upgrade() {
        self.conn.take_error()?;
        return Poll::Ready(Ok(Dispatched::Upgrade(pending)));
    } else if should_shutdown {
        ready!(self.conn.poll_shutdown(cx))...;
    }
    ...
}

pending_upgrade() 取出一个 upgrade::Pending（返回 Option<upgrade::Pending>，定义在 hyper-1.9.0/src/proto/h1/conn.rs:163，调用处在 hyper-1.9.0/src/proto/h1/dispatch.rs:152）——这个 handle 会在未来的某一刻把 raw socket 交出去，用户拿到 Upgraded 后可通过 Upgraded::downcast::<T>() 取回原始 IO。Dispatcher 的工作到这里结束——它不会再读写 socket，让给 upgrade handler。

Upgrade 的完整机制是第 18 章的主题。这里只知道 Dispatcher 通过 Dispatched::Upgrade(pending) 这个 退出码 把"连接控制权"交回给上层。

12.8.3 Shutdown 的优雅顺序

else if should_shutdown {
    ready!(self.conn.poll_shutdown(cx))?;
}

poll_shutdown 调到 hyper::rt::Write::poll_shutdown（tokio 的 AsyncWrite 经 hyper-util 的 TokioIo 适配到这个 trait）——在 TCP 上发一个 FIN，告诉对端"我不再发了"。对端会看到 EOF，自然关闭它的 read half。

这一步是 TCP 半关闭——只关写方向：已写出的字节（比如你的响应最后几个字节）照常送达，之后对端读到 EOF。poll_shutdown 完成后 Dispatcher 返回 Dispatched::Shutdown，IO 对象随连接 future 一起被 drop，socket 这才真正关闭。

12.8.4 源码核对：try_keep_alive 的 4 路状态转移 + Client 特殊路径

§12.3 给了 try_keep_alive 的简化版。真实代码（State::try_keep_alive，hyper-1.9.0/src/proto/h1/conn.rs:1072-1091；注意同名的 Conn::try_keep_alive 在 hyper-1.9.0/src/proto/h1/conn.rs:568 是另一个函数）覆盖4 种状态组合：

fn try_keep_alive<T: Http1Transaction>(&mut self) {
    match (&self.reading, &self.writing) {
        // 情况 1：双方都 KeepAlive → 看 KA flag 决定 idle 或 close
        (&Reading::KeepAlive, &Writing::KeepAlive) => {
            if let KA::Busy = self.keep_alive.status() {
                self.idle::<T>();
            } else {
                trace!("try_keep_alive({}): could keep-alive, but status = {:?}",
                    T::LOG, self.keep_alive);
                self.close();
            }
        }
        // 情况 2/3：一方 Closed 一方 KeepAlive → 直接 close
        (&Reading::Closed, &Writing::KeepAlive)
        | (&Reading::KeepAlive, &Writing::Closed) => {
            self.close()
        }
        // 情况 4：其他组合（一方还在 Init/Body 等）→ 啥都不做、等下次 try
        _ => (),
    }
}

第 2/3 种情况的存在反映了对称性破缺的处理：read 关了但 write 还想 keep-alive——明显矛盾、必须关；反之亦然。如果不处理这种"半关闭一半 keep-alive"的状态，连接会卡在不正常状态——既不真关也不真复用。

fn idle<T: Http1Transaction>（hyper-1.9.0/src/proto/h1/conn.rs:1104-1134）里有一段Client 特有的逻辑：

fn idle<T: Http1Transaction>(&mut self) {
    debug_assert!(!self.is_idle(), "State::idle() called while idle");

    self.method = None;
    self.keep_alive.idle();
    if !self.is_idle() {
        self.close();
        return;
    }

    self.reading = Reading::Init;
    self.writing = Writing::Init;

    // !T::should_read_first() means Client.
    //
    // If Client connection has just gone idle, the Dispatcher
    // should try the poll loop one more time, so as to poll the
    // pending requests stream.
    if !T::should_read_first() {
        self.notify_read = true;
    }

    #[cfg(feature = "server")]
    if self.h1_header_read_timeout.is_some() {
        // Next read will start and poll the header read timeout,
        // so we can close the connection if another header isn't
        // received in a timely manner.
        self.notify_read = true;
    }
}

!T::should_read_first() 是 client side 的标识——server 是"先读 request 再写 response"、client 是"先写 request 再读 response"。Client 在 idle 时设 notify_read = true 让 Dispatcher 立刻再 poll 一次——因为 client 可能有 user 在等着发新请求、需要立刻处理。Server 不需要——它只是被动等下一个请求来。

这个 should_read_first() 是 hyper Server/Client trait 区分的核心方法之一——通过它在编译期 dispatch 到不同的状态机行为。server feature 和 client feature 共享 conn.rs 的代码——dispatch 通过 trait method 实现 zero-cost abstraction。

#[cfg(feature = "server")] 后面的 h1_header_read_timeout 是 server 特有的配置——下一个请求 header 必须在 timeout 内到达、否则关连接（防 slowloris attack）。这条逻辑也通过 notify_read 立刻触发下一次 poll。

短短 30 行 idle 函数同时处理 Client 立刻 poll、Server timeout 启动、双方共用的状态重置——用 trait method + cfg 让两个 use case 共享代码。这种"trait + cfg 双 dispatch" 是 Rust feature gating 的常见 pattern。

12.9 Conn 的 maybe_notify 小玩法

最后看一段鲜为人知但极其 clever 的代码。Conn state 里有一个字段：

notify_read: bool,   // hyper-1.9.0/src/proto/h1/conn.rs:950（在 State 上，初值 false，78 行）

它是一个"我需要被再 poll 一次"的标记。一条消息读完、响应写完，try_keep_alive 把状态复位回 Init 之后会调 maybe_notify（hyper-1.9.0/src/proto/h1/conn.rs:521-566）：如果下一个 request 的字节已经在 buffer 里（或此刻能从 socket 读到），就置位 notify_read——poll_loop 不让 task 挂起、直接循环回 poll_read——这就是 wants_read_again 的意义。

// hyper-1.9.0/src/proto/h1/dispatch.rs:184-187（`wants_read_again` 本体在 hyper-1.9.0/src/proto/h1/conn.rs:429-433：读一次 notify_read 并复位）
if !self.conn.wants_read_again() {
    return Poll::Ready(Ok(()));
}

简单两行，但它关系到 pipelining 的正确性：源码注释（hyper-1.9.0/src/proto/h1/dispatch.rs:176-180）说得很直白——下一条请求的字节已经躺在读缓冲里、socket 上不会再有新的可读事件，如果这里不再跑一轮 poll_read，"we wouldn't ever be woken up again"，连接就挂住了。

有注释特别提到（hyper-1.9.0/src/proto/h1/dispatch.rs:182-183）：

Using this instead of task::current() and notify() inside the Conn is noticeably faster in pipelined benchmarks.

翻译：不在 Conn 里用 waker 重新 notify 自己，而是直接把下一轮 loop 继续跑——在 pipelined benchmark 里明显更快。也就是说，"再跑一轮"是正确性要求，用标记位而不是 waker 实现它是性能选择。

读 hyper 源码你会一再看到这类"为 benchmark 调优"的痕迹。

12.9.1 源码核对：poll_read_keep_alive 的三态——区分"安全 idle"和"异常字节"

§12.8.4 提了 idle 状态下还可能要 read。真实 poll_read_keep_alive（hyper-1.9.0/src/proto/h1/conn.rs:435-445）做的事是在读侧没事可做（不能读 head、也没有 body 可读）时仍然要从 socket 读、判断什么时候是合法 EOF、什么时候是协议错误：

pub(crate) fn poll_read_keep_alive(&mut self, cx: &mut Context<'_>) -> Poll<crate::Result<()>> {
    debug_assert!(!self.can_read_head() && !self.can_read_body());

    if self.is_read_closed() {
        Poll::Pending
    } else if self.is_mid_message() {
        self.mid_message_detect_eof(cx)
    } else {
        self.require_empty_read(cx)
    }
}

三个分支：

1、is_read_closed() → Pending——已经关了 read 端、不应该再尝试任何 IO。返回 Pending 让 task 睡眠（直到外部事件唤醒）。

2、is_mid_message() → mid_message_detect_eof——还在处理一条消息中（reading 或 writing 不是 Init）。这时如果 socket 突然 EOF——是异常——客户端不该在 message 中途断连。报 Error::new_incomplete()。

3、其他 → require_empty_read——纯 idle 状态（只有 client 会走到这里，见下）。期望 socket 没有任何字节——如果有字节、说明服务端在没有请求时发了多余数据（new_unexpected_message）；如果 EOF——可能是合法的"服务端关闭空闲连接"（return Ok）或异常断开（return error，取决于 should_error_on_eof）。

require_empty_read（hyper-1.9.0/src/proto/h1/conn.rs:458-488，开头就是 debug_assert!(T::is_client())）专门给 client side 用——Server 不会主动检测 idle 时是否有 unexpected bytes（因为 server 收到额外数据可能就是 pipelining 的下一个请求）。Client 在 idle 时收到额外字节是 protocol violation——server 不该在 client 没发请求时主动发数据。

mid_message_detect_eof（hyper-1.9.0/src/proto/h1/conn.rs:491-508）有个微妙的 allow_half_close 检查（hyper-1.9.0/src/proto/h1/conn.rs:495-497）：

if self.state.allow_half_close || !self.io.read_buf().is_empty() {
    return Poll::Pending;
}

half-close 模式下——客户端关闭 write 端但 server 还能继续 write response（典型场景：长 polling、SSE、某些 RPC）——hyper 不报错。allow_half_close 是个 server config 选项、默认 false（严格模式）。

这两个函数加起来覆盖了 idle 状态下的4 种 socket 事件：合法 EOF（safe close）、异常 EOF（incomplete）、unexpected bytes（protocol error）、合法 half-close（allow_half_close 模式）。每种事件走不同路径——保证 hyper 在各种异常 client 行为下都有正确响应。

这种"边角情况的精确分类处理"是生产 HTTP server 的核心质量指标——一个 server 的"健壮性"不在 happy path 表现、在 90% 时间不会发生但发生了就崩的边角情况。hyper 的 conn.rs 用 1500 行代码精心覆盖这些 case。

12.10 一张连接生命周期图

把整章讲的东西画成图：

TCP accept
    ↓
Conn::new(io) — state.reading=Init, writing=Init, KA=Busy
    ↓
Dispatcher::new(svc, conn)
    ↓
tokio::spawn(dispatcher) ─── 进入 poll_loop
        │
        ├─ poll_read   ───── role::parse ─── dispatch.recv_msg ─── svc.call(req)
        │                                                              │
        ├─ poll_write  ───── dispatch.poll_msg ─── role::encode ───── writes header
        │                    │
        │                    └── body.poll_frame ─── encoder 写 body
        │
        ├─ poll_flush  ───── underlying socket
        │
        └─ wants_read_again? ─── 继续循环 / yield

事件发生：
  - 读完 body       → Reading::KeepAlive
  - 写完 body       → Writing::KeepAlive
  - try_keep_alive  → 若 KA=Busy，reset 到 Init（开始下一轮）
  - 或 KA=Disabled  → close(); poll_shutdown
  - 或 Upgrade      → Dispatched::Upgrade(pending)，Dispatcher 退出

这张图浓缩了 2339 行代码的指挥流。读懂它你就读懂了 hyper 1.x HTTP/1 的全部顶层逻辑。

12.10.1 源码核对：poll_read_head 的 h1_header_read_timeout——slowloris 防御

§12.8.4 提到 server 有 h1_header_read_timeout。打开 hyper-1.9.0/src/proto/h1/conn.rs:211-277 看实现（略去 trace 行和 ParseContext 的字段，变量名有缩写）：

pub(super) fn poll_read_head(...) -> Poll<...> {
    debug_assert!(self.can_read_head());

    #[cfg(feature = "server")]
    if !self.state.h1_header_read_timeout_running {
        if let Some(h1_header_read_timeout) = self.state.h1_header_read_timeout {
            let deadline = self.state.timer.now() + h1_header_read_timeout;
            self.state.h1_header_read_timeout_running = true;
            match self.state.h1_header_read_timeout_fut {
                Some(ref mut fut) => {
                    self.state.timer.reset(fut, deadline);
                }
                None => {
                    self.state.h1_header_read_timeout_fut =
                        Some(self.state.timer.sleep_until(deadline));
                }
            }
        }
    }

    let msg = match self.io.parse::<T>(cx, ParseContext { ... }) {
        Poll::Ready(Ok(msg)) => msg,
        Poll::Ready(Err(e)) => return self.on_read_head_error(e),
        Poll::Pending => {
            #[cfg(feature = "server")]
            if self.state.h1_header_read_timeout_running {
                if let Some(ref mut fut) = self.state.h1_header_read_timeout_fut {
                    if Pin::new(fut).poll(cx).is_ready() {
                        self.state.h1_header_read_timeout_running = false;
                        warn!("read header from client timeout");
                        return Poll::Ready(Some(Err(crate::Error::new_header_timeout())));
                    }
                }
            }
            return Poll::Pending;
        }
    };

    #[cfg(feature = "server")]
    {
        self.state.h1_header_read_timeout_running = false;
        self.state.h1_header_read_timeout_fut = None;
    }
    ...
}

这套机制专门 mitigate Slowloris 攻击——攻击者建立 TCP 连接、慢慢发送 header（每秒 1 字节）、占用 server connection slot 直到 server 资源耗尽。Slowloris 是 2009 年提出的著名 HTTP DoS 攻击——对每连接一线程/进程的 server（如经典的 Apache prefork）尤其致命。

hyper 的应对：

1、header_read_timeout（server::conn::http1::Builder 选项，默认 30 秒：Dur::Default(Some(Duration::from_secs(30)))）——"header 必须在 N 秒内全部读完、否则关连接"。但它依赖 Builder::timer 提供的 Timer：没设 timer 时默认值会被 Time::check 静默跳过（源码里那句 warn! 在默认编译下是空操作——要打开 tracing feature、以 RUSTFLAGS="--cfg hyper_unstable_tracing" 编译并装上 subscriber 才看得到，见 hyper-1.9.0/src/trace.rs:5-11、:109-116；上面代码里超时时的那句 warn! 同理），而显式调了 header_read_timeout(..) 却没设 timer 会 panic。传 None 可禁用。装了 timer 之后还有一个容易忽略的效果：连接处理完一个请求回到 idle 时，State::idle 会置位 notify_read（hyper-1.9.0/src/proto/h1/conn.rs:1127-1133，§12.8.4），下一轮 poll_read_head 立刻为"下一个请求头"开始计时——所以空闲的 keep-alive 连接等不到新请求，到时也会被关掉。hyper 服务端没有单独的 idle 超时选项，空闲等待就是由它计时的（第 14 章细讲）。

2、正常路径上每个请求都新建一个 sleep：读完 header 后 h1_header_read_timeout_fut 会被置回 None（见第 3 点），所以下一个请求走的是 None 分支、由 timer.sleep_until 新建一个 sleep；Some 分支的 timer.reset 只有 fut 还没被清掉时才会走到。hyper 里真正靠 reset 反复复用同一个 sleep 的是 HTTP/2 的 keep-alive ping（hyper-1.9.0/src/proto/h2/ping.rs:453、:476）。

3、读到 header 后清理 timer：h1_header_read_timeout_running = false + h1_header_read_timeout_fut = None——避免 timer 在已经完成 read 后继续触发。

4、Pending 时 poll timer：在每次 socket Pending 时主动 poll timer——如果 timer 已经 ready（超时了）——立刻 raise header_timeout error 关连接。

这是个多层 future 协调的范例——socket future 和 timer future 同时被 poll、谁先 ready 谁触发对应行为。这种"race between IO and timer"在异步 server 里是常见 pattern（第 5 章 tower 的 Timeout 也是让响应 future 和 sleep 赛跑）。但要注意：hyper 的 HTTP/1 服务端并没有独立的 keep-alive 超时或 idle 超时选项，空闲连接的等待靠的就是上面这个 header 读超时。

12.10.2 源码核对：state.busy() + keep_alive &= msg.keep_alive 的状态推进

poll_read_head 拿到 msg 后执行（hyper-1.9.0/src/proto/h1/conn.rs:293-295）：

self.state.busy();
self.state.keep_alive &= msg.keep_alive;
self.state.version = msg.head.version;

三行代码做了三件事：

1、state.busy() 标记连接正忙——之前可能是 Idle，现在收到 request 头要变 Busy。但有个微妙：busy() 内部检查 if let KA::Disabled = self.keep_alive.status() { return; }——如果之前已经被 disable（比如 connection: close header），不会回退到 Busy——保持 Disabled 状态。

2、keep_alive &= msg.keep_alive——§12.2.4 讲过这条 BitAndAssign。msg.keep_alive 是 parser 根据 HTTP 版本和 Connection header 计算出的"这条 message 是否允许 keep-alive"。&= 让任何一条 false 永久 disable——单向锁语义。

3、state.version = msg.head.version——记录这条 message 的 HTTP 版本。写响应时 enforce_version（hyper-1.9.0/src/proto/h1/conn.rs:682-702）据此调整：对端是 HTTP/1.0 时先由 fix_keep_alive 修正 Connection 头，再把响应版本压成 HTTP/1.0；对端是 HTTP/1.1 而 keep-alive 已被禁用时补一个 Connection: close。

三行配合体现了状态推进的精确性——每条新进 message 都按规则更新 connection state、不会因为某个状态被错误地"重置"而违反协议。

12.11 和其他章节的呼应

  • 第 11 章讲的 Decoder / Encoder 被 Conn::state 的 Reading / Writing 持有。
  • 第 10 章讲的 Incoming::Chan { data_rx, want_tx } 的 sender 端就是 Dispatcher 的 body_tx——want 信号和零容量的 data 通道一起经 body.poll_ready 传到 Dispatcher 的 poll_read，决定是否 poll decoder。
  • 第 13 章（下一章）讲的 hyper::Service trait 是被 Server::recv_msg 这一行调用：self.service.call(req)。
  • 第 18 章会接回 pending_upgrade() 的流程。

整个专栏到这里进入"hyper 核心最难的部分"——协议状态机。读完第 11、12 章你应该感到有些累——因为它们是全专栏最 intricate 的章节。但这也是收获最大的两章——这就是工业级 HTTP/1 实现，没有捷径。

12.11.1 源码核对：poll_drain_or_close_read 的优雅 drain 策略

§12.6.1 提到了 user 早 drop body 时 hyper 走"drain or close"。打开 hyper-1.9.0/src/proto/h1/conn.rs:848-865：

/// If the read side can be cheaply drained, do so. Otherwise, close.
pub(super) fn poll_drain_or_close_read(&mut self, cx: &mut Context<'_>) {
    if let Reading::Continue(ref decoder) = self.state.reading {
        // skip sending the 100-continue
        // just move forward to a read, in case a tiny body was included
        self.state.reading = Reading::Body(decoder.clone());
    }

    let _ = self.poll_read_body(cx);

    // If still in Reading::Body, just give up
    match self.state.reading {
        Reading::Init | Reading::KeepAlive => {
            trace!("body drained")
        }
        _ => self.close_read(),
    }
}

三条工程妙处：

1、Reading::Continue 的特殊路径——客户端发了 Expect: 100-continue 期望 server 同意才发 body。如果 user 已经 drop 了 body 接收者——根本不该回 100-continue（client 收到后会发 body、徒增网络流量）。代码直接把状态转到 Reading::Body、不发 100-continue——这是 server 的"礼貌拒绝"。

2、试一次 poll_read_body 看能不能立刻 drain——如果 socket 缓冲里已经有 body 字节、立刻读完进入 Init / KeepAlive 状态、连接复用。如果 body 还没到、状态保持 Body——_ => close_read() 直接关。

这个权衡背后的考虑：drain 一个未读的小 body（几 KB）几乎免费、能保住 keep-alive；drain 一个大 body（几 MB）耗 CPU + 网络、不如直接关。hyper 用"试一次能不能立刻 drain、不能就 give up"的策略——比 "总是 drain" 或 "总是 close" 更精细。

3、注释 "If still in Reading::Body, just give up" 表达了 hyper 的核心 opinion——不为了 keep-alive 强行付出昂贵代价。Connection 复用是优化、不是宗教——能复用就复用、不能就关。这条 pragmatic 的态度让 hyper 在"客户端各种异常行为"下都不会陷入 hang/wait 状态。

这条 17 行函数浓缩了 HTTP server 写作里最微妙的取舍之一——keep-alive 的成本/收益、用户行为的异常处理、网络 IO 的按需触发。读懂它能让你写自己的 server 时不再纠结"用户中途 drop 我该咋办"——答案是 try-then-give-up。

12.11.2 hyper Connection 编排 vs 其他 HTTP server 实现

横向对比让你看到 hyper 在 Connection 这一层的设计取舍：

Apache httpd（C，prefork/worker MPM）：传统 multi-process / multi-thread 模型——每条连接一个 worker。Connection 状态全在线程局部变量 + apr_pool 里。简单但内存重——并发连接数直接对应进程/线程数。

nginx（C，多个 worker 进程、每个 worker 单线程事件驱动）：同一 worker 里的 connection 共享一个 event loop——每条 connection 是一个 ngx_connection_t struct，状态用 callback chain 表达。极致性能但代码复杂——一个事件回调能跨越几百行代码追踪。

Node.js http module（JavaScript on libuv）：Connection 是个 EventEmitter——状态通过 callback + event 表达。比 nginx 易写，但要背 V8 与 GC 的运行时成本。Connection 状态散落在多个 callback closure 里。

Go net/http（goroutine per connection）：每个 connection 起一个 goroutine——状态在 goroutine 局部变量里。GC + go scheduler 让代码写起来像同步、跑起来异步。简单但连接数极高时会有 GC 压力。

hyper（async/await + tokio）：Connection 是一个 Future——状态在 Conn struct 里、状态机由 poll() 驱动。和 Go 类似的"代码像同步"体验、但没有 GC、也不用给每条连接分配独立的栈——Future 直接编译成状态机代码，调度交给 tokio 这样的运行时。这是 Rust 的核心承诺在 HTTP server 层的体现。

这 5 种实现各有取舍：thread-per-connection 最易写但资源最重，event loop + callback 最省资源但最难读，goroutine 与 Future 状态机都在"代码像同步"与"资源占用"之间取折中，区别在于 Rust 把状态机在编译期展开、不需要 GC，也不需要为每条连接准备一段栈。本章讲的 1500 行 conn.rs + 800 行 dispatch.rs，就是 hyper 在这组取舍里给出的答案。

12.11.3 hyper Connection 的设计哲学三连问

读完本章后你应该能用三句话回答 hyper 的 Connection 设计哲学：

Q1：为什么要把 Conn 和 Dispatcher 拆开？ A：分离 state 和 driver——state 容易测、driver 容易换。测 conn 状态机不需要 mock 整个 IO + Service——只需 mock 一个 IO。不过 1.9.0 里 conn.rs 末尾的 tests 模块（1156 行起）只剩一个 nightly 下才编译的 benchmark（用 tokio_test::io::Builder 构造 Conn），其余旧测试都被 /* */ 注释掉了；现役的用法要看 dispatch.rs 的 tests 模块，那里用 tokio_test::io::Builder 驱动整个 Dispatcher。

Q2：为什么状态机不用 thread-per-connection（每个 connection 一个 thread）？ A：thread-per-connection 简单但内存重。hyper 用 Future 状态机让每条 connection 的常驻状态只是一个 Future 的大小。在性能与资源占用上，future 状态机明显优于 thread-per-connection——这是 Rust async 的核心承诺。

Q3：为什么 Reading/Writing/KA 三 enum 分开、不合一个？ A：正交性。Reading 决定"现在能不能读"、Writing 决定"现在能不能写"、KA 决定"完成后要不要复用"。三件事独立变化、合并到一个 enum 会有 N×M×K 种状态、组合爆炸。三 enum 让每个维度独立 reason about、bug 更少。

这三连问的答案是 hyper 设计的精髓——职责分离 + 性能优先 + 维度正交。任何写 connection-style 协议库的人都该思考这三个问题、给出自己的答案。如果你的设计违反这三条、大概率会有维护问题。

12.11.4 写一个最小 hyper-style Connection driver 的 100 行尝试

理解本章后给读者一个练手项目：用 100 行内代码实现一个 hyper-style 的 Connection driver——只服务一个 echo 协议（client 发什么、server 回什么）、但要包含本章讲过的所有核心 pattern。下面只是骨架示意（AsyncReadWrite 是占位名，task::yield_now 是 hyper 内部函数，poll_read 等都省略了），不能直接编译：

struct Conn { state: State, buf_read: BytesMut, buf_write: BytesMut }
struct State { reading: Reading, writing: Writing, ka: KA }
enum Reading { Init, Body(usize), KeepAlive }
enum Writing { Init, Body(Bytes), KeepAlive }
enum KA { Busy, Idle, Disabled }

impl Conn {
    fn poll_loop<S: AsyncReadWrite>(&mut self, io: &mut S, cx: &mut Context) -> Poll<io::Result<()>> {
        for _ in 0..16 {
            let _ = self.poll_read(io, cx)?;
            let _ = self.poll_write(io, cx)?;
            if !self.wants_more() { return Poll::Ready(Ok(())) }
        }
        task::yield_now(cx).map(|_| Ok(()))
    }
    fn try_keep_alive(&mut self) {
        if matches!((&self.state.reading, &self.state.writing),
                    (Reading::KeepAlive, Writing::KeepAlive)) {
            if matches!(self.state.ka, KA::Busy) {
                self.state.reading = Reading::Init;
                self.state.writing = Writing::Init;
                self.state.ka = KA::Idle;
            } else {
                self.close();
            }
        }
    }
    // ... poll_read / poll_write / close / wants_more 略
}

100 行内不可能 cover 所有 case——但足够让你深刻理解为什么 hyper 要 1500 行。当你写到第 30 行就发现："我的 buf_read 怎么处理 partial frame？" ——hyper 的 Buffered 给了答案。当你写到第 60 行发现："取消怎么传播？" ——hyper 的 dispatch.rs 给了答案。

自己实现一遍是验证理解的好办法——写到卡住的地方再回头对照 hyper 的实现，收获最大。

12.11.5 conn.rs 1500 行的代码结构地图

读完本章后再翻 hyper 的 conn.rs（1500 行）应该能直接 navigate——这里给一份代码区域 map 帮你快速定位：

行号区间 内容 何时去查
1-45 use 声明 + Conn struct 定义 看依赖结构
50-92 Conn::new 构造 + State 初始化 看默认配置
92-160 set_* 配置 setter（set_http1_header_read_timeout 等） 找 Builder 选项落到哪
163-210 pending_upgrade / can_read_head / can_read_body 等状态查询 看读写许可判断
211-362 poll_read_head：timeout + parse + 1xx/100-continue + on_read_head_error 解码请求 header 的全部逻辑
363-428 poll_read_body + decoder 调用 body 流式读取
429-520 wants_read_again + poll_read_keep_alive 三态 + force_io_read idle 状态处理
521-572 maybe_notify + try_keep_alive 读侧的重新唤醒
573-703 can_write_* + write_head / encode_head + fix_keep_alive / enforce_version response header 写入
704-847 write_body / write_trailers / end_body + poll_flush / poll_shutdown body 写入与刷新
849-899 poll_drain_or_close_read + close 系列 + take_error / on_upgrade 异常关闭路径
912-1011 State struct（二十多个字段）+ Reading / Writing enum + Debug impl 找具体配置字段
1013-1153 KA enum + State::busy/idle/close + try_keep_alive + prepare_upgrade keep-alive 状态机
1156-1531 tests 模块（除一个 nightly benchmark 外均已注释掉） 看历史测试写法

1500 行能 navigate 到具体功能区——这就是 "懂源码"的实操体现。任何遇到的 hyper connection 行为问题、你都知道去哪 100 行内找答案。

记 navigation 的核心方法：记关键函数名 + 类型名、不记具体行号。行号会随版本变、函数名 / 类型名相对稳定。grep -n "fn try_keep_alive" 比记 hyper-1.9.0/src/proto/h1/conn.rs:1072 实用得多。

12.11.6 测试 hyper Connection 的几种实战手段

读懂源码后下一步是验证理解——下面 5 种测试 hyper connection 行为的实战手段：

1、hurl 工具发奇怪请求：hurl 能精确控制 HTTP/1 请求的细节——header 顺序、Connection 字段值、Expect: 100-continue 等。比 curl 更适合测协议边界。

2、tcpdump/wireshark 抓包看真实字节：如果 hyper 的行为不符合预期、抓包能 100% 告诉你 wire 上发了什么。tcpdump -i lo -A 'port 8080' 在本地测试一行命令解决。

3、用 tokio-test 写 unit test：tokio_test::io::Builder 能 mock 一个 AsyncRead/AsyncWrite——用它 inject 各种异常字节序列、看 hyper Conn 的状态如何变。hyper 自己的 hyper-1.9.0/src/proto/h1/dispatch.rs 末尾的 tests 模块就是这套用法。

4、tracing 启用 trace 级别日志：hyper 各处都有 trace!() 调用，但它们只在打开 tracing feature 并以 RUSTFLAGS="--cfg hyper_unstable_tracing" 编译时才生效（hyper-1.9.0/src/trace.rs:5-9，见 §12.14）；再装一个按 RUST_LOG 过滤的 subscriber，RUST_LOG=hyper=trace cargo run 就能看到状态转移和 IO。这是定位"hyper 内部到底走到哪了"的最直接手段。

5、miri 跑 connection 单元测试：miri 是 Rust 的 UB checker——能发现 unsafe 代码的边界行为。cargo +nightly miri test 跑一次。

这 5 种手段配合本章源码理解、能让你成为团队里的"HTTP/1 wire-level 专家"——遇到任何 hyper connection 异常都能用源码 + 工具组合定位。这种能力在 cloud-native 时代越来越值钱——不少 API gateway、service mesh proxy 都构建在 hyper 之上。

12.11.7 hyper 1.9 与 0.14 的 try_keep_alive：稳定得出奇

如果你之前读过 hyper 0.14 的 try_keep_alive、再看 1.9 的版本——会发现它一行没变：git show v0.14.0:src/proto/h1/conn.rs 和 v0.14.28 里的 State::try_keep_alive 都是同样的 3 个 match arm，State::busy() 里"KA::Disabled 不回退"的保护也早在 0.14.0 就存在。Reading × Writing × KA 这套状态语义在 0.14 之前就已经稳定下来，1.0 的大改动全在它外面（IO trait、Service trait、body 接口）。

这本身就是一个值得记住的事实：协议状态机是 hyper 最稳定的部分。用 git log -L 追这段代码能看到它最近一次实质修改在 2018 年（0.12 时代），之后只有 2019 年的 rustfmt 动过它——后来的版本只是把它周围的东西换掉。

12.11.8 几个真实生产 bug 的 root cause

几个典型的生产故障场景能用本章知识精确定位：

Bug 1：keep-alive 不工作、每个请求新 TCP

  • 症状：连接数和 TIME_WAIT 暴涨、每个请求都多付一次握手延迟
  • 排查：抓包看连接是不是每个请求后就被关闭（FIN）
  • 根因：response 里误加了 Connection: close、或 client/server 一方 HTTP/1.0 / HTTP/1.1 协商错乱
  • 本章定位：请求侧看 §12.2.4 KA 的 BitAndAssign——请求头让 &= msg.keep_alive 把 KA 置成了 Disabled；响应侧的 Connection: close 则让 Encoder 标记为 last、写完后 Writing 变 Closed，再由 §12.8.4 的 try_keep_alive 情况 2/3 关连接

Bug 2：上传大文件 server hang

  • 症状：客户端 upload 大文件、响应发出后连接卡住、下一个请求永远读不到
  • 根因：user handler 把 Incoming 存了起来既不读也不 drop——零容量的 data 通道一直没人取走，hyper 的 body_tx 等不到空位
  • 本章定位：§12.6.1 的 body.poll_ready 返回 Pending——poll_read 一直停在 body 阶段（drop 掉 Incoming 反而会走 drain-or-close 路径把连接放开）

Bug 3：slowloris 慢速连接拖垮 server

  • 症状：大量连接迟迟发不完 header，把 server 的连接和内存占满
  • 根因：没给 http1::Builder 配 timer——默认 30 秒的 header_read_timeout 因此不生效（§12.10.1）
  • 本章定位：§12.10.1 的 timer race——给 http1::Builder 配上 .timer(TokioTimer::new())，必要时再调 .header_read_timeout(Duration::from_secs(10)) 即可

Bug 4：response 写到一半客户端断开、server CPU 100%

  • 症状：grafana 上 50x errors 不增加但 CPU 飙
  • 根因：socket 已断、但用户的 body 实现在错误后仍不断返回新 frame
  • 本章定位：§12.7 的 poll_write/poll_flush——IO 错误经 new_body_write 上抛到 poll_catch 后连接就会关，busy loop 多半出在 user body 或 handler 自己的重试逻辑里

Bug 5：HTTP/1 server 偶发"connection reset by peer"

  • 症状：客户端报错、server 侧往往看不出异常
  • 根因：server 关闭了一条空闲的 keep-alive 连接（比如 header 读超时到期），client 恰好在同一时刻复用它发出新请求——已关闭的 socket 收到数据会回 RST
  • 本章定位：§12.8.4 的 idle——配置了 header 读超时时，连接空闲等下一个请求的时间也由它计时（§12.10.1）；client 侧对空闲连接上 EOF 的检测见 §12.9.1 的 require_empty_read

这些 bug 的共同特点——没有读 hyper 源码很难定位——hung/reset/high CPU 这些症状指向的不是某行用户代码、是 hyper 内部状态机的某个分支。读懂本章后再遇到这种问题、能直接打开 hyper-1.9.0/src/proto/h1/conn.rs 看哪条状态转移触发了。

12.11.9 hyper 1.x vs 0.14 在 connection 设计上的演进

如果你之前用过 hyper 0.14 再来读 1.x 源码——能看到几个架构级的演进：

1、Service trait 的精简：0.14 直接复用 tower_service::Service（poll_ready + call(&mut self)）。1.x 定义了自己的 hyper::service::Service——去掉 poll_ready、call 改成 &self，关联类型仍是 Response / Error / Future 三个（下一章的主题）。

2、Incoming 取代 Body：0.14 的 hyper::Body 是一个通用的具体类型——既是入站 body 的类型，也常被拿来构造出站 body（Body::from(...)、Body::channel()）。1.x 改成 hyper::body::Incoming（源码内部以 IncomingBody 别名引用）、明确"这是从 connection 进来的 body"——用户的 response 用自己的 B: Body 实现。类型系统更精确、避免"我以为 Body 是 streaming 但实际是 buffer"的混淆。

3、runtime-agnostic 的 IO 抽象：0.14 直接依赖 tokio::io::AsyncRead / AsyncWrite 和 tokio 的定时器；1.x 换成自己的 hyper::rt::{Read, Write, Timer} trait，tokio 类型经 hyper-util 的 TokioIo / TokioTimer 适配——非 tokio runtime 也能用 hyper。（poll_loop 里的 task::yield_now 在 0.14 就有，不是 1.x 新加的。）

4、http-body 1.0 接口：0.14 用 http-body 0.4（poll_data + poll_trailers 两个方法）。1.x 用 http-body 1.0（单一 poll_frame，frame 是 Data / Trailers 的枚举）——trailers 与 data 走同一条流、并为未来扩展留位。

5、Conn 的配置项外移：0.14 的 Conn setter 和 1.x 大体相同、State 的 #[cfg(feature = "server")] / "client" 字段分隔在 0.14 也已存在；1.x 的变化是 server/client 的 Builder 拆到 server::conn::http1 / client::conn::http1 各自的模块里、Timer 成为显式配置。

这些演进反映了 hyper 团队在 0.14 → 1.0 的 3 年里积累的工程教训——什么该精简、什么该精确化、什么该 runtime-agnostic。任何长期维护的库都会经历这种"先 push 大胆新功能、后回头精炼"的循环。读 hyper 不同版本的 changelog 是看这种演进的最佳方式。

12.11.10 几条 HTTP/1 connection 的"反直觉"事实

读完本章你应该获得几条很反直觉但真实的事实——这些在普通 HTTP server 教程里不会讲：

1、HTTP/1.1 keep-alive 默认 ON、HTTP/1.0 默认 OFF——但很多生产 server 错误配置成 HTTP/1.0 风格（每请求一连接）、每个请求都得重付一次握手成本。判断标准是 response header 是否带 Connection: keep-alive——HTTP/1.1 不需要带、HTTP/1.0 必须带才生效。

2、Pipelining 在浏览器里几乎无用——RFC 7230 允许、但实际 Chrome/Firefox 默认禁用 pipelining（因为 head-of-line blocking 风险）。hyper 服务端能正确处理 pipelined 请求，但只是逐个串行处理：上一个请求的 Service future 没完成，dispatch.poll_ready 就是 Pending；上一个响应没写完，Reading 回不到 Init，下一个请求头都不会去解析（§12.6.2）。用 hyper 1.9.0 + tokio 1.46.1 实测，一次写入三个 pipelined 请求、handler 各 sleep 200ms，三个 handler 依次在 0/201/403ms 开始，响应按序返回。hyper 客户端则不会主动 pipeline——Client::should_poll 要求没有在等响应的 callback 才取下一个请求。§12.5.1 的 16 轮上限主要是给 pipelining benchmark 这类场景准备的。

3、Content-Length: 0 和省略 Content-Length 不一样——按 RFC 9112 §6.3，请求既没有 Content-Length 也没有 Transfer-Encoding 时 body 长度视为 0；但 RFC 9110 §8.6 建议方法语义带 content 的请求（POST/PUT）显式带 Content-Length，不少 server 会对没有它的 POST 直接回 411。hyper 的 server 侧（role.rs Server::encode）会给空 body 的响应自动写 content-length: 0（can_have_implicit_zero_content_length 允许时）；client 侧 Client::set_length 对 body: None 的请求不加这个头、只按长度 0 编码——发空 body 的 POST 想带 Content-Length: 0 就自己设、但理解协议帮你 debug 第三方 server 的 weird 行为。

4、Expect: 100-continue 在 HTTP/2 / HTTP/3 里仍然合法但用得少——RFC 9113 §8.1 和 RFC 9114 §4.1 都保留了 1xx 信息响应；这条特性服务上传大文件时"先 OK 再发 body"，HTTP/2 的流控让它不再那么必要。一些 HTTP/1 客户端还在用——hyper 的 Reading::Continue 状态专门处理。

5、Connection 数和并发数不一定相等——HTTP/1.1 即使 keep-alive、一条 connection 同时只处理一个请求；要并发就开多条连接。HTTP/2 一条 connection 多个 stream。所以"server 支持 10K 并发"对 HTTP/1 和 HTTP/2 是不同含义——HTTP/1 的 10K 并发 = 10K connection、HTTP/2 可能是 100 connection × 100 stream。

6、TCP keep-alive 不等于 HTTP keep-alive——TCP keep-alive 是 OS 级别的"心跳检测连接活着"、HTTP keep-alive 是 application 级别的"复用 connection 多个 request"。两者完全无关——很多新手混淆。本章讲的全部是 HTTP keep-alive。

这 6 条事实在生产 server 排错时反复出现——理解它们能让你少走很多弯路。

12.12 本章源码定位索引

为便于按图索骥（基于 hyper 1.9.0）：

主题 源文件 关键行号
Conn struct hyper-1.9.0/src/proto/h1/conn.rs 38-42
State + Reading/Writing/KA enum 同上 912-1028
BitAndAssign for KA 同上 1013-1020
try_keep_alive 4 路 同上 1072-1091
idle + Client/Server 特殊路径 同上 1104-1134
poll_read_head 含 timeout 同上 211-336
poll_read_keep_alive 三态 + require_empty_read / mid_message_detect_eof 同上 435-508
Dispatcher struct hyper-1.9.0/src/proto/h1/dispatch.rs 22-28
Dispatch trait 4 方法 同上 30-43
poll_loop 16 轮 cooperative yield 同上 165-193
poll_read 含取消处理 同上 195-269
poll_read_head + poll_ready 反压 同上 271-326
poll_write + body chunk 处理 同上 328-421
Server / Client 的 Dispatch impl 同上 521-580 / 598-702

12.13 给写自己的 connection-style codec 的 8 条工程启示

如果你将来写自己的协议库（不一定是 HTTP——可能是自定义二进制协议、WebSocket-like、proprietary RPC）——hyper 这套 Conn + Dispatcher 模式给你 8 条可移植的启示：

1、状态用 enum 而不是 boolean——boolean 让 N 种状态变成 2^N 个 bool 组合、容易出现"非法但没人检查"的状态。enum 强制所有状态被穷尽 match。本章 §12.2 的 Reading/Writing/KA 三 enum 就是这个原则。

2、运算符重载表达不变量——§12.2.4 的 BitAndAssign 让"单向锁"语义有代数感。如果你的状态有"一旦 disable 就永久 disable"性质——重载 &= 让代码读起来像"我在 AND 一个布尔"——意图清晰。

3、driver 和 state 分离——本章 §12.4 的 Conn (state) 和 Dispatcher (driver) 拆成两个 struct。Conn 不知道 driver 怎么 poll 它、Dispatcher 不直接读写 socket。职责分离让单元测试容易——Conn 可以脱离 driver 测、Dispatcher 则用 mock IO（tokio_test::io::Builder）测。

4、for _ in 0..N 的 cooperative yielding——§12.5 的 16 轮限制 + yield_now。任何 future 都不应该 starve 整个 executor——不管你的协议多简单、加这条防止饿死。

5、读写缓冲集中在 IO 层——hyper 的 Buffered<I, EncodedBuf> 一边用 read_buf 攒读到的字节、交给 parser 切分，一边用 WriteBuf 攒要写的内容：响应头序列化进一块可复用的 Vec<u8>，body 片段在 Queue 策略下排进 BufList、flush 时一次 vectored write，Flatten 策略下则拷进同一块缓冲（hyper-1.9.0/src/proto/h1/io.rs:32-40、:510-517）。把零碎的读写合并成少量系统调用，在性能敏感的协议库里很常见。

6、用 state.method / state.version 这种 per-message 字段——HTTP/1 的某些行为依赖前一条 message 的字段（比如 HEAD 请求的 response 不该带 body）。connection state 不仅记当前状态、还记跨 message 的契约信息——你写的协议如果有类似跨消息约束、用同样的字段。

7、错误分级：本章看到的几种错误——new_incomplete（mid-message EOF）、new_unexpected_message（idle 收到字节）、new_header_timeout（slowloris）、new_user_body（user 业务错误）、new_io（底层 IO 错）。每种错误有专属构造函数 + 不同诊断——比一刀切 Box<dyn Error> 强得多。

8、#[cfg(feature = "server")] / #[cfg(feature = "client")] 区分——避免 server 部署带 client 代码、反之亦然。用 cargo feature gate 精确控制 binary size——一些 hyper 用户只用 server、不会被 client 代码膨胀。

这 8 条不是 hyper 独家——任何成熟的协议库（h2、tokio-tungstenite、quinn）都用类似套路。协议库写作的"工程哲学"是相对稳定的——学会一种就能迁移到所有同类型项目。

12.14 落到你键盘上

  • 用 RUSTFLAGS="--cfg hyper_unstable_tracing" 编译 hyper 并开 tracing——你会看到 Reading::Init → Reading::Body → Reading::KeepAlive → Reading::Init 这样的状态转移日志。感受状态机实时运转。
  • 尝试手动触发 pipelined requests：用 nc 手写两个连在一起的 HTTP/1.1 请求（GET /a HTTP/1.1\r\nHost: x\r\n\r\nGET /b HTTP/1.1\r\nHost: x\r\n\r\n），观察 hyper 怎么处理。配合 handler 里打的时间戳（或上一条的 trace 日志）能看到：/b 要等 /a 的响应写完才开始处理——两个请求的字节一次就到了，第二个靠的正是 wants_read_again 再跑一轮。
  • 读 dispatch.rs 的 Client impl（598-702 行）——它和 Server impl 对称，但承载的是客户端侧的异步队列。读完一遍两者，你会对"同一个 Dispatcher 如何泛化到 Server/Client"有感性认识。

下一章，我们回到 trait 层面——看 hyper 1.x 为什么定义了自己的 Service trait，以及那个看似只差一个 mut 的差别（&self vs &mut self）背后是怎样一段漫长的讨论。