Introduce async-dispatcher runtime feature

This allows us to use ZeroMQ with native schedulers instead of relying directly on
Tokio or async-std's `spawn`.

Co-Authored-By: Kyle <[email protected]>
Co-Authored-By: Max <[email protected]>

Author: maxbrunsfeld

.github/workflows/main-ci.yml

@@ -59,6 +59,11 @@ jobs:
         with:
           command: test
           args: --all --no-default-features --features async-std-runtime,all-transport
+      - name: Test Async-dispatcher version
+        uses: actions-rs/cargo@v1
+        with:
+          command: test
+          args: --all --no-default-features --features async-dispatcher-runtime,all-transport
 
   fmt:
     name: Formatting

Cargo.toml

@@ -12,11 +12,13 @@ rust-version = "1.62.0"
 default = ["tokio-runtime", "all-transport"]
 tokio-runtime = ["tokio", "tokio-util"]
 async-std-runtime = ["async-std"]
+async-dispatcher-runtime = ["async-std", "async-dispatcher"]
 all-transport = ["ipc-transport", "tcp-transport"]
 ipc-transport = []
 tcp-transport = []
 
 [dependencies]
+async-dispatcher = { version = "0.1", optional = true }
 thiserror = "1"
 futures-channel = { version = "0.3", features = ["sink"] }
 futures-io = "0.3"
@@ -32,13 +34,17 @@ num-traits = "0.2"
 dashmap = "5"
 crossbeam-queue = "0.3"
 uuid = { version = "1", features = ["v4"] }
-regex = { version = "1", default-features = false, features = ["std", "unicode-perl"] }
+regex = { version = "1", default-features = false, features = [
+    "std",
+    "unicode-perl",
+] }
 once_cell = "1"
 log = "0.4"
 asynchronous-codec = "0.7"
 async-std = { version = "1", features = ["attributes"], optional = true }
 
 [dev-dependencies]
+async-dispatcher = { version = "0.1", features = ["macros"] }
 chrono = "0.4"
 criterion = "0.5"
 pretty_env_logger = "0.5"

benches/req_rep.rs

@@ -25,12 +25,12 @@ async fn setup(endpoint: &str) -> (ReqSocket, RepSocket) {
 fn criterion_benchmark(c: &mut Criterion) {
     #[cfg(feature = "tokio-runtime")]
     type Runtime = tokio::runtime::Runtime;
-    #[cfg(feature = "async-std-runtime")]
+    #[cfg(any(feature = "async-std-runtime", feature = "async-dispatcher-runtime"))]
     type Runtime = ();
 
     #[cfg(feature = "tokio-runtime")]
     let mut rt = tokio::runtime::Runtime::new().unwrap();
-    #[cfg(feature = "async-std-runtime")]
+    #[cfg(any(feature = "async-std-runtime", feature = "async-dispatcher-runtime"))]
     let mut rt = ();
 
     const N_MSG: u32 = 512;
@@ -48,6 +48,8 @@ fn criterion_benchmark(c: &mut Criterion) {
         let (req, rep) = rt.block_on(setup(endpoint));
         #[cfg(feature = "async-std-runtime")]
         let (req, rep) = async_std::task::block_on(setup(endpoint));
+        #[cfg(feature = "async-dispatcher-runtime")]
+        let (req, rep) = async_dispatcher::block_on(setup(endpoint));
 
         let (mut req, mut rep) = (Some(req), Some(rep));
 
@@ -57,6 +59,8 @@ fn criterion_benchmark(c: &mut Criterion) {
                 rt.block_on(iter_fn(&mut req, &mut rep));
                 #[cfg(feature = "async-std-runtime")]
                 async_std::task::block_on(iter_fn(&mut req, &mut rep));
+                #[cfg(feature = "async-dispatcher-runtime")]
+                async_dispatcher::block_on(iter_fn(&mut req, &mut rep));
             })
         });
     }

examples/async_helpers/mod.rs

