Files
mesh-controller/vendor/github.com/nats-io/nats-server/v2/server/gsl/gsl.go
T
jochen be92762969 Give every test a bus of its own, at the release the mesh runs (hq ADR 0237)
The live tests reached one shared bus and assert, read and remove the mesh's own objects by
their fixed names, so packages run in parallel deleted what each other read and the suite
passed only one package at a time; a red suite read as noise. internal/testbus starts a server
per test, linked in at the nats-server release go.mod pins, and a test holds that pin to the
catalogue's bus image and to the facts snapshot's bus when there is one, so the tests never run
a bus the mesh does not. The waiter test read a timing (the most connections held at one look)
and now reads the state it means (the fewest held across the wait). make check runs the packages
in parallel under the race detector, with a timeout.
2026-10-06 21:17:02 +02:00

567 lines
12 KiB
Go

// Copyright 2025 The NATS Authors
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package gsl
import (
"errors"
"strings"
"sync"
)
// Sublist is a routing mechanism to handle subject distribution and
// provides a facility to match subjects from published messages to
// interested subscribers. Subscribers can have wildcard subjects to
// match multiple published subjects.
// Common byte variables for wildcards and token separator.
const (
pwc = '*'
pwcs = "*"
fwc = '>'
fwcs = ">"
tsep = "."
btsep = '.'
_EMPTY_ = ""
)
// Sublist related errors
var (
ErrInvalidSubject = errors.New("gsl: invalid subject")
ErrNotFound = errors.New("gsl: no matches found")
ErrNilChan = errors.New("gsl: nil channel")
ErrAlreadyRegistered = errors.New("gsl: notification already registered")
)
// SimpleSublist is an alias type for GenericSublist that takes
// empty values, useful for tracking interest only without any
// unnecessary allocations.
type SimpleSublist = GenericSublist[struct{}]
// NewSimpleSublist will create a simple sublist.
func NewSimpleSublist() *SimpleSublist {
return &GenericSublist[struct{}]{root: newLevel[struct{}]()}
}
// A GenericSublist stores and efficiently retrieves subscriptions.
type GenericSublist[T comparable] struct {
sync.RWMutex
root *level[T]
count uint32
}
// A node contains subscriptions and a pointer to the next level.
type node[T comparable] struct {
next *level[T]
subs map[T]string // value -> subject
}
// A level represents a group of nodes and special pointers to
// wildcard nodes.
type level[T comparable] struct {
nodes map[string]*node[T]
pwc, fwc *node[T]
}
// Create a new default node.
func newNode[T comparable]() *node[T] {
return &node[T]{subs: make(map[T]string)}
}
// Create a new default level.
func newLevel[T comparable]() *level[T] {
return &level[T]{nodes: make(map[string]*node[T])}
}
// NewSublist will create a default sublist with caching enabled per the flag.
func NewSublist[T comparable]() *GenericSublist[T] {
return &GenericSublist[T]{root: newLevel[T]()}
}
// Insert adds a subscription into the sublist
func (s *GenericSublist[T]) Insert(subject string, value T) error {
s.Lock()
var sfwc bool
var n *node[T]
l := s.root
for t := range strings.SplitSeq(subject, tsep) {
lt := len(t)
if lt == 0 || sfwc {
s.Unlock()
return ErrInvalidSubject
}
if lt > 1 {
n = l.nodes[t]
} else {
switch t[0] {
case pwc:
n = l.pwc
case fwc:
n = l.fwc
sfwc = true
default:
n = l.nodes[t]
}
}
if n == nil {
n = newNode[T]()
if lt > 1 {
l.nodes[t] = n
} else {
switch t[0] {
case pwc:
l.pwc = n
case fwc:
l.fwc = n
default:
l.nodes[t] = n
}
}
}
if n.next == nil {
n.next = newLevel[T]()
}
l = n.next
}
n.subs[value] = subject
s.count++
s.Unlock()
return nil
}
// Match will match all entries to the literal subject.
// It will return a set of results for both normal and queue subscribers.
func (s *GenericSublist[T]) Match(subject string, cb func(T)) {
s.match(subject, cb, true)
}
// MatchBytes will match all entries to the literal subject.
// It will return a set of results for both normal and queue subscribers.
func (s *GenericSublist[T]) MatchBytes(subject []byte, cb func(T)) {
s.match(string(subject), cb, true)
}
// 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 *GenericSublist[T]) HasInterest(subject string) bool {
return s.hasInterest(subject, true, nil)
}
// NumInterest will return the number of subs interested in the subject.
// In cases where more detail is not required, this may be faster than Match.
func (s *GenericSublist[T]) NumInterest(subject string) (np int) {
s.hasInterest(subject, true, &np)
return
}
// MatchesFullWildcard returns true if there is top-level ">" interest.
func (s *GenericSublist[T]) MatchesFullWildcard() bool {
if s == nil {
return false
}
s.RLock()
defer s.RUnlock()
return s.root.fwc != nil
}
// MatchesSingleFilter returns the filter when the sublist contains exactly one unique subject.
func (s *GenericSublist[T]) MatchesSingleFilter() (string, bool) {
if s == nil {
return _EMPTY_, false
}
s.RLock()
defer s.RUnlock()
return singleFilter(s.root, _EMPTY_)
}
func singleFilter[T comparable](l *level[T], filter string) (string, bool) {
if l == nil {
return filter, filter != _EMPTY_
}
if len(l.nodes) > 1 {
return _EMPTY_, false
}
var next *node[T]
branches := 0
if l.pwc != nil {
next = l.pwc
branches++
}
if l.fwc != nil {
next = l.fwc
branches++
}
for _, n := range l.nodes {
next = n
branches++
}
if branches != 1 {
return _EMPTY_, false
}
for _, subj := range next.subs {
filter = subj
break
}
if next.next == nil {
return filter, filter != _EMPTY_
}
if filter != _EMPTY_ {
if next.next.numNodes() > 0 {
return _EMPTY_, false
}
return filter, true
}
return singleFilter(next.next, filter)
}
func (s *GenericSublist[T]) match(subject string, cb func(T), doLock bool) {
tsa := [32]string{}
tokens := tsa[:0]
start := 0
for i := 0; i < len(subject); i++ {
if subject[i] == btsep {
if i-start == 0 {
return
}
tokens = append(tokens, subject[start:i])
start = i + 1
}
}
if start >= len(subject) {
return
}
tokens = append(tokens, subject[start:])
if doLock {
s.RLock()
defer s.RUnlock()
}
matchLevel(s.root, tokens, cb)
}
func (s *GenericSublist[T]) hasInterest(subject string, doLock bool, np *int) bool {
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, np)
}
func matchLevelForAny[T comparable](l *level[T], toks []string, np *int) bool {
var pwc, n *node[T]
for i, t := range toks {
if l == nil {
return false
}
if l.fwc != nil {
if np != nil {
*np += len(l.fwc.subs)
}
return true
}
if pwc = l.pwc; pwc != nil {
if match := matchLevelForAny(pwc.next, toks[i+1:], np); match {
return true
}
}
n = l.nodes[t]
if n != nil {
l = n.next
} else {
l = nil
}
}
if n != nil {
if np != nil {
*np += len(n.subs)
}
if len(n.subs) > 0 {
return true
}
}
if pwc != nil {
if np != nil {
*np += len(pwc.subs)
}
return len(pwc.subs) > 0
}
return false
}
// callbacksForResults will make the necessary callbacks for each
// result in this node.
func callbacksForResults[T comparable](n *node[T], cb func(T)) {
for sub := range n.subs {
cb(sub)
}
}
// matchLevel is used to recursively descend into the trie.
func matchLevel[T comparable](l *level[T], toks []string, cb func(T)) {
var pwc, n *node[T]
for i, t := range toks {
if l == nil {
return
}
if l.fwc != nil {
callbacksForResults(l.fwc, cb)
}
if pwc = l.pwc; pwc != nil {
matchLevel(pwc.next, toks[i+1:], cb)
}
n = l.nodes[t]
if n != nil {
l = n.next
} else {
l = nil
}
}
if n != nil {
callbacksForResults(n, cb)
}
if pwc != nil {
callbacksForResults(pwc, cb)
}
}
// lnt is used to track descent into levels for a removal for pruning.
type lnt[T comparable] struct {
l *level[T]
n *node[T]
t string
}
// Raw low level remove, can do batches with lock held outside.
func (s *GenericSublist[T]) remove(subject string, value T, shouldLock bool) error {
if shouldLock {
s.Lock()
defer s.Unlock()
}
var sfwc bool
var n *node[T]
l := s.root
// Track levels for pruning
var lnts [32]lnt[T]
levels := lnts[:0]
for t := range strings.SplitSeq(subject, tsep) {
lt := len(t)
if lt == 0 || sfwc {
return ErrInvalidSubject
}
if l == nil {
return ErrNotFound
}
if lt > 1 {
n = l.nodes[t]
} else {
switch t[0] {
case pwc:
n = l.pwc
case fwc:
n = l.fwc
sfwc = true
default:
n = l.nodes[t]
}
}
if n != nil {
levels = append(levels, lnt[T]{l, n, t})
l = n.next
} else {
l = nil
}
}
if !s.removeFromNode(n, value) {
return ErrNotFound
}
s.count--
for i := len(levels) - 1; i >= 0; i-- {
l, n, t := levels[i].l, levels[i].n, levels[i].t
if n.isEmpty() {
l.pruneNode(n, t)
}
}
return nil
}
// Remove will remove a subscription.
func (s *GenericSublist[T]) Remove(subject string, value T) error {
return s.remove(subject, value, true)
}
// HasInterestStartingIn is a helper for subject tree intersection.
func (s *GenericSublist[T]) HasInterestStartingIn(subj string) bool {
s.RLock()
defer s.RUnlock()
var _tokens [64]string
tokens := tokenizeSubjectIntoSlice(_tokens[:0], subj)
return hasInterestStartingIn(s.root, tokens)
}
func hasInterestStartingIn[T comparable](l *level[T], tokens []string) bool {
if l == nil {
return false
}
if len(tokens) == 0 {
return true
}
token := tokens[0]
if l.fwc != nil {
return true
}
found := false
if pwc := l.pwc; pwc != nil {
found = found || hasInterestStartingIn(pwc.next, tokens[1:])
}
if n := l.nodes[token]; n != nil {
found = found || hasInterestStartingIn(n.next, tokens[1:])
}
return found
}
// pruneNode is used to prune an empty node from the tree.
func (l *level[T]) pruneNode(n *node[T], t string) {
if n == nil {
return
}
if n == l.fwc {
l.fwc = nil
} else if n == l.pwc {
l.pwc = nil
} else {
delete(l.nodes, t)
}
}
// isEmpty will test if the node has any entries. Used
// in pruning.
func (n *node[T]) isEmpty() bool {
return len(n.subs) == 0 && (n.next == nil || n.next.numNodes() == 0)
}
// Return the number of nodes for the given level.
func (l *level[T]) numNodes() int {
if l == nil {
return 0
}
num := len(l.nodes)
if l.pwc != nil {
num++
}
if l.fwc != nil {
num++
}
return num
}
// Remove the sub for the given node.
func (s *GenericSublist[T]) removeFromNode(n *node[T], value T) (found bool) {
if n == nil {
return false
}
if _, found = n.subs[value]; found {
delete(n.subs, value)
}
return found
}
// Count returns the number of subscriptions.
func (s *GenericSublist[T]) Count() uint32 {
s.RLock()
defer s.RUnlock()
return s.count
}
// numLevels will return the maximum number of levels
// contained in the Sublist tree.
func (s *GenericSublist[T]) numLevels() int {
return visitLevel(s.root, 0)
}
// visitLevel is used to descend the Sublist tree structure
// recursively.
func visitLevel[T comparable](l *level[T], depth int) int {
if l == nil || l.numNodes() == 0 {
return depth
}
depth++
maxDepth := depth
for _, n := range l.nodes {
if n == nil {
continue
}
newDepth := visitLevel(n.next, depth)
if newDepth > maxDepth {
maxDepth = newDepth
}
}
if l.pwc != nil {
pwcDepth := visitLevel(l.pwc.next, depth)
if pwcDepth > maxDepth {
maxDepth = pwcDepth
}
}
if l.fwc != nil {
fwcDepth := visitLevel(l.fwc.next, depth)
if fwcDepth > maxDepth {
maxDepth = fwcDepth
}
}
return maxDepth
}
// use similar to append. meaning, the updated slice will be returned
func tokenizeSubjectIntoSlice(tts []string, subject string) []string {
start := 0
for i := 0; i < len(subject); i++ {
if subject[i] == btsep {
tts = append(tts, subject[start:i])
start = i + 1
}
}
tts = append(tts, subject[start:])
return tts
}