whisper: add mutex to mapper

whisper/whisperv5/filter.go

@@ -44,6 +44,7 @@ type Filter struct {
 type Filters struct {
 	watchers       map[string]*Filter
 	watchersTopics map[string]map[string]struct{}
+	topicMatcher   *topicMatcher
 	whisper        *Whisper
 	mutex          sync.RWMutex
 }
@@ -52,9 +53,9 @@ func NewFilters(w *Whisper) *Filters {
 	fs := &Filters{
 		watchers:       make(map[string]*Filter),
 		watchersTopics: make(map[string]map[string]struct{}),
+		topicMatcher:   newTopicMatcher(),
 		whisper:        w,
 	}
-	fs.watchersTopics[ALL_TOPICS] = make(map[string]struct{})
 	return fs
 }
 
@@ -80,56 +81,16 @@ func (fs *Filters) Install(watcher *Filter) (string, error) {
 	}
 
 	fs.watchers[id] = watcher
-	fs.addFilterToTopicsMapping(watcher, id)
+	fs.topicMatcher.addFilterToTopicsMapping(watcher, id)
 	return id, err
 }
 
-func (fs *Filters) addFilterToTopicsMapping(watcher *Filter, id string) {
-	for i := range fs.prepareTopicsMapping(watcher) {
-		topicMapping, ok := fs.watchersTopics[i]
-		if !ok {
-			fs.watchersTopics[i] = make(map[string]struct{})
-			topicMapping = fs.watchersTopics[i]
-		}
-		topicMapping[id] = struct{}{}
-	}
-}
-
-func (fs *Filters) removeTopicFromTopicMapping(id string) {
-	for i := range fs.watchersTopics {
-		delete(fs.watchersTopics[i], id)
-	}
-}
-
-func (fs *Filters) prepareTopicsMapping(watcher *Filter) map[string]struct{} {
-	topics := make(map[string]struct{}, len(watcher.Topics))
-
-	if len(watcher.Topics) == 0 {
-		topics[ALL_TOPICS] = struct{}{}
-		return topics
-	}
-
-	for _, topic := range watcher.Topics {
-		topics[common.ToHex(topic)] = struct{}{}
-	}
-
-	return topics
-}
-
-func (fs *Filters) matchedTopics(topic TopicType) map[string]struct{} {
-	m := fs.watchersTopics[ALL_TOPICS]
-	for i := range fs.watchersTopics[topic.String()] {
-		m[i] = struct{}{}
-	}
-	return m
-}
-
 func (fs *Filters) Uninstall(id string) bool {
 	fs.mutex.Lock()
 	defer fs.mutex.Unlock()
 	if fs.watchers[id] != nil {
 		delete(fs.watchers, id)
-		fs.removeTopicFromTopicMapping(id)
+		fs.topicMatcher.removeTopicFromTopicMapping(id)
 		return true
 	}
 	return false
@@ -147,7 +108,7 @@ func (fs *Filters) NotifyWatchers(env *Envelope, p2pMessage bool) {
 	fs.mutex.RLock()
 	defer fs.mutex.RUnlock()
 
-	for watcherID := range fs.matchedTopics(env.Topic) {
+	for watcherID := range fs.topicMatcher.matchedTopics(env.Topic) {
 		watcher, ok := fs.watchers[watcherID]
 		if !ok {
 			log.Trace(fmt.Sprintf("msg [%x], filter [%s]: filter not exists", env.Hash(), watcherID))
@@ -262,3 +223,70 @@ func IsPubKeyEqual(a, b *ecdsa.PublicKey) bool {
 	// the curve is always the same, just compare the points
 	return a.X.Cmp(b.X) == 0 && a.Y.Cmp(b.Y) == 0
 }
+
+func newTopicMatcher() *topicMatcher {
+	tm := new(topicMatcher)
+	tm.mapper = make(map[string]map[string]struct{})
+	tm.mapper[ALL_TOPICS] = make(map[string]struct{})
+	return tm
+}
+
+type topicMatcher struct {
+	mapper map[string]map[string]struct{}
+	mx     sync.RWMutex
+}
+
+func (fs *topicMatcher) addFilterToTopicsMapping(watcher *Filter, id string) {
+	fs.mx.Lock()
+	defer fs.mx.Unlock()
+
+	for i := range fs.prepareTopicsMapping(watcher) {
+		topicMapping, ok := fs.mapper[i]
+		if !ok {
+			fs.mapper[i] = make(map[string]struct{})
+			topicMapping = fs.mapper[i]
+		}
+		topicMapping[id] = struct{}{}
+	}
+}
+
+func (fs *topicMatcher) removeTopicFromTopicMapping(id string) {
+	fs.mx.Lock()
+	defer fs.mx.Unlock()
+	for i := range fs.mapper {
+		delete(fs.mapper[i], id)
+	}
+}
+
+func (fs *topicMatcher) prepareTopicsMapping(watcher *Filter) map[string]struct{} {
+	fs.mx.RLock()
+	defer fs.mx.RUnlock()
+	topics := make(map[string]struct{}, len(watcher.Topics))
+
+	if len(watcher.Topics) == 0 {
+		topics[ALL_TOPICS] = struct{}{}
+		return topics
+	}
+
+	for _, topic := range watcher.Topics {
+		topics[common.ToHex(topic)] = struct{}{}
+	}
+
+	return topics
+}
+
+func (fs *topicMatcher) matchedTopics(topic TopicType) map[string]struct{} {
+	fs.mx.RLock()
+	defer fs.mx.RUnlock()
+
+	m := make(map[string]struct{}, len(fs.mapper[ALL_TOPICS])+len(fs.mapper[topic.String()]))
+
+	for i := range fs.mapper[ALL_TOPICS] {
+		m[i] = struct{}{}
+	}
+
+	for i := range fs.mapper[topic.String()] {
+		m[i] = struct{}{}
+	}
+	return m
+}

whisper/whisperv5/filter_test.go

@@ -86,9 +86,8 @@ func generateFilter(t *testing.T, symmetric bool) (*Filter, error) {
 func generateFilters() *Filters {
 	fs := Filters{
 		watchers:     make(map[string]*Filter),
-		watchersTopics: make(map[string]map[string]struct{}),
+		topicMatcher: newTopicMatcher(),
 	}
-	fs.watchersTopics[ALL_TOPICS] = make(map[string]struct{})
 	return &fs
 }
 
@@ -496,7 +495,7 @@ func TestMatchMessageAsym(t *testing.T) {
 		t.Fatalf("failed filter install with seed %d: %s.", seed, err)
 	}
 
-	m := fs.matchedTopics(env.Topic)
+	m := fs.topicMatcher.matchedTopics(env.Topic)
 	_, matchedTopic := m[filterID]
 
 	if !matchedTopic {
@@ -509,7 +508,7 @@ func TestMatchMessageAsym(t *testing.T) {
 	}
 	f.Topics[index][0]++
 	filterID, err = fs.Install(f)
-	m = fs.matchedTopics(env.Topic)
+	m = fs.topicMatcher.matchedTopics(env.Topic)
 	_, matchedTopic = m[filterID]
 
 	if matchedTopic {
@@ -762,7 +761,7 @@ func TestVariableTopics(t *testing.T) {
 		env.Topic = BytesToTopic(f.Topics[i])
 
 		//test match
-		m := fs.matchedTopics(env.Topic)
+		m := fs.topicMatcher.matchedTopics(env.Topic)
 		_, ok := m[filterID]
 		match = f.MatchEnvelope(env)
 		if !(match && ok) {
@@ -774,9 +773,9 @@ func TestVariableTopics(t *testing.T) {
 
 		//false positive test
 		match = f.MatchEnvelope(env)
-		m = fs.matchedTopics(env.Topic)
+		m = fs.topicMatcher.matchedTopics(env.Topic)
 		_, ok = m[filterID]
-		if !(match && ok) {
+		if match && ok {
 			t.Fatalf("MatchEnvelope symmetric with seed %d, step %d: false positive.", seed, i)
 		}
 	}
@@ -813,7 +812,7 @@ func TestTopicsMapping(t *testing.T) {
 		if err != nil {
 			t.Fatal(err)
 		}
-		m := fs.matchedTopics(env.Topic)
+		m := fs.topicMatcher.matchedTopics(env.Topic)
 		if _, matchTopic := m[filterID]; !matchTopic {
 			t.Fatalf("failed MatchEnvelope symmetric with seed %d, step %d.", seed, i)
 		}
@@ -822,7 +821,7 @@ func TestTopicsMapping(t *testing.T) {
 		if !fs.Uninstall(filterID) {
 			t.Fatal("Failed to uninstall filter")
 		}
-		m = fs.matchedTopics(env.Topic)
+		m = fs.topicMatcher.matchedTopics(env.Topic)
 		if _, matchTopic := m[filterID]; matchTopic {
 			t.Fatalf("failed MatchEnvelope symmetric with seed %d, step %d.", seed, i)
 		}
@@ -833,7 +832,7 @@ func TestTopicsMapping(t *testing.T) {
 		if err != nil {
 			t.Fatal(err)
 		}
-		m = fs.matchedTopics(env.Topic)
+		m = fs.topicMatcher.matchedTopics(env.Topic)
 		if _, matchTopic := m[filterID]; matchTopic {
 			t.Fatalf("failed MatchEnvelope symmetric with seed %d, step %d.", seed, i)
 		}
@@ -863,25 +862,24 @@ func TestTopicsMapping_MatchAllTopics_Success(t *testing.T) {
 	topic := TopicType{}
 	mrand.Read(topic[:])
 
-	m := fs.matchedTopics(topic)
+	m := fs.topicMatcher.matchedTopics(topic)
 	if _, matchTopic := m[filterID]; !matchTopic {
 		t.Fatalf("failed MatchEnvelope symmetric with seed %d, step %d.", seed)
 	}
-	if _, ok := fs.watchersTopics[ALL_TOPICS][filterID]; !ok {
+	if _, ok := fs.topicMatcher.mapper[ALL_TOPICS][filterID]; !ok {
 		t.Fatal("watcher mapping incorrect")
 	}
 
-	////test match without filter
+	//test match without filter
 	if !fs.Uninstall(filterID) {
 		t.Fatal("Failed to uninstall filter")
 	}
-	m = fs.matchedTopics(topic)
+	m = fs.topicMatcher.matchedTopics(topic)
 	if _, matchTopic := m[filterID]; matchTopic {
 		t.Fatalf("failed MatchEnvelope symmetric with seed %d, step %d.", seed)
 	}
 
-	if _, ok := fs.watchersTopics[ALL_TOPICS][filterID]; ok {
+	if _, ok := fs.topicMatcher.mapper[ALL_TOPICS][filterID]; ok {
 		t.Fatal("watcher mapping incorrect")
 	}
-
 }