6.5840lab3A raft leader election

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

type State int

const (
	LEADER State = iota
	FOLLOWER
	CANDIDATE
)

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 (3A, 3B, 3C).
	// Look at the paper's Figure 2 for a description of what
	// state a Raft server must maintain.
	state           State       // rf当前的状态
	heartbeatTimer  *time.Timer // 心跳timer
	electionTimeout *time.Timer // 超时选举 timer
	currentTerm     int         //当前的任期
	votedFor        int         // 投票给了哪个节点
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

type RequestVoteArgs struct {
	Term         int // Candidate的任期
	CandidateId  int // Candidate的ID
	LastLogIndex int // Candidate最后日志的下标
	LastLogTerm  int // Candidate最后日志的任期
}

type RequestVoteReply struct {
	Term        int  // FOLLOWER的任期
	VoteGranted bool // FOLLOWER是否投票给Candidate
}

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

func (rf *Raft) ticker() {
	for rf.killed() == false {
		select {
		case <-rf.electionTimeout.C:
			rf.mu.Lock()
			if rf.state != LEADER {
				rf.mu.Unlock()
				rf.startElection()
				rf.resetElectionTimer()
			} else {
				rf.mu.Unlock()
			}
		case <-rf.heartbeatTimer.C:
			rf.mu.Lock()
			if rf.state == LEADER {
				rf.requestAppendEntries()
				rf.resetHeartBeatTimer()
			}
			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

func (rf *Raft) startElection() {
	rf.mu.Lock()
	rf.changeState(CANDIDATE)
	rf.persist()
	lastLogIndex, lastLogTerm := rf.getLastLog()
	args := RequestVoteArgs{
		Term:         rf.currentTerm,
		CandidateId:  rf.me,
		LastLogIndex: lastLogIndex,
		LastLogTerm:  lastLogTerm,
	}
	nPeers := len(rf.peers)
	cnt := 1
	DPrintf("%v starts election (lastLogIndex=%d, lastLogTerm=%d)",
		rf, lastLogIndex, lastLogTerm)
	rf.mu.Unlock()

	for peerId := range rf.peers {
		if peerId == rf.me {
			continue
		}
		go func(peerId int) {
			reply := RequestVoteReply{}
			if !rf.sendRequestVote(peerId, &args, &reply) {
				return
			}

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

			if args.Term != rf.currentTerm {
				return
			}

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

			if reply.VoteGranted && rf.state == CANDIDATE {
				cnt += 1
				if cnt > nPeers/2 {
					rf.changeState(LEADER)
					rf.reinitIndex()
					DPrintf("%v becomes LEADER", rf)
					rf.requestAppendEntries()
					return
				}
			}
		}(peerId)
	}
}

 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

func (rf *Raft) RequestVote(args *RequestVoteArgs, reply *RequestVoteReply) {
	rf.mu.Lock()
	defer rf.mu.Unlock()

	reply.Term = rf.currentTerm
	reply.VoteGranted = false

	if args.Term < rf.currentTerm {
		return
	}

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

	lastLogIndex, lastLogTerm := rf.getLastLog()
	if (args.LastLogTerm < lastLogTerm) || (args.LastLogTerm == lastLogTerm && args.LastLogIndex < lastLogIndex) {
		return
	}

	if rf.votedFor == -1 || rf.votedFor == args.CandidateId {
		rf.votedFor = args.CandidateId
		rf.persist()
		reply.VoteGranted = true
		rf.resetElectionTimer()
	}

}