一个请求穿过一摞中间件

主线位置：两条主线都在。本页用一个两层的中间件栈，把就绪主线和请求主线从头到尾走一遍；每一层的细节留给后面的章节。

flowchart TB
  subgraph t["Timeout：最外层"]
    subgraph c["ConcurrencyLimit"]
      s["ServiceFn：echo"]
    end
  end
  caller[调用方] -->|"ready 再 call"| t

生成了什么类型

先看 ServiceBuilder 叠出来的到底是什么。下面的程序打印 builder、服务和响应 future 三个值的类型名：

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

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

fn type_of<T>(_: &T) -> &'static str {
  std::any::type_name::<T>()
}

async fn echo(req: String) -> Result<String, BoxError> {
  Ok(req)
}

#[tokio::main]
async fn main() -> Result<(), BoxError> {
  let builder = ServiceBuilder::new()
    .timeout(Duration::from_secs(1))
    .concurrency_limit(8);
  println!("builder: {}", type_of(&builder));

  let mut svc = builder.service(service_fn(echo));
  println!("service: {}", type_of(&svc));

  svc.ready().await?;
  let fut = svc.call("hi".to_string());
  println!("future:  {}", type_of(&fut));
  println!("size:    {} bytes (Sleep alone: {} bytes)", std::mem::size_of_val(&fut), std::mem::size_of::<tokio::time::Sleep>());
  println!("output:  {:?}", fut.await?);
  Ok(())
}

输出（M2 Max、rustc 1.99.0，跑 3 次相同）：

builder: tower::builder::ServiceBuilder<tower_layer::stack::Stack<tower::limit::concurrency::layer::ConcurrencyLimitLayer, tower_layer::stack::Stack<tower::timeout::layer::TimeoutLayer, tower_layer::identity::Identity>>>
service: tower::timeout::Timeout<tower::limit::concurrency::service::ConcurrencyLimit<tower::util::service_fn::ServiceFn<stack_type::echo>>>
future:  tower::timeout::future::ResponseFuture<tower::limit::concurrency::future::ResponseFuture<stack_type::echo::{{closure}}>>
size:    160 bytes (Sleep alone: 112 bytes)
output:  "hi"

三行类型名各说明一件事。

builder 的类型里，后写的在外面。 .timeout() 先写，却在 Stack 的最里面挨着 Identity；.concurrency_limit() 后写，在最外面。原因在 ServiceBuilder::layer，每个快捷方法最后都调它：

    pub fn layer<T>(self, layer: T) -> ServiceBuilder<Stack<T, L>> {
        ServiceBuilder {
            layer: Stack::new(layer, self.layer),
        }
    }

新加的 layer 成了 Stack 的 inner，之前攒下的 self.layer 成了 outer。

服务的类型里，先写的在外面。 到了 Stack 的 Layer 实现，顺序翻了过来：

    fn layer(&self, service: S) -> Self::Service {
        let inner = self.inner.layer(service);

        self.outer.layer(inner)
    }

inner（最后写的 ConcurrencyLimitLayer）先包住 echo，outer 再包住结果。一层层递归下去，最先写的 TimeoutLayer 最后才包，所以在最外层。结论就是一句话：ServiceBuilder 里先写的，请求先经过。Stack 和 ServiceBuilder 的细节见 Layer 与 Stack 和 ServiceBuilder。

响应 future 的类型和服务一一对应。 每层中间件都有自己的 ResponseFuture，把内层的 future 当作字段包起来：Timeout 的 future 包着 ConcurrencyLimit 的 future，后者包着 echo 这个 async fn 的状态机。整个 future 是一个 160 字节的值，其中 112 字节是 Timeout 带的那个 Sleep。没有 Box，每一层的 poll 都是静态分发。代价就是上面那串类型名，以及更长的编译时间。

两条主线的时序图

假设另一个请求正占着并发许可（上限设成 1 更容易想象）。一个调用方先 ready().await，再 call，再等响应：

