aboutsummaryrefslogtreecommitdiff
path: root/weed/operation/grpc_client.go
diff options
context:
space:
mode:
authorChris Lu <chris.lu@gmail.com>2018-11-14 23:50:46 -0800
committerChris Lu <chris.lu@gmail.com>2018-11-14 23:50:46 -0800
commitd3b7965c76a9f146e6d3db99a836f1bad4ca3c40 (patch)
treef9f09554e3c987e679a47236c709fc684ee4dfab /weed/operation/grpc_client.go
parentd89774cb7f4105223fde3ef0a2fcd85d90524e9c (diff)
downloadseaweedfs-d3b7965c76a9f146e6d3db99a836f1bad4ca3c40.tar.xz
seaweedfs-d3b7965c76a9f146e6d3db99a836f1bad4ca3c40.zip
caching grpc clients
Diffstat (limited to 'weed/operation/grpc_client.go')
-rw-r--r--weed/operation/grpc_client.go35
1 files changed, 33 insertions, 2 deletions
diff --git a/weed/operation/grpc_client.go b/weed/operation/grpc_client.go
index 5e6c23709..6720fe3a3 100644
--- a/weed/operation/grpc_client.go
+++ b/weed/operation/grpc_client.go
@@ -9,6 +9,13 @@ import (
"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
"github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb"
"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 {
@@ -18,11 +25,23 @@ func WithVolumeServerClient(volumeServer string, fn func(volume_server_pb.Volume
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)
if err != nil {
+ grpcClientsLock.Unlock()
return fmt.Errorf("fail to dial %s: %v", grpcAddress, err)
}
- defer grpcConnection.Close()
+
+ grpcClients[grpcAddress] = grpcConnection
+ grpcClientsLock.Unlock()
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 {
+ grpcClientsLock.Lock()
+
+ existingConnection, found := grpcClients[masterServer]
+ if found {
+ grpcClientsLock.Unlock()
+ client := master_pb.NewSeaweedClient(existingConnection)
+ return fn(client)
+ }
+
grpcConnection, err := util.GrpcDial(masterServer)
if err != nil {
+ grpcClientsLock.Unlock()
return fmt.Errorf("fail to dial %s: %v", masterServer, err)
}
- defer grpcConnection.Close()
+
+ grpcClients[masterServer] = grpcConnection
+ grpcClientsLock.Unlock()
client := master_pb.NewSeaweedClient(grpcConnection)