collect 到 Vec

主线位置：主线 2 的终点换成 collect。(0..n).into_par_iter().map(f).collect::<Vec<_>>() 和 sum 走的是同一棵切分树，区别在消费者：每个叶子不再产出一个数，而是把元素直接写进目标 Vec 里属于自己的那一段；回溯时 Reducer 只检查两段是否首尾相接。

flowchart TB
  col["collect::<Vec<T>>()"] --> ext["Vec::par_extend"]
  ext --> q{"par_iter.opt_len()"}
  q -->|"Some(len)：indexed 链"| se["special_extend：reserve(len)"]
  se --> cc["CollectConsumer{start: 裸指针, len}"]
  cc --> split["split_at(i)：左 [start, start+i)，右 [start+i, start+len)"]
  split --> leaf["叶子：CollectResult 逐个 write，initialized_len += 1"]
  leaf --> red["CollectReducer：相邻才合并，不相邻丢掉右边"]
  red --> chk["assert 总数 == len → release_ownership → set_len"]
  q -->|"None：filter / flat_map"| lv["drive_unindexed(ListVecConsumer)"]
  lv --> lst["每个叶子一个 Vec，回溯时拼成 LinkedList"]
  lst --> app["reserve(总长)，逐个 append（再拷贝一次）"]

入口：Vec 的 par_extend

collect() 调 FromParallelIterator::from_par_iter，Vec<T> 的实现是「造一个空 Vec，par_extend 进去」（from_par_iter.rs:25-35）。真正的分叉在 impl ParallelExtend<T> for Vec<T>：

fn par_extend<I>(&mut self, par_iter: I)
where
    I: IntoParallelIterator<Item = T>,
{
    // See the vec_collect benchmarks in rayon-demo for different strategies.
    let par_iter = par_iter.into_par_iter();
    match par_iter.opt_len() {
        Some(len) => {
            // When Rust gets specialization, we can get here for indexed iterators
            // without relying on `opt_len`.  Until then, `special_extend()` fakes
            // an unindexed mode on the promise that `opt_len()` is accurate.
            super::collect::special_extend(par_iter, len, self);
        }
        None => {
            // This works like `extend`, but `Vec::append` is more efficient.
            let list = par_iter.drive_unindexed(ListVecConsumer);
            self.reserve(list.iter().map(Vec::len).sum());
            for mut other in list {
                self.append(&mut other);
            }
        }
    }
}

par_extend 的参数只是 IntoParallelIterator，拿不到 IndexedParallelIterator 的 len()。没有特化，就用 opt_len() 这个「承诺」来区分两条路，注释说得很清楚：快路径建立在 opt_len() 准确的前提上。

快路径：special_extend

special_extend 和 collect_into_vec 都落到 collect_with_consumer：

