in_worker：从哪进池

主线位置：主线 1 的第一步。join、scope、install 都先过这里：当前线程能不能直接干活，还是要把活交给池子、自己等。

两个 in_worker

rayon-core 里有两个同名函数。

全局函数 registry::in_worker，join_context、scope、scope_fifo 用它：

pub(super) fn in_worker<OP, R>(op: OP) -> R
where
    OP: FnOnce(&WorkerThread, bool) -> R + Send,
    R: Send,
{
    unsafe {
        let owner_thread = WorkerThread::current();
        if !owner_thread.is_null() {
            // Perfectly valid to give them a `&T`: this is the
            // current thread, so we know the data structure won't be
            // invalidated until we return.
            op(&*owner_thread, false)
        } else {
            global_registry().in_worker(op)
        }
    }
}

只要当前线程是任意一个池的 worker，就在当前线程上直接调用 op；否则交给全局池。它不问「是不是全局池的 worker」。

方法 Registry::in_worker，ThreadPool::install 用它（self.registry.in_worker(|_, _| op())，thread_pool/mod.rs:137-143），全局函数在池外时也落到它：

    pub(super) fn in_worker<OP, R>(&self, op: OP) -> R
    where
        OP: FnOnce(&WorkerThread, bool) -> R + Send,
        R: Send,
    {
        unsafe {
            let worker_thread = WorkerThread::current();
            if worker_thread.is_null() {
                self.in_worker_cold(op)
            } else if (*worker_thread).registry().id() != self.id() {
                self.in_worker_cross(&*worker_thread, op)
            } else {
                // Perfectly valid to give them a `&T`: this is the
                // current thread, so we know the data structure won't be
                // invalidated until we return.
                op(&*worker_thread, false)
            }
        }
    }

三条路径：

flowchart TB
  start["in_worker(op)"] --> q{"WorkerThread::current()"}
  q -->|"本池的 worker"| direct["① 直接 op(worker, false)"]
  q -->|"null：不在任何池"| cold["② in_worker_cold：注入，LockLatch 上 Condvar 阻塞"]
  q -->|"别的池的 worker"| cross["③ in_worker_cross：注入，SpinLatch::cross，wait_until 边等边干本池的活"]

第二个参数 bool 就是 join_context 拿到的 injected：直接调用时为 false，经 ② ③ 注入时为 true。

① 本池 worker：直接调用

没有任何开销，只是一次线程本地读和一次 registry 地址比较。install 嵌套在同一个池里、join 递归，都走这里。

② 不在任何池：注入后阻塞

registry.rs:516-540：

    #[cold]
    unsafe fn in_worker_cold<OP, R>(&self, op: OP) -> R
    where
        OP: FnOnce(&WorkerThread, bool) -> R + Send,
        R: Send,
    {
        thread_local!(static LOCK_LATCH: LockLatch = const { LockLatch::new() });

        LOCK_LATCH.with(|l| {
            // This thread isn't a member of *any* thread pool, so just block.
            debug_assert!(WorkerThread::current().is_null());
            let job = StackJob::new(
                |injected| {
                    let worker_thread = WorkerThread::current();
                    assert!(injected && !worker_thread.is_null());
                    op(&*worker_thread, true)
                },
                LatchRef::new(l),
            );
            self.inject(job.as_job_ref());
            job.latch.wait_and_reset(); // Make sure we can use the same latch again next time.

            job.into_result()
        })
    }

• latch 用线程本地的一个 LockLatch，const 初始化，每次用完 wait_and_reset 复位，下次复用；不用每次新建 Mutex + Condvar。StackJob 里放的是 LatchRef（借用而不是拥有，Latch 家族）。
• inject（registry.rs:428-444）：记下注入队列原来是不是空的，injected_jobs.push，sleep.new_injected_jobs(1, queue_was_empty)——有线程睡着就叫醒。
• 阻塞：LockLatch::wait_and_reset 是标准的 Mutex<bool> + Condvar 等待。调用线程真的睡在 OS 的条件变量上，不参与计算。
• op 在 worker 上执行：哪个 worker 从注入队列取到它（按找活顺序，注入队列排在最后），就在那个 worker 上调用 op(worker, true)；完成后 LockLatch::set 加锁置 true、notify_all。

这就是 pytorch-to-candle 卷 采样里那个栈：217 个样本里 205 个在 in_worker_cold，栈顶是 __psynch_cvwait（macOS 上 Condvar::wait 的底层）。主线程不是 rayon worker，每次 softmax 的 par_chunks 都走这条路径：注入、叫醒 worker、在 Condvar 上睡、被 notify_all 叫醒。

一次进池的往返至少包括：注入队列的 push、一次可能的唤醒（Condvar::notify_one）、worker 从睡眠或空转中找到这个任务、执行、LockLatch::set（加锁、notify_all）、调用线程从 Condvar::wait 返回。下面的 E13 量这个固定开销。

