Raft 实验 Part B - MIT 6.824 Lab 2

1
2
3
4
5

type LogEntry struct {
	Command interface{} // command for state machine
	Term    int         // when entry is received by leader
	Index   int         // index in the LogEntry
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

func Make(peers []*labrpc.ClientEnd, me int,
	persister *Persister, applyCh chan ApplyMsg) *Raft {
	// ...
	rf.applyCh = applyCh
	rf.nextIndex = make([]int, len(peers))
	rf.matchIndex = make([]int, len(peers))

	go rf.LeaderElection()
	//...
	return rf
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

func (rf *Raft) Start(command interface{}) (int, int, bool) {
	// ... Remember to Lock()
	term = rf.currentTerm
	isLeader = rf.state == Leader

	if isLeader {
		rf.log = append(rf.log, LogEntry{Command: command, Term: term, Index: len(rf.log) + 1})
		index = len(rf.log)
		rf.nextIndex[rf.me] = index + 1
		rf.matchIndex[rf.me] = index
	}
	return index, term, isLeader
}

 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

func (rf *Raft) AppendEntries(args *AppendEntriesArgs, reply *AppendEntriesReply) {
	rf.mu.Lock()
	defer rf.mu.Unlock()

	rf.lastReceive = time.Now()
	reply.Term = rf.currentTerm

	// 1. Reply false if term < currentTerm (§5.1)
	// Switch to Follower if newer term found

	// 2. Reply false if log doesn't contain an entry at prevLogIndex
	// whose term matches prevLogTerm (§5.3)

	// rf.log doesn't contain entry at PrevLogIndex
	// reply false and return

	// 3. 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)
	
	// term at PrevLogIndex disagree with PrevLogTerm
	// reply false and return

	// 4. Append any new entries not already in the log

	// 5. If leaderCommit > commitIndex, set commitIndex =
	// min(leaderCommit, index of last new entry)
	
	// after updating commitIndex, send ApplyMsg if commitIndex > lastApplied
	// update lastApplied
	
	reply.Success = 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

func (rf *Raft) SendHeartbeat(server int) {
	rf.mu.Lock()

	var prevLog LogEntry
	if rf.nextIndex[server] > 1 {
		prevLog = rf.log[(rf.nextIndex[server]-1)-1]
	}
	entries := make([]LogEntry, len(rf.log)-prevLog.Index)
	copy(entries, rf.log[prevLog.Index:])

	args := AppendEntriesArgs{
		Term:         rf.currentTerm,
		LeaderId:     rf.me,
		PrevLogIndex: prevLog.Index,
		PrevLogTerm:  prevLog.Term,
		Entries:      entries,
		LeaderCommit: rf.commitIndex,
	}
	reply := AppendEntriesReply{}

	rf.mu.Unlock()

	ok := rf.sendAppendEntries(server, &args, &reply)

	rf.mu.Lock()
	defer rf.mu.Unlock()

	rf.lastReceive = time.Now()

	// Convert to Follower if reply.Term is newer

	// 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)
	if reply.Success {
		// Update nextIndex and matchIndex
		// ...

		// 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).
	} else {
		// decrement nextIndex
		rf.nextIndex[server] = prevLog.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

func (rf *Raft) sendApplyMsg() {
	// If commitIndex > lastApplied: increment lastApplied, apply
	// log[lastApplied] to state machine (§5.3)
	if rf.commitIndex > rf.lastApplied {
		entries := append([]LogEntry{}, rf.log[rf.lastApplied:rf.commitIndex]...)

		go func(entries []LogEntry) {
			for _, entry := range entries {
				msg := ApplyMsg{
					CommandValid: true,
					Command:      entry.Command,
					CommandIndex: entry.Index,
				}
				rf.applyCh <- msg

				rf.mu.Lock()
				if rf.lastApplied < msg.CommandIndex {
					rf.lastApplied = msg.CommandIndex
				}
				rf.mu.Unlock()
			}
		}(entries)
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

➜  raft git:(lab2) ✗ go test -run 2B
Test (2B): basic agreement ...
  ... Passed --   1.0  3   16    4290    3
Test (2B): RPC byte count ...
  ... Passed --   2.6  3   48  113518   11
Test (2B): agreement despite follower disconnection ...
  ... Passed --   5.5  3   96   25251    7
Test (2B): no agreement if too many followers disconnect ...
  ... Passed --   3.7  5  168   38812    3
Test (2B): concurrent Start()s ...
  ... Passed --   0.8  3   14    3786    6
Test (2B): rejoin of partitioned leader ...
  ... Passed --   4.6  3  140   34273    4
Test (2B): leader backs up quickly over incorrect follower logs ...
  ... Passed --  38.5  5 2634 2335471  106
Test (2B): RPC counts aren't too high ...
  ... Passed --   2.2  3   40   11214   12
PASS