1 import type { TransferListItem
} from
'node:worker_threads'
3 import { AsyncResource
} from
'node:async_hooks'
4 import { randomUUID
} from
'node:crypto'
5 import { EventEmitterAsyncResource
} from
'node:events'
6 import { performance
} from
'node:perf_hooks'
10 PromiseResponseWrapper
,
12 TaskFunctionProperties
,
13 } from
'../utility-types.js'
17 } from
'../worker/task-functions.js'
22 WorkerNodeEventDetail
,
26 import { defaultBucketSize
} from
'../queues/queue-types.js'
29 buildTaskFunctionProperties
,
41 import { KillBehaviors
} from
'../worker/worker-options.js'
49 type TasksQueueOptions
,
53 WorkerChoiceStrategies
,
54 type WorkerChoiceStrategy
,
55 type WorkerChoiceStrategyOptions
,
56 } from
'./selection-strategies/selection-strategies-types.js'
57 import { WorkerChoiceStrategiesContext
} from
'./selection-strategies/worker-choice-strategies-context.js'
61 checkValidTasksQueueOptions
,
62 checkValidWorkerChoiceStrategy
,
63 getDefaultTasksQueueOptions
,
65 updateRunTimeWorkerUsage
,
66 updateTaskStatisticsWorkerUsage
,
67 updateWaitTimeWorkerUsage
,
70 import { version
} from
'./version.js'
71 import { WorkerNode
} from
'./worker-node.js'
74 * Base class that implements some shared logic for all poolifier pools.
75 * @typeParam Worker - Type of worker which manages this pool.
76 * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
77 * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
79 export abstract class AbstractPool
<
80 Worker
extends IWorker
,
83 > implements IPool
<Worker
, Data
, Response
> {
85 * The task execution response promise map:
86 * - `key`: The message id of each submitted task.
87 * - `value`: An object that contains task's worker node key, execution response promise resolve and reject callbacks, async resource.
89 * When we receive a message from the worker, we get a map entry with the promise resolve/reject bound to the message id.
91 protected promiseResponseMap
: Map
<
92 `${string}-${string}-${string}-${string}-${string}`,
93 PromiseResponseWrapper
<Response
>
95 `${string}-${string}-${string}-${string}-${string}`,
96 PromiseResponseWrapper
<Response
>
100 * Worker choice strategies context referencing worker choice algorithms implementation.
102 protected workerChoiceStrategiesContext
?: WorkerChoiceStrategiesContext
<
109 * This method is the message listener registered on each worker.
110 * @param message - The message received from the worker.
112 protected readonly workerMessageListener
= (
113 message
: MessageValue
<Response
>
115 this.checkMessageWorkerId(message
)
116 const { ready
, taskFunctionsProperties
, taskId
, workerId
} = message
117 if (ready
!= null && taskFunctionsProperties
!= null) {
118 // Worker ready response received from worker
119 this.handleWorkerReadyResponse(message
)
120 } else if (taskFunctionsProperties
!= null) {
121 // Task function properties message received from worker
122 const workerNodeKey
= this.getWorkerNodeKeyByWorkerId(workerId
)
123 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
124 if (workerInfo
!= null) {
125 workerInfo
.taskFunctionsProperties
= taskFunctionsProperties
126 this.sendStatisticsMessageToWorker(workerNodeKey
)
127 this.setTasksQueuePriority(workerNodeKey
)
129 } else if (taskId
!= null) {
130 // Task execution response received from worker
131 this.handleTaskExecutionResponse(message
)
136 * Whether the pool back pressure event has been emitted or not.
138 private backPressureEventEmitted
: boolean
141 * Whether the pool busy event has been emitted or not.
143 private busyEventEmitted
: boolean
146 * Whether the pool is destroying or not.
148 private destroying
: boolean
151 * Gets task function worker choice strategy, if any.
152 * @param name - The task function name.
153 * @returns The task function worker choice strategy if the task function worker choice strategy is defined, `undefined` otherwise.
155 private readonly getTaskFunctionWorkerChoiceStrategy
= (
157 ): undefined | WorkerChoiceStrategy
=> {
158 name
= name
?? DEFAULT_TASK_NAME
159 const taskFunctionsProperties
= this.listTaskFunctionsProperties()
160 if (name
=== DEFAULT_TASK_NAME
) {
161 name
= taskFunctionsProperties
[1]?.name
163 return taskFunctionsProperties
.find(
164 (taskFunctionProperties
: TaskFunctionProperties
) =>
165 taskFunctionProperties
.name
=== name
170 * Gets the worker choice strategies registered in this pool.
171 * @returns The worker choice strategies.
173 private readonly getWorkerChoiceStrategies
=
174 (): Set
<WorkerChoiceStrategy
> => {
176 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
177 this.opts
.workerChoiceStrategy
!,
178 ...this.listTaskFunctionsProperties()
180 (taskFunctionProperties
: TaskFunctionProperties
) =>
181 taskFunctionProperties
.strategy
184 (strategy
: undefined | WorkerChoiceStrategy
) => strategy
!= null
190 * Gets worker node task function priority, if any.
191 * @param workerNodeKey - The worker node key.
192 * @param name - The task function name.
193 * @returns The worker node task function priority if the worker node task function priority is defined, `undefined` otherwise.
195 private readonly getWorkerNodeTaskFunctionPriority
= (
196 workerNodeKey
: number,
198 ): number | undefined => {
199 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
200 if (workerInfo
== null) {
203 name
= name
?? DEFAULT_TASK_NAME
204 if (name
=== DEFAULT_TASK_NAME
) {
205 name
= workerInfo
.taskFunctionsProperties
?.[1]?.name
207 return workerInfo
.taskFunctionsProperties
?.find(
208 (taskFunctionProperties
: TaskFunctionProperties
) =>
209 taskFunctionProperties
.name
=== name
214 * Gets worker node task function worker choice strategy, if any.
215 * @param workerNodeKey - The worker node key.
216 * @param name - The task function name.
217 * @returns The worker node task function worker choice strategy if the worker node task function worker choice strategy is defined, `undefined` otherwise.
219 private readonly getWorkerNodeTaskFunctionWorkerChoiceStrategy
= (
220 workerNodeKey
: number,
222 ): undefined | WorkerChoiceStrategy
=> {
223 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
224 if (workerInfo
== null) {
227 name
= name
?? DEFAULT_TASK_NAME
228 if (name
=== DEFAULT_TASK_NAME
) {
229 name
= workerInfo
.taskFunctionsProperties
?.[1]?.name
231 return workerInfo
.taskFunctionsProperties
?.find(
232 (taskFunctionProperties
: TaskFunctionProperties
) =>
233 taskFunctionProperties
.name
=== name
237 private readonly handleWorkerNodeBackPressureEvent
= (
238 eventDetail
: WorkerNodeEventDetail
241 this.cannotStealTask() ||
243 this.isStealingRatioReached()
248 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
249 if (this.opts
.tasksQueueOptions
!.size
! <= sizeOffset
) {
252 const { workerId
} = eventDetail
253 const sourceWorkerNode
=
254 this.workerNodes
[this.getWorkerNodeKeyByWorkerId(workerId
)]
255 const workerNodes
= this.workerNodes
258 (workerNodeA
, workerNodeB
) =>
259 workerNodeA
.usage
.tasks
.queued
- workerNodeB
.usage
.tasks
.queued
261 for (const [workerNodeKey
, workerNode
] of workerNodes
.entries()) {
262 if (sourceWorkerNode
.usage
.tasks
.queued
=== 0) {
266 workerNode
.info
.id
!== workerId
&&
267 !workerNode
.info
.backPressureStealing
&&
268 workerNode
.usage
.tasks
.queued
<
269 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
270 this.opts
.tasksQueueOptions
!.size
! - sizeOffset
272 workerNode
.info
.backPressureStealing
= true
273 this.stealTask(sourceWorkerNode
, workerNodeKey
)
274 workerNode
.info
.backPressureStealing
= false
279 private readonly handleWorkerNodeIdleEvent
= (
280 eventDetail
: WorkerNodeEventDetail
,
281 previousStolenTask
?: Task
<Data
>
283 const { workerNodeKey
} = eventDetail
284 if (workerNodeKey
== null) {
286 "WorkerNode event detail 'workerNodeKey' property must be defined"
289 const workerNode
= this.workerNodes
[workerNodeKey
]
290 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
291 if (workerNode
== null) {
295 !workerNode
.info
.continuousStealing
&&
296 (this.cannotStealTask() || this.isStealingRatioReached())
300 const workerNodeTasksUsage
= workerNode
.usage
.tasks
302 workerNode
.info
.continuousStealing
&&
303 !this.isWorkerNodeIdle(workerNodeKey
)
305 workerNode
.info
.continuousStealing
= false
306 if (workerNodeTasksUsage
.sequentiallyStolen
> 0) {
307 this.resetTaskSequentiallyStolenStatisticsWorkerUsage(
309 previousStolenTask
?.name
314 workerNode
.info
.continuousStealing
= true
315 const stolenTask
= this.workerNodeStealTask(workerNodeKey
)
316 this.updateTaskSequentiallyStolenStatisticsWorkerUsage(
319 previousStolenTask
?.name
321 sleep(exponentialDelay(workerNodeTasksUsage
.sequentiallyStolen
))
323 this.handleWorkerNodeIdleEvent(eventDetail
, stolenTask
)
326 .catch((error
: unknown
) => {
327 this.emitter
?.emit(PoolEvents
.error
, error
)
331 private readonly isStealingRatioReached
= (): boolean => {
333 this.opts
.tasksQueueOptions
?.tasksStealingRatio
=== 0 ||
334 (this.info
.stealingWorkerNodes
?? 0) >
336 this.workerNodes
.length
*
337 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
338 this.opts
.tasksQueueOptions
!.tasksStealingRatio
!
344 * Whether the pool ready event has been emitted or not.
346 private readyEventEmitted
: boolean
349 * Whether the pool is started or not.
351 private started
: boolean
354 * Whether the pool is starting or not.
356 private starting
: boolean
359 * Whether the minimum number of workers is starting or not.
361 private startingMinimumNumberOfWorkers
: boolean
364 * The start timestamp of the pool.
366 private startTimestamp
?: number
368 private readonly stealTask
= (
369 sourceWorkerNode
: IWorkerNode
<Worker
, Data
>,
370 destinationWorkerNodeKey
: number
371 ): Task
<Data
> | undefined => {
372 const destinationWorkerNode
= this.workerNodes
[destinationWorkerNodeKey
]
373 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
374 if (destinationWorkerNode
== null) {
377 // Avoid cross and cascade task stealing. Could be smarter by checking stealing/stolen worker ids pair.
379 !sourceWorkerNode
.info
.ready
||
380 sourceWorkerNode
.info
.stolen
||
381 sourceWorkerNode
.info
.stealing
||
382 !destinationWorkerNode
.info
.ready
||
383 destinationWorkerNode
.info
.stolen
||
384 destinationWorkerNode
.info
.stealing
388 destinationWorkerNode
.info
.stealing
= true
389 sourceWorkerNode
.info
.stolen
= true
390 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
391 const stolenTask
= sourceWorkerNode
.dequeueLastPrioritizedTask()!
392 sourceWorkerNode
.info
.stolen
= false
393 destinationWorkerNode
.info
.stealing
= false
394 this.handleTask(destinationWorkerNodeKey
, stolenTask
)
395 this.updateTaskStolenStatisticsWorkerUsage(
396 destinationWorkerNodeKey
,
397 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
404 * The task functions added at runtime map:
405 * - `key`: The task function name.
406 * - `value`: The task function object.
408 private readonly taskFunctions
: Map
<
410 TaskFunctionObject
<Data
, Response
>
413 private readonly workerNodeStealTask
= (
414 workerNodeKey
: number
415 ): Task
<Data
> | undefined => {
416 const workerNodes
= this.workerNodes
419 (workerNodeA
, workerNodeB
) =>
420 workerNodeB
.usage
.tasks
.queued
- workerNodeA
.usage
.tasks
.queued
422 const sourceWorkerNode
= workerNodes
.find(
423 (sourceWorkerNode
, sourceWorkerNodeKey
) =>
424 sourceWorkerNodeKey
!== workerNodeKey
&&
425 sourceWorkerNode
.usage
.tasks
.queued
> 0
427 if (sourceWorkerNode
!= null) {
428 return this.stealTask(sourceWorkerNode
, workerNodeKey
)
433 public emitter
?: EventEmitterAsyncResource
436 public readonly workerNodes
: IWorkerNode
<Worker
, Data
>[] = []
439 * Constructs a new poolifier pool.
440 * @param minimumNumberOfWorkers - Minimum number of workers that this pool manages.
441 * @param filePath - Path to the worker file.
442 * @param opts - Options for the pool.
443 * @param maximumNumberOfWorkers - Maximum number of workers that this pool manages.
446 protected readonly minimumNumberOfWorkers
: number,
447 protected readonly filePath
: string,
448 protected readonly opts
: PoolOptions
<Worker
>,
449 protected readonly maximumNumberOfWorkers
?: number
451 if (!this.isMain()) {
453 'Cannot start a pool from a worker with the same type as the pool'
457 checkFilePath(this.filePath
)
458 this.checkMinimumNumberOfWorkers(this.minimumNumberOfWorkers
)
459 this.checkPoolOptions(this.opts
)
461 this.chooseWorkerNode
= this.chooseWorkerNode
.bind(this)
462 this.executeTask
= this.executeTask
.bind(this)
463 this.enqueueTask
= this.enqueueTask
.bind(this)
465 if (this.opts
.enableEvents
=== true) {
466 this.initEventEmitter()
468 this.workerChoiceStrategiesContext
= new WorkerChoiceStrategiesContext
<
474 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
475 [this.opts
.workerChoiceStrategy
!],
476 this.opts
.workerChoiceStrategyOptions
481 this.taskFunctions
= new Map
<string, TaskFunctionObject
<Data
, Response
>>()
484 this.starting
= false
485 this.destroying
= false
486 this.readyEventEmitted
= false
487 this.busyEventEmitted
= false
488 this.backPressureEventEmitted
= false
489 this.startingMinimumNumberOfWorkers
= false
490 if (this.opts
.startWorkers
=== true) {
496 * Hook executed after the worker task execution.
498 * @param workerNodeKey - The worker node key.
499 * @param message - The received message.
501 protected afterTaskExecutionHook (
502 workerNodeKey
: number,
503 message
: MessageValue
<Response
>
505 let needWorkerChoiceStrategiesUpdate
= false
506 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
507 if (this.workerNodes
[workerNodeKey
]?.usage
!= null) {
508 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
509 updateTaskStatisticsWorkerUsage(workerUsage
, message
)
510 updateRunTimeWorkerUsage(
511 this.workerChoiceStrategiesContext
,
515 updateEluWorkerUsage(
516 this.workerChoiceStrategiesContext
,
520 needWorkerChoiceStrategiesUpdate
= true
523 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
524 message
.taskPerformance
?.name
!= null &&
525 this.workerNodes
[workerNodeKey
].getTaskFunctionWorkerUsage(
526 message
.taskPerformance
.name
529 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
530 const taskFunctionWorkerUsage
= this.workerNodes
[
532 ].getTaskFunctionWorkerUsage(message
.taskPerformance
.name
)!
533 updateTaskStatisticsWorkerUsage(taskFunctionWorkerUsage
, message
)
534 updateRunTimeWorkerUsage(
535 this.workerChoiceStrategiesContext
,
536 taskFunctionWorkerUsage
,
539 updateEluWorkerUsage(
540 this.workerChoiceStrategiesContext
,
541 taskFunctionWorkerUsage
,
544 needWorkerChoiceStrategiesUpdate
= true
546 if (needWorkerChoiceStrategiesUpdate
) {
547 this.workerChoiceStrategiesContext
?.update(workerNodeKey
)
552 * Method hooked up after a worker node has been newly created.
554 * @param workerNodeKey - The newly created worker node key.
556 protected afterWorkerNodeSetup (workerNodeKey
: number): void {
557 // Listen to worker messages.
558 this.registerWorkerMessageListener(
560 this.workerMessageListener
562 // Send the startup message to worker.
563 this.sendStartupMessageToWorker(workerNodeKey
)
564 // Send the statistics message to worker.
565 this.sendStatisticsMessageToWorker(workerNodeKey
)
566 if (this.opts
.enableTasksQueue
=== true) {
567 if (this.opts
.tasksQueueOptions
?.taskStealing
=== true) {
568 this.workerNodes
[workerNodeKey
].on(
570 this.handleWorkerNodeIdleEvent
573 if (this.opts
.tasksQueueOptions
?.tasksStealingOnBackPressure
=== true) {
574 this.workerNodes
[workerNodeKey
].on(
576 this.handleWorkerNodeBackPressureEvent
583 * Hook executed before the worker task execution.
585 * @param workerNodeKey - The worker node key.
586 * @param task - The task to execute.
588 protected beforeTaskExecutionHook (
589 workerNodeKey
: number,
592 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
593 if (this.workerNodes
[workerNodeKey
]?.usage
!= null) {
594 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
595 ++workerUsage
.tasks
.executing
596 updateWaitTimeWorkerUsage(
597 this.workerChoiceStrategiesContext
,
603 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
604 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
605 this.workerNodes
[workerNodeKey
].getTaskFunctionWorkerUsage(task
.name
!) !=
608 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
609 const taskFunctionWorkerUsage
= this.workerNodes
[
611 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
612 ].getTaskFunctionWorkerUsage(task
.name
!)!
613 ++taskFunctionWorkerUsage
.tasks
.executing
614 updateWaitTimeWorkerUsage(
615 this.workerChoiceStrategiesContext
,
616 taskFunctionWorkerUsage
,
623 * Emits dynamic worker creation events.
625 protected abstract checkAndEmitDynamicWorkerCreationEvents (): void
628 * Emits dynamic worker destruction events.
630 protected abstract checkAndEmitDynamicWorkerDestructionEvents (): void
633 * Creates a new, completely set up dynamic worker node.
634 * @returns New, completely set up dynamic worker node key.
636 protected createAndSetupDynamicWorkerNode (): number {
637 const workerNodeKey
= this.createAndSetupWorkerNode()
638 this.registerWorkerMessageListener(workerNodeKey
, message
=> {
639 this.checkMessageWorkerId(message
)
640 const localWorkerNodeKey
= this.getWorkerNodeKeyByWorkerId(
643 // Kill message received from worker
645 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
646 (isKillBehavior(KillBehaviors
.SOFT
, message
.kill
) &&
647 this.isWorkerNodeIdle(localWorkerNodeKey
) &&
648 !this.isWorkerNodeStealing(localWorkerNodeKey
))
650 // Flag the worker node as not ready immediately
651 this.flagWorkerNodeAsNotReady(localWorkerNodeKey
)
652 this.destroyWorkerNode(localWorkerNodeKey
).catch((error
: unknown
) => {
653 this.emitter
?.emit(PoolEvents
.error
, error
)
657 this.sendToWorker(workerNodeKey
, {
660 if (this.taskFunctions
.size
> 0) {
661 for (const [taskFunctionName
, taskFunctionObject
] of this.taskFunctions
) {
662 this.sendTaskFunctionOperationToWorker(workerNodeKey
, {
663 taskFunction
: taskFunctionObject
.taskFunction
.toString(),
664 taskFunctionOperation
: 'add',
665 taskFunctionProperties
: buildTaskFunctionProperties(
669 }).catch((error
: unknown
) => {
670 this.emitter
?.emit(PoolEvents
.error
, error
)
674 const workerNode
= this.workerNodes
[workerNodeKey
]
675 workerNode
.info
.dynamic
= true
677 this.workerChoiceStrategiesContext
?.getPolicy().dynamicWorkerReady
===
679 this.workerChoiceStrategiesContext
?.getPolicy().dynamicWorkerUsage
===
682 workerNode
.info
.ready
= true
684 this.initWorkerNodeUsage(workerNode
)
685 this.checkAndEmitDynamicWorkerCreationEvents()
690 * Creates a new, completely set up worker node.
691 * @returns New, completely set up worker node key.
693 protected createAndSetupWorkerNode (): number {
694 const workerNode
= this.createWorkerNode()
695 workerNode
.registerWorkerEventHandler(
697 this.opts
.onlineHandler
?? EMPTY_FUNCTION
699 workerNode
.registerWorkerEventHandler(
701 this.opts
.messageHandler
?? EMPTY_FUNCTION
703 workerNode
.registerWorkerEventHandler(
705 this.opts
.errorHandler
?? EMPTY_FUNCTION
707 workerNode
.registerOnceWorkerEventHandler('error', (error
: Error) => {
708 workerNode
.info
.ready
= false
709 this.emitter
?.emit(PoolEvents
.error
, error
)
713 this.opts
.restartWorkerOnError
=== true
715 if (workerNode
.info
.dynamic
) {
716 this.createAndSetupDynamicWorkerNode()
717 } else if (!this.startingMinimumNumberOfWorkers
) {
718 this.startMinimumNumberOfWorkers(true)
724 this.opts
.enableTasksQueue
=== true
726 this.redistributeQueuedTasks(this.workerNodes
.indexOf(workerNode
))
728 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition, promise/no-promise-in-callback
729 workerNode
?.terminate().catch((error
: unknown
) => {
730 this.emitter
?.emit(PoolEvents
.error
, error
)
733 workerNode
.registerWorkerEventHandler(
735 this.opts
.exitHandler
?? EMPTY_FUNCTION
737 workerNode
.registerOnceWorkerEventHandler('exit', () => {
738 this.removeWorkerNode(workerNode
)
741 !this.startingMinimumNumberOfWorkers
&&
744 this.startMinimumNumberOfWorkers(true)
747 const workerNodeKey
= this.addWorkerNode(workerNode
)
748 this.afterWorkerNodeSetup(workerNodeKey
)
753 * Deregisters a listener callback on the worker given its worker node key.
754 * @param workerNodeKey - The worker node key.
755 * @param listener - The message listener callback.
757 protected abstract deregisterWorkerMessageListener
<
758 Message
extends Data
| Response
760 workerNodeKey
: number,
761 listener
: (message
: MessageValue
<Message
>) => void
765 * Terminates the worker node given its worker node key.
766 * @param workerNodeKey - The worker node key.
768 protected async destroyWorkerNode (workerNodeKey
: number): Promise
<void> {
769 this.flagWorkerNodeAsNotReady(workerNodeKey
)
770 const flushedTasks
= this.flushTasksQueue(workerNodeKey
)
771 const workerNode
= this.workerNodes
[workerNodeKey
]
772 await waitWorkerNodeEvents(
776 this.opts
.tasksQueueOptions
?.tasksFinishedTimeout
??
777 getDefaultTasksQueueOptions(
778 this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
779 ).tasksFinishedTimeout
781 await this.sendKillMessageToWorker(workerNodeKey
)
782 await workerNode
.terminate()
785 protected flagWorkerNodeAsNotReady (workerNodeKey
: number): void {
786 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
787 if (workerInfo
!= null) {
788 workerInfo
.ready
= false
792 protected flushTasksQueue (workerNodeKey
: number): number {
794 while (this.tasksQueueSize(workerNodeKey
) > 0) {
795 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
796 this.executeTask(workerNodeKey
, this.dequeueTask(workerNodeKey
)!)
799 this.workerNodes
[workerNodeKey
].clearTasksQueue()
804 * Gets the worker information given its worker node key.
805 * @param workerNodeKey - The worker node key.
806 * @returns The worker information.
808 protected getWorkerInfo (workerNodeKey
: number): undefined | WorkerInfo
{
809 return this.workerNodes
[workerNodeKey
]?.info
813 * Whether the worker nodes are back pressured or not.
814 * @returns Worker nodes back pressure boolean status.
816 protected internalBackPressure (): boolean {
818 this.workerNodes
.reduce(
819 (accumulator
, _
, workerNodeKey
) =>
820 this.isWorkerNodeBackPressured(workerNodeKey
)
824 ) === this.workerNodes
.length
829 * Whether worker nodes are executing concurrently their tasks quota or not.
830 * @returns Worker nodes busyness boolean status.
832 protected internalBusy (): boolean {
834 this.workerNodes
.reduce(
835 (accumulator
, _
, workerNodeKey
) =>
836 this.isWorkerNodeBusy(workerNodeKey
) ? accumulator
+ 1 : accumulator
,
838 ) === this.workerNodes
.length
843 * Returns whether the worker is the main worker or not.
844 * @returns `true` if the worker is the main worker, `false` otherwise.
846 protected abstract isMain (): boolean
849 * Registers once a listener callback on the worker given its worker node key.
850 * @param workerNodeKey - The worker node key.
851 * @param listener - The message listener callback.
853 protected abstract registerOnceWorkerMessageListener
<
854 Message
extends Data
| Response
856 workerNodeKey
: number,
857 listener
: (message
: MessageValue
<Message
>) => void
861 * Registers a listener callback on the worker given its worker node key.
862 * @param workerNodeKey - The worker node key.
863 * @param listener - The message listener callback.
865 protected abstract registerWorkerMessageListener
<
866 Message
extends Data
| Response
868 workerNodeKey
: number,
869 listener
: (message
: MessageValue
<Message
>) => void
873 * Sends the startup message to worker given its worker node key.
874 * @param workerNodeKey - The worker node key.
876 protected abstract sendStartupMessageToWorker (workerNodeKey
: number): void
879 * Sends a message to worker given its worker node key.
880 * @param workerNodeKey - The worker node key.
881 * @param message - The message.
882 * @param transferList - The optional array of transferable objects.
884 protected abstract sendToWorker (
885 workerNodeKey
: number,
886 message
: MessageValue
<Data
>,
887 transferList
?: readonly TransferListItem
[]
891 * Setup hook to execute code before worker nodes are created in the abstract constructor.
894 protected setupHook (): void {
895 /* Intentionally empty */
899 * Conditions for dynamic worker creation.
900 * @returns Whether to create a dynamic worker or not.
902 protected abstract shallCreateDynamicWorker (): boolean
905 * Adds the given worker node in the pool worker nodes.
906 * @param workerNode - The worker node.
907 * @returns The added worker node key.
908 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the added worker node is not found.
910 private addWorkerNode (workerNode
: IWorkerNode
<Worker
, Data
>): number {
911 this.workerNodes
.push(workerNode
)
912 const workerNodeKey
= this.workerNodes
.indexOf(workerNode
)
913 if (workerNodeKey
=== -1) {
914 throw new Error('Worker added not found in worker nodes')
919 private buildTasksQueueOptions (
920 tasksQueueOptions
: TasksQueueOptions
| undefined
921 ): TasksQueueOptions
{
923 ...getDefaultTasksQueueOptions(
924 this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
926 ...this.opts
.tasksQueueOptions
,
927 ...tasksQueueOptions
,
931 private cannotStealTask (): boolean {
932 return this.workerNodes
.length
<= 1 || this.info
.queuedTasks
=== 0
935 private checkAndEmitReadyEvent (): void {
936 if (this.emitter
!= null && !this.readyEventEmitted
&& this.ready
) {
937 this.emitter
.emit(PoolEvents
.ready
, this.info
)
938 this.readyEventEmitted
= true
942 private checkAndEmitTaskDequeuingEvents (): void {
944 this.emitter
!= null &&
945 this.backPressureEventEmitted
&&
948 this.emitter
.emit(PoolEvents
.backPressureEnd
, this.info
)
949 this.backPressureEventEmitted
= false
953 private checkAndEmitTaskExecutionEvents (): void {
954 if (this.emitter
!= null && !this.busyEventEmitted
&& this.busy
) {
955 this.emitter
.emit(PoolEvents
.busy
, this.info
)
956 this.busyEventEmitted
= true
960 private checkAndEmitTaskExecutionFinishedEvents (): void {
961 if (this.emitter
!= null && this.busyEventEmitted
&& !this.busy
) {
962 this.emitter
.emit(PoolEvents
.busyEnd
, this.info
)
963 this.busyEventEmitted
= false
967 private checkAndEmitTaskQueuingEvents (): void {
969 this.emitter
!= null &&
970 !this.backPressureEventEmitted
&&
973 this.emitter
.emit(PoolEvents
.backPressure
, this.info
)
974 this.backPressureEventEmitted
= true
979 * Checks if the worker id sent in the received message from a worker is valid.
980 * @param message - The received message.
981 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the worker id is invalid.
983 private checkMessageWorkerId (message
: MessageValue
<Data
| Response
>): void {
984 if (message
.workerId
== null) {
985 throw new Error('Worker message received without worker id')
986 } else if (this.getWorkerNodeKeyByWorkerId(message
.workerId
) === -1) {
988 `Worker message received from unknown worker '${message.workerId.toString()}'`
993 private checkMinimumNumberOfWorkers (
994 minimumNumberOfWorkers
: number | undefined
996 if (minimumNumberOfWorkers
== null) {
998 'Cannot instantiate a pool without specifying the number of workers'
1000 } else if (!Number.isSafeInteger(minimumNumberOfWorkers
)) {
1001 throw new TypeError(
1002 'Cannot instantiate a pool with a non safe integer number of workers'
1004 } else if (minimumNumberOfWorkers
< 0) {
1005 throw new RangeError(
1006 'Cannot instantiate a pool with a negative number of workers'
1008 } else if (this.type === PoolTypes
.fixed
&& minimumNumberOfWorkers
=== 0) {
1009 throw new RangeError('Cannot instantiate a fixed pool with zero worker')
1013 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
1014 if (isPlainObject(opts
)) {
1015 this.opts
.startWorkers
= opts
.startWorkers
?? true
1016 checkValidWorkerChoiceStrategy(opts
.workerChoiceStrategy
)
1017 this.opts
.workerChoiceStrategy
=
1018 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
1019 this.checkValidWorkerChoiceStrategyOptions(
1020 opts
.workerChoiceStrategyOptions
1022 if (opts
.workerChoiceStrategyOptions
!= null) {
1023 this.opts
.workerChoiceStrategyOptions
= opts
.workerChoiceStrategyOptions
1025 this.opts
.restartWorkerOnError
= opts
.restartWorkerOnError
?? true
1026 this.opts
.enableEvents
= opts
.enableEvents
?? true
1027 this.opts
.enableTasksQueue
= opts
.enableTasksQueue
?? false
1028 if (this.opts
.enableTasksQueue
) {
1029 checkValidTasksQueueOptions(opts
.tasksQueueOptions
)
1030 this.opts
.tasksQueueOptions
= this.buildTasksQueueOptions(
1031 opts
.tasksQueueOptions
1035 throw new TypeError('Invalid pool options: must be a plain object')
1039 private checkPoolType (): void {
1040 if (this.type === PoolTypes
.fixed
&& this.maximumNumberOfWorkers
!= null) {
1042 'Cannot instantiate a fixed pool with a maximum number of workers specified at initialization'
1047 private checkValidWorkerChoiceStrategyOptions (
1048 workerChoiceStrategyOptions
: undefined | WorkerChoiceStrategyOptions
1051 workerChoiceStrategyOptions
!= null &&
1052 !isPlainObject(workerChoiceStrategyOptions
)
1054 throw new TypeError(
1055 'Invalid worker choice strategy options: must be a plain object'
1059 workerChoiceStrategyOptions
?.weights
!= null &&
1060 Object.keys(workerChoiceStrategyOptions
.weights
).length
!==
1061 (this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
)
1064 'Invalid worker choice strategy options: must have a weight for each worker node'
1068 workerChoiceStrategyOptions
?.measurement
!= null &&
1069 !Object.values(Measurements
).includes(
1070 workerChoiceStrategyOptions
.measurement
1074 `Invalid worker choice strategy options: invalid measurement '${workerChoiceStrategyOptions.measurement}'`
1080 * Chooses a worker node for the next task.
1081 * @param name - The task function name.
1082 * @returns The chosen worker node key.
1084 private chooseWorkerNode (name
?: string): number {
1085 if (this.shallCreateDynamicWorker()) {
1086 const workerNodeKey
= this.createAndSetupDynamicWorkerNode()
1088 this.workerChoiceStrategiesContext
?.getPolicy().dynamicWorkerUsage
===
1091 return workerNodeKey
1094 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1095 return this.workerChoiceStrategiesContext
!.execute(
1096 this.getTaskFunctionWorkerChoiceStrategy(name
)
1101 * Creates a worker node.
1102 * @returns The created worker node.
1104 private createWorkerNode (): IWorkerNode
<Worker
, Data
> {
1105 const workerNode
= new WorkerNode
<Worker
, Data
>(
1110 tasksQueueBackPressureSize
:
1111 this.opts
.tasksQueueOptions
?.size
??
1112 getDefaultTasksQueueOptions(
1113 this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
1115 tasksQueueBucketSize
: defaultBucketSize
,
1116 tasksQueuePriority
: this.getTasksQueuePriority(),
1117 workerOptions
: this.opts
.workerOptions
,
1120 // Flag the worker node as ready at pool startup.
1121 if (this.starting
) {
1122 workerNode
.info
.ready
= true
1127 private dequeueTask (workerNodeKey
: number): Task
<Data
> | undefined {
1128 const task
= this.workerNodes
[workerNodeKey
].dequeueTask()
1129 this.checkAndEmitTaskDequeuingEvents()
1133 private enqueueTask (workerNodeKey
: number, task
: Task
<Data
>): number {
1134 const tasksQueueSize
= this.workerNodes
[workerNodeKey
].enqueueTask(task
)
1135 this.checkAndEmitTaskQueuingEvents()
1136 return tasksQueueSize
1140 * Executes the given task on the worker given its worker node key.
1141 * @param workerNodeKey - The worker node key.
1142 * @param task - The task to execute.
1144 private executeTask (workerNodeKey
: number, task
: Task
<Data
>): void {
1145 this.beforeTaskExecutionHook(workerNodeKey
, task
)
1146 this.sendToWorker(workerNodeKey
, task
, task
.transferList
)
1147 this.checkAndEmitTaskExecutionEvents()
1150 private flushTasksQueues (): void {
1151 for (const workerNodeKey
of this.workerNodes
.keys()) {
1152 this.flushTasksQueue(workerNodeKey
)
1156 private getTasksQueuePriority (): boolean {
1157 return this.listTaskFunctionsProperties().some(
1158 taskFunctionProperties
=> taskFunctionProperties
.priority
!= null
1163 * Gets the worker node key given its worker id.
1164 * @param workerId - The worker id.
1165 * @returns The worker node key if the worker id is found in the pool worker nodes, `-1` otherwise.
1167 private getWorkerNodeKeyByWorkerId (workerId
: number | undefined): number {
1168 return this.workerNodes
.findIndex(
1169 workerNode
=> workerNode
.info
.id
=== workerId
1173 private handleTask (workerNodeKey
: number, task
: Task
<Data
>): void {
1174 if (this.shallExecuteTask(workerNodeKey
)) {
1175 this.executeTask(workerNodeKey
, task
)
1177 this.enqueueTask(workerNodeKey
, task
)
1181 private handleTaskExecutionResponse (message
: MessageValue
<Response
>): void {
1182 const { data
, taskId
, workerError
} = message
1183 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1184 const promiseResponse
= this.promiseResponseMap
.get(taskId
!)
1185 if (promiseResponse
!= null) {
1186 const { asyncResource
, reject
, resolve
, workerNodeKey
} = promiseResponse
1187 const workerNode
= this.workerNodes
[workerNodeKey
]
1188 if (workerError
!= null) {
1189 this.emitter
?.emit(PoolEvents
.taskError
, workerError
)
1190 asyncResource
!= null
1191 ? asyncResource
.runInAsyncScope(
1196 : reject(workerError
.message
)
1198 asyncResource
!= null
1199 ? asyncResource
.runInAsyncScope(resolve
, this.emitter
, data
)
1200 : resolve(data
as Response
)
1202 asyncResource
?.emitDestroy()
1203 this.afterTaskExecutionHook(workerNodeKey
, message
)
1204 this.checkAndEmitTaskExecutionFinishedEvents()
1205 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1206 this.promiseResponseMap
.delete(taskId
!)
1207 if (this.opts
.enableTasksQueue
=== true && !this.destroying
) {
1209 !this.isWorkerNodeBusy(workerNodeKey
) &&
1210 this.tasksQueueSize(workerNodeKey
) > 0
1212 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1213 this.executeTask(workerNodeKey
, this.dequeueTask(workerNodeKey
)!)
1215 if (this.isWorkerNodeIdle(workerNodeKey
)) {
1216 workerNode
.emit('idle', {
1221 // FIXME: cannot be theoretically undefined. Schedule in the next tick to avoid race conditions?
1222 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1223 workerNode
?.emit('taskFinished', taskId
)
1227 private handleWorkerReadyResponse (message
: MessageValue
<Response
>): void {
1228 const { ready
, taskFunctionsProperties
, workerId
} = message
1229 if (ready
== null || !ready
) {
1230 // eslint-disable-next-line @typescript-eslint/restrict-template-expressions
1231 throw new Error(`Worker ${workerId?.toString()} failed to initialize`)
1233 const workerNodeKey
= this.getWorkerNodeKeyByWorkerId(workerId
)
1234 const workerNode
= this.workerNodes
[workerNodeKey
]
1235 workerNode
.info
.ready
= ready
1236 workerNode
.info
.taskFunctionsProperties
= taskFunctionsProperties
1237 this.sendStatisticsMessageToWorker(workerNodeKey
)
1238 this.setTasksQueuePriority(workerNodeKey
)
1239 this.checkAndEmitReadyEvent()
1242 private initEventEmitter (): void {
1243 this.emitter
= new EventEmitterAsyncResource({
1244 name
: `poolifier:${this.type}-${this.worker}-pool`,
1249 * Initializes the worker node usage with sensible default values gathered during runtime.
1250 * @param workerNode - The worker node.
1252 private initWorkerNodeUsage (workerNode
: IWorkerNode
<Worker
, Data
>): void {
1254 this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
1255 .runTime
.aggregate
=== true
1257 workerNode
.usage
.runTime
.aggregate
= min(
1258 ...this.workerNodes
.map(
1260 workerNode
.usage
.runTime
.aggregate
?? Number.POSITIVE_INFINITY
1265 this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
1266 .waitTime
.aggregate
=== true
1268 workerNode
.usage
.waitTime
.aggregate
= min(
1269 ...this.workerNodes
.map(
1271 workerNode
.usage
.waitTime
.aggregate
?? Number.POSITIVE_INFINITY
1276 this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements().elu
1279 workerNode
.usage
.elu
.active
.aggregate
= min(
1280 ...this.workerNodes
.map(
1282 workerNode
.usage
.elu
.active
.aggregate
?? Number.POSITIVE_INFINITY
1288 private async internalExecute (
1291 transferList
?: readonly TransferListItem
[]
1292 ): Promise
<Response
> {
1293 return await new Promise
<Response
>((resolve
, reject
) => {
1294 const timestamp
= performance
.now()
1295 const workerNodeKey
= this.chooseWorkerNode(name
)
1296 const task
: Task
<Data
> = {
1297 data
: data
?? ({} as Data
),
1298 name
: name
?? DEFAULT_TASK_NAME
,
1299 priority
: this.getWorkerNodeTaskFunctionPriority(workerNodeKey
, name
),
1300 strategy
: this.getWorkerNodeTaskFunctionWorkerChoiceStrategy(
1304 taskId
: randomUUID(),
1308 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1309 this.promiseResponseMap
.set(task
.taskId
!, {
1313 ...(this.emitter
!= null && {
1314 asyncResource
: new AsyncResource('poolifier:task', {
1315 requireManualDestroy
: true,
1316 triggerAsyncId
: this.emitter
.asyncId
,
1321 this.opts
.enableTasksQueue
=== false ||
1322 (this.opts
.enableTasksQueue
=== true &&
1323 this.shallExecuteTask(workerNodeKey
))
1325 this.executeTask(workerNodeKey
, task
)
1327 this.enqueueTask(workerNodeKey
, task
)
1332 private isWorkerNodeBackPressured (workerNodeKey
: number): boolean {
1333 const workerNode
= this.workerNodes
[workerNodeKey
]
1334 return workerNode
.info
.ready
&& workerNode
.info
.backPressure
1337 private isWorkerNodeBusy (workerNodeKey
: number): boolean {
1338 const workerNode
= this.workerNodes
[workerNodeKey
]
1339 if (this.opts
.enableTasksQueue
=== true) {
1341 workerNode
.info
.ready
&&
1342 workerNode
.usage
.tasks
.executing
>=
1343 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1344 this.opts
.tasksQueueOptions
!.concurrency
!
1347 return workerNode
.info
.ready
&& workerNode
.usage
.tasks
.executing
> 0
1350 private isWorkerNodeIdle (workerNodeKey
: number): boolean {
1351 const workerNode
= this.workerNodes
[workerNodeKey
]
1352 if (this.opts
.enableTasksQueue
=== true) {
1354 workerNode
.info
.ready
&&
1355 workerNode
.usage
.tasks
.executing
=== 0 &&
1356 this.tasksQueueSize(workerNodeKey
) === 0
1359 return workerNode
.info
.ready
&& workerNode
.usage
.tasks
.executing
=== 0
1362 private isWorkerNodeStealing (workerNodeKey
: number): boolean {
1363 const workerNode
= this.workerNodes
[workerNodeKey
]
1365 workerNode
.info
.ready
&&
1366 (workerNode
.info
.continuousStealing
||
1367 workerNode
.info
.backPressureStealing
)
1371 private redistributeQueuedTasks (sourceWorkerNodeKey
: number): void {
1372 if (sourceWorkerNodeKey
=== -1 || this.cannotStealTask()) {
1375 while (this.tasksQueueSize(sourceWorkerNodeKey
) > 0) {
1376 const destinationWorkerNodeKey
= this.workerNodes
.reduce(
1377 (minWorkerNodeKey
, workerNode
, workerNodeKey
, workerNodes
) => {
1378 return sourceWorkerNodeKey
!== workerNodeKey
&&
1379 workerNode
.info
.ready
&&
1380 workerNode
.usage
.tasks
.queued
<
1381 workerNodes
[minWorkerNodeKey
].usage
.tasks
.queued
1388 destinationWorkerNodeKey
,
1389 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1390 this.dequeueTask(sourceWorkerNodeKey
)!
1396 * Removes the worker node from the pool worker nodes.
1397 * @param workerNode - The worker node.
1399 private removeWorkerNode (workerNode
: IWorkerNode
<Worker
, Data
>): void {
1400 const workerNodeKey
= this.workerNodes
.indexOf(workerNode
)
1401 if (workerNodeKey
!== -1) {
1402 this.workerNodes
.splice(workerNodeKey
, 1)
1403 this.workerChoiceStrategiesContext
?.remove(workerNodeKey
)
1404 workerNode
.info
.dynamic
&&
1405 this.checkAndEmitDynamicWorkerDestructionEvents()
1409 private resetTaskSequentiallyStolenStatisticsWorkerUsage (
1410 workerNodeKey
: number,
1413 const workerNode
= this.workerNodes
[workerNodeKey
]
1414 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1415 if (workerNode
?.usage
!= null) {
1416 workerNode
.usage
.tasks
.sequentiallyStolen
= 0
1420 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1421 workerNode
.getTaskFunctionWorkerUsage(taskName
) != null
1423 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1424 workerNode
.getTaskFunctionWorkerUsage(
1426 )!.tasks
.sequentiallyStolen
= 0
1430 private async sendKillMessageToWorker (workerNodeKey
: number): Promise
<void> {
1431 await new Promise
<void>((resolve
, reject
) => {
1432 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1433 if (this.workerNodes
[workerNodeKey
] == null) {
1437 const killMessageListener
= (message
: MessageValue
<Response
>): void => {
1438 this.checkMessageWorkerId(message
)
1439 if (message
.kill
=== 'success') {
1441 } else if (message
.kill
=== 'failure') {
1444 // eslint-disable-next-line @typescript-eslint/restrict-template-expressions
1445 `Kill message handling failed on worker ${message.workerId?.toString()}`
1450 // FIXME: should be registered only once
1451 this.registerWorkerMessageListener(workerNodeKey
, killMessageListener
)
1452 this.sendToWorker(workerNodeKey
, { kill
: true })
1457 * Sends the statistics message to worker given its worker node key.
1458 * @param workerNodeKey - The worker node key.
1460 private sendStatisticsMessageToWorker (workerNodeKey
: number): void {
1461 this.sendToWorker(workerNodeKey
, {
1464 this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
1465 .elu
.aggregate
?? false,
1467 this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
1468 .runTime
.aggregate
?? false,
1473 private async sendTaskFunctionOperationToWorker (
1474 workerNodeKey
: number,
1475 message
: MessageValue
<Data
>
1476 ): Promise
<boolean> {
1477 return await new Promise
<boolean>((resolve
, reject
) => {
1478 const taskFunctionOperationListener
= (
1479 message
: MessageValue
<Response
>
1481 this.checkMessageWorkerId(message
)
1482 const workerId
= this.getWorkerInfo(workerNodeKey
)?.id
1484 message
.taskFunctionOperationStatus
!= null &&
1485 message
.workerId
=== workerId
1487 if (message
.taskFunctionOperationStatus
) {
1492 // eslint-disable-next-line @typescript-eslint/restrict-template-expressions
1493 `Task function operation '${message.taskFunctionOperation?.toString()}' failed on worker ${message.workerId?.toString()} with error: '${
1494 // eslint-disable-next-line @typescript-eslint/restrict-template-expressions
1495 message.workerError?.message
1500 this.deregisterWorkerMessageListener(
1501 this.getWorkerNodeKeyByWorkerId(message
.workerId
),
1502 taskFunctionOperationListener
1506 this.registerWorkerMessageListener(
1508 taskFunctionOperationListener
1510 this.sendToWorker(workerNodeKey
, message
)
1514 private async sendTaskFunctionOperationToWorkers (
1515 message
: MessageValue
<Data
>
1516 ): Promise
<boolean> {
1517 return await new Promise
<boolean>((resolve
, reject
) => {
1518 const responsesReceived
= new Array<MessageValue
<Response
>>()
1519 const taskFunctionOperationsListener
= (
1520 message
: MessageValue
<Response
>
1522 this.checkMessageWorkerId(message
)
1523 if (message
.taskFunctionOperationStatus
!= null) {
1524 responsesReceived
.push(message
)
1525 if (responsesReceived
.length
=== this.workerNodes
.length
) {
1527 responsesReceived
.every(
1528 message
=> message
.taskFunctionOperationStatus
=== true
1533 responsesReceived
.some(
1534 message
=> message
.taskFunctionOperationStatus
=== false
1537 const errorResponse
= responsesReceived
.find(
1538 response
=> response
.taskFunctionOperationStatus
=== false
1542 `Task function operation '${
1543 message.taskFunctionOperation as string
1544 // eslint-disable-next-line @typescript-eslint/restrict-template-expressions
1545 }' failed on worker ${errorResponse?.workerId?.toString()} with error: '${
1546 // eslint-disable-next-line @typescript-eslint/restrict-template-expressions
1547 errorResponse?.workerError?.message
1552 this.deregisterWorkerMessageListener(
1553 this.getWorkerNodeKeyByWorkerId(message
.workerId
),
1554 taskFunctionOperationsListener
1559 for (const workerNodeKey
of this.workerNodes
.keys()) {
1560 this.registerWorkerMessageListener(
1562 taskFunctionOperationsListener
1564 this.sendToWorker(workerNodeKey
, message
)
1569 private setTasksQueuePriority (workerNodeKey
: number): void {
1570 this.workerNodes
[workerNodeKey
].setTasksQueuePriority(
1571 this.getTasksQueuePriority()
1575 private setTasksQueueSize (size
: number): void {
1576 for (const workerNode
of this.workerNodes
) {
1577 workerNode
.tasksQueueBackPressureSize
= size
1581 private setTasksStealingOnBackPressure (): void {
1582 for (const workerNodeKey
of this.workerNodes
.keys()) {
1583 this.workerNodes
[workerNodeKey
].on(
1585 this.handleWorkerNodeBackPressureEvent
1590 private setTaskStealing (): void {
1591 for (const workerNodeKey
of this.workerNodes
.keys()) {
1592 this.workerNodes
[workerNodeKey
].on('idle', this.handleWorkerNodeIdleEvent
)
1596 private shallExecuteTask (workerNodeKey
: number): boolean {
1598 this.tasksQueueSize(workerNodeKey
) === 0 &&
1599 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
<
1600 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1601 this.opts
.tasksQueueOptions
!.concurrency
!
1606 * Whether the worker node shall update its task function worker usage or not.
1607 * @param workerNodeKey - The worker node key.
1608 * @returns `true` if the worker node shall update its task function worker usage, `false` otherwise.
1610 private shallUpdateTaskFunctionWorkerUsage (workerNodeKey
: number): boolean {
1611 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1613 workerInfo
!= null &&
1614 Array.isArray(workerInfo
.taskFunctionsProperties
) &&
1615 workerInfo
.taskFunctionsProperties
.length
> 2
1620 * Starts the minimum number of workers.
1621 * @param initWorkerNodeUsage - Whether to initialize the worker node usage or not. @defaultValue false
1623 private startMinimumNumberOfWorkers (initWorkerNodeUsage
= false): void {
1624 if (this.minimumNumberOfWorkers
=== 0) {
1627 this.startingMinimumNumberOfWorkers
= true
1629 this.workerNodes
.reduce(
1630 (accumulator
, workerNode
) =>
1631 !workerNode
.info
.dynamic
? accumulator
+ 1 : accumulator
,
1633 ) < this.minimumNumberOfWorkers
1635 const workerNodeKey
= this.createAndSetupWorkerNode()
1636 initWorkerNodeUsage
&&
1637 this.initWorkerNodeUsage(this.workerNodes
[workerNodeKey
])
1639 this.startingMinimumNumberOfWorkers
= false
1642 private tasksQueueSize (workerNodeKey
: number): number {
1643 return this.workerNodes
[workerNodeKey
].tasksQueueSize()
1646 private unsetTasksStealingOnBackPressure (): void {
1647 for (const workerNodeKey
of this.workerNodes
.keys()) {
1648 this.workerNodes
[workerNodeKey
].off(
1650 this.handleWorkerNodeBackPressureEvent
1655 private unsetTaskStealing (): void {
1656 for (const workerNodeKey
of this.workerNodes
.keys()) {
1657 this.workerNodes
[workerNodeKey
].off(
1659 this.handleWorkerNodeIdleEvent
1664 private updateTaskSequentiallyStolenStatisticsWorkerUsage (
1665 workerNodeKey
: number,
1667 previousTaskName
?: string
1669 const workerNode
= this.workerNodes
[workerNodeKey
]
1670 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1671 if (workerNode
?.usage
!= null && taskName
!= null) {
1672 ++workerNode
.usage
.tasks
.sequentiallyStolen
1676 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1677 workerNode
.getTaskFunctionWorkerUsage(taskName
) != null
1679 const taskFunctionWorkerUsage
=
1680 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1681 workerNode
.getTaskFunctionWorkerUsage(taskName
)!
1683 taskFunctionWorkerUsage
.tasks
.sequentiallyStolen
=== 0 ||
1684 (previousTaskName
!= null &&
1685 previousTaskName
=== taskName
&&
1686 taskFunctionWorkerUsage
.tasks
.sequentiallyStolen
> 0)
1688 ++taskFunctionWorkerUsage
.tasks
.sequentiallyStolen
1689 } else if (taskFunctionWorkerUsage
.tasks
.sequentiallyStolen
> 0) {
1690 taskFunctionWorkerUsage
.tasks
.sequentiallyStolen
= 0
1695 private updateTaskStolenStatisticsWorkerUsage (
1696 workerNodeKey
: number,
1699 const workerNode
= this.workerNodes
[workerNodeKey
]
1700 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1701 if (workerNode
?.usage
!= null) {
1702 ++workerNode
.usage
.tasks
.stolen
1705 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1706 workerNode
.getTaskFunctionWorkerUsage(taskName
) != null
1708 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1709 ++workerNode
.getTaskFunctionWorkerUsage(taskName
)!.tasks
.stolen
1714 public async addTaskFunction (
1716 fn
: TaskFunction
<Data
, Response
> | TaskFunctionObject
<Data
, Response
>
1717 ): Promise
<boolean> {
1718 if (typeof name
!== 'string') {
1719 throw new TypeError('name argument must be a string')
1721 if (typeof name
=== 'string' && name
.trim().length
=== 0) {
1722 throw new TypeError('name argument must not be an empty string')
1724 if (typeof fn
=== 'function') {
1725 fn
= { taskFunction
: fn
} satisfies TaskFunctionObject
<Data
, Response
>
1727 if (typeof fn
.taskFunction
!== 'function') {
1728 throw new TypeError('taskFunction property must be a function')
1730 checkValidPriority(fn
.priority
)
1731 checkValidWorkerChoiceStrategy(fn
.strategy
)
1732 const opResult
= await this.sendTaskFunctionOperationToWorkers({
1733 taskFunction
: fn
.taskFunction
.toString(),
1734 taskFunctionOperation
: 'add',
1735 taskFunctionProperties
: buildTaskFunctionProperties(name
, fn
),
1737 this.taskFunctions
.set(name
, fn
)
1738 this.workerChoiceStrategiesContext
?.syncWorkerChoiceStrategies(
1739 this.getWorkerChoiceStrategies()
1741 for (const workerNodeKey
of this.workerNodes
.keys()) {
1742 this.sendStatisticsMessageToWorker(workerNodeKey
)
1748 public async destroy (): Promise
<void> {
1749 if (!this.started
) {
1750 throw new Error('Cannot destroy an already destroyed pool')
1752 if (this.starting
) {
1753 throw new Error('Cannot destroy an starting pool')
1755 if (this.destroying
) {
1756 throw new Error('Cannot destroy an already destroying pool')
1758 this.destroying
= true
1760 this.workerNodes
.map(async (_
, workerNodeKey
) => {
1761 await this.destroyWorkerNode(workerNodeKey
)
1764 if (this.emitter
!= null) {
1765 this.emitter
.emit(PoolEvents
.destroy
, this.info
)
1766 this.emitter
.emitDestroy()
1767 this.readyEventEmitted
= false
1769 delete this.startTimestamp
1770 this.destroying
= false
1771 this.started
= false
1775 public enableTasksQueue (
1777 tasksQueueOptions
?: TasksQueueOptions
1779 if (this.opts
.enableTasksQueue
=== true && !enable
) {
1780 this.unsetTaskStealing()
1781 this.unsetTasksStealingOnBackPressure()
1782 this.flushTasksQueues()
1784 this.opts
.enableTasksQueue
= enable
1785 this.setTasksQueueOptions(tasksQueueOptions
)
1789 public async execute (
1792 transferList
?: readonly TransferListItem
[]
1793 ): Promise
<Response
> {
1794 if (!this.started
) {
1795 throw new Error('Cannot execute a task on not started pool')
1797 if (this.destroying
) {
1798 throw new Error('Cannot execute a task on destroying pool')
1800 if (name
!= null && typeof name
!== 'string') {
1801 throw new TypeError('name argument must be a string')
1803 if (name
!= null && typeof name
=== 'string' && name
.trim().length
=== 0) {
1804 throw new TypeError('name argument must not be an empty string')
1806 if (transferList
!= null && !Array.isArray(transferList
)) {
1807 throw new TypeError('transferList argument must be an array')
1809 return await this.internalExecute(data
, name
, transferList
)
1813 public hasTaskFunction (name
: string): boolean {
1814 return this.listTaskFunctionsProperties().some(
1815 taskFunctionProperties
=> taskFunctionProperties
.name
=== name
1820 public listTaskFunctionsProperties (): TaskFunctionProperties
[] {
1821 for (const workerNode
of this.workerNodes
) {
1823 Array.isArray(workerNode
.info
.taskFunctionsProperties
) &&
1824 workerNode
.info
.taskFunctionsProperties
.length
> 0
1826 return workerNode
.info
.taskFunctionsProperties
1833 public async mapExecute (
1834 data
: Iterable
<Data
>,
1836 transferList
?: readonly TransferListItem
[]
1837 ): Promise
<Response
[]> {
1838 if (!this.started
) {
1839 throw new Error('Cannot execute task(s) on not started pool')
1841 if (this.destroying
) {
1842 throw new Error('Cannot execute task(s) on destroying pool')
1844 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1846 throw new TypeError('data argument must be a defined iterable')
1848 if (typeof data
[Symbol
.iterator
] !== 'function') {
1849 throw new TypeError('data argument must be an iterable')
1851 if (name
!= null && typeof name
!== 'string') {
1852 throw new TypeError('name argument must be a string')
1854 if (name
!= null && typeof name
=== 'string' && name
.trim().length
=== 0) {
1855 throw new TypeError('name argument must not be an empty string')
1857 if (transferList
!= null && !Array.isArray(transferList
)) {
1858 throw new TypeError('transferList argument must be an array')
1860 if (!Array.isArray(data
)) {
1863 return await Promise
.all(
1864 (data
as Data
[]).map(data
=>
1865 this.internalExecute(data
, name
, transferList
)
1871 public async removeTaskFunction (name
: string): Promise
<boolean> {
1872 if (!this.taskFunctions
.has(name
)) {
1874 'Cannot remove a task function not handled on the pool side'
1877 const opResult
= await this.sendTaskFunctionOperationToWorkers({
1878 taskFunctionOperation
: 'remove',
1879 taskFunctionProperties
: buildTaskFunctionProperties(
1881 this.taskFunctions
.get(name
)
1884 for (const workerNode
of this.workerNodes
) {
1885 workerNode
.deleteTaskFunctionWorkerUsage(name
)
1887 this.taskFunctions
.delete(name
)
1888 this.workerChoiceStrategiesContext
?.syncWorkerChoiceStrategies(
1889 this.getWorkerChoiceStrategies()
1891 for (const workerNodeKey
of this.workerNodes
.keys()) {
1892 this.sendStatisticsMessageToWorker(workerNodeKey
)
1898 public async setDefaultTaskFunction (name
: string): Promise
<boolean> {
1899 return await this.sendTaskFunctionOperationToWorkers({
1900 taskFunctionOperation
: 'default',
1901 taskFunctionProperties
: buildTaskFunctionProperties(
1903 this.taskFunctions
.get(name
)
1909 public setTasksQueueOptions (
1910 tasksQueueOptions
: TasksQueueOptions
| undefined
1912 if (this.opts
.enableTasksQueue
=== true) {
1913 checkValidTasksQueueOptions(tasksQueueOptions
)
1914 this.opts
.tasksQueueOptions
=
1915 this.buildTasksQueueOptions(tasksQueueOptions
)
1916 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1917 this.setTasksQueueSize(this.opts
.tasksQueueOptions
.size
!)
1918 if (this.opts
.tasksQueueOptions
.taskStealing
=== true) {
1919 this.unsetTaskStealing()
1920 this.setTaskStealing()
1922 this.unsetTaskStealing()
1924 if (this.opts
.tasksQueueOptions
.tasksStealingOnBackPressure
=== true) {
1925 this.unsetTasksStealingOnBackPressure()
1926 this.setTasksStealingOnBackPressure()
1928 this.unsetTasksStealingOnBackPressure()
1930 } else if (this.opts
.tasksQueueOptions
!= null) {
1931 delete this.opts
.tasksQueueOptions
1936 public setWorkerChoiceStrategy (
1937 workerChoiceStrategy
: WorkerChoiceStrategy
,
1938 workerChoiceStrategyOptions
?: WorkerChoiceStrategyOptions
1940 let requireSync
= false
1941 checkValidWorkerChoiceStrategy(workerChoiceStrategy
)
1942 if (workerChoiceStrategyOptions
!= null) {
1943 requireSync
= !this.setWorkerChoiceStrategyOptions(
1944 workerChoiceStrategyOptions
1947 if (workerChoiceStrategy
!== this.opts
.workerChoiceStrategy
) {
1948 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
1949 this.workerChoiceStrategiesContext
?.setDefaultWorkerChoiceStrategy(
1950 this.opts
.workerChoiceStrategy
,
1951 this.opts
.workerChoiceStrategyOptions
1956 this.workerChoiceStrategiesContext
?.syncWorkerChoiceStrategies(
1957 this.getWorkerChoiceStrategies(),
1958 this.opts
.workerChoiceStrategyOptions
1960 for (const workerNodeKey
of this.workerNodes
.keys()) {
1961 this.sendStatisticsMessageToWorker(workerNodeKey
)
1967 public setWorkerChoiceStrategyOptions (
1968 workerChoiceStrategyOptions
: undefined | WorkerChoiceStrategyOptions
1970 this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
1971 if (workerChoiceStrategyOptions
!= null) {
1972 this.opts
.workerChoiceStrategyOptions
= {
1973 ...this.opts
.workerChoiceStrategyOptions
,
1974 ...workerChoiceStrategyOptions
,
1976 this.workerChoiceStrategiesContext
?.setOptions(
1977 this.opts
.workerChoiceStrategyOptions
1979 this.workerChoiceStrategiesContext
?.syncWorkerChoiceStrategies(
1980 this.getWorkerChoiceStrategies(),
1981 this.opts
.workerChoiceStrategyOptions
1983 for (const workerNodeKey
of this.workerNodes
.keys()) {
1984 this.sendStatisticsMessageToWorker(workerNodeKey
)
1992 public start (): void {
1994 throw new Error('Cannot start an already started pool')
1996 if (this.starting
) {
1997 throw new Error('Cannot start an already starting pool')
1999 if (this.destroying
) {
2000 throw new Error('Cannot start a destroying pool')
2002 this.starting
= true
2003 this.startMinimumNumberOfWorkers()
2004 this.startTimestamp
= performance
.now()
2005 this.starting
= false
2010 * Whether the pool is back pressured or not.
2011 * @returns The pool back pressure boolean status.
2013 protected abstract get
backPressure (): boolean
2016 * Whether the pool is busy or not.
2017 * @returns The pool busyness boolean status.
2019 protected abstract get
busy (): boolean
2022 public get
info (): PoolInfo
{
2024 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2025 defaultStrategy
: this.opts
.workerChoiceStrategy
!,
2026 maxSize
: this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
,
2027 minSize
: this.minimumNumberOfWorkers
,
2029 started
: this.started
,
2030 strategyRetries
: this.workerChoiceStrategiesContext
?.retriesCount
?? 0,
2033 worker
: this.worker
,
2034 ...(this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
2035 .runTime
.aggregate
=== true &&
2036 this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
2037 .waitTime
.aggregate
&& {
2038 utilization
: round(this.utilization
),
2040 busyWorkerNodes
: this.workerNodes
.reduce(
2041 (accumulator
, _
, workerNodeKey
) =>
2042 this.isWorkerNodeBusy(workerNodeKey
) ? accumulator
+ 1 : accumulator
,
2045 executedTasks
: this.workerNodes
.reduce(
2046 (accumulator
, workerNode
) =>
2047 accumulator
+ workerNode
.usage
.tasks
.executed
,
2050 executingTasks
: this.workerNodes
.reduce(
2051 (accumulator
, workerNode
) =>
2052 accumulator
+ workerNode
.usage
.tasks
.executing
,
2055 failedTasks
: this.workerNodes
.reduce(
2056 (accumulator
, workerNode
) =>
2057 accumulator
+ workerNode
.usage
.tasks
.failed
,
2060 idleWorkerNodes
: this.workerNodes
.reduce(
2061 (accumulator
, _
, workerNodeKey
) =>
2062 this.isWorkerNodeIdle(workerNodeKey
) ? accumulator
+ 1 : accumulator
,
2065 workerNodes
: this.workerNodes
.length
,
2066 ...(this.type === PoolTypes
.dynamic
&& {
2067 dynamicWorkerNodes
: this.workerNodes
.reduce(
2068 (accumulator
, workerNode
) =>
2069 workerNode
.info
.dynamic
? accumulator
+ 1 : accumulator
,
2073 ...(this.opts
.enableTasksQueue
=== true && {
2074 backPressure
: this.backPressure
,
2075 backPressureWorkerNodes
: this.workerNodes
.reduce(
2076 (accumulator
, _
, workerNodeKey
) =>
2077 this.isWorkerNodeBackPressured(workerNodeKey
)
2082 maxQueuedTasks
: this.workerNodes
.reduce(
2083 (accumulator
, workerNode
) =>
2084 accumulator
+ (workerNode
.usage
.tasks
.maxQueued
?? 0),
2087 queuedTasks
: this.workerNodes
.reduce(
2088 (accumulator
, workerNode
) =>
2089 accumulator
+ workerNode
.usage
.tasks
.queued
,
2092 stealingWorkerNodes
: this.workerNodes
.reduce(
2093 (accumulator
, _
, workerNodeKey
) =>
2094 this.isWorkerNodeStealing(workerNodeKey
)
2099 stolenTasks
: this.workerNodes
.reduce(
2100 (accumulator
, workerNode
) =>
2101 accumulator
+ workerNode
.usage
.tasks
.stolen
,
2105 ...(this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
2106 .runTime
.aggregate
=== true && {
2110 ...this.workerNodes
.map(
2112 workerNode
.usage
.runTime
.maximum
?? Number.NEGATIVE_INFINITY
2118 ...this.workerNodes
.map(
2120 workerNode
.usage
.runTime
.minimum
?? Number.POSITIVE_INFINITY
2124 ...(this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
2125 .runTime
.average
&& {
2128 this.workerNodes
.reduce
<number[]>(
2129 (accumulator
, workerNode
) =>
2131 workerNode
.usage
.runTime
.history
.toArray()
2138 ...(this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
2139 .runTime
.median
&& {
2142 this.workerNodes
.reduce
<number[]>(
2143 (accumulator
, workerNode
) =>
2145 workerNode
.usage
.runTime
.history
.toArray()
2154 ...(this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
2155 .waitTime
.aggregate
=== true && {
2159 ...this.workerNodes
.map(
2161 workerNode
.usage
.waitTime
.maximum
?? Number.NEGATIVE_INFINITY
2167 ...this.workerNodes
.map(
2169 workerNode
.usage
.waitTime
.minimum
?? Number.POSITIVE_INFINITY
2173 ...(this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
2174 .waitTime
.average
&& {
2177 this.workerNodes
.reduce
<number[]>(
2178 (accumulator
, workerNode
) =>
2180 workerNode
.usage
.waitTime
.history
.toArray()
2187 ...(this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
2188 .waitTime
.median
&& {
2191 this.workerNodes
.reduce
<number[]>(
2192 (accumulator
, workerNode
) =>
2194 workerNode
.usage
.waitTime
.history
.toArray()
2203 ...(this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
2204 .elu
.aggregate
=== true && {
2209 ...this.workerNodes
.map(
2211 workerNode
.usage
.elu
.active
.maximum
??
2212 Number.NEGATIVE_INFINITY
2218 ...this.workerNodes
.map(
2220 workerNode
.usage
.elu
.active
.minimum
??
2221 Number.POSITIVE_INFINITY
2225 ...(this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
2229 this.workerNodes
.reduce
<number[]>(
2230 (accumulator
, workerNode
) =>
2232 workerNode
.usage
.elu
.active
.history
.toArray()
2239 ...(this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
2243 this.workerNodes
.reduce
<number[]>(
2244 (accumulator
, workerNode
) =>
2246 workerNode
.usage
.elu
.active
.history
.toArray()
2257 ...this.workerNodes
.map(
2259 workerNode
.usage
.elu
.idle
.maximum
??
2260 Number.NEGATIVE_INFINITY
2266 ...this.workerNodes
.map(
2268 workerNode
.usage
.elu
.idle
.minimum
??
2269 Number.POSITIVE_INFINITY
2273 ...(this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
2277 this.workerNodes
.reduce
<number[]>(
2278 (accumulator
, workerNode
) =>
2280 workerNode
.usage
.elu
.idle
.history
.toArray()
2287 ...(this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
2291 this.workerNodes
.reduce
<number[]>(
2292 (accumulator
, workerNode
) =>
2294 workerNode
.usage
.elu
.idle
.history
.toArray()
2305 this.workerNodes
.map(
2306 workerNode
=> workerNode
.usage
.elu
.utilization
?? 0
2312 this.workerNodes
.map(
2313 workerNode
=> workerNode
.usage
.elu
.utilization
?? 0
2324 * Whether the pool is ready or not.
2325 * @returns The pool readiness boolean status.
2327 private get
ready (): boolean {
2328 if (!this.started
) {
2332 this.workerNodes
.reduce(
2333 (accumulator
, workerNode
) =>
2334 !workerNode
.info
.dynamic
&& workerNode
.info
.ready
2338 ) >= this.minimumNumberOfWorkers
2345 * If it is `'dynamic'`, it provides the `max` property.
2347 protected abstract get
type (): PoolType
2350 * The approximate pool utilization.
2351 * @returns The pool utilization.
2353 private get
utilization (): number {
2354 if (this.startTimestamp
== null) {
2357 const poolTimeCapacity
=
2358 (performance
.now() - this.startTimestamp
) *
2359 (this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
)
2360 const totalTasksRunTime
= this.workerNodes
.reduce(
2361 (accumulator
, workerNode
) =>
2362 accumulator
+ (workerNode
.usage
.runTime
.aggregate
?? 0),
2365 const totalTasksWaitTime
= this.workerNodes
.reduce(
2366 (accumulator
, workerNode
) =>
2367 accumulator
+ (workerNode
.usage
.waitTime
.aggregate
?? 0),
2370 return (totalTasksRunTime
+ totalTasksWaitTime
) / poolTimeCapacity
2376 protected abstract get
worker (): WorkerType