③ 别的池的 worker：注入后边等边干活

registry.rs:542-563：

    #[cold]
    unsafe fn in_worker_cross<OP, R>(&self, current_thread: &WorkerThread, op: OP) -> R
    where
        OP: FnOnce(&WorkerThread, bool) -> R + Send,
        R: Send,
    {
        // This thread is a member of a different pool, so let it process
        // other work while waiting for this `op` to complete.
        debug_assert!(current_thread.registry().id() != self.id());
        let latch = SpinLatch::cross(current_thread);
        let job = StackJob::new(
            |injected| {
                let worker_thread = WorkerThread::current();
                assert!(injected && !worker_thread.is_null());
                op(&*worker_thread, true)
            },
            latch,
        );
        self.inject(job.as_job_ref());
        current_thread.wait_until(&job.latch);
        job.into_result()
    }

当前线程 W 是池 A 的 worker，要把 op 交给池 B：

• 注入池 B 的注入队列，由 B 的某个 worker 执行。
• W 不阻塞在 OS 上，而是 wait_until(&job.latch)，即在池 A 里继续找活：A 的本地 deque、A 的其他线程、A 的注入队列。W 可能睡下，但睡在 A 的 sleep 模块里。
• latch 是 SpinLatch::cross(W)：它记着 W 的 registry（池 A）和 W 的编号。B 的 worker 执行完 op、set latch 时，若 W 正睡着，就去池 A 的 sleep 模块里叫醒 W。cross 标志让 set 先 Arc::clone 一份池 A 的 registry 再置位（latch.rs:194-222）：置位之后 W 可能立刻返回、池 A 可能随之被 drop，而 set 方还要用它去 notify。这是 cbd484c「Keep the registry alive for a cross-pool SpinLatch」（2020-03，随 #746 合并）修的；tests/cross-pool.rs 的注释描述的就是这个场景。

历史。 这条路径是 2017 年 36577d6「Solve inter-pool deadlocks」（#449）加的。从它的 diff 看，之前 in_worker 只分「本池 worker」和「其他」两种，别的池的 worker 也走 in_worker_cold，在 LockLatch 上阻塞；提交说明写的是改成「the current thread can go process other work from its own pool, possibly going to sleep when idle, and still be notified when the other pool finishes the new work」。旧行为具体在什么情况下死锁，提交说明没写；一个可以推出来的情形是：池 A 的 worker 全部阻塞等池 B，而 B 的任务又要回到 A 里执行（推测）。

代价：install 的「execution order」

install 的文档专门有一节「Warning: execution order」（thread_pool/mod.rs:78-113）：全局池只有 1 个线程，rayon::join(|| do_it(), || do_it())，do_it 先打印 one、再 pool.install(|| {})、再打印 two。直觉上全局池只有一个线程，应该是 one two one two；文档说 one one two two 也同样合法。

实测：三条路径与执行顺序（E6）

exp-rayon/cross-pool：全局池 build_global 成 1 个线程，另建一个 2 线程的池 pool。

main: tid=ThreadId(1) index=None
cold (main -> pool): tid=ThreadId(3) index=Some(0) threads=2
same pool nested install: same thread=true
cross (global worker -> pool): same thread=false index=Some(0) threads=2
back on global worker: threads=1
cross_install=true: {"one one two two": 200}
cross_install=false: {"one two one two": 200}

• ② 从 main install：op 跑在 pool 的一个 worker 上（另一个线程），current_num_threads() 是 2。
• ① 在 pool 里再 install 到同一个池：同一个线程，直接调用。
• ③ 在全局池的 worker 上 install 到 pool：op 跑在 pool 的线程上；回到全局池 worker 后 current_num_threads() 又是 1。
• 文档的例子重复 200 次，200 次全是 one one two two，不是「可能出现」，而是在这个配置下必然：唯一的全局 worker 跑 A 的 do_it，打印 one，install 进入 ③，wait_until 时本地 deque 里正好有 B（另一个 do_it），于是在同一个栈上执行它，打印第二个 one；B 的 install 再次进入 ③，等到 pool 做完后打印 two，返回，A 的 install 也早已完成，打印 two。
• 反向验证：把 do_it 里的 pool.install(|| {}) 换成什么都不做，其他不变，200 次全是 one two one two。顺序的变化完全来自 ③ 的「等的时候干本池的活」。两次运行结果相同。

同样的机制让 TLS 和「这段代码不会和自己交错」的假设失效，对应 09 常见坑 里的几条。

「当前池」是怎么定的

把上面的规则合起来：

