JobRef 与几种 Job

主线位置：一次 join。join_context 在自己的栈帧上建一个 StackJob，as_job_ref() 把它变成 16 字节的 JobRef 推进本线程 deque；之后不论是自己弹回来还是被别的线程偷走，拿到它的一方都只做一件事：(execute_fn)(pointer)。

flowchart TB
  jr["JobRef：pointer: *const ()，execute_fn: unsafe fn(*const ())"]
  jr --> sj["StackJob：在发起方栈上，发起方等 latch 后读 result"]
  jr --> hj["HeapJob：Box，execute 时 Box::from_raw 回收"]
  jr --> aj["ArcJob：Arc，每个 JobRef 持一个强引用"]
  jr --> jf["JobFifo：pointer 指向 FIFO 本身，execute 时取 FIFO 队头"]
  sj --- u1["join、install、in_worker_*、broadcast"]
  hj --- u2["spawn、spawn_fifo、scope.spawn"]
  aj --- u3["spawn_broadcast、scope.spawn_broadcast"]
  jf --- u4["worker 上的 spawn_fifo、scope_fifo"]

整个 job.rs 270 行，四种 Job 加一个 JobResult。它是 crossbeam-deque 和 rayon 其余部分的接口：deque 只认 JobRef（Worker<JobRef>、Injector<JobRef>），rayon 其余部分只认 Job trait。

Job 与 JobRef：手写的胖指针

job.rs 第 20–66 行：

pub(super) trait Job {
    /// Unsafe: this may be called from a different thread than the one
    /// which scheduled the job, so the implementer must ensure the
    /// appropriate traits are met, whether `Send`, `Sync`, or both.
    unsafe fn execute(this: *const ());
}

/// Effectively a Job trait object. Each JobRef **must** be executed
/// exactly once, or else data may leak.
pub(super) struct JobRef {
    pointer: *const (),
    execute_fn: unsafe fn(*const ()),
}

unsafe impl Send for JobRef {}
unsafe impl Sync for JobRef {}

impl JobRef {
    pub(super) unsafe fn new<T>(data: *const T) -> JobRef
    where
        T: Job,
    {
        // erase types:
        JobRef {
            pointer: data as *const (),
            execute_fn: <T as Job>::execute,
        }
    }

    pub(super) fn id(&self) -> impl Eq {
        (self.pointer, self.execute_fn)
    }

    pub(super) unsafe fn execute(self) {
        (self.execute_fn)(self.pointer)
    }
}

JobRef 就是 &dyn Job 拆开来自己存：一个数据指针，一个函数指针，64 位平台上 16 字节。和真正的 trait object 比，有三点不同：

• 存的是函数指针本身，不是 vtable 指针。trait object 调方法要先从 vtable 里取出函数指针再调用；JobRef 直接 (self.execute_fn)(self.pointer)。Job 只有一个方法，也用不着 vtable 里的 size、align、drop。
• 类型和生命周期一起擦掉了。pointer 是 *const ()，StackJob<SpinLatch<'r>, F, R> 里的 'r 和闭包 F 借用的栈上数据都不在类型里了。deque 是 Worker<JobRef>，一种元素类型装下所有任务。代价是正确性全靠调用方：JobRef::new 是 unsafe fn，注释写的是「调用方断言 data 在任务执行前一直有效」。
• execute 收 self（按值），JobRef 没有实现 Clone / Copy。注释里加粗的那句「每个 JobRef 必须恰好执行一次」：执行零次，HeapJob 的 Box 泄漏、StackJob 的等待方永远等不到 latch；执行两次，HeapJob 被释放两次。类型上用「不能复制」挡住了一半（同一个 JobRef 不会被执行两次），另一半（每个 JobRef 一定会被执行）靠线程池的终止协议：池在所有 spawn 出去的任务跑完前不会终止（terminate_count，见 spawn 与 spawn_fifo）。

16 字节这个大小在 crossbeam 卷里出现过：deque 扩容时，新 buffer 不小于 1 KiB 就立即 flush epoch，JobRef 16 字节，第一次扩容到 128 个槽就是 2 KiB，所以 rayon 的每一次 deque 扩容都会 flush。

id()：只给 join 用的身份

join 推出 B 之后要从本地 deque 里把它认出来（「弹出的是不是我自己的 B」），所以要比较两个 JobRef。id() 返回一个不透明的 impl Eq，内容是 (pointer, execute_fn)。

