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