1 import { randomUUID
} from
'node:crypto'
2 import { performance
} from
'node:perf_hooks'
3 import { existsSync
} from
'node:fs'
4 import { type TransferListItem
} from
'node:worker_threads'
7 PromiseResponseWrapper
,
9 } from
'../utility-types'
12 DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
,
18 updateMeasurementStatistics
20 import { KillBehaviors
} from
'../worker/worker-options'
29 type TasksQueueOptions
39 type MeasurementStatisticsRequirements
,
41 WorkerChoiceStrategies
,
42 type WorkerChoiceStrategy
,
43 type WorkerChoiceStrategyOptions
44 } from
'./selection-strategies/selection-strategies-types'
45 import { WorkerChoiceStrategyContext
} from
'./selection-strategies/worker-choice-strategy-context'
46 import { version
} from
'./version'
47 import { WorkerNode
} from
'./worker-node'
50 * Base class that implements some shared logic for all poolifier pools.
52 * @typeParam Worker - Type of worker which manages this pool.
53 * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
54 * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
56 export abstract class AbstractPool
<
57 Worker
extends IWorker
,
60 > implements IPool
<Worker
, Data
, Response
> {
62 public readonly workerNodes
: Array<IWorkerNode
<Worker
, Data
>> = []
65 public readonly emitter
?: PoolEmitter
68 * The task execution response promise map.
70 * - `key`: The message id of each submitted task.
71 * - `value`: An object that contains the worker, the execution response promise resolve and reject callbacks.
73 * When we receive a message from the worker, we get a map entry with the promise resolve/reject bound to the message id.
75 protected promiseResponseMap
: Map
<string, PromiseResponseWrapper
<Response
>> =
76 new Map
<string, PromiseResponseWrapper
<Response
>>()
79 * Worker choice strategy context referencing a worker choice algorithm implementation.
81 protected workerChoiceStrategyContext
: WorkerChoiceStrategyContext
<
88 * Dynamic pool maximum size property placeholder.
90 protected readonly max
?: number
93 * Whether the pool is starting or not.
95 private readonly starting
: boolean
97 * The start timestamp of the pool.
99 private readonly startTimestamp
102 * Constructs a new poolifier pool.
104 * @param numberOfWorkers - Number of workers that this pool should manage.
105 * @param filePath - Path to the worker file.
106 * @param opts - Options for the pool.
109 protected readonly numberOfWorkers
: number,
110 protected readonly filePath
: string,
111 protected readonly opts
: PoolOptions
<Worker
>
113 if (!this.isMain()) {
115 'Cannot start a pool from a worker with the same type as the pool'
118 this.checkNumberOfWorkers(this.numberOfWorkers
)
119 this.checkFilePath(this.filePath
)
120 this.checkPoolOptions(this.opts
)
122 this.chooseWorkerNode
= this.chooseWorkerNode
.bind(this)
123 this.executeTask
= this.executeTask
.bind(this)
124 this.enqueueTask
= this.enqueueTask
.bind(this)
126 if (this.opts
.enableEvents
=== true) {
127 this.emitter
= new PoolEmitter()
129 this.workerChoiceStrategyContext
= new WorkerChoiceStrategyContext
<
135 this.opts
.workerChoiceStrategy
,
136 this.opts
.workerChoiceStrategyOptions
143 this.starting
= false
145 this.startTimestamp
= performance
.now()
148 private checkFilePath (filePath
: string): void {
151 typeof filePath
!== 'string' ||
152 (typeof filePath
=== 'string' && filePath
.trim().length
=== 0)
154 throw new Error('Please specify a file with a worker implementation')
156 if (!existsSync(filePath
)) {
157 throw new Error(`Cannot find the worker file '${filePath}'`)
161 private checkNumberOfWorkers (numberOfWorkers
: number): void {
162 if (numberOfWorkers
== null) {
164 'Cannot instantiate a pool without specifying the number of workers'
166 } else if (!Number.isSafeInteger(numberOfWorkers
)) {
168 'Cannot instantiate a pool with a non safe integer number of workers'
170 } else if (numberOfWorkers
< 0) {
171 throw new RangeError(
172 'Cannot instantiate a pool with a negative number of workers'
174 } else if (this.type === PoolTypes
.fixed
&& numberOfWorkers
=== 0) {
175 throw new RangeError('Cannot instantiate a fixed pool with zero worker')
179 protected checkDynamicPoolSize (min
: number, max
: number): void {
180 if (this.type === PoolTypes
.dynamic
) {
183 'Cannot instantiate a dynamic pool without specifying the maximum pool size'
185 } else if (!Number.isSafeInteger(max
)) {
187 'Cannot instantiate a dynamic pool with a non safe integer maximum pool size'
189 } else if (min
> max
) {
190 throw new RangeError(
191 'Cannot instantiate a dynamic pool with a maximum pool size inferior to the minimum pool size'
193 } else if (max
=== 0) {
194 throw new RangeError(
195 'Cannot instantiate a dynamic pool with a maximum pool size equal to zero'
197 } else if (min
=== max
) {
198 throw new RangeError(
199 'Cannot instantiate a dynamic pool with a minimum pool size equal to the maximum pool size. Use a fixed pool instead'
205 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
206 if (isPlainObject(opts
)) {
207 this.opts
.workerChoiceStrategy
=
208 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
209 this.checkValidWorkerChoiceStrategy(this.opts
.workerChoiceStrategy
)
210 this.opts
.workerChoiceStrategyOptions
= {
211 ...DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
,
212 ...opts
.workerChoiceStrategyOptions
214 this.checkValidWorkerChoiceStrategyOptions(
215 this.opts
.workerChoiceStrategyOptions
217 this.opts
.restartWorkerOnError
= opts
.restartWorkerOnError
?? true
218 this.opts
.enableEvents
= opts
.enableEvents
?? true
219 this.opts
.enableTasksQueue
= opts
.enableTasksQueue
?? false
220 if (this.opts
.enableTasksQueue
) {
221 this.checkValidTasksQueueOptions(
222 opts
.tasksQueueOptions
as TasksQueueOptions
224 this.opts
.tasksQueueOptions
= this.buildTasksQueueOptions(
225 opts
.tasksQueueOptions
as TasksQueueOptions
229 throw new TypeError('Invalid pool options: must be a plain object')
233 private checkValidWorkerChoiceStrategy (
234 workerChoiceStrategy
: WorkerChoiceStrategy
236 if (!Object.values(WorkerChoiceStrategies
).includes(workerChoiceStrategy
)) {
238 `Invalid worker choice strategy '${workerChoiceStrategy}'`
243 private checkValidWorkerChoiceStrategyOptions (
244 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
246 if (!isPlainObject(workerChoiceStrategyOptions
)) {
248 'Invalid worker choice strategy options: must be a plain object'
252 workerChoiceStrategyOptions
.choiceRetries
!= null &&
253 !Number.isSafeInteger(workerChoiceStrategyOptions
.choiceRetries
)
256 'Invalid worker choice strategy options: choice retries must be an integer'
260 workerChoiceStrategyOptions
.choiceRetries
!= null &&
261 workerChoiceStrategyOptions
.choiceRetries
<= 0
263 throw new RangeError(
264 `Invalid worker choice strategy options: choice retries '${workerChoiceStrategyOptions.choiceRetries}' must be greater than zero`
268 workerChoiceStrategyOptions
.weights
!= null &&
269 Object.keys(workerChoiceStrategyOptions
.weights
).length
!== this.maxSize
272 'Invalid worker choice strategy options: must have a weight for each worker node'
276 workerChoiceStrategyOptions
.measurement
!= null &&
277 !Object.values(Measurements
).includes(
278 workerChoiceStrategyOptions
.measurement
282 `Invalid worker choice strategy options: invalid measurement '${workerChoiceStrategyOptions.measurement}'`
287 private checkValidTasksQueueOptions (
288 tasksQueueOptions
: TasksQueueOptions
290 if (tasksQueueOptions
!= null && !isPlainObject(tasksQueueOptions
)) {
291 throw new TypeError('Invalid tasks queue options: must be a plain object')
294 tasksQueueOptions
?.concurrency
!= null &&
295 !Number.isSafeInteger(tasksQueueOptions
.concurrency
)
298 'Invalid worker node tasks concurrency: must be an integer'
302 tasksQueueOptions
?.concurrency
!= null &&
303 tasksQueueOptions
.concurrency
<= 0
305 throw new RangeError(
306 `Invalid worker node tasks concurrency: ${tasksQueueOptions.concurrency} is a negative integer or zero`
310 tasksQueueOptions
?.queueMaxSize
!= null &&
311 !Number.isSafeInteger(tasksQueueOptions
.queueMaxSize
)
314 'Invalid worker node tasks queue max size: must be an integer'
318 tasksQueueOptions
?.queueMaxSize
!= null &&
319 tasksQueueOptions
.queueMaxSize
<= 0
321 throw new RangeError(
322 `Invalid worker node tasks queue max size: ${tasksQueueOptions.queueMaxSize} is a negative integer or zero`
327 private startPool (): void {
329 this.workerNodes
.reduce(
330 (accumulator
, workerNode
) =>
331 !workerNode
.info
.dynamic
? accumulator
+ 1 : accumulator
,
333 ) < this.numberOfWorkers
335 this.createAndSetupWorkerNode()
340 public get
info (): PoolInfo
{
346 strategy
: this.opts
.workerChoiceStrategy
as WorkerChoiceStrategy
,
347 minSize
: this.minSize
,
348 maxSize
: this.maxSize
,
349 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
350 .runTime
.aggregate
&&
351 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
352 .waitTime
.aggregate
&& { utilization
: round(this.utilization
) }),
353 workerNodes
: this.workerNodes
.length
,
354 idleWorkerNodes
: this.workerNodes
.reduce(
355 (accumulator
, workerNode
) =>
356 workerNode
.usage
.tasks
.executing
=== 0
361 busyWorkerNodes
: this.workerNodes
.reduce(
362 (accumulator
, workerNode
) =>
363 workerNode
.usage
.tasks
.executing
> 0 ? accumulator
+ 1 : accumulator
,
366 executedTasks
: this.workerNodes
.reduce(
367 (accumulator
, workerNode
) =>
368 accumulator
+ workerNode
.usage
.tasks
.executed
,
371 executingTasks
: this.workerNodes
.reduce(
372 (accumulator
, workerNode
) =>
373 accumulator
+ workerNode
.usage
.tasks
.executing
,
376 ...(this.opts
.enableTasksQueue
=== true && {
377 queuedTasks
: this.workerNodes
.reduce(
378 (accumulator
, workerNode
) =>
379 accumulator
+ workerNode
.usage
.tasks
.queued
,
383 ...(this.opts
.enableTasksQueue
=== true && {
384 maxQueuedTasks
: this.workerNodes
.reduce(
385 (accumulator
, workerNode
) =>
386 accumulator
+ (workerNode
.usage
.tasks
?.maxQueued
?? 0),
390 ...(this.opts
.enableTasksQueue
=== true && {
391 backPressure
: this.hasBackPressure()
393 failedTasks
: this.workerNodes
.reduce(
394 (accumulator
, workerNode
) =>
395 accumulator
+ workerNode
.usage
.tasks
.failed
,
398 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
399 .runTime
.aggregate
&& {
403 ...this.workerNodes
.map(
404 (workerNode
) => workerNode
.usage
.runTime
?.minimum
?? Infinity
410 ...this.workerNodes
.map(
411 (workerNode
) => workerNode
.usage
.runTime
?.maximum
?? -Infinity
416 this.workerNodes
.reduce(
417 (accumulator
, workerNode
) =>
418 accumulator
+ (workerNode
.usage
.runTime
?.aggregate
?? 0),
421 this.workerNodes
.reduce(
422 (accumulator
, workerNode
) =>
423 accumulator
+ (workerNode
.usage
.tasks
?.executed
?? 0),
427 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
431 this.workerNodes
.map(
432 (workerNode
) => workerNode
.usage
.runTime
?.median
?? 0
439 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
440 .waitTime
.aggregate
&& {
444 ...this.workerNodes
.map(
445 (workerNode
) => workerNode
.usage
.waitTime
?.minimum
?? Infinity
451 ...this.workerNodes
.map(
452 (workerNode
) => workerNode
.usage
.waitTime
?.maximum
?? -Infinity
457 this.workerNodes
.reduce(
458 (accumulator
, workerNode
) =>
459 accumulator
+ (workerNode
.usage
.waitTime
?.aggregate
?? 0),
462 this.workerNodes
.reduce(
463 (accumulator
, workerNode
) =>
464 accumulator
+ (workerNode
.usage
.tasks
?.executed
?? 0),
468 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
469 .waitTime
.median
&& {
472 this.workerNodes
.map(
473 (workerNode
) => workerNode
.usage
.waitTime
?.median
?? 0
484 * The pool readiness boolean status.
486 private get
ready (): boolean {
488 this.workerNodes
.reduce(
489 (accumulator
, workerNode
) =>
490 !workerNode
.info
.dynamic
&& workerNode
.info
.ready
499 * The approximate pool utilization.
501 * @returns The pool utilization.
503 private get
utilization (): number {
504 const poolTimeCapacity
=
505 (performance
.now() - this.startTimestamp
) * this.maxSize
506 const totalTasksRunTime
= this.workerNodes
.reduce(
507 (accumulator
, workerNode
) =>
508 accumulator
+ (workerNode
.usage
.runTime
?.aggregate
?? 0),
511 const totalTasksWaitTime
= this.workerNodes
.reduce(
512 (accumulator
, workerNode
) =>
513 accumulator
+ (workerNode
.usage
.waitTime
?.aggregate
?? 0),
516 return (totalTasksRunTime
+ totalTasksWaitTime
) / poolTimeCapacity
522 * If it is `'dynamic'`, it provides the `max` property.
524 protected abstract get
type (): PoolType
529 protected abstract get
worker (): WorkerType
532 * The pool minimum size.
534 protected get
minSize (): number {
535 return this.numberOfWorkers
539 * The pool maximum size.
541 protected get
maxSize (): number {
542 return this.max
?? this.numberOfWorkers
546 * Checks if the worker id sent in the received message from a worker is valid.
548 * @param message - The received message.
549 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the worker id is invalid.
551 private checkMessageWorkerId (message
: MessageValue
<Response
>): void {
552 if (message
.workerId
== null) {
553 throw new Error('Worker message received without worker id')
555 message
.workerId
!= null &&
556 this.getWorkerNodeKeyByWorkerId(message
.workerId
) === -1
559 `Worker message received from unknown worker '${message.workerId}'`
565 * Gets the given worker its worker node key.
567 * @param worker - The worker.
568 * @returns The worker node key if found in the pool worker nodes, `-1` otherwise.
570 private getWorkerNodeKeyByWorker (worker
: Worker
): number {
571 return this.workerNodes
.findIndex(
572 (workerNode
) => workerNode
.worker
=== worker
577 * Gets the worker node key given its worker id.
579 * @param workerId - The worker id.
580 * @returns The worker node key if the worker id is found in the pool worker nodes, `-1` otherwise.
582 private getWorkerNodeKeyByWorkerId (workerId
: number): number {
583 return this.workerNodes
.findIndex(
584 (workerNode
) => workerNode
.info
.id
=== workerId
589 public setWorkerChoiceStrategy (
590 workerChoiceStrategy
: WorkerChoiceStrategy
,
591 workerChoiceStrategyOptions
?: WorkerChoiceStrategyOptions
593 this.checkValidWorkerChoiceStrategy(workerChoiceStrategy
)
594 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
595 this.workerChoiceStrategyContext
.setWorkerChoiceStrategy(
596 this.opts
.workerChoiceStrategy
598 if (workerChoiceStrategyOptions
!= null) {
599 this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
601 for (const [workerNodeKey
, workerNode
] of this.workerNodes
.entries()) {
602 workerNode
.resetUsage()
603 this.sendStatisticsMessageToWorker(workerNodeKey
)
608 public setWorkerChoiceStrategyOptions (
609 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
611 this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
612 this.opts
.workerChoiceStrategyOptions
= {
613 ...DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
,
614 ...workerChoiceStrategyOptions
616 this.workerChoiceStrategyContext
.setOptions(
617 this.opts
.workerChoiceStrategyOptions
622 public enableTasksQueue (
624 tasksQueueOptions
?: TasksQueueOptions
626 if (this.opts
.enableTasksQueue
=== true && !enable
) {
627 this.flushTasksQueues()
629 this.opts
.enableTasksQueue
= enable
630 this.setTasksQueueOptions(tasksQueueOptions
as TasksQueueOptions
)
634 public setTasksQueueOptions (tasksQueueOptions
: TasksQueueOptions
): void {
635 if (this.opts
.enableTasksQueue
=== true) {
636 this.checkValidTasksQueueOptions(tasksQueueOptions
)
637 this.opts
.tasksQueueOptions
=
638 this.buildTasksQueueOptions(tasksQueueOptions
)
639 this.setTasksQueueMaxSize(
640 this.opts
.tasksQueueOptions
.queueMaxSize
as number
642 } else if (this.opts
.tasksQueueOptions
!= null) {
643 delete this.opts
.tasksQueueOptions
647 private setTasksQueueMaxSize (queueMaxSize
: number): void {
648 for (const workerNode
of this.workerNodes
) {
649 workerNode
.tasksQueueBackPressureSize
= queueMaxSize
653 private buildTasksQueueOptions (
654 tasksQueueOptions
: TasksQueueOptions
655 ): TasksQueueOptions
{
658 queueMaxSize
: Math.pow(this.maxSize
, 2),
666 * Whether the pool is full or not.
668 * The pool filling boolean status.
670 protected get
full (): boolean {
671 return this.workerNodes
.length
>= this.maxSize
675 * Whether the pool is busy or not.
677 * The pool busyness boolean status.
679 protected abstract get
busy (): boolean
682 * Whether worker nodes are executing concurrently their tasks quota or not.
684 * @returns Worker nodes busyness boolean status.
686 protected internalBusy (): boolean {
687 if (this.opts
.enableTasksQueue
=== true) {
689 this.workerNodes
.findIndex(
691 workerNode
.info
.ready
&&
692 workerNode
.usage
.tasks
.executing
<
693 (this.opts
.tasksQueueOptions
?.concurrency
as number)
698 this.workerNodes
.findIndex(
700 workerNode
.info
.ready
&& workerNode
.usage
.tasks
.executing
=== 0
707 public listTaskFunctions (): string[] {
708 for (const workerNode
of this.workerNodes
) {
710 Array.isArray(workerNode
.info
.taskFunctions
) &&
711 workerNode
.info
.taskFunctions
.length
> 0
713 return workerNode
.info
.taskFunctions
720 public async execute (
723 transferList
?: TransferListItem
[]
724 ): Promise
<Response
> {
725 return await new Promise
<Response
>((resolve
, reject
) => {
726 if (name
!= null && typeof name
!== 'string') {
727 reject(new TypeError('name argument must be a string'))
731 typeof name
=== 'string' &&
732 name
.trim().length
=== 0
734 reject(new TypeError('name argument must not be an empty string'))
736 if (transferList
!= null && !Array.isArray(transferList
)) {
737 reject(new TypeError('transferList argument must be an array'))
739 const timestamp
= performance
.now()
740 const workerNodeKey
= this.chooseWorkerNode()
741 const workerInfo
= this.getWorkerInfo(workerNodeKey
) as WorkerInfo
744 Array.isArray(workerInfo
.taskFunctions
) &&
745 !workerInfo
.taskFunctions
.includes(name
)
748 new Error(`Task function '${name}' is not registered in the pool`)
751 const task
: Task
<Data
> = {
752 name
: name
?? DEFAULT_TASK_NAME
,
753 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
754 data
: data
?? ({} as Data
),
757 workerId
: workerInfo
.id
as number,
760 this.promiseResponseMap
.set(task
.taskId
as string, {
766 this.opts
.enableTasksQueue
=== false ||
767 (this.opts
.enableTasksQueue
=== true &&
768 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
<
769 (this.opts
.tasksQueueOptions
?.concurrency
as number))
771 this.executeTask(workerNodeKey
, task
)
773 this.enqueueTask(workerNodeKey
, task
)
779 public async destroy (): Promise
<void> {
781 this.workerNodes
.map(async (_
, workerNodeKey
) => {
782 await this.destroyWorkerNode(workerNodeKey
)
785 this.emitter
?.emit(PoolEvents
.destroy
, this.info
)
788 protected async sendKillMessageToWorker (
789 workerNodeKey
: number,
792 await new Promise
<void>((resolve
, reject
) => {
793 this.registerWorkerMessageListener(workerNodeKey
, (message
) => {
794 if (message
.kill
=== 'success') {
796 } else if (message
.kill
=== 'failure') {
797 reject(new Error(`Worker ${workerId} kill message handling failed`))
800 this.sendToWorker(workerNodeKey
, { kill
: true, workerId
})
805 * Terminates the worker node given its worker node key.
807 * @param workerNodeKey - The worker node key.
809 protected abstract destroyWorkerNode (workerNodeKey
: number): Promise
<void>
812 * Setup hook to execute code before worker nodes are created in the abstract constructor.
817 protected setupHook (): void {
818 // Intentionally empty
822 * Should return whether the worker is the main worker or not.
824 protected abstract isMain (): boolean
827 * Hook executed before the worker task execution.
830 * @param workerNodeKey - The worker node key.
831 * @param task - The task to execute.
833 protected beforeTaskExecutionHook (
834 workerNodeKey
: number,
837 if (this.workerNodes
[workerNodeKey
]?.usage
!= null) {
838 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
839 ++workerUsage
.tasks
.executing
840 this.updateWaitTimeWorkerUsage(workerUsage
, task
)
843 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
844 this.workerNodes
[workerNodeKey
].getTaskFunctionWorkerUsage(
848 const taskFunctionWorkerUsage
= this.workerNodes
[
850 ].getTaskFunctionWorkerUsage(task
.name
as string) as WorkerUsage
851 ++taskFunctionWorkerUsage
.tasks
.executing
852 this.updateWaitTimeWorkerUsage(taskFunctionWorkerUsage
, task
)
857 * Hook executed after the worker task execution.
860 * @param workerNodeKey - The worker node key.
861 * @param message - The received message.
863 protected afterTaskExecutionHook (
864 workerNodeKey
: number,
865 message
: MessageValue
<Response
>
867 if (this.workerNodes
[workerNodeKey
]?.usage
!= null) {
868 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
869 this.updateTaskStatisticsWorkerUsage(workerUsage
, message
)
870 this.updateRunTimeWorkerUsage(workerUsage
, message
)
871 this.updateEluWorkerUsage(workerUsage
, message
)
874 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
875 this.workerNodes
[workerNodeKey
].getTaskFunctionWorkerUsage(
876 message
.taskPerformance
?.name
as string
879 const taskFunctionWorkerUsage
= this.workerNodes
[
881 ].getTaskFunctionWorkerUsage(
882 message
.taskPerformance
?.name
as string
884 this.updateTaskStatisticsWorkerUsage(taskFunctionWorkerUsage
, message
)
885 this.updateRunTimeWorkerUsage(taskFunctionWorkerUsage
, message
)
886 this.updateEluWorkerUsage(taskFunctionWorkerUsage
, message
)
891 * Whether the worker node shall update its task function worker usage or not.
893 * @param workerNodeKey - The worker node key.
894 * @returns `true` if the worker node shall update its task function worker usage, `false` otherwise.
896 private shallUpdateTaskFunctionWorkerUsage (workerNodeKey
: number): boolean {
897 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
899 workerInfo
!= null &&
900 Array.isArray(workerInfo
.taskFunctions
) &&
901 workerInfo
.taskFunctions
.length
> 2
905 private updateTaskStatisticsWorkerUsage (
906 workerUsage
: WorkerUsage
,
907 message
: MessageValue
<Response
>
909 const workerTaskStatistics
= workerUsage
.tasks
911 workerTaskStatistics
.executing
!= null &&
912 workerTaskStatistics
.executing
> 0
914 --workerTaskStatistics
.executing
916 workerTaskStatistics
.executing
!= null &&
917 workerTaskStatistics
.executing
< 0
920 'Worker usage statistic for tasks executing cannot be negative'
923 if (message
.taskError
== null) {
924 ++workerTaskStatistics
.executed
926 ++workerTaskStatistics
.failed
930 private updateRunTimeWorkerUsage (
931 workerUsage
: WorkerUsage
,
932 message
: MessageValue
<Response
>
934 updateMeasurementStatistics(
936 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
,
937 message
.taskPerformance
?.runTime
?? 0,
938 workerUsage
.tasks
.executed
942 private updateWaitTimeWorkerUsage (
943 workerUsage
: WorkerUsage
,
946 const timestamp
= performance
.now()
947 const taskWaitTime
= timestamp
- (task
.timestamp
?? timestamp
)
948 updateMeasurementStatistics(
949 workerUsage
.waitTime
,
950 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().waitTime
,
952 workerUsage
.tasks
.executed
956 private updateEluWorkerUsage (
957 workerUsage
: WorkerUsage
,
958 message
: MessageValue
<Response
>
960 const eluTaskStatisticsRequirements
: MeasurementStatisticsRequirements
=
961 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
962 updateMeasurementStatistics(
963 workerUsage
.elu
.active
,
964 eluTaskStatisticsRequirements
,
965 message
.taskPerformance
?.elu
?.active
?? 0,
966 workerUsage
.tasks
.executed
968 updateMeasurementStatistics(
969 workerUsage
.elu
.idle
,
970 eluTaskStatisticsRequirements
,
971 message
.taskPerformance
?.elu
?.idle
?? 0,
972 workerUsage
.tasks
.executed
974 if (eluTaskStatisticsRequirements
.aggregate
) {
975 if (message
.taskPerformance
?.elu
!= null) {
976 if (workerUsage
.elu
.utilization
!= null) {
977 workerUsage
.elu
.utilization
=
978 (workerUsage
.elu
.utilization
+
979 message
.taskPerformance
.elu
.utilization
) /
982 workerUsage
.elu
.utilization
= message
.taskPerformance
.elu
.utilization
989 * Chooses a worker node for the next task.
991 * The default worker choice strategy uses a round robin algorithm to distribute the tasks.
993 * @returns The chosen worker node key
995 private chooseWorkerNode (): number {
996 if (this.shallCreateDynamicWorker()) {
997 const workerNodeKey
= this.createAndSetupDynamicWorkerNode()
999 this.workerChoiceStrategyContext
.getStrategyPolicy().dynamicWorkerUsage
1001 return workerNodeKey
1004 return this.workerChoiceStrategyContext
.execute()
1008 * Conditions for dynamic worker creation.
1010 * @returns Whether to create a dynamic worker or not.
1012 private shallCreateDynamicWorker (): boolean {
1013 return this.type === PoolTypes
.dynamic
&& !this.full
&& this.internalBusy()
1017 * Sends a message to worker given its worker node key.
1019 * @param workerNodeKey - The worker node key.
1020 * @param message - The message.
1021 * @param transferList - The optional array of transferable objects.
1023 protected abstract sendToWorker (
1024 workerNodeKey
: number,
1025 message
: MessageValue
<Data
>,
1026 transferList
?: TransferListItem
[]
1030 * Creates a new worker.
1032 * @returns Newly created worker.
1034 protected abstract createWorker (): Worker
1037 * Creates a new, completely set up worker node.
1039 * @returns New, completely set up worker node key.
1041 protected createAndSetupWorkerNode (): number {
1042 const worker
= this.createWorker()
1044 worker
.on('online', this.opts
.onlineHandler
?? EMPTY_FUNCTION
)
1045 worker
.on('message', this.opts
.messageHandler
?? EMPTY_FUNCTION
)
1046 worker
.on('error', this.opts
.errorHandler
?? EMPTY_FUNCTION
)
1047 worker
.on('error', (error
) => {
1048 const workerNodeKey
= this.getWorkerNodeKeyByWorker(worker
)
1049 const workerInfo
= this.getWorkerInfo(workerNodeKey
) as WorkerInfo
1050 workerInfo
.ready
= false
1051 this.workerNodes
[workerNodeKey
].closeChannel()
1052 this.emitter
?.emit(PoolEvents
.error
, error
)
1053 if (this.opts
.restartWorkerOnError
=== true && !this.starting
) {
1054 if (workerInfo
.dynamic
) {
1055 this.createAndSetupDynamicWorkerNode()
1057 this.createAndSetupWorkerNode()
1060 if (this.opts
.enableTasksQueue
=== true) {
1061 this.redistributeQueuedTasks(workerNodeKey
)
1064 worker
.on('exit', this.opts
.exitHandler
?? EMPTY_FUNCTION
)
1065 worker
.once('exit', () => {
1066 this.removeWorkerNode(worker
)
1069 const workerNodeKey
= this.addWorkerNode(worker
)
1071 this.afterWorkerNodeSetup(workerNodeKey
)
1073 return workerNodeKey
1077 * Creates a new, completely set up dynamic worker node.
1079 * @returns New, completely set up dynamic worker node key.
1081 protected createAndSetupDynamicWorkerNode (): number {
1082 const workerNodeKey
= this.createAndSetupWorkerNode()
1083 this.registerWorkerMessageListener(workerNodeKey
, (message
) => {
1084 const localWorkerNodeKey
= this.getWorkerNodeKeyByWorkerId(
1087 const workerUsage
= this.workerNodes
[localWorkerNodeKey
].usage
1088 // Kill message received from worker
1090 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
1091 (isKillBehavior(KillBehaviors
.SOFT
, message
.kill
) &&
1092 ((this.opts
.enableTasksQueue
=== false &&
1093 workerUsage
.tasks
.executing
=== 0) ||
1094 (this.opts
.enableTasksQueue
=== true &&
1095 workerUsage
.tasks
.executing
=== 0 &&
1096 this.tasksQueueSize(localWorkerNodeKey
) === 0)))
1098 this.destroyWorkerNode(localWorkerNodeKey
).catch((error
) => {
1099 this.emitter
?.emit(PoolEvents
.error
, error
)
1103 const workerInfo
= this.getWorkerInfo(workerNodeKey
) as WorkerInfo
1104 this.sendToWorker(workerNodeKey
, {
1106 workerId
: workerInfo
.id
as number
1108 workerInfo
.dynamic
= true
1110 this.workerChoiceStrategyContext
.getStrategyPolicy().dynamicWorkerReady
||
1111 this.workerChoiceStrategyContext
.getStrategyPolicy().dynamicWorkerUsage
1113 workerInfo
.ready
= true
1115 this.checkAndEmitDynamicWorkerCreationEvents()
1116 return workerNodeKey
1120 * Registers a listener callback on the worker given its worker node key.
1122 * @param workerNodeKey - The worker node key.
1123 * @param listener - The message listener callback.
1125 protected abstract registerWorkerMessageListener
<
1126 Message
extends Data
| Response
1128 workerNodeKey
: number,
1129 listener
: (message
: MessageValue
<Message
>) => void
1133 * Method hooked up after a worker node has been newly created.
1134 * Can be overridden.
1136 * @param workerNodeKey - The newly created worker node key.
1138 protected afterWorkerNodeSetup (workerNodeKey
: number): void {
1139 // Listen to worker messages.
1140 this.registerWorkerMessageListener(workerNodeKey
, this.workerListener())
1141 // Send the startup message to worker.
1142 this.sendStartupMessageToWorker(workerNodeKey
)
1143 // Send the statistics message to worker.
1144 this.sendStatisticsMessageToWorker(workerNodeKey
)
1145 if (this.opts
.enableTasksQueue
=== true) {
1146 this.workerNodes
[workerNodeKey
].onBackPressure
=
1147 this.tasksStealingOnBackPressure
.bind(this)
1152 * Sends the startup message to worker given its worker node key.
1154 * @param workerNodeKey - The worker node key.
1156 protected abstract sendStartupMessageToWorker (workerNodeKey
: number): void
1159 * Sends the statistics message to worker given its worker node key.
1161 * @param workerNodeKey - The worker node key.
1163 private sendStatisticsMessageToWorker (workerNodeKey
: number): void {
1164 this.sendToWorker(workerNodeKey
, {
1167 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1169 elu
: this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1172 workerId
: (this.getWorkerInfo(workerNodeKey
) as WorkerInfo
).id
as number
1176 private redistributeQueuedTasks (workerNodeKey
: number): void {
1177 while (this.tasksQueueSize(workerNodeKey
) > 0) {
1178 let targetWorkerNodeKey
: number = workerNodeKey
1179 let minQueuedTasks
= Infinity
1180 let executeTask
= false
1181 for (const [workerNodeId
, workerNode
] of this.workerNodes
.entries()) {
1183 this.workerNodes
[workerNodeId
].usage
.tasks
.executing
<
1184 (this.opts
.tasksQueueOptions
?.concurrency
as number)
1189 workerNodeId
!== workerNodeKey
&&
1190 workerNode
.info
.ready
&&
1191 workerNode
.usage
.tasks
.queued
=== 0
1193 targetWorkerNodeKey
= workerNodeId
1197 workerNodeId
!== workerNodeKey
&&
1198 workerNode
.info
.ready
&&
1199 workerNode
.usage
.tasks
.queued
< minQueuedTasks
1201 minQueuedTasks
= workerNode
.usage
.tasks
.queued
1202 targetWorkerNodeKey
= workerNodeId
1207 targetWorkerNodeKey
,
1208 this.popTask(workerNodeKey
) as Task
<Data
>
1212 targetWorkerNodeKey
,
1213 this.popTask(workerNodeKey
) as Task
<Data
>
1219 private tasksStealingOnBackPressure (workerId
: number): void {
1220 const sourceWorkerNode
=
1221 this.workerNodes
[this.getWorkerNodeKeyByWorkerId(workerId
)]
1222 const workerNodes
= this.workerNodes
1223 .filter((workerNode
) => workerNode
.info
.id
!== workerId
)
1225 (workerNodeA
, workerNodeB
) =>
1226 workerNodeA
.usage
.tasks
.queued
- workerNodeB
.usage
.tasks
.queued
1228 for (const [workerNodeKey
, workerNode
] of workerNodes
.entries()) {
1230 workerNode
.info
.ready
&&
1231 sourceWorkerNode
.usage
.tasks
.queued
> 0 &&
1232 !workerNode
.hasBackPressure() &&
1233 workerNode
.usage
.tasks
.executing
<
1234 (this.opts
.tasksQueueOptions
?.concurrency
as number)
1238 sourceWorkerNode
.popTask() as Task
<Data
>
1241 workerNode
.info
.ready
&&
1242 sourceWorkerNode
.usage
.tasks
.queued
> 0 &&
1243 !workerNode
.hasBackPressure() &&
1244 workerNode
.usage
.tasks
.executing
>=
1245 (this.opts
.tasksQueueOptions
?.concurrency
as number)
1249 sourceWorkerNode
.popTask() as Task
<Data
>
1256 * This method is the listener registered for each worker message.
1258 * @returns The listener function to execute when a message is received from a worker.
1260 protected workerListener (): (message
: MessageValue
<Response
>) => void {
1261 return (message
) => {
1262 this.checkMessageWorkerId(message
)
1263 if (message
.ready
!= null && message
.taskFunctions
!= null) {
1264 // Worker ready response received from worker
1265 this.handleWorkerReadyResponse(message
)
1266 } else if (message
.taskId
!= null) {
1267 // Task execution response received from worker
1268 this.handleTaskExecutionResponse(message
)
1269 } else if (message
.taskFunctions
!= null) {
1270 // Task functions message received from worker
1273 this.getWorkerNodeKeyByWorkerId(message
.workerId
)
1275 ).taskFunctions
= message
.taskFunctions
1280 private handleWorkerReadyResponse (message
: MessageValue
<Response
>): void {
1281 if (message
.ready
=== false) {
1282 throw new Error(`Worker ${message.workerId} failed to initialize`)
1284 const workerInfo
= this.getWorkerInfo(
1285 this.getWorkerNodeKeyByWorkerId(message
.workerId
)
1287 workerInfo
.ready
= message
.ready
as boolean
1288 workerInfo
.taskFunctions
= message
.taskFunctions
1289 if (this.emitter
!= null && this.ready
) {
1290 this.emitter
.emit(PoolEvents
.ready
, this.info
)
1294 private handleTaskExecutionResponse (message
: MessageValue
<Response
>): void {
1295 const { taskId
, taskError
, data
} = message
1296 const promiseResponse
= this.promiseResponseMap
.get(taskId
as string)
1297 if (promiseResponse
!= null) {
1298 if (taskError
!= null) {
1299 this.emitter
?.emit(PoolEvents
.taskError
, taskError
)
1300 promiseResponse
.reject(taskError
.message
)
1302 promiseResponse
.resolve(data
as Response
)
1304 const workerNodeKey
= promiseResponse
.workerNodeKey
1305 this.afterTaskExecutionHook(workerNodeKey
, message
)
1306 this.promiseResponseMap
.delete(taskId
as string)
1308 this.opts
.enableTasksQueue
=== true &&
1309 this.tasksQueueSize(workerNodeKey
) > 0 &&
1310 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
<
1311 (this.opts
.tasksQueueOptions
?.concurrency
as number)
1315 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1318 this.workerChoiceStrategyContext
.update(workerNodeKey
)
1322 private checkAndEmitTaskExecutionEvents (): void {
1324 this.emitter
?.emit(PoolEvents
.busy
, this.info
)
1328 private checkAndEmitTaskQueuingEvents (): void {
1329 if (this.hasBackPressure()) {
1330 this.emitter
?.emit(PoolEvents
.backPressure
, this.info
)
1334 private checkAndEmitDynamicWorkerCreationEvents (): void {
1335 if (this.type === PoolTypes
.dynamic
) {
1337 this.emitter
?.emit(PoolEvents
.full
, this.info
)
1343 * Gets the worker information given its worker node key.
1345 * @param workerNodeKey - The worker node key.
1346 * @returns The worker information.
1348 protected getWorkerInfo (workerNodeKey
: number): WorkerInfo
| undefined {
1349 return this.workerNodes
[workerNodeKey
]?.info
1353 * Adds the given worker in the pool worker nodes.
1355 * @param worker - The worker.
1356 * @returns The added worker node key.
1357 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the added worker node is not found.
1359 private addWorkerNode (worker
: Worker
): number {
1360 const workerNode
= new WorkerNode
<Worker
, Data
>(
1363 this.opts
.tasksQueueOptions
?.queueMaxSize
?? Math.pow(this.maxSize
, 2)
1365 // Flag the worker node as ready at pool startup.
1366 if (this.starting
) {
1367 workerNode
.info
.ready
= true
1369 this.workerNodes
.push(workerNode
)
1370 const workerNodeKey
= this.getWorkerNodeKeyByWorker(worker
)
1371 if (workerNodeKey
=== -1) {
1372 throw new Error('Worker node added not found')
1374 return workerNodeKey
1378 * Removes the given worker from the pool worker nodes.
1380 * @param worker - The worker.
1382 private removeWorkerNode (worker
: Worker
): void {
1383 const workerNodeKey
= this.getWorkerNodeKeyByWorker(worker
)
1384 if (workerNodeKey
!== -1) {
1385 this.workerNodes
.splice(workerNodeKey
, 1)
1386 this.workerChoiceStrategyContext
.remove(workerNodeKey
)
1391 public hasWorkerNodeBackPressure (workerNodeKey
: number): boolean {
1393 this.opts
.enableTasksQueue
=== true &&
1394 this.workerNodes
[workerNodeKey
].hasBackPressure()
1398 private hasBackPressure (): boolean {
1400 this.opts
.enableTasksQueue
=== true &&
1401 this.workerNodes
.findIndex(
1402 (workerNode
) => !workerNode
.hasBackPressure()
1408 * Executes the given task on the worker given its worker node key.
1410 * @param workerNodeKey - The worker node key.
1411 * @param task - The task to execute.
1413 private executeTask (workerNodeKey
: number, task
: Task
<Data
>): void {
1414 this.beforeTaskExecutionHook(workerNodeKey
, task
)
1415 this.sendToWorker(workerNodeKey
, task
, task
.transferList
)
1416 this.checkAndEmitTaskExecutionEvents()
1419 private enqueueTask (workerNodeKey
: number, task
: Task
<Data
>): number {
1420 const tasksQueueSize
= this.workerNodes
[workerNodeKey
].enqueueTask(task
)
1421 this.checkAndEmitTaskQueuingEvents()
1422 return tasksQueueSize
1425 private dequeueTask (workerNodeKey
: number): Task
<Data
> | undefined {
1426 return this.workerNodes
[workerNodeKey
].dequeueTask()
1429 private popTask (workerNodeKey
: number): Task
<Data
> | undefined {
1430 return this.workerNodes
[workerNodeKey
].popTask()
1433 private tasksQueueSize (workerNodeKey
: number): number {
1434 return this.workerNodes
[workerNodeKey
].tasksQueueSize()
1437 protected flushTasksQueue (workerNodeKey
: number): void {
1438 while (this.tasksQueueSize(workerNodeKey
) > 0) {
1441 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1444 this.workerNodes
[workerNodeKey
].clearTasksQueue()
1447 private flushTasksQueues (): void {
1448 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
1449 this.flushTasksQueue(workerNodeKey
)