FUSE mount: stream read data with buffer

fix https://github.com/chrislusf/seaweedfs/issues/1244

2 files changed, 17 insertions, 4 deletions

diff --git a/weed/filesys/file.go b/weed/filesys/file.go
index 69d440a73..14b9cb208 100644
--- a/weed/filesys/file.go
+++ b/weed/filesys/file.go
@@ -2,6 +2,7 @@ package filesys
 
 import (
 	"context"
+	"io"
 	"os"
 	"sort"
 	"time"
@@ -32,6 +33,7 @@ type File struct {
 	entry          *filer_pb.Entry
 	entryViewCache []filer2.VisibleInterval
 	isOpen         int
+	reader         io.ReadSeeker
 }
 
 func (file *File) fullpath() filer2.FullPath {
@@ -119,6 +121,7 @@ func (file *File) Setattr(ctx context.Context, req *fuse.SetattrRequest, resp *f
 			}
 			file.entry.Chunks = chunks
 			file.entryViewCache = nil
+			file.reader = nil
 		}
 		file.entry.Attributes.FileSize = req.Size
 	}
@@ -245,6 +248,7 @@ func (file *File) addChunks(chunks []*filer_pb.FileChunk) {
 		file.entryViewCache = newVisibles
 		newVisibles = t
 	}
+	file.reader = nil
 
 	glog.V(3).Infof("%s existing %d chunks adds %d more", file.fullpath(), len(file.entry.Chunks), len(chunks))
 
@@ -254,6 +258,7 @@ func (file *File) addChunks(chunks []*filer_pb.FileChunk) {
 func (file *File) setEntry(entry *filer_pb.Entry) {
 	file.entry = entry
 	file.entryViewCache = filer2.NonOverlappingVisibleIntervals(file.entry.Chunks)
+	file.reader = nil
 }
 
 func (file *File) saveEntry() error {
diff --git a/weed/filesys/filehandle.go b/weed/filesys/filehandle.go
index 100c9eba0..bfdafd580 100644
--- a/weed/filesys/filehandle.go
+++ b/weed/filesys/filehandle.go
@@ -3,6 +3,8 @@ package filesys
 import (
 	"context"
 	"fmt"
+	"io"
+	"math"
 	"mime"
 	"path"
 	"time"
@@ -85,17 +87,23 @@ func (fh *FileHandle) readFromChunks(buff []byte, offset int64) (int64, error) {
 
 	if fh.f.entryViewCache == nil {
 		fh.f.entryViewCache = filer2.NonOverlappingVisibleIntervals(fh.f.entry.Chunks)
+		fh.f.reader = nil
+	}
+	if fh.f.reader == nil {
+		chunkViews := filer2.ViewFromVisibleIntervals(fh.f.entryViewCache, 0, math.MaxInt32)
+		fh.f.reader = filer2.NewChunkStreamReaderFromClient(fh.f.wfs, chunkViews)
 	}
 
-	chunkViews := filer2.ViewFromVisibleIntervals(fh.f.entryViewCache, offset, len(buff))
-
-	totalRead, err := filer2.ReadIntoBuffer(fh.f.wfs, fh.f.fullpath(), buff, chunkViews, offset)
+	fh.f.reader.Seek(offset, io.SeekStart)
+	totalRead, err := fh.f.reader.Read(buff)
 
 	if err != nil {
 		glog.Errorf("file handle read %s: %v", fh.f.fullpath(), err)
 	}
 
-	return totalRead, err
+	// glog.V(0).Infof("file handle read %s [%d,%d] %d : %v", fh.f.fullpath(), offset, offset+int64(totalRead), totalRead, err)
+
+	return int64(totalRead), err
 }
 
 // Write to the file handle