diff --git a/pkg/analyzer/analyzer.go b/pkg/analyzer/analyzer.go index 83ac9f8a359..a1a2d9866f4 100644 --- a/pkg/analyzer/analyzer.go +++ b/pkg/analyzer/analyzer.go @@ -5,6 +5,7 @@ import ( "os" "path/filepath" "regexp" + "runtime" "sort" "strings" "sync" @@ -37,6 +38,8 @@ const ( crossplane = "crossplane" knative = "knative" sizeMb = 1048576 + + maxAnalyzerWorkers = 128 ) // move the openApi regex to public to be used on file.go @@ -172,6 +175,7 @@ type Analyzer struct { ExcludeGitIgnore bool MaxFileSize int FallbackMinifiedFileLOC int + MaxAnalyzerWorkers int } // types is a map that contains the regex by type @@ -373,19 +377,34 @@ func Analyze(a *Analyzer) (model.AnalyzedPaths, error) { a.Types, a.ExcludeTypes = typeLower(a.Types, a.ExcludeTypes) - // Start the workers - for _, file := range files { - wg.Add(1) - // analyze the files concurrently - a := &analyzerInfo{ - typesFlag: a.Types, - excludeTypesFlag: a.ExcludeTypes, - filePath: file, - fallbackMinifiedFileLOC: a.FallbackMinifiedFileLOC, - } - go a.worker(results, unwanted, locCount, fileInfo, &wg) + // Start a bounded worker pool. Large repositories can contain tens of + // thousands of candidate files, so one goroutine per file can exhaust + // runtime threads while workers are blocked on file I/O. + filesToAnalyze := make(chan string) + workerCount := analyzerWorkerCount(len(files), a.MaxAnalyzerWorkers) + wg.Add(workerCount) + for range workerCount { + go func() { + defer wg.Done() + for file := range filesToAnalyze { + fileAnalyzer := &analyzerInfo{ + typesFlag: a.Types, + excludeTypesFlag: a.ExcludeTypes, + filePath: file, + fallbackMinifiedFileLOC: a.FallbackMinifiedFileLOC, + } + fileAnalyzer.worker(results, unwanted, locCount, fileInfo) + } + }() } + go func() { + for _, file := range files { + filesToAnalyze <- file + } + close(filesToAnalyze) + }() + go func() { // close channel results when the worker has finished writing into it defer func() { @@ -419,14 +438,12 @@ func (a *analyzerInfo) worker( //nolint: gocyclo unwanted chan<- string, locCount chan<- int, fileInfo chan<- fileTypeInfo, - wg *sync.WaitGroup, ) { defer func() { if err := recover(); err != nil { log.Warn().Msgf("Recovered from analyzing panic for file %s with error: %#v", a.filePath, err.(error).Error()) unwanted <- a.filePath } - wg.Done() }() ext, errExt := utils.GetExtension(a.filePath) @@ -521,6 +538,28 @@ func needsOverride(check bool, returnType, key, ext string) bool { return false } +func analyzerWorkerCount(fileCount int, maxWorkers int) int { + if fileCount < 1 { + return 0 + } + + // Use default constant if maxWorkers is not set + if maxWorkers <= 0 { + maxWorkers = maxAnalyzerWorkers + } + workers := runtime.GOMAXPROCS(0) * 2 + if workers < 1 { + workers = 1 + } + if workers > maxWorkers { + workers = maxWorkers + } + if fileCount < workers { + return fileCount + } + return workers +} + // checkContent will determine the file type by content when worker was unable to // determine by ext, if no type was determined checkContent adds it to unwanted channel func (a *analyzerInfo) checkContent( diff --git a/pkg/analyzer/analyzer_test.go b/pkg/analyzer/analyzer_test.go index ea7adcf7f69..f605a3c95b8 100644 --- a/pkg/analyzer/analyzer_test.go +++ b/pkg/analyzer/analyzer_test.go @@ -1,9 +1,14 @@ package analyzer import ( + "fmt" + "os" "path/filepath" + "runtime" "sort" + "sync/atomic" "testing" + "time" "github.com/stretchr/testify/require" ) @@ -734,3 +739,129 @@ type platformFileStats struct { dirCount int totalLOC int } + +func TestAnalyzerWorkerCount(t *testing.T) { + oldMaxProcs := runtime.GOMAXPROCS(1) + defer runtime.GOMAXPROCS(oldMaxProcs) + + require.Equal(t, 0, analyzerWorkerCount(0, maxAnalyzerWorkers)) + require.Equal(t, 2, analyzerWorkerCount(10, maxAnalyzerWorkers)) + + runtime.GOMAXPROCS(maxAnalyzerWorkers) + require.Equal(t, 5, analyzerWorkerCount(5, maxAnalyzerWorkers)) + require.Equal(t, maxAnalyzerWorkers, analyzerWorkerCount(maxAnalyzerWorkers+1, 0)) + + // A caller-supplied maxWorkers overrides the built-in default, both below and above it. + require.Equal(t, 3, analyzerWorkerCount(maxAnalyzerWorkers+1, 3)) + require.Equal(t, 200, analyzerWorkerCount(300, 200)) +} + +// TestAnalyzer_BoundedWorkerConcurrency guards against Analyze regressing back to +// spawning one goroutine per file. A bounded pool stays near analyzerWorkerCount +// even while a large repository is scanned. +func TestAnalyzer_BoundedWorkerConcurrency(t *testing.T) { + oldMaxProcs := runtime.GOMAXPROCS(2) + defer runtime.GOMAXPROCS(oldMaxProcs) + + dir := t.TempDir() + const fileCount = 3000 + for i := 0; i < fileCount; i++ { + content := []byte(fmt.Sprintf("resource \"null_resource\" \"r%d\" {}\n", i)) + require.NoError(t, os.WriteFile(filepath.Join(dir, fmt.Sprintf("file_%d.tf", i)), content, 0o600)) + } + + expectedWorkers := analyzerWorkerCount(fileCount, 0) + const goroutineSlack = 20 + baseline := runtime.NumGoroutine() + var peak int64 + stop := make(chan struct{}) + monitorDone := make(chan struct{}) + go func() { + defer close(monitorDone) + for { + select { + case <-stop: + return + default: + if n := int64(runtime.NumGoroutine()); n > atomic.LoadInt64(&peak) { + atomic.StoreInt64(&peak, n) + } + time.Sleep(time.Microsecond) + } + } + }() + + analyzer := &Analyzer{ + Paths: []string{dir}, + Types: []string{""}, + ExcludeTypes: []string{""}, + Exc: []string{""}, + MaxFileSize: -1, + } + _, err := Analyze(analyzer) + require.NoError(t, err) + close(stop) + <-monitorDone + + extraGoroutines := int(atomic.LoadInt64(&peak)) - baseline + require.LessOrEqualf(t, extraGoroutines, expectedWorkers+goroutineSlack, + "Analyze spawned %d extra goroutines while processing %d files; expected at most about %d bounded workers", + extraGoroutines, fileCount, expectedWorkers) +} + +// TestAnalyzer_MaxAnalyzerWorkersOverride verifies that a caller-supplied +// Analyzer.MaxAnalyzerWorkers value (used by library embedders) is honored by Analyze +// capping concurrency below the built-in default rather than being ignored. +func TestAnalyzer_MaxAnalyzerWorkersOverride(t *testing.T) { + oldMaxProcs := runtime.GOMAXPROCS(8) + defer runtime.GOMAXPROCS(oldMaxProcs) + + dir := t.TempDir() + const fileCount = 500 + for i := 0; i < fileCount; i++ { + content := []byte(fmt.Sprintf("resource \"null_resource\" \"r%d\" {}\n", i)) + require.NoError(t, os.WriteFile(filepath.Join(dir, fmt.Sprintf("file_%d.tf", i)), content, 0o600)) + } + + const customCap = 2 + expectedWorkers := analyzerWorkerCount(fileCount, customCap) + require.Equal(t, customCap, expectedWorkers, "test setup should exercise the custom cap, not the default") + + const goroutineSlack = 20 + baseline := runtime.NumGoroutine() + var peak int64 + stop := make(chan struct{}) + monitorDone := make(chan struct{}) + go func() { + defer close(monitorDone) + for { + select { + case <-stop: + return + default: + if n := int64(runtime.NumGoroutine()); n > atomic.LoadInt64(&peak) { + atomic.StoreInt64(&peak, n) + } + time.Sleep(time.Microsecond) + } + } + }() + + analyzer := &Analyzer{ + Paths: []string{dir}, + Types: []string{""}, + ExcludeTypes: []string{""}, + Exc: []string{""}, + MaxFileSize: -1, + MaxAnalyzerWorkers: customCap, + } + _, err := Analyze(analyzer) + require.NoError(t, err) + close(stop) + <-monitorDone + + extraGoroutines := int(atomic.LoadInt64(&peak)) - baseline + require.LessOrEqualf(t, extraGoroutines, expectedWorkers+goroutineSlack, + "Analyze spawned %d extra goroutines while processing %d files with MaxAnalyzerWorkers=%d; expected at most about %d bounded workers", + extraGoroutines, fileCount, customCap, expectedWorkers) +}