dope_core/platform/thread/
mod.rs

1use std::{cell, marker, mem, num::NonZero, ptr, vec};
2
3use dope_sys::thread;
4use o3::alloc::{self, VecExt};
5
6use crate::Errno;
7
8mod raw;
9
10union Value<T, O> {
11    task: mem::ManuallyDrop<T>,
12    output: mem::ManuallyDrop<O>,
13}
14
15/// A single owned input transferred into a worker and returned as an output.
16pub trait Task: Send {
17    type Output: Send;
18    fn run(self) -> Self::Output;
19}
20
21#[derive(Clone, Copy, Debug, PartialEq, Eq)]
22pub enum Error {
23    Attributes(Errno),
24    Create(Errno),
25}
26
27/// Builds the one allocation retained for the complete worker scope.
28pub struct Builder<T, O, P> {
29    group: raw::Group<T, O, P>,
30}
31
32impl<T, O, P> Builder<T, O, P> {
33    pub fn try_new(capacity: usize) -> Result<Self, alloc::Error> {
34        Ok(Self {
35            group: raw::Group {
36                slots: Vec::try_new(capacity)?,
37                started: 0,
38            },
39        })
40    }
41    pub fn try_push(&mut self, task: T, parent: P) -> Result<(), alloc::Error> {
42        alloc::try_vacant(&mut self.group.slots)?(Slot {
43            value: cell::UnsafeCell::new(Value {
44                task: mem::ManuallyDrop::new(task),
45            }),
46            thread: cell::Cell::new(mem::MaybeUninit::uninit()),
47            parent,
48        });
49        Ok(())
50    }
51    /// Retains every started worker even if the borrowing view is forgotten.
52    ///
53    pub fn scope<R>(self, run: impl for<'scope> FnOnce(Workers<'scope, T, O, P>) -> R) -> R {
54        let mut group = self.group;
55        run(Workers {
56            group: &mut group,
57            owner: marker::PhantomData,
58        })
59    }
60}
61
62/// A worker's parent-only metadata and private transfer storage.
63pub struct Slot<T, O, P> {
64    value: cell::UnsafeCell<Value<T, O>>,
65    thread: cell::Cell<mem::MaybeUninit<thread::Handle>>,
66    parent: P,
67}
68
69impl<T, O, P> Slot<T, O, P> {
70    pub fn parent(&self) -> &P {
71        &self.parent
72    }
73}
74
75/// Borrowing authority to start and observe the scope-owned workers.
76pub struct Workers<'scope, T, O, P> {
77    group: &'scope mut raw::Group<T, O, P>,
78    owner: marker::PhantomData<fn(&'scope ()) -> &'scope ()>,
79}
80
81impl<T, O, P> Workers<'_, T, O, P> {
82    pub fn entries(&self) -> &[Slot<T, O, P>] {
83        &self.group.slots
84    }
85    /// Mutably borrows only coordinator metadata for the started worker prefix.
86    /// Native entries retain exclusive access to their task and output cells.
87    pub fn parents_mut(&mut self) -> impl ExactSizeIterator<Item = &mut P> {
88        self.group.parents_mut()
89    }
90    /// Joins every started worker and hands back their outputs.
91    ///
92    /// The group retains no started slot afterwards, so its own join pass is empty.
93    pub fn finish(self) -> Finished<T, O, P> {
94        self.group.take()
95    }
96}
97
98/// Joined outputs, in the original registration order.
99pub struct Finished<T, O, P> {
100    slots: vec::IntoIter<Slot<T, O, P>>,
101}
102
103impl<T, O, P> Workers<'_, T, O, P> {
104    /// Spawns the remaining workers. The outer group owns every slot through
105    /// each join, so the context pointer stays valid until its handle joins.
106    /// A failure borrows the exact unstarted worker's parent metadata.
107    pub fn start(&mut self, stack: NonZero<usize>) -> Result<(), (&P, Error)>
108    where
109        T: Task<Output = O>,
110    {
111        // SAFETY: only successful spawns extend started within this fixed
112        // allocation; Workers cannot grow or remove slots before finish.
113        let pending = unsafe { self.group.slots.get_unchecked(self.group.started..) };
114        let Some(first) = pending.first() else {
115            return Ok(());
116        };
117        let attr = thread::Attrs::with_stack(stack)
118            .map_err(|error| (&first.parent, Error::Attributes(error)))?;
119        for slot in pending {
120            // SAFETY: Group owns this fixed slot allocation for the entire
121            // scope. The callback alone accesses task/output until join; the
122            // parent only accesses the disjoint handle and metadata. Store the
123            // handle before any fallible step. Group joins every stored handle
124            // before freeing slots, including when Workers is forgotten.
125            let entry = unsafe { thread::raw::Entry::new(ptr::NonNull::from(slot)) };
126            let handle = entry
127                .spawn(&attr)
128                .map_err(|error| (&slot.parent, Error::Create(error)))?;
129            slot.thread.set(mem::MaybeUninit::new(handle));
130            self.group.started += 1;
131        }
132        Ok(())
133    }
134}
135
136impl<T, O, P> Iterator for Finished<T, O, P> {
137    type Item = (P, O);
138
139    /// Yields each started worker's output; a started worker stores exactly one
140    /// output before its handle joins, so every retained slot below has one.
141    fn next(&mut self) -> Option<Self::Item> {
142        let slot = self.slots.next()?;
143        let Slot { value, parent, .. } = slot;
144        // SAFETY: the group joined this worker, and the worker stores exactly one
145        // output before returning, so the cell holds an initialized value.
146        Some((
147            parent,
148            mem::ManuallyDrop::into_inner(unsafe { value.into_inner().output }),
149        ))
150    }
151    fn size_hint(&self) -> (usize, Option<usize>) {
152        self.slots.size_hint()
153    }
154}
155impl<T, O, P> ExactSizeIterator for Finished<T, O, P> {}
156
157impl<T, O, P> Drop for Finished<T, O, P> {
158    fn drop(&mut self) {
159        for slot in self.slots.as_mut_slice() {
160            // SAFETY: the group joins every retained slot, and next removes
161            // each yielded slot. Every remaining output is initialized and owned.
162            unsafe { mem::ManuallyDrop::drop(&mut slot.value.get_mut().output) };
163        }
164    }
165}
dope_core/platform/thread/mod.rs

dope_core/platform/thread/
mod.rs
