This is an automated email from the ASF dual-hosted git repository.
Similarityoung pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/dubbo-go-pixiu.git
The following commit(s) were added to refs/heads/develop by this push:
new 3687feded perf(cluster): defer Config consistent-hash rebuild to
legacy pick path (#967)
3687feded is described below
commit 3687feded23a26c388733f46107c64c9af0f6289
Author: 承潜 <[email protected]>
AuthorDate: Sun Jul 26 13:19:22 2026 +0800
perf(cluster): defer Config consistent-hash rebuild to legacy pick path
(#967)
* perf(cluster): defer Config consistent-hash rebuild to legacy pick path
prepareClusterConfig rebuilt Config.ConsistentHash.Hash eagerly on every
AddCluster/UpdateCluster/SetEndpoint/DeleteEndpoint. After the snapshot
work in #932, that field is only read by the legacy (non-snapshot) pick
path; all bundled balancers are snapshot-aware and use the snapshot's own
healthy hash. So the eager rebuild is dead work on the common path, and
for a large Maglev table it is milliseconds of wasted CPU on every
endpoint churn in flapping service-discovery environments.
Invalidate the Config-level hash (set to nil) in prepareClusterConfig and
build it lazily via ClusterConfig.EnsureConsistentHash, called from the
legacy branch of pickEndpoint under the existing legacyPickMu. The
in-Handler nil fallback in Maglev/RingHash stays as a safety net.
EnsureConsistentHash uses a nil-check rather than a sync.Once field: a
value Once on ClusterConfig trips go vet copylocks at the legacy path's
`config := *context.Config` shallow copy, and the legacy path is already
serialized by legacyPickMu so no extra synchronization is needed. A fresh
ClusterConfig from CloneStore (whose Hash interface does not survive the
yaml round-trip) rebuilds independently on its first legacy pick.
BenchmarkSetEndpoint, 64-endpoint Maglev cluster, MaglevTableSize=65537:
before 1087 ns/op 1721 B/op
after 413 ns/op 481 B/op
Closes #957
* test(cluster): fix BenchmarkSetEndpoint to actually exercise the rebuild
path
The benchmark reused a single update set, so after the first 64-iteration
cycle every SetEndpoint hit the content-equal idempotent short-circuit and
returned before prepareClusterConfig. Both the eager and deferred builds
therefore measured the same ~410 ns fast path, not the table rebuild the
benchmark claimed to compare.
Alternate between two metadata generations so every SetEndpoint is a genuine
in-place replace that runs prepareClusterConfig on each call. Corrected
benchstat (-count=8, 64-endpoint Maglev, MaglevTableSize=65537):
before (eager) 10505 us/op 18016 KiB/op 864 allocs/op
after (deferred) 53.12 us/op 90.21 KiB/op 614 allocs/op
-99.5% -99.5% -28.9% (p=0.000)
Also note in pickEndpoint that the legacy EnsureConsistentHash branch is
unreached in-tree (all bundled balancers are snapshot-aware); the win comes
from dropping the eager rebuild, and the lazy build is out-of-tree compat.
* fix(cluster): preserve custom consistent hashes
---
pkg/cluster/loadbalancer/load_balancer.go | 12 ++++++
pkg/model/cluster.go | 24 +++++++++++
pkg/model/cluster_test.go | 31 ++++++++++++++
pkg/server/cluster_manager.go | 27 ++++++++++---
pkg/server/cluster_manager_bench_test.go | 46 +++++++++++++++++++++
pkg/server/cluster_manager_test.go | 67 +++++++++++++++++++++++++++++++
6 files changed, 202 insertions(+), 5 deletions(-)
diff --git a/pkg/cluster/loadbalancer/load_balancer.go
b/pkg/cluster/loadbalancer/load_balancer.go
index c07af682e..fb1d3339c 100644
--- a/pkg/cluster/loadbalancer/load_balancer.go
+++ b/pkg/cluster/loadbalancer/load_balancer.go
@@ -167,6 +167,18 @@ func pickEndpoint(balancer LoadBalancer, context
PickContext, policy model.LbPol
legacyPickMu.Lock()
defer legacyPickMu.Unlock()
+ // Build the Config-level consistent hash lazily on the first legacy
pick.
+ // prepareClusterConfig no longer rebuilds it eagerly; snapshot
balancers
+ // never read it. Safe under legacyPickMu, which serializes this path.
+ //
+ // In-tree this branch is currently unreached: every bundled balancer
+ // implements HandlerWithSnapshot and is dispatched above. The actual
win
+ // is dropping the eager rebuild from prepareClusterConfig;
EnsureConsistentHash
+ // exists so an out-of-tree balancer that only implements the legacy
Handler
+ // still observes a non-nil Config.ConsistentHash.Hash, preserving the
old
+ // contract.
+ context.Config.EnsureConsistentHash()
+
allEndpoints := context.AllEndpoints
if allEndpoints == nil {
allEndpoints = context.HealthyEndpoints
diff --git a/pkg/model/cluster.go b/pkg/model/cluster.go
index 12ee82208..dda7e7983 100644
--- a/pkg/model/cluster.go
+++ b/pkg/model/cluster.go
@@ -132,6 +132,30 @@ func (c *ClusterConfig) CreateConsistentHash() {
}
}
+// HasConsistentHashFactory reports whether this cluster's load-balancer policy
+// can rebuild Config-level consistent-hash state from endpoints.
+func (c *ClusterConfig) HasConsistentHashFactory() bool {
+ _, ok := ConsistentHashInitMap[c.LbStr]
+ return ok
+}
+
+// EnsureConsistentHash lazily builds the Config-level consistent hash on first
+// use and reuses it afterwards. It exists only for the legacy (non-snapshot)
+// pick path: snapshot-aware balancers ignore Config.ConsistentHash.Hash and
+// read the snapshot's own healthy hash instead. Building here rather than
+// eagerly on every config mutation avoids rebuilding a large Maglev table on
+// each endpoint add/remove in high-churn discovery environments.
+//
+// Not safe for concurrent use. The legacy pick path serializes callers under
+// loadbalancer's package lock; a fresh ClusterConfig (e.g. a CloneStore copy,
+// whose Hash interface does not survive the yaml round-trip) rebuilds
+// independently on its first legacy pick.
+func (c *ClusterConfig) EnsureConsistentHash() {
+ if c.ConsistentHash.Hash == nil {
+ c.CreateConsistentHash()
+ }
+}
+
func (e Endpoint) GetHost() string {
return fmt.Sprintf("%s:%d", e.Address.Address, e.Address.Port)
}
diff --git a/pkg/model/cluster_test.go b/pkg/model/cluster_test.go
index aeb26495d..b53a6fa6c 100644
--- a/pkg/model/cluster_test.go
+++ b/pkg/model/cluster_test.go
@@ -77,6 +77,37 @@ func TestClusterConfig_CreateConsistentHashRegistersHash(t
*testing.T) {
assert.Equal(t, "127.0.0.1:20880", hash)
}
+func TestClusterConfig_HasConsistentHashFactory(t *testing.T) {
+ assert.True(t, (&model.ClusterConfig{LbStr:
model.LoadBalancerRingHashing}).HasConsistentHashFactory())
+ assert.False(t, (&model.ClusterConfig{LbStr:
model.LbPolicyType("CustomHash")}).HasConsistentHashFactory())
+}
+
+func TestClusterConfig_EnsureConsistentHashBuildsOnce(t *testing.T) {
+ cluster := &model.ClusterConfig{
+ LbStr: model.LoadBalancerRingHashing,
+ ConsistentHash: model.ConsistentHash{
+ ReplicaNum: 32,
+ MaxVnodeNum: 1023,
+ },
+ Endpoints: []*model.Endpoint{
+ {
+ ID: "ep-1",
+ Address: model.SocketAddress{
+ Address: "127.0.0.1",
+ Port: 20880,
+ },
+ },
+ },
+ }
+
+ cluster.EnsureConsistentHash()
+ first := cluster.ConsistentHash.Hash
+ cluster.EnsureConsistentHash()
+
+ require.NotNil(t, first)
+ assert.Same(t, first, cluster.ConsistentHash.Hash)
+}
+
func TestClusterConfig_PrePickEndpointIndexIsRuntimeOnly(t *testing.T) {
cluster := &model.ClusterConfig{
Name: "runtime-cursor",
diff --git a/pkg/server/cluster_manager.go b/pkg/server/cluster_manager.go
index 09a4dd7a5..f7284e9c7 100644
--- a/pkg/server/cluster_manager.go
+++ b/pkg/server/cluster_manager.go
@@ -358,18 +358,35 @@ func (s *ClusterStore) AddCluster(c *model.ClusterConfig)
{
}
// prepareClusterConfig clones operator-supplied endpoints, then rebuilds
-// endpoint defaults and hash from current endpoints.
+// endpoint defaults and invalidates the Config-level consistent hash from the
+// current endpoints. It is the external-input boundary: callers pass
+// operator-owned endpoint pointers, so it deep-clones before defaulting IDs
and
+// names. Store-owned mutation paths call prepareOwnedClusterConfig instead to
+// avoid a second full clone.
func (s *ClusterStore) prepareClusterConfig(c *model.ClusterConfig) {
c.Endpoints = model.CloneEndpoints(c.Endpoints)
s.prepareOwnedClusterConfig(c)
}
-// prepareOwnedClusterConfig rebuilds endpoint defaults and hash for endpoints
-// already owned by ClusterStore. Callers must not pass operator-owned endpoint
-// pointers here; use prepareClusterConfig at external input boundaries.
+// prepareOwnedClusterConfig rebuilds endpoint defaults and invalidates the
+// Config-level consistent hash for endpoints already owned by ClusterStore.
+// Callers must not pass operator-owned endpoint pointers here; use
+// prepareClusterConfig at external input boundaries.
+//
+// The hash is only read by the legacy (non-snapshot) pick path and is rebuilt
+// lazily there via ClusterConfig.EnsureConsistentHash, so eagerly rebuilding
it
+// on every AddCluster/UpdateCluster/SetEndpoint/DeleteEndpoint is dead work
for
+// the common snapshot path (and expensive for large Maglev tables under
+// service-discovery churn). Setting it to nil here keeps the legacy path
correct
+// after endpoint changes: the next legacy pick rebuilds from the current
+// endpoints instead of serving a stale ring. For unregistered/custom policies,
+// preserve any programmatically supplied hash because there is no factory
+// available to rebuild it later.
func (s *ClusterStore) prepareOwnedClusterConfig(c *model.ClusterConfig) {
s.assembleClusterEndpoints(c)
- c.CreateConsistentHash()
+ if c.HasConsistentHashFactory() {
+ c.ConsistentHash.Hash = nil
+ }
}
// assembleClusterEndpoints assembles the cluster endpoints by assigning stable
diff --git a/pkg/server/cluster_manager_bench_test.go
b/pkg/server/cluster_manager_bench_test.go
index 054b0a3ee..523d14e7a 100644
--- a/pkg/server/cluster_manager_bench_test.go
+++ b/pkg/server/cluster_manager_bench_test.go
@@ -242,6 +242,52 @@ func BenchmarkClusterConsistentHashResolve(b *testing.B) {
}
}
+// BenchmarkSetEndpoint measures the per-mutation cost of a registry update on
a
+// Maglev cluster with a large lookup table — the metadata-only re-registration
+// a high-churn discovery environment performs on every heartbeat. Each update
+// keeps the endpoint's address fixed (no healthcheck restart) and only flips
+// metadata, so SetEndpoint takes the in-place replace path and
prepareClusterConfig
+// runs on every iteration. Before this change that meant repopulating the
whole
+// 65537-slot table per mutation; after it, the snapshot path skips the
rebuild.
+//
+// The two metadata generations matter: SetEndpoint short-circuits to an
+// idempotent no-op when the incoming endpoint is content-equal to the slot's
+// current occupant. Alternating generations each cycle guarantees every visit
+// to a slot differs from its previous occupant, so the benchmark exercises the
+// rebuild path on every call instead of collapsing into the fast path after
the
+// first cycle. Inputs are pre-built so no per-iteration input allocation leaks
+// into the measurement.
+func BenchmarkSetEndpoint(b *testing.B) {
+ const (
+ endpointCount = 64
+ generations = 2
+ )
+ cluster := testCluster("set-endpoint-maglev",
model.LoadBalancerMaglevHashing, nil)
+ cluster.ConsistentHash = model.ConsistentHash{MaglevTableSize: 65537}
+ for i := 0; i < endpointCount; i++ {
+ cluster.Endpoints = append(cluster.Endpoints,
testEndpoint(fmt.Sprintf("ep-%d", i), "127.0.0.1", 20000+i))
+ }
+ cm := testClusterManager(cluster)
+ defer stopStoreRuntimes(cm.store)
+
+ updates := make([][]*model.Endpoint, generations)
+ for g := range updates {
+ updates[g] = make([]*model.Endpoint, endpointCount)
+ for i := range updates[g] {
+ endpoint := testEndpoint(fmt.Sprintf("ep-%d", i),
"127.0.0.1", 20000+i)
+ endpoint.Metadata = map[string]string{"generation":
fmt.Sprintf("gen-%d", g)}
+ updates[g][i] = endpoint
+ }
+ }
+
+ b.ReportAllocs()
+ b.ResetTimer()
+ for i := 0; i < b.N; i++ {
+ generation := (i / endpointCount) % generations
+ cm.SetEndpoint(cluster.Name,
updates[generation][i%endpointCount])
+ }
+}
+
func BenchmarkClusterConsistentHashSnapshotRefreshUnchangedHealthySet(b
*testing.B) {
for _, lbType := range
[]model.LbPolicyType{model.LoadBalancerRingHashing,
model.LoadBalancerMaglevHashing} {
for _, endpointCount := range []int{1, 32, 256, 1024} {
diff --git a/pkg/server/cluster_manager_test.go
b/pkg/server/cluster_manager_test.go
index 67f038c60..87ece3de5 100644
--- a/pkg/server/cluster_manager_test.go
+++ b/pkg/server/cluster_manager_test.go
@@ -433,6 +433,9 @@ func
TestClusterManager_SetEndpointExplicitSameIDDifferentAddressRebuildsConsist
assert.Equal(t, "ep-1", endpoints[0].ID)
assert.Equal(t, "127.0.0.2",
endpoints[0].Address.Address)
+ // The Config-level hash is built lazily on the legacy
pick path now,
+ // so trigger the build before inspecting it directly.
+ cm.store.Config[0].EnsureConsistentHash()
hash := cm.store.Config[0].ConsistentHash.Hash
if !assert.NotNil(t, hash) {
return
@@ -481,6 +484,9 @@ func
TestClusterManager_DeleteEndpointRepairsRuntimeAndConsistentHash(t *testing
assert.NotSame(t, remainingEndpoint, config.Endpoints[0])
}
+ // The Config-level hash is built lazily on the legacy pick path now,
+ // so trigger the build before inspecting it directly.
+ config.EnsureConsistentHash()
hash := config.ConsistentHash.Hash
if !assert.NotNil(t, hash) {
return
@@ -494,6 +500,67 @@ func
TestClusterManager_DeleteEndpointRepairsRuntimeAndConsistentHash(t *testing
assert.Contains(t, hosts, remainingHost)
}
+// countingFixedHash is a minimal model.LbConsistentHash fixture for the
+// deferred-rebuild test; only construction is observed (via the registered
+// init func's counter), so the lookup methods are stubs.
+type countingFixedHash struct{}
+
+func (countingFixedHash) Hash(string) uint32 { return 0 }
+func (countingFixedHash) Get(string) (string, error) { return "", nil }
+func (countingFixedHash) GetHash(uint32) (string, error) { return "", nil }
+func (countingFixedHash) Add(string) {}
+func (countingFixedHash) Remove(string) bool { return false }
+
+// TestClusterManager_SetEndpointDefersConsistentHashRebuild verifies that the
+// Config-level consistent hash is no longer rebuilt eagerly on every config
+// mutation. SetEndpoint churn must trigger zero hash builds (the snapshot pick
+// path never reads Config.ConsistentHash.Hash); the hash is built lazily,
once,
+// only when the legacy path calls EnsureConsistentHash.
+func TestClusterManager_SetEndpointDefersConsistentHashRebuild(t *testing.T) {
+ const deferLbPolicy model.LbPolicyType = "DeferRebuildCountingHash"
+ var buildCount int32
+ model.ConsistentHashInitMap[deferLbPolicy] = func(model.ConsistentHash,
[]*model.Endpoint) model.LbConsistentHash {
+ atomic.AddInt32(&buildCount, 1)
+ return countingFixedHash{}
+ }
+ defer delete(model.ConsistentHashInitMap, deferLbPolicy)
+
+ config := testCluster("defer-hash-rebuild", deferLbPolicy,
[]*model.Endpoint{
+ testEndpoint("ep-1", "127.0.0.1", 19360),
+ })
+ cm := testClusterManager(config)
+ defer stopStoreRuntimes(cm.store)
+
+ // Initial assembly must not build the Config-level hash.
+ assert.Equal(t, int32(0), atomic.LoadInt32(&buildCount), "AddCluster
must not eagerly build the consistent hash")
+
+ for i := 0; i < 5; i++ {
+ cm.SetEndpoint(config.Name, testEndpoint("ep-1", "127.0.0.2",
19361+i))
+ }
+ assert.Equal(t, int32(0), atomic.LoadInt32(&buildCount), "SetEndpoint
churn must not build the Config-level hash on the snapshot path")
+
+ stored := cm.store.Config[0]
+ stored.EnsureConsistentHash()
+ stored.EnsureConsistentHash()
+ assert.Equal(t, int32(1), atomic.LoadInt32(&buildCount), "legacy path
must build the hash exactly once and reuse it")
+}
+
+func
TestClusterManager_PrepareClusterConfigPreservesCustomHashWithoutFactory(t
*testing.T) {
+ customHash := &countingFixedHash{}
+ config := testCluster("custom-hash-preserve",
model.LbPolicyType("UnregisteredConsistentHash"), []*model.Endpoint{
+ testEndpoint("ep-1", "127.0.0.1", 19370),
+ })
+ config.ConsistentHash.Hash = customHash
+
+ cm := testClusterManager(config)
+ defer stopStoreRuntimes(cm.store)
+
+ assert.Same(t, customHash, cm.store.Config[0].ConsistentHash.Hash)
+
+ cm.SetEndpoint(config.Name, testEndpoint("ep-1", "127.0.0.2", 19371))
+ assert.Same(t, customHash, cm.store.Config[0].ConsistentHash.Hash)
+}
+
func TestClusterManager_Race_RoundRobinPickEndpoint(t *testing.T) {
cluster := testCluster("race-round-robin",
model.LoadBalancerRoundRobin, []*model.Endpoint{
testEndpoint("ep-1", "127.0.0.1", 19100),
