scope 家族

pub struct Scope<'scope> {
    base: ScopeBase<'scope>,
}

pub struct ScopeFifo<'scope> {
    base: ScopeBase<'scope>,
    fifos: Vec<JobFifo>,
}

struct ScopeBase<'scope> {
    /// thread registry where `scope()` was executed or where `in_place_scope()`
    /// should spawn jobs.
    registry: Arc<Registry>,

    /// if some job panicked, the error is stored here; it will be
    /// propagated to the one who created the scope
    panic: AtomicPtr<Box<dyn Any + Send + 'static>>,

    /// latch to track job counts
    job_completed_latch: CountLatch,

    /// You can think of a scope as containing a list of closures to execute,
    /// all of which outlive `'scope`.  They're not actually required to be
    /// `Sync`, but it's still safe to let the `Scope` implement `Sync` because
    /// the closures are only *moved* across threads to be executed.
    #[allow(clippy::type_complexity)]
    marker: PhantomData<Box<dyn FnOnce(&Scope<'scope>) + Send + Sync + 'scope>>,
}

pub fn scope<'scope, OP, R>(op: OP) -> R
where
    OP: FnOnce(&Scope<'scope>) -> R + Send,
    R: Send,
{
    in_worker(|owner_thread, _| {
        let scope = Scope::<'scope>::new(Some(owner_thread), None);
        scope.base.complete(Some(owner_thread), || op(&scope))
    })
}

pub fn in_place_scope<'scope, OP, R>(op: OP) -> R
where
    OP: FnOnce(&Scope<'scope>) -> R,
{
    do_in_place_scope(None, op)
}

pub(crate) fn do_in_place_scope<'scope, OP, R>(registry: Option<&Arc<Registry>>, op: OP) -> R
where
    OP: FnOnce(&Scope<'scope>) -> R,
{
    let (thread, registry) = get_in_place_thread_registry(registry);
    let scope = Scope::<'scope>::new(thread, registry);
    scope.base.complete(thread, || op(&scope))
}

fn get_in_place_thread_registry(
    registry: Option<&Arc<Registry>>,
) -> (Option<&WorkerThread>, Option<&Arc<Registry>>) {
    let thread = unsafe { WorkerThread::current().as_ref() };
    if thread.is_none() && registry.is_none() {
        // A new global registry may use the current thread, especially on WebAssembly,
        // so we have to re-check our current status after it's built.
        let global = global_registry();
        (global.current_thread(), Some(global))
    } else {
        (thread, registry)
    }
}

fn complete<FUNC, R>(&self, owner: Option<&WorkerThread>, func: FUNC) -> R
where
    FUNC: FnOnce() -> R,
{
    let result = unsafe { Self::execute_job_closure(self, func) };
    self.job_completed_latch.wait(owner);
    self.maybe_propagate_panic();
    result.unwrap() // only None if `op` panicked, and that would have been propagated
}

unsafe fn execute_job_closure<FUNC, R>(this: *const Self, func: FUNC) -> Option<R>
where
    FUNC: FnOnce() -> R,
{
    let result = match unwind::halt_unwinding(func) {
        Ok(r) => Some(r),
        Err(err) => {
            (*this).job_panicked(err);
            None
        }
    };
    Latch::set(&(*this).job_completed_latch);
    result
}

pub fn spawn<BODY>(&self, body: BODY)
where
    BODY: FnOnce(&Scope<'scope>) + Send + 'scope,
{
    let scope_ptr = ScopePtr(self);
    let job = HeapJob::new(move || unsafe {
        // SAFETY: this job will execute before the scope ends.
        let scope = scope_ptr.as_ref();
        ScopeBase::execute_job(&scope.base, move || body(scope))
    });
    let job_ref = self.base.heap_job_ref(job);

    // Since `Scope` implements `Sync`, we can't be sure that we're still in a
    // thread of this pool, so we can't just push to the local worker thread.
    // Also, this might be an in-place scope.
    self.base.registry.inject_or_push(job_ref);
}

fn bad_scope<F>(f: F)
    where F: FnOnce(&i32) + Send,
{
    rayon_core::scope(|s| {
        let x = 22;
        s.spawn(|_| f(&x)); //~ ERROR `x` does not live long enough
    });
}

fn good_scope<F>(f: F)
    where F: FnOnce(&i32) + Send,
{
    let x = 22;
    rayon_core::scope(|s| {
        s.spawn(|_| f(&x));
    });
}

fn job_panicked(&self, err: Box<dyn Any + Send + 'static>) {
    // capture the first error we see, free the rest
    if self.panic.load(Ordering::Relaxed).is_null() {
        let nil = ptr::null_mut();
        let mut err = ManuallyDrop::new(Box::new(err)); // box up the fat ptr
        let err_ptr: *mut Box<dyn Any + Send + 'static> = &mut **err;
        if self
            .panic
            .compare_exchange(nil, err_ptr, Ordering::Release, Ordering::Relaxed)
            .is_ok()
        {
            // ownership now transferred into self.panic
        } else {
            // another panic raced in ahead of us, so drop ours
            let _: Box<Box<_>> = ManuallyDrop::into_inner(err);
        }
    }
}

fn maybe_propagate_panic(&self) {
    // propagate panic, if any occurred; at this point, all
    // outstanding jobs have completed, so we can use a relaxed
    // ordering:
    let panic = self.panic.swap(ptr::null_mut(), Ordering::Relaxed);
    if !panic.is_null() {
        let value = unsafe { Box::from_raw(panic) };
        unwind::resume_unwinding(*value);
    }
}