diff options
| -rw-r--r-- | weed/topology/data_node.go | 16 | ||||
| -rw-r--r-- | weed/topology/disk.go | 11 |
2 files changed, 19 insertions, 8 deletions
diff --git a/weed/topology/data_node.go b/weed/topology/data_node.go index 715b5ed80..de428d38f 100644 --- a/weed/topology/data_node.go +++ b/weed/topology/data_node.go @@ -2,6 +2,8 @@ package topology import ( "fmt" + "sync/atomic" + "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" @@ -9,7 +11,6 @@ import ( "github.com/seaweedfs/seaweedfs/weed/storage/needle" "github.com/seaweedfs/seaweedfs/weed/storage/types" "github.com/seaweedfs/seaweedfs/weed/util" - "sync/atomic" ) type DataNode struct { @@ -79,7 +80,7 @@ func (dn *DataNode) UpdateVolumes(actualVolumes []storage.VolumeInfo) (newVolume if _, ok := actualVolumeMap[vid]; !ok { glog.V(0).Infoln("Deleting volume id:", vid) disk := dn.getOrCreateDisk(v.DiskType) - delete(disk.volumes, vid) + disk.DeleteVolumeById(vid) deletedVolumes = append(deletedVolumes, v) deltaDiskUsages := newDiskUsages() @@ -112,10 +113,12 @@ func (dn *DataNode) DeltaUpdateVolumes(newVolumes, deletedVolumes []storage.Volu for _, v := range deletedVolumes { disk := dn.getOrCreateDisk(v.DiskType) - if _, found := disk.volumes[v.Id]; !found { + + _, err := disk.GetVolumesById(v.Id) + if err != nil { continue } - delete(disk.volumes, v.Id) + disk.DeleteVolumeById(v.Id) deltaDiskUsages := newDiskUsages() deltaDiskUsage := deltaDiskUsages.getOrCreateDisk(types.ToDiskType(v.DiskType)) @@ -170,8 +173,9 @@ func (dn *DataNode) GetVolumesById(id needle.VolumeId) (vInfo storage.VolumeInfo found := false for _, c := range dn.children { disk := c.(*Disk) - vInfo, found = disk.volumes[id] - if found { + vInfo, err = disk.GetVolumesById(id) + if err == nil { + found = true break } } diff --git a/weed/topology/disk.go b/weed/topology/disk.go index f4adc0e72..4597bfc29 100644 --- a/weed/topology/disk.go +++ b/weed/topology/disk.go @@ -2,11 +2,12 @@ package topology import ( "fmt" - "github.com/seaweedfs/seaweedfs/weed/storage/types" - "github.com/seaweedfs/seaweedfs/weed/util" "sync" "sync/atomic" + "github.com/seaweedfs/seaweedfs/weed/storage/types" + "github.com/seaweedfs/seaweedfs/weed/util" + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding" "github.com/seaweedfs/seaweedfs/weed/storage/needle" @@ -201,6 +202,12 @@ func (d *Disk) GetVolumesById(id needle.VolumeId) (storage.VolumeInfo, error) { } } +func (d *Disk) DeleteVolumeById(id needle.VolumeId) { + d.Lock() + defer d.Unlock() + delete(d.volumes, id) +} + func (d *Disk) GetDataCenter() *DataCenter { dn := d.Parent() rack := dn.Parent() |