multiqueue 0.3.2

A fast mpmc broadcast queue
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
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148

extern crate multiqueue;
extern crate crossbeam;

use self::multiqueue::{broadcast_queue};
use self::crossbeam::scope;

fn spsc_example() {
    let (send, recv) = broadcast_queue(4);
    scope(|scope| {
        scope.spawn(move || for val in recv {
            println!("Got {}", val);
        });

        for i in 0..10 {
            // Don't do this busy loop in real stuff unless you're really sure
            loop {
                if send.try_send(i).is_ok() {
                    break;
                }
            }
        }
        drop(send);
    });
}

fn spsc_bcast_example() {
    let (send, recv) = broadcast_queue(4);
    scope(|scope| {
        for i in 0..2 {
            // or n
            let cur_recv = recv.add_stream();
            for j in 0..2 {
                let stream_consumer = cur_recv.clone();
                scope.spawn(move || for val in stream_consumer {
                    println!("Stream {} consumer {} got {}", i, j, val);
                });
            }
        }

        // Take notice that I drop the reader - this removes it from
        // the queue, meaning that the readers in the new threads
        // won't get starved by the lack of progress from recv
        recv.unsubscribe();

        for i in 0..10 {
            // Don't do this busy loop in real stuff unless you're really sure
            loop {
                if send.try_send(i).is_ok() {
                    break;
                }
            }
        }
        drop(send);
    });
}

fn spmc_bcast_example() {
    let (send, recv) = broadcast_queue(4);
    scope(|scope| {
        for i in 0..2 {
            let cur_recv = recv.add_stream();
            for j in 0..2 {
                let stream_consumer = cur_recv.clone();
                scope.spawn(move || for val in stream_consumer {
                    println!("Stream {} consumer {} got {}", i, j, val);
                });
            }
            // cur_recv is dropped here
        }

        // Take notice that I drop the reader - this removes it from
        // the queue, meaning that the readers in the new threads
        // won't get starved by the lack of progress from recv
        recv.unsubscribe();

        for i in 0..10 {
            // Don't do this busy loop in real stuff unless you're really sure
            loop {
                if send.try_send(i).is_ok() {
                    break;
                }
            }
        }
        drop(send);
    });
}

fn wacky_example() {
    let (send, recv) = broadcast_queue(4);
    scope(|scope| {
        for i in 0..2 {
            let cur_recv = recv.add_stream();
            for j in 0..2 {
                let stream_consumer = cur_recv.clone();
                scope.spawn(move || for val in stream_consumer {
                    println!("Stream {} consumer {} got {}", i, j, val);
                });
            }
            // cur_recv is dropped here
        }

        // On this stream, since there's only one consumer,
        // the receiver can be made into a SingleReceiver
        // which can view items inline in the queue
        let single_recv = recv.add_stream().into_single().unwrap();

        scope.spawn(move || for val in single_recv.iter_with(|item_ref| 10 * *item_ref) {
            println!("{}", val);
        });

        // Same as above, except this time we just want to iterate until the receiver is empty
        let single_recv_2 = recv.add_stream().into_single().unwrap();

        scope.spawn(move || for val in
            single_recv_2.try_iter_with(|item_ref| 10 * *item_ref) {
            println!("{}", val);
        });


        // Take notice that I drop the reader - this removes it from
        // the queue, meaning that the readers in the new threads
        // won't get starved by the lack of progress from recv
        recv.unsubscribe();

        for _ in 0..3 {
            for i in 0..10 {
                // Don't do this busy loop in real stuff unless you're really sure
                loop {
                    if send.try_send(i).is_ok() {
                        break;
                    }
                }
            }
        }
        drop(send);
    });
}

fn main() {
    println!("SPSC example");
    spsc_example();
    println!("\n\nSPSC Broadcast example");
    spsc_bcast_example();
    println!("\n\nSPMC Broadcast example");
    spmc_bcast_example();
    println!("\n\nWacky example");
    wacky_example();
}