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 f49ac46a7 perf(loadbalancer): O(1) snapshot pick recheck via
healthy-by-ID index (#970)
f49ac46a7 is described below
commit f49ac46a7fcfa09bf5d87ef147f9225d7116a80d
Author: 承潜 <[email protected]>
AuthorDate: Sun Jul 26 13:36:13 2026 +0800
perf(loadbalancer): O(1) snapshot pick recheck via healthy-by-ID index
(#970)
* perf(loadbalancer): O(1) snapshot pick recheck via healthy-by-ID index
healthyEndpointFromSnapshot validated a balancer's pick by scanning the
healthy slice twice: a pointer-equality pass, then a sameEndpointIdentity
pass (with a per-element SocketAddress.Equal). Zero-copy balancers hit the
pointer pass early, but non-zero-copy snapshot balancers return an element
of the cloned HealthyEndpoints slice, whose pointer never matches the
context slice — so every pick fell through to the full O(N) identity scan.
Expose the snapshot's existing healthyEndpointByID index as an O(1),
allocation-free accessor (HealthyEndpointByIDForPick) and thread it into
PickContext via a small HealthyEndpointByIDLookup interface (declared in
loadbalancer so it keeps no dependency on pkg/cluster). The recheck now
resolves the single candidate by ID and applies the unchanged
sameEndpointIdentity rules to it; it falls back to the original scan when
no index is present (hand-built contexts) or the pick has no ID (the
empty-ID / placeholder-address path), preserving every existing rule
including the blank-domain wildcard and the reject-on-not-found defense.
BenchmarkHealthyEndpointFromSnapshot, 1024 healthy endpoints, non-zero-copy
return (ID at the far end):
identity-scan-without-index 1368 ns/op
identity-recheck-with-index 36 ns/op
The result clone (1 alloc) is unchanged; the win is the removed O(N) walk,
which matters for large clusters under high QPS.
Closes #955
* fix(loadbalancer): clear HealthyByID in defensive snapshot context
defensiveSnapshotPickContext clones the endpoint slices to isolate
non-zero-copy balancers from live snapshot pointers, but it did not clear
the newly-added HealthyByID field. This allowed untrusted balancers to call
HealthyByIDForPick and obtain un-cloned snapshot-owned endpoints, breaking
the memory-isolation guarantee that the defensive copy exists to enforce.
The recheck (healthyEndpointFromSnapshot) still works correctly because both
call sites pass the original context, whose HealthyByID remains intact. Only
the balancer-facing defensive copy is affected.
Current impact: zero (all in-tree balancers are zero-copy and none read
HealthyByID). This change closes a latent trust-boundary leak before
external
plugins can exploit it.
* perf(loadbalancer): keep by-id zero-copy pointer fast path
* chore(loadbalancer): address sonar maintainability issues
---
pkg/cluster/cluster.go | 14 +++
pkg/cluster/cluster_test.go | 25 ++++
pkg/cluster/loadbalancer/load_balancer.go | 58 ++++++++-
pkg/cluster/loadbalancer/load_balancer_test.go | 163 ++++++++++++++++++++++---
pkg/server/cluster_manager.go | 1 +
5 files changed, 242 insertions(+), 19 deletions(-)
diff --git a/pkg/cluster/cluster.go b/pkg/cluster/cluster.go
index 4868107e6..4dc783820 100644
--- a/pkg/cluster/cluster.go
+++ b/pkg/cluster/cluster.go
@@ -479,6 +479,20 @@ func (s *EndpointSnapshot) HealthyEndpointByID(endpointID
string) *model.Endpoin
return model.CloneEndpoint(s.healthyEndpointByID[endpointID])
}
+// HealthyEndpointByIDForPick returns the snapshot-internal healthy endpoint
for
+// endpointID without cloning, or nil when no healthy endpoint carries that ID.
+// Like HealthyEndpointsForPick, the returned endpoint is owned by the
snapshot:
+// callers MUST NOT mutate or retain it past the current pick. This O(1) lookup
+// backs the request-path recheck that validates a balancer's pick by ID
instead
+// of scanning the healthy slice; external callers should use
HealthyEndpointByID
+// (defensive clone).
+func (s *EndpointSnapshot) HealthyEndpointByIDForPick(endpointID string)
*model.Endpoint {
+ if s == nil {
+ return nil
+ }
+ return s.healthyEndpointByID[endpointID]
+}
+
func (s *EndpointSnapshot) withEndpointHealth(
endpointID, endpointAddress string,
healthy bool,
diff --git a/pkg/cluster/cluster_test.go b/pkg/cluster/cluster_test.go
index 196c9da4a..bd0b03f3a 100644
--- a/pkg/cluster/cluster_test.go
+++ b/pkg/cluster/cluster_test.go
@@ -52,6 +52,31 @@ func TestClusterEndpointSnapshotSeedsFromEndpointHealth(t
*testing.T) {
assert.Nil(t, snapshot.HealthyEndpointByID(unhealthy.ID))
}
+// TestClusterEndpointSnapshotHealthyEndpointByIDForPick locks the zero-copy
+// O(1) recheck accessor: it returns the snapshot-owned healthy endpoint
pointer
+// (the same one HealthyEndpointsForPick exposes, not a clone) and nil for
+// unhealthy or unknown IDs. The request-path recheck (issue #955) relies on
+// this being allocation-free and identity-stable.
+func TestClusterEndpointSnapshotHealthyEndpointByIDForPick(t *testing.T) {
+ healthy := testEndpoint("ep-1", "127.0.0.1", 18080)
+ unhealthy := testEndpoint("ep-2", "127.0.0.1", 18081)
+ unhealthy.UnHealthy = true
+
+ runtimeCluster :=
NewCluster(testCluster("snapshot-healthy-by-id-for-pick", healthy, unhealthy))
+ snapshot := runtimeCluster.EndpointSnapshot()
+
+ forPick := snapshot.HealthyEndpointByIDForPick(healthy.ID)
+ if assert.NotNil(t, forPick) {
+ // Same pointer the no-clone pick slice exposes.
+ assert.Same(t, snapshot.HealthyEndpointsForPick()[0], forPick)
+ }
+ assert.Nil(t, snapshot.HealthyEndpointByIDForPick(unhealthy.ID),
"unhealthy ID must not resolve")
+ assert.Nil(t, snapshot.HealthyEndpointByIDForPick("missing"), "unknown
ID must not resolve")
+
+ var nilSnapshot *EndpointSnapshot
+ assert.Nil(t, nilSnapshot.HealthyEndpointByIDForPick(healthy.ID), "nil
snapshot must be safe")
+}
+
func TestClusterEndpointSnapshotReturnsDefensiveEndpointSlices(t *testing.T) {
first := testEndpoint("ep-1", "127.0.0.1", 18080)
second := testEndpoint("ep-2", "127.0.0.1", 18081)
diff --git a/pkg/cluster/loadbalancer/load_balancer.go
b/pkg/cluster/loadbalancer/load_balancer.go
index fb1d3339c..295de13c7 100644
--- a/pkg/cluster/loadbalancer/load_balancer.go
+++ b/pkg/cluster/loadbalancer/load_balancer.go
@@ -49,6 +49,20 @@ type PickContext struct {
// Snapshot-aware balancers must treat endpoints as read-only and return
// the chosen endpoint without mutating or retaining it.
HealthyEndpoints []*model.Endpoint
+ // HealthyByID resolves a healthy snapshot endpoint by ID in O(1) for
the
+ // post-pick identity recheck. It is set by the snapshot-published pick
path;
+ // when nil (e.g. a hand-built context in a test), the recheck falls
back to
+ // scanning HealthyEndpoints. The returned endpoint is snapshot-owned
and
+ // must not be mutated or retained.
+ HealthyByID HealthyEndpointByIDForPicker
+}
+
+// HealthyEndpointByIDForPicker is the O(1) healthy-by-ID accessor the request
path
+// uses to recheck a balancer's pick without scanning the healthy slice. It is
+// satisfied by *cluster.EndpointSnapshot (HealthyEndpointByIDForPick);
declaring
+// it here keeps loadbalancer free of a dependency on the cluster package.
+type HealthyEndpointByIDForPicker interface {
+ HealthyEndpointByIDForPick(endpointID string) *model.Endpoint
}
type LoadBalancer interface {
@@ -134,7 +148,7 @@ func pickEndpoint(balancer LoadBalancer, context
PickContext, policy model.LbPol
snapshotContext = defensiveSnapshotPickContext(context)
}
endpoint :=
snapshotBalancer.HandlerWithSnapshot(snapshotContext, policy)
- return healthyEndpointFromSnapshot(endpoint,
context.HealthyEndpoints)
+ return healthyEndpointFromSnapshot(endpoint, context)
}
// Legacy balancers only understand ClusterConfig. Serialize this
@@ -192,26 +206,60 @@ func pickEndpoint(balancer LoadBalancer, context
PickContext, policy model.LbPol
if cursorAfter != cursorBefore {
atomic.AddUint32(&context.Config.PrePickEndpointIndex,
cursorAfter-cursorBefore)
}
- return healthyEndpointFromSnapshot(endpoint, context.HealthyEndpoints)
+ return healthyEndpointFromSnapshot(endpoint, context)
}
func defensiveSnapshotPickContext(context PickContext) PickContext {
defensive := context
defensive.AllEndpoints = model.CloneEndpoints(context.AllEndpoints)
defensive.HealthyEndpoints =
model.CloneEndpoints(context.HealthyEndpoints)
+ defensive.HealthyByID = nil
return defensive
}
-func healthyEndpointFromSnapshot(endpoint *model.Endpoint, healthyEndpoints
[]*model.Endpoint) *model.Endpoint {
+// healthyEndpointFromSnapshot validates that the balancer's pick is still a
+// member of the snapshot's healthy set and returns a defensive clone of the
+// snapshot-owned endpoint (never the balancer's possibly-cloned return).
+//
+// Resolution order:
+//
+// 1. O(1) by-ID recheck: when the pick carries an ID and the context exposes
+// the snapshot's healthy-by-ID index, resolve the single indexed
candidate.
+// If the indexed candidate is the exact balancer return, keep the
zero-copy
+// pointer fast path; otherwise apply sameEndpointIdentity to it. Every
+// snapshot endpoint carries a unique non-empty ID, so for an ID-bearing
pick
+// this is exactly the candidate the scan below would have found — without
+// the O(N) walk.
+// 2. Fallback scan: used when there is no ID index (e.g. a hand-built
context)
+// or the pick has no ID. Keeps the pointer-equality fast path for
zero-copy
+// balancers and the sameEndpointIdentity slow path (including the
+// empty-ID / placeholder-address rules) unchanged.
+//
+// Returns nil when the pick is not present in the healthy set; the validation
+// is deliberate and must not be skipped by blindly cloning the balancer
return.
+func healthyEndpointFromSnapshot(endpoint *model.Endpoint, context
PickContext) *model.Endpoint {
if endpoint == nil {
return nil
}
- for _, candidate := range healthyEndpoints {
+ if endpoint.ID != "" && context.HealthyByID != nil {
+ candidate :=
context.HealthyByID.HealthyEndpointByIDForPick(endpoint.ID)
+ if candidate == nil {
+ return nil
+ }
+ if candidate == endpoint {
+ return model.CloneEndpoint(candidate)
+ }
+ if sameEndpointIdentity(candidate, endpoint) {
+ return model.CloneEndpoint(candidate)
+ }
+ return nil
+ }
+ for _, candidate := range context.HealthyEndpoints {
if candidate == endpoint {
return model.CloneEndpoint(candidate)
}
}
- for _, candidate := range healthyEndpoints {
+ for _, candidate := range context.HealthyEndpoints {
if sameEndpointIdentity(candidate, endpoint) {
return model.CloneEndpoint(candidate)
}
diff --git a/pkg/cluster/loadbalancer/load_balancer_test.go
b/pkg/cluster/loadbalancer/load_balancer_test.go
index 496f19790..88b0eb4f9 100644
--- a/pkg/cluster/loadbalancer/load_balancer_test.go
+++ b/pkg/cluster/loadbalancer/load_balancer_test.go
@@ -46,6 +46,36 @@ type healthyOnlySnapshotLoadBalancer struct{}
var _ LoadBalancer = (*legacyLoadBalancer)(nil)
+// healthyByIDIndex is a test double for the snapshot's O(1) healthy-by-ID
+// lookup (satisfied in production by *cluster.EndpointSnapshot). It maps the
+// supplied healthy endpoints by ID, mirroring the snapshot's
healthyEndpointByID
+// index so the recheck fast path can be exercised without importing
pkg/cluster.
+type healthyByIDIndex map[string]*model.Endpoint
+
+func (h healthyByIDIndex) HealthyEndpointByIDForPick(endpointID string)
*model.Endpoint {
+ return h[endpointID]
+}
+
+func newHealthyByIDIndex(endpoints []*model.Endpoint) healthyByIDIndex {
+ index := make(healthyByIDIndex, len(endpoints))
+ for _, endpoint := range endpoints {
+ if endpoint != nil && endpoint.ID != "" {
+ index[endpoint.ID] = endpoint
+ }
+ }
+ return index
+}
+
+// snapshotRecheckContext builds the PickContext the request path passes to
+// healthyEndpointFromSnapshot: the healthy slice plus the O(1) by-ID index
over
+// the same endpoints, matching how cluster_manager wires a real snapshot.
+func snapshotRecheckContext(healthy []*model.Endpoint) PickContext {
+ return PickContext{
+ HealthyEndpoints: healthy,
+ HealthyByID: newHealthyByIDIndex(healthy),
+ }
+}
+
type blockingLegacyLoadBalancer struct {
entered chan int
release chan struct{}
@@ -327,6 +357,21 @@ func
TestNeedsAllEndpointsKeepsCompatibilityForUnmarkedSnapshotLoadBalancer(t *t
assert.False(t, NeedsAllEndpoints(healthyOnlySnapshotLoadBalancer{}))
}
+func TestDefensiveSnapshotPickContextClearsHealthyByID(t *testing.T) {
+ healthy := []*model.Endpoint{
+ {ID: "ep-1", Address: model.SocketAddress{Address: "127.0.0.1",
Port: 8080}},
+ }
+ originalContext := PickContext{
+ HealthyEndpoints: healthy,
+ HealthyByID: newHealthyByIDIndex(healthy),
+ }
+
+ defensive := defensiveSnapshotPickContext(originalContext)
+
+ assert.Nil(t, defensive.HealthyByID, "defensive context must not expose
live snapshot pointers")
+ assert.NotNil(t, originalContext.HealthyByID, "original context must be
unchanged")
+}
+
func TestPickEndpointSerializesLegacyLoadBalancerHandlers(t *testing.T) {
harness := newLegacyPickHarness(t)
pickContext := newLegacyPickContext("blocking-legacy-load-balancer",
"first")
@@ -529,7 +574,7 @@ func
TestHealthyEndpointFromSnapshotAcceptsResolvedAddressForBlankPlaceholder(t
Address: model.SocketAddress{Address: "127.0.0.1", Port: 8080},
}
- got := healthyEndpointFromSnapshot(balancerReturn, healthyEndpoints)
+ got := healthyEndpointFromSnapshot(balancerReturn,
snapshotRecheckContext(healthyEndpoints))
if !assert.NotNil(t, got, "balancer return with same ID as snapshot
placeholder must match via wildcard") {
return
}
@@ -546,7 +591,27 @@ func
TestHealthyEndpointFromSnapshotPointerFastPathReturnsClone(t *testing.T) {
},
}
- got := healthyEndpointFromSnapshot(endpoint,
[]*model.Endpoint{endpoint})
+ // No HealthyByID index: exercise the fallback scan's pointer-equality
fast
+ // path (the branch a zero-copy balancer hits when no ID index is
present).
+ got := healthyEndpointFromSnapshot(endpoint,
PickContext{HealthyEndpoints: []*model.Endpoint{endpoint}})
+
+ if !assert.NotNil(t, got) {
+ return
+ }
+ assert.Equal(t, endpoint, got)
+ assert.NotSame(t, endpoint, got, "request path must not return the
snapshot-owned endpoint pointer")
+}
+
+func TestHealthyEndpointFromSnapshotByIDPointerFastPathReturnsClone(t
*testing.T) {
+ endpoint := &model.Endpoint{
+ ID: "zero-copy-id",
+ Address: model.SocketAddress{
+ Address: "127.0.0.1",
+ Port: 8080,
+ },
+ }
+
+ got := healthyEndpointFromSnapshot(endpoint,
snapshotRecheckContext([]*model.Endpoint{endpoint}))
if !assert.NotNil(t, got) {
return
@@ -572,10 +637,68 @@ func
TestHealthyEndpointFromSnapshotRejectsMismatchedRealAddress(t *testing.T) {
Address: model.SocketAddress{Address: "10.0.0.1", Port: 9090},
}
- got := healthyEndpointFromSnapshot(balancerReturn, healthyEndpoints)
+ got := healthyEndpointFromSnapshot(balancerReturn,
snapshotRecheckContext(healthyEndpoints))
assert.Nil(t, got, "real-address mismatch must not match even when IDs
agree")
}
+// TestHealthyEndpointFromSnapshotByIDMatchesScan locks the O(1) by-ID recheck
+// to the fallback scan: for every case, resolving through the HealthyByID
index
+// must produce the same accept/reject decision as scanning HealthyEndpoints,
so
+// the optimization cannot silently change which picks are admitted.
+func TestHealthyEndpointFromSnapshotByIDMatchesScan(t *testing.T) {
+ realAddr := model.SocketAddress{Address: "127.0.0.1", Port: 8080}
+ healthy := []*model.Endpoint{
+ {ID: "ep-real", Address: realAddr},
+ {ID: "ep-placeholder", Address: model.SocketAddress{Domains:
[]string{""}}},
+ }
+
+ cases := []struct {
+ name string
+ ret *model.Endpoint
+ wantMatchID string // "" means expect nil
+ }{
+ {
+ name: "defensive copy of real endpoint matches
by ID",
+ ret: &model.Endpoint{ID: "ep-real", Address:
realAddr},
+ wantMatchID: "ep-real",
+ },
+ {
+ name: "placeholder wildcard accepts resolved
address",
+ ret: &model.Endpoint{ID: "ep-placeholder",
Address: model.SocketAddress{Address: "10.0.0.9", Port: 9090}},
+ wantMatchID: "ep-placeholder",
+ },
+ {
+ name: "same ID different real address rejected",
+ ret: &model.Endpoint{ID: "ep-real", Address:
model.SocketAddress{Address: "10.0.0.1", Port: 9090}},
+ wantMatchID: "",
+ },
+ {
+ name: "unknown ID rejected",
+ ret: &model.Endpoint{ID: "ep-missing", Address:
realAddr},
+ wantMatchID: "",
+ },
+ }
+
+ for _, tc := range cases {
+ t.Run(tc.name, func(t *testing.T) {
+ withIndex := healthyEndpointFromSnapshot(tc.ret,
snapshotRecheckContext(healthy))
+ scanOnly := healthyEndpointFromSnapshot(tc.ret,
PickContext{HealthyEndpoints: healthy})
+
+ if tc.wantMatchID == "" {
+ assert.Nil(t, withIndex, "by-ID recheck must
reject")
+ assert.Nil(t, scanOnly, "scan must reject")
+ return
+ }
+ if assert.NotNil(t, withIndex, "by-ID recheck must
accept") {
+ assert.Equal(t, tc.wantMatchID, withIndex.ID)
+ }
+ if assert.NotNil(t, scanOnly, "scan must accept") {
+ assert.Equal(t, tc.wantMatchID, scanOnly.ID)
+ }
+ })
+ }
+}
+
func BenchmarkHealthyEndpointFromSnapshot(b *testing.B) {
const endpointCount = 1024
healthyEndpoints := make([]*model.Endpoint, endpointCount)
@@ -590,20 +713,32 @@ func BenchmarkHealthyEndpointFromSnapshot(b *testing.B) {
}
zeroCopyEndpoint := healthyEndpoints[endpointCount-1]
defensiveCopyEndpoint := model.CloneEndpoint(zeroCopyEndpoint)
+ scanContext := PickContext{HealthyEndpoints: healthyEndpoints}
+ indexContext := snapshotRecheckContext(healthyEndpoints)
+ // Zero-copy balancers hit the pointer-equality fast path; the index
does
+ // not change that branch, but measure it to confirm no regression.
b.Run("pointer-eq-fast-path", func(b *testing.B) {
- for i := 0; i < b.N; i++ {
- if healthyEndpointFromSnapshot(zeroCopyEndpoint,
healthyEndpoints) == nil {
- b.Fatal("expected match")
- }
- }
+ benchmarkHealthyEndpointFromSnapshot(b, zeroCopyEndpoint,
scanContext)
})
- b.Run("identity-scan", func(b *testing.B) {
- for i := 0; i < b.N; i++ {
- if healthyEndpointFromSnapshot(defensiveCopyEndpoint,
healthyEndpoints) == nil {
- b.Fatal("expected match")
- }
- }
+ // Non-zero-copy balancers return a cloned endpoint whose pointer is
not in
+ // the snapshot slice. Without an ID index this falls through to the
full
+ // O(N) sameEndpointIdentity scan; the index turns it into an O(1)
lookup.
+ b.Run("identity-scan-without-index", func(b *testing.B) {
+ benchmarkHealthyEndpointFromSnapshot(b, defensiveCopyEndpoint,
scanContext)
+ })
+
+ b.Run("identity-recheck-with-index", func(b *testing.B) {
+ benchmarkHealthyEndpointFromSnapshot(b, defensiveCopyEndpoint,
indexContext)
})
}
+
+func benchmarkHealthyEndpointFromSnapshot(b *testing.B, endpoint
*model.Endpoint, context PickContext) {
+ b.Helper()
+ for i := 0; i < b.N; i++ {
+ if healthyEndpointFromSnapshot(endpoint, context) == nil {
+ b.Fatal("expected match")
+ }
+ }
+}
diff --git a/pkg/server/cluster_manager.go b/pkg/server/cluster_manager.go
index f7284e9c7..fb3afb12b 100644
--- a/pkg/server/cluster_manager.go
+++ b/pkg/server/cluster_manager.go
@@ -308,6 +308,7 @@ func (cm *ClusterManager) pickOneEndpoint(runtimeCluster
*cluster.Cluster, polic
HealthyConsistentHash: snapshot.HealthyConsistentHash(),
AllEndpoints: allEndpoints,
HealthyEndpoints: healthyEndpoints,
+ HealthyByID: snapshot,
}, policy)
}
