add leveldb support for needle map

This supposedly should reduce memory consumption. However, for tests
with millions of, this shows consuming more memories. Need to see
whether this will work out. If not, later boltdb will be tested.

6 files changed, 15 insertions, 19 deletions

diff --git a/go/weed/compact.go b/go/weed/compact.go
index 71c4ea90f..6ce55a609 100644
--- a/go/weed/compact.go
+++ b/go/weed/compact.go
@@ -33,7 +33,7 @@ func runCompact(cmd *Command, args []string) bool {
 	}
 
 	vid := storage.VolumeId(*compactVolumeId)
-	v, err := storage.NewVolume(*compactVolumePath, *compactVolumeCollection, vid, nil, nil)
+	v, err := storage.NewVolume(*compactVolumePath, *compactVolumeCollection, vid, false, nil, nil)
 	if err != nil {
 		glog.Fatalf("Load Volume [ERROR] %s\n", err)
 	}
diff --git a/go/weed/fix.go b/go/weed/fix.go
index b2df07554..d2cd40398 100644
--- a/go/weed/fix.go
+++ b/go/weed/fix.go
@@ -52,8 +52,8 @@ func runFix(cmd *Command, args []string) bool {
 	}, 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 {
-			count, pe := nm.Put(n.Id, uint32(offset/storage.NeedlePaddingSize), n.Size)
-			glog.V(2).Infof("saved %d with error %v", count, pe)
+			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)
diff --git a/go/weed/server.go b/go/weed/server.go
index 48612a27b..71346de0a 100644
--- a/go/weed/server.go
+++ b/go/weed/server.go
@@ -68,6 +68,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.")
 	volumeFixJpgOrientation       = cmdServer.Flag.Bool("volume.images.fix.orientation", true, "Adjust jpg orientation when uploading.")
 	isStartingFiler               = cmdServer.Flag.Bool("filer", false, "whether to start filer")
 
@@ -235,6 +236,7 @@ func runServer(cmd *Command, args []string) bool {
 	volumeServer := weed_server.NewVolumeServer(volumeMux, publicVolumeMux,
 		*serverIp, *volumePort, *serverPublicUrl,
 		folders, maxCounts,
+		*volumeUseLevelDb,
 		*serverIp+":"+strconv.Itoa(*masterPort), *volumePulse, *serverDataCenter, *serverRack,
 		serverWhiteList, *volumeFixJpgOrientation,
 	)
diff --git a/go/weed/volume.go b/go/weed/volume.go
index aa2643d20..e2c6ebd94 100644
--- a/go/weed/volume.go
+++ b/go/weed/volume.go
@@ -32,6 +32,7 @@ type VolumeServerOptions struct {
 	dataCenter            *string
 	rack                  *string
 	whiteList             []string
+	useLevelDb            *bool
 	fixJpgOrientation     *bool
 }
 
@@ -48,6 +49,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.fixJpgOrientation = cmdVolume.Flag.Bool("images.fix.orientation", true, "Adjust jpg orientation when uploading.")
 }
 
