Outgoing messages buffer reuse (#150)

* More reasonable defaults

* Adds become_leader() for manual leader takeover.

* try_become_leader() with Ballot increment.

* Changes outgoing_messages() to take a buffer instead of allocating new
vec.

* `take_outgoing_messages` rename and cleaner buffered message logic.

* Crash additional follower in sync tests to ensure entries remain
non-decided.

* Comment out GithuB action cargo caching

* Update docs

Co-authored-by: Harald Ng <[email protected]>

* Renamed buffered to latest and fixed clippy warnings

* Reintroduce GitHub actions cargo caching.

* Update omnipaxos/tests/sync_test.rs

* Revert "Update omnipaxos/tests/sync_test.rs"

This reverts commit 6bfedc1.

---------

Co-authored-by: Harald Ng <[email protected]>

Author: kevin-harrison

docs/omnipaxos/communication.md

@@ -15,7 +15,9 @@ By handling incoming messages and local calls such as `append()`, our local `omn
 
 ```rust
 // send outgoing messages. This should be called periodically, e.g. every ms
-for out_msg in omni_paxos.outgoing_messages() {
+let mut msg_buffer = Vec::with_capacity(1000);
+omni_paxos.take_outgoing_messages(&mut msg_buffer);
+for out_msg in msg_buffer.drain(..) {
     let receiver = out_msg.get_receiver();
     // send out_msg to receiver on network layer
 }

examples/dashboard/src/main.rs

@@ -61,6 +61,7 @@ fn main() {
             omni_paxos: Arc::clone(&omni_paxos),
             incoming: receiver_channels.remove(pid).unwrap(),
             outgoing: sender_channels.clone(),
+            message_buffer: vec![],
         };
         let join_handle = runtime.spawn({
             async move {

examples/dashboard/src/server.rs

@@ -16,12 +16,16 @@ pub struct OmniPaxosServer {
     pub omni_paxos: Arc<Mutex<OmniPaxosLog>>,
     pub incoming: mpsc::Receiver<Message<LogEntry>>,
     pub outgoing: HashMap<NodeId, mpsc::Sender<Message<LogEntry>>>,
+    pub message_buffer: Vec<Message<LogEntry>>,
 }
 
 impl OmniPaxosServer {
     async fn send_outgoing_msgs(&mut self) {
-        let messages = self.omni_paxos.lock().unwrap().outgoing_messages();
-        for msg in messages {
+        self.omni_paxos
+            .lock()
+            .unwrap()
+            .take_outgoing_messages(&mut self.message_buffer);
+        for msg in self.message_buffer.drain(..) {
             let receiver = msg.get_receiver();
             let channel = self
                 .outgoing

examples/kv_store/src/main.rs

@@ -68,6 +68,7 @@ fn main() {
             omni_paxos: Arc::clone(&omni_paxos),
             incoming: receiver_channels.remove(&pid).unwrap(),
             outgoing: sender_channels.clone(),
+            message_buffer: vec![],
         };
         let join_handle = runtime.spawn({
             async move {

examples/kv_store/src/server.rs

@@ -15,12 +15,16 @@ pub struct OmniPaxosServer {
     pub omni_paxos: Arc<Mutex<OmniPaxosKV>>,
     pub incoming: mpsc::Receiver<Message<KeyValue>>,
     pub outgoing: HashMap<NodeId, mpsc::Sender<Message<KeyValue>>>,
+    pub message_buffer: Vec<Message<KeyValue>>,
 }
 
 impl OmniPaxosServer {
     async fn send_outgoing_msgs(&mut self) {
-        let messages = self.omni_paxos.lock().unwrap().outgoing_messages();
-        for msg in messages {
+        self.omni_paxos
+            .lock()
+            .unwrap()
+            .take_outgoing_messages(&mut self.message_buffer);
+        for msg in self.message_buffer.drain(..) {
             let receiver = msg.get_receiver();
             let channel = self
                 .outgoing

omnipaxos/src/ballot_leader_election.rs

@@ -156,9 +156,9 @@ impl BallotLeaderElection {
         self.current_ballot.priority = p;
     }
 
-    /// Returns outgoing messages
-    pub(crate) fn get_outgoing_msgs(&mut self) -> Vec<BLEMessage> {
-        std::mem::take(&mut self.outgoing)
+    /// Returns reference to outgoing messages
+    pub(crate) fn outgoing_mut(&mut self) -> &mut Vec<BLEMessage> {
+        &mut self.outgoing
     }
 
     /// Handle an incoming message.
@@ -205,7 +205,11 @@ impl BallotLeaderElection {
         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
+            // or become_leader() was called.
             self.leader = seq_paxos_promise;
+            if seq_paxos_promise.pid == self.pid {
+                self.current_ballot = seq_paxos_promise;
+            }
             self.happy = true;
         }
         if self.leader == self.current_ballot {

omnipaxos/src/omni_paxos.rs

@@ -289,19 +289,10 @@ where
         self.seq_paxos.get_promise()
     }
 
-    /// Returns the outgoing messages from this server. The messages should then be sent via the network implementation.
-    pub fn outgoing_messages(&mut self) -> Vec<Message<T>> {
-        let paxos_msgs = self
-            .seq_paxos
-            .get_outgoing_msgs()
-            .into_iter()
-            .map(|p| Message::SequencePaxos(p));
-        let ble_msgs = self
-            .ble
-            .get_outgoing_msgs()
-            .into_iter()
-            .map(|b| Message::BLE(b));
-        ble_msgs.chain(paxos_msgs).collect()
+    /// Moves outgoing messages from this server into the buffer. The messages should then be sent via the network implementation.
+    pub fn take_outgoing_messages(&mut self, buffer: &mut Vec<Message<T>>) {
+        self.seq_paxos.take_outgoing_msgs(buffer);
+        buffer.extend(self.ble.outgoing_mut().drain(..).map(|b| Message::BLE(b)));
     }
 
     /// Read entry at index `idx` in the log. Returns `None` if `idx` is out of bounds.
@@ -394,6 +385,15 @@ where
         }
     }
 
+    /// Manually attempt to become the leader by incrementing this instance's Ballot. Calling this
+    /// function may not result in gainig leadership if other instances are competing for
+    /// leadership with higher Ballots.
+    pub fn try_become_leader(&mut self) {
+        let mut my_ballot = self.ble.get_current_ballot();
+        my_ballot.n += 1;
+        self.seq_paxos.handle_leader(my_ballot);
+    }
+
     /*** BLE calls ***/
     /// Update the custom priority used in the Ballot for this server. Note that changing the
     /// priority triggers a leader re-election.

omnipaxos/src/sequence_paxos/follower.rs

@@ -42,11 +42,11 @@ where
                 log_sync,
             };
             self.cached_promise_message = Some(promise.clone());
-            self.outgoing.push(PaxosMessage {
+            self.outgoing.push(Message::SequencePaxos(PaxosMessage {
                 from: self.pid,
                 to: from,
                 msg: PaxosMsg::Promise(promise),
-            });
+            }));
         }
     }
 
@@ -68,11 +68,11 @@ where
             self.current_seq_num = accsync.seq_num;
             let cached_idx = self.outgoing.len();
             self.latest_accepted_meta = Some((accsync.n, cached_idx));
-            self.outgoing.push(PaxosMessage {
+            self.outgoing.push(Message::SequencePaxos(PaxosMessage {
                 from: self.pid,
                 to: from,
                 msg: PaxosMsg::Accepted(accepted),
-            });
+            }));
             #[cfg(feature = "unicache")]
             self.internal_storage.set_unicache(accsync.unicache);
         }
@@ -158,27 +158,38 @@ where
     }
 
     fn reply_accepted(&mut self, n: Ballot, accepted_idx: usize) {
-        match &self.latest_accepted_meta {
-            Some((round, outgoing_idx)) if round == &n => {
-                let PaxosMessage { msg, .. } = self.outgoing.get_mut(*outgoing_idx).unwrap();
-                match msg {
-                    PaxosMsg::Accepted(a) => {
-                        a.accepted_idx = accepted_idx;
-                    }
-                    _ => panic!("Cached idx is not an Accepted Message<T>!"),
-                }
-            }
-            _ => {
+        let latest_accepted = self.get_latest_accepted_message(n);
+        match latest_accepted {
+            Some(acc) => acc.accepted_idx = accepted_idx,
+            None => {
                 let accepted = Accepted { n, accepted_idx };
                 let cached_idx = self.outgoing.len();
                 self.latest_accepted_meta = Some((n, cached_idx));
-                self.outgoing.push(PaxosMessage {
+                self.outgoing.push(Message::SequencePaxos(PaxosMessage {
                     from: self.pid,
                     to: n.pid,
                     msg: PaxosMsg::Accepted(accepted),
-                });
+                }));
             }
-        };
+        }
+    }
+
+    fn get_latest_accepted_message(&mut self, n: Ballot) -> Option<&mut Accepted> {
+        if let Some((ballot, outgoing_idx)) = &self.latest_accepted_meta {
+            if *ballot == n {
+                if let Message::SequencePaxos(PaxosMessage {
+                    msg: PaxosMsg::Accepted(a),
+                    ..
+                }) = self.outgoing.get_mut(*outgoing_idx).unwrap()
+                {
+                    return Some(a);
+                } else {
+                    #[cfg(feature = "logging")]
+                    debug!(self.logger, "Cached idx is not an Accepted message!");
+                }
+            }
+        }
+        None
     }
 
     /// Also returns whether the message's ballot was promised
@@ -195,11 +206,11 @@ where
                     my_promise,
                     message_ballot
                 );
-                self.outgoing.push(PaxosMessage {
+                self.outgoing.push(Message::SequencePaxos(PaxosMessage {
                     from: self.pid,
                     to: message_ballot.pid,
                     msg: PaxosMsg::NotAccepted(not_acc),
-                });
+                }));
                 false
             }
             std::cmp::Ordering::Less => {
@@ -232,11 +243,11 @@ where
                 // Resend Promise
                 match &self.cached_promise_message {
                     Some(promise) => {
-                        self.outgoing.push(PaxosMessage {
+                        self.outgoing.push(Message::SequencePaxos(PaxosMessage {
                             from: self.pid,
                             to: promise.n.pid,
                             msg: PaxosMsg::Promise(promise.clone()),
-                        });
+                        }));
                     }
                     None => {
                         // Shouldn't be possible to be in prepare phase without having
@@ -262,11 +273,11 @@ where
             n: self.get_promise(),
         };
         for peer in &self.peers {
-            self.outgoing.push(PaxosMessage {
+            self.outgoing.push(Message::SequencePaxos(PaxosMessage {
                 from: self.pid,
                 to: *peer,
                 msg: PaxosMsg::PrepareReq(prepreq),
-            });
+            }));
         }
     }