分布式系统作业丨02

func (rf *Raft) sendRequestVote(server int, args RequestVoteArgs, reply *RequestVoteReply) bool {
	ok := rf.peers[server].Call("Raft.RequestVote", args, reply)
	return ok
}

type RequestVoteArgs struct {
	Term         int // candidate的当前任期
	CandidatedId int // 请求投票的candidate的ID
	LastLogIndex int // candidate最后一个日志条目的index（确保安全性的选举限制用）
	LastLogTerm  int // candidate最后一个日志条目的term（确保安全性的选举限制用）
}
 
type RequestVoteReply struct {
	Term        int
	VoteGranted bool
}
 
type Raft struct {
	mu        sync.Mutex
	peers     []*labrpc.ClientEnd
	persister *Persister
	me        int // index into peers[]
	 
	ballot  int           //选票数量
	timer   *time.Timer   // 计时器指针，用于控制选举超时
	state   string        // 当前节点的状态 Follower, Candidate 或者 Leader
	term    int           //当前选举的 term
	voteTo  int           //表示 vote 的对象，如果是-1表示还未投票，如果不是-1不能再投票
	timeout time.Duration //选举超时时间
	log     []LogEntry    // 此Server的日志，包含了若干日志条目，类型是日志条目的切片，第一个日志索引是1
	applyCh chan ApplyMsg //  根据Make()及其他部分的注释，raft server需要维护一个发送ApplyMsg的管道
	 
	// 所有server 容易改变的状态
	commitIndex int
	lastApplied int
	 
	// Leader 容易改变的状态
	nextIndex  []int
	matchIndex []int
}
 
type LogEntry struct {
	Command interface{} // 客户端要求的指令
	Term    int         // 此日志条目的term
}

func (rf *Raft) tick() {
	if rf.timer != nil {
		// 如果 timer 已经初始化，停下之前的 timer
		rf.timer.Stop()
	}
	rf.timer = time.AfterFunc(rf.timeout, func() { rf.ElectionTimeout() }) // 更新 timer
}
 
func (rf *Raft) RequestVote(args RequestVoteArgs, reply *RequestVoteReply) {
	// Your code here.
	rf.mu.Lock()
	defer rf.mu.Unlock()
	 
	// 填充回复参数的默认值
	reply.Term = rf.term // 如果请求者的 term 比当前节点旧，候选者将更新 term
	reply.VoteGranted = false
	 
	//如果请求者的任期 term 比当前节点旧，则直接拒绝投票
	if args.Term < rf.term {
		return
	}
	 
	// 如果请求者的任期比当前节点新，则更新为跟随者（Follower）
	if args.Term > rf.term {
		rf.state = "Follower"
		rf.voteTo = args.CandidatedId
		rf.persist()
	}
	 
	// 更新当前节点的任期为请求者的任期
	rf.term = args.Term
	if (rf.voteTo == -1 || rf.voteTo == args.CandidatedId) && rf.incomingFresher(args) {
		rf.state = "Follower"
		reply.VoteGranted = true
		rf.voteTo = args.CandidatedId
		rf.persist()
		rf.tick()
	}
}
 
func (rf *Raft) RequestVote(args RequestVoteArgs, reply *RequestVoteReply) {
	// Your code here.
	rf.mu.Lock()
	defer rf.mu.Unlock()
	 
	// 构造参数信息
	reply.Term = rf.term // if fresher, candidate will update
	reply.VoteGranted = false
	if args.Term < rf.term { // all ready seen leader for this term
		return
	}
	 
	if args.Term > rf.term { // agree if your term is more recent
		rf.state = "Follower"
		rf.voteTo = args.CandidatedId
		rf.persist()
	}
	 
	rf.term = args.Term // update own term
	if (rf.voteTo == -1 || rf.voteTo == args.CandidatedId) && rf.incomingFresher(args) {
		rf.state = "Follower"
		reply.VoteGranted = true
		rf.voteTo = args.CandidatedId
		rf.persist()
		rf.tick()
	}
}
 
func (rf *Raft) incomingFresher(args RequestVoteArgs) bool {
	if len(rf.log) == 0 {
		return true
	}
	if args.LastLogTerm == rf.log[len(rf.log)-1].Term {
		return len(rf.log)-1 <= args.LastLogIndex
	}
	return args.LastLogTerm > rf.log[len(rf.log)-1].Term
}
 
