fix: high CPU utilization by scheduler while idle

resolves #2155

Signed-off-by: Andrei Aaron <[email protected]>

pkg/scheduler/scheduler.go

@@ -60,7 +60,7 @@ func (pq *generatorsPriorityQueue) Pop() any {
 
 const (
 	rateLimiterScheduler = 400
-	rateLimit            = 5 * time.Second
+	rateLimit            = 50 * time.Millisecond
 	NumWorkersMultiplier = 4
 	sendMetricsInterval  = 5 * time.Second
 )
@@ -241,7 +241,7 @@ func (scheduler *Scheduler) RunScheduler() {
 	ctx, cancel := context.WithCancel(context.Background())
 	scheduler.cancelFunc = cancel
 
-	throttle := time.NewTicker(rateLimit).C
+	throttle := time.NewTicker(scheduler.RateLimit).C
 
 	numWorkers := scheduler.NumWorkers
 
@@ -278,10 +278,14 @@ func (scheduler *Scheduler) RunScheduler() {
 
 				task := scheduler.getTask()
 
-				if task != nil {
-					// push tasks into worker pool until workerChan is full.
-					scheduler.workerChan <- task
+				if task == nil {
+					<-throttle
+
+					continue
 				}
+
+				// push tasks into worker pool until workerChan is full.
+				scheduler.workerChan <- task
 			}
 		}
 	}()

pkg/scheduler/scheduler_test.go

@@ -130,10 +130,10 @@ func TestScheduler(t *testing.T) {
 
 		genH := &shortGenerator{log: logger, priority: "high priority"}
 		// interval has to be higher than throttle value to simulate
-		sch.SubmitGenerator(genH, 6*time.Second, scheduler.HighPriority)
+		sch.SubmitGenerator(genH, 1*time.Second, scheduler.HighPriority)
 
 		sch.RunScheduler()
-		time.Sleep(7 * time.Second)
+		time.Sleep(2 * time.Second)
 		sch.Shutdown()
 
 		data, err := os.ReadFile(logFile.Name())
@@ -152,6 +152,7 @@ func TestScheduler(t *testing.T) {
 		cfg.Scheduler = &config.SchedulerConfig{NumWorkers: 3}
 		metrics := monitoring.NewMetricsServer(true, logger)
 		sch := scheduler.NewScheduler(cfg, metrics, logger)
+		sch.RateLimit = 5 * time.Second
 
 		genL := &generator{log: logger, priority: "low priority"}
 		sch.SubmitGenerator(genL, time.Duration(0), scheduler.LowPriority)
@@ -212,7 +213,7 @@ func TestScheduler(t *testing.T) {
 		sch.SubmitGenerator(genL, 20*time.Millisecond, scheduler.LowPriority)
 
 		sch.RunScheduler()
-		time.Sleep(4 * time.Second)
+		time.Sleep(1 * time.Second)
 		sch.Shutdown()
 
 		data, err := os.ReadFile(logFile.Name())
@@ -275,7 +276,7 @@ func TestScheduler(t *testing.T) {
 		sch.SubmitGenerator(genL, 20*time.Millisecond, scheduler.MediumPriority)
 
 		sch.RunScheduler()
-		time.Sleep(4 * time.Second)
+		time.Sleep(1 * time.Second)
 		sch.Shutdown()
 
 		data, err := os.ReadFile(logFile.Name())