mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2024-01-19 02:48:24 +00:00
125 lines
3.5 KiB
Go
125 lines
3.5 KiB
Go
package topology
|
|
|
|
import (
|
|
"github.com/chrislusf/seaweedfs/weed/glog"
|
|
"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
|
|
"github.com/chrislusf/seaweedfs/weed/storage/erasure_coding"
|
|
"github.com/chrislusf/seaweedfs/weed/storage/needle"
|
|
)
|
|
|
|
type EcShardLocations struct {
|
|
Collection string
|
|
locations [erasure_coding.TotalShardsCount][]*DataNode
|
|
}
|
|
|
|
func (t *Topology) SyncDataNodeEcShards(shardInfos []*master_pb.VolumeEcShardInformationMessage, dn *DataNode) (newShards, deletedShards []*erasure_coding.EcVolumeInfo) {
|
|
// convert into in memory struct storage.VolumeInfo
|
|
var shards []*erasure_coding.EcVolumeInfo
|
|
for _, shardInfo := range shardInfos {
|
|
shards = append(shards,
|
|
erasure_coding.NewEcVolumeInfo(
|
|
shardInfo.Collection,
|
|
needle.VolumeId(shardInfo.Id),
|
|
erasure_coding.ShardBits(shardInfo.EcIndexBits)))
|
|
}
|
|
// find out the delta volumes
|
|
newShards, deletedShards = dn.UpdateEcShards(shards)
|
|
for _, v := range newShards {
|
|
t.RegisterEcShards(v, dn)
|
|
}
|
|
for _, v := range deletedShards {
|
|
t.UnRegisterEcShards(v, dn)
|
|
}
|
|
return
|
|
}
|
|
|
|
func (t *Topology) IncrementalSyncDataNodeEcShards(newEcShards, deletedEcShards []*master_pb.VolumeEcShardInformationMessage, dn *DataNode) {
|
|
// convert into in memory struct storage.VolumeInfo
|
|
var newShards, deletedShards []*erasure_coding.EcVolumeInfo
|
|
for _, shardInfo := range newEcShards {
|
|
newShards = append(newShards,
|
|
erasure_coding.NewEcVolumeInfo(
|
|
shardInfo.Collection,
|
|
needle.VolumeId(shardInfo.Id),
|
|
erasure_coding.ShardBits(shardInfo.EcIndexBits)))
|
|
}
|
|
for _, shardInfo := range deletedEcShards {
|
|
deletedShards = append(deletedShards,
|
|
erasure_coding.NewEcVolumeInfo(
|
|
shardInfo.Collection,
|
|
needle.VolumeId(shardInfo.Id),
|
|
erasure_coding.ShardBits(shardInfo.EcIndexBits)))
|
|
}
|
|
|
|
dn.DeltaUpdateEcShards(newShards, deletedShards)
|
|
|
|
for _, v := range newShards {
|
|
t.RegisterEcShards(v, dn)
|
|
}
|
|
for _, v := range deletedShards {
|
|
t.UnRegisterEcShards(v, dn)
|
|
}
|
|
return
|
|
}
|
|
|
|
func NewEcShardLocations(collection string) *EcShardLocations {
|
|
return &EcShardLocations{
|
|
Collection: collection,
|
|
}
|
|
}
|
|
|
|
func (loc *EcShardLocations) AddShard(shardId erasure_coding.ShardId, dn *DataNode) (added bool) {
|
|
dataNodes := loc.locations[shardId]
|
|
for _, n := range dataNodes {
|
|
if n.Id() == dn.Id() {
|
|
return false
|
|
}
|
|
}
|
|
loc.locations[shardId] = append(dataNodes, dn)
|
|
return true
|
|
}
|
|
|
|
func (loc *EcShardLocations) DeleteShard(shardId erasure_coding.ShardId, dn *DataNode) (deleted bool) {
|
|
dataNodes := loc.locations[shardId]
|
|
foundIndex := -1
|
|
for index, n := range dataNodes {
|
|
if n.Id() == dn.Id() {
|
|
foundIndex = index
|
|
}
|
|
}
|
|
if foundIndex < 0 {
|
|
return false
|
|
}
|
|
loc.locations[shardId] = append(dataNodes[:foundIndex], dataNodes[foundIndex+1:]...)
|
|
return true
|
|
}
|
|
|
|
func (t *Topology) RegisterEcShards(ecShardInfos *erasure_coding.EcVolumeInfo, dn *DataNode) {
|
|
|
|
t.ecShardMapLock.Lock()
|
|
defer t.ecShardMapLock.Unlock()
|
|
|
|
locations, found := t.ecShardMap[ecShardInfos.VolumeId]
|
|
if !found {
|
|
locations = NewEcShardLocations(ecShardInfos.Collection)
|
|
t.ecShardMap[ecShardInfos.VolumeId] = locations
|
|
}
|
|
for _, shardId := range ecShardInfos.ShardIds() {
|
|
locations.AddShard(shardId, dn)
|
|
}
|
|
}
|
|
|
|
func (t *Topology) UnRegisterEcShards(ecShardInfos *erasure_coding.EcVolumeInfo, dn *DataNode) {
|
|
glog.Infof("removing ec shard info:%+v", ecShardInfos)
|
|
t.ecShardMapLock.Lock()
|
|
defer t.ecShardMapLock.Unlock()
|
|
|
|
locations, found := t.ecShardMap[ecShardInfos.VolumeId]
|
|
if !found {
|
|
return
|
|
}
|
|
for _, shardId := range ecShardInfos.ShardIds() {
|
|
locations.DeleteShard(shardId, dn)
|
|
}
|
|
}
|