ringhash: Delegate subchannel creation to pickfirst (#8047)

Author: arjan-bal

balancer/weightedroundrobin/weightedroundrobin.go

@@ -34,7 +34,8 @@ import (
 )
 
 // attributeKey is the type used as the key to store AddrInfo in the
-// BalancerAttributes field of resolver.Address.
+// BalancerAttributes field of resolver.Address or Attributes field of
+// resolver.Endpoint.
 type attributeKey struct{}
 
 // AddrInfo will be stored in the BalancerAttributes field of Address in order
@@ -71,6 +72,14 @@ func GetAddrInfo(addr resolver.Address) AddrInfo {
 	return ai
 }
 
+// AddrInfoFromEndpoint returns the AddrInfo stored in the Attributes field of
+// endpoint.
+func AddrInfoFromEndpoint(endpoint resolver.Endpoint) AddrInfo {
+	v := endpoint.Attributes.Value(attributeKey{})
+	ai, _ := v.(AddrInfo)
+	return ai
+}
+
 func (a AddrInfo) String() string {
 	return fmt.Sprintf("Weight: %d", a.Weight)
 }

xds/internal/balancer/ringhash/e2e/ringhash_balancer_test.go

@@ -32,7 +32,3 @@ var logger = grpclog.Component("xds")
 func prefixLogger(p *ringhashBalancer) *internalgrpclog.PrefixLogger {
 	return internalgrpclog.NewPrefixLogger(logger, fmt.Sprintf(prefix, p))
 }
-
-func subConnPrefixLogger(p *ringhashBalancer, sc *subConn) *internalgrpclog.PrefixLogger {
-	return internalgrpclog.NewPrefixLogger(logger, fmt.Sprintf(prefix, p)+fmt.Sprintf("[subConn %p] ", sc))
-}

xds/internal/balancer/ringhash/logging.go

@@ -22,140 +22,44 @@ import (
 	"fmt"
 
 	"google.golang.org/grpc/balancer"
-	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/connectivity"
 	"google.golang.org/grpc/internal/grpclog"
-	"google.golang.org/grpc/status"
 )
 
 type picker struct {
-	ring          *ring
-	logger        *grpclog.PrefixLogger
-	subConnStates map[*subConn]connectivity.State
-}
-
-func newPicker(ring *ring, logger *grpclog.PrefixLogger) *picker {
-	states := make(map[*subConn]connectivity.State)
-	for _, e := range ring.items {
-		states[e.sc] = e.sc.effectiveState()
-	}
-	return &picker{ring: ring, logger: logger, subConnStates: states}
-}
-
-// handleRICSResult is the return type of handleRICS. It's needed to wrap the
-// returned error from Pick() in a struct. With this, if the return values are
-// `balancer.PickResult, error, bool`, linter complains because error is not the
-// last return value.
-type handleRICSResult struct {
-	pr  balancer.PickResult
-	err error
-}
-
-// handleRICS generates pick result if the entry is in Ready, Idle, Connecting
-// or Shutdown. TransientFailure will be handled specifically after this
-// function returns.
-//
-// The first return value indicates if the state is in Ready, Idle, Connecting
-// or Shutdown. If it's true, the PickResult and error should be returned from
-// Pick() as is.
-func (p *picker) handleRICS(e *ringEntry) (handleRICSResult, bool) {
-	switch state := p.subConnStates[e.sc]; state {
-	case connectivity.Ready:
-		return handleRICSResult{pr: balancer.PickResult{SubConn: e.sc.sc}}, true
-	case connectivity.Idle:
-		// Trigger Connect() and queue the pick.
-		e.sc.queueConnect()
-		return handleRICSResult{err: balancer.ErrNoSubConnAvailable}, true
-	case connectivity.Connecting:
-		return handleRICSResult{err: balancer.ErrNoSubConnAvailable}, true
-	case connectivity.TransientFailure:
-		// Return ok==false, so TransientFailure will be handled afterwards.
-		return handleRICSResult{}, false
-	case connectivity.Shutdown:
-		// Shutdown can happen in a race where the old picker is called. A new
-		// picker should already be sent.
-		return handleRICSResult{err: balancer.ErrNoSubConnAvailable}, true
-	default:
-		// Should never reach this. All the connectivity states are already
-		// handled in the cases.
-		p.logger.Errorf("SubConn has undefined connectivity state: %v", state)
-		return handleRICSResult{err: status.Errorf(codes.Unavailable, "SubConn has undefined connectivity state: %v", state)}, true
-	}
+	ring   *ring
+	logger *grpclog.PrefixLogger
+	// endpointStates is a cache of endpoint connectivity states and pickers.
+	// The ringhash balancer stores endpoint states in a `resolver.EndpointMap`,
+	// with access guarded by `ringhashBalancer.mu`. The `endpointStates` cache
+	// in the picker helps avoid locking the ringhash balancer's mutex when
+	// reading the latest state at RPC time.
+	endpointStates map[string]balancer.State // endpointState.firstAddr -> balancer.State
 }
 
 func (p *picker) Pick(info balancer.PickInfo) (balancer.PickResult, error) {
 	e := p.ring.pick(getRequestHash(info.Ctx))
-	if hr, ok := p.handleRICS(e); ok {
-		return hr.pr, hr.err
-	}
-	// ok was false, the entry is in transient failure.
-	return p.handleTransientFailure(e)
-}
-
-func (p *picker) handleTransientFailure(e *ringEntry) (balancer.PickResult, error) {
-	// Queue a connect on the first picked SubConn.
-	e.sc.queueConnect()
-
-	// Find next entry in the ring, skipping duplicate SubConns.
-	e2 := nextSkippingDuplicates(p.ring, e)
-	if e2 == nil {
-		// There's no next entry available, fail the pick.
-		return balancer.PickResult{}, fmt.Errorf("the only SubConn is in Transient Failure")
-	}
-
-	// For the second SubConn, also check Ready/Idle/Connecting as if it's the
-	// first entry.
-	if hr, ok := p.handleRICS(e2); ok {
-		return hr.pr, hr.err
-	}
-
-	// The second SubConn is also in TransientFailure. Queue a connect on it.
-	e2.sc.queueConnect()
-
-	// If it gets here, this is after the second SubConn, and the second SubConn
-	// was in TransientFailure.
-	//
-	// Loop over all other SubConns:
-	// - If all SubConns so far are all TransientFailure, trigger Connect() on
-	// the TransientFailure SubConns, and keep going.
-	// - If there's one SubConn that's not in TransientFailure, keep checking
-	// the remaining SubConns (in case there's a Ready, which will be returned),
-	// but don't not trigger Connect() on the other SubConns.
-	var firstNonFailedFound bool
-	for ee := nextSkippingDuplicates(p.ring, e2); ee != e; ee = nextSkippingDuplicates(p.ring, ee) {
-		scState := p.subConnStates[ee.sc]
-		if scState == connectivity.Ready {
-			return balancer.PickResult{SubConn: ee.sc.sc}, nil
-		}
-		if firstNonFailedFound {
-			continue
-		}
-		if scState == connectivity.TransientFailure {
-			// This will queue a connect.
-			ee.sc.queueConnect()
-			continue
-		}
-		// This is a SubConn in a non-failure state. We continue to check the
-		// other SubConns, but remember that there was a non-failed SubConn
-		// seen. After this, Pick() will never trigger any SubConn to Connect().
-		firstNonFailedFound = true
-		if scState == connectivity.Idle {
-			// This is the first non-failed SubConn, and it is in a real Idle
-			// state. Trigger it to Connect().
-			ee.sc.queueConnect()
+	ringSize := len(p.ring.items)
+	// Per gRFC A61, because of sticky-TF with PickFirst's auto reconnect on TF,
+	// we ignore all TF subchannels and find the first ring entry in READY,
+	// CONNECTING or IDLE.  If that entry is in IDLE, we need to initiate a
+	// connection. The idlePicker returned by the LazyLB or the new Pickfirst
+	// should do this automatically.
+	for i := 0; i < ringSize; i++ {
+		index := (e.idx + i) % ringSize
+		balState := p.balancerState(p.ring.items[index])
+		switch balState.ConnectivityState {
+		case connectivity.Ready, connectivity.Connecting, connectivity.Idle:
+			return balState.Picker.Pick(info)
+		case connectivity.TransientFailure:
+		default:
+			panic(fmt.Sprintf("Found child balancer in unknown state: %v", balState.ConnectivityState))
 		}
 	}
-	return balancer.PickResult{}, fmt.Errorf("no connection is Ready")
+	// All children are in transient failure. Return the first failure.
+	return p.balancerState(e).Picker.Pick(info)
 }
 
-// nextSkippingDuplicates finds the next entry in the ring, with a different
-// subconn from the given entry.
-func nextSkippingDuplicates(ring *ring, entry *ringEntry) *ringEntry {
-	for next := ring.next(entry); next != entry; next = ring.next(next) {
-		if next.sc != entry.sc {
-			return next
-		}
-	}
-	// There's no qualifying next entry.
-	return nil
+func (p *picker) balancerState(e *ringEntry) balancer.State {
+	return p.endpointStates[e.firstAddr]
 }