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