Add concurrent jobs control and fix flawed logic

commands/compile/compile.go

@@ -115,6 +115,8 @@ func Compile(ctx context.Context, req *rpc.CompileReq, outStream, errStream io.W
 
 	builderCtx.CoreBuildCachePath = paths.TempDir().Join("arduino-core-cache")
 
+	builderCtx.Jobs = int(req.GetJobs())
+
 	builderCtx.USBVidPid = req.GetVidPid()
 	builderCtx.WarningsLevel = req.GetWarnings()
 

legacy/builder/builder_utils/utils.go

@@ -33,6 +33,7 @@ import (
 	"os"
 	"os/exec"
 	"path/filepath"
+	"runtime"
 	"strconv"
 	"strings"
 	"sync"
@@ -169,47 +170,61 @@ func compileFilesWithRecipe(ctx *types.Context, sourcePath *paths.Path, sources
 	if len(sources) == 0 {
 		return objectFiles, nil
 	}
-	objectFilesChan := make(chan *paths.Path)
-	errorsChan := make(chan error)
-	doneChan := make(chan struct{})
+	var objectFilesMux sync.Mutex
+	var errors []error
+	var errorsMux sync.Mutex
 
 	ctx.Progress.Steps = ctx.Progress.Steps / float64(len(sources))
-	var wg sync.WaitGroup
-	wg.Add(len(sources))
 
-	for _, source := range sources {
-		go func(source *paths.Path) {
-			defer wg.Done()
-			PrintProgressIfProgressEnabledAndMachineLogger(ctx)
-			objectFile, err := compileFileWithRecipe(ctx, sourcePath, source, buildPath, buildProperties, includes, recipe)
-			if err != nil {
-				errorsChan <- err
-			} else {
-				objectFilesChan <- objectFile
-			}
-		}(source)
+	queue := make(chan *paths.Path)
+	job := func(source *paths.Path) {
+		PrintProgressIfProgressEnabledAndMachineLogger(ctx)
+		objectFile, err := compileFileWithRecipe(ctx, sourcePath, source, buildPath, buildProperties, includes, recipe)
+		if err != nil {
+			errorsMux.Lock()
+			errors = append(errors, err)
+			errorsMux.Unlock()
+		} else {
+			objectFilesMux.Lock()
+			objectFiles.Add(objectFile)
+			objectFilesMux.Unlock()
+		}
 	}
 
-	go func() {
-		wg.Wait()
-		doneChan <- struct{}{}
-	}()
-
-	for {
-		select {
-		case objectFile := <-objectFilesChan:
-			objectFiles.Add(objectFile)
-		case err := <-errorsChan:
-			return nil, i18n.WrapError(err)
-		case <-doneChan:
-			close(objectFilesChan)
-			for objectFile := range objectFilesChan {
-				objectFiles.Add(objectFile)
+	// Spawn jobs runners
+	var wg sync.WaitGroup
+	jobs := ctx.Jobs
+	if jobs == 0 {
+		jobs = runtime.NumCPU()
+	}
+	for i := 0; i < jobs; i++ {
+		wg.Add(1)
+		go func() {
+			for source := range queue {
+				job(source)
 			}
-			objectFiles.Sort()
-			return objectFiles, nil
+			wg.Done()
+		}()
+	}
+
+	// Feed jobs until error or done
+	for _, source := range sources {
+		errorsMux.Lock()
+		gotError := len(errors) > 0
+		errorsMux.Unlock()
+		if gotError {
+			break
 		}
+		queue <- source
 	}
+	close(queue)
+	wg.Wait()
+	if len(errors) > 0 {
+		// output the first error
+		return nil, i18n.WrapError(errors[0])
+	}
+	objectFiles.Sort()
+	return objectFiles, nil
 }
 
 func compileFileWithRecipe(ctx *types.Context, sourcePath *paths.Path, source *paths.Path, buildPath *paths.Path, buildProperties *properties.Map, includes []string, recipe string) (*paths.Path, error) {

legacy/builder/types/context.go

@@ -111,6 +111,9 @@ type Context struct {
 	// Experimental: use arduino-preprocessor to create prototypes
 	UseArduinoPreprocessor bool
 
+	// Parallel processes
+	Jobs int
+
 	// Out and Err stream to redirect all Exec commands
 	ExecStdout io.Writer
 	ExecStderr io.Writer

rpc/commands/compile.proto

@@ -37,6 +37,7 @@ message CompileReq {
   bool quiet = 11;             // Suppresses almost every output.
   string vidPid = 12;   // VID/PID specific build properties.
   string exportFile = 13;   // The compiled binary is written to this file
+  int32 jobs = 14;   // The max number of concurrent compiler instances to run (as make -jx)
 }
 
 message CompileResp {