Merge pull request #10133 from ndeloof/build_concurrency

limit build concurrency according to --parallel

cmd/compose/compose.go

@@ -22,6 +22,7 @@ import (
 	"os"
 	"os/signal"
 	"path/filepath"
+	"strconv"
 	"strings"
 	"syscall"
 
@@ -324,6 +325,13 @@ func RootCommand(streams api.Streams, backend api.Service) *cobra.Command { //no
 					return err
 				}
 			}
+			if v, ok := os.LookupEnv("COMPOSE_PARALLEL_LIMIT"); ok && !cmd.Flags().Changed("parallel") {
+				i, err := strconv.Atoi(v)
+				if err != nil {
+					return fmt.Errorf("COMPOSE_PARALLEL_LIMIT must be an integer (found: %q)", v)
+				}
+				parallel = i
+			}
 			if parallel > 0 {
 				backend.MaxConcurrency(parallel)
 			}

pkg/compose/build.go

@@ -51,14 +51,16 @@ func (s *composeService) build(ctx context.Context, project *types.Project, opti
 	opts := map[string]build.Options{}
 	args := flatten(options.Args.Resolve(envResolver(project.Environment)))
 
-	services, err := project.GetServices(options.Services...)
+	return InDependencyOrder(ctx, project, func(ctx context.Context, name string) error {
+		if len(options.Services) > 0 && !utils.Contains(options.Services, name) {
+			return nil
+		}
+		service, err := project.GetService(name)
 		if err != nil {
 			return err
 		}
-
-	for _, service := range services {
 		if service.Build == nil {
-			continue
+			return nil
 		}
 		imageName := api.GetImageNameOrDefault(service, project.Name)
 		buildOptions, err := s.toBuildOptions(project, service, imageName, options.SSHs)
@@ -91,10 +93,11 @@ func (s *composeService) build(ctx context.Context, project *types.Project, opti
 			}}
 		}
 		opts[imageName] = buildOptions
-	}
-
 		_, err = s.doBuild(ctx, project, opts, options.Progress)
 		return err
+	}, func(traversal *graphTraversal) {
+		traversal.maxConcurrency = s.maxConcurrency
+	})
 }
 
 func (s *composeService) ensureImagesExists(ctx context.Context, project *types.Project, quietPull bool) error {

pkg/compose/dependencies.go

@@ -48,6 +48,7 @@ type graphTraversal struct {
 	adjacentServiceStatusToSkip ServiceStatus
 
 	visitorFn      func(context.Context, string) error
+	maxConcurrency int
 }
 
 func upDirectionTraversal(visitorFn func(context.Context, string) error) *graphTraversal {
@@ -79,6 +80,9 @@ func InDependencyOrder(ctx context.Context, project *types.Project, fn func(cont
 		return err
 	}
 	t := upDirectionTraversal(fn)
+	for _, option := range options {
+		option(t)
+	}
 	return t.visit(ctx, graph)
 }
 
@@ -96,6 +100,9 @@ func (t *graphTraversal) visit(ctx context.Context, g *Graph) error {
 	nodes := t.extremityNodesFn(g)
 
 	eg, ctx := errgroup.WithContext(ctx)
+	if t.maxConcurrency > 0 {
+		eg.SetLimit(t.maxConcurrency)
+	}
 	t.run(ctx, g, eg, nodes)
 
 	return eg.Wait()