diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 0316f8874f88..03fecacb0892 100644 --- a/.github/workflows/build.yml +++ b/.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 diff --git a/cmd/buildkitd/config/config.go b/cmd/buildkitd/config/config.go index 1841fefe1a01..f296e6cd2c6e 100644 --- a/cmd/buildkitd/config/config.go +++ b/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 { diff --git a/cmd/buildkitd/main_containerd_worker.go b/cmd/buildkitd/main_containerd_worker.go index cde0f8ae7c59..4e096334ab5d 100644 --- a/cmd/buildkitd/main_containerd_worker.go +++ b/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 } diff --git a/cmd/buildkitd/main_oci_worker.go b/cmd/buildkitd/main_oci_worker.go index 450dc3b3a82b..9d7d0172dd9c 100644 --- a/cmd/buildkitd/main_oci_worker.go +++ b/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 } diff --git a/solver/jobs.go b/solver/jobs.go index 1942fdf199be..20479818ed0f 100644 --- a/solver/jobs.go +++ b/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) diff --git a/solver/jobs_test.go b/solver/jobs_test.go new file mode 100644 index 000000000000..198eba64c66a --- /dev/null +++ b/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 = ¶llelismSetterSingle{} + +type parallelismSetterUnlimited struct{} + +func (*parallelismSetterUnlimited) UpdateConfigFile(in string) string { + return in +} + +var maxParallelismUnlimited integration.ConfigUpdater = ¶llelismSetterUnlimited{} diff --git a/solver/llbsolver/ops/build.go b/solver/llbsolver/ops/build.go index 5e7fbc7a15a0..39d2a7707571 100644 --- a/solver/llbsolver/ops/build.go +++ b/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 +} diff --git a/solver/llbsolver/ops/exec.go b/solver/llbsolver/ops/exec.go index d9f9b0ac2f79..b5ef0f16aa1d 100644 --- a/solver/llbsolver/ops/exec.go +++ b/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 +} diff --git a/solver/llbsolver/ops/file.go b/solver/llbsolver/ops/file.go index 554532c1845f..aeabda9bcc27 100644 --- a/solver/llbsolver/ops/file.go +++ b/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 { diff --git a/solver/llbsolver/ops/source.go b/solver/llbsolver/ops/source.go index 4f8b6694f079..6cec6320b58a 100644 --- a/solver/llbsolver/ops/source.go +++ b/solver/llbsolver/ops/source.go @@ -12,32 +12,35 @@ import ( "github.com/moby/buildkit/source" "github.com/moby/buildkit/worker" digest "github.com/opencontainers/go-digest" + "golang.org/x/sync/semaphore" ) const sourceCacheType = "buildkit.source.v0" type sourceOp struct { - mu sync.Mutex - op *pb.Op_Source - platform *pb.Platform - sm *source.Manager - src source.SourceInstance - sessM *session.Manager - w worker.Worker - vtx solver.Vertex + mu sync.Mutex + op *pb.Op_Source + platform *pb.Platform + sm *source.Manager + src source.SourceInstance + sessM *session.Manager + w worker.Worker + vtx solver.Vertex + parallelism *semaphore.Weighted } -func NewSourceOp(vtx solver.Vertex, op *pb.Op_Source, platform *pb.Platform, sm *source.Manager, sessM *session.Manager, w worker.Worker) (solver.Op, error) { +func NewSourceOp(vtx solver.Vertex, op *pb.Op_Source, platform *pb.Platform, sm *source.Manager, parallelism *semaphore.Weighted, sessM *session.Manager, w worker.Worker) (solver.Op, error) { if err := llbsolver.ValidateOp(&pb.Op{Op: op}); err != nil { return nil, err } return &sourceOp{ - op: op, - sm: sm, - w: w, - sessM: sessM, - platform: platform, - vtx: vtx, + op: op, + sm: sm, + w: w, + sessM: sessM, + platform: platform, + vtx: vtx, + parallelism: parallelism, }, nil } @@ -93,3 +96,16 @@ func (s *sourceOp) Exec(ctx context.Context, g session.Group, _ []solver.Result) } return []solver.Result{worker.NewWorkerRefResult(ref, s.w)}, nil } + +func (s *sourceOp) Acquire(ctx context.Context) (solver.ReleaseFunc, error) { + if s.parallelism == nil { + return func() {}, nil + } + err := s.parallelism.Acquire(ctx, 1) + if err != nil { + return nil, err + } + return func() { + s.parallelism.Release(1) + }, nil +} diff --git a/solver/scheduler_test.go b/solver/scheduler_test.go index 4c58f3e1ddbc..92d1d6263c15 100644 --- a/solver/scheduler_test.go +++ b/solver/scheduler_test.go @@ -3511,6 +3511,10 @@ func (v *vertex) Exec(ctx context.Context, g session.Group, inputs []Result) (ou return []Result{&dummyResult{id: identity.NewID(), value: v.opt.value}}, nil } +func (v *vertex) Acquire(ctx context.Context) (ReleaseFunc, error) { + return func() {}, nil +} + func (v *vertex) makeCacheMap() *CacheMap { m := &CacheMap{ Digest: digest.FromBytes([]byte(fmt.Sprintf("seed:%s", v.opt.cacheKeySeed))), @@ -3556,6 +3560,10 @@ func (v *vertexConst) Exec(ctx context.Context, g session.Group, inputs []Result return []Result{&dummyResult{id: identity.NewID(), intValue: v.value}}, nil } +func (v *vertexConst) Acquire(ctx context.Context) (ReleaseFunc, error) { + return func() {}, nil +} + // vtxSum returns a vertex that ourputs sum of its inputs plus a constant func vtxSum(v int, opt vtxOpt) *vertexSum { if opt.cacheKeySeed == "" { @@ -3591,6 +3599,10 @@ func (v *vertexSum) Exec(ctx context.Context, g session.Group, inputs []Result) return []Result{&dummyResult{id: identity.NewID(), intValue: s}}, nil } +func (v *vertexSum) Acquire(ctx context.Context) (ReleaseFunc, error) { + return func() {}, nil +} + func vtxSubBuild(g Edge, opt vtxOpt) *vertexSubBuild { if opt.cacheKeySeed == "" { opt.cacheKeySeed = fmt.Sprintf("sum-%s", identity.NewID()) @@ -3622,6 +3634,10 @@ func (v *vertexSubBuild) Exec(ctx context.Context, g session.Group, inputs []Res return []Result{res}, nil } +func (v *vertexSubBuild) Acquire(ctx context.Context) (ReleaseFunc, error) { + return func() {}, nil +} + //nolint:unused func printGraph(e Edge, pfx string) { name := e.Vertex.Name() diff --git a/solver/types.go b/solver/types.go index f9de6a93b260..a87760692ebc 100644 --- a/solver/types.go +++ b/solver/types.go @@ -135,6 +135,8 @@ type CacheLink struct { Selector digest.Digest `json:",omitempty"` } +type ReleaseFunc func() + // Op defines how the solver can evaluate the properties of a vertex operation. // An op is executed in the worker, and is retrieved from the vertex by the // value of `vertex.Sys()`. The solver is configured with a resolve function to @@ -146,6 +148,9 @@ type Op interface { // Exec runs an operation given results from previous operations. Exec(ctx context.Context, g session.Group, inputs []Result) (outputs []Result, err error) + + // Acquire acquires the necessary resources to execute the `Op`. + Acquire(ctx context.Context) (release ReleaseFunc, err error) } type ResultBasedCacheFunc func(context.Context, Result, session.Group) (digest.Digest, error) diff --git a/worker/base/worker.go b/worker/base/worker.go index 73465db3e92f..e078d105e1e2 100644 --- a/worker/base/worker.go +++ b/worker/base/worker.go @@ -50,6 +50,7 @@ import ( "github.com/pkg/errors" "github.com/sirupsen/logrus" bolt "go.etcd.io/bbolt" + "golang.org/x/sync/semaphore" ) const labelCreatedAt = "buildkit/createdat" @@ -74,6 +75,7 @@ type WorkerOpt struct { IdentityMapping *idtools.IdentityMapping LeaseManager leases.Manager GarbageCollect func(context.Context) (gc.Stats, error) + ParallelismSem *semaphore.Weighted } // Worker is a local worker instance with dedicated snapshotter, cache, and so on. @@ -261,11 +263,11 @@ func (w *Worker) ResolveOp(v solver.Vertex, s frontend.FrontendLLBBridge, sm *se if baseOp, ok := v.Sys().(*pb.Op); ok { switch op := baseOp.Op.(type) { case *pb.Op_Source: - return ops.NewSourceOp(v, op, baseOp.Platform, w.SourceManager, sm, w) + return ops.NewSourceOp(v, op, baseOp.Platform, w.SourceManager, w.ParallelismSem, sm, w) case *pb.Op_Exec: - return ops.NewExecOp(v, op, baseOp.Platform, w.CacheMgr, sm, w.WorkerOpt.MetadataStore, w.WorkerOpt.Executor, w) + return ops.NewExecOp(v, op, baseOp.Platform, w.CacheMgr, w.ParallelismSem, sm, w.WorkerOpt.MetadataStore, w.WorkerOpt.Executor, w) case *pb.Op_File: - return ops.NewFileOp(v, op, w.CacheMgr, w.WorkerOpt.MetadataStore, w) + return ops.NewFileOp(v, op, w.CacheMgr, w.ParallelismSem, w.WorkerOpt.MetadataStore, w) case *pb.Op_Build: return ops.NewBuildOp(v, op, s, w) default: diff --git a/worker/containerd/containerd.go b/worker/containerd/containerd.go index a42894b5a953..5b5ef0795871 100644 --- a/worker/containerd/containerd.go +++ b/worker/containerd/containerd.go @@ -22,19 +22,20 @@ import ( "github.com/moby/buildkit/worker/base" specs "github.com/opencontainers/image-spec/specs-go/v1" "github.com/pkg/errors" + "golang.org/x/sync/semaphore" ) // NewWorkerOpt creates a WorkerOpt. -func NewWorkerOpt(root string, address, snapshotterName, ns string, labels map[string]string, dns *oci.DNSConfig, nopt netproviders.Opt, apparmorProfile string, opts ...containerd.ClientOpt) (base.WorkerOpt, error) { +func NewWorkerOpt(root string, address, snapshotterName, ns string, labels map[string]string, dns *oci.DNSConfig, nopt netproviders.Opt, apparmorProfile string, parallelismSem *semaphore.Weighted, opts ...containerd.ClientOpt) (base.WorkerOpt, error) { opts = append(opts, containerd.WithDefaultNamespace(ns)) client, err := containerd.New(address, opts...) if err != nil { return base.WorkerOpt{}, errors.Wrapf(err, "failed to connect client to %q . make sure containerd is running", address) } - return newContainerd(root, client, snapshotterName, ns, labels, dns, nopt, apparmorProfile) + return newContainerd(root, client, snapshotterName, ns, labels, dns, nopt, apparmorProfile, parallelismSem) } -func newContainerd(root string, client *containerd.Client, snapshotterName, ns string, labels map[string]string, dns *oci.DNSConfig, nopt netproviders.Opt, apparmorProfile string) (base.WorkerOpt, error) { +func newContainerd(root string, client *containerd.Client, snapshotterName, ns string, labels map[string]string, dns *oci.DNSConfig, nopt netproviders.Opt, apparmorProfile string, parallelismSem *semaphore.Weighted) (base.WorkerOpt, error) { if strings.Contains(snapshotterName, "/") { return base.WorkerOpt{}, errors.Errorf("bad snapshotter name: %q", snapshotterName) } @@ -123,6 +124,7 @@ func newContainerd(root string, client *containerd.Client, snapshotterName, ns s Platforms: platforms, LeaseManager: lm, GarbageCollect: gc, + ParallelismSem: parallelismSem, } return opt, nil } diff --git a/worker/containerd/containerd_test.go b/worker/containerd/containerd_test.go index d4c9f5bb695c..f2eb71989e16 100644 --- a/worker/containerd/containerd_test.go +++ b/worker/containerd/containerd_test.go @@ -30,7 +30,7 @@ func newWorkerOpt(t *testing.T, addr string) (base.WorkerOpt, func()) { tmpdir, err := ioutil.TempDir("", "workertest") require.NoError(t, err) cleanup := func() { os.RemoveAll(tmpdir) } - workerOpt, err := NewWorkerOpt(tmpdir, addr, "overlayfs", "buildkit-test", nil, nil, netproviders.Opt{Mode: "host"}, "") + workerOpt, err := NewWorkerOpt(tmpdir, addr, "overlayfs", "buildkit-test", nil, nil, netproviders.Opt{Mode: "host"}, "", nil) require.NoError(t, err) return workerOpt, cleanup } diff --git a/worker/runc/runc.go b/worker/runc/runc.go index 54b8c1255bbd..6d81274c9b81 100644 --- a/worker/runc/runc.go +++ b/worker/runc/runc.go @@ -23,6 +23,7 @@ import ( "github.com/moby/buildkit/worker/base" specs "github.com/opencontainers/image-spec/specs-go/v1" bolt "go.etcd.io/bbolt" + "golang.org/x/sync/semaphore" ) // SnapshotterFactory instantiates a snapshotter @@ -32,7 +33,7 @@ type SnapshotterFactory struct { } // NewWorkerOpt creates a WorkerOpt. -func NewWorkerOpt(root string, snFactory SnapshotterFactory, rootless bool, processMode oci.ProcessMode, labels map[string]string, idmap *idtools.IdentityMapping, nopt netproviders.Opt, dns *oci.DNSConfig, binary, apparmorProfile string) (base.WorkerOpt, error) { +func NewWorkerOpt(root string, snFactory SnapshotterFactory, rootless bool, processMode oci.ProcessMode, labels map[string]string, idmap *idtools.IdentityMapping, nopt netproviders.Opt, dns *oci.DNSConfig, binary, apparmorProfile string, parallelismSem *semaphore.Weighted) (base.WorkerOpt, error) { var opt base.WorkerOpt name := "runc-" + snFactory.Name root = filepath.Join(root, name) @@ -124,6 +125,7 @@ func NewWorkerOpt(root string, snFactory SnapshotterFactory, rootless bool, proc IdentityMapping: idmap, LeaseManager: lm, GarbageCollect: mdb.GarbageCollect, + ParallelismSem: parallelismSem, } return opt, nil } diff --git a/worker/runc/runc_test.go b/worker/runc/runc_test.go index a0321cc7e1d6..3bd249dbd5a0 100644 --- a/worker/runc/runc_test.go +++ b/worker/runc/runc_test.go @@ -40,7 +40,7 @@ func newWorkerOpt(t *testing.T, processMode oci.ProcessMode) (base.WorkerOpt, fu }, } rootless := false - workerOpt, err := NewWorkerOpt(tmpdir, snFactory, rootless, processMode, nil, nil, netproviders.Opt{Mode: "host"}, nil, "", "") + workerOpt, err := NewWorkerOpt(tmpdir, snFactory, rootless, processMode, nil, nil, netproviders.Opt{Mode: "host"}, nil, "", "", nil) require.NoError(t, err) return workerOpt, cleanup