dope_core/platform/thread/
raw.rs

1use std::{mem, ptr};
2
3use dope_sys::thread;
4
5use crate::platform::thread::{Finished, Slot, Task};
6
7pub(super) struct Group<T, O, P> {
8    pub(super) slots: Vec<Slot<T, O, P>>,
9    /// Successful spawn prefix; this owner retains every remaining input.
10    pub(super) started: usize,
11}
12
13impl<T, O, P> Drop for Group<T, O, P> {
14    fn drop(&mut self) {
15        self.take();
16    }
17}
18
19impl<T, O, P> Group<T, O, P> {
20    pub(super) fn parents_mut(&mut self) -> impl ExactSizeIterator<Item = &mut P> {
21        let slots = self.slots.as_mut_ptr();
22        (0..self.started).map(move |index| {
23            // SAFETY: the started prefix lies within this fixed allocation.
24            // Native entries access only task/output cells; this projection
25            // lends only disjoint coordinator metadata for the Group borrow.
26            unsafe {
27                let slot = slots.add(index);
28                &mut *ptr::addr_of_mut!((*slot).parent)
29            }
30        })
31    }
32
33    /// Joins every worker, drops unstarted inputs, and transfers the output prefix.
34    pub(super) fn take(&mut self) -> Finished<T, O, P> {
35        // SAFETY: successful spawns alone extend started within this fixed
36        // allocation; Workers cannot grow or remove slots before this take.
37        for slot in unsafe { self.slots.get_unchecked(..self.started) } {
38            // SAFETY: start stores each handle before extending this prefix.
39            // This take consumes each handle and then clears the prefix before
40            // any later drop; Handle retains thread confinement.
41            let handle = unsafe {
42                slot.thread
43                    .replace(mem::MaybeUninit::uninit())
44                    .assume_init()
45            };
46            handle.join();
47        }
48        let len = self.slots.len();
49        let slots = self.slots.as_mut_ptr();
50        // SAFETY: started fits this fixed allocation. The detached suffix
51        // stays initialized and exclusively owned here until consumed below.
52        unsafe { self.slots.set_len(self.started) };
53        for index in self.started..len {
54            // SAFETY: every original suffix slot is consumed exactly once.
55            let mut slot = unsafe { slots.add(index).read() };
56            // SAFETY: only a successful spawn transfers an input out of this
57            // suffix. Builder and Group retain every other initialized input.
58            unsafe { mem::ManuallyDrop::drop(&mut slot.value.get_mut().task) };
59        }
60        self.started = 0;
61        Finished {
62            slots: mem::take(&mut self.slots).into_iter(),
63        }
64    }
65}
66
67// SAFETY: the native entry accesses only this slot's task and output cells;
68// its Group retains their exclusive worker access until the stored handle joins.
69unsafe impl<T: Task<Output = O>, O, P> thread::raw::Task for Slot<T, O, P> {
70    /// Runs this slot's task once and stores its output. `Workers::start` admits
71    /// each slot once and stores its handle before its index advances, so the
72    /// retained task cell holds exactly one task at this point.
73    unsafe fn run(context: ptr::NonNull<Self>) {
74        let slot = context.as_ptr();
75        // SAFETY: the admitted context grants this invocation sole access to
76        // task and output. The consumed native entry transfers the initialized
77        // input once; Group only drops inputs outside its successful prefix.
78        let task = unsafe { mem::ManuallyDrop::take(&mut (*(*slot).value.get()).task) };
79        let output = task.run();
80        // SAFETY: the same retained context grants exclusive output access.
81        unsafe { (*(*slot).value.get()).output = mem::ManuallyDrop::new(output) };
82    }
83}
dope_core/platform/thread/raw.rs

dope_core/platform/thread/
raw.rs
