Support mixing tracking peers SM mode (#616)

* Support mixing tracking peers SM mode

* Fix

Author: greensky00

src/handle_append_entries.cxx

@@ -1227,6 +1227,7 @@ void raft_server::handle_append_entries_resp(resp_msg& resp) {
             p->set_last_accepted_log_idx(new_matched_idx);
         }
 
+        bool sm_committed_idx_updated = false;
         if (resp.get_ctx() &&
             ctx_->get_params()->track_peers_sm_commit_idx_) {
             // If the response contains appendix, it should be
@@ -1239,6 +1240,7 @@ void raft_server::handle_append_entries_resp(resp_msg& resp) {
                     p_tr("sm committed index of peer %d: %" PRIu64 " -> %" PRIu64,
                          p->get_id(), prev_sm_committed_idx, new_sm_committed_idx);
                     p->set_sm_committed_idx(new_sm_committed_idx);
+                    sm_committed_idx_updated = true;
                 }
                 if (check_sm_commit_notify_ready(new_sm_committed_idx)) {
                     uint64_t target_idx =
@@ -1256,6 +1258,9 @@ void raft_server::handle_append_entries_resp(resp_msg& resp) {
                 }
             }
         }
+        if (!sm_committed_idx_updated) {
+            p->set_sm_committed_idx(0);
+        }
 
         cb_func::Param param(id_, leader_, p->get_id());
         param.ctx = &new_matched_idx;

src/handle_commit.cxx

@@ -41,12 +41,13 @@ namespace nuraft {
 
 void raft_server::commit(ulong target_idx) {
     bool track_peers_sm_commit_idx = ctx_->get_params()->track_peers_sm_commit_idx_;
+    bool same_target_idx = (target_idx == quick_commit_index_);
     if (target_idx > quick_commit_index_ ||
-        (track_peers_sm_commit_idx &&
-         target_idx >= quick_commit_index_)) {
+        (track_peers_sm_commit_idx && same_target_idx)) {
+        p_db( "trigger commit upto %" PRIu64 ", current quick commit index %" PRIu64,
+              target_idx, quick_commit_index_.load() );
         quick_commit_index_ = target_idx;
         lagging_sm_target_index_ = target_idx;
-        p_db( "trigger commit upto %" PRIu64 "", quick_commit_index_.load() );
 
         // if this is a leader notify peers to commit as well
         // for peers that are free, send the request, otherwise,
@@ -60,6 +61,11 @@ void raft_server::commit(ulong target_idx) {
                     // upto the target index. No need to send AE.
                     continue;
                 }
+                if (track_peers_sm_commit_idx &&
+                    same_target_idx &&
+                    pp->get_sm_committed_idx() == 0) {
+                    continue;
+                }
                 if (!request_append_entries(pp)) {
                     pp->set_pending_commit();
                 }
@@ -316,7 +322,8 @@ bool raft_server::commit_in_bg_exec(size_t timeout_ms) {
             }
         }
 
-        if (check_sm_commit_notify_ready(index_to_commit)) {
+        if (ctx_->get_params()->track_peers_sm_commit_idx_ &&
+            check_sm_commit_notify_ready(index_to_commit)) {
             uint64_t target_idx = update_sm_commit_notifier_target_idx(index_to_commit);
             p_tr("sm commit notify ready: %" PRIu64 ", target: %" PRIu64,
                  index_to_commit, target_idx);
@@ -510,9 +517,13 @@ void raft_server::commit_conf(ulong idx_to_commit,
 }
 
 void raft_server::scan_sm_commit_and_notify(uint64_t idx_upto) {
+    auto params = ctx_->get_params();
+    if (params->track_peers_sm_commit_idx_ == false) {
+        return;
+    }
+
     p_tr("sm commit notifier scan start, upto %" PRIu64, idx_upto);
 
-    auto params = ctx_->get_params();
     std::list< ptr<commit_ret_elem> > async_elems;
 
     std::unique_lock<std::mutex> cre_lock(commit_ret_elems_lock_);
@@ -572,8 +583,12 @@ bool raft_server::check_sm_commit_notify_ready(uint64_t idx) {
             // that means it is excluded from the quorum.
             continue;
         }
-        if (pp.second->get_sm_committed_idx() < idx) {
+        if (pp.second->get_sm_committed_idx() &&
+            pp.second->get_sm_committed_idx() < idx) {
             // At least one peer's lag is enough to return false.
+            //
+            // WARNING: We should exclude 0, in case that there are member
+            //          without tracking peer sm mode.
             return false;
         }
     }

src/raft_server.cxx

@@ -417,7 +417,8 @@ void raft_server::apply_and_log_current_params() {
           "snapshot IO: %s, "
           "parallel log appending: %s, "
           "streaming mode max log gap %d, max bytes %" PRIu64 ", "
-          "full consensus mode: %s",
+          "full consensus mode: %s, "
+          "tracking peer sm committed index: %s",
           params->election_timeout_lower_bound_,
           params->election_timeout_upper_bound_,
           params->heart_beat_interval_,
@@ -442,7 +443,9 @@ void raft_server::apply_and_log_current_params() {
           params->parallel_log_appending_ ? "ON" : "OFF",
           params->max_log_gap_in_stream_,
           params->max_bytes_in_flight_in_stream_,
-          params->use_full_consensus_among_healthy_members_ ? "ON" : "OFF" );
+          params->use_full_consensus_among_healthy_members_ ? "ON" : "OFF",
+          params->track_peers_sm_commit_idx_ ? "ON" : "OFF"
+        );
 
     status_check_timer_.set_duration_ms(params->heart_beat_interval_);
     status_check_timer_.reset();