cu_aligner/
buffers.rs

1use circular_buffer::CircularBuffer;
2use cu29::prelude::*;
3
4/// An augmented circular buffer that allows for time-based operations.
5pub struct TimeboundCircularBuffer<const S: usize, P, M>
6where
7    P: CuMsgPayload,
8    M: Metadata,
9{
10    pub inner: CircularBuffer<S, CuStampedData<P, M>>,
11}
12
13#[allow(dead_code)]
14fn extract_tov_time_left(tov: &Tov) -> Option<CuTime> {
15    match tov {
16        Tov::Time(time) => Some(*time),
17        Tov::Range(range) => Some(range.start), // Use the start of the range for alignment
18        Tov::None => None,
19    }
20}
21
22fn extract_tov_time_right(tov: &Tov) -> Option<CuTime> {
23    match tov {
24        Tov::Time(time) => Some(*time),
25        Tov::Range(range) => Some(range.end), // Use the end of the range for alignment
26        Tov::None => None,
27    }
28}
29
30impl<const S: usize, P> Default for TimeboundCircularBuffer<S, P, CuMsgMetadata>
31where
32    P: CuMsgPayload,
33{
34    fn default() -> Self {
35        Self::new()
36    }
37}
38
39impl<const S: usize, P> TimeboundCircularBuffer<S, P, CuMsgMetadata>
40where
41    P: CuMsgPayload,
42{
43    pub fn new() -> Self {
44        Self {
45            // It is assumed to be sorted by time with non overlapping ranges if they are Tov::Range
46            inner: CircularBuffer::<S, CuStampedData<P, CuMsgMetadata>>::new(),
47        }
48    }
49
50    /// Gets a slice of messages that fall within the given time range.
51    /// In case of a Tov::Range, the message is included if its start and end time fall within the range.
52    pub fn iter_window(
53        &self,
54        start_time: CuTime,
55        end_time: CuTime,
56    ) -> impl Iterator<Item = &CuStampedData<P, CuMsgMetadata>> {
57        self.inner.iter().filter(move |msg| match msg.tov {
58            Tov::Time(time) => time >= start_time && time <= end_time,
59            Tov::Range(range) => range.start >= start_time && range.end <= end_time,
60            _ => false,
61        })
62    }
63
64    /// Remove all the messages that are older than the given time horizon.
65    pub fn purge(&mut self, time_horizon: CuTime) {
66        // Find the index of the first element that should be retained
67        let drain_end = self
68            .inner
69            .iter()
70            .position(|msg| match msg.tov {
71                Tov::Time(time) => time >= time_horizon,
72                Tov::Range(range) => range.end >= time_horizon,
73                _ => false,
74            })
75            .unwrap_or(self.inner.len()); // If none match, drain the entire buffer
76
77        // Drain all elements before the `drain_end` index
78        self.inner.drain(..drain_end);
79    }
80
81    /// Get the most recent time of the messages in the buffer.
82    pub fn most_recent_time(&self) -> CuResult<Option<CuTime>> {
83        let mut latest: Option<CuTime> = None;
84        for msg in self.inner.iter() {
85            let time = extract_tov_time_right(&msg.tov).ok_or_else(|| {
86                CuError::from("Trying to align temporal data with no time information")
87            })?;
88            latest = Some(latest.map_or(time, |current_max| current_max.max(time)));
89        }
90        Ok(latest)
91    }
92
93    /// Push a message into the buffer.
94    pub fn push(&mut self, msg: CuStampedData<P, CuMsgMetadata>) {
95        self.inner.push_back(msg);
96    }
97}
98
99#[macro_export]
100macro_rules! alignment_buffers {
101    ($struct_name:ident, $($name:ident: TimeboundCircularBuffer<$size:expr, CuStampedData<$payload:ty, CuMsgMetadata>>),*) => {
102        struct $struct_name {
103            target_alignment_window: cu29::clock::CuDuration, // size of the most recent data window to align
104            stale_data_horizon: cu29::clock::CuDuration,  // time horizon for purging stale data
105            $(pub $name: $crate::buffers::TimeboundCircularBuffer<$size, $payload, CuMsgMetadata>),*
106        }
107
108        impl $struct_name {
109            pub fn new(target_alignment_window: cu29::clock::CuDuration, stale_data_horizon: cu29::clock::CuDuration) -> Self {
110                Self {
111                    target_alignment_window,
112                    stale_data_horizon,
113                    $($name: $crate::buffers::TimeboundCircularBuffer::<$size, $payload, CuMsgMetadata>::new()),*
114                }
115            }
116
117            /// Call this to be sure we discard the old/ non relevant data
118            #[allow(dead_code)]
119            pub fn purge(&mut self, now: cu29::clock::CuTime) {
120                let horizon_time = now - self.stale_data_horizon;
121                // purge all the stale data from the TimeboundCircularBuffers first
122                $(self.$name.purge(horizon_time);)*
123            }
124
125            /// Get the most recent set of aligned data from all the buffers matching the constraints set at construction.
126            #[allow(dead_code)]
127            pub fn get_latest_aligned_data(
128                &mut self,
129            ) -> Option<($(impl Iterator<Item = &cu29::cutask::CuStampedData<$payload, CuMsgMetadata>>),*)> {
130                // Now find the min of the max of the last time for all buffers
131                // meaning the most recent time at which all buffers have data
132                let most_recent_time = [
133                    $(self.$name.most_recent_time().unwrap_or(None)),*
134                ]
135                .into_iter()
136                .flatten()
137                .min()?;
138
139                let time_to_get_complete_window = most_recent_time - self.target_alignment_window;
140                Some(($(self.$name.iter_window(time_to_get_complete_window, most_recent_time)),*))
141            }
142        }
143    };
144}
145
146pub use alignment_buffers;
147
148#[cfg(test)]
149mod tests {
150    use cu29::clock::Tov;
151    use cu29::cutask::*;
152    use std::time::Duration;
153
154    #[test]
155    fn simple_init_test() {
156        alignment_buffers!(AlignmentBuffers, buffer1: TimeboundCircularBuffer<10, CuStampedData<u32, CuMsgMetadata>>, buffer2: TimeboundCircularBuffer<12, CuStampedData<u64, CuMsgMetadata>>);
157
158        let buffers =
159            AlignmentBuffers::new(Duration::from_secs(1).into(), Duration::from_secs(2).into());
160        assert_eq!(buffers.buffer1.inner.capacity(), 10);
161        assert_eq!(buffers.buffer2.inner.capacity(), 12);
162    }
163
164    #[test]
165    fn purge_test() {
166        alignment_buffers!(AlignmentBuffers, buffer1: TimeboundCircularBuffer<10, CuStampedData<u32, CuMsgMetadata>>, buffer2: TimeboundCircularBuffer<12, CuStampedData<u32, CuMsgMetadata>>);
167
168        let mut buffers =
169            AlignmentBuffers::new(Duration::from_secs(1).into(), Duration::from_secs(2).into());
170
171        let mut msg1 = CuStampedData::new(Some(1));
172        msg1.tov = Tov::Time(Duration::from_secs(1).into());
173        buffers.buffer1.inner.push_back(msg1.clone());
174        buffers.buffer2.inner.push_back(msg1);
175        // within the horizon
176        buffers.purge(Duration::from_secs(2).into());
177        assert_eq!(buffers.buffer1.inner.len(), 1);
178        assert_eq!(buffers.buffer2.inner.len(), 1);
179        // outside the horizon
180        buffers.purge(Duration::from_secs(5).into());
181        assert_eq!(buffers.buffer1.inner.len(), 0);
182        assert_eq!(buffers.buffer2.inner.len(), 0);
183    }
184
185    #[test]
186    fn empty_buffers_test() {
187        alignment_buffers!(
188            AlignmentBuffers,
189            buffer1: TimeboundCircularBuffer<10, CuStampedData<u32, CuMsgMetadata>>,
190            buffer2: TimeboundCircularBuffer<12, CuStampedData<u32, CuMsgMetadata>>
191        );
192
193        let mut buffers = AlignmentBuffers::new(
194            Duration::from_secs(2).into(), // 2-second alignment window
195            Duration::from_secs(5).into(), // 5-second stale data horizon
196        );
197
198        // Advance time to 10 seconds
199        assert!(buffers.get_latest_aligned_data().is_none());
200    }
201
202    #[test]
203    fn horizon_and_window_alignment_test() {
204        alignment_buffers!(
205            AlignmentBuffers,
206            buffer1: TimeboundCircularBuffer<10, CuStampedData<u32, CuMsgMetadata>>,
207            buffer2: TimeboundCircularBuffer<12, CuStampedData<u32, CuMsgMetadata>>
208        );
209
210        let mut buffers = AlignmentBuffers::new(
211            Duration::from_secs(2).into(), // 2-second alignment window
212            Duration::from_secs(5).into(), // 5-second stale data horizon
213        );
214
215        // Insert messages with timestamps
216        let mut msg1 = CuStampedData::new(Some(1));
217        msg1.tov = Tov::Time(Duration::from_secs(1).into());
218        buffers.buffer1.inner.push_back(msg1.clone());
219        buffers.buffer2.inner.push_back(msg1);
220
221        let mut msg2 = CuStampedData::new(Some(3));
222        msg2.tov = Tov::Time(Duration::from_secs(3).into());
223        buffers.buffer2.inner.push_back(msg2);
224
225        let mut msg3 = CuStampedData::new(Some(4));
226        msg3.tov = Tov::Time(Duration::from_secs(4).into());
227        buffers.buffer1.inner.push_back(msg3.clone());
228        buffers.buffer2.inner.push_back(msg3);
229
230        // Advance time to 7 seconds; horizon is 7 - 5 = everything 2+ should stay
231        let now = Duration::from_secs(7).into();
232        // Emulate a normal workflow here.
233        buffers.purge(now);
234        if let Some((iter1, iter2)) = buffers.get_latest_aligned_data() {
235            let collected1: Vec<_> = iter1.collect();
236            let collected2: Vec<_> = iter2.collect();
237
238            // Verify only messages within the alignment window [5, 7] are returned
239            assert_eq!(collected1.len(), 1);
240            assert_eq!(collected2.len(), 2);
241
242            assert_eq!(collected1[0].payload(), Some(&4));
243            assert_eq!(collected2[0].payload(), Some(&3));
244            assert_eq!(collected2[1].payload(), Some(&4));
245        } else {
246            panic!("Expected aligned data, but got None");
247        }
248
249        // Ensure older messages outside the horizon [>2 seconds] are purged
250        assert_eq!(buffers.buffer1.inner.len(), 1);
251        assert_eq!(buffers.buffer2.inner.len(), 2);
252    }
253}
cu_aligner/buffers.rs

cu_aligner/
buffers.rs
