Raft 领导人选举实现

// 检查是否确实仅存在一个领导人。
// 为了测试重新选举，所以会多尝试几次查找。
func (cfg *config) checkOneLeader() int {
	// 多迭代几次尝试寻找领导人。
	for iters := 0; iters < 10; iters++ {
		ms := 450 + (rand.Int63() % 100)
		// 随机尝试。
		time.Sleep(time.Duration(ms) * time.Millisecond)
		// 获取领导人和对应上任 Term。
		leaders := make(map[int][]int)
		for i := 0; i < cfg.n; i++ {
			if cfg.connected[i] {
				if term, leader := cfg.rafts[i].GetState(); leader {
					leaders[term] = append(leaders[term], i)
				}
			}
		}
		// 校验没有在一个周期内选出两个领导人，找到最后新的Term。
		lastTermWithLeader := -1
		for term, leaders := range leaders {
			if len(leaders) > 1 {
				cfg.t.Fatalf("term %d has %d (>1) leaders", term, len(leaders))
			}
			if term > lastTermWithLeader {
				lastTermWithLeader = term
			}
		}
		// 找到最新的领导人。
		if len(leaders) != 0 {
			return leaders[lastTermWithLeader][0]
		}
	}
	cfg.t.Fatalf("expected one leader, got none")
	return -1
}

func Make(peers []*labrpc.ClientEnd, me int,
	persister *Persister, applyCh chan ApplyMsg) *Raft {
	rf := &Raft{}
	rf.peers = peers
	rf.persister = persister
	rf.me = me
	// 代表无领导人。
	rf.leaderId = -1
	rf.currentTerm = 0
	// 代表还未投票。
	rf.votedFor = -1
	rf.commitIndex = 0
	rf.lastAppliedIndex = 0
	rf.state = Follower
	rf.lastReceiveTime = time.Now()
	// 开始选举。
	go rf.startLeaderElection()
	rf.readPersist(persister.ReadRaftState())
	return rf
}

// 下面附带一些会用到的一些角色转换的函数。
func (rf *Raft) convertToCandidate() {
	rf.state = Candidate
	rf.currentTerm++
	rf.votedFor = rf.me
	rf.lastReceiveTime = time.Now()
}

func (rf *Raft) convertToFollower(newTerm int) {
	rf.state = Follower
	rf.currentTerm = newTerm
	rf.votedFor = -1
	rf.lastReceiveTime = time.Now()
}

func (rf *Raft) convertToLeader() {
	rf.state = Leader
	rf.leaderId = rf.me
	rf.lastReceiveTime = time.Now()
}

// 开始选举领导人。
// 当发生一次选举超时时，节点就开始一轮新的选举。
func (rf *Raft) startLeaderElection() {
	for {
		electionTimeout := rand.Intn(150)
		startTime := time.Now()
		time.Sleep(time.Duration(HeartbeatInterval+electionTimeout) * time.Millisecond)
		rf.mu.Lock()
		if atomic.LoadInt32(&rf.dead) == Dead {
			rf.mu.Unlock()
			return
		}
		// 在这个地方判断是否应该发起选举。
		if rf.lastReceiveTime.Before(startTime) {
			if rf.state != Leader {
				DPrintf("%d kicks off election on term: %d", rf.me, rf.currentTerm)
				go rf.kickOffElection()
			}
		}
		rf.mu.Unlock()
	}
}

// 触发选举，向所有节点发出投票请求。
func (rf *Raft) kickOffElection() {
	rf.mu.Lock()
	// 将自己转换为候选人。
	rf.convertToCandidate()
	lastLogEntry := rf.getLastLogEntry()
	args := RequestVoteArgs{
		Term:         rf.currentTerm,
		CandidateId:  rf.me,
		LastLogIndex: lastLogEntry.Index,
		LastLogTerm:  lastLogEntry.Term,
	}
	numVote := 1
	rf.mu.Unlock()
	// 开始向每个节点发送投票请求。
	for i := 0; i < len(rf.peers); i++ {
		if i != rf.me {
			go func(peerId int) {
				reply := RequestVoteReply{}
				DPrintf("%d send vote request to %d", rf.me, peerId)
				// 发出 RPC 消息，注意消息和回复字段要求大写。
				ok := rf.sendRequestVoteRPC(peerId, &args, &reply)
				if !ok {
					return
				}
				rf.mu.Lock()
				defer rf.mu.Unlock()
				// 如果投票人的 Term 比当前节点的 Term 大，那将节点转换为 Follower。
				if reply.Term > rf.currentTerm {
					rf.convertToFollower(reply.Term)
					return
				}
				// 收到了有效投票，统计结果，如果获取系统中一半以上的选票，就将该节点转换为领导人，并且开始放发出心跳维持自己的统治。
				if reply.VoteGranted {
					numVote++
					if numVote > len(rf.peers)/2 && rf.state == Candidate {
						rf.convertToLeader()
						DPrintf("%d become the leader on term: %d", rf.me, rf.currentTerm)
						for j := 0; j < len(rf.peers); j++ {
							if j != rf.me {
								go rf.replicaLogToPeer(j)
							}
						}
					}
				}
			}(i)
		}
	}
}

