深入浅出etcd之raft实现

// node is the canonical implementation of the Node interface
type node struct {
	propc      chan msgWithResult
	recvc      chan pb.Message
	confc      chan pb.ConfChange
	confstatec chan pb.ConfState
	readyc     chan Ready
	advancec   chan struct{}
	tickc      chan struct{}
	done       chan struct{}
	stop       chan struct{}
	status     chan chan Status

	logger Logger
}

type Node interface {
	Tick() //时钟的实现，选举超时和心跳超时基于此实现
	Campaign(ctx context.Context) error //参与leader竞争
	Propose(ctx context.Context, data []byte) error //在日志中追加数据，需要实现方保证数据追加的成功
	ProposeConfChange(ctx context.Context, cc pb.ConfChange) error // 集群配置变更
	Step(ctx context.Context, msg pb.Message) error //根据消息变更状态机的状态
	//标志某一状态的完成，收到状态变化的节点必须提交变更
	Ready() <-chan Ready
	//进行状态的提交，收到完成标志后，必须提交过后节点才会实际进行状态机的更新。在包含快照的场景，为了避免快照落地带来的长时间阻塞，允许继续接受和提交其他状态，即使之前的快照状态变更并没有完成。
	Advance()
	//进行集群配置变更
	ApplyConfChange(cc pb.ConfChange) *pb.ConfState
	//变更leader
	TransferLeadership(ctx context.Context, lead, transferee uint64)
	//保证线性一致性读，
	ReadIndex(ctx context.Context, rctx []byte) error
	//状态机当前的配置
	Status() Status
	// ReportUnreachable reports the given node is not reachable for the last send.
	//上报节点的不可达
	ReportUnreachable(id uint64)
	//上报快照状态
	ReportSnapshot(id uint64, status SnapshotStatus)
	//停止节点
	Stop()
}

//启动
func StartNode(c *Config, peers []Peer) Node {
	r := newRaft(c) //初始化raft算法实例
	r.becomeFollower(1, None)
	//将配置中的节点加入集群
	for _, peer := range peers {
		...
	}
	//初始化logindex
	r.raftLog.committed = r.raftLog.lastIndex()
	for _, peer := range peers {
		//初始化节点状态机(progress)
		r.addNode(peer.ID)
	}
	n := newNode()
	n.logger = c.Logger
	go n.run(r)
	return &n
}

//运行
func (n *node) run(r *raft) {
	...

		select {
		//接收到写消息
		case pm := <-propc:
			...
		//接收到readindex 请求
		case m := <-n.recvc:
			...
		//配置变更
		case cc := <-n.confc:
			...
		//超时时间到，包括心跳超时和选举超时等
		case <-n.tickc:
			...
		//数据ready
		case readyc <- rd:
			...
		//可以进行状态变更和日志提交
		case <-advancec:
			...
		//节点状态信号
		case c := <-n.status:
			...
		//收到停止信号
		case <-n.stop:
			...
		}
	}
}

	r := newRaft(c)
	r.becomeFollower(1, None)

func (n *node) run(r *raft) {
    ...
    for {
        select {
            ...
            //触发heartbeat 超时
            case <-n.tickc:
			    r.tick()
            ...
        }
    }
}

//超时触发选举
func (r *raft) tickElection() {
	r.electionElapsed++

	if r.promotable() && r.pastElectionTimeout() {
		r.electionElapsed = 0
		r.Step(pb.Message{From: r.id, Type: pb.MsgHup})
	}
}

//随机超时时间
func (r *raft) pastElectionTimeout() bool {
	return r.electionElapsed >= r.randomizedElectionTimeout
}

func (r *raft) resetRandomizedElectionTimeout() {
	r.randomizedElectionTimeout = r.electionTimeout + globalRand.Intn(r.electionTimeout)
}


