MIT6.824 2022 Raft Lab2B Log Replication

1
2
3
4
5

type Entry struct {
	Index   int
	Term    int
	Command interface{}
}

 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

type AppendEntriesReq struct {
	Term          int
	LeaderId      int
	PrevLogIndex  int
	PrevLogTerm   int
	LeaderComment int
	Entries       []Entry
}

func (req AppendEntriesReq) String() string {
	return fmt.Sprintf("{Term: %d, LeaderId: %v, PreVoteLogIndex: %v, PreVoteLogTerm: %v, LeaderComment: %v, Entries: %v}",
		req.Term, req.LeaderId, req.PrevLogIndex, req.PrevLogTerm, req.LeaderComment, req.Entries)
}

type AppendEntriesResp struct {
	Term    int
	Success bool
	// for fast backup https://mit-public-courses-cn-translatio.gitbook.io/mit6-824/lecture-07-raft2/7.3-hui-fu-jia-su-backup-acceleration
	ConflictIndex int
	ConflictTerm  int
	ConflictLen   int
}

func (resp AppendEntriesResp) String() string {
	return fmt.Sprintf("{Term:%v,Success:%v,ConflictIndex:%v,ConflictTerm:%v}",
		resp.Term, resp.Success, resp.ConflictIndex, resp.ConflictTerm)
}

 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

func (rf *Raft) AppendEntries(req *AppendEntriesReq, resp *AppendEntriesResp) {
	rf.mu.Lock()
	defer rf.mu.Unlock()
	defer rf.persist()
	defer DPrintf("[AppendEntries]- {Node: %v}'s state is {state %v, term %v, commitIndex %v, lastApplied %v, firstLog %v, lastLog %v} before processing AppendEntriesRequest %v and reply AppendEntries %v",
		rf.me, rf.state, rf.currentTerm, rf.commitIndex, rf.lastApplied, rf.getFirstLog(), rf.getLastLog(), req, resp)

	// 如果发现来自leader的rpc中的term比当前peer要小,
	// 说明是该RPC 来自旧的term(leader)，|| 或者 当前leader 需要更新 不处理
	if req.Term < rf.currentTerm {
		resp.Term, resp.Success = rf.currentTerm, false
		return
	}

	// 一般来讲,在vote的时候已经将currentTerm和leader同步
	// 不过,有些peer暂时的掉线或者其他一些情况重连以后,会发现term和leader不一样
	// 以收到大于自己的term的rpc也是第一时间同步.而且要将votefor重新设置为-1
	// 等待将来选举 (说明这个peer 不是之前election 中投的的marjority)
	if req.Term > rf.currentTerm {
		rf.currentTerm, rf.votedFor = req.Term, -1
	}

	rf.ChangeState(StateFollower)
	rf.electionTimer.Reset(RandomizedElectionTimeout())

	// PrevLogIndex 比rf 当前的第一个Log index 还要小
	if req.PrevLogIndex < rf.getFirstLog().Index {
		resp.Term, resp.Success = 0, false
		DPrintf("[AppendEntries] - {Node: %v} receives unexpected AppendEntriesRequest %v from {Node: %v} because prevLogIndex %v < firstLogIndex %v",
			rf.me, req, req.LeaderId, req.PrevLogIndex, rf.getFirstLog().Index)
		return
	}

	if !rf.matchLog(req.PrevLogTerm, req.PrevLogIndex) {
		// 日志的一致性检查失败后，递归找到需要追加日志的位置
		resp.Term, resp.Success = rf.currentTerm, false
		lastIndex := rf.getLastLog().Index

		if lastIndex < req.PrevLogIndex {
			// lastIndex 和 nextIndex[peer] 之间有空洞 scenario3
			// follower 在nextIndex[peer] 没有log
			resp.ConflictTerm = -1
			resp.ConflictIndex = lastIndex + 1
		} else {
			// scenario2, 1
			// 以任期为单位进行回退
			firstIndex := rf.getFirstLog().Index
			resp.ConflictTerm = rf.logs[req.PrevLogIndex-firstIndex].Term
			index := req.PrevLogIndex - 1
			for index >= firstIndex && rf.logs[index-firstIndex].Term == resp.ConflictTerm {
				index--
			}
			resp.ConflictIndex = index
		}
		return
	}

	firstIndex := rf.getFirstLog().Index
	for i, entry := range req.Entries {
		// mergeLog
		// 添加的日志索引位置 比Follower 日志相同 直接添加 此处用大于等于，实际只有==
		// || 要添加的日志索引位置在 Follower 中的任期和AE RPC 中的Term 冲突
		if entry.Index-firstIndex >= len(rf.logs) || rf.logs[entry.Index-firstIndex].Term != entry.Term {
			rf.logs = shrinkEntriesArray(append(rf.logs[:entry.Index-firstIndex], req.Entries[i:]...))
			break
		}
	}

	rf.advanceCommitIndexForFollower(req.LeaderComment)

	resp.Term, resp.Success = rf.currentTerm, true
}

 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

