whisper: bottleneck on matching messages

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
+	watchers       map[string]*Filter
-	whisper  *Whisper
+	watchersTopics map[string]map[string]struct{}
-	mutex    sync.RWMutex
+	whisper        *Whisper
+	mutex          sync.RWMutex
 }
 
 func NewFilters(w *Whisper) *Filters {
-	return &Filters{
+	fs := &Filters{
-		watchers: make(map[string]*Filter),
+		watchers:       make(map[string]*Filter),
-		whisper:  w,
+		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 watcherID := range fs.matchedTopics(env.Topic) {
-	for _, watcher := range fs.watchers {
+		watcher, ok := fs.watchers[watcherID]
-		i++
+		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

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
+	// symmetric +  insufficient PoW: mismatch
-	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
 	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) {
+func TestTopicsMapping(t *testing.T) {
-	bt := []byte("test")
+	InitSingleTest()
-	topic := BytesToTopic(bt)
 
-	if !matchSingleTopic(topic, bt) {
+	const lastTopicByte = 3
-		t.FailNow()
+	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) {
+func TestTopicsMapping_MatchAllTopics_Success(t *testing.T) {
-	bt := []byte("test with tail")
+	InitSingleTest()
-	topic := BytesToTopic([]byte("test"))
 
-	if !matchSingleTopic(topic, bt) {
+	f, err := generateFilter(t, true)
-		t.FailNow()
+	if err != nil {
+		t.Fatalf("failed generateFilter with seed %d: %s.", seed, err)
 	}
-}
+	fs := generateFilters()
 
-func TestMatchSingleTopic_NotEquals_ReturnFalse(t *testing.T) {
+	f.Topics = [][]byte{}
-	bt := []byte("tes")
-	topic := BytesToTopic(bt)
 
-	if matchSingleTopic(topic, bt) {
+	filterID, err := fs.Install(f)
-		t.FailNow()
+	if err != nil {
+		t.Fatal(err)
 	}
-}
 
-func TestMatchSingleTopic_InsufficientLength_ReturnFalse(t *testing.T) {
+	//generate topic
-	bt := []byte("test")
+	topic := TopicType{}
-	topic := BytesToTopic([]byte("not_equal"))
+	mrand.Read(topic[:])
 
-	if matchSingleTopic(topic, bt) {
+	m := fs.matchedTopics(topic)
-		t.FailNow()
+	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")
+	}
+
 }