[refactor]: use an event listener to replace the Shutdown struct.

Closes xline-kv#174

curp/src/lib.rs

@@ -151,8 +151,5 @@ pub(crate) mod conflict_checked_mpmc;
 /// Protobuf generated types that are used in RPC
 mod rpc;
 
-/// Shutdown related
-mod shutdown;
-
 #[cfg(test)]
 pub(crate) mod test_utils;

curp/src/server/bg_tasks.rs

@@ -8,6 +8,7 @@ use std::{
 };
 
 use clippy_utilities::NumericCast;
+use event_listener::Event;
 use futures::{pin_mut, stream::FuturesUnordered, Stream, StreamExt};
 use itertools::Itertools;
 use madsim::rand::{thread_rng, Rng};
@@ -38,7 +39,6 @@ use crate::{
         },
         ServerRole, State,
     },
-    shutdown::Shutdown,
 };
 
 /// Run background tasks
@@ -55,7 +55,7 @@ pub(super) async fn run_bg_tasks<
     cmd_exe_tx: ExeTx,
     cmd_exe_rx: CmdExeReceiver<C>,
     cmd_as_rx: CmdAsReceiver<C>,
-    mut shutdown: Shutdown,
+    shutdown_trigger: Arc<Event>,
     timeout: Arc<ServerTimeout>,
     #[cfg(test)] reachable: Arc<AtomicBool>,
 ) {
@@ -114,7 +114,7 @@ pub(super) async fn run_bg_tasks<
     })
     .collect();
 
-    shutdown.recv().await;
+    shutdown_trigger.listen().await;
     bg_ae_handle.abort();
     bg_election_handle.abort();
     bg_apply_handle.abort();

curp/src/server/mod.rs

@@ -39,7 +39,6 @@ use crate::{
         cmd_worker::{cmd_exe_channel, CmdExeSender},
         spec_pool::SpecPoolRef,
     },
-    shutdown::Shutdown,
 };
 
 /// Background tasks of Curp protocol
@@ -224,7 +223,7 @@ pub struct Protocol<C: Command + 'static> {
     /// Cmd watch board for tracking the cmd sync results
     cmd_board: CmdBoardRef<C>,
     /// Stop channel sender
-    stop_tx: broadcast::Sender<()>,
+    shutdown_trigger: Arc<Event>,
     /// The channel to send cmds to background exe tasks
     cmd_exe_tx: CmdExeSender<C>,
     /// The curp server timeout
@@ -294,7 +293,7 @@ impl<C: 'static + Command> Protocol<C> {
         timeout: Arc<ServerTimeout>,
     ) -> Self {
         let (sync_tx, sync_rx) = flume::unbounded();
-        let (stop_tx, stop_rx) = broadcast::channel(1);
+        let shutdown_trigger = Arc::new(Event::new());
         let (exe_tx, exe_rx, as_rx) = cmd_exe_channel();
 
         let state = State::new(
@@ -320,7 +319,7 @@ impl<C: 'static + Command> Protocol<C> {
             exe_tx.clone(),
             exe_rx,
             as_rx,
-            Shutdown::new(stop_rx.resubscribe()),
+            Arc::clone(&shutdown_trigger),
             Arc::clone(&timeout),
             #[cfg(test)]
             Arc::new(AtomicBool::new(true)),
@@ -334,7 +333,7 @@ impl<C: 'static + Command> Protocol<C> {
             spec,
             sync_tx,
             cmd_board,
-            stop_tx,
+            shutdown_trigger,
             cmd_exe_tx: exe_tx,
             timeout,
         }
@@ -350,7 +349,7 @@ impl<C: 'static + Command> Protocol<C> {
         reachable: Arc<AtomicBool>,
     ) -> Self {
         let (sync_tx, sync_rx) = flume::unbounded();
-        let (stop_tx, stop_rx) = broadcast::channel(1);
+        let shutdown_trigger = Arc::new(Event::new());
         let (exe_tx, exe_rx, as_rx) = cmd_exe_channel();
 
         let state = State::new(
@@ -376,7 +375,7 @@ impl<C: 'static + Command> Protocol<C> {
             exe_tx.clone(),
             exe_rx,
             as_rx,
-            Shutdown::new(stop_rx.resubscribe()),
+            Arc::clone(&shutdown_trigger),
             Arc::clone(&timeout),
             reachable,
         ));
@@ -389,7 +388,7 @@ impl<C: 'static + Command> Protocol<C> {
             spec,
             sync_tx,
             cmd_board,
-            stop_tx,
+            shutdown_trigger,
             cmd_exe_tx: exe_tx,
             timeout,
         }
@@ -710,7 +709,7 @@ impl<C: 'static + Command> Drop for Protocol<C> {
     #[inline]
     fn drop(&mut self) {
         // TODO: async drop is still not supported by Rust(should wait for bg tasks to be stopped?), or we should create an async `stop` function for Protocol
-        let _ = self.stop_tx.send(()).ok();
+        self.shutdown_trigger.notify(usize::MAX);
     }
 }