1 import { AsyncResource
} from
'node:async_hooks'
2 import { randomUUID
} from
'node:crypto'
3 import { EventEmitterAsyncResource
} from
'node:events'
4 import { performance
} from
'node:perf_hooks'
5 import type { TransferListItem
} from
'node:worker_threads'
9 PromiseResponseWrapper
,
11 TaskFunctionProperties
12 } from
'../utility-types.js'
15 buildTaskFunctionProperties
,
30 } from
'../worker/task-functions.js'
31 import { KillBehaviors
} from
'../worker/worker-options.js'
39 type TasksQueueOptions
43 WorkerChoiceStrategies
,
44 type WorkerChoiceStrategy
,
45 type WorkerChoiceStrategyOptions
46 } from
'./selection-strategies/selection-strategies-types.js'
47 import { WorkerChoiceStrategiesContext
} from
'./selection-strategies/worker-choice-strategies-context.js'
51 checkValidTasksQueueOptions
,
52 checkValidWorkerChoiceStrategy
,
53 getDefaultTasksQueueOptions
,
55 updateRunTimeWorkerUsage
,
56 updateTaskStatisticsWorkerUsage
,
57 updateWaitTimeWorkerUsage
,
60 import { version
} from
'./version.js'
65 WorkerNodeEventDetail
,
68 import { WorkerNode
} from
'./worker-node.js'
71 * Base class that implements some shared logic for all poolifier pools.
73 * @typeParam Worker - Type of worker which manages this pool.
74 * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
75 * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
77 export abstract class AbstractPool
<
78 Worker
extends IWorker
,
81 > implements IPool
<Worker
, Data
, Response
> {
83 public readonly workerNodes
: Array<IWorkerNode
<Worker
, Data
>> = []
86 public emitter
?: EventEmitterAsyncResource
89 * The task execution response promise map:
90 * - `key`: The message id of each submitted task.
91 * - `value`: An object that contains task's worker node key, execution response promise resolve and reject callbacks, async resource.
93 * When we receive a message from the worker, we get a map entry with the promise resolve/reject bound to the message id.
95 protected promiseResponseMap
: Map
<
96 `${string}-${string}-${string}-${string}-${string}`,
97 PromiseResponseWrapper
<Response
>
99 `${string}-${string}-${string}-${string}-${string}`,
100 PromiseResponseWrapper
<Response
>
104 * Worker choice strategies context referencing worker choice algorithms implementation.
106 protected workerChoiceStrategiesContext
?: WorkerChoiceStrategiesContext
<
113 * The task functions added at runtime map:
114 * - `key`: The task function name.
115 * - `value`: The task function object.
117 private readonly taskFunctions
: Map
<
119 TaskFunctionObject
<Data
, Response
>
123 * Whether the pool is started or not.
125 private started
: boolean
127 * Whether the pool is starting or not.
129 private starting
: boolean
131 * Whether the pool is destroying or not.
133 private destroying
: boolean
135 * Whether the minimum number of workers is starting or not.
137 private startingMinimumNumberOfWorkers
: boolean
139 * Whether the pool ready event has been emitted or not.
141 private readyEventEmitted
: boolean
143 * The start timestamp of the pool.
145 private startTimestamp
?: number
148 * Constructs a new poolifier pool.
150 * @param minimumNumberOfWorkers - Minimum number of workers that this pool manages.
151 * @param filePath - Path to the worker file.
152 * @param opts - Options for the pool.
153 * @param maximumNumberOfWorkers - Maximum number of workers that this pool manages.
156 protected readonly minimumNumberOfWorkers
: number,
157 protected readonly filePath
: string,
158 protected readonly opts
: PoolOptions
<Worker
>,
159 protected readonly maximumNumberOfWorkers
?: number
161 if (!this.isMain()) {
163 'Cannot start a pool from a worker with the same type as the pool'
167 checkFilePath(this.filePath
)
168 this.checkMinimumNumberOfWorkers(this.minimumNumberOfWorkers
)
169 this.checkPoolOptions(this.opts
)
171 this.chooseWorkerNode
= this.chooseWorkerNode
.bind(this)
172 this.executeTask
= this.executeTask
.bind(this)
173 this.enqueueTask
= this.enqueueTask
.bind(this)
175 if (this.opts
.enableEvents
=== true) {
176 this.initEventEmitter()
178 this.workerChoiceStrategiesContext
= new WorkerChoiceStrategiesContext
<
184 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
185 [this.opts
.workerChoiceStrategy
!],
186 this.opts
.workerChoiceStrategyOptions
191 this.taskFunctions
= new Map
<string, TaskFunctionObject
<Data
, Response
>>()
194 this.starting
= false
195 this.destroying
= false
196 this.readyEventEmitted
= false
197 this.startingMinimumNumberOfWorkers
= false
198 if (this.opts
.startWorkers
=== true) {
203 private checkPoolType (): void {
204 if (this.type === PoolTypes
.fixed
&& this.maximumNumberOfWorkers
!= null) {
206 'Cannot instantiate a fixed pool with a maximum number of workers specified at initialization'
211 private checkMinimumNumberOfWorkers (
212 minimumNumberOfWorkers
: number | undefined
214 if (minimumNumberOfWorkers
== null) {
216 'Cannot instantiate a pool without specifying the number of workers'
218 } else if (!Number.isSafeInteger(minimumNumberOfWorkers
)) {
220 'Cannot instantiate a pool with a non safe integer number of workers'
222 } else if (minimumNumberOfWorkers
< 0) {
223 throw new RangeError(
224 'Cannot instantiate a pool with a negative number of workers'
226 } else if (this.type === PoolTypes
.fixed
&& minimumNumberOfWorkers
=== 0) {
227 throw new RangeError('Cannot instantiate a fixed pool with zero worker')
231 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
232 if (isPlainObject(opts
)) {
233 this.opts
.startWorkers
= opts
.startWorkers
?? true
234 checkValidWorkerChoiceStrategy(opts
.workerChoiceStrategy
)
235 this.opts
.workerChoiceStrategy
=
236 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
237 this.checkValidWorkerChoiceStrategyOptions(
238 opts
.workerChoiceStrategyOptions
240 if (opts
.workerChoiceStrategyOptions
!= null) {
241 this.opts
.workerChoiceStrategyOptions
= opts
.workerChoiceStrategyOptions
243 this.opts
.restartWorkerOnError
= opts
.restartWorkerOnError
?? true
244 this.opts
.enableEvents
= opts
.enableEvents
?? true
245 this.opts
.enableTasksQueue
= opts
.enableTasksQueue
?? false
246 if (this.opts
.enableTasksQueue
) {
247 checkValidTasksQueueOptions(opts
.tasksQueueOptions
)
248 this.opts
.tasksQueueOptions
= this.buildTasksQueueOptions(
249 opts
.tasksQueueOptions
253 throw new TypeError('Invalid pool options: must be a plain object')
257 private checkValidWorkerChoiceStrategyOptions (
258 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
| undefined
261 workerChoiceStrategyOptions
!= null &&
262 !isPlainObject(workerChoiceStrategyOptions
)
265 'Invalid worker choice strategy options: must be a plain object'
269 workerChoiceStrategyOptions
?.weights
!= null &&
270 Object.keys(workerChoiceStrategyOptions
.weights
).length
!==
271 (this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
)
274 'Invalid worker choice strategy options: must have a weight for each worker node'
278 workerChoiceStrategyOptions
?.measurement
!= null &&
279 !Object.values(Measurements
).includes(
280 workerChoiceStrategyOptions
.measurement
284 `Invalid worker choice strategy options: invalid measurement '${workerChoiceStrategyOptions.measurement}'`
289 private initEventEmitter (): void {
290 this.emitter
= new EventEmitterAsyncResource({
291 name
: `poolifier:${this.type}-${this.worker}-pool`
296 public get
info (): PoolInfo
{
301 started
: this.started
,
303 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
304 defaultStrategy
: this.opts
.workerChoiceStrategy
!,
305 strategyRetries
: this.workerChoiceStrategiesContext
?.retriesCount
?? 0,
306 minSize
: this.minimumNumberOfWorkers
,
307 maxSize
: this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
,
308 ...(this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
309 .runTime
.aggregate
=== true &&
310 this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
311 .waitTime
.aggregate
&& {
312 utilization
: round(this.utilization
)
314 workerNodes
: this.workerNodes
.length
,
315 idleWorkerNodes
: this.workerNodes
.reduce(
316 (accumulator
, workerNode
) =>
317 workerNode
.usage
.tasks
.executing
=== 0
322 ...(this.opts
.enableTasksQueue
=== true && {
323 stealingWorkerNodes
: this.workerNodes
.reduce(
324 (accumulator
, workerNode
) =>
325 workerNode
.info
.stealing
? accumulator
+ 1 : accumulator
,
329 busyWorkerNodes
: this.workerNodes
.reduce(
330 (accumulator
, _
, workerNodeKey
) =>
331 this.isWorkerNodeBusy(workerNodeKey
) ? accumulator
+ 1 : accumulator
,
334 executedTasks
: this.workerNodes
.reduce(
335 (accumulator
, workerNode
) =>
336 accumulator
+ workerNode
.usage
.tasks
.executed
,
339 executingTasks
: this.workerNodes
.reduce(
340 (accumulator
, workerNode
) =>
341 accumulator
+ workerNode
.usage
.tasks
.executing
,
344 ...(this.opts
.enableTasksQueue
=== true && {
345 queuedTasks
: this.workerNodes
.reduce(
346 (accumulator
, workerNode
) =>
347 accumulator
+ workerNode
.usage
.tasks
.queued
,
351 ...(this.opts
.enableTasksQueue
=== true && {
352 maxQueuedTasks
: this.workerNodes
.reduce(
353 (accumulator
, workerNode
) =>
354 accumulator
+ (workerNode
.usage
.tasks
.maxQueued
?? 0),
358 ...(this.opts
.enableTasksQueue
=== true && {
359 backPressure
: this.hasBackPressure()
361 ...(this.opts
.enableTasksQueue
=== true && {
362 stolenTasks
: this.workerNodes
.reduce(
363 (accumulator
, workerNode
) =>
364 accumulator
+ workerNode
.usage
.tasks
.stolen
,
368 failedTasks
: this.workerNodes
.reduce(
369 (accumulator
, workerNode
) =>
370 accumulator
+ workerNode
.usage
.tasks
.failed
,
373 ...(this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
374 .runTime
.aggregate
=== true && {
378 ...this.workerNodes
.map(
379 workerNode
=> workerNode
.usage
.runTime
.minimum
?? Infinity
385 ...this.workerNodes
.map(
386 workerNode
=> workerNode
.usage
.runTime
.maximum
?? -Infinity
390 ...(this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
391 .runTime
.average
&& {
394 this.workerNodes
.reduce
<number[]>(
395 (accumulator
, workerNode
) =>
396 accumulator
.concat(workerNode
.usage
.runTime
.history
),
402 ...(this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
406 this.workerNodes
.reduce
<number[]>(
407 (accumulator
, workerNode
) =>
408 accumulator
.concat(workerNode
.usage
.runTime
.history
),
416 ...(this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
417 .waitTime
.aggregate
=== true && {
421 ...this.workerNodes
.map(
422 workerNode
=> workerNode
.usage
.waitTime
.minimum
?? Infinity
428 ...this.workerNodes
.map(
429 workerNode
=> workerNode
.usage
.waitTime
.maximum
?? -Infinity
433 ...(this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
434 .waitTime
.average
&& {
437 this.workerNodes
.reduce
<number[]>(
438 (accumulator
, workerNode
) =>
439 accumulator
.concat(workerNode
.usage
.waitTime
.history
),
445 ...(this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
446 .waitTime
.median
&& {
449 this.workerNodes
.reduce
<number[]>(
450 (accumulator
, workerNode
) =>
451 accumulator
.concat(workerNode
.usage
.waitTime
.history
),
459 ...(this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
460 .elu
.aggregate
=== true && {
465 ...this.workerNodes
.map(
466 workerNode
=> workerNode
.usage
.elu
.idle
.minimum
?? Infinity
472 ...this.workerNodes
.map(
473 workerNode
=> workerNode
.usage
.elu
.idle
.maximum
?? -Infinity
477 ...(this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
481 this.workerNodes
.reduce
<number[]>(
482 (accumulator
, workerNode
) =>
483 accumulator
.concat(workerNode
.usage
.elu
.idle
.history
),
489 ...(this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
493 this.workerNodes
.reduce
<number[]>(
494 (accumulator
, workerNode
) =>
495 accumulator
.concat(workerNode
.usage
.elu
.idle
.history
),
505 ...this.workerNodes
.map(
506 workerNode
=> workerNode
.usage
.elu
.active
.minimum
?? Infinity
512 ...this.workerNodes
.map(
513 workerNode
=> workerNode
.usage
.elu
.active
.maximum
?? -Infinity
517 ...(this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
521 this.workerNodes
.reduce
<number[]>(
522 (accumulator
, workerNode
) =>
523 accumulator
.concat(workerNode
.usage
.elu
.active
.history
),
529 ...(this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
533 this.workerNodes
.reduce
<number[]>(
534 (accumulator
, workerNode
) =>
535 accumulator
.concat(workerNode
.usage
.elu
.active
.history
),
545 this.workerNodes
.map(
546 workerNode
=> workerNode
.usage
.elu
.utilization
?? 0
552 this.workerNodes
.map(
553 workerNode
=> workerNode
.usage
.elu
.utilization
?? 0
564 * The pool readiness boolean status.
566 private get
ready (): boolean {
571 this.workerNodes
.reduce(
572 (accumulator
, workerNode
) =>
573 !workerNode
.info
.dynamic
&& workerNode
.info
.ready
577 ) >= this.minimumNumberOfWorkers
582 * The pool emptiness boolean status.
584 protected get
empty (): boolean {
585 return this.minimumNumberOfWorkers
=== 0 && this.workerNodes
.length
=== 0
589 * The approximate pool utilization.
591 * @returns The pool utilization.
593 private get
utilization (): number {
594 if (this.startTimestamp
== null) {
597 const poolTimeCapacity
=
598 (performance
.now() - this.startTimestamp
) *
599 (this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
)
600 const totalTasksRunTime
= this.workerNodes
.reduce(
601 (accumulator
, workerNode
) =>
602 accumulator
+ (workerNode
.usage
.runTime
.aggregate
?? 0),
605 const totalTasksWaitTime
= this.workerNodes
.reduce(
606 (accumulator
, workerNode
) =>
607 accumulator
+ (workerNode
.usage
.waitTime
.aggregate
?? 0),
610 return (totalTasksRunTime
+ totalTasksWaitTime
) / poolTimeCapacity
616 * If it is `'dynamic'`, it provides the `max` property.
618 protected abstract get
type (): PoolType
623 protected abstract get
worker (): WorkerType
626 * Checks if the worker id sent in the received message from a worker is valid.
628 * @param message - The received message.
629 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the worker id is invalid.
631 private checkMessageWorkerId (message
: MessageValue
<Data
| Response
>): void {
632 if (message
.workerId
== null) {
633 throw new Error('Worker message received without worker id')
634 } else if (this.getWorkerNodeKeyByWorkerId(message
.workerId
) === -1) {
636 `Worker message received from unknown worker '${message.workerId}'`
642 * Gets the worker node key given its worker id.
644 * @param workerId - The worker id.
645 * @returns The worker node key if the worker id is found in the pool worker nodes, `-1` otherwise.
647 private getWorkerNodeKeyByWorkerId (workerId
: number | undefined): number {
648 return this.workerNodes
.findIndex(
649 workerNode
=> workerNode
.info
.id
=== workerId
654 public setWorkerChoiceStrategy (
655 workerChoiceStrategy
: WorkerChoiceStrategy
,
656 workerChoiceStrategyOptions
?: WorkerChoiceStrategyOptions
658 let requireSync
= false
659 checkValidWorkerChoiceStrategy(workerChoiceStrategy
)
660 if (workerChoiceStrategyOptions
!= null) {
661 requireSync
= !this.setWorkerChoiceStrategyOptions(
662 workerChoiceStrategyOptions
665 if (workerChoiceStrategy
!== this.opts
.workerChoiceStrategy
) {
666 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
667 this.workerChoiceStrategiesContext
?.setDefaultWorkerChoiceStrategy(
668 this.opts
.workerChoiceStrategy
,
669 this.opts
.workerChoiceStrategyOptions
674 this.workerChoiceStrategiesContext
?.syncWorkerChoiceStrategies(
675 this.getWorkerChoiceStrategies(),
676 this.opts
.workerChoiceStrategyOptions
678 for (const workerNodeKey
of this.workerNodes
.keys()) {
679 this.sendStatisticsMessageToWorker(workerNodeKey
)
685 public setWorkerChoiceStrategyOptions (
686 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
| undefined
688 this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
689 if (workerChoiceStrategyOptions
!= null) {
690 this.opts
.workerChoiceStrategyOptions
= workerChoiceStrategyOptions
691 this.workerChoiceStrategiesContext
?.setOptions(
692 this.opts
.workerChoiceStrategyOptions
694 this.workerChoiceStrategiesContext
?.syncWorkerChoiceStrategies(
695 this.getWorkerChoiceStrategies(),
696 this.opts
.workerChoiceStrategyOptions
698 for (const workerNodeKey
of this.workerNodes
.keys()) {
699 this.sendStatisticsMessageToWorker(workerNodeKey
)
707 public enableTasksQueue (
709 tasksQueueOptions
?: TasksQueueOptions
711 if (this.opts
.enableTasksQueue
=== true && !enable
) {
712 this.unsetTaskStealing()
713 this.unsetTasksStealingOnBackPressure()
714 this.flushTasksQueues()
716 this.opts
.enableTasksQueue
= enable
717 this.setTasksQueueOptions(tasksQueueOptions
)
721 public setTasksQueueOptions (
722 tasksQueueOptions
: TasksQueueOptions
| undefined
724 if (this.opts
.enableTasksQueue
=== true) {
725 checkValidTasksQueueOptions(tasksQueueOptions
)
726 this.opts
.tasksQueueOptions
=
727 this.buildTasksQueueOptions(tasksQueueOptions
)
728 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
729 this.setTasksQueueSize(this.opts
.tasksQueueOptions
.size
!)
730 if (this.opts
.tasksQueueOptions
.taskStealing
=== true) {
731 this.unsetTaskStealing()
732 this.setTaskStealing()
734 this.unsetTaskStealing()
736 if (this.opts
.tasksQueueOptions
.tasksStealingOnBackPressure
=== true) {
737 this.unsetTasksStealingOnBackPressure()
738 this.setTasksStealingOnBackPressure()
740 this.unsetTasksStealingOnBackPressure()
742 } else if (this.opts
.tasksQueueOptions
!= null) {
743 delete this.opts
.tasksQueueOptions
747 private buildTasksQueueOptions (
748 tasksQueueOptions
: TasksQueueOptions
| undefined
749 ): TasksQueueOptions
{
751 ...getDefaultTasksQueueOptions(
752 this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
758 private setTasksQueueSize (size
: number): void {
759 for (const workerNode
of this.workerNodes
) {
760 workerNode
.tasksQueueBackPressureSize
= size
764 private setTaskStealing (): void {
765 for (const workerNodeKey
of this.workerNodes
.keys()) {
766 this.workerNodes
[workerNodeKey
].on('idle', this.handleWorkerNodeIdleEvent
)
770 private unsetTaskStealing (): void {
771 for (const workerNodeKey
of this.workerNodes
.keys()) {
772 this.workerNodes
[workerNodeKey
].off(
774 this.handleWorkerNodeIdleEvent
779 private setTasksStealingOnBackPressure (): void {
780 for (const workerNodeKey
of this.workerNodes
.keys()) {
781 this.workerNodes
[workerNodeKey
].on(
783 this.handleWorkerNodeBackPressureEvent
788 private unsetTasksStealingOnBackPressure (): void {
789 for (const workerNodeKey
of this.workerNodes
.keys()) {
790 this.workerNodes
[workerNodeKey
].off(
792 this.handleWorkerNodeBackPressureEvent
798 * Whether the pool is full or not.
800 * The pool filling boolean status.
802 protected get
full (): boolean {
804 this.workerNodes
.length
>=
805 (this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
)
810 * Whether the pool is busy or not.
812 * The pool busyness boolean status.
814 protected abstract get
busy (): boolean
817 * Whether worker nodes are executing concurrently their tasks quota or not.
819 * @returns Worker nodes busyness boolean status.
821 protected internalBusy (): boolean {
822 if (this.opts
.enableTasksQueue
=== true) {
824 this.workerNodes
.findIndex(
826 workerNode
.info
.ready
&&
827 workerNode
.usage
.tasks
.executing
<
828 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
829 this.opts
.tasksQueueOptions
!.concurrency
!
834 this.workerNodes
.findIndex(
836 workerNode
.info
.ready
&& workerNode
.usage
.tasks
.executing
=== 0
841 private isWorkerNodeBusy (workerNodeKey
: number): boolean {
842 if (this.opts
.enableTasksQueue
=== true) {
844 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
>=
845 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
846 this.opts
.tasksQueueOptions
!.concurrency
!
849 return this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
> 0
852 private async sendTaskFunctionOperationToWorker (
853 workerNodeKey
: number,
854 message
: MessageValue
<Data
>
855 ): Promise
<boolean> {
856 return await new Promise
<boolean>((resolve
, reject
) => {
857 const taskFunctionOperationListener
= (
858 message
: MessageValue
<Response
>
860 this.checkMessageWorkerId(message
)
861 const workerId
= this.getWorkerInfo(workerNodeKey
)?.id
863 message
.taskFunctionOperationStatus
!= null &&
864 message
.workerId
=== workerId
866 if (message
.taskFunctionOperationStatus
) {
871 `Task function operation '${message.taskFunctionOperation}' failed on worker ${message.workerId} with error: '${message.workerError?.message}'`
875 this.deregisterWorkerMessageListener(
876 this.getWorkerNodeKeyByWorkerId(message
.workerId
),
877 taskFunctionOperationListener
881 this.registerWorkerMessageListener(
883 taskFunctionOperationListener
885 this.sendToWorker(workerNodeKey
, message
)
889 private async sendTaskFunctionOperationToWorkers (
890 message
: MessageValue
<Data
>
891 ): Promise
<boolean> {
892 return await new Promise
<boolean>((resolve
, reject
) => {
893 const responsesReceived
= new Array<MessageValue
<Response
>>()
894 const taskFunctionOperationsListener
= (
895 message
: MessageValue
<Response
>
897 this.checkMessageWorkerId(message
)
898 if (message
.taskFunctionOperationStatus
!= null) {
899 responsesReceived
.push(message
)
900 if (responsesReceived
.length
=== this.workerNodes
.length
) {
902 responsesReceived
.every(
903 message
=> message
.taskFunctionOperationStatus
=== true
908 responsesReceived
.some(
909 message
=> message
.taskFunctionOperationStatus
=== false
912 const errorResponse
= responsesReceived
.find(
913 response
=> response
.taskFunctionOperationStatus
=== false
917 `Task function operation '${
918 message.taskFunctionOperation as string
919 }' failed on worker ${errorResponse?.workerId} with error: '${
920 errorResponse?.workerError?.message
925 this.deregisterWorkerMessageListener(
926 this.getWorkerNodeKeyByWorkerId(message
.workerId
),
927 taskFunctionOperationsListener
932 for (const workerNodeKey
of this.workerNodes
.keys()) {
933 this.registerWorkerMessageListener(
935 taskFunctionOperationsListener
937 this.sendToWorker(workerNodeKey
, message
)
943 public hasTaskFunction (name
: string): boolean {
944 return this.listTaskFunctionsProperties().some(
945 taskFunctionProperties
=> taskFunctionProperties
.name
=== name
950 public async addTaskFunction (
952 fn
: TaskFunction
<Data
, Response
> | TaskFunctionObject
<Data
, Response
>
953 ): Promise
<boolean> {
954 if (typeof name
!== 'string') {
955 throw new TypeError('name argument must be a string')
957 if (typeof name
=== 'string' && name
.trim().length
=== 0) {
958 throw new TypeError('name argument must not be an empty string')
960 if (typeof fn
=== 'function') {
961 fn
= { taskFunction
: fn
} satisfies TaskFunctionObject
<Data
, Response
>
963 if (typeof fn
.taskFunction
!== 'function') {
964 throw new TypeError('taskFunction property must be a function')
966 checkValidPriority(fn
.priority
)
967 checkValidWorkerChoiceStrategy(fn
.strategy
)
968 const opResult
= await this.sendTaskFunctionOperationToWorkers({
969 taskFunctionOperation
: 'add',
970 taskFunctionProperties
: buildTaskFunctionProperties(name
, fn
),
971 taskFunction
: fn
.taskFunction
.toString()
973 this.taskFunctions
.set(name
, fn
)
974 this.workerChoiceStrategiesContext
?.syncWorkerChoiceStrategies(
975 this.getWorkerChoiceStrategies()
977 for (const workerNodeKey
of this.workerNodes
.keys()) {
978 this.sendStatisticsMessageToWorker(workerNodeKey
)
984 public async removeTaskFunction (name
: string): Promise
<boolean> {
985 if (!this.taskFunctions
.has(name
)) {
987 'Cannot remove a task function not handled on the pool side'
990 const opResult
= await this.sendTaskFunctionOperationToWorkers({
991 taskFunctionOperation
: 'remove',
992 taskFunctionProperties
: buildTaskFunctionProperties(
994 this.taskFunctions
.get(name
)
997 for (const workerNode
of this.workerNodes
) {
998 workerNode
.deleteTaskFunctionWorkerUsage(name
)
1000 this.taskFunctions
.delete(name
)
1001 this.workerChoiceStrategiesContext
?.syncWorkerChoiceStrategies(
1002 this.getWorkerChoiceStrategies()
1004 for (const workerNodeKey
of this.workerNodes
.keys()) {
1005 this.sendStatisticsMessageToWorker(workerNodeKey
)
1011 public listTaskFunctionsProperties (): TaskFunctionProperties
[] {
1012 for (const workerNode
of this.workerNodes
) {
1014 Array.isArray(workerNode
.info
.taskFunctionsProperties
) &&
1015 workerNode
.info
.taskFunctionsProperties
.length
> 0
1017 return workerNode
.info
.taskFunctionsProperties
1024 * Gets task function worker choice strategy, if any.
1026 * @param name - The task function name.
1027 * @returns The task function worker choice strategy if the task function worker choice strategy is defined, `undefined` otherwise.
1029 private readonly getTaskFunctionWorkerChoiceStrategy
= (
1031 ): WorkerChoiceStrategy
| undefined => {
1032 name
= name
?? DEFAULT_TASK_NAME
1033 const taskFunctionsProperties
= this.listTaskFunctionsProperties()
1034 if (name
=== DEFAULT_TASK_NAME
) {
1035 name
= taskFunctionsProperties
[1]?.name
1037 return taskFunctionsProperties
.find(
1038 (taskFunctionProperties
: TaskFunctionProperties
) =>
1039 taskFunctionProperties
.name
=== name
1044 * Gets worker node task function worker choice strategy, if any.
1046 * @param workerNodeKey - The worker node key.
1047 * @param name - The task function name.
1048 * @returns The worker node task function worker choice strategy if the worker node task function worker choice strategy is defined, `undefined` otherwise.
1050 private readonly getWorkerNodeTaskFunctionWorkerChoiceStrategy
= (
1051 workerNodeKey
: number,
1053 ): WorkerChoiceStrategy
| undefined => {
1054 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1055 if (workerInfo
== null) {
1058 name
= name
?? DEFAULT_TASK_NAME
1059 if (name
=== DEFAULT_TASK_NAME
) {
1060 name
= workerInfo
.taskFunctionsProperties
?.[1]?.name
1062 return workerInfo
.taskFunctionsProperties
?.find(
1063 (taskFunctionProperties
: TaskFunctionProperties
) =>
1064 taskFunctionProperties
.name
=== name
1069 * Gets worker node task function priority, if any.
1071 * @param workerNodeKey - The worker node key.
1072 * @param name - The task function name.
1073 * @returns The worker node task function priority if the worker node task function priority is defined, `undefined` otherwise.
1075 private readonly getWorkerNodeTaskFunctionPriority
= (
1076 workerNodeKey
: number,
1078 ): number | undefined => {
1079 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1080 if (workerInfo
== null) {
1083 name
= name
?? DEFAULT_TASK_NAME
1084 if (name
=== DEFAULT_TASK_NAME
) {
1085 name
= workerInfo
.taskFunctionsProperties
?.[1]?.name
1087 return workerInfo
.taskFunctionsProperties
?.find(
1088 (taskFunctionProperties
: TaskFunctionProperties
) =>
1089 taskFunctionProperties
.name
=== name
1094 * Gets the worker choice strategies registered in this pool.
1096 * @returns The worker choice strategies.
1098 private readonly getWorkerChoiceStrategies
=
1099 (): Set
<WorkerChoiceStrategy
> => {
1101 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1102 this.opts
.workerChoiceStrategy
!,
1103 ...(this.listTaskFunctionsProperties()
1105 (taskFunctionProperties
: TaskFunctionProperties
) =>
1106 taskFunctionProperties
.strategy
1109 (strategy
: WorkerChoiceStrategy
| undefined) => strategy
!= null
1110 ) as WorkerChoiceStrategy
[])
1115 public async setDefaultTaskFunction (name
: string): Promise
<boolean> {
1116 return await this.sendTaskFunctionOperationToWorkers({
1117 taskFunctionOperation
: 'default',
1118 taskFunctionProperties
: buildTaskFunctionProperties(
1120 this.taskFunctions
.get(name
)
1125 private shallExecuteTask (workerNodeKey
: number): boolean {
1127 this.tasksQueueSize(workerNodeKey
) === 0 &&
1128 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
<
1129 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1130 this.opts
.tasksQueueOptions
!.concurrency
!
1135 public async execute (
1138 transferList
?: readonly TransferListItem
[]
1139 ): Promise
<Response
> {
1140 return await new Promise
<Response
>((resolve
, reject
) => {
1141 if (!this.started
) {
1142 reject(new Error('Cannot execute a task on not started pool'))
1145 if (this.destroying
) {
1146 reject(new Error('Cannot execute a task on destroying pool'))
1149 if (name
!= null && typeof name
!== 'string') {
1150 reject(new TypeError('name argument must be a string'))
1155 typeof name
=== 'string' &&
1156 name
.trim().length
=== 0
1158 reject(new TypeError('name argument must not be an empty string'))
1161 if (transferList
!= null && !Array.isArray(transferList
)) {
1162 reject(new TypeError('transferList argument must be an array'))
1165 const timestamp
= performance
.now()
1166 const workerNodeKey
= this.chooseWorkerNode(name
)
1167 const task
: Task
<Data
> = {
1168 name
: name
?? DEFAULT_TASK_NAME
,
1169 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
1170 data
: data
?? ({} as Data
),
1171 priority
: this.getWorkerNodeTaskFunctionPriority(workerNodeKey
, name
),
1172 strategy
: this.getWorkerNodeTaskFunctionWorkerChoiceStrategy(
1178 taskId
: randomUUID()
1180 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1181 this.promiseResponseMap
.set(task
.taskId
!, {
1185 ...(this.emitter
!= null && {
1186 asyncResource
: new AsyncResource('poolifier:task', {
1187 triggerAsyncId
: this.emitter
.asyncId
,
1188 requireManualDestroy
: true
1193 this.opts
.enableTasksQueue
=== false ||
1194 (this.opts
.enableTasksQueue
=== true &&
1195 this.shallExecuteTask(workerNodeKey
))
1197 this.executeTask(workerNodeKey
, task
)
1199 this.enqueueTask(workerNodeKey
, task
)
1205 * Starts the minimum number of workers.
1207 private startMinimumNumberOfWorkers (initWorkerNodeUsage
= false): void {
1208 this.startingMinimumNumberOfWorkers
= true
1210 this.workerNodes
.reduce(
1211 (accumulator
, workerNode
) =>
1212 !workerNode
.info
.dynamic
? accumulator
+ 1 : accumulator
,
1214 ) < this.minimumNumberOfWorkers
1216 const workerNodeKey
= this.createAndSetupWorkerNode()
1217 initWorkerNodeUsage
&&
1218 this.initWorkerNodeUsage(this.workerNodes
[workerNodeKey
])
1220 this.startingMinimumNumberOfWorkers
= false
1224 public start (): void {
1226 throw new Error('Cannot start an already started pool')
1228 if (this.starting
) {
1229 throw new Error('Cannot start an already starting pool')
1231 if (this.destroying
) {
1232 throw new Error('Cannot start a destroying pool')
1234 this.starting
= true
1235 this.startMinimumNumberOfWorkers()
1236 this.startTimestamp
= performance
.now()
1237 this.starting
= false
1242 public async destroy (): Promise
<void> {
1243 if (!this.started
) {
1244 throw new Error('Cannot destroy an already destroyed pool')
1246 if (this.starting
) {
1247 throw new Error('Cannot destroy an starting pool')
1249 if (this.destroying
) {
1250 throw new Error('Cannot destroy an already destroying pool')
1252 this.destroying
= true
1254 this.workerNodes
.map(async (_
, workerNodeKey
) => {
1255 await this.destroyWorkerNode(workerNodeKey
)
1258 this.emitter
?.emit(PoolEvents
.destroy
, this.info
)
1259 this.emitter
?.emitDestroy()
1260 this.readyEventEmitted
= false
1261 delete this.startTimestamp
1262 this.destroying
= false
1263 this.started
= false
1266 private async sendKillMessageToWorker (workerNodeKey
: number): Promise
<void> {
1267 await new Promise
<void>((resolve
, reject
) => {
1268 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1269 if (this.workerNodes
[workerNodeKey
] == null) {
1273 const killMessageListener
= (message
: MessageValue
<Response
>): void => {
1274 this.checkMessageWorkerId(message
)
1275 if (message
.kill
=== 'success') {
1277 } else if (message
.kill
=== 'failure') {
1280 `Kill message handling failed on worker ${message.workerId}`
1285 // FIXME: should be registered only once
1286 this.registerWorkerMessageListener(workerNodeKey
, killMessageListener
)
1287 this.sendToWorker(workerNodeKey
, { kill
: true })
1292 * Terminates the worker node given its worker node key.
1294 * @param workerNodeKey - The worker node key.
1296 protected async destroyWorkerNode (workerNodeKey
: number): Promise
<void> {
1297 this.flagWorkerNodeAsNotReady(workerNodeKey
)
1298 const flushedTasks
= this.flushTasksQueue(workerNodeKey
)
1299 const workerNode
= this.workerNodes
[workerNodeKey
]
1300 await waitWorkerNodeEvents(
1304 this.opts
.tasksQueueOptions
?.tasksFinishedTimeout
??
1305 getDefaultTasksQueueOptions(
1306 this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
1307 ).tasksFinishedTimeout
1309 await this.sendKillMessageToWorker(workerNodeKey
)
1310 await workerNode
.terminate()
1314 * Setup hook to execute code before worker nodes are created in the abstract constructor.
1315 * Can be overridden.
1319 protected setupHook (): void {
1320 /* Intentionally empty */
1324 * Returns whether the worker is the main worker or not.
1326 * @returns `true` if the worker is the main worker, `false` otherwise.
1328 protected abstract isMain (): boolean
1331 * Hook executed before the worker task execution.
1332 * Can be overridden.
1334 * @param workerNodeKey - The worker node key.
1335 * @param task - The task to execute.
1337 protected beforeTaskExecutionHook (
1338 workerNodeKey
: number,
1341 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1342 if (this.workerNodes
[workerNodeKey
]?.usage
!= null) {
1343 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
1344 ++workerUsage
.tasks
.executing
1345 updateWaitTimeWorkerUsage(
1346 this.workerChoiceStrategiesContext
,
1352 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1353 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1354 this.workerNodes
[workerNodeKey
].getTaskFunctionWorkerUsage(task
.name
!) !=
1357 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1358 const taskFunctionWorkerUsage
= this.workerNodes
[
1360 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1361 ].getTaskFunctionWorkerUsage(task
.name
!)!
1362 ++taskFunctionWorkerUsage
.tasks
.executing
1363 updateWaitTimeWorkerUsage(
1364 this.workerChoiceStrategiesContext
,
1365 taskFunctionWorkerUsage
,
1372 * Hook executed after the worker task execution.
1373 * Can be overridden.
1375 * @param workerNodeKey - The worker node key.
1376 * @param message - The received message.
1378 protected afterTaskExecutionHook (
1379 workerNodeKey
: number,
1380 message
: MessageValue
<Response
>
1382 let needWorkerChoiceStrategiesUpdate
= false
1383 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1384 if (this.workerNodes
[workerNodeKey
]?.usage
!= null) {
1385 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
1386 updateTaskStatisticsWorkerUsage(workerUsage
, message
)
1387 updateRunTimeWorkerUsage(
1388 this.workerChoiceStrategiesContext
,
1392 updateEluWorkerUsage(
1393 this.workerChoiceStrategiesContext
,
1397 needWorkerChoiceStrategiesUpdate
= true
1400 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1401 this.workerNodes
[workerNodeKey
].getTaskFunctionWorkerUsage(
1402 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1403 message
.taskPerformance
!.name
1406 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1407 const taskFunctionWorkerUsage
= this.workerNodes
[
1409 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1410 ].getTaskFunctionWorkerUsage(message
.taskPerformance
!.name
)!
1411 updateTaskStatisticsWorkerUsage(taskFunctionWorkerUsage
, message
)
1412 updateRunTimeWorkerUsage(
1413 this.workerChoiceStrategiesContext
,
1414 taskFunctionWorkerUsage
,
1417 updateEluWorkerUsage(
1418 this.workerChoiceStrategiesContext
,
1419 taskFunctionWorkerUsage
,
1422 needWorkerChoiceStrategiesUpdate
= true
1424 if (needWorkerChoiceStrategiesUpdate
) {
1425 this.workerChoiceStrategiesContext
?.update(workerNodeKey
)
1430 * Whether the worker node shall update its task function worker usage or not.
1432 * @param workerNodeKey - The worker node key.
1433 * @returns `true` if the worker node shall update its task function worker usage, `false` otherwise.
1435 private shallUpdateTaskFunctionWorkerUsage (workerNodeKey
: number): boolean {
1436 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1438 workerInfo
!= null &&
1439 Array.isArray(workerInfo
.taskFunctionsProperties
) &&
1440 workerInfo
.taskFunctionsProperties
.length
> 2
1445 * Chooses a worker node for the next task.
1447 * @param name - The task function name.
1448 * @returns The chosen worker node key
1450 private chooseWorkerNode (name
?: string): number {
1451 if (this.shallCreateDynamicWorker()) {
1452 const workerNodeKey
= this.createAndSetupDynamicWorkerNode()
1454 this.workerChoiceStrategiesContext
?.getPolicy().dynamicWorkerUsage
===
1457 return workerNodeKey
1460 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1461 return this.workerChoiceStrategiesContext
!.execute(
1462 this.getTaskFunctionWorkerChoiceStrategy(name
)
1467 * Conditions for dynamic worker creation.
1469 * @returns Whether to create a dynamic worker or not.
1471 protected abstract shallCreateDynamicWorker (): boolean
1474 * Sends a message to worker given its worker node key.
1476 * @param workerNodeKey - The worker node key.
1477 * @param message - The message.
1478 * @param transferList - The optional array of transferable objects.
1480 protected abstract sendToWorker (
1481 workerNodeKey
: number,
1482 message
: MessageValue
<Data
>,
1483 transferList
?: readonly TransferListItem
[]
1487 * Initializes the worker node usage with sensible default values gathered during runtime.
1489 * @param workerNode - The worker node.
1491 private initWorkerNodeUsage (workerNode
: IWorkerNode
<Worker
, Data
>): void {
1493 this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
1494 .runTime
.aggregate
=== true
1496 workerNode
.usage
.runTime
.aggregate
= min(
1497 ...this.workerNodes
.map(
1498 workerNode
=> workerNode
.usage
.runTime
.aggregate
?? Infinity
1503 this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
1504 .waitTime
.aggregate
=== true
1506 workerNode
.usage
.waitTime
.aggregate
= min(
1507 ...this.workerNodes
.map(
1508 workerNode
=> workerNode
.usage
.waitTime
.aggregate
?? Infinity
1513 this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements().elu
1516 workerNode
.usage
.elu
.active
.aggregate
= min(
1517 ...this.workerNodes
.map(
1518 workerNode
=> workerNode
.usage
.elu
.active
.aggregate
?? Infinity
1525 * Creates a new, completely set up worker node.
1527 * @returns New, completely set up worker node key.
1529 protected createAndSetupWorkerNode (): number {
1530 const workerNode
= this.createWorkerNode()
1531 workerNode
.registerWorkerEventHandler(
1533 this.opts
.onlineHandler
?? EMPTY_FUNCTION
1535 workerNode
.registerWorkerEventHandler(
1537 this.opts
.messageHandler
?? EMPTY_FUNCTION
1539 workerNode
.registerWorkerEventHandler(
1541 this.opts
.errorHandler
?? EMPTY_FUNCTION
1543 workerNode
.registerOnceWorkerEventHandler('error', (error
: Error) => {
1544 workerNode
.info
.ready
= false
1545 this.emitter
?.emit(PoolEvents
.error
, error
)
1549 this.opts
.restartWorkerOnError
=== true
1551 if (workerNode
.info
.dynamic
) {
1552 this.createAndSetupDynamicWorkerNode()
1553 } else if (!this.startingMinimumNumberOfWorkers
) {
1554 this.startMinimumNumberOfWorkers(true)
1560 this.opts
.enableTasksQueue
=== true
1562 this.redistributeQueuedTasks(this.workerNodes
.indexOf(workerNode
))
1564 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1565 workerNode
?.terminate().catch((error
: unknown
) => {
1566 this.emitter
?.emit(PoolEvents
.error
, error
)
1569 workerNode
.registerWorkerEventHandler(
1571 this.opts
.exitHandler
?? EMPTY_FUNCTION
1573 workerNode
.registerOnceWorkerEventHandler('exit', () => {
1574 this.removeWorkerNode(workerNode
)
1577 !this.startingMinimumNumberOfWorkers
&&
1580 this.startMinimumNumberOfWorkers(true)
1583 const workerNodeKey
= this.addWorkerNode(workerNode
)
1584 this.afterWorkerNodeSetup(workerNodeKey
)
1585 return workerNodeKey
1589 * Creates a new, completely set up dynamic worker node.
1591 * @returns New, completely set up dynamic worker node key.
1593 protected createAndSetupDynamicWorkerNode (): number {
1594 const workerNodeKey
= this.createAndSetupWorkerNode()
1595 this.registerWorkerMessageListener(workerNodeKey
, message
=> {
1596 this.checkMessageWorkerId(message
)
1597 const localWorkerNodeKey
= this.getWorkerNodeKeyByWorkerId(
1600 const workerInfo
= this.getWorkerInfo(localWorkerNodeKey
)
1601 const workerUsage
= this.workerNodes
[localWorkerNodeKey
]?.usage
1602 // Kill message received from worker
1604 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
1605 (isKillBehavior(KillBehaviors
.SOFT
, message
.kill
) &&
1606 ((this.opts
.enableTasksQueue
=== false &&
1607 workerUsage
.tasks
.executing
=== 0) ||
1608 (this.opts
.enableTasksQueue
=== true &&
1609 workerInfo
!= null &&
1610 !workerInfo
.stealing
&&
1611 workerUsage
.tasks
.executing
=== 0 &&
1612 this.tasksQueueSize(localWorkerNodeKey
) === 0)))
1614 // Flag the worker node as not ready immediately
1615 this.flagWorkerNodeAsNotReady(localWorkerNodeKey
)
1616 this.destroyWorkerNode(localWorkerNodeKey
).catch((error
: unknown
) => {
1617 this.emitter
?.emit(PoolEvents
.error
, error
)
1621 this.sendToWorker(workerNodeKey
, {
1624 if (this.taskFunctions
.size
> 0) {
1625 for (const [taskFunctionName
, taskFunctionObject
] of this.taskFunctions
) {
1626 this.sendTaskFunctionOperationToWorker(workerNodeKey
, {
1627 taskFunctionOperation
: 'add',
1628 taskFunctionProperties
: buildTaskFunctionProperties(
1632 taskFunction
: taskFunctionObject
.taskFunction
.toString()
1633 }).catch((error
: unknown
) => {
1634 this.emitter
?.emit(PoolEvents
.error
, error
)
1638 const workerNode
= this.workerNodes
[workerNodeKey
]
1639 workerNode
.info
.dynamic
= true
1641 this.workerChoiceStrategiesContext
?.getPolicy().dynamicWorkerReady
===
1643 this.workerChoiceStrategiesContext
?.getPolicy().dynamicWorkerUsage
===
1646 workerNode
.info
.ready
= true
1648 this.initWorkerNodeUsage(workerNode
)
1649 this.checkAndEmitDynamicWorkerCreationEvents()
1650 return workerNodeKey
1654 * Registers a listener callback on the worker given its worker node key.
1656 * @param workerNodeKey - The worker node key.
1657 * @param listener - The message listener callback.
1659 protected abstract registerWorkerMessageListener
<
1660 Message
extends Data
| Response
1662 workerNodeKey
: number,
1663 listener
: (message
: MessageValue
<Message
>) => void
1667 * Registers once a listener callback on the worker given its worker node key.
1669 * @param workerNodeKey - The worker node key.
1670 * @param listener - The message listener callback.
1672 protected abstract registerOnceWorkerMessageListener
<
1673 Message
extends Data
| Response
1675 workerNodeKey
: number,
1676 listener
: (message
: MessageValue
<Message
>) => void
1680 * Deregisters a listener callback on the worker given its worker node key.
1682 * @param workerNodeKey - The worker node key.
1683 * @param listener - The message listener callback.
1685 protected abstract deregisterWorkerMessageListener
<
1686 Message
extends Data
| Response
1688 workerNodeKey
: number,
1689 listener
: (message
: MessageValue
<Message
>) => void
1693 * Method hooked up after a worker node has been newly created.
1694 * Can be overridden.
1696 * @param workerNodeKey - The newly created worker node key.
1698 protected afterWorkerNodeSetup (workerNodeKey
: number): void {
1699 // Listen to worker messages.
1700 this.registerWorkerMessageListener(
1702 this.workerMessageListener
1704 // Send the startup message to worker.
1705 this.sendStartupMessageToWorker(workerNodeKey
)
1706 // Send the statistics message to worker.
1707 this.sendStatisticsMessageToWorker(workerNodeKey
)
1708 if (this.opts
.enableTasksQueue
=== true) {
1709 if (this.opts
.tasksQueueOptions
?.taskStealing
=== true) {
1710 this.workerNodes
[workerNodeKey
].on(
1712 this.handleWorkerNodeIdleEvent
1715 if (this.opts
.tasksQueueOptions
?.tasksStealingOnBackPressure
=== true) {
1716 this.workerNodes
[workerNodeKey
].on(
1718 this.handleWorkerNodeBackPressureEvent
1725 * Sends the startup message to worker given its worker node key.
1727 * @param workerNodeKey - The worker node key.
1729 protected abstract sendStartupMessageToWorker (workerNodeKey
: number): void
1732 * Sends the statistics message to worker given its worker node key.
1734 * @param workerNodeKey - The worker node key.
1736 private sendStatisticsMessageToWorker (workerNodeKey
: number): void {
1737 this.sendToWorker(workerNodeKey
, {
1740 this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
1741 .runTime
.aggregate
?? false,
1743 this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
1744 .elu
.aggregate
?? false
1749 private cannotStealTask (): boolean {
1750 return this.workerNodes
.length
<= 1 || this.info
.queuedTasks
=== 0
1753 private handleTask (workerNodeKey
: number, task
: Task
<Data
>): void {
1754 if (this.shallExecuteTask(workerNodeKey
)) {
1755 this.executeTask(workerNodeKey
, task
)
1757 this.enqueueTask(workerNodeKey
, task
)
1761 private redistributeQueuedTasks (sourceWorkerNodeKey
: number): void {
1762 if (sourceWorkerNodeKey
=== -1 || this.cannotStealTask()) {
1765 while (this.tasksQueueSize(sourceWorkerNodeKey
) > 0) {
1766 const destinationWorkerNodeKey
= this.workerNodes
.reduce(
1767 (minWorkerNodeKey
, workerNode
, workerNodeKey
, workerNodes
) => {
1768 return sourceWorkerNodeKey
!== workerNodeKey
&&
1769 workerNode
.info
.ready
&&
1770 workerNode
.usage
.tasks
.queued
<
1771 workerNodes
[minWorkerNodeKey
].usage
.tasks
.queued
1778 destinationWorkerNodeKey
,
1779 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1780 this.dequeueTask(sourceWorkerNodeKey
)!
1785 private updateTaskStolenStatisticsWorkerUsage (
1786 workerNodeKey
: number,
1789 const workerNode
= this.workerNodes
[workerNodeKey
]
1790 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1791 if (workerNode
?.usage
!= null) {
1792 ++workerNode
.usage
.tasks
.stolen
1795 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1796 workerNode
.getTaskFunctionWorkerUsage(taskName
) != null
1798 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1799 ++workerNode
.getTaskFunctionWorkerUsage(taskName
)!.tasks
.stolen
1803 private updateTaskSequentiallyStolenStatisticsWorkerUsage (
1804 workerNodeKey
: number,
1806 previousTaskName
?: string
1808 const workerNode
= this.workerNodes
[workerNodeKey
]
1809 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1810 if (workerNode
?.usage
!= null) {
1811 ++workerNode
.usage
.tasks
.sequentiallyStolen
1814 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1815 workerNode
.getTaskFunctionWorkerUsage(taskName
) != null
1817 const taskFunctionWorkerUsage
=
1818 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1819 workerNode
.getTaskFunctionWorkerUsage(taskName
)!
1821 taskFunctionWorkerUsage
.tasks
.sequentiallyStolen
=== 0 ||
1822 (previousTaskName
!= null &&
1823 previousTaskName
=== taskName
&&
1824 taskFunctionWorkerUsage
.tasks
.sequentiallyStolen
> 0)
1826 ++taskFunctionWorkerUsage
.tasks
.sequentiallyStolen
1827 } else if (taskFunctionWorkerUsage
.tasks
.sequentiallyStolen
> 0) {
1828 taskFunctionWorkerUsage
.tasks
.sequentiallyStolen
= 0
1833 private resetTaskSequentiallyStolenStatisticsWorkerUsage (
1834 workerNodeKey
: number,
1837 const workerNode
= this.workerNodes
[workerNodeKey
]
1838 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1839 if (workerNode
?.usage
!= null) {
1840 workerNode
.usage
.tasks
.sequentiallyStolen
= 0
1843 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1844 workerNode
.getTaskFunctionWorkerUsage(taskName
) != null
1846 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1847 workerNode
.getTaskFunctionWorkerUsage(
1849 )!.tasks
.sequentiallyStolen
= 0
1853 private readonly handleWorkerNodeIdleEvent
= (
1854 eventDetail
: WorkerNodeEventDetail
,
1855 previousStolenTask
?: Task
<Data
>
1857 const { workerNodeKey
} = eventDetail
1858 if (workerNodeKey
== null) {
1860 "WorkerNode event detail 'workerNodeKey' property must be defined"
1863 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1864 if (workerInfo
== null) {
1866 `Worker node with key '${workerNodeKey}' not found in pool`
1870 this.cannotStealTask() ||
1871 (this.info
.stealingWorkerNodes
?? 0) >
1872 Math.floor(this.workerNodes
.length
/ 2)
1874 if (previousStolenTask
!= null) {
1875 workerInfo
.stealing
= false
1876 this.resetTaskSequentiallyStolenStatisticsWorkerUsage(
1878 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1879 previousStolenTask
.name
!
1884 const workerNodeTasksUsage
= this.workerNodes
[workerNodeKey
].usage
.tasks
1886 previousStolenTask
!= null &&
1887 (workerNodeTasksUsage
.executing
> 0 ||
1888 this.tasksQueueSize(workerNodeKey
) > 0)
1890 workerInfo
.stealing
= false
1891 this.resetTaskSequentiallyStolenStatisticsWorkerUsage(
1893 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1894 previousStolenTask
.name
!
1898 workerInfo
.stealing
= true
1899 const stolenTask
= this.workerNodeStealTask(workerNodeKey
)
1900 if (stolenTask
!= null) {
1901 this.updateTaskSequentiallyStolenStatisticsWorkerUsage(
1903 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1905 previousStolenTask
?.name
1908 sleep(exponentialDelay(workerNodeTasksUsage
.sequentiallyStolen
))
1910 this.handleWorkerNodeIdleEvent(eventDetail
, stolenTask
)
1913 .catch((error
: unknown
) => {
1914 this.emitter
?.emit(PoolEvents
.error
, error
)
1918 private readonly workerNodeStealTask
= (
1919 workerNodeKey
: number
1920 ): Task
<Data
> | undefined => {
1921 const workerNodes
= this.workerNodes
1924 (workerNodeA
, workerNodeB
) =>
1925 workerNodeB
.usage
.tasks
.queued
- workerNodeA
.usage
.tasks
.queued
1927 const sourceWorkerNode
= workerNodes
.find(
1928 (sourceWorkerNode
, sourceWorkerNodeKey
) =>
1929 sourceWorkerNode
.info
.ready
&&
1930 !sourceWorkerNode
.info
.stealing
&&
1931 sourceWorkerNodeKey
!== workerNodeKey
&&
1932 sourceWorkerNode
.usage
.tasks
.queued
> 0
1934 if (sourceWorkerNode
!= null) {
1935 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1936 const task
= sourceWorkerNode
.dequeueLastPrioritizedTask()!
1937 this.handleTask(workerNodeKey
, task
)
1938 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1939 this.updateTaskStolenStatisticsWorkerUsage(workerNodeKey
, task
.name
!)
1944 private readonly handleWorkerNodeBackPressureEvent
= (
1945 eventDetail
: WorkerNodeEventDetail
1948 this.cannotStealTask() ||
1949 this.hasBackPressure() ||
1950 (this.info
.stealingWorkerNodes
?? 0) >
1951 Math.floor(this.workerNodes
.length
/ 2)
1955 const sizeOffset
= 1
1956 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1957 if (this.opts
.tasksQueueOptions
!.size
! <= sizeOffset
) {
1960 const { workerId
} = eventDetail
1961 const sourceWorkerNode
=
1962 this.workerNodes
[this.getWorkerNodeKeyByWorkerId(workerId
)]
1963 const workerNodes
= this.workerNodes
1966 (workerNodeA
, workerNodeB
) =>
1967 workerNodeA
.usage
.tasks
.queued
- workerNodeB
.usage
.tasks
.queued
1969 for (const [workerNodeKey
, workerNode
] of workerNodes
.entries()) {
1971 sourceWorkerNode
.usage
.tasks
.queued
> 0 &&
1972 workerNode
.info
.ready
&&
1973 !workerNode
.info
.stealing
&&
1974 workerNode
.info
.id
!== workerId
&&
1975 workerNode
.usage
.tasks
.queued
<
1976 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1977 this.opts
.tasksQueueOptions
!.size
! - sizeOffset
1979 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1980 if (workerInfo
== null) {
1982 `Worker node with key '${workerNodeKey}' not found in pool`
1985 workerInfo
.stealing
= true
1986 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1987 const task
= sourceWorkerNode
.dequeueLastPrioritizedTask()!
1988 this.handleTask(workerNodeKey
, task
)
1989 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1990 this.updateTaskStolenStatisticsWorkerUsage(workerNodeKey
, task
.name
!)
1991 workerInfo
.stealing
= false
1997 * This method is the message listener registered on each worker.
1999 protected readonly workerMessageListener
= (
2000 message
: MessageValue
<Response
>
2002 this.checkMessageWorkerId(message
)
2003 const { workerId
, ready
, taskId
, taskFunctionsProperties
} = message
2004 if (ready
!= null && taskFunctionsProperties
!= null) {
2005 // Worker ready response received from worker
2006 this.handleWorkerReadyResponse(message
)
2007 } else if (taskFunctionsProperties
!= null) {
2008 // Task function properties message received from worker
2009 const workerNodeKey
= this.getWorkerNodeKeyByWorkerId(workerId
)
2010 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
2011 if (workerInfo
!= null) {
2012 workerInfo
.taskFunctionsProperties
= taskFunctionsProperties
2013 this.sendStatisticsMessageToWorker(workerNodeKey
)
2015 } else if (taskId
!= null) {
2016 // Task execution response received from worker
2017 this.handleTaskExecutionResponse(message
)
2021 private checkAndEmitReadyEvent (): void {
2022 if (!this.readyEventEmitted
&& this.ready
) {
2023 this.emitter
?.emit(PoolEvents
.ready
, this.info
)
2024 this.readyEventEmitted
= true
2028 private handleWorkerReadyResponse (message
: MessageValue
<Response
>): void {
2029 const { workerId
, ready
, taskFunctionsProperties
} = message
2030 if (ready
== null || !ready
) {
2031 throw new Error(`Worker ${workerId} failed to initialize`)
2033 const workerNodeKey
= this.getWorkerNodeKeyByWorkerId(workerId
)
2034 const workerNode
= this.workerNodes
[workerNodeKey
]
2035 workerNode
.info
.ready
= ready
2036 workerNode
.info
.taskFunctionsProperties
= taskFunctionsProperties
2037 this.sendStatisticsMessageToWorker(workerNodeKey
)
2038 this.checkAndEmitReadyEvent()
2041 private handleTaskExecutionResponse (message
: MessageValue
<Response
>): void {
2042 const { workerId
, taskId
, workerError
, data
} = message
2043 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2044 const promiseResponse
= this.promiseResponseMap
.get(taskId
!)
2045 if (promiseResponse
!= null) {
2046 const { resolve
, reject
, workerNodeKey
, asyncResource
} = promiseResponse
2047 const workerNode
= this.workerNodes
[workerNodeKey
]
2048 if (workerError
!= null) {
2049 this.emitter
?.emit(PoolEvents
.taskError
, workerError
)
2050 asyncResource
!= null
2051 ? asyncResource
.runInAsyncScope(
2056 : reject(workerError
.message
)
2058 asyncResource
!= null
2059 ? asyncResource
.runInAsyncScope(resolve
, this.emitter
, data
)
2060 : resolve(data
as Response
)
2062 asyncResource
?.emitDestroy()
2063 this.afterTaskExecutionHook(workerNodeKey
, message
)
2064 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2065 this.promiseResponseMap
.delete(taskId
!)
2066 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
2067 workerNode
?.emit('taskFinished', taskId
)
2069 this.opts
.enableTasksQueue
=== true &&
2071 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
2074 const workerNodeTasksUsage
= workerNode
.usage
.tasks
2076 this.tasksQueueSize(workerNodeKey
) > 0 &&
2077 workerNodeTasksUsage
.executing
<
2078 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2079 this.opts
.tasksQueueOptions
!.concurrency
!
2081 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2082 this.executeTask(workerNodeKey
, this.dequeueTask(workerNodeKey
)!)
2085 workerNodeTasksUsage
.executing
=== 0 &&
2086 this.tasksQueueSize(workerNodeKey
) === 0 &&
2087 workerNodeTasksUsage
.sequentiallyStolen
=== 0
2089 workerNode
.emit('idle', {
2098 private checkAndEmitTaskExecutionEvents (): void {
2100 this.emitter
?.emit(PoolEvents
.busy
, this.info
)
2104 private checkAndEmitTaskQueuingEvents (): void {
2105 if (this.hasBackPressure()) {
2106 this.emitter
?.emit(PoolEvents
.backPressure
, this.info
)
2111 * Emits dynamic worker creation events.
2113 protected abstract checkAndEmitDynamicWorkerCreationEvents (): void
2116 * Gets the worker information given its worker node key.
2118 * @param workerNodeKey - The worker node key.
2119 * @returns The worker information.
2121 protected getWorkerInfo (workerNodeKey
: number): WorkerInfo
| undefined {
2122 return this.workerNodes
[workerNodeKey
]?.info
2126 * Creates a worker node.
2128 * @returns The created worker node.
2130 private createWorkerNode (): IWorkerNode
<Worker
, Data
> {
2131 const workerNode
= new WorkerNode
<Worker
, Data
>(
2136 workerOptions
: this.opts
.workerOptions
,
2137 tasksQueueBackPressureSize
:
2138 this.opts
.tasksQueueOptions
?.size
??
2139 getDefaultTasksQueueOptions(
2140 this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
2142 tasksQueueBucketSize
:
2143 (this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
) * 2
2146 // Flag the worker node as ready at pool startup.
2147 if (this.starting
) {
2148 workerNode
.info
.ready
= true
2154 * Adds the given worker node in the pool worker nodes.
2156 * @param workerNode - The worker node.
2157 * @returns The added worker node key.
2158 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the added worker node is not found.
2160 private addWorkerNode (workerNode
: IWorkerNode
<Worker
, Data
>): number {
2161 this.workerNodes
.push(workerNode
)
2162 const workerNodeKey
= this.workerNodes
.indexOf(workerNode
)
2163 if (workerNodeKey
=== -1) {
2164 throw new Error('Worker added not found in worker nodes')
2166 return workerNodeKey
2169 private checkAndEmitEmptyEvent (): void {
2171 this.emitter
?.emit(PoolEvents
.empty
, this.info
)
2172 this.readyEventEmitted
= false
2177 * Removes the worker node from the pool worker nodes.
2179 * @param workerNode - The worker node.
2181 private removeWorkerNode (workerNode
: IWorkerNode
<Worker
, Data
>): void {
2182 const workerNodeKey
= this.workerNodes
.indexOf(workerNode
)
2183 if (workerNodeKey
!== -1) {
2184 this.workerNodes
.splice(workerNodeKey
, 1)
2185 this.workerChoiceStrategiesContext
?.remove(workerNodeKey
)
2187 this.checkAndEmitEmptyEvent()
2190 protected flagWorkerNodeAsNotReady (workerNodeKey
: number): void {
2191 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
2192 if (workerInfo
!= null) {
2193 workerInfo
.ready
= false
2197 private hasBackPressure (): boolean {
2199 this.opts
.enableTasksQueue
=== true &&
2200 this.workerNodes
.findIndex(
2201 workerNode
=> !workerNode
.hasBackPressure()
2207 * Executes the given task on the worker given its worker node key.
2209 * @param workerNodeKey - The worker node key.
2210 * @param task - The task to execute.
2212 private executeTask (workerNodeKey
: number, task
: Task
<Data
>): void {
2213 this.beforeTaskExecutionHook(workerNodeKey
, task
)
2214 this.sendToWorker(workerNodeKey
, task
, task
.transferList
)
2215 this.checkAndEmitTaskExecutionEvents()
2218 private enqueueTask (workerNodeKey
: number, task
: Task
<Data
>): number {
2219 const tasksQueueSize
= this.workerNodes
[workerNodeKey
].enqueueTask(task
)
2220 this.checkAndEmitTaskQueuingEvents()
2221 return tasksQueueSize
2224 private dequeueTask (workerNodeKey
: number): Task
<Data
> | undefined {
2225 return this.workerNodes
[workerNodeKey
].dequeueTask()
2228 private tasksQueueSize (workerNodeKey
: number): number {
2229 return this.workerNodes
[workerNodeKey
].tasksQueueSize()
2232 protected flushTasksQueue (workerNodeKey
: number): number {
2233 let flushedTasks
= 0
2234 while (this.tasksQueueSize(workerNodeKey
) > 0) {
2235 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2236 this.executeTask(workerNodeKey
, this.dequeueTask(workerNodeKey
)!)
2239 this.workerNodes
[workerNodeKey
].clearTasksQueue()
2243 private flushTasksQueues (): void {
2244 for (const workerNodeKey
of this.workerNodes
.keys()) {
2245 this.flushTasksQueue(workerNodeKey
)