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
		}
	}
}
runFollower runs the FSM for a follower
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()
		}
	}
}