MIT 6.824 Lab2 2021 ABCD完成记录

这里写自定义目录标题

• 说明
• 总体方法
• • paper 与 实验材料阅读
  • Debug & log
  • Time.Timer
  • Log Index
  • 代码逻辑概述
  • 测试时间
• A leader eleciton
• • Raft 基础架构
  • LeaderElection
  • AppendEntries
• B log
• • 补充A
  • ApplyEntries
• C persistence
• D log compaction
• 完整代码（lab1,2,3,4）

说明

• 本人除了Lab1，并没有接触过GO语言。文中有写的不对，不恰当的地方，希望大佬们能及时指出。整个实验花了前后整整4，5天的时间，包括上课时间，大概有40个小时。这里主要是记录一些完成细节、思考逻辑、bug调试心得。

总体方法 paper 与 实验材料阅读

• 论文需要多遍详读。figure2&figure13 中以伪代码的形式给出了Raft的基础实现，最后你的代码的逻辑和这张图几乎就是一致的（除了Hint中提到的几个点），但仍有许多细节问题没有被两张图覆盖，比如section 5.3中的Confliciting entry optimisize的问题。所以建议全篇精读，除了Section 7.8.9这些性能测试章节。
• https://pdos.csail.mit.edu/6.824/labs/lab-raft.html 给出了几个建议阅读链接，建议仔细阅读locking & strcture。教授给了很多额外的提示在里面。

Debug & log

• 实验为我们提供了Debug工具，“Dprintf”。这个工具或者自带的go.log 在调试中帮了我很多，我将不同的逻辑环节打上不同的TAG，这样在日志文件中就能更快地找到是哪个逻辑环节出现了问题。比如这是我的日志的一部分。日志中可以输出尽可能多的信息，比如server具体的log数组的信息，也可以画个图跟着日志来辅助debug。
• 在每一个实验后，我都先不开debug来运行，来改进一些语法和基础逻辑错误，基本都是出现“ one not come to argeement"之类的共识错误才开启日志。

Test (2C): basic persistence ...
2021/04/24 11:25:21 [ElectionTrickerRand++++++]Server 1, term 0, randTime 325
2021/04/24 11:25:21 [JoinElection+++++++] Sever 1, term 1
2021/04/24 11:25:21 [ElectionTrickerRand++++++]Server 2, term 0, randTime 327
2021/04/24 11:25:21 [JoinElection+++++++] Sever 2, term 1
2021/04/24 11:25:21 [ElectionReject+++++++]Server 2 reject 1, Have voter for 2
2021/04/24 11:25:21 [ElectionToFollower++++++]Server 0(term 0) into follower,candidate 1(term 1) 
2021/04/24 11:25:21 [ElectionReject+++++++]Server 1 reject 2, Have voter for 1
2021/04/24 11:25:21 [ElectionSUCCESS+++++++]Server 0 voted for 1!
2021/04/24 11:25:21 [ElectionReject+++++++]Server 0 reject 2, Have voter for 1
2021/04/24 11:25:21 [LeaderSuccess+++++] 1 got votenum: 2, needed >= 2, become leader for term 1
2021/04/24 11:25:21 [LeaderSendHeartBeat]Leader 1 (term 1) to server 0,nextIndex 1, matchIndex 0, lastIndex 0
2021/04/24 11:25:21 [LeaderSendHeartBeat]Leader 1 (term 1) to server 2,nextIndex 1, matchIndex 0, lastIndex 0
2021/04/24 11:25:21 [StartCommand] Leader 1 get command 11,index 1
2021/04/24 11:25:21 [GetHeartBeat]Sever 0, from Leader 1(term 1)
2021/04/24 11:25:21 [GetHeartBeat]Sever 2, from Leader 1(term 1)
2021/04/24 11:25:21 [FinishHeartBeat]Server 0, from leader 1(term 1), me.lastIndex 0
2021/04/24 11:25:21 [FinishHeartBeat]Server 2, from leader 1(term 1), me.lastIndex 0

Time.Timer

