broadcast

主线位置：主线 1 之外。别的 API 都是「把任务交给池，谁有空谁做」；broadcast 是「池里每个线程各做一次」，于是需要一种只有指定线程才会去取的队列，以及一种不靠计数器、直接点名叫醒的唤醒方式。

flowchart LR
  caller["broadcast(op)：N 个 StackJob + CountLatch(N)"] --> lock["锁住 broadcasts（Mutex 里每线程一个 Worker）"]
  lock --> q0["线程 0 的广播队列"]
  lock --> q1["线程 1 的广播队列"]
  lock --> qn["线程 N-1 的广播队列"]
  caller --> wake["对每个线程 notify_worker_latch_is_set"]
  q0 --> w0["线程 0：本地 deque 空了就从这里取"]
  q1 --> w1["线程 1"]
  qn --> wn["线程 N-1"]
  w0 --> cnt["CountLatch 减到 0"]
  w1 --> cnt
  wn --> cnt
  cnt --> back["调用方按线程号取回结果"]

rayon 1.6.0 / rayon-core 1.10.0 加入（eb6c6ef，合并于 #492）。发布说明的描述：「在每个线程用完自己的本地任务队列之后、尝试从别的线程偷之前，以一种降低了的优先级运行」。

每线程一个广播队列

Registry::new 第 263–269 行给每个线程额外建一个 FIFO 的 crossbeam deque：

let (broadcasts, broadcast_stealers): (Vec<_>, Vec<_>) = (0..n_threads)
    .map(|_| {
        let worker = Worker::new_fifo();
        let stealer = worker.stealer();
        (worker, stealer)
    })
    .unzip();

所有权分得很特别：Worker 端（推任务的一端）在 Registry 里，broadcasts: Mutex<Vec<Worker<JobRef>>>；Stealer 端交给对应的线程，就是 WorkerThread 里那个容易误会的 stealer 字段（注释是「the "stealer" half of the worker's broadcast deque」）。线程从自己的广播队列里「偷」，别的线程不碰它。crossbeam 的 Worker 不是 Sync（crossbeam 卷的一个坑），多个线程同时 broadcast 时要往同一个 Worker 里推，所以包了一把 Mutex。

线程什么时候取广播任务，看 take_local_job：先 worker.pop() 本地 deque，空了就从自己的广播队列 steal()，遇到 Retry 就重试。find_work 的顺序是「take_local_job → 偷别人 → 全局注入队列」，所以广播任务排在本地任务之后、窃取之前，和发布说明的描述一致。

inject_broadcast：推完逐个点名

registry.rs 第 460–489 行：

pub(super) fn inject_broadcast(&self, injected_jobs: impl ExactSizeIterator<Item = JobRef>) {
    assert_eq!(self.num_threads(), injected_jobs.len());
    {
        let broadcasts = self.broadcasts.lock().unwrap();
        // ...
        assert_eq!(broadcasts.len(), injected_jobs.len());
        for (worker, job_ref) in broadcasts.iter().zip(injected_jobs) {
            worker.push(job_ref);
        }
    }
    for i in 0..self.num_threads() {
        self.sleep.notify_worker_latch_is_set(i);
    }
}

推完之后对每个线程调 notify_worker_latch_is_set(i)，也就是 wake_specific_thread(i)：线程正睡在 Condvar 上就叫醒，醒着就什么也不做（唤醒策略与实测）。它不走 new_jobs 那套「看计数器决定叫醒几个」的逻辑，也不碰 JEC：普通任务谁来做都行，叫醒几个由计数器估算；广播任务只能由指定线程做，所以每个都点名。

broadcast：阻塞版

broadcast/mod.rs 第 97–120 行：

