MIT 6.824 Llab2B Raft之日志复制

书接上文Raft Part A | MIT 6.824 Lab2A Leader Election。

实验准备

实验代码：git://g.csail.mit.edu/6.824-golabs-2021/src/raft
如何测试：go test -run 2B -race
相关论文：Raft Extended Section 5.3 Section 5.4.1
实验指导：6.824 Lab 2: Raft (mit.edu)

实验目标

在Leader Election的基础上，完成Leader和Follower的关于Log Replication的代码。

一些提示

从实现Start()开始，接着完成AppendEntries RPC用于发送和接收日志。
需要实现Election Restriction，即论文的5.4.1节。
注意在Leader存活期间，不要有节点发起选举。
不要让一些循环不间断的执行，否则会因为速度过慢而无法通过测试，使用条件变量或休眠来解决。
让代码清晰干净，方便后面的实验。
如果未能通过测试，查看test_test.go和config.go。

复制状态机

一条日志的内容可能是这样子：SET A 0，它的涵义是存储一个名字为'A'的数据，他的值为'0'。一个日志只是存储了一个大小为7字节的字符串，数据真正被存储到状态机中。你可以把状态机理解为数据库。

客户端将指令SET A 0发送到Leader，Leader会将指令追加到自己的日志序列中，Leader中未复制到其他节点的日志，会在随着下一次心跳复制到其他节点，被复制到半数以上节点的日志将被标记为Commit状态，并由另一个goroutine负责提交到状态机中，被提交到状态机的日志被标记为Applied状态。

commitIndex & lastApplied

指令是不断追加到日志序列中的，日志的同步要保证顺序性，状态机应该由远及近的执行命令，才能保证节点之间的状态一致。

因此如果log[i]被标记为Commit或Applied，那么log[0] ~ log[i-1]也一定是Commit或Applied状态。论文中使用commitIndex表示最后一个状态为Commit的日志索引，使用lastApplied表示最后一个状态为Applied的日志索引。

并且由于只有Applied只能由Commit转换而来，所以lastApplied <= commitIndex一定成立。

nextIndex

日志从Leader复制到Followers，有如下三种方式。

Leader每增加一个日志，就发送给Followers。
Leader随着每次心跳，将自己所有的日志都发送给Followers。
Leader随着每次心跳，仅发送未复制到Followers的部分日志。

为了效率，选择方式3是显然的，那么Leader如何知道每个Follower都需要哪部分日志呢？

通过各个节点的日志序列可以得出，commitIndex = 7，因为1 ~ 7号日志已经同步到半数以上的节点。

因为在日志同步过程中保证顺序性，Follower相较于Leader缺失的是“后半部分”日志，Leader使用nextIndex[]数组标记每个Follower缺失日志的开始位置（包括Leader自己）。本例中，nextIndex[] = {9, 6, 9, 3, 8}。

在本例中，Leader要向第一个Follower同步的部分日志就是从nextIndex[1]到最后，即log[6]，log[7]，log[8]。

logEntry

你可能注意到上图中日志不止有指令，事实上论文中的日志有两个字段，分别是Term和Command。

type LogEntry struct {

    Term    int

    Command interface{}

}

Command用于存储如SET A 0这样的指令，那么为什么要在日志中加入Term呢？

假设Leader收到了客户端请求，并追加到了自己的日志序列中，还未等待同步到其他节点，就挂掉了，此时剩余节点选出了新的Leader，继续正常工作，状态如下。

Log	Old Leader	New Leader	Follower
1	'SET A 0'	'SET A 0'	'SET A 0'
2	'SET B 1'	'SET C 2'	'SET C 2'
3	OFFLINE	'SET A 1'	'SET A 1'

注意：为了方便处理边界，论文中日志序列索引从1开始。

若此时Old Leader恢复了，感知到New Leader后会转会为Follower，此时nextIndex[0] = 3，New Leader会将log[3:]同步到Old Leader中。

Log	Follower	New Leader	Follower
1	'SET A 0'	'SET A 0'	'SET A 0'
2	'SET B 1'	'SET C 2'	'SET C 2'
3	'SET A 1'	'SET A 1'	'SET A 1'

可以看到，现在集群已经出现了不一致的情况，看样子logEntry应该存储更多的信息，用来保证一致性。

那么为什么是Term呢？回想上文提到的日志的时序性和Leader的唯一性，集群在某一个Term中，Leader是唯一的，在给定Term的前提下，能够确保在该任期内，集群状态是一致的。

Log	Follower	New Leader	Follower
1	[T1] 'SET A 0'	[T1] 'SET A 0'	[T1] 'SET A 0'
2	[T1] 'SET B 1'	[T2] 'SET C 2'	[T2] 'SET C 2'
3	[T2] 'SET A 1'	[T2] 'SET A 1'	[T2] 'SET A 1'

观察在logEntry中引入Term后的状态，不难发现导致不一致的观点就在Term上。

prevLogIndex & prevLogTerm

如何利用Term，在日志同步过程中发现并纠正不一致的日志呢？下图是论文中给出的比较极端的不一致的情况。

目前nextIndex[] = {11, 10, 5, 12, 13, 8, 12}。

nextIndex[a] = 10，Leader需要同步日志序列log[10:]，并且要保证a的log[1:10]和自己的log[1:10]保持一致，关键点在于log[9]，即log[nextIndex[a]-1]。

首先如果要Leader.log[1:10]和a.log[1:10]是一致能，那么Leader.log[9]和a.Log[9]也一定是一致的，根据上一小节的结论，若Leader.log[9].Term == a.log[9].Term，则Leader.log[9].Command == a.log[9].Command一定存在。

