Repository: grpc/grpc-go
File: internal/xds/resolver/xds_resolver.go (Go)

Code before:
	// Send the update to the ClientConn.
	state := iresolver.SetConfigSelector(resolver.State{
		ServiceConfig: r.cc.ParseServiceConfig(string(sc)),
	}, cs)
	if err := r.cc.UpdateState(xdsclient.SetClient(state, r.xdsClient)); err != nil {
		if r.logger.V(2) {
			r.logger.Infof("Channel rejected new state: %+v with error: %v", state, err)
		}
		return false
	}
	return true
}

// newConfigSelector creates a new config selector using the most recently
// received listener and route config updates. May add entries to
// r.activeClusters for previously-unseen clusters.
//
// Only executed in the context of a serializer callback.
func (r *xdsResolver) newConfigSelector() *configSelector {
	cs := &configSelector{
		r:         r,
		xdsNodeID: r.xdsClient.BootstrapConfig().Node().GetId(),
		virtualHost: virtualHost{
			httpFilterConfigOverride: r.currentVirtualHost.HTTPFilterConfigOverride,
			retryConfig:              r.currentVirtualHost.RetryConfig,
		},
		routes:           make([]route, len(r.currentVirtualHost.Routes)),
		clusters:         make(map[string]*clusterInfo),
		httpFilterConfig: r.currentListener.HTTPFilters,
	}

	for i, rt := range r.currentVirtualHost.Routes {
		clusters := rinternal.NewWRR.(func() wrr.WRR)()
		if rt.ClusterSpecifierPlugin != "" {
			clusterName := clusterSpecifierPluginPrefix + rt.ClusterSpecifierPlugin
			clusters.Add(&routeCluster{
				name: clusterName,
			}, 1)
			ci := r.addOrGetActiveClusterInfo(clusterName)
			ci.cfg = xdsChildConfig{ChildPolicy: balancerConfig(r.currentRouteConfig.ClusterSpecifierPlugins[rt.ClusterSpecifierPlugin])}
			cs.clusters[clusterName] = ci
		} else {
			for cluster, wc := range rt.WeightedClusters {
				clusterName := clusterPrefix + cluster
				clusters.Add(&routeCluster{
					name:                     clusterName,
					httpFilterConfigOverride: wc.HTTPFilterConfigOverride,
				}, int64(wc.Weight))
				ci := r.addOrGetActiveClusterInfo(clusterName)
				ci.cfg = xdsChildConfig{ChildPolicy: newBalancerConfig(cdsName, cdsBalancerConfig{Cluster: cluster})}
				cs.clusters[clusterName] = ci

Review comment (suggestion):
I know this is existing behaviour, but it seems a little unexpected to me that we DO NOT use the `configSelector` create here (`cs`) to call  `r.sendNewServiceConfig`. 

Is this to avoid races when `cs.stop()` is called while a call to `sendNewServiceConfig` is queued in the serializer? This present implementation would ensure the latest `configSelector` is called, even if the call is redundant.

I can think of a way to avoid the redundant call in a more explicit way: 

```go
sendNewServiceConfig = func() {
	r.serializer.TrySchedule(func(context.Context) {
		// This makes it explicit that we're trying to avoid calls from stale
		// config selectors. It also avoids redundant updates.
		if cs != r.curConfigSelector {
			return
		}
		r.sendNewServiceConfig(cs)
	})
```

Let me know what you think.

Code after:
	// Send the update to the ClientConn.
	state := iresolver.SetConfigSelector(resolver.State{
		ServiceConfig: r.cc.ParseServiceConfig(string(sc)),
	}, cs)
	if err := r.cc.UpdateState(xdsclient.SetClient(state, r.xdsClient)); err != nil {
		if r.logger.V(2) {
			r.logger.Infof("Channel rejected new state: %+v with error: %v", state, err)
		}
		return false
	}
	return true
}

// newConfigSelector creates a new config selector using the most recently
// received listener and route config updates. May add entries to
// r.activeClusters for previously-unseen clusters.
//
// Only executed in the context of a serializer callback.
func (r *xdsResolver) newConfigSelector() *configSelector {
	cs := &configSelector{
		channelID: r.channelID,
		xdsNodeID: r.xdsClient.BootstrapConfig().Node().GetId(),
		sendNewServiceConfig: func() {
			r.serializer.TrySchedule(func(context.Context) {
				r.sendNewServiceConfig(r.curConfigSelector)
			})
		},
		virtualHost: virtualHost{
			httpFilterConfigOverride: r.currentVirtualHost.HTTPFilterConfigOverride,
			retryConfig:              r.currentVirtualHost.RetryConfig,
		},
		routes:           make([]route, len(r.currentVirtualHost.Routes)),
		clusters:         make(map[string]*clusterInfo),
		httpFilterConfig: r.currentListener.HTTPFilters,
	}

	for i, rt := range r.currentVirtualHost.Routes {
		clusters := rinternal.NewWRR.(func() wrr.WRR)()
		if rt.ClusterSpecifierPlugin != "" {
			clusterName := clusterSpecifierPluginPrefix + rt.ClusterSpecifierPlugin
			clusters.Add(&routeCluster{
				name: clusterName,
			}, 1)
			ci := r.addOrGetActiveClusterInfo(clusterName)
			ci.cfg = xdsChildConfig{ChildPolicy: balancerConfig(r.currentRouteConfig.ClusterSpecifierPlugins[rt.ClusterSpecifierPlugin])}
			cs.clusters[clusterName] = ci
		} else {
			for cluster, wc := range rt.WeightedClusters {
				clusterName := clusterPrefix + cluster
				clusters.Add(&routeCluster{
					name:                     clusterName,