Raft 协议源码阅读

我会带着你远行 2022-12-16 04:00 207阅读 0赞

## Raft源码学习 ##

### 背景 ###

看本篇博客，需要先对Raft协议本身有了解。然后再开始看源码，会有更深入的体会。不建议对Raft协议没有任何的认识就可以直接开始看本篇文章。Raft协议的中英文论文已经有相关比较好的博客。参考引用。  
我看的是githab的[Raft][]代码。分析这个代码是如何实现Raft协议的。

### 体会 ###

看完算法，再去看源码的实现。功能就真的是按照算法一比一去实现的。这是我第一次看源码如此轻松。读源码就像是读小说的感觉一样。感觉还是得看论文可以最快的认识一个算法和协议。全文没有一句废话。

### 代码分析过程 ###

整个项目的核心代码在Raft.go里面。我们根据Raft协议本身的一些情况。然后对照代码分析。

#### Leader选举过程分析，三种状态转化 ####

![在这里插入图片描述][20201018164542348.png_pic_center]

// run is a long running goroutine that runs the Raft FSM.
    func (r *Raft) run() {
    	for {
    		// Check if we are doing a shutdown
    		select {
    		case <-r.shutdownCh:
    			// Clear the leader to prevent forwarding
    			r.setLeader("")
    			return
    		default:
    		}
    
    		// Enter into a sub-FSM、
           // 循环判断状态，是具体的状态，执行具体的逻辑
    		switch r.getState() {
    		case Follower:
    			r.runFollower()
    		case Candidate:
    			r.runCandidate()
    		case Leader:
    			r.runLeader()
    		}
    	}
    }

#### r.runFollower 方法分析 ####

Raft使用心跳机制来触发leader选举。当server启动的时候是处于follower状态，当它可以收到来自leader或者candidate的有效RPC请求时就会保持follower的状态。Leader发送周期性的心跳(不含日志的AppendEntries RPC)给所有的follower来确保自己的权威。如果一个follower一段时间(称为election timeout)没有收到消息，它就会假定leader失效并开始新的选举。  
为了开始新一轮选举，follower会提高自己当前的term并转为candidate状态

// runFollower runs the FSM for a follower.
    func (r *Raft) runFollower() {
    	didWarn := false
    	r.logger.Info("entering follower state", "follower", r, "leader", r.Leader())
    	metrics.IncrCounter([]string{"raft", "state", "follower"}, 1)
    	heartbeatTimer := randomTimeout(r.conf.HeartbeatTimeout)
    
    	for r.getState() == Follower {
    		select {
    		case rpc := <-r.rpcCh:
    			r.processRPC(rpc)
    
    		case c := <-r.configurationChangeCh:
    			// Reject any operations since we are not the leader
    			c.respond(ErrNotLeader)
    
    		... //省略部分不重要代码,都是不需要处理的Reponse
    
    		case <-heartbeatTimer:
    			// Restart the heartbeat timer
    			heartbeatTimer = randomTimeout(r.conf.HeartbeatTimeout)
    
    			// Check if we have had a successful contact
    			lastContact := r.LastContact()
    			if time.Now().Sub(lastContact) < r.conf.HeartbeatTimeout {
    				continue
    			}
    
    			// Heartbeat failed! Transition to the candidate state
    			lastLeader := r.Leader()
    			r.setLeader("")
    
    			if r.configurations.latestIndex == 0 {
    				if !didWarn {
    					r.logger.Warn("no known peers, aborting election")
    					didWarn = true
    				}
    			} else if r.configurations.latestIndex == r.configurations.committedIndex &&
    				!hasVote(r.configurations.latest, r.localID) {
    				if !didWarn {
    					r.logger.Warn("not part of stable configuration, aborting election")
    					didWarn = true
    				}
    			} else {
                
                   //心跳超时了，设置自己为Candidate。并退出Follower的循环
    				r.logger.Warn("heartbeat timeout reached, starting election", "last-leader", lastLeader)
    				metrics.IncrCounter([]string{"raft", "transition", "heartbeat_timeout"}, 1)
    				r.setState(Candidate)
    				return
    			}
    
    		case <-r.shutdownCh:
    			return
    		}
    	}
    }

看一下 randomTimeout方法。这里是心跳定时的逻辑

// randomTimeout returns a value that is between the minVal and 2x minVal.
    func randomTimeout(minVal time.Duration) <-chan time.Time {
    	if minVal == 0 {
    		return nil
    	}
    	extra := (time.Duration(rand.Int63()) % minVal)
    	return time.After(minVal + extra)
    }

#### r.runCandidate 方法分析 ####