//
// 投票请求 RPC 处理.
//
func (rf *Raft) RequestVote(args *RequestVoteArgs, reply *RequestVoteReply) {
	rf.mu.Lock()
	defer rf.mu.Unlock()
	DPrintf("%d start process vote request", rf.me)
	// 设置回复消息的 Term。
	reply.Term = rf.currentTerm
	// 对应我上面说的关于 Term 的那一条规则。
	if args.Term < rf.currentTerm {
		reply.VoteGranted = false
		DPrintf("%d refuse vote for %d", rf.me, args.CandidateId)
	} else {
		// 如果候选人的 Term 比我们的大，那我们应该将自己转换为 Follower。
		if args.Term > rf.currentTerm {
			rf.convertToFollower(args.Term)
		}
		lastEntry := rf.getLastLogEntry()
		// 对应我上面说的关于日志的那一条规则。
		logUpToDate := LogEntry{Command: nil, Index: args.LastLogIndex, Term: args.LastLogTerm}.isMoreUpToDate(lastEntry)
		if rf.votedFor == -1 && logUpToDate {
			rf.votedFor = args.CandidateId
			reply.VoteGranted = true
			DPrintf("%d vote for %d", rf.me, args.CandidateId)
		}
	}
}

// 是否更新更长.
func (l LogEntry) isMoreUpToDate(r LogEntry) bool {
	return (l.Term == r.Term && l.Index >= r.Index) || l.Term > r.Term
}

// 发送日志到附加节点，还是采用 for 循环的形式。
func (rf *Raft) replicaLogToPeer(peerId int) {
	DPrintf("%d start sync log to %d", rf.me, peerId)
	for {
		rf.mu.Lock()
		if rf.state != Leader {
			rf.mu.Unlock()
			DPrintf("%d stop sends append entries to %d", rf.me, peerId)
			return
		}
		rf.mu.Unlock()
		go rf.sendAppendEntry(peerId)
		// 设置一定心跳间隔。
		time.Sleep(HeartbeatInterval * time.Millisecond)
	}
}

// 发送日志给节点.
func (rf *Raft) sendAppendEntry(peerId int) {
	rf.mu.Lock()
	lastLogEntry := rf.getLastLogEntry()
	args := AppendEntriesArgs{
		Term:         rf.currentTerm, // 我们需要填好这个值，在其他节点处理该请求的时候会用到这个值。
		LeaderId:     rf.me, // 设置自己为的编号。
		PrevLogIndex: lastLogEntry.Index, // 这个值对我们目前实验无影响。
		Entries:      nil, // 表示心跳。
		LeaderCommit: -1, // 这个值对我们目前实验无影响，瞎填就可以.
	}
	reply := AppendEntriesReply{}
	rf.mu.Unlock()
	// 发送日志 RPC。
	ok := rf.sendAppendEntryRPC(peerId, &args, &reply)
	if !ok {
		log.Printf("%d send a append PRC to %d failed", rf.me, peerId)
		return
	}
	rf.mu.Lock()
	defer rf.mu.Unlock()
	// 如果发现回复的 Term 比我们自己的高，我们就将自己转换为 Follower。
	if reply.Term > rf.currentTerm {
		rf.convertToFollower(reply.Term)
		return
	}
	log.Printf("%d success send a append to %d", rf.me, peerId)
}

//
// 附加日志 RPC 处理.
//
func (rf *Raft) AppendEntries(args *AppendEntriesArgs, reply *AppendEntriesReply) {
	rf.mu.Lock()
	defer rf.mu.Unlock()
	log.Printf("%d start process append entries request from %d, args term: %d node term: %d", rf.me, args.LeaderId, args.Term, rf.currentTerm)
	// 如果 Term 比改节点的小，那我们就认为这是一条无效的心跳，直接返回。
	if args.Term < rf.currentTerm {
		DPrintf("get append entries form %d, but the term less than %d", args.LeaderId, rf.me)
		reply.Term, reply.Success = rf.currentTerm, false
		return
	}
	// 刷新该节点的 lastReceiveTime.
	reply.Term = rf.currentTerm
	rf.leaderId = args.LeaderId
	rf.lastReceiveTime = time.Now()

	if args.Term > rf.currentTerm {
		DPrintf("%d convert itself as follower", rf.me)
		rf.convertToFollower(args.Term)
	}
}