6.824 lab2 raft A、B实验

//上传给状态机
type ApplyMsg struct {
	CommandValid bool
	Command      interface{}
	CommandIndex int
	CommandTerm  int

	//For 2D:
	SnapshotValid bool
	Snapshot      []byte
	SnapshotTerm  int
	SnapshotIndex int
}

//
// A Go object implementing a single Raft peer.
//
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()
	nPeers    int //所有节点数量

	timeoutInterval time.Duration //follower leader candidate
	lastActiveTime  time.Time     //超时开始计算时间，收到心跳时会更新

	//选举
	term     int
	role     MemberRole
	leaderId int
	votedFor int

	//提交情况
	//log         *RaftLog
	logs        []*LogEntry
	nextIndex   []int
	matchIndex  []int
	commitIndex int 
	lastApplied int

	//状态机
	applyCond *sync.Cond
	applyChan chan ApplyMsg //应用层与raft交互的channel，用于上传数据给应用层
}

type MemberRole int

const (
	Leader    MemberRole = 1
	Follower  MemberRole = 2
	Candidate MemberRole = 3

	RoleNone = -1
	None     = 0
)

func (m MemberRole) String() string {
	switch m {
	case Leader:
		return "Leader"
	case Follower:
		return "Follower"
	case Candidate:
		return "Candidate"
	}
	return "Unknown"
}

type LogType int

const (
	HeartBeatLogType   LogType = 1
	AppendEntryLogType LogType = 2
)

func (l LogType) String() string {
	switch l {
	case HeartBeatLogType:
		return "HeartBeatLogType"
	case AppendEntryLogType:
		return "AppendEntryLogType"
	}
	return "Unknown"
}

type LogEntry struct {
	LogTerm int
	Command interface{}
}

const (
	ElectionTimeout = 200 * time.Millisecond
	HeatBeatTimeout = 150 * time.Millisecond
)

//选举时需要传递自己拥有的最后一条log的term和index
type RequestVoteArgs struct {
	// Your data here (2A, 2B).
	Term         int
	CandidateId  int
	LastLogIndex int
	LastLogTerm  int
}

//
// example RequestVote RPC reply structure.
// field names must start with capital letters!
//
type RequestVoteReply struct {
	Term        int
	VoteGranted bool
}

//
// 心跳或者日志追加
//
type AppendEntriesArgs struct {
	LogType  LogType
	LeaderId int
	Term     int //leader currentTerm
	//用于日志复制，确保前面日志能够匹配
	PrevLogTerm         int
	PrevLogIndex        int
	LeaderCommitedIndex int
	LogEntries          []*LogEntry
}

//
// 心跳或者日志追加
//
type AppendEntriesReply struct {
	Success bool
	Term    int
	Msg     string
}

//初始化raft, 所有raft的任务都要另起协程，测试文件采用的是协程模拟rpc
func Make(peers []*labrpc.ClientEnd, me int,
	persister *Persister, applyCh chan ApplyMsg) *Raft {

	rf := &Raft{
		mu:        sync.Mutex{},
		peers:     peers,
		persister: persister,
		me:        me,
		dead:      -1,
		nPeers:    len(peers),

		leaderId:       RoleNone,
		term:           None,
		votedFor:       RoleNone,
		role:           Follower,
		lastActiveTime: time.Now(),
		//lastHeartBeatTime: time.Now(),
		timeoutInterval: randElectionTimeout(),
		commitIndex:     None,
		lastApplied:     None,
		applyChan:       applyCh,
	}
	rf.applyCond = sync.NewCond(&rf.mu)
	DPrintf("starting new raft node, id[%d], lastActiveTime[%v], timeoutInterval[%d]", me, rf.lastActiveTime.UnixMilli(), rf.timeoutInterval.Milliseconds())

	rf.logs = make([]*LogEntry, 0)
	rf.nextIndex = make([]int, rf.nPeers)
	rf.matchIndex = make([]int, rf.nPeers)
	for i := 0; i < rf.nPeers; i++ {
		rf.nextIndex[i] = 1
		rf.matchIndex[i] = 0
	}
    //有效logIndex、logTerm从1开始，刚开始写入一条dummy log是为了减少判空
	rf.logs = append(rf.logs, &LogEntry{
		LogTerm: 0,
	})

	// initialize from state persisted before a crash
	rf.readPersist(persister.ReadRaftState())
	// start ticker goroutine to start elections
	go rf.electionLoop()
	go rf.heartBeatLoop()
	go rf.applyLogLoop(applyCh)

	DPrintf("starting raft node[%d]", rf.me)

	return rf
}

