Repository: go-co-op/gocron (go)

Issue:
Backend design improvement - Scheduler + Exector
The current design involves a lot of struct accessing across multiple goroutines, resulting in a many mutexes to avoid race conditions.

The proposal here is to move to a new backend design as such:

- Executor: Instead of the scheduler kicking of jobs by calling the `Job.run()` func, the executor will listen on a channel for job functions to execute. This removes the need for multiple goroutines to reference the jobs themselves and we can pass the functions + params across the channel only.
- Scheduler: The scheduler will operate the same as it does, except that it sends on a channel rather than calling `Job.run()` in a goroutine

A start: https://github.com/go-co-op/gocron/compare/master...JohnRoesler:executor?expand=1

Pull request:
implement executor to reduce go routines / mutexes
### What does this do?

Spins up an executor that runs job functions passed to it on a chan. This changes it so that only the function of the job is passed to a separate goroutine and reduces the references to the job object by different goroutines.

### Which issue(s) does this PR fix/relate to?
<!--- Put `Resolves #XXX` here to auto-close the issue that your PR fixes (if such) --->
resolves #101 

### List any changes that modify/break current functionality


### Have you included tests for your changes?
implicitly tested by all tests

### Did you document any new/modified functionality?

- [ ] Updated `example_test.go`
- [x] Updated `README.md`
  - added a link to our examples on the go docs

### Notes


