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.checkAndEmitTaskExecutionEvents
=
122 this.checkAndEmitTaskExecutionEvents
.bind(this)
123 this.checkAndEmitTaskQueuingEvents
=
124 this.checkAndEmitTaskQueuingEvents
.bind(this)
125 this.checkAndEmitDynamicWorkerCreationEvents
=
126 this.checkAndEmitDynamicWorkerCreationEvents
.bind(this)
128 if (this.opts
.enableEvents
=== true) {
129 this.emitter
= new PoolEmitter()
131 this.workerChoiceStrategyContext
= new WorkerChoiceStrategyContext
<
137 this.opts
.workerChoiceStrategy
,
138 this.opts
.workerChoiceStrategyOptions
145 this.starting
= false
147 this.startTimestamp
= performance
.now()
150 private checkFilePath (filePath
: string): void {
153 typeof filePath
!== 'string' ||
154 (typeof filePath
=== 'string' && filePath
.trim().length
=== 0)
156 throw new Error('Please specify a file with a worker implementation')
158 if (!existsSync(filePath
)) {
159 throw new Error(`Cannot find the worker file '${filePath}'`)
163 private checkNumberOfWorkers (numberOfWorkers
: number): void {
164 if (numberOfWorkers
== null) {
166 'Cannot instantiate a pool without specifying the number of workers'
168 } else if (!Number.isSafeInteger(numberOfWorkers
)) {
170 'Cannot instantiate a pool with a non safe integer number of workers'
172 } else if (numberOfWorkers
< 0) {
173 throw new RangeError(
174 'Cannot instantiate a pool with a negative number of workers'
176 } else if (this.type === PoolTypes
.fixed
&& numberOfWorkers
=== 0) {
177 throw new RangeError('Cannot instantiate a fixed pool with zero worker')
181 protected checkDynamicPoolSize (min
: number, max
: number): void {
182 if (this.type === PoolTypes
.dynamic
) {
185 'Cannot instantiate a dynamic pool without specifying the maximum pool size'
187 } else if (!Number.isSafeInteger(max
)) {
189 'Cannot instantiate a dynamic pool with a non safe integer maximum pool size'
191 } else if (min
> max
) {
192 throw new RangeError(
193 'Cannot instantiate a dynamic pool with a maximum pool size inferior to the minimum pool size'
195 } else if (max
=== 0) {
196 throw new RangeError(
197 'Cannot instantiate a dynamic pool with a maximum pool size equal to zero'
199 } else if (min
=== max
) {
200 throw new RangeError(
201 'Cannot instantiate a dynamic pool with a minimum pool size equal to the maximum pool size. Use a fixed pool instead'
207 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
208 if (isPlainObject(opts
)) {
209 this.opts
.workerChoiceStrategy
=
210 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
211 this.checkValidWorkerChoiceStrategy(this.opts
.workerChoiceStrategy
)
212 this.opts
.workerChoiceStrategyOptions
= {
213 ...DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
,
214 ...opts
.workerChoiceStrategyOptions
216 this.checkValidWorkerChoiceStrategyOptions(
217 this.opts
.workerChoiceStrategyOptions
219 this.opts
.restartWorkerOnError
= opts
.restartWorkerOnError
?? true
220 this.opts
.enableEvents
= opts
.enableEvents
?? true
221 this.opts
.enableTasksQueue
= opts
.enableTasksQueue
?? false
222 if (this.opts
.enableTasksQueue
) {
223 this.checkValidTasksQueueOptions(
224 opts
.tasksQueueOptions
as TasksQueueOptions
226 this.opts
.tasksQueueOptions
= this.buildTasksQueueOptions(
227 opts
.tasksQueueOptions
as TasksQueueOptions
231 throw new TypeError('Invalid pool options: must be a plain object')
235 private checkValidWorkerChoiceStrategy (
236 workerChoiceStrategy
: WorkerChoiceStrategy
238 if (!Object.values(WorkerChoiceStrategies
).includes(workerChoiceStrategy
)) {
240 `Invalid worker choice strategy '${workerChoiceStrategy}'`
245 private checkValidWorkerChoiceStrategyOptions (
246 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
248 if (!isPlainObject(workerChoiceStrategyOptions
)) {
250 'Invalid worker choice strategy options: must be a plain object'
254 workerChoiceStrategyOptions
.choiceRetries
!= null &&
255 !Number.isSafeInteger(workerChoiceStrategyOptions
.choiceRetries
)
258 'Invalid worker choice strategy options: choice retries must be an integer'
262 workerChoiceStrategyOptions
.choiceRetries
!= null &&
263 workerChoiceStrategyOptions
.choiceRetries
<= 0
265 throw new RangeError(
266 `Invalid worker choice strategy options: choice retries '${workerChoiceStrategyOptions.choiceRetries}' must be greater than zero`
270 workerChoiceStrategyOptions
.weights
!= null &&
271 Object.keys(workerChoiceStrategyOptions
.weights
).length
!== this.maxSize
274 'Invalid worker choice strategy options: must have a weight for each worker node'
278 workerChoiceStrategyOptions
.measurement
!= null &&
279 !Object.values(Measurements
).includes(
280 workerChoiceStrategyOptions
.measurement
284 `Invalid worker choice strategy options: invalid measurement '${workerChoiceStrategyOptions.measurement}'`
289 private checkValidTasksQueueOptions (
290 tasksQueueOptions
: TasksQueueOptions
292 if (tasksQueueOptions
!= null && !isPlainObject(tasksQueueOptions
)) {
293 throw new TypeError('Invalid tasks queue options: must be a plain object')
296 tasksQueueOptions
?.concurrency
!= null &&
297 !Number.isSafeInteger(tasksQueueOptions
.concurrency
)
300 'Invalid worker tasks concurrency: must be an integer'
304 tasksQueueOptions
?.concurrency
!= null &&
305 tasksQueueOptions
.concurrency
<= 0
308 `Invalid worker tasks concurrency '${tasksQueueOptions.concurrency}' is a negative integer or zero`
313 private startPool (): void {
315 this.workerNodes
.reduce(
316 (accumulator
, workerNode
) =>
317 !workerNode
.info
.dynamic
? accumulator
+ 1 : accumulator
,
319 ) < this.numberOfWorkers
321 this.createAndSetupWorkerNode()
326 public get
info (): PoolInfo
{
332 strategy
: this.opts
.workerChoiceStrategy
as WorkerChoiceStrategy
,
333 minSize
: this.minSize
,
334 maxSize
: this.maxSize
,
335 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
336 .runTime
.aggregate
&&
337 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
338 .waitTime
.aggregate
&& { utilization
: round(this.utilization
) }),
339 workerNodes
: this.workerNodes
.length
,
340 idleWorkerNodes
: this.workerNodes
.reduce(
341 (accumulator
, workerNode
) =>
342 workerNode
.usage
.tasks
.executing
=== 0
347 busyWorkerNodes
: this.workerNodes
.reduce(
348 (accumulator
, workerNode
) =>
349 workerNode
.usage
.tasks
.executing
> 0 ? accumulator
+ 1 : accumulator
,
352 executedTasks
: this.workerNodes
.reduce(
353 (accumulator
, workerNode
) =>
354 accumulator
+ workerNode
.usage
.tasks
.executed
,
357 executingTasks
: this.workerNodes
.reduce(
358 (accumulator
, workerNode
) =>
359 accumulator
+ workerNode
.usage
.tasks
.executing
,
362 ...(this.opts
.enableTasksQueue
=== true && {
363 queuedTasks
: this.workerNodes
.reduce(
364 (accumulator
, workerNode
) =>
365 accumulator
+ workerNode
.usage
.tasks
.queued
,
369 ...(this.opts
.enableTasksQueue
=== true && {
370 maxQueuedTasks
: this.workerNodes
.reduce(
371 (accumulator
, workerNode
) =>
372 accumulator
+ (workerNode
.usage
.tasks
?.maxQueued
?? 0),
376 ...(this.opts
.enableTasksQueue
=== true && {
377 backPressure
: this.hasBackPressure()
379 failedTasks
: this.workerNodes
.reduce(
380 (accumulator
, workerNode
) =>
381 accumulator
+ workerNode
.usage
.tasks
.failed
,
384 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
385 .runTime
.aggregate
&& {
389 ...this.workerNodes
.map(
390 (workerNode
) => workerNode
.usage
.runTime
?.minimum
?? Infinity
396 ...this.workerNodes
.map(
397 (workerNode
) => workerNode
.usage
.runTime
?.maximum
?? -Infinity
402 this.workerNodes
.reduce(
403 (accumulator
, workerNode
) =>
404 accumulator
+ (workerNode
.usage
.runTime
?.aggregate
?? 0),
407 this.workerNodes
.reduce(
408 (accumulator
, workerNode
) =>
409 accumulator
+ (workerNode
.usage
.tasks
?.executed
?? 0),
413 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
417 this.workerNodes
.map(
418 (workerNode
) => workerNode
.usage
.runTime
?.median
?? 0
425 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
426 .waitTime
.aggregate
&& {
430 ...this.workerNodes
.map(
431 (workerNode
) => workerNode
.usage
.waitTime
?.minimum
?? Infinity
437 ...this.workerNodes
.map(
438 (workerNode
) => workerNode
.usage
.waitTime
?.maximum
?? -Infinity
443 this.workerNodes
.reduce(
444 (accumulator
, workerNode
) =>
445 accumulator
+ (workerNode
.usage
.waitTime
?.aggregate
?? 0),
448 this.workerNodes
.reduce(
449 (accumulator
, workerNode
) =>
450 accumulator
+ (workerNode
.usage
.tasks
?.executed
?? 0),
454 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
455 .waitTime
.median
&& {
458 this.workerNodes
.map(
459 (workerNode
) => workerNode
.usage
.waitTime
?.median
?? 0
470 * The pool readiness boolean status.
472 private get
ready (): boolean {
474 this.workerNodes
.reduce(
475 (accumulator
, workerNode
) =>
476 !workerNode
.info
.dynamic
&& workerNode
.info
.ready
485 * The approximate pool utilization.
487 * @returns The pool utilization.
489 private get
utilization (): number {
490 const poolTimeCapacity
=
491 (performance
.now() - this.startTimestamp
) * this.maxSize
492 const totalTasksRunTime
= this.workerNodes
.reduce(
493 (accumulator
, workerNode
) =>
494 accumulator
+ (workerNode
.usage
.runTime
?.aggregate
?? 0),
497 const totalTasksWaitTime
= this.workerNodes
.reduce(
498 (accumulator
, workerNode
) =>
499 accumulator
+ (workerNode
.usage
.waitTime
?.aggregate
?? 0),
502 return (totalTasksRunTime
+ totalTasksWaitTime
) / poolTimeCapacity
508 * If it is `'dynamic'`, it provides the `max` property.
510 protected abstract get
type (): PoolType
515 protected abstract get
worker (): WorkerType
518 * The pool minimum size.
520 protected abstract get
minSize (): number
523 * The pool maximum size.
525 protected abstract get
maxSize (): number
528 * Checks if the worker id sent in the received message from a worker is valid.
530 * @param message - The received message.
531 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the worker id is invalid.
533 private checkMessageWorkerId (message
: MessageValue
<Response
>): void {
534 if (message
.workerId
== null) {
535 throw new Error('Worker message received without worker id')
537 message
.workerId
!= null &&
538 this.getWorkerNodeKeyByWorkerId(message
.workerId
) === -1
541 `Worker message received from unknown worker '${message.workerId}'`
547 * Gets the given worker its worker node key.
549 * @param worker - The worker.
550 * @returns The worker node key if found in the pool worker nodes, `-1` otherwise.
552 private getWorkerNodeKeyByWorker (worker
: Worker
): number {
553 return this.workerNodes
.findIndex(
554 (workerNode
) => workerNode
.worker
=== worker
559 * Gets the worker node key given its worker id.
561 * @param workerId - The worker id.
562 * @returns The worker node key if the worker id is found in the pool worker nodes, `-1` otherwise.
564 private getWorkerNodeKeyByWorkerId (workerId
: number): number {
565 return this.workerNodes
.findIndex(
566 (workerNode
) => workerNode
.info
.id
=== workerId
571 public setWorkerChoiceStrategy (
572 workerChoiceStrategy
: WorkerChoiceStrategy
,
573 workerChoiceStrategyOptions
?: WorkerChoiceStrategyOptions
575 this.checkValidWorkerChoiceStrategy(workerChoiceStrategy
)
576 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
577 this.workerChoiceStrategyContext
.setWorkerChoiceStrategy(
578 this.opts
.workerChoiceStrategy
580 if (workerChoiceStrategyOptions
!= null) {
581 this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
583 for (const [workerNodeKey
, workerNode
] of this.workerNodes
.entries()) {
584 workerNode
.resetUsage()
585 this.sendStatisticsMessageToWorker(workerNodeKey
)
590 public setWorkerChoiceStrategyOptions (
591 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
593 this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
594 this.opts
.workerChoiceStrategyOptions
= {
595 ...DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
,
596 ...workerChoiceStrategyOptions
598 this.workerChoiceStrategyContext
.setOptions(
599 this.opts
.workerChoiceStrategyOptions
604 public enableTasksQueue (
606 tasksQueueOptions
?: TasksQueueOptions
608 if (this.opts
.enableTasksQueue
=== true && !enable
) {
609 this.flushTasksQueues()
611 this.opts
.enableTasksQueue
= enable
612 this.setTasksQueueOptions(tasksQueueOptions
as TasksQueueOptions
)
616 public setTasksQueueOptions (tasksQueueOptions
: TasksQueueOptions
): void {
617 if (this.opts
.enableTasksQueue
=== true) {
618 this.checkValidTasksQueueOptions(tasksQueueOptions
)
619 this.opts
.tasksQueueOptions
=
620 this.buildTasksQueueOptions(tasksQueueOptions
)
621 } else if (this.opts
.tasksQueueOptions
!= null) {
622 delete this.opts
.tasksQueueOptions
626 private buildTasksQueueOptions (
627 tasksQueueOptions
: TasksQueueOptions
628 ): TasksQueueOptions
{
630 concurrency
: tasksQueueOptions
?.concurrency
?? 1
635 * Whether the pool is full or not.
637 * The pool filling boolean status.
639 protected get
full (): boolean {
640 return this.workerNodes
.length
>= this.maxSize
644 * Whether the pool is busy or not.
646 * The pool busyness boolean status.
648 protected abstract get
busy (): boolean
651 * Whether worker nodes are executing concurrently their tasks quota or not.
653 * @returns Worker nodes busyness boolean status.
655 protected internalBusy (): boolean {
656 if (this.opts
.enableTasksQueue
=== true) {
658 this.workerNodes
.findIndex(
660 workerNode
.info
.ready
&&
661 workerNode
.usage
.tasks
.executing
<
662 (this.opts
.tasksQueueOptions
?.concurrency
as number)
667 this.workerNodes
.findIndex(
669 workerNode
.info
.ready
&& workerNode
.usage
.tasks
.executing
=== 0
676 public listTaskFunctions (): string[] {
677 for (const workerNode
of this.workerNodes
) {
679 Array.isArray(workerNode
.info
.taskFunctions
) &&
680 workerNode
.info
.taskFunctions
.length
> 0
682 return workerNode
.info
.taskFunctions
689 public async execute (
692 transferList
?: TransferListItem
[]
693 ): Promise
<Response
> {
694 return await new Promise
<Response
>((resolve
, reject
) => {
695 if (name
!= null && typeof name
!== 'string') {
696 reject(new TypeError('name argument must be a string'))
700 typeof name
=== 'string' &&
701 name
.trim().length
=== 0
703 reject(new TypeError('name argument must not be an empty string'))
705 if (transferList
!= null && !Array.isArray(transferList
)) {
706 reject(new TypeError('transferList argument must be an array'))
708 const timestamp
= performance
.now()
709 const workerNodeKey
= this.chooseWorkerNode()
710 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
713 Array.isArray(workerInfo
.taskFunctions
) &&
714 !workerInfo
.taskFunctions
.includes(name
)
717 new Error(`Task function '${name}' is not registered in the pool`)
720 const task
: Task
<Data
> = {
721 name
: name
?? DEFAULT_TASK_NAME
,
722 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
723 data
: data
?? ({} as Data
),
726 workerId
: workerInfo
.id
as number,
729 this.promiseResponseMap
.set(task
.taskId
as string, {
735 this.opts
.enableTasksQueue
=== false ||
736 (this.opts
.enableTasksQueue
=== true &&
737 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
<
738 (this.opts
.tasksQueueOptions
?.concurrency
as number))
740 this.executeTask(workerNodeKey
, task
)
742 this.enqueueTask(workerNodeKey
, task
)
748 public async destroy (): Promise
<void> {
750 this.workerNodes
.map(async (_
, workerNodeKey
) => {
751 await this.destroyWorkerNode(workerNodeKey
)
754 this.emitter
?.emit(PoolEvents
.destroy
, this.info
)
757 protected async sendKillMessageToWorker (
758 workerNodeKey
: number,
761 await new Promise
<void>((resolve
, reject
) => {
762 this.registerWorkerMessageListener(workerNodeKey
, (message
) => {
763 if (message
.kill
=== 'success') {
765 } else if (message
.kill
=== 'failure') {
766 reject(new Error(`Worker ${workerId} kill message handling failed`))
769 this.sendToWorker(workerNodeKey
, { kill
: true, workerId
})
774 * Terminates the worker node given its worker node key.
776 * @param workerNodeKey - The worker node key.
778 protected abstract destroyWorkerNode (workerNodeKey
: number): Promise
<void>
781 * Setup hook to execute code before worker nodes are created in the abstract constructor.
786 protected setupHook (): void {
787 // Intentionally empty
791 * Should return whether the worker is the main worker or not.
793 protected abstract isMain (): boolean
796 * Hook executed before the worker task execution.
799 * @param workerNodeKey - The worker node key.
800 * @param task - The task to execute.
802 protected beforeTaskExecutionHook (
803 workerNodeKey
: number,
806 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
807 ++workerUsage
.tasks
.executing
808 this.updateWaitTimeWorkerUsage(workerUsage
, task
)
809 if (this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
)) {
810 const taskFunctionWorkerUsage
= this.workerNodes
[
812 ].getTaskFunctionWorkerUsage(task
.name
as string) as WorkerUsage
813 ++taskFunctionWorkerUsage
.tasks
.executing
814 this.updateWaitTimeWorkerUsage(taskFunctionWorkerUsage
, task
)
819 * Hook executed after the worker task execution.
822 * @param workerNodeKey - The worker node key.
823 * @param message - The received message.
825 protected afterTaskExecutionHook (
826 workerNodeKey
: number,
827 message
: MessageValue
<Response
>
829 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
830 this.updateTaskStatisticsWorkerUsage(workerUsage
, message
)
831 this.updateRunTimeWorkerUsage(workerUsage
, message
)
832 this.updateEluWorkerUsage(workerUsage
, message
)
833 if (this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
)) {
834 const taskFunctionWorkerUsage
= this.workerNodes
[
836 ].getTaskFunctionWorkerUsage(
837 message
.taskPerformance
?.name
?? DEFAULT_TASK_NAME
839 this.updateTaskStatisticsWorkerUsage(taskFunctionWorkerUsage
, message
)
840 this.updateRunTimeWorkerUsage(taskFunctionWorkerUsage
, message
)
841 this.updateEluWorkerUsage(taskFunctionWorkerUsage
, message
)
846 * Whether the worker node shall update its task function worker usage or not.
848 * @param workerNodeKey - The worker node key.
849 * @returns `true` if the worker node shall update its task function worker usage, `false` otherwise.
851 private shallUpdateTaskFunctionWorkerUsage (workerNodeKey
: number): boolean {
852 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
854 Array.isArray(workerInfo
.taskFunctions
) &&
855 workerInfo
.taskFunctions
.length
> 2
859 private updateTaskStatisticsWorkerUsage (
860 workerUsage
: WorkerUsage
,
861 message
: MessageValue
<Response
>
863 const workerTaskStatistics
= workerUsage
.tasks
865 workerTaskStatistics
.executing
!= null &&
866 workerTaskStatistics
.executing
> 0
868 --workerTaskStatistics
.executing
870 workerTaskStatistics
.executing
!= null &&
871 workerTaskStatistics
.executing
< 0
874 'Worker usage statistic for tasks executing cannot be negative'
877 if (message
.taskError
== null) {
878 ++workerTaskStatistics
.executed
880 ++workerTaskStatistics
.failed
884 private updateRunTimeWorkerUsage (
885 workerUsage
: WorkerUsage
,
886 message
: MessageValue
<Response
>
888 updateMeasurementStatistics(
890 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
,
891 message
.taskPerformance
?.runTime
?? 0,
892 workerUsage
.tasks
.executed
896 private updateWaitTimeWorkerUsage (
897 workerUsage
: WorkerUsage
,
900 const timestamp
= performance
.now()
901 const taskWaitTime
= timestamp
- (task
.timestamp
?? timestamp
)
902 updateMeasurementStatistics(
903 workerUsage
.waitTime
,
904 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().waitTime
,
906 workerUsage
.tasks
.executed
910 private updateEluWorkerUsage (
911 workerUsage
: WorkerUsage
,
912 message
: MessageValue
<Response
>
914 const eluTaskStatisticsRequirements
: MeasurementStatisticsRequirements
=
915 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
916 updateMeasurementStatistics(
917 workerUsage
.elu
.active
,
918 eluTaskStatisticsRequirements
,
919 message
.taskPerformance
?.elu
?.active
?? 0,
920 workerUsage
.tasks
.executed
922 updateMeasurementStatistics(
923 workerUsage
.elu
.idle
,
924 eluTaskStatisticsRequirements
,
925 message
.taskPerformance
?.elu
?.idle
?? 0,
926 workerUsage
.tasks
.executed
928 if (eluTaskStatisticsRequirements
.aggregate
) {
929 if (message
.taskPerformance
?.elu
!= null) {
930 if (workerUsage
.elu
.utilization
!= null) {
931 workerUsage
.elu
.utilization
=
932 (workerUsage
.elu
.utilization
+
933 message
.taskPerformance
.elu
.utilization
) /
936 workerUsage
.elu
.utilization
= message
.taskPerformance
.elu
.utilization
943 * Chooses a worker node for the next task.
945 * The default worker choice strategy uses a round robin algorithm to distribute the tasks.
947 * @returns The chosen worker node key
949 private chooseWorkerNode (): number {
950 if (this.shallCreateDynamicWorker()) {
951 const workerNodeKey
= this.createAndSetupDynamicWorkerNode()
953 this.workerChoiceStrategyContext
.getStrategyPolicy().useDynamicWorker
958 return this.workerChoiceStrategyContext
.execute()
962 * Conditions for dynamic worker creation.
964 * @returns Whether to create a dynamic worker or not.
966 private shallCreateDynamicWorker (): boolean {
967 return this.type === PoolTypes
.dynamic
&& !this.full
&& this.internalBusy()
971 * Sends a message to worker given its worker node key.
973 * @param workerNodeKey - The worker node key.
974 * @param message - The message.
975 * @param transferList - The optional array of transferable objects.
977 protected abstract sendToWorker (
978 workerNodeKey
: number,
979 message
: MessageValue
<Data
>,
980 transferList
?: TransferListItem
[]
984 * Creates a new worker.
986 * @returns Newly created worker.
988 protected abstract createWorker (): Worker
991 * Creates a new, completely set up worker node.
993 * @returns New, completely set up worker node key.
995 protected createAndSetupWorkerNode (): number {
996 const worker
= this.createWorker()
998 worker
.on('online', this.opts
.onlineHandler
?? EMPTY_FUNCTION
)
999 worker
.on('message', this.opts
.messageHandler
?? EMPTY_FUNCTION
)
1000 worker
.on('error', this.opts
.errorHandler
?? EMPTY_FUNCTION
)
1001 worker
.on('error', (error
) => {
1002 const workerNodeKey
= this.getWorkerNodeKeyByWorker(worker
)
1003 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1004 workerInfo
.ready
= false
1005 this.workerNodes
[workerNodeKey
].closeChannel()
1006 this.emitter
?.emit(PoolEvents
.error
, error
)
1007 if (this.opts
.restartWorkerOnError
=== true && !this.starting
) {
1008 if (workerInfo
.dynamic
) {
1009 this.createAndSetupDynamicWorkerNode()
1011 this.createAndSetupWorkerNode()
1014 if (this.opts
.enableTasksQueue
=== true) {
1015 this.redistributeQueuedTasks(workerNodeKey
)
1018 worker
.on('exit', this.opts
.exitHandler
?? EMPTY_FUNCTION
)
1019 worker
.once('exit', () => {
1020 this.removeWorkerNode(worker
)
1023 const workerNodeKey
= this.addWorkerNode(worker
)
1025 this.afterWorkerNodeSetup(workerNodeKey
)
1027 return workerNodeKey
1031 * Creates a new, completely set up dynamic worker node.
1033 * @returns New, completely set up dynamic worker node key.
1035 protected createAndSetupDynamicWorkerNode (): number {
1036 const workerNodeKey
= this.createAndSetupWorkerNode()
1037 this.registerWorkerMessageListener(workerNodeKey
, (message
) => {
1038 const localWorkerNodeKey
= this.getWorkerNodeKeyByWorkerId(
1041 const workerUsage
= this.workerNodes
[localWorkerNodeKey
].usage
1042 // Kill message received from worker
1044 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
1045 (isKillBehavior(KillBehaviors
.SOFT
, message
.kill
) &&
1046 ((this.opts
.enableTasksQueue
=== false &&
1047 workerUsage
.tasks
.executing
=== 0) ||
1048 (this.opts
.enableTasksQueue
=== true &&
1049 workerUsage
.tasks
.executing
=== 0 &&
1050 this.tasksQueueSize(localWorkerNodeKey
) === 0)))
1052 this.destroyWorkerNode(localWorkerNodeKey
).catch((error
) => {
1053 this.emitter
?.emit(PoolEvents
.error
, error
)
1057 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1058 this.sendToWorker(workerNodeKey
, {
1060 workerId
: workerInfo
.id
as number
1062 workerInfo
.dynamic
= true
1063 if (this.workerChoiceStrategyContext
.getStrategyPolicy().useDynamicWorker
) {
1064 workerInfo
.ready
= true
1066 this.checkAndEmitDynamicWorkerCreationEvents()
1067 return workerNodeKey
1071 * Registers a listener callback on the worker given its worker node key.
1073 * @param workerNodeKey - The worker node key.
1074 * @param listener - The message listener callback.
1076 protected abstract registerWorkerMessageListener
<
1077 Message
extends Data
| Response
1079 workerNodeKey
: number,
1080 listener
: (message
: MessageValue
<Message
>) => void
1084 * Method hooked up after a worker node has been newly created.
1085 * Can be overridden.
1087 * @param workerNodeKey - The newly created worker node key.
1089 protected afterWorkerNodeSetup (workerNodeKey
: number): void {
1090 // Listen to worker messages.
1091 this.registerWorkerMessageListener(workerNodeKey
, this.workerListener())
1092 // Send the startup message to worker.
1093 this.sendStartupMessageToWorker(workerNodeKey
)
1094 // Send the statistics message to worker.
1095 this.sendStatisticsMessageToWorker(workerNodeKey
)
1099 * Sends the startup message to worker given its worker node key.
1101 * @param workerNodeKey - The worker node key.
1103 protected abstract sendStartupMessageToWorker (workerNodeKey
: number): void
1106 * Sends the statistics message to worker given its worker node key.
1108 * @param workerNodeKey - The worker node key.
1110 private sendStatisticsMessageToWorker (workerNodeKey
: number): void {
1111 this.sendToWorker(workerNodeKey
, {
1114 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1116 elu
: this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1119 workerId
: this.getWorkerInfo(workerNodeKey
).id
as number
1123 private redistributeQueuedTasks (workerNodeKey
: number): void {
1124 while (this.tasksQueueSize(workerNodeKey
) > 0) {
1125 let targetWorkerNodeKey
: number = workerNodeKey
1126 let minQueuedTasks
= Infinity
1127 let executeTask
= false
1128 for (const [workerNodeId
, workerNode
] of this.workerNodes
.entries()) {
1129 const workerInfo
= this.getWorkerInfo(workerNodeId
)
1131 workerNodeId
!== workerNodeKey
&&
1133 workerNode
.usage
.tasks
.queued
=== 0
1136 this.workerNodes
[workerNodeId
].usage
.tasks
.executing
<
1137 (this.opts
.tasksQueueOptions
?.concurrency
as number)
1141 targetWorkerNodeKey
= workerNodeId
1145 workerNodeId
!== workerNodeKey
&&
1147 workerNode
.usage
.tasks
.queued
< minQueuedTasks
1149 minQueuedTasks
= workerNode
.usage
.tasks
.queued
1150 targetWorkerNodeKey
= workerNodeId
1155 targetWorkerNodeKey
,
1156 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1160 targetWorkerNodeKey
,
1161 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1168 * This method is the listener registered for each worker message.
1170 * @returns The listener function to execute when a message is received from a worker.
1172 protected workerListener (): (message
: MessageValue
<Response
>) => void {
1173 return (message
) => {
1174 this.checkMessageWorkerId(message
)
1175 if (message
.ready
!= null && message
.taskFunctions
!= null) {
1176 // Worker ready response received from worker
1177 this.handleWorkerReadyResponse(message
)
1178 } else if (message
.taskId
!= null) {
1179 // Task execution response received from worker
1180 this.handleTaskExecutionResponse(message
)
1181 } else if (message
.taskFunctions
!= null) {
1182 // Task functions message received from worker
1184 this.getWorkerNodeKeyByWorkerId(message
.workerId
)
1185 ).taskFunctions
= message
.taskFunctions
1190 private handleWorkerReadyResponse (message
: MessageValue
<Response
>): void {
1191 if (message
.ready
=== false) {
1192 throw new Error(`Worker ${message.workerId} failed to initialize`)
1194 const workerInfo
= this.getWorkerInfo(
1195 this.getWorkerNodeKeyByWorkerId(message
.workerId
)
1197 workerInfo
.ready
= message
.ready
as boolean
1198 workerInfo
.taskFunctions
= message
.taskFunctions
1199 if (this.emitter
!= null && this.ready
) {
1200 this.emitter
.emit(PoolEvents
.ready
, this.info
)
1204 private handleTaskExecutionResponse (message
: MessageValue
<Response
>): void {
1205 const { taskId
, taskError
, data
} = message
1206 const promiseResponse
= this.promiseResponseMap
.get(taskId
as string)
1207 if (promiseResponse
!= null) {
1208 if (taskError
!= null) {
1209 this.emitter
?.emit(PoolEvents
.taskError
, taskError
)
1210 promiseResponse
.reject(taskError
.message
)
1212 promiseResponse
.resolve(data
as Response
)
1214 const workerNodeKey
= promiseResponse
.workerNodeKey
1215 this.afterTaskExecutionHook(workerNodeKey
, message
)
1216 this.promiseResponseMap
.delete(taskId
as string)
1218 this.opts
.enableTasksQueue
=== true &&
1219 this.tasksQueueSize(workerNodeKey
) > 0 &&
1220 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
<
1221 (this.opts
.tasksQueueOptions
?.concurrency
as number)
1225 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1228 this.workerChoiceStrategyContext
.update(workerNodeKey
)
1232 private checkAndEmitTaskExecutionEvents (): void {
1234 this.emitter
?.emit(PoolEvents
.busy
, this.info
)
1238 private checkAndEmitTaskQueuingEvents (): void {
1239 if (this.hasBackPressure()) {
1240 this.emitter
?.emit(PoolEvents
.backPressure
, this.info
)
1244 private checkAndEmitDynamicWorkerCreationEvents (): void {
1245 if (this.type === PoolTypes
.dynamic
) {
1247 this.emitter
?.emit(PoolEvents
.full
, this.info
)
1253 * Gets the worker information given its worker node key.
1255 * @param workerNodeKey - The worker node key.
1256 * @returns The worker information.
1258 protected getWorkerInfo (workerNodeKey
: number): WorkerInfo
{
1259 return this.workerNodes
[workerNodeKey
].info
1263 * Adds the given worker in the pool worker nodes.
1265 * @param worker - The worker.
1266 * @returns The added worker node key.
1267 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the added worker node is not found.
1269 private addWorkerNode (worker
: Worker
): number {
1270 const workerNode
= new WorkerNode
<Worker
, Data
>(
1275 // Flag the worker node as ready at pool startup.
1276 if (this.starting
) {
1277 workerNode
.info
.ready
= true
1279 this.workerNodes
.push(workerNode
)
1280 const workerNodeKey
= this.getWorkerNodeKeyByWorker(worker
)
1281 if (workerNodeKey
=== -1) {
1282 throw new Error('Worker node added not found')
1284 return workerNodeKey
1288 * Removes the given worker from the pool worker nodes.
1290 * @param worker - The worker.
1292 private removeWorkerNode (worker
: Worker
): void {
1293 const workerNodeKey
= this.getWorkerNodeKeyByWorker(worker
)
1294 if (workerNodeKey
!== -1) {
1295 this.workerNodes
.splice(workerNodeKey
, 1)
1296 this.workerChoiceStrategyContext
.remove(workerNodeKey
)
1301 public hasWorkerNodeBackPressure (workerNodeKey
: number): boolean {
1303 this.opts
.enableTasksQueue
=== true &&
1304 this.workerNodes
[workerNodeKey
].hasBackPressure()
1308 private hasBackPressure (): boolean {
1310 this.opts
.enableTasksQueue
=== true &&
1311 this.workerNodes
.findIndex(
1312 (workerNode
) => !workerNode
.hasBackPressure()
1318 * Executes the given task on the worker given its worker node key.
1320 * @param workerNodeKey - The worker node key.
1321 * @param task - The task to execute.
1323 private executeTask (workerNodeKey
: number, task
: Task
<Data
>): void {
1324 this.beforeTaskExecutionHook(workerNodeKey
, task
)
1325 this.sendToWorker(workerNodeKey
, task
, task
.transferList
)
1326 this.checkAndEmitTaskExecutionEvents()
1329 private enqueueTask (workerNodeKey
: number, task
: Task
<Data
>): number {
1330 const tasksQueueSize
= this.workerNodes
[workerNodeKey
].enqueueTask(task
)
1331 this.checkAndEmitTaskQueuingEvents()
1332 return tasksQueueSize
1335 private dequeueTask (workerNodeKey
: number): Task
<Data
> | undefined {
1336 return this.workerNodes
[workerNodeKey
].dequeueTask()
1339 private tasksQueueSize (workerNodeKey
: number): number {
1340 return this.workerNodes
[workerNodeKey
].tasksQueueSize()
1343 protected flushTasksQueue (workerNodeKey
: number): void {
1344 while (this.tasksQueueSize(workerNodeKey
) > 0) {
1347 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1350 this.workerNodes
[workerNodeKey
].clearTasksQueue()
1353 private flushTasksQueues (): void {
1354 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
1355 this.flushTasksQueue(workerNodeKey
)