aboutsummaryrefslogtreecommitdiff
path: root/weed/server/master_grpc_server_volume.go
diff options
context:
space:
mode:
Diffstat (limited to 'weed/server/master_grpc_server_volume.go')
-rw-r--r--weed/server/master_grpc_server_volume.go51
1 files changed, 36 insertions, 15 deletions
diff --git a/weed/server/master_grpc_server_volume.go b/weed/server/master_grpc_server_volume.go
index afe062a2d..f48326459 100644
--- a/weed/server/master_grpc_server_volume.go
+++ b/weed/server/master_grpc_server_volume.go
@@ -3,6 +3,8 @@ package weed_server
import (
"context"
"fmt"
+ "github.com/seaweedfs/seaweedfs/weed/topology"
+ "math/rand"
"reflect"
"strings"
"sync"
@@ -18,8 +20,39 @@ import (
"github.com/seaweedfs/seaweedfs/weed/storage/types"
)
+func (ms *MasterServer) DoAutomaticVolumeGrow(req *topology.VolumeGrowRequest) {
+ glog.V(1).Infoln("starting automatic volume grow")
+ start := time.Now()
+ newVidLocations, err := ms.vg.AutomaticGrowByType(req.Option, ms.grpcDialOption, ms.Topo, req.Count)
+ glog.V(1).Infoln("finished automatic volume grow, cost ", time.Now().Sub(start))
+ if err != nil {
+ glog.V(1).Infof("automatic volume grow failed: %+v", err)
+ return
+ }
+ for _, newVidLocation := range newVidLocations {
+ ms.broadcastToClients(&master_pb.KeepConnectedResponse{VolumeLocation: newVidLocation})
+ }
+}
+
func (ms *MasterServer) ProcessGrowRequest() {
go func() {
+ for {
+ time.Sleep(14*time.Minute + time.Duration(120*rand.Float32())*time.Second)
+ if !ms.Topo.IsLeader() {
+ continue
+ }
+ for _, vl := range ms.Topo.ListVolumeLyauts() {
+ if !vl.HasGrowRequest() && vl.ShouldGrowVolumes(&topology.VolumeGrowOption{}) {
+ vl.AddGrowRequest()
+ ms.volumeGrowthRequestChan <- &topology.VolumeGrowRequest{
+ Option: vl.ToGrowOption(),
+ Count: vl.GetLastGrowCount(),
+ }
+ }
+ }
+ }
+ }()
+ go func() {
filter := sync.Map{}
for {
req, ok := <-ms.volumeGrowthRequestChan
@@ -50,23 +83,11 @@ func (ms *MasterServer) ProcessGrowRequest() {
if !found && vl.ShouldGrowVolumes(option) {
filter.Store(req, nil)
// we have lock called inside vg
- go func() {
- glog.V(1).Infoln("starting automatic volume grow")
- start := time.Now()
- newVidLocations, err := ms.vg.AutomaticGrowByType(req.Option, ms.grpcDialOption, ms.Topo, req.Count)
- glog.V(1).Infoln("finished automatic volume grow, cost ", time.Now().Sub(start))
- if err == nil {
- for _, newVidLocation := range newVidLocations {
- ms.broadcastToClients(&master_pb.KeepConnectedResponse{VolumeLocation: newVidLocation})
- }
- } else {
- glog.V(1).Infof("automatic volume grow failed: %+v", err)
- }
+ go func(req *topology.VolumeGrowRequest, vl *topology.VolumeLayout) {
+ ms.DoAutomaticVolumeGrow(req)
vl.DoneGrowRequest()
-
filter.Delete(req)
- }()
-
+ }(req, vl)
} else {
glog.V(4).Infoln("discard volume grow request")
time.Sleep(time.Millisecond * 211)