Add RealtimeDelay for realtime clock support

src/lib.rs

@@ -36,12 +36,14 @@ mod delay;
 mod delay_queue;
 */
 mod interval;
+mod realtime_delay;
 
 pub use delay::Delay;
 pub use interval::Interval;
 /*
 pub use delay_queue::DelayQueue;
 */
+pub use realtime_delay::RealtimeDelay;
 
 pub struct TimerFd(AsyncFd<InnerTimerFd>);
 

src/realtime_delay.rs

@@ -0,0 +1,123 @@
+use std::future::Future;
+use std::io::{Error as IoError, ErrorKind};
+use std::pin::Pin;
+use std::task::{Context, Poll};
+use std::time::SystemTime;
+
+use crate::TimerFd;
+use futures_core::ready;
+use timerfd::{ClockId, SetTimeFlags, TimerState};
+use tokio::io::{AsyncRead, ReadBuf};
+
+/// A future that completes at a specified instant in realtime.
+/// Instances of [`RealtimeDelay`] perform no work and complete with () once the specified deadline has been reached.
+/// [`RealtimeDelay`] is powered by `timerfd` and has a resolution of 1 nanosecond.
+pub struct RealtimeDelay {
+    timerfd: TimerFd,
+    deadline: SystemTime,
+    cancel_on_set: bool,
+    initialized: bool,
+}
+
+impl RealtimeDelay {
+    /// Create a new [`RealtimeDelay`] instance that elapses at `deadline`.
+    fn create(clock: ClockId, deadline: SystemTime, cancel_on_set: bool) -> Result<Self, IoError> {
+        let timerfd = TimerFd::new(clock)?;
+        Ok(RealtimeDelay {
+            timerfd,
+            deadline,
+            cancel_on_set,
+            initialized: false,
+        })
+    }
+
+    /// Create a new [`RealtimeDelay`] instance with [`ClockId::Realtime`] that elapses at `deadline`.
+    pub fn new(deadline: SystemTime, cancel_on_set: bool) -> Result<Self, IoError> {
+        Self::create(ClockId::Realtime, deadline, cancel_on_set)
+    }
+
+    /// Create a new [`RealtimeDelay`] instance with [`ClockId::RealtimeAlarm`] that elapses at `deadline`.
+    pub fn new_alarm(deadline: SystemTime, cancel_on_set: bool) -> Result<Self, IoError> {
+        Self::create(ClockId::RealtimeAlarm, deadline, cancel_on_set)
+    }
+
+    /// Returns the instant at which the future will complete.
+    pub const fn deadline(&self) -> SystemTime {
+        self.deadline
+    }
+
+    /// Returns true if the [`RealtimeDelay`] has elapsed.
+    pub fn is_elapsed(&self) -> bool {
+        self.deadline <= SystemTime::now()
+    }
+}
+
+impl Future for RealtimeDelay {
+    type Output = Result<(), IoError>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        if !self.initialized {
+            if self.is_elapsed() {
+                return Poll::Ready(Ok(()));
+            }
+            // In order for `deadline` to be before the epoch but not have elapsed,
+            // would require the realtime clock to have also been before the epoch
+            // during `is_elapsed`.
+            let duration = self
+                .deadline
+                .duration_since(SystemTime::UNIX_EPOCH)
+                .map_err(|_| IoError::from(ErrorKind::Other))?;
+            let flags = match self.cancel_on_set {
+                true => SetTimeFlags::TimerCancelOnSet,
+                false => SetTimeFlags::Abstime,
+            };
+            self.timerfd.set_state(TimerState::Oneshot(duration), flags);
+            self.initialized = true;
+        }
+        let mut buf = [0u8; 8];
+        let mut buf = ReadBuf::new(&mut buf);
+        ready!(Pin::new(&mut self.as_mut().timerfd).poll_read(cx, &mut buf)?);
+        Poll::Ready(Ok(()))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::time::{Duration, Instant};
+
+    #[tokio::test]
+    async fn realtime_delay_zero_duration() -> Result<(), std::io::Error> {
+        let now = Instant::now();
+        let delay = RealtimeDelay::new(SystemTime::now(), false)?;
+        delay.await?;
+        let elapsed = now.elapsed();
+        println!("{:?}", elapsed);
+        assert!(elapsed < Duration::from_millis(1));
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn realtime_delay_works() {
+        let now = Instant::now();
+        let deadline = SystemTime::now().checked_add(Duration::from_micros(10)).unwrap();
+        let delay = RealtimeDelay::new(deadline, false).unwrap();
+        delay.await.unwrap();
+        let elapsed = now.elapsed();
+        println!("{:?}", elapsed);
+        assert!(elapsed < Duration::from_millis(1));
+    }
+
+    #[tokio::test]
+    async fn realtime_delay_realtime_1sec_works() {
+        let now = Instant::now();
+        let deadline = SystemTime::now().checked_add(Duration::from_secs(1)).unwrap();
+        let delay = RealtimeDelay::new(deadline, false).unwrap();
+        // Assume system clock is not modified in this period.
+        delay.await.unwrap();
+        let elapsed = now.elapsed();
+        println!("{:?}", elapsed);
+        let expected = Duration::from_secs(1)..Duration::from_secs(2);
+        assert!(expected.contains(&elapsed));
+    }
+}