fn collect_with_consumer<T, F>(vec: &mut Vec<T>, len: usize, scope_fn: F)
where
    T: Send,
    F: FnOnce(CollectConsumer<'_, T>) -> CollectResult<'_, T>,
{
    // Reserve space for `len` more elements in the vector,
    vec.reserve(len);

    // Create the consumer and run the callback for collection.
    let result = scope_fn(CollectConsumer::appender(vec, len));

    // （注释略）
    let actual_writes = result.len();
    assert!(
        actual_writes == len,
        "expected {len} total writes, but got {actual_writes}"
    );

    // Release the result's mutable borrow and "proxy ownership"
    // of the elements, before the vector takes it over.
    result.release_ownership();

    let new_len = vec.len() + len;

    unsafe {
        vec.set_len(new_len);
    }
}

四步：reserve(len) 一次分配够；造一个指向 vec 末尾空闲区的消费者；并行跑完后断言写入总数恰好是 len；把元素的所有权从 CollectResult 交给 Vec，set_len。collect_into_vec(&mut v) 先 v.truncate(0) 再走同一条路，所以 v 原有容量够就不会重新分配。

CollectConsumer：切的是裸指针区间

CollectConsumer 只有 start: SendPtr<T> 和 len，外加一个 PhantomData<&'c mut T> 把它的生命周期绑在目标 Vec 的可变借用上。appender 断言 capacity - len >= 需要的长度，取 vec.as_mut_ptr().add(vec.len())（注释特意说明指针直接从 Vec 取，不经 Deref，以保证 provenance 覆盖整个分配）。

split_at(index) 把区间切成 [start, start + index) 和 [start + index, start + len)。生产者在同一个下标切（plumbing 里两个 split_at 的配对），所以第 i 个元素一定由负责第 i 个槽的那个叶子来写，各叶子的区间互不重叠，不需要任何同步。

CollectResult：既是 Folder，也是 Result

叶子上 into_folder() 产出 CollectResult：start、total_len（这片该写多少）、initialized_len（已经写了多少）。consume 是：

fn consume(mut self, item: T) -> Self {
    assert!(
        self.initialized_len < self.total_len,
        "too many values pushed to consumer"
    );

    // SAFETY: The assert above is a bounds check for this write, and we
    // avoid assignment here so we do not drop an uninitialized T.
    unsafe {
        // Write item and increase the initialized length
        self.start.0.add(self.initialized_len).write(item);
        self.initialized_len += 1;
    }

    self
}

用 ptr::write 而不是赋值，因为目标位置是未初始化内存，赋值会先 drop 那里的「旧值」。initialized_len 记录了哪些槽已经有值，Drop for CollectResult 只析构前 initialized_len 个：某个叶子里 map 的闭包 panic，已经写进去的元素会在展开时被正确析构，Vec 的长度不变。这是 rayon 1.3.1（#759，2020-06-15）的改动，发布说明原话是「Collecting to an indexed Vec now drops any partial writes while unwinding, rather than just leaking them」，并注明旧的泄漏行为也是安全的。

complete() 返回自己：CollectResult 同时是 Folder 和 Result。

CollectReducer：只合并相邻的

fn reduce(
    self,
    mut left: CollectResult<'c, T>,
    right: CollectResult<'c, T>,
) -> CollectResult<'c, T> {
    // Merge if the CollectResults are adjacent and in left to right order
    // else: drop the right piece now and total length will end up short in the end,
    // when the correctness of the collected result is asserted.
    unsafe {
        let left_end = left.start.0.add(left.initialized_len);
        if left_end == right.start.0 {
            left.total_len += right.total_len;
            left.initialized_len += right.release_ownership();
        }
        left
    }
}

（consumer.rs:167-185。）左片写满了，它的末尾才等于右片的开头，两片合成一片，右片交出所有权；左片没写满（某个生产者少产出了元素），就把右片原地 drop（析构它写过的元素），左片继续往上传。回到根时 result.len() 就是「从开头起连续写满的元素个数」，等于 len 才说明每个槽都恰好写了一次。

这个设计把正确性检查放到了最后一步：中间不加锁、不计数，任何生产者「少产出」都会表现为根上的总数不够，触发 assert!；「多产出」在叶子的 consume 里就被 assert! 拦下，不会写出界。两条断言让一个说谎的 opt_len 只会 panic，不会越界写或者把未初始化内存交给 Vec。

慢路径：LinkedList<Vec<T>>

opt_len() 为 None（filter、filter_map、flat_map 之后）时用 ListVecConsumer：每个叶子一个 ListVecFolder { vec: Vec::new() }，consume 就是 vec.push，consume_iter 是 vec.extend；complete 把非空的 Vec 放进一个单元素 LinkedList；ListReducer 是 left.append(&mut right)，链表拼接是 O(1)。回到 par_extend 后先 reserve 总长，再逐个 Vec::append。

代价有两层：每个叶子的 Vec 从空开始按倍数增长（filter 之后叶子上是逐个 consume 也就是逐个 push，FilterFolder 没有覆盖 consume_iter，没法按长度预分配）；最后所有元素还要再拷贝一次进目标 Vec。链表节点本身也是一次分配。顺序是保住的：链表按左右顺序拼，filter().collect() 的结果和顺序版一样。

1.9.0 起有 collect_vec_list()，直接把这个链表交给你，省掉最后一次拷贝；indexed 链调用它会得到只有一个 Vec 的链表。

Box<[T]>、Rc<[T]>、Arc<[T]> 都是先 collect 成 Vec 再 into()；VecDeque、BinaryHeap 等其他集合用 fast_collect（也按 opt_len 分两路，得到一个 Vec 或一串 Vec）再顺序 extend。

unzip_into_vecs 的就地写入是两层嵌套的 collect_with_consumer（collect/mod.rs:45-65）：外层给右边的 Vec 一个消费者，在它的回调里再给左边的 Vec 一个，两个消费者一起交给 unzip_indexed。

