func (c *clusterState) updateCluster() {
	defer c.stats.Latency(stats.SchedUpdateClusterLatency_ms).Time().Stop()
	allUpdates := []cc.NodeUpdate{}
LOOP:
	for i := 0; i < common.DefaultClusterChanSize; i++ {
		select {
		case updates := <-c.nodesUpdatesCh:
			allUpdates = append(allUpdates, updates...)
		default:
			break LOOP
		}
	}
	c.update(allUpdates)
}
update cluster state to reflect added and removed nodes
it will process at most DefaultClusterSize sets of updates with each call
package server
import (
	"fmt"
	"time"
	"github.com/davecgh/go-spew/spew"
	log "github.com/sirupsen/logrus"
	cc "github.com/twitter/scoot/cloud/cluster"
	"github.com/twitter/scoot/common"
	"github.com/twitter/scoot/common/stats"
)
const noJob = ""
const noTask = ""
const defaultMaxLostDuration = time.Minute
const defaultMaxFlakyDuration = 15 * time.Minute
var nilTime = time.Time{}
type ReadyFn func(cc.Node) (ready bool, backoffDuration time.Duration)
type clusterState struct {
	nodesUpdatesCh   chan []cc.NodeUpdate
	nodes            map[cc.NodeId]*nodeState  
	suspendedNodes   map[cc.NodeId]*nodeState  
	offlinedNodes    map[cc.NodeId]*nodeState  
	nodeGroups       map[string]*nodeGroup     
	maxLostDuration  time.Duration             
	maxFlakyDuration time.Duration             
	readyFn          ReadyFn                   
	numRunning       int                       
	stats            stats.StatsReceiver       
	nopUpdateCnt     int
}
func (c *clusterState) isOfflined(ns *nodeState) bool {
	if _, ok := c.offlinedNodes[ns.node.Id()]; ok {
		return true
	}
	return false
}
type nodeGroup struct {
	idle map[cc.NodeId]*nodeState
	busy map[cc.NodeId]*nodeState
}
func newNodeGroup() *nodeGroup {
	return &nodeGroup{idle: map[cc.NodeId]*nodeState{}, busy: map[cc.NodeId]*nodeState{}}
}
type nodeState struct {
	node        cc.Node
	runningJob  string
	runningTask string
	snapshotId  string
	timeLost    time.Time         
	timeFlaky   time.Time         
	readyCh     chan interface{}  
	removedCh   chan interface{}  
}
func (n *nodeState) String() string {
	return fmt.Sprintf("{node:%s, jobId:%s, taskId:%s, snapshotId:%s, timeLost:%v, timeFlaky:%v, ready:%t}",
		spew.Sdump(n.node), n.runningJob, n.runningTask, n.snapshotId, n.timeLost, n.timeFlaky, (n.readyCh == nil))
}
func (ns *nodeState) suspended() bool {
	return ns.readyCh != nil || ns.timeLost != nilTime || ns.timeFlaky != nilTime
}
func (ns *nodeState) ready() bool {
	ready := (ns.readyCh == nil)
	if !ready {
		select {
		case <-ns.readyCh:
			ready = true
		default:
		}
	}
	return ready
}
func (ns *nodeState) startReadyLoop(rfn ReadyFn) {
	ns.readyCh = make(chan interface{})
	go func() {
		done := false
		for !done {
			if ready, backoff := rfn(ns.node); ready {
				close(ns.readyCh)
				done = true
			} else if backoff == 0 {
				done = true
			} else {
				select {
				case <-ns.removedCh:
					done = true
				case <-time.After(backoff):
					break
				}
			}
		}
	}()
}
func newNodeState(node cc.Node) *nodeState {
	return &nodeState{
		node:        node,
		runningJob:  noJob,
		runningTask: noTask,
		snapshotId:  "",
		timeLost:    nilTime,
		timeFlaky:   nilTime,
		readyCh:     nil,
		removedCh:   make(chan interface{}),
	}
}
func newClusterState(nodesUpdatesCh chan []cc.NodeUpdate, rfn ReadyFn, stats stats.StatsReceiver) *clusterState {
	cs := &clusterState{
		nodesUpdatesCh:   nodesUpdatesCh,
		nodes:            make(map[cc.NodeId]*nodeState),
		suspendedNodes:   map[cc.NodeId]*nodeState{},
		offlinedNodes:    make(map[cc.NodeId]*nodeState),
		nodeGroups:       map[string]*nodeGroup{"": newNodeGroup()},
		maxLostDuration:  defaultMaxLostDuration,
		maxFlakyDuration: defaultMaxFlakyDuration,
		readyFn:          rfn,
		stats:            stats,
	}
	cs.updateCluster()
	return cs
}
func (c *clusterState) numFree() int {
	return max(0, len(c.nodes)-c.numRunning)
}
func (c *clusterState) taskScheduled(nodeId cc.NodeId, jobId, taskId, snapshotId string) {
	ns := c.nodes[nodeId]
	delete(c.nodeGroups[ns.snapshotId].idle, nodeId)
	empty := len(c.nodeGroups[ns.snapshotId].idle) == 0 && len(c.nodeGroups[ns.snapshotId].busy) == 0
	if ns.snapshotId != "" && empty {
		delete(c.nodeGroups, ns.snapshotId)
	}
	if _, ok := c.nodeGroups[snapshotId]; !ok {
		c.nodeGroups[snapshotId] = newNodeGroup()
	}
	c.nodeGroups[snapshotId].busy[nodeId] = ns
	ns.runningJob = jobId
	ns.runningTask = taskId
	ns.snapshotId = snapshotId
	c.numRunning++
}
func (c *clusterState) taskCompleted(nodeId cc.NodeId, flaky bool) {
	var ns *nodeState
	var ok bool
	if ns, ok = c.nodes[nodeId]; !ok {
		ns, ok = c.suspendedNodes[nodeId]
	}
	if ok {
		if flaky && !ns.suspended() {
			delete(c.nodes, nodeId)
			c.suspendedNodes[nodeId] = ns
			ns.timeFlaky = time.Now()
		}
		ns.runningJob = noJob
		ns.runningTask = noTask
		delete(c.nodeGroups[ns.snapshotId].busy, nodeId)
		c.nodeGroups[ns.snapshotId].idle[nodeId] = ns
	} else {
		log.Infof("TaskCompleted specified an unknown node: %v (flaky=%t) (likely reaped already)", nodeId, flaky)
	}
	c.numRunning--
}
func (c *clusterState) getNodeState(nodeId cc.NodeId) (*nodeState, bool) {
	ns, ok := c.nodes[nodeId]
	return ns, ok
}