grpc: support channel idleness

balancer_conn_wrappers.go

@@ -56,6 +56,13 @@ type ccBalancerWrapper struct {
 	serializerCancel context.CancelFunc
 	balancer         *gracefulswitch.Balancer
 	curBalancerName  string
+
+	// During the window of time when the channel is entring idle and the
+	// underlying balancer is being shut down, keeping track of whether the
+	// channel is in idle mode to ensure that calls from the underlying balancer
+	// are not forwarded to grpc.
+	idleMu sync.Mutex
+	isIdle bool
 }
 
 // newCCBalancerWrapper creates a new balancer wrapper. The underlying balancer
@@ -125,12 +132,6 @@ func (ccb *ccBalancerWrapper) updateSubConnState(sc balancer.SubConn, s connecti
 	})
 }
 
-func (ccb *ccBalancerWrapper) exitIdle() {
-	ccb.serializer.Schedule(func(_ context.Context) {
-		ccb.balancer.ExitIdle()
-	})
-}
-
 func (ccb *ccBalancerWrapper) resolverError(err error) {
 	ccb.serializer.Schedule(func(_ context.Context) {
 		ccb.balancer.ResolverError(err)
@@ -154,36 +155,93 @@ func (ccb *ccBalancerWrapper) switchTo(name string) {
 		if strings.EqualFold(ccb.curBalancerName, name) {
 			return
 		}
+		ccb.buildLoadBalancingPolicy(name)
+	})
+}
 
-		// Use the default LB policy, pick_first, if no LB policy with name is
-		// found in the registry.
-		builder := balancer.Get(name)
-		if builder == nil {
-			channelz.Warningf(logger, ccb.cc.channelzID, "Channel switches to new LB policy %q, since the specified LB policy %q was not registered", PickFirstBalancerName, name)
-			builder = newPickfirstBuilder()
-		} else {
-			channelz.Infof(logger, ccb.cc.channelzID, "Channel switches to new LB policy %q", name)
-		}
+// buildLoadBalancingPolicy performs the following:
+//   - retrieve a balancer builder for the given name. Use the default LB
+//     policy, pick_first, if no LB policy with name is found in the registry.
+//   - instruct the gracefulswitch balancer to switch to the above builder. This
+//     will actually build the new balancer.
+//   - update the `curBalancerName` field
+func (ccb *ccBalancerWrapper) buildLoadBalancingPolicy(name string) {
+	builder := balancer.Get(name)
+	if builder == nil {
+		channelz.Warningf(logger, ccb.cc.channelzID, "Channel switches to new LB policy %q, since the specified LB policy %q was not registered", PickFirstBalancerName, name)
+		builder = newPickfirstBuilder()
+	} else {
+		channelz.Infof(logger, ccb.cc.channelzID, "Channel switches to new LB policy %q", name)
+	}
 
-		if err := ccb.balancer.SwitchTo(builder); err != nil {
-			channelz.Errorf(logger, ccb.cc.channelzID, "Channel failed to build new LB policy %q: %v", name, err)
-			return
-		}
-		ccb.curBalancerName = builder.Name()
-	})
+	if err := ccb.balancer.SwitchTo(builder); err != nil {
+		channelz.Errorf(logger, ccb.cc.channelzID, "Channel failed to build new LB policy %q: %v", name, err)
+		return
+	}
+	ccb.curBalancerName = builder.Name()
 }
 
 func (ccb *ccBalancerWrapper) close() {
 	// Close the serializer to ensure that no more calls from gRPC are sent to
-	// the balancer. We don't have to worry about suppressing calls from a
-	// closed balancer because these are handled by the ClientConn (balancer
-	// wrapper is only ever closed when the ClientConn is closed).
+	// the balancer, and no more calls from the balancer are sent to gRPC.
 	ccb.serializerCancel()
 	<-ccb.serializer.Done
 	ccb.balancer.Close()
 }
 
