Added feature timer_registration

Author: coder137

Cargo.toml

@@ -10,8 +10,17 @@ categories = ["asynchronous", "concurrency", "game-development", "simulation"]
 readme = "README.md"
 
 [features]
+# Provides a tick event in the form of a `tokio::sync::watch::Receiver<f64>` as per the `delta` provided 
+# to the `TickedAsyncExecutorTicker::tick` API
+# Also provides a timer implementation: `TickedTimerFromTickEvent`
 tick_event = []
 
+# Timers can be registered with the `TickedAsyncExecutorTicker` via the `TickedAsyncExecutorSpawner`
+# The timers count down with every call to `TickedAsyncExecutorTicker::tick` API as per the delta provided
+# Once the timer has elapsed, the corresponding `tokio::sync::oneshot::*` channel is notified
+# Also provides a timer implementation: `TickedTimerFromTimerRegistration`
+timer_registration = []
+
 [dependencies]
 async-task = "4.7"
 pin-project = "1"

src/lib.rs

@@ -16,3 +16,8 @@ pub use ticked_async_executor::*;
 mod ticked_timer_from_tick_event;
 #[cfg(feature = "tick_event")]
 pub use ticked_timer_from_tick_event::*;
+
+#[cfg(feature = "timer_registration")]
+mod ticked_timer_from_timer_registration;
+#[cfg(feature = "timer_registration")]
+pub use ticked_timer_from_timer_registration::*;

src/split_ticked_async_executor.rs

@@ -40,13 +40,18 @@ impl SplitTickedAsyncExecutor {
         #[cfg(feature = "tick_event")]
         let (tick_event_tx, tick_event_rx) = tokio::sync::watch::channel(1.0);
 
+        #[cfg(feature = "timer_registration")]
+        let (timer_registration_tx, timer_registration_rx) = mpsc::channel();
+
         let spawner = TickedAsyncExecutorSpawner {
             task_tx,
             num_woken_tasks: num_woken_tasks.clone(),
             num_spawned_tasks: num_spawned_tasks.clone(),
             observer: observer.clone(),
             #[cfg(feature = "tick_event")]
             tick_event_rx,
+            #[cfg(feature = "timer_registration")]
+            timer_registration_tx,
         };
         let ticker = TickedAsyncExecutorTicker {
             task_rx,
@@ -55,6 +60,10 @@ impl SplitTickedAsyncExecutor {
             observer,
             #[cfg(feature = "tick_event")]
             tick_event_tx,
+            #[cfg(feature = "timer_registration")]
+            timer_registration_rx,
+            #[cfg(feature = "timer_registration")]
+            timers: Vec::new(),
         };
         (spawner, ticker)
     }
@@ -72,6 +81,8 @@ pub struct TickedAsyncExecutorSpawner<O> {
 
     #[cfg(feature = "tick_event")]
     tick_event_rx: tokio::sync::watch::Receiver<f64>,
+    #[cfg(feature = "timer_registration")]
+    timer_registration_tx: mpsc::Sender<(f64, tokio::sync::oneshot::Sender<()>)>,
 }
 
 impl<O> TickedAsyncExecutorSpawner<O>
@@ -96,15 +107,19 @@ where
 
     #[cfg(feature = "tick_event")]
     pub fn create_timer_from_tick_event(&self) -> crate::TickedTimerFromTickEvent {
-        let tick_recv = self.tick_event_rx.clone();
-        crate::TickedTimerFromTickEvent::new(tick_recv)
+        crate::TickedTimerFromTickEvent::new(self.tick_event_rx.clone())
     }
 
     #[cfg(feature = "tick_event")]
     pub fn tick_channel(&self) -> tokio::sync::watch::Receiver<f64> {
         self.tick_event_rx.clone()
     }
 
+    #[cfg(feature = "timer_registration")]
+    pub fn create_timer_from_timer_registration(&self) -> crate::TickedTimerFromTimerRegistration {
+        crate::TickedTimerFromTimerRegistration::new(self.timer_registration_tx.clone())
+    }
+
     pub fn num_tasks(&self) -> usize {
         self.num_spawned_tasks.load(Ordering::Relaxed)
     }
@@ -151,6 +166,11 @@ pub struct TickedAsyncExecutorTicker<O> {
 
     #[cfg(feature = "tick_event")]
     tick_event_tx: tokio::sync::watch::Sender<f64>,
+
+    #[cfg(feature = "timer_registration")]
+    timer_registration_rx: mpsc::Receiver<(f64, tokio::sync::oneshot::Sender<()>)>,
+    #[cfg(feature = "timer_registration")]
+    timers: Vec<(f64, tokio::sync::oneshot::Sender<()>)>,
 }
 
 impl<O> TickedAsyncExecutorTicker<O>
@@ -161,6 +181,9 @@ where
         #[cfg(feature = "tick_event")]
         let _r = self.tick_event_tx.send(delta);
 
+        #[cfg(feature = "timer_registration")]
+        self.timer_registration_tick(delta);
+
         let mut num_woken_tasks = self.num_woken_tasks.load(Ordering::Relaxed);
         if let Some(limit) = limit {
             // Woken tasks should not exceed the allowed limit
@@ -183,4 +206,24 @@ where
             self.tick(constant_delta, None);
         }
     }