func (rf *Raft) handleVoteResult(reply RequestVoteReply) {
	rf.mu.Lock()
	defer rf.mu.Unlock()
	 
	if reply.Term > rf.term {
		rf.term = reply.Term
		rf.state = "Follower"
		rf.voteTo = -1
		rf.persist()
		rf.tick()
		return
	}
	 
	if reply.VoteGranted && rf.state == "Candidate" {
		rf.ballot++
		if rf.ballot > len(rf.peers)/2 { // majority
			rf.state = "Leader" // step up
			rf.persist()
			for i := 0; i < len(rf.peers); i++ {
				if i == rf.me {
					continue
				}
				rf.nextIndex[i] = len(rf.log)
				rf.matchIndex[i] = -1 // index 从0开始, 所以初始化为-1
			}
			rf.tick()
			go func() {
				rf.stepUpCallback() // 发送 heartbeat
			}()
		}
	}
}
 
func (rf *Raft) ElectionTimeout() {
	rf.mu.Lock()
	defer rf.mu.Unlock()
	if rf.state == "Leader" {
		// Leader 仅仅刷新 timer
		rf.tick()
	} else {
		rf.state = "Candidate"
		rf.voteTo = rf.me
		rf.persist()
		rf.ballot = 1
		rf.term += 1
		 
		// 构造参数信息
		var args RequestVoteArgs
		args.Term = rf.term
		args.CandidatedId = rf.me
		args.LastLogIndex = len(rf.log) - 1
		if len(rf.log) == 0 {
			args.LastLogTerm = -1
		} else {
			args.LastLogTerm = rf.log[len(rf.log)-1].Term
		}
		 
		// 发送 RequestForVote 并接受回复
		f := func(server int, args RequestVoteArgs) {
			var resp RequestVoteReply
			var ifSuccess = rf.sendRequestVote(server, args, &resp)
			if ifSuccess {
				rf.handleVoteResult(resp)
			}
		}
		for i := 0; i < len(rf.peers); i++ {
			if i != rf.me {
				go f(i, args)
			}
		}
		rf.tick()
	}
}
 
func Make(peers []*labrpc.ClientEnd, me int,
	persister *Persister, applyCh chan ApplyMsg) *Raft {
	rf := &Raft{}
	rf.peers = peers
	rf.persister = persister
	rf.me = me
	 
	// Your initialization code here.
	rf.term = 0
	rf.state = "Follower" // server刚开始为follower，且current term为0
	rf.voteTo = -1        // 初始时还没投票，就为-1
	rf.readPersist(persister.ReadRaftState())
	rf.log = make([]LogEntry, 0)
	timeout := rand.Intn(150) + 150
	rf.timeout = time.Duration(timeout) * time.Millisecond
	rf.ballot = 0
	 
	rf.commitIndex = -1
	rf.lastApplied = -1
	 
	rf.nextIndex = make([]int, len(rf.peers))
	rf.matchIndex = make([]int, len(rf.peers))
	 
	rf.applyCh = applyCh
	 
	rf.tick()
	rf.readPersist(persister.ReadRaftState())
	return rf
}

type AppendEntriesArgs struct {
	Term        int // leader的任期
	Leader      int
	PreLogIndex int        // 新条目之前的紧接着的日志的索引
	PreLogTerm  int        // 新条目之前的紧接着的日志的任期
	Entries     []LogEntry // 要存储/追加到server的日志条目，为了效率可一次追加多条（若为 heartbeat 则此字段为空）
	CommitTo    int        // leader提交到的日志索引位置
}
 
type AppendEntriesReply struct {
	Term    int  // RPC接收server的current term，leader更新自己用（如果需要的话）
	Success bool // 如果follower包含有匹配leader的preLogIndex以及preLogTerm的日志条目则返回true
}

