piscina Documentation

Technical Documentation: piscinajs/piscina

ℹ️ Provenance: Multi-Tier Ingestion · CodeWiki Reference

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.

Piscina API

Example

In main.js:

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

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

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

In worker.js:

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

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

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

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

ESM is also supported for both Piscina and workers:

import { Piscina } from "piscina";

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

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

In worker.mjs:

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

Exporting multiple worker functions

A single worker file may export multiple named handler functions.

"use strict";

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

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

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

module.exports = add;

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

"use strict";

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

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

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

Cancelable Tasks

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

"use strict";

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

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

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

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

"use strict";

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

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

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

Delaying Availability of Workers

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

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

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

module.exports = initialize();

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

Backpressure

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

Example: Using a Node.js stream to feed a Piscina worker pool:

"use strict";

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

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

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

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

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

Out of scope asynchronous code

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

For example:

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

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

  return a + b;
};

Broadcast a message to all worker threads

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

In main.js

"use strict";

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

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

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

main();

In worker.js

"use 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.
- maxThreads: (number) Sets the maximum number of threads that are
  running for this thread pool. The default is the number provided by os.availableParallelism * 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.
- 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 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 Promises 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;
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.

{
  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.

{
  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.

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,

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:

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;

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 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.

text

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:

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:

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 [email protected]
Anna Henningsen [email protected]
Matteo Collina [email protected]
Rafael Gonzaga [email protected]
Robert Nagy [email protected]
Carlos Fuentes [email protected]

Acknowledgements

Piscina development was initially sponsored by NearForm Research.

Resources

Overview (`docs/README.md`)

Website

This website is built using Docusaurus, a modern static website generator.

Installation

text

$ npm i

Local Development

text

$ 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

text

$ 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:

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:

text

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:

text

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:

text

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!

text

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:

text

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:

text

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.

Docs Api Reference Api Overview (`docs/docs/api-reference/api-overview.md`)

API	Description
Class: `Piscina`	This class extends `EventEmitter` from Node.js.
Constructor: `new Piscina([options])`	Creates a new instance of the `Piscina` class. List of available options
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`)

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, 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

(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

(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

Class: `Piscina`

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.
- maxThreads: (number) Sets the maximum number of threads that are
  running for this thread pool. The default is the number provided by os.availableParallelism * 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.
- 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 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
  - 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 module to be installed.
  See nice(2) and 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) By default, Piscina uses a least-busy algorithm. The loadBalancer
  option can be used to provide an alternative implementation. See Custom Load Balancers 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.
:::

`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:

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.

type PiscinaHistogram = {
  runTime: PiscinaHistogramSummary;
  waitTime: PiscinaHistogramSummary;
  resetRunTime(): void; // Reset Run Time Histogram
  resetWaitTime(): void; // Reset Wait Time Histogram
}

Interface: `PiscinaHistogramSummary`

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.

Interface: `PiscinaLoadBalancer`

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`

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`

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

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

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`)

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`)

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.

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`

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:

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`)

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.

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.

Method: `destroy()`

Stops all Workers and rejects all Promises 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`)

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

{
  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.

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

{
  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`)

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)`

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.

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`)

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<number> => {
  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");
}
})();

text

</TabItem>
<TabItem value="Typescript">
```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");
  }
})();
```
<WorkerWrapperComponent />
</TabItem>
</Tabs>

## Using EventEmitter

The `EventEmitter` instance is passed as the signal option to the `piscina.run` method.

<Tabs>
<TabItem value="Javascript">
```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');
  }
})();
```
</TabItem>
<TabItem value="Typescript">

```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');
  }
})();
```
<WorkerWrapperComponent />
</TabItem>
</Tabs>

## Using a Timer with EventEmitter

<Tabs>
<TabItem value="Javascript">
```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);
  }
})();
```
</TabItem>
<TabItem value="Typescript">
```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);
  }
})();
```
<WorkerWrapperComponent />
</TabItem>
</Tabs>

## Using a Timer with AbortController

