Cluster support

Author: Piotr Oleś

README.md

@@ -59,6 +59,7 @@ TODO
 |**colors**| `boolean` | If `false`, disables colors for logger. Default: `true`. |
 |**logger**| `LoggerInterface` | Logger instance. It should be object that implements method: `error`, `warn`, `info`. Default: `console`.|
 |**silent**| `boolean` | If `true`, logger will not be used. Default: `false`.|
+|**cluster**| `number` | You can split type checking to few workers to speed-up on increment build. But remember: if you don't want type checker to affect build time, you should keep 1 core for build and 1 core for system. Also - node doesn't share memory so keep in mind that memory usage will increase linear. Default: `1`.|
 
 ## License ##
 MIT

lib/FilesRegister.js

@@ -1,27 +1,21 @@
 
-function FilesRegister (oldRegister) {
+function FilesRegister () {
   this.register = {};
-
-  if (oldRegister) {
-    oldRegister.forEach(function (fileName, fileEntry) {
-      if (!fileEntry.modified) {
-        // don't copy - just set reference to save memory
-        this.register[fileName] = fileEntry;
-      }
-    }.bind(this));
-  }
 }
 module.exports = FilesRegister;
 
-FilesRegister.prototype.forEach = function (loop) {
+FilesRegister.prototype.forEach = function (callback) {
   Object.keys(this.register).forEach(function (key) {
-    loop(key, this.register[key]);
+    callback(key, this.register[key]);
   }.bind(this));
 };
 
+FilesRegister.prototype.keys = function() {
+  return Object.keys(this.register);
+};
+
 FilesRegister.prototype.addFile = function (fileName) {
   this.register[fileName] = {
-    modified: false,
     mtime: undefined,
     source: undefined,
     linted: false,
@@ -75,10 +69,12 @@ FilesRegister.prototype.setMtime = function (fileName, mtime) {
   this.ensureFile(fileName);
 
   if (this.register[fileName].mtime !== mtime) {
-    this.register[fileName].modified = true;
     this.register[fileName].linted = false;
     this.register[fileName].lints = [];
     this.register[fileName].mtime = mtime;
+
+    // file has been changed - we are not sure about it's current source
+    this.register[fileName].source = undefined;
   }
 };
 
@@ -96,6 +92,7 @@ FilesRegister.prototype.consumeLint = function (lint) {
   var fileName = lint.getFileName();
 
   this.ensureFile(fileName);
+
   this.register[fileName].linted = true;
   this.register[fileName].lints.push(lint);
 };
@@ -105,3 +102,9 @@ FilesRegister.prototype.consumeLints = function (lints) {
     this.consumeLint(lint);
   }.bind(this));
 };
+
+FilesRegister.prototype.setAllLinted = function () {
+  this.keys().forEach(function (fileName) {
+    this.register[fileName].linted = true;
+  }.bind(this))
+};

lib/IncrementalChecker.js

@@ -4,11 +4,17 @@ var path = require('path');
 var endsWith = require('lodash.endswith');
 var FilesRegister = require('./FilesRegister');
 var FilesWatcher = require('./FilesWatcher');
+var WorkSplitter = require('./WorkSplitter');
 
-function IncrementalChecker (programConfigFile, linterConfigFile, watchPaths) {
+function IncrementalChecker (programConfigFile, linterConfigFile, watchPaths, workNumber, workDivision) {
   this.programConfigFile = programConfigFile;
   this.linterConfigFile = linterConfigFile;
   this.watchPaths = watchPaths;
+  this.workNumber = workNumber || 0;
+  this.workDivision = workDivision || 1;
+
+  // it's shared between compilations
+  this.register = new FilesRegister();
 }
 module.exports = IncrementalChecker;
 
@@ -35,7 +41,7 @@ IncrementalChecker.createProgram = function (programConfig, register, watcher, o
     if (!watcher.isWatchingFile(filePath)) {
       var stats = fs.statSync(filePath);
 
-      register.setMtime(filePath, stats.mtime);
+      register.setMtime(filePath, stats.mtime.valueOf());
     }
 
     // get source file only if there is no source in files register
@@ -64,13 +70,11 @@ IncrementalChecker.createLinter = function (program) {
 };
 
 IncrementalChecker.prototype.nextIteration = function () {
-  this.register = new FilesRegister(this.register);
-
   if (!this.watcher) {
     this.watcher = new FilesWatcher(this.watchPaths, ['.ts', '.tsx']);
 
     // connect watcher with register
-    this.watcher.onChange(function (filePath, stats) { this.register.setMtime(filePath, stats.mtime); }.bind(this));
+    this.watcher.onChange(function (filePath, stats) { this.register.setMtime(filePath, stats.mtime.valueOf()); }.bind(this));
     this.watcher.onUnlink(function (filePath) { this.register.removeFile(filePath); }.bind(this));
 
     this.watcher.watch();
@@ -89,7 +93,27 @@ IncrementalChecker.prototype.nextIteration = function () {
 };
 
 IncrementalChecker.prototype.getDiagnostics = function (cancellationToken) {
-  var diagnostics = this.program.getSemanticDiagnostics(undefined, cancellationToken);
+  var diagnostics = [];
+  var times = [];
+  var filesToCheck = this.program.getSourceFiles();
+  var work = new WorkSplitter(filesToCheck, this.workNumber, this.workDivision);
+
+  work.forEach(function (sourceFile, i) {
+    times[i] = process.hrtime();
+    if (cancellationToken) {
+      cancellationToken.throwIfCancellationRequested();
+    }
+
+    ts.addRange(diagnostics, this.program.getSemanticDiagnostics(sourceFile, cancellationToken));
+    times[i] = process.hrtime(times[i]);
+  }.bind(this));
+
+  var elapsed = times.map(function (time) {
+    return Math.round(Math.round(time[0] * 1E9 + time[1]) / 1E6); // in ms
+  });
+  elapsed.sort(function (a, b) { return b - a; });
+
+  diagnostics = ts.sortAndDeduplicateDiagnostics(diagnostics);
 
   return diagnostics.map(function (diagnostic) {
     var position = diagnostic.file.getLineAndCharacterOfPosition(diagnostic.start);
@@ -108,18 +132,20 @@ IncrementalChecker.prototype.getDiagnostics = function (cancellationToken) {
 };
 
 IncrementalChecker.prototype.getLints = function (cancellationToken) {
-  this.register.forEach(function (fileName, fileEntry) {
+  var filesToLint = this.register.keys().filter(function (fileName) {
+    return !endsWith(fileName, '.d.ts') && !this.register.getFile(fileName).linted;
+  }.bind(this));
+
+  var work = new WorkSplitter(filesToLint, this.workNumber, this.workDivision);
+
+  work.forEach(function (fileName) {
     cancellationToken.throwIfCancellationRequested();
 
-    if (!endsWith(fileName, '.d.ts') && !fileEntry.linted) {
-      this.linter.lint(fileName, undefined, this.linterConfig);
-      fileEntry.linted = true;
-    }
+    this.linter.lint(fileName, undefined, this.linterConfig);
   }.bind(this));
 
-  this.register.consumeLints(
-    this.linter.getResult().failures
-  );
+  this.register.setAllLinted();
+  this.register.consumeLints(this.linter.getResult().failures);
 
   var lints = this.register.getLints();
 

lib/WorkResultSet.js

@@ -0,0 +1,40 @@
+var ts = require('typescript');
+
+function WorkResultSet() {
+  this.result = {};
+}
+module.exports = WorkResultSet;
+
+WorkResultSet.prototype.set = function (key, result) {
+  this.result[key] = result;
+};
+
+WorkResultSet.prototype.has = function (key) {
+  return undefined !== this.result[key];
+};
+
+WorkResultSet.prototype.clear = function () {
+  this.result = {};
+};
+
+WorkResultSet.prototype.done = function (keys) {
+  return keys.every(function (key) {
+    return this.has(key)
+  }.bind(this));
+};
+
+WorkResultSet.prototype.merge = function () {
+  var merged = {
+    diagnostics: [],
+    lints: []
+  };
+
+  Object.keys(this.result).forEach(function (key) {
+    merged.diagnostics.push.apply(merged.diagnostics, this.result[key].diagnostics);
+    merged.lints.push.apply(merged.lints, this.result[key].lints);
+  }.bind(this));
+
+  merged.diagnostics = ts.sortAndDeduplicateDiagnostics(merged.diagnostics);
+
+  return merged;
+};

lib/WorkSplitter.js

@@ -0,0 +1,33 @@
+
+function WorkSplitter (workSet, workNumber, workDivision) {
+  this.workSet = workSet;
+  this.workNumber = workNumber;
+  this.workDivision = workDivision;
+  this.workSize = Math.floor(this.workSet.length / this.workDivision);
+  this.workBegin = this.workNumber * this.workSize;
+  this.workEnd = this.workBegin + this.workSize;
+
+  // be sure that we will process all work for odd workSize.
+  if (this.workNumber === this.workSet.length - 1) {
+    this.workEnd = this.workSet.length;
+  }
+}
+module.exports = WorkSplitter;
+
+WorkSplitter.prototype.forEach = function (callback) {
+  for (var i = this.workBegin; i < this.workEnd; ++i) {
+    callback(this.workSet[i], i);
+  }
+};
+
+WorkSplitter.prototype.getWorkSize = function () {
+  return this.workSize;
+};
+
+WorkSplitter.prototype.getWorkBegin = function () {
+  return this.workBegin;
+};
+
+WorkSplitter.prototype.getWorkEnd = function () {
+  return this.workEnd;
+};

lib/cluster.js

@@ -0,0 +1,41 @@
+var cluster = require('cluster');
+var process = require('process');
+
+if (cluster.isMaster) {
+  var WorkResultSet = require('./WorkResultSet');
+  var resultSet = new WorkResultSet();
+
+  // fork workers...
+  var division = parseInt(process.env.WORK_DIVISION);
+
+  for (var i = 0; i < division; i++) {
+    cluster.fork({ WORK_NUMBER: i });
+  }
+
+  var workerIds = Object.keys(cluster.workers);
+
+  process.on('message', function (message) {
+    // broadcast message to all workers
+    workerIds.forEach(function (workerId) {
+      cluster.workers[workerId].send(message);
+    });
+
+    // clear previous result set
+    resultSet.clear();
+  });
+
+  // listen to all workers
+  workerIds.forEach(function (workerId) {
+    cluster.workers[workerId].on('message', function (message) {
+      // set result from worker
+      resultSet.set(workerId, message);
+
+      // if we have result from all workers, send merged
+      if (resultSet.done(workerIds)) {
+        process.send(resultSet.merge());
+      }
+    });
+  });
+} else {
+  require('./service');
+}

lib/index.js

@@ -22,6 +22,7 @@ function ForkTsCheckerWebpackPlugin (options) {
   this.ignoreLints = options.ignoreLints || [];
   this.logger = options.logger || console;
   this.silent = !!options.silent;
+  this.cluster = options.cluster || 1;
 
   this.tsconfigPath = undefined;
   this.tslintPath = undefined;
@@ -151,13 +152,14 @@ ForkTsCheckerWebpackPlugin.prototype.pluginDone = function () {
 
 ForkTsCheckerWebpackPlugin.prototype.spawnService = function () {
   this.service = childProcess.fork(
-    path.resolve(__dirname, './service.js'),
+    path.resolve(__dirname, this.cluster > 1 ? './cluster.js' : './service.js'),
     [],
     {
       env: {
         TSCONFIG: this.tsconfigPath,
         TSLINT: this.tslintPath,
-        WATCH: this.watchPaths.join('|')
+        WATCH: this.watchPaths.join('|'),
+        WORK_DIVISION: this.cluster > 1 ? this.cluster : 1
       }
     }
   );
@@ -193,6 +195,10 @@ ForkTsCheckerWebpackPlugin.prototype.spawnService = function () {
 };
 
 ForkTsCheckerWebpackPlugin.prototype.handleServiceMessage = function (message) {
+  if (message.type === 'log') {
+    console.log(message.payload);
+    return;
+  }
   if (this.cancellationToken) {
     this.cancellationToken.cleanupCancellation();
     // job is done - nothing to cancel

lib/service.js

@@ -6,7 +6,9 @@ var CancellationToken = require('./CancellationToken');
 var checker = new IncrementalChecker(
   process.env.TSCONFIG,
   process.env.TSLINT,
-  process.env.WATCH === '' ? [] : process.env.WATCH.split('|')
+  process.env.WATCH === '' ? [] : process.env.WATCH.split('|'),
+  parseInt(process.env.WORK_NUMBER, 10),
+  parseInt(process.env.WORK_DIVISION, 10)
 );
 
 var diagnostics = [];