Hyper 与 Tower：工业级 HTTP 栈

第7章 Balance / Discover / ready_cache：负载均衡抽象

作者 杨艺韬 · 12,642 字 · 发布于 · 更新于

7.1 从"一个服务"到"一群服务"

前六章我们讨论的所有中间件——Timeout、Retry、Buffer、LoadShed、ConcurrencyLimit——都围绕一个隐含假设：底层只有一个 Service。到这里我们要打破这个假设。

真实的分布式系统几乎总有多个同等能力的后端。你调用 user-service 其实是调用一个由十几个 Pod 组成的集群中的某一个；你的 gRPC 客户端背后是一个经过 DNS 轮询或 service mesh 发现的服务端点池。

这就引出三个相互咬合的问题：

  1. 怎么表达"后端集合"？Pod 可能随时被调度、新 Pod 可能随时被加进来、旧 Pod 可能因为 health check 失败被剔除——这是动态的。
  2. 怎么选一个后端？随机？轮询？按负载？按延迟？
  3. 怎么保证选到的那个后端"现在"能接请求？已经满了的后端要跳过，已经挂了的要不再选。

Tower 给这三件事提供了三个抽象：

抽象 作用
Discover trait 动态后端集合的"变更流"
ReadyCache<K, S, Req> 维护"就绪 + 待就绪"两个池子
Balance<D, Req> 在 ready 池子里选一个（默认 P2C 算法）

这一章我们把三者串起来读。真实源码全在 tower-0.5.3/tower/src/discover/、tower-0.5.3/tower/src/ready_cache/、tower-0.5.3/tower/src/balance/p2c/，版本 0.5.3。

7.2 Discover：服务集合的"变更流"

// tower-0.5.3/tower/src/discover/mod.rs:55-74
pub trait Discover: Sealed<Change<(), ()>> {
    type Key: Eq;
    type Service;
    type Error;

