Struct Consumer

Source

pub struct Consumer<T> { /* private fields */ }

Expand description

Consumer handle. Move it to the consuming thread.

Implementations§

Source §

impl<T> Consumer<T>

Source

pub fn try_pop(&mut self) -> Option<T>

Pop a value. Returns None if the buffer is empty. Wait-free.

Examples found in repository ?

examples/demo.rs (line 16)

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 88)

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}