Use artificial time instead

rtic-time/Cargo.toml

@@ -23,6 +23,4 @@ rtic-common = { version = "1.0.0-alpha.0", path = "../rtic-common" }
 
 [dev-dependencies]
 parking_lot = "0.12"
-tokio = { version = "1.27", features = ["rt", "macros", "sync", "rt-multi-thread", "time"] }
-pretty_env_logger = "0.4"
-log = "0.4"
+cassette = "0.2"

rtic-time/tests/timer_queue.rs

@@ -2,29 +2,107 @@
 //!
 //! To run this test, you need to activate the `critical-section/std` feature.
 
-use std::{fmt::Debug, time::Duration};
+use std::{
+    fmt::Debug,
+    task::{Poll, Waker},
+};
 
+use cassette::{pin_mut, Cassette};
 use parking_lot::Mutex;
 use rtic_time::{Monotonic, TimerQueue};
-use tokio::sync::watch;
 
-static START: Mutex<Option<std::time::Instant>> = Mutex::new(None);
-pub struct StdTokioMono;
+static NOW: Mutex<Option<Instant>> = Mutex::new(None);
 
-// An instant that "starts" at Duration::ZERO, so we can
-// have a zero value.
 #[derive(PartialEq, Eq, PartialOrd, Ord, Clone, Copy, Debug)]
