Rust in Everything
基于 hyper 1.12.0

ping：保活与 BDP

主线位置：两条主线的旁路。服务端在 poll_server 每圈收流之前、客户端在连接任务每次 poll h2 连接之前，先 poll 一次 Ponger；读到的每一帧都经 Recorder 记一笔。它不处理请求，只决定两件事：这条连接是不是死了，流控窗口该不该放大。

flowchart TB
  frames["读到 HEADERS / DATA / trailers"] --> rec["Recorder：记时间、记字节"]
  rec --> shared["Shared：Mutex 里的共享状态"]
  pong["Ponger::poll"] --> shared
  pong --> ka["keep-alive：到点发 PING"]
  pong --> bdp["BDP：PONG 回来算带宽"]
  ka -->|"超时"| dead["KeepAliveTimedOut：关连接"]
  bdp -->|"窗口变大"| upd["SizeUpdate：调窗口"]

ping.rs 的模块注释开头就说了：hyper 用 HTTP/2 的 PING 做两件事，自适应流控（BDP）和连接保活，两者都是可选的。PING 帧本身由 h2（0.4.20）的 PingPong 发收，对端发来的 PING 也由 h2 自动回 ACK，hyper 只管「什么时候发自己的 PING、PONG 回来后做什么」。

Recorder 与 Ponger

两者都没开时根本不建 ping 通道（Config::is_enabled 为假，服务端 Serving.ping 是 None，客户端拿到的是 ping::disabled()，一个空的 Recorder）。开了任意一个，channel 建一对：

pub(crate) struct Recorder {
    shared: Option<Arc<Mutex<Shared>>>,
}

pub(super) struct Ponger {
    bdp: Option<Bdp>,
    keep_alive: Option<KeepAlive>,
    shared: Arc<Mutex<Shared>>,
}

Shared 放在 Arc<Mutex> 里，字段是两件事的状态：ping_pong（h2 的句柄）、ping_sent_at（在途 PING 的发出时间，Some 就说明有一个 PING 没回来）、bytes 和 next_bdp_at（BDP 用）、last_read_at 和 is_keep_alive_timed_out（keep-alive 用）、timer。

  • Recorder 可以克隆，到处都是。连接本身持一个；每个请求或响应的 Incoming 持一个，record_data 在每交出一块 DATA 时调用，record_non_data 在收到请求头、响应头、trailers 时调用。客户端的 Send 任务和 Pipe 任务也各持一个，即使它们不记录什么。
  • Ponger 只有一个，由驱动连接的那个 future 持有：服务端在 poll_ping 里 poll 它，客户端在包着 h2 连接的 Conn::poll 里 poll 它。结果是 Ponged 的两个变体之一：SizeUpdate(wnd) 或 KeepAliveTimedOut。

Recorder 的数量顺带回答了「连接空不空闲」。is_idle：

    fn is_idle(&self) -> bool {
        Arc::strong_count(&self.shared) <= 2
    }

Ponger 一个引用、连接自己的 Recorder 一个，超过 2 就说明还有流的 body、Send 任务或 Pipe 任务活着。不用另外维护一个计数器，代价是这个「空闲」的定义比「没有打开的流」略宽：只要某个 Incoming 还没被 drop，连接就不算空闲。

keep-alive：读不到帧就发 PING

KeepAlive 带一个 sleep（所以要 Timer，没有就在建通道时 panic，见 Executor 与 Timer）和一个三态的 KeepAliveState：

stateDiagram-v2
  state "Init：没排期" as Init
  state "Scheduled：等间隔到期" as Scheduled
  state "PingSent：等 PONG" as PingSent
  [*] --> Init
  Init --> Scheduled: 不空闲或 while_idle
  Scheduled --> Init: 期间读到过帧
  Scheduled --> PingSent: 到点，发 PING
  PingSent --> Scheduled: PONG 回来
  PingSent --> [*]: 超时

