引言
MIT 6.824的系列实验是构建一个容错Key/Value存储系统，实验2是这个系列实验的第一个。在实验2(Lab 2)中我们将实现Raft这个基于复制状态机(replicated state machine)的共识协议。本文将详细讲解Lab 2B。Lab 2A在这里！
正文
Lab 2B的任务是实现日志复制(log replication)，对应论文的5.3和5.4.1章节。我们的代码要能够选举出“合法”的leader，通过AppendEntries RPC复制日志，已提交(committed)的日志意味着复制到了多数派server，随后要将其正确地返回给上层应用执行。
数据结构
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()
// Your data here (2A, 2B, 2C).
// Look at the paper's Figure 2 for a description of what
// state a Raft server must maintain.
state State
lastReceive int64
currentTerm int
votedFor int
// New data structures in Lab 2B
log []LogEntry
commitIndex int
lastApplied int
applyCh chan ApplyMsg
moreApply bool
applyCond *sync.Cond
nextIndex []int
matchIndex []int
}
type LogEntry struct {
Command interface{}
Term int
}
type ApplyMsg struct {
CommandValid bool
Command interface{}
CommandIndex int
}
复制代码在Raft的数据结构中，我们为Lab 2B新增了LogEntry(日志项)的结构体。根据论文的要求，每一个LogEntry要包含对应的命令，以及leader接收该命令时的term。
根据Figure 2，还要定义以下几个变量：
- commitIndex: 已知被提交的最高日志项对应的index。当日志项被提交(committed)了，意味着该日志项已经成功复制到了集群中的多数派server上，属于“集体记忆”了。如果当前的leader宕机再次发生选举，只有拥有完整已提交日志的server才能够获得多数派选票，才能被选举为leader。根据Leader完整性(Leader Completeness)，如果一个日志项在某个term被提交了，则该Entry会存在于所有更高term的leader日志中。
- lastApplied: 应用(apply)给状态机的最高日志项的index，也就是上层应用“消费”到Raft日志项的最新index。
Leader使用nextIndex和matchIndex两个数组来维护集群中其它server的日志状态。在实现上有一点区别的是，论文中数组从1开始，我们在代码中从0开始（包括log）。
- nextIndex[]: 每个server分别对应着数组中的一个值。下一次要发给对应server的日志项的起始index。
- matchIndex[]: 每个server分别对应着数组中的一个值。已知成功复制到该server的最高日志项的index。
nextIndex可以被看作是乐观估计，值一开始被设置为日志的最高index，随着AppendEntry RPC返回不匹配而逐渐减小。matchIndex是保守估计，初始时认为没有日志项匹配（对应我们代码中的-1，论文中的0），必须AppendEntry RPC匹配上了才能更新值。这样做是为了数据安全：只有当某个日志项被成功复制到了多数派，leader才能更新commitIndex为日志项对应的index。
新Leader在选举成功后要重新初始化nextIndex和matchIndex这两个数组，然后通过AppendEntry RPC收集其它server的日志状态，具体细节我们在下面的日志复制(AppendEntries) 小节配合代码详细讲解。
除了论文中的这些状态，在Lab 2B的代码实现中，我们会单独使用一个goroutine(appMsgApplier)，负责不断将已经被提交的日志项返回给上层应用，所以还需要额外添加以下几个变量用于goroutine同步：
- applyCh: 由实验提供，通过该channel将ApplyMsg发送给上层应用。
- moreApply: 示意有更多的日志项已经被提交，可以apply。
- applyCond: apply时用于多goroutine之间同步的Condition。
type RequestVoteArgs struct {
// Your data here (2A, 2B).
Term int
CandidateId int
LastLogIndex int
LastLogTerm int
}
type RequestVoteReply struct {
// Your data here (2A).
Term int
VoteGranted bool
}
复制代码这里RequestVote RPC的结构体相比于Lab 2A，新增了最后一个日志项的信息。LastLogIndex是 candidate最后一个日志项的index，而LastLogTerm是candidate最后一个日志项的term。这两个参数将用于下文中选举限制(election restriction)的判断。
type AppendEntryArgs struct {
// Your data here (2A, 2B).
Term int
LeaderId int
PrevLogIndex int
PrevLogTerm int
Entries []LogEntry
LeaderCommit int
}
type AppendEntryReply struct {
// Your data here (2A).
Term int
Success bool
// fast back up
XTerm int
XIndex int
XLen int
}
复制代码除了Term和LeaderId，在Lab 2B中AppendEntryArgs结构体新增了如下几个参数：
- Entries[]: 发送给对应server的新日志，如果是心跳则为空。这里要发送给对应server日志的index，是从nextIndex到最后一个日志项的index，注意也可能为空。
- PrevLogIndex: 紧跟在新日志之前的日志项的index，是leader认为follower当前可能已经同步到了的最高日志项的index。对于第i个server，就是nextIndex[i] - 1。
- PrevLogTerm: prevLogIndex对应日志项的term。
- LeaderCommit: leader已经提交的commit index。用于通知follower更新自己的commit index。
AppendEntryReply结构体新增了XTerm、XIndex和XLen几个变量用于nextIndex的快速回退(back up)。我们知道，论文中的nextIndex在AppendEntry RPC返回不匹配后，默认只是回退一个日志项(nextIndex[i]=PrevLogIndex)。如果follower能够返回更多信息，那么leader可以根据这些信息使对应server的nextIndex快速回退，减少AppendEntry RPC通信不匹配的次数，从而加快同步日志的步伐。这几个变量的具体含义：
- XLen: 当前follower所拥有的的日志长度。
- XTerm: 当前follower的日志中，PrevLogIndex所对应日志项的term。可能为空。
- XIndex: 当前follower的日志中，拥有XTerm的日志项的最低index，可能为空。
主要函数
func Make(peers []*labrpc.ClientEnd, me int,
persister *Persister, applyCh chan ApplyMsg) *Raft {
rf := &Raft{}
rf.mu = sync.Mutex{}
rf.peers = peers
rf.persister = persister
rf.me = me
rf.state = Follower
rf.currentTerm = 0
rf.votedFor = -1
rf.lastReceive = -1
// new code in Lab 2B
rf.log = make([]LogEntry, 0)
rf.commitIndex = -1
rf.lastApplied = -1
rf.nextIndex = make([]int, len(peers))
rf.matchIndex = make([]int, len(peers))
rf.applyCh = applyCh
rf.moreApply = false
rf.applyCond = sync.NewCond(&rf.mu)
// Your initialization code here (2A, 2B, 2C).
// initialize from state persisted before a crash
rf.readPersist(persister.ReadRaftState())
go rf.leaderElection()
// new code in Lab 2B
go rf.appMsgApplier()
return rf
}
复制代码Make函数是创建Raft server实例的入口，此处我们初始化Raft实例的各个变量。除了在goroutine中开始选主计时，我们还额外增加了一个appMsgApplier用于各个Raft实例apply已提交的日志给各自的上层应用。
//
// the service using Raft (e.g. a k/v server) wants to start
// agreement on the next command to be appended to Raft's log. if this
// server isn't the leader, returns false. otherwise start the
// agreement and return immediately. there is no guarantee that this
// command will ever be committed to the Raft log, since the leader
// may fail or lose an election. even if the Raft instance has been killed,
// this function should return gracefully.
//
// the first return value is the index that the command will appear at
// if it's ever committed. the second return value is the current
// term. the third return value is true if this server believes it is
// the leader.
//
func (rf *Raft) Start(command interface{}) (int, int, bool) {
index := -1
term := -1
isLeader := true
// Your code here (2B).
if rf.killed() {
return index, term, false
}
rf.mu.Lock()
defer rf.mu.Unlock()
isLeader = rf.state == Leader
if isLeader {
rf.log = append(rf.log, LogEntry{Term: rf.currentTerm, Command: command})
index = len(rf.log) - 1
term = rf.currentTerm
rf.matchIndex[rf.me] = len(rf.log) - 1
rf.nextIndex[rf.me] = len(rf.log)
DPrintf("[%d]: Start received command: index: %d, term: %d", rf.me, index, term)
}
return index + 1, term, isLeader
}
复制代码上层应用接收来自客户端的请求，通过Start函数对将要追加到Raft日志的command发起共识。注意读写要上锁，如果server不是leader则返回false。如果是leader的话，那么将command组装成LogEntry后追加到自己的日志中。此处要同时更新leader自己的matchIndex和nextIndex，目的是防止下面更新commitIndex时对多数派的判断出错。由于我们的日志数组index是从0开始，而论文是从1开始，因此我们返回的index要在原有基础上加一。
func (rf *Raft) convertToLeader() {
DPrintf("[%d]: convert from [%s] to [%s], term [%d]", rf.me, rf.state, Leader, rf.currentTerm)
rf.state = Leader
rf.lastReceive = time.Now().Unix()
for i := 0; i < len(rf.peers); i++ {
rf.nextIndex[i] = len(rf.log)
rf.matchIndex[i] = -1
}
}
复制代码convertToLeader函数，在原有Lab 2A的基础上，需要重新初始化nextIndex[]和matchIndex[]，由调用者负责上锁。
选举限制(Election Restriction)
在Lab 2B中，我们需要为选主环节额外添加一些参数(lastLogIndex和lastLogTerm)，确保满足「只有拥有完整已提交日志的server才能够被选举为leader」的选举限制。
func (rf *Raft) kickOffLeaderElection() {
rf.convertToCandidate()
voteCount := 1
totalCount := 1
cond := sync.NewCond(&rf.mu)
// prepare lastLogIndex and lastLogTerm
lastLogIndex := len(rf.log) - 1
lastLogTerm := -1
if lastLogIndex >= 0 {
lastLogTerm = rf.log[lastLogIndex].Term
}
for i := 0; i < len(rf.peers); i++ {
if i != rf.me {
go func(serverTo int, term int, candidateId int, lastLogIndex int, lastLogTerm int) {
args := RequestVoteArgs{term, candidateId, lastLogIndex, lastLogTerm}
reply := RequestVoteReply{}
DPrintf("[%d]: term: [%d], send request vote to: [%d]", candidateId, term, serverTo)
ok := rf.sendRequestVote(serverTo, &args, &reply)
rf.mu.Lock()
defer rf.mu.Unlock()
totalCount += 1
if !ok {
cond.Broadcast()
return
}
if reply.Term > rf.currentTerm {
rf.convertToFollower(reply.Term)
} else if reply.VoteGranted && reply.Term == rf.currentTerm {
voteCount += 1
}
cond.Broadcast()
}(i, rf.currentTerm, rf.me, lastLogIndex, lastLogTerm)
}
}
go func() {
rf.mu.Lock()
defer rf.mu.Unlock()
for voteCount <= len(rf.peers)/2 && totalCount < len(rf.peers) && rf.state == Candidate {
cond.Wait()
}
if voteCount > len(rf.peers)/2 && rf.state == Candidate {
rf.convertToLeader()
go rf.operateLeaderHeartbeat()
}
}()
}
复制代码kickOffLeaderElection()中正式开始选主，我们在发送RequestVote RPC请求投票的实现中，额外增加了lastLogIndex和lastLogTerm。这里lastLogIndex是candidate最后一个日志项的index，如果日志为空那么为-1。lastLogTerm初始为-1，要在lastLogIndex大于等于零的情况下才能赋值，防止数组越界。计票部分和Lab 2A相同，不再赘述。
下面先看一下RequestVote请求投票这个RPC额外新增了哪些逻辑：
func (rf *Raft) RequestVote(args *RequestVoteArgs, reply *RequestVoteReply) {
// Your code here (2A, 2B).
rf.mu.Lock()
defer rf.mu.Unlock()
DPrintf("[%d]: received vote request from [%d]", rf.me, args.CandidateId)
if args.Term < rf.currentTerm {
reply.Term = rf.currentTerm
reply.VoteGranted = false
return
}
// If RPC request or response contains term T > currentTerm:
// set currentTerm = T, convert to follower (§5.1)
if args.Term > rf.currentTerm {
rf.convertToFollower(args.Term)
}
reply.Term = rf.currentTerm
DPrintf("[%d]: status: term [%d], state [%s], vote for [%d]", rf.me, rf.currentTerm, rf.state, rf.votedFor)
// 新增extra condition in Lab 2B
// 选举限制
if (rf.votedFor < 0 || rf.votedFor == args.CandidateId) &&
(len(rf.log) == 0 || (args.LastLogTerm > rf.log[len(rf.log)-1].Term) ||
(args.LastLogTerm == rf.log[len(rf.log)-1].Term && args.LastLogIndex >= len(rf.log)-1)) {
rf.votedFor = args.CandidateId
rf.lastReceive = time.Now().Unix()
reply.VoteGranted = true
DPrintf("[%d]: voted to [%d]", rf.me, args.CandidateId)
return
}
reply.VoteGranted = false
}
复制代码根据论文中的选举限制，我们在投票时要额外判断：
- 是否没投票或者投给的是这个candidate。
- candidate的log是否至少和接受者的log一样新(up-to-date)。
当全部满足条件才能够投票。
Raft是通过比较两个server日志的最后一个日志项的index和term，来判别哪个更up-to-date的：
- 如果两个server的日志的最后一个日志项的term不同，那么拥有更晚term日志项的server的日志更up-to-date。
- 如果最后一个日志项的term相同，那么日志更长的更up-to-date。
在判别up-to-date的实现中，我们还要额外考虑当前的接受者日志为空的情况。
日志复制(AppendEntries)
在Lab 2A中，我们实现了server选举晋升为leader后，立即并周期性的通过AppendEntry发送心跳。而在Lab 2B中，我们同样要通过AppendEntry RPC进行日志复制，这也是本个实验的重点。
operateLeaderHeartbeat()中，我们还是在新的goroutine中发送AppendEntry RPC。在Lab 2B中新增了prevLogIndex、prevLogTerm、entries和leaderCommit几个参数：
- prevLogIndex，对于第i个server，就是rf.nextIndex[i] - 1，是紧跟在要发送给server[i]日志之前的日志项的index，用于接收者判别日志的同步情况。
- prevLogTerm是prevLogIndex对应日志项的term，为避免数组越界要判断prevLogIndex是否大于等于0。
- entries是发送给对应server的新日志，从rf.nextIndex[i]到最后一个日志项的index，注意也可能为空。
根据论文的日志匹配性质(Log Matching Property)：
- 如果来自不同日志的两个日志项有相同的index和term，那么它们存储了相同的command。
- 如果来自不同日志的两个日志项有相同的index和term，那么它们前面的日志完全相同。
因此PrevLogIndex和PrevLogTerm与follower的日志部分匹配，就能确保follower的PrevLogIndex前的日志一致了。
func (rf *Raft) operateLeaderHeartbeat() {
for {
if rf.killed() {
return
}
rf.mu.Lock()
if rf.state != Leader {
rf.mu.Unlock()
return
}
for i := 0; i < len(rf.peers); i++ {
if i != rf.me {
// // 在Lab 2B中此处新增了prevLogIndex、prevLogTerm、entries和leaderCommit
// If last log index ≥ nextIndex for a follower: send
// AppendEntries RPC with log entries starting at nextIndex
// • If successful: update nextIndex and matchIndex for
// follower (§5.3)
// • If AppendEntries fails because of log inconsistency:
// decrement nextIndex and retry (§5.3)
prevLogIndex := rf.nextIndex[i] - 1
prevLogTerm := -1
if prevLogIndex >= 0 {
prevLogTerm = rf.log[prevLogIndex].Term
}
var entries []LogEntry
if len(rf.log)-1 >= rf.nextIndex[i] {
DPrintf("[%d]: len of log: %d, next index of [%d]: %d", rf.me, len(rf.log), i, rf.nextIndex[i])
entries = rf.log[rf.nextIndex[i]:]
}
go func(serverTo int, term int, leaderId int, prevLogIndex int, prevLogTerm int, entries []LogEntry, leaderCommit int) {
args := AppendEntryArgs{term, leaderId, prevLogIndex, prevLogTerm, entries, leaderCommit}
reply := AppendEntryReply{}
ok := rf.sendAppendEntry(serverTo, &args, &reply)
rf.mu.Lock()
defer rf.mu.Unlock()
if !ok {
return
}
if reply.Term > rf.currentTerm {
rf.convertToFollower(reply.Term)
return
}
// Drop the reply of old term RPCs directly
if rf.currentTerm == term && reply.Term == rf.currentTerm {
if reply.Success {
rf.nextIndex[serverTo] = prevLogIndex + len(entries) + 1
rf.matchIndex[serverTo] = prevLogIndex + len(entries)
// If there exists an N such that N > commitIndex, a majority
// of matchIndex[i] ≥ N, and log[N].term == currentTerm:
// set commitIndex = N (§5.3, §5.4).
matches := make([]int, len(rf.peers))
copy(matches, rf.matchIndex)
sort.Ints(matches)
majority := (len(rf.peers) - 1) / 2
for i := majority; i >= 0 && matches[i] > rf.commitIndex; i-- {
if rf.log[matches[i]].Term == rf.currentTerm {
rf.commitIndex = matches[i]
DPrintf("[%d]: commit index [%d]", rf.me, rf.commitIndex)
rf.sendApplyMsg()
break
}
}
} else {
// In Test (2C): Figure 8 (unreliable), the AppendEntry RPCs are reordered
// So rf.nextIndex[serverTo]-- would be wrong
rf.nextIndex[serverTo] = prevLogIndex
if rf.nextIndex[serverTo]-1 >= reply.XLen {
rf.nextIndex[serverTo] = reply.XLen
} else {
for i := rf.nextIndex[serverTo] - 1; i >= reply.XIndex; i-- {
if rf.log[i].Term != reply.XTerm {
rf.nextIndex[serverTo] -= 1
} else {
break
}
}
}
}
}
}(i, rf.currentTerm, rf.me, prevLogIndex, prevLogTerm, entries, rf.commitIndex)
}
}
rf.mu.Unlock()
time.Sleep(time.Duration(heartBeatInterval) * time.Millisecond)
}
}
复制代码在通过sendAppendEntry()发送AppendEntry RPC并收到对应server响应后，首先判断返回的term看是否降级为follower。
接下来要很重要的一点，由于RPC在网络中可能乱序或者延迟，我们要确保当前RPC发送时的term、当前接收时的currentTerm以及RPC的reply.term三者一致，丢弃过去term的RPC，避免对当前currentTerm产生错误的影响。
当reply.Success为true，说明follower包含了匹配prevLogIndex和prevLogTerm的日志项，更新nextIndex[serverTo]和matchIndex[serverTo]。这里只能用prevLogIndex和entries来更新，而不能用nextIndex及len(log)，因为后两者可能已经被别的RPC更新了，进而导致数据不一致。正确的更新方式应该是：rf.nextIndex[serverTo] = prevLogIndex + len(entries) + 1，rf.matchIndex[serverTo] = prevLogIndex + len(entries)。
由于matchIndex发生了变化，我们要检查是否更新commitIndex。根据论文，如果存在一个N，这个N大于commitIndex，多数派的matchIndex[i]都大于等于N，并且log[N].term等于currentTerm，那么更新commitIndex为N。这里必须注意，日志提交是有限制的，Raft从不提交过去term的日志项，即使已经复制达到了多数派。如果要更新commitIndex为N，那么N所对应的日志项的term必须是当前currentTerm。
论文的Figure 8仔细讲解了「leader只能提交term为curretTerm的日志项」的问题。在(c)中S1的currentTerm为4，不能提交即使已经复制到多数派的term为2的日志项，原因是可能会如(d)所示被term为3的日志项覆盖。但如(e)所示，如果term为4的日志项被复制到了多数派，那么此时S1可以将日志提交。因为S1作为leader，它的currentTerm是当前的最高term，当该currentTerm的日志项被复制到多数派后，根据up-to-date规则，不会再有较低term的server在选举获得多数派选票而成为leader，也就不再会有像(d)中覆盖的情况发生。
在检查是否更新commitIndex的实现上，我们将matchIndex复制到了matches数组中，通过sort升序排序。那么在majority := (len(rf.peers) - 1) / 2时，大于一半的matchIndex大于等于matches[majority]，因此rf.log[matches[majority]]恰好被复制到了多数派server。以majority为初始值自减遍历i，如果rf.log[matches[i]].Term == rf.currentTerm，那么说明满足日志提交限制，找到了上述最大的“N”，随后调用sendApplyMsg()，通知有更多的日志项已经被提交，可以apply。循环的停止条件为i < 0 || matches[i] <= rf.commitIndex，则说明没有找到更大的commitIndex。
当reply.Success为false，说明follower的日志不包含在prevLogIndex处并匹配prevLogTerm的日志项，要将nextIndex缩减。此处更新不宜采用自减的方式更新，因为RPC可能会重发，正确的方式是rf.nextIndex[serverTo] = prevLogIndex。
我们在AppendEntryReply中增加了几个变量，以使nextIndex能够快速回退(back up)。如果接下来要尝试匹配的prevLogIndex比follower当前所拥有的的日志长度（XLen）还要大，那么显然直接从XLen尝试匹配即可。如果接下来要尝试匹配的prevLogIndex在XLen以内，因为我们已经知道了follower的日志从XIndex到当前prevLogIndex的日志项的term都是XTerm，那么我们可以直接在leader侧遍历匹配一遍，而无需多次往返RPC通信。
func (rf *Raft) AppendEntry(args *AppendEntryArgs, reply *AppendEntryReply) {
// Your code here (2A, 2B).
rf.mu.Lock()
defer rf.mu.Unlock()
DPrintf("[%d]: received append entry from [%d], args term: %d, LeaderCommit: %d, prevLogIndex: %d, prevLogTerm: %d, len(entry): %d",
rf.me, args.LeaderId, args.Term, args.LeaderCommit, args.PrevLogIndex, args.PrevLogTerm, len(args.Entries))
if args.Term < rf.currentTerm {
reply.Term = rf.currentTerm
reply.Success = false
return
}
// If RPC request or response contains term T > currentTerm:
// set currentTerm = T, convert to follower (§5.1)
if args.Term > rf.currentTerm || rf.state == Candidate {
rf.convertToFollower(args.Term)
}
// new code in Lab 2B
// Reply false if log doesn’t contain an entry at prevLogIndex
// whose term matches prevLogTerm (§5.3)
if args.PrevLogIndex >= len(rf.log) || (args.PrevLogIndex >= 0 && rf.log[args.PrevLogIndex].Term != args.PrevLogTerm) {
reply.Term = rf.currentTerm
reply.Success = false
reply.XLen = len(rf.log)
if args.PrevLogIndex >= 0 && args.PrevLogIndex < len(rf.log) {
reply.XTerm = rf.log[args.PrevLogIndex].Term
for i := args.PrevLogIndex; i >= 0; i-- {
if rf.log[i].Term == reply.XTerm {
reply.XIndex = i
} else {
break
}
}
}
return
}
// If an existing entry conflicts with a new one (same index
// but different terms), delete the existing entry and all that
// follow it (§5.3)
misMatchIndex := -1
for i := range args.Entries {
if args.PrevLogIndex+1+i >= len(rf.log) || rf.log[args.PrevLogIndex+1+i].Term != args.Entries[i].Term {
misMatchIndex = i
break
}
}
// Append any new entries not already in the log
if misMatchIndex != -1 {
rf.log = append(rf.log[:args.PrevLogIndex+1+misMatchIndex], args.Entries[misMatchIndex:]...)
}
// If leaderCommit > commitIndex, set commitIndex =
// min(leaderCommit, index of last new entry)
if args.LeaderCommit > rf.commitIndex {
newEntryIndex := len(rf.log) - 1
if args.LeaderCommit >= newEntryIndex {
rf.commitIndex = newEntryIndex
} else {
rf.commitIndex = args.LeaderCommit
}
DPrintf("[%d]: commit index [%d]", rf.me, rf.commitIndex)
rf.sendApplyMsg()
}
rf.lastReceive = time.Now().Unix()
reply.Term = rf.currentTerm
reply.Success = true
return
}
复制代码在处理AppendEntry RPC的代码中，我们新增了日志匹配的逻辑。如果日志在prevLogIndex处不包含term为prevLogTerm的日志项，那么返回false。这里有两层意思，一个是接收者的日志没有index为prevLogIndex的日志项，另一个是有对应index的日志项但是term不匹配。同时，根据上面所说的快速回退机制，额外返回XLen、XTerm和XIndex。
此外还要注意prevLogIndex可能为-1，意味着日志全都没有匹配上，或者leader此刻还没有日志，此时接收者就要完全服从。
接下来是PreLogIndex与PrevLogTerm匹配到的情况，还要额外检查新同步过来的日志和已存在的日志是否存在冲突。如果一个已经存在的日志项和新的日志项冲突（相同index但是不同term），那么要删除这个冲突的日志项及其往后的日志，并将新的日志项追加到日志中。这里要注意的一个容易出错的地方是不先进行检查，将全部新日志直接追加到了已有日志上。 这样做一旦有旧的AppendEntry RPC到来，RPC的args.Entries的日志项是旧的，一旦直接把args.Entries追加到日志中，就会出现新数据丢失的不安全问题。
最后，根据论文，如果leaderCommit > commitIndex，说明follower的commitIndex也需要更新。为了防止越界，commitIndex取min(leaderCommit, index of last new entry)。
日志Apply
我们单独使用一个goroutine(appMsgApplier)，负责不断将已经被提交的日志项返回给上层应用。
Leader在将日志项复制到多数派后更新commitIndex的同时，要调用sendApplyMsg()。Follower在AppendEntry RPC收到LeaderCommit的更新时，也要调用sendApplyMsg()。
sendApplyMsg()改变rf.moreApply为true，示意有更多的日志项已经被提交，可以apply，并使用applyCond广播通知appMsgApplier。
func (rf *Raft) sendApplyMsg() {
rf.moreApply = true
rf.applyCond.Broadcast()
}
func (rf *Raft) appMsgApplier() {
for {
rf.mu.Lock()
for !rf.moreApply {
rf.applyCond.Wait()
}
commitIndex := rf.commitIndex
lastApplied := rf.lastApplied
entries := rf.log
rf.moreApply = false
rf.mu.Unlock()
for i := lastApplied + 1; i <= commitIndex; i++ {
msg := ApplyMsg{true, entries[i].Command, i + 1}
DPrintf("[%d]: apply index %d - 1", rf.me, msg.CommandIndex)
rf.applyCh <- msg
rf.mu.Lock()
rf.lastApplied = i
rf.mu.Unlock()
}
}
}
复制代码appMsgApplier在for循环中，如果没有需要apply的新日志项，则不断rf.applyCond.Wait()等待通知。否则，由于应用消费日志项是一个耗时的过程，我们要快速释放锁，主要先将commitIndex拷贝，moreApply置为false，意味着目前的日志项apply工作已经接手，随后释放锁。
在接下来i从lastApplied + 1到commitIndex的循环中，我们组装好ApplyMsg，通过applyCh向上层应用提供日志项，在消费后上锁更新lastApplied。此时如果rf.commitIndex又有更新，sendApplyMsg()会被调用，moreApply又会变为true，所以appMsgApplier会在接下来的循环处理新的待apply的日志项。
总结
本文讲解了MIT 6.824 Lab 2B。按照实验要求讲解了选举限制、日志复制、快速回退和日志Apply，其中也有很多自己的感悟和思考，仅供参考。后续将在Lab 2C中继续讲解持久化。
🏆 技术专题第五期 | 聊聊分布式的那些事......
近期评论