Add and use quorum_value() for Raft commits

Author: erikgrinaker

src/raft/node/leader.rs

@@ -107,11 +107,12 @@ impl RawNode<Leader> {
                 );
 
                 let from = msg.from.unwrap();
-                self.role.progress.entry(from).and_modify(|p| {
-                    p.last = last_index;
-                    p.next = last_index + 1;
-                });
-                self.maybe_commit()?;
+                let progress = self.role.progress.get_mut(&from).unwrap();
+                if last_index > progress.last {
+                    progress.last = last_index;
+                    progress.next = last_index + 1;
+                    self.maybe_commit()?;
+                }
             }
 
             // A follower rejected log entries we sent it, typically because it
@@ -225,16 +226,14 @@ impl RawNode<Leader> {
     fn maybe_commit(&mut self) -> Result<Index> {
         // Determine the new commit index, i.e. the last index replicated to a
         // quorum of peers.
-        let mut last_indexes = self
-            .role
-            .progress
-            .values()
-            .map(|p| p.last)
-            .chain(std::iter::once(self.log.get_last_index().0))
-            .collect::<Vec<_>>();
-        last_indexes.sort_unstable();
-        last_indexes.reverse();
-        let commit_index = last_indexes[self.quorum_size() as usize - 1];
+        let commit_index = self.quorum_value(
+            self.role
+                .progress
+                .values()
+                .map(|p| p.last)
+                .chain(std::iter::once(self.log.get_last_index().0))
+                .collect(),
+        );
 
         // A 0 commit index means we haven't committed anything yet.
         if commit_index == 0 {

src/raft/node/mod.rs

@@ -173,6 +173,13 @@ impl<R: Role> RawNode<R> {
         quorum_size(self.cluster_size())
     }
 
+    /// Returns the quorum value of the given unsorted slice, in descending
+    /// order. The slice must have the same size as the cluster.
+    fn quorum_value<T: Ord + Copy>(&self, values: Vec<T>) -> T {
+        assert!(values.len() == self.cluster_size() as usize, "values must match cluster size");
+        quorum_value(values)
+    }
+
     /// Sends an event
     fn send(&self, to: Address, event: Event) -> Result<()> {
         let msg = Message { term: self.term, from: Address::Node(self.id), to, event };
@@ -228,6 +235,14 @@ fn quorum_size(size: u8) -> u8 {
     size / 2 + 1
 }
 
+/// Returns the quorum (median) value of the given unsorted slice, in descending
+/// order. The slice cannot be empty.
+fn quorum_value<T: Ord + Copy>(mut values: Vec<T>) -> T {
+    assert!(!values.is_empty(), "no values provided");
+    let index = quorum_size(values.len() as u8) as usize - 1;
+    *values.select_nth_unstable_by(index, |a, b: &T| a.cmp(b).reverse()).1
+}
+
 #[cfg(test)]
 mod tests {
     pub use super::super::state::tests::TestState;
@@ -534,4 +549,13 @@ mod tests {
             assert_eq!(super::quorum_size(size), quorum);
         }
     }
+
+    #[test]
+    fn quorum_value() {
+        assert_eq!(super::quorum_value(vec![1]), 1);
+        assert_eq!(super::quorum_value(vec![1, 3, 2]), 2);
+        assert_eq!(super::quorum_value(vec![4, 1, 3, 2]), 2);
+        assert_eq!(super::quorum_value(vec![1, 1, 1, 2, 2]), 1);
+        assert_eq!(super::quorum_value(vec![1, 1, 2, 2, 2]), 2);
+    }
 }