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.git
The following commit(s) were added to refs/heads/develop by this push:
new 0f71ba09e fix(cluster): eliminate data race on StickyInvoker with
atomic.Value (#3439)
0f71ba09e is described below
commit 0f71ba09e67bcaf84a089ae098965c93c933ed57
Author: aias00 <[email protected]>
AuthorDate: Sat Jul 25 18:20:24 2026 -0700
fix(cluster): eliminate data race on StickyInvoker with atomic.Value (#3439)
* fix(cluster): eliminate data race on StickyInvoker with atomic.Value
StickyInvoker in BaseClusterInvoker was a plain field read/written by
multiple goroutines (e.g. failback retry goroutine + Invoke goroutine)
without any synchronization, causing a data race on the RPC hot path.
Replace the unexported StickyInvoker field with an atomic.Value,
accessed via getStickyInvoker()/setStickyInvoker() helpers. This
eliminates the race while keeping the lock-free read path fast.
Key changes:
- StickyInvoker base.Invoker -> stickyInvoker atomic.Value (unexported)
- Add getStickyInvoker() / setStickyInvoker() helpers
- Update IsAvailable() and DoSelect() to use atomic accessors
- DoSelect reads sticky invoker into a local variable to avoid
redundant atomic loads and ensure consistent logic within one call
- Add concurrent tests (TestStickyConcurrentDoSelect,
TestStickyConcurrentIsAvailableAndDoSelect) that pass with -race
Co-Authored-By: Claude <[email protected]>
* fix(cluster): use wrapper type for atomic.Value to prevent nil Store and
type mismatch panics
Address Copilot review feedback:
1. atomic.Value.Store(nil) panics - wrap base.Invoker in
stickyInvokerWrapper
struct so nil invoker is stored as {invoker: nil} (non-nil wrapper)
2. atomic.Value requires consistent concrete type - the wrapper struct
ensures the same type is always stored regardless of Invoker impl
3. Fix test TestStickyConcurrentIsAvailableAndDoSelect to actually call
IsAvailable() with a proper mock Directory (instead of getStickyInvoker)
4. Fix imports formatting per dubbo-go convention (separate import groups)
Co-Authored-By: Claude <[email protected]>
* style: fix import formatting in test file per imports-formatter
Co-Authored-By: Claude <[email protected]>
* refactor(cluster): use atomic.Pointer for StickyInvoker, return
*BaseClusterInvoker from constructor
- Replace atomic.Value + stickyInvokerWrapper + getter/setter with
atomic.Pointer[base.Invoker] — idiomatic Go, nil-safe, type-safe,
no wrapper needed
- NewBaseClusterInvoker now returns *BaseClusterInvoker (pointer),
preventing accidental value copies of the struct containing atomic
fields
- All 9 cluster implementations updated to use pointer embedding
(*base.BaseClusterInvoker) to match the new constructor return type
- StickyInvoker field remains exported (atomic.Pointer is safe for
concurrent access by design)
Co-Authored-By: Claude <[email protected]>
* fix(cluster): fix StickyInvoker data race without breaking exported v3 API
Address [P1] review on PR #3439: the previous fix changed two exported
symbols already released in v3.3.2-20260709 — the StickyInvoker field
type (base.Invoker -> atomic.Pointer[base.Invoker]) and the
NewBaseClusterInvoker return type (value -> pointer) — breaking source
compatibility for external consumers.
Restore the original signatures and implement the race fix via
unexported internal state instead:
- Keep StickyInvoker as the exported base.Invoker field.
- Keep NewBaseClusterInvoker returning BaseClusterInvoker by value.
- Revert cluster invoker embeddings back to value embeddings.
- Guard StickyInvoker access in IsAvailable / DoSelect with an
unexported stickyLock sync.RWMutex via getStickyInvoker /
setStickyInvoker accessors, eliminating the read/write data race
without touching any exported symbol.
The existing TestStickyConcurrent* tests continue to pass under -race.
Co-Authored-By: Claude <[email protected]>
---------
Co-authored-by: Claude <[email protected]>
---
cluster/cluster/base/cluster_invoker.go | 44 ++++++++++----
cluster/cluster/base/cluster_invoker_test.go | 87 ++++++++++++++++++++++++++++
2 files changed, 120 insertions(+), 11 deletions(-)
diff --git a/cluster/cluster/base/cluster_invoker.go
b/cluster/cluster/base/cluster_invoker.go
index ddb9339fc..eacd07771 100644
--- a/cluster/cluster/base/cluster_invoker.go
+++ b/cluster/cluster/base/cluster_invoker.go
@@ -20,6 +20,7 @@ package base
import (
"slices"
+ "sync"
)
import (
@@ -27,7 +28,7 @@ import (
perrors "github.com/pkg/errors"
- "go.uber.org/atomic"
+ uberatomic "go.uber.org/atomic"
)
import (
@@ -42,15 +43,20 @@ import (
type BaseClusterInvoker struct {
Directory directory.Directory
AvailableCheck bool
- Destroyed *atomic.Bool
+ Destroyed *uberatomic.Bool
StickyInvoker base.Invoker
+
+ // stickyLock guards StickyInvoker against the data race between
IsAvailable
+ // (read) and DoSelect (read/write). It is unexported so the public
field and
+ // constructor signatures stay source-compatible across the v3.x line.
+ stickyLock sync.RWMutex
}
func NewBaseClusterInvoker(directory directory.Directory) BaseClusterInvoker {
return BaseClusterInvoker{
Directory: directory,
AvailableCheck: true,
- Destroyed: atomic.NewBool(false),
+ Destroyed: uberatomic.NewBool(false),
}
}
@@ -66,12 +72,26 @@ func (invoker *BaseClusterInvoker) Destroy() {
}
func (invoker *BaseClusterInvoker) IsAvailable() bool {
- if invoker.StickyInvoker != nil {
- return invoker.StickyInvoker.IsAvailable()
+ if sticky := invoker.getStickyInvoker(); sticky != nil {
+ return sticky.IsAvailable()
}
return invoker.Directory.IsAvailable()
}
+// getStickyInvoker returns the sticky invoker under a read lock.
+func (invoker *BaseClusterInvoker) getStickyInvoker() base.Invoker {
+ invoker.stickyLock.RLock()
+ defer invoker.stickyLock.RUnlock()
+ return invoker.StickyInvoker
+}
+
+// setStickyInvoker replaces the sticky invoker under a write lock.
+func (invoker *BaseClusterInvoker) setStickyInvoker(v base.Invoker) {
+ invoker.stickyLock.Lock()
+ defer invoker.stickyLock.Unlock()
+ invoker.StickyInvoker = v
+}
+
// CheckInvokers checks invokers' status if is available or not
func (invoker *BaseClusterInvoker) CheckInvokers(invokers []base.Invoker,
invocation base.Invocation) error {
if len(invokers) == 0 {
@@ -104,19 +124,21 @@ func (invoker *BaseClusterInvoker) DoSelect(lb
loadbalance.LoadBalance, invocati
// Get the service method sticky config if have
sticky = url.GetMethodParamBool(invocation.MethodName(),
constant.StickyKey, sticky)
- if invoker.StickyInvoker != nil && !isInvoked(invoker.StickyInvoker,
invokers) {
- invoker.StickyInvoker = nil
+ stickyInvoker := invoker.getStickyInvoker()
+ if stickyInvoker != nil && !isInvoked(stickyInvoker, invokers) {
+ invoker.setStickyInvoker(nil)
+ stickyInvoker = nil
}
if sticky && invoker.AvailableCheck &&
- invoker.StickyInvoker != nil &&
invoker.StickyInvoker.IsAvailable() &&
- (invoked == nil || !isInvoked(invoker.StickyInvoker, invoked)) {
- return invoker.StickyInvoker
+ stickyInvoker != nil && stickyInvoker.IsAvailable() &&
+ (invoked == nil || !isInvoked(stickyInvoker, invoked)) {
+ return stickyInvoker
}
selectedInvoker = invoker.doSelectInvoker(lb, invocation, invokers,
invoked)
if sticky {
- invoker.StickyInvoker = selectedInvoker
+ invoker.setStickyInvoker(selectedInvoker)
}
return selectedInvoker
}
diff --git a/cluster/cluster/base/cluster_invoker_test.go
b/cluster/cluster/base/cluster_invoker_test.go
index 9907a43c2..813aabcaf 100644
--- a/cluster/cluster/base/cluster_invoker_test.go
+++ b/cluster/cluster/base/cluster_invoker_test.go
@@ -19,6 +19,7 @@ package base
import (
"fmt"
+ "sync"
"testing"
)
@@ -73,3 +74,89 @@ func TestStickyNormalWhenError(t *testing.T) {
result1 := base.DoSelect(random.NewRandomLoadBalance(),
invocation.NewRPCInvocation(baseClusterInvokerMethodName, nil, nil), invokers,
invoked)
assert.NotEqual(t, result, result1)
}
+
+// TestStickyConcurrentDoSelect verifies that concurrent calls to DoSelect
+// with sticky enabled do not cause a data race on StickyInvoker.
+func TestStickyConcurrentDoSelect(t *testing.T) {
+ var invokers []protocolbase.Invoker
+ for i := 0; i < 10; i++ {
+ url, _ := common.NewURL(fmt.Sprintf(baseClusterInvokerFormat,
i))
+ url.SetParam("sticky", "true")
+ invokers = append(invokers, clusterpkg.NewMockInvoker(url, 1))
+ }
+ base := &BaseClusterInvoker{}
+ base.AvailableCheck = true
+
+ lb := random.NewRandomLoadBalance()
+ invocation1 :=
invocation.NewRPCInvocation(baseClusterInvokerMethodName, nil, nil)
+
+ const concurrency = 100
+ var wg sync.WaitGroup
+ wg.Add(concurrency)
+ for i := 0; i < concurrency; i++ {
+ go func() {
+ defer wg.Done()
+ invoked := make([]protocolbase.Invoker, 0)
+ result := base.DoSelect(lb, invocation1, invokers,
invoked)
+ assert.NotNil(t, result)
+ }()
+ }
+ wg.Wait()
+}
+
+// TestStickyConcurrentIsAvailableAndDoSelect verifies that concurrent
+// IsAvailable and DoSelect calls do not cause a data race on StickyInvoker.
+func TestStickyConcurrentIsAvailableAndDoSelect(t *testing.T) {
+ var invokers []protocolbase.Invoker
+ for i := 0; i < 10; i++ {
+ url, _ := common.NewURL(fmt.Sprintf(baseClusterInvokerFormat,
i))
+ url.SetParam("sticky", "true")
+ invokers = append(invokers, clusterpkg.NewMockInvoker(url, 1))
+ }
+
+ // Use NewBaseClusterInvoker so that Directory is initialized,
+ // allowing IsAvailable() to work without panicking.
+ dir := newMockDirectory(invokers)
+ base := NewBaseClusterInvoker(dir)
+ base.AvailableCheck = true
+
+ lb := random.NewRandomLoadBalance()
+ invocation1 :=
invocation.NewRPCInvocation(baseClusterInvokerMethodName, nil, nil)
+
+ // First DoSelect to set the sticky invoker so IsAvailable uses the
sticky path
+ invoked := make([]protocolbase.Invoker, 0)
+ base.DoSelect(lb, invocation1, invokers, invoked)
+
+ const concurrency = 100
+ var wg sync.WaitGroup
+ wg.Add(concurrency * 2)
+ for i := 0; i < concurrency; i++ {
+ go func() {
+ defer wg.Done()
+ base.IsAvailable()
+ }()
+ go func() {
+ defer wg.Done()
+ base.DoSelect(lb, invocation1, invokers, invoked)
+ }()
+ }
+ wg.Wait()
+}
+
+// mockDirectory is a minimal directory.Directory implementation for testing.
+type mockDirectory struct {
+ invokers []protocolbase.Invoker
+ url *common.URL
+}
+
+func newMockDirectory(invokers []protocolbase.Invoker) *mockDirectory {
+ url, _ := common.NewURL(baseClusterInvokerFormat)
+ url.SetParam("sticky", "true")
+ return &mockDirectory{invokers: invokers, url: url}
+}
+
+func (d *mockDirectory) GetURL() *common.URL {
return d.url }
+func (d *mockDirectory) IsAvailable() bool {
return true }
+func (d *mockDirectory) Destroy() {}
+func (d *mockDirectory) List(protocolbase.Invocation) []protocolbase.Invoker {
return d.invokers }
+func (d *mockDirectory) Subscribe(*common.URL) error {
return nil }
