Repository: go-co-op/gocron
Pull request #99 (2020-12-27 21:32:02)

refact: schedules jobs with time.AfterFunc()
### What does this do?
Main change: scheduler now uses `time.AfterFunc` from the standard ```time``` package to schedule jobs instead of manually sorting and looping through all jobs (yes, finally!)

This brings us:

- assertion that our "run jobs" algorithm works without bugs as we're leaving this logic to a standard package
- greater flexibility in what fields we want to keep in jobs - no need to actually keep other fields like `nextRun` or even our job slice for our run logic (they may still be useful for logging or getting meta info about a job or the scheduler, so they're not removed here)
- minor performance gains as we don't sort and loop through our job array every second anymore
- increased cohesion of our scheduler package 

Unfortunately, after doing this refactor and separating the logic correctly, some tests broke, which made this pr a bit larger than what it needed to be. Fortunately, these failures actually brought minor errors to attention; I fixed them in here, so here's a list of what's different:

- calling `.Close()` without listening to the channel returned by `.StartAsync()` would freeze the application;  fixed by making `stopChan` a buffered channel: now the application doesn't freeze up anymore 
- before, the scheduler invoked each job by calling `go job.Run()`. This method not only calls the job function but also updates the run counter, which fails to remove jobs that start immediately (this passed our tests because all tests used `.At()`). This has been fixed, and now we first update our run counter and only later call the job's function; thus it's launch is actually the last thing to be done in the execution chain 
- refactored `TestExecutionSecond` to `TestImmediateExecution`, as this is what was actually being tested here
- refactored `TestAtFuture` to use `t.Run()` and the new `s.start()` method
- fixed `TestRunJobsWithLimit` to tests cases with immediate execution

### List any changes that modify/break current functionality
None

### Have you included tests for your changes?
Yes

### Notes

I did my tests manually here, but if possible guys, take a look at some running examples to also check if they are working correctly. 

A simple example of manual testing function that can be called from any testing file: 
```go
func TestManualTesting(t *testing.T) {
	s := NewScheduler(time.UTC)
	s.Every(1).Second().Do(func() {fmt.Println("job 1: ", time.Now().UTC().Format("15:04:05"))})
	s.Every(5).Second().Do(func() {fmt.Println("job 2: ", time.Now().UTC().Format("15:04:05"))})
	s.StartAsync()
	time.Sleep(11 * time.Second)
}```

Streppel: @JohnRoesler thank you so much for the feedback and cleaning up the data race issues!

Patch:
diff --git a/job.go b/job.go
index 23c1365..1c128fe 100644
--- a/job.go
+++ b/job.go
@@ -26,6 +26,7 @@ type Job struct {
 	tags              []string                 // allow the user to tag Jobs with certain labels
 	runConfig         runConfig                // configuration for how many times to run the job
 	runCount          int                      // number of times the job ran
+	timer             *time.Timer
 }
 
 type runConfig struct {
@@ -51,8 +52,8 @@ func NewJob(interval uint64) *Job {
 func (j *Job) run() {
 	j.Lock()
 	defer j.Unlock()
-	callJobFuncWithParams(j.funcs[j.jobFunc], j.fparams[j.jobFunc])
 	j.runCount++
+	go callJobFuncWithParams(j.funcs[j.jobFunc], j.fparams[j.jobFunc])
 }
 
 func (j *Job) neverRan() bool {
@@ -73,6 +74,18 @@ func (j *Job) setStartsImmediately(b bool) {
 	j.startsImmediately = b
 }
 
+func (j *Job) getTimer() *time.Timer {
+	j.RLock()
+	defer j.RUnlock()
+	return j.timer
+}
+
+func (j *Job) setTimer(t *time.Timer) {
+	j.Lock()
+	defer j.Unlock()
+	j.timer = t
+}
+
 func (j *Job) getAtTime() time.Duration {
 	j.RLock()
 	defer j.RUnlock()
diff --git a/scheduler.go b/scheduler.go
index 0721586..91d6e2b 100644
--- a/scheduler.go
+++ b/scheduler.go
@@ -31,39 +31,50 @@ func NewScheduler(loc *time.Location) *Scheduler {
 		jobs:     make([]*Job, 0),
 		location: loc,
 		running:  false,
-		stopChan: make(chan struct{}),
+		stopChan: make(chan struct{}, 1),
 		time:     &trueTime{},
 	}
 }
 
-// StartBlocking starts all the pending jobs using a second-long ticker and blocks the current thread
+// StartBlocking starts all jobs and blocks the current thread
 func (s *Scheduler) StartBlocking() {
 	<-s.StartAsync()
 }
 
-// StartAsync starts a goroutine that runs all the pending using a second-long ticker
+// StartAsync starts all jobs without blocking the current thread
 func (s *Scheduler) StartAsync() chan struct{} {
 	if s.IsRunning() {
 		return s.stopChan
 	}
+	s.start()
+	go func() {
+		<-s.stopChan
+		s.setRunning(false)
+		return
+	}()
+	return s.stopChan
+}
+
+//start starts the scheduler, scheduling and running jobs
+func (s *Scheduler) start() {
 	s.setRunning(true)
+	s.runJobs(s.Jobs())
+}
 
-	s.scheduleAllJobs()
-	ticker := s.time.NewTicker(1 * time.Second)
-	go func() {
-		for {
-			select {
-			case <-ticker.C:
-				s.RunPending()
-			case <-s.stopChan:
-				ticker.Stop()
-				s.setRunning(false)
-				return
+func (s *Scheduler) runJobs(jobs []*Job) {
+	for _, j := range jobs {
+		if j.getStartsImmediately() {
+			s.run(j)
+			j.setStartsImmediately(false)
+		}
+		if !j.shouldRun() {
+			if j.getRemoveAfterLastRun() { // TODO: this method seems unnecessary as we could always remove after the run cout has expired. Maybe remove this in the future?
+				s.RemoveByReference(j)
 			}
+			continue
 		}
-	}()
-
-	return s.stopChan
+		s.scheduleNextRun(j)
+	}
 }
 
 func (s *Scheduler) setRunning(b bool) {
@@ -126,41 +137,40 @@ func (s *Scheduler) Location() *time.Location {
 
 // scheduleNextRun Compute the instant when this Job should run next
 func (s *Scheduler) scheduleNextRun(job *Job) {
-	now := s.time.Now(s.Location())
+	now := s.now()
+	lastRun := job.LastRun()
 
+	// job can be scheduled with .StartAt()
 	if job.neverRan() {
 		if !job.NextRun().IsZero() {
 			return // scheduled for future run and should skip scheduling
 		}
-		// default is for jobs to start immediately unless scheduled at a specific time or day
-		if job.getStartsImmediately() {
-			job.setNextRun(now)
-			return
-		}
+		lastRun = now
 	}
 
-	job.setLastRun(now)
-
-	durationToNextRun := s.durationToNextRun(job)
-	job.setNextRun(job.LastRun().Add(durationToNextRun))
+	durationToNextRun := s.durationToNextRun(lastRun, job)
+	job.setNextRun(lastRun.Add(durationToNextRun))
+	job.setTimer(time.AfterFunc(durationToNextRun, func() {
+		s.run(job)
+		s.scheduleNextRun(job)
+	}))
 }
 
-func (s *Scheduler) durationToNextRun(job *Job) time.Duration {
-	lastRun := job.LastRun()
+func (s *Scheduler) durationToNextRun(t time.Time, job *Job) time.Duration {
 	var duration time.Duration
 	switch job.unit {
 	case seconds, minutes, hours:
 		duration = s.calculateDuration(job)
 	case days:
-		duration = s.calculateDays(job, lastRun)
+		duration = s.calculateDays(job, t)
 	case weeks:
 		if job.scheduledWeekday != nil { // weekday selected, Every().Monday(), for example
-			duration = s.calculateWeekday(job, lastRun)
+			duration = s.calculateWeekday(job, t)
 		} else {
-			duration = s.calculateWeeks(job, lastRun)
+			duration = s.calculateWeeks(job, t)
 		}
 	case months:
-		duration = s.calculateMonths(job, lastRun)
+		duration = s.calculateMonths(job, t)
 	}
 	return duration
 }
@@ -280,18 +290,6 @@ func (s *Scheduler) roundToMidnight(t time.Time) time.Time {
 	return time.Date(t.Year(), t.Month(), t.Day(), 0, 0, 0, 0, s.Location())
 }
 
-// Get the current runnable Jobs, which shouldRun is True
-func (s *Scheduler) runnableJobs() []*Job {
-	var runnableJobs []*Job
-	sort.Sort(s)
-	for _, job := range s.Jobs() {
-		if s.shouldRun(job) {
-			runnableJobs = append(runnableJobs, job)
-		}
-	}
-	return runnableJobs
-}
-
 // NextRun datetime when the next Job should run.
 func (s *Scheduler) NextRun() (*Job, time.Time) {
 	if len(s.Jobs()) <= 0 {
@@ -310,25 +308,13 @@ func (s *Scheduler) Every(interval uint64) *Scheduler {
 	return s
 }
 
-// RunPending runs all the Jobs that are scheduled to run.
-func (s *Scheduler) RunPending() {
-	for _, job := range s.runnableJobs() {
-		s.runAndReschedule(job) // we should handle this error somehow
+func (s *Scheduler) run(job *Job) {
+	if !s.IsRunning() {
+		return
 	}
-}
 
-func (s *Scheduler) runAndReschedule(job *Job) error {
-	if err := s.run(job); err != nil {
-		return err
-	}
-	s.scheduleNextRun(job)
-	return nil
-}
-
-func (s *Scheduler) run(job *Job) error {
-	job.setLastRun(s.time.Now(s.Location()))
-	go job.run()
-	return nil
+	job.setLastRun(s.now())
+	job.run()
 }
 
 // RunAll run all Jobs regardless if they are scheduled to run or not
@@ -339,10 +325,7 @@ func (s *Scheduler) RunAll() {
 // RunAllWithDelay runs all Jobs with delay seconds
 func (s *Scheduler) RunAllWithDelay(d int) {
 	for _, job := range s.Jobs() {
-		err := s.run(job)
-		if err != nil {
-			continue
-		}
+		s.run(job)
 		s.time.Sleep(time.Duration(d) * time.Second)
 	}
 }
@@ -418,11 +401,12 @@ func (s *Scheduler) Clear() {
 // Stop stops the scheduler. This is a no-op if the scheduler is already stopped .
 func (s *Scheduler) Stop() {
 	if s.IsRunning() {
-		s.stopScheduler()
+		s.stop()
 	}
 }
 
-func (s *Scheduler) stopScheduler() {
+func (s *Scheduler) stop() {
+	s.setRunning(false)
 	s.stopChan <- struct{}{}
 }
 
@@ -626,8 +610,6 @@ func (s *Scheduler) getCurrentJob() *Job {
 	return s.Jobs()[len(s.jobs)-1]
 }
 
-func (s *Scheduler) scheduleAllJobs() {
-	for _, j := range s.Jobs() {
-		s.scheduleNextRun(j)
-	}
+func (s *Scheduler) now() time.Time {
+	return s.time.Now(s.Location())
 }


Test patch:
diff --git a/scheduler_test.go b/scheduler_test.go
index 6e04ae2..c4766b3 100644
--- a/scheduler_test.go
+++ b/scheduler_test.go
@@ -39,20 +39,20 @@ func taskWithParams(a int, b string) {
 	fmt.Println(a, b)
 }
 
-func TestExecutionSecond(t *testing.T) {
+func TestImmediateExecution(t *testing.T) {
 	sched := NewScheduler(time.UTC)
-	success := false
-	mu := sync.Mutex{}
-	sched.Every(1).Second().Do(func(mutableValue *bool, mu *sync.Mutex) {
-		mu.Lock()
-		defer mu.Unlock()
-		*mutableValue = !*mutableValue
-	}, &success, &mu)
-	sched.RunAllWithDelay(1)
+	semaphore := make(chan bool)
+	sched.Every(1).Second().Do(func() {
+		semaphore <- true
+	})
+	sched.StartAsync()
+	select {
+	case <-time.After(1 * time.Second):
+		t.Fatal("job did not run immediately")
+	case <-semaphore:
+		// test passed
+	}
 
-	mu.Lock()
-	defer mu.Unlock()
-	assert.Equal(t, true, success, "Task did not get called")
 }
 
 func TestExecutionSeconds(t *testing.T) {
@@ -63,12 +63,15 @@ func TestExecutionSeconds(t *testing.T) {
 		executions         []int64
 		interval           uint64 = 2
 		expectedExecutions        = 4
+		mu                 sync.RWMutex
 	)
 
 	runTime := time.Duration(6 * time.Second)
 	startTime := time.Now()
 
 	sched.Every(interval).Seconds().Do(func() {
+		mu.Lock()
+		defer mu.Unlock()
 		executions = append(executions, time.Now().UTC().Unix())
 		if time.Now().After(startTime.Add(runTime)) {
 			jobDone <- true
@@ -79,6 +82,8 @@ func TestExecutionSeconds(t *testing.T) {
 	<-jobDone // Wait job done
 	close(stop)
 
+	mu.RLock()
+	defer mu.RUnlock()
 	assert.Equal(t, expectedExecutions, len(executions), "did not run expected number of times")
 
 	for i := 1; i < expectedExecutions; i++ {
@@ -105,36 +110,39 @@ func TestScheduledWithTag(t *testing.T) {
 }
 
 func TestAtFuture(t *testing.T) {
-	s := NewScheduler(time.UTC)
-	now := time.Now().UTC()
-
-	// Schedule to run in next minute
-	nextMinuteTime := now.Add(1 * time.Minute)
-	startAt := fmt.Sprintf("%02d:%02d:%02d", nextMinuteTime.Hour(), nextMinuteTime.Minute(), nextMinuteTime.Second())
-	shouldBeFalse := false
-	dayJob, _ := s.Every(1).Day().At(startAt).Do(func() {
-		shouldBeFalse = true
-	})
-	s.scheduleAllJobs()
+	t.Run("calls to .At() should parse time correctly", func(t *testing.T) {
+
+		s := NewScheduler(time.UTC)
+		now := time.Now().UTC()
 
-	// Check first run
-	expectedStartTime := time.Date(now.Year(), now.Month(), now.Day(), now.Hour(), now.Add(time.Minute).Minute(), now.Second(), 0, time.UTC)
-	nextRun := dayJob.ScheduledTime()
-	assert.Equal(t, expectedStartTime, nextRun)
+		// Schedule to run in next minute
+		nextMinuteTime := now.Add(1 * time.Minute) // fixme: test fails any hour at :59
+		startAt := fmt.Sprintf("%02d:%02d:%02d", nextMinuteTime.Hour(), nextMinuteTime.Minute(), nextMinuteTime.Second())
+		var hasRan bool
+		dayJob, _ := s.Every(1).Day().At(startAt).Do(func() {
+			hasRan = true
+		})
+		s.start()
 
-	s.RunPending()
+		// Check first run
+		expectedStartTime := time.Date(now.Year(), now.Month(), now.Day(), now.Hour(), now.Add(time.Minute).Minute(), now.Second(), 0, time.UTC)
+		nextRun := dayJob.ScheduledTime()
+		assert.Equal(t, expectedStartTime, nextRun)
 
-	// Check next run's scheduled time
-	nextRun = dayJob.ScheduledTime()
-	assert.Equal(t, expectedStartTime, nextRun)
-	assert.Equal(t, false, shouldBeFalse, "Day job was not expected to run as it was in the future")
-	s.RemoveByReference(dayJob)
+		// Check next run's scheduled time
+		nextRun = dayJob.ScheduledTime()
+		assert.Equal(t, expectedStartTime, nextRun)
+		assert.False(t, hasRan, "Day job was not expected to run as it was in the future")
 
-	// error due to bad time format
-	badTime := "0:0"
-	s.Every(1).Day().At(badTime).Do(func() {})
-	assert.Zero(t, len(s.jobs), "The job should be deleted if the time format is wrong")
+	})
 
+	t.Run("error due to bad time format", func(t *testing.T) {
+		s := NewScheduler(time.UTC)
+		badTime := "0:0"
+		_, err := s.Every(1).Day().At(badTime).Do(func() {})
+		assert.Error(t, err, "bad time format should not include jobs to the scheduler")
+		assert.Zero(t, len(s.jobs))
+	})
 }
 
 func schedulerForNextOrPreviousWeekdayEveryNTimes(weekday time.Weekday, next bool, n uint64, s *Scheduler) *Scheduler {
@@ -375,9 +383,18 @@ func TestScheduler_Stop(t *testing.T) {
 	t.Run("stops a running scheduler", func(t *testing.T) {
 		s := NewScheduler(time.UTC)
 		s.StartAsync()
+		assert.True(t, s.IsRunning())
 		s.Stop()
 		assert.False(t, s.IsRunning())
 	})
+	t.Run("stops a running scheduler through StartAsync chan", func(t *testing.T) {
+		s := NewScheduler(time.UTC)
+		c := s.StartAsync()
+		assert.True(t, s.IsRunning())
+		close(c)
+		time.Sleep(1 * time.Millisecond) // wait for stop goroutine to catch up
+		assert.False(t, s.IsRunning())
+	})
 	t.Run("noop on stopped scheduler", func(t *testing.T) {
 		s := NewScheduler(time.UTC)
 		s.Stop()
@@ -391,16 +408,14 @@ func TestScheduler_StartAt(t *testing.T) {
 
 	// With StartAt
 	job, _ := scheduler.Every(3).Seconds().StartAt(now.Add(time.Second * 5)).Do(func() {})
-	scheduler.scheduleAllJobs()
-	_, nextRun := scheduler.NextRun()
-	assert.Equal(t, now.Add(time.Second*5), nextRun)
-	scheduler.Remove(job)
+	assert.False(t, job.getStartsImmediately())
+	scheduler.start()
+	assert.Equal(t, now.Add(time.Second*5), job.NextRun())
+	scheduler.stop()
 
 	// Without StartAt
 	job, _ = scheduler.Every(3).Seconds().Do(func() {})
-	scheduler.scheduleNextRun(job)
-	_, nextRun = scheduler.NextRun()
-	assert.Equal(t, now.Second(), nextRun.Second())
+	assert.True(t, job.getStartsImmediately())
 }
 
 func TestScheduler_CalculateNextRun(t *testing.T) {
@@ -776,7 +791,7 @@ func TestScheduler_CalculateNextRun(t *testing.T) {
 	for i := range tests {
 		t.Run(tests[i].name, func(t *testing.T) {
 			sched := NewScheduler(time.UTC)
-			got := sched.durationToNextRun(&tests[i].job)
+			got := sched.durationToNextRun(tests[i].job.LastRun(), &tests[i].job)
 			assert.Equalf(t, tests[i].wantTimeUntilNextRun, got, fmt.Sprintf("expected %s / got %s", tests[i].wantTimeUntilNextRun.String(), got.String()))
 		})
 	}
@@ -809,7 +824,7 @@ func TestScheduler_Do(t *testing.T) {
 		s.setRunning(false)
 		job, err := s.Every(1).Second().Do(func() {})
 		assert.Equal(t, nil, err)
-		assert.True(t, job.nextRun.IsZero())
+		assert.True(t, job.NextRun().IsZero())
 	})
 
 	t.Run("adding a new job when scheduler is running schedules job", func(t *testing.T) {
@@ -817,7 +832,7 @@ func TestScheduler_Do(t *testing.T) {
 		s.setRunning(true)
 		job, err := s.Every(1).Second().Do(func() {})
 		assert.Equal(t, nil, err)
-		assert.False(t, job.nextRun.IsZero())
+		assert.False(t, job.NextRun().IsZero())
 	})
 }
 
@@ -829,20 +844,20 @@ func TestRunJobsWithLimit(t *testing.T) {
 	}
 
 	s := NewScheduler(time.UTC)
-	s.StartAsync()
 
 	var j1Counter, j2Counter int
 	var j1Mutex, j2Mutex sync.RWMutex
-	j1, err := s.Every(1).StartAt(time.Now().UTC().Add(1*time.Second)).Do(f, &j1Counter, &j1Mutex)
+	j1, err := s.Every(1).Second().Do(f, &j1Counter, &j1Mutex)
 	require.NoError(t, err)
 
 	j1.LimitRunsTo(1)
 
-	j2, err := s.Every(1).StartAt(time.Now().UTC().Add(2*time.Second)).Do(f, &j2Counter, &j2Mutex)
+	j2, err := s.Every(1).Second().Do(f, &j2Counter, &j2Mutex)
 	require.NoError(t, err)
 
 	j2.LimitRunsTo(1)
 
+	s.StartAsync()
 	time.Sleep(3 * time.Second)
 
 	j1Mutex.RLock()
@@ -884,13 +899,13 @@ func TestDo(t *testing.T) {
 
 func TestRemoveAfterExec(t *testing.T) {
 	s := NewScheduler(time.UTC)
-	s.StartAsync()
 
-	job, err := s.Every(1).StartAt(time.Now().Add(1*time.Second)).Do(task, s)
+	job, err := s.Every(1).Second().Do(task, s)
 	require.NoError(t, err)
 
 	job.LimitRunsTo(1)
 	job.RemoveAfterLastRun()
+	s.StartAsync()
 
 	time.Sleep(2 * time.Second)