协议升级

主线位置：服务端主线的一条岔路。正常情况下一个请求处理完，连接回到 Init 等下一个请求；升级时，101 响应写完以后这条连接不再说 HTTP，Dispatcher 退出，IO 交给服务自己。客户端主线上是对称的：收到 101（或 CONNECT 的 2xx）后，Connection 任务把 IO 交给拿着响应的一方。

sequenceDiagram
  participant C as 客户端
  participant D as Dispatcher
  participant S as 服务
  participant U as 升级任务
  C->>D: 请求，带 Upgrade 头
  D->>D: prepare_upgrade：Pending 留在 State
  D->>S: call，extensions 里有 OnUpgrade
  S->>U: spawn，await upgrade::on
  S-->>D: 响应 101
  D->>C: 写出 101 并刷新
  D->>D: 读写都 Closed，返回 Upgrade
  D->>U: fulfill：IO 加剩余字节
  U->>C: 之后说别的协议

hyper 不认识 WebSocket 或任何升级后的协议。它只负责三件事：认出「这个请求可能要升级」，在响应写完后停止解析 HTTP，把 IO 原样交出去，连同已经读进缓冲区、但不属于 HTTP 的那些字节。

起点：解析时认出升级请求

role::Server::parse 里，CONNECT 方法一开始就算升级（wants_upgrade 的初值），带 Upgrade 头的 HTTP/1.1 请求也算：

                header::UPGRADE => {
                    // Upgrades are only allowed with HTTP/1.1
                    wants_upgrade = is_http_11;
                }

Conn::poll_read_head 把它变成 Wants::UPGRADE，Dispatcher 的 poll_read_head 看到这个标志，在调用服务之前往请求头的 extensions 里插一个 OnUpgrade：

                if wants.contains(Wants::UPGRADE) {
                    let upgrade = self.conn.on_upgrade();
                    debug_assert!(!upgrade.is_none(), "empty upgrade");
                    debug_assert!(
                        head.extensions.get::<OnUpgrade>().is_none(),
                        "OnUpgrade already set"
                    );
                    head.extensions.insert(upgrade);
                }

on_upgrade 调 State::prepare_upgrade，用 pending() 建一对 tokio oneshot：

pub(super) fn pending() -> (Pending, OnUpgrade) {
    let (tx, rx) = oneshot::channel();
    (
        Pending { tx },
        OnUpgrade {
            rx: Some(Arc::new(Mutex::new(rx))),
        },
    )
}

发送端 Pending 留在 Conn 的 State.upgrade 里，接收端 OnUpgrade 随请求交给服务。这一步只是「准备」：服务回不回 101、升不升级，此时还不知道。

客户端方向在 role::Client::decoder 里：响应是 101，或者请求是 CONNECT 且响应是 2xx，就算升级，OnUpgrade 放进响应的 extensions。客户端的 Connection 同样要 with_upgrades() 才会交出 IO，见 conn::http1 客户端。

upgrade::on：从 extensions 里取出来

服务拿到请求后，用 upgrade::on 取出这个 future。它对 Request、&mut Request、Response、&mut Response 都能用，四个实现都一样（CanUpgrade）：

    impl<B> CanUpgrade for &'_ mut http::Request<B> {
        fn on_upgrade(self) -> OnUpgrade {
            self.extensions_mut()
                .remove::<OnUpgrade>()
                .unwrap_or_else(OnUpgrade::none)
        }
    }

是 remove，所以第二次调用拿到的是 OnUpgrade::none()。传 &mut req 的好处是请求本身还在，服务可以接着读它的头。

OnUpgrade 是一个 future，poll 有四种结局：

情况	结果	错误的 `Display`
extensions 里没有（请求不是升级请求，或者已经被取走）	立即 `Err`	`no upgrade available`
`Pending::fulfill` 送来了 IO	`Ok(Upgraded)`	—
`Pending::manual` 送来了错误（连接没用 `with_upgrades`）	`Err`	`upgrade expected but low level API in use`
`Pending` 没送任何东西就被 drop	`Err`，`source()` 是 `upgrade expected but not completed`	`operation was canceled`

