Rust in Everything
基于 reqwest 0.13.5

read_timeout 把调用方的处理时间也算了进去

主线位置：响应主线，读 body 的那一段。read_timeout 在 body 上是一层 ReadTimeoutBody，每次交出一帧就把计时器往后推。三种超时各管哪一段、0.12 和 0.13.5 的差别，见 超时。本页讲这个差别在流式消费时的后果。

sequenceDiagram
  participant S as 服务器
  participant B as ReadTimeoutBody
  participant A as 调用方
  S->>B: 4 块一次发完
  A->>B: chunk
  B-->>A: 第 1 块 并重置计时
  Note over A: 处理 300ms
  A->>B: chunk
  B-->>A: 计时已过 报超时

实验 crate：exp-reqwest/pitb-lab（src/bin/readgap.rs）。M2 Max、macOS 27、rustc 1.99.0，reqwest 0.13.5，release 构建，客户端 .no_proxy()，连跑 3 次，结果相同（毫秒数差 1 到 3）。服务器是 testsrv：?chunks=4&size=16 一次发完 4 块，每块 16 字节；加上 body_ms=1000 则每两块之间停 1 秒。

现象

下载一个流，边读边处理：写一个慢盘、往一个有界通道里塞、再转发给另一个慢的下游。客户端设了 read_timeout(200ms)，用意是「服务器 200 毫秒没有新数据就放弃」：

  let c = reqwest::Client::builder().no_proxy().read_timeout(RT).build().unwrap();
  let mut r = c.get(srv.url(path)).send().await.unwrap();
  // ...
  let out = loop {
    match r.chunk().await {
      Ok(Some(_)) => {
        n += 1;
        tokio::time::sleep(Duration::from_millis(per_chunk_ms)).await;
      }
      // ...

服务器一口气把 4 块都发完了，从来没有沉默过：

[read_timeout, consumer 100ms/chunk] all 4 chunks (413ms)
[read_timeout, consumer 300ms/chunk] chunk #2 failed: is_timeout=true error decoding response body <- request or response body error <- operation timed out (304ms)
[read_timeout, 300ms before 1st chunk only] all 4 chunks (308ms)
  • 每块处理 100 毫秒，没事。每块处理 300 毫秒，第 2 次 chunk() 就报超时，这时第 2 块其实已经在缓冲区里了。
  • 错误是 is_timeout() 加 Decode 种类，Display 是 error decoding response body，看上去像是服务器卡住或者数据坏了。
  • 拿到响应头之后、读第一块之前停 300 毫秒，不报错。只有「交出一块」到「再要下一块」之间的时间被算进去。

原因

ReadTimeoutBody::poll_frame：

        let deadline = tokio::time::Instant::now() + *this.timeout;
        let sleep_pinned = if let Some(some) = this.sleep.as_mut().as_pin_mut() {
            some
        } else {
            this.sleep.set(Some(tokio::time::sleep_until(deadline)));
            this.sleep.as_mut().as_pin_mut().unwrap()
        };

        // Error if the timeout has expired.
        if let Poll::Ready(()) = sleep_pinned.poll(cx) {
            return Poll::Ready(Some(Err(crate::error::body(crate::error::TimedOut))));
        }

        let item = ready!(this.inner.poll_frame(cx))
            .map(|opt_chunk| opt_chunk.map_err(crate::error::body));
        // a ready frame means timeout is reset for the next chunk
        if let Some(some) = this.sleep.as_mut().as_pin_mut() {
            some.reset(tokio::time::Instant::now() + *this.timeout);
        }
        Poll::Ready(item)

两处合起来造成这个现象：

  • 交出一帧的同时 reset(now + timeout)：下一次的期限从「交出时刻」算起，调用方拿着这一块做事的时间全在里面。
  • 每次 poll 先看计时器，再去读内层 body：期限一过，就算下一块已经到了，也先报超时。

第一块之前不受影响，是因为 sleep 一开始是 None，第一次 poll 才创建，从那一刻算。

这是 #3085（「perf(body): reuse tokio::time::Sleep timer via reset() in ReadTimeoutBody」，ffda263）带来的，首次发布在 v0.13.5，CHANGELOG 没有提到。改之前，交出一帧后是 this.sleep.set(None)，下一次 poll 才新建计时器，处理时间不算。0.12.28 上同一个实验 4 块都读得到，对照输出在 超时 的动脑筋答案里。tag 之后 master 上 src/async_impl/body.rs 没有新提交。

修法

不设 read_timeout，自己给「等下一块」计时。 tokio::time::timeout 只包住 chunk() 这一个 future，处理数据的时间在它外面：

  let c = reqwest::Client::builder().no_proxy().build().unwrap();
  let mut r = c.get(srv.url(path)).send().await.unwrap();
  // ...
  let out = loop {
    match tokio::time::timeout(RT, r.chunk()).await {
      Ok(Ok(Some(_))) => {
        n += 1;
        tokio::time::sleep(Duration::from_millis(per_chunk_ms)).await;
      }
      Ok(Ok(None)) => break format!("all {n} chunks"),
      Ok(Err(e)) => break format!("chunk #{} failed: {}", n + 1, pitb_lab::chain(&e)),
      Err(elapsed) => break format!("chunk #{}: no data for {RT:?} ({elapsed})", n + 1),
    }
  };

同一个慢消费者，以及一个真的每两块之间停 1 秒的服务器：

[timeout(chunk()), consumer 300ms/chunk] all 4 chunks (1209ms)
[timeout(chunk()), server stalls 1s] chunk #2: no data for 200ms (deadline has elapsed) (203ms)

慢消费者读完了 4 块；服务器真的沉默时，200 毫秒后放弃。这正是当初想要的语义。超时被丢掉的只是 chunk() 这个 future，Response 还在，可以决定重试或者放弃。

响应头之前那一段，read_timeout 本来就是一个从 send() 起算的固定期限（见 超时），去掉它以后用 connect_timeout 和 RequestBuilder::timeout / 外面再包一层 tokio::time::timeout 补上。

用 bytes() / text() 一次读完的代码不受影响：它们在读完之前不交还控制权，两次 poll 之间没有调用方的时间。

怎么检查

  • 错误满足 e.is_timeout()，Display 是 error decoding response body，出现在读 body 的循环里，而且失败的位置和调用方每块的处理时间有关：处理越慢，越早失败。
  • 把处理逻辑临时换成空操作再跑一次，问题消失，就是这个坑。
  • 查 Cargo.lock 里 reqwest 的版本：v0.13.5 有这个行为，0.12.28 没有。

小结

  • 0.13.5 的 read_timeout 在交出一块时重置计时器，下一块的期限包括调用方处理这一块的时间；期限一过，数据已经到了也报超时。
  • 症状是读流时 error decoding response body、is_timeout() 为真，失败点随处理速度变化。
  • 来源是 #3085（ffda263，v0.13.5），一个标题为性能优化的提交。
  • 修法：不设 read_timeout，用 tokio::time::timeout 只包住每次 chunk()；一次读完的 bytes() / text() 不受影响。

💬 关联讨论

发起讨论
加载中...