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
149
150
151
152
153
154
155
156
157
158
159
160
161
|
#![cfg(not(miri))] // https://github.com/rust-lang/miri/issues/1038
use futures::channel::oneshot;
use futures::executor::ThreadPool;
use futures::future::{self, ok, Future, FutureExt, TryFutureExt};
use futures::task::SpawnExt;
use std::sync::mpsc;
use std::thread;
fn run<F: Future + Send + 'static>(future: F) {
let tp = ThreadPool::new().unwrap();
tp.spawn(future.map(drop)).unwrap();
}
#[test]
fn join1() {
let (tx, rx) = mpsc::channel();
run(future::try_join(ok::<i32, i32>(1), ok(2)).map_ok(move |v| tx.send(v).unwrap()));
assert_eq!(rx.recv(), Ok((1, 2)));
assert!(rx.recv().is_err());
}
#[test]
fn join2() {
let (c1, p1) = oneshot::channel::<i32>();
let (c2, p2) = oneshot::channel::<i32>();
let (tx, rx) = mpsc::channel();
run(future::try_join(p1, p2).map_ok(move |v| tx.send(v).unwrap()));
assert!(rx.try_recv().is_err());
c1.send(1).unwrap();
assert!(rx.try_recv().is_err());
c2.send(2).unwrap();
assert_eq!(rx.recv(), Ok((1, 2)));
assert!(rx.recv().is_err());
}
#[test]
fn join3() {
let (c1, p1) = oneshot::channel::<i32>();
let (c2, p2) = oneshot::channel::<i32>();
let (tx, rx) = mpsc::channel();
run(future::try_join(p1, p2).map_err(move |_v| tx.send(1).unwrap()));
assert!(rx.try_recv().is_err());
drop(c1);
assert_eq!(rx.recv(), Ok(1));
assert!(rx.recv().is_err());
drop(c2);
}
#[test]
fn join4() {
let (c1, p1) = oneshot::channel::<i32>();
let (c2, p2) = oneshot::channel::<i32>();
let (tx, rx) = mpsc::channel();
run(future::try_join(p1, p2).map_err(move |v| tx.send(v).unwrap()));
assert!(rx.try_recv().is_err());
drop(c1);
assert!(rx.recv().is_ok());
drop(c2);
assert!(rx.recv().is_err());
}
#[test]
fn join5() {
let (c1, p1) = oneshot::channel::<i32>();
let (c2, p2) = oneshot::channel::<i32>();
let (c3, p3) = oneshot::channel::<i32>();
let (tx, rx) = mpsc::channel();
run(future::try_join(future::try_join(p1, p2), p3).map_ok(move |v| tx.send(v).unwrap()));
assert!(rx.try_recv().is_err());
c1.send(1).unwrap();
assert!(rx.try_recv().is_err());
c2.send(2).unwrap();
assert!(rx.try_recv().is_err());
c3.send(3).unwrap();
assert_eq!(rx.recv(), Ok(((1, 2), 3)));
assert!(rx.recv().is_err());
}
#[test]
fn select1() {
let (c1, p1) = oneshot::channel::<i32>();
let (c2, p2) = oneshot::channel::<i32>();
let (tx, rx) = mpsc::channel();
run(future::try_select(p1, p2).map_ok(move |v| tx.send(v).unwrap()));
assert!(rx.try_recv().is_err());
c1.send(1).unwrap();
let (v, p2) = rx.recv().unwrap().into_inner();
assert_eq!(v, 1);
assert!(rx.recv().is_err());
let (tx, rx) = mpsc::channel();
run(p2.map_ok(move |v| tx.send(v).unwrap()));
c2.send(2).unwrap();
assert_eq!(rx.recv(), Ok(2));
assert!(rx.recv().is_err());
}
#[test]
fn select2() {
let (c1, p1) = oneshot::channel::<i32>();
let (c2, p2) = oneshot::channel::<i32>();
let (tx, rx) = mpsc::channel();
run(future::try_select(p1, p2).map_err(move |v| tx.send((1, v.into_inner().1)).unwrap()));
assert!(rx.try_recv().is_err());
drop(c1);
let (v, p2) = rx.recv().unwrap();
assert_eq!(v, 1);
assert!(rx.recv().is_err());
let (tx, rx) = mpsc::channel();
run(p2.map_ok(move |v| tx.send(v).unwrap()));
c2.send(2).unwrap();
assert_eq!(rx.recv(), Ok(2));
assert!(rx.recv().is_err());
}
#[test]
fn select3() {
let (c1, p1) = oneshot::channel::<i32>();
let (c2, p2) = oneshot::channel::<i32>();
let (tx, rx) = mpsc::channel();
run(future::try_select(p1, p2).map_err(move |v| tx.send((1, v.into_inner().1)).unwrap()));
assert!(rx.try_recv().is_err());
drop(c1);
let (v, p2) = rx.recv().unwrap();
assert_eq!(v, 1);
assert!(rx.recv().is_err());
let (tx, rx) = mpsc::channel();
run(p2.map_err(move |_v| tx.send(2).unwrap()));
drop(c2);
assert_eq!(rx.recv(), Ok(2));
assert!(rx.recv().is_err());
}
#[test]
fn select4() {
let (tx, rx) = mpsc::channel::<oneshot::Sender<i32>>();
let t = thread::spawn(move || {
for c in rx {
c.send(1).unwrap();
}
});
let (tx2, rx2) = mpsc::channel();
for _ in 0..10000 {
let (c1, p1) = oneshot::channel::<i32>();
let (c2, p2) = oneshot::channel::<i32>();
let tx3 = tx2.clone();
run(future::try_select(p1, p2).map_ok(move |_| tx3.send(()).unwrap()));
tx.send(c1).unwrap();
rx2.recv().unwrap();
drop(c2);
}
drop(tx);
t.join().unwrap();
}
|
