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()
	})
}
setAckHandler is used to attach a handler to be invoked when an
ack with a given sequence number is received. If a timeout is reached,
the handler is deleted
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:
		}
	})
}