Ponger::poll 每次先走 keep-alive 的两步，再看 PONG：

        if let Some(ka) = &mut self.keep_alive {
            ka.maybe_schedule(is_idle, &locked);
            ka.maybe_ping(cx, is_idle, &mut locked);
        }

        if !locked.is_ping_sent() {
            // XXX: this doesn't register a waker...?
            return Poll::Pending;
        }

        match locked.ping_pong.poll_pong(cx) {
            Poll::Ready(Ok(_pong)) => {
                // ... 清掉 ping_sent_at，算 rtt
                if let Some(ka) = &mut self.keep_alive {
                    locked.update_last_read_at();
                    ka.maybe_schedule(is_idle, &locked);
                    ka.maybe_ping(cx, is_idle, &mut locked);
                }

                // ... BDP，见下文
            }
            // ...
            Poll::Pending => {
                if let Some(ka) = &mut self.keep_alive {
                    if let Err(KeepAliveTimedOut) = ka.maybe_timeout(cx) {
                        self.keep_alive = None;
                        locked.is_keep_alive_timed_out = true;
                        return Poll::Ready(Ponged::KeepAliveTimedOut);
                    }
                }
            }
        }

maybe_ping 是关键：

            KeepAliveState::Scheduled(at) => {
                if Pin::new(&mut self.sleep).poll(cx).is_pending() {
                    return;
                }
                // check if we've received a frame while we were scheduled
                if shared.last_read_at() + self.interval > at {
                    self.state = KeepAliveState::Init;
                    cx.waker().wake_by_ref(); // schedule us again
                    return;
                }
                if !self.while_idle && is_idle {
                    trace!("keep-alive no need to ping when idle and while_idle=false");
                    return;
                }
                trace!("keep-alive interval ({:?}) reached", self.interval);
                shared.send_ping();
                self.state = KeepAliveState::PingSent;
                let timeout = self.timer.now() + self.timeout;
                self.timer.reset(&mut self.sleep, timeout);
            }
  • 排期按 last_read_at + interval 算（schedule），不是「每隔 interval 发一次」。sleep 醒来时如果这段时间里读到过帧，last_read_at 已经往后挪了，就回到 Init、唤醒自己重新排期，不发 PING。忙碌的连接上数据一直在来，keep-alive 一个 PING 也不会发。
  • 发 PING 之后，同一个 sleep 被 reset 成 now + timeout，从「等间隔」变成「等 PONG」。PONG 回来算作读到一帧，update_last_read_at 后重新排期；sleep 先醒就是超时，maybe_timeout 返回错误。
  • 注释里的 XXX: this doesn't register a waker...? 说的是：没有 PING 在途时直接返回 Pending，不在 poll_pong 上注册。这时它再次被 poll，靠的是 keep-alive 的 sleep（Scheduled 时注册了 waker）或连接上的其他事件。

「读到帧」只包括 hyper 通过 Recorder 看到的帧：头、数据、trailers，以及自己 PING 的 PONG。对端发来的 PING、SETTINGS、WINDOW_UPDATE 由 h2 内部处理，不更新 last_read_at。

📝NOTE
冷知识：while_idle = false 曾经只管第一次

#3380（issue，「eliminates unnecessary ping when the connection is idle」）报告：对着 gRPC 的 Go 服务端，keep_alive_while_idle(false) 的客户端仍在空闲时发 PING，服务端「otherwise output too many ping」。issue 里贴出的修改建议就是在 maybe_ping 里加一个 if !self.while_idle && is_idle { return; }。

在那之前，空闲检查只在 maybe_schedule 的 Init 分支里：空闲时不排期。可是一旦在有流的时候排过期，sleep 到点时 maybe_ping 不再看是否空闲，照发不误；PONG 回来后的 PingSent 分支又直接重新排期，于是请求结束后 PING 一直按间隔发下去。#3381（v1.0.0，「don't send keep-alive ping when idle」）给 maybe_ping 加了 is_idle 参数和上面代码里那三行检查。

默认值与两端的差别

服务端客户端
keep_alive_interval无（关）无（关）
keep_alive_timeout20 秒20 秒
keep_alive_while_idle写死为 true，没有这个选项默认 false，可配
超时后abrupt_shutdown(NO_ERROR)：发 GOAWAY 再关连接任务直接结束，socket 关闭
报错Connection 返回 Err，is_timeout() 为真Connection 和在途请求都返回 Err，is_timeout() 为真