+
+    #[cfg(feature = "timer_registration")]
+    fn timer_registration_tick(&mut self, delta: f64) {
+        // Get new timers
+        let mut new_timers = self.timer_registration_rx.try_iter().collect::<Vec<_>>();
+        self.timers.append(&mut new_timers);
+
+        // Countdown timers
+        self.timers.iter_mut().for_each(|(elapsed, _)| {
+            *elapsed -= delta;
+        });
+
+        // Extract timers that have elapsed
+        // Notify corresponding channels
+        self.timers
+            .extract_if(.., |(elapsed, _)| *elapsed <= 0.0)
+            .for_each(|(_, rx)| {
+                let _ignore = rx.send(());
+            });
+    }
 }

src/ticked_async_executor.rs

@@ -67,6 +67,11 @@ where
         self.spawner.tick_channel()
     }
 
+    #[cfg(feature = "timer_registration")]
+    pub fn create_timer_from_timer_registration(&self) -> crate::TickedTimerFromTimerRegistration {
+        self.spawner.create_timer_from_timer_registration()
+    }
+
     pub fn wait_till_completed(&mut self, delta: f64) {
         self.ticker.wait_till_completed(delta);
     }
@@ -146,7 +151,7 @@ mod tests {
 
     #[cfg(feature = "tick_event")]
     #[test]
-    fn test_ticked_timer() {
+    fn test_ticked_timer_from_tick_event() {
         use std::time::{Duration, Instant};
 
         let mut executor = TickedAsyncExecutor::default();
@@ -221,6 +226,40 @@ mod tests {
         drop(executor);
     }
 
+    #[cfg(feature = "timer_registration")]
+    #[test]
+    fn test_ticked_timer_from_timer_registration() {
+        use std::time::{Duration, Instant};
+
+        let mut executor = TickedAsyncExecutor::default();
+
+        for _ in 0..10 {
+            let timer = executor.create_timer_from_timer_registration();
+            executor
+                .spawn_local("LocalTimer", async move {
+                    timer.sleep_for(256.0).await;
+                })
+                .detach();
+        }
+
+        let now = Instant::now();
+        let mut instances = vec![];
+        while executor.num_tasks() != 0 {
+            let current = Instant::now();
+            executor.tick(DELTA, None);
+            instances.push(current.elapsed());
+            std::thread::sleep(Duration::from_millis(16));
+        }
+        let elapsed = now.elapsed();
+        println!("Elapsed: {:?}", elapsed);
+        println!("Total: {:?}", instances);
+        println!(
+            "Min: {:?}, Max: {:?}",
+            instances.iter().min(),
+            instances.iter().max()
+        );
+    }
+
     #[test]
     fn test_limit() {
         let mut executor = TickedAsyncExecutor::default();

src/ticked_timer_from_timer_registration.rs

@@ -0,0 +1,27 @@
+use std::sync::mpsc;
+
+pub struct TickedTimerFromTimerRegistration {
+    timer_registration_tx: mpsc::Sender<(f64, tokio::sync::oneshot::Sender<()>)>,
+}
+
+impl TickedTimerFromTimerRegistration {
+    pub fn new(
+        timer_registration_tx: mpsc::Sender<(f64, tokio::sync::oneshot::Sender<()>)>,
+    ) -> Self {
+        Self {
+            timer_registration_tx,
+        }
+    }
+
+    pub async fn sleep_for(&self, duration_in_ms: f64) {
+        let (tx, rx) = tokio::sync::oneshot::channel();
+        let _ignore = async {
+            self.timer_registration_tx
+                .send((duration_in_ms, tx))
+                .map_err(|_| ())?;
+            rx.await.map_err(|_| ())?;
+            Ok::<(), ()>(())
+        }
+        .await;
+    }
+}