    fn poll_discover(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Result<Change<Self::Key, Self::Service>, Self::Error>>>;
}

语义：poll_discover 每次 poll 返回一个"集合变更事件"。它不是 "Vec<Service>"，也不是 "HashMap<Key, Service>"——而是流式的增量：

// tower-0.5.3/tower/src/discover/mod.rs:100-107
pub enum Change<K, V> {
    Insert(K, V),   // 新来了一个 K 对应的服务
    Remove(K),      // K 对应的服务消失了
}

为什么这样设计？因为服务发现在现代系统里从来不是"一次性拿到全量列表"的。Kubernetes 的 Endpoints API 用 watch 模式返回一串增量事件；Consul 的 service discovery 用 long polling；DNS SRV 记录有 TTL 要不断重查。把这些异构来源抽象成"一个流"，下游（balancer）就不用关心"这些变化从哪来"——它只看到 Insert / Remove 的连续事件。

Discover trait 的另一个妙处是：它通过 blanket impl 自动覆盖了所有 TryStream<Ok = Change<K, S>>：

// tower-0.5.3/tower/src/discover/mod.rs:83-98
impl<K, S, E, D: ?Sized> Discover for D
where D: TryStream<Ok = Change<K, S>, Error = E>, K: Eq,
{
    type Key = K;
    type Service = S;
    type Error = E;
    fn poll_discover(self: Pin<&mut Self>, cx: &mut Context<'_>)
        -> Poll<Option<Result<D::Ok, D::Error>>>
    {
        TryStream::try_poll_next(self, cx)
    }
}

这意味着你只要有一个能产生 Stream<Item = Result<Change, Error>> 的东西——无论是 async_stream! 包的 coroutine、是 mpsc receiver、是从 Kubernetes API watch 出来的事件流——都自动满足 Discover trait。这种"把一个新抽象嫁接到已经成熟的 Stream 生态上"的设计非常体贴：你不需要为 Tower 专门实现什么，现成的东西就能用。

7.2.1 最简单的 Discover：ServiceList

如果你不需要真正的动态发现——只是想把一个固定列表喂给 balancer——tower 提供了 ServiceList：

// tower-0.5.3/tower/src/discover/list.rs:18 精简版
pin_project! {
    pub struct ServiceList<T>
    where T: IntoIterator,
    {
        inner: Enumerate<T::IntoIter>,
    }
}

impl<T, U> ServiceList<T>
where T: IntoIterator<Item = U>,
{
    pub fn new<Request>(services: T) -> ServiceList<T>
    where U: Service<Request>,
    {
        ServiceList { inner: services.into_iter().enumerate() }
    }
}

impl<T, U> Stream for ServiceList<T>
where T: IntoIterator<Item = U>,
{
    type Item = Result<Change<usize, U>, Infallible>;

    fn poll_next(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        match self.project().inner.next() {
            Some((i, service)) => Poll::Ready(Some(Ok(Change::Insert(i, service)))),
            None => Poll::Ready(None),
        }
    }
}

注意它实现的是 Stream 而不是 Discover——正是靠上面那个 blanket impl 才成为 Discover。它的 poll_next 就是不停地 Insert(i, next_svc)（key 是 enumerate 出来的下标），遍历完之后 stream 结束——对 balancer 来说等价于"这 N 个端点进来，之后永不变化"。适合配置写死的多实例、或者测试场景。

7.2.2 Sealed<Change<(), ()>> 让 trait 无法被外部 impl

看 Discover trait 的真实定义（tower-0.5.3/tower/src/discover/mod.rs:55）：

pub trait Discover: Sealed<Change<(), ()>> {
    type Key: Eq;
    // ...
}

Sealed<Change<(), ()>> 是一个私有 trait——定义在 tower 内部的 mod、用户无法 impl。这就形成了 sealed trait 模式：

  • 用户能使用 Discover trait（写泛型约束、调 poll_discover）
  • 用户无法实现 Discover trait（因为 Sealed 是私有的）
  • 但 Tower 的 blanket impl impl<D: TryStream<..>> Discover for D 让所有 TryStream 自动成为 Discover

为什么要 seal？——因为 Tower 想控制 Discover 的扩展边界——如果任何用户都能 impl Discover、未来 Tower 想加没有默认实现的新方法（比如 poll_discover_many 批量）就是 breaking change。seal 让扩展权保留给 Tower 自己、用户只能通过 blanket impl 的 TryStream 路径 "曲线救国"。

Change<(), ()> 作为类型参数——是一个技巧：tower 只有一个泛型的 Sealed<T>（tower-0.5.3/tower/src/lib.rs:222-224 的私有 mod sealed），靠不同的 T 区分不同的 sealed trait。Sealed<Change<(), ()>> 只给 Discover 用，MakeService 用 Sealed<(Target, Request)>、MakeConnection 用 Sealed<(Target,)>，各自的 blanket impl 互不冲突。

这种 sealed trait 是 Rust 库设计的惯用模式——和 Python 的 _underscore_name 私有约定、Java 的 package-private 机制类似、但用类型系统强制。std 里的 SliceIndex trait 就是用私有 Sealed 超 trait 封住的同款写法——保留内部演进空间。

7.3 ReadyCache：两个池子的协同

Balance 真正干活的地方在 ReadyCache——整整 500 行代码。但它的核心思想可以用一张图说明：

┌──────────────────────────────────────────────┐
│  ReadyCache<K, S, Req>                       │
│                                              │
│   ┌──────────────┐      ┌────────────────┐   │
│   │   pending    │ ---> │    ready       │   │
│   │              │      │                │   │
│   │ K→Pending<S> │      │ K→(S, cancel)  │   │
│   └──────────────┘      └────────────────┘   │
│        ↑                       ↓              │
│   push(key, svc)       call_ready_index()    │
│                                              │
└──────────────────────────────────────────────┘

两个池子：

  • pending: FuturesUnordered<Pending<K, S, Req>>——每一个 pending 是一个 future，它内部反复 poll 自己那个 service 的 poll_ready，直到 Ready 了就"晋升"到 ready 池子。
  • ready: IndexMap<K, (S, CancelPair)>——已经就绪的 service，按 key 索引。
  • pending_cancel_txs: IndexMap<K, CancelTx>——每个 pending service 有一个 cancel 通道，可以在它还在 pending 期间把它剔除。

这里有两个数据结构值得拎出来讲。

7.3.1 ReadyCache 的三个字段构成状态机

ReadyCache 的实际字段（tower-0.5.3/tower/src/ready_cache/cache.rs:58-74）：

pub struct ReadyCache<K, S, Req>
where
    K: Eq + Hash,
{
    pending: FuturesUnordered<Pending<K, S, Req>>,
    pending_cancel_txs: IndexMap<K, CancelTx>,
    ready: IndexMap<K, (S, CancelPair)>,
}

三个字段表达了一个状态机：

  • pending ——未就绪的 service 的 future 集合
  • pending_cancel_txs ——还在 pending 的那些 service 的 cancel sender（按 key 索引）
  • ready ——已就绪的 service（key → (service, cancel_pair)）

一个服务实例在这三者之间通常处于两种状态之一：要么它的 future 在 pending 里、同时它的 cancel sender 在 pending_cancel_txs 里，要么它已经落到 ready 里。两个 pending 侧的集合基本同进同出，例外只有一种：evict 会立刻摘走 cancel sender，而被取消的 future 要等下一次 poll_pending 才从 pending 里出来（tower-0.5.3/tower/src/ready_cache/cache.rs:210-229）。另外，同一个 key 可能短暂地同时出现在两侧：新版本服务被 push 进 pending 时，旧版本仍留在 ready 里，直到被新版本替换（tower-0.5.3/tower/src/ready_cache/cache.rs:243-246）。维护这套对应关系的是 ReadyCache::poll_pending（tower-0.5.3/tower/src/ready_cache/cache.rs:281-315），§7.3.4 那个 promote_pending_to_ready 循环就是在反复调用它。

值得留意的是这个结构体没有 PhantomData。Req 这个类型参数虽然不直接出现在任何字段的类型里，却通过 Pending<K, S, Req> 传了下去，所以编译器认得它被用到了。

真正需要 PhantomData 的是 Pending 自己（tower-0.5.3/tower/src/ready_cache/cache.rs:103-108）：

struct Pending<K, S, Req> {
    key: Option<K>,
    cancel: Option<CancelRx>,
    ready: Option<S>,
    _pd: std::marker::PhantomData<Req>,
}

它的三个字段里一个都没提到 Req，而 Rust 不允许结构体带一个哪儿都没用上的类型参数——_pd 就是用来告诉编译器"这个类型在逻辑上是我的一部分"的。这是 PhantomData 最常见、也最朴素的用途：补上一个类型参数的用处，而不是为了调整变型。

7.3.2 FuturesUnordered：不分先后的并发 poll

FuturesUnordered 是 futures-util 的一个核心工具（回想《Tokio 源码深度解析》专栏第 14 章讨论 select! 时也出现过类似的思想）。它把多个 future 装到同一个集合里，但 poll 时只推进发出过唤醒通知的那些 future（新 push 进来的 future 也会先被排进待 poll 队列）——哪个先 Ready 就先产生 output。futures-util 的文档原话是 "Futures managed by FuturesUnordered will only be polled when they generate wake-up notifications"（futures-util 0.3.34 中 FuturesUnordered 的类型文档；该 crate 不在本专栏 pin 的源码树内，版本取自按 tower 0.5.3 解析出的依赖）。

对 ReadyCache 来说，这完美贴合"推动多个 pending service 的 poll_ready"这个需求——pending.poll_next(cx) 只去 poll 那些被唤醒过的 pending service，不必把整个集合扫一遍，任何一个 Ready 了就被产出。这是很典型的"数据结构即设计"——选对集合类型，算法几乎自动成立。

7.3.3 indexmap::IndexMap：保持插入顺序的 HashMap

ready 字段是 IndexMap<K, (S, CancelPair)> 而不是 HashMap——关键差别是 IndexMap 既支持按 key 查，也支持按 index 查，而且插入顺序稳定。

为什么需要 by-index？因为 P2C 算法要"随机挑两个端点"——最高效的做法是 rng.gen_range(0..n) 两次生成两个 index，然后按 index 直接拿。HashMap 没法按 index 取，需要先 .values().nth(i) 线性扫描。IndexMap 的 get_index(i) 是 O(1)。

这是一个小到不能再小的数据结构选型决定，但它让整个 balancer 的 hot-path 复杂度从 O(n) 降到 O(1)。值得在这种细节处停一停——工业级代码的优秀往往不在算法，而在数据结构。

7.3.4 promote_pending_to_ready 里的 loop + 三分支语义

打开 tower-0.5.3/tower/src/balance/p2c/service.rs:130-155 的 promote_pending_to_ready：

fn promote_pending_to_ready(&mut self, cx: &mut Context<'_>) {
    loop {
        match self.services.poll_pending(cx) {
            Poll::Ready(Ok(())) => {
                // There are no remaining pending services.
                debug_assert_eq!(self.services.pending_len(), 0);
                break;
            }
            Poll::Pending => {
                // None of the pending services are ready.
                debug_assert!(self.services.pending_len() > 0);
                break;
            }
            Poll::Ready(Err(error)) => {
                // An individual service was lost; continue processing
                // pending services.
                debug!(%error, "dropping failed endpoint");
            }
        }
    }
    // …（loop 之后还有一句 trace! 汇报 ready/pending 两个池的长度，略去）
}

看似简单的循环、三种分支各有严格语义：

① Poll::Ready(Ok(())) —— poll_pending 告诉我们"所有 pending 都被提升了、没剩下的"。break 跳出——不是有服务就绪、而是 "pending 集合为空"。这里 pending_len == 0 的断言正是这个意思。

② Poll::Pending —— 至少一个还没就绪、但没一个失败。没有服务就绪、但集合里还有待 poll 的——break 跳出、把 pending_len() > 0 的断言写进来保护。

③ Poll::Ready(Err(error)) —— 某个服务失败了。log 一下、继续 loop 处理剩下的 pending——一个坏端点不能让其他端点受牵连。

三个分支只有 ③ 继续 loop、其他两个都 break——循环终止条件是 "没有更多失败的端点要处理"。这样 promote_pending_to_ready 一次调用能处理多个失败、不会在第一个失败时提前退出。

这就是 fault-tolerant balancer 的核心——单个端点坏了可以无感剔除、其他端点继续提供服务。和 §11.3.8 讲的 hyper InvalidInput vs UnexpectedEof 的处理分层是同一种 "区分错误严重性并对应不同处理"的 Rust 风格。

7.4 P2C：Power of Two Choices 为什么能行

Balance<D, Req> 默认用的算法叫 P2C（Power of Two Choices）：

  1. 从 ready 池随机选两个端点 A、B；
  2. 比较它们的当前 load；
  3. 把请求发给 load 较小的那个。
// tower-0.5.3/tower/src/balance/p2c/service.rs:158-183（省略注释、debug_assert_ne! 与 trace!）
fn p2c_ready_index(&mut self) -> Option<usize> {
    match self.services.ready_len() {
        0 => None,
        1 => Some(0),
        len => {
            let [aidx, bidx] = sample_floyd2(&mut self.rng, len as u64);
            let aload = self.ready_index_load(aidx as usize);
            let bload = self.ready_index_load(bidx as usize);
            let chosen = if aload <= bload { aidx } else { bidx };
            Some(chosen as usize)
        }
    }
}

简单到让人怀疑它真的有用。事实上它好得惊人。

7.4.1 sample_floyd2 的 Floyd 算法：两次 RNG 调用得两个不同索引

sample_floyd2 是 tower 内置的 RNG 采样函数（tower-0.5.3/tower/src/util/rng.rs:120-126）：

pub(crate) fn sample_floyd2<R: Rng>(rng: &mut R, length: u64) -> [u64; 2] {
    debug_assert!(2 <= length);
    let aidx = rng.next_range(0..length - 1);
    let bidx = rng.next_range(0..length);
    let aidx = if aidx == bidx { length - 1 } else { aidx };
    [aidx, bidx]
}

从 [0, length) 里随机选两个不同的索引——用 2 次 RNG 调用、常数时间、零分配。

朴素做法：loop { let a = rand(..n); let b = rand(..n); if a != b { break [a, b] } }——期望循环 n/(n-1) 轮、每轮 2 次调用，但最坏情况无界（每次都 a==b、概率低但存在）。

Floyd 算法：

  1. 第一次 aidx = next_range(0..length-1)——从 [0, length-1) 里选一个 = 从除了最后一个的 N-1 个里选
  2. 第二次 bidx = next_range(0..length)——从全范围 [0, length) 里选
  3. if aidx == bidx { length - 1 }——冲突时把 aidx 替换成 length-1（最后一个）

为什么这样保证不重复？

  • 如果 aidx != bidx：两者直接不同、完事
  • 如果 aidx == bidx：把 aidx 替换成 length-1。

细看这一支：冲突发生时、aidx = bidx = 某个 x < length-1（aidx 的取值范围本来就不含 length-1）。这时把 aidx 改成 length-1——新的 aidx = length-1 ≠ bidx = x。两者不同、√。

如果 aidx != bidx 但 aidx == length-1 和 bidx == 某个 x < length-1——这不可能、因为 aidx 的范围是 [0, length-1)、不包含 length-1。

如果 aidx != bidx 且两者都在 [0, length-1)——两者不同、√。

Floyd 算法用 "[0, n-1) × [0, n)" 的非对称范围 + 冲突映射、2 次 RNG 调用就能得到不重复的两个索引。而且它是均匀的：算一下就知道 aidx 落在任意一个下标（包括 length-1）的概率都是 1/n——这正是 Floyd 组合采样算法的性质（rng.rs 的注释注明它改自 rand 的 sample_floyd）。

对比 Fisher-Yates 洗牌需要 O(n) 时间和 O(n) 空间。balance 每次需要重新选端点时都要采样一次、O(1) 时间、零分配是必须的——Floyd 算法是精确的匹配。

7.4.2 debug_assert_ne!(aidx, bidx) 的双层保证

p2c_ready_index 里有一行看起来多余的断言（tower-0.5.3/tower/src/balance/p2c/service.rs:166）：

let [aidx, bidx] = sample_floyd2(&mut self.rng, len as u64);
debug_assert_ne!(aidx, bidx, "random indices must be distinct");

sample_floyd2 的实现保证了返回两个不同索引——按理说 aidx != bidx 是固定的。为什么还要 debug_assert？

答案有三层：

① 防御 rng 的潜在 bug——如果 rng.next_range 实现有 bug（比如未来 hasher 升级、next_range 的范围算错）、sample_floyd2 的"冲突映射"可能失效。debug_assert 会在测试里立刻捕获——而不是到了生产才发现两个索引一样、p2c 退化成单选。

② 作为不变式的文档——看 service.rs 的人不一定去读 rng.rs 的 Floyd 算法证明——assert 明确写出 "这里两者必须不同"、把不变式当契约暴露。

③ debug_assert 在 release build 会被 compile 掉——成本是零。如果是 prod 的 assert!（不加 debug_）、就要承担运行时成本——tower 选 debug_assert 是"只在测试和 dev build 验证" 的克制。

这种 "外部函数已经保证 + 内部也 assert" 的双层防御是 Rust 库代码的 idiom——不信任任何单一组件、在组件边界上都验证。一旦出错、定位极快——assert 消息直接点明哪个不变式被破坏。

7.4.3 为什么两个就够

P2C 这个名字来自 Mitzenmacher 1996 年在 UC Berkeley 的博士论文《The Power of Two Choices in Randomized Load Balancing》。论文摘要开头复述的是一个反直觉的结论（最早由 Azar、Broder、Karlin、Upfal 在 1994 年证明，见 §7.4.6）：N 个球依次放进 N 个桶，每个球随机选两个桶、放进较空的那个，最大负载以高概率只有 log log N / log 2 + O(1)；而每个球只随机选一个桶时，最大负载约为 log N / log log N。论文本身在此基础上研究了非顺序放置、球进出的动态模型（"supermarket model"）等推广。

换成人话：有 100 个端点、100 个请求要分配——

  • 纯随机：把 N=100 代入渐近式，ln 100 / ln ln 100 ≈ 3。
  • P2C：ln ln 100 / ln 2 ≈ 2.2。
  • 理论最优：每个端点正好 1 个。

渐近式忽略了常数项，只能看量级。我们用 Python 按上面的模型模拟了 2 万次（100 个球、100 个桶，P2C 取两个不同的桶），最大负载的平均值纯随机约 4.2、P2C 约 2.6——差距的方向和量级与渐近式一致。

这个"用 2 换一次 log-reduction"的效果在负载很高、请求数远大于端点数时更显著。Finagle 客户端默认就是 P2C + 最小负载（Finagle 用户指南 Clients 一节），Linkerd 的代理用 P2C 配 PeakEwma（linkerd2-proxy 的 linkerd/proxy/balance），Envoy 的 least-request 策略在各端点权重相同时也是 P2C 采样（默认随机取 2 个）——不只是因为它简单，更因为它用极小的代价换来了接近"选最小"的效果。

7.4.4 HasherRng 零依赖 RNG 的选择

Tower 内置的 HasherRng（tower-0.5.3/tower/src/util/rng.rs:100-110）：

impl<H> Rng for HasherRng<H> where H: BuildHasher, {
    fn next_u64(&mut self) -> u64 {
        let mut hasher = self.hasher.build_hasher();
        hasher.write_u64(self.counter);
        self.counter = self.counter.wrapping_add(1);
        hasher.finish()
    }
}

不引入 rand crate——用标准库的 hasher（DefaultHasher 或用户提供）生成伪随机数。原理：counter 每次递增、用 hasher 把 counter hash 成"看似随机"的 u64。

为什么不用 rand？——因为 rand 会带进一串传递依赖——对 tower 这种轻量核心库并不划算（rng.rs 的模块文档原话："replace tower's internal usage of rand with these smaller, more lightweight methods"）。用标准库 hasher：

  • 零额外依赖——hasher 是 std 必有
  • "够随机"——用于 P2C 的 RNG 不需要密码学强度、hash 伪随机足够
  • deterministic if needed——测试时用 HasherRng::with_hasher 传入一个不带随机种子的 hasher（counter 从 0 起），就能得到可复现序列

代价——生成的随机数不如 rand 的 ThreadRng 好看（分布、周期、独立性有瑕疵）。但对 P2C 这种 "从 100 个里选 2 个" 的场景、瑕疵完全不可感——balance 的统计性质不受影响。

这种"用轻量近似替代重量级专业库"的选择是 Rust 生态轻量化的典型——tower 作为 Rust 异步服务的基础层、依赖树必须窄。每一个多余的 crate 都会传染到所有用 tower 的项目——axum 直接依赖 tower，tonic 的默认 feature（router、transport）也会把 tower 拉进来。

对基础库来说，编译时间和 binary size 的成本是实实在在的，tower 这种选择体现的是基础库对依赖的克制。

7.4.5 为什么不是"选最小的"

朴素直觉会说"直接扫描所有 ready 端点选 load 最小的那个不就行了吗？" 两个原因：

  1. 成本：n 个端点就是 n 次 load() 调用、n 次比较。1000 个 endpoint 的集群上，每请求一次 O(1000) 的开销在 hot path 上可观。
  2. 羊群效应（herd effect）：如果总选最小的，所有并发调用者会同时把请求打到当前最小的那个端点——等到它 load() 更新后为时已晚。P2C 的随机性天然消解了这种 herd。

2 这个数字是收益与成本的折中——把 2 增加到 3 或 4，收益只是常数级（见下一节）。P2C 就叫 "Power of Two"——不是凑数。

7.4.6 "N=3" 收益的精确量化

§7.4.5 提到"把 2 增加到 3 或 4 收益微乎其微"——这个结论值得精确看看。Azar、Broder、Karlin、Upfal 的 "Balanced Allocations"（STOC 1994）给出的是 d-choice 的一般结论（N 个球依次投 N 个桶，以高概率成立），Mitzenmacher 的博士论文摘要也复述了它：

d = 选几个 最大 load（渐近） 相对 d=2 的改善
d = 1（纯随机） log N / log log N 基准
d = 2 log log N / log 2 + O(1) 阶的改善（单 log → 双 log）
d ≥ 3 log log N / log d + O(1) 只改分母上的常数
d = n 最小 load（完美均匀） 退化成 "选最小"

关键的改善发生在 d=1 → d=2——从 log N / log log N 到 log log N 是双 log 的区别。

d=3 以后只是 "系数上的改善"——分母从 log 2 变成 log 3、log 4——不是阶的改善；而代价是多一次 RNG 调用 + 多一次 load() 评估——大部分场景不值。

Power of Two 是论文标题也是实践经典——2 这个数字是 d-choice 中收益 vs 成本最划算的点。Tower 固定用 2，Finagle 默认用 2，Envoy 的 choice_count 默认也是 2——背后是上表的数学：阶的改善在 d=2 就已拿到。

这种"能被量化的 trade-off"在系统设计领域并不多见——大部分选择靠"经验"、"感觉"、"主流这样"。定理本身并没有证明 d=2 "最优"，但它说清了 d 每加 1 换来的只是常数项，让选 2 成为"有依据的取舍"而不是"跟风"。

7.4.7 Load trait：让端点自己报告负载

P2C 需要比较 load——但 load 是什么？Tower 把这件事抽象成 Load trait：

// tower-0.5.3/tower/src/load/mod.rs:80（各成员的 doc 注释已省略）
pub trait Load {
    type Metric: PartialOrd;
    fn load(&self) -> Self::Metric;
}

metric 只需要可比较，不需要是数字——可以是 Duration（用 EWMA 延迟）、usize（in-flight 计数）、自定义 struct（综合指标）。Tower 内置了三种（load/constant.rs 的 Constant 是给测试和"所有端点等价"场景用的常量 metric），其中两种最常见：

  • PendingRequests：当前 in-flight 请求数。最便宜的 metric，几乎免费。
  • PeakEwma：延迟的 EWMA 加权移动平均。对抖动敏感，但能检测"快慢端点"的差异。

在真实系统里，PeakEwma 用得多——因为 "in-flight count" 对延迟不敏感，可能所有端点 in-flight 都是 10，但其中一个实际 p99 是 500ms。

7.4.8 PeakEwma 的时间衰减计算——EWMA 背后的数学

PeakEwma（Peak-Exponentially-Weighted-Moving-Average）是 tower 最精巧的 Load metric。源码在 tower-0.5.3/tower/src/load/peak_ewma.rs：

// 概念性核心（字段名与 tower 0.5.3 的 peak_ewma.rs 一致）
pub struct PeakEwma<S, C = CompleteOnResponse> {
    service: S,
    decay_ns: f64,
    rtt_estimate: Arc<Mutex<RttEstimate>>,
    completion: C,
}

struct RttEstimate {
    update_at: Instant,   // 上次更新的时刻
    rtt_ns: f64,          // 当前 EWMA 值，单位纳秒
}

impl<S, C> Load for PeakEwma<S, C> {
    type Metric = Cost;
    fn load(&self) -> Self::Metric {
        // in-flight 数不另存计数器：就看 rtt_estimate 这个 Arc 被多克隆了几份
        let pending = Arc::strong_count(&self.rtt_estimate) as u32 - 1;
        let estimate = self.update_estimate();   // 先按 decay 衰减到当前时刻
        Cost(estimate * f64::from(pending + 1))
    }
}

EWMA 的核心公式：

new_estimate = alpha * new_sample + (1 - alpha) * old_estimate
其中 alpha = 1 - exp(-delta_t / decay)

关键是 alpha 随时间动态变化——delta_t 是距上一次更新（update_at）的时长，越长、alpha 越接近 1（新样本权重越大）。这里的"更新"不只发生在请求完成时：load() 每次调用 update_estimate() → RttEstimate::decay，也会以一个 0 时长的样本做一次更新，让估计值随时间向 0 衰减（tower-0.5.3/tower/src/load/peak_ewma.rs:227-232）。这比固定 alpha 的普通 EWMA 更准——因为采样间隔不均匀时、固定 alpha 会低估久违样本的重要性。

Peak 的意思——遇到比当前 estimate 大的样本、直接替换（不做 EWMA 混合）、只在样本更小时才做 EWMA decay。效果：延迟上升响应快（峰值被立即看到）、延迟下降响应慢（缓慢 decay 避免被偶然的快请求误导）。

这种 "快速上升、慢速下降" 的非对称响应在网络延迟测量里是一种偏保守的取舍——宁可错报 "延迟高"、不要漏报（漏报会让 load balancer 继续往慢端点打请求）。

加上 (1.0 + pending_count) 作为倍数——正在发送中的请求数也计入 load——这让 balancer 自然避开 "刚刚开始飙升但 EWMA 还没更新" 的端点。

Tower 的 PeakEwma 跟 Finagle 的同名实现是一样的公式（峰值替换、exp(-td/Tau) 衰减、cost * (pending + 1)；Finagle 另有一个 cost 为 0 时的 Penalty 分支，tower 用 default_rtt 起步）——peak_ewma.rs 顶部的文档注明它派生自 Finagle 的 PeakEwma.scala。去掉测试大约 300 行代码、承载的是一套经过生产检验的算法。

7.4.9 Load::Metric 为什么是 PartialOrd 而不是 Ord

Load trait 的 associated type 约束是 PartialOrd：

pub trait Load {
    type Metric: PartialOrd;
    fn load(&self) -> Self::Metric;
}

PartialOrd vs Ord 的区别：

  • Ord 要求 total order——任意两个值都能比较（a < b 或 a > b 或 a == b）
  • PartialOrd 允许 partial order——某些对之间无法比较（返回 None）

为什么 Load metric 要选 PartialOrd？因为某些 metric 不是 total order：

① 浮点数——PeakEwma 的 metric 是 Cost(f64)，而 f64 根本不实现 Ord（NAN 和任何数比较都返回 None）。约束写成 Ord 的话 PeakEwma 就没法当 Load 用。

② 多维 metric——某些 Load 实现可能想同时看 latency 和 pending_count——不同维度的主导可能冲突（a 的 latency 更小但 pending 更多）。注意直接用 tuple 表达不了这一点：tuple 的 PartialOrd 是字典序，照样能比出先后；要为这种 metric 自定义一个 struct、手写 partial_cmp 在两维冲突时返回 None，PartialOrd 才能表达 "这两个 load 没有严格顺序"。

③ future-proofing——留给用户自定义 metric 的自由——不强制他们提供 total order。

选 PartialOrd 的代价是 P2C 里的比较变复杂——实际代码里 if aload <= bload { aidx } else { bidx } 依赖 PartialOrd::le——遇到 None 时会走 else 分支（因为 <= 对 None 返回 false）。等价于 "不能比较时默认选 bidx"——统计上不影响 P2C 的分布性质、只是个小瑕疵。

这种 "约束选弱、代价吸收到使用处" 的设计给用户自定义 Metric 最大自由度——tower 的基础组件不强制自己不需要的性质、允许上层创造性的实现。

7.4.10 PendingRequests vs PeakEwma 的选型决策树

两种内置 Load metric 的选择指南：

选 PendingRequests 的场景：

  • 端点延迟接近均匀（比如同一个 Pod 模板的多副本）——in-flight count 就是最好的 load 信号
  • 极致性能敏感——PendingRequests 只靠一个 Arc<()> 的引用计数（原子自增/自减）数 in-flight 请求、开销微不足道
  • 短连接场景——请求很快完成、in-flight 数变化快、能实时反映 load

选 PeakEwma 的场景：

  • 端点异构（多数据中心、不同代硬件混合）——延迟天然有差别、光 in-flight 误导
  • 长尾敏感——care about P99 / P999、需要检测 "in-flight 少但实际慢" 的端点
  • JIT 语言后端（Java/Node.js/Ruby）——刚启动时 warm up 阶段延迟高、PeakEwma 能自动避开

选择错了的后果：

  • 用 PendingRequests 但端点异构——一个慢 endpoint 会接和快 endpoint 一样多的请求、慢的 p99 飞起
  • 用 PeakEwma 但端点均匀——无用的 EWMA 计算开销 + mutex contention——白白降低单机 QPS 上限

Tower 不替你选默认——两种都暴露、让用户根据自己场景决定——既不强制你用某个算法、也不隐藏实现细节让你无从决策。

7.5 Balance::poll_ready：把三者串起来

// tower-0.5.3/tower/src/balance/p2c/service.rs:208-250（简化：省略原注释、trace!/debug! 日志与一句 debug_assert_eq!）
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
    let _ = self.update_pending_from_discover(cx)?;
    self.promote_pending_to_ready(cx);

    loop {
        if let Some(index) = self.ready_index.take() {
            match self.services.check_ready_index(cx, index) {
                Ok(true) => {
                    self.ready_index = Some(index);
                    return Poll::Ready(Ok(()));
                }
                Ok(false) => { /* no longer ready, try another */ }
                Err(Failed(_, error)) => { /* endpoint failed, try another */ }
            }
        }

        self.ready_index = self.p2c_ready_index();
        if self.ready_index.is_none() {
            return Poll::Pending;
        }
    }
}

四件事：

