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.
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
: 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.
149 * @param minimumNumberOfWorkers - Minimum number of workers that this pool manages.
150 * @param filePath - Path to the worker file.
151 * @param opts - Options for the pool.
152 * @param maximumNumberOfWorkers - Maximum number of workers that this pool manages.
155 protected readonly minimumNumberOfWorkers
: number,
156 protected readonly filePath
: string,
157 protected readonly opts
: PoolOptions
<Worker
>,
158 protected readonly maximumNumberOfWorkers
?: number
160 if (!this.isMain()) {
162 'Cannot start a pool from a worker with the same type as the pool'
166 checkFilePath(this.filePath
)
167 this.checkMinimumNumberOfWorkers(this.minimumNumberOfWorkers
)
168 this.checkPoolOptions(this.opts
)
170 this.chooseWorkerNode
= this.chooseWorkerNode
.bind(this)
171 this.executeTask
= this.executeTask
.bind(this)
172 this.enqueueTask
= this.enqueueTask
.bind(this)
174 if (this.opts
.enableEvents
=== true) {
175 this.initEventEmitter()
177 this.workerChoiceStrategiesContext
= new WorkerChoiceStrategiesContext
<
183 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
184 [this.opts
.workerChoiceStrategy
!],
185 this.opts
.workerChoiceStrategyOptions
190 this.taskFunctions
= new Map
<string, TaskFunctionObject
<Data
, Response
>>()
193 this.starting
= false
194 this.destroying
= false
195 this.readyEventEmitted
= false
196 this.startingMinimumNumberOfWorkers
= false
197 if (this.opts
.startWorkers
=== true) {
202 private checkPoolType (): void {
203 if (this.type === PoolTypes
.fixed
&& this.maximumNumberOfWorkers
!= null) {
205 'Cannot instantiate a fixed pool with a maximum number of workers specified at initialization'
210 private checkMinimumNumberOfWorkers (
211 minimumNumberOfWorkers
: number | undefined
213 if (minimumNumberOfWorkers
== null) {
215 'Cannot instantiate a pool without specifying the number of workers'
217 } else if (!Number.isSafeInteger(minimumNumberOfWorkers
)) {
219 'Cannot instantiate a pool with a non safe integer number of workers'
221 } else if (minimumNumberOfWorkers
< 0) {
222 throw new RangeError(
223 'Cannot instantiate a pool with a negative number of workers'
225 } else if (this.type === PoolTypes
.fixed
&& minimumNumberOfWorkers
=== 0) {
226 throw new RangeError('Cannot instantiate a fixed pool with zero worker')
230 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
231 if (isPlainObject(opts
)) {
232 this.opts
.startWorkers
= opts
.startWorkers
?? true
233 checkValidWorkerChoiceStrategy(opts
.workerChoiceStrategy
)
234 this.opts
.workerChoiceStrategy
=
235 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
236 this.checkValidWorkerChoiceStrategyOptions(
237 opts
.workerChoiceStrategyOptions
239 if (opts
.workerChoiceStrategyOptions
!= null) {
240 this.opts
.workerChoiceStrategyOptions
= opts
.workerChoiceStrategyOptions
242 this.opts
.restartWorkerOnError
= opts
.restartWorkerOnError
?? true
243 this.opts
.enableEvents
= opts
.enableEvents
?? true
244 this.opts
.enableTasksQueue
= opts
.enableTasksQueue
?? false
245 if (this.opts
.enableTasksQueue
) {
246 checkValidTasksQueueOptions(opts
.tasksQueueOptions
)
247 this.opts
.tasksQueueOptions
= this.buildTasksQueueOptions(
248 opts
.tasksQueueOptions
252 throw new TypeError('Invalid pool options: must be a plain object')
256 private checkValidWorkerChoiceStrategyOptions (
257 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
| undefined
260 workerChoiceStrategyOptions
!= null &&
261 !isPlainObject(workerChoiceStrategyOptions
)
264 'Invalid worker choice strategy options: must be a plain object'
268 workerChoiceStrategyOptions
?.weights
!= null &&
269 Object.keys(workerChoiceStrategyOptions
.weights
).length
!==
270 (this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
)
273 'Invalid worker choice strategy options: must have a weight for each worker node'
277 workerChoiceStrategyOptions
?.measurement
!= null &&
278 !Object.values(Measurements
).includes(
279 workerChoiceStrategyOptions
.measurement
283 `Invalid worker choice strategy options: invalid measurement '${workerChoiceStrategyOptions.measurement}'`
288 private initEventEmitter (): void {
289 this.emitter
= new EventEmitterAsyncResource({
290 name
: `poolifier:${this.type}-${this.worker}-pool`,
295 public get
info (): PoolInfo
{
300 started
: this.started
,
302 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
303 defaultStrategy
: this.opts
.workerChoiceStrategy
!,
304 strategyRetries
: this.workerChoiceStrategiesContext
?.retriesCount
?? 0,
305 minSize
: this.minimumNumberOfWorkers
,
306 maxSize
: this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
,
307 ...(this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
308 .runTime
.aggregate
=== true &&
309 this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
310 .waitTime
.aggregate
&& {
311 utilization
: round(this.utilization
),
313 workerNodes
: this.workerNodes
.length
,
314 idleWorkerNodes
: this.workerNodes
.reduce(
315 (accumulator
, workerNode
) =>
316 workerNode
.usage
.tasks
.executing
=== 0
321 ...(this.opts
.enableTasksQueue
=== true && {
322 stealingWorkerNodes
: this.workerNodes
.reduce(
323 (accumulator
, workerNode
) =>
324 workerNode
.info
.stealing
? accumulator
+ 1 : accumulator
,
328 busyWorkerNodes
: this.workerNodes
.reduce(
329 (accumulator
, _
, workerNodeKey
) =>
330 this.isWorkerNodeBusy(workerNodeKey
) ? accumulator
+ 1 : accumulator
,
333 executedTasks
: this.workerNodes
.reduce(
334 (accumulator
, workerNode
) =>
335 accumulator
+ workerNode
.usage
.tasks
.executed
,
338 executingTasks
: this.workerNodes
.reduce(
339 (accumulator
, workerNode
) =>
340 accumulator
+ workerNode
.usage
.tasks
.executing
,
343 ...(this.opts
.enableTasksQueue
=== true && {
344 queuedTasks
: this.workerNodes
.reduce(
345 (accumulator
, workerNode
) =>
346 accumulator
+ workerNode
.usage
.tasks
.queued
,
350 ...(this.opts
.enableTasksQueue
=== true && {
351 maxQueuedTasks
: this.workerNodes
.reduce(
352 (accumulator
, workerNode
) =>
353 accumulator
+ (workerNode
.usage
.tasks
.maxQueued
?? 0),
357 ...(this.opts
.enableTasksQueue
=== true && {
358 backPressure
: this.hasBackPressure(),
360 ...(this.opts
.enableTasksQueue
=== true && {
361 stolenTasks
: this.workerNodes
.reduce(
362 (accumulator
, workerNode
) =>
363 accumulator
+ workerNode
.usage
.tasks
.stolen
,
367 failedTasks
: this.workerNodes
.reduce(
368 (accumulator
, workerNode
) =>
369 accumulator
+ workerNode
.usage
.tasks
.failed
,
372 ...(this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
373 .runTime
.aggregate
=== true && {
377 ...this.workerNodes
.map(
379 workerNode
.usage
.runTime
.minimum
?? Number.POSITIVE_INFINITY
385 ...this.workerNodes
.map(
387 workerNode
.usage
.runTime
.maximum
?? Number.NEGATIVE_INFINITY
391 ...(this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
392 .runTime
.average
&& {
395 this.workerNodes
.reduce
<number[]>(
396 (accumulator
, workerNode
) =>
398 workerNode
.usage
.runTime
.history
.toArray()
405 ...(this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
409 this.workerNodes
.reduce
<number[]>(
410 (accumulator
, workerNode
) =>
412 workerNode
.usage
.runTime
.history
.toArray()
421 ...(this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
422 .waitTime
.aggregate
=== true && {
426 ...this.workerNodes
.map(
428 workerNode
.usage
.waitTime
.minimum
?? Number.POSITIVE_INFINITY
434 ...this.workerNodes
.map(
436 workerNode
.usage
.waitTime
.maximum
?? Number.NEGATIVE_INFINITY
440 ...(this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
441 .waitTime
.average
&& {
444 this.workerNodes
.reduce
<number[]>(
445 (accumulator
, workerNode
) =>
447 workerNode
.usage
.waitTime
.history
.toArray()
454 ...(this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
455 .waitTime
.median
&& {
458 this.workerNodes
.reduce
<number[]>(
459 (accumulator
, workerNode
) =>
461 workerNode
.usage
.waitTime
.history
.toArray()
470 ...(this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
471 .elu
.aggregate
=== true && {
476 ...this.workerNodes
.map(
478 workerNode
.usage
.elu
.idle
.minimum
??
479 Number.POSITIVE_INFINITY
485 ...this.workerNodes
.map(
487 workerNode
.usage
.elu
.idle
.maximum
??
488 Number.NEGATIVE_INFINITY
492 ...(this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
496 this.workerNodes
.reduce
<number[]>(
497 (accumulator
, workerNode
) =>
499 workerNode
.usage
.elu
.idle
.history
.toArray()
506 ...(this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
510 this.workerNodes
.reduce
<number[]>(
511 (accumulator
, workerNode
) =>
513 workerNode
.usage
.elu
.idle
.history
.toArray()
524 ...this.workerNodes
.map(
526 workerNode
.usage
.elu
.active
.minimum
??
527 Number.POSITIVE_INFINITY
533 ...this.workerNodes
.map(
535 workerNode
.usage
.elu
.active
.maximum
??
536 Number.NEGATIVE_INFINITY
540 ...(this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
544 this.workerNodes
.reduce
<number[]>(
545 (accumulator
, workerNode
) =>
547 workerNode
.usage
.elu
.active
.history
.toArray()
554 ...(this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
558 this.workerNodes
.reduce
<number[]>(
559 (accumulator
, workerNode
) =>
561 workerNode
.usage
.elu
.active
.history
.toArray()
572 this.workerNodes
.map(
573 workerNode
=> workerNode
.usage
.elu
.utilization
?? 0
579 this.workerNodes
.map(
580 workerNode
=> workerNode
.usage
.elu
.utilization
?? 0
591 * Whether the pool is ready or not.
592 * @returns The pool readiness boolean status.
594 private get
ready (): boolean {
599 this.workerNodes
.reduce(
600 (accumulator
, workerNode
) =>
601 !workerNode
.info
.dynamic
&& workerNode
.info
.ready
605 ) >= this.minimumNumberOfWorkers
610 * Whether the pool is empty or not.
611 * @returns The pool emptiness boolean status.
613 protected get
empty (): boolean {
614 return this.minimumNumberOfWorkers
=== 0 && this.workerNodes
.length
=== 0
618 * The approximate pool utilization.
619 * @returns The pool utilization.
621 private get
utilization (): number {
622 if (this.startTimestamp
== null) {
625 const poolTimeCapacity
=
626 (performance
.now() - this.startTimestamp
) *
627 (this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
)
628 const totalTasksRunTime
= this.workerNodes
.reduce(
629 (accumulator
, workerNode
) =>
630 accumulator
+ (workerNode
.usage
.runTime
.aggregate
?? 0),
633 const totalTasksWaitTime
= this.workerNodes
.reduce(
634 (accumulator
, workerNode
) =>
635 accumulator
+ (workerNode
.usage
.waitTime
.aggregate
?? 0),
638 return (totalTasksRunTime
+ totalTasksWaitTime
) / poolTimeCapacity
644 * If it is `'dynamic'`, it provides the `max` property.
646 protected abstract get
type (): PoolType
651 protected abstract get
worker (): WorkerType
654 * Checks if the worker id sent in the received message from a worker is valid.
655 * @param message - The received message.
656 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the worker id is invalid.
658 private checkMessageWorkerId (message
: MessageValue
<Data
| Response
>): void {
659 if (message
.workerId
== null) {
660 throw new Error('Worker message received without worker id')
661 } else if (this.getWorkerNodeKeyByWorkerId(message
.workerId
) === -1) {
663 `Worker message received from unknown worker '${message.workerId.toString()}'`
669 * Gets the worker node key given its worker id.
670 * @param workerId - The worker id.
671 * @returns The worker node key if the worker id is found in the pool worker nodes, `-1` otherwise.
673 private getWorkerNodeKeyByWorkerId (workerId
: number | undefined): number {
674 return this.workerNodes
.findIndex(
675 workerNode
=> workerNode
.info
.id
=== workerId
680 public setWorkerChoiceStrategy (
681 workerChoiceStrategy
: WorkerChoiceStrategy
,
682 workerChoiceStrategyOptions
?: WorkerChoiceStrategyOptions
684 let requireSync
= false
685 checkValidWorkerChoiceStrategy(workerChoiceStrategy
)
686 if (workerChoiceStrategyOptions
!= null) {
687 requireSync
= !this.setWorkerChoiceStrategyOptions(
688 workerChoiceStrategyOptions
691 if (workerChoiceStrategy
!== this.opts
.workerChoiceStrategy
) {
692 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
693 this.workerChoiceStrategiesContext
?.setDefaultWorkerChoiceStrategy(
694 this.opts
.workerChoiceStrategy
,
695 this.opts
.workerChoiceStrategyOptions
700 this.workerChoiceStrategiesContext
?.syncWorkerChoiceStrategies(
701 this.getWorkerChoiceStrategies(),
702 this.opts
.workerChoiceStrategyOptions
704 for (const workerNodeKey
of this.workerNodes
.keys()) {
705 this.sendStatisticsMessageToWorker(workerNodeKey
)
711 public setWorkerChoiceStrategyOptions (
712 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
| undefined
714 this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
715 if (workerChoiceStrategyOptions
!= null) {
716 this.opts
.workerChoiceStrategyOptions
= workerChoiceStrategyOptions
717 this.workerChoiceStrategiesContext
?.setOptions(
718 this.opts
.workerChoiceStrategyOptions
720 this.workerChoiceStrategiesContext
?.syncWorkerChoiceStrategies(
721 this.getWorkerChoiceStrategies(),
722 this.opts
.workerChoiceStrategyOptions
724 for (const workerNodeKey
of this.workerNodes
.keys()) {
725 this.sendStatisticsMessageToWorker(workerNodeKey
)
733 public enableTasksQueue (
735 tasksQueueOptions
?: TasksQueueOptions
737 if (this.opts
.enableTasksQueue
=== true && !enable
) {
738 this.unsetTaskStealing()
739 this.unsetTasksStealingOnBackPressure()
740 this.flushTasksQueues()
742 this.opts
.enableTasksQueue
= enable
743 this.setTasksQueueOptions(tasksQueueOptions
)
747 public setTasksQueueOptions (
748 tasksQueueOptions
: TasksQueueOptions
| undefined
750 if (this.opts
.enableTasksQueue
=== true) {
751 checkValidTasksQueueOptions(tasksQueueOptions
)
752 this.opts
.tasksQueueOptions
=
753 this.buildTasksQueueOptions(tasksQueueOptions
)
754 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
755 this.setTasksQueueSize(this.opts
.tasksQueueOptions
.size
!)
756 if (this.opts
.tasksQueueOptions
.taskStealing
=== true) {
757 this.unsetTaskStealing()
758 this.setTaskStealing()
760 this.unsetTaskStealing()
762 if (this.opts
.tasksQueueOptions
.tasksStealingOnBackPressure
=== true) {
763 this.unsetTasksStealingOnBackPressure()
764 this.setTasksStealingOnBackPressure()
766 this.unsetTasksStealingOnBackPressure()
768 } else if (this.opts
.tasksQueueOptions
!= null) {
769 delete this.opts
.tasksQueueOptions
773 private buildTasksQueueOptions (
774 tasksQueueOptions
: TasksQueueOptions
| undefined
775 ): TasksQueueOptions
{
777 ...getDefaultTasksQueueOptions(
778 this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
780 ...tasksQueueOptions
,
784 private setTasksQueueSize (size
: number): void {
785 for (const workerNode
of this.workerNodes
) {
786 workerNode
.tasksQueueBackPressureSize
= size
790 private setTaskStealing (): void {
791 for (const workerNodeKey
of this.workerNodes
.keys()) {
792 this.workerNodes
[workerNodeKey
].on('idle', this.handleWorkerNodeIdleEvent
)
796 private unsetTaskStealing (): void {
797 for (const workerNodeKey
of this.workerNodes
.keys()) {
798 this.workerNodes
[workerNodeKey
].off(
800 this.handleWorkerNodeIdleEvent
805 private setTasksStealingOnBackPressure (): void {
806 for (const workerNodeKey
of this.workerNodes
.keys()) {
807 this.workerNodes
[workerNodeKey
].on(
809 this.handleWorkerNodeBackPressureEvent
814 private unsetTasksStealingOnBackPressure (): void {
815 for (const workerNodeKey
of this.workerNodes
.keys()) {
816 this.workerNodes
[workerNodeKey
].off(
818 this.handleWorkerNodeBackPressureEvent
824 * Whether the pool is full or not.
825 * @returns The pool fullness boolean status.
827 protected get
full (): boolean {
829 this.workerNodes
.length
>=
830 (this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
)
835 * Whether the pool is busy or not.
836 * @returns The pool busyness boolean status.
838 protected abstract get
busy (): boolean
841 * Whether worker nodes are executing concurrently their tasks quota or not.
842 * @returns Worker nodes busyness boolean status.
844 protected internalBusy (): boolean {
845 if (this.opts
.enableTasksQueue
=== true) {
847 this.workerNodes
.findIndex(
849 workerNode
.info
.ready
&&
850 workerNode
.usage
.tasks
.executing
<
851 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
852 this.opts
.tasksQueueOptions
!.concurrency
!
857 this.workerNodes
.findIndex(
859 workerNode
.info
.ready
&& workerNode
.usage
.tasks
.executing
=== 0
864 private isWorkerNodeBusy (workerNodeKey
: number): boolean {
865 if (this.opts
.enableTasksQueue
=== true) {
867 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
>=
868 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
869 this.opts
.tasksQueueOptions
!.concurrency
!
872 return this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
> 0
875 private async sendTaskFunctionOperationToWorker (
876 workerNodeKey
: number,
877 message
: MessageValue
<Data
>
878 ): Promise
<boolean> {
879 return await new Promise
<boolean>((resolve
, reject
) => {
880 const taskFunctionOperationListener
= (
881 message
: MessageValue
<Response
>
883 this.checkMessageWorkerId(message
)
884 const workerId
= this.getWorkerInfo(workerNodeKey
)?.id
886 message
.taskFunctionOperationStatus
!= null &&
887 message
.workerId
=== workerId
889 if (message
.taskFunctionOperationStatus
) {
894 // eslint-disable-next-line @typescript-eslint/restrict-template-expressions
895 `Task function operation '${message.taskFunctionOperation?.toString()}' failed on worker ${message.workerId?.toString()} with error: '${
896 // eslint-disable-next-line @typescript-eslint/restrict-template-expressions
897 message.workerError?.message
902 this.deregisterWorkerMessageListener(
903 this.getWorkerNodeKeyByWorkerId(message
.workerId
),
904 taskFunctionOperationListener
908 this.registerWorkerMessageListener(
910 taskFunctionOperationListener
912 this.sendToWorker(workerNodeKey
, message
)
916 private async sendTaskFunctionOperationToWorkers (
917 message
: MessageValue
<Data
>
918 ): Promise
<boolean> {
919 return await new Promise
<boolean>((resolve
, reject
) => {
920 const responsesReceived
= new Array<MessageValue
<Response
>>()
921 const taskFunctionOperationsListener
= (
922 message
: MessageValue
<Response
>
924 this.checkMessageWorkerId(message
)
925 if (message
.taskFunctionOperationStatus
!= null) {
926 responsesReceived
.push(message
)
927 if (responsesReceived
.length
=== this.workerNodes
.length
) {
929 responsesReceived
.every(
930 message
=> message
.taskFunctionOperationStatus
=== true
935 responsesReceived
.some(
936 message
=> message
.taskFunctionOperationStatus
=== false
939 const errorResponse
= responsesReceived
.find(
940 response
=> response
.taskFunctionOperationStatus
=== false
944 `Task function operation '${
945 message.taskFunctionOperation as string
946 // eslint-disable-next-line @typescript-eslint/restrict-template-expressions
947 }' failed on worker ${errorResponse?.workerId?.toString()} with error: '${
948 // eslint-disable-next-line @typescript-eslint/restrict-template-expressions
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.
1054 * @param name - The task function name.
1055 * @returns The task function worker choice strategy if the task function worker choice strategy is defined, `undefined` otherwise.
1057 private readonly getTaskFunctionWorkerChoiceStrategy
= (
1059 ): WorkerChoiceStrategy
| undefined => {
1060 name
= name
?? DEFAULT_TASK_NAME
1061 const taskFunctionsProperties
= this.listTaskFunctionsProperties()
1062 if (name
=== DEFAULT_TASK_NAME
) {
1063 name
= taskFunctionsProperties
[1]?.name
1065 return taskFunctionsProperties
.find(
1066 (taskFunctionProperties
: TaskFunctionProperties
) =>
1067 taskFunctionProperties
.name
=== name
1072 * Gets worker node task function worker choice strategy, if any.
1073 * @param workerNodeKey - The worker node key.
1074 * @param name - The task function name.
1075 * @returns The worker node task function worker choice strategy if the worker node task function worker choice strategy is defined, `undefined` otherwise.
1077 private readonly getWorkerNodeTaskFunctionWorkerChoiceStrategy
= (
1078 workerNodeKey
: number,
1080 ): WorkerChoiceStrategy
| undefined => {
1081 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1082 if (workerInfo
== null) {
1085 name
= name
?? DEFAULT_TASK_NAME
1086 if (name
=== DEFAULT_TASK_NAME
) {
1087 name
= workerInfo
.taskFunctionsProperties
?.[1]?.name
1089 return workerInfo
.taskFunctionsProperties
?.find(
1090 (taskFunctionProperties
: TaskFunctionProperties
) =>
1091 taskFunctionProperties
.name
=== name
1096 * Gets worker node task function priority, if any.
1097 * @param workerNodeKey - The worker node key.
1098 * @param name - The task function name.
1099 * @returns The worker node task function priority if the worker node task function priority is defined, `undefined` otherwise.
1101 private readonly getWorkerNodeTaskFunctionPriority
= (
1102 workerNodeKey
: number,
1104 ): number | undefined => {
1105 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1106 if (workerInfo
== null) {
1109 name
= name
?? DEFAULT_TASK_NAME
1110 if (name
=== DEFAULT_TASK_NAME
) {
1111 name
= workerInfo
.taskFunctionsProperties
?.[1]?.name
1113 return workerInfo
.taskFunctionsProperties
?.find(
1114 (taskFunctionProperties
: TaskFunctionProperties
) =>
1115 taskFunctionProperties
.name
=== name
1120 * Gets the worker choice strategies registered in this pool.
1121 * @returns The worker choice strategies.
1123 private readonly getWorkerChoiceStrategies
=
1124 (): Set
<WorkerChoiceStrategy
> => {
1126 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1127 this.opts
.workerChoiceStrategy
!,
1128 ...(this.listTaskFunctionsProperties()
1130 (taskFunctionProperties
: TaskFunctionProperties
) =>
1131 taskFunctionProperties
.strategy
1134 (strategy
: WorkerChoiceStrategy
| undefined) => strategy
!= null
1135 ) as WorkerChoiceStrategy
[]),
1140 public async setDefaultTaskFunction (name
: string): Promise
<boolean> {
1141 return await this.sendTaskFunctionOperationToWorkers({
1142 taskFunctionOperation
: 'default',
1143 taskFunctionProperties
: buildTaskFunctionProperties(
1145 this.taskFunctions
.get(name
)
1150 private shallExecuteTask (workerNodeKey
: number): boolean {
1152 this.tasksQueueSize(workerNodeKey
) === 0 &&
1153 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
<
1154 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1155 this.opts
.tasksQueueOptions
!.concurrency
!
1160 public async execute (
1163 transferList
?: readonly TransferListItem
[]
1164 ): Promise
<Response
> {
1165 return await new Promise
<Response
>((resolve
, reject
) => {
1166 if (!this.started
) {
1167 reject(new Error('Cannot execute a task on not started pool'))
1170 if (this.destroying
) {
1171 reject(new Error('Cannot execute a task on destroying pool'))
1174 if (name
!= null && typeof name
!== 'string') {
1175 reject(new TypeError('name argument must be a string'))
1180 typeof name
=== 'string' &&
1181 name
.trim().length
=== 0
1183 reject(new TypeError('name argument must not be an empty string'))
1186 if (transferList
!= null && !Array.isArray(transferList
)) {
1187 reject(new TypeError('transferList argument must be an array'))
1190 const timestamp
= performance
.now()
1191 const workerNodeKey
= this.chooseWorkerNode(name
)
1192 const task
: Task
<Data
> = {
1193 name
: name
?? DEFAULT_TASK_NAME
,
1194 data
: data
?? ({} as Data
),
1195 priority
: this.getWorkerNodeTaskFunctionPriority(workerNodeKey
, name
),
1196 strategy
: this.getWorkerNodeTaskFunctionWorkerChoiceStrategy(
1202 taskId
: randomUUID(),
1204 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1205 this.promiseResponseMap
.set(task
.taskId
!, {
1209 ...(this.emitter
!= null && {
1210 asyncResource
: new AsyncResource('poolifier:task', {
1211 triggerAsyncId
: this.emitter
.asyncId
,
1212 requireManualDestroy
: true,
1217 this.opts
.enableTasksQueue
=== false ||
1218 (this.opts
.enableTasksQueue
=== true &&
1219 this.shallExecuteTask(workerNodeKey
))
1221 this.executeTask(workerNodeKey
, task
)
1223 this.enqueueTask(workerNodeKey
, task
)
1230 data
: Iterable
<Data
>,
1232 transferList
?: readonly TransferListItem
[]
1233 ): Promise
<Response
[]> {
1234 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1236 throw new TypeError('data argument must be a defined iterable')
1238 if (typeof data
[Symbol
.iterator
] !== 'function') {
1239 throw new TypeError('data argument must be an iterable')
1241 if (!Array.isArray(data
)) {
1245 (data
as Data
[]).map(data
=> this.execute(data
, name
, transferList
))
1250 * Starts the minimum number of workers.
1251 * @param initWorkerNodeUsage - Whether to initialize the worker node usage or not. @defaultValue false
1253 private startMinimumNumberOfWorkers (initWorkerNodeUsage
= false): void {
1254 if (this.minimumNumberOfWorkers
=== 0) {
1257 this.startingMinimumNumberOfWorkers
= true
1259 this.workerNodes
.reduce(
1260 (accumulator
, workerNode
) =>
1261 !workerNode
.info
.dynamic
? accumulator
+ 1 : accumulator
,
1263 ) < this.minimumNumberOfWorkers
1265 const workerNodeKey
= this.createAndSetupWorkerNode()
1266 initWorkerNodeUsage
&&
1267 this.initWorkerNodeUsage(this.workerNodes
[workerNodeKey
])
1269 this.startingMinimumNumberOfWorkers
= false
1273 public start (): void {
1275 throw new Error('Cannot start an already started pool')
1277 if (this.starting
) {
1278 throw new Error('Cannot start an already starting pool')
1280 if (this.destroying
) {
1281 throw new Error('Cannot start a destroying pool')
1283 this.starting
= true
1284 this.startMinimumNumberOfWorkers()
1285 this.startTimestamp
= performance
.now()
1286 this.starting
= false
1291 public async destroy (): Promise
<void> {
1292 if (!this.started
) {
1293 throw new Error('Cannot destroy an already destroyed pool')
1295 if (this.starting
) {
1296 throw new Error('Cannot destroy an starting pool')
1298 if (this.destroying
) {
1299 throw new Error('Cannot destroy an already destroying pool')
1301 this.destroying
= true
1303 this.workerNodes
.map(async (_
, workerNodeKey
) => {
1304 await this.destroyWorkerNode(workerNodeKey
)
1307 this.emitter
?.emit(PoolEvents
.destroy
, this.info
)
1308 this.emitter
?.emitDestroy()
1309 this.readyEventEmitted
= false
1310 delete this.startTimestamp
1311 this.destroying
= false
1312 this.started
= false
1315 private async sendKillMessageToWorker (workerNodeKey
: number): Promise
<void> {
1316 await new Promise
<void>((resolve
, reject
) => {
1317 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1318 if (this.workerNodes
[workerNodeKey
] == null) {
1322 const killMessageListener
= (message
: MessageValue
<Response
>): void => {
1323 this.checkMessageWorkerId(message
)
1324 if (message
.kill
=== 'success') {
1326 } else if (message
.kill
=== 'failure') {
1329 // eslint-disable-next-line @typescript-eslint/restrict-template-expressions
1330 `Kill message handling failed on worker ${message.workerId?.toString()}`
1335 // FIXME: should be registered only once
1336 this.registerWorkerMessageListener(workerNodeKey
, killMessageListener
)
1337 this.sendToWorker(workerNodeKey
, { kill
: true })
1342 * Terminates the worker node given its worker node key.
1343 * @param workerNodeKey - The worker node key.
1345 protected async destroyWorkerNode (workerNodeKey
: number): Promise
<void> {
1346 this.flagWorkerNodeAsNotReady(workerNodeKey
)
1347 const flushedTasks
= this.flushTasksQueue(workerNodeKey
)
1348 const workerNode
= this.workerNodes
[workerNodeKey
]
1349 await waitWorkerNodeEvents(
1353 this.opts
.tasksQueueOptions
?.tasksFinishedTimeout
??
1354 getDefaultTasksQueueOptions(
1355 this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
1356 ).tasksFinishedTimeout
1358 await this.sendKillMessageToWorker(workerNodeKey
)
1359 await workerNode
.terminate()
1363 * Setup hook to execute code before worker nodes are created in the abstract constructor.
1364 * Can be overridden.
1366 protected setupHook (): void {
1367 /* Intentionally empty */
1371 * Returns whether the worker is the main worker or not.
1372 * @returns `true` if the worker is the main worker, `false` otherwise.
1374 protected abstract isMain (): boolean
1377 * Hook executed before the worker task execution.
1378 * Can be overridden.
1379 * @param workerNodeKey - The worker node key.
1380 * @param task - The task to execute.
1382 protected beforeTaskExecutionHook (
1383 workerNodeKey
: number,
1386 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1387 if (this.workerNodes
[workerNodeKey
]?.usage
!= null) {
1388 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
1389 ++workerUsage
.tasks
.executing
1390 updateWaitTimeWorkerUsage(
1391 this.workerChoiceStrategiesContext
,
1397 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1398 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1399 this.workerNodes
[workerNodeKey
].getTaskFunctionWorkerUsage(task
.name
!) !=
1402 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1403 const taskFunctionWorkerUsage
= this.workerNodes
[
1405 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1406 ].getTaskFunctionWorkerUsage(task
.name
!)!
1407 ++taskFunctionWorkerUsage
.tasks
.executing
1408 updateWaitTimeWorkerUsage(
1409 this.workerChoiceStrategiesContext
,
1410 taskFunctionWorkerUsage
,
1417 * Hook executed after the worker task execution.
1418 * Can be overridden.
1419 * @param workerNodeKey - The worker node key.
1420 * @param message - The received message.
1422 protected afterTaskExecutionHook (
1423 workerNodeKey
: number,
1424 message
: MessageValue
<Response
>
1426 let needWorkerChoiceStrategiesUpdate
= false
1427 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1428 if (this.workerNodes
[workerNodeKey
]?.usage
!= null) {
1429 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
1430 updateTaskStatisticsWorkerUsage(workerUsage
, message
)
1431 updateRunTimeWorkerUsage(
1432 this.workerChoiceStrategiesContext
,
1436 updateEluWorkerUsage(
1437 this.workerChoiceStrategiesContext
,
1441 needWorkerChoiceStrategiesUpdate
= true
1444 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1445 this.workerNodes
[workerNodeKey
].getTaskFunctionWorkerUsage(
1446 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1447 message
.taskPerformance
!.name
1450 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1451 const taskFunctionWorkerUsage
= this.workerNodes
[
1453 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1454 ].getTaskFunctionWorkerUsage(message
.taskPerformance
!.name
)!
1455 updateTaskStatisticsWorkerUsage(taskFunctionWorkerUsage
, message
)
1456 updateRunTimeWorkerUsage(
1457 this.workerChoiceStrategiesContext
,
1458 taskFunctionWorkerUsage
,
1461 updateEluWorkerUsage(
1462 this.workerChoiceStrategiesContext
,
1463 taskFunctionWorkerUsage
,
1466 needWorkerChoiceStrategiesUpdate
= true
1468 if (needWorkerChoiceStrategiesUpdate
) {
1469 this.workerChoiceStrategiesContext
?.update(workerNodeKey
)
1474 * Whether the worker node shall update its task function worker usage or not.
1475 * @param workerNodeKey - The worker node key.
1476 * @returns `true` if the worker node shall update its task function worker usage, `false` otherwise.
1478 private shallUpdateTaskFunctionWorkerUsage (workerNodeKey
: number): boolean {
1479 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1481 workerInfo
!= null &&
1482 Array.isArray(workerInfo
.taskFunctionsProperties
) &&
1483 workerInfo
.taskFunctionsProperties
.length
> 2
1488 * Chooses a worker node for the next task.
1489 * @param name - The task function name.
1490 * @returns The chosen worker node key.
1492 private chooseWorkerNode (name
?: string): number {
1493 if (this.shallCreateDynamicWorker()) {
1494 const workerNodeKey
= this.createAndSetupDynamicWorkerNode()
1496 this.workerChoiceStrategiesContext
?.getPolicy().dynamicWorkerUsage
===
1499 return workerNodeKey
1502 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1503 return this.workerChoiceStrategiesContext
!.execute(
1504 this.getTaskFunctionWorkerChoiceStrategy(name
)
1509 * Conditions for dynamic worker creation.
1510 * @returns Whether to create a dynamic worker or not.
1512 protected abstract shallCreateDynamicWorker (): boolean
1515 * Sends a message to worker given its worker node key.
1516 * @param workerNodeKey - The worker node key.
1517 * @param message - The message.
1518 * @param transferList - The optional array of transferable objects.
1520 protected abstract sendToWorker (
1521 workerNodeKey
: number,
1522 message
: MessageValue
<Data
>,
1523 transferList
?: readonly TransferListItem
[]
1527 * Initializes the worker node usage with sensible default values gathered during runtime.
1528 * @param workerNode - The worker node.
1530 private initWorkerNodeUsage (workerNode
: IWorkerNode
<Worker
, Data
>): void {
1532 this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
1533 .runTime
.aggregate
=== true
1535 workerNode
.usage
.runTime
.aggregate
= min(
1536 ...this.workerNodes
.map(
1538 workerNode
.usage
.runTime
.aggregate
?? Number.POSITIVE_INFINITY
1543 this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
1544 .waitTime
.aggregate
=== true
1546 workerNode
.usage
.waitTime
.aggregate
= min(
1547 ...this.workerNodes
.map(
1549 workerNode
.usage
.waitTime
.aggregate
?? Number.POSITIVE_INFINITY
1554 this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements().elu
1557 workerNode
.usage
.elu
.active
.aggregate
= min(
1558 ...this.workerNodes
.map(
1560 workerNode
.usage
.elu
.active
.aggregate
?? Number.POSITIVE_INFINITY
1567 * Creates a new, completely set up worker node.
1568 * @returns New, completely set up worker node key.
1570 protected createAndSetupWorkerNode (): number {
1571 const workerNode
= this.createWorkerNode()
1572 workerNode
.registerWorkerEventHandler(
1574 this.opts
.onlineHandler
?? EMPTY_FUNCTION
1576 workerNode
.registerWorkerEventHandler(
1578 this.opts
.messageHandler
?? EMPTY_FUNCTION
1580 workerNode
.registerWorkerEventHandler(
1582 this.opts
.errorHandler
?? EMPTY_FUNCTION
1584 workerNode
.registerOnceWorkerEventHandler('error', (error
: Error) => {
1585 workerNode
.info
.ready
= false
1586 this.emitter
?.emit(PoolEvents
.error
, error
)
1590 this.opts
.restartWorkerOnError
=== true
1592 if (workerNode
.info
.dynamic
) {
1593 this.createAndSetupDynamicWorkerNode()
1594 } else if (!this.startingMinimumNumberOfWorkers
) {
1595 this.startMinimumNumberOfWorkers(true)
1601 this.opts
.enableTasksQueue
=== true
1603 this.redistributeQueuedTasks(this.workerNodes
.indexOf(workerNode
))
1605 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1606 workerNode
?.terminate().catch((error
: unknown
) => {
1607 this.emitter
?.emit(PoolEvents
.error
, error
)
1610 workerNode
.registerWorkerEventHandler(
1612 this.opts
.exitHandler
?? EMPTY_FUNCTION
1614 workerNode
.registerOnceWorkerEventHandler('exit', () => {
1615 this.removeWorkerNode(workerNode
)
1618 !this.startingMinimumNumberOfWorkers
&&
1621 this.startMinimumNumberOfWorkers(true)
1624 const workerNodeKey
= this.addWorkerNode(workerNode
)
1625 this.afterWorkerNodeSetup(workerNodeKey
)
1626 return workerNodeKey
1630 * Creates a new, completely set up dynamic worker node.
1631 * @returns New, completely set up dynamic worker node key.
1633 protected createAndSetupDynamicWorkerNode (): number {
1634 const workerNodeKey
= this.createAndSetupWorkerNode()
1635 this.registerWorkerMessageListener(workerNodeKey
, message
=> {
1636 this.checkMessageWorkerId(message
)
1637 const localWorkerNodeKey
= this.getWorkerNodeKeyByWorkerId(
1640 const workerInfo
= this.getWorkerInfo(localWorkerNodeKey
)
1641 const workerUsage
= this.workerNodes
[localWorkerNodeKey
]?.usage
1642 // Kill message received from worker
1644 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
1645 (isKillBehavior(KillBehaviors
.SOFT
, message
.kill
) &&
1646 ((this.opts
.enableTasksQueue
=== false &&
1647 workerUsage
.tasks
.executing
=== 0) ||
1648 (this.opts
.enableTasksQueue
=== true &&
1649 workerInfo
!= null &&
1650 !workerInfo
.stealing
&&
1651 workerUsage
.tasks
.executing
=== 0 &&
1652 this.tasksQueueSize(localWorkerNodeKey
) === 0)))
1654 // Flag the worker node as not ready immediately
1655 this.flagWorkerNodeAsNotReady(localWorkerNodeKey
)
1656 this.destroyWorkerNode(localWorkerNodeKey
).catch((error
: unknown
) => {
1657 this.emitter
?.emit(PoolEvents
.error
, error
)
1661 this.sendToWorker(workerNodeKey
, {
1664 if (this.taskFunctions
.size
> 0) {
1665 for (const [taskFunctionName
, taskFunctionObject
] of this.taskFunctions
) {
1666 this.sendTaskFunctionOperationToWorker(workerNodeKey
, {
1667 taskFunctionOperation
: 'add',
1668 taskFunctionProperties
: buildTaskFunctionProperties(
1672 taskFunction
: taskFunctionObject
.taskFunction
.toString(),
1673 }).catch((error
: unknown
) => {
1674 this.emitter
?.emit(PoolEvents
.error
, error
)
1678 const workerNode
= this.workerNodes
[workerNodeKey
]
1679 workerNode
.info
.dynamic
= true
1681 this.workerChoiceStrategiesContext
?.getPolicy().dynamicWorkerReady
===
1683 this.workerChoiceStrategiesContext
?.getPolicy().dynamicWorkerUsage
===
1686 workerNode
.info
.ready
= true
1688 this.initWorkerNodeUsage(workerNode
)
1689 this.checkAndEmitDynamicWorkerCreationEvents()
1690 return workerNodeKey
1694 * Registers a listener callback on the worker given its worker node key.
1695 * @param workerNodeKey - The worker node key.
1696 * @param listener - The message listener callback.
1698 protected abstract registerWorkerMessageListener
<
1699 Message
extends Data
| Response
1701 workerNodeKey
: number,
1702 listener
: (message
: MessageValue
<Message
>) => void
1706 * Registers once a listener callback on the worker given its worker node key.
1707 * @param workerNodeKey - The worker node key.
1708 * @param listener - The message listener callback.
1710 protected abstract registerOnceWorkerMessageListener
<
1711 Message
extends Data
| Response
1713 workerNodeKey
: number,
1714 listener
: (message
: MessageValue
<Message
>) => void
1718 * Deregisters a listener callback on the worker given its worker node key.
1719 * @param workerNodeKey - The worker node key.
1720 * @param listener - The message listener callback.
1722 protected abstract deregisterWorkerMessageListener
<
1723 Message
extends Data
| Response
1725 workerNodeKey
: number,
1726 listener
: (message
: MessageValue
<Message
>) => void
1730 * Method hooked up after a worker node has been newly created.
1731 * Can be overridden.
1732 * @param workerNodeKey - The newly created worker node key.
1734 protected afterWorkerNodeSetup (workerNodeKey
: number): void {
1735 // Listen to worker messages.
1736 this.registerWorkerMessageListener(
1738 this.workerMessageListener
1740 // Send the startup message to worker.
1741 this.sendStartupMessageToWorker(workerNodeKey
)
1742 // Send the statistics message to worker.
1743 this.sendStatisticsMessageToWorker(workerNodeKey
)
1744 if (this.opts
.enableTasksQueue
=== true) {
1745 if (this.opts
.tasksQueueOptions
?.taskStealing
=== true) {
1746 this.workerNodes
[workerNodeKey
].on(
1748 this.handleWorkerNodeIdleEvent
1751 if (this.opts
.tasksQueueOptions
?.tasksStealingOnBackPressure
=== true) {
1752 this.workerNodes
[workerNodeKey
].on(
1754 this.handleWorkerNodeBackPressureEvent
1761 * Sends the startup message to worker given its worker node key.
1762 * @param workerNodeKey - The worker node key.
1764 protected abstract sendStartupMessageToWorker (workerNodeKey
: number): void
1767 * Sends the statistics message to worker given its worker node key.
1768 * @param workerNodeKey - The worker node key.
1770 private sendStatisticsMessageToWorker (workerNodeKey
: number): void {
1771 this.sendToWorker(workerNodeKey
, {
1774 this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
1775 .runTime
.aggregate
?? false,
1777 this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
1778 .elu
.aggregate
?? false,
1783 private cannotStealTask (): boolean {
1784 return this.workerNodes
.length
<= 1 || this.info
.queuedTasks
=== 0
1787 private handleTask (workerNodeKey
: number, task
: Task
<Data
>): void {
1788 if (this.shallExecuteTask(workerNodeKey
)) {
1789 this.executeTask(workerNodeKey
, task
)
1791 this.enqueueTask(workerNodeKey
, task
)
1795 private redistributeQueuedTasks (sourceWorkerNodeKey
: number): void {
1796 if (sourceWorkerNodeKey
=== -1 || this.cannotStealTask()) {
1799 while (this.tasksQueueSize(sourceWorkerNodeKey
) > 0) {
1800 const destinationWorkerNodeKey
= this.workerNodes
.reduce(
1801 (minWorkerNodeKey
, workerNode
, workerNodeKey
, workerNodes
) => {
1802 return sourceWorkerNodeKey
!== workerNodeKey
&&
1803 workerNode
.info
.ready
&&
1804 workerNode
.usage
.tasks
.queued
<
1805 workerNodes
[minWorkerNodeKey
].usage
.tasks
.queued
1812 destinationWorkerNodeKey
,
1813 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1814 this.dequeueTask(sourceWorkerNodeKey
)!
1819 private updateTaskStolenStatisticsWorkerUsage (
1820 workerNodeKey
: number,
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
.stolen
1829 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1830 workerNode
.getTaskFunctionWorkerUsage(taskName
) != null
1832 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1833 ++workerNode
.getTaskFunctionWorkerUsage(taskName
)!.tasks
.stolen
1837 private updateTaskSequentiallyStolenStatisticsWorkerUsage (
1838 workerNodeKey
: number,
1840 previousTaskName
?: string
1842 const workerNode
= this.workerNodes
[workerNodeKey
]
1843 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1844 if (workerNode
?.usage
!= null) {
1845 ++workerNode
.usage
.tasks
.sequentiallyStolen
1848 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1849 workerNode
.getTaskFunctionWorkerUsage(taskName
) != null
1851 const taskFunctionWorkerUsage
=
1852 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1853 workerNode
.getTaskFunctionWorkerUsage(taskName
)!
1855 taskFunctionWorkerUsage
.tasks
.sequentiallyStolen
=== 0 ||
1856 (previousTaskName
!= null &&
1857 previousTaskName
=== taskName
&&
1858 taskFunctionWorkerUsage
.tasks
.sequentiallyStolen
> 0)
1860 ++taskFunctionWorkerUsage
.tasks
.sequentiallyStolen
1861 } else if (taskFunctionWorkerUsage
.tasks
.sequentiallyStolen
> 0) {
1862 taskFunctionWorkerUsage
.tasks
.sequentiallyStolen
= 0
1867 private resetTaskSequentiallyStolenStatisticsWorkerUsage (
1868 workerNodeKey
: number,
1871 const workerNode
= this.workerNodes
[workerNodeKey
]
1872 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1873 if (workerNode
?.usage
!= null) {
1874 workerNode
.usage
.tasks
.sequentiallyStolen
= 0
1877 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1878 workerNode
.getTaskFunctionWorkerUsage(taskName
) != null
1880 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1881 workerNode
.getTaskFunctionWorkerUsage(
1883 )!.tasks
.sequentiallyStolen
= 0
1887 private readonly handleWorkerNodeIdleEvent
= (
1888 eventDetail
: WorkerNodeEventDetail
,
1889 previousStolenTask
?: Task
<Data
>
1891 const { workerNodeKey
} = eventDetail
1892 if (workerNodeKey
== null) {
1894 "WorkerNode event detail 'workerNodeKey' property must be defined"
1897 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1898 if (workerInfo
== null) {
1900 `Worker node with key '${workerNodeKey.toString()}' not found in pool`
1904 this.cannotStealTask() ||
1905 (this.info
.stealingWorkerNodes
?? 0) >
1906 Math.floor(this.workerNodes
.length
/ 2)
1908 if (previousStolenTask
!= null) {
1909 workerInfo
.stealing
= false
1910 this.resetTaskSequentiallyStolenStatisticsWorkerUsage(
1912 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1913 previousStolenTask
.name
!
1918 const workerNodeTasksUsage
= this.workerNodes
[workerNodeKey
].usage
.tasks
1920 previousStolenTask
!= null &&
1921 (workerNodeTasksUsage
.executing
> 0 ||
1922 this.tasksQueueSize(workerNodeKey
) > 0)
1924 workerInfo
.stealing
= false
1925 this.resetTaskSequentiallyStolenStatisticsWorkerUsage(
1927 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1928 previousStolenTask
.name
!
1932 workerInfo
.stealing
= true
1933 const stolenTask
= this.workerNodeStealTask(workerNodeKey
)
1934 if (stolenTask
!= null) {
1935 this.updateTaskSequentiallyStolenStatisticsWorkerUsage(
1937 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1939 previousStolenTask
?.name
1942 sleep(exponentialDelay(workerNodeTasksUsage
.sequentiallyStolen
))
1944 this.handleWorkerNodeIdleEvent(eventDetail
, stolenTask
)
1947 .catch((error
: unknown
) => {
1948 this.emitter
?.emit(PoolEvents
.error
, error
)
1952 private readonly workerNodeStealTask
= (
1953 workerNodeKey
: number
1954 ): Task
<Data
> | undefined => {
1955 const workerNodes
= this.workerNodes
1958 (workerNodeA
, workerNodeB
) =>
1959 workerNodeB
.usage
.tasks
.queued
- workerNodeA
.usage
.tasks
.queued
1961 const sourceWorkerNode
= workerNodes
.find(
1962 (sourceWorkerNode
, sourceWorkerNodeKey
) =>
1963 sourceWorkerNode
.info
.ready
&&
1964 !sourceWorkerNode
.info
.stealing
&&
1965 sourceWorkerNodeKey
!== workerNodeKey
&&
1966 sourceWorkerNode
.usage
.tasks
.queued
> 0
1968 if (sourceWorkerNode
!= null) {
1969 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1970 const task
= sourceWorkerNode
.dequeueLastPrioritizedTask()!
1971 this.handleTask(workerNodeKey
, task
)
1972 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1973 this.updateTaskStolenStatisticsWorkerUsage(workerNodeKey
, task
.name
!)
1978 private readonly handleWorkerNodeBackPressureEvent
= (
1979 eventDetail
: WorkerNodeEventDetail
1982 this.cannotStealTask() ||
1983 this.hasBackPressure() ||
1984 (this.info
.stealingWorkerNodes
?? 0) >
1985 Math.floor(this.workerNodes
.length
/ 2)
1989 const sizeOffset
= 1
1990 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1991 if (this.opts
.tasksQueueOptions
!.size
! <= sizeOffset
) {
1994 const { workerId
} = eventDetail
1995 const sourceWorkerNode
=
1996 this.workerNodes
[this.getWorkerNodeKeyByWorkerId(workerId
)]
1997 const workerNodes
= this.workerNodes
2000 (workerNodeA
, workerNodeB
) =>
2001 workerNodeA
.usage
.tasks
.queued
- workerNodeB
.usage
.tasks
.queued
2003 for (const [workerNodeKey
, workerNode
] of workerNodes
.entries()) {
2005 sourceWorkerNode
.usage
.tasks
.queued
> 0 &&
2006 workerNode
.info
.ready
&&
2007 !workerNode
.info
.stealing
&&
2008 workerNode
.info
.id
!== workerId
&&
2009 workerNode
.usage
.tasks
.queued
<
2010 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2011 this.opts
.tasksQueueOptions
!.size
! - sizeOffset
2013 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
2014 if (workerInfo
== null) {
2016 `Worker node with key '${workerNodeKey.toString()}' not found in pool`
2019 workerInfo
.stealing
= true
2020 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2021 const task
= sourceWorkerNode
.dequeueLastPrioritizedTask()!
2022 this.handleTask(workerNodeKey
, task
)
2023 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2024 this.updateTaskStolenStatisticsWorkerUsage(workerNodeKey
, task
.name
!)
2025 workerInfo
.stealing
= false
2030 private setTasksQueuePriority (workerNodeKey
: number): void {
2031 this.workerNodes
[workerNodeKey
].setTasksQueuePriority(
2032 this.getTasksQueuePriority()
2037 * This method is the message listener registered on each worker.
2038 * @param message - The message received from the worker.
2040 protected readonly workerMessageListener
= (
2041 message
: MessageValue
<Response
>
2043 this.checkMessageWorkerId(message
)
2044 const { workerId
, ready
, taskId
, taskFunctionsProperties
} = message
2045 if (ready
!= null && taskFunctionsProperties
!= null) {
2046 // Worker ready response received from worker
2047 this.handleWorkerReadyResponse(message
)
2048 } else if (taskFunctionsProperties
!= null) {
2049 // Task function properties message received from worker
2050 const workerNodeKey
= this.getWorkerNodeKeyByWorkerId(workerId
)
2051 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
2052 if (workerInfo
!= null) {
2053 workerInfo
.taskFunctionsProperties
= taskFunctionsProperties
2054 this.sendStatisticsMessageToWorker(workerNodeKey
)
2055 this.setTasksQueuePriority(workerNodeKey
)
2057 } else if (taskId
!= null) {
2058 // Task execution response received from worker
2059 this.handleTaskExecutionResponse(message
)
2063 private checkAndEmitReadyEvent (): void {
2064 if (!this.readyEventEmitted
&& this.ready
) {
2065 this.emitter
?.emit(PoolEvents
.ready
, this.info
)
2066 this.readyEventEmitted
= true
2070 private handleWorkerReadyResponse (message
: MessageValue
<Response
>): void {
2071 const { workerId
, ready
, taskFunctionsProperties
} = message
2072 if (ready
== null || !ready
) {
2073 // eslint-disable-next-line @typescript-eslint/restrict-template-expressions
2074 throw new Error(`Worker ${workerId?.toString()} failed to initialize`)
2076 const workerNodeKey
= this.getWorkerNodeKeyByWorkerId(workerId
)
2077 const workerNode
= this.workerNodes
[workerNodeKey
]
2078 workerNode
.info
.ready
= ready
2079 workerNode
.info
.taskFunctionsProperties
= taskFunctionsProperties
2080 this.sendStatisticsMessageToWorker(workerNodeKey
)
2081 this.setTasksQueuePriority(workerNodeKey
)
2082 this.checkAndEmitReadyEvent()
2085 private handleTaskExecutionResponse (message
: MessageValue
<Response
>): void {
2086 const { workerId
, taskId
, workerError
, data
} = message
2087 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2088 const promiseResponse
= this.promiseResponseMap
.get(taskId
!)
2089 if (promiseResponse
!= null) {
2090 const { resolve
, reject
, workerNodeKey
, asyncResource
} = promiseResponse
2091 const workerNode
= this.workerNodes
[workerNodeKey
]
2092 if (workerError
!= null) {
2093 this.emitter
?.emit(PoolEvents
.taskError
, workerError
)
2094 asyncResource
!= null
2095 ? asyncResource
.runInAsyncScope(
2100 : reject(workerError
.message
)
2102 asyncResource
!= null
2103 ? asyncResource
.runInAsyncScope(resolve
, this.emitter
, data
)
2104 : resolve(data
as Response
)
2106 asyncResource
?.emitDestroy()
2107 this.afterTaskExecutionHook(workerNodeKey
, message
)
2108 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2109 this.promiseResponseMap
.delete(taskId
!)
2110 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
2111 workerNode
?.emit('taskFinished', taskId
)
2113 this.opts
.enableTasksQueue
=== true &&
2115 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
2118 const workerNodeTasksUsage
= workerNode
.usage
.tasks
2120 this.tasksQueueSize(workerNodeKey
) > 0 &&
2121 workerNodeTasksUsage
.executing
<
2122 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2123 this.opts
.tasksQueueOptions
!.concurrency
!
2125 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2126 this.executeTask(workerNodeKey
, this.dequeueTask(workerNodeKey
)!)
2129 workerNodeTasksUsage
.executing
=== 0 &&
2130 this.tasksQueueSize(workerNodeKey
) === 0 &&
2131 workerNodeTasksUsage
.sequentiallyStolen
=== 0
2133 workerNode
.emit('idle', {
2142 private checkAndEmitTaskExecutionEvents (): void {
2144 this.emitter
?.emit(PoolEvents
.busy
, this.info
)
2148 private checkAndEmitTaskQueuingEvents (): void {
2149 if (this.hasBackPressure()) {
2150 this.emitter
?.emit(PoolEvents
.backPressure
, this.info
)
2155 * Emits dynamic worker creation events.
2157 protected abstract checkAndEmitDynamicWorkerCreationEvents (): void
2160 * Gets the worker information given its worker node key.
2161 * @param workerNodeKey - The worker node key.
2162 * @returns The worker information.
2164 protected getWorkerInfo (workerNodeKey
: number): WorkerInfo
| undefined {
2165 return this.workerNodes
[workerNodeKey
]?.info
2168 private getTasksQueuePriority (): boolean {
2169 return this.listTaskFunctionsProperties().some(
2170 taskFunctionProperties
=> taskFunctionProperties
.priority
!= null
2175 * Creates a worker node.
2176 * @returns The created worker node.
2178 private createWorkerNode (): IWorkerNode
<Worker
, Data
> {
2179 const workerNode
= new WorkerNode
<Worker
, Data
>(
2184 workerOptions
: this.opts
.workerOptions
,
2185 tasksQueueBackPressureSize
:
2186 this.opts
.tasksQueueOptions
?.size
??
2187 getDefaultTasksQueueOptions(
2188 this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
2190 tasksQueueBucketSize
: defaultBucketSize
,
2191 tasksQueuePriority
: this.getTasksQueuePriority(),
2194 // Flag the worker node as ready at pool startup.
2195 if (this.starting
) {
2196 workerNode
.info
.ready
= true
2202 * Adds the given worker node in the pool worker nodes.
2203 * @param workerNode - The worker node.
2204 * @returns The added worker node key.
2205 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the added worker node is not found.
2207 private addWorkerNode (workerNode
: IWorkerNode
<Worker
, Data
>): number {
2208 this.workerNodes
.push(workerNode
)
2209 const workerNodeKey
= this.workerNodes
.indexOf(workerNode
)
2210 if (workerNodeKey
=== -1) {
2211 throw new Error('Worker added not found in worker nodes')
2213 return workerNodeKey
2216 private checkAndEmitEmptyEvent (): void {
2218 this.emitter
?.emit(PoolEvents
.empty
, this.info
)
2219 this.readyEventEmitted
= false
2224 * 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.
2254 * @param workerNodeKey - The worker node key.
2255 * @param task - The task to execute.
2257 private executeTask (workerNodeKey
: number, task
: Task
<Data
>): void {
2258 this.beforeTaskExecutionHook(workerNodeKey
, task
)
2259 this.sendToWorker(workerNodeKey
, task
, task
.transferList
)
2260 this.checkAndEmitTaskExecutionEvents()
2263 private enqueueTask (workerNodeKey
: number, task
: Task
<Data
>): number {
2264 const tasksQueueSize
= this.workerNodes
[workerNodeKey
].enqueueTask(task
)
2265 this.checkAndEmitTaskQueuingEvents()
2266 return tasksQueueSize
2269 private dequeueTask (workerNodeKey
: number): Task
<Data
> | undefined {
2270 return this.workerNodes
[workerNodeKey
].dequeueTask()
2273 private tasksQueueSize (workerNodeKey
: number): number {
2274 return this.workerNodes
[workerNodeKey
].tasksQueueSize()
2277 protected flushTasksQueue (workerNodeKey
: number): number {
2278 let flushedTasks
= 0
2279 while (this.tasksQueueSize(workerNodeKey
) > 0) {
2280 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2281 this.executeTask(workerNodeKey
, this.dequeueTask(workerNodeKey
)!)
2284 this.workerNodes
[workerNodeKey
].clearTasksQueue()
2288 private flushTasksQueues (): void {
2289 for (const workerNodeKey
of this.workerNodes
.keys()) {
2290 this.flushTasksQueue(workerNodeKey
)