summaryrefslogtreecommitdiff
path: root/vendor/google.golang.org/grpc/picker_wrapper.go
diff options
context:
space:
mode:
authorLibravatar tobi <31960611+tsmethurst@users.noreply.github.com>2024-08-26 18:05:54 +0200
committerLibravatar GitHub <noreply@github.com>2024-08-26 18:05:54 +0200
commit28d57d1f13ee61a6ff83ce4beaf238139d20bbac (patch)
tree7946abb44b21f9e2f4267146711760a911072c9b /vendor/google.golang.org/grpc/picker_wrapper.go
parent[chore]: Bump github.com/prometheus/client_golang from 1.20.0 to 1.20.2 (#3239) (diff)
downloadgotosocial-28d57d1f13ee61a6ff83ce4beaf238139d20bbac.tar.xz
[chore] Bump all otel deps (#3241)
Diffstat (limited to 'vendor/google.golang.org/grpc/picker_wrapper.go')
-rw-r--r--vendor/google.golang.org/grpc/picker_wrapper.go84
1 files changed, 40 insertions, 44 deletions
diff --git a/vendor/google.golang.org/grpc/picker_wrapper.go b/vendor/google.golang.org/grpc/picker_wrapper.go
index bf56faa76..bdaa2130e 100644
--- a/vendor/google.golang.org/grpc/picker_wrapper.go
+++ b/vendor/google.golang.org/grpc/picker_wrapper.go
@@ -20,8 +20,9 @@ package grpc
import (
"context"
+ "fmt"
"io"
- "sync"
+ "sync/atomic"
"google.golang.org/grpc/balancer"
"google.golang.org/grpc/codes"
@@ -32,35 +33,43 @@ import (
"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
+ // If pickerGen holds a nil pointer, the pickerWrapper is closed.
+ pickerGen atomic.Pointer[pickerGeneration]
statsHandlers []stats.Handler // to record blocking picker calls
}
func newPickerWrapper(statsHandlers []stats.Handler) *pickerWrapper {
- return &pickerWrapper{
- blockingCh: make(chan struct{}),
+ pw := &pickerWrapper{
statsHandlers: statsHandlers,
}
+ pw.pickerGen.Store(&pickerGeneration{
+ blockingCh: make(chan struct{}),
+ })
+ return pw
}
-// updatePicker is called by UpdateBalancerState. It unblocks all blocked pick.
+// updatePicker is called by UpdateState calls from the LB policy. It
+// unblocks all blocked pick.
func (pw *pickerWrapper) updatePicker(p balancer.Picker) {
- pw.mu.Lock()
- if pw.done {
- pw.mu.Unlock()
- return
- }
- pw.picker = p
- // pw.blockingCh should never be nil.
- close(pw.blockingCh)
- pw.blockingCh = make(chan struct{})
- pw.mu.Unlock()
+ old := pw.pickerGen.Swap(&pickerGeneration{
+ picker: p,
+ blockingCh: make(chan struct{}),
+ })
+ close(old.blockingCh)
}
// doneChannelzWrapper performs the following:
@@ -97,27 +106,24 @@ func (pw *pickerWrapper) pick(ctx context.Context, failfast bool, info balancer.
var lastPickErr error
for {
- pw.mu.Lock()
- if pw.done {
- pw.mu.Unlock()
+ pg := pw.pickerGen.Load()
+ if pg == nil {
return nil, balancer.PickResult{}, ErrClientConnClosing
}
-
- if pw.picker == nil {
- ch = pw.blockingCh
+ if pg.picker == nil {
+ ch = pg.blockingCh
}
- if ch == pw.blockingCh {
+ if ch == pg.blockingCh {
// This could happen when either:
// - pw.picker is nil (the previous if condition), or
- // - has called pick on the current picker.
- pw.mu.Unlock()
+ // - we have already called pick on the current picker.
select {
case <-ctx.Done():
var errStr string
if lastPickErr != nil {
errStr = "latest balancer error: " + lastPickErr.Error()
} else {
- errStr = ctx.Err().Error()
+ errStr = fmt.Sprintf("received context error while waiting for new LB policy update: %s", ctx.Err().Error())
}
switch ctx.Err() {
case context.DeadlineExceeded:
@@ -144,9 +150,8 @@ func (pw *pickerWrapper) pick(ctx context.Context, failfast bool, info balancer.
}
}
- ch = pw.blockingCh
- p := pw.picker
- pw.mu.Unlock()
+ ch = pg.blockingCh
+ p := pg.picker
pickResult, err := p.Pick(info)
if err != nil {
@@ -196,24 +201,15 @@ func (pw *pickerWrapper) pick(ctx context.Context, failfast bool, info balancer.
}
func (pw *pickerWrapper) close() {
- pw.mu.Lock()
- defer pw.mu.Unlock()
- if pw.done {
- return
- }
- pw.done = true
- close(pw.blockingCh)
+ 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() {
- pw.mu.Lock()
- defer pw.mu.Unlock()
- if pw.done {
- return
- }
- pw.blockingCh = make(chan struct{})
+ 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