[VOL-5486] Fix deprecated versions
Change-Id: If0b888d6c2f33b2f415c8b03b08dc994bb3df3f4
Signed-off-by: Abhay Kumar <abhay.kumar@radisys.com>
diff --git a/vendor/google.golang.org/grpc/picker_wrapper.go b/vendor/google.golang.org/grpc/picker_wrapper.go
index 45baa2a..aa52bfe 100644
--- a/vendor/google.golang.org/grpc/picker_wrapper.go
+++ b/vendor/google.golang.org/grpc/picker_wrapper.go
@@ -20,68 +20,64 @@
import (
"context"
+ "fmt"
"io"
- "sync"
+ "sync/atomic"
"google.golang.org/grpc/balancer"
"google.golang.org/grpc/codes"
- "google.golang.org/grpc/grpclog"
"google.golang.org/grpc/internal/channelz"
+ istatus "google.golang.org/grpc/internal/status"
"google.golang.org/grpc/internal/transport"
"google.golang.org/grpc/status"
)
+// pickerGeneration stores a picker and a channel used to signal that a picker
+// newer than this one is available.
+type pickerGeneration struct {
+ // picker is the picker produced by the LB policy. May be nil if a picker
+ // has never been produced.
+ picker balancer.Picker
+ // blockingCh is closed when the picker has been invalidated because there
+ // is a new one available.
+ blockingCh chan struct{}
+}
+
// pickerWrapper is a wrapper of balancer.Picker. It blocks on certain pick
// actions and unblock when there's a picker update.
type pickerWrapper struct {
- mu sync.Mutex
- done bool
- blockingCh chan struct{}
- picker balancer.Picker
-
- // The latest connection happened.
- connErrMu sync.Mutex
- connErr error
+ // If pickerGen holds a nil pointer, the pickerWrapper is closed.
+ pickerGen atomic.Pointer[pickerGeneration]
}
func newPickerWrapper() *pickerWrapper {
- bp := &pickerWrapper{blockingCh: make(chan struct{})}
- return bp
+ pw := &pickerWrapper{}
+ pw.pickerGen.Store(&pickerGeneration{
+ blockingCh: make(chan struct{}),
+ })
+ return pw
}
-func (bp *pickerWrapper) updateConnectionError(err error) {
- bp.connErrMu.Lock()
- bp.connErr = err
- bp.connErrMu.Unlock()
+// updatePicker is called by UpdateState calls from the LB policy. It
+// unblocks all blocked pick.
+func (pw *pickerWrapper) updatePicker(p balancer.Picker) {
+ old := pw.pickerGen.Swap(&pickerGeneration{
+ picker: p,
+ blockingCh: make(chan struct{}),
+ })
+ close(old.blockingCh)
}
-func (bp *pickerWrapper) connectionError() error {
- bp.connErrMu.Lock()
- err := bp.connErr
- bp.connErrMu.Unlock()
- return err
-}
-
-// updatePicker is called by UpdateBalancerState. It unblocks all blocked pick.
-func (bp *pickerWrapper) updatePicker(p balancer.Picker) {
- bp.mu.Lock()
- if bp.done {
- bp.mu.Unlock()
- return
- }
- bp.picker = p
- // bp.blockingCh should never be nil.
- close(bp.blockingCh)
- bp.blockingCh = make(chan struct{})
- bp.mu.Unlock()
-}
-
-func doneChannelzWrapper(acw *acBalancerWrapper, done func(balancer.DoneInfo)) func(balancer.DoneInfo) {
- acw.mu.Lock()
- ac := acw.ac
- acw.mu.Unlock()
+// doneChannelzWrapper performs the following:
+// - increments the calls started channelz counter
+// - wraps the done function in the passed in result to increment the calls
+// failed or calls succeeded channelz counter before invoking the actual
+// done function.
+func doneChannelzWrapper(acbw *acBalancerWrapper, result *balancer.PickResult) {
+ ac := acbw.ac
ac.incrCallsStarted()
- return func(b balancer.DoneInfo) {
+ done := result.Done
+ result.Done = func(b balancer.DoneInfo) {
if b.Err != nil && b.Err != io.EOF {
ac.incrCallsFailed()
} else {
@@ -93,6 +89,12 @@
}
}
+type pick struct {
+ transport transport.ClientTransport // the selected transport
+ result balancer.PickResult // the contents of the pick from the LB policy
+ blocked bool // set if a picker call queued for a new picker
+}
+
// pick returns the transport that will be used for the RPC.
// It may block in the following cases:
// - there's no picker
@@ -100,85 +102,97 @@
// - the current picker returns other errors and failfast is false.
// - the subConn returned by the current picker is not READY
// When one of these situations happens, pick blocks until the picker gets updated.
-func (bp *pickerWrapper) pick(ctx context.Context, failfast bool, opts balancer.PickOptions) (transport.ClientTransport, func(balancer.DoneInfo), error) {
+func (pw *pickerWrapper) pick(ctx context.Context, failfast bool, info balancer.PickInfo) (pick, error) {
var ch chan struct{}
- for {
- bp.mu.Lock()
- if bp.done {
- bp.mu.Unlock()
- return nil, nil, ErrClientConnClosing
- }
+ var lastPickErr error
+ pickBlocked := false
- if bp.picker == nil {
- ch = bp.blockingCh
+ for {
+ pg := pw.pickerGen.Load()
+ if pg == nil {
+ return pick{}, ErrClientConnClosing
}
- if ch == bp.blockingCh {
+ if pg.picker == nil {
+ ch = pg.blockingCh
+ }
+ if ch == pg.blockingCh {
// This could happen when either:
- // - bp.picker is nil (the previous if condition), or
- // - has called pick on the current picker.
- bp.mu.Unlock()
+ // - pw.picker is nil (the previous if condition), or
+ // - we have already called pick on the current picker.
select {
case <-ctx.Done():
- if connectionErr := bp.connectionError(); connectionErr != nil {
- switch ctx.Err() {
- case context.DeadlineExceeded:
- return nil, nil, status.Errorf(codes.DeadlineExceeded, "latest connection error: %v", connectionErr)
- case context.Canceled:
- return nil, nil, status.Errorf(codes.Canceled, "latest connection error: %v", connectionErr)
- }
+ var errStr string
+ if lastPickErr != nil {
+ errStr = "latest balancer error: " + lastPickErr.Error()
+ } else {
+ errStr = fmt.Sprintf("%v while waiting for connections to become ready", ctx.Err())
}
- return nil, nil, ctx.Err()
+ switch ctx.Err() {
+ case context.DeadlineExceeded:
+ return pick{}, status.Error(codes.DeadlineExceeded, errStr)
+ case context.Canceled:
+ return pick{}, status.Error(codes.Canceled, errStr)
+ }
case <-ch:
}
continue
}
- ch = bp.blockingCh
- p := bp.picker
- bp.mu.Unlock()
-
- subConn, done, err := p.Pick(ctx, opts)
-
- if err != nil {
- switch err {
- case balancer.ErrNoSubConnAvailable:
- continue
- case balancer.ErrTransientFailure:
- if !failfast {
- continue
- }
- return nil, nil, status.Errorf(codes.Unavailable, "%v, latest connection error: %v", err, bp.connectionError())
- case context.DeadlineExceeded:
- return nil, nil, status.Error(codes.DeadlineExceeded, err.Error())
- case context.Canceled:
- return nil, nil, status.Error(codes.Canceled, err.Error())
- default:
- if _, ok := status.FromError(err); ok {
- return nil, nil, err
- }
- // err is some other error.
- return nil, nil, status.Error(codes.Unknown, err.Error())
- }
+ // If the channel is set, it means that the pick call had to wait for a
+ // new picker at some point. Either it's the first iteration and this
+ // function received the first picker, or a picker errored with
+ // ErrNoSubConnAvailable or errored with failfast set to false, which
+ // will trigger a continue to the next iteration. In the first case this
+ // conditional will hit if this call had to block (the channel is set).
+ // In the second case, the only way it will get to this conditional is
+ // if there is a new picker.
+ if ch != nil {
+ pickBlocked = true
}
- acw, ok := subConn.(*acBalancerWrapper)
+ ch = pg.blockingCh
+ p := pg.picker
+
+ pickResult, err := p.Pick(info)
+ if err != nil {
+ if err == balancer.ErrNoSubConnAvailable {
+ continue
+ }
+ if st, ok := status.FromError(err); ok {
+ // Status error: end the RPC unconditionally with this status.
+ // First restrict the code to the list allowed by gRFC A54.
+ if istatus.IsRestrictedControlPlaneCode(st) {
+ err = status.Errorf(codes.Internal, "received picker error with illegal status: %v", err)
+ }
+ return pick{}, dropError{error: err}
+ }
+ // For all other errors, wait for ready RPCs should block and other
+ // RPCs should fail with unavailable.
+ if !failfast {
+ lastPickErr = err
+ continue
+ }
+ return pick{}, status.Error(codes.Unavailable, err.Error())
+ }
+
+ acbw, ok := pickResult.SubConn.(*acBalancerWrapper)
if !ok {
- grpclog.Error("subconn returned from pick is not *acBalancerWrapper")
+ logger.Errorf("subconn returned from pick is type %T, not *acBalancerWrapper", pickResult.SubConn)
continue
}
- if t, ok := acw.getAddrConn().getReadyTransport(); ok {
+ if t := acbw.ac.getReadyTransport(); t != nil {
if channelz.IsOn() {
- return t, doneChannelzWrapper(acw, done), nil
+ doneChannelzWrapper(acbw, &pickResult)
}
- return t, done, nil
+ return pick{transport: t, result: pickResult, blocked: pickBlocked}, nil
}
- if done != nil {
+ if pickResult.Done != nil {
// Calling done with nil error, no bytes sent and no bytes received.
// DoneInfo with default value works.
- done(balancer.DoneInfo{})
+ pickResult.Done(balancer.DoneInfo{})
}
- grpclog.Infof("blockingPicker: the picked transport is not ready, loop back to repick")
+ logger.Infof("blockingPicker: the picked transport is not ready, loop back to repick")
// If ok == false, ac.state is not READY.
// A valid picker always returns READY subConn. This means the state of ac
// just changed, and picker will be updated shortly.
@@ -186,12 +200,20 @@
}
}
-func (bp *pickerWrapper) close() {
- bp.mu.Lock()
- defer bp.mu.Unlock()
- if bp.done {
- return
- }
- bp.done = true
- close(bp.blockingCh)
+func (pw *pickerWrapper) close() {
+ old := pw.pickerGen.Swap(nil)
+ close(old.blockingCh)
+}
+
+// reset clears the pickerWrapper and prepares it for being used again when idle
+// mode is exited.
+func (pw *pickerWrapper) reset() {
+ old := pw.pickerGen.Swap(&pickerGeneration{blockingCh: make(chan struct{})})
+ close(old.blockingCh)
+}
+
+// dropError is a wrapper error that indicates the LB policy wishes to drop the
+// RPC and not retry it.
+type dropError struct {
+ error
}