服务端在 Server::new 里把 keep_alive_while_idle 写死：

        let ping_config = ping::Config {
            bdp_initial_window: bdp,
            keep_alive_interval: config.keep_alive_interval,
            keep_alive_timeout: config.keep_alive_timeout,
            // If keep-alive is enabled for servers, always enabled while
            // idle, so it can more aggressively close dead connections.
            keep_alive_while_idle: true,
        };

超时后服务端的 poll_ping 调 h2 的 abrupt_shutdown(h2::Reason::NO_ERROR)；之后 poll_accept 返回 None，poll_server 里的 ensure_not_timed_out 把它变成 KeepAliveTimedOut 错误。客户端的 Conn::poll 遇到 KeepAliveTimedOut 直接 return Poll::Ready(Ok(()))，ConnTask 结束，h2 连接被 drop；ClientTask 和在途请求的 ResponseFutMap 在各自的错误路径上调 ensure_not_timed_out，把 h2 的连接错误换成超时错误。KeepAliveTimedOut 的 source() 是 hyper 的 TimedOut，所以 is_timeout() 为真；Display 只有 http2 error。

实验：PING 没人回

服务端 keep-alive 间隔 200ms、超时 300ms。客户端是手写的：发连接前言和一个空 SETTINGS，回 ACK 给服务端的 SETTINGS，然后只读不回，不理 PING。

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, time::{Duration, Instant}};
use http_body_util::Empty;
use hyper::{body::{Bytes, Incoming}, server, service::service_fn, Request, Response};
use hyper_util::rt::{TokioExecutor, TokioIo, TokioTimer};
use tokio::{io::{AsyncReadExt, AsyncWriteExt}, net::{TcpListener, TcpStream}};

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

#[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();
    let svc = service_fn(|_: Request<Incoming>| async {
      Ok::<_, Infallible>(Response::new(Empty::<Bytes>::new()))
    });
    let r = server::conn::http2::Builder::new(TokioExecutor::new())
      .timer(TokioTimer::new())
      .keep_alive_interval(Duration::from_millis(200))
      .keep_alive_timeout(Duration::from_millis(300))
      .serve_connection(TokioIo::new(stream), svc)
      .await;
    match r {
      Ok(()) => println!("{:>4}ms server conn: Ok(())", ms()),
      Err(e) => println!("{:>4}ms server conn: Err({e}), is_timeout={}", ms(), e.is_timeout()),
    }
  });

  // A raw client: preface + SETTINGS, ACKs the server's SETTINGS,
  // never answers PING.
  let mut c = TcpStream::connect(addr).await.unwrap();
  c.write_all(b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n").await.unwrap();
  c.write_all(&[0, 0, 0, 4, 0, 0, 0, 0, 0]).await.unwrap();
  let mut buf = Vec::new();
  let mut chunk = [0u8; 4096];
  loop {
    let n = c.read(&mut chunk).await.unwrap();
    if n == 0 {
      println!("{:>4}ms client: EOF", ms());
      break;
    }
    buf.extend_from_slice(&chunk[..n]);
    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, flags) = (NAMES[buf[3] as usize], buf[4]);
      let ack = if flags & 1 == 1 && (kind == "SETTINGS" || kind == "PING") { " ACK" } else { "" };
      let code = if kind == "GOAWAY" {
        format!(" error_code={}", u32::from_be_bytes(buf[13..17].try_into().unwrap()))
      } else {
        String::new()
      };
      println!("{:>4}ms client: got {kind}{ack}{code}", ms());
      if kind == "SETTINGS" && ack.is_empty() {
        c.write_all(&[0, 0, 0, 4, 1, 0, 0, 0, 0]).await.unwrap();
      }
      buf.drain(..9 + len);
    }
  }
  tokio::time::sleep(Duration::from_millis(10)).await;
}

M2 Max、rustc 1.99.0、--release，连跑 3 次，除毫秒数差几毫秒外相同：

   0ms client: got SETTINGS
   0ms client: got SETTINGS ACK
   0ms client: got WINDOW_UPDATE
 202ms client: got PING
 506ms server conn: Err(http2 error), is_timeout=true
 506ms client: got GOAWAY error_code=0
 506ms client: EOF

