use super::{latch::Latch, unwind}; use std::{any::Any, cell::UnsafeCell, mem, sync::Arc}; pub(super) enum JobResult { None, Ok(T), Panic(Box), } /// A `Job` is used to advertise work for other threads that they may /// want to steal. In accordance with time honored tradition, jobs are /// arranged in a deque, so that thieves can take from the top of the /// deque while the main worker manages the bottom of the deque. This /// deque is managed by the `thread_pool` module. 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. /// /// Internally, we store the job's data in a `*const ()` pointer. The /// true type is something like `*const StackJob<...>`, but we hide /// it. We also carry the "execute fn" from the `Job` trait. pub(super) struct JobRef { pointer: *const (), execute_fn: unsafe fn(*const ()), } // SAFETY: every constructor of a job reference requires its job to be safe to execute on another // thread, which is all a reference does with its pointer. unsafe impl Send for JobRef {} // SAFETY: a shared reference only exposes the pointer's identity, never the job. unsafe impl Sync for JobRef {} impl JobRef { /// Unsafe: caller asserts that `data` will remain valid until the /// job is executed. pub(super) unsafe fn new(data: *const T) -> Self where T: Job, { // erase types: Self { pointer: data as *const (), execute_fn: ::execute, } } /// Returns an opaque handle that can be saved and compared, /// without making `JobRef` itself `Copy + Eq`. #[inline] pub(super) fn id(&self) -> impl Eq + use<> { (self.pointer, self.execute_fn) } #[inline] pub(super) unsafe fn execute(self) { // SAFETY: the caller upholds the contract of the job this reference was built from, and // consuming the reference executes that job once. unsafe { (self.execute_fn)(self.pointer) } } } /// A job that will be owned by a stack slot. This means that when it /// executes it need not free any heap data, the cleanup occurs when /// the stack frame is later popped. The function parameter indicates /// `true` if the job was stolen and executed on a different thread. pub(super) struct StackJob where L: Latch + Sync, F: FnOnce(bool) -> R + Send, R: Send, { pub(super) latch: L, func: UnsafeCell>, result: UnsafeCell>, } impl StackJob where L: Latch + Sync, F: FnOnce(bool) -> R + Send, R: Send, { pub(super) const fn new(func: F, latch: L) -> Self { Self { latch, func: UnsafeCell::new(Some(func)), result: UnsafeCell::new(JobResult::None), } } pub(super) unsafe fn as_job_ref(&self) -> JobRef { // SAFETY: the caller keeps this job in place until it has executed or been run inline. unsafe { JobRef::new(self) } } 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 Job for StackJob where L: Latch + Sync, F: FnOnce(bool) -> R + Send, R: Send, { unsafe fn execute(this: *const ()) { // SAFETY: `this` came from `as_job_ref` on a job its owner keeps in place until the // latch is set, and a job reference executes once, so nothing else touches the cells. unsafe { 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); } } } /// Represents a job stored in the heap. Used to implement /// `scope`. Unlike `StackJob`, when executed, `HeapJob` simply /// invokes a closure, which then triggers the appropriate logic to /// signal that the job executed. /// /// (Probably `StackJob` should be refactored in a similar fashion.) pub(super) struct HeapJob where BODY: FnOnce() + Send, { job: BODY, } impl HeapJob where BODY: FnOnce() + Send, { pub(super) fn new(job: BODY) -> Box { Box::new(Self { job }) } /// Creates a `JobRef` from this job. This hides all lifetimes, so /// it is up to you to ensure that this JobRef doesn't outlive any /// data that it closes over. pub(super) unsafe fn into_job_ref(self: Box) -> JobRef { // SAFETY: the box is leaked here and reclaimed by `execute`, and the caller keeps the // data the job closes over alive until then. unsafe { JobRef::new(Box::into_raw(self)) } } /// Creates a static `JobRef` from this job. pub(super) fn into_static_job_ref(self: Box) -> JobRef where BODY: 'static, { // SAFETY: a `'static` body closes over nothing that can expire. unsafe { self.into_job_ref() } } } impl Job for HeapJob where BODY: FnOnce() + Send, { unsafe fn execute(this: *const ()) { // SAFETY: `this` is the box leaked by `into_job_ref`, and a job reference executes once. let this = unsafe { Box::from_raw(this as *mut Self) }; (this.job)(); } } /// Represents a job stored in an `Arc`. Like `HeapJob`, but may be /// turned into multiple `JobRef`s and called multiple times. pub(super) struct ArcJob where BODY: Fn() + Send + Sync, { job: BODY, } impl ArcJob where BODY: Fn() + Send + Sync, { pub(super) fn new(job: BODY) -> Arc { Arc::new(Self { job }) } /// Creates a `JobRef` from this job. This hides all lifetimes, so /// it is up to you to ensure that this JobRef doesn't outlive any /// data that it closes over. pub(super) unsafe fn as_job_ref(this: &Arc) -> JobRef { // SAFETY: each reference owns one count of the `Arc`, released by `execute`, and the // caller keeps the data the job closes over alive until then. unsafe { JobRef::new(Arc::into_raw(Arc::clone(this))) } } /// Creates a static `JobRef` from this job. pub(super) fn as_static_job_ref(this: &Arc) -> JobRef where BODY: 'static, { // SAFETY: a `'static` body closes over nothing that can expire. unsafe { Self::as_job_ref(this) } } } impl Job for ArcJob where BODY: Fn() + Send + Sync, { unsafe fn execute(this: *const ()) { // SAFETY: `this` holds the count taken by `as_job_ref`, and a job reference executes once. let this = unsafe { Arc::from_raw(this as *mut Self) }; (this.job)(); } } impl JobResult { fn call(func: impl FnOnce(bool) -> T) -> Self { match unwind::halt_unwinding(|| func(true)) { Ok(x) => Self::Ok(x), Err(x) => Self::Panic(x), } } /// Convert the `JobResult` for a job that has finished (and hence /// its JobResult is populated) into its return value. /// /// NB. This will panic if the job panicked. pub(super) fn into_return_value(self) -> T { match self { Self::None => unreachable!(), Self::Ok(x) => x, Self::Panic(x) => unwind::resume_unwinding(x), } } }