Merge branch 'upstreamMaster' into iamapipr

1 files changed, 56 insertions, 40 deletions

diff --git a/weed/filesys/filehandle.go b/weed/filesys/filehandle.go
index fb073c9cd..4419888c4 100644
--- a/weed/filesys/filehandle.go
+++ b/weed/filesys/filehandle.go
@@ -23,7 +23,7 @@ type FileHandle struct {
 	dirtyPages  *ContinuousDirtyPages
 	contentType string
 	handle      uint64
-	sync.RWMutex
+	sync.Mutex
 
 	f         *File
 	RequestId fuse.RequestID // unique ID for request
@@ -40,8 +40,9 @@ func newFileHandle(file *File, uid, gid uint32) *FileHandle {
 		Uid:        uid,
 		Gid:        gid,
 	}
-	if fh.f.entry != nil {
-		fh.f.entry.Attributes.FileSize = filer.FileSize(fh.f.entry)
+	entry := fh.f.getEntry()
+	if entry != nil {
+		entry.Attributes.FileSize = filer.FileSize(entry)
 	}
 
 	return fh
@@ -58,8 +59,8 @@ var _ = fs.HandleReleaser(&FileHandle{})
 func (fh *FileHandle) Read(ctx context.Context, req *fuse.ReadRequest, resp *fuse.ReadResponse) error {
 
 	glog.V(4).Infof("%s read fh %d: [%d,%d) size %d resp.Data cap=%d", fh.f.fullpath(), fh.handle, req.Offset, req.Offset+int64(req.Size), req.Size, cap(resp.Data))
-	fh.RLock()
-	defer fh.RUnlock()
+	fh.Lock()
+	defer fh.Unlock()
 
 	if req.Size <= 0 {
 		return nil
@@ -104,26 +105,32 @@ func (fh *FileHandle) readFromDirtyPages(buff []byte, startOffset int64) (maxSto
 
 func (fh *FileHandle) readFromChunks(buff []byte, offset int64) (int64, error) {
 
-	fileSize := int64(filer.FileSize(fh.f.entry))
+	entry := fh.f.getEntry()
+	if entry == nil {
+		return 0, io.EOF
+	}
+
+	fileSize := int64(filer.FileSize(entry))
+	fileFullPath := fh.f.fullpath()
 
 	if fileSize == 0 {
-		glog.V(1).Infof("empty fh %v", fh.f.fullpath())
+		glog.V(1).Infof("empty fh %v", fileFullPath)
 		return 0, io.EOF
 	}
 
-	if offset+int64(len(buff)) <= int64(len(fh.f.entry.Content)) {
-		totalRead := copy(buff, fh.f.entry.Content[offset:])
-		glog.V(4).Infof("file handle read cached %s [%d,%d] %d", fh.f.fullpath(), offset, offset+int64(totalRead), totalRead)
+	if offset+int64(len(buff)) <= int64(len(entry.Content)) {
+		totalRead := copy(buff, entry.Content[offset:])
+		glog.V(4).Infof("file handle read cached %s [%d,%d] %d", fileFullPath, offset, offset+int64(totalRead), totalRead)
 		return int64(totalRead), nil
 	}
 
 	var chunkResolveErr error
 	if fh.f.entryViewCache == nil {
-		fh.f.entryViewCache, chunkResolveErr = filer.NonOverlappingVisibleIntervals(fh.f.wfs.LookupFn(), fh.f.entry.Chunks)
+		fh.f.entryViewCache, chunkResolveErr = filer.NonOverlappingVisibleIntervals(fh.f.wfs.LookupFn(), entry.Chunks)
 		if chunkResolveErr != nil {
 			return 0, fmt.Errorf("fail to resolve chunk manifest: %v", chunkResolveErr)
 		}
-		fh.f.reader = nil
+		fh.f.setReader(nil)
 	}
 
 	reader := fh.f.reader
@@ -131,15 +138,15 @@ func (fh *FileHandle) readFromChunks(buff []byte, offset int64) (int64, error) {
 		chunkViews := filer.ViewFromVisibleIntervals(fh.f.entryViewCache, 0, math.MaxInt64)
 		reader = filer.NewChunkReaderAtFromClient(fh.f.wfs.LookupFn(), chunkViews, fh.f.wfs.chunkCache, fileSize)
 	}
-	fh.f.reader = reader
+	fh.f.setReader(reader)
 
 	totalRead, err := reader.ReadAt(buff, offset)
 
 	if err != nil && err != io.EOF {
-		glog.Errorf("file handle read %s: %v", fh.f.fullpath(), err)
+		glog.Errorf("file handle read %s: %v", fileFullPath, err)
 	}
 
-	glog.V(4).Infof("file handle read %s [%d,%d] %d : %v", fh.f.fullpath(), offset, offset+int64(totalRead), totalRead, err)
+	glog.V(4).Infof("file handle read %s [%d,%d] %d : %v", fileFullPath, offset, offset+int64(totalRead), totalRead, err)
 
 	return int64(totalRead), err
 }
@@ -147,6 +154,10 @@ func (fh *FileHandle) readFromChunks(buff []byte, offset int64) (int64, error) {
 // Write to the file handle
 func (fh *FileHandle) Write(ctx context.Context, req *fuse.WriteRequest, resp *fuse.WriteResponse) error {
 
+	if fh.f.wfs.option.ReadOnly {
+		return fuse.EPERM
+	}
+
 	fh.Lock()
 	defer fh.Unlock()
 
@@ -158,8 +169,13 @@ func (fh *FileHandle) Write(ctx context.Context, req *fuse.WriteRequest, resp *f
 		copy(data, req.Data)
 	}
 
-	fh.f.entry.Content = nil
-	fh.f.entry.Attributes.FileSize = uint64(max(req.Offset+int64(len(data)), int64(fh.f.entry.Attributes.FileSize)))
+	entry := fh.f.getEntry()
+	if entry == nil {
+		return fuse.EIO
+	}
+
+	entry.Content = nil
+	entry.Attributes.FileSize = uint64(max(req.Offset+int64(len(data)), int64(entry.Attributes.FileSize)))
 	glog.V(4).Infof("%v write [%d,%d) %d", fh.f.fullpath(), req.Offset, req.Offset+int64(len(req.Data)), len(req.Data))
 
 	fh.dirtyPages.AddPage(req.Offset, data)
@@ -195,12 +211,7 @@ func (fh *FileHandle) Release(ctx context.Context, req *fuse.ReleaseRequest) err
 		fh.f.isOpen--
 
 		fh.f.wfs.ReleaseHandle(fh.f.fullpath(), fuse.HandleID(fh.handle))
-		if closer, ok := fh.f.reader.(io.Closer); ok {
-			if closer != nil {
-				closer.Close()
-			}
-		}
-		fh.f.reader = nil
+		fh.f.setReader(nil)
 	}
 
 	return nil
@@ -242,35 +253,40 @@ func (fh *FileHandle) doFlush(ctx context.Context, header fuse.Header) error {
 
 	err := fh.f.wfs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
 
-		if fh.f.entry.Attributes != nil {
-			fh.f.entry.Attributes.Mime = fh.contentType
-			if fh.f.entry.Attributes.Uid == 0 {
-				fh.f.entry.Attributes.Uid = header.Uid
+		entry := fh.f.getEntry()
+		if entry == nil {
+			return nil
+		}
+
+		if entry.Attributes != nil {
+			entry.Attributes.Mime = fh.contentType
+			if entry.Attributes.Uid == 0 {
+				entry.Attributes.Uid = header.Uid
 			}
-			if fh.f.entry.Attributes.Gid == 0 {
-				fh.f.entry.Attributes.Gid = header.Gid
+			if entry.Attributes.Gid == 0 {
+				entry.Attributes.Gid = header.Gid
 			}
-			if fh.f.entry.Attributes.Crtime == 0 {
-				fh.f.entry.Attributes.Crtime = time.Now().Unix()
+			if entry.Attributes.Crtime == 0 {
+				entry.Attributes.Crtime = time.Now().Unix()
 			}
-			fh.f.entry.Attributes.Mtime = time.Now().Unix()
-			fh.f.entry.Attributes.FileMode = uint32(os.FileMode(fh.f.entry.Attributes.FileMode) &^ fh.f.wfs.option.Umask)
-			fh.f.entry.Attributes.Collection = fh.dirtyPages.collection
-			fh.f.entry.Attributes.Replication = fh.dirtyPages.replication
+			entry.Attributes.Mtime = time.Now().Unix()
+			entry.Attributes.FileMode = uint32(os.FileMode(entry.Attributes.FileMode) &^ fh.f.wfs.option.Umask)
+			entry.Attributes.Collection = fh.dirtyPages.collection
+			entry.Attributes.Replication = fh.dirtyPages.replication
 		}
 
 		request := &filer_pb.CreateEntryRequest{
 			Directory:  fh.f.dir.FullPath(),
-			Entry:      fh.f.entry,
+			Entry:      entry,
 			Signatures: []int32{fh.f.wfs.signature},
 		}
 
-		glog.V(4).Infof("%s set chunks: %v", fh.f.fullpath(), len(fh.f.entry.Chunks))
-		for i, chunk := range fh.f.entry.Chunks {
+		glog.V(4).Infof("%s set chunks: %v", fh.f.fullpath(), len(entry.Chunks))
+		for i, chunk := range entry.Chunks {
 			glog.V(4).Infof("%s chunks %d: %v [%d,%d)", fh.f.fullpath(), i, chunk.GetFileIdString(), chunk.Offset, chunk.Offset+int64(chunk.Size))
 		}
 
-		manifestChunks, nonManifestChunks := filer.SeparateManifestChunks(fh.f.entry.Chunks)
+		manifestChunks, nonManifestChunks := filer.SeparateManifestChunks(entry.Chunks)
 
 		chunks, _ := filer.CompactFileChunks(fh.f.wfs.LookupFn(), nonManifestChunks)
 		chunks, manifestErr := filer.MaybeManifestize(fh.f.wfs.saveDataAsChunk(fh.f.fullpath()), chunks)
@@ -278,7 +294,7 @@ func (fh *FileHandle) doFlush(ctx context.Context, header fuse.Header) error {
 			// not good, but should be ok
 			glog.V(0).Infof("MaybeManifestize: %v", manifestErr)
 		}
-		fh.f.entry.Chunks = append(chunks, manifestChunks...)
+		entry.Chunks = append(chunks, manifestChunks...)
 
 		fh.f.wfs.mapPbIdFromLocalToFiler(request.Entry)
 		defer fh.f.wfs.mapPbIdFromFilerToLocal(request.Entry)