5 } from
'node:worker_threads'
7 import type { MessageValue
} from
'../../utility-types.js'
8 import { AbstractPool
} from
'../abstract-pool.js'
9 import { type PoolOptions
, type PoolType
, PoolTypes
} from
'../pool.js'
10 import { type WorkerType
, WorkerTypes
} from
'../worker.js'
13 * Options for a poolifier thread pool.
15 export type ThreadPoolOptions
= PoolOptions
<Worker
>
18 * A thread pool with a fixed number of threads.
20 * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
21 * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
22 * @author [Alessandro Pio Ardizio](https://github.com/pioardi)
25 export class FixedThreadPool
<
28 > extends AbstractPool
<Worker
, Data
, Response
> {
30 * Constructs a new poolifier fixed thread pool.
32 * @param numberOfThreads - Number of threads for this pool.
33 * @param filePath - Path to an implementation of a `ThreadWorker` file, which can be relative or absolute.
34 * @param opts - Options for this fixed thread pool.
37 numberOfThreads
: number,
39 opts
: ThreadPoolOptions
= {},
40 maximumNumberOfThreads
?: number
42 super(numberOfThreads
, filePath
, opts
, maximumNumberOfThreads
)
46 protected isMain (): boolean {
51 protected sendToWorker (
52 workerNodeKey
: number,
53 message
: MessageValue
<Data
>,
54 transferList
?: TransferListItem
[]
56 this.workerNodes
[workerNodeKey
]?.messageChannel
?.port1
.postMessage(
59 workerId
: this.getWorkerInfo(workerNodeKey
)?.id
60 } satisfies MessageValue
<Data
>,
66 protected sendStartupMessageToWorker (workerNodeKey
: number): void {
67 const workerNode
= this.workerNodes
[workerNodeKey
]
68 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
69 const port2
= workerNode
.messageChannel
!.port2
70 workerNode
.worker
.postMessage(
73 workerId
: this.getWorkerInfo(workerNodeKey
)?.id
,
75 } satisfies MessageValue
<Data
>,
81 protected registerWorkerMessageListener
<Message
extends Data
| Response
>(
82 workerNodeKey
: number,
83 listener
: (message
: MessageValue
<Message
>) => void
85 this.workerNodes
[workerNodeKey
].messageChannel
?.port1
.on(
92 protected registerOnceWorkerMessageListener
<Message
extends Data
| Response
>(
93 workerNodeKey
: number,
94 listener
: (message
: MessageValue
<Message
>) => void
96 this.workerNodes
[workerNodeKey
].messageChannel
?.port1
.once(
103 protected deregisterWorkerMessageListener
<Message
extends Data
| Response
>(
104 workerNodeKey
: number,
105 listener
: (message
: MessageValue
<Message
>) => void
107 this.workerNodes
[workerNodeKey
].messageChannel
?.port1
.off(
114 protected shallCreateDynamicWorker (): boolean {
119 protected checkAndEmitDynamicWorkerCreationEvents (): void {
124 protected get
type (): PoolType
{
125 return PoolTypes
.fixed
129 protected get
worker (): WorkerType
{
130 return WorkerTypes
.thread
134 protected get
busy (): boolean {
135 return this.internalBusy()