func (rf *Raft) electionLoop() {
	for rf.killed() == false {
		time.Sleep(time.Millisecond * 1)
		func() {
			rf.mu.Lock()
			defer rf.mu.Unlock()
            //Follower、Candidate在这个循环，Leader在heartBeatLoop
			if rf.role == Leader {
				return
			}
			if time.Now().Sub(rf.lastActiveTime) < rf.timeoutInterval {
				//不超时不需要进入下一步，只需要接收RequestVote和AppendEntries请求即可
				return
			}
			//超时处理逻辑
			if rf.role == Follower {
				rf.role = Candidate
			}
			DPrintf("become candidate... node[%v] term[%v] role[%v] lastActiveTime[%v], timeoutInterval[%d], now[%v]", rf.me, rf.term, rf.role, rf.lastActiveTime.UnixMilli(), rf.timeoutInterval.Milliseconds(), time.Now().Sub(rf.lastActiveTime).Milliseconds())
			rf.lastActiveTime = time.Now()
			rf.timeoutInterval = randElectionTimeout()
			rf.votedFor = rf.me
			rf.term++
			rf.persist()
			lastLogTerm, lastLogIndex := rf.lastLogTermAndLastLogIndex()
			rf.mu.Unlock()
            
			maxTerm, voteGranted := rf.becomeCandidate(lastLogIndex, lastLogTerm)
			rf.mu.Lock()
			//DPrintf("node[%d] get vote num[%d]", rf.me, totalVote)

			//在这过程中接收到更大term的请求，导致退化为follower
			if rf.role != Candidate {
				//DPrintf("node[%d] role[%v] failed to leader, voteGranted[%d], totalVote[%d]", rf.me, rf.role, voteGranted, totalVote)
				return
			}
            //出现更大term后就退回到Follower
			if maxTerm > rf.term {
				rf.role = Follower
				rf.term = maxTerm
				rf.votedFor = RoleNone
				rf.leaderId = RoleNone
				rf.persist()
			} else if voteGranted > rf.nPeers/2 {
                //成为Leader
				rf.leaderId = rf.me
				rf.role = Leader
                //立刻发送心跳
				rf.lastActiveTime = time.Unix(0, 0)
			}
			DPrintf("node[%d] role[%v] maxTerm[%d] voteGranted[%d] nPeers[%d]", rf.me, rf.role, maxTerm, voteGranted, rf.nPeers)
		}()
	}
}

//并发去请求投票，并返回投票结果
func (rf *Raft) becomeCandidate(lastLogIndex, lastLogTerm int) (int, int) {

	type RequestVoteResult struct {
		peerId int
		resp   *RequestVoteReply
	}
	voteChan := make(chan *RequestVoteResult, rf.nPeers-1)
	args := &RequestVoteArgs{
		Term:         rf.term,
		CandidateId:  rf.me,
		LastLogIndex: lastLogIndex,
		LastLogTerm:  lastLogTerm,
	}
	for i := 0; i < rf.nPeers; i++ {
		if rf.me == i {
			continue
		}
		go func(server int, args *RequestVoteArgs) {
			reply := &RequestVoteReply{}
			ok := rf.sendRequestVote(server, args, reply)
			if ok {
				voteChan <- &RequestVoteResult{
					peerId: server,
					resp:   reply,
				}
			} else {
				voteChan <- &RequestVoteResult{
					peerId: server,
					resp:   nil,
				}
			}
		}(i, args)
	}

	maxTerm := rf.term
	voteGranted := 1
	totalVote := 1
	for i := 0; i < rf.nPeers-1; i++ {
		select {
		case vote := <-voteChan:
			totalVote++
			if vote.resp != nil {
				if vote.resp.VoteGranted {
					voteGranted++
				}
				//出现更大term就退回follower
				if vote.resp.Term > maxTerm {
					maxTerm = vote.resp.Term
				}
			}
		}
        //一旦过半投票就立刻返回，否则会被部分故障节点拖慢达成一致的时间导致超时
		if voteGranted > rf.nPeers/2 || totalVote == rf.nPeers {
			return maxTerm, voteGranted
		}
	}
	return maxTerm, voteGranted
}

