Implemented parallel task runner

Author: cmaglie

Diff for: internal/arduino/builder/internal/runner/runner.go

@@ -0,0 +1,112 @@
+// This file is part of arduino-cli.
+//
+// Copyright 2024 ARDUINO SA (http://www.arduino.cc/)
+//
+// This software is released under the GNU General Public License version 3,
+// which covers the main part of arduino-cli.
+// The terms of this license can be found at:
+// https://www.gnu.org/licenses/gpl-3.0.en.html
+//
+// You can be released from the requirements of the above licenses by purchasing
+// a commercial license. Buying such a license is mandatory if you want to
+// modify or otherwise use the software for commercial activities involving the
+// Arduino software without disclosing the source code of your own applications.
+// To purchase a commercial license, send an email to [email protected].
+
+package runner
+
+import (
+	"context"
+	"runtime"
+	"sync"
+)
+
+// Runner is a helper to run commands in a queue, the commands are immediately exectuded
+// in a goroutine as they are enqueued. The runner can be stopped by calling Cancel.
+type Runner struct {
+	lock      sync.Mutex
+	queue     chan<- *enqueuedCommand
+	results   map[string]<-chan *Result
+	ctx       context.Context
+	ctxCancel func()
+	wg        sync.WaitGroup
+}
+
+type enqueuedCommand struct {
+	task   *Task
+	accept func(*Result)
+}
+
+func (cmd *enqueuedCommand) String() string {
+	return cmd.task.String()
+}
+
+func New(inCtx context.Context) *Runner {
+	ctx, cancel := context.WithCancel(inCtx)
+	queue := make(chan *enqueuedCommand, 1000)
+	r := &Runner{
+		ctx:       ctx,
+		ctxCancel: cancel,
+		queue:     queue,
+		results:   map[string]<-chan *Result{},
+	}
+
+	// Spawn workers
+	for i := 0; i < runtime.NumCPU(); i++ {
+		r.wg.Add(1)
+		go func() {
+			worker(ctx, queue)
+			r.wg.Done()
+		}()
+	}
+
+	return r
+}
+
+func worker(ctx context.Context, queue <-chan *enqueuedCommand) {
+	done := ctx.Done()
+	for {
+		select {
+		case <-done:
+			return
+		default:
+		}
+
+		select {
+		case <-done:
+			return
+		case cmd := <-queue:
+			result := cmd.task.Run(ctx)
+			cmd.accept(result)
+		}
+	}
+}
+
+func (r *Runner) Enqueue(task *Task) {
+	r.lock.Lock()
+	defer r.lock.Unlock()
+
+	result := make(chan *Result, 1)
+	r.results[task.String()] = result
+	r.queue <- &enqueuedCommand{
+		task: task,
+		accept: func(res *Result) {
+			result <- res
+		},
+	}
+}
+
+func (r *Runner) Results(task *Task) *Result {
+	r.lock.Lock()
+	result, ok := r.results[task.String()]
+	r.lock.Unlock()
+	if !ok {
+		return nil
+	}
+	return <-result
+}
+
+func (r *Runner) Cancel() {
+	r.ctxCancel()
+	r.wg.Wait()
+}

Diff for: internal/arduino/builder/internal/runner/runner_test.go

@@ -0,0 +1,38 @@
+package runner_test
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/arduino/arduino-cli/internal/arduino/builder/internal/runner"
+	"github.com/stretchr/testify/require"
+)
+
+func TestRunMultipleTask(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
+	defer cancel()
+	r := runner.New(ctx)
+	r.Enqueue(runner.NewTask("bash", "-c", "sleep 1 ; echo -n 0"))
+	r.Enqueue(runner.NewTask("bash", "-c", "sleep 2 ; echo -n 1"))
+	r.Enqueue(runner.NewTask("bash", "-c", "sleep 3 ; echo -n 2"))
+	r.Enqueue(runner.NewTask("bash", "-c", "sleep 4 ; echo -n 3"))
+	r.Enqueue(runner.NewTask("bash", "-c", "sleep 5 ; echo -n 4"))
+	r.Enqueue(runner.NewTask("bash", "-c", "sleep 6 ; echo -n 5"))
+	r.Enqueue(runner.NewTask("bash", "-c", "sleep 7 ; echo -n 6"))
+	r.Enqueue(runner.NewTask("bash", "-c", "sleep 8 ; echo -n 7"))
+	r.Enqueue(runner.NewTask("bash", "-c", "sleep 9 ; echo -n 8"))
+	r.Enqueue(runner.NewTask("bash", "-c", "sleep 10 ; echo -n 9"))
+	r.Enqueue(runner.NewTask("bash", "-c", "sleep 11 ; echo -n 10"))
+	r.Enqueue(runner.NewTask("bash", "-c", "sleep 12 ; echo -n 11"))
+	r.Enqueue(runner.NewTask("bash", "-c", "sleep 13 ; echo -n 12"))
+	r.Enqueue(runner.NewTask("bash", "-c", "sleep 14 ; echo -n 13"))
+	r.Enqueue(runner.NewTask("bash", "-c", "sleep 15 ; echo -n 14"))
+	r.Enqueue(runner.NewTask("bash", "-c", "sleep 16 ; echo -n 15"))
+	require.Nil(t, r.Results(runner.NewTask("bash", "-c", "echo -n 5")))
+	fmt.Println(string(r.Results(runner.NewTask("bash", "-c", "sleep 3 ; echo -n 2")).Stdout))
+	fmt.Println("Cancelling")
+	r.Cancel()
+	fmt.Println("Runner completed")
+}

Diff for: internal/arduino/builder/internal/runner/task.go

@@ -15,9 +15,46 @@
 
 package runner
 
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"github.com/arduino/go-paths-helper"
+)
+
+// Task is a command to be executed
+type Task struct {
+	Args []string `json:"args"`
+}
+
+// NewTask creates a new Task
+func NewTask(args ...string) *Task {
+	return &Task{Args: args}
+}
+
+func (t *Task) String() string {
+	return strings.Join(t.Args, " ")
+}
+
 // Result contains the output of a command execution
 type Result struct {
 	Args   []string
 	Stdout []byte
 	Stderr []byte
+	Error  error
+}
+
+// Run executes the command and returns the result
+func (t *Task) Run(ctx context.Context) *Result {
+	proc, err := paths.NewProcess(nil, t.Args...)
+	if err != nil {
+		return &Result{Args: t.Args, Error: err}
+	}
+	stdout, stderr, err := proc.RunAndCaptureOutput(ctx)
+
+	// Append arguments to stdout
+	stdout = append([]byte(fmt.Sprintln(t)), stdout...)
+
+	return &Result{Args: proc.GetArgs(), Stdout: stdout, Stderr: stderr, Error: err}
 }