Raft (MIT 6824 Lab 2B) 🏆 技术

发表于

引言

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，随后要将其正确地返回给上层应用执行。

数据结构

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
go复制代码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。
1
2
3
4
5
6
7
8
9
10
11
12
13
go复制代码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)的判断。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
go复制代码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，可能为空。

主要函数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
go复制代码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已提交的日志给各自的上层应用。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
go复制代码//
// 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要在原有基础上加一。

1
2
3
4
5
6
7
8
9
go复制代码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」的选举限制。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
go复制代码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额外新增了哪些逻辑：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
go复制代码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
}

根据论文中的选举限制，我们在投票时要额外判断：

  1. 是否没投票或者投给的是这个candidate。
  2. 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前的日志一致了。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
go复制代码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通信。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
go复制代码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。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
go复制代码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中继续讲解持久化。
🏆 技术专题第五期 | 聊聊分布式的那些事……

参考文献

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

0%