1 files changed, 192 insertions, 87 deletions

diff --git a/weed/filesys/file.go b/weed/filesys/file.go
index afe78ee0f..bb57988cd 100644
--- a/weed/filesys/file.go
+++ b/weed/filesys/file.go
@@ -3,20 +3,22 @@ package filesys
 import (
 	"context"
 	"os"
-	"path/filepath"
 	"sort"
 	"time"
 
-	"github.com/chrislusf/seaweedfs/weed/filer2"
-	"github.com/chrislusf/seaweedfs/weed/glog"
-	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/fuse"
 	"github.com/seaweedfs/fuse/fs"
+
+	"github.com/chrislusf/seaweedfs/weed/filer"
+	"github.com/chrislusf/seaweedfs/weed/glog"
+	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
+	"github.com/chrislusf/seaweedfs/weed/util"
 )
 
 const blockSize = 512
 
 var _ = fs.Node(&File{})
+var _ = fs.NodeIdentifier(&File{})
 var _ = fs.NodeOpener(&File{})
 var _ = fs.NodeFsyncer(&File{})
 var _ = fs.NodeSetattrer(&File{})
@@ -24,35 +26,56 @@ var _ = fs.NodeGetxattrer(&File{})
 var _ = fs.NodeSetxattrer(&File{})
 var _ = fs.NodeRemovexattrer(&File{})
 var _ = fs.NodeListxattrer(&File{})
+var _ = fs.NodeForgetter(&File{})
 
 type File struct {
-	Name           string
-	dir            *Dir
-	wfs            *WFS
-	entry          *filer_pb.Entry
-	entryViewCache []filer2.VisibleInterval
-	isOpen         bool
+	Name          string
+	dir           *Dir
+	wfs           *WFS
+	entry         *filer_pb.Entry
+	isOpen        int
+	dirtyMetadata bool
+	id            uint64
+}
+
+func (file *File) fullpath() util.FullPath {
+	return util.NewFullPath(file.dir.FullPath(), file.Name)
 }
 
