avoid data race read volume.IsEmpty (#4574)

* avoid data race read volume.IsEmpty

-   avoid phantom read isEmpty for onlyEmpty
-   use `v.DataBackend.GetStat()` in v.dataFileAccessLock scope

* add Destroy(onlyEmpty: true) test

* add Destroy(onlyEmpty: false) test

* remove unused `IsEmpty()`

* change literal `8` to `SuperBlockSize`

1 files changed, 18 insertions, 2 deletions

diff --git a/weed/storage/volume_write.go b/weed/storage/volume_write.go
index edee8465c..ae607e691 100644
--- a/weed/storage/volume_write.go
+++ b/weed/storage/volume_write.go
@@ -50,8 +50,24 @@ func (v *Volume) isFileUnchanged(n *needle.Needle) bool {
 	return false
 }
 
+var ErrVolumeNotEmpty = fmt.Errorf("volume not empty")
+
 // Destroy removes everything related to this volume
-func (v *Volume) Destroy() (err error) {
+func (v *Volume) Destroy(onlyEmpty bool) (err error) {
+	v.dataFileAccessLock.Lock()
+	defer v.dataFileAccessLock.Unlock()
+
+	if onlyEmpty {
+		isEmpty, e := v.doIsEmpty()
+		if e != nil {
+			err = fmt.Errorf("failed to read isEmpty %v", e)
+			return
+		}
+		if !isEmpty {
+			err = ErrVolumeNotEmpty
+			return
+		}
+	}
 	if v.isCompacting || v.isCommitCompacting {
 		err = fmt.Errorf("volume %d is compacting", v.Id)
 		return
@@ -63,7 +79,7 @@ func (v *Volume) Destroy() (err error) {
 			backendStorage.DeleteFile(storageKey)
 		}
 	}
-	v.Close()
+	v.doClose()
 	removeVolumeFiles(v.DataFileName())
 	removeVolumeFiles(v.IndexFileName())
 	return