sequenceDiagram
  participant U as 调用方
  participant T as Timeout
  participant C as ConcurrencyLimit
  participant M as 信号量
  participant S as ServiceFn
  Note over U,S: 就绪主线：poll_ready 往里传
  U->>T: poll_ready
  T->>C: poll_ready（原样转发）
  C->>M: poll_acquire
  M-->>C: Pending，记下 waker
  C-->>T: Pending
  T-->>U: Pending，ready() 挂起
  Note over M: 别的请求完成，许可归还，wake
  U->>T: poll_ready
  T->>C: poll_ready
  C->>M: poll_acquire
  M-->>C: 许可，存进 self.permit
  C->>S: poll_ready
  S-->>C: Ready
  C-->>T: Ready
  T-->>U: Ready
  Note over U,S: 请求主线：future 往外包
  U->>T: call(req)
  T->>C: call(req)
  C->>S: call(req)，许可从 self.permit 取出
  S-->>C: echo 的 future
  C-->>T: ResponseFuture，内含许可
  T-->>U: ResponseFuture，再加一个 Sleep
  U->>T: poll
  T->>C: 先 poll 内层
  C->>S: poll
  S-->>C: Ready(resp)
  C-->>T: Ready(resp)
  T-->>U: Ready(resp)，Sleep 没机会触发
  Note over U,C: future 被 drop，许可回到信号量

图里有三处值得对照源码。

Timeout 不管就绪。 Timeout::poll_ready 只是转发内层的结果，把错误转成 BoxError；Sleep 是在 Timeout::call 里才创建的。所以在 ready() 上等许可的时间，不算进超时。ResponseFuture::poll 每次先 poll 内层 future，内层没好再看 Sleep。细节见 Timeout。

ConcurrencyLimit 在 poll_ready 里占资源，在 call 里交出去。 poll_ready 手里没有许可就先向信号量要，要到了存进 self.permit，再问内层；call 把许可 take() 出来，和内层的 future 一起放进 ResponseFuture 的 _permit 字段。许可跟着 future 走，future 被 drop（不管是完成后还是中途放弃），许可才回到信号量。等许可时的唤醒由 tokio 的信号量负责，见 batch_semaphore；ConcurrencyLimit 的其余细节见 ConcurrencyLimit。

ServiceFn 永远就绪。 ServiceFn::poll_ready 直接返回 Ok(())，call 调用闭包得到 future。它是栈的最底层，没有资源要占。

就绪主线上，信号从信号量出发，经 ConcurrencyLimit、Timeout 一路回到调用方；请求主线上，future 从 ServiceFn 出发，每经过一层多包一层，响应再从里往外一层层解开。全卷后面的每个中间件，都是在这两条线上的某一段加自己的逻辑。

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

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

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

  let mut a = svc.clone();
  let mut b = svc.clone();
  a.ready().await.unwrap();
  let fut_a = a.call(80); // A 先拿到唯一的许可，处理 80ms

  let task_b = tokio::spawn(async move {
    b.ready().await.unwrap(); // B 在这里等 A 归还许可
    let ready_at = start.elapsed().as_millis();
    let res = b.call(60).await; // B 自己处理 60ms
    (ready_at, res.map_err(|e| e.to_string()))
  });

  let res_a = fut_a.await;
  println!("A: {:?} at {}ms", res_a.map_err(|e| e.to_string()), start.elapsed().as_millis());
  let (ready_at, res_b) = task_b.await.unwrap();
  println!("B: ready at {ready_at}ms, {:?} at {}ms", res_b, start.elapsed().as_millis());
}

动脑筋答案

A: Ok(80) at 80ms
B: ready at 80ms, Ok(60) at 140ms

跑了 3 次，输出相同。B 没有超时：

• A: Ok(80) at 80ms：A 处理 80ms，正常返回，它的 future 被 drop，许可归还。
• B: ready at 80ms：B 的前 80ms 全花在 ready() 上，也就是 ConcurrencyLimit::poll_ready 等许可。Timeout::poll_ready 只是转发，这段时间没有计时。
• Ok(60) at 140ms：B 的 call 在 80ms 发出，Timeout::call 这时才创建 100ms 的 Sleep；内层 60ms 就完成了，离超时还差 40ms。

所以把 Timeout 放在 ConcurrencyLimit 外面，限制的只是「拿到许可之后」的处理时间，排队等许可的时间要另外管。这对 Buffer 不成立：Buffer 的 call 会立刻返回，等内层就绪发生在 worker 里，放在 Timeout 里面还是外面结果不同，见 Buffer。

小结

• ServiceBuilder::new().timeout(..).concurrency_limit(..).service(svc) 的类型是 Timeout<ConcurrencyLimit<S>>：先写的在最外层，请求先经过。
• builder 的 Stack 类型里顺序是反的（后写的在外面），Stack::layer 先套 inner 再套 outer，又翻了回来。
• 响应 future 同样一层套一层，整个是一个值，没有 Box。
• 就绪主线：Timeout 转发，ConcurrencyLimit 取许可并存在 self.permit，ServiceFn 总是 Ready。
• 请求主线：ConcurrencyLimit 把许可移进 future，Timeout 加一个从 call 开始计时的 Sleep；等许可的时间不算进超时。