基于Raft分布式Kv存储：sendAppendEntries

sendAppendEntries()可以理解为针对某一个 Follower执行一次AppendEntries RPC然后把回复合并回 Leader 的 Raft 状态。它不负责构造请求。请求由doHeartBeat()提前构造它只负责发送 RPC ↓ 等待结果 ↓ 检查任期和 Leader 身份 ↓ 失败回退 nextIndex 成功推进 matchIndex、nextIndex ↓ 尝试推进 commitIndex源码实现可见于该项目的raft.cpp。函数参数函数大致接收四个参数void Raft::sendAppendEntries( int server, std::shared_ptrAppendEntriesArgs args, std::shared_ptrAppendEntriesReply reply, std::shared_ptrint appendNums);含义分别是server 目标 Follower 的编号。 args doHeartBeat() 构造好的 AppendEntries 请求。 包含 term、prevLogIndex、prevLogTerm、entries、leaderCommit。 reply 保存 Follower 返回的响应。 appendNums 本轮心跳中共享的“成功节点数量”。 doHeartBeat() 创建它时通常初始化为 1代表 Leader 自己。使用shared_ptr是因为函数运行在detach()出去的线程中。doHeartBeat()返回后请求、回复和计数器仍必须存活。一、发送 RPC核心调用类似bool ok m_peers[server]-AppendEntries( args.get(), reply.get() );这里值得注意的是执行网络 RPC 时没有持有m_mtx。这是正确的锁边界设计。网络请求可能超时或阻塞如果在 RPC 期间一直持有 Raft 主锁当前节点将无法及时处理其他节点发来的 RPC接收更高任期处理客户端请求更新选举和心跳状态。因此它采用无锁执行网络调用 ↓ RPC 返回 ↓ 加锁处理回复但这也意味着 RPC 飞行期间当前节点的任期和身份可能发生变化所以后面必须重新校验。二、网络失败直接返回if (!ok) { return; }ok false通常表示连接失败、超时或者底层 RPC 没有成功完成。这种情况下函数不会修改 nextIndex 修改 matchIndex 修改 currentTerm 增加 appendNums它也不会立即重试。后续的周期性doHeartBeat()会再次尝试。这是合理的因为网络失败不代表日志不匹配不能因为一次超时就随意回退nextIndex。三、重新获得 Raft 锁std::lock_guardstd::mutex lock(m_mtx);从这里开始回复处理期间的共享状态修改都受同一把锁保护包括m_currentTerm m_status m_votedFor m_nextIndex m_matchIndex m_commitIndex appendNums因此appendNums虽然只是普通int而不是原子变量但它的读写发生在m_mtx内当前实现下不会因为多个回复线程同时执行而产生直接的数据竞争。四、处理更大的任期if (reply-term() m_currentTerm) { m_status Follower; m_currentTerm reply-term(); m_votedFor -1; return; }这是 Raft 非常重要的一条规则任何节点只要发现其他节点的任期比自己大就必须更新任期并退回 Follower。例如当前节点认为 自己是 term6 的 Leader Follower 回复 term7这说明 term 6 已经过期集群至少已经进入 term 7。当前节点不能再继续发送 term 6 的日志必须立即退位。这里同时清空m_votedFor -1;表示新任期中还没有投票。不过这份实现此处分支没有明显调用persist()。由于currentTerm和votedFor属于 Raft 的持久化状态更严格的实现应该在释放锁或返回之前将它们持久化否则节点崩溃重启后可能恢复出旧任期。五、丢弃较小任期的回复else if (reply-term() m_currentTerm) { return; }例如发送请求时 currentTerm 6 等待 RPC 期间 当前节点进入 term 7并且重新成为 Leader 旧请求返回 reply.term 6这个回复属于过去的任期不能再修改当前状态所以直接丢弃。随后通常还有断言assert(reply-term() m_currentTerm);经过前面两个分支后继续执行的回复原则上必须与当前任期一致。六、再次确认自己仍然是 Leaderif (m_status ! Leader) { return; }即使回复的任期等于m_currentTerm也不能说明当前节点仍然是 Leader。RPC 飞行期间可能发生Leader 发送 AppendEntries ↓ 收到合法的同任期 Leader 消息或状态发生变化 ↓ 当前节点转为 Follower ↓ 旧 AppendEntries 回复返回此时不能再更新 Leader 专属的nextIndex[] matchIndex[] commitIndex所以需要独立检查m_status。七、处理日志匹配失败if (!reply-success()) { if (reply-updatenextindex() ! -100) { m_nextIndex[server] reply-updatenextindex(); } }success false一般说明Follower 不存在 prevLogIndex或者Follower 在 prevLogIndex 位置的 term 和 Leader 给出的 prevLogTerm 不同Follower 会通过updateNextIndex告诉 Leader下次应该从哪里尝试。例如Leader nextIndex[F] 8 本次发送 prevLogIndex 7 Follower 实际只有日志 14Follower 可以回复success false updateNextIndex 5Leader于是执行m_nextIndex[F] 5;下一轮请求变成prevLogIndex 4 entries [5, 6, 7, ...]这就是 Raft 的日志回退过程。-100是这份代码使用的特殊哨兵值表示 Follower 没有提供可用的新下标。工程上更清晰的方式是使用 Protobuf 的字段存在性或明确的状态枚举而不是魔法数字。八、成功时更新复制进度成功分支首先增加本轮成功数*appendNums *appendNums 1;然后计算这次请求能够确认的最大日志下标int replicatedIndex args-prevlogindex() args-entries_size();例如prevLogIndex 5 entries [6, 7, 8] entries_size 3那么replicatedIndex 5 3 8意味着 Follower 已经确认拥有截至日志 8 的完整前缀。为什么更新matchIndex要用max代码类似m_matchIndex[server] std::max( m_matchIndex[server], args-prevlogindex() args-entries_size() );原因是网络回复可能乱序。假设同时存在两个请求请求 A确认到日志 8 请求 B确认到日志 12如果 B 先回来matchIndex 12随后旧请求 A 才回来。如果直接赋值就会错误地变成matchIndex 8使用max可以保证matchIndex 只能前进不能后退这是成功回复处理里做得比较稳妥的地方。更新nextIndexm_nextIndex[server] m_matchIndex[server] 1;两个字段的关系是matchIndex[i] 已确认 Follower i 拥有的最后日志下标 nextIndex[i] 下一次应从哪个下标继续发送如果已经确认 Follower 拥有到日志 8matchIndex 8 nextIndex 9下一次心跳如果 Leader 没有新日志就会发送prevLogIndex 8 entries 空这就是纯心跳。九、尝试推进commitIndex当前实现使用if (*appendNums 1 m_peers.size() / 2) { *appendNums 0; if (args-entries_size() 0) { m_commitIndex std::max(m_commitIndex, m_matchIndex[server]); } }假设有 5 个节点多数派数量 1 5 / 2 3appendNums初始是 1因为 Leader 自己已经有日志。收到两个 Follower 的成功回复后appendNums 3于是代码认为获得多数派可以推进提交位置。设置成0是为了避免本轮后续回复再次触发提交逻辑。这里存在一个重要正确性问题appendNums只统计“RPC 成功了几个”但不同 Follower 成功确认的日志位置可能不同。例如 5 节点集群Leader拥有日志到 10 Follower A成功确认到 5 Follower B成功确认到 10成功数量是Leader A B 3已经过半如果 B 的回复正好让appendNums达到 3当前代码可能执行commitIndex 10但日志 10 实际只有Leader Follower B只有两个节点并没有过半。Follower A 只拥有到日志 5。正确算法应该针对每个候选下标N统计有多少节点满足 matchIndex[i] N只有满足多数节点的 matchIndex N 并且 log[N].term currentTerm才能把commitIndex推进到N。这也是 Raft 论文描述的 Leader 提交规则。(usenix.org)该项目其实已经存在类似的leaderUpdateCommitIndex()回复成功后调用它会比appendNums更符合 Raft 语义。