JobRef 原来派生了 Copy, Clone, Debug, PartialEq, Eq，2023-02 的 b6cc378「Stop deriving traits on JobRef」（合并于 #1025）把它们全部去掉，改成在 StackJob 上实现 PartialEq<JobRef>；第二天的 58f7b7e（合并于 #1028）又换成现在的 id()。提交说明写的理由：clippy 的 fn_address_comparisons 指出，同一个函数在不同地方取地址，不保证得到同一个指针；先从推进 deque 的那个 JobRef 里取出 id 保存，之后拿弹出来的 JobRef 的 id 去比，比的就是同一次取到的函数指针。提交说明也提到，不同函数被合并到同一个地址的情况没那么要紧，因为两个 StackJob 的数据指针不同。

StackJob：住在发起方栈上

job.rs 第 72–124 行：

pub(super) struct StackJob<L, F, R>
where
    L: Latch + Sync,
    F: FnOnce(bool) -> R + Send,
    R: Send,
{
    pub(super) latch: L,
    func: UnsafeCell<Option<F>>,
    result: UnsafeCell<JobResult<R>>,
}

impl<L, F, R> StackJob<L, F, R> /* where ... */ {
    pub(super) unsafe fn run_inline(self, stolen: bool) -> R {
        self.func.into_inner().unwrap()(stolen)
    }

    pub(super) unsafe fn into_result(self) -> R {
        self.result.into_inner().into_return_value()
    }
}

impl<L, F, R> Job for StackJob<L, F, R> /* where ... */ {
    unsafe fn execute(this: *const ()) {
        let this = &*(this as *const Self);
        let abort = unwind::AbortIfPanic;
        let func = (*this.func.get()).take().unwrap();
        (*this.result.get()) = JobResult::call(func);
        Latch::set(&this.latch);
        mem::forget(abort);
    }
}

一个 StackJob 有两种结局：

• 被 JobRef 执行（被偷走，或者被本线程在别处弹出）：execute 从 func 里 take 出闭包，用 JobResult::call 包着跑，结果写进 result，然后 Latch::set。发起方看到 latch 被设置后调 into_result 取结果。
• 被发起方自己按值消费：join 发现弹回来的正是自己的 B，就调 run_inline(self, ..) 直接跑闭包。这条路不经过 result 也不碰 latch，B 的 panic 直接在当前栈上展开。

execute 里的顺序是本卷「设置之后不能再碰」那条横线的第一个例子：结果写进 result 在 Latch::set 之前；set 之后发起方可能立刻醒来、读完结果、函数返回，StackJob 所在的栈帧随之消失，所以 set 之后 execute 不再碰 this 的任何字段（最后那句 mem::forget(abort) 动的是局部变量）。latch 一侧怎么配合，见 Latch 家族。

AbortIfPanic 是 unwind.rs 里的一个守卫：它的 Drop 打印「Rayon: detected unexpected panic; aborting」然后 process::abort()。正常路径上用 mem::forget 把它丢掉，Drop 不会运行；只有 execute 中途展开（rayon 自己的代码出了 bug，用户闭包的 panic 已经被 JobResult::call 接住了）才会 abort。原因写在 wait_until_cold 的注释里：如果展开继续往上走，别处的代码会以为 latch 已经设置，接着访问已经无效的内存。

StackJob 的使用者：join_context 的 B（latch 是 SpinLatch），in_worker_cold（LatchRef<LockLatch>），in_worker_cross（SpinLatch::cross），broadcast（每个线程一个，LatchRef<CountLatch>）。共同点是发起方一定会等：不等到 latch 就不返回，所以栈帧一定比任务活得久，unsafe 的前提成立。

JobResult 三态

job.rs 第 9–13、216–235 行：

pub(super) enum JobResult<T> {
    None,
    Ok(T),
    Panic(Box<dyn Any + Send>),
}

impl<T> JobResult<T> {
    fn call(func: impl FnOnce(bool) -> T) -> Self {
        match unwind::halt_unwinding(|| func(true)) {
            Ok(x) => JobResult::Ok(x),
            Err(x) => JobResult::Panic(x),
        }
    }

    pub(super) fn into_return_value(self) -> T {
        match self {
            JobResult::None => unreachable!(),
            JobResult::Ok(x) => x,
            JobResult::Panic(x) => unwind::resume_unwinding(x),
        }
    }
}

