适配器选讲

主线位置：主线 2 里「消费者往下推、生产者往上拉」的那一段。v.par_iter().map(f).sum() 的 map 只给消费者套了一层；换成 filter、zip、enumerate、chain，包法各不相同。本页挑六个代表，其余列表。

flowchart TB
  a["适配器"] --> q1{"需要按下标切上游吗？"}
  q1 -->|"不需要"| c["只包消费者：map / filter / flat_map / fold / inspect ..."]
  q1 -->|"需要"| p["包生产者：zip / enumerate / rev / skip / take / chunks ..."]
  c --> q2{"输出个数等于输入个数？"}
  q2 -->|是| idx["仍是 Indexed：map / inspect / cloned"]
  q2 -->|否| un["退化成 Unindexed：filter / flat_map / fold"]
  p --> idx2["Indexed，要求上游也是 Indexed"]
  ch["chain"] --> j["drive 里直接 join(a.drive, b.drive)"]

src/iter/ 下每个适配器一个文件，大多 100–300 行，其中真正做事的往往只有几行（plumbing 拆过 map.rs：255 行里核心三行）。读适配器只需要回答三个问题：它包的是消费者还是生产者；它还是不是 IndexedParallelIterator；它会不会让消费者 full()。

map：两条路都走

Map<I, F> 同时实现了 ParallelIterator 和（当 I 是 indexed 时）IndexedParallelIterator。输出个数等于输入个数，所以长度和下标都不变，opt_len 原样转发上游（map.rs:48-50）。被消费者驱动时包消费者，被要生产者时包生产者，两条路径见 plumbing。inspect、update、cloned、copied、map_with、map_init 都是同一个形状。

map_with(init, f) 和 map_init(init_fn, f) 多了一份每片一个的状态：map_with 在切分时 clone 一份 init，map_init 在每个叶子的 folder 里调一次 init_fn()。和 fold 一样，状态的个数等于叶子数，叶子数由切分器决定（见 bridge 与自适应切分），所以别在 init_fn 里做昂贵的事，也别指望它「每个线程调一次」。

filter：只包消费者，长度丢了

Filter::drive_unindexed 把消费者包成 FilterConsumer 往上传，FilterFolder::consume 里判断一下，不满足就原样返回自己：

fn consume(self, item: T) -> Self {
    let filter_op = self.filter_op;
    if filter_op(&item) {
        let base = self.base.consume(item);
        FilterFolder { base, filter_op }
    } else {
        self
    }
}

Filter 只实现 ParallelIterator，没有 opt_len（默认 None）。输出个数未知，后面就不能接 zip、enumerate、collect_into_vec、with_min_len；collect 也只能走 LinkedList<Vec<T>> 那条慢路（collect 到 Vec 的 E8：分配次数从 1 次变成几千次）。

但切分还是按上游的长度切的：filter 不改变 bridge 的位置，(0..n).into_par_iter().filter(p) 仍由 Range 的 bridge 用 LengthSplitter 切，每片过滤后剩多少个不影响切分。过滤掉 99% 元素的链，叶子上的 Vec 多数很短甚至为空（ListVecFolder::complete 对空 Vec 不入链表，extend.rs:134-140）。

FilterFolder 没有覆盖 consume_iter，注释说要特化就得在下游的 consume_iter 里检查 full()，不好做（#632）。所以 filter 后面的 folder 拿到的是逐个 consume，而不是一整段 std 迭代器。

flat_map 与 flat_map_iter：内层并不并行

flat_map(f) 的 f 返回一个 IntoParallelIterator。FlatMapFolder::consume 对每个外层元素：

let par_iter = map_op(item).into_par_iter();
let consumer = self.base.split_off_left();
let result = par_iter.drive_unindexed(consumer);

也就是在叶子上对每个外层元素再起一次完整的并行驱动：内层迭代器有自己的 bridge、自己的 LengthSplitter（初值又是线程数），切开时又是 join_context。外层元素很多、内层很短时，这就是成千上万次小的并行启动。

flat_map_iter(f)（flat_map_iter.rs）的 f 返回普通 IntoIterator，folder 里就是 std 的 iter.flat_map(map_op)，内层顺序执行，并行只发生在外层。外层元素已经够多、能把线程喂饱时，flat_map_iter 少一层调度；外层只有几个、每个内层很大时才需要 flat_map。flatten / flatten_iter 同理。

两者都没有 opt_len，都是 unindexed。

zip：两个上游在同一个下标切

zip 必须拿到两个生产者，见 两个 trait。它的 drive_unindexed 直接 bridge(self, consumer)，with_producer 先向 a 要生产者，在回调里再向 b 要，最后把两个合成 ZipProducer（zip.rs:59-120，两层 Callback 嵌套）。

