Merge branch 'master' into support_ssd_volume

1 files changed, 74 insertions, 55 deletions

diff --git a/weed/filesys/dir.go b/weed/filesys/dir.go
index a8481a435..10a0a2b44 100644
--- a/weed/filesys/dir.go
+++ b/weed/filesys/dir.go
@@ -6,6 +6,7 @@ import (
 	"math"
 	"os"
 	"strings"
+	"syscall"
 	"time"
 
 	"github.com/seaweedfs/fuse"
@@ -57,7 +58,7 @@ func (dir *Dir) Attr(ctx context.Context, attr *fuse.Attr) error {
 		return err
 	}
 
-	attr.Inode = util.FullPath(dir.FullPath()).AsInode()
+	// attr.Inode = util.FullPath(dir.FullPath()).AsInode()
 	attr.Mode = os.FileMode(dir.entry.Attributes.FileMode) | os.ModeDir
 	attr.Mtime = time.Unix(dir.entry.Attributes.Mtime, 0)
 	attr.Crtime = time.Unix(dir.entry.Attributes.Crtime, 0)
@@ -81,8 +82,8 @@ func (dir *Dir) Getxattr(ctx context.Context, req *fuse.GetxattrRequest, resp *f
 }
 
 func (dir *Dir) setRootDirAttributes(attr *fuse.Attr) {
-	attr.Inode = 1 // filer2.FullPath(dir.Path).AsInode()
-	attr.Valid = time.Hour
+	// attr.Inode = 1 // filer2.FullPath(dir.Path).AsInode()
+	attr.Valid = time.Second
 	attr.Uid = dir.wfs.option.MountUid
 	attr.Gid = dir.wfs.option.MountGid
 	attr.Mode = dir.wfs.option.MountMode
@@ -90,7 +91,7 @@ func (dir *Dir) setRootDirAttributes(attr *fuse.Attr) {
 	attr.Ctime = dir.wfs.option.MountCtime
 	attr.Mtime = dir.wfs.option.MountMtime
 	attr.Atime = dir.wfs.option.MountMtime
-	attr.BlockSize = 1024 * 1024
+	attr.BlockSize = blockSize
 }
 
 func (dir *Dir) Fsync(ctx context.Context, req *fuse.FsyncRequest) error {
@@ -127,28 +128,59 @@ func (dir *Dir) newDirectory(fullpath util.FullPath, entry *filer_pb.Entry) fs.N
 func (dir *Dir) Create(ctx context.Context, req *fuse.CreateRequest,
 	resp *fuse.CreateResponse) (fs.Node, fs.Handle, error) {
 
+	request, err := dir.doCreateEntry(req.Name, req.Mode, req.Uid, req.Gid, req.Flags&fuse.OpenExclusive != 0)
+
+	if err != nil {
+		return nil, nil, err
+	}
+	var node fs.Node
+	if request.Entry.IsDirectory {
+		node = dir.newDirectory(util.NewFullPath(dir.FullPath(), req.Name), request.Entry)
+		return node, nil, nil
+	}
+
+	node = dir.newFile(req.Name, request.Entry)
+	file := node.(*File)
+	fh := dir.wfs.AcquireHandle(file, req.Uid, req.Gid)
+	return file, fh, nil
+
+}
+
+func (dir *Dir) Mknod(ctx context.Context, req *fuse.MknodRequest) (fs.Node, error) {
+
+	request, err := dir.doCreateEntry(req.Name, req.Mode, req.Uid, req.Gid, false)
+
+	if err != nil {
+		return nil, err
+	}
+	var node fs.Node
+	node = dir.newFile(req.Name, request.Entry)
+	return node, nil
+}
+
+func (dir *Dir) doCreateEntry(name string, mode os.FileMode, uid, gid uint32, exlusive bool) (*filer_pb.CreateEntryRequest, error) {
 	request := &filer_pb.CreateEntryRequest{
 		Directory: dir.FullPath(),
 		Entry: &filer_pb.Entry{
-			Name:        req.Name,
-			IsDirectory: req.Mode&os.ModeDir > 0,
+			Name:        name,
+			IsDirectory: mode&os.ModeDir > 0,
 			Attributes: &filer_pb.FuseAttributes{
 				Mtime:       time.Now().Unix(),
 				Crtime:      time.Now().Unix(),
-				FileMode:    uint32(req.Mode &^ dir.wfs.option.Umask),
-				Uid:         req.Uid,
-				Gid:         req.Gid,
+				FileMode:    uint32(mode &^ dir.wfs.option.Umask),
+				Uid:         uid,
+				Gid:         gid,
 				Collection:  dir.wfs.option.Collection,
 				Replication: dir.wfs.option.Replication,
 				TtlSec:      dir.wfs.option.TtlSec,
 			},
 		},
-		OExcl:      req.Flags&fuse.OpenExclusive != 0,
+		OExcl:      exlusive,
 		Signatures: []int32{dir.wfs.signature},
 	}
-	glog.V(1).Infof("create %s/%s: %v", dir.FullPath(), req.Name, req.Flags)
+	glog.V(1).Infof("create %s/%s", dir.FullPath(), name)
 
-	if err := dir.wfs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
+	err := dir.wfs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
 
 		dir.wfs.mapPbIdFromLocalToFiler(request.Entry)
 		defer dir.wfs.mapPbIdFromFilerToLocal(request.Entry)
@@ -157,41 +189,15 @@ func (dir *Dir) Create(ctx context.Context, req *fuse.CreateRequest,
 			if strings.Contains(err.Error(), "EEXIST") {
 				return fuse.EEXIST
 			}
-			glog.V(0).Infof("create %s/%s: %v", dir.FullPath(), req.Name, err)
+			glog.V(0).Infof("create %s/%s: %v", dir.FullPath(), name, err)
 			return fuse.EIO
 		}
 
 		dir.wfs.metaCache.InsertEntry(context.Background(), filer.FromPbEntry(request.Directory, request.Entry))
 
 		return nil
-	}); err != nil {
-		return nil, nil, err
-	}
-	var node fs.Node
-	if request.Entry.IsDirectory {
-		node = dir.newDirectory(util.NewFullPath(dir.FullPath(), req.Name), request.Entry)
-		return node, nil, nil
-	}
-
-	node = dir.newFile(req.Name, request.Entry)
-	file := node.(*File)
-	fh := dir.wfs.AcquireHandle(file, req.Uid, req.Gid)
-	return file, fh, nil
-
-}
-
-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
+	})
+	return request, err
 }
 
 func (dir *Dir) Mkdir(ctx context.Context, req *fuse.MkdirRequest) (fs.Node, error) {
@@ -279,7 +285,7 @@ func (dir *Dir) Lookup(ctx context.Context, req *fuse.LookupRequest, resp *fuse.
 		}
 
 		// resp.EntryValid = time.Second
-		resp.Attr.Inode = fullFilePath.AsInode()
+		// resp.Attr.Inode = fullFilePath.AsInode()
 		resp.Attr.Valid = time.Second
 		resp.Attr.Mtime = time.Unix(entry.Attributes.Mtime, 0)
 		resp.Attr.Crtime = time.Unix(entry.Attributes.Crtime, 0)
@@ -302,13 +308,11 @@ func (dir *Dir) ReadDirAll(ctx context.Context) (ret []fuse.Dirent, err error) {
 	glog.V(4).Infof("dir ReadDirAll %s", dir.FullPath())
 
 	processEachEntryFn := func(entry *filer_pb.Entry, isLast bool) error {
-		fullpath := util.NewFullPath(dir.FullPath(), entry.Name)
-		inode := fullpath.AsInode()
 		if entry.IsDirectory {
-			dirent := fuse.Dirent{Inode: inode, Name: entry.Name, Type: fuse.DT_Dir}
+			dirent := fuse.Dirent{Name: entry.Name, Type: fuse.DT_Dir}
 			ret = append(ret, dirent)
 		} else {
-			dirent := fuse.Dirent{Inode: inode, Name: entry.Name, Type: fuse.DT_File}
+			dirent := fuse.Dirent{Name: entry.Name, Type: findFileType(uint16(entry.Attributes.FileMode))}
 			ret = append(ret, dirent)
 		}
 		return nil
@@ -319,17 +323,37 @@ func (dir *Dir) ReadDirAll(ctx context.Context) (ret []fuse.Dirent, err error) {
 		glog.Errorf("dir ReadDirAll %s: %v", dirPath, err)
 		return nil, fuse.EIO
 	}
-	listedEntries, listErr := dir.wfs.metaCache.ListDirectoryEntries(context.Background(), util.FullPath(dir.FullPath()), "", false, int(math.MaxInt32))
+	listErr := dir.wfs.metaCache.ListDirectoryEntries(context.Background(), util.FullPath(dir.FullPath()), "", false, int64(math.MaxInt32), func(entry *filer.Entry) bool {
+		processEachEntryFn(entry.ToProtoEntry(), false)
+		return true
+	})
 	if listErr != nil {
 		glog.Errorf("list meta cache: %v", listErr)
 		return nil, fuse.EIO
 	}
-	for _, cachedEntry := range listedEntries {
-		processEachEntryFn(cachedEntry.ToProtoEntry(), false)
-	}
 	return
 }
 
+func findFileType(mode uint16) fuse.DirentType {
+	switch mode & (syscall.S_IFMT & 0xffff) {
+	case syscall.S_IFSOCK:
+		return fuse.DT_Socket
+	case syscall.S_IFLNK:
+		return fuse.DT_Link
+	case syscall.S_IFREG:
+		return fuse.DT_File
+	case syscall.S_IFBLK:
+		return fuse.DT_Block
+	case syscall.S_IFDIR:
+		return fuse.DT_Dir
+	case syscall.S_IFCHR:
+		return fuse.DT_Char
+	case syscall.S_IFIFO:
+		return fuse.DT_FIFO
+	}
+	return fuse.DT_File
+}
+
 func (dir *Dir) Remove(ctx context.Context, req *fuse.RemoveRequest) error {
 
 	if !req.Dir {
@@ -378,11 +402,6 @@ func (dir *Dir) removeOneFile(req *fuse.RemoveRequest) error {
 	inodeId := util.NewFullPath(dir.FullPath(), req.Name).AsInode()
 	delete(dir.wfs.handles, inodeId)
 
-	// delete the chunks last
-	if isDeleteData {
-		dir.wfs.deleteFileChunks(entry.Chunks)
-	}
-
 	return nil
 
 }