repository·current·Indexed 26 days ago
https://github.com/piscinajs/piscinaA fast, efficient Node.js Worker Thread Pool implementation designed to handle fixed and variable task scenarios. It features flexible pool sizing, efficient thread communication, and support for ECMAScript Modules (ESM). Key capabilities include task cancellation via AbortController, backpressure management with maxQueue, custom task queues like FixedQueue, and detailed performance monitoring metrics such as utilization and execution time histograms.
Piscina provides several advanced capabilities for managing worker threads:
To install the latest version of Piscina.js and its dependencies, use your preferred package manager. Note that you must have Node.js version 22 or higher installed on your system.
:::note While Bun can be used to install Piscina, its behavior while running within Bun is not assured. :::
npm install piscina
# or
yarn add piscina
# or
pnpm add piscina
# or
bun add piscinaOn Linux systems, you can set the CPU scheduling priority of workers using the niceIncrement option. This requires the @napi-rs/nice native addon.
npm i @napi-rs/niceniceIncrement to the Piscina constructor.A higher niceIncrement results in lower CPU priority for workers, which helps prevent them from starving the main Node.js event loop thread.
const Piscina = require("piscina");
const pool = new Piscina({
worker: "/absolute/path/to/worker.js",
niceIncrement: 20,
});By default, Piscina makes a worker available as soon as it loads the exported handler. If your worker requires asynchronous initialization (e.g., connecting to a database), you can export a Promise that resolves to the handler function. Piscina will await this Promise before marking the worker as ready to process tasks.
async function initialize() {
await someAsyncInitializationActivity();
return ({ a, b }) => a + b;
}
module.exports = initialize();To use ESM with Piscina, ensure the filename provided to the Piscina constructor is a valid file:// URL. In your worker module, use export default or named exports.
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 });If the rate of incoming tasks is inconsistent, Piscina may experience 'thrashing'—the excessive creation and termination of workers. To prevent this, use one or a combination of the following strategies:
idleTimeout: Increase the time a worker stays alive while idle before being terminated. This prevents immediate shutdown when a brief lull in tasks occurs.minThreads: Increase the minimum number of threads that Piscina always maintains. This ensures a baseline of workers is always available, even during low pressure.You can combine your main application and worker code into a single file by using the isMainThread flag from worker_threads. If isMainThread is true, you initialize Piscina using __filename. Otherwise, you export the worker function as the default export.
import Piscina from 'piscina';
import { isMainThread } from 'worker_threads';
interface Inputs {
a: number;
b: number;
}
if (isMainThread) {
const piscina = new Piscina({ filename: __filename });
(async () => {
const task: Inputs = { a: 1, b: 2 };
console.log(await piscina.run(task));
})();
} else {
export default ({ a, b }: Inputs): number => {
return a + b;
};
}When using the maxQueue option, you can prevent memory exhaustion by implementing backpressure. When pool.queueSize reaches pool.options.maxQueue, you should pause your data source (e.g., a Node.js stream). Listen for the 'drain' event on the Piscina pool instance to know when the queue has been cleared and it is safe to resume the data source.
const Pool = require("piscina");
const { resolve } = require("path");
const pool = new Pool({
filename: resolve(__dirname, "worker.js"),
maxQueue: "auto",
});
const stream = getStreamSomehow();
// Resume stream when queue is empty
pool.on("drain", () => {
if (stream.isPaused()) {
stream.resume();
}
});
stream
.on("data", (data) => {
pool.run(data);
// Pause stream if queue is full
if (pool.queueSize === pool.options.maxQueue) {
stream.pause();
}
})
.on("error", console.error)
.on("end", () => {
console.log("done");
});To use Piscina in a JavaScript project, instantiate Piscina by providing the absolute path to your worker file via the filename option. You can then execute tasks using the .run() method, which returns a Promise that resolves to the worker's result.
const path = require("path");
const Piscina = require("piscina");
// Create a new Piscina instance pointing to your worker file
const piscina = new Piscina({
filename: path.resolve(__dirname, "worker.js"),
});
// Run a task using Piscina
(async () => {
const result = await piscina.run({ a: 4, b: 6 });
console.log(result); // prints 10
})();To generate static content for the documentation website, run npm run build. The output will be placed in the build directory and can be hosted on any static content hosting service.
$ npm run build