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 DataType of data sent to the worker. This can only be serializable data.
  20  * @template ResponseType of response of execution. This can only be serializable 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<ThreadWorkerWithMessageChannel, 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: PoolOptions<ThreadWorkerWithMessageChannel> = {}
  39   ) {
  40     super(numberOfThreads, filePath, opts)
  41   }
  42
  43   /** @inheritdoc */
  44   protected isMain (): boolean {
  45     return isMainThread
  46   }
  47
  48   /** @inheritdoc */
  49   public async destroyWorker (
  50     worker: ThreadWorkerWithMessageChannel
  51   ): Promise<void> {
  52     this.sendToWorker(worker, { kill: 1 })
  53     await worker.terminate()
  54   }
  55
  56   /** @inheritdoc */
  57   protected sendToWorker (
  58     worker: ThreadWorkerWithMessageChannel,
  59     message: MessageValue<Data>
  60   ): void {
  61     worker.postMessage(message)
  62   }
  63
  64   /** @inheritdoc */
  65   public registerWorkerMessageListener<Message extends Data | Response> (
  66     messageChannel: ThreadWorkerWithMessageChannel,
  67     listener: (message: MessageValue<Message>) => void
  68   ): void {
  69     messageChannel.port2?.on('message', listener)
  70   }
  71
  72   /** @inheritdoc */
  73   protected createWorker (): ThreadWorkerWithMessageChannel {
  74     return new Worker(this.filePath, {
  75       env: SHARE_ENV
  76     })
  77   }
  78
  79   /** @inheritdoc */
  80   protected afterWorkerSetup (worker: ThreadWorkerWithMessageChannel): void {
  81     const { port1, port2 } = new MessageChannel()
  82     worker.postMessage({ parent: port1 }, [port1])
  83     worker.port1 = port1
  84     worker.port2 = port2
  85     // Listen worker messages.
  86     this.registerWorkerMessageListener(worker, super.workerListener())
  87   }
  88
  89   /** @inheritdoc */
  90   public get type (): PoolType {
  91     return PoolType.FIXED
  92   }
  93
  94   /** @inheritdoc */
  95   public get busy (): boolean {
  96     return this.internalGetBusyStatus()
  97   }
  98 }