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, 23 insertions, 4 deletions

diff --git a/weed/storage/volume.go b/weed/storage/volume.go
index 79b0059d6..81c466f8f 100644
--- a/weed/storage/volume.go
+++ b/weed/storage/volume.go
@@ -133,6 +133,25 @@ func (v *Volume) ContentSize() uint64 {
 	return v.nm.ContentSize()
 }
 
+func (v *Volume) doIsEmpty() (bool, error) {
+	if v.DataBackend != nil {
+		datFileSize, _, e := v.DataBackend.GetStat()
+		if e != nil {
+			glog.V(0).Infof("Failed to read file size %s %v", v.DataBackend.Name(), e)
+			return false, e
+		}
+		if datFileSize > super_block.SuperBlockSize {
+			return false, nil
+		}
+	}
+	if v.nm != nil {
+		if v.nm.ContentSize() > 0 {
+			return false, nil
+		}
+	}
+	return true, nil
+}
+
 func (v *Volume) DeletedSize() uint64 {
 	v.dataFileAccessLock.RLock()
 	defer v.dataFileAccessLock.RUnlock()
@@ -202,6 +221,10 @@ func (v *Volume) Close() {
 	v.dataFileAccessLock.Lock()
 	defer v.dataFileAccessLock.Unlock()
 
+	v.doClose()
+}
+
+func (v *Volume) doClose() {
 	for v.isCommitCompacting {
 		time.Sleep(521 * time.Millisecond)
 		glog.Warningf("Volume Close wait for compaction %d", v.Id)
@@ -332,7 +355,3 @@ func (v *Volume) IsReadOnly() bool {
 	defer v.noWriteLock.RUnlock()
 	return v.noWriteOrDelete || v.noWriteCanDelete || v.location.isDiskSpaceLow
 }
-func (v *Volume) IsEmpty() bool {
-	datSize, _, _ := v.FileStat()
-	return datSize <= super_block.SuperBlockSize && v.ContentSize() == 0
-}