This is an automated email from the ASF dual-hosted git repository.
AlexStocks 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 59ee0cbc4 Feat/reuse snapshot consistent hash (#960)
59ee0cbc4 is described below
commit 59ee0cbc4b2bd4ca7b80acd9dc507774d8b603f5
Author: wm_03 <[email protected]>
AuthorDate: Mon Jul 6 20:17:16 2026 +0800
Feat/reuse snapshot consistent hash (#960)
* fix(cluster): Reuse consistent hash only for unchanged snapshots
* fix: remove redundant reuseHealthyConsistentHashFrom call in
withEndpointHealthForIDs
---
pkg/cluster/cluster.go | 83 ++++++++++++++-
pkg/cluster/cluster_test.go | 172 +++++++++++++++++++++++++++++++
pkg/server/cluster_manager_bench_test.go | 51 ++++++++-
3 files changed, 302 insertions(+), 4 deletions(-)
diff --git a/pkg/cluster/cluster.go b/pkg/cluster/cluster.go
index e64025c47..c6c27a70f 100644
--- a/pkg/cluster/cluster.go
+++ b/pkg/cluster/cluster.go
@@ -197,6 +197,7 @@ type EndpointSnapshot struct {
healthyByAddress map[string]bool
lbPolicy model.LbPolicyType
consistentHashOnce sync.Once
+ consistentHashMu sync.RWMutex
consistentHash model.LbConsistentHashView
consistentHashConfig model.ConsistentHash
}
@@ -248,6 +249,7 @@ func newEndpointSnapshot(config *model.ClusterConfig,
previous *EndpointSnapshot
healthy := endpointSnapshotHealth(snapshotEndpoint, address,
previous, inheritRuntimeHealth)
snapshot.addEndpoint(snapshotEndpoint, address, healthy)
}
+ snapshot.reuseHealthyConsistentHashFrom(previous)
return snapshot
}
@@ -395,9 +397,36 @@ func (s *EndpointSnapshot) HealthyConsistentHash()
model.LbConsistentHashView {
return nil
}
s.consistentHashOnce.Do(s.rebuildConsistentHash)
+ return s.cachedHealthyConsistentHash()
+}
+
+func (s *EndpointSnapshot) cachedHealthyConsistentHash()
model.LbConsistentHashView {
+ if s == nil {
+ return nil
+ }
+ s.consistentHashMu.RLock()
+ defer s.consistentHashMu.RUnlock()
return s.consistentHash
}
+func (s *EndpointSnapshot) seedHealthyConsistentHash(hash
model.LbConsistentHashView) {
+ if s == nil || hash == nil {
+ return
+ }
+ s.consistentHashOnce.Do(func() {
+ s.storeHealthyConsistentHash(hash)
+ })
+}
+
+func (s *EndpointSnapshot) storeHealthyConsistentHash(hash
model.LbConsistentHashView) {
+ if s == nil || hash == nil {
+ return
+ }
+ s.consistentHashMu.Lock()
+ defer s.consistentHashMu.Unlock()
+ s.consistentHash = hash
+}
+
// PickHealthyEndpoint gives request-path selectors a read-only view of healthy
// endpoints and clones only the selected endpoint before returning it. The
pick
// callback must not mutate or retain the endpoint view.
@@ -545,5 +574,57 @@ func (s *EndpointSnapshot) rebuildConsistentHash() {
if !ok {
return
}
- s.consistentHash =
model.ReadOnlyConsistentHash(newConsistentHash(s.consistentHashConfig,
s.healthy))
+ hash :=
model.ReadOnlyConsistentHash(newConsistentHash(s.consistentHashConfig,
s.healthy))
+ s.storeHealthyConsistentHash(hash)
+}
+
+func (s *EndpointSnapshot) reuseHealthyConsistentHashFrom(previous
*EndpointSnapshot) {
+ if !canReuseHealthyConsistentHash(previous, s) {
+ return
+ }
+ s.seedHealthyConsistentHash(previous.cachedHealthyConsistentHash())
+}
+
+func canReuseHealthyConsistentHash(previous, next *EndpointSnapshot) bool {
+ if previous == nil || next == nil {
+ return false
+ }
+ if previous.lbPolicy != next.lbPolicy {
+ return false
+ }
+ if _, ok := model.ConsistentHashInitMap[next.lbPolicy]; !ok {
+ return false
+ }
+ if !sameConsistentHashConfig(previous.consistentHashConfig,
next.consistentHashConfig) {
+ return false
+ }
+ return sameHealthyEndpointsForConsistentHash(previous, next)
+}
+
+func sameConsistentHashConfig(a, b model.ConsistentHash) bool {
+ return a.ReplicaNum == b.ReplicaNum &&
+ a.MaxVnodeNum == b.MaxVnodeNum &&
+ a.MaglevTableSize == b.MaglevTableSize
+}
+
+func sameHealthyEndpointsForConsistentHash(previous, next *EndpointSnapshot)
bool {
+ if previous == nil || next == nil {
+ return false
+ }
+ if len(previous.healthy) != len(next.healthy) {
+ return false
+ }
+ for i := range previous.healthy {
+ if !sameEndpointHostForConsistentHash(previous.healthy[i],
next.healthy[i]) {
+ return false
+ }
+ }
+ return true
+}
+
+func sameEndpointHostForConsistentHash(a, b *model.Endpoint) bool {
+ if a == nil || b == nil {
+ return a == b
+ }
+ return a.Address.Address == b.Address.Address && a.Address.Port ==
b.Address.Port
}
diff --git a/pkg/cluster/cluster_test.go b/pkg/cluster/cluster_test.go
index d8f7561d6..196c9da4a 100644
--- a/pkg/cluster/cluster_test.go
+++ b/pkg/cluster/cluster_test.go
@@ -189,6 +189,8 @@ func
TestClusterEndpointSnapshotExposesReadOnlyConsistentHash(t *testing.T) {
func TestClusterEndpointSnapshotBuildsConsistentHashLazily(t *testing.T) {
var builds int32
lbPolicy := model.LbPolicyType("test-lazy-hash")
+ // Mutates the global ConsistentHashInitMap; keep usage serial
+ // and do not add t.Parallel.
previousInit, hadPreviousInit := model.ConsistentHashInitMap[lbPolicy]
model.ConsistentHashInitMap[lbPolicy] = func(_ model.ConsistentHash,
endpoints []*model.Endpoint) model.LbConsistentHash {
atomic.AddInt32(&builds, 1)
@@ -247,6 +249,152 @@ func
TestClusterEndpointSnapshotBuildsConsistentHashLazily(t *testing.T) {
"build count is monotonic, but per-flap rebuild count is not
locked here")
}
+func TestClusterEndpointSnapshotReusesConsistentHashForUnchangedHealthySet(t
*testing.T) {
+ var builds int32
+ lbPolicy := registerCountingConsistentHash(t,
"test-reuse-hash-unchanged", &builds)
+
+ first := testEndpoint("ep-1", "127.0.0.1", 18080)
+ second := testEndpoint("ep-2", "127.0.0.1", 18081)
+ config := testCluster("snapshot-reuse-hash-unchanged", first, second)
+ config.LbStr = lbPolicy
+ config.ConsistentHash = model.ConsistentHash{ReplicaNum: 10,
MaxVnodeNum: 1023, MaglevTableSize: 521}
+
+ previous := NewCluster(config).EndpointSnapshot()
+ previousHash := previous.HealthyConsistentHash()
+ assert.NotNil(t, previousHash)
+ assert.Equal(t, int32(1), atomic.LoadInt32(&builds))
+
+ next := newEndpointSnapshot(config, previous, false)
+ assert.Equal(t, previousHash, next.HealthyConsistentHash(),
+ "equivalent snapshot should reuse the previous hash view")
+ assert.Equal(t, int32(1), atomic.LoadInt32(&builds),
+ "reused snapshot must not rebuild on first access")
+}
+
+func
TestClusterEndpointSnapshotDoesNotReuseConsistentHashWhenHealthChangesHealthySet(t
*testing.T) {
+ var builds int32
+ lbPolicy := registerCountingConsistentHash(t,
"test-reuse-hash-health-change", &builds)
+
+ first := testEndpoint("ep-1", "127.0.0.1", 18080)
+ second := testEndpoint("ep-2", "127.0.0.1", 18081)
+ config := testCluster("snapshot-reuse-hash-health-change", first,
second)
+ config.LbStr = lbPolicy
+
+ previous := NewCluster(config).EndpointSnapshot()
+ assert.NotNil(t, previous.HealthyConsistentHash())
+ assert.Equal(t, int32(1), atomic.LoadInt32(&builds))
+
+ next, ok :=
previous.withEndpointHealthForIDs(map[string]struct{}{second.ID: {}}, false)
+ assert.True(t, ok)
+ assert.NotNil(t, next.HealthyConsistentHash())
+ assert.Equal(t, int32(2), atomic.LoadInt32(&builds),
+ "health change that changes the healthy set must force a fresh
consistent hash")
+}
+
+func
TestClusterEndpointSnapshotDoesNotReuseConsistentHashWhenEndpointAddressChanges(t
*testing.T) {
+ var builds int32
+ lbPolicy := registerCountingConsistentHash(t,
"test-reuse-hash-address-change", &builds)
+
+ endpoint := testEndpoint("ep-1", "127.0.0.1", 18080)
+ config := testCluster("snapshot-reuse-hash-address-change", endpoint)
+ config.LbStr = lbPolicy
+
+ previous := NewCluster(config).EndpointSnapshot()
+ assert.NotNil(t, previous.HealthyConsistentHash())
+ assert.Equal(t, int32(1), atomic.LoadInt32(&builds))
+
+ config.Endpoints[0] = testEndpoint("ep-1", "127.0.0.2", 18080)
+ next := newEndpointSnapshot(config, previous, false)
+ assert.NotNil(t, next.HealthyConsistentHash())
+ assert.Equal(t, int32(2), atomic.LoadInt32(&builds),
+ "address change must force a fresh consistent hash")
+}
+
+func
TestClusterEndpointSnapshotDoesNotReuseConsistentHashWhenEndpointCountChanges(t
*testing.T) {
+ tests := []struct {
+ name string
+ endpoints func(first, second, third *model.Endpoint)
[]*model.Endpoint
+ }{
+ {
+ name: "add",
+ endpoints: func(first, second, third *model.Endpoint)
[]*model.Endpoint {
+ return []*model.Endpoint{first, second, third}
+ },
+ },
+ {
+ name: "delete",
+ endpoints: func(first, second, third *model.Endpoint)
[]*model.Endpoint {
+ return []*model.Endpoint{first}
+ },
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ var builds int32
+ lbPolicy := registerCountingConsistentHash(t,
"test-reuse-hash-count-change"+tt.name, &builds)
+
+ first := testEndpoint("ep-1", "127.0.0.1", 18080)
+ second := testEndpoint("ep-2", "127.0.0.1", 18081)
+ third := testEndpoint("ep-3", "127.0.0.1", 18082)
+ config :=
testCluster("snapshot-reuse-hash-count-change-"+tt.name, first, second)
+ config.LbStr = lbPolicy
+
+ previous := NewCluster(config).EndpointSnapshot()
+ assert.NotNil(t, previous.HealthyConsistentHash())
+ assert.Equal(t, int32(1), atomic.LoadInt32(&builds))
+
+ config.Endpoints = tt.endpoints(first, second, third)
+ next := newEndpointSnapshot(config, previous, false)
+ assert.NotNil(t, next.HealthyConsistentHash())
+ assert.Equal(t, int32(2), atomic.LoadInt32(&builds),
+ "count change must force a fresh consistent
hash")
+ })
+ }
+}
+
+func
TestClusterEndpointSnapshotDoesNotReuseConsistentHashWhenHashConfigChanges(t
*testing.T) {
+ var builds int32
+ lbPolicy := registerCountingConsistentHash(t,
"test-reuse-hash-config-change", &builds)
+
+ endpoint := testEndpoint("ep-1", "127.0.0.1", 18080)
+ config := testCluster("snapshot-reuse-hash-config-change", endpoint)
+ config.LbStr = lbPolicy
+ config.ConsistentHash = model.ConsistentHash{ReplicaNum: 10,
MaxVnodeNum: 1023, MaglevTableSize: 521}
+
+ previous := NewCluster(config).EndpointSnapshot()
+ assert.NotNil(t, previous.HealthyConsistentHash())
+ assert.Equal(t, int32(1), atomic.LoadInt32(&builds))
+
+ config.ConsistentHash.ReplicaNum = 20
+ next := newEndpointSnapshot(config, previous, false)
+ assert.NotNil(t, next.HealthyConsistentHash())
+ assert.Equal(t, int32(2), atomic.LoadInt32(&builds),
+ "hash config change must force a fresh consistent hash")
+}
+
+func TestClusterEndpointSnapshotReusesConsistentHashWhenMetadataChanges(t
*testing.T) {
+ var builds int32
+ lbPolicy := registerCountingConsistentHash(t,
"test-reuse-hash-metadata-change", &builds)
+
+ endpoint := testEndpoint("ep-1", "127.0.0.1", 18080)
+ endpoint.Metadata = map[string]string{"weight": "10"}
+ config := testCluster("snapshot-reuse-hash-metadata-change", endpoint)
+ config.LbStr = lbPolicy
+
+ previous := NewCluster(config).EndpointSnapshot()
+ assert.NotNil(t, previous.HealthyConsistentHash())
+ assert.Equal(t, int32(1), atomic.LoadInt32(&builds))
+
+ movedWeight := testEndpoint("ep-1", "127.0.0.1", 18080)
+ movedWeight.Metadata = map[string]string{"weight": "20"}
+ config.Endpoints[0] = movedWeight
+ next := newEndpointSnapshot(config, previous, false)
+ assert.NotNil(t, next.HealthyConsistentHash())
+ assert.Equal(t, int32(1), atomic.LoadInt32(&builds),
+ "metadata-only change with unchanged hosts must reuse the
previous consistent hash instead of rebuilding")
+}
+
func TestClusterEndpointSnapshotClonesConfigEndpointObjects(t *testing.T) {
endpoint := testSnapshotEndpointWithLLMMeta()
runtimeCluster := NewCluster(testCluster("snapshot-clone", endpoint))
@@ -559,6 +707,30 @@ func testSnapshotEndpointWithLLMMeta() *model.Endpoint {
return endpoint
}
+// Mutates the global ConsistentHashInitMap; keep usage serial
+// and do not add t.Parallel.
+func registerCountingConsistentHash(t *testing.T, name string, builds *int32)
model.LbPolicyType {
+ t.Helper()
+
+ lbPolicy := model.LbPolicyType(name)
+ previousInit, hadPreviousInit := model.ConsistentHashInitMap[lbPolicy]
+ model.ConsistentHashInitMap[lbPolicy] = func(_ model.ConsistentHash,
endpoints []*model.Endpoint) model.LbConsistentHash {
+ atomic.AddInt32(builds, 1)
+ if len(endpoints) == 0 {
+ return countingConsistentHash{}
+ }
+ return countingConsistentHash{host: endpoints[0].GetHost()}
+ }
+ t.Cleanup(func() {
+ if hadPreviousInit {
+ model.ConsistentHashInitMap[lbPolicy] = previousInit
+ return
+ }
+ delete(model.ConsistentHashInitMap, lbPolicy)
+ })
+ return lbPolicy
+}
+
type countingConsistentHash struct {
host string
}
diff --git a/pkg/server/cluster_manager_bench_test.go
b/pkg/server/cluster_manager_bench_test.go
index 63faf3db0..054b0a3ee 100644
--- a/pkg/server/cluster_manager_bench_test.go
+++ b/pkg/server/cluster_manager_bench_test.go
@@ -34,9 +34,10 @@ import (
var (
// Keep benchmark results live so the compiler cannot optimize the hot
path away.
- benchmarkEndpointSink *model.Endpoint
- benchmarkEndpointsSink []*model.Endpoint
- benchmarkClusterSink *model.ClusterConfig
+ benchmarkEndpointSink *model.Endpoint
+ benchmarkEndpointsSink []*model.Endpoint
+ benchmarkClusterSink *model.ClusterConfig
+ benchmarkConsistentHashSink model.LbConsistentHashView
)
type benchmarkHashPolicy string
@@ -241,6 +242,50 @@ func BenchmarkClusterConsistentHashResolve(b *testing.B) {
}
}
+func BenchmarkClusterConsistentHashSnapshotRefreshUnchangedHealthySet(b
*testing.B) {
+ for _, lbType := range
[]model.LbPolicyType{model.LoadBalancerRingHashing,
model.LoadBalancerMaglevHashing} {
+ for _, endpointCount := range []int{1, 32, 256, 1024} {
+ name := fmt.Sprintf("%s/endpoints=%d", lbType,
endpointCount)
+ b.Run(name, func(b *testing.B) {
+ b.Run("reuse-cached-previous", func(b
*testing.B) {
+
benchmarkConsistentHashSnapshotRefresh(b, lbType, endpointCount, true)
+ })
+ b.Run("rebuild-uncached-previous", func(b
*testing.B) {
+
benchmarkConsistentHashSnapshotRefresh(b, lbType, endpointCount, false)
+ })
+ })
+ }
+ }
+}
+
+func benchmarkConsistentHashSnapshotRefresh(
+ b *testing.B,
+ lbType model.LbPolicyType,
+ endpointCount int,
+ previousHashBuilt bool,
+) {
+ config := benchmarkClusterConfig("consistent-hash-refresh", lbType,
endpointCount, 0)
+ previous := cluster.NewCluster(config).EndpointSnapshot()
+ if previousHashBuilt {
+ benchmarkConsistentHashSink = previous.HealthyConsistentHash()
+ if benchmarkConsistentHashSink == nil {
+ b.Fatal("expected previous consistent hash")
+ }
+ }
+
+ b.ReportAllocs()
+ b.ResetTimer()
+ for i := 0; i < b.N; i++ {
+ runtimeCluster :=
cluster.NewClusterWithEndpointSnapshot(config, previous)
+ next := runtimeCluster.EndpointSnapshot()
+ benchmarkConsistentHashSink = next.HealthyConsistentHash()
+ }
+ b.StopTimer()
+ if benchmarkConsistentHashSink == nil {
+ b.Fatal("expected consistent hash")
+ }
+}
+
func benchmarkClusterManager(clusterCount int, endpointCount int, lbType
model.LbPolicyType) (*ClusterManager, []string) {
clusters := make([]*model.ClusterConfig, 0, clusterCount)
names := make([]string, 0, clusterCount)
