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
|
#![feature(test)]
extern crate test;
use crate::test::Bencher;
use {
futures::{
channel::mpsc::{self, Sender, UnboundedSender},
ready,
sink::Sink,
stream::{Stream, StreamExt},
task::{Context, Poll},
},
futures_test::task::noop_context,
std::pin::Pin,
};
/// Single producer, single consumer
#[bench]
fn unbounded_1_tx(b: &mut Bencher) {
let mut cx = noop_context();
b.iter(|| {
let (tx, mut rx) = mpsc::unbounded();
// 1000 iterations to avoid measuring overhead of initialization
// Result should be divided by 1000
for i in 0..1000 {
// Poll, not ready, park
assert_eq!(Poll::Pending, rx.poll_next_unpin(&mut cx));
UnboundedSender::unbounded_send(&tx, i).unwrap();
// Now poll ready
assert_eq!(Poll::Ready(Some(i)), rx.poll_next_unpin(&mut cx));
}
})
}
/// 100 producers, single consumer
#[bench]
fn unbounded_100_tx(b: &mut Bencher) {
let mut cx = noop_context();
b.iter(|| {
let (tx, mut rx) = mpsc::unbounded();
let tx: Vec<_> = (0..100).map(|_| tx.clone()).collect();
// 1000 send/recv operations total, result should be divided by 1000
for _ in 0..10 {
for (i, x) in tx.iter().enumerate() {
assert_eq!(Poll::Pending, rx.poll_next_unpin(&mut cx));
UnboundedSender::unbounded_send(x, i).unwrap();
assert_eq!(Poll::Ready(Some(i)), rx.poll_next_unpin(&mut cx));
}
}
})
}
#[bench]
fn unbounded_uncontended(b: &mut Bencher) {
let mut cx = noop_context();
b.iter(|| {
let (tx, mut rx) = mpsc::unbounded();
for i in 0..1000 {
UnboundedSender::unbounded_send(&tx, i).expect("send");
// No need to create a task, because poll is not going to park.
assert_eq!(Poll::Ready(Some(i)), rx.poll_next_unpin(&mut cx));
}
})
}
/// A Stream that continuously sends incrementing number of the queue
struct TestSender {
tx: Sender<u32>,
last: u32, // Last number sent
}
// Could be a Future, it doesn't matter
impl Stream for TestSender {
type Item = u32;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
let this = &mut *self;
let mut tx = Pin::new(&mut this.tx);
ready!(tx.as_mut().poll_ready(cx)).unwrap();
tx.as_mut().start_send(this.last + 1).unwrap();
this.last += 1;
assert_eq!(Poll::Pending, tx.as_mut().poll_flush(cx));
Poll::Ready(Some(this.last))
}
}
/// Single producers, single consumer
#[bench]
fn bounded_1_tx(b: &mut Bencher) {
let mut cx = noop_context();
b.iter(|| {
let (tx, mut rx) = mpsc::channel(0);
let mut tx = TestSender { tx, last: 0 };
for i in 0..1000 {
assert_eq!(Poll::Ready(Some(i + 1)), tx.poll_next_unpin(&mut cx));
assert_eq!(Poll::Pending, tx.poll_next_unpin(&mut cx));
assert_eq!(Poll::Ready(Some(i + 1)), rx.poll_next_unpin(&mut cx));
}
})
}
/// 100 producers, single consumer
#[bench]
fn bounded_100_tx(b: &mut Bencher) {
let mut cx = noop_context();
b.iter(|| {
// Each sender can send one item after specified capacity
let (tx, mut rx) = mpsc::channel(0);
let mut tx: Vec<_> = (0..100).map(|_| TestSender { tx: tx.clone(), last: 0 }).collect();
for i in 0..10 {
for x in &mut tx {
// Send an item
assert_eq!(Poll::Ready(Some(i + 1)), x.poll_next_unpin(&mut cx));
// Then block
assert_eq!(Poll::Pending, x.poll_next_unpin(&mut cx));
// Recv the item
assert_eq!(Poll::Ready(Some(i + 1)), rx.poll_next_unpin(&mut cx));
}
}
})
}
|
