matrix_sdk_common/
ring_buffer.rs

1// Copyright 2023 The Matrix.org Foundation C.I.C.
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::{
16    collections::{
17        VecDeque,
18        vec_deque::{Drain, Iter, IterMut},
19    },
20    num::NonZeroUsize,
21    ops::RangeBounds,
22};
23
24use serde::{Deserialize, Deserializer, Serialize};
25
26/// The capacity assigned to buffers written by older versions, which did not
27/// include the logical capacity in their serialized form.
28const LEGACY_DEFAULT_CAPACITY: NonZeroUsize = NonZeroUsize::new(10).unwrap();
29
30/// A simple fixed-size ring buffer implementation.
31///
32/// A size is provided on creation, and the ring buffer reserves that much
33/// space, and never reallocates. The logical capacity is included in the
34/// serialized representation so it survives a round trip.
35#[derive(Clone, Debug, PartialEq, Serialize)]
36pub struct RingBuffer<T> {
37    #[serde(rename = "items")]
38    inner: VecDeque<T>,
39
40    /// The capacity to restore for the backing `VecDeque` during
41    /// deserialization.
42    capacity: NonZeroUsize,
43}
44
45impl<T> RingBuffer<T> {
46    fn from_parts(mut inner: VecDeque<T>, capacity: NonZeroUsize) -> Self {
47        let capacity_as_usize = capacity.get();
48
49        for _ in capacity_as_usize..inner.len() {
50            inner.pop_front();
51        }
52
53        if let Some(extra_space) = capacity_as_usize.checked_sub(inner.len()) {
54            inner.reserve_exact(extra_space);
55        }
56
57        Self { inner, capacity }
58    }
59
60    /// Create a ring buffer with the supplied capacity, reserving it so we
61    /// never need to reallocate.
62    pub fn new(size: NonZeroUsize) -> Self {
63        Self::from_parts(VecDeque::with_capacity(size.into()), size)
64    }
65
66    /// Returns the number of items that are stored in this ring buffer.
67    ///
68    /// This is the dynamic size indicating how many items are held in the
69    /// buffer, not the fixed capacity.
70    pub fn len(&self) -> usize {
71        self.inner.len()
72    }
73
74    /// Returns true if the ring buffer contains no elements.
75    pub fn is_empty(&self) -> bool {
76        self.inner.is_empty()
77    }
78
79    /// Provides a reference to the element at the given index.
80    ///
81    /// Element at index zero is the "front" i.e. one that will be returned if
82    /// we call pop().
83    pub fn get(&self, index: usize) -> Option<&T> {
84        self.inner.get(index)
85    }
86
87    /// Appends an element to the back of the ring buffer, dropping elements
88    /// from the front if it is full.
89    pub fn push(&mut self, value: T) {
90        if self.inner.len() == self.inner.capacity() {
91            self.inner.pop_front();
92        }
93
94        self.inner.push_back(value);
95    }
96
97    /// Removes the first element and returns it, or None if the ring buffer is
98    /// empty.
99    pub fn pop(&mut self) -> Option<T> {
100        self.inner.pop_front()
101    }
102
103    /// Removes and returns one specific element at `index` if it exists,
104    /// otherwise it returns `None`.
105    pub fn remove(&mut self, index: usize) -> Option<T> {
106        self.inner.remove(index)
107    }
108
109    /// Returns an iterator that provides elements in front-to-back order, i.e.
110    /// the same order you would get if you repeatedly called pop().
111    pub fn iter(&self) -> Iter<'_, T> {
112        self.inner.iter()
113    }
114
115    /// Returns a mutable iterator that provides elements in front-to-back
116    /// order, i.e. the same order you would get if you repeatedly called pop().
117    pub fn iter_mut(&mut self) -> IterMut<'_, T> {
118        self.inner.iter_mut()
119    }
120
121    /// Returns an iterator that drains its items.
122    pub fn drain<R>(&mut self, range: R) -> Drain<'_, T>
123    where
124        R: RangeBounds<usize>,
125    {
126        self.inner.drain(range)
127    }
128
129    /// Clears the ring buffer, removing all values. This does not affect the
130    /// capacity.
131    pub fn clear(&mut self) {
132        self.inner.clear();
133    }
134
135    /// Returns the total number of elements the `RingBuffer` can hold.
136    pub fn capacity(&self) -> usize {
137        self.inner.capacity()
138    }
139
140    /// Retains only the elements specified by the predicate.
141    pub fn retain<F>(&mut self, predicate: F)
142    where
143        F: FnMut(&T) -> bool,
144    {
145        self.inner.retain(predicate)
146    }
147}
148
149// Migrate from the old RingBuffer type without the capacity field to the new
150// type.
151impl<'a, T: Deserialize<'a>> Deserialize<'a> for RingBuffer<T> {
152    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
153    where
154        D: Deserializer<'a>,
155    {
156        #[derive(Deserialize)]
157        #[serde(untagged)]
158        enum SerializedRingBuffer<T> {
159            WithCapacity { items: VecDeque<T>, capacity: NonZeroUsize },
160            Legacy(VecDeque<T>),
161        }
162
163        match SerializedRingBuffer::deserialize(deserializer)? {
164            SerializedRingBuffer::WithCapacity { items, capacity } => {
165                Ok(Self::from_parts(items, capacity))
166            }
167            SerializedRingBuffer::Legacy(items) => {
168                let capacity = NonZeroUsize::new(items.len().max(LEGACY_DEFAULT_CAPACITY.get()))
169                    .expect("legacy capacity is non-zero");
170                Ok(Self::from_parts(items, capacity))
171            }
172        }
173    }
174}
175
176impl<U> Extend<U> for RingBuffer<U> {
177    fn extend<T: IntoIterator<Item = U>>(&mut self, iter: T) {
178        for item in iter.into_iter() {
179            self.push(item);
180        }
181    }
182}
183
184#[cfg(test)]
185mod tests {
186    use std::{num::NonZeroUsize, ops::Not};
187
188    use super::RingBuffer;
189
190    #[test]
191    pub fn test_fixed_size() {
192        let mut ring_buffer = RingBuffer::new(NonZeroUsize::new(5).unwrap());
193
194        assert!(ring_buffer.is_empty());
195
196        ring_buffer.push(1);
197        ring_buffer.push(2);
198        ring_buffer.push(3);
199
200        assert!(ring_buffer.is_empty().not());
201
202        assert_eq!(ring_buffer.get(0), Some(&1));
203        assert_eq!(ring_buffer.get(1), Some(&2));
204        assert_eq!(ring_buffer.get(2), Some(&3));
205
206        ring_buffer.push(4);
207        ring_buffer.push(5);
208
209        assert_eq!(ring_buffer.get(0), Some(&1));
210        assert_eq!(ring_buffer.get(1), Some(&2));
211        assert_eq!(ring_buffer.get(2), Some(&3));
212        assert_eq!(ring_buffer.get(3), Some(&4));
213        assert_eq!(ring_buffer.get(4), Some(&5));
214
215        ring_buffer.push(6);
216
217        assert_eq!(ring_buffer.get(0), Some(&2));
218        assert_eq!(ring_buffer.get(1), Some(&3));
219        assert_eq!(ring_buffer.get(2), Some(&4));
220        assert_eq!(ring_buffer.get(3), Some(&5));
221        assert_eq!(ring_buffer.get(4), Some(&6));
222    }
223
224    #[test]
225    pub fn test_push_and_pop_and_remove_and_length() {
226        let mut ring_buffer = RingBuffer::new(NonZeroUsize::new(3).unwrap());
227
228        ring_buffer.push(1);
229        assert_eq!(ring_buffer.len(), 1);
230
231        ring_buffer.push(2);
232        assert_eq!(ring_buffer.len(), 2);
233
234        ring_buffer.push(3);
235        assert_eq!(ring_buffer.len(), 3);
236
237        assert_eq!(ring_buffer.pop(), Some(1));
238        assert_eq!(ring_buffer.len(), 2);
239        assert_eq!(ring_buffer.get(0), Some(&2));
240        assert_eq!(ring_buffer.get(1), Some(&3));
241        assert_eq!(ring_buffer.get(2), None);
242
243        assert_eq!(ring_buffer.pop(), Some(2));
244        assert_eq!(ring_buffer.len(), 1);
245        assert_eq!(ring_buffer.get(0), Some(&3));
246        assert_eq!(ring_buffer.get(1), None);
247        assert_eq!(ring_buffer.get(2), None);
248
249        assert_eq!(ring_buffer.pop(), Some(3));
250        assert_eq!(ring_buffer.len(), 0);
251        assert_eq!(ring_buffer.get(0), None);
252        assert_eq!(ring_buffer.get(1), None);
253        assert_eq!(ring_buffer.get(2), None);
254
255        assert_eq!(ring_buffer.pop(), None);
256
257        ring_buffer.push(1);
258        ring_buffer.push(2);
259        ring_buffer.push(3);
260        assert_eq!(ring_buffer.len(), 3);
261        assert_eq!(ring_buffer.get(0), Some(&1));
262        assert_eq!(ring_buffer.get(1), Some(&2));
263        assert_eq!(ring_buffer.get(2), Some(&3));
264
265        assert_eq!(ring_buffer.remove(1), Some(2));
266        assert_eq!(ring_buffer.len(), 2);
267        assert_eq!(ring_buffer.get(0), Some(&1));
268        assert_eq!(ring_buffer.get(1), Some(&3));
269        assert_eq!(ring_buffer.get(2), None);
270
271        assert_eq!(ring_buffer.remove(0), Some(1));
272        assert_eq!(ring_buffer.len(), 1);
273        assert_eq!(ring_buffer.get(0), Some(&3));
274        assert_eq!(ring_buffer.get(1), None);
275        assert_eq!(ring_buffer.get(2), None);
276
277        assert_eq!(ring_buffer.remove(1), None);
278        assert_eq!(ring_buffer.remove(10), None);
279    }
280
281    #[test]
282    fn test_iter() {
283        let mut ring_buffer = RingBuffer::new(NonZeroUsize::new(5).unwrap());
284
285        ring_buffer.push(1);
286        ring_buffer.push(2);
287        ring_buffer.push(3);
288
289        let as_vec = ring_buffer.iter().copied().collect::<Vec<_>>();
290        assert_eq!(as_vec, [1, 2, 3]);
291
292        let first_entry = ring_buffer.iter_mut().next().unwrap();
293        *first_entry = 42;
294
295        let as_vec = ring_buffer.iter().copied().collect::<Vec<_>>();
296        assert_eq!(as_vec, [42, 2, 3]);
297    }
298
299    #[test]
300    fn test_drain() {
301        let mut ring_buffer = RingBuffer::new(NonZeroUsize::new(5).unwrap());
302
303        ring_buffer.push(1);
304        ring_buffer.push(2);
305        ring_buffer.push(3);
306        ring_buffer.push(4);
307        ring_buffer.push(5);
308
309        let drained = ring_buffer.drain(0..=2).collect::<Vec<_>>();
310        let left = ring_buffer.iter().map(ToOwned::to_owned).collect::<Vec<_>>();
311
312        assert_eq!(drained, &[1, 2, 3]);
313        assert_eq!(left, &[4, 5]);
314
315        ring_buffer.drain(..);
316
317        assert!(ring_buffer.is_empty());
318    }
319
320    #[test]
321    fn test_clear_on_empty_buffer_is_a_noop() {
322        let mut ring_buffer: RingBuffer<u8> = RingBuffer::new(NonZeroUsize::new(3).unwrap());
323        ring_buffer.clear();
324        assert_eq!(ring_buffer.len(), 0);
325    }
326
327    #[test]
328    fn test_clear_removes_all_items() {
329        // Given a RingBuffer that has been used
330        let mut ring_buffer = RingBuffer::new(NonZeroUsize::new(3).unwrap());
331        ring_buffer.push(4);
332        ring_buffer.push(5);
333        ring_buffer.push(6);
334        ring_buffer.pop();
335        // Sanity: there are 2 items
336        assert_eq!(ring_buffer.len(), 2);
337
338        // When I clear it
339        ring_buffer.clear();
340
341        // Then it is empty
342        assert_eq!(ring_buffer.len(), 0);
343        assert_eq!(ring_buffer.get(0), None);
344        assert_eq!(ring_buffer.pop(), None);
345    }
346
347    #[test]
348    fn test_clear_does_not_affect_capacity() {
349        // Given a RingBuffer that has been used
350        let mut ring_buffer = RingBuffer::new(NonZeroUsize::new(3).unwrap());
351        ring_buffer.push(4);
352        ring_buffer.push(5);
353        ring_buffer.push(6);
354        ring_buffer.pop();
355        // Sanity: capacity is 3
356        assert_eq!(ring_buffer.capacity(), 3);
357
358        // When I clear it
359        ring_buffer.clear();
360
361        // Then its capacity is still 3
362        assert_eq!(ring_buffer.capacity(), 3);
363    }
364
365    #[test]
366    fn test_capacity_is_what_we_passed_to_new() {
367        // Given a RingBuffer
368        let ring_buffer = RingBuffer::<i32>::new(NonZeroUsize::new(13).unwrap());
369        // When I ask for its capacity I get what I provided at the start
370        assert_eq!(ring_buffer.capacity(), 13);
371    }
372
373    #[test]
374    fn test_capacity_is_not_affected_by_overflowing() {
375        // Given a RingBuffer that has been used
376        let mut ring_buffer = RingBuffer::new(NonZeroUsize::new(3).unwrap());
377        ring_buffer.push(4);
378        ring_buffer.push(5);
379        ring_buffer.push(6);
380        ring_buffer.push(7);
381        ring_buffer.pop();
382        ring_buffer.push(8);
383        ring_buffer.push(9);
384
385        // When I ask for its capacity, it gives me what I gave it initially
386        assert_eq!(ring_buffer.capacity(), 3);
387
388        // And even if I extend it
389        ring_buffer.extend(vec![10, 11, 12, 13, 14, 15]);
390
391        // Then its capacity is still 3
392        assert_eq!(ring_buffer.capacity(), 3);
393    }
394
395    #[test]
396    fn test_roundtrip_serialization() {
397        // Given a RingBuffer
398        let mut ring_buffer = RingBuffer::new(NonZeroUsize::new(3).unwrap());
399        ring_buffer.push("1".to_owned());
400        ring_buffer.push("2".to_owned());
401
402        // When I serialize it
403        let json = serde_json::to_string(&ring_buffer).expect("serialisation failed");
404        // Sanity: the JSON includes the logical capacity
405        assert_eq!(json, r#"{"items":["1","2"],"capacity":3}"#);
406
407        // And deserialize it
408        let new_ring_buffer: RingBuffer<String> =
409            serde_json::from_str(&json).expect("deserialisation failed");
410
411        // Then I get back the same items and capacity I started with
412        assert_eq!(ring_buffer, new_ring_buffer);
413        assert_eq!(new_ring_buffer.capacity(), 3);
414        assert_eq!(new_ring_buffer.inner.capacity(), 3);
415    }
416
417    #[test]
418    fn test_deserializes_the_legacy_sequence_format() {
419        let mut ring_buffer: RingBuffer<i32> = serde_json::from_str("[1,2]").unwrap();
420
421        assert_eq!(ring_buffer.iter().copied().collect::<Vec<_>>(), vec![1, 2]);
422        assert_eq!(ring_buffer.capacity(), 10);
423        assert_eq!(ring_buffer.inner.capacity(), 10);
424
425        // A partially filled legacy buffer must retain room for new items.
426        ring_buffer.push(3);
427        assert_eq!(ring_buffer.iter().copied().collect::<Vec<_>>(), vec![1, 2, 3]);
428    }
429
430    #[test]
431    fn test_extending_an_empty_ringbuffer_adds_the_items() {
432        // Given a RingBuffer
433        let mut ring_buffer = RingBuffer::new(NonZeroUsize::new(5).unwrap());
434
435        // When I extend it
436        ring_buffer.extend(vec!["a".to_owned(), "b".to_owned()]);
437
438        // Then the items are added
439        assert_eq!(ring_buffer.iter().map(String::as_str).collect::<Vec<_>>(), vec!["a", "b"]);
440    }
441
442    #[test]
443    fn test_extend_adds_items_to_the_end() {
444        // Given a RingBuffer with something in it
445        let mut ring_buffer = RingBuffer::new(NonZeroUsize::new(5).unwrap());
446        ring_buffer.push("1".to_owned());
447        ring_buffer.push("2".to_owned());
448
449        // When I extend it
450        ring_buffer.extend(vec!["3".to_owned(), "4".to_owned()]);
451
452        // Then the items are added on the end
453        assert_eq!(
454            ring_buffer.iter().map(String::as_str).collect::<Vec<_>>(),
455            vec!["1", "2", "3", "4"]
456        );
457    }
458
459    #[test]
460    fn test_extend_does_not_overflow_max_length() {
461        // Given a RingBuffer with something in it
462        let mut ring_buffer = RingBuffer::new(NonZeroUsize::new(5).unwrap());
463        ring_buffer.push("1".to_owned());
464        ring_buffer.push("2".to_owned());
465
466        // When I extend it with too many items
467        ring_buffer.extend(vec![
468            "3".to_owned(),
469            "4".to_owned(),
470            "5".to_owned(),
471            "6".to_owned(),
472            "7".to_owned(),
473        ]);
474
475        // Then some of previous items are gone, keeping the length to the max
476        assert_eq!(
477            ring_buffer.iter().map(String::as_str).collect::<Vec<_>>(),
478            vec!["3", "4", "5", "6", "7"]
479        );
480    }
481
482    #[test]
483    fn test_extending_a_full_ringbuffer_preserves_max_length() {
484        // Given a full RingBuffer with something in it
485        let mut ring_buffer = RingBuffer::new(NonZeroUsize::new(2).unwrap());
486        ring_buffer.push("1".to_owned());
487        ring_buffer.push("2".to_owned());
488
489        // When I extend it with lots of items
490        ring_buffer.extend(vec![
491            "3".to_owned(),
492            "4".to_owned(),
493            "5".to_owned(),
494            "6".to_owned(),
495            "7".to_owned(),
496        ]);
497
498        // Then only the last N items remain
499        assert_eq!(ring_buffer.iter().map(String::as_str).collect::<Vec<_>>(), vec!["6", "7"]);
500    }
501
502    #[test]
503    fn test_capacity_survives_a_serialization_round_trip() {
504        // Given a partially filled ring buffer with a non-default capacity,
505        let mut ring_buffer = RingBuffer::new(NonZeroUsize::new(3).unwrap());
506        ring_buffer.push(1);
507
508        // When it's deserialized back,
509        let json = serde_json::to_string(&ring_buffer).unwrap();
510        let mut ring_buffer: RingBuffer<i32> = serde_json::from_str(&json).unwrap();
511
512        // Then it still has its original capacity and room for two more items.
513        assert_eq!(ring_buffer.capacity(), 3);
514        assert_eq!(ring_buffer.inner.capacity(), 3);
515        ring_buffer.push(2);
516        ring_buffer.push(3);
517
518        assert_eq!(ring_buffer.iter().copied().collect::<Vec<_>>(), vec![1, 2, 3]);
519    }
520
521    #[test]
522    fn test_retain() {
523        let mut ring_buffer = RingBuffer::new(NonZeroUsize::new(2).unwrap());
524        ring_buffer.push(1);
525        ring_buffer.push(2);
526
527        ring_buffer.retain(|v| v % 2 == 0);
528
529        assert_eq!(ring_buffer.len(), 1);
530        assert_eq!(ring_buffer.get(0).copied().unwrap(), 2);
531    }
532}
matrix_sdk_common/ring_buffer.rs

matrix_sdk_common/
ring_buffer.rs