为了开始新一轮选举，follower会提高自己当前的term并转为candidate状态。它会先给自己投一票然后并行向集群中的其他server发出RequestVote RPC，candidate会保持这个状态，直到下面三种事情之一发生：

(a) 赢得选举。当candidate收到了集群中相同term的多数节点的赞成票时就会选举成功，每一个server在给定的term内至多只能投票给一个candidate，先到先得。收到多数节点的选票可以确保在一个term内至多只能有一个leader选出。一旦一个candidate赢得选举，它就会成为leader。它之后会发送心跳消息来建立自己的权威，并阻止新的选举。

(b) 另一个server被确定为leader。在等待投票的过程中，candidate可能收到来自其他server的AppendEntries RPC，声明它才是leader。如果RPC中的term大于等于candidate的current term，candidate就会认为这个leader是合法的并转为follower状态。如果RPC中的term比自己当前的小，将会拒绝这个请求并保持candidate状态。

// runCandidate runs the FSM for a candidate.
    func (r *Raft) runCandidate() {
    	r.logger.Info("entering candidate state", "node", r, "term", r.getCurrentTerm()+1)
    	metrics.IncrCounter([]string{"raft", "state", "candidate"}, 1)
    
        // 先投自己一票。
    	// Start vote for us, and set a timeout
    	voteCh := r.electSelf()
    
    	// Make sure the leadership transfer flag is reset after each run. Having this
    	// flag will set the field LeadershipTransfer in a RequestVoteRequst to true,
    	// which will make other servers vote even though they have a leader already.
    	// It is important to reset that flag, because this priviledge could be abused
    	// otherwise.
    	defer func() { r.candidateFromLeadershipTransfer = false }()
    
        // 选举超时时间
    	electionTimer := randomTimeout(r.conf.ElectionTimeout)
    
    	// Tally the votes, need a simple majority
    	grantedVotes := 0
    	votesNeeded := r.quorumSize()        // quorumSize 等于 oters/2 + 1
    	r.logger.Debug("votes", "needed", votesNeeded)
    
    	for r.getState() == Candidate {
    		select {
    		case rpc := <-r.rpcCh:
    			r.processRPC(rpc)
    
    		case vote := <-voteCh:
    			// Check if the term is greater than ours, bail
               // 如果接受到Vote请求。发现他的term大于自己的term。把自己改为Follower。修改自己的term
    			if vote.Term > r.getCurrentTerm() {
    				r.logger.Debug("newer term discovered, fallback to follower")
    				r.setState(Follower)
    				r.setCurrentTerm(vote.Term)
    				return
    			}
    
    			// Check if the vote is granted
               // 如果是同意投票了。
    			if vote.Granted {
    				grantedVotes++
    				r.logger.Debug("vote granted", "from", vote.voterID, "term", vote.Term, "tally", grantedVotes)
    			}
    
    			// Check if we've become the leader
                // 获得投投票数超过一半。设置自己为leader
    			if grantedVotes >= votesNeeded {
    				r.logger.Info("election won", "tally", grantedVotes)
    				r.setState(Leader)
    				r.setLeader(r.localAddr)
    				return
    			}
    
    		case c := <-r.configurationChangeCh:
    			// Reject any operations since we are not the leader
    			c.respond(ErrNotLeader)
    
    		... // 省略部分代码
    
    		case <-electionTimer:
    			// Election failed! Restart the election. We simply return,
    			// which will kick us back into runCandidate
    			r.logger.Warn("Election timeout reached, restarting election")
    			return
    
    		case <-r.shutdownCh:
    			return
    		}
    	}
    }

electSelf 方法