func (rf *Raft) RequestVote(args *RequestVoteArgs, reply *RequestVoteReply) {
	// Your code here (2A, 2B).
	rf.mu.Lock()
	defer func() {
		DPrintf("node[%d] role[%v] received vote from node[%d], now[%d], args: %v, reply: %v", rf.me, rf.role, args.CandidateId, time.Now().UnixMilli(), mr.Any2String(args), mr.Any2String(reply))
	}()
	defer rf.mu.Unlock()
	reply.Term = rf.term
	reply.VoteGranted = false
	//不接收小于自己term的请求
	if rf.term > args.Term {
		return
	}

	if args.Term > rf.term {
		rf.role = Follower //leader转换为follower
		rf.term = args.Term
		//需要比较最新一条日志的情况再决定要不要投票
		rf.votedFor = RoleNone
		rf.leaderId = RoleNone
		//rf.lastActiveTime = time.Now()
		//rf.lastHeartBeatTime = time.Now()
		//rf.timeoutInterval = randElectionTimeout()
		rf.persist()
	}
	//避免重复投票
	if rf.votedFor == RoleNone || rf.votedFor == args.CandidateId {
		lastLogTerm, lastLogIndex := rf.lastLogTermAndLastLogIndex()
		//最后一条日志任期更大或者任期一样但是更长
		if args.LastLogTerm > lastLogTerm || (lastLogTerm == args.LastLogTerm && lastLogIndex <= args.LastLogIndex) {
			rf.role = Follower
			rf.votedFor = args.CandidateId
			rf.leaderId = args.CandidateId
			rf.lastActiveTime = time.Now()
			rf.timeoutInterval = randElectionTimeout()
			reply.VoteGranted = true
			rf.persist()
		}
	}
}

func (rf *Raft) heartBeatLoop() {
	for rf.killed() == false {
		//不要太频繁了，否则会导致rpc太多
		time.Sleep(time.Millisecond * 20)
		func() {
			rf.mu.Lock()
			defer rf.mu.Unlock()
            //只有Leader才能进入心跳循环
			if rf.role != Leader {
				return
			}
			//如果没有超时，则直接返回
			if time.Now().Sub(rf.lastActiveTime) < HeatBeatTimeout-50 {
				return
			}
			rf.lastActiveTime = time.Now()
			for i := 0; i < rf.nPeers; i++ {
				if rf.me == i {
					rf.matchIndex[i] = rf.lastLogIndex()
					rf.nextIndex[i] = rf.matchIndex[i] + 1
					continue
				}
				//DPrintf("lastLogIndex: %v, logs: %v", lastLogIndex, mr.Any2String(rf.logs))
				//记录每个node本次发送日志的前一条日志
				prevLogIndex := rf.matchIndex[i]
				argsI := &AppendEntriesArgs{
					LogType:             HeartBeatLogType,
					Term:                rf.term,
					LeaderId:            rf.me,
					PrevLogIndex:        prevLogIndex,
					PrevLogTerm:         rf.logTerm(prevLogIndex),
					LeaderCommitedIndex: rf.commitIndex, //对上一次日志复制请求的二阶段
				}

				//本次复制的最后一条日志
				lastLogIndex := rf.lastLogIndex()
				if rf.matchIndex[i] < lastLogIndex {
					argsI.LogType = AppendEntryLogType
					argsI.LogEntries = make([]*LogEntry, 0)
					//因为此时没有加锁，担心有新日志写入，必须保证每个节点复制的最后一条日志一样才能起到过半提交的效果
					argsI.LogEntries = append(argsI.LogEntries, rf.logs[rf.nextIndex[i]:lastLogIndex+1]...)
				}

				go func(server int, args *AppendEntriesArgs) {
					reply := &AppendEntriesReply{}
					ok := rf.sendAppendEntries(server, args, reply)
					if !ok {
						return
					}
					rf.mu.Lock()
					defer rf.mu.Unlock()
					//如果term变了，表示该结点不再是leader，什么也不做
					if rf.term != args.Term {
						return
					}
					//发现更大的term，本结点是旧leader
					if reply.Term > rf.term {
						rf.term = reply.Term
						rf.votedFor = RoleNone
						rf.leaderId = RoleNone
						rf.role = Follower
						rf.persist()
						return
					}
					if reply.Success {
						rf.nextIndex[server] += len(args.LogEntries)
						rf.matchIndex[server] = rf.nextIndex[server] - 1
						//提交到哪个位置需要根据中位数来判断，中位数表示过半提交的日志位置，
						//每次提交日志向各结点发送的日志并不完全一样，不能光靠是否发送成功来判断
						matchIndexSlice := make([]int, rf.nPeers)
						for index, matchIndex := range rf.matchIndex {
							matchIndexSlice[index] = matchIndex
						}
						sort.Slice(matchIndexSlice, func(i, j int) bool {
							return matchIndexSlice[i] < matchIndexSlice[j]
						})
						//fmt.Printf("matchIndexSlice: %v, newcommitIndex: %v\n", mr.Any2String(matchIndexSlice), matchIndexSlice[rf.nPeers/2])
						newCommitIndex := matchIndexSlice[rf.nPeers/2]
						//不能提交不属于当前term的日志
						if newCommitIndex > rf.commitIndex && rf.logs[newCommitIndex].LogTerm == rf.term {
							DPrintf("id[%d] role[%v] commitIndex %v update to newcommitIndex %v, command: %v", rf.me, rf.role, rf.commitIndex, newCommitIndex, rf.logs[newCommitIndex])
							rf.commitIndex = newCommitIndex
						}
					} else {
						//follower缺少的之前的日志，探测缺少的位置
						//后退策略，可以按term探测，也可以二分，此处采用线性探测，简单一些
						rf.nextIndex[server] -= 1
						if rf.nextIndex[server] < 1 {
							rf.nextIndex[server] = 1
						}
					}
				}(i, argsI)
			}
		}()
	}
}

