feat: parallelize catalogers per-file and hash contents in parallel (#3636)

Signed-off-by: Keith Zantow <kzantow@gmail.com>

internal/file/digest.go

@@ -1,12 +1,15 @@
 package file
 
 import (
+	"context"
 	"crypto"
 	"fmt"
 	"hash"
 	"io"
 	"strings"
 
+	"github.com/anchore/go-sync"
+	"github.com/anchore/syft/syft/cataloging"
 	"github.com/anchore/syft/syft/file"
 )
 
@@ -21,7 +24,7 @@ func supportedHashAlgorithms() []crypto.Hash {
 	}
 }
 
-func NewDigestsFromFile(closer io.ReadCloser, hashes []crypto.Hash) ([]file.Digest, error) {
+func NewDigestsFromFile(ctx context.Context, closer io.ReadCloser, hashes []crypto.Hash) ([]file.Digest, error) {
 	hashes = NormalizeHashes(hashes)
 	// create a set of hasher objects tied together with a single writer to feed content into
 	hashers := make([]hash.Hash, len(hashes))
@@ -31,7 +34,7 @@ func NewDigestsFromFile(closer io.ReadCloser, hashes []crypto.Hash) ([]file.Dige
 		writers[idx] = hashers[idx]
 	}
 
-	size, err := io.Copy(io.MultiWriter(writers...), closer)
+	size, err := io.Copy(sync.ParallelWriter(ctx, cataloging.ExecutorCPU, writers...), closer)
 	if err != nil {
 		return nil, err
 	}

internal/file/digest_test.go

@@ -1,6 +1,7 @@
 package file
 
 import (
+	"context"
 	"crypto"
 	"os"
 	"testing"
@@ -81,7 +82,7 @@ func TestNewDigestsFromFile(t *testing.T) {
 			fh, err := os.Open(tt.fixture)
 			require.NoError(t, err)
 
-			got, err := NewDigestsFromFile(fh, tt.hashes)
+			got, err := NewDigestsFromFile(context.TODO(), fh, tt.hashes)
 			tt.wantErr(t, err)
 			if err != nil {
 				return

internal/task/executor.go

@@ -5,11 +5,8 @@ import (
 	"fmt"
 	"runtime/debug"
 	"slices"
-	"sync"
 	"time"
 
-	"github.com/hashicorp/go-multierror"
-
 	"github.com/anchore/syft/internal/log"
 	"github.com/anchore/syft/internal/sbomsync"
 	"github.com/anchore/syft/internal/unknown"
@@ -18,64 +15,17 @@ import (
 	"github.com/anchore/syft/syft/sbom"
 )
 
-type Executor struct {
-	numWorkers int
-	tasks      chan Task
-}
-
-func NewTaskExecutor(tasks []Task, numWorkers int) *Executor {
-	p := &Executor{
-		numWorkers: numWorkers,
-		tasks:      make(chan Task, len(tasks)),
+func RunTask(ctx context.Context, tsk Task, resolver file.Resolver, s sbomsync.Builder, prog *monitor.CatalogerTaskProgress) error {
+	err := runTaskSafely(ctx, tsk, resolver, s)
+	unknowns, remainingErrors := unknown.ExtractCoordinateErrors(err)
+	if len(unknowns) > 0 {
+		appendUnknowns(s, tsk.Name(), unknowns)
 	}
-
-	for i := range tasks {
-		p.tasks <- tasks[i]
+	if remainingErrors != nil {
+		prog.SetError(remainingErrors)
 	}
-	close(p.tasks)
-
-	return p
-}
-
-func (p *Executor) Execute(ctx context.Context, resolver file.Resolver, s sbomsync.Builder, prog *monitor.CatalogerTaskProgress) error {
-	var lock sync.Mutex
-	withLock := func(fn func()) {
-		lock.Lock()
-		defer lock.Unlock()
-		fn()
-	}
-	var errs error
-	wg := &sync.WaitGroup{}
-	for i := 0; i < p.numWorkers; i++ {
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
-
-			for {
-				tsk, ok := <-p.tasks
-				if !ok {
-					return
-				}
-
-				err := runTaskSafely(ctx, tsk, resolver, s)
-				unknowns, remainingErrors := unknown.ExtractCoordinateErrors(err)
-				if len(unknowns) > 0 {
-					appendUnknowns(s, tsk.Name(), unknowns)
-				}
-				if remainingErrors != nil {
-					withLock(func() {
-						errs = multierror.Append(errs, fmt.Errorf("failed to run task: %w", remainingErrors))
-						prog.SetError(remainingErrors)
-					})
-				}
-				prog.Increment()
-			}
-		}()
-	}
-
-	wg.Wait()
-
-	return errs
+	prog.Increment()
+	return remainingErrors
 }
 
 func appendUnknowns(builder sbomsync.Builder, taskName string, unknowns []unknown.CoordinateError) {

internal/task/executor_test.go

@@ -16,9 +16,8 @@ func Test_TaskExecutor_PanicHandling(t *testing.T) {
 	tsk := NewTask("panicking-cataloger", func(_ context.Context, _ file.Resolver, _ sbomsync.Builder) error {
 		panic("something bad happened")
 	})
-	ex := NewTaskExecutor([]Task{tsk}, 1)
 
-	err := ex.Execute(context.Background(), nil, nil, &monitor.CatalogerTaskProgress{
+	err := RunTask(context.Background(), tsk, nil, nil, &monitor.CatalogerTaskProgress{
 		Manual: progress.NewManual(-1),
 	})
 

internal/task/file_tasks.go

@@ -120,14 +120,14 @@ func newExecutableCatalogerTaskFactory(tags ...string) factory {
 }
 
 func newExecutableCatalogerTask(selection file.Selection, cfg executable.Config, tags ...string) Task {
-	fn := func(_ context.Context, resolver file.Resolver, builder sbomsync.Builder) error {
+	fn := func(ctx context.Context, resolver file.Resolver, builder sbomsync.Builder) error {
 		if selection == file.NoFilesSelection {
 			return nil
 		}
 
 		accessor := builder.(sbomsync.Accessor)
 
-		result, err := executable.NewCataloger(cfg).Catalog(resolver)
+		result, err := executable.NewCataloger(cfg).CatalogCtx(ctx, resolver)
 
 		accessor.WriteToSBOM(func(sbom *sbom.SBOM) {
 			sbom.Artifacts.Executables = result