whisper: add matching to v6

whisper/whisperv5/filter.go

@@ -26,7 +26,10 @@ import (
 	"github.com/ethereum/go-ethereum/log"
 )
 
-const ALL_TOPICS = ""
+const (
+	ALL_TOPICS = ""
+	MAX_POOL_CAPACITY = 1000
+)
 
 type Filter struct {
 	Src        *ecdsa.PublicKey  // Sender of the message
@@ -43,18 +46,16 @@ type Filter struct {
 
 type Filters struct {
 	watchers       map[string]*Filter
-	watchersTopics map[string]map[string]struct{}
-	topicMatcher   *topicMatcher
 	whisper        *Whisper
 	mutex          sync.RWMutex
+	topicMatcher   *topicMatcher
 }
 
 func NewFilters(w *Whisper) *Filters {
 	fs := &Filters{
 		watchers:       make(map[string]*Filter),
-		watchersTopics: make(map[string]map[string]struct{}),
-		topicMatcher:   newTopicMatcher(),
 		whisper:        w,
+		topicMatcher:   newTopicMatcher(),
 	}
 	return fs
 }
@@ -228,6 +229,9 @@ func IsPubKeyEqual(a, b *ecdsa.PublicKey) bool {
 }
 
 type topicMatcher struct {
+	//structure - map[topic]map[filterID]
+	//mapping for topics
+	//"" topic means that the filter allows all topic values
 	mapper map[string]map[string]struct{}
 	mx     sync.RWMutex
 	pool   sync.Pool
@@ -247,7 +251,7 @@ func (fs *topicMatcher) take() []string {
 	return fs.pool.Get().([]string)
 }
 func (fs *topicMatcher) resolve(s []string) {
-	if cap(s) > 1000 {
+	if cap(s) > MAX_POOL_CAPACITY {
 		return
 	}
 	fs.pool.Put(s[:0])

whisper/whisperv5/filter_test.go

@@ -723,6 +723,7 @@ 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)

whisper/whisperv6/filter.go

@@ -26,6 +26,11 @@ import (
 	"github.com/ethereum/go-ethereum/log"
 )
 
+const (
+	ALL_TOPICS        = ""
+	MAX_POOL_CAPACITY = 1000
+)
+
 // Filter represents a Whisper message filter
 type Filter struct {
 	Src        *ecdsa.PublicKey  // Sender of the message
@@ -45,6 +50,7 @@ type Filters struct {
 	watchers     map[string]*Filter
 	whisper      *Whisper
 	mutex        sync.RWMutex
+	topicMatcher *topicMatcher
 }
 
 // NewFilters returns a newly created filter collection
@@ -52,6 +58,7 @@ func NewFilters(w *Whisper) *Filters {
 	return &Filters{
 		watchers:     make(map[string]*Filter),
 		whisper:      w,
+		topicMatcher: newTopicMatcher(),
 	}
 }
 
@@ -82,6 +89,7 @@ func (fs *Filters) Install(watcher *Filter) (string, error) {
 	}
 
 	fs.watchers[id] = watcher
+	fs.topicMatcher.addFilterToTopicsMapping(watcher, id)
 	return id, err
 }
 
@@ -92,6 +100,7 @@ func (fs *Filters) Uninstall(id string) bool {
 	defer fs.mutex.Unlock()
 	if fs.watchers[id] != nil {
 		delete(fs.watchers, id)
+		fs.topicMatcher.removeTopicFromTopicMapping(id)
 		return true
 	}
 	return false
@@ -108,15 +117,22 @@ func (fs *Filters) Get(id string) *Filter {
 // for the envelope's topic.
 func (fs *Filters) NotifyWatchers(env *Envelope, p2pMessage bool) {
 	var msg *ReceivedMessage
+	matchedTopics := fs.topicMatcher.take()
+	defer fs.topicMatcher.resolve(matchedTopics)
 
 	fs.mutex.RLock()
 	defer fs.mutex.RUnlock()
 
-	i := -1 // only used for logging info
+	fs.topicMatcher.matchedTopics(env.Topic, &matchedTopics)
-	for _, watcher := range fs.watchers {
+	for _, watcherID := range matchedTopics {
-		i++
+		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
 		}
 
@@ -128,10 +144,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)
 			}
 		}
 
@@ -201,9 +217,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
 }
@@ -216,38 +232,6 @@ func (f *Filter) MatchEnvelope(envelope *Envelope) bool {
 		return false
 	}
 
-	return f.MatchTopic(envelope.Topic)
-}
-
-// MatchTopic checks that the filter captures a given topic.
-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
 }
 
@@ -261,3 +245,82 @@ 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
 }
+
+type topicMatcher struct {
+	//structure - map[topic]map[filterID]
+	//mapping for topics
+	//"" topic means that the filter allows all topic values
+	mapper map[string]map[string]struct{}
+	mx     sync.RWMutex
+	pool   sync.Pool
+}
+
+func newTopicMatcher() *topicMatcher {
+	tm := new(topicMatcher)
+	tm.mapper = make(map[string]map[string]struct{})
+	tm.mapper[ALL_TOPICS] = make(map[string]struct{})
+	tm.pool.New = func() interface{} {
+		return []string{}
+	}
+	return tm
+}
+
+func (fs *topicMatcher) take() []string {
+	return fs.pool.Get().([]string)
+}
+func (fs *topicMatcher) resolve(s []string) {
+	if cap(s) > MAX_POOL_CAPACITY {
+		return
+	}
+	fs.pool.Put(s[:0])
+}
+
+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{} {
+	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, matched *[]string) {
+	fs.mx.RLock()
+	defer fs.mx.RUnlock()
+
+	for i := range fs.mapper[ALL_TOPICS] {
+		*matched = append(*matched, i)
+	}
+
+	for i := range fs.mapper[topic.String()] {
+		*matched = append(*matched, i)
+	}
+}

whisper/whisperv6/filter_test.go

@@ -83,6 +83,14 @@ func generateFilter(t *testing.T, symmetric bool) (*Filter, error) {
 	return &f, nil
 }
 
+func generateFilters() *Filters {
+	fs := Filters{
+		watchers:     make(map[string]*Filter),
+		topicMatcher: newTopicMatcher(),
+	}
+	return &fs
+}
+
 func generateTestCases(t *testing.T, SizeTestFilters int) []FilterTestCase {
 	cases := make([]FilterTestCase, SizeTestFilters)
 	for i := 0; i < SizeTestFilters; i++ {
@@ -314,19 +322,8 @@ 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)
@@ -336,20 +333,20 @@ func TestMatchEnvelope(t *testing.T) {
 		t.Fatalf("failed Wrap() with seed %d: %s.", seed, err)
 	}
 
-	// symmetric + matching topic: match
+	// symmetric
-	match = fsym.MatchEnvelope(env)
+	match := fsym.MatchEnvelope(env)
 	if !match {
 		t.Fatalf("failed MatchEnvelope() symmetric with seed %d.", seed)
 	}
 
-	// symmetric + matching topic + insufficient PoW: mismatch
+	// symmetric + insufficient PoW: mismatch
 	fsym.PoW = env.PoW() + 1.0
 	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 {
@@ -387,26 +384,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)
@@ -420,20 +397,6 @@ func TestMatchEnvelope(t *testing.T) {
 	if !match {
 		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) {
@@ -485,13 +448,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) {
@@ -578,13 +534,6 @@ func TestMatchMessageAsym(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
 	prev := *f.KeyAsym.PublicKey.X
 	zero := *big.NewInt(0)
@@ -820,53 +769,150 @@ func TestVariableTopics(t *testing.T) {
 		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
+		matched := []string{}
+		fs.topicMatcher.matchedTopics(env.Topic, &matched)
 		match = f.MatchEnvelope(env)
-		if !match {
+		if !(match && hasFilterID(matched, filterID)) {
 			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 {
+
+		matched = matched[:0]
+		fs.topicMatcher.matchedTopics(env.Topic, &matched)
+		if match && hasFilterID(matched, filterID) {
 			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)
+		}
+		matched := []string{}
+		fs.topicMatcher.matchedTopics(env.Topic, &matched)
+		if !hasFilterID(matched, filterID) {
+			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")
+		}
+		matched = matched[:0]
+		fs.topicMatcher.matchedTopics(env.Topic, &matched)
+		if hasFilterID(matched, filterID) {
+			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)
+		}
+		matched = matched[:0]
+		fs.topicMatcher.matchedTopics(env.Topic, &matched)
+		if hasFilterID(matched, filterID) {
+			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()
+
+	f.Topics = [][]byte{}
+
+	filterID, err := fs.Install(f)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	//generate topic
+	topic := TopicType{}
+	mrand.Read(topic[:])
+
+	matched := []string{}
+	fs.topicMatcher.matchedTopics(topic, &matched)
+	if !hasFilterID(matched, filterID) {
+		t.Fatalf("failed MatchEnvelope symmetric with seed %d, step %d.", seed)
+	}
+	if _, ok := fs.topicMatcher.mapper[ALL_TOPICS][filterID]; !ok {
+		t.Fatal("watcher mapping incorrect")
+	}
+
+	//test match without filter
+	if !fs.Uninstall(filterID) {
+		t.Fatal("Failed to uninstall filter")
+	}
+	matched = matched[:0]
+	fs.topicMatcher.matchedTopics(topic, &matched)
+	if hasFilterID(matched, filterID) {
+		t.Fatalf("failed MatchEnvelope symmetric with seed %d, step %d.", seed)
+	}
+
+	if _, ok := fs.topicMatcher.mapper[ALL_TOPICS][filterID]; ok {
+		t.Fatal("watcher mapping incorrect")
 	}
 }
 
-func TestMatchSingleTopic_NotEquals_ReturnFalse(t *testing.T) {
+func hasFilterID(matched []string, filterID string) bool {
-	bt := []byte("tes")
+	for i := range matched {
-	topic := BytesToTopic(bt)
+		if matched[i] == filterID {
-
+			return true
-	if matchSingleTopic(topic, bt) {
-		t.FailNow()
 		}
 	}
-
+	return false
-func TestMatchSingleTopic_InsufficientLength_ReturnFalse(t *testing.T) {
-	bt := []byte("test")
-	topic := BytesToTopic([]byte("not_equal"))
-
-	if matchSingleTopic(topic, bt) {
-		t.FailNow()
-	}
 }