Improve task scheduler sleep mode

metrics/api.go

@@ -128,7 +128,7 @@ func metricsWriter(ctx context.Context) error {
 		select {
 		case <-ctx.Done():
 			return nil
-		case <-ticker.Read():
+		case <-ticker.Wait():
 			err := writeMetricsTo(ctx, pushURL)
 			if err != nil {
 				return err

modules/modules.go

@@ -20,8 +20,7 @@ var (
 	// modulesLocked locks `modules` during starting.
 	modulesLocked = abool.New()
 
-	sleepMode                         = abool.NewBool(false)
-	taskSchedulerSleepModeExitChannel = make(chan struct{})
+	sleepMode = abool.NewBool(false)
 
 	moduleStartTimeout = 2 * time.Minute
 	moduleStopTimeout  = 1 * time.Minute
@@ -112,13 +111,13 @@ func (m *Module) Sleep(enable bool) {
 		return
 	}
 
-	// Notify all waiting tasks that we are not sleeping anymore.
 	m.Lock()
 	defer m.Unlock()
 
 	if enable {
 		m.sleepWaitingChannel = make(chan time.Time)
 	} else {
+		// Notify all waiting tasks that we are not sleeping anymore.
 		close(m.sleepWaitingChannel)
 	}
 }
@@ -129,14 +128,16 @@ func (m *Module) IsSleeping() bool {
 }
 
 // WaitIfSleeping returns channel that will signal when it exits sleep mode.
+// The channel will always return a zero-value time.Time.
+// It uses time.Time to be easier dropped in to replace a time.Ticker.
 func (m *Module) WaitIfSleeping() <-chan time.Time {
 	m.RLock()
 	defer m.RUnlock()
 	return m.sleepWaitingChannel
 }
 
 // NewSleepyTicker returns new sleepyTicker that will respect the modules sleep mode.
-func (m *Module) NewSleepyTicker(normalDuration time.Duration, sleepDuration time.Duration) *SleepyTicker {
+func (m *Module) NewSleepyTicker(normalDuration, sleepDuration time.Duration) *SleepyTicker {
 	return newSleepyTicker(m, normalDuration, sleepDuration)
 }
 
@@ -383,7 +384,7 @@ func initNewModule(name string, prep, start, stop func() error, dependencies ...
 		Name:                name,
 		enabled:             abool.NewBool(false),
 		enabledAsDependency: abool.NewBool(false),
-		sleepMode:           abool.NewBool(false),
+		sleepMode:           abool.NewBool(true),
 		sleepWaitingChannel: make(chan time.Time),
 		prepFn:              prep,
 		startFn:             start,
@@ -400,6 +401,7 @@ func initNewModule(name string, prep, start, stop func() error, dependencies ...
 		depNames:            dependencies,
 	}
 
+	// Sleep mode is disabled by default
 	newModule.Sleep(false)
 
 	return newModule
@@ -425,21 +427,20 @@ func initDependencies() error {
 	return nil
 }
 
-// EnterSleepMode enables or disables sleep mode for all the modules.
-func EnterSleepMode(enabled bool) {
-	// Check if differs with the old state.
-	set := sleepMode.SetToIf(!enabled, enabled)
-	if !set {
-		return
-	}
-
+// SetSleepMode enables or disables sleep mode for all the modules.
+func SetSleepMode(enabled bool) {
 	// Update all modules
 	for _, m := range modules {
 		m.Sleep(enabled)
 	}
 
-	// Send signal to the task schedular.
-	if !enabled {
-		taskSchedulerSleepModeExitChannel <- struct{}{}
+	// Check if differs with the old state.
+	set := sleepMode.SetToIf(!enabled, enabled)
+	if set {
+		// Send signal to the task schedular.
+		select {
+		case notifyTaskScheduler <- struct{}{}:
+		default:
+		}
 	}
 }

modules/sleepyticker.go

@@ -25,8 +25,8 @@ func newSleepyTicker(module *Module, normalDuration time.Duration, sleepDuration
 	return st
 }
 
-// Read waits until the module is not in sleep mode and returns time.Ticker.C channel.
-func (st *SleepyTicker) Read() <-chan time.Time {
+// Wait waits until the module is not in sleep mode and returns time.Ticker.C channel.
+func (st *SleepyTicker) Wait() <-chan time.Time {
 	sleepModeEnabled := st.module.sleepMode.IsSet()
 
 	// Update Sleep mode
@@ -35,10 +35,8 @@ func (st *SleepyTicker) Read() <-chan time.Time {
 	}
 
 	// Wait if until sleep mode exits only if sleepDuration is set to 0.
-	if sleepModeEnabled {
-		if st.sleepDuration == 0 {
-			return st.module.WaitIfSleeping()
-		}
+	if sleepModeEnabled && st.sleepDuration == 0 {
+		return st.module.WaitIfSleeping()
 	}
 
 	return st.ticker.C
@@ -49,12 +47,6 @@ func (st *SleepyTicker) Stop() {
 	st.ticker.Stop()
 }
 
-// Reset stops a ticker and resets its period to the specified duration. The next tick will arrive after the new period elapses. The duration d must be greater than zero; if not, Reset will panic.
-func (st *SleepyTicker) Reset(d time.Duration) {
-	// Reset standard ticker
-	st.ticker.Reset(d)
-}
-
 func (st *SleepyTicker) enterSleepMode(enabled bool) {
 	st.sleepMode = enabled
 	if enabled {

modules/tasks.go

@@ -51,9 +51,9 @@ var (
 
 	waitForever chan time.Time
 
-	queueIsFilled                = make(chan struct{}, 1) // kick off queue handler
-	recalculateNextScheduledTask = make(chan struct{}, 1)
-	taskTimeslot                 = make(chan struct{})
+	queueIsFilled       = make(chan struct{}, 1) // kick off queue handler
+	notifyTaskScheduler = make(chan struct{}, 1)
+	taskTimeslot        = make(chan struct{})
 )
 
 const (
@@ -410,7 +410,7 @@ func (t *Task) addToSchedule(overtime bool) {
 	// notify scheduler
 	defer func() {
 		select {
-		case recalculateNextScheduledTask <- struct{}{}:
+		case notifyTaskScheduler <- struct{}{}:
 		default:
 		}
 	}()
@@ -443,10 +443,6 @@ func (t *Task) addToSchedule(overtime bool) {
 }
 
 func waitUntilNextScheduledTask() <-chan time.Time {
-	if sleepMode.IsSet() {
-		<-taskSchedulerSleepModeExitChannel
-	}
-
 	scheduleLock.Lock()
 	defer scheduleLock.Unlock()
 
@@ -519,10 +515,21 @@ func taskScheduleHandler() {
 	}
 
 	for {
+
+		if sleepMode.IsSet() {
+			select {
+			case <-shutdownSignal:
+				return
+			case <-notifyTaskScheduler:
+				continue
+			}
+		}
+
 		select {
 		case <-shutdownSignal:
 			return
-		case <-recalculateNextScheduledTask:
+		case <-notifyTaskScheduler:
+			continue
 		case <-waitUntilNextScheduledTask():
 			scheduleLock.Lock()
 

notifications/cleaner.go

@@ -13,7 +13,7 @@ func cleaner(ctx context.Context) error { //nolint:unparam // Conforms to worker
 		select {
 		case <-ctx.Done():
 			return nil
-		case <-ticker.Read():
+		case <-ticker.Wait():
 			deleteExpiredNotifs()
 		}
 	}