fix: ensure worker node destroy semantic is always the same
[poolifier.git] / src / pools / thread / fixed.ts
CommitLineData
fc3e6586 1import {
85aeb3f3
JB
2 type MessageChannel,
3 type MessagePort,
fc3e6586 4 SHARE_ENV,
65d7a1c9 5 Worker,
90082c8c 6 type WorkerOptions,
65d7a1c9 7 isMainThread
fc3e6586 8} from 'node:worker_threads'
e102732c 9import type { MessageValue } from '../../utility-types'
c97c7edb 10import { AbstractPool } from '../abstract-pool'
4b628b48
JB
11import { type PoolOptions, type PoolType, PoolTypes } from '../pool'
12import { type WorkerType, WorkerTypes } from '../worker'
4ade5f1f 13
90082c8c
JB
14/**
15 * Options for a poolifier thread pool.
16 */
17export 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
4ade5f1f 26/**
729c563d
S
27 * A thread pool with a fixed number of threads.
28 *
e102732c
JB
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.
4ade5f1f
S
31 * @author [Alessandro Pio Ardizio](https://github.com/pioardi)
32 * @since 0.0.1
33 */
d3c8a1a8 34export class FixedThreadPool<
deb85c12
JB
35 Data = unknown,
36 Response = unknown
e102732c 37> extends AbstractPool<Worker, Data, Response> {
4ade5f1f 38 /**
729c563d
S
39 * Constructs a new poolifier fixed thread pool.
40 *
38e795c1
JB
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.
4ade5f1f
S
44 */
45 public constructor (
5c5a1fb7 46 numberOfThreads: number,
c97c7edb 47 filePath: string,
90082c8c 48 protected readonly opts: ThreadPoolOptions = {}
4ade5f1f 49 ) {
5c5a1fb7 50 super(numberOfThreads, filePath, opts)
c97c7edb 51 }
4ade5f1f 52
afc003b2 53 /** @inheritDoc */
c97c7edb
S
54 protected isMain (): boolean {
55 return isMainThread
4ade5f1f
S
56 }
57
afc003b2 58 /** @inheritDoc */
aa9eede8 59 protected async destroyWorkerNode (workerNodeKey: number): Promise<void> {
81c02522
JB
60 this.flushTasksQueue(workerNodeKey)
61 // FIXME: wait for tasks to be finished
aa9eede8
JB
62 const workerNode = this.workerNodes[workerNodeKey]
63 const worker = workerNode.worker
81c02522
JB
64 const workerExitPromise = new Promise<void>(resolve => {
65 worker.on('exit', () => {
66 resolve()
67 })
68 })
aa9eede8
JB
69 this.sendToWorker(workerNodeKey, { kill: true, workerId: worker.threadId })
70 workerNode.closeChannel()
c97c7edb 71 await worker.terminate()
81c02522 72 await workerExitPromise
4ade5f1f
S
73 }
74
afc003b2 75 /** @inheritDoc */
aa9eede8
JB
76 protected sendToWorker (
77 workerNodeKey: number,
78 message: MessageValue<Data>
79 ): void {
85aeb3f3 80 (
aa9eede8 81 this.getWorkerInfo(workerNodeKey).messageChannel as MessageChannel
85aeb3f3
JB
82 ).port1.postMessage(message)
83 }
84
85 /** @inheritDoc */
aa9eede8
JB
86 protected sendStartupMessageToWorker (workerNodeKey: number): void {
87 const worker = this.workerNodes[workerNodeKey].worker
85aeb3f3 88 const port2: MessagePort = (
aa9eede8 89 this.getWorkerInfo(workerNodeKey).messageChannel as MessageChannel
85aeb3f3
JB
90 ).port2
91 worker.postMessage(
92 {
93 ready: false,
a038b517 94 workerId: worker.threadId,
85aeb3f3
JB
95 port: port2
96 },
97 [port2]
98 )
99 }
100
101 /** @inheritDoc */
102 protected registerWorkerMessageListener<Message extends Data | Response>(
aa9eede8 103 workerNodeKey: number,
85aeb3f3
JB
104 listener: (message: MessageValue<Message>) => void
105 ): void {
106 (
aa9eede8 107 this.getWorkerInfo(workerNodeKey).messageChannel as MessageChannel
85aeb3f3 108 ).port1.on('message', listener)
4ade5f1f
S
109 }
110
afc003b2 111 /** @inheritDoc */
e102732c 112 protected createWorker (): Worker {
c97c7edb 113 return new Worker(this.filePath, {
90082c8c
JB
114 env: SHARE_ENV,
115 ...this.opts.workerOptions
4ade5f1f 116 })
c97c7edb
S
117 }
118
afc003b2 119 /** @inheritDoc */
8881ae32 120 protected get type (): PoolType {
6b27d407 121 return PoolTypes.fixed
7c0ba920
JB
122 }
123
184855e6
JB
124 /** @inheritDoc */
125 protected get worker (): WorkerType {
126 return WorkerTypes.thread
127 }
128
08f3f44c 129 /** @inheritDoc */
6b27d407
JB
130 protected get minSize (): number {
131 return this.numberOfWorkers
132 }
133
134 /** @inheritDoc */
135 protected get maxSize (): number {
08f3f44c
JB
136 return this.numberOfWorkers
137 }
138
afc003b2 139 /** @inheritDoc */
c319c66b 140 protected get busy (): boolean {
c2ade475 141 return this.internalBusy()
7c0ba920 142 }
4ade5f1f 143}