Merge branch 'master' of https://github.com/bingoohuang/seaweedfs

1 files changed, 219 insertions, 78 deletions

diff --git a/weed/filesys/filehandle.go b/weed/filesys/filehandle.go
index 101f5c056..27ffab6e1 100644
--- a/weed/filesys/filehandle.go
+++ b/weed/filesys/filehandle.go
@@ -3,39 +3,51 @@ package filesys
 import (
 	"context"
 	"fmt"
-	"mime"
-	"path"
+	"io"
+	"math"
+	"net/http"
+	"os"
+	"sync"
 	"time"
 
-	"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"
+
+	"github.com/chrislusf/seaweedfs/weed/filer"
+	"github.com/chrislusf/seaweedfs/weed/glog"
+	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
 )
 
 type FileHandle struct {
 	// cache file has been written to
-	dirtyPages    *ContinuousDirtyPages
-	contentType   string
-	dirtyMetadata bool
-	handle        uint64
+	dirtyPages     *ContinuousDirtyPages
+	entryViewCache []filer.VisibleInterval
+	reader         io.ReaderAt
+	contentType    string
+	handle         uint64
+	sync.Mutex
 
 	f         *File
 	RequestId fuse.RequestID // unique ID for request
 	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,
 	}
+	entry := fh.f.getEntry()
+	if entry != nil {
+		entry.Attributes.FileSize = filer.FileSize(entry)
+	}
+
+	return fh
 }
 
 var _ = fs.Handle(&FileHandle{})
@@ -48,134 +60,263 @@ 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)", fh.f.fullpath(), fh.handle, req.Offset, req.Offset+int64(req.Size))
+	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.Lock()
+	defer fh.Unlock()
 
-	// 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)
+	if req.Size <= 0 {
 		return nil
 	}
 
-	buff := make([]byte, req.Size)
-
-	if fh.f.entryViewCache == nil {
-		fh.f.entryViewCache = filer2.NonOverlappingVisibleIntervals(fh.f.entry.Chunks)
+	buff := resp.Data[:cap(resp.Data)]
+	if req.Size > cap(resp.Data) {
+		// should not happen
+		buff = make([]byte, req.Size)
 	}
 
-	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)
+	totalRead, err := fh.readFromChunks(buff, req.Offset)
+	if err == nil || err == io.EOF {
+		maxStop := fh.readFromDirtyPages(buff, req.Offset)
+		totalRead = max(maxStop-req.Offset, totalRead)
+	}
 
-	resp.Data = buff[:totalRead]
+	if err == io.EOF {
+		err = nil
+	}
 
 	if err != nil {
-		glog.Errorf("file handle read %s: %v", fh.f.fullpath(), err)
+		glog.Warningf("file handle read %s %d: %v", fh.f.fullpath(), totalRead, err)
+		return fuse.EIO
+	}
+
+	if totalRead > int64(len(buff)) {
+		glog.Warningf("%s FileHandle Read %d: [%d,%d) size %d totalRead %d", fh.f.fullpath(), fh.handle, req.Offset, req.Offset+int64(req.Size), req.Size, totalRead)
+		totalRead = min(int64(len(buff)), totalRead)
+	}
+	if err == nil {
+		resp.Data = buff[:totalRead]
 	}
 
 	return err
 }
 
-// Write to the file handle
-func (fh *FileHandle) Write(ctx context.Context, req *fuse.WriteRequest, resp *fuse.WriteResponse) error {
+func (fh *FileHandle) readFromDirtyPages(buff []byte, startOffset int64) (maxStop int64) {
+	maxStop = fh.dirtyPages.ReadDirtyDataAt(buff, startOffset)
+	return
+}
 
-	// write the request to volume servers
+func (fh *FileHandle) readFromChunks(buff []byte, offset int64) (int64, error) {
 
-	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)))
+	entry := fh.f.getEntry()
+	if entry == nil {
+		return 0, io.EOF
+	}
 
