EtcdRaft源码分析(心跳)

2019-03-15  本文已影响0人  Pillar_Zhong

这篇我们看下心跳在整个Raft里面是怎么牵一发而动全身,真可谓是Raft的发条。

Client

type Node interface {
   // Tick increments the internal logical clock for the Node by a single tick. Election
   // timeouts and heartbeat timeouts are in units of ticks.
   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})
    }
}

心跳会触发选举超时啦,具体流程可查看EtcdRaft源码分析(选举超时)部分。这里不再详述。

心跳

func (r *raft) tickHeartbeat() {
   r.heartbeatElapsed++
   r.electionElapsed++

   if r.electionElapsed >= r.electionTimeout {
      r.electionElapsed = 0
      if r.checkQuorum {
         r.Step(pb.Message{From: r.id, Type: pb.MsgCheckQuorum})
      }
      // If current leader cannot transfer leadership in electionTimeout, it becomes leader again.
      if r.state == StateLeader && r.leadTransferee != None {
         r.abortLeaderTransfer()
      }
   }

   if r.state != StateLeader {
      return
   }

   if r.heartbeatElapsed >= r.heartbeatTimeout {
      r.heartbeatElapsed = 0
      r.Step(pb.Message{From: r.id, Type: pb.MsgBeat})
   }
}
  • 首先心跳只有Leader才能发出,是Leader保证权力稳固的最重要的手段。
  • 当然,心跳的触发的同时要累加心跳超时和选举超时
  • 一般来说一直发心跳的话,不会可能选举超时,因为选举超时会差心跳一个数量级。不管如何,有可能是网络的原因,导致心跳没有发出。
    • 假如选举超时,那么首先要重置超时时间。
    • 如果设置checkQuorum,那么去调查下当前成员的活跃度,如果活跃度不够,直接转没有leader的follower
    • 如果当前正在进行权力转移,但超过了一个选举周期,那么这次转移认为失败,取消。想象一下,一个成功的权力交接在选举周期内才可能成功,越早越好不是么?假如超过,那么你当选的几率不会比别人高。
  • 不管怎样,有可能健康度检查失败,总之现在不是leader了,那当然就不能继续发心跳了。返回。
  • 最重要的是,心跳计时器到了,现在要立即给成员发心跳MsgBeat了。让人民知道我的存在。

Leader

case pb.MsgBeat:
   r.bcastHeartbeat()
   return nil

前面说了只有Leader才能发心跳,收到MsgBeat后,会去扩散给成员心跳。

bcastHeartbeat

func (r *raft) sendHeartbeat(to uint64, ctx []byte) {
   commit := min(r.getProgress(to).Match, r.raftLog.committed)
   m := pb.Message{
      To:      to,
      Type:    pb.MsgHeartbeat,
      Commit:  commit,
      Context: ctx,
   }

   r.send(m)
}
  • 当然要算取一个commit,最不济是对方进度的Match位置。
  • 给对方发MsgHeartbeat

Candidate

case pb.MsgHeartbeat:
   r.becomeFollower(m.Term, m.From) // always m.Term == r.Term
   r.handleHeartbeat(m)
  • 首先Candidate收到心跳,说明已经选出了Leader,那么你候选人的身份就尴尬了。马上向对方称臣。然后开始处理心跳才是上策。

Follower

case pb.MsgHeartbeat:
   r.electionElapsed = 0
   r.lead = m.From
   r.handleHeartbeat(m)
  • 收到心跳,那么Follower的选举超时就要清零,因为Leader还健在,你需要表忠心,保证不会发起选举。

handleHeartbeat

func (r *raft) handleHeartbeat(m pb.Message) {
   r.raftLog.commitTo(m.Commit)
   r.send(pb.Message{To: m.From, Type: pb.MsgHeartbeatResp, Context: m.Context})
}
  • 心跳包中会包含Leader的committed的index,所以首先先更新自己的commit
  • 给Leader发回MsgHeartbeatResp

Leader

case pb.MsgHeartbeatResp:
   pr.RecentActive = true
   pr.resume()

   // free one slot for the full inflights window to allow progress.
   if pr.State == ProgressStateReplicate && pr.ins.full() {
      pr.ins.freeFirstOne()
   }
   if pr.Match < r.raftLog.lastIndex() {
      r.sendAppend(m.From)
   }

   if r.readOnly.option != ReadOnlySafe || len(m.Context) == 0 {
      return nil
   }

   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 {
         r.send(pb.Message{To: req.From, Type: pb.MsgReadIndexResp, Index: rs.index, Entries: req.Entries})
      }
   }
  • 首先,收到心跳反馈,就说明对方是活跃的,更新本地Progress的RecentActive和pause
  • 如果对方进度状态是ProgressStateReplicate,且ins是满的,释放第一个slot
    • 这里讲讲ins,也就是inflights
    • add的时候有限制,假如已经满了,不止会返回,而且会panic,同步将终止退出。
    • 首先在Leader发起同步日志的时候会往ins里面add最新的index。然后在收到对方响应的时候,会取ins里面移除<=发来的index的slot。一般正常说来,当我Leader往ins里面add的多少,最终接受的时候也是那么多。对方会向我看齐。
    • inflights的作用是干嘛?
      • 一个好处是一个心跳周期内,不至于发得很频繁,有一定的节制
      • 等待心跳的时候释放一个slot,让Leader继续发
  • 如果对方的Match位要小于自己最后一位,那么给对方发日志复制的消息
上一篇下一篇

猜你喜欢

热点阅读