diff --git a/whisper/whisperv5/filter.go b/whisper/whisperv5/filter.go index 920bcf20b8..7e6c87a47c 100644 --- a/whisper/whisperv5/filter.go +++ b/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 +} diff --git a/whisper/whisperv5/filter_test.go b/whisper/whisperv5/filter_test.go index f50c970f5f..e0db720dbb 100644 --- a/whisper/whisperv5/filter_test.go +++ b/whisper/whisperv5/filter_test.go @@ -85,10 +85,9 @@ 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{}), + watchers: make(map[string]*Filter), + 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") } - }