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 * Whether the pool is starting or not.
90 private readonly starting
: boolean
92 * The start timestamp of the pool.
94 private readonly startTimestamp
97 * Constructs a new poolifier pool.
99 * @param numberOfWorkers - Number of workers that this pool should manage.
100 * @param filePath - Path to the worker file.
101 * @param opts - Options for the pool.
104 protected readonly numberOfWorkers
: number,
105 protected readonly filePath
: string,
106 protected readonly opts
: PoolOptions
<Worker
>
108 if (!this.isMain()) {
110 'Cannot start a pool from a worker with the same type as the pool'
113 this.checkNumberOfWorkers(this.numberOfWorkers
)
114 this.checkFilePath(this.filePath
)
115 this.checkPoolOptions(this.opts
)
117 this.chooseWorkerNode
= this.chooseWorkerNode
.bind(this)
118 this.executeTask
= this.executeTask
.bind(this)
119 this.enqueueTask
= this.enqueueTask
.bind(this)
120 this.dequeueTask
= this.dequeueTask
.bind(this)
121 this.checkAndEmitEvents
= this.checkAndEmitEvents
.bind(this)
123 if (this.opts
.enableEvents
=== true) {
124 this.emitter
= new PoolEmitter()
126 this.workerChoiceStrategyContext
= new WorkerChoiceStrategyContext
<
132 this.opts
.workerChoiceStrategy
,
133 this.opts
.workerChoiceStrategyOptions
140 this.starting
= false
142 this.startTimestamp
= performance
.now()
145 private checkFilePath (filePath
: string): void {
148 typeof filePath
!== 'string' ||
149 (typeof filePath
=== 'string' && filePath
.trim().length
=== 0)
151 throw new Error('Please specify a file with a worker implementation')
153 if (!existsSync(filePath
)) {
154 throw new Error(`Cannot find the worker file '${filePath}'`)
158 private checkNumberOfWorkers (numberOfWorkers
: number): void {
159 if (numberOfWorkers
== null) {
161 'Cannot instantiate a pool without specifying the number of workers'
163 } else if (!Number.isSafeInteger(numberOfWorkers
)) {
165 'Cannot instantiate a pool with a non safe integer number of workers'
167 } else if (numberOfWorkers
< 0) {
168 throw new RangeError(
169 'Cannot instantiate a pool with a negative number of workers'
171 } else if (this.type === PoolTypes
.fixed
&& numberOfWorkers
=== 0) {
172 throw new RangeError('Cannot instantiate a fixed pool with zero worker')
176 protected checkDynamicPoolSize (min
: number, max
: number): void {
177 if (this.type === PoolTypes
.dynamic
) {
180 'Cannot instantiate a dynamic pool without specifying the maximum pool size'
182 } else if (!Number.isSafeInteger(max
)) {
184 'Cannot instantiate a dynamic pool with a non safe integer maximum pool size'
186 } else if (min
> max
) {
187 throw new RangeError(
188 'Cannot instantiate a dynamic pool with a maximum pool size inferior to the minimum pool size'
190 } else if (max
=== 0) {
191 throw new RangeError(
192 'Cannot instantiate a dynamic pool with a maximum pool size equal to zero'
194 } else if (min
=== max
) {
195 throw new RangeError(
196 'Cannot instantiate a dynamic pool with a minimum pool size equal to the maximum pool size. Use a fixed pool instead'
202 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
203 if (isPlainObject(opts
)) {
204 this.opts
.workerChoiceStrategy
=
205 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
206 this.checkValidWorkerChoiceStrategy(this.opts
.workerChoiceStrategy
)
207 this.opts
.workerChoiceStrategyOptions
= {
208 ...DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
,
209 ...opts
.workerChoiceStrategyOptions
211 this.checkValidWorkerChoiceStrategyOptions(
212 this.opts
.workerChoiceStrategyOptions
214 this.opts
.restartWorkerOnError
= opts
.restartWorkerOnError
?? true
215 this.opts
.enableEvents
= opts
.enableEvents
?? true
216 this.opts
.enableTasksQueue
= opts
.enableTasksQueue
?? false
217 if (this.opts
.enableTasksQueue
) {
218 this.checkValidTasksQueueOptions(
219 opts
.tasksQueueOptions
as TasksQueueOptions
221 this.opts
.tasksQueueOptions
= this.buildTasksQueueOptions(
222 opts
.tasksQueueOptions
as TasksQueueOptions
226 throw new TypeError('Invalid pool options: must be a plain object')
230 private checkValidWorkerChoiceStrategy (
231 workerChoiceStrategy
: WorkerChoiceStrategy
233 if (!Object.values(WorkerChoiceStrategies
).includes(workerChoiceStrategy
)) {
235 `Invalid worker choice strategy '${workerChoiceStrategy}'`
240 private checkValidWorkerChoiceStrategyOptions (
241 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
243 if (!isPlainObject(workerChoiceStrategyOptions
)) {
245 'Invalid worker choice strategy options: must be a plain object'
249 workerChoiceStrategyOptions
.choiceRetries
!= null &&
250 !Number.isSafeInteger(workerChoiceStrategyOptions
.choiceRetries
)
253 'Invalid worker choice strategy options: choice retries must be an integer'
257 workerChoiceStrategyOptions
.choiceRetries
!= null &&
258 workerChoiceStrategyOptions
.choiceRetries
<= 0
260 throw new RangeError(
261 `Invalid worker choice strategy options: choice retries '${workerChoiceStrategyOptions.choiceRetries}' must be greater than zero`
265 workerChoiceStrategyOptions
.weights
!= null &&
266 Object.keys(workerChoiceStrategyOptions
.weights
).length
!== this.maxSize
269 'Invalid worker choice strategy options: must have a weight for each worker node'
273 workerChoiceStrategyOptions
.measurement
!= null &&
274 !Object.values(Measurements
).includes(
275 workerChoiceStrategyOptions
.measurement
279 `Invalid worker choice strategy options: invalid measurement '${workerChoiceStrategyOptions.measurement}'`
284 private checkValidTasksQueueOptions (
285 tasksQueueOptions
: TasksQueueOptions
287 if (tasksQueueOptions
!= null && !isPlainObject(tasksQueueOptions
)) {
288 throw new TypeError('Invalid tasks queue options: must be a plain object')
291 tasksQueueOptions
?.concurrency
!= null &&
292 !Number.isSafeInteger(tasksQueueOptions
.concurrency
)
295 'Invalid worker tasks concurrency: must be an integer'
299 tasksQueueOptions
?.concurrency
!= null &&
300 tasksQueueOptions
.concurrency
<= 0
303 `Invalid worker tasks concurrency '${tasksQueueOptions.concurrency}'`
308 private startPool (): void {
310 this.workerNodes
.reduce(
311 (accumulator
, workerNode
) =>
312 !workerNode
.info
.dynamic
? accumulator
+ 1 : accumulator
,
314 ) < this.numberOfWorkers
316 this.createAndSetupWorkerNode()
321 public get
info (): PoolInfo
{
327 strategy
: this.opts
.workerChoiceStrategy
as WorkerChoiceStrategy
,
328 minSize
: this.minSize
,
329 maxSize
: this.maxSize
,
330 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
331 .runTime
.aggregate
&&
332 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
333 .waitTime
.aggregate
&& { utilization
: round(this.utilization
) }),
334 workerNodes
: this.workerNodes
.length
,
335 idleWorkerNodes
: this.workerNodes
.reduce(
336 (accumulator
, workerNode
) =>
337 workerNode
.usage
.tasks
.executing
=== 0
342 busyWorkerNodes
: this.workerNodes
.reduce(
343 (accumulator
, workerNode
) =>
344 workerNode
.usage
.tasks
.executing
> 0 ? accumulator
+ 1 : accumulator
,
347 executedTasks
: this.workerNodes
.reduce(
348 (accumulator
, workerNode
) =>
349 accumulator
+ workerNode
.usage
.tasks
.executed
,
352 executingTasks
: this.workerNodes
.reduce(
353 (accumulator
, workerNode
) =>
354 accumulator
+ workerNode
.usage
.tasks
.executing
,
357 ...(this.opts
.enableTasksQueue
=== true && {
358 queuedTasks
: this.workerNodes
.reduce(
359 (accumulator
, workerNode
) =>
360 accumulator
+ workerNode
.usage
.tasks
.queued
,
364 ...(this.opts
.enableTasksQueue
=== true && {
365 maxQueuedTasks
: this.workerNodes
.reduce(
366 (accumulator
, workerNode
) =>
367 accumulator
+ (workerNode
.usage
.tasks
?.maxQueued
?? 0),
371 failedTasks
: this.workerNodes
.reduce(
372 (accumulator
, workerNode
) =>
373 accumulator
+ workerNode
.usage
.tasks
.failed
,
376 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
377 .runTime
.aggregate
&& {
381 ...this.workerNodes
.map(
382 (workerNode
) => workerNode
.usage
.runTime
?.minimum
?? Infinity
388 ...this.workerNodes
.map(
389 (workerNode
) => workerNode
.usage
.runTime
?.maximum
?? -Infinity
394 this.workerNodes
.reduce(
395 (accumulator
, workerNode
) =>
396 accumulator
+ (workerNode
.usage
.runTime
?.aggregate
?? 0),
399 this.workerNodes
.reduce(
400 (accumulator
, workerNode
) =>
401 accumulator
+ (workerNode
.usage
.tasks
?.executed
?? 0),
405 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
409 this.workerNodes
.map(
410 (workerNode
) => workerNode
.usage
.runTime
?.median
?? 0
417 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
418 .waitTime
.aggregate
&& {
422 ...this.workerNodes
.map(
423 (workerNode
) => workerNode
.usage
.waitTime
?.minimum
?? Infinity
429 ...this.workerNodes
.map(
430 (workerNode
) => workerNode
.usage
.waitTime
?.maximum
?? -Infinity
435 this.workerNodes
.reduce(
436 (accumulator
, workerNode
) =>
437 accumulator
+ (workerNode
.usage
.waitTime
?.aggregate
?? 0),
440 this.workerNodes
.reduce(
441 (accumulator
, workerNode
) =>
442 accumulator
+ (workerNode
.usage
.tasks
?.executed
?? 0),
446 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
447 .waitTime
.median
&& {
450 this.workerNodes
.map(
451 (workerNode
) => workerNode
.usage
.waitTime
?.median
?? 0
462 * The pool readiness boolean status.
464 private get
ready (): boolean {
466 this.workerNodes
.reduce(
467 (accumulator
, workerNode
) =>
468 !workerNode
.info
.dynamic
&& workerNode
.info
.ready
477 * The approximate pool utilization.
479 * @returns The pool utilization.
481 private get
utilization (): number {
482 const poolTimeCapacity
=
483 (performance
.now() - this.startTimestamp
) * this.maxSize
484 const totalTasksRunTime
= this.workerNodes
.reduce(
485 (accumulator
, workerNode
) =>
486 accumulator
+ (workerNode
.usage
.runTime
?.aggregate
?? 0),
489 const totalTasksWaitTime
= this.workerNodes
.reduce(
490 (accumulator
, workerNode
) =>
491 accumulator
+ (workerNode
.usage
.waitTime
?.aggregate
?? 0),
494 return (totalTasksRunTime
+ totalTasksWaitTime
) / poolTimeCapacity
500 * If it is `'dynamic'`, it provides the `max` property.
502 protected abstract get
type (): PoolType
507 protected abstract get
worker (): WorkerType
510 * The pool minimum size.
512 protected abstract get
minSize (): number
515 * The pool maximum size.
517 protected abstract get
maxSize (): number
520 * Checks if the worker id sent in the received message from a worker is valid.
522 * @param message - The received message.
523 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the worker id is invalid.
525 private checkMessageWorkerId (message
: MessageValue
<Response
>): void {
526 if (message
.workerId
== null) {
527 throw new Error('Worker message received without worker id')
529 message
.workerId
!= null &&
530 this.getWorkerNodeKeyByWorkerId(message
.workerId
) === -1
533 `Worker message received from unknown worker '${message.workerId}'`
539 * Gets the given worker its worker node key.
541 * @param worker - The worker.
542 * @returns The worker node key if found in the pool worker nodes, `-1` otherwise.
544 private getWorkerNodeKeyByWorker (worker
: Worker
): number {
545 return this.workerNodes
.findIndex(
546 (workerNode
) => workerNode
.worker
=== worker
551 * Gets the worker node key given its worker id.
553 * @param workerId - The worker id.
554 * @returns The worker node key if the worker id is found in the pool worker nodes, `-1` otherwise.
556 private getWorkerNodeKeyByWorkerId (workerId
: number): number {
557 return this.workerNodes
.findIndex(
558 (workerNode
) => workerNode
.info
.id
=== workerId
563 public setWorkerChoiceStrategy (
564 workerChoiceStrategy
: WorkerChoiceStrategy
,
565 workerChoiceStrategyOptions
?: WorkerChoiceStrategyOptions
567 this.checkValidWorkerChoiceStrategy(workerChoiceStrategy
)
568 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
569 this.workerChoiceStrategyContext
.setWorkerChoiceStrategy(
570 this.opts
.workerChoiceStrategy
572 if (workerChoiceStrategyOptions
!= null) {
573 this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
575 for (const [workerNodeKey
, workerNode
] of this.workerNodes
.entries()) {
576 workerNode
.resetUsage()
577 this.sendStatisticsMessageToWorker(workerNodeKey
)
582 public setWorkerChoiceStrategyOptions (
583 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
585 this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
586 this.opts
.workerChoiceStrategyOptions
= {
587 ...DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
,
588 ...workerChoiceStrategyOptions
590 this.workerChoiceStrategyContext
.setOptions(
591 this.opts
.workerChoiceStrategyOptions
596 public enableTasksQueue (
598 tasksQueueOptions
?: TasksQueueOptions
600 if (this.opts
.enableTasksQueue
=== true && !enable
) {
601 this.flushTasksQueues()
603 this.opts
.enableTasksQueue
= enable
604 this.setTasksQueueOptions(tasksQueueOptions
as TasksQueueOptions
)
608 public setTasksQueueOptions (tasksQueueOptions
: TasksQueueOptions
): void {
609 if (this.opts
.enableTasksQueue
=== true) {
610 this.checkValidTasksQueueOptions(tasksQueueOptions
)
611 this.opts
.tasksQueueOptions
=
612 this.buildTasksQueueOptions(tasksQueueOptions
)
613 } else if (this.opts
.tasksQueueOptions
!= null) {
614 delete this.opts
.tasksQueueOptions
618 private buildTasksQueueOptions (
619 tasksQueueOptions
: TasksQueueOptions
620 ): TasksQueueOptions
{
622 concurrency
: tasksQueueOptions
?.concurrency
?? 1
627 * Whether the pool is full or not.
629 * The pool filling boolean status.
631 protected get
full (): boolean {
632 return this.workerNodes
.length
>= this.maxSize
636 * Whether the pool is busy or not.
638 * The pool busyness boolean status.
640 protected abstract get
busy (): boolean
643 * Whether worker nodes are executing concurrently their tasks quota or not.
645 * @returns Worker nodes busyness boolean status.
647 protected internalBusy (): boolean {
648 if (this.opts
.enableTasksQueue
=== true) {
650 this.workerNodes
.findIndex(
652 workerNode
.info
.ready
&&
653 workerNode
.usage
.tasks
.executing
<
654 (this.opts
.tasksQueueOptions
?.concurrency
as number)
659 this.workerNodes
.findIndex(
661 workerNode
.info
.ready
&& workerNode
.usage
.tasks
.executing
=== 0
668 public listTaskFunctions (): string[] {
669 for (const workerNode
of this.workerNodes
) {
671 Array.isArray(workerNode
.info
.taskFunctions
) &&
672 workerNode
.info
.taskFunctions
.length
> 0
674 return workerNode
.info
.taskFunctions
681 public async execute (
684 transferList
?: TransferListItem
[]
685 ): Promise
<Response
> {
686 return await new Promise
<Response
>((resolve
, reject
) => {
687 if (name
!= null && typeof name
!== 'string') {
688 reject(new TypeError('name argument must be a string'))
692 typeof name
=== 'string' &&
693 name
.trim().length
=== 0
695 reject(new TypeError('name argument must not be an empty string'))
697 if (transferList
!= null && !Array.isArray(transferList
)) {
698 reject(new TypeError('transferList argument must be an array'))
700 const timestamp
= performance
.now()
701 const workerNodeKey
= this.chooseWorkerNode()
702 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
705 Array.isArray(workerInfo
.taskFunctions
) &&
706 !workerInfo
.taskFunctions
.includes(name
)
709 new Error(`Task function '${name}' is not registered in the pool`)
712 const task
: Task
<Data
> = {
713 name
: name
?? DEFAULT_TASK_NAME
,
714 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
715 data
: data
?? ({} as Data
),
718 workerId
: workerInfo
.id
as number,
721 this.promiseResponseMap
.set(task
.taskId
as string, {
727 this.opts
.enableTasksQueue
=== false ||
728 (this.opts
.enableTasksQueue
=== true &&
729 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
<
730 (this.opts
.tasksQueueOptions
?.concurrency
as number))
732 this.executeTask(workerNodeKey
, task
)
734 this.enqueueTask(workerNodeKey
, task
)
736 this.checkAndEmitEvents()
741 public async destroy (): Promise
<void> {
743 this.workerNodes
.map(async (_
, workerNodeKey
) => {
744 await this.destroyWorkerNode(workerNodeKey
)
747 this.emitter
?.emit(PoolEvents
.destroy
)
750 protected async sendKillMessageToWorker (
751 workerNodeKey
: number,
754 await new Promise
<void>((resolve
, reject
) => {
755 this.registerWorkerMessageListener(workerNodeKey
, (message
) => {
756 if (message
.kill
=== 'success') {
758 } else if (message
.kill
=== 'failure') {
759 reject(new Error(`Worker ${workerId} kill message handling failed`))
762 this.sendToWorker(workerNodeKey
, { kill
: true, workerId
})
767 * Terminates the worker node given its worker node key.
769 * @param workerNodeKey - The worker node key.
771 protected abstract destroyWorkerNode (workerNodeKey
: number): Promise
<void>
774 * Setup hook to execute code before worker nodes are created in the abstract constructor.
779 protected setupHook (): void {
780 // Intentionally empty
784 * Should return whether the worker is the main worker or not.
786 protected abstract isMain (): boolean
789 * Hook executed before the worker task execution.
792 * @param workerNodeKey - The worker node key.
793 * @param task - The task to execute.
795 protected beforeTaskExecutionHook (
796 workerNodeKey
: number,
799 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
800 ++workerUsage
.tasks
.executing
801 this.updateWaitTimeWorkerUsage(workerUsage
, task
)
802 if (this.canUpdateTaskWorkerUsage(workerNodeKey
)) {
803 const taskWorkerUsage
= this.workerNodes
[
805 ].getTaskWorkerUsage(task
.name
as string) as WorkerUsage
806 ++taskWorkerUsage
.tasks
.executing
807 this.updateWaitTimeWorkerUsage(taskWorkerUsage
, task
)
812 * Hook executed after the worker task execution.
815 * @param workerNodeKey - The worker node key.
816 * @param message - The received message.
818 protected afterTaskExecutionHook (
819 workerNodeKey
: number,
820 message
: MessageValue
<Response
>
822 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
823 this.updateTaskStatisticsWorkerUsage(workerUsage
, message
)
824 this.updateRunTimeWorkerUsage(workerUsage
, message
)
825 this.updateEluWorkerUsage(workerUsage
, message
)
826 if (this.canUpdateTaskWorkerUsage(workerNodeKey
)) {
827 const taskWorkerUsage
= this.workerNodes
[
829 ].getTaskWorkerUsage(
830 message
.taskPerformance
?.name
?? DEFAULT_TASK_NAME
832 this.updateTaskStatisticsWorkerUsage(taskWorkerUsage
, message
)
833 this.updateRunTimeWorkerUsage(taskWorkerUsage
, message
)
834 this.updateEluWorkerUsage(taskWorkerUsage
, message
)
838 private canUpdateTaskWorkerUsage (workerNodeKey
: number): boolean {
839 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
841 Array.isArray(workerInfo
.taskFunctions
) &&
842 workerInfo
.taskFunctions
.length
> 1
846 private updateTaskStatisticsWorkerUsage (
847 workerUsage
: WorkerUsage
,
848 message
: MessageValue
<Response
>
850 const workerTaskStatistics
= workerUsage
.tasks
851 --workerTaskStatistics
.executing
852 if (message
.taskError
== null) {
853 ++workerTaskStatistics
.executed
855 ++workerTaskStatistics
.failed
859 private updateRunTimeWorkerUsage (
860 workerUsage
: WorkerUsage
,
861 message
: MessageValue
<Response
>
863 updateMeasurementStatistics(
865 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
,
866 message
.taskPerformance
?.runTime
?? 0,
867 workerUsage
.tasks
.executed
871 private updateWaitTimeWorkerUsage (
872 workerUsage
: WorkerUsage
,
875 const timestamp
= performance
.now()
876 const taskWaitTime
= timestamp
- (task
.timestamp
?? timestamp
)
877 updateMeasurementStatistics(
878 workerUsage
.waitTime
,
879 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().waitTime
,
881 workerUsage
.tasks
.executed
885 private updateEluWorkerUsage (
886 workerUsage
: WorkerUsage
,
887 message
: MessageValue
<Response
>
889 const eluTaskStatisticsRequirements
: MeasurementStatisticsRequirements
=
890 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
891 updateMeasurementStatistics(
892 workerUsage
.elu
.active
,
893 eluTaskStatisticsRequirements
,
894 message
.taskPerformance
?.elu
?.active
?? 0,
895 workerUsage
.tasks
.executed
897 updateMeasurementStatistics(
898 workerUsage
.elu
.idle
,
899 eluTaskStatisticsRequirements
,
900 message
.taskPerformance
?.elu
?.idle
?? 0,
901 workerUsage
.tasks
.executed
903 if (eluTaskStatisticsRequirements
.aggregate
) {
904 if (message
.taskPerformance
?.elu
!= null) {
905 if (workerUsage
.elu
.utilization
!= null) {
906 workerUsage
.elu
.utilization
=
907 (workerUsage
.elu
.utilization
+
908 message
.taskPerformance
.elu
.utilization
) /
911 workerUsage
.elu
.utilization
= message
.taskPerformance
.elu
.utilization
918 * Chooses a worker node for the next task.
920 * The default worker choice strategy uses a round robin algorithm to distribute the tasks.
922 * @returns The chosen worker node key
924 private chooseWorkerNode (): number {
925 if (this.shallCreateDynamicWorker()) {
926 const workerNodeKey
= this.createAndSetupDynamicWorkerNode()
928 this.workerChoiceStrategyContext
.getStrategyPolicy().useDynamicWorker
933 return this.workerChoiceStrategyContext
.execute()
937 * Conditions for dynamic worker creation.
939 * @returns Whether to create a dynamic worker or not.
941 private shallCreateDynamicWorker (): boolean {
942 return this.type === PoolTypes
.dynamic
&& !this.full
&& this.internalBusy()
946 * Sends a message to worker given its worker node key.
948 * @param workerNodeKey - The worker node key.
949 * @param message - The message.
950 * @param transferList - The optional array of transferable objects.
952 protected abstract sendToWorker (
953 workerNodeKey
: number,
954 message
: MessageValue
<Data
>,
955 transferList
?: TransferListItem
[]
959 * Creates a new worker.
961 * @returns Newly created worker.
963 protected abstract createWorker (): Worker
966 * Creates a new, completely set up worker node.
968 * @returns New, completely set up worker node key.
970 protected createAndSetupWorkerNode (): number {
971 const worker
= this.createWorker()
973 worker
.on('online', this.opts
.onlineHandler
?? EMPTY_FUNCTION
)
974 worker
.on('message', this.opts
.messageHandler
?? EMPTY_FUNCTION
)
975 worker
.on('error', this.opts
.errorHandler
?? EMPTY_FUNCTION
)
976 worker
.on('error', (error
) => {
977 const workerNodeKey
= this.getWorkerNodeKeyByWorker(worker
)
978 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
979 workerInfo
.ready
= false
980 this.workerNodes
[workerNodeKey
].closeChannel()
981 this.emitter
?.emit(PoolEvents
.error
, error
)
982 if (this.opts
.restartWorkerOnError
=== true && !this.starting
) {
983 if (workerInfo
.dynamic
) {
984 this.createAndSetupDynamicWorkerNode()
986 this.createAndSetupWorkerNode()
989 if (this.opts
.enableTasksQueue
=== true) {
990 this.redistributeQueuedTasks(workerNodeKey
)
993 worker
.on('exit', this.opts
.exitHandler
?? EMPTY_FUNCTION
)
994 worker
.once('exit', () => {
995 this.removeWorkerNode(worker
)
998 const workerNodeKey
= this.addWorkerNode(worker
)
1000 this.afterWorkerNodeSetup(workerNodeKey
)
1002 return workerNodeKey
1006 * Creates a new, completely set up dynamic worker node.
1008 * @returns New, completely set up dynamic worker node key.
1010 protected createAndSetupDynamicWorkerNode (): number {
1011 const workerNodeKey
= this.createAndSetupWorkerNode()
1012 this.registerWorkerMessageListener(workerNodeKey
, (message
) => {
1013 const localWorkerNodeKey
= this.getWorkerNodeKeyByWorkerId(
1016 const workerUsage
= this.workerNodes
[localWorkerNodeKey
].usage
1017 // Kill message received from worker
1019 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
1020 (isKillBehavior(KillBehaviors
.SOFT
, message
.kill
) &&
1021 ((this.opts
.enableTasksQueue
=== false &&
1022 workerUsage
.tasks
.executing
=== 0) ||
1023 (this.opts
.enableTasksQueue
=== true &&
1024 workerUsage
.tasks
.executing
=== 0 &&
1025 this.tasksQueueSize(localWorkerNodeKey
) === 0)))
1027 this.destroyWorkerNode(localWorkerNodeKey
).catch((error
) => {
1028 this.emitter
?.emit(PoolEvents
.error
, error
)
1032 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1033 this.sendToWorker(workerNodeKey
, {
1035 workerId
: workerInfo
.id
as number
1037 workerInfo
.dynamic
= true
1038 if (this.workerChoiceStrategyContext
.getStrategyPolicy().useDynamicWorker
) {
1039 workerInfo
.ready
= true
1041 return workerNodeKey
1045 * Registers a listener callback on the worker given its worker node key.
1047 * @param workerNodeKey - The worker node key.
1048 * @param listener - The message listener callback.
1050 protected abstract registerWorkerMessageListener
<
1051 Message
extends Data
| Response
1053 workerNodeKey
: number,
1054 listener
: (message
: MessageValue
<Message
>) => void
1058 * Method hooked up after a worker node has been newly created.
1059 * Can be overridden.
1061 * @param workerNodeKey - The newly created worker node key.
1063 protected afterWorkerNodeSetup (workerNodeKey
: number): void {
1064 // Listen to worker messages.
1065 this.registerWorkerMessageListener(workerNodeKey
, this.workerListener())
1066 // Send the startup message to worker.
1067 this.sendStartupMessageToWorker(workerNodeKey
)
1068 // Send the statistics message to worker.
1069 this.sendStatisticsMessageToWorker(workerNodeKey
)
1073 * Sends the startup message to worker given its worker node key.
1075 * @param workerNodeKey - The worker node key.
1077 protected abstract sendStartupMessageToWorker (workerNodeKey
: number): void
1080 * Sends the statistics message to worker given its worker node key.
1082 * @param workerNodeKey - The worker node key.
1084 private sendStatisticsMessageToWorker (workerNodeKey
: number): void {
1085 this.sendToWorker(workerNodeKey
, {
1088 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1090 elu
: this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1093 workerId
: this.getWorkerInfo(workerNodeKey
).id
as number
1097 private redistributeQueuedTasks (workerNodeKey
: number): void {
1098 while (this.tasksQueueSize(workerNodeKey
) > 0) {
1099 let targetWorkerNodeKey
: number = workerNodeKey
1100 let minQueuedTasks
= Infinity
1101 let executeTask
= false
1102 for (const [workerNodeId
, workerNode
] of this.workerNodes
.entries()) {
1103 const workerInfo
= this.getWorkerInfo(workerNodeId
)
1105 workerNodeId
!== workerNodeKey
&&
1107 workerNode
.usage
.tasks
.queued
=== 0
1110 this.workerNodes
[workerNodeId
].usage
.tasks
.executing
<
1111 (this.opts
.tasksQueueOptions
?.concurrency
as number)
1115 targetWorkerNodeKey
= workerNodeId
1119 workerNodeId
!== workerNodeKey
&&
1121 workerNode
.usage
.tasks
.queued
< minQueuedTasks
1123 minQueuedTasks
= workerNode
.usage
.tasks
.queued
1124 targetWorkerNodeKey
= workerNodeId
1129 targetWorkerNodeKey
,
1130 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1134 targetWorkerNodeKey
,
1135 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1142 * This method is the listener registered for each worker message.
1144 * @returns The listener function to execute when a message is received from a worker.
1146 protected workerListener (): (message
: MessageValue
<Response
>) => void {
1147 return (message
) => {
1148 this.checkMessageWorkerId(message
)
1149 if (message
.ready
!= null && message
.taskFunctions
!= null) {
1150 // Worker ready response received from worker
1151 this.handleWorkerReadyResponse(message
)
1152 } else if (message
.taskId
!= null) {
1153 // Task execution response received from worker
1154 this.handleTaskExecutionResponse(message
)
1155 } else if (message
.taskFunctions
!= null) {
1156 // Task functions message received from worker
1158 this.getWorkerNodeKeyByWorkerId(message
.workerId
)
1159 ).taskFunctions
= message
.taskFunctions
1164 private handleWorkerReadyResponse (message
: MessageValue
<Response
>): void {
1165 if (message
.ready
=== false) {
1166 throw new Error(`Worker ${message.workerId} failed to initialize`)
1168 const workerInfo
= this.getWorkerInfo(
1169 this.getWorkerNodeKeyByWorkerId(message
.workerId
)
1171 workerInfo
.ready
= message
.ready
as boolean
1172 workerInfo
.taskFunctions
= message
.taskFunctions
1173 if (this.emitter
!= null && this.ready
) {
1174 this.emitter
.emit(PoolEvents
.ready
, this.info
)
1178 private handleTaskExecutionResponse (message
: MessageValue
<Response
>): void {
1179 const { taskId
, taskError
, data
} = message
1180 const promiseResponse
= this.promiseResponseMap
.get(taskId
as string)
1181 if (promiseResponse
!= null) {
1182 if (taskError
!= null) {
1183 this.emitter
?.emit(PoolEvents
.taskError
, taskError
)
1184 promiseResponse
.reject(taskError
.message
)
1186 promiseResponse
.resolve(data
as Response
)
1188 const workerNodeKey
= promiseResponse
.workerNodeKey
1189 this.afterTaskExecutionHook(workerNodeKey
, message
)
1190 this.promiseResponseMap
.delete(taskId
as string)
1192 this.opts
.enableTasksQueue
=== true &&
1193 this.tasksQueueSize(workerNodeKey
) > 0 &&
1194 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
<
1195 (this.opts
.tasksQueueOptions
?.concurrency
as number)
1199 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1202 this.workerChoiceStrategyContext
.update(workerNodeKey
)
1206 private checkAndEmitEvents (): void {
1207 if (this.emitter
!= null) {
1209 this.emitter
.emit(PoolEvents
.busy
, this.info
)
1211 if (this.type === PoolTypes
.dynamic
&& this.full
) {
1212 this.emitter
.emit(PoolEvents
.full
, this.info
)
1218 * Gets the worker information given its worker node key.
1220 * @param workerNodeKey - The worker node key.
1221 * @returns The worker information.
1223 protected getWorkerInfo (workerNodeKey
: number): WorkerInfo
{
1224 return this.workerNodes
[workerNodeKey
].info
1228 * Adds the given worker in the pool worker nodes.
1230 * @param worker - The worker.
1231 * @returns The added worker node key.
1232 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the added worker node is not found.
1234 private addWorkerNode (worker
: Worker
): number {
1235 const workerNode
= new WorkerNode
<Worker
, Data
>(
1240 // Flag the worker node as ready at pool startup.
1241 if (this.starting
) {
1242 workerNode
.info
.ready
= true
1244 this.workerNodes
.push(workerNode
)
1245 const workerNodeKey
= this.getWorkerNodeKeyByWorker(worker
)
1246 if (workerNodeKey
=== -1) {
1247 throw new Error('Worker node added not found')
1249 return workerNodeKey
1253 * Removes the given worker from the pool worker nodes.
1255 * @param worker - The worker.
1257 private removeWorkerNode (worker
: Worker
): void {
1258 const workerNodeKey
= this.getWorkerNodeKeyByWorker(worker
)
1259 if (workerNodeKey
!== -1) {
1260 this.workerNodes
.splice(workerNodeKey
, 1)
1261 this.workerChoiceStrategyContext
.remove(workerNodeKey
)
1266 public hasWorkerNodeBackPressure (workerNodeKey
: number): boolean {
1268 this.opts
.enableTasksQueue
=== true &&
1269 this.workerNodes
[workerNodeKey
].hasBackPressure()
1277 * Executes the given task on the worker given its worker node key.
1279 * @param workerNodeKey - The worker node key.
1280 * @param task - The task to execute.
1282 private executeTask (workerNodeKey
: number, task
: Task
<Data
>): void {
1283 this.beforeTaskExecutionHook(workerNodeKey
, task
)
1284 this.sendToWorker(workerNodeKey
, task
, task
.transferList
)
1287 private enqueueTask (workerNodeKey
: number, task
: Task
<Data
>): number {
1288 const tasksQueueSize
= this.workerNodes
[workerNodeKey
].enqueueTask(task
)
1289 if (this.hasWorkerNodeBackPressure(workerNodeKey
)) {
1290 this.emitter
?.emit(PoolEvents
.backPressure
, {
1291 workerId
: this.getWorkerInfo(workerNodeKey
).id
,
1295 return tasksQueueSize
1298 private dequeueTask (workerNodeKey
: number): Task
<Data
> | undefined {
1299 return this.workerNodes
[workerNodeKey
].dequeueTask()
1302 private tasksQueueSize (workerNodeKey
: number): number {
1303 return this.workerNodes
[workerNodeKey
].tasksQueueSize()
1306 protected flushTasksQueue (workerNodeKey
: number): void {
1307 while (this.tasksQueueSize(workerNodeKey
) > 0) {
1310 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1313 this.workerNodes
[workerNodeKey
].clearTasksQueue()
1316 private flushTasksQueues (): void {
1317 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
1318 this.flushTasksQueue(workerNodeKey
)