src/pools/thread/fixed.ts

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