「实验记录」MIT 6.824 Raft Lab2C Persist

I. Source

1. MIT-6.824 2020 课程官网
2. Lab2: Raft 实验主页
3. simviso 精品付费翻译 MIT 6.824 课程
4. Paper - Raft extended version

II. My Code

1. source code 的 Gitee 地址
2. Lab2C: Persist 的 Gitee 地址

课程官网提供的 Lab 代码下载地址，我没有访问成功，于是我从 Github 其他用户那里 clone 到干净的源码，有需要可以访问我的 Gitee 获取

III. Motivation

提出 Raft 的主要目的，是为了解决容错问题，即使集群中有一些机器发生了故障，也不影响整体的运作（对外提供的服务）

我用一个 demo 来说明，假设我们的需求一直都是自己的 PC 能够顺利访问云端的资源（HTTP 或数据库）服务器。在服务器稳定在线的情况下，我们去访问它，一点问题都没有

但是，如果那唯一的一台服务器掉线了，那么我们将无法再访问，即对外的服务到此停止。这是我们无法忍受的，我们希望提供服务的一方能够保持稳定，时时刻刻为我提供访问服务。这就是我们的需求

好，现在问题摆在眼前，提供服务的一方怎样保证稳定性？让唯一的那台服务器永远维持稳定的状态，不允许宕机？这非常地不现实，就好比让一个人练成金刚不坏之身

所以，我们只能琢磨是否可以通过添加服务器的数量来确保对外服务的稳定。更近一步，即是现在服务器不再只有一台，扩充到 3 台，这 3 台中有一台是 primary 服务器，也主要由它对外提供服务；其他 2 台是 secondary 服务器（后备力量），拥有和 primary 服务器相同的数据内容

在 primary 服务器出现故障的时候，secondary 服务器顶上去，替代它的位置。这样就可以保持稳定的对外服务了

这就是我们应对资源服务器崩溃的最常用最有效的法子，但是想实现这个想法，首先要解决数据同步的问题，即如何确保 secondary 服务器拥有和 primary 服务器同样的内容？

这个同步问题，在学术上被称为共识算法，最经典的共识算法是 Paxos，但是它太难理解了。于是，斯坦福那帮人想出了更为简便的共识算法，即 Raft

通过 Raft 算法就可以同步集群中服务器的内容。要实现该算法，分三步走，5 - The Raft consensus algorithm 章节中的 Leader Election、Log Replication 和 Safety

本文主要针对第三步，Lab2C: persist 展开讲解，如有 Lab2A: Leader Election 或 Lab2B: Log Replication 的需要，请移步

IV. Solution

Lab2C: persist 主要就是为了解决一个问题，即网络中的复杂情况会导致集群中的机器掉线 OR 机器本身的宕机。我们希望发生此类的情况，Raft 也能很好地应对

论文中的图 2 也提到了要完成持久化操作，我们需要保存哪些字段放在磁盘中，最重要的莫过于 log[] 、curTerm 和 votedFor 三个字段。其他的例如 nextIdxs[] 和 matchIdxs[] 是可以通过这三个字段即时计算出来的，从工程角度上来讲可以不用保存，这样减少了读写磁盘的时间，从一定程度上提高了 Lab2C: persist 的效率

好，话不多说，我们可以直接开始具体的编码工作了，只要在 Lab2B: Log Replication 确保没有问题的情况下，Lab2C: persist 将会容易很多

S1 - 实现persist()

第一步，就是要实现持久化函数，即 raft.go:persist() ，这个函数干的事情，即是将 log[] 、curTerm 和 votedFor 三个字段写入磁盘，具体如何写入，这不是我们操心的事，我们只需要将其序列化交给 Encoder 即可，

func (rf *Raft) persist() {// Your code here (2C).// Example:// w := new(bytes.Buffer)// e := labgob.NewEncoder(w)// e.Encode(rf.xxx)// e.Encode(rf.yyy)// data := w.Bytes()// rf.persister.SaveRaftState(data)w := new(bytes.Buffer)e := gob.NewEncoder(w)e.Encode(rf.curTerm)e.Encode(rf.votedFor)e.Encode(rf.log)data := w.Bytes()rf.persister.SaveRaftState(data)
}

就像这样，按照上面助教已给出的例子，照猫画虎即可

S2 - 实现readPersist()

我们也要实现读的具体操作，同样如何去读取磁盘也不是我们关心的事，我们只需要调用 Decoder 分解序列化即可，

