Tinypool

A minimal and tiny Node.js Worker Thread Pool implementation (38KB)

Worker thread
tinylibs/tinypooltinypool

README

piscina - the node.js worker pool


CI

✔ Fast communication between threads
✔ Covers both fixed-task and variable-task scenarios
✔ Supports flexible pool sizes
✔ Proper async tracking integration
✔ Tracking statistics for run and wait times
✔ Cancellation Support
✔ Supports enforcing memory resource limits
✔ Supports CommonJS, ESM, and TypeScript
✔ Custom task queues
✔ Optional CPU scheduling priorities on Linux

Written in TypeScript.

For Node.js 12.x and higher.

Piscina API


Example


In main.js:

  1. ```js
  2. const Piscina = require('piscina');

  4. const piscina = new Piscina({
  5.   filename: path.resolve(__dirname, 'worker.js')
  6. });

  8. (async function() {
  9.   const result = await piscina.runTask({ a: 4, b: 6 });
  10.   console.log(result);  // Prints 10
  11. })();
  12. ```

In worker.js:

  1. ```js
  2. module.exports = ({ a, b }) => {
  3.   return a + b;
  4. };
  5. ```

The worker may also be an async function or may return a Promise:

  1. ```js
  2. const { promisify } = require('util');
  3. const sleep = promisify(setTimeout);

  5. module.exports = async ({ a, b } => {
  6.   // Fake some async activity
  7.   await sleep(100);
  8.   return a + b;
  9. })
  10. ```

ESM is also supported for both Piscina and workers:

  1. ```js
  2. import { Piscina } from 'piscina';

  4. const piscina = new Piscina({
  5.   // The URL must be a file:// URL
  6.   filename: new URL('./worker.mjs', import.meta.url).href
  7. });

  9. (async function () {
  10.   const result = await piscina.runTask({ a: 4, b: 6 });
  11.   console.log(result); // Prints 10
  12. })();
  13. ```

In worker.mjs:

  1. ```js
  2. export default ({ a, b }) => {
  3.   return a + b;
  4. };
  5. ```

Cancelable Tasks


Submitted tasks may be canceled using either an AbortController or
an EventEmitter:

  1. ```js
  2. 'use strict';

  4. const Piscina = require('piscina');
  5. const { AbortController } = require('abort-controller');
  6. const { resolve } = require('path');

  8. const piscina = new Piscina({
  9.   filename: resolve(__dirname, 'worker.js')
  10. });

  12. (async function() {
  13.   const abortController = new AbortController();
  14.   try {
  15.     const task = piscina.runTask({ a: 4, b: 6 }, abortController.signal);
  16.     abortController.abort();
  17.     await task;
  18.   } catch (err) {
  19.     console.log('The task was canceled');
  20.   }
  21. })();
  22. ```

To use AbortController, you will need to npm i abort-controller
(or yarn add abort-controller).

Alternatively, any EventEmitter that emits an 'abort' event
may be used as an abort controller:

  1. ```js
  2. 'use strict';

  4. const Piscina = require('piscina');
  5. const EventEmitter = require('events');
  6. const { resolve } = require('path');

  8. const piscina = new Piscina({
  9.   filename: resolve(__dirname, 'worker.js')
  10. });

  12. (async function() {
  13.   const ee = new EventEmitter();
  14.   try {
  15.     const task = piscina.runTask({ a: 4, b: 6 }, ee);
  16.     ee.emit('abort');
  17.     await task;
  18.   } catch (err) {
  19.     console.log('The task was canceled');
  20.   }
  21. })();
  22. ```

Delaying Availability of Workers


A worker thread will not be made available to process tasks until Piscina
determines that it is "ready". By default, a worker is ready as soon as
Piscina loads it and acquires a reference to the exported handler function.

There may be times when the availability of a worker may need to be delayed
longer while the worker initializes any resources it may need to operate.
To support this case, the worker module may export a Promise that resolves
the handler function as opposed to exporting the function directly:

  1. ```js
  2. async function initialize() {
  3.   await someAsyncInitializationActivity();
  4.   return ({ a, b }) => a + b;
  5. }

  7. module.exports = initialize();
  8. ```

Piscina will await the resolution of the exported Promise before marking
the worker thread available.

Backpressure


When the maxQueue option is set, once the Piscina queue is full, no
additional tasks may be submitted until the queue size falls below the
limit. The 'drain' event may be used to receive notification when the
queue is empty and all tasks have been submitted to workers for processing.

Example: Using a Node.js stream to feed a Piscina worker pool:
  1. ```js
  2. 'use strict';

  4. const { resolve } = require('path');
  5. const Pool = require('../..');

  7. const pool = new Pool({
  8.   filename: resolve(__dirname, 'worker.js'),
  9.   maxQueue: 'auto'
  10. });

  12. const stream = getStreamSomehow();
  13. stream.setEncoding('utf8');

  15. pool.on('drain', () => {
  16.   if (stream.isPaused()) {
  17.     console.log('resuming...', counter, pool.queueSize);
  18.     stream.resume();
  19.   }
  20. });

  22. stream
  23.   .on('data', (data) => {
  24.     pool.runTask(data);
  25.     if (pool.queueSize === pool.options.maxQueue) {
  26.       console.log('pausing...', counter, pool.queueSize);
  27.       stream.pause();
  28.     }
  29.   })
  30.   .on('error', console.error)
  31.   .on('end', () => {
  32.     console.log('done');
  33.   });
  34. ```

Additional Examples


Additional examples can be found in the GitHub repo at
https://github.com/jasnell/piscina/tree/master/examples

Class: Piscina


Piscina works by creating a pool of Node.js Worker Threads to which
one or more tasks may be dispatched. Each worker thread executes a
single exported function defined in a separate file. Whenever a
task is dispatched to a worker, the worker invokes the exported
function and reports the return value back to Piscina when the
function completes.

This class extends [EventEmitter][] from Node.js.