Merge pull request #1 from chrislusf/master

update from chrisluf 

1 files changed, 27 insertions, 26 deletions

diff --git a/weed/shell/command_ec_encode.go b/weed/shell/command_ec_encode.go
index f07cb93f9..587b59388 100644
--- a/weed/shell/command_ec_encode.go
+++ b/weed/shell/command_ec_encode.go
@@ -8,13 +8,14 @@ import (
 	"sync"
 	"time"
 
+	"google.golang.org/grpc"
+
 	"github.com/chrislusf/seaweedfs/weed/operation"
 	"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
 	"github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb"
 	"github.com/chrislusf/seaweedfs/weed/storage/erasure_coding"
 	"github.com/chrislusf/seaweedfs/weed/storage/needle"
 	"github.com/chrislusf/seaweedfs/weed/wdclient"
-	"google.golang.org/grpc"
 )
 
 func init() {
@@ -92,10 +93,12 @@ func doEcEncode(ctx context.Context, commandEnv *CommandEnv, collection string,
 		return fmt.Errorf("volume %d not found", vid)
 	}
 
+	// fmt.Printf("found ec %d shards on %v\n", vid, locations)
+
 	// mark the volume as readonly
 	err = markVolumeReadonly(ctx, commandEnv.option.GrpcDialOption, needle.VolumeId(vid), locations)
 	if err != nil {
-		return fmt.Errorf("generate ec shards for volume %d on %s: %v", vid, locations[0].Url, err)
+		return fmt.Errorf("mark volume %d as readonly on %s: %v", vid, locations[0].Url, err)
 	}
 
 	// generate ec shards
@@ -117,7 +120,7 @@ func markVolumeReadonly(ctx context.Context, grpcDialOption grpc.DialOption, vol
 
 	for _, location := range locations {
 
-		err := operation.WithVolumeServerClient(location.Url, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
+		err := operation.WithVolumeServerClient(location.Url, grpcDialOption, func(ctx context.Context, volumeServerClient volume_server_pb.VolumeServerClient) error {
 			_, markErr := volumeServerClient.VolumeMarkReadonly(ctx, &volume_server_pb.VolumeMarkReadonlyRequest{
 				VolumeId: uint32(volumeId),
 			})
@@ -135,7 +138,7 @@ func markVolumeReadonly(ctx context.Context, grpcDialOption grpc.DialOption, vol
 
 func generateEcShards(ctx context.Context, grpcDialOption grpc.DialOption, volumeId needle.VolumeId, collection string, sourceVolumeServer string) error {
 
-	err := operation.WithVolumeServerClient(sourceVolumeServer, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
+	err := operation.WithVolumeServerClient(sourceVolumeServer, grpcDialOption, func(ctx context.Context, volumeServerClient volume_server_pb.VolumeServerClient) error {
 		_, genErr := volumeServerClient.VolumeEcShardsGenerate(ctx, &volume_server_pb.VolumeEcShardsGenerateRequest{
 			VolumeId:   uint32(volumeId),
 			Collection: collection,
@@ -163,10 +166,10 @@ func spreadEcShards(ctx context.Context, commandEnv *CommandEnv, volumeId needle
 	}
 
 	// calculate how many shards to allocate for these servers
-	allocated := balancedEcDistribution(allocatedDataNodes)
+	allocatedEcIds := balancedEcDistribution(allocatedDataNodes)
 
 	// ask the data nodes to copy from the source volume server
-	copiedShardIds, err := parallelCopyEcShardsFromSource(ctx, commandEnv.option.GrpcDialOption, allocatedDataNodes, allocated, volumeId, collection, existingLocations[0])
+	copiedShardIds, err := parallelCopyEcShardsFromSource(ctx, commandEnv.option.GrpcDialOption, allocatedDataNodes, allocatedEcIds, volumeId, collection, existingLocations[0])
 	if err != nil {
 		return err
 	}
@@ -196,31 +199,29 @@ func spreadEcShards(ctx context.Context, commandEnv *CommandEnv, volumeId needle
 }
 
 func parallelCopyEcShardsFromSource(ctx context.Context, grpcDialOption grpc.DialOption,
-	targetServers []*EcNode, allocated []int,
+	targetServers []*EcNode, allocatedEcIds [][]uint32,
 	volumeId needle.VolumeId, collection string, existingLocation wdclient.Location) (actuallyCopied []uint32, err error) {
 
 	// parallelize
 	shardIdChan := make(chan []uint32, len(targetServers))
 	var wg sync.WaitGroup
-	startFromShardId := uint32(0)
 	for i, server := range targetServers {
-		if allocated[i] <= 0 {
+		if len(allocatedEcIds[i]) <= 0 {
 			continue
 		}
 
 		wg.Add(1)
-		go func(server *EcNode, startFromShardId uint32, shardCount int) {
+		go func(server *EcNode, allocatedEcShardIds []uint32) {
 			defer wg.Done()
 			copiedShardIds, copyErr := oneServerCopyAndMountEcShardsFromSource(ctx, grpcDialOption, server,
-				startFromShardId, shardCount, volumeId, collection, existingLocation.Url)
+				allocatedEcShardIds, volumeId, collection, existingLocation.Url)
 			if copyErr != nil {
 				err = copyErr
 			} else {
 				shardIdChan <- copiedShardIds
 				server.addEcVolumeShards(volumeId, collection, copiedShardIds)
 			}
-		}(server, startFromShardId, allocated[i])
-		startFromShardId += uint32(allocated[i])
+		}(server, allocatedEcIds[i])
 	}
 	wg.Wait()
 	close(shardIdChan)
@@ -236,18 +237,18 @@ func parallelCopyEcShardsFromSource(ctx context.Context, grpcDialOption grpc.Dia
 	return
 }
 
-func balancedEcDistribution(servers []*EcNode) (allocated []int) {
-	allocated = make([]int, len(servers))
-	allocatedCount := 0
-	for allocatedCount < erasure_coding.TotalShardsCount {
-		for i, server := range servers {
-			if server.freeEcSlot-allocated[i] > 0 {
-				allocated[i] += 1
-				allocatedCount += 1
-			}
-			if allocatedCount >= erasure_coding.TotalShardsCount {
-				break
-			}
+func balancedEcDistribution(servers []*EcNode) (allocated [][]uint32) {
+	allocated = make([][]uint32, len(servers))
+	allocatedShardIdIndex := uint32(0)
+	serverIndex := 0
+	for allocatedShardIdIndex < erasure_coding.TotalShardsCount {
+		if servers[serverIndex].freeEcSlot > 0 {
+			allocated[serverIndex] = append(allocated[serverIndex], allocatedShardIdIndex)
+			allocatedShardIdIndex++
+		}
+		serverIndex++
+		if serverIndex >= len(servers) {
+			serverIndex = 0
 		}
 	}
 
@@ -281,7 +282,7 @@ func collectVolumeIdsForEcEncode(ctx context.Context, commandEnv *CommandEnv, se
 		}
 	})
 
-	for vid, _ := range vidMap {
+	for vid := range vidMap {
 		vids = append(vids, needle.VolumeId(vid))
 	}