  1. 更新 discover：把最近的 Insert/Remove 吸收到 pending/ready 池子里。
  2. 把 pending 提升到 ready：对所有 pending service 做一次 poll（通过 FuturesUnordered），就绪的进入 ready 池。
  3. check/select：如果上一次已经选了一个 ready_index，确认它还就绪；否则用 P2C 选新的。
  4. 结论：有可用的——Ready；一个都没——Pending（waker 已经被 FuturesUnordered 和 discover 注册好了，服务就绪时会被唤醒）。

注意第 3 步是一个 loop——check 失败时不直接 Pending，而是继续 P2C 找下一个。这保证即使某些 ready endpoints 刚刚变成不可用，只要 ready 池里还有其他可用的，poll_ready 就能报告 Ready。这是 fault tolerance 的细节——一次 poll_ready 调用里可以自动跳过多个坏 endpoint。

7.5.1 check_ready_index 的再次验证——为什么不相信上次的 ready

§7.5 poll_ready 里有一段 subtle 的逻辑：

if let Some(index) = self.ready_index.take() {
    match self.services.check_ready_index(cx, index) {
        Ok(true) => {
            self.ready_index = Some(index);
            return Poll::Ready(Ok(()));
        }
        Ok(false) => { /* no longer ready */ }
        // ...
    }
}

为什么上次选好的 ready_index 还要 check_ready_index 一次？——直接返回 Ready 不更省事？

原因是 ready 的 service 可能在两次 poll_ready 之间变回 pending。cache.rs 顶部的文档给的场景：

