thingbuf 0.1.6

I'm at the buffer pool. I'm at the MPSC channel. I'm at the combination MPSC channel and buffer pool.
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

use std::thread;
use thingbuf::mpsc::blocking;
use thingbuf::mpsc::errors::{TryRecvError, TrySendError};

#[test]
fn basically_works() {
    use std::collections::HashSet;

    const N_SENDS: usize = 10;
    const N_PRODUCERS: usize = 10;

    fn start_producer(tx: blocking::Sender<usize>, n: usize) -> thread::JoinHandle<()> {
        let tag = n * N_SENDS;
        thread::Builder::new()
            .name(format!("producer {}", n))
            .spawn(move || {
                for i in 0..N_SENDS {
                    let msg = i + tag;
                    println!("[producer {}] sending {}...", n, msg);
                    tx.send(msg).unwrap();
                    println!("[producer {}] sent {}!", n, msg);
                }
                println!("[producer {}] DONE!", n);
            })
            .expect("spawning threads should succeed")
    }

    let (tx, rx) = blocking::channel(N_SENDS / 2);
    for n in 0..N_PRODUCERS {
        start_producer(tx.clone(), n);
    }
    drop(tx);

    let mut results = HashSet::new();
    while let Some(val) = {
        println!("receiving...");
        rx.recv()
    } {
        println!("received {}!", val);
        results.insert(val);
    }

    let results = dbg!(results);

    for n in 0..N_PRODUCERS {
        let tag = n * N_SENDS;
        for i in 0..N_SENDS {
            let msg = i + tag;
            assert!(results.contains(&msg), "missing message {:?}", msg);
        }
    }
}

#[test]
fn tx_close_drains_queue() {
    const LEN: usize = 4;
    for i in 0..10000 {
        println!("\n\n--- iteration {} ---\n\n", i);

        let (tx, rx) = blocking::channel(LEN);
        let producer = thread::spawn(move || {
            for i in 0..LEN {
                tx.send(i).unwrap();
            }
        });

        for i in 0..LEN {
            assert_eq!(rx.recv(), Some(i))
        }

        producer.join().unwrap();
    }
}

#[test]
fn spsc_skip_slot() {
    let (tx, rx) = blocking::channel::<usize>(3);
    // 0 lap
    tx.send(0).unwrap();
    assert_eq!(rx.recv(), Some(0));
    tx.send(1).unwrap();
    let msg_ref = rx.try_recv_ref().unwrap();
    tx.send(2).unwrap();
    assert_eq!(rx.recv(), Some(2));
    // 1 lap
    tx.send(3).unwrap();
    assert_eq!(rx.recv(), Some(3));
    tx.send(4).unwrap();
    assert_eq!(rx.recv(), Some(4));
    drop(msg_ref);
    // 2 lap
    tx.send(5).unwrap();
    tx.send(6).unwrap();
    tx.send(7).unwrap();
    assert!(matches!(tx.try_send_ref(), Err(TrySendError::Full(_))));
    assert_eq!(rx.recv(), Some(5));
    assert_eq!(rx.recv(), Some(6));
    assert_eq!(rx.recv(), Some(7));
}

#[test]
fn spsc_full_after_skipped() {
    let (tx, rx) = blocking::channel::<usize>(3);
    // 0 lap
    tx.send(0).unwrap();
    assert_eq!(rx.recv(), Some(0));
    tx.send(1).unwrap();
    let _msg_ref = rx.try_recv_ref().unwrap();
    tx.send(2).unwrap();
    // lap 1
    tx.send(3).unwrap();
    assert!(matches!(tx.try_send_ref(), Err(TrySendError::Full(_))));
}

#[test]
fn spsc_empty_after_skipped() {
    let (tx, rx) = blocking::channel::<usize>(2);
    // 0 lap
    tx.send(0).unwrap();
    tx.send(1).unwrap();
    let _msg_ref = rx.try_recv_ref().unwrap();
    assert_eq!(rx.recv(), Some(1));
    assert!(matches!(rx.try_recv_ref(), Err(TryRecvError::Empty)));
}