reuse processes using a process pool

.editorconfig

@@ -6,5 +6,5 @@ insert_final_newline = true
 indent_style = space
 indent_size = 2
 
-[{*.js,test/index.sh}]
+[{*.js}]
 indent_size = 4

.travis.yml

@@ -3,17 +3,11 @@ node_js:
   - "stable"
   - "8"
   - "9"
-env:
-  - CXX=g++-4.8
-addons:
-  apt:
-    sources:
-    - ubuntu-toolchain-r-test
-    packages:
-    - gcc-4.8
-    - g++-4.8
 before_install: npm install -g greenkeeper-lockfile
 install: npm install
 before_script: greenkeeper-lockfile-update
 after_script: greenkeeper-lockfile-upload
 script: npm run test:ci
+cache:
+  directories:
+    - node_modules

src/main/mocha.ts

@@ -1,17 +1,17 @@
-import { fork } from 'child_process';
+import * as assert from 'assert';
 import * as CircularJSON from 'circular-json';
 import * as debug from 'debug';
 import * as Mocha from 'mocha';
 import { resolve as pathResolve } from 'path';
 
+import ProcessPool from './process-pool';
 import RunnerMain from './runner';
 import TaskManager from './task-manager';
 import {
-  removeDebugArgs,
   subprocessParseReviver,
 } from './util';
 
