optimiz commitig compact (#3388)

* optimiz vacuuming volume

* fix bugx

* rename parameters

* fix conflict

* change copyDataBasedOnIndexFile to an instance method

* close needlemap

* optimiz commiting Vacuum volume for  leveldb index

* fix bugs

* fix leveldb loading bugs

* refactor

* fix leveldb loading bug

* add leveldb recovery

* add test case for levelDB

* modify test case to cover all the new branches

* use one tmpNm instead of two instances

* refactor

* refactor

* move setWatermark to the end

* add test for watermark and updating leveldb

* fix error logic

* refactor, add test

* check nil before close needlemapeer
add test case
fix metric bug

* add tests, fix bugs

* adjust log level
remove wrong test case
refactor

* avoid duplicate  updating metric for leveldb index

1 files changed, 54 insertions, 0 deletions

diff --git a/weed/storage/needle_map_memory.go b/weed/storage/needle_map_memory.go
index 99a9f28e0..93b1fa4f5 100644
--- a/weed/storage/needle_map_memory.go
+++ b/weed/storage/needle_map_memory.go
@@ -7,6 +7,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/storage/idx"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle_map"
 	. "github.com/seaweedfs/seaweedfs/weed/storage/types"
+	"github.com/syndtr/goleveldb/leveldb/opt"
 )
 
 type NeedleMap struct {
@@ -69,6 +70,9 @@ func (nm *NeedleMap) Delete(key NeedleId, offset Offset) error {
 	return nm.appendToIndexFile(key, offset, TombstoneFileSize)
 }
 func (nm *NeedleMap) Close() {
+	if nm.indexFile == nil {
+		return
+	}
 	indexFileName := nm.indexFile.Name()
 	if err := nm.indexFile.Sync(); err != nil {
 		glog.Warningf("sync file %s failed, %v", indexFileName, err)
@@ -79,3 +83,53 @@ func (nm *NeedleMap) Destroy() error {
 	nm.Close()
 	return os.Remove(nm.indexFile.Name())
 }
+
+func (nm *NeedleMap) UpdateNeedleMap(v *Volume, indexFile *os.File, opts *opt.Options) error {
+	if v.nm != nil {
+		v.nm.Close()
+		v.nm = nil
+	}
+	defer func() {
+		if v.tmpNm != nil {
+			v.tmpNm.Close()
+			v.tmpNm = nil
+		}
+	}()
+	nm.indexFile = indexFile
+	stat, err := indexFile.Stat()
+	if err != nil {
+		glog.Fatalf("stat file %s: %v", indexFile.Name(), err)
+		return err
+	}
+	nm.indexFileOffset = stat.Size()
+	v.nm = nm
+	v.tmpNm = nil
+	return nil
+}
+
+func (nm *NeedleMap) DoOffsetLoading(v *Volume, indexFile *os.File, startFrom uint64) error {
+	glog.V(0).Infof("loading idx from offset %d for file: %s", startFrom, indexFile.Name())
+	e := idx.WalkIndexFile(indexFile, startFrom, func(key NeedleId, offset Offset, size Size) error {
+		nm.MaybeSetMaxFileKey(key)
+		nm.FileCounter++
+		if !offset.IsZero() && size.IsValid() {
+			nm.FileByteCounter = nm.FileByteCounter + uint64(size)
+			oldOffset, oldSize := nm.m.Set(NeedleId(key), offset, size)
+			if !oldOffset.IsZero() && oldSize.IsValid() {
+				nm.DeletionCounter++
+				nm.DeletionByteCounter = nm.DeletionByteCounter + uint64(oldSize)
+			}
+		} else {
+			oldSize := nm.m.Delete(NeedleId(key))
+			nm.DeletionCounter++
+			nm.DeletionByteCounter = nm.DeletionByteCounter + uint64(oldSize)
+		}
+		return nil
+	})
+
+	return e
+}
+
+func (m *NeedleMap) UpdateNeedleMapMetric(indexFile *os.File) error {
+	return nil
+}