// handleAppendEntriesResponse peer handle AppendEntries RPC
func (rf *Raft) handleAppendEntriesResponse(peer int, req *AppendEntriesReq, resp *AppendEntriesResp) {
	defer DPrintf("[handleAppendEntriesResponse]-{Node %v}'s state is {state %v,term %v,commitIndex %v,lastApplied %v,firstLog %v,lastLog %v} after handling AppendEntriesResponse %v for AppendEntriesRequest %v",
		rf.me, rf.state, rf.currentTerm, rf.commitIndex, rf.lastApplied, rf.getFirstLog(), rf.getLastLog(), resp, req)

	if rf.state == StateLeader && rf.currentTerm == req.Term {
		if resp.Success {
			// 更新matchIndex, nextIndex
			rf.matchIndex[peer] = req.PrevLogIndex + len(req.Entries)
			rf.nextIndex[peer] = rf.matchIndex[peer] + 1
			rf.advanceCommitIndexForLeader()
		} else {
			// term 太小而失败
			if resp.Term > rf.currentTerm {
				rf.ChangeState(StateFollower)
				rf.currentTerm, rf.votedFor = resp.Term, -1
				rf.persist()
			} else if resp.Term == rf.currentTerm { // 日志不匹配而失败
				rf.nextIndex[peer] = resp.ConflictIndex
				// 1. 如果在Leader 中能找到和Follower 有相同的ConflictTerm，
				// 返回该Leader Term 的最后一个Log 作为nextIndex[peer]
				// 2. 如果找不到相同的Term，返回Follower 中的ConflictTerm 的第一个日志，即ConflictIndex
				if resp.ConflictTerm != -1 {
					firstIndex := rf.getFirstLog().Index
					for i := req.PrevLogIndex; i >= firstIndex; i-- {
						if rf.logs[i-firstIndex].Term == resp.ConflictTerm {
							rf.nextIndex[peer] = i
							break
						}
					}
				}
			}
		}
	}
}

func (rf *Raft) replicator(peer int) {
	rf.replicatorCond[peer].L.Lock()
	defer rf.replicatorCond[peer].L.Unlock()

	for !rf.killed() {
		// if there is no need to replicate entries for this peer,
		// just release CPU and wait other goroutine's signal if service adds new Command
		// if this peer needs replicating entries, this goroutine will call
		// replicateOneRound(peer) multiple times until this peer catches up, and then wait
		// Only Leader 可以Invoke 这个方法，通过.Singal 唤醒各个peer, 不是Leader 不生效
		for !rf.needReplicating(peer) {
			rf.replicatorCond[peer].Wait()
		}
		// maybe a pipeline mechanism is better to trade-off the memory usage and catch up time
		rf.replicateOneRound(peer)
	}
}

 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

// applier a dedicated applier goroutine to guarantee that each log will be push into
// applyCh exactly once, ensuring that service's applying entries and raft's
// committing entries can be parallel
func (rf *Raft) applier() {
	for !rf.killed() {
		rf.mu.Lock()
		// if there is no need to apply entries,
		// just release CPU and wait other goroutine's signal if they commit new entries
		for rf.lastApplied >= rf.commitIndex {
			rf.applyCond.Wait()
		}

		firstIndex, commitIndex, lastApplied := rf.getFirstLog().Index, rf.commitIndex, rf.lastApplied
		entries := make([]Entry, commitIndex-lastApplied)
		copy(entries, rf.logs[lastApplied+1-firstIndex:commitIndex+1-firstIndex])
		rf.mu.Unlock()

		for _, entry := range entries {
			rf.applyCh <- ApplyMsg{
				CommandValid: true,
				Command:      entry.Command,
				CommandTerm:  entry.Term,
				CommandIndex: entry.Index,
			}
		}

		rf.mu.Lock()
		DPrintf("{Node %v} applies entries %v-%v in term %v",
			rf.me, rf.lastApplied, commitIndex, rf.currentTerm)

		rf.lastApplied = Max(rf.lastApplied, commitIndex)
		rf.mu.Unlock()
	}
}

 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

