6.5840-lab2-raft记录

• 2a
• 2b

lab2a主要是完成日志选举和心跳机制
只需要修改/src/raft/raft.go中的内容

• 可交互的raft
• goland的代码自动格式化
  
  测试结果如下
  

考虑测试中有一些不确定因素，可以考虑使用课程助教提供的实现反复测试的脚本，链接在此 。
使用方法：100表示执行100次，4表示4个tester子进程测试

./test-many.sh 100 4 2A

代码如下

package raft

import (
	"math/rand"
	"sync"
	"sync/atomic"
	"time"
	//	"6.5840/labgob"
	"6.5840/labrpc"
)

type state int

// 自定义常量
const (
	FOLLOWER = iota
	CANDIDATE
	LEADER
)
const (
	TimeOut = iota
	VoteDone
	VoteSuccess
)
const (
	INIT_TERM              = 0 //初始的任期
	NULL                   = -1
	VOTE_NOBODY            = -1
	HEARTBEAT_GAP          = 130  //心跳间隔(ms)
	ELECTION_TIMEOUT_FLOOR = 700  //选举超时时间下限(ms)
	ELECTION_TIMEOUT_CEIL  = 1000 //选举超时时间上限(ms)
	RETRY_NUM              = 1    //3次超时重试
	ELECTION_DURATION      = 3000
)

type ApplyMsg struct {
	CommandValid bool
	Command      interface{}
	CommandIndex int

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

type Entry struct {
}

// Raft
// 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()

	// Your data here (2A, 2B, 2C).
	// Look at the paper's Figure 2 for a description of what
	// state a Raft server must maintain.F
	curTerm         int           //当前结点任期
	votedFor        int           //给谁投了票
	curState        state         //当前状态
	msgHandler      chan ApplyMsg //处理传入的请求
	electionTimeout int64         //本结点的选举超时时间
	logs            []Entry       //日志

	ticketGot int //选举阶段获得的票数

	timeLastReceiveHeartBeat time.Time //上一次收到心跳的时间
	timeLastSendHeartBeat    time.Time //作为leader上一次发送心跳的时间
	timeStartElection        time.Time //选举开始时间
}

// GetState
// return currentTerm and whether this server
// believes it is the leader.
func (rf *Raft) GetState() (int, bool) {
	// Your code here (2A).
	rf.mu.Lock()
	defer rf.mu.Unlock()
	return rf.curTerm, rf.curState == LEADER
}

func (rf *Raft) persist() {
	// Your code here (2C).
	// Example:
	// w := new(bytes.Buffer)
	// e := labgob.NewEncoder(w)
	// e.Encode(rf.xxx)
	// e.Encode(rf.yyy)
	// raftstate := w.Bytes()
	// rf.persister.Save(raftstate, nil)
}

// restore previously persisted state.
func (rf *Raft) readPersist(data []byte) {
	if data == nil || len(data) < 1 { // bootstrap without any state?
		return
	}
	// Your code here (2C).
	// Example:
	// r := bytes.NewBuffer(data)
	// d := labgob.NewDecoder(r)
	// var xxx
	// var yyy
	// if d.Decode(&xxx) != nil ||
	//    d.Decode(&yyy) != nil {
	//   error...
	// } else {
	//   rf.xxx = xxx
	//   rf.yyy = yyy
	// }
}

// Snapshot
// the service says it has created a snapshot that has
// all info up to and including index. this means the
// service no longer needs the log through (and including)
// that index. Raft should now trim its log as much as possible.
func (rf *Raft) Snapshot(index int, snapshot []byte) {
	// Your code here (2D).

}

type ReqVoteArgs struct {
	// Your data here (2A, 2B).
	CandidateID   int
	CandidateTerm int
}

type ReqVoteReply struct {
	// Your data here (2A).
	VoteGranted  bool
	FollowerTerm int
}

type ReqAppendEntriesArgs struct {
	LeaderTerm int
	LeaderID   int
}
type ReqAppendEntriesReply struct {
	Success      bool
	FollowerTerm int
}

func (rf *Raft) beFollower(newTerm int) {
	//fmt.Printf("结点 %d 之前的身份是%d\n", rf.me, rf.curState)
	rf.curTerm = newTerm
	rf.curState = FOLLOWER
	rf.votedFor = VOTE_NOBODY
	rf.refreshElectionTimer()
	rf.refreshSendHeartBeatTimer()
}

