hopr_protocol_session/utils/
mod.rs

1pub mod skip_queue;
2
3#[cfg(test)]
4pub mod test;
5
6use std::{
7    cmp::Ordering,
8    sync::Arc,
9    time::{Duration, Instant},
10};
11
12use ringbuffer::{AllocRingBuffer, RingBuffer};
13
14use crate::{
15    errors::SessionError,
16    protocol::{FrameId, Segment, SeqIndicator, SeqNum},
17};
18
19#[derive(Clone)]
20pub(crate) struct RingBufferProducer<T>(Arc<parking_lot::FairMutex<AllocRingBuffer<T>>>);
21
22impl<T> std::fmt::Debug for RingBufferProducer<T> {
23    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
24        f.debug_tuple("RingBufferProducer").finish()
25    }
26}
27
28impl<T> RingBufferProducer<T> {
29    /// Pushes `item` into the ring buffer, returning `true` if there was capacity and `false`
30    /// if the buffer was already full and an existing entry was overwritten.
31    pub fn push(&mut self, item: T) -> bool {
32        let mut rb = self.0.lock();
33        let had_capacity = !rb.is_full();
34        rb.enqueue(item);
35        had_capacity
36    }
37}
38
39#[derive(Debug)]
40pub(crate) struct RingBufferView<T>(Arc<parking_lot::FairMutex<AllocRingBuffer<T>>>);
41
42impl<T> Clone for RingBufferView<T> {
43    fn clone(&self) -> Self {
44        Self(self.0.clone())
45    }
46}
47
48impl<T: Clone> RingBufferView<T> {
49    pub fn find<F: FnMut(&T) -> bool>(&self, mut predicate: F) -> Vec<T> {
50        self.0.lock().iter().filter(|item| predicate(item)).cloned().collect()
51    }
52}
53
54pub(crate) fn searchable_ringbuffer<T: Send + 'static>(capacity: usize) -> (RingBufferProducer<T>, RingBufferView<T>) {
55    let rb = Arc::new(parking_lot::FairMutex::new(AllocRingBuffer::new(capacity)));
56    (RingBufferProducer(rb.clone()), RingBufferView(rb))
57}
58
59const MAX_BACKOFF: Duration = Duration::from_secs(300);
60
61pub(crate) fn next_deadline_with_backoff(n: usize, base: f64, duration: Duration) -> Instant {
62    let backoff = duration.mul_f64(base.powi(n.min((i32::MAX / 2) as usize) as i32 + 1));
63    Instant::now() + backoff.min(MAX_BACKOFF)
64}
65
66#[derive(Debug, Copy, Clone, Eq)]
67pub(crate) struct RetriedFrameId {
68    pub frame_id: FrameId,
69    pub retry_count: usize,
70    max_retries: usize,
71    /// When the frame first entered the retry pipeline; carried across retries, so [`Self::age`]
72    /// measures total time in flight rather than time since the last resend.
73    first_sent: std::time::Instant,
74}
75
76impl RetriedFrameId {
77    pub fn no_retries(frame_id: FrameId) -> Self {
78        Self {
79            frame_id,
80            retry_count: 1,
81            max_retries: 1,
82            first_sent: std::time::Instant::now(),
83        }
84    }
85
86    pub fn with_retries(frame_id: FrameId, max_retries: usize) -> Self {
87        Self {
88            frame_id,
89            retry_count: 1,
90            max_retries,
91            first_sent: std::time::Instant::now(),
92        }
93    }
94
95    pub fn next(self) -> Option<Self> {
96        if self.retry_count < self.max_retries {
97            Some(Self {
98                frame_id: self.frame_id,
99                retry_count: self.retry_count + 1,
100                max_retries: self.max_retries,
101                first_sent: self.first_sent,
102            })
103        } else {
104            None
105        }
106    }
107
108    /// How long ago this frame first entered the retry pipeline.
109    pub fn age(&self) -> std::time::Duration {
110        self.first_sent.elapsed()
111    }
112}
113
114impl PartialEq<Self> for RetriedFrameId {
115    fn eq(&self, other: &Self) -> bool {
116        self.frame_id.eq(&other.frame_id)
117    }
118}
119
120impl PartialOrd<Self> for RetriedFrameId {
121    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
122        Some(self.cmp(other))
123    }
124}
125
126impl Ord for RetriedFrameId {
127    fn cmp(&self, other: &Self) -> Ordering {
128        self.frame_id.cmp(&other.frame_id)
129    }
130}
131
132/// Helper function to segment `data` into segments of a given ` max_segment_size ` length.
133/// All segments are tagged with the same `frame_id` and output into the given ` segments ` buffer.
134pub fn segment_into<T: AsRef<[u8]>, E: Extend<Segment>>(
135    data: T,
136    max_segment_size: usize,
137    frame_id: FrameId,
138    segments: &mut E,
139) -> crate::errors::Result<()> {
140    if frame_id == 0 {
141        return Err(SessionError::InvalidFrameId);
142    }
143
144    if max_segment_size == 0 {
145        return Err(SessionError::IncorrectMessageLength);
146    }
147
148    let data = data.as_ref();
149
150    let num_chunks = data.len().div_ceil(max_segment_size);
151    if num_chunks > SeqNum::MAX as usize {
152        return Err(SessionError::DataTooLong);
153    }
154
155    let chunks = data.chunks(max_segment_size);
156
157    let seq_len = SeqIndicator::try_from(chunks.len() as SeqNum)?;
158    segments.extend(chunks.enumerate().map(|(idx, data)| Segment {
159        frame_id,
160        seq_flags: seq_len,
161        seq_idx: idx as u8,
162        data: data.into(),
163    }));
164
165    Ok(())
166}
167
168/// Convenience wrapper for [`segment_into`] that allocates its own output buffer and returns it.
169#[allow(unused)]
170pub fn segment<T: AsRef<[u8]>>(data: T, max_segment_size: usize, frame_id: u32) -> crate::errors::Result<Vec<Segment>> {
171    let mut out = Vec::with_capacity(data.as_ref().len().div_ceil(max_segment_size));
172    segment_into(data, max_segment_size, frame_id, &mut out)?;
173    Ok(out)
174}
175
176#[cfg(test)]
177mod tests {
178    use hex_literal::hex;
179
180    use super::*;
181
182    #[test]
183    fn ring_buffer_producer_push_returns_false_when_overwriting() {
184        let (mut tx, _rx) = searchable_ringbuffer::<u32>(2);
185        assert!(tx.push(1));
186        assert!(tx.push(2));
187        assert!(!tx.push(3)); // buffer was full; oldest entry overwritten
188    }
189
190    #[test]
191    fn segment_should_split_data_correctly() -> anyhow::Result<()> {
192        let data = hex!("deadbeefcafebabe");
193
194        let segments = segment(data, 3, 1)?;
195        assert_eq!(3, segments.len());
196
197        assert_eq!(hex!("deadbe"), segments[0].data.as_ref());
198        assert_eq!(0, segments[0].seq_idx);
199        assert_eq!(3, segments[0].seq_flags.seq_len());
200        assert_eq!(1, segments[0].frame_id);
201
202        assert_eq!(hex!("efcafe"), segments[1].data.as_ref());
203        assert_eq!(1, segments[1].seq_idx);
204        assert_eq!(3, segments[1].seq_flags.seq_len());
205        assert_eq!(1, segments[1].frame_id);
206
207        assert_eq!(hex!("babe"), segments[2].data.as_ref());
208        assert_eq!(2, segments[2].seq_idx);
209        assert_eq!(3, segments[2].seq_flags.seq_len());
210        assert_eq!(1, segments[2].frame_id);
211
212        Ok(())
213    }
214}
hopr_protocol_session/utils/mod.rs

hopr_protocol_session/utils/
mod.rs