  • 某个端点上次 poll_ready 时 Ready、被放进 ready 池
  • 调用者没立即 call、而是去做了别的事
  • 期间 runtime 处理了别的事件——比如 keepalive 探测到这条连接已经断了——这个端点变成 not-Ready
  • 现在才来 call——如果不重新检查、就会 call 一个 not-Ready 的 service——违反 Tower 协议

check_ready_index 就是重新 poll 那个 service 的 poll_ready——确认 "从上次到现在它一直没变" 再让出 Ready。如果变了、Ok(false) 让 loop 重新 P2C 选一个。

这就是 Tower 协议的严格 recheck 规则——call 之前必须确认 poll_ready 立即就是 Ready。balance 把这个规则落到每个循环、给上层提供了一个铁板钉钉的承诺：从 Ready 返回到 call、中间不会 race。

这个 recheck 是有成本的——每次 poll_ready 多一次 service 内部的 poll_ready 调用。但比起 协议违反导致的难以调试 bug、这个成本完全值得。

7.5.2 call 的简洁

// tower-0.5.3/tower/src/balance/p2c/service.rs:252-257
fn call(&mut self, request: Req) -> Self::Future {
    let index = self.ready_index.take().expect("called before ready");
    self.services
        .call_ready_index(index, request)
        .map_err(Into::into)
}

两行：拿出之前 poll_ready 选好的 index，对那个 service 发起 call。

最神奇的地方在self.services.call_ready_index(index, request)——它内部把那个 service 从 ready 池子 take 出来、调用 .call(request)、把 service 送回 pending 池子等下次 poll_ready。这是 Tower ReadyCache 的核心不变式：每个 service 每次 call 后都要重新 ready——因为它的 poll_ready 可能会因为这次 call 消费了某个 permit 而变回 pending。

这个"call → 重进 pending → FuturesUnordered 自动 poll_ready → 重进 ready"的循环，由 ReadyCache 透明地驱动。你写 Balance 代码时看到的是"选一个、用一个"，背后是整个一套高效的后端状态管理。

7.5.3 update_pending_from_discover 的内部 loop——一次 poll 吸收所有变更

看 update_pending_from_discover（tower-0.5.3/tower/src/balance/p2c/service.rs:105-128）的核心结构：

fn update_pending_from_discover(&mut self, cx: &mut Context<'_>) -> ... {
    loop {
        match ready!(Pin::new(&mut self.discover).poll_discover(cx))
            .transpose()
            .map_err(|e| error::Discover(e.into()))?
        {
            None => return Poll::Ready(None),
            Some(Change::Remove(key)) => self.services.evict(&key),
            Some(Change::Insert(key, svc)) => self.services.push(key, svc),
        }
    }
}

无限 loop——只要 discover 还有 Change 产生、就一直吸收。直到 poll_discover 返回 Pending、ready! 宏把 Pending 往外传（实际是 return Poll::Pending）。

为什么要 loop 吸收所有？——不要让"discover 堆积"发生。如果 discover 产生了 10 个 Change 但 balance 一次只吸收一个、那 9 个留在 discover 的内部队列里、下次 poll_ready 才被处理——导致 balance 对后端集合的认知滞后。

极端情况：k8s endpoints watch 瞬间产生几十个事件（滚动重启），balance 必须一次全部吸收、不能分散到多个 poll_ready 周期。否则新端点迟迟加不进 ready 池、老端点继续被选中、影响用户请求延迟。

loop { ready!(poll_discover) } 把拉取变更和处理变更合成一个紧密循环——discover 生产多快、balance 吸收多快。注意这里的取舍恰恰是不对 discover 施加背压：变更流有多少就吸收多少，换来的是 balance 对后端集合的认知不滞后。

7.6 取消：CancelPair 的设计

ReadyCache 里有一对非常小但设计精巧的类型：CancelTx 和 CancelRx（tower-0.5.3/tower/src/ready_cache/cache.rs:79-91）。

#[derive(Debug)]
struct Cancel {
    waker: AtomicWaker,
    canceled: AtomicBool,
}

#[derive(Debug)]
struct CancelRx(Arc<Cancel>);

#[derive(Debug)]
struct CancelTx(Arc<Cancel>);

type CancelPair = (CancelTx, CancelRx);

一对 Tx/Rx 共享一个 Arc<Cancel>。AtomicWaker 存"谁在等 cancel"，AtomicBool 存"已经 cancel 了"。

为什么不直接用 tokio::sync::oneshot？cancelable() 上面的注释说得很直白：oneshot 的 Receiver 不保证在 Sender 触发后立刻观察到结果，而用 AtomicBool 可以在取消被触发的瞬间就看到状态。顺带地，CancelPair 也不用携带值槽，只有两个原子变量，在高频 insert/evict 场景下开销极小。

它的用法：每个 pending service 带一个 CancelRx，它的 Pending future 在每次 poll 里检查 canceled——一旦上层 evict(&key) 调用了对应的 CancelTx，canceled 被置 true，waker 被触发，pending future 下次被 poll 时发现已 cancel，产出一个 PendingError::Canceled(key)，FuturesUnordered 把它剔除出集合。

ready 里的 service 也绑着一个 CancelPair，但 evict 时并不用它：ready 的 service 不是 future，evict 直接用 swap_remove_full 把它从 IndexMap 里移除（tower-0.5.3/tower/src/ready_cache/cache.rs:225-228）。它留着这对 CancelPair，只是为了 call 之后回到 pending 时不必重新分配（字段注释 tower-0.5.3/tower/src/ready_cache/cache.rs:70-72）。

这段代码的美感不在算法，而在用 Rust 的所有权和原子原语把"动态集合 + 取消"这件看似复杂的事情压缩到极简。Tower 的取消机制（Cancel、CancelTx/CancelRx、cancelable() 与 CancelTx::cancel）总共不到 50 行。

7.6.1 AtomicWaker 和 AtomicBool 为什么不合并

§7.6 Cancel 结构有两个 atomic 字段：

struct Cancel {
    waker: AtomicWaker,
    canceled: AtomicBool,
}

为什么不用单一 AtomicUsize 同时编码 "是否取消"和"waker 指针"？（某些 C++ 无锁结构会这么做）

原因：

① AtomicWaker 本身就是一个非平凡结构——不是一个 raw pointer、而是含有内部状态的 wrapper（处理 "waker 被更新时的并发可见性"）。简单的 AtomicUsize 不能承载 Waker 的全部语义（waker 的 clone、wake、drop 涉及 vtable 调用）。

② 这里本来就谈不上竞争——rx 只注册 waker、tx 只 set canceled，各改各的字段；而且源码注释写明 cancel 的收发两端都由 ReadyCache 持有、在同一个 task 上操作（tower-0.5.3/tower/src/ready_cache/cache.rs:462-465）。合并成一个原子变量换不来什么，分开写反而语义清楚。

③ 简化代码——用 AtomicWaker 的现成 API + 一个 bool、比手写一个合并编码的 atomic dance 安全得多。并发原语的坑多如牛毛——能复用标准库的就复用。

这就是 Rust 并发编程的惯用风格——组合现成的 atomic primitives（AtomicWaker、AtomicUsize、AtomicBool、Arc）、而不是自己手写 SeqCst / Acquire / Release 的复杂 atomic 魔法。专业人士写的原语更可靠、代价是每个原子变量各占几字节内存——在 Cancel 这种 per-service 结构里完全可接受。

这种"优先组合 primitives 而不是手写 lock-free"是应用层和中间件代码的常见取舍——手写 atomic 的活留给 tokio、parking_lot、crossbeam 这类专门做并发原语的库，上层代码平常就组合 Arc<Mutex<_>> 和现成的并发工具。

7.7 组合出一个完整的客户端

把这一章讲的抽象和前面几章串起来，典型的一个"客户端组合"长这样（这里用玩具端点代替真实的 gRPC 客户端，整段可以直接 cargo run；依赖为 tower = { version = "=0.5.3", features = ["balance", "discover", "load", "retry", "timeout", "limit", "buffer", "util"] } 与 tokio = { version = "=1.46.1", features = ["full"] }）：

use std::time::Duration;
use tower::{
    balance::p2c::Balance,
    discover::ServiceList,
    load::{CompleteOnResponse, PendingRequests},
    retry::Policy,
    service_fn, BoxError, Service, ServiceBuilder, ServiceExt,
};

// 一个玩具端点：回显请求，并标出是哪个端点处理的
fn endpoint(id: usize) -> impl Service<
    String,
    Response = String,
    Error = BoxError,
    Future = std::future::Ready<Result<String, BoxError>>,
> + Send {
    service_fn(move |req: String| std::future::ready(Ok(format!("endpoint {id}: {req}"))))
}

// 最多重试 n 次的重试策略
#[derive(Clone)]
struct Attempts(usize);

impl<Req: Clone, Res, E> Policy<Req, Res, E> for Attempts {
    type Future = std::future::Ready<()>;
    fn retry(&mut self, _req: &mut Req, result: &mut Result<Res, E>) -> Option<Self::Future> {
        if result.is_err() && self.0 > 0 {
            self.0 -= 1;
            Some(std::future::ready(()))
        } else {
            None
        }
    }
    fn clone_request(&mut self, req: &Req) -> Option<Req> {
        Some(req.clone())
    }
}

#[tokio::main]
async fn main() -> Result<(), BoxError> {
    // 1. 有一组 endpoints
    let endpoints = vec![endpoint(0), endpoint(1), endpoint(2)];

    // 2. 把每个 endpoint 包成"带 load metric"的 service
    let endpoints = endpoints
        .into_iter()
        .map(|s| PendingRequests::new(s, CompleteOnResponse::default()))
        .collect::<Vec<_>>();

    // 3. 把它们打包成一个静态 Discover
    let discover = ServiceList::new::<String>(endpoints);

    // 4. 用 Balance 做 P2C 负载均衡
    let balanced = Balance::new(discover);

    // 5. 在外面套上常规中间件；Retry 要求内层 Service: Clone，
    //    Balance 本身不是 Clone，所以先用 Buffer 包一层
    let mut svc = ServiceBuilder::new()
        .concurrency_limit(500)
        .timeout(Duration::from_secs(10))
        .retry(Attempts(3))
        .buffer(64)
        .service(balanced);

    for i in 0..3 {
        let resp = svc.ready().await?.call(format!("req-{i}")).await?;
        println!("{resp}");
    }
    Ok(())
}

运行输出形如下面这样（选中哪个端点由随机数决定，每次运行可能不同）：

endpoint 1: req-0
endpoint 2: req-1
endpoint 1: req-2

第 5 步里的 .buffer(64) 不是可有可无的：Retry 的 Service 实现要求内层 S: Service<Request> + Clone（tower-0.5.3/tower/src/retry/mod.rs:75），因为每次重试都要克隆一份内层服务；而 Balance 没有实现 Clone。去掉这一行，svc.ready() 就会因 Retry<Attempts, Balance<..>> 不满足 Service 而编译失败。Buffer 把 Balance 搬进一个后台 worker task，对外给出一个可克隆的句柄（第 6 章）。

请求流：

caller → concurrency_limit → timeout → retry → buffer →（worker task）balance → p2c_pick(A,B) → endpoint.call(req)

任何一个端点临时变慢、不可用，P2C 会自然绕开它；新的 endpoint 通过 discovery 进来后，在下一次 poll_ready 中被吸收、就绪后即可接请求；请求失败时 Retry 会重新发起，而 Balance 会重新做一次 P2C 选择，未必落回原来那个端点。除了 Buffer 这一跳要经过 channel 和 worker task，链条上的其余中间件是可编译期单态化的类型链，彼此之间没有 vtable 跳转（Balance 内部的 Box<dyn Rng> 是一处例外）。

这就是 Rust 后端生态最优雅的一段组合拳。

7.7.1 为什么类型推导能让 ServiceBuilder 链零 annotation

§7.7 的示例代码里，let 绑定一个类型标注都没写——都是 let x = ...；仅有的提示是 collect::<Vec<_>>() 指明集合、ServiceList::new::<String> 给出无法推断的请求类型。这在 ServiceBuilder 链上尤其显眼：

let mut svc = ServiceBuilder::new()
    .concurrency_limit(500)
    .timeout(Duration::from_secs(10))
    .retry(Attempts(3))
    .buffer(64)
    .service(balanced);

svc 的实际类型是（摘自编译器报错里打印的完整类型，TrackCompletionFuture 的参数与 BoxError 的展开略去）：

ConcurrencyLimit<
    Timeout<
        Retry<
            Attempts,
            Buffer<
                String,
                MapErr<TrackCompletionFuture<..>, fn(BoxError) -> BoxError>,
            >,
        >,
    >,
>

注意 Buffer<Req, F> 的第二个参数只是内层服务的 future 类型，Balance<ServiceList<Vec<PendingRequests<..>>>, String> 本身被 Buffer "吞"掉了，不出现在外层类型里。即便如此也有 6 层以上的嵌套泛型——如果要手写、几乎不可能。但 let svc = ... 让 Rust 的 Hindley-Milner 类型推导自动算出——用户不需要知道具体类型是什么。

这是 "zero-ceremony generics" 的精髓——局部变量的类型推导让用户把注意力放在行为上、而不是类型上。

但编译期类型依然是确定的——编译器 monomorphize 时每一层都被具体化。对比 Java/Scala 的 type erasure 生成统一 bytecode、Rust 的 monomorphization 让每一个具体类型组合都有自己的代码段——运行时完全没有 type dispatch 成本。

这是 Rust 泛型系统独特的甜蜜点——写起来像动态语言、跑起来像 C。tower 的 ServiceBuilder 是这种哲学的展示柜——用户看到的是流畅的链式 API、编译器生成的是精确单态化的机器码。

7.8 和其他系统对照

Envoy 的 least-request 策略也是 P2C 采样，核心思想和 Tower 一致——因为论文就是那个论文。差别：Envoy 的负载 metric 是内置的活跃请求数，Tower 允许你自由定义 Load::Metric。

Finagle（Scala）是 Tower 这套 P2C + PeakEwma 的直接源头——peak_ewma.rs 的文档直接链接到 Finagle 的 PeakEwma.scala。Finagle 还有 aperture（只在"一部分"端点里做 P2C），Tower 没有实现。

Go grpc-go 默认策略是 pick_first，常用的是 round_robin；它也带了一个 balancer/leastrequest（按未完成 RPC 数、默认随机取 2 个比较，本质就是 P2C），但要在 service config 里显式启用。Tower 则把 P2C 做成 Balance 的唯一算法、Load metric 可自由替换，让应用代码直接获得工业级负载均衡——不需要外挂 service mesh。

7.9 本章与其他章节的呼应

与《Tokio 源码深度解析》专栏第 12 章（Semaphore）的呼应——tower 的 ReadyCache 维护 "pending vs ready 双池" 和 Tokio Semaphore 维护 "permit 持有者 vs 等待者双队列" 是同构的——都用 "两个池子 + 在中间状态转移" 表达 "可用资源" 的模型。

与第 8 章（Filter/Steer）的呼应——Steer 的 Picker 按请求内容选 service、Balance 的 P2C 按运行时 load 选 service——两者本质都是 "在多个 service 间动态 dispatch"、只是决策依据不同（业务 vs 运行时状态）。

与第 16 章（h2 flow control）的呼应——Balance 的 WINDOW 语义其实也在这里——每个后端的"容量"本质上是它的 poll_ready 可接受速率。flow control 在 HTTP/2 层是每连接 + 每流、在 tower balance 层是每后端 —— 两者是跨协议层的同一种 "容量调度" 思想。

与《Vite 设计与实现》专栏第 15 章 SSR 的呼应——vite 的 ensureBuiltins 启动一次性协商 + tower 的 ServiceList 静态 Discover 都在表达 "配置性的集合"——运行时不会变、Vec 或 list 足够、不需要 watch 机制。

与《React 19 内核探秘》专栏第 6 章 Commit 的呼应——tower 的 check_ready_index 重新验证 ready 状态、React commit 的 finishedWork !== root.current 重新确认双缓冲不变式——都是在核心热路径上做廉价运行时验证、防止上游传来的一致性问题影响当前层。

7.10 关于 Unpin 的一个迂回

细心的读者可能已经注意到：Balance 的 Service impl 里有一条 D: Unpin 约束（源码第 194 行）。注释明确解释：

[Balance] requires that the [Discover] you use is [Unpin] in order to implement [Service]. This is because it needs to be accessed from [Service::poll_ready], which takes &mut self.

原因：Service::poll_ready(&mut self) 拿到的是普通的 &mut self，不是 Pin<&mut Self>。但 Discover::poll_discover 需要 Pin<&mut D>。要把 &mut self.discover 转成 Pin<&mut D>，要么 D 是 Unpin（可以用 Pin::new(&mut self.discover) 零成本转换），要么 Balance 自己持有 Pin<Box<D>>（堆分配）。

Tower 选择前者——要求用户自己把 Discover 包成 Pin<Box<D>> 或者保证它 Unpin。这是一个 "把复杂性推到用户侧" 的决定，换来 Balance 内部实现的简洁。issue #319 有完整讨论。

这个现象在《Rust 编译器与运行时揭秘》专栏第 10 章（Pin / Waker / Future）里已经埋下伏笔——Pin 的传染性在 trait 设计里经常成为需要权衡的焦点。Tower 1.0 如果未来切到 async trait，这类约束可能会彻底简化。

7.11 P2C 和 ReadyCache 的工程密度——十条原则

把本章所有源码观察提炼为 10 条工程原则：

① Sealed trait 保留扩展权（§7.2.2）——Sealed<Change<(), ()>> 让 Discover 只能被 Tower 实现。

② Blanket impl 嫁接生态（§7.2）——任何 TryStream 自动 impl Discover。

③ FuturesUnordered 管理动态 pending 集合（§7.3.2）——N 个 future 一起 poll、先就绪先晋升。

④ IndexMap 兼顾 key 和 index（§7.3.3）——按 key 查、按 index 采样、两个场景都 O(1)。

⑤ Floyd 算法 2 次 RNG 得不同索引（§7.4.1）——常数时间零分配、给 P2C 提供完美采样。

⑥ HasherRng 避免 rand 依赖（§7.4.4）——基础库依赖树克制。

⑦ P2C 选 2 个就拿到阶的改善（§7.4.3、§7.4.5）——2 比 1 好一个阶（Azar 等人证明、Mitzenmacher 博士论文推广的 log-log vs log 差异），又比扫描全部 N 个便宜、还能避开羊群效应。

⑧ PeakEwma 的非对称响应（§7.4.8）——延迟上升立即更新、下降慢慢 decay。

⑨ check_ready_index 再验证 ready 状态（§7.5.1）——Tower 协议的严格性落地。

⑩ update_pending_from_discover 内 loop 排空（§7.5.3）——生产者和消费者紧耦合、不堆积滞后。

这十条加起来就是 Tower Balance + ReadyCache 的设计结晶——3 个小文件（service.rs 258 / cache.rs 503 / rng.rs 187）合计不到 1000 行、承载了客户端负载均衡的核心逻辑（负载度量在 load/、服务发现在 discover/）。

这不到 1000 行在 tower 里由 balance feature 按需开启，可以被任何 Rust 异步服务嵌入，不用时不进依赖树。

这种"小体量承载大能力"的密度在 Rust 生态里很典型——类型系统 + monomorphization 让抽象的零成本成为可能、allocation control 让每一 byte 内存可见、trait sealed 让扩展可控。把这十条内化、你就能自己设计类似质量的其他基础组件。

7.12 Pending<K, S, Req> future：三分支 poll

pending 池里每个元素是 Pending<K, S, Req>——一个自定义 Future，每次 poll 会先看有没有被取消、再调用内部 service 的 poll_ready（tower-0.5.3/tower/src/ready_cache/cache.rs:441-455）。它的字段在 §7.3.1 已经见过（tower-0.5.3/tower/src/ready_cache/cache.rs:99-109）：key: Option<K>、cancel: Option<CancelRx>、ready: Option<S>，外加一个 PhantomData<Req>。三个 Option 的用意是一致的：future 完成时要把 key、service、cancel receiver 三样东西按值交还给 ReadyCache，所以用 take() 搬走。

poll 的实现（tower-0.5.3/tower/src/ready_cache/cache.rs:441-482，仅略去一句 assert）：

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let this = self.project();
        // Before checking whether the service is ready, check to see whether
        // readiness has been canceled.
        let CancelRx(cancel) = this.cancel.as_mut().expect("polled after complete");
        if cancel.canceled.load(Ordering::SeqCst) {
            let key = this.key.take().expect("polled after complete");
            return Err(PendingError::Canceled(key)).into();
        }

