raft: handle separately each timeout durations

bin/darkwikid/src/main.rs

@@ -575,8 +575,6 @@ async fn realmain(settings: Args, executor: Arc<Executor<'_>>) -> Result<()> {
 
     executor.spawn(p2p.clone().run(executor.clone())).detach();
 
-    p2p.clone().wait_for_outbound(executor.clone()).await?;
-
     //
     // Darkwiki start
     //

bin/tau/taud/src/main.rs

@@ -221,8 +221,6 @@ async fn realmain(settings: Args, executor: Arc<Executor<'_>>) -> Result<()> {
 
     executor.spawn(p2p.clone().run(executor.clone())).detach();
 
-    p2p.clone().wait_for_outbound(executor.clone()).await?;
-
     //
     // RPC interface
     //

src/raft/consensus.rs

@@ -1,6 +1,6 @@
 use async_std::{
     sync::{Arc, Mutex},
-    task,
+    task::sleep,
 };
 use std::time::Duration;
 
@@ -14,7 +14,7 @@ use rand::{rngs::OsRng, Rng, RngCore};
 use crate::{
     net,
     util::{
-        self, gen_id,
+        gen_id,
         serial::{deserialize, serialize, Decodable, Encodable},
     },
     Error, Result,
@@ -29,9 +29,9 @@ use super::{
     prune_map, DataStore, RaftSettings,
 };
 
-async fn send_node_id_loop(sender: async_channel::Sender<()>, timeout: i64) -> Result<()> {
+async fn send_loop(sender: async_channel::Sender<()>, timeout: Duration) -> Result<()> {
     loop {
-        util::sleep(timeout as u64).await;
+        sleep(timeout).await;
         sender.send(()).await?;
     }
 }
@@ -52,6 +52,8 @@ pub struct Raft<T> {
 
     pub(super) last_term: u64,
 
+    pub(super) last_heartbeat: i64,
+
     p2p_sender: Sender,
 
     msgs_channel: Channel<T>,
@@ -61,7 +63,7 @@ pub struct Raft<T> {
 
     seen_msgs: Arc<Mutex<FxHashMap<String, i64>>>,
 
-    settings: RaftSettings,
+    pub(super) settings: RaftSettings,
 
     pending_msgs: Vec<T>,
 }
@@ -104,6 +106,7 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
             acked_length: MapLength(FxHashMap::default()),
             nodes: Arc::new(Mutex::new(FxHashMap::default())),
             last_term: 0,
+            last_heartbeat: Utc::now().timestamp(),
             p2p_sender,
             msgs_channel,
             commits_channel,
@@ -126,45 +129,39 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
     ) -> Result<()> {
         let p2p_send_task = executor.spawn(p2p_send_loop(self.p2p_sender.1.clone(), p2p.clone()));
 
-        let prune_seen_messages_task = executor.spawn(prune_map::<String>(
-            self.seen_msgs.clone(),
-            self.settings.prun_messages_duration,
-        ));
-
-        let prune_nodes_id_task = executor
-            .spawn(prune_map::<NodeId>(self.nodes.clone(), self.settings.prun_nodes_ids_duration));
+        let prune_seen_messages_task = executor
+            .spawn(prune_map::<String>(self.seen_msgs.clone(), self.settings.prun_duration));
 
-        let (node_id_sx, node_id_rv) = async_channel::unbounded::<()>();
-        let send_node_id_loop_task =
-            executor.spawn(send_node_id_loop(node_id_sx, self.settings.node_id_timeout));
+        let prune_nodes_id_task =
+            executor.spawn(prune_map::<NodeId>(self.nodes.clone(), self.settings.prun_duration));
 
         let mut rng = rand::thread_rng();
 
+        let (id_sx, id_rv) = async_channel::unbounded::<()>();
+        let (heartbeat_sx, heartbeat_rv) = async_channel::unbounded::<()>();
+        let (timeout_sx, timeout_rv) = async_channel::unbounded::<()>();
+
+        let id_timeout = Duration::from_secs(self.settings.id_timeout);
+        let send_id_task = executor.spawn(send_loop(id_sx, id_timeout));
+
+        let heartbeat_timeout = Duration::from_millis(self.settings.heartbeat_timeout);
+        let send_heartbeat_task = executor.spawn(send_loop(heartbeat_sx, heartbeat_timeout));
+
+        let timeout =
+            Duration::from_secs(rng.gen_range(0..self.settings.timeout) + self.settings.timeout);
+        let send_timeout_task = executor.spawn(send_loop(timeout_sx, timeout));
+
         let broadcast_msg_rv = self.msgs_channel.1.clone();
 
         loop {
-            let timeout = if self.role == Role::Leader {
-                self.settings.heartbeat_timeout
-            } else {
-                rng.gen_range(0..self.settings.timeout) + self.settings.timeout
-            };
-            let timeout = Duration::from_millis(timeout);
-
-            let mut result: Result<()>;
-
-            select! {
-                m =  p2p_recv_channel.recv().fuse() => result = self.handle_method(m?).await,
-                m =  broadcast_msg_rv.recv().fuse() => result = self.broadcast_msg(&m?,None).await,
-                _ =  node_id_rv.recv().fuse() => result = self.send_node_id_msg().await,
-                _ = task::sleep(timeout).fuse() => {
-                    result = if self.role == Role::Leader {
-                        self.send_heartbeat().await
-                    }else {
-                        self.send_vote_request().await
-                    };
-                },
+            let mut result = select! {
+                m =  p2p_recv_channel.recv().fuse() => self.handle_method(m?).await,
+                m =  broadcast_msg_rv.recv().fuse() => self.broadcast_msg(&m?,None).await,
+                _ =  id_rv.recv().fuse() => self.send_id_msg().await,
+                _ = heartbeat_rv.recv().fuse() => self.send_heartbeat().await,
+                _ = timeout_rv.recv().fuse() => self.send_vote_request().await,
                 _ = stop_signal.recv().fuse() => break,
-            }
+            };
 
             // send pending messages
             if !self.pending_msgs.is_empty() && self.role != Role::Candidate {
@@ -175,9 +172,8 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
                 self.pending_msgs = vec![];
             }
 
-            match result {
-                Ok(_) => {}
-                Err(e) => warn!(target: "raft", "warn: {}", e),
+            if let Err(e) = result {
+                warn!(target: "raft", "warn: {}", e);
             }
         }
 
@@ -185,7 +181,9 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
         p2p_send_task.cancel().await;
         prune_seen_messages_task.cancel().await;
         prune_nodes_id_task.cancel().await;
-        send_node_id_loop_task.cancel().await;
+        send_id_task.cancel().await;
+        send_heartbeat_task.cancel().await;
+        send_timeout_task.cancel().await;
         self.datastore.flush().await?;
         Ok(())
     }
@@ -213,9 +211,9 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
         self.id.clone()
     }
 
-    async fn send_node_id_msg(&self) -> Result<()> {
-        let node_id_msg = serialize(&NodeIdMsg { id: self.id.clone() });
-        self.send(None, &node_id_msg, NetMsgMethod::NodeIdMsg, None).await?;
+    async fn send_id_msg(&self) -> Result<()> {
+        let id_msg = serialize(&NodeIdMsg { id: self.id.clone() });
+        self.send(None, &id_msg, NetMsgMethod::NodeIdMsg, None).await?;
         Ok(())
     }
 
@@ -238,7 +236,6 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
                 .await?;
             }
             Role::Candidate => {
-                warn!("The role is Candidate, add the msg to pending_msgs");
                 self.pending_msgs.push(msg.clone());
             }
         }
@@ -255,6 +252,7 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
                 self.receive_log_response(lr).await?;
             }
             NetMsgMethod::LogRequest => {
+                self.last_heartbeat = Utc::now().timestamp();
                 let lr: LogRequest = deserialize(&msg.payload)?;
                 self.receive_log_request(lr).await?;
             }

src/raft/consensus_candidate.rs

@@ -1,3 +1,4 @@
+use chrono::Utc;
 use log::info;
 
 use crate::{
@@ -12,6 +13,16 @@ use super::{
 
 impl<T: Decodable + Encodable + Clone> Raft<T> {
     pub(super) async fn send_vote_request(&mut self) -> Result<()> {
+        if self.role == Role::Leader {
+            return Ok(())
+        }
+
+        let last_heartbeat_duration = Utc::now().timestamp() - self.last_heartbeat;
+
+        if last_heartbeat_duration < self.settings.timeout as i64 {
+            return Ok(())
+        }
+
         let self_id = self.id();
 
         self.set_current_term(&(self.current_term()? + 1))?;

src/raft/consensus_leader.rs

@@ -12,6 +12,10 @@ use super::{
 
 impl<T: Decodable + Encodable + Clone> Raft<T> {
     pub(super) async fn send_heartbeat(&mut self) -> Result<()> {
+        if self.role != Role::Leader {
+            return Ok(())
+        }
+
         let nodes = self.nodes.lock().await;
         let nodes_cloned = nodes.clone();
         drop(nodes);

src/raft/settings.rs

@@ -2,23 +2,19 @@ use std::path::PathBuf;
 
 #[derive(Clone, Debug)]
 pub struct RaftSettings {
-    //
-    // Milliseconds
-    //
+    // the leader duration for sending heartbeat; in milliseconds
     pub heartbeat_timeout: u64,
+
+    // the duration for electing new leader; in seconds
     pub timeout: u64,
 
-    //
-    // Seconds
-    //
-    pub prun_messages_duration: i64,
-    pub prun_nodes_ids_duration: i64,
-    // must be greater than (timeout * 2)
-    pub node_id_timeout: i64,
+    // the duration for sending id to other nodes; in seconds
+    pub id_timeout: u64,
+
+    // this duration used to clean up hashmaps; in seconds
+    pub prun_duration: i64,
 
-    //
     // Datastore path
-    //
     pub datastore_path: PathBuf,
 }
 
@@ -26,10 +22,9 @@ impl Default for RaftSettings {
     fn default() -> Self {
         Self {
             heartbeat_timeout: 500,
-            timeout: 7000,
-            prun_messages_duration: 120,
-            prun_nodes_ids_duration: 120,
-            node_id_timeout: 16,
+            timeout: 6,
+            id_timeout: 12,
+            prun_duration: 240,
             datastore_path: PathBuf::from(""),
         }
     }