func (rf *Raft) readPersist(data []byte) {if data == nil || len(data) < 1 { // bootstrap without any state?return}// Your code here (2C).// Example:// r := bytes.NewBuffer(data)// d := labgob.NewDecoder(r)// var xxx// var yyy// if d.Decode(&xxx) != nil ||//    d.Decode(&yyy) != nil {//   error...// } else {//   rf.xxx = xxx//   rf.yyy = yyy// }r := bytes.NewBuffer(data)d := gob.NewDecoder(r)var curTerm intvar votedFor intvar log []LogEntryif d.Decode(&curTerm) != nil || d.Decode(&votedFor) != nil || d.Decode(&log) != nil {DPrintf("read persist fail\n")} else {rf.curTerm = curTermrf.votedFor = votedForrf.log = log}
}

又是一个照猫画虎，跟着助教的写法来就可以。每一次的 raft.go:Make() 都会调用 readPersist() 读取已持久化的数据用以初始化，所以它不需要我们操心 OR 考虑应该在何处调用，自带的框架已经帮我们安排好了

S3 - 持久化三字段

我们要在 raft.go 、appendEntries.go 和 requestVote.go 中寻找到流程中更新 log[] 、curTerm 和 votedFor 三个字段的地方，然后在它们完成操作之后持久化此类的数据

在 raft.go 中第一处出现在 raft.go:Start() 中，即 client 追加日志条目之后，要持久化，

func (rf *Raft) Start(command interface{}) (int, int, bool) {// Your code here (2B).rf.mu.Lock()defer rf.mu.Unlock()/*------------Lab2C Persist---------------*/defer rf.persist()index := -1term := rf.curTermisLeader := rf.role == Leaderif isLeader {rf.log = append(rf.log, LogEntry{Idx: rf.lastLogIdx() + 1, Term: term, Cmd: command})index = rf.lastLogIdx()}return index, term, isLeader
}

第二处出现在 raft.go:run() 的 candidate 阶段，因为 follower 成为 candidate 之后会更新 curTerm 和 votedFor ，

func (rf *Raft) run() {for !rf.killed() {switch rf.role {case Follower:...case Candidate:rf.mu.Lock()rf.curTerm++rf.votedFor = rf.merf.voteCount = 1rf.persist()rf.mu.Unlock()...}time.Sleep(10 * time.Millisecond)}
}

之后，就是 requestVote.go 中，需要在 RequestVote() 中添加持久化操作，

func (rf *Raft) RequestVote(args *RequestVoteArgs, reply *RequestVoteReply) {// Your code here (2A, 2B).rf.mu.Lock()defer rf.mu.Unlock()/*------------Lab2C Persist---------------*/defer rf.persist()...
}

就像上述一样，调用 defer rf.persist() 即可，这样就可以在离开函数之前将数据写回磁盘，以及在 sendRequestVote() 中，

func (rf *Raft) sendRequestVote(server int, args *RequestVoteArgs, reply *RequestVoteReply) bool {ok := rf.peers[server].Call("Raft.RequestVote", args, reply)rf.mu.Lock()defer rf.mu.Unlock()if !ok {return ok}term := rf.curTerm/* 自身过期的情况下，直接不再唱票 */if rf.role != Candidate || args.Term != term {return ok}/* 碰到一个任期比自己高的人 */if reply.Term > term {rf.curTerm = reply.Termrf.role = Follower /* candidate 主动回滚至 follower */rf.votedFor = NoBodyrf.persist()return ok}if reply.VoteGranted {rf.voteCount++if rf.role == Candidate && rf.voteCount > len(rf.peers)/2 {rf.role = Leader /* 至关重要 */rf.leaderCh <- struct{}{}}}return ok
}

在碰到一个任期比自己高的人之后，candidate 会更新自己的 curTerm 和 votedFor ，这里也需要注意持久化

最后来到 appendEntries.go 中，同样在 AppendEntries() 中添上 defer rf.persist() 即可，

func (rf *Raft) AppendEntries(args *AppendEntriesArgs, reply *AppendEntriesReply) {rf.mu.Lock()defer rf.mu.Unlock()/*------------Lab2C Persist---------------*/defer rf.persist()reply.Success = falsereply.Term = rf.curTerm...
}

在 sendAppendEntries() 中碰见更新任期的情况下，也要持久化，

func (rf *Raft) sendAppendEntries(server int, args *AppendEntriesArgs, reply *AppendEntriesReply) bool {ok := rf.peers[server].Call("Raft.AppendEntries", args, reply)rf.mu.Lock()defer rf.mu.Unlock()if !ok {return ok}term := rf.curTerm/* 自身过期的情况下，不需要在维护 nextIdx 了 */if rf.role != Leader || args.Term != term {return ok}/* 仅仅是被动退位，不涉及到需要投票给谁 */if reply.Term > term {rf.curTerm = reply.Termrf.role = Follower /* 主动回滚至 follower */rf.votedFor = NoBodyrf.persist()return ok}/*------------Lab2B Log Replication----------------*/if reply.Success {...} else {...}return ok
}

至此，已经功成大半，但测试的准确率还不是 100%，是因为有一些问题还需要考虑到

S4 - 在newRaft()中初始化nextIdxs和matchIdxs

记得要在 raft.go:Make() 的子函数 newRaft() 中写上定义 nextIdxs[] 和 matchIdxs[] 的代码，不然会在 boatcastAE() 时出现 index out of range 切片越界的问题，我想不懂为什么会这样，因为即使机器掉线了，它重新上线之后，也需要经过选举才能成为 leader，而成为 leader 的第一件事就是初始化 nextIdxs[] 和 matchIdxs[]

按理说，这两个切片不可能在发送心跳包时为空，具体什么 bug，我称之为玄学，看我代码吧，

func newRaft(peers []*labrpc.ClientEnd, me int, persister *Persister, applyCh chan ApplyMsg) *Raft {rf := &Raft{peers:       peers,persister:   persister,me:          me,role:        Follower,voteCount:   0,curTerm:     0,votedFor:    NoBody,grantVoteCh: make(chan struct{}, ChanCap),leaderCh:    make(chan struct{}, ChanCap),heartBeatCh: make(chan struct{}, ChanCap),commitCh:    make(chan struct{}, ChanCap),nextIdxs:    make([]int, len(peers)),matchIdxs:   make([]int, len(peers)),applyCh:     applyCh,commitIdx:   0,appliedIdx:  0,}/* 下标从 1 开始，这非常重要，0 号位置存放默认题目 */rf.log = append(rf.log, LogEntry{Idx: 0, Term: 0})for i := range rf.peers {rf.nextIdxs[i] = rf.lastLogIdx() + 1rf.matchIdxs[i] = 0}return rf
}

这样就不会再出现 index out of range 的情况了

S5 - 适当缩短心跳时间

上面的几种手段已经能够解决大部分错误了，测试基本能够到达 200 次只出现一个失败的情况，即 one(%v) fail to agreement

这错误提示的意思就是集群不能在有效的时间内选出 leader，进而完成日志同步的工作

很明显，就是选主不够快，最简单的办法，即是缩短心跳时间，虽然在 Lab2: Raft 实验主页 中建议我们心跳 1 s 中不超过十次，但是按照 100 ms 的频率，从工程角度来看是不行的，我改成了 90 ms 就没有此类的错误了，

ElectionTimeOut  = 250 * time.Millisecond /* 要远大于论文中的 150-300 ms 才有意义，当然也要保证在 5 秒之内完成测试 */
HeartBeatTimeOut = 90 * time.Millisecond  /* 心跳 1 秒不超过 10 次 *//* 生成随机超时时间，在 250ms~500 ms 范围之内 */
func randElectionTimeOut() time.Duration {r := rand.New(rand.NewSource(time.Now().UnixNano()))t := time.Duration(r.Int63()) % ElectionTimeOutreturn ElectionTimeOut + t
}/* 生成固定的心跳时间，固定值为 90 ms */
func fixedHeartBeatTimeOut() time.Duration {return HeartBeatTimeOut
}

所以，我还是称之为玄学。模型正确，不一定代表实现正确！

至此，已然讲明白了 Lab2C: persist 整个一套流程

V. Result

golang 比较麻烦，它有 GOPATH 模式，也有 GOMODULE 模式，6.824-golabs-2020 采用的是 GOPATH，所以在运行之前，需要将 golang 默认的 GOMODULE 关掉，

$ export GO111MODULE="off"

随后，就可以进入 src/raft 中开始运行测试程序，

$ go test -run 2C

仅此一次的测试远远不够，可以通过 shell 循环，让测试跑个两百次就差不多了

$ for i in {1..200}; go test -run 2C   

这样，如果还没错误，那应该是真的通过了。分布式的很多 bug 需要通过反复模拟才能复现出来的，它不像单线程程序那样，永远是幂等的情况。也可以用我写的脚本 test_2c.py，

import osntests = 200
nfails = 0
noks = 0if __name__ == "__main__":for i in range(ntests):print("*************ROUND " + str(i+1) + "/" + str(ntests) + "*************")filename = "out" + str(i+1)os.system("go test -run 2C | tee " + filename)with open(filename) as f:if 'FAIL' in f.read():nfails += 1print("✖️fails, " + str(nfails) + "/" + str(ntests))continueelse:noks += 1print("✔️ok, " + str(noks) + "/" + str(ntests))os.system("rm " + filename)

我已经跑过两百次，无一 FAIL。之后的 Lab3: Fault-tolerant Key/Value Service 和 Lab4: Sharded Key/Value Service 都是基于 Lab2: Raft 的，要确保你实现的 Raft 算法没有 bug，不然 Labs 越做到后面越难受