Raft实验报告（2A、2B）

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

// 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()
	applyCh   chan ApplyMsg       // 服务或测试人员希望Raft发送ApplyMsg消息的通道
	applyCond *sync.Cond          // 用于通知applyCh有新的日志条目需要应用

	// 所有服务器上的持久性状态（在响应 RPC 之前更新稳定存储）
	currentTerm int        // 服务器最后一次知道的任期号（初始化为 0，持续递增）
	votedFor    int        // 在当前任期内获得投票的候选者的ID（如果没有则为 null）
	log         []LogEntry // 日志条目集；每个条目包含状态机命令以及领导者收到条目时的任期（第一个索引为 1）
	recvdIndex  int        // 已知收到的最后一个日志条目的索引，即为日志长度-1，（初始化为 0，单调增加）

	// 所有服务器上的易失性状态
	commitIndex     int           // 已知已提交的最高日志条目的索引（初始化为 0，单调增加）
	lastApplied     int           // 应用于状态机的最高日志条目的索引（初始化为 0，单调增加）
	state           int           // 服务器的状态（跟随者、候选者、领导者）
	updateTime      time.Time     // 上次收到心跳的时间，或开始选举的时间
	electionTimeout time.Duration // 选举超时时间

	// 领导者服务器上的易失状态（选举后重新初始化）
	nextIndex     []int         // 对于每个服务器，发送到该服务器的下一个日志条目的索引（初始化为领导者最后一个日志索引 + 1）
	matchIndex    []int         // 对于每个服务器，已知在服务器上复制的最高日志条目的索引（初始化为 0，单调增加）
	heartBeatTime time.Duration // 心跳超时时间
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

