test: improve event emitter tests
[poolifier.git] / src / pools / thread / fixed.ts
CommitLineData
fc3e6586 1import {
85aeb3f3
JB
2 type MessageChannel,
3 type MessagePort,
fc3e6586 4 SHARE_ENV,
7d91a8cd 5 type TransferListItem,
65d7a1c9 6 Worker,
90082c8c 7 type WorkerOptions,
65d7a1c9 8 isMainThread
fc3e6586 9} from 'node:worker_threads'
e102732c 10import type { MessageValue } from '../../utility-types'
c97c7edb 11import { AbstractPool } from '../abstract-pool'
4b628b48
JB
12import { type PoolOptions, type PoolType, PoolTypes } from '../pool'
13import { type WorkerType, WorkerTypes } from '../worker'
4ade5f1f 14
90082c8c
JB
15/**
16 * Options for a poolifier thread pool.
17 */
18export 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
4ade5f1f 27/**
729c563d
S
28 * A thread pool with a fixed number of threads.
29 *
e102732c
JB
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.
4ade5f1f
S
32 * @author [Alessandro Pio Ardizio](https://github.com/pioardi)
33 * @since 0.0.1
34 */
d3c8a1a8 35export class FixedThreadPool<
deb85c12
JB
36 Data = unknown,
37 Response = unknown
e102732c 38> extends AbstractPool<Worker, Data, Response> {
4ade5f1f 39 /**
729c563d
S
40 * Constructs a new poolifier fixed thread pool.
41 *
38e795c1
JB
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.
4ade5f1f
S
45 */
46 public constructor (
5c5a1fb7 47 numberOfThreads: number,
c97c7edb 48 filePath: string,
90082c8c 49 protected readonly opts: ThreadPoolOptions = {}
4ade5f1f 50 ) {
5c5a1fb7 51 super(numberOfThreads, filePath, opts)
c97c7edb 52 }
4ade5f1f 53
afc003b2 54 /** @inheritDoc */
c97c7edb
S
55 protected isMain (): boolean {
56 return isMainThread
4ade5f1f
S
57 }
58
afc003b2 59 /** @inheritDoc */
aa9eede8 60 protected async destroyWorkerNode (workerNodeKey: number): Promise<void> {
81c02522
JB
61 this.flushTasksQueue(workerNodeKey)
62 // FIXME: wait for tasks to be finished
aa9eede8
JB
63 const workerNode = this.workerNodes[workerNodeKey]
64 const worker = workerNode.worker
041dc05b 65 const waitWorkerExit = new Promise<void>(resolve => {
ae036c3e 66 worker.once('exit', () => {
81c02522
JB
67 resolve()
68 })
69 })
72ae84a2 70 await this.sendKillMessageToWorker(workerNodeKey)
aa9eede8 71 workerNode.closeChannel()
c97c7edb 72 await worker.terminate()
c2301b8e 73 await waitWorkerExit
4ade5f1f
S
74 }
75
afc003b2 76 /** @inheritDoc */
aa9eede8
JB
77 protected sendToWorker (
78 workerNodeKey: number,
7d91a8cd
JB
79 message: MessageValue<Data>,
80 transferList?: TransferListItem[]
aa9eede8 81 ): void {
041dc05b 82 (
7884d183 83 this.workerNodes[workerNodeKey].messageChannel as MessageChannel
72ae84a2
JB
84 ).port1.postMessage(
85 { ...message, workerId: this.workerNodes[workerNodeKey].info.id },
86 transferList
87 )
85aeb3f3
JB
88 }
89
90 /** @inheritDoc */
aa9eede8 91 protected sendStartupMessageToWorker (workerNodeKey: number): void {
75de9f41 92 const workerNode = this.workerNodes[workerNodeKey]
75de9f41
JB
93 const port2: MessagePort = (workerNode.messageChannel as MessageChannel)
94 .port2
e9dd5b66 95 workerNode.worker.postMessage(
85aeb3f3
JB
96 {
97 ready: false,
75de9f41 98 workerId: workerNode.info.id,
85aeb3f3
JB
99 port: port2
100 },
101 [port2]
102 )
103 }
104
105 /** @inheritDoc */
106 protected registerWorkerMessageListener<Message extends Data | Response>(
aa9eede8 107 workerNodeKey: number,
85aeb3f3
JB
108 listener: (message: MessageValue<Message>) => void
109 ): void {
041dc05b 110 (
7884d183 111 this.workerNodes[workerNodeKey].messageChannel as MessageChannel
85aeb3f3 112 ).port1.on('message', listener)
4ade5f1f
S
113 }
114
ae036c3e
JB
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
afc003b2 135 /** @inheritDoc */
e102732c 136 protected createWorker (): Worker {
c97c7edb 137 return new Worker(this.filePath, {
90082c8c
JB
138 env: SHARE_ENV,
139 ...this.opts.workerOptions
4ade5f1f 140 })
c97c7edb
S
141 }
142
afc003b2 143 /** @inheritDoc */
8881ae32 144 protected get type (): PoolType {
6b27d407 145 return PoolTypes.fixed
7c0ba920
JB
146 }
147
184855e6
JB
148 /** @inheritDoc */
149 protected get worker (): WorkerType {
150 return WorkerTypes.thread
151 }
152
afc003b2 153 /** @inheritDoc */
c319c66b 154 protected get busy (): boolean {
c2ade475 155 return this.internalBusy()
7c0ba920 156 }
4ade5f1f 157}