-pub struct Instant(std::time::Duration);
+pub struct Duration(u64);
+
+impl Duration {
+    pub fn from_ticks(millis: u64) -> Self {
+        Self(millis)
+    }
+
+    pub fn as_ticks(&self) -> u64 {
+        self.0
+    }
+}
+
+impl core::ops::Add<Duration> for Duration {
+    type Output = Duration;
+
+    fn add(self, rhs: Duration) -> Self::Output {
+        Self(self.0 + rhs.0)
+    }
+}
+
+impl From<Duration> for Instant {
+    fn from(value: Duration) -> Self {
+        Instant(value.0)
+    }
+}
+
+static WAKERS: Mutex<Vec<Waker>> = Mutex::new(Vec::new());
+
+#[derive(PartialEq, Eq, PartialOrd, Ord, Clone, Copy, Debug)]
+pub struct Instant(u64);
 
 impl Instant {
-    pub fn init() {
-        assert!(START.lock().is_none());
-        let _ = START.lock().insert(std::time::Instant::now());
+    const ZERO: Self = Self(0);
+
+    pub fn tick() -> bool {
+        // If we've never ticked before, initialize the clock.
+        if NOW.lock().is_none() {
+            *NOW.lock() = Some(Instant::ZERO);
+        }
+        // We've ticked before, add one to the clock
+        else {
+            let now = Instant::now();
+            let new_time = now + Duration(1);
+            *NOW.lock() = Some(new_time);
+        }
+
+        let had_wakers = !WAKERS.lock().is_empty();
+        // Wake up all things waiting for a specific time to happen.
+        for waker in WAKERS.lock().drain(..) {
+            waker.wake_by_ref();
+        }
+
+        let had_interrupt = TestMono::tick(false);
+
+        had_interrupt || had_wakers
     }
 
     pub fn now() -> Self {
-        let start = channel_read("Instant start not initialized", &START);
-        Self(start.elapsed())
+        NOW.lock().clone().unwrap_or(Instant::ZERO)
+    }
+
+    pub fn from_ticks(ticks: u64) -> Self {
+        Self(ticks)
+    }
+
+    pub fn as_ticks(&self) -> u64 {
+        self.0
+    }
+
+    pub fn elapsed(&self) -> Duration {
+        Duration(Self::now().0 - self.0)
+    }
+
+    pub async fn sleep_until(time: Instant) {
+        core::future::poll_fn(|ctx| {
+            if Instant::now() >= time {
+                Poll::Ready(())
+            } else {
+                WAKERS.lock().push(ctx.waker().clone());
+                Poll::Pending
+            }
+        })
+        .await;
+    }
+}
+
+impl From<u64> for Instant {
+    fn from(value: u64) -> Self {
+        Self::from_ticks(value)
     }
 }
 
@@ -32,7 +110,7 @@ impl core::ops::Add<Duration> for Instant {
     type Output = Instant;
 
     fn add(self, rhs: Duration) -> Self::Output {
-        Self(self.0 + rhs)
+        Self(self.0 + rhs.0)
     }
 }
 
@@ -40,7 +118,7 @@ impl core::ops::Sub<Duration> for Instant {
     type Output = Instant;
 
     fn sub(self, rhs: Duration) -> Self::Output {
-        Self(self.0 - rhs)
+        Self(self.0 - rhs.0)
     }
 }
 
@@ -48,104 +126,49 @@ impl core::ops::Sub<Instant> for Instant {
     type Output = Duration;
 
     fn sub(self, rhs: Instant) -> Self::Output {
-        self.0 - rhs.0
+        Duration(self.0 - rhs.0)
     }
 }
 
-fn channel_read<T: Clone>(msg: &str, channel: &Mutex<Option<T>>) -> T {
-    channel.lock().as_ref().expect(msg).clone()
-}
+static COMPARE: Mutex<Option<Instant>> = Mutex::new(None);
+static TIMER_QUEUE: TimerQueue<TestMono> = TimerQueue::new();
 
-fn event_write<T: Debug>(msg: &str, channel: &Mutex<Option<watch::Sender<T>>>, value: T) {
-    channel.lock().as_ref().expect(msg).send(value).unwrap()
-}
+pub struct TestMono;
 
-static COMPARE_RX: Mutex<Option<watch::Receiver<Instant>>> = Mutex::new(None);
-static COMPARE_TX: Mutex<Option<watch::Sender<Instant>>> = Mutex::new(None);
-static INTERRUPT_RX: Mutex<Option<watch::Receiver<()>>> = Mutex::new(None);
-static INTERRUPT_TX: Mutex<Option<watch::Sender<()>>> = Mutex::new(None);
+impl TestMono {
+    pub fn tick(force_interrupt: bool) -> bool {
+        let now = Instant::now();
+
+        let compare_reached = Some(now) == Self::compare();
+        let interrupt = compare_reached || force_interrupt;
+
+        if interrupt {
+            unsafe {
+                TestMono::queue().on_monotonic_interrupt();
+            }
+            true
+        } else {
+            false
+        }
+    }
 
-impl StdTokioMono {
     /// Initialize the monotonic.
-    ///
-    /// Returns a [`watch::Sender`] that will cause the interrupt
-    /// & compare-change tasks to exit if a value is sent to it or it
-    /// is dropped.
-    #[must_use = "Dropping the returned Sender stops interrupts & compare-change events"]
-    pub fn init() -> watch::Sender<()> {
-        Instant::init();
-        let (compare_tx, compare_rx) = watch::channel(Instant(Duration::ZERO));
-        let (irq_tx, irq_rx) = watch::channel(());
-
-        assert!(COMPARE_RX.lock().is_none());
-        assert!(COMPARE_TX.lock().is_none());
-        let _ = COMPARE_RX.lock().insert(compare_rx);
-        let _ = COMPARE_TX.lock().insert(compare_tx);
-
-        assert!(INTERRUPT_RX.lock().is_none());
-        assert!(INTERRUPT_TX.lock().is_none());
-        let _ = INTERRUPT_RX.lock().insert(irq_rx);
-        let _ = INTERRUPT_TX.lock().insert(irq_tx);
-
+    pub fn init() {
         Self::queue().initialize(Self);
-
-        let (killer_tx, mut killer_rx) = watch::channel(());
-
-        let mut killer_clone = killer_rx.clone();
-        // Set up a task that watches for changes to the COMPARE value,
-        // and re-starts a timeout based on that change
-        tokio::spawn(async move {
-            let mut compare_rx = channel_read("Compare RX not initialized", &COMPARE_RX);
-
-            loop {
-                let compare = compare_rx.borrow().clone();
-
-                let end = channel_read("Start not initialized", &START) + compare.0;
-
-                tokio::select! {
-                    _ = killer_clone.changed() => break,
-                    _ = compare_rx.changed() => {},
-                    _ = tokio::time::sleep_until(end.into()) => {
-                        event_write("Interrupt TX not initialized", &INTERRUPT_TX, ());
-                        // Sleep for a bit to avoid re-firing the interrupt a bunch of
-                        // times.
-                        tokio::time::sleep(Duration::from_millis(1)).await;
-                    },
-                }
-            }
-        });
-
-        // Set up a task that emulates an interrupt handler, calling `on_monotonic_interrupt`
-        // whenever an "interrupt" is generated.
-        tokio::spawn(async move {
-            let mut interrupt_rx = channel_read("Interrupt RX not initialized.", &INTERRUPT_RX);
-
-            loop {
-                tokio::select! {
-                    _ = killer_rx.changed() => break,
-                    _ = interrupt_rx.changed() => {
-                        // TODO: verify that we get interrupts triggered by an
-                        // explicit pend or due to COMPARE at the correct time.
-                    }
-                }
-
-                unsafe {
-                    StdTokioMono::queue().on_monotonic_interrupt();
-                }
-            }
-        });
-
-        killer_tx
     }
 
     /// Used to access the underlying timer queue
-    pub fn queue() -> &'static TimerQueue<StdTokioMono> {
+    pub fn queue() -> &'static TimerQueue<TestMono> {
         &TIMER_QUEUE
     }
+
+    pub fn compare() -> Option<Instant> {
+        COMPARE.lock().clone()
+    }
 }
 
-impl Monotonic for StdTokioMono {
-    const ZERO: Self::Instant = Instant(Duration::ZERO);
+impl Monotonic for TestMono {
+    const ZERO: Self::Instant = Instant::ZERO;
 
     type Instant = Instant;
 
@@ -156,66 +179,108 @@ impl Monotonic for StdTokioMono {
     }
 
     fn set_compare(instant: Self::Instant) {
-        // TODO: verify that we receive the correct amount & values
-        // for `set_compare`.
-
-        log::info!("Setting compare to {} ms", instant.0.as_millis());
-
-        event_write("Compare TX not initialized", &COMPARE_TX, instant);
+        let _ = COMPARE.lock().insert(instant);
     }
 
     fn clear_compare_flag() {}
 
     fn pend_interrupt() {
-        event_write("Interrupt TX not initialized", &INTERRUPT_TX, ());
+        Self::tick(true);
     }
 }
 
-static TIMER_QUEUE: TimerQueue<StdTokioMono> = TimerQueue::new();
+#[test]
+fn timer_queue() {
+    TestMono::init();
+    let start = Instant::ZERO;
 
-#[tokio::test]
-async fn main() {
-    pretty_env_logger::init();
-
-    let _interrupt_killer = StdTokioMono::init();
-
-    let start = std::time::Instant::now();
-
-    let build_delay_test = |threshold: u128, pre_delay: Option<u64>, delay: u64| {
-        let delay = Duration::from_millis(delay);
-        let pre_delay = pre_delay.map(Duration::from_millis);
+    let build_delay_test = |pre_delay: Option<u64>, delay: u64| {
+        let delay = Duration::from_ticks(delay);
+        let pre_delay = pre_delay.map(Duration::from_ticks);
 
         let total = if let Some(pre_delay) = pre_delay {
             pre_delay + delay
         } else {
             delay
         };
-        let total_millis = total.as_millis();
+        let total_millis = total.as_ticks();
+
         async move {
             if let Some(pre_delay) = pre_delay {
-                tokio::time::sleep_until((start + pre_delay).into()).await;
+                Instant::sleep_until(start + pre_delay).await;
             }
 
-            StdTokioMono::queue().delay(delay).await;
+            TestMono::queue().delay(delay).await;
 
-            let elapsed = start.elapsed().as_millis();
-            log::info!("{total_millis} ms delay reached (after {elapsed} ms)");
+            let elapsed = start.elapsed().as_ticks();
+            println!("{total_millis} ticks delay reached after {elapsed} ticks");
 
-            if elapsed > total_millis.saturating_add(threshold)
-                || elapsed < total_millis.saturating_sub(threshold)
-            {
-                panic!("{total_millis} ms delay was not on time ({elapsed} ms passed instead)");
+            if elapsed != total_millis {
+                panic!(
+                    "{total_millis} ticks delay was not on time ({elapsed} ticks passed instead)"
+                );
             }
         }
     };
 
-    // TODO: depending on the precision of the delays that can be used, this threshold
-    // may have to be altered a bit.
-    const TIME_THRESHOLD_MS: u128 = 5;
+    let d1 = build_delay_test(Some(100), 100);
+    pin_mut!(d1);
+    let mut d1 = Cassette::new(d1);
 
-    let sec1 = build_delay_test(TIME_THRESHOLD_MS, Some(100), 100);
-    let sec2 = build_delay_test(TIME_THRESHOLD_MS, None, 300);
-    let sec3 = build_delay_test(TIME_THRESHOLD_MS, None, 400);
+    let d2 = build_delay_test(None, 300);
+    pin_mut!(d2);
+    let mut d2 = Cassette::new(d2);
 
-    tokio::join!(sec2, sec1, sec3);
+    let d3 = build_delay_test(None, 400);
+    pin_mut!(d3);
+    let mut d3 = Cassette::new(d3);
+
+    macro_rules! try_poll {
+        ($fut:ident) => {
+            if !$fut.is_done() {
+                $fut.poll_on();
+            }
+        };
+    }
+
+    // Do an initial poll to set up all of the waiting futures
+    try_poll!(d1);
+    try_poll!(d2);
+    try_poll!(d3);
+
+    for _ in 0..500 {
+        // We only poll the waiting futures if an
+        // interrupt occured or if an artificial delay
+        // has passed.
+        if Instant::tick() {
+            try_poll!(d1);
+            try_poll!(d2);
+            try_poll!(d3);
+        }
+
+        if Instant::now() == 0.into() {
+            // First, we want to be waiting for our 300 tick delay
+            assert_eq!(TestMono::compare(), Some(300.into()));
+        }
+
+        if Instant::now() == 100.into() {
+            // After 100 ticks, we enqueue a new delay that is supposed to last
+            // until the 200-tick-mark
+            assert_eq!(TestMono::compare(), Some(200.into()));
+        }
+
+        if Instant::now() == 200.into() {
+            // After 200 ticks, we dequeue the 200-tick-mark delay and
+            // requeue the 300 tick delay
+            assert_eq!(TestMono::compare(), Some(300.into()));
+        }
+
+        if Instant::now() == 300.into() {
+            // After 300 ticks, we dequeue the 300-tick-mark delay and
+            // go to the 400 tick delay that is already enqueued
+            assert_eq!(TestMono::compare(), Some(400.into()));
+        }
+    }
+
+    assert!(d1.is_done() && d2.is_done() && d3.is_done());
 }