func Make(peers []*labrpc.ClientEnd, me int,
	persister *Persister, applyCh chan ApplyMsg) *Raft {
	rf := &Raft{}
	rf.peers = peers
	rf.persister = persister
	rf.me = me
	// 初始化状态
	rf.state = Follower
	// 心跳时间，因测试要求每秒不多于10次/秒
	// 选举超时时间，大于论文中的300ms，且需要随机化
	rf.heartBeatTime = 120 * time.Millisecond
	rf.electionTimeout = time.Duration(360+rand.Intn(360)) * time.Millisecond
	rf.updateTime = time.Now()
	rf.currentTerm = 0           // 任期号
	rf.votedFor = -1             // 未投票
	rf.log = make([]LogEntry, 1) // 索引从1开始
	rf.log[0] = LogEntry{0, nil} // 第0个日志条目为空
	rf.commitIndex = 0           // 已知已提交的最高日志条目的索引
	rf.recvdIndex = 0            // 已知收到的最后一个日志条目的索引
	rf.nextIndex = make([]int, len(rf.peers))
	rf.matchIndex = make([]int, len(rf.peers))

	rf.applyCh = applyCh
	rf.applyCond = sync.NewCond(&rf.mu)

	rf.readPersist(persister.ReadRaftState())

	go rf.ticker()   // 开辟一个goroutine，用于处理心跳和选举
	go rf.applyLog() // 开辟一个goroutine，用于应用日志
	return rf
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

func (rf *Raft) ticker() {
	for !rf.killed() {
		// 无论是何种状态，都先休眠heartBeatTime时间
		time.Sleep(rf.heartBeatTime)
		rf.mu.Lock()
		// 如果是领导者，则每过一个heartbeat时间发送一次心跳
		if rf.state == Leader {
			rf.entriesToAll()
		}

		// 如果是跟随者，则检测距离上次收到心跳的时间是否超过了选举超时时间
		// 如果超过了，则启动新选举
		if (rf.state == Follower || rf.state == Candidate) && time.Since(rf.updateTime) > rf.electionTimeout {
			rf.leaderElection()
		}
		rf.mu.Unlock()
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

type RequestVoteArgs struct {
	Term         int // 候选者的任期号
	CandidateID  int // 请求选票的候选者的ID
	LastLogIndex int // 候选者的最后日志条目的索引
	LastLogTerm  int //	候选者最后一次日志条目的任期（§5.4）
}

type RequestVoteReply struct {
	Term        int  // currentTerm，供候选者自行更新
	VoteGranted bool // true 表示候选者获得选票
}

1
2
3
4
5
6
7
8

type VoteController struct {
	wg            sync.WaitGroup   // 用于等待所有的RPC请求完成
	receivedCount int              // 用于记录已经接收到的数量
	voteCount     int              // 用于记录已经投票成功的数量
	voteCh        chan bool        // 用于通知RPC请求完成
	timeout       <-chan time.Time // 用于超时控制
	term          int              // 用于记录发出请求时的任期
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

func (rf *Raft) leaderElection() {
	// 设置当前节点为自己选举的候选者，并增加任期编号
	me := rf.me
	rf.currentTerm++
	rf.votedFor = me
	rf.state = Candidate
	rf.resetTime() // 重置选举超时计时器
	// 生成RequestVote RPC参数
	args := RequestVoteArgs{
		Term:         rf.currentTerm, // 当前任期
		CandidateID:  me,
		LastLogIndex: rf.recvdIndex,
		LastLogTerm:  rf.log[rf.recvdIndex].Term,
	}
	// 初始化投票控制结构
	voteCtrl := VoteController{
		wg:        sync.WaitGroup{},
		voteCount: 1,
		voteCh:    make(chan bool, len(rf.peers)-1),
		timeout:   time.After(rf.electionTimeout),
		term:      rf.currentTerm,
	}
	// 向其他服务器发送RequestVote RPC
	for i := range rf.peers {
		if i != me {
			voteCtrl.wg.Add(1)
			go rf.voteToSingle(i, &args, &voteCtrl)
		}
	}
	// 开辟一个goroutine，等待所有投票协程完成
	go func() {
		voteCtrl.wg.Wait()
		close(voteCtrl.voteCh)
	}()
	// 开辟一个goroutine，处理投票回复
	go rf.waitVoteReply(&voteCtrl)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

func (rf *Raft) voteToSingle(server int, args *RequestVoteArgs, voteCtrl *VoteController) {
	defer voteCtrl.wg.Done()
	reply := RequestVoteReply{}
	if rf.sendRequestVote(server, args, &reply) {
		rf.mu.Lock()
		defer rf.mu.Unlock()
		// 如果回复的任期大于当前任期，则更新当前任期
		if reply.Term > rf.currentTerm {
			rf.setNewTerm(reply.Term)
			rf.resetTime()
			rf.updateTime = time.Now()
			voteCtrl.voteCh <- false
			return
		}
		voteCtrl.voteCh <- reply.VoteGranted
	} else {
		// 如果发送失败，则认为投票失败
		voteCtrl.voteCh <- false
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58

func (rf *Raft) waitVoteReply(voteCtrl *VoteController) {
	for {
		rf.mu.Lock()
		state, curTerm := rf.state, rf.currentTerm
		rf.mu.Unlock()
		// 如果当前任期已经改变或者不是候选者状态，则不再处理
		if state == Candidate && curTerm == voteCtrl.term {
			select {
			// 使用 select 语句来同时监听两个通道：
			// 投票通道: 接收其他节点的投票回复。
			// 超时通道: 检测选举是否超时。
			case voteGranted, ok := <-voteCtrl.voteCh:
				rf.mu.Lock()
				if rf.currentTerm != voteCtrl.term || rf.state != Candidate {
					rf.mu.Unlock()
					return
				}
				// 如果投票通道关闭 (!ok)，则将 voteCtrl.voteCh 置为 nil，
				// 表示不再接收新的投票。
				if !ok {
					voteCtrl.voteCh = nil
					rf.mu.Unlock()
					return
				} else if voteGranted {
					voteCtrl.voteCount++
				}
				// 如果获得了大多数的投票，则成为领导者
				if voteCtrl.voteCount > len(rf.peers)/2 {
					rf.state = Leader
					// 向其他服务器发送日志
					// 此日志不一定是空的，因为可能领导者和其他服务器之间的日志不一致
					for i := range rf.nextIndex {
						rf.nextIndex[i] = rf.recvdIndex + 1
						rf.matchIndex[i] = 0
					}
					rf.recvdIndex = len(rf.log) - 1
					rf.entriesToAll()
					rf.mu.Unlock()
					return
				}
				rf.mu.Unlock()
			case <-voteCtrl.timeout:
                // 如果选举超时
				rf.mu.Lock()
				if rf.currentTerm != voteCtrl.term || rf.state != Candidate {
					rf.mu.Unlock()
					return
				}
				rf.setNewTerm(rf.currentTerm)
				rf.resetTime()
				rf.mu.Unlock()
				return
			}
		} else {
			return
		}
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

func (rf *Raft) RequestVote(args *RequestVoteArgs, reply *RequestVoteReply) {
	rf.mu.Lock()
	defer rf.mu.Unlock()

	// 如果请求的任期小于当前任期
	if args.Term < rf.currentTerm {
		reply.Term = rf.currentTerm
		reply.VoteGranted = false
		return
	}
	// 如果请求的任期大于当前任期
	if args.Term > rf.currentTerm {
		rf.setNewTerm(args.Term)
	}

	if (rf.votedFor == -1 || rf.votedFor == args.CandidateID) &&
		(args.LastLogTerm > rf.log[rf.recvdIndex].Term ||
			(args.LastLogTerm == rf.log[rf.recvdIndex].Term &&
				args.LastLogIndex >= rf.recvdIndex)) {
		// 如果 votedFor 为空或为 candidateId，并且候选者的日志至少和接收者一样新
		reply.VoteGranted = true
		rf.votedFor = args.CandidateID
		rf.updateTime = time.Now()
	} else {
		reply.VoteGranted = false
	}
	reply.Term = rf.currentTerm
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

type AppendEntriesArgs struct {
	Term         int        // 领导者的任期号
	LeaderID     int        // 领导者的ID
	PrevLogIndex int        // 新日志条目之前的索引
	PrevLogTerm  int        // 新日志条目之前的任期
	Entries      []LogEntry // 要附加的日志条目
	LeaderCommit int        // 领导者的commitIndex
}
Term, LeaderID, PrevLogIndex, PrevLogTerm, nil, LeaderCommit

type AppendEntriesReply struct {
	Term     int  // currentTerm，用于领导者自我更新
	Success  bool // 如果跟随者包含与 prevLogIndex 和 prevLogTerm 匹配的条目，则为 true
	Conflict bool // 如果跟随者包含与 prevLogIndex 和 prevLogTerm 不匹配的条目，则为 true
	XTerm    int  // 冲突条目的任期
	XIndex   int  // 该任期中存储的第一个索引
	XLen     int  // 跟随者的日志长度
}

1
2
3
4
5
6
7
8
9

type AppendController struct {
	wg            sync.WaitGroup   // 用于等待所有的RPC请求完成
	appendCount   int              // 用于记录已经append成功的数量
	appendCh      chan bool        // 用于通知RPC请求完成
	timeout       <-chan time.Time // 用于超时控制
	term          int              // 用于记录发出请求时的任期
	recvdIndex    int              // 用于记录当前接收到的最大日志索引
	commitIndex   int              // 用于记录当前已经提交的最大日志索引
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

func (rf *Raft) Start(command interface{}) (int, int, bool) {
	rf.mu.Lock()
	defer rf.mu.Unlock()
	me := rf.me
	if rf.state != Leader {
		return 0, 0, false
	}
	curTerm := rf.currentTerm
	rf.log = append(rf.log, LogEntry{Term: curTerm, Command: command})
	rf.recvdIndex++
	recvdIndex := rf.recvdIndex
	rf.nextIndex[me] = recvdIndex + 1
	rf.matchIndex[me] = recvdIndex
	rf.entriesToAll()
	return recvdIndex, curTerm, true
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

func (rf *Raft) entriesToAll() {
	term := rf.currentTerm
	appendCtrl := AppendController{
		wg:            sync.WaitGroup{},
		appendCount:   1,
		appendCh:      make(chan bool, len(rf.peers)-1),
		timeout:       time.After(rf.heartBeatTime),
		term:          term,
		recvdIndex:    rf.recvdIndex,
		commitIndex:   rf.commitIndex,
	}

	for i := range rf.peers {
		if i != rf.me {
			appendCtrl.wg.Add(1)
			var logEntry []LogEntry
			if rf.nextIndex[i] > rf.recvdIndex {
				logEntry = nil
			} else {
				logEntry = rf.log[rf.nextIndex[i]:]
			}
			prevLogIndex := rf.nextIndex[i] - 1
			args := AppendEntriesArgs{
				Term:         rf.currentTerm,
				LeaderID:     rf.me,
				PrevLogIndex: prevLogIndex,
				PrevLogTerm:  rf.log[prevLogIndex].Term,
				Entries:      logEntry,
				LeaderCommit: rf.commitIndex,
			}
			go rf.entriesToSingle(i, &args, &appendCtrl)
		}
	}

	go func() {
		appendCtrl.wg.Wait()
		close(appendCtrl.appendCh)
	}()
	go rf.waitAppendReply(&appendCtrl, term)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

func (rf *Raft) entriesToSingle(server int, args *AppendEntriesArgs, appendCtrl *AppendController) {
	defer appendCtrl.wg.Done()
	reply := AppendEntriesReply{}
	if rf.sendAppendEntries(server, args, &reply) {
		rf.mu.Lock()
		defer rf.mu.Unlock()
		if reply.Term > rf.currentTerm {
			rf.setNewTerm(reply.Term)
			return
		}
		if rf.currentTerm == args.Term {
			if reply.Success {
				// 有时候可能会连续重复发送相同的日志，导致nextIndex不断增加，所以需要取最小值
				rf.nextIndex[server] = min(rf.recvdIndex+1, rf.nextIndex[server]+len(args.Entries))
				rf.matchIndex[server] = rf.nextIndex[server] - 1
				appendCtrl.appendCh <- true
			} else {
				rf.nextIndex[server] = reply.XIndex
				appendCtrl.appendCh <- false
			}
		}
	} else {
		// 如果是因为网络原因导致的失败，则等下次心跳或append时再次尝试
		appendCtrl.appendCh <- false
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

func (rf *Raft) waitAppendReply(appendCtrl *AppendController, term int) {
	for {
		curTerm, isLeader := rf.GetState()
		if curTerm == term && isLeader {
			select {
			case success, ok := <-appendCtrl.appendCh:
				rf.mu.Lock()
				if rf.currentTerm != term || rf.state != Leader {
					rf.mu.Unlock()
					return
				}
				appendCtrl.receivedCount++
				if !ok {
					appendCtrl.appendCh = nil
				} else if success {
					appendCtrl.appendCount++
				}
				if appendCtrl.appendCount > len(rf.peers)/2 {
					// 因为在等待日志提交的过程中，可能有新的日志被leader接收，所以实际上commitIndex应当是旧的recvdIndex
					preCommitIndex := appendCtrl.commitIndex
					rf.commitIndex = appendCtrl.recvdIndex
					if preCommitIndex != rf.commitIndex {
						rf.applyCondSignal()
					}
					rf.mu.Unlock()
					return
				}
				rf.mu.Unlock()
			case <-appendCtrl.timeout:
				return
			}
		} else {
			return
		}
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50

func (rf *Raft) AppendEntries(args *AppendEntriesArgs, reply *AppendEntriesReply) {
	rf.mu.Lock()
	defer rf.mu.Unlock()
	me := rf.me
	reply.Success = false
	reply.Term = rf.currentTerm
	if args.Term > rf.currentTerm {
		rf.setNewTerm(args.Term)
		rf.votedFor = args.LeaderID
	}
	if args.Term < rf.currentTerm {
		return
	}
	if rf.state == Candidate {
		rf.state = Follower
	}

	rf.updateTime = time.Now()
	// 如果日志在 prevLogIndex 处不匹配，则返回 false
	if args.PrevLogIndex > rf.recvdIndex || rf.log[args.PrevLogIndex].Term != args.PrevLogTerm {
		if args.PrevLogIndex <= rf.recvdIndex {
			// 找到冲突条目的任期和该任期中它存储的第一个索引
			reply.XTerm = rf.log[args.PrevLogIndex].Term
			for i := args.PrevLogIndex; i >= 0; i-- {
				if rf.log[i].Term != reply.XTerm {
					reply.XIndex = i + 1
					break
				}
			}
		} else {
			reply.XTerm = rf.currentTerm
			reply.XIndex = rf.recvdIndex + 1
		}
		reply.Term = rf.currentTerm
		return
	}

	// 如果一个已经存在的条目和新条目在相同的索引位置有相同的任期号和索引值，则复制其后的所有条目
	rf.log = rf.log[:args.PrevLogIndex+1]
	rf.log = append(rf.log, args.Entries...)
	rf.recvdIndex = len(rf.log) - 1

	if args.LeaderCommit > rf.commitIndex {
		rf.commitIndex = min(args.LeaderCommit, rf.recvdIndex)
		rf.applyCondSignal()
	}
	reply.Term = rf.currentTerm
	reply.Success = true
	reply.Conflict = false
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

func (rf *Raft) applyCondSignal() {
	rf.applyCond.Broadcast()
}

func (rf *Raft) applyLog() {
	rf.mu.Lock()
	defer rf.mu.Unlock()
	for !rf.killed() {
		if rf.commitIndex > rf.lastApplied && len(rf.log)-1 > rf.lastApplied {
			rf.lastApplied++
			applyMsg := ApplyMsg{
				CommandValid: true,
				Command:      rf.log[rf.lastApplied].Command,
				CommandIndex: rf.lastApplied,
			}
			rf.mu.Unlock()
			rf.applyCh <- applyMsg
			rf.mu.Lock()
		} else {
			rf.applyCond.Wait()
		}
	}
}

1
2
3

#!/usr/bin/zsh

cat test_test.go| grep "2A" | sed 's/(/ /g'|awk '/func/ {printf "%s ",$2;}' | xargs dstest -o .1224_2 -n 100 -p 50 -r

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

def run_tests(
    tests: List[str],
    sequential: bool       = typer.Option(False,  '--sequential',      '-s',    help='Run all test of each group in order'),
    workers: int           = typer.Option(1,      '--workers',         '-p',    help='Number of parallel tasks'),
    iterations: int        = typer.Option(10,     '--iter',            '-n',    help='Number of iterations to run'),
    output: Optional[Path] = typer.Option(None,   '--output',          '-o',    help='Output path to use'),
    verbose: int           = typer.Option(0,      '--verbose',         '-v',    help='Verbosity level', count=True),
    archive: bool          = typer.Option(False,  '--archive',         '-a',    help='Save all logs intead of only failed ones'),
    race: bool             = typer.Option(False,  '--race/--no-race',  '-r/-R', help='Run with race checker'),
    loop: bool             = typer.Option(False,  '--loop',            '-l',    help='Run continuously'),
    growth: int            = typer.Option(10,     '--growth',          '-g',    help='Growth ratio of iterations when using --loop'),
    timing: bool           = typer.Option(False,   '--timing',          '-t',    help='Report timing, only works on macOS'),
    # fmt: on
):

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

Goroutine 7038 (running) created at:
  MyRaft/raft.(*Raft).entriesToAll()
      /home/firefly/Codes/go/myraft/raft/append_entries.go:84 +0x713
  MyRaft/raft.(*Raft).ticker()
      /home/firefly/Codes/go/myraft/raft/raft.go:225 +0xa6
  MyRaft/raft.Make.gowrap1()
      /home/firefly/Codes/go/myraft/raft/raft.go:273 +0x33

Goroutine 7051 (finished) created at:
  MyRaft/labrpc.(*Network).processReq()
      /home/firefly/Codes/go/myraft/labrpc/labrpc.go:242 +0x28a
  MyRaft/labrpc.MakeNetwork.func1.gowrap1()
      /home/firefly/Codes/go/myraft/labrpc/labrpc.go:160 +0x9c
      
--- FAIL: TestBackup2B (20.02s)
    testing.go:1398: race detected during execution of test
FAIL
exit status 1
FAIL	MyRaft/raft	20.202s

1
2
3
4
5

--- FAIL: TestCount2B (1.52s)
    test_test.go:643: too many or few RPCs (32) to elect initial leader
FAIL
exit status 1
FAIL	MyRaft/raft	1.531s

1
2
3
4
5

--- FAIL: TestReElection2A (9.15s)
    config.go:459: expected one leader, got none
FAIL
exit status 1
FAIL	MyRaft/raft	9.163s