func (m *Memberlist) deadNode(d *dead) {
	m.nodeLock.Lock()
	defer m.nodeLock.Unlock()
	state, ok := m.nodeMap[d.Node]
	if !ok {
		return
	}
	if d.Incarnation < state.Incarnation {
		return
	}
	if state.State == stateDead {
		return
	}
	if state.Name == m.config.Name {
		if !m.leave {
			inc := m.nextIncarnation()
			for d.Incarnation >= inc {
				inc = m.nextIncarnation()
			}
			state.Incarnation = inc
			a := alive{
				Incarnation: inc,
				Node:        state.Name,
				Addr:        state.Addr,
				Port:        state.Port,
				Meta:        state.Meta,
				Vsn: []uint8{
					state.PMin, state.PMax, state.PCur,
					state.DMin, state.DMax, state.DCur,
				},
			}
			m.encodeAndBroadcast(d.Node, aliveMsg, a)
			m.logger.Printf("[WARN] memberlist: Refuting a dead message (from: %s)", d.From)
			return  
		}
		m.encodeBroadcastNotify(d.Node, deadMsg, d, m.leaveBroadcast)
	} else {
		m.encodeAndBroadcast(d.Node, deadMsg, d)
	}
	metrics.IncrCounter([]string{"memberlist", "msg", "dead"}, 1)
	state.Incarnation = d.Incarnation
	state.State = stateDead
	state.StateChange = time.Now()
	if m.config.Events != nil {
		m.config.Events.NotifyLeave(&state.Node)
	}
}
deadNode is invoked by the network layer when we get a message
about a dead node
package memberlist
import (
	"bytes"
	"fmt"
	"math"
	"math/rand"
	"net"
	"sync/atomic"
	"time"
	"github.com/socketplane/socketplane/Godeps/_workspace/src/github.com/armon/go-metrics"
)
type nodeStateType int
const (
	stateAlive nodeStateType = iota
	stateSuspect
	stateDead
)
type Node struct {
	Name string
	Addr net.IP
	Port uint16
	Meta []byte  
	PMin uint8   
	PMax uint8   
	PCur uint8   
	DMin uint8   
	DMax uint8   
	DCur uint8   
}
type nodeState struct {
	Node
	Incarnation uint32         
	State       nodeStateType  
	StateChange time.Time      
}
type ackHandler struct {
	handler func()
	timer   *time.Timer
}
func (m *Memberlist) schedule() {
	m.tickerLock.Lock()
	defer m.tickerLock.Unlock()
	if len(m.tickers) > 0 {
		return
	}
	stopCh := make(chan struct{})
	if m.config.ProbeInterval > 0 {
		t := time.NewTicker(m.config.ProbeInterval)
		go m.triggerFunc(m.config.ProbeInterval, t.C, stopCh, m.probe)
		m.tickers = append(m.tickers, t)
	}
	if m.config.PushPullInterval > 0 {
		go m.pushPullTrigger(stopCh)
	}
	if m.config.GossipInterval > 0 && m.config.GossipNodes > 0 {
		t := time.NewTicker(m.config.GossipInterval)
		go m.triggerFunc(m.config.GossipInterval, t.C, stopCh, m.gossip)
		m.tickers = append(m.tickers, t)
	}
	if len(m.tickers) > 0 {
		m.stopTick = stopCh
	}
}
func (m *Memberlist) triggerFunc(stagger time.Duration, C <-chan time.Time, stop <-chan struct{}, f func()) {
	randStagger := time.Duration(uint64(rand.Int63()) % uint64(stagger))
	select {
	case <-time.After(randStagger):
	case <-stop:
		return
	}
	for {
		select {
		case <-C:
			f()
		case <-stop:
			return
		}
	}
}
func (m *Memberlist) pushPullTrigger(stop <-chan struct{}) {
	interval := m.config.PushPullInterval
	randStagger := time.Duration(uint64(rand.Int63()) % uint64(interval))
	select {
	case <-time.After(randStagger):
	case <-stop:
		return
	}
	for {
		m.nodeLock.RLock()
		tickTime := pushPullScale(interval, len(m.nodes))
		m.nodeLock.RUnlock()
		select {
		case <-time.After(tickTime):
			m.pushPull()
		case <-stop:
			return
		}
	}
}
func (m *Memberlist) deschedule() {
	m.tickerLock.Lock()
	defer m.tickerLock.Unlock()
	if len(m.tickers) == 0 {
		return
	}
	close(m.stopTick)
	for _, t := range m.tickers {
		t.Stop()
	}
	m.tickers = nil
}
func (m *Memberlist) probe() {
	numCheck := 0
START:
	m.nodeLock.RLock()
	if numCheck >= len(m.nodes) {
		m.nodeLock.RUnlock()
		return
	}
	if m.probeIndex >= len(m.nodes) {
		m.nodeLock.RUnlock()
		m.resetNodes()
		m.probeIndex = 0
		numCheck++
		goto START
	}
	skip := false
	var node nodeState
	node = *m.nodes[m.probeIndex]
	if node.Name == m.config.Name {
		skip = true
	} else if node.State == stateDead {
		skip = true
	}
	m.nodeLock.RUnlock()
	m.probeIndex++
	if skip {
		numCheck++
		goto START
	}
	m.probeNode(&node)
}
func (m *Memberlist) probeNode(node *nodeState) {
	defer metrics.MeasureSince([]string{"memberlist", "probeNode"}, time.Now())
	ping := ping{SeqNo: m.nextSeqNo(), Node: node.Name}
	destAddr := &net.UDPAddr{IP: node.Addr, Port: int(node.Port)}
	ackCh := make(chan bool, m.config.IndirectChecks+1)
	m.setAckChannel(ping.SeqNo, ackCh, m.config.ProbeInterval)
	if err := m.encodeAndSendMsg(destAddr, pingMsg, &ping); err != nil {
		m.logger.Printf("[ERR] memberlist: Failed to send ping: %s", err)
		return
	}
	select {
	case v := <-ackCh:
		if v == true {
			return
		}
		if v == false {
			ackCh <- v
		}
	case <-time.After(m.config.ProbeTimeout):
	}
	m.nodeLock.RLock()
	excludes := []string{m.config.Name, node.Name}
	kNodes := kRandomNodes(m.config.IndirectChecks, excludes, m.nodes)
	m.nodeLock.RUnlock()
	ind := indirectPingReq{SeqNo: ping.SeqNo, Target: node.Addr, Port: node.Port, Node: node.Name}
	for _, peer := range kNodes {
		destAddr := &net.UDPAddr{IP: peer.Addr, Port: int(peer.Port)}
		if err := m.encodeAndSendMsg(destAddr, indirectPingMsg, &ind); err != nil {
			m.logger.Printf("[ERR] memberlist: Failed to send indirect ping: %s", err)
		}
	}
	select {
	case v := <-ackCh:
		if v == true {
			return
		}
	}
	m.logger.Printf("[INFO] memberlist: Suspect %s has failed, no acks received", node.Name)
	s := suspect{Incarnation: node.Incarnation, Node: node.Name, From: m.config.Name}
	m.suspectNode(&s)
}
func (m *Memberlist) resetNodes() {
	m.nodeLock.Lock()
	defer m.nodeLock.Unlock()
	deadIdx := moveDeadNodes(m.nodes)
	for i := deadIdx; i < len(m.nodes); i++ {
		delete(m.nodeMap, m.nodes[i].Name)
		m.nodes[i] = nil
	}
	m.nodes = m.nodes[0:deadIdx]
	shuffleNodes(m.nodes)
}
func (m *Memberlist) gossip() {
	defer metrics.MeasureSince([]string{"memberlist", "gossip"}, time.Now())
	m.nodeLock.RLock()
	excludes := []string{m.config.Name}
	kNodes := kRandomNodes(m.config.GossipNodes, excludes, m.nodes)
	m.nodeLock.RUnlock()
	bytesAvail := udpSendBuf - compoundHeaderOverhead
	if m.config.EncryptionEnabled() {
		bytesAvail -= encryptOverhead(m.encryptionVersion())
	}
	for _, node := range kNodes {
		msgs := m.getBroadcasts(compoundOverhead, bytesAvail)
		if len(msgs) == 0 {
			return
		}
		compound := makeCompoundMessage(msgs)
		destAddr := &net.UDPAddr{IP: node.Addr, Port: int(node.Port)}
		if err := m.rawSendMsg(destAddr, compound.Bytes()); err != nil {
			m.logger.Printf("[ERR] memberlist: Failed to send gossip to %s: %s", destAddr, err)
		}
	}
}
func (m *Memberlist) pushPull() {
	m.nodeLock.RLock()
	excludes := []string{m.config.Name}
	nodes := kRandomNodes(1, excludes, m.nodes)
	m.nodeLock.RUnlock()
	if len(nodes) == 0 {
		return
	}
	node := nodes[0]
	if err := m.pushPullNode(node.Addr, node.Port, false); err != nil {
		m.logger.Printf("[ERR] memberlist: Push/Pull with %s failed: %s", node.Name, err)
	}
}
func (m *Memberlist) pushPullNode(addr []byte, port uint16, join bool) error {
	defer metrics.MeasureSince([]string{"memberlist", "pushPullNode"}, time.Now())
	remote, userState, err := m.sendAndReceiveState(addr, port, join)
	if err != nil {
		return err
	}
	if err := m.verifyProtocol(remote); err != nil {
		return err
	}
	if join && m.config.Merge != nil {
		nodes := make([]*Node, len(remote))
		for idx, n := range remote {
			nodes[idx] = &Node{
				Name: n.Name,
				Addr: n.Addr,
				Port: n.Port,
				Meta: n.Meta,
				PMin: n.Vsn[0],
				PMax: n.Vsn[1],
				PCur: n.Vsn[2],
				DMin: n.Vsn[3],
				DMax: n.Vsn[4],
				DCur: n.Vsn[5],
			}
		}
		if m.config.Merge.NotifyMerge(nodes) {
			m.logger.Printf("[WARN] memberlist: Cluster merge canceled")
			return fmt.Errorf("Merge canceled")
		}
	}
	m.mergeState(remote)
	if m.config.Delegate != nil {
		m.config.Delegate.MergeRemoteState(userState, join)
	}
	return nil
}
func (m *Memberlist) verifyProtocol(remote []pushNodeState) error {
	m.nodeLock.RLock()
	defer m.nodeLock.RUnlock()
	var maxpmin, minpmax uint8
	var maxdmin, mindmax uint8
	minpmax = math.MaxUint8
	mindmax = math.MaxUint8
	for _, rn := range remote {
		if rn.State != stateAlive {
			continue
		}
		if len(rn.Vsn) == 0 {
			continue
		}
		if rn.Vsn[0] > maxpmin {
			maxpmin = rn.Vsn[0]
		}
		if rn.Vsn[1] < minpmax {
			minpmax = rn.Vsn[1]
		}
		if rn.Vsn[3] > maxdmin {
			maxdmin = rn.Vsn[3]
		}
		if rn.Vsn[4] < mindmax {
			mindmax = rn.Vsn[4]
		}
	}
	for _, n := range m.nodes {
		if n.State != stateAlive {
			continue
		}
		if n.PMin > maxpmin {
			maxpmin = n.PMin
		}
		if n.PMax < minpmax {
			minpmax = n.PMax
		}
		if n.DMin > maxdmin {
			maxdmin = n.DMin
		}
		if n.DMax < mindmax {
			mindmax = n.DMax
		}
	}
	for _, n := range remote {
		var nPCur, nDCur uint8
		if len(n.Vsn) > 0 {
			nPCur = n.Vsn[2]
			nDCur = n.Vsn[5]
		}
		if nPCur < maxpmin || nPCur > minpmax {
			return fmt.Errorf(
				"Node '%s' protocol version (%d) is incompatible: [%d, %d]",
				n.Name, nPCur, maxpmin, minpmax)
		}
		if nDCur < maxdmin || nDCur > mindmax {
			return fmt.Errorf(
				"Node '%s' delegate protocol version (%d) is incompatible: [%d, %d]",
				n.Name, nDCur, maxdmin, mindmax)
		}
	}
	for _, n := range m.nodes {
		nPCur := n.PCur
		nDCur := n.DCur
		if nPCur < maxpmin || nPCur > minpmax {
			return fmt.Errorf(
				"Node '%s' protocol version (%d) is incompatible: [%d, %d]",
				n.Name, nPCur, maxpmin, minpmax)
		}
		if nDCur < maxdmin || nDCur > mindmax {
			return fmt.Errorf(
				"Node '%s' delegate protocol version (%d) is incompatible: [%d, %d]",
				n.Name, nDCur, maxdmin, mindmax)
		}
	}
	return nil
}
func (m *Memberlist) nextSeqNo() uint32 {
	return atomic.AddUint32(&m.sequenceNum, 1)
}
func (m *Memberlist) nextIncarnation() uint32 {
	return atomic.AddUint32(&m.incarnation, 1)
}
func (m *Memberlist) setAckChannel(seqNo uint32, ch chan bool, timeout time.Duration) {
	handler := func() {
		select {
		case ch <- true:
		default:
		}
	}
	ah := &ackHandler{handler, nil}
	m.ackLock.Lock()
	m.ackHandlers[seqNo] = ah
	m.ackLock.Unlock()
	ah.timer = time.AfterFunc(timeout, func() {
		m.ackLock.Lock()
		delete(m.ackHandlers, seqNo)
		m.ackLock.Unlock()
		select {
		case ch <- false:
		default:
		}
	})
}
func (m *Memberlist) setAckHandler(seqNo uint32, handler func(), timeout time.Duration) {
	ah := &ackHandler{handler, nil}
	m.ackLock.Lock()
	m.ackHandlers[seqNo] = ah
	m.ackLock.Unlock()
	ah.timer = time.AfterFunc(timeout, func() {
		m.ackLock.Lock()
		delete(m.ackHandlers, seqNo)
		m.ackLock.Unlock()
	})
}
func (m *Memberlist) invokeAckHandler(seqNo uint32) {
	m.ackLock.Lock()
	ah, ok := m.ackHandlers[seqNo]
	delete(m.ackHandlers, seqNo)
	m.ackLock.Unlock()
	if !ok {
		return
	}
	ah.timer.Stop()
	ah.handler()
}
func (m *Memberlist) aliveNode(a *alive, notify chan struct{}, bootstrap bool) {
	m.nodeLock.Lock()
	defer m.nodeLock.Unlock()
	state, ok := m.nodeMap[a.Node]
	if m.leave && a.Node == m.config.Name {
		return
	}
	if !ok {
		state = &nodeState{
			Node: Node{
				Name: a.Node,
				Addr: a.Addr,
				Port: a.Port,
				Meta: a.Meta,
			},
			State: stateDead,
		}
		m.nodeMap[a.Node] = state
		n := len(m.nodes)
		offset := randomOffset(n)
		m.nodes = append(m.nodes, state)
		m.nodes[offset], m.nodes[n] = m.nodes[n], m.nodes[offset]
	}
	if !bytes.Equal([]byte(state.Addr), a.Addr) || state.Port != a.Port {
		m.logger.Printf("[ERR] memberlist: Conflicting address for %s. Mine: %v:%d Theirs: %v:%d",
			state.Name, state.Addr, state.Port, net.IP(a.Addr), a.Port)
		if m.config.Conflict != nil {
			other := Node{
				Name: a.Node,
				Addr: a.Addr,
				Port: a.Port,
				Meta: a.Meta,
			}
			m.config.Conflict.NotifyConflict(&state.Node, &other)
		}
		return
	}
	isLocalNode := state.Name == m.config.Name
	if a.Incarnation <= state.Incarnation && !isLocalNode {
		return
	}
	if a.Incarnation < state.Incarnation && isLocalNode {
		return
	}
	metrics.IncrCounter([]string{"memberlist", "msg", "alive"}, 1)
	oldState := state.State
	oldMeta := state.Meta
	if !bootstrap && isLocalNode {
		versions := []uint8{
			state.PMin, state.PMax, state.PCur,
			state.DMin, state.DMax, state.DCur,
		}
		if a.Incarnation == state.Incarnation &&
			bytes.Equal(a.Meta, state.Meta) &&
			bytes.Equal(a.Vsn, versions) {
			return
		}
		inc := m.nextIncarnation()
		for a.Incarnation >= inc {
			inc = m.nextIncarnation()
		}
		state.Incarnation = inc
		a := alive{
			Incarnation: inc,
			Node:        state.Name,
			Addr:        state.Addr,
			Port:        state.Port,
			Meta:        state.Meta,
			Vsn:         versions,
		}
		m.encodeBroadcastNotify(a.Node, aliveMsg, a, notify)
		m.logger.Printf("[WARN] memberlist: Refuting an alive message")
	} else {
		m.encodeBroadcastNotify(a.Node, aliveMsg, a, notify)
		if len(a.Vsn) > 0 {
			state.PMin = a.Vsn[0]
			state.PMax = a.Vsn[1]
			state.PCur = a.Vsn[2]
			state.DMin = a.Vsn[3]
			state.DMax = a.Vsn[4]
			state.DCur = a.Vsn[5]
		}
		state.Incarnation = a.Incarnation
		state.Meta = a.Meta
		if state.State != stateAlive {
			state.State = stateAlive
			state.StateChange = time.Now()
		}
	}
	if m.config.Events != nil {
		if oldState == stateDead {
			m.config.Events.NotifyJoin(&state.Node)
		} else if !bytes.Equal(oldMeta, state.Meta) {
			m.config.Events.NotifyUpdate(&state.Node)
		}
	}
}
func (m *Memberlist) suspectNode(s *suspect) {
	m.nodeLock.Lock()
	defer m.nodeLock.Unlock()
	state, ok := m.nodeMap[s.Node]
	if !ok {
		return
	}
	if s.Incarnation < state.Incarnation {
		return
	}
	if state.State != stateAlive {
		return
	}
	if state.Name == m.config.Name {
		inc := m.nextIncarnation()
		for s.Incarnation >= inc {
			inc = m.nextIncarnation()
		}
		state.Incarnation = inc
		a := alive{
			Incarnation: inc,
			Node:        state.Name,
			Addr:        state.Addr,
			Port:        state.Port,
			Meta:        state.Meta,
			Vsn: []uint8{
				state.PMin, state.PMax, state.PCur,
				state.DMin, state.DMax, state.DCur,
			},
		}
		m.encodeAndBroadcast(s.Node, aliveMsg, a)
		m.logger.Printf("[WARN] memberlist: Refuting a suspect message (from: %s)", s.From)
		return  
	} else {
		m.encodeAndBroadcast(s.Node, suspectMsg, s)
	}
	metrics.IncrCounter([]string{"memberlist", "msg", "suspect"}, 1)
	state.Incarnation = s.Incarnation
	state.State = stateSuspect
	changeTime := time.Now()
	state.StateChange = changeTime
	timeout := suspicionTimeout(m.config.SuspicionMult, len(m.nodes), m.config.ProbeInterval)
	time.AfterFunc(timeout, func() {
		m.nodeLock.Lock()
		state, ok := m.nodeMap[s.Node]
		timeout := ok && state.State == stateSuspect && state.StateChange == changeTime
		m.nodeLock.Unlock()
		if timeout {
			m.suspectTimeout(state)
		}
	})
}
func (m *Memberlist) suspectTimeout(n *nodeState) {
	m.logger.Printf("[INFO] memberlist: Marking %s as failed, suspect timeout reached", n.Name)
	d := dead{Incarnation: n.Incarnation, Node: n.Name, From: m.config.Name}
	m.deadNode(&d)
}