bridge 与自适应切分

主线位置：主线 2 的核心。源头拿到生产者以后，bridge_producer_consumer 递归地决定「这一段还切不切」：切就 join_context 两半，不切就在当前线程上顺序跑完。每一层 join_context 是主线 1（见 join 与 join_context），它告诉切分器的唯一信息是 migrated()：这一半是不是换了线程。

flowchart TB
  start["bridge_producer_consumer(len, producer, consumer)"] --> mk["LengthSplitter::new(min_len, max_len, len)：splits = max(线程数, len / max)"]
  mk --> h["helper(len, migrated, splitter, ...)"]
  h --> full{"consumer.full()？"}
  full -->|是| done["into_folder().complete()"]
  full -->|否| ts{"len / 2 >= min 且 Splitter 同意？"}
  ts -->|"否"| leaf["producer.fold_with(consumer.into_folder()).complete()"]
  ts -->|"是"| sp["mid = len / 2；producer.split_at(mid)；consumer.split_at(mid)"]
  sp --> jc["join_context(左 helper, 右 helper)，各自带 context.migrated()"]
  jc --> h
  jc --> red["reducer.reduce(左结果, 右结果)"]

三个 bridge

plumbing/mod.rs 的后半段只有三个公开函数：

bridge_producer_consumer 的递归体：

fn helper<P, C>(
    len: usize,
    migrated: bool,
    mut splitter: LengthSplitter,
    producer: P,
    consumer: C,
) -> C::Result
where
    P: Producer,
    C: Consumer<P::Item>,
{
    if consumer.full() {
        consumer.into_folder().complete()
    } else if splitter.try_split(len, migrated) {
        let mid = len / 2;
        let (left_producer, right_producer) = producer.split_at(mid);
        let (left_consumer, right_consumer, reducer) = consumer.split_at(mid);
        let (left_result, right_result) = join_context(
            |context| {
                helper(
                    mid,
                    context.migrated(),
                    splitter,
                    left_producer,
                    left_consumer,
                )
            },
            |context| {
                helper(
                    len - mid,
                    context.migrated(),
                    splitter,
                    right_producer,
                    right_consumer,
                )
            },
        );
        reducer.reduce(left_result, right_result)
    } else {
        producer.fold_with(consumer.into_folder()).complete()
    }
}

几点：

• 永远对半切：mid = len / 2，不看数据、不看线程忙闲。树的形状只由「在哪一层停」决定。
• splitter 是 Copy 的，按值传进两个闭包，左右各拿一份副本、各自往下减。所以切分决策完全是局部的，没有任何共享状态，也没有原子操作。
• 根调用 helper(len, false, ...)：第一层永远当作「没被偷」。
• 叶子上的 fold_with(...).complete() 在当前线程上执行。(0..1).into_par_iter() 这种切不动的，或者 with_min_len(len)，从头到尾不会进入线程池（E2：threads_used=0）。

Splitter：三行规则

Splitter 只有一个字段 splits: usize，注释叫它「thief-splitting」：先切出够所有线程分的任务，此后每当一个任务真的被偷到别的线程，就重置。

fn new() -> Splitter {
    Splitter {
        splits: crate::current_num_threads(),
    }
}

fn try_split(&mut self, stolen: bool) -> bool {
    let Splitter { splits } = *self;

    if stolen {
        // This job was stolen!  Reset the number of desired splits to the
        // thread count, if that's more than we had remaining anyway.
        self.splits = Ord::max(crate::current_num_threads(), self.splits / 2);
        true
    } else if splits > 0 {
        // We have splits remaining, make it so.
        self.splits /= 2;
        true
    } else {
        // Not stolen, and no more splits -- we're done!
        false
    }
}

（mod.rs:258-284。）

• 初值是 current_num_threads()：在 worker 上是所在池的线程数，不在任何池里是全局池的线程数。
• 没被偷：splits > 0 就切，并减半。12 线程时依次是 12 → 6 → 3 → 1 → 0，共切 4 层，得到 2⁴ = 16 片。一般地，splits = t 时切 ⌊log₂t⌋ + 1 层，片数是大于 t 的最小 2 的幂。
• 被偷（stolen 即 migrated() 为真）：把 splits 重置为 max(t, splits / 2)，并且这一层照切。理由写在注释里：任务被偷说明有线程闲着，值得为它们再切出一批。

