Feature: Add TypeConfigExt to simplify RaftTypeConfig Access

This commit introduces a new trait, `TypeConfigExt`, which extends
`RaftTypeConfig`. The purpose of this trait is to simplify the access to
various functionalities provided by the `RaftTypeConfig` trait,
enhancing code readability and reducing complexity.

**Methods Added to `TypeConfigExt`:**
- `now()`
- `sleep()`
- `sleep_until()`
- `timeout()`
- `timeout_at()`
- `oneshot()`
- `spawn()`

**Usage Improvement:**
- Instead of using the
  `<<C as RaftTypeConfig>::AsyncRuntime as AsyncRuntime>::Instant::now()`,
  you can now simply call `C::now()`.

Author: drmingdrmer

openraft/src/core/raft_core.rs

@@ -88,9 +88,9 @@ use crate::storage::LogFlushed;
 use crate::storage::RaftLogReaderExt;
 use crate::storage::RaftLogStorage;
 use crate::storage::RaftStateMachine;
-use crate::type_config::alias::AsyncRuntimeOf;
 use crate::type_config::alias::InstantOf;
 use crate::type_config::alias::ResponderOf;
+use crate::type_config::TypeConfigExt;
 use crate::AsyncRuntime;
 use crate::ChangeMembers;
 use crate::Instant;