func (r *Raft) electSelf() <-chan *voteResult {
    	// Create a response channel
    	respCh := make(chan *voteResult, len(r.configurations.latest.Servers))
    
    	// Increment the term
    	r.setCurrentTerm(r.getCurrentTerm() + 1)
    
    	// Construct the request
    	lastIdx, lastTerm := r.getLastEntry()
    	req := &RequestVoteRequest{
    		RPCHeader:          r.getRPCHeader(),
    		Term:               r.getCurrentTerm(),
    		Candidate:          r.trans.EncodePeer(r.localID, r.localAddr),
    		LastLogIndex:       lastIdx,
    		LastLogTerm:        lastTerm,
    		LeadershipTransfer: r.candidateFromLeadershipTransfer,
    	}
    
    	// Construct a function to ask for a vote
    	askPeer := func(peer Server) {
    		r.goFunc(func() {
    			defer metrics.MeasureSince([]string{"raft", "candidate", "electSelf"}, time.Now())
    			resp := &voteResult{voterID: peer.ID}
    			err := r.trans.RequestVote(peer.ID, peer.Address, req, &resp.RequestVoteResponse)
    			if err != nil {
    				r.logger.Error("failed to make requestVote RPC",
    					"target", peer,
    					"error", err)
    				resp.Term = req.Term
    				resp.Granted = false
    			}
    			respCh <- resp
    		})
    	}
    
    	// For each peer, request a vote
    	for _, server := range r.configurations.latest.Servers {
    		if server.Suffrage == Voter {
    			if server.ID == r.localID {
    				// Persist a vote for ourselves
    				if err := r.persistVote(req.Term, req.Candidate); err != nil {
    					r.logger.Error("failed to persist vote", "error", err)
    					return nil
    				}
    				// Include our own vote
    				respCh <- &voteResult{
    					RequestVoteResponse: RequestVoteResponse{
    						RPCHeader: r.getRPCHeader(),
    						Term:      req.Term,
    						Granted:   true,
    					},
    					voterID: r.localID,
    				}
    			} else {
    				askPeer(server)
    			}
    		}
    	}
        //返回到外面的voteCh
    	return respCh
    }

#### r.runLeader 方法分析 ####

1.  初始化leader的状态
2.  启动日志复制
3.  循环执行Leader的任务

// the leaderLoop for the hot loop.
    func (r *Raft) runLeader() {
    	r.logger.Info("entering leader state", "leader", r)
    	metrics.IncrCounter([]string{"raft", "state", "leader"}, 1)
    
    	// Notify that we are the leader
    	overrideNotifyBool(r.leaderCh, true)
    
    	// Push to the notify channel if given
    	if notify := r.conf.NotifyCh; notify != nil {
    		select {
    		case notify <- true:
    		case <-r.shutdownCh:
    		}
    	}
    
    	// setup leader state. This is only supposed to be accessed within the
    	// leaderloop.
    	r.setupLeaderState()
    
    	// Start a replication routine for each peer
    	r.startStopReplication()
    
    	// Dispatch a no-op log entry first. This gets this leader up to the latest
    	// possible commit index, even in the absence of client commands. This used
    	// to append a configuration entry instead of a noop. However, that permits
    	// an unbounded number of uncommitted configurations in the log. We now
    	// maintain that there exists at most one uncommitted configuration entry in
    	// any log, so we have to do proper no-ops here.
    	noop := &logFuture{
    		log: Log{
    			Type: LogNoop,
    		},
    	}
    	r.dispatchLogs([]*logFuture{noop})
    
    	// Sit in the leader loop until we step down
    	r.leaderLoop()
    }

leaderLoop() 循环执行

1.  接收日志提交
2.  定期验证是否是leader
3.  日志申请
4.  续约

