lsm/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2025-10-21 07:07:25 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

filer: re-create grpc connections if having transport error

Browse Source 
fix https://github.com/chrislusf/seaweedfs/issues/2070
This commit is contained in:
Chris Lu
2021-05-22 17:46:53 -07:00
parent 2d0a0733ec
commit 3d3fa43542
1 changed files with 28 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
weed/pb/grpc_client_server.go
View File

@@ -4,6 +4,7 @@ import (
"context" "context"
"fmt" "fmt"
"github.com/chrislusf/seaweedfs/weed/glog" "github.com/chrislusf/seaweedfs/weed/glog"
"math/rand"
"net/http" "net/http"
"strconv" "strconv"
"strings" "strings"
@@ -24,10 +25,15 @@ const (
var ( var (
// cache grpc connections // cache grpc connections
grpcClients = make(map[string]*grpc.ClientConn) grpcClients = make(map[string]*versionedGrpcClient)
grpcClientsLock sync.Mutex grpcClientsLock sync.Mutex
) )
type versionedGrpcClient struct {
*grpc.ClientConn
version int
}
func init() { func init() {
http.DefaultTransport.(*http.Transport).MaxIdleConnsPerHost = 1024 http.DefaultTransport.(*http.Transport).MaxIdleConnsPerHost = 1024
http.DefaultTransport.(*http.Transport).MaxIdleConns = 1024 http.DefaultTransport.(*http.Transport).MaxIdleConns = 1024
@@ -79,7 +85,7 @@ func GrpcDial(ctx context.Context, address string, opts ...grpc.DialOption) (*gr
return grpc.DialContext(ctx, address, options...) return grpc.DialContext(ctx, address, options...)
} }
func getOrCreateConnection(address string, opts ...grpc.DialOption) (*grpc.ClientConn, error) { func getOrCreateConnection(address string, opts ...grpc.DialOption) (*versionedGrpcClient, error) {
grpcClientsLock.Lock() grpcClientsLock.Lock()
defer grpcClientsLock.Unlock() defer grpcClientsLock.Unlock()
@@ -94,18 +100,34 @@ func getOrCreateConnection(address string, opts ...grpc.DialOption) (*grpc.Clien
return nil, fmt.Errorf("fail to dial %s: %v", address, err) return nil, fmt.Errorf("fail to dial %s: %v", address, err)
} }
grpcClients[address] = grpcConnection vgc := &versionedGrpcClient{
grpcConnection,
rand.Int(),
}
grpcClients[address] = vgc
return grpcConnection, nil return vgc, nil
} }
func WithCachedGrpcClient(fn func(*grpc.ClientConn) error, address string, opts ...grpc.DialOption) error { func WithCachedGrpcClient(fn func(*grpc.ClientConn) error, address string, opts ...grpc.DialOption) error {
grpcConnection, err := getOrCreateConnection(address, opts...) vgc, err := getOrCreateConnection(address, opts...)
if err != nil { if err != nil {
return fmt.Errorf("getOrCreateConnection %s: %v", address, err) return fmt.Errorf("getOrCreateConnection %s: %v", address, err)
} }
return fn(grpcConnection) executionErr := fn(vgc.ClientConn)
if executionErr != nil && strings.Contains(executionErr.Error(), "transport") {
grpcClientsLock.Lock()
if t, ok := grpcClients[address]; ok {
if t.version == vgc.version {
vgc.Close()
delete(grpcClients, address)
}
}
grpcClientsLock.Unlock()
}
return executionErr
} }
func ParseServerToGrpcAddress(server string) (serverGrpcAddress string, err error) { func ParseServerToGrpcAddress(server string) (serverGrpcAddress string, err error) {