check for closed channel and adjust tests

Using time.Ticker allow us to avoid recursive call that may end up in schedule running and possible deadlock if no routine is listening for cancel calls

switch to closing channel

management/server/scheduler.go

@@ -1,9 +1,10 @@
 package server
 
 import (
-	log "github.com/sirupsen/logrus"
 	"sync"
 	"time"
+
+	log "github.com/sirupsen/logrus"
 )
 
 // Scheduler is an interface which implementations can schedule and cancel jobs
@@ -55,14 +56,8 @@ func (wm *DefaultScheduler) cancel(ID string) bool {
 	cancel, ok := wm.jobs[ID]
 	if ok {
 		delete(wm.jobs, ID)
-		select {
+		close(cancel)
-		case cancel <- struct{}{}:
+		log.Debugf("cancelled scheduled job %s", ID)
-			log.Debugf("cancelled scheduled job %s", ID)
-		default:
-			log.Warnf("couldn't cancel job %s because there was no routine listening on the cancel event", ID)
-			return false
-		}
-
 	}
 	return ok
 }
@@ -90,25 +85,41 @@ func (wm *DefaultScheduler) Schedule(in time.Duration, ID string, job func() (ne
 		return
 	}
 
+	ticker := time.NewTicker(in)
+
 	wm.jobs[ID] = cancel
 	log.Debugf("scheduled a job %s to run in %s. There are %d total jobs scheduled.", ID, in.String(), len(wm.jobs))
 	go func() {
-		select {
+		for {
-		case <-time.After(in):
+			select {
-			log.Debugf("time to do a scheduled job %s", ID)
+			case <-ticker.C:
-			runIn, reschedule := job()
+				select {
-			wm.mu.Lock()
+				case <-cancel:
-			defer wm.mu.Unlock()
+					log.Debugf("scheduled job %s was canceled, stop timer", ID)
-			delete(wm.jobs, ID)
+					ticker.Stop()
-			if reschedule {
+					return
-				go wm.Schedule(runIn, ID, job)
+				default:
+					log.Debugf("time to do a scheduled job %s", ID)
+				}
+				runIn, reschedule := job()
+				if !reschedule {
+					wm.mu.Lock()
+					defer wm.mu.Unlock()
+					delete(wm.jobs, ID)
+					log.Debugf("job %s is not scheduled to run again", ID)
+					ticker.Stop()
+					return
+				}
+				// we need this comparison to avoid resetting the ticker with the same duration and missing the current elapsesed time
+				if runIn != in {
+					ticker.Reset(runIn)
+				}
+			case <-cancel:
+				log.Debugf("job %s was canceled, stopping timer", ID)
+				ticker.Stop()
+				return
 			}
-		case <-cancel:
-			log.Debugf("stopped scheduled job %s ", ID)
-			wm.mu.Lock()
-			defer wm.mu.Unlock()
-			delete(wm.jobs, ID)
-			return
 		}
+
 	}()
 }

management/server/scheduler_test.go

@@ -2,11 +2,12 @@ package server
 
 import (
 	"fmt"
-	"github.com/stretchr/testify/assert"
 	"math/rand"
 	"sync"
 	"testing"
 	"time"
+
+	"github.com/stretchr/testify/assert"
 )
 
 func TestScheduler_Performance(t *testing.T) {
@@ -36,15 +37,24 @@ func TestScheduler_Cancel(t *testing.T) {
 	jobID1 := "test-scheduler-job-1"
 	jobID2 := "test-scheduler-job-2"
 	scheduler := NewDefaultScheduler()
-	scheduler.Schedule(2*time.Second, jobID1, func() (nextRunIn time.Duration, reschedule bool) {
+	tChan := make(chan struct{})
-		return 0, false
+	p := []string{jobID1, jobID2}
+	scheduler.Schedule(2*time.Millisecond, jobID1, func() (nextRunIn time.Duration, reschedule bool) {
+		tt := p[0]
+		<-tChan
+		t.Logf("job %s", tt)
+		return 2 * time.Millisecond, true
 	})
-	scheduler.Schedule(2*time.Second, jobID2, func() (nextRunIn time.Duration, reschedule bool) {
+	scheduler.Schedule(2*time.Millisecond, jobID2, func() (nextRunIn time.Duration, reschedule bool) {
-		return 0, false
+		return 2 * time.Millisecond, true
 	})
 
+	time.Sleep(4 * time.Millisecond)
 	assert.Len(t, scheduler.jobs, 2)
 	scheduler.Cancel([]string{jobID1})
+	close(tChan)
+	p = []string{}
+	time.Sleep(4 * time.Millisecond)
 	assert.Len(t, scheduler.jobs, 1)
 	assert.NotNil(t, scheduler.jobs[jobID2])
 }