spatialrust_runtime/
pipeline.rs

1//! Bounded execution pipelines with explicit capacity.
2
3use crate::{RuntimeError, RuntimeResult, TraceEvent, TraceLevel, TraceLog};
4
5/// Named pipeline stage.
6#[derive(Clone, Debug, PartialEq, Eq, Hash)]
7pub struct PipelineStage(pub String);
8
9impl PipelineStage {
10    /// Creates a stage name.
11    #[must_use]
12    pub fn new(value: impl Into<String>) -> Self {
13        Self(value.into())
14    }
15}
16
17/// Pipeline capacity / timeout configuration.
18#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
19pub struct PipelineConfig {
20    /// Maximum in-flight messages.
21    pub max_inflight: usize,
22}
23
24impl Default for PipelineConfig {
25    fn default() -> Self {
26        Self { max_inflight: 8 }
27    }
28}
29
30/// Bounded FIFO pipeline with tracing.
31#[derive(Clone, Debug)]
32pub struct BoundedPipeline<T> {
33    config: PipelineConfig,
34    queue: Vec<(PipelineStage, T)>,
35    trace: TraceLog,
36}
37
38impl<T> BoundedPipeline<T> {
39    /// Creates an empty pipeline.
40    #[must_use]
41    pub fn new(config: PipelineConfig) -> Self {
42        Self { config, queue: Vec::new(), trace: TraceLog::new() }
43    }
44
45    /// Enqueues a message or rejects when at capacity.
46    pub fn push(&mut self, stage: PipelineStage, value: T) -> RuntimeResult<()> {
47        if self.queue.len() >= self.config.max_inflight {
48            self.trace.push(TraceEvent {
49                level: TraceLevel::Error,
50                stage: stage.0.clone(),
51                message: "capacity exceeded".into(),
52            });
53            return Err(RuntimeError::CapacityExceeded(stage.0));
54        }
55        self.trace.push(TraceEvent {
56            level: TraceLevel::Info,
57            stage: stage.0.clone(),
58            message: "enqueued".into(),
59        });
60        self.queue.push((stage, value));
61        Ok(())
62    }
63
64    /// Pops the next message.
65    pub fn pop(&mut self) -> Option<(PipelineStage, T)> {
66        if self.queue.is_empty() {
67            None
68        } else {
69            Some(self.queue.remove(0))
70        }
71    }
72
73    /// Returns the trace log.
74    #[must_use]
75    pub fn trace(&self) -> &TraceLog {
76        &self.trace
77    }
78
79    /// Returns in-flight count.
80    #[must_use]
81    pub fn len(&self) -> usize {
82        self.queue.len()
83    }
84
85    /// Returns whether the pipeline is empty.
86    #[must_use]
87    pub fn is_empty(&self) -> bool {
88        self.queue.is_empty()
89    }
90}
91
92#[cfg(test)]
93mod tests {
94    use super::{BoundedPipeline, PipelineConfig, PipelineStage};
95
96    #[test]
97    fn rejects_when_full() {
98        let mut pipe = BoundedPipeline::new(PipelineConfig { max_inflight: 1 });
99        pipe.push(PipelineStage::new("a"), 1).unwrap();
100        assert!(pipe.push(PipelineStage::new("b"), 2).is_err());
101    }
102}
spatialrust_runtime/pipeline.rs

spatialrust_runtime/
pipeline.rs