-func (file *File) fullpath() string {
-	return filepath.Join(file.dir.Path, file.Name)
+func (file *File) Id() uint64 {
+	return file.id
 }
 
-func (file *File) Attr(ctx context.Context, attr *fuse.Attr) error {
+func (file *File) Attr(ctx context.Context, attr *fuse.Attr) (err error) {
 
-	glog.V(4).Infof("file Attr %s", file.fullpath())
+	glog.V(4).Infof("file Attr %s, open:%v existing:%v", file.fullpath(), file.isOpen, attr)
 
-	if err := file.maybeLoadEntry(ctx); err != nil {
+	entry, err := file.maybeLoadEntry(ctx)
+	if err != nil {
 		return err
 	}
 
-	attr.Mode = os.FileMode(file.entry.Attributes.FileMode)
-	attr.Size = filer2.TotalSize(file.entry.Chunks)
-	attr.Mtime = time.Unix(file.entry.Attributes.Mtime, 0)
-	attr.Gid = file.entry.Attributes.Gid
-	attr.Uid = file.entry.Attributes.Uid
+	if entry == nil {
+		return fuse.ENOENT
+	}
+
+	attr.Inode = file.Id()
+	attr.Valid = time.Second
+	attr.Mode = os.FileMode(entry.Attributes.FileMode)
+	attr.Size = filer.FileSize(entry)
+	if file.isOpen > 0 {
+		attr.Size = entry.Attributes.FileSize
+		glog.V(4).Infof("file Attr %s, open:%v, size: %d", file.fullpath(), file.isOpen, attr.Size)
+	}
+	attr.Crtime = time.Unix(entry.Attributes.Crtime, 0)
+	attr.Mtime = time.Unix(entry.Attributes.Mtime, 0)
+	attr.Gid = entry.Attributes.Gid
+	attr.Uid = entry.Attributes.Uid
 	attr.Blocks = attr.Size/blockSize + 1
 	attr.BlockSize = uint32(file.wfs.option.ChunkSizeLimit)
+	if entry.HardLinkCounter > 0 {
+		attr.Nlink = uint32(entry.HardLinkCounter)
+	}
 
 	return nil
 
@@ -62,24 +85,23 @@ func (file *File) Getxattr(ctx context.Context, req *fuse.GetxattrRequest, resp
 
 	glog.V(4).Infof("file Getxattr %s", file.fullpath())
 
-	if err := file.maybeLoadEntry(ctx); err != nil {
+	entry, err := file.maybeLoadEntry(ctx)
+	if err != nil {
 		return err
 	}
 
-	return getxattr(file.entry, req, resp)
+	return getxattr(entry, req, resp)
 }
 
 func (file *File) Open(ctx context.Context, req *fuse.OpenRequest, resp *fuse.OpenResponse) (fs.Handle, error) {
 
 	glog.V(4).Infof("file %v open %+v", file.fullpath(), req)
 
-	file.isOpen = true
-
 	handle := file.wfs.AcquireHandle(file, req.Uid, req.Gid)
 
 	resp.Handle = fuse.HandleID(handle.handle)
 
-	glog.V(3).Infof("%v file open handle id = %d", file.fullpath(), handle.handle)
+	glog.V(4).Infof("%v file open handle id = %d", file.fullpath(), handle.handle)
 
 	return handle, nil
 
@@ -87,48 +109,89 @@ func (file *File) Open(ctx context.Context, req *fuse.OpenRequest, resp *fuse.Op
 
 func (file *File) Setattr(ctx context.Context, req *fuse.SetattrRequest, resp *fuse.SetattrResponse) error {
 
-	if err := file.maybeLoadEntry(ctx); err != nil {
+	glog.V(4).Infof("%v file setattr %+v", file.fullpath(), req)
+
+	entry, err := file.maybeLoadEntry(ctx)
+	if err != nil {
 		return err
 	}
+	if file.isOpen > 0 {
+		file.wfs.handlesLock.Lock()
+		fileHandle := file.wfs.handles[file.Id()]
+		file.wfs.handlesLock.Unlock()
+
+		if fileHandle != nil {
+			fileHandle.Lock()
+			defer fileHandle.Unlock()
+		}
+	}
 
-	glog.V(3).Infof("%v file setattr %+v, old:%+v", file.fullpath(), req, file.entry.Attributes)
 	if req.Valid.Size() {
 
-		glog.V(3).Infof("%v file setattr set size=%v", file.fullpath(), req.Size)
-		if req.Size == 0 {
+		glog.V(4).Infof("%v file setattr set size=%v chunks=%d", file.fullpath(), req.Size, len(entry.Chunks))
+		if req.Size < filer.FileSize(entry) {
 			// fmt.Printf("truncate %v \n", fullPath)
-			file.entry.Chunks = nil
-			file.entryViewCache = nil
+			var chunks []*filer_pb.FileChunk
+			var truncatedChunks []*filer_pb.FileChunk
+			for _, chunk := range entry.Chunks {
+				int64Size := int64(chunk.Size)
+				if chunk.Offset+int64Size > int64(req.Size) {
+					// this chunk is truncated
+					int64Size = int64(req.Size) - chunk.Offset
+					if int64Size > 0 {
+						chunks = append(chunks, chunk)
+						glog.V(4).Infof("truncated chunk %+v from %d to %d\n", chunk.GetFileIdString(), chunk.Size, int64Size)
+						chunk.Size = uint64(int64Size)
+					} else {
+						glog.V(4).Infof("truncated whole chunk %+v\n", chunk.GetFileIdString())
+						truncatedChunks = append(truncatedChunks, chunk)
+					}
+				}
+			}
+			entry.Chunks = chunks
 		}
-		file.entry.Attributes.FileSize = req.Size
+		entry.Attributes.FileSize = req.Size
+		file.dirtyMetadata = true
 	}
+
 	if req.Valid.Mode() {
-		file.entry.Attributes.FileMode = uint32(req.Mode)
+		entry.Attributes.FileMode = uint32(req.Mode)
+		file.dirtyMetadata = true
 	}
 
 	if req.Valid.Uid() {
-		file.entry.Attributes.Uid = req.Uid
+		entry.Attributes.Uid = req.Uid
+		file.dirtyMetadata = true
 	}
 
 	if req.Valid.Gid() {
-		file.entry.Attributes.Gid = req.Gid
+		entry.Attributes.Gid = req.Gid
+		file.dirtyMetadata = true
 	}
 
 	if req.Valid.Crtime() {
-		file.entry.Attributes.Crtime = req.Crtime.Unix()
+		entry.Attributes.Crtime = req.Crtime.Unix()
+		file.dirtyMetadata = true
 	}
 
 	if req.Valid.Mtime() {
-		file.entry.Attributes.Mtime = req.Mtime.Unix()
+		entry.Attributes.Mtime = req.Mtime.Unix()
+		file.dirtyMetadata = true
+	}
+
+	if req.Valid.Handle() {
+		// fmt.Printf("file handle => %d\n", req.Handle)
 	}
 
-	if file.isOpen {
+	if file.isOpen > 0 {
 		return nil
 	}
 
-	file.wfs.listDirectoryEntriesCache.Delete(file.fullpath())
+	if !file.dirtyMetadata {
+		return nil
+	}
 
-	return file.saveEntry(ctx)
+	return file.saveEntry(entry)
 
 }
 
@@ -136,17 +199,16 @@ func (file *File) Setxattr(ctx context.Context, req *fuse.SetxattrRequest) error
 
 	glog.V(4).Infof("file Setxattr %s: %s", file.fullpath(), req.Name)
 
-	if err := file.maybeLoadEntry(ctx); err != nil {
+	entry, err := file.maybeLoadEntry(ctx)
+	if err != nil {
 		return err
 	}
 
-	if err := setxattr(file.entry, req); err != nil {
+	if err := setxattr(entry, req); err != nil {
 		return err
 	}
 
-	file.wfs.listDirectoryEntriesCache.Delete(file.fullpath())
-
-	return file.saveEntry(ctx)
+	return file.saveEntry(entry)
 
 }
 
@@ -154,17 +216,16 @@ func (file *File) Removexattr(ctx context.Context, req *fuse.RemovexattrRequest)
 
 	glog.V(4).Infof("file Removexattr %s: %s", file.fullpath(), req.Name)
 
-	if err := file.maybeLoadEntry(ctx); err != nil {
+	entry, err := file.maybeLoadEntry(ctx)
+	if err != nil {
 		return err
 	}
 
-	if err := removexattr(file.entry, req); err != nil {
+	if err := removexattr(entry, req); err != nil {
 		return err
 	}
 
-	file.wfs.listDirectoryEntriesCache.Delete(file.fullpath())
-
-	return file.saveEntry(ctx)
+	return file.saveEntry(entry)
 
 }
 
@@ -172,11 +233,12 @@ func (file *File) Listxattr(ctx context.Context, req *fuse.ListxattrRequest, res
 
 	glog.V(4).Infof("file Listxattr %s", file.fullpath())
 
-	if err := file.maybeLoadEntry(ctx); err != nil {
+	entry, err := file.maybeLoadEntry(ctx)
+	if err != nil {
 		return err
 	}
 
-	if err := listxattr(file.entry, req, resp); err != nil {
+	if err := listxattr(entry, req, resp); err != nil {
 		return err
 	}
 
@@ -187,69 +249,112 @@ func (file *File) Listxattr(ctx context.Context, req *fuse.ListxattrRequest, res
 func (file *File) Fsync(ctx context.Context, req *fuse.FsyncRequest) error {
 	// fsync works at OS level
 	// write the file chunks to the filerGrpcAddress
-	glog.V(3).Infof("%s/%s fsync file %+v", file.dir.Path, file.Name, req)
+	glog.V(4).Infof("%s/%s fsync file %+v", file.dir.FullPath(), file.Name, req)
 
 	return nil
 }
 
-func (file *File) maybeLoadEntry(ctx context.Context) error {
-	if file.entry == nil || !file.isOpen {
-		entry, err := file.wfs.maybeLoadEntry(ctx, file.dir.Path, file.Name)
-		if err != nil {
-			return err
-		}
-		if entry != nil {
-			file.setEntry(entry)
+func (file *File) Forget() {
+	t := util.NewFullPath(file.dir.FullPath(), file.Name)
+	glog.V(4).Infof("Forget file %s", t)
+	file.wfs.ReleaseHandle(t, fuse.HandleID(t.AsInode()))
+}
+
+func (file *File) maybeLoadEntry(ctx context.Context) (entry *filer_pb.Entry, err error) {
+
+	file.wfs.handlesLock.Lock()
+	handle, found := file.wfs.handles[file.Id()]
+	file.wfs.handlesLock.Unlock()
+	entry = file.entry
+	if found {
+		glog.V(4).Infof("maybeLoadEntry found opened file %s/%s: %v %v", file.dir.FullPath(), file.Name, handle.f.entry, entry)
+		entry = handle.f.entry
+	}
+
+	if entry != nil {
+		if len(entry.HardLinkId) == 0 {
+			// only always reload hard link
+			return entry, nil
 		}
 	}
-	return nil
+	entry, err = file.wfs.maybeLoadEntry(file.dir.FullPath(), file.Name)
+	if err != nil {
+		glog.V(3).Infof("maybeLoadEntry file %s/%s: %v", file.dir.FullPath(), file.Name, err)
+		return entry, err
+	}
+	if entry != nil {
+		// file.entry = entry
+	} else {
+		glog.Warningf("maybeLoadEntry not found entry %s/%s: %v", file.dir.FullPath(), file.Name, err)
+	}
+	return entry, nil
 }
 
-func (file *File) addChunk(chunk *filer_pb.FileChunk) {
-	if chunk != nil {
-		file.addChunks([]*filer_pb.FileChunk{chunk})
+func lessThan(a, b *filer_pb.FileChunk) bool {
+	if a.Mtime == b.Mtime {
+		return a.Fid.FileKey < b.Fid.FileKey
 	}
+	return a.Mtime < b.Mtime
 }
 
 func (file *File) addChunks(chunks []*filer_pb.FileChunk) {
 
-	sort.Slice(chunks, func(i, j int) bool {
-		return chunks[i].Mtime < chunks[j].Mtime
-	})
+	// find the earliest incoming chunk
+	newChunks := chunks
+	earliestChunk := newChunks[0]
+	for i := 1; i < len(newChunks); i++ {
+		if lessThan(earliestChunk, newChunks[i]) {
+			earliestChunk = newChunks[i]
+		}
+	}
 
-	var newVisibles []filer2.VisibleInterval
-	for _, chunk := range chunks {
-		newVisibles = filer2.MergeIntoVisibles(file.entryViewCache, newVisibles, chunk)
-		t := file.entryViewCache[:0]
-		file.entryViewCache = newVisibles
-		newVisibles = t
+	entry := file.getEntry()
+	if entry == nil {
+		return
 	}
 
-	glog.V(3).Infof("%s existing %d chunks adds %d more", file.fullpath(), len(file.entry.Chunks), len(chunks))
+	// pick out-of-order chunks from existing chunks
+	for _, chunk := range entry.Chunks {
+		if lessThan(earliestChunk, chunk) {
+			chunks = append(chunks, chunk)
+		}
+	}
 
-	file.entry.Chunks = append(file.entry.Chunks, chunks...)
-}
+	// sort incoming chunks
+	sort.Slice(chunks, func(i, j int) bool {
+		return lessThan(chunks[i], chunks[j])
+	})
 
-func (file *File) setEntry(entry *filer_pb.Entry) {
-	file.entry = entry
-	file.entryViewCache = filer2.NonOverlappingVisibleIntervals(file.entry.Chunks)
+	glog.V(4).Infof("%s existing %d chunks adds %d more", file.fullpath(), len(entry.Chunks), len(chunks))
+
+	entry.Chunks = append(entry.Chunks, newChunks...)
 }
 
-func (file *File) saveEntry(ctx context.Context) error {
-	return file.wfs.WithFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error {
+func (file *File) saveEntry(entry *filer_pb.Entry) error {
+	return file.wfs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
+
+		file.wfs.mapPbIdFromLocalToFiler(entry)
+		defer file.wfs.mapPbIdFromFilerToLocal(entry)
 
 		request := &filer_pb.UpdateEntryRequest{
-			Directory: file.dir.Path,
-			Entry:     file.entry,
+			Directory:  file.dir.FullPath(),
+			Entry:      entry,
+			Signatures: []int32{file.wfs.signature},
 		}
 
-		glog.V(1).Infof("save file entry: %v", request)
-		_, err := client.UpdateEntry(ctx, request)
+		glog.V(4).Infof("save file entry: %v", request)
+		_, err := client.UpdateEntry(context.Background(), request)
 		if err != nil {
-			glog.V(0).Infof("UpdateEntry file %s/%s: %v", file.dir.Path, file.Name, err)
+			glog.Errorf("UpdateEntry file %s/%s: %v", file.dir.FullPath(), file.Name, err)
 			return fuse.EIO
 		}
 
+		file.wfs.metaCache.UpdateEntry(context.Background(), filer.FromPbEntry(request.Directory, request.Entry))
+
 		return nil
 	})
 }
+
+func (file *File) getEntry() *filer_pb.Entry {
+	return file.entry
+}