nexio 0.1.0

A thread-per-core io_uring async runtime for Linux with owned-buffer I/O and a cancellation-safety layer checked by Miri and Kani
Documentation 
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

//! Asynchronous time utilities backed by `io_uring` timeouts.

use std::{fmt, future::Future, time::Duration};

use crate::{
    executor::{select2, Either},
    reactor::REACTOR,
};

/// Error returned when a [`timeout`] expires before the inner future completes.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Elapsed;

impl fmt::Display for Elapsed {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str("deadline has elapsed")
    }
}

impl std::error::Error for Elapsed {}

/// Sleep for at least `duration` using `io_uring` `IORING_OP_TIMEOUT`.
///
/// Dropping the returned future cancels the pending kernel timeout cleanly.
pub async fn sleep(duration: Duration) {
    if duration.is_zero() {
        crate::executor::yield_now().await;
        return;
    }
    if let Ok(op) = REACTOR.with(|r| r.borrow_mut().sleep(duration)) {
        let _ = op.await;
    }
}

/// Require a future to complete within `duration`.
///
/// If the deadline elapses first, the inner future is dropped immediately
/// (triggering cancellation safety for any in-flight `io_uring` ops) and
/// `Err(Elapsed)` is returned.
pub async fn timeout<F>(duration: Duration, future: F) -> Result<F::Output, Elapsed>
where
    F: Future,
{
    match select2(future, sleep(duration)).await {
        Either::Left(output) => Ok(output),
        Either::Right(()) => Err(Elapsed),
    }
}

/// A repeating timer that ticks at a fixed interval.
pub struct Interval {
    period: Duration,
}

impl Interval {
    /// Wait until the next tick elapses.
    pub async fn tick(&mut self) {
        sleep(self.period).await;
    }

    /// Return the interval period.
    pub fn period(&self) -> Duration {
        self.period
    }
}

/// Create an [`Interval`] that yields every `period`.
pub fn interval(period: Duration) -> Interval {
    Interval { period }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::executor::block_on;
    use std::time::Instant;

    #[test]
    fn test_sleep_duration() {
        block_on(async {
            let start = Instant::now();
            sleep(Duration::from_millis(15)).await;
            assert!(start.elapsed() >= Duration::from_millis(10));
        });
    }

    #[test]
    fn test_timeout_completes_before_deadline() {
        block_on(async {
            let res = timeout(Duration::from_millis(200), async {
                sleep(Duration::from_millis(2)).await;
                42
            })
            .await;
            assert_eq!(res, Ok(42));
        });
    }

    #[test]
    fn test_timeout_elapsed_drops_inner() {
        block_on(async {
            let res = timeout(Duration::from_millis(10), async {
                sleep(Duration::from_millis(500)).await;
                99
            })
            .await;
            assert_eq!(res, Err(Elapsed));
        });
    }

    #[test]
    fn test_interval_ticks() {
        block_on(async {
            let mut iv = interval(Duration::from_millis(5));
            let start = Instant::now();
            iv.tick().await;
            iv.tick().await;
            assert!(start.elapsed() >= Duration::from_millis(5));
        });
    }
}