diff --git a/weed/storage/erasure_coding/ec_volume_info.go b/weed/storage/erasure_coding/ec_volume_info.go index 9cebb7d40..645686a92 100644 --- a/weed/storage/erasure_coding/ec_volume_info.go +++ b/weed/storage/erasure_coding/ec_volume_info.go @@ -40,6 +40,16 @@ func (ecInfo *EcVolumeInfo) ShardIds() (ret []ShardId) { return } +func (ecInfo *EcVolumeInfo) Minus(other *EcVolumeInfo) (*EcVolumeInfo) { + ret := &EcVolumeInfo{ + VolumeId: ecInfo.VolumeId, + Collection: ecInfo.Collection, + shardIds: ecInfo.shardIds &^ other.shardIds, + } + + return ret +} + func (ecInfo *EcVolumeInfo) ToVolumeEcShardInformationMessage() (ret []*master_pb.VolumeEcShardInformationMessage) { for _, shard := range ecInfo.ShardIds() { ret = append(ret, &master_pb.VolumeEcShardInformationMessage{ diff --git a/weed/topology/data_node_ec.go b/weed/topology/data_node_ec.go index 5206a3b51..ef395dbdb 100644 --- a/weed/topology/data_node_ec.go +++ b/weed/topology/data_node_ec.go @@ -2,6 +2,7 @@ package topology import ( "github.com/chrislusf/seaweedfs/weed/storage/erasure_coding" + "github.com/chrislusf/seaweedfs/weed/storage/needle" ) func (dn *DataNode) GetEcShards() (ret []*erasure_coding.EcVolumeInfo) { @@ -14,7 +15,43 @@ func (dn *DataNode) GetEcShards() (ret []*erasure_coding.EcVolumeInfo) { } func (dn *DataNode) UpdateEcShards(actualShards []*erasure_coding.EcVolumeInfo) (newShards, deletedShards []*erasure_coding.EcVolumeInfo) { - dn.ecShardsLock.Lock() - dn.ecShardsLock.Unlock() + // prepare the new ec shard map + actualEcShardMap := make(map[needle.VolumeId]*erasure_coding.EcVolumeInfo) + for _, ecShards := range actualShards { + actualEcShardMap[ecShards.VolumeId]= ecShards + } + + // found out the newShards and deletedShards + dn.ecShardsLock.RLock() + for vid, ecShards := range dn.ecShards{ + if actualEcShards, ok := actualEcShardMap[vid]; !ok { + // dn registered ec shards not found in the new set of ec shards + deletedShards = append(deletedShards, ecShards) + } else { + // found, but maybe the actual shard could be missing + a := actualEcShards.Minus(ecShards) + if len(a.ShardIds())>0 { + newShards = append(newShards, a) + } + d := ecShards.Minus(actualEcShards) + if len(d.ShardIds())>0 { + deletedShards = append(deletedShards, d) + } + } + } + for _, ecShards := range actualShards { + if _, found := dn.ecShards[ecShards.VolumeId]; !found { + newShards = append(newShards, ecShards) + } + } + dn.ecShardsLock.RUnlock() + + if len(newShards)>0 || len(deletedShards)>0{ + // if changed, set to the new ec shard map + dn.ecShardsLock.Lock() + dn.ecShards = actualEcShardMap + dn.ecShardsLock.Unlock() + } + return }