// leaderLoop is the hot loop for a leader. It is invoked
    // after all the various leader setup is done.
    func (r *Raft) leaderLoop() {
    	// stepDown is used to track if there is an inflight log that
    	// would cause us to lose leadership (specifically a RemovePeer of
    	// ourselves). If this is the case, we must not allow any logs to
    	// be processed in parallel, otherwise we are basing commit on
    	// only a single peer (ourself) and replicating to an undefined set
    	// of peers.
    	stepDown := false
    	// 租约超时
    	lease := time.After(r.conf.LeaderLeaseTimeout)
    
    	for r.getState() == Leader {
    		select {
    		case rpc := <-r.rpcCh:
    			r.processRPC(rpc)
    
    		case <-r.leaderState.stepDown:
    			r.setState(Follower)
    		// leader的转交。不是重点。先不看
    		case future := <-r.leadershipTransferCh:
           ... //省略代码
    		// 日志的提交。	
    		case <-r.leaderState.commitCh:
    			// Process the newly committed entries
    			oldCommitIndex := r.getCommitIndex()
    			commitIndex := r.leaderState.commitment.getCommitIndex()
    			r.setCommitIndex(commitIndex)
    
    			// New configration has been committed, set it as the committed
    			// value.
    			if r.configurations.latestIndex > oldCommitIndex &&
    				r.configurations.latestIndex <= commitIndex {
    				r.setCommittedConfiguration(r.configurations.latest, r.configurations.latestIndex)
    				if !hasVote(r.configurations.committed, r.localID) {
    					stepDown = true
    				}
    			}
    
    			start := time.Now()
    			var groupReady []*list.Element
    			var groupFutures = make(map[uint64]*logFuture)
    			var lastIdxInGroup uint64
    
    			// Pull all inflight logs that are committed off the queue.
    			for e := r.leaderState.inflight.Front(); e != nil; e = e.Next() {
    				commitLog := e.Value.(*logFuture)
    				idx := commitLog.log.Index
    				if idx > commitIndex {
    					// Don't go past the committed index
    					break
    				}
    
    				// Measure the commit time
    				metrics.MeasureSince([]string{"raft", "commitTime"}, commitLog.dispatch)
    				groupReady = append(groupReady, e)
    				groupFutures[idx] = commitLog
    				lastIdxInGroup = idx
    			}
    
    			// Process the group
    			if len(groupReady) != 0 {
    				r.processLogs(lastIdxInGroup, groupFutures)
    
    				for _, e := range groupReady {
    					r.leaderState.inflight.Remove(e)
    				}
    			}
    
    			// Measure the time to enqueue batch of logs for FSM to apply
    			metrics.MeasureSince([]string{"raft", "fsm", "enqueue"}, start)
    
    			// Count the number of logs enqueued
    			metrics.SetGauge([]string{"raft", "commitNumLogs"}, float32(len(groupReady)))
    
    			if stepDown {
    				if r.conf.ShutdownOnRemove {
    					r.logger.Info("removed ourself, shutting down")
    					r.Shutdown()
    				} else {
    					r.logger.Info("removed ourself, transitioning to follower")
    					r.setState(Follower)
    				}
    			}
    			
    			
    		// 定期验证是否还可以作为leader	
    		case v := <-r.verifyCh:
    			if v.quorumSize == 0 {
    				// Just dispatched, start the verification
    				r.verifyLeader(v)
    
    			} else if v.votes < v.quorumSize {
    				// Early return, means there must be a new leader
    				r.logger.Warn("new leader elected, stepping down")
    				r.setState(Follower)
    				delete(r.leaderState.notify, v)
    				for _, repl := range r.leaderState.replState {
    					repl.cleanNotify(v)
    				}
    				v.respond(ErrNotLeader)
    
    			} else {
    				// Quorum of members agree, we are still leader
    				delete(r.leaderState.notify, v)
    				for _, repl := range r.leaderState.replState {
    					repl.cleanNotify(v)
    				}
    				v.respond(nil)
    			}
    
            ... // 省略部分代码
    
    		case b := <-r.bootstrapCh:
    			b.respond(ErrCantBootstrap)
    
    		// 提交日志申请	
    		case newLog := <-r.applyCh:
    			if r.getLeadershipTransferInProgress() {
    				r.logger.Debug(ErrLeadershipTransferInProgress.Error())
    				newLog.respond(ErrLeadershipTransferInProgress)
    				continue
    			}
    			// Group commit, gather all the ready commits.
                //多个提交一起，批处理。
    			ready := []*logFuture{newLog}
    		GROUP_COMMIT_LOOP:
    			for i := 0; i < r.conf.MaxAppendEntries; i++ {
    				select {
    				case newLog := <-r.applyCh:
    					ready = append(ready, newLog)
    				default:
    					break GROUP_COMMIT_LOOP
    				}
    			}
    
    			// Dispatch the logs
    			if stepDown {
    				// we're in the process of stepping down as leader, don't process anything new
    				for i := range ready {
    					ready[i].respond(ErrNotLeader)
    				}
    			} else {
    				r.dispatchLogs(ready)
    			}
    
    		//续约	
    		case <-lease:
    			// Check if we've exceeded the lease, potentially stepping down
    			maxDiff := r.checkLeaderLease()
    
    			// Next check interval should adjust for the last node we've
    			// contacted, without going negative
    			checkInterval := r.conf.LeaderLeaseTimeout - maxDiff
    			if checkInterval < minCheckInterval {
    				checkInterval = minCheckInterval
    			}
    
    			// Renew the lease timer
    			lease = time.After(checkInterval)
    
    		case <-r.shutdownCh:
    			return
    		}
    	}
    }

#### Leader 将日志同步给Follower过程，r.startStopReplication() ####

1.  创建协程，每个Follower一个
2.  replState 记录 replicator的状态

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MjEwNzY3_size_16_color_FFFFFF_t_70_pic_center]

