revert webp support due to "go test"

related to https://github.com/chai2010/webp/issues/9

.travis.yml

@@ -1,8 +1,8 @@
 sudo: false
 language: go
 go:
-- 1.5
-- 1.6
+- 1.7
+- 1.8.3
 - tip
 
 before_install:

README.md

@@ -12,7 +12,7 @@
 - [Wiki Documentation](https://github.com/chrislusf/seaweedfs/wiki)
 
 
-[![](https://www.paypalobjects.com/en_US/i/btn/btn_donateCC_LG.gif)](https://www.paypal.com/cgi-bin/webscr?cmd=_s-xclick&hosted_button_id=EEECLJ8QGTTPC) 
+[![](https://www.paypalobjects.com/en_US/i/btn/btn_donateCC_LG.gif)](https://www.paypal.com/cgi-bin/webscr?cmd=_s-xclick&hosted_button_id=EEECLJ8QGTTPC)
 
 ## Introduction
 
@@ -39,7 +39,7 @@ SeaweedFS started by implementing [Facebook's Haystack design paper](http://www.
 * Optional [filer server][Filer] provides "normal" directories and files via http
 * Optionally fix the orientation for jpeg pictures
 * Support Etag, Accept-Range, Last-Modified, etc.
-* Support in-memory/leveldb/boltdb mode tuning for memory/performance balance.
+* Support in-memory/leveldb/boltdb/btree mode tuning for memory/performance balance.
 
 [Filer]: https://github.com/chrislusf/seaweedfs/wiki/Filer
 
@@ -135,6 +135,8 @@ If you want get a scaled version of an image, you can add some params:
 
 ```
 http://localhost:8080/3/01637037d6.jpg?height=200&width=200
+http://localhost:8080/3/01637037d6.jpg?height=200&width=200&mode=fit
+http://localhost:8080/3/01637037d6.jpg?height=200&width=200&mode=fill
 ```
 
 ### Rack-Aware and Data Center-Aware Replication ###

docker/Dockerfile

@@ -1,4 +1,4 @@
-FROM frolvlad/alpine-glibc:alpine-3.4
+FROM frolvlad/alpine-glibc:alpine-3.5
 
 RUN apk add --no-cache --virtual=build-dependencies --update curl wget ca-certificates && \
     wget -P /tmp https://github.com/$(curl -s -L https://github.com/chrislusf/seaweedfs/releases/latest | egrep -o '/chrislusf/seaweedfs/releases/download/.*/linux_amd64.tar.gz') && \

weed/command/benchmark.go

@@ -201,9 +201,11 @@ func writeFiles(idChan chan int, fileIdLineChan chan string, s *stat) {
 		}()
 	}
 
+	random := rand.New(rand.NewSource(time.Now().UnixNano()))
+
 	for id := range idChan {
 		start := time.Now()
-		fileSize := int64(*b.fileSize + rand.Intn(64))
+		fileSize := int64(*b.fileSize + random.Intn(64))
 		fp := &operation.FilePart{Reader: &FakeReader{id: uint64(id), size: fileSize}, FileSize: fileSize}
 		ar := &operation.VolumeAssignRequest{
 			Count:      1,
@@ -212,7 +214,7 @@ func writeFiles(idChan chan int, fileIdLineChan chan string, s *stat) {
 		if assignResult, err := operation.Assign(*b.server, ar); err == nil {
 			fp.Server, fp.Fid, fp.Collection = assignResult.Url, assignResult.Fid, *b.collection
 			if _, err := fp.Upload(0, *b.server, secret); err == nil {
-				if rand.Intn(100) < *b.deletePercentage {
+				if random.Intn(100) < *b.deletePercentage {
 					s.total++
 					delayedDeleteChan <- &delayedFile{time.Now().Add(time.Second), fp}
 				} else {
@@ -239,6 +241,8 @@ func writeFiles(idChan chan int, fileIdLineChan chan string, s *stat) {
 
 func readFiles(fileIdLineChan chan string, s *stat) {
 	defer wait.Done()
+	random := rand.New(rand.NewSource(time.Now().UnixNano()))
+
 	for fid := range fileIdLineChan {
 		if len(fid) == 0 {
 			continue
@@ -258,7 +262,7 @@ func readFiles(fileIdLineChan chan string, s *stat) {
 			println("!!!! volume id ", vid, " location not found!!!!!")
 			continue
 		}
-		server := ret.Locations[rand.Intn(len(ret.Locations))].Url
+		server := ret.Locations[random.Intn(len(ret.Locations))].Url
 		url := "http://" + server + "/" + fid
 		if bytesRead, err := util.Get(url); err == nil {
 			s.completed++
@@ -297,6 +301,8 @@ func readFileIds(fileName string, fileIdLineChan chan string) {
 	}
 	defer file.Close()
 
+	random := rand.New(rand.NewSource(time.Now().UnixNano()))
+
 	r := bufio.NewReader(file)
 	if *b.sequentialRead {
 		for {
@@ -317,7 +323,7 @@ func readFileIds(fileName string, fileIdLineChan chan string) {
 		}
 		if len(lines) > 0 {
 			for i := 0; i < readStats.total; i++ {
-				fileIdLineChan <- lines[rand.Intn(len(lines))]
+				fileIdLineChan <- lines[random.Intn(len(lines))]
 			}
 		}
 	}

weed/command/export.go

@@ -118,7 +118,7 @@ func runExport(cmd *Command, args []string) bool {
 	}
 	defer indexFile.Close()
 
-	needleMap, err := storage.LoadNeedleMap(indexFile)
+	needleMap, err := storage.LoadBtreeNeedleMap(indexFile)
 	if err != nil {
 		glog.Fatalf("cannot load needle map from %s: %s", indexFile.Name(), err)
 	}

weed/command/filer.go

@@ -19,6 +19,7 @@ type FilerOptions struct {
 	master                  *string
 	ip                      *string
 	port                    *int
+	publicPort              *int
 	collection              *string
 	defaultReplicaPlacement *string
 	dir                     *string
@@ -40,6 +41,7 @@ func init() {
 	f.collection = cmdFiler.Flag.String("collection", "", "all data will be stored in this collection")
 	f.ip = cmdFiler.Flag.String("ip", "", "filer server http listen ip address")
 	f.port = cmdFiler.Flag.Int("port", 8888, "filer server http listen port")
+	f.publicPort = cmdFiler.Flag.Int("port.public", 0, "port opened to public")
 	f.dir = cmdFiler.Flag.String("dir", os.TempDir(), "directory to store meta data")
 	f.defaultReplicaPlacement = cmdFiler.Flag.String("defaultReplicaPlacement", "000", "default replication type if not specified")
 	f.redirectOnRead = cmdFiler.Flag.Bool("redirectOnRead", false, "whether proxy or redirect to volume server during file GET request")
@@ -83,29 +85,57 @@ func runFiler(cmd *Command, args []string) bool {
 		glog.Fatalf("Check Meta Folder (-dir) Writable %s : %s", *f.dir, err)
 	}
 
-	r := http.NewServeMux()
-	_, nfs_err := weed_server.NewFilerServer(r, *f.ip, *f.port, *f.master, *f.dir, *f.collection,
-		*f.defaultReplicaPlacement, *f.redirectOnRead, *f.disableDirListing,
-		*f.confFile,
-		*f.maxMB,
-		*f.secretKey,
-		*f.cassandra_server, *f.cassandra_keyspace,
-		*f.redis_server, *f.redis_password, *f.redis_database,
+	f.start()
+
+	return true
+}
+
+func (fo *FilerOptions) start() {
+
+	defaultMux := http.NewServeMux()
+	publicVolumeMux := defaultMux
+
+	if *fo.publicPort != 0 {
+		publicVolumeMux = http.NewServeMux()
+	}
+
+	_, nfs_err := weed_server.NewFilerServer(defaultMux, publicVolumeMux,
+		*fo.ip, *fo.port, *fo.master, *fo.dir, *fo.collection,
+		*fo.defaultReplicaPlacement, *fo.redirectOnRead, *fo.disableDirListing,
+		*fo.confFile,
+		*fo.maxMB,
+		*fo.secretKey,
+		*fo.cassandra_server, *fo.cassandra_keyspace,
+		*fo.redis_server, *fo.redis_password, *fo.redis_database,
 	)
 	if nfs_err != nil {
 		glog.Fatalf("Filer startup error: %v", nfs_err)
 	}
-	glog.V(0).Infoln("Start Seaweed Filer", util.VERSION, "at port", strconv.Itoa(*f.port))
+
+	if *fo.publicPort != 0 {
+		publicListeningAddress := *fo.ip + ":" + strconv.Itoa(*fo.publicPort)
+		glog.V(0).Infoln("Start Seaweed filer server", util.VERSION, "public at", publicListeningAddress)
+		publicListener, e := util.NewListener(publicListeningAddress, 0)
+		if e != nil {
+			glog.Fatalf("Filer server public listener error on port %d:%v", *fo.publicPort, e)
+		}
+		go func() {
+			if e := http.Serve(publicListener, publicVolumeMux); e != nil {
+				glog.Fatalf("Volume server fail to serve public: %v", e)
+			}
+		}()
+	}
+
+	glog.V(0).Infoln("Start Seaweed Filer", util.VERSION, "at port", strconv.Itoa(*fo.port))
 	filerListener, e := util.NewListener(
-		":"+strconv.Itoa(*f.port),
+		":"+strconv.Itoa(*fo.port),
 		time.Duration(10)*time.Second,
 	)
 	if e != nil {
 		glog.Fatalf("Filer listener error: %v", e)
 	}
-	if e := http.Serve(filerListener, r); e != nil {
+	if e := http.Serve(filerListener, defaultMux); e != nil {
 		glog.Fatalf("Filer Fail to serve: %v", e)
 	}
 
-	return true
 }

weed/command/fix.go

@@ -43,7 +43,7 @@ func runFix(cmd *Command, args []string) bool {
 	}
 	defer indexFile.Close()
 
-	nm := storage.NewNeedleMap(indexFile)
+	nm := storage.NewBtreeNeedleMap(indexFile)
 	defer nm.Close()
 
 	vid := storage.VolumeId(*fixVolumeId)

weed/command/master.go

@@ -41,7 +41,6 @@ var (
 	volumeSizeLimitMB       = cmdMaster.Flag.Uint("volumeSizeLimitMB", 30*1000, "Master stops directing writes to oversized volumes.")
 	volumePreallocate       = cmdMaster.Flag.Bool("volumePreallocate", false, "Preallocate disk space for volumes.")
 	mpulse                  = cmdMaster.Flag.Int("pulseSeconds", 5, "number of seconds between heartbeats")
-	confFile                = cmdMaster.Flag.String("conf", "/etc/weedfs/weedfs.conf", "Deprecating! xml configuration file")
 	defaultReplicaPlacement = cmdMaster.Flag.String("defaultReplication", "000", "Default replication type if not specified.")
 	// mTimeout                = cmdMaster.Flag.Int("idleTimeout", 30, "connection idle seconds")
 	mMaxCpu               = cmdMaster.Flag.Int("maxCpu", 0, "maximum number of CPUs. 0 means all available CPUs")
@@ -79,7 +78,7 @@ func runMaster(cmd *Command, args []string) bool {
 	r := mux.NewRouter()
 	ms := weed_server.NewMasterServer(r, *mport, *metaFolder,
 		*volumeSizeLimitMB, *volumePreallocate,
-		*mpulse, *confFile, *defaultReplicaPlacement, *garbageThreshold,
+		*mpulse, *defaultReplicaPlacement, *garbageThreshold,
 		masterWhiteList, *masterSecureKey,
 	)
 

weed/command/mount.go

@@ -26,7 +26,7 @@ var cmdMount = &Command{
   2) have a "weed filer" running
   These 2 requirements can be achieved with one command "weed server -filer=true"
 
-  This uses bazil.org/fuse, whichenables writing FUSE file systems on
+  This uses bazil.org/fuse, which enables writing FUSE file systems on
   Linux, and OS X.
 
   On OS X, it requires OSXFUSE (http://osxfuse.github.com/).

weed/command/mount_std.go

@@ -4,8 +4,11 @@ package command
 
 import (
 	"fmt"
+	"os"
 	"runtime"
 
+	"path"
+
 	"bazil.org/fuse"
 	"bazil.org/fuse/fs"
 	"github.com/chrislusf/seaweedfs/weed/filer"
@@ -22,7 +25,7 @@ func runMount(cmd *Command, args []string) bool {
 		return false
 	}
 
-	c, err := fuse.Mount(*mountOptions.dir)
+	c, err := fuse.Mount(*mountOptions.dir, fuse.LocalVolume())
 	if err != nil {
 		glog.Fatal(err)
 		return false
@@ -47,60 +50,98 @@ func runMount(cmd *Command, args []string) bool {
 	return true
 }
 
-type File struct {
-	FileId filer.FileId
-	Name   string
-}
-
-func (File) Attr(context context.Context, attr *fuse.Attr) error {
-	return nil
-}
-func (File) ReadAll(ctx context.Context) ([]byte, error) {
-	return []byte("hello, world\n"), nil
-}
-
-type Dir struct {
-	Path string
-	Id   uint64
-}
-
-func (dir Dir) Attr(context context.Context, attr *fuse.Attr) error {
-	return nil
-}
-
-func (dir Dir) Lookup(ctx context.Context, name string) (fs.Node, error) {
-	files_result, e := filer.ListFiles(*mountOptions.filer, dir.Path, name)
-	if e != nil {
-		return nil, fuse.ENOENT
-	}
-	if len(files_result.Files) > 0 {
-		return File{files_result.Files[0].Id, files_result.Files[0].Name}, nil
-	}
-	return nil, fmt.Errorf("File Not Found for %s", name)
-}
-
 type WFS struct{}
 
 func (WFS) Root() (fs.Node, error) {
-	return Dir{}, nil
+	return &Dir{Path: "/"}, nil
 }
 
-func (dir *Dir) ReadDir(ctx context.Context) ([]fuse.Dirent, error) {
+var fileIdMap = make(map[uint64]filer.FileId)
+
+type Dir struct {
+	Id        uint64
+	Path      string
+	DirentMap map[string]*fuse.Dirent
+}
+
+func (dir *Dir) Attr(context context.Context, attr *fuse.Attr) error {
+	attr.Inode = dir.Id
+	attr.Mode = os.ModeDir | 0555
+	return nil
+}
+
+func (dir *Dir) Lookup(ctx context.Context, name string) (fs.Node, error) {
+	if dirent, ok := dir.DirentMap[name]; ok {
+		if dirent.Type == fuse.DT_File {
+			return &File{Id: dirent.Inode, FileId: fileIdMap[dirent.Inode], Name: dirent.Name}, nil
+		}
+		return &Dir{
+			Id:   dirent.Inode,
+			Path: path.Join(dir.Path, dirent.Name),
+		}, nil
+	}
+	return nil, fuse.ENOENT
+}
+
+func (dir *Dir) ReadDirAll(ctx context.Context) ([]fuse.Dirent, error) {
 	var ret []fuse.Dirent
+	if dir.DirentMap == nil {
+		dir.DirentMap = make(map[string]*fuse.Dirent)
+	}
 	if dirs, e := filer.ListDirectories(*mountOptions.filer, dir.Path); e == nil {
 		for _, d := range dirs.Directories {
 			dirId := uint64(d.Id)
-			ret = append(ret, fuse.Dirent{Inode: dirId, Name: d.Name, Type: fuse.DT_Dir})
+			dirent := fuse.Dirent{Inode: dirId, Name: d.Name, Type: fuse.DT_Dir}
+			ret = append(ret, dirent)
+			dir.DirentMap[d.Name] = &dirent
 		}
 	}
 	if files, e := filer.ListFiles(*mountOptions.filer, dir.Path, ""); e == nil {
 		for _, f := range files.Files {
 			if fileId, e := storage.ParseFileId(string(f.Id)); e == nil {
 				fileInode := uint64(fileId.VolumeId)<<48 + fileId.Key
-				ret = append(ret, fuse.Dirent{Inode: fileInode, Name: f.Name, Type: fuse.DT_File})
+				dirent := fuse.Dirent{Inode: fileInode, Name: f.Name, Type: fuse.DT_File}
+				ret = append(ret, dirent)
+				dir.DirentMap[f.Name] = &dirent
+				fileIdMap[fileInode] = f.Id
 			}
-
 		}
 	}
 	return ret, nil
 }
+
+func (dir *Dir) Remove(ctx context.Context, req *fuse.RemoveRequest) error {
+	name := path.Join(dir.Path, req.Name)
+	err := filer.DeleteDirectoryOrFile(*mountOptions.filer, name, req.Dir)
+	if err != nil {
+		fmt.Printf("Delete file %s [ERROR] %s\n", name, err)
+	}
+	return err
+}
+
+type File struct {
+	Id     uint64
+	FileId filer.FileId
+	Name   string
+}
+
+func (file *File) Attr(context context.Context, attr *fuse.Attr) error {
+	attr.Inode = file.Id
+	attr.Mode = 0444
+	ret, err := filer.GetFileSize(*mountOptions.filer, string(file.FileId))
+	if err == nil {
+		attr.Size = ret.Size
+	} else {
+		fmt.Printf("Get file %s attr [ERROR] %s\n", file.Name, err)
+	}
+	return err
+}
+
+func (file *File) ReadAll(ctx context.Context) ([]byte, error) {
+	ret, err := filer.GetFileContent(*mountOptions.filer, string(file.FileId))
+	if err == nil {
+		return ret.Content, nil
+	}
+	fmt.Printf("Get file %s content [ERROR] %s\n", file.Name, err)
+	return nil, err
+}

weed/command/server.go

@@ -66,14 +66,13 @@ var (
 	masterMetaFolder              = cmdServer.Flag.String("master.dir", "", "data directory to store meta data, default to same as -dir specified")
 	masterVolumeSizeLimitMB       = cmdServer.Flag.Uint("master.volumeSizeLimitMB", 30*1000, "Master stops directing writes to oversized volumes.")
 	masterVolumePreallocate       = cmdServer.Flag.Bool("master.volumePreallocate", false, "Preallocate disk space for volumes.")
-	masterConfFile                = cmdServer.Flag.String("master.conf", "/etc/weedfs/weedfs.conf", "xml configuration file")
 	masterDefaultReplicaPlacement = cmdServer.Flag.String("master.defaultReplicaPlacement", "000", "Default replication type if not specified.")
 	volumePort                    = cmdServer.Flag.Int("volume.port", 8080, "volume server http listen port")
 	volumePublicPort              = cmdServer.Flag.Int("volume.port.public", 0, "volume server public port")
 	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")
-	volumeIndexType               = cmdServer.Flag.String("volume.index", "memory", "Choose [memory|leveldb|boltdb] mode for memory~performance balance.")
+	volumeIndexType               = cmdServer.Flag.String("volume.index", "memory", "Choose [memory|leveldb|boltdb|btree] mode for memory~performance balance.")
 	volumeFixJpgOrientation       = cmdServer.Flag.Bool("volume.images.fix.orientation", true, "Adjust jpg orientation when uploading.")
 	volumeReadRedirect            = cmdServer.Flag.Bool("volume.read.redirect", true, "Redirect moved or non-local volumes.")
 	volumeServerPublicUrl         = cmdServer.Flag.String("volume.publicUrl", "", "publicly accessible address")
@@ -88,6 +87,7 @@ func init() {
 	filerOptions.master = cmdServer.Flag.String("filer.master", "", "default to current master server")
 	filerOptions.collection = cmdServer.Flag.String("filer.collection", "", "all data will be stored in this collection")
 	filerOptions.port = cmdServer.Flag.Int("filer.port", 8888, "filer server http listen port")
+	filerOptions.publicPort = cmdServer.Flag.Int("filer.port.public", 0, "filer server public http listen port")
 	filerOptions.dir = cmdServer.Flag.String("filer.dir", "", "directory to store meta data, default to a 'filer' sub directory of what -dir is specified")
 	filerOptions.defaultReplicaPlacement = cmdServer.Flag.String("filer.defaultReplicaPlacement", "", "Default replication type if not specified during runtime.")
 	filerOptions.redirectOnRead = cmdServer.Flag.Bool("filer.redirectOnRead", false, "whether proxy or redirect to volume server during file GET request")
@@ -117,6 +117,7 @@ func runServer(cmd *Command, args []string) bool {
 	}
 
 	*filerOptions.master = *serverIp + ":" + strconv.Itoa(*masterPort)
+	filerOptions.ip = serverIp
 
 	if *filerOptions.defaultReplicaPlacement == "" {
 		*filerOptions.defaultReplicaPlacement = *masterDefaultReplicaPlacement
@@ -173,30 +174,9 @@ func runServer(cmd *Command, args []string) bool {
 	if *isStartingFiler {
 		go func() {
 			time.Sleep(1 * time.Second)
-			r := http.NewServeMux()
-			_, nfs_err := weed_server.NewFilerServer(r, *serverBindIp, *filerOptions.port, *filerOptions.master, *filerOptions.dir, *filerOptions.collection,
-				*filerOptions.defaultReplicaPlacement,
-				*filerOptions.redirectOnRead, *filerOptions.disableDirListing,
-				*filerOptions.confFile,
-				*filerOptions.maxMB,
-				*filerOptions.secretKey,
-				*filerOptions.cassandra_server, *filerOptions.cassandra_keyspace,
-				*filerOptions.redis_server, *filerOptions.redis_password, *filerOptions.redis_database,
-			)
-			if nfs_err != nil {
-				glog.Fatalf("Filer startup error: %v", nfs_err)
-			}
-			glog.V(0).Infoln("Start Seaweed Filer", util.VERSION, "at port", strconv.Itoa(*filerOptions.port))
-			filerListener, e := util.NewListener(
-				":"+strconv.Itoa(*filerOptions.port),
-				time.Duration(10)*time.Second,
-			)
-			if e != nil {
-				glog.Fatalf("Filer listener error: %v", e)
-			}
-			if e := http.Serve(filerListener, r); e != nil {
-				glog.Fatalf("Filer Fail to serve: %v", e)
-			}
+
+			filerOptions.start()
+
 		}()
 	}
 
@@ -210,7 +190,7 @@ func runServer(cmd *Command, args []string) bool {
 		r := mux.NewRouter()
 		ms := weed_server.NewMasterServer(r, *masterPort, *masterMetaFolder,
 			*masterVolumeSizeLimitMB, *masterVolumePreallocate,
-			*volumePulse, *masterConfFile, *masterDefaultReplicaPlacement, *serverGarbageThreshold,
+			*volumePulse, *masterDefaultReplicaPlacement, *serverGarbageThreshold,
 			serverWhiteList, *serverSecureKey,
 		)
 
@@ -277,6 +257,8 @@ func runServer(cmd *Command, args []string) bool {
 		volumeNeedleMapKind = storage.NeedleMapLevelDb
 	case "boltdb":
 		volumeNeedleMapKind = storage.NeedleMapBoltDb
+	case "btree":
+		volumeNeedleMapKind = storage.NeedleMapBtree
 	}
 	volumeServer := weed_server.NewVolumeServer(volumeMux, publicVolumeMux,
 		*serverIp, *volumePort, *volumeServerPublicUrl,

weed/command/volume.go

@@ -52,7 +52,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.indexType = cmdVolume.Flag.String("index", "memory", "Choose [memory|leveldb|boltdb] mode for memory~performance balance.")
+	v.indexType = cmdVolume.Flag.String("index", "memory", "Choose [memory|leveldb|boltdb|btree] mode for memory~performance balance.")
 	v.fixJpgOrientation = cmdVolume.Flag.Bool("images.fix.orientation", true, "Adjust jpg orientation when uploading.")
 	v.readRedirect = cmdVolume.Flag.Bool("read.redirect", true, "Redirect moved or non-local volumes.")
 	v.enableBytesCache = cmdVolume.Flag.Bool("cache.enable", false, "direct cache instead of OS cache, cost more memory.")
@@ -126,6 +126,8 @@ func runVolume(cmd *Command, args []string) bool {
 		volumeNeedleMapKind = storage.NeedleMapLevelDb
 	case "boltdb":
 		volumeNeedleMapKind = storage.NeedleMapBoltDb
+	case "btree":
+		volumeNeedleMapKind = storage.NeedleMapBtree
 	}
 	volumeServer := weed_server.NewVolumeServer(volumeMux, publicVolumeMux,
 		*v.ip, *v.port, *v.publicUrl,

weed/filer/client_operations.go

@@ -4,16 +4,16 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"net/url"
 
 	"github.com/chrislusf/seaweedfs/weed/util"
-
-	"net/url"
 )
 
 type ApiRequest struct {
-	Command   string //"listFiles", "listDirectories"
+	Command   string //"listFiles", "listDirectories", "getFileSize"
 	Directory string
 	FileName  string
+	FileId    string
 }
 
 type ListFilesResult struct {
@@ -21,16 +21,47 @@ type ListFilesResult struct {
 	Error string `json:"error,omitempty"`
 }
 
-func ListFiles(server string, directory string, fileName string) (*ListFilesResult, error) {
-	var ret ListFilesResult
-	if err := call(server, ApiRequest{Command: "listFiles", Directory: directory, FileName: fileName}, &ret); err == nil {
+func ListFiles(server string, directory string, fileName string) (ret *ListFilesResult, err error) {
+	ret = new(ListFilesResult)
+	if err = call(server, ApiRequest{Command: "listFiles", Directory: directory, FileName: fileName}, ret); err == nil {
 		if ret.Error != "" {
 			return nil, errors.New(ret.Error)
 		}
-		return &ret, nil
-	} else {
-		return nil, err
+		return ret, nil
 	}
+	return nil, err
+}
+
+type GetFileSizeResult struct {
+	Size  uint64
+	Error string `json:"error,omitempty"`
+}
+
+func GetFileSize(server string, fileId string) (ret *GetFileSizeResult, err error) {
+	ret = new(GetFileSizeResult)
+	if err = call(server, ApiRequest{Command: "getFileSize", FileId: fileId}, ret); err == nil {
+		if ret.Error != "" {
+			return nil, errors.New(ret.Error)
+		}
+		return ret, nil
+	}
+	return nil, err
+}
+
+type GetFileContentResult struct {
+	Content []byte
+	Error   string `json:"error,omitempty"`
+}
+
+func GetFileContent(server string, fileId string) (ret *GetFileContentResult, err error) {
+	ret = new(GetFileContentResult)
+	if err = call(server, ApiRequest{Command: "getFileContent", FileId: fileId}, ret); err == nil {
+		if ret.Error != "" {
+			return nil, errors.New(ret.Error)
+		}
+		return ret, nil
+	}
+	return nil, err
 }
 
 type ListDirectoriesResult struct {
@@ -38,16 +69,23 @@ type ListDirectoriesResult struct {
 	Error       string `json:"error,omitempty"`
 }
 
-func ListDirectories(server string, directory string) (*ListDirectoriesResult, error) {
-	var ret ListDirectoriesResult
-	if err := call(server, ApiRequest{Command: "listDirectories", Directory: directory}, &ret); err == nil {
+func ListDirectories(server string, directory string) (ret *ListDirectoriesResult, err error) {
+	ret = new(ListDirectoriesResult)
+	if err := call(server, ApiRequest{Command: "listDirectories", Directory: directory}, ret); err == nil {
 		if ret.Error != "" {
 			return nil, errors.New(ret.Error)
 		}
-		return &ret, nil
-	} else {
-		return nil, err
+		return ret, nil
 	}
+	return nil, err
+}
+
+func DeleteDirectoryOrFile(server string, path string, isDir bool) error {
+	destUrl := fmt.Sprintf("http://%s%s", server, path)
+	if isDir {
+		destUrl += "/?recursive=true"
+	}
+	return util.Delete(destUrl, "")
 }
 
 func call(server string, request ApiRequest, ret interface{}) error {

weed/filer/embedded_filer/filer_embedded.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"path/filepath"
 	"strings"
+	"sync"
 
 	"github.com/chrislusf/seaweedfs/weed/filer"
 	"github.com/chrislusf/seaweedfs/weed/operation"
@@ -14,6 +15,7 @@ type FilerEmbedded struct {
 	master      string
 	directories *DirectoryManagerInMap
 	files       *FileListInLevelDb
+	mvMutex     sync.Mutex
 }
 
 func NewFilerEmbedded(master string, dir string) (filer *FilerEmbedded, err error) {
@@ -121,6 +123,9 @@ mv fromFile toDir
 mv fromFile toFile
 */
 func (filer *FilerEmbedded) Move(fromPath string, toPath string) error {
+	filer.mvMutex.Lock()
+	defer filer.mvMutex.Unlock()
+
 	if _, dir_err := filer.FindDirectory(fromPath); dir_err == nil {
 		if _, err := filer.FindDirectory(toPath); err == nil {
 			// move folder under an existing folder

weed/filer/mysql_store/mysql_store.go

@@ -184,7 +184,7 @@ func (s *MySqlStore) Close() {
 var createTable = `
 CREATE TABLE IF NOT EXISTS %s (
   id bigint(20) NOT NULL AUTO_INCREMENT,
-  uriPath char(256) NOT NULL DEFAULT "" COMMENT 'http uriPath',
+  uriPath char(255) NOT NULL DEFAULT "" COMMENT 'http uriPath',
   fid char(36) NOT NULL DEFAULT "" COMMENT 'seaweedfs fid',
   createTime int(10) NOT NULL DEFAULT 0 COMMENT 'createdTime in unix timestamp',
   updateTime int(10) NOT NULL DEFAULT 0 COMMENT 'updatedTime in unix timestamp',

weed/images/preprocess.go

@@ -18,10 +18,10 @@ func MaybePreprocessImage(filename string, data []byte, width, height int) (resi
 	ext = strings.ToLower(ext)
 	switch ext {
 	case ".png", ".gif":
-		return Resized(ext, data, width, height)
+		return Resized(ext, data, width, height, "")
 	case ".jpg", ".jpeg":
 		data = FixJpgOrientation(data)
-		return Resized(ext, data, width, height)
+		return Resized(ext, data, width, height, "")
 	}
 	return data, 0, 0
 }

weed/images/resizing.go

@@ -11,7 +11,7 @@ import (
 	"github.com/disintegration/imaging"
 )
 
-func Resized(ext string, data []byte, width, height int) (resized []byte, w int, h int) {
+func Resized(ext string, data []byte, width, height int, mode string) (resized []byte, w int, h int) {
 	if width == 0 && height == 0 {
 		return data, 0, 0
 	}
@@ -20,11 +20,18 @@ func Resized(ext string, data []byte, width, height int) (resized []byte, w int,
 		bounds := srcImage.Bounds()
 		var dstImage *image.NRGBA
 		if bounds.Dx() > width && width != 0 || bounds.Dy() > height && height != 0 {
-			if width == height && bounds.Dx() != bounds.Dy() {
-				dstImage = imaging.Thumbnail(srcImage, width, height, imaging.Lanczos)
-				w, h = width, height
-			} else {
-				dstImage = imaging.Resize(srcImage, width, height, imaging.Lanczos)
+			switch mode {
+			case "fit":
+				dstImage = imaging.Fit(srcImage, width, height, imaging.Lanczos)
+			case "fill":
+				dstImage = imaging.Fill(srcImage, width, height, imaging.Center, imaging.Lanczos)
+			default:
+				if width == height && bounds.Dx() != bounds.Dy() {
+					dstImage = imaging.Thumbnail(srcImage, width, height, imaging.Lanczos)
+					w, h = width, height
+				} else {
+					dstImage = imaging.Resize(srcImage, width, height, imaging.Lanczos)
+				}
 			}
 		} else {
 			return data, bounds.Dx(), bounds.Dy()

weed/server/filer_server.go

@@ -57,7 +57,7 @@ type FilerServer struct {
 	masterNodes        *storage.MasterNodes
 }
 
-func NewFilerServer(r *http.ServeMux, ip string, port int, master string, dir string, collection string,
+func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, ip string, port int, master string, dir string, collection string,
 	replication string, redirectOnRead bool, disableDirListing bool,
 	confFile string,
 	maxMB int,
@@ -105,11 +105,15 @@ func NewFilerServer(r *http.ServeMux, ip string, port int, master string, dir st
 			return
 		}
 
-		r.HandleFunc("/admin/mv", fs.moveHandler)
-		r.HandleFunc("/admin/register", fs.registerHandler)
+		defaultMux.HandleFunc("/admin/mv", fs.moveHandler)
+		defaultMux.HandleFunc("/admin/register", fs.registerHandler)
+		defaultMux.HandleFunc("/__api__", fs.apiHandler)
 	}
 
-	r.HandleFunc("/", fs.filerHandler)
+	defaultMux.HandleFunc("/", fs.filerHandler)
+	if defaultMux != readonlyMux {
+		readonlyMux.HandleFunc("/", fs.readonlyFilerHandler)
+	}
 
 	go func() {
 		connected := true

weed/server/filer_server_handlers.go

@@ -18,3 +18,12 @@ func (fs *FilerServer) filerHandler(w http.ResponseWriter, r *http.Request) {
 		fs.PostHandler(w, r)
 	}
 }
+
+func (fs *FilerServer) readonlyFilerHandler(w http.ResponseWriter, r *http.Request) {
+	switch r.Method {
+	case "GET":
+		fs.GetOrHeadHandler(w, r, true)
+	case "HEAD":
+		fs.GetOrHeadHandler(w, r, false)
+	}
+}

weed/server/filer_server_handlers_api.go

@@ -0,0 +1,73 @@
+package weed_server
+
+import (
+	"encoding/json"
+	"net/http"
+	"strconv"
+
+	"github.com/chrislusf/seaweedfs/weed/filer"
+	"github.com/chrislusf/seaweedfs/weed/glog"
+	"github.com/chrislusf/seaweedfs/weed/operation"
+	"github.com/chrislusf/seaweedfs/weed/util"
+)
+
+func (fs *FilerServer) apiHandler(w http.ResponseWriter, r *http.Request) {
+	request := r.FormValue("request")
+	apiRequest := filer.ApiRequest{}
+
+	err := json.Unmarshal([]byte(request), &apiRequest)
+	if err != nil {
+		glog.V(0).Infoln("failing to read request", r.RequestURI, request)
+		writeJsonError(w, r, http.StatusInternalServerError, err)
+	}
+	switch apiRequest.Command {
+	case "listDirectories":
+		res := filer.ListDirectoriesResult{}
+		res.Directories, err = fs.filer.ListDirectories(apiRequest.Directory)
+		if err != nil {
+			res.Error = err.Error()
+		}
+		writeJsonQuiet(w, r, http.StatusOK, res)
+	case "listFiles":
+		res := filer.ListFilesResult{}
+		res.Files, err = fs.filer.ListFiles(apiRequest.Directory, apiRequest.FileName, 100)
+		if err != nil {
+			res.Error = err.Error()
+		}
+		writeJsonQuiet(w, r, http.StatusOK, res)
+	case "getFileSize":
+		res := filer.GetFileSizeResult{}
+		server, err := operation.LookupFileId(fs.getMasterNode(), apiRequest.FileId)
+		if err != nil {
+			res.Error = err.Error()
+			writeJsonQuiet(w, r, http.StatusOK, res)
+			return
+		}
+		head, err := util.Head(server)
+		if err != nil {
+			res.Error = err.Error()
+			writeJsonQuiet(w, r, http.StatusOK, res)
+			return
+		}
+		res.Size, err = strconv.ParseUint(head.Get("Content-Length"), 10, 0)
+		if err != nil {
+			res.Error = err.Error()
+		}
+		writeJsonQuiet(w, r, http.StatusOK, res)
+	case "getFileContent":
+		res := filer.GetFileContentResult{}
+		server, err := operation.LookupFileId(fs.getMasterNode(), apiRequest.FileId)
+		if err != nil {
+			res.Error = err.Error()
+			writeJsonQuiet(w, r, http.StatusOK, res)
+			return
+		}
+		res.Content, err = util.Get(server)
+		if err != nil {
+			res.Error = err.Error()
+			writeJsonQuiet(w, r, http.StatusOK, res)
+			return
+		}
+		writeJsonQuiet(w, r, http.StatusOK, res)
+	}
+}

weed/server/filer_server_handlers_write.go

@@ -532,7 +532,7 @@ func (fs *FilerServer) doUpload(urlLocation string, w http.ResponseWriter, r *ht
 }
 
 // curl -X DELETE http://localhost:8888/path/to
-// curl -X DELETE http://localhost:8888/path/to?recursive=true
+// curl -X DELETE http://localhost:8888/path/to/?recursive=true
 func (fs *FilerServer) DeleteHandler(w http.ResponseWriter, r *http.Request) {
 	var err error
 	var fid string

weed/server/master_grpc_server.go

@@ -59,23 +59,21 @@ func (ms MasterServer) SendHeartbeat(stream pb.Seaweed_SendHeartbeatServer) erro
 			}
 
 		} else {
-			glog.V(0).Infof("lost volume server %s:%d", dn.Ip, dn.Port)
 			if dn != nil {
+				glog.V(0).Infof("lost volume server %s:%d", dn.Ip, dn.Port)
 				t.UnRegisterDataNode(dn)
 			}
 			return err
 		}
 
-		if !t.IsLeader() {
-			newLeader, err := t.Leader()
-			if err == nil {
-				if err := stream.Send(&pb.HeartbeatResponse{
-					Leader: newLeader,
-				}); err != nil {
-					return err
-				}
+		// tell the volume servers about the leader
+		newLeader, err := t.Leader()
+		if err == nil {
+			if err := stream.Send(&pb.HeartbeatResponse{
+				Leader: newLeader,
+			}); err != nil {
+				return err
 			}
 		}
-
 	}
 }

weed/server/master_server.go

@@ -37,7 +37,6 @@ func NewMasterServer(r *mux.Router, port int, metaFolder string,
 	volumeSizeLimitMB uint,
 	preallocate bool,
 	pulseSeconds int,
-	confFile string,
 	defaultReplicaPlacement string,
 	garbageThreshold string,
 	whiteList []string,
@@ -58,11 +57,7 @@ func NewMasterServer(r *mux.Router, port int, metaFolder string,
 	}
 	ms.bounedLeaderChan = make(chan int, 16)
 	seq := sequence.NewMemorySequencer()
-	var e error
-	if ms.Topo, e = topology.NewTopology("topo", confFile, seq,
-		uint64(volumeSizeLimitMB)*1024*1024, pulseSeconds); e != nil {
-		glog.Fatalf("cannot create topology:%s", e)
-	}
+	ms.Topo = topology.NewTopology("topo", seq, uint64(volumeSizeLimitMB)*1024*1024, pulseSeconds)
 	ms.vg = topology.NewDefaultVolumeGrowth()
 	glog.V(0).Infoln("Volume Size Limit is", volumeSizeLimitMB, "MB")
 

weed/server/raft_server.go

@@ -65,8 +65,8 @@ func NewRaftServer(r *mux.Router, peers []string, httpAddr string, dataDir strin
 		return nil
 	}
 	transporter.Install(s.raftServer, s)
-	s.raftServer.SetHeartbeatInterval(1 * time.Second)
-	s.raftServer.SetElectionTimeout(time.Duration(pulseSeconds) * 3450 * time.Millisecond)
+	s.raftServer.SetHeartbeatInterval(500 * time.Millisecond)
+	s.raftServer.SetElectionTimeout(time.Duration(pulseSeconds) * 500 * time.Millisecond)
 	s.raftServer.Start()
 
 	s.router.HandleFunc("/cluster/join", s.joinHandler).Methods("POST")

weed/server/volume_grpc_client.go

@@ -63,8 +63,12 @@ func (vs *VolumeServer) doHeartbeat(sleepInterval time.Duration) error {
 				doneChan <- err
 				return
 			}
-			vs.store.VolumeSizeLimit = in.GetVolumeSizeLimit()
-			vs.guard.SecretKey = security.Secret(in.GetSecretKey())
+			if in.GetVolumeSizeLimit() != 0 {
+				vs.store.VolumeSizeLimit = in.GetVolumeSizeLimit()
+			}
+			if in.GetSecretKey() != "" {
+				vs.guard.SecretKey = security.Secret(in.GetSecretKey())
+			}
 			if in.GetLeader() != "" && masterNode != in.GetLeader() {
 				vs.masterNodes.SetPossibleLeader(in.GetLeader())
 				doneChan <- nil
@@ -78,7 +82,7 @@ func (vs *VolumeServer) doHeartbeat(sleepInterval time.Duration) error {
 		return err
 	}
 
-	tickChan := time.NewTimer(sleepInterval).C
+	tickChan := time.Tick(sleepInterval)
 
 	for {
 		select {
@@ -88,6 +92,7 @@ func (vs *VolumeServer) doHeartbeat(sleepInterval time.Duration) error {
 				return err
 			}
 		case err := <-doneChan:
+			glog.V(0).Infof("Volume Server heart beat stops with %v", err)
 			return err
 		}
 	}

weed/server/volume_server.go

@@ -58,6 +58,9 @@ func NewVolumeServer(adminMux, publicMux *http.ServeMux, ip string,
 	adminMux.HandleFunc("/admin/sync/status", vs.guard.WhiteList(vs.getVolumeSyncStatusHandler))
 	adminMux.HandleFunc("/admin/sync/index", vs.guard.WhiteList(vs.getVolumeIndexContentHandler))
 	adminMux.HandleFunc("/admin/sync/data", vs.guard.WhiteList(vs.getVolumeDataContentHandler))
+	adminMux.HandleFunc("/admin/volume/mount", vs.guard.WhiteList(vs.getVolumeMountHandler))
+	adminMux.HandleFunc("/admin/volume/unmount", vs.guard.WhiteList(vs.getVolumeUnmountHandler))
+	adminMux.HandleFunc("/admin/volume/delete", vs.guard.WhiteList(vs.getVolumeDeleteHandler))
 	adminMux.HandleFunc("/stats/counter", vs.guard.WhiteList(statsCounterHandler))
 	adminMux.HandleFunc("/stats/memory", vs.guard.WhiteList(statsMemoryHandler))
 	adminMux.HandleFunc("/stats/disk", vs.guard.WhiteList(vs.statsDiskHandler))

weed/server/volume_server_handlers_admin.go

@@ -1,6 +1,7 @@
 package weed_server
 
 import (
+	"fmt"
 	"net/http"
 	"path/filepath"
 	"strconv"
@@ -8,6 +9,7 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/stats"
 	"github.com/chrislusf/seaweedfs/weed/util"
+	"github.com/chrislusf/seaweedfs/weed/storage"
 )
 
 func (vs *VolumeServer) statusHandler(w http.ResponseWriter, r *http.Request) {
@@ -65,3 +67,45 @@ func (vs *VolumeServer) statsDiskHandler(w http.ResponseWriter, r *http.Request)
 	m["DiskStatuses"] = ds
 	writeJsonQuiet(w, r, http.StatusOK, m)
 }
+
+func (vs *VolumeServer) getVolume(volumeParameterName string, r *http.Request) (*storage.Volume, error) {
+	vid, err := vs.getVolumeId(volumeParameterName, r)
+	if err != nil {
+		return nil, err
+	}
+	v := vs.store.GetVolume(vid)
+	if v == nil {
+		return nil, fmt.Errorf("Not Found Volume Id %d", vid)
+	}
+	return v, nil
+}
+
+func (vs *VolumeServer) getVolumeMountHandler(w http.ResponseWriter, r *http.Request) {
+	vid, err := vs.getVolumeId("volume", r)
+	if err != nil {
+		writeJsonError(w, r, http.StatusNotFound, err)
+		return
+	}
+	vs.store.MountVolume(vid)
+	writeJsonQuiet(w, r, http.StatusOK, "Volume mounted")
+}
+
+func (vs *VolumeServer) getVolumeUnmountHandler(w http.ResponseWriter, r *http.Request) {
+	vid, err := vs.getVolumeId("volume", r)
+	if err != nil {
+		writeJsonError(w, r, http.StatusNotFound, err)
+		return
+	}
+	vs.store.UnmountVolume(vid)
+	writeJsonQuiet(w, r, http.StatusOK, "Volume unmounted")
+}
+
+func (vs *VolumeServer) getVolumeDeleteHandler(w http.ResponseWriter, r *http.Request) {
+	vid, err := vs.getVolumeId("volume", r)
+	if err != nil {
+		writeJsonError(w, r, http.StatusNotFound, err)
+		return
+	}
+	vs.store.DeleteVolume(vid)
+	writeJsonQuiet(w, r, http.StatusOK, "Volume deleted")
+}

weed/server/volume_server_handlers_read.go

@@ -67,7 +67,7 @@ func (vs *VolumeServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request)
 	cookie := n.Cookie
 	count, e := vs.store.ReadVolumeNeedle(volumeId, n)
 	glog.V(4).Infoln("read bytes", count, "error", e)
-	if e != nil || count <= 0 {
+	if e != nil || count < 0 {
 		glog.V(0).Infoln("read error:", e, r.URL.Path)
 		w.WriteHeader(http.StatusNotFound)
 		return
@@ -144,7 +144,7 @@ func (vs *VolumeServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request)
 		if r.FormValue("height") != "" {
 			height, _ = strconv.Atoi(r.FormValue("height"))
 		}
-		n.Data, _, _ = images.Resized(ext, n.Data, width, height)
+		n.Data, _, _ = images.Resized(ext, n.Data, width, height, r.FormValue("mode"))
 	}
 
 	if e := writeResponseContent(filename, mtype, bytes.NewReader(n.Data), w, r); e != nil {
@@ -165,7 +165,7 @@ func (vs *VolumeServer) FaviconHandler(w http.ResponseWriter, r *http.Request) {
 }
 
 func (vs *VolumeServer) tryHandleChunkedFile(n *storage.Needle, fileName string, w http.ResponseWriter, r *http.Request) (processed bool) {
-	if !n.IsChunkedManifest() {
+	if !n.IsChunkedManifest() || r.URL.Query().Get("cm") == "false" {
 		return false
 	}
 

weed/server/volume_server_handlers_sync.go

@@ -68,20 +68,19 @@ func (vs *VolumeServer) getVolumeDataContentHandler(w http.ResponseWriter, r *ht
 	w.Write(content)
 }
 
-func (vs *VolumeServer) getVolume(volumeParameterName string, r *http.Request) (*storage.Volume, error) {
+func (vs *VolumeServer) getVolumeId(volumeParameterName string, r *http.Request) (storage.VolumeId, error) {
 	volumeIdString := r.FormValue(volumeParameterName)
+
 	if volumeIdString == "" {
 		err := fmt.Errorf("Empty Volume Id: Need to pass in %s=the_volume_id.", volumeParameterName)
-		return nil, err
+		return 0, err
 	}
+
 	vid, err := storage.NewVolumeId(volumeIdString)
 	if err != nil {
 		err = fmt.Errorf("Volume Id %s is not a valid unsigned integer", volumeIdString)
-		return nil, err
+		return 0, err
 	}
-	v := vs.store.GetVolume(vid)
-	if v == nil {
-		return nil, fmt.Errorf("Not Found Volume Id %s: %d", volumeIdString, vid)
-	}
-	return v, nil
-}
+
+	return vid, err
+}

weed/storage/disk_location.go

@@ -6,6 +6,8 @@ import (
 	"strings"
 	"sync"
 
+	"fmt"
+
 	"github.com/chrislusf/seaweedfs/weed/glog"
 )
 
@@ -22,7 +24,7 @@ func NewDiskLocation(dir string, maxVolumeCount int) *DiskLocation {
 	return location
 }
 
-func (l *DiskLocation) loadExistingVolume(dir os.FileInfo, needleMapKind NeedleMapType, mutex *sync.RWMutex) {
+func (l *DiskLocation) volumeIdFromPath(dir os.FileInfo) (VolumeId, string, error) {
 	name := dir.Name()
 	if !dir.IsDir() && strings.HasSuffix(name, ".dat") {
 		collection := ""
@@ -31,7 +33,18 @@ func (l *DiskLocation) loadExistingVolume(dir os.FileInfo, needleMapKind NeedleM
 		if i > 0 {
 			collection, base = base[0:i], base[i+1:]
 		}
-		if vid, err := NewVolumeId(base); err == nil {
+		vol, err := NewVolumeId(base)
+		return vol, collection, err
+	}
+
+	return 0, "", fmt.Errorf("Path is not a volume: %s", name)
+}
+
+func (l *DiskLocation) loadExistingVolume(dir os.FileInfo, needleMapKind NeedleMapType, mutex *sync.RWMutex) {
+	name := dir.Name()
+	if !dir.IsDir() && strings.HasSuffix(name, ".dat") {
+		vid, collection, err := l.volumeIdFromPath(dir)
+		if err == nil {
 			mutex.RLock()
 			_, found := l.volumes[vid]
 			mutex.RUnlock()
@@ -42,10 +55,6 @@ func (l *DiskLocation) loadExistingVolume(dir os.FileInfo, needleMapKind NeedleM
 					mutex.Unlock()
 					glog.V(0).Infof("data file %s, replicaPlacement=%s v=%d size=%d ttl=%s",
 						l.Directory+"/"+name, v.ReplicaPlacement, v.Version(), v.Size(), v.Ttl.String())
-					if v.Size() != v.dataFileSize {
-						glog.V(0).Infof("data file %s, size=%d expected=%d",
-							l.Directory+"/"+name, v.Size(), v.dataFileSize)
-					}
 				} else {
 					glog.V(0).Infof("new volume %s error %s", name, e)
 				}
@@ -125,6 +134,44 @@ func (l *DiskLocation) deleteVolumeById(vid VolumeId) (e error) {
 	return
 }
 
+func (l *DiskLocation) LoadVolume(vid VolumeId, needleMapKind NeedleMapType) bool {
+	if dirs, err := ioutil.ReadDir(l.Directory); err == nil {
+		for _, dir := range dirs {
+			volId, _, err := l.volumeIdFromPath(dir)
+			if vid == volId && err == nil {
+				var mutex sync.RWMutex
+				l.loadExistingVolume(dir, needleMapKind, &mutex)
+				return true
+			}
+		}
+	}
+
+	return false
+}
+
+func (l *DiskLocation) DeleteVolume(vid VolumeId) error {
+	l.Lock()
+	defer l.Unlock()
+
+	_, ok := l.volumes[vid]
+	if !ok {
+		return fmt.Errorf("Volume not found, VolumeId: %d", vid)
+	}
+	return l.deleteVolumeById(vid)
+}
+
+func (l *DiskLocation) UnloadVolume(vid VolumeId) error {
+	l.Lock()
+	defer l.Unlock()
+
+	_, ok := l.volumes[vid]
+	if !ok {
+		return fmt.Errorf("Volume not loaded, VolumeId: %d", vid)
+	}
+	delete(l.volumes, vid)
+	return nil
+}
+
 func (l *DiskLocation) SetVolume(vid VolumeId, volume *Volume) {
 	l.Lock()
 	defer l.Unlock()

weed/storage/needle/btree_map.go

@@ -0,0 +1,52 @@
+package needle
+
+import (
+	"github.com/google/btree"
+)
+
+//This map assumes mostly inserting increasing keys
+type BtreeMap struct {
+	tree *btree.BTree
+}
+
+func NewBtreeMap() *BtreeMap {
+	return &BtreeMap{
+		tree: btree.New(32),
+	}
+}
+
+func (cm *BtreeMap) Set(key Key, offset, size uint32) (oldOffset, oldSize uint32) {
+	found := cm.tree.ReplaceOrInsert(NeedleValue{key, offset, size})
+	if found != nil {
+		old := found.(NeedleValue)
+		return old.Offset, old.Size
+	}
+	return
+}
+
+func (cm *BtreeMap) Delete(key Key) (oldSize uint32) {
+	found := cm.tree.Delete(NeedleValue{key, 0, 0})
+	if found != nil {
+		old := found.(NeedleValue)
+		return old.Size
+	}
+	return
+}
+func (cm *BtreeMap) Get(key Key) (*NeedleValue, bool) {
+	found := cm.tree.Get(NeedleValue{key, 0, 0})
+	if found != nil {
+		old := found.(NeedleValue)
+		return &old, true
+	}
+	return nil, false
+}
+
+// Visit visits all entries or stop if any error when visiting
+func (cm *BtreeMap) Visit(visit func(NeedleValue) error) (ret error) {
+	cm.tree.Ascend(func(item btree.Item) bool {
+		needle := item.(NeedleValue)
+		ret = visit(needle)
+		return ret == nil
+	})
+	return ret
+}

weed/storage/needle/compact_map.go

@@ -0,0 +1,194 @@
+package needle
+
+import (
+	"sync"
+)
+
+type CompactSection struct {
+	sync.RWMutex
+	values   []NeedleValue
+	overflow map[Key]NeedleValue
+	start    Key
+	end      Key
+	counter  int
+}
+
+func NewCompactSection(start Key) *CompactSection {
+	return &CompactSection{
+		values:   make([]NeedleValue, batch),
+		overflow: make(map[Key]NeedleValue),
+		start:    start,
+	}
+}
+
+//return old entry size
+func (cs *CompactSection) Set(key Key, offset, size uint32) (oldOffset, oldSize uint32) {
+	cs.Lock()
+	if key > cs.end {
+		cs.end = key
+	}
+	if i := cs.binarySearchValues(key); i >= 0 {
+		oldOffset, oldSize = cs.values[i].Offset, cs.values[i].Size
+		//println("key", key, "old size", ret)
+		cs.values[i].Offset, cs.values[i].Size = offset, size
+	} else {
+		needOverflow := cs.counter >= batch
+		needOverflow = needOverflow || cs.counter > 0 && cs.values[cs.counter-1].Key > key
+		if needOverflow {
+			//println("start", cs.start, "counter", cs.counter, "key", key)
+			if oldValue, found := cs.overflow[key]; found {
+				oldOffset, oldSize = oldValue.Offset, oldValue.Size
+			}
+			cs.overflow[key] = NeedleValue{Key: key, Offset: offset, Size: size}
+		} else {
+			p := &cs.values[cs.counter]
+			p.Key, p.Offset, p.Size = key, offset, size
+			//println("added index", cs.counter, "key", key, cs.values[cs.counter].Key)
+			cs.counter++
+		}
+	}
+	cs.Unlock()
+	return
+}
+
+//return old entry size
+func (cs *CompactSection) Delete(key Key) uint32 {
+	cs.Lock()
+	ret := uint32(0)
+	if i := cs.binarySearchValues(key); i >= 0 {
+		if cs.values[i].Size > 0 {
+			ret = cs.values[i].Size
+			cs.values[i].Size = 0
+		}
+	}
+	if v, found := cs.overflow[key]; found {
+		delete(cs.overflow, key)
+		ret = v.Size
+	}
+	cs.Unlock()
+	return ret
+}
+func (cs *CompactSection) Get(key Key) (*NeedleValue, bool) {
+	cs.RLock()
+	if v, ok := cs.overflow[key]; ok {
+		cs.RUnlock()
+		return &v, true
+	}
+	if i := cs.binarySearchValues(key); i >= 0 {
+		cs.RUnlock()
+		return &cs.values[i], true
+	}
+	cs.RUnlock()
+	return nil, false
+}
+func (cs *CompactSection) binarySearchValues(key Key) int {
+	l, h := 0, cs.counter-1
+	if h >= 0 && cs.values[h].Key < key {
+		return -2
+	}
+	//println("looking for key", key)
+	for l <= h {
+		m := (l + h) / 2
+		//println("mid", m, "key", cs.values[m].Key, cs.values[m].Offset, cs.values[m].Size)
+		if cs.values[m].Key < key {
+			l = m + 1
+		} else if key < cs.values[m].Key {
+			h = m - 1
+		} else {
+			//println("found", m)
+			return m
+		}
+	}
+	return -1
+}
+
+//This map assumes mostly inserting increasing keys
+//This map assumes mostly inserting increasing keys
+type CompactMap struct {
+	list []*CompactSection
+}
+
+func NewCompactMap() *CompactMap {
+	return &CompactMap{}
+}
+
+func (cm *CompactMap) Set(key Key, offset, size uint32) (oldOffset, oldSize uint32) {
+	x := cm.binarySearchCompactSection(key)
+	if x < 0 {
+		//println(x, "creating", len(cm.list), "section, starting", key)
+		cm.list = append(cm.list, NewCompactSection(key))
+		x = len(cm.list) - 1
+		//keep compact section sorted by start
+		for x > 0 {
+			if cm.list[x-1].start > cm.list[x].start {
+				cm.list[x-1], cm.list[x] = cm.list[x], cm.list[x-1]
+				x = x - 1
+			} else {
+				break
+			}
+		}
+	}
+	return cm.list[x].Set(key, offset, size)
+}
+func (cm *CompactMap) Delete(key Key) uint32 {
+	x := cm.binarySearchCompactSection(key)
+	if x < 0 {
+		return uint32(0)
+	}
+	return cm.list[x].Delete(key)
+}
+func (cm *CompactMap) Get(key Key) (*NeedleValue, bool) {
+	x := cm.binarySearchCompactSection(key)
+	if x < 0 {
+		return nil, false
+	}
+	return cm.list[x].Get(key)
+}
+func (cm *CompactMap) binarySearchCompactSection(key Key) int {
+	l, h := 0, len(cm.list)-1
+	if h < 0 {
+		return -5
+	}
+	if cm.list[h].start <= key {
+		if cm.list[h].counter < batch || key <= cm.list[h].end {
+			return h
+		}
+		return -4
+	}
+	for l <= h {
+		m := (l + h) / 2
+		if key < cm.list[m].start {
+			h = m - 1
+		} else { // cm.list[m].start <= key
+			if cm.list[m+1].start <= key {
+				l = m + 1
+			} else {
+				return m
+			}
+		}
+	}
+	return -3
+}
+
+// Visit visits all entries or stop if any error when visiting
+func (cm *CompactMap) Visit(visit func(NeedleValue) error) error {
+	for _, cs := range cm.list {
+		cs.RLock()
+		for _, v := range cs.overflow {
+			if err := visit(v); err != nil {
+				cs.RUnlock()
+				return err
+			}
+		}
+		for _, v := range cs.values {
+			if _, found := cs.overflow[v.Key]; !found {
+				if err := visit(v); err != nil {
+					cs.RUnlock()
+					return err
+				}
+			}
+		}
+		cs.RUnlock()
+	}
+	return nil
+}

weed/storage/needle/compact_map_perf_test.go

@@ -1,4 +1,4 @@
-package storage
+package needle
 
 import (
 	"log"
@@ -11,15 +11,15 @@ import (
 
 func TestMemoryUsage(t *testing.T) {
 
-	indexFile, ie := os.OpenFile("../../test/sample.idx", os.O_RDWR|os.O_RDONLY, 0644)
+	indexFile, ie := os.OpenFile("../../../test/sample.idx", os.O_RDWR|os.O_RDONLY, 0644)
 	if ie != nil {
 		log.Fatalln(ie)
 	}
-	LoadNewNeedleMap(indexFile)
+	loadNewNeedleMap(indexFile)
 
 }
 
-func LoadNewNeedleMap(file *os.File) CompactMap {
+func loadNewNeedleMap(file *os.File) {
 	m := NewCompactMap()
 	bytes := make([]byte, 16*1024)
 	count, e := file.Read(bytes)
@@ -41,5 +41,4 @@ func LoadNewNeedleMap(file *os.File) CompactMap {
 
 		count, e = file.Read(bytes)
 	}
-	return m
 }

weed/storage/needle/compact_map_test.go

@@ -1,4 +1,4 @@
-package storage
+package needle
 
 import (
 	"testing"

weed/storage/needle/needle_value.go

@@ -0,0 +1,28 @@
+package needle
+
+import (
+	"strconv"
+
+	"github.com/google/btree"
+)
+
+const (
+	batch = 100000
+)
+
+type NeedleValue struct {
+	Key    Key
+	Offset uint32 `comment:"Volume offset"` //since aligned to 8 bytes, range is 4G*8=32G
+	Size   uint32 `comment:"Size of the data portion"`
+}
+
+func (this NeedleValue) Less(than btree.Item) bool {
+	that := than.(NeedleValue)
+	return this.Key < that.Key
+}
+
+type Key uint64
+
+func (k Key) String() string {
+	return strconv.FormatUint(uint64(k), 10)
+}

weed/storage/needle/needle_value_map.go

@@ -0,0 +1,8 @@
+package needle
+
+type NeedleValueMap interface {
+	Set(key Key, offset, size uint32) (oldOffset, oldSize uint32)
+	Delete(key Key) uint32
+	Get(key Key) (*NeedleValue, bool)
+	Visit(visit func(NeedleValue) error) error
+}

weed/storage/needle_map.go

@@ -6,6 +6,7 @@ import (
 	"os"
 	"sync"
 
+	"github.com/chrislusf/seaweedfs/weed/storage/needle"
 	"github.com/chrislusf/seaweedfs/weed/util"
 )
 
@@ -15,6 +16,7 @@ const (
 	NeedleMapInMemory NeedleMapType = iota
 	NeedleMapLevelDb
 	NeedleMapBoltDb
+	NeedleMapBtree
 )
 
 const (
@@ -23,7 +25,7 @@ const (
 
 type NeedleMapper interface {
 	Put(key uint64, offset uint32, size uint32) error
-	Get(key uint64) (element *NeedleValue, ok bool)
+	Get(key uint64) (element *needle.NeedleValue, ok bool)
 	Delete(key uint64, offset uint32) error
 	Close()
 	Destroy() error

weed/storage/needle_map_boltdb.go

@@ -7,6 +7,7 @@ import (
 	"github.com/boltdb/bolt"
 
 	"github.com/chrislusf/seaweedfs/weed/glog"
+	"github.com/chrislusf/seaweedfs/weed/storage/needle"
 	"github.com/chrislusf/seaweedfs/weed/util"
 )
 
@@ -31,7 +32,7 @@ func NewBoltDbNeedleMap(dbFileName string, indexFile *os.File) (m *BoltDbNeedleM
 		return
 	}
 	glog.V(1).Infof("Loading %s...", indexFile.Name())
-	nm, indexLoadError := LoadNeedleMap(indexFile)
+	nm, indexLoadError := LoadBtreeNeedleMap(indexFile)
 	if indexLoadError != nil {
 		return nil, indexLoadError
 	}
@@ -72,7 +73,7 @@ func generateBoltDbFile(dbFileName string, indexFile *os.File) error {
 	})
 }
 
-func (m *BoltDbNeedleMap) Get(key uint64) (element *NeedleValue, ok bool) {
+func (m *BoltDbNeedleMap) Get(key uint64) (element *needle.NeedleValue, ok bool) {
 	bytes := make([]byte, 8)
 	var data []byte
 	util.Uint64toBytes(bytes, key)
@@ -91,7 +92,7 @@ func (m *BoltDbNeedleMap) Get(key uint64) (element *NeedleValue, ok bool) {
 	}
 	offset := util.BytesToUint32(data[0:4])
 	size := util.BytesToUint32(data[4:8])
-	return &NeedleValue{Key: Key(key), Offset: offset, Size: size}, true
+	return &needle.NeedleValue{Key: needle.Key(key), Offset: offset, Size: size}, true
 }
 
 func (m *BoltDbNeedleMap) Put(key uint64, offset uint32, size uint32) error {

weed/storage/needle_map_leveldb.go

@@ -6,6 +6,7 @@ import (
 	"path/filepath"
 
 	"github.com/chrislusf/seaweedfs/weed/glog"
+	"github.com/chrislusf/seaweedfs/weed/storage/needle"
 	"github.com/chrislusf/seaweedfs/weed/util"
 	"github.com/syndtr/goleveldb/leveldb"
 )
@@ -29,7 +30,7 @@ func NewLevelDbNeedleMap(dbFileName string, indexFile *os.File) (m *LevelDbNeedl
 		return
 	}
 	glog.V(1).Infof("Loading %s...", indexFile.Name())
-	nm, indexLoadError := LoadNeedleMap(indexFile)
+	nm, indexLoadError := LoadBtreeNeedleMap(indexFile)
 	if indexLoadError != nil {
 		return nil, indexLoadError
 	}
@@ -70,7 +71,7 @@ func generateLevelDbFile(dbFileName string, indexFile *os.File) error {
 	})
 }
 
-func (m *LevelDbNeedleMap) Get(key uint64) (element *NeedleValue, ok bool) {
+func (m *LevelDbNeedleMap) Get(key uint64) (element *needle.NeedleValue, ok bool) {
 	bytes := make([]byte, 8)
 	util.Uint64toBytes(bytes, key)
 	data, err := m.db.Get(bytes, nil)
@@ -79,7 +80,7 @@ func (m *LevelDbNeedleMap) Get(key uint64) (element *NeedleValue, ok bool) {
 	}
 	offset := util.BytesToUint32(data[0:4])
 	size := util.BytesToUint32(data[4:8])
-	return &NeedleValue{Key: Key(key), Offset: offset, Size: size}, true
+	return &needle.NeedleValue{Key: needle.Key(key), Offset: offset, Size: size}, true
 }
 
 func (m *LevelDbNeedleMap) Put(key uint64, offset uint32, size uint32) error {

weed/storage/needle_map_memory.go

@@ -5,17 +5,26 @@ import (
 	"os"
 
 	"github.com/chrislusf/seaweedfs/weed/glog"
+	"github.com/chrislusf/seaweedfs/weed/storage/needle"
 )
 
 type NeedleMap struct {
-	m CompactMap
+	m needle.NeedleValueMap
 
 	baseNeedleMapper
 }
 
-func NewNeedleMap(file *os.File) *NeedleMap {
+func NewCompactNeedleMap(file *os.File) *NeedleMap {
 	nm := &NeedleMap{
-		m: NewCompactMap(),
+		m: needle.NewCompactMap(),
+	}
+	nm.indexFile = file
+	return nm
+}
+
+func NewBtreeNeedleMap(file *os.File) *NeedleMap {
+	nm := &NeedleMap{
+		m: needle.NewBtreeMap(),
 	}
 	nm.indexFile = file
 	return nm
@@ -25,8 +34,17 @@ const (
 	RowsToRead = 1024
 )
 
-func LoadNeedleMap(file *os.File) (*NeedleMap, error) {
-	nm := NewNeedleMap(file)
+func LoadCompactNeedleMap(file *os.File) (*NeedleMap, error) {
+	nm := NewCompactNeedleMap(file)
+	return doLoading(file, nm)
+}
+
+func LoadBtreeNeedleMap(file *os.File) (*NeedleMap, error) {
+	nm := NewBtreeNeedleMap(file)
+	return doLoading(file, nm)
+}
+
+func doLoading(file *os.File, nm *NeedleMap) (*NeedleMap, error) {
 	e := WalkIndexFile(file, func(key uint64, offset, size uint32) error {
 		if key > nm.MaximumFileKey {
 			nm.MaximumFileKey = key
@@ -34,14 +52,14 @@ func LoadNeedleMap(file *os.File) (*NeedleMap, error) {
 		if offset > 0 && size != TombstoneFileSize {
 			nm.FileCounter++
 			nm.FileByteCounter = nm.FileByteCounter + uint64(size)
-			oldOffset, oldSize := nm.m.Set(Key(key), offset, size)
+			oldOffset, oldSize := nm.m.Set(needle.Key(key), offset, size)
 			glog.V(3).Infoln("reading key", key, "offset", offset*NeedlePaddingSize, "size", size, "oldSize", oldSize)
 			if oldOffset > 0 && oldSize != TombstoneFileSize {
 				nm.DeletionCounter++
 				nm.DeletionByteCounter = nm.DeletionByteCounter + uint64(oldSize)
 			}
 		} else {
-			oldSize := nm.m.Delete(Key(key))
+			oldSize := nm.m.Delete(needle.Key(key))
 			glog.V(3).Infoln("removing key", key, "offset", offset*NeedlePaddingSize, "size", size, "oldSize", oldSize)
 			nm.DeletionCounter++
 			nm.DeletionByteCounter = nm.DeletionByteCounter + uint64(oldSize)
@@ -84,16 +102,16 @@ func WalkIndexFile(r *os.File, fn func(key uint64, offset, size uint32) error) e
 }
 
 func (nm *NeedleMap) Put(key uint64, offset uint32, size uint32) error {
-	_, oldSize := nm.m.Set(Key(key), offset, size)
+	_, oldSize := nm.m.Set(needle.Key(key), offset, size)
 	nm.logPut(key, oldSize, size)
 	return nm.appendToIndexFile(key, offset, size)
 }
-func (nm *NeedleMap) Get(key uint64) (element *NeedleValue, ok bool) {
-	element, ok = nm.m.Get(Key(key))
+func (nm *NeedleMap) Get(key uint64) (element *needle.NeedleValue, ok bool) {
+	element, ok = nm.m.Get(needle.Key(key))
 	return
 }
 func (nm *NeedleMap) Delete(key uint64, offset uint32) error {
-	deletedBytes := nm.m.Delete(Key(key))
+	deletedBytes := nm.m.Delete(needle.Key(key))
 	nm.logDelete(deletedBytes)
 	return nm.appendToIndexFile(key, offset, TombstoneFileSize)
 }

weed/storage/store.go

@@ -77,6 +77,7 @@ type Store struct {
 	connected       bool
 	VolumeSizeLimit uint64 //read from the master
 	Client          pb.Seaweed_SendHeartbeatClient
+	NeedleMapType   NeedleMapType
 }
 
 func (s *Store) String() (str string) {
@@ -85,7 +86,7 @@ func (s *Store) String() (str string) {
 }
 
 func NewStore(port int, ip, publicUrl string, dirnames []string, maxVolumeCounts []int, needleMapKind NeedleMapType) (s *Store) {
-	s = &Store{Port: port, Ip: ip, PublicUrl: publicUrl}
+	s = &Store{Port: port, Ip: ip, PublicUrl: publicUrl, NeedleMapType: needleMapKind}
 	s.Locations = make([]*DiskLocation, 0)
 	for i := 0; i < len(dirnames); i++ {
 		location := NewDiskLocation(dirnames[i], maxVolumeCounts[i])
@@ -261,6 +262,7 @@ func (s *Store) Close() {
 		location.Close()
 	}
 }
+
 func (s *Store) Write(i VolumeId, n *Needle) (size uint32, err error) {
 	if v := s.findVolume(i); v != nil {
 		if v.readOnly {
@@ -273,31 +275,33 @@ func (s *Store) Write(i VolumeId, n *Needle) (size uint32, err error) {
 		} else {
 			err = fmt.Errorf("Volume Size Limit %d Exceeded! Current size is %d", s.VolumeSizeLimit, v.ContentSize())
 		}
-		if s.VolumeSizeLimit < v.ContentSize()+3*uint64(size) {
-			glog.V(0).Infoln("volume", i, "size", v.ContentSize(), "will exceed limit", s.VolumeSizeLimit)
-			if s.Client != nil {
-				if e := s.Client.Send(s.CollectHeartbeat()); e != nil {
-					glog.V(0).Infoln("error when reporting size:", e)
-				}
-			}
-		}
 		return
 	}
 	glog.V(0).Infoln("volume", i, "not found!")
 	err = fmt.Errorf("Volume %d not found!", i)
 	return
 }
+
+func (s *Store) updateMaster() {
+	if s.Client != nil {
+		if e := s.Client.Send(s.CollectHeartbeat()); e != nil {
+			glog.V(0).Infoln("error when reporting size:", e)
+		}
+	}
+}
+
 func (s *Store) Delete(i VolumeId, n *Needle) (uint32, error) {
 	if v := s.findVolume(i); v != nil && !v.readOnly {
 		return v.deleteNeedle(n)
 	}
 	return 0, nil
 }
+
 func (s *Store) ReadVolumeNeedle(i VolumeId, n *Needle) (int, error) {
 	if v := s.findVolume(i); v != nil {
 		return v.readNeedle(n)
 	}
-	return 0, fmt.Errorf("Volume %v not found!", i)
+	return 0, fmt.Errorf("Volume %d not found!", i)
 }
 func (s *Store) GetVolume(i VolumeId) *Volume {
 	return s.findVolume(i)
@@ -307,3 +311,36 @@ func (s *Store) HasVolume(i VolumeId) bool {
 	v := s.findVolume(i)
 	return v != nil
 }
+
+func (s *Store) MountVolume(i VolumeId) error {
+	for _, location := range s.Locations {
+		if found := location.LoadVolume(i, s.NeedleMapType); found == true {
+			s.updateMaster()
+			return nil
+		}
+	}
+
+	return fmt.Errorf("Volume %d not found on disk", i)
+}
+
+func (s *Store) UnmountVolume(i VolumeId) error {
+	for _, location := range s.Locations {
+		if err := location.UnloadVolume(i); err == nil {
+			s.updateMaster()
+			return nil
+		}
+	}
+
+	return fmt.Errorf("Volume %d not found on disk", i)
+}
+
+func (s *Store) DeleteVolume(i VolumeId) error {
+	for _, location := range s.Locations {
+		if error := location.deleteVolumeById(i); error == nil {
+			s.updateMaster()
+			return nil
+		}
+	}
+
+	return fmt.Errorf("Volume %d not found on disk", i)
+}

weed/storage/volume.go

@@ -15,7 +15,6 @@ type Volume struct {
 	dir           string
 	Collection    string
 	dataFile      *os.File
-	dataFileSize  int64
 	nm            NeedleMapper
 	needleMapKind NeedleMapType
 	readOnly      bool

weed/storage/volume_checking.go

@@ -12,28 +12,28 @@ func getActualSize(size uint32) int64 {
 	return NeedleHeaderSize + int64(size) + NeedleChecksumSize + int64(padding)
 }
 
-func CheckVolumeDataIntegrity(v *Volume, indexFile *os.File) (int64, error) {
+func CheckVolumeDataIntegrity(v *Volume, indexFile *os.File) (error) {
 	var indexSize int64
 	var e error
 	if indexSize, e = verifyIndexFileIntegrity(indexFile); e != nil {
-		return 0, fmt.Errorf("verifyIndexFileIntegrity %s failed: %v", indexFile.Name(), e)
+		return fmt.Errorf("verifyIndexFileIntegrity %s failed: %v", indexFile.Name(), e)
 	}
 	if indexSize == 0 {
-		return int64(SuperBlockSize), nil
+		return nil
 	}
 	var lastIdxEntry []byte
 	if lastIdxEntry, e = readIndexEntryAtOffset(indexFile, indexSize-NeedleIndexSize); e != nil {
-		return 0, fmt.Errorf("readLastIndexEntry %s failed: %v", indexFile.Name(), e)
+		return fmt.Errorf("readLastIndexEntry %s failed: %v", indexFile.Name(), e)
 	}
 	key, offset, size := idxFileEntry(lastIdxEntry)
 	if offset == 0 || size == TombstoneFileSize {
-		return 0, nil
+		return nil
 	}
 	if e = verifyNeedleIntegrity(v.dataFile, v.Version(), int64(offset)*NeedlePaddingSize, key, size); e != nil {
-		return 0, fmt.Errorf("verifyNeedleIntegrity %s failed: %v", indexFile.Name(), e)
+		return fmt.Errorf("verifyNeedleIntegrity %s failed: %v", indexFile.Name(), e)
 	}
 
-	return int64(offset)*int64(NeedlePaddingSize) + getActualSize(size), nil
+	return nil
 }
 
 func verifyIndexFileIntegrity(indexFile *os.File) (indexSize int64, err error) {

weed/storage/volume_loading.go

@@ -64,26 +64,31 @@ func (v *Volume) load(alsoLoadIndex bool, createDatIfMissing bool, needleMapKind
 				return fmt.Errorf("cannot write Volume Index %s.idx: %v", fileName, e)
 			}
 		}
-		if v.dataFileSize, e = CheckVolumeDataIntegrity(v, indexFile); e != nil {
+		if e = CheckVolumeDataIntegrity(v, indexFile); e != nil {
 			v.readOnly = true
 			glog.V(0).Infof("volumeDataIntegrityChecking failed %v", e)
 		}
 		switch needleMapKind {
 		case NeedleMapInMemory:
-			glog.V(0).Infoln("loading index file", fileName+".idx", "readonly", v.readOnly)
-			if v.nm, e = LoadNeedleMap(indexFile); e != nil {
-				glog.V(0).Infof("loading index %s error: %v", fileName+".idx", e)
+			glog.V(0).Infoln("loading index", fileName+".idx", "to memory readonly", v.readOnly)
+			if v.nm, e = LoadCompactNeedleMap(indexFile); e != nil {
+				glog.V(0).Infof("loading index %s to memory error: %v", fileName+".idx", e)
 			}
 		case NeedleMapLevelDb:
-			glog.V(0).Infoln("loading leveldb file", fileName+".ldb")
+			glog.V(0).Infoln("loading leveldb", fileName+".ldb")
 			if v.nm, e = NewLevelDbNeedleMap(fileName+".ldb", indexFile); e != nil {
 				glog.V(0).Infof("loading leveldb %s error: %v", fileName+".ldb", e)
 			}
 		case NeedleMapBoltDb:
-			glog.V(0).Infoln("loading boltdb file", fileName+".bdb")
+			glog.V(0).Infoln("loading boltdb", fileName+".bdb")
 			if v.nm, e = NewBoltDbNeedleMap(fileName+".bdb", indexFile); e != nil {
 				glog.V(0).Infof("loading boltdb %s error: %v", fileName+".bdb", e)
 			}
+		case NeedleMapBtree:
+			glog.V(0).Infoln("loading index", fileName+".idx", "to btree readonly", v.readOnly)
+			if v.nm, e = LoadBtreeNeedleMap(indexFile); e != nil {
+				glog.V(0).Infof("loading index %s to btree error: %v", fileName+".idx", e)
+			}
 		}
 	}
 

weed/storage/volume_read_write.go

@@ -60,8 +60,6 @@ func (v *Volume) AppendBlob(b []byte) (offset int64, err error) {
 	if offset, err = v.dataFile.Seek(0, 2); err != nil {
 		glog.V(0).Infof("failed to seek the end of file: %v", err)
 		return
-	} else if offset != int64(v.dataFileSize) {
-		glog.V(0).Infof("dataFileSize %d != actual data file size: %d", v.dataFileSize, offset)
 	}
 	//ensure file writing starting from aligned positions
 	if offset%NeedlePaddingSize != 0 {
@@ -69,12 +67,9 @@ func (v *Volume) AppendBlob(b []byte) (offset int64, err error) {
 		if offset, err = v.dataFile.Seek(offset, 0); err != nil {
 			glog.V(0).Infof("failed to align in datafile %s: %v", v.dataFile.Name(), err)
 			return
-		} else if offset != int64(v.dataFileSize) {
-			glog.V(0).Infof("dataFileSize %d != actual data file size: %d", v.dataFileSize, offset)
 		}
 	}
 	_, err = v.dataFile.Write(b)
-	v.dataFileSize += int64(len(b))
 	return
 }
 
@@ -91,12 +86,10 @@ func (v *Volume) writeNeedle(n *Needle) (size uint32, err error) {
 		glog.V(4).Infof("needle is unchanged!")
 		return
 	}
-	var offset, actualSize int64
+	var offset int64
 	if offset, err = v.dataFile.Seek(0, 2); err != nil {
 		glog.V(0).Infof("failed to seek the end of file: %v", err)
 		return
-	} else if offset != int64(v.dataFileSize) {
-		glog.V(0).Infof("dataFileSize %d != actual data file size: %d", v.dataFileSize, offset)
 	}
 
 	//ensure file writing starting from aligned positions
@@ -108,13 +101,12 @@ func (v *Volume) writeNeedle(n *Needle) (size uint32, err error) {
 		}
 	}
 
-	if size, actualSize, err = n.Append(v.dataFile, v.Version()); err != nil {
+	if size, _, err = n.Append(v.dataFile, v.Version()); err != nil {
 		if e := v.dataFile.Truncate(offset); e != nil {
 			err = fmt.Errorf("%s\ncannot truncate %s: %v", err, v.dataFile.Name(), e)
 		}
 		return
 	}
-	v.dataFileSize += actualSize
 
 	nv, ok := v.nm.Get(n.Id)
 	if !ok || int64(nv.Offset)*NeedlePaddingSize < offset {
@@ -139,18 +131,15 @@ func (v *Volume) deleteNeedle(n *Needle) (uint32, error) {
 	//fmt.Println("key", n.Id, "volume offset", nv.Offset, "data_size", n.Size, "cached size", nv.Size)
 	if ok && nv.Size != TombstoneFileSize {
 		size := nv.Size
-		// println("adding tombstone", n.Id, "at offset", v.dataFileSize)
-		if err := v.nm.Delete(n.Id, uint32(v.dataFileSize/NeedlePaddingSize)); err != nil {
+		offset, err := v.dataFile.Seek(0, 2)
+		if err != nil {
 			return size, err
 		}
-		if offset, err := v.dataFile.Seek(0, 2); err != nil {
+		if err := v.nm.Delete(n.Id, uint32(offset/NeedlePaddingSize)); err != nil {
 			return size, err
-		} else if offset != int64(v.dataFileSize) {
-			glog.V(0).Infof("dataFileSize %d != actual data file size: %d, deleteMarker: %d", v.dataFileSize, offset, getActualSize(0))
 		}
 		n.Data = nil
-		_, actualSize, err := n.Append(v.dataFile, v.Version())
-		v.dataFileSize += actualSize
+		_, _, err = n.Append(v.dataFile, v.Version())
 		return size, err
 	}
 	return 0, nil

weed/storage/volume_super_block.go

@@ -56,7 +56,6 @@ func (v *Volume) maybeWriteSuperBlock() error {
 				}
 			}
 		}
-		v.dataFileSize = SuperBlockSize
 	}
 	return e
 }

weed/storage/volume_sync.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/operation"
+	"github.com/chrislusf/seaweedfs/weed/storage/needle"
 	"github.com/chrislusf/seaweedfs/weed/util"
 )
 
@@ -48,7 +49,7 @@ optimized more later).
 func (v *Volume) Synchronize(volumeServer string) (err error) {
 	var lastCompactRevision uint16 = 0
 	var compactRevision uint16 = 0
-	var masterMap CompactMap
+	var masterMap *needle.CompactMap
 	for i := 0; i < 3; i++ {
 		if masterMap, _, compactRevision, err = fetchVolumeFileEntries(volumeServer, v.Id); err != nil {
 			return fmt.Errorf("Failed to sync volume %d entries with %s: %v", v.Id, volumeServer, err)
@@ -69,7 +70,7 @@ func (v *Volume) Synchronize(volumeServer string) (err error) {
 	return
 }
 
-type ByOffset []NeedleValue
+type ByOffset []needle.NeedleValue
 
 func (a ByOffset) Len() int           { return len(a) }
 func (a ByOffset) Swap(i, j int)      { a[i], a[j] = a[j], a[i] }
@@ -77,18 +78,18 @@ func (a ByOffset) Less(i, j int) bool { return a[i].Offset < a[j].Offset }
 
 // trySynchronizing sync with remote volume server incrementally by
 // make up the local and remote delta.
-func (v *Volume) trySynchronizing(volumeServer string, masterMap CompactMap, compactRevision uint16) error {
+func (v *Volume) trySynchronizing(volumeServer string, masterMap *needle.CompactMap, compactRevision uint16) error {
 	slaveIdxFile, err := os.Open(v.nm.IndexFileName())
 	if err != nil {
 		return fmt.Errorf("Open volume %d index file: %v", v.Id, err)
 	}
 	defer slaveIdxFile.Close()
-	slaveMap, err := LoadNeedleMap(slaveIdxFile)
+	slaveMap, err := LoadBtreeNeedleMap(slaveIdxFile)
 	if err != nil {
 		return fmt.Errorf("Load volume %d index file: %v", v.Id, err)
 	}
-	var delta []NeedleValue
-	if err := masterMap.Visit(func(needleValue NeedleValue) error {
+	var delta []needle.NeedleValue
+	if err := masterMap.Visit(func(needleValue needle.NeedleValue) error {
 		if needleValue.Key == 0 {
 			return nil
 		}
@@ -100,7 +101,7 @@ func (v *Volume) trySynchronizing(volumeServer string, masterMap CompactMap, com
 	}); err != nil {
 		return fmt.Errorf("Add master entry: %v", err)
 	}
-	if err := slaveMap.m.Visit(func(needleValue NeedleValue) error {
+	if err := slaveMap.m.Visit(func(needleValue needle.NeedleValue) error {
 		if needleValue.Key == 0 {
 			return nil
 		}
@@ -137,8 +138,8 @@ func (v *Volume) trySynchronizing(volumeServer string, masterMap CompactMap, com
 	return nil
 }
 
-func fetchVolumeFileEntries(volumeServer string, vid VolumeId) (m CompactMap, lastOffset uint64, compactRevision uint16, err error) {
-	m = NewCompactMap()
+func fetchVolumeFileEntries(volumeServer string, vid VolumeId) (m *needle.CompactMap, lastOffset uint64, compactRevision uint16, err error) {
+	m = needle.NewCompactMap()
 
 	syncStatus, err := operation.GetVolumeSyncStatus(volumeServer, vid.String())
 	if err != nil {
@@ -149,9 +150,9 @@ func fetchVolumeFileEntries(volumeServer string, vid VolumeId) (m CompactMap, la
 	err = operation.GetVolumeIdxEntries(volumeServer, vid.String(), func(key uint64, offset, size uint32) {
 		// println("remote key", key, "offset", offset*NeedlePaddingSize, "size", size)
 		if offset > 0 && size != TombstoneFileSize {
-			m.Set(Key(key), offset, size)
+			m.Set(needle.Key(key), offset, size)
 		} else {
-			m.Delete(Key(key))
+			m.Delete(needle.Key(key))
 		}
 		total++
 	})
@@ -178,7 +179,7 @@ func (v *Volume) IndexFileContent() ([]byte, error) {
 }
 
 // removeNeedle removes one needle by needle key
-func (v *Volume) removeNeedle(key Key) {
+func (v *Volume) removeNeedle(key needle.Key) {
 	n := new(Needle)
 	n.Id = uint64(key)
 	v.deleteNeedle(n)
@@ -188,7 +189,7 @@ func (v *Volume) removeNeedle(key Key) {
 // The compact revision is checked first in case the remote volume
 // is compacted and the offset is invalid any more.
 func (v *Volume) fetchNeedle(volumeDataContentHandlerUrl string,
-	needleValue NeedleValue, compactRevision uint16) error {
+	needleValue needle.NeedleValue, compactRevision uint16) error {
 	// add master file entry to local data file
 	values := make(url.Values)
 	values.Add("revision", strconv.Itoa(int(compactRevision)))

weed/storage/volume_vacuum.go

@@ -24,7 +24,7 @@ func (v *Volume) Compact() error {
 	v.lastCompactIndexOffset = v.nm.IndexFileSize()
 	v.lastCompactRevision = v.SuperBlock.CompactRevision
 	glog.V(3).Infof("creating copies for volume %d ,last offset %d...", v.Id, v.lastCompactIndexOffset)
-	return v.copyDataAndGenerateIndexFile(filePath+".cpd", filePath+".cpx", v.dataFileSize)
+	return v.copyDataAndGenerateIndexFile(filePath+".cpd", filePath+".cpx", 0)
 }
 
 func (v *Volume) Compact2() error {
@@ -221,7 +221,7 @@ func (v *Volume) copyDataAndGenerateIndexFile(dstName, idxName string, prealloca
 	}
 	defer idx.Close()
 
-	nm := NewNeedleMap(idx)
+	nm := NewBtreeNeedleMap(idx)
 	new_offset := int64(SuperBlockSize)
 
 	now := uint64(time.Now().Unix())
@@ -272,7 +272,7 @@ func (v *Volume) copyDataBasedOnIndexFile(dstName, idxName string) (err error) {
 	}
 	defer oldIndexFile.Close()
 
-	nm := NewNeedleMap(idx)
+	nm := NewBtreeNeedleMap(idx)
 	now := uint64(time.Now().Unix())
 
 	v.SuperBlock.CompactRevision++

weed/topology/configuration.go

@@ -25,20 +25,6 @@ type Configuration struct {
 	ip2location map[string]loc
 }
 
-func NewConfiguration(b []byte) (*Configuration, error) {
-	c := &Configuration{}
-	err := xml.Unmarshal(b, c)
-	c.ip2location = make(map[string]loc)
-	for _, dc := range c.Topo.DataCenters {
-		for _, rack := range dc.Racks {
-			for _, ip := range rack.Ips {
-				c.ip2location[ip] = loc{dcName: dc.Name, rackName: rack.Name}
-			}
-		}
-	}
-	return c, err
-}
-
 func (c *Configuration) String() string {
 	if b, e := xml.MarshalIndent(c, "  ", "  "); e == nil {
 		return string(b)

weed/topology/configuration_test.go

@@ -1,42 +0,0 @@
-package topology
-
-import (
-	"fmt"
-	"testing"
-)
-
-func TestLoadConfiguration(t *testing.T) {
-
-	confContent := `
-
-<?xml version="1.0" encoding="UTF-8" ?>
-<Configuration>
-  <Topology>
-    <DataCenter name="dc1">
-      <Rack name="rack1">
-        <Ip>192.168.1.1</Ip>
-      </Rack>
-    </DataCenter>
-    <DataCenter name="dc2">
-      <Rack name="rack1">
-        <Ip>192.168.1.2</Ip>
-      </Rack>
-      <Rack name="rack2">
-        <Ip>192.168.1.3</Ip>
-        <Ip>192.168.1.4</Ip>
-      </Rack>
-    </DataCenter>
-  </Topology>
-</Configuration>
-`
-	c, err := NewConfiguration([]byte(confContent))
-
-	fmt.Printf("%s\n", c)
-	if err != nil {
-		t.Fatalf("unmarshal error:%v", err)
-	}
-
-	if len(c.Topo.DataCenters) <= 0 || c.Topo.DataCenters[0].Name != "dc1" {
-		t.Fatalf("unmarshal error:%s", c)
-	}
-}

weed/topology/store_replicate.go

@@ -58,10 +58,14 @@ func ReplicatedWrite(masterNode string, s *storage.Store,
 
 				pairMap := make(map[string]string)
 				if needle.HasPairs() {
-					err := json.Unmarshal(needle.Pairs, &pairMap)
+					tmpMap := make(map[string]string)
+					err := json.Unmarshal(needle.Pairs, &tmpMap)
 					if err != nil {
 						glog.V(0).Infoln("Unmarshal pairs error:", err)
 					}
+					for k, v := range tmpMap {
+						pairMap[storage.PairNamePrefix+k] = v
+					}
 				}
 
 				_, err := operation.Upload(u.String(),

weed/topology/topology.go

@@ -2,7 +2,6 @@ package topology
 
 import (
 	"errors"
-	"io/ioutil"
 	"math/rand"
 
 	"github.com/chrislusf/raft"
@@ -30,7 +29,7 @@ type Topology struct {
 	RaftServer raft.Server
 }
 
-func NewTopology(id string, confFile string, seq sequence.Sequencer, volumeSizeLimit uint64, pulse int) (*Topology, error) {
+func NewTopology(id string, seq sequence.Sequencer, volumeSizeLimit uint64, pulse int) *Topology {
 	t := &Topology{}
 	t.id = NodeId(id)
 	t.nodeType = "Topology"
@@ -44,9 +43,9 @@ func NewTopology(id string, confFile string, seq sequence.Sequencer, volumeSizeL
 
 	t.chanFullVolumes = make(chan storage.VolumeInfo)
 
-	err := t.loadConfiguration(confFile)
+	t.Configuration = &Configuration{}
 
-	return t, err
+	return t
 }
 
 func (t *Topology) IsLeader() bool {
@@ -72,16 +71,6 @@ func (t *Topology) Leader() (string, error) {
 	return l, nil
 }
 
-func (t *Topology) loadConfiguration(configurationFile string) error {
-	b, e := ioutil.ReadFile(configurationFile)
-	if e == nil {
-		t.Configuration, e = NewConfiguration(b)
-		return e
-	}
-	glog.V(0).Infoln("Using default configurations.")
-	return nil
-}
-
 func (t *Topology) Lookup(collection string, vid storage.VolumeId) []*DataNode {
 	//maybe an issue if lots of collections?
 	if collection == "" {

weed/topology/topology_vacuum.go

@@ -86,14 +86,16 @@ func (t *Topology) Vacuum(garbageThreshold string) int {
 		for _, vl := range c.storageType2VolumeLayout.Items() {
 			if vl != nil {
 				volumeLayout := vl.(*VolumeLayout)
-				writableSet := make(map[storage.VolumeId]bool)
-				for _, id := range volumeLayout.writables {
-					writableSet[id] = true
-				}
 				for vid, locationlist := range volumeLayout.vid2location {
-					if _, isWritable := writableSet[vid]; !isWritable {
+
+					volumeLayout.accessLock.RLock()
+					isReadOnly, hasValue := volumeLayout.readonlyVolumes[vid]
+					volumeLayout.accessLock.RUnlock()
+
+					if hasValue && isReadOnly {
 						continue
 					}
+
 					glog.V(0).Infof("check vacuum on collection:%s volume:%d", c.Name, vid)
 					if batchVacuumVolumeCheck(volumeLayout, vid, locationlist, garbageThreshold) {
 						if batchVacuumVolumeCompact(volumeLayout, vid, locationlist) {

weed/topology/volume_growth_test.go

@@ -79,11 +79,7 @@ func setup(topologyLayout string) *Topology {
 	fmt.Println("data:", data)
 
 	//need to connect all nodes first before server adding volumes
-	topo, err := NewTopology("weedfs", "/etc/weedfs/weedfs.conf",
-		sequence.NewMemorySequencer(), 32*1024, 5)
-	if err != nil {
-		panic("error: " + err.Error())
-	}
+	topo := NewTopology("weedfs", sequence.NewMemorySequencer(), 32*1024, 5)
 	mTopology := data.(map[string]interface{})
 	for dcKey, dcValue := range mTopology {
 		dc := NewDataCenter(dcKey)

weed/topology/volume_layout.go

@@ -16,6 +16,7 @@ type VolumeLayout struct {
 	ttl              *storage.TTL
 	vid2location     map[storage.VolumeId]*VolumeLocationList
 	writables        []storage.VolumeId        // transient array of writable volume id
+	readonlyVolumes  map[storage.VolumeId]bool // transient set of readonly volumes
 	oversizedVolumes map[storage.VolumeId]bool // set of oversized volumes
 	volumeSizeLimit  uint64
 	accessLock       sync.RWMutex
@@ -27,6 +28,7 @@ func NewVolumeLayout(rp *storage.ReplicaPlacement, ttl *storage.TTL, volumeSizeL
 		ttl:              ttl,
 		vid2location:     make(map[storage.VolumeId]*VolumeLocationList),
 		writables:        *new([]storage.VolumeId),
+		readonlyVolumes:  make(map[storage.VolumeId]bool),
 		oversizedVolumes: make(map[storage.VolumeId]bool),
 		volumeSizeLimit:  volumeSizeLimit,
 	}
@@ -50,11 +52,15 @@ func (vl *VolumeLayout) RegisterVolume(v *storage.VolumeInfo, dn *DataNode) {
 			if v_info.ReadOnly {
 				glog.V(3).Infof("vid %d removed from writable", v.Id)
 				vl.removeFromWritable(v.Id)
+				vl.readonlyVolumes[v.Id] = true
 				return
+			} else {
+				delete(vl.readonlyVolumes, v.Id)
 			}
 		} else {
 			glog.V(3).Infof("vid %d removed from writable", v.Id)
 			vl.removeFromWritable(v.Id)
+			delete(vl.readonlyVolumes, v.Id)
 			return
 		}
 	}

weed/util/constants.go

@@ -1,5 +1,5 @@
 package util
 
 const (
-	VERSION = "0.74"
+	VERSION = "0.76"
 )

weed/util/http_util.go

@@ -74,6 +74,17 @@ func Get(url string) ([]byte, error) {
 	return b, nil
 }
 
+func Head(url string) (http.Header, error) {
+	r, err := client.Head(url)
+	if err != nil {
+		return nil, err
+	}
+	if r.StatusCode >= 400 {
+		return nil, fmt.Errorf("%s: %s", url, r.Status)
+	}
+	return r.Header, nil
+}
+
 func Delete(url string, jwt security.EncodedJwt) error {
 	req, err := http.NewRequest("DELETE", url, nil)
 	if jwt != "" {