ec shard balancing

1 files changed, 334 insertions, 121 deletions

diff --git a/weed/shell/command_ec_balance.go b/weed/shell/command_ec_balance.go
index 15750b61f..2f292f11d 100644
--- a/weed/shell/command_ec_balance.go
+++ b/weed/shell/command_ec_balance.go
@@ -5,9 +5,8 @@ import (
 	"flag"
 	"fmt"
 	"io"
-	"math"
+	"sort"
 
-	"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
 	"github.com/chrislusf/seaweedfs/weed/storage/erasure_coding"
 	"github.com/chrislusf/seaweedfs/weed/storage/needle"
 )
@@ -24,27 +23,72 @@ func (c *commandEcBalance) Name() string {
 }
 
 func (c *commandEcBalance) Help() string {
-	return `balance all ec shards among volume servers
+	return `balance all ec shards among all racks and volume servers
 
 	ec.balance [-c EACH_COLLECTION|<collection_name>] [-force] [-dataCenter <data_center>]
 
 	Algorithm:
 
 	For each type of volume server (different max volume count limit){
-		for each collection {
-			balanceEcVolumes()
+		for each collection:
+			balanceEcVolumes(collectionName)
+		for each rack:
+			balanceEcRack(rack)
+	}
+
+	func balanceEcVolumes(collectionName){
+		for each volume:
+			doDeduplicateEcShards(volumeId)
+
+		tracks rack~shardCount mapping
+		for each volume:
+			doBalanceEcShardsAcrossRacks(volumeId)
+
+		for each volume:
+			doBalanceEcShardsWithinRacks(volumeId)
+	}
+
+	// spread ec shards into more racks
+	func doBalanceEcShardsAcrossRacks(volumeId){
+		tracks rack~volumeIdShardCount mapping
+		averageShardsPerEcRack = totalShardNumber / numRacks  // totalShardNumber is 14 for now, later could varies for each dc
+		ecShardsToMove = select overflown ec shards from racks with ec shard counts > averageShardsPerEcRack
+		for each ecShardsToMove {
+			destRack = pickOneRack(rack~shardCount, rack~volumeIdShardCount, averageShardsPerEcRack)
+			destVolumeServers = volume servers on the destRack
+			pickOneEcNodeAndMoveOneShard(destVolumeServers)
+		}
+	}
+
+	func doBalanceEcShardsWithinRacks(volumeId){
+		racks = collect all racks that the volume id is on
+		for rack, shards := range racks
+			doBalanceEcShardsWithinOneRack(volumeId, shards, rack)
+	}
+
+	// move ec shards 
+	func doBalanceEcShardsWithinOneRack(volumeId, shards, rackId){
+		tracks volumeServer~volumeIdShardCount mapping
+		averageShardCount = len(shards) / numVolumeServers
+		volumeServersOverAverage = volume servers with volumeId's ec shard counts > averageShardsPerEcRack
+		ecShardsToMove = select overflown ec shards from volumeServersOverAverage
+		for each ecShardsToMove {
+			destVolumeServer = pickOneVolumeServer(volumeServer~shardCount, volumeServer~volumeIdShardCount, averageShardCount)
+			pickOneEcNodeAndMoveOneShard(destVolumeServers)
 		}
 	}
 
-	func balanceEcVolumes(){
-		idealWritableVolumes = totalWritableVolumes / numVolumeServers
-		for {
-			sort all volume servers ordered by the number of local writable volumes
-			pick the volume server A with the lowest number of writable volumes x
-			pick the volume server B with the highest number of writable volumes y
-			if y > idealWritableVolumes and x +1 <= idealWritableVolumes {
-				if B has a writable volume id v that A does not have {
-					move writable volume v from A to B
+	// move ec shards while keeping shard distribution for the same volume unchanged or more even
+	func balanceEcRack(rack){
+		averageShardCount = total shards / numVolumeServers
+		for hasMovedOneEcShard {
+			sort all volume servers ordered by the number of local ec shards
+			pick the volume server A with the lowest number of ec shards x
+			pick the volume server B with the highest number of ec shards y
+			if y > averageShardCount and x +1 <= averageShardCount {
+				if B has a ec shard with volume id v that A does not have {
+					move one ec shard v from B to A
+					hasMovedOneEcShard = true
 				}
 			}
 		}
@@ -63,112 +107,86 @@ func (c *commandEcBalance) Do(args []string, commandEnv *CommandEnv, writer io.W
 		return nil
 	}
 
-	var resp *master_pb.VolumeListResponse
 	ctx := context.Background()
-	err = commandEnv.MasterClient.WithClient(ctx, func(client master_pb.SeaweedClient) error {
-		resp, err = client.VolumeList(ctx, &master_pb.VolumeListRequest{})
-		return err
-	})
+
+	// collect all ec nodes
+	allEcNodes, totalFreeEcSlots, err := collectEcNodes(ctx, commandEnv, *dc)
 	if err != nil {
 		return err
 	}
+	if totalFreeEcSlots < 1 {
+		return fmt.Errorf("no free ec shard slots. only %d left", totalFreeEcSlots)
+	}
 
-	typeToNodes := collectVolumeServersByType(resp.TopologyInfo, *dc)
-	for _, volumeServers := range typeToNodes {
-
-		fmt.Printf("balanceEcVolumes servers %d\n", len(volumeServers))
+	racks := collectRacks(allEcNodes)
 
-		if len(volumeServers) < 2 {
-			continue
+	if *collection == "EACH_COLLECTION" {
+		collections, err := ListCollectionNames(commandEnv, false, true)
+		if err != nil {
+			return err
 		}
-
-		if *collection == "EACH_COLLECTION" {
-			collections, err := ListCollectionNames(commandEnv, false, true)
-			if err != nil {
-				return err
-			}
-			fmt.Printf("balanceEcVolumes collections %+v\n", len(collections))
-			for _, c := range collections {
-				fmt.Printf("balanceEcVolumes collection %+v\n", c)
-				if err = balanceEcVolumes(commandEnv, c, *applyBalancing); err != nil {
-					return err
-				}
-			}
-		} else {
-			if err = balanceEcVolumes(commandEnv, *collection, *applyBalancing); err != nil {
+		fmt.Printf("balanceEcVolumes collections %+v\n", len(collections))
+		for _, c := range collections {
+			fmt.Printf("balanceEcVolumes collection %+v\n", c)
+			if err = balanceEcVolumes(commandEnv, c, allEcNodes, racks, *applyBalancing); err != nil {
 				return err
 			}
 		}
+	} else {
+		if err = balanceEcVolumes(commandEnv, *collection, allEcNodes, racks, *applyBalancing); err != nil {
+			return err
+		}
+	}
 
+	if err := balanceEcRacks(ctx, commandEnv, racks, *applyBalancing); err != nil {
+		return fmt.Errorf("balance ec racks: %v", err)
 	}
+
 	return nil
 }
 
-func balanceEcVolumes(commandEnv *CommandEnv, collection string, applyBalancing bool) error {
+func collectRacks(allEcNodes []*EcNode) map[RackId]*EcRack {
+	// collect racks info
+	racks := make(map[RackId]*EcRack)
+	for _, ecNode := range allEcNodes {
+		if racks[ecNode.rack] == nil {
+			racks[ecNode.rack] = &EcRack{
+				ecNodes: make(map[EcNodeId]*EcNode),
+			}
+		}
+		racks[ecNode.rack].ecNodes[EcNodeId(ecNode.info.Id)] = ecNode
+		racks[ecNode.rack].freeEcSlot += ecNode.freeEcSlot
+	}
+	return racks
+}
+
+func balanceEcVolumes(commandEnv *CommandEnv, collection string, allEcNodes []*EcNode, racks map[RackId]*EcRack, applyBalancing bool) error {
 
 	ctx := context.Background()
 
 	fmt.Printf("balanceEcVolumes %s\n", collection)
 
-	// collect all ec nodes
-	allEcNodes, totalFreeEcSlots, err := collectEcNodes(ctx, commandEnv)
-	if err != nil {
-		return err
-	}
-	if totalFreeEcSlots < 1 {
-		return fmt.Errorf("no free ec shard slots. only %d left", totalFreeEcSlots)
+	if err := deleteDuplicatedEcShards(ctx, commandEnv, allEcNodes, collection, applyBalancing); err != nil {
+		return fmt.Errorf("delete duplicated collection %s ec shards: %v", collection, err)
 	}
 
-	// vid => []ecNode
-	vidLocations := make(map[needle.VolumeId][]*EcNode)
-	for _, ecNode := range allEcNodes {
-		for _, shardInfo := range ecNode.info.EcShardInfos {
-			vidLocations[needle.VolumeId(shardInfo.Id)] = append(vidLocations[needle.VolumeId(shardInfo.Id)], ecNode)
-		}
+	if err := balanceEcShardsAcrossRacks(ctx, commandEnv, allEcNodes, racks, collection, applyBalancing); err != nil {
+		return fmt.Errorf("balance across racks collection %s ec shards: %v", collection, err)
 	}
 
-	for vid, locations := range vidLocations {
-
-		if err := doDeduplicateEcShards(ctx, commandEnv, collection, vid, locations, applyBalancing); err != nil {
-			return err
-		}
-
-		if err := doBalanceEcShards(ctx, commandEnv, collection, vid, locations, allEcNodes, applyBalancing); err != nil {
-			return err
-		}
-
+	if err := balanceEcShardsWithinRacks(ctx, commandEnv, allEcNodes, racks, collection, applyBalancing); err != nil {
+		return fmt.Errorf("balance across racks collection %s ec shards: %v", collection, err)
 	}
 
 	return nil
 }
 
-func doBalanceEcShards(ctx context.Context, commandEnv *CommandEnv, collection string, vid needle.VolumeId, locations []*EcNode, allEcNodes []*EcNode, applyBalancing bool) error {
-	// collect all ec nodes with at least one free slot
-	var possibleDestinationEcNodes []*EcNode
-	possibleDataCenters := make(map[string]int)
-	possibleRacks := make(map[string]int)
-	for _, ecNode := range allEcNodes {
-		if ecNode.freeEcSlot > 0 {
-			possibleDestinationEcNodes = append(possibleDestinationEcNodes, ecNode)
-			possibleDataCenters[ecNode.dc] += ecNode.freeEcSlot
-			possibleRacks[ecNode.dc+"/"+ecNode.rack] += ecNode.freeEcSlot
-		}
-	}
-	// calculate average number of shards an ec node should have for one volume
-	averageShardsPerEcNode := int(math.Ceil(float64(erasure_coding.TotalShardsCount) / float64(len(possibleDestinationEcNodes))))
-	fmt.Printf("vid %d averageShards Per EcNode:%d\n", vid, averageShardsPerEcNode)
-	// check whether this volume has ecNodes that are over average
-	isOverPerNodeAverage := false
-	for _, ecNode := range locations {
-		shardBits := findEcVolumeShards(ecNode, vid)
-		if shardBits.ShardIdCount() > averageShardsPerEcNode {
-			isOverPerNodeAverage = true
-			fmt.Printf("vid %d %s has %d shards, isOverPerNodeAverage %+v\n", vid, ecNode.info.Id, shardBits.ShardIdCount(), isOverPerNodeAverage)
-			break
-		}
-	}
-	if isOverPerNodeAverage {
-		if err := spreadShardsIntoMoreDataNodes(ctx, commandEnv, averageShardsPerEcNode, collection, vid, locations, possibleDestinationEcNodes, applyBalancing); err != nil {
+func deleteDuplicatedEcShards(ctx context.Context, commandEnv *CommandEnv, allEcNodes []*EcNode, collection string, applyBalancing bool) error {
+	// vid => []ecNode
+	vidLocations := collectVolumeIdToEcNodes(allEcNodes)
+	// deduplicate ec shards
+	for vid, locations := range vidLocations {
+		if err := doDeduplicateEcShards(ctx, commandEnv, collection, vid, locations, applyBalancing); err != nil {
 			return err
 		}
 	}
@@ -203,14 +221,120 @@ func doDeduplicateEcShards(ctx context.Context, commandEnv *CommandEnv, collecti
 			if err := sourceServerDeleteEcShards(ctx, commandEnv.option.GrpcDialOption, collection, vid, ecNode.info.Id, duplicatedShardIds); err != nil {
 				return err
 			}
-			deleteEcVolumeShards(ecNode, vid, duplicatedShardIds)
-			ecNode.freeEcSlot++
+			ecNode.deleteEcVolumeShards(vid, duplicatedShardIds)
 		}
 	}
 	return nil
 }
 
-func spreadShardsIntoMoreDataNodes(ctx context.Context, commandEnv *CommandEnv, averageShardsPerEcNode int, collection string, vid needle.VolumeId, existingLocations, possibleDestinationEcNodes []*EcNode, applyBalancing bool) error {
+func balanceEcShardsAcrossRacks(ctx context.Context, commandEnv *CommandEnv, allEcNodes []*EcNode, racks map[RackId]*EcRack, collection string, applyBalancing bool) error {
+	// collect vid => []ecNode, since previous steps can change the locations
+	vidLocations := collectVolumeIdToEcNodes(allEcNodes)
+	// spread the ec shards evenly
+	for vid, locations := range vidLocations {
+		if err := doBalanceEcShardsAcrossRacks(ctx, commandEnv, collection, vid, locations, racks, applyBalancing); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func doBalanceEcShardsAcrossRacks(ctx context.Context, commandEnv *CommandEnv, collection string, vid needle.VolumeId, locations []*EcNode, racks map[RackId]*EcRack, applyBalancing bool) error {
+
+	// calculate average number of shards an ec rack should have for one volume
+	averageShardsPerEcRack := ceilDivide(erasure_coding.TotalShardsCount, len(racks))
+
+	// see the volume's shards are in how many racks, and how many in each rack
+	rackToShardCount := groupByCount(locations, func(ecNode *EcNode) (id string, count int) {
+		shardBits := findEcVolumeShards(ecNode, vid)
+		return string(ecNode.rack), shardBits.ShardIdCount()
+	})
+	rackEcNodesWithVid := groupBy(locations, func(ecNode *EcNode) string {
+		return string(ecNode.rack)
+	})
+
+	// ecShardsToMove = select overflown ec shards from racks with ec shard counts > averageShardsPerEcRack
+	ecShardsToMove := make(map[erasure_coding.ShardId]*EcNode)
+	for rackId, count := range rackToShardCount {
+		if count > averageShardsPerEcRack {
+			possibleEcNodes := rackEcNodesWithVid[rackId]
+			for shardId, ecNode := range pickNEcShardsToMoveFrom(possibleEcNodes, vid, count-averageShardsPerEcRack) {
+				ecShardsToMove[shardId] = ecNode
+			}
+		}
+	}
+
+	for shardId, ecNode := range ecShardsToMove {
+		rackId := pickOneRack(racks, rackToShardCount, averageShardsPerEcRack)
+		var possibleDestinationEcNodes []*EcNode
+		for _, n := range racks[rackId].ecNodes {
+			possibleDestinationEcNodes = append(possibleDestinationEcNodes, n)
+		}
+		err := pickOneEcNodeAndMoveOneShard(ctx, commandEnv, averageShardsPerEcRack, ecNode, collection, vid, shardId, possibleDestinationEcNodes, applyBalancing)
+		if err != nil {
+			return err
+		}
+		rackToShardCount[string(rackId)] += 1
+		rackToShardCount[string(ecNode.rack)] -= 1
+		racks[rackId].freeEcSlot -= 1
+		racks[ecNode.rack].freeEcSlot += 1
+	}
+
+	return nil
+}
+
+func pickOneRack(rackToEcNodes map[RackId]*EcRack, rackToShardCount map[string]int, averageShardsPerEcRack int) RackId {
+
+	// TODO later may need to add some randomness
+
+	for rackId, rack := range rackToEcNodes {
+		if rackToShardCount[string(rackId)] >= averageShardsPerEcRack {
+			continue
+		}
+
+		if rack.freeEcSlot <= 0 {
+			continue
+		}
+
+		return rackId
+	}
+
+	return ""
+}
+
+func balanceEcShardsWithinRacks(ctx context.Context, commandEnv *CommandEnv, allEcNodes []*EcNode, racks map[RackId]*EcRack, collection string, applyBalancing bool) error {
+	// collect vid => []ecNode, since previous steps can change the locations
+	vidLocations := collectVolumeIdToEcNodes(allEcNodes)
+
+	// spread the ec shards evenly
+	for vid, locations := range vidLocations {
+
+		// see the volume's shards are in how many racks, and how many in each rack
+		rackToShardCount := groupByCount(locations, func(ecNode *EcNode) (id string, count int) {
+			shardBits := findEcVolumeShards(ecNode, vid)
+			return string(ecNode.rack), shardBits.ShardIdCount()
+		})
+		rackEcNodesWithVid := groupBy(locations, func(ecNode *EcNode) string {
+			return string(ecNode.rack)
+		})
+
+		for rackId, _ := range rackToShardCount {
+
+			var possibleDestinationEcNodes []*EcNode
+			for _, n := range racks[RackId(rackId)].ecNodes {
+				possibleDestinationEcNodes = append(possibleDestinationEcNodes, n)
+			}
+			sourceEcNodes := rackEcNodesWithVid[rackId]
+			averageShardsPerEcNode := ceilDivide(rackToShardCount[rackId], len(possibleDestinationEcNodes))
+			if err := doBalanceEcShardsWithinOneRack(ctx, commandEnv, averageShardsPerEcNode, collection, vid, sourceEcNodes, possibleDestinationEcNodes, applyBalancing); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
+func doBalanceEcShardsWithinOneRack(ctx context.Context, commandEnv *CommandEnv, averageShardsPerEcNode int, collection string, vid needle.VolumeId, existingLocations, possibleDestinationEcNodes []*EcNode, applyBalancing bool) error {
 
 	for _, ecNode := range existingLocations {
 
@@ -237,9 +361,85 @@ func spreadShardsIntoMoreDataNodes(ctx context.Context, commandEnv *CommandEnv,
 	return nil
 }
 
-func pickOneEcNodeAndMoveOneShard(ctx context.Context, commandEnv *CommandEnv, averageShardsPerEcNode int, existingLocation *EcNode, collection string, vid needle.VolumeId, shardId erasure_coding.ShardId, possibleDestinationEcNodes []*EcNode, applyBalancing bool) error {
+func balanceEcRacks(ctx context.Context, commandEnv *CommandEnv, racks map[RackId]*EcRack, applyBalancing bool) error {
+
+	// balance one rack for all ec shards
+	for _, ecRack := range racks {
+		if err := doBalanceEcRack(ctx, commandEnv, ecRack, applyBalancing); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func doBalanceEcRack(ctx context.Context, commandEnv *CommandEnv, ecRack *EcRack, applyBalancing bool) error {
+
+	if len(ecRack.ecNodes) <= 1 {
+		return nil
+	}
+
+	var rackEcNodes []*EcNode
+	for _, node := range ecRack.ecNodes {
+		rackEcNodes = append(rackEcNodes, node)
+	}
+
+	ecNodeIdToShardCount := groupByCount(rackEcNodes, func(node *EcNode) (id string, count int) {
+		for _, ecShardInfo := range node.info.EcShardInfos {
+			count += erasure_coding.ShardBits(ecShardInfo.EcIndexBits).ShardIdCount()
+		}
+		return node.info.Id, count
+	})
+
+	var totalShardCount int
+	for _, count := range ecNodeIdToShardCount {
+		totalShardCount += count
+	}
+
+	averageShardCount := ceilDivide(totalShardCount, len(rackEcNodes))
+
+	hasMove := true
+	for hasMove {
+		hasMove = false
+		sort.Slice(rackEcNodes, func(i, j int) bool {
+			return rackEcNodes[i].freeEcSlot > rackEcNodes[j].freeEcSlot
+		})
+		emptyNode, fullNode := rackEcNodes[0], rackEcNodes[len(rackEcNodes)-1]
+		emptyNodeShardCount, fullNodeShardCount := ecNodeIdToShardCount[emptyNode.info.Id], ecNodeIdToShardCount[fullNode.info.Id]
+		if fullNodeShardCount > averageShardCount && emptyNodeShardCount+1 <= averageShardCount {
+
+			emptyNodeIds := make(map[uint32]bool)
+			for _, shards := range emptyNode.info.EcShardInfos {
+				emptyNodeIds[shards.Id] = true
+			}
+			for _, shards := range fullNode.info.EcShardInfos {
+				if _, found := emptyNodeIds[shards.Id]; !found {
+					for _, shardId := range erasure_coding.ShardBits(shards.EcIndexBits).ShardIds() {
+
+						fmt.Printf("%s moves ec shards %d.%d to %s\n", fullNode.info.Id, shards.Id, shardId, emptyNode.info.Id)
+
+						err := moveMountedShardToEcNode(ctx, commandEnv, fullNode, shards.Collection, needle.VolumeId(shards.Id), shardId, emptyNode, applyBalancing)
+						if err != nil {
+							return err
+						}
+
+						ecNodeIdToShardCount[emptyNode.info.Id]++
+						ecNodeIdToShardCount[fullNode.info.Id]--
+						hasMove = true
+						break
+					}
+					break
+				}
+			}
+		}
+	}
+
+	return nil
+}
+
+func pickOneEcNodeAndMoveOneShard(ctx context.Context, commandEnv *CommandEnv, expectedTotalEcShards int, existingLocation *EcNode, collection string, vid needle.VolumeId, shardId erasure_coding.ShardId, possibleDestinationEcNodes []*EcNode, applyBalancing bool) error {
 
 	sortEcNodes(possibleDestinationEcNodes)
+	averageShardsPerEcNode := ceilDivide(expectedTotalEcShards, len(possibleDestinationEcNodes))
 
 	for _, destEcNode := range possibleDestinationEcNodes {
 		if destEcNode.info.Id == existingLocation.info.Id {
@@ -260,45 +460,58 @@ func pickOneEcNodeAndMoveOneShard(ctx context.Context, commandEnv *CommandEnv, a
 			return err
 		}
 
-		destEcNode.freeEcSlot--
-		existingLocation.freeEcSlot++
 		return nil
 	}
 
 	return nil
 }
 
-func findEcVolumeShards(ecNode *EcNode, vid needle.VolumeId) erasure_coding.ShardBits {
-
-	for _, shardInfo := range ecNode.info.EcShardInfos {
-		if needle.VolumeId(shardInfo.Id) == vid {
-			return erasure_coding.ShardBits(shardInfo.EcIndexBits)
+func pickNEcShardsToMoveFrom(ecNodes []*EcNode, vid needle.VolumeId, n int) (map[erasure_coding.ShardId]*EcNode) {
+	picked := make(map[erasure_coding.ShardId]*EcNode)
+	var candidateEcNodes []*CandidateEcNode
+	for _, ecNode := range ecNodes {
+		shardBits := findEcVolumeShards(ecNode, vid)
+		if shardBits.ShardIdCount() > 0 {
+			candidateEcNodes = append(candidateEcNodes, &CandidateEcNode{
+				ecNode:     ecNode,
+				shardCount: shardBits.ShardIdCount(),
+			})
 		}
 	}
-
-	return 0
-}
-
-func addEcVolumeShards(ecNode *EcNode, vid needle.VolumeId, shardIds []uint32) {
-
-	for _, shardInfo := range ecNode.info.EcShardInfos {
-		if needle.VolumeId(shardInfo.Id) == vid {
-			for _, shardId := range shardIds {
-				shardInfo.EcIndexBits = uint32(erasure_coding.ShardBits(shardInfo.EcIndexBits).AddShardId(erasure_coding.ShardId(shardId)))
+	sort.Slice(candidateEcNodes, func(i, j int) bool {
+		return candidateEcNodes[i].shardCount > candidateEcNodes[j].shardCount
+	})
+	for i := 0; i < n; i++ {
+		selectedEcNodeIndex := -1
+		for i, candidateEcNode := range candidateEcNodes {
+			shardBits := findEcVolumeShards(candidateEcNode.ecNode, vid)
+			if shardBits > 0 {
+				selectedEcNodeIndex = i
+				for _, shardId := range shardBits.ShardIds() {
+					candidateEcNode.shardCount--
+					picked[shardId] = candidateEcNode.ecNode
+					candidateEcNode.ecNode.deleteEcVolumeShards(vid, []uint32{uint32(shardId)})
+					break
+				}
+				break
 			}
 		}
-	}
+		if selectedEcNodeIndex >= 0 {
+			ensureSortedEcNodes(candidateEcNodes, selectedEcNodeIndex, func(i, j int) bool {
+				return candidateEcNodes[i].shardCount > candidateEcNodes[j].shardCount
+			})
+		}
 
+	}
+	return picked
 }
 
-func deleteEcVolumeShards(ecNode *EcNode, vid needle.VolumeId, shardIds []uint32) {
-
-	for _, shardInfo := range ecNode.info.EcShardInfos {
-		if needle.VolumeId(shardInfo.Id) == vid {
-			for _, shardId := range shardIds {
-				shardInfo.EcIndexBits = uint32(erasure_coding.ShardBits(shardInfo.EcIndexBits).RemoveShardId(erasure_coding.ShardId(shardId)))
-			}
+func collectVolumeIdToEcNodes(allEcNodes []*EcNode) map[needle.VolumeId][]*EcNode {
+	vidLocations := make(map[needle.VolumeId][]*EcNode)
+	for _, ecNode := range allEcNodes {
+		for _, shardInfo := range ecNode.info.EcShardInfos {
+			vidLocations[needle.VolumeId(shardInfo.Id)] = append(vidLocations[needle.VolumeId(shardInfo.Id)], ecNode)
 		}
 	}
-
+	return vidLocations
 }