<Tabs>
<TabItem value="Javascript">
```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);
  }
})();
```
</TabItem>
<TabItem value="Typescript">
    ```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);
  }
})();
```

<WorkerWrapperComponent />
    </TabItem>
</Tabs>

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.

<Tabs>
<TabItem  value="Javascript">
```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);
}
})();

text

<WorkerWrapperComponent/>

</TabItem>
</Tabs>

Both worker files demonstrate asynchronous loading. They use a sleep function to simulate an asynchronous operation that takes 500 milliseconds.

<Tabs>
<TabItem value="Javascript">
```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");
}

text

</TabItem>

</Tabs>

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.

Docs Examples Broadcast (`docs/docs/examples/broadcast.mdx`)

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.

Docs Examples Electron (`docs/docs/examples/electron.mdx`)

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.

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.

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.

Docs Examples Move (`docs/docs/examples/move.mdx`)

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.

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<number, TaskInput>({ a: 2, b: 3 }, { name: 'add' }),
    pool.run<number, TaskInput>({ 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.

Docs Examples Multiple Workers (`docs/docs/examples/multiple-workers.mdx`)

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.

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.

Docs Examples N Api (`docs/docs/examples/n-api.mdx`)

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.

Then install the dependencies:

console

npm i

Build the native addon artifacts:

console

npm run prebuild

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.

Docs Examples Named (`docs/docs/examples/named.mdx`)

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<string, Function>) {
  return async ({ op, args }: Task) => {
    return await obj[op](...args);
  };
}

You can also check out this example on github.

Docs Examples Progress (`docs/docs/examples/progress.mdx`)

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<void> {
  // 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);

text

// Send a message to indicate progress
port.postMessage(n);

}

// Close the port when the task is complete
port.close();
};

text

</TabItem>
<TabItem value="Typescript">
```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<void> {
  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.

Docs Examples React Ssr (`docs/docs/examples/react-ssr.mdx`)

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 `
  <!doctype html>
    <html>
    <body>
    <div id="root">${
      ReactDOMServer.renderToString(React.createElement(Greeting, { name }))
    }</div>
    ${
      ReactDOMServer.renderToString(React.createElement(Lorem))
    }
    <script src="/static/home.js"></script>
  </body>
  </html>`;
};

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 `
  <!doctype html>
    <html>
    <body>
    <div id="root">${
      ReactDOMServer.renderToString(React.createElement(Greeting, { name }))
    }</div>
    ${
      ReactDOMServer.renderToString(React.createElement(Lorem))
    }
    <script src="/static/home.js"></script>
  </body>
  </html>`;
}

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<GreetingProps> {
  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 `
  <!doctype html>
    <html>
    <body>
    <div id="root">${
      ReactDOMServer.renderToString(React.createElement(Greeting, { name }))
    }</div>
    ${
      ReactDOMServer.renderToString(React.createElement(Lorem))
    }
    <script src="/static/home.js"></script>
  </body>
  </html>`;
});

// 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 `
  <!doctype html>
    <html>
    <body>
    <div id="root">${
      ReactDOMServer.renderToString(React.createElement(Greeting, { name }))
    }</div>
    ${
      ReactDOMServer.renderToString(React.createElement(Lorem))
    }
    <script src="/static/home.js"></script>
  </body>
  </html>`;
});

// 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

text

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

text

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.

Docs Examples Resource Limit (`docs/docs/examples/resource-limit.mdx`)

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]);
}
}

text

</TabItem>
</Tabs>

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.

<Tabs>
<TabItem value="Javascript">

```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));
});

text

</TabItem>
</Tabs>

### Pooled Asynchronous

The pooled versions use Piscina to distribute the scrypt operations across multiple worker threads, potentially improving performance on multi-core systems.

<Tabs>
<TabItem value="Javascript">

/* Detailed source-code truncated for AI context efficiency. */

text

```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');
})();

text

```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<string> {
  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');
})();

text

</TabItem>
</Tabs>