        match this
            .ready
            .as_mut()
            .expect("polled after ready")
            .poll_ready(cx)
        {
            Poll::Pending => {
                // Before returning Pending, register interest in cancelation so
                // that this future is polled again if the state changes.
                let CancelRx(cancel) = this.cancel.as_mut().expect("polled after complete");
                cancel.waker.register(cx.waker());
                // …（这里还有一句 assert!，断言轮询 pending service 期间取消状态不会变，略去）
                Poll::Pending
            }
            Poll::Ready(Ok(())) => {
                let key = this.key.take().expect("polled after complete");
                let cancel = this.cancel.take().expect("polled after complete");
                Ok((key, this.ready.take().expect("polled after ready"), cancel)).into()
            }
            Poll::Ready(Err(e)) => {
                let key = this.key.take().expect("polled after compete");
                Err(PendingError::Inner(key, e)).into()
            }
        }
    }

三个出口对应 poll_pending（§7.3.4 引用的那个函数）的三个分支：Ok((key, svc, cancel_rx)) 晋升进 ready 池（cancel 对儿一起存进去，下次 call 后回 pending 时不用重新分配）、Canceled(key) 静默丢弃（它的 cancel sender 已经在 evict 时被摘走）、Inner(key, e) 变成 error::Failed(key, e) 往上报——也就是 promote_pending_to_ready 里 "log 一下继续 loop" 的那条分支。

