src/pools/thread/fixed.ts

   1 import {
   2   type MessageChannel,
   3   type MessagePort,
   4   SHARE_ENV,
   5   Worker,
   6   type WorkerOptions,
   7   isMainThread
   8 } from 'node:worker_threads'
   9 import type { MessageValue } from '../../utility-types'
  10 import { AbstractPool } from '../abstract-pool'
  11 import { type PoolOptions, type PoolType, PoolTypes } from '../pool'
  12 import { type WorkerType, WorkerTypes } from '../worker'
  13
  14 /**
  15  * Options for a poolifier thread pool.
  16  */
  17 export interface ThreadPoolOptions extends PoolOptions<Worker> {
  18   /**
  19    * Worker options.
  20    *
  21    * @see https://nodejs.org/api/worker_threads.html#new-workerfilename-options
  22    */
  23   workerOptions?: WorkerOptions
  24 }
  25
  26 /**
  27  * A thread pool with a fixed number of threads.
  28  *
  29  * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
  30  * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
  31  * @author [Alessandro Pio Ardizio](https://github.com/pioardi)
  32  * @since 0.0.1
  33  */
  34 export class FixedThreadPool<
  35   Data = unknown,
  36   Response = unknown
  37 > extends AbstractPool<Worker, Data, Response> {
  38   /**
  39    * Constructs a new poolifier fixed thread pool.
  40    *
  41    * @param numberOfThreads - Number of threads for this pool.
  42    * @param filePath - Path to an implementation of a `ThreadWorker` file, which can be relative or absolute.
  43    * @param opts - Options for this fixed thread pool.
  44    */
  45   public constructor (
  46     numberOfThreads: number,
  47     filePath: string,
  48     protected readonly opts: ThreadPoolOptions = {}
  49   ) {
  50     super(numberOfThreads, filePath, opts)
  51   }
  52
  53   /** @inheritDoc */
  54   protected isMain (): boolean {
  55     return isMainThread
  56   }
  57
  58   /** @inheritDoc */
  59   protected async destroyWorkerNode (workerNodeKey: number): Promise<void> {
  60     this.flushTasksQueue(workerNodeKey)
  61     // FIXME: wait for tasks to be finished
  62     const workerNode = this.workerNodes[workerNodeKey]
  63     const worker = workerNode.worker
  64     const waitWorkerExit = new Promise<void>(resolve => {
  65       worker.on('exit', () => {
  66         resolve()
  67       })
  68     })
  69     this.sendToWorker(workerNodeKey, { kill: true, workerId: worker.threadId })
  70     workerNode.closeChannel()
  71     await worker.terminate()
  72     await waitWorkerExit
  73   }
  74
  75   /** @inheritDoc */
  76   protected sendToWorker (
  77     workerNodeKey: number,
  78     message: MessageValue<Data>
  79   ): void {
  80     (
  81       this.getWorkerInfo(workerNodeKey).messageChannel as MessageChannel
  82     ).port1.postMessage(message)
  83   }
  84
  85   /** @inheritDoc */
  86   protected sendStartupMessageToWorker (workerNodeKey: number): void {
  87     const worker = this.workerNodes[workerNodeKey].worker
  88     const port2: MessagePort = (
  89       this.getWorkerInfo(workerNodeKey).messageChannel as MessageChannel
  90     ).port2
  91     worker.postMessage(
  92       {
  93         ready: false,
  94         workerId: worker.threadId,
  95         port: port2
  96       },
  97       [port2]
  98     )
  99   }
 100
 101   /** @inheritDoc */
 102   protected registerWorkerMessageListener<Message extends Data | Response>(
 103     workerNodeKey: number,
 104     listener: (message: MessageValue<Message>) => void
 105   ): void {
 106     (
 107       this.getWorkerInfo(workerNodeKey).messageChannel as MessageChannel
 108     ).port1.on('message', listener)
 109   }
 110
 111   /** @inheritDoc */
 112   protected createWorker (): Worker {
 113     return new Worker(this.filePath, {
 114       env: SHARE_ENV,
 115       ...this.opts.workerOptions
 116     })
 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 minSize (): number {
 131     return this.numberOfWorkers
 132   }
 133
 134   /** @inheritDoc */
 135   protected get maxSize (): number {
 136     return this.numberOfWorkers
 137   }
 138
 139   /** @inheritDoc */
 140   protected get busy (): boolean {
 141     return this.internalBusy()
 142   }
 143 }