internal: fix deadlock during switch_balancer and NewSubConn() (#4536)

clientconn.go

@@ -711,7 +711,12 @@ func (cc *ClientConn) switchBalancer(name string) {
 		return
 	}
 	if cc.balancerWrapper != nil {
+		// Don't hold cc.mu while closing the balancers. The balancers may call
+		// methods that require cc.mu (e.g. cc.NewSubConn()). Holding the mutex
+		// would cause a deadlock in that case.
+		cc.mu.Unlock()
 		cc.balancerWrapper.close()
+		cc.mu.Lock()
 	}
 
 	builder := balancer.Get(name)

resolver_conn_wrapper.go

@@ -39,6 +39,8 @@ type ccResolverWrapper struct {
 	resolver   resolver.Resolver
 	done       *grpcsync.Event
 	curState   resolver.State
+
+	incomingMu sync.Mutex // Synchronizes all the incoming calls.
 }
 
 // newCCResolverWrapper uses the resolver.Builder to build a Resolver and
@@ -90,6 +92,8 @@ func (ccr *ccResolverWrapper) close() {
 }
 
 func (ccr *ccResolverWrapper) UpdateState(s resolver.State) error {
+	ccr.incomingMu.Lock()
+	defer ccr.incomingMu.Unlock()
 	if ccr.done.HasFired() {
 		return nil
 	}
@@ -105,6 +109,8 @@ func (ccr *ccResolverWrapper) UpdateState(s resolver.State) error {
 }
 
 func (ccr *ccResolverWrapper) ReportError(err error) {
+	ccr.incomingMu.Lock()
+	defer ccr.incomingMu.Unlock()
 	if ccr.done.HasFired() {
 		return
 	}
@@ -114,6 +120,8 @@ func (ccr *ccResolverWrapper) ReportError(err error) {
 
 // NewAddress is called by the resolver implementation to send addresses to gRPC.
 func (ccr *ccResolverWrapper) NewAddress(addrs []resolver.Address) {
+	ccr.incomingMu.Lock()
+	defer ccr.incomingMu.Unlock()
 	if ccr.done.HasFired() {
 		return
 	}
@@ -128,6 +136,8 @@ func (ccr *ccResolverWrapper) NewAddress(addrs []resolver.Address) {
 // NewServiceConfig is called by the resolver implementation to send service
 // configs to gRPC.
 func (ccr *ccResolverWrapper) NewServiceConfig(sc string) {
+	ccr.incomingMu.Lock()
+	defer ccr.incomingMu.Unlock()
 	if ccr.done.HasFired() {
 		return
 	}