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
use base::TimerTrait;
use crate::AsyncError;
use crate::AsyncResult;
use crate::IntoAsync;
use crate::TimerAsync;
impl<T: TimerTrait + IntoAsync> TimerAsync<T> {
pub async fn wait_sys(&self) -> AsyncResult<()> {
let (n, _) = self
.io_source
.read_to_vec(None, 0u64.to_ne_bytes().to_vec())
.await?;
if n != 8 {
return Err(AsyncError::EventAsync(base::Error::new(libc::ENODATA)));
}
Ok(())
}
}
#[cfg(test)]
mod tests {
use std::sync::Arc;
use std::time::Duration;
use std::time::Instant;
use base::Timer;
use super::super::fd_executor::EpollReactor;
use super::super::uring_executor::UringReactor;
use super::*;
use crate::common_executor::RawExecutor;
use crate::sys::unix::uring_executor::is_uring_stable;
use crate::Executor;
impl TimerAsync<Timer> {
pub(crate) fn new_poll(
timer: Timer,
ex: &Arc<RawExecutor<EpollReactor>>,
) -> AsyncResult<TimerAsync<Timer>> {
ex.new_source(timer)
.map(|io_source| TimerAsync { io_source })
}
pub(crate) fn new_uring(
timer: Timer,
ex: &Arc<RawExecutor<UringReactor>>,
) -> AsyncResult<TimerAsync<Timer>> {
ex.new_source(timer)
.map(|io_source| TimerAsync { io_source })
}
}
#[test]
fn timer() {
async fn this_test(ex: &Executor) {
let dur = Duration::from_millis(200);
let now = Instant::now();
TimerAsync::sleep(ex, dur).await.expect("unable to sleep");
assert!(now.elapsed() >= dur);
}
let ex = Executor::new().expect("creating an executor failed");
ex.run_until(this_test(&ex)).unwrap();
}
#[test]
fn one_shot() {
if !is_uring_stable() {
return;
}
async fn this_test(ex: &Arc<RawExecutor<UringReactor>>) {
let mut tfd = Timer::new().expect("failed to create timerfd");
let dur = Duration::from_millis(200);
let now = Instant::now();
tfd.reset(dur, None).expect("failed to arm timer");
let t = TimerAsync::new_uring(tfd, ex).unwrap();
t.wait().await.expect("unable to wait for timer");
assert!(now.elapsed() >= dur);
}
let ex = RawExecutor::<UringReactor>::new().unwrap();
ex.run_until(this_test(&ex)).unwrap();
}
#[test]
fn one_shot_fd() {
async fn this_test(ex: &Arc<RawExecutor<EpollReactor>>) {
let mut tfd = Timer::new().expect("failed to create timerfd");
let dur = Duration::from_millis(200);
let now = Instant::now();
tfd.reset(dur, None).expect("failed to arm timer");
let t = TimerAsync::new_poll(tfd, ex).unwrap();
t.wait().await.expect("unable to wait for timer");
assert!(now.elapsed() >= dur);
}
let ex = RawExecutor::<EpollReactor>::new().unwrap();
ex.run_until(this_test(&ex)).unwrap();
}
}