piscina - the node.js worker pool

• ✔ 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 20.x and higher.

[MIT Licensed][].

Documentation

• Website

• piscina - the node.js worker pool

  • Documentation
  • Piscina API
    • Example
    • Exporting multiple worker functions
    • Cancelable Tasks
    • Delaying Availability of Workers
    • Backpressure
    • Out of scope asynchronous code
    • Broadcast a message to all worker threads
    • Additional Examples
  • Class: Piscina
    • Constructor: new Piscina([options])
    • Method: run(task[, options])
    • Method: destroy()
    • Method: close([options])
    • Event: 'error'
    • Event: 'drain'
    • Event: 'needsDrain'
    • Event: 'message'
    • Property: completed (readonly)
    • Property: duration (readonly)
    • Property: options (readonly)
    • Property: runTime (readonly)
    • Property: threads (readonly)
    • Property: queueSize (readonly)
    • Property: needsDrain (readonly)
    • Property: utilization (readonly)
    • Property: waitTime (readonly)
    • Static property: isWorkerThread (readonly)
    • Static property: version (readonly)
    • Static method: move(value)
      • Interface: Transferable
  • Custom Task Queues
    • Built-In Queues
      • Using FixedQueue Example
  • Current Limitations (Things we're working on / would love help with)
  • Performance Notes
    • Queue Size
    • Queue Pressure and Idle Threads
    • Thread priority on Linux systems
    • Multiple Thread Pools and Embedding Piscina as a Dependency
  • The Team
  • Acknowledgements
  • Sponsors
    • Bronze Sponsors
  • Resources

Piscina API

Example

In main.js:

const path = require("path");
const Piscina = require("piscina");

const piscina = new Piscina({
  filename: path.resolve(__dirname, "worker.js"),
});

(async function () {
  const result = await piscina.run({ a: 4, b: 6 });
  console.log(result); // Prints 10
})();

In worker.js:

module.exports = ({ a, b }) => {
  return a + b;
};

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

const { setTimeout } = require("timers/promises");

module.exports = async ({ a, b }) => {
  // Fake some async activity
  await setTimeout(100);
  return a + b;
};

ESM is also supported for both Piscina and workers:

import { Piscina } from "piscina";

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

const result = await piscina.run({ a: 4, b: 6 });
console.log(result); // Prints 10

In worker.mjs:

export default ({ a, b }) => {
  return a + b;
};

Exporting multiple worker functions

A single worker file may export multiple named handler functions.

"use strict";

function add({ a, b }) {
  return a + b;
}

function multiply({ a, b }) {
  return a * b;
}

add.add = add;
add.multiply = multiply;

module.exports = add;

The export to target can then be specified when the task is submitted:

"use strict";

const Piscina = require("piscina");
const { resolve } = require("path");

const piscina = new Piscina({
  filename: resolve(__dirname, "worker.js"),
});

(async function () {
  const res = await Promise.all([
    piscina.run({ a: 4, b: 6 }, { name: "add" }),
    piscina.run({ a: 4, b: 6 }, { name: "multiply" }),
  ]);
})();

Cancelable Tasks

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

"use strict";

const Piscina = require("piscina");
const { resolve } = require("path");

const piscina = new Piscina({
  filename: resolve(__dirname, "worker.js"),
});

(async function () {
  const abortController = new AbortController();
  try {
    const { signal } = abortController;
    const task = piscina.run({ a: 4, b: 6 }, { signal });
    abortController.abort();
    await task;
  } catch (err) {
    console.log("The task was canceled");
  }
})();

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

"use strict";

const Piscina = require("piscina");
const EventEmitter = require("events");
const { resolve } = require("path");

const piscina = new Piscina({
  filename: resolve(__dirname, "worker.js"),
});

(async function () {
  const ee = new EventEmitter();
  try {
    const task = piscina.run({ a: 4, b: 6 }, { signal: ee });
    ee.emit("abort");
    await task;
  } catch (err) {
    console.log("The task was canceled");
  }
})();

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:

async function initialize() {
  await someAsyncInitializationActivity();
  return ({ a, b }) => a + b;
}

module.exports = initialize();

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:

"use strict";

const { resolve } = require("path");
const Pool = require("../..");

const pool = new Pool({
  filename: resolve(__dirname, "worker.js"),
  maxQueue: "auto",
});

const stream = getStreamSomehow();
stream.setEncoding("utf8");

pool.on("drain", () => {
  if (stream.isPaused()) {
    console.log("resuming...", counter, pool.queueSize);
    stream.resume();
  }
});

stream
  .on("data", (data) => {
    pool.run(data);
    if (pool.queueSize === pool.options.maxQueue) {
      console.log("pausing...", counter, pool.queueSize);
      stream.pause();
    }
  })
  .on("error", console.error)
  .on("end", () => {
    console.log("done");
  });

Out of scope asynchronous code

A worker thread is only active until the moment it returns a result, it can be a result of a synchronous call or a Promise that will be fulfilled/rejected in the future. Once this is done, Piscina will wait for stdout and stderr to be flushed, and then pause the worker's event-loop until the next call. If async code is scheduled without being awaited before returning since Piscina has no way of detecting this, that code execution will be resumed on the next call. Thus, it is highly recommended to properly handle all async tasks before returning a result as it could make your code unpredictable.

For example:

const { setTimeout } = require("timers/promises");

module.exports = ({ a, b }) => {
  // This promise should be awaited
  setTimeout(1000).then(() => {
    console.log("Working"); // This will **not** run during the same worker call
  });

  return a + b;
};

Broadcast a message to all worker threads

Piscina supports broadcast communication via BroadcastChannel(Node v18+). Here is an example, the main thread sends a message, and other threads the receive message.

In main.js

"use strict";

const { BroadcastChannel } = require("worker_threads");
const { resolve } = require("path");

const Piscina = require("piscina");
const piscina = new Piscina({
  filename: resolve(__dirname, "worker.js"),
  atomics: "disabled",
});

async function main() {
  const bc = new BroadcastChannel("my_channel");
  // start worker
  Promise.all([piscina.run("thread 1"), piscina.run("thread 2")]);
  // post message in one second
  setTimeout(() => {
    bc.postMessage("Main thread message");
  }, 1000);
}

main();

In worker.js

"use stric