7.13 为什么 Pending 一个 #[pin] 都不需要

Pending 是用 pin_project_lite::pin_project! 声明的，但没有任何字段标 #[pin]。原因很朴素：

  • Service::poll_ready 的签名是 &mut self，不是 Pin<&mut Self>——所以对 ready: Option<S> 只需要普通的 &mut 访问；
  • Option<K>、Option<CancelRx>（内部是 Arc<Cancel>）都是可以随便搬动的普通值；
  • 唯一"像 future"的等待——等待取消——走的是 AtomicWaker::register，也不需要 pin。

于是 self.project() 得到的每个字段都是普通 &mut，take()、as_mut() 都能直接用。ReadyCache 本身也顺势声明了 impl Unpin（tower-0.5.3/tower/src/ready_cache/cache.rs:76-77，注释是 "This is safe because we do not use Pin::new_unchecked"）。

这和上一章 Buffer 的 ResponseFuture、第 5 章 Timeout 的 ResponseFuture 形成对照：那两个 future 里装的是别人的 future（#[pin] fut、#[pin] sleep），必须 pin 投影才能 poll；而 Pending 里装的是一个 Service，Service 的就绪协议是 &mut self 的 poll_ready——Tower 把 poll_ready 设计成 &mut self 而不是 Pin<&mut Self>，在这里直接换来了一个不需要 pin 的 future。这是 Rust 手写 future 的一条实用判断：只有当字段本身是 !Unpin 的 future/stream 时才标 #[pin]；能用 &mut 驱动的东西就别给自己加 Pin 的负担。

