replication.go
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(fmt.Sprintf("Failed to heartbeat to %v: %v", s.peer.Address, err))
			failures++
			select {
			case <-time.After(backoff(failureWait, failures, maxFailureScale)):
			case <-stopCh:
			}
		} else {
			s.setLastContact()
			failures = 0
			metrics.MeasureSince([]string{"raft", "replication", "heartbeat", string(s.peer.ID)}, start)
			s.notifyAll(resp.Success)
		}
	}
}
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(fmt.Sprintf("Failed to heartbeat to %v: %v", s.peer.Address, err))
			failures++
			select {
			case <-time.After(backoff(failureWait, failures, maxFailureScale)):
			case <-stopCh:
			}
		} else {
			s.setLastContact()
			failures = 0
			metrics.MeasureSince([]string{"raft", "replication", "heartbeat", string(s.peer.ID)}, start)
			s.notifyAll(resp.Success)
		}
	}
}
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 ( fmt . Sprintf ( " " , s . peer . Address , err ) ) failures ++ select { case <- time . After ( backoff ( failureWait , failures , maxFailureScale ) ) : case <- stopCh : } } else { s . setLastContact ( ) failures = 0 metrics . MeasureSince ( [ ] string { " " , " " , " " , string ( s . peer . ID ) } , start ) s . notifyAll ( resp . Success ) } } }
// heartbeat is used to periodically invoke AppendEntries on a peer
// to ensure they don't time out. This is done async of replicate(),
// since that routine could potentially be blocked on disk IO.
heartbeat is used to periodically invoke AppendEntries on a peer to ensure they don t time out . This is done async of replicate () since that routine could potentially be blocked on disk IO .