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

remoting/exchange_client.go

@@ -19,6 +19,7 @@ package remoting
 
 import (
 	"errors"
+	"sync"
 	"time"
 )
 
@@ -61,7 +62,9 @@ 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.
+	initState      uatomic.Int32  // init state: 0=uninitialized, 1=initializing, 2=initialized
+	initCond       *sync.Cond     // signals waiters when init completes (success or failure)
+	initErr        error          // stored init error for waiters to observe (guarded by initCond.L)
 	activeNum      uatomic.Uint32 // the number of service using the exchangeClient
 }
 
@@ -71,6 +74,7 @@ func NewExchangeClient(url *common.URL, client Client, connectTimeout time.Durat
 		ConnectTimeout: connectTimeout,
 		address:        url.Location,
 		client:         client,
+		initCond:       sync.NewCond(&sync.Mutex{}),
 	}
 	if !lazyInit {
 		if err := exchangeClient.doInit(url); err != nil {
@@ -82,20 +86,43 @@ func NewExchangeClient(url *common.URL, client Client, connectTimeout time.Durat
 }
 
 func (cl *ExchangeClient) doInit(url *common.URL) error {
-	if cl.init {
+	// Fast path: already initialized.
+	if cl.initState.Load() == 2 {
 		return nil
 	}
+	// Try to become the initializer: transition 0 -> 1 (uninitialized -> initializing).
+	if !cl.initState.CAS(0, 1) {
+		// Another goroutine is initializing. Wait for it to complete.
+		cl.initCond.L.Lock()
+		for cl.initState.Load() == 1 {
+			cl.initCond.Wait()
+		}
+		err := cl.initErr
+		cl.initCond.L.Unlock()
+		return err
+	}
+	// This goroutine won the CAS — perform the actual initialization.
+	var initErr error
 	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)
-			return errors.New("Failed to connect server " + url.Location)
+			initErr = errors.New("Failed to connect server " + url.Location)
 		}
 	}
-	// FIXME atomic operation
-	cl.init = true
-	return nil
+	cl.initCond.L.Lock()
+	if initErr != nil {
+		// Reset to uninitialized so future calls can retry.
+		cl.initState.Store(0)
+		cl.initErr = initErr
+	} else {
+		cl.initState.Store(2)
+		cl.initErr = nil
+	}
+	cl.initCond.L.Unlock()
+	cl.initCond.Broadcast()
+	return initErr
 }
 
 // IncreaseActiveNumber increase number of service using client.
@@ -197,7 +224,11 @@ 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.initCond.L.Lock()
+	client.initState.Store(0)
+	client.initErr = nil
+	client.initCond.L.Unlock()
+	client.initCond.Broadcast()
 }
 
 // IsAvailable to check if the underlying network client is available yet.

remoting/exchange_client_test.go

@@ -99,7 +99,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.Equal(t, int32(0), ec.initState.Load(), "initState should be 0 (uninitialized) after Close")
 }
 
 func TestExchangeClientIsAvailable(t *testing.T) {
@@ -112,6 +112,83 @@ func TestExchangeClientIsAvailable(t *testing.T) {
 	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
+	errCh := make(chan error, goroutines)
+	wg.Add(goroutines)
+	for i := 0; i < goroutines; i++ {
+		go func() {
+			defer wg.Done()
+			if err := ec.doInit(testURL()); err != nil {
+				errCh <- err
+			}
+		}()
+	}
+	wg.Wait()
+	close(errCh)
+
+	for err := range errCh {
+		t.Errorf("doInit failed: %v", err)
+	}
+
+	m.mu.Lock()
+	count := m.connCount
+	m.mu.Unlock()
+	assert.Equal(t, 1, count, "Connect should be called exactly once despite concurrent doInit")
+	assert.Equal(t, int32(2), ec.initState.Load(), "initState should be 2 (initialized) after doInit")
+}
+
+func TestExchangeClientDoInitFailureResets(t *testing.T) {
+	// Verify that a failed doInit resets the init state so future calls can retry.
+	m := &mockClient{connectErr: errors.New("fail")}
+	ec := NewExchangeClient(testURL(), m, 5*time.Second, true)
+
+	err := ec.doInit(testURL())
+	require.Error(t, err)
+	assert.Equal(t, int32(0), ec.initState.Load(), "initState should be 0 (uninitialized) after failed doInit")
+
+	// Now make Connect succeed and retry
+	m.mu.Lock()
+	m.connectErr = nil
+	m.mu.Unlock()
+	err = ec.doInit(testURL())
+	require.NoError(t, err)
+	assert.Equal(t, int32(2), ec.initState.Load(), "initState should be 2 (initialized) after successful retry")
+}
+
+func TestExchangeClientConcurrentDoInitPropagatesError(t *testing.T) {
+	// Verify that when initialization fails, all concurrent waiters receive the same error.
+	m := &mockClient{connectErr: errors.New("connect refused")}
+	ec := NewExchangeClient(testURL(), m, 5*time.Second, true)
+
+	var wg sync.WaitGroup
+	const goroutines = 10
+	errCh := make(chan error, goroutines)
+	wg.Add(goroutines)
+	for i := 0; i < goroutines; i++ {
+		go func() {
+			defer wg.Done()
+			errCh <- ec.doInit(testURL())
+		}()
+	}
+	wg.Wait()
+	close(errCh)
+
+	errorCount := 0
+	for err := range errCh {
+		if err != nil {
+			errorCount++
+		}
+	}
+	assert.Equal(t, goroutines, errorCount, "all goroutines should receive the init error")
+	assert.Equal(t, int32(0), ec.initState.Load(), "initState should be 0 (uninitialized) after failure")
+}
+
 // newTestInvocation builds a *base.Invocation usable by ExchangeClient.Request/AsyncRequest.
 func newTestInvocation() *base.Invocation {
 	var inv base.Invocation = invocation.NewRPCInvocation("test", nil, nil)