这可能是全网最透彻的raftelection+hearbeat（mit6.824parta）代码解析（1）

前言&＃xff1a;
虽然mit6.824这门课说不让大家把代码公布出来&＃xff0c;因为它们觉得自己写学习到的才是最多的。我不否认它们这个观点&＃xff0c;但当一个go实践并不多的学生开始写的时候&＃xff0c;它会发现毫无思路。因此&＃xff0c;对于一些想学raft并实践的学生&＃xff08;比如说我&＃xff09;&＃xff0c;是非常需要一份详细到不能再详细的代码解析的。然而&＃xff0c;现状是网上的raft 各个part解析都是只讲重点而不注重细节&＃xff0c;因此我认为很有必要把细节缕清楚&＃xff01;

参考资料&＃xff1a;
我这里的代码是参考thu大神的&＃xff0c;但这位小哥哥也没有详细地解释清楚&＃xff0c;我就借花献佛&＃xff0c;将它的代码逻辑讲明白
在实现之前&＃xff0c;可以先参考我之前写的一些博客&＃xff0c;包括raft动图的介绍以及论文翻译&＃xff1a;
详见我的论文学习模块
mit 6.824 lab2 raft

内容概括&＃xff1a;
本文主要从以下几个方面展开对parta实验的源码解析&＃xff1a;
1.raft结构体
2.make初始化
2.5 changestate&＃xff08;&＃xff09;函数的实现
3.requestvoterpc的实现
4.appendentries的实现&＃xff08;仅heartbeat&＃xff09;
5.ticker的逻辑
6.测试样例的讲解与测试结果的解析&＃xff08;正常选取&＃xff0c;再选举&＃xff0c;多节点宕机然后再进入&＃xff09;

preparation
一些常数的设定&＃xff1a;

// 常量的设置&＃xff0c;包括状态对应的int&＃xff0c;以及timeout时间
const (FOLLOWER &＃61; 0CANDIDATE &＃61; 1LEADER &＃61; 2TO_FOLLOWER &＃61; 0TO_CANDIDATE &＃61; 1TO_LEADER &＃61; 2// 设置选举时间的随机区间ELECTION_TIMEOUT_MAX &＃61; 600ELECTION_TIMEOUT_MIN &＃61; 400// 设置心跳的timeoutHEARTBEAT_TIMEOUT &＃61; 100// 设置apply的timeout// APPLIED_TIMEOUT &＃61; 28
)


设置follower、candidate、leader三个状态对应的int&＃xff0c;以及心跳的timeout为100ms&＃xff08;raft实验要求1s不超过十次心跳&＃xff09;&＃xff0c;这样我们就要适当增加论文中150-300ms的electiontimeout的限度了&＃xff0c;这里设置为400-600ms

raft结构体
先看看论文中的结构体描述&＃xff1a;

