Add a configuration item to limit parallelism

.github/workflows/build.yml

@@ -90,6 +90,9 @@ jobs:
           -
             pkg: ./cmd/buildctl ./worker/containerd
             typ: integration
+          -
+            pkg: ./solver
+            typ: integration
           -
             pkg: ''
             skip-integration-tests: 1

cmd/buildkitd/config/config.go

@@ -91,6 +91,8 @@ type OCIConfig struct {
 	// ApparmorProfile is the name of the apparmor profile that should be used to constrain build containers.
 	// The profile should already be loaded (by a higher level system) before creating a worker.
 	ApparmorProfile string `toml:"apparmor-profile"`
+
+	MaxParallelism int `toml:"max-parallelism"`
 }
 
 type ContainerdConfig struct {
@@ -106,6 +108,8 @@ type ContainerdConfig struct {
 	// ApparmorProfile is the name of the apparmor profile that should be used to constrain build containers.
 	// The profile should already be loaded (by a higher level system) before creating a worker.
 	ApparmorProfile string `toml:"apparmor-profile"`
+
+	MaxParallelism int `toml:"max-parallelism"`
 }
 
 type GCPolicy struct {

cmd/buildkitd/main_containerd_worker.go

@@ -18,6 +18,7 @@ import (
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 	"github.com/urfave/cli"
+	"golang.org/x/sync/semaphore"
 )
 
 const (
@@ -225,11 +226,16 @@ func containerdWorkerInitializer(c *cli.Context, common workerInitializerOpt) ([
 		},
 	}
 
+	var parallelismSem *semaphore.Weighted
+	if cfg.MaxParallelism > 0 {
+		parallelismSem = semaphore.NewWeighted(int64(cfg.MaxParallelism))
+	}
+
 	snapshotter := ctd.DefaultSnapshotter
 	if cfg.Snapshotter != "" {
 		snapshotter = cfg.Snapshotter
 	}
-	opt, err := containerd.NewWorkerOpt(common.config.Root, cfg.Address, snapshotter, cfg.Namespace, cfg.Labels, dns, nc, common.config.Workers.Containerd.ApparmorProfile, ctd.WithTimeout(60*time.Second))
+	opt, err := containerd.NewWorkerOpt(common.config.Root, cfg.Address, snapshotter, cfg.Namespace, cfg.Labels, dns, nc, common.config.Workers.Containerd.ApparmorProfile, parallelismSem, ctd.WithTimeout(60*time.Second))
 	if err != nil {
 		return nil, err
 	}

cmd/buildkitd/main_oci_worker.go

@@ -36,6 +36,7 @@ import (
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 	"github.com/urfave/cli"
+	"golang.org/x/sync/semaphore"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/backoff"
 )
@@ -276,7 +277,12 @@ func ociWorkerInitializer(c *cli.Context, common workerInitializerOpt) ([]worker
 		},
 	}
 
-	opt, err := runc.NewWorkerOpt(common.config.Root, snFactory, cfg.Rootless, processMode, cfg.Labels, idmapping, nc, dns, cfg.Binary, cfg.ApparmorProfile)
+	var parallelismSem *semaphore.Weighted
+	if cfg.MaxParallelism > 0 {
+		parallelismSem = semaphore.NewWeighted(int64(cfg.MaxParallelism))
+	}
+
+	opt, err := runc.NewWorkerOpt(common.config.Root, snFactory, cfg.Rootless, processMode, cfg.Labels, idmapping, nc, dns, cfg.Binary, cfg.ApparmorProfile, parallelismSem)
 	if err != nil {
 		return nil, err
 	}

solver/jobs.go

@@ -766,6 +766,11 @@ func (s *sharedOp) Exec(ctx context.Context, inputs []Result) (outputs []Result,
 		if s.execRes != nil || s.execErr != nil {
 			return s.execRes, s.execErr
 		}
+		release, err := op.Acquire(ctx)
+		if err != nil {
+			return nil, errors.Wrap(err, "acquire op resources")
+		}
+		defer release()
 
 		ctx = opentracing.ContextWithSpan(progress.WithProgress(ctx, s.st.mpw), s.st.mspan)
 		ctx = withAncestorCacheOpts(ctx, s.st)

solver/jobs_test.go

@@ -0,0 +1,107 @@
+package solver
+
+import (
+	"context"
+	"io/ioutil"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/moby/buildkit/client"
+	"github.com/moby/buildkit/client/llb"
+	"github.com/moby/buildkit/util/testutil/integration"
+	"github.com/stretchr/testify/require"
+	"golang.org/x/sync/errgroup"
+)
+
+func init() {
+	integration.InitOCIWorker()
+	integration.InitContainerdWorker()
+}
+
+func TestJobsIntegration(t *testing.T) {
+	mirrors := integration.WithMirroredImages(integration.OfficialImages("busybox:latest"))
+	integration.Run(t, []integration.Test{
+		testParallelism,
+	},
+		mirrors,
+		integration.WithMatrix("max-parallelism", map[string]interface{}{
+			"single":    maxParallelismSingle,
+			"unlimited": maxParallelismUnlimited,
+		}),
+	)
+}
+
+func testParallelism(t *testing.T, sb integration.Sandbox) {
+	ctx := context.TODO()
+
+	c, err := client.New(ctx, sb.Address())
+	require.NoError(t, err)
+	defer c.Close()
+
+	cacheMount := llb.AddMount(
+		"/shared", llb.Scratch(),
+		llb.AsPersistentCacheDir("shared", llb.CacheMountShared))
+	run1 := llb.Image("busybox:latest").Run(
+		llb.Args([]string{
+			"/bin/sh", "-c",
+			"touch /shared/signal1 && i=0; while [ ! -f /shared/signal2 ] && [ $i -lt 10 ]; do i=$((i+1)); sleep 1; done",
+		}),
+		cacheMount,
+	).Root()
+	d1, err := run1.Marshal(ctx)
+	require.NoError(t, err)
+	run2 := llb.Image("busybox:latest").Run(
+		llb.Args([]string{
+			"/bin/sh", "-c",
+			"touch /shared/signal2 && i=0; while [ ! -f /shared/signal1 ] && [ $i -lt 10 ]; do i=$((i+1)); sleep 1; done",
+		}),
+		cacheMount,
+	).Root()
+	d2, err := run2.Marshal(ctx)
+	require.NoError(t, err)
+
+	timeStart := time.Now()
+	eg, egCtx := errgroup.WithContext(ctx)
+	tmpDir, err := ioutil.TempDir("", "solver-jobs-test-")
+	require.NoError(t, err)
+	defer os.RemoveAll(tmpDir)
+	solveOpt := client.SolveOpt{
+		LocalDirs: map[string]string{"cache": tmpDir},
+	}
+	eg.Go(func() error {
+		_, err := c.Solve(egCtx, d1, solveOpt, nil)
+		return err
+	})
+	eg.Go(func() error {
+		_, err := c.Solve(egCtx, d2, solveOpt, nil)
+		return err
+	})
+	err = eg.Wait()
+	require.NoError(t, err)
+
+	elapsed := time.Since(timeStart)
+
+	maxParallelism := sb.Value("max-parallelism")
+	if maxParallelism == maxParallelismSingle {
+		require.Greater(t, elapsed, 10*time.Second, "parallelism not restricted")
+	} else if maxParallelism == maxParallelismUnlimited {
+		require.Less(t, elapsed, 10*time.Second, "parallelism hindered")
+	}
+}
+
+type parallelismSetterSingle struct{}
+
+func (*parallelismSetterSingle) UpdateConfigFile(in string) string {
+	return in + "\n\n[worker.oci]\n  max-parallelism = 1\n\n[worker.containerd]\n  max-parallelism = 1\n"
+}
+
+var maxParallelismSingle integration.ConfigUpdater = &parallelismSetterSingle{}
+
+type parallelismSetterUnlimited struct{}
+
+func (*parallelismSetterUnlimited) UpdateConfigFile(in string) string {
+	return in
+}
+
+var maxParallelismUnlimited integration.ConfigUpdater = &parallelismSetterUnlimited{}

solver/llbsolver/ops/build.go

@@ -141,3 +141,8 @@ func (b *buildOp) Exec(ctx context.Context, g session.Group, inputs []solver.Res
 
 	return []solver.Result{r}, err
 }
+
+func (b *buildOp) Acquire(ctx context.Context) (solver.ReleaseFunc, error) {
+	// buildOp itself does not count towards parallelism budget.
+	return func() {}, nil
+}

solver/llbsolver/ops/exec.go

@@ -29,33 +29,36 @@ import (
 	specs "github.com/opencontainers/image-spec/specs-go/v1"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
+	"golang.org/x/sync/semaphore"
 )
 
 const execCacheType = "buildkit.exec.v0"
 
 type execOp struct {
-	op        *pb.ExecOp
-	cm        cache.Manager
-	mm        *mounts.MountManager
-	exec      executor.Executor
-	w         worker.Worker
-	platform  *pb.Platform
-	numInputs int
+	op          *pb.ExecOp
+	cm          cache.Manager
+	mm          *mounts.MountManager
+	exec        executor.Executor
+	w           worker.Worker
+	platform    *pb.Platform
+	numInputs   int
+	parallelism *semaphore.Weighted
 }
 
-func NewExecOp(v solver.Vertex, op *pb.Op_Exec, platform *pb.Platform, cm cache.Manager, sm *session.Manager, md *metadata.Store, exec executor.Executor, w worker.Worker) (solver.Op, error) {
+func NewExecOp(v solver.Vertex, op *pb.Op_Exec, platform *pb.Platform, cm cache.Manager, parallelism *semaphore.Weighted, sm *session.Manager, md *metadata.Store, exec executor.Executor, w worker.Worker) (solver.Op, error) {
 	if err := llbsolver.ValidateOp(&pb.Op{Op: op}); err != nil {
 		return nil, err
 	}
 	name := fmt.Sprintf("exec %s", strings.Join(op.Exec.Meta.Args, " "))
 	return &execOp{
-		op:        op.Exec,
-		mm:        mounts.NewMountManager(name, cm, sm, md),
-		cm:        cm,
-		exec:      exec,
-		numInputs: len(v.Inputs()),
-		w:         w,
-		platform:  platform,
+		op:          op.Exec,
+		mm:          mounts.NewMountManager(name, cm, sm, md),
+		cm:          cm,
+		exec:        exec,
+		numInputs:   len(v.Inputs()),
+		w:           w,
+		platform:    platform,
+		parallelism: parallelism,
 	}, nil
 }
 
@@ -388,3 +391,16 @@ func parseExtraHosts(ips []*pb.HostIP) ([]executor.HostIP, error) {
 	}
 	return out, nil
 }
+
+func (e *execOp) Acquire(ctx context.Context) (solver.ReleaseFunc, error) {
+	if e.parallelism == nil {
+		return func() {}, nil
+	}
+	err := e.parallelism.Acquire(ctx, 1)
+	if err != nil {
+		return nil, err
+	}
+	return func() {
+		e.parallelism.Release(1)
+	}, nil
+}

solver/llbsolver/ops/file.go

@@ -24,28 +24,31 @@ import (
 	digest "github.com/opencontainers/go-digest"
 	"github.com/pkg/errors"
 	"golang.org/x/sync/errgroup"
+	"golang.org/x/sync/semaphore"
 )
 
 const fileCacheType = "buildkit.file.v0"
 
 type fileOp struct {
-	op        *pb.FileOp
-	md        *metadata.Store
-	w         worker.Worker
-	solver    *FileOpSolver
-	numInputs int
+	op          *pb.FileOp
+	md          *metadata.Store
+	w           worker.Worker
+	solver      *FileOpSolver
+	numInputs   int
+	parallelism *semaphore.Weighted
 }
 
-func NewFileOp(v solver.Vertex, op *pb.Op_File, cm cache.Manager, md *metadata.Store, w worker.Worker) (solver.Op, error) {
+func NewFileOp(v solver.Vertex, op *pb.Op_File, cm cache.Manager, parallelism *semaphore.Weighted, md *metadata.Store, w worker.Worker) (solver.Op, error) {
 	if err := llbsolver.ValidateOp(&pb.Op{Op: op}); err != nil {
 		return nil, err
 	}
 	return &fileOp{
-		op:        op.File,
-		md:        md,
-		numInputs: len(v.Inputs()),
-		w:         w,
-		solver:    NewFileOpSolver(w, &file.Backend{}, file.NewRefManager(cm)),
+		op:          op.File,
+		md:          md,
+		numInputs:   len(v.Inputs()),
+		w:           w,
+		solver:      NewFileOpSolver(w, &file.Backend{}, file.NewRefManager(cm)),
+		parallelism: parallelism,
 	}, nil
 }
 
@@ -179,6 +182,19 @@ func (f *fileOp) Exec(ctx context.Context, g session.Group, inputs []solver.Resu
 	return outResults, nil
 }
 
+func (f *fileOp) Acquire(ctx context.Context) (solver.ReleaseFunc, error) {
+	if f.parallelism == nil {
+		return func() {}, nil
+	}
+	err := f.parallelism.Acquire(ctx, 1)
+	if err != nil {
+		return nil, err
+	}
+	return func() {
+		f.parallelism.Release(1)
+	}, nil
+}
+
 func addSelector(m map[int]map[llbsolver.Selector]struct{}, idx int, sel string, wildcard, followLinks bool) {
 	mm, ok := m[idx]
 	if !ok {