连接上一个流都没有，服务端照样在 200ms 发 PING（while_idle 写死为真）。PING 发出后 300ms 没有 PONG，服务端发 GOAWAY(NO_ERROR) 关连接，Connection 以超时错误结束。开头的 WINDOW_UPDATE 是服务端把连接级窗口从规范默认的 65535 放大到 1 MiB：用 sniff.rs 看，它在流 0 上的增量是 983041，正好是 1048576 − 65535（SETTINGS 只能改流级窗口的初始值，连接级窗口只能靠 WINDOW_UPDATE）。

反过来，hyper 客户端（间隔 100ms、超时 200ms）对着一个只读不回的手写服务端发一个请求：

use std::time::{Duration, Instant};
use http_body_util::Empty;
use hyper::{body::Bytes, client, Request};
use hyper_util::rt::{TokioExecutor, TokioIo, TokioTimer};
use tokio::{io::{AsyncReadExt, AsyncWriteExt}, net::{TcpListener, TcpStream}};

#[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();

  // A raw server: sends SETTINGS, ACKs every SETTINGS it gets, reads and
  // ignores everything else (never answers PING, never responds).
  tokio::spawn(async move {
    let (mut s, _) = listener.accept().await.unwrap();
    s.write_all(&[0, 0, 0, 4, 0, 0, 0, 0, 0]).await.unwrap();
    let mut buf = Vec::new();
    let mut chunk = [0u8; 4096];
    let mut skip = 24;
    loop {
      let n = s.read(&mut chunk).await.unwrap();
      if n == 0 {
        println!("{:>4}ms server: EOF", ms());
        return;
      }
      buf.extend_from_slice(&chunk[..n]);
      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;
        }
        if buf[3] == 4 && buf[4] & 1 == 0 {
          s.write_all(&[0, 0, 0, 4, 1, 0, 0, 0, 0]).await.unwrap();
        }
        match buf[3] {
          1 => println!("{:>4}ms server: got HEADERS", ms()),
          6 => println!("{:>4}ms server: got PING{}", ms(), if buf[4] & 1 == 1 { " ACK" } else { "" }),
          7 => println!("{:>4}ms server: got GOAWAY", ms()),
          _ => {}
        }
        buf.drain(..9 + len);
      }
    }
  });

  let stream = TcpStream::connect(addr).await.unwrap();
  let (mut sender, conn) = client::conn::http2::Builder::new(TokioExecutor::new())
    .timer(TokioTimer::new())
    .keep_alive_interval(Duration::from_millis(100))
    .keep_alive_timeout(Duration::from_millis(200))
    .handshake::<_, Empty<Bytes>>(TokioIo::new(stream))
    .await
    .unwrap();
  let conn = tokio::spawn(async move {
    let r = conn.await;
    println!("{:>4}ms client conn: {r:?}", ms());
  });
  let req = Request::get("http://x/").body(Empty::new()).unwrap();
  match sender.send_request(req).await {
    Ok(res) => println!("{:>4}ms client: {}", ms(), res.status()),
    Err(e) => println!("{:>4}ms client: Err({e}), is_timeout={}", ms(), e.is_timeout()),
  }
  conn.await.unwrap();
  tokio::time::sleep(Duration::from_millis(20)).await;
}

连跑 3 次，除毫秒数差 1ms 以内外相同：

   0ms server: got HEADERS
 104ms server: got PING
 306ms client conn: Err(hyper::Error(Http2, KeepAliveTimedOut))
 306ms client: Err(http2 error), is_timeout=true
 306ms server: EOF

客户端默认 while_idle = false，但这里有一个请求在等响应（Send 任务持着一个 Recorder，连接不空闲），所以 100ms 发了 PING。300ms 时超时：对端看到的是直接 EOF，没有 GOAWAY；Connection 和在途请求都得到超时错误。

📝NOTE
没有蠢问题

问： 客户端默认空闲时不发 PING，服务端却写死空闲时也发，为什么两边不一样？