func (rf *Raft) AppendEntries(args *AppendEntriesArgs, reply *AppendEntriesReply) {

	rf.mu.Lock()
	defer rf.mu.Unlock()
	defer func() {
		DPrintf("id[%d] role[%v] args: %v, reply: %v", rf.me, rf.role, mr.Any2String(args), mr.Any2String(reply))
	}()

	reply.Term = rf.term
	reply.Success = false
	//拒绝旧leader请求
	if args.Term < rf.term {
		return
	}
	//发现一个更大的任期，转变成这个term的follower，leader、follower--> follower
	if args.Term > rf.term {
		rf.term = args.Term
		rf.role = Follower
		//发现term大于等于自己的日志复制请求，则认其为主
		rf.votedFor = RoleNone
		rf.leaderId = RoleNone
		rf.persist()
	}
	rf.leaderId = args.LeaderId
    rf.votedFor = args.LeaderId
	rf.lastActiveTime = time.Now()
	//还缺少前面的日志或者前一条日志匹配不上
	if args.PrevLogIndex > rf.lastLogIndex() || args.PrevLogTerm != rf.logTerm(args.PrevLogIndex) {
		return
	}
	//args.PrevLogIndex<=lastLogIndex，有可能发生截断的情况
	if rf.lastLogIndex() > args.PrevLogIndex {
		rf.logs = rf.logs[:args.PrevLogIndex+1]
	}
	rf.logs = append(rf.logs, args.LogEntries...)
	rf.persist()
	if args.LeaderCommitedIndex > rf.commitIndex {
		rf.commitIndex = args.LeaderCommitedIndex
		if rf.lastLogIndex() < rf.commitIndex {
			rf.commitIndex = rf.lastLogIndex()
		}
	}
	reply.Success = true
}

//写入数据
func (rf *Raft) Start(command interface{}) (int, int, bool) {
	rf.mu.Lock()
	defer rf.mu.Unlock()

	if rf.role != Leader {
		return -1, -1, false
	}
	entry := &LogEntry{
		LogTerm: rf.term,
		Command: command,
	}
	rf.logs = append(rf.logs, entry)
	index := rf.lastLogIndex()
	term := rf.logTerm(index)
	//写入后立刻持久化
	rf.persist()
	DPrintf("node[%d] term[%d] role[%v] add entry: %v, logIndex[%d]", rf.me, rf.term, rf.role, mr.Any2String(entry), index)
	return index, term, true
}

