Repository: grpc/grpc-go
File: xds/internal/balancer/clusterimpl/clusterimpl.go (Go)

Code before:

	bOpts     balancer.BuildOptions
	logger    *grpclog.PrefixLogger
	xdsClient xdsclient.XDSClient

	config           *LBConfig
	child            *gracefulswitch.Balancer
	cancelLoadReport func()
	edsServiceName   string
	lrsServer        *bootstrap.ServerConfig
	loadWrapper      *loadstore.Wrapper

	clusterNameMu sync.Mutex
	clusterName   string

	childStateMu sync.Mutex // guards childState and inhibitPickerUpdates
	// Set during UpdateClientConnState when pushing updates to child policy.
	// Prevents state updates from child policy causing new pickers to be sent
	// up the channel. Cleared after child policy have processed the
	// updates sent to them, after which a new picker is sent up the channel.
	inhibitPickerUpdates bool
	// childState contains the most recently pushed state update from the
	// child LB policy
	childState            balancer.State
	dropCategories        []DropConfig // The categories for drops.
	configMu              sync.Mutex   // guards drops and request count config.
	drops                 []*dropper
	requestCounterCluster string // The cluster name for the request counter.
	requestCounterService string // The service name for the request counter.
	requestCounter        *xdsclient.ClusterRequestsCounter
	requestCountMax       uint32
	telemetryLabels       map[string]string
}

// updateLoadStore checks the config for load store, and decides whether it
// needs to restart the load reporting stream.
func (b *clusterImplBalancer) updateLoadStore(newConfig *LBConfig) error {
	var updateLoadClusterAndService bool

	// ClusterName is different, restart. ClusterName is from ClusterName and
	// EDSServiceName.
	clusterName := b.getClusterName()
	if clusterName != newConfig.Cluster {
		updateLoadClusterAndService = true
		b.setClusterName(newConfig.Cluster)
		clusterName = newConfig.Cluster
	}
	if b.edsServiceName != newConfig.EDSServiceName {
		updateLoadClusterAndService = true
		b.edsServiceName = newConfig.EDSServiceName
	}

Review comment (bug):
Having multiple mutexes in a single type introduces all sorts of complexity and can very easily lead us down the path to a deadlock.

Do we really need multiple mutexes? Do we even need two mutexes? Can we do with a single mutex?

Code after:
	// The following fields are set at creation time, and are read-only after
	// that, and therefore need not be protected by a mutex.
	logger      *grpclog.PrefixLogger
	loadWrapper *loadstore.Wrapper

	// The following fields are only accessed from balancer API methods, which
	// are guaranteed to be called serially by gRPC.
	xdsClient        xdsclient.XDSClient     // Sent down in ResolverState attributes.
	cancelLoadReport func()                  // To stop reporting load through the above xDS client.
	edsServiceName   string                  // EDS service name to report load for.
	lrsServer        *bootstrap.ServerConfig // Load reporting server configuration.
	dropCategories   []DropConfig            // The categories for drops.
	child            *gracefulswitch.Balancer

	// The following fields are protected by mu, since they are accessed in
	// balancer API methods and in methods called from the child policy.
	mu                    sync.Mutex
	clusterName           string                            // The cluster name for credentials handshaking.
	inhibitPickerUpdates  bool                              // Inhibits state updates from child policy when processing an update from the parent.
	childState            balancer.State                    // Most recent state update from the child policy.
	drops                 []*dropper                        // Drops implementation.
	requestCounterCluster string                            // The cluster name for the request counter, from LB config.
	requestCounterService string                            // The service name for the request counter, from LB config.
	requestCountMax       uint32                            // Max concurrent requests, from LB config.
	requestCounter        *xdsclient.ClusterRequestsCounter // Tracks total inflight requests for a given service.
	telemetryLabels       map[string]string                 // Telemetry labels to set on picks, from LB config.
}

// handleDropAndRequestCountLocked compares drop and request counter in newConfig with
// the one currently used by picker, and is protected by b.mu. It returns a boolean
// indicating if a new picker needs to be generated.
func (b *clusterImplBalancer) handleDropAndRequestCountLocked(newConfig *LBConfig) bool {
	var updatePicker bool
	if !equalDropCategories(b.dropCategories, newConfig.DropCategories) {
		b.dropCategories = newConfig.DropCategories
		b.drops = make([]*dropper, 0, len(newConfig.DropCategories))
		for _, c := range newConfig.DropCategories {
			b.drops = append(b.drops, newDropper(c))
		}
		updatePicker = true
	}

	if b.requestCounterCluster != newConfig.Cluster || b.requestCounterService != newConfig.EDSServiceName {
		b.requestCounterCluster = newConfig.Cluster
		b.requestCounterService = newConfig.EDSServiceName
		b.requestCounter = xdsclient.GetClusterRequestsCounter(newConfig.Cluster, newConfig.EDSServiceName)
		updatePicker = true
	}
	var newRequestCountMax uint32 = 1024
	if newConfig.MaxConcurrentRequests != nil {
		newRequestCountMax = *newConfig.MaxConcurrentRequests