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
|
#![warn(rust_2018_idioms)]
#![cfg(feature = "full")]
use tokio::sync::oneshot;
use tokio::time::{self, timeout, timeout_at, Instant};
use tokio_test::*;
use futures::future::pending;
use std::time::Duration;
#[tokio::test]
async fn simultaneous_deadline_future_completion() {
// Create a future that is immediately ready
let mut fut = task::spawn(timeout_at(Instant::now(), async {}));
// Ready!
assert_ready_ok!(fut.poll());
}
#[tokio::test]
async fn completed_future_past_deadline() {
// Wrap it with a deadline
let mut fut = task::spawn(timeout_at(Instant::now() - ms(1000), async {}));
// Ready!
assert_ready_ok!(fut.poll());
}
#[tokio::test]
async fn future_and_deadline_in_future() {
time::pause();
// Not yet complete
let (tx, rx) = oneshot::channel();
// Wrap it with a deadline
let mut fut = task::spawn(timeout_at(Instant::now() + ms(100), rx));
assert_pending!(fut.poll());
// Turn the timer, it runs for the elapsed time
time::advance(ms(90)).await;
assert_pending!(fut.poll());
// Complete the future
tx.send(()).unwrap();
assert!(fut.is_woken());
assert_ready_ok!(fut.poll()).unwrap();
}
#[tokio::test]
async fn future_and_timeout_in_future() {
time::pause();
// Not yet complete
let (tx, rx) = oneshot::channel();
// Wrap it with a deadline
let mut fut = task::spawn(timeout(ms(100), rx));
// Ready!
assert_pending!(fut.poll());
// Turn the timer, it runs for the elapsed time
time::advance(ms(90)).await;
assert_pending!(fut.poll());
// Complete the future
tx.send(()).unwrap();
assert_ready_ok!(fut.poll()).unwrap();
}
#[tokio::test]
async fn very_large_timeout() {
time::pause();
// Not yet complete
let (tx, rx) = oneshot::channel();
// copy-paste unstable `Duration::MAX`
let duration_max = Duration::from_secs(u64::MAX) + Duration::from_nanos(999_999_999);
// Wrap it with a deadline
let mut fut = task::spawn(timeout(duration_max, rx));
// Ready!
assert_pending!(fut.poll());
// Turn the timer, it runs for the elapsed time
time::advance(Duration::from_secs(86400 * 365 * 10)).await;
assert_pending!(fut.poll());
// Complete the future
tx.send(()).unwrap();
assert_ready_ok!(fut.poll()).unwrap();
}
#[tokio::test]
async fn deadline_now_elapses() {
use futures::future::pending;
time::pause();
// Wrap it with a deadline
let mut fut = task::spawn(timeout_at(Instant::now(), pending::<()>()));
// Factor in jitter
// TODO: don't require this
time::advance(ms(1)).await;
assert_ready_err!(fut.poll());
}
#[tokio::test]
async fn deadline_future_elapses() {
time::pause();
// Wrap it with a deadline
let mut fut = task::spawn(timeout_at(Instant::now() + ms(300), pending::<()>()));
assert_pending!(fut.poll());
time::advance(ms(301)).await;
assert!(fut.is_woken());
assert_ready_err!(fut.poll());
}
fn ms(n: u64) -> Duration {
Duration::from_millis(n)
}
#[tokio::test]
async fn timeout_is_not_exhausted_by_future() {
let fut = timeout(ms(1), async {
let mut buffer = [0u8; 1];
loop {
use tokio::io::AsyncReadExt;
let _ = tokio::io::empty().read(&mut buffer).await;
}
});
assert!(fut.await.is_err());
}
|
