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                use cu29::prelude::SaturatingSub;
121                let horizon_time = now.saturating_sub(self.stale_data_horizon);
122                // purge all the stale data from the TimeboundCircularBuffers first
123                $(self.$name.purge(horizon_time);)*
124            }
125
126            /// Get the most recent set of aligned data from all the buffers matching the constraints set at construction.
127            #[allow(dead_code)]
128            pub fn get_latest_aligned_data(
129                &mut self,
130            ) -> Option<($(impl Iterator<Item = &cu29::cutask::CuStampedData<$payload, CuMsgMetadata>>),*)> {
131                use cu29::prelude::SaturatingSub;
132                // Now find the min of the max of the last time for all buffers
133                // meaning the most recent time at which all buffers have data
134                let most_recent_time = [
135                    $(self.$name.most_recent_time().unwrap_or(None)),*
136                ]
137                .into_iter()
138                .flatten()
139                .min()?;
140
141                let time_to_get_complete_window =
142                    most_recent_time.saturating_sub(self.target_alignment_window);
143                Some(($(self.$name.iter_window(time_to_get_complete_window, most_recent_time)),*))
144            }
145        }
146    };
147}
148
149pub use alignment_buffers;
150
151#[cfg(test)]
152mod tests {
153    use cu29::clock::Tov;
154    use cu29::cutask::*;
155    use std::time::Duration;
156
157    #[test]
158    fn simple_init_test() {
159        alignment_buffers!(AlignmentBuffers, buffer1: TimeboundCircularBuffer<10, CuStampedData<u32, CuMsgMetadata>>, buffer2: TimeboundCircularBuffer<12, CuStampedData<u64, CuMsgMetadata>>);
160
161        let buffers =
162            AlignmentBuffers::new(Duration::from_secs(1).into(), Duration::from_secs(2).into());
163        assert_eq!(buffers.buffer1.inner.capacity(), 10);
164        assert_eq!(buffers.buffer2.inner.capacity(), 12);
165    }
166
167    #[test]
168    fn purge_test() {
169        alignment_buffers!(AlignmentBuffers, buffer1: TimeboundCircularBuffer<10, CuStampedData<u32, CuMsgMetadata>>, buffer2: TimeboundCircularBuffer<12, CuStampedData<u32, CuMsgMetadata>>);
170
171        let mut buffers =
172            AlignmentBuffers::new(Duration::from_secs(1).into(), Duration::from_secs(2).into());
173
174        let mut msg1 = CuStampedData::new(Some(1));
175        msg1.tov = Tov::Time(Duration::from_secs(1).into());
176        buffers.buffer1.inner.push_back(msg1.clone());
177        buffers.buffer2.inner.push_back(msg1);
178        // within the horizon
179        buffers.purge(Duration::from_secs(2).into());
180        assert_eq!(buffers.buffer1.inner.len(), 1);
181        assert_eq!(buffers.buffer2.inner.len(), 1);
182        // outside the horizon
183        buffers.purge(Duration::from_secs(5).into());
184        assert_eq!(buffers.buffer1.inner.len(), 0);
185        assert_eq!(buffers.buffer2.inner.len(), 0);
186    }
187
188    #[test]
189    fn empty_buffers_test() {
190        alignment_buffers!(
191            AlignmentBuffers,
192            buffer1: TimeboundCircularBuffer<10, CuStampedData<u32, CuMsgMetadata>>,
193            buffer2: TimeboundCircularBuffer<12, CuStampedData<u32, CuMsgMetadata>>
194        );
195
196        let mut buffers = AlignmentBuffers::new(
197            Duration::from_secs(2).into(), // 2-second alignment window
198            Duration::from_secs(5).into(), // 5-second stale data horizon
199        );
200
201        // Advance time to 10 seconds
202        assert!(buffers.get_latest_aligned_data().is_none());
203    }
204
205    #[test]
206    fn horizon_and_window_alignment_test() {
207        alignment_buffers!(
208            AlignmentBuffers,
209            buffer1: TimeboundCircularBuffer<10, CuStampedData<u32, CuMsgMetadata>>,
210            buffer2: TimeboundCircularBuffer<12, CuStampedData<u32, CuMsgMetadata>>
211        );
212
213        let mut buffers = AlignmentBuffers::new(
214            Duration::from_secs(2).into(), // 2-second alignment window
215            Duration::from_secs(5).into(), // 5-second stale data horizon
216        );
217
218        // Insert messages with timestamps
219        let mut msg1 = CuStampedData::new(Some(1));
220        msg1.tov = Tov::Time(Duration::from_secs(1).into());
221        buffers.buffer1.inner.push_back(msg1.clone());
222        buffers.buffer2.inner.push_back(msg1);
223
224        let mut msg2 = CuStampedData::new(Some(3));
225        msg2.tov = Tov::Time(Duration::from_secs(3).into());
226        buffers.buffer2.inner.push_back(msg2);
227
228        let mut msg3 = CuStampedData::new(Some(4));
229        msg3.tov = Tov::Time(Duration::from_secs(4).into());
230        buffers.buffer1.inner.push_back(msg3.clone());
231        buffers.buffer2.inner.push_back(msg3);
232
233        // Advance time to 7 seconds; horizon is 7 - 5 = everything 2+ should stay
234        let now = Duration::from_secs(7).into();
235        // Emulate a normal workflow here.
236        buffers.purge(now);
237        if let Some((iter1, iter2)) = buffers.get_latest_aligned_data() {
238            let collected1: Vec<_> = iter1.collect();
239            let collected2: Vec<_> = iter2.collect();
240
241            // Verify only messages within the alignment window [5, 7] are returned
242            assert_eq!(collected1.len(), 1);
243            assert_eq!(collected2.len(), 2);
244
245            assert_eq!(collected1[0].payload(), Some(&4));
246            assert_eq!(collected2[0].payload(), Some(&3));
247            assert_eq!(collected2[1].payload(), Some(&4));
248        } else {
249            panic!("Expected aligned data, but got None");
250        }
251
252        // Ensure older messages outside the horizon [>2 seconds] are purged
253        assert_eq!(buffers.buffer1.inner.len(), 1);
254        assert_eq!(buffers.buffer2.inner.len(), 2);
255    }
256}
cu_aligner/buffers.rs

cu_aligner/
buffers.rs
