cornerstone中raft_server源码解析

1.概述

cornerstone中核心即为raft_server的实现。
在raft里面有follower，leader，candidate三种角色，且角色身份还可以相互切换。
写三个类follower，leader，candidate显得没必要，因为三个类可以共享许多成员变量，如term，log_store等等。因此在cornerstone中抽象出raft_server这一个类，而raft_server的角色可以在三种状态相互切换。
下图为cornerstone中关于管理三种角色的示意图。

在本文中我们先解析单个raft_server节点中角色变化的过程，再关注leader与follower的通信。

2. raft_server节点中角色变化

2.1.1 逻辑概览

示意图如下

• 1.当election_timeout事件发生后，followerA便按照上图的顺序先向自己的peer-follower发送prevote，得到半数以上的同意后开始下一步。
• 2.followerA通过prevote知道自己网络状态良好，成为candidate，然后发送正式的request_vote请求，得到半数以上的同意后开始下一步。
• 3.followerA调用become_leader，成为leader

2.1.2 election_timeout代码解析：

void raft_server::handle_election_timeout() {     recur_lock(lock_);     if (steps_to_down_ > 0)     {         if (--steps_to_down_ == 0)         {             l_->info("no hearing further news from leader, remove this server from cluster and step down");             for (std::list<ptr<srv_config>>::iterator it = config_->get_servers().begin();                  it != config_->get_servers().end();                  ++it)             {                 if ((*it)->get_id() == id_)                 {                     config_->get_servers().erase(it);                     ctx_->state_mgr_->save_config(*config_);                     break;                 }             }              ctx_->state_mgr_->system_exit(-1);             return;         }          l_->info(sstrfmt("stepping down (cycles left: %d), skip this election timeout event").fmt(steps_to_down_));         restart_election_timer();         return;     }      if (catching_up_)     {         // this is a new server for the cluster, will not send out vote req until conf that includes this srv is         // committed         l_->info("election timeout while joining the cluster, ignore it.");         restart_election_timer();         return;     }      if (role_ == srv_role::leader)     {         l_->err("A leader should never encounter election timeout, illegal application state, stop the application");         ctx_->state_mgr_->system_exit(-1);         return;     }      if (ctx_->params_->prevote_enabled_ && role_ == srv_role::follower)     {         if (prevote_state_ && !prevote_state_->empty())         {             l_->debug("Election timeout, but there is already a prevote ongoing, ignore this event");         }         else         {             l_->debug("Election timeout, start prevoting");             request_prevote();         }     }     else     {         l_->debug("Election timeout, change to Candidate");         become_candidate();     } } 

• 1.首先steps_to_down_--，判断steps_to_down_是否减为0了，为0则继续下一步，不为0则不处理，重置election_timer。
• 2.判断是不是新加入的server在catching-up集群的log_entry及相应配置信息，是则不处理，重置election_timer，否则继续下一步。
• 3.判断进行了prevote没有，进行了就become_candidate，否则就去prevote。

知识点：
采用step_down机制，给server可能因偶然网络故障一次缓冲的机会，初始化step_down为2，先给step_down--，如果是偶然故障减为1依然还有1次机会。

2.1.3 request_prevote源码解析

void raft_server::request_prevote() {     l_->info(sstrfmt("prevote started with term %llu").fmt(state_->get_term()));     bool change_to_candidate(false);     {         read_lock(peers_lock_);         if (peers_.size() == 0)         {             change_to_candidate = true;         }     }      if (change_to_candidate)     {         l_->info("prevote done, change to candidate and start voting");         become_candidate();         return;     }      if (!prevote_state_)     {         prevote_state_ = std::make_unique<prevote_state>();     }      prevote_state_->inc_accepted_votes();     prevote_state_->add_voted_server(id_);     {         read_lock(peers_lock_);         for (peer_itor it = peers_.begin(); it != peers_.end(); ++it)         {             ptr<req_msg> req(cs_new<req_msg>(                 state_->get_term(),                 msg_type::prevote_request,                 id_,                 it->second->get_id(),                 term_for_log(log_store_->next_slot() - 1),                 log_store_->next_slot() - 1,                 quick_commit_idx_));             l_->debug(sstrfmt("send %s to server %d with term %llu")                           .fmt(__msg_type_str[req->get_type()], it->second->get_id(), state_->get_term()));             it->second->send_req(req, ex_resp_handler_);         }     } } 

• 1.特判peer的大小是否为0，为0直接跳过prevote与vote阶段，直接become_candidate，否则继续。
• 2.遍历每一个peer，向peer发送req_msg，类型为msg_type::prevote_request，req_msg里面包含自身的log_store中entry的last_idx，last_term，commit_idx情况给peer决定是否投票。

知识点：
为什么peer的大小为0就直接become_candidate而不是报持follower状态呢？

2.1.4 request_vote源码解析

void raft_server::request_vote() {     l_->info(sstrfmt("requestVote started with term %llu").fmt(state_->get_term()));     state_->set_voted_for(id_);     ctx_->state_mgr_->save_state(*state_);     votes_granted_ += 1;     voted_servers_.insert(id_);      bool change_to_leader(false);     {         read_lock(peers_lock_);          // is this the only server?         if (votes_granted_ > (int32)(peers_.size() + 1) / 2)         {             election_completed_ = true;             change_to_leader = true;         }         else         {             for (peer_itor it = peers_.begin(); it != peers_.end(); ++it)             {                 ptr<req_msg> req(cs_new<req_msg>(                     state_->get_term(),                     msg_type::vote_request,                     id_,                     it->second->get_id(),                     term_for_log(log_store_->next_slot() - 1),                     log_store_->next_slot() - 1,                     quick_commit_idx_));                 l_->debug(sstrfmt("send %s to server %d with term %llu")                               .fmt(__msg_type_str[req->get_type()], it->second->get_id(), state_->get_term()));                 it->second->send_req(req, resp_handler_);             }         }     }      if (change_to_leader)     {         become_leader();     } } 

• 整体与prevote类似，关键点在于计算是否有一半以上的节点支持的技巧：if (votes_granted_ > (int32)(peers_.size() + 1) / 2)。不管奇数还是偶数，一半以上都是⌊(x + 1) / 2⌋。

2.1.5 become_leader源码解析

void raft_server::become_leader() {     stop_election_timer();     role_ = srv_role::leader;     leader_ = id_;     srv_to_join_.reset();     ptr<snapshot> nil_snp;     {         read_lock(peers_lock_);         for (peer_itor it = peers_.begin(); it != peers_.end(); ++it)         {             it->second->set_next_log_idx(log_store_->next_slot());             it->second->set_snapshot_in_sync(nil_snp);             it->second->set_free();             enable_hb_for_peer(*(it->second));         }     }      if (config_->get_log_idx() == 0)     {         config_->set_log_idx(log_store_->next_slot());         bufptr conf_buf = config_->serialize();         ptr<log_entry> entry(cs_new<log_entry>(state_->get_term(), std::move(conf_buf), log_val_type::conf));         log_store_->append(entry);         l_->info("save initial config to log store");         config_changing_ = true;     }      if (ctx_->event_listener_)     {         ctx_->event_listener_->on_event(raft_event::become_leader);     }      request_append_entries(); } 

• 1.把election_timer给停了，同时更新自身的role等属性。
• 2.清空每一个peer原有leader的信息，同时给每个peer设置hb来宣示自己主权。
• 3.如果config_为空，更新config_

知识点：
这里的election_timeout事件其实不发生在election里面，而是在正常任期内发生的，用于触发election。follower在给定时间内没收到leader消息那么就启动vote，就是通过election_timer来实现的，如果收到了leader消息就restart_election_timer继续定时。

3.leader向follower发送消息

3.1 request_append_entries源码解析

void raft_server::request_append_entries() {     read_lock(peers_lock_);     if (peers_.size() == 0)     {         commit(log_store_->next_slot() - 1);         return;     }      for (peer_itor it = peers_.begin(); it != peers_.end(); ++it)     {         request_append_entries(*it->second);     } }  bool raft_server::request_append_entries(peer& p) {     if (p.make_busy())     {         ptr<req_msg> msg = create_append_entries_req(p);         p.send_req(msg, resp_handler_);         return true;     }      l_->debug(sstrfmt("Server %d is busy, skip the request").fmt(p.get_id()));     return false; } ptr<req_msg> raft_server::create_append_entries_req(peer& p) {     ulong cur_nxt_idx(0L);     ulong commit_idx(0L);     ulong last_log_idx(0L);     ulong term(0L);     ulong starting_idx(1L);      {         recur_lock(lock_);         starting_idx = log_store_->start_index();         cur_nxt_idx = log_store_->next_slot();         commit_idx = quick_commit_idx_;         term = state_->get_term();     }      {         std::lock_guard<std::mutex> guard(p.get_lock());         if (p.get_next_log_idx() == 0L)         {             p.set_next_log_idx(cur_nxt_idx);         }          last_log_idx = p.get_next_log_idx() - 1;     }      if (last_log_idx >= cur_nxt_idx)     {         l_->err(             sstrfmt("Peer's lastLogIndex is too large %llu v.s. %llu, server exits").fmt(last_log_idx, cur_nxt_idx));         ctx_->state_mgr_->system_exit(-1);         return ptr<req_msg>();     }      // for syncing the snapshots, for starting_idx - 1, we can check with last snapshot     if (last_log_idx > 0 && last_log_idx < starting_idx - 1)     {         return create_sync_snapshot_req(p, last_log_idx, term, commit_idx);     }      ulong last_log_term = term_for_log(last_log_idx);     ulong end_idx = std::min(cur_nxt_idx, last_log_idx + 1 + ctx_->params_->max_append_size_);     ptr<std::vector<ptr<log_entry>>> log_entries(         (last_log_idx + 1) >= cur_nxt_idx ? ptr<std::vector<ptr<log_entry>>>()                                           : log_store_->log_entries(last_log_idx + 1, end_idx));     l_->debug(         lstrfmt("An AppendEntries Request for %d with LastLogIndex=%llu, LastLogTerm=%llu, EntriesLength=%d, "                 "CommitIndex=%llu and Term=%llu")             .fmt(p.get_id(), last_log_idx, last_log_term, log_entries ? log_entries->size() : 0, commit_idx, term));     ptr<req_msg> req(cs_new<req_msg>(         term, msg_type::append_entries_request, id_, p.get_id(), last_log_term, last_log_idx, commit_idx));     std::vector<ptr<log_entry>>& v = req->log_entries();     if (log_entries)     {         v.insert(v.end(), log_entries->begin(), log_entries->end());     }      return req; } 

• 1.cornerstone无处不体现封装隔离的思想，将append-entry向所有peer的请求的实现下放到更小粒度的针对单个peer的append-entry，而即使是针对单个peer的append-entry，依然把底层的发送请求与对peer的状态管理分隔开来。
• 2.create_append_entries_req才是底层的发送请求，这里要分三种情况讨论
  (1).follower的last_log_idx >= leader的cur_nxt_idx，说明follower
  (2).last_log_idx > 0 && last_log_idx < starting_idx - 1，说明follower的log_store差太多，直接给follower安装snapshot而不是按传统发送leader的log_store。
  (3).最后一种情况说明follower与leader的log_store有重合，选出非重合的log_store发送给follower。

知识点：
follower的日志落后很多的时候，可以直接发送snapshot加快同步速度。

3.2 create_sync_snapshot_req源码解析

ptr<req_msg> raft_server::create_sync_snapshot_req(peer& p, ulong last_log_idx, ulong term, ulong commit_idx) {     std::lock_guard<std::mutex> guard(p.get_lock());     ptr<snapshot_sync_ctx> sync_ctx = p.get_snapshot_sync_ctx();     ptr<snapshot> snp;     if (sync_ctx != nilptr)     {         snp = sync_ctx->get_snapshot();     }      if (!snp || (last_snapshot_ && last_snapshot_->get_last_log_idx() > snp->get_last_log_idx()))     {         snp = last_snapshot_;         if (snp == nilptr || last_log_idx > snp->get_last_log_idx())         {             l_->err(lstrfmt("system is running into fatal errors, failed to find a snapshot for peer %d(snapshot null: "                             "%d, snapshot doesn't contais lastLogIndex: %d")                         .fmt(p.get_id(), snp == nilptr ? 1 : 0, last_log_idx > snp->get_last_log_idx() ? 1 : 0));             ctx_->state_mgr_->system_exit(-1);             return ptr<req_msg>();         }          if (snp->size() < 1L)         {             l_->err("invalid snapshot, this usually means a bug from state machine implementation, stop the system to "                     "prevent further errors");             ctx_->state_mgr_->system_exit(-1);             return ptr<req_msg>();         }          l_->info(sstrfmt("trying to sync snapshot with last index %llu to peer %d")                      .fmt(snp->get_last_log_idx(), p.get_id()));         p.set_snapshot_in_sync(snp);     }      ulong offset = p.get_snapshot_sync_ctx()->get_offset();     int32 sz_left = (int32)(snp->size() - offset);     int32 blk_sz = get_snapshot_sync_block_size();     bufptr data = buffer::alloc((size_t)(std::min(blk_sz, sz_left)));     int32 sz_rd = state_machine_->read_snapshot_data(*snp, offset, *data);     if ((size_t)sz_rd < data->size())     {         l_->err(             lstrfmt(                 "only %d bytes could be read from snapshot while %d bytes are expected, must be something wrong, exit.")                 .fmt(sz_rd, data->size()));         ctx_->state_mgr_->system_exit(-1);         return ptr<req_msg>();     }      bool done = (offset + (ulong)data->size()) >= snp->size();     std::unique_ptr<snapshot_sync_req> sync_req(new snapshot_sync_req(snp, offset, std::move(data), done));     ptr<req_msg> req(cs_new<req_msg>(         term,         msg_type::install_snapshot_request,         id_,         p.get_id(),         snp->get_last_log_term(),         snp->get_last_log_idx(),         commit_idx));     req->log_entries().push_back(cs_new<log_entry>(term, sync_req->serialize(), log_val_type::snp_sync_req));     return req; } 

• 1.首先获取旧的snapshot，判断是否能更新，能的话就更新。
• 2.把snapshot绑定到peer身上，因为snapshot挺大，需要分段发，所以要绑定到peer身上。
• 3.offset记录snapshot发送到哪里了，bool done就是记录是否发送完了snapshot。
• 4.发送snapshot_req。

知识点：
即使使用了offset记录发送的偏移，但是根据这里的代码很明显只发送了一次，那怎么能做到分段发送呢？
答案在cornerstone对于resp的处理里面，因为客户端接受snapshot，安装snapshot需要一定时间。不可能leader发送完一段snapshot紧跟着又发送下一段，leader需要等待follower处理完当前一段snapshot发送ack过来后再发送下一段，收到follower的resp后leader会再次调用这个函数，实现分段发送。

4.集群cluster的变更

4.1 cluster添加server

ptr<async_result<bool>> raft_server::add_srv(const srv_config& srv) {     bufptr buf(srv.serialize());     ptr<log_entry> log(cs_new<log_entry>(0, std::move(buf), log_val_type::cluster_server));     ptr<req_msg> req(cs_new<req_msg>((ulong)0, msg_type::add_server_request, 0, 0, (ulong)0, (ulong)0, (ulong)0));     req->log_entries().push_back(log);     return send_msg_to_leader(req); }  ptr<async_result<bool>> raft_server::send_msg_to_leader(ptr<req_msg>& req) {     typedef std::unordered_map<int32, ptr<rpc_client>>::const_iterator rpc_client_itor;     int32 leader_id = leader_;     ptr<cluster_config> cluster = config_;     bool result(false);     if (leader_id == -1)     {         return cs_new<async_result<bool>>(result);     }      if (leader_id == id_)     {         ptr<resp_msg> resp = process_req(*req);         result = resp->get_accepted();         return cs_new<async_result<bool>>(result);     }      ptr<rpc_client> rpc_cli;     {         auto_lock(rpc_clients_lock_);         rpc_client_itor itor = rpc_clients_.find(leader_id);         if (itor == rpc_clients_.end())         {             ptr<srv_config> srv_conf = config_->get_server(leader_id);             if (!srv_conf)             {                 return cs_new<async_result<bool>>(result);             }              rpc_cli = ctx_->rpc_cli_factory_->create_client(srv_conf->get_endpoint());             rpc_clients_.insert(std::make_pair(leader_id, rpc_cli));         }         else         {             rpc_cli = itor->second;         }     }      if (!rpc_cli)     {         return cs_new<async_result<bool>>(result);     }      ptr<async_result<bool>> presult(cs_new<async_result<bool>>());     rpc_handler handler = [presult](ptr<resp_msg>& resp, const ptr<rpc_exception>& err) -> void     {         bool rpc_success(false);         ptr<std::exception> perr;         if (err)         {             perr = err;         }         else         {             rpc_success = resp && resp->get_accepted();         }          presult->set_result(rpc_success, perr);     };     rpc_cli->send(req, handler);     return presult; } 

• add_srv先生成一个req，把变更的srv信息存到req附带的log里面。由于不是用于follower与leader之间的log_store同步，所以原来的last_log_idx，last_log_term，commit_idx全部为0。
• 调用send_msg_to_leader向leader发送变更srv的信息

4.2 cluster移除server

ptr<async_result<bool>> raft_server::remove_srv(const int srv_id) {     bufptr buf(buffer::alloc(sz_int));     buf->put(srv_id);     buf->pos(0);     ptr<log_entry> log(cs_new<log_entry>(0, std::move(buf), log_val_type::cluster_server));     ptr<req_msg> req(cs_new<req_msg>((ulong)0, msg_type::remove_server_request, 0, 0, (ulong)0, (ulong)0, (ulong)0));     req->log_entries().push_back(log);     return send_msg_to_leader(req); } 

• 同add_srv的分析。

5.总结

• 1.合理架构raft中各角色关系，采用一个server外加peers的组合，server内部可follower，candidate，leader相互转换。
• 2.采用step_down机制，给server可能因偶然网络故障一次缓冲的机会。
• 3.计算是否有一半以上的节点支持的技巧：if (votes_granted_ > (int32)(peers_.size() + 1) / 2)。不管奇数还是偶数，一半以上都是⌊(x + 1) / 2⌋。
• 4.follower的日志落后很多的时候，可以直接发送snapshot加快同步速度。
• 5.发送大文件采用offset机制分段传送。