Raft算法规定，当Leader.log[N-1].Term == Follower.log[N-1].Term时，就可以把Leader.log[N]同步到Follower。

注意，这其实是一个动态规划的思想。

if L.log[N-1].Term == F.log[N-1].Term {

    F.log[N] = L.log[N]

}

由此可得出日志匹配原则：如果两个日志在相同索引的位置的Term相同，那么从开始到该索引位置的日志一定相同。

注意N和Term是由Leader.nextIndex和Leader.log得来，而上面的代码逻辑是在Follower中，Follower如何得知呢，论文中在AppendEntriesArgs中引入prevLogIndex和prevLogTerm，分别表示N-1和L.log[N-1].Term。

因此在AppendEntries中有如下逻辑。

/* Reply false if log doesn't contain an entry at pervLogIndex whose term matches pervLogTerm. */

if len(rf.log) <= args.PrevLogIndex || rf.log[args.PrevLogIndex].Term != args.PrevLogTerm {

    return false

}

Follower拒绝后，Leader如何处理呢？Follower错误和缺失的日志部分依旧是需要修正和同步的。一个简单的办法是让nextIndex递减，直到找到Leader和Follower日志序列中一致的位置。

因此在heartbeat（此处Part A中的heartbeat名字或许改为sync更恰当）中有如下逻辑。

/* If AppendEntries fails because of log inconsistency: decrement nextIndex and retry */

if !reply.Success {

    rf.nextIndex[i] = max(1, rf.nextIndex[i]-1)

}

注意log索引最小为1的问题，如果你觉得这种方式效率太低，可以尝试在AppendEntriesReply中添加更多的字段提升效率。

matchIndex

commitIndex是由Leader根据日志同步情况确定的，同步到半数以上节点的日志会被标记为commitIndex，论文在Leader中引入了matchIndex[]数组用于辅助计算commitIndex，在日志同步成功时，会同步更新nextIndex和matchIndex，两者的关系为matchIndex = nextIndex - 1。matchIndex是个相对不是那么重要的概念，你也可以用其它方式计算commitIndex，这里使用排序取中位数的方式。

因此在heartbeat（此处Part A中的heartbeat名字或许改为sync更恰当）有如下逻辑。

if reply.Success {

    /* If successful: update nextIndex and matchIndex for follower. */

    rf.nextIndex[i] += len(args.Entries)

    rf.matchIndex[i] = rf.nextIndex[i] - 1

    match := make([]int, len(rf.peers))

    copy(match, rf.matchIndex)

    sort.Ints(match)

    majority := match[(len(rf.peers)-1)/2]

    /* If there exists an N such that N > commitIndex, a majority of matchIndex[i] >= N, and log[N].Term == currentTerm: set commitIndex = N */

    if majority > rf.commitIndex && rf.log[majority].Term == rf.currentTerm {

        rf.commitIndex = majority

    }

}

commitIndex是不只是Leader的概念，集群中所有节点的commitIndex应当保持一致。论文在AppendEntriesArgs中引入了leaderCommit字段，用于Follower更新commitIndex。

因此在AppendEntries中有如下逻辑。

/* If leaderCommit > commitIndex, set commitIndex = min(leaderCommit, index of last new entry) */

if args.LeaderCommit > rf.commitIndex {

    rf.commitIndex = min(args.LeaderCommit, len(rf.log)-1)

}

如何理解rf.log[majority].Term == rf.currentTerm？

图中黑色框选表示Leader，注意(c)，此时S1 Leader，Term为2，Index为2的日志在半数以上节点中存在，但是S1不能将该日志标记为commit状态。因为如果此时Leader转为S5，该日志将会被覆盖。

所以在标记commit状态时需要判断Term是否和currentTerm相等，只有当前Term的日志才可以标记为commit状态。也就是(e)的状态。

lastLogTerm & lastLogIndex

让我们再次观察上文中提过的图片，不知道你在之前是否发现了一些问题。

当Leader挂掉后，会从Followers中选举新的Leader，在Part A中，每个Follower的选票会投给第一个向自己索要选票的Candidate，也就是说每个Follower都有可能被选举为新的Leader。

假设第三个Follower，即日志序列长度为2的Follower竞选成功，那么自log[3]开始向后的所有日志都会在同步中丢失。但是已经标记为commit状态的日志是不应该丢失的。

Raft算法规定，投票者只会将选票投给日志和自己至少一样新的成员。这样保证了commit状态的日志不会丢失。即论文5.4.1节的选举限制。

论文中在RequestVoteArgs中引入了lastLogIndex和lastLogTerm两个字段用于辅助实现选举限制，分别表示候选人最后一个日志的索引和任期。

因此在RequestVote中有如下逻辑。

/* If votedFor is null or candidateId, and candidate's log is at least as up-to-date as receiver's log, grant vote. */

if rf.votedFor == -1 || rf.votedFor == args.CandidateId {

    if args.LastLogTerm > rf.log[len(rf.log)-1].Term || args.LastLogIndex >= len(rf.log)-1 && rf.log[len(rf.log)-1].Term == args.LastLogTerm {

        rf.votedFor = args.CandidateId

        reply.VoteGranted = true

    }

}

实验总结

我用绿色的线框选了Part B中需要实现的内容，和本文代码中的注释是一致的，你可以通过关键字搜索定位到相应的代码。

除了本文中给出的代码外，想要通过测试还需要实现Start()和Apply()两个函数，分别用于客户端发送指令和提交日志到状态机，这两个函数并不难实现所以本文就不给出具体代码了。

最后，为了证明我不是在乱写，附上我的测试结果。

巴特西