service_fn 与 map 一族

主线位置：几乎全在请求主线上。map_request 改进去的请求，其余的改出来的 future 或它的结果；就绪主线上它们都原样转发 poll_ready，最多把错误换个类型。

flowchart TB
  req[请求] -->|map_request| inner["内层 call"]
  inner --> fut[内层 future]
  fut -->|map_future| whole["换掉整个 future"]
  fut --> res["Result 出来"]
  res -->|map_response| okv["只改 Ok"]
  res -->|map_err| errv["只改 Err"]
  res -->|map_result| both["Ok 和 Err 都改"]
  res -->|then| thenf["再跑一个 future"]
  res -->|and_then| andf["Ok 时再跑 future"]

这些组合子都在 ServiceExt 上，都由 util feature 打开，每个都对应一个 Layer（MapRequestLayer 等），ServiceBuilder 上也有同名方法，见 ServiceBuilder。

service_fn：永远就绪

service_fn 只是把闭包存进 ServiceFn，Service 实现是全卷最短的一个：

#[derive(Copy, Clone)]
pub struct ServiceFn<T> {
    f: T,
}

impl<T, F, Request, R, E> Service<Request> for ServiceFn<T>
where
    T: FnMut(Request) -> F,
    F: Future<Output = Result<R, E>>,
{
    type Response = R;
    type Error = E;
    type Future = F;

    fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), E>> {
        Ok(()).into()
    }

    fn call(&mut self, req: Request) -> Self::Future {
        (self.f)(req)
    }
}

• poll_ready 不看 cx，永远 Ready(Ok(()))。
• Future 就是闭包返回的那个类型 F。async move { .. } 块的类型是编译器生成的匿名类型，ServiceFn 原样交出去，不包、不装箱。
• Request 是 impl 上的泛型参数，所以一个闭包只要参数类型推得出来就能用；T: FnMut 让闭包可以改自己捕获的状态。
• Clone / Copy 是 derive 的，闭包能 Clone 它就能 Clone。04 部分的实验里大量用 service_fn 当内层，正是因为它不带任何就绪状态，复制一份也不会出错。

一张表：每个组合子改哪一段

下表的「闭包约束」取自各自的 Service 实现；「future 类型」是 call 返回的类型。

有两组对照值得记住。

闭包什么时候被调用。 map_request 和 map_future 的闭包在 call 里当场调用，所以是 FnMut，存一份就够。其余五个的闭包要等内层 future 完成才用得上，而那时 call 早就返回了，所以每次 call 都 self.f.clone() 一份移进 future，future 用完即弃，于是约束是 FnOnce + Clone。MapResponse::call 就一行：

    fn call(&mut self, request: Request) -> Self::Future {
        MapResponseFuture::new(self.inner.call(request).map_ok(self.f.clone()))
    }

闭包捕获了大对象时，每个请求都要复制一次；捕获 Arc 就只是加一次引用计数。

map_future 能做别的组合子做不到的事。 其他几个只看得到 future 的输出，map_future 拿到的是 future 本身，可以给它套一层再交回去，比如加超时、加 tracing 的 span。#542 加它时的动机正是这个：「I wanted to write a middleware that wraps all futures produced by a service in a tracing::Span. So something like self.inner.call(req).instrument(span).」下一页的 BoxService 也是用它把 future 装箱的。

poll_ready 的错误由谁转换

poll_ready 返回的是 Result<(), Self::Error>。组合子一旦改了 Error 类型，就绪主线上的错误也得跟着换类型。七个组合子在这里分成两派：

• MapErr::poll_ready 把同一个闭包也用在 poll_ready 的错误上：self.inner.poll_ready(cx).map_err(self.f.clone())。
• 其余改错误类型的组合子，poll_ready 的错误只走 From / Into，比如 MapResult::poll_ready 是 .map_err(Into::into)。闭包的参数是 Result<S::Response, S::Error>，poll_ready 根本没有 Response 可给，只能另找一条转换路径，这就是 Error: From<S::Error> 这条约束的来历。map_result 的文档写明了这一点：「because the map_result function is not applied to the errors returned by the service's poll_ready method, it must be possible to convert the service's Error type into the error type returned by the map_result function」。

map_err 的文档说它「does not alter the behaviour of the poll_ready method」。这句话只对一半：何时 Pending、何时 Ready 确实不变，但错误值会经过你的闭包。闭包里如果有副作用（打日志、计数），就绪失败也会触发。

