Repository: nats-io/nats-server
Pull request #4864 (2023-12-08 09:56:27)

Do more efficient matching of filter subjects in consumer
Avoid tokenizing the message subject again for every filter subject that it is matched against. Also keep the tokenized version of the filter subjects in the subjectFilter struct.

Signed-off-by: Sven Neumann <sven.neumann@holoplot.com>


svenfoo: Benchmark (from the first commit) before doing the actual changes:
```
goos: linux
goarch: amd64
pkg: github.com/nats-io/nats-server/v2/server
cpu: 11th Gen Intel(R) Core(TM) i7-1185G7 @ 3.00GHz
Benchmark____________ConsumerIsFilteredMatch/1_filter_subjects-8         	1000000000	         0.0000020 ns/op	       0 B/op	       0 allocs/op
Benchmark____________ConsumerIsFilteredMatch/2_filter_subjects-8         	1000000000	         0.0000027 ns/op	       0 B/op	       0 allocs/op
Benchmark____________ConsumerIsFilteredMatch/4_filter_subjects-8         	1000000000	         0.0000042 ns/op	       0 B/op	       0 allocs/op
Benchmark____________ConsumerIsFilteredMatch/8_filter_subjects-8         	1000000000	         0.0000049 ns/op	       0 B/op	       0 allocs/op
Benchmark____________ConsumerIsFilteredMatch/16_filter_subjects-8        	1000000000	         0.0000043 ns/op	       0 B/op	       0 allocs/op
Benchmark____________ConsumerIsFilteredMatch/32_filter_subjects-8        	1000000000	         0.0000093 ns/op	       0 B/op	       0 allocs/op
Benchmark____________ConsumerIsFilteredMatch/64_filter_subjects-8        	1000000000	         0.0000142 ns/op	       0 B/op	       0 allocs/op
Benchmark____________ConsumerIsFilteredMatch/128_filter_subjects-8       	1000000000	         0.0000176 ns/op	       0 B/op	       0 allocs/op
Benchmark____________ConsumerIsFilteredMatch/256_filter_subjects-8       	1000000000	         0.0000438 ns/op	       0 B/op	       0 allocs/op
Benchmark____________ConsumerIsFilteredMatch/512_filter_subjects-8       	1000000000	         0.0000944 ns/op	       0 B/op	       0 allocs/op
Benchmark____________ConsumerIsFilteredMatch/1024_filter_subjects-8      	1000000000	         0.0001221 ns/op	       0 B/op	       0 allocs/op
PASS
ok  	github.com/nats-io/nats-server/v2/server	0.774s
```