//参与选举
func (r *raft) campaign(t CampaignType) {
	var term uint64
	var voteMsg pb.MessageType
	//成为candicate,将任期id加1
	if t == campaignPreElection {
		r.becomePreCandidate()
		voteMsg = pb.MsgPreVote
		term = r.Term + 1
	} else {
		r.becomeCandidate()
		voteMsg = pb.MsgVote
		term = r.Term
	}
	//判断获取的票数是否超过半数，如果是当选为leader
	if r.quorum() == r.poll(r.id, voteRespMsgType(voteMsg), true) {
		if t == campaignPreElection {
			r.campaign(campaignElection)
		} else {
			r.becomeLeader()
		}
		return
	}
	//向其他节点发送竞选消息
	for id := range r.prs {
		if id == r.id {
			continue
		}
		var ctx []byte
		if t == campaignTransfer {
			ctx = []byte(t)
		}
		r.send(pb.Message{Term: term, To: id, Type: voteMsg, Index: r.raftLog.lastIndex(), LogTerm: r.raftLog.lastTerm(), Context: ctx})
	}
}

//节点投票过程
func (r *raft) Step(m pb.Message) error {
	...
	//比较任期id
	case m.Term > r.Term:
		if m.Type == pb.MsgVote || m.Type == pb.MsgPreVote {
			force := bytes.Equal(m.Context, []byte(campaignTransfer))
			inLease := r.checkQuorum && r.lead != None && r.electionElapsed < r.electionTimeout
			if !force && inLease {
				return nil
			}
		}
		
	switch m.Type {
	case pb.MsgVote, pb.MsgPreVote:
		...
		//与本地最新的持久化日志比较
		if canVote && r.raftLog.isUpToDate(m.Index, m.LogTerm) {
			//发送投票信息
			r.send(pb.Message{To: m.From, Term: m.Term, Type: voteRespMsgType(m.Type)})
			if m.Type == pb.MsgVote {
				// Only record real votes.
				r.electionElapsed = 0
				r.Vote = m.From
			}
		} 
		...
	return nil
}

func (l *raftLog) isUpToDate(lasti, term uint64) bool {
	return term > l.lastTerm() || (term == l.lastTerm() && lasti >= l.lastIndex())
}

//投票结果判断
	case myVoteRespType:
		gr := r.poll(m.From, m.Type, !m.Reject)
		//计算票数是否超过半数
		switch r.quorum() {
		case gr:
			if r.state == StatePreCandidate {
				r.campaign(campaignElection)
			} else {
				r.becomeLeader()
				r.bcastAppend()
			}
		case len(r.votes) - gr:
			r.becomeFollower(r.Term, None)
		}

func (n *node) Propose(ctx context.Context, data []byte) error {
	return n.stepWait(ctx, pb.Message{Type: pb.MsgProp, Entries: []pb.Entry{{Data: data}}})
}

func (n *node) stepWithWaitOption(ctx context.Context, m pb.Message, wait bool) error {
    ...
	//提交日志数据至 node的 propc channel 队列
	select {
	case ch <- pm:
		if !wait {
		    //非阻塞直接返回
			return nil
		}
	case <-ctx.Done():
		return ctx.Err()
	case <-n.done:
		return ErrStopped
	}
	select {
	//等待结果的返回
	case rsp := <-pm.result:
		if rsp != nil {
			return rsp
		}
	case <-ctx.Done():
		return ctx.Err()
	case <-n.done:
		return ErrStopped
	}
	return nil
}

func stepLeader(r *raft, m pb.Message) error {
	case pb.MsgProp:
		...
		if !r.appendEntry(m.Entries...) {
			return ErrProposalDropped
		}
		r.bcastAppend()
		...
}

func (r *raft) handleAppendEntries(m pb.Message) {
    //leader提交的logindex小于本地已经提交的logindex
	if m.Index < r.raftLog.committed {
		r.send(pb.Message{To: m.From, Type: pb.MsgAppResp, Index: r.raftLog.committed})
		return
	}
	//追加日志，可能存在冲突的情况，需要找到冲突的位置用leader的日志进行覆盖
	if mlastIndex, ok := r.raftLog.maybeAppend(m.Index, m.LogTerm, m.Commit, m.Entries...); ok {
	    //mlastIndex表示最佳成功的最新位置
		r.send(pb.Message{To: m.From, Type: pb.MsgAppResp, Index: mlastIndex})
	} else {
	    //任期信息不一致，拒绝此次追加请求，并把最新的logindex回复给leader,便于进行追加
		r.send(pb.Message{To: m.From, Type: pb.MsgAppResp, Index: m.Index, Reject: true, RejectHint: r.raftLog.lastIndex()})
	}
}

   func stepLeader(r *raft, m pb.Message) error {
	    case pb.MsgAppResp:
		    pr.RecentActive = true

		    if m.Reject {
			...
		} else {
			oldPaused := pr.IsPaused()
			//更新索引信息，更新该follower的match index 和next index.
			if pr.maybeUpdate(m.Index) {
				switch {
				//日志追加成功，状态由复制探测状态变成复制状态，加快日志的追加
				case pr.State == ProgressStateProbe:
					pr.becomeReplicate()
				case pr.State == ProgressStateSnapshot && pr.needSnapshotAbort():
					r.logger.Debugf("%x snapshot aborted, resumed sending replication messages to %x [%s]", r.id, m.From, pr)
					pr.becomeProbe()
				//pr.ins用于限制消息发送的速率，用于统计当前处于发送状态的日志数量
				case pr.State == ProgressStateReplicate:
					pr.ins.freeTo(m.Index)
				}
				//leader进行本地的提交
				if r.maybeCommit() {
					//广播至所有follower 通知进行log的提交
					r.bcastAppend()
				} else if oldPaused {
					//append请求被中止，则重新发送最新的请求
					r.sendAppend(m.From)
				}
				}
			}
		}
    }
    
    func (r *raft) maybeCommit() bool {
		if cap(r.matchBuf) < len(r.prs) {
			r.matchBuf = make(uint64Slice, len(r.prs))
		}
		mis := r.matchBuf[:len(r.prs)]
		idx := 0
		for _, p := range r.prs {
			mis[idx] = p.Match
			idx++
		}
		//排序取取中间值
		sort.Sort(mis)
		mci := mis[len(mis)-r.quorum()]
		return r.raftLog.maybeCommit(mci, r.Term)
	}
	
	func (l *raftLog) maybeCommit(maxIndex, term uint64) bool {
		//match的中间值是否已经大于本地已经commit的matchindex
		if maxIndex > l.committed && l.zeroTermOnErrCompacted(l.term(maxIndex)) == term {
			l.commitTo(maxIndex)
			return true
		}
		return false
	}

   //follower接收leader的请求
   func (l *raftLog) maybeAppend(index, logTerm, committed uint64, ents ...pb.Entry) (lastnewi uint64, ok bool) {
       if l.matchTerm(index, logTerm) {
   	    ...
       }
       //拒绝leader当前的append请求
       return 0, false
    }
   //对leader提交append请求中的logindex和termid进行判断
   func (l *raftLog) matchTerm(i, term uint64) bool {
       t, err := l.term(i)
       if err != nil {
   	    return false
       }
       return t == term
    }
    
    func stepLeader(r *raft, m pb.Message) error {
       case pb.MsgAppResp:
   	    pr.RecentActive = true
   
   	    if m.Reject {
   		    if pr.maybeDecrTo(m.Index, m.RejectHint) {
   			    //由复制状态进入探测状态，探测follower最新的匹配位置
   			    if pr.State == ProgressStateReplicate {
   				    pr.becomeProbe()
   			    }
   			    r.sendAppend(m.From)
   		    }
    }

func (pr *Progress) maybeDecrTo(rejected, last uint64) bool {
       if pr.State == ProgressStateReplicate {
   	    if rejected <= pr.Match {
   		    return false
   	    }
   	    // directly decrease next to match + 1
   	    //复制状态将pr的next置为当前匹配位置+1
   	    pr.Next = pr.Match + 1
   	    return true
       }
   
       if pr.Next-1 != rejected {
   	    return false
       }
   
    	//如果是探测状态，则将next置为follower中最新的logindex  和当前index中的最小值。
       if pr.Next = min(rejected, last+1); pr.Next < 1 {
   	    pr.Next = 1
       }
       pr.resume()
       return true
    }
      日志推送的具体实现位于maybeSendAppend.func (r *raft) maybeSendAppend(to uint64, sendIfEmpty bool) bool {
	pr := r.getProgress(to)
	if pr.IsPaused() {
		return false
	}
	m := pb.Message{}
	m.To = to

	//发送给follower的最后一条日志对应的任期
	term, errt := r.raftLog.term(pr.Next - 1)
	//需要发送给follower的日志条数
	ents, erre := r.raftLog.entries(pr.Next, r.maxMsgSize)
	if len(ents) == 0 && !sendIfEmpty {
		return false
	}

	if errt != nil || erre != nil { // send snapshot if we failed to get term or entries
			...
	} else {
		m.Type = pb.MsgApp
		m.Index = pr.Next - 1
		m.LogTerm = term
		m.Entries = ents
		//leader 已经提交的最新index
		m.Commit = r.raftLog.committed
		if n := len(m.Entries); n != 0 {
			switch pr.State {
			//在日志复制状态，乐观地增加next, 加快日志的推送速度
			case ProgressStateReplicate:
				last := m.Entries[n-1].Index
				pr.optimisticUpdate(last)
				pr.ins.add(last)
			case ProgressStateProbe:
				pr.pause()
			default:
				r.logger.Panicf("%x is sending append in unhandled state %s", r.id, pr.State)
			}
		}
	}
	r.send(m)
	return true
}

func newReady(r *raft, prevSoftSt *SoftState, prevHardSt pb.HardState) Ready {
		...
		//日志数据
		rd := Ready{
			Entries:          r.raftLog.unstableEntries(),
			CommittedEntries: r.raftLog.nextEnts(),
			Messages:         r.msgs,
		}
		...
	}

 //etcd/raft/raft.go
 func (n *node) TransferLeadership(ctx context.Context, lead, transferee uint64) {
	    select {
    	//通过recvc发送MsgTransferLeader消息至集群中节点
	    case n.recvc <- pb.Message{Type: pb.MsgTransferLeader, From: transferee, To: lead}:
	    case <-n.done:
	    case <-ctx.Done():
	    }
    }

 //etcd/raft/raft.go
 func stepLeader(r *raft, m pb.Message) error {	
		switch m.Type {
		...
		case pb.MsgTransferLeader:
			if pr.IsLearner {
				r.logger.Debugf("%x is learner. Ignored transferring leadership", r.id)
				return nil
			}
			leadTransferee := m.From
			lastLeadTransferee := r.leadTransferee
			//上一次transfer正在进行
			if lastLeadTransferee != None {
				if lastLeadTransferee == leadTransferee {
					r.logger.Infof("%x [term %d] transfer leadership to %x is in progress, ignores request to same node %x",
						r.id, r.Term, leadTransferee, leadTransferee)
					return nil
				}
				r.abortLeaderTransfer()
				r.logger.Infof("%x [term %d] abort previous transferring leadership to %x", r.id, r.Term, lastLeadTransferee)
			}
			//transfee和当前leader相同
			if leadTransferee == r.id {
				r.logger.Debugf("%x is already leader. Ignored transferring leadership to self", r.id)
				return nil
			}
			// Transfer leadership to third party.
			// Transfer leadership should be finished in one electionTimeout, so reset r.electionElapsed.
			r.electionElapsed = 0
			r.leadTransferee = leadTransferee
			if pr.Match == r.raftLog.lastIndex() {
				//transfee的日志已经是最新和leader保持一致了，则立刻发送MsgTimeoutNow，触发选举
				r.sendTimeoutNow(leadTransferee)
				r.logger.Infof("%x sends MsgTimeoutNow to %x immediately as %x already has up-to-date log", r.id, leadTransferee, leadTransferee)
			} else {
				//日志非最新进行日志的同步
				r.sendAppend(leadTransferee)
			}
		}
		return nil
	}

   func newRaft(c *Config) *raft {
   	...
   	}
   	r := &raft{
   		id:                        c.ID,
   		...
   		//初始化readOnly配置
   		readOnly:                  newReadOnly(c.ReadOnlyOption),
   		disableProposalForwarding: c.DisableProposalForwarding,
   	}
   }
   
   const (
   	//ReadIndex方案
   	ReadOnlySafe ReadOnlyOption = iota
   	//leaseRead方案
   	ReadOnlyLeaseBased
   )

   	func stepLeader(r *raft, m pb.Message) error {
   	switch m.Type {
   	...
   	case pb.MsgReadIndex:
   			switch r.readOnly.option {
   			case ReadOnlySafe:
   				//加入请求队列
   				r.readOnly.addRequest(r.raftLog.committed, m)
   				//广播心跳消息
   				r.bcastHeartbeatWithCtx(m.Entries[0].Data)
   			}
   		} else {
   			r.readStates = append(r.readStates, ReadState{Index: r.raftLog.committed, RequestCtx: m.Entries[0].Data})
   		}
   	}
   }
   
   func (ro *readOnly) addRequest(index uint64, m pb.Message) {
   	ctx := string(m.Entries[0].Data)
   	if _, ok := ro.pendingReadIndex[ctx]; ok {
   		return
   	}
   	//index是当前集群的committedIndex,acks 用来收集节点心跳回复包
   	ro.pendingReadIndex[ctx] = &readIndexStatus{index: index, req: m, acks: make(map[uint64]struct{})}
   	ro.readIndexQueue = append(ro.readIndexQueue, ctx)
   }

   	func stepLeader(r *raft, m pb.Message) error {
   	case pb.MsgHeartbeatResp:
   		...
   		}
   		//统计回复结果，如果未超过半数则直接返回
   		ackCount := r.readOnly.recvAck(m)
   		if ackCount < r.quorum() {
   			return nil
   		}
   
   		rss := r.readOnly.advance(m)
   		for _, rs := range rss {
   			req := rs.req
   			//如果是本地的请求
   			if req.From == None || req.From == r.id { // from local member
   				r.readStates = append(r.readStates, ReadState{Index: rs.index, RequestCtx: req.Entries[0].Data})
   			} else {
   				//如果是来自follower的请求，将结果返回给follower
   				r.send(pb.Message{To: req.From, Type: pb.MsgReadIndexResp, Index: rs.index, Entries: req.Entries})
   			}
   		}
   	}
   	
   	func newReady(r *raft, prevSoftSt *SoftState, prevHardSt pb.HardState) Ready {
   		rd := Ready{
   			Entries:          r.raftLog.unstableEntries(),
   			CommittedEntries: r.raftLog.nextEnts(),
   			Messages:         r.msgs,
   		}
   		...
   		//readIndex消息追加
   		if len(r.readStates) != 0 {
   			rd.ReadStates = r.readStates
   		}
   		rd.MustSync = MustSync(r.hardState(), prevHardSt, len(rd.Entries))
   		return rd
   	}
   	
   	func (n *node) run(r *raft) {
   		....
   		for {
   			if advancec != nil {
   				readyc = nil
   			} else {
   				//消息加入readyc队列
   				rd = newReady(r, prevSoftSt, prevHardSt)
   				if rd.containsUpdates() {
   					readyc = n.readyc
   				} else {
   					readyc = nil
   				}
   			}
   			....
   	}	

   func stepFollower(r *raft, m pb.Message) error {	
   	...
   	case pb.MsgReadIndex:
   		if r.lead == None {
   			r.logger.Infof("%x no leader at term %d; dropping index reading msg", r.id, r.Term)
   			return nil
   		}
   		//将ReadIndex请求转发给leader
   		m.To = r.lead
   		r.send(m)
   	case pb.MsgReadIndexResp:
   		if len(m.Entries) != 1 {
   			r.logger.Errorf("%x invalid format of MsgReadIndexResp from %x, entries count: %d", r.id, m.From, len(m.Entries))
   			return nil
   		}
   		//收到leader回复后将消息加入readStatus
   		r.readStates = append(r.readStates, ReadState{Index: m.Index, RequestCtx: m.Entries[0].Data})
   		...
   }