1 files changed, 100 insertions, 143 deletions

diff --git a/weed/filesys/filehandle.go b/weed/filesys/filehandle.go
index 0f6ca1164..9b9df916c 100644
--- a/weed/filesys/filehandle.go
+++ b/weed/filesys/filehandle.go
@@ -3,17 +3,17 @@ package filesys
 import (
 	"context"
 	"fmt"
+	"io"
+	"math"
+	"net/http"
+	"os"
+	"time"
+
 	"github.com/chrislusf/seaweedfs/weed/filer2"
 	"github.com/chrislusf/seaweedfs/weed/glog"
-	"github.com/chrislusf/seaweedfs/weed/operation"
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
-	"github.com/chrislusf/seaweedfs/weed/util"
 	"github.com/seaweedfs/fuse"
 	"github.com/seaweedfs/fuse/fs"
-	"net/http"
-	"strings"
-	"sync"
-	"time"
 )
 
 type FileHandle struct {
@@ -28,15 +28,20 @@ type FileHandle struct {
 	NodeId    fuse.NodeID    // file or directory the request is about
 	Uid       uint32         // user ID of process making request
 	Gid       uint32         // group ID of process making request
+
 }
 
 func newFileHandle(file *File, uid, gid uint32) *FileHandle {
-	return &FileHandle{
+	fh := &FileHandle{
 		f:          file,
 		dirtyPages: newDirtyPages(file),
 		Uid:        uid,
 		Gid:        gid,
 	}
+	if fh.f.entry != nil {
+		fh.f.entry.Attributes.FileSize = filer2.TotalSize(fh.f.entry.Chunks)
+	}
+	return fh
 }
 
 var _ = fs.Handle(&FileHandle{})
@@ -51,115 +56,91 @@ 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))
 
-	// 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
-	}
-
 	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)
-
-	var vids []string
-	for _, chunkView := range chunkViews {
-		vids = append(vids, volumeId(chunkView.FileId))
-	}
-
-	vid2Locations := make(map[string]*filer_pb.Locations)
-
-	err := fh.f.wfs.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
-
-		glog.V(4).Infof("read fh lookup volume id locations: %v", vids)
-		resp, err := client.LookupVolume(ctx, &filer_pb.LookupVolumeRequest{
-			VolumeIds: vids,
-		})
-		if err != nil {
-			return err
+	totalRead, err := fh.readFromChunks(buff, req.Offset)
+	if err == nil {
+		dirtyOffset, dirtySize := fh.readFromDirtyPages(buff, req.Offset)
+		if totalRead+req.Offset < dirtyOffset+int64(dirtySize) {
+			totalRead = dirtyOffset + int64(dirtySize) - req.Offset
 		}
+	}
 
-		vid2Locations = resp.LocationsMap
-
-		return nil
-	})
+	resp.Data = buff[:totalRead]
 
 	if err != nil {
-		glog.V(4).Infof("%v/%v read fh lookup volume ids: %v", fh.f.dir.Path, fh.f.Name, err)
-		return fmt.Errorf("failed to lookup volume ids %v: %v", vids, err)
+		glog.Errorf("file handle read %s: %v", fh.f.fullpath(), err)
+		return fuse.EIO
 	}
 
-	var totalRead int64
-	var wg sync.WaitGroup
-	for _, chunkView := range chunkViews {
-		wg.Add(1)
-		go func(chunkView *filer2.ChunkView) {
-			defer wg.Done()
+	return err
+}
 
-			glog.V(4).Infof("read fh reading chunk: %+v", chunkView)
+func (fh *FileHandle) readFromDirtyPages(buff []byte, startOffset int64) (offset int64, size int) {
+	return fh.dirtyPages.ReadDirtyData(buff, startOffset)
+}
 
-			locations := vid2Locations[volumeId(chunkView.FileId)]
-			if locations == nil || len(locations.Locations) == 0 {
-				glog.V(0).Infof("failed to locate %s", chunkView.FileId)
-				err = fmt.Errorf("failed to locate %s", chunkView.FileId)
-				return
-			}
+func (fh *FileHandle) readFromChunks(buff []byte, offset int64) (int64, error) {
 
-			var n int64
-			n, err = util.ReadUrl(
-				fmt.Sprintf("http://%s/%s", locations.Locations[0].Url, chunkView.FileId),
-				chunkView.Offset,
-				int(chunkView.Size),
-				buff[chunkView.LogicOffset-req.Offset:chunkView.LogicOffset-req.Offset+int64(chunkView.Size)],
-				!chunkView.IsFullChunk)
+	// 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", fh.f.fullpath())
+		return 0, nil
+	}
 
-			if err != nil {
+	if fh.f.entryViewCache == nil {
+		fh.f.entryViewCache = filer2.NonOverlappingVisibleIntervals(fh.f.entry.Chunks)
+		fh.f.reader = nil
+	}
 
-				glog.V(0).Infof("%v/%v read http://%s/%v %v bytes: %v", fh.f.dir.Path, fh.f.Name, locations.Locations[0].Url, chunkView.FileId, n, err)
+	if fh.f.reader == nil {
+		chunkViews := filer2.ViewFromVisibleIntervals(fh.f.entryViewCache, 0, math.MaxInt32)
+		fh.f.reader = filer2.NewChunkReaderAtFromClient(fh.f.wfs, chunkViews, fh.f.wfs.chunkCache)
+	}
 
-				err = fmt.Errorf("failed to read http://%s/%s: %v",
-					locations.Locations[0].Url, chunkView.FileId, err)
-				return
-			}
+	totalRead, err := fh.f.reader.ReadAt(buff, offset)
 
-			glog.V(4).Infof("read fh read %d bytes: %+v", n, chunkView)
-			totalRead += n
+	if err == io.EOF {
+		err = nil
+	}
 
-		}(chunkView)
+	if err != nil {
+		glog.Errorf("file handle read %s: %v", fh.f.fullpath(), err)
 	}