func (rf *Raft) beCandidate() {
	// 可能会变多次
	if rf.curState != CANDIDATE {
		// 任期+1
		//fmt.Printf("结点 %d 任期+1\n", rf.me)
		rf.curTerm += 1
		//变状态
		rf.curState = CANDIDATE
	}
	// 先给自己投票，防止待会给别人投了
	rf.votedFor = rf.me
	// 初始化自己的票数
	rf.ticketGot = 1
}

func (rf *Raft) beLeader() {
	rf.curState = LEADER
	rf.votedFor = VOTE_NOBODY
	// 这里不用重置，让马上发心跳
	// rf.timeLastSendHeartBeat = time.Now()
}

// 异步发消息
func (rf *Raft) sendAndHandleRequestVote(server int) {
	rf.mu.Lock()
	tmpTerm := rf.curTerm
	rf.mu.Unlock()
	reqArgs := ReqVoteArgs{
		CandidateID:   rf.me,
		CandidateTerm: tmpTerm,
	}
	reply := ReqVoteReply{}
	// 重试发请求
	ok := false
	tmpRetryNum := 0
	for ok == false && tmpRetryNum < RETRY_NUM {
		ok = rf.peers[server].Call("Raft.HandleVote", &reqArgs, &reply)
		tmpRetryNum += 1
	}
	rf.mu.Lock()
	if ok == false {
		//fmt.Printf("结点 %d 发请求给 %d 失败\n", reqArgs.CandidateID, server)
		// 对方结点宕机也算没同意投票
	} else {
		if rf.curState == CANDIDATE {
			// 别人同意投票
			if reply.VoteGranted {
				// 可能会有很久以前的请求，需要丢弃他们
				if reply.FollowerTerm < rf.curTerm {
					//不处理
				} else if reply.FollowerTerm == rf.curTerm {
					rf.ticketGot += 1
					// 查看投票结果，超过结点数一半就算赢
					//log.Printf("结点 %d 当前票数：%d/%d\n", rf.me, rf.ticketGot, len(rf.peers))
					if rf.ticketGot*2 > len(rf.peers) {
						//log.Printf("------结点 %d(%d) : 变成leader------\n", rf.me, rf.curTerm)
						rf.beLeader()
					}
				}
			} else {
				// 我out了
				if reply.FollowerTerm > rf.curTerm {
					rf.beFollower(reply.FollowerTerm)
					rf.mu.Unlock()
				} else {
					//不处理
				}
			}
		}
	}
	rf.mu.Unlock()
}

func (rf *Raft) HandleVote(args *ReqVoteArgs, reply *ReqVoteReply) {
	// Your code here (2A, 2B).
	//log.Printf("结点(id:%d,term:%d)收到(id:%d,term:%d)的投票请求\n", rf.me, rf.curTerm, args.CandidateID, args.CandidateTerm)
	rf.mu.Lock()
	if rf.curTerm > args.CandidateTerm {
		//不投票
		reply.VoteGranted = false
		reply.FollowerTerm = rf.curTerm
		//log.Printf("拒绝")
	} else if rf.curTerm < args.CandidateTerm {
		// 投票
		reply.VoteGranted = true
		reply.FollowerTerm = args.CandidateTerm
		//log.Printf("同意")
		rf.beFollower(args.CandidateTerm)
		rf.votedFor = args.CandidateID
	} else {
		// 先投票
		if rf.votedFor == VOTE_NOBODY || rf.votedFor == args.CandidateID {
			reply.VoteGranted = true
			reply.FollowerTerm = args.CandidateTerm
			//log.Printf("同意\n")
			rf.beFollower(args.CandidateTerm)
			rf.votedFor = args.CandidateID
		}
	}
	rf.mu.Unlock()
}

// 选举超过一段时间后的操作
func (rf *Raft) electionTimer() {
	time.Sleep(time.Duration(ELECTION_DURATION) * time.Millisecond)
	rf.mu.Lock()
	if rf.curState == CANDIDATE {
		rf.beFollower(rf.curTerm - 1)
	}
	rf.mu.Unlock()
}

// 异步发心跳
func (rf *Raft) sendAppendEntries(server int) {
	rf.mu.Lock()
	tmpTerm := rf.curTerm
	rf.mu.Unlock()
	reqArgs := ReqAppendEntriesArgs{
		LeaderTerm: tmpTerm,
		LeaderID:   rf.me,
	}
	reply := ReqAppendEntriesReply{}
	// 重试发请求
	ok := false
	tmpRetryNum := 0
	for ok == false && tmpRetryNum < RETRY_NUM {
		ok = rf.peers[server].Call("Raft.HandleAppendEntries", &reqArgs, &reply)
		tmpRetryNum += 1
	}
	if ok {
		rf.mu.Lock()
		if reply.Success {

		} else {
			rf.beFollower(reply.FollowerTerm)
		}
		rf.mu.Unlock()
	}
}