//日志提交循环，不采用定期唤醒，而是条件变量
func (rf *Raft) applyLogLoop(applyCh chan ApplyMsg) {

	for !rf.killed() {
		time.Sleep(time.Millisecond * 10)
		applyMsgs := make([]ApplyMsg, 0)
		func() {
			rf.mu.Lock()
			defer rf.mu.Unlock()

			//没有数据需要上传给应用层
			if rf.lastApplied >= rf.commitIndex {
				return
			}
			for rf.lastApplied < rf.commitIndex {
				rf.lastApplied++
				applyMsgs = append(applyMsgs, ApplyMsg{
					CommandValid: true,
					Command:      rf.logs[rf.lastApplied].Command,
					CommandIndex: rf.lastApplied,
				})
			}
		}()
		go func() {
			//锁外提交给应用
			for i := 0; i < len(applyMsgs); i++ {
				DPrintf("id[%v] role[%v] upload log to application, lastApplied[%d], commitIndex[%d]", rf.me, rf.role, applyMsgs[i].CommandIndex, rf.commitIndex)

				applyCh <- applyMsgs[i]
			}
		}()
	}
}

func randElectionTimeout() time.Duration {
	return ElectionTimeout + time.Duration(rand.Uint32())%ElectionTimeout
}

// return currentTerm and whether this server
// believes it is the leader.
//上层用来检测节点的状态
func (rf *Raft) GetState() (int, bool) {
	rf.mu.Lock()
	defer rf.mu.Unlock()

	return rf.term, rf.role == Leader
}

func (rf *Raft) lastLogTermAndLastLogIndex() (int, int) {
	logIndex := len(rf.logs) - 1
	logTerm := rf.logs[logIndex].LogTerm
	return logTerm, logIndex
}

func (rf *Raft) lastLogIndex() int {
	return len(rf.logs) - 1
}
func (rf *Raft) logTerm(logIndex int) int {
	return rf.logs[logIndex].LogTerm
}

func (cfg *config) checkOneLeader() int {
	for iters := 0; iters < 10; iters++ {
		ms := 450 + (rand.Int63() % 100)
		time.Sleep(time.Duration(ms) * time.Millisecond)

		leaders := make(map[int][]int)
		for i := 0; i < cfg.n; i++ {
            //connected表示这个节点是否在网络中
			if cfg.connected[i] {
				if term, leader := cfg.rafts[i].GetState(); leader {
					leaders[term] = append(leaders[term], i)
				}
			}
		}

		lastTermWithLeader := -1
		for term, leaders := range leaders {
            //Leader节点超过1则Panic
			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]
		}
	}
    //重复十次，如果都没有leader，表示选主有问题
	cfg.t.Fatalf("expected one leader, got none")
	return -1
}

//写入日志cmd，10秒内成功
func (cfg *config) one(cmd interface{}, expectedServers int, retry bool) int {
	t0 := time.Now()
	starts := 0
	for time.Since(t0).Seconds() < 10 && cfg.checkFinished() == false {
		// try all the servers, maybe one is the leader.
		index := -1
		for si := 0; si < cfg.n; si++ {
			starts = (starts + 1) % cfg.n
			var rf *Raft
			cfg.mu.Lock()
			if cfg.connected[starts] {
				rf = cfg.rafts[starts]
			}
			cfg.mu.Unlock()
			if rf != nil {
				//fmt.Printf("connected: %v, raft: %v\n", mr.Any2String(cfg.connected), mr.Any2String(rf))
                //写入日志
				index1, _, ok := rf.Start(cmd)
				if ok {
					index = index1
					break
				}
			}
		}
		//fmt.Printf("cfg index: %v\n", index)
		if index != -1 {
			// somebody claimed to be the leader and to have
			// submitted our command; wait a while for agreement.
			t1 := time.Now()
            //判断2s内是否能够将日志复制到全部结点
			for time.Since(t1).Seconds() < 2 {
				nd, cmd1 := cfg.nCommitted(index)
				if nd > 0 && nd >= expectedServers {
					// committed
					if cmd1 == cmd {
						// and it was the command we submitted.
						return index
					}
				}
				time.Sleep(20 * time.Millisecond)
			}
			if retry == false {
				cfg.t.Fatalf("one(%v) failed to reach agreement", cmd)
			}
		} else {
			time.Sleep(50 * time.Millisecond)
		}
	}
	//fmt.Printf("cfg.logs: %v\n", mr.Any2String(cfg.logs))
	if cfg.checkFinished() == false {
		cfg.t.Fatalf("one(%v) failed to reach agreement", cmd)
	}
	return -1
}