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  * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
  33  * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
  34  * @author [Alessandro Pio Ardizio](https://github.com/pioardi)
  35  * @since 0.0.1
  36  */
  37 export class FixedThreadPool<
  38   Data = unknown,
  39   Response = unknown
  40 > extends AbstractPool<Worker, Data, Response> {
  41   /**
  42    * Constructs a new poolifier fixed thread pool.
  43    *
  44    * @param numberOfThreads - Number of threads for this pool.
  45    * @param filePath - Path to an implementation of a `ThreadWorker` file, which can be relative or absolute.
  46    * @param opts - Options for this fixed thread pool.
  47    */
  48   public constructor (
  49     numberOfThreads: number,
  50     filePath: string,
  51     protected readonly opts: ThreadPoolOptions = {}
  52   ) {
  53     super(numberOfThreads, filePath, opts)
  54   }
  55
  56   /** @inheritDoc */
  57   protected isMain (): boolean {
  58     return isMainThread
  59   }
  60
  61   /** @inheritDoc */
  62   protected async destroyWorker (worker: Worker): Promise<void> {
  63     this.sendToWorker(worker, { kill: 1 })
  64     await worker.terminate()
  65   }
  66
  67   /** @inheritDoc */
  68   protected sendToWorker (worker: Worker, message: MessageValue<Data>): void {
  69     worker.postMessage(message)
  70   }
  71
  72   /** @inheritDoc */
  73   protected createWorker (): Worker {
  74     return new Worker(this.filePath, {
  75       env: SHARE_ENV,
  76       ...this.opts.workerOptions
  77     })
  78   }
  79
  80   /** @inheritDoc */
  81   protected get type (): PoolType {
  82     return PoolTypes.fixed
  83   }
  84
  85   /** @inheritDoc */
  86   protected get worker (): WorkerType {
  87     return WorkerTypes.thread
  88   }
  89
  90   /** @inheritDoc */
  91   protected get minSize (): number {
  92     return this.numberOfWorkers
  93   }
  94
  95   /** @inheritDoc */
  96   protected get maxSize (): number {
  97     return this.numberOfWorkers
  98   }
  99
 100   /** @inheritDoc */
 101   protected get busy (): boolean {
 102     return this.internalBusy()
 103   }
 104 }