Skip to content

Commit 3dfa7fb

Browse files
Aias00claude
andcommitted
fix(remoting): use atomic.Bool for ExchangeClient.init to fix data race
The init field was a plain bool with a FIXME comment acknowledging the race. It is written in doInit/Close and read in Request/AsyncRequest/Send without synchronization. This causes a data race when multiple goroutines concurrently trigger lazy initialization. Replace init bool with uatomic.Bool (already imported in this file) and: - Use CAS in doInit to ensure only one goroutine performs Connect - Spin-wait for in-flight initialization to complete - Reset init flag on Connect failure so retries can succeed - Use atomic Store in Close This follows the same pattern used by BaseInvoker (uatomic.Bool), BaseClusterInvoker (atomic.Bool with CAS), and Directory (atomic.Bool with CAS + mutex) elsewhere in the codebase. Co-Authored-By: Claude <noreply@anthropic.com>
1 parent a7b6ee6 commit 3dfa7fb

2 files changed

Lines changed: 59 additions & 6 deletions

File tree

remoting/exchange_client.go

Lines changed: 14 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -61,7 +61,7 @@ type ExchangeClient struct {
6161
ConnectTimeout time.Duration // timeout for connecting server
6262
address string // server address for dialing. The format: ip:port
6363
client Client // dealing with the transport
64-
init bool // the tag for init.
64+
init uatomic.Bool // the tag for init, protected by atomic operations
6565
activeNum uatomic.Uint32 // the number of service using the exchangeClient
6666
}
6767

@@ -82,19 +82,28 @@ func NewExchangeClient(url *common.URL, client Client, connectTimeout time.Durat
8282
}
8383

8484
func (cl *ExchangeClient) doInit(url *common.URL) error {
85-
if cl.init {
85+
if cl.init.Load() {
8686
return nil
8787
}
88+
// Use CompareAndSwap to ensure only one goroutine performs initialization.
89+
// If CAS fails, another goroutine already initialized; spin-wait until init completes.
90+
if !cl.init.CAS(false, true) {
91+
// Another goroutine is initializing; wait for it to complete.
92+
for !cl.init.Load() {
93+
time.Sleep(10 * time.Millisecond)
94+
}
95+
return nil
96+
}
97+
// This goroutine won the CAS — perform the actual initialization.
8898
if cl.client.Connect(url) != nil {
8999
// retry for a while
90100
time.Sleep(100 * time.Millisecond)
91101
if cl.client.Connect(url) != nil {
92102
logger.Errorf("[Remoting] failed to connect server, url=%v", url.Location)
103+
cl.init.Store(false) // reset on failure so future calls can retry
93104
return errors.New("Failed to connect server " + url.Location)
94105
}
95106
}
96-
// FIXME atomic operation
97-
cl.init = true
98107
return nil
99108
}
100109

@@ -195,7 +204,7 @@ func (client *ExchangeClient) Send(invocation *base.Invocation, url *common.URL,
195204
// Close close the client.
196205
func (client *ExchangeClient) Close() {
197206
client.client.Close()
198-
client.init = false
207+
client.init.Store(false)
199208
}
200209

201210
// IsAvailable to check if the underlying network client is available yet.

remoting/exchange_client_test.go

Lines changed: 45 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -92,7 +92,7 @@ func TestExchangeClientClose(t *testing.T) {
9292
m := &mockClient{available: true}
9393
ec := NewExchangeClient(testURL(), m, 5*time.Second, true)
9494
ec.Close()
95-
assert.False(t, ec.init)
95+
assert.False(t, ec.init.Load())
9696
}
9797

9898
func TestExchangeClientIsAvailable(t *testing.T) {
@@ -104,3 +104,47 @@ func TestExchangeClientIsAvailable(t *testing.T) {
104104
m.mu.Unlock()
105105
assert.False(t, ec.IsAvailable())
106106
}
107+
108+
func TestExchangeClientConcurrentDoInit(t *testing.T) {
109+
// Verify that concurrent calls to doInit only result in a single Connect call.
110+
m := &mockClient{available: true}
111+
ec := NewExchangeClient(testURL(), m, 5*time.Second, true)
112+
113+
var wg sync.WaitGroup
114+
const goroutines = 20
115+
wg.Add(goroutines)
116+
for i := 0; i < goroutines; i++ {
117+
go func() {
118+
defer wg.Done()
119+
// All goroutines trigger lazy init concurrently via Request-like paths
120+
if err := ec.doInit(testURL()); err != nil {
121+
t.Errorf("doInit failed: %v", err)
122+
}
123+
}()
124+
}
125+
wg.Wait()
126+
127+
m.mu.Lock()
128+
count := m.connCount
129+
m.mu.Unlock()
130+
assert.Equal(t, 1, count, "Connect should be called exactly once despite concurrent doInit")
131+
assert.True(t, ec.init.Load(), "init flag should be true after doInit")
132+
}
133+
134+
func TestExchangeClientDoInitFailureResets(t *testing.T) {
135+
// Verify that a failed doInit resets the init flag so future calls can retry.
136+
m := &mockClient{connectErr: errors.New("fail")}
137+
ec := NewExchangeClient(testURL(), m, 5*time.Second, true)
138+
139+
err := ec.doInit(testURL())
140+
assert.Error(t, err)
141+
assert.False(t, ec.init.Load(), "init flag should be reset after failed doInit")
142+
143+
// Now make Connect succeed and retry
144+
m.mu.Lock()
145+
m.connectErr = nil
146+
m.mu.Unlock()
147+
err = ec.doInit(testURL())
148+
assert.NoError(t, err)
149+
assert.True(t, ec.init.Load(), "init flag should be true after successful retry")
150+
}

0 commit comments

Comments
 (0)