Lab2A. Raft 选主实现

type Raft struct {
	mu        sync.Mutex          // 共享锁
	peers     []*labrpc.ClientEnd // 集群中的全部节点
	persister *Persister          // 持久化工具
	me        int                 // 本节点在 peers 中的索引

	curTerm  int           // 节点目前的任期号
	votedFor int           // 节点目前的投票对象
	entries  []LogEntry    // 本地日志
	state    PeerState     // 节点状态
	timer    *RaftTimer    // 选举超时定时器
	entryCh  chan LogEntry // 日志处理 channel
}

// 投票参数
type RequestVoteArgs struct {
	Term        int // 候选人的任期号
	CandidateId int // 候选人 id
}

// 响应投票
type RequestVoteReply struct {
	Term        int  // 选民节点的任期号
	VoteGranted bool // 是否赢得该选票
}

// 候选人发起投票
func (rf *Raft) vote() {
	rf.curTerm++
	rf.state = Candidate
	rf.votedFor = rf.me

	args := RequestVoteArgs{
		Term:        rf.curTerm,
		CandidateId: rf.me,
	}
	replyCh := make(chan RequestVoteReply, len(rf.peers))
	var wg sync.WaitGroup
	for i := range rf.peers {
		if i == rf.me {
			continue
		}

		wg.Add(1)
		go func(server int) {
			defer wg.Done()
			var reply RequestVoteReply
			if succ := rf.sendRequestVote(server, &args, &reply); !succ {
				return
			}
			replyCh <- reply
		}(i)
	}
	go func() {
		wg.Wait()
		close(replyCh) // 避免资源泄漏
	}()

	votes := 1
	targetVotes := len(rf.peers)/2 + 1
	for reply := range replyCh {
		// 已有更新 leader，回退到 follower
		if reply.Term > rf.curTerm {
			rf.back2Follower(reply.Term)
			return
		}
		if reply.VoteGranted {
			votes++
		}

		// 如果选票已过半，不再等待已 crash 的节点调用超时
		if votes >= targetVotes {
			break
		}
	}

	// 因 split vote 等原因未达到多数票
	if votes < len(rf.peers)/2+1 {
		rf.resetElectTimer()
		return
	}

	// 成功当选，立刻发送心跳
	rf.state = Leader
	go rf.heartbeat()
}

func (rf *Raft) RequestVote(args *RequestVoteArgs, reply *RequestVoteReply) {
	reply.Voter = rf.me
	reply.Term = rf.curTerm

	switch {
	case args.Term < rf.curTerm: // 拒绝处理
		reply.VoteGranted = false
		return
	case args.Term == rf.curTerm: // 每个任期只能投一票
		if rf.votedFor == VOTE_NIL || rf.votedFor == args.CandidateId {
			reply.VoteGranted = true
			rf.votedFor = args.CandidateId
			rf.back2Follower(args.Term)
		}
	case args.Term > rf.curTerm: // 直接投票
		reply.VoteGranted = true
		rf.votedFor = args.CandidateId
		rf.back2Follower(args.Term)
	}

	return
}

type LogEntry struct {
	Index   int         // 日志索引号
	Term    int         // 写入日志时节点的任期号
	Command interface{} // 客户端命令
}

// 心跳请求
type AppendEntriesArgs struct {
	Term         int        // leader 任期号
	LeaderId     int        // leader id
	PrevLogIndex int        // 暂时不用
	PrevLogTerm  int        //
	Entries      []LogEntry // 批量日志，心跳时为空
}

// 心跳响应
type AppendEntriesReply struct {
	Term int  // 节点任期号
	Succ bool // 心跳是否成功响应
}

// leader 发送心跳
func (rf *Raft) heartbeat() {
	t := time.NewTicker(HEARTBEAT_INTERVAL) // 100ms
	for {
		if !rf.isLeader() {
			return
		}

		args := AppendEntriesArgs{
			Term:         rf.curTerm,
			LeaderId:     rf.me,
			PrevLogIndex: 0,
			PrevLogTerm:  0,
			Entries:      nil, // 心跳时为空日志
		}
		replyCh := make(chan AppendEntriesReply, len(rf.peers))
		var wg sync.WaitGroup
		for i := range rf.peers {
			if i == rf.me {
				continue
			}
			wg.Add(1)

			go func(server int) {
				defer wg.Done()
				var reply AppendEntriesReply
				if succ := rf.sendAppendEntries(server, &args, &reply); !succ {
					return
				}
				replyCh <- reply
			}(i)
		}
		wg.Wait()
		close(replyCh)

		var lived int
		for reply := range replyCh {
			if reply.Term > rf.curTerm {
				// 发现新 leader，如网络分区恢复
				rf.back2Follower(reply.Term)
				return
			}
			lived++
		}

		// 未收到来自大多数节点的心跳，重新开始选举
		if lived < len(rf.peers)/2+1 {
			rf.vote() // 重新开始投票
			return
		}

		<-t.C
	}
}

func (rf *Raft) AppendEntries(args *AppendEntriesArgs, reply *AppendEntriesReply) {
	if len(args.Entries) > 0 {
		log.Fatal("invalid entry in 2A")
	}

	reply.Term = rf.curTerm
	if rf.curTerm > args.Term {
		reply.Succ = false
		return
	}

	// 检查双方日志的一致性
	if i := len(rf.entries) - 1; i >= 0 {
		switch {
		case i < args.PrevLogIndex: // 本地少日志，让 leader nextIndex[i]-- 后再同步
			reply.Succ = false
			return
		case i == args.PrevLogIndex:
			if rf.entries[i].Term != args.PrevLogTerm { // term 不匹配
				reply.Succ = false
				return
			}
		case i > args.PrevLogIndex: // 强制删除
			rf.entries = rf.entries[args.PrevLogIndex:]
		}
	}
	rf.entries = append(rf.entries, args.Entries...)
	rf.entryCh <- LogEntry{Term: args.Term}

	reply.Succ = true
	return
}