• 这是我实验中最大的一个坑。Hint中教授表示：You’ll need to write code that takes actions periodically or after delays in time. The easiest way to do this is to create a goroutine with a loop that calls time.Sleep(); (see the ticker() goroutine that Make() creates for this purpose). Don’t use Go’s time.Timer or time.Ticker, which are difficult to use correctly.
• 我这人偏偏不信邪，用time.Timer完成了A，在B中遇到了共识错误，经过再三确认逻辑正确后，我发现是我调用Timer的Reset方法无法刷新计时器。跳进源码后，提示其返回值不代表是否刷新。其Stop, Reset方法不熟悉的人真的不太好使用对。我就改弦更张，采用Time.Sleep的方法了。

Log Index

• raft 对consensus最重要的就是peers之间各种Index的同步。但需要注意的一点是：logEntry的Index 是从1开始的，而不同与log[]的数组下标。在完成 D Snapshot时，需要在数据中截断Index小于SnapShotIndex的Entry，而只保留Index大于的Entry在log[]数组中，因此，我们要正确区分EntryIndex(包括figure2中提及的所有Index)和ArrayIndex(用作下表去取log[]中的Entry）。例如，这是我对我的Index的使用说明：

• 我后来想到了更好的解决方案：是在Entry{}添加Index属性，这样可以直接get到，然后可以把log[0]用于存储lastSnapShotIndex 和 lastSnapShotTerm，这样后续的代码会有一定的简便性，不至于像我一样总是考虑边界条件，以防去拿到Entry信息的数据造成数据下标越界。

代码逻辑概述

• 我们需要在Raft对象中实现各种方法来满足test.test.go中的测试逻辑。
• 这些方法我将其分为以下几类：
  • Util 方法：一些通用的工具方法，例如

  func getRand(server int64) int{
  	rand.Seed(time.Now().Unix()+server)
  	return rand.Intn(ELECTION_TIMEOUT_MAX-ELECTION_TIMEOUT_MIN)+ELECTION_TIMEOUT_MIN
  }
  

  • 三个与Timeout相关的循环逻辑：Leader_election , Append_entries, Apply_entries
  • 持久化逻辑：persister.go & func persistData, persist, readPersist
  • 快照逻辑：InstallSnapShot RPC & Handler
  • 测试系统（Service)调用的接口：例如start, kill , CondInsallSnapShot

测试时间

• 整体ABCD的测试通过时间应该在8分钟，这样保证你的Lab3能够达到一个high performance（还没做3，猜的），这是我的，代码应该不是很solid, 勉强达标

go test -run 2A -race
Test (2A): initial election ...
  ... Passed --   3.0  3   46   11980    0
Test (2A): election after network failure ...
  ... Passed --   4.9  3  120   22160    0
Test (2A): multiple elections ...
  ... Passed --   5.4  7  420   79120    0
PASS
ok   6.824/raft 14.019s

go test -run 2B -race
Test (2B): basic agreement ...
  ... Passed --   1.0  3   16    4184    3
Test (2B): RPC byte count ...
  ... Passed --   3.0  3   48  113156   11
Test (2B): agreement despite follower disconnection ...
  ... Passed --   6.3  3  108   26570    8
Test (2B): no agreement if too many followers disconnect ...
  ... Passed --   3.7  5  192   36340    3
Test (2B): concurrent Start()s ...
  ... Passed --   0.9  3   14    3682    6
Test (2B): rejoin of partitioned leader ...
  ... Passed --   4.5  3  124   27537    4
Test (2B): leader backs up quickly over incorrect follower logs ...
  ... Passed --  31.4  5 2272 1619107  103
Test (2B): RPC counts aren't too high ...
  ... Passed --   2.2  3   36    9832   12
PASS
ok   6.824/raft 53.254s

go test -run 2C -race
Test (2C): basic persistence ...
  ... Passed --   4.1  3   78   18444    6
Test (2C): more persistence ...
  ... Passed --  18.1  5  972  185988   17
Test (2C): partitioned leader and one follower crash, leader restarts ...
  ... Passed --   1.9  3   32    7475    4
Test (2C): Figure 8 ...
  ... Passed --  33.7  5  972  192655   27
Test (2C): unreliable agreement ...
  ... Passed --   6.4  5  208   69991  246
Test (2C): Figure 8 (unreliable) ...
  ... Passed --  39.0  5 3160 6365101  525
Test (2C): churn ...
  ... Passed --  16.4  5  656  335440  237
Test (2C): unreliable churn ...
  ... Passed --  16.3  5  536  203178  112
PASS
ok   6.824/raft 136.328s

go test -run 2D -race
Test (2D): snapshots basic ...
  ... Passed --   8.3  3  134   46608  251
