Merge pull request #879 from poolifier/dependabot/npm_and_yarn/examples/typescript...
[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
8ebe6c30 65 const waitWorkerExit = new Promise<void>((resolve) => {
81c02522
JB
66 worker.on('exit', () => {
67 resolve()
68 })
69 })
1e3214b6 70 await this.sendKillMessageToWorker(workerNodeKey, worker.threadId)
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 {
85aeb3f3 82 (
7884d183 83 this.workerNodes[workerNodeKey].messageChannel as MessageChannel
7d91a8cd 84 ).port1.postMessage(message, transferList)
85aeb3f3
JB
85 }
86
87 /** @inheritDoc */
aa9eede8
JB
88 protected sendStartupMessageToWorker (workerNodeKey: number): void {
89 const worker = this.workerNodes[workerNodeKey].worker
85aeb3f3 90 const port2: MessagePort = (
7884d183 91 this.workerNodes[workerNodeKey].messageChannel as MessageChannel
85aeb3f3
JB
92 ).port2
93 worker.postMessage(
94 {
95 ready: false,
a038b517 96 workerId: worker.threadId,
85aeb3f3
JB
97 port: port2
98 },
99 [port2]
100 )
101 }
102
103 /** @inheritDoc */
104 protected registerWorkerMessageListener<Message extends Data | Response>(
aa9eede8 105 workerNodeKey: number,
85aeb3f3
JB
106 listener: (message: MessageValue<Message>) => void
107 ): void {
108 (
7884d183 109 this.workerNodes[workerNodeKey].messageChannel as MessageChannel
85aeb3f3 110 ).port1.on('message', listener)
4ade5f1f
S
111 }
112
afc003b2 113 /** @inheritDoc */
e102732c 114 protected createWorker (): Worker {
c97c7edb 115 return new Worker(this.filePath, {
90082c8c
JB
116 env: SHARE_ENV,
117 ...this.opts.workerOptions
4ade5f1f 118 })
c97c7edb
S
119 }
120
afc003b2 121 /** @inheritDoc */
8881ae32 122 protected get type (): PoolType {
6b27d407 123 return PoolTypes.fixed
7c0ba920
JB
124 }
125
184855e6
JB
126 /** @inheritDoc */
127 protected get worker (): WorkerType {
128 return WorkerTypes.thread
129 }
130
afc003b2 131 /** @inheritDoc */
c319c66b 132 protected get busy (): boolean {
c2ade475 133 return this.internalBusy()
7c0ba920 134 }
4ade5f1f 135}