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