plumbing

主线位置：主线 2 的中段。sum 造好 SumConsumer、源头拿出 IterProducer 之后，bridge_producer_consumer 只做四件事：问消费者 full() 没有；切生产者和消费者；在叶子上 producer.fold_with(consumer.into_folder())；回溯时 reducer.reduce(left, right)。本页讲这几个调用背后的 trait。

flowchart LR
  subgraph prod["生产者（数据）"]
    P["Producer：split_at(i)、into_iter、fold_with"]
    UP["UnindexedProducer：split() → (Self, Option<Self>)"]
  end
  subgraph cons["消费者（怎么接）"]
    C["Consumer：split_at(i) → (L, R, Reducer)、into_folder、full"]
    UC["UnindexedConsumer：split_off_left、to_reducer"]
  end
  F["Folder：consume、consume_iter、complete、full"]
  R["Reducer：reduce(left, right)"]
  C -->|叶子| F
  C -->|切开时| R
  P -->|"fold_with(folder)"| F
  UC -->|继承| C

src/iter/plumbing/mod.rs 一共 476 行，前 243 行是六个 trait，后面是 Splitter 和三个 bridge 函数。模块注释说这些是「low-level details」，普通用户不需要直接碰；自己写并行数据结构、或者想看懂任何一个适配器的源码，就得先读这六个 trait。

Producer：可按下标切的数据

Producer：

pub trait Producer: Send + Sized {
    type Item;
    type IntoIter: Iterator<Item = Self::Item> + DoubleEndedIterator + ExactSizeIterator;

    fn into_iter(self) -> Self::IntoIter;
    fn min_len(&self) -> usize { 1 }
    fn max_len(&self) -> usize { usize::MAX }
    fn split_at(self, index: usize) -> (Self, Self);
    fn fold_with<F>(self, folder: F) -> F
    where
        F: Folder<Self::Item>,
    {
        folder.consume_iter(self.into_iter())
    }
}

• split_at(index) 按值消费自己，返回左右两半。切片的实现就是 slice.split_at(index)（slice/mod.rs:871），区间是 start..mid 和 mid..end。左右两半会被 join_context 送到可能不同的线程，所以 Producer: Send。
• into_iter() 返回一个普通的顺序迭代器，而且要求是 DoubleEndedIterator + ExactSizeIterator：rev() 需要从后往前，enumerate() 需要知道这一段有多长。
• fold_with(folder) 是叶子上真正干活的地方，默认实现就是把 into_iter() 整个交给 folder.consume_iter。生产者可以覆盖它做得更快，比如 MapProducer 把 map_op 包进 folder 里，让内层用 std 的 Iterator::map。
• min_len / max_len 是给切分器的提示，默认 1 和 usize::MAX，由 with_min_len / with_max_len 包一层改掉（见 bridge 与自适应切分）。

Producer 只有 indexed 的迭代器才有。长度不确定的数据源（par_split、par_chunk_by、walk_tree、par_bridge、u64 的超长区间）实现的是 UnindexedProducer：

pub trait UnindexedProducer: Send + Sized {
    type Item;
    fn split(self) -> (Self, Option<Self>);
    fn fold_with<F>(self, folder: F) -> F
    where
        F: Folder<Self::Item>;
}

split() 没有下标参数，自己决定切在哪；返回 None 表示「切不动了」。par_split(sep) 的 split 从中点往后找分隔符，找不到再往前找（split_producer.rs:104-144）；par_bridge 的 split 只是把一个计数器减一（见 par_bridge）。

Consumer：下游怎么接

Consumer<Item>：

pub trait Consumer<Item>: Send + Sized {
    type Folder: Folder<Item, Result = Self::Result>;
    type Reducer: Reducer<Self::Result>;
    type Result: Send;

    fn split_at(self, index: usize) -> (Self, Self, Self::Reducer);
    fn into_folder(self) -> Self::Folder;
    fn full(&self) -> bool;
}

三个关联类型讲的是消费者的生命周期：被切开时产出两个子消费者和一个 Reducer；到了叶子变成 Folder；Folder 最后 complete() 得到 Result，Result 一路被 Reducer 两两合并回根。Result: Send，因为右边的结果在另一个线程上算出来，要送回来合并。

split_at(index) 带下标，是因为有的消费者要知道左边有多少个元素：collect 的 CollectConsumer 拿着目标缓冲区的裸指针，切开时左边拿 [start, start + index)，右边拿剩下的（见 collect 到 Vec）。sum 不关心下标，split_at 直接返回三个新的 SumConsumer（sum.rs:41）。

UnindexedConsumer 在 Consumer 上加两个方法：

pub trait UnindexedConsumer<I>: Consumer<I> {
    fn split_off_left(&self) -> Self;
    fn to_reducer(&self) -> Self::Reducer;
}

切的时候不知道左边多长，只能「从我这里分出一个左边的」，自己当右边。bridge_unindexed 里就是 (consumer.to_reducer(), consumer.split_off_left(), consumer) 三件套（mod.rs:462-463）。

CollectConsumer 也实现了 UnindexedConsumer，但它的 split_off_left 是 unreachable!("CollectConsumer must be indexed!")（collect/consumer.rs:156）：类型上要求它能当 unindexed 消费者用（drive_unindexed 的签名），实际上只有 opt_len() 返回 Some 的迭代器才会拿到它，而这些迭代器都走 indexed 的 bridge。