halt_unwinding 就是 catch_unwind(AssertUnwindSafe(func))。别的线程上的 panic 被装进 JobResult::Panic，等发起方 into_result 时再 resume_unwind，于是 panic 看起来像是在发起方线程上发生的。None 只在任务还没执行时存在，into_return_value 遇到它就是 rayon 自己的 bug，所以是 unreachable!()。

migrated 参数：「经过了 JobRef」而不是「换了线程」

注意 call 里那个写死的 func(true)：只要是经 execute 跑的，闭包收到的 stolen（在 join_context 里变成 FnContext::migrated()）就是 true。而 run_inline(stolen) 收到的是调用方传进来的值，join_context 传的是 injected（整个 join 是不是从池外注入进来的）。

所以 migrated() 的真实含义是「这个闭包没有在 join_context 的原栈帧上被直接调用」，和「是否换了线程」并不完全一致。实验 E14（exp-rayon/join-context/）在 RAYON_NUM_THREADS=1 下跑三种情况，所有闭包都在 0 号线程上执行（没有别的线程可以偷）：

三次运行结果完全相同。4 线程下第三种情况也一样（B 在同一个线程上，migrated 为 true）。FnContext::migrated 的文档说「闭包在一个与提供它的线程不同的线程上被调用时返回 true」，实际是这个条件的超集。rayon 自己只拿它做一件事：并行迭代器的切分器在 migrated 为真时把切分预算重置为线程数（见 bridge 与自适应切分），多报一次 true 只是多切几刀，不影响正确性。

HeapJob：spawn 出去就不管了

job.rs 第 132–171 行：

pub(super) struct HeapJob<BODY>
where
    BODY: FnOnce() + Send,
{
    job: BODY,
}

impl<BODY> HeapJob<BODY> where BODY: FnOnce() + Send {
    pub(super) fn new(job: BODY) -> Box<Self> {
        Box::new(HeapJob { job })
    }

    pub(super) unsafe fn into_job_ref(self: Box<Self>) -> JobRef {
        JobRef::new(Box::into_raw(self))
    }

    pub(super) fn into_static_job_ref(self: Box<Self>) -> JobRef
    where
        BODY: 'static,
    {
        unsafe { self.into_job_ref() }
    }
}

impl<BODY> Job for HeapJob<BODY> where BODY: FnOnce() + Send {
    unsafe fn execute(this: *const ()) {
        let this = Box::from_raw(this as *mut Self);
        (this.job)();
    }
}

HeapJob 没有 latch 也没有 result。execute 用 Box::from_raw 把所有权拿回来，闭包跑完 Box 就释放了。「做完了」的信号由闭包自己负责：spawn 的闭包里包着 registry.catch_unwind(func) 和 registry.terminate()，scope.spawn 的闭包里包着 ScopeBase::execute_job，后者在最后 Latch::set(&scope.job_completed_latch)。

两个构造函数的区别就是生命周期：

• into_static_job_ref 要求 BODY: 'static，所以是安全函数。spawn / spawn_fifo 用它。
• into_job_ref 是 unsafe，允许闭包借用栈上数据。只有 scope 用它，安全性由「scope 在所有任务完成前不返回」保证（ScopeBase::heap_job_ref 先给 CountLatch 加一，再转成 JobRef）。

ArcJob：同一个任务，每个线程一份 JobRef

job.rs 第 175–214 行：

impl<BODY> ArcJob<BODY> where BODY: Fn() + Send + Sync {
    pub(super) unsafe fn as_job_ref(this: &Arc<Self>) -> JobRef {
        JobRef::new(Arc::into_raw(Arc::clone(this)))
    }
}

impl<BODY> Job for ArcJob<BODY> where BODY: Fn() + Send + Sync {
    unsafe fn execute(this: *const ()) {
        let this = Arc::from_raw(this as *mut Self);
        (this.job)();
    }
}

spawn_broadcast 要在 N 个线程上各跑一次同一个闭包，又不等它们。每生成一个 JobRef 就 Arc::clone 一次，引用计数加一；每次执行 Arc::from_raw 拿回一个强引用，跑完 drop。N 个 JobRef 都执行完，计数归零，闭包被释放。所以闭包是 Fn（会被调用多次）且 Sync（多个线程同时调用）。

阻塞的 broadcast 不用 ArcJob：它要等，于是在自己栈上建一个 Vec<StackJob>，每个线程一个，结果各自放在自己的 StackJob 里（见 broadcast）。

