ringbuf 0.5.3

Lock-free SPSC FIFO ring buffer with direct access to inner data
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133

#![allow(deprecated)]

use super::Rb;
use crate::{
    CachingProd, Cons, Prod,
    storage::Array,
    traits::*,
    wrap::{FrozenCons, FrozenProd, Wrap},
};
use core::{cell::Cell, mem};

#[test]
fn producer_publishes_without_commit() {
    let mut rb = Rb::<Array<i32, 2>>::default();
    let (prod, mut cons) = rb.split_ref();
    let mut prod = prod.freeze();
    prod.try_push(0).unwrap();
    prod.try_push(1).unwrap();
    assert_eq!(cons.try_pop(), Some(0));
    // The cached producer must fetch progress when its old snapshot is full.
    prod.try_push(2).unwrap();
    prod.commit();
    prod.fetch();
    prod.sync();
    assert_eq!(cons.try_pop(), Some(1));
    assert_eq!(cons.try_pop(), Some(2));
    assert_eq!(cons.try_pop(), None);
}

#[test]
fn discard_does_not_retract_published_items() {
    let mut rb = Rb::<Array<i32, 2>>::default();
    let (prod, mut cons) = rb.split_ref();
    let mut prod = prod.freeze();
    prod.try_push(0).unwrap();
    prod.discard();
    assert_eq!(cons.try_pop(), Some(0));
    prod.try_push(1).unwrap();
    assert_eq!(cons.try_pop(), Some(1));
    // Discarding after the consumer has advanced must not rewind the write index.
    prod.discard();
    assert_eq!(cons.try_pop(), None);
    prod.try_push(2).unwrap();
    assert_eq!(cons.try_pop(), Some(2));
}

#[test]
fn consumer_publishes_without_commit() {
    let mut rb = Rb::<Array<i32, 2>>::default();
    let (mut prod, cons) = rb.split_ref();
    let mut cons = cons.freeze();
    prod.push_slice(&[0, 1]);
    // The consumer starts with an empty cached snapshot and must fetch automatically.
    assert_eq!(cons.try_pop(), Some(0));
    prod.try_push(2).unwrap();
    cons.commit();
    cons.fetch();
    cons.sync();
    assert_eq!(cons.try_pop(), Some(1));
    assert_eq!(cons.try_pop(), Some(2));
    assert_eq!(cons.try_pop(), None);
}

struct CountDrop<'a>(&'a Cell<usize>);
impl Drop for CountDrop<'_> {
    fn drop(&mut self) {
        self.0.set(self.0.get() + 1);
    }
}

#[test]
fn forgotten_consumer_does_not_drop_items_twice() {
    for mode in 0..3 {
        for offset in [0, 3] {
            let drops: [Cell<usize>; 3] = core::array::from_fn(|_| Cell::new(0));
            let mut rb = Rb::<Array<CountDrop<'_>, 4>>::default();
            for _ in 0..offset {
                rb.try_push(CountDrop(&drops[0])).ok().unwrap();
                drop(rb.try_pop().unwrap());
            }
            drops[0].set(0);
            let mut prod = CachingProd::new(&rb);
            for counter in &drops {
                prod.try_push(CountDrop(counter)).ok().unwrap();
            }
            let mut cons = match mode {
                0 => crate::CachingCons::new(&rb).freeze(),
                1 => Cons::new(&rb).freeze(),
                _ => FrozenCons::new(&rb),
            };
            let item = cons.try_pop().unwrap();
            mem::forget(cons);
            drop(item);
            drop(prod);
            assert_eq!(rb.occupied_len(), 2);
            assert_eq!(drops[0].get(), 1);
            drop(rb);
            for counter in &drops {
                assert_eq!(counter.get(), 1);
            }
        }
    }
}

#[test]
fn forgotten_producer_keeps_items_owned_by_buffer() {
    let drops = Cell::new(0);
    let rb = Rb::<Array<CountDrop<'_>, 2>>::default();
    let mut prod = FrozenProd::new(&rb);
    prod.try_push(CountDrop(&drops)).ok().unwrap();
    mem::forget(prod);
    assert_eq!(rb.occupied_len(), 1);
    drop(rb);
    assert_eq!(drops.get(), 1);
}

#[test]
fn conversion_and_extraction_preserve_endpoint_rights() {
    let rb = Rb::<Array<i32, 2>>::default();
    let prod = Prod::new(&rb).freeze();
    let cons = Cons::new(&rb).freeze();
    let obs = prod.observe();
    assert!(obs.write_is_held() && obs.read_is_held());
    let rb_ref = prod.into_rb_ref();
    assert!(!obs.write_is_held() && obs.read_is_held());
    let mut prod = FrozenProd::new(rb_ref);
    prod.try_push(1).unwrap();
    drop(prod);
    drop(cons);
    assert!(!obs.write_is_held() && !obs.read_is_held());
    let mut cons = FrozenCons::new(&rb);
    assert_eq!(cons.try_pop(), Some(1));
}