spawn 与 spawn_fifo

主线位置：主线 1 之外的分支。join 和 scope 都会等任务做完；spawn 把任务交出去就返回，没有人等它，于是要回答两个 join 不用回答的问题：池什么时候可以终止，任务 panic 了交给谁。

flowchart LR
  call["spawn(f)：f 必须 'static"] --> inc["increment_terminate_count"]
  inc --> hj["HeapJob：catch_unwind(f)，然后 terminate()"]
  hj --> where{"当前线程是这个池的 worker？"}
  where -->|"是"| push["推本地 deque"]
  where -->|"否"| inj["注入全局队列"]
  push --> run["某个 worker 执行"]
  inj --> run
  run --> ok{"f panic？"}
  ok -->|"否"| term["terminate：计数 -1"]
  ok -->|"是"| h{"有 panic_handler？"}
  h -->|"有"| handler["handler(err)"] --> term
  h -->|"没有"| abort["打印后 abort，退出码 134"]

代码在 spawn/mod.rs，163 行，大半是文档。

spawn：三件事

spawn/mod.rs 第 58–100 行：

pub fn spawn<F>(func: F)
where
    F: FnOnce() + Send + 'static,
{
    // We assert that current registry has not terminated.
    unsafe { spawn_in(func, &Registry::current()) }
}

pub(super) unsafe fn spawn_in<F>(func: F, registry: &Arc<Registry>)
where
    F: FnOnce() + Send + 'static,
{
    // We assert that this does not hold any references (we know
    // this because of the `'static` bound in the interface);
    // moreover, we assert that the code below is not supposed to
    // be able to panic, and hence the data won't leak but will be
    // enqueued into some deque for later execution.
    let abort_guard = unwind::AbortIfPanic; // just in case we are wrong, and code CAN panic
    let job_ref = spawn_job(func, registry);
    registry.inject_or_push(job_ref);
    mem::forget(abort_guard);
}

unsafe fn spawn_job<F>(func: F, registry: &Arc<Registry>) -> JobRef
where
    F: FnOnce() + Send + 'static,
{
    // Ensure that registry cannot terminate until this job has
    // executed. This ref is decremented at the (*) below.
    registry.increment_terminate_count();

    HeapJob::new({
        let registry = Arc::clone(registry);
        move || {
            registry.catch_unwind(func);
            registry.terminate(); // (*) permit registry to terminate now
        }
    })
    .into_static_job_ref()
}

1. 选池：Registry::current() 是当前 worker 所在的池，不在任何池里就是全局池。ThreadPool::spawn 直接传自己的 registry。
2. 占住池：increment_terminate_count，然后把 HeapJob 包成「先 catch_unwind(func)，再 terminate()」。HeapJob 里还克隆了一份 Arc<Registry>，任务执行时 registry 一定还在。
3. 入队：inject_or_push，在这个池的 worker 上就推本地 deque，否则注入全局队列（registry.rs 第 414–423 行）。

'static 约束是因为没有人等：调用方的栈帧随时可能消失。要借用栈上数据就用 scope。

spawn_in 外面套的 AbortIfPanic 是防御性的：注释说下面的代码「不应该会 panic」（increment_terminate_count 里有一个计数溢出的 assert!），万一 panic 了，HeapJob 已经分配、计数已经加上却没入队，就直接 abort，不让池处于计数永远不归零的状态。

terminate_count：池什么时候能停

registry.rs 第 565–600 行：

pub(super) fn increment_terminate_count(&self) {
    let previous = self.terminate_count.fetch_add(1, Ordering::AcqRel);
    debug_assert!(previous != 0, "registry ref count incremented from zero");
    assert!(previous != usize::MAX, "overflow in registry ref count");
}

pub(super) fn terminate(&self) {
    if self.terminate_count.fetch_sub(1, Ordering::AcqRel) == 1 {
        for (i, thread_info) in self.thread_infos.iter().enumerate() {
            unsafe { OnceLatch::set_and_tickle_one(&thread_info.terminate, self, i) };
        }
    }
}