为什么「被偷」能代表「有人闲着」：join_context 把 B 推进本地 deque，A 在本线程跑；只有别的线程没活干、去偷的时候，B 才会换线程（见 WorkerThread 主循环）。B 被偷，它那一支的 splits 回到线程数，在新线程上再展开一棵小树。负载均匀时几乎不偷，树就停在 16 片；某一支特别慢，空闲线程会不断偷它的后代，那一支就被切得更细。这就是「自适应」：不需要事先知道每个元素的代价。

bridge_unindexed 直接用 Splitter；bridge_producer_consumer 用的是外面包了一层的 LengthSplitter。

LengthSplitter：再加两条

LengthSplitter：

fn new(min: usize, max: usize, len: usize) -> LengthSplitter {
    let mut splitter = LengthSplitter {
        inner: Splitter::new(),
        min: Ord::max(min, 1),
    };

    // Divide the given length by the max working length to get the minimum
    // number of splits we need to get under that max.  This rounds down,
    // but the splitter actually gives `next_power_of_two()` pieces anyway.
    // e.g. len 12345 / max 100 = 123 min_splits -> 128 pieces.
    let min_splits = len / Ord::max(max, 1);

    // Only update the value if it's not splitting enough already.
    if min_splits > splitter.inner.splits {
        splitter.inner.splits = min_splits;
    }

    splitter
}

fn try_split(&mut self, len: usize, stolen: bool) -> bool {
    // If splitting wouldn't make us too small, try the inner splitter.
    len / 2 >= self.min && self.inner.try_split(stolen)
}

• min 是硬下限：len / 2 >= min 不成立就不切，而且 && 短路，内层的 splits 也不会被消耗。默认 min = 1，所以 len >= 2 才切，n 个元素最多 n 片。
• max 只在开头起作用：splits 至少是 len / max，于是不被偷时也至少切到每片 ≤ max。
• min / max 来自 producer.min_len() / max_len()，也就是 with_min_len / with_max_len 包上去的 MinLen / MaxLen。它们的生产者只改这两个方法，split_at 原样转发。zip 取两边 min_len 的较大者、max_len 的较小者。

两条冲突时 min 赢：with_max_len 的文档举的例子是 min = 10、max = 15，长度 16 的片不会再切（切了就是 8 < 10）。

规划时的发现：从池外调用多切一层

按上面的规则，12 线程应该最少切出 16 片。但从 main 线程调 (0..n).into_par_iter()... 时，最少是 64 片。原因在 rayon-core 里：

1. main 不是 worker。根 helper 在 main 上执行，try_split(len, false)：12 → 6，切。
2. 它调 join_context，进 registry::in_worker，当前线程不是 worker，走 in_worker_cold：把整个 join_context 的闭包包成 StackJob 注入全局队列，main 在 LockLatch 上阻塞。某个 worker 取到它，以 op(&*worker_thread, true) 调用，即 injected = true。
3. join_context 里，A 的 FnContext 用的就是这个 injected；B 在本线程被弹出时是 job_b.run_inline(injected)，被偷时 StackJob::execute 经 JobResult::call 传 func(true)（job.rs:217-218）。两边的 migrated() 都是 true。
4. 于是第二层两个 helper 都走 stolen 分支：splits = max(12, 6 / 2) = 12，切，而且 splits 没有减少。
5. 之后每一支从 12 开始再切 4 层。总层数 1 + 1 + 4 = 6，2⁶ = 64 片。

在 worker 上调用（ThreadPool::install 里、rayon::scope 里、或者一个 rayon 任务里嵌套的 par_iter），根 join_context 走 in_worker 的第一条路，injected = false，没有这一层重置，最少 16 片。

一般地，t 个线程时设 L = ⌊log₂t⌋ + 1：池内最少 2ᴸ 片，池外最少 2ᴸ⁺² 片。

这是「最少」：只要发生窃取，被偷的那一支就会多切。表里的数字是在没有窃取时才会精确出现的值；1 线程没有别的线程可偷，结果完全确定。

这多出来的一层是否有意为之，代码和注释里都没有说明（in_worker_cold 里 injected 的用途是让 join 的两个闭包知道自己「被换了线程」，这个语义本身是对的：A 确实不在调用者的线程上跑）。效果是从池外发起的并行迭代切得更细，叶子更多；对计算量大的循环影响不大，对只有几十个元素的小循环意味着更多次 join（见常见坑 tiny-tasks）。

实验 E2：数叶子