OnUpgrade 里是 Arc<Mutex<oneshot::Receiver>> 而不是一个裸的 Receiver，因为它要放进 http::Extensions，而 http 1.x 的 Extensions::insert 要求 T: Clone + Send + Sync + 'static。oneshot::Receiver 不能 Clone，hyper 升级到 http 1.0 时（#3406，v1.0.0，提交说明「To do so, implements Clone for OnUpgrade」）给它套上了 Arc<Mutex<…>>。

📝NOTE

冷知识：升级的入口曾经在 Body 上

0.14 之前，拿升级 future 的方法是 Body::on_upgrade()。#2086 列了它的两个问题：「It makes the Body type fatter」，以及它要求调用方一直留着 Body，「they can't use stream combinators to read the body first, and then wait for the upgrade to finish」，想把 Request<Body> 转成别的 body 类型也很麻烦。提议是改成一个自由函数，参数是请求或响应本身，状态放进 http::Extensions。

#2337（v0.14.0）做了对外的部分：「The method Body::on_upgrade() is gone. It is essentially replaced with hyper::upgrade::on(msg)」。#2353（v0.14.0）做了内部的部分：给内部的 MessageHead 加了 extensions 字段，proto::h1 把升级状态从 body 移到头上。上面 poll_read_head 里的 head.extensions.insert(upgrade) 和 CanUpgrade 的 remove::<OnUpgrade>() 就是这次改动的结果。

101 写出去之后：读写两端都关

服务回 101 Switching Protocols，role::Server::encode 把它标成最后一条消息（encode）：

        let (ret, is_last) = if msg.head.subject == StatusCode::SWITCHING_PROTOCOLS {
            (Ok(()), true)
        } else if msg.req_method == &Some(Method::CONNECT) && msg.head.subject.is_success() {
            // Sending content-length or transfer-encoding header on 2xx response
            // to CONNECT is forbidden in RFC 7231.
            wrote_len = true;
            (Ok(()), true)
        } else if msg.head.subject.is_informational() {
            // ...

CONNECT 的 2xx 也是 is_last，并且不写 Content-Length / Transfer-Encoding。之后的过程完全是 Conn 的读写状态机的常规路径：头写完，写端 Closed；升级请求一般没有 body，读端早已是 KeepAlive；刷新成功后 try_keep_alive 看到「读 KeepAlive、写 Closed」，两端都关。Dispatcher::is_done 为真，poll_inner 先查有没有挂着的升级：

        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 {
                // ...

有就返回 Dispatched::Upgrade(pending)，不对 IO 调 poll_shutdown。try_keep_alive 在刷新成功之后才调，所以 101 的最后一个字节一定已经写出去了，IO 交出去时写缓冲区是空的。

谁来交出 IO：with_upgrades

Dispatcher 只把 Pending 交到外层，IO 还在它自己身上。外层有两种。

普通的 Connection，它的 poll：

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        match ready!(Pin::new(&mut self.conn).poll(cx)) {
            Ok(done) => {
                match done {
                    proto::Dispatched::Shutdown => {}
                    proto::Dispatched::Upgrade(pending) => {
                        // With no `Send` bound on `I`, we can't try to do
                        // upgrades here. In case a user was trying to use
                        // `Body::on_upgrade` with this API, send a special
                        // error letting them know about that.
                        pending.manual();
                    }
                }
                Poll::Ready(Ok(()))
            }
            Err(e) => Poll::Ready(Err(e)),
        }
    }

它调 manual，等升级的一方收到 upgrade expected but low level API in use。注释给了原因：Connection 对 I 没有 Send 约束，而 Upgraded 里装的是 Box<dyn Io + Send>，造不出来。注释里的 Body::on_upgrade 是上面冷知识里那个旧 API 的名字。

with_upgrades 返回的 UpgradeableConnection，它的 poll 要求 I: Read + Write + Unpin + Send + 'static：

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        if let Some(conn) = self.inner.as_mut() {
            match ready!(Pin::new(&mut conn.conn).poll(cx)) {
                Ok(proto::Dispatched::Shutdown) => Poll::Ready(Ok(())),
                Ok(proto::Dispatched::Upgrade(pending)) => {
                    let (io, buf, _) = self
                        .inner
                        .take()
                        .expect("upgradeable server connection missing after upgrade")
                        .conn
                        .into_inner();
                    pending.fulfill(Upgraded::new(io, buf));
                    Poll::Ready(Ok(()))
                }
                Err(e) => Poll::Ready(Err(e)),
            }
        } else {
            // inner is `None`, meaning the connection was upgraded, thus it's `Poll::Ready(Ok(()))`
            Poll::Ready(Ok(()))
        }
    }