-	wg.Wait()
 
-	resp.Data = buff[:totalRead]
+	// glog.V(0).Infof("file handle read %s [%d,%d] %d : %v", fh.f.fullpath(), offset, offset+int64(totalRead), totalRead, err)
 
-	return err
+	return int64(totalRead), err
 }
 
 // Write to the file handle
 func (fh *FileHandle) Write(ctx context.Context, req *fuse.WriteRequest, resp *fuse.WriteResponse) error {
 
 	// write the request to volume servers
+	data := make([]byte, len(req.Data))
+	copy(data, req.Data)
 
-	glog.V(4).Infof("%+v/%v write fh %d: [%d,%d)", fh.f.dir.Path, fh.f.Name, fh.handle, req.Offset, req.Offset+int64(len(req.Data)))
+	fh.f.entry.Attributes.FileSize = uint64(max(req.Offset+int64(len(data)), int64(fh.f.entry.Attributes.FileSize)))
+	// glog.V(0).Infof("%v write [%d,%d)", fh.f.fullpath(), req.Offset, req.Offset+int64(len(req.Data)))
 
-	chunks, err := fh.dirtyPages.AddPage(ctx, req.Offset, req.Data)
+	chunks, err := fh.dirtyPages.AddPage(req.Offset, data)
 	if err != nil {
-		glog.Errorf("%+v/%v write fh %d: [%d,%d): %v", fh.f.dir.Path, fh.f.Name, fh.handle, req.Offset, req.Offset+int64(len(req.Data)), err)
-		return fmt.Errorf("write %s/%s at [%d,%d): %v", fh.f.dir.Path, fh.f.Name, req.Offset, req.Offset+int64(len(req.Data)), err)
+		glog.Errorf("%v write fh %d: [%d,%d): %v", fh.f.fullpath(), fh.handle, req.Offset, req.Offset+int64(len(data)), err)
+		return fuse.EIO
 	}
 
-	resp.Size = len(req.Data)
+	resp.Size = len(data)
 
 	if req.Offset == 0 {
-		fh.contentType = http.DetectContentType(req.Data)
+		// detect mime type
+		fh.contentType = http.DetectContentType(data)
 		fh.dirtyMetadata = true
 	}
 
-	fh.f.addChunks(chunks)
-
 	if len(chunks) > 0 {
+
+		fh.f.addChunks(chunks)
+
 		fh.dirtyMetadata = true
 	}
 
@@ -170,11 +151,14 @@ 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.wfs.ReleaseHandle(fh.f.fullpath(), fuse.HandleID(fh.handle))
+	fh.f.isOpen--
 
-	fh.f.isOpen = false
+	if fh.f.isOpen <= 0 {
+		fh.dirtyPages.releaseResource()
+		fh.f.wfs.ReleaseHandle(fh.f.fullpath(), fuse.HandleID(fh.handle))
+	}
+	fh.f.entryViewCache = nil
+	fh.f.reader = nil
 
 	return nil
 }
