Rust in Everything
基于 tower 0.5.3

request timed out，内层却只用了 100ms：超时算进了排队

主线位置：请求主线。Timeout 只在 call 返回的 future 上挂一个 Sleep。排队发生在 call 之前（就绪主线上）还是之后（请求主线上），决定了它算不算进超时。

flowchart TB
  q{排队发生在哪}
  q -->|"poll_ready 里"| pr["等 ConcurrencyLimit 许可"]
  q -->|"call 之后"| af["Buffer 通道里、worker 手里"]
  pr --> no["Timeout 还没开始计时"]
  af --> yes["Timeout 在 Buffer 外：已经在计时"]
  af --> inner["Timeout 在 Buffer 里：还没开始"]

Timeout 一页已经用 A、B、C 三种摆法比较过超时的个数。这一页从排查的角度看同一件事：日志里内层每个请求都很快，调用方却收到 request timed out；或者反过来，请求明明等了很久，Timeout 一次也没触发。

现象：内层不慢，调用方超时

内层每个请求固定耗时 100ms，并打印开始和结束的时间；外面是 Timeout(150ms) > Buffer > ConcurrencyLimit(1)，三个任务同时发请求：

tower = { version = "=0.5.3", features = ["timeout", "buffer", "limit", "util"] }
tokio = { version = "=1.54.1", features = ["full", "test-util"] }

use std::time::Duration;
use tokio::time::{Instant, sleep};
use tower::{BoxError, Service, ServiceBuilder, ServiceExt, service_fn};

// Each request takes 100ms; prints when the inner service starts and finishes it.
fn handler(start: Instant) -> impl Service<u32, Response = u32, Error = BoxError, Future: Send> + Clone + Send + 'static {
  service_fn(move |id: u32| async move {
    println!("    inner: req {id} starts at {}ms", start.elapsed().as_millis());
    sleep(Duration::from_millis(100)).await;
    println!("    inner: req {id} done   at {}ms", start.elapsed().as_millis());
    Ok::<_, BoxError>(id)
  })
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
  let start = Instant::now();
  let svc = ServiceBuilder::new()
    .timeout(Duration::from_millis(150))
    .buffer(8)
    .concurrency_limit(1)
    .service(handler(start));

  let mut tasks = Vec::new();
  for id in 1..=3 {
    let mut svc = svc.clone();
    tasks.push(tokio::spawn(async move {
      let r = svc.ready().await?.call(id).await;
      Ok::<_, BoxError>((id, r))
    }));
  }
  for t in tasks {
    let (id, r) = t.await.unwrap().unwrap();
    let ms = start.elapsed().as_millis();
    match r {
      Ok(_) => println!("caller: req {id} ok at {ms}ms"),
      Err(e) => println!("caller: req {id} \"{e}\" at {ms}ms"),
    }
  }
}

时钟是暂停的，输出是确定的。M2 Max、rustc 1.99.0、--release，连跑 3 次输出相同：

    inner: req 1 starts at 0ms
    inner: req 1 done   at 100ms
    inner: req 2 starts at 100ms
caller: req 1 ok at 100ms
caller: req 2 "request timed out" at 150ms
caller: req 3 "request timed out" at 150ms

内层的日志里没有一个请求超过 100ms：请求 1 用了 100ms，请求 2 在 100ms 开始、做到一半被取消，请求 3 根本没到内层。可调用方三个里有两个超时。

源码里的原因

计时的起点在 Timeout::call：先 self.inner.call(request)，再 tokio::time::sleep(self.timeout)。所以问题变成：内层（这里是 Buffer）的 call 什么时候返回。

Buffer::call 只是把请求发进通道，立刻返回一个等 oneshot 的 future。三个任务在 0ms 都完成了 call，三个 Sleep 都从 0ms 开始。真正的排队发生在后面，在 worker 里：

时刻worker调用方
0ms请求 1 拿到许可，调内层，future 送回任务 1；请求 2 等许可，留在 worker 手里三个 Sleep 开始计时
100ms任务 1 的 future 完成，许可归还；请求 2 拿到许可，future 送回任务 2；请求 3 等许可任务 2 开始 poll 内层 future
150ms—任务 2、3 的 Sleep 到期，返回 Elapsed，响应 future 被 drop
150ms 之后请求 2 的 future 被 drop，许可归还；请求 3 的 oneshot 接收端已关闭，被跳过—

最后一行就是请求 3 没到内层的原因：worker 取消息时检查 msg.tx.is_closed()，调用方已经放弃的请求直接丢掉（见 Buffer 的「调用方放弃的请求」）。

所以 Timeout 在 Buffer 外面时，request timed out 的意思是「从交给 Buffer 到拿到响应超过了 150ms」，其中大部分可能是排队。内层的耗时日志、内层自己的指标都看不到这段时间。

反过来：等许可不算超时