into_inner 拆出 IO 和读缓冲（Buffered::into_inner 把 BytesMut 冻结成 Bytes），服务被丢掉；fulfill 经 oneshot 把 Upgraded 送给等待的一方。UpgradeableConnection 自己随即以 Ok(()) 完成：从这一刻起，这条连接不再归 hyper 管，它的生命由拿到 Upgraded 的那个任务决定。

还有一条旧路：poll_without_shutdown / without_shutdown 加 into_parts，在连接「HTTP 意义上结束」时把 IO、读缓冲和服务一起拿回来，它内部同样对 Upgrade 调 manual()。这条路属于 Connection 的 API，见服务端 Builder 与 Connection。hyper-util 的 auto::Builder 也有升级版本的入口，见 auto：一个端口两个协议。

Upgraded：Rewind 加类型擦除

Upgraded：

pub struct Upgraded {
    io: Rewind<Box<dyn Io + Send>>,
}

两层包装各管一件事。

Rewind 管多读的字节。 客户端完全可以把升级请求和新协议的第一批字节一次写过来（升级到 WebSocket 时，客户端发完请求通常要等 101，但协议没有禁止）；服务端读请求头时，这些字节一起进了 hyper 的读缓冲。直接把 socket 交出去，它们就丢了。Rewind 把它们放在 pre 里，poll_read 先吐 pre，吐完才读底层 IO：

    fn poll_read(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        mut buf: ReadBufCursor<'_>,
    ) -> Poll<io::Result<()>> {
        if let Some(mut prefix) = self.pre.take() {
            // If there are no remaining bytes, let the bytes get dropped.
            if !prefix.is_empty() {
                let copy_len = cmp::min(prefix.len(), buf.remaining());
                // TODO: There should be a way to do following two lines cleaner...
                buf.put_slice(&prefix[..copy_len]);
                prefix.advance(copy_len);
                // Put back what's left
                if !prefix.is_empty() {
                    self.pre = Some(prefix);
                }

                return Poll::Ready(Ok(()));
            }
        }
        Pin::new(&mut self.inner).poll_read(cx, buf)
    }

写方向原样转发。hyper-util 的 auto 用同一个思路把判断协议时「偷看」过的字节（最多 24 个）塞回去，见 auto：一个端口两个协议。

Box<dyn Io + Send> 管类型。 OnUpgrade 的输出类型不能带 I：它装在 http::Extensions 里，服务拿到的是 Request<Incoming>，类型上不知道连接的 IO 是什么。于是 IO 被装箱成 trait object，Upgraded 自己实现 rt::Read / rt::Write，可以直接当 IO 用。

知道具体类型时，可以用 downcast 拆回来：

    pub fn downcast<T: Read + Write + Unpin + 'static>(self) -> Result<Parts<T>, Self> {
        let (io, buf) = self.io.into_inner();
        match io.__hyper_downcast() {
            Ok(t) => Ok(Parts {
                io: *t,
                read_buf: buf,
            }),
            Err(io) => Err(Upgraded {
                io: Rewind::new_buffered(io, buf),
            }),
        }
    }

