Rust in Everything
基于 hyper 1.12.0

HTTP/2 服务端

主线位置：服务端主线的 HTTP/2 版本。「字节 → 请求」交给 h2：它解析帧、拼出 http::Request；hyper 在连接任务里调用服务，把「等响应 → 写响应头 → 写 body」放进每个流自己的任务。HTTP/1 那条线上 Dispatcher 一个 future 包办的事，在这里分到了 1 + N 个任务里。

flowchart TB
  sc["serve_connection"] --> hs["State::Handshaking"]
  hs --> sv["State::Serving"]
  sv --> acc["poll_accept 收到新流"]
  acc --> callreq["service.call：连接任务里"]
  callreq --> spawn["execute_h2stream"]
  spawn --> st["H2Stream 任务：poll 服务 future"]
  st --> head["send_response 写响应头"]
  head --> pipe["PipeToSendStream 写 body"]
  acc --> acc
  sv --> gs["graceful_shutdown：GOAWAY"]

帧的编解码、HPACK、流状态、流控记账都在 h2 crate（本卷对照 h2 0.4.20）里。hyper 的 proto/h2/server.rs 只有 556 行，做三件事：握手、收流并把每个流交给执行器、把服务的响应写回 h2。

Builder 与 Connection：一层壳

http2::Builder 只有三个字段：

pub struct Builder<E> {
    exec: E,
    timer: Time,
    h2_builder: proto::h2::server::Config,
}

和 http1::Builder 不同，Builder::new 必须传一个执行器。timer 从 Time::Empty 开始，只有 keep-alive 非要它不可（BDP 只取 now()，没有 timer 时退回 Instant::now()），见 ping：保活与 BDP。serve_connection 用这三样建一个 proto::h2::Server，包进 Connection 返回，和 HTTP/1 一样不 spawn 任何东西。Connection::poll 把内部的 Dispatched 丢掉，只剩 Result<()>：HTTP/2 的 CONNECT 隧道只占一个流，不会把整条连接的 IO 交出去，所以内部永远只返回 Dispatched::Shutdown，留着这个类型是历史原因（源码里一条 TODO 这么说）。

执行器的约束写成 E: Http2ServerConnExec<S::Future, Bd>，它为什么是一个 sealed trait、用 TokioExecutor 时为什么服务的 future 和响应 body 都必须是 Send，见 Executor 与 Timer。hyper-util 的 auto::Builder 里的 HTTP/2 分支用的就是这个 Builder，见 auto：一个端口两个协议。

选项都写进 Config，默认值：

选项默认说明
initial_connection_window_size1 MiB规范默认是 65535，源码注释说这对性能太小
initial_stream_window_size1 MiB同上
max_frame_size16 KiB
max_send_buf_size400 KiB每个流在 h2 里缓冲待发数据的上限
max_header_list_size16 KiB
max_concurrent_streams2001.0.0 起（#3362），之前不限
max_local_error_reset_streams1024
keep_alive_interval无设了就要 timer
keep_alive_timeout20 秒
adaptive_window关打开时两个窗口都改成 65535，由 BDP 往上调
auto_date_header开

两个窗口选项和 adaptive_window 互相覆盖：调 initial_stream_window_size 会把 adaptive_window 关掉，调 adaptive_window(true) 会把两个窗口改回 SPEC_WINDOW_SIZE（65535）。后调的那个说了算。

Server：握手，然后收流

Server 的状态只有两个，State：

enum State<T, B>
where
    B: Body,
{
    Handshaking {
        ping_config: ping::Config,
        hs: Handshake<Compat<T>, SendBuf<B::Data>>,
    },
    Serving(Serving<T, B>),
}

Server::new 把 Config 一项项抄进 h2::server::Builder，然后 builder.handshake(Compat::new(io))。h2 要的是 tokio 的 AsyncRead / AsyncWrite，hyper 手里是自己的 rt::Read / rt::Write，Compat 就是两者之间的转接（两个 trait 为什么不同，见 IO trait）。h2 依赖 tokio 的 io-util，所以开了 http2 特性，依赖树里就会有 tokio 的 IO trait，但仍然没有 tokio 的运行时。

poll 是一个两步的状态机：

            let next = match &mut me.state {
                State::Handshaking { hs, ping_config } => {
                    let mut conn = ready!(Pin::new(hs).poll(cx).map_err(crate::Error::new_h2))?;
                    let ping = if ping_config.is_enabled() {
                        let pp = conn.ping_pong().expect("conn.ping_pong");
                        Some(ping::channel(pp, ping_config.clone(), me.timer.clone()))
                    } else {
                        None
                    };
                    State::Serving(Serving {
                        ping,
                        conn,
                        closing: None,
                        date_header: me.date_header,
                    })
                }
                State::Serving(srv) => {
                    // graceful_shutdown was called before handshaking finished,
                    if me.close_pending && srv.closing.is_none() {
                        srv.conn.graceful_shutdown();
                    }
                    ready!(srv.poll_server(cx, &mut me.service, &mut me.exec))?;
                    return Poll::Ready(Ok(Dispatched::Shutdown));
                }
            };

