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