-import { DEBUG_SUBPROCESS, SUITE_OWN_OPTIONS } from '../config';
+import { SUITE_OWN_OPTIONS } from '../config';
 import {
   IRetriedTest,
   ISubprocessOutputMessage,
@@ -24,6 +24,7 @@ import {
 const debugLog = debug('mocha-parallel-tests');
 
 export default class MochaWrapper extends Mocha {
+  private pool = new ProcessPool();
   private isTypescriptRunMode = false;
   private maxParallel: number | undefined;
   private requires: string[] = [];
@@ -50,6 +51,7 @@ export default class MochaWrapper extends Mocha {
 
   setMaxParallel(maxParallel: number) {
     this.maxParallel = maxParallel;
+    this.pool.setMaxParallel(maxParallel);
   }
 
   enableExitMode() {
@@ -136,6 +138,7 @@ export default class MochaWrapper extends Mocha {
         };
 
         runner.emitFinishEvents(done);
+        this.pool.destroyAll();
       });
 
     return runner;
@@ -162,89 +165,99 @@ export default class MochaWrapper extends Mocha {
   }
 
   private spawnTestProcess(file: string): Promise<ISubprocessResult> {
-    return new Promise((resolve) => {
-      const nodeFlags: string[] = [];
-      const extension = this.isTypescriptRunMode ? 'ts' : 'js';
-      const runnerPath = pathResolve(__dirname, `../subprocess/runner.${extension}`);
+    return new Promise<ISubprocessResult>(async (resolve) => {
       const resolvedFilePath = pathResolve(file);
 
-      const forkArgs: string[] = ['--test', resolvedFilePath];
+      const testOptions: {[key: string]: any} = { test: resolvedFilePath };
+
       for (const option of SUITE_OWN_OPTIONS) {
         const propValue = this.suite[option]();
         // bail is undefined by default, we need to somehow pass its value to the subprocess
-        forkArgs.push(`--${option}`, propValue === undefined ? false : propValue);
+        testOptions[option] = propValue === undefined ? false : propValue;
       }
 
       for (const requirePath of this.requires) {
-        forkArgs.push('--require', requirePath);
+        testOptions.require = requirePath;
       }
 
-      for (const compilerPath of this.compilers) {
-        forkArgs.push('--compilers', compilerPath);
-      }
+      testOptions.compilers = this.compilers || [];
 
       if (this.options.delay) {
-        forkArgs.push('--delay');
+        testOptions.delay = true;
       }
 
       if (this.options.grep) {
-        forkArgs.push('--grep', this.options.grep.toString());
+        testOptions.grep = this.options.grep.toString();
       }
 
       if (this.exitImmediately) {
-        forkArgs.push('--exit');
+        testOptions.exit = true;
       }
 
       if (this.options.fullStackTrace) {
-        forkArgs.push('--full-trace');
+        testOptions.fullStackTrace = true;
       }
 
-      const test = fork(runnerPath, forkArgs, {
-        // otherwise `--inspect-brk` and other params will be passed to subprocess
-        execArgv: process.execArgv.filter(removeDebugArgs),
-        stdio: ['ipc'],
-      });
-
-      if (this.isTypescriptRunMode) {
-        nodeFlags.push('--require', 'ts-node/register');
+      let test;
+      try {
+        test = await this.pool.getOrCreate(this.isTypescriptRunMode);
+      } catch (e) {
+        throw e;
       }
 
-      debugLog('Process spawned. You can run it manually with this command:');
-      debugLog(`node ${nodeFlags.join(' ')} ${runnerPath} ${forkArgs.concat([DEBUG_SUBPROCESS.argument]).join(' ')}`);
+      test.send(JSON.stringify({ type: 'start', testOptions }));
 
       const events: Array<ISubprocessOutputMessage | ISubprocessRunnerMessage> = [];
       let syncedSubprocessData: ISubprocessSyncedData | undefined;
       const startedAt = Date.now();
 
-      test.on('message', function onMessageHandler({ event, data }) {
+      const clean = () => {
+        test.removeListener('message', onMessageHandler);
+        test.stdout.removeListener('data', onStdoutData);
+        test.stderr.removeListener('data', onStderrData);
+        test.removeListener('close', onClose);
+        test.destroy();
+      };
+
+      const onMessageHandler = ({ event, data }: { event: string, data: any }) => {
         if (event === 'sync') {
           syncedSubprocessData = data;
+        } else if (event === 'end') {
+          clean();
+          resolve({
+            code: data.code || 0,
+            events,
+            execTime: Date.now() - startedAt,
+            file,
+            syncedSubprocessData,
+          });
         } else {
+          assert(event);
           events.push({
             data,
             event,
             type: 'runner',
           });
         }
-      });
+      };
 
-      test.stdout.on('data', function onStdoutData(data: Buffer) {
+      const onStdoutData = (data: Buffer) => {
         events.push({
           data,
           event: undefined,
           type: 'stdout',
         });
-      });
+      };
 
-      test.stderr.on('data', function onStderrData(data: Buffer) {
+      const onStderrData = (data: Buffer) => {
         events.push({
           data,
           event: undefined,
           type: 'stderr',
         });
-      });
+      };
 
-      test.on('close', (code) => {
+      const onClose = (code: number) => {
         debugLog(`Process for ${file} exited with code ${code}`);
 
         resolve({
@@ -254,7 +267,12 @@ export default class MochaWrapper extends Mocha {
           file,
           syncedSubprocessData,
         });
-      });
+      };
+
+      test.on('message', onMessageHandler);
+      test.stdout.on('data', onStdoutData);
+      test.stderr.on('data', onStderrData);
+      test.on('close', onClose);
     });
   }
 }

src/main/process-pool.ts

@@ -0,0 +1,92 @@
+import { fork } from 'child_process';
+import * as os from 'os';
+import { resolve as pathResolve } from 'path';
+import { removeDebugArgs } from './util';
+
+type eventFn = (data: any) => void;
+
+interface IMochaProcess {
+  send: (msg: any) => void;
+  kill: () => void;
+  destroy: () => void;
+  on: (event: string, fn: eventFn) => void;
+  removeListener: (event: string, eventFn) => void;
+  stdout: NodeJS.ReadableStream;
+  stderr: NodeJS.ReadableStream;
+}
+
+export default class ProcessPool {
+  private maxParallel: number;
+  private waitingList: Array<(proc: IMochaProcess) => void> = [];
+  private unusedProcesses: IMochaProcess[] = [];
+  private processes: IMochaProcess[] = [];
+
+  constructor() {
+    this.maxParallel = os.cpus().length;
+  }
+
+  setMaxParallel(n: number) {
+    this.maxParallel = n;
+  }
+
+  async getOrCreate(isTypescriptRunMode: boolean): Promise<IMochaProcess> {
+    const extension = isTypescriptRunMode ? 'ts' : 'js';
+    const runnerPath = pathResolve(__dirname, `../subprocess/runner.${extension}`);
+
+    const lastUnusedProcess = this.unusedProcesses.pop();
+    if (lastUnusedProcess) {
+      return lastUnusedProcess;
+    }
+
+    if (this.processes.length >= this.maxParallel) {
+      const process: IMochaProcess = await new Promise<IMochaProcess>((resolve) => {
+        this.waitingList.push(resolve);
+      });
+      return process;
+    }
+    return this.create(runnerPath, {
+      // otherwise `--inspect-brk` and other params will be passed to subprocess
+      execArgv: process.execArgv.filter(removeDebugArgs),
+      stdio: ['ipc'],
+    });
+  }
+
+  create(runnerPath: string, opt: object): IMochaProcess {
+    const process = fork(runnerPath, [], opt);
+
+    const handle = {
+      destroy: () => {
+        const nextOnWaitingList = this.waitingList.pop();
+        if (nextOnWaitingList) {
+          nextOnWaitingList(handle);
+        } else {
+          this.unusedProcesses.push(handle);
+        }
+      },
+      kill: () => {
+        process.kill();
+      },
+      on: (ev, fn) => {
+        process.on(ev, fn);
+      },
+      removeListener: (ev, fn) => {
+        process.removeListener(ev, fn);
+      },
+      send: (msg: any) => {
+        process.send(msg);
+      },
+      stderr: process.stderr,
+      stdout: process.stdout,
+    };
+
+    this.processes.push(handle);
+
+    return handle;
+  }
+
+  destroyAll() {
+    this.processes.forEach((proc: IMochaProcess) => {
+      proc.kill();
+    });
+  }
+}