raft: introduce group commit

harness/tests/integration_cases/test_raft.rs

@@ -4728,32 +4728,31 @@ fn test_request_snapshot_on_role_change() {
 /// Tests group commit.
 ///
 /// 1. Logs should be replicated to at least different groups before committed;
-/// 2. If no group is configured or all peers are configured to the same group,
-///     simple quorum should be used.
+/// 2. all peers are configured to the same group, simple quorum should be used.
 #[test]
 fn test_group_commit() {
     let l = default_logger();
     let mut tests = vec![
         // Single
-        (vec![1], vec![0], 1),
-        (vec![1], vec![1], 1),
+        (vec![1], vec![0], 1, 1),
+        (vec![1], vec![1], 1, 1),
         // Odd
-        (vec![2, 2, 1], vec![1, 2, 1], 2),
-        (vec![2, 2, 1], vec![1, 1, 2], 1),
-        (vec![2, 2, 1], vec![1, 0, 1], 1),
-        (vec![2, 2, 1], vec![0, 0, 0], 2),
+        (vec![2, 2, 1], vec![1, 2, 1], 2, 2),
+        (vec![2, 2, 1], vec![1, 1, 2], 1, 2),
+        (vec![2, 2, 1], vec![1, 0, 1], 1, 2),
+        (vec![2, 2, 1], vec![0, 0, 0], 1, 2),
         // Even
-        (vec![4, 2, 1, 3], vec![0, 0, 0, 0], 2),
-        (vec![4, 2, 1, 3], vec![1, 0, 0, 0], 1),
-        (vec![4, 2, 1, 3], vec![0, 1, 0, 2], 2),
-        (vec![4, 2, 1, 3], vec![0, 2, 1, 0], 1),
-        (vec![4, 2, 1, 3], vec![1, 1, 1, 1], 2),
-        (vec![4, 2, 1, 3], vec![1, 1, 2, 1], 1),
-        (vec![4, 2, 1, 3], vec![1, 2, 1, 1], 2),
-        (vec![4, 2, 1, 3], vec![4, 3, 2, 1], 2),
+        (vec![4, 2, 1, 3], vec![0, 0, 0, 0], 1, 2),
+        (vec![4, 2, 1, 3], vec![1, 0, 0, 0], 1, 2),
+        (vec![4, 2, 1, 3], vec![0, 1, 0, 2], 2, 2),
+        (vec![4, 2, 1, 3], vec![0, 2, 1, 0], 1, 2),
+        (vec![4, 2, 1, 3], vec![1, 1, 1, 1], 2, 2),
+        (vec![4, 2, 1, 3], vec![1, 1, 2, 1], 1, 2),
+        (vec![4, 2, 1, 3], vec![1, 2, 1, 1], 2, 2),
+        (vec![4, 2, 1, 3], vec![4, 3, 2, 1], 2, 2),
     ];
 
-    for (i, (matches, group_ids, w)) in tests.drain(..).enumerate() {
+    for (i, (matches, group_ids, g_w, q_w)) in tests.drain(..).enumerate() {
         let store = MemStorage::new_with_conf_state((vec![1], vec![]));
         let min_index = *matches.iter().min().unwrap();
         let max_index = *matches.iter().max().unwrap();
@@ -4775,10 +4774,28 @@ fn test_group_commit() {
                 groups.push((id, g));
             }
         }
+        sm.enable_group_commit(true);
         sm.assign_commit_groups(&groups);
-        sm.maybe_commit();
-        if sm.raft_log.committed != w {
-            panic!("#{}: committed = {}, want {}", i, sm.raft_log.committed, w);
+        if sm.raft_log.committed != 0 {
+            panic!(
+                "#{}: follower group committed {}, want 0",
+                i, sm.raft_log.committed
+            );
+        }
+        sm.state = StateRole::Leader;
+        sm.assign_commit_groups(&groups);
+        if sm.raft_log.committed != g_w {
+            panic!(
+                "#{}: leader group committed {}, want {}",
+                i, sm.raft_log.committed, g_w
+            );
+        }
+        sm.enable_group_commit(false);
+        if sm.raft_log.committed != q_w {
+            panic!(
+                "#{}: quorum committed {}, want {}",
+                i, sm.raft_log.committed, q_w
+            );
         }
     }
 }
@@ -4887,6 +4904,16 @@ fn test_group_commit_consistent() {
             }
         }
         sm.assign_commit_groups(&groups);
+        if Some(true) == exp {
+            let is_consistent = sm.check_group_commit_consistent();
+            if is_consistent != Some(false) {
+                panic!(
+                    "#{}: consistency = {:?}, want Some(false)",
+                    i, is_consistent
+                );
+            }
+        }
+        sm.enable_group_commit(true);
         let is_consistent = sm.check_group_commit_consistent();
         if is_consistent != exp {
             panic!("#{}: consistency = {:?}, want {:?}", i, is_consistent, exp);

src/progress/progress_set.rs

@@ -142,6 +142,7 @@ pub struct ProgressSet {
     // You should not depend on these values unless you just set them.
     // We use a cell to avoid taking a `&mut self`.
     sort_buffer: Vec<(u64, u64)>,
+    group_commit: bool,
     pub(crate) logger: Logger,
 }
 
@@ -160,10 +161,21 @@ impl ProgressSet {
             ),
             sort_buffer: Vec::with_capacity(voters),
             configuration: Configuration::with_capacity(voters, learners),
+            group_commit: false,
             logger,
         }
     }
 