fn split_at(self, index: usize) -> (Self, Self) {
    let (a_left, a_right) = self.a.split_at(index);
    let (b_left, b_right) = self.b.split_at(index);
    (
        ZipProducer { a: a_left, b: b_left },
        ZipProducer { a: a_right, b: b_right },
    )
}

长度取两边的较小值（Ord::min(self.a.len(), self.b.len())）；min_len 取两边较大的、max_len 取较小的（zip.rs:131-137），所以 a.with_min_len(100).zip(b) 的下限对整条链生效。zip_eq 在构造时断言两边等长。candle 的 softmax 就是 src.par_chunks(dim_m1).zip(dst.par_chunks_mut(dim_m1))（见 slice producer）：两个切片在同一个行号上切开，每片拿到的输入行和输出行一一对应。

enumerate：生产者带一个偏移

EnumerateProducer 只有 base 和 offset 两个字段：

fn split_at(self, index: usize) -> (Self, Self) {
    let (left, right) = self.base.split_at(index);
    (
        EnumerateProducer { base: left, offset: self.offset },
        EnumerateProducer { base: right, offset: self.offset + index },
    )
}

叶子上 into_iter() 返回 (offset..offset + len).zip(base)。下标是切分时算出来的，不需要任何共享计数器，所以 enumerate 之后的下标是确定的，和哪个线程跑、什么时候跑无关。rev 也是类似的思路：RevProducer::split_at(index) 对底层切在 len - index，然后左右对调（rev.rs）。

chain：自己调一次 join

Chain::drive 不经过 bridge：

let Chain { a, b } = self;
let (left, right, reducer) = consumer.split_at(a.len());
let (a, b) = join(|| a.drive(left), || b.drive(right));
reducer.reduce(a, b)

先按 a 的长度把消费者切成两半，join 两边各自驱动，各自走自己的 bridge、各自一个 LengthSplitter。所以 a.chain(b) 的叶子数大约是 a、b 各自叶子数之和，with_max_len 文档里专门写了「Chained iterators ... may each use their own maximum length」。

drive_unindexed（chain.rs:31-50）多一个判断：a.opt_len() 有值就用 consumer.split_at(len) 切，否则用 split_off_left()。注释说明了原因：collect 的 CollectConsumer 不能被当成真正的 unindexed 消费者来切（它的 split_off_left 是 unreachable!），知道长度时必须按下标切。

被要生产者时（chain 后面接了 zip 之类），ChainProducer 记着 a_len，split_at(index) 在 index <= a_len 时只切 a、把 b 切成空和全部，反之亦然（chain.rs:178-196）。

总表

按 src/iter/*.rs 里是否有 impl ... IndexedParallelIterator for 分类（v1.12.0）。「包什么」指主要实现方式；「短路」指会不会让消费者 full() 提前返回 true。

终结方法这一侧，会短路的有 find_any / find_first / find_last（及 find_map_*、position_*）、any、all、try_for_each*、try_reduce*，机制都是 plumbing 里那个共享标志。

三个「按顺序推进」的工具

• by_exponential_blocks() / by_uniform_blocks(n)（1.9.0）：BlocksCallback::callback 拿到生产者后，在当前线程上一个 while 循环：从剩余部分切出一块，对这一块调 bridge_producer_consumer 并行跑完，再切下一块，消费者 full() 了就停。指数版的块大小从 current_num_threads() 开始每次翻倍（exponential_size 是 saturating_mul(2)）。用途是「找到就停」的搜索：普通 find_first 一开始就把整个区间切给所有线程，后半段的线程可能白干很多活；按块推进时，前面的块找到了，后面的块根本不开始。代价是块与块之间是串行的屏障。
• walk_tree(root, children_of)（1.9.0）：对树形结构做并行遍历，walk_tree_prefix / walk_tree_postfix 保证前序 / 后序的产出顺序，walk_tree 不保证。实现是一个 UnindexedProducer，split() 把待访问的子节点分成两半。
• panic_fuse()：普通情况下一个线程 panic，其他线程仍会把手上的片跑完，join 等它们都结束才把 panic 传回来；panic_fuse 让其他线程在下一次 full() 检查时停下，panic 传回来得更快。

小结

• 适配器只回答三个问题：包消费者还是生产者；还是不是 indexed；会不会让 full() 提前为真。
• 输出个数等于输入个数的（map、inspect、cloned）保持 indexed；filter、flat_map、fold 退化成 unindexed，collect 失去就地写入。
• zip、enumerate、rev、skip、take 必须拿生产者，在同一个下标切上游；enumerate 的下标在切分时算出，与调度无关。
• flat_map 对每个外层元素都起一次完整的并行驱动；外层已经够多时用 flat_map_iter。
• chain 自己 join 一次，两边各一个切分器；skip 会执行被跳过的部分，take 不执行被丢掉的部分。