test: improve event emitter tests
[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.flushTasksQueue(workerNodeKey)
62 // FIXME: wait for tasks to be finished
63 const workerNode = this.workerNodes[workerNodeKey]
64 const worker = workerNode.worker
65 const waitWorkerExit = new Promise<void>(resolve => {
66 worker.once('exit', () => {
67 resolve()
68 })
69 })
70 await this.sendKillMessageToWorker(workerNodeKey)
71 workerNode.closeChannel()
72 await worker.terminate()
73 await waitWorkerExit
74 }
75
76 /** @inheritDoc */
77 protected sendToWorker (
78 workerNodeKey: number,
79 message: MessageValue<Data>,
80 transferList?: TransferListItem[]
81 ): void {
82 (
83 this.workerNodes[workerNodeKey].messageChannel as MessageChannel
84 ).port1.postMessage(
85 { ...message, workerId: this.workerNodes[workerNodeKey].info.id },
86 transferList
87 )
88 }
89
90 /** @inheritDoc */
91 protected sendStartupMessageToWorker (workerNodeKey: number): void {
92 const workerNode = this.workerNodes[workerNodeKey]
93 const port2: MessagePort = (workerNode.messageChannel as MessageChannel)
94 .port2
95 workerNode.worker.postMessage(
96 {
97 ready: false,
98 workerId: workerNode.info.id,
99 port: port2
100 },
101 [port2]
102 )
103 }
104
105 /** @inheritDoc */
106 protected registerWorkerMessageListener<Message extends Data | Response>(
107 workerNodeKey: number,
108 listener: (message: MessageValue<Message>) => void
109 ): void {
110 (
111 this.workerNodes[workerNodeKey].messageChannel as MessageChannel
112 ).port1.on('message', listener)
113 }
114
115 /** @inheritDoc */
116 protected registerOnceWorkerMessageListener<Message extends Data | Response>(
117 workerNodeKey: number,
118 listener: (message: MessageValue<Message>) => void
119 ): void {
120 (
121 this.workerNodes[workerNodeKey].messageChannel as MessageChannel
122 ).port1.once('message', listener)
123 }
124
125 /** @inheritDoc */
126 protected deregisterWorkerMessageListener<Message extends Data | Response>(
127 workerNodeKey: number,
128 listener: (message: MessageValue<Message>) => void
129 ): void {
130 (
131 this.workerNodes[workerNodeKey].messageChannel as MessageChannel
132 ).port1.off('message', listener)
133 }
134
135 /** @inheritDoc */
136 protected createWorker (): Worker {
137 return new Worker(this.filePath, {
138 env: SHARE_ENV,
139 ...this.opts.workerOptions
140 })
141 }
142
143 /** @inheritDoc */
144 protected get type (): PoolType {
145 return PoolTypes.fixed
146 }
147
148 /** @inheritDoc */
149 protected get worker (): WorkerType {
150 return WorkerTypes.thread
151 }
152
153 /** @inheritDoc */
154 protected get busy (): boolean {
155 return this.internalBusy()
156 }
157 }