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.
73 * - `key`: The message id of each submitted task.
74 * - `value`: An object that contains the worker, the execution response promise resolve and reject callbacks.
76 * When we receive a message from the worker, we get a map entry with the promise resolve/reject bound to the message id.
78 protected promiseResponseMap
: Map
<string, PromiseResponseWrapper
<Response
>> =
79 new Map
<string, PromiseResponseWrapper
<Response
>>()
82 * Worker choice strategy context referencing a worker choice algorithm implementation.
84 protected workerChoiceStrategyContext
: WorkerChoiceStrategyContext
<
91 * Dynamic pool maximum size property placeholder.
93 protected readonly max
?: number
96 * Whether the pool is started or not.
98 private started
: boolean
100 * Whether the pool is starting or not.
102 private starting
: boolean
104 * The start timestamp of the pool.
106 private readonly startTimestamp
109 * Constructs a new poolifier pool.
111 * @param numberOfWorkers - Number of workers that this pool should manage.
112 * @param filePath - Path to the worker file.
113 * @param opts - Options for the pool.
116 protected readonly numberOfWorkers
: number,
117 protected readonly filePath
: string,
118 protected readonly opts
: PoolOptions
<Worker
>
120 if (!this.isMain()) {
122 'Cannot start a pool from a worker with the same type as the pool'
125 this.checkNumberOfWorkers(this.numberOfWorkers
)
126 this.checkFilePath(this.filePath
)
127 this.checkPoolOptions(this.opts
)
129 this.chooseWorkerNode
= this.chooseWorkerNode
.bind(this)
130 this.executeTask
= this.executeTask
.bind(this)
131 this.enqueueTask
= this.enqueueTask
.bind(this)
133 if (this.opts
.enableEvents
=== true) {
134 this.emitter
= new PoolEmitter()
136 this.workerChoiceStrategyContext
= new WorkerChoiceStrategyContext
<
142 this.opts
.workerChoiceStrategy
,
143 this.opts
.workerChoiceStrategyOptions
149 this.starting
= false
150 if (this.opts
.startWorkers
=== true) {
154 this.startTimestamp
= performance
.now()
157 private checkFilePath (filePath
: string): void {
160 typeof filePath
!== 'string' ||
161 (typeof filePath
=== 'string' && filePath
.trim().length
=== 0)
163 throw new Error('Please specify a file with a worker implementation')
165 if (!existsSync(filePath
)) {
166 throw new Error(`Cannot find the worker file '${filePath}'`)
170 private checkNumberOfWorkers (numberOfWorkers
: number): void {
171 if (numberOfWorkers
== null) {
173 'Cannot instantiate a pool without specifying the number of workers'
175 } else if (!Number.isSafeInteger(numberOfWorkers
)) {
177 'Cannot instantiate a pool with a non safe integer number of workers'
179 } else if (numberOfWorkers
< 0) {
180 throw new RangeError(
181 'Cannot instantiate a pool with a negative number of workers'
183 } else if (this.type === PoolTypes
.fixed
&& numberOfWorkers
=== 0) {
184 throw new RangeError('Cannot instantiate a fixed pool with zero worker')
188 protected checkDynamicPoolSize (min
: number, max
: number): void {
189 if (this.type === PoolTypes
.dynamic
) {
192 'Cannot instantiate a dynamic pool without specifying the maximum pool size'
194 } else if (!Number.isSafeInteger(max
)) {
196 'Cannot instantiate a dynamic pool with a non safe integer maximum pool size'
198 } else if (min
> max
) {
199 throw new RangeError(
200 'Cannot instantiate a dynamic pool with a maximum pool size inferior to the minimum pool size'
202 } else if (max
=== 0) {
203 throw new RangeError(
204 'Cannot instantiate a dynamic pool with a maximum pool size equal to zero'
206 } else if (min
=== max
) {
207 throw new RangeError(
208 'Cannot instantiate a dynamic pool with a minimum pool size equal to the maximum pool size. Use a fixed pool instead'
214 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
215 if (isPlainObject(opts
)) {
216 this.opts
.startWorkers
= opts
.startWorkers
?? true
217 this.opts
.workerChoiceStrategy
=
218 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
219 this.checkValidWorkerChoiceStrategy(this.opts
.workerChoiceStrategy
)
220 this.opts
.workerChoiceStrategyOptions
= {
221 ...DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
,
222 ...opts
.workerChoiceStrategyOptions
224 this.checkValidWorkerChoiceStrategyOptions(
225 this.opts
.workerChoiceStrategyOptions
227 this.opts
.restartWorkerOnError
= opts
.restartWorkerOnError
?? true
228 this.opts
.enableEvents
= opts
.enableEvents
?? true
229 this.opts
.enableTasksQueue
= opts
.enableTasksQueue
?? false
230 if (this.opts
.enableTasksQueue
) {
231 this.checkValidTasksQueueOptions(
232 opts
.tasksQueueOptions
as TasksQueueOptions
234 this.opts
.tasksQueueOptions
= this.buildTasksQueueOptions(
235 opts
.tasksQueueOptions
as TasksQueueOptions
239 throw new TypeError('Invalid pool options: must be a plain object')
243 private checkValidWorkerChoiceStrategy (
244 workerChoiceStrategy
: WorkerChoiceStrategy
246 if (!Object.values(WorkerChoiceStrategies
).includes(workerChoiceStrategy
)) {
248 `Invalid worker choice strategy '${workerChoiceStrategy}'`
253 private checkValidWorkerChoiceStrategyOptions (
254 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
256 if (!isPlainObject(workerChoiceStrategyOptions
)) {
258 'Invalid worker choice strategy options: must be a plain object'
262 workerChoiceStrategyOptions
.retries
!= null &&
263 !Number.isSafeInteger(workerChoiceStrategyOptions
.retries
)
266 'Invalid worker choice strategy options: retries must be an integer'
270 workerChoiceStrategyOptions
.retries
!= null &&
271 workerChoiceStrategyOptions
.retries
< 0
273 throw new RangeError(
274 `Invalid worker choice strategy options: retries '${workerChoiceStrategyOptions.retries}' must be greater or equal than zero`
278 workerChoiceStrategyOptions
.weights
!= null &&
279 Object.keys(workerChoiceStrategyOptions
.weights
).length
!== this.maxSize
282 'Invalid worker choice strategy options: must have a weight for each worker node'
286 workerChoiceStrategyOptions
.measurement
!= null &&
287 !Object.values(Measurements
).includes(
288 workerChoiceStrategyOptions
.measurement
292 `Invalid worker choice strategy options: invalid measurement '${workerChoiceStrategyOptions.measurement}'`
297 private checkValidTasksQueueOptions (
298 tasksQueueOptions
: TasksQueueOptions
300 if (tasksQueueOptions
!= null && !isPlainObject(tasksQueueOptions
)) {
301 throw new TypeError('Invalid tasks queue options: must be a plain object')
304 tasksQueueOptions
?.concurrency
!= null &&
305 !Number.isSafeInteger(tasksQueueOptions
?.concurrency
)
308 'Invalid worker node tasks concurrency: must be an integer'
312 tasksQueueOptions
?.concurrency
!= null &&
313 tasksQueueOptions
?.concurrency
<= 0
315 throw new RangeError(
316 `Invalid worker node tasks concurrency: ${tasksQueueOptions?.concurrency} is a negative integer or zero`
320 tasksQueueOptions
?.size
!= null &&
321 !Number.isSafeInteger(tasksQueueOptions
?.size
)
324 'Invalid worker node tasks queue size: must be an integer'
327 if (tasksQueueOptions
?.size
!= null && tasksQueueOptions
?.size
<= 0) {
328 throw new RangeError(
329 `Invalid worker node tasks queue size: ${tasksQueueOptions?.size} is a negative integer or zero`
335 public get
info (): PoolInfo
{
340 started
: this.started
,
342 strategy
: this.opts
.workerChoiceStrategy
as WorkerChoiceStrategy
,
343 minSize
: this.minSize
,
344 maxSize
: this.maxSize
,
345 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
346 .runTime
.aggregate
&&
347 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
348 .waitTime
.aggregate
&& { utilization
: round(this.utilization
) }),
349 workerNodes
: this.workerNodes
.length
,
350 idleWorkerNodes
: this.workerNodes
.reduce(
351 (accumulator
, workerNode
) =>
352 workerNode
.usage
.tasks
.executing
=== 0
357 busyWorkerNodes
: this.workerNodes
.reduce(
358 (accumulator
, workerNode
) =>
359 workerNode
.usage
.tasks
.executing
> 0 ? accumulator
+ 1 : accumulator
,
362 executedTasks
: this.workerNodes
.reduce(
363 (accumulator
, workerNode
) =>
364 accumulator
+ workerNode
.usage
.tasks
.executed
,
367 executingTasks
: this.workerNodes
.reduce(
368 (accumulator
, workerNode
) =>
369 accumulator
+ workerNode
.usage
.tasks
.executing
,
372 ...(this.opts
.enableTasksQueue
=== true && {
373 queuedTasks
: this.workerNodes
.reduce(
374 (accumulator
, workerNode
) =>
375 accumulator
+ workerNode
.usage
.tasks
.queued
,
379 ...(this.opts
.enableTasksQueue
=== true && {
380 maxQueuedTasks
: this.workerNodes
.reduce(
381 (accumulator
, workerNode
) =>
382 accumulator
+ (workerNode
.usage
.tasks
?.maxQueued
?? 0),
386 ...(this.opts
.enableTasksQueue
=== true && {
387 backPressure
: this.hasBackPressure()
389 ...(this.opts
.enableTasksQueue
=== true && {
390 stolenTasks
: this.workerNodes
.reduce(
391 (accumulator
, workerNode
) =>
392 accumulator
+ workerNode
.usage
.tasks
.stolen
,
396 failedTasks
: this.workerNodes
.reduce(
397 (accumulator
, workerNode
) =>
398 accumulator
+ workerNode
.usage
.tasks
.failed
,
401 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
402 .runTime
.aggregate
&& {
406 ...this.workerNodes
.map(
407 workerNode
=> workerNode
.usage
.runTime
?.minimum
?? Infinity
413 ...this.workerNodes
.map(
414 workerNode
=> workerNode
.usage
.runTime
?.maximum
?? -Infinity
418 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
419 .runTime
.average
&& {
422 this.workerNodes
.reduce
<number[]>(
423 (accumulator
, workerNode
) =>
424 accumulator
.concat(workerNode
.usage
.runTime
.history
),
430 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
434 this.workerNodes
.reduce
<number[]>(
435 (accumulator
, workerNode
) =>
436 accumulator
.concat(workerNode
.usage
.runTime
.history
),
444 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
445 .waitTime
.aggregate
&& {
449 ...this.workerNodes
.map(
450 workerNode
=> workerNode
.usage
.waitTime
?.minimum
?? Infinity
456 ...this.workerNodes
.map(
457 workerNode
=> workerNode
.usage
.waitTime
?.maximum
?? -Infinity
461 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
462 .waitTime
.average
&& {
465 this.workerNodes
.reduce
<number[]>(
466 (accumulator
, workerNode
) =>
467 accumulator
.concat(workerNode
.usage
.waitTime
.history
),
473 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
474 .waitTime
.median
&& {
477 this.workerNodes
.reduce
<number[]>(
478 (accumulator
, workerNode
) =>
479 accumulator
.concat(workerNode
.usage
.waitTime
.history
),
491 * The pool readiness boolean status.
493 private get
ready (): boolean {
495 this.workerNodes
.reduce(
496 (accumulator
, workerNode
) =>
497 !workerNode
.info
.dynamic
&& workerNode
.info
.ready
506 * The approximate pool utilization.
508 * @returns The pool utilization.
510 private get
utilization (): number {
511 const poolTimeCapacity
=
512 (performance
.now() - this.startTimestamp
) * this.maxSize
513 const totalTasksRunTime
= this.workerNodes
.reduce(
514 (accumulator
, workerNode
) =>
515 accumulator
+ (workerNode
.usage
.runTime
?.aggregate
?? 0),
518 const totalTasksWaitTime
= this.workerNodes
.reduce(
519 (accumulator
, workerNode
) =>
520 accumulator
+ (workerNode
.usage
.waitTime
?.aggregate
?? 0),
523 return (totalTasksRunTime
+ totalTasksWaitTime
) / poolTimeCapacity
529 * If it is `'dynamic'`, it provides the `max` property.
531 protected abstract get
type (): PoolType
536 protected abstract get
worker (): WorkerType
539 * The pool minimum size.
541 protected get
minSize (): number {
542 return this.numberOfWorkers
546 * The pool maximum size.
548 protected get
maxSize (): number {
549 return this.max
?? this.numberOfWorkers
553 * Checks if the worker id sent in the received message from a worker is valid.
555 * @param message - The received message.
556 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the worker id is invalid.
558 private checkMessageWorkerId (message
: MessageValue
<Response
>): void {
559 if (message
.workerId
== null) {
560 throw new Error('Worker message received without worker id')
562 message
.workerId
!= null &&
563 this.getWorkerNodeKeyByWorkerId(message
.workerId
) === -1
566 `Worker message received from unknown worker '${message.workerId}'`
572 * Gets the given worker its worker node key.
574 * @param worker - The worker.
575 * @returns The worker node key if found in the pool worker nodes, `-1` otherwise.
577 private getWorkerNodeKeyByWorker (worker
: Worker
): number {
578 return this.workerNodes
.findIndex(
579 workerNode
=> workerNode
.worker
=== worker
584 * Gets the worker node key given its worker id.
586 * @param workerId - The worker id.
587 * @returns The worker node key if the worker id is found in the pool worker nodes, `-1` otherwise.
589 private getWorkerNodeKeyByWorkerId (workerId
: number): number {
590 return this.workerNodes
.findIndex(
591 workerNode
=> workerNode
.info
.id
=== workerId
596 public setWorkerChoiceStrategy (
597 workerChoiceStrategy
: WorkerChoiceStrategy
,
598 workerChoiceStrategyOptions
?: WorkerChoiceStrategyOptions
600 this.checkValidWorkerChoiceStrategy(workerChoiceStrategy
)
601 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
602 this.workerChoiceStrategyContext
.setWorkerChoiceStrategy(
603 this.opts
.workerChoiceStrategy
605 if (workerChoiceStrategyOptions
!= null) {
606 this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
608 for (const [workerNodeKey
, workerNode
] of this.workerNodes
.entries()) {
609 workerNode
.resetUsage()
610 this.sendStatisticsMessageToWorker(workerNodeKey
)
615 public setWorkerChoiceStrategyOptions (
616 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
618 this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
619 this.opts
.workerChoiceStrategyOptions
= {
620 ...DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
,
621 ...workerChoiceStrategyOptions
623 this.workerChoiceStrategyContext
.setOptions(
624 this.opts
.workerChoiceStrategyOptions
629 public enableTasksQueue (
631 tasksQueueOptions
?: TasksQueueOptions
633 if (this.opts
.enableTasksQueue
=== true && !enable
) {
634 this.flushTasksQueues()
636 this.opts
.enableTasksQueue
= enable
637 this.setTasksQueueOptions(tasksQueueOptions
as TasksQueueOptions
)
641 public setTasksQueueOptions (tasksQueueOptions
: TasksQueueOptions
): void {
642 if (this.opts
.enableTasksQueue
=== true) {
643 this.checkValidTasksQueueOptions(tasksQueueOptions
)
644 this.opts
.tasksQueueOptions
=
645 this.buildTasksQueueOptions(tasksQueueOptions
)
646 this.setTasksQueueSize(this.opts
.tasksQueueOptions
.size
as number)
647 } else if (this.opts
.tasksQueueOptions
!= null) {
648 delete this.opts
.tasksQueueOptions
652 private setTasksQueueSize (size
: number): void {
653 for (const workerNode
of this.workerNodes
) {
654 workerNode
.tasksQueueBackPressureSize
= size
658 private buildTasksQueueOptions (
659 tasksQueueOptions
: TasksQueueOptions
660 ): TasksQueueOptions
{
663 size
: Math.pow(this.maxSize
, 2),
666 tasksStealingOnBackPressure
: true
673 * Whether the pool is full or not.
675 * The pool filling boolean status.
677 protected get
full (): boolean {
678 return this.workerNodes
.length
>= this.maxSize
682 * Whether the pool is busy or not.
684 * The pool busyness boolean status.
686 protected abstract get
busy (): boolean
689 * Whether worker nodes are executing concurrently their tasks quota or not.
691 * @returns Worker nodes busyness boolean status.
693 protected internalBusy (): boolean {
694 if (this.opts
.enableTasksQueue
=== true) {
696 this.workerNodes
.findIndex(
698 workerNode
.info
.ready
&&
699 workerNode
.usage
.tasks
.executing
<
700 (this.opts
.tasksQueueOptions
?.concurrency
as number)
705 this.workerNodes
.findIndex(
707 workerNode
.info
.ready
&& workerNode
.usage
.tasks
.executing
=== 0
714 public listTaskFunctions (): string[] {
715 for (const workerNode
of this.workerNodes
) {
717 Array.isArray(workerNode
.info
.taskFunctions
) &&
718 workerNode
.info
.taskFunctions
.length
> 0
720 return workerNode
.info
.taskFunctions
726 private shallExecuteTask (workerNodeKey
: number): boolean {
728 this.tasksQueueSize(workerNodeKey
) === 0 &&
729 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
<
730 (this.opts
.tasksQueueOptions
?.concurrency
as number)
735 public async execute (
738 transferList
?: TransferListItem
[]
739 ): Promise
<Response
> {
740 return await new Promise
<Response
>((resolve
, reject
) => {
742 reject(new Error('Cannot execute a task on not started pool'))
745 if (name
!= null && typeof name
!== 'string') {
746 reject(new TypeError('name argument must be a string'))
751 typeof name
=== 'string' &&
752 name
.trim().length
=== 0
754 reject(new TypeError('name argument must not be an empty string'))
757 if (transferList
!= null && !Array.isArray(transferList
)) {
758 reject(new TypeError('transferList argument must be an array'))
761 const timestamp
= performance
.now()
762 const workerNodeKey
= this.chooseWorkerNode()
763 const task
: Task
<Data
> = {
764 name
: name
?? DEFAULT_TASK_NAME
,
765 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
766 data
: data
?? ({} as Data
),
769 workerId
: this.getWorkerInfo(workerNodeKey
).id
as number,
772 this.promiseResponseMap
.set(task
.taskId
as string, {
778 this.opts
.enableTasksQueue
=== false ||
779 (this.opts
.enableTasksQueue
=== true &&
780 this.shallExecuteTask(workerNodeKey
))
782 this.executeTask(workerNodeKey
, task
)
784 this.enqueueTask(workerNodeKey
, task
)
790 public start (): void {
793 this.workerNodes
.reduce(
794 (accumulator
, workerNode
) =>
795 !workerNode
.info
.dynamic
? accumulator
+ 1 : accumulator
,
797 ) < this.numberOfWorkers
799 this.createAndSetupWorkerNode()
801 this.starting
= false
806 public async destroy (): Promise
<void> {
808 this.workerNodes
.map(async (_
, workerNodeKey
) => {
809 await this.destroyWorkerNode(workerNodeKey
)
812 this.emitter
?.emit(PoolEvents
.destroy
, this.info
)
816 protected async sendKillMessageToWorker (
817 workerNodeKey
: number,
820 await new Promise
<void>((resolve
, reject
) => {
821 this.registerWorkerMessageListener(workerNodeKey
, message
=> {
822 if (message
.kill
=== 'success') {
824 } else if (message
.kill
=== 'failure') {
825 reject(new Error(`Worker ${workerId} kill message handling failed`))
828 this.sendToWorker(workerNodeKey
, { kill
: true, workerId
})
833 * Terminates the worker node given its worker node key.
835 * @param workerNodeKey - The worker node key.
837 protected abstract destroyWorkerNode (workerNodeKey
: number): Promise
<void>
840 * Setup hook to execute code before worker nodes are created in the abstract constructor.
845 protected setupHook (): void {
846 /* Intentionally empty */
850 * Should return whether the worker is the main worker or not.
852 protected abstract isMain (): boolean
855 * Hook executed before the worker task execution.
858 * @param workerNodeKey - The worker node key.
859 * @param task - The task to execute.
861 protected beforeTaskExecutionHook (
862 workerNodeKey
: number,
865 if (this.workerNodes
[workerNodeKey
]?.usage
!= null) {
866 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
867 ++workerUsage
.tasks
.executing
868 this.updateWaitTimeWorkerUsage(workerUsage
, task
)
871 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
872 this.workerNodes
[workerNodeKey
].getTaskFunctionWorkerUsage(
876 const taskFunctionWorkerUsage
= this.workerNodes
[
878 ].getTaskFunctionWorkerUsage(task
.name
as string) as WorkerUsage
879 ++taskFunctionWorkerUsage
.tasks
.executing
880 this.updateWaitTimeWorkerUsage(taskFunctionWorkerUsage
, task
)
885 * Hook executed after the worker task execution.
888 * @param workerNodeKey - The worker node key.
889 * @param message - The received message.
891 protected afterTaskExecutionHook (
892 workerNodeKey
: number,
893 message
: MessageValue
<Response
>
895 if (this.workerNodes
[workerNodeKey
]?.usage
!= null) {
896 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
897 this.updateTaskStatisticsWorkerUsage(workerUsage
, message
)
898 this.updateRunTimeWorkerUsage(workerUsage
, message
)
899 this.updateEluWorkerUsage(workerUsage
, message
)
902 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
903 this.workerNodes
[workerNodeKey
].getTaskFunctionWorkerUsage(
904 message
.taskPerformance
?.name
as string
907 const taskFunctionWorkerUsage
= this.workerNodes
[
909 ].getTaskFunctionWorkerUsage(
910 message
.taskPerformance
?.name
as string
912 this.updateTaskStatisticsWorkerUsage(taskFunctionWorkerUsage
, message
)
913 this.updateRunTimeWorkerUsage(taskFunctionWorkerUsage
, message
)
914 this.updateEluWorkerUsage(taskFunctionWorkerUsage
, message
)
919 * Whether the worker node shall update its task function worker usage or not.
921 * @param workerNodeKey - The worker node key.
922 * @returns `true` if the worker node shall update its task function worker usage, `false` otherwise.
924 private shallUpdateTaskFunctionWorkerUsage (workerNodeKey
: number): boolean {
925 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
927 workerInfo
!= null &&
928 Array.isArray(workerInfo
.taskFunctions
) &&
929 workerInfo
.taskFunctions
.length
> 2
933 private updateTaskStatisticsWorkerUsage (
934 workerUsage
: WorkerUsage
,
935 message
: MessageValue
<Response
>
937 const workerTaskStatistics
= workerUsage
.tasks
939 workerTaskStatistics
.executing
!= null &&
940 workerTaskStatistics
.executing
> 0
942 --workerTaskStatistics
.executing
944 if (message
.taskError
== null) {
945 ++workerTaskStatistics
.executed
947 ++workerTaskStatistics
.failed
951 private updateRunTimeWorkerUsage (
952 workerUsage
: WorkerUsage
,
953 message
: MessageValue
<Response
>
955 if (message
.taskError
!= null) {
958 updateMeasurementStatistics(
960 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
,
961 message
.taskPerformance
?.runTime
?? 0
965 private updateWaitTimeWorkerUsage (
966 workerUsage
: WorkerUsage
,
969 const timestamp
= performance
.now()
970 const taskWaitTime
= timestamp
- (task
.timestamp
?? timestamp
)
971 updateMeasurementStatistics(
972 workerUsage
.waitTime
,
973 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().waitTime
,
978 private updateEluWorkerUsage (
979 workerUsage
: WorkerUsage
,
980 message
: MessageValue
<Response
>
982 if (message
.taskError
!= null) {
985 const eluTaskStatisticsRequirements
: MeasurementStatisticsRequirements
=
986 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
987 updateMeasurementStatistics(
988 workerUsage
.elu
.active
,
989 eluTaskStatisticsRequirements
,
990 message
.taskPerformance
?.elu
?.active
?? 0
992 updateMeasurementStatistics(
993 workerUsage
.elu
.idle
,
994 eluTaskStatisticsRequirements
,
995 message
.taskPerformance
?.elu
?.idle
?? 0
997 if (eluTaskStatisticsRequirements
.aggregate
) {
998 if (message
.taskPerformance
?.elu
!= null) {
999 if (workerUsage
.elu
.utilization
!= null) {
1000 workerUsage
.elu
.utilization
=
1001 (workerUsage
.elu
.utilization
+
1002 message
.taskPerformance
.elu
.utilization
) /
1005 workerUsage
.elu
.utilization
= message
.taskPerformance
.elu
.utilization
1012 * Chooses a worker node for the next task.
1014 * The default worker choice strategy uses a round robin algorithm to distribute the tasks.
1016 * @returns The chosen worker node key
1018 private chooseWorkerNode (): number {
1019 if (this.shallCreateDynamicWorker()) {
1020 const workerNodeKey
= this.createAndSetupDynamicWorkerNode()
1022 this.workerChoiceStrategyContext
.getStrategyPolicy().dynamicWorkerUsage
1024 return workerNodeKey
1027 return this.workerChoiceStrategyContext
.execute()
1031 * Conditions for dynamic worker creation.
1033 * @returns Whether to create a dynamic worker or not.
1035 private shallCreateDynamicWorker (): boolean {
1036 return this.type === PoolTypes
.dynamic
&& !this.full
&& this.internalBusy()
1040 * Sends a message to worker given its worker node key.
1042 * @param workerNodeKey - The worker node key.
1043 * @param message - The message.
1044 * @param transferList - The optional array of transferable objects.
1046 protected abstract sendToWorker (
1047 workerNodeKey
: number,
1048 message
: MessageValue
<Data
>,
1049 transferList
?: TransferListItem
[]
1053 * Creates a new worker.
1055 * @returns Newly created worker.
1057 protected abstract createWorker (): Worker
1060 * Creates a new, completely set up worker node.
1062 * @returns New, completely set up worker node key.
1064 protected createAndSetupWorkerNode (): number {
1065 const worker
= this.createWorker()
1067 worker
.on('online', this.opts
.onlineHandler
?? EMPTY_FUNCTION
)
1068 worker
.on('message', this.opts
.messageHandler
?? EMPTY_FUNCTION
)
1069 worker
.on('error', this.opts
.errorHandler
?? EMPTY_FUNCTION
)
1070 worker
.on('error', error
=> {
1071 const workerNodeKey
= this.getWorkerNodeKeyByWorker(worker
)
1072 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1073 workerInfo
.ready
= false
1074 this.workerNodes
[workerNodeKey
].closeChannel()
1075 this.emitter
?.emit(PoolEvents
.error
, error
)
1077 this.opts
.restartWorkerOnError
=== true &&
1081 if (workerInfo
.dynamic
) {
1082 this.createAndSetupDynamicWorkerNode()
1084 this.createAndSetupWorkerNode()
1087 if (this.opts
.enableTasksQueue
=== true) {
1088 this.redistributeQueuedTasks(workerNodeKey
)
1091 worker
.on('exit', this.opts
.exitHandler
?? EMPTY_FUNCTION
)
1092 worker
.once('exit', () => {
1093 this.removeWorkerNode(worker
)
1096 const workerNodeKey
= this.addWorkerNode(worker
)
1098 this.afterWorkerNodeSetup(workerNodeKey
)
1100 return workerNodeKey
1104 * Creates a new, completely set up dynamic worker node.
1106 * @returns New, completely set up dynamic worker node key.
1108 protected createAndSetupDynamicWorkerNode (): number {
1109 const workerNodeKey
= this.createAndSetupWorkerNode()
1110 this.registerWorkerMessageListener(workerNodeKey
, message
=> {
1111 const localWorkerNodeKey
= this.getWorkerNodeKeyByWorkerId(
1114 const workerUsage
= this.workerNodes
[localWorkerNodeKey
].usage
1115 // Kill message received from worker
1117 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
1118 (isKillBehavior(KillBehaviors
.SOFT
, message
.kill
) &&
1119 ((this.opts
.enableTasksQueue
=== false &&
1120 workerUsage
.tasks
.executing
=== 0) ||
1121 (this.opts
.enableTasksQueue
=== true &&
1122 workerUsage
.tasks
.executing
=== 0 &&
1123 this.tasksQueueSize(localWorkerNodeKey
) === 0)))
1125 this.destroyWorkerNode(localWorkerNodeKey
).catch(error
=> {
1126 this.emitter
?.emit(PoolEvents
.error
, error
)
1130 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1131 this.sendToWorker(workerNodeKey
, {
1133 workerId
: workerInfo
.id
as number
1135 workerInfo
.dynamic
= true
1137 this.workerChoiceStrategyContext
.getStrategyPolicy().dynamicWorkerReady
||
1138 this.workerChoiceStrategyContext
.getStrategyPolicy().dynamicWorkerUsage
1140 workerInfo
.ready
= true
1142 this.checkAndEmitDynamicWorkerCreationEvents()
1143 return workerNodeKey
1147 * Registers a listener callback on the worker given its worker node key.
1149 * @param workerNodeKey - The worker node key.
1150 * @param listener - The message listener callback.
1152 protected abstract registerWorkerMessageListener
<
1153 Message
extends Data
| Response
1155 workerNodeKey
: number,
1156 listener
: (message
: MessageValue
<Message
>) => void
1160 * Method hooked up after a worker node has been newly created.
1161 * Can be overridden.
1163 * @param workerNodeKey - The newly created worker node key.
1165 protected afterWorkerNodeSetup (workerNodeKey
: number): void {
1166 // Listen to worker messages.
1167 this.registerWorkerMessageListener(workerNodeKey
, this.workerListener())
1168 // Send the startup message to worker.
1169 this.sendStartupMessageToWorker(workerNodeKey
)
1170 // Send the statistics message to worker.
1171 this.sendStatisticsMessageToWorker(workerNodeKey
)
1172 if (this.opts
.enableTasksQueue
=== true) {
1173 if (this.opts
.tasksQueueOptions
?.tasksStealing
=== true) {
1174 this.workerNodes
[workerNodeKey
].onEmptyQueue
=
1175 this.taskStealingOnEmptyQueue
.bind(this)
1177 if (this.opts
.tasksQueueOptions
?.tasksStealingOnBackPressure
=== true) {
1178 this.workerNodes
[workerNodeKey
].onBackPressure
=
1179 this.tasksStealingOnBackPressure
.bind(this)
1185 * Sends the startup message to worker given its worker node key.
1187 * @param workerNodeKey - The worker node key.
1189 protected abstract sendStartupMessageToWorker (workerNodeKey
: number): void
1192 * Sends the statistics message to worker given its worker node key.
1194 * @param workerNodeKey - The worker node key.
1196 private sendStatisticsMessageToWorker (workerNodeKey
: number): void {
1197 this.sendToWorker(workerNodeKey
, {
1200 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1202 elu
: this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1205 workerId
: this.getWorkerInfo(workerNodeKey
).id
as number
1209 private redistributeQueuedTasks (workerNodeKey
: number): void {
1210 while (this.tasksQueueSize(workerNodeKey
) > 0) {
1211 const destinationWorkerNodeKey
= this.workerNodes
.reduce(
1212 (minWorkerNodeKey
, workerNode
, workerNodeKey
, workerNodes
) => {
1213 return workerNode
.info
.ready
&&
1214 workerNode
.usage
.tasks
.queued
<
1215 workerNodes
[minWorkerNodeKey
].usage
.tasks
.queued
1221 const destinationWorkerNode
= this.workerNodes
[destinationWorkerNodeKey
]
1223 ...(this.dequeueTask(workerNodeKey
) as Task
<Data
>),
1224 workerId
: destinationWorkerNode
.info
.id
as number
1226 if (this.shallExecuteTask(destinationWorkerNodeKey
)) {
1227 this.executeTask(destinationWorkerNodeKey
, task
)
1229 this.enqueueTask(destinationWorkerNodeKey
, task
)
1234 private updateTaskStolenStatisticsWorkerUsage (
1235 workerNodeKey
: number,
1238 const workerNode
= this.workerNodes
[workerNodeKey
]
1239 if (workerNode
?.usage
!= null) {
1240 ++workerNode
.usage
.tasks
.stolen
1243 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1244 workerNode
.getTaskFunctionWorkerUsage(taskName
) != null
1246 const taskFunctionWorkerUsage
= workerNode
.getTaskFunctionWorkerUsage(
1249 ++taskFunctionWorkerUsage
.tasks
.stolen
1253 private taskStealingOnEmptyQueue (workerId
: number): void {
1254 const destinationWorkerNodeKey
= this.getWorkerNodeKeyByWorkerId(workerId
)
1255 const destinationWorkerNode
= this.workerNodes
[destinationWorkerNodeKey
]
1256 const workerNodes
= this.workerNodes
1259 (workerNodeA
, workerNodeB
) =>
1260 workerNodeB
.usage
.tasks
.queued
- workerNodeA
.usage
.tasks
.queued
1262 const sourceWorkerNode
= workerNodes
.find(
1264 workerNode
.info
.ready
&&
1265 workerNode
.info
.id
!== workerId
&&
1266 workerNode
.usage
.tasks
.queued
> 0
1268 if (sourceWorkerNode
!= null) {
1270 ...(sourceWorkerNode
.popTask() as Task
<Data
>),
1271 workerId
: destinationWorkerNode
.info
.id
as number
1273 if (this.shallExecuteTask(destinationWorkerNodeKey
)) {
1274 this.executeTask(destinationWorkerNodeKey
, task
)
1276 this.enqueueTask(destinationWorkerNodeKey
, task
)
1278 this.updateTaskStolenStatisticsWorkerUsage(
1279 destinationWorkerNodeKey
,
1285 private tasksStealingOnBackPressure (workerId
: number): void {
1286 const sizeOffset
= 1
1287 if ((this.opts
.tasksQueueOptions
?.size
as number) <= sizeOffset
) {
1290 const sourceWorkerNode
=
1291 this.workerNodes
[this.getWorkerNodeKeyByWorkerId(workerId
)]
1292 const workerNodes
= this.workerNodes
1295 (workerNodeA
, workerNodeB
) =>
1296 workerNodeA
.usage
.tasks
.queued
- workerNodeB
.usage
.tasks
.queued
1298 for (const [workerNodeKey
, workerNode
] of workerNodes
.entries()) {
1300 sourceWorkerNode
.usage
.tasks
.queued
> 0 &&
1301 workerNode
.info
.ready
&&
1302 workerNode
.info
.id
!== workerId
&&
1303 workerNode
.usage
.tasks
.queued
<
1304 (this.opts
.tasksQueueOptions
?.size
as number) - sizeOffset
1307 ...(sourceWorkerNode
.popTask() as Task
<Data
>),
1308 workerId
: workerNode
.info
.id
as number
1310 if (this.shallExecuteTask(workerNodeKey
)) {
1311 this.executeTask(workerNodeKey
, task
)
1313 this.enqueueTask(workerNodeKey
, task
)
1315 this.updateTaskStolenStatisticsWorkerUsage(
1324 * This method is the listener registered for each worker message.
1326 * @returns The listener function to execute when a message is received from a worker.
1328 protected workerListener (): (message
: MessageValue
<Response
>) => void {
1330 this.checkMessageWorkerId(message
)
1331 if (message
.ready
!= null && message
.taskFunctions
!= null) {
1332 // Worker ready response received from worker
1333 this.handleWorkerReadyResponse(message
)
1334 } else if (message
.taskId
!= null) {
1335 // Task execution response received from worker
1336 this.handleTaskExecutionResponse(message
)
1337 } else if (message
.taskFunctions
!= null) {
1338 // Task functions message received from worker
1340 this.getWorkerNodeKeyByWorkerId(message
.workerId
)
1341 ).taskFunctions
= message
.taskFunctions
1346 private handleWorkerReadyResponse (message
: MessageValue
<Response
>): void {
1347 if (message
.ready
=== false) {
1348 throw new Error(`Worker ${message.workerId} failed to initialize`)
1350 const workerInfo
= this.getWorkerInfo(
1351 this.getWorkerNodeKeyByWorkerId(message
.workerId
)
1353 workerInfo
.ready
= message
.ready
as boolean
1354 workerInfo
.taskFunctions
= message
.taskFunctions
1355 if (this.emitter
!= null && this.ready
) {
1356 this.emitter
.emit(PoolEvents
.ready
, this.info
)
1360 private handleTaskExecutionResponse (message
: MessageValue
<Response
>): void {
1361 const { taskId
, taskError
, data
} = message
1362 const promiseResponse
= this.promiseResponseMap
.get(taskId
as string)
1363 if (promiseResponse
!= null) {
1364 if (taskError
!= null) {
1365 this.emitter
?.emit(PoolEvents
.taskError
, taskError
)
1366 promiseResponse
.reject(taskError
.message
)
1368 promiseResponse
.resolve(data
as Response
)
1370 const workerNodeKey
= promiseResponse
.workerNodeKey
1371 this.afterTaskExecutionHook(workerNodeKey
, message
)
1372 this.workerChoiceStrategyContext
.update(workerNodeKey
)
1373 this.promiseResponseMap
.delete(taskId
as string)
1375 this.opts
.enableTasksQueue
=== true &&
1376 this.tasksQueueSize(workerNodeKey
) > 0 &&
1377 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
<
1378 (this.opts
.tasksQueueOptions
?.concurrency
as number)
1382 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1388 private checkAndEmitTaskExecutionEvents (): void {
1390 this.emitter
?.emit(PoolEvents
.busy
, this.info
)
1394 private checkAndEmitTaskQueuingEvents (): void {
1395 if (this.hasBackPressure()) {
1396 this.emitter
?.emit(PoolEvents
.backPressure
, this.info
)
1400 private checkAndEmitDynamicWorkerCreationEvents (): void {
1401 if (this.type === PoolTypes
.dynamic
) {
1403 this.emitter
?.emit(PoolEvents
.full
, this.info
)
1409 * Gets the worker information given its worker node key.
1411 * @param workerNodeKey - The worker node key.
1412 * @returns The worker information.
1414 protected getWorkerInfo (workerNodeKey
: number): WorkerInfo
{
1415 return this.workerNodes
[workerNodeKey
].info
1419 * Adds the given worker in the pool worker nodes.
1421 * @param worker - The worker.
1422 * @returns The added worker node key.
1423 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the added worker node is not found.
1425 private addWorkerNode (worker
: Worker
): number {
1426 const workerNode
= new WorkerNode
<Worker
, Data
>(
1428 this.opts
.tasksQueueOptions
?.size
?? Math.pow(this.maxSize
, 2)
1430 // Flag the worker node as ready at pool startup.
1431 if (this.starting
) {
1432 workerNode
.info
.ready
= true
1434 this.workerNodes
.push(workerNode
)
1435 const workerNodeKey
= this.getWorkerNodeKeyByWorker(worker
)
1436 if (workerNodeKey
=== -1) {
1437 throw new Error('Worker added not found in worker nodes')
1439 return workerNodeKey
1443 * Removes the given worker from the pool worker nodes.
1445 * @param worker - The worker.
1447 private removeWorkerNode (worker
: Worker
): void {
1448 const workerNodeKey
= this.getWorkerNodeKeyByWorker(worker
)
1449 if (workerNodeKey
!== -1) {
1450 this.workerNodes
.splice(workerNodeKey
, 1)
1451 this.workerChoiceStrategyContext
.remove(workerNodeKey
)
1456 public hasWorkerNodeBackPressure (workerNodeKey
: number): boolean {
1458 this.opts
.enableTasksQueue
=== true &&
1459 this.workerNodes
[workerNodeKey
].hasBackPressure()
1463 private hasBackPressure (): boolean {
1465 this.opts
.enableTasksQueue
=== true &&
1466 this.workerNodes
.findIndex(
1467 workerNode
=> !workerNode
.hasBackPressure()
1473 * Executes the given task on the worker given its worker node key.
1475 * @param workerNodeKey - The worker node key.
1476 * @param task - The task to execute.
1478 private executeTask (workerNodeKey
: number, task
: Task
<Data
>): void {
1479 this.beforeTaskExecutionHook(workerNodeKey
, task
)
1480 this.sendToWorker(workerNodeKey
, task
, task
.transferList
)
1481 this.checkAndEmitTaskExecutionEvents()
1484 private enqueueTask (workerNodeKey
: number, task
: Task
<Data
>): number {
1485 const tasksQueueSize
= this.workerNodes
[workerNodeKey
].enqueueTask(task
)
1486 this.checkAndEmitTaskQueuingEvents()
1487 return tasksQueueSize
1490 private dequeueTask (workerNodeKey
: number): Task
<Data
> | undefined {
1491 return this.workerNodes
[workerNodeKey
].dequeueTask()
1494 private tasksQueueSize (workerNodeKey
: number): number {
1495 return this.workerNodes
[workerNodeKey
].tasksQueueSize()
1498 protected flushTasksQueue (workerNodeKey
: number): void {
1499 while (this.tasksQueueSize(workerNodeKey
) > 0) {
1502 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1505 this.workerNodes
[workerNodeKey
].clearTasksQueue()
1508 private flushTasksQueues (): void {
1509 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
1510 this.flushTasksQueue(workerNodeKey
)