fixes #361

When the `task.spin()` goroutine returns, it leaves the one running
`task.waitForSchedule()` blocked.  When the task is restarted, its
response chan is available again, causing the hung `waitForSchedule`
goroutine to begin execution again.

At this time, the _new_ call to `spin()` spawns a _new_ goroutine for
`waitForSchedule`, causing the spin loop to be executed twice.

This pattern continued for each restart of a task.

mgmt/rest/rest_func_test.go

@@ -938,6 +938,8 @@ func TestPluginRestCalls(t *testing.T) {
 				plr4 := r4.Body.(*rbody.ScheduledTaskStopped)
 				So(plr4.ID, ShouldEqual, id)
 
+				time.Sleep(1 * time.Second)
+
 				r5 := getTasks(port)
 				So(r5.Body, ShouldHaveSameTypeAs, new(rbody.ScheduledTaskListReturned))
 				plr5 := r5.Body.(*rbody.ScheduledTaskListReturned)

scheduler/task.go

@@ -68,7 +68,6 @@ func newTask(s schedule.Schedule, wf *schedulerWorkflow, m *workManager, mm mana
 		id:               taskId,
 		name:             name,
 		schResponseChan:  make(chan schedule.Response),
-		killChan:         make(chan struct{}),
 		schedule:         s,
 		state:            core.TaskStopped,
 		creationTime:     time.Now(),
@@ -174,6 +173,7 @@ func (t *task) Spin() {
 	defer t.Unlock()
 	if t.state == core.TaskStopped {
 		t.state = core.TaskSpinning
+		t.killChan = make(chan struct{})
 		// spin in a goroutine
 		go t.spin()
 	}
@@ -183,15 +183,15 @@ func (t *task) Stop() {
 	t.Lock()
 	defer t.Unlock()
 	if t.state == core.TaskFiring || t.state == core.TaskSpinning {
-		t.killChan <- struct{}{}
+		close(t.killChan)
 	}
 }
 
 func (t *task) Kill() {
 	t.Lock()
 	defer t.Unlock()
 	if t.state == core.TaskFiring || t.state == core.TaskSpinning {
-		t.killChan <- struct{}{}
+		close(t.killChan)
 		t.state = core.TaskDisabled
 	}
 }
@@ -274,6 +274,7 @@ func (t *task) spin() {
 		case <-t.killChan:
 			// Only here can it truly be stopped
 			t.state = core.TaskStopped
+			t.lastFireTime = time.Time{}
 			return
 		}
 	}
@@ -289,7 +290,11 @@ func (t *task) fire() {
 }
 
 func (t *task) waitForSchedule() {
-	t.schResponseChan <- t.schedule.Wait(t.lastFireTime)
+	select {
+	case <-t.killChan:
+		return
+	case t.schResponseChan <- t.schedule.Wait(t.lastFireTime):
+	}
 }
 
 type taskCollection struct {

scheduler/task_test.go

@@ -89,18 +89,6 @@ func TestTask(t *testing.T) {
 				task.Stop()
 				time.Sleep(time.Millisecond * 10) // it is a race so we slow down the test
 				So(task.state, ShouldEqual, core.TaskStopped)
-				Convey("Stopping a stopped tasks should not send to kill channel", func() {
-					task.Stop()
-					b := false
-					select {
-					case <-task.killChan:
-						b = true
-					default:
-						b = false
-					}
-					So(task.state, ShouldEqual, core.TaskStopped)
-					So(b, ShouldBeFalse)
-				})
 			})
 		})