Repository: go-co-op/gocron
Pull request #473 (2023-05-06 12:47:45)

fix: deadlock for blocking channel
### What does this do?

This is a scenario that occurs with a very small probability. The `StartBlocking()` and `Stop()` methods are executed almost concurrently. When the stop is notified through the channel, the start has not been monitored, so a deadlock problem occurs.

When will the stop method and start be executed concurrently, such as closing gocron after receiving a linux signal.

![image](https://user-images.githubusercontent.com/3785409/236624840-46d4e1a1-fd59-481f-87c7-c0b313e76e4c.png)


### Which issue(s) does this PR fix/relate to?
<!--- Put `Resolves #XXX` here to auto-close the issue that your PR fixes (if such) --->


### List any changes that modify/break current functionality


### Have you included tests for your changes?


### Did you document any new/modified functionality?

- [ ] Updated `example_test.go`
- [ ] Updated `README.md`

### Notes



Patch:
diff --git a/README.md b/README.md
index e9a4c70..45cd39c 100644
--- a/README.md
+++ b/README.md
@@ -88,16 +88,16 @@ For more examples, take a look in our [go docs](https://pkg.go.dev/github.com/go
 
 There are several options available to restrict how jobs run:
 
-| Mode                       | Function                  | Behavior                                                                                             |
-|----------------------------|---------------------------|------------------------------------------------------------------------------------------------------|
-| Default                    |                           | jobs are rescheduled at every interval                                                               |
-| Job singleton              | `SingletonMode()`         | a long running job will not be rescheduled until the current run is completed                        |
-| Scheduler limit            | `SetMaxConcurrentJobs()`  | set a collective maximum number of concurrent jobs running across the scheduler                      |
-| Distributed locking (BETA) | `WithDistributedLocker()` | prevents the same job from being run more than once when running multiple instances of the scheduler |
+| Mode                | Function                  | Behavior                                                                                             |
+|---------------------|---------------------------|------------------------------------------------------------------------------------------------------|
+| Default             |                           | jobs are rescheduled at every interval                                                               |
+| Job singleton       | `SingletonMode()`         | a long running job will not be rescheduled until the current run is completed                        |
+| Scheduler limit     | `SetMaxConcurrentJobs()`  | set a collective maximum number of concurrent jobs running across the scheduler                      |
+| Distributed locking | `WithDistributedLocker()` | prevents the same job from being run more than once when running multiple instances of the scheduler |
 
 ## Distributed Locker Implementations
 
-- Redis: [redis.go](lockers/redislock/redislock.go) `go get github.com/go-co-op/gocron/lockers/redislock`
+- Redis: [redislock](lockers/redislock/README.md) `go get github.com/go-co-op/gocron/lockers/redislock`
 
 ## Tags
 
@@ -125,8 +125,8 @@ s.RunByTag("tag")
 
 - Q: I'm running multiple pods on a distributed environment. How can I make a job not run once per pod causing duplication?
   - We recommend using your own lock solution within the jobs themselves (you could use [Redis](https://redis.io/topics/distlock), for example)
-  - A2: Currently in BETA (please provide feedback): Use the scheduler option `WithDistributedLocker` and either use an implemented backend
-    or implement your own and contribute it back in a PR (we hope)!
+  - A2: Use the scheduler option `WithDistributedLocker` and either use an implemented [backend](lockers)
+    or implement your own and contribute it back in a PR!
 
 - Q: I've removed my job from the scheduler, but how can I stop a long-running job that has already been triggered?
   - A: We recommend using a means of canceling your job, e.g. a `context.WithCancel()`.
diff --git a/executor.go b/executor.go
index 708e7cb..aa16b84 100644
--- a/executor.go
+++ b/executor.go
@@ -48,6 +48,7 @@ type executor struct {
 	limitModeFuncsRunning   *atomic.Int64    // tracks the count of limited mode funcs running
 	limitModeFuncWg         *sync.WaitGroup  // allow the executor to wait for limit mode functions to wrap up
 	limitModeQueue          chan jobFunction // pass job functions to the limit mode workers
+	limitModeQueueMu        *sync.Mutex      // for protecting the limitModeQueue
 	limitModeRunningJobs    *atomic.Int64    // tracks the count of running jobs to check against the max
 	stopped                 *atomic.Bool     // allow workers to drain the buffered limitModeQueue
 
@@ -61,7 +62,7 @@ func newExecutor() executor {
 		limitModeFuncsRunning: &atomic.Int64{},
 		limitModeFuncWg:       &sync.WaitGroup{},
 		limitModeRunningJobs:  &atomic.Int64{},
-		limitModeQueue:        make(chan jobFunction, 1000),
+		limitModeQueueMu:      &sync.Mutex{},
 	}
 	return e
 }
@@ -121,14 +122,22 @@ func (e *executor) start() {
 	e.jobsWg = &sync.WaitGroup{}
 
 	e.stopped = &atomic.Bool{}
+
+	e.limitModeQueueMu.Lock()
+	e.limitModeQueue = make(chan jobFunction, 1000)
+	e.limitModeQueueMu.Unlock()
 	go e.run()
 }
 
 func (e *executor) runJob(f jobFunction) {
 	switch f.runConfig.mode {
 	case defaultMode:
+		lockKey := f.jobName
+		if lockKey == "" {
+			lockKey = f.funcName
+		}
 		if e.distributedLocker != nil {
-			l, err := e.distributedLocker.Lock(f.ctx, f.name)
+			l, err := e.distributedLocker.Lock(f.ctx, lockKey)
 			if err != nil || l == nil {
 				return
 			}
@@ -185,7 +194,7 @@ func (e *executor) run() {
 				if panicHandler != nil {
 					defer func() {
 						if r := recover(); r != any(nil) {
-							panicHandler(f.name, r)
+							panicHandler(f.funcName, r)
 						}
 					}()
 				}
@@ -194,10 +203,16 @@ func (e *executor) run() {
 					switch e.limitMode {
 					case RescheduleMode:
 						if e.limitModeRunningJobs.Load() < int64(e.limitModeMaxRunningJobs) {
-							e.limitModeQueue <- f
+							select {
+							case e.limitModeQueue <- f:
+							case <-e.ctx.Done():
+							}
 						}
 					case WaitMode:
-						e.limitModeQueue <- f
+						select {
+						case e.limitModeQueue <- f:
+						case <-e.ctx.Done():
+						}
 					}
 					return
 				}
@@ -226,5 +241,8 @@ func (e *executor) stop() {
 	}
 	if e.limitModeMaxRunningJobs > 0 {
 		e.limitModeFuncWg.Wait()
+		e.limitModeQueueMu.Lock()
+		e.limitModeQueue = nil
+		e.limitModeQueueMu.Unlock()
 	}
 }
diff --git a/gocron.go b/gocron.go
index 58931ea..ec97f45 100644
--- a/gocron.go
+++ b/gocron.go
@@ -28,7 +28,7 @@ var (
 // SetPanicHandler sets the global panicHandler to the given function.
 // Leaving it nil or setting it to nil disables automatic panic handling.
 // If the panicHandler is not nil, any panic that occurs during executing a job will be recovered
-// and the panicHandlerFunc will be called with the job's name and the recover data.
+// and the panicHandlerFunc will be called with the job's funcName and the recover data.
 func SetPanicHandler(handler PanicHandlerFunc) {
 	panicHandlerMutex.Lock()
 	defer panicHandlerMutex.Unlock()
diff --git a/job.go b/job.go
index 653d391..74a2381 100644
--- a/job.go
+++ b/job.go
@@ -45,7 +45,8 @@ type jobFunction struct {
 	function          any                // task's function
 	parameters        []any              // task's function parameters
 	parametersLen     int                // length of the passed parameters
-	name              string             // nolint the function name to run
+	jobName           string             // key of the distributed lock
+	funcName          string             // the name of the function - e.g. main.func1
 	runConfig         runConfig          // configuration for how many times to run the job
 	singletonQueue    chan struct{}      // queues jobs for the singleton runner to handle
 	singletonRunnerOn *atomic.Bool       // whether the runner function for singleton is running
@@ -74,7 +75,8 @@ func (jf *jobFunction) copy() jobFunction {
 		function:          jf.function,
 		parameters:        nil,
 		parametersLen:     jf.parametersLen,
-		name:              jf.name,
+		funcName:          jf.funcName,
+		jobName:           jf.jobName,
 		runConfig:         jf.runConfig,
 		singletonQueue:    jf.singletonQueue,
 		ctx:               jf.ctx,
@@ -135,6 +137,16 @@ func newJob(interval int, startImmediately bool, singletonMode bool) *Job {
 	return job
 }
 
+// Name sets the name of the current job.
+//
+// If the scheduler is running using WithDistributedLocker(),
+// the job name is used as the distributed lock key.
+func (j *Job) Name(name string) {
+	j.mu.Lock()
+	defer j.mu.Unlock()
+	j.jobName = name
+}
+
 func (j *Job) setRandomInterval(a, b int) {
 	j.random.rand = rand.New(rand.NewSource(time.Now().UnixNano())) // nolint
 
diff --git a/lockers/redislock/README.md b/lockers/redislock/README.md
new file mode 100644
index 0000000..cb50446
--- /dev/null
+++ b/lockers/redislock/README.md
@@ -0,0 +1,46 @@
+# redislock
+
+## install
+
+```
+go get github.com/go-co-op/gocron/lockers/redislock
+```
+
+## usage
+
+Here is an example usage that would be deployed in multiple instances
+
+```go
+package main
+
+import (
+	"time"
+
+	"github.com/go-co-op/gocron"
+	"github.com/go-co-op/gocron/lockers/redislock"
+	"github.com/redis/go-redis/v9"
+)
+
+func main() {
+	redisOptions := &redis.Options{
+		Addr: "localhost:6379",
+	}
+	redisClient := redis.NewClient(redisOptions)
+	locker, err := redislock.NewRedisLocker(redisClient)
+	if err != nil {
+		// handle the error
+	}
+
+	s := gocron.NewScheduler(time.UTC)
+	s.WithDistributedLocker(locker)
+
+	_, err = s.Every("1s").Name("unique_name").Do(func() {
+		// task to do
+	})
+	if err != nil {
+		// handle the error
+	}
+
+	s.StartBlocking()
+}
+```
\ No newline at end of file
diff --git a/docker-compose.yml b/lockers/redislock/docker-compose.yml
similarity index 100%
rename from docker-compose.yml
rename to lockers/redislock/docker-compose.yml
diff --git a/scheduler.go b/scheduler.go
index 36498b6..39325b1 100644
--- a/scheduler.go
+++ b/scheduler.go
@@ -82,7 +82,12 @@ func (s *Scheduler) StartBlocking() {
 	s.startBlockingStopChanMutex.Lock()
 	s.startBlockingStopChan = make(chan struct{}, 1)
 	s.startBlockingStopChanMutex.Unlock()
+
 	<-s.startBlockingStopChan
+
+	s.startBlockingStopChanMutex.Lock()
+	s.startBlockingStopChan = nil
+	s.startBlockingStopChanMutex.Unlock()
 }
 
 // StartAsync starts all jobs without blocking the current thread
@@ -126,6 +131,16 @@ func (s *Scheduler) Jobs() []*Job {
 	return s.jobs
 }
 
+// Name sets the name of the current job.
+//
+// If the scheduler is running using WithDistributedLocker(), the job name is used
+// as the distributed lock key.
+func (s *Scheduler) Name(name string) *Scheduler {
+	job := s.getCurrentJob()
+	job.jobName = name
+	return s
+}
+
 func (s *Scheduler) setJobs(jobs []*Job) {
 	s.jobsMutex.Lock()
 	defer s.jobsMutex.Unlock()
@@ -658,7 +673,7 @@ func (s *Scheduler) Remove(job any) {
 	j := s.findJobByTaskName(fName)
 	s.removeJobsUniqueTags(j)
 	s.removeByCondition(func(someJob *Job) bool {
-		return someJob.name == fName
+		return someJob.funcName == fName
 	})
 }
 
@@ -674,7 +689,7 @@ func (s *Scheduler) RemoveByReference(job *Job) {
 
 func (s *Scheduler) findJobByTaskName(name string) *Job {
 	for _, job := range s.Jobs() {
-		if job.name == name {
+		if job.funcName == name {
 			return job
 		}
 	}
@@ -820,7 +835,7 @@ func (s *Scheduler) SingletonModeAll() {
 // TaskPresent checks if specific job's function was added to the scheduler.
 func (s *Scheduler) TaskPresent(j any) bool {
 	for _, job := range s.Jobs() {
-		if job.name == getFunctionName(j) {
+		if job.funcName == getFunctionName(j) {
 			return true
 		}
 	}
@@ -868,10 +883,10 @@ func (s *Scheduler) Stop() {
 }
 
 func (s *Scheduler) stop() {
-	s.setRunning(false)
 	s.stopJobs(s.jobs)
 	s.executor.stop()
 	s.StopBlockingChan()
+	s.setRunning(false)
 }
 
 func (s *Scheduler) stopJobs(jobs []*Job) {
@@ -915,10 +930,10 @@ func (s *Scheduler) doCommon(jobFun any, params ...any) (*Job, error) {
 	}
 
 	fname := getFunctionName(jobFun)
-	if job.name != fname {
+	if job.funcName != fname {
 		job.function = jobFun
 		job.parameters = params
-		job.name = fname
+		job.funcName = fname
 	}
 
 	f := reflect.ValueOf(jobFun)
@@ -1376,8 +1391,8 @@ func (s *Scheduler) CustomTimer(customTimer func(d time.Duration, f func()) *tim
 
 func (s *Scheduler) StopBlockingChan() {
 	s.startBlockingStopChanMutex.Lock()
-	if s.startBlockingStopChan != nil {
-		s.startBlockingStopChan <- struct{}{}
+	if s.IsRunning() && s.startBlockingStopChan != nil {
+		close(s.startBlockingStopChan)
 	}
 	s.startBlockingStopChanMutex.Unlock()
 }
@@ -1396,8 +1411,13 @@ func (s *Scheduler) StopBlockingChan() {
 // Another strategy is to use the Cron or CronWithSeconds methods as they
 // use the same behavior described above using StartAt.
 //
-// NOTE - the Locker will NOT lock jobs using any of the limiting functions:
-// SingletonMode, SingletonModeAll or SetMaxConcurrentJobs
+// NOTE - the Locker will NOT lock jobs using the singleton options:
+// SingletonMode, or SingletonModeAll
+//
+// NOTE - beware of potential race conditions when running the Locker
+// with SetMaxConcurrentJobs and WaitMode as jobs are not guaranteed
+// to be locked when each scheduler's is below its limit and able
+// to run the job.
 func (s *Scheduler) WithDistributedLocker(l Locker) {
 	s.executor.distributedLocker = l
 }


Test patch:
diff --git a/executor_test.go b/executor_test.go
index 21d76a5..7ed1bfd 100644
--- a/executor_test.go
+++ b/executor_test.go
@@ -16,7 +16,7 @@ func Test_ExecutorExecute(t *testing.T) {
 	e.start()
 
 	e.jobFunctions <- jobFunction{
-		name: "test_fn",
+		funcName: "test_fn",
 		function: func(arg string) {
 			assert.Equal(t, arg, "test")
 			wg.Done()
@@ -47,7 +47,7 @@ func Test_ExecutorPanicHandling(t *testing.T) {
 	e.start()
 
 	e.jobFunctions <- jobFunction{
-		name: "test_fn",
+		funcName: "test_fn",
 		function: func() {
 			defer wg.Done()
 			a := make([]string, 0)
diff --git a/gocron_test.go b/gocron_test.go
index 4951256..5c25397 100644
--- a/gocron_test.go
+++ b/gocron_test.go
@@ -181,7 +181,7 @@ func Test_getFunctionName(t *testing.T) {
 		want string
 	}{
 		{
-			name: "test get function name",
+			name: "test get function funcName",
 			args: args{
 				fn: Test_getFunctionName,
 			},
diff --git a/scheduler_test.go b/scheduler_test.go
index 2f3b075..d4513c0 100644
--- a/scheduler_test.go
+++ b/scheduler_test.go
@@ -2625,3 +2625,66 @@ func runTestWithDistributedLocking(t *testing.T, maxConcurrentJobs int) {
 	}
 	assert.Len(t, results, 4)
 }
+
+func TestScheduler_WithDistributedLocker_With_Name(t *testing.T) {
+	testCases := []struct {
+		description string
+		name        string
+		expected    int
+	}{
+		{"with job name set", "foo", 4},
+		// The distributed locking will use the function funcName as its lock key
+		// if the job funcName is empty. So the result will be 8 instead of 4 since
+		// they are regarded as different jobs.
+		{"without job name", "", 8},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.description, func(t *testing.T) {
+			resultChan := make(chan int, 10)
+			f1 := func(schedulerInstance int) {
+				resultChan <- schedulerInstance
+				time.Sleep(100 * time.Millisecond)
+			}
+			f2 := func(schedulerInstance int) {
+				resultChan <- schedulerInstance
+				time.Sleep(100 * time.Millisecond)
+			}
+
+			l := &locker{
+				store: make(map[string]struct{}, 0),
+			}
+
+			schedulers := make([]*Scheduler, 0)
+
+			var err error
+			var job *Job
+			for i := 0; i < 2; i++ {
+				s := NewScheduler(time.UTC)
+				s.WithDistributedLocker(l)
+				if i%2 == 0 {
+					job, err = s.Every("500ms").Name(tc.name).Do(f1, 1)
+				} else {
+					job, err = s.Every("500ms").Name(tc.name).Do(f2, 1)
+				}
+				require.NoError(t, err)
+				require.Equal(t, job.jobName, tc.name)
+				schedulers = append(schedulers, s)
+			}
+			for i := range schedulers {
+				schedulers[i].StartAsync()
+			}
+			time.Sleep(1700 * time.Millisecond)
+			for i := range schedulers {
+				schedulers[i].Stop()
+			}
+			close(resultChan)
+
+			var results []int
+			for r := range resultChan {
+				results = append(results, r)
+			}
+			assert.Len(t, results, tc.expected)
+		})
+	}
+}