Timeout

主线位置：请求主线。call 把内层 future 和一个 Sleep 包在一起；就绪主线上它只是转发。

flowchart TB
  pr["poll_ready"] --> fwd[原样问内层]
  call["call(req)"] --> ic[内层 call]
  ic --> sl[建 Sleep 开始计时]
  sl --> rf[ResponseFuture]
  rf --> p1{内层完成？}
  p1 -->|是| out[返回内层结果]
  p1 -->|否| p2{Sleep 到期？}
  p2 -->|否| pend[Pending]
  p2 -->|是| el[返回 Elapsed]

Timeout 是 tower 里最短的中间件之一，timeout/ 目录下四个文件加起来不到 200 行。它值得单独一页，是因为「超时从什么时候开始算」这个问题，答案不在 Timeout 自己，而在它周围的中间件。

结构：一个内层，一个时长

Timeout：

#[derive(Debug, Clone)]
pub struct Timeout<T> {
    inner: T,
    timeout: Duration,
}

没有截止时间，没有计时器。Clone 是 derive 的：服务身上没有任何就绪状态，复制一份完全没问题。TimeoutLayer 只存一个 Duration，layer 就是 Timeout::new(service, self.timeout)。

poll_ready：只转发

poll_ready 和 call：

    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        match self.inner.poll_ready(cx) {
            Poll::Pending => Poll::Pending,
            Poll::Ready(r) => Poll::Ready(r.map_err(Into::into)),
        }
    }

    fn call(&mut self, request: Request) -> Self::Future {
        let response = self.inner.call(request);
        let sleep = tokio::time::sleep(self.timeout);

        ResponseFuture::new(response, sleep)
    }

poll_ready 唯一做的事是把内层的错误类型转成 BoxError。内层 Pending 多久，它就 Pending 多久，不计时，也不会因为等太久而报错。

Sleep 在 call 里创建，而且在 self.inner.call(request) 之后。所以计时的起点是「内层已经接下这个请求、返回了 future」的那一刻。内层 call 本身如果同步地做了很多事，这段时间也不算。

📝NOTE

没有蠢问题

问：我想限制「从开始等到拿到响应」的总时间，包括等内层就绪。Timeout 为什么不管 poll_ready？

答：它没有地方放这个起点。结构体里只有 inner 和 timeout，poll_ready 可能被调用很多次，也可能 Ready 之后很久才 call，甚至永远不 call。Sleep 只能挂在一个具体请求的 future 上，而请求要到 call 才存在。要给等就绪也设上限，在调用方那一侧把 svc.ready() 这个 future 用 tokio::time::timeout 包起来。

ResponseFuture：先内层，后 Sleep

ResponseFuture 的两个字段都 pin 住，poll：

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let this = self.project();

        // First, try polling the future
        match this.response.poll(cx) {
            Poll::Ready(v) => return Poll::Ready(v.map_err(Into::into)),
            Poll::Pending => {}
        }

        // Now check the sleep
        match this.sleep.poll(cx) {
            Poll::Pending => Poll::Pending,
            Poll::Ready(_) => Poll::Ready(Err(Elapsed(()).into())),
        }
    }

两个都用同一个 cx 去 poll，所以内层的 waker 和定时器的 waker 都指向调用方的任务，谁先到谁把任务叫醒。顺序是先内层后 Sleep：同一次 poll 里两边都已经就绪时，响应赢。这和 tokio 自己的 tokio::time::timeout 是同一个顺序。Sleep 怎么登记到时间轮、到期后怎么叫醒任务，见那一页。

超时后，内层 future 不会被显式取消，它随 ResponseFuture 一起被 drop。内层如果是 ConcurrencyLimit 的 future，许可就在这时归还。

错误：装箱的 Elapsed

Elapsed：

#[derive(Debug, Default)]
pub struct Elapsed(pub(super) ());

impl fmt::Display for Elapsed {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.pad("request timed out")
    }
}

Timeout 的 Error 是 crate::BoxError，也就是 Box<dyn Error + Send + Sync>。内层的错误经 Into 装箱，超时则装箱一个 Elapsed。两种错误进了同一个类型，调用方要区分只能 downcast_ref::<Elapsed>()：

match svc.ready().await?.call(req).await {
  Ok(rsp) => { /* ... */ }
  Err(e) if e.is::<tower::timeout::error::Elapsed>() => { /* timed out */ }
  Err(e) => { /* inner error */ }
}

外层中间件通常也是 Error = BoxError，靠 Into<BoxError> 转换错误；BoxError 转 BoxError 走的是标准库的 impl<T> From<T> for T，原样返回，不会再套一层 Box。所以 Elapsed 穿过多少层都还能 downcast 出来，除非某一层把它包进了自己定义的错误类型。为什么 tower 的错误都统一成 BoxError，见第 08 部分。

📝NOTE

冷知识：Elapsed::new 是为了测试重试策略才加的

Elapsed 的字段是 pub(super) ()，模块外面构造不出来。2019 年 7 月以前，它也没有公开的构造函数，tower 之外的代码想要一个 Elapsed 值，只能真的等一次超时。PR #308 加了 pub fn new() -> Self，PR 描述只有一句：「The motivation for this change is to support the ability to test that my tower-retry policy will properly retry on Elapsed errors.」也就是让重试策略的单元测试能直接造一个超时错误。这个改动随独立的 tower-timeout 0.1.1 发布，tower 本体里第一个包含它的版本是 0.3.0。现在它是 const fn（#760 加的 const），结构体上也多了 derive(Default)（#505 修 clippy 警告时加的）。三年后 LoadShed 的 Overloaded::new 也因为同样的理由公开（PR #661：「This allows for mocking. This also matches what is done for the timeout Elapsed error.」）。

