mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-21 22:12:36 +00:00
mount: bound each filer call so a wedged filer cannot hold the failover walk
WithFilerClient callers run their RPCs on context.Background(), so a filer whose transport stalls without refusing (stopped peer, black-holed connection) blocks the call forever and the walk never reaches the remaining filer addresses. Run each attempt under a timeout; on expiry drop the cached channel so the abandoned call is unblocked and the next address is tried. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
co-authored by
Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
parent
e337443176
commit
69ec4c3f0a
@@ -145,6 +145,7 @@ type WFS struct {
|
||||
manifestCache *filer.ChunkManifestCache
|
||||
writeBufferAccountant *page_writer.WriteBufferAccountant
|
||||
signature int32
|
||||
filerCallTimeout time.Duration // bounds one filer call; tests only, 0 = filerRPCTimeout
|
||||
concurrentWriters *util.LimitedConcurrentExecutor
|
||||
copyBufferPool sync.Pool
|
||||
concurrentCopiersSem chan struct{}
|
||||
|
||||
@@ -2,9 +2,13 @@ package mount
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"google.golang.org/grpc"
|
||||
"google.golang.org/grpc/codes"
|
||||
"google.golang.org/grpc/status"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
||||
@@ -22,10 +26,7 @@ func (wfs *WFS) WithFilerClient(streamingMode bool, fn func(filer_pb.SeaweedFile
|
||||
for x := 0; x < n; x++ {
|
||||
|
||||
filerGrpcAddress := wfs.option.FilerAddresses[i].ToGrpcAddress()
|
||||
err = pb.WithGrpcClient(context.Background(), streamingMode, wfs.signature, func(grpcConnection *grpc.ClientConn) error {
|
||||
client := filer_pb.NewSeaweedFilerClient(grpcConnection)
|
||||
return fn(client)
|
||||
}, filerGrpcAddress, false, wfs.option.GrpcDialOption)
|
||||
err = wfs.callFiler(streamingMode, filerGrpcAddress, fn)
|
||||
|
||||
if err != nil {
|
||||
// glog.V(0).Infof("WithFilerClient %d %v: %v", x, filerGrpcAddress, err)
|
||||
@@ -45,6 +46,44 @@ func (wfs *WFS) WithFilerClient(streamingMode bool, fn func(filer_pb.SeaweedFile
|
||||
|
||||
}
|
||||
|
||||
// filerRPCWait bounds one filer call when the callers' RPCs carry no deadline
|
||||
// of their own. filerCallTimeout is only settable on the WFS for tests.
|
||||
func (wfs *WFS) filerRPCWait() time.Duration {
|
||||
if wfs.filerCallTimeout > 0 {
|
||||
return wfs.filerCallTimeout
|
||||
}
|
||||
return filerRPCTimeout
|
||||
}
|
||||
|
||||
// callFiler runs fn against one filer. Callers issue their RPCs on
|
||||
// context.Background(), so a filer whose transport silently stalls (stopped
|
||||
// peer, black-holed connection) would otherwise block forever and keep the
|
||||
// walk from reaching the remaining filer addresses. Bound the call; on
|
||||
// timeout drop the cached channel so the abandoned attempt is unblocked and
|
||||
// the next attempt redials.
|
||||
func (wfs *WFS) callFiler(streamingMode bool, filerGrpcAddress string, fn func(filer_pb.SeaweedFilerClient) error) error {
|
||||
if streamingMode {
|
||||
return pb.WithGrpcClient(context.Background(), streamingMode, wfs.signature, func(grpcConnection *grpc.ClientConn) error {
|
||||
return fn(filer_pb.NewSeaweedFilerClient(grpcConnection))
|
||||
}, filerGrpcAddress, false, wfs.option.GrpcDialOption)
|
||||
}
|
||||
|
||||
done := make(chan error, 1)
|
||||
go func() {
|
||||
done <- pb.WithGrpcClient(context.Background(), streamingMode, wfs.signature, func(grpcConnection *grpc.ClientConn) error {
|
||||
return fn(filer_pb.NewSeaweedFilerClient(grpcConnection))
|
||||
}, filerGrpcAddress, false, wfs.option.GrpcDialOption)
|
||||
}()
|
||||
|
||||
select {
|
||||
case err := <-done:
|
||||
return err
|
||||
case <-time.After(wfs.filerRPCWait()):
|
||||
pb.InvalidateGrpcConnection(filerGrpcAddress)
|
||||
return status.Error(codes.Unavailable, fmt.Sprintf("filer %s: call timed out", filerGrpcAddress))
|
||||
}
|
||||
}
|
||||
|
||||
func (wfs *WFS) AdjustedUrl(location *filer_pb.Location) string {
|
||||
if wfs.option.VolumeServerAccess == "publicUrl" {
|
||||
return location.PublicUrl
|
||||
|
||||
Reference in New Issue
Block a user