reduce heartbeat message

src/raft.rs

@@ -545,41 +545,42 @@ impl<T: Storage> Raft<T> {
         is_batched
     }
 
-    /// Sends RPC, with entries to the given peer.
-    pub fn send_append(&mut self, to: u64, pr: &mut Progress) {
+    /// Sends RPC, with entries to the given peer. Returns true if a message was sent.
+    pub fn send_append(&mut self, to: u64, pr: &mut Progress) -> bool {
         if pr.is_paused() {
             trace!(
                 self.logger,
                 "Skipping sending to {to}, it's paused",
                 to = to;
                 "progress" => ?pr,
             );
-            return;
+            return false;
         }
         let mut m = Message::default();
         m.to = to;
         if pr.pending_request_snapshot != INVALID_INDEX {
             // Check pending request snapshot first to avoid unnecessary loading entries.
             if !self.prepare_send_snapshot(&mut m, pr, to) {
-                return;
+                return false;
             }
         } else {
             let term = self.raft_log.term(pr.next_idx - 1);
             let ents = self.raft_log.entries(pr.next_idx, self.max_msg_size);
             if term.is_err() || ents.is_err() {
                 // send snapshot if we failed to get term or entries.
                 if !self.prepare_send_snapshot(&mut m, pr, to) {
-                    return;
+                    return false;
                 }
             } else {
                 let mut ents = ents.unwrap();
                 if self.batch_append && self.try_batching(to, pr, &mut ents) {
-                    return;
+                    return true;
                 }
                 self.prepare_send_entries(&mut m, pr, term.unwrap(), ents);
             }
         }
         self.send(m);
+        true
     }
 
     // send_heartbeat sends an empty MsgAppend
@@ -606,10 +607,21 @@ impl<T: Storage> Raft<T> {
     pub fn bcast_append(&mut self) {
         let self_id = self.id;
         let mut prs = self.take_prs();
+        let mut bcast_msg = true;
         prs.iter_mut()
             .filter(|&(id, _)| *id != self_id)
-            .for_each(|(id, pr)| self.send_append(*id, pr));
+            .for_each(|(id, pr)| {
+                if !self.send_append(*id, pr) && bcast_msg {
+                    bcast_msg = false;
+                }
+            });
         self.set_prs(prs);
+        // When there is no pending Readindex, MsgAppend will do the
+        // work of MsgHeartbeat, so we can reset heartbeat_elapsed
+        // here to reduce MsgHeartbeat
+        if bcast_msg && self.read_only.last_pending_request_ctx().is_none() {
+            self.heartbeat_elapsed = 0;
+        }
     }
 
     /// Broadcasts heartbeats to all the followers if it's leader.