答： 两边要防的东西不同。服务端开 keep-alive 是为了清理死连接：客户端掉线、NAT 表项过期之后，连接上不会再有任何帧，它恰恰是「空闲」的，空闲时不发 PING 就永远发现不了，注释说的就是「so it can more aggressively close dead connections」。客户端开 keep-alive 主要是为了别让在途请求挂在一条死连接上；空闲连接死了，下次用时自然会失败，而且很多服务端把客户端在没有流时频繁发 PING 当作滥用（上面冷知识里的 gRPC 就是），所以默认不发。连接池里的空闲连接需要保活时，再显式打开 keep_alive_while_idle。

BDP：用 PING 量出窗口该多大

HTTP/2 的接收窗口决定了对端在等 WINDOW_UPDATE 之前最多能发多少字节。窗口小于带宽 × 往返时间（BDP），发送方就会时不时停下来等，吞吐上不去。adaptive_window(true) 让 hyper 自己量：模块注释里的算法是，收到 DATA 时若没有 BDP PING 在途，记下时间、发一个 PING；之后累加收到的字节；PONG 回来时，用往返时间和这段时间收到的字节估算 BDP，大于当前窗口的 2/3 就把窗口调大。

record_data 负责采样：

        let mut locked = shared.lock().panic_if_poisoned();

        locked.update_last_read_at();

        // are we ready to send another bdp ping?
        // if not, we don't need to record bytes either

        if let Some(next_bdp_at) = &locked.next_bdp_at {
            if locked.timer.now() < *next_bdp_at {
                return;
            } else {
                locked.next_bdp_at = None;
            }
        }

        if let Some(bytes) = &mut locked.bytes {
            *bytes += len;
        } else {
            // no need to send bdp ping if bdp is disabled
            return;
        }

        if !locked.is_ping_sent() {
            locked.send_ping();
        }

BDP 的 PING 由读数据的一方顺手发出（Incoming 被 poll 时），不需要 timer 的 sleep，只用 timer.now()。这就是为什么只开 adaptive_window 不给 timer 不会 panic。keep-alive 和 BDP 共用 ping_sent_at：同一时刻最多一个 PING 在途，它的 PONG 两边都用。

PONG 回来后，Bdp::calculate：

        // average the rtt
        let rtt = seconds(rtt);
        if self.rtt == 0.0 {
            // First sample means rtt is first rtt.
            self.rtt = rtt;
        } else {
            // Weigh this rtt as 1/8 for a moving average.
            self.rtt += (rtt - self.rtt) * 0.125;
        }

        // calculate the current bandwidth
        let bw = (bytes as f64) / (self.rtt * 1.5);
        trace!("current bandwidth = {:.1}B/s", bw);

        if bw < self.max_bandwidth {
            // not a faster bandwidth, so don't update
            self.stabilize_delay();
            return None;
        } else {
            self.max_bandwidth = bw;
        }

        // if the current `bytes` sample is at least 2/3 the previous
        // bdp, increase to double the current sample.
        // ...
        if bytes >= self.bdp as usize * 2 / 3 {
            self.bdp =
                WindowSize::try_from((bytes * 2).min(BDP_LIMIT as usize)).unwrap_or(BDP_LIMIT);
            trace!("BDP increased to {}", self.bdp);

            self.stable_count = 0;
            self.ping_delay /= 2;
            Some(self.bdp)
        } else {
            self.stabilize_delay();
            None
        }

规则归纳一下：

量规则
RTT第一个样本直接用，之后按 1/8 的权重做指数移动平均
带宽bytes / (rtt × 1.5)，只认创新高的样本
调窗口这个样本的字节数 ≥ 当前窗口的 2/3，新窗口 = 字节数 × 2，上限 BDP_LIMIT 16 MiB（注释：再大就撞上 TCP 自己的流控了）
下次采样ping_delay 从 100ms 起；调了窗口就减半，否则（stabilize_delay）在延迟还不到 10 秒时，连续两次没调窗口就乘 4

ping_delay 的伸缩是 v0.14.8 的 #2550 加的，提交说明说目的是窗口在变时快速提速、稳定后少发 PING，「hopefully preventing hyper from triggering ENHANCE_YOUR_CALM errors」。