@@ -22,3 +22,11 @@ pub async fn sleep(duration: std::time::Duration) {
 pub async fn sleep(duration: std::time::Duration) {
     async_std::task::sleep(duration).await
 }
+
+#[allow(unused_imports)]
+#[cfg(feature = "async-dispatcher-runtime")]
+pub use async_dispatcher::{main, test};
+
+#[allow(unused)]
+#[cfg(feature = "async-dispatcher-runtime")]
+pub use async_dispatcher::sleep;

src/async_rt/mod.rs

@@ -11,3 +11,6 @@ pub use tokio::{main, test};
 extern crate async_std;
 #[cfg(feature = "async-std-runtime")]
 pub use async_std::{main, test};
+
+#[cfg(feature = "async-dispatcher-runtime")]
+pub use async_dispatcher::{main, test};

src/async_rt/task/join_handle.rs

@@ -1,3 +1,5 @@
+#[cfg(feature = "async-dispatcher-runtime")]
+use async_dispatcher as rt_task;
 #[cfg(feature = "async-std-runtime")]
 use async_std::task as rt_task;
 #[cfg(feature = "tokio-runtime")]
@@ -17,7 +19,7 @@ impl<T> Future for JoinHandle<T> {
         // In async-std, the program aborts on panic so results arent returned. To
         // unify with tokio, we simply make an `Ok` result.
         let result = rt_task::JoinHandle::poll(Pin::new(&mut self.0), cx);
-        #[cfg(feature = "async-std-runtime")]
+        #[cfg(any(feature = "async-std-runtime", feature = "async-dispatcher-runtime"))]
         return result.map(Ok);
         #[cfg(feature = "tokio-runtime")]
         return result.map_err(|e| e.into());

src/async_rt/task/mod.rs

@@ -15,6 +15,8 @@ where
     let result = tokio::task::spawn(task).into();
     #[cfg(feature = "async-std-runtime")]
     let result = async_std::task::spawn(task).into();
+    #[cfg(feature = "async-dispatcher-runtime")]
+    let result = async_dispatcher::spawn(task).into();
 
     result
 }
@@ -54,7 +56,9 @@ pub async fn sleep(duration: std::time::Duration) {
     #[cfg(feature = "tokio-runtime")]
     ::tokio::time::sleep(duration).await;
     #[cfg(feature = "async-std-runtime")]
-    ::async_std::task::sleep(duration).await
+    ::async_std::task::sleep(duration).await;
+    #[cfg(feature = "async-dispatcher-runtime")]
+    ::async_dispatcher::sleep(duration).await;
 }
 
 pub async fn timeout<F, T>(
@@ -68,6 +72,8 @@ where
     let result = ::tokio::time::timeout(duration, f).await?;
     #[cfg(feature = "async-std-runtime")]
     let result = ::async_std::future::timeout(duration, f).await?;
+    #[cfg(feature = "async-dispatcher-runtime")]
+    let result = ::async_dispatcher::timeout(duration, f).await?;
 
     Ok(result)
 }

src/transport/ipc.rs

@@ -1,7 +1,7 @@
 #[cfg(feature = "tokio-runtime")]
 use tokio::net::{UnixListener, UnixStream};
 
-#[cfg(feature = "async-std-runtime")]
+#[cfg(any(feature = "async-std-runtime", feature = "async-dispatcher-runtime"))]
 use async_std::os::unix::net::{UnixListener, UnixStream};
 
 use super::make_framed;
@@ -39,7 +39,7 @@ where
 
     #[cfg(feature = "tokio-runtime")]
     let listener = UnixListener::bind(path)?;
-    #[cfg(feature = "async-std-runtime")]
+    #[cfg(any(feature = "async-std-runtime", feature = "async-dispatcher-runtime"))]
     let listener = UnixListener::bind(path).await?;
 
     let resolved_addr = listener.local_addr()?;
@@ -65,7 +65,7 @@ where
         }
         drop(listener);
         if let Some(listener_addr) = listener_addr {
-            #[cfg(feature = "async-std-runtime")]
+            #[cfg(any(feature = "async-std-runtime", feature = "async-dispatcher-runtime"))]
             use async_std::fs::remove_file;
             #[cfg(feature = "tokio-runtime")]
             use tokio::fs::remove_file;

src/transport/mod.rs

@@ -101,7 +101,7 @@ where
 }
 
 #[allow(unused)]
-#[cfg(feature = "async-std-runtime")]
+#[cfg(any(feature = "async-std-runtime", feature = "async-dispatcher-runtime"))]
 fn make_framed<T>(stream: T) -> FramedIo
 where
     T: futures_io::AsyncRead + futures_io::AsyncWrite + Send + Sync + 'static,

src/transport/tcp.rs

@@ -1,7 +1,7 @@
 #[cfg(feature = "tokio-runtime")]
 use tokio::net::{TcpListener, TcpStream};
 
-#[cfg(feature = "async-std-runtime")]
+#[cfg(any(feature = "async-std-runtime", feature = "async-dispatcher-runtime"))]
 use async_std::net::{TcpListener, TcpStream};
 
 use super::make_framed;