func ParseComputationID(s StreamID) (string, string, error) {
	c := strings.TrimSpace(string(s))
	firstParen := strings.Index(c, "(")
	if firstParen < 1 || c[len(c)-1] != ')' {
		return "", "", fmt.Errorf("Computation %s invalid!", c)
	}
	return c[:firstParen], c[firstParen+1 : len(c)-1], nil
}
ParseComputationID parses a computation definition
package dagger
import (
	"fmt"
	log "github.com/Sirupsen/logrus"
	"math/rand"
	"strings"
	"github.com/rcrowley/go-metrics"
)
type Task interface {
	RecordProcessor
	GetSnapshot() ([]byte, error)
	Sync() (Timestamp, error)
	Run(chan error)
	Stop()
}
type TaskManager interface {
	ManageTasks() error
	GetTaskSnapshot(StreamID) ([]byte, error)
	Stop()
}
type taskManager struct {
	tasks        map[StreamID]Task
	coordinator  Coordinator
	receiver     InputManager
	taskStarter  TaskStarter
	done         chan struct{}
	taskFailedCh chan *TaskInfo
	counter metrics.Counter
}
func NewTaskManager(coordinator Coordinator, receiver InputManager, taskStarter TaskStarter) TaskManager {
	c := metrics.NewCounter()
	metrics.Register("processing", c)
	tm := &taskManager{
		tasks:        make(map[StreamID]Task),
		coordinator:  coordinator,
		receiver:     receiver,
		taskStarter:  taskStarter,
		counter:      c,
		done:         make(chan struct{}),
		taskFailedCh: make(chan *TaskInfo),
	}
	receiver.SetTaskManager(tm)
	return tm
}