Merge pull request #1876 from wswsmao/main

fuse passthrough: fix cache files closed by lru in passthrough model

cache/cache.go

@@ -97,7 +97,8 @@ type Writer interface {
 }
 
 type cacheOpt struct {
-	direct bool
+	direct      bool
+	passThrough bool
 }
 
 type Option func(o *cacheOpt) *cacheOpt
@@ -113,6 +114,15 @@ func Direct() Option {
 	}
 }
 
+// PassThrough option indicates whether to enable FUSE passthrough mode
+// to improve local file read performance.
+func PassThrough() Option {
+	return func(o *cacheOpt) *cacheOpt {
+		o.passThrough = true
+		return o
+	}
+}
+
 func NewDirectoryCache(directory string, config DirectoryCacheConfig) (BlobCache, error) {
 	if !filepath.IsAbs(directory) {
 		return nil, fmt.Errorf("dir cache path must be an absolute path; got %q", directory)
@@ -232,8 +242,16 @@ func (dc *directoryCache) Get(key string, opts ...Option) (Reader, error) {
 	// that won't be accessed immediately.
 	if dc.direct || opt.direct {
 		return &reader{
-			ReaderAt:  file,
-			closeFunc: func() error { return file.Close() },
+			ReaderAt: file,
+			closeFunc: func() error {
+				// In passthough model, close will be toke over by go-fuse
+				// If "passThrough" option is specified, "direct" option also will
+				// be specified, so adding this branch here is enough
+				if opt.passThrough {
+					return nil
+				}
+				return file.Close()
+			},
 		}, nil
 	}
 

fs/reader/reader.go

@@ -516,7 +516,8 @@ func (sf *file) GetPassthroughFd() (uintptr, error) {
 	id := genID(sf.id, firstChunkOffset, totalSize)
 
 	for {
-		r, err := sf.gr.cache.Get(id)
+		// cache.PassThrough() is necessary to take over files
+		r, err := sf.gr.cache.Get(id, cache.PassThrough())
 		if err != nil {
 			if err := sf.prefetchEntireFile(); err != nil {
 				return 0, err
@@ -563,6 +564,7 @@ func (sf *file) prefetchEntireFile() error {
 		ip := b.Bytes()[:chunkSize]
 
 		// Check if the content exists in the cache
+		// Just read it and merge to a new files, so cache.PassThrough() should not be used here
 		if r, err := sf.gr.cache.Get(id); err == nil {
 			n, err := r.ReadAt(ip, 0)
 			if (err == nil || err == io.EOF) && int64(n) == chunkSize {