+    /// Configures group commit.
+    pub fn enable_group_commit(&mut self, enable: bool) {
+        self.group_commit = enable;
+    }
+
+    /// Whether enable group commit.
+    pub fn group_commit(&self) -> bool {
+        self.group_commit
+    }
+
     fn clear(&mut self) {
         self.progress.clear();
         self.configuration.voters.clear();
@@ -376,18 +388,16 @@ impl ProgressSet {
     pub fn maximal_committed_index(&mut self) -> (u64, bool) {
         let matched = &mut self.sort_buffer;
         matched.clear();
-        let mut use_group_commit = false;
         let progress = &self.progress;
         self.configuration.voters().iter().for_each(|id| {
             let p = &progress[id];
-            use_group_commit |= p.commit_group_id != 0;
             matched.push((p.matched, p.commit_group_id));
         });
         // Reverse sort.
         matched.sort_by(|a, b| b.0.cmp(&a.0));
 
         let quorum = crate::majority(matched.len());
-        if !use_group_commit {
+        if !self.group_commit {
             return (matched[quorum - 1].0, false);
         }
         let (quorum_commit_index, mut checked_group_id) = matched[quorum - 1];

src/raft.rs

@@ -395,20 +395,41 @@ impl<T: Storage> Raft<T> {
         self.batch_append = batch_append;
     }
 
-    /// When committing logs, only logs replicated to at least two different
-    /// groups are committed.
+    /// Configures group commit.
+    ///
+    /// If group commit is enabled, only logs replicated to at least two
+    /// different groups are committed.
+    ///
+    /// You should use `assign_commit_groups` to configure peer groups.
+    pub fn enable_group_commit(&mut self, enable: bool) {
+        self.mut_prs().enable_group_commit(enable);
+        if StateRole::Leader == self.state && !enable && self.maybe_commit() {
+            self.bcast_append();
+        }
+    }
+
+    /// Whether enable group commit.
+    pub fn group_commit(&self) -> bool {
+        self.prs().group_commit()
+    }
+
+    /// Assigns groups to peers.
     ///
     /// The tuple is (`peer_id`, `group_id`). `group_id` should be larger than 0.
+    ///
+    /// The group information is only stored in memory. So you need to configure
+    /// it every time a raft state machine is initialized or a snapshot is applied.
     pub fn assign_commit_groups(&mut self, ids: &[(u64, u64)]) {
+        let prs = self.mut_prs();
         for (peer_id, group_id) in ids {
             assert!(*group_id > 0);
-            if let Some(pr) = self.mut_prs().get_mut(*peer_id) {
+            if let Some(pr) = prs.get_mut(*peer_id) {
                 pr.commit_group_id = *group_id;
             } else {
                 continue;
             }
         }
-        if StateRole::Leader == self.state && self.maybe_commit() {
+        if StateRole::Leader == self.state && self.group_commit() && self.maybe_commit() {
             self.bcast_append();
         }
     }
@@ -418,9 +439,6 @@ impl<T: Storage> Raft<T> {
         for (_, pr) in self.mut_prs().iter_mut() {
             pr.commit_group_id = 0;
         }
-        if StateRole::Leader == self.state && self.maybe_commit() {
-            self.bcast_append();
-        }
     }
 
     /// Checks whether the raft group is using group commit and consistent