Merge branch 'upstreamMaster' into iamapipr

1 files changed, 66 insertions, 21 deletions

diff --git a/weed/filesys/file.go b/weed/filesys/file.go
index a210c5152..2433be590 100644
--- a/weed/filesys/file.go
+++ b/weed/filesys/file.go
@@ -5,6 +5,7 @@ import (
 	"io"
 	"os"
 	"sort"
+	"sync"
 	"time"
 
 	"github.com/seaweedfs/fuse"
@@ -33,6 +34,7 @@ type File struct {
 	dir            *Dir
 	wfs            *WFS
 	entry          *filer_pb.Entry
+	entryLock      sync.RWMutex
 	entryViewCache []filer.VisibleInterval
 	isOpen         int
 	reader         io.ReaderAt
@@ -47,13 +49,17 @@ func (file *File) Attr(ctx context.Context, attr *fuse.Attr) (err error) {
 
 	glog.V(4).Infof("file Attr %s, open:%v existing:%v", file.fullpath(), file.isOpen, attr)
 
-	entry := file.entry
+	entry := file.getEntry()
 	if file.isOpen <= 0 || entry == nil {
 		if entry, err = file.maybeLoadEntry(ctx); err != nil {
 			return err
 		}
 	}
 
+	if entry == nil {
+		return fuse.ENOENT
+	}
+
 	// attr.Inode = file.fullpath().AsInode()
 	attr.Valid = time.Second
 	attr.Mode = os.FileMode(entry.Attributes.FileMode)
@@ -104,9 +110,13 @@ 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 file.wfs.option.ReadOnly {
+		return fuse.EPERM
+	}
+
 	glog.V(4).Infof("%v file setattr %+v", file.fullpath(), req)
 
-	_, err := file.maybeLoadEntry(ctx)
+	entry, err := file.maybeLoadEntry(ctx)
 	if err != nil {
 		return err
 	}
@@ -123,12 +133,12 @@ func (file *File) Setattr(ctx context.Context, req *fuse.SetattrRequest, resp *f
 
 	if req.Valid.Size() {
 
-		glog.V(4).Infof("%v file setattr set size=%v chunks=%d", file.fullpath(), req.Size, len(file.entry.Chunks))
-		if req.Size < filer.FileSize(file.entry) {
+		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)
 			var chunks []*filer_pb.FileChunk
 			var truncatedChunks []*filer_pb.FileChunk
-			for _, chunk := range file.entry.Chunks {
+			for _, chunk := range entry.Chunks {
 				int64Size := int64(chunk.Size)
 				if chunk.Offset+int64Size > int64(req.Size) {
 					// this chunk is truncated
@@ -143,36 +153,36 @@ func (file *File) Setattr(ctx context.Context, req *fuse.SetattrRequest, resp *f
 					}
 				}
 			}
-			file.entry.Chunks = chunks
+			entry.Chunks = chunks
 			file.entryViewCache, _ = filer.NonOverlappingVisibleIntervals(file.wfs.LookupFn(), chunks)
-			file.reader = nil
+			file.setReader(nil)
 		}
-		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
 	}
 
@@ -188,12 +198,16 @@ func (file *File) Setattr(ctx context.Context, req *fuse.SetattrRequest, resp *f
 		return nil
 	}
 
-	return file.saveEntry(file.entry)
+	return file.saveEntry(entry)
 
 }
 
 func (file *File) Setxattr(ctx context.Context, req *fuse.SetxattrRequest) error {
 
+	if file.wfs.option.ReadOnly {
+		return fuse.EPERM
+	}
+
 	glog.V(4).Infof("file Setxattr %s: %s", file.fullpath(), req.Name)
 
 	entry, err := file.maybeLoadEntry(ctx)
@@ -211,6 +225,10 @@ func (file *File) Setxattr(ctx context.Context, req *fuse.SetxattrRequest) error
 
 func (file *File) Removexattr(ctx context.Context, req *fuse.RemovexattrRequest) error {
 
+	if file.wfs.option.ReadOnly {
+		return fuse.EPERM
+	}
+
 	glog.V(4).Infof("file Removexattr %s: %s", file.fullpath(), req.Name)
 
 	entry, err := file.maybeLoadEntry(ctx)
@@ -255,10 +273,12 @@ func (file *File) Forget() {
 	t := util.NewFullPath(file.dir.FullPath(), file.Name)
 	glog.V(4).Infof("Forget file %s", t)
 	file.wfs.fsNodeCache.DeleteFsNode(t)
+	file.wfs.ReleaseHandle(t, 0)
+	file.setReader(nil)
 }
 
 func (file *File) maybeLoadEntry(ctx context.Context) (entry *filer_pb.Entry, err error) {
-	entry = file.entry
+	entry = file.getEntry()
 	if file.isOpen > 0 {
 		return entry, nil
 	}
@@ -299,8 +319,13 @@ func (file *File) addChunks(chunks []*filer_pb.FileChunk) {
 		}
 	}
 
+	entry := file.getEntry()
+	if entry == nil {
+		return
+	}
+
 	// pick out-of-order chunks from existing chunks
-	for _, chunk := range file.entry.Chunks {
+	for _, chunk := range entry.Chunks {
 		if lessThan(earliestChunk, chunk) {
 			chunks = append(chunks, chunk)
 		}
@@ -316,23 +341,37 @@ func (file *File) addChunks(chunks []*filer_pb.FileChunk) {
 		file.entryViewCache = filer.MergeIntoVisibles(file.entryViewCache, chunk)
 	}
 
-	file.reader = nil
+	file.setReader(nil)
 
-	glog.V(4).Infof("%s existing %d chunks adds %d more", file.fullpath(), len(file.entry.Chunks), len(chunks))
+	glog.V(4).Infof("%s existing %d chunks adds %d more", file.fullpath(), len(entry.Chunks), len(chunks))
 
-	file.entry.Chunks = append(file.entry.Chunks, newChunks...)
+	entry.Chunks = append(entry.Chunks, newChunks...)
+}
+
+func (file *File) setReader(reader io.ReaderAt) {
+	r := file.reader
+	if r != nil {
+		if closer, ok := r.(io.Closer); ok {
+			closer.Close()
+		}
+	}
+	file.reader = reader
 }
 
 func (file *File) setEntry(entry *filer_pb.Entry) {
+	file.entryLock.Lock()
+	defer file.entryLock.Unlock()
 	file.entry = entry
 	file.entryViewCache, _ = filer.NonOverlappingVisibleIntervals(file.wfs.LookupFn(), entry.Chunks)
-	file.reader = nil
+	file.setReader(nil)
 }
 
 func (file *File) clearEntry() {
+	file.entryLock.Lock()
+	defer file.entryLock.Unlock()
 	file.entry = nil
 	file.entryViewCache = nil
-	file.reader = nil
+	file.setReader(nil)
 }
 
 func (file *File) saveEntry(entry *filer_pb.Entry) error {
@@ -359,3 +398,9 @@ func (file *File) saveEntry(entry *filer_pb.Entry) error {
 		return nil
 	})
 }
+
+func (file *File) getEntry() *filer_pb.Entry {
+	file.entryLock.RLock()
+	defer file.entryLock.RUnlock()
+	return file.entry
+}