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
,
18 updateMeasurementStatistics
20 import { KillBehaviors
} from
'../worker/worker-options'
29 type TasksQueueOptions
39 type MeasurementStatisticsRequirements
,
41 WorkerChoiceStrategies
,
42 type WorkerChoiceStrategy
,
43 type WorkerChoiceStrategyOptions
44 } from
'./selection-strategies/selection-strategies-types'
45 import { WorkerChoiceStrategyContext
} from
'./selection-strategies/worker-choice-strategy-context'
46 import { version
} from
'./version'
47 import { WorkerNode
} from
'./worker-node'
50 * Base class that implements some shared logic for all poolifier pools.
52 * @typeParam Worker - Type of worker which manages this pool.
53 * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
54 * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
56 export abstract class AbstractPool
<
57 Worker
extends IWorker
,
60 > implements IPool
<Worker
, Data
, Response
> {
62 public readonly workerNodes
: Array<IWorkerNode
<Worker
, Data
>> = []
65 public readonly emitter
?: PoolEmitter
68 * The task execution response promise map.
70 * - `key`: The message id of each submitted task.
71 * - `value`: An object that contains the worker, the execution response promise resolve and reject callbacks.
73 * When we receive a message from the worker, we get a map entry with the promise resolve/reject bound to the message id.
75 protected promiseResponseMap
: Map
<string, PromiseResponseWrapper
<Response
>> =
76 new Map
<string, PromiseResponseWrapper
<Response
>>()
79 * Worker choice strategy context referencing a worker choice algorithm implementation.
81 protected workerChoiceStrategyContext
: WorkerChoiceStrategyContext
<
88 * Dynamic pool maximum size property placeholder.
90 protected readonly max
?: number
93 * Whether the pool is starting or not.
95 private readonly starting
: boolean
97 * Whether the pool is started or not.
99 private started
: boolean
101 * The start timestamp of the pool.
103 private readonly startTimestamp
106 * Constructs a new poolifier pool.
108 * @param numberOfWorkers - Number of workers that this pool should manage.
109 * @param filePath - Path to the worker file.
110 * @param opts - Options for the pool.
113 protected readonly numberOfWorkers
: number,
114 protected readonly filePath
: string,
115 protected readonly opts
: PoolOptions
<Worker
>
117 if (!this.isMain()) {
119 'Cannot start a pool from a worker with the same type as the pool'
122 this.checkNumberOfWorkers(this.numberOfWorkers
)
123 this.checkFilePath(this.filePath
)
124 this.checkPoolOptions(this.opts
)
126 this.chooseWorkerNode
= this.chooseWorkerNode
.bind(this)
127 this.executeTask
= this.executeTask
.bind(this)
128 this.enqueueTask
= this.enqueueTask
.bind(this)
130 if (this.opts
.enableEvents
=== true) {
131 this.emitter
= new PoolEmitter()
133 this.workerChoiceStrategyContext
= new WorkerChoiceStrategyContext
<
139 this.opts
.workerChoiceStrategy
,
140 this.opts
.workerChoiceStrategyOptions
147 this.starting
= false
150 this.startTimestamp
= performance
.now()
153 private checkFilePath (filePath
: string): void {
156 typeof filePath
!== 'string' ||
157 (typeof filePath
=== 'string' && filePath
.trim().length
=== 0)
159 throw new Error('Please specify a file with a worker implementation')
161 if (!existsSync(filePath
)) {
162 throw new Error(`Cannot find the worker file '${filePath}'`)
166 private checkNumberOfWorkers (numberOfWorkers
: number): void {
167 if (numberOfWorkers
== null) {
169 'Cannot instantiate a pool without specifying the number of workers'
171 } else if (!Number.isSafeInteger(numberOfWorkers
)) {
173 'Cannot instantiate a pool with a non safe integer number of workers'
175 } else if (numberOfWorkers
< 0) {
176 throw new RangeError(
177 'Cannot instantiate a pool with a negative number of workers'
179 } else if (this.type === PoolTypes
.fixed
&& numberOfWorkers
=== 0) {
180 throw new RangeError('Cannot instantiate a fixed pool with zero worker')
184 protected checkDynamicPoolSize (min
: number, max
: number): void {
185 if (this.type === PoolTypes
.dynamic
) {
188 'Cannot instantiate a dynamic pool without specifying the maximum pool size'
190 } else if (!Number.isSafeInteger(max
)) {
192 'Cannot instantiate a dynamic pool with a non safe integer maximum pool size'
194 } else if (min
> max
) {
195 throw new RangeError(
196 'Cannot instantiate a dynamic pool with a maximum pool size inferior to the minimum pool size'
198 } else if (max
=== 0) {
199 throw new RangeError(
200 'Cannot instantiate a dynamic pool with a maximum pool size equal to zero'
202 } else if (min
=== max
) {
203 throw new RangeError(
204 'Cannot instantiate a dynamic pool with a minimum pool size equal to the maximum pool size. Use a fixed pool instead'
210 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
211 if (isPlainObject(opts
)) {
212 this.opts
.workerChoiceStrategy
=
213 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
214 this.checkValidWorkerChoiceStrategy(this.opts
.workerChoiceStrategy
)
215 this.opts
.workerChoiceStrategyOptions
= {
216 ...DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
,
217 ...opts
.workerChoiceStrategyOptions
219 this.checkValidWorkerChoiceStrategyOptions(
220 this.opts
.workerChoiceStrategyOptions
222 this.opts
.restartWorkerOnError
= opts
.restartWorkerOnError
?? true
223 this.opts
.enableEvents
= opts
.enableEvents
?? true
224 this.opts
.enableTasksQueue
= opts
.enableTasksQueue
?? false
225 if (this.opts
.enableTasksQueue
) {
226 this.checkValidTasksQueueOptions(
227 opts
.tasksQueueOptions
as TasksQueueOptions
229 this.opts
.tasksQueueOptions
= this.buildTasksQueueOptions(
230 opts
.tasksQueueOptions
as TasksQueueOptions
234 throw new TypeError('Invalid pool options: must be a plain object')
238 private checkValidWorkerChoiceStrategy (
239 workerChoiceStrategy
: WorkerChoiceStrategy
241 if (!Object.values(WorkerChoiceStrategies
).includes(workerChoiceStrategy
)) {
243 `Invalid worker choice strategy '${workerChoiceStrategy}'`
248 private checkValidWorkerChoiceStrategyOptions (
249 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
251 if (!isPlainObject(workerChoiceStrategyOptions
)) {
253 'Invalid worker choice strategy options: must be a plain object'
257 workerChoiceStrategyOptions
.choiceRetries
!= null &&
258 !Number.isSafeInteger(workerChoiceStrategyOptions
.choiceRetries
)
261 'Invalid worker choice strategy options: choice retries must be an integer'
265 workerChoiceStrategyOptions
.choiceRetries
!= null &&
266 workerChoiceStrategyOptions
.choiceRetries
<= 0
268 throw new RangeError(
269 `Invalid worker choice strategy options: choice retries '${workerChoiceStrategyOptions.choiceRetries}' must be greater than zero`
273 workerChoiceStrategyOptions
.weights
!= null &&
274 Object.keys(workerChoiceStrategyOptions
.weights
).length
!== this.maxSize
277 'Invalid worker choice strategy options: must have a weight for each worker node'
281 workerChoiceStrategyOptions
.measurement
!= null &&
282 !Object.values(Measurements
).includes(
283 workerChoiceStrategyOptions
.measurement
287 `Invalid worker choice strategy options: invalid measurement '${workerChoiceStrategyOptions.measurement}'`
292 private checkValidTasksQueueOptions (
293 tasksQueueOptions
: TasksQueueOptions
295 if (tasksQueueOptions
!= null && !isPlainObject(tasksQueueOptions
)) {
296 throw new TypeError('Invalid tasks queue options: must be a plain object')
299 tasksQueueOptions
?.concurrency
!= null &&
300 !Number.isSafeInteger(tasksQueueOptions
.concurrency
)
303 'Invalid worker node tasks concurrency: must be an integer'
307 tasksQueueOptions
?.concurrency
!= null &&
308 tasksQueueOptions
.concurrency
<= 0
310 throw new RangeError(
311 `Invalid worker node tasks concurrency: ${tasksQueueOptions.concurrency} is a negative integer or zero`
315 tasksQueueOptions
?.queueMaxSize
!= null &&
316 !Number.isSafeInteger(tasksQueueOptions
.queueMaxSize
)
319 'Invalid worker node tasks queue max size: must be an integer'
323 tasksQueueOptions
?.queueMaxSize
!= null &&
324 tasksQueueOptions
.queueMaxSize
<= 0
326 throw new RangeError(
327 `Invalid worker node tasks queue max size: ${tasksQueueOptions.queueMaxSize} is a negative integer or zero`
332 private startPool (): void {
334 this.workerNodes
.reduce(
335 (accumulator
, workerNode
) =>
336 !workerNode
.info
.dynamic
? accumulator
+ 1 : accumulator
,
338 ) < this.numberOfWorkers
340 this.createAndSetupWorkerNode()
345 public get
info (): PoolInfo
{
351 strategy
: this.opts
.workerChoiceStrategy
as WorkerChoiceStrategy
,
352 minSize
: this.minSize
,
353 maxSize
: this.maxSize
,
354 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
355 .runTime
.aggregate
&&
356 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
357 .waitTime
.aggregate
&& { utilization
: round(this.utilization
) }),
358 workerNodes
: this.workerNodes
.length
,
359 idleWorkerNodes
: this.workerNodes
.reduce(
360 (accumulator
, workerNode
) =>
361 workerNode
.usage
.tasks
.executing
=== 0
366 busyWorkerNodes
: this.workerNodes
.reduce(
367 (accumulator
, workerNode
) =>
368 workerNode
.usage
.tasks
.executing
> 0 ? accumulator
+ 1 : accumulator
,
371 executedTasks
: this.workerNodes
.reduce(
372 (accumulator
, workerNode
) =>
373 accumulator
+ workerNode
.usage
.tasks
.executed
,
376 executingTasks
: this.workerNodes
.reduce(
377 (accumulator
, workerNode
) =>
378 accumulator
+ workerNode
.usage
.tasks
.executing
,
381 ...(this.opts
.enableTasksQueue
=== true && {
382 queuedTasks
: this.workerNodes
.reduce(
383 (accumulator
, workerNode
) =>
384 accumulator
+ workerNode
.usage
.tasks
.queued
,
388 ...(this.opts
.enableTasksQueue
=== true && {
389 maxQueuedTasks
: this.workerNodes
.reduce(
390 (accumulator
, workerNode
) =>
391 accumulator
+ (workerNode
.usage
.tasks
?.maxQueued
?? 0),
395 ...(this.opts
.enableTasksQueue
=== true && {
396 backPressure
: this.hasBackPressure()
398 failedTasks
: this.workerNodes
.reduce(
399 (accumulator
, workerNode
) =>
400 accumulator
+ workerNode
.usage
.tasks
.failed
,
403 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
404 .runTime
.aggregate
&& {
408 ...this.workerNodes
.map(
409 (workerNode
) => workerNode
.usage
.runTime
?.minimum
?? Infinity
415 ...this.workerNodes
.map(
416 (workerNode
) => workerNode
.usage
.runTime
?.maximum
?? -Infinity
421 this.workerNodes
.reduce(
422 (accumulator
, workerNode
) =>
423 accumulator
+ (workerNode
.usage
.runTime
?.aggregate
?? 0),
426 this.workerNodes
.reduce(
427 (accumulator
, workerNode
) =>
428 accumulator
+ (workerNode
.usage
.tasks
?.executed
?? 0),
432 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
436 this.workerNodes
.map(
437 (workerNode
) => workerNode
.usage
.runTime
?.median
?? 0
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
462 this.workerNodes
.reduce(
463 (accumulator
, workerNode
) =>
464 accumulator
+ (workerNode
.usage
.waitTime
?.aggregate
?? 0),
467 this.workerNodes
.reduce(
468 (accumulator
, workerNode
) =>
469 accumulator
+ (workerNode
.usage
.tasks
?.executed
?? 0),
473 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
474 .waitTime
.median
&& {
477 this.workerNodes
.map(
478 (workerNode
) => workerNode
.usage
.waitTime
?.median
?? 0
489 * The pool readiness boolean status.
491 private get
ready (): boolean {
493 this.workerNodes
.reduce(
494 (accumulator
, workerNode
) =>
495 !workerNode
.info
.dynamic
&& workerNode
.info
.ready
504 * The approximate pool utilization.
506 * @returns The pool utilization.
508 private get
utilization (): number {
509 const poolTimeCapacity
=
510 (performance
.now() - this.startTimestamp
) * this.maxSize
511 const totalTasksRunTime
= this.workerNodes
.reduce(
512 (accumulator
, workerNode
) =>
513 accumulator
+ (workerNode
.usage
.runTime
?.aggregate
?? 0),
516 const totalTasksWaitTime
= this.workerNodes
.reduce(
517 (accumulator
, workerNode
) =>
518 accumulator
+ (workerNode
.usage
.waitTime
?.aggregate
?? 0),
521 return (totalTasksRunTime
+ totalTasksWaitTime
) / poolTimeCapacity
527 * If it is `'dynamic'`, it provides the `max` property.
529 protected abstract get
type (): PoolType
534 protected abstract get
worker (): WorkerType
537 * The pool minimum size.
539 protected get
minSize (): number {
540 return this.numberOfWorkers
544 * The pool maximum size.
546 protected get
maxSize (): number {
547 return this.max
?? this.numberOfWorkers
551 * Checks if the worker id sent in the received message from a worker is valid.
553 * @param message - The received message.
554 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the worker id is invalid.
556 private checkMessageWorkerId (message
: MessageValue
<Response
>): void {
557 if (message
.workerId
== null) {
558 throw new Error('Worker message received without worker id')
560 message
.workerId
!= null &&
561 this.getWorkerNodeKeyByWorkerId(message
.workerId
) === -1
564 `Worker message received from unknown worker '${message.workerId}'`
570 * Gets the given worker its worker node key.
572 * @param worker - The worker.
573 * @returns The worker node key if found in the pool worker nodes, `-1` otherwise.
575 private getWorkerNodeKeyByWorker (worker
: Worker
): number {
576 return this.workerNodes
.findIndex(
577 (workerNode
) => workerNode
.worker
=== worker
582 * Gets the worker node key given its worker id.
584 * @param workerId - The worker id.
585 * @returns The worker node key if the worker id is found in the pool worker nodes, `-1` otherwise.
587 private getWorkerNodeKeyByWorkerId (workerId
: number): number {
588 return this.workerNodes
.findIndex(
589 (workerNode
) => workerNode
.info
.id
=== workerId
594 public setWorkerChoiceStrategy (
595 workerChoiceStrategy
: WorkerChoiceStrategy
,
596 workerChoiceStrategyOptions
?: WorkerChoiceStrategyOptions
598 this.checkValidWorkerChoiceStrategy(workerChoiceStrategy
)
599 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
600 this.workerChoiceStrategyContext
.setWorkerChoiceStrategy(
601 this.opts
.workerChoiceStrategy
603 if (workerChoiceStrategyOptions
!= null) {
604 this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
606 for (const [workerNodeKey
, workerNode
] of this.workerNodes
.entries()) {
607 workerNode
.resetUsage()
608 this.sendStatisticsMessageToWorker(workerNodeKey
)
613 public setWorkerChoiceStrategyOptions (
614 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
616 this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
617 this.opts
.workerChoiceStrategyOptions
= {
618 ...DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
,
619 ...workerChoiceStrategyOptions
621 this.workerChoiceStrategyContext
.setOptions(
622 this.opts
.workerChoiceStrategyOptions
627 public enableTasksQueue (
629 tasksQueueOptions
?: TasksQueueOptions
631 if (this.opts
.enableTasksQueue
=== true && !enable
) {
632 this.flushTasksQueues()
634 this.opts
.enableTasksQueue
= enable
635 this.setTasksQueueOptions(tasksQueueOptions
as TasksQueueOptions
)
639 public setTasksQueueOptions (tasksQueueOptions
: TasksQueueOptions
): void {
640 if (this.opts
.enableTasksQueue
=== true) {
641 this.checkValidTasksQueueOptions(tasksQueueOptions
)
642 this.opts
.tasksQueueOptions
=
643 this.buildTasksQueueOptions(tasksQueueOptions
)
644 this.setTasksQueueMaxSize(
645 this.opts
.tasksQueueOptions
.queueMaxSize
as number
647 } else if (this.opts
.tasksQueueOptions
!= null) {
648 delete this.opts
.tasksQueueOptions
652 private setTasksQueueMaxSize (queueMaxSize
: number): void {
653 for (const workerNode
of this.workerNodes
) {
654 workerNode
.tasksQueueBackPressureSize
= queueMaxSize
658 private buildTasksQueueOptions (
659 tasksQueueOptions
: TasksQueueOptions
660 ): TasksQueueOptions
{
663 queueMaxSize
: Math.pow(this.maxSize
, 2),
671 * Whether the pool is full or not.
673 * The pool filling boolean status.
675 protected get
full (): boolean {
676 return this.workerNodes
.length
>= this.maxSize
680 * Whether the pool is busy or not.
682 * The pool busyness boolean status.
684 protected abstract get
busy (): boolean
687 * Whether worker nodes are executing concurrently their tasks quota or not.
689 * @returns Worker nodes busyness boolean status.
691 protected internalBusy (): boolean {
692 if (this.opts
.enableTasksQueue
=== true) {
694 this.workerNodes
.findIndex(
696 workerNode
.info
.ready
&&
697 workerNode
.usage
.tasks
.executing
<
698 (this.opts
.tasksQueueOptions
?.concurrency
as number)
703 this.workerNodes
.findIndex(
705 workerNode
.info
.ready
&& workerNode
.usage
.tasks
.executing
=== 0
712 public listTaskFunctions (): string[] {
713 for (const workerNode
of this.workerNodes
) {
715 Array.isArray(workerNode
.info
.taskFunctions
) &&
716 workerNode
.info
.taskFunctions
.length
> 0
718 return workerNode
.info
.taskFunctions
725 public async execute (
728 transferList
?: TransferListItem
[]
729 ): Promise
<Response
> {
730 return await new Promise
<Response
>((resolve
, reject
) => {
732 reject(new Error('Cannot execute a task on destroyed pool'))
734 if (name
!= null && typeof name
!== 'string') {
735 reject(new TypeError('name argument must be a string'))
739 typeof name
=== 'string' &&
740 name
.trim().length
=== 0
742 reject(new TypeError('name argument must not be an empty string'))
744 if (transferList
!= null && !Array.isArray(transferList
)) {
745 reject(new TypeError('transferList argument must be an array'))
747 const timestamp
= performance
.now()
748 const workerNodeKey
= this.chooseWorkerNode()
749 const workerInfo
= this.getWorkerInfo(workerNodeKey
) as WorkerInfo
752 Array.isArray(workerInfo
.taskFunctions
) &&
753 !workerInfo
.taskFunctions
.includes(name
)
756 new Error(`Task function '${name}' is not registered in the pool`)
759 const task
: Task
<Data
> = {
760 name
: name
?? DEFAULT_TASK_NAME
,
761 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
762 data
: data
?? ({} as Data
),
765 workerId
: workerInfo
.id
as number,
768 this.promiseResponseMap
.set(task
.taskId
as string, {
774 this.opts
.enableTasksQueue
=== false ||
775 (this.opts
.enableTasksQueue
=== true &&
776 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
<
777 (this.opts
.tasksQueueOptions
?.concurrency
as number))
779 this.executeTask(workerNodeKey
, task
)
781 this.enqueueTask(workerNodeKey
, task
)
787 public async destroy (): Promise
<void> {
789 this.workerNodes
.map(async (_
, workerNodeKey
) => {
790 await this.destroyWorkerNode(workerNodeKey
)
793 this.emitter
?.emit(PoolEvents
.destroy
, this.info
)
797 protected async sendKillMessageToWorker (
798 workerNodeKey
: number,
801 await new Promise
<void>((resolve
, reject
) => {
802 this.registerWorkerMessageListener(workerNodeKey
, (message
) => {
803 if (message
.kill
=== 'success') {
805 } else if (message
.kill
=== 'failure') {
806 reject(new Error(`Worker ${workerId} kill message handling failed`))
809 this.sendToWorker(workerNodeKey
, { kill
: true, workerId
})
814 * Terminates the worker node given its worker node key.
816 * @param workerNodeKey - The worker node key.
818 protected abstract destroyWorkerNode (workerNodeKey
: number): Promise
<void>
821 * Setup hook to execute code before worker nodes are created in the abstract constructor.
826 protected setupHook (): void {
827 // Intentionally empty
831 * Should return whether the worker is the main worker or not.
833 protected abstract isMain (): boolean
836 * Hook executed before the worker task execution.
839 * @param workerNodeKey - The worker node key.
840 * @param task - The task to execute.
842 protected beforeTaskExecutionHook (
843 workerNodeKey
: number,
846 if (this.workerNodes
[workerNodeKey
]?.usage
!= null) {
847 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
848 ++workerUsage
.tasks
.executing
849 this.updateWaitTimeWorkerUsage(workerUsage
, task
)
852 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
853 this.workerNodes
[workerNodeKey
].getTaskFunctionWorkerUsage(
857 const taskFunctionWorkerUsage
= this.workerNodes
[
859 ].getTaskFunctionWorkerUsage(task
.name
as string) as WorkerUsage
860 ++taskFunctionWorkerUsage
.tasks
.executing
861 this.updateWaitTimeWorkerUsage(taskFunctionWorkerUsage
, task
)
866 * Hook executed after the worker task execution.
869 * @param workerNodeKey - The worker node key.
870 * @param message - The received message.
872 protected afterTaskExecutionHook (
873 workerNodeKey
: number,
874 message
: MessageValue
<Response
>
876 if (this.workerNodes
[workerNodeKey
]?.usage
!= null) {
877 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
878 this.updateTaskStatisticsWorkerUsage(workerUsage
, message
)
879 this.updateRunTimeWorkerUsage(workerUsage
, message
)
880 this.updateEluWorkerUsage(workerUsage
, message
)
883 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
884 this.workerNodes
[workerNodeKey
].getTaskFunctionWorkerUsage(
885 message
.taskPerformance
?.name
as string
888 const taskFunctionWorkerUsage
= this.workerNodes
[
890 ].getTaskFunctionWorkerUsage(
891 message
.taskPerformance
?.name
as string
893 this.updateTaskStatisticsWorkerUsage(taskFunctionWorkerUsage
, message
)
894 this.updateRunTimeWorkerUsage(taskFunctionWorkerUsage
, message
)
895 this.updateEluWorkerUsage(taskFunctionWorkerUsage
, message
)
900 * Whether the worker node shall update its task function worker usage or not.
902 * @param workerNodeKey - The worker node key.
903 * @returns `true` if the worker node shall update its task function worker usage, `false` otherwise.
905 private shallUpdateTaskFunctionWorkerUsage (workerNodeKey
: number): boolean {
906 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
908 workerInfo
!= null &&
909 Array.isArray(workerInfo
.taskFunctions
) &&
910 workerInfo
.taskFunctions
.length
> 2
914 private updateTaskStatisticsWorkerUsage (
915 workerUsage
: WorkerUsage
,
916 message
: MessageValue
<Response
>
918 const workerTaskStatistics
= workerUsage
.tasks
920 workerTaskStatistics
.executing
!= null &&
921 workerTaskStatistics
.executing
> 0
923 --workerTaskStatistics
.executing
925 if (message
.taskError
== null) {
926 ++workerTaskStatistics
.executed
928 ++workerTaskStatistics
.failed
932 private updateRunTimeWorkerUsage (
933 workerUsage
: WorkerUsage
,
934 message
: MessageValue
<Response
>
936 updateMeasurementStatistics(
938 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
,
939 message
.taskPerformance
?.runTime
?? 0,
940 workerUsage
.tasks
.executed
944 private updateWaitTimeWorkerUsage (
945 workerUsage
: WorkerUsage
,
948 const timestamp
= performance
.now()
949 const taskWaitTime
= timestamp
- (task
.timestamp
?? timestamp
)
950 updateMeasurementStatistics(
951 workerUsage
.waitTime
,
952 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().waitTime
,
954 workerUsage
.tasks
.executed
958 private updateEluWorkerUsage (
959 workerUsage
: WorkerUsage
,
960 message
: MessageValue
<Response
>
962 const eluTaskStatisticsRequirements
: MeasurementStatisticsRequirements
=
963 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
964 updateMeasurementStatistics(
965 workerUsage
.elu
.active
,
966 eluTaskStatisticsRequirements
,
967 message
.taskPerformance
?.elu
?.active
?? 0,
968 workerUsage
.tasks
.executed
970 updateMeasurementStatistics(
971 workerUsage
.elu
.idle
,
972 eluTaskStatisticsRequirements
,
973 message
.taskPerformance
?.elu
?.idle
?? 0,
974 workerUsage
.tasks
.executed
976 if (eluTaskStatisticsRequirements
.aggregate
) {
977 if (message
.taskPerformance
?.elu
!= null) {
978 if (workerUsage
.elu
.utilization
!= null) {
979 workerUsage
.elu
.utilization
=
980 (workerUsage
.elu
.utilization
+
981 message
.taskPerformance
.elu
.utilization
) /
984 workerUsage
.elu
.utilization
= message
.taskPerformance
.elu
.utilization
991 * Chooses a worker node for the next task.
993 * The default worker choice strategy uses a round robin algorithm to distribute the tasks.
995 * @returns The chosen worker node key
997 private chooseWorkerNode (): number {
998 if (this.shallCreateDynamicWorker()) {
999 const workerNodeKey
= this.createAndSetupDynamicWorkerNode()
1001 this.workerChoiceStrategyContext
.getStrategyPolicy().dynamicWorkerUsage
1003 return workerNodeKey
1006 return this.workerChoiceStrategyContext
.execute()
1010 * Conditions for dynamic worker creation.
1012 * @returns Whether to create a dynamic worker or not.
1014 private shallCreateDynamicWorker (): boolean {
1015 return this.type === PoolTypes
.dynamic
&& !this.full
&& this.internalBusy()
1019 * Sends a message to worker given its worker node key.
1021 * @param workerNodeKey - The worker node key.
1022 * @param message - The message.
1023 * @param transferList - The optional array of transferable objects.
1025 protected abstract sendToWorker (
1026 workerNodeKey
: number,
1027 message
: MessageValue
<Data
>,
1028 transferList
?: TransferListItem
[]
1032 * Creates a new worker.
1034 * @returns Newly created worker.
1036 protected abstract createWorker (): Worker
1039 * Creates a new, completely set up worker node.
1041 * @returns New, completely set up worker node key.
1043 protected createAndSetupWorkerNode (): number {
1044 const worker
= this.createWorker()
1046 worker
.on('online', this.opts
.onlineHandler
?? EMPTY_FUNCTION
)
1047 worker
.on('message', this.opts
.messageHandler
?? EMPTY_FUNCTION
)
1048 worker
.on('error', this.opts
.errorHandler
?? EMPTY_FUNCTION
)
1049 worker
.on('error', (error
) => {
1050 const workerNodeKey
= this.getWorkerNodeKeyByWorker(worker
)
1051 const workerInfo
= this.getWorkerInfo(workerNodeKey
) as WorkerInfo
1052 workerInfo
.ready
= false
1053 this.workerNodes
[workerNodeKey
].closeChannel()
1054 this.emitter
?.emit(PoolEvents
.error
, error
)
1056 this.opts
.restartWorkerOnError
=== true &&
1060 if (workerInfo
.dynamic
) {
1061 this.createAndSetupDynamicWorkerNode()
1063 this.createAndSetupWorkerNode()
1066 if (this.opts
.enableTasksQueue
=== true) {
1067 this.redistributeQueuedTasks(workerNodeKey
)
1070 worker
.on('exit', this.opts
.exitHandler
?? EMPTY_FUNCTION
)
1071 worker
.once('exit', () => {
1072 this.removeWorkerNode(worker
)
1075 const workerNodeKey
= this.addWorkerNode(worker
)
1077 this.afterWorkerNodeSetup(workerNodeKey
)
1079 return workerNodeKey
1083 * Creates a new, completely set up dynamic worker node.
1085 * @returns New, completely set up dynamic worker node key.
1087 protected createAndSetupDynamicWorkerNode (): number {
1088 const workerNodeKey
= this.createAndSetupWorkerNode()
1089 this.registerWorkerMessageListener(workerNodeKey
, (message
) => {
1090 const localWorkerNodeKey
= this.getWorkerNodeKeyByWorkerId(
1093 const workerUsage
= this.workerNodes
[localWorkerNodeKey
].usage
1094 // Kill message received from worker
1096 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
1097 (isKillBehavior(KillBehaviors
.SOFT
, message
.kill
) &&
1098 ((this.opts
.enableTasksQueue
=== false &&
1099 workerUsage
.tasks
.executing
=== 0) ||
1100 (this.opts
.enableTasksQueue
=== true &&
1101 workerUsage
.tasks
.executing
=== 0 &&
1102 this.tasksQueueSize(localWorkerNodeKey
) === 0)))
1104 this.destroyWorkerNode(localWorkerNodeKey
).catch((error
) => {
1105 this.emitter
?.emit(PoolEvents
.error
, error
)
1109 const workerInfo
= this.getWorkerInfo(workerNodeKey
) as WorkerInfo
1110 this.sendToWorker(workerNodeKey
, {
1112 workerId
: workerInfo
.id
as number
1114 workerInfo
.dynamic
= true
1116 this.workerChoiceStrategyContext
.getStrategyPolicy().dynamicWorkerReady
||
1117 this.workerChoiceStrategyContext
.getStrategyPolicy().dynamicWorkerUsage
1119 workerInfo
.ready
= true
1121 this.checkAndEmitDynamicWorkerCreationEvents()
1122 return workerNodeKey
1126 * Registers a listener callback on the worker given its worker node key.
1128 * @param workerNodeKey - The worker node key.
1129 * @param listener - The message listener callback.
1131 protected abstract registerWorkerMessageListener
<
1132 Message
extends Data
| Response
1134 workerNodeKey
: number,
1135 listener
: (message
: MessageValue
<Message
>) => void
1139 * Method hooked up after a worker node has been newly created.
1140 * Can be overridden.
1142 * @param workerNodeKey - The newly created worker node key.
1144 protected afterWorkerNodeSetup (workerNodeKey
: number): void {
1145 // Listen to worker messages.
1146 this.registerWorkerMessageListener(workerNodeKey
, this.workerListener())
1147 // Send the startup message to worker.
1148 this.sendStartupMessageToWorker(workerNodeKey
)
1149 // Send the statistics message to worker.
1150 this.sendStatisticsMessageToWorker(workerNodeKey
)
1151 if (this.opts
.enableTasksQueue
=== true) {
1152 this.workerNodes
[workerNodeKey
].onBackPressure
=
1153 this.tasksStealingOnBackPressure
.bind(this)
1158 * Sends the startup message to worker given its worker node key.
1160 * @param workerNodeKey - The worker node key.
1162 protected abstract sendStartupMessageToWorker (workerNodeKey
: number): void
1165 * Sends the statistics message to worker given its worker node key.
1167 * @param workerNodeKey - The worker node key.
1169 private sendStatisticsMessageToWorker (workerNodeKey
: number): void {
1170 this.sendToWorker(workerNodeKey
, {
1173 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1175 elu
: this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1178 workerId
: (this.getWorkerInfo(workerNodeKey
) as WorkerInfo
).id
as number
1182 private redistributeQueuedTasks (workerNodeKey
: number): void {
1183 while (this.tasksQueueSize(workerNodeKey
) > 0) {
1184 let targetWorkerNodeKey
: number = workerNodeKey
1185 let minQueuedTasks
= Infinity
1186 let executeTask
= false
1187 for (const [workerNodeId
, workerNode
] of this.workerNodes
.entries()) {
1189 this.workerNodes
[workerNodeId
].usage
.tasks
.executing
<
1190 (this.opts
.tasksQueueOptions
?.concurrency
as number)
1195 workerNodeId
!== workerNodeKey
&&
1196 workerNode
.info
.ready
&&
1197 workerNode
.usage
.tasks
.queued
=== 0
1199 targetWorkerNodeKey
= workerNodeId
1203 workerNodeId
!== workerNodeKey
&&
1204 workerNode
.info
.ready
&&
1205 workerNode
.usage
.tasks
.queued
< minQueuedTasks
1207 minQueuedTasks
= workerNode
.usage
.tasks
.queued
1208 targetWorkerNodeKey
= workerNodeId
1213 targetWorkerNodeKey
,
1214 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1218 targetWorkerNodeKey
,
1219 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1225 private tasksStealingOnBackPressure (workerId
: number): void {
1226 const sourceWorkerNode
=
1227 this.workerNodes
[this.getWorkerNodeKeyByWorkerId(workerId
)]
1228 const workerNodes
= this.workerNodes
1229 .filter((workerNode
) => workerNode
.info
.id
!== workerId
)
1231 (workerNodeA
, workerNodeB
) =>
1232 workerNodeA
.usage
.tasks
.queued
- workerNodeB
.usage
.tasks
.queued
1234 for (const [workerNodeKey
, workerNode
] of workerNodes
.entries()) {
1236 workerNode
.info
.ready
&&
1237 sourceWorkerNode
.usage
.tasks
.queued
> 0 &&
1238 !workerNode
.hasBackPressure() &&
1239 workerNode
.usage
.tasks
.executing
<
1240 (this.opts
.tasksQueueOptions
?.concurrency
as number)
1244 sourceWorkerNode
.popTask() as Task
<Data
>
1247 workerNode
.info
.ready
&&
1248 sourceWorkerNode
.usage
.tasks
.queued
> 0 &&
1249 !workerNode
.hasBackPressure() &&
1250 workerNode
.usage
.tasks
.executing
>=
1251 (this.opts
.tasksQueueOptions
?.concurrency
as number)
1255 sourceWorkerNode
.popTask() as Task
<Data
>
1262 * This method is the listener registered for each worker message.
1264 * @returns The listener function to execute when a message is received from a worker.
1266 protected workerListener (): (message
: MessageValue
<Response
>) => void {
1267 return (message
) => {
1268 this.checkMessageWorkerId(message
)
1269 if (message
.ready
!= null && message
.taskFunctions
!= null) {
1270 // Worker ready response received from worker
1271 this.handleWorkerReadyResponse(message
)
1272 } else if (message
.taskId
!= null) {
1273 // Task execution response received from worker
1274 this.handleTaskExecutionResponse(message
)
1275 } else if (message
.taskFunctions
!= null) {
1276 // Task functions message received from worker
1279 this.getWorkerNodeKeyByWorkerId(message
.workerId
)
1281 ).taskFunctions
= message
.taskFunctions
1286 private handleWorkerReadyResponse (message
: MessageValue
<Response
>): void {
1287 if (message
.ready
=== false) {
1288 throw new Error(`Worker ${message.workerId} failed to initialize`)
1290 const workerInfo
= this.getWorkerInfo(
1291 this.getWorkerNodeKeyByWorkerId(message
.workerId
)
1293 workerInfo
.ready
= message
.ready
as boolean
1294 workerInfo
.taskFunctions
= message
.taskFunctions
1295 if (this.emitter
!= null && this.ready
) {
1296 this.emitter
.emit(PoolEvents
.ready
, this.info
)
1300 private handleTaskExecutionResponse (message
: MessageValue
<Response
>): void {
1301 const { taskId
, taskError
, data
} = message
1302 const promiseResponse
= this.promiseResponseMap
.get(taskId
as string)
1303 if (promiseResponse
!= null) {
1304 if (taskError
!= null) {
1305 this.emitter
?.emit(PoolEvents
.taskError
, taskError
)
1306 promiseResponse
.reject(taskError
.message
)
1308 promiseResponse
.resolve(data
as Response
)
1310 const workerNodeKey
= promiseResponse
.workerNodeKey
1311 this.afterTaskExecutionHook(workerNodeKey
, message
)
1312 this.promiseResponseMap
.delete(taskId
as string)
1314 this.opts
.enableTasksQueue
=== true &&
1315 this.tasksQueueSize(workerNodeKey
) > 0 &&
1316 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
<
1317 (this.opts
.tasksQueueOptions
?.concurrency
as number)
1321 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1324 this.workerChoiceStrategyContext
.update(workerNodeKey
)
1328 private checkAndEmitTaskExecutionEvents (): void {
1330 this.emitter
?.emit(PoolEvents
.busy
, this.info
)
1334 private checkAndEmitTaskQueuingEvents (): void {
1335 if (this.hasBackPressure()) {
1336 this.emitter
?.emit(PoolEvents
.backPressure
, this.info
)
1340 private checkAndEmitDynamicWorkerCreationEvents (): void {
1341 if (this.type === PoolTypes
.dynamic
) {
1343 this.emitter
?.emit(PoolEvents
.full
, this.info
)
1349 * Gets the worker information given its worker node key.
1351 * @param workerNodeKey - The worker node key.
1352 * @returns The worker information.
1354 protected getWorkerInfo (workerNodeKey
: number): WorkerInfo
| undefined {
1355 return this.workerNodes
[workerNodeKey
]?.info
1359 * Adds the given worker in the pool worker nodes.
1361 * @param worker - The worker.
1362 * @returns The added worker node key.
1363 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the added worker node is not found.
1365 private addWorkerNode (worker
: Worker
): number {
1366 const workerNode
= new WorkerNode
<Worker
, Data
>(
1369 this.opts
.tasksQueueOptions
?.queueMaxSize
?? Math.pow(this.maxSize
, 2)
1371 // Flag the worker node as ready at pool startup.
1372 if (this.starting
) {
1373 workerNode
.info
.ready
= true
1375 this.workerNodes
.push(workerNode
)
1376 const workerNodeKey
= this.getWorkerNodeKeyByWorker(worker
)
1377 if (workerNodeKey
=== -1) {
1378 throw new Error('Worker node added not found')
1380 return workerNodeKey
1384 * Removes the given worker from the pool worker nodes.
1386 * @param worker - The worker.
1388 private removeWorkerNode (worker
: Worker
): void {
1389 const workerNodeKey
= this.getWorkerNodeKeyByWorker(worker
)
1390 if (workerNodeKey
!== -1) {
1391 this.workerNodes
.splice(workerNodeKey
, 1)
1392 this.workerChoiceStrategyContext
.remove(workerNodeKey
)
1397 public hasWorkerNodeBackPressure (workerNodeKey
: number): boolean {
1399 this.opts
.enableTasksQueue
=== true &&
1400 this.workerNodes
[workerNodeKey
].hasBackPressure()
1404 private hasBackPressure (): boolean {
1406 this.opts
.enableTasksQueue
=== true &&
1407 this.workerNodes
.findIndex(
1408 (workerNode
) => !workerNode
.hasBackPressure()
1414 * Executes the given task on the worker given its worker node key.
1416 * @param workerNodeKey - The worker node key.
1417 * @param task - The task to execute.
1419 private executeTask (workerNodeKey
: number, task
: Task
<Data
>): void {
1420 this.beforeTaskExecutionHook(workerNodeKey
, task
)
1421 this.sendToWorker(workerNodeKey
, task
, task
.transferList
)
1422 this.checkAndEmitTaskExecutionEvents()
1425 private enqueueTask (workerNodeKey
: number, task
: Task
<Data
>): number {
1426 const tasksQueueSize
= this.workerNodes
[workerNodeKey
].enqueueTask(task
)
1427 this.checkAndEmitTaskQueuingEvents()
1428 return tasksQueueSize
1431 private dequeueTask (workerNodeKey
: number): Task
<Data
> | undefined {
1432 return this.workerNodes
[workerNodeKey
].dequeueTask()
1435 private tasksQueueSize (workerNodeKey
: number): number {
1436 return this.workerNodes
[workerNodeKey
].tasksQueueSize()
1439 protected flushTasksQueue (workerNodeKey
: number): void {
1440 while (this.tasksQueueSize(workerNodeKey
) > 0) {
1443 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1446 this.workerNodes
[workerNodeKey
].clearTasksQueue()
1449 private flushTasksQueues (): void {
1450 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
1451 this.flushTasksQueue(workerNodeKey
)