### Pooled Synchronous

The synchronous versions use `scryptSync` and `randomFillSync`, which can be more efficient but may block the event loop.

<Tabs>
<TabItem value="Javascript">

```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;

text

<WorkerWrapperComponent/>
</TabItem>
</Tabs>

### Unpooled Synchronous

<Tabs>
<TabItem value="Javascript">

```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);

text

</TabItem>
</Tabs>

### Running the Example

The `package.json` file includes scripts to run each variant of the scrypt implementation with the monitor.

<Tabs>
<TabItem value="Javascript">
```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.

Docs Examples Server (`docs/docs/examples/server.mdx`)

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.

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:

text

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:

text

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 │
└───────────┴─────────┴─────────┴─────────┴─────────┴─────────┴───────┴─────────┘

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:

text

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 │
└───────────┴─────────┴─────────┴─────────┴─────────┴─────────┴────────┴─────────┘

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!

text

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 │
└───────────┴─────────┴─────────┴─────────┴─────────┴─────────┴───────┴─────────┘

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.

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:

text

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:

text

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.

You can also check out this example on github.

Docs Examples Simple (`docs/docs/examples/simple.mdx`)

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.

Docs Examples Simple Async (`docs/docs/examples/simple_async.mdx`)

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 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<number> => {
  // Fake some async activity
  await sleep(100);
  return a + b;
};
export const filename = __filename;

You can also check out this example on github.

Docs Examples Stream In (`docs/docs/examples/stream-in.mdx`)

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.

text

/* Detailed source-code truncated for AI context efficiency. */

text

/* 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;

text

</TabItem>
</Tabs>


A custom `Progress` class (`progress.js`) is used to track the progress of the task, emitting events and providing status messages.

<Tabs>
<TabItem value="Javascript">
```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;

text

</TabItem>
</Tabs>

A separate script (`generate.js`) is provided to generate sample CSV data for testing purposes.

<Tabs>
<TabItem value="Javascript">
```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');

text

</TabItem>
</Tabs>

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.

<Tabs>
<TabItem value="Javascript">
```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.

text

