DOCSEARCHbyAlgolia
Submit Docs

tinypool

repository·main·Indexed 23 days ago

https://github.com/tinylibs/tinypool

A minimal, zero-dependency Node.js worker pool implementation and fork of piscina. It supports both `node:worker_threads` and `node:child_process` runtimes, allowing for task execution in separate threads or processes. Key features include configurable thread limits, worker recycling based on memory limits, task queue management, and support for Transferable objects via `transferList`.

Tokens
3.5K
Snippets
4
Records
20
Agent score
82%

What's inside tinypool

  1. Basic usage with node:worker_threads

    main

    By default, Tinypool uses node:worker_threads. You initialize a pool by providing a filename pointing to your worker script. You then use pool.run(data) to execute tasks. Always remember to call await pool.destroy() when the pool is no longer needed to terminate idle workers and free resources.

    // main.mjs
import Tinypool from 'tinypool'

const pool = new Tinypool({
  filename: new URL('./worker.mjs', import.meta.url).href,
})
const result = await pool.run({ a: 4, b: 6 })
console.log(result) // Prints 10

// Make sure to destroy pool once it's not needed anymore
// This terminates all pool's idle workers
await pool.destroy()
    // worker.mjs
export default ({ a, b }) => {
  return a + b
}

  2. Basic usage with node:child_process

    main

    To use node:child_process instead of worker threads, set the runtime option to 'child_process' in the Tinypool constructor. This is useful if you need to run tasks in separate processes rather than threads.

    // main.mjs
import Tinypool from 'tinypool'

const pool = new Tinypool({
  runtime: 'child_process',
  filename: new URL('./worker.mjs', import.meta.url).href,
})
const result = await pool.run({ a: 4, b: 6 })
console.log(result) // Prints 10
    // worker.mjs
export default ({ a, b }) => {
  return a + b
}

  3. Configure the Tinypool constructor

    main

    The Tinypool constructor accepts an options object to configure the pool behavior.

    Key options include:

    • runtime: Specifies the worker runtime. Defaults to 'worker_threads'. Supported values: 'worker_threads', 'child_process'.
    • isolateWorkers: (Boolean) If enabled, starts a fresh worker for every task. Disabled by default.
    • terminateTimeout: (Number) Milliseconds to wait for a worker to terminate before raising an error. Disabled by default.
    • maxMemoryLimitBeforeRecycle: (Number) If a worker's heap memory exceeds this value after a task, the worker is replaced. Useful for managing memory leaks.
    • teardown: (String) The name of a named export in the worker file that should be called before termination.
    • serialization: (String) Serialization type for child_process runtime. Values: 'json', 'advanced'.

  4. Identify and handle Tinypool internal messages

    main

    Tinypool uses internal messaging for communication between the main thread and workers. If you are intercepting or inspecting all messages sent to a worker, you can identify Tinypool's internal messages by checking for the __tinypool_worker_message__ property. To avoid side effects, you should ignore messages that contain this property.

    Internal messages follow this structure:

    {
  __tinypool_worker_message__: true,
  source: 'port' | 'pool'
}

  5. Handle errors and task cancellation

    main

    Tasks in Tinypool can fail in several ways:

    • Task Errors: If the worker script throws an error, the promise returned by .run() will reject with that error.
    • AbortError: If you provide an AbortSignal and it triggers, the task promise rejects with an AbortError.
    • CancelError: If a task is explicitly cancelled via the queue, it rejects with a CancelError.
    • Thread Termination: If a worker thread is terminated while a task is running, the task will reject with an error stating 'Terminating worker thread'.

  6. Communicate between main and worker threads using MessageChannel

    main

    When using the default worker_threads runtime, you can establish two-way communication by passing a MessagePort (from a MessageChannel) through the pool.run() method's transferList option. This allows the main thread and the worker to exchange messages directly.

    // main.mjs
import Tinypool from 'tinypool'
import { MessageChannel } from 'node:worker_threads'

const pool = new Tinypool({
  filename: new URL('./worker.mjs', import.meta.url).href,
})
const { port1, port2 } = new MessageChannel()
const promise = pool.run({ port: port1 }, { transferList: [port1] })

