sim_lib_stream_host/
ring.rs

1//! Preallocated bounded ring buffer for process-site stream boundaries.
2
3use sim_kernel::{Error, Result};
4use sim_lib_stream_core::StreamStats;
5
6/// Push result for the bounded process-site ring.
7#[derive(Clone, Debug, PartialEq, Eq)]
8pub enum ProcessRingPush<T> {
9    /// The item was stored in the ring.
10    Accepted,
11    /// The ring was full; the rejected item is returned.
12    DroppedNewest(T),
13    /// The ring was closed; the rejected item is returned.
14    Closed(T),
15}
16
17/// Capacity snapshot for steady-state process-site checks.
18#[derive(Clone, Copy, Debug, PartialEq, Eq)]
19pub struct ProcessRingSnapshot {
20    capacity: usize,
21    len: usize,
22    allocated_slots: usize,
23}
24
25/// Preallocated bounded ring used at process-site stream boundaries.
26#[derive(Clone, Debug)]
27pub struct ProcessSharedRing<T> {
28    slots: Vec<Option<T>>,
29    head: usize,
30    len: usize,
31    closed: bool,
32    stats: StreamStats,
33}
34
35impl<T> ProcessSharedRing<T> {
36    /// Creates a ring preallocated to hold `capacity` items.
37    ///
38    /// Returns an evaluation error when `capacity` is zero.
39    ///
40    /// # Examples
41    ///
42    /// ```
43    /// use sim_lib_stream_host::{ProcessRingPush, ProcessSharedRing};
44    ///
45    /// let mut ring = ProcessSharedRing::with_capacity(2).unwrap();
46    /// assert_eq!(ring.try_push(1), ProcessRingPush::Accepted);
47    /// assert_eq!(ring.try_push(2), ProcessRingPush::Accepted);
48    /// assert_eq!(ring.try_push(3), ProcessRingPush::DroppedNewest(3));
49    /// assert_eq!(ring.try_pop(), Some(1));
50    /// ```
51    pub fn with_capacity(capacity: usize) -> Result<Self> {
52        if capacity == 0 {
53            return Err(Error::Eval(
54                "process ring capacity must be greater than zero".to_owned(),
55            ));
56        }
57        let slots = std::iter::repeat_with(|| None).take(capacity).collect();
58        Ok(Self {
59            slots,
60            head: 0,
61            len: 0,
62            closed: false,
63            stats: StreamStats::default(),
64        })
65    }
66
67    /// Pushes an item, returning whether it was accepted, dropped (full), or
68    /// rejected (closed).
69    pub fn try_push(&mut self, item: T) -> ProcessRingPush<T> {
70        self.stats.pushed = self.stats.pushed.saturating_add(1);
71        if self.closed {
72            self.stats.closed = true;
73            return ProcessRingPush::Closed(item);
74        }
75        if self.len == self.capacity() {
76            self.stats.dropped_newest = self.stats.dropped_newest.saturating_add(1);
77            return ProcessRingPush::DroppedNewest(item);
78        }
79        let tail = (self.head + self.len) % self.capacity();
80        self.slots[tail] = Some(item);
81        self.len += 1;
82        self.stats.accepted = self.stats.accepted.saturating_add(1);
83        ProcessRingPush::Accepted
84    }
85
86    /// Pops the oldest item, or returns `None` when the ring is empty.
87    pub fn try_pop(&mut self) -> Option<T> {
88        if self.len == 0 {
89            return None;
90        }
91        let item = self.slots[self.head].take();
92        self.head = (self.head + 1) % self.capacity();
93        self.len -= 1;
94        if item.is_some() {
95            self.stats.yielded = self.stats.yielded.saturating_add(1);
96        }
97        item
98    }
99
100    /// Closes the ring; further pushes are rejected.
101    pub fn close(&mut self) {
102        self.closed = true;
103        self.stats.closed = true;
104    }
105
106    /// Drains all buffered items and closes the ring, recording cancellation.
107    pub fn cancel(&mut self) {
108        while self.try_pop().is_some() {}
109        self.closed = true;
110        self.stats.closed = true;
111        self.stats.cancelled = true;
112    }
113
114    /// Returns the number of buffered items.
115    pub fn len(&self) -> usize {
116        self.len
117    }
118
119    /// Returns whether the ring is empty.
120    pub fn is_empty(&self) -> bool {
121        self.len == 0
122    }
123
124    /// Returns the ring capacity in items.
125    pub fn capacity(&self) -> usize {
126        self.slots.len()
127    }
128
129    /// Returns the number of slots backing storage has allocated.
130    pub fn allocated_slots(&self) -> usize {
131        self.slots.capacity()
132    }
133
134    /// Returns whether the ring has been closed.
135    pub fn is_closed(&self) -> bool {
136        self.closed
137    }
138
139    /// Captures a capacity snapshot for steady-state checks.
140    pub fn snapshot(&self) -> ProcessRingSnapshot {
141        ProcessRingSnapshot {
142            capacity: self.capacity(),
143            len: self.len,
144            allocated_slots: self.allocated_slots(),
145        }
146    }
147
148    /// Returns a clone of the ring's running statistics.
149    pub fn stats(&self) -> StreamStats {
150        self.stats.clone()
151    }
152}
153
154impl ProcessRingSnapshot {
155    /// Returns the captured ring capacity.
156    pub fn capacity(self) -> usize {
157        self.capacity
158    }
159
160    /// Returns the captured buffered length.
161    pub fn len(self) -> usize {
162        self.len
163    }
164
165    /// Returns whether the ring was empty when captured.
166    pub fn is_empty(self) -> bool {
167        self.len == 0
168    }
169
170    /// Returns the captured number of allocated slots.
171    pub fn allocated_slots(self) -> usize {
172        self.allocated_slots
173    }
174}
sim_lib_stream_host/ring.rs

sim_lib_stream_host/
ring.rs