JobFifo：在 LIFO 的 deque 上做出 FIFO

worker 的本地 deque 默认是 LIFO：本线程从栈顶弹，窃取者从栈底偷（crossbeam 卷的 Chase-Lev 一章讲过两种 flavor）。spawn_fifo 要求「本线程上的执行顺序是 FIFO」，rayon 没有为它另开一个 deque，而是加了一层间接（job.rs 第 237–270 行）：

pub(super) struct JobFifo {
    inner: Injector<JobRef>,
}

impl JobFifo {
    pub(super) unsafe fn push(&self, job_ref: JobRef) -> JobRef {
        // A little indirection ensures that spawns are always prioritized in FIFO order.  The
        // jobs in a thread's deque may be popped from the back (LIFO) or stolen from the front
        // (FIFO), but either way they will end up popping from the front of this queue.
        self.inner.push(job_ref);
        JobRef::new(self)
    }
}

impl Job for JobFifo {
    unsafe fn execute(this: *const ()) {
        // We "execute" a queue by executing its first job, FIFO.
        let this = &*(this as *const Self);
        loop {
            match this.inner.steal() {
                Steal::Success(job_ref) => break job_ref.execute(),
                Steal::Empty => panic!("FIFO is empty"),
                Steal::Retry => {}
            }
        }
    }
}

调用点是 WorkerThread::push_fifo：self.push(self.fifo.push(job))。真正的任务进 FIFO（一个 crossbeam Injector），deque 里放的是一个指向 FIFO 本身的 JobRef。push 三次就有三个一模一样的 JobRef(&fifo, JobFifo::execute) 在 deque 里。不管 deque 里的哪一个先被谁拿到（本线程从栈顶弹，或者别的线程从栈底偷），执行的都是「FIFO 的队头」。deque 里的条目数和 FIFO 里的任务数始终相等，所以 "FIFO is empty" 不应该出现。

这样做的效果是：本线程弹出顺序是 FIFO；被别的线程偷走的那个条目，执行的也是 FIFO 队头，整体上 FIFO 队列里的任务按入队顺序开始。代价是每个 spawn_fifo 多一次 Injector 的 push 和 steal。Injector 是多生产者多消费者的，所以窃取者在别的线程上 steal 这个 FIFO 没问题。这是 crossbeam 卷 Injector 一章没提到的第二处用法：每个 WorkerThread 一个（fifo 字段），每个 ScopeFifo 再按线程数各建一个（fifos: Vec<JobFifo>，scope/mod.rs 第 562–567 行）。ScopeFifo 自带一组，是为了让不同 scope 的 FIFO 任务互不插队。

这层间接来自 8ccc028「Add spawn_fifo」（2018-12，合并于 #615），rayon 1.1.0 / rayon-core 1.5.0 发布。

JobFifo 的 JobRef 指针指向 WorkerThread 里的 fifo 字段，而 WorkerThread 在 worker 线程主循环的栈上（main_loop 第一行 let worker_thread = &WorkerThread::from(thread);）。别的线程偷到它，访问的是另一个线程栈上的数据；这个栈在线程退出前一直有效，而线程要等 terminate latch 被设置才退出，spawn_fifo 的任务又各持一份 terminate_count，所以任务都执行完之前栈不会消失。

四种 Job 对照

HeapJob 上的一行注释（「Probably StackJob should be refactored in a similar fashion.」）说明作者也觉得 StackJob 把 latch 和 result 内置的做法不如 HeapJob 灵活，目前没有改。

小结

• JobRef = 数据指针 + execute 函数指针，16 字节，不可复制；类型和生命周期全部擦掉，deque 只装一种元素。安全性靠「每个 JobRef 恰好执行一次」和「数据在执行前一直有效」两条约定，前者部分由「不可复制」保证，后者由各调用方保证。
• StackJob 在发起方栈上，先写 result 再 set latch，set 之后不碰自己；发起方自己弹回 B 时走 run_inline，不经过 latch。
• HeapJob / ArcJob 执行时回收内存，没有结果字段；'static 版本是安全函数，借用版本只有 scope 用。
• JobFifo 用「deque 里放指向 FIFO 的引用，执行时取 FIFO 队头」在 LIFO deque 上实现了 FIFO 的执行顺序。
• 经 execute 跑的闭包一律收到 migrated = true，即使在同一个线程上（E14）；从池外调 join_context 时 A、B 也都是 true。