实验 E8：分配次数、缓冲区与两条断言

exp-rayon/collect-paths/ 装一个计数的全局分配器，统计 alloc / realloc 次数和请求的字节数。n = 10⁷ 个 u64（map 一个乘法），12 线程，从 main 调用：

（八次运行，跨四种负载。indexed 那一行是第二次起的稳定值：进程里第一次调用多 2–10 次、合计几百字节的小分配，来自 rayon 内部第一次用到的结构，之后不再出现。）

• indexed 链和顺序 collect 一样只分配一次、恰好 8000 万字节，没有 realloc：先 reserve 再就地写入，验证了「不重新分配」。
• filter(|_| true) 什么也没过滤掉，分配次数却从 1 变成几千：每个叶子一个 Vec（叶子数随窃取变化，所以次数每次不同），每个 Vec 增长时多次 realloc，最后 reserve 一次 8000 万字节再拷贝。「请求字节」是每次 alloc / realloc 新大小的累加，约为最终大小的 4 倍多。
• 只有一个叶子时 3 次 alloc：叶子的 Vec、链表节点、目标 Vec；22 次 realloc 是那个 Vec 从小翻倍长到 1000 万。

缓冲区不换：

collect_into_vec(cap=n)    same_ptr=true len=10000000 allocs=0 reallocs=0
par_extend(reserved)       same_ptr=true len=10001000 allocs=0 reallocs=0

v 预先 with_capacity(n) 并放进 1 个元素，collect_into_vec 先 truncate(0)，容量够，指针不变、零分配；已有 1000 个元素再 reserve(n) 的 Vec 做 par_extend，新元素接在后面，零分配。容量只有 n/2 时有 1 次 realloc（这次指针碰巧没变，是系统分配器原地扩容，不是 rayon 的行为）。

耗时（中位数，见下方负载）：

（n = 10⁷，12 线程，7 次取中位数，测前 1 分钟 / 5 分钟负载 1.97 / 2.98。此前两轮在 5 分钟负载很高时测过，比例相同，不列出。）indexed 链比顺序版快不到 2 倍，因为这个循环几乎只是写内存；filter 链慢 8 倍，比顺序版还慢 4.5 倍，多出来的是几千次分配、realloc 时的拷贝和最后一次整体拷贝。同一进程里反复分配同样大小的 8000 万字节，系统分配器可能直接复用刚释放的区域，所以这几个数里缺页的开销很少（推测，没有单独测）。

反向验证：谎报长度。 写一个 Liar 迭代器，内部就是 (0..1000).into_par_iter() 的 bridge，但 opt_len() 返回 1000 + delta：

delta = +1：panic: expected 1001 total writes, but got 1000
delta = -1：panic: too many values pushed to consumer

多报一个：每个叶子都写满了自己的区间，但最后一个槽没人写，根上总数 1000 ≠ 1001，collect_with_consumer 的断言触发。少报一个：CollectConsumer 的区间比生产者少一格，最右边那个叶子写第 n 个元素时 initialized_len == total_len，consume 的断言触发，不会写出界。两种情况都是 panic，Vec 的长度没有被改动。这两条消息就是源码里的原文。

小结

• Vec::par_extend 按 opt_len() 分两路：有值就 special_extend，否则 ListVecConsumer。没有特化，opt_len 就是「我是 indexed」的承诺。
• 快路径：一次 reserve，CollectConsumer 与生产者在同一个下标切出不相交的裸指针区间，叶子用 ptr::write 写入并记 initialized_len，CollectReducer 只合并首尾相接的两片，最后断言总数再 set_len。panic 时已写的元素由 CollectResult::drop 析构（1.3.1 起）。
• 慢路径：每个叶子一个 Vec，拼成 LinkedList，最后再拷贝一次。E8：10⁷ 个元素 indexed 1 次分配，filter(|_| true) 四千到八千多次分配加一两万次 realloc，耗时多 8 倍。
• 谎报 opt_len 只会触发「expected N total writes」或「too many values pushed to consumer」两条断言之一，不会越界。
• collect_into_vec / 预留过容量的 par_extend 不换缓冲区（E8：指针不变、零分配），适合在循环里复用同一个 Vec。