aboutsummaryrefslogtreecommitdiff
path: root/weed/wdclient
diff options
context:
space:
mode:
Diffstat (limited to 'weed/wdclient')
-rw-r--r--weed/wdclient/masterclient.go63
-rw-r--r--weed/wdclient/vid_map.go23
-rw-r--r--weed/wdclient/wdclient.go15
3 files changed, 52 insertions, 49 deletions
diff --git a/weed/wdclient/masterclient.go b/weed/wdclient/masterclient.go
index b26853945..e16356e9d 100644
--- a/weed/wdclient/masterclient.go
+++ b/weed/wdclient/masterclient.go
@@ -3,29 +3,32 @@ package wdclient
import (
"context"
"fmt"
+ "math/rand"
"time"
"github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
"github.com/chrislusf/seaweedfs/weed/util"
- "math/rand"
+ "google.golang.org/grpc"
)
type MasterClient struct {
- ctx context.Context
- name string
- currentMaster string
- masters []string
+ ctx context.Context
+ name string
+ currentMaster string
+ masters []string
+ grpcDialOption grpc.DialOption
vidMap
}
-func NewMasterClient(ctx context.Context, clientName string, masters []string) *MasterClient {
+func NewMasterClient(ctx context.Context, grpcDialOption grpc.DialOption, clientName string, masters []string) *MasterClient {
return &MasterClient{
- ctx: ctx,
- name: clientName,
- masters: masters,
- vidMap: newVidMap(),
+ ctx: ctx,
+ name: clientName,
+ masters: masters,
+ grpcDialOption: grpcDialOption,
+ vidMap: newVidMap(),
}
}
@@ -40,7 +43,7 @@ func (mc *MasterClient) WaitUntilConnected() {
}
func (mc *MasterClient) KeepConnectedToMaster() {
- glog.V(0).Infof("%s bootstraps with masters %v", mc.name, mc.masters)
+ glog.V(1).Infof("%s bootstraps with masters %v", mc.name, mc.masters)
for {
mc.tryAllMasters()
time.Sleep(time.Second)
@@ -49,28 +52,28 @@ func (mc *MasterClient) KeepConnectedToMaster() {
func (mc *MasterClient) tryAllMasters() {
for _, master := range mc.masters {
- glog.V(0).Infof("Connecting to master %v", master)
- gprcErr := withMasterClient(master, func(client master_pb.SeaweedClient) error {
+ glog.V(1).Infof("%s Connecting to master %v", mc.name, master)
+ gprcErr := withMasterClient(context.Background(), master, mc.grpcDialOption, func(ctx context.Context, client master_pb.SeaweedClient) error {
- stream, err := client.KeepConnected(context.Background())
+ stream, err := client.KeepConnected(ctx)
if err != nil {
- glog.V(0).Infof("failed to keep connected to %s: %v", master, err)
+ glog.V(0).Infof("%s failed to keep connected to %s: %v", mc.name, master, err)
return err
}
if err = stream.Send(&master_pb.ClientListenRequest{Name: mc.name}); err != nil {
- glog.V(0).Infof("failed to send to %s: %v", master, err)
+ glog.V(0).Infof("%s failed to send to %s: %v", mc.name, master, err)
return err
}
if mc.currentMaster == "" {
- glog.V(0).Infof("Connected to %v", master)
+ glog.V(1).Infof("%s Connected to %v", mc.name, master)
mc.currentMaster = master
}
for {
if volumeLocation, err := stream.Recv(); err != nil {
- glog.V(0).Infof("failed to receive from %s: %v", master, err)
+ glog.V(0).Infof("%s failed to receive from %s: %v", mc.name, master, err)
return err
} else {
loc := Location{
@@ -78,9 +81,11 @@ func (mc *MasterClient) tryAllMasters() {
PublicUrl: volumeLocation.PublicUrl,
}
for _, newVid := range volumeLocation.NewVids {
+ glog.V(1).Infof("%s: %s adds volume %d", mc.name, loc.Url, newVid)
mc.addLocation(newVid, loc)
}
for _, deletedVid := range volumeLocation.DeletedVids {
+ glog.V(1).Infof("%s: %s removes volume %d", mc.name, loc.Url, deletedVid)
mc.deleteLocation(deletedVid, loc)
}
}
@@ -96,20 +101,22 @@ func (mc *MasterClient) tryAllMasters() {
}
}
-func withMasterClient(master string, fn func(client master_pb.SeaweedClient) error) error {
+func withMasterClient(ctx context.Context, master string, grpcDialOption grpc.DialOption, fn func(ctx context.Context, client master_pb.SeaweedClient) error) error {
- masterGrpcAddress, parseErr := util.ParseServerToGrpcAddress(master, 0)
+ masterGrpcAddress, parseErr := util.ParseServerToGrpcAddress(master)
if parseErr != nil {
- return fmt.Errorf("failed to parse master grpc %v", master)
+ return fmt.Errorf("failed to parse master grpc %v: %v", master, parseErr)
}
- grpcConnection, err := util.GrpcDial(masterGrpcAddress)
- if err != nil {
- return fmt.Errorf("fail to dial %s: %v", master, err)
- }
- defer grpcConnection.Close()
+ return util.WithCachedGrpcClient(ctx, func(grpcConnection *grpc.ClientConn) error {
+ client := master_pb.NewSeaweedClient(grpcConnection)
+ return fn(ctx, client)
+ }, masterGrpcAddress, grpcDialOption)
- client := master_pb.NewSeaweedClient(grpcConnection)
+}
- return fn(client)
+func (mc *MasterClient) WithClient(ctx context.Context, fn func(client master_pb.SeaweedClient) error) error {
+ return withMasterClient(ctx, mc.currentMaster, mc.grpcDialOption, func(ctx context.Context, client master_pb.SeaweedClient) error {
+ return fn(client)
+ })
}
diff --git a/weed/wdclient/vid_map.go b/weed/wdclient/vid_map.go
index aef29f56f..06308944d 100644
--- a/weed/wdclient/vid_map.go
+++ b/weed/wdclient/vid_map.go
@@ -7,6 +7,7 @@ import (
"strconv"
"strings"
"sync"
+ "time"
"github.com/chrislusf/seaweedfs/weed/glog"
)
@@ -19,11 +20,13 @@ type Location struct {
type vidMap struct {
sync.RWMutex
vid2Locations map[uint32][]Location
+ r *rand.Rand
}
func newVidMap() vidMap {
return vidMap{
vid2Locations: make(map[uint32][]Location),
+ r: rand.New(rand.NewSource(time.Now().UnixNano())),
}
}
@@ -34,12 +37,7 @@ func (vc *vidMap) LookupVolumeServerUrl(vid string) (serverUrl string, err error
return "", err
}
- locations := vc.GetLocations(uint32(id))
- if len(locations) == 0 {
- return "", fmt.Errorf("volume %d not found", id)
- }
-
- return locations[rand.Intn(len(locations))].Url, nil
+ return vc.GetRandomLocation(uint32(id))
}
func (vc *vidMap) LookupFileId(fileId string) (fullUrl string, err error) {
@@ -82,6 +80,18 @@ func (vc *vidMap) GetLocations(vid uint32) (locations []Location) {
return vc.vid2Locations[vid]
}
+func (vc *vidMap) GetRandomLocation(vid uint32) (serverUrl string, err error) {
+ vc.RLock()
+ defer vc.RUnlock()
+
+ locations := vc.vid2Locations[vid]
+ if len(locations) == 0 {
+ return "", fmt.Errorf("volume %d not found", vid)
+ }
+
+ return locations[vc.r.Intn(len(locations))].Url, nil
+}
+
func (vc *vidMap) addLocation(vid uint32, location Location) {
vc.Lock()
defer vc.Unlock()
@@ -114,6 +124,7 @@ func (vc *vidMap) deleteLocation(vid uint32, location Location) {
for i, loc := range locations {
if loc.Url == location.Url {
vc.vid2Locations[vid] = append(locations[0:i], locations[i+1:]...)
+ break
}
}
diff --git a/weed/wdclient/wdclient.go b/weed/wdclient/wdclient.go
deleted file mode 100644
index 722f4d061..000000000
--- a/weed/wdclient/wdclient.go
+++ /dev/null
@@ -1,15 +0,0 @@
-package wdclient
-
-import (
- "context"
-)
-
-type SeaweedClient struct {
- *MasterClient
-}
-
-func NewSeaweedClient(ctx context.Context, clientName string, masters []string) *SeaweedClient {
- return &SeaweedClient{
- MasterClient: NewMasterClient(ctx, clientName, masters),
- }
-}
generated by cgit v1.2.3 (git 2.25.1) at 2025-12-24 16:18:19 +0000