cornerstone中raft_server_req_handlers源码解析
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，集群成员变更一次只能变更一个节点，同时如果前一个集群变更还没完成不能开始下一个。