(cache-dir) can set cache concurrency, now defaults to :shared

pkg/bass/cache_path.go

@@ -12,13 +12,22 @@ import (
 
 // CachePath is a Path within an ephemeral directory managed by the runtime.
 type CachePath struct {
-	ID   string
-	Path FileOrDirPath
+	ID              string
+	ConcurrencyMode ConcurrencyMode
+	Path            FileOrDirPath
 }
 
+type ConcurrencyMode int
+
+const (
+	ConcurrencyModeShared = iota
+	ConcurrencyModePrivate
+	ConcurrencyModeLocked
+)
+
 var _ Value = CachePath{}
 
-func NewCacheDir(id string) CachePath {
+func NewCacheDir(id string, mode ...ConcurrencyMode) CachePath {
 	return NewCachePath(id, ParseFileOrDirPath("."))
 }
 
@@ -86,6 +95,7 @@ func (path *CachePath) UnmarshalProto(msg proto.Message) error {
 	}
 
 	path.ID = p.Id
+	path.ConcurrencyMode = ConcurrencyMode(p.Concurrency)
 
 	return path.Path.UnmarshalProto(p.Path)
 }

pkg/bass/ground.go

@@ -801,9 +801,31 @@ func init() {
 	)
 
 	Ground.Set("cache-dir",
-		Func("cache-dir", "[id]", NewCacheDir),
-		`returns a cache directory corresponding to the string identifier`,
-		`Cache directories may be mounted to thunks. Their content persists across thunk runs.`)
+		Func("cache-dir", "[id & mode]", func(id string, mode ...Symbol) (CachePath, error) {
+			cache := NewCacheDir(id)
+			if len(mode) > 0 {
+				switch mode[0] {
+				case "locked":
+					cache.ConcurrencyMode = ConcurrencyModeLocked
+				case "private":
+					cache.ConcurrencyMode = ConcurrencyModePrivate
+				case "shared":
+					cache.ConcurrencyMode = ConcurrencyModeShared
+				default:
+					return CachePath{}, fmt.Errorf("invalid cache mode: %s", mode[0])
+				}
+			}
+
+			return cache, nil
+		}),
+		`returns a cache directory with the given identifier and concurrency mode`,
+		`Cache directories may be mounted to thunks. Their content persists across thunk runs.`,
+		`The :shared concurrency mode allows multiple thunks to run concurrently with the same cache.`,
+		`The :locked mode allows only one thunk to run at a time with the same cache.`,
+		`The :private mode creates a new mount for concurrent thunks.`,
+		`The default mode is :shared.`,
+		`=> (cache-dir "foo")`,
+		`=> (cache-dir "foo" :locked)`)
 
 	Ground.Set("binds?",
 		Func("binds?", "[scope sym]", (*Scope).Binds),

pkg/bass/proto.go

@@ -77,8 +77,9 @@ func FromProto(val *proto.Value) (Value, error) {
 		}, nil
 	case *proto.Value_CachePath:
 		return CachePath{
-			ID:   x.CachePath.Id,
-			Path: fod(x.CachePath.Path),
+			ID:              x.CachePath.Id,
+			Path:            fod(x.CachePath.Path),
+			ConcurrencyMode: ConcurrencyMode(x.CachePath.Concurrency),
 		}, nil
 	case *proto.Value_LogicalPath:
 		fsp := &FSPath{}
@@ -308,7 +309,8 @@ func (value *FSPath) MarshalProto() (proto.Message, error) {
 
 func (value CachePath) MarshalProto() (proto.Message, error) {
 	pv := &proto.CachePath{
-		Id: value.ID,
+		Id:          value.ID,
+		Concurrency: proto.ConcurrencyMode(value.ConcurrencyMode),
 	}
 
 	pathp, err := value.Path.MarshalProto()

pkg/runtimes/buildkit.go

@@ -1312,10 +1312,20 @@ func (b *buildkitBuilder) initializeMount(ctx context.Context, source bass.Thunk
 		return llb.AddMount(targetPath, st, llb.SourcePath(sp)), sp, false, nil
 
 	case source.Cache != nil:
+		var mode llb.CacheMountSharingMode
+		switch source.Cache.ConcurrencyMode {
+		case bass.ConcurrencyModeShared:
+			mode = llb.CacheMountShared
+		case bass.ConcurrencyModePrivate:
+			mode = llb.CacheMountPrivate
+		case bass.ConcurrencyModeLocked:
+			mode = llb.CacheMountLocked
+		}
+
 		return llb.AddMount(
 			targetPath,
 			llb.Scratch(),
-			llb.AsPersistentCacheDir(source.Cache.ID, llb.CacheMountLocked),
+			llb.AsPersistentCacheDir(source.Cache.ID, mode),
 			llb.SourcePath(source.Cache.Path.FilesystemPath().FromSlash()),
 		), "", false, nil
 

pkg/runtimes/dagger.go

@@ -413,11 +413,21 @@ func (runtime *Dagger) mount(ctx context.Context, client *dagger.Client, ctr *da
 			return nil, fmt.Errorf("mounting subpaths of cache not implemented yet: %s", fsp.Slash())
 		}
 
+		var mode dagger.CacheSharingMode
+		switch src.Cache.ConcurrencyMode {
+		case bass.ConcurrencyModeShared:
+			mode = dagger.Shared
+		case bass.ConcurrencyModePrivate:
+			mode = dagger.Private
+		case bass.ConcurrencyModeLocked:
+			mode = dagger.Locked
+		}
+
 		return ctr.WithMountedCache(
 			target,
 			client.CacheVolume(src.Cache.ID),
 			dagger.ContainerWithMountedCacheOpts{
-				Sharing: dagger.Locked,
+				Sharing: mode,
 			},
 		), nil
 	case src.FSPath != nil:

pkg/runtimes/suite.go

@@ -169,18 +169,6 @@ func Suite(t *testing.T, runtimeConfig bass.RuntimeConfig, opts ...SuiteOpt) {
 		},
 		{
 			File: "cache-sync.bass",
-			Result: bass.NewList(
-				bass.NewList(bass.String("1")),
-				bass.NewList(bass.String("2")),
-				bass.NewList(bass.String("3")),
-				bass.NewList(bass.String("4")),
-				bass.NewList(bass.String("5")),
-				bass.NewList(bass.String("6")),
-				bass.NewList(bass.String("7")),
-				bass.NewList(bass.String("8")),
-				bass.NewList(bass.String("9")),
-				bass.NewList(bass.String("10")),
-			),
 		},
 		{
 			File:   "cache-cmd.bass",

pkg/runtimes/testdata/cache-sync.bass

@@ -1,5 +1,5 @@
 (def test-cache-path
-  (cache-dir (str "test-cache-sync-" (now 0))))
+  (cache-dir (str "test-cache-sync-" (now 0)) :locked))
 
 (defn counter [tag]
   (subpath
@@ -18,7 +18,8 @@
   (let [files (map counter tags)]
     (-> ($ sh -c "cat $@ | sort -n" ignored & $files)
         (with-image (linux/alpine))
-        (read :unix-table)
+        (read :lines)
         all)))
 
-(counts "a" "b" "c" "d" "e" "f" "g" "h" "i" "j")
+(assert = ["1" "2" "3" "4" "5" "6" "7" "8" "9" "10"]
+  (counts "a" "b" "c" "d" "e" "f" "g" "h" "i" "j"))

proto/bass.proto

@@ -163,6 +163,13 @@ message String {
 message CachePath {
   string id = 1;
   FilesystemPath path = 2;
+  ConcurrencyMode concurrency = 3;
+};
+
+enum ConcurrencyMode {
+  CONCURRENCY_MODE_SHARED = 0;
+  CONCURRENCY_MODE_PRIVATE = 1;
+  CONCURRENCY_MODE_LOCKED = 2;
 };
 
 message Secret {