Add boltdb for volume needle map

boltdb is fairly slow to write, about 6 minutes for recreating index
for 1553934 files. Boltdb loads 1,553,934 x 16 = 24,862,944bytes from
disk, and generate the boltdb as large as 134,217,728 bytes in 6
minutes.

To compare, for leveldb, it recreates index in leveldb as large as
27,188,148 bytes in 8 seconds.
For in memory version, it loads the index in

To test the memory consumption, the leveldb or boltdb index are
created. And the server is restarted. Using the benchmark tool to read
lots of files. There are 7 volumes in benchmark collection, each with
about 1553K files.
For leveldb, the memory starts at 142,884KB, and stays at 179,340KB.
For boltdb, the memory starts at 73,756KB, and stays at 144,564KB.
For in-memory, the memory starts at 368,152KB, and stays at 448,032KB.

7 files changed, 63 insertions, 42 deletions

diff --git a/go/weed/compact.go b/go/weed/compact.go
index 6ce55a609..a5ea8a529 100644
--- a/go/weed/compact.go
+++ b/go/weed/compact.go
@@ -33,7 +33,8 @@ func runCompact(cmd *Command, args []string) bool {
 	}
 
 	vid := storage.VolumeId(*compactVolumeId)
-	v, err := storage.NewVolume(*compactVolumePath, *compactVolumeCollection, vid, false, nil, nil)
+	v, err := storage.NewVolume(*compactVolumePath, *compactVolumeCollection, vid,
+		storage.NeedleMapInMemory, nil, nil)
 	if err != nil {
 		glog.Fatalf("Load Volume [ERROR] %s\n", err)
 	}
