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
,
10 } from
'../utility-types'
13 DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
,
19 updateMeasurementStatistics
21 import { KillBehaviors
} from
'../worker/worker-options'
30 type TasksQueueOptions
40 type MeasurementStatisticsRequirements
,
42 WorkerChoiceStrategies
,
43 type WorkerChoiceStrategy
,
44 type WorkerChoiceStrategyOptions
45 } from
'./selection-strategies/selection-strategies-types'
46 import { WorkerChoiceStrategyContext
} from
'./selection-strategies/worker-choice-strategy-context'
47 import { version
} from
'./version'
48 import { WorkerNode
} from
'./worker-node'
51 * Base class that implements some shared logic for all poolifier pools.
53 * @typeParam Worker - Type of worker which manages this pool.
54 * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
55 * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
57 export abstract class AbstractPool
<
58 Worker
extends IWorker
,
61 > implements IPool
<Worker
, Data
, Response
> {
63 public readonly workerNodes
: Array<IWorkerNode
<Worker
, Data
>> = []
66 public readonly emitter
?: PoolEmitter
69 * The task execution response promise map.
71 * - `key`: The message id of each submitted task.
72 * - `value`: An object that contains the worker, the execution response promise resolve and reject callbacks.
74 * When we receive a message from the worker, we get a map entry with the promise resolve/reject bound to the message id.
76 protected promiseResponseMap
: Map
<string, PromiseResponseWrapper
<Response
>> =
77 new Map
<string, PromiseResponseWrapper
<Response
>>()
80 * Worker choice strategy context referencing a worker choice algorithm implementation.
82 protected workerChoiceStrategyContext
: WorkerChoiceStrategyContext
<
89 * Dynamic pool maximum size property placeholder.
91 protected readonly max
?: number
94 * Whether the pool is starting or not.
96 private readonly starting
: boolean
98 * Whether the pool is started or not.
100 private started
: boolean
102 * The start timestamp of the pool.
104 private readonly startTimestamp
107 * Constructs a new poolifier pool.
109 * @param numberOfWorkers - Number of workers that this pool should manage.
110 * @param filePath - Path to the worker file.
111 * @param opts - Options for the pool.
114 protected readonly numberOfWorkers
: number,
115 protected readonly filePath
: string,
116 protected readonly opts
: PoolOptions
<Worker
>
118 if (!this.isMain()) {
120 'Cannot start a pool from a worker with the same type as the pool'
123 this.checkNumberOfWorkers(this.numberOfWorkers
)
124 this.checkFilePath(this.filePath
)
125 this.checkPoolOptions(this.opts
)
127 this.chooseWorkerNode
= this.chooseWorkerNode
.bind(this)
128 this.executeTask
= this.executeTask
.bind(this)
129 this.enqueueTask
= this.enqueueTask
.bind(this)
131 if (this.opts
.enableEvents
=== true) {
132 this.emitter
= new PoolEmitter()
134 this.workerChoiceStrategyContext
= new WorkerChoiceStrategyContext
<
140 this.opts
.workerChoiceStrategy
,
141 this.opts
.workerChoiceStrategyOptions
148 this.starting
= false
151 this.startTimestamp
= performance
.now()
154 private checkFilePath (filePath
: string): void {
157 typeof filePath
!== 'string' ||
158 (typeof filePath
=== 'string' && filePath
.trim().length
=== 0)
160 throw new Error('Please specify a file with a worker implementation')
162 if (!existsSync(filePath
)) {
163 throw new Error(`Cannot find the worker file '${filePath}'`)
167 private checkNumberOfWorkers (numberOfWorkers
: number): void {
168 if (numberOfWorkers
== null) {
170 'Cannot instantiate a pool without specifying the number of workers'
172 } else if (!Number.isSafeInteger(numberOfWorkers
)) {
174 'Cannot instantiate a pool with a non safe integer number of workers'
176 } else if (numberOfWorkers
< 0) {
177 throw new RangeError(
178 'Cannot instantiate a pool with a negative number of workers'
180 } else if (this.type === PoolTypes
.fixed
&& numberOfWorkers
=== 0) {
181 throw new RangeError('Cannot instantiate a fixed pool with zero worker')
185 protected checkDynamicPoolSize (min
: number, max
: number): void {
186 if (this.type === PoolTypes
.dynamic
) {
189 'Cannot instantiate a dynamic pool without specifying the maximum pool size'
191 } else if (!Number.isSafeInteger(max
)) {
193 'Cannot instantiate a dynamic pool with a non safe integer maximum pool size'
195 } else if (min
> max
) {
196 throw new RangeError(
197 'Cannot instantiate a dynamic pool with a maximum pool size inferior to the minimum pool size'
199 } else if (max
=== 0) {
200 throw new RangeError(
201 'Cannot instantiate a dynamic pool with a maximum pool size equal to zero'
203 } else if (min
=== max
) {
204 throw new RangeError(
205 'Cannot instantiate a dynamic pool with a minimum pool size equal to the maximum pool size. Use a fixed pool instead'
211 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
212 if (isPlainObject(opts
)) {
213 this.opts
.workerChoiceStrategy
=
214 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
215 this.checkValidWorkerChoiceStrategy(this.opts
.workerChoiceStrategy
)
216 this.opts
.workerChoiceStrategyOptions
= {
217 ...DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
,
218 ...opts
.workerChoiceStrategyOptions
220 this.checkValidWorkerChoiceStrategyOptions(
221 this.opts
.workerChoiceStrategyOptions
223 this.opts
.restartWorkerOnError
= opts
.restartWorkerOnError
?? true
224 this.opts
.enableEvents
= opts
.enableEvents
?? true
225 this.opts
.enableTasksQueue
= opts
.enableTasksQueue
?? false
226 if (this.opts
.enableTasksQueue
) {
227 this.checkValidTasksQueueOptions(
228 opts
.tasksQueueOptions
as TasksQueueOptions
230 this.opts
.tasksQueueOptions
= this.buildTasksQueueOptions(
231 opts
.tasksQueueOptions
as TasksQueueOptions
235 throw new TypeError('Invalid pool options: must be a plain object')
239 private checkValidWorkerChoiceStrategy (
240 workerChoiceStrategy
: WorkerChoiceStrategy
242 if (!Object.values(WorkerChoiceStrategies
).includes(workerChoiceStrategy
)) {
244 `Invalid worker choice strategy '${workerChoiceStrategy}'`
249 private checkValidWorkerChoiceStrategyOptions (
250 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
252 if (!isPlainObject(workerChoiceStrategyOptions
)) {
254 'Invalid worker choice strategy options: must be a plain object'
258 workerChoiceStrategyOptions
.choiceRetries
!= null &&
259 !Number.isSafeInteger(workerChoiceStrategyOptions
.choiceRetries
)
262 'Invalid worker choice strategy options: choice retries must be an integer'
266 workerChoiceStrategyOptions
.choiceRetries
!= null &&
267 workerChoiceStrategyOptions
.choiceRetries
<= 0
269 throw new RangeError(
270 `Invalid worker choice strategy options: choice retries '${workerChoiceStrategyOptions.choiceRetries}' must be greater than zero`
274 workerChoiceStrategyOptions
.weights
!= null &&
275 Object.keys(workerChoiceStrategyOptions
.weights
).length
!== this.maxSize
278 'Invalid worker choice strategy options: must have a weight for each worker node'
282 workerChoiceStrategyOptions
.measurement
!= null &&
283 !Object.values(Measurements
).includes(
284 workerChoiceStrategyOptions
.measurement
288 `Invalid worker choice strategy options: invalid measurement '${workerChoiceStrategyOptions.measurement}'`
293 private checkValidTasksQueueOptions (
294 tasksQueueOptions
: Writable
<TasksQueueOptions
>
296 if (tasksQueueOptions
!= null && !isPlainObject(tasksQueueOptions
)) {
297 throw new TypeError('Invalid tasks queue options: must be a plain object')
300 tasksQueueOptions
?.concurrency
!= null &&
301 !Number.isSafeInteger(tasksQueueOptions
.concurrency
)
304 'Invalid worker node tasks concurrency: must be an integer'
308 tasksQueueOptions
?.concurrency
!= null &&
309 tasksQueueOptions
.concurrency
<= 0
311 throw new RangeError(
312 `Invalid worker node tasks concurrency: ${tasksQueueOptions.concurrency} is a negative integer or zero`
316 tasksQueueOptions
?.queueMaxSize
!= null &&
317 tasksQueueOptions
?.size
!= null
320 'Invalid tasks queue options: cannot specify both queueMaxSize and size'
323 if (tasksQueueOptions
?.queueMaxSize
!= null) {
324 tasksQueueOptions
.size
= tasksQueueOptions
.queueMaxSize
327 tasksQueueOptions
?.size
!= null &&
328 !Number.isSafeInteger(tasksQueueOptions
.size
)
331 'Invalid worker node tasks queue max size: must be an integer'
334 if (tasksQueueOptions
?.size
!= null && tasksQueueOptions
.size
<= 0) {
335 throw new RangeError(
336 `Invalid worker node tasks queue max size: ${tasksQueueOptions.size} is a negative integer or zero`
341 private startPool (): void {
343 this.workerNodes
.reduce(
344 (accumulator
, workerNode
) =>
345 !workerNode
.info
.dynamic
? accumulator
+ 1 : accumulator
,
347 ) < this.numberOfWorkers
349 this.createAndSetupWorkerNode()
354 public get
info (): PoolInfo
{
360 strategy
: this.opts
.workerChoiceStrategy
as WorkerChoiceStrategy
,
361 minSize
: this.minSize
,
362 maxSize
: this.maxSize
,
363 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
364 .runTime
.aggregate
&&
365 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
366 .waitTime
.aggregate
&& { utilization
: round(this.utilization
) }),
367 workerNodes
: this.workerNodes
.length
,
368 idleWorkerNodes
: this.workerNodes
.reduce(
369 (accumulator
, workerNode
) =>
370 workerNode
.usage
.tasks
.executing
=== 0
375 busyWorkerNodes
: this.workerNodes
.reduce(
376 (accumulator
, workerNode
) =>
377 workerNode
.usage
.tasks
.executing
> 0 ? accumulator
+ 1 : accumulator
,
380 executedTasks
: this.workerNodes
.reduce(
381 (accumulator
, workerNode
) =>
382 accumulator
+ workerNode
.usage
.tasks
.executed
,
385 executingTasks
: this.workerNodes
.reduce(
386 (accumulator
, workerNode
) =>
387 accumulator
+ workerNode
.usage
.tasks
.executing
,
390 ...(this.opts
.enableTasksQueue
=== true && {
391 queuedTasks
: this.workerNodes
.reduce(
392 (accumulator
, workerNode
) =>
393 accumulator
+ workerNode
.usage
.tasks
.queued
,
397 ...(this.opts
.enableTasksQueue
=== true && {
398 maxQueuedTasks
: this.workerNodes
.reduce(
399 (accumulator
, workerNode
) =>
400 accumulator
+ (workerNode
.usage
.tasks
?.maxQueued
?? 0),
404 ...(this.opts
.enableTasksQueue
=== true && {
405 backPressure
: this.hasBackPressure()
407 failedTasks
: this.workerNodes
.reduce(
408 (accumulator
, workerNode
) =>
409 accumulator
+ workerNode
.usage
.tasks
.failed
,
412 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
413 .runTime
.aggregate
&& {
417 ...this.workerNodes
.map(
418 (workerNode
) => workerNode
.usage
.runTime
?.minimum
?? Infinity
424 ...this.workerNodes
.map(
425 (workerNode
) => workerNode
.usage
.runTime
?.maximum
?? -Infinity
430 this.workerNodes
.reduce(
431 (accumulator
, workerNode
) =>
432 accumulator
+ (workerNode
.usage
.runTime
?.aggregate
?? 0),
435 this.workerNodes
.reduce(
436 (accumulator
, workerNode
) =>
437 accumulator
+ (workerNode
.usage
.tasks
?.executed
?? 0),
441 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
445 this.workerNodes
.map(
446 (workerNode
) => workerNode
.usage
.runTime
?.median
?? 0
453 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
454 .waitTime
.aggregate
&& {
458 ...this.workerNodes
.map(
459 (workerNode
) => workerNode
.usage
.waitTime
?.minimum
?? Infinity
465 ...this.workerNodes
.map(
466 (workerNode
) => workerNode
.usage
.waitTime
?.maximum
?? -Infinity
471 this.workerNodes
.reduce(
472 (accumulator
, workerNode
) =>
473 accumulator
+ (workerNode
.usage
.waitTime
?.aggregate
?? 0),
476 this.workerNodes
.reduce(
477 (accumulator
, workerNode
) =>
478 accumulator
+ (workerNode
.usage
.tasks
?.executed
?? 0),
482 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
483 .waitTime
.median
&& {
486 this.workerNodes
.map(
487 (workerNode
) => workerNode
.usage
.waitTime
?.median
?? 0
498 * The pool readiness boolean status.
500 private get
ready (): boolean {
502 this.workerNodes
.reduce(
503 (accumulator
, workerNode
) =>
504 !workerNode
.info
.dynamic
&& workerNode
.info
.ready
513 * The approximate pool utilization.
515 * @returns The pool utilization.
517 private get
utilization (): number {
518 const poolTimeCapacity
=
519 (performance
.now() - this.startTimestamp
) * this.maxSize
520 const totalTasksRunTime
= this.workerNodes
.reduce(
521 (accumulator
, workerNode
) =>
522 accumulator
+ (workerNode
.usage
.runTime
?.aggregate
?? 0),
525 const totalTasksWaitTime
= this.workerNodes
.reduce(
526 (accumulator
, workerNode
) =>
527 accumulator
+ (workerNode
.usage
.waitTime
?.aggregate
?? 0),
530 return (totalTasksRunTime
+ totalTasksWaitTime
) / poolTimeCapacity
536 * If it is `'dynamic'`, it provides the `max` property.
538 protected abstract get
type (): PoolType
543 protected abstract get
worker (): WorkerType
546 * The pool minimum size.
548 protected get
minSize (): number {
549 return this.numberOfWorkers
553 * The pool maximum size.
555 protected get
maxSize (): number {
556 return this.max
?? this.numberOfWorkers
560 * Checks if the worker id sent in the received message from a worker is valid.
562 * @param message - The received message.
563 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the worker id is invalid.
565 private checkMessageWorkerId (message
: MessageValue
<Response
>): void {
566 if (message
.workerId
== null) {
567 throw new Error('Worker message received without worker id')
569 message
.workerId
!= null &&
570 this.getWorkerNodeKeyByWorkerId(message
.workerId
) === -1
573 `Worker message received from unknown worker '${message.workerId}'`
579 * Gets the given worker its worker node key.
581 * @param worker - The worker.
582 * @returns The worker node key if found in the pool worker nodes, `-1` otherwise.
584 private getWorkerNodeKeyByWorker (worker
: Worker
): number {
585 return this.workerNodes
.findIndex(
586 (workerNode
) => workerNode
.worker
=== worker
591 * Gets the worker node key given its worker id.
593 * @param workerId - The worker id.
594 * @returns The worker node key if the worker id is found in the pool worker nodes, `-1` otherwise.
596 private getWorkerNodeKeyByWorkerId (workerId
: number): number {
597 return this.workerNodes
.findIndex(
598 (workerNode
) => workerNode
.info
.id
=== workerId
603 public setWorkerChoiceStrategy (
604 workerChoiceStrategy
: WorkerChoiceStrategy
,
605 workerChoiceStrategyOptions
?: WorkerChoiceStrategyOptions
607 this.checkValidWorkerChoiceStrategy(workerChoiceStrategy
)
608 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
609 this.workerChoiceStrategyContext
.setWorkerChoiceStrategy(
610 this.opts
.workerChoiceStrategy
612 if (workerChoiceStrategyOptions
!= null) {
613 this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
615 for (const [workerNodeKey
, workerNode
] of this.workerNodes
.entries()) {
616 workerNode
.resetUsage()
617 this.sendStatisticsMessageToWorker(workerNodeKey
)
622 public setWorkerChoiceStrategyOptions (
623 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
625 this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
626 this.opts
.workerChoiceStrategyOptions
= {
627 ...DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
,
628 ...workerChoiceStrategyOptions
630 this.workerChoiceStrategyContext
.setOptions(
631 this.opts
.workerChoiceStrategyOptions
636 public enableTasksQueue (
638 tasksQueueOptions
?: TasksQueueOptions
640 if (this.opts
.enableTasksQueue
=== true && !enable
) {
641 this.flushTasksQueues()
643 this.opts
.enableTasksQueue
= enable
644 this.setTasksQueueOptions(tasksQueueOptions
as TasksQueueOptions
)
648 public setTasksQueueOptions (tasksQueueOptions
: TasksQueueOptions
): void {
649 if (this.opts
.enableTasksQueue
=== true) {
650 this.checkValidTasksQueueOptions(tasksQueueOptions
)
651 this.opts
.tasksQueueOptions
=
652 this.buildTasksQueueOptions(tasksQueueOptions
)
653 this.setTasksQueueMaxSize(this.opts
.tasksQueueOptions
.size
as number)
654 } else if (this.opts
.tasksQueueOptions
!= null) {
655 delete this.opts
.tasksQueueOptions
659 private setTasksQueueMaxSize (size
: number): void {
660 for (const workerNode
of this.workerNodes
) {
661 workerNode
.tasksQueueBackPressureSize
= size
665 private buildTasksQueueOptions (
666 tasksQueueOptions
: TasksQueueOptions
667 ): TasksQueueOptions
{
670 size
: Math.pow(this.maxSize
, 2),
678 * Whether the pool is full or not.
680 * The pool filling boolean status.
682 protected get
full (): boolean {
683 return this.workerNodes
.length
>= this.maxSize
687 * Whether the pool is busy or not.
689 * The pool busyness boolean status.
691 protected abstract get
busy (): boolean
694 * Whether worker nodes are executing concurrently their tasks quota or not.
696 * @returns Worker nodes busyness boolean status.
698 protected internalBusy (): boolean {
699 if (this.opts
.enableTasksQueue
=== true) {
701 this.workerNodes
.findIndex(
703 workerNode
.info
.ready
&&
704 workerNode
.usage
.tasks
.executing
<
705 (this.opts
.tasksQueueOptions
?.concurrency
as number)
710 this.workerNodes
.findIndex(
712 workerNode
.info
.ready
&& workerNode
.usage
.tasks
.executing
=== 0
719 public listTaskFunctions (): string[] {
720 for (const workerNode
of this.workerNodes
) {
722 Array.isArray(workerNode
.info
.taskFunctions
) &&
723 workerNode
.info
.taskFunctions
.length
> 0
725 return workerNode
.info
.taskFunctions
732 public async execute (
735 transferList
?: TransferListItem
[]
736 ): Promise
<Response
> {
737 return await new Promise
<Response
>((resolve
, reject
) => {
739 reject(new Error('Cannot execute a task on destroyed pool'))
741 if (name
!= null && typeof name
!== 'string') {
742 reject(new TypeError('name argument must be a string'))
746 typeof name
=== 'string' &&
747 name
.trim().length
=== 0
749 reject(new TypeError('name argument must not be an empty string'))
751 if (transferList
!= null && !Array.isArray(transferList
)) {
752 reject(new TypeError('transferList argument must be an array'))
754 const timestamp
= performance
.now()
755 const workerNodeKey
= this.chooseWorkerNode()
756 const workerInfo
= this.getWorkerInfo(workerNodeKey
) as WorkerInfo
759 Array.isArray(workerInfo
.taskFunctions
) &&
760 !workerInfo
.taskFunctions
.includes(name
)
763 new Error(`Task function '${name}' is not registered in the pool`)
766 const task
: Task
<Data
> = {
767 name
: name
?? DEFAULT_TASK_NAME
,
768 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
769 data
: data
?? ({} as Data
),
772 workerId
: workerInfo
.id
as number,
775 this.promiseResponseMap
.set(task
.taskId
as string, {
781 this.opts
.enableTasksQueue
=== false ||
782 (this.opts
.enableTasksQueue
=== true &&
783 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
<
784 (this.opts
.tasksQueueOptions
?.concurrency
as number))
786 this.executeTask(workerNodeKey
, task
)
788 this.enqueueTask(workerNodeKey
, task
)
794 public async destroy (): Promise
<void> {
796 this.workerNodes
.map(async (_
, workerNodeKey
) => {
797 await this.destroyWorkerNode(workerNodeKey
)
800 this.emitter
?.emit(PoolEvents
.destroy
, this.info
)
804 protected async sendKillMessageToWorker (
805 workerNodeKey
: number,
808 await new Promise
<void>((resolve
, reject
) => {
809 this.registerWorkerMessageListener(workerNodeKey
, (message
) => {
810 if (message
.kill
=== 'success') {
812 } else if (message
.kill
=== 'failure') {
813 reject(new Error(`Worker ${workerId} kill message handling failed`))
816 this.sendToWorker(workerNodeKey
, { kill
: true, workerId
})
821 * Terminates the worker node given its worker node key.
823 * @param workerNodeKey - The worker node key.
825 protected abstract destroyWorkerNode (workerNodeKey
: number): Promise
<void>
828 * Setup hook to execute code before worker nodes are created in the abstract constructor.
833 protected setupHook (): void {
834 // Intentionally empty
838 * Should return whether the worker is the main worker or not.
840 protected abstract isMain (): boolean
843 * Hook executed before the worker task execution.
846 * @param workerNodeKey - The worker node key.
847 * @param task - The task to execute.
849 protected beforeTaskExecutionHook (
850 workerNodeKey
: number,
853 if (this.workerNodes
[workerNodeKey
]?.usage
!= null) {
854 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
855 ++workerUsage
.tasks
.executing
856 this.updateWaitTimeWorkerUsage(workerUsage
, task
)
859 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
860 this.workerNodes
[workerNodeKey
].getTaskFunctionWorkerUsage(
864 const taskFunctionWorkerUsage
= this.workerNodes
[
866 ].getTaskFunctionWorkerUsage(task
.name
as string) as WorkerUsage
867 ++taskFunctionWorkerUsage
.tasks
.executing
868 this.updateWaitTimeWorkerUsage(taskFunctionWorkerUsage
, task
)
873 * Hook executed after the worker task execution.
876 * @param workerNodeKey - The worker node key.
877 * @param message - The received message.
879 protected afterTaskExecutionHook (
880 workerNodeKey
: number,
881 message
: MessageValue
<Response
>
883 if (this.workerNodes
[workerNodeKey
]?.usage
!= null) {
884 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
885 this.updateTaskStatisticsWorkerUsage(workerUsage
, message
)
886 this.updateRunTimeWorkerUsage(workerUsage
, message
)
887 this.updateEluWorkerUsage(workerUsage
, message
)
890 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
891 this.workerNodes
[workerNodeKey
].getTaskFunctionWorkerUsage(
892 message
.taskPerformance
?.name
as string
895 const taskFunctionWorkerUsage
= this.workerNodes
[
897 ].getTaskFunctionWorkerUsage(
898 message
.taskPerformance
?.name
as string
900 this.updateTaskStatisticsWorkerUsage(taskFunctionWorkerUsage
, message
)
901 this.updateRunTimeWorkerUsage(taskFunctionWorkerUsage
, message
)
902 this.updateEluWorkerUsage(taskFunctionWorkerUsage
, message
)
907 * Whether the worker node shall update its task function worker usage or not.
909 * @param workerNodeKey - The worker node key.
910 * @returns `true` if the worker node shall update its task function worker usage, `false` otherwise.
912 private shallUpdateTaskFunctionWorkerUsage (workerNodeKey
: number): boolean {
913 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
915 workerInfo
!= null &&
916 Array.isArray(workerInfo
.taskFunctions
) &&
917 workerInfo
.taskFunctions
.length
> 2
921 private updateTaskStatisticsWorkerUsage (
922 workerUsage
: WorkerUsage
,
923 message
: MessageValue
<Response
>
925 const workerTaskStatistics
= workerUsage
.tasks
927 workerTaskStatistics
.executing
!= null &&
928 workerTaskStatistics
.executing
> 0
930 --workerTaskStatistics
.executing
932 if (message
.taskError
== null) {
933 ++workerTaskStatistics
.executed
935 ++workerTaskStatistics
.failed
939 private updateRunTimeWorkerUsage (
940 workerUsage
: WorkerUsage
,
941 message
: MessageValue
<Response
>
943 updateMeasurementStatistics(
945 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
,
946 message
.taskPerformance
?.runTime
?? 0,
947 workerUsage
.tasks
.executed
951 private updateWaitTimeWorkerUsage (
952 workerUsage
: WorkerUsage
,
955 const timestamp
= performance
.now()
956 const taskWaitTime
= timestamp
- (task
.timestamp
?? timestamp
)
957 updateMeasurementStatistics(
958 workerUsage
.waitTime
,
959 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().waitTime
,
961 workerUsage
.tasks
.executed
965 private updateEluWorkerUsage (
966 workerUsage
: WorkerUsage
,
967 message
: MessageValue
<Response
>
969 const eluTaskStatisticsRequirements
: MeasurementStatisticsRequirements
=
970 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
971 updateMeasurementStatistics(
972 workerUsage
.elu
.active
,
973 eluTaskStatisticsRequirements
,
974 message
.taskPerformance
?.elu
?.active
?? 0,
975 workerUsage
.tasks
.executed
977 updateMeasurementStatistics(
978 workerUsage
.elu
.idle
,
979 eluTaskStatisticsRequirements
,
980 message
.taskPerformance
?.elu
?.idle
?? 0,
981 workerUsage
.tasks
.executed
983 if (eluTaskStatisticsRequirements
.aggregate
) {
984 if (message
.taskPerformance
?.elu
!= null) {
985 if (workerUsage
.elu
.utilization
!= null) {
986 workerUsage
.elu
.utilization
=
987 (workerUsage
.elu
.utilization
+
988 message
.taskPerformance
.elu
.utilization
) /
991 workerUsage
.elu
.utilization
= message
.taskPerformance
.elu
.utilization
998 * Chooses a worker node for the next task.
1000 * The default worker choice strategy uses a round robin algorithm to distribute the tasks.
1002 * @returns The chosen worker node key
1004 private chooseWorkerNode (): number {
1005 if (this.shallCreateDynamicWorker()) {
1006 const workerNodeKey
= this.createAndSetupDynamicWorkerNode()
1008 this.workerChoiceStrategyContext
.getStrategyPolicy().dynamicWorkerUsage
1010 return workerNodeKey
1013 return this.workerChoiceStrategyContext
.execute()
1017 * Conditions for dynamic worker creation.
1019 * @returns Whether to create a dynamic worker or not.
1021 private shallCreateDynamicWorker (): boolean {
1022 return this.type === PoolTypes
.dynamic
&& !this.full
&& this.internalBusy()
1026 * Sends a message to worker given its worker node key.
1028 * @param workerNodeKey - The worker node key.
1029 * @param message - The message.
1030 * @param transferList - The optional array of transferable objects.
1032 protected abstract sendToWorker (
1033 workerNodeKey
: number,
1034 message
: MessageValue
<Data
>,
1035 transferList
?: TransferListItem
[]
1039 * Creates a new worker.
1041 * @returns Newly created worker.
1043 protected abstract createWorker (): Worker
1046 * Creates a new, completely set up worker node.
1048 * @returns New, completely set up worker node key.
1050 protected createAndSetupWorkerNode (): number {
1051 const worker
= this.createWorker()
1053 worker
.on('online', this.opts
.onlineHandler
?? EMPTY_FUNCTION
)
1054 worker
.on('message', this.opts
.messageHandler
?? EMPTY_FUNCTION
)
1055 worker
.on('error', this.opts
.errorHandler
?? EMPTY_FUNCTION
)
1056 worker
.on('error', (error
) => {
1057 const workerNodeKey
= this.getWorkerNodeKeyByWorker(worker
)
1058 const workerInfo
= this.getWorkerInfo(workerNodeKey
) as WorkerInfo
1059 workerInfo
.ready
= false
1060 this.workerNodes
[workerNodeKey
].closeChannel()
1061 this.emitter
?.emit(PoolEvents
.error
, error
)
1063 this.opts
.restartWorkerOnError
=== true &&
1067 if (workerInfo
.dynamic
) {
1068 this.createAndSetupDynamicWorkerNode()
1070 this.createAndSetupWorkerNode()
1073 if (this.opts
.enableTasksQueue
=== true) {
1074 this.redistributeQueuedTasks(workerNodeKey
)
1077 worker
.on('exit', this.opts
.exitHandler
?? EMPTY_FUNCTION
)
1078 worker
.once('exit', () => {
1079 this.removeWorkerNode(worker
)
1082 const workerNodeKey
= this.addWorkerNode(worker
)
1084 this.afterWorkerNodeSetup(workerNodeKey
)
1086 return workerNodeKey
1090 * Creates a new, completely set up dynamic worker node.
1092 * @returns New, completely set up dynamic worker node key.
1094 protected createAndSetupDynamicWorkerNode (): number {
1095 const workerNodeKey
= this.createAndSetupWorkerNode()
1096 this.registerWorkerMessageListener(workerNodeKey
, (message
) => {
1097 const localWorkerNodeKey
= this.getWorkerNodeKeyByWorkerId(
1100 const workerUsage
= this.workerNodes
[localWorkerNodeKey
].usage
1101 // Kill message received from worker
1103 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
1104 (isKillBehavior(KillBehaviors
.SOFT
, message
.kill
) &&
1105 ((this.opts
.enableTasksQueue
=== false &&
1106 workerUsage
.tasks
.executing
=== 0) ||
1107 (this.opts
.enableTasksQueue
=== true &&
1108 workerUsage
.tasks
.executing
=== 0 &&
1109 this.tasksQueueSize(localWorkerNodeKey
) === 0)))
1111 this.destroyWorkerNode(localWorkerNodeKey
).catch((error
) => {
1112 this.emitter
?.emit(PoolEvents
.error
, error
)
1116 const workerInfo
= this.getWorkerInfo(workerNodeKey
) as WorkerInfo
1117 this.sendToWorker(workerNodeKey
, {
1119 workerId
: workerInfo
.id
as number
1121 workerInfo
.dynamic
= true
1123 this.workerChoiceStrategyContext
.getStrategyPolicy().dynamicWorkerReady
||
1124 this.workerChoiceStrategyContext
.getStrategyPolicy().dynamicWorkerUsage
1126 workerInfo
.ready
= true
1128 this.checkAndEmitDynamicWorkerCreationEvents()
1129 return workerNodeKey
1133 * Registers a listener callback on the worker given its worker node key.
1135 * @param workerNodeKey - The worker node key.
1136 * @param listener - The message listener callback.
1138 protected abstract registerWorkerMessageListener
<
1139 Message
extends Data
| Response
1141 workerNodeKey
: number,
1142 listener
: (message
: MessageValue
<Message
>) => void
1146 * Method hooked up after a worker node has been newly created.
1147 * Can be overridden.
1149 * @param workerNodeKey - The newly created worker node key.
1151 protected afterWorkerNodeSetup (workerNodeKey
: number): void {
1152 // Listen to worker messages.
1153 this.registerWorkerMessageListener(workerNodeKey
, this.workerListener())
1154 // Send the startup message to worker.
1155 this.sendStartupMessageToWorker(workerNodeKey
)
1156 // Send the statistics message to worker.
1157 this.sendStatisticsMessageToWorker(workerNodeKey
)
1158 if (this.opts
.enableTasksQueue
=== true) {
1159 this.workerNodes
[workerNodeKey
].onEmptyQueue
=
1160 this.taskStealingOnEmptyQueue
.bind(this)
1161 this.workerNodes
[workerNodeKey
].onBackPressure
=
1162 this.tasksStealingOnBackPressure
.bind(this)
1167 * Sends the startup message to worker given its worker node key.
1169 * @param workerNodeKey - The worker node key.
1171 protected abstract sendStartupMessageToWorker (workerNodeKey
: number): void
1174 * Sends the statistics message to worker given its worker node key.
1176 * @param workerNodeKey - The worker node key.
1178 private sendStatisticsMessageToWorker (workerNodeKey
: number): void {
1179 this.sendToWorker(workerNodeKey
, {
1182 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1184 elu
: this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1187 workerId
: (this.getWorkerInfo(workerNodeKey
) as WorkerInfo
).id
as number
1191 private redistributeQueuedTasks (workerNodeKey
: number): void {
1192 while (this.tasksQueueSize(workerNodeKey
) > 0) {
1193 let destinationWorkerNodeKey
: number = workerNodeKey
1194 let minQueuedTasks
= Infinity
1195 let executeTask
= false
1196 for (const [workerNodeId
, workerNode
] of this.workerNodes
.entries()) {
1197 if (workerNode
.info
.ready
&& workerNodeId
!== workerNodeKey
) {
1199 workerNode
.usage
.tasks
.executing
<
1200 (this.opts
.tasksQueueOptions
?.concurrency
as number)
1204 if (workerNode
.usage
.tasks
.queued
=== 0) {
1205 destinationWorkerNodeKey
= workerNodeId
1208 if (workerNode
.usage
.tasks
.queued
< minQueuedTasks
) {
1209 minQueuedTasks
= workerNode
.usage
.tasks
.queued
1210 destinationWorkerNodeKey
= workerNodeId
1215 ...(this.dequeueTask(workerNodeKey
) as Task
<Data
>),
1216 workerId
: (this.getWorkerInfo(destinationWorkerNodeKey
) as WorkerInfo
)
1220 this.executeTask(destinationWorkerNodeKey
, task
)
1222 this.enqueueTask(destinationWorkerNodeKey
, task
)
1227 private taskStealingOnEmptyQueue (workerId
: number): void {
1228 const destinationWorkerNodeKey
= this.getWorkerNodeKeyByWorkerId(workerId
)
1229 const destinationWorkerNode
= this.workerNodes
[destinationWorkerNodeKey
]
1230 const workerNodes
= this.workerNodes
1233 (workerNodeA
, workerNodeB
) =>
1234 workerNodeB
.usage
.tasks
.queued
- workerNodeA
.usage
.tasks
.queued
1236 for (const sourceWorkerNode
of workerNodes
) {
1238 sourceWorkerNode
.info
.ready
&&
1239 sourceWorkerNode
.info
.id
!== workerId
&&
1240 sourceWorkerNode
.usage
.tasks
.queued
> 0
1243 ...(sourceWorkerNode
.popTask() as Task
<Data
>),
1244 workerId
: destinationWorkerNode
.info
.id
as number
1247 destinationWorkerNode
.usage
.tasks
.executing
<
1248 (this.opts
.tasksQueueOptions
?.concurrency
as number)
1250 this.executeTask(destinationWorkerNodeKey
, task
)
1252 this.enqueueTask(destinationWorkerNodeKey
, task
)
1259 private tasksStealingOnBackPressure (workerId
: number): void {
1260 const sourceWorkerNode
=
1261 this.workerNodes
[this.getWorkerNodeKeyByWorkerId(workerId
)]
1262 const workerNodes
= this.workerNodes
1265 (workerNodeA
, workerNodeB
) =>
1266 workerNodeA
.usage
.tasks
.queued
- workerNodeB
.usage
.tasks
.queued
1268 for (const [workerNodeKey
, workerNode
] of workerNodes
.entries()) {
1270 workerNode
.info
.ready
&&
1271 workerNode
.info
.id
!== workerId
&&
1272 sourceWorkerNode
.usage
.tasks
.queued
> 0 &&
1273 !workerNode
.hasBackPressure()
1276 ...(sourceWorkerNode
.popTask() as Task
<Data
>),
1277 workerId
: workerNode
.info
.id
as number
1280 workerNode
.usage
.tasks
.executing
<
1281 (this.opts
.tasksQueueOptions
?.concurrency
as number)
1283 this.executeTask(workerNodeKey
, task
)
1285 this.enqueueTask(workerNodeKey
, task
)
1292 * This method is the listener registered for each worker message.
1294 * @returns The listener function to execute when a message is received from a worker.
1296 protected workerListener (): (message
: MessageValue
<Response
>) => void {
1297 return (message
) => {
1298 this.checkMessageWorkerId(message
)
1299 if (message
.ready
!= null && message
.taskFunctions
!= null) {
1300 // Worker ready response received from worker
1301 this.handleWorkerReadyResponse(message
)
1302 } else if (message
.taskId
!= null) {
1303 // Task execution response received from worker
1304 this.handleTaskExecutionResponse(message
)
1305 } else if (message
.taskFunctions
!= null) {
1306 // Task functions message received from worker
1309 this.getWorkerNodeKeyByWorkerId(message
.workerId
)
1311 ).taskFunctions
= message
.taskFunctions
1316 private handleWorkerReadyResponse (message
: MessageValue
<Response
>): void {
1317 if (message
.ready
=== false) {
1318 throw new Error(`Worker ${message.workerId} failed to initialize`)
1320 const workerInfo
= this.getWorkerInfo(
1321 this.getWorkerNodeKeyByWorkerId(message
.workerId
)
1323 workerInfo
.ready
= message
.ready
as boolean
1324 workerInfo
.taskFunctions
= message
.taskFunctions
1325 if (this.emitter
!= null && this.ready
) {
1326 this.emitter
.emit(PoolEvents
.ready
, this.info
)
1330 private handleTaskExecutionResponse (message
: MessageValue
<Response
>): void {
1331 const { taskId
, taskError
, data
} = message
1332 const promiseResponse
= this.promiseResponseMap
.get(taskId
as string)
1333 if (promiseResponse
!= null) {
1334 if (taskError
!= null) {
1335 this.emitter
?.emit(PoolEvents
.taskError
, taskError
)
1336 promiseResponse
.reject(taskError
.message
)
1338 promiseResponse
.resolve(data
as Response
)
1340 const workerNodeKey
= promiseResponse
.workerNodeKey
1341 this.afterTaskExecutionHook(workerNodeKey
, message
)
1342 this.promiseResponseMap
.delete(taskId
as string)
1344 this.opts
.enableTasksQueue
=== true &&
1345 this.tasksQueueSize(workerNodeKey
) > 0 &&
1346 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
<
1347 (this.opts
.tasksQueueOptions
?.concurrency
as number)
1351 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1354 this.workerChoiceStrategyContext
.update(workerNodeKey
)
1358 private checkAndEmitTaskExecutionEvents (): void {
1360 this.emitter
?.emit(PoolEvents
.busy
, this.info
)
1364 private checkAndEmitTaskQueuingEvents (): void {
1365 if (this.hasBackPressure()) {
1366 this.emitter
?.emit(PoolEvents
.backPressure
, this.info
)
1370 private checkAndEmitDynamicWorkerCreationEvents (): void {
1371 if (this.type === PoolTypes
.dynamic
) {
1373 this.emitter
?.emit(PoolEvents
.full
, this.info
)
1379 * Gets the worker information given its worker node key.
1381 * @param workerNodeKey - The worker node key.
1382 * @returns The worker information.
1384 protected getWorkerInfo (workerNodeKey
: number): WorkerInfo
| undefined {
1385 return this.workerNodes
[workerNodeKey
]?.info
1389 * Adds the given worker in the pool worker nodes.
1391 * @param worker - The worker.
1392 * @returns The added worker node key.
1393 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the added worker node is not found.
1395 private addWorkerNode (worker
: Worker
): number {
1396 const workerNode
= new WorkerNode
<Worker
, Data
>(
1399 this.opts
.tasksQueueOptions
?.size
?? Math.pow(this.maxSize
, 2)
1401 // Flag the worker node as ready at pool startup.
1402 if (this.starting
) {
1403 workerNode
.info
.ready
= true
1405 this.workerNodes
.push(workerNode
)
1406 const workerNodeKey
= this.getWorkerNodeKeyByWorker(worker
)
1407 if (workerNodeKey
=== -1) {
1408 throw new Error('Worker node added not found')
1410 return workerNodeKey
1414 * Removes the given worker from the pool worker nodes.
1416 * @param worker - The worker.
1418 private removeWorkerNode (worker
: Worker
): void {
1419 const workerNodeKey
= this.getWorkerNodeKeyByWorker(worker
)
1420 if (workerNodeKey
!== -1) {
1421 this.workerNodes
.splice(workerNodeKey
, 1)
1422 this.workerChoiceStrategyContext
.remove(workerNodeKey
)
1427 public hasWorkerNodeBackPressure (workerNodeKey
: number): boolean {
1429 this.opts
.enableTasksQueue
=== true &&
1430 this.workerNodes
[workerNodeKey
].hasBackPressure()
1434 private hasBackPressure (): boolean {
1436 this.opts
.enableTasksQueue
=== true &&
1437 this.workerNodes
.findIndex(
1438 (workerNode
) => !workerNode
.hasBackPressure()
1444 * Executes the given task on the worker given its worker node key.
1446 * @param workerNodeKey - The worker node key.
1447 * @param task - The task to execute.
1449 private executeTask (workerNodeKey
: number, task
: Task
<Data
>): void {
1450 this.beforeTaskExecutionHook(workerNodeKey
, task
)
1451 this.sendToWorker(workerNodeKey
, task
, task
.transferList
)
1452 this.checkAndEmitTaskExecutionEvents()
1455 private enqueueTask (workerNodeKey
: number, task
: Task
<Data
>): number {
1456 const tasksQueueSize
= this.workerNodes
[workerNodeKey
].enqueueTask(task
)
1457 this.checkAndEmitTaskQueuingEvents()
1458 return tasksQueueSize
1461 private dequeueTask (workerNodeKey
: number): Task
<Data
> | undefined {
1462 return this.workerNodes
[workerNodeKey
].dequeueTask()
1465 private tasksQueueSize (workerNodeKey
: number): number {
1466 return this.workerNodes
[workerNodeKey
].tasksQueueSize()
1469 protected flushTasksQueue (workerNodeKey
: number): void {
1470 while (this.tasksQueueSize(workerNodeKey
) > 0) {
1473 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1476 this.workerNodes
[workerNodeKey
].clearTasksQueue()
1479 private flushTasksQueues (): void {
1480 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
1481 this.flushTasksQueue(workerNodeKey
)