Add index for dynamic config constraints (temporalio#8766)

## What changed?
For dynamic config settings with many constrained values, linear search
can be slow. This adds an on-demand index when the number of constrained
values crosses a threshold.

Also fix convertCache so that the cleanup closures don't reference the
whole Collection, which would prevent it from getting colllected.

## Why?
Optimization.

## How did you test it?
- [ ] built
- [ ] run locally and tested manually
- [x] covered by existing tests
- [x] added new unit test(s)
- [ ] added new functional test(s)

## Potential risks
Increased memory usage for the indexes.

Author: dnr

common/dynamicconfig/client.go

@@ -88,9 +88,9 @@ type (
 		Namespace     string
 		NamespaceID   string
 		TaskQueueName string
+		Destination   string
 		TaskQueueType enumspb.TaskQueueType
 		ShardID       int32
 		TaskType      enumsspb.TaskType
-		Destination   string
 	}
 )

common/dynamicconfig/collection.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"math"
 	"reflect"
 	"runtime"
 	"strconv"
@@ -43,8 +44,12 @@ type (
 		poller goro.Group
 
 		// cache converted values. use weak pointers to avoid holding on to values in the cache
-		// that are no longer in use.
-		convertCache sync.Map // map[weak.Pointer[ConstrainedValue]]any
+		// that are no longer in use. this must be a pointer since the cleanup closures need to
+		// reference this without referencing Collection.
+		convertCache *sync.Map // map[weak.Pointer[ConstrainedValue]]any
+
+		// index by constraints
+		indexCache *sync.Map // map[weak.Pointer[ConstrainedValue]]map[Constraints]int32
 	}
 
 	subscription[T any] struct {
@@ -87,6 +92,9 @@ type (
 
 const (
 	errCountLogThreshold = 1000
+	// After this many constraints, switch to a cached lookup. This value was determined
+	// empirically on my machine using BenchmarkCollectionIndexed.
+	constraintsCacheThreshold = 32
 )
 
 var (
@@ -112,6 +120,8 @@ func NewCollection(client Client, logger log.Logger) *Collection {
 		logger:        logger,
 		errCount:      -1,
 		subscriptions: make(map[Key]map[int]any),
+		convertCache:  new(sync.Map),
+		indexCache:    new(sync.Map),
 	}
 }
 
@@ -216,10 +226,17 @@ func (c *Collection) throttleLog() bool {
 	return errCount < errCountLogThreshold || errCount%errCountLogThreshold == 0
 }
 
-func findMatch(cvs []ConstrainedValue, precedence []Constraints) (*ConstrainedValue, error) {
+func findMatch(
+	cache *sync.Map,
+	cvs []ConstrainedValue,
+	precedence []Constraints,
+) (*ConstrainedValue, error) {
 	if len(cvs) == 0 {
 		return nil, errKeyNotPresent
+	} else if len(cvs) > constraintsCacheThreshold && len(cvs) <= math.MaxInt32 {
+		return findMatchWithCache(cache, cvs, precedence)
 	}
+
 	for _, m := range precedence {
 		for idx, cv := range cvs {
 			if m == cv.Constraints {
@@ -235,6 +252,43 @@ func findMatch(cvs []ConstrainedValue, precedence []Constraints) (*ConstrainedVa
 	return nil, errNoMatchingConstraint
 }
 
+func findMatchWithCache(
+	cache *sync.Map,
+	cvs []ConstrainedValue,
+	precedence []Constraints,
+) (*ConstrainedValue, error) {
+	var cached map[Constraints]int32
+	weakcvp := weak.Make(&cvs[0])
+	if v, ok := cache.Load(weakcvp); ok {
+		cached = v.(map[Constraints]int32) // nolint:revive // unchecked-type-assertion
+	} else {
+		cached = make(map[Constraints]int32, len(cvs))
+		for i := range cvs {
+			// pick first one to match behavior if multiple match
+			if _, ok := cached[cvs[i].Constraints]; !ok {
+				cached[cvs[i].Constraints] = int32(i)
+			}
+		}
+		if _, loaded := cache.LoadOrStore(weakcvp, cached); !loaded {
+			runtime.AddCleanup(&cvs[0], func(w weak.Pointer[ConstrainedValue]) {
+				cache.Delete(w)
+			}, weakcvp)
+		}
+	}
+
+	for _, m := range precedence {
+		if i, ok := cached[m]; ok {
+			// Note: cvs here is the slice returned by Client.GetValue. We want to return a
+			// pointer into that slice so that the converted value is cached as long as the
+			// Client keeps the []ConstrainedValue alive. See the comment on
+			// Client.GetValue.
+			return &cvs[i], nil
+		}
+	}
+	// key is present but no constraint section matches
+	return nil, errNoMatchingConstraint
+}
+
 // matchAndConvert can't be a method of Collection because methods can't be generic, but we can
 // take a *Collection as an argument.
 func matchAndConvert[T any](
@@ -257,7 +311,7 @@ func matchAndConvertCvs[T any](
 	precedence []Constraints,
 	cvs []ConstrainedValue,
 ) (T, any) {
-	cvp, err := findMatch(cvs, precedence)
+	cvp, err := findMatch(c.indexCache, cvs, precedence)
 	if err != nil {
 		// couldn't find a constrained match, use default
 		return def, usingDefaultValue
@@ -449,7 +503,7 @@ func dispatchUpdate[T any](
 	cvs []ConstrainedValue,
 ) {
 	var raw any
-	cvp, err := findMatch(cvs, sub.prec)
+	cvp, err := findMatch(c.indexCache, cvs, sub.prec)
 	if err != nil {
 		raw = usingDefaultValue
 	} else {
@@ -529,11 +583,12 @@ func convertWithCache[T any](c *Collection, key Key, convert func(any) (T, error
 		return zero, err
 	}
 
-	c.convertCache.Store(weakcvp, t)
-
-	runtime.AddCleanup(cvp, func(w weak.Pointer[ConstrainedValue]) {
-		c.convertCache.Delete(w)
-	}, weakcvp)
+	if _, loaded := c.convertCache.LoadOrStore(weakcvp, t); !loaded {
+		cc := c.convertCache // capture only this pointer, not the whole Collection
+		runtime.AddCleanup(cvp, func(w weak.Pointer[ConstrainedValue]) {
+			cc.Delete(w)
+		}, weakcvp)
+	}
 
 	return t, nil
 }

common/dynamicconfig/collection_bench_test.go

@@ -1,6 +1,7 @@
 package dynamicconfig_test
 
 import (
+	"fmt"
 	"testing"
 
 	"go.temporal.io/server/common/dynamicconfig"
@@ -122,3 +123,40 @@ func BenchmarkCollection(b *testing.B) {
 		}
 	})
 }
+
+func BenchmarkCollectionIndexed(b *testing.B) {
+	// You might want to set constraintsCacheThreshold to a high value before running this to
+	// measure the performance of linear search.
+
+	var nums []int
+	for v := 1.0; v < 1000; v *= 1.5 {
+		nums = append(nums, int(v+0.999))
+	}
+	for _, numNs := range nums {
+		// query for the middle one to measure the average
+		queryNs := numNs / 2
+
+		b.Run(fmt.Sprintf("num%d", numNs), func(b *testing.B) {
+			cvs := make([]dynamicconfig.ConstrainedValue, numNs)
+			for i := range cvs {
+				cvs[i] = dynamicconfig.ConstrainedValue{
+					Constraints: dynamicconfig.Constraints{
+						Namespace: fmt.Sprintf("namespace%d", i),
+					},
+					Value: 1000 + i,
+				}
+			}
+
+			cli := dynamicconfig.StaticClient{
+				dynamicconfig.FrontendGlobalNamespaceRPS.Key(): cvs,
+			}
+			cln := dynamicconfig.NewCollection(cli, log.NewNoopLogger())
+			get := dynamicconfig.FrontendGlobalNamespaceRPS.Get(cln)
+			query := fmt.Sprintf("namespace%d", queryNs)
+
+			for b.Loop() {
+				get(query)
+			}
+		})
+	}
+}

common/dynamicconfig/find_match_test.go

@@ -1,9 +1,12 @@
 package dynamicconfig
 
 import (
+	"fmt"
+	"sync"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 // These two tests are in a separate file in the 'dynamicconfig' package to access the private
@@ -54,11 +57,37 @@ func TestFindMatch(t *testing.T) {
 	}
 
 	for _, tc := range testCases {
-		_, err := findMatch(tc.v, tc.filters)
+		var cache sync.Map
+		_, err := findMatch(&cache, tc.v, tc.filters)
 		assert.Equal(t, tc.matched, err == nil)
 	}
 }
 
+func TestFindMatchIndexed(t *testing.T) {
+	var cvs []ConstrainedValue
+	for i := range 100 {
+		cvs = append(cvs, ConstrainedValue{
+			Constraints: Constraints{
+				Namespace: fmt.Sprintf("namespace%d", i),
+			},
+			Value: 1000 + i,
+		})
+	}
+
+	have := []Constraints{{Namespace: "namespace75"}}
+	notHave := []Constraints{{Namespace: "othernamespace"}}
+
+	var cache sync.Map
+	v, err := findMatch(&cache, cvs, have)
+	require.NoError(t, err)
+	require.NotNil(t, v)
+	assert.EqualValues(t, 1075, v.Value)
+	assert.Equal(t, &cvs[75], v)
+
+	_, err = findMatch(&cache, cvs, notHave)
+	assert.Error(t, err)
+}
+
 func TestFindMatchWithTyped(t *testing.T) {
 	testCases := []struct {
 		val      []ConstrainedValue