Gathering detailed insights and metrics for piscina
Gathering detailed insights and metrics for piscina
Gathering detailed insights and metrics for piscina
Gathering detailed insights and metrics for piscina
tinypool
A minimal and tiny Node.js Worker Thread Pool implementation, a fork of piscina, but with fewer features
fastify-piscina
A Fastify Piscina Plugin
@delight-rpc/piscina
```sh npm install --save @delight-rpc/piscina # or yarn add @delight-rpc/piscina ```
piscina-microjob
CPU bound routines over Piscina thread pool
A fast, efficient Node.js Worker Thread Pool implementation
npm install piscina
Typescript
Module System
Min. Node Version
Node Version
NPM Version
60.4
Supply Chain
99.6
Quality
95.1
Maintenance
100
Vulnerability
100
License
TypeScript (96.85%)
JavaScript (3.15%)
Total
323,220,917
Last Day
130,482
Last Week
3,859,690
Last Month
15,630,022
Last Year
154,164,935
4,416 Stars
300 Commits
111 Forks
26 Watching
13 Branches
36 Contributors
Latest Version
5.0.0-alpha.0
Package Id
piscina@5.0.0-alpha.0
Unpacked Size
366.49 kB
Size
88.20 kB
File Count
136
NPM Version
10.8.3
Node Version
22.9.0
Publised On
04 Dec 2024
Cumulative downloads
Total Downloads
Last day
20.4%
130,482
Compared to previous day
Last week
8.5%
3,859,690
Compared to previous week
Last month
0.5%
15,630,022
Compared to previous month
Last year
54.2%
154,164,935
Compared to previous year
11
1
Written in TypeScript.
For Node.js 18.x and higher.
In main.js
:
1const path = require('path'); 2const Piscina = require('piscina'); 3 4const piscina = new Piscina({ 5 filename: path.resolve(__dirname, 'worker.js') 6}); 7 8(async function() { 9 const result = await piscina.run({ a: 4, b: 6 }); 10 console.log(result); // Prints 10 11})();
In worker.js
:
1module.exports = ({ a, b }) => { 2 return a + b; 3};
The worker may also be an async function or may return a Promise:
1const { setTimeout } = require('timers/promises'); 2 3module.exports = async ({ a, b }) => { 4 // Fake some async activity 5 await setTimeout(100); 6 return a + b; 7};
ESM is also supported for both Piscina and workers:
1import { Piscina } from 'piscina'; 2 3const piscina = new Piscina({ 4 // The URL must be a file:// URL 5 filename: new URL('./worker.mjs', import.meta.url).href 6}); 7 8const result = await piscina.run({ a: 4, b: 6 }); 9console.log(result); // Prints 10
In worker.mjs
:
1export default ({ a, b }) => { 2 return a + b; 3};
A single worker file may export multiple named handler functions.
1'use strict'; 2 3function add({ a, b }) { return a + b; } 4 5function multiply({ a, b }) { return a * b; } 6 7add.add = add; 8add.multiply = multiply; 9 10module.exports = add;
The export to target can then be specified when the task is submitted:
1'use strict'; 2 3const Piscina = require('piscina'); 4const { resolve } = require('path'); 5 6const piscina = new Piscina({ 7 filename: resolve(__dirname, 'worker.js') 8}); 9 10(async function() { 11 const res = await Promise.all([ 12 piscina.run({ a: 4, b: 6 }, { name: 'add' }), 13 piscina.run({ a: 4, b: 6 }, { name: 'multiply' }) 14 ]); 15})();
Submitted tasks may be canceled using either an AbortController
or
an EventEmitter
:
1'use strict'; 2 3const Piscina = require('piscina'); 4const { resolve } = require('path'); 5 6const piscina = new Piscina({ 7 filename: resolve(__dirname, 'worker.js') 8}); 9 10(async function() { 11 const abortController = new AbortController(); 12 try { 13 const { signal } = abortController; 14 const task = piscina.run({ a: 4, b: 6 }, { signal }); 15 abortController.abort(); 16 await task; 17 } catch (err) { 18 console.log('The task was canceled'); 19 } 20})();
Alternatively, any EventEmitter
that emits an 'abort'
event
may be used as an abort controller:
1'use strict'; 2 3const Piscina = require('piscina'); 4const EventEmitter = require('events'); 5const { resolve } = require('path'); 6 7const piscina = new Piscina({ 8 filename: resolve(__dirname, 'worker.js') 9}); 10 11(async function() { 12 const ee = new EventEmitter(); 13 try { 14 const task = piscina.run({ a: 4, b: 6 }, { signal: ee }); 15 ee.emit('abort'); 16 await task; 17 } catch (err) { 18 console.log('The task was canceled'); 19 } 20})();
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:
1async function initialize() { 2 await someAsyncInitializationActivity(); 3 return ({ a, b }) => a + b; 4} 5 6module.exports = initialize();
Piscina will await the resolution of the exported Promise before marking the worker thread available.
When the maxQueue
option is set, once the Piscina
queue is full, no
additional tasks may be submitted until the queue size falls below the
limit. The 'drain'
event may be used to receive notification when the
queue is empty and all tasks have been submitted to workers for processing.
Example: Using a Node.js stream to feed a Piscina worker pool:
1'use strict'; 2 3const { resolve } = require('path'); 4const Pool = require('../..'); 5 6const pool = new Pool({ 7 filename: resolve(__dirname, 'worker.js'), 8 maxQueue: 'auto' 9}); 10 11const stream = getStreamSomehow(); 12stream.setEncoding('utf8'); 13 14pool.on('drain', () => { 15 if (stream.isPaused()) { 16 console.log('resuming...', counter, pool.queueSize); 17 stream.resume(); 18 } 19}); 20 21stream 22 .on('data', (data) => { 23 pool.run(data); 24 if (pool.queueSize === pool.options.maxQueue) { 25 console.log('pausing...', counter, pool.queueSize); 26 stream.pause(); 27 } 28 }) 29 .on('error', console.error) 30 .on('end', () => { 31 console.log('done'); 32 });
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:
1const { setTimeout } = require('timers/promises'); 2 3module.exports = ({ a, b }) => { 4 // This promise should be awaited 5 setTimeout(1000).then(() => { 6 console.log('Working'); // This will **not** run during the same worker call 7 }); 8 9 return a + b; 10};
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
1'use strict'; 2 3const { BroadcastChannel } = require('worker_threads'); 4const { resolve } = require('path'); 5 6const Piscina = require('piscina'); 7const piscina = new Piscina({ 8 filename: resolve(__dirname, 'worker.js'), 9 atomics: 'disabled' 10}); 11 12async function main () { 13 const bc = new BroadcastChannel('my_channel'); 14 // start worker 15 Promise.all([ 16 piscina.run('thread 1'), 17 piscina.run('thread 2') 18 ]); 19 // post message in one second 20 setTimeout(() => { 21 bc.postMessage('Main thread message'); 22 }, 1000); 23} 24 25main(); 26
In worker.js
1'use strict'; 2const { BroadcastChannel } = require('worker_threads'); 3 4module.exports = async (thread) => { 5 const bc = new BroadcastChannel('my_channel'); 6 bc.onmessage = (event) => { 7 console.log(thread + ' Received from:' + event.data); 8 }; 9 await new Promise((resolve) => { 10 setTimeout(resolve, 2000); 11 }); 12}; 13
Additional examples can be found in the GitHub repo at https://github.com/piscinajs/piscina/tree/master/examples
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.
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. 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 (stoping 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 asynchrnous 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: 4env
: (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
.Use caution when setting resource limits. Setting limits that are too low may
result in the Piscina
worker threads being unusable.
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.
destroy()
Stops all Workers and rejects all Promise
s for pending tasks.
This returns a Promise
that is fulfilled once all threads have stopped.
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.
'error'
An 'error'
event is emitted by instances of this class when:
All other errors are reported by rejecting the Promise
returned from
run()
, including rejections reported by the handler function
itself.
'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.
'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.
'message'
A 'message'
event is emitted whenever a message is received from a worker thread.
completed
(readonly)The current number of completed tasks.
duration
(readonly)The length of time (in milliseconds) since this Piscina
instance was
created.
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.
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 tasksrunTime.mean
{number
} The mean run time of all tasksrunTime.stddev
{number
} The standard deviation of collected run timesrunTime.min
{number
} The fastest recorded run timerunTime.max
{number
} The slowest recorded run timeAll 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.
1{ 2 average: 1880.25, 3 mean: 1880.25, 4 stddev: 1.93, 5 min: 1877, 6 max: 1882.0190887451172, 7 p0_001: 1877, 8 p0_01: 1877, 9 p0_1: 1877, 10 p1: 1877, 11 p2_5: 1877, 12 p10: 1877, 13 p25: 1877, 14 p50: 1881, 15 p75: 1881, 16 p90: 1882, 17 p97_5: 1882, 18 p99: 1882, 19 p99_9: 1882, 20 p99_99: 1882, 21 p99_999: 1882 22}
threads
(readonly)An Array of the Worker
instances used by this pool.
queueSize
(readonly)The current number of tasks waiting to be assigned to a Worker thread.
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.
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
.
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 taskswaitTime.mean
{number
} The mean wait time of all taskswaitTime.stddev
{number
} The standard deviation of collected wait timeswaitTime.min
{number
} The fastest recorded wait timewaitTime.max
{number
} The longest recorded wait timeAll 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.
1{ 2 average: 1880.25, 3 mean: 1880.25, 4 stddev: 1.93, 5 min: 1877, 6 max: 1882.0190887451172, 7 p0_001: 1877, 8 p0_01: 1877, 9 p0_1: 1877, 10 p1: 1877, 11 p2_5: 1877, 12 p10: 1877, 13 p25: 1877, 14 p50: 1881, 15 p75: 1881, 16 p90: 1882, 17 p97_5: 1882, 18 p99: 1882, 19 p99_9: 1882, 20 p99_99: 1882, 21 p99_999: 1882 22}
isWorkerThread
(readonly)Is true
if this code runs inside a Piscina
threadpool as a Worker.
version
(readonly)Provides the current version of this library as a semver string.
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 transfered. The Piscina.move()
method can be used to wrap and
mark transferable values such that they will by transfered 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.
1const { move } = require('piscina'); 2 3module.exports = () => { 4 return move(new ArrayBuffer(10)); 5}
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 transfering the object it wraps.
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,
1const { 2 move, 3 transferableSymbol, 4 valueSymbol 5} = require('piscina'); 6 7module.exports = () => { 8 const obj = { 9 a: { b: new Uint8Array(5); }, 10 c: { new Uint8Array(10); }, 11 12 get [transferableSymbol]() { 13 // Transfer the two underlying ArrayBuffers 14 return [this.a.b.buffer, this.c.buffer]; 15 } 16 17 get [valueSymbol]() { 18 return { a: { b: this.a.b }, c: this.c }; 19 } 20 }; 21 return move(obj); 22};
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:
1interface Task { 2 readonly [Piscina.queueOptionsSymbol] : object | null; 3} 4 5interface TaskQueue { 6 readonly size : number; 7 shift () : Task | null; 8 remove (task : Task) : void; 9 push (task : Task) : void; 10}
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).
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.
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.
Here's an example of how to use the FixedQueue
:
1const { Piscina, FixedQueue } = require('piscina'); 2const { resolve } = require('path'); 3 4// Create a Piscina pool with FixedQueue 5const piscina = new Piscina({ 6 filename: resolve(__dirname, 'worker.js'), 7 taskQueue: new FixedQueue() 8}); 9 10// Submit tasks to the pool 11for (let i = 0; i < 10; i++) { 12 piscina.run({ data: i }).then((result) => { 13 console.log(result); 14 }).catch((error) => { 15 console.error(error); 16 }); 17}
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).
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.
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.
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:
1const Piscina = require('piscina'); 2const pool = new Piscina({ 3 worker: '/absolute/path/to/worker.js', 4 niceIncrement: 20 5});
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.
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.
Piscina development is sponsored by NearForm Research.
No vulnerabilities found.
Reason
29 commit(s) and 7 issue activity found in the last 90 days -- score normalized to 10
Reason
no dangerous workflow patterns detected
Reason
no binaries found in the repo
Reason
license file detected
Details
Reason
Found 7/10 approved changesets -- score normalized to 7
Reason
3 existing vulnerabilities detected
Details
Reason
security policy file not detected
Details
Reason
project is not fuzzed
Details
Reason
no effort to earn an OpenSSF best practices badge detected
Reason
detected GitHub workflow tokens with excessive permissions
Details
Reason
dependency not pinned by hash detected -- score normalized to 0
Details
Reason
SAST tool is not run on all commits -- score normalized to 0
Details
Score
Last Scanned on 2024-12-02
The Open Source Security Foundation is a cross-industry collaboration to improve the security of open source software (OSS). The Scorecard provides security health metrics for open source projects.
Learn More