Lab2B. Raft 日志复制实现

type LogEntry struct {
	Term    int          
	Command interface{}
}

type AppendEntriesArgs struct {
	Term         int        // leader term
	LeaderID     int        // so follower can redirect clients
	PrevLogIndex int        // index of log entry immediately preceding new ones
	PrevLogTerm  int        // term of prevLogIndex entry
	Entries      []LogEntry // log entries to store (empty for heartbeat;may send more than one for efficiency)
	LeaderCommit int        // leader’s commitIndex
}

type AppendEntriesReply struct {
	Term          int  // currentTerm, for leader to update itself
	Succ          bool // true if follower contained entry matching prevLogIndex and prevLogTerm
}

func (rf *Raft) AppendEntries(args *AppendEntriesArgs, reply *AppendEntriesReply) {
	reply.Term = rf.curTerm
	reply.Succ = false

	if args.Term < rf.curTerm {
		return // leader expired
	}

	if args.Term > rf.curTerm {
		rf.curTerm = args.Term
		rf.back2Follower(args.Term, VOTE_NIL)
	}
	// now terms are same
	rf.resetElectTimer()

	// consistency check
	last := len(rf.logs) - 1
	if last < args.PrevLogIndex { // missing logs
		return
	}
	// now peer and leader have same prevIndex and same prevTerm

	// check conflict and append new logs
	committed := prevIdx
	for i, e := range args.Entries {
		cur := prevIdx + 1 + i
		if cur <= last && rf.logs[cur].Term != e.Term { // term conflict, overwrite it
			rf.logs[cur] = e
			committed = cur
		}
		if cur > last {
			rf.logs = append(rf.logs, e) // new log, just append
			committed = len(rf.logs) - 1
		}
	}

	// if leaderCommit > commitIndex, set commitIndex =min(leaderCommit, index of last new entry)
	if args.LeaderCommit > rf.commitIndex {
		rf.commitIndex = min(committed, args.LeaderCommit) // need to commit
		rf.applyCond.Broadcast() // trigger apply
	}

	rf.back2Follower(args.Term, args.LeaderID)
	reply.Succ = true
}

// leader replicate logs or send heartneat to other nodes
func (rf *Raft) sync() {
	for i := range rf.peers {
		if i == rf.me {
			rf.resetElectTimer()
			continue
		}

		go func(server int) {
			for {
				if !rf.isLeader() {
					return
				}

				rf.mu.Lock()
				rf.syncConds[server].Wait() // wait for heartbeat or Start to trigger

				// sync new log or missing logs to server
				next := rf.nextIndex[server]
				args := AppendEntriesArgs{
					Term:         rf.curTerm,
					LeaderID:     rf.me,
					Entries:      nil,
					LeaderCommit: rf.commitIndex,
				}
				if next < len(rf.logs) { // logs need sync
					args.PrevLogIndex = next - 1
					args.PrevLogTerm = rf.logs[next-1].Term
					args.Entries = append(args.Entries, rf.logs[next:]...)
				}
				rf.mu.Unlock()

				// do not depend on labrpc to call timeout(it may more bigger than heartbeat), so should be check manually
				var reply AppendEntriesReply
				respCh := make(chan struct{})
				go func() {
					rf.sendAppendEntries(server, &args, &reply)
					respCh <- struct{}{}
				}()
				select {
				case <-time.After(RPC_CALL_TIMEOUT): // After() with currency may be inefficient
					continue
				case <-respCh:
				}

				if !reply.Succ {
					if reply.Term > rf.curTerm { // higher term
						rf.back2Follower(reply.Term, VOTE_NIL)
						return
					}
					continue
				}

				// append succeed
				rf.nextIndex[server] += len(args.Entries)
				rf.matchIndex[server] = rf.nextIndex[server] - 1 // replicate succeed
			}
		}(i)
	}
}

// apply (lastApplied, commitIndex]
func (rf *Raft) waitApply() {
	for {
		rf.mu.Lock()
		rf.applyCond.Wait() // wait for new commit log trigger

		var logs []LogEntry // un apply logs
		applied := rf.lastApplied
		committed := rf.commitIndex
		if applied < committed {
			for i := applied + 1; i <= committed; i++ {
				logs = append(logs, rf.logs[i])
			}
			rf.lastApplied = committed // update applied
		}
		rf.mu.Unlock()

		for i, l := range logs {
			msg := ApplyMsg{
				Command:      l.Command,
				CommandIndex: applied + 1 + i, // apply to state machine
				CommandValid: true,
			}
			rf.applyCh <- msg
		}
	}
}

// leader daemon detect and commit log which has been replicated on majority successfully
func (rf *Raft) leaderCommit() {
	for {
		if !rf.isLeader() {
			return
		}

		rf.mu.Lock()
		majority := len(rf.peers)/2 + 1
		n := len(rf.logs)
		for i := n - 1; i > rf.commitIndex; i-- { // looking for newest commit index from tail to head
			// in current term, if replicated on majority, commit it
			replicated := 0
			if rf.logs[i].Term == rf.curTerm {
				for server := range rf.peers {
					if rf.matchIndex[server] >= i {
						replicated += 1
					}
				}
			}

			if replicated >= majority {
				// all (commitIndex, newest commitIndex] logs are committed
				// leader now apply them
				rf.applyCond.Broadcast()
				rf.commitIndex = i
				break
			}
		}
		rf.mu.Unlock()
	}
}

// election restrictions
func (rf *Raft) RequestVote(args *RequestVoteArgs, reply *RequestVoteReply) {
	reply.Term = rf.curTerm
	reply.VoteGranted = false

	if args.Term < rf.curTerm {
		return // candidate expired
	}
	if args.Term > rf.curTerm {
		rf.back2Follower(args.Term, VOTE_NIL)
	}
	// now the term are same

	// check up-to-date, from Paper:
	// if the logs have last entries with different terms, then the log with the later term is more up-to-date.
	// if the logs end with the same term, then whichever log is longer is more up-to-date.
	i := len(rf.logs) - 1
	lastTerm := rf.logs[i].Term
	if lastTerm > args.LastLogTerm {
		return
	}
	if lastTerm == args.LastLogTerm && i > args.LastLogIndex {
		return
	}
	// now last index and term both matched

	if rf.votedFor == VOTE_NIL || rf.votedFor == args.CandidateID {
		reply.VoteGranted = true
		rf.back2Follower(args.Term, args.CandidateID)
	}

	return
}

➜  raft git:(master) tree
.
├── config.go
├── persister.go
├── raft.go          # 节点初始化，超时选举机制
├── raft_entry.go    # AppendEntries RPC 逻辑
├── raft_leader.go   # sync 日志，心跳通信等
├── raft_peer.go     # 定义超时时间
├── raft_vote.go     # RequestVote RPC 逻辑
├── test_test.go
└── util.go          # 自定义的调试函数等