func (t *taskCollection) add(task *task) error {
	t.Lock()
	defer t.Unlock()
	if _, ok := t.table[task.id]; !ok {
		t.table[task.id] = task
	} else {
		taskLogger.WithFields(log.Fields{
			"_module": "scheduler-taskCollection",
			"_block":  "add",
			"task id": task.id,
		}).Error(ErrTaskHasAlreadyBeenAdded.Error())
		return ErrTaskHasAlreadyBeenAdded
	}
	return nil
}
Add given a reference to a task adds it to the collection of tasks.  An
error is returned if the task already exists in the collection.
package scheduler
import (
	"errors"
	"fmt"
	"net"
	"strconv"
	"sync"
	"time"
	"github.com/intelsdi-x/gomit"
	"github.com/pborman/uuid"
	log "github.com/sirupsen/logrus"
	"github.com/intelsdi-x/snap/core"
	"github.com/intelsdi-x/snap/core/scheduler_event"
	"github.com/intelsdi-x/snap/core/serror"
	"github.com/intelsdi-x/snap/grpc/controlproxy"
	"github.com/intelsdi-x/snap/pkg/schedule"
	"github.com/intelsdi-x/snap/scheduler/wmap"
)
const (
	DefaultDeadlineDuration = time.Second * 5
	DefaultStopOnFailure = 10
)
var (
	taskLogger = schedulerLogger.WithField("_module", "scheduler-task")
	ErrTaskNotFound = errors.New("Task not found")
	ErrTaskNotStopped = errors.New("Task must be stopped")
	ErrTaskHasAlreadyBeenAdded = errors.New("Task has already been added")
	ErrTaskDisabledOnFailures = errors.New("Task disabled due to consecutive failures")
	ErrTaskNotDisabled = errors.New("Task must be disabled")
)
type task struct {
	sync.Mutex  
	id                 string
	name               string
	schResponseChan    chan schedule.Response
	killChan           chan struct{}
	schedule           schedule.Schedule
	workflow           *schedulerWorkflow
	state              core.TaskState
	creationTime       time.Time
	lastFireTime       time.Time
	manager            managesWork
	metricsManager     managesMetrics
	deadlineDuration   time.Duration
	hitCount           uint
	missedIntervals    uint
	failureMutex       sync.Mutex
	failedRuns         uint
	lastFailureMessage string
	lastFailureTime    time.Time
	stopOnFailure      int
	eventEmitter       gomit.Emitter
	RemoteManagers     managers
	isStream           bool
	maxCollectDuration time.Duration
	maxMetricsBuffer   int64
}
func newTask(s schedule.Schedule, wf *schedulerWorkflow, m *workManager, mm managesMetrics, emitter gomit.Emitter, opts ...core.TaskOption) (*task, error) {
	taskID := uuid.New()
	name := fmt.Sprintf("Task-%s", taskID)
	wf.eventEmitter = emitter
	mgrs := newManagers(mm)
	err := createTaskClients(&mgrs, wf)
	if err != nil {
		return nil, err
	}
	_, stream := s.(*schedule.StreamingSchedule)
	task := &task{
		id:               taskID,
		name:             name,
		schResponseChan:  make(chan schedule.Response),
		schedule:         s,
		state:            core.TaskStopped,
		creationTime:     time.Now(),
		workflow:         wf,
		manager:          m,
		metricsManager:   mm,
		deadlineDuration: DefaultDeadlineDuration,
		stopOnFailure:    DefaultStopOnFailure,
		eventEmitter:     emitter,
		RemoteManagers:   mgrs,
		isStream:         stream,
	}
	for _, opt := range opts {
		opt(task)
	}
	return task, nil
}
func (t *task) Option(opts ...core.TaskOption) core.TaskOption {
	var previous core.TaskOption
	for _, opt := range opts {
		previous = opt(t)
	}
	return previous
}
func (t *task) MaxCollectDuration() time.Duration {
	return t.maxCollectDuration
}
func (t *task) SetMaxCollectDuration(ti time.Duration) {
	t.maxCollectDuration = ti
}
func (t *task) MaxMetricsBuffer() int64 {
	return t.maxMetricsBuffer
}
func (t *task) SetMaxMetricsBuffer(i int64) {
	t.maxMetricsBuffer = i
}
func (t *task) GetName() string {
	return t.name
}
func (t *task) SetName(name string) {
	t.name = name
}
func (t *task) CreationTime() *time.Time {
	return &t.creationTime
}
func (t *task) DeadlineDuration() time.Duration {
	return t.deadlineDuration
}
func (t *task) SetDeadlineDuration(d time.Duration) {
	t.deadlineDuration = d
}
func (t *task) SetTaskID(id string) {
	t.id = id
}
func (t *task) HitCount() uint {
	return t.hitCount
}
func (t *task) ID() string {
	return t.id
}
func (t *task) LastRunTime() *time.Time {
	return &t.lastFireTime
}
func (t *task) MissedCount() uint {
	return t.missedIntervals
}
func (t *task) FailedCount() uint {
	return t.failedRuns
}
func (t *task) LastFailureMessage() string {
	return t.lastFailureMessage
}
func (t *task) State() core.TaskState {
	return t.state
}
func (t *task) Status() WorkflowState {
	return t.workflow.State()
}
func (t *task) SetStopOnFailure(v int) {
	t.stopOnFailure = v
}
func (t *task) SetID(id string) {
	t.id = id
}
func (t *task) GetStopOnFailure() int {
	return t.stopOnFailure
}
func (t *task) Spin() {
	t.Lock()
	defer t.Unlock()
	if t.isStream {
		t.state = core.TaskSpinning
		t.killChan = make(chan struct{})
		go t.stream()
		return
	}
	t.lastFireTime = time.Time{}
	if t.state == core.TaskStopped || t.state == core.TaskEnded {
		t.state = core.TaskSpinning
		t.killChan = make(chan struct{})
		go t.spin()
	}
}
func (t *task) stream() {
	var consecutiveFailures int
	resetTime := time.Second * 3
	for {
		metricsChan, errChan, err := t.metricsManager.StreamMetrics(
			t.id,
			t.workflow.tags,
			t.maxCollectDuration,
			t.maxMetricsBuffer)
		if err != nil {
			consecutiveFailures++
			if t.stopOnFailure >= 0 && consecutiveFailures >= t.stopOnFailure {
				taskLogger.WithFields(log.Fields{
					"_block":               "stream",
					"task-id":              t.id,
					"task-name":            t.name,
					"consecutive failures": consecutiveFailures,
					"error":                t.lastFailureMessage,
				}).Error(ErrTaskDisabledOnFailures)
				t.disable(t.lastFailureMessage)
				return
			}
			time.Sleep(resetTime)
			continue
		} else {
			consecutiveFailures = 0
		}
		done := false
		for !done {
			if errChan == nil {
				break
			}
			select {
			case <-t.killChan:
				t.Lock()
				t.state = core.TaskStopped
				t.Unlock()
				done = true
				event := new(scheduler_event.TaskStoppedEvent)
				event.TaskID = t.id
				defer t.eventEmitter.Emit(event)
				return
			case mts, ok := <-metricsChan:
				if !ok {
					metricsChan = nil
					break
				}
				if len(mts) == 0 {
					continue
				}
				t.hitCount++
				consecutiveFailures = 0
				t.workflow.StreamStart(t, mts)
			case err := <-errChan:
				taskLogger.WithFields(log.Fields{
					"_block":    "stream",
					"task-id":   t.id,
					"task-name": t.name,
				}).Error("Error: " + err.Error())
				consecutiveFailures++
				if err.Error() == "connection broken" {
					time.Sleep(resetTime)
					done = true
				}
				if t.stopOnFailure >= 0 && consecutiveFailures >= t.stopOnFailure {
					taskLogger.WithFields(log.Fields{
						"_block":               "stream",
						"task-id":              t.id,
						"task-name":            t.name,
						"consecutive failures": consecutiveFailures,
						"error":                t.lastFailureMessage,
					}).Error(ErrTaskDisabledOnFailures)
					t.disable(t.lastFailureMessage)
					return
				}
			}
		}
	}
}
func (t *task) Stop() {
	t.Lock()
	defer t.Unlock()
	if t.state == core.TaskFiring || t.state == core.TaskSpinning {
		t.state = core.TaskStopping
		close(t.killChan)
	}
}
func (t *task) UnsubscribePlugins() []serror.SnapError {
	depGroups := getWorkflowPlugins(t.workflow.processNodes, t.workflow.publishNodes, t.workflow.metrics)
	var errs []serror.SnapError
	for k := range depGroups {
		event := &scheduler_event.PluginsUnsubscribedEvent{
			TaskID:  t.ID(),
			Plugins: depGroups[k].subscribedPlugins,
		}
		defer t.eventEmitter.Emit(event)
		mgr, err := t.RemoteManagers.Get(k)
		if err != nil {
			errs = append(errs, serror.New(err))
		} else {
			uerrs := mgr.UnsubscribeDeps(t.ID())
			if len(uerrs) > 0 {
				errs = append(errs, uerrs...)
			}
		}
	}
	for _, err := range errs {
		taskLogger.WithFields(log.Fields{
			"_block":     "UnsubscribePlugins",
			"task-id":    t.id,
			"task-name":  t.name,
			"task-state": t.state,
		}).Error(err)
	}
	return errs
}
func (t *task) SubscribePlugins() ([]string, []serror.SnapError) {
	depGroups := getWorkflowPlugins(t.workflow.processNodes, t.workflow.publishNodes, t.workflow.metrics)
	var subbedDeps []string
	for k := range depGroups {
		var errs []serror.SnapError
		mgr, err := t.RemoteManagers.Get(k)
		if err != nil {
			errs = append(errs, serror.New(err))
		} else {
			errs = mgr.SubscribeDeps(t.ID(), depGroups[k].requestedMetrics, depGroups[k].subscribedPlugins, t.workflow.configTree)
		}
		if len(errs) > 0 {
			for _, key := range subbedDeps {
				mgr, err := t.RemoteManagers.Get(key)
				if err != nil {
					errs = append(errs, serror.New(err))
				} else {
					uerrs := mgr.UnsubscribeDeps(t.ID())
					errs = append(errs, uerrs...)
				}
			}
			return nil, errs
		}
		subbedDeps = append(subbedDeps, k)
	}
	return subbedDeps, nil
}
func (t *task) Enable() error {
	t.Lock()
	defer t.Unlock()
	if t.state != core.TaskDisabled {
		return ErrTaskNotDisabled
	}
	t.state = core.TaskStopped
	return nil
}
func (t *task) Kill() {
	t.Lock()
	defer t.Unlock()
	if t.state == core.TaskFiring || t.state == core.TaskSpinning {
		close(t.killChan)
		t.state = core.TaskDisabled
	}
}
func (t *task) WMap() *wmap.WorkflowMap {
	return t.workflow.workflowMap
}
func (t *task) Schedule() schedule.Schedule {
	return t.schedule
}
func (t *task) spin() {
	var consecutiveFailures int
	for {
		taskLogger.Debug("task spin loop")
		go t.waitForSchedule()
		select {
		case sr := <-t.schResponseChan:
			switch sr.State() {
			case schedule.Active:
				t.missedIntervals += sr.Missed()
				t.fire()
				if t.lastFailureTime == t.lastFireTime {
					consecutiveFailures++
					taskLogger.WithFields(log.Fields{
						"_block":                    "spin",
						"task-id":                   t.id,
						"task-name":                 t.name,
						"consecutive failures":      consecutiveFailures,
						"consecutive failure limit": t.stopOnFailure,
						"error":                     t.lastFailureMessage,
					}).Warn("Task failed")
				} else {
					consecutiveFailures = 0
				}
				if t.stopOnFailure >= 0 && consecutiveFailures >= t.stopOnFailure {
					taskLogger.WithFields(log.Fields{
						"_block":               "spin",
						"task-id":              t.id,
						"task-name":            t.name,
						"consecutive failures": consecutiveFailures,
						"error":                t.lastFailureMessage,
					}).Error(ErrTaskDisabledOnFailures)
					t.disable(t.lastFailureMessage)
					return
				}
			case schedule.Ended:
				t.Lock()
				t.state = core.TaskEnded
				t.Unlock()
				event := new(scheduler_event.TaskEndedEvent)
				event.TaskID = t.id
				defer t.eventEmitter.Emit(event)
				return  
			case schedule.Error:
				failureMessage := sr.Error().Error()
				t.disable(failureMessage)
				return  
			}
		case <-t.killChan:
			t.Lock()
			t.state = core.TaskStopped
			t.lastFireTime = time.Time{}
			t.Unlock()
			event := new(scheduler_event.TaskStoppedEvent)
			event.TaskID = t.id
			defer t.eventEmitter.Emit(event)
			return
		}
	}
}
func (t *task) fire() {
	t.Lock()
	defer t.Unlock()
	t.state = core.TaskFiring
	t.lastFireTime = time.Now()
	t.workflow.Start(t)
	t.hitCount++
	t.state = core.TaskSpinning
}
func (t *task) disable(failureMsg string) {
	t.Lock()
	t.state = core.TaskDisabled
	t.Unlock()
	event := new(scheduler_event.TaskDisabledEvent)
	event.TaskID = t.id
	event.Why = fmt.Sprintf("Task disabled with error: %s", failureMsg)
	defer t.eventEmitter.Emit(event)
}
func (t *task) waitForSchedule() {
	select {
	case <-t.killChan:
		return
	case t.schResponseChan <- t.schedule.Wait(t.lastFireTime):
	}
}
func (t *task) RecordFailure(e []error) {
	t.failureMutex.Lock()
	defer t.failureMutex.Unlock()
	t.failedRuns++
	t.lastFailureTime = t.lastFireTime
	t.lastFailureMessage = e[len(e)-1].Error()
}
type taskCollection struct {
	*sync.Mutex
	table map[string]*task
}
func newTaskCollection() *taskCollection {
	return &taskCollection{
		Mutex: &sync.Mutex{},
		table: make(map[string]*task),
	}
}
func (t *taskCollection) Get(id string) *task {
	t.Lock()
	defer t.Unlock()
	if t, ok := t.table[id]; ok {
		return t
	}
	return nil
}