func (rf *Raft) HandleAppendEntries(args *ReqAppendEntriesArgs, reply *ReqAppendEntriesReply) {
	// Your code here (2A, 2B).
	rf.mu.Lock()
	//fmt.Printf("结点 (%d,%d) 收到结点 (%d,%d) 的心跳\n", rf.me, rf.curTerm, args.LeaderID, args.LeaderTerm)
	if rf.curTerm > args.LeaderTerm {
		reply.FollowerTerm = rf.curTerm
		reply.Success = false
	} else {
		rf.beFollower(args.LeaderTerm)
		reply.FollowerTerm = args.LeaderTerm
		reply.Success = true
		//log.Printf("结点 %d 重置心跳\n", rf.me)
	}
	rf.mu.Unlock()
}

func (rf *Raft) Start(command interface{}) (int, int, bool) {
	index := -1
	term := -1
	isLeader := true
	//if isLeader ==false{
	//	return NULL,NULL,false
	//}
	// Your code here (2B).

	return index, term, isLeader
}

func (rf *Raft) Kill() {
	atomic.StoreInt32(&rf.dead, 1)
	// Your code here, if desired.
}

func (rf *Raft) killed() bool {
	z := atomic.LoadInt32(&rf.dead)
	return z == 1
}

// 选举和发送心跳计时器
func (rf *Raft) ticker() {
	for rf.killed() == false {
		// pause for a random amount of time between 50 and 350 milliseconds.
		// 稍微卡一下时间
		ms := 50 + (rand.Int63() % 50)
		time.Sleep(time.Duration(ms) * time.Millisecond)
		// Your code here (2A)
		// Check if a leader election should be started.
		rf.mu.Lock()
		// 只有follower才会想抢
		if rf.curState == FOLLOWER {
			// 心跳超时,开始选举
			//fmt.Printf("节点%d距上一次心跳%d,预设超时为%d\n", rf.me, time.Now().Sub(rf.timeLastReceiveHeartBeat).Milliseconds(), rf.electionTimeout)
			if time.Now().Sub(rf.timeLastReceiveHeartBeat).Milliseconds() > rf.electionTimeout {
				rf.beCandidate()
				//log.Printf("结点%d开始选举,term:%d\n", rf.me, rf.curTerm)
				// 广播投票请求
				for serverID := range rf.peers {
					if serverID == rf.me {
						continue
					}
					go rf.sendAndHandleRequestVote(serverID)
				}
				go rf.electionTimer()
			}
		} else if rf.curState == LEADER {
			//leader发心跳
			//fmt.Printf("leader结点%d发心跳\n", rf.me)
			//fmt.Printf("leader距上一次心跳%v，阈值%v\n", time.Now().Sub(rf.timeLastSendHeartBeat).Milliseconds(), HEARTBEAT_GAP)
			if time.Now().Sub(rf.timeLastSendHeartBeat).Milliseconds() > HEARTBEAT_GAP {
				for idx := range rf.peers {
					if idx == rf.me {
						continue
					}
					go rf.sendAppendEntries(idx)
				}
				//重置计时
				rf.refreshSendHeartBeatTimer()
			}
		}
		rf.mu.Unlock()
	}
}

func (rf *Raft) refreshElectionTimer() {
	rf.timeLastReceiveHeartBeat = time.Now()
}

func (rf *Raft) refreshSendHeartBeatTimer() {
	rf.timeLastSendHeartBeat = time.Now()
}

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 (2A, 2B, 2C).
	rf.logs = make([]Entry, 1024)
	rf.msgHandler = applyCh
	rf.electionTimeout = int64(ELECTION_TIMEOUT_FLOOR + rand.Intn(ELECTION_TIMEOUT_CEIL-ELECTION_TIMEOUT_FLOOR))
	//fmt.Printf("超时时间为%d\n", rf.electionTimeout)
	rf.beFollower(INIT_TERM)

	// initialize from state persisted before a crash
	rf.readPersist(persister.ReadRaftState())
	// start ticker goroutine to start elections
	go rf.ticker()
	return rf
}

内容