Merge pull request #34 from chrislusf/master

sync

40 files changed, 716 insertions, 184 deletions

diff --git a/weed/command/filer_copy.go b/weed/command/filer_copy.go
index 2295faa8a..9afa65d23 100644
--- a/weed/command/filer_copy.go
+++ b/weed/command/filer_copy.go
@@ -219,7 +219,7 @@ func genFileCopyTask(fileOrDir string, destPath string, fileCopyTaskChan chan Fi
 
 	fileCopyTaskChan <- FileCopyTask{
 		sourceLocation:     fileOrDir,
-		destinationUrlPath: destPath+fi.Name(),
+		destinationUrlPath: destPath,
 		fileSize:           fi.Size(),
 		fileMode:           fi.Mode(),
 		uid:                uid,
@@ -405,7 +405,7 @@ func (worker *FileCopyWorker) uploadFileInChunks(task FileCopyTask, f *os.File,
 					Replication: *worker.options.replication,
 					Collection:  *worker.options.collection,
 					TtlSec:      worker.options.ttlSec,
-					Path:        task.destinationUrlPath+fileName,
+					Path:        task.destinationUrlPath + fileName,
 				}
 
 				assignResult, assignError = client.AssignVolume(context.Background(), request)
diff --git a/weed/command/mount.go b/weed/command/mount.go
index 7fdb21254..f325cb0a5 100644
--- a/weed/command/mount.go
+++ b/weed/command/mount.go
@@ -14,6 +14,7 @@ type MountOptions struct {
 	replication                 *string
 	ttlSec                      *int
 	chunkSizeLimitMB            *int
+	concurrentWriters           *int
 	cacheDir                    *string
 	cacheSizeMB                 *int64
 	dataCenter                  *string
@@ -42,6 +43,7 @@ func init() {
 	mountOptions.replication = cmdMount.Flag.String("replication", "", "replication(e.g. 000, 001) to create to files. If empty, let filer decide.")
 	mountOptions.ttlSec = cmdMount.Flag.Int("ttl", 0, "file ttl in seconds")
 	mountOptions.chunkSizeLimitMB = cmdMount.Flag.Int("chunkSizeLimitMB", 2, "local write buffer size, also chunk large files")
+	mountOptions.concurrentWriters = cmdMount.Flag.Int("concurrentWriters", 0, "limit concurrent goroutine writers if not 0")
 	mountOptions.cacheDir = cmdMount.Flag.String("cacheDir", os.TempDir(), "local cache directory for file chunks and meta data")
 	mountOptions.cacheSizeMB = cmdMount.Flag.Int64("cacheCapacityMB", 1000, "local file chunk cache capacity in MB (0 will disable cache)")
 	mountOptions.dataCenter = cmdMount.Flag.String("dataCenter", "", "prefer to write to the data center")
diff --git a/weed/command/mount_std.go b/weed/command/mount_std.go
index 20d08314c..83cb352ff 100644
--- a/weed/command/mount_std.go
+++ b/weed/command/mount_std.go
@@ -5,8 +5,6 @@ package command
 import (
 	"context"
 	"fmt"
-	"github.com/chrislusf/seaweedfs/weed/filer"
-	"github.com/chrislusf/seaweedfs/weed/filesys/meta_cache"
 	"os"
 	"os/user"
 	"path"
@@ -15,6 +13,8 @@ import (
 	"strings"
 	"time"
 
+	"github.com/chrislusf/seaweedfs/weed/filesys/meta_cache"
+
 	"github.com/seaweedfs/fuse"
 	"github.com/seaweedfs/fuse/fs"
 
@@ -33,7 +33,7 @@ func runMount(cmd *Command, args []string) bool {
 	if *mountReadRetryTime < time.Second {
 		*mountReadRetryTime = time.Second
 	}
-	filer.ReadWaitTime = *mountReadRetryTime
+	util.RetryWaitTime = *mountReadRetryTime
 
 	umask, umaskErr := strconv.ParseUint(*mountOptions.umaskString, 8, 64)
 	if umaskErr != nil {
@@ -175,6 +175,7 @@ func RunMount(option *MountOptions, umask os.FileMode) bool {
 		Replication:                 *option.replication,
 		TtlSec:                      int32(*option.ttlSec),
 		ChunkSizeLimit:              int64(chunkSizeLimitMB) * 1024 * 1024,
+		ConcurrentWriters:           *option.concurrentWriters,
 		CacheDir:                    *option.cacheDir,
 		CacheSizeMB:                 *option.cacheSizeMB,
 		DataCenter:                  *option.dataCenter,
diff --git a/weed/command/server.go b/weed/command/server.go
index 6a78fb3f4..4a9c2411a 100644
--- a/weed/command/server.go
+++ b/weed/command/server.go
@@ -2,13 +2,14 @@ package command
 
 import (
 	"fmt"
-	stats_collect "github.com/chrislusf/seaweedfs/weed/stats"
 	"os"
 	"runtime"
 	"runtime/pprof"
 	"strings"
 	"time"
 
+	stats_collect "github.com/chrislusf/seaweedfs/weed/stats"
+
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/util"
 )
@@ -60,9 +61,10 @@ var (
 	serverMetricsHttpPort     = cmdServer.Flag.Int("metricsPort", 0, "Prometheus metrics listen port")
 
 	// pulseSeconds              = cmdServer.Flag.Int("pulseSeconds", 5, "number of seconds between heartbeats")
-	isStartingFiler     = cmdServer.Flag.Bool("filer", false, "whether to start filer")
-	isStartingS3        = cmdServer.Flag.Bool("s3", false, "whether to start S3 gateway")
-	isStartingMsgBroker = cmdServer.Flag.Bool("msgBroker", false, "whether to start message broker")
+	isStartingVolumeServer = cmdServer.Flag.Bool("volume", true, "whether to start volume server")
+	isStartingFiler        = cmdServer.Flag.Bool("filer", false, "whether to start filer")
+	isStartingS3           = cmdServer.Flag.Bool("s3", false, "whether to start S3 gateway")
+	isStartingMsgBroker    = cmdServer.Flag.Bool("msgBroker", false, "whether to start message broker")
 
 	serverWhiteList []string
 
@@ -99,7 +101,7 @@ func init() {
 	serverOptions.v.fixJpgOrientation = cmdServer.Flag.Bool("volume.images.fix.orientation", false, "Adjust jpg orientation when uploading.")
 	serverOptions.v.readRedirect = cmdServer.Flag.Bool("volume.read.redirect", true, "Redirect moved or non-local volumes.")
 	serverOptions.v.compactionMBPerSecond = cmdServer.Flag.Int("volume.compactionMBps", 0, "limit compaction speed in mega bytes per second")
-	serverOptions.v.fileSizeLimitMB = cmdServer.Flag.Int("volume.fileSizeLimitMB", 1024, "limit file size to avoid out of memory")
+	serverOptions.v.fileSizeLimitMB = cmdServer.Flag.Int("volume.fileSizeLimitMB", 256, "limit file size to avoid out of memory")
 	serverOptions.v.publicUrl = cmdServer.Flag.String("volume.publicUrl", "", "publicly accessible address")
 	serverOptions.v.preStopSeconds = cmdServer.Flag.Int("volume.preStopSeconds", 10, "number of seconds between stop send heartbeats and stop volume server")
 	serverOptions.v.pprof = cmdServer.Flag.Bool("volume.pprof", false, "enable pprof http handlers. precludes --memprofile and --cpuprofile")
@@ -214,7 +216,7 @@ func runServer(cmd *Command, args []string) bool {
 	}
 
 	// start volume server
-	{
+	if *isStartingVolumeServer {
 		go serverOptions.v.startVolumeServer(*volumeDataFolders, *volumeMaxDataVolumeCounts, *serverWhiteListOption, *volumeMinFreeSpacePercent)
 
 	}
diff --git a/weed/command/volume.go b/weed/command/volume.go
index d73c24ed1..ce5992665 100644
--- a/weed/command/volume.go
+++ b/weed/command/volume.go
@@ -80,7 +80,7 @@ func init() {
 	v.cpuProfile = cmdVolume.Flag.String("cpuprofile", "", "cpu profile output file")
 	v.memProfile = cmdVolume.Flag.String("memprofile", "", "memory profile output file")
 	v.compactionMBPerSecond = cmdVolume.Flag.Int("compactionMBps", 0, "limit background compaction or copying speed in mega bytes per second")
-	v.fileSizeLimitMB = cmdVolume.Flag.Int("fileSizeLimitMB", 1024, "limit file size to avoid out of memory")
+	v.fileSizeLimitMB = cmdVolume.Flag.Int("fileSizeLimitMB", 256, "limit file size to avoid out of memory")
 	v.pprof = cmdVolume.Flag.Bool("pprof", false, "enable pprof http handlers. precludes --memprofile and --cpuprofile")
 	v.metricsHttpPort = cmdVolume.Flag.Int("metricsPort", 0, "Prometheus metrics listen port")
 }
diff --git a/weed/filer/filechunk_manifest.go b/weed/filer/filechunk_manifest.go
index 0d01a4a36..f5ab36d37 100644
--- a/weed/filer/filechunk_manifest.go
+++ b/weed/filer/filechunk_manifest.go
@@ -99,7 +99,7 @@ func retriedFetchChunkData(urlStrings []string, cipherKey []byte, isGzipped bool
 	var buffer bytes.Buffer
 	var shouldRetry bool
 
-	for waitTime := time.Second; waitTime < ReadWaitTime; waitTime += waitTime / 2 {
+	for waitTime := time.Second; waitTime < util.RetryWaitTime; waitTime += waitTime / 2 {
 		for _, urlString := range urlStrings {
 			shouldRetry, err = util.ReadUrlAsStream(urlString+"?readDeleted=true", cipherKey, isGzipped, isFullChunk, offset, size, func(data []byte) {
 				buffer.Write(data)
diff --git a/weed/filer/reader_at.go b/weed/filer/reader_at.go
index 04c64d449..ccc746b90 100644
--- a/weed/filer/reader_at.go
+++ b/weed/filer/reader_at.go
@@ -3,19 +3,16 @@ package filer
 import (
 	"context"
 	"fmt"
+	"io"
+	"math/rand"
+	"sync"
+
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
+	"github.com/chrislusf/seaweedfs/weed/util"
 	"github.com/chrislusf/seaweedfs/weed/util/chunk_cache"
 	"github.com/chrislusf/seaweedfs/weed/wdclient"
 	"github.com/golang/groupcache/singleflight"
-	"io"
-	"math/rand"
-	"sync"
-	"time"
-)
-
-var (
-	ReadWaitTime = 6 * time.Second
 )
 
 type ChunkReadAt struct {
@@ -45,34 +42,29 @@ func LookupFn(filerClient filer_pb.FilerClient) LookupFileIdFunctionType {
 		locations, found := vidCache[vid]
 		vicCacheLock.RUnlock()
 
-		waitTime := time.Second
-		for !found && waitTime < ReadWaitTime {
-			// println("looking up volume", vid)
-			err = filerClient.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
-				resp, err := client.LookupVolume(context.Background(), &filer_pb.LookupVolumeRequest{
-					VolumeIds: []string{vid},
+		if !found {
+			util.Retry("lookup volume "+vid, func() error {
+				err = filerClient.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
+					resp, err := client.LookupVolume(context.Background(), &filer_pb.LookupVolumeRequest{
+						VolumeIds: []string{vid},
+					})
+					if err != nil {
+						return err
+					}
+
+					locations = resp.LocationsMap[vid]
+					if locations == nil || len(locations.Locations) == 0 {
+						glog.V(0).Infof("failed to locate %s", fileId)
+						return fmt.Errorf("failed to locate %s", fileId)
+					}
+					vicCacheLock.Lock()
+					vidCache[vid] = locations
+					vicCacheLock.Unlock()
+
+					return nil
 				})
-				if err != nil {
-					return err
-				}
-
-				locations = resp.LocationsMap[vid]
-				if locations == nil || len(locations.Locations) == 0 {
-					glog.V(0).Infof("failed to locate %s", fileId)
-					return fmt.Errorf("failed to locate %s", fileId)
-				}
-				vicCacheLock.Lock()
-				vidCache[vid] = locations
-				vicCacheLock.Unlock()
-
-				return nil
+				return err
 			})
-			if err == nil {
-				break
-			}
-			glog.V(1).Infof("wait for volume %s", vid)
-			time.Sleep(waitTime)
-			waitTime += waitTime / 2
 		}
 
 		if err != nil {
diff --git a/weed/filesys/dir.go b/weed/filesys/dir.go
index ae2ae3418..a8481a435 100644
--- a/weed/filesys/dir.go
+++ b/weed/filesys/dir.go
@@ -27,6 +27,7 @@ type Dir struct {
 
 var _ = fs.Node(&Dir{})
 var _ = fs.NodeCreater(&Dir{})
+var _ = fs.NodeMknoder(&Dir{})
 var _ = fs.NodeMkdirer(&Dir{})
 var _ = fs.NodeFsyncer(&Dir{})
 var _ = fs.NodeRequestLookuper(&Dir{})
@@ -179,6 +180,20 @@ func (dir *Dir) Create(ctx context.Context, req *fuse.CreateRequest,
 
 }
 
+func (dir *Dir) Mknod(ctx context.Context, req *fuse.MknodRequest) (fs.Node, error) {
+	if req.Mode&os.ModeNamedPipe != 0 {
+		glog.V(1).Infof("mknod named pipe %s", req.String())
+		return nil, fuse.ENOSYS
+	}
+	if req.Mode&req.Mode&os.ModeSocket != 0 {
+		glog.V(1).Infof("mknod socket %s", req.String())
+		return nil, fuse.ENOSYS
+	}
+	// not going to support mknod for normal files either
+	glog.V(1).Infof("mknod %s", req.String())
+	return nil, fuse.ENOSYS
+}
+
 func (dir *Dir) Mkdir(ctx context.Context, req *fuse.MkdirRequest) (fs.Node, error) {
 
 	glog.V(4).Infof("mkdir %s: %s", dir.FullPath(), req.Name)
@@ -347,8 +362,22 @@ func (dir *Dir) removeOneFile(req *fuse.RemoveRequest) error {
 
 	// then, delete meta cache and fsNode cache
 	dir.wfs.metaCache.DeleteEntry(context.Background(), filePath)
+
+	// clear entry inside the file
+	fsNode := dir.wfs.fsNodeCache.GetFsNode(filePath)
+	if fsNode != nil {
+		if file, ok := fsNode.(*File); ok {
+			file.clearEntry()
+		}
+	}
 	dir.wfs.fsNodeCache.DeleteFsNode(filePath)
 
+	// remove current file handle if any
+	dir.wfs.handlesLock.Lock()
+	defer dir.wfs.handlesLock.Unlock()
+	inodeId := util.NewFullPath(dir.FullPath(), req.Name).AsInode()
+	delete(dir.wfs.handles, inodeId)
+
 	// delete the chunks last
 	if isDeleteData {
 		dir.wfs.deleteFileChunks(entry.Chunks)
diff --git a/weed/filesys/dir_link.go b/weed/filesys/dir_link.go
index f6bc41b56..ba3280f03 100644
--- a/weed/filesys/dir_link.go
+++ b/weed/filesys/dir_link.go
@@ -31,7 +31,7 @@ func (dir *Dir) Link(ctx context.Context, req *fuse.LinkRequest, old fs.Node) (f
 
 	glog.V(4).Infof("Link: %v/%v -> %v/%v", oldFile.dir.FullPath(), oldFile.Name, dir.FullPath(), req.NewName)
 
-	if err := oldFile.maybeLoadEntry(ctx); err != nil {
+	if _, err := oldFile.maybeLoadEntry(ctx); err != nil {
 		return nil, err
 	}
 
@@ -86,7 +86,7 @@ func (dir *Dir) Link(ctx context.Context, req *fuse.LinkRequest, old fs.Node) (f
 	// create new file node
 	newNode := dir.newFile(req.NewName, request.Entry)
 	newFile := newNode.(*File)
-	if err := newFile.maybeLoadEntry(ctx); err != nil {
+	if _, err := newFile.maybeLoadEntry(ctx); err != nil {
 		return nil, err
 	}
 
@@ -138,16 +138,17 @@ func (dir *Dir) Symlink(ctx context.Context, req *fuse.SymlinkRequest) (fs.Node,
 
 func (file *File) Readlink(ctx context.Context, req *fuse.ReadlinkRequest) (string, error) {
 
-	if err := file.maybeLoadEntry(ctx); err != nil {
+	entry, err := file.maybeLoadEntry(ctx)
+	if err != nil {
 		return "", err
 	}
 
-	if os.FileMode(file.entry.Attributes.FileMode)&os.ModeSymlink == 0 {
+	if os.FileMode(entry.Attributes.FileMode)&os.ModeSymlink == 0 {
 		return "", fuse.Errno(syscall.EINVAL)
 	}
 
-	glog.V(4).Infof("Readlink: %v/%v => %v", file.dir.FullPath(), file.Name, file.entry.Attributes.SymlinkTarget)
+	glog.V(4).Infof("Readlink: %v/%v => %v", file.dir.FullPath(), file.Name, entry.Attributes.SymlinkTarget)
 
-	return file.entry.Attributes.SymlinkTarget, nil
+	return entry.Attributes.SymlinkTarget, nil
 
 }
diff --git a/weed/filesys/dirty_page.go b/weed/filesys/dirty_page.go
index dd0c48796..11089186f 100644
--- a/weed/filesys/dirty_page.go
+++ b/weed/filesys/dirty_page.go
@@ -9,22 +9,16 @@ import (
 
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
-	"github.com/chrislusf/seaweedfs/weed/util"
-)
-
-var (
-	concurrentWriterLimit = runtime.NumCPU()
-	concurrentWriters     = util.NewLimitedConcurrentExecutor(4 * concurrentWriterLimit)
 )
 
 type ContinuousDirtyPages struct {
 	intervals              *ContinuousIntervals
 	f                      *File
 	writeWaitGroup         sync.WaitGroup
+	chunkAddLock           sync.Mutex
 	chunkSaveErrChan       chan error
 	chunkSaveErrChanClosed bool
 	lastErr                error
-	lock                   sync.Mutex
 	collection             string
 	replication            string
 }
@@ -33,7 +27,7 @@ func newDirtyPages(file *File) *ContinuousDirtyPages {
 	dirtyPages := &ContinuousDirtyPages{
 		intervals:        &ContinuousIntervals{},
 		f:                file,
-		chunkSaveErrChan: make(chan error, concurrentWriterLimit),
+		chunkSaveErrChan: make(chan error, runtime.NumCPU()),
 	}
 	go func() {
 		for t := range dirtyPages.chunkSaveErrChan {
@@ -100,14 +94,18 @@ func (pages *ContinuousDirtyPages) saveExistingLargestPageToStorage() (hasSavedD
 
 func (pages *ContinuousDirtyPages) saveToStorage(reader io.Reader, offset int64, size int64) {
 
+	errChanSize := pages.f.wfs.option.ConcurrentWriters
+	if errChanSize == 0 {
+		errChanSize = runtime.NumCPU()
+	}
 	if pages.chunkSaveErrChanClosed {
-		pages.chunkSaveErrChan = make(chan error, concurrentWriterLimit)
+		pages.chunkSaveErrChan = make(chan error, errChanSize)
 		pages.chunkSaveErrChanClosed = false
 	}
 
 	mtime := time.Now().UnixNano()
 	pages.writeWaitGroup.Add(1)
-	go func() {
+	writer := func() {
 		defer pages.writeWaitGroup.Done()
 
 		reader = io.LimitReader(reader, size)
@@ -119,9 +117,17 @@ func (pages *ContinuousDirtyPages) saveToStorage(reader io.Reader, offset int64,
 		}
 		chunk.Mtime = mtime
 		pages.collection, pages.replication = collection, replication
+		pages.chunkAddLock.Lock()
+		defer pages.chunkAddLock.Unlock()
 		pages.f.addChunks([]*filer_pb.FileChunk{chunk})
 		glog.V(3).Infof("%s saveToStorage [%d,%d)", pages.f.fullpath(), offset, offset+size)
-	}()
+	}
+
+	if pages.f.wfs.concurrentWriters != nil {
+		pages.f.wfs.concurrentWriters.Execute(writer)
+	} else {
+		go writer()
+	}
 }
 
 func max(x, y int64) int64 {
diff --git a/weed/filesys/file.go b/weed/filesys/file.go
index 7aa1016d7..9e1342370 100644
--- a/weed/filesys/file.go
+++ b/weed/filesys/file.go
@@ -43,32 +43,33 @@ func (file *File) fullpath() util.FullPath {
 	return util.NewFullPath(file.dir.FullPath(), file.Name)
 }
 
-func (file *File) Attr(ctx context.Context, attr *fuse.Attr) error {
+func (file *File) Attr(ctx context.Context, attr *fuse.Attr) (err error) {
 
 	glog.V(4).Infof("file Attr %s, open:%v, existing attr: %+v", file.fullpath(), file.isOpen, attr)
 
-	if file.isOpen <= 0 {
-		if err := file.maybeLoadEntry(ctx); err != nil {
+	entry := file.entry
+	if file.isOpen <= 0 || entry == nil {
+		if entry, err = file.maybeLoadEntry(ctx); err != nil {
 			return err
 		}
 	}
 
 	attr.Inode = file.fullpath().AsInode()
 	attr.Valid = time.Second
-	attr.Mode = os.FileMode(file.entry.Attributes.FileMode)
-	attr.Size = filer.FileSize(file.entry)
+	attr.Mode = os.FileMode(entry.Attributes.FileMode)
+	attr.Size = filer.FileSize(entry)
 	if file.isOpen > 0 {
-		attr.Size = file.entry.Attributes.FileSize
+		attr.Size = entry.Attributes.FileSize
 		glog.V(4).Infof("file Attr %s, open:%v, size: %d", file.fullpath(), file.isOpen, attr.Size)
 	}
-	attr.Crtime = time.Unix(file.entry.Attributes.Crtime, 0)
-	attr.Mtime = time.Unix(file.entry.Attributes.Mtime, 0)
-	attr.Gid = file.entry.Attributes.Gid
-	attr.Uid = file.entry.Attributes.Uid
+	attr.Crtime = time.Unix(entry.Attributes.Crtime, 0)
+	attr.Mtime = time.Unix(entry.Attributes.Mtime, 0)
+	attr.Gid = entry.Attributes.Gid
+	attr.Uid = entry.Attributes.Uid
 	attr.Blocks = attr.Size/blockSize + 1
 	attr.BlockSize = uint32(file.wfs.option.ChunkSizeLimit)
-	if file.entry.HardLinkCounter > 0 {
-		attr.Nlink = uint32(file.entry.HardLinkCounter)
+	if entry.HardLinkCounter > 0 {
+		attr.Nlink = uint32(entry.HardLinkCounter)
 	}
 
 	return nil
@@ -79,11 +80,12 @@ func (file *File) Getxattr(ctx context.Context, req *fuse.GetxattrRequest, resp
 
 	glog.V(4).Infof("file Getxattr %s", file.fullpath())
 
-	if err := file.maybeLoadEntry(ctx); err != nil {
+	entry, err := file.maybeLoadEntry(ctx)
+	if err != nil {
 		return err
 	}
 
-	return getxattr(file.entry, req, resp)
+	return getxattr(entry, req, resp)
 }
 
 func (file *File) Open(ctx context.Context, req *fuse.OpenRequest, resp *fuse.OpenResponse) (fs.Handle, error) {
@@ -104,7 +106,8 @@ func (file *File) Setattr(ctx context.Context, req *fuse.SetattrRequest, resp *f
 
 	glog.V(4).Infof("%v file setattr %+v", file.fullpath(), req)
 
-	if err := file.maybeLoadEntry(ctx); err != nil {
+	_, err := file.maybeLoadEntry(ctx)
+	if err != nil {
 		return err
 	}
 	if file.isOpen > 0 {
@@ -141,7 +144,7 @@ func (file *File) Setattr(ctx context.Context, req *fuse.SetattrRequest, resp *f
 				}
 			}
 			file.entry.Chunks = chunks
-			file.entryViewCache = nil
+			file.entryViewCache, _ = filer.NonOverlappingVisibleIntervals(filer.LookupFn(file.wfs), chunks)
 			file.reader = nil
 			file.wfs.deleteFileChunks(truncatedChunks)
 		}
@@ -186,7 +189,7 @@ func (file *File) Setattr(ctx context.Context, req *fuse.SetattrRequest, resp *f
 		return nil
 	}
 
-	return file.saveEntry()
+	return file.saveEntry(file.entry)
 
 }
 
@@ -194,15 +197,16 @@ func (file *File) Setxattr(ctx context.Context, req *fuse.SetxattrRequest) error
 
 	glog.V(4).Infof("file Setxattr %s: %s", file.fullpath(), req.Name)
 
-	if err := file.maybeLoadEntry(ctx); err != nil {
+	entry, err := file.maybeLoadEntry(ctx)
+	if err != nil {
 		return err
 	}
 
-	if err := setxattr(file.entry, req); err != nil {
+	if err := setxattr(entry, req); err != nil {
 		return err
 	}
 
-	return file.saveEntry()
+	return file.saveEntry(entry)
 
 }
 
@@ -210,15 +214,16 @@ func (file *File) Removexattr(ctx context.Context, req *fuse.RemovexattrRequest)
 
 	glog.V(4).Infof("file Removexattr %s: %s", file.fullpath(), req.Name)
 
-	if err := file.maybeLoadEntry(ctx); err != nil {
+	entry, err := file.maybeLoadEntry(ctx)
+	if err != nil {
 		return err
 	}
 
-	if err := removexattr(file.entry, req); err != nil {
+	if err := removexattr(entry, req); err != nil {
 		return err
 	}
 
-	return file.saveEntry()
+	return file.saveEntry(entry)
 
 }
 
@@ -226,11 +231,12 @@ func (file *File) Listxattr(ctx context.Context, req *fuse.ListxattrRequest, res
 
 	glog.V(4).Infof("file Listxattr %s", file.fullpath())
 
-	if err := file.maybeLoadEntry(ctx); err != nil {
+	entry, err := file.maybeLoadEntry(ctx)
+	if err != nil {
 		return err
 	}
 
-	if err := listxattr(file.entry, req, resp); err != nil {
+	if err := listxattr(entry, req, resp); err != nil {
 		return err
 	}
 
@@ -252,30 +258,61 @@ func (file *File) Forget() {
 	file.wfs.fsNodeCache.DeleteFsNode(t)
 }
 
-func (file *File) maybeLoadEntry(ctx context.Context) error {
-	if (file.entry != nil && len(file.entry.HardLinkId) != 0) || file.isOpen > 0 {
-		return nil
+func (file *File) maybeLoadEntry(ctx context.Context) (entry *filer_pb.Entry, err error) {
+	entry = file.entry
+	if file.isOpen > 0 {
+		return entry, nil
 	}
-	entry, err := file.wfs.maybeLoadEntry(file.dir.FullPath(), file.Name)
+	if entry != nil {
+		if len(entry.HardLinkId) == 0 {
+			// only always reload hard link
+			return entry, nil
+		}
+	}
+	entry, err = file.wfs.maybeLoadEntry(file.dir.FullPath(), file.Name)
 	if err != nil {
 		glog.V(3).Infof("maybeLoadEntry file %s/%s: %v", file.dir.FullPath(), file.Name, err)
-		return err
+		return entry, err
 	}
 	if entry != nil {
 		file.setEntry(entry)
+	} else {
+		glog.Warningf("maybeLoadEntry not found entry %s/%s: %v", file.dir.FullPath(), file.Name, err)
 	}
-	return nil
+	return entry, nil
+}
+
+func lessThan(a, b *filer_pb.FileChunk) bool {
+	if a.Mtime == b.Mtime {
+		return a.Fid.FileKey < b.Fid.FileKey
+	}
+	return a.Mtime < b.Mtime
 }
 
 func (file *File) addChunks(chunks []*filer_pb.FileChunk) {
 
-	sort.Slice(chunks, func(i, j int) bool {
-		if chunks[i].Mtime == chunks[j].Mtime {
-			return chunks[i].Fid.FileKey < chunks[j].Fid.FileKey
+	// find the earliest incoming chunk
+	newChunks := chunks
+	earliestChunk := newChunks[0]
+	for i:=1;i<len(newChunks);i++{
+		if lessThan(earliestChunk, newChunks[i]) {
+			earliestChunk = newChunks[i]
+		}
+	}
+
+	// pick out-of-order chunks from existing chunks
+	for _, chunk := range file.entry.Chunks {
+		if lessThan(earliestChunk, chunk) {
+			chunks = append(chunks, chunk)
 		}
-		return chunks[i].Mtime < chunks[j].Mtime
+	}
+
+	// sort incoming chunks
+	sort.Slice(chunks, func(i, j int) bool {
+		return lessThan(chunks[i], chunks[j])
 	})
 
+	// add to entry view cache
 	for _, chunk := range chunks {
 		file.entryViewCache = filer.MergeIntoVisibles(file.entryViewCache, chunk)
 	}
@@ -284,24 +321,30 @@ func (file *File) addChunks(chunks []*filer_pb.FileChunk) {
 
 	glog.V(4).Infof("%s existing %d chunks adds %d more", file.fullpath(), len(file.entry.Chunks), len(chunks))
 
-	file.entry.Chunks = append(file.entry.Chunks, chunks...)
+	file.entry.Chunks = append(file.entry.Chunks, newChunks...)
 }
 
 func (file *File) setEntry(entry *filer_pb.Entry) {
 	file.entry = entry
-	file.entryViewCache, _ = filer.NonOverlappingVisibleIntervals(filer.LookupFn(file.wfs), file.entry.Chunks)
+	file.entryViewCache, _ = filer.NonOverlappingVisibleIntervals(filer.LookupFn(file.wfs), entry.Chunks)
+	file.reader = nil
+}
+
+func (file *File) clearEntry() {
+	file.entry = nil
+	file.entryViewCache = nil
 	file.reader = nil
 }
 
-func (file *File) saveEntry() error {
+func (file *File) saveEntry(entry *filer_pb.Entry) error {
 	return file.wfs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
 
-		file.wfs.mapPbIdFromLocalToFiler(file.entry)
-		defer file.wfs.mapPbIdFromFilerToLocal(file.entry)
+		file.wfs.mapPbIdFromLocalToFiler(entry)
+		defer file.wfs.mapPbIdFromFilerToLocal(entry)
 
 		request := &filer_pb.UpdateEntryRequest{
 			Directory:  file.dir.FullPath(),
-			Entry:      file.entry,
+			Entry:      entry,
 			Signatures: []int32{file.wfs.signature},
 		}
 
diff --git a/weed/filesys/filehandle.go b/weed/filesys/filehandle.go
index 54bde3494..54410a0ba 100644
--- a/weed/filesys/filehandle.go
+++ b/weed/filesys/filehandle.go
@@ -183,16 +183,18 @@ func (fh *FileHandle) Release(ctx context.Context, req *fuse.ReleaseRequest) err
 	}
 
 	if fh.f.isOpen == 0 {
+
 		if err := fh.doFlush(ctx, req.Header); err != nil {
 			glog.Errorf("Release doFlush %s: %v", fh.f.Name, err)
 		}
-		fh.f.wfs.ReleaseHandle(fh.f.fullpath(), fuse.HandleID(fh.handle))
-	}
 
-	// stop the goroutine
-	if !fh.dirtyPages.chunkSaveErrChanClosed {
-		fh.dirtyPages.chunkSaveErrChanClosed = true
-		close(fh.dirtyPages.chunkSaveErrChan)
+		// stop the goroutine
+		if !fh.dirtyPages.chunkSaveErrChanClosed {
+			fh.dirtyPages.chunkSaveErrChanClosed = true
+			close(fh.dirtyPages.chunkSaveErrChan)
+		}
+
+		fh.f.wfs.ReleaseHandle(fh.f.fullpath(), fuse.HandleID(fh.handle))
 	}
 
 	return nil
@@ -262,7 +264,6 @@ func (fh *FileHandle) doFlush(ctx context.Context, header fuse.Header) error {
 			glog.V(0).Infof("MaybeManifestize: %v", manifestErr)
 		}
 		fh.f.entry.Chunks = append(chunks, manifestChunks...)
-		fh.f.entryViewCache = nil
 
 		fh.f.wfs.mapPbIdFromLocalToFiler(request.Entry)
 		defer fh.f.wfs.mapPbIdFromFilerToLocal(request.Entry)
diff --git a/weed/filesys/meta_cache/meta_cache_init.go b/weed/filesys/meta_cache/meta_cache_init.go
index f42d61230..4089cea28 100644
--- a/weed/filesys/meta_cache/meta_cache_init.go
+++ b/weed/filesys/meta_cache/meta_cache_init.go
@@ -3,8 +3,6 @@ package meta_cache
 import (
 	"context"
 	"fmt"
-	"strings"
-	"time"
 
 	"github.com/chrislusf/seaweedfs/weed/filer"
 	"github.com/chrislusf/seaweedfs/weed/glog"
@@ -18,7 +16,7 @@ func EnsureVisited(mc *MetaCache, client filer_pb.FilerClient, dirPath util.Full
 
 		glog.V(4).Infof("ReadDirAllEntries %s ...", path)
 
-		for waitTime := time.Second; waitTime < filer.ReadWaitTime; waitTime += waitTime / 2 {
+		util.Retry("ReadDirAllEntries", func() error {
 			err = filer_pb.ReadDirAllEntries(client, dirPath, "", func(pbEntry *filer_pb.Entry, isLast bool) error {
 				entry := filer.FromPbEntry(string(dirPath), pbEntry)
 				if err := mc.doInsertEntry(context.Background(), entry); err != nil {
@@ -30,17 +28,13 @@ func EnsureVisited(mc *MetaCache, client filer_pb.FilerClient, dirPath util.Full
 				}
 				return nil
 			})
-			if err == nil {
-				break
-			}
-			if strings.Contains(err.Error(), "transport: ") {
-				glog.V(0).Infof("ReadDirAllEntries %s: %v. Retry in %v", path, err, waitTime)
-				time.Sleep(waitTime)
-				continue
-			}
+			return err
+		})
+
+		if err != nil {
 			err = fmt.Errorf("list %s: %v", dirPath, err)
-			break
 		}
+
 		return
 	})
 }
diff --git a/weed/filesys/wfs.go b/weed/filesys/wfs.go
index 759e21b15..cd14e8032 100644
--- a/weed/filesys/wfs.go
+++ b/weed/filesys/wfs.go
@@ -31,6 +31,7 @@ type Option struct {
 	Replication        string
 	TtlSec             int32
 	ChunkSizeLimit     int64
+	ConcurrentWriters  int
 	CacheDir           string
 	CacheSizeMB        int64
 	DataCenter         string
@@ -68,6 +69,9 @@ type WFS struct {
 	chunkCache *chunk_cache.TieredChunkCache
 	metaCache  *meta_cache.MetaCache
 	signature  int32
+
+	// throttle writers
+	concurrentWriters *util.LimitedConcurrentExecutor
 }
 type statsCache struct {
 	filer_pb.StatisticsResponse
@@ -96,7 +100,7 @@ func NewSeaweedFileSystem(option *Option) *WFS {
 		fsNode := wfs.fsNodeCache.GetFsNode(filePath)
 		if fsNode != nil {
 			if file, ok := fsNode.(*File); ok {
-				file.entry = nil
+				file.clearEntry()
 			}
 		}
 	})
@@ -110,6 +114,10 @@ func NewSeaweedFileSystem(option *Option) *WFS {
 	wfs.root = &Dir{name: wfs.option.FilerMountRootPath, wfs: wfs, entry: entry}
 	wfs.fsNodeCache = newFsCache(wfs.root)
 
+	if wfs.option.ConcurrentWriters > 0 {
+		wfs.concurrentWriters = util.NewLimitedConcurrentExecutor(wfs.option.ConcurrentWriters)
+	}
+
 	return wfs
 }
 
diff --git a/weed/filesys/wfs_filer_client.go b/weed/filesys/wfs_filer_client.go
index 096ee555f..dd76f5669 100644
--- a/weed/filesys/wfs_filer_client.go
+++ b/weed/filesys/wfs_filer_client.go
@@ -1,6 +1,7 @@
 package filesys
 
 import (
+	"github.com/chrislusf/seaweedfs/weed/util"
 	"google.golang.org/grpc"
 
 	"github.com/chrislusf/seaweedfs/weed/pb"
@@ -11,10 +12,12 @@ var _ = filer_pb.FilerClient(&WFS{})
 
 func (wfs *WFS) WithFilerClient(fn func(filer_pb.SeaweedFilerClient) error) error {
 
-	err := pb.WithCachedGrpcClient(func(grpcConnection *grpc.ClientConn) error {
-		client := filer_pb.NewSeaweedFilerClient(grpcConnection)
-		return fn(client)
-	}, wfs.option.FilerGrpcAddress, wfs.option.GrpcDialOption)
+	err := util.Retry("filer grpc", func() error {
+		return pb.WithCachedGrpcClient(func(grpcConnection *grpc.ClientConn) error {
+			client := filer_pb.NewSeaweedFilerClient(grpcConnection)
+			return fn(client)
+		}, wfs.option.FilerGrpcAddress, wfs.option.GrpcDialOption)
+	})
 
 	if err == nil {
 		return nil
diff --git a/weed/ftpd/ftp_server.go b/weed/ftpd/ftp_server.go
new file mode 100644
index 000000000..4a0dca2c3
--- /dev/null
+++ b/weed/ftpd/ftp_server.go
@@ -0,0 +1,81 @@
+package ftpd
+
+import (
+	"crypto/tls"
+	"errors"
+	"fmt"
+	"net"
+
+	ftpserver "github.com/fclairamb/ftpserverlib"
+	"google.golang.org/grpc"
+)
+
+type FtpServerOption struct {
+	Filer            string
+	IP               string
+	IpBind           string
+	Port             int
+	FilerGrpcAddress string
+	FtpRoot          string
+	GrpcDialOption   grpc.DialOption
+	PassivePortStart int
+	PassivePortStop  int
+}
+
+type SftpServer struct {
+	option      *FtpServerOption
+	ftpListener net.Listener
+}
+
+var _ = ftpserver.MainDriver(&SftpServer{})
+
+// NewServer returns a new FTP server driver
+func NewFtpServer(ftpListener net.Listener, option *FtpServerOption) (*SftpServer, error) {
+	var err error
+	server := &SftpServer{
+		option:      option,
+		ftpListener: ftpListener,
+	}
+	return server, err
+}
+
+// GetSettings returns some general settings around the server setup
+func (s *SftpServer) GetSettings() (*ftpserver.Settings, error) {
+	var portRange *ftpserver.PortRange
+	if s.option.PassivePortStart > 0 && s.option.PassivePortStop > s.option.PassivePortStart {
+		portRange = &ftpserver.PortRange{
+			Start: s.option.PassivePortStart,
+			End:   s.option.PassivePortStop,
+		}
+	}
+
+	return &ftpserver.Settings{
+		Listener:                 s.ftpListener,
+		ListenAddr:               fmt.Sprintf("%s:%d", s.option.IpBind, s.option.Port),
+		PublicHost:               s.option.IP,
+		PassiveTransferPortRange: portRange,
+		ActiveTransferPortNon20:  true,
+		IdleTimeout:              -1,
+		ConnectionTimeout:        20,
+	}, nil
+}
+
+// ClientConnected is called to send the very first welcome message
+func (s *SftpServer) ClientConnected(cc ftpserver.ClientContext) (string, error) {
+	return "Welcome to SeaweedFS FTP Server", nil
+}
+
+// ClientDisconnected is called when the user disconnects, even if he never authenticated
+func (s *SftpServer) ClientDisconnected(cc ftpserver.ClientContext) {
+}
+
+// AuthUser authenticates the user and selects an handling driver
+func (s *SftpServer) AuthUser(cc ftpserver.ClientContext, username, password string) (ftpserver.ClientDriver, error) {
+	return nil, nil
+}
+
+// GetTLSConfig returns a TLS Certificate to use
+// The certificate could frequently change if we use something like "let's encrypt"
+func (s *SftpServer) GetTLSConfig() (*tls.Config, error) {
+	return nil, errors.New("no TLS certificate configured")
+}
diff --git a/weed/operation/upload_content.go b/weed/operation/upload_content.go
index a4148cb22..ac0b477cb 100644
--- a/weed/operation/upload_content.go
+++ b/weed/operation/upload_content.go
@@ -81,21 +81,18 @@ func doUpload(uploadUrl string, filename string, cipher bool, reader io.Reader,
 	if ok {
 		data = bytesReader.Bytes
 	} else {
-		buf := bytebufferpool.Get()
-		_, err = buf.ReadFrom(reader)
-		defer bytebufferpool.Put(buf)
+		data, err = ioutil.ReadAll(reader)
 		if err != nil {
 			err = fmt.Errorf("read input: %v", err)
 			return
 		}
-		data = buf.Bytes()
 	}
 	uploadResult, uploadErr := retriedUploadData(uploadUrl, filename, cipher, data, isInputCompressed, mtype, pairMap, jwt)
 	return uploadResult, uploadErr, data
 }
 
 func retriedUploadData(uploadUrl string, filename string, cipher bool, data []byte, isInputCompressed bool, mtype string, pairMap map[string]string, jwt security.EncodedJwt) (uploadResult *UploadResult, err error) {
-	for i := 0; i < 1; i++ {
+	for i := 0; i < 3; i++ {
 		uploadResult, err = doUploadData(uploadUrl, filename, cipher, data, isInputCompressed, mtype, pairMap, jwt)
 		if err == nil {
 			return
diff --git a/weed/pb/grpc_client_server.go b/weed/pb/grpc_client_server.go
index ce706e282..f19af43b2 100644
--- a/weed/pb/grpc_client_server.go
+++ b/weed/pb/grpc_client_server.go
@@ -62,6 +62,7 @@ func GrpcDial(ctx context.Context, address string, opts ...grpc.DialOption) (*gr
 		grpc.WithDefaultCallOptions(
 			grpc.MaxCallSendMsgSize(Max_Message_Size),
 			grpc.MaxCallRecvMsgSize(Max_Message_Size),
+			grpc.WaitForReady(true),
 		),
 		grpc.WithKeepaliveParams(keepalive.ClientParameters{
 			Time:                30 * time.Second, // client ping server if no activity for this long
diff --git a/weed/s3api/filer_util_tags.go b/weed/s3api/filer_util_tags.go
index 3d4da7825..75d3b37d0 100644
--- a/weed/s3api/filer_util_tags.go
+++ b/weed/s3api/filer_util_tags.go
@@ -4,10 +4,11 @@ import (
 	"strings"
 
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
+	xhttp "github.com/chrislusf/seaweedfs/weed/s3api/http"
 )
 
 const (
-	S3TAG_PREFIX = "s3-"
+	S3TAG_PREFIX = xhttp.AmzObjectTagging + "-"
 )
 
 func (s3a *S3ApiServer) getTags(parentDirectoryPath string, entryName string) (tags map[string]string, err error) {
diff --git a/weed/s3api/http/header.go b/weed/s3api/http/header.go
new file mode 100644
index 000000000..2802b560f
--- /dev/null
+++ b/weed/s3api/http/header.go
@@ -0,0 +1,30 @@
+/*
+ * MinIO Cloud Storage, (C) 2019 MinIO, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package http
+
+// Standard S3 HTTP request constants
+const (
+	// S3 storage class
+	AmzStorageClass = "x-amz-storage-class"
+
+	// S3 user-defined metadata
+	AmzUserMetaPrefix = "X-Amz-Meta-"
+
+	// S3 object tagging
+	AmzObjectTagging = "X-Amz-Tagging"
+	AmzTagCount      = "x-amz-tagging-count"
+)
diff --git a/weed/s3api/s3api_object_handlers.go b/weed/s3api/s3api_object_handlers.go
index fa628f44e..fe134c102 100644
--- a/weed/s3api/s3api_object_handlers.go
+++ b/weed/s3api/s3api_object_handlers.go
@@ -266,7 +266,7 @@ func (s3a *S3ApiServer) proxyToFiler(w http.ResponseWriter, r *http.Request, des
 
 	resp, postErr := client.Do(proxyReq)
 
-	if resp.ContentLength == -1 && !strings.HasSuffix(destUrl, "/") {
+	if (resp.ContentLength == -1 || resp.StatusCode == 404) && !strings.HasSuffix(destUrl, "/") {
 		writeErrorResponse(w, s3err.ErrNoSuchKey, r.URL)
 		return
 	}
diff --git a/weed/s3api/s3api_objects_list_handlers.go b/weed/s3api/s3api_objects_list_handlers.go
index 23406d6df..5d63f1039 100644
--- a/weed/s3api/s3api_objects_list_handlers.go
+++ b/weed/s3api/s3api_objects_list_handlers.go
@@ -4,7 +4,6 @@ import (
 	"context"
 	"encoding/xml"
 	"fmt"
-	"github.com/chrislusf/seaweedfs/weed/s3api/s3err"
 	"io"
 	"net/http"
 	"net/url"
@@ -15,6 +14,8 @@ import (
 
 	"github.com/chrislusf/seaweedfs/weed/filer"
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
+	xhttp "github.com/chrislusf/seaweedfs/weed/s3api/http"
+	"github.com/chrislusf/seaweedfs/weed/s3api/s3err"
 )
 
 type ListBucketResultV2 struct {
@@ -137,6 +138,10 @@ func (s3a *S3ApiServer) listFilerEntries(bucket string, originalPrefix string, m
 					})
 				}
 			} else {
+				storageClass := "STANDARD"
+				if v, ok := entry.Extended[xhttp.AmzStorageClass]; ok {
+					storageClass = string(v)
+				}
 				contents = append(contents, ListEntry{
 					Key:          fmt.Sprintf("%s/%s", dir, entry.Name)[len(bucketPrefix):],
 					LastModified: time.Unix(entry.Attributes.Mtime, 0).UTC(),
@@ -146,7 +151,7 @@ func (s3a *S3ApiServer) listFilerEntries(bucket string, originalPrefix string, m
 						ID:          fmt.Sprintf("%x", entry.Attributes.Uid),
 						DisplayName: entry.Attributes.UserName,
 					},
-					StorageClass: "STANDARD",
+					StorageClass: StorageClass(storageClass),
 				})
 			}
 		})
diff --git a/weed/server/filer_server_handlers.go b/weed/server/filer_server_handlers.go
index 18f78881c..555036feb 100644
--- a/weed/server/filer_server_handlers.go
+++ b/weed/server/filer_server_handlers.go
@@ -10,6 +10,10 @@ import (
 
 func (fs *FilerServer) filerHandler(w http.ResponseWriter, r *http.Request) {
 	w.Header().Set("Server", "SeaweedFS Filer "+util.VERSION)
+	if r.Header.Get("Origin") != "" {
+		w.Header().Set("Access-Control-Allow-Origin", "*")
+		w.Header().Set("Access-Control-Allow-Credentials", "true")
+	}
 	start := time.Now()
 	switch r.Method {
 	case "GET":
@@ -32,11 +36,19 @@ func (fs *FilerServer) filerHandler(w http.ResponseWriter, r *http.Request) {
 		stats.FilerRequestCounter.WithLabelValues("post").Inc()
 		fs.PostHandler(w, r)
 		stats.FilerRequestHistogram.WithLabelValues("post").Observe(time.Since(start).Seconds())
+	case "OPTIONS":
+		stats.FilerRequestCounter.WithLabelValues("options").Inc()
+		OptionsHandler(w, r, false)
+		stats.FilerRequestHistogram.WithLabelValues("head").Observe(time.Since(start).Seconds())
 	}
 }
 
 func (fs *FilerServer) readonlyFilerHandler(w http.ResponseWriter, r *http.Request) {
 	w.Header().Set("Server", "SeaweedFS Filer "+util.VERSION)
+	if r.Header.Get("Origin") != "" {
+		w.Header().Set("Access-Control-Allow-Origin", "*")
+		w.Header().Set("Access-Control-Allow-Credentials", "true")
+	}
 	start := time.Now()
 	switch r.Method {
 	case "GET":
@@ -47,5 +59,18 @@ func (fs *FilerServer) readonlyFilerHandler(w http.ResponseWriter, r *http.Reque
 		stats.FilerRequestCounter.WithLabelValues("head").Inc()
 		fs.GetOrHeadHandler(w, r, false)
 		stats.FilerRequestHistogram.WithLabelValues("head").Observe(time.Since(start).Seconds())
+	case "OPTIONS":
+		stats.FilerRequestCounter.WithLabelValues("options").Inc()
+		OptionsHandler(w, r, true)
+		stats.FilerRequestHistogram.WithLabelValues("head").Observe(time.Since(start).Seconds())
+	}
+}
+
+func OptionsHandler(w http.ResponseWriter, r *http.Request, isReadOnly bool) {
+	if isReadOnly {
+		w.Header().Add("Access-Control-Allow-Methods", "GET, OPTIONS")
+	} else {
+		w.Header().Add("Access-Control-Allow-Methods", "PUT, POST, GET, DELETE, OPTIONS")
 	}
+	w.Header().Add("Access-Control-Allow-Headers", "*")
 }
diff --git a/weed/server/filer_server_handlers_read.go b/weed/server/filer_server_handlers_read.go
index fbd45d6b9..69d485e90 100644
--- a/weed/server/filer_server_handlers_read.go
+++ b/weed/server/filer_server_handlers_read.go
@@ -15,6 +15,7 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/images"
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
+	xhttp "github.com/chrislusf/seaweedfs/weed/s3api/http"
 	"github.com/chrislusf/seaweedfs/weed/stats"
 	"github.com/chrislusf/seaweedfs/weed/util"
 )
@@ -93,6 +94,24 @@ func (fs *FilerServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request,
 		}
 	}
 
+	// print out the header from extended properties
+	for k, v := range entry.Extended {
+		w.Header().Set(k, string(v))
+	}
+
+	//set tag count
+	if r.Method == "GET" {
+		tagCount := 0
+		for k := range entry.Extended {
+			if strings.HasPrefix(k, xhttp.AmzObjectTagging+"-") {
+				tagCount++
+			}
+		}
+		if tagCount > 0 {
+			w.Header().Set(xhttp.AmzTagCount, strconv.Itoa(tagCount))
+		}
+	}
+
 	// set etag
 	etag := filer.ETagEntry(entry)
 	if inm := r.Header.Get("If-None-Match"); inm == "\""+etag+"\"" {
diff --git a/weed/server/filer_server_handlers_write_autochunk.go b/weed/server/filer_server_handlers_write_autochunk.go
index d86d49b2a..ee0af9aab 100644
--- a/weed/server/filer_server_handlers_write_autochunk.go
+++ b/weed/server/filer_server_handlers_write_autochunk.go
@@ -18,8 +18,10 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/operation"
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
+	xhttp "github.com/chrislusf/seaweedfs/weed/s3api/http"
 	"github.com/chrislusf/seaweedfs/weed/security"
 	"github.com/chrislusf/seaweedfs/weed/stats"
+	"github.com/chrislusf/seaweedfs/weed/storage/needle"
 	"github.com/chrislusf/seaweedfs/weed/util"
 )
 
@@ -176,6 +178,18 @@ func (fs *FilerServer) saveMetaData(ctx context.Context, r *http.Request, fileNa
 		Size: chunkOffset,
 	}
 
+	if entry.Extended == nil {
+		entry.Extended = make(map[string][]byte)
+	}
+
+	fs.saveAmzMetaData(r, entry)
+
+	for k, v := range r.Header {
+		if len(v) > 0 && strings.HasPrefix(k, needle.PairNamePrefix) {
+			entry.Extended[k] = []byte(v[0])
+		}
+	}
+
 	if dbErr := fs.filer.CreateEntry(ctx, entry, false, false, nil); dbErr != nil {
 		fs.filer.DeleteChunks(entry.Chunks)
 		replyerr = dbErr
@@ -308,3 +322,27 @@ func (fs *FilerServer) mkdir(ctx context.Context, w http.ResponseWriter, r *http
 	}
 	return filerResult, replyerr
 }
+
+func (fs *FilerServer) saveAmzMetaData(r *http.Request, entry *filer.Entry) {
+
+	if sc := r.Header.Get(xhttp.AmzStorageClass); sc != "" {
+		entry.Extended[xhttp.AmzStorageClass] = []byte(sc)
+	}
+
+	if tags := r.Header.Get(xhttp.AmzObjectTagging); tags != "" {
+		for _, v := range strings.Split(tags, "&") {
+			tag := strings.Split(v, "=")
+			if len(tag) == 2 {
+				entry.Extended[xhttp.AmzObjectTagging+"-"+tag[0]] = []byte(tag[1])
+			}
+		}
+	}
+
+	for header, values := range r.Header {
+		if strings.HasPrefix(header, xhttp.AmzUserMetaPrefix) {
+			for _, value := range values {
+				entry.Extended[header] = []byte(value)
+			}
+		}
+	}
+}
diff --git a/weed/server/filer_ui/templates.go b/weed/server/filer_ui/templates.go
index f86dde5b1..3f0647119 100644
--- a/weed/server/filer_ui/templates.go
+++ b/weed/server/filer_ui/templates.go
@@ -25,7 +25,7 @@ var StatusTpl = template.Must(template.New("status").Funcs(funcMap).Parse(`<!DOC
   <meta name="viewport" content="width=device-width, initial-scale=1">
   <link rel="stylesheet" href="/seaweedfsstatic/bootstrap/3.3.1/css/bootstrap.min.css">
 <style>
-body { padding-bottom: 70px; }
+body { padding-bottom: 128px; }
 #drop-area {
   border: 1px transparent;
 }
diff --git a/weed/server/volume_grpc_client_to_master.go b/weed/server/volume_grpc_client_to_master.go
index 199f8faba..2f594fa2b 100644
--- a/weed/server/volume_grpc_client_to_master.go
+++ b/weed/server/volume_grpc_client_to_master.go
@@ -203,6 +203,7 @@ func (vs *VolumeServer) doHeartbeat(masterNode, masterGrpcAddress string, grpcDi
 			}
 		case <-volumeTickChan:
 			glog.V(4).Infof("volume server %s:%d heartbeat", vs.store.Ip, vs.store.Port)
+			vs.store.MaybeAdjustVolumeMax()
 			if err = stream.Send(vs.store.CollectHeartbeat()); err != nil {
 				glog.V(0).Infof("Volume Server Failed to talk with master %s: %v", masterNode, err)
 				return "", err
@@ -216,6 +217,23 @@ func (vs *VolumeServer) doHeartbeat(masterNode, masterGrpcAddress string, grpcDi
 		case err = <-doneChan:
 			return
 		case <-vs.stopChan:
+			var volumeMessages []*master_pb.VolumeInformationMessage
+			emptyBeat := &master_pb.Heartbeat{
+				Ip:             vs.store.Ip,
+				Port:           uint32(vs.store.Port),
+				PublicUrl:      vs.store.PublicUrl,
+				MaxVolumeCount: uint32(0),
+				MaxFileKey:     uint64(0),
+				DataCenter:     vs.store.GetDataCenter(),
+				Rack:           vs.store.GetRack(),
+				Volumes:        volumeMessages,
+				HasNoVolumes:   len(volumeMessages) == 0,
+			}
+			glog.V(1).Infof("volume server %s:%d stops and deletes all volumes", vs.store.Ip, vs.store.Port)
+			if err = stream.Send(emptyBeat); err != nil {
+				glog.V(0).Infof("Volume Server Failed to update to master %s: %v", masterNode, err)
+				return "", err
+			}
 			return
 		}
 	}
diff --git a/weed/server/volume_grpc_copy.go b/weed/server/volume_grpc_copy.go
index cd2b53c8a..2aecb140f 100644
--- a/weed/server/volume_grpc_copy.go
+++ b/weed/server/volume_grpc_copy.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"io/ioutil"
 	"math"
 	"os"
 	"time"
@@ -60,13 +61,14 @@ func (vs *VolumeServer) VolumeCopy(ctx context.Context, req *volume_server_pb.Vo
 
 		volumeFileName = storage.VolumeFileName(location.Directory, volFileInfoResp.Collection, int(req.VolumeId))
 
+		ioutil.WriteFile(volumeFileName+".note", []byte(fmt.Sprintf("copying from %s", req.SourceDataNode)), 0755)
+
 		// println("source:", volFileInfoResp.String())
-		// copy ecx file
-		if err := vs.doCopyFile(client, false, req.Collection, req.VolumeId, volFileInfoResp.CompactionRevision, volFileInfoResp.IdxFileSize, volumeFileName, ".idx", false, false); err != nil {
+		if err := vs.doCopyFile(client, false, req.Collection, req.VolumeId, volFileInfoResp.CompactionRevision, volFileInfoResp.DatFileSize, volumeFileName, ".dat", false, true); err != nil {
 			return err
 		}
 
-		if err := vs.doCopyFile(client, false, req.Collection, req.VolumeId, volFileInfoResp.CompactionRevision, volFileInfoResp.DatFileSize, volumeFileName, ".dat", false, true); err != nil {
+		if err := vs.doCopyFile(client, false, req.Collection, req.VolumeId, volFileInfoResp.CompactionRevision, volFileInfoResp.IdxFileSize, volumeFileName, ".idx", false, false); err != nil {
 			return err
 		}
 
@@ -74,6 +76,8 @@ func (vs *VolumeServer) VolumeCopy(ctx context.Context, req *volume_server_pb.Vo
 			return err
 		}
 
+		os.Remove(volumeFileName + ".note")
+
 		return nil
 	})
 
diff --git a/weed/server/volume_server_handlers.go b/weed/server/volume_server_handlers.go
index ad13cdf3b..7852c950a 100644
--- a/weed/server/volume_server_handlers.go
+++ b/weed/server/volume_server_handlers.go
@@ -1,10 +1,11 @@
 package weed_server
 
 import (
-	"github.com/chrislusf/seaweedfs/weed/util"
 	"net/http"
 	"strings"
 
+	"github.com/chrislusf/seaweedfs/weed/util"
+
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/security"
 	"github.com/chrislusf/seaweedfs/weed/stats"
@@ -27,6 +28,10 @@ security settings:
 
 func (vs *VolumeServer) privateStoreHandler(w http.ResponseWriter, r *http.Request) {
 	w.Header().Set("Server", "SeaweedFS Volume "+util.VERSION)
+	if r.Header.Get("Origin") != "" {
+		w.Header().Set("Access-Control-Allow-Origin", "*")
+		w.Header().Set("Access-Control-Allow-Credentials", "true")
+	}
 	switch r.Method {
 	case "GET", "HEAD":
 		stats.ReadRequest()
@@ -37,11 +42,19 @@ func (vs *VolumeServer) privateStoreHandler(w http.ResponseWriter, r *http.Reque
 	case "PUT", "POST":
 		stats.WriteRequest()
 		vs.guard.WhiteList(vs.PostHandler)(w, r)
+	case "OPTIONS":
+		stats.ReadRequest()
+		w.Header().Add("Access-Control-Allow-Methods", "PUT, POST, GET, DELETE, OPTIONS")
+		w.Header().Add("Access-Control-Allow-Headers", "*")
 	}
 }
 
 func (vs *VolumeServer) publicReadOnlyHandler(w http.ResponseWriter, r *http.Request) {
 	w.Header().Set("Server", "SeaweedFS Volume "+util.VERSION)
+	if r.Header.Get("Origin") != "" {
+		w.Header().Set("Access-Control-Allow-Origin", "*")
+		w.Header().Set("Access-Control-Allow-Credentials", "true")
+	}
 	switch r.Method {
 	case "GET":
 		stats.ReadRequest()
@@ -49,6 +62,10 @@ func (vs *VolumeServer) publicReadOnlyHandler(w http.ResponseWriter, r *http.Req
 	case "HEAD":
 		stats.ReadRequest()
 		vs.GetOrHeadHandler(w, r)
+	case "OPTIONS":
+		stats.ReadRequest()
+		w.Header().Add("Access-Control-Allow-Methods", "GET, OPTIONS")
+		w.Header().Add("Access-Control-Allow-Headers", "*")
 	}
 }
 
diff --git a/weed/shell/command_volume_mark.go b/weed/shell/command_volume_mark.go
new file mode 100644
index 000000000..19b614310
--- /dev/null
+++ b/weed/shell/command_volume_mark.go
@@ -0,0 +1,55 @@
+package shell
+
+import (
+	"flag"
+	"fmt"
+	"io"
+
+	"github.com/chrislusf/seaweedfs/weed/storage/needle"
+)
+
+func init() {
+	Commands = append(Commands, &commandVolumeMark{})
+}
+
+type commandVolumeMark struct {
+}
+
+func (c *commandVolumeMark) Name() string {
+	return "volume.mark"
+}
+
+func (c *commandVolumeMark) Help() string {
+	return `Mark volume writable or readonly from one volume server
+
+	volume.mark -node <volume server host:port> -volumeId <volume id> -writable or -readonly
+`
+}
+
+func (c *commandVolumeMark) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
+
+	if err = commandEnv.confirmIsLocked(); err != nil {
+		return
+	}
+
+	volMarkCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
+	volumeIdInt := volMarkCommand.Int("volumeId", 0, "the volume id")
+	nodeStr := volMarkCommand.String("node", "", "the volume server <host>:<port>")
+	writable := volMarkCommand.Bool("writable", false, "volume mark writable")
+	readonly := volMarkCommand.Bool("readonly", false, "volume mark readonly")
+	if err = volMarkCommand.Parse(args); err != nil {
+		return nil
+	}
+	markWritable := false
+	if (*writable && *readonly) || (!*writable && !*readonly) {
+		return fmt.Errorf("use -readonly or -writable")
+	} else if *writable {
+		markWritable = true
+	}
+
+	sourceVolumeServer := *nodeStr
+
+	volumeId := needle.VolumeId(*volumeIdInt)
+
+	return markVolumeWritable(commandEnv.option.GrpcDialOption, volumeId, sourceVolumeServer, markWritable)
+}
diff --git a/weed/shell/command_volume_move.go b/weed/shell/command_volume_move.go
index b136604e5..2bc8dfad8 100644
--- a/weed/shell/command_volume_move.go
+++ b/weed/shell/command_volume_move.go
@@ -166,3 +166,18 @@ func deleteVolume(grpcDialOption grpc.DialOption, volumeId needle.VolumeId, sour
 		return deleteErr
 	})
 }
+
+func markVolumeWritable(grpcDialOption grpc.DialOption, volumeId needle.VolumeId, sourceVolumeServer string, writable bool) (err error) {
+	return operation.WithVolumeServerClient(sourceVolumeServer, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
+		if writable {
+			_, err = volumeServerClient.VolumeMarkWritable(context.Background(), &volume_server_pb.VolumeMarkWritableRequest{
+				VolumeId: uint32(volumeId),
+			})
+		} else {
+			_, err = volumeServerClient.VolumeMarkReadonly(context.Background(), &volume_server_pb.VolumeMarkReadonlyRequest{
+				VolumeId: uint32(volumeId),
+			})
+		}
+		return err
+	})
+}
diff --git a/weed/storage/disk_location.go b/weed/storage/disk_location.go
index ed57aa54b..775ebf092 100644
--- a/weed/storage/disk_location.go
+++ b/weed/storage/disk_location.go
@@ -13,14 +13,16 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/stats"
 	"github.com/chrislusf/seaweedfs/weed/storage/erasure_coding"
 	"github.com/chrislusf/seaweedfs/weed/storage/needle"
+	"github.com/chrislusf/seaweedfs/weed/util"
 )
 
 type DiskLocation struct {
-	Directory           string
-	MaxVolumeCount      int
-	MinFreeSpacePercent float32
-	volumes             map[needle.VolumeId]*Volume
-	volumesLock         sync.RWMutex
+	Directory              string
+	MaxVolumeCount         int
+	OriginalMaxVolumeCount int
+	MinFreeSpacePercent    float32
+	volumes                map[needle.VolumeId]*Volume
+	volumesLock            sync.RWMutex
 
 	// erasure coding
 	ecVolumes     map[needle.VolumeId]*erasure_coding.EcVolume
@@ -30,7 +32,7 @@ type DiskLocation struct {
 }
 
 func NewDiskLocation(dir string, maxVolumeCount int, minFreeSpacePercent float32) *DiskLocation {
-	location := &DiskLocation{Directory: dir, MaxVolumeCount: maxVolumeCount, MinFreeSpacePercent: minFreeSpacePercent}
+	location := &DiskLocation{Directory: dir, MaxVolumeCount: maxVolumeCount, OriginalMaxVolumeCount: maxVolumeCount, MinFreeSpacePercent: minFreeSpacePercent}
 	location.volumes = make(map[needle.VolumeId]*Volume)
 	location.ecVolumes = make(map[needle.VolumeId]*erasure_coding.EcVolume)
 	go location.CheckDiskSpace()
@@ -60,6 +62,14 @@ func parseCollectionVolumeId(base string) (collection string, vid needle.VolumeI
 func (l *DiskLocation) loadExistingVolume(fileInfo os.FileInfo, needleMapKind NeedleMapType) bool {
 	name := fileInfo.Name()
 	if !fileInfo.IsDir() && strings.HasSuffix(name, ".idx") {
+		name := name[:len(name)-len(".idx")]
+		noteFile := l.Directory + "/" + name + ".note"
+		if util.FileExists(noteFile) {
+			note, _ := ioutil.ReadFile(noteFile)
+			glog.Warningf("volume %s was not completed: %s", name, string(note))
+			removeVolumeFiles(l.Directory + "/" + name)
+			return false
+		}
 		vid, collection, err := l.volumeIdFromPath(fileInfo)
 		if err != nil {
 			glog.Warningf("get volume id failed, %s, err : %s", name, err)
@@ -85,7 +95,7 @@ func (l *DiskLocation) loadExistingVolume(fileInfo os.FileInfo, needleMapKind Ne
 
 		size, _, _ := v.FileStat()
 		glog.V(0).Infof("data file %s, replicaPlacement=%s v=%d size=%d ttl=%s",
-			l.Directory+"/"+name, v.ReplicaPlacement, v.Version(), size, v.Ttl.String())
+			l.Directory+"/"+name+".dat", v.ReplicaPlacement, v.Version(), size, v.Ttl.String())
 		return true
 	}
 	return false
diff --git a/weed/storage/needle/needle_read_write.go b/weed/storage/needle/needle_read_write.go
index 89fc85b0d..e758a6fee 100644
--- a/weed/storage/needle/needle_read_write.go
+++ b/weed/storage/needle/needle_read_write.go
@@ -24,6 +24,8 @@ const (
 	TtlBytesLength          = 2
 )
 
+var ErrorSizeMismatch = errors.New("size mismatch")
+
 func (n *Needle) DiskSize(version Version) int64 {
 	return GetActualSize(n.Size, version)
 }
@@ -168,6 +170,11 @@ func ReadNeedleBlob(r backend.BackendStorageFile, offset int64, size Size, versi
 func (n *Needle) ReadBytes(bytes []byte, offset int64, size Size, version Version) (err error) {
 	n.ParseNeedleHeader(bytes)
 	if n.Size != size {
+		// cookie is not always passed in for this API. Use size to do preliminary checking.
+		if OffsetSize == 4 && offset < int64(MaxPossibleVolumeSize) {
+			glog.Errorf("entry not found1: offset %d found id %x size %d, expected size %d", offset, n.Id, n.Size, size)
+			return ErrorSizeMismatch
+		}
 		return fmt.Errorf("entry not found: offset %d found id %x size %d, expected size %d", offset, n.Id, n.Size, size)
 	}
 	switch version {
diff --git a/weed/storage/store.go b/weed/storage/store.go
index b9fcfcba9..38f167cef 100644
--- a/weed/storage/store.go
+++ b/weed/storage/store.go
@@ -194,13 +194,19 @@ func (s *Store) SetDataCenter(dataCenter string) {
 func (s *Store) SetRack(rack string) {
 	s.rack = rack
 }
+func (s *Store) GetDataCenter() string {
+	return s.dataCenter
+}
+func (s *Store) GetRack() string {
+	return s.rack
+}
 
 func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
 	var volumeMessages []*master_pb.VolumeInformationMessage
 	maxVolumeCount := 0
 	var maxFileKey NeedleId
 	collectionVolumeSize := make(map[string]uint64)
-	collectionVolumeReadOnlyCount := make(map[string]uint8)
+	collectionVolumeReadOnlyCount := make(map[string]map[string]uint8)
 	for _, location := range s.Locations {
 		var deleteVids []needle.VolumeId
 		maxVolumeCount = maxVolumeCount + location.MaxVolumeCount
@@ -220,11 +226,24 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
 				}
 			}
 			collectionVolumeSize[v.Collection] += volumeMessage.Size
+			if _, exist := collectionVolumeReadOnlyCount[v.Collection]; !exist {
+				collectionVolumeReadOnlyCount[v.Collection] = map[string]uint8{
+					"IsReadOnly":       0,
+					"noWriteOrDelete":  0,
+					"noWriteCanDelete": 0,
+					"isDiskSpaceLow":   0,
+				}
+			}
 			if v.IsReadOnly() {
-				collectionVolumeReadOnlyCount[v.Collection] += 1
-			} else {
-				if _, exist := collectionVolumeReadOnlyCount[v.Collection]; !exist {
-					collectionVolumeReadOnlyCount[v.Collection] = 0
+				collectionVolumeReadOnlyCount[v.Collection]["IsReadOnly"] += 1
+				if v.noWriteOrDelete {
+					collectionVolumeReadOnlyCount[v.Collection]["noWriteOrDelete"] += 1
+				}
+				if v.noWriteCanDelete {
+					collectionVolumeReadOnlyCount[v.Collection]["noWriteCanDelete"] += 1
+				}
+				if v.location.isDiskSpaceLow {
+					collectionVolumeReadOnlyCount[v.Collection]["isDiskSpaceLow"] += 1
 				}
 			}
 		}
@@ -251,8 +270,10 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
 		stats.VolumeServerDiskSizeGauge.WithLabelValues(col, "normal").Set(float64(size))
 	}
 
-	for col, count := range collectionVolumeReadOnlyCount {
-		stats.VolumeServerReadOnlyVolumeGauge.WithLabelValues(col, "normal").Set(float64(count))
+	for col, types := range collectionVolumeReadOnlyCount {
+		for t, count := range types {
+			stats.VolumeServerReadOnlyVolumeGauge.WithLabelValues(col, t).Set(float64(count))
+		}
 	}
 
 	return &master_pb.Heartbeat{
@@ -440,7 +461,8 @@ func (s *Store) GetVolumeSizeLimit() uint64 {
 func (s *Store) MaybeAdjustVolumeMax() (hasChanges bool) {
 	volumeSizeLimit := s.GetVolumeSizeLimit()
 	for _, diskLocation := range s.Locations {
-		if diskLocation.MaxVolumeCount == 0 {
+		if diskLocation.OriginalMaxVolumeCount == 0 {
+			currentMaxVolumeCount := diskLocation.MaxVolumeCount
 			diskStatus := stats.NewDiskStatus(diskLocation.Directory)
 			unusedSpace := diskLocation.UnUsedSpace(volumeSizeLimit)
 			unclaimedSpaces := int64(diskStatus.Free) - int64(unusedSpace)
@@ -452,7 +474,7 @@ func (s *Store) MaybeAdjustVolumeMax() (hasChanges bool) {
 			diskLocation.MaxVolumeCount = maxVolumeCount
 			glog.V(0).Infof("disk %s max %d unclaimedSpace:%dMB, unused:%dMB volumeSizeLimit:%dMB",
 				diskLocation.Directory, maxVolumeCount, unclaimedSpaces/1024/1024, unusedSpace/1024/1024, volumeSizeLimit/1024/1024)
-			hasChanges = true
+			hasChanges = hasChanges || currentMaxVolumeCount != diskLocation.MaxVolumeCount
 		}
 	}
 	return
diff --git a/weed/storage/volume_checking.go b/weed/storage/volume_checking.go
index e42fb238b..00e04047f 100644
--- a/weed/storage/volume_checking.go
+++ b/weed/storage/volume_checking.go
@@ -2,8 +2,10 @@ package storage
 
 import (
 	"fmt"
+	"io"
 	"os"
 
+	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/storage/backend"
 	"github.com/chrislusf/seaweedfs/weed/storage/idx"
 	"github.com/chrislusf/seaweedfs/weed/storage/needle"
@@ -11,17 +13,40 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/util"
 )
 
-func CheckVolumeDataIntegrity(v *Volume, indexFile *os.File) (lastAppendAtNs uint64, e error) {
+func CheckAndFixVolumeDataIntegrity(v *Volume, indexFile *os.File) (lastAppendAtNs uint64, err error) {
 	var indexSize int64
-	if indexSize, e = verifyIndexFileIntegrity(indexFile); e != nil {
-		return 0, fmt.Errorf("verifyIndexFileIntegrity %s failed: %v", indexFile.Name(), e)
+	if indexSize, err = verifyIndexFileIntegrity(indexFile); err != nil {
+		return 0, fmt.Errorf("verifyIndexFileIntegrity %s failed: %v", indexFile.Name(), err)
 	}
 	if indexSize == 0 {
 		return 0, nil
 	}
+	healthyIndexSize := indexSize
+	for i := 1; i <= 10 && indexSize >= int64(i)*NeedleMapEntrySize; i++ {
+		// check and fix last 10 entries
+		lastAppendAtNs, err = doCheckAndFixVolumeData(v, indexFile, indexSize-int64(i)*NeedleMapEntrySize)
+		if err == io.EOF {
+			healthyIndexSize = indexSize - int64(i)*NeedleMapEntrySize
+			continue
+		}
+		if err != ErrorSizeMismatch {
+			break
+		}
+	}
+	if healthyIndexSize < indexSize {
+		glog.Warningf("CheckAndFixVolumeDataIntegrity truncate idx file %s from %d to %d", indexFile.Name(), indexSize, healthyIndexSize)
+		err = indexFile.Truncate(healthyIndexSize)
+		if err != nil {
+			glog.Warningf("CheckAndFixVolumeDataIntegrity truncate idx file %s from %d to %d: %v", indexFile.Name(), indexSize, healthyIndexSize, err)
+		}
+	}
+	return
+}
+
+func doCheckAndFixVolumeData(v *Volume, indexFile *os.File, indexOffset int64) (lastAppendAtNs uint64, err error) {
 	var lastIdxEntry []byte
-	if lastIdxEntry, e = readIndexEntryAtOffset(indexFile, indexSize-NeedleMapEntrySize); e != nil {
-		return 0, fmt.Errorf("readLastIndexEntry %s failed: %v", indexFile.Name(), e)
+	if lastIdxEntry, err = readIndexEntryAtOffset(indexFile, indexOffset); err != nil {
+		return 0, fmt.Errorf("readLastIndexEntry %s failed: %v", indexFile.Name(), err)
 	}
 	key, offset, size := idx.IdxFileEntry(lastIdxEntry)
 	if offset.IsZero() {
@@ -29,15 +54,15 @@ func CheckVolumeDataIntegrity(v *Volume, indexFile *os.File) (lastAppendAtNs uin
 	}
 	if size < 0 {
 		// read the deletion entry
-		if lastAppendAtNs, e = verifyDeletedNeedleIntegrity(v.DataBackend, v.Version(), key); e != nil {
-			return lastAppendAtNs, fmt.Errorf("verifyNeedleIntegrity %s failed: %v", indexFile.Name(), e)
+		if lastAppendAtNs, err = verifyDeletedNeedleIntegrity(v.DataBackend, v.Version(), key); err != nil {
+			return lastAppendAtNs, fmt.Errorf("verifyNeedleIntegrity %s failed: %v", indexFile.Name(), err)
 		}
 	} else {
-		if lastAppendAtNs, e = verifyNeedleIntegrity(v.DataBackend, v.Version(), offset.ToAcutalOffset(), key, size); e != nil {
-			return lastAppendAtNs, fmt.Errorf("verifyNeedleIntegrity %s failed: %v", indexFile.Name(), e)
+		if lastAppendAtNs, err = verifyNeedleIntegrity(v.DataBackend, v.Version(), offset.ToAcutalOffset(), key, size); err != nil {
+			return lastAppendAtNs, err
 		}
 	}
-	return
+	return lastAppendAtNs, nil
 }
 
 func verifyIndexFileIntegrity(indexFile *os.File) (indexSize int64, err error) {
@@ -60,7 +85,44 @@ func readIndexEntryAtOffset(indexFile *os.File, offset int64) (bytes []byte, err
 }
 
 func verifyNeedleIntegrity(datFile backend.BackendStorageFile, v needle.Version, offset int64, key NeedleId, size Size) (lastAppendAtNs uint64, err error) {
-	n := new(needle.Needle)
+	n, _, _, err := needle.ReadNeedleHeader(datFile, v, offset)
+	if err == io.EOF {
+		return 0, err
+	}
+	if err != nil {
+		return 0, fmt.Errorf("read %s at %d", datFile.Name(), offset)
+	}
+	if n.Size != size {
+		return 0, ErrorSizeMismatch
+	}
+	if v == needle.Version3 {
+		bytes := make([]byte, TimestampSize)
+		_, err = datFile.ReadAt(bytes, offset+NeedleHeaderSize+int64(size)+needle.NeedleChecksumSize)
+		if err == io.EOF {
+			return 0, err
+		}
+		if err != nil {
+			return 0, fmt.Errorf("verifyNeedleIntegrity check %s entry offset %d size %d: %v", datFile.Name(), offset, size, err)
+		}
+		n.AppendAtNs = util.BytesToUint64(bytes)
+		fileTailOffset := offset + needle.GetActualSize(size, v)
+		fileSize, _, err := datFile.GetStat()
+		if err != nil {
+			return 0, fmt.Errorf("stat file %s: %v", datFile.Name(), err)
+		}
+		if fileSize == fileTailOffset {
+			return n.AppendAtNs, nil
+		}
+		if fileSize > fileTailOffset {
+			glog.Warningf("Truncate %s from %d bytes to %d bytes!", datFile.Name(), fileSize, fileTailOffset)
+			err = datFile.Truncate(fileTailOffset)
+			if err == nil {
+				return n.AppendAtNs, nil
+			}
+			return n.AppendAtNs, fmt.Errorf("truncate file %s: %v", datFile.Name(), err)
+		}
+		glog.Warningf("data file %s has %d bytes, less than expected %d bytes!", datFile.Name(), fileSize, fileTailOffset)
+	}
 	if err = n.ReadData(datFile, offset, size, v); err != nil {
 		return n.AppendAtNs, fmt.Errorf("read data [%d,%d) : %v", offset, offset+int64(size), err)
 	}
diff --git a/weed/storage/volume_loading.go b/weed/storage/volume_loading.go
index 73e2de02b..05684cbdb 100644
--- a/weed/storage/volume_loading.go
+++ b/weed/storage/volume_loading.go
@@ -89,7 +89,7 @@ func (v *Volume) load(alsoLoadIndex bool, createDatIfMissing bool, needleMapKind
 				return fmt.Errorf("cannot write Volume Index %s.idx: %v", fileName, err)
 			}
 		}
-		if v.lastAppendAtNs, err = CheckVolumeDataIntegrity(v, indexFile); err != nil {
+		if v.lastAppendAtNs, err = CheckAndFixVolumeDataIntegrity(v, indexFile); err != nil {
 			v.noWriteOrDelete = true
 			glog.V(0).Infof("volumeDataIntegrityChecking failed %v", err)
 		}
diff --git a/weed/storage/volume_read_write.go b/weed/storage/volume_read_write.go
index 94c1d0ea1..869796a3f 100644
--- a/weed/storage/volume_read_write.go
+++ b/weed/storage/volume_read_write.go
@@ -17,6 +17,7 @@ import (
 
 var ErrorNotFound = errors.New("not found")
 var ErrorDeleted = errors.New("already deleted")
+var ErrorSizeMismatch = errors.New("size mismatch")
 
 // isFileUnchanged checks whether this needle to write is same as last one.
 // It requires serialized access in the same volume.
@@ -55,16 +56,21 @@ func (v *Volume) Destroy() (err error) {
 		}
 	}
 	v.Close()
-	os.Remove(v.FileName() + ".dat")
-	os.Remove(v.FileName() + ".idx")
-	os.Remove(v.FileName() + ".vif")
-	os.Remove(v.FileName() + ".sdx")
-	os.Remove(v.FileName() + ".cpd")
-	os.Remove(v.FileName() + ".cpx")
-	os.RemoveAll(v.FileName() + ".ldb")
+	removeVolumeFiles(v.FileName())
 	return
 }
 
+func removeVolumeFiles(filename string) {
+	os.Remove(filename + ".dat")
+	os.Remove(filename + ".idx")
+	os.Remove(filename + ".vif")
+	os.Remove(filename + ".sdx")
+	os.Remove(filename + ".cpd")
+	os.Remove(filename + ".cpx")
+	os.RemoveAll(filename + ".ldb")
+	os.Remove(filename + ".note")
+}
+
 func (v *Volume) asyncRequestAppend(request *needle.AsyncRequest) {
 	v.asyncRequestsChan <- request
 }
@@ -274,6 +280,9 @@ func (v *Volume) readNeedle(n *needle.Needle, readOption *ReadOption) (int, erro
 		return 0, nil
 	}
 	err := n.ReadData(v.DataBackend, nv.Offset.ToAcutalOffset(), readSize, v.Version())
+	if err == needle.ErrorSizeMismatch && OffsetSize == 4 {
+		err = n.ReadData(v.DataBackend, nv.Offset.ToAcutalOffset()+int64(MaxPossibleVolumeSize), readSize, v.Version())
+	}
 	if err != nil {
 		return 0, err
 	}
diff --git a/weed/util/constants.go b/weed/util/constants.go
index 177c20a60..498ef11a2 100644
--- a/weed/util/constants.go
+++ b/weed/util/constants.go
@@ -5,7 +5,7 @@ import (
 )
 
 var (
-	VERSION = fmt.Sprintf("%s %d.%02d", sizeLimit, 2, 07)
+	VERSION = fmt.Sprintf("%s %d.%02d", sizeLimit, 2, 8)
 	COMMIT  = ""
 )
 
diff --git a/weed/util/retry.go b/weed/util/retry.go
new file mode 100644
index 000000000..faaab0351
--- /dev/null
+++ b/weed/util/retry.go
@@ -0,0 +1,31 @@
+package util
+
+import (
+	"strings"
+	"time"
+
+	"github.com/chrislusf/seaweedfs/weed/glog"
+)
+
+var RetryWaitTime = 6 * time.Second
+
+func Retry(name string, job func() error) (err error) {
+	waitTime := time.Second
+	hasErr := false
+	for waitTime < RetryWaitTime {
+		err = job()
+		if err == nil {
+			if hasErr {
+				glog.V(0).Infof("retry %s successfully", name)
+			}
+			break
+		}
+		if strings.Contains(err.Error(), "transport") {
+			hasErr = true
+			glog.V(0).Infof("retry %s", name)
+			time.Sleep(waitTime)
+			waitTime += waitTime / 2
+		}
+	}
+	return err
+}
diff --git a/weed/wdclient/masterclient.go b/weed/wdclient/masterclient.go
index 60753e582..df8c186f2 100644
--- a/weed/wdclient/masterclient.go
+++ b/weed/wdclient/masterclient.go
@@ -5,6 +5,7 @@ import (
 	"math/rand"
 	"time"
 
+	"github.com/chrislusf/seaweedfs/weed/util"
 	"google.golang.org/grpc"
 
 	"github.com/chrislusf/seaweedfs/weed/glog"
@@ -150,10 +151,12 @@ func (mc *MasterClient) tryConnectToMaster(master string) (nextHintedLeader stri
 }
 
 func (mc *MasterClient) WithClient(fn func(client master_pb.SeaweedClient) error) error {
-	for mc.currentMaster == "" {
-		time.Sleep(3 * time.Second)
-	}
-	return pb.WithMasterClient(mc.currentMaster, mc.grpcDialOption, func(client master_pb.SeaweedClient) error {
-		return fn(client)
+	return util.Retry("master grpc", func() error {
+		for mc.currentMaster == "" {
+			time.Sleep(3 * time.Second)
+		}
+		return pb.WithMasterClient(mc.currentMaster, mc.grpcDialOption, func(client master_pb.SeaweedClient) error {
+			return fn(client)
+		})
 	})
 }