Add a Monoio async runtime

.github/workflows/ci.yaml

@@ -15,7 +15,7 @@ jobs:
       matrix:
         include:
           - toolchain: "nightly"
-            features: "bench,serde,bt,singlethreaded"
+            features: "bench,serde,bt,singlethreaded,monoio"
 
     steps:
       - name: Setup | Checkout

Cargo.toml

@@ -24,7 +24,9 @@ clap = { version = "4.1.11", features = ["derive", "env"] }
 derive_more = { version="0.99.9" }
 futures = "0.3"
 lazy_static = "1.4.0"
+local-sync = "0.1"
 maplit = "1.0.2"
+monoio = "0.2.2"
 pretty_assertions = "1.0.0"
 proc-macro2 = "1.0"
 quote = "1.0"
@@ -38,7 +40,7 @@ tokio = { version="1.8", default-features=false, features=["fs", "io-util", "mac
 tracing = { version = "0.1.40" }
 tracing-appender = "0.2.0"
 tracing-futures = "0.2.4"
-tracing-subscriber = { version = "0.3.3",  features=["env-filter"] }
+tracing-subscriber = { version = "0.3.18",  features=["env-filter"] }
 validit = { version = "0.2.2" }
 
 [workspace]

README.md

@@ -1,7 +1,8 @@
 <div align="center">
     <h1>Openraft</h1>
     <h4>
-        Advanced <a href="https://raft.github.io/">Raft</a> in 🦀 Rust using <a href="https://tokio.rs/">Tokio</a>. Please ⭐ on <a href="https://github.com/datafuselabs/openraft">github</a>!
+        Advanced <a href="https://raft.github.io/">Raft</a> in 🦀 Rust using <a href="https://tokio.rs/">Tokio</a> or 
+<a href="https://github.com/bytedance/monoio/">Monoio</a>. Please ⭐ on <a href="https://github.com/datafuselabs/openraft">github</a>!
     </h4>
 
 

memstore/Cargo.toml

@@ -24,6 +24,7 @@ tracing         = { workspace = true }
 [dev-dependencies]
 
 [features]
+monoio = ["openraft/monoio"]
 
 [package.metadata.docs.rs]
 all-features = true

memstore/src/lib.rs

@@ -15,6 +15,7 @@ use openraft::storage::LogState;
 use openraft::storage::RaftLogReader;
 use openraft::storage::RaftSnapshotBuilder;
 use openraft::storage::Snapshot;
+use openraft::AsyncRuntime;
 use openraft::Entry;
 use openraft::EntryPayload;
 use openraft::LogId;
@@ -26,7 +27,6 @@ use openraft::SnapshotMeta;
 use openraft::StorageError;
 use openraft::StorageIOError;
 use openraft::StoredMembership;
-use openraft::TokioRuntime;
 use openraft::Vote;
 use serde::Deserialize;
 use serde::Serialize;
@@ -72,6 +72,7 @@ pub struct ClientResponse(pub Option<String>);
 
 pub type MemNodeId = u64;
 
+#[cfg(not(feature = "monoio"))]
 openraft::declare_raft_types!(
     /// Declare the type configuration for `MemStore`.
     pub TypeConfig:
@@ -81,7 +82,20 @@ openraft::declare_raft_types!(
         Node = (),
         Entry = Entry<TypeConfig>,
         SnapshotData = Cursor<Vec<u8>>,
-        AsyncRuntime = TokioRuntime
+        AsyncRuntime = openraft::TokioRuntime
+);
+
+#[cfg(feature = "monoio")]
+openraft::declare_raft_types!(
+    /// Declare the type configuration for `MemStore`.
+    pub TypeConfig:
+        D = ClientRequest,
+        R = ClientResponse,
+        NodeId = MemNodeId,
+        Node = (),
+        Entry = Entry<TypeConfig>,
+        SnapshotData = Cursor<Vec<u8>>,
+        AsyncRuntime = openraft::MonoioRuntime
 );
 
 /// The application snapshot type which the `MemStore` works with.
@@ -232,7 +246,7 @@ impl RaftSnapshotBuilder<TypeConfig> for Arc<MemStore> {
 
         if let Some(d) = self.get_blocking(&BlockOperation::DelayBuildingSnapshot) {
             tracing::info!(?d, "delay snapshot build");
-            tokio::time::sleep(d).await;
+            <TypeConfig as RaftTypeConfig>::AsyncRuntime::sleep(d).await;
         }
 
         {
@@ -245,7 +259,7 @@ impl RaftSnapshotBuilder<TypeConfig> for Arc<MemStore> {
 
             if let Some(d) = self.get_blocking(&BlockOperation::BuildSnapshot) {
                 tracing::info!(?d, "blocking snapshot build");
-                tokio::time::sleep(d).await;
+                <TypeConfig as RaftTypeConfig>::AsyncRuntime::sleep(d).await;
             }
         }
 
@@ -368,7 +382,7 @@ impl RaftStorage<TypeConfig> for Arc<MemStore> {
 
         if let Some(d) = self.get_blocking(&BlockOperation::PurgeLog) {
             tracing::info!(?d, "block purging log");
-            tokio::time::sleep(d).await;
+            <TypeConfig as RaftTypeConfig>::AsyncRuntime::sleep(d).await;
         }
 
         {

openraft/Cargo.toml

@@ -26,6 +26,8 @@ rand            = { workspace = true }
 serde           = { workspace = true, optional = true }
 serde_json      = { workspace = true, optional = true }
 tempfile        = { workspace = true, optional = true }
+monoio          = { workspace = true, optional = true }
+local-sync      = { workspace = true, optional = true }
 thiserror       = { workspace = true }
 tokio           = { workspace = true }
 tracing         = { workspace = true }
@@ -78,6 +80,9 @@ compat = []
 compat-07 = ["compat", "serde", "dep:or07", "compat-07-testing"]
 compat-07-testing = ["dep:tempfile", "anyhow", "dep:serde_json"]
 
+# Turn on monoio Runtime with singlethreaded monoio tasks
+monoio = ["dep:monoio", "singlethreaded", "dep:local-sync"]
+
 # Allows an application to implement a custom the v2 storage API.
 # See `openraft::storage::v2` for more details.
 # V2 API are unstable and may change in the future.
@@ -115,6 +120,7 @@ generic-snapshot-data = []
 tracing-log = [ "tracing/log" ]
 
 # default = ["single-term-leader"]
+# default = ["monoio"]
 
 [package.metadata.docs.rs]
 

openraft/src/async_runtime.rs → openraft/src/async_runtime/mod.rs

@@ -6,7 +6,10 @@ use std::time::Duration;
 use crate::Instant;
 use crate::OptionalSend;
 use crate::OptionalSync;
-use crate::TokioInstant;
+
+#[cfg(not(feature = "monoio"))] pub mod tokio;
+
+#[cfg(feature = "monoio")] pub mod monoio;
 
 /// A trait defining interfaces with an asynchronous runtime.
 ///
@@ -96,78 +99,6 @@ pub trait AsyncRuntime: Debug + Default + PartialEq + Eq + OptionalSend + Option
     where T: OptionalSend;
 }
 
-/// `Tokio` is the default asynchronous executor.
-#[derive(Debug, Default, PartialEq, Eq)]
-pub struct TokioRuntime;
-
-pub struct TokioOneShotSender<T: OptionalSend>(pub tokio::sync::oneshot::Sender<T>);
-
-impl AsyncRuntime for TokioRuntime {
-    type JoinError = tokio::task::JoinError;
-    type JoinHandle<T: OptionalSend + 'static> = tokio::task::JoinHandle<T>;
-    type Sleep = tokio::time::Sleep;
-    type Instant = TokioInstant;
-    type TimeoutError = tokio::time::error::Elapsed;
-    type Timeout<R, T: Future<Output = R> + OptionalSend> = tokio::time::Timeout<T>;
-    type ThreadLocalRng = rand::rngs::ThreadRng;
-    type OneshotSender<T: OptionalSend> = TokioOneShotSender<T>;
-    type OneshotReceiver<T: OptionalSend> = tokio::sync::oneshot::Receiver<T>;
-    type OneshotReceiverError = tokio::sync::oneshot::error::RecvError;
-
-    #[inline]
-    fn spawn<T>(future: T) -> Self::JoinHandle<T::Output>
-    where
-        T: Future + OptionalSend + 'static,
-        T::Output: OptionalSend + 'static,
-    {
-        #[cfg(feature = "singlethreaded")]
-        {
-            tokio::task::spawn_local(future)
-        }
-        #[cfg(not(feature = "singlethreaded"))]
-        {
-            tokio::task::spawn(future)
-        }
-    }
-
-    #[inline]
-    fn sleep(duration: Duration) -> Self::Sleep {
-        tokio::time::sleep(duration)
-    }
-
-    #[inline]
-    fn sleep_until(deadline: Self::Instant) -> Self::Sleep {
-        tokio::time::sleep_until(deadline)
-    }
-
-    #[inline]
-    fn timeout<R, F: Future<Output = R> + OptionalSend>(duration: Duration, future: F) -> Self::Timeout<R, F> {
-        tokio::time::timeout(duration, future)
-    }
-
-    #[inline]
-    fn timeout_at<R, F: Future<Output = R> + OptionalSend>(deadline: Self::Instant, future: F) -> Self::Timeout<R, F> {
-        tokio::time::timeout_at(deadline, future)
-    }
-
-    #[inline]
-    fn is_panic(join_error: &Self::JoinError) -> bool {
-        join_error.is_panic()
-    }
-
-    #[inline]
-    fn thread_rng() -> Self::ThreadLocalRng {
-        rand::thread_rng()
-    }
-
-    #[inline]
-    fn oneshot<T>() -> (Self::OneshotSender<T>, Self::OneshotReceiver<T>)
-    where T: OptionalSend {
-        let (tx, rx) = tokio::sync::oneshot::channel();
-        (TokioOneShotSender(tx), rx)
-    }
-}
-
 pub trait AsyncOneshotSendExt<T>: Unpin {
     /// Attempts to send a value on this channel, returning it back if it could
     /// not be sent.
@@ -179,16 +110,3 @@ pub trait AsyncOneshotSendExt<T>: Unpin {
     /// problems.
     fn send(self, t: T) -> Result<(), T>;
 }
-
-impl<T: OptionalSend> AsyncOneshotSendExt<T> for TokioOneShotSender<T> {
-    #[inline]
-    fn send(self, t: T) -> Result<(), T> {
-        self.0.send(t)
-    }
-}
-
-impl<T: OptionalSend> Debug for TokioOneShotSender<T> {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        f.debug_tuple("TokioSendWrapper").finish()
-    }
-}

openraft/src/async_runtime/monoio.rs

@@ -0,0 +1,94 @@
+use std::fmt::Debug;
+use std::future::Future;
+use std::time::Duration;
+
+use super::AsyncOneshotSendExt;
+use crate::AsyncRuntime;
+use crate::OptionalSend;
+
+#[derive(Debug, Default, PartialEq, Eq)]
+pub struct MonoioRuntime;
+
+pub type MonoioInstant = monoio::time::Instant;
+
+impl crate::Instant for monoio::time::Instant {
+    #[inline]
+    fn now() -> Self {
+        monoio::time::Instant::now()
+    }
+}
+
+pub struct MonoioOneshotSender<T: OptionalSend>(pub local_sync::oneshot::Sender<T>);
+
+impl AsyncRuntime for MonoioRuntime {
+    type JoinError = crate::error::Infallible;
+    type JoinHandle<T: OptionalSend + 'static> = monoio::task::JoinHandle<Result<T, Self::JoinError>>;
+    type Sleep = monoio::time::Sleep;
+    type Instant = MonoioInstant;
+    type TimeoutError = monoio::time::error::Elapsed;
+    type Timeout<R, T: Future<Output = R> + OptionalSend> = monoio::time::Timeout<T>;
+    type ThreadLocalRng = rand::rngs::ThreadRng;
+    type OneshotSender<T: OptionalSend> = MonoioOneshotSender<T>;
+    type OneshotReceiver<T: OptionalSend> = local_sync::oneshot::Receiver<T>;
+    type OneshotReceiverError = local_sync::oneshot::error::RecvError;
+
+    #[inline]
+    fn spawn<T>(future: T) -> Self::JoinHandle<T::Output>
+    where
+        T: Future + OptionalSend + 'static,
+        T::Output: OptionalSend + 'static,
+    {
+        monoio::spawn(async move { Ok(future.await) })
+    }
+
+    #[inline]
+    fn sleep(duration: Duration) -> Self::Sleep {
+        monoio::time::sleep(duration)
+    }
+
+    #[inline]
+    fn sleep_until(deadline: Self::Instant) -> Self::Sleep {
+        monoio::time::sleep_until(deadline)
+    }
+
+    #[inline]
+    fn timeout<R, F: Future<Output = R> + OptionalSend>(duration: Duration, future: F) -> Self::Timeout<R, F> {
+        monoio::time::timeout(duration, future)
+    }
+
+    #[inline]
+    fn timeout_at<R, F: Future<Output = R> + OptionalSend>(deadline: Self::Instant, future: F) -> Self::Timeout<R, F> {
+        monoio::time::timeout_at(deadline, future)
+    }
+
+    #[inline]
+    fn is_panic(_join_error: &Self::JoinError) -> bool {
+        // A monoio task shouldn't panic or it would bubble the panic in case of a join
+        false
+    }
+
+    #[inline]
+    fn thread_rng() -> Self::ThreadLocalRng {
+        rand::thread_rng()
+    }
+
+    #[inline]
+    fn oneshot<T>() -> (Self::OneshotSender<T>, Self::OneshotReceiver<T>)
+    where T: OptionalSend {
+        let (tx, rx) = local_sync::oneshot::channel();
+        (MonoioOneshotSender(tx), rx)
+    }
+}
+
+impl<T: OptionalSend> AsyncOneshotSendExt<T> for MonoioOneshotSender<T> {
+    #[inline]
+    fn send(self, t: T) -> Result<(), T> {
+        self.0.send(t)
+    }
+}
+
+impl<T: OptionalSend> Debug for MonoioOneshotSender<T> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_tuple("MonoioSendWrapper").finish()
+    }
+}