func (r *Raft) leaderLoop() {
	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:
			committed := r.leaderState.inflight.Committed()
			for e := committed.Front(); e != nil; e = e.Next() {
				commitLog := e.Value.(*logFuture)
				metrics.MeasureSince([]string{"raft", "commitTime"}, commitLog.dispatch)
				idx := commitLog.log.Index
				r.setCommitIndex(idx)
				r.processLogs(idx, commitLog)
			}
		case v := <-r.verifyCh:
			if v.quorumSize == 0 {
				r.verifyLeader(v)
			} else if v.votes < v.quorumSize {
				r.logger.Printf("[WARN] raft: New leader elected, stepping down")
				r.setState(Follower)
				delete(r.leaderState.notify, v)
				v.respond(ErrNotLeader)
			} else {
				delete(r.leaderState.notify, v)
				v.respond(nil)
			}
		case p := <-r.peerCh:
			p.respond(ErrLeader)
		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
				}
			}
			n := len(ready)
			for i := 0; i < n; i++ {
				log := ready[i]
				if log.log.Type != LogAddPeer && log.log.Type != LogRemovePeer {
					continue
				}
				if !r.preparePeerChange(log) {
					ready[i], ready[n-1] = ready[n-1], nil
					n--
					i--
					continue
				}
				r.processLog(&log.log, nil, true)
			}
			if n == 0 {
				continue
			}
			ready = ready[:n]
			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
		}
	}
}
leaderLoop is the hot loop for a leader, it is invoked
after all the various leader setup is done
package raft
import (
	"bytes"
	"errors"
	"fmt"
	"io"
	"log"
	"net"
	"os"
	"strconv"
	"sync"
	"time"
	"github.com/socketplane/socketplane/Godeps/_workspace/src/github.com/armon/go-metrics"
)
const (
	minCheckInterval = 10 * time.Millisecond
)
var (
	keyCurrentTerm  = []byte("CurrentTerm")
	keyLastVoteTerm = []byte("LastVoteTerm")
	keyLastVoteCand = []byte("LastVoteCand")
	ErrLeader = errors.New("node is the leader")
	ErrNotLeader = errors.New("node is not the leader")
	ErrLeadershipLost = errors.New("leadership lost while committing log")
	ErrRaftShutdown = errors.New("raft is already shutdown")
	ErrEnqueueTimeout = errors.New("timed out enqueuing operation")
	ErrKnownPeer = errors.New("peer already known")
	ErrUnknownPeer = errors.New("peer is unknown")
)
type commitTuple struct {
	log    *Log
	future *logFuture
}
type leaderState struct {
	commitCh  chan struct{}
	inflight  *inflight
	replState map[string]*followerReplication
	notify    map[*verifyFuture]struct{}
	stepDown  chan struct{}
}
type Raft struct {
	raftState
	applyCh chan *logFuture
	conf *Config
	fsm FSM
	fsmCommitCh chan commitTuple
	fsmRestoreCh chan *restoreFuture
	fsmSnapshotCh chan *reqSnapshotFuture
	lastContact     time.Time
	lastContactLock sync.RWMutex
	leader     net.Addr
	leaderLock sync.RWMutex
	leaderCh chan bool
	leaderState leaderState
	localAddr net.Addr
	logger *log.Logger
	logs LogStore
	peerCh    chan *peerFuture
	peers     []net.Addr
	peerStore PeerStore
	rpcCh <-chan RPC
	shutdown     bool
	shutdownCh   chan struct{}
	shutdownLock sync.Mutex
	snapshots SnapshotStore
	snapshotCh chan *snapshotFuture
	stable StableStore
	trans Transport
	verifyCh chan *verifyFuture
}
func NewRaft(conf *Config, fsm FSM, logs LogStore, stable StableStore, snaps SnapshotStore,
	peerStore PeerStore, trans Transport) (*Raft, error) {
	if err := ValidateConfig(conf); err != nil {
		return nil, err
	}
	if conf.LogOutput == nil {
		conf.LogOutput = os.Stderr
	}
	currentTerm, err := stable.GetUint64(keyCurrentTerm)
	if err != nil && err.Error() != "not found" {
		return nil, fmt.Errorf("failed to load current term: %v", err)
	}
	lastIdx, err := logs.LastIndex()
	if err != nil {
		return nil, fmt.Errorf("failed to find last log: %v", err)
	}
	var lastLog Log
	if lastIdx > 0 {
		if err := logs.GetLog(lastIdx, &lastLog); err != nil {
			return nil, fmt.Errorf("failed to get last log: %v", err)
		}
	}
	localAddr := trans.LocalAddr()
	peers, err := peerStore.Peers()
	if err != nil {
		return nil, fmt.Errorf("failed to get list of peers: %v", err)
	}
	peers = ExcludePeer(peers, localAddr)
	r := &Raft{
		applyCh:       make(chan *logFuture),
		conf:          conf,
		fsm:           fsm,
		fsmCommitCh:   make(chan commitTuple, 128),
		fsmRestoreCh:  make(chan *restoreFuture),
		fsmSnapshotCh: make(chan *reqSnapshotFuture),
		leaderCh:      make(chan bool),
		localAddr:     localAddr,
		logger:        log.New(conf.LogOutput, "", log.LstdFlags),
		logs:          logs,
		peerCh:        make(chan *peerFuture),
		peers:         peers,
		peerStore:     peerStore,
		rpcCh:         trans.Consumer(),
		snapshots:     snaps,
		snapshotCh:    make(chan *snapshotFuture),
		shutdownCh:    make(chan struct{}),
		stable:        stable,
		trans:         trans,
		verifyCh:      make(chan *verifyFuture, 64),
	}
	r.setState(Follower)
	r.setCurrentTerm(currentTerm)
	r.setLastLogIndex(lastLog.Index)
	r.setLastLogTerm(lastLog.Term)
	if err := r.restoreSnapshot(); err != nil {
		return nil, err
	}
	trans.SetHeartbeatHandler(r.processHeartbeat)
	r.goFunc(r.run)
	r.goFunc(r.runFSM)
	r.goFunc(r.runSnapshots)
	return r, nil
}
func (r *Raft) Leader() net.Addr {
	r.leaderLock.RLock()
	leader := r.leader
	r.leaderLock.RUnlock()
	return leader
}
func (r *Raft) setLeader(leader net.Addr) {
	r.leaderLock.Lock()
	r.leader = leader
	r.leaderLock.Unlock()
}
func (r *Raft) Apply(cmd []byte, timeout time.Duration) ApplyFuture {
	metrics.IncrCounter([]string{"raft", "apply"}, 1)
	var timer <-chan time.Time
	if timeout > 0 {
		timer = time.After(timeout)
	}
	logFuture := &logFuture{
		log: Log{
			Type: LogCommand,
			Data: cmd,
		},
	}
	logFuture.init()
	select {
	case <-timer:
		return errorFuture{ErrEnqueueTimeout}
	case <-r.shutdownCh:
		return errorFuture{ErrRaftShutdown}
	case r.applyCh <- logFuture:
		return logFuture
	}
}
func (r *Raft) Barrier(timeout time.Duration) Future {
	metrics.IncrCounter([]string{"raft", "barrier"}, 1)
	var timer <-chan time.Time
	if timeout > 0 {
		timer = time.After(timeout)
	}
	logFuture := &logFuture{
		log: Log{
			Type: LogBarrier,
		},
	}
	logFuture.init()
	select {
	case <-timer:
		return errorFuture{ErrEnqueueTimeout}
	case <-r.shutdownCh:
		return errorFuture{ErrRaftShutdown}
	case r.applyCh <- logFuture:
		return logFuture
	}
}
func (r *Raft) VerifyLeader() Future {
	metrics.IncrCounter([]string{"raft", "verify_leader"}, 1)
	verifyFuture := &verifyFuture{}
	verifyFuture.init()
	select {
	case <-r.shutdownCh:
		return errorFuture{ErrRaftShutdown}
	case r.verifyCh <- verifyFuture:
		return verifyFuture
	}
}
func (r *Raft) AddPeer(peer net.Addr) Future {
	logFuture := &logFuture{
		log: Log{
			Type: LogAddPeer,
			peer: peer,
		},
	}
	logFuture.init()
	select {
	case r.applyCh <- logFuture:
		return logFuture
	case <-r.shutdownCh:
		return errorFuture{ErrRaftShutdown}
	}
}
func (r *Raft) RemovePeer(peer net.Addr) Future {
	logFuture := &logFuture{
		log: Log{
			Type: LogRemovePeer,
			peer: peer,
		},
	}
	logFuture.init()
	select {
	case r.applyCh <- logFuture:
		return logFuture
	case <-r.shutdownCh:
		return errorFuture{ErrRaftShutdown}
	}
}
func (r *Raft) SetPeers(p []net.Addr) Future {
	peerFuture := &peerFuture{
		peers: p,
	}
	peerFuture.init()
	select {
	case r.peerCh <- peerFuture:
		return peerFuture
	case <-r.shutdownCh:
		return errorFuture{ErrRaftShutdown}
	}
}
func (r *Raft) Shutdown() Future {
	r.shutdownLock.Lock()
	defer r.shutdownLock.Unlock()
	if !r.shutdown {
		close(r.shutdownCh)
		r.shutdown = true
		r.setState(Shutdown)
	}
	return &shutdownFuture{r}
}
func (r *Raft) Snapshot() Future {
	snapFuture := &snapshotFuture{}
	snapFuture.init()
	select {
	case r.snapshotCh <- snapFuture:
		return snapFuture
	case <-r.shutdownCh:
		return errorFuture{ErrRaftShutdown}
	}
}
func (r *Raft) State() RaftState {
	return r.getState()
}
func (r *Raft) LeaderCh() <-chan bool {
	return r.leaderCh
}
func (r *Raft) String() string {
	return fmt.Sprintf("Node at %s [%v]", r.localAddr.String(), r.getState())
}
func (r *Raft) LastContact() time.Time {
	r.lastContactLock.RLock()
	last := r.lastContact
	r.lastContactLock.RUnlock()
	return last
}
func (r *Raft) Stats() map[string]string {
	toString := func(v uint64) string {
		return strconv.FormatUint(v, 10)
	}
	s := map[string]string{
		"state":               r.getState().String(),
		"term":                toString(r.getCurrentTerm()),
		"last_log_index":      toString(r.getLastLogIndex()),
		"last_log_term":       toString(r.getLastLogTerm()),
		"commit_index":        toString(r.getCommitIndex()),
		"applied_index":       toString(r.getLastApplied()),
		"fsm_pending":         toString(uint64(len(r.fsmCommitCh))),
		"last_snapshot_index": toString(r.getLastSnapshotIndex()),
		"last_snapshot_term":  toString(r.getLastSnapshotTerm()),
		"num_peers":           toString(uint64(len(r.peers))),
	}
	last := r.LastContact()
	if last.IsZero() {
		s["last_contact"] = "never"
	} else if r.getState() == Leader {
		s["last_contact"] = "0"
	} else {
		s["last_contact"] = fmt.Sprintf("%v", time.Now().Sub(last))
	}
	return s
}
func (r *Raft) LastIndex() uint64 {
	return r.getLastIndex()
}
func (r *Raft) runFSM() {
	var lastIndex, lastTerm uint64
	for {
		select {
		case req := <-r.fsmRestoreCh:
			meta, source, err := r.snapshots.Open(req.ID)
			if err != nil {
				req.respond(fmt.Errorf("failed to open snapshot %v: %v", req.ID, err))
				continue
			}
			start := time.Now()
			if err := r.fsm.Restore(source); err != nil {
				req.respond(fmt.Errorf("failed to restore snapshot %v: %v", req.ID, err))
				source.Close()
				continue
			}
			source.Close()
			metrics.MeasureSince([]string{"raft", "fsm", "restore"}, start)
			lastIndex = meta.Index
			lastTerm = meta.Term
			req.respond(nil)
		case req := <-r.fsmSnapshotCh:
			peers, err := r.peerStore.Peers()
			if err != nil {
				req.respond(err)
			}
			start := time.Now()
			snap, err := r.fsm.Snapshot()
			metrics.MeasureSince([]string{"raft", "fsm", "snapshot"}, start)
			req.index = lastIndex
			req.term = lastTerm
			req.peers = peers
			req.snapshot = snap
			req.respond(err)
		case commitTuple := <-r.fsmCommitCh:
			var resp interface{}
			if commitTuple.log.Type == LogCommand {
				start := time.Now()
				resp = r.fsm.Apply(commitTuple.log)
				metrics.MeasureSince([]string{"raft", "fsm", "apply"}, start)
			}
			lastIndex = commitTuple.log.Index
			lastTerm = commitTuple.log.Term
			if commitTuple.future != nil {
				commitTuple.future.response = resp
				commitTuple.future.respond(nil)
			}
		case <-r.shutdownCh:
			return
		}
	}
}
func (r *Raft) run() {
	for {
		select {
		case <-r.shutdownCh:
			r.setLeader(nil)
			return
		default:
		}
		switch r.getState() {
		case Follower:
			r.runFollower()
		case Candidate:
			r.runCandidate()
		case Leader:
			r.runLeader()
		}
	}
}
func (r *Raft) runFollower() {
	didWarn := false
	r.logger.Printf("[INFO] raft: %v entering Follower state", r)
	heartbeatTimer := randomTimeout(r.conf.HeartbeatTimeout)
	for {
		select {
		case rpc := <-r.rpcCh:
			r.processRPC(rpc)
		case a := <-r.applyCh:
			a.respond(ErrNotLeader)
		case v := <-r.verifyCh:
			v.respond(ErrNotLeader)
		case p := <-r.peerCh:
			r.peers = ExcludePeer(p.peers, r.localAddr)
			p.respond(r.peerStore.SetPeers(p.peers))
		case <-heartbeatTimer:
			heartbeatTimer = randomTimeout(r.conf.HeartbeatTimeout)
			lastContact := r.LastContact()
			if time.Now().Sub(lastContact) < r.conf.HeartbeatTimeout {
				continue
			}
			r.setLeader(nil)
			if len(r.peers) == 0 && !r.conf.EnableSingleNode {
				if !didWarn {
					r.logger.Printf("[WARN] raft: EnableSingleNode disabled, and no known peers. Aborting election.")
					didWarn = true
				}
			} else {
				r.logger.Printf("[WARN] raft: Heartbeat timeout reached, starting election")
				r.setState(Candidate)
				return
			}
		case <-r.shutdownCh:
			return
		}
	}
}
func (r *Raft) runCandidate() {
	r.logger.Printf("[INFO] raft: %v entering Candidate state", r)
	voteCh := r.electSelf()
	electionTimer := randomTimeout(r.conf.ElectionTimeout)
	grantedVotes := 0
	votesNeeded := r.quorumSize()
	r.logger.Printf("[DEBUG] raft: Votes needed: %d", votesNeeded)
	for r.getState() == Candidate {
		select {
		case rpc := <-r.rpcCh:
			r.processRPC(rpc)
		case vote := <-voteCh:
			if vote.Term > r.getCurrentTerm() {
				r.logger.Printf("[DEBUG] raft: Newer term discovered, fallback to follower")
				r.setState(Follower)
				r.setCurrentTerm(vote.Term)
				return
			}
			if vote.Granted {
				grantedVotes++
				r.logger.Printf("[DEBUG] raft: Vote granted. Tally: %d", grantedVotes)
			}
			if grantedVotes >= votesNeeded {
				r.logger.Printf("[INFO] raft: Election won. Tally: %d", grantedVotes)
				r.setLeader(r.localAddr)
				r.setState(Leader)
				return
			}
		case a := <-r.applyCh:
			a.respond(ErrNotLeader)
		case v := <-r.verifyCh:
			v.respond(ErrNotLeader)
		case p := <-r.peerCh:
			r.peers = ExcludePeer(p.peers, r.localAddr)
			p.respond(r.peerStore.SetPeers(p.peers))
			r.setState(Follower)
			return
		case <-electionTimer:
			r.logger.Printf("[WARN] raft: Election timeout reached, restarting election")
			return
		case <-r.shutdownCh:
			return
		}
	}
}
func (r *Raft) runLeader() {
	r.logger.Printf("[INFO] raft: %v entering Leader state", r)
	asyncNotifyBool(r.leaderCh, true)
	r.leaderState.commitCh = make(chan struct{}, 1)
	r.leaderState.inflight = newInflight(r.leaderState.commitCh)
	r.leaderState.replState = make(map[string]*followerReplication)
	r.leaderState.notify = make(map[*verifyFuture]struct{})
	r.leaderState.stepDown = make(chan struct{}, 1)
	defer func() {
		for _, p := range r.leaderState.replState {
			close(p.stopCh)
		}
		r.leaderState.inflight.Cancel(ErrLeadershipLost)
		for future := range r.leaderState.notify {
			future.respond(ErrLeadershipLost)
		}
		r.leaderState.commitCh = nil
		r.leaderState.inflight = nil
		r.leaderState.replState = nil
		r.leaderState.notify = nil
		r.leaderState.stepDown = nil
		r.leaderLock.Lock()
		if r.leader == r.localAddr {
			r.leader = nil
		}
		r.leaderLock.Unlock()
		asyncNotifyBool(r.leaderCh, false)
	}()
	for _, peer := range r.peers {
		r.startReplication(peer)
	}
	peerSet := append([]net.Addr{r.localAddr}, r.peers...)
	noop := &logFuture{
		log: Log{
			Type: LogAddPeer,
			Data: encodePeers(peerSet, r.trans),
		},
	}
	r.dispatchLogs([]*logFuture{noop})
	if r.conf.DisableBootstrapAfterElect && r.conf.EnableSingleNode {
		r.logger.Printf("[INFO] raft: Disabling EnableSingleNode (bootstrap)")
		r.conf.EnableSingleNode = false
	}
	r.leaderLoop()
}
func (r *Raft) startReplication(peer net.Addr) {
	lastIdx := r.getLastIndex()
	s := &followerReplication{
		peer:        peer,
		inflight:    r.leaderState.inflight,
		stopCh:      make(chan uint64, 1),
		triggerCh:   make(chan struct{}, 1),
		currentTerm: r.getCurrentTerm(),
		matchIndex:  0,
		nextIndex:   lastIdx + 1,
		lastContact: time.Now(),
		notifyCh:    make(chan struct{}, 1),
		stepDown:    r.leaderState.stepDown,
	}
	r.leaderState.replState[peer.String()] = s
	r.goFunc(func() { r.replicate(s) })
	asyncNotifyCh(s.triggerCh)
}