exp-rayon/split-count/ 自写一个 Producer（Counted，对 0..n 的区间），在 split_at 里计数切分次数、在 fold_with 里计数叶子数并记录每个叶子的长度和执行线程；同一次运行再用 fold(|| (), ...).count() 数一遍（每个叶子的 FoldFolder 恰好产出一个元素，count 就是叶子数）。每次运行都检查两条恒等式：fold().count() == 叶子数，切分次数 + 1 == 叶子数（二叉树）。所有配置、所有重复里两条都成立，所以 fold().count() 这个计数法没有偏差。

「池内」是在 rayon::scope(|_| ...) 的闭包里调用（闭包在全局池的 worker 上执行），「池外」是直接在 main 里调用。线程数用 RAYON_NUM_THREADS 控制。

1 线程（反向验证：没有窃取，必须完全确定）：

main     n=1         leaves=[1, 1, 1, 1, 1]  threads_used=[0, 0, 0, 0, 0]
main     n=3         leaves=[3, 3, 3, 3, 3]
inside   n=3         leaves=[2, 2, 2, 2, 2]
main     n=1048576   leaves=[8, 8, 8, 8, 8]  leaf_len=[131072..131072]
inside   n=1048576   leaves=[2, 2, 2, 2, 2]  leaf_len=[524288..524288]
main     n=1000000   min_len=100000   leaves=[8, 8, 8]
main     n=1000000   min_len=1000000  leaves=[1, 1, 1]  threads_used=[0, 0, 0]
main     n=1000000   max_len=10000    leaves=[128, 128, 128]  leaf_len=[7812..7813]
main     n=1000000   max_len=100      leaves=[16384, 16384, 16384]  leaf_len=[61..62]

每个数都能用上面的规则手算出来：n = 3 从 main 调用时根切成 1 + 2，右边被「重置」后再切成 1 + 1；min_len = 100000 时 10⁶ → 5×10⁵ → 2.5×10⁵ → 1.25×10⁵，再切就小于下限，8 片；max_len = 100 使 splits 初值为 10⁶ / 100 = 10000，减半 14 次到 0，2¹⁴ = 16384 片，每片 61–62 个，正是注释里「next_power_of_two() pieces」。threads_used=0 的两行（n = 1、min_len = n）说明没有切分时整个计算都在 main 上完成，根本没进线程池。

2 / 4 线程也符合表里的下限：2 线程池内最少 4 片、池外 16 片，4 线程池外 n = 1000 最少 32 片（一次运行恰好是 32）。

12 线程（受窃取影响，每格 5 次；测前 1 分钟 / 5 分钟负载 2.62 / 2.99）：

• 池内的下限 16 直接出现了（n = 1000 五次里三次恰好 16：这三次没有发生窃取）。池外这一轮没有出现 64，每次都有窃取。其他几轮里出现过：负载约 7 时池外 n = 1000 有一次恰好 64；本机负载一度超过 100（别的会话误开了大量线程）时，池外 n = 1000 五次里三次恰好 64。这四次 64 的叶子全部在同一个 worker 上执行（threads_used=1）：其他 worker 没被操作系统调度上来，没有窃取，树就停在下限。
• n = 17 池内是 16：12 线程本来只切 4 层，17 个元素切 4 层是 16 片（最后一片 2 个）；另一轮里偶尔是 17，那是有窃取把最后一片又切开了。
• 大数组上的叶子数主要由窃取决定，远多于下限，每次运行不同，随负载变化：n = 2²⁰ 在这一轮 1177–1727，另两轮 1258–1877、1133–1731，规划阶段 1460–3032。窃取发生得越多（每元素工作量越重、空闲线程越多），切得越细。
• with_min_len(10⁵) 在 12 线程下也恰好 8 片：下限让窃取无片可切。
• with_max_len(100) 在 16384 的基础上多出 1000 多片：max 只定了「至少切到这么细」，窃取照样会在它下面继续切到 min。

实验 E4：min_len / max_len 与耗时

exp-rayon/min-max-len/：n = 10⁶，(0..n).into_par_iter().with_min_len(m)（或 with_max_len(m)）.map(work).reduce(|| 0, wrapping_add)，从 main 调用，12 线程，5 次取中位数。两档工作量：轻（每元素 1 次乘加，可向量化，顺序版不到 1 ms）、重（每元素 200 次乘加，顺序版 168 ms，每元素约 170 ns）。测前 1 分钟 / 5 分钟负载 2.19 / 2.92；「叶子」是同一配置单独跑一次 fold().count() 的值，只作参考。

