mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2024-01-19 02:48:24 +00:00
broadcast vid->location map to master clients for ec chards
This commit is contained in:
parent
af67d99ca4
commit
b58e25e588
|
@ -9,6 +9,7 @@ import (
|
||||||
"github.com/chrislusf/raft"
|
"github.com/chrislusf/raft"
|
||||||
"github.com/chrislusf/seaweedfs/weed/glog"
|
"github.com/chrislusf/seaweedfs/weed/glog"
|
||||||
"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
|
"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
|
||||||
|
"github.com/chrislusf/seaweedfs/weed/storage/needle"
|
||||||
"github.com/chrislusf/seaweedfs/weed/topology"
|
"github.com/chrislusf/seaweedfs/weed/topology"
|
||||||
"google.golang.org/grpc/peer"
|
"google.golang.org/grpc/peer"
|
||||||
)
|
)
|
||||||
|
@ -108,18 +109,32 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ
|
||||||
|
|
||||||
// update master internal volume layouts
|
// update master internal volume layouts
|
||||||
t.IncrementalSyncDataNodeEcShards(heartbeat.NewEcShards, heartbeat.DeletedEcShards, dn)
|
t.IncrementalSyncDataNodeEcShards(heartbeat.NewEcShards, heartbeat.DeletedEcShards, dn)
|
||||||
|
|
||||||
|
for _, s := range heartbeat.NewEcShards {
|
||||||
|
message.NewVids = append(message.NewVids, s.Id)
|
||||||
|
}
|
||||||
|
for _, s := range heartbeat.DeletedEcShards {
|
||||||
|
if dn.HasVolumesById(needle.VolumeId(s.Id)) {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
message.DeletedVids = append(message.DeletedVids, s.Id)
|
||||||
|
}
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
if len(heartbeat.EcShards) > 0 {
|
if len(heartbeat.EcShards) > 0 {
|
||||||
glog.V(0).Infof("master recieved ec shards from %s: %+v", dn.Url(), heartbeat.EcShards)
|
glog.V(0).Infof("master recieved ec shards from %s: %+v", dn.Url(), heartbeat.EcShards)
|
||||||
newShards, deletedShards := t.SyncDataNodeEcShards(heartbeat.EcShards, dn)
|
newShards, deletedShards := t.SyncDataNodeEcShards(heartbeat.EcShards, dn)
|
||||||
|
|
||||||
//TODO broadcast the ec vid
|
// broadcast the ec vid changes to master clients
|
||||||
if len(newShards)>0{
|
for _, s := range newShards {
|
||||||
|
message.NewVids = append(message.NewVids, uint32(s.VolumeId))
|
||||||
}
|
}
|
||||||
if len(deletedShards)>0{
|
for _, s := range deletedShards {
|
||||||
|
if dn.HasVolumesById(s.VolumeId) {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
message.DeletedVids = append(message.DeletedVids, uint32(s.VolumeId))
|
||||||
}
|
}
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
|
@ -92,3 +92,29 @@ func (dn *DataNode) DeleteEcShard(s *erasure_coding.EcVolumeInfo) {
|
||||||
}
|
}
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (dn *DataNode) HasVolumesById(id needle.VolumeId) (hasVolumeId bool) {
|
||||||
|
|
||||||
|
// check whether normal volumes has this volume id
|
||||||
|
dn.RLock()
|
||||||
|
_, ok := dn.volumes[id]
|
||||||
|
if ok {
|
||||||
|
hasVolumeId = true
|
||||||
|
}
|
||||||
|
dn.RUnlock()
|
||||||
|
|
||||||
|
if hasVolumeId {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
// check whether ec shards has this volume id
|
||||||
|
dn.ecShardsLock.RLock()
|
||||||
|
_, ok = dn.ecShards[id]
|
||||||
|
if ok {
|
||||||
|
hasVolumeId = true
|
||||||
|
}
|
||||||
|
dn.ecShardsLock.RUnlock()
|
||||||
|
|
||||||
|
return
|
||||||
|
|
||||||
|
}
|
||||||
|
|
Loading…
Reference in a new issue