@@ -148,7 +148,7 @@ pub(crate) struct LeaderData<C: RaftTypeConfig> {
 impl<C: RaftTypeConfig> LeaderData<C> {
     pub(crate) fn new() -> Self {
         Self {
-            next_heartbeat: InstantOf::<C>::now(),
+            next_heartbeat: C::now(),
         }
     }
 }
@@ -509,19 +509,12 @@ where
     /// Currently heartbeat is a blank log
     #[tracing::instrument(level = "debug", skip_all, fields(id = display(self.id)))]
     pub fn send_heartbeat(&mut self, emitter: impl Display) -> bool {
-        tracing::debug!(
-            now = debug(<C::AsyncRuntime as AsyncRuntime>::Instant::now()),
-            "send_heartbeat"
-        );
+        tracing::debug!(now = debug(C::now()), "send_heartbeat");
 
         let mut lh = if let Some((lh, _)) = self.engine.get_leader_handler_or_reject(None) {
             lh
         } else {
-            tracing::debug!(
-                now = debug(<C::AsyncRuntime as AsyncRuntime>::Instant::now()),
-                "{} failed to send heartbeat",
-                emitter
-            );
+            tracing::debug!(now = debug(C::now()), "{} failed to send heartbeat", emitter);
             return false;
         };
 
@@ -1150,7 +1143,7 @@ where
                 self.handle_append_entries_request(rpc, tx);
             }
             RaftMsg::RequestVote { rpc, tx } => {
-                let now = <C::AsyncRuntime as AsyncRuntime>::Instant::now();
+                let now = C::now();
                 tracing::info!(
                     now = display(now.display()),
                     vote_request = display(&rpc),
@@ -1237,7 +1230,7 @@ where
                 resp,
                 sender_vote,
             } => {
-                let now = <C::AsyncRuntime as AsyncRuntime>::Instant::now();
+                let now = C::now();
 
                 tracing::info!(
                     now = display(now.display()),
@@ -1273,7 +1266,7 @@ where
             Notify::Tick { i } => {
                 // check every timer
 
-                let now = <C::AsyncRuntime as AsyncRuntime>::Instant::now();
+                let now = C::now();
                 tracing::debug!("received tick: {}, now: {:?}", i, now);
 
                 self.handle_tick_election();
@@ -1291,8 +1284,7 @@ where
 
                         // Install next heartbeat
                         if let Some(l) = &mut self.leader_data {
-                            l.next_heartbeat = <C::AsyncRuntime as AsyncRuntime>::Instant::now()
-                                + Duration::from_millis(self.config.heartbeat_interval);
+                            l.next_heartbeat = C::now() + Duration::from_millis(self.config.heartbeat_interval);
                         }
                     }
                 }
@@ -1431,7 +1423,7 @@ where
 
     #[tracing::instrument(level = "debug", skip_all)]
     fn handle_tick_election(&mut self) {
-        let now = <C::AsyncRuntime as AsyncRuntime>::Instant::now();
+        let now = C::now();
 
         tracing::debug!("try to trigger election by tick, now: {:?}", now);
 
@@ -1660,7 +1652,7 @@ where
 
                     // False positive lint warning(`non-binding `let` on a future`): https://github.com/rust-lang/rust-clippy/issues/9932
                     #[allow(clippy::let_underscore_future)]
-                    let _ = AsyncRuntimeOf::<C>::spawn(async move {
+                    let _ = C::spawn(async move {
                         for (log_index, tx) in removed.into_iter() {
                             tx.send(Err(ClientWriteError::ForwardToLeader(ForwardToLeader {
                                 leader_id,

openraft/src/core/sm/handle.rs

@@ -3,9 +3,8 @@
 use tokio::sync::mpsc;
 
 use crate::core::sm;
-use crate::type_config::alias::AsyncRuntimeOf;
 use crate::type_config::alias::JoinHandleOf;
-use crate::AsyncRuntime;
+use crate::type_config::TypeConfigExt;
 use crate::RaftTypeConfig;
 use crate::Snapshot;
 
@@ -55,7 +54,7 @@ where C: RaftTypeConfig
     /// If the state machine worker has shutdown, it will return an error.
     /// If there is not snapshot available, it will return `Ok(None)`.
     pub(crate) async fn get_snapshot(&self) -> Result<Option<Snapshot<C>>, &'static str> {
-        let (tx, rx) = AsyncRuntimeOf::<C>::oneshot();
+        let (tx, rx) = C::oneshot();
 
         let cmd = sm::Command::get_snapshot(tx);
         tracing::debug!("SnapshotReader sending command to sm::Worker: {:?}", cmd);

openraft/src/core/tick.rs

@@ -14,11 +14,8 @@ use tracing::Level;
 use tracing::Span;
 
 use crate::core::notify::Notify;
-use crate::type_config::alias::AsyncRuntimeOf;
-use crate::type_config::alias::InstantOf;
 use crate::type_config::alias::JoinHandleOf;
-use crate::AsyncRuntime;
-use crate::Instant;
+use crate::type_config::TypeConfigExt;
 use crate::RaftTypeConfig;
 
 /// Emit RaftMsg::Tick event at regular `interval`.
@@ -68,7 +65,7 @@ where C: RaftTypeConfig
 
         let shutdown = Mutex::new(Some(shutdown));
 
-        let join_handle = AsyncRuntimeOf::<C>::spawn(this.tick_loop(shutdown_rx).instrument(tracing::span!(
+        let join_handle = C::spawn(this.tick_loop(shutdown_rx).instrument(tracing::span!(
             parent: &Span::current(),
             Level::DEBUG,
             "tick"
@@ -87,8 +84,8 @@ where C: RaftTypeConfig
         let mut cancel = std::pin::pin!(cancel_rx);
 
         loop {
-            let at = InstantOf::<C>::now() + self.interval;
-            let mut sleep_fut = AsyncRuntimeOf::<C>::sleep_until(at);
+            let at = C::now() + self.interval;
+            let mut sleep_fut = C::sleep_until(at);
             let sleep_fut = std::pin::pin!(sleep_fut);
             let cancel_fut = cancel.as_mut();
 
@@ -159,8 +156,7 @@ mod tests {
     use tokio::time::Duration;
 
     use crate::core::Tick;
-    use crate::type_config::alias::AsyncRuntimeOf;
-    use crate::AsyncRuntime;
+    use crate::type_config::TypeConfigExt;
     use crate::RaftTypeConfig;
     use crate::TokioRuntime;
 
@@ -187,9 +183,9 @@ mod tests {
         let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
         let th = Tick::<TickUTConfig>::spawn(Duration::from_millis(100), tx, true);
 
-        AsyncRuntimeOf::<TickUTConfig>::sleep(Duration::from_millis(500)).await;
+        TickUTConfig::sleep(Duration::from_millis(500)).await;
         let _ = th.shutdown().unwrap().await;
-        AsyncRuntimeOf::<TickUTConfig>::sleep(Duration::from_millis(500)).await;
+        TickUTConfig::sleep(Duration::from_millis(500)).await;
 
         let mut received = vec![];
         while let Some(x) = rx.recv().await {

openraft/src/display_ext.rs

@@ -1,4 +1,4 @@
-//! Implement [`fmt::Display`] for types such as `Option<T>` and slice `&[T]`.
+//! Implement [`std::fmt::Display`] for types such as `Option<T>` and slice `&[T]`.
 
 pub(crate) mod display_instant;
 pub(crate) mod display_option;

openraft/src/engine/engine_impl.rs

@@ -42,11 +42,10 @@ use crate::raft::VoteResponse;
 use crate::raft_state::LogStateReader;
 use crate::raft_state::RaftState;
 use crate::summary::MessageSummary;
-use crate::type_config::alias::InstantOf;
 use crate::type_config::alias::ResponderOf;
 use crate::type_config::alias::SnapshotDataOf;
+use crate::type_config::TypeConfigExt;
 use crate::AsyncRuntime;
-use crate::Instant;
 use crate::LogId;
 use crate::LogIdOptionExt;
 use crate::Membership;
@@ -128,7 +127,7 @@ where C: RaftTypeConfig
     /// The candidate `last_log_id` is initialized with the attributes of Acceptor part:
     /// [`RaftState`]
     pub(crate) fn new_candidate(&mut self, vote: Vote<C::NodeId>) -> &mut Candidate<C, LeaderQuorumSet<C::NodeId>> {
-        let now = InstantOf::<C>::now();
+        let now = C::now();
         let last_log_id = self.state.last_log_id().copied();
 
         let membership = self.state.membership_state.effective().membership();
@@ -290,7 +289,7 @@ where C: RaftTypeConfig
 
     #[tracing::instrument(level = "debug", skip_all)]
     pub(crate) fn handle_vote_req(&mut self, req: VoteRequest<C::NodeId>) -> VoteResponse<C::NodeId> {
-        let now = <C::AsyncRuntime as AsyncRuntime>::Instant::now();
+        let now = C::now();
         let lease = self.config.timer_config.leader_lease;
         let vote = self.state.vote_ref();
 

openraft/src/engine/handler/vote_handler/accept_vote_test.rs

@@ -11,9 +11,8 @@ use crate::engine::Respond;
 use crate::error::Infallible;
 use crate::raft::VoteResponse;
 use crate::testing::log_id;
-use crate::type_config::alias::AsyncRuntimeOf;
+use crate::type_config::TypeConfigExt;
 use crate::utime::UTime;
-use crate::AsyncRuntime;
 use crate::EffectiveMembership;
 use crate::Membership;
 use crate::TokioInstant;
@@ -48,12 +47,12 @@ fn test_accept_vote_reject_smaller_vote() -> anyhow::Result<()> {
     let mut eng = eng();
     eng.output.take_commands();
 
-    let (tx, _rx) = AsyncRuntimeOf::<UTConfig>::oneshot();
+    let (tx, _rx) = UTConfig::oneshot();
     let resp = eng.vote_handler().accept_vote(&Vote::new(1, 2), tx, |_state, _err| mk_res());
 
     assert!(resp.is_none());
 
-    let (tx, _rx) = AsyncRuntimeOf::<UTConfig>::oneshot();
+    let (tx, _rx) = UTConfig::oneshot();
     assert_eq!(
         vec![
             //
@@ -74,7 +73,7 @@ fn test_accept_vote_granted_greater_vote() -> anyhow::Result<()> {
     let mut eng = eng();
     eng.output.take_commands();
 
-    let (tx, _rx) = AsyncRuntimeOf::<UTConfig>::oneshot();
+    let (tx, _rx) = UTConfig::oneshot();
     let resp = eng.vote_handler().accept_vote(&Vote::new(3, 3), tx, |_state, _err| mk_res());
 
     assert!(resp.is_some());

openraft/src/engine/handler/vote_handler/mod.rs

@@ -15,8 +15,8 @@ use crate::proposer::CandidateState;
 use crate::proposer::LeaderState;
 use crate::raft_state::LogStateReader;
 use crate::type_config::alias::InstantOf;
+use crate::type_config::TypeConfigExt;
 use crate::AsyncRuntime;
-use crate::Instant;
 use crate::LogId;
 use crate::OptionalSend;
 use crate::RaftState;
@@ -134,19 +134,15 @@ where C: RaftTypeConfig
         if vote > self.state.vote_ref() {
             tracing::info!("vote is changing from {} to {}", self.state.vote_ref(), vote);
 
-            self.state.vote.update(<C::AsyncRuntime as AsyncRuntime>::Instant::now(), *vote);
+            self.state.vote.update(C::now(), *vote);
             self.output.push_command(Command::SaveVote { vote: *vote });
         } else {
-            self.state.vote.touch(<C::AsyncRuntime as AsyncRuntime>::Instant::now());
+            self.state.vote.touch(C::now());
         }
 
         // Update vote related timer and lease.
 
-        tracing::debug!(
-            now = debug(<C::AsyncRuntime as AsyncRuntime>::Instant::now()),
-            "{}",
-            func_name!()
-        );
+        tracing::debug!(now = debug(C::now()), "{}", func_name!());
 
         self.update_internal_server_state();
 

openraft/src/engine/tests/install_full_snapshot_test.rs

@@ -12,8 +12,7 @@ use crate::engine::LogIdList;
 use crate::engine::Respond;
 use crate::raft::SnapshotResponse;
 use crate::testing::log_id;
-use crate::type_config::alias::AsyncRuntimeOf;
-use crate::AsyncRuntime;
+use crate::type_config::TypeConfigExt;
 use crate::Membership;
 use crate::Snapshot;
 use crate::SnapshotMeta;
@@ -63,7 +62,7 @@ fn test_handle_install_full_snapshot_lt_last_snapshot() -> anyhow::Result<()> {
 
     let curr_vote = *eng.state.vote_ref();
 
-    let (tx, _rx) = AsyncRuntimeOf::<UTConfig>::oneshot();
+    let (tx, _rx) = UTConfig::oneshot();
 
     eng.handle_install_full_snapshot(
         curr_vote,
@@ -87,7 +86,7 @@ fn test_handle_install_full_snapshot_lt_last_snapshot() -> anyhow::Result<()> {
         eng.state.snapshot_meta
     );
 
-    let (dummy_tx, _rx) = AsyncRuntimeOf::<UTConfig>::oneshot();
+    let (dummy_tx, _rx) = UTConfig::oneshot();
     assert_eq!(
         vec![
             //
@@ -111,7 +110,7 @@ fn test_handle_install_full_snapshot_no_conflict() -> anyhow::Result<()> {
 
     let curr_vote = *eng.state.vote_ref();
 
-    let (tx, _rx) = AsyncRuntimeOf::<UTConfig>::oneshot();
+    let (tx, _rx) = UTConfig::oneshot();
 
     eng.handle_install_full_snapshot(
         curr_vote,
@@ -135,7 +134,7 @@ fn test_handle_install_full_snapshot_no_conflict() -> anyhow::Result<()> {
         eng.state.snapshot_meta
     );
 
-    let (dummy_tx, _rx) = AsyncRuntimeOf::<UTConfig>::oneshot();
+    let (dummy_tx, _rx) = UTConfig::oneshot();
     assert_eq!(
         vec![
             //

openraft/src/lib.rs

@@ -46,7 +46,6 @@ pub(crate) mod log_id_range;
 pub(crate) mod proposer;
 pub(crate) mod raft_state;
 pub(crate) mod timer;
-pub(crate) mod type_config;
 pub(crate) mod utime;
 
 pub mod async_runtime;
@@ -63,6 +62,7 @@ pub mod network;
 pub mod raft;
 pub mod storage;
 pub mod testing;
+pub mod type_config;
 
 #[cfg(test)]
 mod feature_serde_test;

openraft/src/network/snapshot_transport.rs

@@ -20,8 +20,7 @@ use crate::error::StreamingError;
 use crate::network::RPCOption;
 use crate::raft::InstallSnapshotRequest;
 use crate::raft::SnapshotResponse;
-use crate::type_config::alias::AsyncRuntimeOf;
-use crate::AsyncRuntime;
+use crate::type_config::TypeConfigExt;
 use crate::ErrorSubject;
 use crate::ErrorVerb;
 use crate::OptionalSend;
@@ -124,7 +123,7 @@ where C::SnapshotData: tokio::io::AsyncRead + tokio::io::AsyncWrite + tokio::io:
             // Sleep a short time otherwise in test environment it is a dead-loop that never
             // yields.
             // Because network implementation does not yield.
-            AsyncRuntimeOf::<C>::sleep(Duration::from_millis(1)).await;
+            C::sleep(Duration::from_millis(1)).await;
 
             snapshot.snapshot.seek(SeekFrom::Start(offset)).await.sto_res(subject_verb)?;
 
@@ -160,7 +159,7 @@ where C::SnapshotData: tokio::io::AsyncRead + tokio::io::AsyncWrite + tokio::io:
             );
 
             #[allow(deprecated)]
-            let res = AsyncRuntimeOf::<C>::timeout(option.hard_ttl(), net.install_snapshot(req, option.clone())).await;
+            let res = C::timeout(option.hard_ttl(), net.install_snapshot(req, option.clone())).await;
 
             let resp = match res {
                 Ok(outer_res) => match outer_res {