feat: add built-in support for concurrency limiting (#559)

Inspired by fork-ts-checker-webpack-plugin-limiter plugin - thanks @vicary

✅ Closes: #441

Author: piotr-oles

src/ForkTsCheckerWebpackPlugin.ts

@@ -20,9 +20,18 @@ import { tapAfterCompileToAddDependencies } from './hooks/tapAfterCompileToAddDe
 import { tapErrorToLogMessage } from './hooks/tapErrorToLogMessage';
 import { getForkTsCheckerWebpackPluginHooks } from './hooks/pluginHooks';
 import { tapAfterEnvironmentToPatchWatching } from './hooks/tapAfterEnvironmentToPatchWatching';
+import { createPool, Pool } from './utils/async/pool';
+import os from 'os';
 
 class ForkTsCheckerWebpackPlugin implements webpack.Plugin {
+  /**
+   * Current version of the plugin
+   */
   static readonly version: string = '{{VERSION}}'; // will be replaced by the @semantic-release/exec
+  /**
+   * Default pool for the plugin concurrency limit
+   */
+  static readonly pool: Pool = createPool(Math.max(1, os.cpus().length));
 
   private readonly options: ForkTsCheckerWebpackPluginOptions;
 

src/hooks/tapAfterCompileToGetIssues.ts

@@ -1,5 +1,4 @@
 import webpack from 'webpack';
-import path from 'path';
 import { ForkTsCheckerWebpackPluginConfiguration } from '../ForkTsCheckerWebpackPluginConfiguration';
 import { ForkTsCheckerWebpackPluginState } from '../ForkTsCheckerWebpackPluginState';
 import { getForkTsCheckerWebpackPluginHooks } from './pluginHooks';

src/hooks/tapDoneToAsyncGetIssues.ts

@@ -1,6 +1,5 @@
 import webpack from 'webpack';
 import chalk from 'chalk';
-import path from 'path';
 import { ForkTsCheckerWebpackPluginConfiguration } from '../ForkTsCheckerWebpackPluginConfiguration';
 import { ForkTsCheckerWebpackPluginState } from '../ForkTsCheckerWebpackPluginState';
 import { getForkTsCheckerWebpackPluginHooks } from './pluginHooks';

src/hooks/tapStartToConnectAndRunReporter.ts

@@ -8,6 +8,7 @@ import { tapDoneToAsyncGetIssues } from './tapDoneToAsyncGetIssues';
 import { tapAfterCompileToGetIssues } from './tapAfterCompileToGetIssues';
 import { interceptDoneToGetWebpackDevServerTap } from './interceptDoneToGetWebpackDevServerTap';
 import { Issue } from '../issue';
+import { ForkTsCheckerWebpackPlugin } from '../ForkTsCheckerWebpackPlugin';
 
 function tapStartToConnectAndRunReporter(
   compiler: webpack.Compiler,
@@ -76,40 +77,44 @@ function tapStartToConnectAndRunReporter(
       rejectIssues = reject;
     });
     const previousReportPromise = state.reportPromise;
-    state.reportPromise = new Promise(async (resolve) => {
-      change = await hooks.start.promise(change, compilation);
-
-      try {
-        await reporter.connect();
-
-        const previousReport = await previousReportPromise;
-        if (previousReport) {
-          await previousReport.close();
-        }
-
-        const report = await reporter.getReport(change);
-        resolve(report);
-
-        report
-          .getDependencies()
-          .then(resolveDependencies)
-          .catch(rejectedDependencies)
-          .finally(() => {
-            // get issues after dependencies are resolved as it can be blocking
-            report.getIssues().then(resolveIssues).catch(rejectIssues);
-          });
-      } catch (error) {
-        if (error instanceof OperationCanceledError) {
-          hooks.canceled.call(compilation);
-        } else {
-          hooks.error.call(error, compilation);
-        }
-
-        resolve(undefined);
-        resolveDependencies(undefined);
-        resolveIssues(undefined);
-      }
-    });
+    state.reportPromise = ForkTsCheckerWebpackPlugin.pool.submit(
+      (done) =>
+        new Promise(async (resolve) => {
+          change = await hooks.start.promise(change, compilation);
+
+          try {
+            await reporter.connect();
+
+            const previousReport = await previousReportPromise;
+            if (previousReport) {
+              await previousReport.close();
+            }
+
+            const report = await reporter.getReport(change);
+            resolve(report);
+
+            report
+              .getDependencies()
+              .then(resolveDependencies)
+              .catch(rejectedDependencies)
+              .finally(() => {
+                // get issues after dependencies are resolved as it can be blocking
+                report.getIssues().then(resolveIssues).catch(rejectIssues).finally(done);
+              });
+          } catch (error) {
+            if (error instanceof OperationCanceledError) {
+              hooks.canceled.call(compilation);
+            } else {
+              hooks.error.call(error, compilation);
+            }
+
+            resolve(undefined);
+            resolveDependencies(undefined);
+            resolveIssues(undefined);
+            done();
+          }
+        })
+    );
   });
 }
 

