fix(scheduler): synchronize the started flag read (#149)

reugn · Feb 1, 2025 · 9dff931 · 9dff931
1 parent 4bb9433
commit 9dff931
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 8 deletions.
diff --git a/quartz/scheduler.go b/quartz/scheduler.go
@@ -68,7 +68,7 @@ type Scheduler interface {
 
 // StdScheduler implements the [Scheduler] interface.
 type StdScheduler struct {
-	mtx sync.Mutex
+	mtx sync.RWMutex
 	wg  sync.WaitGroup
 
 	interrupt chan struct{}
@@ -300,7 +300,7 @@ func (sched *StdScheduler) ScheduleJob(
 
 	if err = sched.queue.Push(toSchedule); err == nil {
 		sched.logger.Debug("Successfully added job", "key", jobDetail.jobKey.String())
-		if sched.started {
+		if sched.IsStarted() {
 			sched.Reset()
 		}
 	}
@@ -342,8 +342,8 @@ func (sched *StdScheduler) Wait(ctx context.Context) {
 
 // IsStarted determines whether the scheduler has been started.
 func (sched *StdScheduler) IsStarted() bool {
-	sched.mtx.Lock()
-	defer sched.mtx.Unlock()
+	sched.mtx.RLock()
+	defer sched.mtx.RUnlock()
 
 	return sched.started
 }
@@ -391,7 +391,7 @@ func (sched *StdScheduler) DeleteJob(jobKey *JobKey) error {
 	_, err := sched.queue.Remove(jobKey)
 	if err == nil {
 		sched.logger.Debug("Successfully deleted job", "key", jobKey.String())
-		if sched.started {
+		if sched.IsStarted() {
 			sched.Reset()
 		}
 	}
@@ -426,7 +426,7 @@ func (sched *StdScheduler) PauseJob(jobKey *JobKey) error {
 		}
 		if err = sched.queue.Push(paused); err == nil {
 			sched.logger.Debug("Successfully paused job", "key", jobKey.String())
-			if sched.started {
+			if sched.IsStarted() {
 				sched.Reset()
 			}
 		}
@@ -466,7 +466,7 @@ func (sched *StdScheduler) ResumeJob(jobKey *JobKey) error {
 		}
 		if err = sched.queue.Push(resumed); err == nil {
 			sched.logger.Debug("Successfully resumed job", "key", jobKey.String())
-			if sched.started {
+			if sched.IsStarted() {
 				sched.Reset()
 			}
 		}
@@ -483,7 +483,7 @@ func (sched *StdScheduler) Clear() error {
 	err := sched.queue.Clear()
 	if err == nil {
 		sched.logger.Debug("Successfully cleared job queue")
-		if sched.started {
+		if sched.IsStarted() {
 			sched.Reset()
 		}
 	}

diff --git a/quartz/scheduler_test.go b/quartz/scheduler_test.go
@@ -105,8 +105,12 @@ func TestScheduler(t *testing.T) {
 }
 
 func TestScheduler_BlockingSemantics(t *testing.T) {
+	t.Parallel()
+
 	for _, tt := range []string{"Blocking", "NonBlocking", "WorkerSmall", "WorkerLarge"} {
+		tt := tt
 		t.Run(tt, func(t *testing.T) {
+			t.Parallel()
 			var (
 				workerLimit int
 				opt         quartz.SchedulerOpt