判断靠 TypeId：私有 trait Io 有一个默认方法 __hyper_type_id 返回 TypeId::of::<Self>()，__hyper_downcast 比对相等后用裸指针转换，和 std::error::Error::downcast 的做法一样（源码注释原话「Taken from std::error::Error::downcast()」）。TypeId 要求 'static，这就是 with_upgrades 要 I: 'static 的原因。类型必须和交给 serve_connection 的完全一致：传的是 TokioIo<TcpStream>，就要 downcast 成 TokioIo<TcpStream>，不是 TcpStream。失败时 Upgraded 原样还回来，pre 里的字节也还在。

拆出来的 Parts 有 io 和 read_buf 两个字段（#[non_exhaustive]，以后可能加字段）。read_buf 里就是那些多读的字节，绕过 Rewind 直接用 io 时，要自己先处理它们。

📝NOTE

没有蠢问题

问：服务返回 101 的那一刻，升级就完成了吗？为什么还要另起一个任务去 await OnUpgrade？

答：没有。服务返回的只是一个 Response，它还要经过 Dispatcher 编码、写进缓冲区、刷到 socket，然后读写两端关闭，Dispatcher 才退出，IO 才空出来。这些都发生在连接的任务里，而且发生在服务的 future 完成之后。如果在服务的 future 里直接 await OnUpgrade，就成了服务等连接交出 IO，连接等服务给出响应，互相等。所以 upgrade 模块文档的说法是：调用 on()，然后「can spawn a task awaiting it」。

实验：三种连接，同一个服务

服务对每个请求都 spawn 一个任务等 upgrade::on；路径是 /up 时回 101。第 1 条连接用 with_upgrades()，客户端把升级请求和 hello 一次写过去，收到之后再发 world；第 2 条连接不用 with_upgrades()；第 3 条连接发一个普通请求：

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

use std::{convert::Infallible, time::{Duration, Instant}};
use http_body_util::Empty;
use hyper::{body::{Bytes, Incoming}, server::conn::http1, service::service_fn, upgrade, Request, Response, StatusCode};
use hyper_util::rt::TokioIo;
use tokio::{io::{AsyncReadExt, AsyncWriteExt}, net::{TcpListener, TcpStream}, time::timeout};

static START: std::sync::OnceLock<Instant> = std::sync::OnceLock::new();
fn ms() -> u128 {
  START.get().unwrap().elapsed().as_millis()
}

async fn handle(mut req: Request<Incoming>) -> Result<Response<Empty<Bytes>>, Infallible> {
  let path = req.uri().path().to_string();
  let is_up = path == "/up";
  let on = upgrade::on(&mut req);
  tokio::spawn(async move {
    match on.await {
      Err(e) => println!("{:>4}ms server {path}: on_upgrade error: {e}", ms()),
      Ok(up) => match up.downcast::<TokioIo<TcpStream>>() {
        Err(_) => println!("{:>4}ms server {path}: downcast failed", ms()),
        Ok(parts) => {
          println!("{:>4}ms server {path}: upgraded, read_buf = {:?}", ms(), parts.read_buf);
          let mut io = parts.io.into_inner();
          io.write_all(&parts.read_buf).await.unwrap(); // echo what hyper had buffered
          let mut buf = [0u8; 64];
          let n = io.read(&mut buf).await.unwrap();
          io.write_all(&buf[..n]).await.unwrap(); // then echo from the raw socket
        }
      },
    }
  });
  let mut rsp = Response::new(Empty::new());
  if is_up {
    *rsp.status_mut() = StatusCode::SWITCHING_PROTOCOLS;
    rsp.headers_mut().insert("upgrade", "echo".parse().unwrap());
    rsp.headers_mut().insert("connection", "upgrade".parse().unwrap());
  }
  Ok(rsp)
}