7.14 落到你键盘上

读完这一章，你能做的事：

  • 阅读 tower-0.5.3/tower/src/load/ 下的 PendingRequests 和 PeakEwma 实现。两者都不长，但 PeakEwma 有一个时间衰减的精巧计算——读它你会学到如何在 Arc<Mutex<RttEstimate>> 里按时间衰减地维护"移动平均"，以及如何用 Arc 的引用计数顺带数出 in-flight 请求。
  • 实验一个动态 Discover。用 tokio mpsc channel 做一个手写 Discover，用 tokio::spawn 模拟"每 5 秒随机增删一个端点"，用 tracing::debug! 看 Balance 如何响应。
  • 对照阅读 linkerd2-proxy（GitHub 开源）的负载均衡代码（linkerd/proxy/balance）。它目前的主干直接复用 tower 的 PeakEwma 做负载度量，P2C 选择则放在自己的 linkerd-pool-p2c 里，外面再套上请求队列——是学习"工业级应用如何复用和扩展 Tower 抽象"的好参考。

7.15 应用层的三个典型问题

P2C、Floyd、pin projection 这几条知识不是只对库作者有用——应用层也会在日常排障里碰上。

① axum + tower + tonic 的 p99 毛刺。tonic 的 balancer 选中的某个端点响应持续慢、整体 QPS 没变、p99 却飙。懂 P2C 的话你会先去看 load metric：tonic 自带的 Channel::balance_* 用的是 Balance::new(discover)，而它的 Connection 的 Load 实现恒返回 0（tonic/src/transport/channel/mod.rs:191-204、tonic/src/transport/channel/service/connection.rs:118-124），P2C 比不出高低，实际上退化成随机选。想让 balancer 自动降低慢端点权重，就得自己用 tower::balance::p2c::Balance 组装，把每个端点包成 PeakEwma——pending 数能看到"排了多少请求"但看不到"每个请求多慢"，PeakEwma 两者都算进去。

