[master] fix master panic (seaweedfs#5893)

weed/topology/data_node.go

@@ -2,14 +2,15 @@ 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"
 	"github.com/seaweedfs/seaweedfs/weed/storage"
 	"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
 		}
 	}

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()