Merge pull request #1 from chrislusf/master

sync

1 files changed, 228 insertions, 170 deletions

diff --git a/weed/filesys/dir.go b/weed/filesys/dir.go
index 79cf45385..483229b3f 100644
--- a/weed/filesys/dir.go
+++ b/weed/filesys/dir.go
@@ -3,7 +3,7 @@ package filesys
 import (
 	"context"
 	"os"
-	"path"
+	"strings"
 	"time"
 
 	"github.com/chrislusf/seaweedfs/weed/filer2"
@@ -14,9 +14,9 @@ import (
 )
 
 type Dir struct {
-	Path       string
-	wfs        *WFS
-	attributes *filer_pb.FuseAttributes
+	Path  string
+	wfs   *WFS
+	entry *filer_pb.Entry
 }
 
 var _ = fs.Node(&Dir{})
@@ -27,6 +27,11 @@ var _ = fs.HandleReadDirAller(&Dir{})
 var _ = fs.NodeRemover(&Dir{})
 var _ = fs.NodeRenamer(&Dir{})
 var _ = fs.NodeSetattrer(&Dir{})
+var _ = fs.NodeGetxattrer(&Dir{})
+var _ = fs.NodeSetxattrer(&Dir{})
+var _ = fs.NodeRemovexattrer(&Dir{})
+var _ = fs.NodeListxattrer(&Dir{})
+var _ = fs.NodeForgetter(&Dir{})
 
 func (dir *Dir) Attr(ctx context.Context, attr *fuse.Attr) error {
 
@@ -35,42 +40,41 @@ func (dir *Dir) Attr(ctx context.Context, attr *fuse.Attr) error {
 
 	if dir.Path == dir.wfs.option.FilerMountRootPath {
 		dir.setRootDirAttributes(attr)
+		glog.V(3).Infof("root dir Attr %s, attr: %+v", dir.Path, attr)
 		return nil
 	}
 
-	item := dir.wfs.listDirectoryEntriesCache.Get(dir.Path)
-	if item != nil && !item.Expired() {
-		entry := item.Value().(*filer_pb.Entry)
+	if err := dir.maybeLoadEntry(); err != nil {
+		glog.V(3).Infof("dir Attr %s,err: %+v", dir.Path, err)
+		return err
+	}
 
-		attr.Mtime = time.Unix(entry.Attributes.Mtime, 0)
-		attr.Ctime = time.Unix(entry.Attributes.Crtime, 0)
-		attr.Mode = os.FileMode(entry.Attributes.FileMode)
-		attr.Gid = entry.Attributes.Gid
-		attr.Uid = entry.Attributes.Uid
+	attr.Inode = filer2.FullPath(dir.Path).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)
+	attr.Gid = dir.entry.Attributes.Gid
+	attr.Uid = dir.entry.Attributes.Uid
 
-		return nil
-	}
+	glog.V(3).Infof("dir Attr %s, attr: %+v", dir.Path, attr)
 
-	entry, err := filer2.GetEntry(ctx, dir.wfs, dir.Path)
-	if err != nil {
-		glog.V(2).Infof("read dir %s attr: %v, error: %v", dir.Path, dir.attributes, err)
-		return err
-	}
-	dir.attributes = entry.Attributes
+	return nil
+}
 
-	glog.V(2).Infof("dir %s: %v perm: %v", dir.Path, dir.attributes, os.FileMode(dir.attributes.FileMode))
+func (dir *Dir) Getxattr(ctx context.Context, req *fuse.GetxattrRequest, resp *fuse.GetxattrResponse) error {
 
-	attr.Mode = os.FileMode(dir.attributes.FileMode) | os.ModeDir
+	glog.V(4).Infof("dir Getxattr %s", dir.Path)
 
-	attr.Mtime = time.Unix(dir.attributes.Mtime, 0)
-	attr.Ctime = time.Unix(dir.attributes.Crtime, 0)
-	attr.Gid = dir.attributes.Gid
-	attr.Uid = dir.attributes.Uid
+	if err := dir.maybeLoadEntry(); err != nil {
+		return err
+	}
 
-	return nil
+	return getxattr(dir.entry, req, resp)
 }
 
 func (dir *Dir) setRootDirAttributes(attr *fuse.Attr) {
+	attr.Inode = 1 // filer2.FullPath(dir.Path).AsInode()
+	attr.Valid = time.Hour
 	attr.Uid = dir.wfs.option.MountUid
 	attr.Gid = dir.wfs.option.MountGid
 	attr.Mode = dir.wfs.option.MountMode
@@ -78,16 +82,25 @@ 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
 }
 
-func (dir *Dir) newFile(name string, entry *filer_pb.Entry) *File {
-	return &File{
-		Name:           name,
-		dir:            dir,
-		wfs:            dir.wfs,
-		entry:          entry,
-		entryViewCache: nil,
-	}
+func (dir *Dir) newFile(name string, entry *filer_pb.Entry) fs.Node {
+	return dir.wfs.getNode(filer2.NewFullPath(dir.Path, name), func() fs.Node {
+		return &File{
+			Name:           name,
+			dir:            dir,
+			wfs:            dir.wfs,
+			entry:          entry,
+			entryViewCache: nil,
+		}
+	})
+}
+
+func (dir *Dir) newDirectory(fullpath filer2.FullPath, entry *filer_pb.Entry) fs.Node {
+	return dir.wfs.getNode(fullpath, func() fs.Node {
+		return &Dir{Path: string(fullpath), wfs: dir.wfs, entry: entry}
+	})
 }
 
 func (dir *Dir) Create(ctx context.Context, req *fuse.CreateRequest,
@@ -109,94 +122,104 @@ func (dir *Dir) Create(ctx context.Context, req *fuse.CreateRequest,
 				TtlSec:      dir.wfs.option.TtlSec,
 			},
 		},
+		OExcl: req.Flags&fuse.OpenExclusive != 0,
 	}
-	glog.V(1).Infof("create: %v", request)
+	glog.V(1).Infof("create %s/%s: %v", dir.Path, req.Name, req.Flags)
 
-	if request.Entry.IsDirectory {
-		if err := dir.wfs.WithFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error {
-			if _, err := client.CreateEntry(ctx, request); err != nil {
-				glog.V(0).Infof("create %s/%s: %v", dir.Path, req.Name, err)
-				return fuse.EIO
+	if err := dir.wfs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
+		if err := filer_pb.CreateEntry(client, request); err != nil {
+			if strings.Contains(err.Error(), "EEXIST") {
+				return fuse.EEXIST
 			}
-			return nil
-		}); err != nil {
-			return nil, nil, err
+			return fuse.EIO
 		}
+		return nil
+	}); err != nil {
+		return nil, nil, err
 	}
-
-	file := dir.newFile(req.Name, request.Entry)
-	if !request.Entry.IsDirectory {
-		file.isOpen = true
+	var node fs.Node
+	if request.Entry.IsDirectory {
+		node = dir.newDirectory(filer2.NewFullPath(dir.Path, req.Name), request.Entry)
+		return node, nil, nil
 	}
+
+	node = dir.newFile(req.Name, request.Entry)
+	file := node.(*File)
+	file.isOpen++
 	fh := dir.wfs.AcquireHandle(file, req.Uid, req.Gid)
-	fh.dirtyMetadata = true
 	return file, fh, nil
 
 }
 
 func (dir *Dir) Mkdir(ctx context.Context, req *fuse.MkdirRequest) (fs.Node, error) {
 
-	err := dir.wfs.WithFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error {
+	newEntry := &filer_pb.Entry{
+		Name:        req.Name,
+		IsDirectory: true,
+		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,
+		},
+	}
+
+	err := dir.wfs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
 
 		request := &filer_pb.CreateEntryRequest{
 			Directory: dir.Path,
-			Entry: &filer_pb.Entry{
-				Name:        req.Name,
-				IsDirectory: true,
-				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,
-				},
-			},
+			Entry:     newEntry,
 		}
 
 		glog.V(1).Infof("mkdir: %v", request)
-		if _, err := client.CreateEntry(ctx, request); err != nil {
+		if err := filer_pb.CreateEntry(client, request); err != nil {
 			glog.V(0).Infof("mkdir %s/%s: %v", dir.Path, req.Name, err)
-			return fuse.EIO
+			return err
 		}
 
 		return nil
 	})
 
 	if err == nil {
-		node := &Dir{Path: path.Join(dir.Path, req.Name), wfs: dir.wfs}
+		node := dir.newDirectory(filer2.NewFullPath(dir.Path, req.Name), newEntry)
 		return node, nil
 	}
 
