@Nest/workers-pool

"NestJS Workers Pool" is a robust library for NestJS that enables you to execute functions in separate processes using worker_threads. This prevents the main thread from being blocked in heavy computation process and also allows you to control the number of functions running concurrently.

• It manages a pool of workers, automatically assigning tasks and queuing them when all workers are busy.

• runAsyncTask for awaiting task results, and runTask for non-blocking execution, you have the flexibility to choose based on your needs.

• When using the waitUntilAvailable option, tasks wait in the queue until a worker is free, preventing overloading.

• Workers are restarted after reaching their maximum tasks or lifetime, maintaining optimal performance.

• Extend or inject WorkerPoolService as needed, and use decorators for handling custom events in worker scripts.

• Use of decorador as @OnMessage or @onError to handle custom events in your service.

Installation

npm install nestjs-workers-pool

Usage as a extendable service

import { Injectable } from '@nestjs/common';
import { WorkerPoolService, onEmitted, onceEmitted, onError } from '@nestjs/workers-pool';

@Injectable()
export class FibonacciService extends WorkerPoolService {
  constructor() {
    super({
      scriptPath: './src/workers/heavy_task_worker.js',
      waitUntilAvailable: true,
      workerOptions: {
        execArgv: ['--max-old-space-size=4096'],
      },
    });
  }

  async calcFibonacci(n: number): Promise<boolean> {
    return this.runTask(n);
  }

  // by default the name of the method
  @onEmitted()
  progress(data: { progress: number }): void {
    console.log(`Progress: ${data.progress}`);
  }

  @onEmitted('message')
  message(data: { result: string }): void {
    console.log(`Message: ${data.result}`);
  }

  @onError()
    error(data: { error: string }): void {
        console.log(`Error: ${data.error}`);
    }
}

Usage as a inyectable service

// module.ts
import { Module } from '@nestjs/common';
import { WorkerPoolService, WORKER_OPTIONS_TOKEN } from '@nestjs/workers-pool';

@Module({
  providers: [
    {
      provide: WORKER_OPTIONS_TOKEN,
      useValue: {
        scriptPath: './src/workers/heavy_task_worker.js',
        waitUntilAvailable: true,
        maxWorkers: 2,
        maxLifetime: 10000,
        maxTasks: 10,
      },
    },
    WorkerPoolService,
  ],
})
export class AppModule {}

// app.service.ts
import { Injectable } from '@nestjs/common';

import { WorkersPool } from '@nest/workers-pool';

@Injectable()
export class AppService {
  constructor(private readonly workersPool: WorkersPool) {}

  async run() {
    // run task in a separate process, returns a boolean if the task was queued
    this.workersPool.runTask()

    // run task in a separate process, returns a result of the task
    const result = this.workersPool.runSyncTask()
    console.log(result)
  }
}

Worker script

// heavy_task_worker.js
const { parentPort } = require('worker_threads');

parentPort.on('message', (data) => {
  parentPort.postMessage({ progress: 0.1, type: 'progress' });
  // do something
  parentPort.postMessage({ progress: 1, type: 'progress' });
  parentPort.postMessage({ result: 'done', type: 'message' });
});

• To trigger the custom events you should use the type property in the message object with the name of the event.
• If you send an unhandled event, it will be ignored.
• You should always send a type: 'message' in the final result, or the worker will never be flag as available again and the task will be queued forever, once you send the type: 'message' it will be stop listening for events and the worker will continue with the lifecycle.

Configuration

| Option | Type | Default value | Description | | ----------------- | -------- | ------------- | --------------------------------------------------------------------------- | | scriptPath | string | | Path to the worker script | | maxWorkers | number | os.cpus().length | The maximum number of workers that can be started simultaneously | | waitUntilAvailable | boolean | false | Wait until the worker is ready to accept tasks, if is false, it will throw an error |
| workerOptions | object | {} | Options for the worker thread | | maxTasks | number | 0 | The maximum number of tasks that can be run | | maxLifetime | number | 0 | The maximum lifetime of a worker in milliseconds | | verbose | boolean | false | Show debug information |

handling custom events

There are several decorators available for handling different types of events:

• Use the @onEmitted() decorator to handle custom events. This will be executed each time the specified event is emitted.

• The @onceEmitted() decorator is also for handling custom events, but it will only be executed the first time the event is emitted.

• To handle errors, use the @onError() decorator. This will be executed each time an error raised in the worker.

• The @onExit() decorator is for handling the exit event. This will be executed each time the exit event is emitted.

• You can destroy all workers with the destroyWorkerPool() method if you need to, bt any pending task will be lost.

import { Injectable } from '@nestjs/common';
import { WorkerPoolService, onEmitted, onceEmitted, onError } from '@nestjs/workers-pool';

@Injectable()
export class FibonacciService extends WorkerPoolService {
  constructor() {
    super({
      scriptPath: './src/workers/heavy_task_worker.js',
      waitUntilAvailable: true,
      maxWorkers: 2,
    });
  }

  async calcFibonacci(n: number): Promise<boolean> {
    // you can send a transferable object to the worker
    const transferList = [n];
    return this.runTask(n, transferList);
  }

  // by default the name of the method
  @onEmitted()
  progress(data: { progress: number }): void {
    console.log(`Progress: ${data.progress}`);
  }

  @onEmitted('message')
  message(data: { result: string }): void {
    console.log(`Message: ${data.result}`);
  }

  @onError()
    error(data: { error: string }): void {
        console.log(`Error: ${data.error}`);
    }

  @onExit()
    exit(data: { exitCode: number }): void {
        console.log(`Exit: ${data.exitCode}`);
    }

}

// heavy_task_worker.js
const { parentPort } = require('worker_threads');

parentPort.on('message', (data) => {
  parentPort.postMessage({ progress: 0.1, type: 'progress' });
  // do something
  parentPort.postMessage({ progress: 1, type: 'progress' });
  parentPort.postMessage({ result: 'done', type: 'message' });
});

Future improvements

• Integrations with services like Bull or RabbitMQ
• Send intransferible objects to the worker like streams or sockets
• Integrations with TypeORM to handle database connections

This is my first library so any feedback is welcome, I will be happy to receive your comments and pull requests! :D @bueno12223