1 import { randomUUID
} from
'node:crypto'
2 import { performance
} from
'node:perf_hooks'
3 import type { MessageValue
, PromiseResponseWrapper
} from
'../utility-types'
6 DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
,
13 import { KillBehaviors
} from
'../worker/worker-options'
22 type TasksQueueOptions
35 WorkerChoiceStrategies
,
36 type WorkerChoiceStrategy
,
37 type WorkerChoiceStrategyOptions
38 } from
'./selection-strategies/selection-strategies-types'
39 import { WorkerChoiceStrategyContext
} from
'./selection-strategies/worker-choice-strategy-context'
40 import { version
} from
'./version'
41 import { WorkerNode
} from
'./worker-node'
44 * Base class that implements some shared logic for all poolifier pools.
46 * @typeParam Worker - Type of worker which manages this pool.
47 * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
48 * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
50 export abstract class AbstractPool
<
51 Worker
extends IWorker
,
54 > implements IPool
<Worker
, Data
, Response
> {
56 public readonly workerNodes
: Array<IWorkerNode
<Worker
, Data
>> = []
59 public readonly emitter
?: PoolEmitter
62 * The execution response promise map.
64 * - `key`: The message id of each submitted task.
65 * - `value`: An object that contains the worker, the execution response promise resolve and reject callbacks.
67 * When we receive a message from the worker, we get a map entry with the promise resolve/reject bound to the message id.
69 protected promiseResponseMap
: Map
<
71 PromiseResponseWrapper
<Worker
, Response
>
72 > = new Map
<string, PromiseResponseWrapper
<Worker
, Response
>>()
75 * Worker choice strategy context referencing a worker choice algorithm implementation.
77 protected workerChoiceStrategyContext
: WorkerChoiceStrategyContext
<
84 * The start timestamp of the pool.
86 private readonly startTimestamp
89 * Constructs a new poolifier pool.
91 * @param numberOfWorkers - Number of workers that this pool should manage.
92 * @param filePath - Path to the worker file.
93 * @param opts - Options for the pool.
96 protected readonly numberOfWorkers
: number,
97 protected readonly filePath
: string,
98 protected readonly opts
: PoolOptions
<Worker
>
100 if (!this.isMain()) {
101 throw new Error('Cannot start a pool from a worker!')
103 this.checkNumberOfWorkers(this.numberOfWorkers
)
104 this.checkFilePath(this.filePath
)
105 this.checkPoolOptions(this.opts
)
107 this.chooseWorkerNode
= this.chooseWorkerNode
.bind(this)
108 this.executeTask
= this.executeTask
.bind(this)
109 this.enqueueTask
= this.enqueueTask
.bind(this)
110 this.checkAndEmitEvents
= this.checkAndEmitEvents
.bind(this)
112 if (this.opts
.enableEvents
=== true) {
113 this.emitter
= new PoolEmitter()
115 this.workerChoiceStrategyContext
= new WorkerChoiceStrategyContext
<
121 this.opts
.workerChoiceStrategy
,
122 this.opts
.workerChoiceStrategyOptions
127 while (this.workerNodes
.length
< this.numberOfWorkers
) {
128 this.createAndSetupWorker()
131 this.startTimestamp
= performance
.now()
134 private checkFilePath (filePath
: string): void {
137 (typeof filePath
=== 'string' && filePath
.trim().length
=== 0)
139 throw new Error('Please specify a file with a worker implementation')
143 private checkNumberOfWorkers (numberOfWorkers
: number): void {
144 if (numberOfWorkers
== null) {
146 'Cannot instantiate a pool without specifying the number of workers'
148 } else if (!Number.isSafeInteger(numberOfWorkers
)) {
150 'Cannot instantiate a pool with a non safe integer number of workers'
152 } else if (numberOfWorkers
< 0) {
153 throw new RangeError(
154 'Cannot instantiate a pool with a negative number of workers'
156 } else if (this.type === PoolTypes
.fixed
&& numberOfWorkers
=== 0) {
157 throw new RangeError('Cannot instantiate a fixed pool with zero worker')
161 protected checkDynamicPoolSize (min
: number, max
: number): void {
162 if (this.type === PoolTypes
.dynamic
) {
164 throw new RangeError(
165 'Cannot instantiate a dynamic pool with a maximum pool size inferior to the minimum pool size'
167 } else if (min
=== 0 && max
=== 0) {
168 throw new RangeError(
169 'Cannot instantiate a dynamic pool with a minimum pool size and a maximum pool size equal to zero'
171 } else if (min
=== max
) {
172 throw new RangeError(
173 'Cannot instantiate a dynamic pool with a minimum pool size equal to the maximum pool size. Use a fixed pool instead'
179 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
180 if (isPlainObject(opts
)) {
181 this.opts
.workerChoiceStrategy
=
182 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
183 this.checkValidWorkerChoiceStrategy(this.opts
.workerChoiceStrategy
)
184 this.opts
.workerChoiceStrategyOptions
=
185 opts
.workerChoiceStrategyOptions
??
186 DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
187 this.checkValidWorkerChoiceStrategyOptions(
188 this.opts
.workerChoiceStrategyOptions
190 this.opts
.restartWorkerOnError
= opts
.restartWorkerOnError
?? true
191 this.opts
.enableEvents
= opts
.enableEvents
?? true
192 this.opts
.enableTasksQueue
= opts
.enableTasksQueue
?? false
193 if (this.opts
.enableTasksQueue
) {
194 this.checkValidTasksQueueOptions(
195 opts
.tasksQueueOptions
as TasksQueueOptions
197 this.opts
.tasksQueueOptions
= this.buildTasksQueueOptions(
198 opts
.tasksQueueOptions
as TasksQueueOptions
202 throw new TypeError('Invalid pool options: must be a plain object')
206 private checkValidWorkerChoiceStrategy (
207 workerChoiceStrategy
: WorkerChoiceStrategy
209 if (!Object.values(WorkerChoiceStrategies
).includes(workerChoiceStrategy
)) {
211 `Invalid worker choice strategy '${workerChoiceStrategy}'`
216 private checkValidWorkerChoiceStrategyOptions (
217 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
219 if (!isPlainObject(workerChoiceStrategyOptions
)) {
221 'Invalid worker choice strategy options: must be a plain object'
225 workerChoiceStrategyOptions
.weights
!= null &&
226 Object.keys(workerChoiceStrategyOptions
.weights
).length
!== this.maxSize
229 'Invalid worker choice strategy options: must have a weight for each worker node'
233 workerChoiceStrategyOptions
.measurement
!= null &&
234 !Object.values(Measurements
).includes(
235 workerChoiceStrategyOptions
.measurement
239 `Invalid worker choice strategy options: invalid measurement '${workerChoiceStrategyOptions.measurement}'`
244 private checkValidTasksQueueOptions (
245 tasksQueueOptions
: TasksQueueOptions
247 if (tasksQueueOptions
!= null && !isPlainObject(tasksQueueOptions
)) {
248 throw new TypeError('Invalid tasks queue options: must be a plain object')
251 tasksQueueOptions
?.concurrency
!= null &&
252 !Number.isSafeInteger(tasksQueueOptions
.concurrency
)
255 'Invalid worker tasks concurrency: must be an integer'
259 tasksQueueOptions
?.concurrency
!= null &&
260 tasksQueueOptions
.concurrency
<= 0
263 `Invalid worker tasks concurrency '${tasksQueueOptions.concurrency}'`
269 public get
info (): PoolInfo
{
275 strategy
: this.opts
.workerChoiceStrategy
as WorkerChoiceStrategy
,
276 minSize
: this.minSize
,
277 maxSize
: this.maxSize
,
278 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
279 .runTime
.aggregate
&&
280 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
281 .waitTime
.aggregate
&& { utilization
: round(this.utilization
) }),
282 workerNodes
: this.workerNodes
.length
,
283 idleWorkerNodes
: this.workerNodes
.reduce(
284 (accumulator
, workerNode
) =>
285 workerNode
.usage
.tasks
.executing
=== 0
290 busyWorkerNodes
: this.workerNodes
.reduce(
291 (accumulator
, workerNode
) =>
292 workerNode
.usage
.tasks
.executing
> 0 ? accumulator
+ 1 : accumulator
,
295 executedTasks
: this.workerNodes
.reduce(
296 (accumulator
, workerNode
) =>
297 accumulator
+ workerNode
.usage
.tasks
.executed
,
300 executingTasks
: this.workerNodes
.reduce(
301 (accumulator
, workerNode
) =>
302 accumulator
+ workerNode
.usage
.tasks
.executing
,
305 queuedTasks
: this.workerNodes
.reduce(
306 (accumulator
, workerNode
) =>
307 accumulator
+ workerNode
.usage
.tasks
.queued
,
310 maxQueuedTasks
: this.workerNodes
.reduce(
311 (accumulator
, workerNode
) =>
312 accumulator
+ (workerNode
.usage
.tasks
?.maxQueued
?? 0),
315 failedTasks
: this.workerNodes
.reduce(
316 (accumulator
, workerNode
) =>
317 accumulator
+ workerNode
.usage
.tasks
.failed
,
320 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
321 .runTime
.aggregate
&& {
325 ...this.workerNodes
.map(
326 workerNode
=> workerNode
.usage
.runTime
?.minimum
?? Infinity
332 ...this.workerNodes
.map(
333 workerNode
=> workerNode
.usage
.runTime
?.maximum
?? -Infinity
338 this.workerNodes
.reduce(
339 (accumulator
, workerNode
) =>
340 accumulator
+ (workerNode
.usage
.runTime
?.aggregate
?? 0),
343 this.workerNodes
.reduce(
344 (accumulator
, workerNode
) =>
345 accumulator
+ (workerNode
.usage
.tasks
?.executed
?? 0),
349 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
353 this.workerNodes
.map(
354 workerNode
=> workerNode
.usage
.runTime
?.median
?? 0
361 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
362 .waitTime
.aggregate
&& {
366 ...this.workerNodes
.map(
367 workerNode
=> workerNode
.usage
.waitTime
?.minimum
?? Infinity
373 ...this.workerNodes
.map(
374 workerNode
=> workerNode
.usage
.waitTime
?.maximum
?? -Infinity
379 this.workerNodes
.reduce(
380 (accumulator
, workerNode
) =>
381 accumulator
+ (workerNode
.usage
.waitTime
?.aggregate
?? 0),
384 this.workerNodes
.reduce(
385 (accumulator
, workerNode
) =>
386 accumulator
+ (workerNode
.usage
.tasks
?.executed
?? 0),
390 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
391 .waitTime
.median
&& {
394 this.workerNodes
.map(
395 workerNode
=> workerNode
.usage
.waitTime
?.median
?? 0
405 private get
starting (): boolean {
407 this.workerNodes
.length
< this.minSize
||
408 (this.workerNodes
.length
>= this.minSize
&&
409 this.workerNodes
.some(workerNode
=> !workerNode
.info
.ready
))
413 private get
ready (): boolean {
415 this.workerNodes
.length
>= this.minSize
&&
416 this.workerNodes
.every(workerNode
=> workerNode
.info
.ready
)
421 * Gets the approximate pool utilization.
423 * @returns The pool utilization.
425 private get
utilization (): number {
426 const poolRunTimeCapacity
=
427 (performance
.now() - this.startTimestamp
) * this.maxSize
428 const totalTasksRunTime
= this.workerNodes
.reduce(
429 (accumulator
, workerNode
) =>
430 accumulator
+ (workerNode
.usage
.runTime
?.aggregate
?? 0),
433 const totalTasksWaitTime
= this.workerNodes
.reduce(
434 (accumulator
, workerNode
) =>
435 accumulator
+ (workerNode
.usage
.waitTime
?.aggregate
?? 0),
438 return (totalTasksRunTime
+ totalTasksWaitTime
) / poolRunTimeCapacity
444 * If it is `'dynamic'`, it provides the `max` property.
446 protected abstract get
type (): PoolType
449 * Gets the worker type.
451 protected abstract get
worker (): WorkerType
456 protected abstract get
minSize (): number
461 protected abstract get
maxSize (): number
464 * Get the worker given its id.
466 * @param workerId - The worker id.
467 * @returns The worker if found in the pool worker nodes, `undefined` otherwise.
469 private getWorkerById (workerId
: number): Worker
| undefined {
470 return this.workerNodes
.find(workerNode
=> workerNode
.info
.id
=== workerId
)
474 private checkMessageWorkerId (message
: MessageValue
<Response
>): void {
476 message
.workerId
!= null &&
477 this.getWorkerById(message
.workerId
) == null
480 `Worker message received from unknown worker '${message.workerId}'`
486 * Gets the given worker its worker node key.
488 * @param worker - The worker.
489 * @returns The worker node key if found in the pool worker nodes, `-1` otherwise.
491 private getWorkerNodeKey (worker
: Worker
): number {
492 return this.workerNodes
.findIndex(
493 workerNode
=> workerNode
.worker
=== worker
498 public setWorkerChoiceStrategy (
499 workerChoiceStrategy
: WorkerChoiceStrategy
,
500 workerChoiceStrategyOptions
?: WorkerChoiceStrategyOptions
502 this.checkValidWorkerChoiceStrategy(workerChoiceStrategy
)
503 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
504 this.workerChoiceStrategyContext
.setWorkerChoiceStrategy(
505 this.opts
.workerChoiceStrategy
507 if (workerChoiceStrategyOptions
!= null) {
508 this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
510 for (const workerNode
of this.workerNodes
) {
511 workerNode
.resetUsage()
512 this.setWorkerStatistics(workerNode
.worker
)
517 public setWorkerChoiceStrategyOptions (
518 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
520 this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
521 this.opts
.workerChoiceStrategyOptions
= workerChoiceStrategyOptions
522 this.workerChoiceStrategyContext
.setOptions(
523 this.opts
.workerChoiceStrategyOptions
528 public enableTasksQueue (
530 tasksQueueOptions
?: TasksQueueOptions
532 if (this.opts
.enableTasksQueue
=== true && !enable
) {
533 this.flushTasksQueues()
535 this.opts
.enableTasksQueue
= enable
536 this.setTasksQueueOptions(tasksQueueOptions
as TasksQueueOptions
)
540 public setTasksQueueOptions (tasksQueueOptions
: TasksQueueOptions
): void {
541 if (this.opts
.enableTasksQueue
=== true) {
542 this.checkValidTasksQueueOptions(tasksQueueOptions
)
543 this.opts
.tasksQueueOptions
=
544 this.buildTasksQueueOptions(tasksQueueOptions
)
545 } else if (this.opts
.tasksQueueOptions
!= null) {
546 delete this.opts
.tasksQueueOptions
550 private buildTasksQueueOptions (
551 tasksQueueOptions
: TasksQueueOptions
552 ): TasksQueueOptions
{
554 concurrency
: tasksQueueOptions
?.concurrency
?? 1
559 * Whether the pool is full or not.
561 * The pool filling boolean status.
563 protected get
full (): boolean {
564 return this.workerNodes
.length
>= this.maxSize
568 * Whether the pool is busy or not.
570 * The pool busyness boolean status.
572 protected abstract get
busy (): boolean
575 * Whether worker nodes are executing at least one task.
577 * @returns Worker nodes busyness boolean status.
579 protected internalBusy (): boolean {
581 this.workerNodes
.findIndex(workerNode
=> {
582 return workerNode
.usage
.tasks
.executing
=== 0
588 public async execute (data
?: Data
, name
?: string): Promise
<Response
> {
589 const timestamp
= performance
.now()
590 const workerNodeKey
= this.chooseWorkerNode()
591 const submittedTask
: Task
<Data
> = {
592 name
: name
?? DEFAULT_TASK_NAME
,
593 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
594 data
: data
?? ({} as Data
),
596 workerId
: this.getWorkerInfo(workerNodeKey
).id
as number,
599 const res
= new Promise
<Response
>((resolve
, reject
) => {
600 this.promiseResponseMap
.set(submittedTask
.id
as string, {
603 worker
: this.workerNodes
[workerNodeKey
].worker
607 this.opts
.enableTasksQueue
=== true &&
609 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
>=
610 ((this.opts
.tasksQueueOptions
as TasksQueueOptions
)
611 .concurrency
as number))
613 this.enqueueTask(workerNodeKey
, submittedTask
)
615 this.executeTask(workerNodeKey
, submittedTask
)
617 this.checkAndEmitEvents()
618 // eslint-disable-next-line @typescript-eslint/return-await
623 public async destroy (): Promise
<void> {
625 this.workerNodes
.map(async (workerNode
, workerNodeKey
) => {
626 this.flushTasksQueue(workerNodeKey
)
627 // FIXME: wait for tasks to be finished
628 const workerExitPromise
= new Promise
<void>(resolve
=> {
629 workerNode
.worker
.on('exit', () => {
633 await this.destroyWorker(workerNode
.worker
)
634 await workerExitPromise
640 * Terminates the given worker.
642 * @param worker - A worker within `workerNodes`.
644 protected abstract destroyWorker (worker
: Worker
): void | Promise
<void>
647 * Setup hook to execute code before worker nodes are created in the abstract constructor.
652 protected setupHook (): void {
653 // Intentionally empty
657 * Should return whether the worker is the main worker or not.
659 protected abstract isMain (): boolean
662 * Hook executed before the worker task execution.
665 * @param workerNodeKey - The worker node key.
666 * @param task - The task to execute.
668 protected beforeTaskExecutionHook (
669 workerNodeKey
: number,
672 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
673 ++workerUsage
.tasks
.executing
674 this.updateWaitTimeWorkerUsage(workerUsage
, task
)
675 const tasksWorkerUsage
= this.workerNodes
[
677 ].getTasksWorkerUsage(task
.name
as string) as WorkerUsage
678 ++tasksWorkerUsage
.tasks
.executing
679 this.updateWaitTimeWorkerUsage(tasksWorkerUsage
, task
)
683 * Hook executed after the worker task execution.
686 * @param worker - The worker.
687 * @param message - The received message.
689 protected afterTaskExecutionHook (
691 message
: MessageValue
<Response
>
693 const workerNodeKey
= this.getWorkerNodeKey(worker
)
694 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
695 this.updateTaskStatisticsWorkerUsage(workerUsage
, message
)
696 this.updateRunTimeWorkerUsage(workerUsage
, message
)
697 this.updateEluWorkerUsage(workerUsage
, message
)
698 const tasksWorkerUsage
= this.workerNodes
[
700 ].getTasksWorkerUsage(message
.name
as string) as WorkerUsage
701 this.updateTaskStatisticsWorkerUsage(tasksWorkerUsage
, message
)
702 this.updateRunTimeWorkerUsage(tasksWorkerUsage
, message
)
703 this.updateEluWorkerUsage(tasksWorkerUsage
, message
)
706 private updateTaskStatisticsWorkerUsage (
707 workerUsage
: WorkerUsage
,
708 message
: MessageValue
<Response
>
710 const workerTaskStatistics
= workerUsage
.tasks
711 --workerTaskStatistics
.executing
712 if (message
.taskError
== null) {
713 ++workerTaskStatistics
.executed
715 ++workerTaskStatistics
.failed
719 private updateRunTimeWorkerUsage (
720 workerUsage
: WorkerUsage
,
721 message
: MessageValue
<Response
>
724 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
727 const taskRunTime
= message
.taskPerformance
?.runTime
?? 0
728 workerUsage
.runTime
.aggregate
=
729 (workerUsage
.runTime
.aggregate
?? 0) + taskRunTime
730 workerUsage
.runTime
.minimum
= Math.min(
732 workerUsage
.runTime
?.minimum
?? Infinity
734 workerUsage
.runTime
.maximum
= Math.max(
736 workerUsage
.runTime
?.maximum
?? -Infinity
739 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
741 workerUsage
.tasks
.executed
!== 0
743 workerUsage
.runTime
.average
=
744 workerUsage
.runTime
.aggregate
/ workerUsage
.tasks
.executed
747 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
749 message
.taskPerformance
?.runTime
!= null
751 workerUsage
.runTime
.history
.push(message
.taskPerformance
.runTime
)
752 workerUsage
.runTime
.median
= median(workerUsage
.runTime
.history
)
757 private updateWaitTimeWorkerUsage (
758 workerUsage
: WorkerUsage
,
761 const timestamp
= performance
.now()
762 const taskWaitTime
= timestamp
- (task
.timestamp
?? timestamp
)
764 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().waitTime
767 workerUsage
.waitTime
.aggregate
=
768 (workerUsage
.waitTime
?.aggregate
?? 0) + taskWaitTime
769 workerUsage
.waitTime
.minimum
= Math.min(
771 workerUsage
.waitTime
?.minimum
?? Infinity
773 workerUsage
.waitTime
.maximum
= Math.max(
775 workerUsage
.waitTime
?.maximum
?? -Infinity
778 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
780 workerUsage
.tasks
.executed
!== 0
782 workerUsage
.waitTime
.average
=
783 workerUsage
.waitTime
.aggregate
/ workerUsage
.tasks
.executed
786 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
789 workerUsage
.waitTime
.history
.push(taskWaitTime
)
790 workerUsage
.waitTime
.median
= median(workerUsage
.waitTime
.history
)
795 private updateEluWorkerUsage (
796 workerUsage
: WorkerUsage
,
797 message
: MessageValue
<Response
>
800 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
803 if (message
.taskPerformance
?.elu
!= null) {
804 workerUsage
.elu
.idle
.aggregate
=
805 (workerUsage
.elu
.idle
?.aggregate
?? 0) +
806 message
.taskPerformance
.elu
.idle
807 workerUsage
.elu
.active
.aggregate
=
808 (workerUsage
.elu
.active
?.aggregate
?? 0) +
809 message
.taskPerformance
.elu
.active
810 if (workerUsage
.elu
.utilization
!= null) {
811 workerUsage
.elu
.utilization
=
812 (workerUsage
.elu
.utilization
+
813 message
.taskPerformance
.elu
.utilization
) /
816 workerUsage
.elu
.utilization
= message
.taskPerformance
.elu
.utilization
818 workerUsage
.elu
.idle
.minimum
= Math.min(
819 message
.taskPerformance
.elu
.idle
,
820 workerUsage
.elu
.idle
?.minimum
?? Infinity
822 workerUsage
.elu
.idle
.maximum
= Math.max(
823 message
.taskPerformance
.elu
.idle
,
824 workerUsage
.elu
.idle
?.maximum
?? -Infinity
826 workerUsage
.elu
.active
.minimum
= Math.min(
827 message
.taskPerformance
.elu
.active
,
828 workerUsage
.elu
.active
?.minimum
?? Infinity
830 workerUsage
.elu
.active
.maximum
= Math.max(
831 message
.taskPerformance
.elu
.active
,
832 workerUsage
.elu
.active
?.maximum
?? -Infinity
835 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
837 workerUsage
.tasks
.executed
!== 0
839 workerUsage
.elu
.idle
.average
=
840 workerUsage
.elu
.idle
.aggregate
/ workerUsage
.tasks
.executed
841 workerUsage
.elu
.active
.average
=
842 workerUsage
.elu
.active
.aggregate
/ workerUsage
.tasks
.executed
845 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
848 workerUsage
.elu
.idle
.history
.push(message
.taskPerformance
.elu
.idle
)
849 workerUsage
.elu
.active
.history
.push(
850 message
.taskPerformance
.elu
.active
852 workerUsage
.elu
.idle
.median
= median(workerUsage
.elu
.idle
.history
)
853 workerUsage
.elu
.active
.median
= median(workerUsage
.elu
.active
.history
)
860 * Chooses a worker node for the next task.
862 * The default worker choice strategy uses a round robin algorithm to distribute the tasks.
864 * @returns The worker node key
866 private chooseWorkerNode (): number {
867 if (this.shallCreateDynamicWorker()) {
868 const worker
= this.createAndSetupDynamicWorker()
870 this.workerChoiceStrategyContext
.getStrategyPolicy().useDynamicWorker
872 return this.getWorkerNodeKey(worker
)
875 return this.workerChoiceStrategyContext
.execute()
879 * Conditions for dynamic worker creation.
881 * @returns Whether to create a dynamic worker or not.
883 private shallCreateDynamicWorker (): boolean {
884 return this.type === PoolTypes
.dynamic
&& !this.full
&& this.internalBusy()
888 * Sends a message to the given worker.
890 * @param worker - The worker which should receive the message.
891 * @param message - The message.
893 protected abstract sendToWorker (
895 message
: MessageValue
<Data
>
899 * Registers a listener callback on the given worker.
901 * @param worker - The worker which should register a listener.
902 * @param listener - The message listener callback.
904 private registerWorkerMessageListener
<Message
extends Data
| Response
>(
906 listener
: (message
: MessageValue
<Message
>) => void
908 worker
.on('message', listener
as MessageHandler
<Worker
>)
912 * Creates a new worker.
914 * @returns Newly created worker.
916 protected abstract createWorker (): Worker
919 * Function that can be hooked up when a worker has been newly created and moved to the pool worker nodes.
922 * @param worker - The newly created worker.
924 protected afterWorkerSetup (worker
: Worker
): void {
925 // Listen to worker messages.
926 this.registerWorkerMessageListener(worker
, this.workerListener())
927 // Send startup message to worker.
928 this.sendToWorker(worker
, {
930 workerId
: this.getWorkerInfo(this.getWorkerNodeKey(worker
)).id
as number
932 // Setup worker task statistics computation.
933 this.setWorkerStatistics(worker
)
937 * Creates a new worker and sets it up completely in the pool worker nodes.
939 * @returns New, completely set up worker.
941 protected createAndSetupWorker (): Worker
{
942 const worker
= this.createWorker()
944 worker
.on('message', this.opts
.messageHandler
?? EMPTY_FUNCTION
)
945 worker
.on('error', this.opts
.errorHandler
?? EMPTY_FUNCTION
)
946 worker
.on('error', error
=> {
947 if (this.emitter
!= null) {
948 this.emitter
.emit(PoolEvents
.error
, error
)
950 if (this.opts
.enableTasksQueue
=== true) {
951 this.redistributeQueuedTasks(worker
)
953 if (this.opts
.restartWorkerOnError
=== true && !this.starting
) {
954 if (this.getWorkerInfo(this.getWorkerNodeKey(worker
)).dynamic
) {
955 this.createAndSetupDynamicWorker()
957 this.createAndSetupWorker()
961 worker
.on('online', this.opts
.onlineHandler
?? EMPTY_FUNCTION
)
962 worker
.on('exit', this.opts
.exitHandler
?? EMPTY_FUNCTION
)
963 worker
.once('exit', () => {
964 this.removeWorkerNode(worker
)
967 this.pushWorkerNode(worker
)
969 this.afterWorkerSetup(worker
)
974 private redistributeQueuedTasks (worker
: Worker
): void {
975 const workerNodeKey
= this.getWorkerNodeKey(worker
)
976 while (this.tasksQueueSize(workerNodeKey
) > 0) {
977 let targetWorkerNodeKey
: number = workerNodeKey
978 let minQueuedTasks
= Infinity
979 for (const [workerNodeId
, workerNode
] of this.workerNodes
.entries()) {
981 workerNodeId
!== workerNodeKey
&&
982 workerNode
.usage
.tasks
.queued
=== 0
984 targetWorkerNodeKey
= workerNodeId
988 workerNodeId
!== workerNodeKey
&&
989 workerNode
.usage
.tasks
.queued
< minQueuedTasks
991 minQueuedTasks
= workerNode
.usage
.tasks
.queued
992 targetWorkerNodeKey
= workerNodeId
997 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1003 * Creates a new dynamic worker and sets it up completely in the pool worker nodes.
1005 * @returns New, completely set up dynamic worker.
1007 protected createAndSetupDynamicWorker (): Worker
{
1008 const worker
= this.createAndSetupWorker()
1009 this.registerWorkerMessageListener(worker
, message
=> {
1010 const workerNodeKey
= this.getWorkerNodeKey(worker
)
1012 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
1013 (message
.kill
!= null &&
1014 ((this.opts
.enableTasksQueue
=== false &&
1015 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
=== 0) ||
1016 (this.opts
.enableTasksQueue
=== true &&
1017 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
=== 0 &&
1018 this.tasksQueueSize(workerNodeKey
) === 0)))
1020 // Kill message received from the worker: no new tasks are submitted to that worker for a while ( > maxInactiveTime)
1021 void (this.destroyWorker(worker
) as Promise
<void>)
1024 const workerInfo
= this.getWorkerInfo(this.getWorkerNodeKey(worker
))
1025 workerInfo
.dynamic
= true
1026 this.sendToWorker(worker
, {
1028 workerId
: workerInfo
.id
as number
1034 * This function is the listener registered for each worker message.
1036 * @returns The listener function to execute when a message is received from a worker.
1038 protected workerListener (): (message
: MessageValue
<Response
>) => void {
1040 this.checkMessageWorkerId(message
)
1041 if (message
.ready
!= null && message
.workerId
!= null) {
1042 // Worker ready message received
1043 this.handleWorkerReadyMessage(message
)
1044 } else if (message
.id
!= null) {
1045 // Task execution response received
1046 this.handleTaskExecutionResponse(message
)
1051 private handleWorkerReadyMessage (message
: MessageValue
<Response
>): void {
1052 const worker
= this.getWorkerById(message
.workerId
)
1053 this.getWorkerInfo(this.getWorkerNodeKey(worker
as Worker
)).ready
=
1054 message
.ready
as boolean
1055 if (this.emitter
!= null && this.ready
) {
1056 this.emitter
.emit(PoolEvents
.ready
, this.info
)
1060 private handleTaskExecutionResponse (message
: MessageValue
<Response
>): void {
1061 const promiseResponse
= this.promiseResponseMap
.get(message
.id
as string)
1062 if (promiseResponse
!= null) {
1063 if (message
.taskError
!= null) {
1064 if (this.emitter
!= null) {
1065 this.emitter
.emit(PoolEvents
.taskError
, message
.taskError
)
1067 promiseResponse
.reject(message
.taskError
.message
)
1069 promiseResponse
.resolve(message
.data
as Response
)
1071 this.afterTaskExecutionHook(promiseResponse
.worker
, message
)
1072 this.promiseResponseMap
.delete(message
.id
as string)
1073 const workerNodeKey
= this.getWorkerNodeKey(promiseResponse
.worker
)
1075 this.opts
.enableTasksQueue
=== true &&
1076 this.tasksQueueSize(workerNodeKey
) > 0
1080 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1083 this.workerChoiceStrategyContext
.update(workerNodeKey
)
1087 private checkAndEmitEvents (): void {
1088 if (this.emitter
!= null) {
1090 this.emitter
.emit(PoolEvents
.busy
, this.info
)
1092 if (this.type === PoolTypes
.dynamic
&& this.full
) {
1093 this.emitter
.emit(PoolEvents
.full
, this.info
)
1099 * Gets the worker information.
1101 * @param workerNodeKey - The worker node key.
1103 private getWorkerInfo (workerNodeKey
: number): WorkerInfo
{
1104 return this.workerNodes
[workerNodeKey
].info
1108 * Pushes the given worker in the pool worker nodes.
1110 * @param worker - The worker.
1111 * @returns The worker nodes length.
1113 private pushWorkerNode (worker
: Worker
): number {
1114 return this.workerNodes
.push(new WorkerNode(worker
, this.worker
))
1118 * Removes the given worker from the pool worker nodes.
1120 * @param worker - The worker.
1122 private removeWorkerNode (worker
: Worker
): void {
1123 const workerNodeKey
= this.getWorkerNodeKey(worker
)
1124 if (workerNodeKey
!== -1) {
1125 this.workerNodes
.splice(workerNodeKey
, 1)
1126 this.workerChoiceStrategyContext
.remove(workerNodeKey
)
1130 private executeTask (workerNodeKey
: number, task
: Task
<Data
>): void {
1131 this.beforeTaskExecutionHook(workerNodeKey
, task
)
1132 this.sendToWorker(this.workerNodes
[workerNodeKey
].worker
, task
)
1135 private enqueueTask (workerNodeKey
: number, task
: Task
<Data
>): number {
1136 return this.workerNodes
[workerNodeKey
].enqueueTask(task
)
1139 private dequeueTask (workerNodeKey
: number): Task
<Data
> | undefined {
1140 return this.workerNodes
[workerNodeKey
].dequeueTask()
1143 private tasksQueueSize (workerNodeKey
: number): number {
1144 return this.workerNodes
[workerNodeKey
].tasksQueueSize()
1147 private flushTasksQueue (workerNodeKey
: number): void {
1148 while (this.tasksQueueSize(workerNodeKey
) > 0) {
1151 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1154 this.workerNodes
[workerNodeKey
].clearTasksQueue()
1157 private flushTasksQueues (): void {
1158 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
1159 this.flushTasksQueue(workerNodeKey
)
1163 private setWorkerStatistics (worker
: Worker
): void {
1164 this.sendToWorker(worker
, {
1167 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1169 elu
: this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1172 workerId
: this.getWorkerInfo(this.getWorkerNodeKey(worker
)).id
as number