Compute signatures of dependent files in parallel on incremental rebuilds

tsc/internal/compiler/checkerpool.go

@@ -367,7 +367,7 @@ func (p *checkerPool) getCheckerNonExclusive() (*checker.Checker, func()) {
 func (p *checkerPool) createCheckers() {
 	p.createCheckersOnce.Do(func() {
 		checkerCount := len(p.checkers)
-		wg := core.NewWorkGroup(p.program.SingleThreaded())
+		wg := core.NewWorkGroup(core.WorkGroupConcurrency(p.program.SingleThreaded()))
 		for i := range checkerCount {
 			wg.Queue(func() {
 				var tracer *checker.Tracer
@@ -450,7 +450,7 @@ func (p *checkerPool) getImportAdjacency() [][]int {
 // making it safe to call `forEachCheckerParallel` from many threads simultaneously.
 func (p *checkerPool) forEachCheckerParallel(cb func(idx int, c *checker.Checker)) {
 	p.createCheckers()
-	wg := core.NewWorkGroup(p.program.SingleThreaded())
+	wg := core.NewWorkGroup(core.WorkGroupConcurrency(p.program.SingleThreaded()))
 	for idx, checker := range p.checkers {
 		wg.Queue(func() {
 			p.locks[idx].Lock()
@@ -477,7 +477,7 @@ func (p *checkerPool) forEachCheckerGroupDo(ctx context.Context, files []*ast.So
 	p.createCheckers()
 
 	checkerCount := len(p.checkers)
-	wg := core.NewWorkGroup(singleThreaded)
+	wg := core.NewWorkGroup(core.WorkGroupConcurrency(singleThreaded))
 	for checkerIdx := range checkerCount {
 		wg.Queue(func() {
 			p.locks[checkerIdx].Lock()

tsc/internal/compiler/fileloader.go

@@ -182,7 +182,7 @@ func processAllProgramFiles(
 		defaultLibraryPath: opts.Host.DefaultLibraryPath(),
 		caseSensitivity:    opts.Host.FS().CaseSensitivity(),
 		filesParser: &filesParser{
-			wg:       core.NewWorkGroup(singleThreaded),
+			wg:       core.NewWorkGroup(core.WorkGroupConcurrency(singleThreaded)),
 			maxDepth: maxNodeModuleJsDepth,
 		},
 		rootTasks:           make([]*parseTask, 0, len(rootFiles)+len(compilerOptions.Lib)),
@@ -352,7 +352,7 @@ func (p *fileLoader) addProjectReferenceTasks(singleThreaded bool) {
 
 	parser := &projectReferenceParser{
 		loader: p,
-		wg:     core.NewWorkGroup(singleThreaded),
+		wg:     core.NewWorkGroup(core.WorkGroupConcurrency(singleThreaded)),
 	}
 	rootTasks := createProjectReferenceParseTasks(projectReferences)
 	parser.parse(rootTasks)

tsc/internal/compiler/program.go

@@ -593,7 +593,7 @@ func (p *Program) SingleThreaded() bool {
 }
 
 func (p *Program) BindSourceFiles() {
-	wg := core.NewWorkGroup(p.SingleThreaded())
+	wg := core.NewWorkGroup(core.WorkGroupConcurrency(p.SingleThreaded()))
 	for _, file := range p.files {
 		if !file.IsBound() {
 			wg.Queue(func() {
@@ -701,7 +701,7 @@ func (p *Program) collectDiagnostics(ctx context.Context, sourceFile *ast.Source
 
 func (p *Program) collectDiagnosticsFromFiles(ctx context.Context, sourceFiles []*ast.SourceFile, concurrent bool, collect func(context.Context, *ast.SourceFile) []*ast.Diagnostic) [][]*ast.Diagnostic {
 	diagnostics := make([][]*ast.Diagnostic, len(sourceFiles))
-	wg := core.NewWorkGroup(!concurrent || p.SingleThreaded())
+	wg := core.NewWorkGroup(core.WorkGroupConcurrency(!concurrent || p.SingleThreaded()))
 	for i, file := range sourceFiles {
 		wg.Queue(func() {
 			diagnostics[i] = collect(ctx, file)
@@ -748,7 +748,7 @@ func (p *Program) collectCheckerDiagnosticsFromFiles(ctx context.Context, source
 			diagnostics[fileIndex] = collect(ctx, c, file)
 		})
 	} else {
-		wg := core.NewWorkGroup(p.SingleThreaded())
+		wg := core.NewWorkGroup(core.WorkGroupConcurrency(p.SingleThreaded()))
 		for i, file := range sourceFiles {
 			if p.SkipTypeChecking(file, false) {
 				continue
@@ -1910,7 +1910,7 @@ func (p *Program) Emit(ctx context.Context, options EmitOptions) *EmitResult {
 			return printer.NewTextWriter(newLine, 0)
 		},
 	}
-	wg := core.NewWorkGroup(p.SingleThreaded())
+	wg := core.NewWorkGroup(core.WorkGroupConcurrency(p.SingleThreaded()))
 	var emitters []*emitter
 	forceDtsEmit := options.EmitOnly == EmitOnlyBuilderSignature || options.ForceEmit && options.EmitOnly == EmitOnlyDts
 	forceJsEmit := options.ForceEmit && options.EmitOnly == EmitOnlyJs

tsc/internal/core/workgroup.go

@@ -17,16 +17,28 @@ type WorkGroup interface {
 	RunAndWait()
 }
 
-func NewWorkGroup(singleThreaded bool) WorkGroup {
-	if singleThreaded {
+// A concurrency of 1 runs the functions on the goroutine that calls RunAndWait; 0 means no limit.
+func NewWorkGroup(concurrency int) WorkGroup {
+	switch concurrency {
+	case 0:
+		return &parallelWorkGroup{semaphore: UnlimitedSemaphore{}}
+	case 1:
 		return &singleThreadedWorkGroup{}
 	}
-	return &parallelWorkGroup{}
+	return &parallelWorkGroup{semaphore: NewLimitedSemaphore(concurrency)}
+}
+
+func WorkGroupConcurrency(singleThreaded bool) int {
+	if singleThreaded {
+		return 1
+	}
+	return 0
 }
 
 type parallelWorkGroup struct {
-	done atomic.Bool
-	wg   sync.WaitGroup
+	done      atomic.Bool
+	wg        sync.WaitGroup
+	semaphore Semaphore
 }
 
 var _ WorkGroup = (*parallelWorkGroup)(nil)
@@ -37,6 +49,8 @@ func (w *parallelWorkGroup) Queue(fn func()) {
 	}
 
 	w.wg.Go(func() {
+		// Running functions call Queue, so waiting for a slot there would deadlock at the limit.
+		defer w.semaphore.Acquire()()
 		fn()
 	})
 }

tsc/internal/core/workgroup_test.go

@@ -0,0 +1,52 @@
+package core_test
+
+import (
+	"runtime"
+	"sync"
+	"sync/atomic"
+	"testing"
+
+	"github.com/microsoft/TypeScript/tsc/internal/core"
+	"gotest.tools/v3/assert"
+)
+
+func TestWorkGroupConcurrencyLimit(t *testing.T) {
+	t.Parallel()
+	const limit = 3
+	const depth = 2
+	const children = 4
+	wg := core.NewWorkGroup(limit)
+	var running, maxRunning, completed atomic.Int32
+	var allRootsRunning sync.WaitGroup
+	allRootsRunning.Add(limit)
+	var queue func(remaining int)
+	queue = func(remaining int) {
+		wg.Queue(func() {
+			now := running.Add(1)
+			for {
+				seen := maxRunning.Load()
+				if now <= seen || maxRunning.CompareAndSwap(seen, now) {
+					break
+				}
+			}
+			if remaining == depth {
+				allRootsRunning.Done()
+				allRootsRunning.Wait()
+			}
+			if remaining > 0 {
+				for range children {
+					queue(remaining - 1)
+				}
+			}
+			runtime.Gosched()
+			running.Add(-1)
+			completed.Add(1)
+		})
+	}
+	for range limit {
+		queue(depth)
+	}
+	wg.RunAndWait()
+	assert.Equal(t, completed.Load(), int32(limit*(1+children+children*children)))
+	assert.Equal(t, maxRunning.Load(), int32(limit))
+}

tsc/internal/execute/build/orchestrator.go

@@ -260,7 +260,7 @@ func (o *Orchestrator) GenerateGraphReusingOldTasks() {
 func (o *Orchestrator) GenerateGraph(oldTasks *collections.SyncMap[tspath.PathKey, *BuildTask]) {
 	projects := o.opts.Command.ResolvedProjectPaths()
 	// Parse all config files in parallel
-	wg := core.NewWorkGroup(o.opts.Command.CompilerOptions.SingleThreaded.IsTrue())
+	wg := core.NewWorkGroup(core.WorkGroupConcurrency(o.opts.Command.CompilerOptions.SingleThreaded.IsTrue()))
 	o.createBuildTasks(oldTasks, projects, wg)
 	wg.RunAndWait()
 
@@ -940,7 +940,7 @@ func (o *Orchestrator) rangeTasks(order []*BuildTask, f func(path tspath.PathKey
 	if numRoutines == 1 {
 		runTask()
 	} else {
-		wg := core.NewWorkGroup(false)
+		wg := core.NewWorkGroup(0)
 		for range numRoutines {
 			wg.Queue(runTask)
 		}

tsc/internal/execute/incremental/affectedfileshandler.go

@@ -2,7 +2,7 @@ package incremental
 
 import (
 	"context"
-	"maps"
+	"runtime"
 	"slices"
 	"sync"
 	"sync/atomic"
@@ -88,12 +88,11 @@ func (h *affectedFilesHandler) updateShapeSignature(file *ast.SourceFile, useFil
 	update := &updatedSignature{}
 	update.mu.Lock()
 	defer update.mu.Unlock()
-	// If we have cached the result for this file, that means hence forth we should assume file shape is uptodate
 	if existing, ok := h.updatedSignatures.LoadOrStore(file.PathKey(), update); ok {
 		// Ensure calculations for existing ones are complete before using the value
 		existing.mu.Lock()
 		defer existing.mu.Unlock()
-		return false
+		return h.isChangedSignature(file.PathKey())
 	}
 
 	info, _ := h.program.snapshot.fileInfos.Load(file.PathKey())
@@ -111,42 +110,45 @@ func (h *affectedFilesHandler) updateShapeSignature(file *ast.SourceFile, useFil
 	return update.signature != prevSignature
 }
 
-func (h *affectedFilesHandler) getFilesAffectedBy(path tspath.PathKey) []*ast.SourceFile {
+func (h *affectedFilesHandler) collectFilesAffectedBy(path tspath.PathKey, wg core.WorkGroup, result *collections.SyncSet[*ast.SourceFile]) {
 	file := h.program.program.GetSourceFileByPath(path)
 	if file == nil {
-		return nil
+		return
 	}
 
 	if !h.updateShapeSignature(file, false) {
-		return []*ast.SourceFile{file}
+		result.Add(file)
+		return
 	}
 
 	if info, _ := h.program.snapshot.fileInfos.Load(file.PathKey()); info.affectsGlobalScope {
 		h.hasAllFilesExcludingDefaultLibraryFile.Store(true)
-		return h.program.snapshot.getAllFilesExcludingDefaultLibraryFile(h.program.program, file)
+		for _, affectedFile := range h.program.snapshot.getAllFilesExcludingDefaultLibraryFile(h.program.program, file) {
+			result.Add(affectedFile)
+		}
+		return
 	}
 
+	result.Add(file)
 	if h.program.snapshot.options.IsolatedModules.IsTrue() {
-		return []*ast.SourceFile{file}
+		return
 	}
 
-	// Now we need to if each file in the referencedBy list has a shape change as well.
-	// Because if so, its own referencedBy files need to be saved as well to make the
-	// emitting result consistent with files on disk.
-	seenFileNamesMap := h.forEachFileReferencedBy(
-		file,
-		func(currentFile *ast.SourceFile, currentPath tspath.PathKey) (queueForFile bool, fastReturn bool) {
-			// If the current file is not nil and has a shape change, we need to queue it for processing
-			if currentFile != nil && h.updateShapeSignature(currentFile, false) {
-				return true, false
+	h.collectReferencingFiles(file, wg, result)
+}
+
+func (h *affectedFilesHandler) collectReferencingFiles(file *ast.SourceFile, wg core.WorkGroup, result *collections.SyncSet[*ast.SourceFile]) {
+	for path := range h.program.snapshot.referencedMap.getReferencedBy(file.PathKey()) {
+		currentFile := h.program.program.GetSourceFileByPath(path)
+		if currentFile == nil || !result.AddIfAbsent(currentFile) {
+			continue
+		}
+		wg.Queue(func() {
+			if h.updateShapeSignature(currentFile, false) {
+				h.collectReferencingFiles(currentFile, wg, result)
 			}
-			return false, false
-		},
-	)
-	// Return array of values that needs emit
-	return core.Filter(slices.Collect(maps.Values(seenFileNamesMap)), func(file *ast.SourceFile) bool {
-		return file != nil
-	})
+		})
+	}
 }
 
 func (h *affectedFilesHandler) forEachFileReferencedBy(file *ast.SourceFile, fn func(currentFile *ast.SourceFile, currentPath tspath.PathKey) (queueForFile bool, fastReturn bool)) map[tspath.PathKey]*ast.SourceFile {
@@ -362,13 +364,13 @@ func collectAllAffectedFiles(ctx context.Context, program *Program) {
 	}
 
 	handler := affectedFilesHandler{ctx: ctx, program: program}
-	wg := core.NewWorkGroup(handler.program.program.SingleThreaded())
+	// Starting every signature at once raises peak memory without finishing sooner.
+	concurrency := core.IfElse(program.program.SingleThreaded(), 1, runtime.GOMAXPROCS(0))
+	wg := core.NewWorkGroup(concurrency)
 	var result collections.SyncSet[*ast.SourceFile]
 	program.snapshot.changedFilesSet.Range(func(file tspath.PathKey) bool {
 		wg.Queue(func() {
-			for _, affectedFile := range handler.getFilesAffectedBy(file) {
-				result.Add(affectedFile)
-			}
+			handler.collectFilesAffectedBy(file, wg, &result)
 		})
 		return true
 	})
@@ -380,7 +382,7 @@ func collectAllAffectedFiles(ctx context.Context, program *Program) {
 
 	// For all the affected files, get all the files that would need to change their dts or js files,
 	// update their diagnostics
-	wg = core.NewWorkGroup(program.program.SingleThreaded())
+	wg = core.NewWorkGroup(concurrency)
 	emitKind := GetFileEmitKind(program.snapshot.options)
 	result.Range(func(file *ast.SourceFile) bool {
 		// remove the cached semantic diagnostics and handle dts emit and js emit if needed