refactor(@angular-devkit/build-angular): add initial support for parallel TS/NG compilation

By default Angular compilations will now use a Node.js Worker thread to load and execute the
TypeScript and Angular compilers when using esbuild-based builders (`application`/`browser-esbuild`).
This allows for longer synchronous actions such as semantic and template diagnostics to be calculated
in parallel to the other aspects of the application bundling process. The worker thread also has
a separate memory pool which significantly reduces the need for adjusting the main Node.js CLI process
memory settings with large application code sizes. This can be disabled via the `NG_BUILD_PARALLEL_TS`
environment variable currently to support performance benchmarking. However, this is an unsupported
environment variable option and may be removed in a future version.

Author: clydin

packages/angular_devkit/build_angular/src/tools/esbuild/angular/compilation/angular-compilation.ts

@@ -10,7 +10,7 @@ import type ng from '@angular/compiler-cli';
 import type { PartialMessage } from 'esbuild';
 import ts from 'typescript';
 import { loadEsmModule } from '../../../../utils/load-esm';
-import { profileSync } from '../../profiling';
+import { profileAsync, profileSync } from '../../profiling';
 import type { AngularHostOptions } from '../angular-host';
 import { convertTypeScriptDiagnostic } from '../diagnostics';
 
@@ -26,9 +26,8 @@ export abstract class AngularCompilation {
   static async loadCompilerCli(): Promise<typeof ng> {
     // This uses a wrapped dynamic import to load `@angular/compiler-cli` which is ESM.
     // Once TypeScript provides support for retaining dynamic imports this workaround can be dropped.
-    AngularCompilation.#angularCompilerCliModule ??= await loadEsmModule<typeof ng>(
-      '@angular/compiler-cli',
-    );
+    AngularCompilation.#angularCompilerCliModule ??=
+      await loadEsmModule<typeof ng>('@angular/compiler-cli');
 
     return AngularCompilation.#angularCompilerCliModule;
   }
@@ -63,15 +62,17 @@ export abstract class AngularCompilation {
     referencedFiles: readonly string[];
   }>;
 
-  abstract emitAffectedFiles(): Iterable<EmitFileResult>;
+  abstract emitAffectedFiles(): Iterable<EmitFileResult> | Promise<Iterable<EmitFileResult>>;
 