排队时间算不算

Timeout 只量「call 之后」。所以问题变成：排队发生在 call 之前还是之后？这取决于外面是谁。

ConcurrencyLimit：等许可发生在 poll_ready 里（见 ConcurrencyLimit）。不管 Timeout 在它外面还是里面，等许可时都还没 call，不计时。
Buffer：poll_ready 只在通道里预留一个位置，call 立即返回一个 future；真正等内层就绪、调用内层，发生在后台 worker 里（见 Buffer）。Timeout 在 Buffer 外面，call 一返回就开始计时，排队的时间全算进去；Timeout 在 Buffer 里面，worker 调到它时才开始计时，排队不算。

实验：内层每个请求耗时 100ms，前面放一个并发上限 1 的 ConcurrencyLimit，超时 150ms，三个任务同时发请求。时钟是暂停的（start_paused），输出是确定的。

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

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

// each request takes 100ms; at most one runs at a time
async fn work(id: u32) -> Result<u32, BoxError> {
  sleep(Duration::from_millis(100)).await;
  Ok(id)
}

async fn run<S>(name: &str, svc: S)
where
  S: Service<u32, Response = u32, Error = BoxError> + Clone + Send + 'static,
  S::Future: Send,
{
  let start = Instant::now();
  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))
    }));
  }
  println!("{name}");
  for t in tasks {
    let (id, r) = t.await.unwrap().unwrap();
    let ms = start.elapsed().as_millis();
    match r {
      Ok(v) => println!("  req {id}: ok {v} at {ms}ms"),
      Err(e) => println!(
        "  req {id}: err \"{e}\", is Elapsed = {}, at {ms}ms",
        e.downcast_ref::<Elapsed>().is_some()
      ),
    }
  }
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
  let t = Duration::from_millis(150);

  let a = ServiceBuilder::new()
    .timeout(t)
    .concurrency_limit(1)
    .service_fn(work);
  run("A: Timeout > ConcurrencyLimit(1)", a).await;

  let b = ServiceBuilder::new()
    .timeout(t)
    .buffer(8)
    .concurrency_limit(1)
    .service_fn(work);
  run("B: Timeout > Buffer > ConcurrencyLimit(1)", b).await;

  let c = ServiceBuilder::new()
    .buffer(8)
    .timeout(t)
    .concurrency_limit(1)
    .service(service_fn(work));
  run("C: Buffer > Timeout > ConcurrencyLimit(1)", c).await;
}

ServiceBuilder 先写的在外层（见 ServiceBuilder），> 表示「套在外面」。

❗IMPORTANT

动脑筋：三种摆法里，各有几个请求超时？

三个请求一共要 300ms，每个只给 150ms。A 里 Timeout 直接套在 ConcurrencyLimit 外面；B 里中间隔了一个 Buffer；C 把 Buffer 挪到了 Timeout 外面。先猜 A、B、C 各有几个 request timed out，答案在文末。

小心「超时」两个字

把上面的结论换成配置时的说法：

想要的语义	摆法
只限制内层处理一个请求的时间	`Timeout` 放在最里面，或者至少在 `Buffer` 里面
限制「交给这一摞服务之后」到拿到响应的总时间	`Timeout` 放在 `Buffer` 外面
连等 `ConcurrencyLimit` 许可的时间也限制	`Timeout` 做不到，在调用方用 `tokio::time::timeout` 包住 `ready()` 和响应

动脑筋答案

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

A: Timeout > ConcurrencyLimit(1)
  req 1: ok 1 at 100ms
  req 2: ok 2 at 200ms
  req 3: ok 3 at 300ms
B: Timeout > Buffer > ConcurrencyLimit(1)
  req 1: ok 1 at 100ms
  req 2: err "request timed out", is Elapsed = true, at 150ms
  req 3: err "request timed out", is Elapsed = true, at 150ms
C: Buffer > Timeout > ConcurrencyLimit(1)
  req 1: ok 1 at 100ms
  req 2: ok 2 at 200ms
  req 3: ok 3 at 300ms

A：0 个。 请求 2、3 在 ready().await 里等许可，这时还没 call，Timeout 没开始计时。拿到许可后 call，内层 100ms 完成，远小于 150ms。三个请求分别在 100、200、300ms 完成。
B：2 个。 Buffer::poll_ready 只预留通道位置，三个任务在 0ms 就都 call 了，三个 Sleep 都从 0ms 开始。worker 一次只能把一个请求送进 ConcurrencyLimit(1)，请求 2 要排到 100ms 才开始处理，200ms 才能完成，150ms 时就超时了，请求 3 同理。错误能 downcast_ref::<Elapsed>()。
C：0 个。 Timeout 在 Buffer 里面，worker 等到 ConcurrencyLimit 放行、调用 Timeout::call 时才开始计时，每个请求都只量到自己的 100ms。

B 里超时的两个请求，调用方手里的响应 future 被 drop，等 worker 回传的 oneshot 接收端也随之关闭；worker 之后怎么对待这样的请求，见 Buffer。

小结

Timeout 只有 inner 和 timeout 两个字段，没有就绪状态，Clone 直接 derive。
poll_ready 原样转发，等就绪多久都不计时。Sleep 在 call 里、内层 call 之后创建。
响应 future 先 poll 内层再 poll Sleep，同时就绪时响应赢。超时后内层 future 随之被 drop。
错误统一是 BoxError，超时是装箱的 Elapsed（request timed out），用 is / downcast_ref 区分。
排队算不算超时：等 ConcurrencyLimit 许可发生在 poll_ready，永远不算；Buffer 的 call 立即返回，Timeout 在 Buffer 外面就算，在里面就不算。