6.5840lab3C

 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

func (rf *Raft) persist(snapshot []byte) {
	raftstate := rf.encodeState()
	if snapshot == nil {
		snapshot = rf.persister.ReadSnapshot()
	}
	rf.persister.Save(raftstate, snapshot)
}

// restore previously persisted state.
func (rf *Raft) readPersist(data []byte) {
	rf.mu.Lock()
	defer rf.mu.Unlock()
	DPrintf(dPersist, "readPersist\n")
	// DPrintf("readPersist\n")
	if data == nil || len(data) < 1 {
		return
	}

	r := bytes.NewBuffer(data)
	d := labgob.NewDecoder(r)
	var term int
	var votedFor int
	var logs []LogEntry
	if d.Decode(&term) != nil ||
		d.Decode(&votedFor) != nil ||
		d.Decode(&logs) != nil {
		return
	}
	DPrintf(dPersist, "%v load len of log is %d\n", rf, len(logs))
	rf.currentTerm = term
	rf.votedFor = votedFor
	rf.logs = logs
	rf.lastIncludedIndex = rf.logs[0].Index
	rf.lastIncludedTerm = rf.logs[0].Term
	rf.lastApplied = rf.lastIncludedIndex
	rf.commitIndex = rf.lastIncludedIndex
}

// the service says it has created a snapshot that has
// all info up to and including index. this means the
// service no longer needs the log through (and including)
// that index. Raft should now trim its log as much as possible.
func (rf *Raft) Snapshot(index int, snapshot []byte) {
	// Your code here (3D).
	rf.mu.Lock()
	defer rf.mu.Unlock()

	if rf.killed() || index <= rf.lastIncludedIndex || index > rf.lastApplied {
		return
	}

	newIndex := index - rf.lastIncludedIndex
	rf.lastIncludedTerm = rf.logs[newIndex].Term
	rf.lastIncludedIndex = index

	logs := make([]LogEntry, len(rf.logs[newIndex+1:]))
	copy(logs, rf.logs[newIndex+1:])

	rf.logs = append([]LogEntry{{rf.lastIncludedIndex, rf.lastIncludedTerm, nil}}, logs...)
	rf.persist(snapshot)
}

 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

func (rf *Raft) InstallSnapshot(peerId int) {
	rf.mu.Lock()
	args := SnapshotArgs{
		Term:              rf.currentTerm,
		LeaderId:          rf.me,
		LastIncludedIndex: rf.lastIncludedIndex,
		LastIncludedTerm:  rf.lastIncludedTerm,
		Snapshot:          rf.persister.ReadSnapshot(),
	}
	reply := SnapshotReply{}

	DPrintf(dSnap, "%v send snapshot to S%d\n", rf, peerId)

	rf.mu.Unlock()
	if !rf.sendSnapshot(peerId, &args, &reply) {
		return
	}
	rf.mu.Lock()
	defer rf.mu.Unlock()

	DPrintf(dSnap, "S%d get reply from S%d\n", rf.me, peerId)

	if args.Term != rf.currentTerm || rf.state != LEADER {
		return
	}
	if reply.Term > rf.currentTerm {
		rf.changeState(FOLLOWER)
		rf.currentTerm = reply.Term
		rf.persist(nil)
		return
	}
	rf.nextIndex[peerId] = rf.lastIncludedIndex + 1
	rf.matchIndex[peerId] = rf.lastIncludedIndex
}

func (rf *Raft) HandleSnapshot(args *SnapshotArgs, reply *SnapshotReply) {
	rf.mu.Lock()

	reply.Term = rf.currentTerm
	if args.Term < rf.currentTerm {
		rf.mu.Unlock()
		return
	}

	if args.Term > rf.currentTerm {
		rf.changeState(FOLLOWER)
		rf.currentTerm = args.Term
		rf.persist(nil)
		reply.Term = rf.currentTerm
	}

	rf.resetElectionTimer()

	if args.LastIncludedIndex <= rf.lastIncludedIndex || args.LastIncludedIndex <= rf.commitIndex {
		rf.mu.Unlock()
		return
	}

	DPrintf(dSnap, "%v HandleSnapshot, leaderID: %d, LastIncludedIndex: %d, LastIncludedTerm: %d\n", rf, args.LeaderId, args.LastIncludedIndex, args.LastIncludedTerm)
	index := args.LastIncludedIndex - rf.lastIncludedIndex

	if index < len(rf.logs) && rf.logs[index].Index == args.LastIncludedIndex && rf.logs[index].Term == args.LastIncludedTerm {
		logs := make([]LogEntry, len(rf.logs[index+1:]))
		copy(logs, rf.logs[index+1:])
		rf.logs = append([]LogEntry{{args.LastIncludedIndex, args.LastIncludedTerm, nil}}, logs...)
	} else {
		rf.logs = []LogEntry{{args.LastIncludedIndex, args.LastIncludedTerm, nil}}
	}

	rf.lastIncludedIndex = args.LastIncludedIndex
	rf.lastIncludedTerm = args.LastIncludedTerm
	rf.commitIndex = args.LastIncludedIndex
	rf.lastApplied = args.LastIncludedIndex

	rf.persist(args.Snapshot)
	rf.mu.Unlock()

	applyMsg := ApplyMsg{
		SnapshotValid: true,
		Snapshot:      args.Snapshot,
		SnapshotTerm:  args.LastIncludedTerm,
		SnapshotIndex: args.LastIncludedIndex,
	}

	rf.applyCh <- applyMsg
}