-  protected abstract collectDiagnostics(): Iterable<ts.Diagnostic>;
+  protected abstract collectDiagnostics():
+    | Iterable<ts.Diagnostic>
+    | Promise<Iterable<ts.Diagnostic>>;
 
   async diagnoseFiles(): Promise<{ errors?: PartialMessage[]; warnings?: PartialMessage[] }> {
     const result: { errors?: PartialMessage[]; warnings?: PartialMessage[] } = {};
 
-    profileSync('NG_DIAGNOSTICS_TOTAL', () => {
-      for (const diagnostic of this.collectDiagnostics()) {
+    await profileAsync('NG_DIAGNOSTICS_TOTAL', async () => {
+      for (const diagnostic of await this.collectDiagnostics()) {
         const message = convertTypeScriptDiagnostic(diagnostic);
         if (diagnostic.category === ts.DiagnosticCategory.Error) {
           (result.errors ??= []).push(message);
@@ -83,4 +84,8 @@ export abstract class AngularCompilation {
 
     return result;
   }
+
+  update?(files: Set<string>): Promise<void>;
+
+  close?(): Promise<void>;
 }

packages/angular_devkit/build_angular/src/tools/esbuild/angular/compilation/factory.ts

@@ -0,0 +1,35 @@
+/**
+ * @license
+ * Copyright Google LLC All Rights Reserved.
+ *
+ * Use of this source code is governed by an MIT-style license that can be
+ * found in the LICENSE file at https://angular.io/license
+ */
+
+import { useParallelTs } from '../../../../utils/environment-options';
+import type { AngularCompilation } from './angular-compilation';
+
+/**
+ * Creates an Angular compilation object that can be used to perform Angular application
+ * compilation either for AOT or JIT mode. By default a parallel compilation is created
+ * that uses a Node.js worker thread.
+ * @param jit True, for Angular JIT compilation; False, for Angular AOT compilation.
+ * @returns An instance of an Angular compilation object.
+ */
+export async function createAngularCompilation(jit: boolean): Promise<AngularCompilation> {
+  if (useParallelTs) {
+    const { ParallelCompilation } = await import('./parallel-compilation');
+
+    return new ParallelCompilation(jit);
+  }
+
+  if (jit) {
+    const { JitCompilation } = await import('./jit-compilation');
+
+    return new JitCompilation();
+  } else {
+    const { AotCompilation } = await import('./aot-compilation');
+
+    return new AotCompilation();
+  }
+}

packages/angular_devkit/build_angular/src/tools/esbuild/angular/compilation/index.ts

@@ -7,6 +7,5 @@
  */
 
 export { AngularCompilation } from './angular-compilation';
-export { AotCompilation } from './aot-compilation';
-export { JitCompilation } from './jit-compilation';
+export { createAngularCompilation } from './factory';
 export { NoopCompilation } from './noop-compilation';

packages/angular_devkit/build_angular/src/tools/esbuild/angular/compilation/parallel-compilation.ts

@@ -0,0 +1,140 @@
+/**
+ * @license
+ * Copyright Google LLC All Rights Reserved.
+ *
+ * Use of this source code is governed by an MIT-style license that can be
+ * found in the LICENSE file at https://angular.io/license
+ */
+
+import type { CompilerOptions } from '@angular/compiler-cli';
+import type { PartialMessage } from 'esbuild';
+import { createRequire } from 'node:module';
+import { MessageChannel } from 'node:worker_threads';
+import Piscina from 'piscina';
+import type { SourceFile } from 'typescript';
+import type { AngularHostOptions } from '../angular-host';
+import { AngularCompilation, EmitFileResult } from './angular-compilation';
+
+/**
+ * An Angular compilation which uses a Node.js Worker thread to load and execute
+ * the TypeScript and Angular compilers. This allows for longer synchronous actions
+ * such as semantic and template diagnostics to be calculated in parallel to the
+ * other aspects of the application bundling process. The worker thread also has
+ * a separate memory pool which significantly reduces the need for adjusting the
+ * main Node.js CLI process memory settings with large application code sizes.
+ */
+export class ParallelCompilation extends AngularCompilation {
+  readonly #worker: Piscina;
+
+  constructor(readonly jit: boolean) {
+    super();
+
+    // TODO: Convert to import.meta usage during ESM transition
+    const localRequire = createRequire(__filename);
+
+    this.#worker = new Piscina({
+      minThreads: 1,
+      maxThreads: 1,
+      idleTimeout: Infinity,
+      filename: localRequire.resolve('./parallel-worker'),
+    });
+  }
+
+  override initialize(
+    tsconfig: string,
+    hostOptions: AngularHostOptions,
+    compilerOptionsTransformer?:
+      | ((compilerOptions: CompilerOptions) => CompilerOptions)
+      | undefined,
+  ): Promise<{
+    affectedFiles: ReadonlySet<SourceFile>;
+    compilerOptions: CompilerOptions;
+    referencedFiles: readonly string[];
+  }> {
+    const stylesheetChannel = new MessageChannel();
+    // The request identifier is required because Angular can issue multiple concurrent requests
+    stylesheetChannel.port1.on('message', ({ requestId, data, containingFile, stylesheetFile }) => {
+      hostOptions
+        .transformStylesheet(data, containingFile, stylesheetFile)
+        .then((value) => stylesheetChannel.port1.postMessage({ requestId, value }))
+        .catch((error) => stylesheetChannel.port1.postMessage({ requestId, error }));
+    });
+
+    // The web worker processing is a synchronous operation and uses shared memory combined with
+    // the Atomics API to block execution here until a response is received.
+    const webWorkerChannel = new MessageChannel();
+    const webWorkerSignal = new Int32Array(new SharedArrayBuffer(4));
+    webWorkerChannel.port1.on('message', ({ workerFile, containingFile }) => {
+      try {
+        const workerCodeFile = hostOptions.processWebWorker(workerFile, containingFile);
+        webWorkerChannel.port1.postMessage({ workerCodeFile });
+      } catch (error) {
+        webWorkerChannel.port1.postMessage({ error });
+      } finally {
+        Atomics.store(webWorkerSignal, 0, 1);
+        Atomics.notify(webWorkerSignal, 0);
+      }
+    });
+
+    // The compiler options transformation is a synchronous operation and uses shared memory combined
+    // with the Atomics API to block execution here until a response is received.
+    const optionsChannel = new MessageChannel();
+    const optionsSignal = new Int32Array(new SharedArrayBuffer(4));
+    optionsChannel.port1.on('message', (compilerOptions) => {
+      try {
+        const transformedOptions = compilerOptionsTransformer?.(compilerOptions) ?? compilerOptions;
+        optionsChannel.port1.postMessage({ transformedOptions });
+      } catch (error) {
+        webWorkerChannel.port1.postMessage({ error });
+      } finally {
+        Atomics.store(optionsSignal, 0, 1);
+        Atomics.notify(optionsSignal, 0);
+      }
+    });
+
+    // Execute the initialize function in the worker thread
+    return this.#worker.run(
+      {
+        fileReplacements: hostOptions.fileReplacements,
+        tsconfig,
+        jit: this.jit,
+        stylesheetPort: stylesheetChannel.port2,
+        optionsPort: optionsChannel.port2,
+        optionsSignal,
+        webWorkerPort: webWorkerChannel.port2,
+        webWorkerSignal,
+      },
+      {
+        name: 'initialize',
+        transferList: [stylesheetChannel.port2, optionsChannel.port2, webWorkerChannel.port2],
+      },
+    );
+  }
+
+  /**
+   * This is not needed with this compilation type since the worker will already send a response
+   * with the serializable esbuild compatible diagnostics.
+   */
+  protected override collectDiagnostics(): never {
+    throw new Error('Not implemented in ParallelCompilation.');
+  }
+
+  override diagnoseFiles(): Promise<{ errors?: PartialMessage[]; warnings?: PartialMessage[] }> {
+    return this.#worker.run(undefined, { name: 'diagnose' });
+  }
+
+  override emitAffectedFiles(): Promise<Iterable<EmitFileResult>> {
+    return this.#worker.run(undefined, { name: 'emit' });
+  }
+
+  override update(files: Set<string>): Promise<void> {
+    return this.#worker.run(files, { name: 'update' });
+  }
+
+  override close() {
+    // Workaround piscina bug where a worker thread will be recreated after destroy to meet the minimum.
+    this.#worker.options.minThreads = 0;
+
+    return this.#worker.destroy();
+  }
+}

packages/angular_devkit/build_angular/src/tools/esbuild/angular/compilation/parallel-worker.ts

@@ -0,0 +1,119 @@
+/**
+ * @license
+ * Copyright Google LLC All Rights Reserved.
+ *
+ * Use of this source code is governed by an MIT-style license that can be
+ * found in the LICENSE file at https://angular.io/license
+ */
+
+import assert from 'node:assert';
+import { randomUUID } from 'node:crypto';
+import { type MessagePort, receiveMessageOnPort } from 'node:worker_threads';
+import { SourceFileCache } from '../source-file-cache';
+import type { AngularCompilation } from './angular-compilation';
+import { AotCompilation } from './aot-compilation';
+import { JitCompilation } from './jit-compilation';
+
+export interface InitRequest {
+  jit: boolean;
+  tsconfig: string;
+  fileReplacements?: Record<string, string>;
+  stylesheetPort: MessagePort;
+  optionsPort: MessagePort;
+  optionsSignal: Int32Array;
+  webWorkerPort: MessagePort;
+  webWorkerSignal: Int32Array;
+}
+
+let compilation: AngularCompilation | undefined;
+
+const sourceFileCache = new SourceFileCache();
+
+export async function initialize(request: InitRequest) {
+  compilation ??= request.jit ? new JitCompilation() : new AotCompilation();
+
+  const stylesheetRequests = new Map<string, [(value: string) => void, (reason: Error) => void]>();
+  request.stylesheetPort.on('message', ({ requestId, value, error }) => {
+    if (error) {
+      stylesheetRequests.get(requestId)?.[1](error);
+    } else {
+      stylesheetRequests.get(requestId)?.[0](value);
+    }
+  });
+
+  const { compilerOptions, referencedFiles } = await compilation.initialize(
+    request.tsconfig,
+    {
+      fileReplacements: request.fileReplacements,
+      sourceFileCache,
+      modifiedFiles: sourceFileCache.modifiedFiles,
+      transformStylesheet(data, containingFile, stylesheetFile) {
+        const requestId = randomUUID();
+        const resultPromise = new Promise<string>((resolve, reject) =>
+          stylesheetRequests.set(requestId, [resolve, reject]),
+        );
+
+        request.stylesheetPort.postMessage({
+          requestId,
+          data,
+          containingFile,
+          stylesheetFile,
+        });
+
+        return resultPromise;
+      },
+      processWebWorker(workerFile, containingFile) {
+        Atomics.store(request.webWorkerSignal, 0, 0);
+        request.webWorkerPort.postMessage({ workerFile, containingFile });
+
+        Atomics.wait(request.webWorkerSignal, 0, 0);
+        const result = receiveMessageOnPort(request.webWorkerPort)?.message;
+
+        if (result?.error) {
+          throw result.error;
+        }
+
+        return result?.workerCodeFile ?? workerFile;
+      },
+    },
+    (compilerOptions) => {
+      Atomics.store(request.optionsSignal, 0, 0);
+      request.optionsPort.postMessage(compilerOptions);
+
+      Atomics.wait(request.optionsSignal, 0, 0);
+      const result = receiveMessageOnPort(request.optionsPort)?.message;
+
+      if (result?.error) {
+        throw result.error;
+      }
+
+      return result?.transformedOptions ?? compilerOptions;
+    },
+  );
+
+  return {
+    referencedFiles,
+    // TODO: Expand? `allowJs` is the only field needed currently.
+    compilerOptions: { allowJs: compilerOptions.allowJs },
+  };
+}
+
+export async function diagnose() {
+  assert(compilation);
+
+  const diagnostics = await compilation.diagnoseFiles();
+
+  return diagnostics;
+}
+
+export async function emit() {
+  assert(compilation);
+
+  const files = await compilation.emitAffectedFiles();
+
+  return [...files];
+}
+
+export function update(files: Set<string>): void {
+  sourceFileCache.invalidate(files);
+}