After the changes:
```
goos: linux
goarch: amd64
pkg: github.com/nats-io/nats-server/v2/server
cpu: 11th Gen Intel(R) Core(TM) i7-1185G7 @ 3.00GHz
Benchmark____________ConsumerIsFilteredMatch/1_filter_subjects-8         	1000000000	         0.0000030 ns/op	       0 B/op	       0 allocs/op
Benchmark____________ConsumerIsFilteredMatch/2_filter_subjects-8         	1000000000	         0.0000025 ns/op	       0 B/op	       0 allocs/op
Benchmark____________ConsumerIsFilteredMatch/4_filter_subjects-8         	1000000000	         0.0000029 ns/op	       0 B/op	       0 allocs/op
Benchmark____________ConsumerIsFilteredMatch/8_filter_subjects-8         	1000000000	         0.0000033 ns/op	       0 B/op	       0 allocs/op
Benchmark____________ConsumerIsFilteredMatch/16_filter_subjects-8        	1000000000	         0.0000038 ns/op	       0 B/op	       0 allocs/op
Benchmark____________ConsumerIsFilteredMatch/32_filter_subjects-8        	1000000000	         0.0000046 ns/op	       0 B/op	       0 allocs/op
Benchmark____________ConsumerIsFilteredMatch/64_filter_subjects-8        	1000000000	         0.0000058 ns/op	       0 B/op	       0 allocs/op
Benchmark____________ConsumerIsFilteredMatch/128_filter_subjects-8       	1000000000	         0.0000102 ns/op	       0 B/op	       0 allocs/op
Benchmark____________ConsumerIsFilteredMatch/256_filter_subjects-8       	1000000000	         0.0000256 ns/op	       0 B/op	       0 allocs/op
Benchmark____________ConsumerIsFilteredMatch/512_filter_subjects-8       	1000000000	         0.0000330 ns/op	       0 B/op	       0 allocs/op
Benchmark____________ConsumerIsFilteredMatch/1024_filter_subjects-8      	1000000000	         0.0000626 ns/op	       0 B/op	       0 allocs/op
PASS
ok  	github.com/nats-io/nats-server/v2/server	0.521s
```
svenfoo: We, at [Holoplot](https://holoplot.com/), would like to see this merged as we would like to start using multiple filter subjects on consumers in our project. However we found that the performance degrades significantly with larger numbers of filter subjects.


svenfoo: I do not really embrace the way this is done, but I tried to follow the example of the existing code that does subject sub-matching.

I started working on a more holistic approach where I introduced a `Subject` type that holds a string and its tokenized form and then provides methods for matching etc.  This quickly turned out to become a larger change, so I decided to try to get this much smaller change merged first. If there is interest I can clean up the work I've done so far and present it as a proposal for a refactoring that would potentially lead to some smaller performance wins.



Patch:
diff --git a/server/consumer.go b/server/consumer.go
index f71184bc..ef0eb7d2 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -386,12 +386,13 @@ type consumer struct {
 
 // A single subject filter.
 type subjectFilter struct {
-	subject     string
-	nextSeq     uint64
-	currentSeq  uint64
-	pmsg        *jsPubMsg
-	err         error
-	hasWildcard bool
+	subject          string
+	nextSeq          uint64
+	currentSeq       uint64
+	pmsg             *jsPubMsg
+	err              error
+	hasWildcard      bool
+	tokenizedSubject []string
 }
 
 type subjectFilters []*subjectFilter
@@ -936,8 +937,9 @@ func (mset *stream) addConsumerWithAssignment(config *ConsumerConfig, oname stri
 	subjects := gatherSubjectFilters(o.cfg.FilterSubject, o.cfg.FilterSubjects)
 	for _, filter := range subjects {
 		sub := &subjectFilter{
-			subject:     filter,
-			hasWildcard: subjectHasWildcard(filter),
+			subject:          filter,
+			hasWildcard:      subjectHasWildcard(filter),
+			tokenizedSubject: tokenizeSubjectIntoSlice(nil, filter),
 		}
 		o.subjf = append(o.subjf, sub)
 	}
@@ -1858,8 +1860,9 @@ func (o *consumer) updateConfig(cfg *ConsumerConfig) error {
 		newSubjf := make(subjectFilters, 0, len(newSubjects))
 		for _, newFilter := range newSubjects {
 			fs := &subjectFilter{
-				subject:     newFilter,
-				hasWildcard: subjectHasWildcard(newFilter),
+				subject:          newFilter,
+				hasWildcard:      subjectHasWildcard(newFilter),
+				tokenizedSubject: tokenizeSubjectIntoSlice(nil, newFilter),
 			}
 			// If given subject was present, we will retain its fields values
 			// so `getNextMgs` can take advantage of already buffered `pmsgs`.
@@ -3361,9 +3364,10 @@ func (o *consumer) isFilteredMatch(subj string) bool {
 	}
 	// It's quicker to first check for non-wildcard filters, then
 	// iterate again to check for subset match.
-	// TODO(dlc) at speed might be better to just do a sublist with L2 and/or possibly L1.
+	tsa := [32]string{}
+	tts := tokenizeSubjectIntoSlice(tsa[:0], subj)
 	for _, filter := range o.subjf {
-		if subjectIsSubsetMatch(subj, filter.subject) {
+		if isSubsetMatchTokenized(tts, filter.tokenizedSubject) {
 			return true
 		}
 	}
@@ -3945,8 +3949,10 @@ func (o *consumer) loopAndGatherMsgs(qch chan struct{}) {
 			}
 		} else {
 			if o.subjf != nil {
+				tsa := [32]string{}
+				tts := tokenizeSubjectIntoSlice(tsa[:0], pmsg.subj)
 				for i, filter := range o.subjf {
-					if subjectIsSubsetMatch(pmsg.subj, filter.subject) {
+					if isSubsetMatchTokenized(tts, filter.tokenizedSubject) {
 						o.subjf[i].currentSeq--
 						o.subjf[i].nextSeq--
 						break
diff --git a/server/stream.go b/server/stream.go
index 4bdd4868..16861bf0 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -5291,6 +5291,8 @@ func (mset *stream) Store() StreamStore {
 // Lock should be held.
 func (mset *stream) partitionUnique(name string, partitions []string) bool {
 	for _, partition := range partitions {
+		psa := [32]string{}
+		pts := tokenizeSubjectIntoSlice(psa[:0], partition)
 		for n, o := range mset.consumers {
 			// Skip the consumer being checked.
 			if n == name {
@@ -5300,8 +5302,8 @@ func (mset *stream) partitionUnique(name string, partitions []string) bool {
 				return false
 			}
 			for _, filter := range o.subjf {
-				if subjectIsSubsetMatch(partition, filter.subject) ||
-					subjectIsSubsetMatch(filter.subject, partition) {
+				if isSubsetMatchTokenized(pts, filter.tokenizedSubject) ||
+					isSubsetMatchTokenized(filter.tokenizedSubject, pts) {
 					return false
 				}
 			}


Test patch:
diff --git a/server/jetstream_consumer_test.go b/server/jetstream_consumer_test.go
index 56df26ce..ed3de3ad 100644
--- a/server/jetstream_consumer_test.go
+++ b/server/jetstream_consumer_test.go
@@ -21,12 +21,14 @@ import (
 	"fmt"
 	"math/rand"
 	"sort"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
 
 	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nuid"
 )
 
 func TestJetStreamConsumerMultipleFiltersRemoveFilters(t *testing.T) {
@@ -792,3 +794,90 @@ func TestJetStreamConsumerMultipleFiltersLastPerSubject(t *testing.T) {
 	require_Equal(t, "two", msg.Subject)
 
 }
+
+func consumerWithFilterSubjects(filterSubjects []string) *consumer {
+	c := consumer{}
+	for _, filter := range filterSubjects {
+		sub := &subjectFilter{
+			subject:          filter,
+			hasWildcard:      subjectHasWildcard(filter),
+			tokenizedSubject: tokenizeSubjectIntoSlice(nil, filter),
+		}
+		c.subjf = append(c.subjf, sub)
+	}
+
+	return &c
+}
+
+func filterSubjects(n int) []string {
+	fs := make([]string, 0, n)
+	for {
+		literals := []string{"foo", "bar", nuid.Next(), "xyz", "abcdef"}
+		fs = append(fs, strings.Join(literals, "."))
+		if len(fs) == n {
+			return fs
+		}
+		// Create more filterSubjects by going through the literals and replacing one with the '*' wildcard.
+		l := len(literals)
+		for i := 0; i < l; i++ {
+			e := make([]string, l)
+			for j := 0; j < l; j++ {
+				if j == i {
+					e[j] = "*"
+				} else {
+					e[j] = literals[j]
+				}
+			}
+			fs = append(fs, strings.Join(e, "."))
+			if len(fs) == n {
+				return fs
+			}
+		}
+	}
+}
+
+func TestJetStreamConsumerIsFilteredMatch(t *testing.T) {
+	for _, test := range []struct {
+		name           string
+		filterSubjects []string
+		subject        string
+		result         bool
+	}{
+		{"no filter", []string{}, "foo.bar", true},
+		{"literal match", []string{"foo.baz", "foo.bar"}, "foo.bar", true},
+		{"literal mismatch", []string{"foo.baz", "foo.bar"}, "foo.ban", false},
+		{"wildcard > match", []string{"bar.>", "foo.>"}, "foo.bar", true},
+		{"wildcard > match", []string{"bar.>", "foo.>"}, "bar.foo", true},
+		{"wildcard > mismatch", []string{"bar.>", "foo.>"}, "baz.foo", false},
+		{"wildcard * match", []string{"bar.*", "foo.*"}, "foo.bar", true},
+		{"wildcard * match", []string{"bar.*", "foo.*"}, "bar.foo", true},
+		{"wildcard * mismatch", []string{"bar.*", "foo.*"}, "baz.foo", false},
+		{"wildcard * match", []string{"foo.*.x", "foo.*.y"}, "foo.bar.x", true},
+		{"wildcard * match", []string{"foo.*.x", "foo.*.y", "foo.*.z"}, "foo.bar.z", true},
+		{"many mismatch", filterSubjects(100), "foo.bar.do.not.match.any.filter.subject", false},
+		{"many match", filterSubjects(100), "foo.bar.12345.xyz.abcdef", true}, // will be matched by "foo.bar.*.xyz.abcdef"
+	} {
+		test := test
+
+		t.Run(test.name, func(t *testing.T) {
+			t.Parallel()
+
+			c := consumerWithFilterSubjects(test.filterSubjects)
+			if res := c.isFilteredMatch(test.subject); res != test.result {
+				t.Fatalf("Subject %q filtered match of %v, should be %v, got %v",
+					test.subject, test.filterSubjects, test.result, res)
+			}
+		})
+	}
+}
+
+func Benchmark____JetStreamConsumerIsFilteredMatch(b *testing.B) {
+	subject := "foo.bar.do.not.match.any.filter.subject"
+	for n := 1; n <= 1024; n *= 2 {
+		name := fmt.Sprintf("%d filter subjects", int(n))
+		c := consumerWithFilterSubjects(filterSubjects(int(n)))
+		b.Run(name, func(b *testing.B) {
+			c.isFilteredMatch(subject)
+		})
+	}
+}