Added TickedAsyncLocalExecutor implementation

Author: coder137

Cargo.toml

@@ -0,0 +1,10 @@
+[package]
+name = "ticked_async_executor"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+async-task = "4.7"
+
+[dev-dependencies]
+tokio = { version = "1", features = ["full"] }

src/lib.rs

@@ -0,0 +1,99 @@
+use std::{
+    future::Future,
+    sync::{
+        atomic::{AtomicUsize, Ordering},
+        mpsc, Arc,
+    },
+};
+
+use async_task::{Runnable, Task};
+
+pub struct TickedAsyncLocalExecutor {
+    channel: (mpsc::Sender<Runnable>, mpsc::Receiver<Runnable>),
+    num_woken_tasks: Arc<AtomicUsize>,
+}
+
+impl Default for TickedAsyncLocalExecutor {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl TickedAsyncLocalExecutor {
+    pub fn new() -> Self {
+        Self {
+            channel: mpsc::channel(),
+            num_woken_tasks: Arc::new(AtomicUsize::new(0)),
+        }
+    }
+
+    pub fn spawn_local<T>(&self, future: impl Future<Output = T> + 'static) -> Task<T>
+    where
+        T: 'static,
+    {
+        let sender = self.channel.0.clone();
+        let num_woken_tasks = self.num_woken_tasks.clone();
+        let schedule = move |runnable| {
+            sender.send(runnable).unwrap_or(());
+            num_woken_tasks.fetch_add(1, Ordering::Relaxed);
+        };
+        let (runnable, task) = async_task::spawn_local(future, schedule);
+        runnable.schedule();
+        task
+    }
+
+    /// Run the woken tasks once
+    ///
+    /// NOTE: Will not run tasks that are woken/scheduled immediately after `Runnable::run`
+    pub fn tick(&self) {
+        let num_woken_tasks = self.num_woken_tasks.load(Ordering::Relaxed);
+        self.channel
+            .1
+            .try_iter()
+            .take(num_woken_tasks)
+            .for_each(|runnable| {
+                runnable.run();
+            });
+        self.num_woken_tasks
+            .fetch_sub(num_woken_tasks, Ordering::Relaxed);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_multiple_tasks() {
+        let executor = TickedAsyncLocalExecutor::new();
+        executor
+            .spawn_local(async move {
+                println!("A: Start");
+                tokio::task::yield_now().await;
+                println!("A: End");
+            })
+            .detach();
+
+        executor
+            .spawn_local(async move {
+                println!("B: Start");
+                tokio::task::yield_now().await;
+                println!("B: End");
+            })
+            .detach();
+
+        executor
+            .spawn_local(async move {
+                println!("C: Start");
+                tokio::task::yield_now().await;
+                println!("C: End");
+            })
+            .detach();
+
+        // A, B, C: Start
+        executor.tick();
+
+        // A, B, C: End
+        executor.tick();
+    }
+}