1 import { randomUUID
} from
'node:crypto'
2 import { performance
} from
'node:perf_hooks'
3 import { existsSync
} from
'node:fs'
6 PromiseResponseWrapper
,
8 } from
'../utility-types'
11 DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
,
18 import { KillBehaviors
} from
'../worker/worker-options'
27 type TasksQueueOptions
39 WorkerChoiceStrategies
,
40 type WorkerChoiceStrategy
,
41 type WorkerChoiceStrategyOptions
42 } from
'./selection-strategies/selection-strategies-types'
43 import { WorkerChoiceStrategyContext
} from
'./selection-strategies/worker-choice-strategy-context'
44 import { version
} from
'./version'
45 import { WorkerNode
} from
'./worker-node'
48 * Base class that implements some shared logic for all poolifier pools.
50 * @typeParam Worker - Type of worker which manages this pool.
51 * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
52 * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
54 export abstract class AbstractPool
<
55 Worker
extends IWorker
,
58 > implements IPool
<Worker
, Data
, Response
> {
60 public readonly workerNodes
: Array<IWorkerNode
<Worker
, Data
>> = []
63 public readonly emitter
?: PoolEmitter
66 * The execution response promise map.
68 * - `key`: The message id of each submitted task.
69 * - `value`: An object that contains the worker, the execution response promise resolve and reject callbacks.
71 * When we receive a message from the worker, we get a map entry with the promise resolve/reject bound to the message id.
73 protected promiseResponseMap
: Map
<
75 PromiseResponseWrapper
<Worker
, Response
>
76 > = new Map
<string, PromiseResponseWrapper
<Worker
, Response
>>()
79 * Worker choice strategy context referencing a worker choice algorithm implementation.
81 protected workerChoiceStrategyContext
: WorkerChoiceStrategyContext
<
88 * The start timestamp of the pool.
90 private readonly startTimestamp
93 * Constructs a new poolifier pool.
95 * @param numberOfWorkers - Number of workers that this pool should manage.
96 * @param filePath - Path to the worker file.
97 * @param opts - Options for the pool.
100 protected readonly numberOfWorkers
: number,
101 protected readonly filePath
: string,
102 protected readonly opts
: PoolOptions
<Worker
>
104 if (!this.isMain()) {
105 throw new Error('Cannot start a pool from a worker!')
107 this.checkNumberOfWorkers(this.numberOfWorkers
)
108 this.checkFilePath(this.filePath
)
109 this.checkPoolOptions(this.opts
)
111 this.chooseWorkerNode
= this.chooseWorkerNode
.bind(this)
112 this.executeTask
= this.executeTask
.bind(this)
113 this.enqueueTask
= this.enqueueTask
.bind(this)
114 this.checkAndEmitEvents
= this.checkAndEmitEvents
.bind(this)
116 if (this.opts
.enableEvents
=== true) {
117 this.emitter
= new PoolEmitter()
119 this.workerChoiceStrategyContext
= new WorkerChoiceStrategyContext
<
125 this.opts
.workerChoiceStrategy
,
126 this.opts
.workerChoiceStrategyOptions
131 while (this.workerNodes
.length
< this.numberOfWorkers
) {
132 this.createAndSetupWorker()
135 this.startTimestamp
= performance
.now()
138 private checkFilePath (filePath
: string): void {
141 typeof filePath
!== 'string' ||
142 (typeof filePath
=== 'string' && filePath
.trim().length
=== 0)
144 throw new Error('Please specify a file with a worker implementation')
146 if (!existsSync(filePath
)) {
147 throw new Error(`Cannot find the worker file '${filePath}'`)
151 private checkNumberOfWorkers (numberOfWorkers
: number): void {
152 if (numberOfWorkers
== null) {
154 'Cannot instantiate a pool without specifying the number of workers'
156 } else if (!Number.isSafeInteger(numberOfWorkers
)) {
158 'Cannot instantiate a pool with a non safe integer number of workers'
160 } else if (numberOfWorkers
< 0) {
161 throw new RangeError(
162 'Cannot instantiate a pool with a negative number of workers'
164 } else if (this.type === PoolTypes
.fixed
&& numberOfWorkers
=== 0) {
165 throw new RangeError('Cannot instantiate a fixed pool with zero worker')
169 protected checkDynamicPoolSize (min
: number, max
: number): void {
170 if (this.type === PoolTypes
.dynamic
) {
172 throw new RangeError(
173 'Cannot instantiate a dynamic pool with a maximum pool size inferior to the minimum pool size'
175 } else if (min
=== 0 && max
=== 0) {
176 throw new RangeError(
177 'Cannot instantiate a dynamic pool with a minimum pool size and a maximum pool size equal to zero'
179 } else if (min
=== max
) {
180 throw new RangeError(
181 'Cannot instantiate a dynamic pool with a minimum pool size equal to the maximum pool size. Use a fixed pool instead'
187 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
188 if (isPlainObject(opts
)) {
189 this.opts
.workerChoiceStrategy
=
190 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
191 this.checkValidWorkerChoiceStrategy(this.opts
.workerChoiceStrategy
)
192 this.opts
.workerChoiceStrategyOptions
=
193 opts
.workerChoiceStrategyOptions
??
194 DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
195 this.checkValidWorkerChoiceStrategyOptions(
196 this.opts
.workerChoiceStrategyOptions
198 this.opts
.restartWorkerOnError
= opts
.restartWorkerOnError
?? true
199 this.opts
.enableEvents
= opts
.enableEvents
?? true
200 this.opts
.enableTasksQueue
= opts
.enableTasksQueue
?? false
201 if (this.opts
.enableTasksQueue
) {
202 this.checkValidTasksQueueOptions(
203 opts
.tasksQueueOptions
as TasksQueueOptions
205 this.opts
.tasksQueueOptions
= this.buildTasksQueueOptions(
206 opts
.tasksQueueOptions
as TasksQueueOptions
210 throw new TypeError('Invalid pool options: must be a plain object')
214 private checkValidWorkerChoiceStrategy (
215 workerChoiceStrategy
: WorkerChoiceStrategy
217 if (!Object.values(WorkerChoiceStrategies
).includes(workerChoiceStrategy
)) {
219 `Invalid worker choice strategy '${workerChoiceStrategy}'`
224 private checkValidWorkerChoiceStrategyOptions (
225 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
227 if (!isPlainObject(workerChoiceStrategyOptions
)) {
229 'Invalid worker choice strategy options: must be a plain object'
233 workerChoiceStrategyOptions
.weights
!= null &&
234 Object.keys(workerChoiceStrategyOptions
.weights
).length
!== this.maxSize
237 'Invalid worker choice strategy options: must have a weight for each worker node'
241 workerChoiceStrategyOptions
.measurement
!= null &&
242 !Object.values(Measurements
).includes(
243 workerChoiceStrategyOptions
.measurement
247 `Invalid worker choice strategy options: invalid measurement '${workerChoiceStrategyOptions.measurement}'`
252 private checkValidTasksQueueOptions (
253 tasksQueueOptions
: TasksQueueOptions
255 if (tasksQueueOptions
!= null && !isPlainObject(tasksQueueOptions
)) {
256 throw new TypeError('Invalid tasks queue options: must be a plain object')
259 tasksQueueOptions
?.concurrency
!= null &&
260 !Number.isSafeInteger(tasksQueueOptions
.concurrency
)
263 'Invalid worker tasks concurrency: must be an integer'
267 tasksQueueOptions
?.concurrency
!= null &&
268 tasksQueueOptions
.concurrency
<= 0
271 `Invalid worker tasks concurrency '${tasksQueueOptions.concurrency}'`
277 public get
info (): PoolInfo
{
283 strategy
: this.opts
.workerChoiceStrategy
as WorkerChoiceStrategy
,
284 minSize
: this.minSize
,
285 maxSize
: this.maxSize
,
286 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
287 .runTime
.aggregate
&&
288 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
289 .waitTime
.aggregate
&& { utilization
: round(this.utilization
) }),
290 workerNodes
: this.workerNodes
.length
,
291 idleWorkerNodes
: this.workerNodes
.reduce(
292 (accumulator
, workerNode
) =>
293 workerNode
.usage
.tasks
.executing
=== 0
298 busyWorkerNodes
: this.workerNodes
.reduce(
299 (accumulator
, workerNode
) =>
300 workerNode
.usage
.tasks
.executing
> 0 ? accumulator
+ 1 : accumulator
,
303 executedTasks
: this.workerNodes
.reduce(
304 (accumulator
, workerNode
) =>
305 accumulator
+ workerNode
.usage
.tasks
.executed
,
308 executingTasks
: this.workerNodes
.reduce(
309 (accumulator
, workerNode
) =>
310 accumulator
+ workerNode
.usage
.tasks
.executing
,
313 queuedTasks
: this.workerNodes
.reduce(
314 (accumulator
, workerNode
) =>
315 accumulator
+ workerNode
.usage
.tasks
.queued
,
318 maxQueuedTasks
: this.workerNodes
.reduce(
319 (accumulator
, workerNode
) =>
320 accumulator
+ (workerNode
.usage
.tasks
?.maxQueued
?? 0),
323 failedTasks
: this.workerNodes
.reduce(
324 (accumulator
, workerNode
) =>
325 accumulator
+ workerNode
.usage
.tasks
.failed
,
328 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
329 .runTime
.aggregate
&& {
333 ...this.workerNodes
.map(
334 workerNode
=> workerNode
.usage
.runTime
?.minimum
?? Infinity
340 ...this.workerNodes
.map(
341 workerNode
=> workerNode
.usage
.runTime
?.maximum
?? -Infinity
346 this.workerNodes
.reduce(
347 (accumulator
, workerNode
) =>
348 accumulator
+ (workerNode
.usage
.runTime
?.aggregate
?? 0),
351 this.workerNodes
.reduce(
352 (accumulator
, workerNode
) =>
353 accumulator
+ (workerNode
.usage
.tasks
?.executed
?? 0),
357 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
361 this.workerNodes
.map(
362 workerNode
=> workerNode
.usage
.runTime
?.median
?? 0
369 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
370 .waitTime
.aggregate
&& {
374 ...this.workerNodes
.map(
375 workerNode
=> workerNode
.usage
.waitTime
?.minimum
?? Infinity
381 ...this.workerNodes
.map(
382 workerNode
=> workerNode
.usage
.waitTime
?.maximum
?? -Infinity
387 this.workerNodes
.reduce(
388 (accumulator
, workerNode
) =>
389 accumulator
+ (workerNode
.usage
.waitTime
?.aggregate
?? 0),
392 this.workerNodes
.reduce(
393 (accumulator
, workerNode
) =>
394 accumulator
+ (workerNode
.usage
.tasks
?.executed
?? 0),
398 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
399 .waitTime
.median
&& {
402 this.workerNodes
.map(
403 workerNode
=> workerNode
.usage
.waitTime
?.median
?? 0
413 private get
starting (): boolean {
414 return this.workerNodes
.length
< this.minSize
417 private get
ready (): boolean {
419 this.workerNodes
.length
>= this.minSize
&&
420 this.workerNodes
.every(
421 (workerNode
, workerNodeKey
) =>
422 workerNodeKey
< this.minSize
&& workerNode
.info
.ready
428 * Gets the approximate pool utilization.
430 * @returns The pool utilization.
432 private get
utilization (): number {
433 const poolTimeCapacity
=
434 (performance
.now() - this.startTimestamp
) * this.maxSize
435 const totalTasksRunTime
= this.workerNodes
.reduce(
436 (accumulator
, workerNode
) =>
437 accumulator
+ (workerNode
.usage
.runTime
?.aggregate
?? 0),
440 const totalTasksWaitTime
= this.workerNodes
.reduce(
441 (accumulator
, workerNode
) =>
442 accumulator
+ (workerNode
.usage
.waitTime
?.aggregate
?? 0),
445 return (totalTasksRunTime
+ totalTasksWaitTime
) / poolTimeCapacity
451 * If it is `'dynamic'`, it provides the `max` property.
453 protected abstract get
type (): PoolType
456 * Gets the worker type.
458 protected abstract get
worker (): WorkerType
463 protected abstract get
minSize (): number
468 protected abstract get
maxSize (): number
471 * Get the worker given its id.
473 * @param workerId - The worker id.
474 * @returns The worker if found in the pool worker nodes, `undefined` otherwise.
476 private getWorkerById (workerId
: number): Worker
| undefined {
477 return this.workerNodes
.find(workerNode
=> workerNode
.info
.id
=== workerId
)
482 * Checks if the worker id sent in the received message from a worker is valid.
484 * @param message - The received message.
485 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the worker id is invalid.
487 private checkMessageWorkerId (message
: MessageValue
<Response
>): void {
489 message
.workerId
!= null &&
490 this.getWorkerById(message
.workerId
) == null
493 `Worker message received from unknown worker '${message.workerId}'`
499 * Gets the given worker its worker node key.
501 * @param worker - The worker.
502 * @returns The worker node key if found in the pool worker nodes, `-1` otherwise.
504 private getWorkerNodeKey (worker
: Worker
): number {
505 return this.workerNodes
.findIndex(
506 workerNode
=> workerNode
.worker
=== worker
511 public setWorkerChoiceStrategy (
512 workerChoiceStrategy
: WorkerChoiceStrategy
,
513 workerChoiceStrategyOptions
?: WorkerChoiceStrategyOptions
515 this.checkValidWorkerChoiceStrategy(workerChoiceStrategy
)
516 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
517 this.workerChoiceStrategyContext
.setWorkerChoiceStrategy(
518 this.opts
.workerChoiceStrategy
520 if (workerChoiceStrategyOptions
!= null) {
521 this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
523 for (const workerNode
of this.workerNodes
) {
524 workerNode
.resetUsage()
525 this.setWorkerStatistics(workerNode
.worker
)
530 public setWorkerChoiceStrategyOptions (
531 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
533 this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
534 this.opts
.workerChoiceStrategyOptions
= workerChoiceStrategyOptions
535 this.workerChoiceStrategyContext
.setOptions(
536 this.opts
.workerChoiceStrategyOptions
541 public enableTasksQueue (
543 tasksQueueOptions
?: TasksQueueOptions
545 if (this.opts
.enableTasksQueue
=== true && !enable
) {
546 this.flushTasksQueues()
548 this.opts
.enableTasksQueue
= enable
549 this.setTasksQueueOptions(tasksQueueOptions
as TasksQueueOptions
)
553 public setTasksQueueOptions (tasksQueueOptions
: TasksQueueOptions
): void {
554 if (this.opts
.enableTasksQueue
=== true) {
555 this.checkValidTasksQueueOptions(tasksQueueOptions
)
556 this.opts
.tasksQueueOptions
=
557 this.buildTasksQueueOptions(tasksQueueOptions
)
558 } else if (this.opts
.tasksQueueOptions
!= null) {
559 delete this.opts
.tasksQueueOptions
563 private buildTasksQueueOptions (
564 tasksQueueOptions
: TasksQueueOptions
565 ): TasksQueueOptions
{
567 concurrency
: tasksQueueOptions
?.concurrency
?? 1
572 * Whether the pool is full or not.
574 * The pool filling boolean status.
576 protected get
full (): boolean {
577 return this.workerNodes
.length
>= this.maxSize
581 * Whether the pool is busy or not.
583 * The pool busyness boolean status.
585 protected abstract get
busy (): boolean
588 * Whether worker nodes are executing at least one task.
590 * @returns Worker nodes busyness boolean status.
592 protected internalBusy (): boolean {
594 this.workerNodes
.findIndex(workerNode
=> {
595 return workerNode
.usage
.tasks
.executing
=== 0
601 public async execute (data
?: Data
, name
?: string): Promise
<Response
> {
602 const timestamp
= performance
.now()
603 const workerNodeKey
= this.chooseWorkerNode()
604 const submittedTask
: Task
<Data
> = {
605 name
: name
?? DEFAULT_TASK_NAME
,
606 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
607 data
: data
?? ({} as Data
),
609 workerId
: this.getWorkerInfo(workerNodeKey
).id
as number,
612 const res
= new Promise
<Response
>((resolve
, reject
) => {
613 this.promiseResponseMap
.set(submittedTask
.id
as string, {
616 worker
: this.workerNodes
[workerNodeKey
].worker
620 this.opts
.enableTasksQueue
=== true &&
622 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
>=
623 ((this.opts
.tasksQueueOptions
as TasksQueueOptions
)
624 .concurrency
as number))
626 this.enqueueTask(workerNodeKey
, submittedTask
)
628 this.executeTask(workerNodeKey
, submittedTask
)
630 this.checkAndEmitEvents()
631 // eslint-disable-next-line @typescript-eslint/return-await
636 public async destroy (): Promise
<void> {
638 this.workerNodes
.map(async (workerNode
, workerNodeKey
) => {
639 this.flushTasksQueue(workerNodeKey
)
640 // FIXME: wait for tasks to be finished
641 const workerExitPromise
= new Promise
<void>(resolve
=> {
642 workerNode
.worker
.on('exit', () => {
646 await this.destroyWorker(workerNode
.worker
)
647 await workerExitPromise
653 * Terminates the given worker.
655 * @param worker - A worker within `workerNodes`.
657 protected abstract destroyWorker (worker
: Worker
): void | Promise
<void>
660 * Setup hook to execute code before worker nodes are created in the abstract constructor.
665 protected setupHook (): void {
666 // Intentionally empty
670 * Should return whether the worker is the main worker or not.
672 protected abstract isMain (): boolean
675 * Hook executed before the worker task execution.
678 * @param workerNodeKey - The worker node key.
679 * @param task - The task to execute.
681 protected beforeTaskExecutionHook (
682 workerNodeKey
: number,
685 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
686 ++workerUsage
.tasks
.executing
687 this.updateWaitTimeWorkerUsage(workerUsage
, task
)
688 const taskWorkerUsage
= this.workerNodes
[workerNodeKey
].getTaskWorkerUsage(
691 ++taskWorkerUsage
.tasks
.executing
692 this.updateWaitTimeWorkerUsage(taskWorkerUsage
, task
)
696 * Hook executed after the worker task execution.
699 * @param worker - The worker.
700 * @param message - The received message.
702 protected afterTaskExecutionHook (
704 message
: MessageValue
<Response
>
706 const workerNodeKey
= this.getWorkerNodeKey(worker
)
707 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
708 this.updateTaskStatisticsWorkerUsage(workerUsage
, message
)
709 this.updateRunTimeWorkerUsage(workerUsage
, message
)
710 this.updateEluWorkerUsage(workerUsage
, message
)
711 const taskWorkerUsage
= this.workerNodes
[workerNodeKey
].getTaskWorkerUsage(
712 message
.taskPerformance
?.name
?? DEFAULT_TASK_NAME
714 this.updateTaskStatisticsWorkerUsage(taskWorkerUsage
, message
)
715 this.updateRunTimeWorkerUsage(taskWorkerUsage
, message
)
716 this.updateEluWorkerUsage(taskWorkerUsage
, message
)
719 private updateTaskStatisticsWorkerUsage (
720 workerUsage
: WorkerUsage
,
721 message
: MessageValue
<Response
>
723 const workerTaskStatistics
= workerUsage
.tasks
724 --workerTaskStatistics
.executing
725 if (message
.taskError
== null) {
726 ++workerTaskStatistics
.executed
728 ++workerTaskStatistics
.failed
732 private updateRunTimeWorkerUsage (
733 workerUsage
: WorkerUsage
,
734 message
: MessageValue
<Response
>
737 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
740 const taskRunTime
= message
.taskPerformance
?.runTime
?? 0
741 workerUsage
.runTime
.aggregate
=
742 (workerUsage
.runTime
.aggregate
?? 0) + taskRunTime
743 workerUsage
.runTime
.minimum
= Math.min(
745 workerUsage
.runTime
?.minimum
?? Infinity
747 workerUsage
.runTime
.maximum
= Math.max(
749 workerUsage
.runTime
?.maximum
?? -Infinity
752 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
754 workerUsage
.tasks
.executed
!== 0
756 workerUsage
.runTime
.average
=
757 workerUsage
.runTime
.aggregate
/ workerUsage
.tasks
.executed
760 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
762 message
.taskPerformance
?.runTime
!= null
764 workerUsage
.runTime
.history
.push(message
.taskPerformance
.runTime
)
765 workerUsage
.runTime
.median
= median(workerUsage
.runTime
.history
)
770 private updateWaitTimeWorkerUsage (
771 workerUsage
: WorkerUsage
,
774 const timestamp
= performance
.now()
775 const taskWaitTime
= timestamp
- (task
.timestamp
?? timestamp
)
777 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().waitTime
780 workerUsage
.waitTime
.aggregate
=
781 (workerUsage
.waitTime
?.aggregate
?? 0) + taskWaitTime
782 workerUsage
.waitTime
.minimum
= Math.min(
784 workerUsage
.waitTime
?.minimum
?? Infinity
786 workerUsage
.waitTime
.maximum
= Math.max(
788 workerUsage
.waitTime
?.maximum
?? -Infinity
791 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
793 workerUsage
.tasks
.executed
!== 0
795 workerUsage
.waitTime
.average
=
796 workerUsage
.waitTime
.aggregate
/ workerUsage
.tasks
.executed
799 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
802 workerUsage
.waitTime
.history
.push(taskWaitTime
)
803 workerUsage
.waitTime
.median
= median(workerUsage
.waitTime
.history
)
808 private updateEluWorkerUsage (
809 workerUsage
: WorkerUsage
,
810 message
: MessageValue
<Response
>
813 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
816 if (message
.taskPerformance
?.elu
!= null) {
817 workerUsage
.elu
.idle
.aggregate
=
818 (workerUsage
.elu
.idle
?.aggregate
?? 0) +
819 message
.taskPerformance
.elu
.idle
820 workerUsage
.elu
.active
.aggregate
=
821 (workerUsage
.elu
.active
?.aggregate
?? 0) +
822 message
.taskPerformance
.elu
.active
823 if (workerUsage
.elu
.utilization
!= null) {
824 workerUsage
.elu
.utilization
=
825 (workerUsage
.elu
.utilization
+
826 message
.taskPerformance
.elu
.utilization
) /
829 workerUsage
.elu
.utilization
= message
.taskPerformance
.elu
.utilization
831 workerUsage
.elu
.idle
.minimum
= Math.min(
832 message
.taskPerformance
.elu
.idle
,
833 workerUsage
.elu
.idle
?.minimum
?? Infinity
835 workerUsage
.elu
.idle
.maximum
= Math.max(
836 message
.taskPerformance
.elu
.idle
,
837 workerUsage
.elu
.idle
?.maximum
?? -Infinity
839 workerUsage
.elu
.active
.minimum
= Math.min(
840 message
.taskPerformance
.elu
.active
,
841 workerUsage
.elu
.active
?.minimum
?? Infinity
843 workerUsage
.elu
.active
.maximum
= Math.max(
844 message
.taskPerformance
.elu
.active
,
845 workerUsage
.elu
.active
?.maximum
?? -Infinity
848 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
850 workerUsage
.tasks
.executed
!== 0
852 workerUsage
.elu
.idle
.average
=
853 workerUsage
.elu
.idle
.aggregate
/ workerUsage
.tasks
.executed
854 workerUsage
.elu
.active
.average
=
855 workerUsage
.elu
.active
.aggregate
/ workerUsage
.tasks
.executed
858 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
861 workerUsage
.elu
.idle
.history
.push(message
.taskPerformance
.elu
.idle
)
862 workerUsage
.elu
.active
.history
.push(
863 message
.taskPerformance
.elu
.active
865 workerUsage
.elu
.idle
.median
= median(workerUsage
.elu
.idle
.history
)
866 workerUsage
.elu
.active
.median
= median(workerUsage
.elu
.active
.history
)
873 * Chooses a worker node for the next task.
875 * The default worker choice strategy uses a round robin algorithm to distribute the tasks.
877 * @returns The worker node key
879 private chooseWorkerNode (): number {
880 if (this.shallCreateDynamicWorker()) {
881 const worker
= this.createAndSetupDynamicWorker()
883 this.workerChoiceStrategyContext
.getStrategyPolicy().useDynamicWorker
885 return this.getWorkerNodeKey(worker
)
888 return this.workerChoiceStrategyContext
.execute()
892 * Conditions for dynamic worker creation.
894 * @returns Whether to create a dynamic worker or not.
896 private shallCreateDynamicWorker (): boolean {
897 return this.type === PoolTypes
.dynamic
&& !this.full
&& this.internalBusy()
901 * Sends a message to the given worker.
903 * @param worker - The worker which should receive the message.
904 * @param message - The message.
906 protected abstract sendToWorker (
908 message
: MessageValue
<Data
>
912 * Creates a new worker.
914 * @returns Newly created worker.
916 protected abstract createWorker (): Worker
919 * Creates a new worker and sets it up completely in the pool worker nodes.
921 * @returns New, completely set up worker.
923 protected createAndSetupWorker (): Worker
{
924 const worker
= this.createWorker()
926 worker
.on('message', this.opts
.messageHandler
?? EMPTY_FUNCTION
)
927 worker
.on('error', this.opts
.errorHandler
?? EMPTY_FUNCTION
)
928 worker
.on('error', error
=> {
929 const workerNodeKey
= this.getWorkerNodeKey(worker
)
930 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
931 workerInfo
.ready
= false
932 if (this.emitter
!= null) {
933 this.emitter
.emit(PoolEvents
.error
, error
)
935 if (this.opts
.restartWorkerOnError
=== true && !this.starting
) {
936 if (workerInfo
.dynamic
) {
937 this.createAndSetupDynamicWorker()
939 this.createAndSetupWorker()
942 if (this.opts
.enableTasksQueue
=== true) {
943 this.redistributeQueuedTasks(workerNodeKey
)
946 worker
.on('online', this.opts
.onlineHandler
?? EMPTY_FUNCTION
)
947 worker
.on('exit', this.opts
.exitHandler
?? EMPTY_FUNCTION
)
948 worker
.once('exit', () => {
949 this.removeWorkerNode(worker
)
952 this.pushWorkerNode(worker
)
954 this.afterWorkerSetup(worker
)
960 * Creates a new dynamic worker and sets it up completely in the pool worker nodes.
962 * @returns New, completely set up dynamic worker.
964 protected createAndSetupDynamicWorker (): Worker
{
965 const worker
= this.createAndSetupWorker()
966 this.registerWorkerMessageListener(worker
, message
=> {
967 const workerNodeKey
= this.getWorkerNodeKey(worker
)
969 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
970 (message
.kill
!= null &&
971 ((this.opts
.enableTasksQueue
=== false &&
972 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
=== 0) ||
973 (this.opts
.enableTasksQueue
=== true &&
974 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
=== 0 &&
975 this.tasksQueueSize(workerNodeKey
) === 0)))
977 // Kill message received from the worker: no new tasks are submitted to that worker for a while ( > maxInactiveTime)
978 void (this.destroyWorker(worker
) as Promise
<void>)
981 const workerInfo
= this.getWorkerInfo(this.getWorkerNodeKey(worker
))
982 workerInfo
.ready
= true
983 workerInfo
.dynamic
= true
984 this.sendToWorker(worker
, {
986 workerId
: workerInfo
.id
as number
992 * Registers a listener callback on the given worker.
994 * @param worker - The worker which should register a listener.
995 * @param listener - The message listener callback.
997 private registerWorkerMessageListener
<Message
extends Data
| Response
>(
999 listener
: (message
: MessageValue
<Message
>) => void
1001 worker
.on('message', listener
as MessageHandler
<Worker
>)
1005 * Function that can be hooked up when a worker has been newly created and moved to the pool worker nodes.
1006 * Can be overridden.
1008 * @param worker - The newly created worker.
1010 protected afterWorkerSetup (worker
: Worker
): void {
1011 // Listen to worker messages.
1012 this.registerWorkerMessageListener(worker
, this.workerListener())
1013 // Send startup message to worker.
1014 this.sendWorkerStartupMessage(worker
)
1015 // Setup worker task statistics computation.
1016 this.setWorkerStatistics(worker
)
1019 private sendWorkerStartupMessage (worker
: Worker
): void {
1020 this.sendToWorker(worker
, {
1022 workerId
: this.getWorkerInfo(this.getWorkerNodeKey(worker
)).id
as number
1026 private redistributeQueuedTasks (workerNodeKey
: number): void {
1027 while (this.tasksQueueSize(workerNodeKey
) > 0) {
1028 let targetWorkerNodeKey
: number = workerNodeKey
1029 let minQueuedTasks
= Infinity
1030 for (const [workerNodeId
, workerNode
] of this.workerNodes
.entries()) {
1031 const workerInfo
= this.getWorkerInfo(workerNodeId
)
1033 workerNodeId
!== workerNodeKey
&&
1035 workerNode
.usage
.tasks
.queued
=== 0
1037 targetWorkerNodeKey
= workerNodeId
1041 workerNodeId
!== workerNodeKey
&&
1043 workerNode
.usage
.tasks
.queued
< minQueuedTasks
1045 minQueuedTasks
= workerNode
.usage
.tasks
.queued
1046 targetWorkerNodeKey
= workerNodeId
1050 targetWorkerNodeKey
,
1051 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1057 * This function is the listener registered for each worker message.
1059 * @returns The listener function to execute when a message is received from a worker.
1061 protected workerListener (): (message
: MessageValue
<Response
>) => void {
1063 this.checkMessageWorkerId(message
)
1064 if (message
.ready
!= null) {
1065 // Worker ready message received
1066 this.handleWorkerReadyMessage(message
)
1067 } else if (message
.id
!= null) {
1068 // Task execution response received
1069 this.handleTaskExecutionResponse(message
)
1074 private handleWorkerReadyMessage (message
: MessageValue
<Response
>): void {
1075 const worker
= this.getWorkerById(message
.workerId
)
1076 this.getWorkerInfo(this.getWorkerNodeKey(worker
as Worker
)).ready
=
1077 message
.ready
as boolean
1078 if (this.emitter
!= null && this.ready
) {
1079 this.emitter
.emit(PoolEvents
.ready
, this.info
)
1083 private handleTaskExecutionResponse (message
: MessageValue
<Response
>): void {
1084 const promiseResponse
= this.promiseResponseMap
.get(message
.id
as string)
1085 if (promiseResponse
!= null) {
1086 if (message
.taskError
!= null) {
1087 if (this.emitter
!= null) {
1088 this.emitter
.emit(PoolEvents
.taskError
, message
.taskError
)
1090 promiseResponse
.reject(message
.taskError
.message
)
1092 promiseResponse
.resolve(message
.data
as Response
)
1094 this.afterTaskExecutionHook(promiseResponse
.worker
, message
)
1095 this.promiseResponseMap
.delete(message
.id
as string)
1096 const workerNodeKey
= this.getWorkerNodeKey(promiseResponse
.worker
)
1098 this.opts
.enableTasksQueue
=== true &&
1099 this.tasksQueueSize(workerNodeKey
) > 0
1103 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1106 this.workerChoiceStrategyContext
.update(workerNodeKey
)
1110 private checkAndEmitEvents (): void {
1111 if (this.emitter
!= null) {
1113 this.emitter
.emit(PoolEvents
.busy
, this.info
)
1115 if (this.type === PoolTypes
.dynamic
&& this.full
) {
1116 this.emitter
.emit(PoolEvents
.full
, this.info
)
1122 * Gets the worker information.
1124 * @param workerNodeKey - The worker node key.
1126 private getWorkerInfo (workerNodeKey
: number): WorkerInfo
{
1127 return this.workerNodes
[workerNodeKey
].info
1131 * Pushes the given worker in the pool worker nodes.
1133 * @param worker - The worker.
1134 * @returns The worker nodes length.
1136 private pushWorkerNode (worker
: Worker
): number {
1137 const workerNode
= new WorkerNode(worker
, this.worker
)
1138 if (this.starting
) {
1139 workerNode
.info
.ready
= true
1141 return this.workerNodes
.push(new WorkerNode(worker
, this.worker
))
1145 * Removes the given worker from the pool worker nodes.
1147 * @param worker - The worker.
1149 private removeWorkerNode (worker
: Worker
): void {
1150 const workerNodeKey
= this.getWorkerNodeKey(worker
)
1151 if (workerNodeKey
!== -1) {
1152 this.workerNodes
.splice(workerNodeKey
, 1)
1153 this.workerChoiceStrategyContext
.remove(workerNodeKey
)
1157 private executeTask (workerNodeKey
: number, task
: Task
<Data
>): void {
1158 this.beforeTaskExecutionHook(workerNodeKey
, task
)
1159 this.sendToWorker(this.workerNodes
[workerNodeKey
].worker
, task
)
1162 private enqueueTask (workerNodeKey
: number, task
: Task
<Data
>): number {
1163 return this.workerNodes
[workerNodeKey
].enqueueTask(task
)
1166 private dequeueTask (workerNodeKey
: number): Task
<Data
> | undefined {
1167 return this.workerNodes
[workerNodeKey
].dequeueTask()
1170 private tasksQueueSize (workerNodeKey
: number): number {
1171 return this.workerNodes
[workerNodeKey
].tasksQueueSize()
1174 private flushTasksQueue (workerNodeKey
: number): void {
1175 while (this.tasksQueueSize(workerNodeKey
) > 0) {
1178 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1181 this.workerNodes
[workerNodeKey
].clearTasksQueue()
1184 private flushTasksQueues (): void {
1185 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
1186 this.flushTasksQueue(workerNodeKey
)
1190 private setWorkerStatistics (worker
: Worker
): void {
1191 this.sendToWorker(worker
, {
1194 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1196 elu
: this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1199 workerId
: this.getWorkerInfo(this.getWorkerNodeKey(worker
)).id
as number