calculate 返回 Some(wnd) 时 Ponger::poll 给出 Ponged::SizeUpdate(wnd)，两端都对 h2 连接调 set_target_window_size(wnd) 和 set_initial_window_size(wnd)：前者放大连接级窗口，后者通过一个新的 SETTINGS 帧改所有流的初始窗口。

实验：窗口怎么长

客户端开 adaptive_window，服务端回一个 64 MiB 的 body，中间的 sniff.rs（HTTP/2 服务端 那页的代码）只打印 SETTINGS 帧。SETTINGS 参数按 id=值 打印：2 是 ENABLE_PUSH，3 是 MAX_CONCURRENT_STREAMS，4 是 INITIAL_WINDOW_SIZE，5 是 MAX_FRAME_SIZE，6 是 MAX_HEADER_LIST_SIZE。

use std::{convert::Infallible, time::Instant};
use http_body_util::{BodyExt, 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]
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, &["SETTINGS"]).await;
  tokio::spawn(async move {
    let (stream, _) = listener.accept().await.unwrap();
    let svc = service_fn(|_: Request<Incoming>| async {
      Ok::<_, Infallible>(Response::new(Full::new(Bytes::from(vec![0u8; 64 << 20]))))
    });
    server::conn::http2::Builder::new(TokioExecutor::new())
      .serve_connection(TokioIo::new(stream), svc)
      .await
      .unwrap();
  });

  let stream = TcpStream::connect(proxy).await.unwrap();
  let (mut sender, conn) = client::conn::http2::Builder::new(TokioExecutor::new())
    .adaptive_window(true)
    .handshake::<_, Empty<Bytes>>(TokioIo::new(stream))
    .await
    .unwrap();
  tokio::spawn(conn);
  let res = sender.send_request(Request::get("http://x/").body(Empty::new()).unwrap()).await.unwrap();
  let n = res.into_body().collect().await.unwrap().to_bytes().len();
  println!("{:>5}ms client: read {n} bytes", start.elapsed().as_millis());
}

多线程运行时，其中一次的输出：

    0ms s->c SETTINGS stream=0 3=200 4=1048576 5=16384 6=16384
    0ms c->s SETTINGS stream=0 2=0 4=65535 5=16384 6=16384
    0ms c->s SETTINGS stream=0 ACK
    0ms s->c SETTINGS stream=0 ACK
    1ms c->s SETTINGS stream=0 4=131070
    1ms s->c SETTINGS stream=0 ACK
   51ms c->s SETTINGS stream=0 4=425980
   51ms s->c SETTINGS stream=0 ACK
   77ms c->s SETTINGS stream=0 4=1605624
   78ms s->c SETTINGS stream=0 ACK
   92ms c->s SETTINGS stream=0 4=2719728
   92ms s->c SETTINGS stream=0 ACK
  100ms client: read 67108864 bytes
  • 头两行是两边的初始 SETTINGS：服务端用 hyper 的默认值（并发 200、窗口 1 MiB），客户端开了自适应窗口，从规范默认的 65535 起步，并且总是声明 ENABLE_PUSH=0。
  • 之后每个 4= 都是一次 SizeUpdate。第一次在 1ms：第一个样本收满了一个 65535 的窗口，65535 ≥ 2/3 × 65535，新窗口 = 65535 × 2 = 131070。
  • 更新的时间间隔约 50ms、26ms、15ms：ping_delay 从 100ms 每次减半，下一次采样要等 next_bdp_at，所以更新越来越密。每次的新值都是样本字节数的两倍（都是偶数）。

一共跑了 8 次。更新的时间点稳定（约 1ms、51ms、77ms，其中 2 次在 90ms 左右还有第四次，这 2 次里又有 1 次在 100ms 有第五次），窗口的具体值每次不同（第一次更新是 131070 或 229374，第二次在 425980 到 851964 之间，第三次在 819200 到 3178496 之间），因为一个 PING 往返里收到多少字节取决于调度。64 MiB 在本机回环上约 100ms 就读完了，窗口还远没到 16 MiB 的上限。

❗IMPORTANT
动脑筋：空闲连接会发几个 PING？