type Raft struct {mu sync.Mutex // Lock to protect shared access to this peer&＃39;s statepeers []*labrpc.ClientEnd // RPC end points of all peerspersister *Persister // Object to hold this peer&＃39;s persisted stateme int // this peer&＃39;s index into peers[]dead int32 // set by Kill()// Your data here (2A, 2B, 2C).// Look at the paper&＃39;s Figure 2 for a description of what// state a Raft server must maintain.// 对照state表currentTerm int //当前看到的最新termvotedFor int //给哪个candidate投票log []Entry //log entriesgetVoteNum int //获得多少的票commitIndex int //最新的committed的log entrylastApplied int //最新的applied给state machine的log entrystate int // 当前的角色lastResetElectionTime time.Time // 上次重新选举的时间nextIndex []int // leader记录的发给每个server的next log entry的indexmatchIndex []int // leader记录的每个server复制的最高indexapplyCh chan ApplyMsg // 结构体使用chan&＃xff0c;由leader返回给tester&＃xff1f;//SnapShot 用到的lastSSPointIndex intlastSSPointTerm int}


详细的描述可以看注释&＃xff0c;在parta中我们不会用到后面的这几个index和term相关的属性&＃xff0c;都是后面几个part用到的。persister也不用管&＃xff0c;也是后面的partd才用到&＃xff08;用来存快照的&＃xff09;
我们着重看一下这个peers
peers数组我们只需要关注它们的index&＃xff0c;其中该raft节点本身的index就是me&＃xff1b;然后&＃xff0c;每个raft节点中存储的peers的index所指向的实际raft节点都是一致的&＃xff08;他这里没有明显指出来&＃xff0c;但我们需要知道这一点才好理解&＃xff09;
然后&＃xff0c;相比论文中的state表&＃xff0c;我们新增了state变量表示当前该raft节点的实际状态&＃xff0c;以及它上一次重置时间的时间戳lastResetElectionTime&＃xff08;用来判断是否超时&＃xff09;

make初始化
我们在raft.go中需要提供make接口来方面tester调用make_config构建raft集群&＃xff0c;主要的作用就是初始化&＃xff1a;

func Make(peers []*labrpc.ClientEnd, me int,persister *Persister, applyCh chan ApplyMsg) *Raft {rf :&＃61; &Raft{}// peers对应的index就是每个server唯一的idrf.peers &＃61; peersrf.persister &＃61; persisterrf.me &＃61; me// Your initialization code here (2A, 2B, 2C).// 还是要加锁rf.mu.Lock()rf.state &＃61; FOLLOWERrf.currentTerm &＃61; 0rf.getVoteNum &＃61; 0rf.votedFor &＃61; -1rf.lastApplied &＃61; 0rf.commitIndex &＃61; 0rf.log &＃61; []Entry{}rf.log &＃61; append(rf.log, Entry{}) // 能不能删掉&＃xff1f;rf.lastSSPointIndex &＃61; 0rf.lastSSPointTerm &＃61; 0rf.applyCh &＃61; applyChrf.mu.Unlock()// initialize from state persisted before a crashrf.readPersist(persister.ReadRaftState())// start ticker goroutine to start elections// go rf.ticker()// 定时选举candidatego rf.candidateElectionTicker()// leader定时发心跳&＃xff0c;和log entriesgo rf.leaderAppendEntriesTicker()// 定时将committed变成applied//go rf.commitedToAppliedTicker()return rf}


很显然&＃xff0c;前面就是各种状态的初始化。然后最后用go关键字拉起了两个协程&＃xff0c;分别是用来控制进行定期的election选举的检查以及进行定期的leader发送心跳提醒。

changestate&＃xff08;&＃xff09;函数的实现
在介绍changestate函数之前&＃xff0c;我们先归纳一下在parta中所有状态变换的可能以及它们是否需要重置时间

1. server x收到reqeustvote rpc&＃xff0c; 且rpc中的term比当前的currentTerm要大&＃xff0c;也就是说当前server x手上的信息不是最新的&＃xff0c;那么server x -> follower(server x 之前可能是leader或者candidate哦&＃xff09;&＃xff0c;然后是不重置时间的&＃xff08;因为可能它就是一个follower&＃xff09;
2. candidate自己的getvoteNum过半&＃xff0c;选举成功&＃xff0c;candidate -> leader&＃xff0c; 重置时间
3. candidate发现传rpc的term比reply的小&＃xff0c; candidate -> follower, 不重置时间
4. follower -> candidate 重置时间
5. server x 给 candidate 投票后&＃xff0c;x重置时间
6. server x 收到leader心跳&＃xff0c;重置时间
7. leader看到更大的term&＃xff0c;降级为follower&＃xff0c; 重置时间

接下来介绍一下changestate函数

// 更换raft节点的状态&＃xff0c;并初始化一些东西
func (rf *Raft) changeState(howtochange int, resetTime bool) {// 如果要变成followerif howtochange &＃61;&＃61; TO_FOLLOWER {rf.state &＃61; FOLLOWERrf.votedFor &＃61; -1rf.getVoteNum &＃61; 0// 如果要重置TIMEOUT起始时间if resetTime {rf.lastResetElectionTime &＃61; time.Now()}}// 如果要变成candidateif howtochange &＃61;&＃61; TO_CANDIDATE {rf.state &＃61; CANDIDATE// 自己给自己投票rf.votedFor &＃61; rf.merf.getVoteNum &＃61; 1// term更新rf.currentTerm &＃43;&＃61; 1// 变成candidate一定会resetrf.lastResetElectionTime &＃61; time.Now()// 加入选举过程rf.candidateJoinElection()}// 如果要变成leaderif howtochange &＃61;&＃61; TO_LEADER {rf.state &＃61; LEADERrf.votedFor &＃61; -1rf.getVoteNum &＃61; 0// 后续用到的nextIndex和matchIndexrf.nextIndex &＃61; make([]int, len(rf.peers))rf.matchIndex &＃61; make([]int, len(rf.peers))//重置时间rf.lastResetElectionTime &＃61; time.Now()}
}


这个函数两个参数&＃xff0c;一个是需要变成的state&＃xff0c;一个是是否需要重置时间。我们可以发现只有follower才可以选择是否需要变更时间&＃xff0c;其他的candidate和leader都是变了之后时间默认重置的&＃xff0c;然后这主要涉及到状态的再初始化&＃xff0c;这就不需要赘述了。然后就是candidate中涉及到的一个candidateJoinElection函数&＃xff0c;因为candidate只可能是follower变得&＃xff0c;因此变成candidate的同时就可以发起requestvoterpc了

candidatejoinelection函数&＃xff1a;

这里面涉及到rpc的处理我们后面再一起讲

模拟rpc
我们先看看两个调用rpc接口的函数哈&＃xff08;requestvoterpc和appendentriesrpc&＃xff09;

func (rf *Raft) sendRequestVote(server int, args *RequestVoteArgs, reply *RequestVoteReply) bool {ok :&＃61; rf.peers[server].Call("Raft.RequestVote", args, reply)return ok
}func (rf *Raft) sendAppendEntries(server int, args *AppendEntriesArgs, reply *AppendEntriesReply) bool {ok :&＃61; rf.peers[server].Call("Raft.AppendEntries", args, reply)return ok
}


这里面都把调用rpc定义成raft绑定的方法了&＃xff0c;这里是rf发rpc&＃xff0c;然后peers[server]收到rpc&＃xff0c;然后args是rf发出去的参数&＃xff0c;通过我们编写的rpc接口处理器处理后&＃xff0c;也就是call的第一个参数&＃xff0c;server就会根据具体情况返回对应的reply结果&＃xff0c;然后ok如果为true就说明网络是ok的&＃xff0c;完成了一次有效的通信。

因此&＃xff0c;我们为了实现rpc&＃xff0c;要定义输入输出参数的格式&＃xff0c;以及定义rpc处理函数&＃xff1b;然后通过ticker来调用一个协程负责向其他节点发送rpc&＃xff08;这里又要拉起新的协程&＃xff09;&＃xff0c;然后再完成后续的判断

未完待续。。。