Test (2D): install snapshots (disconnect) ...
  ... Passed --  74.5  3 1370  350694  357
Test (2D): install snapshots (disconnect+unreliable) ...
  ... Passed --  81.4  3 1552  367242  389
Test (2D): install snapshots (crash) ...
  ... Passed --  40.6  3  666  179049  377
Test (2D): install snapshots (unreliable+crash) ...
  ... Passed --  51.0  3  794  199084  356
PASS
ok   6.824/raft 256.051s

A leader eleciton

• 需要我们完成 raft基础架构、LeaderElection、AppendEntries三部分

Raft 基础架构

• 理解并完成figure2中那些server需要维持的参数及后续RPC中需要传递的参数。我理解figure2中给出的是其MVP产品，后续我们为了实现的方便，可以自由向Raft对象添加属性。
• 为了使代码简洁，需要做一些顶层设计，抽象出共有变量和公用方法。
• 举例：

  type Raft struct {
  mu        sync.Mutex          // Lock to protect shared access to this peer's state
  peers     []*labrpc.ClientEnd // RPC end points of all peers
  persister *Persister          // Object to hold this peer's persisted state
  me        int                 // this peer's index into peers[]
  dead      int32               // set by Kill()
  
  currentTerm int
  votedFor int
  getVoteNum int
  log []Entry
  
  commitIndex int
  lastApplied int
  
  state int
  //election_timer *time.Timer
  //heartbeat_timer *time.Timer
  lastResetElectionTime time.Time
  
  nextIndex []int
  matchIndex []int
  
  applyCh chan ApplyMsg
  // SnapShot Point use
  lastSSPointIndex int
  lastSSPointTerm int
  }
  
  // use to change rafte state
  func (rf *Raft) changeState(howtochange int, resetTime bool) {
  
  if howtochange == TO_FOLLOWER {
  	//...
  }
  
  if howtochange == TO_CANDIDATE {
  	//...
  }
  
  if howtochange == TO_LEADER {
  	//...
  }
  

LeaderElection

• 这个逻辑是一个循环，需要在一个Ticker中进行超时触发，如果超时，server转换为Candidate,并自动 Join Election
• Tricker

func (rf *Raft) candidateElectionTicker() {
	for rf.killed() == false {
		nowTime := time.Now()
		time.Sleep(time.Duration(getRand(int64(rf.me)))*time.Millisecond)
		rf.mu.Lock()
		if rf.lastResetElectionTime.Before(nowTime) && rf.state != LEADER{
			rf.changeState(TO_CANDIDATE,true)
		}
		rf.mu.Unlock()
	}
}

• Join election 后，会投票给自己，向所有的server 并发发送RequestVote RPC，并等待其回复

func (rf *Raft) candidateJoinElection(){
	for index := range rf.peers{
		if index == rf.me{
			continue
		}
		// parallel send request
		go func(server int){
			rf.mu.Lock()
			rvArgs := RequestVoteArgs{
				rf.currentTerm,
				rf.me,
				rf.getLastIndex(),
				rf.getLastTerm(),
			}
			rvReply := RequestVoteReply{}
			rf.mu.Unlock()
			
			// waiting code should free lock first.
			re := rf.sendRequestVote(server,&rvArgs,&rvReply)
			if re == true {
				rf.mu.Lock()
				defer rf.mu.Unlock()
				// your code handle RPC reply
				return
			}
		}(index)
	}
}

• RequestVote中直接处理接受逻辑即可

AppendEntries

• 同样的逻辑，用一个Tricker来触发其执行，A中我们只需要携带空的Entry过去，即HeartBeat，还不需要处理log的逻辑。
• 最开始我想把heartbeat和appendentries需要分开写，以便heartbeat不用处理appendentries的额外逻辑而加速RPC返回速度。但后来发现，那样会造成大量的代码冗余和复杂性，还容易出bug。所以两者放在一个函数就可以，heartbeat处理逻辑和appendentries完全相同即可。

func (rf *Raft) leaderAppendEntriesTicker() {
	for rf.killed() == false {
		time.Sleep(HEARTBEAT_TIMEOUT * time.Millisecond)
		rf.mu.Lock()
		defer rf.mu.Unlock()
		if rf.state == LEADER {
			rf.leaderAppendEntries()
		} 
	}
}

• 在leaderAppendEntries函数中，同样的逻辑，向其余所有peers节点发送appendentries消息，再在RPCHandler中处理接受逻辑即可。

func (rf *Raft) leaderAppendEntries(){
	for index := range rf.peers{
		if index == rf.me {
			continue
		}
		// parallel replicate logs to sever
		go func(server int){
			rf.mu.Lock()
			if rf.state!=LEADER{
				rf.mu.Unlock()
				return
			}
			// for 2D ,leader send snapshot to peers
			prevLogIndextemp := rf.nextIndex[server]-1
			if prevLogIndextemp < rf.lastSSPointIndex{
				go rf.leaderSendSnapShot(server)
				rf.mu.Unlock()
				return
			}

			aeArgs := AppendEntriesArgs{}

			if rf.getLastIndex() >= rf.nextIndex[server] {
				// appendEntries 
				rf.printLogsForDebug()
				entriesNeeded := make([]Entry,0)
				entriesNeeded = append(entriesNeeded,rf.log[rf.nextIndex[server]-rf.lastSSPointIndex:]...)
				prevLogIndex,prevLogTerm := rf.getPrevLogInfo(server)
				aeArgs = AppendEntriesArgs{
					rf.currentTerm,
					rf.me,
					prevLogIndex,
					prevLogTerm,
					entriesNeeded,
					rf.commitIndex,
				}
			}else {
			// heartBeat
				prevLogIndex,prevLogTerm := rf.getPrevLogInfo(server)
				aeArgs = AppendEntriesArgs{
					rf.currentTerm,
					rf.me,
					prevLogIndex,
					prevLogTerm,
					[] Entry{},
					rf.commitIndex,
				}
			}
			aeReply := AppendEntriesReply{}
			rf.mu.Unlock()

			re := rf.sendAppendEntries(server,&aeArgs,&aeReply)
			
			if re == true {
				rf.mu.Lock()
				defer rf.mu.Unlock()
			// your code handle appendEntries
			}

		}(index)

	}
}

B log 补充A

• B中涉及到 log replication的逻辑，因此我们需要完善A中的投票选举和发送心跳的逻辑。
• 选举这部分主要是加入了一个 UpToDate的概念，即section5.4.1中的eletion restriction, 从而实现figure8中的fault tolerance属性。这里需要好好理解“up-to-date"的含义，peers只会投票给那些拥有”least as up-to-date than itself"的候选人。我的uptodate逻辑如下：

func (rf* Raft) UpToDate(index int, term int) bool {
	//lastEntry := rf.log[len(rf.log)-1]
	lastIndex := rf.getLastIndex()
	lastTerm := rf.getLastTerm()
	return term > lastTerm || (term == lastTerm && index >= lastIndex)
}

• 而AppendEntries 对于A的修改就比较大了，最主要的就是要判断以下几件事：
  • leader对于自己的同步状态是否准确
  • 自己的log是否与leader冲突
  • 从leader得知commit进度
• 这里建议再仔细阅读论文关于appendEntries规则，并充分理解作者如何设计使得raft达到其safe gurantee。还有一个值得注意的是，再论文第7页末尾提到了一个”Conflict Index Optimiszation“ 这个再figuer2中并没有出现，但建议这里直接实现，否则实验D中SnapShot时会发生大量重传，不能达到测试时间限制。
• 这里最重要的就是充分考虑 args.PrevLogIndex, len(rf.log)的关系，考虑在各种情况下，follower应该返回给leader什么样的信息来完成二者的高效协作同步。

ApplyEntries

• 这个就是模拟集群把已经commited 的Entry应用在状态虚拟机上的过程。我们要做的就是用一个循环，检测是否存在已经被commited但没有Applied的Entry，将Entry信息发送给测试系统提供的chan中，以视为Apply过程。

func (rf *Raft) committedToAppliedTicker(){
	// put the committed entry to apply on the state machine
	for rf.killed() == false{
		time.Sleep(APPLIED_TIMEOUT*time.Millisecond)
		rf.mu.Lock()
		if rf.lastApplied >= rf.commitIndex {
			rf.mu.Unlock()
			continue
		}
		Messages := make([]ApplyMsg,0)
		for rf.lastApplied < rf.commitIndex {
			rf.lastApplied +=1
			Messages = append(Messages,ApplyMsg{
				CommandValid: true,
				CommandIndex: rf.lastApplied,
				Command: rf.getLogWithIndex(rf.lastApplied).Command,
			})
		}
		rf.mu.Unlock()
		for _,messages := range Messages{
			rf.applyCh<-messages
		}
	}

}

C persistence

• 这是四个实验中代码量最小的一个环节，也相对比较简单。
• 我们需要理解persistence的作用，其实就是将一些不能自动推算出的”metadata”装载在存储中，在peers 崩溃后恢复自身状态时读取。figure2中已经指出，我们需要持久化currentTerm, votedFor, log[]三个变量，因为其他的状态都可以通过集群传递过来的信息进行推测和selfrecovery,因此不需要持久化。当然你为了图方便和效率也可以持久化尽可能多的状态。
• 这里要求我们阅读理解persiter.go，并完成readpersist 和persist函数,这两段代码基本骨架都在了，我们只需要理解它就可以。可以简单理解成一个persist编码，readpersist解码的过程。
• 在每一次任何一个需要持久化的量发生改变时触发persist, 在peers Init or Recover 时调用readpersist即可。
• 基本C实验本身不会造成错误，但测试C中有复杂的网络随机性变化情况，没有通过测试的原因可能仍然出现在你的LeaderElection 和 AppendEntries上。

D log compaction

• 这里需要仔细阅读论文section 7,理解如何通过SnapShot 来实现log compaction,限制无限制的log增长，减轻peers的存储压力。以及这种方式的优点。
• 仔细阅读raft_diagram.pdf, 理解service如何与raft code交互，来实现SnapShot。
• 注意引入SnapShot后会对原有的Index系统造成比较大的改变，与Index取log信息有关的地方，都需要考虑lastSnapShotIndex来更改。但这个在你理解了SnapShotIndex的涵义后就不难修改，就变成了一个体力活了。
• SnapShot() 是peers主动去存储Snapshot的调用借口，而InstallSnapShotRPC 则是Leader将自己的SnapShot同步给大家的过程，这两者并不矛盾。
• leader 向 follower发送Snapshot的条件是 ：rf.nextIndex[server]-1 < rf.lastSnapShotIndex。 即server需要同步的下一个Entry已经被我给存储持久化了，因为持久化的状态肯定及已经经过majority commite 然后apply了，你还在要求它，说明你来的太晚，进度照大家差得太多了，我就给你加个速，直接告诉你SnapShotIndex前的都已经apply了，你直接全部接受就行了。

func (rf *Raft) Snapshot(index int, snapshot []byte) {
	// Your code here (2D).
	rf.mu.Lock()
	defer rf.mu.Unlock()
	if rf.lastSSPointIndex >= index || index > rf.commitIndex{
		return
	}
	// trim the peers log and update rf.SnapShot State, commitIndex, lastApplied
	//...
	// user persister to persist SnapShot
	rf.persister.SaveStateAndSnapshot(rf.persistData(),snapshot)
}

// InstallSnapShot RPC Handler
func (rf *Raft) InstallSnapShot(args *InstallSnapshotArgs, reply *InstallSnapshotReply){
	rf.mu.Lock()
	if rf.currentTerm > args.Term{
		reply.Term = rf.currentTerm
		rf.mu.Unlock()
		return
	}
	// your code handle InstallSnapShot RPC
	rf.persister.SaveStateAndSnapshot(rf.persistData(),args.Data)
	msg := ApplyMsg{
		SnapshotValid: true,
		Snapshot: args.Data,
		SnapshotTerm: rf.lastSSPointTerm,
		SnapshotIndex: rf.lastSSPointIndex,
	}
	rf.mu.Unlock()
	rf.applyCh <- msg
}

func (rf *Raft) leaderSendSnapShot(server int){
	rf.mu.Lock()
	ssArgs := InstallSnapshotArgs{
		rf.currentTerm,
		rf.me,
		rf.lastSSPointIndex,
		rf.lastSSPointTerm,
		rf.persister.ReadSnapshot(),
	}
	ssReply := InstallSnapshotReply{}
	rf.mu.Unlock()
	
	re := rf.sendSnapShot(server,&ssArgs,&ssReply)
	
		if re == true {
			rf.mu.Lock()
			defer rf.mu.Unlock()
			// your code handle InstallSnapshot
			return
		}
}

完整代码（lab1,2,3,4）

• https://github.com/AlohaPoster/MIT6.824_2021_spring