src/utils/async/pool.ts

@@ -0,0 +1,49 @@
+// provide done callback because our promise chain is a little bit complicated
+type Task<T> = (done: () => void) => Promise<T>;
+
+interface Pool {
+  submit<T>(task: Task<T>): Promise<T>;
+  size: number;
+  readonly pending: number;
+}
+
+function createPool(size: number): Pool {
+  let pendingPromises: Promise<unknown>[] = [];
+
+  const pool = {
+    async submit<T>(task: Task<T>): Promise<T> {
+      while (pendingPromises.length >= pool.size) {
+        await Promise.race(pendingPromises).catch(() => undefined);
+      }
+
+      let resolve: (result: T) => void;
+      let reject: (error: Error) => void;
+      const taskPromise = new Promise<T>((taskResolve, taskReject) => {
+        resolve = taskResolve;
+        reject = taskReject;
+      });
+
+      const donePromise = new Promise((doneResolve) => {
+        task(() => {
+          doneResolve(undefined);
+          pendingPromises = pendingPromises.filter(
+            (pendingPromise) => pendingPromise !== donePromise
+          );
+        })
+          .then(resolve)
+          .catch(reject);
+      });
+      pendingPromises.push(donePromise);
+
+      return taskPromise;
+    },
+    size,
+    get pending() {
+      return pendingPromises.length;
+    },
+  };
+
+  return pool;
+}
+
+export { Pool, createPool };

test/unit/utils/async/pool.spec.ts

@@ -0,0 +1,35 @@
+import { createPool } from 'lib/utils/async/pool';
+
+describe('createPool', () => {
+  it('creates new pool', () => {
+    const pool = createPool(10);
+    expect(pool).toBeDefined();
+    expect(pool.size).toEqual(10);
+    expect(pool.pending).toEqual(0);
+    expect(pool.submit).toBeInstanceOf(Function);
+  });
+
+  it('limits concurrency', async () => {
+    const pool = createPool(2);
+    const shortTask = jest.fn(async (done: () => void) => {
+      setTimeout(done, 10);
+    });
+    const longTask = jest.fn(async (done: () => void) => {
+      setTimeout(done, 10000);
+    });
+
+    pool.submit(shortTask);
+    pool.submit(shortTask);
+    pool.submit(longTask);
+    pool.submit(longTask);
+    pool.submit(longTask);
+
+    expect(shortTask).toHaveBeenCalledTimes(2);
+    expect(longTask).toHaveBeenCalledTimes(0);
+
+    await new Promise((resolve) => setTimeout(resolve, 200));
+
+    expect(shortTask).toHaveBeenCalledTimes(2);
+    expect(longTask).toHaveBeenCalledTimes(2);
+  });
+});