Improve performance of parallel bundle processing

packages/angular_devkit/build_angular/package.json

@@ -27,6 +27,7 @@
     "find-cache-dir": "3.0.0",
     "glob": "7.1.4",
     "istanbul-instrumenter-loader": "3.0.1",
+    "jest-worker": "24.9.0",
     "karma-source-map-support": "1.4.0",
     "less": "3.10.3",
     "less-loader": "5.0.0",
@@ -61,7 +62,6 @@
     "webpack-merge": "4.2.2",
     "webpack-sources": "1.4.3",
     "webpack-subresource-integrity": "1.3.3",
-    "worker-farm": "1.7.0",
     "worker-plugin": "3.2.0"
   },
   "devDependencies": {

packages/angular_devkit/build_angular/src/browser/action-executor.ts

@@ -0,0 +1,53 @@
+/**
+ * @license
+ * Copyright Google Inc. 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 JestWorker from 'jest-worker';
+import * as os from 'os';
+
+export class ActionExecutor<Input extends { size: number }, Output> {
+  private largeWorker: JestWorker;
+  private smallWorker: JestWorker;
+
+  private smallThreshold = 32 * 1024;
+
+  constructor(actionFile: string, private readonly actionName: string) {
+    // larger files are processed in a separate process to limit memory usage in the main process
+    this.largeWorker = new JestWorker(actionFile, {
+      exposedMethods: [actionName],
+    });
+
+    // small files are processed in a limited number of threads to improve speed
+    // The limited number also prevents a large increase in memory usage for an otherwise short operation
+    this.smallWorker = new JestWorker(actionFile, {
+      exposedMethods: [actionName],
+      numWorkers: os.cpus().length < 2 ? 1 : 2,
+      // Will automatically fallback to processes if not supported
+      enableWorkerThreads: true,
+    });
+  }
+
+  execute(options: Input): Promise<Output> {
+    if (options.size > this.smallThreshold) {
+      return ((this.largeWorker as unknown) as Record<string, (options: Input) => Promise<Output>>)[
+        this.actionName
+      ](options);
+    } else {
+      return ((this.smallWorker as unknown) as Record<string, (options: Input) => Promise<Output>>)[
+        this.actionName
+      ](options);
+    }
+  }
+
+  executeAll(options: Input[]): Promise<Output[]> {
+    return Promise.all(options.map(o => this.execute(o)));
+  }
+
+  stop() {
+    this.largeWorker.end();
+    this.smallWorker.end();
+  }
+}

packages/angular_devkit/build_angular/src/browser/index.ts

@@ -31,7 +31,6 @@ import { Observable, from, of } from 'rxjs';
 import { bufferCount, catchError, concatMap, map, mergeScan, switchMap } from 'rxjs/operators';
 import { ScriptTarget } from 'typescript';
 import * as webpack from 'webpack';
-import * as workerFarm from 'worker-farm';
 import { NgBuildAnalyticsPlugin } from '../../plugins/webpack/analytics';
 import { WebpackConfigOptions } from '../angular-cli-files/models/build-options';
 import {
@@ -78,6 +77,7 @@ import {
   getIndexInputFile,
   getIndexOutputFile,
 } from '../utils/webpack-browser-config';
+import { ActionExecutor } from './action-executor';
 import { Schema as BrowserBuilderSchema } from './schema';
 
 const cacache = require('cacache');
@@ -115,7 +115,7 @@ export async function buildBrowserWebpackConfigFromContext(
   options: BrowserBuilderSchema,
   context: BuilderContext,
   host: virtualFs.Host<fs.Stats> = new NodeJsSyncHost(),
-): Promise<{ config: webpack.Configuration[], projectRoot: string, projectSourceRoot?: string }> {
+): Promise<{ config: webpack.Configuration[]; projectRoot: string; projectSourceRoot?: string }> {
   return generateBrowserWebpackConfigFromContext(
     options,
     context,
@@ -449,23 +449,20 @@ export function buildWebpackBrowser(
 
                 // Attempt to get required cache entries
                 const cacheEntries = [];
+                let cached = cacheKeys.length > 0;
                 for (const key of cacheKeys) {
                   if (key) {
-                    cacheEntries.push(await cacache.get.info(cacheDownlevelPath, key));
+                    const entry = await cacache.get.info(cacheDownlevelPath, key);
+                    if (!entry) {
+                      cached = false;
+                      break;
+                    }
+                    cacheEntries.push(entry);
                   } else {
                     cacheEntries.push(null);
                   }
                 }
 
-                // Check if required cache entries are present
-                let cached = cacheKeys.length > 0;
-                for (let i = 0; i < cacheKeys.length; ++i) {
-                  if (cacheKeys[i] && !cacheEntries[i]) {
-                    cached = false;
-                    break;
-                  }
-                }
-
                 // If all required cached entries are present, use the cached entries
                 // Otherwise process the files
                 // If SRI is enabled always process the runtime bundle
@@ -587,35 +584,25 @@ export function buildWebpackBrowser(
               }
 
               if (processActions.length > 0) {
-                await new Promise<void>((resolve, reject) => {
-                  const workerFile = require.resolve('../utils/process-bundle');
-                  const workers = workerFarm(
-                    {
-                      maxRetries: 1,
-                    },
-                    path.extname(workerFile) !== '.ts'
-                      ? workerFile
-                      : require.resolve('../utils/process-bundle-bootstrap'),
-                    ['process'],
-                  );
-                  let completed = 0;
-                  const workCallback = (error: Error | null, result: ProcessBundleResult) => {
-                    if (error) {
-                      workerFarm.end(workers);
-                      reject(error);
-
-                      return;
-                    }
-
-                    processResults.push(result);
-                    if (++completed === processActions.length) {
-                      workerFarm.end(workers);
-                      resolve();
-                    }
-                  };
+                const workerFile = require.resolve('../utils/process-bundle');
+                const executor = new ActionExecutor<
+                  ProcessBundleOptions & { size: number },
+                  ProcessBundleResult
+                >(
+                  path.extname(workerFile) !== '.ts'
+                    ? workerFile
+                    : require.resolve('../utils/process-bundle-bootstrap'),
+                  'process',
+                );
 
-                  processActions.forEach(action => workers['process'](action, workCallback));
-                });
+                try {
+                  const results = await executor.executeAll(
+                    processActions.map(a => ({ ...a, size: a.code.length })),
+                  );
+                  results.forEach(result => processResults.push(result));
+                } finally {
+                  executor.stop();
+                }
               }
 
               // Runtime must be processed after all other files
@@ -625,7 +612,7 @@ export function buildWebpackBrowser(
                   runtimeData: processResults,
                 };
                 processResults.push(
-                  await import('../utils/process-bundle').then(m => m.processAsync(runtimeOptions)),
+                  await import('../utils/process-bundle').then(m => m.process(runtimeOptions)),
                 );
               }
 

packages/angular_devkit/build_angular/src/utils/process-bundle.ts

@@ -57,14 +57,7 @@ export const enum CacheKey {
   DownlevelMap = 3,
 }
 
-export function process(
-  options: ProcessBundleOptions,
-  callback: (error: Error | null, result?: ProcessBundleResult) => void,
-): void {
-  processAsync(options).then(result => callback(null, result), error => callback(error));
-}
-
-export async function processAsync(options: ProcessBundleOptions): Promise<ProcessBundleResult> {
+export async function process(options: ProcessBundleOptions): Promise<ProcessBundleResult> {
   if (!options.cacheKeys) {
     options.cacheKeys = [];
   }

yarn.lock

@@ -5860,7 +5860,7 @@ jasminewd2@^2.1.0:
   resolved "https://registry.yarnpkg.com/jasminewd2/-/jasminewd2-2.2.0.tgz#e37cf0b17f199cce23bea71b2039395246b4ec4e"
   integrity sha1-43zwsX8ZnM4jvqcbIDk5Uka07E4=
 
-jest-worker@^24.9.0:
+jest-worker@24.9.0, jest-worker@^24.9.0:
   version "24.9.0"
   resolved "https://registry.yarnpkg.com/jest-worker/-/jest-worker-24.9.0.tgz#5dbfdb5b2d322e98567898238a9697bcce67b3e5"
   integrity sha512-51PE4haMSXcHohnSMdM42anbvZANYTqMrr52tVKPqqsPJMzoP6FYYDVqahX/HrAoKEKz3uUPzSvKs9A3qR4iVw==
@@ -8959,7 +8959,6 @@ sauce-connect-launcher@^1.2.4:
 
 "sauce-connect-proxy@https://saucelabs.com/downloads/sc-4.5.4-linux.tar.gz":
   version "0.0.0"
-  uid dc5efcd2be24ddb099a85b923d6e754754651fa8
   resolved "https://saucelabs.com/downloads/sc-4.5.4-linux.tar.gz#dc5efcd2be24ddb099a85b923d6e754754651fa8"
 
 saucelabs@^1.5.0:
@@ -10890,7 +10889,7 @@ wordwrap@~0.0.2:
   resolved "https://registry.yarnpkg.com/wordwrap/-/wordwrap-0.0.3.tgz#a3d5da6cd5c0bc0008d37234bbaf1bed63059107"
   integrity sha1-o9XabNXAvAAI03I0u68b7WMFkQc=
 
-worker-farm@1.7.0, worker-farm@^1.7.0:
+worker-farm@^1.7.0:
   version "1.7.0"
   resolved "https://registry.yarnpkg.com/worker-farm/-/worker-farm-1.7.0.tgz#26a94c5391bbca926152002f69b84a4bf772e5a8"
   integrity sha512-rvw3QTZc8lAxyVrqcSGVm5yP/IJ2UcB3U0graE3LCFoZ0Yn2x4EoVSqJKdB/T5M+FLcRPjz4TDacRf3OCfNUzw==