-	return nil, err
+	return nil, fuse.EIO
 }
 
 func (dir *Dir) Lookup(ctx context.Context, req *fuse.LookupRequest, resp *fuse.LookupResponse) (node fs.Node, err error) {
 
-	var entry *filer_pb.Entry
-	fullFilePath := path.Join(dir.Path, req.Name)
+	glog.V(4).Infof("dir Lookup %s: %s", dir.Path, req.Name)
 
-	item := dir.wfs.listDirectoryEntriesCache.Get(fullFilePath)
-	if item != nil && !item.Expired() {
-		entry = item.Value().(*filer_pb.Entry)
-	}
+	fullFilePath := filer2.NewFullPath(dir.Path, req.Name)
+	entry := dir.wfs.cacheGet(fullFilePath)
 
 	if entry == nil {
-		entry, err = filer2.GetEntry(ctx, dir.wfs, fullFilePath)
+		// glog.V(3).Infof("dir Lookup cache miss %s", fullFilePath)
+		entry, err = filer2.GetEntry(dir.wfs, fullFilePath)
 		if err != nil {
-			return nil, err
+			glog.V(1).Infof("dir GetEntry %s: %v", fullFilePath, err)
+			return nil, fuse.ENOENT
 		}
+		dir.wfs.cacheSet(fullFilePath, entry, 5*time.Minute)
+	} else {
+		glog.V(4).Infof("dir Lookup cache hit %s", fullFilePath)
 	}
 
 	if entry != nil {
 		if entry.IsDirectory {
-			node = &Dir{Path: path.Join(dir.Path, req.Name), wfs: dir.wfs, attributes: entry.Attributes}
+			node = dir.newDirectory(fullFilePath, entry)
 		} else {
 			node = dir.newFile(req.Name, entry)
 		}
 
-		resp.EntryValid = time.Duration(0)
+		// resp.EntryValid = time.Second
+		resp.Attr.Inode = fullFilePath.AsInode()
+		resp.Attr.Valid = time.Second
 		resp.Attr.Mtime = time.Unix(entry.Attributes.Mtime, 0)
-		resp.Attr.Ctime = time.Unix(entry.Attributes.Crtime, 0)
+		resp.Attr.Crtime = time.Unix(entry.Attributes.Crtime, 0)
 		resp.Attr.Mode = os.FileMode(entry.Attributes.FileMode)
 		resp.Attr.Gid = entry.Attributes.Gid
 		resp.Attr.Uid = entry.Attributes.Uid
@@ -204,57 +227,32 @@ func (dir *Dir) Lookup(ctx context.Context, req *fuse.LookupRequest, resp *fuse.
 		return node, nil
 	}
 
+	glog.V(4).Infof("not found dir GetEntry %s: %v", fullFilePath, err)
 	return nil, fuse.ENOENT
 }
 
 func (dir *Dir) ReadDirAll(ctx context.Context) (ret []fuse.Dirent, err error) {
 
-	err = dir.wfs.WithFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error {
-
-		paginationLimit := 1024
-		remaining := dir.wfs.option.DirListingLimit
-
-		lastEntryName := ""
-
-		for remaining >= 0 {
+	glog.V(3).Infof("dir ReadDirAll %s", dir.Path)
 
-			request := &filer_pb.ListEntriesRequest{
-				Directory:         dir.Path,
-				StartFromFileName: lastEntryName,
-				Limit:             uint32(paginationLimit),
-			}
-
-			glog.V(4).Infof("read directory: %v", request)
-			resp, err := client.ListEntries(ctx, request)
-			if err != nil {
-				glog.V(0).Infof("list %s: %v", dir.Path, err)
-				return fuse.EIO
-			}
-
-			cacheTtl := estimatedCacheTtl(len(resp.Entries))
-
-			for _, entry := range resp.Entries {
-				if entry.IsDirectory {
-					dirent := fuse.Dirent{Name: entry.Name, Type: fuse.DT_Dir}
-					ret = append(ret, dirent)
-				} else {
-					dirent := fuse.Dirent{Name: entry.Name, Type: fuse.DT_File}
-					ret = append(ret, dirent)
-				}
-				dir.wfs.listDirectoryEntriesCache.Set(path.Join(dir.Path, entry.Name), entry, cacheTtl)
-				lastEntryName = entry.Name
-			}
-
-			remaining -= len(resp.Entries)
-
-			if len(resp.Entries) < paginationLimit {
-				break
-			}
+	cacheTtl := 5 * time.Minute
 
+	readErr := filer2.ReadDirAllEntries(dir.wfs, filer2.FullPath(dir.Path), "", func(entry *filer_pb.Entry, isLast bool) {
+		fullpath := filer2.NewFullPath(dir.Path, entry.Name)
+		inode := fullpath.AsInode()
+		if entry.IsDirectory {
+			dirent := fuse.Dirent{Inode: inode, Name: entry.Name, Type: fuse.DT_Dir}
+			ret = append(ret, dirent)
+		} else {
+			dirent := fuse.Dirent{Inode: inode, Name: entry.Name, Type: fuse.DT_File}
+			ret = append(ret, dirent)
 		}
-
-		return nil
+		dir.wfs.cacheSet(fullpath, entry, cacheTtl)
 	})
+	if readErr != nil {
+		glog.V(0).Infof("list %s: %v", dir.Path, err)
+		return ret, fuse.EIO
+	}
 
 	return ret, err
 }
@@ -262,23 +260,29 @@ func (dir *Dir) ReadDirAll(ctx context.Context) (ret []fuse.Dirent, err error) {
 func (dir *Dir) Remove(ctx context.Context, req *fuse.RemoveRequest) error {
 
 	if !req.Dir {
-		return dir.removeOneFile(ctx, req)
+		return dir.removeOneFile(req)
 	}
 
-	return dir.removeFolder(ctx, req)
+	return dir.removeFolder(req)
 
 }
 
-func (dir *Dir) removeOneFile(ctx context.Context, req *fuse.RemoveRequest) error {
+func (dir *Dir) removeOneFile(req *fuse.RemoveRequest) error {
 
-	entry, err := filer2.GetEntry(ctx, dir.wfs, path.Join(dir.Path, req.Name))
+	filePath := filer2.NewFullPath(dir.Path, req.Name)
+	entry, err := filer2.GetEntry(dir.wfs, filePath)
 	if err != nil {
 		return err
 	}
+	if entry == nil {
+		return nil
+	}
+
+	dir.wfs.deleteFileChunks(entry.Chunks)
 
-	dir.wfs.deleteFileChunks(ctx, entry.Chunks)
+	dir.wfs.cacheDelete(filePath)
 
-	return dir.wfs.WithFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error {
+	return dir.wfs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
 
 		request := &filer_pb.DeleteEntryRequest{
 			Directory:    dir.Path,
@@ -287,22 +291,22 @@ func (dir *Dir) removeOneFile(ctx context.Context, req *fuse.RemoveRequest) erro
 		}
 
 		glog.V(3).Infof("remove file: %v", request)
-		_, err := client.DeleteEntry(ctx, request)
+		_, err := client.DeleteEntry(context.Background(), request)
 		if err != nil {
-			glog.V(3).Infof("remove file %s/%s: %v", dir.Path, req.Name, err)
+			glog.V(3).Infof("not found remove file %s/%s: %v", dir.Path, req.Name, err)
 			return fuse.ENOENT
 		}
 
-		dir.wfs.listDirectoryEntriesCache.Delete(path.Join(dir.Path, req.Name))
-
 		return nil
 	})
 
 }
 
-func (dir *Dir) removeFolder(ctx context.Context, req *fuse.RemoveRequest) error {
+func (dir *Dir) removeFolder(req *fuse.RemoveRequest) error {
 
-	return dir.wfs.WithFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error {
+	dir.wfs.cacheDelete(filer2.NewFullPath(dir.Path, req.Name))
+
+	return dir.wfs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
 
 		request := &filer_pb.DeleteEntryRequest{
 			Directory:    dir.Path,
@@ -311,14 +315,12 @@ func (dir *Dir) removeFolder(ctx context.Context, req *fuse.RemoveRequest) error
 		}
 
 		glog.V(3).Infof("remove directory entry: %v", request)
-		_, err := client.DeleteEntry(ctx, request)
+		_, err := client.DeleteEntry(context.Background(), request)
 		if err != nil {
-			glog.V(3).Infof("remove %s/%s: %v", dir.Path, req.Name, err)
+			glog.V(3).Infof("not found remove %s/%s: %v", dir.Path, req.Name, err)
 			return fuse.ENOENT
 		}
 
-		dir.wfs.listDirectoryEntriesCache.Delete(path.Join(dir.Path, req.Name))
-
 		return nil
 	})
 
@@ -326,66 +328,122 @@ func (dir *Dir) removeFolder(ctx context.Context, req *fuse.RemoveRequest) error
 
 func (dir *Dir) Setattr(ctx context.Context, req *fuse.SetattrRequest, resp *fuse.SetattrResponse) error {
 
-	if dir.attributes == nil {
-		return nil
+	glog.V(3).Infof("%v dir setattr %+v", dir.Path, req)
+
+	if err := dir.maybeLoadEntry(); err != nil {
+		return err
 	}
 
-	glog.V(3).Infof("%v dir setattr %+v, fh=%d", dir.Path, req, req.Handle)
 	if req.Valid.Mode() {
-		dir.attributes.FileMode = uint32(req.Mode)
+		dir.entry.Attributes.FileMode = uint32(req.Mode)
 	}
 
 	if req.Valid.Uid() {
-		dir.attributes.Uid = req.Uid
+		dir.entry.Attributes.Uid = req.Uid
 	}
 
 	if req.Valid.Gid() {
-		dir.attributes.Gid = req.Gid
+		dir.entry.Attributes.Gid = req.Gid
 	}
 
 	if req.Valid.Mtime() {
-		dir.attributes.Mtime = req.Mtime.Unix()
+		dir.entry.Attributes.Mtime = req.Mtime.Unix()
+	}
+
+	dir.wfs.cacheDelete(filer2.FullPath(dir.Path))
+
+	return dir.saveEntry()
+
+}
+
+func (dir *Dir) Setxattr(ctx context.Context, req *fuse.SetxattrRequest) error {
+
+	glog.V(4).Infof("dir Setxattr %s: %s", dir.Path, req.Name)
+
+	if err := dir.maybeLoadEntry(); err != nil {
+		return err
+	}
+
+	if err := setxattr(dir.entry, req); err != nil {
+		return err
+	}
+
+	dir.wfs.cacheDelete(filer2.FullPath(dir.Path))
+
+	return dir.saveEntry()
+
+}
+
+func (dir *Dir) Removexattr(ctx context.Context, req *fuse.RemovexattrRequest) error {
+
+	glog.V(4).Infof("dir Removexattr %s: %s", dir.Path, req.Name)
+
+	if err := dir.maybeLoadEntry(); err != nil {
+		return err
+	}
+
+	if err := removexattr(dir.entry, req); err != nil {
+		return err
+	}
+
+	dir.wfs.cacheDelete(filer2.FullPath(dir.Path))
+
+	return dir.saveEntry()
+
+}
+
+func (dir *Dir) Listxattr(ctx context.Context, req *fuse.ListxattrRequest, resp *fuse.ListxattrResponse) error {
+
+	glog.V(4).Infof("dir Listxattr %s", dir.Path)
+
+	if err := dir.maybeLoadEntry(); err != nil {
+		return err
+	}
+
+	if err := listxattr(dir.entry, req, resp); err != nil {
+		return err
 	}
 
+	return nil
+
+}
+
+func (dir *Dir) Forget() {
+	glog.V(3).Infof("Forget dir %s", dir.Path)
+
+	dir.wfs.forgetNode(filer2.FullPath(dir.Path))
+}
+
+func (dir *Dir) maybeLoadEntry() error {
+	if dir.entry == nil {
+		parentDirPath, name := filer2.FullPath(dir.Path).DirAndName()
+		entry, err := dir.wfs.maybeLoadEntry(parentDirPath, name)
+		if err != nil {
+			return err
+		}
+		dir.entry = entry
+	}
+	return nil
+}
+
+func (dir *Dir) saveEntry() error {
+
 	parentDir, name := filer2.FullPath(dir.Path).DirAndName()
-	return dir.wfs.WithFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error {
+
+	return dir.wfs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
 
 		request := &filer_pb.UpdateEntryRequest{
 			Directory: parentDir,
-			Entry: &filer_pb.Entry{
-				Name:       name,
-				Attributes: dir.attributes,
-			},
+			Entry:     dir.entry,
 		}
 
-		glog.V(1).Infof("set attr directory entry: %v", request)
-		_, err := client.UpdateEntry(ctx, request)
+		glog.V(1).Infof("save dir entry: %v", request)
+		_, err := client.UpdateEntry(context.Background(), request)
 		if err != nil {
-			glog.V(0).Infof("UpdateEntry %s: %v", dir.Path, err)
+			glog.V(0).Infof("UpdateEntry dir %s/%s: %v", parentDir, name, err)
 			return fuse.EIO
 		}
 
-		dir.wfs.listDirectoryEntriesCache.Delete(dir.Path)
-
 		return nil
 	})
-
-}
-
-func estimatedCacheTtl(numEntries int) time.Duration {
-	if numEntries < 100 {
-		// 30 ms per entry
-		return 3 * time.Second
-	}
-	if numEntries < 1000 {
-		// 10 ms per entry
-		return 10 * time.Second
-	}
-	if numEntries < 10000 {
-		// 10 ms per entry
-		return 100 * time.Second
-	}
-
-	// 2 ms per entry
-	return time.Duration(numEntries*2) * time.Millisecond
 }