Method UpdateClientConnState

balancer/rls/balancer.go:296–354

(ccs balancer.ClientConnState)

}

func (b *rlsBalancer) UpdateClientConnState(ccs balancer.ClientConnState) error {
	defer clientConnUpdateHook()

	b.stateMu.Lock()
	if b.closed.HasFired() {
		b.stateMu.Unlock()
		b.logger.Warningf("Received service config after balancer close: %s", pretty.ToJSON(ccs.BalancerConfig))
		return errBalancerClosed
	}

	newCfg := ccs.BalancerConfig.(*lbConfig)
	if b.lbCfg.Equal(newCfg) {
		b.stateMu.Unlock()
		b.logger.Infof("New service config matches existing config")
		return nil
	}

	b.logger.Infof("Delaying picker updates until config is propagated to and processed by child policies")
	b.inhibitPickerUpdates = true

	// When the RLS server name changes, the old control channel needs to be
	// swapped out for a new one. All state associated with the throttling
	// algorithm is stored on a per-control-channel basis; when we swap out
	// channels, we also swap out the throttling state.
	b.handleControlChannelUpdate(newCfg)

	// Any changes to child policy name or configuration needs to be handled by
	// either creating new child policies or pushing updates to existing ones.
	b.resolverState = ccs.ResolverState
	b.handleChildPolicyConfigUpdate(newCfg, &ccs)

	// Resize the cache if the size in the config has changed.
	resizeCache := newCfg.cacheSizeBytes != b.lbCfg.cacheSizeBytes

	// Update the copy of the config in the LB policy before releasing the lock.
	b.lbCfg = newCfg
	b.stateMu.Unlock()

	// We cannot do cache operations above because `cacheMu` needs to be grabbed
	// before `stateMu` if we are to hold both locks at the same time.
	b.cacheMu.Lock()
	b.dataCache.updateRLSServerTarget(newCfg.lookupService)
	if resizeCache {
		// If the new config changes reduces the size of the data cache, we
		// might have to evict entries to get the cache size down to the newly
		// specified size. If we do evict an entry with valid backoff timer,
		// the new picker needs to be sent to the channel to re-process any
		// RPCs queued as a result of this backoff timer.
		b.dataCache.resize(newCfg.cacheSizeBytes)
	}
	b.cacheMu.Unlock()
	// Enqueue an event which will notify us when the above update has been
	// propagated to all child policies, and the child policies have all
	// processed their updates, and we have sent a picker update.
	done := make(chan struct{})
	b.updateCh.Put(resumePickerUpdates{done: done})
	<-done
	return nil