async fn client(addr: std::net::SocketAddr, name: &str, steps: &[&str]) {
  println!("== {name}");
  let mut s = TcpStream::connect(addr).await.unwrap();
  let mut buf = [0u8; 1024];
  for step in steps {
    s.write_all(step.as_bytes()).await.unwrap();
    loop {
      match timeout(Duration::from_millis(100), s.read(&mut buf)).await {
        Ok(Ok(0)) => {
          println!("{:>4}ms client: EOF", ms());
          return;
        }
        Ok(Ok(n)) => {
          let text = String::from_utf8_lossy(&buf[..n]).replace("\r\n", "\\r\\n");
          println!("{:>4}ms client: got {text}", ms());
        }
        Ok(Err(e)) => panic!("{e}"),
        Err(_) => break, // quiet for 100ms: next step
      }
    }
  }
  drop(s);
  tokio::time::sleep(Duration::from_millis(50)).await;
}

#[tokio::main]
async fn main() {
  START.set(Instant::now()).unwrap();
  let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
  let addr = listener.local_addr().unwrap();
  tokio::spawn(async move {
    for i in 1.. {
      let (stream, _) = listener.accept().await.unwrap();
      let mut b = http1::Builder::new();
      b.auto_date_header(false);
      let conn = b.serve_connection(TokioIo::new(stream), service_fn(handle));
      tokio::spawn(async move {
        // connection 2 is served without with_upgrades()
        let r = if i == 2 { conn.await } else { conn.with_upgrades().await };
        println!("{:>4}ms server: connection {i} ended {r:?}", ms());
      });
    }
  });

  let up = "GET /up HTTP/1.1\r\nHost: x\r\nUpgrade: echo\r\nConnection: upgrade\r\n\r\nhello";
  client(addr, "1: with_upgrades, 101", &[up, "world"]).await;
  client(addr, "2: without with_upgrades, 101", &[up]).await;
  client(addr, "3: no Upgrade header", &["GET /plain HTTP/1.1\r\nHost: x\r\n\r\n"]).await;
}

M2 Max、rustc 1.99.0、--release。连跑 4 次，顺序相同，毫秒数差 2ms 以内：

== 1: with_upgrades, 101
   0ms server: connection 1 ended Ok(())
   0ms server /up: upgraded, read_buf = b"hello"
   0ms client: got HTTP/1.1 101 Switching Protocols\r\nupgrade: echo\r\nconnection: upgrade\r\n\r\n
   0ms client: got hello
 104ms client: got world
 104ms client: EOF
== 2: without with_upgrades, 101
 105ms server: connection 2 ended Ok(())
 105ms server /up: on_upgrade error: upgrade expected but low level API in use
 105ms client: got HTTP/1.1 101 Switching Protocols\r\nupgrade: echo\r\nconnection: upgrade\r\n\r\n
 105ms client: EOF
== 3: no Upgrade header
 105ms server /plain: on_upgrade error: no upgrade available
 105ms client: got HTTP/1.1 200 OK\r\ncontent-length: 0\r\n\r\n
 207ms server: connection 3 ended Ok(())

第 1 条：hello 和请求头在同一次写里到达，被 hyper 读进了缓冲区，downcast 后出现在 read_buf 里。UpgradeableConnection 在交出 IO 的同时就以 Ok(()) 结束了，比客户端收到 101 打印得还早。101 响应里没有 Content-Length，只有服务设置的两个头（实验关掉了 Date）。之后的 world 由升级任务直接从 socket 读、再写回。升级任务结束、drop 掉 socket，客户端看到 EOF。
第 2 条：101 照样写出去了，但 Connection 拿到 Upgrade 后只调 manual()，升级任务拿到错误，IO 随 Connection 一起被 drop，客户端在 101 之后立刻看到 EOF。客户端以为升级成功了，服务端却已经把连接扔了，这个错误只在服务端的日志里。
第 3 条：普通请求的 extensions 里没有 OnUpgrade，on 立即返回 no upgrade available。连接照常 keep-alive，客户端 drop 之后才结束。

HTTP/2 也能升级：HTTP/2 服务端对所有 CONNECT 请求（包括扩展 CONNECT）同样放一个 OnUpgrade，Upgraded 里装的是一个把 h2 流包装成 IO 的类型，downcast 成 TCP 类型会失败。见 HTTP/2 服务端。