func (r *Raft) startStopReplication() {
    	inConfig := make(map[ServerID]bool, len(r.configurations.latest.Servers))
    	lastIdx := r.getLastIndex()
    
    	// Start replication goroutines that need starting
    	for _, server := range r.configurations.latest.Servers {
    		if server.ID == r.localID {
    			continue
    		}
    
    		inConfig[server.ID] = true
    
    		s, ok := r.leaderState.replState[server.ID]
    		if !ok {
    			r.logger.Info("added peer, starting replication", "peer", server.ID)
    			s = &followerReplication{
    				peer:                server,
    				commitment:          r.leaderState.commitment,
    				stopCh:              make(chan uint64, 1),
    				triggerCh:           make(chan struct{}, 1),
    				triggerDeferErrorCh: make(chan *deferError, 1),
    				currentTerm:         r.getCurrentTerm(),
    				nextIndex:           lastIdx + 1,
    				lastContact:         time.Now(),
    				notify:              make(map[*verifyFuture]struct{}),
    				notifyCh:            make(chan struct{}, 1),
    				stepDown:            r.leaderState.stepDown,
    			}
    
    			r.leaderState.replState[server.ID] = s
    			r.goFunc(func() { r.replicate(s) })
    			asyncNotifyCh(s.triggerCh)
    			r.observe(PeerObservation{Peer: server, Removed: false})
    		} else if ok && s.peer.Address != server.Address {
    			r.logger.Info("updating peer", "peer", server.ID)
    			s.peer = server
    		}
    	}
        // ... 省略代码
    }

r.replicate方法。启动协程，触发心跳

func (r *Raft) heartbeat(s *followerReplication, stopCh chan struct{}) {
    	var failures uint64
    	req := AppendEntriesRequest{
    		RPCHeader: r.getRPCHeader(),
    		Term:      s.currentTerm,
    		Leader:    r.trans.EncodePeer(r.localID, r.localAddr),
    	}
    	var resp AppendEntriesResponse
    	for {
    		// Wait for the next heartbeat interval or forced notify
    		select {
    		case <-s.notifyCh:
    		case <-randomTimeout(r.conf.HeartbeatTimeout / 10):
    		case <-stopCh:
    			return
    		}
    
    		start := time.Now()
    		if err := r.trans.AppendEntries(s.peer.ID, s.peer.Address, &req, &resp); err != nil {
    			r.logger.Error("failed to heartbeat to", "peer", s.peer.Address, "error", err)
    			failures++
    			select {
    			case <-time.After(backoff(failureWait, failures, maxFailureScale)):
    			case <-stopCh:
    			}
    		} else {
    			s.setLastContact()
    			failures = 0
    			labels := []metrics.Label{
       {Name: "peer_id", Value: string(s.peer.ID)}}
    			metrics.MeasureSinceWithLabels([]string{"raft", "replication", "heartbeat"}, start, labels)
    			// Duplicated information. Kept for backward compatibility.
    			metrics.MeasureSince([]string{"raft", "replication", "heartbeat", string(s.peer.ID)}, start)
    			s.notifyAll(resp.Success)
    		}
    	}
    }

// replicate is a long running routine that replicates log entries to a single
    // follower.
    func (r *Raft) replicate(s *followerReplication) {
    	// Start an async heartbeating routing
    	stopHeartbeat := make(chan struct{})
    	defer close(stopHeartbeat)
    	r.goFunc(func() { r.heartbeat(s, stopHeartbeat) })
    
    RPC:
    	shouldStop := false
    	for !shouldStop {
    		select {
    		case maxIndex := <-s.stopCh:
    			// Make a best effort to replicate up to this index
    			if maxIndex > 0 {
    				r.replicateTo(s, maxIndex)
    			}
    			return
           ... //省略代码
    			//有新的Log。会触发triggerCh
    		case <-s.triggerCh:
    			lastLogIdx, _ := r.getLastLog()
    			shouldStop = r.replicateTo(s, lastLogIdx)
    		// This is _not_ our heartbeat mechanism but is to ensure
    		// followers quickly learn the leader's commit index when
    		// raft commits stop flowing naturally. The actual heartbeats
    		// can't do this to keep them unblocked by disk IO on the
    		// follower. See https://github.com/hashicorp/raft/issues/282.
    		case <-randomTimeout(r.conf.CommitTimeout):
    			lastLogIdx, _ := r.getLastLog()
    			shouldStop = r.replicateTo(s, lastLogIdx)
    		}
           ... //省略代码
    	return
    }

### 引用 ###

[Raft协议论文中文(翻译版)][Raft 1]  
[Raft协议在线动画][Raft 2]  
[Raft协议动画][Raft 3]  
[知乎上写的博客][Link 1]

[Raft]: https://github.com/hashicorp/raft.git
[20201018164542348.png_pic_center]: /images/20221123/c1bf24f36c1e4ccfba505b3ff6669832.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MjEwNzY3_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/c0af4e39c25b41a080810dea3cff9636.png
[Raft 1]: https://docs.qq.com/doc/DY0VxSkVGWHFYSlZJ
[Raft 2]: http://thesecretlivesofdata.com/raft/
[Raft 3]: https://raft.github.io/
[Link 1]: https://zhuanlan.zhihu.com/p/27207160