From 133b772fb5c8df6b4455c6f457800d4baa309518 Mon Sep 17 00:00:00 2001
From: Chris Lu <chris.lu@gmail.com>
Date: Sat, 1 Jun 2019 01:51:28 -0700
Subject: [PATCH] destroy ec volume if it is empty

---
 weed/server/volume_grpc_erasure_coding.go |  2 +-
 weed/storage/disk_location_ec.go          | 11 +++++++++++
 weed/storage/store_ec.go                  |  6 ++++++
 3 files changed, 18 insertions(+), 1 deletion(-)

diff --git a/weed/server/volume_grpc_erasure_coding.go b/weed/server/volume_grpc_erasure_coding.go
index c0d666b6d..a2ba10323 100644
--- a/weed/server/volume_grpc_erasure_coding.go
+++ b/weed/server/volume_grpc_erasure_coding.go
@@ -142,7 +142,7 @@ func (vs *VolumeServer) doDeleteMountedShards(ctx context.Context, req *volume_s
 	}
 
 	if len(ecv.Shards) == 0 {
-		ecv.Destroy()
+		vs.store.DestroyEcVolume(needle.VolumeId(req.VolumeId))
 	}
 
 	return nil
diff --git a/weed/storage/disk_location_ec.go b/weed/storage/disk_location_ec.go
index 21f26e093..1cbb2c380 100644
--- a/weed/storage/disk_location_ec.go
+++ b/weed/storage/disk_location_ec.go
@@ -27,6 +27,17 @@ func (l *DiskLocation) FindEcVolume(vid needle.VolumeId) (*erasure_coding.EcVolu
 	return nil, false
 }
 
+func (l *DiskLocation) DestroyEcVolume(vid needle.VolumeId) {
+	l.ecVolumesLock.Lock()
+	defer l.ecVolumesLock.Unlock()
+
+	ecVolume, found := l.ecVolumes[vid]
+	if found {
+		ecVolume.Destroy()
+		delete(l.ecVolumes,vid)
+	}
+}
+
 func (l *DiskLocation) FindEcShard(vid needle.VolumeId, shardId erasure_coding.ShardId) (*erasure_coding.EcVolumeShard, bool) {
 	l.ecVolumesLock.RLock()
 	defer l.ecVolumesLock.RUnlock()
diff --git a/weed/storage/store_ec.go b/weed/storage/store_ec.go
index 9f6609ec6..0c9f674d5 100644
--- a/weed/storage/store_ec.go
+++ b/weed/storage/store_ec.go
@@ -94,6 +94,12 @@ func (s *Store) FindEcVolume(vid needle.VolumeId) (*erasure_coding.EcVolume, boo
 	return nil, false
 }
 
+func (s *Store) DestroyEcVolume(vid needle.VolumeId) {
+	for _, location := range s.Locations {
+		location.DestroyEcVolume(vid)
+	}
+}
+
 func (s *Store) ReadEcShardNeedle(ctx context.Context, vid needle.VolumeId, n *needle.Needle) (int, error) {
 	for _, location := range s.Locations {
 		if localEcVolume, found := location.FindEcVolume(vid); found {