nexus-queue 1.3.0

High-performance lock-free SPSC and MPSC queues for low-latency systems
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

//! Miri tests for nexus-queue lock-free ring buffers.
//!
//! Run: `cargo +nightly miri test -p nexus-queue --test miri_tests`

use std::cell::Cell;
use std::rc::Rc;

#[derive(Clone, Debug)]
struct DropCounter(Rc<Cell<u32>>);

impl Drop for DropCounter {
    fn drop(&mut self) {
        self.0.set(self.0.get() + 1);
    }
}

#[test]
fn spsc_push_pop() {
    let (tx, rx) = nexus_queue::spsc::ring_buffer::<u32>(4);
    for i in 0..16 {
        tx.push(i).unwrap();
        assert_eq!(rx.pop(), Some(i));
    }
}

#[test]
fn spsc_drop_with_pending() {
    let count = Rc::new(Cell::new(0u32));
    {
        let (tx, rx) = nexus_queue::spsc::ring_buffer::<DropCounter>(4);
        for _ in 0..4 {
            tx.push(DropCounter(Rc::clone(&count))).unwrap();
        }
        drop(rx);
        drop(tx);
    }
    // 4 items pushed, none popped — all must be dropped by the ring buffer.
    assert_eq!(count.get(), 4);
}

#[test]
fn spsc_fill_and_drain() {
    let (tx, rx) = nexus_queue::spsc::ring_buffer::<u32>(4);
    for i in 0..4 {
        tx.push(i).unwrap();
    }
    for i in 0..4 {
        assert_eq!(rx.pop(), Some(i));
    }
    assert_eq!(rx.pop(), None);
}

#[test]
fn mpsc_basic() {
    let (tx, rx) = nexus_queue::mpsc::ring_buffer::<u32>(4);
    for i in 0..4 {
        tx.push(i).unwrap();
    }
    for i in 0..4 {
        assert_eq!(rx.pop(), Some(i));
    }
    assert_eq!(rx.pop(), None);
}

#[test]
fn mpsc_clone_producer() {
    let (tx, rx) = nexus_queue::mpsc::ring_buffer::<u32>(8);
    let tx2 = tx.clone();
    for i in 0..4 {
        tx.push(i).unwrap();
        tx2.push(i + 100).unwrap();
    }
    let mut values = Vec::new();
    while let Some(v) = rx.pop() {
        values.push(v);
    }
    assert_eq!(values.len(), 8);
}

#[test]
fn spmc_basic() {
    let (tx, rx) = nexus_queue::spmc::ring_buffer::<u32>(4);
    for i in 0..4 {
        tx.push(i).unwrap();
    }
    let mut values = Vec::new();
    while let Some(v) = rx.pop() {
        values.push(v);
    }
    assert_eq!(values.len(), 4);
}