🧵 threadpool

thread pool with ergonomic IPC

Install

npm i @nicholaswmin/threadpool

Example

messaging between the primary and 4 threads:

// primary.js

import { Threadpool } from '@nicholaswmin/threadpool'

const pool = new Threadpool('thread.js', 4)

await pool.start()

pool
  .on('pong', () => {
    console.log('🏓 pong')
    pool.emit('ping')
  })
  .emit('ping')

and:

// thread.js 

import { primary } from '@nicholaswmin/threadpool'

primary.on('ping', () => {
  console.log('ping 🏓')

  setTimeout(() => primary.emit('pong'), 100)
})

then:

node primary.js

logs:

# ping 🏓
# 🏓 pong
# ping 🏓
# 🏓 pong
# ...

API

`new Threadpool(path, size, env)`

Creates a pool.

name	type	description	default
`path`	`String`	file path of thread code	current path
`size`	`Number`	number of threads	available cores
`env`	`Object`	Thread env. variables	primary `process.env`

`await pool.start()`

Starts the pool.

`await pool.stop()`

Shutsdown all threads, removes all listeners and stops the pool

Returns array of exit codes.

Messaging

primary-to-thread IPC:

`pool.on(name, listener)`

Listens for an emitted event, across all threads.

name	type	description
`name`	`String`	name of event
`listener`	`Function`	callback function

`pool.once(name, listener)`

Listens for an emitted event once, across all threads.
As soon as the listener fires it is removed.

`pool.off(name, listener)`

Removes a listener of a given event, across all threads.

`pool.removeAllListeners(name)`

Removes all listeners of a given event, across all threads.

`pool.emit(name, data)`

Sends the event to a single thread, chosen in round-robin.

`pool.broadcast(name, data)`

Sends the event to every thread, in fan-out

Emitted Events

`'pool-error'`

Emitted if an uncaught error is thrown in a thread.
The error is provided as an Error in a listener argument.

A shutdown is attempted before emitting this event.

If the shutdown fails, the Error instance will contain the shutdown error and the error.cause will contain the originating thread error.

Thread API

`thread.pid`

Thread's Process ID

`thread.exitCode`

null: is alive
0: exited with exit-code: 0
1: threw uncaught exception or killed with any signal other than SIGTERM.

Primary API

For thread-to-primary IPC, for usage in the thread code file

`primary.on(name, listener)`

Listen for events emitted from the primary.

`primary.emit(name, data)`

Emit an event to the primary.

Graceful exits

Use beforeExit to call pool.stop(), like so:

// primary.js

process.on('beforeExit', async () => {
  try {
    await pool.stop()
    process.exit(0)
  } catch (err) {
    console.error(err)
    process.exit(1)
  }
})

Timeouts

Threads which block the event loop or delay their termination are issued a SIGKILL signal, after a set timeout.

timeouts are in ms and can be set like so:

import { Threadpool } from '@nicholaswmin/threadpool'

Threadpool.spawnTimeout = 500
Threadpool.readyTimeout = 500
Threadpool.killTimeout  = 500

const pool = new Threadpool('thread.js')

// ... rest of code

Gotchas

Runtime exceptions trigger a shutdown/stop.
Cyclic pool.broadcasts can create an exponentially-increasing send rate.
Based on fork() so technically it's multi-processing, each "thread" being an isolated V8 instance.

Test

NODE_ENV=test node --run test

test coverage saved as: test/lcov.info

Benchmark

Run a ping/pong benchmark

node --run benchmark -- --size=4 --kibs=10

4 threads, each ping sending 10 kilobytes of event data

Contributing

Follows Semver, Github Flow & Conventional Commits.
Changes must be accompanied by 100% unit-test coverage.

Authors

@nicholaswmin

License

The MIT License