Repository: go-co-op/gocron
Pull request #212 (2021-08-21 08:26:33)

Replace tag map with thread safe sync.map
### What does this do?
This PR replaces that `uniqueTag` map with the [sync.Map](https://pkg.go.dev/sync#Map) struct which allows safe concurrent operations against the map. Before this, scheduled jobs that executed on the same schedule were prone to causing a "concurrent map" panic.

### Have you included tests for your changes?
Tests have been updated to properly use the sync.map.

### Notes
This change should produce a no-op, just fixes the potential panic from jobs scheduled at the same time that require `uniqueTags`.



Patch:
diff --git a/scheduler.go b/scheduler.go
index 1d9244c..e217610 100644
--- a/scheduler.go
+++ b/scheduler.go
@@ -28,8 +28,9 @@ type Scheduler struct {
 	time     timeWrapper // wrapper around time.Time
 	executor *executor   // executes jobs passed via chan
 
-	tags map[string]struct{} // for storing tags when unique tags is set
+	tags sync.Map // for storing tags when unique tags is set
 
+	tagsUnique      bool // defines whether tags should be unique
 	updateJob       bool // so the scheduler knows to create a new job or update the current
 	waitForInterval bool // defaults jobs to waiting for first interval to start
 }
@@ -42,11 +43,12 @@ func NewScheduler(loc *time.Location) *Scheduler {
 	executor := newExecutor()
 
 	return &Scheduler{
-		jobs:     make([]*Job, 0),
-		location: loc,
-		running:  false,
-		time:     &trueTime{},
-		executor: &executor,
+		jobs:       make([]*Job, 0),
+		location:   loc,
+		running:    false,
+		time:       &trueTime{},
+		executor:   &executor,
+		tagsUnique: false,
 	}
 }
 
@@ -536,9 +538,9 @@ func (s *Scheduler) removeJobsUniqueTags(job *Job) {
 	if job == nil {
 		return
 	}
-	if s.tags != nil && len(job.tags) > 0 {
+	if s.tagsUnique && len(job.tags) > 0 {
 		for _, tag := range job.tags {
-			delete(s.tags, tag)
+			s.tags.Delete(tag)
 		}
 	}
 }
@@ -637,8 +639,12 @@ func (s *Scheduler) Clear() {
 		job.stop()
 	}
 	s.setJobs(make([]*Job, 0))
-	if s.tags != nil {
-		s.TagsUnique()
+	// If unique tags was enabled, delete all the tags loaded in the tags sync.Map
+	if s.tagsUnique {
+		s.tags.Range(func(key interface{}, value interface{}) bool {
+			s.tags.Delete(key)
+			return true
+		})
 	}
 }
 
@@ -730,13 +736,13 @@ func (s *Scheduler) At(i interface{}) *Scheduler {
 func (s *Scheduler) Tag(t ...string) *Scheduler {
 	job := s.getCurrentJob()
 
-	if s.tags != nil {
+	if s.tagsUnique {
 		for _, tag := range t {
-			if _, ok := s.tags[tag]; ok {
+			if _, ok := s.tags.Load(tag); ok {
 				job.error = wrapOrError(job.error, ErrTagsUnique(tag))
 				return s
 			}
-			s.tags[tag] = struct{}{}
+			s.tags.Store(tag, struct{}{})
 		}
 	}
 
@@ -912,7 +918,7 @@ func (s *Scheduler) now() time.Time {
 // This does not enforce uniqueness on tags added via
 // (j *Job) Tag()
 func (s *Scheduler) TagsUnique() {
-	s.tags = make(map[string]struct{})
+	s.tagsUnique = true
 }
 
 // Job puts the provided job in focus for the purpose


Test patch:
diff --git a/scheduler_test.go b/scheduler_test.go
index 4701ced..31c56f4 100644
--- a/scheduler_test.go
+++ b/scheduler_test.go
@@ -343,8 +343,6 @@ func TestScheduler_Remove(t *testing.T) {
 
 		s.Remove(task)
 		assert.Equal(t, 1, s.Len(), "Incorrect number of jobs after removing non-existent job")
-
-		assert.Zero(t, len(s.tags))
 	})
 
 	t.Run("remove from running scheduler", func(t *testing.T) {
@@ -402,7 +400,8 @@ func TestScheduler_RemoveByReference(t *testing.T) {
 			t.Fatal("job ran after being removed")
 		}
 
-		assert.Zero(t, len(s.tags))
+		_, ok := s.tags.Load("tag1")
+		assert.False(t, ok)
 	})
 }
 
@@ -541,6 +540,41 @@ func TestClear(t *testing.T) {
 	assert.Equal(t, 1, counter)
 }
 
+func TestClearUnique(t *testing.T) {
+	s := NewScheduler(time.UTC)
+	s.TagsUnique()
+	semaphore := make(chan bool)
+
+	_, err := s.Every("100ms").Tag("tag1").Do(func() {
+		semaphore <- true
+	})
+	require.NoError(t, err)
+
+	s.StartAsync()
+
+	s.Clear()
+	assert.Equal(t, 0, s.Len())
+
+	var counter int
+	now := time.Now()
+	for time.Now().Before(now.Add(200 * time.Millisecond)) {
+		select {
+		case <-semaphore:
+			counter++
+		default:
+		}
+	}
+
+	// job should run only once - immediately and then
+	// be stopped on s.Clear()
+	assert.Equal(t, 1, counter)
+
+	s.tags.Range(func(key, value interface{}) bool {
+		assert.FailNow(t, "map should be empty")
+		return true
+	})
+}
+
 func TestSetUnit(t *testing.T) {
 
 	testCases := []struct {
@@ -866,7 +900,10 @@ func TestRunJobsWithLimit(t *testing.T) {
 			require.LessOrEqual(t, counter, 1)
 		}
 
-		assert.Zero(t, len(s.tags))
+		s.tags.Range(func(key, value interface{}) bool {
+			assert.FailNow(t, "map should be empty")
+			return true
+		})
 	})
 }