fuse passthrough: fix oom when running huge images

Signed-off-by: abushwang <[email protected]>

Author: wswsmao

fs/reader/reader.go

@@ -519,7 +519,7 @@ func (sf *file) GetPassthroughFd() (uintptr, error) {
 	// cache.PassThrough() is necessary to take over files
 	r, err := sf.gr.cache.Get(id, cache.PassThrough())
 	if err != nil {
-		if id, err = sf.prefetchEntireFile(); err != nil {
+		if err := sf.prefetchEntireFile(id); err != nil {
 			return 0, err
 		}
 
@@ -542,15 +542,12 @@ func (sf *file) GetPassthroughFd() (uintptr, error) {
 	return fd, nil
 }
 
-func (sf *file) prefetchEntireFile() (string, error) {
+func (sf *file) prefetchEntireFile(entireCacheID string) error {
 	var (
 		offset           int64
 		firstChunkOffset int64 = -1
 		totalSize        int64
 	)
-	combinedBuffer := sf.gr.bufPool.Get().(*bytes.Buffer)
-	combinedBuffer.Reset()
-	defer sf.gr.putBuffer(combinedBuffer)
 
 	for {
 		chunkOffset, chunkSize, chunkDigestStr, ok := sf.fr.ChunkEntryForOffset(offset)
@@ -572,7 +569,11 @@ func (sf *file) prefetchEntireFile() (string, error) {
 		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 {
-				combinedBuffer.Write(ip[:n])
+				if err := sf.gr.appendData(ip[:n], entireCacheID); err != nil {
+					r.Close()
+					sf.gr.putBuffer(b)
+					return fmt.Errorf("failed to append cached data: %w", err)
+				}
 				totalSize += int64(n)
 				offset = chunkOffset + int64(n)
 				r.Close()
@@ -585,21 +586,21 @@ func (sf *file) prefetchEntireFile() (string, error) {
 		// cache miss, prefetch the whole chunk
 		if _, err := sf.fr.ReadAt(ip, chunkOffset); err != nil && err != io.EOF {
 			sf.gr.putBuffer(b)
-			return "", fmt.Errorf("failed to read data: %w", err)
+			return fmt.Errorf("failed to read data: %w", err)
 		}
 		if err := sf.gr.verifyOneChunk(sf.id, ip, chunkDigestStr); err != nil {
 			sf.gr.putBuffer(b)
-			return "", err
+			return err
+		}
+		if err := sf.gr.appendData(ip, entireCacheID); err != nil {
+			sf.gr.putBuffer(b)
+			return fmt.Errorf("failed to append fetched data: %w", err)
 		}
-		combinedBuffer.Write(ip)
 		totalSize += chunkSize
 		offset = chunkOffset + chunkSize
 		sf.gr.putBuffer(b)
 	}
-	combinedIP := combinedBuffer.Bytes()
-	combinedID := genID(sf.id, firstChunkOffset, totalSize)
-	sf.gr.cacheData(combinedIP, combinedID)
-	return combinedID, nil
+	return nil
 }
 
 func (gr *reader) verifyOneChunk(entryID uint32, ip []byte, chunkDigestStr string) error {
@@ -624,6 +625,31 @@ func (gr *reader) cacheData(ip []byte, cacheID string) {
 	}
 }
 
+func (gr *reader) appendData(ip []byte, cacheID string) error {
+	r, err := gr.cache.Get(cacheID)
+	if err != nil {
+		gr.cacheData(ip, cacheID)
+		return nil
+	}
+	defer r.Close()
+
+	readerAt := r.GetReaderAt()
+	file, ok := readerAt.(*os.File)
+	if !ok {
+		return fmt.Errorf("the cached ReaderAt is not of type *os.File")
+	}
+
+	if _, err := file.Seek(0, io.SeekEnd); err != nil {
+		return fmt.Errorf("failed to seek to end of file: %w", err)
+	}
+
+	if _, err := file.Write(ip); err != nil {
+		return fmt.Errorf("failed to append data to file: %w", err)
+	}
+
+	return nil
+}
+
 func (gr *reader) verifyAndCache(entryID uint32, ip []byte, chunkDigestStr string, cacheID string) error {
 	if err := gr.verifyOneChunk(entryID, ip, chunkDigestStr); err != nil {
 		return err