② reqwest 批量抓取遇到单 IP 限流。某个 upstream 返回 429、reqwest 的 pool 继续往那个 endpoint 打——后续都失败。一种做法是给每个端点包一层 circuit breaker middleware，失败累计到阈值后让它的 poll_ready 返回错误——ReadyCache 对 poll_ready 报错的端点会直接移出（pending 侧见 tower-0.5.3/tower/src/ready_cache/cache.rs:305-311，ready 侧见 :364-371）；另一种是由你自己的 Discover 在熔断时产出 Remove。

③ service mesh 里某些 endpoint load 偏高但延迟不高。这个现象八成是 Load metric 选错：PendingRequests 看不到 GPU-bound 后端的真实负载（请求先排队、之后才真正 busy GPU）。换成 PeakEwma 或者自定义 Load::Metric（从后端 x-load header 读取）能解决。

7.16 一个典型场景：Balance 和 HPA 的互动

在 k8s 集群里、Balance + HorizontalPodAutoscaler（HPA）有一种耐人寻味的互动（下面是一个示意场景，数字只为说明量级）——直接关系到本章设计原则在生产的意义。

场景：某服务 10 个 Pod、QPS 突然从 1000 翻到 5000。HPA 扩到 20 个 Pod——新 Pod 启动需要 30-60 秒。这 60 秒里发生了什么？

有 P2C 的情况：

  • Balance 通过 Discover（比如 linkerd 或 kuma）立即看到 10 个新 endpoint Insert
  • 新 endpoint 经过 push → pending → poll_ready → ready 的流程——第一次 poll_ready 返回 Ready 就进 ready 池
  • P2C 开始把新请求 spread 到 20 个端点——旧 10 个的 load 压力立刻减半
  • 60 秒内整个集群从 5000 QPS / 10 pod 过渡到 5000 QPS / 20 pod——平滑

没有 P2C（纯 RR）的情况：

  • RR 按固定顺序轮询、不看 load——新 endpoint 和老 endpoint 被公平对待
  • 新 endpoint 刚启动时 JVM/Golang runtime 在 warm up、p99 latency 可能 10x 于老 endpoint
  • 用户的 P99 latency 出现 spike——PagerDuty 炸
  • SRE 开始调 HPA 参数"再激进点预热 Pod"——治标不治本

P2C 自带 "避开慢 endpoint" 的能力——新 endpoint 因为 warm up 阶段 PeakEwma 较高、被 P2C 自动少选，直到它稳定后再承接正常份额。两个机制各自工作、叠加起来产生远超单独工作的效果——这也是 Load metric 不能选错的原因：如果 balance 用 PendingRequests 而不是 PeakEwma，新 Pod 在 warm up 时 pending 数反而低（因为还没接几个请求），balance 反倒会优先把新请求打到慢 endpoint 上。

下一章我们读 Tower 的一组更 low-level 的中间件：Filter、MapRequest、Steer——它们不涉及容量管理，专注于"请求变换"。