fix(remoting): use atomic.Bool for ExchangeClient.init to fix data race

remoting/exchange_client.go

@@ -61,7 +61,7 @@ type ExchangeClient struct {
 	ConnectTimeout time.Duration  // timeout for connecting server
 	address        string         // server address for dialing. The format: ip:port
 	client         Client         // dealing with the transport
-	init           bool           // the tag for init.
+	init           uatomic.Bool  // the tag for init, protected by atomic operations
 	activeNum      uatomic.Uint32 // the number of service using the exchangeClient
 }
 
@@ -82,19 +82,28 @@ func NewExchangeClient(url *common.URL, client Client, connectTimeout time.Durat
 }
 
 func (cl *ExchangeClient) doInit(url *common.URL) error {
-	if cl.init {
+	if cl.init.Load() {
 		return nil
 	}
+	// Use CompareAndSwap to ensure only one goroutine performs initialization.
+	// If CAS fails, another goroutine already initialized; spin-wait until init completes.
+	if !cl.init.CAS(false, true) {
+		// Another goroutine is initializing; wait for it to complete.
+		for !cl.init.Load() {
+			time.Sleep(10 * time.Millisecond)
+		}
+		return nil
+	}
+	// This goroutine won the CAS — perform the actual initialization.
 	if cl.client.Connect(url) != nil {
 		// retry for a while
 		time.Sleep(100 * time.Millisecond)
 		if cl.client.Connect(url) != nil {
 			logger.Errorf("[Remoting] failed to connect server, url=%v", url.Location)
+			cl.init.Store(false) // reset on failure so future calls can retry
 			return errors.New("Failed to connect server " + url.Location)
 		}
 	}
-	// FIXME atomic operation
-	cl.init = true
 	return nil
 }
 
@@ -195,7 +204,7 @@ func (client *ExchangeClient) Send(invocation *base.Invocation, url *common.URL,
 // Close close the client.
 func (client *ExchangeClient) Close() {
 	client.client.Close()
-	client.init = false
+	client.init.Store(false)
 }
 
 // IsAvailable to check if the underlying network client is available yet.

remoting/exchange_client_test.go

@@ -92,7 +92,7 @@ func TestExchangeClientClose(t *testing.T) {
 	m := &mockClient{available: true}
 	ec := NewExchangeClient(testURL(), m, 5*time.Second, true)
 	ec.Close()
-	assert.False(t, ec.init)
+	assert.False(t, ec.init.Load())
 }
 
 func TestExchangeClientIsAvailable(t *testing.T) {
@@ -104,3 +104,47 @@ func TestExchangeClientIsAvailable(t *testing.T) {
 	m.mu.Unlock()
 	assert.False(t, ec.IsAvailable())
 }
+
+func TestExchangeClientConcurrentDoInit(t *testing.T) {
+	// Verify that concurrent calls to doInit only result in a single Connect call.
+	m := &mockClient{available: true}
+	ec := NewExchangeClient(testURL(), m, 5*time.Second, true)
+
+	var wg sync.WaitGroup
+	const goroutines = 20
+	wg.Add(goroutines)
+	for i := 0; i < goroutines; i++ {
+		go func() {
+			defer wg.Done()
+			// All goroutines trigger lazy init concurrently via Request-like paths
+			if err := ec.doInit(testURL()); err != nil {
+				t.Errorf("doInit failed: %v", err)
+			}
+		}()
+	}
+	wg.Wait()
+
+	m.mu.Lock()
+	count := m.connCount
+	m.mu.Unlock()
+	assert.Equal(t, 1, count, "Connect should be called exactly once despite concurrent doInit")
+	assert.True(t, ec.init.Load(), "init flag should be true after doInit")
+}
+
+func TestExchangeClientDoInitFailureResets(t *testing.T) {
+	// Verify that a failed doInit resets the init flag so future calls can retry.
+	m := &mockClient{connectErr: errors.New("fail")}
+	ec := NewExchangeClient(testURL(), m, 5*time.Second, true)
+
+	err := ec.doInit(testURL())
+	assert.Error(t, err)
+	assert.False(t, ec.init.Load(), "init flag should be reset after failed doInit")
+
+	// Now make Connect succeed and retry
+	m.mu.Lock()
+	m.connectErr = nil
+	m.mu.Unlock()
+	err = ec.doInit(testURL())
+	assert.NoError(t, err)
+	assert.True(t, ec.init.Load(), "init flag should be true after successful retry")
+}