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'
7 import { defaultBucketSize
} from
'../priority-queue.js'
10 PromiseResponseWrapper
,
12 TaskFunctionProperties
13 } from
'../utility-types.js'
16 buildTaskFunctionProperties
,
31 } from
'../worker/task-functions.js'
32 import { KillBehaviors
} from
'../worker/worker-options.js'
40 type TasksQueueOptions
44 WorkerChoiceStrategies
,
45 type WorkerChoiceStrategy
,
46 type WorkerChoiceStrategyOptions
47 } from
'./selection-strategies/selection-strategies-types.js'
48 import { WorkerChoiceStrategiesContext
} from
'./selection-strategies/worker-choice-strategies-context.js'
52 checkValidTasksQueueOptions
,
53 checkValidWorkerChoiceStrategy
,
54 getDefaultTasksQueueOptions
,
56 updateRunTimeWorkerUsage
,
57 updateTaskStatisticsWorkerUsage
,
58 updateWaitTimeWorkerUsage
,
61 import { version
} from
'./version.js'
66 WorkerNodeEventDetail
,
69 import { WorkerNode
} from
'./worker-node.js'
72 * Base class that implements some shared logic for all poolifier pools.
74 * @typeParam Worker - Type of worker which manages this pool.
75 * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
76 * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
78 export abstract class AbstractPool
<
79 Worker
extends IWorker
,
82 > implements IPool
<Worker
, Data
, Response
> {
84 public readonly workerNodes
: Array<IWorkerNode
<Worker
, Data
>> = []
87 public emitter
?: EventEmitterAsyncResource
90 * The task execution response promise map:
91 * - `key`: The message id of each submitted task.
92 * - `value`: An object that contains task's worker node key, execution response promise resolve and reject callbacks, async resource.
94 * When we receive a message from the worker, we get a map entry with the promise resolve/reject bound to the message id.
96 protected promiseResponseMap
: Map
<
97 `${string}-${string}-${string}-${string}-${string}`,
98 PromiseResponseWrapper
<Response
>
100 `${string}-${string}-${string}-${string}-${string}`,
101 PromiseResponseWrapper
<Response
>
105 * Worker choice strategies context referencing worker choice algorithms implementation.
107 protected workerChoiceStrategiesContext
?: WorkerChoiceStrategiesContext
<
114 * The task functions added at runtime map:
115 * - `key`: The task function name.
116 * - `value`: The task function object.
118 private readonly taskFunctions
: Map
<
120 TaskFunctionObject
<Data
, Response
>
124 * Whether the pool is started or not.
126 private started
: boolean
128 * Whether the pool is starting or not.
130 private starting
: boolean
132 * Whether the pool is destroying or not.
134 private destroying
: boolean
136 * Whether the minimum number of workers is starting or not.
138 private startingMinimumNumberOfWorkers
: boolean
140 * Whether the pool ready event has been emitted or not.
142 private readyEventEmitted
: boolean
144 * The start timestamp of the pool.
146 private startTimestamp
?: number
149 * Constructs a new poolifier pool.
151 * @param minimumNumberOfWorkers - Minimum number of workers that this pool manages.
152 * @param filePath - Path to the worker file.
153 * @param opts - Options for the pool.
154 * @param maximumNumberOfWorkers - Maximum number of workers that this pool manages.
157 protected readonly minimumNumberOfWorkers
: number,
158 protected readonly filePath
: string,
159 protected readonly opts
: PoolOptions
<Worker
>,
160 protected readonly maximumNumberOfWorkers
?: number
162 if (!this.isMain()) {
164 'Cannot start a pool from a worker with the same type as the pool'
168 checkFilePath(this.filePath
)
169 this.checkMinimumNumberOfWorkers(this.minimumNumberOfWorkers
)
170 this.checkPoolOptions(this.opts
)
172 this.chooseWorkerNode
= this.chooseWorkerNode
.bind(this)
173 this.executeTask
= this.executeTask
.bind(this)
174 this.enqueueTask
= this.enqueueTask
.bind(this)
176 if (this.opts
.enableEvents
=== true) {
177 this.initEventEmitter()
179 this.workerChoiceStrategiesContext
= new WorkerChoiceStrategiesContext
<
185 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
186 [this.opts
.workerChoiceStrategy
!],
187 this.opts
.workerChoiceStrategyOptions
192 this.taskFunctions
= new Map
<string, TaskFunctionObject
<Data
, Response
>>()
195 this.starting
= false
196 this.destroying
= false
197 this.readyEventEmitted
= false
198 this.startingMinimumNumberOfWorkers
= false
199 if (this.opts
.startWorkers
=== true) {
204 private checkPoolType (): void {
205 if (this.type === PoolTypes
.fixed
&& this.maximumNumberOfWorkers
!= null) {
207 'Cannot instantiate a fixed pool with a maximum number of workers specified at initialization'
212 private checkMinimumNumberOfWorkers (
213 minimumNumberOfWorkers
: number | undefined
215 if (minimumNumberOfWorkers
== null) {
217 'Cannot instantiate a pool without specifying the number of workers'
219 } else if (!Number.isSafeInteger(minimumNumberOfWorkers
)) {
221 'Cannot instantiate a pool with a non safe integer number of workers'
223 } else if (minimumNumberOfWorkers
< 0) {
224 throw new RangeError(
225 'Cannot instantiate a pool with a negative number of workers'
227 } else if (this.type === PoolTypes
.fixed
&& minimumNumberOfWorkers
=== 0) {
228 throw new RangeError('Cannot instantiate a fixed pool with zero worker')
232 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
233 if (isPlainObject(opts
)) {
234 this.opts
.startWorkers
= opts
.startWorkers
?? true
235 checkValidWorkerChoiceStrategy(opts
.workerChoiceStrategy
)
236 this.opts
.workerChoiceStrategy
=
237 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
238 this.checkValidWorkerChoiceStrategyOptions(
239 opts
.workerChoiceStrategyOptions
241 if (opts
.workerChoiceStrategyOptions
!= null) {
242 this.opts
.workerChoiceStrategyOptions
= opts
.workerChoiceStrategyOptions
244 this.opts
.restartWorkerOnError
= opts
.restartWorkerOnError
?? true
245 this.opts
.enableEvents
= opts
.enableEvents
?? true
246 this.opts
.enableTasksQueue
= opts
.enableTasksQueue
?? false
247 if (this.opts
.enableTasksQueue
) {
248 checkValidTasksQueueOptions(opts
.tasksQueueOptions
)
249 this.opts
.tasksQueueOptions
= this.buildTasksQueueOptions(
250 opts
.tasksQueueOptions
254 throw new TypeError('Invalid pool options: must be a plain object')
258 private checkValidWorkerChoiceStrategyOptions (
259 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
| undefined
262 workerChoiceStrategyOptions
!= null &&
263 !isPlainObject(workerChoiceStrategyOptions
)
266 'Invalid worker choice strategy options: must be a plain object'
270 workerChoiceStrategyOptions
?.weights
!= null &&
271 Object.keys(workerChoiceStrategyOptions
.weights
).length
!==
272 (this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
)
275 'Invalid worker choice strategy options: must have a weight for each worker node'
279 workerChoiceStrategyOptions
?.measurement
!= null &&
280 !Object.values(Measurements
).includes(
281 workerChoiceStrategyOptions
.measurement
285 `Invalid worker choice strategy options: invalid measurement '${workerChoiceStrategyOptions.measurement}'`
290 private initEventEmitter (): void {
291 this.emitter
= new EventEmitterAsyncResource({
292 name
: `poolifier:${this.type}-${this.worker}-pool`
297 public get
info (): PoolInfo
{
302 started
: this.started
,
304 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
305 defaultStrategy
: this.opts
.workerChoiceStrategy
!,
306 strategyRetries
: this.workerChoiceStrategiesContext
?.retriesCount
?? 0,
307 minSize
: this.minimumNumberOfWorkers
,
308 maxSize
: this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
,
309 ...(this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
310 .runTime
.aggregate
=== true &&
311 this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
312 .waitTime
.aggregate
&& {
313 utilization
: round(this.utilization
)
315 workerNodes
: this.workerNodes
.length
,
316 idleWorkerNodes
: this.workerNodes
.reduce(
317 (accumulator
, workerNode
) =>
318 workerNode
.usage
.tasks
.executing
=== 0
323 ...(this.opts
.enableTasksQueue
=== true && {
324 stealingWorkerNodes
: this.workerNodes
.reduce(
325 (accumulator
, workerNode
) =>
326 workerNode
.info
.stealing
? accumulator
+ 1 : accumulator
,
330 busyWorkerNodes
: this.workerNodes
.reduce(
331 (accumulator
, _
, workerNodeKey
) =>
332 this.isWorkerNodeBusy(workerNodeKey
) ? accumulator
+ 1 : accumulator
,
335 executedTasks
: this.workerNodes
.reduce(
336 (accumulator
, workerNode
) =>
337 accumulator
+ workerNode
.usage
.tasks
.executed
,
340 executingTasks
: this.workerNodes
.reduce(
341 (accumulator
, workerNode
) =>
342 accumulator
+ workerNode
.usage
.tasks
.executing
,
345 ...(this.opts
.enableTasksQueue
=== true && {
346 queuedTasks
: this.workerNodes
.reduce(
347 (accumulator
, workerNode
) =>
348 accumulator
+ workerNode
.usage
.tasks
.queued
,
352 ...(this.opts
.enableTasksQueue
=== true && {
353 maxQueuedTasks
: this.workerNodes
.reduce(
354 (accumulator
, workerNode
) =>
355 accumulator
+ (workerNode
.usage
.tasks
.maxQueued
?? 0),
359 ...(this.opts
.enableTasksQueue
=== true && {
360 backPressure
: this.hasBackPressure()
362 ...(this.opts
.enableTasksQueue
=== true && {
363 stolenTasks
: this.workerNodes
.reduce(
364 (accumulator
, workerNode
) =>
365 accumulator
+ workerNode
.usage
.tasks
.stolen
,
369 failedTasks
: this.workerNodes
.reduce(
370 (accumulator
, workerNode
) =>
371 accumulator
+ workerNode
.usage
.tasks
.failed
,
374 ...(this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
375 .runTime
.aggregate
=== true && {
379 ...this.workerNodes
.map(
381 workerNode
.usage
.runTime
.minimum
?? Number.POSITIVE_INFINITY
387 ...this.workerNodes
.map(
389 workerNode
.usage
.runTime
.maximum
?? Number.NEGATIVE_INFINITY
393 ...(this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
394 .runTime
.average
&& {
397 this.workerNodes
.reduce
<number[]>(
398 (accumulator
, workerNode
) =>
400 workerNode
.usage
.runTime
.history
.toArray()
407 ...(this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
411 this.workerNodes
.reduce
<number[]>(
412 (accumulator
, workerNode
) =>
414 workerNode
.usage
.runTime
.history
.toArray()
423 ...(this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
424 .waitTime
.aggregate
=== true && {
428 ...this.workerNodes
.map(
430 workerNode
.usage
.waitTime
.minimum
?? Number.POSITIVE_INFINITY
436 ...this.workerNodes
.map(
438 workerNode
.usage
.waitTime
.maximum
?? Number.NEGATIVE_INFINITY
442 ...(this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
443 .waitTime
.average
&& {
446 this.workerNodes
.reduce
<number[]>(
447 (accumulator
, workerNode
) =>
449 workerNode
.usage
.waitTime
.history
.toArray()
456 ...(this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
457 .waitTime
.median
&& {
460 this.workerNodes
.reduce
<number[]>(
461 (accumulator
, workerNode
) =>
463 workerNode
.usage
.waitTime
.history
.toArray()
472 ...(this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
473 .elu
.aggregate
=== true && {
478 ...this.workerNodes
.map(
480 workerNode
.usage
.elu
.idle
.minimum
??
481 Number.POSITIVE_INFINITY
487 ...this.workerNodes
.map(
489 workerNode
.usage
.elu
.idle
.maximum
??
490 Number.NEGATIVE_INFINITY
494 ...(this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
498 this.workerNodes
.reduce
<number[]>(
499 (accumulator
, workerNode
) =>
501 workerNode
.usage
.elu
.idle
.history
.toArray()
508 ...(this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
512 this.workerNodes
.reduce
<number[]>(
513 (accumulator
, workerNode
) =>
515 workerNode
.usage
.elu
.idle
.history
.toArray()
526 ...this.workerNodes
.map(
528 workerNode
.usage
.elu
.active
.minimum
??
529 Number.POSITIVE_INFINITY
535 ...this.workerNodes
.map(
537 workerNode
.usage
.elu
.active
.maximum
??
538 Number.NEGATIVE_INFINITY
542 ...(this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
546 this.workerNodes
.reduce
<number[]>(
547 (accumulator
, workerNode
) =>
549 workerNode
.usage
.elu
.active
.history
.toArray()
556 ...(this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
560 this.workerNodes
.reduce
<number[]>(
561 (accumulator
, workerNode
) =>
563 workerNode
.usage
.elu
.active
.history
.toArray()
574 this.workerNodes
.map(
575 workerNode
=> workerNode
.usage
.elu
.utilization
?? 0
581 this.workerNodes
.map(
582 workerNode
=> workerNode
.usage
.elu
.utilization
?? 0
593 * The pool readiness boolean status.
595 private get
ready (): boolean {
600 this.workerNodes
.reduce(
601 (accumulator
, workerNode
) =>
602 !workerNode
.info
.dynamic
&& workerNode
.info
.ready
606 ) >= this.minimumNumberOfWorkers
611 * The pool emptiness boolean status.
613 protected get
empty (): boolean {
614 return this.minimumNumberOfWorkers
=== 0 && this.workerNodes
.length
=== 0
618 * The approximate pool utilization.
620 * @returns The pool utilization.
622 private get
utilization (): number {
623 if (this.startTimestamp
== null) {
626 const poolTimeCapacity
=
627 (performance
.now() - this.startTimestamp
) *
628 (this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
)
629 const totalTasksRunTime
= this.workerNodes
.reduce(
630 (accumulator
, workerNode
) =>
631 accumulator
+ (workerNode
.usage
.runTime
.aggregate
?? 0),
634 const totalTasksWaitTime
= this.workerNodes
.reduce(
635 (accumulator
, workerNode
) =>
636 accumulator
+ (workerNode
.usage
.waitTime
.aggregate
?? 0),
639 return (totalTasksRunTime
+ totalTasksWaitTime
) / poolTimeCapacity
645 * If it is `'dynamic'`, it provides the `max` property.
647 protected abstract get
type (): PoolType
652 protected abstract get
worker (): WorkerType
655 * Checks if the worker id sent in the received message from a worker is valid.
657 * @param message - The received message.
658 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the worker id is invalid.
660 private checkMessageWorkerId (message
: MessageValue
<Data
| Response
>): void {
661 if (message
.workerId
== null) {
662 throw new Error('Worker message received without worker id')
663 } else if (this.getWorkerNodeKeyByWorkerId(message
.workerId
) === -1) {
665 `Worker message received from unknown worker '${message.workerId}'`
671 * Gets the worker node key given its worker id.
673 * @param workerId - The worker id.
674 * @returns The worker node key if the worker id is found in the pool worker nodes, `-1` otherwise.
676 private getWorkerNodeKeyByWorkerId (workerId
: number | undefined): number {
677 return this.workerNodes
.findIndex(
678 workerNode
=> workerNode
.info
.id
=== workerId
683 public setWorkerChoiceStrategy (
684 workerChoiceStrategy
: WorkerChoiceStrategy
,
685 workerChoiceStrategyOptions
?: WorkerChoiceStrategyOptions
687 let requireSync
= false
688 checkValidWorkerChoiceStrategy(workerChoiceStrategy
)
689 if (workerChoiceStrategyOptions
!= null) {
690 requireSync
= !this.setWorkerChoiceStrategyOptions(
691 workerChoiceStrategyOptions
694 if (workerChoiceStrategy
!== this.opts
.workerChoiceStrategy
) {
695 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
696 this.workerChoiceStrategiesContext
?.setDefaultWorkerChoiceStrategy(
697 this.opts
.workerChoiceStrategy
,
698 this.opts
.workerChoiceStrategyOptions
703 this.workerChoiceStrategiesContext
?.syncWorkerChoiceStrategies(
704 this.getWorkerChoiceStrategies(),
705 this.opts
.workerChoiceStrategyOptions
707 for (const workerNodeKey
of this.workerNodes
.keys()) {
708 this.sendStatisticsMessageToWorker(workerNodeKey
)
714 public setWorkerChoiceStrategyOptions (
715 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
| undefined
717 this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
718 if (workerChoiceStrategyOptions
!= null) {
719 this.opts
.workerChoiceStrategyOptions
= workerChoiceStrategyOptions
720 this.workerChoiceStrategiesContext
?.setOptions(
721 this.opts
.workerChoiceStrategyOptions
723 this.workerChoiceStrategiesContext
?.syncWorkerChoiceStrategies(
724 this.getWorkerChoiceStrategies(),
725 this.opts
.workerChoiceStrategyOptions
727 for (const workerNodeKey
of this.workerNodes
.keys()) {
728 this.sendStatisticsMessageToWorker(workerNodeKey
)
736 public enableTasksQueue (
738 tasksQueueOptions
?: TasksQueueOptions
740 if (this.opts
.enableTasksQueue
=== true && !enable
) {
741 this.unsetTaskStealing()
742 this.unsetTasksStealingOnBackPressure()
743 this.flushTasksQueues()
745 this.opts
.enableTasksQueue
= enable
746 this.setTasksQueueOptions(tasksQueueOptions
)
750 public setTasksQueueOptions (
751 tasksQueueOptions
: TasksQueueOptions
| undefined
753 if (this.opts
.enableTasksQueue
=== true) {
754 checkValidTasksQueueOptions(tasksQueueOptions
)
755 this.opts
.tasksQueueOptions
=
756 this.buildTasksQueueOptions(tasksQueueOptions
)
757 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
758 this.setTasksQueueSize(this.opts
.tasksQueueOptions
.size
!)
759 if (this.opts
.tasksQueueOptions
.taskStealing
=== true) {
760 this.unsetTaskStealing()
761 this.setTaskStealing()
763 this.unsetTaskStealing()
765 if (this.opts
.tasksQueueOptions
.tasksStealingOnBackPressure
=== true) {
766 this.unsetTasksStealingOnBackPressure()
767 this.setTasksStealingOnBackPressure()
769 this.unsetTasksStealingOnBackPressure()
771 } else if (this.opts
.tasksQueueOptions
!= null) {
772 delete this.opts
.tasksQueueOptions
776 private buildTasksQueueOptions (
777 tasksQueueOptions
: TasksQueueOptions
| undefined
778 ): TasksQueueOptions
{
780 ...getDefaultTasksQueueOptions(
781 this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
787 private setTasksQueueSize (size
: number): void {
788 for (const workerNode
of this.workerNodes
) {
789 workerNode
.tasksQueueBackPressureSize
= size
793 private setTaskStealing (): void {
794 for (const workerNodeKey
of this.workerNodes
.keys()) {
795 this.workerNodes
[workerNodeKey
].on('idle', this.handleWorkerNodeIdleEvent
)
799 private unsetTaskStealing (): void {
800 for (const workerNodeKey
of this.workerNodes
.keys()) {
801 this.workerNodes
[workerNodeKey
].off(
803 this.handleWorkerNodeIdleEvent
808 private setTasksStealingOnBackPressure (): void {
809 for (const workerNodeKey
of this.workerNodes
.keys()) {
810 this.workerNodes
[workerNodeKey
].on(
812 this.handleWorkerNodeBackPressureEvent
817 private unsetTasksStealingOnBackPressure (): void {
818 for (const workerNodeKey
of this.workerNodes
.keys()) {
819 this.workerNodes
[workerNodeKey
].off(
821 this.handleWorkerNodeBackPressureEvent
827 * Whether the pool is full or not.
829 * The pool filling boolean status.
831 protected get
full (): boolean {
833 this.workerNodes
.length
>=
834 (this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
)
839 * Whether the pool is busy or not.
841 * The pool busyness boolean status.
843 protected abstract get
busy (): boolean
846 * Whether worker nodes are executing concurrently their tasks quota or not.
848 * @returns Worker nodes busyness boolean status.
850 protected internalBusy (): boolean {
851 if (this.opts
.enableTasksQueue
=== true) {
853 this.workerNodes
.findIndex(
855 workerNode
.info
.ready
&&
856 workerNode
.usage
.tasks
.executing
<
857 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
858 this.opts
.tasksQueueOptions
!.concurrency
!
863 this.workerNodes
.findIndex(
865 workerNode
.info
.ready
&& workerNode
.usage
.tasks
.executing
=== 0
870 private isWorkerNodeBusy (workerNodeKey
: number): boolean {
871 if (this.opts
.enableTasksQueue
=== true) {
873 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
>=
874 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
875 this.opts
.tasksQueueOptions
!.concurrency
!
878 return this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
> 0
881 private async sendTaskFunctionOperationToWorker (
882 workerNodeKey
: number,
883 message
: MessageValue
<Data
>
884 ): Promise
<boolean> {
885 return await new Promise
<boolean>((resolve
, reject
) => {
886 const taskFunctionOperationListener
= (
887 message
: MessageValue
<Response
>
889 this.checkMessageWorkerId(message
)
890 const workerId
= this.getWorkerInfo(workerNodeKey
)?.id
892 message
.taskFunctionOperationStatus
!= null &&
893 message
.workerId
=== workerId
895 if (message
.taskFunctionOperationStatus
) {
900 `Task function operation '${message.taskFunctionOperation}' failed on worker ${message.workerId} with error: '${message.workerError?.message}'`
904 this.deregisterWorkerMessageListener(
905 this.getWorkerNodeKeyByWorkerId(message
.workerId
),
906 taskFunctionOperationListener
910 this.registerWorkerMessageListener(
912 taskFunctionOperationListener
914 this.sendToWorker(workerNodeKey
, message
)
918 private async sendTaskFunctionOperationToWorkers (
919 message
: MessageValue
<Data
>
920 ): Promise
<boolean> {
921 return await new Promise
<boolean>((resolve
, reject
) => {
922 const responsesReceived
= new Array<MessageValue
<Response
>>()
923 const taskFunctionOperationsListener
= (
924 message
: MessageValue
<Response
>
926 this.checkMessageWorkerId(message
)
927 if (message
.taskFunctionOperationStatus
!= null) {
928 responsesReceived
.push(message
)
929 if (responsesReceived
.length
=== this.workerNodes
.length
) {
931 responsesReceived
.every(
932 message
=> message
.taskFunctionOperationStatus
=== true
937 responsesReceived
.some(
938 message
=> message
.taskFunctionOperationStatus
=== false
941 const errorResponse
= responsesReceived
.find(
942 response
=> response
.taskFunctionOperationStatus
=== false
946 `Task function operation '${
947 message.taskFunctionOperation as string
948 }' failed on worker ${errorResponse?.workerId} with error: '${
949 errorResponse?.workerError?.message
954 this.deregisterWorkerMessageListener(
955 this.getWorkerNodeKeyByWorkerId(message
.workerId
),
956 taskFunctionOperationsListener
961 for (const workerNodeKey
of this.workerNodes
.keys()) {
962 this.registerWorkerMessageListener(
964 taskFunctionOperationsListener
966 this.sendToWorker(workerNodeKey
, message
)
972 public hasTaskFunction (name
: string): boolean {
973 return this.listTaskFunctionsProperties().some(
974 taskFunctionProperties
=> taskFunctionProperties
.name
=== name
979 public async addTaskFunction (
981 fn
: TaskFunction
<Data
, Response
> | TaskFunctionObject
<Data
, Response
>
982 ): Promise
<boolean> {
983 if (typeof name
!== 'string') {
984 throw new TypeError('name argument must be a string')
986 if (typeof name
=== 'string' && name
.trim().length
=== 0) {
987 throw new TypeError('name argument must not be an empty string')
989 if (typeof fn
=== 'function') {
990 fn
= { taskFunction
: fn
} satisfies TaskFunctionObject
<Data
, Response
>
992 if (typeof fn
.taskFunction
!== 'function') {
993 throw new TypeError('taskFunction property must be a function')
995 checkValidPriority(fn
.priority
)
996 checkValidWorkerChoiceStrategy(fn
.strategy
)
997 const opResult
= await this.sendTaskFunctionOperationToWorkers({
998 taskFunctionOperation
: 'add',
999 taskFunctionProperties
: buildTaskFunctionProperties(name
, fn
),
1000 taskFunction
: fn
.taskFunction
.toString()
1002 this.taskFunctions
.set(name
, fn
)
1003 this.workerChoiceStrategiesContext
?.syncWorkerChoiceStrategies(
1004 this.getWorkerChoiceStrategies()
1006 for (const workerNodeKey
of this.workerNodes
.keys()) {
1007 this.sendStatisticsMessageToWorker(workerNodeKey
)
1013 public async removeTaskFunction (name
: string): Promise
<boolean> {
1014 if (!this.taskFunctions
.has(name
)) {
1016 'Cannot remove a task function not handled on the pool side'
1019 const opResult
= await this.sendTaskFunctionOperationToWorkers({
1020 taskFunctionOperation
: 'remove',
1021 taskFunctionProperties
: buildTaskFunctionProperties(
1023 this.taskFunctions
.get(name
)
1026 for (const workerNode
of this.workerNodes
) {
1027 workerNode
.deleteTaskFunctionWorkerUsage(name
)
1029 this.taskFunctions
.delete(name
)
1030 this.workerChoiceStrategiesContext
?.syncWorkerChoiceStrategies(
1031 this.getWorkerChoiceStrategies()
1033 for (const workerNodeKey
of this.workerNodes
.keys()) {
1034 this.sendStatisticsMessageToWorker(workerNodeKey
)
1040 public listTaskFunctionsProperties (): TaskFunctionProperties
[] {
1041 for (const workerNode
of this.workerNodes
) {
1043 Array.isArray(workerNode
.info
.taskFunctionsProperties
) &&
1044 workerNode
.info
.taskFunctionsProperties
.length
> 0
1046 return workerNode
.info
.taskFunctionsProperties
1053 * Gets task function worker choice strategy, if any.
1055 * @param name - The task function name.
1056 * @returns The task function worker choice strategy if the task function worker choice strategy is defined, `undefined` otherwise.
1058 private readonly getTaskFunctionWorkerChoiceStrategy
= (
1060 ): WorkerChoiceStrategy
| undefined => {
1061 name
= name
?? DEFAULT_TASK_NAME
1062 const taskFunctionsProperties
= this.listTaskFunctionsProperties()
1063 if (name
=== DEFAULT_TASK_NAME
) {
1064 name
= taskFunctionsProperties
[1]?.name
1066 return taskFunctionsProperties
.find(
1067 (taskFunctionProperties
: TaskFunctionProperties
) =>
1068 taskFunctionProperties
.name
=== name
1073 * Gets worker node task function worker choice strategy, if any.
1075 * @param workerNodeKey - The worker node key.
1076 * @param name - The task function name.
1077 * @returns The worker node task function worker choice strategy if the worker node task function worker choice strategy is defined, `undefined` otherwise.
1079 private readonly getWorkerNodeTaskFunctionWorkerChoiceStrategy
= (
1080 workerNodeKey
: number,
1082 ): WorkerChoiceStrategy
| undefined => {
1083 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1084 if (workerInfo
== null) {
1087 name
= name
?? DEFAULT_TASK_NAME
1088 if (name
=== DEFAULT_TASK_NAME
) {
1089 name
= workerInfo
.taskFunctionsProperties
?.[1]?.name
1091 return workerInfo
.taskFunctionsProperties
?.find(
1092 (taskFunctionProperties
: TaskFunctionProperties
) =>
1093 taskFunctionProperties
.name
=== name
1098 * Gets worker node task function priority, if any.
1100 * @param workerNodeKey - The worker node key.
1101 * @param name - The task function name.
1102 * @returns The worker node task function priority if the worker node task function priority is defined, `undefined` otherwise.
1104 private readonly getWorkerNodeTaskFunctionPriority
= (
1105 workerNodeKey
: number,
1107 ): number | undefined => {
1108 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1109 if (workerInfo
== null) {
1112 name
= name
?? DEFAULT_TASK_NAME
1113 if (name
=== DEFAULT_TASK_NAME
) {
1114 name
= workerInfo
.taskFunctionsProperties
?.[1]?.name
1116 return workerInfo
.taskFunctionsProperties
?.find(
1117 (taskFunctionProperties
: TaskFunctionProperties
) =>
1118 taskFunctionProperties
.name
=== name
1123 * Gets the worker choice strategies registered in this pool.
1125 * @returns The worker choice strategies.
1127 private readonly getWorkerChoiceStrategies
=
1128 (): Set
<WorkerChoiceStrategy
> => {
1130 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1131 this.opts
.workerChoiceStrategy
!,
1132 ...(this.listTaskFunctionsProperties()
1134 (taskFunctionProperties
: TaskFunctionProperties
) =>
1135 taskFunctionProperties
.strategy
1138 (strategy
: WorkerChoiceStrategy
| undefined) => strategy
!= null
1139 ) as WorkerChoiceStrategy
[])
1144 public async setDefaultTaskFunction (name
: string): Promise
<boolean> {
1145 return await this.sendTaskFunctionOperationToWorkers({
1146 taskFunctionOperation
: 'default',
1147 taskFunctionProperties
: buildTaskFunctionProperties(
1149 this.taskFunctions
.get(name
)
1154 private shallExecuteTask (workerNodeKey
: number): boolean {
1156 this.tasksQueueSize(workerNodeKey
) === 0 &&
1157 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
<
1158 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1159 this.opts
.tasksQueueOptions
!.concurrency
!
1164 public async execute (
1167 transferList
?: readonly TransferListItem
[]
1168 ): Promise
<Response
> {
1169 return await new Promise
<Response
>((resolve
, reject
) => {
1170 if (!this.started
) {
1171 reject(new Error('Cannot execute a task on not started pool'))
1174 if (this.destroying
) {
1175 reject(new Error('Cannot execute a task on destroying pool'))
1178 if (name
!= null && typeof name
!== 'string') {
1179 reject(new TypeError('name argument must be a string'))
1184 typeof name
=== 'string' &&
1185 name
.trim().length
=== 0
1187 reject(new TypeError('name argument must not be an empty string'))
1190 if (transferList
!= null && !Array.isArray(transferList
)) {
1191 reject(new TypeError('transferList argument must be an array'))
1194 const timestamp
= performance
.now()
1195 const workerNodeKey
= this.chooseWorkerNode(name
)
1196 const task
: Task
<Data
> = {
1197 name
: name
?? DEFAULT_TASK_NAME
,
1198 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
1199 data
: data
?? ({} as Data
),
1200 priority
: this.getWorkerNodeTaskFunctionPriority(workerNodeKey
, name
),
1201 strategy
: this.getWorkerNodeTaskFunctionWorkerChoiceStrategy(
1207 taskId
: randomUUID()
1209 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1210 this.promiseResponseMap
.set(task
.taskId
!, {
1214 ...(this.emitter
!= null && {
1215 asyncResource
: new AsyncResource('poolifier:task', {
1216 triggerAsyncId
: this.emitter
.asyncId
,
1217 requireManualDestroy
: true
1222 this.opts
.enableTasksQueue
=== false ||
1223 (this.opts
.enableTasksQueue
=== true &&
1224 this.shallExecuteTask(workerNodeKey
))
1226 this.executeTask(workerNodeKey
, task
)
1228 this.enqueueTask(workerNodeKey
, task
)
1234 * Starts the minimum number of workers.
1236 private startMinimumNumberOfWorkers (initWorkerNodeUsage
= false): void {
1237 this.startingMinimumNumberOfWorkers
= true
1239 this.workerNodes
.reduce(
1240 (accumulator
, workerNode
) =>
1241 !workerNode
.info
.dynamic
? accumulator
+ 1 : accumulator
,
1243 ) < this.minimumNumberOfWorkers
1245 const workerNodeKey
= this.createAndSetupWorkerNode()
1246 initWorkerNodeUsage
&&
1247 this.initWorkerNodeUsage(this.workerNodes
[workerNodeKey
])
1249 this.startingMinimumNumberOfWorkers
= false
1253 public start (): void {
1255 throw new Error('Cannot start an already started pool')
1257 if (this.starting
) {
1258 throw new Error('Cannot start an already starting pool')
1260 if (this.destroying
) {
1261 throw new Error('Cannot start a destroying pool')
1263 this.starting
= true
1264 this.startMinimumNumberOfWorkers()
1265 this.startTimestamp
= performance
.now()
1266 this.starting
= false
1271 public async destroy (): Promise
<void> {
1272 if (!this.started
) {
1273 throw new Error('Cannot destroy an already destroyed pool')
1275 if (this.starting
) {
1276 throw new Error('Cannot destroy an starting pool')
1278 if (this.destroying
) {
1279 throw new Error('Cannot destroy an already destroying pool')
1281 this.destroying
= true
1283 this.workerNodes
.map(async (_
, workerNodeKey
) => {
1284 await this.destroyWorkerNode(workerNodeKey
)
1287 this.emitter
?.emit(PoolEvents
.destroy
, this.info
)
1288 this.emitter
?.emitDestroy()
1289 this.readyEventEmitted
= false
1290 delete this.startTimestamp
1291 this.destroying
= false
1292 this.started
= false
1295 private async sendKillMessageToWorker (workerNodeKey
: number): Promise
<void> {
1296 await new Promise
<void>((resolve
, reject
) => {
1297 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1298 if (this.workerNodes
[workerNodeKey
] == null) {
1302 const killMessageListener
= (message
: MessageValue
<Response
>): void => {
1303 this.checkMessageWorkerId(message
)
1304 if (message
.kill
=== 'success') {
1306 } else if (message
.kill
=== 'failure') {
1309 `Kill message handling failed on worker ${message.workerId}`
1314 // FIXME: should be registered only once
1315 this.registerWorkerMessageListener(workerNodeKey
, killMessageListener
)
1316 this.sendToWorker(workerNodeKey
, { kill
: true })
1321 * Terminates the worker node given its worker node key.
1323 * @param workerNodeKey - The worker node key.
1325 protected async destroyWorkerNode (workerNodeKey
: number): Promise
<void> {
1326 this.flagWorkerNodeAsNotReady(workerNodeKey
)
1327 const flushedTasks
= this.flushTasksQueue(workerNodeKey
)
1328 const workerNode
= this.workerNodes
[workerNodeKey
]
1329 await waitWorkerNodeEvents(
1333 this.opts
.tasksQueueOptions
?.tasksFinishedTimeout
??
1334 getDefaultTasksQueueOptions(
1335 this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
1336 ).tasksFinishedTimeout
1338 await this.sendKillMessageToWorker(workerNodeKey
)
1339 await workerNode
.terminate()
1343 * Setup hook to execute code before worker nodes are created in the abstract constructor.
1344 * Can be overridden.
1348 protected setupHook (): void {
1349 /* Intentionally empty */
1353 * Returns whether the worker is the main worker or not.
1355 * @returns `true` if the worker is the main worker, `false` otherwise.
1357 protected abstract isMain (): boolean
1360 * Hook executed before the worker task execution.
1361 * Can be overridden.
1363 * @param workerNodeKey - The worker node key.
1364 * @param task - The task to execute.
1366 protected beforeTaskExecutionHook (
1367 workerNodeKey
: number,
1370 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1371 if (this.workerNodes
[workerNodeKey
]?.usage
!= null) {
1372 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
1373 ++workerUsage
.tasks
.executing
1374 updateWaitTimeWorkerUsage(
1375 this.workerChoiceStrategiesContext
,
1381 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1382 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1383 this.workerNodes
[workerNodeKey
].getTaskFunctionWorkerUsage(task
.name
!) !=
1386 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1387 const taskFunctionWorkerUsage
= this.workerNodes
[
1389 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1390 ].getTaskFunctionWorkerUsage(task
.name
!)!
1391 ++taskFunctionWorkerUsage
.tasks
.executing
1392 updateWaitTimeWorkerUsage(
1393 this.workerChoiceStrategiesContext
,
1394 taskFunctionWorkerUsage
,
1401 * Hook executed after the worker task execution.
1402 * Can be overridden.
1404 * @param workerNodeKey - The worker node key.
1405 * @param message - The received message.
1407 protected afterTaskExecutionHook (
1408 workerNodeKey
: number,
1409 message
: MessageValue
<Response
>
1411 let needWorkerChoiceStrategiesUpdate
= false
1412 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1413 if (this.workerNodes
[workerNodeKey
]?.usage
!= null) {
1414 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
1415 updateTaskStatisticsWorkerUsage(workerUsage
, message
)
1416 updateRunTimeWorkerUsage(
1417 this.workerChoiceStrategiesContext
,
1421 updateEluWorkerUsage(
1422 this.workerChoiceStrategiesContext
,
1426 needWorkerChoiceStrategiesUpdate
= true
1429 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1430 this.workerNodes
[workerNodeKey
].getTaskFunctionWorkerUsage(
1431 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1432 message
.taskPerformance
!.name
1435 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1436 const taskFunctionWorkerUsage
= this.workerNodes
[
1438 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1439 ].getTaskFunctionWorkerUsage(message
.taskPerformance
!.name
)!
1440 updateTaskStatisticsWorkerUsage(taskFunctionWorkerUsage
, message
)
1441 updateRunTimeWorkerUsage(
1442 this.workerChoiceStrategiesContext
,
1443 taskFunctionWorkerUsage
,
1446 updateEluWorkerUsage(
1447 this.workerChoiceStrategiesContext
,
1448 taskFunctionWorkerUsage
,
1451 needWorkerChoiceStrategiesUpdate
= true
1453 if (needWorkerChoiceStrategiesUpdate
) {
1454 this.workerChoiceStrategiesContext
?.update(workerNodeKey
)
1459 * Whether the worker node shall update its task function worker usage or not.
1461 * @param workerNodeKey - The worker node key.
1462 * @returns `true` if the worker node shall update its task function worker usage, `false` otherwise.
1464 private shallUpdateTaskFunctionWorkerUsage (workerNodeKey
: number): boolean {
1465 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1467 workerInfo
!= null &&
1468 Array.isArray(workerInfo
.taskFunctionsProperties
) &&
1469 workerInfo
.taskFunctionsProperties
.length
> 2
1474 * Chooses a worker node for the next task.
1476 * @param name - The task function name.
1477 * @returns The chosen worker node key.
1479 private chooseWorkerNode (name
?: string): number {
1480 if (this.shallCreateDynamicWorker()) {
1481 const workerNodeKey
= this.createAndSetupDynamicWorkerNode()
1483 this.workerChoiceStrategiesContext
?.getPolicy().dynamicWorkerUsage
===
1486 return workerNodeKey
1489 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1490 return this.workerChoiceStrategiesContext
!.execute(
1491 this.getTaskFunctionWorkerChoiceStrategy(name
)
1496 * Conditions for dynamic worker creation.
1498 * @returns Whether to create a dynamic worker or not.
1500 protected abstract shallCreateDynamicWorker (): boolean
1503 * Sends a message to worker given its worker node key.
1505 * @param workerNodeKey - The worker node key.
1506 * @param message - The message.
1507 * @param transferList - The optional array of transferable objects.
1509 protected abstract sendToWorker (
1510 workerNodeKey
: number,
1511 message
: MessageValue
<Data
>,
1512 transferList
?: readonly TransferListItem
[]
1516 * Initializes the worker node usage with sensible default values gathered during runtime.
1518 * @param workerNode - The worker node.
1520 private initWorkerNodeUsage (workerNode
: IWorkerNode
<Worker
, Data
>): void {
1522 this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
1523 .runTime
.aggregate
=== true
1525 workerNode
.usage
.runTime
.aggregate
= min(
1526 ...this.workerNodes
.map(
1528 workerNode
.usage
.runTime
.aggregate
?? Number.POSITIVE_INFINITY
1533 this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
1534 .waitTime
.aggregate
=== true
1536 workerNode
.usage
.waitTime
.aggregate
= min(
1537 ...this.workerNodes
.map(
1539 workerNode
.usage
.waitTime
.aggregate
?? Number.POSITIVE_INFINITY
1544 this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements().elu
1547 workerNode
.usage
.elu
.active
.aggregate
= min(
1548 ...this.workerNodes
.map(
1550 workerNode
.usage
.elu
.active
.aggregate
?? Number.POSITIVE_INFINITY
1557 * Creates a new, completely set up worker node.
1559 * @returns New, completely set up worker node key.
1561 protected createAndSetupWorkerNode (): number {
1562 const workerNode
= this.createWorkerNode()
1563 workerNode
.registerWorkerEventHandler(
1565 this.opts
.onlineHandler
?? EMPTY_FUNCTION
1567 workerNode
.registerWorkerEventHandler(
1569 this.opts
.messageHandler
?? EMPTY_FUNCTION
1571 workerNode
.registerWorkerEventHandler(
1573 this.opts
.errorHandler
?? EMPTY_FUNCTION
1575 workerNode
.registerOnceWorkerEventHandler('error', (error
: Error) => {
1576 workerNode
.info
.ready
= false
1577 this.emitter
?.emit(PoolEvents
.error
, error
)
1581 this.opts
.restartWorkerOnError
=== true
1583 if (workerNode
.info
.dynamic
) {
1584 this.createAndSetupDynamicWorkerNode()
1585 } else if (!this.startingMinimumNumberOfWorkers
) {
1586 this.startMinimumNumberOfWorkers(true)
1592 this.opts
.enableTasksQueue
=== true
1594 this.redistributeQueuedTasks(this.workerNodes
.indexOf(workerNode
))
1596 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1597 workerNode
?.terminate().catch((error
: unknown
) => {
1598 this.emitter
?.emit(PoolEvents
.error
, error
)
1601 workerNode
.registerWorkerEventHandler(
1603 this.opts
.exitHandler
?? EMPTY_FUNCTION
1605 workerNode
.registerOnceWorkerEventHandler('exit', () => {
1606 this.removeWorkerNode(workerNode
)
1609 !this.startingMinimumNumberOfWorkers
&&
1612 this.startMinimumNumberOfWorkers(true)
1615 const workerNodeKey
= this.addWorkerNode(workerNode
)
1616 this.afterWorkerNodeSetup(workerNodeKey
)
1617 return workerNodeKey
1621 * Creates a new, completely set up dynamic worker node.
1623 * @returns New, completely set up dynamic worker node key.
1625 protected createAndSetupDynamicWorkerNode (): number {
1626 const workerNodeKey
= this.createAndSetupWorkerNode()
1627 this.registerWorkerMessageListener(workerNodeKey
, message
=> {
1628 this.checkMessageWorkerId(message
)
1629 const localWorkerNodeKey
= this.getWorkerNodeKeyByWorkerId(
1632 const workerInfo
= this.getWorkerInfo(localWorkerNodeKey
)
1633 const workerUsage
= this.workerNodes
[localWorkerNodeKey
]?.usage
1634 // Kill message received from worker
1636 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
1637 (isKillBehavior(KillBehaviors
.SOFT
, message
.kill
) &&
1638 ((this.opts
.enableTasksQueue
=== false &&
1639 workerUsage
.tasks
.executing
=== 0) ||
1640 (this.opts
.enableTasksQueue
=== true &&
1641 workerInfo
!= null &&
1642 !workerInfo
.stealing
&&
1643 workerUsage
.tasks
.executing
=== 0 &&
1644 this.tasksQueueSize(localWorkerNodeKey
) === 0)))
1646 // Flag the worker node as not ready immediately
1647 this.flagWorkerNodeAsNotReady(localWorkerNodeKey
)
1648 this.destroyWorkerNode(localWorkerNodeKey
).catch((error
: unknown
) => {
1649 this.emitter
?.emit(PoolEvents
.error
, error
)
1653 this.sendToWorker(workerNodeKey
, {
1656 if (this.taskFunctions
.size
> 0) {
1657 for (const [taskFunctionName
, taskFunctionObject
] of this.taskFunctions
) {
1658 this.sendTaskFunctionOperationToWorker(workerNodeKey
, {
1659 taskFunctionOperation
: 'add',
1660 taskFunctionProperties
: buildTaskFunctionProperties(
1664 taskFunction
: taskFunctionObject
.taskFunction
.toString()
1665 }).catch((error
: unknown
) => {
1666 this.emitter
?.emit(PoolEvents
.error
, error
)
1670 const workerNode
= this.workerNodes
[workerNodeKey
]
1671 workerNode
.info
.dynamic
= true
1673 this.workerChoiceStrategiesContext
?.getPolicy().dynamicWorkerReady
===
1675 this.workerChoiceStrategiesContext
?.getPolicy().dynamicWorkerUsage
===
1678 workerNode
.info
.ready
= true
1680 this.initWorkerNodeUsage(workerNode
)
1681 this.checkAndEmitDynamicWorkerCreationEvents()
1682 return workerNodeKey
1686 * Registers a listener callback on the worker given its worker node key.
1688 * @param workerNodeKey - The worker node key.
1689 * @param listener - The message listener callback.
1691 protected abstract registerWorkerMessageListener
<
1692 Message
extends Data
| Response
1694 workerNodeKey
: number,
1695 listener
: (message
: MessageValue
<Message
>) => void
1699 * Registers once a listener callback on the worker given its worker node key.
1701 * @param workerNodeKey - The worker node key.
1702 * @param listener - The message listener callback.
1704 protected abstract registerOnceWorkerMessageListener
<
1705 Message
extends Data
| Response
1707 workerNodeKey
: number,
1708 listener
: (message
: MessageValue
<Message
>) => void
1712 * Deregisters a listener callback on the worker given its worker node key.
1714 * @param workerNodeKey - The worker node key.
1715 * @param listener - The message listener callback.
1717 protected abstract deregisterWorkerMessageListener
<
1718 Message
extends Data
| Response
1720 workerNodeKey
: number,
1721 listener
: (message
: MessageValue
<Message
>) => void
1725 * Method hooked up after a worker node has been newly created.
1726 * Can be overridden.
1728 * @param workerNodeKey - The newly created worker node key.
1730 protected afterWorkerNodeSetup (workerNodeKey
: number): void {
1731 // Listen to worker messages.
1732 this.registerWorkerMessageListener(
1734 this.workerMessageListener
1736 // Send the startup message to worker.
1737 this.sendStartupMessageToWorker(workerNodeKey
)
1738 // Send the statistics message to worker.
1739 this.sendStatisticsMessageToWorker(workerNodeKey
)
1740 if (this.opts
.enableTasksQueue
=== true) {
1741 if (this.opts
.tasksQueueOptions
?.taskStealing
=== true) {
1742 this.workerNodes
[workerNodeKey
].on(
1744 this.handleWorkerNodeIdleEvent
1747 if (this.opts
.tasksQueueOptions
?.tasksStealingOnBackPressure
=== true) {
1748 this.workerNodes
[workerNodeKey
].on(
1750 this.handleWorkerNodeBackPressureEvent
1757 * Sends the startup message to worker given its worker node key.
1759 * @param workerNodeKey - The worker node key.
1761 protected abstract sendStartupMessageToWorker (workerNodeKey
: number): void
1764 * Sends the statistics message to worker given its worker node key.
1766 * @param workerNodeKey - The worker node key.
1768 private sendStatisticsMessageToWorker (workerNodeKey
: number): void {
1769 this.sendToWorker(workerNodeKey
, {
1772 this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
1773 .runTime
.aggregate
?? false,
1775 this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
1776 .elu
.aggregate
?? false
1781 private cannotStealTask (): boolean {
1782 return this.workerNodes
.length
<= 1 || this.info
.queuedTasks
=== 0
1785 private handleTask (workerNodeKey
: number, task
: Task
<Data
>): void {
1786 if (this.shallExecuteTask(workerNodeKey
)) {
1787 this.executeTask(workerNodeKey
, task
)
1789 this.enqueueTask(workerNodeKey
, task
)
1793 private redistributeQueuedTasks (sourceWorkerNodeKey
: number): void {
1794 if (sourceWorkerNodeKey
=== -1 || this.cannotStealTask()) {
1797 while (this.tasksQueueSize(sourceWorkerNodeKey
) > 0) {
1798 const destinationWorkerNodeKey
= this.workerNodes
.reduce(
1799 (minWorkerNodeKey
, workerNode
, workerNodeKey
, workerNodes
) => {
1800 return sourceWorkerNodeKey
!== workerNodeKey
&&
1801 workerNode
.info
.ready
&&
1802 workerNode
.usage
.tasks
.queued
<
1803 workerNodes
[minWorkerNodeKey
].usage
.tasks
.queued
1810 destinationWorkerNodeKey
,
1811 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1812 this.dequeueTask(sourceWorkerNodeKey
)!
1817 private updateTaskStolenStatisticsWorkerUsage (
1818 workerNodeKey
: number,
1821 const workerNode
= this.workerNodes
[workerNodeKey
]
1822 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1823 if (workerNode
?.usage
!= null) {
1824 ++workerNode
.usage
.tasks
.stolen
1827 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1828 workerNode
.getTaskFunctionWorkerUsage(taskName
) != null
1830 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1831 ++workerNode
.getTaskFunctionWorkerUsage(taskName
)!.tasks
.stolen
1835 private updateTaskSequentiallyStolenStatisticsWorkerUsage (
1836 workerNodeKey
: number,
1838 previousTaskName
?: string
1840 const workerNode
= this.workerNodes
[workerNodeKey
]
1841 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1842 if (workerNode
?.usage
!= null) {
1843 ++workerNode
.usage
.tasks
.sequentiallyStolen
1846 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1847 workerNode
.getTaskFunctionWorkerUsage(taskName
) != null
1849 const taskFunctionWorkerUsage
=
1850 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1851 workerNode
.getTaskFunctionWorkerUsage(taskName
)!
1853 taskFunctionWorkerUsage
.tasks
.sequentiallyStolen
=== 0 ||
1854 (previousTaskName
!= null &&
1855 previousTaskName
=== taskName
&&
1856 taskFunctionWorkerUsage
.tasks
.sequentiallyStolen
> 0)
1858 ++taskFunctionWorkerUsage
.tasks
.sequentiallyStolen
1859 } else if (taskFunctionWorkerUsage
.tasks
.sequentiallyStolen
> 0) {
1860 taskFunctionWorkerUsage
.tasks
.sequentiallyStolen
= 0
1865 private resetTaskSequentiallyStolenStatisticsWorkerUsage (
1866 workerNodeKey
: number,
1869 const workerNode
= this.workerNodes
[workerNodeKey
]
1870 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1871 if (workerNode
?.usage
!= null) {
1872 workerNode
.usage
.tasks
.sequentiallyStolen
= 0
1875 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1876 workerNode
.getTaskFunctionWorkerUsage(taskName
) != null
1878 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1879 workerNode
.getTaskFunctionWorkerUsage(
1881 )!.tasks
.sequentiallyStolen
= 0
1885 private readonly handleWorkerNodeIdleEvent
= (
1886 eventDetail
: WorkerNodeEventDetail
,
1887 previousStolenTask
?: Task
<Data
>
1889 const { workerNodeKey
} = eventDetail
1890 if (workerNodeKey
== null) {
1892 "WorkerNode event detail 'workerNodeKey' property must be defined"
1895 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1896 if (workerInfo
== null) {
1898 `Worker node with key '${workerNodeKey}' not found in pool`
1902 this.cannotStealTask() ||
1903 (this.info
.stealingWorkerNodes
?? 0) >
1904 Math.floor(this.workerNodes
.length
/ 2)
1906 if (previousStolenTask
!= null) {
1907 workerInfo
.stealing
= false
1908 this.resetTaskSequentiallyStolenStatisticsWorkerUsage(
1910 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1911 previousStolenTask
.name
!
1916 const workerNodeTasksUsage
= this.workerNodes
[workerNodeKey
].usage
.tasks
1918 previousStolenTask
!= null &&
1919 (workerNodeTasksUsage
.executing
> 0 ||
1920 this.tasksQueueSize(workerNodeKey
) > 0)
1922 workerInfo
.stealing
= false
1923 this.resetTaskSequentiallyStolenStatisticsWorkerUsage(
1925 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1926 previousStolenTask
.name
!
1930 workerInfo
.stealing
= true
1931 const stolenTask
= this.workerNodeStealTask(workerNodeKey
)
1932 if (stolenTask
!= null) {
1933 this.updateTaskSequentiallyStolenStatisticsWorkerUsage(
1935 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1937 previousStolenTask
?.name
1940 sleep(exponentialDelay(workerNodeTasksUsage
.sequentiallyStolen
))
1942 this.handleWorkerNodeIdleEvent(eventDetail
, stolenTask
)
1945 .catch((error
: unknown
) => {
1946 this.emitter
?.emit(PoolEvents
.error
, error
)
1950 private readonly workerNodeStealTask
= (
1951 workerNodeKey
: number
1952 ): Task
<Data
> | undefined => {
1953 const workerNodes
= this.workerNodes
1956 (workerNodeA
, workerNodeB
) =>
1957 workerNodeB
.usage
.tasks
.queued
- workerNodeA
.usage
.tasks
.queued
1959 const sourceWorkerNode
= workerNodes
.find(
1960 (sourceWorkerNode
, sourceWorkerNodeKey
) =>
1961 sourceWorkerNode
.info
.ready
&&
1962 !sourceWorkerNode
.info
.stealing
&&
1963 sourceWorkerNodeKey
!== workerNodeKey
&&
1964 sourceWorkerNode
.usage
.tasks
.queued
> 0
1966 if (sourceWorkerNode
!= null) {
1967 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1968 const task
= sourceWorkerNode
.dequeueLastPrioritizedTask()!
1969 this.handleTask(workerNodeKey
, task
)
1970 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1971 this.updateTaskStolenStatisticsWorkerUsage(workerNodeKey
, task
.name
!)
1976 private readonly handleWorkerNodeBackPressureEvent
= (
1977 eventDetail
: WorkerNodeEventDetail
1980 this.cannotStealTask() ||
1981 this.hasBackPressure() ||
1982 (this.info
.stealingWorkerNodes
?? 0) >
1983 Math.floor(this.workerNodes
.length
/ 2)
1987 const sizeOffset
= 1
1988 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1989 if (this.opts
.tasksQueueOptions
!.size
! <= sizeOffset
) {
1992 const { workerId
} = eventDetail
1993 const sourceWorkerNode
=
1994 this.workerNodes
[this.getWorkerNodeKeyByWorkerId(workerId
)]
1995 const workerNodes
= this.workerNodes
1998 (workerNodeA
, workerNodeB
) =>
1999 workerNodeA
.usage
.tasks
.queued
- workerNodeB
.usage
.tasks
.queued
2001 for (const [workerNodeKey
, workerNode
] of workerNodes
.entries()) {
2003 sourceWorkerNode
.usage
.tasks
.queued
> 0 &&
2004 workerNode
.info
.ready
&&
2005 !workerNode
.info
.stealing
&&
2006 workerNode
.info
.id
!== workerId
&&
2007 workerNode
.usage
.tasks
.queued
<
2008 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2009 this.opts
.tasksQueueOptions
!.size
! - sizeOffset
2011 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
2012 if (workerInfo
== null) {
2014 `Worker node with key '${workerNodeKey}' not found in pool`
2017 workerInfo
.stealing
= true
2018 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2019 const task
= sourceWorkerNode
.dequeueLastPrioritizedTask()!
2020 this.handleTask(workerNodeKey
, task
)
2021 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2022 this.updateTaskStolenStatisticsWorkerUsage(workerNodeKey
, task
.name
!)
2023 workerInfo
.stealing
= false
2028 private setTasksQueuePriority (workerNodeKey
: number): void {
2029 this.workerNodes
[workerNodeKey
].setTasksQueuePriority(
2030 this.getTasksQueuePriority()
2035 * This method is the message listener registered on each worker.
2037 protected readonly workerMessageListener
= (
2038 message
: MessageValue
<Response
>
2040 this.checkMessageWorkerId(message
)
2041 const { workerId
, ready
, taskId
, taskFunctionsProperties
} = message
2042 if (ready
!= null && taskFunctionsProperties
!= null) {
2043 // Worker ready response received from worker
2044 this.handleWorkerReadyResponse(message
)
2045 } else if (taskFunctionsProperties
!= null) {
2046 // Task function properties message received from worker
2047 const workerNodeKey
= this.getWorkerNodeKeyByWorkerId(workerId
)
2048 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
2049 if (workerInfo
!= null) {
2050 workerInfo
.taskFunctionsProperties
= taskFunctionsProperties
2051 this.sendStatisticsMessageToWorker(workerNodeKey
)
2052 this.setTasksQueuePriority(workerNodeKey
)
2054 } else if (taskId
!= null) {
2055 // Task execution response received from worker
2056 this.handleTaskExecutionResponse(message
)
2060 private checkAndEmitReadyEvent (): void {
2061 if (!this.readyEventEmitted
&& this.ready
) {
2062 this.emitter
?.emit(PoolEvents
.ready
, this.info
)
2063 this.readyEventEmitted
= true
2067 private handleWorkerReadyResponse (message
: MessageValue
<Response
>): void {
2068 const { workerId
, ready
, taskFunctionsProperties
} = message
2069 if (ready
== null || !ready
) {
2070 throw new Error(`Worker ${workerId} failed to initialize`)
2072 const workerNodeKey
= this.getWorkerNodeKeyByWorkerId(workerId
)
2073 const workerNode
= this.workerNodes
[workerNodeKey
]
2074 workerNode
.info
.ready
= ready
2075 workerNode
.info
.taskFunctionsProperties
= taskFunctionsProperties
2076 this.sendStatisticsMessageToWorker(workerNodeKey
)
2077 this.setTasksQueuePriority(workerNodeKey
)
2078 this.checkAndEmitReadyEvent()
2081 private handleTaskExecutionResponse (message
: MessageValue
<Response
>): void {
2082 const { workerId
, taskId
, workerError
, data
} = message
2083 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2084 const promiseResponse
= this.promiseResponseMap
.get(taskId
!)
2085 if (promiseResponse
!= null) {
2086 const { resolve
, reject
, workerNodeKey
, asyncResource
} = promiseResponse
2087 const workerNode
= this.workerNodes
[workerNodeKey
]
2088 if (workerError
!= null) {
2089 this.emitter
?.emit(PoolEvents
.taskError
, workerError
)
2090 asyncResource
!= null
2091 ? asyncResource
.runInAsyncScope(
2096 : reject(workerError
.message
)
2098 asyncResource
!= null
2099 ? asyncResource
.runInAsyncScope(resolve
, this.emitter
, data
)
2100 : resolve(data
as Response
)
2102 asyncResource
?.emitDestroy()
2103 this.afterTaskExecutionHook(workerNodeKey
, message
)
2104 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2105 this.promiseResponseMap
.delete(taskId
!)
2106 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
2107 workerNode
?.emit('taskFinished', taskId
)
2109 this.opts
.enableTasksQueue
=== true &&
2111 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
2114 const workerNodeTasksUsage
= workerNode
.usage
.tasks
2116 this.tasksQueueSize(workerNodeKey
) > 0 &&
2117 workerNodeTasksUsage
.executing
<
2118 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2119 this.opts
.tasksQueueOptions
!.concurrency
!
2121 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2122 this.executeTask(workerNodeKey
, this.dequeueTask(workerNodeKey
)!)
2125 workerNodeTasksUsage
.executing
=== 0 &&
2126 this.tasksQueueSize(workerNodeKey
) === 0 &&
2127 workerNodeTasksUsage
.sequentiallyStolen
=== 0
2129 workerNode
.emit('idle', {
2138 private checkAndEmitTaskExecutionEvents (): void {
2140 this.emitter
?.emit(PoolEvents
.busy
, this.info
)
2144 private checkAndEmitTaskQueuingEvents (): void {
2145 if (this.hasBackPressure()) {
2146 this.emitter
?.emit(PoolEvents
.backPressure
, this.info
)
2151 * Emits dynamic worker creation events.
2153 protected abstract checkAndEmitDynamicWorkerCreationEvents (): void
2156 * Gets the worker information given its worker node key.
2158 * @param workerNodeKey - The worker node key.
2159 * @returns The worker information.
2161 protected getWorkerInfo (workerNodeKey
: number): WorkerInfo
| undefined {
2162 return this.workerNodes
[workerNodeKey
]?.info
2165 private getTasksQueuePriority (): boolean {
2166 return this.listTaskFunctionsProperties().some(
2167 taskFunctionProperties
=> taskFunctionProperties
.priority
!= null
2172 * Creates a worker node.
2174 * @returns The created worker node.
2176 private createWorkerNode (): IWorkerNode
<Worker
, Data
> {
2177 const workerNode
= new WorkerNode
<Worker
, Data
>(
2182 workerOptions
: this.opts
.workerOptions
,
2183 tasksQueueBackPressureSize
:
2184 this.opts
.tasksQueueOptions
?.size
??
2185 getDefaultTasksQueueOptions(
2186 this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
2188 tasksQueueBucketSize
: defaultBucketSize
,
2189 tasksQueuePriority
: this.getTasksQueuePriority()
2192 // Flag the worker node as ready at pool startup.
2193 if (this.starting
) {
2194 workerNode
.info
.ready
= true
2200 * Adds the given worker node in the pool worker nodes.
2202 * @param workerNode - The worker node.
2203 * @returns The added worker node key.
2204 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the added worker node is not found.
2206 private addWorkerNode (workerNode
: IWorkerNode
<Worker
, Data
>): number {
2207 this.workerNodes
.push(workerNode
)
2208 const workerNodeKey
= this.workerNodes
.indexOf(workerNode
)
2209 if (workerNodeKey
=== -1) {
2210 throw new Error('Worker added not found in worker nodes')
2212 return workerNodeKey
2215 private checkAndEmitEmptyEvent (): void {
2217 this.emitter
?.emit(PoolEvents
.empty
, this.info
)
2218 this.readyEventEmitted
= false
2223 * Removes the worker node from the pool worker nodes.
2225 * @param workerNode - The worker node.
2227 private removeWorkerNode (workerNode
: IWorkerNode
<Worker
, Data
>): void {
2228 const workerNodeKey
= this.workerNodes
.indexOf(workerNode
)
2229 if (workerNodeKey
!== -1) {
2230 this.workerNodes
.splice(workerNodeKey
, 1)
2231 this.workerChoiceStrategiesContext
?.remove(workerNodeKey
)
2233 this.checkAndEmitEmptyEvent()
2236 protected flagWorkerNodeAsNotReady (workerNodeKey
: number): void {
2237 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
2238 if (workerInfo
!= null) {
2239 workerInfo
.ready
= false
2243 private hasBackPressure (): boolean {
2245 this.opts
.enableTasksQueue
=== true &&
2246 this.workerNodes
.findIndex(
2247 workerNode
=> !workerNode
.hasBackPressure()
2253 * Executes the given task on the worker given its worker node key.
2255 * @param workerNodeKey - The worker node key.
2256 * @param task - The task to execute.
2258 private executeTask (workerNodeKey
: number, task
: Task
<Data
>): void {
2259 this.beforeTaskExecutionHook(workerNodeKey
, task
)
2260 this.sendToWorker(workerNodeKey
, task
, task
.transferList
)
2261 this.checkAndEmitTaskExecutionEvents()
2264 private enqueueTask (workerNodeKey
: number, task
: Task
<Data
>): number {
2265 const tasksQueueSize
= this.workerNodes
[workerNodeKey
].enqueueTask(task
)
2266 this.checkAndEmitTaskQueuingEvents()
2267 return tasksQueueSize
2270 private dequeueTask (workerNodeKey
: number): Task
<Data
> | undefined {
2271 return this.workerNodes
[workerNodeKey
].dequeueTask()
2274 private tasksQueueSize (workerNodeKey
: number): number {
2275 return this.workerNodes
[workerNodeKey
].tasksQueueSize()
2278 protected flushTasksQueue (workerNodeKey
: number): number {
2279 let flushedTasks
= 0
2280 while (this.tasksQueueSize(workerNodeKey
) > 0) {
2281 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2282 this.executeTask(workerNodeKey
, this.dequeueTask(workerNodeKey
)!)
2285 this.workerNodes
[workerNodeKey
].clearTasksQueue()
2289 private flushTasksQueues (): void {
2290 for (const workerNodeKey
of this.workerNodes
.keys()) {
2291 this.flushTasksQueue(workerNodeKey
)