par_bridge

主线位置：主线 2 的一个旁支。源头不是能按下标切的数据，而是一个只有 next() 的普通 Iterator（文件行、channel、数据库游标）。par_bridge() 把它包成一个 UnindexedProducer，照样走 bridge_unindexed 和 join_context，只是每个叶子都去同一把锁后面抢元素。

flowchart TB
  it["iter.par_bridge()：IterBridge{iter}"] --> du["drive_unindexed：在自己的栈上建 Mutex<Fuse<Iter>>、split_count = 线程数、threads_started[线程数]"]
  du --> bu["bridge_unindexed(&IterParallelProducer, consumer)"]
  bu --> sp{"split()：split_count 还能减 1？"}
  sp -->|是| two["(self, Some(self))：两半是同一个引用"]
  two --> bu
  sp -->|否| leaf["fold_with：本线程已经在跑一个叶子？"]
  leaf -->|"是（重入）"| ret["直接返回 folder"]
  leaf -->|否| loop["循环：lock → next() → unlock → folder.consume"]
  loop -->|"next() 为 None 或 folder.full()"| done["返回"]

实现：一个计数器和一把锁

par_bridge.rs 157 行。IterBridge::drive_unindexed：

fn drive_unindexed<C>(self, consumer: C) -> C::Result
where
    C: UnindexedConsumer<Self::Item>,
{
    let num_threads = current_num_threads();
    let threads_started: Vec<_> = (0..num_threads).map(|_| AtomicBool::new(false)).collect();

    bridge_unindexed(
        &IterParallelProducer {
            split_count: AtomicUsize::new(num_threads),
            iter: Mutex::new(self.iter.fuse()),
            threads_started: &threads_started,
        },
        consumer,
    )
}

生产者是 &IterParallelProducer，一个共享引用：所有「切」出来的生产者都是同一个对象，Mutex 和计数器都在 drive_unindexed 的栈帧上（两个 trait 里说的「在自己栈上准备资源、以引用交出去」）。fuse() 保证迭代器返回 None 之后再调 next() 仍是 None，因为会有好几个叶子在最后各调一次。

split 只是把 split_count 减一：

fn split(self) -> (Self, Option<Self>) {
    // Check if the iterator is exhausted
    let update = self
        .split_count
        .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |c| c.checked_sub(1));
    (self, update.is_ok().then_some(self))
}

从线程数开始，减到 0 就不再切。bridge_unindexed 的 Splitter 也在限制，两者取先到者；不管树长什么样，成功的 split 最多是线程数次，叶子最多线程数 + 1 个。实测 12 线程时 fold(|| (), ..).count() 五次都是 13。

fold_with 是每个叶子的主循环：

loop {
    if let Ok(mut iter) = self.iter.lock() {
        if let Some(it) = iter.next() {
            drop(iter);
            folder = folder.consume(it);
            if folder.full() {
                return folder;
            }
        } else {
            return folder;
        }
    } else {
        // any panics from other threads will have been caught by the pool,
        // and will be re-thrown when joined - just exit
        return folder;
    }
}

每个叶子都一直取到迭代器耗尽为止，所以 13 个叶子并不是「各处理 1/13」，而是 12 个线程同时从一把锁后面抢元素，谁抢到谁处理。每取一个元素：加锁、next()、解锁、处理。锁中毒（某个 consume panic 了）时直接退出，panic 会经 join 传回调用方。

这是 1.6.1（2022-12-09）的版本，发布说明：「Simplified par_bridge to only pull one item at a time from the iterator, without batching. Threads that are waiting for iterator items will now block appropriately rather than spinning CPU.」等元素的线程阻塞在 Mutex 上，而不是自旋。

防递归死锁

同一个 1.6.1 的另一条：「Added protection against recursion in par_bridge, so iterators that also invoke rayon will not cause mutex recursion deadlocks」（168d5a7，#997）。

场景：迭代器的 next() 里又调用了 rayon（比如每取一个元素都要用 par_iter 做点预处理）。next() 是在持锁时调用的；这个线程在内层 join 里等待时会去偷任务（见 WorkerThread 主循环），偷到的可能正好是同一个 par_bridge 的另一个叶子，于是它去 lock() 一把本线程已经持有的 Mutex。std::sync::Mutex 不可重入，死锁。下游的 consume 里调用 rayon 也会发生同样的重入，但那时锁已经释放（drop(iter) 在 consume 之前），注释说这种重入没有互斥锁的危险，只是统一检查一次开销更低。

