Joint Consensus (#101)

* Enable Joint Consensus.

This commit enables Joint Consensus as described by the Raft paper
(https://raft.github.io/raft.pdf).

As of this commit it is possible to undergo arbitrary peer membership
changes in a safe way.

Notably, this gives us a resilient "Replace Node" functionality, which
is able to progress in the situation of a loss of both the old (removed)
and new (added) peers go down. This is not possible with our previous
one-at-a-time strategy.

Unfortunately, this feature is fairly large in scope. Thankfully it's
mostly testing code!

There is some new API surface, notably the
`Raft::begin_membership_change` function.

There is also some moved API surface. The `RaftLog::applied_to` function
has been moved to `Raft::commit_apply`. The old function, while it still
works, carries a depreaction warning. In the future we should make it a
`pub(crate)` function.

Finally, many tests were added to ensure Joint Consensus can work.

TODOs:
- Consider renaming `progress::Configuration` to `progress::Topology`.
- Test interaction between Joint Consensus and one-by-one method.
- Think of more cases we might need.
- Documentation update.

Signed-off-by: Hoverbear <operator@hoverbear.org>

Cargo.toml

@@ -22,6 +22,7 @@ quick-error = "1.2.2"
 rand = "0.5.4"
 hashbrown = "0.1"
 fail = { version = "0.2", optional = true }
+getset = "0.0.6"
 
 [dev-dependencies]
 env_logger = "0.5"

benches/benches.rs

@@ -1,5 +1,5 @@
 #![allow(dead_code)] // Due to criterion we need this to avoid warnings.
-#![cfg_attr(feature = "cargo-clippy", allow(let_and_return))] // Benches often artificially return values. Allow it.
+#![cfg_attr(feature = "cargo-clippy", allow(clippy::let_and_return))] // Benches often artificially return values. Allow it.
 
 extern crate criterion;
 extern crate env_logger;

benches/suites/progress_set.rs

@@ -11,7 +11,8 @@ pub fn bench_progress_set(c: &mut Criterion) {
     bench_progress_set_remove(c);
     bench_progress_set_iter(c);
     bench_progress_set_get(c);
-    bench_progress_set_nodes(c);
+    bench_progress_set_voters(c);
+    bench_progress_set_learners(c);
 }
 
 fn quick_progress_set(voters: usize, learners: usize) -> ProgressSet {
@@ -146,14 +147,40 @@ pub fn bench_progress_set_iter(c: &mut Criterion) {
     });
 }
 
-pub fn bench_progress_set_nodes(c: &mut Criterion) {
+pub fn bench_progress_set_voters(c: &mut Criterion) {
     let bench = |voters, learners| {
         move |b: &mut Bencher| {
             let set = quick_progress_set(voters, learners);
             b.iter(|| {
                 let set = set.clone();
-                let agg = set.iter().all(|_| true);
-                agg
+                let sum = set.voters().fold(0, |mut sum, _| {
+                    sum += 1;
+                    sum
+                });
+                sum
+            });
+        }
+    };
+
+    DEFAULT_RAFT_SETS.iter().for_each(|(voters, learners)| {
+        c.bench_function(
+            &format!("ProgressSet::nodes ({}, {})", voters, learners),
+            bench(*voters, *learners),
+        );
+    });
+}
+
+pub fn bench_progress_set_learners(c: &mut Criterion) {
+    let bench = |voters, learners| {
+        move |b: &mut Bencher| {
+            let set = quick_progress_set(voters, learners);
+            b.iter(|| {
+                let set = set.clone();
+                let sum = set.voters().fold(0, |mut sum, _| {
+                    sum += 1;
+                    sum
+                });
+                sum
             });
         }
     };

benches/suites/raft.rs

@@ -13,10 +13,10 @@ fn quick_raft(voters: usize, learners: usize) -> Raft<MemStorage> {
     let config = Config::new(id);
     let mut raft = Raft::new(&config, storage).unwrap();
     (0..voters).for_each(|id| {
-        raft.add_node(id as u64);
+        raft.add_node(id as u64).unwrap();
     });
     (voters..learners).for_each(|id| {
-        raft.add_learner(id as u64);
+        raft.add_learner(id as u64).unwrap();
     });
     raft
 }

benches/suites/raw_node.rs

@@ -10,8 +10,7 @@ fn quick_raw_node() -> RawNode<MemStorage> {
     let peers = vec![];
     let storage = MemStorage::default();
     let config = Config::new(id);
-    let node = RawNode::new(&config, storage, peers).unwrap();
-    node
+    RawNode::new(&config, storage, peers).unwrap()
 }
 
 pub fn bench_raw_node_new(c: &mut Criterion) {

proto/eraftpb.proto

@@ -30,6 +30,8 @@ message Entry {
 
 message SnapshotMetadata {
     ConfState conf_state = 1;
+    ConfState pending_membership_change = 4;
+    uint64 pending_membership_change_index = 5;
     uint64 index = 2;
     uint64 term = 3;
 }
@@ -91,11 +93,20 @@ enum ConfChangeType {
     AddNode    = 0;
     RemoveNode = 1;
     AddLearnerNode = 2;
+    BeginMembershipChange = 3;
+    FinalizeMembershipChange = 4;
 }
 
 message ConfChange {
     uint64 id = 1;
     ConfChangeType change_type = 2;
+    // Used in `AddNode`, `RemoveNode`, and `AddLearnerNode`.
     uint64 node_id = 3;
     bytes context = 4;
+    // Used in `BeginMembershipChange` and `FinalizeMembershipChange`.
+    ConfState configuration = 5;
+    // Used in `BeginMembershipChange` and `FinalizeMembershipChange`.
+    // Because `RawNode::apply_conf_change` takes a `ConfChange` instead of an `Entry` we must
+    // include this index so it can be known.
+    uint64 start_index = 6;
 }