Skip to content

Commit

Permalink
balancer: add V2Picker, ClientConn.UpdateState, SubConnState.Connecti…
Browse files Browse the repository at this point in the history 
…onError (grpc#3186)

Also implement V2 versions of base.*, xds, pickfirst, grpclb, and round robin balancers.
  • Loading branch information
@dfawley
dfawley authored Nov 21, 2019
1 parent 7c1d326 commit dc49de8
Show file tree
Hide file tree
Showing 34 changed files with 701 additions and 560 deletions.
110 changes: 98 additions & 12 deletions balancer/balancer.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -117,6 +117,15 @@ type NewSubConnOptions struct {
HealthCheckEnabled bool
}

// State contains the balancer's state relevant to the gRPC ClientConn.
type State struct {
// State contains the connectivity state of the balancer, which is used to
// determine the state of the ClientConn.
ConnectivityState connectivity.State
// Picker is used to choose connections (SubConns) for RPCs.
Picker V2Picker
}

// ClientConn represents a gRPC ClientConn.
//
// This interface is to be implemented by gRPC. Users should not need a
Expand All @@ -137,8 +146,17 @@ type ClientConn interface {
//
// gRPC will update the connectivity state of the ClientConn, and will call pick
// on the new picker to pick new SubConn.
//
// Deprecated: use UpdateState instead
UpdateBalancerState(s connectivity.State, p Picker)

// UpdateState notifies gRPC that the balancer's internal state has
// changed.
//
// gRPC will update the connectivity state of the ClientConn, and will call pick
// on the new picker to pick new SubConns.
UpdateState(State)

// ResolveNow is called by balancer to notify gRPC to do a name resolving.
ResolveNow(resolver.ResolveNowOptions)

Expand Down Expand Up @@ -185,11 +203,19 @@ type ConfigParser interface {
ParseConfig(LoadBalancingConfigJSON json.RawMessage) (serviceconfig.LoadBalancingConfig, error)
}

// PickOptions contains addition information for the Pick operation.
type PickOptions struct {
// PickOptions is a type alias of PickInfo for legacy reasons.
//
// Deprecated: use PickInfo instead.
type PickOptions = PickInfo

// PickInfo contains additional information for the Pick operation.
type PickInfo struct {
// FullMethodName is the method name that NewClientStream() is called
// with. The canonical format is /service/Method.
FullMethodName string
// Ctx is the RPC's context, and may contain relevant RPC-level information
// like the outgoing header metadata.
Ctx context.Context
}

// DoneInfo contains additional information for done.
Expand All @@ -215,14 +241,16 @@ var (
ErrNoSubConnAvailable = errors.New("no SubConn is available")
// ErrTransientFailure indicates all SubConns are in TransientFailure.
// WaitForReady RPCs will block, non-WaitForReady RPCs will fail.
ErrTransientFailure = errors.New("all SubConns are in TransientFailure")
ErrTransientFailure = TransientFailureError(errors.New("all SubConns are in TransientFailure"))
)

// Picker is used by gRPC to pick a SubConn to send an RPC.
// Balancer is expected to generate a new picker from its snapshot every time its
// internal state has changed.
//
// The pickers used by gRPC can be updated by ClientConn.UpdateBalancerState().
//
// Deprecated: use V2Picker instead
type Picker interface {
// Pick returns the SubConn to be used to send the RPC.
// The returned SubConn must be one returned by NewSubConn().
Expand All @@ -243,18 +271,76 @@ type Picker interface {
//
// If the returned error is not nil:
// - If the error is ErrNoSubConnAvailable, gRPC will block until UpdateBalancerState()
// - If the error is ErrTransientFailure:
// - If the error is ErrTransientFailure or implements IsTransientFailure()
// bool, returning true:
// - If the RPC is wait-for-ready, gRPC will block until UpdateBalancerState()
// is called to pick again;
// - Otherwise, RPC will fail with unavailable error.
// - Else (error is other non-nil error):
// - The RPC will fail with unavailable error.
// - The RPC will fail with the error's status code, or Unknown if it is
// not a status error.
//
// The returned done() function will be called once the rpc has finished,
// with the final status of that RPC. If the SubConn returned is not a
// valid SubConn type, done may not be called. done may be nil if balancer
// doesn't care about the RPC status.
Pick(ctx context.Context, opts PickOptions) (conn SubConn, done func(DoneInfo), err error)
Pick(ctx context.Context, info PickInfo) (conn SubConn, done func(DoneInfo), err error)
}

// PickResult contains information related to a connection chosen for an RPC.
type PickResult struct {
// SubConn is the connection to use for this pick, if its state is Ready.
// If the state is not Ready, gRPC will block the RPC until a new Picker is
// provided by the balancer (using ClientConn.UpdateState). The SubConn
// must be one returned by ClientConn.NewSubConn.
SubConn SubConn

// Done is called when the RPC is completed. If the SubConn is not ready,
// this will be called with a nil parameter. If the SubConn is not a valid
// type, Done may not be called. May be nil if the balancer does not wish
// to be notified when the RPC completes.
Done func(DoneInfo)
}

type transientFailureError struct {
error
}

func (e *transientFailureError) IsTransientFailure() bool { return true }

// TransientFailureError wraps err in an error implementing
// IsTransientFailure() bool, returning true.
func TransientFailureError(err error) error {
return &transientFailureError{error: err}
}

// V2Picker is used by gRPC to pick a SubConn to send an RPC.
// Balancer is expected to generate a new picker from its snapshot every time its
// internal state has changed.
//
// The pickers used by gRPC can be updated by ClientConn.UpdateBalancerState().
type V2Picker interface {
// Pick returns the connection to use for this RPC and related information.
//
// Pick should not block. If the balancer needs to do I/O or any blocking
// or time-consuming work to service this call, it should return
// ErrNoSubConnAvailable, and the Pick call will be repeated by gRPC when
// the Picker is updated (using ClientConn.UpdateState).
//
// If an error is returned:
//
// - If the error is ErrNoSubConnAvailable, gRPC will block until a new
// Picker is provided by the balancer (using ClientConn.UpdateState).
//
// - If the error implements IsTransientFailure() bool, returning true,
// wait for ready RPCs will wait, but non-wait for ready RPCs will be
// terminated with this error's Error() string and status code
// Unavailable.
//
// - Any other errors terminate all RPCs with the code and message
// provided. If the error is not a status error, it will be converted by
// gRPC to a status error with code Unknown.
Pick(info PickInfo) (PickResult, error)
}

// Balancer takes input from gRPC, manages SubConns, and collects and aggregates
Expand Down Expand Up @@ -292,8 +378,11 @@ type Balancer interface {

// SubConnState describes the state of a SubConn.
type SubConnState struct {
// ConnectivityState is the connectivity state of the SubConn.
ConnectivityState connectivity.State
// TODO: add last connection error
// ConnectionError is set if the ConnectivityState is TransientFailure,
// describing the reason the SubConn failed. Otherwise, it is nil.
ConnectionError error
}

// ClientConnState describes the state of a ClientConn relevant to the
Expand Down Expand Up @@ -335,9 +424,8 @@ type V2Balancer interface {
//
// It's not thread safe.
type ConnectivityStateEvaluator struct {
numReady uint64 // Number of addrConns in ready state.
numConnecting uint64 // Number of addrConns in connecting state.
numTransientFailure uint64 // Number of addrConns in transientFailure.
numReady uint64 // Number of addrConns in ready state.
numConnecting uint64 // Number of addrConns in connecting state.
}

// RecordTransition records state change happening in subConn and based on that
Expand All @@ -357,8 +445,6 @@ func (cse *ConnectivityStateEvaluator) RecordTransition(oldState, newState conne
cse.numReady += updateVal
case connectivity.Connecting:
cse.numConnecting += updateVal
case connectivity.TransientFailure:
cse.numTransientFailure += updateVal
}
}

Expand Down
110 changes: 84 additions & 26 deletions balancer/base/balancer.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ package base

import (
"context"
"errors"

"google.golang.org/grpc/balancer"
"google.golang.org/grpc/connectivity"
Expand All @@ -28,25 +29,32 @@ import (
)

type baseBuilder struct {
name string
pickerBuilder PickerBuilder
config Config
name string
pickerBuilder PickerBuilder
v2PickerBuilder V2PickerBuilder
config Config
}

func (bb *baseBuilder) Build(cc balancer.ClientConn, opt balancer.BuildOptions) balancer.Balancer {
return &baseBalancer{
cc: cc,
pickerBuilder: bb.pickerBuilder,
bal := &baseBalancer{
cc: cc,
pickerBuilder: bb.pickerBuilder,
v2PickerBuilder: bb.v2PickerBuilder,

subConns: make(map[resolver.Address]balancer.SubConn),
scStates: make(map[balancer.SubConn]connectivity.State),
csEvltr: &balancer.ConnectivityStateEvaluator{},
// Initialize picker to a picker that always return
// ErrNoSubConnAvailable, because when state of a SubConn changes, we
// may call UpdateBalancerState with this picker.
picker: NewErrPicker(balancer.ErrNoSubConnAvailable),
config: bb.config,
config: bb.config,
}
// Initialize picker to a picker that always returns
// ErrNoSubConnAvailable, because when state of a SubConn changes, we
// may call UpdateState with this picker.
if bb.pickerBuilder != nil {
bal.picker = NewErrPicker(balancer.ErrNoSubConnAvailable)
} else {
bal.v2Picker = NewErrPickerV2(balancer.ErrNoSubConnAvailable)
}
return bal
}

func (bb *baseBuilder) Name() string {
Expand All @@ -56,24 +64,33 @@ func (bb *baseBuilder) Name() string {
var _ balancer.V2Balancer = (*baseBalancer)(nil) // Assert that we implement V2Balancer

type baseBalancer struct {
cc balancer.ClientConn
pickerBuilder PickerBuilder
cc balancer.ClientConn
pickerBuilder PickerBuilder
v2PickerBuilder V2PickerBuilder

csEvltr *balancer.ConnectivityStateEvaluator
state connectivity.State

subConns map[resolver.Address]balancer.SubConn
scStates map[balancer.SubConn]connectivity.State
picker balancer.Picker
v2Picker balancer.V2Picker
config Config
}

func (b *baseBalancer) HandleResolvedAddrs(addrs []resolver.Address, err error) {
panic("not implemented")
}

func (b *baseBalancer) ResolverError(error) {
// Ignore
func (b *baseBalancer) ResolverError(err error) {
switch b.state {
case connectivity.TransientFailure, connectivity.Idle, connectivity.Connecting:
if b.picker != nil {
b.picker = NewErrPicker(err)
} else {
b.v2Picker = NewErrPickerV2(err)
}
}
}

func (b *baseBalancer) UpdateClientConnState(s balancer.ClientConnState) error {
Expand Down Expand Up @@ -114,20 +131,44 @@ func (b *baseBalancer) UpdateClientConnState(s balancer.ClientConnState) error {
// from it. The picker is
// - errPicker with ErrTransientFailure if the balancer is in TransientFailure,
// - built by the pickerBuilder with all READY SubConns otherwise.
func (b *baseBalancer) regeneratePicker() {
func (b *baseBalancer) regeneratePicker(err error) {
if b.state == connectivity.TransientFailure {
b.picker = NewErrPicker(balancer.ErrTransientFailure)
if b.pickerBuilder != nil {
b.picker = NewErrPicker(balancer.ErrTransientFailure)
} else {
if err != nil {
b.v2Picker = NewErrPickerV2(balancer.TransientFailureError(err))
} else {
// This means the last subchannel transition was not to
// TransientFailure (otherwise err must be set), but the
// aggregate state of the balancer is TransientFailure, meaning
// there are no other addresses.
b.v2Picker = NewErrPickerV2(balancer.TransientFailureError(errors.New("resolver returned no addresses")))
}
}
return
}
readySCs := make(map[resolver.Address]balancer.SubConn)
if b.pickerBuilder != nil {
readySCs := make(map[resolver.Address]balancer.SubConn)

// Filter out all ready SCs from full subConn map.
for addr, sc := range b.subConns {
if st, ok := b.scStates[sc]; ok && st == connectivity.Ready {
readySCs[addr] = sc
// Filter out all ready SCs from full subConn map.
for addr, sc := range b.subConns {
if st, ok := b.scStates[sc]; ok && st == connectivity.Ready {
readySCs[addr] = sc
}
}
b.picker = b.pickerBuilder.Build(readySCs)
} else {
readySCs := make(map[balancer.SubConn]SubConnInfo)

// Filter out all ready SCs from full subConn map.
for addr, sc := range b.subConns {
if st, ok := b.scStates[sc]; ok && st == connectivity.Ready {
readySCs[sc] = SubConnInfo{Address: addr}
}
}
b.v2Picker = b.v2PickerBuilder.Build(PickerBuildInfo{ReadySCs: readySCs})
}
b.picker = b.pickerBuilder.Build(readySCs)
}

func (b *baseBalancer) HandleSubConnStateChange(sc balancer.SubConn, s connectivity.State) {
Expand Down Expand Up @@ -166,10 +207,14 @@ func (b *baseBalancer) UpdateSubConnState(sc balancer.SubConn, state balancer.Su
// - the aggregated state of balancer became non-TransientFailure from TransientFailure
if (s == connectivity.Ready) != (oldS == connectivity.Ready) ||
(b.state == connectivity.TransientFailure) != (oldAggrState == connectivity.TransientFailure) {
b.regeneratePicker()
b.regeneratePicker(state.ConnectionError)
}

b.cc.UpdateBalancerState(b.state, b.picker)
if b.picker != nil {
b.cc.UpdateBalancerState(b.state, b.picker)
} else {
b.cc.UpdateState(balancer.State{ConnectivityState: b.state, Picker: b.v2Picker})
}
}

// Close is a nop because base balancer doesn't have internal state to clean up,
Expand All @@ -186,6 +231,19 @@ type errPicker struct {
err error // Pick() always returns this err.
}

func (p *errPicker) Pick(ctx context.Context, opts balancer.PickOptions) (balancer.SubConn, func(balancer.DoneInfo), error) {
func (p *errPicker) Pick(context.Context, balancer.PickInfo) (balancer.SubConn, func(balancer.DoneInfo), error) {
return nil, nil, p.err
}

// NewErrPickerV2 returns a V2Picker that always returns err on Pick().
func NewErrPickerV2(err error) balancer.V2Picker {
return &errPickerV2{err: err}
}

type errPickerV2 struct {
err error // Pick() always returns this err.
}

func (p *errPickerV2) Pick(info balancer.PickInfo) (balancer.PickResult, error) {
return balancer.PickResult{}, p.err
}
Loading

0 comments on commit dc49de8

Please sign in to comment.