@@ -116,6 +118,7 @@ func runVolume(cmd *Command, args []string) bool {
 	volumeServer := weed_server.NewVolumeServer(volumeMux, publicVolumeMux,
 		*v.ip, *v.port, *v.publicUrl,
 		v.folders, v.folderMaxLimits,
+		*v.useLevelDb,
 		*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 e3878fac4..d84b39808 100644
--- a/go/weed/weed_server/volume_server.go
+++ b/go/weed/weed_server/volume_server.go
@@ -20,12 +20,14 @@ type VolumeServer struct {
 	store        *storage.Store
 	guard        *security.Guard
 
+	UseLevelDb        bool
 	FixJpgOrientation bool
 }
 
 func NewVolumeServer(adminMux, publicMux *http.ServeMux, ip string,
 	port int, publicUrl string,
 	folders []string, maxCounts []int,
+	useLevelDb bool,
 	masterNode string, pulseSeconds int,
 	dataCenter string, rack string,
 	whiteList []string,
@@ -34,10 +36,11 @@ func NewVolumeServer(adminMux, publicMux *http.ServeMux, ip string,
 		pulseSeconds:      pulseSeconds,
 		dataCenter:        dataCenter,
 		rack:              rack,
+		UseLevelDb:        useLevelDb,
 		FixJpgOrientation: fixJpgOrientation,
 	}
 	vs.SetMasterNode(masterNode)
-	vs.store = storage.NewStore(port, ip, publicUrl, folders, maxCounts)
+	vs.store = storage.NewStore(port, ip, publicUrl, folders, maxCounts, vs.UseLevelDb)
 
 	vs.guard = security.NewGuard(whiteList, "")
 
@@ -47,7 +50,6 @@ func NewVolumeServer(adminMux, publicMux *http.ServeMux, ip string,
 	adminMux.HandleFunc("/admin/vacuum_volume_check", vs.guard.WhiteList(vs.vacuumVolumeCheckHandler))
 	adminMux.HandleFunc("/admin/vacuum_volume_compact", vs.guard.WhiteList(vs.vacuumVolumeCompactHandler))
 	adminMux.HandleFunc("/admin/vacuum_volume_commit", vs.guard.WhiteList(vs.vacuumVolumeCommitHandler))
-	adminMux.HandleFunc("/admin/freeze_volume", vs.guard.WhiteList(vs.freezeVolumeHandler))
 	adminMux.HandleFunc("/admin/delete_collection", vs.guard.WhiteList(vs.deleteCollectionHandler))
 	adminMux.HandleFunc("/stats/counter", vs.guard.WhiteList(statsCounterHandler))
 	adminMux.HandleFunc("/stats/memory", vs.guard.WhiteList(statsMemoryHandler))
@@ -66,7 +68,7 @@ func NewVolumeServer(adminMux, publicMux *http.ServeMux, ip string,
 		vs.store.SetDataCenter(vs.dataCenter)
 		vs.store.SetRack(vs.rack)
 		for {
-			master, secretKey, err := vs.store.Join()
+			master, secretKey, err := vs.store.SendHeartbeatToMaster()
 			if err == nil {
 				if !connected {
 					connected = true
@@ -75,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(4).Infoln("Volume Server Failed to talk with master:", err.Error())
+				glog.V(0).Infof("Volume Server Failed to talk with master %s: %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 c84b72db0..0d70a757e 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"), r.FormValue("replication"), r.FormValue("ttl"))
+	err := vs.store.AddVolume(r.FormValue("volume"), r.FormValue("collection"), vs.UseLevelDb, r.FormValue("replication"), r.FormValue("ttl"))
 	if err == nil {
 		writeJsonQuiet(w, r, http.StatusAccepted, map[string]string{"error": ""})
 	} else {
@@ -40,17 +40,6 @@ func (vs *VolumeServer) deleteCollectionHandler(w http.ResponseWriter, r *http.R
 	glog.V(2).Infoln("deleting collection =", r.FormValue("collection"), ", error =", err)
 }
 
-func (vs *VolumeServer) freezeVolumeHandler(w http.ResponseWriter, r *http.Request) {
-	//TODO: notify master that this volume will be read-only
-	err := vs.store.FreezeVolume(r.FormValue("volume"))
-	if err == nil {
-		writeJsonQuiet(w, r, http.StatusOK, map[string]string{"error": ""})
-	} else {
-		writeJsonError(w, r, http.StatusInternalServerError, err)
-	}
-	glog.V(2).Infoln("freeze volume =", r.FormValue("volume"), ", error =", err)
-}
-
 func (vs *VolumeServer) statsDiskHandler(w http.ResponseWriter, r *http.Request) {
 	m := make(map[string]interface{})
 	m["Version"] = util.VERSION