wait to read again if the volume is compacting

weed/storage/volume.go

@@ -16,6 +16,7 @@ type Volume struct {
 	Collection    string
 	dataFile      *os.File
 	nm            NeedleMapper
+	compactingWg  sync.WaitGroup
 	needleMapKind NeedleMapType
 	readOnly      bool
 

weed/storage/volume_read_write.go

@@ -12,6 +12,8 @@ import (
 	. "github.com/chrislusf/seaweedfs/weed/storage/types"
 )
 
+var ErrorNotFound = errors.New("not found")
+
 // isFileUnchanged checks whether this needle to write is same as last one.
 // It requires serialized access in the same volume.
 func (v *Volume) isFileUnchanged(n *Needle) bool {
@@ -134,7 +136,11 @@ func (v *Volume) deleteNeedle(n *Needle) (uint32, error) {
 func (v *Volume) readNeedle(n *Needle) (int, error) {
 	nv, ok := v.nm.Get(n.Id)
 	if !ok || nv.Offset == 0 {
-		return -1, errors.New("Not Found")
+		v.compactingWg.Wait()
+		nv, ok = v.nm.Get(n.Id)
+		if !ok || nv.Offset == 0 {
+			return -1, ErrorNotFound
+		}
 	}
 	if nv.Size == TombstoneFileSize {
 		return -1, errors.New("Already Deleted")

weed/storage/volume_vacuum.go

@@ -43,6 +43,8 @@ func (v *Volume) commitCompact() error {
 	v.dataFileAccessLock.Lock()
 	defer v.dataFileAccessLock.Unlock()
 	glog.V(3).Infof("Got volume %d committing lock...", v.Id)
+	v.compactingWg.Add(1)
+	defer v.compactingWg.Done()
 	v.nm.Close()
 	if err := v.dataFile.Close(); err != nil {
 		glog.V(0).Infof("fail to close volume %d", v.Id)