（min_len = 10、1000 与相邻两行同量级，表里省略。轻工作量的顺序基线很不稳：同一程序另一轮是 0.437 ms，这一轮 0.844 ms，推测与主线程被调度到性能核还是能效核有关，所以轻工作量只比较各并行配置之间的耗时，不算加速比。另两轮在 5 分钟负载很高时测过，形状相同：轻工作量最快的都是 min_len = 10⁴，max_len = 10 都是并行配置里最慢的。）

• 反向验证：min_len = n 时只有一个叶子，在 main 上顺序执行：重工作量 167.4 对 167.9 ms（1.00×）；轻工作量 0.738 对 0.844 ms，在顺序基线本身的波动范围内。
• 轻工作量：整个计算不到 1 ms，从 main 进池一次就要十几微秒，再加上几百个叶子各自的 join。把 min_len 提到 10²–10⁴ 能减少叶子，耗时从 0.32 ms 降到 0.11 ms；反方向把 max_len 压到 10，13 万个叶子让它慢到 0.49 ms，在另一轮里比顺序版还慢（0.638 对 0.437 ms）。
• 重工作量：默认已经是最好的（9.97×，12 个核里 4 个是能效核），各种 min_len / max_len 在 9.0–9.5× 之间；另一轮里 max_len 略好于默认，方向不一致，差别在 ±10% 以内。min_len = 10⁵ 只有 8 片，12 个线程里只有 8 个有活，掉到 7.5×（另一轮 7.0×），接近 8。
• 结论：with_min_len 是给每元素很便宜的循环用的，设到「每片的工作量明显大于一次 join」即可；对重计算设得太大，会让片数少于线程数。with_max_len 很少需要，除非元素代价极不均匀、又不希望依赖窃取来平衡。

历史

• rayon 0.7（2017-04-06）的发布说明写：「Parallel iterators now allow you to customize the min/max number of items to be processed in a given thread. This mechanism replaces the older weight mechanism」。对应 #260（合并提交 f75e2e6，2017-03-21），其中 c339b53「Remove the weight APIs」删掉了旧的「权重」接口，133e0cc 改名为 with_min_len / with_max_len。
• 那时的 Splitter 已经是 thief-splitting，但判断「被偷」的办法是比较线程：Splitter 多存一个 origin，取一个线程本地变量的地址当线程 ID，try() 里发现当前线程和 origin 不同就重置（f75e2e6 版本的 src/iter/internal.rs）。
• 7970d16「Introduce join_notify which tells whether jobs were stolen」（2017-05-31，经 #354 合并）去掉了这个线程本地变量，改由 join 直接告诉闭包是否被偷；它随 rayon 0.9.0 / rayon-core 1.3.0（2017-11-09）以 join_context + FnContext 的名字发布。按旧代码推算，从池外调用时根的两个孩子在 worker 上执行、线程 ID 与 main 不同，同样会重置，所以「池外多一层」在换成 migrated() 之前就存在（推算，没有运行旧版本验证）。
• 03e3d3a（2019-04-26）把 try 改名为 try_split（try 成了保留字）。此后到 v1.12.0，plumbing/mod.rs 的改动都是语法层面的（Rust 2018 路径、Ord::max 代替 cmp::max、文档链接），切分规则没有变。

小结

• bridge → with_producer → bridge_producer_consumer：永远对半切，LengthSplitter 是 Copy 的，每一支各自决定，没有共享状态。
• Splitter：初值线程数，不被偷就减半，被偷（migrated()）就重置为 max(线程数, splits/2) 并照切。t 个线程时池内最少 2^(⌊log₂t⌋+1) 片，12 线程是 16。
• 从池外调用时根 join 经 in_worker_cold 注入，两个孩子的 migrated() 都为真，多重置一层：12 线程最少 64 片，1 线程 8 片对 2 片（E2 完全确定）。
• 有窃取时叶子数远大于下限且每次不同（n = 2²⁰ 约 1100–1900，每元素 200 次乘法时 2200–5100，随负载变化）。
• E4：轻工作量时 with_min_len(10⁴) 把 0.32 ms 降到 0.11 ms，with_max_len(10) 升到 0.49 ms；重工作量时默认已是最好，片数少于线程数（min_len = 10⁵，8 片）掉到 7.5×。
• with_min_len 是硬下限，能把叶子数钉死（10⁵ → 8 片），设成 len 就完全不进线程池；with_max_len 只保证「至少切到这么细」。