fix(worker): fix concurrent map iteration and map write (#698)

Co-authored-by: zhenghaoz <zhangzhenghao@hotmail.com>

config/config.go

@@ -334,6 +334,7 @@ func (config *Config) OfflineRecommendDigest(option ...DigestOption) string {
 	})
 
 	var builder strings.Builder
+	config.Recommend.Offline.Lock()
 	builder.WriteString(fmt.Sprintf("%v-%v-%v-%v-%v-%v-%v-%v",
 		config.Recommend.Offline.ExploreRecommend,
 		config.Recommend.Offline.EnableLatestRecommend,
@@ -344,6 +345,7 @@ func (config *Config) OfflineRecommendDigest(option ...DigestOption) string {
 		options.enableRanking,
 		config.Recommend.Replacement.EnableReplacement,
 	))
+	config.Recommend.Offline.UnLock()
 	if config.Recommend.Offline.EnablePopularRecommend {
 		builder.WriteString(fmt.Sprintf("-%v", config.Recommend.Popular.PopularWindow))
 	}

worker/worker_test.go

@@ -727,6 +727,55 @@ func TestWorker_Sync(t *testing.T) {
 	done <- struct{}{}
 }
 
+func TestWorker_SyncRecommend(t *testing.T) {
+	cfg := config.GetDefaultConfig()
+	cfg.Recommend.Offline.ExploreRecommend = map[string]float64{"popular": 0.5}
+	master := newMockMaster(t)
+	master.meta.Config = marshal(t, cfg)
+	go master.Start(t)
+	address := <-master.addr
+	conn, err := grpc.Dial(address, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	assert.NoError(t, err)
+	worker := &Worker{
+		Settings:     config.NewSettings(),
+		jobs:         1,
+		testMode:     true,
+		masterClient: protocol.NewMasterClient(conn),
+		syncedChan:   parallel.NewConditionChannel(),
+		ticker:       time.NewTicker(time.Minute),
+	}
+	worker.Sync()
+
+	stopSync := make(chan struct{})
+	go func() {
+		for {
+			select {
+			case <-stopSync:
+				return
+			default:
+				worker.Sync()
+			}
+		}
+	}()
+
+	stopRecommend := make(chan struct{})
+	go func() {
+		for {
+			select {
+			case <-stopRecommend:
+				return
+			default:
+				worker.Settings.Config.OfflineRecommendDigest()
+			}
+		}
+	}()
+
+	time.Sleep(time.Second)
+	stopSync <- struct{}{}
+	stopRecommend <- struct{}{}
+	master.Stop()
+}
+
 type mockFactorizationMachine struct {
 	click.BaseFactorizationMachine
 }