@@ -184,19 +168,22 @@ func (fh *FileHandle) Flush(ctx context.Context, req *fuse.FlushRequest) error {
 	// send the data to the OS
 	glog.V(4).Infof("%s fh %d flush %v", fh.f.fullpath(), fh.handle, req)
 
-	chunk, err := fh.dirtyPages.FlushToStorage(ctx)
+	chunks, err := fh.dirtyPages.FlushToStorage()
 	if err != nil {
-		glog.Errorf("flush %s/%s: %v", fh.f.dir.Path, fh.f.Name, err)
-		return fmt.Errorf("flush %s/%s: %v", fh.f.dir.Path, fh.f.Name, err)
+		glog.Errorf("flush %s: %v", fh.f.fullpath(), err)
+		return fuse.EIO
 	}
 
-	fh.f.addChunk(chunk)
+	if len(chunks) > 0 {
+		fh.f.addChunks(chunks)
+		fh.dirtyMetadata = true
+	}
 
 	if !fh.dirtyMetadata {
 		return nil
 	}
 
-	return fh.f.wfs.withFilerClient(func(client filer_pb.SeaweedFilerClient) 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
@@ -204,78 +191,48 @@ func (fh *FileHandle) Flush(ctx context.Context, req *fuse.FlushRequest) error {
 			fh.f.entry.Attributes.Gid = req.Gid
 			fh.f.entry.Attributes.Mtime = time.Now().Unix()
 			fh.f.entry.Attributes.Crtime = time.Now().Unix()
-			fh.f.entry.Attributes.FileMode = uint32(0770)
+			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
 		}
 
 		request := &filer_pb.CreateEntryRequest{
-			Directory: fh.f.dir.Path,
+			Directory: fh.f.dir.FullPath(),
 			Entry:     fh.f.entry,
 		}
 
-		//glog.V(1).Infof("%s/%s set chunks: %v", fh.f.dir.Path, fh.f.Name, len(fh.f.entry.Chunks))
-		//for i, chunk := range fh.f.entry.Chunks {
-		//	glog.V(4).Infof("%s/%s chunks %d: %v [%d,%d)", fh.f.dir.Path, fh.f.Name, i, chunk.FileId, chunk.Offset, chunk.Offset+int64(chunk.Size))
-		//}
+		glog.V(3).Infof("%s set chunks: %v", fh.f.fullpath(), len(fh.f.entry.Chunks))
+		for i, chunk := range fh.f.entry.Chunks {
+			glog.V(3).Infof("%s chunks %d: %v [%d,%d)", fh.f.fullpath(), i, chunk.FileId, chunk.Offset, chunk.Offset+int64(chunk.Size))
+		}
 
 		chunks, garbages := filer2.CompactFileChunks(fh.f.entry.Chunks)
 		fh.f.entry.Chunks = chunks
 		// fh.f.entryViewCache = nil
-		fh.f.wfs.deleteFileChunks(garbages)
 
-		if _, err := client.CreateEntry(ctx, request); err != nil {
-			return fmt.Errorf("update fh: %v", err)
+		if err := filer_pb.CreateEntry(client, request); err != nil {
+			glog.Errorf("fh flush create %s: %v", fh.f.fullpath(), err)
+			return fmt.Errorf("fh flush create %s: %v", fh.f.fullpath(), err)
 		}
 
-		return nil
-	})
-}
-
-func deleteFileIds(ctx context.Context, client filer_pb.SeaweedFilerClient, fileIds []string) error {
-
-	var vids []string
-	for _, fileId := range fileIds {
-		vids = append(vids, volumeId(fileId))
-	}
-
-	lookupFunc := func(vids []string) (map[string]operation.LookupResult, error) {
+		fh.f.wfs.metaCache.InsertEntry(context.Background(), filer2.FromPbEntry(request.Directory, request.Entry))
 
-		m := make(map[string]operation.LookupResult)
-
-		glog.V(4).Infof("remove file lookup volume id locations: %v", vids)
-		resp, err := client.LookupVolume(ctx, &filer_pb.LookupVolumeRequest{
-			VolumeIds: vids,
-		})
-		if err != nil {
-			return m, err
+		fh.f.wfs.deleteFileChunks(garbages)
+		for i, chunk := range garbages {
+			glog.V(3).Infof("garbage %s chunks %d: %v [%d,%d)", fh.f.fullpath(), i, chunk.FileId, chunk.Offset, chunk.Offset+int64(chunk.Size))
 		}
 
-		for _, vid := range vids {
-			lr := operation.LookupResult{
-				VolumeId:  vid,
-				Locations: nil,
-			}
-			locations := resp.LocationsMap[vid]
-			for _, loc := range locations.Locations {
-				lr.Locations = append(lr.Locations, operation.Location{
-					Url:       loc.Url,
-					PublicUrl: loc.PublicUrl,
-				})
-			}
-			m[vid] = lr
-		}
+		return nil
+	})
 
-		return m, err
+	if err == nil {
+		fh.dirtyMetadata = false
 	}
 
-	_, err := operation.DeleteFilesWithLookupVolumeId(fileIds, lookupFunc)
-
-	return err
-}
-
-func volumeId(fileId string) string {
-	lastCommaIndex := strings.LastIndex(fileId, ",")
-	if lastCommaIndex > 0 {
-		return fileId[:lastCommaIndex]
+	if err != nil {
+		glog.Errorf("%v fh %d flush: %v", fh.f.fullpath(), fh.handle, err)
+		return fuse.EIO
 	}
-	return fileId
+
+	return nil
 }