Struct SpscRingBuffer

Source

pub struct SpscRingBuffer;

Expand description

Constructor namespace. Use SpscRingBuffer::with_capacity.

Implementations§

Source §

impl SpscRingBuffer

Source

pub fn with_capacity<T>(requested_capacity: usize) -> (Producer<T>, Consumer<T>)

Build a (producer, consumer) pair backed by a buffer of at least requested_capacity slots, rounded up to the next power of two. A floor of 2 is enforced.

Examples found in repository ?

examples/demo.rs (line 5)

4fn main() {
5    let (mut tx, mut rx) = SpscRingBuffer::with_capacity::<u32>(16);
6
7    let producer = thread::spawn(move || {
8        for i in 0..10u32 {
9            while tx.try_push(i).is_err() {}
10        }
11    });
12
13    let consumer = thread::spawn(move || {
14        let mut seen = Vec::new();
15        while seen.len() < 10 {
16            if let Some(v) = rx.try_pop() {
17                seen.push(v);
18            }
19        }
20        seen
21    });
22
23    producer.join().unwrap();
24    let seen = consumer.join().unwrap();
25    println!("consumed: {seen:?}");
26}

More examples

Hide additional examples

examples/sample_app.rs (line 54)

49fn base_feed_to_strategy() {
50    println!("== base: feed-handler -> strategy handoff ==");
51
52    // Drop-on-full is the caller's decision. Shown deterministically on a
53    // small ring: capacity 4, six ticks offered, the last two are dropped.
54    let (mut tx, _rx) = SpscRingBuffer::with_capacity::<Tick>(4);
55    let mut dropped = 0usize;
56    for seq in 0..6u64 {
57        let tick = Tick {
58            seq,
59            price_cents: 10_000 + seq as u32,
60        };
61        if tx.try_push(tick).is_err() {
62            dropped += 1;
63        }
64    }
65    println!("  cap-4 ring, 6 offered -> {dropped} dropped under backpressure");
66    assert_eq!(
67        dropped, 2,
68        "two ticks past capacity are dropped, not blocked"
69    );
70
71    // Steady state: a drained ring loses nothing and preserves feed order.
72    let n = 50_000u64;
73    let (mut tx, mut rx) = SpscRingBuffer::with_capacity::<Tick>(1024);
74    let feed = thread::spawn(move || {
75        for seq in 0..n {
76            let tick = Tick {
77                seq,
78                price_cents: 10_000 + (seq % 500) as u32,
79            };
80            while tx.try_push(tick).is_err() {
81                std::hint::spin_loop();
82            }
83        }
84    });
85    let strategy = thread::spawn(move || {
86        let mut expected = 0u64;
87        while expected < n {
88            if let Some(tick) = rx.try_pop() {
89                assert_eq!(tick.seq, expected, "ticks arrive in feed order");
90                expected += 1;
91            }
92        }
93        expected
94    });
95    feed.join().unwrap();
96    let received = strategy.join().unwrap();
97    println!("  streamed {received} ticks in order, zero loss when drained");
98    assert_eq!(received, n);
99}