BLE optimizations (#117)

* Updates leader election. Also adds std::time::Duration parsing to TestConfig TOML.

* Non-gossip leader election

* Cleaned up code

* Ensures stale leaders become followers in Paxos

* Restructure code and send BleState to OmniPaxos.

* Corrected leader happiness definition. Solution for HBRequest happiness.

* Removed need for ballots in HeartbeatRequests.

* Fixes, recovering leader edge case. Adds comments.

* Remove need for PendingLeader by syncing with Paxos promise,

* Addresses review comments

* Removes StaleLeader signaling. Fixes BLE-Paxos leader sync.

* Adds NotAccepted message.

* Stops leader from sending prepares to followers promised with a greater ballot.

* Rework promises_meta to Vec<PromiseState>

* Removed reset_promises()

* Refines follower ballot checking. Removes leader_state.decided_indexes.

* Addresses comments.

* Add comment for previous_replies in ble.

---------

Co-authored-by: Harald Ng <harald.ng@hotmail.com>

Author: kevin-harrison

omnipaxos/src/ballot_leader_election.rs

@@ -1,7 +1,10 @@
 use std::cmp::Ordering;
 
 /// Ballot Leader Election algorithm for electing new leaders
-use crate::util::{defaults::*, ConfigurationId, FlexibleQuorum, Quorum};
+use crate::{
+    sequence_paxos::{Phase, Role},
+    util::{defaults::*, ConfigurationId, FlexibleQuorum, Quorum},
+};
 
 #[cfg(feature = "logging")]
 use crate::utils::logger::create_logger;
@@ -15,7 +18,7 @@ use crate::{
 #[cfg(feature = "serde")]
 use serde::{Deserialize, Serialize};
 #[cfg(feature = "logging")]
-use slog::{debug, info, trace, warn, Logger};
+use slog::{info, trace, warn, Logger};
 
 /// Used to define a Sequence Paxos epoch
 #[derive(Clone, Copy, Eq, Debug, Default, PartialEq)]
@@ -59,8 +62,8 @@ impl PartialOrd for Ballot {
     }
 }
 
-/// The connectivity of an OmniPaxos node
-pub(crate) type Connectivity = u8;
+const INITIAL_ROUND: u32 = 1;
+const RECOVERY_ROUND: u32 = 0;
 
 /// A Ballot Leader Election component. Used in conjunction with OmniPaxos to handle the election of a leader for a cluster of OmniPaxos servers,
 /// incoming messages and produces outgoing messages that the user has to fetch periodically and send using a network implementation.
@@ -74,18 +77,18 @@ pub(crate) struct BallotLeaderElection {
     peers: Vec<NodeId>,
     /// The current round of the heartbeat cycle.
     hb_round: u32,
-    /// Vector which temporarily holds all the received heartbeats from one heartbeat round, including the current node.
-    ballots: Vec<(Ballot, Connectivity)>,
-    /// Vector that holds all the received heartbeats from the previous heartbeat round, including the current node.
+    /// The heartbeat replies this instance received during the current round.
+    heartbeat_replies: Vec<HeartbeatReply>,
+    /// Vector that holds all the received heartbeats from the previous heartbeat round, including the current node. Only used to display the connectivity of this node in the UI.
     /// Represents nodes that are currently alive from the view of the current node.
-    prev_round_ballots: Vec<(Ballot, Connectivity)>,
+    prev_replies: Vec<HeartbeatReply>,
     /// Holds the current ballot of this instance.
     current_ballot: Ballot,
-    /// The number of replicas inside the cluster that this instance is
-    /// connected to (based on heartbeats received) including itself.
-    connectivity: Connectivity,
-    /// Current elected leader.
-    leader: Option<Ballot>,
+    /// The current leader of this instance.
+    leader: Ballot,
+    /// A happy node either sees that it is, is connected to, or sees evidence of a potential leader
+    /// for the cluster. If a node is unhappy then it is seeking a new leader.
+    happy: bool,
     /// The number of replicas inside the cluster whose heartbeats are needed to become and remain the leader.
     quorum: Quorum,
     /// Vector which holds all the outgoing messages of the BLE instance.
@@ -97,23 +100,31 @@ pub(crate) struct BallotLeaderElection {
 
 impl BallotLeaderElection {
     /// Construct a new BallotLeaderElection node
-    pub(crate) fn with(config: BLEConfig, initial_leader: Option<Ballot>) -> Self {
+    pub(crate) fn with(config: BLEConfig, recovered_leader: Option<Ballot>) -> Self {
         let config_id = config.configuration_id;
         let pid = config.pid;
         let peers = config.peers;
         let num_nodes = &peers.len() + 1;
         let quorum = Quorum::with(config.flexible_quorum, num_nodes);
-        let initial_ballot = Ballot::with(config_id, 0, config.priority, pid);
+        let mut initial_ballot = Ballot::with(config_id, INITIAL_ROUND, config.priority, pid);
+        let initial_leader = match recovered_leader {
+            Some(b) if b != Ballot::default() => {
+                // Prevents a recovered server from retaining BLE leadership with the same ballot.
+                initial_ballot.n = RECOVERY_ROUND;
+                b
+            }
+            _ => initial_ballot,
+        };
         let mut ble = BallotLeaderElection {
             configuration_id: config_id,
             pid,
             peers,
             hb_round: 0,
-            ballots: Vec::with_capacity(num_nodes),
-            prev_round_ballots: Vec::with_capacity(num_nodes),
+            heartbeat_replies: Vec::with_capacity(num_nodes),
+            prev_replies: Vec::with_capacity(num_nodes),
             current_ballot: initial_ballot,
-            connectivity: num_nodes as Connectivity,
             leader: initial_leader,
+            happy: true,
             quorum,
             outgoing: Vec::with_capacity(config.buffer_size),
             #[cfg(feature = "logging")]
@@ -160,70 +171,20 @@ impl BallotLeaderElection {
         }
     }
 
-    fn check_leader(&mut self) -> Option<Ballot> {
-        let ballots = std::mem::take(&mut self.ballots);
-        let top_accept_ballot = ballots
-            .iter()
-            .filter_map(|&(ballot, connectivity)| {
-                if self.quorum.is_accept_quorum(connectivity as usize) {
-                    Some(ballot)
-                } else {
-                    None
-                }
-            })
-            .max()
-            .unwrap_or_default();
-        let leader_ballot = self.leader.unwrap_or_default();
-        if top_accept_ballot == leader_ballot {
-            // leader is still alive and has accept quorum
-            None
-        } else {
-            // leader is dead || changed priority || doesn't have an accept quorum
-            let top_prepare_ballot = ballots
-                .iter()
-                .filter_map(|&(ballot, connectivity)| {
-                    if self.quorum.is_prepare_quorum(connectivity as usize) {
-                        Some(ballot)
-                    } else {
-                        None
-                    }
-                })
-                .max()
-                .unwrap_or_default();
-            if top_prepare_ballot > leader_ballot {
-                // new leader with prepare quorum
-                let new_leader = top_prepare_ballot;
-                self.leader = Some(new_leader);
-                #[cfg(feature = "logging")]
-                debug!(
-                    self.logger,
-                    "BLE {}, New Leader elected: {:?}", self.pid, new_leader
-                );
-                Some(new_leader)
-            } else {
-                // nobody has taken over leadership, let's try to ourselves
-                self.current_ballot.n = leader_ballot.n + 1;
-                self.leader = None;
-                None
-            }
-        }
-    }
-
     /// Initiates a new heartbeat round.
     pub(crate) fn new_hb_round(&mut self) {
+        self.prev_replies = std::mem::take(&mut self.heartbeat_replies);
         self.hb_round += 1;
         #[cfg(feature = "logging")]
         trace!(
             self.logger,
             "Initiate new heartbeat round: {}",
             self.hb_round
         );
-
         for peer in &self.peers {
             let hb_request = HeartbeatRequest {
                 round: self.hb_round,
             };
-
             self.outgoing.push(BLEMessage {
                 from: self.pid,
                 to: *peer,
@@ -232,41 +193,89 @@ impl BallotLeaderElection {
         }
     }
 
-    pub(crate) fn hb_timeout(&mut self) -> Option<Ballot> {
-        let my_connectivity = self.ballots.len() + 1;
-        self.connectivity = my_connectivity as Connectivity;
-        // Add our own ballot to the list of received ballots of current hb round
-        self.ballots.push((self.current_ballot, self.connectivity));
-        self.prev_round_ballots = self.ballots.clone();
-        let result: Option<Ballot> = if self.quorum.is_prepare_quorum(my_connectivity) {
-            #[cfg(feature = "logging")]
-            debug!(
-                self.logger,
-                "Received a majority of heartbeats, round: {}, {:?}", self.hb_round, self.ballots
-            );
-            self.check_leader()
+    /// End of a heartbeat round. Returns current leader and election status.
+    pub(crate) fn hb_timeout(
+        &mut self,
+        seq_paxos_state: &(Role, Phase),
+        seq_paxos_promise: Ballot,
+    ) -> Option<Ballot> {
+        self.update_leader();
+        self.update_happiness(seq_paxos_state);
+        self.check_takeover();
+        self.new_hb_round();
+        if seq_paxos_promise > self.leader {
+            // Sync leader with Paxos promise in case ballot didn't make it to BLE followers
+            self.leader = seq_paxos_promise;
+            self.happy = true;
+        }
+        if self.leader == self.current_ballot {
+            Some(self.current_ballot)
         } else {
-            #[cfg(feature = "logging")]
-            warn!(
-                self.logger,
-                "Did not receive a majority of heartbeats, round: {}, {:?}",
-                self.hb_round,
-                self.ballots
-            );
-            self.ballots.clear();
             None
+        }
+    }
+
+    fn update_leader(&mut self) {
+        let max_reply_ballot = self.heartbeat_replies.iter().map(|r| r.ballot).max();
+        if let Some(max) = max_reply_ballot {
+            if max > self.leader {
+                self.leader = max;
+            }
+        }
+    }
+
+    fn update_happiness(&mut self, seq_paxos_state: &(Role, Phase)) {
+        self.happy = if self.leader == self.current_ballot {
+            let potential_followers = self
+                .heartbeat_replies
+                .iter()
+                .filter(|hb_reply| hb_reply.leader <= self.current_ballot)
+                .count();
+            let can_form_quorum = match seq_paxos_state {
+                (Role::Leader, Phase::Accept) => {
+                    self.quorum.is_accept_quorum(potential_followers + 1)
+                }
+                _ => self.quorum.is_prepare_quorum(potential_followers + 1),
+            };
+            if can_form_quorum {
+                true
+            } else {
+                let see_larger_happy_leader = self
+                    .heartbeat_replies
+                    .iter()
+                    .any(|r| r.leader > self.current_ballot && r.happy);
+                see_larger_happy_leader
+            }
+        } else {
+            self.heartbeat_replies
+                .iter()
+                .any(|r| r.ballot == self.leader && r.happy)
         };
-        self.new_hb_round();
-        result
+    }
+
+    fn check_takeover(&mut self) {
+        if !self.happy {
+            let all_neighbors_unhappy = self.heartbeat_replies.iter().all(|r| !r.happy);
+            let im_quorum_connected = self
+                .quorum
+                .is_prepare_quorum(self.heartbeat_replies.len() + 1);
+            if all_neighbors_unhappy && im_quorum_connected {
+                // We increment past our leader instead of max of unhappy ballots because we
+                // assume we have already checked leader for this round so they should be equal
+                self.current_ballot.n = self.leader.n + 1;
+                self.leader = self.current_ballot;
+                self.happy = true;
+            }
+        }
     }
 
     fn handle_request(&mut self, from: NodeId, req: HeartbeatRequest) {
         let hb_reply = HeartbeatReply {
             round: req.round,
             ballot: self.current_ballot,
-            connectivity: self.connectivity,
+            leader: self.leader,
+            happy: self.happy,
         };
-
         self.outgoing.push(BLEMessage {
             from: self.pid,
             to: from,
@@ -276,7 +285,7 @@ impl BallotLeaderElection {
 
     fn handle_reply(&mut self, rep: HeartbeatReply) {
         if rep.round == self.hb_round && rep.ballot.config_id == self.configuration_id {
-            self.ballots.push((rep.ballot, rep.connectivity));
+            self.heartbeat_replies.push(rep);
         } else {
             #[cfg(feature = "logging")]
             warn!(
@@ -290,8 +299,8 @@ impl BallotLeaderElection {
         self.current_ballot
     }
 
-    pub(crate) fn get_ballots(&self) -> Vec<(Ballot, Connectivity)> {
-        self.prev_round_ballots.clone()
+    pub(crate) fn get_ballots(&self) -> Vec<HeartbeatReply> {
+        self.prev_replies.clone()
     }
 }
 

omnipaxos/src/messages.rs

@@ -17,6 +17,14 @@ pub mod sequence_paxos {
     use serde::{Deserialize, Serialize};
     use std::fmt::Debug;
 
+    /// Message sent by a follower on crash-recovery or dropped messages to request its leader to re-prepare them.
+    #[derive(Copy, Clone, Debug)]
+    #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+    pub struct PrepareReq {
+        /// The current round.
+        pub n: Ballot,
+    }
+
     /// Prepare message sent by a newly-elected leader to initiate the Prepare phase.
     #[derive(Copy, Clone, Debug)]
     #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
@@ -128,6 +136,15 @@ pub mod sequence_paxos {
         pub ss: StopSign,
     }
 
+    /// Message sent by follower to leader when accepting an entry is rejected.
+    /// This happens when the follower is promised to a greater leader.
+    #[derive(Clone, Debug)]
+    #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+    pub struct NotAccepted {
+        /// The follower's current ballot
+        pub n: Ballot,
+    }
+
     /// Compaction Request
     #[allow(missing_docs)]
     #[derive(Clone, Debug)]
@@ -146,13 +163,14 @@ pub mod sequence_paxos {
         T: Entry,
     {
         /// Request a [`Prepare`] to be sent from the leader. Used for fail-recovery.
-        PrepareReq,
+        PrepareReq(PrepareReq),
         #[allow(missing_docs)]
         Prepare(Prepare),
         Promise(Promise<T>),
         AcceptSync(AcceptSync<T>),
         AcceptDecide(AcceptDecide<T>),
         Accepted(Accepted),
+        NotAccepted(NotAccepted),
         Decide(Decide),
         /// Forward client proposals to the leader.
         ProposalForward(Vec<T>),
@@ -207,10 +225,12 @@ pub mod ballot_leader_election {
     pub struct HeartbeatReply {
         /// Number of the current heartbeat round.
         pub round: u32,
-        /// Ballot of a server.
+        /// Ballot of replying server.
         pub ballot: Ballot,
-        /// The number of replicas inside the cluster the sender is connected to (including itself)
-        pub connectivity: u8,
+        /// Leader this server is following
+        pub leader: Ballot,
+        /// Whether the replying server sees a need for a new leader
+        pub happy: bool,
     }
 
     /// A struct for a Paxos message that also includes sender and receiver.

omnipaxos/src/omni_paxos.rs

@@ -391,21 +391,24 @@ where
     /// It is also used for the election process, where the server checks if it can become the leader.
     /// For instance if `election_timeout()` is called every 100ms, then if the leader fails, the servers will detect it after 100ms and elect a new server after another 100ms if possible.
     fn election_timeout(&mut self) {
-        if let Some(b) = self.ble.hb_timeout() {
-            self.seq_paxos.handle_leader(b);
+        if let Some(new_leader) = self
+            .ble
+            .hb_timeout(self.seq_paxos.get_state(), self.seq_paxos.get_promise())
+        {
+            self.seq_paxos.handle_leader(new_leader);
         }
     }
 
     /// Returns the current states of the OmniPaxos instance for OmniPaxos UI to display.
     pub fn get_ui_states(&self) -> ui::OmniPaxosStates {
         let mut cluster_state = ClusterState::from(self.seq_paxos.get_leader_state());
-        cluster_state.ballots = self.ble.get_ballots();
+        cluster_state.heartbeats = self.ble.get_ballots();
 
         ui::OmniPaxosStates {
             current_ballot: self.ble.get_current_ballot(),
             current_leader: self.get_current_leader(),
             decided_idx: self.get_decided_idx(),
-            ballots: self.ble.get_ballots(),
+            heartbeats: self.ble.get_ballots(),
             cluster_state,
         }
     }