gitlab-com-seaweedfs/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2024-01-19 02:48:24 +00:00
Code Issues Projects Releases Wiki Activity

caching grpc clients

Browse source
This commit is contained in:
Chris Lu 2018-11-14 23:50:46 -08:00
parent d89774cb7f
commit d3b7965c76
2 changed files with 34 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
weed/filesys/wfs.go
View file

@ -66,6 +66,7 @@ func (wfs *WFS) withFilerClient(fn func(filer_pb.SeaweedFilerClient) error) erro
grpcConnection, err := util.GrpcDial(wfs.option.FilerGrpcAddress) grpcConnection, err := util.GrpcDial(wfs.option.FilerGrpcAddress)
if err != nil { if err != nil {
wfs.grpcClientsLock.Unlock()
return fmt.Errorf("fail to dial %s: %v", wfs.option.FilerGrpcAddress, err) return fmt.Errorf("fail to dial %s: %v", wfs.option.FilerGrpcAddress, err)
} }

35
weed/operation/grpc_client.go
View file

@ -9,6 +9,13 @@ import (
"github.com/chrislusf/seaweedfs/weed/pb/master_pb" "github.com/chrislusf/seaweedfs/weed/pb/master_pb"
"github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb" "github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb"
"github.com/chrislusf/seaweedfs/weed/util" "github.com/chrislusf/seaweedfs/weed/util"
"sync"
"google.golang.org/grpc"
)
var (
grpcClients = make(map[string]*grpc.ClientConn)
grpcClientsLock sync.Mutex
) )
func WithVolumeServerClient(volumeServer string, fn func(volume_server_pb.VolumeServerClient) error) error { func WithVolumeServerClient(volumeServer string, fn func(volume_server_pb.VolumeServerClient) error) error {
@ -18,11 +25,23 @@ func WithVolumeServerClient(volumeServer string, fn func(volume_server_pb.Volume
return err return err
} }
grpcClientsLock.Lock()
existingConnection, found := grpcClients[grpcAddress]
if found {
grpcClientsLock.Unlock()
client := volume_server_pb.NewVolumeServerClient(existingConnection)
return fn(client)
}
grpcConnection, err := util.GrpcDial(grpcAddress) grpcConnection, err := util.GrpcDial(grpcAddress)
if err != nil { if err != nil {
grpcClientsLock.Unlock()
return fmt.Errorf("fail to dial %s: %v", grpcAddress, err) return fmt.Errorf("fail to dial %s: %v", grpcAddress, err)
} }
defer grpcConnection.Close()
grpcClients[grpcAddress] = grpcConnection
grpcClientsLock.Unlock()
client := volume_server_pb.NewVolumeServerClient(grpcConnection) client := volume_server_pb.NewVolumeServerClient(grpcConnection)
@ -41,11 +60,23 @@ func toVolumeServerGrpcAddress(volumeServer string) (grpcAddress string, err err
func withMasterServerClient(masterServer string, fn func(masterClient master_pb.SeaweedClient) error) error { func withMasterServerClient(masterServer string, fn func(masterClient master_pb.SeaweedClient) error) error {
grpcClientsLock.Lock()
existingConnection, found := grpcClients[masterServer]
if found {
grpcClientsLock.Unlock()
client := master_pb.NewSeaweedClient(existingConnection)
return fn(client)
}
grpcConnection, err := util.GrpcDial(masterServer) grpcConnection, err := util.GrpcDial(masterServer)
if err != nil { if err != nil {
grpcClientsLock.Unlock()
return fmt.Errorf("fail to dial %s: %v", masterServer, err) return fmt.Errorf("fail to dial %s: %v", masterServer, err)
} }
defer grpcConnection.Close()
grpcClients[masterServer] = grpcConnection
grpcClientsLock.Unlock()
client := master_pb.NewSeaweedClient(grpcConnection) client := master_pb.NewSeaweedClient(grpcConnection)