idle_e2e_test.go

Documentation: google.golang.org/grpc/internal/idle

     1  /*
     2   *
     3   * Copyright 2023 gRPC authors.
     4   *
     5   * Licensed under the Apache License, Version 2.0 (the "License");
     6   * you may not use this file except in compliance with the License.
     7   * You may obtain a copy of the License at
     8   *
     9   *     http://www.apache.org/licenses/LICENSE-2.0
    10   *
    11   * Unless required by applicable law or agreed to in writing, software
    12   * distributed under the License is distributed on an "AS IS" BASIS,
    13   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    14   * See the License for the specific language governing permissions and
    15   * limitations under the License.
    16   *
    17   */
    18  
    19  package idle_test
    20  
    21  import (
    22  	"context"
    23  	"fmt"
    24  	"io"
    25  	"strings"
    26  	"sync"
    27  	"testing"
    28  	"time"
    29  
    30  	"google.golang.org/grpc"
    31  	"google.golang.org/grpc/balancer"
    32  	"google.golang.org/grpc/balancer/roundrobin"
    33  	"google.golang.org/grpc/codes"
    34  	"google.golang.org/grpc/connectivity"
    35  	"google.golang.org/grpc/credentials/insecure"
    36  	"google.golang.org/grpc/internal"
    37  	"google.golang.org/grpc/internal/balancer/stub"
    38  	"google.golang.org/grpc/internal/channelz"
    39  	"google.golang.org/grpc/internal/grpctest"
    40  	"google.golang.org/grpc/internal/stubserver"
    41  	"google.golang.org/grpc/internal/testutils"
    42  	"google.golang.org/grpc/resolver"
    43  	"google.golang.org/grpc/resolver/manual"
    44  	"google.golang.org/grpc/status"
    45  
    46  	testgrpc "google.golang.org/grpc/interop/grpc_testing"
    47  	testpb "google.golang.org/grpc/interop/grpc_testing"
    48  )
    49  
    50  func init() {
    51  	channelz.TurnOn()
    52  }
    53  
    54  type s struct {
    55  	grpctest.Tester
    56  }
    57  
    58  func Test(t *testing.T) {
    59  	grpctest.RunSubTests(t, s{})
    60  }
    61  
    62  const (
    63  	defaultTestTimeout          = 10 * time.Second
    64  	defaultTestShortTimeout     = 100 * time.Millisecond
    65  	defaultTestShortIdleTimeout = 500 * time.Millisecond
    66  )
    67  
    68  // channelzTraceEventFound looks up the top-channels in channelz (expects a
    69  // single one), and checks if there is a trace event on the channel matching the
    70  // provided description string.
    71  func channelzTraceEventFound(ctx context.Context, wantDesc string) error {
    72  	for ctx.Err() == nil {
    73  		tcs, _ := channelz.GetTopChannels(0, 0)
    74  		if l := len(tcs); l != 1 {
    75  			return fmt.Errorf("when looking for channelz trace event with description %q, found %d top-level channels, want 1", wantDesc, l)
    76  		}
    77  		trace := tcs[0].Trace()
    78  		if trace == nil {
    79  			return fmt.Errorf("when looking for channelz trace event with description %q, no trace events found for top-level channel", wantDesc)
    80  		}
    81  
    82  		for _, e := range trace.Events {
    83  			if strings.Contains(e.Desc, wantDesc) {
    84  				return nil
    85  			}
    86  		}
    87  	}
    88  	return fmt.Errorf("when looking for channelz trace event with description %q, %w", wantDesc, ctx.Err())
    89  }
    90  
    91  // Registers a wrapped round_robin LB policy for the duration of this test that
    92  // retains all the functionality of the round_robin LB policy and makes the
    93  // balancer close event available for inspection by the test.
    94  //
    95  // Returns a channel that gets pinged when the balancer is closed.
    96  func registerWrappedRoundRobinPolicy(t *testing.T) chan struct{} {
    97  	rrBuilder := balancer.Get(roundrobin.Name)
    98  	closeCh := make(chan struct{}, 1)
    99  	stub.Register(roundrobin.Name, stub.BalancerFuncs{
   100  		Init: func(bd *stub.BalancerData) {
   101  			bd.Data = rrBuilder.Build(bd.ClientConn, bd.BuildOptions)
   102  		},
   103  		UpdateClientConnState: func(bd *stub.BalancerData, ccs balancer.ClientConnState) error {
   104  			bal := bd.Data.(balancer.Balancer)
   105  			return bal.UpdateClientConnState(ccs)
   106  		},
   107  		Close: func(bd *stub.BalancerData) {
   108  			select {
   109  			case closeCh <- struct{}{}:
   110  			default:
   111  			}
   112  			bal := bd.Data.(balancer.Balancer)
   113  			bal.Close()
   114  		},
   115  	})
   116  	t.Cleanup(func() { balancer.Register(rrBuilder) })
   117  
   118  	return closeCh
   119  }
   120  
   121  // Tests the case where channel idleness is disabled by passing an idle_timeout
   122  // of 0. Verifies that a READY channel with no RPCs does not move to IDLE.
   123  func (s) TestChannelIdleness_Disabled_NoActivity(t *testing.T) {
   124  	closeCh := registerWrappedRoundRobinPolicy(t)
   125  
   126  	// Create a ClientConn with idle_timeout set to 0.
   127  	r := manual.NewBuilderWithScheme("whatever")
   128  	dopts := []grpc.DialOption{
   129  		grpc.WithTransportCredentials(insecure.NewCredentials()),
   130  		grpc.WithResolvers(r),
   131  		grpc.WithIdleTimeout(0), // Disable idleness.
   132  		grpc.WithDefaultServiceConfig(`{"loadBalancingConfig": [{"round_robin":{}}]}`),
   133  	}
   134  	cc, err := grpc.Dial(r.Scheme()+":///test.server", dopts...)
   135  	if err != nil {
   136  		t.Fatalf("grpc.Dial() failed: %v", err)
   137  	}
   138  	defer cc.Close()
   139  
   140  	// Start a test backend and push an address update via the resolver.
   141  	backend := stubserver.StartTestService(t, nil)
   142  	defer backend.Stop()
   143  	r.UpdateState(resolver.State{Addresses: []resolver.Address{{Addr: backend.Address}}})
   144  
   145  	// Verify that the ClientConn moves to READY.
   146  	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
   147  	defer cancel()
   148  	testutils.AwaitState(ctx, t, cc, connectivity.Ready)
   149  
   150  	// Verify that the ClientConn stays in READY.
   151  	sCtx, sCancel := context.WithTimeout(ctx, 3*defaultTestShortIdleTimeout)
   152  	defer sCancel()
   153  	testutils.AwaitNoStateChange(sCtx, t, cc, connectivity.Ready)
   154  
   155  	// Verify that the LB policy is not closed which is expected to happen when
   156  	// the channel enters IDLE.
   157  	sCtx, sCancel = context.WithTimeout(ctx, defaultTestShortIdleTimeout)
   158  	defer sCancel()
   159  	select {
   160  	case <-sCtx.Done():
   161  	case <-closeCh:
   162  		t.Fatal("LB policy closed when expected not to")
   163  	}
   164  }
   165  
   166  // Tests the case where channel idleness is enabled by passing a small value for
   167  // idle_timeout. Verifies that a READY channel with no RPCs moves to IDLE, and
   168  // the connection to the backend is closed.
   169  func (s) TestChannelIdleness_Enabled_NoActivity(t *testing.T) {
   170  	closeCh := registerWrappedRoundRobinPolicy(t)
   171  
   172  	// Create a ClientConn with a short idle_timeout.
   173  	r := manual.NewBuilderWithScheme("whatever")
   174  	dopts := []grpc.DialOption{
   175  		grpc.WithTransportCredentials(insecure.NewCredentials()),
   176  		grpc.WithResolvers(r),
   177  		grpc.WithIdleTimeout(defaultTestShortIdleTimeout),
   178  		grpc.WithDefaultServiceConfig(`{"loadBalancingConfig": [{"round_robin":{}}]}`),
   179  	}
   180  	cc, err := grpc.Dial(r.Scheme()+":///test.server", dopts...)
   181  	if err != nil {
   182  		t.Fatalf("grpc.Dial() failed: %v", err)
   183  	}
   184  	defer cc.Close()
   185  
   186  	// Start a test backend and push an address update via the resolver.
   187  	lis := testutils.NewListenerWrapper(t, nil)
   188  	backend := stubserver.StartTestService(t, &stubserver.StubServer{Listener: lis})
   189  	defer backend.Stop()
   190  	r.UpdateState(resolver.State{Addresses: []resolver.Address{{Addr: backend.Address}}})
   191  
   192  	// Verify that the ClientConn moves to READY.
   193  	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
   194  	defer cancel()
   195  	testutils.AwaitState(ctx, t, cc, connectivity.Ready)
   196  
   197  	// Retrieve the wrapped conn from the listener.
   198  	v, err := lis.NewConnCh.Receive(ctx)
   199  	if err != nil {
   200  		t.Fatalf("Failed to retrieve conn from test listener: %v", err)
   201  	}
   202  	conn := v.(*testutils.ConnWrapper)
   203  
   204  	// Verify that the ClientConn moves to IDLE as there is no activity.
   205  	testutils.AwaitState(ctx, t, cc, connectivity.Idle)
   206  
   207  	// Verify idleness related channelz events.
   208  	if err := channelzTraceEventFound(ctx, "entering idle mode"); err != nil {
   209  		t.Fatal(err)
   210  	}
   211  
   212  	// Verify that the previously open connection is closed.
   213  	if _, err := conn.CloseCh.Receive(ctx); err != nil {
   214  		t.Fatalf("Failed when waiting for connection to be closed after channel entered IDLE: %v", err)
   215  	}
   216  
   217  	// Verify that the LB policy is closed.
   218  	select {
   219  	case <-ctx.Done():
   220  		t.Fatal("Timeout waiting for LB policy to be closed after the channel enters IDLE")
   221  	case <-closeCh:
   222  	}
   223  }
   224  
   225  // Tests the case where channel idleness is enabled by passing a small value for
   226  // idle_timeout. Verifies that a READY channel with an ongoing RPC stays READY.
   227  func (s) TestChannelIdleness_Enabled_OngoingCall(t *testing.T) {
   228  	tests := []struct {
   229  		name    string
   230  		makeRPC func(ctx context.Context, client testgrpc.TestServiceClient) error
   231  	}{
   232  		{
   233  			name: "unary",
   234  			makeRPC: func(ctx context.Context, client testgrpc.TestServiceClient) error {
   235  				if _, err := client.EmptyCall(ctx, &testpb.Empty{}); err != nil {
   236  					return fmt.Errorf("EmptyCall RPC failed: %v", err)
   237  				}
   238  				return nil
   239  			},
   240  		},
   241  		{
   242  			name: "streaming",
   243  			makeRPC: func(ctx context.Context, client testgrpc.TestServiceClient) error {
   244  				stream, err := client.FullDuplexCall(ctx)
   245  				if err != nil {
   246  					t.Fatalf("FullDuplexCall RPC failed: %v", err)
   247  				}
   248  				if _, err := stream.Recv(); err != nil && err != io.EOF {
   249  					t.Fatalf("stream.Recv() failed: %v", err)
   250  				}
   251  				return nil
   252  			},
   253  		},
   254  	}
   255  
   256  	for _, test := range tests {
   257  		t.Run(test.name, func(t *testing.T) {
   258  			closeCh := registerWrappedRoundRobinPolicy(t)
   259  
   260  			// Create a ClientConn with a short idle_timeout.
   261  			r := manual.NewBuilderWithScheme("whatever")
   262  			dopts := []grpc.DialOption{
   263  				grpc.WithTransportCredentials(insecure.NewCredentials()),
   264  				grpc.WithResolvers(r),
   265  				grpc.WithIdleTimeout(defaultTestShortIdleTimeout),
   266  				grpc.WithDefaultServiceConfig(`{"loadBalancingConfig": [{"round_robin":{}}]}`),
   267  			}
   268  			cc, err := grpc.Dial(r.Scheme()+":///test.server", dopts...)
   269  			if err != nil {
   270  				t.Fatalf("grpc.Dial() failed: %v", err)
   271  			}
   272  			defer cc.Close()
   273  
   274  			// Start a test backend that keeps the RPC call active by blocking
   275  			// on a channel that is closed by the test later on.
   276  			blockCh := make(chan struct{})
   277  			backend := &stubserver.StubServer{
   278  				EmptyCallF: func(ctx context.Context, in *testpb.Empty) (*testpb.Empty, error) {
   279  					<-blockCh
   280  					return &testpb.Empty{}, nil
   281  				},
   282  				FullDuplexCallF: func(stream testgrpc.TestService_FullDuplexCallServer) error {
   283  					<-blockCh
   284  					return nil
   285  				},
   286  			}
   287  			if err := backend.StartServer(); err != nil {
   288  				t.Fatalf("Failed to start backend: %v", err)
   289  			}
   290  			defer backend.Stop()
   291  
   292  			// Push an address update containing the address of the above
   293  			// backend via the manual resolver.
   294  			r.UpdateState(resolver.State{Addresses: []resolver.Address{{Addr: backend.Address}}})
   295  
   296  			// Verify that the ClientConn moves to READY.
   297  			ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
   298  			defer cancel()
   299  			testutils.AwaitState(ctx, t, cc, connectivity.Ready)
   300  
   301  			// Spawn a goroutine to check for expected behavior while a blocking
   302  			// RPC all is made from the main test goroutine.
   303  			errCh := make(chan error, 1)
   304  			go func() {
   305  				defer close(blockCh)
   306  
   307  				// Verify that the ClientConn stays in READY.
   308  				sCtx, sCancel := context.WithTimeout(ctx, 3*defaultTestShortIdleTimeout)
   309  				defer sCancel()
   310  				if cc.WaitForStateChange(sCtx, connectivity.Ready) {
   311  					errCh <- fmt.Errorf("state changed from %q to %q when no state change was expected", connectivity.Ready, cc.GetState())
   312  					return
   313  				}
   314  
   315  				// Verify that the LB policy is not closed which is expected to happen when
   316  				// the channel enters IDLE.
   317  				sCtx, sCancel = context.WithTimeout(ctx, defaultTestShortIdleTimeout)
   318  				defer sCancel()
   319  				select {
   320  				case <-sCtx.Done():
   321  				case <-closeCh:
   322  					errCh <- fmt.Errorf("LB policy closed when expected not to")
   323  				}
   324  				errCh <- nil
   325  			}()
   326  
   327  			if err := test.makeRPC(ctx, testgrpc.NewTestServiceClient(cc)); err != nil {
   328  				t.Fatalf("%s rpc failed: %v", test.name, err)
   329  			}
   330  
   331  			select {
   332  			case err := <-errCh:
   333  				if err != nil {
   334  					t.Fatal(err)
   335  				}
   336  			case <-ctx.Done():
   337  				t.Fatalf("Timeout when trying to verify that an active RPC keeps channel from moving to IDLE")
   338  			}
   339  		})
   340  	}
   341  }
   342  
   343  // Tests the case where channel idleness is enabled by passing a small value for
   344  // idle_timeout. Verifies that activity on a READY channel (frequent and short
   345  // RPCs) keeps it from moving to IDLE.
   346  func (s) TestChannelIdleness_Enabled_ActiveSinceLastCheck(t *testing.T) {
   347  	closeCh := registerWrappedRoundRobinPolicy(t)
   348  
   349  	// Create a ClientConn with a short idle_timeout.
   350  	r := manual.NewBuilderWithScheme("whatever")
   351  	dopts := []grpc.DialOption{
   352  		grpc.WithTransportCredentials(insecure.NewCredentials()),
   353  		grpc.WithResolvers(r),
   354  		grpc.WithIdleTimeout(defaultTestShortIdleTimeout),
   355  		grpc.WithDefaultServiceConfig(`{"loadBalancingConfig": [{"round_robin":{}}]}`),
   356  	}
   357  	cc, err := grpc.Dial(r.Scheme()+":///test.server", dopts...)
   358  	if err != nil {
   359  		t.Fatalf("grpc.Dial() failed: %v", err)
   360  	}
   361  	defer cc.Close()
   362  
   363  	// Start a test backend and push an address update via the resolver.
   364  	backend := stubserver.StartTestService(t, nil)
   365  	defer backend.Stop()
   366  	r.UpdateState(resolver.State{Addresses: []resolver.Address{{Addr: backend.Address}}})
   367  
   368  	// Verify that the ClientConn moves to READY.
   369  	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
   370  	defer cancel()
   371  	testutils.AwaitState(ctx, t, cc, connectivity.Ready)
   372  
   373  	// For a duration of three times the configured idle timeout, making RPCs
   374  	// every now and then and ensure that the channel does not move out of
   375  	// READY.
   376  	sCtx, sCancel := context.WithTimeout(ctx, 3*defaultTestShortIdleTimeout)
   377  	defer sCancel()
   378  	go func() {
   379  		for ; sCtx.Err() == nil; <-time.After(defaultTestShortIdleTimeout / 4) {
   380  			client := testgrpc.NewTestServiceClient(cc)
   381  			if _, err := client.EmptyCall(sCtx, &testpb.Empty{}); err != nil {
   382  				// While iterating through this for loop, at some point in time,
   383  				// the context deadline will expire. It is safe to ignore that
   384  				// error code.
   385  				if status.Code(err) != codes.DeadlineExceeded {
   386  					t.Errorf("EmptyCall RPC failed: %v", err)
   387  					return
   388  				}
   389  			}
   390  		}
   391  	}()
   392  
   393  	// Verify that the ClientConn stays in READY.
   394  	testutils.AwaitNoStateChange(sCtx, t, cc, connectivity.Ready)
   395  
   396  	// Verify that the LB policy is not closed which is expected to happen when
   397  	// the channel enters IDLE.
   398  	select {
   399  	case <-sCtx.Done():
   400  	case <-closeCh:
   401  		t.Fatal("LB policy closed when expected not to")
   402  	}
   403  }
   404  
   405  // Tests the case where channel idleness is enabled by passing a small value for
   406  // idle_timeout. Verifies that a READY channel with no RPCs moves to IDLE. Also
   407  // verifies that a subsequent RPC on the IDLE channel kicks it out of IDLE.
   408  func (s) TestChannelIdleness_Enabled_ExitIdleOnRPC(t *testing.T) {
   409  	closeCh := registerWrappedRoundRobinPolicy(t)
   410  
   411  	// Start a test backend and set the bootstrap state of the resolver to
   412  	// include this address. This will ensure that when the resolver is
   413  	// restarted when exiting idle, it will push the same address to grpc again.
   414  	r := manual.NewBuilderWithScheme("whatever")
   415  	backend := stubserver.StartTestService(t, nil)
   416  	defer backend.Stop()
   417  	r.InitialState(resolver.State{Addresses: []resolver.Address{{Addr: backend.Address}}})
   418  
   419  	// Create a ClientConn with a short idle_timeout.
   420  	dopts := []grpc.DialOption{
   421  		grpc.WithTransportCredentials(insecure.NewCredentials()),
   422  		grpc.WithResolvers(r),
   423  		grpc.WithIdleTimeout(defaultTestShortIdleTimeout),
   424  		grpc.WithDefaultServiceConfig(`{"loadBalancingConfig": [{"round_robin":{}}]}`),
   425  	}
   426  	cc, err := grpc.Dial(r.Scheme()+":///test.server", dopts...)
   427  	if err != nil {
   428  		t.Fatalf("grpc.Dial() failed: %v", err)
   429  	}
   430  	defer cc.Close()
   431  
   432  	// Verify that the ClientConn moves to READY.
   433  	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
   434  	defer cancel()
   435  	testutils.AwaitState(ctx, t, cc, connectivity.Ready)
   436  
   437  	// Verify that the ClientConn moves to IDLE as there is no activity.
   438  	testutils.AwaitState(ctx, t, cc, connectivity.Idle)
   439  
   440  	// Verify idleness related channelz events.
   441  	if err := channelzTraceEventFound(ctx, "entering idle mode"); err != nil {
   442  		t.Fatal(err)
   443  	}
   444  
   445  	// Verify that the LB policy is closed.
   446  	select {
   447  	case <-ctx.Done():
   448  		t.Fatal("Timeout waiting for LB policy to be closed after the channel enters IDLE")
   449  	case <-closeCh:
   450  	}
   451  
   452  	// Make an RPC and ensure that it succeeds and moves the channel back to
   453  	// READY.
   454  	client := testgrpc.NewTestServiceClient(cc)
   455  	if _, err := client.EmptyCall(ctx, &testpb.Empty{}); err != nil {
   456  		t.Fatalf("EmptyCall RPC failed: %v", err)
   457  	}
   458  	testutils.AwaitState(ctx, t, cc, connectivity.Ready)
   459  	if err := channelzTraceEventFound(ctx, "exiting idle mode"); err != nil {
   460  		t.Fatal(err)
   461  	}
   462  }
   463  
   464  // Tests the case where channel idleness is enabled by passing a small value for
   465  // idle_timeout. Simulates a race between the idle timer firing and RPCs being
   466  // initiated, after a period of inactivity on the channel.
   467  //
   468  // After a period of inactivity (for the configured idle timeout duration), when
   469  // RPCs are started, there are two possibilities:
   470  //   - the idle timer wins the race and puts the channel in idle. The RPCs then
   471  //     kick it out of idle.
   472  //   - the RPCs win the race, and therefore the channel never moves to idle.
   473  //
   474  // In either of these cases, all RPCs must succeed.
   475  func (s) TestChannelIdleness_Enabled_IdleTimeoutRacesWithRPCs(t *testing.T) {
   476  	// Start a test backend and set the bootstrap state of the resolver to
   477  	// include this address. This will ensure that when the resolver is
   478  	// restarted when exiting idle, it will push the same address to grpc again.
   479  	r := manual.NewBuilderWithScheme("whatever")
   480  	backend := stubserver.StartTestService(t, nil)
   481  	defer backend.Stop()
   482  	r.InitialState(resolver.State{Addresses: []resolver.Address{{Addr: backend.Address}}})
   483  
   484  	// Create a ClientConn with a short idle_timeout.
   485  	dopts := []grpc.DialOption{
   486  		grpc.WithTransportCredentials(insecure.NewCredentials()),
   487  		grpc.WithResolvers(r),
   488  		grpc.WithIdleTimeout(defaultTestShortTimeout),
   489  		grpc.WithDefaultServiceConfig(`{"loadBalancingConfig": [{"round_robin":{}}]}`),
   490  	}
   491  	cc, err := grpc.NewClient(r.Scheme()+":///test.server", dopts...)
   492  	if err != nil {
   493  		t.Fatalf("grpc.NewClient() failed: %v", err)
   494  	}
   495  	defer cc.Close()
   496  
   497  	// Verify that the ClientConn moves to READY.
   498  	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
   499  	defer cancel()
   500  	client := testgrpc.NewTestServiceClient(cc)
   501  	if _, err := client.EmptyCall(ctx, &testpb.Empty{}); err != nil {
   502  		t.Errorf("EmptyCall RPC failed: %v", err)
   503  	}
   504  
   505  	// Make an RPC every defaultTestShortTimeout duration so as to race with the
   506  	// idle timeout. Whether the idle timeout wins the race or the RPC wins the
   507  	// race, RPCs must succeed.
   508  	for i := 0; i < 20; i++ {
   509  		<-time.After(defaultTestShortTimeout)
   510  		if _, err := client.EmptyCall(ctx, &testpb.Empty{}); err != nil {
   511  			t.Fatalf("EmptyCall RPC failed: %v", err)
   512  		}
   513  		t.Logf("Iteration %d succeeded", i)
   514  	}
   515  }
   516  
   517  // Tests the case where the channel is IDLE and we call cc.Connect.
   518  func (s) TestChannelIdleness_Connect(t *testing.T) {
   519  	// Start a test backend and set the bootstrap state of the resolver to
   520  	// include this address. This will ensure that when the resolver is
   521  	// restarted when exiting idle, it will push the same address to grpc again.
   522  	r := manual.NewBuilderWithScheme("whatever")
   523  	backend := stubserver.StartTestService(t, nil)
   524  	defer backend.Stop()
   525  	r.InitialState(resolver.State{Addresses: []resolver.Address{{Addr: backend.Address}}})
   526  
   527  	// Create a ClientConn with a short idle_timeout.
   528  	dopts := []grpc.DialOption{
   529  		grpc.WithTransportCredentials(insecure.NewCredentials()),
   530  		grpc.WithResolvers(r),
   531  		grpc.WithIdleTimeout(defaultTestShortIdleTimeout),
   532  		grpc.WithDefaultServiceConfig(`{"loadBalancingConfig": [{"round_robin":{}}]}`),
   533  	}
   534  	cc, err := grpc.NewClient(r.Scheme()+":///test.server", dopts...)
   535  	if err != nil {
   536  		t.Fatalf("grpc.NewClient() failed: %v", err)
   537  	}
   538  	defer cc.Close()
   539  
   540  	// Verify that the ClientConn moves to IDLE.
   541  	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
   542  	defer cancel()
   543  
   544  	testutils.AwaitState(ctx, t, cc, connectivity.Idle)
   545  
   546  	// Connect should exit channel idleness.
   547  	cc.Connect()
   548  
   549  	// Verify that the ClientConn moves back to READY.
   550  	testutils.AwaitState(ctx, t, cc, connectivity.Ready)
   551  }
   552  
   553  // runFunc runs f repeatedly until the context expires.
   554  func runFunc(ctx context.Context, f func()) {
   555  	for {
   556  		select {
   557  		case <-ctx.Done():
   558  			return
   559  		case <-time.After(10 * time.Millisecond):
   560  			f()
   561  		}
   562  	}
   563  }
   564  
   565  // Tests the scenario where there are concurrent calls to exit and enter idle
   566  // mode on the ClientConn. Verifies that there is no race under this scenario.
   567  func (s) TestChannelIdleness_RaceBetweenEnterAndExitIdleMode(t *testing.T) {
   568  	// Start a test backend and set the bootstrap state of the resolver to
   569  	// include this address. This will ensure that when the resolver is
   570  	// restarted when exiting idle, it will push the same address to grpc again.
   571  	r := manual.NewBuilderWithScheme("whatever")
   572  	backend := stubserver.StartTestService(t, nil)
   573  	defer backend.Stop()
   574  	r.InitialState(resolver.State{Addresses: []resolver.Address{{Addr: backend.Address}}})
   575  
   576  	// Create a ClientConn with a long idle_timeout. We will explicitly trigger
   577  	// entering and exiting IDLE mode from the test.
   578  	dopts := []grpc.DialOption{
   579  		grpc.WithTransportCredentials(insecure.NewCredentials()),
   580  		grpc.WithResolvers(r),
   581  		grpc.WithIdleTimeout(30 * time.Minute),
   582  		grpc.WithDefaultServiceConfig(`{"loadBalancingConfig": [{"pick_first":{}}]}`),
   583  	}
   584  	cc, err := grpc.NewClient(r.Scheme()+":///test.server", dopts...)
   585  	if err != nil {
   586  		t.Fatalf("grpc.NewClient() failed: %v", err)
   587  	}
   588  	defer cc.Close()
   589  
   590  	enterIdle := internal.EnterIdleModeForTesting.(func(*grpc.ClientConn))
   591  	enterIdleFunc := func() { enterIdle(cc) }
   592  	exitIdle := internal.ExitIdleModeForTesting.(func(*grpc.ClientConn) error)
   593  	exitIdleFunc := func() {
   594  		if err := exitIdle(cc); err != nil {
   595  			t.Errorf("Failed to exit idle mode: %v", err)
   596  		}
   597  	}
   598  	// Spawn goroutines that call methods on the ClientConn to enter and exit
   599  	// idle mode concurrently for one second.
   600  	ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
   601  	defer cancel()
   602  	var wg sync.WaitGroup
   603  	wg.Add(4)
   604  	go func() {
   605  		runFunc(ctx, enterIdleFunc)
   606  		wg.Done()
   607  	}()
   608  	go func() {
   609  		runFunc(ctx, enterIdleFunc)
   610  		wg.Done()
   611  	}()
   612  	go func() {
   613  		runFunc(ctx, exitIdleFunc)
   614  		wg.Done()
   615  	}()
   616  	go func() {
   617  		runFunc(ctx, exitIdleFunc)
   618  		wg.Done()
   619  	}()
   620  	wg.Wait()
   621  }
   622
View as plain text