From e40634e6b49e8ab20720c7b8625b9333bf2bab37 Mon Sep 17 00:00:00 2001
From: Chris Lu <chris.lu@gmail.com>
Date: Mon, 12 Aug 2019 00:53:50 -0700
Subject: [PATCH] volume: fail the volume deletion if compaction is in progress

fix https://github.com/chrislusf/seaweedfs/issues/1035
---
 weed/storage/volume.go            |  2 ++
 weed/storage/volume_read_write.go |  4 ++++
 weed/storage/volume_vacuum.go     | 17 +++++++++++++++++
 3 files changed, 23 insertions(+)

diff --git a/weed/storage/volume.go b/weed/storage/volume.go
index a567b2e14..a5e923547 100644
--- a/weed/storage/volume.go
+++ b/weed/storage/volume.go
@@ -33,6 +33,8 @@ type Volume struct {
 
 	lastCompactIndexOffset uint64
 	lastCompactRevision    uint16
+
+	isCompacting bool
 }
 
 func NewVolume(dirname string, collection string, id needle.VolumeId, needleMapKind NeedleMapType, replicaPlacement *ReplicaPlacement, ttl *needle.TTL, preallocate int64) (v *Volume, e error) {
diff --git a/weed/storage/volume_read_write.go b/weed/storage/volume_read_write.go
index 2f2eb81dc..2c67b2dc4 100644
--- a/weed/storage/volume_read_write.go
+++ b/weed/storage/volume_read_write.go
@@ -43,6 +43,10 @@ func (v *Volume) Destroy() (err error) {
 		err = fmt.Errorf("%s is read-only", v.dataFile.Name())
 		return
 	}
+	if v.isCompacting {
+		err = fmt.Errorf("volume %d is compacting", v.Id)
+		return
+	}
 	v.Close()
 	os.Remove(v.FileName() + ".dat")
 	os.Remove(v.FileName() + ".idx")
diff --git a/weed/storage/volume_vacuum.go b/weed/storage/volume_vacuum.go
index 020dc7c74..ff09df42d 100644
--- a/weed/storage/volume_vacuum.go
+++ b/weed/storage/volume_vacuum.go
@@ -27,6 +27,10 @@ func (v *Volume) Compact(preallocate int64, compactionBytePerSecond int64) error
 	//v.accessLock.Lock()
 	//defer v.accessLock.Unlock()
 	//glog.V(3).Infof("Got Compaction lock...")
+	v.isCompacting = true
+	defer func() {
+		v.isCompacting = false
+	}()
 
 	filePath := v.FileName()
 	v.lastCompactIndexOffset = v.nm.IndexFileSize()
@@ -37,6 +41,12 @@ func (v *Volume) Compact(preallocate int64, compactionBytePerSecond int64) error
 
 func (v *Volume) Compact2() error {
 	glog.V(3).Infof("Compact2 volume %d ...", v.Id)
+
+	v.isCompacting = true
+	defer func() {
+		v.isCompacting = false
+	}()
+
 	filePath := v.FileName()
 	glog.V(3).Infof("creating copies for volume %d ...", v.Id)
 	return v.copyDataBasedOnIndexFile(filePath+".cpd", filePath+".cpx")
@@ -44,8 +54,15 @@ func (v *Volume) Compact2() error {
 
 func (v *Volume) CommitCompact() error {
 	glog.V(0).Infof("Committing volume %d vacuuming...", v.Id)
+
+	v.isCompacting = true
+	defer func() {
+		v.isCompacting = false
+	}()
+
 	v.dataFileAccessLock.Lock()
 	defer v.dataFileAccessLock.Unlock()
+
 	glog.V(3).Infof("Got volume %d committing lock...", v.Id)
 	v.nm.Close()
 	if err := v.dataFile.Close(); err != nil {