src/pools/thread/fixed.ts

   1 import {
   2   isMainThread,
   3   type Transferable,
   4   type Worker,
   5 } from 'node:worker_threads'
   6
   7 import type { MessageValue } from '../../utility-types.js'
   8
   9 import { AbstractPool } from '../abstract-pool.js'
  10 import { type PoolOptions, type PoolType, PoolTypes } from '../pool.js'
  11 import { type WorkerType, WorkerTypes } from '../worker.js'
  12
  13 /**
  14  * Options for a poolifier thread pool.
  15  */
  16 export type ThreadPoolOptions = PoolOptions<Worker>
  17
  18 /**
  19  * A thread pool with a fixed number of threads.
  20  * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
  21  * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
  22  * @author [Alessandro Pio Ardizio](https://github.com/pioardi)
  23  * @since 0.0.1
  24  */
  25 export class FixedThreadPool<
  26   Data = unknown,
  27   Response = unknown
  28 > extends AbstractPool<Worker, Data, Response> {
  29   /** @inheritDoc */
  30   protected get backPressure (): boolean {
  31     return this.internalBackPressure()
  32   }
  33
  34   /** @inheritDoc */
  35   protected get busy (): boolean {
  36     return this.internalBusy()
  37   }
  38
  39   /** @inheritDoc */
  40   protected get type (): PoolType {
  41     return PoolTypes.fixed
  42   }
  43
  44   /** @inheritDoc */
  45   protected get worker (): WorkerType {
  46     return WorkerTypes.thread
  47   }
  48
  49   /**
  50    * Constructs a new poolifier fixed thread pool.
  51    * @param numberOfThreads - Number of threads for this pool.
  52    * @param filePath - Path to an implementation of a `ThreadWorker` file, which can be relative or absolute.
  53    * @param opts - Options for this fixed thread pool.
  54    * @param maximumNumberOfThreads - The maximum number of threads for this pool.
  55    */
  56   public constructor (
  57     numberOfThreads: number,
  58     filePath: string,
  59     opts: ThreadPoolOptions = {},
  60     maximumNumberOfThreads?: number
  61   ) {
  62     super(numberOfThreads, filePath, opts, maximumNumberOfThreads)
  63   }
  64
  65   /** @inheritDoc */
  66   protected checkAndEmitDynamicWorkerCreationEvents (): void {
  67     /* noop */
  68   }
  69
  70   /** @inheritDoc */
  71   protected checkAndEmitDynamicWorkerDestructionEvents (): void {
  72     /* noop */
  73   }
  74
  75   /** @inheritDoc */
  76   protected deregisterWorkerMessageListener<Message extends Data | Response>(
  77     workerNodeKey: number,
  78     listener: (message: MessageValue<Message>) => void
  79   ): void {
  80     this.workerNodes[workerNodeKey].messageChannel?.port1.off(
  81       'message',
  82       listener
  83     )
  84   }
  85
  86   /** @inheritDoc */
  87   protected isMain (): boolean {
  88     return isMainThread
  89   }
  90
  91   /** @inheritDoc */
  92   protected registerOnceWorkerMessageListener<Message extends Data | Response>(
  93     workerNodeKey: number,
  94     listener: (message: MessageValue<Message>) => void
  95   ): void {
  96     this.workerNodes[workerNodeKey].messageChannel?.port1.once(
  97       'message',
  98       listener
  99     )
 100   }
 101
 102   /** @inheritDoc */
 103   protected registerWorkerMessageListener<Message extends Data | Response>(
 104     workerNodeKey: number,
 105     listener: (message: MessageValue<Message>) => void
 106   ): void {
 107     this.workerNodes[workerNodeKey].messageChannel?.port1.on(
 108       'message',
 109       listener
 110     )
 111   }
 112
 113   /** @inheritDoc */
 114   protected sendStartupMessageToWorker (workerNodeKey: number): void {
 115     const workerNode = this.workerNodes[workerNodeKey]
 116     // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
 117     const port2 = workerNode.messageChannel!.port2
 118     workerNode.worker.postMessage(
 119       {
 120         port: port2,
 121         ready: false,
 122         workerId: this.getWorkerInfo(workerNodeKey)?.id,
 123       } satisfies MessageValue<Data>,
 124       [port2]
 125     )
 126   }
 127
 128   /** @inheritDoc */
 129   protected sendToWorker (
 130     workerNodeKey: number,
 131     message: MessageValue<Data>,
 132     transferList?: readonly Transferable[]
 133   ): void {
 134     this.workerNodes[workerNodeKey]?.messageChannel?.port1.postMessage(
 135       {
 136         ...message,
 137         workerId: this.getWorkerInfo(workerNodeKey)?.id,
 138       } satisfies MessageValue<Data>,
 139       transferList
 140     )
 141   }
 142
 143   /** @inheritDoc */
 144   protected shallCreateDynamicWorker (): boolean {
 145     return false
 146   }
 147 }