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
|
use std::sync::Arc;
use std::thread;
use async_lock::Barrier;
use futures_lite::future;
#[test]
#[cfg_attr(miri, ignore)]
fn smoke() {
future::block_on(async move {
const N: usize = 10;
let barrier = Arc::new(Barrier::new(N));
for _ in 0..10 {
let (tx, rx) = flume::unbounded();
for _ in 0..N - 1 {
let c = barrier.clone();
let tx = tx.clone();
thread::spawn(move || {
future::block_on(async move {
let res = c.wait().await;
tx.send_async(res.is_leader()).await.unwrap();
})
});
}
// At this point, all spawned threads should be blocked,
// so we shouldn't get anything from the channel.
let res = rx.try_recv();
assert!(res.is_err());
let mut leader_found = barrier.wait().await.is_leader();
// Now, the barrier is cleared and we should get data.
for _ in 0..N - 1 {
if rx.recv_async().await.unwrap() {
assert!(!leader_found);
leader_found = true;
}
}
assert!(leader_found);
}
});
}
#[cfg(all(feature = "std", not(target_family = "wasm")))]
#[test]
#[cfg_attr(miri, ignore)]
fn smoke_blocking() {
future::block_on(async move {
const N: usize = 10;
let barrier = Arc::new(Barrier::new(N));
for _ in 0..10 {
let (tx, rx) = flume::unbounded();
for _ in 0..N - 1 {
let c = barrier.clone();
let tx = tx.clone();
thread::spawn(move || {
let res = c.wait_blocking();
tx.send(res.is_leader()).unwrap();
});
}
// At this point, all spawned threads should be blocked,
// so we shouldn't get anything from the channel.
let res = rx.try_recv();
assert!(res.is_err());
let mut leader_found = barrier.wait_blocking().is_leader();
// Now, the barrier is cleared and we should get data.
for _ in 0..N - 1 {
if rx.recv_async().await.unwrap() {
assert!(!leader_found);
leader_found = true;
}
}
assert!(leader_found);
}
});
}
|
