src/pools/thread/fixed.ts

   1 import {
   2   SHARE_ENV,
   3   Worker,
   4   type WorkerOptions,
   5   isMainThread
   6 } from 'node:worker_threads'
   7 import type { MessageValue } from '../../utility-types'
   8 import { AbstractPool } from '../abstract-pool'
   9 import {
  10   type PoolOptions,
  11   type PoolType,
  12   PoolTypes,
  13   type WorkerType,
  14   WorkerTypes
  15 } from '../pool'
  16
  17 /**
  18  * Options for a poolifier thread pool.
  19  */
  20 export interface ThreadPoolOptions extends PoolOptions<Worker> {
  21   /**
  22    * Worker options.
  23    *
  24    * @see https://nodejs.org/api/worker_threads.html#new-workerfilename-options
  25    */
  26   workerOptions?: WorkerOptions
  27 }
  28
  29 /**
  30  * A thread pool with a fixed number of threads.
  31  *
  32  * It is possible to perform tasks in sync or asynchronous mode as you prefer.
  33  *
  34  * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
  35  * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
  36  * @author [Alessandro Pio Ardizio](https://github.com/pioardi)
  37  * @since 0.0.1
  38  */
  39 export class FixedThreadPool<
  40   Data = unknown,
  41   Response = unknown
  42 > extends AbstractPool<Worker, Data, Response> {
  43   /**
  44    * Constructs a new poolifier fixed thread pool.
  45    *
  46    * @param numberOfThreads - Number of threads for this pool.
  47    * @param filePath - Path to an implementation of a `ThreadWorker` file, which can be relative or absolute.
  48    * @param opts - Options for this fixed thread pool.
  49    */
  50   public constructor (
  51     numberOfThreads: number,
  52     filePath: string,
  53     protected readonly opts: ThreadPoolOptions = {}
  54   ) {
  55     super(numberOfThreads, filePath, opts)
  56   }
  57
  58   /** @inheritDoc */
  59   protected isMain (): boolean {
  60     return isMainThread
  61   }
  62
  63   /** @inheritDoc */
  64   protected async destroyWorker (worker: Worker): Promise<void> {
  65     this.sendToWorker(worker, { kill: 1 })
  66     await worker.terminate()
  67   }
  68
  69   /** @inheritDoc */
  70   protected sendToWorker (worker: Worker, message: MessageValue<Data>): void {
  71     worker.postMessage(message)
  72   }
  73
  74   /** @inheritDoc */
  75   protected createWorker (): Worker {
  76     return new Worker(this.filePath, {
  77       env: SHARE_ENV,
  78       ...this.opts.workerOptions
  79     })
  80   }
  81
  82   /** @inheritDoc */
  83   protected get type (): PoolType {
  84     return PoolTypes.fixed
  85   }
  86
  87   /** @inheritDoc */
  88   protected get worker (): WorkerType {
  89     return WorkerTypes.thread
  90   }
  91
  92   /** @inheritDoc */
  93   protected get minSize (): number {
  94     return this.numberOfWorkers
  95   }
  96
  97   /** @inheritDoc */
  98   protected get maxSize (): number {
  99     return this.numberOfWorkers
 100   }
 101
 102   /** @inheritDoc */
 103   protected get busy (): boolean {
 104     return this.internalBusy()
 105   }
 106 }