tower = { version = "=0.5.3", features = ["util"] }
futures = "0.3"

use std::future::{ready, Ready};
use std::task::{Context, Poll};
use tower::{Service, ServiceExt};

// poll_ready fails when `broken`; call(0) fails, other requests echo
#[derive(Clone)]
struct Flaky {
  broken: bool,
}

impl Service<u32> for Flaky {
  type Response = u32;
  type Error = String;
  type Future = Ready<Result<u32, String>>;

  fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), String>> {
    if self.broken {
      Poll::Ready(Err("ready-err".to_string()))
    } else {
      Poll::Ready(Ok(()))
    }
  }

  fn call(&mut self, req: u32) -> Self::Future {
    if req == 0 { ready(Err("call-err".to_string())) } else { ready(Ok(req)) }
  }
}

fn probe<S>(name: &str, mk: impl Fn(bool) -> S)
where
  S: Service<u32>,
  S::Error: std::fmt::Debug,
  S::Response: std::fmt::Debug,
{
  let waker = futures::task::noop_waker();
  let mut cx = Context::from_waker(&waker);
  let mut broken = mk(true);
  let r = broken.poll_ready(&mut cx);
  let mut ok = mk(false);
  let _ = ok.poll_ready(&mut cx);
  let c = futures::executor::block_on(ok.call(0));
  println!("{name:<12} poll_ready: {r:?}");
  println!("{:<12} call(0):    {c:?}", "");
}

fn main() {
  probe("map_err", |b| Flaky { broken: b }.map_err(|e: String| format!("map_err({e})")));
  probe("map_result", |b| {
    Flaky { broken: b }.map_result(|r: Result<u32, String>| r.map_err(|e| format!("map_result({e})")))
  });
  probe("then", |b| {
    Flaky { broken: b }.then(|r: Result<u32, String>| async move { r.map_err(|e| format!("then({e})")) })
  });
  probe("and_then", |b| {
    Flaky { broken: b }.and_then(|v: u32| async move { Ok::<_, String>(v + 1) })
  });
  probe("map_future", |b| {
    Flaky { broken: b }.map_future(|f| async move {
      f.await.map_err(|e| format!("map_future({e})"))
    })
  });
}

future 类型：一个都不装箱

Service::Future 是关联类型，组合子必须给出一个具体类型。tower 的做法分三种，下面的程序把每种组合子的 future 类型名（只取最外层）和大小打印出来：

tower = { version = "=0.5.3", features = ["util"] }

use std::any::type_name;
use std::convert::Infallible;
use std::future::ready;
use std::mem::size_of;
use tower::{service_fn, Service, ServiceExt};

fn show<S: Service<u32>>(name: &str, _svc: &S) {
  let full = type_name::<S::Future>();
  // keep the outermost path only, the generic arguments are long
  let head = full.split('<').next().unwrap_or(full);
  println!("{name:<12} {:>3} bytes  {head}", size_of::<S::Future>());
}

fn main() {
  let echo = service_fn(|x: u32| ready(Ok::<u32, Infallible>(x)));
  show("service_fn", &echo);
  show("map_request", &echo.map_request(|x: u32| x + 1));
  show("map_response", &echo.map_response(|x| x + 1));
  show("map_err", &echo.map_err(|e| e));
  show("map_result", &echo.map_result(|r| r));
  show("map_future", &echo.map_future(|f| f));
  show("then", &echo.then(|r| ready(r)));
  show("and_then", &echo.and_then(|x| ready(Ok::<u32, Infallible>(x))));
  show("boxed", &echo.boxed());
}

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

service_fn     8 bytes  core::future::ready::Ready
map_request    8 bytes  core::future::ready::Ready
map_response   8 bytes  tower::util::map_response::MapResponseFuture
map_err        8 bytes  tower::util::map_err::MapErrFuture
map_result     8 bytes  tower::util::map_result::MapResultFuture
map_future     8 bytes  core::future::ready::Ready
then          12 bytes  tower::util::then::ThenFuture
and_then      12 bytes  tower::util::and_then::AndThenFuture
boxed         16 bytes  core::pin::Pin

第一种：直接沿用别人的类型。 map_request 只改请求，future 就是内层的 S::Future；map_future 的 future 是闭包返回的 Fut（这里闭包原样返回，所以还是 Ready）。它们没有自己的 future 类型，tower::util::future 模块里也就没有 MapRequestFuture、MapFutureFuture。

