1.概述
之前说过raft_server是cornerstone的核心，其中充满了很多req的发送，那么follower收到leader的req会怎么处理呢？
本文就是来解析cornerstone中处理req的源码。
2.process_req源码解析
ptr<resp_msg> raft_server::process_req(req_msg& req) { ptr<resp_msg> resp; l_->debug(lstrfmt("Receive a %s message from %d with LastLogIndex=%llu, LastLogTerm=%llu, EntriesLength=%d, " "CommitIndex=%llu and Term=%llu") .fmt( __msg_type_str[req.get_type()], req.get_src(), req.get_last_log_idx(), req.get_last_log_term(), req.log_entries().size(), req.get_commit_idx(), req.get_term())); { recur_lock(lock_); if (req.get_type() == msg_type::append_entries_request || req.get_type() == msg_type::vote_request || req.get_type() == msg_type::install_snapshot_request) { // we allow the server to be continue after term updated to save a round message update_term(req.get_term()); // Reset stepping down value to prevent this server goes down when leader crashes after sending a // LeaveClusterRequest if (steps_to_down_ > 0) { steps_to_down_ = 2; } } if (req.get_type() == msg_type::append_entries_request) { resp = handle_append_entries(req); } else if (req.get_type() == msg_type::vote_request) { resp = handle_vote_req(req); } else if (req.get_type() == msg_type::client_request) { resp = handle_cli_req(req); } else { // extended requests resp = handle_extended_msg(req); } } if (resp) { l_->debug(lstrfmt("Response back a %s message to %d with Accepted=%d, Term=%llu, NextIndex=%llu") .fmt( __msg_type_str[resp->get_type()], resp->get_dst(), resp->get_accepted() ? 1 : 0, resp->get_term(), resp->get_next_idx())); } return resp; } ptr<resp_msg> raft_server::handle_extended_msg(req_msg& req) { switch (req.get_type()) { case msg_type::add_server_request: return handle_add_srv_req(req); case msg_type::remove_server_request: return handle_rm_srv_req(req); case msg_type::sync_log_request: return handle_log_sync_req(req); case msg_type::join_cluster_request: return handle_join_cluster_req(req); case msg_type::leave_cluster_request: return handle_leave_cluster_req(req); case msg_type::install_snapshot_request: return handle_install_snapshot_req(req); case msg_type::prevote_request: return handle_prevote_req(req); default: l_->err( sstrfmt("receive an unknown request %s, for safety, step down.").fmt(__msg_type_str[req.get_type()])); ctx_->state_mgr_->system_exit(-1); ::exit(-1); break; } return ptr<resp_msg>(); } - 1.判断是不是append-entry，vote或者install-snapshot类型请求，如果是可以用req的term来更新自己，同时更新step_down = 2。
- 2.接着便是用switch来分流处理req。
知识点：
这里更新term的方法很巧妙，append-entry或者install-snapshot都是用于leader与follower数据间的同步。在election结束后，follower其实是不知道election是否已经结束。为了更新自己的term，follower不采用轮询election是否结束这样占用时间且低效的方式更新term，而是采用事件驱动模型，收到append-entry或install-snapshot的rpc请求后顺带更新term，高效快捷。而对于vote_request也更新自己的term则是考虑到了在网络分区的情况下某些节点可能一直处于candidate状态，在某个时间点网络又正常了，这时候通过vote_request便可以更新term到最新状态。（对于append-entry或者install-snapshot也可以帮助网络分区的节点在网络恢复后更新自己的term。）
3.handle_append_entries源码解析
ptr<resp_msg> raft_server::handle_append_entries(req_msg& req) { if (req.get_term() == state_->get_term()) { if (role_ == srv_role::candidate) { become_follower(); } else if (role_ == srv_role::leader) { l_->debug(lstrfmt("Receive AppendEntriesRequest from another leader(%d) with same term, there must be a " "bug, server exits") .fmt(req.get_src())); ctx_->state_mgr_->system_exit(-1); ::exit(-1); } else { restart_election_timer(); } } // After a snapshot the req.get_last_log_idx() may less than log_store_->next_slot() but equals to // log_store_->next_slot() -1 In this case, log is Okay if req.get_last_log_idx() == lastSnapshot.get_last_log_idx() // && req.get_last_log_term() == lastSnapshot.get_last_log_term() In not accepted case, we will return // log_store_->next_slot() for the leader to quick jump to the index that might aligned ptr<resp_msg> resp(cs_new<resp_msg>( state_->get_term(), msg_type::append_entries_response, id_, req.get_src(), log_store_->next_slot())); bool log_okay = req.get_last_log_idx() == 0 || (req.get_last_log_idx() < log_store_->next_slot() && req.get_last_log_term() == term_for_log(req.get_last_log_idx())); if (req.get_term() < state_->get_term() || !log_okay) { return resp; } // follower & log is okay if (req.log_entries().size() > 0) { // write logs to store, start from overlapped logs ulong idx = req.get_last_log_idx() + 1; size_t log_idx = 0; while (idx < log_store_->next_slot() && log_idx < req.log_entries().size()) { if (log_store_->term_at(idx) == req.log_entries().at(log_idx)->get_term()) { idx++; log_idx++; } else { break; } } // dealing with overwrites while (idx < log_store_->next_slot() && log_idx < req.log_entries().size()) { ptr<log_entry> old_entry(log_store_->entry_at(idx)); if (old_entry->get_val_type() == log_val_type::app_log) { state_machine_->rollback(idx, old_entry->get_buf(), old_entry->get_cookie()); } else if (old_entry->get_val_type() == log_val_type::conf) { l_->info(sstrfmt("revert from a prev config change to config at %llu").fmt(config_->get_log_idx())); config_changing_ = false; } ptr<log_entry> entry = req.log_entries().at(log_idx); log_store_->write_at(idx, entry); if (entry->get_val_type() == log_val_type::app_log) { state_machine_->pre_commit(idx, entry->get_buf(), entry->get_cookie()); } else if (entry->get_val_type() == log_val_type::conf) { l_->info(sstrfmt("receive a config change from leader at %llu").fmt(idx)); config_changing_ = true; } idx += 1; log_idx += 1; } // append new log entries while (log_idx < req.log_entries().size()) { ptr<log_entry> entry = req.log_entries().at(log_idx++); ulong idx_for_entry = log_store_->append(entry); if (entry->get_val_type() == log_val_type::conf) { l_->info(sstrfmt("receive a config change from leader at %llu").fmt(idx_for_entry)); config_changing_ = true; } else if (entry->get_val_type() == log_val_type::app_log) { state_machine_->pre_commit(idx_for_entry, entry->get_buf(), entry->get_cookie()); } } } leader_ = req.get_src(); commit(req.get_commit_idx()); resp->accept(req.get_last_log_idx() + req.log_entries().size() + 1); return resp; } - 1.根据不同角色判断（正常来说结束选举后的follower，在前面process_req就已经更新自己的term与leader一致了）
(1)是candidate，说明election已经结束，已经出来了leader，退回到follower。
(2)是leader，说明出bug了，终止程序。
(3)是follower，说明已经收到了来自leader的消息，通过restart_election_timer来重新定时，避免election_timeout后触发election事件。 - 2.最难理解的log_okay的判断，
bool log_okay = req.get_last_log_idx() == 0 || (req.get_last_log_idx() < log_store_->next_slot() && req.get_last_log_term() == term_for_log(req.get_last_log_idx())); (1)这里的req的last_log_idx = leader对于follower猜测的next_idx - 1，如果last_log_idx = 0，说明需要从follower的log_store的头（idx = 0）开始更新log_store。
(2)req.get_last_log_idx() < log_store_->next_slot()，说明last_log_idx在follower的log_store范围内。如果last_log_idx超过follower的log_store范围，说明猜测不准，需要更正。这时候通过resp发送follower的log_store_->next_slot()可以快速帮助leader更正next_idx。（具体可见cornerstone中msg类型解析）
- 3.接着便是更新follower的log_store，其中相同的就跳过加快速度，需要overwrite就覆盖，需要在末尾append的就先分配空间再追加。
知识点：
对于follower与leader数据的同步，如果相同可以不用更新，直接跳过加快速度。
4.handle_vote_req源码解析
ptr<resp_msg> raft_server::handle_vote_req(req_msg& req) { ptr<resp_msg> resp(cs_new<resp_msg>(state_->get_term(), msg_type::vote_response, id_, req.get_src())); bool log_okay = req.get_last_log_term() > log_store_->last_entry()->get_term() || (req.get_last_log_term() == log_store_->last_entry()->get_term() && log_store_->next_slot() - 1 <= req.get_last_log_idx()); bool grant = req.get_term() == state_->get_term() && log_okay && (state_->get_voted_for() == req.get_src() || state_->get_voted_for() == -1); if (grant) { resp->accept(log_store_->next_slot()); state_->set_voted_for(req.get_src()); ctx_->state_mgr_->save_state(*state_); } return resp; } - 1.重点是log_okay的判断
bool log_okay = req.get_last_log_term() > log_store_->last_entry()->get_term() || (req.get_last_log_term() == log_store_->last_entry()->get_term() && log_store_->next_slot() - 1 <= req.get_last_log_idx()); (1)req.get_last_log_term() > log_store_->last_entry()->get_term()，根据req最新的term与follower最新的term来比较，如果req的term更高，直接log_okay = true。
(2)如果term相同，判断req的idx是不是要比follower的更高，是的话log_okay = true。
- 2.log_okay通过后，还要进一步判断，
bool grant = req.get_term() == state_->get_term() && log_okay && (state_->get_voted_for() == req.get_src() || state_->get_voted_for() == -1); (1)判断任期是否相同。
(2)判断是否未投票，因为raft规定一个follower只能投给一个candidate，否则会导致split-vote从而可能有多个leader。
知识点：
1.根据candidate的最后一个log_entry来给follower判断是否要投票。
2.一个follower只能投给一个candidate，否则造成split-vote导致多个leader。
5.handle_cli_req源码解析
ptr<resp_msg> raft_server::handle_cli_req(req_msg& req) { bool leader = is_leader(); // check if leader has expired. // there could be a case that the leader just elected, in that case, client can // just simply retry, no safety issue here. if (role_ == srv_role::leader && !leader) { return cs_new<resp_msg>(state_->get_term(), msg_type::append_entries_response, id_, -1); } ptr<resp_msg> resp(cs_new<resp_msg>(state_->get_term(), msg_type::append_entries_response, id_, leader_)); if (!leader) { return resp; } std::vector<ptr<log_entry>>& entries = req.log_entries(); for (size_t i = 0; i < entries.size(); ++i) { // force the log's term to current term entries.at(i)->set_term(state_->get_term()); log_store_->append(entries.at(i)); state_machine_->pre_commit(log_store_->next_slot() - 1, entries.at(i)->get_buf(), entries.at(i)->get_cookie()); } // urgent commit, so that the commit will not depend on hb request_append_entries(); resp->accept(log_store_->next_slot()); return resp; } - 1.首先判断是不是leader，不是则不处理，否则继续。
- 2.是leader的话就把client要上传的entry先下载到自己的log_store里面，然后再通过request_append_entries广播给自己的follower。
知识点：
raft采用强leader机制，对于client的请求只能由leader处理，leader将req应用到自己状态机再广播给follower。
6.handle_add_srv_req源码解析
ptr<resp_msg> raft_server::handle_add_srv_req(req_msg& req) { std::vector<ptr<log_entry>>& entries(req.log_entries()); ptr<resp_msg> resp(cs_new<resp_msg>(state_->get_term(), msg_type::add_server_response, id_, leader_)); if (entries.size() != 1 || entries[0]->get_val_type() != log_val_type::cluster_server) { l_->debug("bad add server request as we are expecting one log entry with value type of ClusterServer"); return resp; } if (role_ != srv_role::leader) { l_->info("this is not a leader, cannot handle AddServerRequest"); return resp; } ptr<srv_config> srv_conf(srv_config::deserialize(entries[0]->get_buf())); { read_lock(peers_lock_); if (peers_.find(srv_conf->get_id()) != peers_.end() || id_ == srv_conf->get_id()) { l_->warn( lstrfmt("the server to be added has a duplicated id with existing server %d").fmt(srv_conf->get_id())); return resp; } } if (config_changing_) { // the previous config has not committed yet l_->info("previous config has not committed yet"); return resp; } conf_to_add_ = std::move(srv_conf); timer_task<peer&>::executor exec = [this](peer& p) { this->handle_hb_timeout(p); }; srv_to_join_ = cs_new<peer>(conf_to_add_, *ctx_, exec); invite_srv_to_join_cluster(); resp->accept(log_store_->next_slot()); return resp; } void raft_server::invite_srv_to_join_cluster() { ptr<req_msg> req(cs_new<req_msg>( state_->get_term(), msg_type::join_cluster_request, id_, srv_to_join_->get_id(), 0L, log_store_->next_slot() - 1, quick_commit_idx_)); req->log_entries().push_back(cs_new<log_entry>(state_->get_term(), config_->serialize(), log_val_type::conf)); srv_to_join_->send_req(req, ex_resp_handler_); } ptr<resp_msg> raft_server::handle_join_cluster_req(req_msg& req) { std::vector<ptr<log_entry>>& entries = req.log_entries(); ptr<resp_msg> resp(cs_new<resp_msg>(state_->get_term(), msg_type::join_cluster_response, id_, req.get_src())); if (entries.size() != 1 || entries[0]->get_val_type() != log_val_type::conf) { l_->info("receive an invalid JoinClusterRequest as the log entry value doesn't meet the requirements"); return resp; } if (catching_up_) { l_->info("this server is already in log syncing mode"); return resp; } catching_up_ = true; role_ = srv_role::follower; leader_ = req.get_src(); sm_commit_index_ = 0; quick_commit_idx_ = 0; state_->set_voted_for(-1); state_->set_term(req.get_term()); ctx_->state_mgr_->save_state(*state_); reconfigure(cluster_config::deserialize(entries[0]->get_buf())); resp->accept(log_store_->next_slot()); return resp; } - 1.首先判断是不是只添加一个srv，
if (entries.size() != 1 || entries[0]->get_val_type() != log_val_type::cluster_server)。raft为了保证election的时候只选出一个leader，强制要求每次cluster变更只能变更一个srv。具体可见Raft算法（三）：如何解决成员变更的问题？ - 2.接着判断role是不是leader，因为集群成员的变更是需要广播给各个节点的，只有leader能通过把集群的config写入自己的log_store然后调用append-entry广播给follower做到这点。
- 3.if (config_changing_)判断是不是正在修改集群成员，保证一次只更改一个srv。
- 4.调用invite_srv_to_join_cluster更新新节点的状态。
- 5.在leader收到的resp里面，leader将新的cluster_config写入自己的log_store，调用append-entry将cluster集群成员变更同步给follower。（具体可见raft_server_resp_handlers源码解析）
知识点：
为了确保只有一个leader，集群成员变更一次只能变更一个节点，同时如果前一个集群变更还没完成不能开始下一个。
7.handle_rm_srv_req源码解析
ptr<resp_msg> raft_server::handle_rm_srv_req(req_msg& req) { std::vector<ptr<log_entry>>& entries(req.log_entries()); ptr<resp_msg> resp(cs_new<resp_msg>(state_->get_term(), msg_type::remove_server_response, id_, leader_)); if (entries.size() != 1 || entries[0]->get_buf().size() != sz_int) { l_->info("bad remove server request as we are expecting one log entry with value type of int"); return resp; } if (role_ != srv_role::leader) { l_->info("this is not a leader, cannot handle RemoveServerRequest"); return resp; } if (config_changing_) { // the previous config has not committed yet l_->info("previous config has not committed yet"); return resp; } int32 srv_id = entries[0]->get_buf().get_int(); if (srv_id == id_) { l_->info("cannot request to remove leader"); return resp; } ptr<peer> p; { read_lock(peers_lock_); peer_itor pit = peers_.find(srv_id); if (pit == peers_.end()) { l_->info(sstrfmt("server %d does not exist").fmt(srv_id)); return resp; } p = pit->second; } ptr<req_msg> leave_req(cs_new<req_msg>( state_->get_term(), msg_type::leave_cluster_request, id_, srv_id, 0, log_store_->next_slot() - 1, quick_commit_idx_)); p->send_req(leave_req, ex_resp_handler_); resp->accept(log_store_->next_slot()); return resp; } 与handle_add_srv_req同理，不再赘述。
8.handle_log_sync_req源码解析
ptr<resp_msg> raft_server::handle_log_sync_req(req_msg& req) { std::vector<ptr<log_entry>>& entries = req.log_entries(); ptr<resp_msg> resp(cs_new<resp_msg>(state_->get_term(), msg_type::sync_log_response, id_, req.get_src())); if (entries.size() != 1 || entries[0]->get_val_type() != log_val_type::log_pack) { l_->info("receive an invalid LogSyncRequest as the log entry value doesn't meet the requirements"); return resp; } if (!catching_up_) { l_->info("This server is ready for cluster, ignore the request"); return resp; } log_store_->apply_pack(req.get_last_log_idx() + 1, entries[0]->get_buf()); commit(log_store_->next_slot() - 1); resp->accept(log_store_->next_slot()); return resp; } - 1.这里log_sync是特指新加入的srv的log_store同步，普通的follower与leader的同步是通过append-entry进行的。
- 2.特判是不是只有一个srv的变更，然后再判断这个新加的srv是不是在catch-up，如果catch-up = 0，说明已经更新好了，无需更新。否则更新新加的srv。
9.handle_install_snapshot_req源码解析
ptr<resp_msg> raft_server::handle_install_snapshot_req(req_msg& req) { if (req.get_term() == state_->get_term() && !catching_up_) { if (role_ == srv_role::candidate) { become_follower(); } else if (role_ == srv_role::leader) { l_->err(lstrfmt("Receive InstallSnapshotRequest from another leader(%d) with same term, there must be a " "bug, server exits") .fmt(req.get_src())); ctx_->state_mgr_->system_exit(-1); ::exit(-1); return ptr<resp_msg>(); } else { restart_election_timer(); } } ptr<resp_msg> resp(cs_new<resp_msg>(state_->get_term(), msg_type::install_snapshot_response, id_, req.get_src())); if (!catching_up_ && req.get_term() < state_->get_term()) { l_->info("received an install snapshot request which has lower term than this server, decline the request"); return resp; } std::vector<ptr<log_entry>>& entries(req.log_entries()); if (entries.size() != 1 || entries[0]->get_val_type() != log_val_type::snp_sync_req) { l_->warn("Receive an invalid InstallSnapshotRequest due to bad log entries or bad log entry value"); return resp; } ptr<snapshot_sync_req> sync_req(snapshot_sync_req::deserialize(entries[0]->get_buf())); if (sync_req->get_snapshot().get_last_log_idx() <= sm_commit_index_) { l_->warn(sstrfmt("received a snapshot (%llu) that is older than current log store") .fmt(sync_req->get_snapshot().get_last_log_idx())); return resp; } if (handle_snapshot_sync_req(*sync_req)) { resp->accept(sync_req->get_offset() + sync_req->get_data().size()); } return resp; } - 1.判断req的term与节点的term是否相同，相同则分情况（前面说过install_snapshot的rpc请求是会调用update_term，所以正常情况term是相同的）
(1)节点是candidate，说明已经出现了leader，退回到follower。
(2)节点是leader，出现了两个leader，出bug了。
(3)节点是follower，说明收到了消息，重新定时。 - 2.if (!catching_up_ && req.get_term() < state_->get_term())判断req是否合法，这里不单纯只判断req.get_term() < state_->get_term()是因为可能出现这个节点是新加的srv，term还没更新完，所以还要加一个!catching_up_的判断。
- 3.snapshot应该只有一项，所以判断 if (entries.size() != 1 || entries[0]->get_val_type() != log_val_type::snp_sync_req)。
- 4.判断snapshot的idx与节点的commit_idx大小关系，如果snapshot的小说明snapshot是旧的，不要。否则应用到自己状态机。
10.handle_prevote_req源码解析
ptr<resp_msg> raft_server::handle_prevote_req(req_msg& req) { ptr<resp_msg> resp(cs_new<resp_msg>(state_->get_term(), msg_type::prevote_response, id_, req.get_src())); bool log_okay = req.get_last_log_term() > log_store_->last_entry()->get_term() || (req.get_last_log_term() == log_store_->last_entry()->get_term() && log_store_->next_slot() - 1 <= req.get_last_log_idx()); bool grant = req.get_term() >= state_->get_term() && log_okay; if (ctx_->params_->defensive_prevote_) { // In defensive mode, server will deny the prevote when it's operating well. grant = grant && prevote_state_; } if (grant) { resp->accept(log_store_->next_slot()); } return resp; } 这里代码与handle_vote_req很相似，只是没有判断vote_for，意味着一个节点可能投多个票。
11. 总结
- 1.采用事件驱动模型更新节点的term，高效快捷。
- 2.对于follower与leader数据的同步，如果相同可以不用更新，直接跳过加快速度。
- 3.一个follower只能投给一个candidate，否则造成split-vote导致多个leader。
- 4.raft采用强leader机制，对于client的请求只能由leader处理，leader将req应用到自己状态机再广播给follower。
- 5.为了确保只有一个leader，集群成员变更一次只能变更一个节点，同时如果前一个集群变更还没完成不能开始下一个。
