text: // Copyright 2015 CoreOS, Inc.
  //
  // Licensed under the Apache License, Version 2.0 (the "License");
  // you may not use this file except in compliance with the License.
  // You may obtain a copy of the License at
  //
  //     http://www.apache.org/licenses/LICENSE-2.0
  //
  // Unless required by applicable law or agreed to in writing, software
  // distributed under the License is distributed on an "AS IS" BASIS,
  // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  // See the License for the specific language governing permissions and
  // limitations under the License.
  
  package raft
  
  import (
  	"errors"
  
  	"github.com/coreos/etcd/Godeps/_workspace/src/golang.org/x/net/context"
  	pb "github.com/coreos/etcd/raft/raftpb"
  )
  
  type SnapshotStatus int
  
  const (
  	SnapshotFinish  SnapshotStatus = 1
  	SnapshotFailure SnapshotStatus = 2
  )
  
  var (
  	emptyState = pb.HardState{}
  
  	// ErrStopped is returned by methods on Nodes that have been stopped.
  	ErrStopped = errors.New("raft: stopped")
  )
  
  // SoftState provides state that is useful for logging and debugging.
  // The state is volatile and does not need to be persisted to the WAL.
  type SoftState struct {
  	Lead      uint64
  	RaftState StateType
  }
  
  func (a *SoftState) equal(b *SoftState) bool {
  	return a.Lead == b.Lead && a.RaftState == b.RaftState
  }
  
  // Ready encapsulates the entries and messages that are ready to read,
  // be saved to stable storage, committed or sent to other peers.
  // All fields in Ready are read-only.
  type Ready struct {
  	// The current volatile state of a Node.
  	// SoftState will be nil if there is no update.
  	// It is not required to consume or store SoftState.
  	*SoftState
  
  	// The current state of a Node to be saved to stable storage BEFORE
  	// Messages are sent.
  	// HardState will be equal to empty state if there is no update.
  	pb.HardState
  
  	// Entries specifies entries to be saved to stable storage BEFORE
  	// Messages are sent.
  	Entries []pb.Entry
  
  	// Snapshot specifies the snapshot to be saved to stable storage.
  	Snapshot pb.Snapshot
  
  	// CommittedEntries specifies entries to be committed to a
  	// store/state-machine. These have previously been committed to stable
  	// store.
  	CommittedEntries []pb.Entry
  
  	// Messages specifies outbound messages to be sent AFTER Entries are
  	// committed to stable storage.
  	// If it contains a MsgSnap message, the application MUST report back to raft
  	// when the snapshot has been received or has failed by calling ReportSnapshot.
  	Messages []pb.Message
  }
  
  func isHardStateEqual(a, b pb.HardState) bool {
  	return a.Term == b.Term && a.Vote == b.Vote && a.Commit == b.Commit
  }
  
  // IsEmptyHardState returns true if the given HardState is empty.
  func IsEmptyHardState(st pb.HardState) bool {
  	return isHardStateEqual(st, emptyState)
  }
  
  // IsEmptySnap returns true if the given Snapshot is empty.
  func IsEmptySnap(sp pb.Snapshot) bool {
  	return sp.Metadata.Index == 0
  }
  
  func (rd Ready) containsUpdates() bool {
  	return rd.SoftState != nil || !IsEmptyHardState(rd.HardState) ||
  		!IsEmptySnap(rd.Snapshot) || len(rd.Entries) > 0 ||
  		len(rd.CommittedEntries) > 0 || len(rd.Messages) > 0
  }
  
  // Node represents a node in a raft cluster.
  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()
  	// Campaign causes the Node to transition to candidate state and start campaigning to become leader.
  	Campaign(ctx context.Context) error
  	// Propose proposes that data be appended to the log.
  	Propose(ctx context.Context, data []byte) error
  	// ProposeConfChange proposes config change.
  	// At most one ConfChange can be in the process of going through consensus.
  	// Application needs to call ApplyConfChange when applying EntryConfChange type entry.
  	ProposeConfChange(ctx context.Context, cc pb.ConfChange) error
  	// Step advances the state machine using the given message. ctx.Err() will be returned, if any.
  	Step(ctx context.Context, msg pb.Message) error
  	// Ready returns a channel that returns the current point-in-time state
  	// Users of the Node must call Advance after applying the state returned by Ready
  	Ready() <-chan Ready
  	// Advance notifies the Node that the application has applied and saved progress up to the last Ready.
  	// It prepares the node to return the next available Ready.
  	Advance()
  	// ApplyConfChange applies config change to the local node.
  	// Returns an opaque ConfState protobuf which must be recorded
  	// in snapshots. Will never return nil; it returns a pointer only
  	// to match MemoryStorage.Compact.
  	ApplyConfChange(cc pb.ConfChange) *pb.ConfState
  	// Status returns the current status of the raft state machine.
  	Status() Status
  	// Report reports the given node is not reachable for the last send.
  	ReportUnreachable(id uint64)
  	// ReportSnapshot reports the stutus of the sent snapshot.
  	ReportSnapshot(id uint64, status SnapshotStatus)
  	// Stop performs any necessary termination of the Node
  	Stop()
  }
  
  type Peer struct {
  	ID      uint64
  	Context []byte
  }
  
  // StartNode returns a new Node given configuration and a list of raft peers.
  // It appends a ConfChangeAddNode entry for each given peer to the initial log.
  func StartNode(c *Config, peers []Peer) Node {
  	r := newRaft(c)
  	// become the follower at term 1 and apply initial configuration
  	// entires of term 1
  	r.becomeFollower(1, None)
  	for _, peer := range peers {
  		cc := pb.ConfChange{Type: pb.ConfChangeAddNode, NodeID: peer.ID, Context: peer.Context}
  		d, err := cc.Marshal()
  		if err != nil {
  			panic("unexpected marshal error")
  		}
  		e := pb.Entry{Type: pb.EntryConfChange, Term: 1, Index: r.raftLog.lastIndex() + 1, Data: d}
  		r.raftLog.append(e)
  	}
  	// Mark these initial entries as committed.
  	// TODO(bdarnell): These entries are still unstable; do we need to preserve
  	// the invariant that committed < unstable?
  	r.raftLog.committed = r.raftLog.lastIndex()
  	r.Commit = r.raftLog.committed
  	// Now apply them, mainly so that the application can call Campaign
  	// immediately after StartNode in tests. Note that these nodes will
  	// be added to raft twice: here and when the application's Ready
  	// loop calls ApplyConfChange. The calls to addNode must come after
  	// all calls to raftLog.append so progress.next is set after these
  	// bootstrapping entries (it is an error if we try to append these
  	// entries since they have already been committed).
  	// We do not set raftLog.applied so the application will be able
  	// to observe all conf changes via Ready.CommittedEntries.
  	for _, peer := range peers {
  		r.addNode(peer.ID)
  	}
  
  	n := newNode()
  	go n.run(r)
  	return &n
  }
  
  // RestartNode is similar to StartNode but does not take a list of peers.
  // The current membership of the cluster will be restored from the Storage.
  // If the caller has an existing state machine, pass in the last log index that
  // has been applied to it; otherwise use zero.
  func RestartNode(c *Config) Node {
  	r := newRaft(c)
  
  	n := newNode()
  	go n.run(r)
  	return &n
  }
  
  // node is the canonical implementation of the Node interface
  type node struct {
  	propc      chan pb.Message
  	recvc      chan pb.Message
  	confc      chan pb.ConfChange
  	confstatec chan pb.ConfState
  	readyc     chan Ready
  	advancec   chan struct{}
  	tickc      chan struct{}
  	done       chan struct{}
  	stop       chan struct{}
  	status     chan chan Status
  }
  
  func newNode() node {
  	return node{
  		propc:      make(chan pb.Message),
  		recvc:      make(chan pb.Message),
  		confc:      make(chan pb.ConfChange),
  		confstatec: make(chan pb.ConfState),
  		readyc:     make(chan Ready),
  		advancec:   make(chan struct{}),
  		tickc:      make(chan struct{}),
  		done:       make(chan struct{}),
  		stop:       make(chan struct{}),
  		status:     make(chan chan Status),
  	}
  }
  
  func (n *node) Stop() {
  	select {
  	case n.stop <- struct{}{}:
  		// Not already stopped, so trigger it
  	case <-n.done:
  		// Node has already been stopped - no need to do anything
  		return
  	}
  	// Block until the stop has been acknowledged by run()
  	<-n.done
  }
  
  func (n *node) run(r *raft) {
  	var propc chan pb.Message
  	var readyc chan Ready
  	var advancec chan struct{}
  	var prevLastUnstablei, prevLastUnstablet uint64
  	var havePrevLastUnstablei bool
  	var prevSnapi uint64
  	var rd Ready
  
  	lead := None
  	prevSoftSt := r.softState()
  	prevHardSt := emptyState
  
  	for {
  		if advancec != nil {
  			readyc = nil
  		} else {
  			rd = newReady(r, prevSoftSt, prevHardSt)
  			if rd.containsUpdates() {
  				readyc = n.readyc
  			} else {
  				readyc = nil
  			}
  		}
  
  		if lead != r.lead {
  			if r.hasLeader() {
  				if lead == None {
  					raftLogger.Infof("raft.node: %x elected leader %x at term %d", r.id, r.lead, r.Term)
  				} else {
  					raftLogger.Infof("raft.node: %x changed leader from %x to %x at term %d", r.id, lead, r.lead, r.Term)
  				}
  				propc = n.propc
  			} else {
  				raftLogger.Infof("raft.node: %x lost leader %x at term %d", r.id, lead, r.Term)
  				propc = nil
  			}
  			lead = r.lead
  		}
  
  		select {
  		// TODO: maybe buffer the config propose if there exists one (the way
  		// described in raft dissertation)
  		// Currently it is dropped in Step silently.
  		case m := <-propc:
  			m.From = r.id
  			r.Step(m)
  		case m := <-n.recvc:
  			// filter out response message from unknown From.
  			if _, ok := r.prs[m.From]; ok || !IsResponseMsg(m) {
  				r.Step(m) // raft never returns an error
  			}
  		case cc := <-n.confc:
  			if cc.NodeID == None {
  				r.resetPendingConf()
  				select {
  				case n.confstatec <- pb.ConfState{Nodes: r.nodes()}:
  				case <-n.done:
  				}
  				break
  			}
  			switch cc.Type {
  			case pb.ConfChangeAddNode:
  				r.addNode(cc.NodeID)
  			case pb.ConfChangeRemoveNode:
  				// block incoming proposal when local node is
  				// removed
  				if cc.NodeID == r.id {
  					n.propc = nil
  				}
  				r.removeNode(cc.NodeID)
  			case pb.ConfChangeUpdateNode:
  				r.resetPendingConf()
  			default:
  				panic("unexpected conf type")
  			}
  			select {
  			case n.confstatec <- pb.ConfState{Nodes: r.nodes()}:
  			case <-n.done:
  			}
  		case <-n.tickc:
  			r.tick()
  		case readyc <- rd:
  			if rd.SoftState != nil {
  				prevSoftSt = rd.SoftState
  			}
  			if len(rd.Entries) > 0 {
  				prevLastUnstablei = rd.Entries[len(rd.Entries)-1].Index
  				prevLastUnstablet = rd.Entries[len(rd.Entries)-1].Term
  				havePrevLastUnstablei = true
  			}
  			if !IsEmptyHardState(rd.HardState) {
  				prevHardSt = rd.HardState
  			}
  			if !IsEmptySnap(rd.Snapshot) {
  				prevSnapi = rd.Snapshot.Metadata.Index
  			}
  			r.msgs = nil
  			advancec = n.advancec
  		case <-advancec:
  			if prevHardSt.Commit != 0 {
  				r.raftLog.appliedTo(prevHardSt.Commit)
  			}
  			if havePrevLastUnstablei {
  				r.raftLog.stableTo(prevLastUnstablei, prevLastUnstablet)
  				havePrevLastUnstablei = false
  			}
  			r.raftLog.stableSnapTo(prevSnapi)
  			advancec = nil
  		case c := <-n.status:
  			c <- getStatus(r)
  		case <-n.stop:
  			close(n.done)
  			return
  		}
  	}
  }
  
  // Tick increments the internal logical clock for this Node. Election timeouts
  // and heartbeat timeouts are in units of ticks.
  func (n *node) Tick() {
  	select {
  	case n.tickc <- struct{}{}:
  	case <-n.done:
  	}
  }
  
  func (n *node) Campaign(ctx context.Context) error { return n.step(ctx, pb.Message{Type: pb.MsgHup}) }
  
  func (n *node) Propose(ctx context.Context, data []byte) error {
  	return n.step(ctx, pb.Message{Type: pb.MsgProp, Entries: []pb.Entry{{Data: data}}})
  }
  
  func (n *node) Step(ctx context.Context, m pb.Message) error {
  	// ignore unexpected local messages receiving over network
  	if IsLocalMsg(m) {
  		// TODO: return an error?
  		return nil
  	}
  	return n.step(ctx, m)
  }
  
  func (n *node) ProposeConfChange(ctx context.Context, cc pb.ConfChange) error {
  	data, err := cc.Marshal()
  	if err != nil {
  		return err
  	}
  	return n.Step(ctx, pb.Message{Type: pb.MsgProp, Entries: []pb.Entry{{Type: pb.EntryConfChange, Data: data}}})
  }
  
  // Step advances the state machine using msgs. The ctx.Err() will be returned,
  // if any.
  func (n *node) step(ctx context.Context, m pb.Message) error {
  	ch := n.recvc
  	if m.Type == pb.MsgProp {
  		ch = n.propc
  	}
  
  	select {
  	case ch <- m:
  		return nil
  	case <-ctx.Done():
  		return ctx.Err()
  	case <-n.done:
  		return ErrStopped
  	}
  }
  
  func (n *node) Ready() <-chan Ready { return n.readyc }
  
  func (n *node) Advance() {
  	select {
  	case n.advancec <- struct{}{}:
  	case <-n.done:
  	}
  }
  
  func (n *node) ApplyConfChange(cc pb.ConfChange) *pb.ConfState {
  	var cs pb.ConfState
  	select {
  	case n.confc <- cc:
  	case <-n.done:
  	}
  	select {
  	case cs = <-n.confstatec:
  	case <-n.done:
  	}
  	return &cs
  }
  
  func (n *node) Status() Status {
  	c := make(chan Status)
  	n.status <- c
  	return <-c
  }
  
  func (n *node) ReportUnreachable(id uint64) {
  	select {
  	case n.recvc <- pb.Message{Type: pb.MsgUnreachable, From: id}:
  	case <-n.done:
  	}
  }
  
  func (n *node) ReportSnapshot(id uint64, status SnapshotStatus) {
  	rej := status == SnapshotFailure
  
  	select {
  	case n.recvc <- pb.Message{Type: pb.MsgSnapStatus, From: id, Reject: rej}:
  	case <-n.done:
  	}
  }
  
  func newReady(r *raft, prevSoftSt *SoftState, prevHardSt pb.HardState) Ready {
  	rd := Ready{
  		Entries:          r.raftLog.unstableEntries(),
  		CommittedEntries: r.raftLog.nextEnts(),
  		Messages:         r.msgs,
  	}
  	if softSt := r.softState(); !softSt.equal(prevSoftSt) {
  		rd.SoftState = softSt
  	}
  	if !isHardStateEqual(r.HardState, prevHardSt) {
  		rd.HardState = r.HardState
  	}
  	if r.raftLog.unstable.snapshot != nil {
  		rd.Snapshot = *r.raftLog.unstable.snapshot
  	}
  	return rd
  }
  