Let the user choose worker implementation

Readme.md

@@ -105,6 +105,7 @@ Describes a WorkerNodes options.
     * [.workerEndurance](#WorkerNodesOptions+workerEndurance) : <code>Number</code>
     * [.workerStopTimeout](#WorkerNodesOptions+workerStopTimeout) : <code>Number</code>
     * [.resourceLimits](#WorkerNodesOptions+resourceLimits) : <code>Object</code>
+    * [.workerType](#WorkerNodesOptions+workerType) : <code>string</code>
 
 <a name="WorkerNodesOptions+autoStart"></a>
 
@@ -195,7 +196,7 @@ The timeout value (in milliseconds) for the worker to stop before sending SIGKIL
 <a name="WorkerNodesOptions+resourceLimits"></a>
 
 ### options.resourceLimits : <code>Object</code>
-Provides the set of JS engine resource constraints inside this Worker thread.
+Provides the set of JS engine resource constraints inside this Worker thread. (Usable when using `workerType: thread` only)
 
 **Kind**: instance property of [<code>WorkerNodesOptions</code>](#WorkerNodesOptions)  
 **Properties**
@@ -207,6 +208,10 @@ Provides the set of JS engine resource constraints inside this Worker thread.
 | codeRangeSizeMb | <code>Number</code> | The size of a pre-allocated memory range used for generated code |
 | stackSizeMb | <code>Number</code> | The default maximum stack size for the thread. Small values may lead to unusable Worker instances |
 
+### options.workerType : <code>string</code>
+Can be either `process` or `thread` (default), that controls the underlying implementation used, either `child_process` or `worker_threads`.
+Most usecases are perfectly fine with `thread` implementation, some work loads though, might need to use `process`, for example, if you are using
+`process.chdir()` call which is [not supported](https://github.com/nodejs/node/issues/41673) in `worker_threads`.
 
 ## Example
 

e2e/async-initialization-process.spec.js

@@ -0,0 +1,3 @@
+const testBase = require("./async-initialization.base");
+
+testBase("process");

e2e/async-initialization-thread.spec.js

@@ -0,0 +1,3 @@
+const testBase = require("./async-initialization.base");
+
+testBase("thread");

e2e/async-initialization.base.js

@@ -0,0 +1,35 @@
+const test = require('ava');
+
+const WorkerNodes = require('../');
+const { fixture } = require('./utils');
+
+module.exports = function describe(workerType) {
+    test(`should not mark worker as ready until module fully initialized`, t => {
+        // given
+        const workerNodes = new WorkerNodes(fixture('async-initialization'), {
+            maxWorkers: 1,
+            asyncWorkerInitialization: true,
+            autoStart: true,
+            workerType
+        });
+
+        // then
+        t.falsy(workerNodes.pickWorker());
+    });
+
+    test(`should correctly handle task after initialization`, async t => {
+        // given
+        const workerNodes = new WorkerNodes(fixture('async-initialization'), {
+            maxWorkers: 1,
+            asyncWorkerInitialization: true,
+            autoStart: true,
+            workerType
+        });
+
+        // when 
+        const result = await workerNodes.call.result();
+
+        // then
+        t.is(result, 'result');
+    });    
+}

e2e/auto-start-process.spec.js

@@ -0,0 +1,3 @@
+const testBase = require("./auto-start.base");
+
+testBase("process");

e2e/auto-start-thread.spec.js

@@ -0,0 +1,3 @@
+const testBase = require("./auto-start.base");
+
+testBase("thread");

e2e/auto-start.base.js

@@ -0,0 +1,76 @@
+const test = require('ava');
+
+const WorkerNodes = require('../');
+const { fixture, unique, repeatCall, eventually } = require('./utils');
+
+module.exports = function describe(workerType) {
+    test(`should be disabled by default`, async t => {
+        // given
+        const workerNodes = new WorkerNodes(fixture('process-info'), { workerType });
+
+        // when
+        const callTime = new Date();
+        await workerNodes.call.noop();
+
+        // then
+        workerNodes.workersQueue.forEach(worker => {
+            const startDate = worker.process.startDate;
+
+            if (startDate) {
+                t.true(startDate >= callTime);
+            }
+        });
+    });
+
+    test(`should result in spawn of the workers before the first call if active`, async t => {
+        // given
+        const workerNodes = new WorkerNodes(fixture('process-info'), { autoStart: true, minWorkers: 1, maxWorkers: 1, workerType });
+        await workerNodes.ready();
+
+        // when
+        const callTime = Date.now();
+        const workerStartTime = await workerNodes.call.getStartTime();
+
+        // then
+        t.true(workerStartTime <= callTime);
+    });
+
+    test(`should force the workerNodes to wait for all the required workers to start before reporting ready`, async t => {
+        // given
+        const workerNodes = new WorkerNodes(fixture('process-info'), { autoStart: true, minWorkers: 4, maxWorkers: 4, workerType });
+        await workerNodes.ready();
+        const callStartTime = Date.now();
+
+        // when
+        const results = await repeatCall(workerNodes.call.getStartTime, 4);
+
+        // then
+        t.is(results.length, 4);
+
+        results.forEach(result => t.true(result <= callStartTime));
+    });
+
+    test(`should only use workers that are fully initialized`, async t => {
+        // given
+        const workerNodes = new WorkerNodes(fixture('slow-module'), {
+            autoStart: true,
+            minWorkers: 2,
+            maxWorkers: 2,
+            taskMaxRetries: Infinity,
+            workerType
+        });
+        await workerNodes.ready();
+
+        await repeatCall(workerNodes.call.getPid, 4);
+
+        // when
+        workerNodes.workersQueue.storage[0].process.exit();
+        await repeatCall(workerNodes.call.getPid, 4);
+
+        // then
+        // we wait for all workers to come back a live
+        const getLiveWorkers = () => workerNodes.workersQueue.filter(worker => worker.isProcessAlive);
+        await eventually(() => unique(getLiveWorkers()).length === 2);
+        t.is(unique(getLiveWorkers()).length, 2);
+    });
+}

e2e/call-timeout-process.spec.js

@@ -0,0 +1,3 @@
+const testBase = require("./call-timeout.base");
+
+testBase("process");

e2e/call-timeout-thread.spec.js

@@ -0,0 +1,3 @@
+const testBase = require("./call-timeout.base");
+
+testBase("thread");

e2e/call-timeout.base.js

@@ -0,0 +1,92 @@
+const test = require('ava');
+
+const WorkerNodes = require('../');
+const errors = require('../lib/errors');
+const { fixture, wait, eventually } = require('./utils');
+
+module.exports = function describe(workerType) {
+    test(`should not affect fast method calls`, async t => {
+        // given
+        const workerNodes = new WorkerNodes(fixture('async-tasks'), { taskTimeout: 500, maxWorkers: 1, workerType });
+        await workerNodes.ready();
+
+        // then
+        await t.notThrowsAsync(workerNodes.call.task100ms);
+    });
+
+    test(`should result in an error when a single method call takes too long`, async t => {
+        // given
+        const workerNodes = new WorkerNodes(fixture('async-tasks'), { taskTimeout: 250, maxWorkers: 1, workerType });
+        await workerNodes.ready();
+
+        // then
+        await t.throwsAsync(workerNodes.call.task500ms, { instanceOf: errors.TimeoutError });
+    });
+
+    test(`should kill the worker that was involved in processing the task`, async t => {
+        // given
+        const workerNodes = new WorkerNodes(fixture('async-tasks'), { taskTimeout: 250, maxWorkers: 1, workerType });
+        await workerNodes.ready();
+
+        // when
+        let executingWorkerId;
+        await t.throwsAsync(async () => {
+            const p = workerNodes.call.task500ms();
+            workerNodes.workersQueue.storage.forEach(worker => executingWorkerId = worker.id);
+            await p;
+        }, { instanceOf: errors.TimeoutError });
+        await eventually(() => workerNodes.workersQueue.storage.filter(worker => worker.id === executingWorkerId).length === 0);
+
+        // then
+        t.is(workerNodes.workersQueue.storage.filter(worker => worker.id === executingWorkerId).length, 0);
+    });
+
+    test(`should result with rejection of all the calls that the worker was processing at the moment`, async t => {
+        // given
+        const workerNodes = new WorkerNodes(fixture('async-tasks'), {
+            autoStart: true,
+            minWorkers: 1,
+            maxWorkers: 1,
+            maxTasksPerWorker: 2,
+            taskTimeout: 250,
+            workerType
+        });
+        await workerNodes.ready();
+
+        // when
+
+        const failingCall = workerNodes.call.task500ms().catch(error => error);
+        await wait(200);
+
+        const secondCall = workerNodes.call.task100ms().catch(error => error);
+        const results = await Promise.all([failingCall, secondCall]);
+
+        // then
+        results.forEach(result => {
+            t.true(result instanceof errors.TimeoutError);
+        });
+    });
+
+    test(`should result in the spawn of a new worker`, async t => {
+        // given
+        const workerNodes = new WorkerNodes(fixture('async-tasks'), {
+            maxWorkers: 1,
+            maxTasksPerWorker: Infinity,
+            taskTimeout: 250,
+            workerType
+        });
+        await workerNodes.ready();
+
+        // when
+        await workerNodes.call.noop();
+        const idBefore = workerNodes.workersQueue.storage[0].id;
+        const callResult = await workerNodes.call.task500ms().catch(error => error);
+        await workerNodes.call.noop();
+        const idAfter = workerNodes.workersQueue.storage[0].id;
+
+        // then
+        t.true(callResult instanceof errors.TimeoutError);
+        t.not(idBefore, idAfter);
+        t.is(typeof idBefore, 'number');
+    });
+}