握手完成时，只要开了 keep-alive 或自适应窗口，就从 h2 连接上取出 PingPong，建 ping 通道；这一步会调 timer.sleep，没给 timer 就在这里 panic（Executor 与 Timer 的实验 1）。close_pending 见本页最后一节。

poll_server：一个流一个任务

poll_server 是服务端的主循环：

        if self.closing.is_none() {
            loop {
                self.poll_ping(cx);

                match ready!(self.conn.poll_accept(cx)) {
                    Some(Ok((req, mut respond))) => {
                        trace!("incoming request");
                        let content_length = headers::content_length_parse_all(req.headers());
                        let ping = self
                            .ping
                            .as_ref()
                            .map_or_else(ping::disabled, |ping| ping.0.clone());

                        // Record the headers received
                        ping.record_non_data();

                        let is_connect = req.method() == Method::CONNECT;
                        let (mut parts, stream) = req.into_parts();
                        let (mut req, connect_parts) = if !is_connect {
                            (
                                Request::from_parts(
                                    parts,
                                    IncomingBody::h2(stream, content_length.into(), ping),
                                ),
                                None,
                            )
                        } else {
                            // ... CONNECT：body 为空，extensions 里放 OnUpgrade
                        };

                        // ...
                        let fut = H2Stream::new(
                            service.call(req),
                            connect_parts,
                            respond,
                            self.date_header,
                            exec.clone(),
                        );

                        exec.execute_h2stream(fut);
                    }
                    Some(Err(e)) => {
                        return Poll::Ready(Err(crate::Error::new_h2(e)));
                    }
                    None => {
                        // no more incoming streams...
                        if let Some((ping, _)) = &self.ping {
                            ping.ensure_not_timed_out()?;
                        }

                        trace!("incoming connection complete");
                        return Poll::Ready(Ok(()));
                    }
                }
            }
        }

每圈先 poll_ping（keep-alive 和 BDP 的结果在这里生效，见 ping），再向 h2 要下一个新流。h2 的 poll_accept 在内部推动整条连接：读 socket、解帧、把 DATA 分给各个流、把各个流排队的帧写出去。所以这个 Connection future 虽然只「收流」，却是整条连接 IO 的驱动者，必须一直被 poll。

拿到一个新流，hyper 做的事：

  • 请求 body 包成 Incoming 的 h2 变体（IncomingBody::h2），带上 Content-Length 解析出的长度和一个 ping Recorder。
  • 调 service.call(req)，同步地，就在连接任务里。
  • 把返回的 future、h2 给的 SendResponse（respond）和一份执行器包成 H2Stream，交给执行器。

poll_accept 返回 None 表示连接上不会再有新流（对端关了、GOAWAY 走完了、或者 keep-alive 超时后 hyper 自己关了连接），poll_server 正常返回，Connection 以 Ok(()) 结束；唯一的例外是 keep-alive 超时，ensure_not_timed_out 把它变成错误。

实验：任务归属与并发

和 Dispatcher 那页的 HTTP/1 实验一样，三个请求、服务每个睡 200ms，这次走 HTTP/2，客户端用 hyper 自己的 HTTP/2 客户端在同一条连接上并发发出。两边的执行器都换成 LogExec：它在每次 execute 时打印 future 的类型名和 tokio 分配的任务 id。服务端把「call 在哪个任务里」和「返回的 future 在哪个任务里被 poll」分开打印。

hyper = { version = "=1.12.0", features = ["server", "client", "http2"] }
hyper-util = { version = "=0.1.21", features = ["tokio"] }
http-body-util = "0.1"
tokio = { version = "=1.54.1", features = ["full"] }

use std::{convert::Infallible, future::Future, time::{Duration, Instant}};
use http_body_util::{BodyExt, Empty, Full};
use hyper::{body::{Bytes, Incoming}, client, rt::Executor, server, service::service_fn, Request, Response};
use hyper_util::rt::TokioIo;
use tokio::net::{TcpListener, TcpStream};

