1 import { randomUUID
} from
'node:crypto'
2 import { performance
} from
'node:perf_hooks'
3 import { existsSync
} from
'node:fs'
4 import { type TransferListItem
} from
'node:worker_threads'
7 PromiseResponseWrapper
,
9 } from
'../utility-types'
12 DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
,
21 updateMeasurementStatistics
23 import { KillBehaviors
} from
'../worker/worker-options'
32 type TasksQueueOptions
42 type MeasurementStatisticsRequirements
,
44 WorkerChoiceStrategies
,
45 type WorkerChoiceStrategy
,
46 type WorkerChoiceStrategyOptions
47 } from
'./selection-strategies/selection-strategies-types'
48 import { WorkerChoiceStrategyContext
} from
'./selection-strategies/worker-choice-strategy-context'
49 import { version
} from
'./version'
50 import { WorkerNode
} from
'./worker-node'
53 * Base class that implements some shared logic for all poolifier pools.
55 * @typeParam Worker - Type of worker which manages this pool.
56 * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
57 * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
59 export abstract class AbstractPool
<
60 Worker
extends IWorker
,
63 > implements IPool
<Worker
, Data
, Response
> {
65 public readonly workerNodes
: Array<IWorkerNode
<Worker
, Data
>> = []
68 public readonly emitter
?: PoolEmitter
71 * The task execution response promise map:
72 * - `key`: The message id of each submitted task.
73 * - `value`: An object that contains the worker, the execution response promise resolve and reject callbacks.
75 * When we receive a message from the worker, we get a map entry with the promise resolve/reject bound to the message id.
77 protected promiseResponseMap
: Map
<string, PromiseResponseWrapper
<Response
>> =
78 new Map
<string, PromiseResponseWrapper
<Response
>>()
81 * Worker choice strategy context referencing a worker choice algorithm implementation.
83 protected workerChoiceStrategyContext
: WorkerChoiceStrategyContext
<
90 * Dynamic pool maximum size property placeholder.
92 protected readonly max
?: number
95 * Whether the pool is started or not.
97 private started
: boolean
99 * Whether the pool is starting or not.
101 private starting
: boolean
103 * The start timestamp of the pool.
105 private readonly startTimestamp
108 * Constructs a new poolifier pool.
110 * @param numberOfWorkers - Number of workers that this pool should manage.
111 * @param filePath - Path to the worker file.
112 * @param opts - Options for the pool.
115 protected readonly numberOfWorkers
: number,
116 protected readonly filePath
: string,
117 protected readonly opts
: PoolOptions
<Worker
>
119 if (!this.isMain()) {
121 'Cannot start a pool from a worker with the same type as the pool'
124 this.checkNumberOfWorkers(this.numberOfWorkers
)
125 this.checkFilePath(this.filePath
)
126 this.checkPoolOptions(this.opts
)
128 this.chooseWorkerNode
= this.chooseWorkerNode
.bind(this)
129 this.executeTask
= this.executeTask
.bind(this)
130 this.enqueueTask
= this.enqueueTask
.bind(this)
132 if (this.opts
.enableEvents
=== true) {
133 this.emitter
= new PoolEmitter()
135 this.workerChoiceStrategyContext
= new WorkerChoiceStrategyContext
<
141 this.opts
.workerChoiceStrategy
,
142 this.opts
.workerChoiceStrategyOptions
148 this.starting
= false
149 if (this.opts
.startWorkers
=== true) {
153 this.startTimestamp
= performance
.now()
156 private checkFilePath (filePath
: string): void {
159 typeof filePath
!== 'string' ||
160 (typeof filePath
=== 'string' && filePath
.trim().length
=== 0)
162 throw new Error('Please specify a file with a worker implementation')
164 if (!existsSync(filePath
)) {
165 throw new Error(`Cannot find the worker file '${filePath}'`)
169 private checkNumberOfWorkers (numberOfWorkers
: number): void {
170 if (numberOfWorkers
== null) {
172 'Cannot instantiate a pool without specifying the number of workers'
174 } else if (!Number.isSafeInteger(numberOfWorkers
)) {
176 'Cannot instantiate a pool with a non safe integer number of workers'
178 } else if (numberOfWorkers
< 0) {
179 throw new RangeError(
180 'Cannot instantiate a pool with a negative number of workers'
182 } else if (this.type === PoolTypes
.fixed
&& numberOfWorkers
=== 0) {
183 throw new RangeError('Cannot instantiate a fixed pool with zero worker')
187 protected checkDynamicPoolSize (min
: number, max
: number): void {
188 if (this.type === PoolTypes
.dynamic
) {
191 'Cannot instantiate a dynamic pool without specifying the maximum pool size'
193 } else if (!Number.isSafeInteger(max
)) {
195 'Cannot instantiate a dynamic pool with a non safe integer maximum pool size'
197 } else if (min
> max
) {
198 throw new RangeError(
199 'Cannot instantiate a dynamic pool with a maximum pool size inferior to the minimum pool size'
201 } else if (max
=== 0) {
202 throw new RangeError(
203 'Cannot instantiate a dynamic pool with a maximum pool size equal to zero'
205 } else if (min
=== max
) {
206 throw new RangeError(
207 'Cannot instantiate a dynamic pool with a minimum pool size equal to the maximum pool size. Use a fixed pool instead'
213 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
214 if (isPlainObject(opts
)) {
215 this.opts
.startWorkers
= opts
.startWorkers
?? true
216 this.checkValidWorkerChoiceStrategy(
217 opts
.workerChoiceStrategy
as WorkerChoiceStrategy
219 this.opts
.workerChoiceStrategy
=
220 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
221 this.checkValidWorkerChoiceStrategyOptions(
222 opts
.workerChoiceStrategyOptions
as WorkerChoiceStrategyOptions
224 this.opts
.workerChoiceStrategyOptions
= {
225 ...DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
,
226 ...opts
.workerChoiceStrategyOptions
228 this.opts
.restartWorkerOnError
= opts
.restartWorkerOnError
?? true
229 this.opts
.enableEvents
= opts
.enableEvents
?? true
230 this.opts
.enableTasksQueue
= opts
.enableTasksQueue
?? false
231 if (this.opts
.enableTasksQueue
) {
232 this.checkValidTasksQueueOptions(
233 opts
.tasksQueueOptions
as TasksQueueOptions
235 this.opts
.tasksQueueOptions
= this.buildTasksQueueOptions(
236 opts
.tasksQueueOptions
as TasksQueueOptions
240 throw new TypeError('Invalid pool options: must be a plain object')
244 private checkValidWorkerChoiceStrategy (
245 workerChoiceStrategy
: WorkerChoiceStrategy
248 workerChoiceStrategy
!= null &&
249 !Object.values(WorkerChoiceStrategies
).includes(workerChoiceStrategy
)
252 `Invalid worker choice strategy '${workerChoiceStrategy}'`
257 private checkValidWorkerChoiceStrategyOptions (
258 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
261 workerChoiceStrategyOptions
!= null &&
262 !isPlainObject(workerChoiceStrategyOptions
)
265 'Invalid worker choice strategy options: must be a plain object'
269 workerChoiceStrategyOptions
?.retries
!= null &&
270 !Number.isSafeInteger(workerChoiceStrategyOptions
.retries
)
273 'Invalid worker choice strategy options: retries must be an integer'
277 workerChoiceStrategyOptions
?.retries
!= null &&
278 workerChoiceStrategyOptions
.retries
< 0
280 throw new RangeError(
281 `Invalid worker choice strategy options: retries '${workerChoiceStrategyOptions.retries}' must be greater or equal than zero`
285 workerChoiceStrategyOptions
?.weights
!= null &&
286 Object.keys(workerChoiceStrategyOptions
.weights
).length
!== this.maxSize
289 'Invalid worker choice strategy options: must have a weight for each worker node'
293 workerChoiceStrategyOptions
?.measurement
!= null &&
294 !Object.values(Measurements
).includes(
295 workerChoiceStrategyOptions
.measurement
299 `Invalid worker choice strategy options: invalid measurement '${workerChoiceStrategyOptions.measurement}'`
304 private checkValidTasksQueueOptions (
305 tasksQueueOptions
: TasksQueueOptions
307 if (tasksQueueOptions
!= null && !isPlainObject(tasksQueueOptions
)) {
308 throw new TypeError('Invalid tasks queue options: must be a plain object')
311 tasksQueueOptions
?.concurrency
!= null &&
312 !Number.isSafeInteger(tasksQueueOptions
.concurrency
)
315 'Invalid worker node tasks concurrency: must be an integer'
319 tasksQueueOptions
?.concurrency
!= null &&
320 tasksQueueOptions
.concurrency
<= 0
322 throw new RangeError(
323 `Invalid worker node tasks concurrency: ${tasksQueueOptions.concurrency} is a negative integer or zero`
327 tasksQueueOptions
?.size
!= null &&
328 !Number.isSafeInteger(tasksQueueOptions
.size
)
331 'Invalid worker node tasks queue size: must be an integer'
334 if (tasksQueueOptions
?.size
!= null && tasksQueueOptions
.size
<= 0) {
335 throw new RangeError(
336 `Invalid worker node tasks queue size: ${tasksQueueOptions.size} is a negative integer or zero`
342 public get
info (): PoolInfo
{
347 started
: this.started
,
349 strategy
: this.opts
.workerChoiceStrategy
as WorkerChoiceStrategy
,
350 minSize
: this.minSize
,
351 maxSize
: this.maxSize
,
352 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
353 .runTime
.aggregate
&&
354 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
355 .waitTime
.aggregate
&& { utilization
: round(this.utilization
) }),
356 workerNodes
: this.workerNodes
.length
,
357 idleWorkerNodes
: this.workerNodes
.reduce(
358 (accumulator
, workerNode
) =>
359 workerNode
.usage
.tasks
.executing
=== 0
364 busyWorkerNodes
: this.workerNodes
.reduce(
365 (accumulator
, workerNode
) =>
366 workerNode
.usage
.tasks
.executing
> 0 ? accumulator
+ 1 : accumulator
,
369 executedTasks
: this.workerNodes
.reduce(
370 (accumulator
, workerNode
) =>
371 accumulator
+ workerNode
.usage
.tasks
.executed
,
374 executingTasks
: this.workerNodes
.reduce(
375 (accumulator
, workerNode
) =>
376 accumulator
+ workerNode
.usage
.tasks
.executing
,
379 ...(this.opts
.enableTasksQueue
=== true && {
380 queuedTasks
: this.workerNodes
.reduce(
381 (accumulator
, workerNode
) =>
382 accumulator
+ workerNode
.usage
.tasks
.queued
,
386 ...(this.opts
.enableTasksQueue
=== true && {
387 maxQueuedTasks
: this.workerNodes
.reduce(
388 (accumulator
, workerNode
) =>
389 accumulator
+ (workerNode
.usage
.tasks
?.maxQueued
?? 0),
393 ...(this.opts
.enableTasksQueue
=== true && {
394 backPressure
: this.hasBackPressure()
396 ...(this.opts
.enableTasksQueue
=== true && {
397 stolenTasks
: this.workerNodes
.reduce(
398 (accumulator
, workerNode
) =>
399 accumulator
+ workerNode
.usage
.tasks
.stolen
,
403 failedTasks
: this.workerNodes
.reduce(
404 (accumulator
, workerNode
) =>
405 accumulator
+ workerNode
.usage
.tasks
.failed
,
408 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
409 .runTime
.aggregate
&& {
413 ...this.workerNodes
.map(
414 workerNode
=> workerNode
.usage
.runTime
?.minimum
?? Infinity
420 ...this.workerNodes
.map(
421 workerNode
=> workerNode
.usage
.runTime
?.maximum
?? -Infinity
425 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
426 .runTime
.average
&& {
429 this.workerNodes
.reduce
<number[]>(
430 (accumulator
, workerNode
) =>
431 accumulator
.concat(workerNode
.usage
.runTime
.history
),
437 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
441 this.workerNodes
.reduce
<number[]>(
442 (accumulator
, workerNode
) =>
443 accumulator
.concat(workerNode
.usage
.runTime
.history
),
451 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
452 .waitTime
.aggregate
&& {
456 ...this.workerNodes
.map(
457 workerNode
=> workerNode
.usage
.waitTime
?.minimum
?? Infinity
463 ...this.workerNodes
.map(
464 workerNode
=> workerNode
.usage
.waitTime
?.maximum
?? -Infinity
468 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
469 .waitTime
.average
&& {
472 this.workerNodes
.reduce
<number[]>(
473 (accumulator
, workerNode
) =>
474 accumulator
.concat(workerNode
.usage
.waitTime
.history
),
480 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
481 .waitTime
.median
&& {
484 this.workerNodes
.reduce
<number[]>(
485 (accumulator
, workerNode
) =>
486 accumulator
.concat(workerNode
.usage
.waitTime
.history
),
498 * The pool readiness boolean status.
500 private get
ready (): boolean {
502 this.workerNodes
.reduce(
503 (accumulator
, workerNode
) =>
504 !workerNode
.info
.dynamic
&& workerNode
.info
.ready
513 * The approximate pool utilization.
515 * @returns The pool utilization.
517 private get
utilization (): number {
518 const poolTimeCapacity
=
519 (performance
.now() - this.startTimestamp
) * this.maxSize
520 const totalTasksRunTime
= this.workerNodes
.reduce(
521 (accumulator
, workerNode
) =>
522 accumulator
+ (workerNode
.usage
.runTime
?.aggregate
?? 0),
525 const totalTasksWaitTime
= this.workerNodes
.reduce(
526 (accumulator
, workerNode
) =>
527 accumulator
+ (workerNode
.usage
.waitTime
?.aggregate
?? 0),
530 return (totalTasksRunTime
+ totalTasksWaitTime
) / poolTimeCapacity
536 * If it is `'dynamic'`, it provides the `max` property.
538 protected abstract get
type (): PoolType
543 protected abstract get
worker (): WorkerType
546 * The pool minimum size.
548 protected get
minSize (): number {
549 return this.numberOfWorkers
553 * The pool maximum size.
555 protected get
maxSize (): number {
556 return this.max
?? this.numberOfWorkers
560 * Checks if the worker id sent in the received message from a worker is valid.
562 * @param message - The received message.
563 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the worker id is invalid.
565 private checkMessageWorkerId (message
: MessageValue
<Response
>): void {
566 if (message
.workerId
== null) {
567 throw new Error('Worker message received without worker id')
569 message
.workerId
!= null &&
570 this.getWorkerNodeKeyByWorkerId(message
.workerId
) === -1
573 `Worker message received from unknown worker '${message.workerId}'`
579 * Gets the given worker its worker node key.
581 * @param worker - The worker.
582 * @returns The worker node key if found in the pool worker nodes, `-1` otherwise.
584 private getWorkerNodeKeyByWorker (worker
: Worker
): number {
585 return this.workerNodes
.findIndex(
586 workerNode
=> workerNode
.worker
=== worker
591 * Gets the worker node key given its worker id.
593 * @param workerId - The worker id.
594 * @returns The worker node key if the worker id is found in the pool worker nodes, `-1` otherwise.
596 private getWorkerNodeKeyByWorkerId (workerId
: number): number {
597 return this.workerNodes
.findIndex(
598 workerNode
=> workerNode
.info
.id
=== workerId
603 public setWorkerChoiceStrategy (
604 workerChoiceStrategy
: WorkerChoiceStrategy
,
605 workerChoiceStrategyOptions
?: WorkerChoiceStrategyOptions
607 this.checkValidWorkerChoiceStrategy(workerChoiceStrategy
)
608 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
609 this.workerChoiceStrategyContext
.setWorkerChoiceStrategy(
610 this.opts
.workerChoiceStrategy
612 if (workerChoiceStrategyOptions
!= null) {
613 this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
615 for (const [workerNodeKey
, workerNode
] of this.workerNodes
.entries()) {
616 workerNode
.resetUsage()
617 this.sendStatisticsMessageToWorker(workerNodeKey
)
622 public setWorkerChoiceStrategyOptions (
623 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
625 this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
626 this.opts
.workerChoiceStrategyOptions
= {
627 ...DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
,
628 ...workerChoiceStrategyOptions
630 this.workerChoiceStrategyContext
.setOptions(
631 this.opts
.workerChoiceStrategyOptions
636 public enableTasksQueue (
638 tasksQueueOptions
?: TasksQueueOptions
640 if (this.opts
.enableTasksQueue
=== true && !enable
) {
641 this.unsetTaskStealing()
642 this.unsetTasksStealingOnBackPressure()
643 this.flushTasksQueues()
645 this.opts
.enableTasksQueue
= enable
646 this.setTasksQueueOptions(tasksQueueOptions
as TasksQueueOptions
)
650 public setTasksQueueOptions (tasksQueueOptions
: TasksQueueOptions
): void {
651 if (this.opts
.enableTasksQueue
=== true) {
652 this.checkValidTasksQueueOptions(tasksQueueOptions
)
653 this.opts
.tasksQueueOptions
=
654 this.buildTasksQueueOptions(tasksQueueOptions
)
655 this.setTasksQueueSize(this.opts
.tasksQueueOptions
.size
as number)
656 if (this.opts
.tasksQueueOptions
.taskStealing
=== true) {
657 this.setTaskStealing()
659 this.unsetTaskStealing()
661 if (this.opts
.tasksQueueOptions
.tasksStealingOnBackPressure
=== true) {
662 this.setTasksStealingOnBackPressure()
664 this.unsetTasksStealingOnBackPressure()
666 } else if (this.opts
.tasksQueueOptions
!= null) {
667 delete this.opts
.tasksQueueOptions
671 private buildTasksQueueOptions (
672 tasksQueueOptions
: TasksQueueOptions
673 ): TasksQueueOptions
{
676 size
: Math.pow(this.maxSize
, 2),
679 tasksStealingOnBackPressure
: true
685 private setTasksQueueSize (size
: number): void {
686 for (const workerNode
of this.workerNodes
) {
687 workerNode
.tasksQueueBackPressureSize
= size
691 private setTaskStealing (): void {
692 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
693 this.workerNodes
[workerNodeKey
].onEmptyQueue
=
694 this.taskStealingOnEmptyQueue
.bind(this)
698 private unsetTaskStealing (): void {
699 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
700 delete this.workerNodes
[workerNodeKey
].onEmptyQueue
704 private setTasksStealingOnBackPressure (): void {
705 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
706 this.workerNodes
[workerNodeKey
].onBackPressure
=
707 this.tasksStealingOnBackPressure
.bind(this)
711 private unsetTasksStealingOnBackPressure (): void {
712 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
713 delete this.workerNodes
[workerNodeKey
].onBackPressure
718 * Whether the pool is full or not.
720 * The pool filling boolean status.
722 protected get
full (): boolean {
723 return this.workerNodes
.length
>= this.maxSize
727 * Whether the pool is busy or not.
729 * The pool busyness boolean status.
731 protected abstract get
busy (): boolean
734 * Whether worker nodes are executing concurrently their tasks quota or not.
736 * @returns Worker nodes busyness boolean status.
738 protected internalBusy (): boolean {
739 if (this.opts
.enableTasksQueue
=== true) {
741 this.workerNodes
.findIndex(
743 workerNode
.info
.ready
&&
744 workerNode
.usage
.tasks
.executing
<
745 (this.opts
.tasksQueueOptions
?.concurrency
as number)
750 this.workerNodes
.findIndex(
752 workerNode
.info
.ready
&& workerNode
.usage
.tasks
.executing
=== 0
758 public listTaskFunctions (): string[] {
759 for (const workerNode
of this.workerNodes
) {
761 Array.isArray(workerNode
.info
.taskFunctions
) &&
762 workerNode
.info
.taskFunctions
.length
> 0
764 return workerNode
.info
.taskFunctions
770 private shallExecuteTask (workerNodeKey
: number): boolean {
772 this.tasksQueueSize(workerNodeKey
) === 0 &&
773 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
<
774 (this.opts
.tasksQueueOptions
?.concurrency
as number)
779 public async execute (
782 transferList
?: TransferListItem
[]
783 ): Promise
<Response
> {
784 return await new Promise
<Response
>((resolve
, reject
) => {
786 reject(new Error('Cannot execute a task on not started pool'))
789 if (name
!= null && typeof name
!== 'string') {
790 reject(new TypeError('name argument must be a string'))
795 typeof name
=== 'string' &&
796 name
.trim().length
=== 0
798 reject(new TypeError('name argument must not be an empty string'))
801 if (transferList
!= null && !Array.isArray(transferList
)) {
802 reject(new TypeError('transferList argument must be an array'))
805 const timestamp
= performance
.now()
806 const workerNodeKey
= this.chooseWorkerNode()
807 const task
: Task
<Data
> = {
808 name
: name
?? DEFAULT_TASK_NAME
,
809 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
810 data
: data
?? ({} as Data
),
813 workerId
: this.getWorkerInfo(workerNodeKey
).id
as number,
816 this.promiseResponseMap
.set(task
.taskId
as string, {
822 this.opts
.enableTasksQueue
=== false ||
823 (this.opts
.enableTasksQueue
=== true &&
824 this.shallExecuteTask(workerNodeKey
))
826 this.executeTask(workerNodeKey
, task
)
828 this.enqueueTask(workerNodeKey
, task
)
834 public start (): void {
837 this.workerNodes
.reduce(
838 (accumulator
, workerNode
) =>
839 !workerNode
.info
.dynamic
? accumulator
+ 1 : accumulator
,
841 ) < this.numberOfWorkers
843 this.createAndSetupWorkerNode()
845 this.starting
= false
850 public async destroy (): Promise
<void> {
852 this.workerNodes
.map(async (_
, workerNodeKey
) => {
853 await this.destroyWorkerNode(workerNodeKey
)
856 this.emitter
?.emit(PoolEvents
.destroy
, this.info
)
860 protected async sendKillMessageToWorker (
861 workerNodeKey
: number,
864 await new Promise
<void>((resolve
, reject
) => {
865 this.registerWorkerMessageListener(workerNodeKey
, message
=> {
866 if (message
.kill
=== 'success') {
868 } else if (message
.kill
=== 'failure') {
869 reject(new Error(`Worker ${workerId} kill message handling failed`))
872 this.sendToWorker(workerNodeKey
, { kill
: true, workerId
})
877 * Terminates the worker node given its worker node key.
879 * @param workerNodeKey - The worker node key.
881 protected abstract destroyWorkerNode (workerNodeKey
: number): Promise
<void>
884 * Setup hook to execute code before worker nodes are created in the abstract constructor.
889 protected setupHook (): void {
890 /* Intentionally empty */
894 * Should return whether the worker is the main worker or not.
896 protected abstract isMain (): boolean
899 * Hook executed before the worker task execution.
902 * @param workerNodeKey - The worker node key.
903 * @param task - The task to execute.
905 protected beforeTaskExecutionHook (
906 workerNodeKey
: number,
909 if (this.workerNodes
[workerNodeKey
]?.usage
!= null) {
910 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
911 ++workerUsage
.tasks
.executing
912 this.updateWaitTimeWorkerUsage(workerUsage
, task
)
915 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
916 this.workerNodes
[workerNodeKey
].getTaskFunctionWorkerUsage(
920 const taskFunctionWorkerUsage
= this.workerNodes
[
922 ].getTaskFunctionWorkerUsage(task
.name
as string) as WorkerUsage
923 ++taskFunctionWorkerUsage
.tasks
.executing
924 this.updateWaitTimeWorkerUsage(taskFunctionWorkerUsage
, task
)
929 * Hook executed after the worker task execution.
932 * @param workerNodeKey - The worker node key.
933 * @param message - The received message.
935 protected afterTaskExecutionHook (
936 workerNodeKey
: number,
937 message
: MessageValue
<Response
>
939 if (this.workerNodes
[workerNodeKey
]?.usage
!= null) {
940 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
941 this.updateTaskStatisticsWorkerUsage(workerUsage
, message
)
942 this.updateRunTimeWorkerUsage(workerUsage
, message
)
943 this.updateEluWorkerUsage(workerUsage
, message
)
946 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
947 this.workerNodes
[workerNodeKey
].getTaskFunctionWorkerUsage(
948 message
.taskPerformance
?.name
as string
951 const taskFunctionWorkerUsage
= this.workerNodes
[
953 ].getTaskFunctionWorkerUsage(
954 message
.taskPerformance
?.name
as string
956 this.updateTaskStatisticsWorkerUsage(taskFunctionWorkerUsage
, message
)
957 this.updateRunTimeWorkerUsage(taskFunctionWorkerUsage
, message
)
958 this.updateEluWorkerUsage(taskFunctionWorkerUsage
, message
)
963 * Whether the worker node shall update its task function worker usage or not.
965 * @param workerNodeKey - The worker node key.
966 * @returns `true` if the worker node shall update its task function worker usage, `false` otherwise.
968 private shallUpdateTaskFunctionWorkerUsage (workerNodeKey
: number): boolean {
969 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
971 workerInfo
!= null &&
972 Array.isArray(workerInfo
.taskFunctions
) &&
973 workerInfo
.taskFunctions
.length
> 2
977 private updateTaskStatisticsWorkerUsage (
978 workerUsage
: WorkerUsage
,
979 message
: MessageValue
<Response
>
981 const workerTaskStatistics
= workerUsage
.tasks
983 workerTaskStatistics
.executing
!= null &&
984 workerTaskStatistics
.executing
> 0
986 --workerTaskStatistics
.executing
988 if (message
.taskError
== null) {
989 ++workerTaskStatistics
.executed
991 ++workerTaskStatistics
.failed
995 private updateRunTimeWorkerUsage (
996 workerUsage
: WorkerUsage
,
997 message
: MessageValue
<Response
>
999 if (message
.taskError
!= null) {
1002 updateMeasurementStatistics(
1003 workerUsage
.runTime
,
1004 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
,
1005 message
.taskPerformance
?.runTime
?? 0
1009 private updateWaitTimeWorkerUsage (
1010 workerUsage
: WorkerUsage
,
1013 const timestamp
= performance
.now()
1014 const taskWaitTime
= timestamp
- (task
.timestamp
?? timestamp
)
1015 updateMeasurementStatistics(
1016 workerUsage
.waitTime
,
1017 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().waitTime
,
1022 private updateEluWorkerUsage (
1023 workerUsage
: WorkerUsage
,
1024 message
: MessageValue
<Response
>
1026 if (message
.taskError
!= null) {
1029 const eluTaskStatisticsRequirements
: MeasurementStatisticsRequirements
=
1030 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
1031 updateMeasurementStatistics(
1032 workerUsage
.elu
.active
,
1033 eluTaskStatisticsRequirements
,
1034 message
.taskPerformance
?.elu
?.active
?? 0
1036 updateMeasurementStatistics(
1037 workerUsage
.elu
.idle
,
1038 eluTaskStatisticsRequirements
,
1039 message
.taskPerformance
?.elu
?.idle
?? 0
1041 if (eluTaskStatisticsRequirements
.aggregate
) {
1042 if (message
.taskPerformance
?.elu
!= null) {
1043 if (workerUsage
.elu
.utilization
!= null) {
1044 workerUsage
.elu
.utilization
=
1045 (workerUsage
.elu
.utilization
+
1046 message
.taskPerformance
.elu
.utilization
) /
1049 workerUsage
.elu
.utilization
= message
.taskPerformance
.elu
.utilization
1056 * Chooses a worker node for the next task.
1058 * The default worker choice strategy uses a round robin algorithm to distribute the tasks.
1060 * @returns The chosen worker node key
1062 private chooseWorkerNode (): number {
1063 if (this.shallCreateDynamicWorker()) {
1064 const workerNodeKey
= this.createAndSetupDynamicWorkerNode()
1066 this.workerChoiceStrategyContext
.getStrategyPolicy().dynamicWorkerUsage
1068 return workerNodeKey
1071 return this.workerChoiceStrategyContext
.execute()
1075 * Conditions for dynamic worker creation.
1077 * @returns Whether to create a dynamic worker or not.
1079 private shallCreateDynamicWorker (): boolean {
1080 return this.type === PoolTypes
.dynamic
&& !this.full
&& this.internalBusy()
1084 * Sends a message to worker given its worker node key.
1086 * @param workerNodeKey - The worker node key.
1087 * @param message - The message.
1088 * @param transferList - The optional array of transferable objects.
1090 protected abstract sendToWorker (
1091 workerNodeKey
: number,
1092 message
: MessageValue
<Data
>,
1093 transferList
?: TransferListItem
[]
1097 * Creates a new worker.
1099 * @returns Newly created worker.
1101 protected abstract createWorker (): Worker
1104 * Creates a new, completely set up worker node.
1106 * @returns New, completely set up worker node key.
1108 protected createAndSetupWorkerNode (): number {
1109 const worker
= this.createWorker()
1111 worker
.on('online', this.opts
.onlineHandler
?? EMPTY_FUNCTION
)
1112 worker
.on('message', this.opts
.messageHandler
?? EMPTY_FUNCTION
)
1113 worker
.on('error', this.opts
.errorHandler
?? EMPTY_FUNCTION
)
1114 worker
.on('error', error
=> {
1115 const workerNodeKey
= this.getWorkerNodeKeyByWorker(worker
)
1116 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1117 workerInfo
.ready
= false
1118 this.workerNodes
[workerNodeKey
].closeChannel()
1119 this.emitter
?.emit(PoolEvents
.error
, error
)
1123 this.opts
.restartWorkerOnError
=== true
1125 if (workerInfo
.dynamic
) {
1126 this.createAndSetupDynamicWorkerNode()
1128 this.createAndSetupWorkerNode()
1131 if (this.started
&& this.opts
.enableTasksQueue
=== true) {
1132 this.redistributeQueuedTasks(workerNodeKey
)
1135 worker
.on('exit', this.opts
.exitHandler
?? EMPTY_FUNCTION
)
1136 worker
.once('exit', () => {
1137 this.removeWorkerNode(worker
)
1140 const workerNodeKey
= this.addWorkerNode(worker
)
1142 this.afterWorkerNodeSetup(workerNodeKey
)
1144 return workerNodeKey
1148 * Creates a new, completely set up dynamic worker node.
1150 * @returns New, completely set up dynamic worker node key.
1152 protected createAndSetupDynamicWorkerNode (): number {
1153 const workerNodeKey
= this.createAndSetupWorkerNode()
1154 this.registerWorkerMessageListener(workerNodeKey
, message
=> {
1155 const localWorkerNodeKey
= this.getWorkerNodeKeyByWorkerId(
1158 const workerUsage
= this.workerNodes
[localWorkerNodeKey
].usage
1159 // Kill message received from worker
1161 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
1162 (isKillBehavior(KillBehaviors
.SOFT
, message
.kill
) &&
1163 ((this.opts
.enableTasksQueue
=== false &&
1164 workerUsage
.tasks
.executing
=== 0) ||
1165 (this.opts
.enableTasksQueue
=== true &&
1166 workerUsage
.tasks
.executing
=== 0 &&
1167 this.tasksQueueSize(localWorkerNodeKey
) === 0)))
1169 this.destroyWorkerNode(localWorkerNodeKey
).catch(error
=> {
1170 this.emitter
?.emit(PoolEvents
.error
, error
)
1174 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1175 this.sendToWorker(workerNodeKey
, {
1177 workerId
: workerInfo
.id
as number
1179 workerInfo
.dynamic
= true
1181 this.workerChoiceStrategyContext
.getStrategyPolicy().dynamicWorkerReady
||
1182 this.workerChoiceStrategyContext
.getStrategyPolicy().dynamicWorkerUsage
1184 workerInfo
.ready
= true
1186 this.checkAndEmitDynamicWorkerCreationEvents()
1187 return workerNodeKey
1191 * Registers a listener callback on the worker given its worker node key.
1193 * @param workerNodeKey - The worker node key.
1194 * @param listener - The message listener callback.
1196 protected abstract registerWorkerMessageListener
<
1197 Message
extends Data
| Response
1199 workerNodeKey
: number,
1200 listener
: (message
: MessageValue
<Message
>) => void
1204 * Method hooked up after a worker node has been newly created.
1205 * Can be overridden.
1207 * @param workerNodeKey - The newly created worker node key.
1209 protected afterWorkerNodeSetup (workerNodeKey
: number): void {
1210 // Listen to worker messages.
1211 this.registerWorkerMessageListener(workerNodeKey
, this.workerListener())
1212 // Send the startup message to worker.
1213 this.sendStartupMessageToWorker(workerNodeKey
)
1214 // Send the statistics message to worker.
1215 this.sendStatisticsMessageToWorker(workerNodeKey
)
1216 if (this.opts
.enableTasksQueue
=== true) {
1217 if (this.opts
.tasksQueueOptions
?.taskStealing
=== true) {
1218 this.workerNodes
[workerNodeKey
].onEmptyQueue
=
1219 this.taskStealingOnEmptyQueue
.bind(this)
1221 if (this.opts
.tasksQueueOptions
?.tasksStealingOnBackPressure
=== true) {
1222 this.workerNodes
[workerNodeKey
].onBackPressure
=
1223 this.tasksStealingOnBackPressure
.bind(this)
1229 * Sends the startup message to worker given its worker node key.
1231 * @param workerNodeKey - The worker node key.
1233 protected abstract sendStartupMessageToWorker (workerNodeKey
: number): void
1236 * Sends the statistics message to worker given its worker node key.
1238 * @param workerNodeKey - The worker node key.
1240 private sendStatisticsMessageToWorker (workerNodeKey
: number): void {
1241 this.sendToWorker(workerNodeKey
, {
1244 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1246 elu
: this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1249 workerId
: this.getWorkerInfo(workerNodeKey
).id
as number
1253 private redistributeQueuedTasks (workerNodeKey
: number): void {
1254 while (this.tasksQueueSize(workerNodeKey
) > 0) {
1255 const destinationWorkerNodeKey
= this.workerNodes
.reduce(
1256 (minWorkerNodeKey
, workerNode
, workerNodeKey
, workerNodes
) => {
1257 return workerNode
.info
.ready
&&
1258 workerNode
.usage
.tasks
.queued
<
1259 workerNodes
[minWorkerNodeKey
].usage
.tasks
.queued
1265 const destinationWorkerNode
= this.workerNodes
[destinationWorkerNodeKey
]
1267 ...(this.dequeueTask(workerNodeKey
) as Task
<Data
>),
1268 workerId
: destinationWorkerNode
.info
.id
as number
1270 if (this.shallExecuteTask(destinationWorkerNodeKey
)) {
1271 this.executeTask(destinationWorkerNodeKey
, task
)
1273 this.enqueueTask(destinationWorkerNodeKey
, task
)
1278 private updateTaskStolenStatisticsWorkerUsage (
1279 workerNodeKey
: number,
1282 const workerNode
= this.workerNodes
[workerNodeKey
]
1283 if (workerNode
?.usage
!= null) {
1284 ++workerNode
.usage
.tasks
.stolen
1287 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1288 workerNode
.getTaskFunctionWorkerUsage(taskName
) != null
1290 const taskFunctionWorkerUsage
= workerNode
.getTaskFunctionWorkerUsage(
1293 ++taskFunctionWorkerUsage
.tasks
.stolen
1297 private taskStealingOnEmptyQueue (workerId
: number): void {
1298 const destinationWorkerNodeKey
= this.getWorkerNodeKeyByWorkerId(workerId
)
1299 const destinationWorkerNode
= this.workerNodes
[destinationWorkerNodeKey
]
1300 const workerNodes
= this.workerNodes
1303 (workerNodeA
, workerNodeB
) =>
1304 workerNodeB
.usage
.tasks
.queued
- workerNodeA
.usage
.tasks
.queued
1306 const sourceWorkerNode
= workerNodes
.find(
1308 workerNode
.info
.ready
&&
1309 workerNode
.info
.id
!== workerId
&&
1310 workerNode
.usage
.tasks
.queued
> 0
1312 if (sourceWorkerNode
!= null) {
1314 ...(sourceWorkerNode
.popTask() as Task
<Data
>),
1315 workerId
: destinationWorkerNode
.info
.id
as number
1317 if (this.shallExecuteTask(destinationWorkerNodeKey
)) {
1318 this.executeTask(destinationWorkerNodeKey
, task
)
1320 this.enqueueTask(destinationWorkerNodeKey
, task
)
1322 this.updateTaskStolenStatisticsWorkerUsage(
1323 destinationWorkerNodeKey
,
1329 private tasksStealingOnBackPressure (workerId
: number): void {
1330 const sizeOffset
= 1
1331 if ((this.opts
.tasksQueueOptions
?.size
as number) <= sizeOffset
) {
1334 const sourceWorkerNode
=
1335 this.workerNodes
[this.getWorkerNodeKeyByWorkerId(workerId
)]
1336 const workerNodes
= this.workerNodes
1339 (workerNodeA
, workerNodeB
) =>
1340 workerNodeA
.usage
.tasks
.queued
- workerNodeB
.usage
.tasks
.queued
1342 for (const [workerNodeKey
, workerNode
] of workerNodes
.entries()) {
1344 sourceWorkerNode
.usage
.tasks
.queued
> 0 &&
1345 workerNode
.info
.ready
&&
1346 workerNode
.info
.id
!== workerId
&&
1347 workerNode
.usage
.tasks
.queued
<
1348 (this.opts
.tasksQueueOptions
?.size
as number) - sizeOffset
1351 ...(sourceWorkerNode
.popTask() as Task
<Data
>),
1352 workerId
: workerNode
.info
.id
as number
1354 if (this.shallExecuteTask(workerNodeKey
)) {
1355 this.executeTask(workerNodeKey
, task
)
1357 this.enqueueTask(workerNodeKey
, task
)
1359 this.updateTaskStolenStatisticsWorkerUsage(
1368 * This method is the listener registered for each worker message.
1370 * @returns The listener function to execute when a message is received from a worker.
1372 protected workerListener (): (message
: MessageValue
<Response
>) => void {
1374 this.checkMessageWorkerId(message
)
1375 if (message
.ready
!= null && message
.taskFunctions
!= null) {
1376 // Worker ready response received from worker
1377 this.handleWorkerReadyResponse(message
)
1378 } else if (message
.taskId
!= null) {
1379 // Task execution response received from worker
1380 this.handleTaskExecutionResponse(message
)
1381 } else if (message
.taskFunctions
!= null) {
1382 // Task functions message received from worker
1384 this.getWorkerNodeKeyByWorkerId(message
.workerId
)
1385 ).taskFunctions
= message
.taskFunctions
1390 private handleWorkerReadyResponse (message
: MessageValue
<Response
>): void {
1391 if (message
.ready
=== false) {
1392 throw new Error(`Worker ${message.workerId} failed to initialize`)
1394 const workerInfo
= this.getWorkerInfo(
1395 this.getWorkerNodeKeyByWorkerId(message
.workerId
)
1397 workerInfo
.ready
= message
.ready
as boolean
1398 workerInfo
.taskFunctions
= message
.taskFunctions
1400 this.emitter
?.emit(PoolEvents
.ready
, this.info
)
1404 private handleTaskExecutionResponse (message
: MessageValue
<Response
>): void {
1405 const { taskId
, taskError
, data
} = message
1406 const promiseResponse
= this.promiseResponseMap
.get(taskId
as string)
1407 if (promiseResponse
!= null) {
1408 if (taskError
!= null) {
1409 this.emitter
?.emit(PoolEvents
.taskError
, taskError
)
1410 promiseResponse
.reject(taskError
.message
)
1412 promiseResponse
.resolve(data
as Response
)
1414 const workerNodeKey
= promiseResponse
.workerNodeKey
1415 this.afterTaskExecutionHook(workerNodeKey
, message
)
1416 this.workerChoiceStrategyContext
.update(workerNodeKey
)
1417 this.promiseResponseMap
.delete(taskId
as string)
1419 this.opts
.enableTasksQueue
=== true &&
1420 this.tasksQueueSize(workerNodeKey
) > 0 &&
1421 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
<
1422 (this.opts
.tasksQueueOptions
?.concurrency
as number)
1426 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1432 private checkAndEmitTaskExecutionEvents (): void {
1434 this.emitter
?.emit(PoolEvents
.busy
, this.info
)
1438 private checkAndEmitTaskQueuingEvents (): void {
1439 if (this.hasBackPressure()) {
1440 this.emitter
?.emit(PoolEvents
.backPressure
, this.info
)
1444 private checkAndEmitDynamicWorkerCreationEvents (): void {
1445 if (this.type === PoolTypes
.dynamic
) {
1447 this.emitter
?.emit(PoolEvents
.full
, this.info
)
1453 * Gets the worker information given its worker node key.
1455 * @param workerNodeKey - The worker node key.
1456 * @returns The worker information.
1458 protected getWorkerInfo (workerNodeKey
: number): WorkerInfo
{
1459 return this.workerNodes
[workerNodeKey
].info
1463 * Adds the given worker in the pool worker nodes.
1465 * @param worker - The worker.
1466 * @returns The added worker node key.
1467 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the added worker node is not found.
1469 private addWorkerNode (worker
: Worker
): number {
1470 const workerNode
= new WorkerNode
<Worker
, Data
>(
1472 this.opts
.tasksQueueOptions
?.size
?? Math.pow(this.maxSize
, 2)
1474 // Flag the worker node as ready at pool startup.
1475 if (this.starting
) {
1476 workerNode
.info
.ready
= true
1478 this.workerNodes
.push(workerNode
)
1479 const workerNodeKey
= this.getWorkerNodeKeyByWorker(worker
)
1480 if (workerNodeKey
=== -1) {
1481 throw new Error('Worker added not found in worker nodes')
1483 return workerNodeKey
1487 * Removes the given worker from the pool worker nodes.
1489 * @param worker - The worker.
1491 private removeWorkerNode (worker
: Worker
): void {
1492 const workerNodeKey
= this.getWorkerNodeKeyByWorker(worker
)
1493 if (workerNodeKey
!== -1) {
1494 this.workerNodes
.splice(workerNodeKey
, 1)
1495 this.workerChoiceStrategyContext
.remove(workerNodeKey
)
1500 public hasWorkerNodeBackPressure (workerNodeKey
: number): boolean {
1502 this.opts
.enableTasksQueue
=== true &&
1503 this.workerNodes
[workerNodeKey
].hasBackPressure()
1507 private hasBackPressure (): boolean {
1509 this.opts
.enableTasksQueue
=== true &&
1510 this.workerNodes
.findIndex(
1511 workerNode
=> !workerNode
.hasBackPressure()
1517 * Executes the given task on the worker given its worker node key.
1519 * @param workerNodeKey - The worker node key.
1520 * @param task - The task to execute.
1522 private executeTask (workerNodeKey
: number, task
: Task
<Data
>): void {
1523 this.beforeTaskExecutionHook(workerNodeKey
, task
)
1524 this.sendToWorker(workerNodeKey
, task
, task
.transferList
)
1525 this.checkAndEmitTaskExecutionEvents()
1528 private enqueueTask (workerNodeKey
: number, task
: Task
<Data
>): number {
1529 const tasksQueueSize
= this.workerNodes
[workerNodeKey
].enqueueTask(task
)
1530 this.checkAndEmitTaskQueuingEvents()
1531 return tasksQueueSize
1534 private dequeueTask (workerNodeKey
: number): Task
<Data
> | undefined {
1535 return this.workerNodes
[workerNodeKey
].dequeueTask()
1538 private tasksQueueSize (workerNodeKey
: number): number {
1539 return this.workerNodes
[workerNodeKey
].tasksQueueSize()
1542 protected flushTasksQueue (workerNodeKey
: number): void {
1543 while (this.tasksQueueSize(workerNodeKey
) > 0) {
1546 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1549 this.workerNodes
[workerNodeKey
].clearTasksQueue()
1552 private flushTasksQueues (): void {
1553 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
1554 this.flushTasksQueue(workerNodeKey
)