terminate_count 是一个引用计数：ThreadPool 本身持 1（全局池的那个 1 永远不释放），每个 spawn 出去的任务持 1。ThreadPool::drop 只调一次 terminate()（thread_pool/mod.rs 第 399–403 行），减到 0 才给每个线程设置 terminate latch（Latch 家族里的 OnceLatch）。所以 drop 一个还有 spawn 任务没跑完的 ThreadPool，线程会把这些任务跑完再退出；drop 本身不等线程退出。increment_terminate_count 上的注释解释了 join / scope 为什么不用加计数：它们阻塞调用方，调用方要么在池里（池的 ThreadPool 还活着），要么在 install 里等着。

spawn_broadcast 用同样的方式，每个线程的那份任务各持 1（见 broadcast）。

spawn_fifo：本线程按 FIFO 执行

spawn/mod.rs 第 130–160 行：

pub(super) unsafe fn spawn_fifo_in<F>(func: F, registry: &Arc<Registry>)
where
    F: FnOnce() + Send + 'static,
{
    let abort_guard = unwind::AbortIfPanic; // just in case we are wrong, and code CAN panic
    let job_ref = spawn_job(func, registry);

    // If we're in the pool, use our thread's private fifo for this thread to execute
    // in a locally-FIFO order.  Otherwise, just use the pool's global injector.
    match registry.current_thread() {
        Some(worker) => worker.push_fifo(job_ref),
        None => registry.inject(job_ref),
    }
    mem::forget(abort_guard);
}

和 spawn 只差在入队：在这个池的 worker 上走 push_fifo（任务进本线程的 JobFifo，deque 里放一个指向 FIFO 的 JobRef，见 JobRef 与几种 Job），否则注入全局队列（Injector 本来就是 FIFO）。文档把它和废弃的 breadth_first 选项对比：breadth_first 把整个池的 deque 都换成 FIFO，spawn_fifo 只影响 spawn_fifo 提交的那些任务之间的相对顺序。1.1.0 / rayon-core 1.5.0 加入（8ccc028，#615）。

panic：交给 handler，没有就 abort

任务里的 panic 由 Registry::catch_unwind 处理：

pub(super) fn catch_unwind(&self, f: impl FnOnce()) {
    if let Err(err) = unwind::halt_unwinding(f) {
        // If there is no handler, or if that handler itself panics, then we abort.
        let abort_guard = unwind::AbortIfPanic;
        if let Some(ref handler) = self.panic_handler {
            handler(err);
            mem::forget(abort_guard);
        }
    }
}

接住 panic 后先放一个 AbortIfPanic 守卫。有 handler：调用它，正常返回就 forget 守卫；handler 自己 panic，展开时守卫被 drop，abort。没有 handler：函数结束，守卫被 drop，abort。AbortIfPanic 的 Drop 是：

impl Drop for AbortIfPanic {
    fn drop(&mut self) {
        eprintln!("Rayon: detected unexpected panic; aborting");
        ::std::process::abort();
    }
}

ThreadPoolBuilder::panic_handler 的文档（lib.rs 第 557–570 行）写的原则是「panic 不应该悄无声息地丢掉」：join、scope 能把 panic 传回调用方，spawn 没有调用方可传，宁可让进程退出。start_handler 和 exit_handler 也经过同一个 catch_unwind（main_loop，registry.rs 第 924–938 行），它们 panic 时同样交给 panic_handler 或 abort。

实验 E11（exp-rayon/panics/），主线程 spawn 后睡 300 ms 再打印一行：

和 tokio 对照：tokio::spawn 返回 JoinHandle，任务 panic 被运行时接住，await 这个 handle 时得到 Err(JoinError)（is_panic() 为真），没人 await 就没人知道，进程继续。rayon 的 spawn 没有句柄，默认选择了另一个极端。这在依赖链里有实际后果：任何一个第三方库里的 rayon::spawn 闭包 panic，在没设 handler 的进程里都会让整个进程退出，catch_unwind 包在外面也拦不住（panic 发生在另一个线程上）。

