diff --git a/.travis.yml b/.travis.yml index 7af5642..0a71f4a 100644 --- a/.travis.yml +++ b/.travis.yml @@ -2,6 +2,7 @@ language: node_js node_js: - 6 - 8 + - 9 - 10 - 12 branches: diff --git a/README.md b/README.md index b629c06..737fdf1 100644 --- a/README.md +++ b/README.md @@ -3,7 +3,7 @@ [![NPM](https://nodei.co/npm/worker-farm.png?downloads=true&downloadRank=true&stars=true)](https://nodei.co/npm/worker-farm/) -Distribute processing tasks to child processes with an über-simple API and baked-in durability & custom concurrency options. *Available in npm as worker-farm*. +Distribute processing tasks to child processes or worker threads with an über-simple API and baked-in durability & custom concurrency options. *Available in npm as worker-farm*. ## Example @@ -48,6 +48,58 @@ We'll get an output something like the following: This example is contained in the *[examples/basic](https://github.com/rvagg/node-worker-farm/tree/master/examples/basic/)* directory. +### Using worker_threads + +Since version 2.0.0 you can also make use of node's [worker_threads](https://nodejs.org/api/worker_threads.html) module. +This requires you to use the module with node 12.x or higher, or 10.5 or higher with the ```--experimental-worker``` flag. +If ```worker_threads``` is not available, the module will fall back to the default implementation using child_processes, but a warning will be emited. + +Given a file, *child.js*: + +```js +'use strict' +const {threadId} = require('worker_threads') + +module.exports = function (inp, callback) { + callback(null, inp + ' BAR (' + [process.pid, threadId].join(':') + ')') +} +``` + +And a main file: + +```js +'use strict' + +let workerFarm = require('../../') + , workers = workerFarm.threaded(require.resolve('./child')) + , ret = 0 + +for (let i = 0; i < 10; i++) { + workers('#' + i + ' FOO', function(err, outp) { + console.log(outp) + if (++ret == 10) + workerFarm.end(workers) + }) +} +``` + +We'll get an output something like the following: + +``` +#3 FOO BAR (22236:4) +#5 FOO BAR (22236:6) +#1 FOO BAR (22236:2) +#9 FOO BAR (22236:2) +#4 FOO BAR (22236:5) +#7 FOO BAR (22236:8) +#2 FOO BAR (22236:3) +#0 FOO BAR (22236:1) +#8 FOO BAR (22236:1) +#6 FOO BAR (22236:7) +``` + +This example is contained in the *[examples/threaded](https://github.com/rvagg/node-worker-farm/tree/master/examples/threaded/)* directory. + ### Example #1: Estimating π using child workers You will also find a more complex example in *[examples/pi](https://github.com/rvagg/node-worker-farm/tree/master/examples/pi/)* that estimates the value of **π** by using a Monte Carlo *area-under-the-curve* method and compares the speed of doing it all in-process vs using child workers to complete separate portions. @@ -63,6 +115,52 @@ Doing it the fast (multi-process) way... took 1985 milliseconds ``` +### Example #2: Using transferLists with worker threads + +The [benefit](https://nodejs.org/docs/latest-v10.x/api/worker_threads.html#worker_threads_worker_threads) of using worker threads compared to child processes is that we can make use of transferLists and SharedArrayBuffers to efficiently pass data around. + +When passing data from a child to the main thread, you can specify a transferList as third argument to the callback like +```js +module.exports = function(callback) { + let result = getLargeDataStructure() + let transferList = getTransferListForResult(result) + callback(null, result, transferList) +} +``` + +When passing data to a child, you can specify a transferList after the callback like +```js +let workers = workerFarm(require.resolve('path/to/child')) +workers(all, your, args, function(err, result) { + // Do something with the result +}, transferList) +``` + +Beware that **after** transferring, the data is no longer available on the sending side. +However, because we're using a queue it might be possible that after calling a worker, data may still be available. +Consider +```js +// Let's run a task very often, much more than we have child threads. +for (let i = 0; i < 100; i++) { + let arr = new Uint8Array(1024); + workers(function(err, result) { + + }, [arr.buffer]) + + // What will be the length of arr here? If the array was transferred + // immediately to the child, arr.length === 0, but if the call is still in + // the queue - arr will still be available because it hasn't been + // transferred yet! In that case arr.length === 1024! + let schrodingersCat = { + dead: arr.length === 0 + } + +} +``` +This behavior can hence lead to subtle bugs and therefore **YOU SHOULD NEVER** use an ArrayBuffer anymore after you've specified it in a transferList. + +Have a look at the *[examples/transfer](https://github.com/rvagg/node-worker-farm/tree/master/examples/transfer/)* directory for an example that shows that transferLists can pass data around more efficiently. + ## Durability An important feature of Worker Farm is **call durability**. If a child process dies for any reason during the execution of call(s), those calls will be re-queued and taken care of by other child processes. In this way, when you ask for something to be done, unless there is something *seriously* wrong with what you're doing, you should get a result on your callback function. @@ -83,6 +181,12 @@ Worker Farm exports a main function and an `end()` method. The main function set In its most basic form, you call `workerFarm()` with the path to a module file to be invoked by the child process. You should use an **absolute path** to the module file, the best way to obtain the path is with `require.resolve('./path/to/module')`, this function can be used in exactly the same way as `require('./path/to/module')` but it returns an absolute path. +### workerFarm.threaded([options, ]pathToModule[, exportedMethods]) + +Using this method will try to use node's `worker_threads` module if it's available. +If it's not available, this will be detected and the module will default to the default implementation with child processes, displaying a warning that the `worker_threads` module is not available. +The api of the method is exactly the same as the `workerFarm()` function. + #### `exportedMethods` If your module exports a single function on `module.exports` then you should omit the final parameter. However, if you are exporting multiple functions on `module.exports` then you should list them in an Array of Strings: @@ -115,7 +219,7 @@ If you don't provide an `options` object then the following defaults will be use } ``` - * **workerOptions** allows you to customize all the parameters passed to child nodes. This object supports [all possible options of `child_process.fork`](https://nodejs.org/api/child_process.html#child_process_child_process_fork_modulepath_args_options). The default options passed are the parent `execArgv`, `cwd` and `env`. Any (or all) of them can be overridden, and others can be added as well. + * **workerOptions** allows you to customize all the parameters passed to child nodes. This object supports [all possible options of `child_process.fork`](https://nodejs.org/api/child_process.html#child_process_child_process_fork_modulepath_args_options) or [all possible options of `Worker`](https://nodejs.org/api/worker_threads.html#worker_threads_new_worker_filename_options) when using the threaded implementation. The default options passed are the parent `execArgv`, `cwd` and `env`. Any (or all) of them can be overridden, and others can be added as well. * **maxCallsPerWorker** allows you to control the lifespan of your child processes. A positive number will indicate that you only want each child to accept that many calls before it is terminated. This may be useful if you need to control memory leaks or similar in child processes. @@ -131,7 +235,7 @@ If you don't provide an `options` object then the following defaults will be use * **autoStart** when set to `true` will start the workers as early as possible. Use this when your workers have to do expensive initialization. That way they'll be ready when the first request comes through. - * **onChild** when new child process starts this callback will be called with subprocess object as an argument. Use this when you need to add some custom communication with child processes. + * **onChild** when new child process (or worker when using threads) starts this callback will be called with subprocess (or worker) object as an argument. Use this when you need to add some custom communication with children. ### workerFarm.end(farm) @@ -141,6 +245,25 @@ Any calls that are queued and not yet being handled by a child process will be d Once you end a farm, it won't handle any more calls, so don't even try! +## Breaking changes in v2 + +Although not explicitly documented, in v1 it was possible to pass multiple arguments to the callback +```js +// BEWARE! CODE BELOW NO LONGER WORKS IN v2! + +// child.js +module.exports = function(callback) { + callback(null, 'result', 'another result') +} + +// parent.js +workers(function(err, one, two) { + console.log(one, two) // Logs 'result' 'another result' +}) +``` +This is no longer possible in v2 because the third argument is considered to be a transferList. +If you're using the default implementation that uses child processes, the third (and fourth, and fifth, ...) argument specified to the callback will simply be ignored. + ## Related * [farm-cli](https://github.com/Kikobeats/farm-cli) – Launch a farm of workers from CLI. diff --git a/examples/threads/child.js b/examples/threads/child.js new file mode 100644 index 0000000..9ff5cad --- /dev/null +++ b/examples/threads/child.js @@ -0,0 +1,6 @@ +'use strict' +const {threadId} = require('worker_threads') + +module.exports = function (inp, callback) { + callback(null, inp + ' BAR (' + [process.pid, threadId].join(':') + ')') +} \ No newline at end of file diff --git a/examples/threads/index.js b/examples/threads/index.js new file mode 100644 index 0000000..f3a9231 --- /dev/null +++ b/examples/threads/index.js @@ -0,0 +1,14 @@ +'use strict' + +// Beware! Run with node 12.x or higher, or 10.5 with --experimental-worker +let workerFarm = require('../../') + , workers = workerFarm.threaded(require.resolve('./child')) + , ret = 0 + +for (let i = 0; i < 10; i++) { + workers('#' + i + ' FOO', function(err, outp) { + console.log(outp) + if (++ret == 10) + workerFarm.end(workers) + }) +} \ No newline at end of file diff --git a/examples/transfer/child.js b/examples/transfer/child.js new file mode 100644 index 0000000..c2d8dc4 --- /dev/null +++ b/examples/transfer/child.js @@ -0,0 +1,9 @@ +'use strict'; + +module.exports = function (size, useList, callback) { + let arr = new Float64Array(size[0]) + for (let i = 0; i < arr.length; i++) { + arr[i] = Math.random() + } + callback(null, arr, useList ? [arr.buffer] : null) +} \ No newline at end of file diff --git a/examples/transfer/index.js b/examples/transfer/index.js new file mode 100644 index 0000000..0fa94c0 --- /dev/null +++ b/examples/transfer/index.js @@ -0,0 +1,40 @@ +'use strict' + +// Beware! Run with node 12.x or higher, or 10.5 with --experimental-worker +let workerFarm = require('../../') + , workers = workerFarm.threaded(require.resolve('./child')) + , ret = 0 + , n = 100 + , size = 5*1024*1024 + +;(async function() { + + console.time('With transferList') + await new Promise(function(resolve, reject) { + for (let i = 0; i < n; i++) { + let data = new Uint32Array([size]) + workers(data, true, function(err, result) { + if (++ret == n) + resolve() + }, [data.buffer]) + } + }) + console.timeEnd('With transferList') + + ret = 0 + + console.time('Without transferList') + await new Promise(function(resolve, reject) { + for (let i = 0; i < n; i++) { + let data = new Uint32Array([size]) + workers(data, false, function(err, result) { + if (++ret == n) + resolve() + }) + } + }) + console.timeEnd('Without transferList') + + workerFarm.end(workers) + +})().catch(console.error) \ No newline at end of file diff --git a/lib/child/handle.js b/lib/child/handle.js new file mode 100644 index 0000000..55ece49 --- /dev/null +++ b/lib/child/handle.js @@ -0,0 +1,46 @@ +'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({ owner: 'farm', idx: idx, child: child, args: _args.slice(0,2) }, _args[2]); + } + , 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 (data.owner !== 'farm') { + return; + } + if (!$module) return $module = require(data.module) + if (data.event == 'die') return process.exit(0) + handle(data, send) + } +} \ No newline at end of file diff --git a/lib/child/index.js b/lib/child/index.js index 78f6337..f7fad7f 100644 --- a/lib/child/index.js +++ b/lib/child/index.js @@ -1,6 +1,5 @@ 'use strict' - -let $module +const handle = require('./handle') /* let contextProto = this.context; @@ -8,49 +7,8 @@ let $module 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({ owner: 'farm', 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 (data.owner !== 'farm') { - return; - } - - if (!$module) return $module = require(data.module) - if (data.event == 'die') return process.exit(0) - handle(data) -}) +process.on('message', handle(function(data) { + // Ignore any additional arguments because process send only allows to + // send handles. + process.send(data) +})) \ No newline at end of file diff --git a/lib/child/thread.js b/lib/child/thread.js new file mode 100644 index 0000000..ca4d5c5 --- /dev/null +++ b/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))) diff --git a/lib/farm.js b/lib/farm.js index 60720dc..25b2407 100644 --- a/lib/farm.js +++ b/lib/farm.js @@ -10,6 +10,7 @@ const DEFAULT_OPTIONS = { , maxRetries : Infinity , forcedKillTime : 100 , autoStart : false + , workerThreads : false , onChild : function() {} } @@ -36,11 +37,18 @@ Farm.prototype.mkhandle = function (method) { return process.nextTick(args[args.length - 1].bind(null, err)) throw err } + let cb = args.pop() + let transferList = null + if (typeof cb !== 'function') { + transferList = cb + cb = args.pop() + } this.addCall({ - method : method - , callback : args.pop() - , args : args - , retries : 0 + method : method + , callback : cb + , args : args + , retries : 0 + , transferList : transferList }) }.bind(this) } @@ -104,7 +112,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.workerThreads) , id = this.childId , c = { send : forked.send @@ -138,8 +146,15 @@ Farm.prototype.stopChild = function (childId) { if (child) { child.send({owner: 'farm', event: 'die'}) setTimeout(function () { - if (child.exitCode === null) - child.child.kill('SIGKILL') + if (child.exitCode === null) { + + // Difference between worker_threads and child_process. + if (child.child.kill) + child.child.kill('SIGKILL') + else + child.child.terminate() + + } }, this.options.forcedKillTime).unref() ;delete this.children[childId] this.activeChildren-- @@ -242,11 +257,12 @@ Farm.prototype.send = function (childId, call) { this.activeCalls++ child.send({ - owner : 'farm' - , idx : idx - , child : childId - , method : call.method - , args : call.args + owner : 'farm' + , idx : idx + , child : childId + , method : call.method + , args : call.args + , transferList : call.transferList }) if (this.options.maxCallTime !== Infinity) { diff --git a/lib/fork.js b/lib/fork.js index 5a035d9..7d26786 100644 --- a/lib/fork.js +++ b/lib/fork.js @@ -3,8 +3,28 @@ const childProcess = require('child_process') , childModule = require.resolve('./child/index') +let checked, available +function fork (forkModule, workerOptions, workerThreads) { + + // Check whether we need to run using worker_threads. + if (workerThreads) { + + 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 12.x or higher, or 10.5 with the --experimental-worker flag!') + } + 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) @@ -24,10 +44,27 @@ function fork (forkModule, workerOptions) { // return a send() function for this child return { - send : child.send.bind(child) + send : function(call) { + delete call.transferList + child.send(call) + } , child : child } } +function thread (forkModule, workerOptions) { + const {Worker} = require('worker_threads'); + let child = new Worker(require.resolve('./child/thread'), workerOptions) + child.on('error', function() {}) + child.postMessage({ owner: 'farm', module: forkModule }) + return { + send : function(call) { + let transferList = call.transferList + ;delete call.transferList + child.postMessage(call, transferList) + } + , child : child + } +} module.exports = fork diff --git a/lib/index.js b/lib/index.js index 8c14222..117009e 100644 --- a/lib/index.js +++ b/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({ workerThreads: 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 \ No newline at end of file diff --git a/package.json b/package.json index 1c4a1b8..77671d8 100644 --- a/package.json +++ b/package.json @@ -1,7 +1,7 @@ { "name": "worker-farm", "description": "Distribute processing tasks to child processes with an über-simple API and baked-in durability & custom concurrency options.", - "version": "1.7.0", + "version": "2.0.0", "homepage": "https://github.com/rvagg/node-worker-farm", "authors": [ "Rod Vagg @rvagg (https://github.com/rvagg)" @@ -24,7 +24,7 @@ "tape": "~4.10.1" }, "scripts": { - "test": "node ./tests/" + "test": "node ./tests/run" }, "types": "./index.d.ts", "license": "MIT" diff --git a/tests/child.js b/tests/child.js index 71cb728..afc8564 100644 --- a/tests/child.js +++ b/tests/child.js @@ -3,9 +3,19 @@ const fs = require('fs') const started = Date.now() +// Check if we're running as a worker thread or child process. It doesn't make +// much sense to send along the process id for worker threads because it will +// be the same as the on in the main thread. +let pid = process.pid +try { + const {threadId, isMainThread} = require('worker_threads') + if (!isMainThread) { + pid = threadId + } +} catch (e) {} module.exports = function (timeout, callback) { - callback = callback.bind(null, null, process.pid, Math.random(), timeout) + callback = callback.bind(null, null, [pid, Math.random(), timeout]) if (timeout) return setTimeout(callback, timeout) callback() @@ -29,7 +39,7 @@ module.exports.run0 = function (callback) { module.exports.killable = function (id, callback) { if (Math.random() < 0.5) return process.exit(-1) - callback(null, id, process.pid) + callback(null, [id, pid]) } @@ -85,3 +95,19 @@ module.exports.stubborn = function (path, callback) { module.exports.uptime = function (callback) { callback(null, Date.now() - started) } + +module.exports.transfer = function (one, two, callback) { + let sum = 0 + for (let buffer of [one, two]) { + for (let value of buffer) sum += value + } + let arr = new Uint32Array([sum]) + callback(null, arr, [arr.buffer]) +} + +module.exports.shared = function (buffer, callback) { + setTimeout(function() { + buffer[0] = Math.E + callback(null) + }, 15) +} \ No newline at end of file diff --git a/tests/index.js b/tests/index.js index e6be7bb..aa6876b 100644 --- a/tests/index.js +++ b/tests/index.js @@ -1,8 +1,8 @@ 'use strict' +module.exports = function(workerFarm, threaded) { const tape = require('tape') , child_process = require('child_process') - , workerFarm = require('../') , childPath = require.resolve('./child') , fs = require('fs') , os = require('os') @@ -19,12 +19,12 @@ function uniq (ar) { // a child where module.exports = function ... tape('simple, exports=function test', function (t) { - t.plan(4) + t.plan(2) let child = workerFarm(childPath) - child(0, function (err, pid, rnd) { - t.ok(pid > process.pid, 'pid makes sense') - t.ok(pid < process.pid + 750, 'pid makes sense') + child(0, function (err, [pid, rnd]) { + // t.ok(pid > process.pid, 'pid makes sense') + // t.ok(pid < process.pid + 750, 'pid makes sense') t.ok(rnd >= 0 && rnd < 1, 'rnd result makes sense') }) @@ -36,12 +36,12 @@ tape('simple, exports=function test', function (t) { // a child where we have module.exports.fn = function ... tape('simple, exports.fn test', function (t) { - t.plan(4) + t.plan(2) let child = workerFarm(childPath, [ 'run0' ]) - child.run0(function (err, pid, rnd) { - t.ok(pid > process.pid, 'pid makes sense') - t.ok(pid < process.pid + 750, 'pid makes sense') + child.run0(function (err, [pid, rnd]) { + // t.ok(pid > process.pid, 'pid makes sense') + // t.ok(pid < process.pid + 750, 'pid makes sense') t.ok(rnd >= 0 && rnd < 1, 'rnd result makes sense') }) @@ -52,18 +52,42 @@ tape('simple, exports.fn test', function (t) { tape('on child', function (t) { + + if (threaded) { + + t.plan(2) + let child = workerFarm({ + onChild: function(worker) { + threadId = worker.threadId + } + }, childPath) + let threadId = null + + child(0, function(err, [tid]) { + t.equal(threadId, tid) + }) + + workerFarm.end(child, function() { + t.ok(true, 'workerFarm ended') + }) + + } + else { + t.plan(2) let child = workerFarm({ onChild: function(subprocess) { childPid = subprocess.pid } }, childPath) , childPid = null; - child(0, function(err, pid) { + child(0, function(err, [pid]) { t.equal(childPid, pid) }) workerFarm.end(child, function () { t.ok(true, 'workerFarm ended') }) + + } }) @@ -77,7 +101,7 @@ tape('single worker', function (t) { , i = 10 while (i--) { - child(0, function (err, pid) { + child(0, function (err, [pid]) { pids.push(pid) if (pids.length == 10) { t.equal(1, uniq(pids).length, 'only a single process (by pid)') @@ -102,7 +126,7 @@ tape('two workers', function (t) { , i = 10 while (i--) { - child(0, function (err, pid) { + child(0, function (err, [pid]) { pids.push(pid) if (pids.length == 10) { t.equal(2, uniq(pids).length, 'only two child processes (by pid)') @@ -127,7 +151,7 @@ tape('many workers', function (t) { , i = 10 while (i--) { - child(1, function (err, pid) { + child(1, function (err, [pid]) { pids.push(pid) if (pids.length == 10) { t.equal(10, uniq(pids).length, 'pids are all the same (by pid)') @@ -180,7 +204,7 @@ tape('single call per worker', function (t) { , i = count while (i--) { - child(0, function (err, pid) { + child(0, function (err, [pid]) { pids.push(pid) if (pids.length == count) { t.equal(count, uniq(pids).length, 'one process for each call (by pid)') @@ -210,7 +234,7 @@ tape('two calls per worker', function (t) { , i = count while (i--) { - child(0, function (err, pid) { + child(0, function (err, [pid]) { pids.push(pid) if (pids.length == count) { t.equal(count / 2, uniq(pids).length, 'one process for each call (by pid)') @@ -348,7 +372,7 @@ tape('durability', function (t) { , i = count while (i--) { - child.killable(i, function (err, id, pid) { + child.killable(i, function (err, [id, pid]) { ids.push(id) pids.push(pid) if (ids.length == count) { @@ -366,12 +390,12 @@ tape('durability', function (t) { // a callback provided to .end() can and will be called (uses "simple, exports=function test" to create a child) tape('simple, end callback', function (t) { - t.plan(4) + t.plan(2) let child = workerFarm(childPath) - child(0, function (err, pid, rnd) { - t.ok(pid > process.pid, 'pid makes sense ' + pid + ' vs ' + process.pid) - t.ok(pid < process.pid + 750, 'pid makes sense ' + pid + ' vs ' + process.pid) + child(0, function (err, [pid, rnd]) { + // t.ok(pid > process.pid, 'pid makes sense ' + pid + ' vs ' + process.pid) + // t.ok(pid < process.pid + 750, 'pid makes sense ' + pid + ' vs ' + process.pid) t.ok(rnd >= 0 && rnd < 1, 'rnd result makes sense') }) @@ -382,26 +406,27 @@ tape('simple, end callback', function (t) { tape('call timeout test', function (t) { - t.plan(3 + 3 + 4 + 4 + 4 + 3 + 1) + t.plan(3 + 3 + 4 + 4 + 4 + 3 + 1 - 6) let child = workerFarm({ maxCallTime: 250, maxConcurrentWorkers: 1 }, childPath) // should come back ok - child(50, function (err, pid, rnd) { - t.ok(pid > process.pid, 'pid makes sense ' + pid + ' vs ' + process.pid) - t.ok(pid < process.pid + 750, 'pid makes sense ' + pid + ' vs ' + process.pid) + child(50, function (err, [pid, rnd]) { + // t.ok(pid > process.pid, 'pid makes sense ' + pid + ' vs ' + process.pid) + // t.ok(pid < process.pid + 750, 'pid makes sense ' + pid + ' vs ' + process.pid) t.ok(rnd > 0 && rnd < 1, 'rnd result makes sense ' + rnd) }) // should come back ok - child(50, function (err, pid, rnd) { - t.ok(pid > process.pid, 'pid makes sense ' + pid + ' vs ' + process.pid) - t.ok(pid < process.pid + 750, 'pid makes sense ' + pid + ' vs ' + process.pid) + child(50, function (err, [pid, rnd]) { + // t.ok(pid > process.pid, 'pid makes sense ' + pid + ' vs ' + process.pid) + // t.ok(pid < process.pid + 750, 'pid makes sense ' + pid + ' vs ' + process.pid) t.ok(rnd > 0 && rnd < 1, 'rnd result makes sense ' + rnd) }) // should die - child(500, function (err, pid, rnd) { + child(500, function (err, arr) { + let [pid, rnd] = arr || []; t.ok(err, 'got an error') t.equal(err.type, 'TimeoutError', 'correct error type') t.ok(pid === undefined, 'no pid') @@ -409,7 +434,8 @@ tape('call timeout test', function (t) { }) // should die - child(1000, function (err, pid, rnd) { + child(1000, function (err, arr) { + let [pid, rnd] = arr || []; t.ok(err, 'got an error') t.equal(err.type, 'TimeoutError', 'correct error type') t.ok(pid === undefined, 'no pid') @@ -419,7 +445,8 @@ tape('call timeout test', function (t) { // should die even though it is only a 100ms task, it'll get caught up // in a dying worker setTimeout(function () { - child(100, function (err, pid, rnd) { + child(100, function (err, arr) { + let [pid, rnd] = arr || []; t.ok(err, 'got an error') t.equal(err.type, 'TimeoutError', 'correct error type') t.ok(pid === undefined, 'no pid') @@ -429,9 +456,9 @@ tape('call timeout test', function (t) { // should be ok, new worker setTimeout(function () { - child(50, function (err, pid, rnd) { - t.ok(pid > process.pid, 'pid makes sense ' + pid + ' vs ' + process.pid) - t.ok(pid < process.pid + 750, 'pid makes sense ' + pid + ' vs ' + process.pid) + child(50, function (err, [pid, rnd]) { + // t.ok(pid > process.pid, 'pid makes sense ' + pid + ' vs ' + process.pid) + // t.ok(pid < process.pid + 750, 'pid makes sense ' + pid + ' vs ' + process.pid) t.ok(rnd > 0 && rnd < 1, 'rnd result makes sense ' + rnd) }) workerFarm.end(child, function () { @@ -573,26 +600,37 @@ tape('test max retries after process terminate', function (t) { }) }) - tape('custom arguments can be passed to "fork"', function (t) { - t.plan(3) - // allocate a real, valid path, in any OS - let cwd = fs.realpathSync(os.tmpdir()) - , workerOptions = { - cwd : cwd - , execArgv : ['--expose-gc'] - } - , child = workerFarm({ maxConcurrentWorkers: 1, maxRetries: 5, workerOptions: workerOptions}, childPath, ['args']) + // Passing --expose-gc won't work with threads, so in that case expect an error. + function test() { + // allocate a real, valid path, in any OS + let cwd = fs.realpathSync(os.tmpdir()) + , workerOptions = { + cwd : cwd + , execArgv : ['--expose-gc'] + } + , child = workerFarm({ maxConcurrentWorkers: 1, maxRetries: 5, workerOptions: workerOptions}, childPath, ['args']) + + child.args(function (err, result) { + t.equal(result.execArgv[0], '--expose-gc', 'flags passed (overridden default)') + t.equal(result.cwd, cwd, 'correct cwd folder') + }) - child.args(function (err, result) { - t.equal(result.execArgv[0], '--expose-gc', 'flags passed (overridden default)') - t.equal(result.cwd, cwd, 'correct cwd folder') - }) + workerFarm.end(child, function () { + t.ok(true, 'workerFarm ended') + }) - workerFarm.end(child, function () { - t.ok(true, 'workerFarm ended') - }) + } + + if (threaded) { + t.plan(1) + t.throws(test) + } + else { + t.plan(3) + test() + } }) @@ -614,3 +652,54 @@ tape('ensure --debug/--inspect not propagated to children', function (t) { t.ok(stdout.indexOf('--debug') === -1, 'child does not receive debug flag') }) }) + +tape('pass a transferList when running in threaded mode', function (t) { + + // Won't work with child_processes. + if (!threaded) { + t.end() + return + } + + t.plan(5) + + let child = workerFarm(childPath, ['transfer']) + let one = new Float64Array([0,1,2]) + let two = new Uint8Array([3,4,5]) + + child.transfer(one, two, function(err, result) { + t.equal(one.byteLength, 0, 'Float64Array has been transferred') + t.equal(two.byteLength, 0, 'Uint8Array has been transferred') + t.ok(result instanceof Uint32Array, 'result should be a Uint32Array') + t.equal(result[0], 15, 'sum should be 15') + }, [one.buffer, two.buffer]) + + workerFarm.end(child, function() { + t.ok(true, 'workerFarm ended') + }) + +}) + +tape('modify a shared array buffer', function (t) { + + // Won't work with child_processes. + if (!threaded) { + t.end() + return + } + + t.plan(2) + let child = workerFarm(childPath, ['shared']) + let arr = new Float64Array(new SharedArrayBuffer(8)) + arr[0] = Math.PI + child.shared(arr, function() { + t.equal(arr[0], Math.E, 'shared array buffer should have been modified') + }) + + workerFarm.end(child, function() { + t.ok(true, 'workerFarm.end') + }) + +}) + +} \ No newline at end of file diff --git a/tests/run.js b/tests/run.js new file mode 100644 index 0000000..00cd6ff --- /dev/null +++ b/tests/run.js @@ -0,0 +1,8 @@ +const test = require('.') +const workerFarm = require('..') + +test(workerFarm) +try { + require('worker_threads'); + test(workerFarm.threaded, true) +} catch (e) {} \ No newline at end of file