From 03116ba7b092d9658b42c87764ade010d425d2ef Mon Sep 17 00:00:00 2001 From: b00ris Date: Wed, 21 Feb 2018 11:55:17 +0300 Subject: [PATCH] whisper: bottleneck on matching messages --- whisper/whisperv5/filter.go | 113 ++++++++------ whisper/whisperv5/filter_test.go | 243 ++++++++++++++++++------------- 2 files changed, 208 insertions(+), 148 deletions(-) diff --git a/whisper/whisperv5/filter.go b/whisper/whisperv5/filter.go index 3190334ebb..920bcf20b8 100644 --- a/whisper/whisperv5/filter.go +++ b/whisper/whisperv5/filter.go @@ -26,6 +26,8 @@ import ( "github.com/ethereum/go-ethereum/log" ) +const ALL_TOPICS = "" + type Filter struct { Src *ecdsa.PublicKey // Sender of the message KeyAsym *ecdsa.PrivateKey // Private Key of recipient @@ -40,16 +42,20 @@ type Filter struct { } type Filters struct { - watchers map[string]*Filter - whisper *Whisper - mutex sync.RWMutex + watchers map[string]*Filter + watchersTopics map[string]map[string]struct{} + whisper *Whisper + mutex sync.RWMutex } func NewFilters(w *Whisper) *Filters { - return &Filters{ - watchers: make(map[string]*Filter), - whisper: w, + fs := &Filters{ + watchers: make(map[string]*Filter), + watchersTopics: make(map[string]map[string]struct{}), + whisper: w, } + fs.watchersTopics[ALL_TOPICS] = make(map[string]struct{}) + return fs } func (fs *Filters) Install(watcher *Filter) (string, error) { @@ -74,14 +80,56 @@ func (fs *Filters) Install(watcher *Filter) (string, error) { } fs.watchers[id] = watcher + fs.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) return true } return false @@ -99,11 +147,15 @@ func (fs *Filters) NotifyWatchers(env *Envelope, p2pMessage bool) { fs.mutex.RLock() defer fs.mutex.RUnlock() - i := -1 // only used for logging info - for _, watcher := range fs.watchers { - i++ + for watcherID := range fs.matchedTopics(env.Topic) { + watcher, ok := fs.watchers[watcherID] + if !ok { + log.Trace(fmt.Sprintf("msg [%x], filter [%s]: filter not exists", env.Hash(), watcherID)) + continue + } + if p2pMessage && !watcher.AllowP2P { - log.Trace(fmt.Sprintf("msg [%x], filter [%d]: p2p messages are not allowed", env.Hash(), i)) + log.Trace(fmt.Sprintf("msg [%x], filter [%s]: p2p messages are not allowed", env.Hash(), watcherID)) continue } @@ -115,10 +167,10 @@ func (fs *Filters) NotifyWatchers(env *Envelope, p2pMessage bool) { if match { msg = env.Open(watcher) if msg == nil { - log.Trace("processing message: failed to open", "message", env.Hash().Hex(), "filter", i) + log.Trace("processing message: failed to open", "message", env.Hash().Hex(), "filter", watcherID) } } else { - log.Trace("processing message: does not match", "message", env.Hash().Hex(), "filter", i) + log.Trace("processing message: does not match", "message", env.Hash().Hex(), "filter", watcherID) } } @@ -181,9 +233,9 @@ func (f *Filter) MatchMessage(msg *ReceivedMessage) bool { } if f.expectsAsymmetricEncryption() && msg.isAsymmetricEncryption() { - return IsPubKeyEqual(&f.KeyAsym.PublicKey, msg.Dst) && f.MatchTopic(msg.Topic) + return IsPubKeyEqual(&f.KeyAsym.PublicKey, msg.Dst) } else if f.expectsSymmetricEncryption() && msg.isSymmetricEncryption() { - return f.SymKeyHash == msg.SymKeyHash && f.MatchTopic(msg.Topic) + return f.SymKeyHash == msg.SymKeyHash } return false } @@ -194,44 +246,13 @@ func (f *Filter) MatchEnvelope(envelope *Envelope) bool { } if f.expectsAsymmetricEncryption() && envelope.isAsymmetric() { - return f.MatchTopic(envelope.Topic) + return true } else if f.expectsSymmetricEncryption() && envelope.IsSymmetric() { - return f.MatchTopic(envelope.Topic) - } - return false -} - -func (f *Filter) MatchTopic(topic TopicType) bool { - if len(f.Topics) == 0 { - // any topic matches return true } - - for _, bt := range f.Topics { - if matchSingleTopic(topic, bt) { - return true - } - } return false } -func matchSingleTopic(topic TopicType, bt []byte) bool { - if len(bt) > TopicLength { - bt = bt[:TopicLength] - } - - if len(bt) < TopicLength { - return false - } - - for j, b := range bt { - if topic[j] != b { - return false - } - } - return true -} - func IsPubKeyEqual(a, b *ecdsa.PublicKey) bool { if !ValidatePublicKey(a) { return false diff --git a/whisper/whisperv5/filter_test.go b/whisper/whisperv5/filter_test.go index 01034a3513..f50c970f5f 100644 --- a/whisper/whisperv5/filter_test.go +++ b/whisper/whisperv5/filter_test.go @@ -83,6 +83,15 @@ func generateFilter(t *testing.T, symmetric bool) (*Filter, error) { return &f, nil } +func generateFilters() *Filters { + fs := Filters{ + watchers: make(map[string]*Filter), + watchersTopics: make(map[string]map[string]struct{}), + } + fs.watchersTopics[ALL_TOPICS] = make(map[string]struct{}) + return &fs +} + func generateTestCases(t *testing.T, SizeTestFilters int) []FilterTestCase { cases := make([]FilterTestCase, SizeTestFilters) for i := 0; i < SizeTestFilters; i++ { @@ -284,19 +293,7 @@ func TestMatchEnvelope(t *testing.T) { if err != nil { t.Fatalf("failed Wrap with seed %d: %s.", seed, err) } - match := fsym.MatchEnvelope(env) - if match { - t.Fatalf("failed MatchEnvelope symmetric with seed %d.", seed) - } - match = fasym.MatchEnvelope(env) - if match { - t.Fatalf("failed MatchEnvelope asymmetric with seed %d.", seed) - } - // encrypt symmetrically - i := mrand.Int() % 4 - fsym.Topics[i] = params.Topic[:] - fasym.Topics[i] = params.Topic[:] msg, err = NewSentMessage(params) if err != nil { t.Fatalf("failed to create new message with seed %d: %s.", seed, err) @@ -306,41 +303,20 @@ func TestMatchEnvelope(t *testing.T) { t.Fatalf("failed Wrap() with seed %d: %s.", seed, err) } - // symmetric + matching topic: match - match = fsym.MatchEnvelope(env) - if !match { - t.Fatalf("failed MatchEnvelope() symmetric with seed %d.", seed) - } - - // asymmetric + matching topic: mismatch - match = fasym.MatchEnvelope(env) - if match { - t.Fatalf("failed MatchEnvelope() asymmetric with seed %d.", seed) - } - - // symmetric + matching topic + insufficient PoW: mismatch + // symmetric + insufficient PoW: mismatch fsym.PoW = env.PoW() + 1.0 - match = fsym.MatchEnvelope(env) + match := fsym.MatchEnvelope(env) if match { t.Fatalf("failed MatchEnvelope(symmetric + matching topic + insufficient PoW) asymmetric with seed %d.", seed) } - // symmetric + matching topic + sufficient PoW: match + // symmetric + sufficient PoW: match fsym.PoW = env.PoW() / 2 match = fsym.MatchEnvelope(env) if !match { t.Fatalf("failed MatchEnvelope(symmetric + matching topic + sufficient PoW) with seed %d.", seed) } - // symmetric + topics are nil (wildcard): match - prevTopics := fsym.Topics - fsym.Topics = nil - match = fsym.MatchEnvelope(env) - if !match { - t.Fatalf("failed MatchEnvelope(symmetric + topics are nil) with seed %d.", seed) - } - fsym.Topics = prevTopics - // encrypt asymmetrically key, err := crypto.GenerateKey() if err != nil { @@ -363,26 +339,6 @@ func TestMatchEnvelope(t *testing.T) { t.Fatalf("failed MatchEnvelope(encryption method mismatch) with seed %d.", seed) } - // asymmetric + mismatching topic: mismatch - match = fasym.MatchEnvelope(env) - if !match { - t.Fatalf("failed MatchEnvelope(asymmetric + mismatching topic) with seed %d.", seed) - } - - // asymmetric + matching topic: match - fasym.Topics[i] = fasym.Topics[i+1] - match = fasym.MatchEnvelope(env) - if match { - t.Fatalf("failed MatchEnvelope(asymmetric + matching topic) with seed %d.", seed) - } - - // asymmetric + filter without topic (wildcard): match - fasym.Topics = nil - match = fasym.MatchEnvelope(env) - if !match { - t.Fatalf("failed MatchEnvelope(asymmetric + filter without topic) with seed %d.", seed) - } - // asymmetric + insufficient PoW: mismatch fasym.PoW = env.PoW() + 1.0 match = fasym.MatchEnvelope(env) @@ -397,19 +353,6 @@ func TestMatchEnvelope(t *testing.T) { t.Fatalf("failed MatchEnvelope(asymmetric + sufficient PoW) with seed %d.", seed) } - // filter without topic + envelope without topic: match - env.Topic = TopicType{} - match = fasym.MatchEnvelope(env) - if !match { - t.Fatalf("failed MatchEnvelope(filter without topic + envelope without topic) with seed %d.", seed) - } - - // filter with topic + envelope without topic: mismatch - fasym.Topics = fsym.Topics - match = fasym.MatchEnvelope(env) - if match { - t.Fatalf("failed MatchEnvelope(filter without topic + envelope without topic) with seed %d.", seed) - } } func TestMatchMessageSym(t *testing.T) { @@ -461,13 +404,6 @@ func TestMatchMessageSym(t *testing.T) { t.Fatalf("failed MatchEnvelope(sufficient PoW) with seed %d.", seed) } - // topic mismatch - f.Topics[index][0]++ - if f.MatchMessage(msg) { - t.Fatalf("failed MatchEnvelope(topic mismatch) with seed %d.", seed) - } - f.Topics[index][0]-- - // key mismatch f.SymKeyHash[0]++ if f.MatchMessage(msg) { @@ -554,9 +490,29 @@ func TestMatchMessageAsym(t *testing.T) { t.Fatalf("failed MatchEnvelope(sufficient PoW) with seed %d.", seed) } + fs := generateFilters() + filterID, err := fs.Install(f) + if err != nil { + t.Fatalf("failed filter install with seed %d: %s.", seed, err) + } + + m := fs.matchedTopics(env.Topic) + _, matchedTopic := m[filterID] + + if !matchedTopic { + t.Fatalf("failed MatchEnvelope(topic mismatch) with seed %d.", seed) + } + // topic mismatch + if !fs.Uninstall(filterID) { + t.Fatal("failed to uninstall filter") + } f.Topics[index][0]++ - if f.MatchMessage(msg) { + filterID, err = fs.Install(f) + m = fs.matchedTopics(env.Topic) + _, matchedTopic = m[filterID] + + if matchedTopic { t.Fatalf("failed MatchEnvelope(topic mismatch) with seed %d.", seed) } f.Topics[index][0]-- @@ -795,54 +751,137 @@ func TestVariableTopics(t *testing.T) { if err != nil { t.Fatalf("failed generateFilter with seed %d: %s.", seed, err) } + fs := generateFilters() + + filterID, err := fs.Install(f) + if err != nil { + t.Fatal(err) + } for i := 0; i < 4; i++ { env.Topic = BytesToTopic(f.Topics[i]) + + //test match + m := fs.matchedTopics(env.Topic) + _, ok := m[filterID] match = f.MatchEnvelope(env) - if !match { + if !(match && ok) { t.Fatalf("failed MatchEnvelope symmetric with seed %d, step %d.", seed, i) } - f.Topics[i][lastTopicByte]++ + //change envelop topic + env.Topic[lastTopicByte]++ + + //false positive test match = f.MatchEnvelope(env) - if match { + m = fs.matchedTopics(env.Topic) + _, ok = m[filterID] + if !(match && ok) { t.Fatalf("MatchEnvelope symmetric with seed %d, step %d: false positive.", seed, i) } } } -func TestMatchSingleTopic_ReturnTrue(t *testing.T) { - bt := []byte("test") - topic := BytesToTopic(bt) +func TestTopicsMapping(t *testing.T) { + InitSingleTest() - if !matchSingleTopic(topic, bt) { - t.FailNow() + const lastTopicByte = 3 + params, err := generateMessageParams() + if err != nil { + t.Fatalf("failed generateMessageParams with seed %d: %s.", seed, err) + } + msg, err := NewSentMessage(params) + if err != nil { + t.Fatalf("failed to create new message with seed %d: %s.", seed, err) + } + env, err := msg.Wrap(params) + if err != nil { + t.Fatalf("failed Wrap with seed %d: %s.", seed, err) + } + + f, err := generateFilter(t, true) + if err != nil { + t.Fatalf("failed generateFilter with seed %d: %s.", seed, err) + } + fs := generateFilters() + + for i := 0; i < len(f.Topics); i++ { + env.Topic = BytesToTopic(f.Topics[i]) + + //test match + filterID, err := fs.Install(f) + if err != nil { + t.Fatal(err) + } + m := fs.matchedTopics(env.Topic) + if _, matchTopic := m[filterID]; !matchTopic { + t.Fatalf("failed MatchEnvelope symmetric with seed %d, step %d.", seed, i) + } + + //test match without filter + if !fs.Uninstall(filterID) { + t.Fatal("Failed to uninstall filter") + } + m = fs.matchedTopics(env.Topic) + if _, matchTopic := m[filterID]; matchTopic { + t.Fatalf("failed MatchEnvelope symmetric with seed %d, step %d.", seed, i) + } + + //test match with changed topic + f.Topics[i][lastTopicByte]++ + filterID, err = fs.Install(f) + if err != nil { + t.Fatal(err) + } + m = fs.matchedTopics(env.Topic) + if _, matchTopic := m[filterID]; matchTopic { + t.Fatalf("failed MatchEnvelope symmetric with seed %d, step %d.", seed, i) + } + if !fs.Uninstall(filterID) { + t.Fatal("Failed to uninstall filter") + } } } -func TestMatchSingleTopic_WithTail_ReturnTrue(t *testing.T) { - bt := []byte("test with tail") - topic := BytesToTopic([]byte("test")) +func TestTopicsMapping_MatchAllTopics_Success(t *testing.T) { + InitSingleTest() - if !matchSingleTopic(topic, bt) { - t.FailNow() + f, err := generateFilter(t, true) + if err != nil { + t.Fatalf("failed generateFilter with seed %d: %s.", seed, err) } -} + fs := generateFilters() -func TestMatchSingleTopic_NotEquals_ReturnFalse(t *testing.T) { - bt := []byte("tes") - topic := BytesToTopic(bt) + f.Topics = [][]byte{} - if matchSingleTopic(topic, bt) { - t.FailNow() + filterID, err := fs.Install(f) + if err != nil { + t.Fatal(err) } -} -func TestMatchSingleTopic_InsufficientLength_ReturnFalse(t *testing.T) { - bt := []byte("test") - topic := BytesToTopic([]byte("not_equal")) + //generate topic + topic := TopicType{} + mrand.Read(topic[:]) - if matchSingleTopic(topic, bt) { - t.FailNow() + m := fs.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 { + t.Fatal("watcher mapping incorrect") + } + + ////test match without filter + if !fs.Uninstall(filterID) { + t.Fatal("Failed to uninstall filter") + } + m = fs.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 { + t.Fatal("watcher mapping incorrect") + } + }