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.opts
.workerChoiceStrategy
=
217 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
218 this.checkValidWorkerChoiceStrategy(this.opts
.workerChoiceStrategy
)
219 this.opts
.workerChoiceStrategyOptions
= {
220 ...DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
,
221 ...opts
.workerChoiceStrategyOptions
223 this.checkValidWorkerChoiceStrategyOptions(
224 this.opts
.workerChoiceStrategyOptions
226 this.opts
.restartWorkerOnError
= opts
.restartWorkerOnError
?? true
227 this.opts
.enableEvents
= opts
.enableEvents
?? true
228 this.opts
.enableTasksQueue
= opts
.enableTasksQueue
?? false
229 if (this.opts
.enableTasksQueue
) {
230 this.checkValidTasksQueueOptions(
231 opts
.tasksQueueOptions
as TasksQueueOptions
233 this.opts
.tasksQueueOptions
= this.buildTasksQueueOptions(
234 opts
.tasksQueueOptions
as TasksQueueOptions
238 throw new TypeError('Invalid pool options: must be a plain object')
242 private checkValidWorkerChoiceStrategy (
243 workerChoiceStrategy
: WorkerChoiceStrategy
245 if (!Object.values(WorkerChoiceStrategies
).includes(workerChoiceStrategy
)) {
247 `Invalid worker choice strategy '${workerChoiceStrategy}'`
252 private checkValidWorkerChoiceStrategyOptions (
253 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
255 if (!isPlainObject(workerChoiceStrategyOptions
)) {
257 'Invalid worker choice strategy options: must be a plain object'
261 workerChoiceStrategyOptions
.retries
!= null &&
262 !Number.isSafeInteger(workerChoiceStrategyOptions
.retries
)
265 'Invalid worker choice strategy options: retries must be an integer'
269 workerChoiceStrategyOptions
.retries
!= null &&
270 workerChoiceStrategyOptions
.retries
< 0
272 throw new RangeError(
273 `Invalid worker choice strategy options: retries '${workerChoiceStrategyOptions.retries}' must be greater or equal than zero`
277 workerChoiceStrategyOptions
.weights
!= null &&
278 Object.keys(workerChoiceStrategyOptions
.weights
).length
!== this.maxSize
281 'Invalid worker choice strategy options: must have a weight for each worker node'
285 workerChoiceStrategyOptions
.measurement
!= null &&
286 !Object.values(Measurements
).includes(
287 workerChoiceStrategyOptions
.measurement
291 `Invalid worker choice strategy options: invalid measurement '${workerChoiceStrategyOptions.measurement}'`
296 private checkValidTasksQueueOptions (
297 tasksQueueOptions
: TasksQueueOptions
299 if (tasksQueueOptions
!= null && !isPlainObject(tasksQueueOptions
)) {
300 throw new TypeError('Invalid tasks queue options: must be a plain object')
303 tasksQueueOptions
?.concurrency
!= null &&
304 !Number.isSafeInteger(tasksQueueOptions
?.concurrency
)
307 'Invalid worker node tasks concurrency: must be an integer'
311 tasksQueueOptions
?.concurrency
!= null &&
312 tasksQueueOptions
?.concurrency
<= 0
314 throw new RangeError(
315 `Invalid worker node tasks concurrency: ${tasksQueueOptions?.concurrency} is a negative integer or zero`
319 tasksQueueOptions
?.size
!= null &&
320 !Number.isSafeInteger(tasksQueueOptions
?.size
)
323 'Invalid worker node tasks queue size: must be an integer'
326 if (tasksQueueOptions
?.size
!= null && tasksQueueOptions
?.size
<= 0) {
327 throw new RangeError(
328 `Invalid worker node tasks queue size: ${tasksQueueOptions?.size} is a negative integer or zero`
334 public get
info (): PoolInfo
{
339 started
: this.started
,
341 strategy
: this.opts
.workerChoiceStrategy
as WorkerChoiceStrategy
,
342 minSize
: this.minSize
,
343 maxSize
: this.maxSize
,
344 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
345 .runTime
.aggregate
&&
346 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
347 .waitTime
.aggregate
&& { utilization
: round(this.utilization
) }),
348 workerNodes
: this.workerNodes
.length
,
349 idleWorkerNodes
: this.workerNodes
.reduce(
350 (accumulator
, workerNode
) =>
351 workerNode
.usage
.tasks
.executing
=== 0
356 busyWorkerNodes
: this.workerNodes
.reduce(
357 (accumulator
, workerNode
) =>
358 workerNode
.usage
.tasks
.executing
> 0 ? accumulator
+ 1 : accumulator
,
361 executedTasks
: this.workerNodes
.reduce(
362 (accumulator
, workerNode
) =>
363 accumulator
+ workerNode
.usage
.tasks
.executed
,
366 executingTasks
: this.workerNodes
.reduce(
367 (accumulator
, workerNode
) =>
368 accumulator
+ workerNode
.usage
.tasks
.executing
,
371 ...(this.opts
.enableTasksQueue
=== true && {
372 queuedTasks
: this.workerNodes
.reduce(
373 (accumulator
, workerNode
) =>
374 accumulator
+ workerNode
.usage
.tasks
.queued
,
378 ...(this.opts
.enableTasksQueue
=== true && {
379 maxQueuedTasks
: this.workerNodes
.reduce(
380 (accumulator
, workerNode
) =>
381 accumulator
+ (workerNode
.usage
.tasks
?.maxQueued
?? 0),
385 ...(this.opts
.enableTasksQueue
=== true && {
386 backPressure
: this.hasBackPressure()
388 ...(this.opts
.enableTasksQueue
=== true && {
389 stolenTasks
: this.workerNodes
.reduce(
390 (accumulator
, workerNode
) =>
391 accumulator
+ workerNode
.usage
.tasks
.stolen
,
395 failedTasks
: this.workerNodes
.reduce(
396 (accumulator
, workerNode
) =>
397 accumulator
+ workerNode
.usage
.tasks
.failed
,
400 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
401 .runTime
.aggregate
&& {
405 ...this.workerNodes
.map(
406 workerNode
=> workerNode
.usage
.runTime
?.minimum
?? Infinity
412 ...this.workerNodes
.map(
413 workerNode
=> workerNode
.usage
.runTime
?.maximum
?? -Infinity
417 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
418 .runTime
.average
&& {
421 this.workerNodes
.reduce
<number[]>(
422 (accumulator
, workerNode
) =>
423 accumulator
.concat(workerNode
.usage
.runTime
.history
),
429 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
433 this.workerNodes
.reduce
<number[]>(
434 (accumulator
, workerNode
) =>
435 accumulator
.concat(workerNode
.usage
.runTime
.history
),
443 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
444 .waitTime
.aggregate
&& {
448 ...this.workerNodes
.map(
449 workerNode
=> workerNode
.usage
.waitTime
?.minimum
?? Infinity
455 ...this.workerNodes
.map(
456 workerNode
=> workerNode
.usage
.waitTime
?.maximum
?? -Infinity
460 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
461 .waitTime
.average
&& {
464 this.workerNodes
.reduce
<number[]>(
465 (accumulator
, workerNode
) =>
466 accumulator
.concat(workerNode
.usage
.waitTime
.history
),
472 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
473 .waitTime
.median
&& {
476 this.workerNodes
.reduce
<number[]>(
477 (accumulator
, workerNode
) =>
478 accumulator
.concat(workerNode
.usage
.waitTime
.history
),
490 * The pool readiness boolean status.
492 private get
ready (): boolean {
494 this.workerNodes
.reduce(
495 (accumulator
, workerNode
) =>
496 !workerNode
.info
.dynamic
&& workerNode
.info
.ready
505 * The approximate pool utilization.
507 * @returns The pool utilization.
509 private get
utilization (): number {
510 const poolTimeCapacity
=
511 (performance
.now() - this.startTimestamp
) * this.maxSize
512 const totalTasksRunTime
= this.workerNodes
.reduce(
513 (accumulator
, workerNode
) =>
514 accumulator
+ (workerNode
.usage
.runTime
?.aggregate
?? 0),
517 const totalTasksWaitTime
= this.workerNodes
.reduce(
518 (accumulator
, workerNode
) =>
519 accumulator
+ (workerNode
.usage
.waitTime
?.aggregate
?? 0),
522 return (totalTasksRunTime
+ totalTasksWaitTime
) / poolTimeCapacity
528 * If it is `'dynamic'`, it provides the `max` property.
530 protected abstract get
type (): PoolType
535 protected abstract get
worker (): WorkerType
538 * The pool minimum size.
540 protected get
minSize (): number {
541 return this.numberOfWorkers
545 * The pool maximum size.
547 protected get
maxSize (): number {
548 return this.max
?? this.numberOfWorkers
552 * Checks if the worker id sent in the received message from a worker is valid.
554 * @param message - The received message.
555 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the worker id is invalid.
557 private checkMessageWorkerId (message
: MessageValue
<Response
>): void {
558 if (message
.workerId
== null) {
559 throw new Error('Worker message received without worker id')
561 message
.workerId
!= null &&
562 this.getWorkerNodeKeyByWorkerId(message
.workerId
) === -1
565 `Worker message received from unknown worker '${message.workerId}'`
571 * Gets the given worker its worker node key.
573 * @param worker - The worker.
574 * @returns The worker node key if found in the pool worker nodes, `-1` otherwise.
576 private getWorkerNodeKeyByWorker (worker
: Worker
): number {
577 return this.workerNodes
.findIndex(
578 workerNode
=> workerNode
.worker
=== worker
583 * Gets the worker node key given its worker id.
585 * @param workerId - The worker id.
586 * @returns The worker node key if the worker id is found in the pool worker nodes, `-1` otherwise.
588 private getWorkerNodeKeyByWorkerId (workerId
: number): number {
589 return this.workerNodes
.findIndex(
590 workerNode
=> workerNode
.info
.id
=== workerId
595 public setWorkerChoiceStrategy (
596 workerChoiceStrategy
: WorkerChoiceStrategy
,
597 workerChoiceStrategyOptions
?: WorkerChoiceStrategyOptions
599 this.checkValidWorkerChoiceStrategy(workerChoiceStrategy
)
600 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
601 this.workerChoiceStrategyContext
.setWorkerChoiceStrategy(
602 this.opts
.workerChoiceStrategy
604 if (workerChoiceStrategyOptions
!= null) {
605 this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
607 for (const [workerNodeKey
, workerNode
] of this.workerNodes
.entries()) {
608 workerNode
.resetUsage()
609 this.sendStatisticsMessageToWorker(workerNodeKey
)
614 public setWorkerChoiceStrategyOptions (
615 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
617 this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
618 this.opts
.workerChoiceStrategyOptions
= {
619 ...DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
,
620 ...workerChoiceStrategyOptions
622 this.workerChoiceStrategyContext
.setOptions(
623 this.opts
.workerChoiceStrategyOptions
628 public enableTasksQueue (
630 tasksQueueOptions
?: TasksQueueOptions
632 if (this.opts
.enableTasksQueue
=== true && !enable
) {
633 this.unsetTaskStealing()
634 this.unsetTasksStealingOnBackPressure()
635 this.flushTasksQueues()
637 this.opts
.enableTasksQueue
= enable
638 this.setTasksQueueOptions(tasksQueueOptions
as TasksQueueOptions
)
642 public setTasksQueueOptions (tasksQueueOptions
: TasksQueueOptions
): void {
643 if (this.opts
.enableTasksQueue
=== true) {
644 this.checkValidTasksQueueOptions(tasksQueueOptions
)
645 this.opts
.tasksQueueOptions
=
646 this.buildTasksQueueOptions(tasksQueueOptions
)
647 this.setTasksQueueSize(this.opts
.tasksQueueOptions
.size
as number)
648 if (this.opts
.tasksQueueOptions
.taskStealing
=== true) {
649 this.setTaskStealing()
651 this.unsetTaskStealing()
653 if (this.opts
.tasksQueueOptions
.tasksStealingOnBackPressure
=== true) {
654 this.setTasksStealingOnBackPressure()
656 this.unsetTasksStealingOnBackPressure()
658 } else if (this.opts
.tasksQueueOptions
!= null) {
659 delete this.opts
.tasksQueueOptions
663 private setTasksQueueSize (size
: number): void {
664 for (const workerNode
of this.workerNodes
) {
665 workerNode
.tasksQueueBackPressureSize
= size
669 private setTaskStealing (): void {
670 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
671 this.workerNodes
[workerNodeKey
].onEmptyQueue
=
672 this.taskStealingOnEmptyQueue
.bind(this)
676 private unsetTaskStealing (): void {
677 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
678 delete this.workerNodes
[workerNodeKey
].onEmptyQueue
682 private setTasksStealingOnBackPressure (): void {
683 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
684 this.workerNodes
[workerNodeKey
].onBackPressure
=
685 this.tasksStealingOnBackPressure
.bind(this)
689 private unsetTasksStealingOnBackPressure (): void {
690 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
691 delete this.workerNodes
[workerNodeKey
].onBackPressure
695 private buildTasksQueueOptions (
696 tasksQueueOptions
: TasksQueueOptions
697 ): TasksQueueOptions
{
700 size
: Math.pow(this.maxSize
, 2),
703 tasksStealingOnBackPressure
: true
710 * Whether the pool is full or not.
712 * The pool filling boolean status.
714 protected get
full (): boolean {
715 return this.workerNodes
.length
>= this.maxSize
719 * Whether the pool is busy or not.
721 * The pool busyness boolean status.
723 protected abstract get
busy (): boolean
726 * Whether worker nodes are executing concurrently their tasks quota or not.
728 * @returns Worker nodes busyness boolean status.
730 protected internalBusy (): boolean {
731 if (this.opts
.enableTasksQueue
=== true) {
733 this.workerNodes
.findIndex(
735 workerNode
.info
.ready
&&
736 workerNode
.usage
.tasks
.executing
<
737 (this.opts
.tasksQueueOptions
?.concurrency
as number)
742 this.workerNodes
.findIndex(
744 workerNode
.info
.ready
&& workerNode
.usage
.tasks
.executing
=== 0
750 public listTaskFunctions (): string[] {
751 for (const workerNode
of this.workerNodes
) {
753 Array.isArray(workerNode
.info
.taskFunctions
) &&
754 workerNode
.info
.taskFunctions
.length
> 0
756 return workerNode
.info
.taskFunctions
762 private shallExecuteTask (workerNodeKey
: number): boolean {
764 this.tasksQueueSize(workerNodeKey
) === 0 &&
765 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
<
766 (this.opts
.tasksQueueOptions
?.concurrency
as number)
771 public async execute (
774 transferList
?: TransferListItem
[]
775 ): Promise
<Response
> {
776 return await new Promise
<Response
>((resolve
, reject
) => {
778 reject(new Error('Cannot execute a task on not started pool'))
781 if (name
!= null && typeof name
!== 'string') {
782 reject(new TypeError('name argument must be a string'))
787 typeof name
=== 'string' &&
788 name
.trim().length
=== 0
790 reject(new TypeError('name argument must not be an empty string'))
793 if (transferList
!= null && !Array.isArray(transferList
)) {
794 reject(new TypeError('transferList argument must be an array'))
797 const timestamp
= performance
.now()
798 const workerNodeKey
= this.chooseWorkerNode()
799 const task
: Task
<Data
> = {
800 name
: name
?? DEFAULT_TASK_NAME
,
801 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
802 data
: data
?? ({} as Data
),
805 workerId
: this.getWorkerInfo(workerNodeKey
).id
as number,
808 this.promiseResponseMap
.set(task
.taskId
as string, {
814 this.opts
.enableTasksQueue
=== false ||
815 (this.opts
.enableTasksQueue
=== true &&
816 this.shallExecuteTask(workerNodeKey
))
818 this.executeTask(workerNodeKey
, task
)
820 this.enqueueTask(workerNodeKey
, task
)
826 public start (): void {
829 this.workerNodes
.reduce(
830 (accumulator
, workerNode
) =>
831 !workerNode
.info
.dynamic
? accumulator
+ 1 : accumulator
,
833 ) < this.numberOfWorkers
835 this.createAndSetupWorkerNode()
837 this.starting
= false
842 public async destroy (): Promise
<void> {
844 this.workerNodes
.map(async (_
, workerNodeKey
) => {
845 await this.destroyWorkerNode(workerNodeKey
)
848 this.emitter
?.emit(PoolEvents
.destroy
, this.info
)
852 protected async sendKillMessageToWorker (
853 workerNodeKey
: number,
856 await new Promise
<void>((resolve
, reject
) => {
857 this.registerWorkerMessageListener(workerNodeKey
, message
=> {
858 if (message
.kill
=== 'success') {
860 } else if (message
.kill
=== 'failure') {
861 reject(new Error(`Worker ${workerId} kill message handling failed`))
864 this.sendToWorker(workerNodeKey
, { kill
: true, workerId
})
869 * Terminates the worker node given its worker node key.
871 * @param workerNodeKey - The worker node key.
873 protected abstract destroyWorkerNode (workerNodeKey
: number): Promise
<void>
876 * Setup hook to execute code before worker nodes are created in the abstract constructor.
881 protected setupHook (): void {
882 /* Intentionally empty */
886 * Should return whether the worker is the main worker or not.
888 protected abstract isMain (): boolean
891 * Hook executed before the worker task execution.
894 * @param workerNodeKey - The worker node key.
895 * @param task - The task to execute.
897 protected beforeTaskExecutionHook (
898 workerNodeKey
: number,
901 if (this.workerNodes
[workerNodeKey
]?.usage
!= null) {
902 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
903 ++workerUsage
.tasks
.executing
904 this.updateWaitTimeWorkerUsage(workerUsage
, task
)
907 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
908 this.workerNodes
[workerNodeKey
].getTaskFunctionWorkerUsage(
912 const taskFunctionWorkerUsage
= this.workerNodes
[
914 ].getTaskFunctionWorkerUsage(task
.name
as string) as WorkerUsage
915 ++taskFunctionWorkerUsage
.tasks
.executing
916 this.updateWaitTimeWorkerUsage(taskFunctionWorkerUsage
, task
)
921 * Hook executed after the worker task execution.
924 * @param workerNodeKey - The worker node key.
925 * @param message - The received message.
927 protected afterTaskExecutionHook (
928 workerNodeKey
: number,
929 message
: MessageValue
<Response
>
931 if (this.workerNodes
[workerNodeKey
]?.usage
!= null) {
932 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
933 this.updateTaskStatisticsWorkerUsage(workerUsage
, message
)
934 this.updateRunTimeWorkerUsage(workerUsage
, message
)
935 this.updateEluWorkerUsage(workerUsage
, message
)
938 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
939 this.workerNodes
[workerNodeKey
].getTaskFunctionWorkerUsage(
940 message
.taskPerformance
?.name
as string
943 const taskFunctionWorkerUsage
= this.workerNodes
[
945 ].getTaskFunctionWorkerUsage(
946 message
.taskPerformance
?.name
as string
948 this.updateTaskStatisticsWorkerUsage(taskFunctionWorkerUsage
, message
)
949 this.updateRunTimeWorkerUsage(taskFunctionWorkerUsage
, message
)
950 this.updateEluWorkerUsage(taskFunctionWorkerUsage
, message
)
955 * Whether the worker node shall update its task function worker usage or not.
957 * @param workerNodeKey - The worker node key.
958 * @returns `true` if the worker node shall update its task function worker usage, `false` otherwise.
960 private shallUpdateTaskFunctionWorkerUsage (workerNodeKey
: number): boolean {
961 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
963 workerInfo
!= null &&
964 Array.isArray(workerInfo
.taskFunctions
) &&
965 workerInfo
.taskFunctions
.length
> 2
969 private updateTaskStatisticsWorkerUsage (
970 workerUsage
: WorkerUsage
,
971 message
: MessageValue
<Response
>
973 const workerTaskStatistics
= workerUsage
.tasks
975 workerTaskStatistics
.executing
!= null &&
976 workerTaskStatistics
.executing
> 0
978 --workerTaskStatistics
.executing
980 if (message
.taskError
== null) {
981 ++workerTaskStatistics
.executed
983 ++workerTaskStatistics
.failed
987 private updateRunTimeWorkerUsage (
988 workerUsage
: WorkerUsage
,
989 message
: MessageValue
<Response
>
991 if (message
.taskError
!= null) {
994 updateMeasurementStatistics(
996 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
,
997 message
.taskPerformance
?.runTime
?? 0
1001 private updateWaitTimeWorkerUsage (
1002 workerUsage
: WorkerUsage
,
1005 const timestamp
= performance
.now()
1006 const taskWaitTime
= timestamp
- (task
.timestamp
?? timestamp
)
1007 updateMeasurementStatistics(
1008 workerUsage
.waitTime
,
1009 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().waitTime
,
1014 private updateEluWorkerUsage (
1015 workerUsage
: WorkerUsage
,
1016 message
: MessageValue
<Response
>
1018 if (message
.taskError
!= null) {
1021 const eluTaskStatisticsRequirements
: MeasurementStatisticsRequirements
=
1022 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
1023 updateMeasurementStatistics(
1024 workerUsage
.elu
.active
,
1025 eluTaskStatisticsRequirements
,
1026 message
.taskPerformance
?.elu
?.active
?? 0
1028 updateMeasurementStatistics(
1029 workerUsage
.elu
.idle
,
1030 eluTaskStatisticsRequirements
,
1031 message
.taskPerformance
?.elu
?.idle
?? 0
1033 if (eluTaskStatisticsRequirements
.aggregate
) {
1034 if (message
.taskPerformance
?.elu
!= null) {
1035 if (workerUsage
.elu
.utilization
!= null) {
1036 workerUsage
.elu
.utilization
=
1037 (workerUsage
.elu
.utilization
+
1038 message
.taskPerformance
.elu
.utilization
) /
1041 workerUsage
.elu
.utilization
= message
.taskPerformance
.elu
.utilization
1048 * Chooses a worker node for the next task.
1050 * The default worker choice strategy uses a round robin algorithm to distribute the tasks.
1052 * @returns The chosen worker node key
1054 private chooseWorkerNode (): number {
1055 if (this.shallCreateDynamicWorker()) {
1056 const workerNodeKey
= this.createAndSetupDynamicWorkerNode()
1058 this.workerChoiceStrategyContext
.getStrategyPolicy().dynamicWorkerUsage
1060 return workerNodeKey
1063 return this.workerChoiceStrategyContext
.execute()
1067 * Conditions for dynamic worker creation.
1069 * @returns Whether to create a dynamic worker or not.
1071 private shallCreateDynamicWorker (): boolean {
1072 return this.type === PoolTypes
.dynamic
&& !this.full
&& this.internalBusy()
1076 * Sends a message to worker given its worker node key.
1078 * @param workerNodeKey - The worker node key.
1079 * @param message - The message.
1080 * @param transferList - The optional array of transferable objects.
1082 protected abstract sendToWorker (
1083 workerNodeKey
: number,
1084 message
: MessageValue
<Data
>,
1085 transferList
?: TransferListItem
[]
1089 * Creates a new worker.
1091 * @returns Newly created worker.
1093 protected abstract createWorker (): Worker
1096 * Creates a new, completely set up worker node.
1098 * @returns New, completely set up worker node key.
1100 protected createAndSetupWorkerNode (): number {
1101 const worker
= this.createWorker()
1103 worker
.on('online', this.opts
.onlineHandler
?? EMPTY_FUNCTION
)
1104 worker
.on('message', this.opts
.messageHandler
?? EMPTY_FUNCTION
)
1105 worker
.on('error', this.opts
.errorHandler
?? EMPTY_FUNCTION
)
1106 worker
.on('error', error
=> {
1107 const workerNodeKey
= this.getWorkerNodeKeyByWorker(worker
)
1108 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1109 workerInfo
.ready
= false
1110 this.workerNodes
[workerNodeKey
].closeChannel()
1111 this.emitter
?.emit(PoolEvents
.error
, error
)
1113 this.opts
.restartWorkerOnError
=== true &&
1117 if (workerInfo
.dynamic
) {
1118 this.createAndSetupDynamicWorkerNode()
1120 this.createAndSetupWorkerNode()
1123 if (this.opts
.enableTasksQueue
=== true) {
1124 this.redistributeQueuedTasks(workerNodeKey
)
1127 worker
.on('exit', this.opts
.exitHandler
?? EMPTY_FUNCTION
)
1128 worker
.once('exit', () => {
1129 this.removeWorkerNode(worker
)
1132 const workerNodeKey
= this.addWorkerNode(worker
)
1134 this.afterWorkerNodeSetup(workerNodeKey
)
1136 return workerNodeKey
1140 * Creates a new, completely set up dynamic worker node.
1142 * @returns New, completely set up dynamic worker node key.
1144 protected createAndSetupDynamicWorkerNode (): number {
1145 const workerNodeKey
= this.createAndSetupWorkerNode()
1146 this.registerWorkerMessageListener(workerNodeKey
, message
=> {
1147 const localWorkerNodeKey
= this.getWorkerNodeKeyByWorkerId(
1150 const workerUsage
= this.workerNodes
[localWorkerNodeKey
].usage
1151 // Kill message received from worker
1153 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
1154 (isKillBehavior(KillBehaviors
.SOFT
, message
.kill
) &&
1155 ((this.opts
.enableTasksQueue
=== false &&
1156 workerUsage
.tasks
.executing
=== 0) ||
1157 (this.opts
.enableTasksQueue
=== true &&
1158 workerUsage
.tasks
.executing
=== 0 &&
1159 this.tasksQueueSize(localWorkerNodeKey
) === 0)))
1161 this.destroyWorkerNode(localWorkerNodeKey
).catch(error
=> {
1162 this.emitter
?.emit(PoolEvents
.error
, error
)
1166 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1167 this.sendToWorker(workerNodeKey
, {
1169 workerId
: workerInfo
.id
as number
1171 workerInfo
.dynamic
= true
1173 this.workerChoiceStrategyContext
.getStrategyPolicy().dynamicWorkerReady
||
1174 this.workerChoiceStrategyContext
.getStrategyPolicy().dynamicWorkerUsage
1176 workerInfo
.ready
= true
1178 this.checkAndEmitDynamicWorkerCreationEvents()
1179 return workerNodeKey
1183 * Registers a listener callback on the worker given its worker node key.
1185 * @param workerNodeKey - The worker node key.
1186 * @param listener - The message listener callback.
1188 protected abstract registerWorkerMessageListener
<
1189 Message
extends Data
| Response
1191 workerNodeKey
: number,
1192 listener
: (message
: MessageValue
<Message
>) => void
1196 * Method hooked up after a worker node has been newly created.
1197 * Can be overridden.
1199 * @param workerNodeKey - The newly created worker node key.
1201 protected afterWorkerNodeSetup (workerNodeKey
: number): void {
1202 // Listen to worker messages.
1203 this.registerWorkerMessageListener(workerNodeKey
, this.workerListener())
1204 // Send the startup message to worker.
1205 this.sendStartupMessageToWorker(workerNodeKey
)
1206 // Send the statistics message to worker.
1207 this.sendStatisticsMessageToWorker(workerNodeKey
)
1208 if (this.opts
.enableTasksQueue
=== true) {
1209 if (this.opts
.tasksQueueOptions
?.taskStealing
=== true) {
1210 this.workerNodes
[workerNodeKey
].onEmptyQueue
=
1211 this.taskStealingOnEmptyQueue
.bind(this)
1213 if (this.opts
.tasksQueueOptions
?.tasksStealingOnBackPressure
=== true) {
1214 this.workerNodes
[workerNodeKey
].onBackPressure
=
1215 this.tasksStealingOnBackPressure
.bind(this)
1221 * Sends the startup message to worker given its worker node key.
1223 * @param workerNodeKey - The worker node key.
1225 protected abstract sendStartupMessageToWorker (workerNodeKey
: number): void
1228 * Sends the statistics message to worker given its worker node key.
1230 * @param workerNodeKey - The worker node key.
1232 private sendStatisticsMessageToWorker (workerNodeKey
: number): void {
1233 this.sendToWorker(workerNodeKey
, {
1236 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1238 elu
: this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1241 workerId
: this.getWorkerInfo(workerNodeKey
).id
as number
1245 private redistributeQueuedTasks (workerNodeKey
: number): void {
1246 while (this.tasksQueueSize(workerNodeKey
) > 0) {
1247 const destinationWorkerNodeKey
= this.workerNodes
.reduce(
1248 (minWorkerNodeKey
, workerNode
, workerNodeKey
, workerNodes
) => {
1249 return workerNode
.info
.ready
&&
1250 workerNode
.usage
.tasks
.queued
<
1251 workerNodes
[minWorkerNodeKey
].usage
.tasks
.queued
1257 const destinationWorkerNode
= this.workerNodes
[destinationWorkerNodeKey
]
1259 ...(this.dequeueTask(workerNodeKey
) as Task
<Data
>),
1260 workerId
: destinationWorkerNode
.info
.id
as number
1262 if (this.shallExecuteTask(destinationWorkerNodeKey
)) {
1263 this.executeTask(destinationWorkerNodeKey
, task
)
1265 this.enqueueTask(destinationWorkerNodeKey
, task
)
1270 private updateTaskStolenStatisticsWorkerUsage (
1271 workerNodeKey
: number,
1274 const workerNode
= this.workerNodes
[workerNodeKey
]
1275 if (workerNode
?.usage
!= null) {
1276 ++workerNode
.usage
.tasks
.stolen
1279 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1280 workerNode
.getTaskFunctionWorkerUsage(taskName
) != null
1282 const taskFunctionWorkerUsage
= workerNode
.getTaskFunctionWorkerUsage(
1285 ++taskFunctionWorkerUsage
.tasks
.stolen
1289 private taskStealingOnEmptyQueue (workerId
: number): void {
1290 const destinationWorkerNodeKey
= this.getWorkerNodeKeyByWorkerId(workerId
)
1291 const destinationWorkerNode
= this.workerNodes
[destinationWorkerNodeKey
]
1292 const workerNodes
= this.workerNodes
1295 (workerNodeA
, workerNodeB
) =>
1296 workerNodeB
.usage
.tasks
.queued
- workerNodeA
.usage
.tasks
.queued
1298 const sourceWorkerNode
= workerNodes
.find(
1300 workerNode
.info
.ready
&&
1301 workerNode
.info
.id
!== workerId
&&
1302 workerNode
.usage
.tasks
.queued
> 0
1304 if (sourceWorkerNode
!= null) {
1306 ...(sourceWorkerNode
.popTask() as Task
<Data
>),
1307 workerId
: destinationWorkerNode
.info
.id
as number
1309 if (this.shallExecuteTask(destinationWorkerNodeKey
)) {
1310 this.executeTask(destinationWorkerNodeKey
, task
)
1312 this.enqueueTask(destinationWorkerNodeKey
, task
)
1314 this.updateTaskStolenStatisticsWorkerUsage(
1315 destinationWorkerNodeKey
,
1321 private tasksStealingOnBackPressure (workerId
: number): void {
1322 const sizeOffset
= 1
1323 if ((this.opts
.tasksQueueOptions
?.size
as number) <= sizeOffset
) {
1326 const sourceWorkerNode
=
1327 this.workerNodes
[this.getWorkerNodeKeyByWorkerId(workerId
)]
1328 const workerNodes
= this.workerNodes
1331 (workerNodeA
, workerNodeB
) =>
1332 workerNodeA
.usage
.tasks
.queued
- workerNodeB
.usage
.tasks
.queued
1334 for (const [workerNodeKey
, workerNode
] of workerNodes
.entries()) {
1336 sourceWorkerNode
.usage
.tasks
.queued
> 0 &&
1337 workerNode
.info
.ready
&&
1338 workerNode
.info
.id
!== workerId
&&
1339 workerNode
.usage
.tasks
.queued
<
1340 (this.opts
.tasksQueueOptions
?.size
as number) - sizeOffset
1343 ...(sourceWorkerNode
.popTask() as Task
<Data
>),
1344 workerId
: workerNode
.info
.id
as number
1346 if (this.shallExecuteTask(workerNodeKey
)) {
1347 this.executeTask(workerNodeKey
, task
)
1349 this.enqueueTask(workerNodeKey
, task
)
1351 this.updateTaskStolenStatisticsWorkerUsage(
1360 * This method is the listener registered for each worker message.
1362 * @returns The listener function to execute when a message is received from a worker.
1364 protected workerListener (): (message
: MessageValue
<Response
>) => void {
1366 this.checkMessageWorkerId(message
)
1367 if (message
.ready
!= null && message
.taskFunctions
!= null) {
1368 // Worker ready response received from worker
1369 this.handleWorkerReadyResponse(message
)
1370 } else if (message
.taskId
!= null) {
1371 // Task execution response received from worker
1372 this.handleTaskExecutionResponse(message
)
1373 } else if (message
.taskFunctions
!= null) {
1374 // Task functions message received from worker
1376 this.getWorkerNodeKeyByWorkerId(message
.workerId
)
1377 ).taskFunctions
= message
.taskFunctions
1382 private handleWorkerReadyResponse (message
: MessageValue
<Response
>): void {
1383 if (message
.ready
=== false) {
1384 throw new Error(`Worker ${message.workerId} failed to initialize`)
1386 const workerInfo
= this.getWorkerInfo(
1387 this.getWorkerNodeKeyByWorkerId(message
.workerId
)
1389 workerInfo
.ready
= message
.ready
as boolean
1390 workerInfo
.taskFunctions
= message
.taskFunctions
1391 if (this.emitter
!= null && this.ready
) {
1392 this.emitter
.emit(PoolEvents
.ready
, this.info
)
1396 private handleTaskExecutionResponse (message
: MessageValue
<Response
>): void {
1397 const { taskId
, taskError
, data
} = message
1398 const promiseResponse
= this.promiseResponseMap
.get(taskId
as string)
1399 if (promiseResponse
!= null) {
1400 if (taskError
!= null) {
1401 this.emitter
?.emit(PoolEvents
.taskError
, taskError
)
1402 promiseResponse
.reject(taskError
.message
)
1404 promiseResponse
.resolve(data
as Response
)
1406 const workerNodeKey
= promiseResponse
.workerNodeKey
1407 this.afterTaskExecutionHook(workerNodeKey
, message
)
1408 this.workerChoiceStrategyContext
.update(workerNodeKey
)
1409 this.promiseResponseMap
.delete(taskId
as string)
1411 this.opts
.enableTasksQueue
=== true &&
1412 this.tasksQueueSize(workerNodeKey
) > 0 &&
1413 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
<
1414 (this.opts
.tasksQueueOptions
?.concurrency
as number)
1418 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1424 private checkAndEmitTaskExecutionEvents (): void {
1426 this.emitter
?.emit(PoolEvents
.busy
, this.info
)
1430 private checkAndEmitTaskQueuingEvents (): void {
1431 if (this.hasBackPressure()) {
1432 this.emitter
?.emit(PoolEvents
.backPressure
, this.info
)
1436 private checkAndEmitDynamicWorkerCreationEvents (): void {
1437 if (this.type === PoolTypes
.dynamic
) {
1439 this.emitter
?.emit(PoolEvents
.full
, this.info
)
1445 * Gets the worker information given its worker node key.
1447 * @param workerNodeKey - The worker node key.
1448 * @returns The worker information.
1450 protected getWorkerInfo (workerNodeKey
: number): WorkerInfo
{
1451 return this.workerNodes
[workerNodeKey
].info
1455 * Adds the given worker in the pool worker nodes.
1457 * @param worker - The worker.
1458 * @returns The added worker node key.
1459 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the added worker node is not found.
1461 private addWorkerNode (worker
: Worker
): number {
1462 const workerNode
= new WorkerNode
<Worker
, Data
>(
1464 this.opts
.tasksQueueOptions
?.size
?? Math.pow(this.maxSize
, 2)
1466 // Flag the worker node as ready at pool startup.
1467 if (this.starting
) {
1468 workerNode
.info
.ready
= true
1470 this.workerNodes
.push(workerNode
)
1471 const workerNodeKey
= this.getWorkerNodeKeyByWorker(worker
)
1472 if (workerNodeKey
=== -1) {
1473 throw new Error('Worker added not found in worker nodes')
1475 return workerNodeKey
1479 * Removes the given worker from the pool worker nodes.
1481 * @param worker - The worker.
1483 private removeWorkerNode (worker
: Worker
): void {
1484 const workerNodeKey
= this.getWorkerNodeKeyByWorker(worker
)
1485 if (workerNodeKey
!== -1) {
1486 this.workerNodes
.splice(workerNodeKey
, 1)
1487 this.workerChoiceStrategyContext
.remove(workerNodeKey
)
1492 public hasWorkerNodeBackPressure (workerNodeKey
: number): boolean {
1494 this.opts
.enableTasksQueue
=== true &&
1495 this.workerNodes
[workerNodeKey
].hasBackPressure()
1499 private hasBackPressure (): boolean {
1501 this.opts
.enableTasksQueue
=== true &&
1502 this.workerNodes
.findIndex(
1503 workerNode
=> !workerNode
.hasBackPressure()
1509 * Executes the given task on the worker given its worker node key.
1511 * @param workerNodeKey - The worker node key.
1512 * @param task - The task to execute.
1514 private executeTask (workerNodeKey
: number, task
: Task
<Data
>): void {
1515 this.beforeTaskExecutionHook(workerNodeKey
, task
)
1516 this.sendToWorker(workerNodeKey
, task
, task
.transferList
)
1517 this.checkAndEmitTaskExecutionEvents()
1520 private enqueueTask (workerNodeKey
: number, task
: Task
<Data
>): number {
1521 const tasksQueueSize
= this.workerNodes
[workerNodeKey
].enqueueTask(task
)
1522 this.checkAndEmitTaskQueuingEvents()
1523 return tasksQueueSize
1526 private dequeueTask (workerNodeKey
: number): Task
<Data
> | undefined {
1527 return this.workerNodes
[workerNodeKey
].dequeueTask()
1530 private tasksQueueSize (workerNodeKey
: number): number {
1531 return this.workerNodes
[workerNodeKey
].tasksQueueSize()
1534 protected flushTasksQueue (workerNodeKey
: number): void {
1535 while (this.tasksQueueSize(workerNodeKey
) > 0) {
1538 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1541 this.workerNodes
[workerNodeKey
].clearTasksQueue()
1544 private flushTasksQueues (): void {
1545 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
1546 this.flushTasksQueue(workerNodeKey
)