1 import crypto from
'node:crypto'
2 import type { MessageValue
, PromiseResponseWrapper
} from
'../utility-types'
3 import { EMPTY_FUNCTION
} from
'../utils'
4 import { KillBehaviors
, isKillBehavior
} from
'../worker/worker-options'
5 import { PoolEvents
, type PoolOptions
} from
'./pool'
6 import { PoolEmitter
} from
'./pool'
7 import type { IPoolInternal
, TasksUsage
, WorkerType
} from
'./pool-internal'
8 import { PoolType
} from
'./pool-internal'
9 import type { IPoolWorker
} from
'./pool-worker'
11 WorkerChoiceStrategies
,
12 type WorkerChoiceStrategy
13 } from
'./selection-strategies/selection-strategies-types'
14 import { WorkerChoiceStrategyContext
} from
'./selection-strategies/worker-choice-strategy-context'
17 * Base class that implements some shared logic for all poolifier pools.
19 * @typeParam Worker - Type of worker which manages this pool.
20 * @typeParam Data - Type of data sent to the worker. This can only be serializable data.
21 * @typeParam Response - Type of response of execution. This can only be serializable data.
23 export abstract class AbstractPool
<
24 Worker
extends IPoolWorker
,
27 > implements IPoolInternal
<Worker
, Data
, Response
> {
29 public readonly workers
: Array<WorkerType
<Worker
>> = []
32 public readonly emitter
?: PoolEmitter
35 * The promise response map.
37 * - `key`: The message id of each submitted task.
38 * - `value`: An object that contains the worker, the promise resolve and reject callbacks.
40 * When we receive a message from the worker we get a map entry with the promise resolve/reject bound to the message.
42 protected promiseResponseMap
: Map
<
44 PromiseResponseWrapper
<Worker
, Response
>
45 > = new Map
<string, PromiseResponseWrapper
<Worker
, Response
>>()
48 * Worker choice strategy context referencing a worker choice algorithm implementation.
50 * Default to a round robin algorithm.
52 protected workerChoiceStrategyContext
: WorkerChoiceStrategyContext
<
59 * Constructs a new poolifier pool.
61 * @param numberOfWorkers - Number of workers that this pool should manage.
62 * @param filePath - Path to the worker-file.
63 * @param opts - Options for the pool.
66 public readonly numberOfWorkers
: number,
67 public readonly filePath
: string,
68 public readonly opts
: PoolOptions
<Worker
>
71 throw new Error('Cannot start a pool from a worker!')
73 this.checkNumberOfWorkers(this.numberOfWorkers
)
74 this.checkFilePath(this.filePath
)
75 this.checkPoolOptions(this.opts
)
77 this.chooseWorker
.bind(this)
78 this.internalExecute
.bind(this)
79 this.checkAndEmitFull
.bind(this)
80 this.checkAndEmitBusy
.bind(this)
81 this.sendToWorker
.bind(this)
85 for (let i
= 1; i
<= this.numberOfWorkers
; i
++) {
86 this.createAndSetupWorker()
89 if (this.opts
.enableEvents
=== true) {
90 this.emitter
= new PoolEmitter()
92 this.workerChoiceStrategyContext
= new WorkerChoiceStrategyContext
<
96 >(this, this.opts
.workerChoiceStrategy
)
99 private checkFilePath (filePath
: string): void {
102 (typeof filePath
=== 'string' && filePath
.trim().length
=== 0)
104 throw new Error('Please specify a file with a worker implementation')
108 private checkNumberOfWorkers (numberOfWorkers
: number): void {
109 if (numberOfWorkers
== null) {
111 'Cannot instantiate a pool without specifying the number of workers'
113 } else if (!Number.isSafeInteger(numberOfWorkers
)) {
115 'Cannot instantiate a pool with a non integer number of workers'
117 } else if (numberOfWorkers
< 0) {
118 throw new RangeError(
119 'Cannot instantiate a pool with a negative number of workers'
121 } else if (this.type === PoolType
.FIXED
&& numberOfWorkers
=== 0) {
122 throw new Error('Cannot instantiate a fixed pool with no worker')
126 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
127 this.opts
.workerChoiceStrategy
=
128 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
129 this.checkValidWorkerChoiceStrategy(this.opts
.workerChoiceStrategy
)
130 this.opts
.enableEvents
= opts
.enableEvents
?? true
133 private checkValidWorkerChoiceStrategy (
134 workerChoiceStrategy
: WorkerChoiceStrategy
136 if (!Object.values(WorkerChoiceStrategies
).includes(workerChoiceStrategy
)) {
138 `Invalid worker choice strategy '${workerChoiceStrategy}'`
144 public abstract get
type (): PoolType
147 * Number of tasks concurrently running in the pool.
149 private get
numberOfRunningTasks (): number {
150 return this.promiseResponseMap
.size
154 * Gets the given worker key.
156 * @param worker - The worker.
157 * @returns The worker key if the worker is found in the pool, `-1` otherwise.
159 private getWorkerKey (worker
: Worker
): number {
160 return this.workers
.findIndex(workerItem
=> workerItem
.worker
=== worker
)
164 public setWorkerChoiceStrategy (
165 workerChoiceStrategy
: WorkerChoiceStrategy
167 this.checkValidWorkerChoiceStrategy(workerChoiceStrategy
)
168 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
169 for (const [index
, workerItem
] of this.workers
.entries()) {
170 this.setWorker(index
, workerItem
.worker
, {
178 this.workerChoiceStrategyContext
.setWorkerChoiceStrategy(
184 public abstract get
full (): boolean
187 public abstract get
busy (): boolean
189 protected internalBusy (): boolean {
191 this.numberOfRunningTasks
>= this.numberOfWorkers
&&
192 this.findFreeWorkerKey() === -1
197 public findFreeWorkerKey (): number {
198 return this.workers
.findIndex(workerItem
=> {
199 return workerItem
.tasksUsage
.running
=== 0
204 public async execute (data
: Data
): Promise
<Response
> {
205 const [workerKey
, worker
] = this.chooseWorker()
206 const messageId
= crypto
.randomUUID()
207 const res
= this.internalExecute(workerKey
, worker
, messageId
)
208 this.checkAndEmitFull()
209 this.checkAndEmitBusy()
210 this.sendToWorker(worker
, {
211 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
212 data
: data
?? ({} as Data
),
215 // eslint-disable-next-line @typescript-eslint/return-await
220 public async destroy (): Promise
<void> {
222 this.workers
.map(async workerItem
=> {
223 await this.destroyWorker(workerItem
.worker
)
229 * Shutdowns given worker in the pool.
231 * @param worker - A worker within `workers`.
233 protected abstract destroyWorker (worker
: Worker
): void | Promise
<void>
236 * Setup hook that can be overridden by a Poolifier pool implementation
237 * to run code before workers are created in the abstract constructor.
242 protected setupHook (): void {
243 // Intentionally empty
247 * Should return whether the worker is the main worker or not.
249 protected abstract isMain (): boolean
252 * Hook executed before the worker task promise resolution.
255 * @param workerKey - The worker key.
257 protected beforePromiseResponseHook (workerKey
: number): void {
258 ++this.workers
[workerKey
].tasksUsage
.running
262 * Hook executed after the worker task promise resolution.
265 * @param worker - The worker.
266 * @param message - The received message.
268 protected afterPromiseResponseHook (
270 message
: MessageValue
<Response
>
272 const workerTasksUsage
= this.getWorkerTasksUsage(worker
) as TasksUsage
273 --workerTasksUsage
.running
274 ++workerTasksUsage
.run
275 if (message
.error
!= null) {
276 ++workerTasksUsage
.error
278 if (this.workerChoiceStrategyContext
.getRequiredStatistics().runTime
) {
279 workerTasksUsage
.runTime
+= message
.runTime
?? 0
281 this.workerChoiceStrategyContext
.getRequiredStatistics().avgRunTime
&&
282 workerTasksUsage
.run
!== 0
284 workerTasksUsage
.avgRunTime
=
285 workerTasksUsage
.runTime
/ workerTasksUsage
.run
291 * Chooses a worker for the next task.
293 * The default uses a round robin algorithm to distribute the load.
295 * @returns [worker key, worker].
297 protected chooseWorker (): [number, Worker
] {
298 let workerKey
: number
300 this.type === PoolType
.DYNAMIC
&&
302 this.findFreeWorkerKey() === -1
304 const createdWorker
= this.createAndSetupWorker()
305 this.registerWorkerMessageListener(createdWorker
, message
=> {
307 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
308 (message
.kill
!= null &&
309 this.getWorkerTasksUsage(createdWorker
)?.running
=== 0)
311 // Kill message received from the worker, means that no new tasks are submitted to that worker for a while ( > maxInactiveTime)
312 void this.destroyWorker(createdWorker
)
315 workerKey
= this.getWorkerKey(createdWorker
)
317 workerKey
= this.workerChoiceStrategyContext
.execute()
319 return [workerKey
, this.workers
[workerKey
].worker
]
323 * Sends a message to the given worker.
325 * @param worker - The worker which should receive the message.
326 * @param message - The message.
328 protected abstract sendToWorker (
330 message
: MessageValue
<Data
>
334 * Registers a listener callback on a given worker.
336 * @param worker - The worker which should register a listener.
337 * @param listener - The message listener callback.
339 protected abstract registerWorkerMessageListener
<
340 Message
extends Data
| Response
341 >(worker
: Worker
, listener
: (message
: MessageValue
<Message
>) => void): void
344 * Returns a newly created worker.
346 protected abstract createWorker (): Worker
349 * Function that can be hooked up when a worker has been newly created and moved to the workers registry.
351 * Can be used to update the `maxListeners` or binding the `main-worker`\<-\>`worker` connection if not bind by default.
353 * @param worker - The newly created worker.
356 protected abstract afterWorkerSetup (worker
: Worker
): void
359 * Creates a new worker for this pool and sets it up completely.
361 * @returns New, completely set up worker.
363 protected createAndSetupWorker (): Worker
{
364 const worker
= this.createWorker()
366 worker
.on('message', this.opts
.messageHandler
?? EMPTY_FUNCTION
)
367 worker
.on('error', this.opts
.errorHandler
?? EMPTY_FUNCTION
)
368 worker
.on('online', this.opts
.onlineHandler
?? EMPTY_FUNCTION
)
369 worker
.on('exit', this.opts
.exitHandler
?? EMPTY_FUNCTION
)
370 worker
.once('exit', () => {
371 this.removeWorker(worker
)
374 this.pushWorker(worker
, {
382 this.afterWorkerSetup(worker
)
388 * This function is the listener registered for each worker.
390 * @returns The listener function to execute when a message is received from a worker.
392 protected workerListener (): (message
: MessageValue
<Response
>) => void {
394 if (message
.id
!= null) {
395 // Task response received
396 const promiseResponse
= this.promiseResponseMap
.get(message
.id
)
397 if (promiseResponse
!= null) {
398 if (message
.error
!= null) {
399 promiseResponse
.reject(message
.error
)
401 promiseResponse
.resolve(message
.data
as Response
)
403 this.afterPromiseResponseHook(promiseResponse
.worker
, message
)
404 this.promiseResponseMap
.delete(message
.id
)
410 private async internalExecute (
414 ): Promise
<Response
> {
415 this.beforePromiseResponseHook(workerKey
)
416 return await new Promise
<Response
>((resolve
, reject
) => {
417 this.promiseResponseMap
.set(messageId
, { resolve
, reject
, worker
})
421 private checkAndEmitBusy (): void {
422 if (this.opts
.enableEvents
=== true && this.busy
) {
423 this.emitter
?.emit(PoolEvents
.busy
)
427 private checkAndEmitFull (): void {
429 this.type === PoolType
.DYNAMIC
&&
430 this.opts
.enableEvents
=== true &&
433 this.emitter
?.emit(PoolEvents
.full
)
438 * Gets the given worker tasks usage in the pool.
440 * @param worker - The worker.
441 * @returns The worker tasks usage.
443 private getWorkerTasksUsage (worker
: Worker
): TasksUsage
| undefined {
444 const workerKey
= this.getWorkerKey(worker
)
445 if (workerKey
!== -1) {
446 return this.workers
[workerKey
].tasksUsage
448 throw new Error('Worker could not be found in the pool')
452 * Pushes the given worker in the pool.
454 * @param worker - The worker.
455 * @param tasksUsage - The worker tasks usage.
457 private pushWorker (worker
: Worker
, tasksUsage
: TasksUsage
): void {
465 * Sets the given worker in the pool.
467 * @param workerKey - The worker key.
468 * @param worker - The worker.
469 * @param tasksUsage - The worker tasks usage.
474 tasksUsage
: TasksUsage
476 this.workers
[workerKey
] = {
483 * Removes the given worker from the pool.
485 * @param worker - The worker that will be removed.
487 protected removeWorker (worker
: Worker
): void {
488 const workerKey
= this.getWorkerKey(worker
)
489 this.workers
.splice(workerKey
, 1)
490 this.workerChoiceStrategyContext
.remove(workerKey
)