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

reverting and working

Browse Source
This commit is contained in:
Chris Lu
2020-06-28 10:14:17 -07:00
parent b813fac4a3
commit 1bb8cae65d
11 changed files with 478 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files

101
weed/filesys/dir.go
View File

@@ -7,13 +7,14 @@ import (
"strings"
"time"
"github.com/seaweedfs/fuse"
"github.com/seaweedfs/fuse/fs"
"github.com/chrislusf/seaweedfs/weed/filer2"
"github.com/chrislusf/seaweedfs/weed/filesys/meta_cache"
"github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
"github.com/chrislusf/seaweedfs/weed/util"
"github.com/seaweedfs/fuse"
"github.com/seaweedfs/fuse/fs"
)
type Dir struct {
@@ -90,18 +91,22 @@ func (dir *Dir) setRootDirAttributes(attr *fuse.Attr) {
}
func (dir *Dir) newFile(name string, entry *filer_pb.Entry) fs.Node {
return &File{
Name: name,
dir: dir,
wfs: dir.wfs,
entry: entry,
entryViewCache: nil,
}
return dir.wfs.fsNodeCache.EnsureFsNode(util.NewFullPath(dir.FullPath(), name), func() fs.Node {
return &File{
Name: name,
dir: dir,
wfs: dir.wfs,
entry: entry,
entryViewCache: nil,
}
})
}
func (dir *Dir) newDirectory(fullpath util.FullPath, entry *filer_pb.Entry) fs.Node {
return &Dir{name: entry.Name, wfs: dir.wfs, entry: entry, parent: dir}
return dir.wfs.fsNodeCache.EnsureFsNode(fullpath, func() fs.Node {
return &Dir{name: entry.Name, wfs: dir.wfs, entry: entry, parent: dir}
})
}
@@ -136,7 +141,9 @@ func (dir *Dir) Create(ctx context.Context, req *fuse.CreateRequest,
return fuse.EIO
}
dir.wfs.metaCache.InsertEntry(context.Background(), filer2.FromPbEntry(request.Directory, request.Entry))
if dir.wfs.option.AsyncMetaDataCaching {
dir.wfs.metaCache.InsertEntry(context.Background(), filer2.FromPbEntry(request.Directory, request.Entry))
}
return nil
}); err != nil {
@@ -185,7 +192,9 @@ func (dir *Dir) Mkdir(ctx context.Context, req *fuse.MkdirRequest) (fs.Node, err
return err
}
dir.wfs.metaCache.InsertEntry(context.Background(), filer2.FromPbEntry(request.Directory, request.Entry))
if dir.wfs.option.AsyncMetaDataCaching {
dir.wfs.metaCache.InsertEntry(context.Background(), filer2.FromPbEntry(request.Directory, request.Entry))
}
return nil
})
@@ -205,15 +214,18 @@ func (dir *Dir) Lookup(ctx context.Context, req *fuse.LookupRequest, resp *fuse.
glog.V(4).Infof("dir Lookup %s: %s by %s", dir.FullPath(), req.Name, req.Header.String())
dirPath := dir.FullPath()
fullFilePath := util.NewFullPath(dirPath, req.Name)
fullFilePath := util.NewFullPath(dir.FullPath(), req.Name)
entry := dir.wfs.cacheGet(fullFilePath)
dirPath := util.FullPath(dir.FullPath())
meta_cache.EnsureVisited(dir.wfs.metaCache, dir.wfs, util.FullPath(dirPath))
cachedEntry, cacheErr := dir.wfs.metaCache.FindEntry(context.Background(), fullFilePath)
if cacheErr == filer_pb.ErrNotFound {
return nil, fuse.ENOENT
if dir.wfs.option.AsyncMetaDataCaching {
cachedEntry, cacheErr := dir.wfs.metaCache.FindEntry(context.Background(), fullFilePath)
if cacheErr == filer_pb.ErrNotFound {
return nil, fuse.ENOENT
}
entry = cachedEntry.ToProtoEntry()
}
entry := cachedEntry.ToProtoEntry()
if entry == nil {
// glog.V(3).Infof("dir Lookup cache miss %s", fullFilePath)
@@ -222,6 +234,7 @@ func (dir *Dir) Lookup(ctx context.Context, req *fuse.LookupRequest, resp *fuse.
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)
}
@@ -253,6 +266,7 @@ func (dir *Dir) ReadDirAll(ctx context.Context) (ret []fuse.Dirent, err error) {
glog.V(3).Infof("dir ReadDirAll %s", dir.FullPath())
cacheTtl := 5 * time.Minute
processEachEntryFn := func(entry *filer_pb.Entry, isLast bool) error {
fullpath := util.NewFullPath(dir.FullPath(), entry.Name)
inode := fullpath.AsInode()
@@ -263,21 +277,31 @@ func (dir *Dir) ReadDirAll(ctx context.Context) (ret []fuse.Dirent, err error) {
dirent := fuse.Dirent{Inode: inode, Name: entry.Name, Type: fuse.DT_File}
ret = append(ret, dirent)
}
dir.wfs.cacheSet(fullpath, entry, cacheTtl)
return nil
}
dirPath := util.FullPath(dir.FullPath())
meta_cache.EnsureVisited(dir.wfs.metaCache, dir.wfs, dirPath)
listedEntries, listErr := dir.wfs.metaCache.ListDirectoryEntries(context.Background(), dirPath, "", false, int(dir.wfs.option.DirListCacheLimit))
if listErr != nil {
glog.Errorf("list meta cache: %v", listErr)
return nil, fuse.EIO
if dir.wfs.option.AsyncMetaDataCaching {
listedEntries, listErr := dir.wfs.metaCache.ListDirectoryEntries(context.Background(), util.FullPath(dir.FullPath()), "", false, int(dir.wfs.option.DirListCacheLimit))
if listErr != nil {
glog.Errorf("list meta cache: %v", listErr)
return nil, fuse.EIO
}
for _, cachedEntry := range listedEntries {
processEachEntryFn(cachedEntry.ToProtoEntry(), false)
}
return
}
for _, cachedEntry := range listedEntries {
processEachEntryFn(cachedEntry.ToProtoEntry(), false)
}
return
readErr := filer_pb.ReadDirAllEntries(dir.wfs, util.FullPath(dir.FullPath()), "", processEachEntryFn)
if readErr != nil {
glog.V(0).Infof("list %s: %v", dir.FullPath(), err)
return ret, fuse.EIO
}
return ret, err
}
func (dir *Dir) Remove(ctx context.Context, req *fuse.RemoveRequest) error {
@@ -303,7 +327,12 @@ func (dir *Dir) removeOneFile(req *fuse.RemoveRequest) error {
dir.wfs.deleteFileChunks(entry.Chunks)
dir.wfs.metaCache.DeleteEntry(context.Background(), filePath)
dir.wfs.cacheDelete(filePath)
dir.wfs.fsNodeCache.DeleteFsNode(filePath)
if dir.wfs.option.AsyncMetaDataCaching {
dir.wfs.metaCache.DeleteEntry(context.Background(), filePath)
}
glog.V(3).Infof("remove file: %v", req)
err = filer_pb.Remove(dir.wfs, dir.FullPath(), req.Name, false, false, false)
@@ -319,8 +348,12 @@ func (dir *Dir) removeOneFile(req *fuse.RemoveRequest) error {
func (dir *Dir) removeFolder(req *fuse.RemoveRequest) error {
t := util.NewFullPath(dir.FullPath(), req.Name)
dir.wfs.cacheDelete(t)
dir.wfs.fsNodeCache.DeleteFsNode(t)
dir.wfs.metaCache.DeleteEntry(context.Background(), t)
if dir.wfs.option.AsyncMetaDataCaching {
dir.wfs.metaCache.DeleteEntry(context.Background(), t)
}
glog.V(3).Infof("remove directory entry: %v", req)
err := filer_pb.Remove(dir.wfs, dir.FullPath(), req.Name, true, false, false)
@@ -357,6 +390,8 @@ 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.FullPath()))
return dir.saveEntry()
}
@@ -373,6 +408,8 @@ func (dir *Dir) Setxattr(ctx context.Context, req *fuse.SetxattrRequest) error {
return err
}
dir.wfs.cacheDelete(util.FullPath(dir.FullPath()))
return dir.saveEntry()
}
@@ -389,6 +426,8 @@ func (dir *Dir) Removexattr(ctx context.Context, req *fuse.RemovexattrRequest) e
return err
}
dir.wfs.cacheDelete(util.FullPath(dir.FullPath()))
return dir.saveEntry()
}
@@ -411,6 +450,8 @@ func (dir *Dir) Listxattr(ctx context.Context, req *fuse.ListxattrRequest, resp
func (dir *Dir) Forget() {
glog.V(3).Infof("Forget dir %s", dir.FullPath())
dir.wfs.fsNodeCache.DeleteFsNode(util.FullPath(dir.FullPath()))
}
func (dir *Dir) maybeLoadEntry() error {
@@ -443,7 +484,9 @@ func (dir *Dir) saveEntry() error {
return fuse.EIO
}
dir.wfs.metaCache.UpdateEntry(context.Background(), filer2.FromPbEntry(request.Directory, request.Entry))
if dir.wfs.option.AsyncMetaDataCaching {
dir.wfs.metaCache.UpdateEntry(context.Background(), filer2.FromPbEntry(request.Directory, request.Entry))
}
return nil
})