Merge startup cancellation and parallel OCI pulls

Author: luke-lombardi

pkg/clip/clip.go

@@ -66,6 +66,7 @@ type MountOptions struct {
 	CachePath             string
 	ContentCache          storage.ContentCache
 	ContentCacheAvailable bool
+	StorageModeOverride   common.StorageMode
 	StorageInfo           common.ClipStorageInfo
 	Credentials           storage.ClipStorageCredentials
 	UseCheckpoints        bool        // Enable checkpoint-based partial decompression for OCI layers
@@ -229,12 +230,15 @@ func MountArchive(options MountOptions) (func() error, <-chan error, *fuse.Serve
 			go server.Serve()
 
 			if err := server.WaitMount(); err != nil {
+				_ = clipfs.closeViewFiles()
+				_ = archiveStorage.Cleanup()
 				serverError <- err
 				return
 			}
 
 			server.Wait()
-			archiveStorage.Cleanup()
+			_ = clipfs.closeViewFiles()
+			_ = archiveStorage.Cleanup()
 
 			close(serverError)
 		}()
@@ -266,6 +270,7 @@ func openArchiveStorage(options MountOptions) (storage.ClipStorageInterface, err
 		ArchivePath:           options.ArchivePath,
 		CachePath:             options.CachePath,
 		Metadata:              metadata,
+		StorageModeOverride:   options.StorageModeOverride,
 		Credentials:           options.Credentials,
 		StorageInfo:           s3Info,
 		ContentCache:          options.ContentCache,

pkg/clip/clipfs.go

@@ -2,7 +2,9 @@ package clip
 
 import (
 	"fmt"
+	"os"
 	"sync"
+	"syscall"
 
 	"github.com/beam-cloud/clip/pkg/common"
 	"github.com/beam-cloud/clip/pkg/storage"
@@ -30,6 +32,8 @@ type ClipFileSystem struct {
 	cachingStatus         map[string]bool
 	cacheEventChan        chan cacheEvent
 	cachingStatusMu       sync.Mutex
+	viewFilesMu           sync.Mutex
+	viewFiles             map[string]*os.File
 }
 
 type lookupCacheEntry struct {
@@ -49,6 +53,7 @@ func NewFileSystem(s storage.ClipStorageInterface, opts ClipFileSystemOpts) (*Cl
 		contentCacheReadAhead: storage.NewContentCacheReadAhead(opts.ContentCache, storage.ContentCacheReadAheadOptions{}),
 		cacheEventChan:        make(chan cacheEvent, 10000),
 		cachingStatus:         make(map[string]bool),
+		viewFiles:             make(map[string]*os.File),
 		contentCacheAvailable: opts.ContentCacheAvailable,
 		readTraceObserver:     opts.ReadTraceObserver,
 	}
@@ -70,6 +75,39 @@ func NewFileSystem(s storage.ClipStorageInterface, opts ClipFileSystemOpts) (*Cl
 	return cfs, nil
 }
 
+func (cfs *ClipFileSystem) openViewFile(path string) (*os.File, error) {
+	cfs.viewFilesMu.Lock()
+	defer cfs.viewFilesMu.Unlock()
+
+	if file := cfs.viewFiles[path]; file != nil {
+		return file, nil
+	}
+	if len(cfs.viewFiles) >= clipFileViewFDCacheSize {
+		return nil, syscall.EMFILE
+	}
+
+	file, err := os.Open(path)
+	if err != nil {
+		return nil, err
+	}
+	cfs.viewFiles[path] = file
+	return file, nil
+}
+
+func (cfs *ClipFileSystem) closeViewFiles() error {
+	cfs.viewFilesMu.Lock()
+	defer cfs.viewFilesMu.Unlock()
+
+	var firstErr error
+	for path, file := range cfs.viewFiles {
+		if err := file.Close(); err != nil && firstErr == nil {
+			firstErr = err
+		}
+		delete(cfs.viewFiles, path)
+	}
+	return firstErr
+}
+
 func (cfs *ClipFileSystem) Root() (fs.InodeEmbedder, error) {
 	if cfs.root == nil {
 		return nil, fmt.Errorf("root not initialized")

pkg/clip/fsnode.go

@@ -2,7 +2,6 @@ package clip
 
 import (
 	"context"
-	"os"
 	"path"
 	"strconv"
 	"sync"
@@ -25,19 +24,14 @@ type FSNode struct {
 	dirEntries     []fuse.DirEntry
 }
 
-const clipFileHandleFDCacheSize = 2048
+const clipFileViewFDCacheSize = 2048
 
 type clipFileHandle struct {
-	node  *FSNode
-	mu    sync.Mutex
-	files map[string]*os.File
+	node *FSNode
 }
 
 func newClipFileHandle(node *FSNode) *clipFileHandle {
-	return &clipFileHandle{
-		node:  node,
-		files: make(map[string]*os.File),
-	}
+	return &clipFileHandle{node: node}
 }
 
 func (fh *clipFileHandle) Read(ctx context.Context, dest []byte, off int64) (fuse.ReadResult, syscall.Errno) {
@@ -57,33 +51,7 @@ func (fh *clipFileHandle) Read(ctx context.Context, dest []byte, off int64) (fus
 }
 
 func (fh *clipFileHandle) Release(ctx context.Context) syscall.Errno {
-	fh.mu.Lock()
-	defer fh.mu.Unlock()
-	var firstErr syscall.Errno
-	for path, file := range fh.files {
-		if err := file.Close(); err != nil && firstErr == fs.OK {
-			firstErr = fs.ToErrno(err)
-		}
-		delete(fh.files, path)
-	}
-	return firstErr
-}
-
-func (fh *clipFileHandle) openViewFile(path string) (*os.File, error) {
-	fh.mu.Lock()
-	defer fh.mu.Unlock()
-	if file := fh.files[path]; file != nil {
-		return file, nil
-	}
-	if len(fh.files) >= clipFileHandleFDCacheSize {
-		return nil, syscall.EMFILE
-	}
-	file, err := os.Open(path)
-	if err != nil {
-		return nil, err
-	}
-	fh.files[path] = file
-	return file, nil
+	return fs.OK
 }
 
 func (fh *clipFileHandle) readClientLocalFileView(ctx context.Context, dest []byte, off int64) (fuse.ReadResult, bool, syscall.Errno) {
@@ -119,7 +87,7 @@ func (fh *clipFileHandle) readClientLocalFileView(ctx context.Context, dest []by
 		return nil, false, fs.OK
 	}
 
-	file, err := fh.openViewFile(view.Path)
+	file, err := fh.node.filesystem.openViewFile(view.Path)
 	if err != nil {
 		if traceRead {
 			fh.node.observeRead(ctx, fh.node.clientLocalFileViewReadTrace(view, off, readLen, 0, started, err))

pkg/clip/fsnode_test.go

@@ -8,6 +8,7 @@ import (
 	"errors"
 	"fmt"
 	"os"
+	"path/filepath"
 	"strings"
 	"sync"
 	"testing"
@@ -272,6 +273,21 @@ func TestLegacyClientLocalFileViewReadWarmsContentCache(t *testing.T) {
 	require.Equal(t, testData, cache.store[contentHash])
 }
 
+func TestClientLocalFileViewsShareBackingFile(t *testing.T) {
+	archivePath := filepath.Join(t.TempDir(), "layer")
+	require.NoError(t, os.WriteFile(archivePath, []byte("data"), 0644))
+
+	filesystem := &ClipFileSystem{viewFiles: make(map[string]*os.File)}
+	first, err := filesystem.openViewFile(archivePath)
+	require.NoError(t, err)
+	second, err := filesystem.openViewFile(archivePath)
+	require.NoError(t, err)
+	require.Same(t, first, second)
+	require.Len(t, filesystem.viewFiles, 1)
+	require.NoError(t, filesystem.closeViewFiles())
+	require.Empty(t, filesystem.viewFiles)
+}
+
 func TestImmutableFileOpenSkipsFlush(t *testing.T) {
 	node := &FSNode{clipNode: &common.ClipNode{NodeType: common.FileNode}}