also remove volumeLayout if collection is empty

weed/topology/configuration.go

@@ -20,8 +20,8 @@ type topology struct {
 	DataCenters []dataCenter `xml:"DataCenter"`
 }
 type Configuration struct {
-	XMLName     xml.Name `xml:"Configuration"`
-	Topo        topology `xml:"Topology"`
+	XMLName     xml.Name       `xml:"Configuration"`
+	Topo        topology       `xml:"Topology"`
 	ip2location map[string]loc // this is not used any more. leave it here for later.
 }
 

weed/topology/topology.go

@@ -6,10 +6,10 @@ import (
 
 	"github.com/chrislusf/raft"
 	"github.com/chrislusf/seaweedfs/weed/glog"
+	"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
 	"github.com/chrislusf/seaweedfs/weed/sequence"
 	"github.com/chrislusf/seaweedfs/weed/storage"
 	"github.com/chrislusf/seaweedfs/weed/util"
-	"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
 )
 
 type Topology struct {
@@ -132,7 +132,11 @@ func (t *Topology) RegisterVolumeLayout(v storage.VolumeInfo, dn *DataNode) {
 }
 func (t *Topology) UnRegisterVolumeLayout(v storage.VolumeInfo, dn *DataNode) {
 	glog.Infof("removing volume info:%+v", v)
-	t.GetVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl).UnRegisterVolume(&v, dn)
+	volumeLayout := t.GetVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl)
+	volumeLayout.UnRegisterVolume(&v, dn)
+	if volumeLayout.isEmpty() {
+		t.DeleteCollection(v.Collection)
+	}
 }
 
 func (t *Topology) GetOrCreateDataCenter(dcName string) *DataCenter {

weed/topology/topology_test.go

@@ -1,9 +1,12 @@
 package topology
 
 import (
-	"testing"
-	"github.com/chrislusf/seaweedfs/weed/sequence"
+	"fmt"
 	"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
+	"github.com/chrislusf/seaweedfs/weed/sequence"
+	"github.com/chrislusf/seaweedfs/weed/storage"
+	"github.com/kr/pretty"
+	"testing"
 )
 
 func TestRemoveDataCenter(t *testing.T) {
@@ -26,7 +29,7 @@ func TestHandlingVolumeServerHeartbeat(t *testing.T) {
 	dn := rack.GetOrCreateDataNode("127.0.0.1", 34534, "127.0.0.1", 25)
 
 	{
-		volumeCount := 700
+		volumeCount := 7
 		var volumeMessages []*master_pb.VolumeInformationMessage
 		for k := 1; k <= volumeCount; k++ {
 			volumeMessage := &master_pb.VolumeInformationMessage{
@@ -51,7 +54,7 @@ func TestHandlingVolumeServerHeartbeat(t *testing.T) {
 	}
 
 	{
-		volumeCount := 700 - 1
+		volumeCount := 7 - 1
 		var volumeMessages []*master_pb.VolumeInformationMessage
 		for k := 1; k <= volumeCount; k++ {
 			volumeMessage := &master_pb.VolumeInformationMessage{
@@ -85,3 +88,35 @@ func assert(t *testing.T, message string, actual, expected int) {
 		t.Fatalf("unexpected %s: %d, expected: %d", message, actual, expected)
 	}
 }
+
+func TestAddRemoveVolume(t *testing.T) {
+
+	topo := NewTopology("weedfs", sequence.NewMemorySequencer(), 32*1024, 5)
+
+	dc := topo.GetOrCreateDataCenter("dc1")
+	rack := dc.GetOrCreateRack("rack1")
+	dn := rack.GetOrCreateDataNode("127.0.0.1", 34534, "127.0.0.1", 25)
+
+	v := storage.VolumeInfo{
+		Id:               storage.VolumeId(1),
+		Size:             100,
+		Collection:       "xcollection",
+		FileCount:        123,
+		DeleteCount:      23,
+		DeletedByteCount: 45,
+		ReadOnly:         false,
+		Version:          storage.CurrentVersion,
+		ReplicaPlacement: &storage.ReplicaPlacement{},
+		Ttl:              storage.EMPTY_TTL,
+	}
+
+	dn.UpdateVolumes([]storage.VolumeInfo{v})
+	topo.RegisterVolumeLayout(v, dn)
+
+	fmt.Printf("added volume 1 \n%# v\n", pretty.Formatter(topo.ToMap()))
+
+	topo.UnRegisterVolumeLayout(v, dn)
+
+	fmt.Printf("removed volume 1 \n%# v\n", pretty.Formatter(topo.ToMap()))
+
+}

weed/topology/volume_layout.go

@@ -107,6 +107,13 @@ func (vl *VolumeLayout) isWritable(v *storage.VolumeInfo) bool {
 		!v.ReadOnly
 }
 
+func (vl *VolumeLayout) isEmpty() bool {
+	vl.accessLock.RLock()
+	defer vl.accessLock.RUnlock()
+
+	return len(vl.vid2location) == 0
+}
+
 func (vl *VolumeLayout) Lookup(vid storage.VolumeId) []*DataNode {
 	vl.accessLock.RLock()
 	defer vl.accessLock.RUnlock()