# Technical Documentation: piscinajs/piscina > ℹ️ **Provenance:** Multi-Tier Ingestion · [CodeWiki Reference](https://codewiki.google/github.com/piscinajs/piscina) ### Overview (`README.md`) # 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 24.x and higher. [MIT Licensed][]. ## Documentation - [Website](https://piscinajs.github.io/piscina/) - [piscina - the node.js worker pool](#piscina---the-nodejs-worker-pool) - [Documentation](#documentation) - [Piscina API](#piscina-api) - [Example](#example) - [Exporting multiple worker functions](#exporting-multiple-worker-functions) - [Cancelable Tasks](#cancelable-tasks) - [Delaying Availability of Workers](#delaying-availability-of-workers) - [Backpressure](#backpressure) - [Out of scope asynchronous code](#out-of-scope-asynchronous-code) - [Broadcast a message to all worker threads](#broadcast-a-message-to-all-worker-threads) - [Additional Examples](#additional-examples) - [Class: `Piscina`](#class-piscina) - [Constructor: `new Piscina([options])`](#constructor-new-piscinaoptions) - [Method: `run(task[, options])`](#method-runtask-options) - [Method: `destroy()`](#method-destroy) - [Method: `close([options])`](#method-closeoptions) - [Event: `'error'`](#event-error) - [Event: `'drain'`](#event-drain) - [Event: `'needsDrain'`](#event-needsdrain) - [Event: `'message'`](#event-message) - [Property: `completed` (readonly)](#property-completed-readonly) - [Property: `duration` (readonly)](#property-duration-readonly) - [Property: `options` (readonly)](#property-options-readonly) - [Property: `runTime` (readonly)](#property-runtime-readonly) - [Property: `threads` (readonly)](#property-threads-readonly) - [Property: `idleThreads` (readonly)](#property-idlethreads-readonly) - [Property: `queueSize` (readonly)](#property-queuesize-readonly) - [Property: `needsDrain` (readonly)](#property-needsdrain-readonly) - [Property: `utilization` (readonly)](#property-utilization-readonly) - [Property: `waitTime` (readonly)](#property-waittime-readonly) - [Static property: `isWorkerThread` (readonly)](#static-property-isworkerthread-readonly) - [Static property: `version` (readonly)](#static-property-version-readonly) - [Static method: `move(value)`](#static-method-movevalue) - [Interface: `Transferable`](#interface-transferable) - [Custom Task Queues](#custom-task-queues) - [Built-In Queues](#built-in-queues) - [Using FixedQueue Example](#using-fixedqueue-example) - [Current Limitations (Things we're working on / would love help with)](#current-limitations-things-were-working-on--would-love-help-with) - [Performance Notes](#performance-notes) - [Queue Size](#queue-size) - [Queue Pressure and Idle Threads](#queue-pressure-and-idle-threads) - [Thread priority on Linux systems](#thread-priority-on-linux-systems) - [Multiple Thread Pools and Embedding Piscina as a Dependency](#multiple-thread-pools-and-embedding-piscina-as-a-dependency) - [The Team](#the-team) - [Acknowledgements](#acknowledgements) - [Sponsors](#sponsors) - [Bronze Sponsors](#bronze-sponsors) - [Resources](#resources) ## Piscina API ### Example In `main.js`: ```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`: ```js module.exports = ({ a, b }) => { return a + b; }; ``` The worker may also be an async function or may return a Promise: ```js 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: ```js 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`: ```js export default ({ a, b }) => { return a + b; }; ``` ### Exporting multiple worker functions A single worker file may export multiple named handler functions. ```js "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: ```js "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`: ```js "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: ```js "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: ```js 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: ```js "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: ```js 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` ```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` ```js "use strict"; const { BroadcastChannel } = require("worker_threads"); module.exports = async (thread) => { const bc = new BroadcastChannel("my_channel"); bc.onmessage = (event) => { console.log(thread + " Received from:" + event.data); }; await new Promise((resolve) => { setTimeout(resolve, 2000); }); }; ``` ### Additional Examples Additional examples can be found in the GitHub repo at https://github.com/piscinajs/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. ### Constructor: `new Piscina([options])` - The following optional configuration is supported: - `filename`: (`string | null`) Provides the default source for the code that runs the tasks on Worker threads. This should be an absolute path or an absolute `file://` URL to a file that exports a JavaScript `function` or `async function` as its default export or `module.exports`. [ES modules][] are supported. - `name`: (`string | null`) Provides the name of the default exported worker function. The default is `'default'`, indicating the default export of the worker module. - `minThreads`: (`number`) Sets the minimum number of threads that are always running for this thread pool. The default is the number provided by [`os.availableParallelism`](https://nodejs.org/api/os.html#osavailableparallelism). - `maxThreads`: (`number`) Sets the maximum number of threads that are running for this thread pool. The default is the number provided by [`os.availableParallelism`](https://nodejs.org/api/os.html#osavailableparallelism) \* 1.5. - `idleTimeout`: (`number`) A timeout in milliseconds that specifies how long a `Worker` is allowed to be idle, i.e. not handling any tasks, before it is shut down. By default, this is immediate. If `Infinity` is passed as the value, the `Worker` never shuts down. Be careful when using `Infinity`, as it can lead to resource overuse. **Tip**: _The default `idleTimeout` can lead to some performance loss in the application because of the overhead involved with stopping and starting new worker threads. To improve performance, try setting the `idleTimeout` explicitly._ - `maxQueue`: (`number` | `string`) The maximum number of tasks that may be scheduled to run, but not yet running due to lack of available threads, at a given time. By default, there is no limit. The special value `'auto'` may be used to have Piscina calculate the maximum as the square of `maxThreads`. When `'auto'` is used, the calculated `maxQueue` value may be found by checking the [`options.maxQueue`](#property-options-readonly) property. - `concurrentTasksPerWorker`: (`number`) Specifies how many tasks can share a single Worker thread simultaneously. The default is `1`. This generally only makes sense to specify if there is some kind of asynchronous component to the task. Keep in mind that Worker threads are generally not built for handling I/O in parallel. - `atomics`: (`sync` | `async` | `disabled`) Use the [`Atomics`](https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Atomics) API for faster communication between threads. This is on by default. You can disable `Atomics` globally by setting the environment variable `PISCINA_DISABLE_ATOMICS` to `1` . If `atomics` is `sync`, it will cause to pause threads (stopping all execution) between tasks. Ideally, threads should wait for all operations to finish before returning control to the main thread (avoid having open handles within a thread). If still want to have the possibility of having open handles or handle asynchronous tasks, you can set the environment variable `PISCINA_ENABLE_ASYNC_ATOMICS` to `1` or setting `options.atomics` to `async`. > **Note**: The `async` mode comes with performance penalties and can lead to undesired behaviour if open handles are not tracked correctly. - `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. - `stackSizeMb` : (`number`) The default maximum stack size for the thread. Small values may lead to unusable Worker instances. Default: 4 - `env`: (`object`) If set, specifies the initial value of `process.env` inside the worker threads. See [Node.js new Worker options][] for details. - `argv`: (`any[]`) List of arguments that will be stringified and appended to `process.argv` in the worker. See [Node.js new Worker options][] for details. - `execArgv`: (`string[]`) List of Node.js CLI options passed to the worker. See [Node.js new Worker options][] for details. - `workerData`: (`any`) Any JavaScript value that can be cloned and made available as `require('piscina').workerData`. See [Node.js new Worker options][] for details. Unlike regular Node.js Worker Threads, `workerData` must not specify any value requiring a `transferList`. This is because the `workerData` will be cloned for each pooled worker. - `taskQueue`: (`TaskQueue`) By default, Piscina uses a first-in-first-out queue for submitted tasks. The `taskQueue` option can be used to provide an alternative implementation. See [Custom Task Queues][] for additional detail. - `niceIncrement`: (`number`) An optional value that decreases priority for the individual threads, i.e. the higher the value, the lower the priority of the Worker threads. This value is used on Unix/Windows and requires the optional [`@napi-rs/nice`][https://www.npmjs.com/package/@napi-rs/nice] module to be installed. See [`nice(2)`][https://linux.die.net/man/2/nice] for more details. - `trackUnmanagedFds`: (`boolean`) An optional setting that, when `true`, will cause Workers to track file descriptors managed using `fs.open()` and `fs.close()`, and will close them automatically when the Worker exits. Defaults to `true`. (This option is only supported on Node.js 12.19+ and all Node.js versions higher than 14.6.0). - `closeTimeout`: (`number`) An optional time (in milliseconds) to wait for the pool to complete all in-flight tasks when `close()` is called. The default is `30000` - `recordTiming`: (`boolean`) By default, run and wait time will be recorded for the pool. To disable, set to `false`. - `stricterFIFO`: (`boolean`) When `true`, tasks that cannot be dispatched immediately are returned to the front of the queue instead of the back. This avoids head-of-line blocking under sustained load (especially with a single worker). Defaults to `false`. Use caution when setting resource limits. Setting limits that are too low may result in the `Piscina` worker threads being unusable. ### Method: `run(task[, options])` Schedules a task to be run on a Worker thread. - `task`: Any value. This will be passed to the function that is exported from `filename`. - `options`: - `transferList`: An optional lists of objects that is passed to [`postMessage()`] when posting `task` to the Worker, which are transferred rather than cloned. - `filename`: Optionally overrides the `filename` option passed to the constructor for this task. If no `filename` was specified to the constructor, this is mandatory. - `name`: Optionally overrides the exported worker function used for the task. - `signal`: An [`AbortSignal`][] instance. If passed, this can be used to cancel a task. If the task is already running, the corresponding `Worker` thread will be stopped. (More generally, any `EventEmitter` or `EventTarget` that emits `'abort'` events can be passed here.) Abortable tasks cannot share threads regardless of the `concurrentTasksPerWorker` options. This returns a `Promise` for the return value of the (async) function call made to the function exported from `filename`. If the (async) function throws an error, the returned `Promise` will be rejected with that error. If the task is aborted, the returned `Promise` is rejected with an error as well. ### Method: `destroy()` Stops all Workers and rejects all `Promise`s for pending tasks. This returns a `Promise` that is fulfilled once all threads have stopped. ### Method: `close([options])` - `options`: - `force`: A `boolean` value that indicates whether to abort all tasks that are enqueued but not started yet. The default is `false`. Stops all Workers gracefully. This returns a `Promise` that is fulfilled once all tasks that were started have completed and all threads have stopped. This method is similar to `destroy()`, but with the difference that `close()` will wait for the worker tasks to finish, while `destroy()` will abort them immediately. ### Event: `'error'` An `'error'` event is emitted by instances of this class when: - Uncaught exceptions occur inside Worker threads that do not currently handle tasks. - Unexpected messages are sent from from Worker threads. All other errors are reported by rejecting the `Promise` returned from `run()`, including rejections reported by the handler function itself. ### Event: `'drain'` A `'drain'` event is emitted when the current usage of the pool is below the maximum capacity of the same. The intended goal is to provide backpressure to the task source so creating tasks that can not be executed at immediately can be avoided. ### Event: `'needsDrain'` Similar to [`Piscina#needsDrain`](#property-needsdrain-readonly); this event is triggered once the total capacity of the pool is exceeded by number of tasks enqueued that are pending of execution. ### Event: `'message'` A `'message'` event is emitted whenever a message is received from a worker thread. ### Property: `completed` (readonly) The current number of completed tasks. ### Property: `duration` (readonly) The length of time (in milliseconds) since this `Piscina` instance was created. ### Property: `options` (readonly) A copy of the options that are currently being used by this instance. This object has the same properties as the options object passed to the constructor. ### Property: `runTime` (readonly) A histogram summary object summarizing the collected run times of completed tasks. All values are expressed in milliseconds. - `runTime.average` {`number`} The average run time of all tasks - `runTime.mean` {`number`} The mean run time of all tasks - `runTime.stddev` {`number`} The standard deviation of collected run times - `runTime.min` {`number`} The fastest recorded run time - `runTime.max` {`number`} The slowest recorded run time All properties following the pattern `p{N}` where N is a number (e.g. `p1`, `p99`) represent the percentile distributions of run time observations. For example, `p99` is the 99th percentile indicating that 99% of the observed run times were faster or equal to the given value. ```js { average: 1880.25, mean: 1880.25, stddev: 1.93, min: 1877, max: 1882.0190887451172, p0_001: 1877, p0_01: 1877, p0_1: 1877, p1: 1877, p2_5: 1877, p10: 1877, p25: 1877, p50: 1881, p75: 1881, p90: 1882, p97_5: 1882, p99: 1882, p99_9: 1882, p99_99: 1882, p99_999: 1882 } ``` ### Property: `threads` (readonly) An Array of the `Worker` instances used by this pool. ### Property: `idleThreads` (readonly) The current number of ready (warmed up) Worker threads that are not running any task. ### Property: `queueSize` (readonly) The current number of tasks waiting to be assigned to a Worker thread. ### Property: `needsDrain` (readonly) Boolean value that specifies whether the capacity of the pool has been exceeded by the number of tasks submitted. This property is helpful to make decisions towards creating backpressure over the number of tasks submitted to the pool. ### Property: `utilization` (readonly) A point-in-time ratio comparing the approximate total mean run time of completed tasks to the total runtime capacity of the pool. A pools runtime capacity is determined by multiplying the `duration` by the `options.maxThread` count. This provides an absolute theoretical maximum aggregate compute time that the pool would be capable of. The approximate total mean run time is determined by multiplying the mean run time of all completed tasks by the total number of completed tasks. This number represents the approximate amount of time the pool as been actively processing tasks. The utilization is then calculated by dividing the approximate total mean run time by the capacity, yielding a fraction between `0` and `1`. ### Property: `waitTime` (readonly) A histogram summary object summarizing the collected times tasks spent waiting in the queue. All values are expressed in milliseconds. - `waitTime.average` {`number`} The average wait time of all tasks - `waitTime.mean` {`number`} The mean wait time of all tasks - `waitTime.stddev` {`number`} The standard deviation of collected wait times - `waitTime.min` {`number`} The fastest recorded wait time - `waitTime.max` {`number`} The longest recorded wait time All properties following the pattern `p{N}` where N is a number (e.g. `p1`, `p99`) represent the percentile distributions of wait time observations. For example, `p99` is the 99th percentile indicating that 99% of the observed wait times were faster or equal to the given value. ```js { average: 1880.25, mean: 1880.25, stddev: 1.93, min: 1877, max: 1882.0190887451172, p0_001: 1877, p0_01: 1877, p0_1: 1877, p1: 1877, p2_5: 1877, p10: 1877, p25: 1877, p50: 1881, p75: 1881, p90: 1882, p97_5: 1882, p99: 1882, p99_9: 1882, p99_99: 1882, p99_999: 1882 } ``` ### Static property: `isWorkerThread` (readonly) Is `true` if this code runs inside a `Piscina` threadpool as a Worker. ### Static property: `version` (readonly) Provides the current version of this library as a semver string. ### Static method: `move(value)` By default, any value returned by a worker function will be cloned when returned back to the Piscina pool, even if that object is capable of being transferred. The `Piscina.move()` method can be used to wrap and mark transferable values such that they will be transferred rather than cloned. The `value` may be any object supported by Node.js to be transferable (e.g. `ArrayBuffer`, any `TypedArray`, or `MessagePort`), or any object implementing the `Transferable` interface. ```js const { move } = require("piscina"); module.exports = () => { return move(new ArrayBuffer(10)); }; ``` The `move()` method will throw if the `value` is not transferable. The object returned by the `move()` method should not be set as a nested value in an object. If it is used, the `move()` object itself will be cloned as opposed to transferring the object it wraps. #### Interface: `Transferable` Objects may implement the `Transferable` interface to create their own custom transferable objects. This is useful when an object being passed into or from a worker contains a deeply nested transferable object such as an `ArrayBuffer` or `MessagePort`. `Transferable` objects expose two properties inspected by Piscina to determine how to transfer the object. These properties are named using the special static `Piscina.transferableSymbol` and `Piscina.valueSymbol` properties: - The `Piscina.transferableSymbol` property provides the object (or objects) that are to be included in the `transferList`. - The `Piscina.valueSymbol` property provides a surrogate value to transmit in place of the `Transferable` itself. Both properties are required. For example, ```js const { move, transferableSymbol, valueSymbol } = require('piscina'); module.exports = () => { const obj = { a: { b: new Uint8Array(5); }, c: { new Uint8Array(10); }, get [transferableSymbol]() { // Transfer the two underlying ArrayBuffers return [this.a.b.buffer, this.c.buffer]; } get [valueSymbol]() { return { a: { b: this.a.b }, c: this.c }; } }; return move(obj); }; ``` ## Custom Task Queues By default, Piscina uses a simple array-based first-in-first-out (fifo) task queue. When a new task is submitted and there are no available workers, tasks are pushed on to the queue until a worker becomes available. If the default fifo queue is not sufficient, user code may replace the task queue implementation with a custom implementation using the `taskQueue` option on the Piscina constructor. Custom task queue objects _must_ implement the `TaskQueue` interface, described below using TypeScript syntax: ```ts interface Task { readonly [Piscina.queueOptionsSymbol]: object | null; } interface TaskQueue { readonly size: number; shift(): Task | null; remove(task: Task): void; push(task: Task): void; } ``` An example of a custom task queue that uses a shuffled priority queue is available in [`examples/task-queue`](./examples/task-queue/index.js); The special symbol `Piscina.queueOptionsSymbol` may be set as a property on tasks submitted to `run()` as a way of passing additional options on to the custom `TaskQueue` implementation. (Note that because the queue options are set as a property on the task, tasks with queue options cannot be submitted as JavaScript primitives). ### Built-In Queues Piscina also provides the `FixedQueue`, a more performant task queue implementation based on [`FixedQueue`](https://github.com/nodejs/node/blob/de7b37880f5a541d5f874c1c2362a65a4be76cd0/lib/internal/fixed_queue.js) from Node.js project. Here are some benchmarks to compare new `FixedQueue` with `ArrayTaskQueue` (current default). The benchmarks demonstrate substantial improvements in push and shift operations, especially with larger queue sizes. ``` Queue size = 1000 ┌─────────┬─────────────────────────────────────────┬───────────┬────────────────────┬──────────┬─────────┐ │ (index) │ Task Name │ ops/sec │ Average Time (ns) │ Margin │ Samples │ ├─────────┼─────────────────────────────────────────┼───────────┼────────────────────┼──────────┼─────────┤ │ 0 │ 'ArrayTaskQueue full push + full shift' │ '9 692' │ 103175.15463917515 │ '±0.80%' │ 970 │ │ 1 │ 'FixedQueue full push + full shift' │ '131 879' │ 7582.696390658352 │ '±1.81%' │ 13188 │ └─────────┴─────────────────────────────────────────┴───────────┴────────────────────┴──────────┴─────────┘ Queue size = 100_000 ┌─────────┬─────────────────────────────────────────┬─────────┬────────────────────┬──────────┬─────────┐ │ (index) │ Task Name │ ops/sec │ Average Time (ns) │ Margin │ Samples │ ├─────────┼─────────────────────────────────────────┼─────────┼────────────────────┼──────────┼─────────┤ │ 0 │ 'ArrayTaskQueue full push + full shift' │ '0' │ 1162376920.0000002 │ '±1.77%' │ 10 │ │ 1 │ 'FixedQueue full push + full shift' │ '1 026' │ 974328.1553396407 │ '±2.51%' │ 103 │ └─────────┴─────────────────────────────────────────┴─────────┴────────────────────┴──────────┴─────────┘ ``` In terms of Piscina performance itself, using `FixedQueue` with a queue size of 100,000 queued tasks can result in up to 6 times faster execution times. Users can import `FixedQueue` from the `Piscina` package and pass it as the `taskQueue` option to leverage its benefits. #### Using FixedQueue Example Here's an example of how to use the `FixedQueue`: ```js const { Piscina, FixedQueue } = require("piscina"); const { resolve } = require("path"); // Create a Piscina pool with FixedQueue const piscina = new Piscina({ filename: resolve(__dirname, "worker.js"), taskQueue: new FixedQueue(), }); // Submit tasks to the pool for (let i = 0; i < 10; i++) { piscina .run({ data: i }) .then((result) => { console.log(result); }) .catch((error) => { console.error(error); }); } ``` ## Current Limitations (Things we're working on / would love help with) - Improved Documentation - Benchmarks ## Performance Notes Workers are generally optimized for offloading synchronous, compute-intensive operations off the main Node.js event loop thread. While it is possible to perform asynchronous operations and I/O within a Worker, the performance advantages of doing so will be minimal. Specifically, it is worth noting that asynchronous operations within Node.js, including I/O such as file system operations or CPU-bound tasks such as crypto operations or compression algorithms, are already performed in parallel by Node.js and libuv on a per-process level. This means that there will be little performance impact on moving such async operations into a Piscina worker (see examples/scrypt for example). ### Queue Size Piscina provides the ability to configure the minimum and maximum number of worker threads active in the pool, as well as set limits on the number of tasks that may be queued up waiting for a free worker. It is important to note that setting the `maxQueue` size too high relative to the number of worker threads can have a detrimental impact on performance and memory usage. Setting the `maxQueue` size too small can also be problematic as doing so could cause your worker threads to become idle and be shutdown. Our testing has shown that a `maxQueue` size of approximately the square of the maximum number of threads is generally sufficient and performs well for many cases, but this will vary significantly depending on your workload. It will be important to test and benchmark your worker pools to ensure you've effectively balanced queue wait times, memory usage, and worker pool utilization. ### Queue Pressure and Idle Threads The thread pool maintained by Piscina has both a minimum and maximum limit to the number of threads that may be created. When a Piscina instance is created, it will spawn the minimum number of threads immediately, then create additional threads as needed up to the limit set by `maxThreads`. Whenever a worker completes a task, a check is made to determine if there is additional work for it to perform. If there is no additional work, the thread is marked idle. By default, idle threads are shutdown immediately, with Piscina ensuring that the pool always maintains at least the minimum. When a Piscina pool is processing a stream of tasks (for instance, processing http server requests as in the React server-side rendering example in examples/react-ssr), if the rate in which new tasks are received and queued is not sufficient to keep workers from going idle and terminating, the pool can experience a thrashing effect -- excessively creating and terminating workers that will cause a net performance loss. There are a couple of strategies to avoid this churn: Strategy 1: Ensure that the queue rate of new tasks is sufficient to keep workers from going idle. We refer to this as "queue pressure". If the queue pressure is too low, workers will go idle and terminate. If the queue pressure is too high, tasks will stack up, experience increased wait latency, and consume additional memory. Strategy 2: Increase the `idleTimeout` configuration option. By default, idle threads terminate immediately. The `idleTimeout` option can be used to specify a longer period of time to wait for additional tasks to be submitted before terminating the worker. If the queue pressure is not maintained, this could result in workers sitting idle but those will have less of a performance impact than the thrashing that occurs when threads are repeatedly terminated and recreated. Strategy 3: Increase the `minThreads` configuration option. This has the same basic effect as increasing the `idleTimeout`. If the queue pressure is not high enough, workers may sit idle indefinitely but there will be less of a performance hit. In applications using Piscina, it will be most effective to use a combination of these three approaches and tune the various configuration parameters to find the optimum combination both for the application workload and the capabilities of the deployment environment. There are no one set of options that are going to work best. ### Thread priority on Linux systems On Linux systems that support [`nice(2)`][], Piscina is capable of setting the priority of every worker in the pool. To use this mechanism, an additional optional native addon dependency (`@napi-rs/nice`, `npm i @napi-rs/nice`) is required. Once [`@napi-rs/nice`][] is installed, creating a `Piscina` instance with the `niceIncrement` configuration option will set the priority for the pool: ```js const Piscina = require("piscina"); const pool = new Piscina({ worker: "/absolute/path/to/worker.js", niceIncrement: 20, }); ``` The higher the `niceIncrement`, the lower the CPU scheduling priority will be for the pooled workers which will generally extend the execution time of CPU-bound tasks but will help prevent those threads from stealing CPU time from the main Node.js event loop thread. Whether this is a good thing or not depends entirely on your application and will require careful profiling to get correct. The key metrics to pay attention to when tuning the `niceIncrement` are the sampled run times of the tasks in the worker pool (using the [`runTime`][] property) and the [delay of the Node.js main thread event loop][]. ### Multiple Thread Pools and Embedding Piscina as a Dependency Every `Piscina` instance creates a separate pool of threads and operates without any awareness of the other. When multiple pools are created in a single application the various threads may contend with one another, and with the Node.js main event loop thread, and may cause an overall reduction in system performance. Modules that embed Piscina as a dependency _should_ make it clear via documentation that threads are being used. It would be ideal if those would make it possible for users to provide an existing `Piscina` instance as a configuration option in lieu of always creating their own. ## The Team - James M Snell - Anna Henningsen - Matteo Collina - Rafael Gonzaga - Robert Nagy - Carlos Fuentes ## Acknowledgements Piscina development was initially sponsored by [NearForm Research][]. ## Sponsors Looking to support the development of Piscina? Consider sponsoring us on [Open Collective](https://opencollective.com/piscinajs). We appreciate all levels of support! ### Bronze Sponsors ## Resources [`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 [`examples/task-queue`]: https://github.com/jasnell/piscina/blob/master/examples/task-queue/index.js [`nice(2)`]: https://linux.die.net/man/2/nice [`@napi-rs/nice`]: https://npmjs.org/package/@napi-rs/nice [`runTime`]: #property-runtime-readonly [Custom Task Queues]: #custom-task-queues [ES modules]: https://nodejs.org/api/esm.html [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/ [delay of the Node.js main thread event loop]: https://nodejs.org/dist/latest-v14.x/docs/api/perf_hooks.html#perf_hooks_perf_hooks_monitoreventloopdelay_options --- ### Overview (`docs/README.md`) # Website This website is built using [Docusaurus](https://docusaurus.io/), a modern static website generator. ### Installation ``` $ npm i ``` ### Local Development ``` $ npm run start ``` This command starts a local development server and opens up a browser window. Most changes are reflected live without having to restart the server. ### Build ``` $ npm run build ``` This command generates static content into the `build` directory and can be served using any static contents hosting service. --- ### Overview (`examples/n-api/README.md`) # Piscina N-API (native addon) example A Piscina worker can be implemented using a native addon, allowing work to performed by native code. This provides a number of interesting options including implementing workers using other languages such as Rust. To get started with this example, first install the dependencies: ```console $ npm i ``` Then build the artifacts: ```console $ npm run prebuild ``` The `prebuild` command will build the binary artifacts for the native addon and will put them in the `prebuilds` folder. Because of how prebuilds work, we need to use an intermediate JavaScript file to load and export them. For this example native addon, you'll find that in the `examples` folder. The index.js illustrates how to load and use the native addon as the worker implementation: ```js const Piscina = require('piscina'); const { resolve } = require('path'); const pool = new Piscina({ filename: resolve(__dirname, 'example') }); (async () => { // Submit 5 concurrent tasks console.log(await Promise.all([ pool.run(), pool.run(), pool.run(), pool.run(), pool.run() ])); })(); ``` --- ### Overview (`examples/scrypt/README.md`) # scrypt Example This is an example comparing the performance of pooled and unpooled sync and async scrypt operations. ```console $ npm run pooled 100 $ npm run unpooled 100 $ npm run pooled-sync 100 $ npm run unpooled-sync 100 ``` --- ### Overview (`examples/server/README.md`) # Server example The benefit of offloading work to a worker pool will vary significantly based on work load. In some cases, the worker pool may actually be a detriment to performance. Careful consideration must be given. In this example, we create a simple Fastify http server that responds with a simple JSON object after introducing an artificial delay of 100 milliseconds. There are four variants to the server: * async-sleep-unpooled -- Uses an async delay using a Promise-wrapped setTimeout. The event loop is allowed to keep turning during the delay allowing the server to respond to additional requests. The delay occurs within the main thread. * async-sleep-pooled -- Moves the async delay to the worker threads for processing. * sync-sleep-unpooled -- Uses a sync delay using Atomics.wait to block the main thread for 100 milliseconds before responding. Because the main thread is blocked, the event loop is not turning while we are waiting. This simulates are particularly expensive event loop blocking scenario. * sync-sleep-pooled -- Moves the sync delay into the worker threads for processing. Note that for both `async-sleep-pooled` and `sync-sleep-pooled`, the main thread uses a Promise to wait on the worker to complete it's task before responding to the request. To set up for this example, run `npm i` to install dependencies. We use [`autocannon`][] for benchmarking the four variants: ```console $ npm i autocannon ``` Let's start with the two async sleep variants. First run: ```console $ node async-sleep-unpooled ``` And in a separate terminal, run autocannon: ```console $ autocannon localhost:3000 ``` You should see results similar to: ``` Running 10s test @ http://localhost:3000 10 connections ┌─────────┬────────┬────────┬────────┬────────┬──────────┬─────────┬───────────┐ │ Stat │ 2.5% │ 50% │ 97.5% │ 99% │ Avg │ Stdev │ Max │ ├─────────┼────────┼────────┼────────┼────────┼──────────┼─────────┼───────────┤ │ Latency │ 100 ms │ 101 ms │ 104 ms │ 125 ms │ 101.2 ms │ 3.09 ms │ 130.66 ms │ └─────────┴────────┴────────┴────────┴────────┴──────────┴─────────┴───────────┘ ┌───────────┬─────────┬─────────┬─────────┬─────────┬───────┬───────┬─────────┐ │ Stat │ 1% │ 2.5% │ 50% │ 97.5% │ Avg │ Stdev │ Min │ ├───────────┼─────────┼─────────┼─────────┼─────────┼───────┼───────┼─────────┤ │ Req/Sec │ 90 │ 90 │ 99 │ 100 │ 97.8 │ 3.32 │ 90 │ ├───────────┼─────────┼─────────┼─────────┼─────────┼───────┼───────┼─────────┤ │ Bytes/Sec │ 14.8 kB │ 14.8 kB │ 16.2 kB │ 16.4 kB │ 16 kB │ 545 B │ 14.8 kB │ └───────────┴─────────┴─────────┴─────────┴─────────┴───────┴───────┴─────────┘ ``` Note that this server is not particularly fast at all due to the artificially imposed 100 millisecond delay. (It's possible to achieve significantly higher results by tuning the way autocannon is sending requests. For instance, running autocannon with the `-c 100 -p 2` options increases performance of the `async-sleep-unpooled` example significantly.) Let's see how the Piscina version does in comparison: Run: ```console $ node async-sleep-pooled ``` And autocannon again: ```console $ autocannon localhost:3000 ``` Your result should be similar to: ``` Running 10s test @ http://localhost:3000 10 connections ┌─────────┬────────┬────────┬────────┬────────┬───────────┬──────────┬───────────┐ │ Stat │ 2.5% │ 50% │ 97.5% │ 99% │ Avg │ Stdev │ Max │ ├─────────┼────────┼────────┼────────┼────────┼───────────┼──────────┼───────────┤ │ Latency │ 126 ms │ 169 ms │ 197 ms │ 339 ms │ 170.13 ms │ 31.41 ms │ 425.33 ms │ └─────────┴────────┴────────┴────────┴────────┴───────────┴──────────┴───────────┘ ┌───────────┬─────────┬─────────┬─────────┬─────────┬─────────┬───────┬─────────┐ │ Stat │ 1% │ 2.5% │ 50% │ 97.5% │ Avg │ Stdev │ Min │ ├───────────┼─────────┼─────────┼─────────┼─────────┼─────────┼───────┼─────────┤ │ Req/Sec │ 43 │ 43 │ 60 │ 60 │ 58.2 │ 5.08 │ 43 │ ├───────────┼─────────┼─────────┼─────────┼─────────┼─────────┼───────┼─────────┤ │ Bytes/Sec │ 7.05 kB │ 7.05 kB │ 9.85 kB │ 9.85 kB │ 9.55 kB │ 833 B │ 7.05 kB │ └───────────┴─────────┴─────────┴─────────┴─────────┴─────────┴───────┴─────────┘ ``` Notice that the pooled version is slower! This is because in both async examples, the event loop on the main thread is still active, allowing it to service more requests. The worker pool here, however, adds additional performance overhead marshalling data back and forth between threads and allocating additional Promises. In this scenario, the worker pool does not add much benefit. Let's look as the sync cases and see what happens with those. Run: ```console $ node sync-sleep-unpooled ``` And run autocannon again: ```console $ autocannon localhost:3000 ``` The results should be fairly awful in comparison to the first two cases: ``` Running 10s test @ http://localhost:3000 10 connections ┌─────────┬────────┬─────────┬─────────┬─────────┬──────────┬───────────┬────────────┐ │ Stat │ 2.5% │ 50% │ 97.5% │ 99% │ Avg │ Stdev │ Max │ ├─────────┼────────┼─────────┼─────────┼─────────┼──────────┼───────────┼────────────┤ │ Latency │ 338 ms │ 1007 ms │ 1811 ms │ 1919 ms │ 962.2 ms │ 255.54 ms │ 1919.77 ms │ └─────────┴────────┴─────────┴─────────┴─────────┴──────────┴───────────┴────────────┘ ┌───────────┬─────────┬─────────┬─────────┬─────────┬─────────┬────────┬─────────┐ │ Stat │ 1% │ 2.5% │ 50% │ 97.5% │ Avg │ Stdev │ Min │ ├───────────┼─────────┼─────────┼─────────┼─────────┼─────────┼────────┼─────────┤ │ Req/Sec │ 9 │ 9 │ 10 │ 10 │ 9.81 │ 0.4 │ 9 │ ├───────────┼─────────┼─────────┼─────────┼─────────┼─────────┼────────┼─────────┤ │ Bytes/Sec │ 1.48 kB │ 1.48 kB │ 1.64 kB │ 1.64 kB │ 1.61 kB │ 65.6 B │ 1.48 kB │ └───────────┴─────────┴─────────┴─────────┴─────────┴─────────┴────────┴─────────┘ ``` The reason for the performance drop should be apparent. Synchronously sleeping blocks the event loop from turning while a request is being processed, which means the server is unable to do anything else while it waits. To see how the pooled version fares, run: ```console $ node sync-sleep-pooled ``` And run autocannon again ```console $ autocannon localhost:3000 ``` The results should be nearly identical to the `async-sleep-pooled` version! ``` Running 10s test @ http://localhost:3000 10 connections ┌─────────┬────────┬────────┬────────┬────────┬───────────┬──────────┬──────────┐ │ Stat │ 2.5% │ 50% │ 97.5% │ 99% │ Avg │ Stdev │ Max │ ├─────────┼────────┼────────┼────────┼────────┼───────────┼──────────┼──────────┤ │ Latency │ 124 ms │ 174 ms │ 200 ms │ 335 ms │ 169.99 ms │ 36.22 ms │ 422.2 ms │ └─────────┴────────┴────────┴────────┴────────┴───────────┴──────────┴──────────┘ ┌───────────┬─────────┬─────────┬─────────┬─────────┬─────────┬───────┬─────────┐ │ Stat │ 1% │ 2.5% │ 50% │ 97.5% │ Avg │ Stdev │ Min │ ├───────────┼─────────┼─────────┼─────────┼─────────┼─────────┼───────┼─────────┤ │ Req/Sec │ 42 │ 42 │ 60 │ 60 │ 58.2 │ 5.4 │ 42 │ ├───────────┼─────────┼─────────┼─────────┼─────────┼─────────┼───────┼─────────┤ │ Bytes/Sec │ 6.89 kB │ 6.89 kB │ 9.85 kB │ 9.85 kB │ 9.55 kB │ 886 B │ 6.89 kB │ └───────────┴─────────┴─────────┴─────────┴─────────┴─────────┴───────┴─────────┘ ``` The reason `sync-sleep-pooled` and `async-sleep-pooled` yield identical results is because, in both cases, the main thread is doing identical work -- that is, receiving a request that is dispatched to a worker, waiting about 100 milliseconds, the returning the response. The main thread here does not care whether the workers are sleeping synchronously or asynchronously. ## Tuning pool performance It is possible to tune the performance of the Piscina worker pool using a variety of options: * `minThreads` - The minimum number of threads always running * `maxThreads` - The maximum number of threads allowed * `idleTimeout` - The number of millisecondsa thread is permitted to remain idle * `maxQueue` -- The maximum number of pending work items * `concurrentTasksPerWorker` -- The number of work items to dispatch concurrently to a single thread. We'll use `idleTimeout` and `concurrentTasksPerWorker` in this example. By default, as soon as a Piscina worker thread has nothing to do, it will terminate. This means that if the work queue is empty, the thread will terminate. If we're not keeping our queue filled, this will incur additional overhead as Node.js spins up new worker threads to handle incoming requests. Also by default, Piscina will assume that jobs are synchronous in nature and will dispatch only a single job per thread at any time. If the workload is asynchronous, setting `concurrentTasksPerWorker` to a higher number will increase the number of jobs Piscina will send to a single worker, allowing it to process multiple tasks. The `async-sleep-pooled` example has been written to accept two optional parameters. The first is the `concurrentTasksPerWorker`, and the second is the `idleTimeout`. To see the effect setting each has on the performance of the example, run: ```console $ node async-sleep-pooled 10 1000 ``` Then run autocannon again: ```console $ autocannon localhost:3000 ``` You should see that the performance of the pooled example improves significantly, but is still slightly less than the async-sleep-unpooled version: ``` Running 10s test @ http://localhost:3000 10 connections ┌─────────┬────────┬────────┬────────┬────────┬───────────┬──────────┬───────────┐ │ Stat │ 2.5% │ 50% │ 97.5% │ 99% │ Avg │ Stdev │ Max │ ├─────────┼────────┼────────┼────────┼────────┼───────────┼──────────┼───────────┤ │ Latency │ 100 ms │ 101 ms │ 126 ms │ 135 ms │ 105.16 ms │ 23.23 ms │ 391.75 ms │ └─────────┴────────┴────────┴────────┴────────┴───────────┴──────────┴───────────┘ ┌───────────┬─────────┬─────────┬─────────┬─────────┬─────────┬─────────┬─────────┐ │ Stat │ 1% │ 2.5% │ 50% │ 97.5% │ Avg │ Stdev │ Min │ ├───────────┼─────────┼─────────┼─────────┼─────────┼─────────┼─────────┼─────────┤ │ Req/Sec │ 76 │ 76 │ 97 │ 99 │ 94.2 │ 6.42 │ 76 │ ├───────────┼─────────┼─────────┼─────────┼─────────┼─────────┼─────────┼─────────┤ │ Bytes/Sec │ 12.5 kB │ 12.5 kB │ 15.9 kB │ 16.2 kB │ 15.5 kB │ 1.05 kB │ 12.5 kB │ └───────────┴─────────┴─────────┴─────────┴─────────┴─────────┴─────────┴─────────┘ ``` Let's use the same options for the `sync-sleep-pooled.js` example: ```console $ node sync-sleep-pooled 10 1000 ``` Then run autocannon again: ```console $ autocannon localhost:3000 ``` With the results: ``` Running 10s test @ http://localhost:3000 10 connections ┌─────────┬────────┬────────┬────────┬────────┬───────────┬──────────┬──────────┐ │ Stat │ 2.5% │ 50% │ 97.5% │ 99% │ Avg │ Stdev │ Max │ ├─────────┼────────┼────────┼────────┼────────┼───────────┼──────────┼──────────┤ │ Latency │ 100 ms │ 200 ms │ 201 ms │ 329 ms │ 169.05 ms │ 52.78 ms │ 456.5 ms │ └─────────┴────────┴────────┴────────┴────────┴───────────┴──────────┴──────────┘ ┌───────────┬─────────┬─────────┬─────────┬─────────┬─────────┬───────┬─────────┐ │ Stat │ 1% │ 2.5% │ 50% │ 97.5% │ Avg │ Stdev │ Min │ ├───────────┼─────────┼─────────┼─────────┼─────────┼─────────┼───────┼─────────┤ │ Req/Sec │ 46 │ 46 │ 60 │ 60 │ 58.4 │ 4.16 │ 46 │ ├───────────┼─────────┼─────────┼─────────┼─────────┼─────────┼───────┼─────────┤ │ Bytes/Sec │ 7.55 kB │ 7.55 kB │ 9.85 kB │ 9.85 kB │ 9.58 kB │ 681 B │ 7.54 kB │ └───────────┴─────────┴─────────┴─────────┴─────────┴─────────┴───────┴─────────┘ ``` The options have no noticeable impact on the sync sleep version. The reason is because the workload is fully synchronous and cannot be processed concurrently no matter what the concurrency settings are. The performance, in other words, in bound entirely to the synchronous delay and can only be improved by reducing the event loop block. [`autocannon`]: https://www.npmjs.com/package/autocannon --- ### Docs Api Reference Api Overview (`docs/docs/api-reference/api-overview.md`) --- id: API Overview sidebar_position: 1 --- | API | Description | | ---------------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | **Class: `Piscina`** | This class extends [`EventEmitter`](https://nodejs.org/api/events.html#events_class_eventemitter) from Node.js. | | **Constructor: `new Piscina([options])`** | Creates a new instance of the `Piscina` class. [List of available options](./class.md) | | **Method: `run(task[, options])`** | Schedules a task to be run on a Worker thread. | | **Method: `runTask(task[, transferList][, filename][, abortSignal])`** | **Deprecated** -- Use `run(task, options)` instead. | | **Method: `destroy()`** | Stops all Workers and rejects all `Promises` for pending tasks. | | **Method: `close([options])`** | Stops all Workers gracefully. | | **Event: `'error'`** | Emitted when uncaught exceptions occur inside Worker threads or unexpected messages are sent from Worker threads. | | **Event: `'drain'`** | Emitted whenever the `queueSize` reaches `0`. | | **Event: `'needsDrain'`** | Triggered once the total capacity of the pool is exceeded by the number of tasks enqueued that are pending execution. | | **Event: `'message'`** | Emitted whenever a message is received from a worker thread. | | **Property: `completed` (readonly)** | The current number of completed tasks. | | **Property: `duration` (readonly)** | The length of time (in milliseconds) since this `Piscina` instance was created. | | **Property: `options` (readonly)** | A copy of the options that are currently being used by this instance. | | **Property: `runTime` (readonly)** | A histogram summary object summarizing the collected run times of completed tasks. | | **Property: `threads` (readonly)** | An Array of the `Worker` instances used by this pool. | | **Property: `idleThreads` (readonly)** | The current number of ready (warmed up) Worker threads that are not running any task. | | **Property: `queueSize` (readonly)** | The current number of tasks waiting to be assigned to a Worker thread. | | **Property: `needsDrain` (readonly)** | Boolean value that specifies whether the capacity of the pool has been exceeded by the number of tasks submitted. | | **Property: `utilization` (readonly)** | A point-in-time ratio comparing the approximate total mean run time of completed tasks to the total runtime capacity of the pool. | | **Property: `waitTime` (readonly)** | A histogram summary object summarizing the collected times tasks spent waiting in the queue. | | **Static Property: `isWorkerThread` (readonly)** | Is `true` if this code runs inside a `Piscina` threadpool as a Worker. | | **Static Property: `version` (readonly)** | Provides the current version of this library as a semver string. | | **Static Method: `move(value)`** | By default, any value returned by a worker function will be cloned when returned back to the Piscina pool, even if that object is capable of being transferred. The `Piscina.move()` method can be used to wrap and mark transferable values such that they will be transferred rather than cloned. | | **Interface: `Transferable`** | Objects may implement the `Transferable` interface to create their own custom transferable objects. | --- ### Package.Json (`docs/package.json`) { "name": "docs", "version": "0.0.0", "private": true, "scripts": { "docusaurus": "docusaurus", "start": "docusaurus start", "build": "docusaurus build", "swizzle": "docusaurus swizzle", "deploy": "docusaurus deploy", "clear": "docusaurus clear", "serve": "docusaurus serve", "write-translations": "docusaurus write-translations", "write-heading-ids": "docusaurus write-heading-ids", "typecheck": "tsc" }, "dependencies": { "@docusaurus/core": "3.10.2", "@docusaurus/preset-classic": "3.10.2", "@easyops-cn/docusaurus-search-local": "^0.51.1", "@mdx-js/react": "^3.1.0", "prism-react-renderer": "^2.4.1", "react": "^19.1.0", "react-dom": "^19.1.0" }, "devDependencies": { "@docusaurus/module-type-aliases": "3.8.1", "@docusaurus/tsconfig": "3.8.1", "@docusaurus/types": "3.8.1", "typescript": "~5.8.3" }, "browserslist": { "production": [ ">0.5%", "not dead", "not op_mini all" ], "development": [ "last 3 chrome version", "last 3 firefox version", "last 5 safari version" ] }, "engines": { "node": ">=22.0" } } --- ### Tsconfig.Json (`docs/tsconfig.json`) { "extends": "@docusaurus/tsconfig", "compilerOptions": { "baseUrl": "." } } --- ### Docs Introduction (`docs/docs/Introduction.md`) --- sidebar_position: 1 slug: / --- # Introduction Piscina.js is a powerful Node.js worker pool library that allows you to efficiently run CPU-intensive tasks in parallel using worker threads. It provides a simple API for offloading computationally expensive tasks to a pool of worker threads, thereby improving the performance and scalability of your Node.js applications. ## Why Piscina? In the early days of worker threads, the Node.js core team encountered an issue where a user's application was spinning up thousands of concurrent worker threads, leading to performance issues. While this specific issue helped identify a minor memory leak in the worker implementation, it highlighted a broader problem: the misuse of worker threads due to a lack of understanding. While worker threads have matured and their usage has become more widespread, there is still a need for better examples and education around their correct usage. This realization led to the creation of Piscina, an open-source project sponsored by [NearForm Research](https://www.nearform.com/), focused on providing guidance and best practices for using worker threads in Node.js applications. With worker threads now a well-established feature in Node.js, Piscina aims to bridge the gap between the potential of worker threads and their practical implementation. ## Key features ✔ 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 ## Sponsors Looking to support the development of Piscina? Consider sponsoring us on [Open Collective](https://opencollective.com/piscinajs). We appreciate all levels of support! ### Bronze Sponsors TestMu.ai logo TestMu.ai logo --- ### Docs Advanced Topics Category .Json (`docs/docs/advanced-topics/_category_.json`) { "label": "Advanced topics", "position": 4, "link": { "type": "generated-index", "description": "In this section, you'll find information on more complex features of Piscina like setting up your own task queues and tips for making your application run faster. This part of the documentation is great for when you want to get more out of Piscina and understand how to manage tasks and performance better." } } --- ### Docs Advanced Topics Custom Task Queues (`docs/docs/advanced-topics/custom-task-queues.mdx`) --- id: Custom Task Queues sidebar_position: 1 --- import {WorkerWrapperComponent} from '@site/src/components/WorkerWrapper.mdx'; import Tabs from "@theme/Tabs"; import TabItem from "@theme/TabItem"; By default, Piscina uses a simple array-based first-in-first-out (fifo) task queue. When a new task is submitted and there are no available workers, tasks are pushed on to the queue until a worker becomes available. If the default fifo queue is not sufficient, user code may replace the task queue implementation with a custom implementation using the `taskQueue` option on the Piscina constructor. Custom task queue objects _must_ implement the `TaskQueue` interface, described below using TypeScript syntax: ```ts interface Task { readonly [Piscina.queueOptionsSymbol]: object | null; } interface TaskQueue { readonly size: number; shift(): Task | null; remove(task: Task): void; push(task: Task): void; } ``` An example of a custom task queue that uses a shuffled priority queue is available in [`examples/task-queue`](https://github.com/piscinajs/piscina/tree/current/examples/task-queue); The special symbol `Piscina.queueOptionsSymbol` may be set as a property on tasks submitted to `run()` or `runTask()` as a way of passing additional options on to the custom `TaskQueue` implementation. (Note that because the queue options are set as a property on the task, tasks with queue options cannot be submitted as JavaScript primitives). ### Built-In Queues Piscina also provides the `FixedQueue`, a more performant task queue implementation based on [`FixedQueue`](https://github.com/nodejs/node/blob/de7b37880f5a541d5f874c1c2362a65a4be76cd0/lib/internal/fixed_queue.js) from Node.js project. Here are some benchmarks to compare new `FixedQueue` with `ArrayTaskQueue` (current default). The benchmarks demonstrate substantial improvements in push and shift operations, especially with larger queue sizes.
**Queue size = 1000** | (index) | Task Name | ops/sec | Average Time (ns) | Margin | Samples | |---------|-----------------------------------------|-----------|--------------------|----------|---------| | 0 | 'ArrayTaskQueue full push + full shift' | '9 692' | 103175.15463917515 | '±0.80%' | 970 | | 1 | 'FixedQueue full push + full shift' | '131 879' | 7582.696390658352 | '±1.81%' | 13188 |
**Queue size = 100,000** | (index) | Task Name | ops/sec | Average Time (ns) | Margin | Samples | |---------|-----------------------------------------|---------|--------------------|----------|---------| | 0 | 'ArrayTaskQueue full push + full shift' | '0' | 1162376920.0000002 | '±1.77%' | 10 | | 1 | 'FixedQueue full push + full shift' | '1 026' | 974328.1553396407 | '±2.51%' | 103 | In terms of Piscina performance itself, using `FixedQueue` with a queue size of 100,000 queued tasks can result in up to 6 times faster execution times. Users can import `FixedQueue` from the `Piscina` package and pass it as the `taskQueue` option to leverage its benefits. ### Using FixedQueue Here's an example of how to use the `FixedQueue`: ```js const { Piscina, FixedQueue } = require("piscina"); const { resolve } = require("path"); // Create a Piscina pool with FixedQueue const piscina = new Piscina({ filename: resolve(__dirname, "worker.js"), taskQueue: new FixedQueue(), }); // Submit tasks to the pool for (let i = 0; i < 10; i++) { piscina .runTask({ data: i }) .then((result) => { console.log(result); }) .catch((error) => { console.error(error); }); } ``` ```js title="main.ts" import Piscina from "piscina"; import { resolve } from "path"; import { filename } from "./worker"; const piscina = new Piscina({ filename: resolve(__dirname, "./workerWrapper.js"), workerData: { fullpath: filename }, }); // Submit tasks to the pool for (let i = 0; i < 10; i++) { piscina .runTask({ data: i }) .then((result) => { console.log(result); }) .catch((error) => { console.error(error); }); } ``` :::info The `FixedQueue` will become the default task queue implementation in a next major version. ::: --- ### Docs Advanced Topics Loadbalancer (`docs/docs/advanced-topics/loadbalancer.mdx`) --- id: Custom Load Balancers sidebar_position: 2 --- [Load Balancers](https://www.cloudflare.com/learning/performance/what-is-load-balancing/) are a crucial part of any distributed workload. Piscina by default uses a [Resource Based](https://www.cloudflare.com/en-gb/learning/performance/types-of-load-balancing-algorithms/) algorithm (least-busy) to distribute the load across the different workers set by the pool. ### Choosing the Load Balancing Algorithm Choosing the right algorithm heavily depends on the requirements of each individual problem, and to assess them by testing the implementation against several variations of workloads. The algorithms can be grouped on: #### Dynamic Focused on taking decision based on heuristics from the workload. It aims to balance the work by adapting itself to the environment and distribute the workloads equally across the different workers/nodes attempting to make better use of resources or to minimize the time to finish each individual workload. > Piscina uses a Dynamic algorithm, based on how busy the worker is (least-busy) #### Static Aims to distribute the workload based on a group of predefined factors. It does not adapt to the environment, but rather aims to preserve the state of the distribution accordingly to the values set beforehand. This can be helpful under several situations where we want the workload to be distributed evenly or stick to a specific worker/node over time. > Round Robin, Ring Hash, are just examples of these algorithms It is heavily advised to understand the problem and the workload your pool will be facing, and tests heavily against the different algorithms that matches your use case with several combinations to better understand its impact and how to manage it. ### Resources - [Load Balancing Algorithms](https://en.wikipedia.org/wiki/Load_balancing_(computing)) - [Types of Load Balancing Algorithms](https://www.cloudflare.com/en-gb/learning/performance/types-of-load-balancing-algorithms/) --- ### Docs Advanced Topics Performance (`docs/docs/advanced-topics/performance.mdx`) --- id: Performance Notes sidebar_position: 3 --- Workers are generally optimized for offloading synchronous, compute-intensive operations off the main Node.js event loop thread. While it is possible to perform asynchronous operations and I/O within a Worker, the performance advantages of doing so will be minimal. Specifically, it is worth noting that asynchronous operations within Node.js, including I/O such as file system operations or CPU-bound tasks such as crypto operations or compression algorithms, are already performed in parallel by Node.js and libuv on a per-process level. This means that there will be little performance impact on moving such async operations into a Piscina worker (see examples/scrypt for example). ### Queue Size Piscina provides the ability to configure the minimum and maximum number of worker threads active in the pool, as well as set limits on the number of tasks that may be queued up waiting for a free worker. It is important to note that setting the `maxQueue` size too high relative to the number of worker threads can have a detrimental impact on performance and memory usage. Setting the `maxQueue` size too small can also be problematic as doing so could cause your worker threads to become idle and be shutdown. Our testing has shown that a `maxQueue` size of approximately the square of the maximum number of threads is generally sufficient and performs well for many cases, but this will vary significantly depending on your workload. It will be important to test and benchmark your worker pools to ensure you've effectively balanced queue wait times, memory usage, and worker pool utilization. ### Queue Pressure and Idle Threads The thread pool maintained by Piscina has both a minimum and maximum limit to the number of threads that may be created. When a Piscina instance is created, it will spawn the minimum number of threads immediately, then create additional threads as needed up to the limit set by `maxThreads`. Whenever a worker completes a task, a check is made to determine if there is additional work for it to perform. If there is no additional work, the thread is marked idle. By default, idle threads are shutdown immediately, with Piscina ensuring that the pool always maintains at least the minimum. When a Piscina pool is processing a stream of tasks (for instance, processing http server requests as in the React server-side rendering example in examples/react-ssr), if the rate in which new tasks are received and queued is not sufficient to keep workers from going idle and terminating, the pool can experience a thrashing effect -- excessively creating and terminating workers that will cause a net performance loss. There are a couple of strategies to avoid this churn: - Strategy 1: Ensure that the queue rate of new tasks is sufficient to keep workers from going idle. We refer to this as "queue pressure". If the queue pressure is too low, workers will go idle and terminate. If the queue pressure is too high, tasks will stack up, experience increased wait latency, and consume additional memory. - Strategy 2: Increase the `idleTimeout` configuration option. By default, idle threads terminate immediately. The `idleTimeout` option can be used to specify a longer period of time to wait for additional tasks to be submitted before terminating the worker. If the queue pressure is not maintained, this could result in workers sitting idle but those will have less of a performance impact than the thrashing that occurs when threads are repeatedly terminated and recreated. - Strategy 3: Increase the `minThreads` configuration option. This has the same basic effect as increasing the `idleTimeout`. If the queue pressure is not high enough, workers may sit idle indefinitely but there will be less of a performance hit. In applications using Piscina, it will be most effective to use a combination of these three approaches and tune the various configuration parameters to find the optimum combination both for the application workload and the capabilities of the deployment environment. There are no one set of options that are going to work best. ### Thread priority on Linux systems On Unix systems that support [`nice(2)`][https://linux.die.net/man/2/nice] and Windows, Piscina is capable of setting the priority of every worker in the pool. To use this mechanism, an additional optional native addon dependency (`@napi-rs/nice`, `npm i @napi-rs/nice`) is required. Once [`@napi-rs/nice`][https://www.npmjs.com/package/@napi-rs/nice] is installed, creating a `Piscina`instance with the`niceIncrement` configuration option will set the priority for the pool: ```js const Piscina = require('piscina') const { WindowsThreadPriority } = require('@napi-rs/nice') const pool = new Piscina({ worker: '/absolute/path/to/worker.js', niceIncrement: process.platform !== 'win32' ? 20 : WindowsThreadPriority.ThreadPriorityHighest, }) ``` The higher the `niceIncrement`, the lower the CPU scheduling priority will be for the pooled workers which will generally extend the execution time of CPU-bound tasks but will help prevent those threads from stealing CPU time from the main Node.js event loop thread. Whether this is a good thing or not depends entirely on your application and will require careful profiling to get correct. The key metrics to pay attention to when tuning the `niceIncrement` are the sampled run times of the tasks in the worker pool (using the [`runTime`][] property) and the [delay of the Node.js main thread event loop][]. ### Multiple Thread Pools and Embedding Piscina as a Dependency Every `Piscina` instance creates a separate pool of threads and operates without any awareness of the other. When multiple pools are created in a single application the various threads may contend with one another, and with the Node.js main event loop thread, and may cause an overall reduction in system performance. Modules that embed Piscina as a dependency *should* make it clear via documentation that threads are being used. It would be ideal if those would make it possible for users to provide an existing `Piscina` instance as a configuration option in lieu of always creating their own. --- ### Docs Api Reference Category .Json (`docs/docs/api-reference/_category_.json`) { "label": "API Reference", "position": 5, "link": { "type": "generated-index", "description": "This section provides a detailed explanation of the classes, methods, properties, events, and other components that make up the library" } } --- ### Docs Api Reference Class (`docs/docs/api-reference/class.md`) --- id: Instance sidebar_position: 2 --- ## 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`](https://nodejs.org/api/events.html) from Node.js. ### Constructor: `new Piscina([options])` - The following optional configuration is supported: - `filename`: (`string | null`) Provides the default source for the code that runs the tasks on Worker threads. This should be an absolute path or an absolute `file://` URL to a file that exports a JavaScript `function` or `async function` as its default export or `module.exports`. [ES modules](https://nodejs.org/api/esm.html) are supported. - `name`: (`string | null`) Provides the name of the default exported worker function. The default is `'default'`, indicating the default export of the worker module. - `minThreads`: (`number`) Sets the minimum number of threads that are always running for this thread pool. The default is the number provided by [`os.availableParallelism`](https://nodejs.org/api/os.html#osavailableparallelism). - `maxThreads`: (`number`) Sets the maximum number of threads that are running for this thread pool. The default is the number provided by [`os.availableParallelism`](https://nodejs.org/api/os.html#osavailableparallelism) \* 1.5. - `idleTimeout`: (`number`) A timeout in milliseconds that specifies how long a `Worker` is allowed to be idle, i.e. not handling any tasks, before it is shut down. By default, this is immediate. If `Infinity` is passed as the value, the `Worker` never shuts down. :::info The default `idleTimeout` can lead to some performance loss in the application because of the overhead involved with stopping and starting new worker threads. To improve performance, try setting the `idleTimeout` explicitly. ::: :::info Be careful when when setting `idleTimeout` to `Infinity`, as this will prevent the worker from shutting down, even when idle, potentially leading to resource overuse. ::: - `maxQueue`: (`number` | `string`) The maximum number of tasks that may be scheduled to run, but not yet running due to lack of available threads, at a given time. By default, there is no limit. The special value `'auto'` may be used to have Piscina calculate the maximum as the square of `maxThreads`. When `'auto'` is used, the calculated `maxQueue` value may be found by checking the [`options.maxQueue`](#property-options-readonly) property. - `concurrentTasksPerWorker`: (`number`) Specifies how many tasks can share a single Worker thread simultaneously. The default is `1`. This generally only makes sense to specify if there is some kind of asynchronous component to the task. Keep in mind that Worker threads are generally not built for handling I/O in parallel. - `atomics`: (`sync` | `async` | `disabled`) Use the [`Atomics`](https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Atomics) API for faster communication between threads. This is on by default. You can disable `Atomics` globally by setting the environment variable `PISCINA_DISABLE_ATOMICS` to `1` . If `atomics` is `sync`, it will cause to pause threads (stopping all execution) between tasks. Ideally, threads should wait for all operations to finish before returning control to the main thread (avoid having open handles within a thread). If still want to have the possibility of having open handles or handle asynchronous tasks, you can set the environment variable `PISCINA_ENABLE_ASYNC_ATOMICS` to `1` or setting `options.atomics` to `async`. :::info **Note**: The `async` mode comes with performance penalties and can lead to undesired behaviour if open handles are not tracked correctly. Workers should be designed to wait for all operations to finish before returning control to the main thread, if any background operations are still running `async` can be of help (e.g. for cache warming, etc). ::: - `resourceLimits`: (`object`) See [Node.js new Worker options](https://nodejs.org/api/worker_threads.html#worker_threads_new_worker_filename_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. - `stackSizeMb` : (`number`) The default maximum stack size for the thread. Small values may lead to unusable Worker instances. Default: 4 - `env`: (`object`) If set, specifies the initial value of `process.env` inside the worker threads. See [Node.js new Worker options](https://nodejs.org/api/worker_threads.html#worker_threads_new_worker_filename_options) for details. - `argv`: (`any[]`) List of arguments that will be stringified and appended to `process.argv` in the worker. See [Node.js new Worker options](https://nodejs.org/api/worker_threads.html#worker_threads_new_worker_filename_options) for details. - `execArgv`: (`string[]`) List of Node.js CLI options passed to the worker. See [Node.js new Worker options](https://nodejs.org/api/worker_threads.html#worker_threads_new_worker_filename_options) for details. - `workerData`: (`any`) Any JavaScript value that can be cloned and made available as `require('piscina').workerData`. See [Node.js new Worker options](https://nodejs.org/api/worker_threads.html#worker_threads_new_worker_filename_options) for details. Unlike regular Node.js Worker Threads, `workerData` must not specify any value requiring a `transferList`. This is because the `workerData` will be cloned for each pooled worker. - `taskQueue`: (`TaskQueue`) By default, Piscina uses a first-in-first-out queue for submitted tasks. The `taskQueue` option can be used to provide an alternative implementation. See [Custom Task Queues](https://github.com/piscinajs/piscina#custom_task_queues) for additional detail. - `niceIncrement`: (`number`) An optional value that decreases priority for the individual threads, i.e. the higher the value, the lower the priority of the Worker threads. This value is used on Unix/Windows and requires the optional [`@napi-rs/nice`](https://npmjs.org/package/@napi-rs/nice) module to be installed. See [`nice(2)`](https://linux.die.net/man/2/nice) and [`SetThreadPriority`](https://learn.microsoft.com/en-us/windows/win32/api/processthreadsapi/nf-processthreadsapi-setthreadpriority) for more details. - `trackUnmanagedFds`: (`boolean`) An optional setting that, when `true`, will cause Workers to track file descriptors managed using `fs.open()` and `fs.close()`, and will close them automatically when the Worker exits. Defaults to `true`. (This option is only supported on Node.js 12.19+ and all Node.js versions higher than 14.6.0). - `closeTimeout`: (`number`) An optional time (in milliseconds) to wait for the pool to complete all in-flight tasks when `close()` is called. The default is `30000` - `recordTiming`: (`boolean`) By default, run and wait time will be recorded for the pool. To disable, set to `false`. - `workerHistogram`: (`boolean`) By default `false`. It will hint the Worker pool to record statistics for each individual Worker - `loadBalancer`: ([`PiscinaLoadBalancer`](#piscinaloadbalancer)) By default, Piscina uses a least-busy algorithm. The `loadBalancer` option can be used to provide an alternative implementation. See [Custom Load Balancers](../advanced-topics/loadbalancer.mdx) for additional detail. :::caution Use caution when setting resource limits. Setting limits that are too low may result in the `Piscina` worker threads being unusable. ::: :::info **Note on Explicit Resource Management**: Piscina does has support for `Symbol.dispose` and `Symbol.asyncDispose` for explicit resource management for its usage with the `using` keyword. This is only available on Node.js 24 and higher. For more information, see the [Explicit Resource Management](https://github.com/tc39/proposal-explicit-resource-management). ::: ## `PiscinaHistogram` The `PiscinaHistogram` allows you to access the histogram data for the pool of worker threads. It can be reset upon request in case of a need to clear the data. **Example**: ```js import { Piscina } from 'piscina'; const pool = new Piscina({ filename: resolve(__dirname, 'path/to/worker.js'), }); const firstBatch = []; for (let n = 0; n < 10; n++) { firstBatch.push(pool.run('42')); } await Promise.all(firstBatch); console.log(pool.histogram.runTime); // Print run time histogram summary console.log(pool.histogram.waitTime); // Print wait time histogram summary // If in need to reset the histogram data for a new set of tasks pool.histogram.resetRunTime(); pool.histogram.resetWaitTime(); const secondBatch = []; for (let n = 0; n < 10; n++) { secondBatch.push(pool.run('42')); } await Promise.all(secondBatch); // The histogram data will only contain the data for the second batch of tasks console.log(pool.histogram.runTime); console.log(pool.histogram.waitTime); ``` ### Interface: `PiscinaHistogram` - `runTime`: (`PiscinaHistogramSummary`) Run Time Histogram Summary. Time taken to execute a task. - `waitTime`: (`PiscinaHistogramSummary`) Wait Time Histogram Summary. Time between a task being submitted and the task starting to run. > **Note**: The histogram data is only available if `recordTiming` is set to `true`. ```ts type PiscinaHistogram = { runTime: PiscinaHistogramSummary; waitTime: PiscinaHistogramSummary; resetRunTime(): void; // Reset Run Time Histogram resetWaitTime(): void; // Reset Wait Time Histogram } ``` ### Interface: `PiscinaHistogramSummary` ```ts type PiscinaHistogramSummary = { average: number; mean: number; stddev: number; min: number; max: number; p0_001: number; p0_01: number; p0_1: number; p1: number; p2_5: number; p10: number; p25: number; p50: number; p75: number; p90: number; p97_5: number; p99: number; p99_9: number; p99_99: number; p99_999: number; } ``` ## `PiscinaLoadBalancer` The `PiscinaLoadBalancer` interface is used to implement custom load balancing algorithm that determines which worker thread should be assigned a task. > For more information, see [Custom Load Balancers](../advanced-topics/loadbalancer.mdx). ### Interface: `PiscinaLoadBalancer` ```ts type PiscinaLoadBalancer = ( task: PiscinaTask, // Task to be distributed workers: PiscinaWorker[] // Array of Worker instances ) => PiscinaWorker | null; // Worker instance to be assigned the task ``` If the `PiscinaLoadBalancer` returns `null`, `Piscina` will attempt to spawn a new worker, otherwise the task will be queued until a worker is available. ### Interface: `PiscinaTask` ```ts interface PiscinaTask { taskId: string; // Unique identifier for the task filename: string; // Filename of the worker module name: string; // Name of the worker function created: number; // Timestamp when the task was created isAbortable: boolean; // Indicates if the task can be aborted through AbortSignal } ``` ### Interface: `PiscinaWorker` ```ts interface PiscinaWorker { id: number; // Unique identifier for the worker currentUsage: number; // Number of tasks currently running on the worker isRunningAbortableTask: boolean; // Indicates if the worker is running an abortable task histogram: HistogramSummary | null; // Worker histogram terminating: boolean; // Indicates if the worker is terminating destroyed: boolean; // Indicates if the worker has been destroyed } ``` ### Example: Custom Load Balancer #### JavaScript ```js const { Piscina } = require('piscina'); function LeastBusyBalancer(opts) { const { maximumUsage } = opts; return (task, workers) => { let candidate = null; let checkpoint = maximumUsage; for (const worker of workers) { if (worker.currentUsage === 0) { candidate = worker; break; } if (worker.isRunningAbortableTask) continue; if (!task.isAbortable && worker.currentUsage < checkpoint) { candidate = worker; checkpoint = worker.currentUsage; } } return candidate; }; } const piscina = new Piscina({ loadBalancer: LeastBusyBalancer({ maximumUsage: 2 }), }); piscina .runTask({ filename: 'worker.js', name: 'default' }) .then((result) => console.log(result)) .catch((err) => console.error(err)); ``` #### TypeScript ```ts import { Piscina } from 'piscina'; function LeastBusyBalancer( opts: LeastBusyBalancerOptions ): PiscinaLoadBalancer { const { maximumUsage } = opts; return (task, workers) => { let candidate: PiscinaWorker | null = null; let checkpoint = maximumUsage; for (const worker of workers) { if (worker.currentUsage === 0) { candidate = worker; break; } if (worker.isRunningAbortableTask) continue; if (!task.isAbortable && worker.currentUsage < checkpoint) { candidate = worker; checkpoint = worker.currentUsage; } } return candidate; }; } const piscina = new Piscina({ loadBalancer: LeastBusyBalancer({ maximumUsage: 2 }), }); piscina .runTask({ filename: 'worker.js', name: 'default' }) .then((result) => console.log(result)) .catch((err) => console.error(err)); ``` --- ### Docs Api Reference Error Codes (`docs/docs/api-reference/error-codes.md`) --- id: Errors sidebar_position: 8 --- ## Errors The following section aggregates all the error codes thrown from `Piscina` and an explanation of them ### Constructor All errors thrown by `Piscina` follows are child of `PiscinaError` parent class. - `name`: (`string`) Name of the error - `code`: (`string`) Error code. Prefixed with `PISCINA_ERR_*` - `message`: (`string | null`) Summary of the cause of the error - `cause`: (`string | null`) Root cause of the error. ### Error Codes | Code | Description | | ----------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------- | | PISCINA_ERR_ABORT | Indicates a task has been aborted. Seek for the `cause` property to find more about the root cause. | | PISCINA_ERR_TASK_QUEUE_LIMIT | Task queue is at limit. No more tasks are being enqueued. Listen to the `drain` and `needsDrain` event to continue submitting tasks | | PISCINA_ERR_THREAD_TERMINATION | Thread is being terminated. | | PISCINA_ERR_VALIDATION | Validation for given values has failed. Review the `message` property for more information | | PISCINA_ERR_NO_TASK_QUEUE_AVAILABLE | Task queue has been disabled and workers are at full capacity. Listen to `drain` and `needsDrain` events to continue submitting tasks | | PISCINA_ERR_CLOSE_TIMEOUT | Pool has exceeded timeout for closing up. Pool will be destroyed immediately. | --- ### Docs Api Reference Event (`docs/docs/api-reference/event.md`) --- id: Events sidebar_position: 4 --- ## Event: `'error'` An `'error'` event is emitted by instances of this class when: - Uncaught exceptions occur inside Worker threads that do not currently handle tasks. - Unexpected messages are sent from from Worker threads. All other errors are reported by rejecting the `Promise` returned from `run()` or `runTask()`, including rejections reported by the handler function itself. ## Event: `'drain'` A `'drain'` event is emitted whenever the `queueSize` reaches `0`. ## Event: `'needsDrain'` Similar to [`Piscina#needsDrain`](https://github.com/piscinajs/piscina#property-needsdrain-readonly). ## Event: `'message'` A `'message'` event is emitted whenever a message is received from a worker thread. ## Event: `'workerCreate'` Event that is triggered when a new worker is created. As argument, it receives the worker instance. ## Event: `'workerDestroy'` Event that is triggered when a worker is destroyed. As argument, it receives the worker instance that has been destroyed. --- ### Docs Api Reference Interface (`docs/docs/api-reference/interface.md`) --- id: Interface sidebar_position: 7 --- ## Interface: `Transferable` Objects may implement the `Transferable` interface to create their own custom transferable objects. This is useful when an object being passed into or from a worker contains a deeply nested transferable object such as an `ArrayBuffer` or `MessagePort`. `Transferable` objects expose two properties inspected by Piscina to determine how to transfer the object. These properties are named using the special static `Piscina.transferableSymbol` and `Piscina.valueSymbol` properties: * The `Piscina.transferableSymbol` property provides the object (or objects) that are to be included in the `transferList`. * The `Piscina.valueSymbol` property provides a surrogate value to transmit in place of the `Transferable` itself. Both properties are required. For example: ```js const { move, transferableSymbol, valueSymbol } = require('piscina'); module.exports = () => { const obj = { a: { b: new Uint8Array(5); }, c: { new Uint8Array(10); }, get [transferableSymbol]() { // Transfer the two underlying ArrayBuffers return [this.a.b.buffer, this.c.buffer]; } get [valueSymbol]() { return { a: { b: this.a.b }, c: this.c }; } }; return move(obj); }; ``` --- ### Docs Api Reference Method (`docs/docs/api-reference/method.md`) --- id: Methods sidebar_position: 3 --- ## Method: `run(task[, options])` Schedules a task to be run on a Worker thread. * `task`: Any value. This will be passed to the function that is exported from `filename`. * `options`: * `transferList`: An optional lists of objects that is passed to [`postMessage()`] when posting `task` to the Worker, which are transferred rather than cloned. * `filename`: Optionally overrides the `filename` option passed to the constructor for this task. If no `filename` was specified to the constructor, this is mandatory. * `name`: Optionally overrides the exported worker function used for the task. * `abortSignal`: An `AbortSignal` instance. If passed, this can be used to cancel a task. If the task is already running, the corresponding `Worker` thread will be stopped. (More generally, any `EventEmitter` or `EventTarget` that emits `'abort'` events can be passed here.) Abortable tasks cannot share threads regardless of the `concurrentTasksPerWorker` options. This returns a `Promise` for the return value of the (async) function call made to the function exported from `filename`. If the (async) function throws an error, the returned `Promise` will be rejected with that error. If the task is aborted, the returned `Promise` is rejected with an error as well. ## Method: `runTask(task[, transferList][, filename][, abortSignal])` **Deprecated** -- Use `run(task, options)` instead. Schedules a task to be run on a Worker thread. * `task`: Any value. This will be passed to the function that is exported from `filename`. * `transferList`: An optional lists of objects that is passed to [`postMessage()`] when posting `task` to the Worker, which are transferred rather than cloned. * `filename`: Optionally overrides the `filename` option passed to the constructor for this task. If no `filename` was specified to the constructor, this is mandatory. * `signal`: An [`AbortSignal`][] instance. If passed, this can be used to cancel a task. If the task is already running, the corresponding `Worker` thread will be stopped. (More generally, any `EventEmitter` or `EventTarget` that emits `'abort'` events can be passed here.) Abortable tasks cannot share threads regardless of the `concurrentTasksPerWorker` options. This returns a `Promise` for the return value of the (async) function call made to the function exported from `filename`. If the (async) function throws an error, the returned `Promise` will be rejected with that error. If the task is aborted, the returned `Promise` is rejected with an error as well. ## Method: `destroy()` Stops all Workers and rejects all `Promise`s for pending tasks. This returns a `Promise` that is fulfilled once all threads have stopped. ## Method: `close([options])` * `options`: * `force`: A `boolean` value that indicates whether to abort all tasks that are enqueued but not started yet. The default is `false`. It stops all Workers gracefully. This returns a `Promise` that is fulfilled once all tasks that were started have completed and all threads have stopped. This method is similar to `destroy()`, but with the difference that `close()` will wait for the worker tasks to finish, while `destroy()` will abort them immediately. --- ### Docs Api Reference Property (`docs/docs/api-reference/property.md`) --- id: Properties sidebar_position: 5 --- ## Property: `completed` (readonly) The current number of completed tasks. ## Property: `duration` (readonly) The length of time (in milliseconds) since this `Piscina` instance was created. ## Property: `options` (readonly) A copy of the options that are currently being used by this instance. This object has the same properties as the options object passed to the constructor. ## Property: `runTime` (readonly) A histogram summary object summarizing the collected run times of completed tasks. All values are expressed in milliseconds. * `runTime.average` {`number`} The average run time of all tasks * `runTime.mean` {`number`} The mean run time of all tasks * `runTime.stddev` {`number`} The standard deviation of collected run times * `runTime.min` {`number`} The fastest recorded run time * `runTime.max` {`number`} The slowest recorded run time All properties following the pattern `p{N}` where N is a number (e.g. `p1`, `p99`) represent the percentile distributions of run time observations. For example, `p99` is the 99th percentile indicating that 99% of the observed run times were faster or equal to the given value. ```js { average: 1880.25, mean: 1880.25, stddev: 1.93, min: 1877, max: 1882.0190887451172, p0_001: 1877, p0_01: 1877, p0_1: 1877, p1: 1877, p2_5: 1877, p10: 1877, p25: 1877, p50: 1881, p75: 1881, p90: 1882, p97_5: 1882, p99: 1882, p99_9: 1882, p99_99: 1882, p99_999: 1882 } ``` ## Property: `threads` (readonly) An Array of the `Worker` instances used by this pool. ## Property: `idleThreads` (readonly) The current number of ready (warmed up) Worker threads that are not running any task. ## Property: `queueSize` (readonly) The current number of tasks waiting to be assigned to a Worker thread. ## Property: `needsDrain` (readonly) Boolean value that specifies whether the pool requires draining before processing new tasks. This is `true` when there is not enough capacity to process more tasks without exceeding the maximum queue size or within the maximum number of threads. This is often calculated based on the number of threads (current and pending if number of threads is below `maxThreads` threshold), the maximum queue size, and the current number of in-flight tasks. This property is helpful to make decisions towards creating backpressure. > **Note**: Piscina does not buffer tasks beyond the `maxQueue` threshold, so this property is not a guarantee that the pool will accept more tasks. ## Property: `utilization` (readonly) A point-in-time ratio comparing the approximate total mean run time of completed tasks to the total runtime capacity of the pool. A pools runtime capacity is determined by multiplying the `duration` by the `options.maxThread` count. This provides an absolute theoretical maximum aggregate compute time that the pool would be capable of. The approximate total mean run time is determined by multiplying the mean run time of all completed tasks by the total number of completed tasks. This number represents the approximate amount of time the pool as been actively processing tasks. The utilization is then calculated by dividing the approximate total mean run time by the capacity, yielding a fraction between `0` and `1`. ## Property: `waitTime` (readonly) A histogram summary object summarizing the collected times tasks spent waiting in the queue. All values are expressed in milliseconds. * `waitTime.average` {`number`} The average wait time of all tasks * `waitTime.mean` {`number`} The mean wait time of all tasks * `waitTime.stddev` {`number`} The standard deviation of collected wait times * `waitTime.min` {`number`} The fastest recorded wait time * `waitTime.max` {`number`} The longest recorded wait time All properties following the pattern `p{N}` where N is a number (e.g. `p1`, `p99`) represent the percentile distributions of wait time observations. For example, `p99` is the 99th percentile indicating that 99% of the observed wait times were faster or equal to the given value. ```js { average: 1880.25, mean: 1880.25, stddev: 1.93, min: 1877, max: 1882.0190887451172, p0_001: 1877, p0_01: 1877, p0_1: 1877, p1: 1877, p2_5: 1877, p10: 1877, p25: 1877, p50: 1881, p75: 1881, p90: 1882, p97_5: 1882, p99: 1882, p99_9: 1882, p99_99: 1882, p99_999: 1882 } ``` --- ### Docs Api Reference Static Property (`docs/docs/api-reference/static-property.md`) --- id: Static Properties and Methods sidebar_position: 6 --- ## Static property: `isWorkerThread` (readonly) Is `true` if this code runs inside a `Piscina` threadpool as a Worker. ## Static property: `version` (readonly) Provides the current version of this library as a semver string. ## Static method: `move(value)` By default, any value returned by a worker function will be cloned when returned back to the Piscina pool, even if that object is capable of being transferred. The `Piscina.move()` method can be used to wrap and mark transferable values such that they will be transferred rather than cloned. The `value` may be any object supported by Node.js to be transferable (e.g. `ArrayBuffer`, any `TypedArray`, or `MessagePort`), or any object implementing the `Transferable` interface. ```js const { move } = require('piscina'); module.exports = () => { return move(new ArrayBuffer(10)); } ``` The `move()` method will throw if the `value` is not transferable. The object returned by the `move()` method should not be set as a nested value in an object. If it is used, the `move()` object itself will be cloned as opposed to transferring the object it wraps. --- ### Docs Examples Category .Json (`docs/docs/examples/_category_.json`) { "label": "Examples", "position": 3, "link": { "type": "generated-index", "description": "Here are more ways to integrate Piscina into your projects. " } } --- ### Docs Examples Abort (`docs/docs/examples/abort.mdx`) --- id: Aborting Tasks sidebar_position: 1 --- import { WorkerWrapperComponent } from '@site/src/components/WorkerWrapper.mdx'; import Tabs from '@theme/Tabs'; import TabItem from '@theme/TabItem'; Piscina allows submitted tasks to be cancelled even if the task has already been submitted to a worker and is being actively processed. This can be used, for instance, to cancel tasks that are taking too long to run. In this example, we'll simulate a long running task using `setTimeout`: ```js title="worker.js" tab={"label":"Javascript"} 'use strict'; const { promisify } = require('util'); const sleep = promisify(setTimeout); module.exports = async ({ a, b }) => { await sleep(10000); return a + b; }; ``` ```js title="worker.ts" tab={"label":"Typescript"} import { promisify } from 'util'; import { resolve } from 'path'; export const filename = resolve(__filename); const sleep = promisify(setTimeout); interface OperationInput { a: number; b: number; } export default async ({ a, b }: OperationInput): Promise => { await sleep(10000); return a + b; }; ``` Tasks can be canceled using an `AbortController` or an `EventEmitter`. ## Using AbortController In this example, the `AbortController` instance is passed as the signal option to the `piscina.run` method. After submitting the task, we call `abortController.abort()` to cancel the task. ```js "use strict"; const Piscina = require("piscina"); const { AbortController } = require("abort-controller"); const { resolve } = require("path"); const piscina = new Piscina({ filename: resolve(__dirname, "worker.js"), }); (async function () { const abortController = new AbortController(); try { const task = piscina.run( { a: 4, b: 6 }, { signal: abortController.signal } ); abortController.abort(); await task; } catch (err) { console.log("The task was cancelled"); } })(); ```` ```js import Piscina from "piscina"; // Adjust the import path as necessary import { AbortController } from "abort-controller"; import { resolve } from "path"; import { filename } from "./worker.js"; const piscina = new Piscina({ filename: resolve(__dirname, "./workerWrapper.js"), workerData: { fullpath: filename }, }); (async () => { const abortController = new AbortController(); try { const task = piscina.run( { a: 4, b: 6 }, { signal: abortController.signal } ); abortController.abort(); // Cancel the task await task; // This will now throw due to the task being cancelled } catch (err) { console.log("The task was cancelled"); } })(); ``` ## Using EventEmitter The `EventEmitter` instance is passed as the signal option to the `piscina.run` method. ```js 'use strict'; const Piscina = require('../..'); 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 cancelled'); } })(); ``` ```js import Piscina from 'piscina'; import { resolve } from 'path'; import { filename } from './worker.js'; import { EventEmitter } from 'events'; const piscina = new Piscina({ filename: resolve(__dirname, './workerWrapper.js'), workerData: { fullpath: filename } }); (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 cancelled'); } })(); ``` ## Using a Timer with EventEmitter ```js 'use strict'; const Piscina = require('../..'); const EventEmitter = require('events'); const { resolve } = require('path'); const piscina = new Piscina({ filename: resolve(__dirname, 'worker.js') }); (async function () { const ee = new EventEmitter(); // Use a timer to limit task processing length const t = setTimeout(() => ee.emit('abort'), 500); try { await piscina.run({ a: 4, b: 6 }, { signal: ee }); } catch (err) { console.log('The task timed out'); } finally { // Clear the timeout in a finally to make sure // it is definitely cleared. clearTimeout(t); } })(); ``` ```js tab={"label":"Typescript","span":2} import Piscina from 'piscina'; import { resolve } from 'path'; import { filename } from './worker.js'; import { EventEmitter } from 'events'; const piscina = new Piscina({ filename: resolve(__dirname, './workerWrapper.js'), workerData: { fullpath: filename } }); (async function () { const ee = new EventEmitter(); // Use a timer to limit task processing length const t = setTimeout(() => ee.emit('abort'), 500); try { await piscina.run({ a: 4, b: 6 }, { signal: ee }); } catch (err) { console.log('The task timed out'); } finally { // Clear the timeout in a finally to make sure // it is definitely cleared. clearTimeout(t); } })(); ``` ## Using a Timer with AbortController ```js tab={"label":"Javascript"} 'use strict'; const Piscina = require('../..'); const { resolve } = require('path'); const piscina = new Piscina({ filename: resolve(__dirname, 'worker.js') }); (async function () { // Using the new built-in Node.js AbortController // Node.js version 15.0 or higher const ac = new AbortController(); // Use a timer to limit task processing length const t = setTimeout(() => ac.abort(), 500); try { await piscina.run({ a: 4, b: 6 }, { signal: ac.signal }); } catch (err) { console.log('The task timed out'); } finally { // Clear the timeout in a finally to make sure // it is definitely cleared. clearTimeout(t); } })(); ``` ```js tab={"label":"Typescript","span":2} import Piscina from 'piscina'; import { resolve } from 'path'; import { filename } from './worker.js'; import { EventEmitter } from 'events'; const piscina = new Piscina({ filename: resolve(__dirname, './workerWrapper.js'), workerData: { fullpath: filename } }); (async function () { // Using the new built-in Node.js AbortController // Node.js version 15.0 or higher const ac = new AbortController(); // Use a timer to limit task processing length const t = setTimeout(() => ac.abort(), 500); try { await piscina.run({ a: 4, b: 6 }, { signal: ac.signal }); } catch (err) { console.log('The task timed out'); } finally { // Clear the timeout in a finally to make sure // it is definitely cleared. clearTimeout(t); } })(); ``` You can also check out this example on [github](https://github.com/piscinajs/piscina/tree/current/examples/abort). --- ### Docs Examples Async Load (`docs/docs/examples/async_load.mdx`) --- id: Async Load sidebar_position: 2 --- import { WorkerWrapperComponent } from "@site/src/components/WorkerWrapper.mdx"; import Tabs from "@theme/Tabs"; import TabItem from "@theme/TabItem"; Piscina supports asynchronously loaded workers. This feature allows you to perform asynchronous operations during worker initialization, such as loading configurations or establishing database connections. For example, the main script below creates a Piscina pool and runs tasks from two worker files. ```js title="main.js" "use strict"; const Piscina = require("../.."); const pool = new Piscina(); const { resolve } = require("path"); (async () => { await Promise.all([ pool.run({}, { filename: resolve(__dirname, "worker") }), pool.run({}, { filename: resolve(__dirname, "worker.mjs") }), ]); })(); ```` ```javascript title="main.ts" import Piscina from "piscina"; import { resolve } from "path"; import { filename } from "./worker"; const pool = new Piscina({ workerData: { fullpath: filename }, }); (async () => { try { await Promise.all([ pool.run( {}, { name: "callSleep", filename: resolve(__dirname, "./workerWrapper.js"), } ), pool.run({}, { filename: resolve(__dirname, "worker.mjs") }), ]); } catch (error) { console.log(error); } })(); ``` Both worker files demonstrate asynchronous loading. They use a sleep function to simulate an asynchronous operation that takes 500 milliseconds. ```js title="worker.js" "use strict"; const { promisify } = require("util"); const sleep = promisify(setTimeout); module.exports = (async () => { await sleep(500); return () => console.log("hello from an async loaded CommonJS worker"); })(); ``` ```js title="worker.ts" import { promisify } from "util"; import { resolve } from "path"; export const filename = resolve(__filename); const sleep = promisify(setTimeout); export async function callSleep(): Promise { await sleep(500); return console.log("hello from an async loaded TypeScript worker"); } ``` In `worker.mjs`: ```js title="worker.mjs" // eslint-disable-next-line no-eval import util from "util"; const sleep = util.promisify(setTimeout); async function load() { await sleep(500); return () => console.log("hello from an async loaded ESM worker"); } export default load(); ``` You can also check out this example on [github](https://github.com/piscinajs/piscina/tree/current/examples/async_load). --- ### Docs Examples Broadcast (`docs/docs/examples/broadcast.mdx`) --- id: Broadcast sidebar_position: 3 --- import Tabs from '@theme/Tabs'; import TabItem from '@theme/TabItem'; import { WorkerWrapperComponent } from '@site/src/components/WorkerWrapper.mdx' In this example, we create a Piscina instance that uses BroadcastChannel(Node v18+) to implement broadcast communication. The main thread sends a message, and other threads the receive message. ```javascript title="index.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(); ``` ```javascript title="worker.js" 'use strict'; const { BroadcastChannel } = require('worker_threads'); module.exports = async (thread) => { const bc = new BroadcastChannel('my_channel'); bc.onmessage = (event) => { console.log(thread + ' Received from:' + event.data); }; await new Promise((resolve) => { setTimeout(resolve, 2000); }); }; ``` You can also check out this example on [github](https://github.com/piscinajs/piscina/tree/current/examples/broadcast). --- ### Docs Examples Electron (`docs/docs/examples/electron.mdx`) --- id: Electron sidebar_position: 4 --- import { WorkerWrapperComponent } from "@site/src/components/WorkerWrapper.mdx"; import Tabs from "@theme/Tabs"; import TabItem from "@theme/TabItem"; Using workers and pooling within your application can greatly enhance performance and stability. The renderer is processed on the main thread, so any processor intensive tasks are going to cause hangups for users, either delivering a poor UX or in many cases, causing the dreaded `pthread_kill` and crashing the app. Implementing workers is relatively straight-forward in development, but you might run into some issues when packaging your build. ## electron-vite NB: While Electron / electron-vite officially support ESM, with the way that rollup packages the worker, electron ends up not being able read the worker. You must be using CJS in electron to read the worker modules. Besides, the bytecode plugin for obfuscating your code only works with CJS too. In your main file, import the path to your worker (not the wrapper - no wrapper is needed because of rollup) by appending `\*?modulePath` to the end of the path. ```typescript import workerPath from './worker?modulePath'; ``` During compilation, rollup will now automatically create a separate file for the worker in your main folder output at the given path. The path above resolves, and we can now use this in Piscina at run time. Create a pool with Piscina, and use the imported path instead. For those using TypeScript, the filename export referenced in the guide is not needed. ```typescript title="index.ts" const pool = new Piscina({ filename: workerPath, workerData: { userPath: path.join(app.getPath("userData")), }, }); ``` Electron-specific modules are NOT accessible inside workers. If you need access to information that electron provides, (e.g. `app.getPath("userData")`), you'll need to pass in that data as workerData. That's it! You can now use pool like normal in the Piscina docs. For further help, reference the worker section of the docs from electron-vite. https://electron-vite.org/guide/dev#worker-threads --- ### Docs Examples Es Module (`docs/docs/examples/es-module.mdx`) --- id: ES Module sidebar_position: 5 --- import { WorkerWrapperComponent } from "@site/src/components/WorkerWrapper.mdx"; import Tabs from "@theme/Tabs"; import TabItem from "@theme/TabItem"; Piscina supports ES modules (ESM) out of the box. This example demonstrates how to use Piscina with ES modules in both the main script and the worker file. ```javascript title="index.mjs" import { Piscina } from "piscina"; const piscina = new Piscina({ filename: new URL("./worker.mjs", import.meta.url).href, }); (async () => { const result = await piscina.run({ a: 4, b: 6 }); console.log(result); })(); // Prints 10 ``` ```typescript title="index.ts" import Piscina from "piscina"; import { resolve } from "path"; import { filename } from "./worker"; const piscina = new Piscina({ filename: resolve(__dirname, "./workerWrapper.js"), workerData: { fullpath: filename }, }); (async () => { const result = await piscina.run({ a: 4, b: 6 }); console.log(result); })(); ``` The worker file is a simple ES module that exports a default function: ```javascript title="worker.mjs" export default ({ a, b }) => { return a + b; }; ``` ```typescript title="worker.ts" interface Input { a: number; b: number; } export default ({ a, b }: Input): number => { return a + b; }; ``` You can also check out this example on [github](https://github.com/piscinajs/piscina/tree/current/examples/es-module). --- ### Docs Examples Message Port (`docs/docs/examples/message-port.mdx`) --- id: Message Port sidebar_position: 6 --- import { WorkerWrapperComponent } from "@site/src/components/WorkerWrapper.mdx"; import Tabs from "@theme/Tabs"; import TabItem from "@theme/TabItem"; Worker threads can receive `MessagePort` objects, enabling direct communication channels with the main thread. This feature is useful for scenarios that require continuous communication or transfer of large data sets between threads. The example below shows how to use `MessagePort` with Piscina. ```javascript title="index.js" 'use strict'; const Piscina = require('piscina'); const { resolve } = require('path'); const { MessageChannel } = require('worker_threads'); const piscina = new Piscina({ filename: resolve(__dirname, 'worker.js') }); (async function () { const channel = new MessageChannel(); channel.port2.on('message', (message) => { console.log(message); channel.port2.close(); }); await piscina.run({ port: channel.port1 }, { transferList: [channel.port1] }); })(); ``` ```typescript title="index.ts" import Piscina from "piscina"; import { resolve } from "path"; import { filename } from "./worker"; import { TransferListItem } from "piscina/dist/types"; const piscina = new Piscina({ filename: resolve(__dirname, "./workerWrapper.js"), workerData: { fullpath: filename }, }); interface CustomMessagePort extends MessagePort { on(event: string, listener: (message: any) => void): void; } (async function () { const channel = new MessageChannel(); (channel.port2 as CustomMessagePort).on('message', (message: string) => { console.log(message); channel.port2.close(); }); await piscina.run({ port: channel.port1 }, { transferList:[channel.port1] as TransferListItem}); })(); ``` The worker file receives the `MessagePort` and uses it to send a message back to the main thread: ```javascript title="worker.js" 'use strict'; module.exports = ({ port }) => { port.postMessage('hello from the worker pool'); }; ``` ```typescript title="worker.ts" import { MessagePort } from 'worker_threads'; interface WorkerInput { port: MessagePort; } export default ({ port }: WorkerInput): void => { port.postMessage('hello from the worker pool'); }; ``` You can also check out this example on [github](https://github.com/piscinajs/piscina/tree/current/examples/message_port). --- ### Docs Examples Messages (`docs/docs/examples/messages.mdx`) --- id: Messages sidebar_position: 7 --- import { WorkerWrapperComponent } from "@site/src/components/WorkerWrapper.mdx"; import Tabs from "@theme/Tabs"; import TabItem from "@theme/TabItem"; Piscina allows workers to send messages back to the main thread using the `parentPort.postMessage()` method. This can be useful for sending progress updates, or intermediate results during the execution of a long-running task. ```javascript title="index.js" 'use strict'; const Piscina = require('piscina'); const { resolve } = require('path'); const piscina = new Piscina({ filename: resolve(__dirname, 'worker.js') }); (async function () { piscina.on('message', (event) => { console.log('Message received from worker: ', event); }); await piscina.run(); })(); ``` ```typescript title="index.ts" import Piscina from "piscina"; import { resolve } from "path"; import { filename } from "./worker"; import { TransferListItem } from "piscina/dist/types"; const piscina = new Piscina({ filename: resolve(__dirname, "./workerWrapper.js"), workerData: { fullpath: filename }, }); (async function () { piscina.on('message', (event: string) => { console.log('Message received from worker: ', event); }); await piscina.run({}); })(); ``` ```javascript title="worker.js" 'use strict'; const { parentPort } = require('worker_threads'); module.exports = () => { parentPort.postMessage('hello from the worker pool'); }; ``` ```typescript title="worker.ts" import { parentPort } from 'worker_threads'; export default (): void => { parentPort?.postMessage('hello from the worker pool'); }; ``` You can also check out this example on [github](https://github.com/piscinajs/piscina/tree/current/examples/messages). --- ### Docs Examples Move (`docs/docs/examples/move.mdx`) --- id: Move sidebar_position: 8 --- import { WorkerWrapperComponent } from "@site/src/components/WorkerWrapper.mdx"; import Tabs from "@theme/Tabs"; import TabItem from "@theme/TabItem"; Piscina provides a `move()` function that allows the transfer of data between the main thread and worker threads. The example below will show you how to use `Piscina.move()` to transfer `ArrayBuffer` without cloning, which can significantly improve performance for large data transfers. ```javascript title="index.js" const Piscina = require('piscina'); const { resolve } = require('path'); const pool = new Piscina({ filename: resolve(__dirname, 'worker.js'), idleTimeout: 1000 }); (async () => { // The task will transfer an ArrayBuffer // back to the main thread rather than // cloning it. const u8 = await pool.run(Piscina.move(new Uint8Array(2))); console.log(u8.length); })(); ``` ```typescript title="index.ts" import Piscina from 'piscina'; import { resolve } from 'path'; const pool = new Piscina({ filename: resolve(__dirname, 'worker.ts'), idleTimeout: 1000 }); (async () => { // The task will transfer an ArrayBuffer // back to the main thread rather than // cloning it. const u8 = await pool.run(Piscina.move(new Uint8Array(2))); console.log(u8.length); })(); ``` The worker file uses `move()` to transfer data back to the main thread: ```javascript title="worker.js" const { move } = require('piscina'); module.exports = () => { // Using move causes the Uint8Array to be // transferred rather than cloned. return move(new Uint8Array(10)); }; ``` ```typescript title="worker.ts" import { resolve } from "path"; import { move } from "piscina"; export const filename = resolve(__filename); export default (): Uint8Array => { // Using move causes the Uint8Array to be // transferred rather than cloned. return move(new Uint8Array(10)) as Uint8Array; }; ``` You can also check out this example on [github](https://github.com/piscinajs/piscina/tree/current/examples/move). --- ### Docs Examples Multiple Workers One File (`docs/docs/examples/multiple-workers-one-file.mdx`) --- id: Multiple Workers in One File sidebar_position: 9 --- import { WorkerWrapperComponent } from "@site/src/components/WorkerWrapper.mdx"; import Tabs from "@theme/Tabs"; import TabItem from "@theme/TabItem"; Piscina allows you to define multiple worker functions within a single file. This approach can be useful when you have related tasks that you want to keep in one module. In the example below, for each task, `pool.run()` is called with two arguments: - The task data `({ a: 2, b: 3 })` - An options object specifying the name of the function to use. ```javascript title="index.js" 'use strict'; const Piscina = require('piscina'); const { resolve } = require('path'); const pool = new Piscina({ filename: resolve(__dirname, 'worker.js') }); (async () => { console.log(await Promise.all([ pool.run({ a: 2, b: 3 }, { name: 'add' }), pool.run({ a: 2, b: 3 }, { name: 'multiply' }) ])); })(); ``` ```javascript title="index.mjs" import { Piscina } from 'piscina'; const pool = new Piscina({ filename: new URL('./worker.mjs', import.meta.url).href }); console.log(await Promise.all([ pool.run({ a: 2, b: 3 }, { name: 'add' }), pool.run({ a: 2, b: 3 }, { name: 'multiply' }) ])); ``` ```typescript title="index.ts" import Piscina from 'piscina'; import { resolve } from 'path'; interface TaskInput { a: number; b: number; } const pool = new Piscina({ filename: resolve(__dirname, 'worker.ts') }); (async () => { console.log(await Promise.all([ pool.run({ a: 2, b: 3 }, { name: 'add' }), pool.run({ a: 2, b: 3 }, { name: 'multiply' }) ])); })(); ``` Here's the worker file that defines multiple functions: ```javascript title="worker.js" 'use strict'; function add ({ a, b }) { return a + b; } function multiply ({ a, b }) { return a * b; }; add.add = add; add.multiply = multiply; // The add function is the default handler, but // additional named handlers are exported. module.exports = add; ``` ```javascript title="worker.mjs" export function add ({ a, b }) { return a + b; } export function multiply ({ a, b }) { return a * b; }; ``` ```typescript title="worker.ts" import { resolve } from "path"; interface TaskInput { a: number; b: number; } function add({ a, b }: TaskInput): number { return a + b; } function multiply({ a, b }: TaskInput): number { return a * b; } export { add, multiply }; export const filename = resolve(__filename); ``` You can also check out this example on [github](https://github.com/piscinajs/piscina/tree/current/examples/multiple-workers-one-file). --- ### Docs Examples Multiple Workers (`docs/docs/examples/multiple-workers.mdx`) --- id: Multiple Workers sidebar_position: 10 --- import { WorkerWrapperComponent } from "@site/src/components/WorkerWrapper.mdx"; import Tabs from "@theme/Tabs"; import TabItem from "@theme/TabItem"; It is possible for a single Piscina pool to run multiple workers at the same time. To do so, pass the worker `filename` to `run` method rather than to the Piscina constructor. To work with multiple workers using Typescript, check out [multiple workers in one file](./multiple-workers-one-file.mdx). ```javascript title="index.js" "use strict"; const Piscina = require("piscina"); const { resolve } = require("path"); const pool = new Piscina(); (async () => { console.log( await Promise.all([ pool.run({ a: 2, b: 3 }, { filename: resolve(__dirname, "add_worker") }), pool.run( { a: 2, b: 3 }, { filename: resolve(__dirname, "multiply_worker") } ), ]) ); })(); ``` Here are the two worker files used in this example: ```javascript title="add_worker.js" "use strict"; module.exports = ({ a, b }) => a + b; ``` ```javascript title="multiply_worker.js" "use strict"; module.exports = ({ a, b }) => a * b; ``` You can also check out this example on [github](https://github.com/piscinajs/piscina/tree/current/examples/multiple-workers). --- ### Docs Examples N Api (`docs/docs/examples/n-api.mdx`) --- id: N-API Native Addon sidebar_position: 11 --- For CPU-intensive tasks or when implementing workers in languages such as C++ or Rust, you can leverage Piscina's support for native addons as worker implementations. ## Setup First, get the example from the [Piscina GitHub repository](https://github.com/piscinajs/piscina/tree/current/examples/n-api). Then install the dependencies: ```console npm i ``` Build the native addon artifacts: ```console npm run prebuild ``` The `prebuild` command will build the binary artifacts for the native addon and will put them in the `prebuilds` folder. Because of how prebuilds work, we need to use an intermediate JavaScript file to load and export them. For this example native addon, you'll find that in the `examples` folder. The `index.js` illustrates how to load and use the native addon as the worker implementation: ```js title="index.js" const Piscina = require('piscina'); const { resolve } = require('path'); const pool = new Piscina({ filename: resolve(__dirname, 'example') }); (async () => { // Submit 5 concurrent tasks console.log(await Promise.all([ pool.run(), pool.run(), pool.run(), pool.run(), pool.run() ])); })(); ``` You can also check out this example on [github](https://github.com/piscinajs/piscina/tree/current/examples/n-api). --- ### Docs Examples Named (`docs/docs/examples/named.mdx`) --- id: Named Tasks sidebar_position: 12 --- import { WorkerWrapperComponent } from "@site/src/components/WorkerWrapper.mdx"; import Tabs from "@theme/Tabs"; import TabItem from "@theme/TabItem"; Piscina supports running named tasks within a single worker file. This example demonstrates how to use a dispatcher pattern to execute different operations based on the task name. ```javascript title="index.js" 'use strict'; const Piscina = require('piscina'); const { resolve } = require('path'); const { makeTask } = require('./helper'); const piscina = new Piscina({ filename: resolve(__dirname, 'worker.js') }); (async function () { const result = await Promise.all([ piscina.run(makeTask('add', 4, 6)), piscina.run(makeTask('sub', 4, 6)) ]); console.log(result); })(); ``` ```typescript title="index.ts" import Piscina from "piscina"; import { resolve } from "path"; import { filename } from "./worker"; import { makeTask } from "./helper"; const piscina = new Piscina({ filename: resolve(__dirname, "./workerWrapper.js"), workerData: { fullpath: filename }, }); (async function () { const result = await Promise.all([ piscina.run(makeTask('add', 4, 6)), piscina.run(makeTask('sub', 4, 6)) ]); console.log(result); })(); ``` The worker file uses a dispatcher to handle different operations: ```javascript title="worker.js" 'use strict'; const { dispatcher } = require('./helper'); module.exports = dispatcher({ add (a, b) { return a + b; }, sub (a, b) { return a - b; } }); ``` ```typescript title="worker.ts" import { dispatcher } from './helper'; export default dispatcher({ add (a: number, b: number): number { return a + b; }, sub (a: number, b: number): number { return a - b; } }); ``` The helper file provides utility functions for creating tasks and dispatching them: ```javascript title="helper.js" function makeTask (op, ...args) { return { op, args }; } function dispatcher (obj) { return async ({ op, args }) => { return await obj[op](...args); }; } module.exports = { dispatcher, makeTask }; ``` ```typescript title="helper.ts" type Task = { op: string; args: any[]; }; export function makeTask (op: string, ...args: any[]): Task { return { op, args }; } export function dispatcher (obj: Record) { return async ({ op, args }: Task) => { return await obj[op](...args); }; } ``` You can also check out this example on [github](https://github.com/piscinajs/piscina/tree/current/examples/named). --- ### Docs Examples Progress (`docs/docs/examples/progress.mdx`) --- id: Progress sidebar_position: 13 --- import {WorkerWrapperComponent} from '@site/src/components/WorkerWrapper.mdx'; import Tabs from "@theme/Tabs"; import TabItem from "@theme/TabItem"; You can track the progress of long-running tasks using a `MessageChannel`. This can be adapted for use cases where you need to track the progress of long-running tasks executed in worker threads. The example below shows how to add a progress bar for tasks running in worker threads. To get started, install the required dependency: ```bash npm install progress ``` The main script creates progress bars for each task and updates them based on messages received from the workers. ```javascript title="index.js" 'use strict'; const Piscina = require('piscina'); const { resolve } = require('path'); const ProgressBar = require('progress'); // Initialize Piscina with the worker file const piscina = new Piscina({ filename: resolve(__dirname, 'worker.js') }); // Function to create and run a task with progress tracking async function task(a, b) { // Create a progress bar for this task const bar = new ProgressBar(':bar [:current/:total]', { total: b }); // Set up a MessageChannel for communication with the worker const mc = new MessageChannel(); // Update the progress bar when a message is received from the worker mc.port2.onmessage = () => bar.tick(); // Prevent the port from keeping the event loop alive mc.port2.unref(); // Run the task in a worker, passing one port of the channel return await piscina.run({ a, b, port: mc.port1 }, { transferList: [mc.port1] }); } // Run multiple tasks concurrently Promise.all([ task(0, 50), task(0, 25), task(0, 90) ]).catch((err) => console.error(err)); ``` ```typescript title="index.ts" import Piscina from "piscina"; import { resolve } from "path"; import { filename } from "./worker"; import ProgressBar from 'progress'; const piscina = new Piscina({ filename: resolve(__dirname, "./workerWrapper.js"), workerData: { fullpath: filename }, }); async function task(a: number, b: number): Promise { // Create a progress bar for this task const bar = new ProgressBar(':bar [:current/:total]', { total: b }); // Set up a MessageChannel for communication with the worker const mc = new MessageChannel(); // Update the progress bar when a message is received from the worker mc.port2.onmessage = () => bar.tick(); // Prevent the port from keeping the event loop alive mc.port2.unref(); // Run the task in a worker, passing one port of the channel await piscina.run({ a, b, port: mc.port1 }, { transferList: [mc.port1], name: 'progressTask' }); } // Run multiple tasks concurrently Promise.all([ task(0, 50), task(0, 25), task(0, 90) ]).catch((err) => console.error(err)); ``` In the worker file, the worker receives the `port` as part of its input. As it processes the task, it sends messages through the port to indicate progress. When the task is complete, it closes the port. ```javascript title="worker.js" const { setTimeout } = require('timers/promises'); // Worker function that simulates a time-consuming task module.exports = async ({ a, b, port }) => { for (let n = a; n < b; n++) { // Simulate some work await setTimeout(10); // Send a message to indicate progress port.postMessage(n); } // Close the port when the task is complete port.close(); }; ``` ```typescript title="worker.ts" import { resolve } from "path"; import { setTimeout } from 'timers/promises'; export const filename = resolve(__filename); interface TaskInput { a: number; b: number; port: MessagePort; } // Worker function that simulates a time-consuming task export async function progressTask({ a, b, port }: TaskInput): Promise { for (let n = a; n < b; n++) { // Simulate some work await setTimeout(10); // Send a message to indicate progress port.postMessage(n); } // Close the port when the task is complete port.close(); } ``` When you run this code, you should see progress bars for three concurrent tasks updating in real-time. You can also check out this example on [github](https://github.com/piscinajs/piscina/tree/current/examples/progress). --- ### Docs Examples React Ssr (`docs/docs/examples/react-ssr.mdx`) --- id: React Server Side Rendering sidebar_position: 14 --- import Tabs from "@theme/Tabs"; import TabItem from "@theme/TabItem"; import { WorkerWrapperComponent } from "@site/src/components/WorkerWrapper.mdx"; This example explains how to use Piscina for server-side rendering (SSR) of React components. We'll compare a pooled version using Piscina with an unpooled version to highlight the benefits of using a thread pool for SSR. To get started, make sure you have the following dependencies installed: ```bash npm install fastify fastify-piscina react react-dom lorem-ipsum ``` If you are using TypeScript in your project, also install: ```bash npm install -D typescript @types/react @types/react-dom @types/node ``` The function in the worker file uses `ReactDOMServer.renderToString()` to render two React components: `Greeting` and `Lorem`. ```javascript title="worker.js" 'use strict'; const React = require('react'); const ReactDOMServer = require('react-dom/server'); const { Greeting, Lorem } = require('./components'); module.exports = ({ name }) => { return `
${ ReactDOMServer.renderToString(React.createElement(Greeting, { name })) }
${ ReactDOMServer.renderToString(React.createElement(Lorem)) } `; }; ``` ```typescript title="worker.ts" import React from 'react'; import ReactDOMServer from 'react-dom/server'; import { Greeting, Lorem } from './components'; export const filename = __filename; interface WorkerInput { name: string; } export function renderPage({ name }: WorkerInput): string { return `
${ ReactDOMServer.renderToString(React.createElement(Greeting, { name })) }
${ ReactDOMServer.renderToString(React.createElement(Lorem)) } `; } ``` Here are the components used in this example: ```javascript title="components/greeting.js" 'use strict'; const React = require('react'); class Greeting extends React.Component { render () { return React.createElement('div', null, 'hello ' + this.props.name); } } module.exports = Greeting; ``` ```javascript title="components/lorem.js" 'use strict'; const React = require('react'); const { LoremIpsum } = require('lorem-ipsum'); class Paragraph extends React.Component { #lorem; constructor (props) { super(props); this.#lorem = new LoremIpsum({ sentencesPerParagraph: { max: 8, min: 4 }, wordsPerSentence: { max: 16, min: 4 } }); } render () { return React.createElement('div', null, this.#lorem.generateParagraphs(1)); } } class Lorem extends React.Component { render () { const children = []; for (let n = 0; n < Math.floor(Math.random() * 50); n++) { children.push(React.createElement(Paragraph, { key: n })); } return React.createElement('div', null, children); } } module.exports = Lorem; ``` ```typescript title="components/greeting.tsx" import React from 'react'; interface GreetingProps { name: string; } class Greeting extends React.Component { render() { return React.createElement('div', null, 'hello ' + this.props.name); } } export default Greeting; ``` ```typescript title="components/lorem.tsx" import React from 'react'; import { LoremIpsum } from 'lorem-ipsum'; class Paragraph extends React.Component { private lorem: LoremIpsum; constructor(props: {}) { super(props); this.lorem = new LoremIpsum({ sentencesPerParagraph: { max: 8, min: 4 }, wordsPerSentence: { max: 16, min: 4 } }); } render() { return React.createElement('div', null, this.lorem.generateParagraphs(1)); } } class Lorem extends React.Component { render() { const children = []; for (let n = 0; n < Math.floor(Math.random() * 50); n++) { children.push(React.createElement(Paragraph, { key: n })); } return React.createElement('div', null, children); } } export default Lorem; ``` ### Pooled Version In pooled version of the code, we set up a Fastify server with Piscina integration. We register the `fastify-piscina` plugin, configuring it to use the `worker.js` file with 6 threads. The root route (`/`) uses `fastify.runTask()` to execute the rendering in a worker thread. ```javascript title="pooled.js" 'use strict'; const fastify = require('fastify')(); const { resolve } = require('path'); fastify.register(require('fastify-piscina'), { filename: resolve(__dirname, 'worker.js'), execArgv: [], minThreads: 6, maxThreads: 6 }); // Declare a route fastify.get('/', async () => fastify.runTask({ name: 'James' })); // Run the server! const start = async () => { try { await fastify.listen({ port: 3000 }); } catch (err) { process.exit(1); } }; start(); process.on('SIGINT', () => { const waitTime = fastify.piscina.waitTime; console.log('\nMax Queue Wait Time:', waitTime.max); console.log('Mean Queue Wait Time:', waitTime.mean); process.exit(0); }); ``` ```typescript title="pooled.ts" import { resolve } from 'path'; import { filename } from './worker'; const fastify = require('fastify')(); fastify.register(require('fastify-piscina'), { filename: resolve(__dirname, 'workerWrapper.js'), workerData: { fullpath: filename }, execArgv: [], minThreads: 6, maxThreads: 6 }); // Declare a route fastify.get('/', async () => fastify.runTask({ name: 'James' }, { name: 'renderPage' })); // Run the server! const start = async () => { try { await fastify.listen({ port: 3000 }); } catch (err) { process.exit(1); } }; start(); process.on('SIGINT', () => { const waitTime = (fastify as any).piscina.waitTime; console.log('\nMax Queue Wait Time:', waitTime.max); console.log('Mean Queue Wait Time:', waitTime.mean); process.exit(0); }); ``` ### Unpooled Version In the unpooled version of the server, we perform the same rendering as the pooled version, but with no thread pool or worker management. The root route (`/`) renders the React components synchronously in the request handler. ```javascript title="unpooled.js" 'use strict'; const fastify = require('fastify')(); const React = require('react'); const ReactDOMServer = require('react-dom/server'); const { Greeting, Lorem } = require('./components'); // Declare a route fastify.get('/', async () => { const name = 'James'; return `
${ ReactDOMServer.renderToString(React.createElement(Greeting, { name })) }
${ ReactDOMServer.renderToString(React.createElement(Lorem)) } `; }); // Run the server! const start = async () => { try { await fastify.listen({ port: 3000 }); } catch (err) { process.exit(1); } }; start(); ``` ```typescript title="unpooled.ts" import fastify from 'fastify'; import React from 'react'; import ReactDOMServer from 'react-dom/server'; import { Greeting, Lorem } from './components'; const server = fastify(); // Declare a route server.get('/', async () => { const name = 'James'; return `
${ ReactDOMServer.renderToString(React.createElement(Greeting, { name })) }
${ ReactDOMServer.renderToString(React.createElement(Lorem)) } `; }); // Run the server! const start = async () => { try { await server.listen(3000); } catch (err) { process.exit(1); } }; start(); ``` Run either the pooled or unpooled version: ``` node pooled.js ``` ``` node unpooled.js ``` Access `http://localhost:3000` in your browser to see the rendered page. ## Benchmarking Results We used `autocannon` to benchmark both the pooled and unpooled versions. Install autocannon: ```bash npm i autocannon ``` Then benchmark the results by running the script below: ```bash autocannon http://localhost:3000 ``` Here are the results: ### Pooled Version Result ``` Running 10s test @ http://localhost:3000 10 connections ┌─────────┬──────┬──────┬───────┬───────┬────────┬──────────┬────────┐ │ Stat │ 2.5% │ 50% │ 97.5% │ 99% │ Avg │ Stdev │ Max │ ├─────────┼──────┼──────┼───────┼───────┼────────┼──────────┼────────┤ │ Latency │ 2 ms │ 4 ms │ 44 ms │ 63 ms │ 6.9 ms │ 10.98 ms │ 190 ms │ └─────────┴──────┴──────┴───────┴───────┴────────┴──────────┴────────┘ ┌───────────┬────────┬────────┬─────────┬─────────┬─────────┬─────────┬────────┐ │ Stat │ 1% │ 2.5% │ 50% │ 97.5% │ Avg │ Stdev │ Min │ ├───────────┼────────┼────────┼─────────┼─────────┼─────────┼─────────┼────────┤ │ Req/Sec │ 657 │ 657 │ 1,225 │ 2,067 │ 1,348.9 │ 454.88 │ 657 │ ├───────────┼────────┼────────┼─────────┼─────────┼─────────┼─────────┼────────┤ │ Bytes/Sec │ 2.3 MB │ 2.3 MB │ 4.35 MB │ 7.12 MB │ 4.68 MB │ 1.55 MB │ 2.3 MB │ └───────────┴────────┴────────┴─────────┴─────────┴─────────┴─────────┴────────┘ Req/Bytes counts sampled once per second. # of samples: 10 13k requests in 10.07s, 46.8 MB read ``` ### Unpooled Version Result ``` Running 10s test @ http://localhost:3000 10 connections ┌─────────┬──────┬───────┬───────┬───────┬──────────┬──────────┬────────┐ │ Stat │ 2.5% │ 50% │ 97.5% │ 99% │ Avg │ Stdev │ Max │ ├─────────┼──────┼───────┼───────┼───────┼──────────┼──────────┼────────┤ │ Latency │ 8 ms │ 22 ms │ 63 ms │ 75 ms │ 25.03 ms │ 14.78 ms │ 165 ms │ └─────────┴──────┴───────┴───────┴───────┴──────────┴──────────┴────────┘ ┌───────────┬────────┬────────┬─────────┬─────────┬─────────┬────────┬────────┐ │ Stat │ 1% │ 2.5% │ 50% │ 97.5% │ Avg │ Stdev │ Min │ ├───────────┼────────┼────────┼─────────┼─────────┼─────────┼────────┼────────┤ │ Req/Sec │ 234 │ 234 │ 419 │ 590 │ 395.6 │ 127.26 │ 234 │ ├───────────┼────────┼────────┼─────────┼─────────┼─────────┼────────┼────────┤ │ Bytes/Sec │ 785 kB │ 785 kB │ 1.47 MB │ 1.97 MB │ 1.36 MB │ 439 kB │ 785 kB │ └───────────┴────────┴────────┴─────────┴─────────┴─────────┴────────┴────────┘ Req/Bytes counts sampled once per second. # of samples: 10 4k requests in 10.22s, 13.6 MB read ``` The pooled version using Piscina offers several advantages for server-side rendering: - It handles significantly more requests per second (1,348.9 vs 395.6 on average). - It has lower latency (6.9ms vs 25.03ms on average). - It processes more data (4.68 MB/sec vs 1.36 MB/sec on average). You can also check out this example on [github](https://github.com/piscinajs/piscina/tree/current/examples/react-ssr). --- ### Docs Examples Resource Limit (`docs/docs/examples/resource-limit.mdx`) --- id: Resource Limits sidebar_position: 15 --- import Tabs from '@theme/Tabs'; import TabItem from '@theme/TabItem'; import { WorkerWrapperComponent } from '@site/src/components/WorkerWrapper.mdx'; Piscina allows you to set resource limits on worker threads to prevent them from consuming excessive memory. The example below will show you how to configure and use resource limits to handle out-of-memory scenarios. The main file sets up a Piscina instance with specific resource limits: - `maxOldGenerationSizeMb`: Limits the maximum size of the old generation in the V8 heap to 16 MB. - `maxYoungGenerationSizeMb`: Limits the maximum size of the young generation in the V8 heap to 4 MB. - `codeRangeSizeMb`: Limits the size of the code range to 16 MB. ```javascript title="index.js" 'use strict'; const Piscina = require('piscina'); 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.run(); } catch (err) { console.log('Worker terminated due to resource limits'); strictEqual(err.code, 'ERR_WORKER_OUT_OF_MEMORY'); } })(); ``` ```typescript title="index.ts" import Piscina from 'piscina'; import { resolve } from 'path'; import { strictEqual } from 'assert'; import { filename } from './worker'; const piscina = new Piscina({ filename: resolve(__dirname, 'workerWrapper.js'), workerData: { fullpath: filename }, resourceLimits: { maxOldGenerationSizeMb: 16, maxYoungGenerationSizeMb: 4, codeRangeSizeMb: 16 } }); (async function () { try { await piscina.run({}, { name: 'memoryLeak' }); } catch (err) { console.log('Worker terminated due to resource limits'); strictEqual((err as any).code, 'ERR_WORKER_OUT_OF_MEMORY'); } })(); ``` The worker file contains a function that deliberately causes a memory leak by creating an infinitely growing array. This will eventually exceed the memory limits set in the main file. ```javascript title="worker.js" 'use strict'; module.exports = () => { const array = []; while (true) { array.push([array]); } }; ``` ```typescript title="worker.ts" export const filename = __filename; export function memoryLeak(): void { const array: any[] = []; while (true) { array.push([array]); } } ``` You can also check out this example on [github](https://github.com/piscinajs/piscina/tree/current/examples/resourceLimits). --- ### Docs Examples Scrypt (`docs/docs/examples/scrypt.mdx`) --- id: Scrypt sidebar_position: 16 --- import Tabs from '@theme/Tabs'; import TabItem from '@theme/TabItem'; import { WorkerWrapperComponent } from '@site/src/components/WorkerWrapper.mdx'; This example demonstrates the performance benefits of using Piscina for CPU-intensive cryptographic operations, specifically the scrypt key derivation function. It compares four different implementations: pooled and unpooled versions, each with both synchronous and asynchronous variants. ### Setup The monitor script measures event loop delay, helping us to understand the impact of each implementation on the main thread's responsiveness. ```javascript title="monitor.js" 'use strict'; const { monitorEventLoopDelay } = require('perf_hooks'); const { isMainThread } = require('worker_threads'); if (!isMainThread) return; const monitor = monitorEventLoopDelay({ resolution: 20 }); monitor.enable(); process.on('exit', () => { monitor.disable(); console.log('Main Thread Mean/Max/99% Event Loop Delay:', monitor.mean, monitor.max, monitor.percentile(99)); }); ``` ```js title="monitor.ts" import { monitorEventLoopDelay } from 'perf_hooks'; import { isMainThread } from 'worker_threads'; if (!isMainThread) process.exit(); const monitor = monitorEventLoopDelay({ resolution: 20 }); monitor.enable(); process.on('exit', () => { monitor.disable(); console.log('Main Thread Mean/Max/99% Event Loop Delay:', monitor.mean, monitor.max, monitor.percentile(99)); }); ``` ### Pooled Asynchronous The pooled versions use Piscina to distribute the scrypt operations across multiple worker threads, potentially improving performance on multi-core systems. ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ```javascript title="scrypt.js" // eslint-disable no-unused-vars 'use strict'; const crypto = require('crypto'); const { promisify } = require('util'); const scrypt = promisify(crypto.scrypt); const randomFill = promisify(crypto.randomFill); const salt = Buffer.allocUnsafe(16); module.exports = async function ({ input, keylen, N = 16384, r = 8, p = 1, maxmem = 32 * 1024 * 1024 }) { return (await scrypt( input, await randomFill(salt), keylen, { N, r, p, maxmem })).toString('hex'); }; ``` ```js title="pooled.ts" import { resolve } from 'path'; import { promisify } from 'util'; import { randomFill } from 'crypto'; import { performance } from 'perf_hooks'; import Piscina from 'piscina'; import { filename } from './scrypt'; // ... (performance observer setup) const piscina = new Piscina({ filename: resolve(__dirname, 'workerWrapper.js'), workerData: { fullpath: filename }, concurrentTasksPerWorker: 10 }); // ... (process exit handler for statistics) async function* generateInput() { let max = parseInt(process.argv[2] || '10', 10); const data = Buffer.allocUnsafe(10); while (max-- > 0) { yield promisify(randomFill)(data); } } (async function () { performance.mark('start'); const keylen = 64; for await (const input of generateInput()) { await piscina.run({ input, keylen }); } performance.mark('end'); performance.measure('start to end', 'start', 'end'); })(); ``` ```js title="scrypt.ts" import crypto from 'crypto'; import { promisify } from 'util'; const scrypt: any = promisify(crypto.scrypt); const randomFill = promisify(crypto.randomFill); const salt = Buffer.allocUnsafe(16); export default async function ({ input, keylen, N = 16384, r = 8, p = 1, maxmem = 32 * 1024 * 1024 }: { input: Buffer; keylen: number; N?: number; r?: number; p?: number; maxmem?: number; }): Promise { return ( await scrypt(input, await randomFill(salt), keylen, { N, r, p, maxmem }) ).toString('hex'); } export const filename = __filename; ``` ### Unpooled Asynchronous The asynchronous versions use `promisify` versions of `scrypt` and `randomFill`, which don't block the event loop but may have slightly higher overhead. ```javascript title="unpooled.js" 'use strict'; const crypto = require('crypto'); const { promisify } = require('util'); const randomFill = promisify(crypto.randomFill); const scrypt = promisify(crypto.scrypt); const { performance, PerformanceObserver } = require('perf_hooks'); const salt = Buffer.allocUnsafe(16); const obs = new PerformanceObserver((entries) => { console.log(entries.getEntries()[0].duration); }); obs.observe({ entryTypes: ['measure'] }); async function * generateInput () { let max = parseInt(process.argv[2] || 10); const data = Buffer.allocUnsafe(10); while (max-- > 0) { yield randomFill(data); } } (async function () { performance.mark('start'); const keylen = 64; for await (const input of generateInput()) { (await scrypt(input, await randomFill(salt), keylen)).toString('hex'); } performance.mark('end'); performance.measure('start to end', 'start', 'end'); })(); ``` ```js title="unpooled.ts" 'use strict'; import crypto from 'crypto'; import { promisify } from 'util'; const randomFill = promisify(crypto.randomFill); const scrypt:any = promisify(crypto.scrypt); const { performance, PerformanceObserver } = require('perf_hooks'); const salt = Buffer.allocUnsafe(16); const obs = new PerformanceObserver((entries: { getEntries: () => { duration: any; }[]; }) => { console.log(entries.getEntries()[0].duration); }); obs.observe({ entryTypes: ['measure'] }); async function* generateInput() { let max = parseInt(process.argv[2] || '10', 10); const data = Buffer.allocUnsafe(10); while (max-- > 0) { yield randomFill(data); } } (async function () { performance.mark('start'); const keylen = 64; for await (const input of generateInput()) { (await scrypt(input, await randomFill(salt), keylen)).toString('hex'); } performance.mark('end'); performance.measure('start to end', 'start', 'end'); })(); ``` ### Pooled Synchronous The synchronous versions use `scryptSync` and `randomFillSync`, which can be more efficient but may block the event loop. ```javascript title="pooled_sync.js" 'use strict'; const Piscina = require('../..'); const { resolve } = require('path'); const crypto = require('crypto'); const { promisify } = require('util'); const randomFill = promisify(crypto.randomFill); const { performance, PerformanceObserver } = require('perf_hooks'); const obs = new PerformanceObserver((entries) => { console.log(entries.getEntries()[0].duration); }); obs.observe({ entryTypes: ['measure'] }); const piscina = new Piscina({ filename: resolve(__dirname, 'scrypt_sync.js') }); process.on('exit', () => { const { runTime, waitTime } = piscina; console.log('Run Time Average:', runTime.average); console.log('Run Time Mean/Stddev:', runTime.mean, runTime.stddev); console.log('Run Time Min:', runTime.min); console.log('Run Time Max:', runTime.max); console.log('Wait Time Average:', waitTime.average); console.log('Wait Time Mean/Stddev:', waitTime.mean, waitTime.stddev); console.log('Wait Time Min:', waitTime.min); console.log('Wait Time Max:', waitTime.max); }); async function * generateInput () { let max = parseInt(process.argv[2] || 10); const data = Buffer.allocUnsafe(10); while (max-- > 0) { yield randomFill(data); } } (async function () { performance.mark('start'); const keylen = 64; for await (const input of generateInput()) { await piscina.run({ input, keylen }); } performance.mark('end'); performance.measure('start to end', 'start', 'end'); })(); ``` ```javascript title="scrypt_sync.js" 'use strict'; const { scryptSync, randomFillSync } = require('crypto'); const salt = Buffer.allocUnsafe(16); module.exports = function ({ input, keylen, N = 16384, r = 8, p = 1, maxmem = 32 * 1024 * 1024 }) { return scryptSync(input, randomFillSync(salt), keylen, { N, r, p, maxmem }).toString('hex'); }; ``` ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ```js title="scrypt_sync.ts" 'use strict'; import { scryptSync, randomFillSync } from 'crypto'; const salt = Buffer.allocUnsafe(16); export default function ({ input, keylen, N = 16384, r = 8, p = 1, maxmem = 32 * 1024 * 1024 }: { input: Buffer; keylen: number; N?: number; r?: number; p?: number; maxmem?: number; }): string { return scryptSync(input, randomFillSync(salt), keylen, { N, r, p, maxmem }).toString('hex'); } export const filename = __filename; ``` ### Unpooled Synchronous ```javascript title="unpooled_sync.js" 'use strict'; const crypto = require('crypto'); const { promisify } = require('util'); const { scryptSync, randomFillSync } = crypto; const randomFill = promisify(crypto.randomFill); const { performance, PerformanceObserver } = require('perf_hooks'); const salt = Buffer.allocUnsafe(16); const obs = new PerformanceObserver((entries) => { console.log(entries.getEntries()[0].duration); }); obs.observe({ entryTypes: ['measure'] }); async function * generateInput () { let max = parseInt(process.argv[2] || 10); const data = Buffer.allocUnsafe(10); while (max-- > 0) { yield randomFill(data); } } (async function () { performance.mark('start'); const keylen = 64; for await (const input of generateInput()) { // Everything in here is intentionally sync scryptSync(input, randomFillSync(salt), keylen).toString('hex'); } performance.mark('end'); performance.measure('start to end', 'start', 'end'); })(); ``` ```js title="unpooled_sync.ts" 'use strict'; import { scryptSync, randomFillSync } from 'crypto'; const salt = Buffer.allocUnsafe(16); export default function ({ input, keylen, N = 16384, r = 8, p = 1, maxmem = 32 * 1024 * 1024 }: { input: Buffer; keylen: number; N?: number; r?: number; p?: number; maxmem?: number; }): string { return scryptSync(input, randomFillSync(salt), keylen, { N, r, p, maxmem }).toString('hex'); } ``` ### Running the Example The `package.json` file includes scripts to run each variant of the scrypt implementation with the monitor. ```json title="package.json" { "name": "scrypt", "version": "1.0.0", "scripts": { "pooled": "node -r ./monitor pooled", "unpooled": "node -r ./monitor unpooled", "pooled-sync": "node -r ./monitor pooled_sync", "unpooled-sync": "node -r ./monitor unpooled_sync" }, "keywords": [], "author": "", "license": "MIT", "description": "" } ``` ```json title="package.json" { "name": "scrypt", "version": "1.0.0", "scripts": { "pooled": "ts-node -r ./monitor pooled", "unpooled": "ts-node -r ./monitor unpooled", "pooled-sync": "ts-node -r ./monitor pooled_sync", "unpooled-sync": "ts-node -r ./monitor unpooled_sync" }, "keywords": [], "author": "", "license": "MIT", "description": "", "dependencies": { "@types/node": "^20.14.10", "ts-node": "^10.9.2", "typescript": "^5.5.3" } } ``` To run the different implementations and compare their performance: ```bash npm run pooled 100 npm run unpooled 100 npm run pooled-sync 100 npm run unpooled-sync 100 ``` You can also check out this example on [github](https://github.com/piscinajs/piscina/tree/current/examples/scrypt). --- ### Docs Examples Server (`docs/docs/examples/server.mdx`) --- id: Server sidebar_position: 17 --- The benefit of offloading work to a worker pool will vary significantly based on work load. In some cases, the worker pool may actually be a detriment to performance. Careful consideration must be given. In this example, we create a simple Fastify http server that responds with a simple JSON object after introducing an artificial delay of 100 milliseconds. There are four variants to the server: * async-sleep-unpooled - Uses an async delay using a Promise-wrapped `setTimeout`. The event loop is allowed to keep turning during the delay allowing the server to respond to additional requests. The delay occurs within the main thread. * async-sleep-pooled - Moves the async delay to the worker threads for processing. * sync-sleep-unpooled - Uses a sync delay using `Atomics.wait` to block the main thread for 100 milliseconds before responding. Because the main thread is blocked, the event loop is not turning while we are waiting. This simulates a particularly expensive event loop blocking scenario. * sync-sleep-pooled - Moves the sync delay into the worker threads for processing. :::note For both `async-sleep-pooled` and `sync-sleep-pooled`, the main thread uses a Promise to wait on the worker to complete it's task before responding to the request. ::: ## Setting up The code files for this example can be found on [github](https://github.com/piscinajs/piscina/tree/current/examples/server). To get started, run `npm i` to install dependencies. We use [`autocannon`][] for benchmarking the four variants: ```console $ npm i autocannon ``` Let's start with the two async sleep variants. First run: ```console $ node async-sleep-unpooled ``` And in a separate terminal, run autocannon: ```console $ autocannon localhost:3000 ``` You should see results similar to: ``` Running 10s test @ http://localhost:3000 10 connections ┌─────────┬────────┬────────┬────────┬────────┬──────────┬─────────┬───────────┐ │ Stat │ 2.5% │ 50% │ 97.5% │ 99% │ Avg │ Stdev │ Max │ ├─────────┼────────┼────────┼────────┼────────┼──────────┼─────────┼───────────┤ │ Latency │ 100 ms │ 101 ms │ 104 ms │ 125 ms │ 101.2 ms │ 3.09 ms │ 130.66 ms │ └─────────┴────────┴────────┴────────┴────────┴──────────┴─────────┴───────────┘ ┌───────────┬─────────┬─────────┬─────────┬─────────┬───────┬───────┬─────────┐ │ Stat │ 1% │ 2.5% │ 50% │ 97.5% │ Avg │ Stdev │ Min │ ├───────────┼─────────┼─────────┼─────────┼─────────┼───────┼───────┼─────────┤ │ Req/Sec │ 90 │ 90 │ 99 │ 100 │ 97.8 │ 3.32 │ 90 │ ├───────────┼─────────┼─────────┼─────────┼─────────┼───────┼───────┼─────────┤ │ Bytes/Sec │ 14.8 kB │ 14.8 kB │ 16.2 kB │ 16.4 kB │ 16 kB │ 545 B │ 14.8 kB │ └───────────┴─────────┴─────────┴─────────┴─────────┴───────┴───────┴─────────┘ ``` :::info Note that this server is not particularly fast at all due to the artificially imposed 100 millisecond delay. It's possible to achieve significantly higher results by tuning the way autocannon is sending requests. For instance, running autocannon with the `-c 100 -p 2` options increases performance of the `async-sleep-unpooled` example significantly. ::: Let's see how the Piscina version does in comparison: Run: ```console $ node async-sleep-pooled ``` And autocannon again: ```console $ autocannon localhost:3000 ``` Your result should be similar to: ``` Running 10s test @ http://localhost:3000 10 connections ┌─────────┬────────┬────────┬────────┬────────┬───────────┬──────────┬───────────┐ │ Stat │ 2.5% │ 50% │ 97.5% │ 99% │ Avg │ Stdev │ Max │ ├─────────┼────────┼────────┼────────┼────────┼───────────┼──────────┼───────────┤ │ Latency │ 126 ms │ 169 ms │ 197 ms │ 339 ms │ 170.13 ms │ 31.41 ms │ 425.33 ms │ └─────────┴────────┴────────┴────────┴────────┴───────────┴──────────┴───────────┘ ┌───────────┬─────────┬─────────┬─────────┬─────────┬─────────┬───────┬─────────┐ │ Stat │ 1% │ 2.5% │ 50% │ 97.5% │ Avg │ Stdev │ Min │ ├───────────┼─────────┼─────────┼─────────┼─────────┼─────────┼───────┼─────────┤ │ Req/Sec │ 43 │ 43 │ 60 │ 60 │ 58.2 │ 5.08 │ 43 │ ├───────────┼─────────┼─────────┼─────────┼─────────┼─────────┼───────┼─────────┤ │ Bytes/Sec │ 7.05 kB │ 7.05 kB │ 9.85 kB │ 9.85 kB │ 9.55 kB │ 833 B │ 7.05 kB │ └───────────┴─────────┴─────────┴─────────┴─────────┴─────────┴───────┴─────────┘ ``` Notice that the pooled version is slower! This is because in both async examples, the event loop on the main thread is still active, allowing it to service more requests. The worker pool here, however, adds additional performance overhead marshalling data back and forth between threads and allocating additional Promises. In this scenario, the worker pool does not add much benefit. Let's look as the sync cases and see what happens with those. Run: ```console $ node sync-sleep-unpooled ``` And run autocannon again: ```console $ autocannon localhost:3000 ``` The results should be fairly awful in comparison to the first two cases: ``` Running 10s test @ http://localhost:3000 10 connections ┌─────────┬────────┬─────────┬─────────┬─────────┬──────────┬───────────┬────────────┐ │ Stat │ 2.5% │ 50% │ 97.5% │ 99% │ Avg │ Stdev │ Max │ ├─────────┼────────┼─────────┼─────────┼─────────┼──────────┼───────────┼────────────┤ │ Latency │ 338 ms │ 1007 ms │ 1811 ms │ 1919 ms │ 962.2 ms │ 255.54 ms │ 1919.77 ms │ └─────────┴────────┴─────────┴─────────┴─────────┴──────────┴───────────┴────────────┘ ┌───────────┬─────────┬─────────┬─────────┬─────────┬─────────┬────────┬─────────┐ │ Stat │ 1% │ 2.5% │ 50% │ 97.5% │ Avg │ Stdev │ Min │ ├───────────┼─────────┼─────────┼─────────┼─────────┼─────────┼────────┼─────────┤ │ Req/Sec │ 9 │ 9 │ 10 │ 10 │ 9.81 │ 0.4 │ 9 │ ├───────────┼─────────┼─────────┼─────────┼─────────┼─────────┼────────┼─────────┤ │ Bytes/Sec │ 1.48 kB │ 1.48 kB │ 1.64 kB │ 1.64 kB │ 1.61 kB │ 65.6 B │ 1.48 kB │ └───────────┴─────────┴─────────┴─────────┴─────────┴─────────┴────────┴─────────┘ ``` The reason for the performance drop should be apparent. Synchronously sleeping blocks the event loop from turning while a request is being processed, which means the server is unable to do anything else while it waits. To see how the pooled version fares, run: ```console $ node sync-sleep-pooled ``` And run autocannon again ```console $ autocannon localhost:3000 ``` The results should be nearly identical to the `async-sleep-pooled` version! ``` Running 10s test @ http://localhost:3000 10 connections ┌─────────┬────────┬────────┬────────┬────────┬───────────┬──────────┬──────────┐ │ Stat │ 2.5% │ 50% │ 97.5% │ 99% │ Avg │ Stdev │ Max │ ├─────────┼────────┼────────┼────────┼────────┼───────────┼──────────┼──────────┤ │ Latency │ 124 ms │ 174 ms │ 200 ms │ 335 ms │ 169.99 ms │ 36.22 ms │ 422.2 ms │ └─────────┴────────┴────────┴────────┴────────┴───────────┴──────────┴──────────┘ ┌───────────┬─────────┬─────────┬─────────┬─────────┬─────────┬───────┬─────────┐ │ Stat │ 1% │ 2.5% │ 50% │ 97.5% │ Avg │ Stdev │ Min │ ├───────────┼─────────┼─────────┼─────────┼─────────┼─────────┼───────┼─────────┤ │ Req/Sec │ 42 │ 42 │ 60 │ 60 │ 58.2 │ 5.4 │ 42 │ ├───────────┼─────────┼─────────┼─────────┼─────────┼─────────┼───────┼─────────┤ │ Bytes/Sec │ 6.89 kB │ 6.89 kB │ 9.85 kB │ 9.85 kB │ 9.55 kB │ 886 B │ 6.89 kB │ └───────────┴─────────┴─────────┴─────────┴─────────┴─────────┴───────┴─────────┘ ``` The reason `sync-sleep-pooled` and `async-sleep-pooled` yield identical results is because, in both cases, the main thread is doing identical work -- that is, receiving a request that is dispatched to a worker, waiting about 100 milliseconds, the returning the response. The main thread here does not care whether the workers are sleeping synchronously or asynchronously. ## Tuning pool performance It is possible to tune the performance of the Piscina worker pool using a variety of options: * `minThreads` - The minimum number of threads always running * `maxThreads` - The maximum number of threads allowed * `idleTimeout` - The number of millisecondsa thread is permitted to remain idle * `maxQueue` -- The maximum number of pending work items * `concurrentTasksPerWorker` -- The number of work items to dispatch concurrently to a single thread. We'll use `idleTimeout` and `concurrentTasksPerWorker` in this example. By default, as soon as a Piscina worker thread has nothing to do, it will terminate. This means that if the work queue is empty, the thread will terminate. If we're not keeping our queue filled, this will incur additional overhead as Node.js spins up new worker threads to handle incoming requests. Also by default, Piscina will assume that jobs are synchronous in nature and will dispatch only a single job per thread at any time. If the workload is asynchronous, setting `concurrentTasksPerWorker` to a higher number will increase the number of jobs Piscina will send to a single worker, allowing it to process multiple tasks. The `async-sleep-pooled` example has been written to accept two optional parameters. The first is the `concurrentTasksPerWorker`, and the second is the `idleTimeout`. To see the effect setting each has on the performance of the example, run: ```console $ node async-sleep-pooled 10 1000 ``` Then run autocannon again: ```console $ autocannon localhost:3000 ``` You should see that the performance of the pooled example improves significantly, but is still slightly less than the async-sleep-unpooled version: ``` Running 10s test @ http://localhost:3000 10 connections ┌─────────┬────────┬────────┬────────┬────────┬───────────┬──────────┬───────────┐ │ Stat │ 2.5% │ 50% │ 97.5% │ 99% │ Avg │ Stdev │ Max │ ├─────────┼────────┼────────┼────────┼────────┼───────────┼──────────┼───────────┤ │ Latency │ 100 ms │ 101 ms │ 126 ms │ 135 ms │ 105.16 ms │ 23.23 ms │ 391.75 ms │ └─────────┴────────┴────────┴────────┴────────┴───────────┴──────────┴───────────┘ ┌───────────┬─────────┬─────────┬─────────┬─────────┬─────────┬─────────┬─────────┐ │ Stat │ 1% │ 2.5% │ 50% │ 97.5% │ Avg │ Stdev │ Min │ ├───────────┼─────────┼─────────┼─────────┼─────────┼─────────┼─────────┼─────────┤ │ Req/Sec │ 76 │ 76 │ 97 │ 99 │ 94.2 │ 6.42 │ 76 │ ├───────────┼─────────┼─────────┼─────────┼─────────┼─────────┼─────────┼─────────┤ │ Bytes/Sec │ 12.5 kB │ 12.5 kB │ 15.9 kB │ 16.2 kB │ 15.5 kB │ 1.05 kB │ 12.5 kB │ └───────────┴─────────┴─────────┴─────────┴─────────┴─────────┴─────────┴─────────┘ ``` Let's use the same options for the `sync-sleep-pooled.js` example: ```console $ node sync-sleep-pooled 10 1000 ``` Then run autocannon again: ```console $ autocannon localhost:3000 ``` With the results: ``` Running 10s test @ http://localhost:3000 10 connections ┌─────────┬────────┬────────┬────────┬────────┬───────────┬──────────┬──────────┐ │ Stat │ 2.5% │ 50% │ 97.5% │ 99% │ Avg │ Stdev │ Max │ ├─────────┼────────┼────────┼────────┼────────┼───────────┼──────────┼──────────┤ │ Latency │ 100 ms │ 200 ms │ 201 ms │ 329 ms │ 169.05 ms │ 52.78 ms │ 456.5 ms │ └─────────┴────────┴────────┴────────┴────────┴───────────┴──────────┴──────────┘ ┌───────────┬─────────┬─────────┬─────────┬─────────┬─────────┬───────┬─────────┐ │ Stat │ 1% │ 2.5% │ 50% │ 97.5% │ Avg │ Stdev │ Min │ ├───────────┼─────────┼─────────┼─────────┼─────────┼─────────┼───────┼─────────┤ │ Req/Sec │ 46 │ 46 │ 60 │ 60 │ 58.4 │ 4.16 │ 46 │ ├───────────┼─────────┼─────────┼─────────┼─────────┼─────────┼───────┼─────────┤ │ Bytes/Sec │ 7.55 kB │ 7.55 kB │ 9.85 kB │ 9.85 kB │ 9.58 kB │ 681 B │ 7.54 kB │ └───────────┴─────────┴─────────┴─────────┴─────────┴─────────┴───────┴─────────┘ ``` The options have no noticeable impact on the sync sleep version. The reason is because the workload is fully synchronous and cannot be processed concurrently no matter what the concurrency settings are. The performance, in other words, is bound entirely to the synchronous delay and can only be improved by reducing the event loop block. [`autocannon`]: https://www.npmjs.com/package/autocannon You can also check out this example on [github](https://github.com/piscinajs/piscina/tree/current/examples/server). --- ### Docs Examples Simple (`docs/docs/examples/simple.mdx`) --- id: Simple sidebar_position: 19 --- import Tabs from '@theme/Tabs'; import TabItem from '@theme/TabItem'; import { WorkerWrapperComponent } from '@site/src/components/WorkerWrapper.mdx' In this example, we create a Piscina instance that uses a worker file to perform a simple addition operation. The main script (`index.js`) creates the Piscina instance and runs a task, while the worker script (`worker.js`) defines the task to be executed. ```javascript title="index.js" 'use strict'; const Piscina = require('../..'); const { resolve } = require('path'); const piscina = new Piscina({ filename: resolve(__dirname, 'worker.js') }); (async function () { const result = await piscina.run({ a: 4, b: 6 }); console.log(result); // Prints 10 })(); ``` ```javascript title="worker.js" 'use strict'; module.exports = ({ a, b }) => { return a + b; }; ``` ```typescript title="index.ts" import Piscina from 'piscina'; import { resolve } from 'path'; import { filename } from './worker'; const piscina = new Piscina({ filename: resolve(__dirname, 'workerWrapper.js'), workerData: { fullpath: filename }, }); (async function () { const result = await piscina.run({ a: 4, b: 6 }); console.log(result); // Prints 10 })(); ``` ```typescript title="worker.ts" export const filename = __filename; interface AdditionParams { a: number; b: number; } export default ({ a, b }: AdditionParams): number => { return a + b; }; ``` You can also check out this example on [github](https://github.com/piscinajs/piscina/tree/current/examples/simple). --- ### Docs Examples Simple Async (`docs/docs/examples/simple_async.mdx`) --- id: Simple Async sidebar_position: 18 --- import Tabs from '@theme/Tabs'; import TabItem from '@theme/TabItem'; import { WorkerWrapperComponent } from '@site/src/components/WorkerWrapper.mdx' This example builds upon the [simple addition](./simple.mdx) scenario in the previous section. In this section, we simulated an asynchronous operation in the worker file. The simulated delay (100ms) represents any asynchronous operation that might occur in a real-world scenario, such as database queries, file I/O, or network requests. ```javascript title="index.js" 'use strict'; const Piscina = require('../..'); const { resolve } = require('path'); const piscina = new Piscina({ filename: resolve(__dirname, 'worker.js') }); (async function () { const result = await piscina.run({ a: 4, b: 6 }); console.log(result); // Prints 10 })(); ``` ```javascript title="worker.js" 'use strict'; const { promisify } = require('util'); const sleep = promisify(setTimeout); module.exports = async ({ a, b }) => { // Fake some async activity await sleep(100); return a + b; }; ``` ```typescript title="index.ts" import Piscina from 'piscina'; import { resolve } from 'path'; import { filename } from './worker'; const piscina = new Piscina({ filename: resolve(__dirname, 'workerWrapper.js'), workerData: { fullpath: filename }, }); (async function () { const result = await piscina.run({ a: 4, b: 6 }); console.log(result); // Prints 10 })(); ``` ```typescript title="worker.ts" import { promisify } from 'util'; const sleep = promisify(setTimeout); interface AdditionParams { a: number; b: number; } export default async ({ a, b }: AdditionParams): Promise => { // Fake some async activity await sleep(100); return a + b; }; export const filename = __filename; ``` You can also check out this example on [github](https://github.com/piscinajs/piscina/tree/current/examples/simple_async). --- ### Docs Examples Stream In (`docs/docs/examples/stream-in.mdx`) --- id: Stream-In sidebar_position: 20 --- import Tabs from '@theme/Tabs'; import TabItem from '@theme/TabItem'; import { WorkerWrapperComponent } from '@site/src/components/WorkerWrapper.mdx'; Piscina can be effectively used in data processing pipelines, handling large volumes of data efficiently by leveraging worker threads and implementing proper flow control. The main script in the example below sets up a Piscina pool and a CSV stream, then processes each line of the CSV file through a worker thread. We implement flow control by pausing the stream when the Piscina queue reaches its maximum size and resuming it when the queue drains. This ensures that memory usage remains under control even when processing very large files. ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ``` /* Detailed source-code truncated for AI context efficiency. */ ``` The worker (`worker.js`) is a simple function that takes a line of data and returns its JSON stringified version. ```javascript title="worker.js" module.exports = (data) => { return JSON.stringify(data); }; ``` ```typescript title="worker.ts" export default function toJson(data: any): string { return JSON.stringify(data); } export const filename = __filename; ``` A custom `Progress` class (`progress.js`) is used to track the progress of the task, emitting events and providing status messages. ```javascript title="progress.js" 'use strict'; const { EventEmitter } = require('events'); class Progress extends EventEmitter { #tasksSubmitted = 0; #tasksCompleted = 0; #tasksFailed = 0; #done = false; done () { this.#done = true; } incSubmitted () { this.#tasksSubmitted++; } incCompleted () { this.#tasksCompleted++; if (this.#done && this.completed === this.#tasksSubmitted) { process.nextTick(() => this.emit('finished')); } } incFailed () { this.#tasksFailed++; } get completed () { return this.#tasksCompleted + this.#tasksFailed; } get message () { return `${this.#tasksCompleted} of ${this.#tasksSubmitted} completed` + ` ${((this.#tasksCompleted / this.#tasksSubmitted) * 100).toFixed(2)}%` + ` [${this.#tasksFailed} failed]`; } }; module.exports = Progress; ```` ```typescript title="progress.ts" import { EventEmitter } from 'events'; class Progress extends EventEmitter { #tasksSubmitted = 0; #tasksCompleted = 0; #tasksFailed = 0; #done = false; done() { this.#done = true; } incSubmitted() { this.#tasksSubmitted++; } incCompleted() { this.#tasksCompleted++; if (this.#done && this.completed === this.#tasksSubmitted) { process.nextTick(() => this.emit('finished')); } } incFailed() { this.#tasksFailed++; } get completed() { return this.#tasksCompleted + this.#tasksFailed; } get message() { return `${this.#tasksCompleted} of ${this.#tasksSubmitted} completed` + ` ${((this.#tasksCompleted / this.#tasksSubmitted) * 100).toFixed(2)}%` + ` [${this.#tasksFailed} failed]`; } } export default Progress; ``` A separate script (`generate.js`) is provided to generate sample CSV data for testing purposes. ```javascript title="generate.js" 'use strict'; const { createWriteStream } = require('fs'); const out = createWriteStream('./data.csv'); const count = parseInt(process.argv[2] || '5000') || 5000; out.write('a,b,c,d,e,f,g\n'); for (let n = 0; n < count; n++) { out.write('1,2,3,4,5,6,7\n'); } out.end(); console.log('done'); ``` ```typescript title="generate.ts" import { createWriteStream } from 'fs'; const out = createWriteStream('./data.csv'); const count = parseInt(process.argv[2] || '5000') || 5000; out.write('a,b,c,d,e,f,g\n'); for (let n = 0; n < count; n++) { out.write('1,2,3,4,5,6,7\n'); } out.end(); console.log('done'); ``` You can also check out this example on [github](https://github.com/piscinajs/piscina/tree/current/examples/es-module). --- ### Docs Examples Stream (`docs/docs/examples/stream.mdx`) --- id: Stream sidebar_position: 21 --- import Tabs from '@theme/Tabs'; import TabItem from '@theme/TabItem'; import { WorkerWrapperComponent } from '@site/src/components/WorkerWrapper.mdx' This example explains how Piscina can be used for more complex scenarios involving stream processing. Piscina allows for efficient data transfer between the main thread and worker threads using `MessagePort`. The main script (`index.mjs`) sets up a pipeline that reads a file, processes its contents through a worker thread (converting text to uppercase), and outputs the result. ```javascript title="index.mjs" import Piscina from 'piscina'; import { MessagePortDuplex } from './stream.mjs'; import { createReadStream } from 'fs'; import { pipeline } from 'stream'; const pool = new Piscina({ filename: new URL('./worker.mjs', import.meta.url).href }); const { port1, port2 } = new MessageChannel(); pool.run(port2, { transferList: [port2] }); const duplex = new MessagePortDuplex(port1); pipeline( createReadStream(new URL('./index.mjs', import.meta.url).pathname), duplex, process.stdout, (err) => { if (err) throw err; }); ``` ```typescript title="index.ts" import Piscina from 'piscina'; import { resolve } from 'path'; import { createReadStream } from 'fs'; import { pipeline } from 'stream'; import { MessageChannel } from 'worker_threads'; import { filename } from './worker'; import { MessagePortDuplex } from './stream'; const pool = new Piscina({ filename: resolve(__dirname, 'workerWrapper.js'), workerData: { fullpath: filename }, }); const { port1, port2 } = new MessageChannel(); pool.run(port2, { transferList: [port2] }); const duplex = new MessagePortDuplex(port1); pipeline( createReadStream(resolve(__dirname, "index.ts")), duplex, process.stdout, (err: NodeJS.ErrnoException | null) => { if (err) throw err; } ); ``` The stream file defines custom stream classes that extend standard Node.js streams to work with `MessagePort`. These classes enable efficient communication between the main thread and worker threads. ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ```typescript title="stream.ts" import { Writable, Readable, Duplex, WritableOptions, ReadableOptions, DuplexOptions } from 'stream'; import { MessagePort } from 'worker_threads'; const kPort = Symbol('kPort'); export class MessagePortWritable extends Writable { private [kPort]: MessagePort; constructor(port: MessagePort, options?: WritableOptions) { super(options); this[kPort] = port; } _write(buf: Buffer, _: string, cb: (error?: Error | null) => void): void { this[kPort].postMessage(buf, [buf.buffer]); cb(); } // ... (other methods) } export class MessagePortReadable extends Readable { private [kPort]: MessagePort; constructor(port: MessagePort, options?: ReadableOptions) { super(options); this[kPort] = port; port.onmessage = ({ data }) => this.push(data); } // ... (other methods) } export class MessagePortDuplex extends Duplex { private [kPort]: MessagePort; constructor(port: MessagePort, options?: DuplexOptions) { super(options); this[kPort] = port; port.onmessage = ({ data }) => this.push(data); } // ... (other methods) } ``` The worker script defines a function that processes data through a `MessagePortDuplex` stream, converting incoming text to uppercase and writing it back. ```javascript title="worker.mjs" import { MessagePortDuplex } from './stream.mjs'; export default function (port) { let res; const ret = new Promise((resolve) => { res = resolve; }); const duplex = new MessagePortDuplex(port); duplex.setEncoding('utf8'); duplex.on('data', (chunk) => duplex.write(chunk.toUpperCase())); duplex.on('end', () => { duplex.end(); res(); }); return ret; } ``` ```typescript title="worker.ts" import { MessagePortDuplex } from './stream'; import { MessagePort } from 'worker_threads'; export default function (port: MessagePort): Promise { let res: () => void; const ret = new Promise((resolve) => { res = resolve; }); const duplex = new MessagePortDuplex(port); duplex.setEncoding('utf8'); duplex.on('data', (chunk: string) => duplex.write(chunk.toUpperCase())); duplex.on('end', () => { duplex.end(); res(); }); return ret; } ``` You can also check out this example on [github](https://github.com/piscinajs/piscina/tree/current/examples/stream). --- ### Docs Examples Task Queue (`docs/docs/examples/task-queue.mdx`) --- id: Task Queue sidebar_position: 22 --- import {WorkerWrapperComponent} from '@site/src/components/WorkerWrapper.mdx'; import Tabs from "@theme/Tabs"; import TabItem from "@theme/TabItem"; You can extend Piscina's functionality by implementing a custom task queue. Instead of the default FIFO (First-In-First-Out) queue, it uses a priority queue to manage tasks. The main script (`index.js`) defines a `PriorityTaskQueue` class that wraps the `shuffled-priority-queue` package. This class implements the necessary methods (`push`, `remove`, `shift`) that Piscina expects from a task queue. The queue prioritizes tasks based on a `priority` value specified in the task options. :::note Task priorities only affect queued tasks. If a worker is immediately available when a task is submitted, it will be processed regardless of its priority. The priority queue comes into play when there are more tasks than available workers. ::: ```javascript title="index.js" 'use strict'; const spq = require('shuffled-priority-queue'); const Piscina = require('piscina'); const { resolve } = require('path'); const kItem = Symbol('item'); class PriorityTaskQueue { queue = spq(); get size () { return this.queue.length; } push (value) { const queueOptions = value[Piscina.queueOptionsSymbol]; const priority = queueOptions ? (queueOptions.priority || 0) : 0; value[kItem] = this.queue.add({ priority, value }); } remove (value) { this.queue.remove(value[kItem]); } shift () { return this.queue.shift().value; } } const pool = new Piscina({ filename: resolve(__dirname, 'worker.js'), taskQueue: new PriorityTaskQueue(), maxThreads: 4 }); function makeTask (task, priority) { return { ...task, [Piscina.queueOptionsSymbol]: { priority } }; } (async () => { // Submit enough tasks to ensure that at least some are queued console.log(await Promise.all([ pool.run(makeTask({ priority: 1 }, 1)), pool.run(makeTask({ priority: 2 }, 2)), pool.run(makeTask({ priority: 3 }, 3)), pool.run(makeTask({ priority: 4 }, 4)), pool.run(makeTask({ priority: 5 }, 5)), pool.run(makeTask({ priority: 6 }, 6)), pool.run(makeTask({ priority: 7 }, 7)), pool.run(makeTask({ priority: 8 }, 8)) ])); })(); ``` ``` /* Detailed source-code truncated for AI context efficiency. */ ``` The worker script (`worker.js`) simulates some work by sleeping for 100ms, then logs and returns the priority of the task it processed. This allows us to observe the order in which tasks are executed. ```javascript title="worker.js" 'use strict'; const { promisify } = require('util'); const sleep = promisify(setTimeout); module.exports = async ({ priority }) => { await sleep(100); process._rawDebug(`${priority}`); return priority; }; ``` ```typescript title="worker.ts" import { promisify } from 'util'; const sleep = promisify(setTimeout); export default async ({ priority }: { priority: number }): Promise => { await sleep(100); process._rawDebug(`${priority}`); return priority; }; export const filename = __filename; ``` You can also check out this example on [github](https://github.com/piscinajs/piscina/tree/current/examples/task-queue). --- ### Docs Examples Webstreams Transform (`docs/docs/examples/webstreams-transform.mdx`) --- id: Web Streams Transfer sidebar_position: 24 --- import {WorkerWrapperComponent} from '@site/src/components/WorkerWrapper.mdx'; import Tabs from "@theme/Tabs"; import TabItem from "@theme/TabItem"; Using [Web Streams](https://developer.mozilla.org/en-US/docs/Web/API/Streams_API) and Piscina, you can create a data processing pipeline. This is useful for cases involving real-time data transformation or analysis. In the main script below, we set up three key components of the Web Streams API: * A `ReadableStream` that generates a sequence of numbers from 1 to 9, and then 0. * A `WritableStream` that simply logs each chunk it receives to the console. * A `TransformStream` that processes each chunk using a Piscina worker pool. The TransformStream's `transform` method uses `pool.run()` to process each chunk in a worker thread. This allows for parallel processing of the stream data. Finally, the script sets up a pipeline using `pipeThrough()` and `pipeTo()` methods, connecting the `ReadableStream` to the `TransformStream`, and then to the `WritableStream`. ```javascript title="worker.mjs" export default async function (num) { return 'ABC'.repeat(num * num); } ``` ```javascript title="index.mjs" import Piscina from 'piscina'; import { ReadableStream, TransformStream, WritableStream } from 'node:stream/web'; const pool = new Piscina({ filename: new URL('./worker.mjs', import.meta.url).href }); const readable = new ReadableStream({ start () { this.chunks = [1, 2, 3, 4, 5, 6, 7, 8, 9, 0]; }, pull (controller) { const chunk = this.chunks.shift(); controller.enqueue(chunk); if (this.chunks.length === 0) { controller.close(); } } }); const writable = new WritableStream({ write (chunk) { console.log(chunk); } }); const transform = new TransformStream({ async transform (chunk, controller) { controller.enqueue(await pool.run(chunk)); } }); readable.pipeThrough(transform).pipeTo(writable); ``` ```typescript title="worker.ts" export default async function (num: number): Promise { return 'ABC'.repeat(num * num); } export const filename = __filename; ``` ```typescript title="index.ts" import { resolve } from 'path'; import Piscina from 'piscina'; import { ReadableStream, WritableStream } from 'node:stream/web'; import { filename } from './worker'; const pool = new Piscina({ filename: resolve(__dirname, 'workerWrapper.js'), workerData: { fullpath: filename }, }); const readable = new ReadableStream({ start () { this.chunks = [1, 2, 3, 4, 5, 6, 7, 8, 9, 0]; }, pull (controller) { const chunk = this.chunks.shift(); controller.enqueue(chunk); if (this.chunks.length === 0) { controller.close(); } } }); const writable = new WritableStream({ write (chunk) { console.log(chunk); } }); const transform = new TransformStream({ async transform (chunk, controller) { controller.enqueue(await pool.run(chunk)); } }); readable.pipeThrough(transform).pipeTo(writable); ``` You can also check out this example on [github](https://github.com/piscinajs/piscina/tree/current/examples/webstreams-transform). --- ### Docs Examples Webstreams (`docs/docs/examples/webstreams.mdx`) --- id: Web Streams sidebar_position: 25 --- import {WorkerWrapperComponent} from '@site/src/components/WorkerWrapper.mdx'; import Tabs from "@theme/Tabs"; import TabItem from "@theme/TabItem"; You can work with modern Web APIs like [Web Streams](https://developer.mozilla.org/en-US/docs/Web/API/Streams_API) using Piscina. Web Streams enable efficient processing of streaming data across multiple threads. It's particularly useful for scenarios involving large datasets or real-time data processing where the benefits of multi-threading can be significant. In the main script (`index.mjs`), we create a Piscina instance and set up a `ReadableStream` that generates a sequence of numbers. We also create a `WritableStream` that simply logs each chunk it receives. The script then runs a task in the worker pool, passing both streams as arguments. The worker script (`worker.mjs`) defines an async function that reads from the provided `ReadableStream` and writes to the `WritableStream`. It uses a for-await loop to iterate over the chunks in the `ReadableStream`, writing each chunk to the `WritableStream`. ```javascript title="worker.mjs" export default async function ({ readable, writable }) { const writer = writable.getWriter(); for await (const chunk of readable) { await writer.write(chunk); } writer.close(); } ``` ```javascript title="index.mjs" import Piscina from 'piscina'; import { ReadableStream, WritableStream } from 'node:stream/web'; const pool = new Piscina({ filename: new URL('./worker.mjs', import.meta.url).href }); const readable = new ReadableStream({ start () { this.chunks = [1, 2, 3, 4, 5, 6, 7, 8, 9, 0]; }, pull (controller) { const chunk = this.chunks.shift(); controller.enqueue(chunk); if (this.chunks.length === 0) { controller.close(); } } }); const writable = new WritableStream({ write (chunk) { console.log(chunk); } }); (async function () { await pool.run({ readable, writable }, { transferList: [readable, writable] }); })() ``` ```typescript title="worker.ts" import { ReadableStream, WritableStream } from 'node:stream/web'; export default async function ({ readable, writable }: { readable: ReadableStream, writable: WritableStream }): Promise { const writer = writable.getWriter(); for await (const chunk of readable) { await writer.write(chunk); } writer.close(); } ``` ```typescript title="index.ts" import { resolve } from 'path'; import Piscina from 'piscina'; import { ReadableStream, WritableStream } from 'node:stream/web'; import { filename } from './worker'; const pool = new Piscina({ filename: resolve(__dirname, 'workerWrapper.js'), workerData: { fullpath: filename }, }); const readable = new ReadableStream({ start () { this.chunks = [1, 2, 3, 4, 5, 6, 7, 8, 9, 0]; }, pull (controller) { const chunk = this.chunks.shift(); controller.enqueue(chunk); if (this.chunks.length === 0) { controller.close(); } } }); const writable = new WritableStream({ write (chunk) { console.log(chunk); } }); (async function () { await pool.run({ readable, writable }, { transferList: [readable, writable] }); })() ``` A key aspect of this example is the use of the `transferList` option when running the task. This allows the `ReadableStream` and `WritableStream` instances to be transferred to the worker thread, rather than cloned. This is crucial for maintaining the integrity of the streams across threads. You can also check out this example on [github](https://github.com/piscinajs/piscina/tree/current/examples/webstreams). --- ### Docs Examples Worker Options (`docs/docs/examples/worker-options.mdx`) --- id: Worker Options sidebar_position: 26 --- import {WorkerWrapperComponent} from '@site/src/components/WorkerWrapper.mdx'; import Tabs from "@theme/Tabs"; import TabItem from "@theme/TabItem"; Piscina allows you to customize the environment and runtime options for worker threads. You can set environment variables, command-line arguments, and other options for the worker processes. This is useful when you need to: - Pass configuration or environment-specific data to workers - Control Node.js runtime behavior in workers - Provide initial data to workers without including it in every task ```javascript title="index.js" 'use strict'; const Piscina = require('../..'); const { resolve } = require('path'); const piscina = new Piscina({ filename: resolve(__dirname, 'worker.js'), env: { A: '123' }, argv: ['a', 'b', 'c'], execArgv: ['--no-warnings'], workerData: 'ABC' }); (async function () { const result = await piscina.run({ a: 4, b: 6 }); console.log(result); // Prints 10 })(); ``` ```javascript title="worker.js" 'use strict'; const Piscina = require('../..'); const { format } = require('util'); module.exports = ({ a, b }) => { console.log(` process.argv: ${process.argv.slice(2)} process.execArgv: ${process.execArgv} process.env: ${format({ ...process.env })} workerData: ${Piscina.workerData}`); return a + b; }; ``` ```typescript title="index.ts" import { resolve } from 'path'; import Piscina from 'piscina'; import { filename } from './worker'; const piscina = new Piscina({ filename: resolve(__dirname, 'workerWrapper.js'), workerData: { fullpath: filename }, env: { A: '123' }, argv: ['a', 'b', 'c'], execArgv: ['--no-warnings'], }); (async function () { const result = await piscina.run({ a: 4, b: 6 }); console.log(result); // Prints 10 })(); ``` ```typescript title="worker.ts" import Piscina from 'piscina'; import { format } from 'util'; export default ({ a, b }: { a: number, b: number }): number => { console.log(` process.argv: ${process.argv.slice(2)} process.execArgv: ${process.execArgv} process.env: ${format({ ...process.env })} workerData: ${Piscina.workerData}`); return a + b; }; ``` You can also check out this example on [github](https://github.com/piscinajs/piscina/tree/current/examples/worker_options). --- ### Docs Getting Started Installation (`docs/docs/getting-started/Installation.mdx`) --- id: Installation sidebar_position: 1 --- import Tabs from "@theme/Tabs"; import TabItem from "@theme/TabItem"; This section will guide you through the process of getting started with Piscina.js. We'll cover installation, basic usage, and configuration options to help you quickly integrate Piscina.js into your project. To get started, you'll need to have Node.js version 22 or higher installed on your system. Open your terminal and run the following command: ```bash tab={"label":"npm"} npm install piscina ``` ```bash tab={"label":"yarn"} yarn add piscina ``` ```bash tab={"label":"pnpm"} pnpm add piscina ``` ```bash tab={"label":"Bun"} bun add piscina ``` :::note While Bun can be used to install Piscina, its behavior while running within Bun is not assured. ::: This will download and install the latest version of Piscina.js and its dependencies. Once you have Piscina.js installed, you can start using it in your code. --- ### Docs Getting Started Category .Json (`docs/docs/getting-started/_category_.json`) { "label": "Getting Started", "position": 2, "link": { "type": "generated-index", "description": "Learn how to install and use Piscina in your application." } } --- ### Docs Getting Started Basic Usage (`docs/docs/getting-started/basic-usage.mdx`) --- id: Basic Usage sidebar_position: 2 --- import {WorkerWrapperComponent} from '@site/src/components/WorkerWrapper.mdx'; import Tabs from "@theme/Tabs"; import TabItem from "@theme/TabItem"; ## Setting Up Your Main File ```javascript title='main.js' 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 })(); ``` ```javascript title='main.ts' import Piscina from "piscina"; import { resolve } from "path"; import { filename } from "./worker"; const piscina = new Piscina({ filename: resolve(__dirname, "./workerWrapper.js"), workerData: { fullpath: filename }, }); (async () => { const result = await piscina.run({ a: 2, b: 3 }, { name: "addNumbers" }); console.log("Result:", result); })(); ``` ## Using a Worker Wrapper File in a TypeScript Project When working with TypeScript, you need to use a `workerWrapper.js` file to load your worker file correctly. This file checks if the provided file path ends with `.ts` and registers the `ts-node` compiler to handle TypeScript files. You can learn more about using Typescript with Piscina in the following [guide](./typescript.mdx). ## Creating a Worker File ```javascript tab={"label":"Javascript"} title='worker.js' // A simple worker function that adds two numbers module.exports = ({ a, b }) => a + b; ``` ```javascript tab={"label":"Typescript"} title='worker.ts' // A simple worker function that adds two numbers const { resolve } = require("path"); export const filename = resolve(__filename); interface Inputs { a: number; b: number; } export function addNumbers({ a, b }: Inputs): number { return a + b; } ``` ## Using Async Functions or Promises Workers can be asynchronous or return a promise: ```javascript tab={"label":"Javascript"} const { setTimeout } = require("timers/promises"); // An async worker function with simulated delay module.exports = async ({ a, b }) => { // Fake some async activity with a delay await setTimeout(100); return a + b; }; ``` ```javascript const { setTimeout } = require("timers/promises"); interface WorkerInput { a: number; b: number; } // An async worker function with simulated delay module.exports = async ({ a, b }: WorkerInput): Promise => { // Fake some async activity with a delay await setTimeout(100); return a + b; }; ``` ## Supporting ECMAScript Modules (ESM) Piscina.js also works with ECMAScript modules: ```javascript 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 ``` ```javascript import { Piscina } from "piscina"; import { filename } from "./worker"; const workerWrapperURL = new URL("workerWrapper.js", import.meta.url); const piscina = new Piscina({ filename: workerWrapperURL.href, workerData: { fullpath: filename }, }); (async () => { const result = await piscina.run({ a: 4, b: 6 }, { name: "addNumbers" }); console.log(result); // Prints 10 })(); ``` In your worker module file: ```javascript title="worker.mjs" // Default export of an addition function export default ({ a, b }) => a + b; ``` ```javascript tab={"label":"Typescript"} title="worker.ts" // Default export of an addition function interface Inputs { a: number; b: number; } export default function addNumbers({ a, b }: Inputs): number { return a + b; } ``` ## Exporting Multiple Worker Functions A single worker file may export multiple named handler functions: ```javascript tab={"label":"Javascript"} title="worker.js" "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; ``` ```javascript title="worker.ts" import { resolve } from "path"; export const filename = resolve(__filename); function add({ a, b }: { a: number, b: number }): number { return a + b; } function multiply({ a, b }: { a: number, b: number }): number { return a * b; } export { multiply, add }; ``` The export to target can then be specified when the task is submitted: ```javascript tab={"label":"Javascript"} title="main.js" "use strict"; const Piscina = require("piscina"); const { resolve } = require("path"); // Initialize Piscina with the worker file const piscina = new Piscina({ filename: resolve(__dirname, "worker.js"), }); // Run multiple tasks concurrently (async () => { const [sum, product] = await Promise.all([ piscina.run({ a: 4, b: 6 }, { name: "add" }), piscina.run({ a: 4, b: 6 }, { name: "multiply" }), ]); })(); ``` ```javascript tab={"label":"Typescript","span":2} title="main.ts" import Piscina from "piscina"; import { resolve } from "path"; import { filename } from "./worker"; const piscina = new Piscina({ filename: resolve(__dirname, "./workerWrapper.js"), workerData: { fullpath: filename }, }); // Run multiple tasks concurrently (async () => { const [sum, product] = await Promise.all([ piscina.run({ a: 4, b: 6 }, { name: "add" }), piscina.run({ a: 4, b: 6 }, { name: "multiply" }), ]); })(); ``` ## Cancellable Tasks Piscina supports task cancellation even if the task has already been submitted to a worker and is being actively processed. This can be used, for instance, to cancel tasks that are taking too long to run. Tasks can be canceled using an `AbortController` or an `EventEmitter`. ### Using `AbortController` ```javascript title="main.js" "use strict"; const Piscina = require("piscina"); const { resolve } = require("path"); // Set up Piscina with the worker file const piscina = new Piscina({ filename: resolve(__dirname, "worker.js"), }); // Run a task and cancel it using AbortController (async () => { const abortController = new AbortController(); const { signal } = abortController; const task = piscina.run({ a: 4, b: 6 }, { signal }); abortController.abort(); // Cancel the task try { await task; } catch (err) { console.log("The task was canceled"); // Handle the cancellation } })(); ``` ```javascript title="main.ts" import Piscina from "piscina"; import { resolve } from "path"; import { filename } from "./worker"; // Set up Piscina with the worker and workerWrapper file const piscina = new Piscina({ filename: resolve(__dirname, "./workerWrapper.js"), workerData: { fullpath: filename }, }); // Run a task and cancel it using AbortController (async () => { const abortController = new AbortController(); const { signal } = abortController; const task = piscina.run({ a: 4, b: 6 }, { signal }); abortController.abort(); // Cancel the task try { await task; } catch (err) { console.log("The task was canceled"); // Handle the cancellation } })(); ``` ### Using `EventEmitter` A plain Node.js EventEmitter can also be used here, in which case, emitting the abort event on it will have the same effect. ```javascript tab={"label":"Javascript"} title="main.js" "use strict"; const Piscina = require("piscina"); const EventEmitter = require("events"); const { resolve } = require("path"); // Initialize Piscina with the worker file const piscina = new Piscina({ filename: resolve(__dirname, "worker.js"), }); // Run a task and cancel it using an EventEmitter (async () => { const ee = new EventEmitter(); const task = piscina.run({ a: 4, b: 6 }, { signal: ee }); ee.emit("abort"); // Emit an 'abort' event to cancel the task try { await task; } catch (err) { console.log("The task was canceled"); // Handle the cancellation } })(); ``` ```javascript tab={"label":"Typescript","span":2} title="main.ts" import Piscina from "piscina"; import { resolve } from "path"; import { filename } from "./worker"; // Initialize Piscina with the worker and workerWrapper file const piscina = new Piscina({ filename: resolve(__dirname, "./workerWrapper.js"), workerData: { fullpath: filename }, }); // Run a task and cancel it using an EventEmitter (async () => { const ee = new EventEmitter(); const task = piscina.run({ a: 4, b: 6 }, { signal: ee }); ee.emit("abort"); // Emit an 'abort' event to cancel the task try { await task; } catch (err) { console.log("The task was canceled"); // Handle the cancellation } })(); ``` Find more use cases in the [Examples](../category/examples/) section of the documentation --- ### Docs Getting Started Managing Worker Threads (`docs/docs/getting-started/managing-worker-threads.mdx`) --- id: Managing Worker Threads sidebar_position: 3 --- import {WorkerWrapperComponent} from '@site/src/components/WorkerWrapper.mdx'; import Tabs from "@theme/Tabs"; import TabItem from "@theme/TabItem"; ## 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: ```js tab={"label":"Javascript"} title="worker.js" async function initialize() { await someAsyncInitializationActivity(); return ({ a, b }) => a + b; } module.exports = initialize(); ``` ```js tab={"label":"Typescript"} title="worker.ts" interface OperationInput { a: number; b: number; } export async function initialize(): Promise<(input: OperationInput) => number> { await someAsyncInitializationActivity(); return ({ a, b }) => a + b; } ``` Piscina will await the resolution of the exported Promise before marking the worker thread available. ## Managing Task Overload with 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. The example below shows how to use a Node.js stream to feed a Piscina worker pool: ```js tab={"label":"Javascript"} "use strict"; const { resolve } = require("path"); const Pool = require("piscina"); 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"); }); ``` ```js tab={"label":"Typescript","span":2} import Pool from "piscina"; import { resolve } from "path"; import { filename } from "./worker.js"; const pool = new Pool({ filename: resolve(__dirname, "./workerWrapper.js"), workerData: { fullpath: filename }, maxQueue: "auto", }); const stream = getStreamSomehow(); stream.setEncoding("utf8"); let counter = 0; pool.on("drain", () => { if (stream.isPaused()) { console.log("resuming...", counter, pool.queueSize); stream.resume(); } }); stream .on("data", (data: string) => { 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: ```js tab={"label":"Javascript"} 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; }; ``` ```js tab={"label":"Typescript"} import { setTimeout } from "timers/promises"; interface OperationInput { a: number; b: number; } export async ({ a, b }: OperationInput): Promise => { // This promise should be awaited await setTimeout(1000); console.log("Working"); // This will **not** run during the same worker call return a + b; }; ``` --- ### Docs Getting Started Typescript (`docs/docs/getting-started/typescript.mdx`) --- id: Typescript sidebar_position: 4 --- import {WorkerWrapperComponent} from '@site/src/components/WorkerWrapper.mdx'; Although Piscina itself is written in TypeScript and supports TypeScript out of the box, complication arises when trying to use `.ts` files directly as worker files because Node.js does not support TypeScript natively. To work around this, you would typically need to compile your TypeScript worker files to JavaScript first, and then point Piscina’s `filename` option to the compiled JavaScript files. Consider the following methods: ## Method 1: Using a Worker Wrapper File and `ts_node` The worker wrapper checks if the provided file path ends with `.ts` and registers the `ts-node` compiler to handle TypeScript files. In your `worker.ts`: ```typescript title='worker.ts' import { resolve } from 'path'; export const filename = resolve(__filename); interface Inputs { a: number; b: number; } export function addNumbers({ a, b }: Inputs): number { return a + b; } ``` Inside the main application: ```typescript title='main.ts' import Piscina from 'piscina'; import { resolve } from 'path'; import { filename } from './worker'; const piscina = new Piscina({ filename: resolve(__dirname, './workerWrapper.js'), workerData: { fullpath: filename }, }); (async () => { const result = await piscina.run({ a: 2, b: 3 }, { name: 'addNumbers' }); console.log('Result:', result); })(); ``` ## Method 2: Inline Worker Code Alternatively, you can include the worker code in the same file as your main code and use the `isMainThread` flag from `worker_threads` to determine whether the code is running in the main thread or the worker thread. ```typescript title="main.ts" 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; }; } ``` You can also check out this example on [github](https://github.com/piscinajs/piscina/tree/current/examples/typescript). ## Method 3: Node.js Native TypeScript Support Starting with Node.js 23.6.0+, you can run TypeScript files directly using Node.js's [experimental TypeScript support](https://nodejs.org/api/typescript.html). This eliminates the need for external TypeScript compilers or build steps. For Node.js versions 22.6.0 to 23.5.0, you need to explicitly enable the `--experimental-strip-types` flag. However, starting from Node.js 23.6.0+, type stripping is enabled by default. ```typescript title='worker.ts' interface Inputs { a: number; b: number; } export function addNumbers({ a, b }: Inputs): number { return a + b; } ``` ```typescript title='main.ts' import Piscina from 'piscina'; const piscina = new Piscina({ filename: new URL("./worker.ts", import.meta.url).href, }); (async () => { const result = await piscina.run({ a: 2, b: 3 }, { name: 'addNumbers' }); console.log('Result:', result); })(); ``` ## Using generics You can use generics to the types for the worker function: ```typescript title='worker.ts' interface Inputs { a: number; b: number; } export default function addNumbers({ a, b }: Inputs): number { return a + b; } ``` ```typescript title='main.ts' import Piscina from 'piscina'; const piscina = new Piscina({ filename: new URL("./worker.ts", import.meta.url).href, }); (async () => { const result = await piscina.run({ a: 2, b: 3 }); console.log('Result:', result); })(); ``` --- ### Docs Update Log Category .Json (`docs/docs/update-log/_category_.json`) { "label": "Update Log", "position": 6, "link": { "type": "generated-index", "description": "This section helps you stay updated about latest changes and bug fixes." } } --- ### Docs Update Log Changelog (`docs/docs/update-log/changelog.md`) --- title: "ChangeLog" sidebar_position: 1 --- import Changelog from '../../../CHANGELOG.md' --- ### Docs Update Log Release Notes (`docs/docs/update-log/release-notes.md`) --- id: Release Notes sidebar_position: 1 --- ### 4.1.0 #### Features * add `needsDrain` property ([#368](https://github.com/piscinajs/piscina/issues/368)) ([2d49b63](https://github.com/piscinajs/piscina/commit/2d49b63368116c172a52e2019648049b4d280162)) * correctly handle process.exit calls outside of a task ([#361](https://github.com/piscinajs/piscina/issues/361)) ([8e6d16e](https://github.com/piscinajs/piscina/commit/8e6d16e1dc23f8bb39772ed954f6689852ad435f)) #### Bug Fixes * Fix types for TypeScript 4.7 ([#239](https://github.com/piscinajs/piscina/issues/239)) ([a38fb29](https://github.com/piscinajs/piscina/commit/a38fb292e8fcc45cc20abab8668f82d908a24dc0)) * use CJS imports ([#374](https://github.com/piscinajs/piscina/issues/374)) ([edf8dc4](https://github.com/piscinajs/piscina/commit/edf8dc4f1a19e9b49e266109cdb70d9acc86f3ca)) ### 4.0.0 * Drop Node.js 14.x support * Add Node.js 20.x to CI ### 3.2.0 * Adds a new `PISCINA_DISABLE_ATOMICS` environment variable as an alternative way of disabling Piscina's internal use of the `Atomics` API. (https://github.com/piscinajs/piscina/pull/163) * Fixes a bug with transferable objects. (https://github.com/piscinajs/piscina/pull/155) * Fixes CI issues with TypeScript. (https://github.com/piscinajs/piscina/pull/161) ### 3.1.0 * Deprecates `piscina.runTask()`; adds `piscina.run()` as an alternative. https://github.com/piscinajs/piscina/commit/d7fa24d7515789001f7237ad6ae9ad42d582fc75 * Allows multiple exported handler functions from a single file. https://github.com/piscinajs/piscina/commit/d7fa24d7515789001f7237ad6ae9ad42d582fc75 ### 3.0.0 * Drops Node.js 10.x support * Updates minimum TypeScript target to ES2019 ### 2.1.0 * Adds name property to indicate `AbortError` when tasks are canceled using an `AbortController` (or similar) * More examples ### 2.0.0 * Added unmanaged file descriptor tracking * Updated dependencies ### 1.6.1 * Bug fix: Reject if AbortSignal is already aborted * Bug Fix: Use once listener for abort event ### 1.6.0 * Add the `niceIncrement` configuration parameter. ### 1.5.1 * Bug fixes around abortable task selection. ### 1.5.0 * Added `Piscina.move()` * Added Custom Task Queues * Added utilization metric * Wait for workers to be ready before considering them as candidates * Additional examples ### 1.4.0 * Added `maxQueue = 'auto'` to autocalculate the maximum queue size. * Added more examples, including an example of implementing a worker as a Node.js native addon. ### 1.3.0 * Added the `'drain'` event ### 1.2.0 * Added support for ESM and file:// URLs * Added `env`, `argv`, `execArgv`, and `workerData` options * More examples ### 1.1.0 * Added support for Worker Thread `resourceLimits` ### 1.0.0 * Initial release! --- ### Src Components WorkerWrapper (`docs/src/components/WorkerWrapper.mdx`) // WorkerWrapperComponent.mdx import React from 'react'; export const WorkerWrapperComponent = () => ( 
    {`const { workerData } = require('worker_threads');

if (workerData.fullpath.endsWith(".ts")) {
require("ts-node").register();
}
module.exports = require(workerData.fullpath);
`}
); export default WorkerWrapperComponent; --- --- METRICS --- - Files Extracted: 56 - Estimated Token Budget: ~58131 tokens - Canonical Reference: https://codewiki.google/github.com/piscinajs/piscina