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
117    /// pop().
118    pub fn iter_mut(&mut self) -> IterMut<'_, T> {
119        self.inner.iter_mut()
120    }
121
122    /// Returns an iterator that drains its items.
123    pub fn drain<R>(&mut self, range: R) -> Drain<'_, T>
124    where
125        R: RangeBounds<usize>,
126    {
127        self.inner.drain(range)
128    }
129
130    /// Clears the ring buffer, removing all values. This does not affect the
131    /// capacity.
132    pub fn clear(&mut self) {
133        self.inner.clear();
134    }
135
136    /// Returns the total number of elements the `RingBuffer` can hold.
137    pub fn capacity(&self) -> usize {
138        self.inner.capacity()
139    }
140
141    /// Retains only the elements specified by the predicate.
142    pub fn retain<F>(&mut self, predicate: F)
143    where
144        F: FnMut(&T) -> bool,
145    {
146        self.inner.retain(predicate)
147    }
148}
149
150// Migrate from the old RingBuffer type without the capacity field to the new
151// type.
152impl<'a, T: Deserialize<'a>> Deserialize<'a> for RingBuffer<T> {
153    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
154    where
155        D: Deserializer<'a>,
156    {
157        #[derive(Deserialize)]
158        #[serde(untagged)]
159        enum SerializedRingBuffer<T> {
160            WithCapacity { items: VecDeque<T>, capacity: NonZeroUsize },
161            Legacy(VecDeque<T>),
162        }
163
164        match SerializedRingBuffer::deserialize(deserializer)? {
165            SerializedRingBuffer::WithCapacity { items, capacity } => {
166                Ok(Self::from_parts(items, capacity))
167            }
168            SerializedRingBuffer::Legacy(items) => {
169                let capacity = NonZeroUsize::new(items.len().max(LEGACY_DEFAULT_CAPACITY.get()))
170                    .expect("legacy capacity is non-zero");
171                Ok(Self::from_parts(items, capacity))
172            }
173        }
174    }
175}
176
177impl<U> Extend<U> for RingBuffer<U> {
178    fn extend<T: IntoIterator<Item = U>>(&mut self, iter: T) {
179        for item in iter.into_iter() {
180            self.push(item);
181        }
182    }
183}
184
185#[cfg(test)]
186mod tests {
187    use std::{num::NonZeroUsize, ops::Not};
188
189    use super::RingBuffer;
190
191    #[test]
192    pub fn test_fixed_size() {
193        let mut ring_buffer = RingBuffer::new(NonZeroUsize::new(5).unwrap());
194
195        assert!(ring_buffer.is_empty());
196
197        ring_buffer.push(1);
198        ring_buffer.push(2);
199        ring_buffer.push(3);
200
201        assert!(ring_buffer.is_empty().not());
202
203        assert_eq!(ring_buffer.get(0), Some(&1));
204        assert_eq!(ring_buffer.get(1), Some(&2));
205        assert_eq!(ring_buffer.get(2), Some(&3));
206
207        ring_buffer.push(4);
208        ring_buffer.push(5);
209
210        assert_eq!(ring_buffer.get(0), Some(&1));
211        assert_eq!(ring_buffer.get(1), Some(&2));
212        assert_eq!(ring_buffer.get(2), Some(&3));
213        assert_eq!(ring_buffer.get(3), Some(&4));
214        assert_eq!(ring_buffer.get(4), Some(&5));
215
216        ring_buffer.push(6);
217
218        assert_eq!(ring_buffer.get(0), Some(&2));
219        assert_eq!(ring_buffer.get(1), Some(&3));
220        assert_eq!(ring_buffer.get(2), Some(&4));
221        assert_eq!(ring_buffer.get(3), Some(&5));
222        assert_eq!(ring_buffer.get(4), Some(&6));
223    }
224
225    #[test]
226    pub fn test_push_and_pop_and_remove_and_length() {
227        let mut ring_buffer = RingBuffer::new(NonZeroUsize::new(3).unwrap());
228
229        ring_buffer.push(1);
230        assert_eq!(ring_buffer.len(), 1);
231
232        ring_buffer.push(2);
233        assert_eq!(ring_buffer.len(), 2);
234
235        ring_buffer.push(3);
236        assert_eq!(ring_buffer.len(), 3);
237
238        assert_eq!(ring_buffer.pop(), Some(1));
239        assert_eq!(ring_buffer.len(), 2);
240        assert_eq!(ring_buffer.get(0), Some(&2));
241        assert_eq!(ring_buffer.get(1), Some(&3));
242        assert_eq!(ring_buffer.get(2), None);
243
244        assert_eq!(ring_buffer.pop(), Some(2));
245        assert_eq!(ring_buffer.len(), 1);
246        assert_eq!(ring_buffer.get(0), Some(&3));
247        assert_eq!(ring_buffer.get(1), None);
248        assert_eq!(ring_buffer.get(2), None);
249
250        assert_eq!(ring_buffer.pop(), Some(3));
251        assert_eq!(ring_buffer.len(), 0);
252        assert_eq!(ring_buffer.get(0), None);
253        assert_eq!(ring_buffer.get(1), None);
254        assert_eq!(ring_buffer.get(2), None);
255
256        assert_eq!(ring_buffer.pop(), None);
257
258        ring_buffer.push(1);
259        ring_buffer.push(2);
260        ring_buffer.push(3);
261        assert_eq!(ring_buffer.len(), 3);
262        assert_eq!(ring_buffer.get(0), Some(&1));
263        assert_eq!(ring_buffer.get(1), Some(&2));
264        assert_eq!(ring_buffer.get(2), Some(&3));
265
266        assert_eq!(ring_buffer.remove(1), Some(2));
267        assert_eq!(ring_buffer.len(), 2);
268        assert_eq!(ring_buffer.get(0), Some(&1));
269        assert_eq!(ring_buffer.get(1), Some(&3));
270        assert_eq!(ring_buffer.get(2), None);
271
272        assert_eq!(ring_buffer.remove(0), Some(1));
273        assert_eq!(ring_buffer.len(), 1);
274        assert_eq!(ring_buffer.get(0), Some(&3));
275        assert_eq!(ring_buffer.get(1), None);
276        assert_eq!(ring_buffer.get(2), None);
277
278        assert_eq!(ring_buffer.remove(1), None);
279        assert_eq!(ring_buffer.remove(10), None);
280    }
281
282    #[test]
283    fn test_iter() {
284        let mut ring_buffer = RingBuffer::new(NonZeroUsize::new(5).unwrap());
285
286        ring_buffer.push(1);
287        ring_buffer.push(2);
288        ring_buffer.push(3);
289
290        let as_vec = ring_buffer.iter().copied().collect::<Vec<_>>();
291        assert_eq!(as_vec, [1, 2, 3]);
292
293        let first_entry = ring_buffer.iter_mut().next().unwrap();
294        *first_entry = 42;
295
296        let as_vec = ring_buffer.iter().copied().collect::<Vec<_>>();
297        assert_eq!(as_vec, [42, 2, 3]);
298    }
299
300    #[test]
301    fn test_drain() {
302        let mut ring_buffer = RingBuffer::new(NonZeroUsize::new(5).unwrap());
303
304        ring_buffer.push(1);
305        ring_buffer.push(2);
306        ring_buffer.push(3);
307        ring_buffer.push(4);
308        ring_buffer.push(5);
309
310        let drained = ring_buffer.drain(0..=2).collect::<Vec<_>>();
311        let left = ring_buffer.iter().map(ToOwned::to_owned).collect::<Vec<_>>();
312
313        assert_eq!(drained, &[1, 2, 3]);
314        assert_eq!(left, &[4, 5]);
315
316        ring_buffer.drain(..);
317
318        assert!(ring_buffer.is_empty());
319    }
320
321    #[test]
322    fn test_clear_on_empty_buffer_is_a_noop() {
323        let mut ring_buffer: RingBuffer<u8> = RingBuffer::new(NonZeroUsize::new(3).unwrap());
324        ring_buffer.clear();
325        assert_eq!(ring_buffer.len(), 0);
326    }
327
328    #[test]
329    fn test_clear_removes_all_items() {
330        // Given a RingBuffer that has been used
331        let mut ring_buffer = RingBuffer::new(NonZeroUsize::new(3).unwrap());
332        ring_buffer.push(4);
333        ring_buffer.push(5);
334        ring_buffer.push(6);
335        ring_buffer.pop();
336        // Sanity: there are 2 items
337        assert_eq!(ring_buffer.len(), 2);
338
339        // When I clear it
340        ring_buffer.clear();
341
342        // Then it is empty
343        assert_eq!(ring_buffer.len(), 0);
344        assert_eq!(ring_buffer.get(0), None);
345        assert_eq!(ring_buffer.pop(), None);
346    }
347
348    #[test]
349    fn test_clear_does_not_affect_capacity() {
350        // Given a RingBuffer that has been used
351        let mut ring_buffer = RingBuffer::new(NonZeroUsize::new(3).unwrap());
352        ring_buffer.push(4);
353        ring_buffer.push(5);
354        ring_buffer.push(6);
355        ring_buffer.pop();
356        // Sanity: capacity is 3
357        assert_eq!(ring_buffer.capacity(), 3);
358
359        // When I clear it
360        ring_buffer.clear();
361
362        // Then its capacity is still 3
363        assert_eq!(ring_buffer.capacity(), 3);
364    }
365
366    #[test]
367    fn test_capacity_is_what_we_passed_to_new() {
368        // Given a RingBuffer
369        let ring_buffer = RingBuffer::<i32>::new(NonZeroUsize::new(13).unwrap());
370        // When I ask for its capacity I get what I provided at the start
371        assert_eq!(ring_buffer.capacity(), 13);
372    }
373
374    #[test]
375    fn test_capacity_is_not_affected_by_overflowing() {
376        // Given a RingBuffer that has been used
377        let mut ring_buffer = RingBuffer::new(NonZeroUsize::new(3).unwrap());
378        ring_buffer.push(4);
379        ring_buffer.push(5);
380        ring_buffer.push(6);
381        ring_buffer.push(7);
382        ring_buffer.pop();
383        ring_buffer.push(8);
384        ring_buffer.push(9);
385
386        // When I ask for its capacity, it gives me what I gave it initially
387        assert_eq!(ring_buffer.capacity(), 3);
388
389        // And even if I extend it
390        ring_buffer.extend(vec![10, 11, 12, 13, 14, 15]);
391
392        // Then its capacity is still 3
393        assert_eq!(ring_buffer.capacity(), 3);
394    }
395
396    #[test]
397    fn test_roundtrip_serialization() {
398        // Given a RingBuffer
399        let mut ring_buffer = RingBuffer::new(NonZeroUsize::new(3).unwrap());
400        ring_buffer.push("1".to_owned());
401        ring_buffer.push("2".to_owned());
402
403        // When I serialize it
404        let json = serde_json::to_string(&ring_buffer).expect("serialisation failed");
405        // Sanity: the JSON includes the logical capacity
406        assert_eq!(json, r#"{"items":["1","2"],"capacity":3}"#);
407
408        // And deserialize it
409        let new_ring_buffer: RingBuffer<String> =
410            serde_json::from_str(&json).expect("deserialisation failed");
411
412        // Then I get back the same items and capacity I started with
413        assert_eq!(ring_buffer, new_ring_buffer);
414        assert_eq!(new_ring_buffer.capacity(), 3);
415        assert_eq!(new_ring_buffer.inner.capacity(), 3);
416    }
417
418    #[test]
419    fn test_deserializes_the_legacy_sequence_format() {
420        let mut ring_buffer: RingBuffer<i32> = serde_json::from_str("[1,2]").unwrap();
421
422        assert_eq!(ring_buffer.iter().copied().collect::<Vec<_>>(), vec![1, 2]);
423        assert_eq!(ring_buffer.capacity(), 10);
424        assert_eq!(ring_buffer.inner.capacity(), 10);
425
426        // A partially filled legacy buffer must retain room for new items.
427        ring_buffer.push(3);
428        assert_eq!(ring_buffer.iter().copied().collect::<Vec<_>>(), vec![1, 2, 3]);
429    }
430
431    #[test]
432    fn test_extending_an_empty_ringbuffer_adds_the_items() {
433        // Given a RingBuffer
434        let mut ring_buffer = RingBuffer::new(NonZeroUsize::new(5).unwrap());
435
436        // When I extend it
437        ring_buffer.extend(vec!["a".to_owned(), "b".to_owned()]);
438
439        // Then the items are added
440        assert_eq!(ring_buffer.iter().map(String::as_str).collect::<Vec<_>>(), vec!["a", "b"]);
441    }
442
443    #[test]
444    fn test_extend_adds_items_to_the_end() {
445        // Given a RingBuffer with something in it
446        let mut ring_buffer = RingBuffer::new(NonZeroUsize::new(5).unwrap());
447        ring_buffer.push("1".to_owned());
448        ring_buffer.push("2".to_owned());
449
450        // When I extend it
451        ring_buffer.extend(vec!["3".to_owned(), "4".to_owned()]);
452
453        // Then the items are added on the end
454        assert_eq!(
455            ring_buffer.iter().map(String::as_str).collect::<Vec<_>>(),
456            vec!["1", "2", "3", "4"]
457        );
458    }
459
460    #[test]
461    fn test_extend_does_not_overflow_max_length() {
462        // Given a RingBuffer with something in it
463        let mut ring_buffer = RingBuffer::new(NonZeroUsize::new(5).unwrap());
464        ring_buffer.push("1".to_owned());
465        ring_buffer.push("2".to_owned());
466
467        // When I extend it with too many items
468        ring_buffer.extend(vec![
469            "3".to_owned(),
470            "4".to_owned(),
471            "5".to_owned(),
472            "6".to_owned(),
473            "7".to_owned(),
474        ]);
475
476        // Then some of previous items are gone, keeping the length to the max
477        assert_eq!(
478            ring_buffer.iter().map(String::as_str).collect::<Vec<_>>(),
479            vec!["3", "4", "5", "6", "7"]
480        );
481    }
482
483    #[test]
484    fn test_extending_a_full_ringbuffer_preserves_max_length() {
485        // Given a full RingBuffer with something in it
486        let mut ring_buffer = RingBuffer::new(NonZeroUsize::new(2).unwrap());
487        ring_buffer.push("1".to_owned());
488        ring_buffer.push("2".to_owned());
489
490        // When I extend it with lots of items
491        ring_buffer.extend(vec![
492            "3".to_owned(),
493            "4".to_owned(),
494            "5".to_owned(),
495            "6".to_owned(),
496            "7".to_owned(),
497        ]);
498
499        // Then only the last N items remain
500        assert_eq!(ring_buffer.iter().map(String::as_str).collect::<Vec<_>>(), vec!["6", "7"]);
501    }
502
503    #[test]
504    fn test_capacity_survives_a_serialization_round_trip() {
505        // Given a partially filled ring buffer with a non-default capacity,
506        let mut ring_buffer = RingBuffer::new(NonZeroUsize::new(3).unwrap());
507        ring_buffer.push(1);
508
509        // When it's deserialized back,
510        let json = serde_json::to_string(&ring_buffer).unwrap();
511        let mut ring_buffer: RingBuffer<i32> = serde_json::from_str(&json).unwrap();
512
513        // Then it still has its original capacity and room for two more items.
514        assert_eq!(ring_buffer.capacity(), 3);
515        assert_eq!(ring_buffer.inner.capacity(), 3);
516        ring_buffer.push(2);
517        ring_buffer.push(3);
518
519        assert_eq!(ring_buffer.iter().copied().collect::<Vec<_>>(), vec![1, 2, 3]);
520    }
521
522    #[test]
523    fn test_retain() {
524        let mut ring_buffer = RingBuffer::new(NonZeroUsize::new(2).unwrap());
525        ring_buffer.push(1);
526        ring_buffer.push(2);
527
528        ring_buffer.retain(|v| v % 2 == 0);
529
530        assert_eq!(ring_buffer.len(), 1);
531        assert_eq!(ring_buffer.get(0).copied().unwrap(), 2);
532    }
533}
matrix_sdk_common/ring_buffer.rs

matrix_sdk_common/
ring_buffer.rs
