chore: upgrade dep and fix data race in ut (#19)

* chore: upgrade dep and fix data race in ut

.github/workflows/tests.yml

@@ -26,7 +26,7 @@ jobs:
 #            ${{ runner.os }}-go-
 
       - name: Unit Test
-        run: go test -covermode=atomic -coverprofile=coverage.out ./...
+        run: go test -race -covermode=atomic -coverprofile=coverage.out ./...
 
       - name: Benchmark
         run: go test -bench=. -benchmem -run=none ./...

core/manager/client.go

@@ -553,6 +553,18 @@ func (c *xdsClient) handleResponse(msg interface{}) error {
 	return err
 }
 
+func (c *xdsClient) version(t xdsresource.ResourceType) string {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.versionMap[t]
+}
+
+func (c *xdsClient) nonce(t xdsresource.ResourceType) string {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.nonceMap[t]
+}
+
 func clearRequestCh(ch chan *discoveryv3.DiscoveryRequest, length int) {
 	for i := 0; i < length; i++ {
 		select {

core/manager/client_test.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"fmt"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -128,21 +129,21 @@ func Test_xdsClient_handleResponse(t *testing.T) {
 	// handle before watch
 	err := c.handleResponse(mock.LdsResp1)
 	assert.Nil(t, err)
-	assert.Equal(t, c.versionMap[xdsresource.ListenerType], "")
-	assert.Equal(t, c.nonceMap[xdsresource.ListenerType], "")
+	assert.Equal(t, c.version(xdsresource.ListenerType), "")
+	assert.Equal(t, c.nonce(xdsresource.ListenerType), "")
 
 	// handle after watch
 	c.watchedResource[xdsresource.ListenerType] = make(map[string]bool)
 	err = c.handleResponse(mock.LdsResp1)
 	assert.Nil(t, err)
-	assert.Equal(t, c.versionMap[xdsresource.ListenerType], mock.LDSVersion1)
-	assert.Equal(t, c.nonceMap[xdsresource.ListenerType], mock.LDSNonce1)
+	assert.Equal(t, c.version(xdsresource.ListenerType), mock.LDSVersion1)
+	assert.Equal(t, c.nonce(xdsresource.ListenerType), mock.LDSNonce1)
 
 	c.watchedResource[xdsresource.RouteConfigType] = make(map[string]bool)
 	err = c.handleResponse(mock.RdsResp1)
 	assert.Nil(t, err)
-	assert.Equal(t, c.versionMap[xdsresource.RouteConfigType], mock.RDSVersion1)
-	assert.Equal(t, c.nonceMap[xdsresource.RouteConfigType], mock.RDSNonce1)
+	assert.Equal(t, c.version(xdsresource.RouteConfigType), mock.RDSVersion1)
+	assert.Equal(t, c.nonce(xdsresource.RouteConfigType), mock.RDSNonce1)
 }
 
 func TestReconnect(t *testing.T) {
@@ -152,8 +153,8 @@ func TestReconnect(t *testing.T) {
 	}
 	sendCh := make(chan *mockStatus)
 	recvCh := make(chan *mockStatus)
-	sendCnt, recvCnt := 0, 0
-	closed := false
+	sendCnt, recvCnt := int64(0), int64(0)
+	closed := int32(0)
 	defer func() {
 		close(sendCh)
 		close(recvCh)
@@ -162,20 +163,20 @@ func TestReconnect(t *testing.T) {
 	ac := &mockADSClient{
 		opt: &streamOpt{
 			sendFunc: func(req *discoveryv3.DiscoveryRequest) error {
-				sendCnt++
+				atomic.AddInt64(&sendCnt, 1)
 				return nil
 			},
 			recvFunc: func() (response *discoveryv3.DiscoveryResponse, err error) {
 				s := <-recvCh
-				recvCnt++
+				atomic.AddInt64(&recvCnt, 1)
 				if s.err != nil {
 					return nil, s.err
 				}
 				// handle eds will not trigger new send
 				return mock.EdsResp1, nil
 			},
 			closeFunc: func() error {
-				closed = true
+				atomic.StoreInt32(&closed, 1)
 				return nil
 			},
 		},
@@ -189,21 +190,22 @@ func TestReconnect(t *testing.T) {
 		},
 	}, ac, mockUpdater)
 	assert.Nil(t, err)
-	assert.Equal(t, cli.versionMap[xdsresource.EndpointsType], "")
+
+	assert.Equal(t, cli.version(xdsresource.EndpointsType), "")
 	cli.Watch(xdsresource.EndpointsType, xdsresource.EndpointName1, false)
 	// mock recv succeed
 	recvCh <- &mockStatus{err: nil}
 	time.Sleep(10 * time.Millisecond)
-	assert.Equal(t, cli.nonceMap[xdsresource.EndpointsType], mock.EdsResp1.Nonce)
+	assert.Equal(t, cli.nonce(xdsresource.EndpointsType), mock.EdsResp1.Nonce)
 	// mock recv failed, reconnect
 	recvCh <- &mockStatus{err: fmt.Errorf("recv failed")}
 	time.Sleep(10 * time.Millisecond)
-	assert.Equal(t, 2, recvCnt)
-	assert.Equal(t, true, closed)
-	assert.Equal(t, 3, sendCnt) // watch&ack and reconnect
+	assert.Equal(t, 2, int(atomic.LoadInt64(&recvCnt)))
+	assert.Equal(t, 1, int(atomic.LoadInt32(&closed)))
+	assert.Equal(t, 3, int(atomic.LoadInt64(&sendCnt))) // watch&ack and reconnect
 	// without resp, the nonce should be reset to empty. the version should not be reset.
-	assert.Equal(t, cli.nonceMap[xdsresource.EndpointsType], "")
-	assert.Equal(t, cli.versionMap[xdsresource.EndpointsType], mock.EdsResp1.VersionInfo)
+	assert.Equal(t, cli.nonce(xdsresource.ListenerType), "")
+	assert.Equal(t, cli.version(xdsresource.EndpointsType), mock.EdsResp1.VersionInfo)
 	_ = cli
 }
 

go.mod

@@ -3,47 +3,61 @@ module github.com/kitex-contrib/xds
 go 1.17
 
 require (
-	github.com/bytedance/gopkg v0.0.0-20220817015305-b879a72dc90f
+	github.com/bytedance/gopkg v0.0.0-20230728082804-614d0af6619b
 	github.com/cenkalti/backoff/v4 v4.1.0
-	github.com/cloudwego/kitex v0.5.1
-	github.com/envoyproxy/go-control-plane v0.10.3
-	github.com/golang/protobuf v1.5.2
-	github.com/stretchr/testify v1.7.1
-	google.golang.org/genproto v0.0.0-20220819174105-e9f053255caa
-	google.golang.org/protobuf v1.28.1
+	github.com/cloudwego/kitex v0.7.3
+	github.com/envoyproxy/go-control-plane v0.11.1
+	github.com/golang/protobuf v1.5.3
+	github.com/stretchr/testify v1.8.3
+	google.golang.org/protobuf v1.30.0
 )
 
 require (
 	github.com/apache/thrift v0.13.0 // indirect
-	github.com/census-instrumentation/opencensus-proto v0.3.0 // indirect
-	github.com/chenzhuoyu/iasm v0.0.0-20230222070914-0b1b64b0e762 // indirect
-	github.com/choleraehyq/pid v0.0.16 // indirect
+	github.com/bytedance/sonic v1.9.1 // indirect
+	github.com/census-instrumentation/opencensus-proto v0.4.1 // indirect
+	github.com/chenzhuoyu/base64x v0.0.0-20221115062448-fe3a3abad311 // indirect
+	github.com/chenzhuoyu/iasm v0.9.0 // indirect
+	github.com/choleraehyq/pid v0.0.17 // indirect
+	github.com/cloudwego/configmanager v0.2.0 // indirect
+	github.com/cloudwego/dynamicgo v0.1.3 // indirect
 	github.com/cloudwego/fastpb v0.0.4 // indirect
-	github.com/cloudwego/frugal v0.1.6 // indirect
-	github.com/cloudwego/netpoll v0.3.2 // indirect
-	github.com/cloudwego/thriftgo v0.2.8 // indirect
-	github.com/cncf/xds/go v0.0.0-20220520190051-1e77728a1eaa // indirect
-	github.com/davecgh/go-spew v1.1.1 // indirect
-	github.com/envoyproxy/protoc-gen-validate v0.6.7 // indirect
-	github.com/google/go-cmp v0.5.8 // indirect
+	github.com/cloudwego/frugal v0.1.8 // indirect
+	github.com/cloudwego/localsession v0.0.2 // indirect
+	github.com/cloudwego/netpoll v0.5.1 // indirect
+	github.com/cloudwego/thriftgo v0.3.2-0.20230828085742-edaddf2c17af // indirect
+	github.com/cncf/xds/go v0.0.0-20230428030218-4003588d1b74 // indirect
+	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
+	github.com/envoyproxy/protoc-gen-validate v1.0.1 // indirect
+	github.com/fatih/structtag v1.2.0 // indirect
 	github.com/google/pprof v0.0.0-20220608213341-c488b8fa1db3 // indirect
+	github.com/iancoleman/strcase v0.2.0 // indirect
 	github.com/jhump/protoreflect v1.8.2 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
+	github.com/klauspost/cpuid/v2 v2.2.4 // indirect
 	github.com/kr/pretty v0.3.0 // indirect
-	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
+	github.com/modern-go/gls v0.0.0-20220109145502-612d0167dce5 // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/oleiade/lane v1.0.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
-	github.com/rogpeppe/go-internal v1.8.0 // indirect
+	github.com/rogpeppe/go-internal v1.9.0 // indirect
 	github.com/tidwall/gjson v1.9.3 // indirect
 	github.com/tidwall/match v1.1.1 // indirect
 	github.com/tidwall/pretty v1.2.0 // indirect
+	github.com/twitchyliquid64/golang-asm v0.15.1 // indirect
 	golang.org/x/arch v0.2.0 // indirect
-	golang.org/x/net v0.0.0-20220722155237-a158d28d115b // indirect
-	golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4 // indirect
-	golang.org/x/sys v0.0.0-20220817070843-5a390386f1f2 // indirect
-	golang.org/x/text v0.6.0 // indirect
-	google.golang.org/grpc v1.48.0 // indirect
+	golang.org/x/net v0.10.0 // indirect
+	golang.org/x/sync v0.1.0 // indirect
+	golang.org/x/sys v0.8.0 // indirect
+	golang.org/x/text v0.9.0 // indirect
+	google.golang.org/genproto v0.0.0-20230526203410-71b5a4ffd15e // indirect
+	google.golang.org/genproto/googleapis/api v0.0.0-20230526203410-71b5a4ffd15e // indirect
+	google.golang.org/grpc v1.55.0 // indirect
 	gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )
+
+require (
+	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20230526203410-71b5a4ffd15e
+)