automation_structures/
execution_api.rs

1//! Checked public entry points for reusable execution modalities.
2
3use crate::modalities::fork_join::ForkJoin as ForkJoinCarrier;
4use crate::modalities::sequential::Sequential as SequentialCarrier;
5use crate::modalities::step_graph::StepGraph as StepGraphCarrier;
6use crate::modalities::stream_graph::StreamGraph as StreamGraphCarrier;
7use vstd::prelude::*;
8
9verus! {
10
11/// Invalid sequential-execution configuration.
12#[derive(Clone, Copy, Debug, Eq, PartialEq)]
13#[non_exhaustive]
14pub enum SequentialBuildError {
15    /// Sequential execution requires at least one step.
16    NoSteps,
17    /// The value domain must contain at least one value.
18    EmptyValueDomain,
19    /// The initial value is outside the configured domain.
20    InitialValueOutOfRange,
21}
22
23/// A totally ordered, finite-step execution modality.
24///
25/// # Examples
26///
27/// ```rust
28/// use automation_structures::Sequential;
29///
30/// let mut execution = Sequential::new(1, 3, 0)?;
31/// assert!(execution.begin_step());
32/// assert!(execution.complete_step(2));
33/// assert!(execution.is_done());
34/// # Ok::<(), automation_structures::SequentialBuildError>(())
35/// ```
36pub struct Sequential {
37    inner: SequentialCarrier,
38}
39
40impl Sequential {
41    #[verifier::type_invariant]
42    closed spec fn well_formed(&self) -> bool { self.inner.inv() }
43
44    /// Validate and construct an inactive sequential execution.
45    ///
46    /// # Errors
47    ///
48    /// Returns an error for zero steps, an empty value domain, or an out-of-domain
49    /// initial value.
50    pub fn new(steps: usize, value_domain_size: u64, initial_value: u64)
51        -> (result: Result<Self, SequentialBuildError>) {
52        if steps == 0 { return Err(SequentialBuildError::NoSteps); }
53        if value_domain_size == 0 { return Err(SequentialBuildError::EmptyValueDomain); }
54        if initial_value >= value_domain_size {
55            return Err(SequentialBuildError::InitialValueOutOfRange);
56        }
57        Ok(Self { inner: SequentialCarrier::new(steps, value_domain_size, initial_value) })
58    }
59
60    /// Total number of steps.
61    pub fn steps(&self) -> usize { self.inner.steps }
62
63    /// Number of completed steps.
64    pub fn completed(&self) -> usize { self.inner.pc }
65
66    /// Current carried value.
67    pub fn value(&self) -> u64 { self.inner.value }
68
69    /// Whether a step is active.
70    pub fn is_active(&self) -> bool { self.inner.active }
71
72    /// Whether all steps are complete and inactive.
73    pub fn is_done(&self) -> bool { self.inner.pc == self.inner.steps && !self.inner.active }
74
75    /// Read a completed-step value by execution order.
76    #[expect(clippy::indexing_slicing, reason = "the branch proves the history index is in bounds")]
77    pub fn history(&self, index: usize) -> Option<u64> {
78        if index < self.inner.history.len() { Some(self.inner.history[index]) } else { None }
79    }
80
81    /// Begin the next step if execution is inactive and incomplete.
82    #[must_use]
83    pub fn begin_step(&mut self) -> (accepted: bool) {
84        proof { use_type_invariant(&*self); }
85        let mut carrier = sequential_sentinel();
86        core::mem::swap(&mut self.inner, &mut carrier);
87        let accepted = carrier.begin_step();
88        core::mem::swap(&mut self.inner, &mut carrier);
89        accepted
90    }
91
92    /// Complete the active step with a value in the configured domain.
93    #[must_use]
94    pub fn complete_step(&mut self, next_value: u64) -> (accepted: bool) {
95        proof { use_type_invariant(&*self); }
96        let mut carrier = sequential_sentinel();
97        core::mem::swap(&mut self.inner, &mut carrier);
98        let accepted = carrier.complete_step(next_value);
99        core::mem::swap(&mut self.inner, &mut carrier);
100        accepted
101    }
102}
103
104/// Invalid fork-join configuration.
105#[derive(Clone, Copy, Debug, Eq, PartialEq)]
106#[non_exhaustive]
107pub enum ForkJoinBuildError {
108    /// The value domain must contain at least one value.
109    EmptyValueDomain,
110    /// The initial worker value is outside the configured domain.
111    InitialValueOutOfRange,
112}
113
114/// One worker's fork-join lifecycle state.
115#[derive(Clone, Copy, Debug, Eq, PartialEq)]
116pub enum WorkerState {
117    /// The worker may be started.
118    Ready,
119    /// The worker is running.
120    Running,
121    /// The worker has produced its value.
122    Complete,
123}
124
125/// The global fork-join phase.
126#[derive(Clone, Copy, Debug, Eq, PartialEq)]
127pub enum ForkJoinPhase {
128    /// Workers may be started and completed.
129    Fork,
130    /// All workers are complete and the barrier has committed.
131    Join,
132    /// A stable output snapshot has been produced.
133    Done,
134}
135
136/// A barriered fork-join execution with a stable output snapshot.
137///
138/// # Examples
139///
140/// ```rust
141/// use automation_structures::ForkJoin;
142///
143/// let mut execution = ForkJoin::new(1, 4, 0)?;
144/// assert!(execution.start_worker(0));
145/// assert!(execution.complete_worker(0, 3));
146/// assert!(execution.barrier());
147/// assert!(execution.produce_output());
148/// assert_eq!(execution.outputs(), Some(&[3][..]));
149/// # Ok::<(), automation_structures::ForkJoinBuildError>(())
150/// ```
151pub struct ForkJoin {
152    inner: ForkJoinCarrier,
153}
154
155impl ForkJoin {
156    #[verifier::type_invariant]
157    closed spec fn well_formed(&self) -> bool { self.inner.inv() }
158
159    /// Validate and construct a fork-join execution.
160    ///
161    /// # Errors
162    ///
163    /// Returns an error for an empty value domain or an out-of-domain initial value.
164    pub fn new(workers: usize, value_domain_size: u64, initial_value: u64)
165        -> (result: Result<Self, ForkJoinBuildError>) {
166        if value_domain_size == 0 { return Err(ForkJoinBuildError::EmptyValueDomain); }
167        if initial_value >= value_domain_size {
168            return Err(ForkJoinBuildError::InitialValueOutOfRange);
169        }
170        Ok(Self { inner: ForkJoinCarrier::new(workers, value_domain_size, initial_value) })
171    }
172
173    /// Number of workers.
174    pub fn len(&self) -> usize { self.inner.wstate.len() }
175
176    /// Whether no workers are configured.
177    pub fn is_empty(&self) -> bool { self.inner.wstate.is_empty() }
178
179    /// Current global phase.
180    pub fn phase(&self) -> ForkJoinPhase {
181        self.inner.phase
182    }
183
184    /// Whether the output snapshot is ready.
185    pub fn output_ready(&self) -> bool { self.inner.output_ready }
186
187    /// Read one worker lifecycle state.
188    #[expect(clippy::indexing_slicing, reason = "the branch proves the worker index is in bounds")]
189    pub fn worker_state(&self, worker: usize) -> Option<WorkerState> {
190        if worker >= self.inner.wstate.len() { return None; }
191        Some(self.inner.wstate[worker])
192    }
193
194    /// Read one worker's current value.
195    #[expect(clippy::indexing_slicing, reason = "the branch proves the worker index is in bounds")]
196    pub fn worker_value(&self, worker: usize) -> Option<u64> {
197        if worker < self.inner.wvalue.len() { Some(self.inner.wvalue[worker]) } else { None }
198    }
199
200    /// Read one stable output value after output production.
201    #[expect(clippy::indexing_slicing, reason = "the branch proves the output index is in bounds")]
202    pub fn output(&self, worker: usize) -> Option<u64> {
203        if self.inner.output_ready && worker < self.inner.output_snapshot.len() {
204            Some(self.inner.output_snapshot[worker])
205        } else { None }
206    }
207
208    /// Start one ready worker during the fork phase.
209    #[must_use]
210    pub fn start_worker(&mut self, worker: usize) -> (accepted: bool) {
211        proof { use_type_invariant(&*self); }
212        let mut carrier = fork_join_sentinel();
213        core::mem::swap(&mut self.inner, &mut carrier);
214        let accepted = carrier.start_worker(worker);
215        core::mem::swap(&mut self.inner, &mut carrier);
216        accepted
217    }
218
219    /// Complete one running worker with an in-domain value.
220    #[must_use]
221    pub fn complete_worker(&mut self, worker: usize, value: u64) -> (accepted: bool) {
222        proof { use_type_invariant(&*self); }
223        let mut carrier = fork_join_sentinel();
224        core::mem::swap(&mut self.inner, &mut carrier);
225        let accepted = carrier.complete_worker(worker, value);
226        core::mem::swap(&mut self.inner, &mut carrier);
227        accepted
228    }
229
230    /// Commit the barrier when every worker is complete.
231    #[must_use]
232    pub fn barrier(&mut self) -> (accepted: bool) {
233        proof { use_type_invariant(&*self); }
234        let mut carrier = fork_join_sentinel();
235        core::mem::swap(&mut self.inner, &mut carrier);
236        let accepted = carrier.barrier();
237        core::mem::swap(&mut self.inner, &mut carrier);
238        accepted
239    }
240
241    /// Produce the immutable output snapshot from the joined worker values.
242    #[must_use]
243    pub fn produce_output(&mut self) -> (accepted: bool) {
244        proof { use_type_invariant(&*self); }
245        let mut carrier = fork_join_sentinel();
246        core::mem::swap(&mut self.inner, &mut carrier);
247        let accepted = carrier.produce_output();
248        core::mem::swap(&mut self.inner, &mut carrier);
249        accepted
250    }
251}
252
253/// Invalid step-graph configuration.
254#[derive(Clone, Copy, Debug, Eq, PartialEq)]
255#[non_exhaustive]
256pub enum StepGraphBuildError {
257    /// At least one edge endpoint is outside the node universe.
258    EdgeEndpointOutOfRange,
259    /// Duplicate edges are not admitted.
260    DuplicateEdge,
261}
262
263/// One step's lifecycle state.
264#[derive(Clone, Copy, Debug, Eq, PartialEq)]
265pub enum StepState {
266    /// A predecessor is incomplete.
267    NotReady,
268    /// All predecessors are complete.
269    Ready,
270    /// The step is running.
271    Running,
272    /// The step is complete.
273    Complete,
274}
275
276/// A predecessor-governed directed step graph.
277///
278/// # Examples
279///
280/// ```rust
281/// use automation_structures::StepGraph;
282///
283/// let mut graph = StepGraph::new(2, vec![(0, 1)])?;
284/// assert!(graph.start(0));
285/// assert!(graph.complete(0));
286/// assert!(graph.become_ready(1));
287/// # Ok::<(), automation_structures::StepGraphBuildError>(())
288/// ```
289pub struct StepGraph {
290    inner: StepGraphCarrier,
291}
292
293impl StepGraph {
294    #[verifier::type_invariant]
295    closed spec fn well_formed(&self) -> bool { self.inner.inv() }
296
297    /// Validate edges and construct initial readiness states.
298    ///
299    /// # Errors
300    ///
301    /// Returns an error when an endpoint is outside the node universe or an edge is duplicated.
302    pub fn new(num_nodes: usize, edges: Vec<(usize, usize)>)
303        -> (result: Result<Self, StepGraphBuildError>) {
304        if !step_edges_valid(&edges, num_nodes) {
305            return Err(StepGraphBuildError::EdgeEndpointOutOfRange);
306        }
307        if !step_edges_distinct(&edges) { return Err(StepGraphBuildError::DuplicateEdge); }
308        Ok(Self { inner: StepGraphCarrier::new(num_nodes, edges) })
309    }
310
311    /// Number of steps.
312    pub fn len(&self) -> usize { self.inner.num_nodes }
313
314    /// Whether the graph has no steps.
315    pub fn is_empty(&self) -> bool { self.inner.num_nodes == 0 }
316
317    /// Number of directed predecessor edges.
318    pub fn edge_count(&self) -> usize { self.inner.edges.len() }
319
320    /// Read one directed predecessor edge.
321    #[expect(clippy::indexing_slicing, reason = "the branch proves the edge index is in bounds")]
322    pub fn edge(&self, index: usize) -> Option<(usize, usize)> {
323        if index < self.inner.edges.len() { Some(self.inner.edges[index]) } else { None }
324    }
325
326    /// Read one step lifecycle state.
327    #[expect(clippy::indexing_slicing, reason = "the branch proves the node index is in bounds")]
328    pub fn state(&self, node: usize) -> Option<StepState> {
329        if node >= self.inner.nstate.len() { return None; }
330        Some(self.inner.nstate[node])
331    }
332
333    /// Promote a blocked node after every predecessor completes.
334    #[must_use]
335    pub fn become_ready(&mut self, node: usize) -> (accepted: bool) {
336        proof { use_type_invariant(&*self); }
337        let mut carrier = step_graph_sentinel();
338        core::mem::swap(&mut self.inner, &mut carrier);
339        let accepted = carrier.become_ready(node);
340        core::mem::swap(&mut self.inner, &mut carrier);
341        accepted
342    }
343
344    /// Start one ready step.
345    #[must_use]
346    pub fn start(&mut self, node: usize) -> (accepted: bool) {
347        proof { use_type_invariant(&*self); }
348        let mut carrier = step_graph_sentinel();
349        core::mem::swap(&mut self.inner, &mut carrier);
350        let accepted = carrier.start_running(node);
351        core::mem::swap(&mut self.inner, &mut carrier);
352        accepted
353    }
354
355    /// Complete one running step.
356    #[must_use]
357    pub fn complete(&mut self, node: usize) -> (accepted: bool) {
358        proof { use_type_invariant(&*self); }
359        let mut carrier = step_graph_sentinel();
360        core::mem::swap(&mut self.inner, &mut carrier);
361        let accepted = carrier.complete_node(node);
362        core::mem::swap(&mut self.inner, &mut carrier);
363        accepted
364    }
365
366    /// Whether every step is complete.
367    #[expect(clippy::indexing_slicing, reason = "the loop proves the state index is in bounds")]
368    #[expect(clippy::arithmetic_side_effects, reason = "the loop proves the cursor remains within the vector")]
369    pub fn is_done(&self) -> bool {
370        let mut index = 0;
371        while index < self.inner.nstate.len()
372            invariant index <= self.inner.nstate.len(),
373            decreases self.inner.nstate.len() - index,
374        {
375            if !matches!(self.inner.nstate[index], StepState::Complete) { return false; }
376            index += 1;
377        }
378        true
379    }
380}
381
382/// Invalid stream-graph configuration.
383#[derive(Clone, Copy, Debug, Eq, PartialEq)]
384#[non_exhaustive]
385pub enum StreamGraphBuildError {
386    /// Only three- and four-stage chains are currently represented.
387    UnsupportedChainLength,
388    /// Every inter-stage queue needs positive capacity.
389    ZeroCapacity,
390    /// The record value domain must be nonempty.
391    EmptyRecordDomain,
392}
393
394/// A bounded three- or four-stage FIFO stream graph.
395///
396/// # Examples
397///
398/// ```rust
399/// use automation_structures::StreamGraph;
400///
401/// let mut graph = StreamGraph::new(3, 1, 1, 4)?;
402/// assert!(graph.ingest(3));
403/// assert!(graph.advance_first());
404/// assert_eq!(graph.consume(), Some(3));
405/// assert!(graph.is_done());
406/// # Ok::<(), automation_structures::StreamGraphBuildError>(())
407/// ```
408pub struct StreamGraph {
409    inner: StreamGraphCarrier,
410}
411
412impl StreamGraph {
413    #[verifier::type_invariant]
414    closed spec fn well_formed(&self) -> bool { self.inner.inv() }
415
416    /// Validate and construct an empty stream graph.
417    ///
418    /// # Errors
419    ///
420    /// Returns an error for an unsupported chain length, zero queue capacity, or an
421    /// empty record domain.
422    pub fn new(chain_length: usize, capacity: usize, max_inputs: usize, record_domain_size: u64)
423        -> (result: Result<Self, StreamGraphBuildError>) {
424        if chain_length != 3 && chain_length != 4 {
425            return Err(StreamGraphBuildError::UnsupportedChainLength);
426        }
427        if capacity == 0 { return Err(StreamGraphBuildError::ZeroCapacity); }
428        if record_domain_size == 0 { return Err(StreamGraphBuildError::EmptyRecordDomain); }
429        Ok(Self { inner: StreamGraphCarrier::new(
430            chain_length, capacity, max_inputs, record_domain_size,
431        ) })
432    }
433
434    /// Number of execution stages.
435    pub fn chain_length(&self) -> usize { self.inner.chain_length }
436
437    /// Per-edge FIFO capacity.
438    pub fn capacity(&self) -> usize { self.inner.capacity() }
439
440    /// Records admitted at the source.
441    pub fn ingested(&self) -> usize { self.inner.ingested.value() as usize }
442
443    /// Records consumed at the sink.
444    pub fn emitted(&self) -> usize { self.inner.emitted.value() as usize }
445
446    /// Current depth of the first queue.
447    pub fn first_queue_len(&self) -> usize { self.inner.q1.len() }
448
449    /// Current depth of the second queue.
450    pub fn second_queue_len(&self) -> usize { self.inner.q2.len() }
451
452    /// Current depth of the optional third queue.
453    pub fn third_queue_len(&self) -> usize { self.inner.q3.len() }
454
455    /// Admit one source record if its value, input bound, and backpressure permit it.
456    #[must_use]
457    pub fn ingest(&mut self, value: u64) -> (accepted: bool) {
458        proof { use_type_invariant(&*self); }
459        let mut carrier = stream_graph_sentinel();
460        core::mem::swap(&mut self.inner, &mut carrier);
461        let accepted = carrier.source_ingest(value);
462        core::mem::swap(&mut self.inner, &mut carrier);
463        accepted
464    }
465
466    /// Transfer one FIFO record across the first internal stage.
467    #[must_use]
468    pub fn advance_first(&mut self) -> (accepted: bool) {
469        proof { use_type_invariant(&*self); }
470        let mut carrier = stream_graph_sentinel();
471        core::mem::swap(&mut self.inner, &mut carrier);
472        let accepted = carrier.middle2_fire();
473        core::mem::swap(&mut self.inner, &mut carrier);
474        accepted
475    }
476
477    /// Transfer one FIFO record across the optional four-stage link.
478    #[must_use]
479    pub fn advance_second(&mut self) -> (accepted: bool) {
480        proof { use_type_invariant(&*self); }
481        let mut carrier = stream_graph_sentinel();
482        core::mem::swap(&mut self.inner, &mut carrier);
483        let accepted = carrier.middle3_fire();
484        core::mem::swap(&mut self.inner, &mut carrier);
485        accepted
486    }
487
488    /// Consume and return the next FIFO record at the sink.
489    #[expect(clippy::indexing_slicing, reason = "the queue guards prove the sink head is present")]
490    pub fn consume(&mut self) -> (value: Option<u64>) {
491        proof { use_type_invariant(&*self); }
492        let value = if self.inner.chain_length == 3 {
493            if self.inner.q2.is_empty() { return None; }
494            self.inner.q2.values[0]
495        } else {
496            if self.inner.q3.is_empty() { return None; }
497            self.inner.q3.values[0]
498        };
499        let mut carrier = stream_graph_sentinel();
500        core::mem::swap(&mut self.inner, &mut carrier);
501        let accepted = carrier.sink_consume();
502        if !accepted {
503            core::mem::swap(&mut self.inner, &mut carrier);
504            return None;
505        }
506        core::mem::swap(&mut self.inner, &mut carrier);
507        Some(value)
508    }
509
510    /// Whether the input bound is reached and every queue is drained.
511    pub fn is_done(&self) -> bool {
512        self.inner.ingested.value() == self.inner.max_inputs as u64
513            && self.inner.q1.is_empty()
514            && self.inner.q2.is_empty()
515            && self.inner.q3.is_empty()
516    }
517}
518
519#[expect(clippy::indexing_slicing, reason = "the loop proves the edge index is in bounds")]
520#[expect(clippy::arithmetic_side_effects, reason = "the loop proves the cursor remains within the vector")]
521#[expect(clippy::ptr_arg, reason = "Verus sequence-view contracts are stated over Vec in this checked boundary")]
522fn step_edges_valid(edges: &Vec<(usize, usize)>, num_nodes: usize) -> (valid: bool)
523    ensures valid == StepGraphCarrier::edges_valid(edges@, num_nodes),
524{
525    let mut index = 0;
526    while index < edges.len()
527        invariant
528            index <= edges.len(),
529            forall|i: int| 0 <= i < index ==>
530                #[trigger] edges@[i].0 < num_nodes && edges@[i].1 < num_nodes,
531        decreases edges.len() - index,
532    {
533        if edges[index].0 >= num_nodes || edges[index].1 >= num_nodes {
534            assert(!StepGraphCarrier::edges_valid(edges@, num_nodes));
535            return false;
536        }
537        index = index + 1;
538    }
539    true
540}
541
542#[expect(clippy::indexing_slicing, reason = "the nested loops prove both edge indices are in bounds")]
543#[expect(clippy::arithmetic_side_effects, reason = "the loops prove both cursors remain within the vector")]
544#[expect(clippy::ptr_arg, reason = "Verus sequence-view contracts are stated over Vec in this checked boundary")]
545fn step_edges_distinct(edges: &Vec<(usize, usize)>) -> (distinct: bool)
546    ensures distinct == StepGraphCarrier::edges_distinct(edges@),
547{
548    let mut left = 0;
549    while left < edges.len()
550        invariant
551            left <= edges.len(),
552            forall|i: int, j: int| 0 <= i < left && 0 <= j < edges.len() && i != j
553                ==> #[trigger] edges@[i] != #[trigger] edges@[j],
554        decreases edges.len() - left,
555    {
556        let mut right = left + 1;
557        while right < edges.len()
558            invariant
559                left < edges.len(),
560                left + 1 <= right <= edges.len(),
561                forall|j: int| left < j < right ==> edges@[left as int] != edges@[j],
562            decreases edges.len() - right,
563        {
564            if edges[left].0 == edges[right].0 && edges[left].1 == edges[right].1 {
565                assert(!StepGraphCarrier::edges_distinct(edges@));
566                return false;
567            }
568            right = right + 1;
569        }
570        left = left + 1;
571    }
572    true
573}
574
575fn sequential_sentinel() -> (carrier: SequentialCarrier)
576    ensures carrier.inv(),
577{ SequentialCarrier::new(1, 1, 0) }
578
579fn fork_join_sentinel() -> (carrier: ForkJoinCarrier)
580    ensures carrier.inv(),
581{ ForkJoinCarrier::new(0, 1, 0) }
582
583fn step_graph_sentinel() -> (carrier: StepGraphCarrier)
584    ensures carrier.inv(),
585{
586    let edges: Vec<(usize, usize)> = Vec::new();
587    StepGraphCarrier::new(0, edges)
588}
589
590fn stream_graph_sentinel() -> (carrier: StreamGraphCarrier)
591    ensures carrier.inv(),
592{ StreamGraphCarrier::new(3, 1, 0, 1) }
593
594}
595
596impl Sequential {
597    /// Exclusive upper bound of carried values.
598    pub fn value_domain_size(&self) -> u64 {
599        self.inner.value_domain_size
600    }
601
602    /// Borrow completed-step values in execution order.
603    pub fn history_values(&self) -> &[u64] {
604        self.inner.history.as_slice()
605    }
606}
607
608impl ForkJoin {
609    /// Exclusive upper bound of worker values.
610    pub fn value_domain_size(&self) -> u64 {
611        self.inner.value_domain_size
612    }
613
614    /// Borrow worker lifecycle states by worker index.
615    pub fn worker_states(&self) -> &[WorkerState] {
616        self.inner.wstate.as_slice()
617    }
618
619    /// Borrow current worker values by worker index.
620    pub fn worker_values(&self) -> &[u64] {
621        self.inner.wvalue.as_slice()
622    }
623
624    /// Borrow the stable output snapshot after output production.
625    pub fn outputs(&self) -> Option<&[u64]> {
626        self.inner
627            .output_ready
628            .then_some(self.inner.output_snapshot.as_slice())
629    }
630}
631
632impl StepGraph {
633    /// Borrow directed predecessor edges in configured order.
634    pub fn edges(&self) -> &[(usize, usize)] {
635        self.inner.edges.as_slice()
636    }
637
638    /// Borrow all current step states by node index.
639    pub fn states(&self) -> &[StepState] {
640        self.inner.nstate.as_slice()
641    }
642}
643
644impl StreamGraph {
645    /// Maximum number of source records admitted by this run.
646    pub fn max_inputs(&self) -> usize {
647        self.inner.max_inputs
648    }
649
650    /// Exclusive upper bound of record values.
651    pub fn record_domain_size(&self) -> u64 {
652        self.inner.record_domain_size
653    }
654
655    /// Borrow records waiting at the first FIFO edge.
656    pub fn first_queue(&self) -> &[u64] {
657        self.inner.q1.values.as_slice()
658    }
659
660    /// Borrow records waiting at the second FIFO edge.
661    pub fn second_queue(&self) -> &[u64] {
662        self.inner.q2.values.as_slice()
663    }
664
665    /// Borrow records waiting at the optional third FIFO edge.
666    pub fn third_queue(&self) -> &[u64] {
667        self.inner.q3.values.as_slice()
668    }
669}
670
671impl_observational_debug!(Sequential, "Sequential",
672    "steps" => steps,
673    "completed" => completed,
674    "value" => value,
675    "active" => is_active,
676    "done" => is_done,
677);
678impl_observational_debug!(ForkJoin, "ForkJoin",
679    "len" => len,
680    "phase" => phase,
681    "output_ready" => output_ready,
682);
683impl_observational_debug!(StepGraph, "StepGraph",
684    "len" => len,
685    "edge_count" => edge_count,
686    "done" => is_done,
687);
688impl_observational_debug!(StreamGraph, "StreamGraph",
689    "chain_length" => chain_length,
690    "capacity" => capacity,
691    "ingested" => ingested,
692    "emitted" => emitted,
693    "first_queue_len" => first_queue_len,
694    "second_queue_len" => second_queue_len,
695    "third_queue_len" => third_queue_len,
696    "done" => is_done,
697);
698
699impl_public_error!(SequentialBuildError, {
700    Self::NoSteps => "sequential execution requires at least one step",
701    Self::EmptyValueDomain => "sequential value domain is empty",
702    Self::InitialValueOutOfRange => "initial sequential value is outside its domain",
703});
704impl_public_error!(ForkJoinBuildError, {
705    Self::EmptyValueDomain => "fork-join value domain is empty",
706    Self::InitialValueOutOfRange => "initial worker value is outside its domain",
707});
708impl_public_error!(StepGraphBuildError, {
709    Self::EdgeEndpointOutOfRange => "step-graph edge endpoint is outside the node universe",
710    Self::DuplicateEdge => "step graph contains a duplicate edge",
711});
712impl_public_error!(StreamGraphBuildError, {
713    Self::UnsupportedChainLength => "stream graph supports only three- or four-stage chains",
714    Self::ZeroCapacity => "stream-graph queue capacity must be positive",
715    Self::EmptyRecordDomain => "stream-graph record domain is empty",
716});
automation_structures/execution_api.rs

automation_structures/
execution_api.rs