Folder 与 Reducer

Folder<Item> 是叶子上的顺序累加器：

pub trait Folder<Item>: Sized {
    type Result;
    fn consume(self, item: Item) -> Self;
    fn consume_iter<I>(mut self, iter: I) -> Self
    where
        I: IntoIterator<Item = Item>,
    {
        for item in iter {
            self = self.consume(item);
            if self.full() {
                break;
            }
        }
        self
    }
    fn complete(self) -> Self::Result;
    fn full(&self) -> bool;
}

consume 按值拿走 self 再还回来，和 std 的 fold 一个形状，状态全在值里，不需要 &mut。默认的 consume_iter 每个元素后问一次 full()；不会短路的消费者通常覆盖它，直接用 std 的迭代器方法，比如 SumFolder::consume_iter 是 add(self.sum, iter.into_iter().sum())（sum.rs:94-101），让编译器去向量化。

Reducer<Result> 只有一个方法 fn reduce(self, left: Result, right: Result) -> Result。left 一定是下标靠前的那一半，所以 collect、reduce 这类要保序的操作只要在 reduce 里按左右顺序拼就行，不管右边是在哪个线程上、什么时候算完的。

以 map 为例：四层包装

map 在 src/iter/map.rs（255 行）里实现了全部四个角色，是读其他适配器的模板：

核心逻辑一共三行：MapFolder::consume 里的 (self.map_op)(item)，consume_iter 里的 iter.into_iter().map(self.map_op)，MapProducer::into_iter 里的 self.base.into_iter().map(self.map_op)。其余 200 多行都是把这三行接到推、拉两条路径上的转发。

map_op 存成 &'f F 而不是 F：切成几千份时每份只复制一个指针，F 本身留在根上那个 Map（或 Callback）里，所以只需要 F: Sync，不需要 F: Clone。这也是 rayon 的闭包普遍要求 Fn + Sync + Send 的原因：Send 让 Map 结构体能整体送到池里，Sync 让各线程同时拿着 &F 调用。

两条路径在 Map 上都走得通，取决于下游：map(f).sum() 走消费者路径（Map::drive_unindexed），到源头 bridge 时生产者是裸的 IterProducer、消费者是 MapConsumer<SumConsumer>；map(f).zip(b) 走生产者路径（Zip 调 Map::with_producer），生产者是 MapProducer<IterProducer>。两种组合在叶子上做的事一样。

full()：短路

消费者的 full() 被问三个地方：bridge_producer_consumer 每一层开头（满了就直接 consumer.into_folder().complete()，不再切也不再跑，mod.rs:404-405）；Folder::consume_iter 的默认实现每个元素之后；以及各个生产者 fold_with 里（比如 ChainProducer 跑完 a 后先问一下再跑 b）。

find_any 的实现（find.rs）把这个机制用到底：

pub(super) fn find<I, P>(pi: I, find_op: P) -> Option<I::Item>
where
    I: ParallelIterator,
    P: Fn(&I::Item) -> bool + Sync,
{
    let found = AtomicBool::new(false);
    let consumer = FindConsumer::new(&find_op, &found);
    pi.drive_unindexed(consumer)
}

found 在调用者的栈上，所有子消费者、所有 folder 都拿着同一个 &AtomicBool。full() 就是 found.load(Relaxed)；某个 folder 找到了就 store(true, Relaxed)；consume_iter 用 take_while(|_| !found.load(Relaxed)) 包住迭代器再 find（find.rs:86-103），每取一个元素看一眼标志。FindReducer 是 left.or(right)。

为什么 Relaxed 就够：标志只是「可以停了」的提示，不用它发布任何数据。找到的元素本身作为 Result 经 join_context 的返回值传回来，join 的完成（latch 的 set / probe）已经建立了 happens-before。读到旧值的代价只是多看几个元素。

这也说明了 find_any 的「any」：哪个叶子先找到、哪个 Some 先被 or 到，取决于调度，返回的不一定是下标最小的那个。要最靠前的用 find_first，它在 find_first_last 里多记了一个「目前找到的最小下标」，只让比它靠前的部分继续跑。

try_fold、try_reduce、any、all、while_some、take_any 都是同一套：一个共享的 AtomicBool（或计数器），full() 读它。panic_fuse 也是：某个线程 panic 时置标志，其他线程尽快停下，免得等一个已经注定失败的任务跑完。

小结

• 六个 trait 分两边：生产者（Producer / UnindexedProducer）代表数据和怎么切，消费者（Consumer / UnindexedConsumer）代表怎么接；叶子上消费者变成 Folder，回溯时用 Reducer 合并。
• Producer::split_at(index) 与 Consumer::split_at(index) 在同一个下标切，所以生产者的第 i 个元素一定落到消费者的第 i 个位置上，collect 的就地写入靠的就是这一点。
• map 的核心逻辑三行，其余是把它接到推、拉两条路径的转发；闭包以 &F 的形式复制进每一片，所以要求 Sync。
• full() 在每层切分前、每个元素后被问；find_any 用一个调用者栈上的 AtomicBool（Relaxed）让所有线程尽快停下，返回哪个元素取决于调度。