Merge pull request #25 from andreev-io/ilya/immediate-broadcast

Broadcast new transitions immediately, not waiting for heartbeat timer
andreev-io · Dec 1, 2021 · ba7a2e5 · ba7a2e5
2 parents 0e84c50 + 8f0e866
commit ba7a2e5
Show file tree

Hide file tree

Showing 3 changed files with 15 additions and 10 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/little_raft/Cargo.toml b/little_raft/Cargo.toml
@@ -1,7 +1,7 @@
 [package]
 description = "The lightest distributed consensus library. Run your own replicated state machine!"
 name = "little_raft"
-version = "0.1.5"
+version = "0.1.6"
 authors = ["Ilya Andreev <[email protected]>"]
 edition = "2018"
 license = "MIT"

diff --git a/little_raft/src/replica.rs b/little_raft/src/replica.rs
@@ -212,25 +212,30 @@ where
                 oper.recv(recv_transition)
                     .expect("could not react to a new transition");
                 self.load_new_transitions();
+                self.broadcast_append_entry_request();
             }
             // Broadcast heartbeat messages.
             i if i == heartbeat => {
                 oper.recv(recv_heartbeat)
                     .expect("could not react to the heartbeat");
-                self.broadcast_message(|peer_id: ReplicaID| Message::AppendEntryRequest {
-                    term: self.current_term,
-                    from_id: self.id,
-                    prev_log_index: self.next_index[&peer_id] - 1,
-                    prev_log_term: self.log[self.next_index[&peer_id] - 1].term,
-                    entries: self.get_entries_for_peer(peer_id),
-                    commit_index: self.commit_index,
-                });
+                self.broadcast_append_entry_request();
                 self.heartbeat_timer.renew();
             }
             _ => unreachable!(),
         }
     }
 
+    fn broadcast_append_entry_request(&mut self) {
+        self.broadcast_message(|peer_id: ReplicaID| Message::AppendEntryRequest {
+            term: self.current_term,
+            from_id: self.id,
+            prev_log_index: self.next_index[&peer_id] - 1,
+            prev_log_term: self.log[self.next_index[&peer_id] - 1].term,
+            entries: self.get_entries_for_peer(peer_id),
+            commit_index: self.commit_index,
+        });
+    }
+
     fn poll_as_follower(&mut self, recv_msg: &Receiver<()>) {
         match recv_msg.recv_deadline(self.next_election_deadline) {
             // Process pending messages.