1 import { AsyncResource
} from
'node:async_hooks'
2 import { randomUUID
} from
'node:crypto'
3 import { EventEmitterAsyncResource
} from
'node:events'
4 import { performance
} from
'node:perf_hooks'
5 import type { TransferListItem
} from
'node:worker_threads'
9 PromiseResponseWrapper
,
11 TaskFunctionProperties
12 } from
'../utility-types.js'
15 buildTaskFunctionProperties
,
30 } from
'../worker/task-functions.js'
31 import { KillBehaviors
} from
'../worker/worker-options.js'
39 type TasksQueueOptions
43 WorkerChoiceStrategies
,
44 type WorkerChoiceStrategy
,
45 type WorkerChoiceStrategyOptions
46 } from
'./selection-strategies/selection-strategies-types.js'
47 import { WorkerChoiceStrategiesContext
} from
'./selection-strategies/worker-choice-strategies-context.js'
51 checkValidTasksQueueOptions
,
52 checkValidWorkerChoiceStrategy
,
53 getDefaultTasksQueueOptions
,
55 updateRunTimeWorkerUsage
,
56 updateTaskStatisticsWorkerUsage
,
57 updateWaitTimeWorkerUsage
,
60 import { version
} from
'./version.js'
65 WorkerNodeEventDetail
,
68 import { WorkerNode
} from
'./worker-node.js'
71 * Base class that implements some shared logic for all poolifier pools.
73 * @typeParam Worker - Type of worker which manages this pool.
74 * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
75 * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
77 export abstract class AbstractPool
<
78 Worker
extends IWorker
,
81 > implements IPool
<Worker
, Data
, Response
> {
83 public readonly workerNodes
: Array<IWorkerNode
<Worker
, Data
>> = []
86 public emitter
?: EventEmitterAsyncResource
89 * The task execution response promise map:
90 * - `key`: The message id of each submitted task.
91 * - `value`: An object that contains task's worker node key, execution response promise resolve and reject callbacks, async resource.
93 * When we receive a message from the worker, we get a map entry with the promise resolve/reject bound to the message id.
95 protected promiseResponseMap
: Map
<
96 `${string}-${string}-${string}-${string}-${string}`,
97 PromiseResponseWrapper
<Response
>
99 `${string}-${string}-${string}-${string}-${string}`,
100 PromiseResponseWrapper
<Response
>
104 * Worker choice strategies context referencing worker choice algorithms implementation.
106 protected workerChoiceStrategiesContext
?: WorkerChoiceStrategiesContext
<
113 * The task functions added at runtime map:
114 * - `key`: The task function name.
115 * - `value`: The task function object.
117 private readonly taskFunctions
: Map
<
119 TaskFunctionObject
<Data
, Response
>
123 * Whether the pool is started or not.
125 private started
: boolean
127 * Whether the pool is starting or not.
129 private starting
: boolean
131 * Whether the pool is destroying or not.
133 private destroying
: boolean
135 * Whether the minimum number of workers is starting or not.
137 private startingMinimumNumberOfWorkers
: boolean
139 * Whether the pool ready event has been emitted or not.
141 private readyEventEmitted
: boolean
143 * The start timestamp of the pool.
145 private startTimestamp
?: number
148 * Constructs a new poolifier pool.
150 * @param minimumNumberOfWorkers - Minimum number of workers that this pool manages.
151 * @param filePath - Path to the worker file.
152 * @param opts - Options for the pool.
153 * @param maximumNumberOfWorkers - Maximum number of workers that this pool manages.
156 protected readonly minimumNumberOfWorkers
: number,
157 protected readonly filePath
: string,
158 protected readonly opts
: PoolOptions
<Worker
>,
159 protected readonly maximumNumberOfWorkers
?: number
161 if (!this.isMain()) {
163 'Cannot start a pool from a worker with the same type as the pool'
167 checkFilePath(this.filePath
)
168 this.checkMinimumNumberOfWorkers(this.minimumNumberOfWorkers
)
169 this.checkPoolOptions(this.opts
)
171 this.chooseWorkerNode
= this.chooseWorkerNode
.bind(this)
172 this.executeTask
= this.executeTask
.bind(this)
173 this.enqueueTask
= this.enqueueTask
.bind(this)
175 if (this.opts
.enableEvents
=== true) {
176 this.initEventEmitter()
178 this.workerChoiceStrategiesContext
= new WorkerChoiceStrategiesContext
<
184 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
185 [this.opts
.workerChoiceStrategy
!],
186 this.opts
.workerChoiceStrategyOptions
191 this.taskFunctions
= new Map
<string, TaskFunctionObject
<Data
, Response
>>()
194 this.starting
= false
195 this.destroying
= false
196 this.readyEventEmitted
= false
197 this.startingMinimumNumberOfWorkers
= false
198 if (this.opts
.startWorkers
=== true) {
203 private checkPoolType (): void {
204 if (this.type === PoolTypes
.fixed
&& this.maximumNumberOfWorkers
!= null) {
206 'Cannot instantiate a fixed pool with a maximum number of workers specified at initialization'
211 private checkMinimumNumberOfWorkers (
212 minimumNumberOfWorkers
: number | undefined
214 if (minimumNumberOfWorkers
== null) {
216 'Cannot instantiate a pool without specifying the number of workers'
218 } else if (!Number.isSafeInteger(minimumNumberOfWorkers
)) {
220 'Cannot instantiate a pool with a non safe integer number of workers'
222 } else if (minimumNumberOfWorkers
< 0) {
223 throw new RangeError(
224 'Cannot instantiate a pool with a negative number of workers'
226 } else if (this.type === PoolTypes
.fixed
&& minimumNumberOfWorkers
=== 0) {
227 throw new RangeError('Cannot instantiate a fixed pool with zero worker')
231 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
232 if (isPlainObject(opts
)) {
233 this.opts
.startWorkers
= opts
.startWorkers
?? true
234 checkValidWorkerChoiceStrategy(opts
.workerChoiceStrategy
)
235 this.opts
.workerChoiceStrategy
=
236 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
237 this.checkValidWorkerChoiceStrategyOptions(
238 opts
.workerChoiceStrategyOptions
240 if (opts
.workerChoiceStrategyOptions
!= null) {
241 this.opts
.workerChoiceStrategyOptions
= opts
.workerChoiceStrategyOptions
243 this.opts
.restartWorkerOnError
= opts
.restartWorkerOnError
?? true
244 this.opts
.enableEvents
= opts
.enableEvents
?? true
245 this.opts
.enableTasksQueue
= opts
.enableTasksQueue
?? false
246 if (this.opts
.enableTasksQueue
) {
247 checkValidTasksQueueOptions(opts
.tasksQueueOptions
)
248 this.opts
.tasksQueueOptions
= this.buildTasksQueueOptions(
249 opts
.tasksQueueOptions
253 throw new TypeError('Invalid pool options: must be a plain object')
257 private checkValidWorkerChoiceStrategyOptions (
258 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
| undefined
261 workerChoiceStrategyOptions
!= null &&
262 !isPlainObject(workerChoiceStrategyOptions
)
265 'Invalid worker choice strategy options: must be a plain object'
269 workerChoiceStrategyOptions
?.weights
!= null &&
270 Object.keys(workerChoiceStrategyOptions
.weights
).length
!==
271 (this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
)
274 'Invalid worker choice strategy options: must have a weight for each worker node'
278 workerChoiceStrategyOptions
?.measurement
!= null &&
279 !Object.values(Measurements
).includes(
280 workerChoiceStrategyOptions
.measurement
284 `Invalid worker choice strategy options: invalid measurement '${workerChoiceStrategyOptions.measurement}'`
289 private initEventEmitter (): void {
290 this.emitter
= new EventEmitterAsyncResource({
291 name
: `poolifier:${this.type}-${this.worker}-pool`
296 public get
info (): PoolInfo
{
301 started
: this.started
,
303 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
304 defaultStrategy
: this.opts
.workerChoiceStrategy
!,
305 strategyRetries
: this.workerChoiceStrategiesContext
?.retriesCount
?? 0,
306 minSize
: this.minimumNumberOfWorkers
,
307 maxSize
: this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
,
308 ...(this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
309 .runTime
.aggregate
=== true &&
310 this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
311 .waitTime
.aggregate
&& {
312 utilization
: round(this.utilization
)
314 workerNodes
: this.workerNodes
.length
,
315 idleWorkerNodes
: this.workerNodes
.reduce(
316 (accumulator
, workerNode
) =>
317 workerNode
.usage
.tasks
.executing
=== 0
322 ...(this.opts
.enableTasksQueue
=== true && {
323 stealingWorkerNodes
: this.workerNodes
.reduce(
324 (accumulator
, workerNode
) =>
325 workerNode
.info
.stealing
? accumulator
+ 1 : accumulator
,
329 busyWorkerNodes
: this.workerNodes
.reduce(
330 (accumulator
, _workerNode
, workerNodeKey
) =>
331 this.isWorkerNodeBusy(workerNodeKey
) ? accumulator
+ 1 : accumulator
,
334 executedTasks
: this.workerNodes
.reduce(
335 (accumulator
, workerNode
) =>
336 accumulator
+ workerNode
.usage
.tasks
.executed
,
339 executingTasks
: this.workerNodes
.reduce(
340 (accumulator
, workerNode
) =>
341 accumulator
+ workerNode
.usage
.tasks
.executing
,
344 ...(this.opts
.enableTasksQueue
=== true && {
345 queuedTasks
: this.workerNodes
.reduce(
346 (accumulator
, workerNode
) =>
347 accumulator
+ workerNode
.usage
.tasks
.queued
,
351 ...(this.opts
.enableTasksQueue
=== true && {
352 maxQueuedTasks
: this.workerNodes
.reduce(
353 (accumulator
, workerNode
) =>
354 accumulator
+ (workerNode
.usage
.tasks
.maxQueued
?? 0),
358 ...(this.opts
.enableTasksQueue
=== true && {
359 backPressure
: this.hasBackPressure()
361 ...(this.opts
.enableTasksQueue
=== true && {
362 stolenTasks
: this.workerNodes
.reduce(
363 (accumulator
, workerNode
) =>
364 accumulator
+ workerNode
.usage
.tasks
.stolen
,
368 failedTasks
: this.workerNodes
.reduce(
369 (accumulator
, workerNode
) =>
370 accumulator
+ workerNode
.usage
.tasks
.failed
,
373 ...(this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
374 .runTime
.aggregate
=== true && {
378 ...this.workerNodes
.map(
379 workerNode
=> workerNode
.usage
.runTime
.minimum
?? Infinity
385 ...this.workerNodes
.map(
386 workerNode
=> workerNode
.usage
.runTime
.maximum
?? -Infinity
390 ...(this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
391 .runTime
.average
&& {
394 this.workerNodes
.reduce
<number[]>(
395 (accumulator
, workerNode
) =>
396 accumulator
.concat(workerNode
.usage
.runTime
.history
),
402 ...(this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
406 this.workerNodes
.reduce
<number[]>(
407 (accumulator
, workerNode
) =>
408 accumulator
.concat(workerNode
.usage
.runTime
.history
),
416 ...(this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
417 .waitTime
.aggregate
=== true && {
421 ...this.workerNodes
.map(
422 workerNode
=> workerNode
.usage
.waitTime
.minimum
?? Infinity
428 ...this.workerNodes
.map(
429 workerNode
=> workerNode
.usage
.waitTime
.maximum
?? -Infinity
433 ...(this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
434 .waitTime
.average
&& {
437 this.workerNodes
.reduce
<number[]>(
438 (accumulator
, workerNode
) =>
439 accumulator
.concat(workerNode
.usage
.waitTime
.history
),
445 ...(this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
446 .waitTime
.median
&& {
449 this.workerNodes
.reduce
<number[]>(
450 (accumulator
, workerNode
) =>
451 accumulator
.concat(workerNode
.usage
.waitTime
.history
),
463 * The pool readiness boolean status.
465 private get
ready (): boolean {
470 this.workerNodes
.reduce(
471 (accumulator
, workerNode
) =>
472 !workerNode
.info
.dynamic
&& workerNode
.info
.ready
476 ) >= this.minimumNumberOfWorkers
481 * The pool emptiness boolean status.
483 protected get
empty (): boolean {
484 return this.minimumNumberOfWorkers
=== 0 && this.workerNodes
.length
=== 0
488 * The approximate pool utilization.
490 * @returns The pool utilization.
492 private get
utilization (): number {
493 if (this.startTimestamp
== null) {
496 const poolTimeCapacity
=
497 (performance
.now() - this.startTimestamp
) *
498 (this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
)
499 const totalTasksRunTime
= this.workerNodes
.reduce(
500 (accumulator
, workerNode
) =>
501 accumulator
+ (workerNode
.usage
.runTime
.aggregate
?? 0),
504 const totalTasksWaitTime
= this.workerNodes
.reduce(
505 (accumulator
, workerNode
) =>
506 accumulator
+ (workerNode
.usage
.waitTime
.aggregate
?? 0),
509 return (totalTasksRunTime
+ totalTasksWaitTime
) / poolTimeCapacity
515 * If it is `'dynamic'`, it provides the `max` property.
517 protected abstract get
type (): PoolType
522 protected abstract get
worker (): WorkerType
525 * Checks if the worker id sent in the received message from a worker is valid.
527 * @param message - The received message.
528 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the worker id is invalid.
530 private checkMessageWorkerId (message
: MessageValue
<Data
| Response
>): void {
531 if (message
.workerId
== null) {
532 throw new Error('Worker message received without worker id')
533 } else if (this.getWorkerNodeKeyByWorkerId(message
.workerId
) === -1) {
535 `Worker message received from unknown worker '${message.workerId}'`
541 * Gets the worker node key given its worker id.
543 * @param workerId - The worker id.
544 * @returns The worker node key if the worker id is found in the pool worker nodes, `-1` otherwise.
546 private getWorkerNodeKeyByWorkerId (workerId
: number | undefined): number {
547 return this.workerNodes
.findIndex(
548 workerNode
=> workerNode
.info
.id
=== workerId
553 public setWorkerChoiceStrategy (
554 workerChoiceStrategy
: WorkerChoiceStrategy
,
555 workerChoiceStrategyOptions
?: WorkerChoiceStrategyOptions
557 let requireSync
= false
558 checkValidWorkerChoiceStrategy(workerChoiceStrategy
)
559 if (workerChoiceStrategyOptions
!= null) {
560 requireSync
= !this.setWorkerChoiceStrategyOptions(
561 workerChoiceStrategyOptions
564 if (workerChoiceStrategy
!== this.opts
.workerChoiceStrategy
) {
565 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
566 this.workerChoiceStrategiesContext
?.setDefaultWorkerChoiceStrategy(
567 this.opts
.workerChoiceStrategy
,
568 this.opts
.workerChoiceStrategyOptions
573 this.workerChoiceStrategiesContext
?.syncWorkerChoiceStrategies(
574 this.getWorkerWorkerChoiceStrategies(),
575 this.opts
.workerChoiceStrategyOptions
577 for (const workerNodeKey
of this.workerNodes
.keys()) {
578 this.sendStatisticsMessageToWorker(workerNodeKey
)
584 public setWorkerChoiceStrategyOptions (
585 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
| undefined
587 this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
588 if (workerChoiceStrategyOptions
!= null) {
589 this.opts
.workerChoiceStrategyOptions
= workerChoiceStrategyOptions
590 this.workerChoiceStrategiesContext
?.setOptions(
591 this.opts
.workerChoiceStrategyOptions
593 this.workerChoiceStrategiesContext
?.syncWorkerChoiceStrategies(
594 this.getWorkerWorkerChoiceStrategies(),
595 this.opts
.workerChoiceStrategyOptions
597 for (const workerNodeKey
of this.workerNodes
.keys()) {
598 this.sendStatisticsMessageToWorker(workerNodeKey
)
606 public enableTasksQueue (
608 tasksQueueOptions
?: TasksQueueOptions
610 if (this.opts
.enableTasksQueue
=== true && !enable
) {
611 this.unsetTaskStealing()
612 this.unsetTasksStealingOnBackPressure()
613 this.flushTasksQueues()
615 this.opts
.enableTasksQueue
= enable
616 this.setTasksQueueOptions(tasksQueueOptions
)
620 public setTasksQueueOptions (
621 tasksQueueOptions
: TasksQueueOptions
| undefined
623 if (this.opts
.enableTasksQueue
=== true) {
624 checkValidTasksQueueOptions(tasksQueueOptions
)
625 this.opts
.tasksQueueOptions
=
626 this.buildTasksQueueOptions(tasksQueueOptions
)
627 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
628 this.setTasksQueueSize(this.opts
.tasksQueueOptions
.size
!)
629 if (this.opts
.tasksQueueOptions
.taskStealing
=== true) {
630 this.unsetTaskStealing()
631 this.setTaskStealing()
633 this.unsetTaskStealing()
635 if (this.opts
.tasksQueueOptions
.tasksStealingOnBackPressure
=== true) {
636 this.unsetTasksStealingOnBackPressure()
637 this.setTasksStealingOnBackPressure()
639 this.unsetTasksStealingOnBackPressure()
641 } else if (this.opts
.tasksQueueOptions
!= null) {
642 delete this.opts
.tasksQueueOptions
646 private buildTasksQueueOptions (
647 tasksQueueOptions
: TasksQueueOptions
| undefined
648 ): TasksQueueOptions
{
650 ...getDefaultTasksQueueOptions(
651 this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
657 private setTasksQueueSize (size
: number): void {
658 for (const workerNode
of this.workerNodes
) {
659 workerNode
.tasksQueueBackPressureSize
= size
663 private setTaskStealing (): void {
664 for (const workerNodeKey
of this.workerNodes
.keys()) {
665 this.workerNodes
[workerNodeKey
].on('idle', this.handleWorkerNodeIdleEvent
)
669 private unsetTaskStealing (): void {
670 for (const workerNodeKey
of this.workerNodes
.keys()) {
671 this.workerNodes
[workerNodeKey
].off(
673 this.handleWorkerNodeIdleEvent
678 private setTasksStealingOnBackPressure (): void {
679 for (const workerNodeKey
of this.workerNodes
.keys()) {
680 this.workerNodes
[workerNodeKey
].on(
682 this.handleWorkerNodeBackPressureEvent
687 private unsetTasksStealingOnBackPressure (): void {
688 for (const workerNodeKey
of this.workerNodes
.keys()) {
689 this.workerNodes
[workerNodeKey
].off(
691 this.handleWorkerNodeBackPressureEvent
697 * Whether the pool is full or not.
699 * The pool filling boolean status.
701 protected get
full (): boolean {
703 this.workerNodes
.length
>=
704 (this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
)
709 * Whether the pool is busy or not.
711 * The pool busyness boolean status.
713 protected abstract get
busy (): boolean
716 * Whether worker nodes are executing concurrently their tasks quota or not.
718 * @returns Worker nodes busyness boolean status.
720 protected internalBusy (): boolean {
721 if (this.opts
.enableTasksQueue
=== true) {
723 this.workerNodes
.findIndex(
725 workerNode
.info
.ready
&&
726 workerNode
.usage
.tasks
.executing
<
727 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
728 this.opts
.tasksQueueOptions
!.concurrency
!
733 this.workerNodes
.findIndex(
735 workerNode
.info
.ready
&& workerNode
.usage
.tasks
.executing
=== 0
740 private isWorkerNodeBusy (workerNodeKey
: number): boolean {
741 if (this.opts
.enableTasksQueue
=== true) {
743 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
>=
744 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
745 this.opts
.tasksQueueOptions
!.concurrency
!
748 return this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
> 0
751 private async sendTaskFunctionOperationToWorker (
752 workerNodeKey
: number,
753 message
: MessageValue
<Data
>
754 ): Promise
<boolean> {
755 return await new Promise
<boolean>((resolve
, reject
) => {
756 const taskFunctionOperationListener
= (
757 message
: MessageValue
<Response
>
759 this.checkMessageWorkerId(message
)
760 const workerId
= this.getWorkerInfo(workerNodeKey
)?.id
762 message
.taskFunctionOperationStatus
!= null &&
763 message
.workerId
=== workerId
765 if (message
.taskFunctionOperationStatus
) {
770 `Task function operation '${message.taskFunctionOperation}' failed on worker ${message.workerId} with error: '${message.workerError?.message}'`
774 this.deregisterWorkerMessageListener(
775 this.getWorkerNodeKeyByWorkerId(message
.workerId
),
776 taskFunctionOperationListener
780 this.registerWorkerMessageListener(
782 taskFunctionOperationListener
784 this.sendToWorker(workerNodeKey
, message
)
788 private async sendTaskFunctionOperationToWorkers (
789 message
: MessageValue
<Data
>
790 ): Promise
<boolean> {
791 return await new Promise
<boolean>((resolve
, reject
) => {
792 const responsesReceived
= new Array<MessageValue
<Response
>>()
793 const taskFunctionOperationsListener
= (
794 message
: MessageValue
<Response
>
796 this.checkMessageWorkerId(message
)
797 if (message
.taskFunctionOperationStatus
!= null) {
798 responsesReceived
.push(message
)
799 if (responsesReceived
.length
=== this.workerNodes
.length
) {
801 responsesReceived
.every(
802 message
=> message
.taskFunctionOperationStatus
=== true
807 responsesReceived
.some(
808 message
=> message
.taskFunctionOperationStatus
=== false
811 const errorResponse
= responsesReceived
.find(
812 response
=> response
.taskFunctionOperationStatus
=== false
816 `Task function operation '${
817 message.taskFunctionOperation as string
818 }' failed on worker ${errorResponse?.workerId} with error: '${
819 errorResponse?.workerError?.message
824 this.deregisterWorkerMessageListener(
825 this.getWorkerNodeKeyByWorkerId(message
.workerId
),
826 taskFunctionOperationsListener
831 for (const workerNodeKey
of this.workerNodes
.keys()) {
832 this.registerWorkerMessageListener(
834 taskFunctionOperationsListener
836 this.sendToWorker(workerNodeKey
, message
)
842 public hasTaskFunction (name
: string): boolean {
843 return this.listTaskFunctionsProperties().some(
844 taskFunctionProperties
=> taskFunctionProperties
.name
=== name
849 public async addTaskFunction (
851 fn
: TaskFunction
<Data
, Response
> | TaskFunctionObject
<Data
, Response
>
852 ): Promise
<boolean> {
853 if (typeof name
!== 'string') {
854 throw new TypeError('name argument must be a string')
856 if (typeof name
=== 'string' && name
.trim().length
=== 0) {
857 throw new TypeError('name argument must not be an empty string')
859 if (typeof fn
=== 'function') {
860 fn
= { taskFunction
: fn
} satisfies TaskFunctionObject
<Data
, Response
>
862 if (typeof fn
.taskFunction
!== 'function') {
863 throw new TypeError('taskFunction property must be a function')
865 checkValidPriority(fn
.priority
)
866 checkValidWorkerChoiceStrategy(fn
.strategy
)
867 const opResult
= await this.sendTaskFunctionOperationToWorkers({
868 taskFunctionOperation
: 'add',
869 taskFunctionProperties
: buildTaskFunctionProperties(name
, fn
),
870 taskFunction
: fn
.taskFunction
.toString()
872 this.taskFunctions
.set(name
, fn
)
873 this.workerChoiceStrategiesContext
?.syncWorkerChoiceStrategies(
874 this.getWorkerWorkerChoiceStrategies()
876 for (const workerNodeKey
of this.workerNodes
.keys()) {
877 this.sendStatisticsMessageToWorker(workerNodeKey
)
883 public async removeTaskFunction (name
: string): Promise
<boolean> {
884 if (!this.taskFunctions
.has(name
)) {
886 'Cannot remove a task function not handled on the pool side'
889 const opResult
= await this.sendTaskFunctionOperationToWorkers({
890 taskFunctionOperation
: 'remove',
891 taskFunctionProperties
: buildTaskFunctionProperties(
893 this.taskFunctions
.get(name
)
896 for (const workerNode
of this.workerNodes
) {
897 workerNode
.deleteTaskFunctionWorkerUsage(name
)
899 this.taskFunctions
.delete(name
)
900 this.workerChoiceStrategiesContext
?.syncWorkerChoiceStrategies(
901 this.getWorkerWorkerChoiceStrategies()
903 for (const workerNodeKey
of this.workerNodes
.keys()) {
904 this.sendStatisticsMessageToWorker(workerNodeKey
)
910 public listTaskFunctionsProperties (): TaskFunctionProperties
[] {
911 for (const workerNode
of this.workerNodes
) {
913 Array.isArray(workerNode
.info
.taskFunctionsProperties
) &&
914 workerNode
.info
.taskFunctionsProperties
.length
> 0
916 return workerNode
.info
.taskFunctionsProperties
923 * Gets task function strategy, if any.
925 * @param name - The task function name.
926 * @returns The task function worker choice strategy if the task function worker choice strategy is defined, `undefined` otherwise.
928 private readonly getTaskFunctionWorkerWorkerChoiceStrategy
= (
930 ): WorkerChoiceStrategy
| undefined => {
932 return this.listTaskFunctionsProperties().find(
933 (taskFunctionProperties
: TaskFunctionProperties
) =>
934 taskFunctionProperties
.name
=== name
940 * Gets worker node task function priority, if any.
942 * @param workerNodeKey - The worker node key.
943 * @param name - The task function name.
944 * @returns The task function worker choice priority if the task function worker choice priority is defined, `undefined` otherwise.
946 private readonly getWorkerNodeTaskFunctionPriority
= (
947 workerNodeKey
: number,
949 ): number | undefined => {
951 return this.getWorkerInfo(workerNodeKey
)?.taskFunctionsProperties
?.find(
952 (taskFunctionProperties
: TaskFunctionProperties
) =>
953 taskFunctionProperties
.name
=== name
959 * Gets the worker choice strategies registered in this pool.
961 * @returns The worker choice strategies.
963 private readonly getWorkerWorkerChoiceStrategies
=
964 (): Set
<WorkerChoiceStrategy
> => {
966 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
967 this.opts
.workerChoiceStrategy
!,
968 ...(this.listTaskFunctionsProperties()
970 (taskFunctionProperties
: TaskFunctionProperties
) =>
971 taskFunctionProperties
.strategy
974 (strategy
: WorkerChoiceStrategy
| undefined) => strategy
!= null
975 ) as WorkerChoiceStrategy
[])
980 public async setDefaultTaskFunction (name
: string): Promise
<boolean> {
981 return await this.sendTaskFunctionOperationToWorkers({
982 taskFunctionOperation
: 'default',
983 taskFunctionProperties
: buildTaskFunctionProperties(
985 this.taskFunctions
.get(name
)
990 private shallExecuteTask (workerNodeKey
: number): boolean {
992 this.tasksQueueSize(workerNodeKey
) === 0 &&
993 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
<
994 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
995 this.opts
.tasksQueueOptions
!.concurrency
!
1000 public async execute (
1003 transferList
?: readonly TransferListItem
[]
1004 ): Promise
<Response
> {
1005 return await new Promise
<Response
>((resolve
, reject
) => {
1006 if (!this.started
) {
1007 reject(new Error('Cannot execute a task on not started pool'))
1010 if (this.destroying
) {
1011 reject(new Error('Cannot execute a task on destroying pool'))
1014 if (name
!= null && typeof name
!== 'string') {
1015 reject(new TypeError('name argument must be a string'))
1020 typeof name
=== 'string' &&
1021 name
.trim().length
=== 0
1023 reject(new TypeError('name argument must not be an empty string'))
1026 if (transferList
!= null && !Array.isArray(transferList
)) {
1027 reject(new TypeError('transferList argument must be an array'))
1030 const timestamp
= performance
.now()
1031 const taskFunctionStrategy
=
1032 this.getTaskFunctionWorkerWorkerChoiceStrategy(name
)
1033 const workerNodeKey
= this.chooseWorkerNode(taskFunctionStrategy
)
1034 const task
: Task
<Data
> = {
1035 name
: name
?? DEFAULT_TASK_NAME
,
1036 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
1037 data
: data
?? ({} as Data
),
1038 priority
: this.getWorkerNodeTaskFunctionPriority(workerNodeKey
, name
),
1039 strategy
: taskFunctionStrategy
,
1042 taskId
: randomUUID()
1044 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1045 this.promiseResponseMap
.set(task
.taskId
!, {
1049 ...(this.emitter
!= null && {
1050 asyncResource
: new AsyncResource('poolifier:task', {
1051 triggerAsyncId
: this.emitter
.asyncId
,
1052 requireManualDestroy
: true
1057 this.opts
.enableTasksQueue
=== false ||
1058 (this.opts
.enableTasksQueue
=== true &&
1059 this.shallExecuteTask(workerNodeKey
))
1061 this.executeTask(workerNodeKey
, task
)
1063 this.enqueueTask(workerNodeKey
, task
)
1069 * Starts the minimum number of workers.
1071 private startMinimumNumberOfWorkers (initWorkerNodeUsage
= false): void {
1072 this.startingMinimumNumberOfWorkers
= true
1074 this.workerNodes
.reduce(
1075 (accumulator
, workerNode
) =>
1076 !workerNode
.info
.dynamic
? accumulator
+ 1 : accumulator
,
1078 ) < this.minimumNumberOfWorkers
1080 const workerNodeKey
= this.createAndSetupWorkerNode()
1081 initWorkerNodeUsage
&&
1082 this.initWorkerNodeUsage(this.workerNodes
[workerNodeKey
])
1084 this.startingMinimumNumberOfWorkers
= false
1088 public start (): void {
1090 throw new Error('Cannot start an already started pool')
1092 if (this.starting
) {
1093 throw new Error('Cannot start an already starting pool')
1095 if (this.destroying
) {
1096 throw new Error('Cannot start a destroying pool')
1098 this.starting
= true
1099 this.startMinimumNumberOfWorkers()
1100 this.startTimestamp
= performance
.now()
1101 this.starting
= false
1106 public async destroy (): Promise
<void> {
1107 if (!this.started
) {
1108 throw new Error('Cannot destroy an already destroyed pool')
1110 if (this.starting
) {
1111 throw new Error('Cannot destroy an starting pool')
1113 if (this.destroying
) {
1114 throw new Error('Cannot destroy an already destroying pool')
1116 this.destroying
= true
1118 this.workerNodes
.map(async (_workerNode
, workerNodeKey
) => {
1119 await this.destroyWorkerNode(workerNodeKey
)
1122 this.emitter
?.emit(PoolEvents
.destroy
, this.info
)
1123 this.emitter
?.emitDestroy()
1124 this.readyEventEmitted
= false
1125 delete this.startTimestamp
1126 this.destroying
= false
1127 this.started
= false
1130 private async sendKillMessageToWorker (workerNodeKey
: number): Promise
<void> {
1131 await new Promise
<void>((resolve
, reject
) => {
1132 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1133 if (this.workerNodes
[workerNodeKey
] == null) {
1137 const killMessageListener
= (message
: MessageValue
<Response
>): void => {
1138 this.checkMessageWorkerId(message
)
1139 if (message
.kill
=== 'success') {
1141 } else if (message
.kill
=== 'failure') {
1144 `Kill message handling failed on worker ${message.workerId}`
1149 // FIXME: should be registered only once
1150 this.registerWorkerMessageListener(workerNodeKey
, killMessageListener
)
1151 this.sendToWorker(workerNodeKey
, { kill
: true })
1156 * Terminates the worker node given its worker node key.
1158 * @param workerNodeKey - The worker node key.
1160 protected async destroyWorkerNode (workerNodeKey
: number): Promise
<void> {
1161 this.flagWorkerNodeAsNotReady(workerNodeKey
)
1162 const flushedTasks
= this.flushTasksQueue(workerNodeKey
)
1163 const workerNode
= this.workerNodes
[workerNodeKey
]
1164 await waitWorkerNodeEvents(
1168 this.opts
.tasksQueueOptions
?.tasksFinishedTimeout
??
1169 getDefaultTasksQueueOptions(
1170 this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
1171 ).tasksFinishedTimeout
1173 await this.sendKillMessageToWorker(workerNodeKey
)
1174 await workerNode
.terminate()
1178 * Setup hook to execute code before worker nodes are created in the abstract constructor.
1179 * Can be overridden.
1183 protected setupHook (): void {
1184 /* Intentionally empty */
1188 * Returns whether the worker is the main worker or not.
1190 * @returns `true` if the worker is the main worker, `false` otherwise.
1192 protected abstract isMain (): boolean
1195 * Hook executed before the worker task execution.
1196 * Can be overridden.
1198 * @param workerNodeKey - The worker node key.
1199 * @param task - The task to execute.
1201 protected beforeTaskExecutionHook (
1202 workerNodeKey
: number,
1205 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1206 if (this.workerNodes
[workerNodeKey
]?.usage
!= null) {
1207 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
1208 ++workerUsage
.tasks
.executing
1209 updateWaitTimeWorkerUsage(
1210 this.workerChoiceStrategiesContext
,
1216 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1217 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1218 this.workerNodes
[workerNodeKey
].getTaskFunctionWorkerUsage(task
.name
!) !=
1221 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1222 const taskFunctionWorkerUsage
= this.workerNodes
[
1224 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1225 ].getTaskFunctionWorkerUsage(task
.name
!)!
1226 ++taskFunctionWorkerUsage
.tasks
.executing
1227 updateWaitTimeWorkerUsage(
1228 this.workerChoiceStrategiesContext
,
1229 taskFunctionWorkerUsage
,
1236 * Hook executed after the worker task execution.
1237 * Can be overridden.
1239 * @param workerNodeKey - The worker node key.
1240 * @param message - The received message.
1242 protected afterTaskExecutionHook (
1243 workerNodeKey
: number,
1244 message
: MessageValue
<Response
>
1246 let needWorkerChoiceStrategiesUpdate
= false
1247 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1248 if (this.workerNodes
[workerNodeKey
]?.usage
!= null) {
1249 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
1250 updateTaskStatisticsWorkerUsage(workerUsage
, message
)
1251 updateRunTimeWorkerUsage(
1252 this.workerChoiceStrategiesContext
,
1256 updateEluWorkerUsage(
1257 this.workerChoiceStrategiesContext
,
1261 needWorkerChoiceStrategiesUpdate
= true
1264 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1265 this.workerNodes
[workerNodeKey
].getTaskFunctionWorkerUsage(
1266 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1267 message
.taskPerformance
!.name
1270 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1271 const taskFunctionWorkerUsage
= this.workerNodes
[
1273 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1274 ].getTaskFunctionWorkerUsage(message
.taskPerformance
!.name
)!
1275 updateTaskStatisticsWorkerUsage(taskFunctionWorkerUsage
, message
)
1276 updateRunTimeWorkerUsage(
1277 this.workerChoiceStrategiesContext
,
1278 taskFunctionWorkerUsage
,
1281 updateEluWorkerUsage(
1282 this.workerChoiceStrategiesContext
,
1283 taskFunctionWorkerUsage
,
1286 needWorkerChoiceStrategiesUpdate
= true
1288 if (needWorkerChoiceStrategiesUpdate
) {
1289 this.workerChoiceStrategiesContext
?.update(workerNodeKey
)
1294 * Whether the worker node shall update its task function worker usage or not.
1296 * @param workerNodeKey - The worker node key.
1297 * @returns `true` if the worker node shall update its task function worker usage, `false` otherwise.
1299 private shallUpdateTaskFunctionWorkerUsage (workerNodeKey
: number): boolean {
1300 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1302 workerInfo
!= null &&
1303 Array.isArray(workerInfo
.taskFunctionsProperties
) &&
1304 workerInfo
.taskFunctionsProperties
.length
> 2
1309 * Chooses a worker node for the next task given the worker choice strategy.
1311 * @param workerChoiceStrategy - The worker choice strategy.
1312 * @returns The chosen worker node key
1314 private chooseWorkerNode (
1315 workerChoiceStrategy
?: WorkerChoiceStrategy
1317 if (this.shallCreateDynamicWorker()) {
1318 const workerNodeKey
= this.createAndSetupDynamicWorkerNode()
1320 this.workerChoiceStrategiesContext
?.getPolicy().dynamicWorkerUsage
===
1323 return workerNodeKey
1326 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1327 return this.workerChoiceStrategiesContext
!.execute(workerChoiceStrategy
)
1331 * Conditions for dynamic worker creation.
1333 * @returns Whether to create a dynamic worker or not.
1335 protected abstract shallCreateDynamicWorker (): boolean
1338 * Sends a message to worker given its worker node key.
1340 * @param workerNodeKey - The worker node key.
1341 * @param message - The message.
1342 * @param transferList - The optional array of transferable objects.
1344 protected abstract sendToWorker (
1345 workerNodeKey
: number,
1346 message
: MessageValue
<Data
>,
1347 transferList
?: readonly TransferListItem
[]
1351 * Initializes the worker node usage with sensible default values gathered during runtime.
1353 * @param workerNode - The worker node.
1355 private initWorkerNodeUsage (workerNode
: IWorkerNode
<Worker
, Data
>): void {
1357 this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
1358 .runTime
.aggregate
=== true
1360 workerNode
.usage
.runTime
.aggregate
= min(
1361 ...this.workerNodes
.map(
1362 workerNode
=> workerNode
.usage
.runTime
.aggregate
?? Infinity
1367 this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
1368 .waitTime
.aggregate
=== true
1370 workerNode
.usage
.waitTime
.aggregate
= min(
1371 ...this.workerNodes
.map(
1372 workerNode
=> workerNode
.usage
.waitTime
.aggregate
?? Infinity
1377 this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements().elu
1380 workerNode
.usage
.elu
.active
.aggregate
= min(
1381 ...this.workerNodes
.map(
1382 workerNode
=> workerNode
.usage
.elu
.active
.aggregate
?? Infinity
1389 * Creates a new, completely set up worker node.
1391 * @returns New, completely set up worker node key.
1393 protected createAndSetupWorkerNode (): number {
1394 const workerNode
= this.createWorkerNode()
1395 workerNode
.registerWorkerEventHandler(
1397 this.opts
.onlineHandler
?? EMPTY_FUNCTION
1399 workerNode
.registerWorkerEventHandler(
1401 this.opts
.messageHandler
?? EMPTY_FUNCTION
1403 workerNode
.registerWorkerEventHandler(
1405 this.opts
.errorHandler
?? EMPTY_FUNCTION
1407 workerNode
.registerOnceWorkerEventHandler('error', (error
: Error) => {
1408 workerNode
.info
.ready
= false
1409 this.emitter
?.emit(PoolEvents
.error
, error
)
1413 this.opts
.restartWorkerOnError
=== true
1415 if (workerNode
.info
.dynamic
) {
1416 this.createAndSetupDynamicWorkerNode()
1417 } else if (!this.startingMinimumNumberOfWorkers
) {
1418 this.startMinimumNumberOfWorkers(true)
1424 this.opts
.enableTasksQueue
=== true
1426 this.redistributeQueuedTasks(this.workerNodes
.indexOf(workerNode
))
1428 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1429 workerNode
?.terminate().catch((error
: unknown
) => {
1430 this.emitter
?.emit(PoolEvents
.error
, error
)
1433 workerNode
.registerWorkerEventHandler(
1435 this.opts
.exitHandler
?? EMPTY_FUNCTION
1437 workerNode
.registerOnceWorkerEventHandler('exit', () => {
1438 this.removeWorkerNode(workerNode
)
1441 !this.startingMinimumNumberOfWorkers
&&
1444 this.startMinimumNumberOfWorkers(true)
1447 const workerNodeKey
= this.addWorkerNode(workerNode
)
1448 this.afterWorkerNodeSetup(workerNodeKey
)
1449 return workerNodeKey
1453 * Creates a new, completely set up dynamic worker node.
1455 * @returns New, completely set up dynamic worker node key.
1457 protected createAndSetupDynamicWorkerNode (): number {
1458 const workerNodeKey
= this.createAndSetupWorkerNode()
1459 this.registerWorkerMessageListener(workerNodeKey
, message
=> {
1460 this.checkMessageWorkerId(message
)
1461 const localWorkerNodeKey
= this.getWorkerNodeKeyByWorkerId(
1464 const workerUsage
= this.workerNodes
[localWorkerNodeKey
]?.usage
1465 // Kill message received from worker
1467 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
1468 (isKillBehavior(KillBehaviors
.SOFT
, message
.kill
) &&
1469 ((this.opts
.enableTasksQueue
=== false &&
1470 workerUsage
.tasks
.executing
=== 0) ||
1471 (this.opts
.enableTasksQueue
=== true &&
1472 workerUsage
.tasks
.executing
=== 0 &&
1473 this.tasksQueueSize(localWorkerNodeKey
) === 0)))
1475 // Flag the worker node as not ready immediately
1476 this.flagWorkerNodeAsNotReady(localWorkerNodeKey
)
1477 this.destroyWorkerNode(localWorkerNodeKey
).catch((error
: unknown
) => {
1478 this.emitter
?.emit(PoolEvents
.error
, error
)
1482 this.sendToWorker(workerNodeKey
, {
1485 if (this.taskFunctions
.size
> 0) {
1486 for (const [taskFunctionName
, taskFunctionObject
] of this.taskFunctions
) {
1487 this.sendTaskFunctionOperationToWorker(workerNodeKey
, {
1488 taskFunctionOperation
: 'add',
1489 taskFunctionProperties
: buildTaskFunctionProperties(
1493 taskFunction
: taskFunctionObject
.taskFunction
.toString()
1494 }).catch((error
: unknown
) => {
1495 this.emitter
?.emit(PoolEvents
.error
, error
)
1499 const workerNode
= this.workerNodes
[workerNodeKey
]
1500 workerNode
.info
.dynamic
= true
1502 this.workerChoiceStrategiesContext
?.getPolicy().dynamicWorkerReady
===
1504 this.workerChoiceStrategiesContext
?.getPolicy().dynamicWorkerUsage
===
1507 workerNode
.info
.ready
= true
1509 this.initWorkerNodeUsage(workerNode
)
1510 this.checkAndEmitDynamicWorkerCreationEvents()
1511 return workerNodeKey
1515 * Registers a listener callback on the worker given its worker node key.
1517 * @param workerNodeKey - The worker node key.
1518 * @param listener - The message listener callback.
1520 protected abstract registerWorkerMessageListener
<
1521 Message
extends Data
| Response
1523 workerNodeKey
: number,
1524 listener
: (message
: MessageValue
<Message
>) => void
1528 * Registers once a listener callback on the worker given its worker node key.
1530 * @param workerNodeKey - The worker node key.
1531 * @param listener - The message listener callback.
1533 protected abstract registerOnceWorkerMessageListener
<
1534 Message
extends Data
| Response
1536 workerNodeKey
: number,
1537 listener
: (message
: MessageValue
<Message
>) => void
1541 * Deregisters a listener callback on the worker given its worker node key.
1543 * @param workerNodeKey - The worker node key.
1544 * @param listener - The message listener callback.
1546 protected abstract deregisterWorkerMessageListener
<
1547 Message
extends Data
| Response
1549 workerNodeKey
: number,
1550 listener
: (message
: MessageValue
<Message
>) => void
1554 * Method hooked up after a worker node has been newly created.
1555 * Can be overridden.
1557 * @param workerNodeKey - The newly created worker node key.
1559 protected afterWorkerNodeSetup (workerNodeKey
: number): void {
1560 // Listen to worker messages.
1561 this.registerWorkerMessageListener(
1563 this.workerMessageListener
1565 // Send the startup message to worker.
1566 this.sendStartupMessageToWorker(workerNodeKey
)
1567 // Send the statistics message to worker.
1568 this.sendStatisticsMessageToWorker(workerNodeKey
)
1569 if (this.opts
.enableTasksQueue
=== true) {
1570 if (this.opts
.tasksQueueOptions
?.taskStealing
=== true) {
1571 this.workerNodes
[workerNodeKey
].on(
1573 this.handleWorkerNodeIdleEvent
1576 if (this.opts
.tasksQueueOptions
?.tasksStealingOnBackPressure
=== true) {
1577 this.workerNodes
[workerNodeKey
].on(
1579 this.handleWorkerNodeBackPressureEvent
1586 * Sends the startup message to worker given its worker node key.
1588 * @param workerNodeKey - The worker node key.
1590 protected abstract sendStartupMessageToWorker (workerNodeKey
: number): void
1593 * Sends the statistics message to worker given its worker node key.
1595 * @param workerNodeKey - The worker node key.
1597 private sendStatisticsMessageToWorker (workerNodeKey
: number): void {
1598 this.sendToWorker(workerNodeKey
, {
1601 this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
1602 .runTime
.aggregate
?? false,
1604 this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
1605 .elu
.aggregate
?? false
1610 private cannotStealTask (): boolean {
1611 return this.workerNodes
.length
<= 1 || this.info
.queuedTasks
=== 0
1614 private handleTask (workerNodeKey
: number, task
: Task
<Data
>): void {
1615 if (this.shallExecuteTask(workerNodeKey
)) {
1616 this.executeTask(workerNodeKey
, task
)
1618 this.enqueueTask(workerNodeKey
, task
)
1622 private redistributeQueuedTasks (sourceWorkerNodeKey
: number): void {
1623 if (sourceWorkerNodeKey
=== -1 || this.cannotStealTask()) {
1626 while (this.tasksQueueSize(sourceWorkerNodeKey
) > 0) {
1627 const destinationWorkerNodeKey
= this.workerNodes
.reduce(
1628 (minWorkerNodeKey
, workerNode
, workerNodeKey
, workerNodes
) => {
1629 return sourceWorkerNodeKey
!== workerNodeKey
&&
1630 workerNode
.info
.ready
&&
1631 workerNode
.usage
.tasks
.queued
<
1632 workerNodes
[minWorkerNodeKey
].usage
.tasks
.queued
1639 destinationWorkerNodeKey
,
1640 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1641 this.dequeueTask(sourceWorkerNodeKey
)!
1646 private updateTaskStolenStatisticsWorkerUsage (
1647 workerNodeKey
: number,
1650 const workerNode
= this.workerNodes
[workerNodeKey
]
1651 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1652 if (workerNode
?.usage
!= null) {
1653 ++workerNode
.usage
.tasks
.stolen
1656 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1657 workerNode
.getTaskFunctionWorkerUsage(taskName
) != null
1659 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1660 ++workerNode
.getTaskFunctionWorkerUsage(taskName
)!.tasks
.stolen
1664 private updateTaskSequentiallyStolenStatisticsWorkerUsage (
1665 workerNodeKey
: number,
1667 previousTaskName
?: string
1669 const workerNode
= this.workerNodes
[workerNodeKey
]
1670 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1671 if (workerNode
?.usage
!= null) {
1672 ++workerNode
.usage
.tasks
.sequentiallyStolen
1675 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1676 workerNode
.getTaskFunctionWorkerUsage(taskName
) != null
1678 const taskFunctionWorkerUsage
=
1679 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1680 workerNode
.getTaskFunctionWorkerUsage(taskName
)!
1682 taskFunctionWorkerUsage
.tasks
.sequentiallyStolen
=== 0 ||
1683 (previousTaskName
!= null &&
1684 previousTaskName
=== taskName
&&
1685 taskFunctionWorkerUsage
.tasks
.sequentiallyStolen
> 0)
1687 ++taskFunctionWorkerUsage
.tasks
.sequentiallyStolen
1688 } else if (taskFunctionWorkerUsage
.tasks
.sequentiallyStolen
> 0) {
1689 taskFunctionWorkerUsage
.tasks
.sequentiallyStolen
= 0
1694 private resetTaskSequentiallyStolenStatisticsWorkerUsage (
1695 workerNodeKey
: number,
1698 const workerNode
= this.workerNodes
[workerNodeKey
]
1699 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1700 if (workerNode
?.usage
!= null) {
1701 workerNode
.usage
.tasks
.sequentiallyStolen
= 0
1704 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1705 workerNode
.getTaskFunctionWorkerUsage(taskName
) != null
1707 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1708 workerNode
.getTaskFunctionWorkerUsage(
1710 )!.tasks
.sequentiallyStolen
= 0
1714 private readonly handleWorkerNodeIdleEvent
= (
1715 eventDetail
: WorkerNodeEventDetail
,
1716 previousStolenTask
?: Task
<Data
>
1718 const { workerNodeKey
} = eventDetail
1719 if (workerNodeKey
== null) {
1721 "WorkerNode event detail 'workerNodeKey' property must be defined"
1724 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1726 this.cannotStealTask() ||
1727 (this.info
.stealingWorkerNodes
?? 0) >
1728 Math.floor(this.workerNodes
.length
/ 2)
1730 if (workerInfo
!= null && previousStolenTask
!= null) {
1731 workerInfo
.stealing
= false
1732 this.resetTaskSequentiallyStolenStatisticsWorkerUsage(
1734 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1735 previousStolenTask
.name
!
1740 const workerNodeTasksUsage
= this.workerNodes
[workerNodeKey
].usage
.tasks
1742 workerInfo
!= null &&
1743 previousStolenTask
!= null &&
1744 (workerNodeTasksUsage
.executing
> 0 ||
1745 this.tasksQueueSize(workerNodeKey
) > 0)
1747 workerInfo
.stealing
= false
1748 this.resetTaskSequentiallyStolenStatisticsWorkerUsage(
1750 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1751 previousStolenTask
.name
!
1755 if (workerInfo
== null) {
1757 `Worker node with key '${workerNodeKey}' not found in pool`
1760 workerInfo
.stealing
= true
1761 const stolenTask
= this.workerNodeStealTask(workerNodeKey
)
1762 if (stolenTask
!= null) {
1763 this.updateTaskSequentiallyStolenStatisticsWorkerUsage(
1765 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1767 previousStolenTask
?.name
1770 sleep(exponentialDelay(workerNodeTasksUsage
.sequentiallyStolen
))
1772 this.handleWorkerNodeIdleEvent(eventDetail
, stolenTask
)
1775 .catch((error
: unknown
) => {
1776 this.emitter
?.emit(PoolEvents
.error
, error
)
1780 private readonly workerNodeStealTask
= (
1781 workerNodeKey
: number
1782 ): Task
<Data
> | undefined => {
1783 const workerNodes
= this.workerNodes
1786 (workerNodeA
, workerNodeB
) =>
1787 workerNodeB
.usage
.tasks
.queued
- workerNodeA
.usage
.tasks
.queued
1789 const sourceWorkerNode
= workerNodes
.find(
1790 (sourceWorkerNode
, sourceWorkerNodeKey
) =>
1791 sourceWorkerNode
.info
.ready
&&
1792 !sourceWorkerNode
.info
.stealing
&&
1793 sourceWorkerNodeKey
!== workerNodeKey
&&
1794 sourceWorkerNode
.usage
.tasks
.queued
> 0
1796 if (sourceWorkerNode
!= null) {
1797 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1798 const task
= sourceWorkerNode
.dequeueLastPrioritizedTask()!
1799 this.handleTask(workerNodeKey
, task
)
1800 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1801 this.updateTaskStolenStatisticsWorkerUsage(workerNodeKey
, task
.name
!)
1806 private readonly handleWorkerNodeBackPressureEvent
= (
1807 eventDetail
: WorkerNodeEventDetail
1810 this.cannotStealTask() ||
1811 this.hasBackPressure() ||
1812 (this.info
.stealingWorkerNodes
?? 0) >
1813 Math.floor(this.workerNodes
.length
/ 2)
1817 const { workerId
} = eventDetail
1818 const sizeOffset
= 1
1819 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1820 if (this.opts
.tasksQueueOptions
!.size
! <= sizeOffset
) {
1823 const sourceWorkerNode
=
1824 this.workerNodes
[this.getWorkerNodeKeyByWorkerId(workerId
)]
1825 const workerNodes
= this.workerNodes
1828 (workerNodeA
, workerNodeB
) =>
1829 workerNodeA
.usage
.tasks
.queued
- workerNodeB
.usage
.tasks
.queued
1831 for (const [workerNodeKey
, workerNode
] of workerNodes
.entries()) {
1833 sourceWorkerNode
.usage
.tasks
.queued
> 0 &&
1834 workerNode
.info
.ready
&&
1835 !workerNode
.info
.stealing
&&
1836 workerNode
.info
.id
!== workerId
&&
1837 workerNode
.usage
.tasks
.queued
<
1838 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1839 this.opts
.tasksQueueOptions
!.size
! - sizeOffset
1841 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1842 if (workerInfo
== null) {
1844 `Worker node with key '${workerNodeKey}' not found in pool`
1847 workerInfo
.stealing
= true
1848 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1849 const task
= sourceWorkerNode
.dequeueLastPrioritizedTask()!
1850 this.handleTask(workerNodeKey
, task
)
1851 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1852 this.updateTaskStolenStatisticsWorkerUsage(workerNodeKey
, task
.name
!)
1853 workerInfo
.stealing
= false
1859 * This method is the message listener registered on each worker.
1861 protected readonly workerMessageListener
= (
1862 message
: MessageValue
<Response
>
1864 this.checkMessageWorkerId(message
)
1865 const { workerId
, ready
, taskId
, taskFunctionsProperties
} = message
1866 if (ready
!= null && taskFunctionsProperties
!= null) {
1867 // Worker ready response received from worker
1868 this.handleWorkerReadyResponse(message
)
1869 } else if (taskFunctionsProperties
!= null) {
1870 // Task function properties message received from worker
1871 const workerNodeKey
= this.getWorkerNodeKeyByWorkerId(workerId
)
1872 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1873 if (workerInfo
!= null) {
1874 workerInfo
.taskFunctionsProperties
= taskFunctionsProperties
1875 this.sendStatisticsMessageToWorker(workerNodeKey
)
1877 } else if (taskId
!= null) {
1878 // Task execution response received from worker
1879 this.handleTaskExecutionResponse(message
)
1883 private checkAndEmitReadyEvent (): void {
1884 if (!this.readyEventEmitted
&& this.ready
) {
1885 this.emitter
?.emit(PoolEvents
.ready
, this.info
)
1886 this.readyEventEmitted
= true
1890 private handleWorkerReadyResponse (message
: MessageValue
<Response
>): void {
1891 const { workerId
, ready
, taskFunctionsProperties
} = message
1892 if (ready
== null || !ready
) {
1893 throw new Error(`Worker ${workerId} failed to initialize`)
1895 const workerNodeKey
= this.getWorkerNodeKeyByWorkerId(workerId
)
1896 const workerNode
= this.workerNodes
[workerNodeKey
]
1897 workerNode
.info
.ready
= ready
1898 workerNode
.info
.taskFunctionsProperties
= taskFunctionsProperties
1899 this.sendStatisticsMessageToWorker(workerNodeKey
)
1900 this.checkAndEmitReadyEvent()
1903 private handleTaskExecutionResponse (message
: MessageValue
<Response
>): void {
1904 const { workerId
, taskId
, workerError
, data
} = message
1905 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1906 const promiseResponse
= this.promiseResponseMap
.get(taskId
!)
1907 if (promiseResponse
!= null) {
1908 const { resolve
, reject
, workerNodeKey
, asyncResource
} = promiseResponse
1909 const workerNode
= this.workerNodes
[workerNodeKey
]
1910 if (workerError
!= null) {
1911 this.emitter
?.emit(PoolEvents
.taskError
, workerError
)
1912 asyncResource
!= null
1913 ? asyncResource
.runInAsyncScope(
1918 : reject(workerError
.message
)
1920 asyncResource
!= null
1921 ? asyncResource
.runInAsyncScope(resolve
, this.emitter
, data
)
1922 : resolve(data
as Response
)
1924 asyncResource
?.emitDestroy()
1925 this.afterTaskExecutionHook(workerNodeKey
, message
)
1926 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1927 this.promiseResponseMap
.delete(taskId
!)
1928 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1929 workerNode
?.emit('taskFinished', taskId
)
1931 this.opts
.enableTasksQueue
=== true &&
1933 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1936 const workerNodeTasksUsage
= workerNode
.usage
.tasks
1938 this.tasksQueueSize(workerNodeKey
) > 0 &&
1939 workerNodeTasksUsage
.executing
<
1940 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1941 this.opts
.tasksQueueOptions
!.concurrency
!
1943 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1944 this.executeTask(workerNodeKey
, this.dequeueTask(workerNodeKey
)!)
1947 workerNodeTasksUsage
.executing
=== 0 &&
1948 this.tasksQueueSize(workerNodeKey
) === 0 &&
1949 workerNodeTasksUsage
.sequentiallyStolen
=== 0
1951 workerNode
.emit('idle', {
1960 private checkAndEmitTaskExecutionEvents (): void {
1962 this.emitter
?.emit(PoolEvents
.busy
, this.info
)
1966 private checkAndEmitTaskQueuingEvents (): void {
1967 if (this.hasBackPressure()) {
1968 this.emitter
?.emit(PoolEvents
.backPressure
, this.info
)
1973 * Emits dynamic worker creation events.
1975 protected abstract checkAndEmitDynamicWorkerCreationEvents (): void
1978 * Gets the worker information given its worker node key.
1980 * @param workerNodeKey - The worker node key.
1981 * @returns The worker information.
1983 protected getWorkerInfo (workerNodeKey
: number): WorkerInfo
| undefined {
1984 return this.workerNodes
[workerNodeKey
]?.info
1988 * Creates a worker node.
1990 * @returns The created worker node.
1992 private createWorkerNode (): IWorkerNode
<Worker
, Data
> {
1993 const workerNode
= new WorkerNode
<Worker
, Data
>(
1998 workerOptions
: this.opts
.workerOptions
,
1999 tasksQueueBackPressureSize
:
2000 this.opts
.tasksQueueOptions
?.size
??
2001 getDefaultTasksQueueOptions(
2002 this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
2004 tasksQueueBucketSize
:
2005 (this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
) * 2
2008 // Flag the worker node as ready at pool startup.
2009 if (this.starting
) {
2010 workerNode
.info
.ready
= true
2016 * Adds the given worker node in the pool worker nodes.
2018 * @param workerNode - The worker node.
2019 * @returns The added worker node key.
2020 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the added worker node is not found.
2022 private addWorkerNode (workerNode
: IWorkerNode
<Worker
, Data
>): number {
2023 this.workerNodes
.push(workerNode
)
2024 const workerNodeKey
= this.workerNodes
.indexOf(workerNode
)
2025 if (workerNodeKey
=== -1) {
2026 throw new Error('Worker added not found in worker nodes')
2028 return workerNodeKey
2031 private checkAndEmitEmptyEvent (): void {
2033 this.emitter
?.emit(PoolEvents
.empty
, this.info
)
2034 this.readyEventEmitted
= false
2039 * Removes the worker node from the pool worker nodes.
2041 * @param workerNode - The worker node.
2043 private removeWorkerNode (workerNode
: IWorkerNode
<Worker
, Data
>): void {
2044 const workerNodeKey
= this.workerNodes
.indexOf(workerNode
)
2045 if (workerNodeKey
!== -1) {
2046 this.workerNodes
.splice(workerNodeKey
, 1)
2047 this.workerChoiceStrategiesContext
?.remove(workerNodeKey
)
2049 this.checkAndEmitEmptyEvent()
2052 protected flagWorkerNodeAsNotReady (workerNodeKey
: number): void {
2053 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
2054 if (workerInfo
!= null) {
2055 workerInfo
.ready
= false
2059 private hasBackPressure (): boolean {
2061 this.opts
.enableTasksQueue
=== true &&
2062 this.workerNodes
.findIndex(
2063 workerNode
=> !workerNode
.hasBackPressure()
2069 * Executes the given task on the worker given its worker node key.
2071 * @param workerNodeKey - The worker node key.
2072 * @param task - The task to execute.
2074 private executeTask (workerNodeKey
: number, task
: Task
<Data
>): void {
2075 this.beforeTaskExecutionHook(workerNodeKey
, task
)
2076 this.sendToWorker(workerNodeKey
, task
, task
.transferList
)
2077 this.checkAndEmitTaskExecutionEvents()
2080 private enqueueTask (workerNodeKey
: number, task
: Task
<Data
>): number {
2081 const tasksQueueSize
= this.workerNodes
[workerNodeKey
].enqueueTask(task
)
2082 this.checkAndEmitTaskQueuingEvents()
2083 return tasksQueueSize
2086 private dequeueTask (workerNodeKey
: number): Task
<Data
> | undefined {
2087 return this.workerNodes
[workerNodeKey
].dequeueTask()
2090 private tasksQueueSize (workerNodeKey
: number): number {
2091 return this.workerNodes
[workerNodeKey
].tasksQueueSize()
2094 protected flushTasksQueue (workerNodeKey
: number): number {
2095 let flushedTasks
= 0
2096 while (this.tasksQueueSize(workerNodeKey
) > 0) {
2097 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2098 this.executeTask(workerNodeKey
, this.dequeueTask(workerNodeKey
)!)
2101 this.workerNodes
[workerNodeKey
].clearTasksQueue()
2105 private flushTasksQueues (): void {
2106 for (const workerNodeKey
of this.workerNodes
.keys()) {
2107 this.flushTasksQueue(workerNodeKey
)