This is an automated email from the ASF dual-hosted git repository.
Alanxtl pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/dubbo-go.git
The following commit(s) were added to refs/heads/develop by this push:
new ab00bd408 feat(router): add router chain snapshot cache (#3305)
ab00bd408 is described below
commit ab00bd40810a944593738270838e8273a44bcb56
Author: Aether <[email protected]>
AuthorDate: Wed Jun 10 21:36:58 2026 +0800
feat(router): add router chain snapshot cache (#3305)
* feat(router): add invoker-snapshot cache for Poolable route
Introduce routerCache on RouterChain, rebuilt in SetInvokers. Add
FindAddrPoolWithInvokers to atomically read snapshot and pool.
* test(router): add unit tests for tag router bitmap cache path
* test(router): add benchmarks for tag router snapshot cache
* fix: strengthen cache invalidation checks to avoid wrong invoker selection
* fix(router): make FindAddrPool return pool and invokers atomically
* fix(router): tag router failover should return untagged providers only
* docs: add comment explaining cache only works when TagRouter is first in
chain
* fix: use atomic.Value for TagRouter.cache to prevent data race
* fix: add defensive check and TODO comment for TagRouter cache
* chore: use roaring.Iterator to avoid intermediate allocation in
collectInvokers
* chore: enable memory allocation reporting in tag router benchmarks
* chore: extract bitmap key prefixes to constants for Poolable routers
* chore: remove unnecessary copy in FindAddrPool
* fix(router): incorrect tag routing and regex overhead for match field
with non-indexed keys
* refactor(router): split tag router code into router.go, cache.go, and
match.go
* fix(router): guard tag router cache path with chain snapshot generation
* fix router cache injection for added routers
---
cluster/router/chain/cache.go | 100 ++++++++++
cluster/router/chain/chain.go | 40 +++-
cluster/router/chain/chain_test.go | 134 +++++++++++++
cluster/router/router.go | 15 +-
cluster/router/tag/cache.go | 213 +++++++++++++++++++++
cluster/router/tag/cache_benchmarks_test.go | 219 +++++++++++++++++++++
cluster/router/tag/match.go | 2 +-
cluster/router/tag/router.go | 21 ++
cluster/router/tag/router_test.go | 286 ++++++++++++++++++++++++++++
common/constant/key.go | 8 +
10 files changed, 1030 insertions(+), 8 deletions(-)
diff --git a/cluster/router/chain/cache.go b/cluster/router/chain/cache.go
new file mode 100644
index 000000000..24ddfb99a
--- /dev/null
+++ b/cluster/router/chain/cache.go
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 chain
+
+import (
+ "sync"
+)
+
+import (
+ "dubbo.apache.org/dubbo-go/v3/cluster/router"
+ "dubbo.apache.org/dubbo-go/v3/protocol/base"
+)
+
+// routerCache implements router.Cache by storing pre-computed AddrPool
+// keyed by Poolable.Name(). It is rebuilt in its entirety on each
+// SetInvokers call so that bitmap indices stay aligned with the current
+// invoker snapshot.
+type routerCache struct {
+ mu sync.RWMutex
+ // generation identifies the invoker snapshot this cache was built
from. It mirrors
+ // RouterChain.generation and is set under the same write lock as
invokers/pools so that
+ // FindAddrPool can hand all three out as one consistent snapshot.
+ generation uint64
+ invokers []base.Invoker
+ pools map[string]*poolEntry
+}
+
+type poolEntry struct {
+ pool router.AddrPool
+}
+
+func newRouterCache() *routerCache {
+ return &routerCache{
+ pools: make(map[string]*poolEntry),
+ }
+}
+
+func (c *routerCache) GetInvokers() []base.Invoker {
+ c.mu.RLock()
+ defer c.mu.RUnlock()
+ ret := make([]base.Invoker, len(c.invokers))
+ copy(ret, c.invokers)
+ return ret
+}
+
+// FindAddrPool returns the address pool, the invoker snapshot, and the
generation of that
+// snapshot for the given Poolable. The returned invokers slice is shared and
must not be
+// modified by the caller. The generation is always returned (even on a miss)
so callers can
+// detect a snapshot rebuilt by a concurrent SetInvokers.
+func (c *routerCache) FindAddrPool(p router.Poolable) (router.AddrPool,
[]base.Invoker, uint64) {
+ c.mu.RLock()
+ defer c.mu.RUnlock()
+ entry, ok := c.pools[p.Name()]
+ if !ok {
+ return nil, nil, c.generation
+ }
+ return entry.pool, c.invokers, c.generation
+}
+
+// FindAddrMeta is a no-op, reserved for future use.
+func (c *routerCache) FindAddrMeta(p router.Poolable) router.AddrMetadata {
+ return nil
+}
+
+// rebuild iterates all Poolable routers whose ShouldPool returns true,
+// calls Pool on each with current invokers, and atomically swaps in the
+// new pools and generation under write lock. The generation must match the
+// RouterChain.generation produced by the same SetInvokers call.
+func (c *routerCache) rebuild(generation uint64, invokers []base.Invoker,
routers []router.PriorityRouter) {
+ newPools := make(map[string]*poolEntry, len(routers))
+ for _, r := range routers {
+ p, ok := r.(router.Poolable)
+ if !ok || !p.ShouldPool() {
+ continue
+ }
+ pool, _ := p.Pool(invokers) // TODO: store AddrMetadata when
needed
+ newPools[p.Name()] = &poolEntry{pool: pool}
+ }
+
+ c.mu.Lock()
+ c.generation = generation
+ c.invokers = invokers
+ c.pools = newPools
+ c.mu.Unlock()
+}
diff --git a/cluster/router/chain/chain.go b/cluster/router/chain/chain.go
index 428e36d56..adbc7a74b 100644
--- a/cluster/router/chain/chain.go
+++ b/cluster/router/chain/chain.go
@@ -49,12 +49,24 @@ type RouterChain struct {
// instance will never delete or recreate.
builtinRouters []router.PriorityRouter
- mutex sync.RWMutex
+ cache *routerCache
+ // generation is bumped under mutex on every SetInvokers and identifies
the current
+ // invoker snapshot. It is handed to Route callers (via invocation
attribute) and stored
+ // in the cache so a router can prove the cached pool belongs to the
snapshot it is routing
+ // for. 0 means SetInvokers has never run, so no cache exists yet.
+ generation uint64
+ mutex sync.RWMutex
}
// Route Loop routers in RouterChain and call Route method to determine the
target invokers list.
func (c *RouterChain) Route(url *common.URL, invocation base.Invocation)
[]base.Invoker {
- invokers := c.snapshotInvokers()
+ invokers, generation := c.snapshotInvokers()
+ // Publish the snapshot generation so Poolable routers can verify their
cache was built
+ // from this same snapshot before taking the bitmap fast path.
generation == 0 means no
+ // SetInvokers has run yet, hence no cache exists, so there is nothing
to publish.
+ if generation > 0 {
+ invocation.SetAttribute(constant.RouterChainCacheGeneration,
generation)
+ }
finalInvokers := make([]base.Invoker, 0, len(invokers))
// multiple invoker may include different methods, find correct invoker
otherwise
// will return the invoker without methods
@@ -66,10 +78,13 @@ func (c *RouterChain) Route(url *common.URL, invocation
base.Invocation) []base.
if len(finalInvokers) == 0 {
finalInvokers = invokers
+ } else if len(finalInvokers) != len(invokers) {
+ invocation.SetAttribute(constant.RouterCacheDisable, true)
}
for _, r := range c.copyRouters() {
finalInvokers = r.Route(finalInvokers, url, invocation)
+ invocation.SetAttribute(constant.RouterCacheDisable, true)
}
return finalInvokers
}
@@ -93,12 +108,27 @@ func (c *RouterChain) AddRouters(routers
[]router.PriorityRouter) {
func (c *RouterChain) SetInvokers(invokers []base.Invoker) {
c.mutex.Lock()
defer c.mutex.Unlock()
+ c.generation++
c.invokers = invokers
+ c.rebuildCache(invokers)
for _, v := range c.routers {
v.Notify(c.invokers)
}
}
+func (c *RouterChain) rebuildCache(invokers []base.Invoker) {
+ if c.cache == nil {
+ c.cache = newRouterCache()
+ }
+ // Re-inject cache so routers added via AddRouters can receive it.
+ for _, r := range c.routers {
+ if accessor, ok := r.(router.CacheAccessor); ok {
+ accessor.SetCache(c.cache)
+ }
+ }
+ c.cache.rebuild(c.generation, invokers, c.routers)
+}
+
// copyRouters make a snapshot copy from RouterChain's router list.
func (c *RouterChain) copyRouters() []router.PriorityRouter {
c.mutex.RLock()
@@ -108,13 +138,13 @@ func (c *RouterChain) copyRouters()
[]router.PriorityRouter {
return ret
}
-// snapshotInvokers returns a copy of current invokers under lock.
-func (c *RouterChain) snapshotInvokers() []base.Invoker {
+// snapshotInvokers returns a copy of current invokers and their generation
under lock.
+func (c *RouterChain) snapshotInvokers() ([]base.Invoker, uint64) {
c.mutex.RLock()
defer c.mutex.RUnlock()
ret := make([]base.Invoker, len(c.invokers))
copy(ret, c.invokers)
- return ret
+ return ret, c.generation
}
// injectStaticRouters injects static router configurations into the router
chain.
diff --git a/cluster/router/chain/chain_test.go
b/cluster/router/chain/chain_test.go
index 96557c608..5ea72d187 100644
--- a/cluster/router/chain/chain_test.go
+++ b/cluster/router/chain/chain_test.go
@@ -18,10 +18,15 @@
package chain
import (
+ "strconv"
+ "sync"
+ "sync/atomic"
"testing"
)
import (
+ "github.com/RoaringBitmap/roaring"
+
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
@@ -342,3 +347,132 @@ func TestRouteAppliesRoutersOnSnapshot(t *testing.T) {
assert.Equal(t, 1, r2.called)
assert.Equal(t, 1, r2.lastSize)
}
+
+// TestSetInvokersIncrementsAndPublishesGeneration verifies that each
SetInvokers bumps the
+// chain generation and that Route publishes the current generation into the
invocation so
+// Poolable routers can validate their cache against it.
+func TestSetInvokersIncrementsAndPublishesGeneration(t *testing.T) {
+ consumerURL, err := common.NewURL(testConsumerServiceURL)
+ require.NoError(t, err)
+
+ var publishedGen uint64
+ r := &testPriorityRouter{priority: 1, routeFn: func(invokers
[]base.Invoker, _ *common.URL, inv base.Invocation) []base.Invoker {
+ publishedGen =
inv.GetAttributeWithDefaultValue(constant.RouterChainCacheGeneration,
uint64(0)).(uint64)
+ return invokers
+ }}
+ chain := &RouterChain{routers: []router.PriorityRouter{r}}
+
+ invoker := buildInvoker(t, "dubbo://127.0.0.1:20000/com.demo.Service")
+ for i := 1; i <= 3; i++ {
+ chain.SetInvokers([]base.Invoker{invoker})
+ assert.Equal(t, uint64(i), chain.generation)
+ }
+
+ chain.Route(consumerURL, invocation.NewRPCInvocation("Say", nil, nil))
+ assert.Equal(t, uint64(3), publishedGen, "Route should publish the
current chain generation")
+}
+
+// genCheckRouter is a Poolable router that mirrors TagRouter's cache guard.
On the fast path it
+// asserts the invoker snapshot it received from the cache belongs to the same
generation the
+// chain published, catching any snapshot/generation skew under concurrency.
+type genCheckRouter struct {
+ cache router.Cache
+ violations int64
+ fastPaths int64
+}
+
+func (r *genCheckRouter) Name() string { return "gen-check" }
+func (r *genCheckRouter) ShouldPool() bool { return true }
+func (r *genCheckRouter) SetCache(c router.Cache) { r.cache = c }
+func (r *genCheckRouter) URL() *common.URL { return nil }
+func (r *genCheckRouter) Priority() int64 { return 0 }
+func (r *genCheckRouter) Notify(_ []base.Invoker) {}
+
+func (r *genCheckRouter) Pool(invokers []base.Invoker) (router.AddrPool,
router.AddrMetadata) {
+ // Index every invoker under a single key; the test only needs
FindAddrPool to hit.
+ bm := roaring.New()
+ for i := range invokers {
+ bm.Add(uint32(i))
+ }
+ return router.AddrPool{"all": bm}, nil
+}
+
+func (r *genCheckRouter) Route(invokers []base.Invoker, _ *common.URL, inv
base.Invocation) []base.Invoker {
+ if r.cache == nil {
+ return invokers
+ }
+ pool, full, cacheGen := r.cache.FindAddrPool(r)
+ snapGen :=
inv.GetAttributeWithDefaultValue(constant.RouterChainCacheGeneration,
uint64(0)).(uint64)
+ if pool == nil || cacheGen != snapGen {
+ return invokers // fall back, exactly like TagRouter
+ }
+ atomic.AddInt64(&r.fastPaths, 1)
+ // On the fast path every invoker in the cache snapshot must carry the
published generation.
+ // A mismatch would mean the guard let a snapshot from a concurrent
SetInvokers through.
+ for _, ivk := range full {
+ if ivk.GetURL().GetParam("snapgen", "") !=
strconv.FormatUint(snapGen, 10) {
+ atomic.AddInt64(&r.violations, 1)
+ }
+ }
+ return full
+}
+
+// TestRouteCacheGenerationRace hammers SetInvokers and Route concurrently.
Each invoker set is
+// tagged with the generation that produced it, so the Poolable router can
detect any skew
+// between the published generation and the cache snapshot it routes over. Run
with -race.
+func TestRouteCacheGenerationRace(t *testing.T) {
+ consumerURL, err := common.NewURL(testConsumerServiceURL)
+ require.NoError(t, err)
+
+ r := &genCheckRouter{}
+ chain := &RouterChain{routers: []router.PriorityRouter{r}}
+
+ makeSet := func(gen uint64, n int) []base.Invoker {
+ out := make([]base.Invoker, 0, n)
+ for i := 0; i < n; i++ {
+ u, _ := common.NewURL("dubbo://127.0.0.1:2000" +
strconv.Itoa(i) + "/com.demo.Service")
+ u.SetParam("snapgen", strconv.FormatUint(gen, 10))
+ out = append(out, base.NewBaseInvoker(u))
+ }
+ return out
+ }
+
+ // Seed generation 1 so the cache exists before readers start.
+ chain.SetInvokers(makeSet(1, 3))
+
+ stop := make(chan struct{})
+ var writer sync.WaitGroup
+ writer.Add(1)
+ go func() {
+ defer writer.Done()
+ gen := uint64(1)
+ for {
+ select {
+ case <-stop:
+ return
+ default:
+ gen++
+ size := 2 + int(gen%4)
+ chain.SetInvokers(makeSet(gen, size))
+ }
+ }
+ }()
+
+ var readers sync.WaitGroup
+ for i := 0; i < 4; i++ {
+ readers.Add(1)
+ go func() {
+ defer readers.Done()
+ for j := 0; j < 2000; j++ {
+ chain.Route(consumerURL,
invocation.NewRPCInvocation("Say", nil, nil))
+ }
+ }()
+ }
+
+ readers.Wait()
+ close(stop)
+ writer.Wait()
+
+ assert.Equal(t, int64(0), atomic.LoadInt64(&r.violations),
+ "cache snapshot generation must always match the published
generation on the fast path")
+}
diff --git a/cluster/router/router.go b/cluster/router/router.go
index 8c5f1b226..80c4765fc 100644
--- a/cluster/router/router.go
+++ b/cluster/router/router.go
@@ -95,9 +95,20 @@ type Cache interface {
// GetInvokers returns the snapshot of received invokers.
GetInvokers() []base.Invoker
- // FindAddrPool returns address pool associated with the given Poolable
instance.
- FindAddrPool(Poolable) AddrPool
+ // FindAddrPool returns the address pool, the invoker snapshot, and the
generation of that
+ // snapshot in a single locked read. The generation lets callers verify
the pool/invokers
+ // belong to the same generation the chain snapshotted for the current
route, so the bitmap
+ // indices stay aligned with the invoker slice and the route is not
served from a snapshot
+ // produced by a concurrent SetInvokers.
+ FindAddrPool(Poolable) (AddrPool, []base.Invoker, uint64)
// FindAddrMeta returns address metadata associated with the given
Poolable instance.
FindAddrMeta(Poolable) AddrMetadata
}
+
+// CacheAccessor allow routers to receive the invoker-snapshot cache.
+// Implemented by Poolable routers so that RouterChain can pass the cache
+// reference after it is built in SetInvokers.
+type CacheAccessor interface {
+ SetCache(Cache)
+}
diff --git a/cluster/router/tag/cache.go b/cluster/router/tag/cache.go
new file mode 100644
index 000000000..e665d2f61
--- /dev/null
+++ b/cluster/router/tag/cache.go
@@ -0,0 +1,213 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 tag
+
+import (
+ "strings"
+)
+
+import (
+ "github.com/RoaringBitmap/roaring"
+)
+
+import (
+ "dubbo.apache.org/dubbo-go/v3/cluster/router"
+ "dubbo.apache.org/dubbo-go/v3/common"
+ "dubbo.apache.org/dubbo-go/v3/common/constant"
+ "dubbo.apache.org/dubbo-go/v3/global"
+ "dubbo.apache.org/dubbo-go/v3/protocol/base"
+)
+
+func (p *PriorityRouter) Name() string {
+ return "tag"
+}
+
+func (p *PriorityRouter) ShouldPool() bool {
+ return true
+}
+
+// Pool builds bitmap indices for tag, address, and port keys.
+// Key space uses "\x00" as separator.
+// Other Poolable routers should use different prefixes to avoid conflicts.
+func (p *PriorityRouter) Pool(invokers []base.Invoker) (router.AddrPool,
router.AddrMetadata) {
+ pool := make(router.AddrPool)
+ for i, invoker := range invokers {
+ url := invoker.GetURL()
+ upsertBM(pool, constant.PoolKeyAll, i)
+ tag := url.GetParam(constant.Tagkey, "")
+ upsertBM(pool, constant.PoolKeyTagPrefix+tag, i)
+ addr := url.Location
+ if addr != "" {
+ upsertBM(pool, constant.PoolKeyAddrPrefix+addr, i)
+ if idx := strings.LastIndex(addr, ":"); idx > 0 {
+ upsertBM(pool,
constant.PoolKeyPortPrefix+addr[idx+1:], i)
+ }
+ }
+ }
+ return pool, nil
+}
+
+func (p *PriorityRouter) SetCache(cache router.Cache) {
+ if cache != nil {
+ p.cache.Store(cache)
+ }
+}
+
+func (p *PriorityRouter) routeWithPool(invokers []base.Invoker, pool
router.AddrPool, url *common.URL, invocation base.Invocation) []base.Invoker {
+ if len(invokers) == 0 {
+ return invokers
+ }
+ tag := invocation.GetAttachmentWithDefaultValue(constant.Tagkey,
url.GetParam(constant.Tagkey, ""))
+
+ application := invokers[0].GetURL().GetParam(constant.ApplicationKey,
"")
+ key := strings.Join([]string{application,
constant.TagRouterRuleSuffix}, "")
+ value, ok := p.routerConfigs.Load(key)
+ if !ok {
+ return collectInvokers(invokers, p.staticTagMatchBM(pool, tag,
url, invocation))
+ }
+ routerCfg := value.(global.RouterConfig)
+ enabled := routerCfg.Enabled == nil || *routerCfg.Enabled
+ valid := (routerCfg.Valid != nil && *routerCfg.Valid) ||
(routerCfg.Valid == nil && len(routerCfg.Tags) > 0)
+ if !enabled || !valid {
+ return collectInvokers(invokers, p.staticTagMatchBM(pool, tag,
url, invocation))
+ }
+ if tag == "" {
+ return collectInvokers(invokers, p.emptyTagMatchBM(pool,
routerCfg))
+ }
+ bm := p.requestTagMatchBM(pool, url, invocation, routerCfg, tag)
+ if bm == nil {
+ return requestTag(invokers, url, invocation, routerCfg, tag)
+ }
+ return collectInvokers(invokers, bm)
+}
+
+func (p *PriorityRouter) staticTagMatchBM(pool router.AddrPool, tag string,
url *common.URL, invocation base.Invocation) *roaring.Bitmap {
+ if tag != "" {
+ if bm := pool[constant.PoolKeyTagPrefix+tag]; bm != nil &&
!bm.IsEmpty() {
+ return bm
+ }
+ if requestIsForce(url, invocation) {
+ return nil
+ }
+ }
+ return pool[constant.PoolKeyTagPrefix]
+}
+
+func (p *PriorityRouter) requestTagMatchBM(pool router.AddrPool, url
*common.URL, invocation base.Invocation,
+ cfg global.RouterConfig, tag string) *roaring.Bitmap {
+ var (
+ addresses []string
+ match []*common.ParamMatch
+ )
+ for _, tagCfg := range cfg.Tags {
+ if tagCfg.Name == tag {
+ addresses = tagCfg.Addresses
+ match = tagCfg.Match
+ break
+ }
+ }
+
+ // ParamMatch is not bitmap-cached; fall back to requestTag.
+ if len(match) != 0 {
+ return nil
+ }
+
+ var resultBM *roaring.Bitmap
+ if len(addresses) != 0 {
+ resultBM = p.addressesBM(pool, addresses)
+ } else {
+ resultBM = pool[constant.PoolKeyTagPrefix+tag]
+ }
+
+ if (cfg.Force != nil && *cfg.Force) || requestIsForce(url, invocation) {
+ return resultBM
+ }
+ if resultBM != nil && !resultBM.IsEmpty() {
+ return resultBM
+ }
+
+ emptyBM := pool[constant.PoolKeyTagPrefix]
+ if emptyBM == nil {
+ return nil
+ }
+ if len(addresses) == 0 {
+ return emptyBM
+ }
+ addrBM := p.addressesBM(pool, addresses)
+ result := emptyBM.Clone()
+ if addrBM != nil {
+ result.AndNot(addrBM)
+ }
+ return result
+}
+
+func (p *PriorityRouter) emptyTagMatchBM(pool router.AddrPool, cfg
global.RouterConfig) *roaring.Bitmap {
+ bm := pool[constant.PoolKeyTagPrefix]
+ if bm == nil || bm.IsEmpty() {
+ return bm
+ }
+ for _, tagCfg := range cfg.Tags {
+ if len(tagCfg.Addresses) == 0 {
+ continue
+ }
+ addrBM := p.addressesBM(pool, tagCfg.Addresses)
+ if addrBM != nil && !addrBM.IsEmpty() && bm != nil {
+ result := bm.Clone()
+ result.AndNot(addrBM)
+ bm = result
+ }
+ }
+ return bm
+}
+
+func (p *PriorityRouter) addressesBM(pool router.AddrPool, addrs []string)
*roaring.Bitmap {
+ bm := roaring.NewBitmap()
+ for _, addr := range addrs {
+ if ab := pool[constant.PoolKeyAddrPrefix+addr]; ab != nil {
+ bm.Or(ab)
+ }
+ if idx := strings.LastIndex(addr, ":"); idx > 0 && addr[:idx]
== constant.AnyHostValue {
+ if pb := pool[constant.PoolKeyPortPrefix+addr[idx+1:]];
pb != nil {
+ bm.Or(pb)
+ }
+ }
+ }
+ return bm
+}
+
+func collectInvokers(invokers []base.Invoker, bm *roaring.Bitmap)
[]base.Invoker {
+ if bm == nil || bm.IsEmpty() {
+ return []base.Invoker{}
+ }
+ result := make([]base.Invoker, 0, bm.GetCardinality())
+ it := bm.Iterator()
+ for it.HasNext() {
+ idx := it.Next()
+ if int(idx) < len(invokers) {
+ result = append(result, invokers[int(idx)])
+ }
+ }
+ return result
+}
+
+func upsertBM(pool router.AddrPool, key string, idx int) {
+ if _, ok := pool[key]; !ok {
+ pool[key] = roaring.NewBitmap()
+ }
+ pool[key].Add(uint32(idx))
+}
diff --git a/cluster/router/tag/cache_benchmarks_test.go
b/cluster/router/tag/cache_benchmarks_test.go
new file mode 100644
index 000000000..5d07941ae
--- /dev/null
+++ b/cluster/router/tag/cache_benchmarks_test.go
@@ -0,0 +1,219 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 tag
+
+import (
+ "fmt"
+ "testing"
+)
+
+import (
+ "dubbo.apache.org/dubbo-go/v3/common"
+ "dubbo.apache.org/dubbo-go/v3/common/constant"
+ "dubbo.apache.org/dubbo-go/v3/global"
+ "dubbo.apache.org/dubbo-go/v3/protocol/base"
+ "dubbo.apache.org/dubbo-go/v3/protocol/invocation"
+)
+
+// makeBenchInvokers creates n invokers where the first tagged ones carry
tag=gray.
+// All invokers get the given app as ApplicationKey.
+func makeBenchInvokers(n, tagged int, app string) []base.Invoker {
+ invokers := make([]base.Invoker, n)
+ for i := 0; i < n; i++ {
+ u, _ :=
common.NewURL(fmt.Sprintf("dubbo://192.168.0.%d:20000/com.xxx.xxx.UserProvider?interface=com.xxx.xxx.UserProvider&group=&version=3.1.0",
i+1))
+ u.SetParam(constant.ApplicationKey, app)
+ if i < tagged {
+ u.SetParam(constant.Tagkey, "gray")
+ }
+ invokers[i] = base.NewBaseInvoker(u)
+ }
+ return invokers
+}
+
+func newBenchConsumerURL(app string) *common.URL {
+ u, _ :=
common.NewURL(fmt.Sprintf("consumer://127.0.0.1:20000/com.xxx.xxx.UserProvider?interface=com.xxx.xxx.UserProvider&group=&version=3.1.0&%s=%s",
constant.ApplicationKey, app))
+ return u
+}
+
+// BenchmarkRouteStaticTag compares cached vs no-cache paths for static tag
routing.
+// The tag is embedded in invoker URLs; the request carries tag=gray.
+//
+// cached: pool["tag\x00"+tag] single map lookup + bitmap iteration
+// no_cache: filterInvokers linear scan with GetURL().GetParam per invoker
+func BenchmarkRouteStaticTag(b *testing.B) {
+ b.ReportAllocs()
+ for _, n := range []int{10, 100, 1000} {
+ tagged := n * 30 / 100
+ invokers := makeBenchInvokers(n, tagged, "bench-static")
+ url := newBenchConsumerURL("bench-static")
+ invoc := invocation.NewRPCInvocation("GetUser", nil,
map[string]any{constant.Tagkey: "gray"})
+
+ b.Run(fmt.Sprintf("no_cache/N=%d", n), func(b *testing.B) {
+ p, _ := NewTagPriorityRouter()
+ b.ResetTimer()
+ for i := 0; i < b.N; i++ {
+ p.Route(invokers, url, invoc)
+ }
+ })
+ }
+ for _, n := range []int{10, 100, 1000} {
+ tagged := n * 30 / 100
+ invokers := makeBenchInvokers(n, tagged, "bench-static")
+ url := newBenchConsumerURL("bench-static")
+ invoc := invocation.NewRPCInvocation("GetUser", nil,
map[string]any{constant.Tagkey: "gray"})
+
+ b.Run(fmt.Sprintf("cached/N=%d", n), func(b *testing.B) {
+ p, _ := NewTagPriorityRouter()
+ withCache(p, invokers)
+ b.ResetTimer()
+ for i := 0; i < b.N; i++ {
+ p.Route(invokers, url, invoc)
+ }
+ })
+ }
+}
+
+// BenchmarkRouteDynamicTagAddr compares cached vs no-cache paths for dynamic
tag
+// routing with an address list. The request carries tag=gray; the dynamic
config
+// specifies 3 addresses for the gray tag.
+//
+// cached: requestTagMatchBM -> addressesBM map lookups + bitmap OR
+// no_cache: requestTag -> filterInvokers + getAddressPredicate O(N*A) nested
loop
+func BenchmarkRouteDynamicTagAddr(b *testing.B) {
+ b.ReportAllocs()
+ for _, n := range []int{10, 100, 1000} {
+ tagged := n * 30 / 100
+ app := "bench-dynaddr"
+ invokers := makeBenchInvokers(n, tagged, app)
+ url := newBenchConsumerURL(app)
+ invoc := invocation.NewRPCInvocation("GetUser", nil,
map[string]any{constant.Tagkey: "gray"})
+ cfgKey := app + constant.TagRouterRuleSuffix
+ cfg := global.RouterConfig{
+ Key: cfgKey,
+ Force: boolPtr(false),
+ Enabled: boolPtr(true),
+ Valid: boolPtr(true),
+ Tags: []global.Tag{{
+ Name: "gray",
+ Addresses: []string{"192.168.0.1:20000",
"192.168.0.2:20000", "192.168.0.3:20000"},
+ }},
+ }
+
+ b.Run(fmt.Sprintf("no_cache/N=%d", n), func(b *testing.B) {
+ p, _ := NewTagPriorityRouter()
+ p.routerConfigs.Store(cfgKey, cfg)
+ b.ResetTimer()
+ for i := 0; i < b.N; i++ {
+ p.Route(invokers, url, invoc)
+ }
+ })
+ }
+ for _, n := range []int{10, 100, 1000} {
+ tagged := n * 30 / 100
+ app := "bench-dynaddr"
+ invokers := makeBenchInvokers(n, tagged, app)
+ url := newBenchConsumerURL(app)
+ invoc := invocation.NewRPCInvocation("GetUser", nil,
map[string]any{constant.Tagkey: "gray"})
+ cfgKey := app + constant.TagRouterRuleSuffix
+ cfg := global.RouterConfig{
+ Key: cfgKey,
+ Force: boolPtr(false),
+ Enabled: boolPtr(true),
+ Valid: boolPtr(true),
+ Tags: []global.Tag{{
+ Name: "gray",
+ Addresses: []string{"192.168.0.1:20000",
"192.168.0.2:20000", "192.168.0.3:20000"},
+ }},
+ }
+
+ b.Run(fmt.Sprintf("cached/N=%d", n), func(b *testing.B) {
+ p, _ := NewTagPriorityRouter()
+ withCache(p, invokers)
+ p.routerConfigs.Store(cfgKey, cfg)
+ b.ResetTimer()
+ for i := 0; i < b.N; i++ {
+ p.Route(invokers, url, invoc)
+ }
+ })
+ }
+}
+
+// BenchmarkRouteCacheHitBenefit measures cache benefit at fixed N=100 with
varying
+// tag density (10/50/90 tagged invokers). The dynamic config specifies 3
addresses
+// for the gray tag; the request carries tag=gray.
+//
+// At low tagged ratio the no-cache path scans all N but finds few matches
(O(N));
+// the cached path directly locates a small bitmap (O(result_size)).
+// At high tagged ratio the result set is large and the gap narrows.
+func BenchmarkRouteCacheHitBenefit(b *testing.B) {
+ b.ReportAllocs()
+ n := 100
+ for _, tagged := range []int{10, 50, 90} {
+ app := "bench-hitbenefit"
+ invokers := makeBenchInvokers(n, tagged, app)
+ url := newBenchConsumerURL(app)
+ invoc := invocation.NewRPCInvocation("GetUser", nil,
map[string]any{constant.Tagkey: "gray"})
+ cfgKey := app + constant.TagRouterRuleSuffix
+ cfg := global.RouterConfig{
+ Key: cfgKey,
+ Force: boolPtr(false),
+ Enabled: boolPtr(true),
+ Valid: boolPtr(true),
+ Tags: []global.Tag{{
+ Name: "gray",
+ Addresses: []string{"192.168.0.1:20000",
"192.168.0.2:20000", "192.168.0.3:20000"},
+ }},
+ }
+
+ b.Run(fmt.Sprintf("no_cache/tagged=%d", tagged), func(b
*testing.B) {
+ p, _ := NewTagPriorityRouter()
+ p.routerConfigs.Store(cfgKey, cfg)
+ b.ResetTimer()
+ for i := 0; i < b.N; i++ {
+ p.Route(invokers, url, invoc)
+ }
+ })
+ }
+ for _, tagged := range []int{10, 50, 90} {
+ app := "bench-hitbenefit"
+ invokers := makeBenchInvokers(n, tagged, app)
+ url := newBenchConsumerURL(app)
+ invoc := invocation.NewRPCInvocation("GetUser", nil,
map[string]any{constant.Tagkey: "gray"})
+ cfgKey := app + constant.TagRouterRuleSuffix
+ cfg := global.RouterConfig{
+ Key: cfgKey,
+ Force: boolPtr(false),
+ Enabled: boolPtr(true),
+ Valid: boolPtr(true),
+ Tags: []global.Tag{{
+ Name: "gray",
+ Addresses: []string{"192.168.0.1:20000",
"192.168.0.2:20000", "192.168.0.3:20000"},
+ }},
+ }
+
+ b.Run(fmt.Sprintf("cached/tagged=%d", tagged), func(b
*testing.B) {
+ p, _ := NewTagPriorityRouter()
+ withCache(p, invokers)
+ p.routerConfigs.Store(cfgKey, cfg)
+ b.ResetTimer()
+ for i := 0; i < b.N; i++ {
+ p.Route(invokers, url, invoc)
+ }
+ })
+ }
+}
diff --git a/cluster/router/tag/match.go b/cluster/router/tag/match.go
index 621f9afe8..dea37af9e 100644
--- a/cluster/router/tag/match.go
+++ b/cluster/router/tag/match.go
@@ -146,7 +146,7 @@ func requestTag(invokers []base.Invoker, url *common.URL,
invocation base.Invoca
if len(addresses) == 0 {
return result
}
- result = filterInvokers(invokers, addresses, getAddressPredicate(true))
+ result = filterInvokers(result, addresses, getAddressPredicate(true))
logger.Debugf("[Router][Tag] failover match all providers without any
tags, invokers=%+v", result)
return result
}
diff --git a/cluster/router/tag/router.go b/cluster/router/tag/router.go
index bc67be186..635020fd9 100644
--- a/cluster/router/tag/router.go
+++ b/cluster/router/tag/router.go
@@ -20,6 +20,7 @@ package tag
import (
"strings"
"sync"
+ "sync/atomic"
)
import (
@@ -29,6 +30,7 @@ import (
)
import (
+ "dubbo.apache.org/dubbo-go/v3/cluster/router"
"dubbo.apache.org/dubbo-go/v3/common"
conf "dubbo.apache.org/dubbo-go/v3/common/config"
"dubbo.apache.org/dubbo-go/v3/common/constant"
@@ -40,6 +42,7 @@ import (
type PriorityRouter struct {
routerConfigs sync.Map
+ cache atomic.Value // router.Cache
}
func NewTagPriorityRouter() (*PriorityRouter, error) {
@@ -52,6 +55,24 @@ func (p *PriorityRouter) Route(invokers []base.Invoker, url
*common.URL, invocat
logger.Warn("[Router][Tag] invokers from previous router is
empty")
return invokers
}
+
+ // Cache only takes effect when TagRouter is the first router in the
chain.
+ // RouterChain sets RouterCacheDisable=true after each router, so later
routers always skip cache.
+ if v := p.cache.Load(); v != nil {
+ if
!invocation.GetAttributeWithDefaultValue(constant.RouterCacheDisable,
false).(bool) {
+ c := v.(router.Cache)
+ pool, fullInvokers, cacheGen := c.FindAddrPool(p)
+ snapshotGen :=
invocation.GetAttributeWithDefaultValue(constant.RouterChainCacheGeneration,
uint64(0)).(uint64)
+ // Only take the bitmap fast path when the cache
snapshot is the same generation the
+ // chain snapshotted for this call. Otherwise a
concurrent SetInvokers may have rebuilt
+ // the cache between the chain snapshot and this
lookup, so fall through to the original
+ // matcher over the invokers passed in (finalInvokers).
+ if pool != nil && fullInvokers != nil && cacheGen ==
snapshotGen {
+ return p.routeWithPool(fullInvokers, pool, url,
invocation)
+ }
+ }
+ }
+
// get application name from invoker to look up tag routing config
application := invokers[0].GetURL().GetParam(constant.ApplicationKey,
"")
key := strings.Join([]string{application,
constant.TagRouterRuleSuffix}, "")
diff --git a/cluster/router/tag/router_test.go
b/cluster/router/tag/router_test.go
index 630d573a9..6a9a291aa 100644
--- a/cluster/router/tag/router_test.go
+++ b/cluster/router/tag/router_test.go
@@ -28,6 +28,7 @@ import (
)
import (
+ "dubbo.apache.org/dubbo-go/v3/cluster/router"
"dubbo.apache.org/dubbo-go/v3/common"
common_cfg "dubbo.apache.org/dubbo-go/v3/common/config"
"dubbo.apache.org/dubbo-go/v3/common/constant"
@@ -546,3 +547,288 @@ func TestRouteNilDefaults(t *testing.T) {
assert.Len(t, result, 3)
}
+
+type mockCache struct {
+ invokers []base.Invoker
+ pool router.AddrPool
+ generation uint64
+}
+
+func (m *mockCache) GetInvokers() []base.Invoker { return m.invokers }
+func (m *mockCache) FindAddrPool(_ router.Poolable) (router.AddrPool,
[]base.Invoker, uint64) {
+ return m.pool, m.invokers, m.generation
+}
+func (m *mockCache) FindAddrMeta(_ router.Poolable) router.AddrMetadata {
return nil }
+
+func newCacheRouter() *PriorityRouter {
+ initUrl()
+ p, _ := NewTagPriorityRouter()
+ return p
+}
+
+func makeInvokers(tag1, tag2, tag3 string) []base.Invoker {
+ u1, _ :=
common.NewURL("dubbo://192.168.0.1:20000/com.xxx.xxx.UserProvider?interface=com.xxx.xxx.UserProvider&group=&version=3.1.0")
+ u2, _ :=
common.NewURL("dubbo://192.168.0.2:20000/com.xxx.xxx.UserProvider?interface=com.xxx.xxx.UserProvider&group=&version=3.1.0")
+ u3, _ :=
common.NewURL("dubbo://192.168.0.3:20000/com.xxx.xxx.UserProvider?interface=com.xxx.xxx.UserProvider&group=&version=3.1.0")
+ if tag1 != "" {
+ u1.SetParam(constant.Tagkey, tag1)
+ }
+ if tag2 != "" {
+ u2.SetParam(constant.Tagkey, tag2)
+ }
+ if tag3 != "" {
+ u3.SetParam(constant.Tagkey, tag3)
+ }
+ return []base.Invoker{
+ base.NewBaseInvoker(u1),
+ base.NewBaseInvoker(u2),
+ base.NewBaseInvoker(u3),
+ }
+}
+
+func withCache(p *PriorityRouter, invokers []base.Invoker) {
+ pool, _ := p.Pool(invokers)
+ p.cache.Store(&mockCache{invokers: invokers, pool: pool})
+}
+
+func boolPtr(v bool) *bool { return &v }
+
+func TestRouteBitmapStaticTag(t *testing.T) {
+ p := newCacheRouter()
+ invokers := makeInvokers("gray", "gray", "")
+ withCache(p, invokers)
+
+ t.Run("request has tag, returns only matching invokers", func(t
*testing.T) {
+ attachments := map[string]any{constant.Tagkey: "gray"}
+ result := p.Route(invokers, consumerUrl,
invocation.NewRPCInvocation("GetUser", nil, attachments))
+ assert.Len(t, result, 2)
+ for _, r := range result {
+ assert.Equal(t, "gray",
r.GetURL().GetParam(constant.Tagkey, ""))
+ }
+ })
+
+ t.Run("request has no tag, returns untagged invokers", func(t
*testing.T) {
+ result := p.Route(invokers, consumerUrl,
invocation.NewRPCInvocation("GetUser", nil, nil))
+ assert.Len(t, result, 1)
+ assert.Empty(t, result[0].GetURL().GetParam(constant.Tagkey,
""))
+ })
+
+ t.Run("request has non-matching tag with force, returns empty", func(t
*testing.T) {
+ attachments := map[string]any{constant.Tagkey: "nonexistent",
constant.ForceUseTag: "true"}
+ result := p.Route(invokers, consumerUrl,
invocation.NewRPCInvocation("GetUser", nil, attachments))
+ assert.Empty(t, result)
+ })
+}
+
+func TestRouteBitmapDynamicTagAddress(t *testing.T) {
+ p := newCacheRouter()
+ invokers := makeInvokers("gray", "gray", "")
+ withCache(p, invokers)
+
+ p.routerConfigs.Store(consumerUrl.GetParam(constant.ApplicationKey,
"")+constant.TagRouterRuleSuffix, global.RouterConfig{
+ Key: consumerUrl.Service() + constant.TagRouterRuleSuffix,
+ Force: boolPtr(false),
+ Enabled: boolPtr(true),
+ Valid: boolPtr(true),
+ Tags: []global.Tag{{
+ Name: "gray",
+ Addresses: []string{"192.168.0.1:20000"},
+ }},
+ })
+
+ t.Run("address matches only selected invoker", func(t *testing.T) {
+ attachments := map[string]any{constant.Tagkey: "gray"}
+ result := p.Route(invokers, consumerUrl,
invocation.NewRPCInvocation("GetUser", nil, attachments))
+ assert.Len(t, result, 1)
+ assert.Equal(t, "192.168.0.1:20000",
result[0].GetURL().Location)
+ })
+}
+
+func TestRouteBitmapDynamicTagAnyHost(t *testing.T) {
+ p := newCacheRouter()
+ invokers := makeInvokers("gray", "gray", "gray")
+ withCache(p, invokers)
+
+ p.routerConfigs.Store(consumerUrl.GetParam(constant.ApplicationKey,
"")+constant.TagRouterRuleSuffix, global.RouterConfig{
+ Key: consumerUrl.Service() + constant.TagRouterRuleSuffix,
+ Force: boolPtr(true),
+ Enabled: boolPtr(true),
+ Valid: boolPtr(true),
+ Tags: []global.Tag{{
+ Name: "gray",
+ Addresses: []string{constant.AnyHostValue + ":20000"},
+ }},
+ })
+
+ t.Run("anyhost address matches all invokers on that port", func(t
*testing.T) {
+ attachments := map[string]any{constant.Tagkey: "gray"}
+ result := p.Route(invokers, consumerUrl,
invocation.NewRPCInvocation("GetUser", nil, attachments))
+ assert.Len(t, result, 3)
+ })
+}
+
+func TestRouteBitmapDynamicEmptyTag(t *testing.T) {
+ p := newCacheRouter()
+ invokers := makeInvokers("gray", "", "")
+ withCache(p, invokers)
+
+ p.routerConfigs.Store(consumerUrl.GetParam(constant.ApplicationKey,
"")+constant.TagRouterRuleSuffix, global.RouterConfig{
+ Key: consumerUrl.Service() + constant.TagRouterRuleSuffix,
+ Force: boolPtr(false),
+ Enabled: boolPtr(true),
+ Valid: boolPtr(true),
+ Tags: []global.Tag{{
+ Addresses: []string{"192.168.0.1:20000"},
+ }},
+ })
+
+ t.Run("empty request tag with address exclusion", func(t *testing.T) {
+ result := p.Route(invokers, consumerUrl,
invocation.NewRPCInvocation("GetUser", nil, nil))
+ assert.Len(t, result, 2)
+ for _, r := range result {
+ assert.NotEqual(t, "192.168.0.1:20000",
r.GetURL().Location)
+ }
+ })
+}
+
+func TestRouteDynamicParamExactFallback(t *testing.T) {
+ p := newCacheRouter()
+ invokers := makeInvokers("gray", "gray", "")
+ invokers[1].GetURL().SetParam("version", "v2")
+ withCache(p, invokers)
+
+ p.routerConfigs.Store(consumerUrl.GetParam(constant.ApplicationKey,
"")+constant.TagRouterRuleSuffix, global.RouterConfig{
+ Key: consumerUrl.Service() + constant.TagRouterRuleSuffix,
+ Force: boolPtr(false),
+ Enabled: boolPtr(true),
+ Valid: boolPtr(true),
+ Tags: []global.Tag{{
+ Name: "gray",
+ Match: []*common.ParamMatch{
+ {Key: "version", Value:
common.StringMatch{Exact: "v2"}},
+ },
+ }},
+ })
+
+ t.Run("match falls back to requestTag and filters correctly", func(t
*testing.T) {
+ attachments := map[string]any{constant.Tagkey: "gray"}
+ result := p.Route(invokers, consumerUrl,
invocation.NewRPCInvocation("GetUser", nil, attachments))
+ assert.Len(t, result, 1)
+ assert.Equal(t, "192.168.0.2:20000",
result[0].GetURL().Location)
+ })
+}
+
+func TestRouteBitmapMatchFallback(t *testing.T) {
+ p := newCacheRouter()
+ invokers := makeInvokers("gray", "gray", "")
+ withCache(p, invokers)
+
+ p.routerConfigs.Store(consumerUrl.GetParam(constant.ApplicationKey,
"")+constant.TagRouterRuleSuffix, global.RouterConfig{
+ Key: consumerUrl.Service() + constant.TagRouterRuleSuffix,
+ Force: boolPtr(false),
+ Enabled: boolPtr(true),
+ Valid: boolPtr(true),
+ Tags: []global.Tag{{
+ Name: "gray",
+ Match: []*common.ParamMatch{
+ {Key: "environment", Value:
common.StringMatch{Exact: "prod"}},
+ },
+ }},
+ })
+
+ t.Run("unindexed param key falls back to original path", func(t
*testing.T) {
+ attachments := map[string]any{constant.Tagkey: "gray"}
+ result := p.Route(invokers, consumerUrl,
invocation.NewRPCInvocation("GetUser", nil, attachments))
+ assert.Len(t, result, 1)
+ assert.Empty(t, result[0].GetURL().GetParam(constant.Tagkey,
""))
+ })
+}
+
+func TestRouteBitmapEquivalence(t *testing.T) {
+ // Same scenario via bitmap and original path should produce identical
results.
+ invokers := makeInvokers("gray", "gray", "")
+
+ pCache := newCacheRouter()
+ withCache(pCache, invokers)
+
pCache.routerConfigs.Store(consumerUrl.GetParam(constant.ApplicationKey,
"")+constant.TagRouterRuleSuffix, global.RouterConfig{
+ Key: consumerUrl.Service() + constant.TagRouterRuleSuffix,
+ Force: boolPtr(false),
+ Enabled: boolPtr(true),
+ Valid: boolPtr(true),
+ Tags: []global.Tag{{
+ Name: "gray",
+ Addresses: []string{"192.168.0.1:20000"},
+ }},
+ })
+
+ pFallback := newCacheRouter()
+
pFallback.routerConfigs.Store(consumerUrl.GetParam(constant.ApplicationKey,
"")+constant.TagRouterRuleSuffix, global.RouterConfig{
+ Key: consumerUrl.Service() + constant.TagRouterRuleSuffix,
+ Force: boolPtr(false),
+ Enabled: boolPtr(true),
+ Valid: boolPtr(true),
+ Tags: []global.Tag{{
+ Name: "gray",
+ Addresses: []string{"192.168.0.1:20000"},
+ }},
+ })
+
+ t.Run("bitmap and fallback produce same result", func(t *testing.T) {
+ attachments := map[string]any{constant.Tagkey: "gray"}
+ invoc := invocation.NewRPCInvocation("GetUser", nil,
attachments)
+
+ rCache := pCache.Route(invokers, consumerUrl, invoc)
+ rFallback := pFallback.Route(invokers, consumerUrl, invoc)
+
+ assert.Len(t, rCache, len(rFallback))
+ for i := range rCache {
+ assert.Equal(t, rCache[i].GetURL().Location,
rFallback[i].GetURL().Location)
+ }
+ })
+}
+
+// withCacheGen stores a cache built from invokers at the given generation,
mirroring what
+// RouterChain.SetInvokers does so a test can simulate a specific cache
generation.
+func withCacheGen(p *PriorityRouter, invokers []base.Invoker, generation
uint64) {
+ pool, _ := p.Pool(invokers)
+ p.cache.Store(&mockCache{invokers: invokers, pool: pool, generation:
generation})
+}
+
+// TestRouteGenerationGuard reproduces the race Alan flagged: the cache is
built on [a,b,c],
+// but Route is called with the chain's current snapshot [a]. When the
invocation's published
+// generation does not match the cache generation (a concurrent SetInvokers
rebuilt the cache
+// between the chain snapshot and this lookup), the bitmap fast path must be
skipped so the
+// route stays confined to the invokers passed in.
+func TestRouteGenerationGuard(t *testing.T) {
+ p := newCacheRouter()
+ full := makeInvokers("gray", "gray", "") // [a(gray), b(gray), c()]
+ withCacheGen(p, full, 5) // cache snapshot is
generation 5
+ subset := []base.Invoker{full[0]} // chain snapshot for this
call is just [a]
+
+ newInvoc := func(gen uint64) base.Invocation {
+ inv := invocation.NewRPCInvocation("GetUser", nil,
map[string]any{constant.Tagkey: "gray"})
+ inv.SetAttribute(constant.RouterChainCacheGeneration, gen)
+ return inv
+ }
+
+ t.Run("stale generation falls back to the invokers passed in", func(t
*testing.T) {
+ // snapshot generation 4 != cache generation 5 -> must not use
the [a,b,c] cache.
+ result := p.Route(subset, consumerUrl, newInvoc(4))
+ assert.Len(t, result, 1)
+ assert.Equal(t, full[0].GetURL().Location,
result[0].GetURL().Location)
+ })
+
+ t.Run("missing generation falls back to the invokers passed in", func(t
*testing.T) {
+ // No RouterChainCacheGeneration published -> default 0 !=
cache generation 5 -> fall back.
+ inv := invocation.NewRPCInvocation("GetUser", nil,
map[string]any{constant.Tagkey: "gray"})
+ result := p.Route(subset, consumerUrl, inv)
+ assert.Len(t, result, 1)
+ assert.Equal(t, full[0].GetURL().Location,
result[0].GetURL().Location)
+ })
+
+ t.Run("matching generation takes the bitmap path over the cached
snapshot", func(t *testing.T) {
+ // snapshot generation 5 == cache generation 5 -> bitmap path
over [a,b,c] -> [a,b].
+ result := p.Route(subset, consumerUrl, newInvoc(5))
+ assert.Len(t, result, 2)
+ })
+}
diff --git a/common/constant/key.go b/common/constant/key.go
index d680cf986..bab746aaf 100644
--- a/common/constant/key.go
+++ b/common/constant/key.go
@@ -357,6 +357,8 @@ const (
RouterScopeApplication = "application"
ForceKey = "force"
TrafficDisableKey = "trafficDisable"
+ RouterCacheDisable = "routerCacheDisable"
+ RouterChainCacheGeneration = "routerChainCacheGeneration"
Arguments = "arguments"
Attachments = "attachments"
Param = "param"
@@ -364,6 +366,12 @@ const (
Wildcard = "wildcard"
MeshRouterFactoryKey = "mesh"
DefaultRouteConditionSubSetWeight = 100
+
+ // Poolable router bitmap key prefixes
+ PoolKeyTagPrefix = "tag\x00"
+ PoolKeyAddrPrefix = "addr\x00"
+ PoolKeyPortPrefix = "port\x00"
+ PoolKeyAll = "*"
)
// Auth filter
