Repository: grpc/grpc-go
File: balancer/rls/control_channel.go (Go)

Code before:
	//   - adaptive throttling for the channel as a whole
	//   - exponential backoff on a per-request basis
	// we need a way to avoid double-penalizing requests by counting failures
	// toward both mechanisms when the RLS server is unreachable.
	//
	// To accomplish this, we monitor the state of the control plane channel. If
	// the state has been TRANSIENT_FAILURE since the last time it was in state
	// READY, and it then transitions into state READY, we push on a channel
	// which is being read by the LB policy.
	//
	// The LB the policy will iterate through the cache to reset the backoff
	// timeouts in all cache entries. Specifically, this means that it will
	// reset the backoff state and cancel the pending backoff timer. Note that
	// when cancelling the backoff timer, just like when the backoff timer fires
	// normally, a new picker is returned to the channel, to force it to
	// re-process any wait-for-ready RPCs that may still be queued if we failed
	// them while we were in backoff. However, we should optimize this case by
	// returning only one new picker, regardless of how many backoff timers are
	// cancelled.

	// Using the background context is fine here since we check for the ClientConn
	// entering SHUTDOWN and return early in that case.
	stateSubscriber := &ccStateSubscriber{
		state: buffer.NewUnbounded(),
	}
	unsubscribe := internal.SubscribeToConnectivityStateChanges.(func(cc *grpc.ClientConn, s grpcsync.Subscriber) func())(cc.cc, stateSubscriber)
	first := true
	defer func() {
		unsubscribe()
		stateSubscriber.state.Close()
	}()
	for {
		// Wait for the control channel to become READY.
		var s any
		for s = <-stateSubscriber.state.Get(); s != connectivity.Ready; s = <-stateSubscriber.state.Get() {
			stateSubscriber.state.Load()
			if s == connectivity.Shutdown {
				return
			}
		}
		stateSubscriber.state.Load()
		cc.logger.Infof("Connectivity state is READY")

		if !first {
			cc.logger.Infof("Control channel back to READY")
			cc.backToReadyFunc()
		}
		first = false

		// Wait for the control channel to move out of READY.
		for s = <-stateSubscriber.state.Get(); s == connectivity.Ready; s = <-stateSubscriber.state.Get() {

Review comment (suggestion):
When a new subscriber is added to the pubsub, it receives the most recent message posted on the pubsub. This means that if there were N messages posted on the pubsub when a new subscriber is added, the subscriber only receives the most recently posted message. This might not be good enough for our purposes here. So, I suggest making the following changes.

- Get rid of the `ccStateSubscriber`. Instead store the `buffer.Unbounded` as a field of `controlChannel`.
- Initialize the unbounded buffer when the control channel is created in `newControlChannel`.
- Change `grpc.Dial` to `grpc.NewClient` in `newControlChannel`.
- Register the subscriber right after creating the ClientConn to the RLS server, but before calling `Connect` on it. This will ensure that the subscriber will receive every single state change on the ClientConn.
  - Implement the `OnMessage` method on the `controlChannel` type and pass it to the call to `internal.SubscribeToConnectivityStateChanges`

Code after:
		dopts = append(dopts, grpc.WithDisableServiceConfig(), grpc.WithDefaultServiceConfig(serviceConfig))
	}

	return dopts, nil
}

func (cc *controlChannel) monitorConnectivityState() {
	cc.logger.Infof("Starting connectivity state monitoring goroutine")
	defer close(cc.monitorDoneCh)

	// Since we use two mechanisms to deal with RLS server being down:
	//   - adaptive throttling for the channel as a whole
	//   - exponential backoff on a per-request basis
	// we need a way to avoid double-penalizing requests by counting failures
	// toward both mechanisms when the RLS server is unreachable.
	//
	// To accomplish this, we monitor the state of the control plane channel. If
	// the state has been TRANSIENT_FAILURE since the last time it was in state
	// READY, and it then transitions into state READY, we push on a channel
	// which is being read by the LB policy.
	//
	// The LB the policy will iterate through the cache to reset the backoff
	// timeouts in all cache entries. Specifically, this means that it will
	// reset the backoff state and cancel the pending backoff timer. Note that
	// when cancelling the backoff timer, just like when the backoff timer fires
	// normally, a new picker is returned to the channel, to force it to
	// re-process any wait-for-ready RPCs that may still be queued if we failed
	// them while we were in backoff. However, we should optimize this case by
	// returning only one new picker, regardless of how many backoff timers are
	// cancelled.

	// Wait for the control channel to become READY for the first time.
	for s, ok := <-cc.connectivityStateCh.Get(); s != connectivity.Ready; s, ok = <-cc.connectivityStateCh.Get() {
		if !ok {
			return
		}

		cc.connectivityStateCh.Load()
		if s == connectivity.Shutdown {
			return
		}
	}
	cc.connectivityStateCh.Load()
	cc.logger.Infof("Connectivity state is READY")

	for {
		s, ok := <-cc.connectivityStateCh.Get()
		if !ok {
			return
		}
		cc.connectivityStateCh.Load()