1 import { randomUUID
} from
'node:crypto'
2 import { performance
} from
'node:perf_hooks'
5 PromiseResponseWrapper
,
7 } from
'../utility-types'
10 DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
,
17 import { KillBehaviors
} from
'../worker/worker-options'
26 type TasksQueueOptions
38 WorkerChoiceStrategies
,
39 type WorkerChoiceStrategy
,
40 type WorkerChoiceStrategyOptions
41 } from
'./selection-strategies/selection-strategies-types'
42 import { WorkerChoiceStrategyContext
} from
'./selection-strategies/worker-choice-strategy-context'
43 import { version
} from
'./version'
44 import { WorkerNode
} from
'./worker-node'
47 * Base class that implements some shared logic for all poolifier pools.
49 * @typeParam Worker - Type of worker which manages this pool.
50 * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
51 * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
53 export abstract class AbstractPool
<
54 Worker
extends IWorker
,
57 > implements IPool
<Worker
, Data
, Response
> {
59 public readonly workerNodes
: Array<IWorkerNode
<Worker
, Data
>> = []
62 public readonly emitter
?: PoolEmitter
65 * The execution response promise map.
67 * - `key`: The message id of each submitted task.
68 * - `value`: An object that contains the worker, the execution response promise resolve and reject callbacks.
70 * When we receive a message from the worker, we get a map entry with the promise resolve/reject bound to the message id.
72 protected promiseResponseMap
: Map
<
74 PromiseResponseWrapper
<Worker
, Response
>
75 > = new Map
<string, PromiseResponseWrapper
<Worker
, Response
>>()
78 * Worker choice strategy context referencing a worker choice algorithm implementation.
80 protected workerChoiceStrategyContext
: WorkerChoiceStrategyContext
<
87 * The start timestamp of the pool.
89 private readonly startTimestamp
92 * Constructs a new poolifier pool.
94 * @param numberOfWorkers - Number of workers that this pool should manage.
95 * @param filePath - Path to the worker file.
96 * @param opts - Options for the pool.
99 protected readonly numberOfWorkers
: number,
100 protected readonly filePath
: string,
101 protected readonly opts
: PoolOptions
<Worker
>
103 if (!this.isMain()) {
104 throw new Error('Cannot start a pool from a worker!')
106 this.checkNumberOfWorkers(this.numberOfWorkers
)
107 this.checkFilePath(this.filePath
)
108 this.checkPoolOptions(this.opts
)
110 this.chooseWorkerNode
= this.chooseWorkerNode
.bind(this)
111 this.executeTask
= this.executeTask
.bind(this)
112 this.enqueueTask
= this.enqueueTask
.bind(this)
113 this.checkAndEmitEvents
= this.checkAndEmitEvents
.bind(this)
115 if (this.opts
.enableEvents
=== true) {
116 this.emitter
= new PoolEmitter()
118 this.workerChoiceStrategyContext
= new WorkerChoiceStrategyContext
<
124 this.opts
.workerChoiceStrategy
,
125 this.opts
.workerChoiceStrategyOptions
130 while (this.workerNodes
.length
< this.numberOfWorkers
) {
131 this.createAndSetupWorker()
134 this.startTimestamp
= performance
.now()
137 private checkFilePath (filePath
: string): void {
140 (typeof filePath
=== 'string' && filePath
.trim().length
=== 0)
142 throw new Error('Please specify a file with a worker implementation')
146 private checkNumberOfWorkers (numberOfWorkers
: number): void {
147 if (numberOfWorkers
== null) {
149 'Cannot instantiate a pool without specifying the number of workers'
151 } else if (!Number.isSafeInteger(numberOfWorkers
)) {
153 'Cannot instantiate a pool with a non safe integer number of workers'
155 } else if (numberOfWorkers
< 0) {
156 throw new RangeError(
157 'Cannot instantiate a pool with a negative number of workers'
159 } else if (this.type === PoolTypes
.fixed
&& numberOfWorkers
=== 0) {
160 throw new RangeError('Cannot instantiate a fixed pool with zero worker')
164 protected checkDynamicPoolSize (min
: number, max
: number): void {
165 if (this.type === PoolTypes
.dynamic
) {
167 throw new RangeError(
168 'Cannot instantiate a dynamic pool with a maximum pool size inferior to the minimum pool size'
170 } else if (min
=== 0 && max
=== 0) {
171 throw new RangeError(
172 'Cannot instantiate a dynamic pool with a minimum pool size and a maximum pool size equal to zero'
174 } else if (min
=== max
) {
175 throw new RangeError(
176 'Cannot instantiate a dynamic pool with a minimum pool size equal to the maximum pool size. Use a fixed pool instead'
182 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
183 if (isPlainObject(opts
)) {
184 this.opts
.workerChoiceStrategy
=
185 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
186 this.checkValidWorkerChoiceStrategy(this.opts
.workerChoiceStrategy
)
187 this.opts
.workerChoiceStrategyOptions
=
188 opts
.workerChoiceStrategyOptions
??
189 DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
190 this.checkValidWorkerChoiceStrategyOptions(
191 this.opts
.workerChoiceStrategyOptions
193 this.opts
.restartWorkerOnError
= opts
.restartWorkerOnError
?? true
194 this.opts
.enableEvents
= opts
.enableEvents
?? true
195 this.opts
.enableTasksQueue
= opts
.enableTasksQueue
?? false
196 if (this.opts
.enableTasksQueue
) {
197 this.checkValidTasksQueueOptions(
198 opts
.tasksQueueOptions
as TasksQueueOptions
200 this.opts
.tasksQueueOptions
= this.buildTasksQueueOptions(
201 opts
.tasksQueueOptions
as TasksQueueOptions
205 throw new TypeError('Invalid pool options: must be a plain object')
209 private checkValidWorkerChoiceStrategy (
210 workerChoiceStrategy
: WorkerChoiceStrategy
212 if (!Object.values(WorkerChoiceStrategies
).includes(workerChoiceStrategy
)) {
214 `Invalid worker choice strategy '${workerChoiceStrategy}'`
219 private checkValidWorkerChoiceStrategyOptions (
220 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
222 if (!isPlainObject(workerChoiceStrategyOptions
)) {
224 'Invalid worker choice strategy options: must be a plain object'
228 workerChoiceStrategyOptions
.weights
!= null &&
229 Object.keys(workerChoiceStrategyOptions
.weights
).length
!== this.maxSize
232 'Invalid worker choice strategy options: must have a weight for each worker node'
236 workerChoiceStrategyOptions
.measurement
!= null &&
237 !Object.values(Measurements
).includes(
238 workerChoiceStrategyOptions
.measurement
242 `Invalid worker choice strategy options: invalid measurement '${workerChoiceStrategyOptions.measurement}'`
247 private checkValidTasksQueueOptions (
248 tasksQueueOptions
: TasksQueueOptions
250 if (tasksQueueOptions
!= null && !isPlainObject(tasksQueueOptions
)) {
251 throw new TypeError('Invalid tasks queue options: must be a plain object')
254 tasksQueueOptions
?.concurrency
!= null &&
255 !Number.isSafeInteger(tasksQueueOptions
.concurrency
)
258 'Invalid worker tasks concurrency: must be an integer'
262 tasksQueueOptions
?.concurrency
!= null &&
263 tasksQueueOptions
.concurrency
<= 0
266 `Invalid worker tasks concurrency '${tasksQueueOptions.concurrency}'`
272 public get
info (): PoolInfo
{
278 strategy
: this.opts
.workerChoiceStrategy
as WorkerChoiceStrategy
,
279 minSize
: this.minSize
,
280 maxSize
: this.maxSize
,
281 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
282 .runTime
.aggregate
&&
283 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
284 .waitTime
.aggregate
&& { utilization
: round(this.utilization
) }),
285 workerNodes
: this.workerNodes
.length
,
286 idleWorkerNodes
: this.workerNodes
.reduce(
287 (accumulator
, workerNode
) =>
288 workerNode
.usage
.tasks
.executing
=== 0
293 busyWorkerNodes
: this.workerNodes
.reduce(
294 (accumulator
, workerNode
) =>
295 workerNode
.usage
.tasks
.executing
> 0 ? accumulator
+ 1 : accumulator
,
298 executedTasks
: this.workerNodes
.reduce(
299 (accumulator
, workerNode
) =>
300 accumulator
+ workerNode
.usage
.tasks
.executed
,
303 executingTasks
: this.workerNodes
.reduce(
304 (accumulator
, workerNode
) =>
305 accumulator
+ workerNode
.usage
.tasks
.executing
,
308 queuedTasks
: this.workerNodes
.reduce(
309 (accumulator
, workerNode
) =>
310 accumulator
+ workerNode
.usage
.tasks
.queued
,
313 maxQueuedTasks
: this.workerNodes
.reduce(
314 (accumulator
, workerNode
) =>
315 accumulator
+ (workerNode
.usage
.tasks
?.maxQueued
?? 0),
318 failedTasks
: this.workerNodes
.reduce(
319 (accumulator
, workerNode
) =>
320 accumulator
+ workerNode
.usage
.tasks
.failed
,
323 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
324 .runTime
.aggregate
&& {
328 ...this.workerNodes
.map(
329 workerNode
=> workerNode
.usage
.runTime
?.minimum
?? Infinity
335 ...this.workerNodes
.map(
336 workerNode
=> workerNode
.usage
.runTime
?.maximum
?? -Infinity
341 this.workerNodes
.reduce(
342 (accumulator
, workerNode
) =>
343 accumulator
+ (workerNode
.usage
.runTime
?.aggregate
?? 0),
346 this.workerNodes
.reduce(
347 (accumulator
, workerNode
) =>
348 accumulator
+ (workerNode
.usage
.tasks
?.executed
?? 0),
352 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
356 this.workerNodes
.map(
357 workerNode
=> workerNode
.usage
.runTime
?.median
?? 0
364 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
365 .waitTime
.aggregate
&& {
369 ...this.workerNodes
.map(
370 workerNode
=> workerNode
.usage
.waitTime
?.minimum
?? Infinity
376 ...this.workerNodes
.map(
377 workerNode
=> workerNode
.usage
.waitTime
?.maximum
?? -Infinity
382 this.workerNodes
.reduce(
383 (accumulator
, workerNode
) =>
384 accumulator
+ (workerNode
.usage
.waitTime
?.aggregate
?? 0),
387 this.workerNodes
.reduce(
388 (accumulator
, workerNode
) =>
389 accumulator
+ (workerNode
.usage
.tasks
?.executed
?? 0),
393 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
394 .waitTime
.median
&& {
397 this.workerNodes
.map(
398 workerNode
=> workerNode
.usage
.waitTime
?.median
?? 0
408 private get
starting (): boolean {
410 this.workerNodes
.length
< this.minSize
||
411 (this.workerNodes
.length
>= this.minSize
&&
412 this.workerNodes
.some(workerNode
=> !workerNode
.info
.ready
))
416 private get
ready (): boolean {
418 this.workerNodes
.length
>= this.minSize
&&
419 this.workerNodes
.every(workerNode
=> workerNode
.info
.ready
)
424 * Gets the approximate pool utilization.
426 * @returns The pool utilization.
428 private get
utilization (): number {
429 const poolTimeCapacity
=
430 (performance
.now() - this.startTimestamp
) * this.maxSize
431 const totalTasksRunTime
= this.workerNodes
.reduce(
432 (accumulator
, workerNode
) =>
433 accumulator
+ (workerNode
.usage
.runTime
?.aggregate
?? 0),
436 const totalTasksWaitTime
= this.workerNodes
.reduce(
437 (accumulator
, workerNode
) =>
438 accumulator
+ (workerNode
.usage
.waitTime
?.aggregate
?? 0),
441 return (totalTasksRunTime
+ totalTasksWaitTime
) / poolTimeCapacity
447 * If it is `'dynamic'`, it provides the `max` property.
449 protected abstract get
type (): PoolType
452 * Gets the worker type.
454 protected abstract get
worker (): WorkerType
459 protected abstract get
minSize (): number
464 protected abstract get
maxSize (): number
467 * Get the worker given its id.
469 * @param workerId - The worker id.
470 * @returns The worker if found in the pool worker nodes, `undefined` otherwise.
472 private getWorkerById (workerId
: number): Worker
| undefined {
473 return this.workerNodes
.find(workerNode
=> workerNode
.info
.id
=== workerId
)
478 * Checks if the worker id sent in the received message from a worker is valid.
480 * @param message - The received message.
481 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the worker id is invalid.
483 private checkMessageWorkerId (message
: MessageValue
<Response
>): void {
485 message
.workerId
!= null &&
486 this.getWorkerById(message
.workerId
) == null
489 `Worker message received from unknown worker '${message.workerId}'`
495 * Gets the given worker its worker node key.
497 * @param worker - The worker.
498 * @returns The worker node key if found in the pool worker nodes, `-1` otherwise.
500 private getWorkerNodeKey (worker
: Worker
): number {
501 return this.workerNodes
.findIndex(
502 workerNode
=> workerNode
.worker
=== worker
507 public setWorkerChoiceStrategy (
508 workerChoiceStrategy
: WorkerChoiceStrategy
,
509 workerChoiceStrategyOptions
?: WorkerChoiceStrategyOptions
511 this.checkValidWorkerChoiceStrategy(workerChoiceStrategy
)
512 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
513 this.workerChoiceStrategyContext
.setWorkerChoiceStrategy(
514 this.opts
.workerChoiceStrategy
516 if (workerChoiceStrategyOptions
!= null) {
517 this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
519 for (const workerNode
of this.workerNodes
) {
520 workerNode
.resetUsage()
521 this.setWorkerStatistics(workerNode
.worker
)
526 public setWorkerChoiceStrategyOptions (
527 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
529 this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
530 this.opts
.workerChoiceStrategyOptions
= workerChoiceStrategyOptions
531 this.workerChoiceStrategyContext
.setOptions(
532 this.opts
.workerChoiceStrategyOptions
537 public enableTasksQueue (
539 tasksQueueOptions
?: TasksQueueOptions
541 if (this.opts
.enableTasksQueue
=== true && !enable
) {
542 this.flushTasksQueues()
544 this.opts
.enableTasksQueue
= enable
545 this.setTasksQueueOptions(tasksQueueOptions
as TasksQueueOptions
)
549 public setTasksQueueOptions (tasksQueueOptions
: TasksQueueOptions
): void {
550 if (this.opts
.enableTasksQueue
=== true) {
551 this.checkValidTasksQueueOptions(tasksQueueOptions
)
552 this.opts
.tasksQueueOptions
=
553 this.buildTasksQueueOptions(tasksQueueOptions
)
554 } else if (this.opts
.tasksQueueOptions
!= null) {
555 delete this.opts
.tasksQueueOptions
559 private buildTasksQueueOptions (
560 tasksQueueOptions
: TasksQueueOptions
561 ): TasksQueueOptions
{
563 concurrency
: tasksQueueOptions
?.concurrency
?? 1
568 * Whether the pool is full or not.
570 * The pool filling boolean status.
572 protected get
full (): boolean {
573 return this.workerNodes
.length
>= this.maxSize
577 * Whether the pool is busy or not.
579 * The pool busyness boolean status.
581 protected abstract get
busy (): boolean
584 * Whether worker nodes are executing at least one task.
586 * @returns Worker nodes busyness boolean status.
588 protected internalBusy (): boolean {
590 this.workerNodes
.findIndex(workerNode
=> {
591 return workerNode
.usage
.tasks
.executing
=== 0
597 public async execute (data
?: Data
, name
?: string): Promise
<Response
> {
598 const timestamp
= performance
.now()
599 const workerNodeKey
= this.chooseWorkerNode()
600 const submittedTask
: Task
<Data
> = {
601 name
: name
?? DEFAULT_TASK_NAME
,
602 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
603 data
: data
?? ({} as Data
),
605 workerId
: this.getWorkerInfo(workerNodeKey
).id
as number,
608 const res
= new Promise
<Response
>((resolve
, reject
) => {
609 this.promiseResponseMap
.set(submittedTask
.id
as string, {
612 worker
: this.workerNodes
[workerNodeKey
].worker
616 this.opts
.enableTasksQueue
=== true &&
618 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
>=
619 ((this.opts
.tasksQueueOptions
as TasksQueueOptions
)
620 .concurrency
as number))
622 this.enqueueTask(workerNodeKey
, submittedTask
)
624 this.executeTask(workerNodeKey
, submittedTask
)
626 this.checkAndEmitEvents()
627 // eslint-disable-next-line @typescript-eslint/return-await
632 public async destroy (): Promise
<void> {
634 this.workerNodes
.map(async (workerNode
, workerNodeKey
) => {
635 this.flushTasksQueue(workerNodeKey
)
636 // FIXME: wait for tasks to be finished
637 const workerExitPromise
= new Promise
<void>(resolve
=> {
638 workerNode
.worker
.on('exit', () => {
642 await this.destroyWorker(workerNode
.worker
)
643 await workerExitPromise
649 * Terminates the given worker.
651 * @param worker - A worker within `workerNodes`.
653 protected abstract destroyWorker (worker
: Worker
): void | Promise
<void>
656 * Setup hook to execute code before worker nodes are created in the abstract constructor.
661 protected setupHook (): void {
662 // Intentionally empty
666 * Should return whether the worker is the main worker or not.
668 protected abstract isMain (): boolean
671 * Hook executed before the worker task execution.
674 * @param workerNodeKey - The worker node key.
675 * @param task - The task to execute.
677 protected beforeTaskExecutionHook (
678 workerNodeKey
: number,
681 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
682 ++workerUsage
.tasks
.executing
683 this.updateWaitTimeWorkerUsage(workerUsage
, task
)
684 const taskWorkerUsage
= this.workerNodes
[workerNodeKey
].getTaskWorkerUsage(
687 ++taskWorkerUsage
.tasks
.executing
688 this.updateWaitTimeWorkerUsage(taskWorkerUsage
, task
)
692 * Hook executed after the worker task execution.
695 * @param worker - The worker.
696 * @param message - The received message.
698 protected afterTaskExecutionHook (
700 message
: MessageValue
<Response
>
702 const workerNodeKey
= this.getWorkerNodeKey(worker
)
703 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
704 this.updateTaskStatisticsWorkerUsage(workerUsage
, message
)
705 this.updateRunTimeWorkerUsage(workerUsage
, message
)
706 this.updateEluWorkerUsage(workerUsage
, message
)
707 const taskWorkerUsage
= this.workerNodes
[workerNodeKey
].getTaskWorkerUsage(
708 message
.taskPerformance
?.name
?? DEFAULT_TASK_NAME
710 this.updateTaskStatisticsWorkerUsage(taskWorkerUsage
, message
)
711 this.updateRunTimeWorkerUsage(taskWorkerUsage
, message
)
712 this.updateEluWorkerUsage(taskWorkerUsage
, message
)
715 private updateTaskStatisticsWorkerUsage (
716 workerUsage
: WorkerUsage
,
717 message
: MessageValue
<Response
>
719 const workerTaskStatistics
= workerUsage
.tasks
720 --workerTaskStatistics
.executing
721 if (message
.taskError
== null) {
722 ++workerTaskStatistics
.executed
724 ++workerTaskStatistics
.failed
728 private updateRunTimeWorkerUsage (
729 workerUsage
: WorkerUsage
,
730 message
: MessageValue
<Response
>
733 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
736 const taskRunTime
= message
.taskPerformance
?.runTime
?? 0
737 workerUsage
.runTime
.aggregate
=
738 (workerUsage
.runTime
.aggregate
?? 0) + taskRunTime
739 workerUsage
.runTime
.minimum
= Math.min(
741 workerUsage
.runTime
?.minimum
?? Infinity
743 workerUsage
.runTime
.maximum
= Math.max(
745 workerUsage
.runTime
?.maximum
?? -Infinity
748 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
750 workerUsage
.tasks
.executed
!== 0
752 workerUsage
.runTime
.average
=
753 workerUsage
.runTime
.aggregate
/ workerUsage
.tasks
.executed
756 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
758 message
.taskPerformance
?.runTime
!= null
760 workerUsage
.runTime
.history
.push(message
.taskPerformance
.runTime
)
761 workerUsage
.runTime
.median
= median(workerUsage
.runTime
.history
)
766 private updateWaitTimeWorkerUsage (
767 workerUsage
: WorkerUsage
,
770 const timestamp
= performance
.now()
771 const taskWaitTime
= timestamp
- (task
.timestamp
?? timestamp
)
773 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().waitTime
776 workerUsage
.waitTime
.aggregate
=
777 (workerUsage
.waitTime
?.aggregate
?? 0) + taskWaitTime
778 workerUsage
.waitTime
.minimum
= Math.min(
780 workerUsage
.waitTime
?.minimum
?? Infinity
782 workerUsage
.waitTime
.maximum
= Math.max(
784 workerUsage
.waitTime
?.maximum
?? -Infinity
787 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
789 workerUsage
.tasks
.executed
!== 0
791 workerUsage
.waitTime
.average
=
792 workerUsage
.waitTime
.aggregate
/ workerUsage
.tasks
.executed
795 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
798 workerUsage
.waitTime
.history
.push(taskWaitTime
)
799 workerUsage
.waitTime
.median
= median(workerUsage
.waitTime
.history
)
804 private updateEluWorkerUsage (
805 workerUsage
: WorkerUsage
,
806 message
: MessageValue
<Response
>
809 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
812 if (message
.taskPerformance
?.elu
!= null) {
813 workerUsage
.elu
.idle
.aggregate
=
814 (workerUsage
.elu
.idle
?.aggregate
?? 0) +
815 message
.taskPerformance
.elu
.idle
816 workerUsage
.elu
.active
.aggregate
=
817 (workerUsage
.elu
.active
?.aggregate
?? 0) +
818 message
.taskPerformance
.elu
.active
819 if (workerUsage
.elu
.utilization
!= null) {
820 workerUsage
.elu
.utilization
=
821 (workerUsage
.elu
.utilization
+
822 message
.taskPerformance
.elu
.utilization
) /
825 workerUsage
.elu
.utilization
= message
.taskPerformance
.elu
.utilization
827 workerUsage
.elu
.idle
.minimum
= Math.min(
828 message
.taskPerformance
.elu
.idle
,
829 workerUsage
.elu
.idle
?.minimum
?? Infinity
831 workerUsage
.elu
.idle
.maximum
= Math.max(
832 message
.taskPerformance
.elu
.idle
,
833 workerUsage
.elu
.idle
?.maximum
?? -Infinity
835 workerUsage
.elu
.active
.minimum
= Math.min(
836 message
.taskPerformance
.elu
.active
,
837 workerUsage
.elu
.active
?.minimum
?? Infinity
839 workerUsage
.elu
.active
.maximum
= Math.max(
840 message
.taskPerformance
.elu
.active
,
841 workerUsage
.elu
.active
?.maximum
?? -Infinity
844 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
846 workerUsage
.tasks
.executed
!== 0
848 workerUsage
.elu
.idle
.average
=
849 workerUsage
.elu
.idle
.aggregate
/ workerUsage
.tasks
.executed
850 workerUsage
.elu
.active
.average
=
851 workerUsage
.elu
.active
.aggregate
/ workerUsage
.tasks
.executed
854 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
857 workerUsage
.elu
.idle
.history
.push(message
.taskPerformance
.elu
.idle
)
858 workerUsage
.elu
.active
.history
.push(
859 message
.taskPerformance
.elu
.active
861 workerUsage
.elu
.idle
.median
= median(workerUsage
.elu
.idle
.history
)
862 workerUsage
.elu
.active
.median
= median(workerUsage
.elu
.active
.history
)
869 * Chooses a worker node for the next task.
871 * The default worker choice strategy uses a round robin algorithm to distribute the tasks.
873 * @returns The worker node key
875 private chooseWorkerNode (): number {
876 if (this.shallCreateDynamicWorker()) {
877 const worker
= this.createAndSetupDynamicWorker()
879 this.workerChoiceStrategyContext
.getStrategyPolicy().useDynamicWorker
881 return this.getWorkerNodeKey(worker
)
884 return this.workerChoiceStrategyContext
.execute()
888 * Conditions for dynamic worker creation.
890 * @returns Whether to create a dynamic worker or not.
892 private shallCreateDynamicWorker (): boolean {
893 return this.type === PoolTypes
.dynamic
&& !this.full
&& this.internalBusy()
897 * Sends a message to the given worker.
899 * @param worker - The worker which should receive the message.
900 * @param message - The message.
902 protected abstract sendToWorker (
904 message
: MessageValue
<Data
>
908 * Registers a listener callback on the given worker.
910 * @param worker - The worker which should register a listener.
911 * @param listener - The message listener callback.
913 private registerWorkerMessageListener
<Message
extends Data
| Response
>(
915 listener
: (message
: MessageValue
<Message
>) => void
917 worker
.on('message', listener
as MessageHandler
<Worker
>)
921 * Creates a new worker.
923 * @returns Newly created worker.
925 protected abstract createWorker (): Worker
928 * Function that can be hooked up when a worker has been newly created and moved to the pool worker nodes.
931 * @param worker - The newly created worker.
933 protected afterWorkerSetup (worker
: Worker
): void {
934 // Listen to worker messages.
935 this.registerWorkerMessageListener(worker
, this.workerListener())
936 // Send startup message to worker.
937 this.sendToWorker(worker
, {
939 workerId
: this.getWorkerInfo(this.getWorkerNodeKey(worker
)).id
as number
941 // Setup worker task statistics computation.
942 this.setWorkerStatistics(worker
)
946 * Creates a new worker and sets it up completely in the pool worker nodes.
948 * @returns New, completely set up worker.
950 protected createAndSetupWorker (): Worker
{
951 const worker
= this.createWorker()
953 worker
.on('message', this.opts
.messageHandler
?? EMPTY_FUNCTION
)
954 worker
.on('error', this.opts
.errorHandler
?? EMPTY_FUNCTION
)
955 worker
.on('error', error
=> {
956 if (this.emitter
!= null) {
957 this.emitter
.emit(PoolEvents
.error
, error
)
959 if (this.opts
.restartWorkerOnError
=== true && !this.starting
) {
960 if (this.getWorkerInfo(this.getWorkerNodeKey(worker
)).dynamic
) {
961 this.createAndSetupDynamicWorker()
963 this.createAndSetupWorker()
966 if (this.opts
.enableTasksQueue
=== true) {
967 this.redistributeQueuedTasks(worker
)
970 worker
.on('online', this.opts
.onlineHandler
?? EMPTY_FUNCTION
)
971 worker
.on('exit', this.opts
.exitHandler
?? EMPTY_FUNCTION
)
972 worker
.once('exit', () => {
973 this.removeWorkerNode(worker
)
976 this.pushWorkerNode(worker
)
978 this.afterWorkerSetup(worker
)
983 private redistributeQueuedTasks (worker
: Worker
): void {
984 const workerNodeKey
= this.getWorkerNodeKey(worker
)
985 while (this.tasksQueueSize(workerNodeKey
) > 0) {
986 let targetWorkerNodeKey
: number = workerNodeKey
987 let minQueuedTasks
= Infinity
988 for (const [workerNodeId
, workerNode
] of this.workerNodes
.entries()) {
990 workerNodeId
!== workerNodeKey
&&
991 workerNode
.usage
.tasks
.queued
=== 0
993 targetWorkerNodeKey
= workerNodeId
997 workerNodeId
!== workerNodeKey
&&
998 workerNode
.usage
.tasks
.queued
< minQueuedTasks
1000 minQueuedTasks
= workerNode
.usage
.tasks
.queued
1001 targetWorkerNodeKey
= workerNodeId
1005 targetWorkerNodeKey
,
1006 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1012 * Creates a new dynamic worker and sets it up completely in the pool worker nodes.
1014 * @returns New, completely set up dynamic worker.
1016 protected createAndSetupDynamicWorker (): Worker
{
1017 const worker
= this.createAndSetupWorker()
1018 this.registerWorkerMessageListener(worker
, message
=> {
1019 const workerNodeKey
= this.getWorkerNodeKey(worker
)
1021 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
1022 (message
.kill
!= null &&
1023 ((this.opts
.enableTasksQueue
=== false &&
1024 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
=== 0) ||
1025 (this.opts
.enableTasksQueue
=== true &&
1026 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
=== 0 &&
1027 this.tasksQueueSize(workerNodeKey
) === 0)))
1029 // Kill message received from the worker: no new tasks are submitted to that worker for a while ( > maxInactiveTime)
1030 void (this.destroyWorker(worker
) as Promise
<void>)
1033 const workerInfo
= this.getWorkerInfo(this.getWorkerNodeKey(worker
))
1034 workerInfo
.dynamic
= true
1035 this.sendToWorker(worker
, {
1037 workerId
: workerInfo
.id
as number
1043 * This function is the listener registered for each worker message.
1045 * @returns The listener function to execute when a message is received from a worker.
1047 protected workerListener (): (message
: MessageValue
<Response
>) => void {
1049 this.checkMessageWorkerId(message
)
1050 if (message
.ready
!= null && message
.workerId
!= null) {
1051 // Worker ready message received
1052 this.handleWorkerReadyMessage(message
)
1053 } else if (message
.id
!= null) {
1054 // Task execution response received
1055 this.handleTaskExecutionResponse(message
)
1060 private handleWorkerReadyMessage (message
: MessageValue
<Response
>): void {
1061 const worker
= this.getWorkerById(message
.workerId
)
1062 this.getWorkerInfo(this.getWorkerNodeKey(worker
as Worker
)).ready
=
1063 message
.ready
as boolean
1064 if (this.emitter
!= null && this.ready
) {
1065 this.emitter
.emit(PoolEvents
.ready
, this.info
)
1069 private handleTaskExecutionResponse (message
: MessageValue
<Response
>): void {
1070 const promiseResponse
= this.promiseResponseMap
.get(message
.id
as string)
1071 if (promiseResponse
!= null) {
1072 if (message
.taskError
!= null) {
1073 if (this.emitter
!= null) {
1074 this.emitter
.emit(PoolEvents
.taskError
, message
.taskError
)
1076 promiseResponse
.reject(message
.taskError
.message
)
1078 promiseResponse
.resolve(message
.data
as Response
)
1080 this.afterTaskExecutionHook(promiseResponse
.worker
, message
)
1081 this.promiseResponseMap
.delete(message
.id
as string)
1082 const workerNodeKey
= this.getWorkerNodeKey(promiseResponse
.worker
)
1084 this.opts
.enableTasksQueue
=== true &&
1085 this.tasksQueueSize(workerNodeKey
) > 0
1089 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1092 this.workerChoiceStrategyContext
.update(workerNodeKey
)
1096 private checkAndEmitEvents (): void {
1097 if (this.emitter
!= null) {
1099 this.emitter
.emit(PoolEvents
.busy
, this.info
)
1101 if (this.type === PoolTypes
.dynamic
&& this.full
) {
1102 this.emitter
.emit(PoolEvents
.full
, this.info
)
1108 * Gets the worker information.
1110 * @param workerNodeKey - The worker node key.
1112 private getWorkerInfo (workerNodeKey
: number): WorkerInfo
{
1113 return this.workerNodes
[workerNodeKey
].info
1117 * Pushes the given worker in the pool worker nodes.
1119 * @param worker - The worker.
1120 * @returns The worker nodes length.
1122 private pushWorkerNode (worker
: Worker
): number {
1123 return this.workerNodes
.push(new WorkerNode(worker
, this.worker
))
1127 * Removes the given worker from the pool worker nodes.
1129 * @param worker - The worker.
1131 private removeWorkerNode (worker
: Worker
): void {
1132 const workerNodeKey
= this.getWorkerNodeKey(worker
)
1133 if (workerNodeKey
!== -1) {
1134 this.workerNodes
.splice(workerNodeKey
, 1)
1135 this.workerChoiceStrategyContext
.remove(workerNodeKey
)
1139 private executeTask (workerNodeKey
: number, task
: Task
<Data
>): void {
1140 this.beforeTaskExecutionHook(workerNodeKey
, task
)
1141 this.sendToWorker(this.workerNodes
[workerNodeKey
].worker
, task
)
1144 private enqueueTask (workerNodeKey
: number, task
: Task
<Data
>): number {
1145 return this.workerNodes
[workerNodeKey
].enqueueTask(task
)
1148 private dequeueTask (workerNodeKey
: number): Task
<Data
> | undefined {
1149 return this.workerNodes
[workerNodeKey
].dequeueTask()
1152 private tasksQueueSize (workerNodeKey
: number): number {
1153 return this.workerNodes
[workerNodeKey
].tasksQueueSize()
1156 private flushTasksQueue (workerNodeKey
: number): void {
1157 while (this.tasksQueueSize(workerNodeKey
) > 0) {
1160 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1163 this.workerNodes
[workerNodeKey
].clearTasksQueue()
1166 private flushTasksQueues (): void {
1167 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
1168 this.flushTasksQueue(workerNodeKey
)
1172 private setWorkerStatistics (worker
: Worker
): void {
1173 this.sendToWorker(worker
, {
1176 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1178 elu
: this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1181 workerId
: this.getWorkerInfo(this.getWorkerNodeKey(worker
)).id
as number