diff --git a/notification-dingtalk/dingtalk_notification.go b/notification-dingtalk/dingtalk_notification.go index 8db367705..f295a20ab 100644 --- a/notification-dingtalk/dingtalk_notification.go +++ b/notification-dingtalk/dingtalk_notification.go @@ -63,12 +63,7 @@ func (n *Notification) Info() plugin.Info { // GetNewQuestionSubscribers returns the subscribers of the new question notification func (n *Notification) GetNewQuestionSubscribers() (userIDs []string) { - for userID, conf := range n.UserConfigCache.userConfigMapping { - if conf.AllNewQuestions { - userIDs = append(userIDs, userID) - } - } - return userIDs + return n.UserConfigCache.NewQuestionSubscriberIDs() } // Notify sends a notification to the user diff --git a/notification-dingtalk/user_config.go b/notification-dingtalk/user_config.go index 5cfdad4b1..8ec58f817 100644 --- a/notification-dingtalk/user_config.go +++ b/notification-dingtalk/user_config.go @@ -39,7 +39,7 @@ type UserConfig struct { type UserConfigCache struct { // key: userID value: user config userConfigMapping map[string]*UserConfig - sync.Mutex + sync.RWMutex } func NewUserConfigCache() *UserConfigCache { @@ -55,6 +55,19 @@ func (ucc *UserConfigCache) SetUserConfig(userID string, config *UserConfig) { ucc.userConfigMapping[userID] = config } +func (ucc *UserConfigCache) NewQuestionSubscriberIDs() []string { + ucc.RLock() + defer ucc.RUnlock() + + userIDs := make([]string, 0) + for userID, conf := range ucc.userConfigMapping { + if conf.AllNewQuestions { + userIDs = append(userIDs, userID) + } + } + return userIDs +} + func (n *Notification) UserConfigFields() []plugin.ConfigField { fields := make([]plugin.ConfigField, 0) // Show tip for user, if the notification service is disabled diff --git a/notification-dingtalk/user_config_concurrency_test.go b/notification-dingtalk/user_config_concurrency_test.go new file mode 100644 index 000000000..3da38343a --- /dev/null +++ b/notification-dingtalk/user_config_concurrency_test.go @@ -0,0 +1,51 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0. + */ + +package dingtalk + +import ( + "fmt" + "sync" + "testing" +) + +func TestUserConfigCacheConcurrentNewQuestionSubscribers(t *testing.T) { + cache := NewUserConfigCache() + const workers, iterations = 8, 500 + start := make(chan struct{}) + var wg sync.WaitGroup + for worker := range workers { + wg.Add(1) + go func() { + defer wg.Done() + <-start + for i := range iterations { + cache.SetUserConfig(fmt.Sprintf("writer-%d-user-%d", worker, i%32), &UserConfig{AllNewQuestions: i%2 == 0}) + } + }() + } + for range workers { + wg.Add(1) + go func() { + defer wg.Done() + <-start + for range iterations { + _ = cache.NewQuestionSubscriberIDs() + } + }() + } + close(start) + wg.Wait() + + cache.SetUserConfig("subscriber", &UserConfig{AllNewQuestions: true}) + for _, userID := range cache.NewQuestionSubscriberIDs() { + if userID == "subscriber" { + return + } + } + t.Fatal("subscriber missing from snapshot") +} diff --git a/notification-lark/notification.go b/notification-lark/notification.go index c9954cb75..3e599889e 100644 --- a/notification-lark/notification.go +++ b/notification-lark/notification.go @@ -211,12 +211,7 @@ func makeCardMsg(args plugin.NotificationMessage) Card { // GetNewQuestionSubscribers returns the subscribers of the new question notification func (n *Notification) GetNewQuestionSubscribers() (userIDs []string) { - for userID, conf := range n.userConfigCache.userConfigMapping { - if conf.AllNewQuestions { - userIDs = append(userIDs, userID) - } - } - return userIDs + return n.userConfigCache.NewQuestionSubscriberIDs() } // Notify sends a notification to the user diff --git a/notification-lark/user_config.go b/notification-lark/user_config.go index 2a6246861..57f046885 100644 --- a/notification-lark/user_config.go +++ b/notification-lark/user_config.go @@ -39,7 +39,7 @@ type UserConfig struct { type UserConfigCache struct { userConfigMapping map[string]*UserConfig - sync.Mutex + sync.RWMutex } func NewUserConfigCache() *UserConfigCache { @@ -55,6 +55,19 @@ func (ucc *UserConfigCache) SetUserConfig(userID string, config *UserConfig) { ucc.userConfigMapping[userID] = config } +func (ucc *UserConfigCache) NewQuestionSubscriberIDs() []string { + ucc.RLock() + defer ucc.RUnlock() + + userIDs := make([]string, 0) + for userID, conf := range ucc.userConfigMapping { + if conf.AllNewQuestions { + userIDs = append(userIDs, userID) + } + } + return userIDs +} + func (n *Notification) UserConfigFields() []plugin.ConfigField { return []plugin.ConfigField{ { diff --git a/notification-lark/user_config_concurrency_test.go b/notification-lark/user_config_concurrency_test.go new file mode 100644 index 000000000..a466c086f --- /dev/null +++ b/notification-lark/user_config_concurrency_test.go @@ -0,0 +1,51 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0. + */ + +package lark + +import ( + "fmt" + "sync" + "testing" +) + +func TestUserConfigCacheConcurrentNewQuestionSubscribers(t *testing.T) { + cache := NewUserConfigCache() + const workers, iterations = 8, 500 + start := make(chan struct{}) + var wg sync.WaitGroup + for worker := range workers { + wg.Add(1) + go func() { + defer wg.Done() + <-start + for i := range iterations { + cache.SetUserConfig(fmt.Sprintf("writer-%d-user-%d", worker, i%32), &UserConfig{AllNewQuestions: i%2 == 0}) + } + }() + } + for range workers { + wg.Add(1) + go func() { + defer wg.Done() + <-start + for range iterations { + _ = cache.NewQuestionSubscriberIDs() + } + }() + } + close(start) + wg.Wait() + + cache.SetUserConfig("subscriber", &UserConfig{AllNewQuestions: true}) + for _, userID := range cache.NewQuestionSubscriberIDs() { + if userID == "subscriber" { + return + } + } + t.Fatal("subscriber missing from snapshot") +} diff --git a/notification-slack/slack_notification.go b/notification-slack/slack_notification.go index 6d1ba9ac9..dfb3ea391 100644 --- a/notification-slack/slack_notification.go +++ b/notification-slack/slack_notification.go @@ -63,12 +63,7 @@ func (n *Notification) Info() plugin.Info { // GetNewQuestionSubscribers returns the subscribers of the new question notification func (n *Notification) GetNewQuestionSubscribers() (userIDs []string) { - for userID, conf := range n.UserConfigCache.userConfigMapping { - if conf.AllNewQuestions { - userIDs = append(userIDs, userID) - } - } - return userIDs + return n.UserConfigCache.NewQuestionSubscriberIDs() } // Notify sends a notification to the user diff --git a/notification-slack/user_config.go b/notification-slack/user_config.go index 7ea077efc..d37c02e05 100644 --- a/notification-slack/user_config.go +++ b/notification-slack/user_config.go @@ -43,7 +43,7 @@ type UserConfig struct { type UserConfigCache struct { // key: userID value: user config userConfigMapping map[string]*UserConfig - sync.Mutex + sync.RWMutex } func NewUserConfigCache() *UserConfigCache { @@ -59,6 +59,19 @@ func (ucc *UserConfigCache) SetUserConfig(userID string, config *UserConfig) { ucc.userConfigMapping[userID] = config } +func (ucc *UserConfigCache) NewQuestionSubscriberIDs() []string { + ucc.RLock() + defer ucc.RUnlock() + + userIDs := make([]string, 0) + for userID, conf := range ucc.userConfigMapping { + if conf.AllNewQuestions { + userIDs = append(userIDs, userID) + } + } + return userIDs +} + func (n *Notification) UserConfigFields() []plugin.ConfigField { fields := make([]plugin.ConfigField, 0) // Show tip for user, if the notification service is disabled diff --git a/notification-slack/user_config_concurrency_test.go b/notification-slack/user_config_concurrency_test.go new file mode 100644 index 000000000..4afb8912a --- /dev/null +++ b/notification-slack/user_config_concurrency_test.go @@ -0,0 +1,51 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0. + */ + +package slack_notification + +import ( + "fmt" + "sync" + "testing" +) + +func TestUserConfigCacheConcurrentNewQuestionSubscribers(t *testing.T) { + cache := NewUserConfigCache() + const workers, iterations = 8, 500 + start := make(chan struct{}) + var wg sync.WaitGroup + for worker := range workers { + wg.Add(1) + go func() { + defer wg.Done() + <-start + for i := range iterations { + cache.SetUserConfig(fmt.Sprintf("writer-%d-user-%d", worker, i%32), &UserConfig{AllNewQuestions: i%2 == 0}) + } + }() + } + for range workers { + wg.Add(1) + go func() { + defer wg.Done() + <-start + for range iterations { + _ = cache.NewQuestionSubscriberIDs() + } + }() + } + close(start) + wg.Wait() + + cache.SetUserConfig("subscriber", &UserConfig{AllNewQuestions: true}) + for _, userID := range cache.NewQuestionSubscriberIDs() { + if userID == "subscriber" { + return + } + } + t.Fatal("subscriber missing from snapshot") +} diff --git a/notification-wecom/notification_wecom.go b/notification-wecom/notification_wecom.go index 0de5a8b2e..afdc0fded 100644 --- a/notification-wecom/notification_wecom.go +++ b/notification-wecom/notification_wecom.go @@ -62,12 +62,7 @@ func (*Notification) Info() plugin.Info { // GetNewQuestionSubscribers returns the subscribers of the new question notification func (n *Notification) GetNewQuestionSubscribers() (userIDs []string) { - for userID, conf := range n.UserConfigCache.userConfigMapping { - if conf.AllNewQuestions { - userIDs = append(userIDs, userID) - } - } - return userIDs + return n.UserConfigCache.NewQuestionSubscriberIDs() } // Notify sends a notification to the user diff --git a/notification-wecom/user_config.go b/notification-wecom/user_config.go index f61e2492d..17ff018c8 100644 --- a/notification-wecom/user_config.go +++ b/notification-wecom/user_config.go @@ -39,7 +39,7 @@ type UserConfig struct { type UserConfigCache struct { // key: userID value: user config userConfigMapping map[string]*UserConfig - sync.Mutex + sync.RWMutex } func NewUserConfigCache() *UserConfigCache { @@ -55,6 +55,19 @@ func (ucc *UserConfigCache) SetUserConfig(userID string, config *UserConfig) { ucc.userConfigMapping[userID] = config } +func (ucc *UserConfigCache) NewQuestionSubscriberIDs() []string { + ucc.RLock() + defer ucc.RUnlock() + + userIDs := make([]string, 0) + for userID, conf := range ucc.userConfigMapping { + if conf.AllNewQuestions { + userIDs = append(userIDs, userID) + } + } + return userIDs +} + func (n *Notification) UserConfigFields() []plugin.ConfigField { fields := make([]plugin.ConfigField, 0) // Show tip for user, if the notification service is disabled diff --git a/notification-wecom/user_config_concurrency_test.go b/notification-wecom/user_config_concurrency_test.go new file mode 100644 index 000000000..8ea44c619 --- /dev/null +++ b/notification-wecom/user_config_concurrency_test.go @@ -0,0 +1,51 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0. + */ + +package wecom + +import ( + "fmt" + "sync" + "testing" +) + +func TestUserConfigCacheConcurrentNewQuestionSubscribers(t *testing.T) { + cache := NewUserConfigCache() + const workers, iterations = 8, 500 + start := make(chan struct{}) + var wg sync.WaitGroup + for worker := range workers { + wg.Add(1) + go func() { + defer wg.Done() + <-start + for i := range iterations { + cache.SetUserConfig(fmt.Sprintf("writer-%d-user-%d", worker, i%32), &UserConfig{AllNewQuestions: i%2 == 0}) + } + }() + } + for range workers { + wg.Add(1) + go func() { + defer wg.Done() + <-start + for range iterations { + _ = cache.NewQuestionSubscriberIDs() + } + }() + } + close(start) + wg.Wait() + + cache.SetUserConfig("subscriber", &UserConfig{AllNewQuestions: true}) + for _, userID := range cache.NewQuestionSubscriberIDs() { + if userID == "subscriber" { + return + } + } + t.Fatal("subscriber missing from snapshot") +} diff --git a/user-center-slack/notification.go b/user-center-slack/notification.go index c647e7308..a75a50290 100644 --- a/user-center-slack/notification.go +++ b/user-center-slack/notification.go @@ -31,12 +31,7 @@ import ( // GetNewQuestionSubscribers returns the subscribers of the new question notification func (uc *UserCenter) GetNewQuestionSubscribers() (userIDs []string) { - for userID, conf := range uc.UserConfigCache.userConfigMapping { - if conf.AllNewQuestions { - userIDs = append(userIDs, userID) - } - } - return userIDs + return uc.UserConfigCache.NewQuestionSubscriberIDs() } // Notify sends a notification to the user diff --git a/user-center-slack/user_config.go b/user-center-slack/user_config.go index 5f2543430..92d1f48fa 100644 --- a/user-center-slack/user_config.go +++ b/user-center-slack/user_config.go @@ -43,7 +43,7 @@ type UserConfig struct { type UserConfigCache struct { // key: userID value: user config userConfigMapping map[string]*UserConfig - sync.Mutex + sync.RWMutex } func NewUserConfigCache() *UserConfigCache { @@ -59,6 +59,19 @@ func (ucc *UserConfigCache) SetUserConfig(userID string, config *UserConfig) { ucc.userConfigMapping[userID] = config } +func (ucc *UserConfigCache) NewQuestionSubscriberIDs() []string { + ucc.RLock() + defer ucc.RUnlock() + + userIDs := make([]string, 0) + for userID, conf := range ucc.userConfigMapping { + if conf.AllNewQuestions { + userIDs = append(userIDs, userID) + } + } + return userIDs +} + func (uc *UserCenter) UserConfigFields() []plugin.ConfigField { fields := make([]plugin.ConfigField, 0) // Show tip for user, if the notification service is disabled diff --git a/user-center-slack/user_config_concurrency_test.go b/user-center-slack/user_config_concurrency_test.go new file mode 100644 index 000000000..00c7c1f12 --- /dev/null +++ b/user-center-slack/user_config_concurrency_test.go @@ -0,0 +1,51 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0. + */ + +package slack_user_center + +import ( + "fmt" + "sync" + "testing" +) + +func TestUserConfigCacheConcurrentNewQuestionSubscribers(t *testing.T) { + cache := NewUserConfigCache() + const workers, iterations = 8, 500 + start := make(chan struct{}) + var wg sync.WaitGroup + for worker := range workers { + wg.Add(1) + go func() { + defer wg.Done() + <-start + for i := range iterations { + cache.SetUserConfig(fmt.Sprintf("writer-%d-user-%d", worker, i%32), &UserConfig{AllNewQuestions: i%2 == 0}) + } + }() + } + for range workers { + wg.Add(1) + go func() { + defer wg.Done() + <-start + for range iterations { + _ = cache.NewQuestionSubscriberIDs() + } + }() + } + close(start) + wg.Wait() + + cache.SetUserConfig("subscriber", &UserConfig{AllNewQuestions: true}) + for _, userID := range cache.NewQuestionSubscriberIDs() { + if userID == "subscriber" { + return + } + } + t.Fatal("subscriber missing from snapshot") +} diff --git a/user-center-wecom/notification.go b/user-center-wecom/notification.go index d22f0421a..4ce63c882 100644 --- a/user-center-wecom/notification.go +++ b/user-center-wecom/notification.go @@ -31,12 +31,7 @@ import ( // GetNewQuestionSubscribers returns the subscribers of the new question notification func (uc *UserCenter) GetNewQuestionSubscribers() (userIDs []string) { - for userID, conf := range uc.UserConfigCache.userConfigMapping { - if conf.AllNewQuestions { - userIDs = append(userIDs, userID) - } - } - return userIDs + return uc.UserConfigCache.NewQuestionSubscriberIDs() } // Notify sends a notification to the user diff --git a/user-center-wecom/user_config.go b/user-center-wecom/user_config.go index 2a7cabec2..5f6106129 100644 --- a/user-center-wecom/user_config.go +++ b/user-center-wecom/user_config.go @@ -38,7 +38,7 @@ type UserConfig struct { type UserConfigCache struct { // key: userID value: user config userConfigMapping map[string]*UserConfig - sync.Mutex + sync.RWMutex } func NewUserConfigCache() *UserConfigCache { @@ -54,6 +54,19 @@ func (ucc *UserConfigCache) SetUserConfig(userID string, config *UserConfig) { ucc.userConfigMapping[userID] = config } +func (ucc *UserConfigCache) NewQuestionSubscriberIDs() []string { + ucc.RLock() + defer ucc.RUnlock() + + userIDs := make([]string, 0) + for userID, conf := range ucc.userConfigMapping { + if conf.AllNewQuestions { + userIDs = append(userIDs, userID) + } + } + return userIDs +} + func (uc *UserCenter) UserConfigFields() []plugin.ConfigField { fields := make([]plugin.ConfigField, 0) // Show tip for user, if the notification service is disabled diff --git a/user-center-wecom/user_config_concurrency_test.go b/user-center-wecom/user_config_concurrency_test.go new file mode 100644 index 000000000..8ea44c619 --- /dev/null +++ b/user-center-wecom/user_config_concurrency_test.go @@ -0,0 +1,51 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0. + */ + +package wecom + +import ( + "fmt" + "sync" + "testing" +) + +func TestUserConfigCacheConcurrentNewQuestionSubscribers(t *testing.T) { + cache := NewUserConfigCache() + const workers, iterations = 8, 500 + start := make(chan struct{}) + var wg sync.WaitGroup + for worker := range workers { + wg.Add(1) + go func() { + defer wg.Done() + <-start + for i := range iterations { + cache.SetUserConfig(fmt.Sprintf("writer-%d-user-%d", worker, i%32), &UserConfig{AllNewQuestions: i%2 == 0}) + } + }() + } + for range workers { + wg.Add(1) + go func() { + defer wg.Done() + <-start + for range iterations { + _ = cache.NewQuestionSubscriberIDs() + } + }() + } + close(start) + wg.Wait() + + cache.SetUserConfig("subscriber", &UserConfig{AllNewQuestions: true}) + for _, userID := range cache.NewQuestionSubscriberIDs() { + if userID == "subscriber" { + return + } + } + t.Fatal("subscriber missing from snapshot") +}