第二种：包着 futures_util 类型的 newtype。 map_response、map_err、map_result、then 四个都是一行宏调用，比如 MapErrFuture：

opaque_future! {
    /// Response future from [`MapErr`] services.
    ///
    /// [`MapErr`]: crate::util::MapErr
    pub type MapErrFuture<F, N> = future::MapErr<F, N>;
}

写法像类型别名，展开后却是一个新结构体。opaque_future! 用 pin_project! 生成 pub struct MapErrFuture<F, N> { #[pin] inner: future::MapErr<F, N> }，Future::poll 只是 self.project().inner.poll(cx)，Debug 只打印 "..."。包这一层没有运行时开销：大小和内层一样是 8 字节（闭包不捕获东西，大小为 0）。

第三种：手写。 AndThenFuture 做的事和宏一样，但宏的参数写的是 $($param:ident),+，只能接光秃秃的类型参数；AndThenFuture 的内层类型是 future::AndThen<future::ErrInto<F1, F2::Error>, F2, N>，要写 F2::Error 就得在参数上带约束 F2: TryFuture，宏表达不了，只能手写一遍。

then 和 and_then 多出 4 字节，是因为 futures_util 里它们是两段式的状态机（Then 是 Flatten<Map<Fut1, F>, Fut2>）：先等内层 future，再等闭包返回的 future，要记住自己在哪一段。

三种都不装箱，所以一摞组合子的 future 是一个嵌套的结构体，整体放在调用方的栈上或调用方的 future 里，不需要每个请求分配一次堆内存，Send 也由编译器按内层自动推导。最后一行 boxed 是对照：Pin<Box<dyn Future>> 是一个胖指针，16 字节，真正的 future 在堆上。什么时候值得付这个代价，见 类型擦除。

Layer 版本

七个组合子各有一个 XxxLayer，结构都是存一份闭包，layer 时 self.f.clone() 塞进新服务，所以 Layer 实现都要求 F: Clone。比如 MapRequestLayer：

impl<S, F> Layer<S> for MapRequestLayer<F>
where
    F: Clone,
{
    type Service = MapRequest<S, F>;

    fn layer(&self, inner: S) -> Self::Service {
        MapRequest {
            f: self.f.clone(),
            inner,
        }
    }
}

Layer 上没有对 S 和请求类型的任何约束，类型对不上的错误要等到把栈当 Service 用时才报出来。怎么把这种错误定位到具体哪一层，见 ServiceBuilder 里的 check_service。

动脑筋答案

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

map_err      poll_ready: Ready(Err("map_err(ready-err)"))
             call(0):    Err("map_err(call-err)")
map_result   poll_ready: Ready(Err("ready-err"))
             call(0):    Err("map_result(call-err)")
then         poll_ready: Ready(Err("ready-err"))
             call(0):    Err("then(call-err)")
and_then     poll_ready: Ready(Err("ready-err"))
             call(0):    Err("call-err")
map_future   poll_ready: Ready(Err("ready-err"))
             call(0):    Err("map_future(call-err)")

改写过的有 5 行：

• map_err：两行都被改写。poll_ready 的错误也经过了闭包，这是唯一这样做的组合子。
• map_result、then、map_future：call(0) 的错误被闭包改写，poll_ready 的错误原样出来。这里闭包的错误类型还是 String，From<String> for String 就是原样返回；如果换成别的错误类型，这一行就是 From 转换的结果，你的闭包照样不会被调用。
• and_then：两行都没改。它的闭包只在 Ok 时调用，call(0) 的 Err("call-err") 直接穿过去，poll_ready 的错误也只走 Into。

小结

• service_fn 永远就绪，future 就是闭包返回的类型；需要背压就在外面套中间件，或者手写 poll_ready。
• map_request 改请求，map_response / map_err / map_result 改结果，then / and_then 在结果之后再跑一个 future，map_future 换掉整个 future。
• 在 future 完成时才用的闭包要 FnOnce + Clone，每次 call 复制一份；map_request、map_future 的闭包在 call 里当场用，是 FnMut。
• poll_ready 的错误只有 map_err 会经过闭包；map_result、then、and_then、map_future 只做 From / Into 转换。
• 没有一个组合子装箱：map_request、map_future 沿用现成的 future 类型，四个用 opaque_future! 包住 futures_util 类型（#509，为了 API 稳定），AndThenFuture 手写。