全局函数没有「池」参数，靠调用线程所在的池。所以 pool.install(|| data.par_iter()...) 能把迭代器送进 pool：install 把闭包送到 pool 的 worker 上，闭包里的 par_iter 再调全局 in_worker，发现自己在 pool 的 worker 上，于是整棵 join 树都在 pool 里展开。candle 的 Device::with_context 就是 candle_pool().install(f)，gemm 里的 into_par_iter 因此进 candle 的私有池（candle 卷 CPU 后端）；tenferro 把闭包 install 进自己建的私有池，同理（tenferro 卷 CPU 引擎）。反过来，在 install 外面调用的 par_iter，哪怕在同一个函数里，也会去全局池。

实测：进池的固定开销（E13）

exp-rayon/outside-call：先 rayon::broadcast(|_| ()) 让所有线程起来，再另建一个默认大小的 pool 并 broadcast 一次。每项循环 2 万次取平均，每次运行 5 轮。测的项目：

每次运行前要求 1 分钟负载 < 3（实测开始时 2.78–2.98），12 线程和 RAYON_NUM_THREADS=1 各跑 3 次，每次取 5 轮的中位数；表里是 3 次的中位数，括号里是 3 次各自的值。

读法：

• 池外比池内贵三个数量级。 12 线程时从 main 调一次空 join 约 17 µs，池内约 30 ns，差 500 多倍。install_main 和 join_main 几乎一样，说明开销在「进池」本身，不在 join。规划阶段在负载 3–4 时测得的 14–18 µs、23–27 ns 与此一致。
• 16 个元素的 sum：串行 1 ns，从 main 并行 21 µs，慢两万倍。candle 卷里 9 行 softmax「走 rayon 要几十微秒」就是这个量级（candle 卷 quantized-llama），采样里看到的 in_worker_cold / __psynch_cvwait 正是这段时间。
• 池内调用也不是免费的。 par16_inside 在 12 线程时是 1.6 µs，1 线程时 15 ns。池内 16 个元素会切成至少 16 片、15 次 join；每次 push 都经 new_internal_jobs 考虑叫醒睡着的线程，被叫醒的线程来偷、偷到后跨线程 set latch。1 线程时这些全没有，15 次 join 每次约 1 ns 量级的本地 push / pop。
• 反向验证：1 线程。 只有一个 worker 时，池内 join 12 ns，和 join 一章 E1 实测的「每次 join 约 6.6 ns」同量级，说明 join 本身的固定开销确实只有十几纳秒。从 main 进池仍要 3.8 µs：注入、唤醒唯一的 worker、Condvar 往返一样不少。
• 12 线程时进池为什么比 1 线程贵 4 倍多（17 µs 对 3.8 µs），本实验分辨不出来。可能的来源有：注入队列在找活顺序里排最后，空闲线程要先轮询其他 11 个线程的 deque；12 个线程在两次调用之间各自处在空转或睡眠的不同阶段，唤醒和抢同一个注入任务的竞争更多。以上是推测，需要 sleep 模块的实测（E3，05 sleep 模块）配合才能拆开。

结论只有一条：一次并行调用的活要远大于这几十微秒，从池外调用才划算；已经在池里时，门槛低得多，但 12 线程下也有微秒级。

在 async 任务里直接调用

tokio 的 worker 线程不是 rayon worker，在 async 任务里直接 par_iter() 走的就是 ②：tokio worker 阻塞在 LockLatch 的 Condvar 上，直到整个并行计算完成，期间这个 tokio worker 上的其他任务全部停摆。实验 E5 在 current_thread 运行时上，5 次各 59 ms 的计算让一个 1 ms 心跳任务的最大间隔达到 245 ms，spawn_blocking 或 rayon::spawn + oneshot 时约 11 ms（数字见 spawn）。改法和 tokio 一侧的分析见 tokio 卷 阻塞在 async 里，坑的完整复现在 常见坑：在 async 里直接调 par_iter。

小结

• 全局 in_worker：在任何池的 worker 上就直接调用，否则进全局池。Registry::in_worker：本池直接调用、池外注入后在 LockLatch 上阻塞、别的池注入后 wait_until 边等边干本池的活。
• ② 冷路径复用线程本地的 LockLatch，调用线程睡在 Condvar 上；pytorch-to-candle 卷采样里的 in_worker_cold / __psynch_cvwait 就是这里。
• ③ 跨池路径用 SpinLatch::cross，set 前先 Arc::clone 等待方的 registry；等待时执行本池任务，文档例子在 1 线程全局池上 200 次全是 one one two two，去掉 install 后 200 次全是 one two one two（E6）。
• 「当前池」= 调用线程所在的池；install 能把内部的 par_iter 带进私有池，靠的就是这一条。
• 从池外进池一次是微秒级，池内是纳秒级（E13），这是小任务不该从池外并行的根本原因。