func (rf *Raft) AppendEntries(req *AppendEntriesReq, resp *AppendEntriesResp) {
	rf.mu.Lock()
	defer rf.mu.Unlock()
	defer rf.persist()
	defer DPrintf("[AppendEntries]- {Node: %v}'s state is {state %v, term %v, commitIndex %v, lastApplied %v, firstLog %v, lastLog %v} before processing AppendEntriesRequest %v and reply AppendEntries %v",
		rf.me, rf.state, rf.currentTerm, rf.commitIndex, rf.lastApplied, rf.getFirstLog(), rf.getLastLog(), req, resp)

	// 如果发现来自leader的rpc中的term比当前peer要小,
	// 说明是该RPC 来自旧的term(leader)，|| 或者 当前leader 需要更新 不处理
	if req.Term < rf.currentTerm {
		resp.Term, resp.Success = rf.currentTerm, false
		return
	}

	// 一般来讲,在vote的时候已经将currentTerm和leader同步
	// 不过,有些peer暂时的掉线或者其他一些情况重连以后,会发现term和leader不一样
	// 以收到大于自己的term的rpc也是第一时间同步.而且要将votefor重新设置为-1
	// 等待将来选举 (说明这个peer 不是之前election 中投的的marjority)
	if req.Term > rf.currentTerm {
		rf.currentTerm, rf.votedFor = req.Term, -1
	}

	rf.ChangeState(StateFollower)
	rf.electionTimer.Reset(RandomizedElectionTimeout())

	// PrevLogIndex 比rf 当前的第一个Log index 还要小
	if req.PrevLogIndex < rf.getFirstLog().Index {
		resp.Term, resp.Success = 0, false
		DPrintf("[AppendEntries] - {Node: %v} receives unexpected AppendEntriesRequest %v from {Node: %v} because prevLogIndex %v < firstLogIndex %v",
			rf.me, req, req.LeaderId, req.PrevLogIndex, rf.getFirstLog().Index)
		return
	}

	if !rf.matchLog(req.PrevLogTerm, req.PrevLogIndex) {
		// 日志的一致性检查失败后，递归找到需要追加日志的位置
		resp.Term, resp.Success = rf.currentTerm, false
		lastIndex := rf.getLastLog().Index

		if lastIndex < req.PrevLogIndex {
			// lastIndex 和 nextIndex[peer] 之间有空洞 scenario3
			// follower 在nextIndex[peer] 没有log
			resp.ConflictTerm = -1
			resp.ConflictIndex = lastIndex + 1
		} else {
			// scenario2, 1
			// 以任期为单位进行回退
			firstIndex := rf.getFirstLog().Index
			resp.ConflictTerm = rf.logs[req.PrevLogIndex-firstIndex].Term
			index := req.PrevLogIndex - 1
			for index >= firstIndex && rf.logs[index-firstIndex].Term == resp.ConflictTerm {
				index--
			}
			resp.ConflictIndex = index
		}
		return
	}

	firstIndex := rf.getFirstLog().Index
	for i, entry := range req.Entries {
		// mergeLog
		// 添加的日志索引位置 比Follower 日志相同 直接添加 此处用大于等于，实际只有==
		// || 要添加的日志索引位置在 Follower 中的任期和AE RPC 中的Term 冲突
		if entry.Index-firstIndex >= len(rf.logs) || rf.logs[entry.Index-firstIndex].Term != entry.Term {
			rf.logs = shrinkEntriesArray(append(rf.logs[:entry.Index-firstIndex], req.Entries[i:]...))
			break
		}
	}

	rf.advanceCommitIndexForFollower(req.LeaderComment)

	resp.Term, resp.Success = rf.currentTerm, true
}