-	chunks, err := fh.dirtyPages.AddPage(ctx, req.Offset, req.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)
+	fileSize := int64(filer.FileSize(entry))
+	fileFullPath := fh.f.fullpath()
+
+	if fileSize == 0 {
+		glog.V(1).Infof("empty fh %v", fileFullPath)
+		return 0, io.EOF
 	}
 
-	resp.Size = len(req.Data)
+	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
+	}
 
-	if req.Offset == 0 {
-		// detect mime type
-		detectedMIME := mimetype.Detect(req.Data)
-		fh.contentType = detectedMIME.String()
-		if ext := path.Ext(fh.f.Name); ext != detectedMIME.Extension() {
-			fh.contentType = mime.TypeByExtension(ext)
+	var chunkResolveErr error
+	if fh.entryViewCache == nil {
+		fh.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.reader = nil
+	}
+
+	reader := fh.reader
+	if reader == nil {
+		chunkViews := filer.ViewFromVisibleIntervals(fh.entryViewCache, 0, math.MaxInt64)
+		reader = filer.NewChunkReaderAtFromClient(fh.f.wfs.LookupFn(), chunkViews, fh.f.wfs.chunkCache, fileSize)
+	}
+	fh.reader = reader
 
-		fh.dirtyMetadata = true
+	totalRead, err := reader.ReadAt(buff, offset)
+
+	if err != nil && err != io.EOF {
+		glog.Errorf("file handle read %s: %v", fileFullPath, err)
 	}
 
-	if len(chunks) > 0 {
+	glog.V(4).Infof("file handle read %s [%d,%d] %d : %v", fileFullPath, offset, offset+int64(totalRead), totalRead, err)
+
+	return int64(totalRead), err
+}
 
-		fh.f.addChunks(chunks)
+// Write to the file handle
+func (fh *FileHandle) Write(ctx context.Context, req *fuse.WriteRequest, resp *fuse.WriteResponse) error {
+
+	fh.Lock()
+	defer fh.Unlock()
+
+	// write the request to volume servers
+	data := req.Data
+	if len(data) <= 512 {
+		// fuse message cacheable size
+		data = make([]byte, len(req.Data))
+		copy(data, req.Data)
+	}
 
-		fh.dirtyMetadata = true
+	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)
+
+	resp.Size = len(data)
+
+	if req.Offset == 0 {
+		// detect mime type
+		fh.contentType = http.DetectContentType(data)
+		fh.f.dirtyMetadata = true
+	}
+
+	fh.f.dirtyMetadata = true
+
 	return nil
 }
 
 func (fh *FileHandle) Release(ctx context.Context, req *fuse.ReleaseRequest) error {
 
-	glog.V(4).Infof("%v release fh %d", fh.f.fullpath(), fh.handle)
+	glog.V(4).Infof("Release %v fh %d open=%d", fh.f.fullpath(), fh.handle, fh.f.isOpen)
+
+	fh.Lock()
+	defer fh.Unlock()
 
-	fh.dirtyPages.releaseResource()
+	fh.f.isOpen--
 
-	fh.f.wfs.ReleaseHandle(fh.f.fullpath(), fuse.HandleID(fh.handle))
+	if fh.f.isOpen <= 0 {
+		fh.f.entry = nil
+		fh.entryViewCache = nil
+		fh.reader = nil
 
-	fh.f.isOpen = false
+		fh.f.wfs.ReleaseHandle(fh.f.fullpath(), fuse.HandleID(fh.handle))
+	}
+
+	if fh.f.isOpen < 0 {
+		glog.V(0).Infof("Release reset %s open count %d => %d", fh.f.Name, fh.f.isOpen, 0)
+		fh.f.isOpen = 0
+		return nil
+	}
 
 	return nil
 }
 
 func (fh *FileHandle) Flush(ctx context.Context, req *fuse.FlushRequest) error {
-	// fflush works at fh level
-	// 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)
-	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.V(4).Infof("Flush %v fh %d", fh.f.fullpath(), fh.handle)
+
+	fh.Lock()
+	defer fh.Unlock()
+
+	if err := fh.doFlush(ctx, req.Header); err != nil {
+		glog.Errorf("Flush doFlush %s: %v", fh.f.Name, err)
+		return err
 	}
 
-	fh.f.addChunk(chunk)
+	glog.V(4).Infof("Flush %v fh %d success", fh.f.fullpath(), fh.handle)
+	return nil
+}
+
+func (fh *FileHandle) doFlush(ctx context.Context, header fuse.Header) error {
+	// flush works at fh level
+	// send the data to the OS
+	glog.V(4).Infof("doFlush %s fh %d", fh.f.fullpath(), fh.handle)
+
+	fh.dirtyPages.saveExistingPagesToStorage()
+
+	fh.dirtyPages.writeWaitGroup.Wait()
 
-	if !fh.dirtyMetadata {
+	if fh.dirtyPages.lastErr != nil {
+		glog.Errorf("%v doFlush last err: %v", fh.f.fullpath(), fh.dirtyPages.lastErr)
+		return fuse.EIO
+	}
+
+	if !fh.f.dirtyMetadata {
 		return nil
 	}
 
-	return fh.f.wfs.WithFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error {
+	err := fh.f.wfs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
+
+		entry := fh.f.getEntry()
+		if entry == nil {
+			return nil
+		}
 
-		if fh.f.entry.Attributes != nil {
-			fh.f.entry.Attributes.Mime = fh.contentType
-			fh.f.entry.Attributes.Uid = req.Uid
-			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(0777 &^ fh.f.wfs.option.Umask)
+		if entry.Attributes != nil {
+			entry.Attributes.Mime = fh.contentType
+			if entry.Attributes.Uid == 0 {
+				entry.Attributes.Uid = header.Uid
+			}
+			if entry.Attributes.Gid == 0 {
+				entry.Attributes.Gid = header.Gid
+			}
+			if entry.Attributes.Crtime == 0 {
+				entry.Attributes.Crtime = time.Now().Unix()
+			}
+			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.Path,
-			Entry:     fh.f.entry,
+			Directory:  fh.f.dir.FullPath(),
+			Entry:      entry,
+			Signatures: []int32{fh.f.wfs.signature},
 		}
 
-		glog.V(3).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(3).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(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))
 		}
 
-		chunks, garbages := filer2.CompactFileChunks(fh.f.entry.Chunks)
-		fh.f.entry.Chunks = chunks
-		// fh.f.entryViewCache = nil
+		manifestChunks, nonManifestChunks := filer.SeparateManifestChunks(entry.Chunks)
 
-		if _, err := client.CreateEntry(ctx, request); err != nil {
-			glog.Errorf("update fh: %v", err)
-			return fmt.Errorf("update fh: %v", err)
+		chunks, _ := filer.CompactFileChunks(fh.f.wfs.LookupFn(), nonManifestChunks)
+		chunks, manifestErr := filer.MaybeManifestize(fh.f.wfs.saveDataAsChunk(fh.f.fullpath()), chunks)
+		if manifestErr != nil {
+			// not good, but should be ok
+			glog.V(0).Infof("MaybeManifestize: %v", manifestErr)
 		}
+		entry.Chunks = append(chunks, manifestChunks...)
 
-		fh.f.wfs.deleteFileChunks(ctx, garbages)
-		for i, chunk := range garbages {
-			glog.V(3).Infof("garbage %s/%s chunks %d: %v [%d,%d)", fh.f.dir.Path, fh.f.Name, i, chunk.FileId, chunk.Offset, chunk.Offset+int64(chunk.Size))
+		fh.f.wfs.mapPbIdFromLocalToFiler(request.Entry)
+		defer fh.f.wfs.mapPbIdFromFilerToLocal(request.Entry)
+
+		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)
 		}
 
+		fh.f.wfs.metaCache.InsertEntry(context.Background(), filer.FromPbEntry(request.Directory, request.Entry))
+
 		return nil
 	})
+
+	if err == nil {
+		fh.f.dirtyMetadata = false
+	}
+
+	if err != nil {
+		glog.Errorf("%v fh %d flush: %v", fh.f.fullpath(), fh.handle, err)
+		return fuse.EIO
+	}
+
+	return nil
 }