tests: avoid relying on socket reads/writes for synchronization (#9033)
### Background
In #9032, we will transition from standard `net.Conn.Read` methods to
`syscall` UNIX APIs to enable non-memory-pinning reads. Due to this
change, the Go race detector beings failing on tests that share state
between client and server goroutines without standard synchronization
primitives (mutexes, channels, etc.).
Because these tests rely on the network request itself as a memory
barrier, they are logically safe but technically racy from the Go
runtime's perspective. The standard `net.Conn` leverages Go's internal
network poller, which inadvertently provides the "happens-before" edges
the race detector looks for. Dropping down to raw syscalls bypasses
this, causing the detector to flag the accesses. (Minimal repro:
https://go.dev/play/p/yvEtBmLTOJ2)
### Solution
Introduce explicit synchronization to the affected tests. Tests now
properly coordinate shared state access between clients and servers
without relying on socket I/O timing.
RELEASE NOTES: N/A
diff --git a/encoding/encoding_test.go b/encoding/encoding_test.go
index a3ff145..dbdca59 100644
--- a/encoding/encoding_test.go
+++ b/encoding/encoding_test.go
@@ -119,30 +119,35 @@
ec := &errProtoCodec{name: t.Name(), encodingErr: encodingErr}
// Start a server with the above codec.
- backend := stubserver.StartTestService(t, nil, grpc.ForceServerCodecV2(ec))
- defer backend.Stop()
-
- // Create a channel to the above server.
- cc, err := grpc.NewClient(backend.Address, grpc.WithTransportCredentials(insecure.NewCredentials()))
- if err != nil {
- t.Fatalf("Failed to dial test backend at %q: %v", backend.Address, err)
+ backend1 := stubserver.StubServer{
+ EmptyCallF: func(context.Context, *testpb.Empty) (*testpb.Empty, error) { return &testpb.Empty{}, nil },
}
- defer cc.Close()
+ if err := backend1.Start([]grpc.ServerOption{grpc.ForceServerCodecV2(ec)}, grpc.WithTransportCredentials(insecure.NewCredentials())); err != nil {
+ t.Fatal(err)
+ }
+ defer backend1.Stop()
// Make an RPC and expect it to fail. Since we do not specify any codec
// here, the proto codec will get automatically used.
ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
defer cancel()
- client := testgrpc.NewTestServiceClient(cc)
- _, err = client.EmptyCall(ctx, &testpb.Empty{})
+ _, err := backend1.Client.EmptyCall(ctx, &testpb.Empty{})
if err == nil || !strings.Contains(err.Error(), encodingErr.Error()) {
t.Fatalf("RPC failed with error: %v, want: %v", err, encodingErr)
}
// Configure the codec on the server to not return errors anymore and expect
// the RPC to succeed.
- ec.encodingErr = nil
- if _, err := client.EmptyCall(ctx, &testpb.Empty{}); err != nil {
+ ec = &errProtoCodec{name: t.Name()}
+ backend2 := stubserver.StubServer{
+ EmptyCallF: func(context.Context, *testpb.Empty) (*testpb.Empty, error) { return &testpb.Empty{}, nil },
+ }
+ if err := backend2.Start([]grpc.ServerOption{grpc.ForceServerCodecV2(ec)}, grpc.WithTransportCredentials(insecure.NewCredentials())); err != nil {
+ t.Fatal(err)
+ }
+ defer backend2.Stop()
+
+ if _, err := backend2.Client.EmptyCall(ctx, &testpb.Empty{}); err != nil {
t.Fatalf("RPC failed with error: %v", err)
}
}
@@ -154,32 +159,36 @@
decodingErr := errors.New("decoding failed")
ec := &errProtoCodec{name: t.Name(), decodingErr: decodingErr}
- // Start a server with the above codec.
- backend := stubserver.StartTestService(t, nil, grpc.ForceServerCodecV2(ec))
- defer backend.Stop()
-
- // Create a channel to the above server. Since we do not specify any codec
- // here, the proto codec will get automatically used.
- cc, err := grpc.NewClient(backend.Address, grpc.WithTransportCredentials(insecure.NewCredentials()))
- if err != nil {
- t.Fatalf("Failed to dial test backend at %q: %v", backend.Address, err)
+ // Start a server with the above codec and a channel to the server.
+ backend1 := stubserver.StubServer{
+ EmptyCallF: func(context.Context, *testpb.Empty) (*testpb.Empty, error) { return &testpb.Empty{}, nil },
}
- defer cc.Close()
+ if err := backend1.Start([]grpc.ServerOption{grpc.ForceServerCodecV2(ec)}, grpc.WithTransportCredentials(insecure.NewCredentials())); err != nil {
+ t.Fatal(err)
+ }
+ defer backend1.Stop()
// Make an RPC and expect it to fail. Since we do not specify any codec
// here, the proto codec will get automatically used.
ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
defer cancel()
- client := testgrpc.NewTestServiceClient(cc)
- _, err = client.EmptyCall(ctx, &testpb.Empty{})
+ _, err := backend1.Client.EmptyCall(ctx, &testpb.Empty{})
if err == nil || !strings.Contains(err.Error(), decodingErr.Error()) || !strings.Contains(err.Error(), "grpc: error unmarshalling request") {
t.Fatalf("RPC failed with error: %v, want: %v", err, decodingErr)
}
// Configure the codec on the server to not return errors anymore and expect
// the RPC to succeed.
- ec.decodingErr = nil
- if _, err := client.EmptyCall(ctx, &testpb.Empty{}); err != nil {
+ ec = &errProtoCodec{name: t.Name()}
+ backend2 := stubserver.StubServer{
+ EmptyCallF: func(context.Context, *testpb.Empty) (*testpb.Empty, error) { return &testpb.Empty{}, nil },
+ }
+ if err := backend2.Start([]grpc.ServerOption{grpc.ForceServerCodecV2(ec)}, grpc.WithTransportCredentials(insecure.NewCredentials())); err != nil {
+ t.Fatal(err)
+ }
+ defer backend2.Stop()
+
+ if _, err := backend2.Client.EmptyCall(ctx, &testpb.Empty{}); err != nil {
t.Fatalf("RPC failed with error: %v", err)
}
}
diff --git a/internal/xds/balancer/clusterimpl/tests/balancer_test.go b/internal/xds/balancer/clusterimpl/tests/balancer_test.go
index 627996a..f6554a7 100644
--- a/internal/xds/balancer/clusterimpl/tests/balancer_test.go
+++ b/internal/xds/balancer/clusterimpl/tests/balancer_test.go
@@ -1372,12 +1372,12 @@
mgmtServer, resolverBuilder, nodeID := setupManagementServerAndResolver(t)
// Start a server backend exposing the test service.
- var gotAuthority string
+ gotAuthority := atomic.Pointer[string]{}
f := &stubserver.StubServer{
EmptyCallF: func(ctx context.Context, _ *testpb.Empty) (*testpb.Empty, error) {
if md, ok := metadata.FromIncomingContext(ctx); ok {
if authVals := md.Get(":authority"); len(authVals) > 0 {
- gotAuthority = authVals[0]
+ gotAuthority.Store(&authVals[0])
}
}
return &testpb.Empty{}, nil
@@ -1415,8 +1415,8 @@
} else {
wantAuthority = server.Address
}
- if gotAuthority != wantAuthority {
- t.Errorf("invalid authority got: %q, want: %q", gotAuthority, wantAuthority)
+ if got, want := *gotAuthority.Load(), wantAuthority; got != want {
+ t.Errorf("invalid authority got: %q, want: %q", got, want)
}
// The authority specified via the `CallAuthority` CallOption takes the
@@ -1426,8 +1426,8 @@
t.Fatalf("client.EmptyCall() failed: %v", err)
}
- if gotAuthority != userAuthorityOverride {
- t.Errorf("Server received authority %q, want %q (user override)", gotAuthority, userAuthorityOverride)
+ if got, want := *gotAuthority.Load(), userAuthorityOverride; got != want {
+ t.Errorf("Server received authority %q, want %q (user override)", got, want)
}
})
}
diff --git a/internal/xds/clients/lrsclient/loadreport_test.go b/internal/xds/clients/lrsclient/loadreport_test.go
index 374e1af..9754cdb 100644
--- a/internal/xds/clients/lrsclient/loadreport_test.go
+++ b/internal/xds/clients/lrsclient/loadreport_test.go
@@ -21,6 +21,7 @@
import (
"context"
"net"
+ "sync"
"testing"
"time"
@@ -665,9 +666,12 @@
t.Fatalf("Timeout when waiting for LRS stream to be created: %v", err)
}
+ stateMu := sync.Mutex{}
// Initial time for reporter creation
currentTime := time.Now()
lrsclientinternal.TimeNow = func() time.Time {
+ stateMu.Lock()
+ defer stateMu.Unlock()
return currentTime
}
@@ -676,7 +680,9 @@
// Update currentTime to simulate the passage of time between the reporter
// creation and first stats() call.
+ stateMu.Lock()
currentTime = currentTime.Add(5 * time.Second)
+ stateMu.Unlock()
// Ensure the initial load reporting request is received at the server.
req, err := lrsServer.LRSRequestChan.Receive(ctx)
diff --git a/internal/xds/xdsdepmgr/xds_dependency_manager_test.go b/internal/xds/xdsdepmgr/xds_dependency_manager_test.go
index af09849..d1187c6 100644
--- a/internal/xds/xdsdepmgr/xds_dependency_manager_test.go
+++ b/internal/xds/xdsdepmgr/xds_dependency_manager_test.go
@@ -1505,6 +1505,13 @@
// Configure a endpoint resource that is expected to be NACKed because it
// does not contain the `Locality` field. Since a valid one is already
// cached, this should result in an ambient error.
+ resources = e2e.DefaultClientResources(e2e.ResourceParams{
+ NodeID: nodeID,
+ DialTarget: defaultTestServiceName,
+ Host: "localhost",
+ Port: 8080,
+ SecLevel: e2e.SecurityLevelNone,
+ })
resources.Endpoints[0].Endpoints[0].Locality = nil
if err := mgmtServer.Update(ctx, resources); err != nil {
t.Fatal(err)
diff --git a/orca/producer_test.go b/orca/producer_test.go
index 61425d6..9a0a68d 100644
--- a/orca/producer_test.go
+++ b/orca/producer_test.go
@@ -20,6 +20,7 @@
"context"
"fmt"
"sync"
+ "sync/atomic"
"testing"
"time"
@@ -279,14 +280,15 @@
// value.
const backoffShouldNotBeCalled = 9999 // Use to assert backoff function is not called.
const backoffAllowAny = -1 // Use to ignore any backoff calls.
- expectedBackoff := backoffAllowAny
+ expectedBackoff := atomic.Int32{}
+ expectedBackoff.Store(backoffAllowAny)
oldBackoff := internal.DefaultBackoffFunc
internal.DefaultBackoffFunc = func(got int) time.Duration {
- if expectedBackoff == backoffShouldNotBeCalled {
+ if expectedBackoff.Load() == backoffShouldNotBeCalled {
t.Errorf("Unexpected backoff call; parameter = %v", got)
- } else if expectedBackoff != backoffAllowAny {
- if got != expectedBackoff {
- t.Errorf("Unexpected backoff received; got %v want %v", got, expectedBackoff)
+ } else if expectedBackoff.Load() != backoffAllowAny {
+ if got != int(expectedBackoff.Load()) {
+ t.Errorf("Unexpected backoff received; got %v want %v", got, expectedBackoff.Load())
}
}
return time.Millisecond
@@ -363,23 +365,23 @@
// The next request should be immediate, since there was a message
// received.
- expectedBackoff = backoffShouldNotBeCalled
+ expectedBackoff.Store(backoffShouldNotBeCalled)
fake.respCh <- status.Errorf(codes.Internal, "injected error")
awaitRequest(reportInterval)
// The next requests will need to backoff.
- expectedBackoff = 0
+ expectedBackoff.Store(0)
fake.respCh <- status.Errorf(codes.Internal, "injected error")
awaitRequest(reportInterval)
- expectedBackoff = 1
+ expectedBackoff.Store(1)
fake.respCh <- status.Errorf(codes.Internal, "injected error")
awaitRequest(reportInterval)
- expectedBackoff = 2
+ expectedBackoff.Store(2)
fake.respCh <- status.Errorf(codes.Internal, "injected error")
awaitRequest(reportInterval)
// The next request should be immediate, since there was a message
// received.
- expectedBackoff = backoffShouldNotBeCalled
+ expectedBackoff.Store(backoffShouldNotBeCalled)
// Send another valid response and wait for it on the client.
fake.respCh <- loadReportWant
diff --git a/test/end2end_test.go b/test/end2end_test.go
index ddf6fd7..4b15f8d 100644
--- a/test/end2end_test.go
+++ b/test/end2end_test.go
@@ -2077,14 +2077,14 @@
}
type myTap struct {
- cnt int
+ cnt atomic.Int32
}
func (t *myTap) handle(ctx context.Context, info *tap.Info) (context.Context, error) {
if info != nil {
switch info.FullMethodName {
case "/grpc.testing.TestService/EmptyCall":
- t.cnt++
+ t.cnt.Add(1)
if vals := info.Header.Get("return-error"); len(vals) > 0 && vals[0] == "true" {
return nil, status.Errorf(codes.Unknown, "tap error")
@@ -2114,22 +2114,22 @@
if _, err := tc.EmptyCall(ctx, &testpb.Empty{}); err != nil {
t.Fatalf("TestService/EmptyCall(_, _) = _, %v, want _, <nil>", err)
}
- if ttap.cnt != 1 {
- t.Fatalf("Get the count in ttap %d, want 1", ttap.cnt)
+ if ttap.cnt.Load() != 1 {
+ t.Fatalf("Get the count in ttap %d, want 1", ttap.cnt.Load())
}
if _, err := tc.EmptyCall(metadata.AppendToOutgoingContext(ctx, "return-error", "false"), &testpb.Empty{}); err != nil {
t.Fatalf("TestService/EmptyCall(_, _) = _, %v, want _, <nil>", err)
}
- if ttap.cnt != 2 {
- t.Fatalf("Get the count in ttap %d, want 2", ttap.cnt)
+ if ttap.cnt.Load() != 2 {
+ t.Fatalf("Get the count in ttap %d, want 2", ttap.cnt.Load())
}
if _, err := tc.EmptyCall(metadata.AppendToOutgoingContext(ctx, "return-error", "true"), &testpb.Empty{}); status.Code(err) != codes.Unknown {
t.Fatalf("TestService/EmptyCall(_, _) = _, %v, want _, %s", err, codes.Unknown)
}
- if ttap.cnt != 3 {
- t.Fatalf("Get the count in ttap %d, want 3", ttap.cnt)
+ if ttap.cnt.Load() != 3 {
+ t.Fatalf("Get the count in ttap %d, want 3", ttap.cnt.Load())
}
payload, err := newPayload(testpb.PayloadType_COMPRESSABLE, 31)
@@ -5167,11 +5167,14 @@
}
func (s) TestGRPCMethod(t *testing.T) {
+ mu := sync.Mutex{}
var method string
var ok bool
ss := &stubserver.StubServer{
EmptyCallF: func(ctx context.Context, _ *testpb.Empty) (*testpb.Empty, error) {
+ mu.Lock()
+ defer mu.Unlock()
method, ok = grpc.Method(ctx)
return &testpb.Empty{}, nil
},
@@ -5188,6 +5191,8 @@
t.Fatalf("ss.Client.EmptyCall(_, _) = _, %v; want _, nil", err)
}
+ mu.Lock()
+ defer mu.Unlock()
if want := "/grpc.testing.TestService/EmptyCall"; !ok || method != want {
t.Fatalf("grpc.Method(_) = %q, %v; want %q, true", method, ok, want)
}
@@ -5614,9 +5619,12 @@
const testMethod = "/package.service/method"
e := tcpClearRREnv
te := newTest(t, e)
+ var mu sync.Mutex
var method string
var ok bool
te.unknownHandler = func(_ any, stream grpc.ServerStream) error {
+ mu.Lock()
+ defer mu.Unlock()
method, ok = grpc.MethodFromServerStream(stream)
return nil
}
@@ -5626,6 +5634,8 @@
ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
defer cancel()
_ = te.clientConn().Invoke(ctx, testMethod, nil, nil)
+ mu.Lock()
+ defer mu.Unlock()
if !ok || method != testMethod {
t.Fatalf("Invoke with method %q, got %q, %v, want %q, true", testMethod, method, ok, testMethod)
}
@@ -6213,13 +6223,6 @@
}
func testLargeTimeout(t *testing.T, e env) {
- te := newTest(t, e)
- te.declareLogNoise("Server.processUnaryRPC failed to write status")
-
- ts := &funcServer{}
- te.startServer(ts)
- defer te.tearDown()
- tc := testgrpc.NewTestServiceClient(te.clientConn())
timeouts := []time.Duration{
time.Duration(math.MaxInt64), // will be (correctly) converted to
@@ -6228,23 +6231,33 @@
}
for i, maxTimeout := range timeouts {
- ts.unaryCall = func(ctx context.Context, _ *testpb.SimpleRequest) (*testpb.SimpleResponse, error) {
- deadline, ok := ctx.Deadline()
- timeout := time.Until(deadline)
- minTimeout := maxTimeout - 5*time.Second
- if !ok || timeout < minTimeout || timeout > maxTimeout {
- t.Errorf("ctx.Deadline() = (now+%v), %v; want [%v, %v], true", timeout, ok, minTimeout, maxTimeout)
- return nil, status.Error(codes.OutOfRange, "deadline error")
+ func() {
+ te := newTest(t, e)
+ te.declareLogNoise("Server.processUnaryRPC failed to write status")
+
+ ts := &funcServer{
+ unaryCall: func(ctx context.Context, _ *testpb.SimpleRequest) (*testpb.SimpleResponse, error) {
+ deadline, ok := ctx.Deadline()
+ timeout := time.Until(deadline)
+ minTimeout := maxTimeout - 5*time.Second
+ if !ok || timeout < minTimeout || timeout > maxTimeout {
+ t.Errorf("ctx.Deadline() = (now+%v), %v; want [%v, %v], true", timeout, ok, minTimeout, maxTimeout)
+ return nil, status.Error(codes.OutOfRange, "deadline error")
+ }
+ return &testpb.SimpleResponse{}, nil
+ },
}
- return &testpb.SimpleResponse{}, nil
- }
+ te.startServer(ts)
+ defer te.tearDown()
+ tc := testgrpc.NewTestServiceClient(te.clientConn())
- ctx, cancel := context.WithTimeout(context.Background(), maxTimeout)
- defer cancel()
+ ctx, cancel := context.WithTimeout(context.Background(), maxTimeout)
+ defer cancel()
- if _, err := tc.UnaryCall(ctx, &testpb.SimpleRequest{}); err != nil {
- t.Errorf("case %v: UnaryCall(_) = _, %v; want _, nil", i, err)
- }
+ if _, err := tc.UnaryCall(ctx, &testpb.SimpleRequest{}); err != nil {
+ t.Errorf("case %v: UnaryCall(_) = _, %v; want _, nil", i, err)
+ }
+ }()
}
}
@@ -6960,11 +6973,11 @@
}
type mockMethodLogger struct {
- events uint64
+ events atomic.Uint64
}
func (mml *mockMethodLogger) Log(context.Context, binarylog.LogEntryConfig) {
- atomic.AddUint64(&mml.events, 1)
+ mml.events.Add(1)
}
// TestGlobalBinaryLoggingOptions tests the binary logging options for client
@@ -7008,11 +7021,11 @@
if _, err := ss.Client.UnaryCall(ctx, &testpb.SimpleRequest{}); err != nil {
t.Fatalf("Unexpected error from UnaryCall: %v", err)
}
- if csbl.mml.events != 5 {
- t.Fatalf("want 5 client side binary logging events, got %v", csbl.mml.events)
+ if got := csbl.mml.events.Load(); got != 5 {
+ t.Fatalf("want 5 client side binary logging events, got %v", got)
}
- if ssbl.mml.events != 5 {
- t.Fatalf("want 5 server side binary logging events, got %v", ssbl.mml.events)
+ if got := ssbl.mml.events.Load(); got != 5 {
+ t.Fatalf("want 5 server side binary logging events, got %v", got)
}
// Make a streaming RPC. This should cause Log calls on the MethodLogger.
@@ -7026,11 +7039,11 @@
t.Fatalf("unexpected error: %v, expected an EOF error", err)
}
- if csbl.mml.events != 8 {
- t.Fatalf("want 8 client side binary logging events, got %v", csbl.mml.events)
+ if got := csbl.mml.events.Load(); got != 8 {
+ t.Fatalf("want 8 client side binary logging events, got %v", got)
}
- if ssbl.mml.events != 8 {
- t.Fatalf("want 8 server side binary logging events, got %v", ssbl.mml.events)
+ if got := ssbl.mml.events.Load(); got != 8 {
+ t.Fatalf("want 8 server side binary logging events, got %v", got)
}
}
diff --git a/test/retry_test.go b/test/retry_test.go
index af97be2..62f873b 100644
--- a/test/retry_test.go
+++ b/test/retry_test.go
@@ -46,9 +46,12 @@
)
func (s) TestRetryUnary(t *testing.T) {
+ serverMu := sync.Mutex{}
i := -1
ss := &stubserver.StubServer{
EmptyCallF: func(context.Context, *testpb.Empty) (r *testpb.Empty, err error) {
+ serverMu.Lock()
+ defer serverMu.Unlock()
defer func() { t.Logf("server call %v returning err %v", i, err) }()
i++
switch i {
@@ -78,8 +81,8 @@
defer ss.Stop()
testCases := []struct {
- code codes.Code
- count int
+ wantCode codes.Code
+ wantCount int
}{
{codes.OK, 0},
{codes.OK, 2},
@@ -94,19 +97,26 @@
ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
_, err := ss.Client.EmptyCall(ctx, &testpb.Empty{})
cancel()
- if status.Code(err) != tc.code {
- t.Fatalf("EmptyCall(_, _) = _, %v; want _, <Code() = %v>", err, tc.code)
+ if status.Code(err) != tc.wantCode {
+ t.Fatalf("EmptyCall(_, _) = _, %v; want _, <Code() = %v>", err, tc.wantCode)
}
- if i != tc.count {
- t.Fatalf("i = %v; want %v", i, tc.count)
+ serverMu.Lock()
+ if i != tc.wantCount {
+ serverMu.Unlock()
+ t.Fatalf("i = %v; want %v", i, tc.wantCount)
}
+ serverMu.Unlock()
}
}
func (s) TestRetryThrottling(t *testing.T) {
+ serverMu := sync.Mutex{}
i := -1
+
ss := &stubserver.StubServer{
EmptyCallF: func(context.Context, *testpb.Empty) (*testpb.Empty, error) {
+ serverMu.Lock()
+ defer serverMu.Unlock()
i++
switch i {
case 0, 3, 6, 10, 11, 12, 13, 14, 16, 18:
@@ -138,8 +148,8 @@
defer ss.Stop()
testCases := []struct {
- code codes.Code
- count int
+ wantCode codes.Code
+ wantCount int
}{
{codes.OK, 0}, // tokens = 10
{codes.OK, 3}, // tokens = 8.5 (10 - 2 failures + 0.5 success)
@@ -158,12 +168,14 @@
ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
_, err := ss.Client.EmptyCall(ctx, &testpb.Empty{})
cancel()
- if status.Code(err) != tc.code {
- t.Errorf("EmptyCall(_, _) = _, %v; want _, <Code() = %v>", err, tc.code)
+ if status.Code(err) != tc.wantCode {
+ t.Errorf("EmptyCall(_, _) = _, %v; want _, <Code() = %v>", err, tc.wantCode)
}
- if i != tc.count {
- t.Errorf("i = %v; want %v", i, tc.count)
+ serverMu.Lock()
+ if i != tc.wantCount {
+ t.Errorf("i = %v; want %v", i, tc.wantCount)
}
+ serverMu.Unlock()
}
}
@@ -414,10 +426,14 @@
clientOps: []clientOp{cReqPayload(largePayload), cErr(codes.Unavailable)},
}}
+ serverMu := sync.Mutex{}
var serverOpIter int
var serverOps []serverOp
+
ss := &stubserver.StubServer{
FullDuplexCallF: func(stream testgrpc.TestService_FullDuplexCallServer) error {
+ serverMu.Lock()
+ defer serverMu.Unlock()
for serverOpIter < len(serverOps) {
op := serverOps[serverOpIter]
serverOpIter++
@@ -458,8 +474,10 @@
for i, tc := range testCases {
func() {
+ serverMu.Lock()
serverOpIter = 0
serverOps = tc.serverOps
+ serverMu.Unlock()
stream, err := ss.Client.FullDuplexCall(ctx)
if err != nil {
@@ -471,6 +489,8 @@
break
}
}
+ serverMu.Lock()
+ defer serverMu.Unlock()
if serverOpIter != len(serverOps) {
t.Errorf("%v: serverOpIter = %v; want %v", tc.desc, serverOpIter, len(serverOps))
}
@@ -509,15 +529,20 @@
),
}
+ serverMu := sync.Mutex{}
streamCallCount := 0
unaryCallCount := 0
ss := &stubserver.StubServer{
FullDuplexCallF: func(testgrpc.TestService_FullDuplexCallServer) error {
+ serverMu.Lock()
+ defer serverMu.Unlock()
streamCallCount++
return status.New(codes.Unavailable, "this is a test error").Err()
},
EmptyCallF: func(context.Context, *testpb.Empty) (r *testpb.Empty, err error) {
+ serverMu.Lock()
+ defer serverMu.Unlock()
unaryCallCount++
return nil, status.New(codes.Unavailable, "this is a test error").Err()
},
@@ -555,9 +580,12 @@
t.Fatalf("want: ErrRetriesExhausted, got: %v", err)
}
+ serverMu.Lock()
if streamCallCount != tc.expectedAttempts {
+ serverMu.Unlock()
t.Fatalf("stream expectedAttempts = %v; want %v", streamCallCount, tc.expectedAttempts)
}
+ serverMu.Unlock()
// Test unary RPC
if ugot, err := ss.Client.EmptyCall(ctx, &testpb.Empty{}); err == nil {
@@ -565,6 +593,8 @@
} else if status.Code(err) != codes.Unavailable {
t.Fatalf("client: EmptyCall() = _, %v; want _, Unavailable", err)
}
+ serverMu.Lock()
+ defer serverMu.Unlock()
if unaryCallCount != tc.expectedAttempts {
t.Fatalf("unary expectedAttempts = %v; want %v", unaryCallCount, tc.expectedAttempts)
}
diff --git a/test/server_test.go b/test/server_test.go
index f04b4a5..0441c08 100644
--- a/test/server_test.go
+++ b/test/server_test.go
@@ -21,6 +21,7 @@
import (
"context"
"io"
+ "sync/atomic"
"testing"
"google.golang.org/grpc"
@@ -230,56 +231,56 @@
}
func (s) TestChainStreamServerInterceptor(t *testing.T) {
- callCounts := make([]int, 4)
+ callCounts := make([]atomic.Int32, 4)
firstInt := func(srv any, stream grpc.ServerStream, _ *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
- if callCounts[0] != 0 {
- return status.Errorf(codes.Internal, "callCounts[0] should be 0, but got=%d", callCounts[0])
+ if callCounts[0].Load() != 0 {
+ return status.Errorf(codes.Internal, "callCounts[0] should be 0, but got=%d", callCounts[0].Load())
}
- if callCounts[1] != 0 {
- return status.Errorf(codes.Internal, "callCounts[1] should be 0, but got=%d", callCounts[1])
+ if callCounts[1].Load() != 0 {
+ return status.Errorf(codes.Internal, "callCounts[1] should be 0, but got=%d", callCounts[1].Load())
}
- if callCounts[2] != 0 {
- return status.Errorf(codes.Internal, "callCounts[2] should be 0, but got=%d", callCounts[2])
+ if callCounts[2].Load() != 0 {
+ return status.Errorf(codes.Internal, "callCounts[2] should be 0, but got=%d", callCounts[2].Load())
}
- if callCounts[3] != 0 {
- return status.Errorf(codes.Internal, "callCounts[3] should be 0, but got=%d", callCounts[3])
+ if callCounts[3].Load() != 0 {
+ return status.Errorf(codes.Internal, "callCounts[3] should be 0, but got=%d", callCounts[3].Load())
}
- callCounts[0]++
+ callCounts[0].Add(1)
return handler(srv, stream)
}
secondInt := func(srv any, stream grpc.ServerStream, _ *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
- if callCounts[0] != 1 {
- return status.Errorf(codes.Internal, "callCounts[0] should be 1, but got=%d", callCounts[0])
+ if callCounts[0].Load() != 1 {
+ return status.Errorf(codes.Internal, "callCounts[0] should be 1, but got=%d", callCounts[0].Load())
}
- if callCounts[1] != 0 {
- return status.Errorf(codes.Internal, "callCounts[1] should be 0, but got=%d", callCounts[1])
+ if callCounts[1].Load() != 0 {
+ return status.Errorf(codes.Internal, "callCounts[1] should be 0, but got=%d", callCounts[1].Load())
}
- if callCounts[2] != 0 {
- return status.Errorf(codes.Internal, "callCounts[2] should be 0, but got=%d", callCounts[2])
+ if callCounts[2].Load() != 0 {
+ return status.Errorf(codes.Internal, "callCounts[2] should be 0, but got=%d", callCounts[2].Load())
}
- if callCounts[3] != 0 {
- return status.Errorf(codes.Internal, "callCounts[3] should be 0, but got=%d", callCounts[3])
+ if callCounts[3].Load() != 0 {
+ return status.Errorf(codes.Internal, "callCounts[3] should be 0, but got=%d", callCounts[3].Load())
}
- callCounts[1]++
+ callCounts[1].Add(1)
return handler(srv, stream)
}
lastInt := func(srv any, stream grpc.ServerStream, _ *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
- if callCounts[0] != 1 {
- return status.Errorf(codes.Internal, "callCounts[0] should be 1, but got=%d", callCounts[0])
+ if callCounts[0].Load() != 1 {
+ return status.Errorf(codes.Internal, "callCounts[0] should be 1, but got=%d", callCounts[0].Load())
}
- if callCounts[1] != 1 {
- return status.Errorf(codes.Internal, "callCounts[1] should be 1, but got=%d", callCounts[1])
+ if callCounts[1].Load() != 1 {
+ return status.Errorf(codes.Internal, "callCounts[1] should be 1, but got=%d", callCounts[1].Load())
}
- if callCounts[2] != 0 {
- return status.Errorf(codes.Internal, "callCounts[2] should be 0, but got=%d", callCounts[2])
+ if callCounts[2].Load() != 0 {
+ return status.Errorf(codes.Internal, "callCounts[2] should be 0, but got=%d", callCounts[2].Load())
}
- if callCounts[3] != 0 {
- return status.Errorf(codes.Internal, "callCounts[3] should be 0, but got=%d", callCounts[3])
+ if callCounts[3].Load() != 0 {
+ return status.Errorf(codes.Internal, "callCounts[3] should be 0, but got=%d", callCounts[3].Load())
}
- callCounts[2]++
+ callCounts[2].Add(1)
return handler(srv, stream)
}
@@ -289,19 +290,19 @@
ss := &stubserver.StubServer{
FullDuplexCallF: func(testgrpc.TestService_FullDuplexCallServer) error {
- if callCounts[0] != 1 {
- return status.Errorf(codes.Internal, "callCounts[0] should be 1, but got=%d", callCounts[0])
+ if callCounts[0].Load() != 1 {
+ return status.Errorf(codes.Internal, "callCounts[0] should be 1, but got=%d", callCounts[0].Load())
}
- if callCounts[1] != 1 {
- return status.Errorf(codes.Internal, "callCounts[1] should be 1, but got=%d", callCounts[1])
+ if callCounts[1].Load() != 1 {
+ return status.Errorf(codes.Internal, "callCounts[1] should be 1, but got=%d", callCounts[1].Load())
}
- if callCounts[2] != 1 {
- return status.Errorf(codes.Internal, "callCounts[2] should be 0, but got=%d", callCounts[2])
+ if callCounts[2].Load() != 1 {
+ return status.Errorf(codes.Internal, "callCounts[2] should be 0, but got=%d", callCounts[2].Load())
}
- if callCounts[3] != 0 {
- return status.Errorf(codes.Internal, "callCounts[3] should be 0, but got=%d", callCounts[3])
+ if callCounts[3].Load() != 0 {
+ return status.Errorf(codes.Internal, "callCounts[3] should be 0, but got=%d", callCounts[3].Load())
}
- callCounts[3]++
+ callCounts[3].Add(1)
return nil
},
}
@@ -322,7 +323,7 @@
t.Fatalf("failed to recv from stream: %v", err)
}
- if callCounts[3] != 1 {
- t.Fatalf("callCounts[3] should be 1, but got=%d", callCounts[3])
+ if callCounts[3].Load() != 1 {
+ t.Fatalf("callCounts[3] should be 1, but got=%d", callCounts[3].Load())
}
}
diff --git a/test/xds/xds_client_retry_test.go b/test/xds/xds_client_retry_test.go
index 746964e..aebcef6 100644
--- a/test/xds/xds_client_retry_test.go
+++ b/test/xds/xds_client_retry_test.go
@@ -21,6 +21,7 @@
import (
"context"
"fmt"
+ "sync"
"testing"
"google.golang.org/grpc"
@@ -39,6 +40,8 @@
)
func (s) TestClientSideRetry(t *testing.T) {
+ // serverMu guards access to ctr and errs.
+ serverMu := sync.Mutex{}
ctr := 0
errs := []codes.Code{codes.ResourceExhausted}
@@ -46,7 +49,11 @@
server := stubserver.StartTestService(t, &stubserver.StubServer{
EmptyCallF: func(context.Context, *testpb.Empty) (*testpb.Empty, error) {
- defer func() { ctr++ }()
+ serverMu.Lock()
+ defer func() {
+ ctr++
+ serverMu.Unlock()
+ }()
if ctr < len(errs) {
return nil, status.Errorf(errs[ctr], "this should be retried")
}
@@ -151,10 +158,12 @@
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
+ serverMu.Lock()
errs = tc.errs
// Confirm tryAgainErr is correct before updating resources.
ctr = 0
+ serverMu.Unlock()
_, err := client.EmptyCall(ctx, &testpb.Empty{})
if code := status.Code(err); code != tc.tryAgainErr {
t.Fatalf("with old retry policy: EmptyCall() = _, %v; want _, %v", err, tc.tryAgainErr)
@@ -167,7 +176,9 @@
}
for {
+ serverMu.Lock()
ctr = 0
+ serverMu.Unlock()
_, err := client.EmptyCall(ctx, &testpb.Empty{})
if code := status.Code(err); code == tc.tryAgainErr {
continue
diff --git a/test/xds/xds_server_rbac_test.go b/test/xds/xds_server_rbac_test.go
index dcb749f..fe2ab8f 100644
--- a/test/xds/xds_server_rbac_test.go
+++ b/test/xds/xds_server_rbac_test.go
@@ -25,6 +25,7 @@
"net"
"strconv"
"strings"
+ "sync"
"testing"
v3xdsxdstypepb "github.com/cncf/xds/go/xds/type/v3"
@@ -749,6 +750,8 @@
t.Fatalf("UnaryCall() returned err with status: %v, wantStatusUnaryCall: %v", err, test.wantStatusUnaryCall)
}
+ lb.mu.Lock()
+ defer lb.mu.Unlock()
if test.wantAuthzOutcomes != nil {
if diff := cmp.Diff(lb.authzDecisionStat, test.wantAuthzOutcomes); diff != "" {
t.Fatalf("authorization decision do not match\ndiff (-got +want):\n%s", diff)
@@ -956,28 +959,30 @@
}
type statAuditLogger struct {
+ builder *loggerBuilder
+}
+
+func (s *statAuditLogger) Log(event *audit.Event) {
+ b := s.builder
+ b.mu.Lock()
+ defer b.mu.Unlock()
+ b.authzDecisionStat[event.Authorized]++
+ *b.lastEvent = *event
+}
+
+type loggerBuilder struct {
+ mu sync.Mutex
authzDecisionStat map[bool]int // Map to hold counts of authorization decisions
lastEvent *audit.Event // Field to store last received event
}
-func (s *statAuditLogger) Log(event *audit.Event) {
- s.authzDecisionStat[event.Authorized]++
- *s.lastEvent = *event
-}
-
-type loggerBuilder struct {
- authzDecisionStat map[bool]int
- lastEvent *audit.Event
-}
-
-func (loggerBuilder) Name() string {
+func (*loggerBuilder) Name() string {
return "stat_logger"
}
func (lb *loggerBuilder) Build(audit.LoggerConfig) audit.Logger {
return &statAuditLogger{
- authzDecisionStat: lb.authzDecisionStat,
- lastEvent: lb.lastEvent,
+ builder: lb,
}
}
diff --git a/xds/server_test.go b/xds/server_test.go
index df63141..fc94356 100644
--- a/xds/server_test.go
+++ b/xds/server_test.go
@@ -442,6 +442,10 @@
// Update the listener resource on the management server in such a way that
// it will be NACKed by our xDS client. The listener_filters field is
// unsupported and will be NACKed.
+ resources = e2e.UpdateOptions{
+ NodeID: nodeID,
+ Listeners: []*v3listenerpb.Listener{e2e.DefaultServerListener(host, port, e2e.SecurityLevelNone, "routeName")},
+ }
resources.Listeners[0].ListenerFilters = []*v3listenerpb.ListenerFilter{{Name: "foo"}}
if err := mgmtServer.Update(ctx, resources); err != nil {
t.Fatal(err)
