diff --git a/weed/filesys/dir.go b/weed/filesys/dir.go index 3c1672911..91e42fc0a 100644 --- a/weed/filesys/dir.go +++ b/weed/filesys/dir.go @@ -146,9 +146,8 @@ func (dir *Dir) Create(ctx context.Context, req *fuse.CreateRequest, } file := node.(*File) - file.isOpen = true + file.isOpen++ fh := dir.wfs.AcquireHandle(file, req.Uid, req.Gid) - fh.dirtyMetadata = true return file, fh, nil } diff --git a/weed/filesys/dirty_page.go b/weed/filesys/dirty_page.go index f83944678..7a466b506 100644 --- a/weed/filesys/dirty_page.go +++ b/weed/filesys/dirty_page.go @@ -222,3 +222,26 @@ func max(x, y int64) int64 { } return y } +func min(x, y int64) int64 { + if x < y { + return x + } + return y +} + +func (pages *ContinuousDirtyPages) ReadDirtyData(ctx context.Context, data []byte, startOffset int64) (offset int64, size int, err error) { + bufSize := int64(len(data)) + if startOffset+bufSize < pages.Offset { + return + } + if startOffset >= pages.Offset+pages.Size { + return + } + + offset = max(pages.Offset, startOffset) + stopOffset := min(pages.Offset+pages.Size, startOffset+bufSize) + size = int(stopOffset - offset) + copy(data[offset-startOffset:], pages.Data[offset-pages.Offset:stopOffset-pages.Offset]) + + return +} diff --git a/weed/filesys/file.go b/weed/filesys/file.go index d811cb179..e15d55b5b 100644 --- a/weed/filesys/file.go +++ b/weed/filesys/file.go @@ -31,7 +31,7 @@ type File struct { wfs *WFS entry *filer_pb.Entry entryViewCache []filer2.VisibleInterval - isOpen bool + isOpen int } func (file *File) fullpath() filer2.FullPath { @@ -42,7 +42,7 @@ func (file *File) Attr(ctx context.Context, attr *fuse.Attr) error { glog.V(4).Infof("file Attr %s, open:%v, existing attr: %+v", file.fullpath(), file.isOpen, attr) - if !file.isOpen { + if file.isOpen <=0 { if err := file.maybeLoadEntry(ctx); err != nil { return err } @@ -52,7 +52,7 @@ func (file *File) Attr(ctx context.Context, attr *fuse.Attr) error { attr.Valid = time.Second attr.Mode = os.FileMode(file.entry.Attributes.FileMode) attr.Size = filer2.TotalSize(file.entry.Chunks) - if file.isOpen { + if file.isOpen > 0 { attr.Size = file.entry.Attributes.FileSize } attr.Crtime = time.Unix(file.entry.Attributes.Crtime, 0) @@ -81,7 +81,7 @@ func (file *File) Open(ctx context.Context, req *fuse.OpenRequest, resp *fuse.Op glog.V(4).Infof("file %v open %+v", file.fullpath(), req) - file.isOpen = true + file.isOpen++ handle := file.wfs.AcquireHandle(file, req.Uid, req.Gid) @@ -140,7 +140,7 @@ func (file *File) Setattr(ctx context.Context, req *fuse.SetattrRequest, resp *f file.entry.Attributes.Mtime = req.Mtime.Unix() } - if file.isOpen { + if file.isOpen > 0 { return nil } @@ -218,7 +218,7 @@ func (file *File) Forget() { } func (file *File) maybeLoadEntry(ctx context.Context) error { - if file.entry == nil || !file.isOpen { + if file.entry == nil || file.isOpen <= 0{ entry, err := file.wfs.maybeLoadEntry(ctx, file.dir.Path, file.Name) if err != nil { return err diff --git a/weed/filesys/filehandle.go b/weed/filesys/filehandle.go index 101f5c056..981de7ea2 100644 --- a/weed/filesys/filehandle.go +++ b/weed/filesys/filehandle.go @@ -7,10 +7,11 @@ import ( "path" "time" + "github.com/gabriel-vasile/mimetype" + "github.com/chrislusf/seaweedfs/weed/filer2" "github.com/chrislusf/seaweedfs/weed/glog" "github.com/chrislusf/seaweedfs/weed/pb/filer_pb" - "github.com/gabriel-vasile/mimetype" "github.com/seaweedfs/fuse" "github.com/seaweedfs/fuse/fs" ) @@ -50,29 +51,50 @@ func (fh *FileHandle) Read(ctx context.Context, req *fuse.ReadRequest, resp *fus glog.V(4).Infof("%s read fh %d: [%d,%d)", fh.f.fullpath(), fh.handle, req.Offset, req.Offset+int64(req.Size)) + buff := make([]byte, req.Size) + + totalRead, err := fh.readFromChunks(ctx, buff, req.Offset) + if err == nil { + dirtyOffset, dirtySize, dirtyReadErr := fh.readFromDirtyPages(ctx, buff, req.Offset) + if dirtyReadErr == nil && totalRead+req.Offset < dirtyOffset+int64(dirtySize) { + totalRead = dirtyOffset + int64(dirtySize) - req.Offset + } + } + + resp.Data = buff[:totalRead] + + if err != nil { + glog.Errorf("file handle read %s: %v", fh.f.fullpath(), err) + } + + return err +} + +func (fh *FileHandle) readFromDirtyPages(ctx context.Context, buff []byte, startOffset int64) (offset int64, size int, err error) { + return fh.dirtyPages.ReadDirtyData(ctx, buff, startOffset) +} + +func (fh *FileHandle) readFromChunks(ctx context.Context, buff []byte, offset int64) (int64, error) { + // this value should come from the filer instead of the old f if len(fh.f.entry.Chunks) == 0 { glog.V(1).Infof("empty fh %v/%v", fh.f.dir.Path, fh.f.Name) - return nil + return 0, nil } - buff := make([]byte, req.Size) - if fh.f.entryViewCache == nil { fh.f.entryViewCache = filer2.NonOverlappingVisibleIntervals(fh.f.entry.Chunks) } - chunkViews := filer2.ViewFromVisibleIntervals(fh.f.entryViewCache, req.Offset, req.Size) - - totalRead, err := filer2.ReadIntoBuffer(ctx, fh.f.wfs, fh.f.fullpath(), buff, chunkViews, req.Offset) + chunkViews := filer2.ViewFromVisibleIntervals(fh.f.entryViewCache, offset, len(buff)) - resp.Data = buff[:totalRead] + totalRead, err := filer2.ReadIntoBuffer(ctx, fh.f.wfs, fh.f.fullpath(), buff, chunkViews, offset) if err != nil { glog.Errorf("file handle read %s: %v", fh.f.fullpath(), err) } - return err + return totalRead, err } // Write to the file handle @@ -115,11 +137,12 @@ func (fh *FileHandle) Release(ctx context.Context, req *fuse.ReleaseRequest) err glog.V(4).Infof("%v release fh %d", fh.f.fullpath(), fh.handle) - fh.dirtyPages.releaseResource() + fh.f.isOpen-- - fh.f.wfs.ReleaseHandle(fh.f.fullpath(), fuse.HandleID(fh.handle)) - - fh.f.isOpen = false + if fh.f.isOpen <= 0 { + fh.dirtyPages.releaseResource() + fh.f.wfs.ReleaseHandle(fh.f.fullpath(), fuse.HandleID(fh.handle)) + } return nil } @@ -141,7 +164,7 @@ func (fh *FileHandle) Flush(ctx context.Context, req *fuse.FlushRequest) error { return nil } - return fh.f.wfs.WithFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error { + err = fh.f.wfs.WithFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error { if fh.f.entry.Attributes != nil { fh.f.entry.Attributes.Mime = fh.contentType @@ -178,4 +201,10 @@ func (fh *FileHandle) Flush(ctx context.Context, req *fuse.FlushRequest) error { return nil }) + + if err == nil { + fh.dirtyMetadata = false + } + + return err } diff --git a/weed/filesys/wfs.go b/weed/filesys/wfs.go index a2e5a9073..4cfab811b 100644 --- a/weed/filesys/wfs.go +++ b/weed/filesys/wfs.go @@ -49,6 +49,7 @@ type WFS struct { // contains all open handles, protected by handlesLock handlesLock sync.Mutex handles []*FileHandle + pathToHandleIndex map[filer2.FullPath]int bufPool sync.Pool @@ -68,6 +69,7 @@ func NewSeaweedFileSystem(option *Option) *WFS { wfs := &WFS{ option: option, listDirectoryEntriesCache: ccache.New(ccache.Configure().MaxSize(option.DirListCacheLimit * 3).ItemsToPrune(100)), + pathToHandleIndex: make(map[filer2.FullPath]int), bufPool: sync.Pool{ New: func() interface{} { return make([]byte, option.ChunkSizeLimit) @@ -102,11 +104,18 @@ func (wfs *WFS) AcquireHandle(file *File, uid, gid uint32) (fileHandle *FileHand wfs.handlesLock.Lock() defer wfs.handlesLock.Unlock() + index, found := wfs.pathToHandleIndex[fullpath] + if found && wfs.handles[index] != nil { + glog.V(2).Infoln(fullpath, "found fileHandle id", index) + return wfs.handles[index] + } + fileHandle = newFileHandle(file, uid, gid) for i, h := range wfs.handles { if h == nil { wfs.handles[i] = fileHandle fileHandle.handle = uint64(i) + wfs.pathToHandleIndex[fullpath] = i glog.V(4).Infof( "%s reuse fh %d", fullpath,fileHandle.handle) return } @@ -114,6 +123,7 @@ func (wfs *WFS) AcquireHandle(file *File, uid, gid uint32) (fileHandle *FileHand wfs.handles = append(wfs.handles, fileHandle) fileHandle.handle = uint64(len(wfs.handles) - 1) + wfs.pathToHandleIndex[fullpath] = int(fileHandle.handle) glog.V(4).Infof( "%s new fh %d", fullpath,fileHandle.handle) return @@ -124,6 +134,7 @@ func (wfs *WFS) ReleaseHandle(fullpath filer2.FullPath, handleId fuse.HandleID) defer wfs.handlesLock.Unlock() glog.V(4).Infof("%s ReleaseHandle id %d current handles length %d", fullpath, handleId, len(wfs.handles)) + delete(wfs.pathToHandleIndex, fullpath) if int(handleId) < len(wfs.handles) { wfs.handles[int(handleId)] = nil }