sendAppendEntries()可以理解為針對(duì)某一個(gè) Follower執(zhí)行一次AppendEntries RPC然后把回復(fù)合并回 Leader 的 Raft 狀態(tài)。它不負(fù)責(zé)構(gòu)造請(qǐng)求。請(qǐng)求由doHeartBeat()提前構(gòu)造它只負(fù)責(zé)發(fā)送 RPC ↓ 等待結(jié)果 ↓ 檢查任期和 Leader 身份 ↓ 失敗回退 nextIndex 成功推進(jìn) matchIndex、nextIndex ↓ 嘗試推進(jìn) commitIndex源碼實(shí)現(xiàn)可見于該項(xiàng)目的raft.cpp。函數(shù)參數(shù)函數(shù)大致接收四個(gè)參數(shù)void Raft::sendAppendEntries( int server, std::shared_ptrAppendEntriesArgs args, std::shared_ptrAppendEntriesReply reply, std::shared_ptrint appendNums);含義分別是server 目標(biāo) Follower 的編號(hào)。 args doHeartBeat() 構(gòu)造好的 AppendEntries 請(qǐng)求。 包含 term、prevLogIndex、prevLogTerm、entries、leaderCommit。 reply 保存 Follower 返回的響應(yīng)。 appendNums 本輪心跳中共享的“成功節(jié)點(diǎn)數(shù)量”。 doHeartBeat() 創(chuàng)建它時(shí)通常初始化為 1代表 Leader 自己。使用shared_ptr是因?yàn)楹瘮?shù)運(yùn)行在detach()出去的線程中。doHeartBeat()返回后請(qǐng)求、回復(fù)和計(jì)數(shù)器仍必須存活。一、發(fā)送 RPC核心調(diào)用類似bool ok m_peers[server]-AppendEntries( args.get(), reply.get() );這里值得注意的是執(zhí)行網(wǎng)絡(luò) RPC 時(shí)沒有持有m_mtx。這是正確的鎖邊界設(shè)計(jì)。網(wǎng)絡(luò)請(qǐng)求可能超時(shí)或阻塞如果在 RPC 期間一直持有 Raft 主鎖當(dāng)前節(jié)點(diǎn)將無法及時(shí)處理其他節(jié)點(diǎn)發(fā)來的 RPC接收更高任期處理客戶端請(qǐng)求更新選舉和心跳狀態(tài)。因此它采用無鎖執(zhí)行網(wǎng)絡(luò)調(diào)用 ↓ RPC 返回 ↓ 加鎖處理回復(fù)但這也意味著 RPC 飛行期間當(dāng)前節(jié)點(diǎn)的任期和身份可能發(fā)生變化所以后面必須重新校驗(yàn)。二、網(wǎng)絡(luò)失敗直接返回if (!ok) { return; }ok false通常表示連接失敗、超時(shí)或者底層 RPC 沒有成功完成。這種情況下函數(shù)不會(huì)修改 nextIndex 修改 matchIndex 修改 currentTerm 增加 appendNums它也不會(huì)立即重試。后續(xù)的周期性doHeartBeat()會(huì)再次嘗試。這是合理的因?yàn)榫W(wǎng)絡(luò)失敗不代表日志不匹配不能因?yàn)橐淮纬瑫r(shí)就隨意回退nextIndex。三、重新獲得 Raft 鎖std::lock_guardstd::mutex lock(m_mtx);從這里開始回復(fù)處理期間的共享狀態(tài)修改都受同一把鎖保護(hù)包括m_currentTerm m_status m_votedFor m_nextIndex m_matchIndex m_commitIndex appendNums因此appendNums雖然只是普通int而不是原子變量但它的讀寫發(fā)生在m_mtx內(nèi)當(dāng)前實(shí)現(xiàn)下不會(huì)因?yàn)槎鄠€(gè)回復(fù)線程同時(shí)執(zhí)行而產(chǎn)生直接的數(shù)據(jù)競爭。四、處理更大的任期if (reply-term() m_currentTerm) { m_status Follower; m_currentTerm reply-term(); m_votedFor -1; return; }這是 Raft 非常重要的一條規(guī)則任何節(jié)點(diǎn)只要發(fā)現(xiàn)其他節(jié)點(diǎn)的任期比自己大就必須更新任期并退回 Follower。例如當(dāng)前節(jié)點(diǎn)認(rèn)為 自己是 term6 的 Leader Follower 回復(fù) term7這說明 term 6 已經(jīng)過期集群至少已經(jīng)進(jìn)入 term 7。當(dāng)前節(jié)點(diǎn)不能再繼續(xù)發(fā)送 term 6 的日志必須立即退位。這里同時(shí)清空m_votedFor -1;表示新任期中還沒有投票。不過這份實(shí)現(xiàn)此處分支沒有明顯調(diào)用persist()。由于currentTerm和votedFor屬于 Raft 的持久化狀態(tài)更嚴(yán)格的實(shí)現(xiàn)應(yīng)該在釋放鎖或返回之前將它們持久化否則節(jié)點(diǎn)崩潰重啟后可能恢復(fù)出舊任期。五、丟棄較小任期的回復(fù)else if (reply-term() m_currentTerm) { return; }例如發(fā)送請(qǐng)求時(shí) currentTerm 6 等待 RPC 期間 當(dāng)前節(jié)點(diǎn)進(jìn)入 term 7并且重新成為 Leader 舊請(qǐng)求返回 reply.term 6這個(gè)回復(fù)屬于過去的任期不能再修改當(dāng)前狀態(tài)所以直接丟棄。隨后通常還有斷言assert(reply-term() m_currentTerm);經(jīng)過前面兩個(gè)分支后繼續(xù)執(zhí)行的回復(fù)原則上必須與當(dāng)前任期一致。六、再次確認(rèn)自己仍然是 Leaderif (m_status ! Leader) { return; }即使回復(fù)的任期等于m_currentTerm也不能說明當(dāng)前節(jié)點(diǎn)仍然是 Leader。RPC 飛行期間可能發(fā)生Leader 發(fā)送 AppendEntries ↓ 收到合法的同任期 Leader 消息或狀態(tài)發(fā)生變化 ↓ 當(dāng)前節(jié)點(diǎn)轉(zhuǎn)為 Follower ↓ 舊 AppendEntries 回復(fù)返回此時(shí)不能再更新 Leader 專屬的nextIndex[] matchIndex[] commitIndex所以需要獨(dú)立檢查m_status。七、處理日志匹配失敗if (!reply-success()) { if (reply-updatenextindex() ! -100) { m_nextIndex[server] reply-updatenextindex(); } }success false一般說明Follower 不存在 prevLogIndex或者Follower 在 prevLogIndex 位置的 term 和 Leader 給出的 prevLogTerm 不同F(xiàn)ollower 會(huì)通過updateNextIndex告訴 Leader下次應(yīng)該從哪里嘗試。例如Leader nextIndex[F] 8 本次發(fā)送 prevLogIndex 7 Follower 實(shí)際只有日志 14Follower 可以回復(fù)success false updateNextIndex 5Leader于是執(zhí)行m_nextIndex[F] 5;下一輪請(qǐng)求變成prevLogIndex 4 entries [5, 6, 7, ...]這就是 Raft 的日志回退過程。-100是這份代碼使用的特殊哨兵值表示 Follower 沒有提供可用的新下標(biāo)。工程上更清晰的方式是使用 Protobuf 的字段存在性或明確的狀態(tài)枚舉而不是魔法數(shù)字。八、成功時(shí)更新復(fù)制進(jìn)度成功分支首先增加本輪成功數(shù)*appendNums *appendNums 1;然后計(jì)算這次請(qǐng)求能夠確認(rèn)的最大日志下標(biāo)int replicatedIndex args-prevlogindex() args-entries_size();例如prevLogIndex 5 entries [6, 7, 8] entries_size 3那么replicatedIndex 5 3 8意味著 Follower 已經(jīng)確認(rèn)擁有截至日志 8 的完整前綴。為什么更新matchIndex要用max代碼類似m_matchIndex[server] std::max( m_matchIndex[server], args-prevlogindex() args-entries_size() );原因是網(wǎng)絡(luò)回復(fù)可能亂序。假設(shè)同時(shí)存在兩個(gè)請(qǐng)求請(qǐng)求 A確認(rèn)到日志 8 請(qǐng)求 B確認(rèn)到日志 12如果 B 先回來matchIndex 12隨后舊請(qǐng)求 A 才回來。如果直接賦值就會(huì)錯(cuò)誤地變成matchIndex 8使用max可以保證matchIndex 只能前進(jìn)不能后退這是成功回復(fù)處理里做得比較穩(wěn)妥的地方。更新nextIndexm_nextIndex[server] m_matchIndex[server] 1;兩個(gè)字段的關(guān)系是matchIndex[i] 已確認(rèn) Follower i 擁有的最后日志下標(biāo) nextIndex[i] 下一次應(yīng)從哪個(gè)下標(biāo)繼續(xù)發(fā)送如果已經(jīng)確認(rèn) Follower 擁有到日志 8matchIndex 8 nextIndex 9下一次心跳如果 Leader 沒有新日志就會(huì)發(fā)送prevLogIndex 8 entries 空這就是純心跳。九、嘗試推進(jìn)commitIndex當(dāng)前實(shí)現(xiàn)使用if (*appendNums 1 m_peers.size() / 2) { *appendNums 0; if (args-entries_size() 0) { m_commitIndex std::max(m_commitIndex, m_matchIndex[server]); } }假設(shè)有 5 個(gè)節(jié)點(diǎn)多數(shù)派數(shù)量 1 5 / 2 3appendNums初始是 1因?yàn)?Leader 自己已經(jīng)有日志。收到兩個(gè) Follower 的成功回復(fù)后appendNums 3于是代碼認(rèn)為獲得多數(shù)派可以推進(jìn)提交位置。設(shè)置成0是為了避免本輪后續(xù)回復(fù)再次觸發(fā)提交邏輯。這里存在一個(gè)重要正確性問題appendNums只統(tǒng)計(jì)“RPC 成功了幾個(gè)”但不同 Follower 成功確認(rèn)的日志位置可能不同。例如 5 節(jié)點(diǎn)集群Leader擁有日志到 10 Follower A成功確認(rèn)到 5 Follower B成功確認(rèn)到 10成功數(shù)量是Leader A B 3已經(jīng)過半如果 B 的回復(fù)正好讓appendNums達(dá)到 3當(dāng)前代碼可能執(zhí)行commitIndex 10但日志 10 實(shí)際只有Leader Follower B只有兩個(gè)節(jié)點(diǎn)并沒有過半。Follower A 只擁有到日志 5。正確算法應(yīng)該針對(duì)每個(gè)候選下標(biāo)N統(tǒng)計(jì)有多少節(jié)點(diǎn)滿足 matchIndex[i] N只有滿足多數(shù)節(jié)點(diǎn)的 matchIndex N 并且 log[N].term currentTerm才能把commitIndex推進(jìn)到N。這也是 Raft 論文描述的 Leader 提交規(guī)則。(usenix.org)該項(xiàng)目其實(shí)已經(jīng)存在類似的leaderUpdateCommitIndex()回復(fù)成功后調(diào)用它會(huì)比appendNums更符合 Raft 語義。