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   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   protected sendToWorker (
  57     worker: ThreadWorkerWithMessageChannel,
  58     message: MessageValue<Data>
  59   ): void {
  60     worker.postMessage(message)
  61   }
  62
  63   /** @inheritdoc */
  64   public registerWorkerMessageListener<Message extends Data | Response> (
  65     messageChannel: ThreadWorkerWithMessageChannel,
  66     listener: (message: MessageValue<Message>) => void
  67   ): void {
  68     messageChannel.port2?.on('message', listener)
  69   }
  70
  71   protected createWorker (): ThreadWorkerWithMessageChannel {
  72     return new Worker(this.filePath, {
  73       env: SHARE_ENV
  74     })
  75   }
  76
  77   protected afterWorkerSetup (worker: ThreadWorkerWithMessageChannel): void {
  78     const { port1, port2 } = new MessageChannel()
  79     worker.postMessage({ parent: port1 }, [port1])
  80     worker.port1 = port1
  81     worker.port2 = port2
  82     // Listen worker messages.
  83     this.registerWorkerMessageListener(worker, super.workerListener())
  84   }
  85
  86   /** @inheritdoc */
  87   public get type (): PoolType {
  88     return PoolType.FIXED
  89   }
  90
  91   /** @inheritdoc */
  92   public get busy (): boolean {
  93     return this.internalGetBusyStatus()
  94   }
  95 }