diff --git a/go/weed/export.go b/go/weed/export.go
index 9e7012937..b120c2fa7 100644
--- a/go/weed/export.go
+++ b/go/weed/export.go
@@ -99,23 +99,25 @@ func runExport(cmd *Command, args []string) bool {
 
 	var version storage.Version
 
-	err = storage.ScanVolumeFile(*exportVolumePath, *exportCollection, vid, func(superBlock storage.SuperBlock) error {
-		version = superBlock.Version()
-		return nil
-	}, true, func(n *storage.Needle, offset int64) error {
-		nv, ok := nm.Get(n.Id)
-		glog.V(3).Infof("key %d offset %d size %d disk_size %d gzip %v ok %v nv %+v",
-			n.Id, offset, n.Size, n.DiskSize(), n.IsGzipped(), ok, nv)
-		if ok && nv.Size > 0 && int64(nv.Offset)*8 == offset {
-			return walker(vid, n, version)
-		}
-		if !ok {
-			glog.V(2).Infof("This seems deleted %d size %d", n.Id, n.Size)
-		} else {
-			glog.V(2).Infof("Skipping later-updated Id %d size %d", n.Id, n.Size)
-		}
-		return nil
-	})
+	err = storage.ScanVolumeFile(*exportVolumePath, *exportCollection, vid,
+		storage.NeedleMapInMemory,
+		func(superBlock storage.SuperBlock) error {
+			version = superBlock.Version()
+			return nil
+		}, true, func(n *storage.Needle, offset int64) error {
+			nv, ok := nm.Get(n.Id)
+			glog.V(3).Infof("key %d offset %d size %d disk_size %d gzip %v ok %v nv %+v",
+				n.Id, offset, n.Size, n.DiskSize(), n.IsGzipped(), ok, nv)
+			if ok && nv.Size > 0 && int64(nv.Offset)*8 == offset {
+				return walker(vid, n, version)
+			}
+			if !ok {
+				glog.V(2).Infof("This seems deleted %d size %d", n.Id, n.Size)
+			} else {
+				glog.V(2).Infof("Skipping later-updated Id %d size %d", n.Id, n.Size)
+			}
+			return nil
+		})
 	if err != nil {
 		glog.Fatalf("Export Volume File [ERROR] %s\n", err)
 	}
diff --git a/go/weed/fix.go b/go/weed/fix.go
index d2cd40398..f51dc1bf2 100644
--- a/go/weed/fix.go
+++ b/go/weed/fix.go
@@ -47,19 +47,21 @@ func runFix(cmd *Command, args []string) bool {
 	defer nm.Close()
 
 	vid := storage.VolumeId(*fixVolumeId)
-	err = storage.ScanVolumeFile(*fixVolumePath, *fixVolumeCollection, vid, func(superBlock storage.SuperBlock) error {
-		return nil
-	}, false, func(n *storage.Needle, offset int64) error {
-		glog.V(2).Infof("key %d offset %d size %d disk_size %d gzip %v", n.Id, offset, n.Size, n.DiskSize(), n.IsGzipped())
-		if n.Size > 0 {
-			pe := nm.Put(n.Id, uint32(offset/storage.NeedlePaddingSize), n.Size)
-			glog.V(2).Infof("saved %d with error %v", n.Size, pe)
-		} else {
-			glog.V(2).Infof("skipping deleted file ...")
-			return nm.Delete(n.Id)
-		}
-		return nil
-	})
+	err = storage.ScanVolumeFile(*fixVolumePath, *fixVolumeCollection, vid,
+		storage.NeedleMapInMemory,
+		func(superBlock storage.SuperBlock) error {
+			return nil
+		}, false, func(n *storage.Needle, offset int64) error {
+			glog.V(2).Infof("key %d offset %d size %d disk_size %d gzip %v", n.Id, offset, n.Size, n.DiskSize(), n.IsGzipped())
+			if n.Size > 0 {
+				pe := nm.Put(n.Id, uint32(offset/storage.NeedlePaddingSize), n.Size)
+				glog.V(2).Infof("saved %d with error %v", n.Size, pe)
+			} else {
+				glog.V(2).Infof("skipping deleted file ...")
+				return nm.Delete(n.Id)
+			}
+			return nil
+		})
 	if err != nil {
 		glog.Fatalf("Export Volume File [ERROR] %s\n", err)
 	}
diff --git a/go/weed/server.go b/go/weed/server.go
index 71346de0a..39d02597b 100644
--- a/go/weed/server.go
+++ b/go/weed/server.go
@@ -11,6 +11,7 @@ import (
 	"time"
 
 	"github.com/chrislusf/weed-fs/go/glog"
+	"github.com/chrislusf/weed-fs/go/storage"
 	"github.com/chrislusf/weed-fs/go/util"
 	"github.com/chrislusf/weed-fs/go/weed/weed_server"
 	"github.com/gorilla/mux"
@@ -68,7 +69,7 @@ var (
 	volumeDataFolders             = cmdServer.Flag.String("dir", os.TempDir(), "directories to store data files. dir[,dir]...")
 	volumeMaxDataVolumeCounts     = cmdServer.Flag.String("volume.max", "7", "maximum numbers of volumes, count[,count]...")
 	volumePulse                   = cmdServer.Flag.Int("pulseSeconds", 5, "number of seconds between heartbeats")
-	volumeUseLevelDb              = cmdServer.Flag.Bool("volume.leveldb", false, "Change to leveldb mode to save memory with reduced performance of read and write.")
+	volumeIndexType               = cmdServer.Flag.String("volume.index", "memory", "Choose [memory|leveldb|boltdb] mode for memory~performance balance.")
 	volumeFixJpgOrientation       = cmdServer.Flag.Bool("volume.images.fix.orientation", true, "Adjust jpg orientation when uploading.")
 	isStartingFiler               = cmdServer.Flag.Bool("filer", false, "whether to start filer")
 
@@ -233,10 +234,17 @@ func runServer(cmd *Command, args []string) bool {
 	if isSeperatedPublicPort {
 		publicVolumeMux = http.NewServeMux()
 	}
+	volumeNeedleMapKind := storage.NeedleMapInMemory
+	switch *volumeIndexType {
+	case "leveldb":
+		volumeNeedleMapKind = storage.NeedleMapLevelDb
+	case "boltdb":
+		volumeNeedleMapKind = storage.NeedleMapBoltDb
+	}
 	volumeServer := weed_server.NewVolumeServer(volumeMux, publicVolumeMux,
 		*serverIp, *volumePort, *serverPublicUrl,
 		folders, maxCounts,
-		*volumeUseLevelDb,
+		volumeNeedleMapKind,
 		*serverIp+":"+strconv.Itoa(*masterPort), *volumePulse, *serverDataCenter, *serverRack,
 		serverWhiteList, *volumeFixJpgOrientation,
 	)
diff --git a/go/weed/volume.go b/go/weed/volume.go
index e2c6ebd94..2d3ecbb4d 100644
--- a/go/weed/volume.go
+++ b/go/weed/volume.go
@@ -9,6 +9,7 @@ import (
 	"time"
 
 	"github.com/chrislusf/weed-fs/go/glog"
+	"github.com/chrislusf/weed-fs/go/storage"
 	"github.com/chrislusf/weed-fs/go/util"
 	"github.com/chrislusf/weed-fs/go/weed/weed_server"
 )
@@ -32,7 +33,7 @@ type VolumeServerOptions struct {
 	dataCenter            *string
 	rack                  *string
 	whiteList             []string
-	useLevelDb            *bool
+	indexType             *string
 	fixJpgOrientation     *bool
 }
 
@@ -49,7 +50,7 @@ func init() {
 	v.maxCpu = cmdVolume.Flag.Int("maxCpu", 0, "maximum number of CPUs. 0 means all available CPUs")
 	v.dataCenter = cmdVolume.Flag.String("dataCenter", "", "current volume server's data center name")
 	v.rack = cmdVolume.Flag.String("rack", "", "current volume server's rack name")
-	v.useLevelDb = cmdVolume.Flag.Bool("leveldb", false, "Change to leveldb mode to save memory with reduced performance of read and write.")
+	v.indexType = cmdVolume.Flag.String("index", "memory", "Choose [memory|leveldb|boltdb] mode for memory~performance balance.")
 	v.fixJpgOrientation = cmdVolume.Flag.Bool("images.fix.orientation", true, "Adjust jpg orientation when uploading.")
 }
 
@@ -115,10 +116,17 @@ func runVolume(cmd *Command, args []string) bool {
 		publicVolumeMux = http.NewServeMux()
 	}
 
+	volumeNeedleMapKind := storage.NeedleMapInMemory
+	switch *v.indexType {
+	case "leveldb":
+		volumeNeedleMapKind = storage.NeedleMapLevelDb
+	case "boltdb":
+		volumeNeedleMapKind = storage.NeedleMapBoltDb
+	}
 	volumeServer := weed_server.NewVolumeServer(volumeMux, publicVolumeMux,
 		*v.ip, *v.port, *v.publicUrl,
 		v.folders, v.folderMaxLimits,
-		*v.useLevelDb,
+		volumeNeedleMapKind,
 		*v.master, *v.pulseSeconds, *v.dataCenter, *v.rack,
 		v.whiteList,
 		*v.fixJpgOrientation,
diff --git a/go/weed/weed_server/volume_server.go b/go/weed/weed_server/volume_server.go
index d84b39808..69c944c99 100644
--- a/go/weed/weed_server/volume_server.go
+++ b/go/weed/weed_server/volume_server.go
@@ -20,14 +20,14 @@ type VolumeServer struct {
 	store        *storage.Store
 	guard        *security.Guard
 
-	UseLevelDb        bool
+	needleMapKind     storage.NeedleMapType
 	FixJpgOrientation bool
 }
 
 func NewVolumeServer(adminMux, publicMux *http.ServeMux, ip string,
 	port int, publicUrl string,
 	folders []string, maxCounts []int,
-	useLevelDb bool,
+	needleMapKind storage.NeedleMapType,
 	masterNode string, pulseSeconds int,
 	dataCenter string, rack string,
 	whiteList []string,
@@ -36,11 +36,11 @@ func NewVolumeServer(adminMux, publicMux *http.ServeMux, ip string,
 		pulseSeconds:      pulseSeconds,
 		dataCenter:        dataCenter,
 		rack:              rack,
-		UseLevelDb:        useLevelDb,
+		needleMapKind:     needleMapKind,
 		FixJpgOrientation: fixJpgOrientation,
 	}
 	vs.SetMasterNode(masterNode)
-	vs.store = storage.NewStore(port, ip, publicUrl, folders, maxCounts, vs.UseLevelDb)
+	vs.store = storage.NewStore(port, ip, publicUrl, folders, maxCounts, vs.needleMapKind)
 
 	vs.guard = security.NewGuard(whiteList, "")
 
@@ -77,7 +77,7 @@ func NewVolumeServer(adminMux, publicMux *http.ServeMux, ip string,
 					glog.V(0).Infoln("Volume Server Connected with master at", master)
 				}
 			} else {
-				glog.V(0).Infof("Volume Server Failed to talk with master %s: %v", vs, err)
+				glog.V(0).Infof("Volume Server Failed to talk with master %+v: %v", vs, err)
 				if connected {
 					connected = false
 				}
diff --git a/go/weed/weed_server/volume_server_handlers_admin.go b/go/weed/weed_server/volume_server_handlers_admin.go
index 0d70a757e..eb8222ff8 100644
--- a/go/weed/weed_server/volume_server_handlers_admin.go
+++ b/go/weed/weed_server/volume_server_handlers_admin.go
@@ -17,7 +17,7 @@ func (vs *VolumeServer) statusHandler(w http.ResponseWriter, r *http.Request) {
 }
 
 func (vs *VolumeServer) assignVolumeHandler(w http.ResponseWriter, r *http.Request) {
-	err := vs.store.AddVolume(r.FormValue("volume"), r.FormValue("collection"), vs.UseLevelDb, r.FormValue("replication"), r.FormValue("ttl"))
+	err := vs.store.AddVolume(r.FormValue("volume"), r.FormValue("collection"), vs.needleMapKind, r.FormValue("replication"), r.FormValue("ttl"))
 	if err == nil {
 		writeJsonQuiet(w, r, http.StatusAccepted, map[string]string{"error": ""})
 	} else {