把 Buffer 去掉，Timeout(150ms) > ConcurrencyLimit(1)，三个请求分别在 100、200、300ms 完成，一个都不超时（Timeout 的 A 组）。等许可发生在 ConcurrencyLimit::poll_ready 里，那时还没 call；Timeout::poll_ready 只是原样转发，不计时。等多久都不会触发 Timeout。RateLimit 用完次数后等下一个窗口，也发生在 poll_ready 里，同样不计。

修法

先决定「超时」要量的是哪一段，再决定放在哪里：

想量的时间做法
内层处理一个请求的时间Timeout 放在 Buffer 里面（.buffer(n).timeout(t)...），worker 调到它时才开始计时，即 Timeout 一页的 C 组
从交给 Buffer 到拿到响应，含通道和 worker 里的排队Timeout 放在 Buffer 外面，也就是上面的程序。这时超时多半说明排队太长，该看的是并发上限和 bound
从调用方开始等到拿到响应，含等许可、等窗口中间件做不到；在调用方用 tokio::time::timeout 同时包住 ready() 和 call

第三种写法，内层和上面一样每个请求 100ms，栈换成 Timeout(150ms) > ConcurrencyLimit(1)：

tower = { version = "=0.5.3", features = ["timeout", "limit", "util"] }
tokio = { version = "=1.54.1", features = ["full", "test-util"] }

use std::time::Duration;
use tokio::time::{Instant, sleep, timeout};
use tower::{BoxError, Service, ServiceBuilder, ServiceExt, service_fn};

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
  let start = Instant::now();
  let svc = ServiceBuilder::new()
    .timeout(Duration::from_millis(150))
    .concurrency_limit(1)
    .service(service_fn(|id: u32| async move {
      sleep(Duration::from_millis(100)).await;
      Ok::<_, BoxError>(id)
    }));

  let mut tasks = Vec::new();
  for id in 1..=3 {
    let mut svc = svc.clone();
    tasks.push(tokio::spawn(async move {
      // The deadline covers waiting for the permit, too.
      let r = timeout(Duration::from_millis(150), async {
        svc.ready().await?.call(id).await
      })
      .await;
      (id, r)
    }));
  }
  for t in tasks {
    let (id, r) = t.await.unwrap();
    let ms = start.elapsed().as_millis();
    match r {
      Ok(Ok(_)) => println!("req {id} ok at {ms}ms"),
      Ok(Err(e)) => println!("req {id} service error \"{e}\" at {ms}ms"),
      Err(e) => println!("req {id} caller deadline \"{e}\" at {ms}ms"),
    }
  }
}

连跑 3 次输出相同：

req 1 ok at 100ms
req 2 caller deadline "deadline has elapsed" at 150ms
req 3 caller deadline "deadline has elapsed" at 150ms

请求 3 在等许可时就被截断了，这是中间件里的 Timeout 做不到的。注意错误不一样：这里是 tokio 的 tokio::time::error::Elapsed（deadline has elapsed），类型上是外面那一层 Result 的 Err；中间件 Timeout 的是 tower 的 tower::timeout::error::Elapsed（request timed out），装在 BoxError 里，要 downcast_ref 才认得出（见 Timeout）。两者同名，按类型判断时别导错了模块。

📝NOTE
没有蠢问题

问： 为什么 tower 不提供一个连 poll_ready 也计时的超时中间件？

答： 中间件在 poll_ready 里超时，唯一能做的是返回 Err，而 poll_ready 的 Err 表示这个服务实例报废了（Service trait 的规则 3）。2022 年有人在 issue #648 里问过怎么给 poll_ready 计时，回答是交给调用方，写法就是上面的 tokio::time::timeout(timeout, svc.ready()) 或者包住 svc.oneshot(req)，理由是：「all it can really do is cause the whole stack to become defunct (by returning an error in poll_ready), so it's generally up to the caller to figure out how long it wants to wait.」同年的 PR #677 尝试加一个也管 poll_ready 的 GlobalTimeout，评审意见是「this strikes me as kind of a footgun, since a lot of Service implementations treat poll_ready returning an error as a terminal state for the service」，至今没有合并。放在 Buffer 里面，这样一个错误会让 worker 失败，之后所有请求都是 buffered service failed，见 buffer's worker closed unexpectedly。调用方的 timeout 只丢掉自己这一次等待，服务本身不受影响。

小结

  • Timeout 的计时从 call 开始（内层 call 返回之后）。Buffer::call 立即返回，所以 Timeout 在 Buffer 外面时，通道和 worker 里的排队全算进超时，内层每个请求都不慢也会 request timed out。
  • 超时的请求如果还在 worker 手里，会被 worker 丢掉，内层看不到它；已经交给内层的，内层 future 随响应 future 一起被 drop。
  • 等 ConcurrencyLimit 许可、等 RateLimit 窗口都发生在 poll_ready 里，Timeout 放在哪里都不计。
  • 只量处理时间：Timeout 放进 Buffer 里面。要量总时间：在调用方用 tokio::time::timeout 包住 ready() 和 call，错误是 tokio 的 Elapsed（deadline has elapsed），不是 tower 的。

💬 关联讨论

发起讨论
加载中...