Repository: nats-io/nats-server
Pull request #5329 (2024-04-22 16:21:34)

Implement `(*Sublist).HasInterest()`
This may end up being more efficient than `MatchWithResult()` in some cases as we will stop processing on the first match and not spend time building up a `*SublistResult`. May want to give more thought on the caching here — right now it checks the cache, but `HasInterest()` won't populate it, only `Match()` will.

/cc @kozlovic 

Signed-off-by: Neil Twigg <neil@nats.io>

derekcollison: Let me know when no longer draft, etc.

Patch:
diff --git a/server/filestore.go b/server/filestore.go
index fe994d71..650af5f2 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -2257,7 +2257,6 @@ func (mb *msgBlock) firstMatchingMulti(sl *Sublist, start uint64, sm *StoreMsg)
 		sm = new(StoreMsg)
 	}
 
-	var result SublistResult
 	for seq := start; seq <= lseq; seq++ {
 		llseq := mb.llseq
 		fsm, err := mb.cacheLookup(seq, sm)
@@ -2266,7 +2265,7 @@ func (mb *msgBlock) firstMatchingMulti(sl *Sublist, start uint64, sm *StoreMsg)
 		}
 		expireOk := seq == lseq && mb.llseq == seq
 
-		if r := sl.MatchWithResult(fsm.subj, &result); len(r.psubs) > 0 {
+		if sl.HasInterest(fsm.subj) {
 			return fsm, expireOk, nil
 		}
 		// If we are here we did not match, so put the llseq back.
diff --git a/server/memstore.go b/server/memstore.go
index 7a81847d..723520fa 100644
--- a/server/memstore.go
+++ b/server/memstore.go
@@ -1055,13 +1055,12 @@ func (ms *memStore) LoadNextMsgMulti(sl *Sublist, start uint64, smp *StoreMsg) (
 	// Initial setup.
 	fseq, lseq := start, ms.state.LastSeq
 
-	var result SublistResult
 	for nseq := fseq; nseq <= lseq; nseq++ {
 		sm, ok := ms.msgs[nseq]
 		if !ok {
 			continue
 		}
-		if r := sl.MatchWithResult(sm.subj, &result); len(r.psubs) > 0 {
+		if sl.HasInterest(sm.subj) {
 			if smp == nil {
 				smp = new(StoreMsg)
 			}
diff --git a/server/sublist.go b/server/sublist.go
index 8d5859f4..71c256e4 100644
--- a/server/sublist.go
+++ b/server/sublist.go
@@ -527,30 +527,27 @@ var emptyResult = &SublistResult{}
 // Match will match all entries to the literal subject.
 // It will return a set of results for both normal and queue subscribers.
 func (s *Sublist) Match(subject string) *SublistResult {
-	return s.match(subject, true, nil)
+	return s.match(subject, true)
 }
 
-// MatchWithResult will match all entries to the literal subject, reusing the
-// supplied SublistResult to reduce allocations on hot paths.
-// It will return a set of results for both normal and queue subscribers.
-func (s *Sublist) MatchWithResult(subject string, result *SublistResult) *SublistResult {
-	return s.match(subject, true, result)
+// HasInterest will return whether or not there is any interest in the subject.
+// In cases where more detail is not required, this may be faster than Match.
+func (s *Sublist) HasInterest(subject string) bool {
+	return s.hasInterest(subject, true)
 }
 
 func (s *Sublist) matchNoLock(subject string) *SublistResult {
-	return s.match(subject, false, nil)
+	return s.match(subject, false)
 }
 
-func (s *Sublist) match(subject string, doLock bool, result *SublistResult) *SublistResult {
+func (s *Sublist) match(subject string, doLock bool) *SublistResult {
 	atomic.AddUint64(&s.matches, 1)
 
 	// Check cache first.
 	if doLock {
 		s.RLock()
 	}
-	// Writing to the cache is only allowed if not supplying our
-	// own SublistResult, i.e. via call to MatchWithResult.
-	cacheEnabled := result == nil && s.cache != nil
+	cacheEnabled := s.cache != nil
 	r, ok := s.cache[subject]
 	if doLock {
 		s.RUnlock()
@@ -578,11 +575,7 @@ func (s *Sublist) match(subject string, doLock bool, result *SublistResult) *Sub
 	tokens = append(tokens, subject[start:])
 
 	// FIXME(dlc) - Make shared pool between sublist and client readLoop?
-	if result == nil {
-		result = &SublistResult{}
-	}
-	result.psubs = result.psubs[:0]
-	result.qsubs = result.qsubs[:0]
+	result := &SublistResult{}
 
 	// Get result from the main structure and place into the shared cache.
 	// Hold the read lock to avoid race between match and store.
@@ -621,6 +614,47 @@ func (s *Sublist) match(subject string, doLock bool, result *SublistResult) *Sub
 	return result
 }
 
+func (s *Sublist) hasInterest(subject string, doLock bool) bool {
+	// Check cache first.
+	if doLock {
+		s.RLock()
+	}
+	var matched bool
+	if s.cache != nil {
+		_, matched = s.cache[subject]
+	}
+	if doLock {
+		s.RUnlock()
+	}
+	if matched {
+		atomic.AddUint64(&s.cacheHits, 1)
+		return true
+	}
+
+	tsa := [32]string{}
+	tokens := tsa[:0]
+	start := 0
+	for i := 0; i < len(subject); i++ {
+		if subject[i] == btsep {
+			if i-start == 0 {
+				return false
+			}
+			tokens = append(tokens, subject[start:i])
+			start = i + 1
+		}
+	}
+	if start >= len(subject) {
+		return false
+	}
+	tokens = append(tokens, subject[start:])
+
+	if doLock {
+		s.RLock()
+		defer s.RUnlock()
+	}
+	return matchLevelForAny(s.root, tokens)
+}
+
 // Remove entries in the cache until we are under the maximum.
 // TODO(dlc) this could be smarter now that its not inline.
 func (s *Sublist) reduceCacheCount() {
@@ -733,6 +767,36 @@ func matchLevel(l *level, toks []string, results *SublistResult) {
 	}
 }
 
+func matchLevelForAny(l *level, toks []string) bool {
+	var pwc, n *node
+	for i, t := range toks {
+		if l == nil {
+			return false
+		}
+		if l.fwc != nil {
+			return true
+		}
+		if pwc = l.pwc; pwc != nil {
+			if match := matchLevelForAny(pwc.next, toks[i+1:]); match {
+				return true
+			}
+		}
+		n = l.nodes[t]
+		if n != nil {
+			l = n.next
+		} else {
+			l = nil
+		}
+	}
+	if n != nil {
+		return true
+	}
+	if pwc != nil {
+		return true
+	}
+	return false
+}
+
 // lnt is used to track descent into levels for a removal for pruning.
 type lnt struct {
 	l *level


Test patch:
diff --git a/server/sublist_test.go b/server/sublist_test.go
index 1e2e0288..16444b92 100644
--- a/server/sublist_test.go
+++ b/server/sublist_test.go
@@ -1602,6 +1602,28 @@ func TestMain(m *testing.M) {
 	os.Exit(m.Run())
 }
 
+func TestSublistHasInterest(t *testing.T) {
+	sl := NewSublistWithCache()
+	fooSub := newSub("foo")
+	sl.Insert(fooSub)
+
+	// Expect to find that "foo" matches but "bar" doesn't.
+	// At this point nothing should be in the cache.
+	require_True(t, sl.HasInterest("foo"))
+	require_False(t, sl.HasInterest("bar"))
+	require_Equal(t, sl.cacheHits, 0)
+
+	// Now call Match(), which will populate the cache.
+	sl.Match("foo")
+	require_Equal(t, sl.cacheHits, 0)
+
+	// Future calls to HasInterest() should hit the cache now.
+	for i := uint64(1); i <= 5; i++ {
+		require_True(t, sl.HasInterest("foo"))
+		require_Equal(t, sl.cacheHits, i)
+	}
+}
+
 func subsInit(pre string, toks []string) {
 	var sub string
 	for _, t := range toks {