cornerstone中raft_server_req_handlers源码解析

888974111

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更新新节点的状态，新节点收到了join_cluster_req以后更新自己的role_，leader_等状态，并调用reconfigure重置cluster的config。
  
• 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，集群成员变更一次只能变更一个节点，同时如果前一个集群变更还没完成不能开始下一个。