src/pools/thread/fixed.ts

   1 import { isMainThread, MessageChannel, SHARE_ENV, Worker } from 'worker_threads'
   2 import type { Draft, MessageValue } from '../../utility-types'
   3 import type { PoolOptions } from '../abstract-pool'
   4 import { AbstractPool } from '../abstract-pool'
   5 import { PoolType } from '../pool-internal'
   6
   7 /**
   8  * A thread worker with message channels for communication between main thread and thread worker.
   9  */
  10 export type ThreadWorkerWithMessageChannel = Worker & Draft<MessageChannel>
  11
  12 /**
  13  * A thread pool with a fixed number of threads.
  14  *
  15  * It is possible to perform tasks in sync or asynchronous mode as you prefer.
  16  *
  17  * This pool selects the threads in a round robin fashion.
  18  *
  19  * @template Data Type of data sent to the worker. This can only be serializable data.
  20  * @template Response Type of response of execution. This can only be serializable data.
  21  *
  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<ThreadWorkerWithMessageChannel, Data, Response> {
  29   /**
  30    * Constructs a new poolifier fixed thread pool.
  31    *
  32    * @param numberOfThreads Number of threads for this pool.
  33    * @param filePath Path to an implementation of a `ThreadWorker` file, which can be relative or absolute.
  34    * @param opts Options for this fixed thread pool. Default: `{}`
  35    */
  36   public constructor (
  37     numberOfThreads: number,
  38     filePath: string,
  39     opts: PoolOptions<ThreadWorkerWithMessageChannel> = {}
  40   ) {
  41     super(numberOfThreads, filePath, opts)
  42   }
  43
  44   /** @inheritdoc */
  45   protected isMain (): boolean {
  46     return isMainThread
  47   }
  48
  49   /** @inheritdoc */
  50   public async destroyWorker (
  51     worker: ThreadWorkerWithMessageChannel
  52   ): Promise<void> {
  53     this.sendToWorker(worker, { kill: 1 })
  54     await worker.terminate()
  55   }
  56
  57   /** @inheritdoc */
  58   protected sendToWorker (
  59     worker: ThreadWorkerWithMessageChannel,
  60     message: MessageValue<Data>
  61   ): void {
  62     worker.postMessage(message)
  63   }
  64
  65   /** @inheritdoc */
  66   public registerWorkerMessageListener<Message extends Data | Response> (
  67     messageChannel: ThreadWorkerWithMessageChannel,
  68     listener: (message: MessageValue<Message>) => void
  69   ): void {
  70     messageChannel.port2?.on('message', listener)
  71   }
  72
  73   /** @inheritdoc */
  74   protected createWorker (): ThreadWorkerWithMessageChannel {
  75     return new Worker(this.filePath, {
  76       env: SHARE_ENV
  77     })
  78   }
  79
  80   /** @inheritdoc */
  81   protected afterWorkerSetup (worker: ThreadWorkerWithMessageChannel): void {
  82     const { port1, port2 } = new MessageChannel()
  83     worker.postMessage({ parent: port1 }, [port1])
  84     worker.port1 = port1
  85     worker.port2 = port2
  86     // Listen worker messages.
  87     this.registerWorkerMessageListener(worker, super.workerListener())
  88   }
  89
  90   /** @inheritdoc */
  91   public get type (): PoolType {
  92     return PoolType.FIXED
  93   }
  94
  95   /** @inheritdoc */
  96   public get busy (): boolean {
  97     return this.internalGetBusyStatus()
  98   }
  99 }