// An executor that logs the type of every future hyper hands it.
#[derive(Clone)]
struct LogExec(&'static str);

impl<F> Executor<F> for LogExec
where
  F: Future + Send + 'static,
  F::Output: Send + 'static,
{
  fn execute(&self, fut: F) {
    let name = std::any::type_name::<F>();
    let short = name.split('<').next().unwrap().rsplit("::").next().unwrap();
    let id = tokio::spawn(fut).id();
    println!("       {} execute {short} -> task {id}", self.0);
  }
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
  let start = Instant::now();
  let ms = move || start.elapsed().as_millis();
  let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
  let addr = listener.local_addr().unwrap();

  tokio::spawn(async move {
    let (stream, _) = listener.accept().await.unwrap();
    println!("{:>4}ms server: connection task {}", ms(), tokio::task::id());
    let svc = service_fn(move |req: Request<Incoming>| {
      let path = req.uri().path().to_string();
      println!("{:>4}ms server: call {path} in task {}", ms(), tokio::task::id());
      async move {
        println!("{:>4}ms server: future {path} polled in task {}", ms(), tokio::task::id());
        tokio::time::sleep(Duration::from_millis(200)).await;
        println!("{:>4}ms server: handler {path} done", ms());
        Ok::<_, Infallible>(Response::new(Full::new(Bytes::from(path))))
      }
    });
    server::conn::http2::Builder::new(LogExec("server"))
      .serve_connection(TokioIo::new(stream), svc)
      .await
      .unwrap();
  });

  let stream = TcpStream::connect(addr).await.unwrap();
  let (sender, conn) = client::conn::http2::Builder::new(LogExec("client"))
    .handshake(TokioIo::new(stream))
    .await
    .unwrap();
  println!("       client: Connection future -> task {}", tokio::spawn(conn).id());

  let mut tasks = Vec::new();
  for path in ["/a", "/b", "/c"] {
    let mut sender = sender.clone(); // SendRequest is Clone for HTTP/2
    let task = tokio::spawn(async move {
      let req = Request::get(format!("http://x{path}")).body(Empty::<Bytes>::new()).unwrap();
      let res = sender.send_request(req).await.unwrap();
      let body = res.into_body().collect().await.unwrap().to_bytes();
      println!("{:>4}ms client: got {body:?}", ms());
    });
    println!("       client: request {path} -> task {}", task.id());
    tasks.push(task);
  }
  for t in tasks {
    t.await.unwrap();
  }
}

M2 Max、rustc 1.99.0、--release，单线程运行时。连跑 3 次，除毫秒数差 1ms 以内外完全相同：

       client execute H2ClientFuture -> task 2
       client: Connection future -> task 3
       client: request /a -> task 4
       client: request /b -> task 5
       client: request /c -> task 6
       client execute H2ClientFuture -> task 7
       client execute H2ClientFuture -> task 8
       client execute H2ClientFuture -> task 9
   1ms server: connection task 1
   1ms server: call /a in task 1
       server execute H2Stream -> task 10
   1ms server: call /b in task 1
       server execute H2Stream -> task 11
   1ms server: call /c in task 1
       server execute H2Stream -> task 12
   1ms server: future /a polled in task 10
   1ms server: future /b polled in task 11
   1ms server: future /c polled in task 12
 204ms server: handler /a done
 204ms server: handler /b done
 204ms server: handler /c done
 204ms client: got b"/a"
 205ms client: got b"/b"
 205ms client: got b"/c"

服务端这一半：

  • 连接 future 跑在任务 1。三个流的 call 都在任务 1 里执行，每次 call 之后紧跟一次 execute H2Stream，得到任务 10、11、12。
  • 服务返回的 future 分别在任务 10、11、12 里第一次被 poll，三个 sleep 同时开始，200ms 后同时结束。
  • 三个请求一共约 200ms。HTTP/1 的同一个实验要 600ms：/b 要等 /a 的响应写完才被调用。

客户端那一半（任务 2 是握手时 spawn 的连接任务，任务 7、8、9 是每个请求一个的等响应任务）在 HTTP/2 客户端 里讲。

📝NOTE
没有蠢问题

问： 每个流都有自己的任务了，那在服务里做一点同步的重活（解析大 JSON、算哈希）也不会影响别的请求吧？

答： 取决于活放在哪儿。service_fn 的闭包体、Service::call 的函数体，是在 poll_server 的循环里同步执行的，也就是在连接任务里：它跑多久，这条连接上的读、写、收新流就停多久，所有流一起等。只有 call 返回的 future 才在各自的 H2Stream 任务里被 poll。所以重活要放进 async 块里（或者交给 spawn_blocking），而 call 本身应该立即返回。tower 的服务经过 TowerToHyperService 适配后，call 里只有 clone 和构造 Oneshot（见 Service 与 Body），poll_ready 和 tower 服务自己的 call 都发生在 Oneshot 第一次被 poll 时，也就是在流任务里。

H2Stream：等响应，写响应头

H2Stream 是交给执行器的那个 future：

    pub struct H2Stream<F, B, E>
    where
        B: Body,
    {
        reply: SendResponse<SendBuf<B::Data>>,
        #[pin]
        state: H2StreamState<F, B>,
        date_header: bool,
        exec: E,
    }

state 先是 Service { fut, connect_parts }，拿到响应后变成 Body { pipe }。poll2 的前一半：

                    let res = match h.poll(cx) {
                        Poll::Ready(Ok(r)) => r,
                        Poll::Pending => {
                            // Response is not yet ready, so we want to check if the client has sent a
                            // RST_STREAM frame which would cancel the current request.
                            if let Poll::Ready(reason) =
                                me.reply.poll_reset(cx).map_err(crate::Error::new_h2)?
                            {
                                debug!("stream received RST_STREAM: {:?}", reason);
                                return Poll::Ready(Err(crate::Error::new_h2(reason.into())));
                            }
                            return Poll::Pending;
                        }
                        Poll::Ready(Err(e)) => {
                            let err = crate::Error::new_user_service(e);
                            warn!("http2 service errored: {}", err);
                            me.reply.send_reset(err.h2_reason());
                            return Poll::Ready(Err(err));
                        }
                    };

                    let (head, body) = res.into_parts();
                    let mut res = ::http::Response::from_parts(head, ());
                    super::strip_connection_headers(
                        res.headers_mut(),
                        super::MessageKind::Response,
                    );

                    // set Date header if it isn't already set if instructed
                    if *me.date_header {
                        res.headers_mut()
                            .entry(::http::header::DATE)
                            .or_insert_with(date::update_and_header_value);
                    }

                    // ... CONNECT 的成功响应：交出升级后的流，见下文

                    if !body.is_end_stream() {
                        // automatically set Content-Length from body...
                        if let Some(len) = body.size_hint().exact() {
                            headers::set_content_length_if_missing(res.headers_mut(), len);
                        }

                        let body_tx = reply!(me, res, false);
                        H2StreamState::Body {
                            pipe: PipeToSendStream::new(body, body_tx),
                        }
                    } else {
                        reply!(me, res, true);
                        return Poll::Ready(Ok(()));
                    }

几处值得看：

  • 服务 future 没好时，顺便盯着对端的 RST_STREAM。poll_reset 注册了 waker，客户端取消请求时这个任务会被唤醒，返回 Err，H2Stream 结束，服务的 future 随之被 drop。服务里正在等的数据库查询、下游请求也就跟着被取消。客户端那边发生了什么、线上的帧长什么样，见 HTTP/2 客户端 的实验。
  • 服务返回 Err，只重置这一个流。错误码是 h2_reason：错误链里有 h2::Error 就用它的原因，否则是 INTERNAL_ERROR。HTTP/1 里服务出错会让整条连接以错误结束（见 Dispatcher 的 poll_write），HTTP/2 不会，连接继续服务别的流。H2Stream::poll 把 poll2 的错误只记一条 debug! 就丢掉，它的 Output 是 ()，没人能收到这个错误。
  • 连接级的头被删掉。strip_connection_headers 删 keep-alive、proxy-connection、transfer-encoding、upgrade、响应里的 te，以及 connection 头和它列出的头名，每删一个记一条 warn!。HTTP/2 里这些信息由帧表达，RFC 9113 不允许消息里出现连接级的头。
  • body 长度：is_end_stream() 为真就把 END_STREAM 标在 HEADERS 帧上，没有 DATA 帧；否则 size_hint().exact() 有值时补 Content-Length（已经有就不动）。HTTP/2 没有 chunked，长度未知时什么也不加，靠 END_STREAM 结束。

实验：服务出错

为了看到线上的帧，后面几个实验都在客户端和服务端之间夹一个小代理 sniff.rs：原样转发字节，同时按 HTTP/2 帧头（3 字节长度、1 字节类型、1 字节标志、4 字节流 id）把指定类型的帧打出来，客户端方向先跳过 24 字节的连接前言。

// A one-connection TCP proxy that prints the HTTP/2 frame headers going
// through it. Only frame types listed in `show` are printed.
use std::{net::SocketAddr, sync::atomic::{AtomicUsize, Ordering}, time::Instant};
use tokio::{
  io::{AsyncReadExt, AsyncWriteExt},
  net::{tcp::{OwnedReadHalf, OwnedWriteHalf}, TcpListener, TcpStream},
};

/// Number of PING frames (not ACKs) seen from the client to the server.
pub static CLIENT_PINGS: AtomicUsize = AtomicUsize::new(0);

const NAMES: [&str; 10] = [
  "DATA", "HEADERS", "PRIORITY", "RST_STREAM", "SETTINGS",
  "PUSH_PROMISE", "PING", "GOAWAY", "WINDOW_UPDATE", "CONTINUATION",
];

fn reason(code: u32) -> String {
  match code {
    0 => "NO_ERROR".into(),
    2 => "INTERNAL_ERROR".into(),
    7 => "REFUSED_STREAM".into(),
    8 => "CANCEL".into(),
    c => format!("code {c}"),
  }
}

fn describe(kind: &str, flags: u8, payload: &[u8]) -> String {
  let u32_at = |i: usize| u32::from_be_bytes(payload[i..i + 4].try_into().unwrap());
  match kind {
    "DATA" => format!("len={}{}", payload.len(), if flags & 1 != 0 { " END_STREAM" } else { "" }),
    "HEADERS" => (if flags & 1 != 0 { "END_STREAM" } else { "" }).into(),
    "RST_STREAM" => reason(u32_at(0)),
    "GOAWAY" => format!("last_stream={} {}", u32_at(0) & 0x7fff_ffff, reason(u32_at(4))),
    "PING" | "SETTINGS" if flags & 1 != 0 => "ACK".into(),
    "SETTINGS" => payload
      .chunks(6)
      .map(|p| format!("{}={}", u16::from_be_bytes([p[0], p[1]]), u32::from_be_bytes([p[2], p[3], p[4], p[5]])))
      .collect::<Vec<_>>()
      .join(" "),
    "WINDOW_UPDATE" => format!("+{}", u32_at(0)),
    _ => String::new(),
  }
}

async fn pump(mut from: OwnedReadHalf, mut to: OwnedWriteHalf, dir: &'static str, start: Instant, show: &'static [&'static str]) {
  let mut buf = Vec::new();
  let mut skip = if dir == "c->s" { 24 } else { 0 }; // client connection preface
  let mut chunk = [0u8; 16384];
  loop {
    let n = match from.read(&mut chunk).await {
      Ok(0) | Err(_) => {
        let _ = to.shutdown().await;
        if show.contains(&"EOF") {
          println!("{:>5}ms {dir} EOF", start.elapsed().as_millis());
        }
        return;
      }
      Ok(n) => n,
    };
    if to.write_all(&chunk[..n]).await.is_err() {
      return;
    }
    buf.extend_from_slice(&chunk[..n]);
    if skip > 0 {
      let k = skip.min(buf.len());
      buf.drain(..k);
      skip -= k;
    }
    while buf.len() >= 9 {
      let len = u32::from_be_bytes([0, buf[0], buf[1], buf[2]]) as usize;
      if buf.len() < 9 + len {
        break;
      }
      let kind = NAMES.get(buf[3] as usize).copied().unwrap_or("UNKNOWN");
      let flags = buf[4];
      let stream = u32::from_be_bytes([buf[5], buf[6], buf[7], buf[8]]) & 0x7fff_ffff;
      if kind == "PING" && flags & 1 == 0 && dir == "c->s" {
        CLIENT_PINGS.fetch_add(1, Ordering::SeqCst);
      }
      if show.contains(&kind) {
        let d = describe(kind, flags, &buf[9..9 + len]);
        println!("{:>5}ms {dir} {kind} stream={stream} {d}", start.elapsed().as_millis());
      }
      buf.drain(..9 + len);
    }
  }
}

/// Listens on 127.0.0.1:0, forwards the first accepted connection to
/// `upstream`, and returns the proxy's address.
pub async fn proxy(upstream: SocketAddr, start: Instant, show: &'static [&'static str]) -> SocketAddr {
  let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
  let addr = listener.local_addr().unwrap();
  tokio::spawn(async move {
    let (client, _) = listener.accept().await.unwrap();
    let server = TcpStream::connect(upstream).await.unwrap();
    let (cr, cw) = client.into_split();
    let (sr, sw) = server.into_split();
    tokio::spawn(pump(cr, sw, "c->s", start, show));
    tokio::spawn(pump(sr, cw, "s->c", start, show));
  });
  addr
}

（CLIENT_PINGS 给 ping 那页的实验计数用。）依赖和上一个实验相同。服务对 /err 返回 Err，客户端先请求 /err 再请求 /ok：

use std::time::Instant;
use http_body_util::Empty;
use hyper::{body::{Bytes, Incoming}, client, server, service::service_fn, Request, Response};
use hyper_util::rt::{TokioExecutor, TokioIo};
use tokio::net::{TcpListener, TcpStream};

#[path = "../sniff.rs"]
mod sniff;

#[tokio::main(flavor = "current_thread")]
async fn main() {
  let start = Instant::now();
  let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
  let proxy = sniff::proxy(listener.local_addr().unwrap(), start, &["HEADERS", "RST_STREAM", "GOAWAY"]).await;
  tokio::spawn(async move {
    let (stream, _) = listener.accept().await.unwrap();
    let svc = service_fn(|req: Request<Incoming>| async move {
      if req.uri().path() == "/err" {
        return Err("handler failed");
      }
      Ok(Response::new(Empty::<Bytes>::new()))
    });
    let r = server::conn::http2::Builder::new(TokioExecutor::new())
      .serve_connection(TokioIo::new(stream), svc)
      .await;
    println!("server conn: {r:?}");
  });

  let stream = TcpStream::connect(proxy).await.unwrap();
  let (mut sender, conn) = client::conn::http2::handshake(TokioExecutor::new(), TokioIo::new(stream))
    .await
    .unwrap();
  tokio::spawn(conn);
  for path in ["/err", "/ok"] {
    let req = Request::get(format!("http://x{path}")).body(Empty::<Bytes>::new()).unwrap();
    match sender.send_request(req).await {
      Ok(res) => println!("client: {path} -> {}", res.status()),
      Err(e) => println!("client: {path} -> Err({e}: {})", std::error::Error::source(&e).unwrap()),
    }
  }
}

连跑 3 次，除毫秒数外相同：

    1ms c->s HEADERS stream=1 END_STREAM
    1ms s->c RST_STREAM stream=1 INTERNAL_ERROR
client: /err -> Err(http2 error: stream error received: unexpected internal error encountered)
    2ms c->s HEADERS stream=3 END_STREAM
    2ms s->c HEADERS stream=3 END_STREAM
client: /ok -> 200 OK

服务的错误在线上只是流 1 上的一个 RST_STREAM(INTERNAL_ERROR)，错误内容（handler failed）不会发给客户端。流 3 照常得到 200，没有 GOAWAY，server conn 那行没有打印，因为程序结束时服务端连接还活着。/ok 的空 body 让响应只有一个带 END_STREAM 的 HEADERS 帧。

PipeToSendStream：按流控容量写 body

响应 body 不为空时，H2Stream 进入 Body 状态，后面的事全交给 PipeToSendStream。客户端发请求 body 用的也是它，所以它在 mod.rs 里而不在 server.rs 里。

    pub(crate) struct PipeToSendStream<S>
    where
        S: Body,
    {
        body_tx: SendStream<SendBuf<S::Data>>,
        data_done: bool,
        // A data chunk that has been polled from the body but is still waiting
        // for stream-level capacity before it can be shipped. Stored here so
        // it survives across `Poll::Pending` returns from `poll_capacity`; if
        // we left the chunk in a local, it would be dropped on every repoll.
        buffered_data: Option<Peeked<S::Data>>,
        #[pin]
        stream: S,
    }

HTTP/2 的发送受两层窗口限制：每个流一个，整条连接一个，都由对端用 WINDOW_UPDATE 放大。h2 把这个约束暴露成「容量」：reserve_capacity(n) 表示想发 n 字节，poll_capacity 等 h2 从窗口里分到容量，send_data 把数据交给 h2 排队。poll 每圈的顺序：

  1. poll_reset：对端重置了这个流就停下，返回错误。等数据、等容量时都注册着这个 waker。
  2. 手里有一块等容量的数据（buffered_data）：capacity() 为 0 就 poll_capacity，分到容量后 send_data。
  3. 否则 poll_frame 向 body 要下一帧：
            match ready!(me.stream.as_mut().poll_frame(cx)) {
                Some(Ok(frame)) => {
                    if frame.is_data() {
                        let chunk = frame.into_data().unwrap_or_else(|_| unreachable!());
                        let is_eos = me.stream.is_end_stream();
                        let len = chunk.remaining();
                        // ...
                        if len == 0 {
                            // ... 空数据帧不需要容量，直接 send_data
                        }

                        // ...
                        me.body_tx.reserve_capacity(1);
                        *me.buffered_data = Some(Peeked {
                            data: chunk,
                            is_eos,
                        });
                    } else if frame.is_trailers() {
                        // no more DATA, so give any capacity back
                        me.body_tx.reserve_capacity(0);
                        me.body_tx
                            .send_trailers(frame.into_trailers().unwrap_or_else(|_| unreachable!()))
                            .map_err(crate::Error::new_body_write)?;
                        return Poll::Ready(Ok(()));
                    } else {
                        trace!("discarding unknown frame");
                        // loop again
                    }
                }
                Some(Err(e)) => return Poll::Ready(Err(me.body_tx.on_user_err(e))),
                None => {
                    // no more frames means we're done here
                    // but at this point, we haven't sent an EOS DATA, or
                    // any trailers, so send an empty EOS DATA.
                    return Poll::Ready(me.body_tx.send_eos_frame());
                }
            }
  • 先拿到数据，再要容量，而且只要 1 字节。h2 在 send_data 时会把请求的容量提高到实际缓冲的长度，所以对端最终看到的需求不变；下面的冷知识讲为什么是这个顺序、这个数。
  • body 出错：on_user_err 用 h2_reason 给这个流发 RST_STREAM，对端看到的是流被重置，不是一个截断的 body。
  • body 结束但最后一帧没带结束标记：补一个空的、带 END_STREAM 的 DATA 帧。

背压来自两个方向。发送方向是这里：对端不放大窗口，poll_capacity 一直 Pending，PipeToSendStream 不再 poll_frame，响应 body 也就不再被拉取。接收方向在 Incoming 里：h2 变体每交出一块数据才对 h2 调一次 release_capacity，服务不读请求 body，窗口就不会被放大，对端最多发完一个窗口就得停（见 Incoming 与 body 通道）。HTTP/1 靠的是「不读 socket」，这里靠的是 HTTP/2 自己的流控，同一条连接上其他流不受影响，只要连接级窗口还有余量。

📝NOTE
冷知识：先要 1 字节容量，曾经让第二个流饿死

#4003（issue）报告生产环境里偶发死锁：一条 HTTP/2 连接发出「exactly 65534 bytes of body data」之后，第二个请求发不出去。原因是 PipeToSendStream「always keeps reserved one byte of flow control capacity while its waiting for more data from the body」：当时每圈一开始就 reserve_capacity(1)，再去问 body 要数据。body 暂时没数据时，这 1 字节一直从连接级窗口里占着；对端如果只在窗口完全耗尽时才发 WINDOW_UPDATE（issue 里举的是 Bun 的 HTTP/2 服务端），它眼里窗口还剩 1 字节，永远不会放大，第二个流就一个字节也拿不到。

#4061（v1.10.0，「do not reserve capacity before body data is available」）把顺序倒过来：先 poll_frame，拿到数据块后再按数据块的长度预留，数据块存进新加的 buffered_data，「so it survives the poll_capacity wait across Poll::Pending returns」。

#4149（v1.11.1，「reserve minimal send capacity when piping request bodies」）又把预留量从整块数据改回 1 字节。PR 说按整块预留在一个压测环境里带来「a ~10% increase in CPU time per request and a ~12% drop in throughput」，只在连接上的流加起来要的容量超过对端窗口时出现；死锁的修复不受影响，因为「The reservation is still made only once the chunk is in hand」。上面代码里的 reserve_capacity(1) 和 buffered_data 就是这两次改动的结果。

CONNECT：流变成 IO

HTTP/2 没有 Upgrade 头，隧道用 CONNECT 方法在一个流上建立（RFC 8441 的扩展 CONNECT 要 enable_connect_protocol() 打开，:protocol 伪头以 hyper::ext::Protocol 出现在 extensions 里）。poll_server 对所有 CONNECT 请求都放一个 OnUpgrade，请求 body 给空的 Incoming，带非零 Content-Length 的 CONNECT 直接 RST_STREAM(INTERNAL_ERROR)。服务回 2xx 时，H2Stream 删掉 Content-Length、写响应头（不带 END_STREAM），然后把 h2 的 SendStream 和 RecvStream 拼成一个 IO（proto/h2/upgrade.rs），装进 Upgraded 交给等待者，同时 execute_upgrade 再 spawn 一个任务。HTTP/1 要显式 with_upgrades() 才会交出 IO（见 协议升级），HTTP/2 不用，因为交出去的只是一个流，连接还在 hyper 手里。

graceful_shutdown：两次 GOAWAY

Connection::graceful_shutdown 转到 Server::graceful_shutdown：

    pub(crate) fn graceful_shutdown(&mut self) {
        trace!("graceful_shutdown");
        match &mut self.state {
            State::Handshaking { .. } => {
                self.close_pending = true;
            }
            State::Serving(srv) => {
                if srv.closing.is_none() {
                    srv.conn.graceful_shutdown();
                }
            }
        }
    }

握手还没完成时只记一个 close_pending，等进入 Serving 后的第一次 poll 再真正开始（前面 poll 代码里那个分支）。这是 v1.5.1 的 #3729 改的，提交说明写的是之前「the connection would just be dropped instantly」，握手期间已经在路上的请求跟着丢了。

真正的关闭流程在 h2 里。h2 0.4.20 的 graceful_shutdown 照 RFC 9113 的建议，先发一个 last_stream_id = 2^31-1、NO_ERROR 的 GOAWAY，告诉客户端别再开新流；再发一个 PING，收到 PONG（即至少过了一个往返，路上的新流都已到达）后发第二个 GOAWAY，带上真实的最后一个流 id。之后 poll_accept 不再给出新流，已有的流照常跑完，全部结束后连接关闭，poll_accept 返回 None，Connection 以 Ok(()) 完成。和 HTTP/1 一样，graceful_shutdown 只改状态，之后必须继续 poll Connection。

实验：在途的请求和迟到的请求

服务每个请求睡 300ms。客户端在 1ms 发 /first；服务端在 100ms 调 graceful_shutdown；客户端在 200ms 用同一个 SendRequest 再发 /late。依赖同上。

use std::{convert::Infallible, pin::pin, time::{Duration, Instant}};
use http_body_util::{Empty, Full};
use hyper::{body::{Bytes, Incoming}, client, server, service::service_fn, Request, Response};
use hyper_util::rt::{TokioExecutor, TokioIo};
use tokio::net::{TcpListener, TcpStream};

#[path = "../sniff.rs"]
mod sniff;

#[tokio::main(flavor = "current_thread")]
async fn main() {
  let start = Instant::now();
  let ms = move || start.elapsed().as_millis();
  let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
  let show = &["HEADERS", "DATA", "RST_STREAM", "GOAWAY", "PING", "EOF"];
  let proxy = sniff::proxy(listener.local_addr().unwrap(), start, show).await;

  let server = tokio::spawn(async move {
    let (stream, _) = listener.accept().await.unwrap();
    let svc = service_fn(move |req: Request<Incoming>| async move {
      let path = req.uri().path().to_string();
      println!("{:>5}ms server: handler {path} start", ms());
      tokio::time::sleep(Duration::from_millis(300)).await;
      Ok::<_, Infallible>(Response::new(Full::new(Bytes::from(path))))
    });
    let mut conn = pin!(server::conn::http2::Builder::new(TokioExecutor::new())
      .serve_connection(TokioIo::new(stream), svc));
    tokio::select! {
      r = conn.as_mut() => return r,
      _ = tokio::time::sleep(Duration::from_millis(100)) => {}
    }
    println!("{:>5}ms server: graceful_shutdown()", ms());
    conn.as_mut().graceful_shutdown();
    conn.await
  });

  let stream = TcpStream::connect(proxy).await.unwrap();
  let (mut sender, conn) = client::conn::http2::handshake(TokioExecutor::new(), TokioIo::new(stream))
    .await
    .unwrap();
  let client_conn = tokio::spawn(async move {
    let r = conn.await;
    println!("{:>5}ms client conn: {r:?}", ms());
  });

  let req = |p: &str| Request::get(format!("http://x{p}")).body(Empty::<Bytes>::new()).unwrap();
  let mut s1 = sender.clone();
  let first = tokio::spawn(async move { s1.send_request(req("/first")).await.map(|r| r.status()) });

  tokio::time::sleep(Duration::from_millis(200)).await;
  println!("{:>5}ms client: is_closed={}", ms(), sender.is_closed());
  let late = sender.send_request(req("/late")).await;
  println!("{:>5}ms client: /late -> {late:?}", ms());
  let first = first.await.unwrap();
  println!("{:>5}ms client: /first -> {first:?}", ms());
  let server = server.await.unwrap();
  println!("{:>5}ms server conn: {server:?}", ms());
  client_conn.await.unwrap();
}

连跑 3 次，顺序相同，毫秒数差 2ms 以内：

    1ms c->s HEADERS stream=1 END_STREAM
    1ms server: handler /first start
  104ms server: graceful_shutdown()
  104ms s->c GOAWAY stream=0 last_stream=2147483647 NO_ERROR
  104ms s->c PING stream=0 
  104ms c->s PING stream=0 ACK
  104ms s->c GOAWAY stream=0 last_stream=1 NO_ERROR
  202ms client: is_closed=false
  202ms client conn: Ok(())
  202ms client: /late -> Err(hyper::Error(Canceled, "connection closed"))
  304ms s->c HEADERS stream=1 
  304ms s->c DATA stream=1 len=6 END_STREAM
  304ms s->c EOF
  304ms client: /first -> Ok(200)
  304ms server conn: Ok(())
  • 104ms 的四个帧就是上面说的两阶段：GOAWAY(2147483647)、PING、客户端的 PING ACK、GOAWAY(1)。本地回环上一个往返不到 1ms，两个 GOAWAY 几乎同时出现。
  • /first（流 1）不受影响，300ms 后照常拿到 200。最后一个流结束后服务端关掉连接（s->c EOF），Connection 以 Ok(()) 结束。
  • /late 根本没上线：线上没有流 3 的 HEADERS。客户端在 200ms 时 is_closed() 还是 false，发出 /late 后它的连接 future 才发现连接已经不接受新流，以 Ok(()) 结束，/late 收到 Canceled 错误。这一段为什么是这样，见 HTTP/2 客户端 的「收到 GOAWAY 之后」。

hyper-util 的 GracefulShutdown 对 HTTP/2 连接调用的就是这个方法，「什么时候调、怎么等所有连接结束」在 优雅停机 里。keep-alive 超时走的是另一条路：hyper 调 h2 的 abrupt_shutdown(NO_ERROR)，只发一个 GOAWAY 就关，见 ping。

小结

  • http2::Builder 必须带执行器；serve_connection 只组装 proto::h2::Server，不 spawn。默认窗口 1 MiB、最大并发流 200，默认值不算稳定 API。
  • Server 先握手（IO 经 Compat 转成 tokio 的 IO trait 交给 h2），再在 poll_server 里循环 poll_accept。这个 future 驱动整条连接的 IO，必须一直 poll。
  • 每个新流：service.call 在连接任务里同步执行，返回的 future 包成 H2Stream 交给执行器。实验里三个 200ms 的请求一共 200ms，HTTP/1 要 600ms。call 本身要快，重活放进它返回的 future。
  • H2Stream 等服务 future 时监听 RST_STREAM；服务返回 Err 只给这个流发 RST_STREAM（默认 INTERNAL_ERROR），连接继续服务别的流。
  • PipeToSendStream 先从 body 拿到数据，再向 h2 要 1 字节容量、等分配、send_data；这个顺序是修死锁（#4003）的结果。
  • graceful_shutdown 交给 h2：GOAWAY(2^31-1) → PING → GOAWAY(最后流 id)，在途的流跑完再关；握手中调用会推迟到握手完成。

💬 关联讨论

发起讨论
加载中...