fix: maybe discard crob job if execute job delay

zhenzou · Nov 1, 2023 · f80911a · f80911a
1 parent 871a6e9
commit f80911a
Show file tree

Hide file tree

Showing 3 changed files with 39 additions and 20 deletions.
diff --git a/cron/cron.go b/cron/cron.go
@@ -54,14 +54,14 @@ func (t *task[T]) ready() bool {
 
 func (t *task[T]) scheduleNextRun() {
 	t.LastRunTime = t.NextRunTime
-	t.NextRunTime = t.Expr.Next(t.LastRunTime.In(t.Location)).Truncate(time.Microsecond)
+	t.NextRunTime = t.Expr.Next(t.LastRunTime.In(t.Location))
 }
 
 func (t *task[T]) untilNextRun() time.Duration {
 	if t.ready() {
 		return 0
 	}
-	return t.NextRunTime.Sub(t.now()).Truncate(time.Microsecond)
+	return t.NextRunTime.Sub(t.now())
 }
 
 type Dispatcher[T any] interface {

diff --git a/cron/dispatcher.go b/cron/dispatcher.go
@@ -126,22 +126,26 @@ func (d *dispatcher[T]) takeReadyTask() (time.Duration, bool) {
 	d.locker.Lock()
 	defer d.locker.Unlock()
 
-	t, ok := d.heap.Peek()
+	for {
+		t, ok := d.heap.Peek()
 
-	if !ok {
-		return maxYieldDuration, false
-	}
-	if !t.ready() {
-		return getYieldDuration(t), false
-	}
+		d.logger.Debug("peek task: ",
+			slog.Int("id", int(t.ID)),
+			slog.String("next_run_time", t.NextRunTime.String()),
+			slog.Bool("ready", t.ready()),
+		)
 
-	_, _ = d.heap.Pop()
-	t.scheduleNextRun()
-	d.heap.Push(t)
+		if !ok {
+			return maxYieldDuration, false
+		}
+		if !t.ready() {
+			return getYieldDuration(t), false
+		}
+
+		_, _ = d.heap.Pop()
+		t.scheduleNextRun()
+		d.heap.Push(t)
 
-	select {
-	case d.readyChan <- t.Task:
-	default:
+		d.readyChan <- t.Task
 	}
-	return 0, true
 }
diff --git a/example/schedule/main.go → example/scheduler/main.go b/example/schedule/main.go → example/scheduler/main.go
@@ -22,24 +22,39 @@ func main() {
 	callable1 := executors.RunnableFunc(func(ctx context.Context) {
 		println("1:", time.Now().Format(time.RFC3339))
 	})
+
 	callable2 := executors.RunnableFunc(func(ctx context.Context) {
 		println("2:", time.Now().Format(time.RFC3339))
 	})
 
+	callable3 := executors.RunnableFunc(func(ctx context.Context) {
+		println("3:", time.Now().Format(time.RFC3339))
+	})
+
 	_, err := executor.ScheduleAtCronRate(callable1, executors.CRONRule{
-		Expr: "0 4 * * *",
+		// Expr: "0/1 * * * *",
 		// Expr: "*/1 * * * *",
-		// Expr:     "*/1 * * * * * *",
+		Expr:     "*/5 * * * * * *",
 		Timezone: "Local",
 	})
 	if err != nil {
 		panic(err)
 	}
 
 	_, err = executor.ScheduleAtCronRate(callable2, executors.CRONRule{
-		Expr: "0 4 * * *",
+		// Expr: "0 4 * * *",
+		// Expr: "*/1 * * * *",
+		Expr:     "*/5 * * * * * *",
+		Timezone: "Local",
+	})
+	if err != nil {
+		panic(err)
+	}
+
+	_, err = executor.ScheduleAtCronRate(callable3, executors.CRONRule{
+		// Expr: "0 4 * * *",
 		// Expr: "*/1 * * * *",
-		// Expr:     "*/1 * * * * * *",
+		Expr:     "*/5 * * * * * *",
 		Timezone: "Local",
 	})
 	if err != nil {