port2.on('message', (message) => console.log('Main thread received:', message))
setTimeout(() => port2.postMessage('Hello from main thread!'), 1000)

await promise

port1.close()
port2.close()
    // worker.mjs
export default ({ port }) => {
  return new Promise((resolve) => {
    port.on('message', (message) => {
      console.log('Worker received:', message)

      port.postMessage('Hello from worker thread!')
      resolve()
    })
  })
}

  7. Communicate between main and worker processes using TinypoolChannel

    main

    When using the child_process runtime, you can implement custom communication by passing a channel object to the pool.run() method. The worker communicates via process.send(). Note that you must filter out Tinypool's internal messages by checking for the __tinypool_worker_message__ property on incoming messages.

    // main.mjs
import Tinypool from 'tinypool'

const pool = new Tinypool({
  runtime: 'child_process',
  filename: new URL('./worker.mjs', import.meta.url).href,
})

const messages = []
const listeners = []
const channel = {
  onMessage: (listener) => listeners.push(listener),
  postMessage: (message) => messages.push(message),
}

const promise = pool.run({}, { channel })

// Send message to worker
setTimeout(
  () => listeners.forEach((listener) => listener('Hello from main process')),
  1000
)

// Wait for task to finish
await promise

console.log(messages)
// [{ received: 'Hello from main process', response: 'Hello from worker' }]
    // worker.mjs
export default async function run() {
  return new Promise((resolve) => {
    process.on('message', (message) => {
      // Ignore Tinypool's internal messages
      if (message?.__tinypool_worker_message__) return

      process.send({ received: message, response: 'Hello from worker' })
      resolve()
    })
  })
}

  8. Use Tinypool pool methods

    main

    The Tinypool instance provides methods to manage the lifecycle and tasks of the pool:

    • cancelPendingTasks(): Gracefully cancels all tasks currently waiting in the queue without affecting tasks that are already running. Use this when tasks have side effects that shouldn't be interrupted forcefully.
    • recycleWorkers(options): Waits for all current tasks to finish and then re-creates all workers. This can be used to force environment isolation even if isolateWorkers is disabled. Accepts a { runtime } option.

  9. Configure Tinypool via Options

    main

    When creating a Tinypool instance, you can provide an Options object to control the pool's behavior. Key options include:

    • filename: The path to the worker script (required if not provided in .run()).
    • runtime: Either 'worker_threads' (default) or 'child_process'.
    • minThreads: Minimum number of workers to keep alive. Supports fractional values (e.g., 0.5 will be calculated as Math.floor(0.5 * cpuCount)).
    • maxThreads: Maximum number of workers. Supports fractional values.
    • idleTimeout: Time in milliseconds before an idle worker is removed.
    • maxQueue: Maximum number of tasks allowed in the queue. Use 'auto' to set it to maxThreads ** 2.
    • concurrentTasksPerWorker: How many tasks a single worker can handle simultaneously.
    • isolateWorkers: If true, workers are recycled (destroyed) immediately after a task completes.
    • maxMemoryLimitBeforeRecycle: Threshold for memory usage to trigger worker recycling.
    • workerData: Data passed to the workers upon initialization.
    • env: Environment variables for the workers.
    • argv: Command line arguments for the workers.

  10. Cancel pending tasks in the pool

    main
    You can clear all tasks currently waiting in the queue by calling pool.cancelPendingTasks(). This will prevent queued tasks from starting, but will not affect tasks that are already running in workers.

  11. Run tasks with RunOptions

    main

    The .run(task, options) method allows you to override or extend pool settings for a specific task execution.

    Key RunOptions include:

    • transferList: A list of objects to transfer (e.g., ArrayBuffer, MessagePort) to the worker to avoid copying.
    • signal: An AbortSignal to allow cancelling the specific task.
    • filename: Override the pool's default worker filename.
    • runtime: Override the pool's default runtime.
    • channel: A TinypoolChannel for direct communication.

  12. Recycle workers in the pool

    main
    Use pool.recycleWorkers(options) to force the pool to replace its current workers. This is useful if you need to change the runtime (e.g., switching from worker_threads to child_process) or if you want to clear out workers that have been running for a long time. If isolateWorkers is enabled, workers are recycled automatically after tasks.