tenferro_cpu/
domain_executor.rs

1use std::fmt::Debug;
2use std::num::NonZeroUsize;
3use std::sync::atomic::{AtomicU8, AtomicUsize, Ordering};
4use std::sync::Arc;
5
6use rayon::prelude::*;
7
8/// Inner parallel-region support offered by a CPU domain executor.
9///
10/// # Examples
11///
12/// ```rust
13/// use tenferro_cpu::CpuInnerParallelism;
14///
15/// assert_ne!(CpuInnerParallelism::None, CpuInnerParallelism::Rayon);
16/// ```
17#[derive(Clone, Copy, Debug, Eq, PartialEq)]
18pub enum CpuInnerParallelism {
19    /// The executor cannot host provider-owned inner parallel regions.
20    None,
21    /// The executor can host a Rayon-compatible inner parallel region.
22    Rayon,
23}
24
25/// Re-entry capability of one CPU domain executor.
26///
27/// This describes executor-level same-executor entry only. It never grants
28/// permission for recursive public [`crate::CpuBackend`] entry, which remains a
29/// separate backend contract.
30///
31/// # Examples
32///
33/// ```rust
34/// use tenferro_cpu::CpuExecutorReentrancy;
35///
36/// let policy = CpuExecutorReentrancy::Rejected;
37/// assert_eq!(policy, CpuExecutorReentrancy::Rejected);
38/// ```
39#[derive(Clone, Copy, Debug, Eq, PartialEq)]
40pub enum CpuExecutorReentrancy {
41    /// Nested entry into the same executor is rejected.
42    Rejected,
43    /// The executor supports nested entry into that same executor.
44    SameExecutor,
45}
46
47/// Affinity claim made by a CPU domain executor.
48///
49/// # Examples
50///
51/// ```rust
52/// use tenferro_cpu::CpuExecutorAffinity;
53///
54/// let affinity = CpuExecutorAffinity::CallerDeclaredUnverified;
55/// assert_ne!(affinity, CpuExecutorAffinity::TenferroPinnedVerified);
56/// ```
57#[derive(Clone, Copy, Debug, Eq, PartialEq)]
58pub enum CpuExecutorAffinity {
59    /// Tenferro pinned the workers and verified their placement.
60    TenferroPinnedVerified,
61    /// The caller declared worker placement, but tenferro did not verify it.
62    CallerDeclaredUnverified,
63    /// The executor makes no worker-placement claim.
64    None,
65}
66
67/// Ownership of CPU executor shutdown.
68///
69/// # Examples
70///
71/// ```rust
72/// use tenferro_cpu::CpuExecutorShutdown;
73///
74/// assert_ne!(
75///     CpuExecutorShutdown::TenferroOwned,
76///     CpuExecutorShutdown::CallerOwned,
77/// );
78/// ```
79#[derive(Clone, Copy, Debug, Eq, PartialEq)]
80pub enum CpuExecutorShutdown {
81    /// Tenferro owns executor shutdown.
82    TenferroOwned,
83    /// The caller owns executor shutdown and executor lifetime policy.
84    CallerOwned,
85}
86
87/// Immutable construction-time capabilities of a CPU domain executor.
88///
89/// # Examples
90///
91/// ```rust
92/// use std::num::NonZeroUsize;
93/// use tenferro_cpu::{
94///     CpuDomainExecutorCapabilities, CpuExecutorAffinity, CpuExecutorReentrancy,
95///     CpuExecutorShutdown, CpuInnerParallelism,
96/// };
97///
98/// let capabilities = CpuDomainExecutorCapabilities {
99///     worker_count: NonZeroUsize::new(4).unwrap(),
100///     outer_parallelism: true,
101///     inner_parallelism: CpuInnerParallelism::Rayon,
102///     reentrancy: CpuExecutorReentrancy::Rejected,
103///     affinity: CpuExecutorAffinity::TenferroPinnedVerified,
104///     shutdown: CpuExecutorShutdown::TenferroOwned,
105/// };
106/// assert_eq!(capabilities.worker_count.get(), 4);
107/// ```
108#[derive(Clone, Copy, Debug, Eq, PartialEq)]
109pub struct CpuDomainExecutorCapabilities {
110    /// Number of workers made available to this domain.
111    pub worker_count: NonZeroUsize,
112    /// Whether indexed outer fork/join submission is supported.
113    pub outer_parallelism: bool,
114    /// Provider-owned inner parallel-region support.
115    pub inner_parallelism: CpuInnerParallelism,
116    /// Same-executor re-entry capability.
117    pub reentrancy: CpuExecutorReentrancy,
118    /// Worker-affinity claim and verification level.
119    pub affinity: CpuExecutorAffinity,
120    /// Executor shutdown owner.
121    pub shutdown: CpuExecutorShutdown,
122}
123
124/// Failure at the CPU executor admission or scheduling boundary.
125///
126/// Operation and provider errors do not belong in this type. Executors use
127/// these variants only for their own admission, scheduling, cancellation, and
128/// panic-bridge failures.
129///
130/// # Examples
131///
132/// ```rust
133/// use tenferro_cpu::CpuDomainExecutorError;
134///
135/// let error = CpuDomainExecutorError::Admission {
136///     message: "domain is busy".to_string(),
137/// };
138/// assert!(matches!(error, CpuDomainExecutorError::Admission { .. }));
139/// ```
140#[derive(Clone, Debug, Eq, PartialEq, thiserror::Error)]
141pub enum CpuDomainExecutorError {
142    /// The executor rejected entry before scheduling work.
143    #[error("CPU domain executor admission failed: {message}")]
144    Admission {
145        /// Executor-owned diagnostic.
146        message: String,
147    },
148    /// The executor could not schedule or complete submitted work.
149    #[error("CPU domain executor scheduling failed: {message}")]
150    Scheduling {
151        /// Executor-owned diagnostic.
152        message: String,
153    },
154    /// The executor cancelled submitted work.
155    #[error("CPU domain executor cancelled work: {message}")]
156    Cancellation {
157        /// Executor-owned diagnostic.
158        message: String,
159    },
160    /// The executor converted a worker panic into a typed failure.
161    #[error("CPU domain executor worker panicked: {message}")]
162    PanicBridge {
163        /// Executor-owned diagnostic.
164        message: String,
165    },
166}
167
168/// One borrowed job installed synchronously into a CPU domain executor.
169///
170/// The executor must finish using the job before [`CpuDomainExecutor::install`]
171/// returns; a borrowed job never escapes that call.
172///
173/// # Examples
174///
175/// ```rust
176/// use tenferro_cpu::{CpuDomainExecutorError, ScopedCpuJob};
177///
178/// struct Job(bool);
179/// impl ScopedCpuJob for Job {
180///     fn run(&mut self) -> Result<(), CpuDomainExecutorError> {
181///         self.0 = true;
182///         Ok(())
183///     }
184/// }
185/// let mut job = Job(false);
186/// job.run().unwrap();
187/// assert!(job.0);
188/// ```
189pub trait ScopedCpuJob: Send {
190    /// Run this job once on the executor-selected calling context.
191    ///
192    /// # Examples
193    ///
194    /// ```rust
195    /// use tenferro_cpu::{CpuDomainExecutorError, ScopedCpuJob};
196    ///
197    /// struct Job;
198    /// impl ScopedCpuJob for Job {
199    ///     fn run(&mut self) -> Result<(), CpuDomainExecutorError> { Ok(()) }
200    /// }
201    /// assert!(Job.run().is_ok());
202    /// ```
203    ///
204    /// # Errors
205    ///
206    /// Returns [`CpuDomainExecutorError::Admission`],
207    /// [`CpuDomainExecutorError::Scheduling`],
208    /// [`CpuDomainExecutorError::Cancellation`], or
209    /// [`CpuDomainExecutorError::PanicBridge`] when that failure is observed
210    /// while running the job.
211    fn run(&mut self) -> Result<(), CpuDomainExecutorError>;
212}
213
214/// Synchronously submitted indexed jobs for engine-owned outer scheduling.
215///
216/// Implementations expose a borrowed logical range `0..len()` without
217/// allocating a job collection. Every indexed call must finish before
218/// [`CpuDomainExecutor::submit`] returns.
219///
220/// # Examples
221///
222/// ```rust
223/// use tenferro_cpu::{CpuDomainExecutorError, ScopedCpuJobs};
224///
225/// struct Jobs;
226/// impl ScopedCpuJobs for Jobs {
227///     fn len(&self) -> usize { 2 }
228///     fn run(&self, index: usize) -> Result<(), CpuDomainExecutorError> {
229///         assert!(index < self.len());
230///         Ok(())
231///     }
232/// }
233/// let jobs: &dyn ScopedCpuJobs = &Jobs;
234/// assert_eq!(jobs.len(), 2);
235/// jobs.run(1).unwrap();
236/// ```
237pub trait ScopedCpuJobs: Sync {
238    /// Return the number of indexed jobs in this synchronous submission.
239    ///
240    /// # Examples
241    ///
242    /// ```rust
243    /// use tenferro_cpu::{CpuDomainExecutorError, ScopedCpuJobs};
244    ///
245    /// struct Jobs;
246    /// impl ScopedCpuJobs for Jobs {
247    ///     fn len(&self) -> usize { 3 }
248    ///     fn run(&self, _index: usize) -> Result<(), CpuDomainExecutorError> { Ok(()) }
249    /// }
250    /// assert_eq!(Jobs.len(), 3);
251    /// ```
252    fn len(&self) -> usize;
253
254    /// Return whether this submission contains no indexed jobs.
255    ///
256    /// # Examples
257    ///
258    /// ```rust
259    /// use tenferro_cpu::{CpuDomainExecutorError, ScopedCpuJobs};
260    ///
261    /// struct Jobs;
262    /// impl ScopedCpuJobs for Jobs {
263    ///     fn len(&self) -> usize { 0 }
264    ///     fn run(&self, _index: usize) -> Result<(), CpuDomainExecutorError> { Ok(()) }
265    /// }
266    /// assert!(Jobs.is_empty());
267    /// ```
268    fn is_empty(&self) -> bool {
269        self.len() == 0
270    }
271
272    /// Run one indexed job synchronously.
273    ///
274    /// # Examples
275    ///
276    /// ```rust
277    /// use tenferro_cpu::{CpuDomainExecutorError, ScopedCpuJobs};
278    ///
279    /// struct Jobs;
280    /// impl ScopedCpuJobs for Jobs {
281    ///     fn len(&self) -> usize { 1 }
282    ///     fn run(&self, index: usize) -> Result<(), CpuDomainExecutorError> {
283    ///         assert_eq!(index, 0);
284    ///         Ok(())
285    ///     }
286    /// }
287    /// Jobs.run(0).unwrap();
288    /// ```
289    ///
290    /// # Errors
291    ///
292    /// Returns [`CpuDomainExecutorError::Admission`],
293    /// [`CpuDomainExecutorError::Scheduling`],
294    /// [`CpuDomainExecutorError::Cancellation`], or
295    /// [`CpuDomainExecutorError::PanicBridge`] when that failure is observed
296    /// while running the indexed job.
297    fn run(&self, index: usize) -> Result<(), CpuDomainExecutorError>;
298}
299
300/// Object-safe synchronous executor for one CPU resource domain.
301///
302/// `submit` is an indexed fork/join boundary and `install` is one borrowed
303/// provider-owned inner-region entry. Neither method may retain its borrowed
304/// job after returning.
305///
306/// # Examples
307///
308/// ```rust
309/// use std::num::NonZeroUsize;
310/// use tenferro_cpu::{
311///     CpuDomainExecutor, CpuDomainExecutorCapabilities, CpuDomainExecutorError,
312///     CpuExecutorAffinity, CpuExecutorReentrancy, CpuExecutorShutdown,
313///     CpuInnerParallelism, ScopedCpuJob, ScopedCpuJobs,
314/// };
315///
316/// #[derive(Debug)]
317/// struct Inline;
318/// impl CpuDomainExecutor for Inline {
319///     fn capabilities(&self) -> CpuDomainExecutorCapabilities {
320///         CpuDomainExecutorCapabilities {
321///             worker_count: NonZeroUsize::new(1).unwrap(),
322///             outer_parallelism: false,
323///             inner_parallelism: CpuInnerParallelism::None,
324///             reentrancy: CpuExecutorReentrancy::Rejected,
325///             affinity: CpuExecutorAffinity::None,
326///             shutdown: CpuExecutorShutdown::CallerOwned,
327///         }
328///     }
329///     fn submit(&self, _jobs: &dyn ScopedCpuJobs) -> Result<(), CpuDomainExecutorError> {
330///         Ok(())
331///     }
332///     fn install(&self, job: &mut dyn ScopedCpuJob) -> Result<(), CpuDomainExecutorError> {
333///         job.run()
334///     }
335/// }
336/// let executor: &dyn CpuDomainExecutor = &Inline;
337/// assert_eq!(executor.capabilities().worker_count.get(), 1);
338/// ```
339pub trait CpuDomainExecutor: Debug + Send + Sync + 'static {
340    /// Return immutable construction-time executor capabilities.
341    ///
342    /// # Examples
343    ///
344    /// ```rust
345    /// use std::num::NonZeroUsize;
346    /// use tenferro_cpu::{
347    ///     CpuDomainExecutor, CpuDomainExecutorCapabilities, CpuDomainExecutorError,
348    ///     CpuExecutorAffinity, CpuExecutorReentrancy, CpuExecutorShutdown,
349    ///     CpuInnerParallelism, ScopedCpuJob, ScopedCpuJobs,
350    /// };
351    /// # #[derive(Debug)] struct Inline;
352    /// # impl CpuDomainExecutor for Inline {
353    /// # fn capabilities(&self) -> CpuDomainExecutorCapabilities {
354    /// # CpuDomainExecutorCapabilities { worker_count: NonZeroUsize::new(2).unwrap(),
355    /// # outer_parallelism: true, inner_parallelism: CpuInnerParallelism::None,
356    /// # reentrancy: CpuExecutorReentrancy::Rejected, affinity: CpuExecutorAffinity::None,
357    /// # shutdown: CpuExecutorShutdown::CallerOwned }
358    /// # }
359    /// # fn submit(&self, _jobs: &dyn ScopedCpuJobs) -> Result<(), CpuDomainExecutorError> { Ok(()) }
360    /// # fn install(&self, job: &mut dyn ScopedCpuJob) -> Result<(), CpuDomainExecutorError> { job.run() }
361    /// # }
362    /// let executor: &dyn CpuDomainExecutor = &Inline;
363    /// assert_eq!(executor.capabilities().worker_count.get(), 2);
364    /// ```
365    fn capabilities(&self) -> CpuDomainExecutorCapabilities;
366
367    /// Submit all indexed jobs as one synchronous fork/join operation.
368    ///
369    /// All `0..jobs.len()` jobs must be complete when this method returns.
370    ///
371    /// # Examples
372    ///
373    /// ```rust
374    /// use std::num::NonZeroUsize;
375    /// use std::sync::atomic::{AtomicUsize, Ordering};
376    /// use tenferro_cpu::{
377    ///     CpuDomainExecutor, CpuDomainExecutorCapabilities, CpuDomainExecutorError,
378    ///     CpuExecutorAffinity, CpuExecutorReentrancy, CpuExecutorShutdown,
379    ///     CpuInnerParallelism, ScopedCpuJob, ScopedCpuJobs,
380    /// };
381    /// # #[derive(Debug)] struct Inline;
382    /// # impl CpuDomainExecutor for Inline {
383    /// # fn capabilities(&self) -> CpuDomainExecutorCapabilities {
384    /// # CpuDomainExecutorCapabilities { worker_count: NonZeroUsize::new(1).unwrap(),
385    /// # outer_parallelism: true, inner_parallelism: CpuInnerParallelism::None,
386    /// # reentrancy: CpuExecutorReentrancy::Rejected, affinity: CpuExecutorAffinity::None,
387    /// # shutdown: CpuExecutorShutdown::CallerOwned }
388    /// # }
389    /// # fn submit(&self, jobs: &dyn ScopedCpuJobs) -> Result<(), CpuDomainExecutorError> {
390    /// # for index in 0..jobs.len() { jobs.run(index)?; } Ok(())
391    /// # }
392    /// # fn install(&self, job: &mut dyn ScopedCpuJob) -> Result<(), CpuDomainExecutorError> { job.run() }
393    /// # }
394    /// struct Jobs<'a>(&'a AtomicUsize);
395    /// impl ScopedCpuJobs for Jobs<'_> {
396    ///     fn len(&self) -> usize { 2 }
397    ///     fn run(&self, _index: usize) -> Result<(), CpuDomainExecutorError> {
398    ///         self.0.fetch_add(1, Ordering::Relaxed);
399    ///         Ok(())
400    ///     }
401    /// }
402    /// let count = AtomicUsize::new(0);
403    /// Inline.submit(&Jobs(&count)).unwrap();
404    /// assert_eq!(count.load(Ordering::Relaxed), 2);
405    /// ```
406    ///
407    /// # Errors
408    ///
409    /// Returns [`CpuDomainExecutorError::Admission`],
410    /// [`CpuDomainExecutorError::Scheduling`],
411    /// [`CpuDomainExecutorError::Cancellation`], or
412    /// [`CpuDomainExecutorError::PanicBridge`] for executor-owned failures.
413    fn submit(&self, jobs: &dyn ScopedCpuJobs) -> Result<(), CpuDomainExecutorError>;
414
415    /// Enter one synchronous provider-owned inner parallel region.
416    ///
417    /// # Examples
418    ///
419    /// ```rust
420    /// use std::num::NonZeroUsize;
421    /// use tenferro_cpu::{
422    ///     CpuDomainExecutor, CpuDomainExecutorCapabilities, CpuDomainExecutorError,
423    ///     CpuExecutorAffinity, CpuExecutorReentrancy, CpuExecutorShutdown,
424    ///     CpuInnerParallelism, ScopedCpuJob, ScopedCpuJobs,
425    /// };
426    /// # #[derive(Debug)] struct Inline;
427    /// # impl CpuDomainExecutor for Inline {
428    /// # fn capabilities(&self) -> CpuDomainExecutorCapabilities {
429    /// # CpuDomainExecutorCapabilities { worker_count: NonZeroUsize::new(1).unwrap(),
430    /// # outer_parallelism: false, inner_parallelism: CpuInnerParallelism::None,
431    /// # reentrancy: CpuExecutorReentrancy::Rejected, affinity: CpuExecutorAffinity::None,
432    /// # shutdown: CpuExecutorShutdown::CallerOwned }
433    /// # }
434    /// # fn submit(&self, _jobs: &dyn ScopedCpuJobs) -> Result<(), CpuDomainExecutorError> { Ok(()) }
435    /// # fn install(&self, job: &mut dyn ScopedCpuJob) -> Result<(), CpuDomainExecutorError> { job.run() }
436    /// # }
437    /// struct Job(bool);
438    /// impl ScopedCpuJob for Job {
439    ///     fn run(&mut self) -> Result<(), CpuDomainExecutorError> {
440    ///         self.0 = true;
441    ///         Ok(())
442    ///     }
443    /// }
444    /// let mut job = Job(false);
445    /// Inline.install(&mut job).unwrap();
446    /// assert!(job.0);
447    /// ```
448    ///
449    /// # Errors
450    ///
451    /// Returns [`CpuDomainExecutorError::Admission`],
452    /// [`CpuDomainExecutorError::Scheduling`],
453    /// [`CpuDomainExecutorError::Cancellation`], or
454    /// [`CpuDomainExecutorError::PanicBridge`] for executor-owned failures.
455    fn install(&self, job: &mut dyn ScopedCpuJob) -> Result<(), CpuDomainExecutorError>;
456}
457
458/// Adapter that executes CPU-domain jobs on one caller-owned Rayon pool.
459///
460/// The adapter retains the supplied pool and never creates, reconfigures, or
461/// shuts it down. [`CpuDomainExecutor`] remains the primary injection contract;
462/// this type is only a convenience for Rayon hosts.
463///
464/// # Examples
465///
466/// ```rust
467/// use std::sync::Arc;
468/// use tenferro_cpu::{CpuDomainExecutor, RayonCpuDomainExecutor};
469///
470/// let pool = Arc::new(rayon::ThreadPoolBuilder::new().num_threads(2).build()?);
471/// let executor = RayonCpuDomainExecutor::new(Arc::clone(&pool));
472/// assert_eq!(executor.capabilities().worker_count.get(), 2);
473/// assert_eq!(Arc::strong_count(&pool), 2);
474/// # Ok::<(), rayon::ThreadPoolBuildError>(())
475/// ```
476pub struct RayonCpuDomainExecutor {
477    pool: Arc<rayon::ThreadPool>,
478}
479
480impl std::fmt::Debug for RayonCpuDomainExecutor {
481    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
482        formatter
483            .debug_struct("RayonCpuDomainExecutor")
484            .field("worker_count", &self.pool.current_num_threads())
485            .finish_non_exhaustive()
486    }
487}
488
489impl RayonCpuDomainExecutor {
490    /// Retain one caller-owned Rayon pool as a CPU-domain executor.
491    ///
492    /// # Examples
493    ///
494    /// ```rust
495    /// use std::sync::Arc;
496    /// use tenferro_cpu::{CpuDomainExecutor, RayonCpuDomainExecutor};
497    ///
498    /// let pool = Arc::new(rayon::ThreadPoolBuilder::new().num_threads(2).build()?);
499    /// let executor = RayonCpuDomainExecutor::new(pool);
500    /// assert_eq!(executor.capabilities().worker_count.get(), 2);
501    /// # Ok::<(), rayon::ThreadPoolBuildError>(())
502    /// ```
503    pub fn new(pool: Arc<rayon::ThreadPool>) -> Self {
504        Self { pool }
505    }
506}
507
508impl CpuDomainExecutor for RayonCpuDomainExecutor {
509    fn capabilities(&self) -> CpuDomainExecutorCapabilities {
510        // INVARIANT: Rayon rejects thread pools with zero workers.
511        let worker_count =
512            NonZeroUsize::new(self.pool.current_num_threads()).unwrap_or(NonZeroUsize::MIN);
513        CpuDomainExecutorCapabilities {
514            worker_count,
515            outer_parallelism: worker_count.get() > 1,
516            inner_parallelism: CpuInnerParallelism::Rayon,
517            reentrancy: CpuExecutorReentrancy::SameExecutor,
518            affinity: CpuExecutorAffinity::None,
519            shutdown: CpuExecutorShutdown::CallerOwned,
520        }
521    }
522
523    fn submit(&self, jobs: &dyn ScopedCpuJobs) -> Result<(), CpuDomainExecutorError> {
524        self.pool.install(|| {
525            (0..jobs.len())
526                .into_par_iter()
527                .try_for_each(|index| jobs.run(index))
528        })
529    }
530
531    fn install(&self, job: &mut dyn ScopedCpuJob) -> Result<(), CpuDomainExecutorError> {
532        self.pool.install(|| job.run())
533    }
534}
535
536pub(crate) struct ScopedJob<F, R> {
537    operation: Option<F>,
538    result: Option<R>,
539}
540
541pub(crate) fn scoped_job<F, R>(operation: F) -> ScopedJob<F, R>
542where
543    F: FnOnce() -> R + Send,
544    R: Send,
545{
546    ScopedJob {
547        operation: Some(operation),
548        result: None,
549    }
550}
551
552impl<F, R> ScopedJob<F, R> {
553    fn into_result(self) -> Result<R, CpuDomainExecutorError> {
554        self.result
555            .ok_or_else(|| CpuDomainExecutorError::Scheduling {
556                message: "executor returned success without running the scoped CPU job".to_string(),
557            })
558    }
559}
560
561impl<F, R> ScopedCpuJob for ScopedJob<F, R>
562where
563    F: FnOnce() -> R + Send,
564    R: Send,
565{
566    fn run(&mut self) -> Result<(), CpuDomainExecutorError> {
567        let operation =
568            self.operation
569                .take()
570                .ok_or_else(|| CpuDomainExecutorError::Scheduling {
571                    message: "executor attempted to run a scoped CPU job more than once"
572                        .to_string(),
573                })?;
574        self.result = Some(operation());
575        Ok(())
576    }
577}
578
579pub(crate) fn install_scoped<F, R>(
580    executor: &dyn CpuDomainExecutor,
581    operation: F,
582) -> Result<R, CpuDomainExecutorError>
583where
584    F: FnOnce() -> R + Send,
585    R: Send,
586{
587    let mut job = scoped_job(operation);
588    executor.install(&mut job)?;
589    job.into_result()
590}
591
592pub(crate) struct IndexedJobs<F> {
593    len: usize,
594    run: F,
595    invalid_index_attempt: InvalidIndexAudit,
596}
597
598const INVALID_INDEX_EMPTY: u8 = 0;
599const INVALID_INDEX_WRITING: u8 = 1;
600const INVALID_INDEX_READY: u8 = 2;
601
602// INVARIANT: the two-phase state publishes every usize value, including
603// usize::MAX, without a sentinel collision. Valid `run` calls never touch this
604// audit, and the post-submit Acquire observes the selected invalid index after
605// its Release publication without locking or allocating.
606struct InvalidIndexAudit {
607    state: AtomicU8,
608    index: AtomicUsize,
609}
610
611impl InvalidIndexAudit {
612    const fn new() -> Self {
613        Self {
614            state: AtomicU8::new(INVALID_INDEX_EMPTY),
615            index: AtomicUsize::new(0),
616        }
617    }
618
619    fn record(&self, index: usize) {
620        if self
621            .state
622            .compare_exchange(
623                INVALID_INDEX_EMPTY,
624                INVALID_INDEX_WRITING,
625                Ordering::AcqRel,
626                Ordering::Acquire,
627            )
628            .is_ok()
629        {
630            self.index.store(index, Ordering::Relaxed);
631            self.state.store(INVALID_INDEX_READY, Ordering::Release);
632        }
633    }
634
635    fn load(&self) -> Option<usize> {
636        (self.state.load(Ordering::Acquire) == INVALID_INDEX_READY)
637            .then(|| self.index.load(Ordering::Relaxed))
638    }
639}
640
641pub(crate) fn indexed_jobs<F>(len: usize, run: F) -> IndexedJobs<F>
642where
643    F: Fn(usize) -> Result<(), CpuDomainExecutorError> + Sync,
644{
645    IndexedJobs {
646        len,
647        run,
648        invalid_index_attempt: InvalidIndexAudit::new(),
649    }
650}
651
652impl<F> IndexedJobs<F> {
653    pub(crate) fn invalid_index_attempt(&self) -> Option<usize> {
654        self.invalid_index_attempt.load()
655    }
656}
657
658impl<F> ScopedCpuJobs for IndexedJobs<F>
659where
660    F: Fn(usize) -> Result<(), CpuDomainExecutorError> + Sync,
661{
662    fn len(&self) -> usize {
663        self.len
664    }
665
666    fn run(&self, index: usize) -> Result<(), CpuDomainExecutorError> {
667        if index >= self.len {
668            self.invalid_index_attempt.record(index);
669            return Err(CpuDomainExecutorError::Scheduling {
670                message: format!(
671                    "executor requested scoped CPU job index {index}, but the submission has {} jobs",
672                    self.len
673                ),
674            });
675        }
676        (self.run)(index)
677    }
678}
679
680#[cfg(test)]
681mod tests;
tenferro_cpu/domain_executor.rs

tenferro_cpu/
domain_executor.rs
