idle_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
    20  
    21  import (
    22  	"context"
    23  	"fmt"
    24  	"sync"
    25  	"sync/atomic"
    26  	"testing"
    27  	"time"
    28  
    29  	"google.golang.org/grpc/internal/grpctest"
    30  )
    31  
    32  const (
    33  	defaultTestTimeout      = 10 * time.Second
    34  	defaultTestIdleTimeout  = 500 * time.Millisecond // A short idle_timeout for tests.
    35  	defaultTestShortTimeout = 10 * time.Millisecond  // A small deadline to wait for events expected to not happen.
    36  )
    37  
    38  type s struct {
    39  	grpctest.Tester
    40  }
    41  
    42  func Test(t *testing.T) {
    43  	grpctest.RunSubTests(t, s{})
    44  }
    45  
    46  type testEnforcer struct {
    47  	exitIdleCh  chan struct{}
    48  	enterIdleCh chan struct{}
    49  }
    50  
    51  func (ti *testEnforcer) ExitIdleMode() error {
    52  	ti.exitIdleCh <- struct{}{}
    53  	return nil
    54  
    55  }
    56  
    57  func (ti *testEnforcer) EnterIdleMode() {
    58  	ti.enterIdleCh <- struct{}{}
    59  }
    60  
    61  func newTestEnforcer() *testEnforcer {
    62  	return &testEnforcer{
    63  		exitIdleCh:  make(chan struct{}, 1),
    64  		enterIdleCh: make(chan struct{}, 1),
    65  	}
    66  }
    67  
    68  // overrideNewTimer overrides the new timer creation function by ensuring that a
    69  // message is pushed on the returned channel everytime the timer fires.
    70  func overrideNewTimer(t *testing.T) <-chan struct{} {
    71  	t.Helper()
    72  
    73  	ch := make(chan struct{}, 1)
    74  	origTimeAfterFunc := timeAfterFunc
    75  	timeAfterFunc = func(d time.Duration, callback func()) *time.Timer {
    76  		return time.AfterFunc(d, func() {
    77  			select {
    78  			case ch <- struct{}{}:
    79  			default:
    80  			}
    81  			callback()
    82  		})
    83  	}
    84  	t.Cleanup(func() { timeAfterFunc = origTimeAfterFunc })
    85  	return ch
    86  }
    87  
    88  // TestManager_Disabled tests the case where the idleness manager is
    89  // disabled by passing an idle_timeout of 0. Verifies the following things:
    90  //   - timer callback does not fire
    91  //   - an RPC triggers a call to ExitIdleMode on the ClientConn
    92  //   - more calls to RPC termination (as compared to RPC initiation) does not
    93  //     result in an error log
    94  func (s) TestManager_Disabled(t *testing.T) {
    95  	callbackCh := overrideNewTimer(t)
    96  
    97  	// Create an idleness manager that is disabled because of idleTimeout being
    98  	// set to `0`.
    99  	enforcer := newTestEnforcer()
   100  	mgr := NewManager(enforcer, time.Duration(0))
   101  
   102  	// Ensure that the timer callback does not fire within a short deadline.
   103  	select {
   104  	case <-callbackCh:
   105  		t.Fatal("Idle timer callback fired when manager is disabled")
   106  	case <-time.After(defaultTestShortTimeout):
   107  	}
   108  
   109  	// The first invocation of OnCallBegin() should lead to a call to
   110  	// ExitIdleMode() on the enforcer.
   111  	go mgr.OnCallBegin()
   112  	select {
   113  	case <-enforcer.exitIdleCh:
   114  	case <-time.After(defaultTestShortTimeout):
   115  		t.Fatal("Timeout waiting for channel to move out of idle mode")
   116  	}
   117  
   118  	// If the number of calls to OnCallEnd() exceeds the number of calls to
   119  	// OnCallBegin(), the idleness manager is expected to throw an error log
   120  	// (which will cause our TestLogger to fail the test). But since the manager
   121  	// is disabled, this should not happen.
   122  	mgr.OnCallEnd()
   123  	mgr.OnCallEnd()
   124  
   125  	// The idleness manager is explicitly not closed here. But since the manager
   126  	// is disabled, it will not start the run goroutine, and hence we expect the
   127  	// leakchecker to not find any leaked goroutines.
   128  }
   129  
   130  // TestManager_Enabled_TimerFires tests the case where the idle manager
   131  // is enabled. Ensures that when there are no RPCs, the timer callback is
   132  // invoked and the EnterIdleMode() method is invoked on the enforcer.
   133  func (s) TestManager_Enabled_TimerFires(t *testing.T) {
   134  	callbackCh := overrideNewTimer(t)
   135  
   136  	enforcer := newTestEnforcer()
   137  	mgr := NewManager(enforcer, time.Duration(defaultTestIdleTimeout))
   138  	defer mgr.Close()
   139  	mgr.ExitIdleMode()
   140  
   141  	// Ensure that the timer callback fires within a appropriate amount of time.
   142  	select {
   143  	case <-callbackCh:
   144  	case <-time.After(2 * defaultTestIdleTimeout):
   145  		t.Fatal("Timeout waiting for idle timer callback to fire")
   146  	}
   147  
   148  	// Ensure that the channel moves to idle mode eventually.
   149  	select {
   150  	case <-enforcer.enterIdleCh:
   151  	case <-time.After(defaultTestTimeout):
   152  		t.Fatal("Timeout waiting for channel to move to idle")
   153  	}
   154  }
   155  
   156  // TestManager_Enabled_OngoingCall tests the case where the idle manager
   157  // is enabled. Ensures that when there is an ongoing RPC, the channel does not
   158  // enter idle mode.
   159  func (s) TestManager_Enabled_OngoingCall(t *testing.T) {
   160  	callbackCh := overrideNewTimer(t)
   161  
   162  	enforcer := newTestEnforcer()
   163  	mgr := NewManager(enforcer, time.Duration(defaultTestIdleTimeout))
   164  	defer mgr.Close()
   165  	mgr.ExitIdleMode()
   166  
   167  	// Fire up a goroutine that simulates an ongoing RPC that is terminated
   168  	// after the timer callback fires for the first time.
   169  	timerFired := make(chan struct{})
   170  	go func() {
   171  		mgr.OnCallBegin()
   172  		<-timerFired
   173  		mgr.OnCallEnd()
   174  	}()
   175  
   176  	// Ensure that the timer callback fires and unblock the above goroutine.
   177  	select {
   178  	case <-callbackCh:
   179  		close(timerFired)
   180  	case <-time.After(2 * defaultTestIdleTimeout):
   181  		t.Fatal("Timeout waiting for idle timer callback to fire")
   182  	}
   183  
   184  	// The invocation of the timer callback should not put the channel in idle
   185  	// mode since we had an ongoing RPC.
   186  	select {
   187  	case <-enforcer.enterIdleCh:
   188  		t.Fatalf("EnterIdleMode() called on enforcer when active RPC exists")
   189  	case <-time.After(defaultTestShortTimeout):
   190  	}
   191  
   192  	// Since we terminated the ongoing RPC and we have no other active RPCs, the
   193  	// channel must move to idle eventually.
   194  	select {
   195  	case <-enforcer.enterIdleCh:
   196  	case <-time.After(defaultTestTimeout):
   197  		t.Fatal("Timeout waiting for channel to move to idle")
   198  	}
   199  }
   200  
   201  // TestManager_Enabled_ActiveSinceLastCheck tests the case where the
   202  // idle manager is enabled. Ensures that when there are active RPCs in the last
   203  // period (even though there is no active call when the timer fires), the
   204  // channel does not enter idle mode.
   205  func (s) TestManager_Enabled_ActiveSinceLastCheck(t *testing.T) {
   206  	callbackCh := overrideNewTimer(t)
   207  
   208  	enforcer := newTestEnforcer()
   209  	mgr := NewManager(enforcer, time.Duration(defaultTestIdleTimeout))
   210  	defer mgr.Close()
   211  	mgr.ExitIdleMode()
   212  
   213  	// Fire up a goroutine that simulates unary RPCs until the timer callback
   214  	// fires.
   215  	timerFired := make(chan struct{})
   216  	go func() {
   217  		for ; ; <-time.After(defaultTestShortTimeout) {
   218  			mgr.OnCallBegin()
   219  			mgr.OnCallEnd()
   220  
   221  			select {
   222  			case <-timerFired:
   223  				return
   224  			default:
   225  			}
   226  		}
   227  	}()
   228  
   229  	// Ensure that the timer callback fires, and that we don't enter idle as
   230  	// part of this invocation of the timer callback, since we had some RPCs in
   231  	// this period.
   232  	select {
   233  	case <-callbackCh:
   234  		close(timerFired)
   235  	case <-time.After(2 * defaultTestIdleTimeout):
   236  		close(timerFired)
   237  		t.Fatal("Timeout waiting for idle timer callback to fire")
   238  	}
   239  	select {
   240  	case <-enforcer.enterIdleCh:
   241  		t.Fatalf("EnterIdleMode() called on enforcer when one RPC completed in the last period")
   242  	case <-time.After(defaultTestShortTimeout):
   243  	}
   244  
   245  	// Since the unrary RPC terminated and we have no other active RPCs, the
   246  	// channel must move to idle eventually.
   247  	select {
   248  	case <-enforcer.enterIdleCh:
   249  	case <-time.After(defaultTestTimeout):
   250  		t.Fatal("Timeout waiting for channel to move to idle")
   251  	}
   252  }
   253  
   254  // TestManager_Enabled_ExitIdleOnRPC tests the case where the idle
   255  // manager is enabled. Ensures that the channel moves out of idle when an RPC is
   256  // initiated.
   257  func (s) TestManager_Enabled_ExitIdleOnRPC(t *testing.T) {
   258  	overrideNewTimer(t)
   259  
   260  	enforcer := newTestEnforcer()
   261  	mgr := NewManager(enforcer, time.Duration(defaultTestIdleTimeout))
   262  	defer mgr.Close()
   263  
   264  	mgr.ExitIdleMode()
   265  	<-enforcer.exitIdleCh
   266  	// Ensure that the channel moves to idle since there are no RPCs.
   267  	select {
   268  	case <-enforcer.enterIdleCh:
   269  	case <-time.After(2 * defaultTestIdleTimeout):
   270  		t.Fatal("Timeout waiting for channel to move to idle mode")
   271  	}
   272  
   273  	for i := 0; i < 100; i++ {
   274  		// A call to OnCallBegin and OnCallEnd simulates an RPC.
   275  		go func() {
   276  			if err := mgr.OnCallBegin(); err != nil {
   277  				t.Errorf("OnCallBegin() failed: %v", err)
   278  			}
   279  			mgr.OnCallEnd()
   280  		}()
   281  	}
   282  
   283  	// Ensure that the channel moves out of idle as a result of the above RPC.
   284  	select {
   285  	case <-enforcer.exitIdleCh:
   286  	case <-time.After(2 * defaultTestIdleTimeout):
   287  		t.Fatal("Timeout waiting for channel to move out of idle mode")
   288  	}
   289  
   290  	// Ensure that only one call to exit idle mode is made to the CC.
   291  	sCtx, sCancel := context.WithTimeout(context.Background(), defaultTestShortTimeout)
   292  	defer sCancel()
   293  	select {
   294  	case <-enforcer.exitIdleCh:
   295  		t.Fatal("More than one call to exit idle mode on the ClientConn; only one expected")
   296  	case <-sCtx.Done():
   297  	}
   298  }
   299  
   300  type racyState int32
   301  
   302  const (
   303  	stateInitial racyState = iota
   304  	stateEnteredIdle
   305  	stateExitedIdle
   306  	stateActiveRPCs
   307  )
   308  
   309  // racyIdlnessEnforcer is a test idleness enforcer used specifically to test the
   310  // race between idle timeout and incoming RPCs.
   311  type racyEnforcer struct {
   312  	t       *testing.T
   313  	state   *racyState // Accessed atomically.
   314  	started bool
   315  }
   316  
   317  // ExitIdleMode sets the internal state to stateExitedIdle. We should only ever
   318  // exit idle when we are currently in idle.
   319  func (ri *racyEnforcer) ExitIdleMode() error {
   320  	// Set only on the initial ExitIdleMode
   321  	if ri.started == false {
   322  		if !atomic.CompareAndSwapInt32((*int32)(ri.state), int32(stateInitial), int32(stateInitial)) {
   323  			return fmt.Errorf("idleness enforcer's first ExitIdleMode after EnterIdleMode")
   324  		}
   325  		ri.started = true
   326  		return nil
   327  	}
   328  	if !atomic.CompareAndSwapInt32((*int32)(ri.state), int32(stateEnteredIdle), int32(stateExitedIdle)) {
   329  		return fmt.Errorf("idleness enforcer asked to exit idle when it did not enter idle earlier")
   330  	}
   331  	return nil
   332  }
   333  
   334  // EnterIdleMode attempts to set the internal state to stateEnteredIdle. We should only ever enter idle before RPCs start.
   335  func (ri *racyEnforcer) EnterIdleMode() {
   336  	if !atomic.CompareAndSwapInt32((*int32)(ri.state), int32(stateInitial), int32(stateEnteredIdle)) {
   337  		ri.t.Errorf("idleness enforcer asked to enter idle after rpcs started")
   338  	}
   339  }
   340  
   341  // TestManager_IdleTimeoutRacesWithOnCallBegin tests the case where firing of
   342  // the idle timeout races with an incoming RPC. The test verifies that if the
   343  // timer callback wins the race and puts the channel in idle, the RPCs can kick
   344  // it out of idle. And if the RPCs win the race and keep the channel active,
   345  // then the timer callback should not attempt to put the channel in idle mode.
   346  func (s) TestManager_IdleTimeoutRacesWithOnCallBegin(t *testing.T) {
   347  	// Run multiple iterations to simulate different possibilities.
   348  	for i := 0; i < 20; i++ {
   349  		t.Run(fmt.Sprintf("iteration=%d", i), func(t *testing.T) {
   350  			var idlenessState racyState
   351  			enforcer := &racyEnforcer{t: t, state: &idlenessState}
   352  
   353  			// Configure a large idle timeout so that we can control the
   354  			// race between the timer callback and RPCs.
   355  			mgr := NewManager(enforcer, time.Duration(10*time.Minute))
   356  			defer mgr.Close()
   357  			mgr.ExitIdleMode()
   358  
   359  			var wg sync.WaitGroup
   360  			wg.Add(1)
   361  			go func() {
   362  				defer wg.Done()
   363  				<-time.After(defaultTestIdleTimeout / 50)
   364  				mgr.handleIdleTimeout()
   365  			}()
   366  			for j := 0; j < 5; j++ {
   367  				wg.Add(1)
   368  				go func() {
   369  					defer wg.Done()
   370  					// Wait for the configured idle timeout and simulate an RPC to
   371  					// race with the idle timeout timer callback.
   372  					<-time.After(defaultTestIdleTimeout / 50)
   373  					if err := mgr.OnCallBegin(); err != nil {
   374  						t.Errorf("OnCallBegin() failed: %v", err)
   375  					}
   376  					atomic.StoreInt32((*int32)(&idlenessState), int32(stateActiveRPCs))
   377  					mgr.OnCallEnd()
   378  				}()
   379  			}
   380  			wg.Wait()
   381  		})
   382  	}
   383  }
   384
View as plain text