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)

		case <-r.leaderState.commitCh:
			// Process the newly committed entries
			oldCommitIndex := r.getCommitIndex()
			commitIndex := r.leaderState.commitment.getCommitIndex()
			r.setCommitIndex(commitIndex)

			if r.configurations.latestIndex > oldCommitIndex &&
				r.configurations.latestIndex <= commitIndex {
				r.configurations.committed = r.configurations.latest
				r.configurations.committedIndex = r.configurations.latestIndex
				if !hasVote(r.configurations.committed, r.localID) {
					stepDown = true
				}
			}

			var numProcessed int
			start := time.Now()

			for {
				e := r.leaderState.inflight.Front()
				if e == nil {
					break
				}
				commitLog := e.Value.(*logFuture)
				idx := commitLog.log.Index
				if idx > commitIndex {
					break
				}
				// Measure the commit time
				metrics.MeasureSince([]string{"raft", "commitTime"}, commitLog.dispatch)

				r.processLogs(idx, commitLog)

				r.leaderState.inflight.Remove(e)
				numProcessed++
			}

			// 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(numProcessed))

			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)
				}
			}

		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 future := <-r.userRestoreCh:
			err := r.restoreUserSnapshot(future.meta, future.reader)
			future.respond(err)

		case c := <-r.configurationsCh:
			c.configurations = r.configurations.Clone()
			c.respond(nil)

		case future := <-r.configurationChangeChIfStable():
			r.appendConfigurationEntry(future)

		case b := <-r.bootstrapCh:
			b.respond(ErrCantBootstrap)

		case newLog := <-r.applyCh:
			// Group commit, gather all the ready commits
			ready := []*logFuture{newLog}
			for i := 0; i < r.conf.MaxAppendEntries; i++ {
				select {
				case newLog := <-r.applyCh:
					ready = append(ready, newLog)
				default:
					break
				}
			}

			// 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
		}
	}
}
func (r *Raft) leaderLoop() {
	
	
	
	
	
	
	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)

		case <-r.leaderState.commitCh:
			
			oldCommitIndex := r.getCommitIndex()
			commitIndex := r.leaderState.commitment.getCommitIndex()
			r.setCommitIndex(commitIndex)

			if r.configurations.latestIndex > oldCommitIndex &&
				r.configurations.latestIndex <= commitIndex {
				r.configurations.committed = r.configurations.latest
				r.configurations.committedIndex = r.configurations.latestIndex
				if !hasVote(r.configurations.committed, r.localID) {
					stepDown = true
				}
			}

			var numProcessed int
			start := time.Now()

			for {
				e := r.leaderState.inflight.Front()
				if e == nil {
					break
				}
				commitLog := e.Value.(*logFuture)
				idx := commitLog.log.Index
				if idx > commitIndex {
					break
				}
				
				metrics.MeasureSince([]string{"raft", "commitTime"}, commitLog.dispatch)

				r.processLogs(idx, commitLog)

				r.leaderState.inflight.Remove(e)
				numProcessed++
			}

			
			metrics.MeasureSince([]string{"raft", "fsm", "enqueue"}, start)

			
			metrics.SetGauge([]string{"raft", "commitNumLogs"}, float32(numProcessed))

			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)
				}
			}

		case v := <-r.verifyCh:
			if v.quorumSize == 0 {
				
				r.verifyLeader(v)

			} else if v.votes < v.quorumSize {
				
				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 {
				
				delete(r.leaderState.notify, v)
				for _, repl := range r.leaderState.replState {
					repl.cleanNotify(v)
				}
				v.respond(nil)
			}

		case future := <-r.userRestoreCh:
			err := r.restoreUserSnapshot(future.meta, future.reader)
			future.respond(err)

		case c := <-r.configurationsCh:
			c.configurations = r.configurations.Clone()
			c.respond(nil)

		case future := <-r.configurationChangeChIfStable():
			r.appendConfigurationEntry(future)

		case b := <-r.bootstrapCh:
			b.respond(ErrCantBootstrap)

		case newLog := <-r.applyCh:
			
			ready := []*logFuture{newLog}
			for i := 0; i < r.conf.MaxAppendEntries; i++ {
				select {
				case newLog := <-r.applyCh:
					ready = append(ready, newLog)
				default:
					break
				}
			}

			
			if stepDown {
				
				for i := range ready {
					ready[i].respond(ErrNotLeader)
				}
			} else {
				r.dispatchLogs(ready)
			}

		case <-lease:
			
			maxDiff := r.checkLeaderLease()

			
			
			checkInterval := r.conf.LeaderLeaseTimeout - maxDiff
			if checkInterval < minCheckInterval {
				checkInterval = minCheckInterval
			}

			
			lease = time.After(checkInterval)

		case <-r.shutdownCh:
			return
		}
	}
}
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 ) case <- r . leaderState . commitCh : // Process the newly committed entries oldCommitIndex := r . getCommitIndex ( ) commitIndex := r . leaderState . commitment . getCommitIndex ( ) r . setCommitIndex ( commitIndex ) if r . configurations . latestIndex > oldCommitIndex && r . configurations . latestIndex <= commitIndex { r . configurations . committed = r . configurations . latest r . configurations . committedIndex = r . configurations . latestIndex if ! hasVote ( r . configurations . committed , r . localID ) { stepDown = true } } var numProcessed int start := time . Now ( ) for { e := r . leaderState . inflight . Front ( ) if e == nil { break } commitLog := e . Value . ( * logFuture ) idx := commitLog . log . Index if idx > commitIndex { break } // Measure the commit time metrics . MeasureSince ( [ ] string { " " , " " } , commitLog . dispatch ) r . processLogs ( idx , commitLog ) r . leaderState . inflight . Remove ( e ) numProcessed ++ } // Measure the time to enqueue batch of logs for FSM to apply metrics . MeasureSince ( [ ] string { " " , " " , " " } , start ) // Count the number of logs enqueued metrics . SetGauge ( [ ] string { " " , " " } , float32 ( numProcessed ) ) if stepDown { if r . conf . ShutdownOnRemove { r . logger . Info ( " " ) r . Shutdown ( ) } else { r . logger . Info ( " " ) r . setState ( Follower ) } } 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 ( " " ) 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 future := <- r . userRestoreCh : err := r . restoreUserSnapshot ( future . meta , future . reader ) future . respond ( err ) case c := <- r . configurationsCh : c . configurations = r . configurations . Clone ( ) c . respond ( nil ) case future := <- r . configurationChangeChIfStable ( ) : r . appendConfigurationEntry ( future ) case b := <- r . bootstrapCh : b . respond ( ErrCantBootstrap ) case newLog := <- r . applyCh : // Group commit, gather all the ready commits ready := [ ] * logFuture { newLog } for i := 0 ; i < r . conf . MaxAppendEntries ; i ++ { select { case newLog := <- r . applyCh : ready = append ( ready , newLog ) default : break } } // 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 } } }
// leaderLoop is the hot loop for a leader. It is invoked
// after all the various leader setup is done.
leaderLoop is the hot loop for a leader . It is invoked after all the various leader setup is done .