❗IMPORTANT

动脑筋：带了 Upgrade 头，却回了 200

服务和上面的 handle 一样，连接用 with_upgrades()。客户端发一个带 Upgrade: echo 的请求，路径是 /no-101，服务回 200；然后在同一条连接上发一个普通请求，再关闭连接：

// handle() and client() are the same as above
#[tokio::main]
async fn main() {
  START.set(Instant::now()).unwrap();
  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();
    let mut b = http1::Builder::new();
    b.auto_date_header(false);
    let r = b.serve_connection(TokioIo::new(stream), service_fn(handle)).with_upgrades().await;
    println!("{:>4}ms server: connection ended {r:?}", ms());
  });

  client(
    addr,
    "Upgrade header, answered 200, then another request",
    &[
      "GET /no-101 HTTP/1.1\r\nHost: x\r\nUpgrade: echo\r\nConnection: upgrade\r\n\r\n",
      "GET /plain HTTP/1.1\r\nHost: x\r\n\r\n",
    ],
  )
  .await;
}

为 /no-101 spawn 的那个任务，on.await 什么时候返回、返回什么？答案在本页末尾。

动脑筋答案

M2 Max、rustc 1.99.0、--release。连跑 5 次，顺序相同，毫秒数差 2ms 以内：

== Upgrade header, answered 200, then another request
   0ms client: got HTTP/1.1 200 OK\r\ncontent-length: 0\r\n\r\n
 103ms server /plain: on_upgrade error: no upgrade available
 103ms client: got HTTP/1.1 200 OK\r\ncontent-length: 0\r\n\r\n
 206ms server: connection ended Ok(())
 206ms server /no-101: upgraded, read_buf = b""

/no-101 的 on.await 一直等到连接结束（206ms），返回的是 Ok(Upgraded)，read_buf 为空，里面的 socket 对端已经关了。

回 200 时 is_last 不为真，连接照常 keep-alive，读写回到 Init。Conn 里挂着的 Pending 没有人清：源码里只有 prepare_upgrade 写它、pending_upgrade 在连接结束时取它。
第二个请求没有 Upgrade 头，不会调 prepare_upgrade，旧的 Pending 继续挂着；这个请求的 on 拿到 no upgrade available。
客户端关闭后，服务端在空闲状态读到 EOF，读写两端关闭，is_done 为真。poll_inner 先查 pending_upgrade()，取到了那个旧的 Pending，于是返回 Dispatched::Upgrade，UpgradeableConnection 把一个已经结束的连接当作升级结果交了出去。

所以 on.await 返回 Ok 不能说明升级真的发生了。判断依据应该是自己写出的响应：只在回了 101 的那条路径上等 OnUpgrade，或者拿到 Upgraded 后先确认这是一个回了 101 的请求。

小结

CONNECT 或带 Upgrade 头的 HTTP/1.1 请求会被标成 Wants::UPGRADE；Dispatcher 建一对 oneshot，Pending 留在 Conn，OnUpgrade 放进请求的 extensions，upgrade::on 把它 remove 出来。
回 101（或 CONNECT 的 2xx）让写端在头写完后 Closed，刷新成功后读写都关，Dispatcher 返回 Dispatched::Upgrade，不关 IO。
只有 with_upgrades() 的 UpgradeableConnection 会把 IO 和读缓冲交出去（要求 I: Send + 'static）；普通 Connection 调 manual()，等待方得到 upgrade expected but low level API in use，而 101 已经发给了客户端。
Upgraded 是 Rewind<Box<dyn Io + Send>>：Rewind 先吐出 HTTP 解析时多读的字节，trait object 擦掉 IO 类型；downcast 按 TypeId 拆回原类型，多读的字节在 Parts::read_buf 里，要自己处理。
OnUpgrade 要在服务的 future 之外等（另起任务），因为 IO 要等响应写完才会交出。挂着的升级状态不会因为回了非 101 而清除，连接结束时仍会被当成升级交出去。