scheduler: avoid dispatching jobs too early

quartz/function_job_test.go

@@ -26,9 +26,10 @@ func TestFunctionJob(t *testing.T) {
 
 	sched := quartz.NewStdScheduler()
 	sched.Start(ctx)
-	sched.ScheduleJob(ctx, funcJob1, quartz.NewRunOnceTrigger(time.Millisecond*300))
-	sched.ScheduleJob(ctx, funcJob2, quartz.NewRunOnceTrigger(time.Millisecond*800))
+	sched.ScheduleJob(ctx, funcJob1, quartz.NewRunOnceTrigger(100*time.Millisecond))
+	sched.ScheduleJob(ctx, funcJob2, quartz.NewRunOnceTrigger(200*time.Millisecond))
 	time.Sleep(time.Second)
+
 	sched.Clear()
 	sched.Stop()
 

quartz/scheduler.go

@@ -58,7 +58,7 @@ type StdScheduler struct {
 	mtx       sync.Mutex
 	wg        *sync.WaitGroup
 	queue     *priorityQueue
-	interrupt chan struct{}
+	interrupt chan time.Time
 	cancel    context.CancelFunc
 	feeder    chan *item
 	dispatch  chan *item
@@ -95,7 +95,7 @@ func NewStdSchedulerWithOptions(opts StdSchedulerOptions) *StdScheduler {
 	return &StdScheduler{
 		queue:     &priorityQueue{},
 		wg:        &sync.WaitGroup{},
-		interrupt: make(chan struct{}, 1),
+		interrupt: make(chan time.Time, 1),
 		feeder:    make(chan *item),
 		dispatch:  make(chan *item),
 		opts:      opts,
@@ -233,30 +233,53 @@ func (sched *StdScheduler) Stop() {
 
 func (sched *StdScheduler) startExecutionLoop(ctx context.Context) {
 	defer sched.wg.Done()
+
+	t := time.NewTimer(0)
+	defer t.Stop()
+
 	for {
 		if sched.queueLen() == 0 {
 			select {
-			case <-sched.interrupt:
+			case nextJobAt := <-sched.interrupt:
+				safeSetTimer(t, nextJobAt)
 			case <-ctx.Done():
 				log.Printf("Exit the empty execution loop.")
 				return
 			}
-		} else {
-			t := time.NewTimer(sched.calculateNextTick())
-			select {
-			case <-t.C:
-				sched.executeAndReschedule(ctx)
-
-			case <-sched.interrupt:
-				t.Stop()
+			continue
+		}
+		select {
+		case <-t.C:
+			sched.executeAndReschedule(ctx)
+			safeSetTimer(t, sched.calculateNextTick())
+		case nextJobAt := <-sched.interrupt:
+			safeSetTimer(t, nextJobAt)
+		case <-ctx.Done():
+			log.Printf("Exit the execution loop.")
+			return
+		}
+	}
+}
 
-			case <-ctx.Done():
-				log.Printf("Exit the execution loop.")
-				t.Stop()
-				return
-			}
+func safeSetTimer(timer *time.Timer, next time.Time) {
+	// reset/stop the timer
+	if !timer.Stop() {
+		// drain if needed
+		select {
+		case <-timer.C:
+		default:
 		}
+
 	}
+
+	// if the "next" time is in the future, we reset the timer to
+	// this point.
+	if wait := time.Until(next); wait >= 0 {
+		timer.Reset(wait)
+		return
+	}
+
+	timer.Reset(0)
 }
 
 func (sched *StdScheduler) startWorkers(ctx context.Context) {
@@ -285,32 +308,42 @@ func (sched *StdScheduler) queueLen() int {
 	return sched.queue.Len()
 }
 
-func (sched *StdScheduler) calculateNextTick() time.Duration {
-	var interval int64
-
+func (sched *StdScheduler) calculateNextTick() time.Time {
 	sched.mtx.Lock()
 	defer sched.mtx.Unlock()
+
 	if sched.queue.Len() > 0 {
-		interval = parkTime(sched.queue.Head().priority)
+		return time.Unix(0, sched.queue.Head().priority)
 	}
 
-	return time.Duration(interval)
+	return time.Now()
 }
 
 func (sched *StdScheduler) executeAndReschedule(ctx context.Context) {
-	// return if the job queue is empty
-	if sched.queueLen() == 0 {
-		return
-	}
-
 	// fetch an item
 	var it *item
 	func() {
 		sched.mtx.Lock()
 		defer sched.mtx.Unlock()
+		if sched.queue.Len() == 0 {
+			// return if the job queue is empty
+			return
+		}
+
+		if next := time.Unix(0, sched.queue.Head().priority); time.Until(next) > 0 {
+			// return early
+			sched.reset(ctx, next)
+			return
+		}
 		it = heap.Pop(sched.queue).(*item)
 	}()
 
+	// if there isn't actually a job ready to run now, we'll
+	// return early and try again.
+	if it == nil {
+		return
+	}
+
 	// execute the Job
 	if !isOutdated(it.priority) {
 		switch {
@@ -335,6 +368,7 @@ func (sched *StdScheduler) executeAndReschedule(ctx context.Context) {
 	nextRunTime, err := it.Trigger.NextFireTime(it.priority)
 	if err != nil {
 		log.Printf("The Job '%s' got out the execution loop: %q", it.Job.Description(), err.Error())
+		sched.reset(ctx, time.Now().Add(-time.Millisecond))
 		return
 	}
 	it.priority = nextRunTime
@@ -354,7 +388,7 @@ func (sched *StdScheduler) startFeedReader(ctx context.Context) {
 				defer sched.mtx.Unlock()
 
 				heap.Push(sched.queue, item)
-				sched.reset(ctx)
+				sched.reset(ctx, time.Unix(0, sched.queue.Head().priority))
 			}()
 		case <-ctx.Done():
 			log.Printf("Exit the feed reader.")
@@ -363,18 +397,10 @@ func (sched *StdScheduler) startFeedReader(ctx context.Context) {
 	}
 }
 
-func (sched *StdScheduler) reset(ctx context.Context) {
+func (sched *StdScheduler) reset(ctx context.Context, next time.Time) {
 	select {
-	case sched.interrupt <- struct{}{}:
+	case sched.interrupt <- next:
 	case <-ctx.Done():
 	default:
 	}
 }
-
-func parkTime(ts int64) int64 {
-	now := NowNano()
-	if ts > now {
-		return ts - now
-	}
-	return 0
-}

quartz/scheduler_test.go

@@ -96,23 +96,27 @@ func TestSchedulerBlockingSemantics(t *testing.T) {
 					defer timer.Stop()
 					select {
 					case <-timer.C:
+						t.Error("should never reach this")
 						return false, nil
 					case <-ctx.Done():
 						return true, nil
 					}
 				}),
-				quartz.NewSimpleTrigger(time.Millisecond))
+				quartz.NewSimpleTrigger(time.Millisecond),
+			)
 
 			ticker := time.NewTicker(4 * time.Millisecond)
+			defer ticker.Stop()
 			<-ticker.C
 			if atomic.LoadInt64(&n) == 0 {
 				t.Error("job should have run at least once")
 			}
 
+			const attempts = 100
 			switch tt {
 			case "Blocking":
 			BLOCKING:
-				for iters := 0; iters < 100; iters++ {
+				for iters := 0; iters < attempts; iters++ {
 					iters++
 					select {
 					case <-ctx.Done():
@@ -127,23 +131,29 @@ func TestSchedulerBlockingSemantics(t *testing.T) {
 			case "NonBlocking":
 				var lastN int64
 			NONBLOCKING:
-				for iters := 0; iters < 100; iters++ {
+				for iters := 0; iters < attempts; iters++ {
 					select {
 					case <-ctx.Done():
 						break NONBLOCKING
 					case <-ticker.C:
 						num := atomic.LoadInt64(&n)
-						if num <= lastN {
-							t.Errorf("on iter %d n did not increase %d",
-								iters, num,
-							)
+						if num > lastN {
+							break NONBLOCKING
 						}
+
 						lastN = num
 					}
 				}
+				num := atomic.LoadInt64(&n)
+				if num <= lastN {
+					t.Errorf("on iter %d n did not increase %d",
+						attempts, num,
+					)
+				}
+
 			case "WorkerSmall", "WorkerLarge":
 			WORKERS:
-				for iters := 0; iters < 100; iters++ {
+				for iters := 0; iters < attempts; iters++ {
 					select {
 					case <-ctx.Done():
 						break WORKERS

quartz/util_test.go

@@ -8,12 +8,14 @@ import (
 )
 
 func assertEqual[T any](t *testing.T, a T, b T) {
+	t.Helper()
 	if !reflect.DeepEqual(a, b) {
 		t.Fatalf("%v != %v", a, b)
 	}
 }
 
 func assertNotEqual[T any](t *testing.T, a T, b T) {
+	t.Helper()
 	if reflect.DeepEqual(a, b) {
 		t.Fatalf("%v == %v", a, b)
 	}