/* 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)
}

text

</TabItem>
</Tabs>


The worker script defines a function that processes data through a `MessagePortDuplex` stream, converting incoming text to uppercase and writing it back.
<Tabs>
<TabItem value="Javascript">
```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;
}

text

</TabItem>
</Tabs>

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.
:::

<Tabs>
<TabItem value="Javascript">

```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))
  ]));
})();

text

/* 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;
};

text

</TabItem>
<TabItem value="Typescript">
```typescript title="worker.ts"
import { promisify } from 'util';

const sleep = promisify(setTimeout);

export default async ({ priority }: { priority: number }): Promise<number> => {
  await sleep(100);
  process._rawDebug(`${priority}`);
  return priority;
};
export const filename = __filename;

You can also check out this example on github.

Docs Examples Webstreams Transform (`docs/docs/examples/webstreams-transform.mdx`)

import {WorkerWrapperComponent} from '@site/src/components/WorkerWrapper.mdx';
import Tabs from "@theme/Tabs";
import TabItem from "@theme/TabItem";

Using Web Streams 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<string> {
  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.

Docs Examples Webstreams (`docs/docs/examples/webstreams.mdx`)

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 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<void> {
  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.

Docs Examples Worker Options (`docs/docs/examples/worker-options.mdx`)

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.

Docs Getting Started Installation (`docs/docs/getting-started/Installation.mdx`)

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`)

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
})();

text

</TabItem>
<TabItem value="Typescript">
```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.

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;
}

text

</TabItem>
</Tabs>

## Using Async Functions or Promises

Workers can be asynchronous or return a promise:
<Tabs>
<TabItem value="Javascript">
```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;
};

text

</TabItem>
</Tabs>

## Supporting ECMAScript Modules (ESM)

Piscina.js also works with ECMAScript modules:
<Tabs>
<TabItem value="Javascript">
```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);

text

<WorkerWrapperComponent/>
</TabItem>
</Tabs>

In your worker module file:
<Tabs>
<TabItem value="Javascript">

```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;
}

text

</TabItem>
</Tabs>


## Exporting Multiple Worker Functions

A single worker file may export multiple named handler functions:
<Tabs>
<TabItem value="Javascript">
```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 };

text

</TabItem>
</Tabs>


The export to target can then be specified when the task is submitted:
<Tabs>
<TabItem value="Javascript">
```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
}
})();

text

</TabItem>
<TabItem value="Typescript">
```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
}
})();

text

<WorkerWrapperComponent/>

</TabItem>
</Tabs>

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:

<Tabs>
<TabItem value="Javascript">

```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

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");
});

text

<WorkerWrapperComponent/>
</TabItem>
</Tabs>


## 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:
<Tabs>
<TabItem value="Javascript">
```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;
};

text

</TabItem>
</Tabs>

---

### 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.

<WorkerWrapperComponent/>

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.

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. 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<typeof import("./worker.ts")>({
  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`)

import Changelog from '../../../CHANGELOG.md'

Docs Update Log Release Notes (`docs/docs/update-log/release-notes.md`)

4.1.0

Features

add needsDrain property (#368) (2d49b63)
correctly handle process.exit calls outside of a task (#361) (8e6d16e)

Bug Fixes

Fix types for TypeScript 4.7 (#239) (a38fb29)
use CJS imports (#374) (edf8dc4)

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 = () => (

text

<code className="language-js" title="workerWrapper.js">{`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

Technical Documentation: piscinajs/piscina

Overview (`README.md`) Copy Section

piscina - the node.js worker pool

Documentation Copy Chapter

Piscina API Copy Chapter

Example Copy Section

Exporting multiple worker functions Copy Section

Cancelable Tasks Copy Section

Delaying Availability of Workers Copy Section

Backpressure Copy Section

Out of scope asynchronous code Copy Section

Broadcast a message to all worker threads Copy Section

Additional Examples Copy Section

Class: `Piscina` Copy Chapter

Constructor: `new Piscina([options])` Copy Section

Method: `run(task[, options])` Copy Section

Method: `destroy()` Copy Section

Method: `close([options])` Copy Section

Event: `'error'` Copy Section

Event: `'drain'` Copy Section

Event: `'needsDrain'` Copy Section

Event: `'message'` Copy Section

Property: `completed` (readonly) Copy Section

Property: `duration` (readonly) Copy Section

Property: `options` (readonly) Copy Section

Property: `runTime` (readonly) Copy Section

Property: `threads` (readonly) Copy Section

Property: `idleThreads` (readonly) Copy Section

Property: `queueSize` (readonly) Copy Section

Property: `needsDrain` (readonly) Copy Section

Property: `utilization` (readonly) Copy Section

Property: `waitTime` (readonly) Copy Section

Static property: `isWorkerThread` (readonly) Copy Section

Static property: `version` (readonly) Copy Section

Static method: `move(value)` Copy Section

Interface: `Transferable`

Custom Task Queues Copy Chapter

Built-In Queues Copy Section

Using FixedQueue Example

Current Limitations (Things we're working on / would love help with) Copy Chapter

Performance Notes Copy Chapter

Queue Size Copy Section

Queue Pressure and Idle Threads Copy Section

Thread priority on Linux systems Copy Section

Multiple Thread Pools and Embedding Piscina as a Dependency Copy Section

The Team Copy Chapter

Acknowledgements Copy Chapter

Sponsors Copy Chapter

Bronze Sponsors Copy Section

Resources Copy Chapter

Overview (`docs/README.md`) Copy Section

Website

Installation Copy Section

Local Development Copy Section

Build Copy Section

Overview (`examples/n-api/README.md`) Copy Section

Piscina N-API (native addon) example

Overview (`examples/scrypt/README.md`) Copy Section

scrypt Example

Overview (`examples/server/README.md`) Copy Section

Server example

Tuning pool performance Copy Chapter

Docs Api Reference Api Overview (`docs/docs/api-reference/api-overview.md`) Copy Section

id: API Overview sidebar_position: 1 Copy Chapter

Package.Json (`docs/package.json`) Copy Section

Tsconfig.Json (`docs/tsconfig.json`) Copy Section

Docs Introduction (`docs/docs/Introduction.md`) Copy Section

sidebar_position: 1 slug: / Copy Chapter

Introduction

Why Piscina? Copy Chapter

Key features Copy Chapter

Sponsors Copy Chapter

Bronze Sponsors Copy Section

Docs Advanced Topics Category .Json (`docs/docs/advanced-topics/_category_.json`) Copy Section

Docs Advanced Topics Custom Task Queues (`docs/docs/advanced-topics/custom-task-queues.mdx`) Copy Section

id: Custom Task Queues sidebar_position: 1 Copy Chapter

Built-In Queues Copy Section

Using FixedQueue Copy Section

Docs Advanced Topics Loadbalancer (`docs/docs/advanced-topics/loadbalancer.mdx`) Copy Section

id: Custom Load Balancers sidebar_position: 2 Copy Chapter

Overview (`README.md`)

Documentation

Piscina API

Example

Exporting multiple worker functions

Cancelable Tasks

Delaying Availability of Workers

Backpressure

Out of scope asynchronous code

Broadcast a message to all worker threads

Additional Examples

Class: `Piscina`

Constructor: `new Piscina([options])`

Method: `run(task[, options])`

Method: `destroy()`

Method: `close([options])`

Event: `'error'`

Event: `'drain'`

Event: `'needsDrain'`

Event: `'message'`

Property: `completed` (readonly)

Property: `duration` (readonly)

Property: `options` (readonly)

Property: `runTime` (readonly)

Property: `threads` (readonly)

Property: `idleThreads` (readonly)

Property: `queueSize` (readonly)

Property: `needsDrain` (readonly)

Property: `utilization` (readonly)

Property: `waitTime` (readonly)

Static property: `isWorkerThread` (readonly)

Static property: `version` (readonly)

Static method: `move(value)`

Custom Task Queues

Built-In Queues

Current Limitations (Things we're working on / would love help with)

Performance Notes

Queue Size

Queue Pressure and Idle Threads

Thread priority on Linux systems

Multiple Thread Pools and Embedding Piscina as a Dependency

The Team

Acknowledgements

Sponsors

Bronze Sponsors

Resources

Overview (`docs/README.md`)

Installation

Local Development

Build

Overview (`examples/n-api/README.md`)

Overview (`examples/scrypt/README.md`)

Overview (`examples/server/README.md`)

Tuning pool performance

Docs Api Reference Api Overview (`docs/docs/api-reference/api-overview.md`)

id: API Overview sidebar_position: 1

Package.Json (`docs/package.json`)

Tsconfig.Json (`docs/tsconfig.json`)

Docs Introduction (`docs/docs/Introduction.md`)

sidebar_position: 1 slug: /

Why Piscina?

Key features

Sponsors

Bronze Sponsors

Docs Advanced Topics Category .Json (`docs/docs/advanced-topics/_category_.json`)

Docs Advanced Topics Custom Task Queues (`docs/docs/advanced-topics/custom-task-queues.mdx`)

id: Custom Task Queues sidebar_position: 1

Built-In Queues

Using FixedQueue

Docs Advanced Topics Loadbalancer (`docs/docs/advanced-topics/loadbalancer.mdx`)

id: Custom Load Balancers sidebar_position: 2

Choosing the Load Balancing Algorithm

Resources

Docs Advanced Topics Performance (`docs/docs/advanced-topics/performance.mdx`)

id: Performance Notes sidebar_position: 3

Queue Size

Queue Pressure and Idle Threads

Thread priority on Linux systems

Multiple Thread Pools and Embedding Piscina as a Dependency

Docs Api Reference Category .Json (`docs/docs/api-reference/_category_.json`)