+// exitIdleMode is invoked by grpc when the channel exits idle mode either
+// because of an RPC or because of an invocation of the Connect() API. This
+// recreates the balancer that was closed previously when entering idle mode.
+//
+// If the channel is not in idle mode, we know for a fact that we are here as a
+// result of the user calling the Connect() method on the ClientConn. Forward
+// the call to the underlying balancer, instructing it to reconnect to the
+// backends.
+func (ccb *ccBalancerWrapper) exitIdleMode() {
+	channelz.Info(logger, ccb.cc.channelzID, "ccBalancerWrapper: exiting idle mode")
+
+	done := make(chan struct{})
+	ccb.serializer.Schedule(func(_ context.Context) {
+		defer close(done)
+
+		ccb.idleMu.Lock()
+		defer ccb.idleMu.Unlock()
+
+		if !ccb.isIdle {
+			ccb.balancer.ExitIdle()
+			return
+		}
+
+		ccb.buildLoadBalancingPolicy(ccb.curBalancerName)
+		ccb.isIdle = false
+	})
+	<-done
+}
+
+// enterIdleMode is invoked by grpc when the channel enters idle mode upon
+// expiry of idle_timeout. This call blocks until the balancer is closed.
+func (ccb *ccBalancerWrapper) enterIdleMode() {
+	channelz.Info(logger, ccb.cc.channelzID, "ccBalancerWrapper: entering idle mode")
+
+	done := make(chan struct{})
+	ccb.serializer.Schedule(func(_ context.Context) {
+		ccb.idleMu.Lock()
+		defer ccb.idleMu.Unlock()
+
+		ccb.close()
+		ccb.isIdle = true
+		close(done)
+	})
+	<-done
+}
+
 func (ccb *ccBalancerWrapper) NewSubConn(addrs []resolver.Address, opts balancer.NewSubConnOptions) (balancer.SubConn, error) {
+	ccb.idleMu.Lock()
+	defer ccb.idleMu.Unlock()
+	if ccb.isIdle {
+		return nil, fmt.Errorf("grpc: cannote create a SubConn in idle mode")
+	}
+
 	if len(addrs) <= 0 {
 		return nil, fmt.Errorf("grpc: cannot create SubConn with empty address list")
 	}
@@ -200,6 +258,12 @@ func (ccb *ccBalancerWrapper) NewSubConn(addrs []resolver.Address, opts balancer
 }
 
 func (ccb *ccBalancerWrapper) RemoveSubConn(sc balancer.SubConn) {
+	ccb.idleMu.Lock()
+	defer ccb.idleMu.Unlock()
+	if ccb.isIdle {
+		return
+	}
+
 	acbw, ok := sc.(*acBalancerWrapper)
 	if !ok {
 		return
@@ -208,6 +272,12 @@ func (ccb *ccBalancerWrapper) RemoveSubConn(sc balancer.SubConn) {
 }
 
 func (ccb *ccBalancerWrapper) UpdateAddresses(sc balancer.SubConn, addrs []resolver.Address) {
+	ccb.idleMu.Lock()
+	defer ccb.idleMu.Unlock()
+	if ccb.isIdle {
+		return
+	}
+
 	acbw, ok := sc.(*acBalancerWrapper)
 	if !ok {
 		return
@@ -216,6 +286,12 @@ func (ccb *ccBalancerWrapper) UpdateAddresses(sc balancer.SubConn, addrs []resol
 }
 
 func (ccb *ccBalancerWrapper) UpdateState(s balancer.State) {
+	ccb.idleMu.Lock()
+	defer ccb.idleMu.Unlock()
+	if ccb.isIdle {
+		return
+	}
+
 	// Update picker before updating state.  Even though the ordering here does
 	// not matter, it can lead to multiple calls of Pick in the common start-up
 	// case where we wait for ready and then perform an RPC.  If the picker is
@@ -226,6 +302,12 @@ func (ccb *ccBalancerWrapper) UpdateState(s balancer.State) {
 }
 
 func (ccb *ccBalancerWrapper) ResolveNow(o resolver.ResolveNowOptions) {
+	ccb.idleMu.Lock()
+	defer ccb.idleMu.Unlock()
+	if ccb.isIdle {
+		return
+	}
+
 	ccb.cc.resolveNow(o)
 }
 

call.go

@@ -27,6 +27,11 @@ import (
 //
 // All errors returned by Invoke are compatible with the status package.
 func (cc *ClientConn) Invoke(ctx context.Context, method string, args, reply interface{}, opts ...CallOption) error {
+	if err := cc.idlenessMgr.onCallBegin(); err != nil {
+		return err
+	}
+	defer cc.idlenessMgr.onCallEnd()
+
 	// allow interceptor to see all applicable call options, which means those
 	// configured as defaults from dial option as well as per-call options
 	opts = combine(cc.dopts.callOptions, opts)