在 async 代码里调 rayon（E5）

要从 spawn 出去的任务里拿结果，常见写法是配一个 channel，在 async 代码里就是 tokio 的 oneshot：

let (tx, rx) = tokio::sync::oneshot::channel();
rayon::spawn(move || {
    let _ = tx.send(cpu_job());
});
let r = rx.await?;

tokio 卷的「阻塞 async」一章把它列为 CPU 密集计算的推荐写法之一。反过来，如果在 async 函数里直接调 par_iter，调用线程（tokio 的 worker）不是 rayon 的 worker，走的是 in_worker_cold：把任务注入全局池，然后在线程本地的 LockLatch 上阻塞（in_worker：从哪进池）。这段时间 tokio worker 什么也干不了。

exp-rayon/tokio-rayon/：cpu_job 是一个 400 万元素的 par_iter().map(..).reduce(..)，单独跑一次约 50 ms（12 线程）；连续算 5 次，同一个运行时上一个 1 ms 的心跳任务记录最大间隔，每种组合 3 次取中位数。心跳任务从 block_on 里 spawn（不在被测任务内部 spawn，避免它落进被阻塞 worker 的 LIFO 槽）。开始前 1 / 5 分钟负载 2.39 / 2.75、2.54 / 2.87、2.34 / 2.83（三轮），单位 ms：

几点：

• current_thread 上直接调：心跳停了 245 ms，约等于 5 次计算之和（每次约 50 ms，8 线程时约 63 ms，五次 317 ms）。5 次计算之间没有 .await，唯一的 worker 线程一直睡在 LockLatch 上。
• 另两种写法的约 11 ms：这是 rayon 的 12 个线程占满全部 12 个核，tokio 的线程（和定时器线程）被 OS 挤掉造成的，不是 rayon 阻塞了谁。反向验证：把 rayon 池缩到 8 个线程（留出 4 个核），三种不阻塞的写法都回到 2.7–2.9 ms，和底数（2.3 ms，tokio::time::sleep(1 ms) 本身的粒度）几乎相同。
• multi_thread(2) 上直接调也只有 11.5 ms：一个 worker 被阻塞，心跳在另一个 worker 上照跑。规划阶段的 survey 程序在这里量到 289 ms，是因为心跳任务是被测任务自己 spawn 的，进了那个 worker 的 LIFO 槽，worker 一阻塞它就偷不走（tokio 卷的 LIFO 槽）。同一个程序加参数 inside 复现了这一点：multi_thread(2) + 直接调 = 265 ms（三次 257–278 ms），其余写法不受影响（约 10 ms）。

所以「在 async 里直接调 par_iter」的后果取决于运行时：current_thread 上整个运行时停住；multi_thread 上只占住一个 worker，但这个 worker 本地队列和 LIFO 槽里的任务都要等它。不论哪种，rayon::spawn + oneshot 和 spawn_blocking 都不会占住 tokio worker；两者的区别是线程从哪来：spawn_blocking 在 tokio 的阻塞线程池里新开（或复用）一个线程，计算本身仍由 rayon 的池执行（cpu_job 内部的 par_iter 从阻塞线程进 rayon，走 in_worker_cold），rayon::spawn 则直接把整段计算交给 rayon 的 worker。

小结

• spawn = HeapJob（'static）+ terminate_count 加一 + inject_or_push；任务结束时 terminate() 减一。ThreadPool::drop 只减自己那一份，不等线程。
• spawn_fifo 在 worker 上经本线程的 JobFifo 入队，本线程上按 FIFO 执行；不在池里就直接注入。
• 任务 panic 交给 panic_handler；没有 handler、或 handler 自己 panic，打印「Rayon: detected unexpected panic; aborting」后 abort，退出码 134（E11）。
• 在 async 函数里直接调 par_iter 会让 tokio worker 阻塞在 LockLatch 上；rayon::spawn + oneshot 或 spawn_blocking 不会（E5）。