caching ec shard locations

8 files changed, 45 insertions, 22 deletions

diff --git a/weed/operation/assign_file_id.go b/weed/operation/assign_file_id.go
index dc5c0c5d7..4c50eaa26 100644
--- a/weed/operation/assign_file_id.go
+++ b/weed/operation/assign_file_id.go
@@ -43,7 +43,7 @@ func Assign(server string, grpcDialOption grpc.DialOption, primaryRequest *Volum
 			continue
 		}
 
-		lastError = withMasterServerClient(server, grpcDialOption, func(masterClient master_pb.SeaweedClient) error {
+		lastError = WithMasterServerClient(server, grpcDialOption, func(masterClient master_pb.SeaweedClient) error {
 
 			req := &master_pb.AssignRequest{
 				Count:       primaryRequest.Count,
diff --git a/weed/operation/grpc_client.go b/weed/operation/grpc_client.go
index bf4d4bec3..f6b2b69e9 100644
--- a/weed/operation/grpc_client.go
+++ b/weed/operation/grpc_client.go
@@ -38,7 +38,7 @@ func toVolumeServerGrpcAddress(volumeServer string) (grpcAddress string, err err
 	return fmt.Sprintf("%s:%d", volumeServer[0:sepIndex], port+10000), nil
 }
 
-func withMasterServerClient(masterServer string, grpcDialOption grpc.DialOption, fn func(masterClient master_pb.SeaweedClient) error) error {
+func WithMasterServerClient(masterServer string, grpcDialOption grpc.DialOption, fn func(masterClient master_pb.SeaweedClient) error) error {
 
 	ctx := context.Background()
 
diff --git a/weed/operation/lookup.go b/weed/operation/lookup.go
index ca9056dbb..d0773e7fd 100644
--- a/weed/operation/lookup.go
+++ b/weed/operation/lookup.go
@@ -99,7 +99,7 @@ func LookupVolumeIds(server string, grpcDialOption grpc.DialOption, vids []strin
 
 	//only query unknown_vids
 
-	err := withMasterServerClient(server, grpcDialOption, func(masterClient master_pb.SeaweedClient) error {
+	err := WithMasterServerClient(server, grpcDialOption, func(masterClient master_pb.SeaweedClient) error {
 
 		req := &master_pb.LookupVolumeRequest{
 			VolumeIds: unknown_vids,
diff --git a/weed/operation/stats.go b/weed/operation/stats.go
index 08719cdb5..b69a33750 100644
--- a/weed/operation/stats.go
+++ b/weed/operation/stats.go
@@ -9,7 +9,7 @@ import (
 
 func Statistics(server string, grpcDialOption grpc.DialOption, req *master_pb.StatisticsRequest) (resp *master_pb.StatisticsResponse, err error) {
 
-	err = withMasterServerClient(server, grpcDialOption, func(masterClient master_pb.SeaweedClient) error {
+	err = WithMasterServerClient(server, grpcDialOption, func(masterClient master_pb.SeaweedClient) error {
 
 		grpcResponse, grpcErr := masterClient.Statistics(context.Background(), req)
 		if grpcErr != nil {
diff --git a/weed/server/volume_grpc_client_to_master.go b/weed/server/volume_grpc_client_to_master.go
index 595db73e0..d493b7233 100644
--- a/weed/server/volume_grpc_client_to_master.go
+++ b/weed/server/volume_grpc_client_to_master.go
@@ -39,13 +39,13 @@ func (vs *VolumeServer) heartbeat() {
 				glog.V(0).Infof("failed to parse master grpc %v: %v", masterGrpcAddress, parseErr)
 				continue
 			}
-			vs.store.MasterGrpcAddress = masterGrpcAddress
+			vs.store.MasterAddress = master
 			newLeader, err = vs.doHeartbeat(context.Background(), master, masterGrpcAddress, grpcDialOption, time.Duration(vs.pulseSeconds)*time.Second)
 			if err != nil {
 				glog.V(0).Infof("heartbeat error: %v", err)
 				time.Sleep(time.Duration(vs.pulseSeconds) * time.Second)
 				newLeader = ""
-				vs.store.MasterGrpcAddress = ""
+				vs.store.MasterAddress = ""
 			}
 		}
 	}
diff --git a/weed/storage/erasure_coding/ec_volume.go b/weed/storage/erasure_coding/ec_volume.go
index b4432c884..ecdc372bb 100644
--- a/weed/storage/erasure_coding/ec_volume.go
+++ b/weed/storage/erasure_coding/ec_volume.go
@@ -5,6 +5,8 @@ import (
 	"math"
 	"os"
 	"sort"
+	"sync"
+	"time"
 
 	"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
 	"github.com/chrislusf/seaweedfs/weed/storage/idx"
@@ -13,12 +15,15 @@ import (
 )
 
 type EcVolume struct {
-	Shards      []*EcVolumeShard
-	VolumeId    needle.VolumeId
-	Collection  string
-	dir         string
-	ecxFile     *os.File
-	ecxFileSize int64
+	VolumeId                  needle.VolumeId
+	Collection                string
+	dir                       string
+	ecxFile                   *os.File
+	ecxFileSize               int64
+	Shards                    []*EcVolumeShard
+	ShardLocations            map[ShardId][]string
+	ShardLocationsRefreshTime time.Time
+	ShardLocationsLock        sync.RWMutex
 }
 
 func NewEcVolume(dir string, collection string, vid needle.VolumeId) (ev *EcVolume, err error) {
diff --git a/weed/storage/store.go b/weed/storage/store.go
index d4fdb1a03..85ea2ae4c 100644
--- a/weed/storage/store.go
+++ b/weed/storage/store.go
@@ -19,7 +19,7 @@ const (
  * A VolumeServer contains one Store
  */
 type Store struct {
-	MasterGrpcAddress   string
+	MasterAddress       string
 	grpcDialOption      grpc.DialOption
 	volumeSizeLimit     uint64 //read from the master
 	Ip                  string
diff --git a/weed/storage/store_ec.go b/weed/storage/store_ec.go
index a75529fbd..daa8bbd19 100644
--- a/weed/storage/store_ec.go
+++ b/weed/storage/store_ec.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"time"
 
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/operation"
@@ -119,13 +120,13 @@ func (s *Store) ReadEcShardNeedle(ctx context.Context, vid needle.VolumeId, n *n
 }
 
 func (s *Store) readEcShardIntervals(ctx context.Context, vid needle.VolumeId, ecVolume *erasure_coding.EcVolume, intervals []erasure_coding.Interval) (data []byte, err error) {
-	shardLocations, err := s.cachedLookupEcShardLocations(ctx, vid)
-	if err != nil {
-		return nil, fmt.Errorf("failed to locate shard via master grpc %s: %v", s.MasterGrpcAddress, err)
+
+	if err = s.cachedLookupEcShardLocations(ctx, ecVolume); err != nil {
+		return nil, fmt.Errorf("failed to locate shard via master grpc %s: %v", s.MasterAddress, err)
 	}
 
 	for i, interval := range intervals {
-		if d, e := s.readOneEcShardInterval(ctx, vid, ecVolume, shardLocations, interval); e != nil {
+		if d, e := s.readOneEcShardInterval(ctx, ecVolume, interval); e != nil {
 			return nil, e
 		} else {
 			if i == 0 {
@@ -138,7 +139,7 @@ func (s *Store) readEcShardIntervals(ctx context.Context, vid needle.VolumeId, e
 	return
 }
 
-func (s *Store) readOneEcShardInterval(ctx context.Context, vid needle.VolumeId, ecVolume *erasure_coding.EcVolume, shardLocations map[erasure_coding.ShardId]string, interval erasure_coding.Interval) (data []byte, err error) {
+func (s *Store) readOneEcShardInterval(ctx context.Context, ecVolume *erasure_coding.EcVolume, interval erasure_coding.Interval) (data []byte, err error) {
 	shardId, actualOffset := interval.ToShardIdAndOffset(erasure_coding.ErasureCodingLargeBlockSize, erasure_coding.ErasureCodingSmallBlockSize)
 	data = make([]byte, interval.Size)
 	if shard, found := ecVolume.FindEcVolumeShard(shardId); found {
@@ -146,16 +147,33 @@ func (s *Store) readOneEcShardInterval(ctx context.Context, vid needle.VolumeId,
 			return
 		}
 	} else {
-		sourceDataNode := shardLocations[shardId]
-		_, err = s.readOneRemoteEcShardInterval(ctx, sourceDataNode, vid, shardId, data, actualOffset)
+		ecVolume.ShardLocationsLock.RLock()
+		sourceDataNodes, found := ecVolume.ShardLocations[shardId]
+		ecVolume.ShardLocationsLock.RUnlock()
+		if !found || len(sourceDataNodes) == 0 {
+			return nil, fmt.Errorf("failed to find ec shard %d.%d", ecVolume.VolumeId, shardId)
+		}
+		_, err = s.readOneRemoteEcShardInterval(ctx, sourceDataNodes[0], ecVolume.VolumeId, shardId, data, actualOffset)
 		if err != nil {
-			glog.V(1).Infof("failed to read from %s for ec shard %d.%d : %v", sourceDataNode, vid, shardId, err)
+			glog.V(1).Infof("failed to read from %s for ec shard %d.%d : %v", sourceDataNodes[0], ecVolume.VolumeId, shardId, err)
 		}
 	}
 	return
 }
 
-func (s *Store) cachedLookupEcShardLocations(ctx context.Context, vid needle.VolumeId) (shardLocations map[erasure_coding.ShardId]string, err error) {
+func (s *Store) cachedLookupEcShardLocations(ctx context.Context, ecVolume *erasure_coding.EcVolume) (err error) {
+
+	if ecVolume.ShardLocationsRefreshTime.Add(10 * time.Minute).After(time.Now()) {
+		// still fresh
+		return nil
+	}
+
+	ecVolume.ShardLocationsLock.Lock()
+	defer ecVolume.ShardLocationsLock.Unlock()
+
+	err = operation.WithMasterServerClient(s.MasterAddress, s.grpcDialOption, func(masterClient master_pb.SeaweedClient) error {
+		return nil
+	})
 	return
 }