func (rf *Raft) AppendEntries(args *AppendEntriesArgs, reply *AppendEntriesReply) {
	rf.mu.Lock()
	defer rf.mu.Unlock()
	 
	reply.Term = rf.term
	if rf.term > args.Term {
		reply.Success = false
		return
	} else {
		rf.term = args.Term
		rf.state = "Follower" // in case disconnected leader wake up
		rf.voteTo = -1
		rf.persist()
		reply.Term = rf.term
		rf.tick()
		 
		if args.PreLogIndex > -1 {
			if len(rf.log) <= args.PreLogIndex || 
				rf.log[args.PreLogIndex].Term != args.PreLogTerm {
				// 不匹配
				reply.Success = false
				return
			}
		}
		 
		// prevLogIndex 值为-1或者成功匹配
		rf.log = rf.log[:args.PreLogIndex+1] // discard all following
		rf.log = append(rf.log, args.Entries...)
		rf.persist()
		if rf.commitIndex < args.CommitTo {
			rf.commitIndex = args.CommitTo
			go rf.refreshCommits()
		}
		reply.Success = true
	}
}
 
func (rf *Raft) handleAppendEntriesResult(reply AppendEntriesReply, nodeId int) {
	rf.mu.Lock()
	defer rf.mu.Unlock()
	 
	if rf.state != "Leader" {
		return
	}
	 
	if reply.Term > rf.term {
		// 自己的 term 已经过时
		rf.term = reply.Term
		rf.state = "Follower"
		rf.voteTo = -1
		rf.persist()
	}
	 
	if reply.Success {
		rf.matchIndex[nodeId] = rf.nextIndex[nodeId] - 1
		rf.nextIndex[nodeId] = len(rf.log) // 更新, 用于在下次检查是否是最新的
		 
		for i := len(rf.log) - 1; i >= 0 && i > rf.commitIndex; i-- {
			// term 从领导者尚未提交的最后一个任期开始倒推。
			if rf.log[i].Term != rf.term {
				break
			}
			 
			count := 1
			for server := 0; server < len(rf.peers); server++ {
				if server == rf.me {
					continue
				}
				if rf.matchIndex[server] >= i {
					count++
				}
			}
			if count > len(rf.peers)/2 {
				rf.commitIndex = i // 更新 commitIndex
				rf.refreshCommits()
			}
		}
	} else {
		rf.nextIndex[nodeId]--
	}
}
 
func (rf *Raft) sendAppendEntry2Index(nodeId int) {
	if nodeId == rf.me {
		return
	}
	var args AppendEntriesArgs // construct args
	args.Term = rf.term
	args.Leader = rf.me
	args.PreLogIndex = rf.nextIndex[nodeId] - 1
	if args.PreLogIndex >= 0 {
		args.PreLogTerm = rf.log[args.PreLogIndex].Term
	} else { // -1
		args.PreLogTerm = -1
	}
	args.Entries = rf.log[rf.nextIndex[nodeId]:]
	args.CommitTo = rf.commitIndex
	 
	go func(server int, args AppendEntriesArgs) { // parallel sending
		var reply AppendEntriesReply
		ok := rf.sendAppendEntries(server, &args, &reply)
		if ok {
			rf.handleAppendEntriesResult(reply, server)
		}
	}(nodeId, args)
}
 
func (rf *Raft) sendAppendEntries(server int, args *AppendEntriesArgs, reply *AppendEntriesReply) bool {
	ok := rf.peers[server].Call("Raft.AppendEntries", args, reply) // 调用对应server的Raft.AppendEntries方法进行请求日志追加处理
	return ok
}
 
func (rf *Raft) refreshCommits() {
	for logIndex := rf.lastApplied + 1; logIndex <= rf.commitIndex; logIndex++ {
		rf.applyCh <- ApplyMsg{Index: logIndex + 1, Command: rf.log[logIndex].Command}
	}
	rf.lastApplied = rf.commitIndex
}

func (rf *Raft) persist() {
	// Your code here.
	// Example:
	w := new(bytes.Buffer)
	e := gob.NewEncoder(w)
	e.Encode(rf.term)
	e.Encode(rf.voteTo)
	e.Encode(rf.log)
	data := w.Bytes()
	rf.persister.SaveRaftState(data)
}
 
func (rf *Raft) readPersist(data []byte) {
	// Your code here.
	// Example:
	r := bytes.NewBuffer(data)
	d := gob.NewDecoder(r)
	d.Decode(&rf.term)
	d.Decode(&rf.voteTo)
	d.Decode(&rf.log)
}