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
|
extern crate futures;
extern crate futures_cpupool;
use std::sync::atomic::{AtomicUsize, Ordering, ATOMIC_USIZE_INIT};
use std::thread;
use std::time::Duration;
use futures::future::Future;
use futures_cpupool::{CpuPool, Builder};
fn done<T: Send + 'static>(t: T) -> Box<Future<Item = T, Error = ()> + Send> {
Box::new(futures::future::ok(t))
}
#[test]
fn join() {
let pool = CpuPool::new(2);
let a = pool.spawn(done(1));
let b = pool.spawn(done(2));
let res = a.join(b).map(|(a, b)| a + b).wait();
assert_eq!(res.unwrap(), 3);
}
#[test]
fn select() {
let pool = CpuPool::new(2);
let a = pool.spawn(done(1));
let b = pool.spawn(done(2));
let (item1, next) = a.select(b).wait().ok().unwrap();
let item2 = next.wait().unwrap();
assert!(item1 != item2);
assert!((item1 == 1 && item2 == 2) || (item1 == 2 && item2 == 1));
}
#[test]
fn threads_go_away() {
static CNT: AtomicUsize = ATOMIC_USIZE_INIT;
struct A;
impl Drop for A {
fn drop(&mut self) {
CNT.fetch_add(1, Ordering::SeqCst);
}
}
thread_local!(static FOO: A = A);
let pool = CpuPool::new(2);
let _handle = pool.spawn_fn(|| {
FOO.with(|_| ());
Ok::<(), ()>(())
});
drop(pool);
for _ in 0..100 {
if CNT.load(Ordering::SeqCst) == 1 {
return
}
thread::sleep(Duration::from_millis(10));
}
panic!("thread didn't exit");
}
#[test]
fn lifecycle_test() {
static NUM_STARTS: AtomicUsize = ATOMIC_USIZE_INIT;
static NUM_STOPS: AtomicUsize = ATOMIC_USIZE_INIT;
fn after_start() {
NUM_STARTS.fetch_add(1, Ordering::SeqCst);
}
fn before_stop() {
NUM_STOPS.fetch_add(1, Ordering::SeqCst);
}
let pool = Builder::new()
.pool_size(4)
.after_start(after_start)
.before_stop(before_stop)
.create();
let _handle = pool.spawn_fn(|| {
Ok::<(), ()>(())
});
drop(pool);
for _ in 0..100 {
if NUM_STOPS.load(Ordering::SeqCst) == 4 {
assert_eq!(NUM_STARTS.load(Ordering::SeqCst), 4);
return;
}
thread::sleep(Duration::from_millis(10));
}
panic!("thread didn't exit");
}
#[test]
fn thread_name() {
let pool = Builder::new()
.name_prefix("my-pool-")
.create();
let future = pool.spawn_fn(|| {
assert!(thread::current().name().unwrap().starts_with("my-pool-"));
Ok::<(), ()>(())
});
let _ = future.wait();
}
|