修法在 fold_with 开头（par_bridge.rs:121-137）：

if let Some(i) = current_thread_index() {
    let thread_started = &self.threads_started[i % self.threads_started.len()];
    if thread_started.swap(true, Ordering::Relaxed) {
        // We can't make progress with a nested mutex, so just return and let
        // the outermost loop continue with the rest of the iterator items.
        return folder;
    }
}

每个 worker 一个标志，第一次进 fold_with 时置位；同一个线程再次进入（只可能是嵌套）时直接返回空 folder，让外层那个循环去把剩下的元素取完。标志从不复位，因为一个线程的第一个叶子一定会一直取到迭代器耗尽，不需要第二个叶子。注释还说明了两点：如果线程池以后能动态增长，下标取模会让几个线程共用一个标志，最多是误判为重入、少一点并行，不影响正确性；不在 worker 上（current_thread_index() 为 None）时不检查，因为那种线程不会偷任务。

不保序

元素按「哪个线程先抢到锁」分配，下游 collect 时的顺序是叶子的左右顺序，而每个叶子拿到的元素在原序列里是交错的。实测 (0..100_000u64).par_bridge().collect::<Vec<_>>()，12 线程时（四次运行）有 99973–100000 个位置上的元素不等于下标；在 1 线程池里是 0 个（反向验证：只有一个线程抢锁，第一个叶子按顺序取完全部元素）。

需要顺序的话：在迭代器上先 enumerate()，处理后按下标排序或写回；或者干脆先 collect 成 Vec 再 par_iter()，两者都恢复成 indexed。

实验：每元素的代价

exp-rayon/par-bridge/，12 线程，从 main 调用，每个元素做 rounds 次乘加（rounds = 1000 约 0.94 µs），最后 reduce 求和。四种写法：顺序、(0..n).into_par_iter()、(0..n).par_bridge()、先 collect::<Vec<_>>() 再 into_par_iter()。5 次取中位数，测前 1 分钟 / 5 分钟负载 1.97 / 2.98（测完 3.15 / 3.19）。此前两轮在 5 分钟负载很高、正在回落时测过，数字同量级，不列出：

• 每元素约 76 ns 时，par_bridge 比顺序执行还慢一倍多；约 1 µs 时也只比顺序快 1.4 倍，而 par_iter 快 10 倍。
• 每元素约 10 µs 时 par_bridge 才接近 par_iter（121 对 96 ms）。
• 「先 collect 再 par_iter」在每一档都和 par_iter 几乎一样：多一次顺序的 collect（10⁶ 个 u64 不到 1 ms），换来 indexed 的切分和零锁。

par_bridge 慢在锁上。12 个线程每取一个元素就抢一次同一把 Mutex，本机的 std::sync::Mutex 在 macOS 上是 pthread mutex（rust-src 1.99.1 sys/sync/mutex/mod.rs:21-27，Linux 用 futex）。从「每元素约 1 µs 时 10⁶ 个元素 674 ms」反推，锁的吞吐大约是每 0.67 µs 交出一个元素，而且工作越轻、争用越激烈，单个元素的代价反而越高（推测是争用下锁在线程间交接要经过内核的阻塞与唤醒；没有用 profiler 拆分验证）。

什么时候用

• 元素来自一个本来就是顺序的源（IO、channel、外部库的迭代器），每个元素的处理在几微秒以上，元素总数事先不知道或者太多不想全部放进内存：par_bridge 合适。
• 元素能先放进内存：先 collect 成 Vec 再 par_iter()，几乎总是更快，而且保序。
• 每个元素的处理很轻：别并行，或者先把元素攒成批（比如 chunks 之后对每批 par_bridge），把锁的次数降下来。

小结

• par_bridge 的生产者是栈上 IterParallelProducer 的共享引用：split 只减计数，最多切线程数次、线程数 + 1 个叶子（实测 12 线程 13 个）。
• 每个叶子循环「加锁 → next() → 解锁 → 处理」直到迭代器耗尽，1.6.1 起一次只取一个元素、等锁时阻塞。
• 1.6.1 的 threads_started 标志让同一线程重入时直接返回，避免在 next() 或 consume 里调用 rayon 导致的不可重入 Mutex 死锁。
• 不保序（实测 10⁵ 个元素几乎全部错位，1 线程时 0 个）。
• 实测每元素约 76 ns 时比顺序还慢，约 1 µs 时只有 1.4 倍，约 10 µs 才接近 par_iter；能先 collect 就先 collect。