Add resourceLimits support

README.md

@@ -8,6 +8,7 @@
 * ✔ Proper async tracking integration
 * ✔ Tracking statistics for run and wait times
 * ✔ Cancelation Support
+* ✔ Supports enforcing memory resource limits
 
 For Node.js 12.x and higher.
 
@@ -147,6 +148,16 @@ This class extends [`EventEmitter`][] from Node.js.
     handling I/O in parallel.
   * `useAtomics`: (`boolean`) Use the [`Atomics`][] API for faster communication
     between threads. This is on by default.
+  * `resourceLimits`: (`object`) See [Node.js new Worker options][]
+    * `maxOldGenerationSizeMb`: (`number`) The maximum size of each worker threads
+      main heap in MB.
+    * `maxYoungGenerationSizeMb`: (`number`) The maximum size of a heap space for
+      recently created objects.
+    * `codeRangeSizeMb`: (`number`) The size of a pre-allocated memory range used
+      for generated code.
+
+Use caution when setting resource limits. Setting limits that are too low may
+result in the `Piscina` worker threads being unusable.
 
 ### Method: `runTask(task[, transferList][, filename][, abortSignal])`
 
@@ -335,5 +346,6 @@ Piscina development is sponsored by [NearForm Research][].
 [`Atomics`]: https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Atomics
 [`EventEmitter`]: https://nodejs.org/api/events.html
 [`postMessage`]: https://nodejs.org/api/worker_threads.html#worker_threads_port_postmessage_value_transferlist
+[Node.js new Worker options]: https://nodejs.org/api/worker_threads.html#worker_threads_new_worker_filename_options
 [MIT Licensed]: LICENSE.md
 [NearForm Research]: https://www.nearform.com/research/

examples/resourceLimits/index.js

@@ -0,0 +1,23 @@
+'use strict';
+
+const Piscina = require('../..');
+const { resolve } = require('path');
+const { strictEqual } = require('assert');
+
+const piscina = new Piscina({
+  filename: resolve(__dirname, 'worker.js'),
+  resourceLimits: {
+    maxOldGenerationSizeMb: 16,
+    maxYoungGenerationSizeMb: 4,
+    codeRangeSizeMb: 16
+  }
+});
+
+(async function() {
+  try {
+    await piscina.runTask();
+  } catch (err) {
+    console.log('Worker terminated due to resource limits');
+    strictEqual(err.code, 'ERR_WORKER_OUT_OF_MEMORY');
+  }
+})();

examples/resourceLimits/worker.js

@@ -0,0 +1,8 @@
+'use strict';
+
+module.exports = () => {
+  const array = [];
+  while (true) {
+    array.push([array]);
+  }
+};

src/index.ts

@@ -43,15 +43,20 @@ class AbortError extends Error {
   }
 }
 
+type ResourceLimits = Worker extends {
+  resourceLimits? : infer T;
+} ? T : {};
+
 interface Options {
   // Probably also support URL here
   filename? : string | null,
   minThreads? : number,
   maxThreads? : number,
   idleTimeout? : number,
   maxQueue? : number,
-  concurrentTasksPerWorker? : number
-  useAtomics? : boolean
+  concurrentTasksPerWorker? : number,
+  useAtomics? : boolean,
+  resourceLimits? : ResourceLimits
 }
 
 interface FilledOptions extends Options {
@@ -279,7 +284,9 @@ class ThreadPool {
 
   _addNewWorker () : WorkerInfo {
     const pool = this;
-    const worker = new Worker(resolve(__dirname, 'worker.js'));
+    const worker = new Worker(resolve(__dirname, 'worker.js'), {
+      resourceLimits: this.options.resourceLimits
+    });
 
     const { port1, port2 } = new MessageChannel();
     const workerInfo = new WorkerInfo(worker, port1, onMessage);
@@ -542,6 +549,11 @@ class Piscina extends EventEmitter {
         typeof options.useAtomics !== 'boolean') {
       throw new TypeError('options.useAtomics must be a boolean value');
     }
+    if (options.resourceLimits !== undefined &&
+        (typeof options.resourceLimits !== 'object' ||
+         options.resourceLimits === null)) {
+      throw new TypeError('options.resourceLimits must be an object');
+    }
 
     this.#pool = new ThreadPool(this, options);
   }

test/fixtures/resource-limits.js

@@ -0,0 +1,8 @@
+'use strict';
+
+module.exports = () => {
+  const array = [];
+  while (true) {
+    array.push([array]);
+  }
+};

test/option-validation.ts

@@ -75,3 +75,9 @@ test('useAtomics must be a boolean', async ({ throws }) => {
     useAtomics: 'string'
   }) as any), /options.useAtomics must be a boolean/);
 });
+
+test('resourceLimits must be an object', async ({ throws }) => {
+  throws(() => new Piscina(({
+    resourceLimits: 0
+  }) as any), /options.resourceLimits must be an object/);
+});

test/test-resourcelimits.ts

@@ -0,0 +1,34 @@
+import Piscina from '..';
+import { test } from 'tap';
+import { resolve } from 'path';
+
+test('resourceLimits causes task to reject', async ({ is, rejects }) => {
+  const worker = new Piscina({
+    filename: resolve(__dirname, 'fixtures/resource-limits.js'),
+    resourceLimits: {
+      maxOldGenerationSizeMb: 16,
+      maxYoungGenerationSizeMb: 4,
+      codeRangeSizeMb: 16
+    }
+  });
+  worker.on('error', () => {
+    // Ignore any additional errors that may occur.
+    // This may happen because when the Worker is
+    // killed a new worker is created that may hit
+    // the memory limits immediately. When that
+    // happens, there is no associated Promise to
+    // reject so we emit an error event instead.
+    // We don't care so much about that here. We
+    // could potentially avoid the issue by setting
+    // higher limits above but rather than try to
+    // guess at limits that may work consistently,
+    // let's just ignore the additional error for
+    // now.
+  });
+  const limits : any = worker.options.resourceLimits;
+  is(limits.maxOldGenerationSizeMb, 16);
+  is(limits.maxYoungGenerationSizeMb, 4);
+  is(limits.codeRangeSizeMb, 16);
+  rejects(worker.runTask(null),
+    /Worker terminated due to reaching memory limit: JS heap out of memory/);
+});