pub(super) unsafe fn broadcast_in<OP, R>(op: OP, registry: &Arc<Registry>) -> Vec<R>
where
    OP: Fn(BroadcastContext<'_>) -> R + Sync,
    R: Send,
{
    let f = move |injected: bool| {
        debug_assert!(injected);
        BroadcastContext::with(&op)
    };

    let n_threads = registry.num_threads();
    let current_thread = WorkerThread::current().as_ref();
    let latch = CountLatch::with_count(n_threads, current_thread);
    let jobs: Vec<_> = (0..n_threads)
        .map(|_| StackJob::new(&f, LatchRef::new(&latch)))
        .collect();
    let job_refs = jobs.iter().map(|job| job.as_job_ref());

    registry.inject_broadcast(job_refs);

    // Wait for all jobs to complete, then collect the results, maybe propagating a panic.
    latch.wait(current_thread);
    jobs.into_iter().map(|job| job.into_result()).collect()
}

• 每个线程一个 StackJob，放在调用方栈上的 Vec 里；闭包是 &f（op 只要求 Fn + Sync，被 N 个线程同时以共享引用调用，不需要 Send 也不需要 'static）。
• 所有 StackJob 共用一个计数为 N 的 CountLatch，经 LatchRef 借用（Latch 家族最后一节说了为什么不能直接存 &CountLatch）。
• current_thread 是任何池的 worker 时，CountLatch 是 Stealing，等的时候继续干活；调用线程是这个池自己的 worker 时，它自己那一份广播任务就在自己的广播队列里，wait_until 的第一步 take_local_job 会把它取出来执行，不会卡住。池外线程调用时是 Blocking，用 LockLatch 阻塞。
• 结果按线程号顺序收集。某个线程的 op panic 时，into_result 在遍历到它时 resume_unwind；所有线程都已执行完（latch 归零后才收集），后面的结果随 Vec 一起丢弃。

BroadcastContext（第 45–79 行）只有两个方法：index()（当前线程在池里的下标）和 num_threads()。它带一个 PhantomData<&'a mut dyn Fn()>，不是 Send 也不是 Sync，只能在广播闭包里用。

实验 E11 的 broadcast 模式：12 线程，3 号和 7 号线程 panic，5 次运行传播的都是 "worker3"，即线程号较小的那一个，和「按线程号顺序收集」一致。

spawn_broadcast：不等的版本

broadcast/mod.rs 第 128–150 行：一个 ArcJob，闭包里是 registry.catch_unwind(|| BroadcastContext::with(&op)) 加 registry.terminate()；每生成一个 JobRef 就 increment_terminate_count 一次。和 spawn 一样，池要等这 N 份都执行完才能终止，panic 交给 panic_handler，没有就 abort（E11：0 号线程 panic 时退出码 134；设了 handler 时 0、1 号两个线程的 panic 各调用一次 handler）。Scope / ScopeFifo 上的 spawn_broadcast 则是每份加一次 scope 的计数，scope 结束前等它们。

代价：最慢的线程决定广播什么时候结束

广播任务只能由指定线程执行，而线程只在「本地 deque 空了」时才去看广播队列，正在跑一个长任务的线程不会中途去看。实验 E15（exp-rayon/broadcast-busy/）：4 线程的池，先 spawn 一个睡 200 ms 的任务占住一个线程，10 ms 后 broadcast，记录每个线程开始执行广播闭包的时刻：

三次运行前的 1 / 5 分钟负载 2.57 / 2.86。

yield_local 调的就是 take_local_job，会取到广播队列（见 yield_now 与 yield_local）。由此也能推出一种死锁：如果某个 worker 正在阻塞等待一件「要等 broadcast 返回才会发生」的事，broadcast 永远等不到它（推测，未单独构造实验）。

一次竞态修复：睡前漏看广播队列

rayon-core 1.10.0 与 1.10.1 同一天发布（2022-11-18）。1.10.1 的发布说明只有一句：「修复了线程正要睡下时 broadcast 开始的竞态」。修复是 be12615「Fix a sleep race with broadcast jobs」（合并于 #992），改动只有一处：

+    fn has_injected_job(&self) -> bool {
+        !self.stealer.is_empty() || self.registry.has_injected_job()
+    }
 ...
-                    .no_work_found(&mut idle_state, latch, || self.registry.has_injected_job())
+                    .no_work_found(&mut idle_state, latch, || self.has_injected_job())

has_injected_jobs 是 sleep 模块在线程真正阻塞之前做的最后一次检查（睡眠与唤醒的握手）。提交说明：这个检查原来只看全局注入队列；有了 broadcast，就多了一种直接注入到每个线程的任务，线程睡前也该看自己那一份。现在的代码在 registry.rs 第 765–767 行。

按代码推一遍为什么原来会丢（以下是推理，提交说明没有展开）：广播方推完队列后的 notify_worker_latch_is_set(i) 只叫醒 is_blocked 为真的线程，不改 JEC。线程 S 正在空转、已经宣布 sleepy，还没拿自己的 is_blocked 锁：广播方此时 wake_specific_thread(S) 看到 is_blocked == false，什么也不做。S 接着拿锁、fall_asleep、看 JEC（广播不改 JEC，没变）、登记 sleeping、最后检查「有没有注入的任务」（只看全局队列，空的），然后睡下。广播任务留在 S 的广播队列里，broadcast 的调用方在 CountLatch 上永远等不到 S 那一份。

修复后，S 睡前看自己的广播队列。这次检查为什么一定能看到广播方推的任务：wake_specific_thread 要拿 S 的 is_blocked 锁。如果广播方先拿到锁（S 还没拿），那么广播方的 push 先于它放锁，S 之后拿同一把锁，push 对 S 的最后检查可见；如果 S 先拿到锁，S 一直持锁到 Condvar::wait 放锁，广播方拿到锁时 is_blocked 已经是真，会把 S 叫醒。两种顺序都不丢。修复提交还加了一个测试 broadcast_sleep_race：7 线程的池，1 秒内反复 broadcast，每个线程按下标睡几微秒，制造「有的线程刚好在睡前」的时机。

小结

• 每个线程一个 FIFO 广播队列：Worker 端在 Registry 里、用 Mutex 保护，Stealer 端给对应线程；线程在本地 deque 空了之后、偷别人之前取它。
• inject_broadcast 推完后对每个线程点名唤醒，不经过 JEC 和「叫醒几个」的估算。
• broadcast 每线程一个 StackJob，共用一个计数 N 的 CountLatch，按线程号收集结果，panic 时传播线程号最小的那个（E11）；spawn_broadcast 是一个 ArcJob 加 N 份 terminate_count，无 handler 时 panic 即 abort。
• 广播要等每个线程回到自己的队列：一个线程忙 200 ms，broadcast 就要 188–193 ms（E15）。
• 1.10.1 修了睡前最后检查漏看广播队列的竞态（be12615，#992）；推完队列再拿对方 is_blocked 锁的顺序，保证修复后的检查不丢任务。