5 } from
'node:worker_threads'
7 import type { MessageValue
} from
'../../utility-types.js'
9 import { AbstractPool
} from
'../abstract-pool.js'
10 import { type PoolOptions
, type PoolType
, PoolTypes
} from
'../pool.js'
11 import { type WorkerType
, WorkerTypes
} from
'../worker.js'
14 * Options for a poolifier thread pool.
16 export type ThreadPoolOptions
= PoolOptions
<Worker
>
19 * 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.
31 * @param numberOfThreads - Number of threads for this pool.
32 * @param filePath - Path to an implementation of a `ThreadWorker` file, which can be relative or absolute.
33 * @param opts - Options for this fixed thread pool.
34 * @param maximumNumberOfThreads - The maximum number of threads for this pool.
37 numberOfThreads
: number,
39 opts
: ThreadPoolOptions
= {},
40 maximumNumberOfThreads
?: number
42 super(numberOfThreads
, filePath
, opts
, maximumNumberOfThreads
)
46 protected checkAndEmitDynamicWorkerCreationEvents (): void {
51 protected checkAndEmitDynamicWorkerDestructionEvents (): void {
56 protected deregisterWorkerMessageListener
<Message
extends Data
| Response
>(
57 workerNodeKey
: number,
58 listener
: (message
: MessageValue
<Message
>) => void
60 this.workerNodes
[workerNodeKey
].messageChannel
?.port1
.off(
67 protected isMain (): boolean {
72 protected registerOnceWorkerMessageListener
<Message
extends Data
| Response
>(
73 workerNodeKey
: number,
74 listener
: (message
: MessageValue
<Message
>) => void
76 this.workerNodes
[workerNodeKey
].messageChannel
?.port1
.once(
83 protected registerWorkerMessageListener
<Message
extends Data
| Response
>(
84 workerNodeKey
: number,
85 listener
: (message
: MessageValue
<Message
>) => void
87 this.workerNodes
[workerNodeKey
].messageChannel
?.port1
.on(
94 protected sendStartupMessageToWorker (workerNodeKey
: number): void {
95 const workerNode
= this.workerNodes
[workerNodeKey
]
96 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
97 const port2
= workerNode
.messageChannel
!.port2
98 workerNode
.worker
.postMessage(
102 workerId
: this.getWorkerInfo(workerNodeKey
)?.id
,
103 } satisfies MessageValue
<Data
>,
109 protected sendToWorker (
110 workerNodeKey
: number,
111 message
: MessageValue
<Data
>,
112 transferList
?: readonly TransferListItem
[]
114 this.workerNodes
[workerNodeKey
]?.messageChannel
?.port1
.postMessage(
117 workerId
: this.getWorkerInfo(workerNodeKey
)?.id
,
118 } satisfies MessageValue
<Data
>,
124 protected shallCreateDynamicWorker (): boolean {
129 protected get
backPressure (): boolean {
130 return this.internalBackPressure()
134 protected get
busy (): boolean {
135 return this.internalBusy()
139 protected get
type (): PoolType
{
140 return PoolTypes
.fixed
144 protected get
worker (): WorkerType
{
145 return WorkerTypes
.thread