lsm/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2025-10-22 00:47:24 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

tree structured fs cache

Browse Source 
FsCache for FsNode, wrapping fs.Node
This commit is contained in:
Chris Lu
2020-03-26 00:08:14 -07:00
parent 2e4fadd10a
commit 7f0d87b206
10 changed files with 150 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files

106
weed/filesys/dir.go
View File

@@ -1,6 +1,7 @@
package filesys
import (
"bytes"
"context"
"os"
"strings"
@@ -14,9 +15,10 @@ import (
)
type Dir struct {
Path string
wfs *WFS
entry *filer_pb.Entry
name string
wfs *WFS
entry *filer_pb.Entry
parent *Dir
}
var _ = fs.Node(&Dir{})
@@ -38,32 +40,32 @@ func (dir *Dir) Attr(ctx context.Context, attr *fuse.Attr) error {
// https://github.com/bazil/fuse/issues/196
attr.Valid = time.Second
if dir.Path == dir.wfs.option.FilerMountRootPath {
if dir.FullPath() == dir.wfs.option.FilerMountRootPath {
dir.setRootDirAttributes(attr)
glog.V(3).Infof("root dir Attr %s, attr: %+v", dir.Path, attr)
glog.V(3).Infof("root dir Attr %s, attr: %+v", dir.FullPath(), attr)
return nil
}
if err := dir.maybeLoadEntry(); err != nil {
glog.V(3).Infof("dir Attr %s,err: %+v", dir.Path, err)
glog.V(3).Infof("dir Attr %s,err: %+v", dir.FullPath(), err)
return err
}
attr.Inode = util.FullPath(dir.Path).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)
attr.Gid = dir.entry.Attributes.Gid
attr.Uid = dir.entry.Attributes.Uid
glog.V(3).Infof("dir Attr %s, attr: %+v", dir.Path, attr)
glog.V(3).Infof("dir Attr %s, attr: %+v", dir.FullPath(), attr)
return nil
}
func (dir *Dir) Getxattr(ctx context.Context, req *fuse.GetxattrRequest, resp *fuse.GetxattrResponse) error {
glog.V(4).Infof("dir Getxattr %s", dir.Path)
glog.V(4).Infof("dir Getxattr %s", dir.FullPath())
if err := dir.maybeLoadEntry(); err != nil {
return err
@@ -86,7 +88,7 @@ func (dir *Dir) setRootDirAttributes(attr *fuse.Attr) {
}
func (dir *Dir) newFile(name string, entry *filer_pb.Entry) fs.Node {
return dir.wfs.fsNodeCache.EnsureFsNode(util.NewFullPath(dir.Path, name), func() fs.Node {
return dir.wfs.fsNodeCache.EnsureFsNode(util.NewFullPath(dir.FullPath(), name), func() fs.Node {
return &File{
Name: name,
dir: dir,
@@ -100,7 +102,7 @@ func (dir *Dir) newFile(name string, entry *filer_pb.Entry) fs.Node {
func (dir *Dir) newDirectory(fullpath util.FullPath, entry *filer_pb.Entry) fs.Node {
return dir.wfs.fsNodeCache.EnsureFsNode(fullpath, func() fs.Node {
return &Dir{Path: string(fullpath), wfs: dir.wfs, entry: entry}
return &Dir{name: entry.Name, wfs: dir.wfs, entry: entry, parent: dir,}
})
}
@@ -109,7 +111,7 @@ func (dir *Dir) Create(ctx context.Context, req *fuse.CreateRequest,
resp *fuse.CreateResponse) (fs.Node, fs.Handle, error) {
request := &filer_pb.CreateEntryRequest{
Directory: dir.Path,
Directory: dir.FullPath(),
Entry: &filer_pb.Entry{
Name: req.Name,
IsDirectory: req.Mode&os.ModeDir > 0,
@@ -126,7 +128,7 @@ func (dir *Dir) Create(ctx context.Context, req *fuse.CreateRequest,
},
OExcl: req.Flags&fuse.OpenExclusive != 0,
}
glog.V(1).Infof("create %s/%s: %v", dir.Path, req.Name, req.Flags)
glog.V(1).Infof("create %s/%s: %v", dir.FullPath(), req.Name, req.Flags)
if err := dir.wfs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
if err := filer_pb.CreateEntry(client, request); err != nil {
@@ -141,7 +143,7 @@ func (dir *Dir) Create(ctx context.Context, req *fuse.CreateRequest,
}
var node fs.Node
if request.Entry.IsDirectory {
node = dir.newDirectory(util.NewFullPath(dir.Path, req.Name), request.Entry)
node = dir.newDirectory(util.NewFullPath(dir.FullPath(), req.Name), request.Entry)
return node, nil, nil
}
@@ -170,13 +172,13 @@ func (dir *Dir) Mkdir(ctx context.Context, req *fuse.MkdirRequest) (fs.Node, err
err := dir.wfs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
request := &filer_pb.CreateEntryRequest{
Directory: dir.Path,
Directory: dir.FullPath(),
Entry: newEntry,
}
glog.V(1).Infof("mkdir: %v", request)
if err := filer_pb.CreateEntry(client, request); err != nil {
glog.V(0).Infof("mkdir %s/%s: %v", dir.Path, req.Name, err)
glog.V(0).Infof("mkdir %s/%s: %v", dir.FullPath(), req.Name, err)
return err
}
@@ -184,7 +186,7 @@ func (dir *Dir) Mkdir(ctx context.Context, req *fuse.MkdirRequest) (fs.Node, err
})
if err == nil {
node := dir.newDirectory(util.NewFullPath(dir.Path(), req.Name), newEntry)
node := dir.newDirectory(util.NewFullPath(dir.FullPath(), req.Name), newEntry)
return node, nil
}
@@ -193,9 +195,9 @@ func (dir *Dir) Mkdir(ctx context.Context, req *fuse.MkdirRequest) (fs.Node, err
func (dir *Dir) Lookup(ctx context.Context, req *fuse.LookupRequest, resp *fuse.LookupResponse) (node fs.Node, err error) {
glog.V(4).Infof("dir Lookup %s: %s", dir.Path, req.Name)
glog.V(4).Infof("dir Lookup %s: %s", dir.FullPath(), req.Name)
fullFilePath := util.NewFullPath(dir.Path, req.Name)
fullFilePath := util.NewFullPath(dir.FullPath(), req.Name)
entry := dir.wfs.cacheGet(fullFilePath)
if entry == nil {
@@ -235,12 +237,12 @@ func (dir *Dir) Lookup(ctx context.Context, req *fuse.LookupRequest, resp *fuse.
func (dir *Dir) ReadDirAll(ctx context.Context) (ret []fuse.Dirent, err error) {
glog.V(3).Infof("dir ReadDirAll %s", dir.Path)
glog.V(3).Infof("dir ReadDirAll %s", dir.FullPath())
cacheTtl := 5 * time.Minute
readErr := filer_pb.ReadDirAllEntries(dir.wfs, util.FullPath(dir.Path), "", func(entry *filer_pb.Entry, isLast bool) {
fullpath := util.NewFullPath(dir.Path, entry.Name)
readErr := filer_pb.ReadDirAllEntries(dir.wfs, util.FullPath(dir.FullPath()), "", func(entry *filer_pb.Entry, isLast bool) {
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}
@@ -252,7 +254,7 @@ func (dir *Dir) ReadDirAll(ctx context.Context) (ret []fuse.Dirent, err error) {
dir.wfs.cacheSet(fullpath, entry, cacheTtl)
})
if readErr != nil {
glog.V(0).Infof("list %s: %v", dir.Path, err)
glog.V(0).Infof("list %s: %v", dir.FullPath(), err)
return ret, fuse.EIO
}
@@ -271,7 +273,7 @@ func (dir *Dir) Remove(ctx context.Context, req *fuse.RemoveRequest) error {
func (dir *Dir) removeOneFile(req *fuse.RemoveRequest) error {
filePath := util.NewFullPath(dir.Path, req.Name)
filePath := util.NewFullPath(dir.FullPath(), req.Name)
entry, err := filer_pb.GetEntry(dir.wfs, filePath)
if err != nil {
return err
@@ -285,9 +287,9 @@ func (dir *Dir) removeOneFile(req *fuse.RemoveRequest) error {
dir.wfs.cacheDelete(filePath)
glog.V(3).Infof("remove file: %v", req)
err = filer_pb.Remove(dir.wfs, dir.Path, req.Name, false, false, false)
err = filer_pb.Remove(dir.wfs, dir.FullPath(), req.Name, false, false, false)
if err != nil {
glog.V(3).Infof("not found remove file %s/%s: %v", dir.Path, req.Name, err)
glog.V(3).Infof("not found remove file %s/%s: %v", dir.FullPath(), req.Name, err)
return fuse.ENOENT
}
@@ -297,12 +299,12 @@ func (dir *Dir) removeOneFile(req *fuse.RemoveRequest) error {
func (dir *Dir) removeFolder(req *fuse.RemoveRequest) error {
dir.wfs.cacheDelete(util.NewFullPath(dir.Path, req.Name))
dir.wfs.cacheDelete(util.NewFullPath(dir.FullPath(), req.Name))
glog.V(3).Infof("remove directory entry: %v", req)
err := filer_pb.Remove(dir.wfs, dir.Path, req.Name, true, false, false)
err := filer_pb.Remove(dir.wfs, dir.FullPath(), req.Name, true, false, false)
if err != nil {
glog.V(3).Infof("not found remove %s/%s: %v", dir.Path, req.Name, err)
glog.V(3).Infof("not found remove %s/%s: %v", dir.FullPath(), req.Name, err)
return fuse.ENOENT
}
return nil
@@ -311,7 +313,7 @@ func (dir *Dir) removeFolder(req *fuse.RemoveRequest) error {
func (dir *Dir) Setattr(ctx context.Context, req *fuse.SetattrRequest, resp *fuse.SetattrResponse) error {
glog.V(3).Infof("%v dir setattr %+v", dir.Path, req)
glog.V(3).Infof("%v dir setattr %+v", dir.FullPath(), req)
if err := dir.maybeLoadEntry(); err != nil {
return err
@@ -333,7 +335,7 @@ func (dir *Dir) Setattr(ctx context.Context, req *fuse.SetattrRequest, resp *fus
dir.entry.Attributes.Mtime = req.Mtime.Unix()
}
dir.wfs.cacheDelete(util.FullPath(dir.Path))
dir.wfs.cacheDelete(util.FullPath(dir.FullPath()))
return dir.saveEntry()
@@ -341,7 +343,7 @@ func (dir *Dir) Setattr(ctx context.Context, req *fuse.SetattrRequest, resp *fus
func (dir *Dir) Setxattr(ctx context.Context, req *fuse.SetxattrRequest) error {
glog.V(4).Infof("dir Setxattr %s: %s", dir.Path, req.Name)
glog.V(4).Infof("dir Setxattr %s: %s", dir.FullPath(), req.Name)
if err := dir.maybeLoadEntry(); err != nil {
return err
@@ -351,7 +353,7 @@ func (dir *Dir) Setxattr(ctx context.Context, req *fuse.SetxattrRequest) error {
return err
}
dir.wfs.cacheDelete(util.FullPath(dir.Path))
dir.wfs.cacheDelete(util.FullPath(dir.FullPath()))
return dir.saveEntry()
@@ -359,7 +361,7 @@ func (dir *Dir) Setxattr(ctx context.Context, req *fuse.SetxattrRequest) error {
func (dir *Dir) Removexattr(ctx context.Context, req *fuse.RemovexattrRequest) error {
glog.V(4).Infof("dir Removexattr %s: %s", dir.Path, req.Name)
glog.V(4).Infof("dir Removexattr %s: %s", dir.FullPath(), req.Name)
if err := dir.maybeLoadEntry(); err != nil {
return err
@@ -369,7 +371,7 @@ func (dir *Dir) Removexattr(ctx context.Context, req *fuse.RemovexattrRequest) e
return err
}
dir.wfs.cacheDelete(util.FullPath(dir.Path))
dir.wfs.cacheDelete(util.FullPath(dir.FullPath()))
return dir.saveEntry()
@@ -377,7 +379,7 @@ func (dir *Dir) Removexattr(ctx context.Context, req *fuse.RemovexattrRequest) e
func (dir *Dir) Listxattr(ctx context.Context, req *fuse.ListxattrRequest, resp *fuse.ListxattrResponse) error {
glog.V(4).Infof("dir Listxattr %s", dir.Path)
glog.V(4).Infof("dir Listxattr %s", dir.FullPath())
if err := dir.maybeLoadEntry(); err != nil {
return err
@@ -392,14 +394,14 @@ func (dir *Dir) Listxattr(ctx context.Context, req *fuse.ListxattrRequest, resp
}
func (dir *Dir) Forget() {
glog.V(3).Infof("Forget dir %s", dir.Path)
glog.V(3).Infof("Forget dir %s", dir.FullPath())
dir.wfs.fsNodeCache.DeleteFsNode(util.FullPath(dir.Path))
dir.wfs.fsNodeCache.DeleteFsNode(util.FullPath(dir.FullPath()))
}
func (dir *Dir) maybeLoadEntry() error {
if dir.entry == nil {
parentDirPath, name := util.FullPath(dir.Path).DirAndName()
parentDirPath, name := util.FullPath(dir.FullPath()).DirAndName()
entry, err := dir.wfs.maybeLoadEntry(parentDirPath, name)
if err != nil {
return err
@@ -411,7 +413,7 @@ func (dir *Dir) maybeLoadEntry() error {
func (dir *Dir) saveEntry() error {
parentDir, name := util.FullPath(dir.Path).DirAndName()
parentDir, name := util.FullPath(dir.FullPath()).DirAndName()
return dir.wfs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
@@ -430,3 +432,27 @@ func (dir *Dir) saveEntry() error {
return nil
})
}
func (dir *Dir) FullPath() string {
var parts []string
for p := dir; p != nil; p = p.parent {
if strings.HasPrefix(p.name, "/") {
if len(p.name) > 1 {
parts = append(parts, p.name[1:])
}
} else {
parts = append(parts, p.name)
}
}
if len(parts) == 0 {
return "/"
}
var buf bytes.Buffer
for i := len(parts) - 1; i >= 0; i-- {
buf.WriteString("/")
buf.WriteString(parts[i])
}
return buf.String()
}