Patch:
diff --git a/README.md b/README.md
index edd7a90..f41bc9a 100644
--- a/README.md
+++ b/README.md
@@ -13,6 +13,10 @@ See also these two great articles:
 
 If you want to chat, you can find us at Slack! [<img src="https://img.shields.io/badge/gophers-gocron-brightgreen?logo=slack">](https://gophers.slack.com/archives/CQ7T0T1FW)
 
+## Examples
+
+Take a look in our [go docs](https://pkg.go.dev/github.com/go-co-op/gocron#pkg-examples)
+
 ## FAQ
 
 * Q: I'm running multiple pods on a distributed environment. How can I make a job not run once per pod causing duplication? 
diff --git a/executor.go b/executor.go
new file mode 100644
index 0000000..c3d3950
--- /dev/null
+++ b/executor.go
@@ -0,0 +1,34 @@
+package gocron
+
+import (
+	"sync"
+)
+
+type executor struct {
+	jobFunctions chan jobFunction
+	stop         chan struct{}
+}
+
+func newExecutor() executor {
+	return executor{
+		jobFunctions: make(chan jobFunction, 1),
+		stop:         make(chan struct{}, 1),
+	}
+}
+
+func (e *executor) start() {
+	wg := sync.WaitGroup{}
+	for {
+		select {
+		case f := <-e.jobFunctions:
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				callJobFuncWithParams(f.functions[f.name], f.params[f.name])
+			}()
+		case <-e.stop:
+			wg.Wait()
+			return
+		}
+	}
+}
diff --git a/gocron.go b/gocron.go
index 5253805..88a1e4b 100644
--- a/gocron.go
+++ b/gocron.go
@@ -17,7 +17,6 @@ import (
 // Error declarations for gocron related errors
 var (
 	ErrTimeFormat            = errors.New("time format error")
-	ErrParamsNotAdapted      = errors.New("the number of params is not adapted")
 	ErrNotAFunction          = errors.New("only functions can be schedule into the job queue")
 	ErrNotScheduledWeekday   = errors.New("job not scheduled weekly on a weekday")
 	ErrJobNotFoundWithTag    = errors.New("no jobs found with given tag")
@@ -46,16 +45,16 @@ const (
 	duration
 )
 
-func callJobFuncWithParams(jobFunc interface{}, params []interface{}) ([]reflect.Value, error) {
+func callJobFuncWithParams(jobFunc interface{}, params []interface{}) {
 	f := reflect.ValueOf(jobFunc)
 	if len(params) != f.Type().NumIn() {
-		return nil, ErrParamsNotAdapted
+		return
 	}
 	in := make([]reflect.Value, len(params))
 	for k, param := range params {
 		in[k] = reflect.ValueOf(param)
 	}
-	return f.Call(in), nil
+	f.Call(in)
 }
 
 func getFunctionName(fn interface{}) string {
diff --git a/job.go b/job.go
index c5a3de2..c6a9b87 100644
--- a/job.go
+++ b/job.go
@@ -9,26 +9,30 @@ import (
 // Job struct stores the information necessary to run a Job
 type Job struct {
 	sync.RWMutex
-	interval          int                      // pause interval * unit between runs
-	duration          time.Duration            // time duration between runs
-	unit              timeUnit                 // time units, ,e.g. 'minutes', 'hours'...
-	startsImmediately bool                     // if the Job should run upon scheduler start
-	jobFunc           string                   // the Job jobFunc to run, func[jobFunc]
-	atTime            time.Duration            // optional time at which this Job runs when interval is day
-	startAtTime       time.Time                // optional time at which the Job starts
-	err               error                    // error related to Job
-	lastRun           time.Time                // datetime of last run
-	nextRun           time.Time                // datetime of next run
-	scheduledWeekday  *time.Weekday            // Specific day of the week to start on
-	dayOfTheMonth     int                      // Specific day of the month to run the job
-	funcs             map[string]interface{}   // Map for the function task store
-	fparams           map[string][]interface{} // Map for function and params of function
-	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
+	jobFunction
+	interval          int           // pause interval * unit between runs
+	duration          time.Duration // time duration between runs
+	unit              timeUnit      // time units, ,e.g. 'minutes', 'hours'...
+	startsImmediately bool          // if the Job should run upon scheduler start
+	atTime            time.Duration // optional time at which this Job runs when interval is day
+	startAtTime       time.Time     // optional time at which the Job starts
+	err               error         // error related to Job
+	lastRun           time.Time     // datetime of last run
+	nextRun           time.Time     // datetime of next run
+	scheduledWeekday  *time.Weekday // Specific day of the week to start on
+	dayOfTheMonth     int           // Specific day of the month to run the job
+	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 jobFunction struct {
+	functions map[string]interface{}   // Map for the function task store
+	params    map[string][]interface{} // Map for function and params of function
+	name      string                   // the Job name to run, func[jobFunc]
+}
+
 type runConfig struct {
 	finiteRuns         bool
 	maxRuns            int
@@ -38,41 +42,27 @@ type runConfig struct {
 // NewJob creates a new Job with the provided interval
 func NewJob(interval int) *Job {
 	return &Job{
-		interval:          interval,
-		lastRun:           time.Time{},
-		nextRun:           time.Time{},
-		funcs:             make(map[string]interface{}),
-		fparams:           make(map[string][]interface{}),
+		interval: interval,
+		lastRun:  time.Time{},
+		nextRun:  time.Time{},
+		jobFunction: jobFunction{
+			functions: make(map[string]interface{}),
+			params:    make(map[string][]interface{}),
+		},
 		tags:              []string{},
 		startsImmediately: true,
 	}
 }
 
-// Run the Job and immediately reschedule it
-func (j *Job) run() {
-	j.Lock()
-	defer j.Unlock()
-	j.runCount++
-	go func() {
-		_, _ = callJobFuncWithParams(j.funcs[j.jobFunc], j.fparams[j.jobFunc])
-	}()
-}
-
 func (j *Job) neverRan() bool {
-	j.RLock()
-	defer j.RUnlock()
 	return j.lastRun.IsZero()
 }
 
 func (j *Job) getStartsImmediately() bool {
-	j.RLock()
-	defer j.RUnlock()
 	return j.startsImmediately
 }
 
 func (j *Job) setStartsImmediately(b bool) {
-	j.Lock()
-	defer j.Unlock()
 	j.startsImmediately = b
 }
 
@@ -83,48 +73,34 @@ func (j *Job) setTimer(t *time.Timer) {
 }
 
 func (j *Job) getAtTime() time.Duration {
-	j.RLock()
-	defer j.RUnlock()
 	return j.atTime
 }
 
 func (j *Job) setAtTime(t time.Duration) {
-	j.Lock()
-	defer j.Unlock()
 	j.atTime = t
 }
 
 func (j *Job) getStartAtTime() time.Time {
-	j.RLock()
-	defer j.RUnlock()
 	return j.startAtTime
 }
 
 func (j *Job) setStartAtTime(t time.Time) {
-	j.Lock()
-	defer j.Unlock()
 	j.startAtTime = t
 }
 
 // Err returns an error if one occurred while creating the Job
 func (j *Job) Err() error {
-	j.RLock()
-	defer j.RUnlock()
 	return j.err
 }
 
 // Tag allows you to add arbitrary labels to a Job that do not
 // impact the functionality of the Job
 func (j *Job) Tag(tags ...string) {
-	j.Lock()
-	defer j.Unlock()
 	j.tags = append(j.tags, tags...)
 }
 
 // Untag removes a tag from a Job
 func (j *Job) Untag(t string) {
-	j.Lock()
-	defer j.Unlock()
 	var newTags []string
 	for _, tag := range j.tags {
 		if t != tag {
@@ -137,30 +113,22 @@ func (j *Job) Untag(t string) {
 
 // Tags returns the tags attached to the Job
 func (j *Job) Tags() []string {
-	j.RLock()
-	defer j.RUnlock()
 	return j.tags
 }
 
 // ScheduledTime returns the time of the Job's next scheduled run
 func (j *Job) ScheduledTime() time.Time {
-	j.RLock()
-	defer j.RUnlock()
 	return j.nextRun
 }
 
 // ScheduledAtTime returns the specific time of day the Job will run at
 func (j *Job) ScheduledAtTime() string {
-	j.RLock()
-	defer j.RUnlock()
 	return fmt.Sprintf("%d:%d", j.atTime/time.Hour, (j.atTime%time.Hour)/time.Minute)
 }
 
 // Weekday returns which day of the week the Job will run on and
 // will return an error if the Job is not scheduled weekly
 func (j *Job) Weekday() (time.Weekday, error) {
-	j.RLock()
-	defer j.RUnlock()
 	if j.scheduledWeekday == nil {
 		return time.Sunday, ErrNotScheduledWeekday
 	}
@@ -185,15 +153,11 @@ func (j *Job) LimitRunsTo(n int) {
 
 // RemoveAfterLastRun sets the job to be removed after it's last run (when limited)
 func (j *Job) RemoveAfterLastRun() *Job {
-	j.Lock()
-	defer j.Unlock()
 	j.runConfig.removeAfterLastRun = true
 	return j
 }
 
 func (j *Job) getRemoveAfterLastRun() bool {
-	j.RLock()
-	defer j.RUnlock()
 	return j.runConfig.removeAfterLastRun
 }
 
@@ -207,14 +171,10 @@ func (j *Job) shouldRun() bool {
 
 // LastRun returns the time the job was run last
 func (j *Job) LastRun() time.Time {
-	j.RLock()
-	defer j.RUnlock()
 	return j.lastRun
 }
 
 func (j *Job) setLastRun(t time.Time) {
-	j.Lock()
-	defer j.Unlock()
 	j.lastRun = t
 }
 
@@ -233,14 +193,10 @@ func (j *Job) setNextRun(t time.Time) {
 
 // RunCount returns the number of time the job ran so far
 func (j *Job) RunCount() int {
-	j.RLock()
-	defer j.RUnlock()
 	return j.runCount
 }
 
 func (j *Job) stopTimer() {
-	j.Lock()
-	defer j.Unlock()
 	if j.timer != nil {
 		j.timer.Stop()
 	}
diff --git a/scheduler.go b/scheduler.go
index e631adf..7d41e91 100644
--- a/scheduler.go
+++ b/scheduler.go
@@ -21,16 +21,20 @@ type Scheduler struct {
 	runningMutex sync.RWMutex
 	running      bool // represents if the scheduler is running at the moment or not
 
-	time timeWrapper // wrapper around time.Time
+	time     timeWrapper // wrapper around time.Time
+	executor *executor   // executes jobs passed via chan
 }
 
 // NewScheduler creates a new Scheduler
 func NewScheduler(loc *time.Location) *Scheduler {
+	executor := newExecutor()
+
 	return &Scheduler{
 		jobs:     make([]*Job, 0),
 		location: loc,
 		running:  false,
 		time:     &trueTime{},
+		executor: &executor,
 	}
 }
 
@@ -49,6 +53,7 @@ func (s *Scheduler) StartAsync() {
 
 //start starts the scheduler, scheduling and running jobs
 func (s *Scheduler) start() {
+	go s.executor.start()
 	s.setRunning(true)
 	s.runJobs(s.Jobs())
 }
@@ -342,7 +347,8 @@ func (s *Scheduler) run(job *Job) {
 	}
 
 	job.setLastRun(s.now())
-	job.run()
+	job.runCount++
+	s.executor.jobFunctions <- job.jobFunction
 }
 
 // RunAll run all Jobs regardless if they are scheduled to run or not
@@ -367,7 +373,7 @@ func (s *Scheduler) RunAllWithDelay(d time.Duration) {
 // stopping, e.g. using a context.WithCancel().
 func (s *Scheduler) Remove(j interface{}) {
 	s.removeByCondition(func(someJob *Job) bool {
-		return someJob.jobFunc == getFunctionName(j)
+		return someJob.name == getFunctionName(j)
 	})
 }
 
@@ -398,7 +404,7 @@ func (s *Scheduler) RemoveByTag(tag string) error {
 	if err != nil {
 		return err
 	}
-	// Remove job if jobindex is valid
+	// Remove job if job index is valid
 	s.jobs[jobindex].stopTimer()
 	s.setJobs(removeAtIndex(s.jobs, jobindex))
 	return nil
@@ -440,7 +446,7 @@ func (s *Scheduler) RemoveAfterLastRun() *Scheduler {
 // TaskPresent checks if specific job's function was added to the scheduler.
 func (s *Scheduler) TaskPresent(j interface{}) bool {
 	for _, job := range s.Jobs() {
-		if job.jobFunc == getFunctionName(j) {
+		if job.name == getFunctionName(j) {
 			return true
 		}
 	}
@@ -470,6 +476,7 @@ func (s *Scheduler) Stop() {
 
 func (s *Scheduler) stop() {
 	s.setRunning(false)
+	s.executor.stop <- struct{}{}
 }
 
 // Do specifies the jobFunc that should be called every time the Job runs
@@ -490,9 +497,9 @@ func (s *Scheduler) Do(jobFun interface{}, params ...interface{}) (*Job, error)
 	}
 
 	fname := getFunctionName(jobFun)
-	j.funcs[fname] = jobFun
-	j.fparams[fname] = params
-	j.jobFunc = fname
+	j.functions[fname] = jobFun
+	j.params[fname] = params
+	j.name = fname
 
 	// we should not schedule if not running since we cant foresee how long it will take for the scheduler to start
 	if s.IsRunning() {


Test patch:
diff --git a/job_test.go b/job_test.go
index e788207..849ae66 100644
--- a/job_test.go
+++ b/job_test.go
@@ -102,9 +102,9 @@ func TestJob_LimitRunsTo(t *testing.T) {
 	j, _ := NewScheduler(time.Local).Every(1).Second().Do(func() {})
 	j.LimitRunsTo(2)
 	assert.Equal(t, j.shouldRun(), true, "Expecting it to run again")
-	j.run()
+	j.runCount++
 	assert.Equal(t, j.shouldRun(), true, "Expecting it to run again")
-	j.run()
+	j.runCount++
 	assert.Equal(t, j.shouldRun(), false, "Not expecting it to run again")
 }
 
diff --git a/scheduler_test.go b/scheduler_test.go
index 5a8acee..e66f12c 100644
--- a/scheduler_test.go
+++ b/scheduler_test.go
@@ -354,7 +354,7 @@ func TestScheduler_Remove(t *testing.T) {
 
 		s.StartAsync()
 
-		s.Remove(j.funcs[j.jobFunc])
+		s.Remove(j.functions[j.name])
 
 		select {
 		case <-time.After(2 * time.Second):