Add support for worker_threads

lib/child/handle.js

@@ -0,0 +1,43 @@
+'use strict'
+let $module
+function handle (data, send) {
+  let idx      = data.idx
+    , child    = data.child
+    , method   = data.method
+    , args     = data.args
+    , callback = function () {
+        let _args = Array.prototype.slice.call(arguments)
+        if (_args[0] instanceof Error) {
+          let e = _args[0]
+          _args[0] = {
+              '$error'  : '$error'
+            , 'type'    : e.constructor.name
+            , 'message' : e.message
+            , 'stack'   : e.stack
+          }
+          Object.keys(e).forEach(function(key) {
+            _args[0][key] = e[key]
+          })
+        }
+        send({ idx: idx, child: child, args: _args })
+      }
+    , exec
+
+  if (method == null && typeof $module == 'function')
+    exec = $module
+  else if (typeof $module[method] == 'function')
+    exec = $module[method]
+
+  if (!exec)
+    return console.error('NO SUCH METHOD:', method)
+
+  exec.apply(null, args.concat([ callback ]))
+}
+
+module.exports = function(send) {
+  return function(data) {
+    if (!$module) return $module = require(data.module)
+    if (data == 'die') return process.exit(0)
+    handle(data, send)
+  }
+}

lib/child/index.js

@@ -1,52 +1,10 @@
 'use strict'
-
-let $module
+const handle = require('./handle')
 
 /*
   let contextProto = this.context;
   while (contextProto = Object.getPrototypeOf(contextProto)) {
     completionGroups.push(Object.getOwnPropertyNames(contextProto));
   }
 */
-
-
-function handle (data) {
-  let idx      = data.idx
-    , child    = data.child
-    , method   = data.method
-    , args     = data.args
-    , callback = function () {
-        let _args = Array.prototype.slice.call(arguments)
-        if (_args[0] instanceof Error) {
-          let e = _args[0]
-          _args[0] = {
-              '$error'  : '$error'
-            , 'type'    : e.constructor.name
-            , 'message' : e.message
-            , 'stack'   : e.stack
-          }
-          Object.keys(e).forEach(function(key) {
-            _args[0][key] = e[key]
-          })
-        }
-        process.send({ idx: idx, child: child, args: _args })
-      }
-    , exec
-
-  if (method == null && typeof $module == 'function')
-    exec = $module
-  else if (typeof $module[method] == 'function')
-    exec = $module[method]
-
-  if (!exec)
-    return console.error('NO SUCH METHOD:', method)
-
-  exec.apply(null, args.concat([ callback ]))
-}
-
-
-process.on('message', function (data) {
-  if (!$module) return $module = require(data.module)
-  if (data == 'die') return process.exit(0)
-  handle(data)
-})
+process.on('message', handle(process.send.bind(process)))

lib/child/thread.js

@@ -0,0 +1,4 @@
+'use strict'
+const handle = require('./handle')
+const {parentPort} = require('worker_threads')
+parentPort.on('message', handle(parentPort.postMessage.bind(parentPort)))

lib/farm.js

@@ -10,6 +10,7 @@ const DEFAULT_OPTIONS = {
         , maxRetries                  : Infinity
         , forcedKillTime              : 100
         , autoStart                   : false
+        , worker_threads              : false
       }
 
 const fork                    = require('./fork')
@@ -103,7 +104,7 @@ Farm.prototype.onExit = function (childId) {
 Farm.prototype.startChild = function () {
   this.childId++
 
-  let forked = fork(this.path, this.options.workerOptions)
+  let forked = fork(this.path, this.options.workerOptions, this.options.worker_threads)
     , id     = this.childId
     , c      = {
           send        : forked.send

lib/fork.js

@@ -3,8 +3,27 @@
 const childProcess = require('child_process')
     , childModule  = require.resolve('./child/index')
 
+let checked, available
+function fork (forkModule, workerOptions, worker_threads) {
+
+  // Check whether we need to run using worker_threads.
+  if (worker_threads) {
+
+    if (!checked) {
+      try {
+        require('worker_threads')
+        available = true
+      } catch (e) {
+        available = false
+        console.warn('[WARNING] Worker threads are not available! Make sure you run node 10.x or higher with --experimental-worker!')
+      }
+      checked = true
+    }
+
+    if (available) return thread(forkModule, workerOptions);
+
+  }
 
-function fork (forkModule, workerOptions) {
   // suppress --debug / --inspect flags while preserving others (like --harmony)
   let filteredArgs = process.execArgv.filter(function (v) {
         return !(/^--(debug|inspect)/).test(v)
@@ -29,5 +48,15 @@ function fork (forkModule, workerOptions) {
   }
 }
 
+function thread (forkModule, workerOptions) {
+  const {Worker} = require('worker_threads');
+  let child = new Worker(require.resolve('./child/thread'), workerOptions)
+  child.on('error', function() {})
+  child.postMessage({ module: forkModule })
+  return {
+      send  : child.postMessage.bind(child)
+    , child : child
+  }
+}
 
 module.exports = fork

lib/index.js

@@ -21,6 +21,15 @@ function farm (options, path, methods) {
   return api
 }
 
+function threaded (options, path, methods) {
+  if (typeof options == 'string') {
+    methods = path
+    path = options
+    options = {}
+  }
+  options = Object.assign({ worker_threads: true }, options)
+  return farm(options, path, methods)
+}
 
 function end (api, callback) {
   for (let i = 0; i < farms.length; i++)
@@ -31,4 +40,5 @@ function end (api, callback) {
 
 
 module.exports     = farm
-module.exports.end = end
+module.exports.threaded = threaded
+module.exports.end = threaded.end = end

package.json

@@ -24,7 +24,7 @@
     "tape": "~4.9.0"
   },
   "scripts": {
-    "test": "node ./tests/"
+    "test": "node ./tests/run"
   },
   "types": "./index.d.ts",
   "license": "MIT"

tests/index.js

@@ -1,8 +1,8 @@
 'use strict'
+module.exports = function(workerFarm) {
 
 const tape          = require('tape')
     , child_process = require('child_process')
-    , workerFarm    = require('../')
     , childPath     = require.resolve('./child')
     , fs            = require('fs')
     , os            = require('os')
@@ -562,3 +562,5 @@ tape('ensure --debug/--inspect not propagated to children', function (t) {
     t.ok(stdout.indexOf('--debug') === -1, 'child does not receive debug flag')
   })
 })
+
+}

tests/run.js

@@ -0,0 +1,4 @@
+const test = require('.')
+const workerFarm = require('..')
+test(workerFarm)
+test(workerFarm.threaded)