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(
380 workerNode
.usage
.runTime
.minimum
?? Number.POSITIVE_INFINITY
386 ...this.workerNodes
.map(
388 workerNode
.usage
.runTime
.maximum
?? Number.NEGATIVE_INFINITY
392 ...(this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
393 .runTime
.average
&& {
396 this.workerNodes
.reduce
<number[]>(
397 (accumulator
, workerNode
) =>
398 accumulator
.concat(workerNode
.usage
.runTime
.history
),
404 ...(this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
408 this.workerNodes
.reduce
<number[]>(
409 (accumulator
, workerNode
) =>
410 accumulator
.concat(workerNode
.usage
.runTime
.history
),
418 ...(this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
419 .waitTime
.aggregate
=== true && {
423 ...this.workerNodes
.map(
425 workerNode
.usage
.waitTime
.minimum
?? Number.POSITIVE_INFINITY
431 ...this.workerNodes
.map(
433 workerNode
.usage
.waitTime
.maximum
?? Number.NEGATIVE_INFINITY
437 ...(this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
438 .waitTime
.average
&& {
441 this.workerNodes
.reduce
<number[]>(
442 (accumulator
, workerNode
) =>
443 accumulator
.concat(workerNode
.usage
.waitTime
.history
),
449 ...(this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
450 .waitTime
.median
&& {
453 this.workerNodes
.reduce
<number[]>(
454 (accumulator
, workerNode
) =>
455 accumulator
.concat(workerNode
.usage
.waitTime
.history
),
463 ...(this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
464 .elu
.aggregate
=== true && {
469 ...this.workerNodes
.map(
471 workerNode
.usage
.elu
.idle
.minimum
??
472 Number.POSITIVE_INFINITY
478 ...this.workerNodes
.map(
480 workerNode
.usage
.elu
.idle
.maximum
??
481 Number.NEGATIVE_INFINITY
485 ...(this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
489 this.workerNodes
.reduce
<number[]>(
490 (accumulator
, workerNode
) =>
491 accumulator
.concat(workerNode
.usage
.elu
.idle
.history
),
497 ...(this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
501 this.workerNodes
.reduce
<number[]>(
502 (accumulator
, workerNode
) =>
503 accumulator
.concat(workerNode
.usage
.elu
.idle
.history
),
513 ...this.workerNodes
.map(
515 workerNode
.usage
.elu
.active
.minimum
??
516 Number.POSITIVE_INFINITY
522 ...this.workerNodes
.map(
524 workerNode
.usage
.elu
.active
.maximum
??
525 Number.NEGATIVE_INFINITY
529 ...(this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
533 this.workerNodes
.reduce
<number[]>(
534 (accumulator
, workerNode
) =>
535 accumulator
.concat(workerNode
.usage
.elu
.active
.history
),
541 ...(this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
545 this.workerNodes
.reduce
<number[]>(
546 (accumulator
, workerNode
) =>
547 accumulator
.concat(workerNode
.usage
.elu
.active
.history
),
557 this.workerNodes
.map(
558 workerNode
=> workerNode
.usage
.elu
.utilization
?? 0
564 this.workerNodes
.map(
565 workerNode
=> workerNode
.usage
.elu
.utilization
?? 0
576 * The pool readiness boolean status.
578 private get
ready (): boolean {
583 this.workerNodes
.reduce(
584 (accumulator
, workerNode
) =>
585 !workerNode
.info
.dynamic
&& workerNode
.info
.ready
589 ) >= this.minimumNumberOfWorkers
594 * The pool emptiness boolean status.
596 protected get
empty (): boolean {
597 return this.minimumNumberOfWorkers
=== 0 && this.workerNodes
.length
=== 0
601 * The approximate pool utilization.
603 * @returns The pool utilization.
605 private get
utilization (): number {
606 if (this.startTimestamp
== null) {
609 const poolTimeCapacity
=
610 (performance
.now() - this.startTimestamp
) *
611 (this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
)
612 const totalTasksRunTime
= this.workerNodes
.reduce(
613 (accumulator
, workerNode
) =>
614 accumulator
+ (workerNode
.usage
.runTime
.aggregate
?? 0),
617 const totalTasksWaitTime
= this.workerNodes
.reduce(
618 (accumulator
, workerNode
) =>
619 accumulator
+ (workerNode
.usage
.waitTime
.aggregate
?? 0),
622 return (totalTasksRunTime
+ totalTasksWaitTime
) / poolTimeCapacity
628 * If it is `'dynamic'`, it provides the `max` property.
630 protected abstract get
type (): PoolType
635 protected abstract get
worker (): WorkerType
638 * Checks if the worker id sent in the received message from a worker is valid.
640 * @param message - The received message.
641 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the worker id is invalid.
643 private checkMessageWorkerId (message
: MessageValue
<Data
| Response
>): void {
644 if (message
.workerId
== null) {
645 throw new Error('Worker message received without worker id')
646 } else if (this.getWorkerNodeKeyByWorkerId(message
.workerId
) === -1) {
648 `Worker message received from unknown worker '${message.workerId}'`
654 * Gets the worker node key given its worker id.
656 * @param workerId - The worker id.
657 * @returns The worker node key if the worker id is found in the pool worker nodes, `-1` otherwise.
659 private getWorkerNodeKeyByWorkerId (workerId
: number | undefined): number {
660 return this.workerNodes
.findIndex(
661 workerNode
=> workerNode
.info
.id
=== workerId
666 public setWorkerChoiceStrategy (
667 workerChoiceStrategy
: WorkerChoiceStrategy
,
668 workerChoiceStrategyOptions
?: WorkerChoiceStrategyOptions
670 let requireSync
= false
671 checkValidWorkerChoiceStrategy(workerChoiceStrategy
)
672 if (workerChoiceStrategyOptions
!= null) {
673 requireSync
= !this.setWorkerChoiceStrategyOptions(
674 workerChoiceStrategyOptions
677 if (workerChoiceStrategy
!== this.opts
.workerChoiceStrategy
) {
678 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
679 this.workerChoiceStrategiesContext
?.setDefaultWorkerChoiceStrategy(
680 this.opts
.workerChoiceStrategy
,
681 this.opts
.workerChoiceStrategyOptions
686 this.workerChoiceStrategiesContext
?.syncWorkerChoiceStrategies(
687 this.getWorkerChoiceStrategies(),
688 this.opts
.workerChoiceStrategyOptions
690 for (const workerNodeKey
of this.workerNodes
.keys()) {
691 this.sendStatisticsMessageToWorker(workerNodeKey
)
697 public setWorkerChoiceStrategyOptions (
698 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
| undefined
700 this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
701 if (workerChoiceStrategyOptions
!= null) {
702 this.opts
.workerChoiceStrategyOptions
= workerChoiceStrategyOptions
703 this.workerChoiceStrategiesContext
?.setOptions(
704 this.opts
.workerChoiceStrategyOptions
706 this.workerChoiceStrategiesContext
?.syncWorkerChoiceStrategies(
707 this.getWorkerChoiceStrategies(),
708 this.opts
.workerChoiceStrategyOptions
710 for (const workerNodeKey
of this.workerNodes
.keys()) {
711 this.sendStatisticsMessageToWorker(workerNodeKey
)
719 public enableTasksQueue (
721 tasksQueueOptions
?: TasksQueueOptions
723 if (this.opts
.enableTasksQueue
=== true && !enable
) {
724 this.unsetTaskStealing()
725 this.unsetTasksStealingOnBackPressure()
726 this.flushTasksQueues()
728 this.opts
.enableTasksQueue
= enable
729 this.setTasksQueueOptions(tasksQueueOptions
)
733 public setTasksQueueOptions (
734 tasksQueueOptions
: TasksQueueOptions
| undefined
736 if (this.opts
.enableTasksQueue
=== true) {
737 checkValidTasksQueueOptions(tasksQueueOptions
)
738 this.opts
.tasksQueueOptions
=
739 this.buildTasksQueueOptions(tasksQueueOptions
)
740 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
741 this.setTasksQueueSize(this.opts
.tasksQueueOptions
.size
!)
742 if (this.opts
.tasksQueueOptions
.taskStealing
=== true) {
743 this.unsetTaskStealing()
744 this.setTaskStealing()
746 this.unsetTaskStealing()
748 if (this.opts
.tasksQueueOptions
.tasksStealingOnBackPressure
=== true) {
749 this.unsetTasksStealingOnBackPressure()
750 this.setTasksStealingOnBackPressure()
752 this.unsetTasksStealingOnBackPressure()
754 } else if (this.opts
.tasksQueueOptions
!= null) {
755 delete this.opts
.tasksQueueOptions
759 private buildTasksQueueOptions (
760 tasksQueueOptions
: TasksQueueOptions
| undefined
761 ): TasksQueueOptions
{
763 ...getDefaultTasksQueueOptions(
764 this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
770 private setTasksQueueSize (size
: number): void {
771 for (const workerNode
of this.workerNodes
) {
772 workerNode
.tasksQueueBackPressureSize
= size
776 private setTaskStealing (): void {
777 for (const workerNodeKey
of this.workerNodes
.keys()) {
778 this.workerNodes
[workerNodeKey
].on('idle', this.handleWorkerNodeIdleEvent
)
782 private unsetTaskStealing (): void {
783 for (const workerNodeKey
of this.workerNodes
.keys()) {
784 this.workerNodes
[workerNodeKey
].off(
786 this.handleWorkerNodeIdleEvent
791 private setTasksStealingOnBackPressure (): void {
792 for (const workerNodeKey
of this.workerNodes
.keys()) {
793 this.workerNodes
[workerNodeKey
].on(
795 this.handleWorkerNodeBackPressureEvent
800 private unsetTasksStealingOnBackPressure (): void {
801 for (const workerNodeKey
of this.workerNodes
.keys()) {
802 this.workerNodes
[workerNodeKey
].off(
804 this.handleWorkerNodeBackPressureEvent
810 * Whether the pool is full or not.
812 * The pool filling boolean status.
814 protected get
full (): boolean {
816 this.workerNodes
.length
>=
817 (this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
)
822 * Whether the pool is busy or not.
824 * The pool busyness boolean status.
826 protected abstract get
busy (): boolean
829 * Whether worker nodes are executing concurrently their tasks quota or not.
831 * @returns Worker nodes busyness boolean status.
833 protected internalBusy (): boolean {
834 if (this.opts
.enableTasksQueue
=== true) {
836 this.workerNodes
.findIndex(
838 workerNode
.info
.ready
&&
839 workerNode
.usage
.tasks
.executing
<
840 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
841 this.opts
.tasksQueueOptions
!.concurrency
!
846 this.workerNodes
.findIndex(
848 workerNode
.info
.ready
&& workerNode
.usage
.tasks
.executing
=== 0
853 private isWorkerNodeBusy (workerNodeKey
: number): boolean {
854 if (this.opts
.enableTasksQueue
=== true) {
856 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
>=
857 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
858 this.opts
.tasksQueueOptions
!.concurrency
!
861 return this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
> 0
864 private async sendTaskFunctionOperationToWorker (
865 workerNodeKey
: number,
866 message
: MessageValue
<Data
>
867 ): Promise
<boolean> {
868 return await new Promise
<boolean>((resolve
, reject
) => {
869 const taskFunctionOperationListener
= (
870 message
: MessageValue
<Response
>
872 this.checkMessageWorkerId(message
)
873 const workerId
= this.getWorkerInfo(workerNodeKey
)?.id
875 message
.taskFunctionOperationStatus
!= null &&
876 message
.workerId
=== workerId
878 if (message
.taskFunctionOperationStatus
) {
883 `Task function operation '${message.taskFunctionOperation}' failed on worker ${message.workerId} with error: '${message.workerError?.message}'`
887 this.deregisterWorkerMessageListener(
888 this.getWorkerNodeKeyByWorkerId(message
.workerId
),
889 taskFunctionOperationListener
893 this.registerWorkerMessageListener(
895 taskFunctionOperationListener
897 this.sendToWorker(workerNodeKey
, message
)
901 private async sendTaskFunctionOperationToWorkers (
902 message
: MessageValue
<Data
>
903 ): Promise
<boolean> {
904 return await new Promise
<boolean>((resolve
, reject
) => {
905 const responsesReceived
= new Array<MessageValue
<Response
>>()
906 const taskFunctionOperationsListener
= (
907 message
: MessageValue
<Response
>
909 this.checkMessageWorkerId(message
)
910 if (message
.taskFunctionOperationStatus
!= null) {
911 responsesReceived
.push(message
)
912 if (responsesReceived
.length
=== this.workerNodes
.length
) {
914 responsesReceived
.every(
915 message
=> message
.taskFunctionOperationStatus
=== true
920 responsesReceived
.some(
921 message
=> message
.taskFunctionOperationStatus
=== false
924 const errorResponse
= responsesReceived
.find(
925 response
=> response
.taskFunctionOperationStatus
=== false
929 `Task function operation '${
930 message.taskFunctionOperation as string
931 }' failed on worker ${errorResponse?.workerId} with error: '${
932 errorResponse?.workerError?.message
937 this.deregisterWorkerMessageListener(
938 this.getWorkerNodeKeyByWorkerId(message
.workerId
),
939 taskFunctionOperationsListener
944 for (const workerNodeKey
of this.workerNodes
.keys()) {
945 this.registerWorkerMessageListener(
947 taskFunctionOperationsListener
949 this.sendToWorker(workerNodeKey
, message
)
955 public hasTaskFunction (name
: string): boolean {
956 return this.listTaskFunctionsProperties().some(
957 taskFunctionProperties
=> taskFunctionProperties
.name
=== name
962 public async addTaskFunction (
964 fn
: TaskFunction
<Data
, Response
> | TaskFunctionObject
<Data
, Response
>
965 ): Promise
<boolean> {
966 if (typeof name
!== 'string') {
967 throw new TypeError('name argument must be a string')
969 if (typeof name
=== 'string' && name
.trim().length
=== 0) {
970 throw new TypeError('name argument must not be an empty string')
972 if (typeof fn
=== 'function') {
973 fn
= { taskFunction
: fn
} satisfies TaskFunctionObject
<Data
, Response
>
975 if (typeof fn
.taskFunction
!== 'function') {
976 throw new TypeError('taskFunction property must be a function')
978 checkValidPriority(fn
.priority
)
979 checkValidWorkerChoiceStrategy(fn
.strategy
)
980 const opResult
= await this.sendTaskFunctionOperationToWorkers({
981 taskFunctionOperation
: 'add',
982 taskFunctionProperties
: buildTaskFunctionProperties(name
, fn
),
983 taskFunction
: fn
.taskFunction
.toString()
985 this.taskFunctions
.set(name
, fn
)
986 this.workerChoiceStrategiesContext
?.syncWorkerChoiceStrategies(
987 this.getWorkerChoiceStrategies()
989 for (const workerNodeKey
of this.workerNodes
.keys()) {
990 this.sendStatisticsMessageToWorker(workerNodeKey
)
996 public async removeTaskFunction (name
: string): Promise
<boolean> {
997 if (!this.taskFunctions
.has(name
)) {
999 'Cannot remove a task function not handled on the pool side'
1002 const opResult
= await this.sendTaskFunctionOperationToWorkers({
1003 taskFunctionOperation
: 'remove',
1004 taskFunctionProperties
: buildTaskFunctionProperties(
1006 this.taskFunctions
.get(name
)
1009 for (const workerNode
of this.workerNodes
) {
1010 workerNode
.deleteTaskFunctionWorkerUsage(name
)
1012 this.taskFunctions
.delete(name
)
1013 this.workerChoiceStrategiesContext
?.syncWorkerChoiceStrategies(
1014 this.getWorkerChoiceStrategies()
1016 for (const workerNodeKey
of this.workerNodes
.keys()) {
1017 this.sendStatisticsMessageToWorker(workerNodeKey
)
1023 public listTaskFunctionsProperties (): TaskFunctionProperties
[] {
1024 for (const workerNode
of this.workerNodes
) {
1026 Array.isArray(workerNode
.info
.taskFunctionsProperties
) &&
1027 workerNode
.info
.taskFunctionsProperties
.length
> 0
1029 return workerNode
.info
.taskFunctionsProperties
1036 * Gets task function worker choice strategy, if any.
1038 * @param name - The task function name.
1039 * @returns The task function worker choice strategy if the task function worker choice strategy is defined, `undefined` otherwise.
1041 private readonly getTaskFunctionWorkerChoiceStrategy
= (
1043 ): WorkerChoiceStrategy
| undefined => {
1044 name
= name
?? DEFAULT_TASK_NAME
1045 const taskFunctionsProperties
= this.listTaskFunctionsProperties()
1046 if (name
=== DEFAULT_TASK_NAME
) {
1047 name
= taskFunctionsProperties
[1]?.name
1049 return taskFunctionsProperties
.find(
1050 (taskFunctionProperties
: TaskFunctionProperties
) =>
1051 taskFunctionProperties
.name
=== name
1056 * Gets worker node task function worker choice strategy, if any.
1058 * @param workerNodeKey - The worker node key.
1059 * @param name - The task function name.
1060 * @returns The worker node task function worker choice strategy if the worker node task function worker choice strategy is defined, `undefined` otherwise.
1062 private readonly getWorkerNodeTaskFunctionWorkerChoiceStrategy
= (
1063 workerNodeKey
: number,
1065 ): WorkerChoiceStrategy
| undefined => {
1066 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1067 if (workerInfo
== null) {
1070 name
= name
?? DEFAULT_TASK_NAME
1071 if (name
=== DEFAULT_TASK_NAME
) {
1072 name
= workerInfo
.taskFunctionsProperties
?.[1]?.name
1074 return workerInfo
.taskFunctionsProperties
?.find(
1075 (taskFunctionProperties
: TaskFunctionProperties
) =>
1076 taskFunctionProperties
.name
=== name
1081 * Gets worker node task function priority, if any.
1083 * @param workerNodeKey - The worker node key.
1084 * @param name - The task function name.
1085 * @returns The worker node task function priority if the worker node task function priority is defined, `undefined` otherwise.
1087 private readonly getWorkerNodeTaskFunctionPriority
= (
1088 workerNodeKey
: number,
1090 ): number | undefined => {
1091 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1092 if (workerInfo
== null) {
1095 name
= name
?? DEFAULT_TASK_NAME
1096 if (name
=== DEFAULT_TASK_NAME
) {
1097 name
= workerInfo
.taskFunctionsProperties
?.[1]?.name
1099 return workerInfo
.taskFunctionsProperties
?.find(
1100 (taskFunctionProperties
: TaskFunctionProperties
) =>
1101 taskFunctionProperties
.name
=== name
1106 * Gets the worker choice strategies registered in this pool.
1108 * @returns The worker choice strategies.
1110 private readonly getWorkerChoiceStrategies
=
1111 (): Set
<WorkerChoiceStrategy
> => {
1113 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1114 this.opts
.workerChoiceStrategy
!,
1115 ...(this.listTaskFunctionsProperties()
1117 (taskFunctionProperties
: TaskFunctionProperties
) =>
1118 taskFunctionProperties
.strategy
1121 (strategy
: WorkerChoiceStrategy
| undefined) => strategy
!= null
1122 ) as WorkerChoiceStrategy
[])
1127 public async setDefaultTaskFunction (name
: string): Promise
<boolean> {
1128 return await this.sendTaskFunctionOperationToWorkers({
1129 taskFunctionOperation
: 'default',
1130 taskFunctionProperties
: buildTaskFunctionProperties(
1132 this.taskFunctions
.get(name
)
1137 private shallExecuteTask (workerNodeKey
: number): boolean {
1139 this.tasksQueueSize(workerNodeKey
) === 0 &&
1140 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
<
1141 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1142 this.opts
.tasksQueueOptions
!.concurrency
!
1147 public async execute (
1150 transferList
?: readonly TransferListItem
[]
1151 ): Promise
<Response
> {
1152 return await new Promise
<Response
>((resolve
, reject
) => {
1153 if (!this.started
) {
1154 reject(new Error('Cannot execute a task on not started pool'))
1157 if (this.destroying
) {
1158 reject(new Error('Cannot execute a task on destroying pool'))
1161 if (name
!= null && typeof name
!== 'string') {
1162 reject(new TypeError('name argument must be a string'))
1167 typeof name
=== 'string' &&
1168 name
.trim().length
=== 0
1170 reject(new TypeError('name argument must not be an empty string'))
1173 if (transferList
!= null && !Array.isArray(transferList
)) {
1174 reject(new TypeError('transferList argument must be an array'))
1177 const timestamp
= performance
.now()
1178 const workerNodeKey
= this.chooseWorkerNode(name
)
1179 const task
: Task
<Data
> = {
1180 name
: name
?? DEFAULT_TASK_NAME
,
1181 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
1182 data
: data
?? ({} as Data
),
1183 priority
: this.getWorkerNodeTaskFunctionPriority(workerNodeKey
, name
),
1184 strategy
: this.getWorkerNodeTaskFunctionWorkerChoiceStrategy(
1190 taskId
: randomUUID()
1192 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1193 this.promiseResponseMap
.set(task
.taskId
!, {
1197 ...(this.emitter
!= null && {
1198 asyncResource
: new AsyncResource('poolifier:task', {
1199 triggerAsyncId
: this.emitter
.asyncId
,
1200 requireManualDestroy
: true
1205 this.opts
.enableTasksQueue
=== false ||
1206 (this.opts
.enableTasksQueue
=== true &&
1207 this.shallExecuteTask(workerNodeKey
))
1209 this.executeTask(workerNodeKey
, task
)
1211 this.enqueueTask(workerNodeKey
, task
)
1217 * Starts the minimum number of workers.
1219 private startMinimumNumberOfWorkers (initWorkerNodeUsage
= false): void {
1220 this.startingMinimumNumberOfWorkers
= true
1222 this.workerNodes
.reduce(
1223 (accumulator
, workerNode
) =>
1224 !workerNode
.info
.dynamic
? accumulator
+ 1 : accumulator
,
1226 ) < this.minimumNumberOfWorkers
1228 const workerNodeKey
= this.createAndSetupWorkerNode()
1229 initWorkerNodeUsage
&&
1230 this.initWorkerNodeUsage(this.workerNodes
[workerNodeKey
])
1232 this.startingMinimumNumberOfWorkers
= false
1236 public start (): void {
1238 throw new Error('Cannot start an already started pool')
1240 if (this.starting
) {
1241 throw new Error('Cannot start an already starting pool')
1243 if (this.destroying
) {
1244 throw new Error('Cannot start a destroying pool')
1246 this.starting
= true
1247 this.startMinimumNumberOfWorkers()
1248 this.startTimestamp
= performance
.now()
1249 this.starting
= false
1254 public async destroy (): Promise
<void> {
1255 if (!this.started
) {
1256 throw new Error('Cannot destroy an already destroyed pool')
1258 if (this.starting
) {
1259 throw new Error('Cannot destroy an starting pool')
1261 if (this.destroying
) {
1262 throw new Error('Cannot destroy an already destroying pool')
1264 this.destroying
= true
1266 this.workerNodes
.map(async (_
, workerNodeKey
) => {
1267 await this.destroyWorkerNode(workerNodeKey
)
1270 this.emitter
?.emit(PoolEvents
.destroy
, this.info
)
1271 this.emitter
?.emitDestroy()
1272 this.readyEventEmitted
= false
1273 delete this.startTimestamp
1274 this.destroying
= false
1275 this.started
= false
1278 private async sendKillMessageToWorker (workerNodeKey
: number): Promise
<void> {
1279 await new Promise
<void>((resolve
, reject
) => {
1280 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1281 if (this.workerNodes
[workerNodeKey
] == null) {
1285 const killMessageListener
= (message
: MessageValue
<Response
>): void => {
1286 this.checkMessageWorkerId(message
)
1287 if (message
.kill
=== 'success') {
1289 } else if (message
.kill
=== 'failure') {
1292 `Kill message handling failed on worker ${message.workerId}`
1297 // FIXME: should be registered only once
1298 this.registerWorkerMessageListener(workerNodeKey
, killMessageListener
)
1299 this.sendToWorker(workerNodeKey
, { kill
: true })
1304 * Terminates the worker node given its worker node key.
1306 * @param workerNodeKey - The worker node key.
1308 protected async destroyWorkerNode (workerNodeKey
: number): Promise
<void> {
1309 this.flagWorkerNodeAsNotReady(workerNodeKey
)
1310 const flushedTasks
= this.flushTasksQueue(workerNodeKey
)
1311 const workerNode
= this.workerNodes
[workerNodeKey
]
1312 await waitWorkerNodeEvents(
1316 this.opts
.tasksQueueOptions
?.tasksFinishedTimeout
??
1317 getDefaultTasksQueueOptions(
1318 this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
1319 ).tasksFinishedTimeout
1321 await this.sendKillMessageToWorker(workerNodeKey
)
1322 await workerNode
.terminate()
1326 * Setup hook to execute code before worker nodes are created in the abstract constructor.
1327 * Can be overridden.
1331 protected setupHook (): void {
1332 /* Intentionally empty */
1336 * Returns whether the worker is the main worker or not.
1338 * @returns `true` if the worker is the main worker, `false` otherwise.
1340 protected abstract isMain (): boolean
1343 * Hook executed before the worker task execution.
1344 * Can be overridden.
1346 * @param workerNodeKey - The worker node key.
1347 * @param task - The task to execute.
1349 protected beforeTaskExecutionHook (
1350 workerNodeKey
: number,
1353 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1354 if (this.workerNodes
[workerNodeKey
]?.usage
!= null) {
1355 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
1356 ++workerUsage
.tasks
.executing
1357 updateWaitTimeWorkerUsage(
1358 this.workerChoiceStrategiesContext
,
1364 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1365 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1366 this.workerNodes
[workerNodeKey
].getTaskFunctionWorkerUsage(task
.name
!) !=
1369 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1370 const taskFunctionWorkerUsage
= this.workerNodes
[
1372 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1373 ].getTaskFunctionWorkerUsage(task
.name
!)!
1374 ++taskFunctionWorkerUsage
.tasks
.executing
1375 updateWaitTimeWorkerUsage(
1376 this.workerChoiceStrategiesContext
,
1377 taskFunctionWorkerUsage
,
1384 * Hook executed after the worker task execution.
1385 * Can be overridden.
1387 * @param workerNodeKey - The worker node key.
1388 * @param message - The received message.
1390 protected afterTaskExecutionHook (
1391 workerNodeKey
: number,
1392 message
: MessageValue
<Response
>
1394 let needWorkerChoiceStrategiesUpdate
= false
1395 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1396 if (this.workerNodes
[workerNodeKey
]?.usage
!= null) {
1397 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
1398 updateTaskStatisticsWorkerUsage(workerUsage
, message
)
1399 updateRunTimeWorkerUsage(
1400 this.workerChoiceStrategiesContext
,
1404 updateEluWorkerUsage(
1405 this.workerChoiceStrategiesContext
,
1409 needWorkerChoiceStrategiesUpdate
= true
1412 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1413 this.workerNodes
[workerNodeKey
].getTaskFunctionWorkerUsage(
1414 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1415 message
.taskPerformance
!.name
1418 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1419 const taskFunctionWorkerUsage
= this.workerNodes
[
1421 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1422 ].getTaskFunctionWorkerUsage(message
.taskPerformance
!.name
)!
1423 updateTaskStatisticsWorkerUsage(taskFunctionWorkerUsage
, message
)
1424 updateRunTimeWorkerUsage(
1425 this.workerChoiceStrategiesContext
,
1426 taskFunctionWorkerUsage
,
1429 updateEluWorkerUsage(
1430 this.workerChoiceStrategiesContext
,
1431 taskFunctionWorkerUsage
,
1434 needWorkerChoiceStrategiesUpdate
= true
1436 if (needWorkerChoiceStrategiesUpdate
) {
1437 this.workerChoiceStrategiesContext
?.update(workerNodeKey
)
1442 * Whether the worker node shall update its task function worker usage or not.
1444 * @param workerNodeKey - The worker node key.
1445 * @returns `true` if the worker node shall update its task function worker usage, `false` otherwise.
1447 private shallUpdateTaskFunctionWorkerUsage (workerNodeKey
: number): boolean {
1448 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1450 workerInfo
!= null &&
1451 Array.isArray(workerInfo
.taskFunctionsProperties
) &&
1452 workerInfo
.taskFunctionsProperties
.length
> 2
1457 * Chooses a worker node for the next task.
1459 * @param name - The task function name.
1460 * @returns The chosen worker node key
1462 private chooseWorkerNode (name
?: string): number {
1463 if (this.shallCreateDynamicWorker()) {
1464 const workerNodeKey
= this.createAndSetupDynamicWorkerNode()
1466 this.workerChoiceStrategiesContext
?.getPolicy().dynamicWorkerUsage
===
1469 return workerNodeKey
1472 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1473 return this.workerChoiceStrategiesContext
!.execute(
1474 this.getTaskFunctionWorkerChoiceStrategy(name
)
1479 * Conditions for dynamic worker creation.
1481 * @returns Whether to create a dynamic worker or not.
1483 protected abstract shallCreateDynamicWorker (): boolean
1486 * Sends a message to worker given its worker node key.
1488 * @param workerNodeKey - The worker node key.
1489 * @param message - The message.
1490 * @param transferList - The optional array of transferable objects.
1492 protected abstract sendToWorker (
1493 workerNodeKey
: number,
1494 message
: MessageValue
<Data
>,
1495 transferList
?: readonly TransferListItem
[]
1499 * Initializes the worker node usage with sensible default values gathered during runtime.
1501 * @param workerNode - The worker node.
1503 private initWorkerNodeUsage (workerNode
: IWorkerNode
<Worker
, Data
>): void {
1505 this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
1506 .runTime
.aggregate
=== true
1508 workerNode
.usage
.runTime
.aggregate
= min(
1509 ...this.workerNodes
.map(
1511 workerNode
.usage
.runTime
.aggregate
?? Number.POSITIVE_INFINITY
1516 this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
1517 .waitTime
.aggregate
=== true
1519 workerNode
.usage
.waitTime
.aggregate
= min(
1520 ...this.workerNodes
.map(
1522 workerNode
.usage
.waitTime
.aggregate
?? Number.POSITIVE_INFINITY
1527 this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements().elu
1530 workerNode
.usage
.elu
.active
.aggregate
= min(
1531 ...this.workerNodes
.map(
1533 workerNode
.usage
.elu
.active
.aggregate
?? Number.POSITIVE_INFINITY
1540 * Creates a new, completely set up worker node.
1542 * @returns New, completely set up worker node key.
1544 protected createAndSetupWorkerNode (): number {
1545 const workerNode
= this.createWorkerNode()
1546 workerNode
.registerWorkerEventHandler(
1548 this.opts
.onlineHandler
?? EMPTY_FUNCTION
1550 workerNode
.registerWorkerEventHandler(
1552 this.opts
.messageHandler
?? EMPTY_FUNCTION
1554 workerNode
.registerWorkerEventHandler(
1556 this.opts
.errorHandler
?? EMPTY_FUNCTION
1558 workerNode
.registerOnceWorkerEventHandler('error', (error
: Error) => {
1559 workerNode
.info
.ready
= false
1560 this.emitter
?.emit(PoolEvents
.error
, error
)
1564 this.opts
.restartWorkerOnError
=== true
1566 if (workerNode
.info
.dynamic
) {
1567 this.createAndSetupDynamicWorkerNode()
1568 } else if (!this.startingMinimumNumberOfWorkers
) {
1569 this.startMinimumNumberOfWorkers(true)
1575 this.opts
.enableTasksQueue
=== true
1577 this.redistributeQueuedTasks(this.workerNodes
.indexOf(workerNode
))
1579 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1580 workerNode
?.terminate().catch((error
: unknown
) => {
1581 this.emitter
?.emit(PoolEvents
.error
, error
)
1584 workerNode
.registerWorkerEventHandler(
1586 this.opts
.exitHandler
?? EMPTY_FUNCTION
1588 workerNode
.registerOnceWorkerEventHandler('exit', () => {
1589 this.removeWorkerNode(workerNode
)
1592 !this.startingMinimumNumberOfWorkers
&&
1595 this.startMinimumNumberOfWorkers(true)
1598 const workerNodeKey
= this.addWorkerNode(workerNode
)
1599 this.afterWorkerNodeSetup(workerNodeKey
)
1600 return workerNodeKey
1604 * Creates a new, completely set up dynamic worker node.
1606 * @returns New, completely set up dynamic worker node key.
1608 protected createAndSetupDynamicWorkerNode (): number {
1609 const workerNodeKey
= this.createAndSetupWorkerNode()
1610 this.registerWorkerMessageListener(workerNodeKey
, message
=> {
1611 this.checkMessageWorkerId(message
)
1612 const localWorkerNodeKey
= this.getWorkerNodeKeyByWorkerId(
1615 const workerInfo
= this.getWorkerInfo(localWorkerNodeKey
)
1616 const workerUsage
= this.workerNodes
[localWorkerNodeKey
]?.usage
1617 // Kill message received from worker
1619 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
1620 (isKillBehavior(KillBehaviors
.SOFT
, message
.kill
) &&
1621 ((this.opts
.enableTasksQueue
=== false &&
1622 workerUsage
.tasks
.executing
=== 0) ||
1623 (this.opts
.enableTasksQueue
=== true &&
1624 workerInfo
!= null &&
1625 !workerInfo
.stealing
&&
1626 workerUsage
.tasks
.executing
=== 0 &&
1627 this.tasksQueueSize(localWorkerNodeKey
) === 0)))
1629 // Flag the worker node as not ready immediately
1630 this.flagWorkerNodeAsNotReady(localWorkerNodeKey
)
1631 this.destroyWorkerNode(localWorkerNodeKey
).catch((error
: unknown
) => {
1632 this.emitter
?.emit(PoolEvents
.error
, error
)
1636 this.sendToWorker(workerNodeKey
, {
1639 if (this.taskFunctions
.size
> 0) {
1640 for (const [taskFunctionName
, taskFunctionObject
] of this.taskFunctions
) {
1641 this.sendTaskFunctionOperationToWorker(workerNodeKey
, {
1642 taskFunctionOperation
: 'add',
1643 taskFunctionProperties
: buildTaskFunctionProperties(
1647 taskFunction
: taskFunctionObject
.taskFunction
.toString()
1648 }).catch((error
: unknown
) => {
1649 this.emitter
?.emit(PoolEvents
.error
, error
)
1653 const workerNode
= this.workerNodes
[workerNodeKey
]
1654 workerNode
.info
.dynamic
= true
1656 this.workerChoiceStrategiesContext
?.getPolicy().dynamicWorkerReady
===
1658 this.workerChoiceStrategiesContext
?.getPolicy().dynamicWorkerUsage
===
1661 workerNode
.info
.ready
= true
1663 this.initWorkerNodeUsage(workerNode
)
1664 this.checkAndEmitDynamicWorkerCreationEvents()
1665 return workerNodeKey
1669 * Registers a listener callback on the worker given its worker node key.
1671 * @param workerNodeKey - The worker node key.
1672 * @param listener - The message listener callback.
1674 protected abstract registerWorkerMessageListener
<
1675 Message
extends Data
| Response
1677 workerNodeKey
: number,
1678 listener
: (message
: MessageValue
<Message
>) => void
1682 * Registers once a listener callback on the worker given its worker node key.
1684 * @param workerNodeKey - The worker node key.
1685 * @param listener - The message listener callback.
1687 protected abstract registerOnceWorkerMessageListener
<
1688 Message
extends Data
| Response
1690 workerNodeKey
: number,
1691 listener
: (message
: MessageValue
<Message
>) => void
1695 * Deregisters a listener callback on the worker given its worker node key.
1697 * @param workerNodeKey - The worker node key.
1698 * @param listener - The message listener callback.
1700 protected abstract deregisterWorkerMessageListener
<
1701 Message
extends Data
| Response
1703 workerNodeKey
: number,
1704 listener
: (message
: MessageValue
<Message
>) => void
1708 * Method hooked up after a worker node has been newly created.
1709 * Can be overridden.
1711 * @param workerNodeKey - The newly created worker node key.
1713 protected afterWorkerNodeSetup (workerNodeKey
: number): void {
1714 // Listen to worker messages.
1715 this.registerWorkerMessageListener(
1717 this.workerMessageListener
1719 // Send the startup message to worker.
1720 this.sendStartupMessageToWorker(workerNodeKey
)
1721 // Send the statistics message to worker.
1722 this.sendStatisticsMessageToWorker(workerNodeKey
)
1723 if (this.opts
.enableTasksQueue
=== true) {
1724 if (this.opts
.tasksQueueOptions
?.taskStealing
=== true) {
1725 this.workerNodes
[workerNodeKey
].on(
1727 this.handleWorkerNodeIdleEvent
1730 if (this.opts
.tasksQueueOptions
?.tasksStealingOnBackPressure
=== true) {
1731 this.workerNodes
[workerNodeKey
].on(
1733 this.handleWorkerNodeBackPressureEvent
1740 * Sends the startup message to worker given its worker node key.
1742 * @param workerNodeKey - The worker node key.
1744 protected abstract sendStartupMessageToWorker (workerNodeKey
: number): void
1747 * Sends the statistics message to worker given its worker node key.
1749 * @param workerNodeKey - The worker node key.
1751 private sendStatisticsMessageToWorker (workerNodeKey
: number): void {
1752 this.sendToWorker(workerNodeKey
, {
1755 this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
1756 .runTime
.aggregate
?? false,
1758 this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
1759 .elu
.aggregate
?? false
1764 private cannotStealTask (): boolean {
1765 return this.workerNodes
.length
<= 1 || this.info
.queuedTasks
=== 0
1768 private handleTask (workerNodeKey
: number, task
: Task
<Data
>): void {
1769 if (this.shallExecuteTask(workerNodeKey
)) {
1770 this.executeTask(workerNodeKey
, task
)
1772 this.enqueueTask(workerNodeKey
, task
)
1776 private redistributeQueuedTasks (sourceWorkerNodeKey
: number): void {
1777 if (sourceWorkerNodeKey
=== -1 || this.cannotStealTask()) {
1780 while (this.tasksQueueSize(sourceWorkerNodeKey
) > 0) {
1781 const destinationWorkerNodeKey
= this.workerNodes
.reduce(
1782 (minWorkerNodeKey
, workerNode
, workerNodeKey
, workerNodes
) => {
1783 return sourceWorkerNodeKey
!== workerNodeKey
&&
1784 workerNode
.info
.ready
&&
1785 workerNode
.usage
.tasks
.queued
<
1786 workerNodes
[minWorkerNodeKey
].usage
.tasks
.queued
1793 destinationWorkerNodeKey
,
1794 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1795 this.dequeueTask(sourceWorkerNodeKey
)!
1800 private updateTaskStolenStatisticsWorkerUsage (
1801 workerNodeKey
: number,
1804 const workerNode
= this.workerNodes
[workerNodeKey
]
1805 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1806 if (workerNode
?.usage
!= null) {
1807 ++workerNode
.usage
.tasks
.stolen
1810 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1811 workerNode
.getTaskFunctionWorkerUsage(taskName
) != null
1813 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1814 ++workerNode
.getTaskFunctionWorkerUsage(taskName
)!.tasks
.stolen
1818 private updateTaskSequentiallyStolenStatisticsWorkerUsage (
1819 workerNodeKey
: number,
1821 previousTaskName
?: string
1823 const workerNode
= this.workerNodes
[workerNodeKey
]
1824 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1825 if (workerNode
?.usage
!= null) {
1826 ++workerNode
.usage
.tasks
.sequentiallyStolen
1829 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1830 workerNode
.getTaskFunctionWorkerUsage(taskName
) != null
1832 const taskFunctionWorkerUsage
=
1833 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1834 workerNode
.getTaskFunctionWorkerUsage(taskName
)!
1836 taskFunctionWorkerUsage
.tasks
.sequentiallyStolen
=== 0 ||
1837 (previousTaskName
!= null &&
1838 previousTaskName
=== taskName
&&
1839 taskFunctionWorkerUsage
.tasks
.sequentiallyStolen
> 0)
1841 ++taskFunctionWorkerUsage
.tasks
.sequentiallyStolen
1842 } else if (taskFunctionWorkerUsage
.tasks
.sequentiallyStolen
> 0) {
1843 taskFunctionWorkerUsage
.tasks
.sequentiallyStolen
= 0
1848 private resetTaskSequentiallyStolenStatisticsWorkerUsage (
1849 workerNodeKey
: number,
1852 const workerNode
= this.workerNodes
[workerNodeKey
]
1853 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1854 if (workerNode
?.usage
!= null) {
1855 workerNode
.usage
.tasks
.sequentiallyStolen
= 0
1858 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1859 workerNode
.getTaskFunctionWorkerUsage(taskName
) != null
1861 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1862 workerNode
.getTaskFunctionWorkerUsage(
1864 )!.tasks
.sequentiallyStolen
= 0
1868 private readonly handleWorkerNodeIdleEvent
= (
1869 eventDetail
: WorkerNodeEventDetail
,
1870 previousStolenTask
?: Task
<Data
>
1872 const { workerNodeKey
} = eventDetail
1873 if (workerNodeKey
== null) {
1875 "WorkerNode event detail 'workerNodeKey' property must be defined"
1878 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1879 if (workerInfo
== null) {
1881 `Worker node with key '${workerNodeKey}' not found in pool`
1885 this.cannotStealTask() ||
1886 (this.info
.stealingWorkerNodes
?? 0) >
1887 Math.floor(this.workerNodes
.length
/ 2)
1889 if (previousStolenTask
!= null) {
1890 workerInfo
.stealing
= false
1891 this.resetTaskSequentiallyStolenStatisticsWorkerUsage(
1893 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1894 previousStolenTask
.name
!
1899 const workerNodeTasksUsage
= this.workerNodes
[workerNodeKey
].usage
.tasks
1901 previousStolenTask
!= null &&
1902 (workerNodeTasksUsage
.executing
> 0 ||
1903 this.tasksQueueSize(workerNodeKey
) > 0)
1905 workerInfo
.stealing
= false
1906 this.resetTaskSequentiallyStolenStatisticsWorkerUsage(
1908 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1909 previousStolenTask
.name
!
1913 workerInfo
.stealing
= true
1914 const stolenTask
= this.workerNodeStealTask(workerNodeKey
)
1915 if (stolenTask
!= null) {
1916 this.updateTaskSequentiallyStolenStatisticsWorkerUsage(
1918 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1920 previousStolenTask
?.name
1923 sleep(exponentialDelay(workerNodeTasksUsage
.sequentiallyStolen
))
1925 this.handleWorkerNodeIdleEvent(eventDetail
, stolenTask
)
1928 .catch((error
: unknown
) => {
1929 this.emitter
?.emit(PoolEvents
.error
, error
)
1933 private readonly workerNodeStealTask
= (
1934 workerNodeKey
: number
1935 ): Task
<Data
> | undefined => {
1936 const workerNodes
= this.workerNodes
1939 (workerNodeA
, workerNodeB
) =>
1940 workerNodeB
.usage
.tasks
.queued
- workerNodeA
.usage
.tasks
.queued
1942 const sourceWorkerNode
= workerNodes
.find(
1943 (sourceWorkerNode
, sourceWorkerNodeKey
) =>
1944 sourceWorkerNode
.info
.ready
&&
1945 !sourceWorkerNode
.info
.stealing
&&
1946 sourceWorkerNodeKey
!== workerNodeKey
&&
1947 sourceWorkerNode
.usage
.tasks
.queued
> 0
1949 if (sourceWorkerNode
!= null) {
1950 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1951 const task
= sourceWorkerNode
.dequeueLastPrioritizedTask()!
1952 this.handleTask(workerNodeKey
, task
)
1953 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1954 this.updateTaskStolenStatisticsWorkerUsage(workerNodeKey
, task
.name
!)
1959 private readonly handleWorkerNodeBackPressureEvent
= (
1960 eventDetail
: WorkerNodeEventDetail
1963 this.cannotStealTask() ||
1964 this.hasBackPressure() ||
1965 (this.info
.stealingWorkerNodes
?? 0) >
1966 Math.floor(this.workerNodes
.length
/ 2)
1970 const sizeOffset
= 1
1971 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1972 if (this.opts
.tasksQueueOptions
!.size
! <= sizeOffset
) {
1975 const { workerId
} = eventDetail
1976 const sourceWorkerNode
=
1977 this.workerNodes
[this.getWorkerNodeKeyByWorkerId(workerId
)]
1978 const workerNodes
= this.workerNodes
1981 (workerNodeA
, workerNodeB
) =>
1982 workerNodeA
.usage
.tasks
.queued
- workerNodeB
.usage
.tasks
.queued
1984 for (const [workerNodeKey
, workerNode
] of workerNodes
.entries()) {
1986 sourceWorkerNode
.usage
.tasks
.queued
> 0 &&
1987 workerNode
.info
.ready
&&
1988 !workerNode
.info
.stealing
&&
1989 workerNode
.info
.id
!== workerId
&&
1990 workerNode
.usage
.tasks
.queued
<
1991 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1992 this.opts
.tasksQueueOptions
!.size
! - sizeOffset
1994 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1995 if (workerInfo
== null) {
1997 `Worker node with key '${workerNodeKey}' not found in pool`
2000 workerInfo
.stealing
= true
2001 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2002 const task
= sourceWorkerNode
.dequeueLastPrioritizedTask()!
2003 this.handleTask(workerNodeKey
, task
)
2004 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2005 this.updateTaskStolenStatisticsWorkerUsage(workerNodeKey
, task
.name
!)
2006 workerInfo
.stealing
= false
2012 * This method is the message listener registered on each worker.
2014 protected readonly workerMessageListener
= (
2015 message
: MessageValue
<Response
>
2017 this.checkMessageWorkerId(message
)
2018 const { workerId
, ready
, taskId
, taskFunctionsProperties
} = message
2019 if (ready
!= null && taskFunctionsProperties
!= null) {
2020 // Worker ready response received from worker
2021 this.handleWorkerReadyResponse(message
)
2022 } else if (taskFunctionsProperties
!= null) {
2023 // Task function properties message received from worker
2024 const workerNodeKey
= this.getWorkerNodeKeyByWorkerId(workerId
)
2025 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
2026 if (workerInfo
!= null) {
2027 workerInfo
.taskFunctionsProperties
= taskFunctionsProperties
2028 this.sendStatisticsMessageToWorker(workerNodeKey
)
2030 } else if (taskId
!= null) {
2031 // Task execution response received from worker
2032 this.handleTaskExecutionResponse(message
)
2036 private checkAndEmitReadyEvent (): void {
2037 if (!this.readyEventEmitted
&& this.ready
) {
2038 this.emitter
?.emit(PoolEvents
.ready
, this.info
)
2039 this.readyEventEmitted
= true
2043 private handleWorkerReadyResponse (message
: MessageValue
<Response
>): void {
2044 const { workerId
, ready
, taskFunctionsProperties
} = message
2045 if (ready
== null || !ready
) {
2046 throw new Error(`Worker ${workerId} failed to initialize`)
2048 const workerNodeKey
= this.getWorkerNodeKeyByWorkerId(workerId
)
2049 const workerNode
= this.workerNodes
[workerNodeKey
]
2050 workerNode
.info
.ready
= ready
2051 workerNode
.info
.taskFunctionsProperties
= taskFunctionsProperties
2052 this.sendStatisticsMessageToWorker(workerNodeKey
)
2053 this.checkAndEmitReadyEvent()
2056 private handleTaskExecutionResponse (message
: MessageValue
<Response
>): void {
2057 const { workerId
, taskId
, workerError
, data
} = message
2058 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2059 const promiseResponse
= this.promiseResponseMap
.get(taskId
!)
2060 if (promiseResponse
!= null) {
2061 const { resolve
, reject
, workerNodeKey
, asyncResource
} = promiseResponse
2062 const workerNode
= this.workerNodes
[workerNodeKey
]
2063 if (workerError
!= null) {
2064 this.emitter
?.emit(PoolEvents
.taskError
, workerError
)
2065 asyncResource
!= null
2066 ? asyncResource
.runInAsyncScope(
2071 : reject(workerError
.message
)
2073 asyncResource
!= null
2074 ? asyncResource
.runInAsyncScope(resolve
, this.emitter
, data
)
2075 : resolve(data
as Response
)
2077 asyncResource
?.emitDestroy()
2078 this.afterTaskExecutionHook(workerNodeKey
, message
)
2079 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2080 this.promiseResponseMap
.delete(taskId
!)
2081 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
2082 workerNode
?.emit('taskFinished', taskId
)
2084 this.opts
.enableTasksQueue
=== true &&
2086 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
2089 const workerNodeTasksUsage
= workerNode
.usage
.tasks
2091 this.tasksQueueSize(workerNodeKey
) > 0 &&
2092 workerNodeTasksUsage
.executing
<
2093 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2094 this.opts
.tasksQueueOptions
!.concurrency
!
2096 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2097 this.executeTask(workerNodeKey
, this.dequeueTask(workerNodeKey
)!)
2100 workerNodeTasksUsage
.executing
=== 0 &&
2101 this.tasksQueueSize(workerNodeKey
) === 0 &&
2102 workerNodeTasksUsage
.sequentiallyStolen
=== 0
2104 workerNode
.emit('idle', {
2113 private checkAndEmitTaskExecutionEvents (): void {
2115 this.emitter
?.emit(PoolEvents
.busy
, this.info
)
2119 private checkAndEmitTaskQueuingEvents (): void {
2120 if (this.hasBackPressure()) {
2121 this.emitter
?.emit(PoolEvents
.backPressure
, this.info
)
2126 * Emits dynamic worker creation events.
2128 protected abstract checkAndEmitDynamicWorkerCreationEvents (): void
2131 * Gets the worker information given its worker node key.
2133 * @param workerNodeKey - The worker node key.
2134 * @returns The worker information.
2136 protected getWorkerInfo (workerNodeKey
: number): WorkerInfo
| undefined {
2137 return this.workerNodes
[workerNodeKey
]?.info
2141 * Creates a worker node.
2143 * @returns The created worker node.
2145 private createWorkerNode (): IWorkerNode
<Worker
, Data
> {
2146 const workerNode
= new WorkerNode
<Worker
, Data
>(
2151 workerOptions
: this.opts
.workerOptions
,
2152 tasksQueueBackPressureSize
:
2153 this.opts
.tasksQueueOptions
?.size
??
2154 getDefaultTasksQueueOptions(
2155 this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
2157 tasksQueueBucketSize
:
2158 (this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
) * 2
2161 // Flag the worker node as ready at pool startup.
2162 if (this.starting
) {
2163 workerNode
.info
.ready
= true
2169 * Adds the given worker node in the pool worker nodes.
2171 * @param workerNode - The worker node.
2172 * @returns The added worker node key.
2173 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the added worker node is not found.
2175 private addWorkerNode (workerNode
: IWorkerNode
<Worker
, Data
>): number {
2176 this.workerNodes
.push(workerNode
)
2177 const workerNodeKey
= this.workerNodes
.indexOf(workerNode
)
2178 if (workerNodeKey
=== -1) {
2179 throw new Error('Worker added not found in worker nodes')
2181 return workerNodeKey
2184 private checkAndEmitEmptyEvent (): void {
2186 this.emitter
?.emit(PoolEvents
.empty
, this.info
)
2187 this.readyEventEmitted
= false
2192 * Removes the worker node from the pool worker nodes.
2194 * @param workerNode - The worker node.
2196 private removeWorkerNode (workerNode
: IWorkerNode
<Worker
, Data
>): void {
2197 const workerNodeKey
= this.workerNodes
.indexOf(workerNode
)
2198 if (workerNodeKey
!== -1) {
2199 this.workerNodes
.splice(workerNodeKey
, 1)
2200 this.workerChoiceStrategiesContext
?.remove(workerNodeKey
)
2202 this.checkAndEmitEmptyEvent()
2205 protected flagWorkerNodeAsNotReady (workerNodeKey
: number): void {
2206 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
2207 if (workerInfo
!= null) {
2208 workerInfo
.ready
= false
2212 private hasBackPressure (): boolean {
2214 this.opts
.enableTasksQueue
=== true &&
2215 this.workerNodes
.findIndex(
2216 workerNode
=> !workerNode
.hasBackPressure()
2222 * Executes the given task on the worker given its worker node key.
2224 * @param workerNodeKey - The worker node key.
2225 * @param task - The task to execute.
2227 private executeTask (workerNodeKey
: number, task
: Task
<Data
>): void {
2228 this.beforeTaskExecutionHook(workerNodeKey
, task
)
2229 this.sendToWorker(workerNodeKey
, task
, task
.transferList
)
2230 this.checkAndEmitTaskExecutionEvents()
2233 private enqueueTask (workerNodeKey
: number, task
: Task
<Data
>): number {
2234 const tasksQueueSize
= this.workerNodes
[workerNodeKey
].enqueueTask(task
)
2235 this.checkAndEmitTaskQueuingEvents()
2236 return tasksQueueSize
2239 private dequeueTask (workerNodeKey
: number): Task
<Data
> | undefined {
2240 return this.workerNodes
[workerNodeKey
].dequeueTask()
2243 private tasksQueueSize (workerNodeKey
: number): number {
2244 return this.workerNodes
[workerNodeKey
].tasksQueueSize()
2247 protected flushTasksQueue (workerNodeKey
: number): number {
2248 let flushedTasks
= 0
2249 while (this.tasksQueueSize(workerNodeKey
) > 0) {
2250 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2251 this.executeTask(workerNodeKey
, this.dequeueTask(workerNodeKey
)!)
2254 this.workerNodes
[workerNodeKey
].clearTasksQueue()
2258 private flushTasksQueues (): void {
2259 for (const workerNodeKey
of this.workerNodes
.keys()) {
2260 this.flushTasksQueue(workerNodeKey
)