2019-11-07 19:05:39 +00:00
|
|
|
/*
|
|
|
|
*
|
|
|
|
* Copyright 2017 gRPC authors.
|
|
|
|
*
|
|
|
|
* Licensed under the Apache License, Version 2.0 (the "License");
|
|
|
|
* you may not use this file except in compliance with the License.
|
|
|
|
* You may obtain a copy of the License at
|
|
|
|
*
|
|
|
|
* http://www.apache.org/licenses/LICENSE-2.0
|
|
|
|
*
|
|
|
|
* Unless required by applicable law or agreed to in writing, software
|
|
|
|
* distributed under the License is distributed on an "AS IS" BASIS,
|
|
|
|
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
|
|
* See the License for the specific language governing permissions and
|
|
|
|
* limitations under the License.
|
|
|
|
*
|
|
|
|
*/
|
|
|
|
|
|
|
|
package grpc
|
|
|
|
|
|
|
|
import (
|
|
|
|
"context"
|
|
|
|
"io"
|
|
|
|
"sync"
|
|
|
|
|
|
|
|
"google.golang.org/grpc/balancer"
|
|
|
|
"google.golang.org/grpc/codes"
|
|
|
|
"google.golang.org/grpc/internal/channelz"
|
2022-11-22 09:26:16 +00:00
|
|
|
istatus "google.golang.org/grpc/internal/status"
|
2019-11-07 19:05:39 +00:00
|
|
|
"google.golang.org/grpc/internal/transport"
|
|
|
|
"google.golang.org/grpc/status"
|
|
|
|
)
|
|
|
|
|
|
|
|
// 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
|
2023-07-07 07:04:32 +00:00
|
|
|
idle bool
|
2019-11-07 19:05:39 +00:00
|
|
|
blockingCh chan struct{}
|
2020-06-25 20:40:43 +00:00
|
|
|
picker balancer.Picker
|
2019-11-07 19:05:39 +00:00
|
|
|
}
|
|
|
|
|
2019-12-26 17:41:02 +00:00
|
|
|
func newPickerWrapper() *pickerWrapper {
|
2020-06-25 20:40:43 +00:00
|
|
|
return &pickerWrapper{blockingCh: make(chan struct{})}
|
2019-12-26 17:41:02 +00:00
|
|
|
}
|
|
|
|
|
2019-11-07 19:05:39 +00:00
|
|
|
// updatePicker is called by UpdateBalancerState. It unblocks all blocked pick.
|
2019-12-26 17:41:02 +00:00
|
|
|
func (pw *pickerWrapper) updatePicker(p balancer.Picker) {
|
|
|
|
pw.mu.Lock()
|
2023-07-07 07:04:32 +00:00
|
|
|
if pw.done || pw.idle {
|
|
|
|
// There is a small window where a picker update from the LB policy can
|
|
|
|
// race with the channel going to idle mode. If the picker is idle here,
|
|
|
|
// it is because the channel asked it to do so, and therefore it is sage
|
|
|
|
// to ignore the update from the LB policy.
|
2019-12-26 17:41:02 +00:00
|
|
|
pw.mu.Unlock()
|
2019-11-07 19:05:39 +00:00
|
|
|
return
|
|
|
|
}
|
2019-12-26 17:41:02 +00:00
|
|
|
pw.picker = p
|
|
|
|
// pw.blockingCh should never be nil.
|
|
|
|
close(pw.blockingCh)
|
|
|
|
pw.blockingCh = make(chan struct{})
|
|
|
|
pw.mu.Unlock()
|
2019-11-07 19:05:39 +00:00
|
|
|
}
|
|
|
|
|
2023-02-08 16:55:14 +00:00
|
|
|
// 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.
|
2023-07-07 07:04:32 +00:00
|
|
|
func doneChannelzWrapper(acbw *acBalancerWrapper, result *balancer.PickResult) {
|
|
|
|
ac := acbw.ac
|
2019-11-07 19:05:39 +00:00
|
|
|
ac.incrCallsStarted()
|
2023-02-08 16:55:14 +00:00
|
|
|
done := result.Done
|
|
|
|
result.Done = func(b balancer.DoneInfo) {
|
2019-11-07 19:05:39 +00:00
|
|
|
if b.Err != nil && b.Err != io.EOF {
|
|
|
|
ac.incrCallsFailed()
|
|
|
|
} else {
|
|
|
|
ac.incrCallsSucceeded()
|
|
|
|
}
|
|
|
|
if done != nil {
|
|
|
|
done(b)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// pick returns the transport that will be used for the RPC.
|
|
|
|
// It may block in the following cases:
|
|
|
|
// - there's no picker
|
|
|
|
// - the current picker returns ErrNoSubConnAvailable
|
|
|
|
// - 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.
|
2023-02-08 16:55:14 +00:00
|
|
|
func (pw *pickerWrapper) pick(ctx context.Context, failfast bool, info balancer.PickInfo) (transport.ClientTransport, balancer.PickResult, error) {
|
2019-11-07 19:05:39 +00:00
|
|
|
var ch chan struct{}
|
|
|
|
|
2019-12-26 17:41:02 +00:00
|
|
|
var lastPickErr error
|
2019-11-07 19:05:39 +00:00
|
|
|
for {
|
2019-12-26 17:41:02 +00:00
|
|
|
pw.mu.Lock()
|
|
|
|
if pw.done {
|
|
|
|
pw.mu.Unlock()
|
2023-02-08 16:55:14 +00:00
|
|
|
return nil, balancer.PickResult{}, ErrClientConnClosing
|
2019-11-07 19:05:39 +00:00
|
|
|
}
|
|
|
|
|
2019-12-26 17:41:02 +00:00
|
|
|
if pw.picker == nil {
|
|
|
|
ch = pw.blockingCh
|
2019-11-07 19:05:39 +00:00
|
|
|
}
|
2019-12-26 17:41:02 +00:00
|
|
|
if ch == pw.blockingCh {
|
2019-11-07 19:05:39 +00:00
|
|
|
// This could happen when either:
|
2019-12-26 17:41:02 +00:00
|
|
|
// - pw.picker is nil (the previous if condition), or
|
2019-11-07 19:05:39 +00:00
|
|
|
// - has called pick on the current picker.
|
2019-12-26 17:41:02 +00:00
|
|
|
pw.mu.Unlock()
|
2019-11-07 19:05:39 +00:00
|
|
|
select {
|
|
|
|
case <-ctx.Done():
|
2019-12-26 17:41:02 +00:00
|
|
|
var errStr string
|
|
|
|
if lastPickErr != nil {
|
|
|
|
errStr = "latest balancer error: " + lastPickErr.Error()
|
|
|
|
} else {
|
|
|
|
errStr = ctx.Err().Error()
|
|
|
|
}
|
|
|
|
switch ctx.Err() {
|
|
|
|
case context.DeadlineExceeded:
|
2023-02-08 16:55:14 +00:00
|
|
|
return nil, balancer.PickResult{}, status.Error(codes.DeadlineExceeded, errStr)
|
2019-12-26 17:41:02 +00:00
|
|
|
case context.Canceled:
|
2023-02-08 16:55:14 +00:00
|
|
|
return nil, balancer.PickResult{}, status.Error(codes.Canceled, errStr)
|
2019-11-07 19:05:39 +00:00
|
|
|
}
|
|
|
|
case <-ch:
|
|
|
|
}
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
|
2019-12-26 17:41:02 +00:00
|
|
|
ch = pw.blockingCh
|
|
|
|
p := pw.picker
|
|
|
|
pw.mu.Unlock()
|
2019-11-07 19:05:39 +00:00
|
|
|
|
2019-12-26 17:41:02 +00:00
|
|
|
pickResult, err := p.Pick(info)
|
2019-11-07 19:05:39 +00:00
|
|
|
if err != nil {
|
2019-12-26 17:41:02 +00:00
|
|
|
if err == balancer.ErrNoSubConnAvailable {
|
2019-11-07 19:05:39 +00:00
|
|
|
continue
|
2019-12-26 17:41:02 +00:00
|
|
|
}
|
2022-11-22 09:26:16 +00:00
|
|
|
if st, ok := status.FromError(err); ok {
|
2020-06-25 20:40:43 +00:00
|
|
|
// Status error: end the RPC unconditionally with this status.
|
2022-11-22 09:26:16 +00:00
|
|
|
// 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)
|
|
|
|
}
|
2023-02-08 16:55:14 +00:00
|
|
|
return nil, balancer.PickResult{}, dropError{error: err}
|
2019-12-26 17:41:02 +00:00
|
|
|
}
|
2020-06-25 20:40:43 +00:00
|
|
|
// For all other errors, wait for ready RPCs should block and other
|
|
|
|
// RPCs should fail with unavailable.
|
|
|
|
if !failfast {
|
|
|
|
lastPickErr = err
|
|
|
|
continue
|
|
|
|
}
|
2023-02-08 16:55:14 +00:00
|
|
|
return nil, balancer.PickResult{}, status.Error(codes.Unavailable, err.Error())
|
2019-11-07 19:05:39 +00:00
|
|
|
}
|
|
|
|
|
2023-07-07 07:04:32 +00:00
|
|
|
acbw, ok := pickResult.SubConn.(*acBalancerWrapper)
|
2019-11-07 19:05:39 +00:00
|
|
|
if !ok {
|
2021-11-08 10:39:57 +00:00
|
|
|
logger.Errorf("subconn returned from pick is type %T, not *acBalancerWrapper", pickResult.SubConn)
|
2019-11-07 19:05:39 +00:00
|
|
|
continue
|
|
|
|
}
|
2023-07-07 07:04:32 +00:00
|
|
|
if t := acbw.ac.getReadyTransport(); t != nil {
|
2019-11-07 19:05:39 +00:00
|
|
|
if channelz.IsOn() {
|
2023-07-07 07:04:32 +00:00
|
|
|
doneChannelzWrapper(acbw, &pickResult)
|
2023-02-08 16:55:14 +00:00
|
|
|
return t, pickResult, nil
|
2019-11-07 19:05:39 +00:00
|
|
|
}
|
2023-02-08 16:55:14 +00:00
|
|
|
return t, pickResult, nil
|
2019-11-07 19:05:39 +00:00
|
|
|
}
|
2019-12-26 17:41:02 +00:00
|
|
|
if pickResult.Done != nil {
|
2019-11-07 19:05:39 +00:00
|
|
|
// Calling done with nil error, no bytes sent and no bytes received.
|
|
|
|
// DoneInfo with default value works.
|
2019-12-26 17:41:02 +00:00
|
|
|
pickResult.Done(balancer.DoneInfo{})
|
2019-11-07 19:05:39 +00:00
|
|
|
}
|
2020-08-05 08:10:10 +00:00
|
|
|
logger.Infof("blockingPicker: the picked transport is not ready, loop back to repick")
|
2019-11-07 19:05:39 +00:00
|
|
|
// 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.
|
|
|
|
// continue back to the beginning of the for loop to repick.
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2019-12-26 17:41:02 +00:00
|
|
|
func (pw *pickerWrapper) close() {
|
|
|
|
pw.mu.Lock()
|
|
|
|
defer pw.mu.Unlock()
|
|
|
|
if pw.done {
|
2019-11-07 19:05:39 +00:00
|
|
|
return
|
|
|
|
}
|
2019-12-26 17:41:02 +00:00
|
|
|
pw.done = true
|
|
|
|
close(pw.blockingCh)
|
2019-11-07 19:05:39 +00:00
|
|
|
}
|
2022-05-20 11:45:24 +00:00
|
|
|
|
2023-07-07 07:04:32 +00:00
|
|
|
func (pw *pickerWrapper) enterIdleMode() {
|
|
|
|
pw.mu.Lock()
|
|
|
|
defer pw.mu.Unlock()
|
|
|
|
if pw.done {
|
|
|
|
return
|
|
|
|
}
|
|
|
|
pw.idle = true
|
|
|
|
}
|
|
|
|
|
|
|
|
func (pw *pickerWrapper) exitIdleMode() {
|
|
|
|
pw.mu.Lock()
|
|
|
|
defer pw.mu.Unlock()
|
|
|
|
if pw.done {
|
|
|
|
return
|
|
|
|
}
|
|
|
|
pw.blockingCh = make(chan struct{})
|
|
|
|
pw.idle = false
|
|
|
|
}
|
|
|
|
|
2022-05-20 11:45:24 +00:00
|
|
|
// dropError is a wrapper error that indicates the LB policy wishes to drop the
|
|
|
|
// RPC and not retry it.
|
|
|
|
type dropError struct {
|
|
|
|
error
|
|
|
|
}
|