客户端 keep-alive 间隔 100ms、带 TokioTimer，服务端不开 keep-alive，处理函数睡 550ms。三种情况各观察 550ms，用 sniff.rs 的 CLIENT_PINGS 数客户端发出的 PING（不算 ACK）：A 不发请求，keep_alive_while_idle(false)；B 不发请求，keep_alive_while_idle(true)；C while_idle 为 false，发一个请求等它的响应。依赖同上。

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

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

async fn run_case(name: &str, while_idle: bool, busy: bool) {
  let start = Instant::now();
  let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
  let proxy = sniff::proxy(listener.local_addr().unwrap(), start, &[]).await;
  tokio::spawn(async move {
    let (stream, _) = listener.accept().await.unwrap();
    let svc = service_fn(|_: Request<Incoming>| async {
      tokio::time::sleep(Duration::from_millis(550)).await;
      Ok::<_, Infallible>(Response::new(Empty::<Bytes>::new()))
    });
    let _ = server::conn::http2::Builder::new(TokioExecutor::new())
      .serve_connection(TokioIo::new(stream), svc)
      .await;
  });

  sniff::CLIENT_PINGS.store(0, Ordering::SeqCst);
  let stream = TcpStream::connect(proxy).await.unwrap();
  let (mut sender, conn) = client::conn::http2::Builder::new(TokioExecutor::new())
    .timer(TokioTimer::new())
    .keep_alive_interval(Duration::from_millis(100))
    .keep_alive_while_idle(while_idle)
    .handshake::<_, Empty<Bytes>>(TokioIo::new(stream))
    .await
    .unwrap();
  tokio::spawn(conn);
  if busy {
    // one request whose response takes 550ms
    let req = Request::get("http://x/").body(Empty::new()).unwrap();
    sender.send_request(req).await.unwrap();
  } else {
    tokio::time::sleep(Duration::from_millis(550)).await;
  }
  println!("{name}: client sent {} PINGs", sniff::CLIENT_PINGS.load(Ordering::SeqCst));
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
  run_case("A idle, while_idle=false", false, false).await;
  run_case("B idle, while_idle=true ", true, false).await;
  run_case("C busy, while_idle=false", false, true).await;
}
三种情况各是几个？答案在本页末尾。

动脑筋答案

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

A idle, while_idle=false: client sent 0 PINGs
B idle, while_idle=true : client sent 5 PINGs
C busy, while_idle=false: client sent 5 PINGs
  • A：连接上只有 Ponger 和 ClientTask 的 Recorder，is_idle 为真，maybe_schedule 在 Init 就返回，从不排期。
  • B：允许空闲时发。每次 PONG 回来都更新 last_read_at 并按 last_read_at + 100ms 重新排期，550ms 内在约 100、200、300、400、500ms 发了 5 个。服务端没开 keep-alive，但 h2 会自动回 PING，所以每个都有 PONG，连接不会超时。
  • C：请求在等响应，Send 任务持着一个 Recorder，引用数是 3，连接不空闲，和 B 一样发 5 个。响应头在 550ms 才来，这期间没有别的帧更新 last_read_at，排期只靠 PONG 推进。

小结

  • Ponger 只有一个，由驱动连接的 future 每次 poll；Recorder 到处都是，读到头、数据、trailers 时记一笔。两者共享一个 Mutex 里的状态，同一时刻最多一个 PING 在途，keep-alive 和 BDP 共用它的 PONG。
  • keep-alive 按 last_read_at + interval 排期，期间读到过帧就不发；发出后等 timeout，没有 PONG 就超时。需要 Timer。服务端 while_idle 写死为真，超时后发 GOAWAY 再关；客户端默认空闲不发，超时直接断开。两边的错误 is_timeout() 都为真。
  • is_idle 就是 Arc::strong_count(&shared) <= 2：还有 body、Send 或 Pipe 任务活着，连接就不空闲。
  • BDP 由收数据的一方顺手发 PING，只用 timer.now()；样本字节数达到当前窗口的 2/3 就把窗口设为样本的两倍（上限 16 MiB），采样间隔从 100ms 起，增长时减半、稳定时乘 4。

💬 关联讨论

发起讨论
加载中...