1 import { randomUUID
} from
'node:crypto'
2 import { performance
} from
'node:perf_hooks'
3 import type { MessageValue
, PromiseResponseWrapper
} from
'../utility-types'
5 DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
,
12 import { KillBehaviors
} from
'../worker/worker-options'
21 type TasksQueueOptions
34 WorkerChoiceStrategies
,
35 type WorkerChoiceStrategy
,
36 type WorkerChoiceStrategyOptions
37 } from
'./selection-strategies/selection-strategies-types'
38 import { WorkerChoiceStrategyContext
} from
'./selection-strategies/worker-choice-strategy-context'
39 import { version
} from
'./version'
40 import { WorkerNode
} from
'./worker-node'
43 * Base class that implements some shared logic for all poolifier pools.
45 * @typeParam Worker - Type of worker which manages this pool.
46 * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
47 * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
49 export abstract class AbstractPool
<
50 Worker
extends IWorker
,
53 > implements IPool
<Worker
, Data
, Response
> {
55 public readonly workerNodes
: Array<IWorkerNode
<Worker
, Data
>> = []
58 public readonly emitter
?: PoolEmitter
61 * The execution response promise map.
63 * - `key`: The message id of each submitted task.
64 * - `value`: An object that contains the worker, the execution response promise resolve and reject callbacks.
66 * When we receive a message from the worker, we get a map entry with the promise resolve/reject bound to the message id.
68 protected promiseResponseMap
: Map
<
70 PromiseResponseWrapper
<Worker
, Response
>
71 > = new Map
<string, PromiseResponseWrapper
<Worker
, Response
>>()
74 * Worker choice strategy context referencing a worker choice algorithm implementation.
76 protected workerChoiceStrategyContext
: WorkerChoiceStrategyContext
<
83 * The start timestamp of the pool.
85 private readonly startTimestamp
88 * Constructs a new poolifier pool.
90 * @param numberOfWorkers - Number of workers that this pool should manage.
91 * @param filePath - Path to the worker file.
92 * @param opts - Options for the pool.
95 protected readonly numberOfWorkers
: number,
96 protected readonly filePath
: string,
97 protected readonly opts
: PoolOptions
<Worker
>
100 throw new Error('Cannot start a pool from a worker!')
102 this.checkNumberOfWorkers(this.numberOfWorkers
)
103 this.checkFilePath(this.filePath
)
104 this.checkPoolOptions(this.opts
)
106 this.chooseWorkerNode
= this.chooseWorkerNode
.bind(this)
107 this.executeTask
= this.executeTask
.bind(this)
108 this.enqueueTask
= this.enqueueTask
.bind(this)
109 this.checkAndEmitEvents
= this.checkAndEmitEvents
.bind(this)
111 if (this.opts
.enableEvents
=== true) {
112 this.emitter
= new PoolEmitter()
114 this.workerChoiceStrategyContext
= new WorkerChoiceStrategyContext
<
120 this.opts
.workerChoiceStrategy
,
121 this.opts
.workerChoiceStrategyOptions
126 while (this.workerNodes
.length
< this.numberOfWorkers
) {
127 this.createAndSetupWorker()
130 this.startTimestamp
= performance
.now()
133 private checkFilePath (filePath
: string): void {
136 (typeof filePath
=== 'string' && filePath
.trim().length
=== 0)
138 throw new Error('Please specify a file with a worker implementation')
142 private checkNumberOfWorkers (numberOfWorkers
: number): void {
143 if (numberOfWorkers
== null) {
145 'Cannot instantiate a pool without specifying the number of workers'
147 } else if (!Number.isSafeInteger(numberOfWorkers
)) {
149 'Cannot instantiate a pool with a non safe integer number of workers'
151 } else if (numberOfWorkers
< 0) {
152 throw new RangeError(
153 'Cannot instantiate a pool with a negative number of workers'
155 } else if (this.type === PoolTypes
.fixed
&& numberOfWorkers
=== 0) {
156 throw new RangeError('Cannot instantiate a fixed pool with zero worker')
160 protected checkDynamicPoolSize (min
: number, max
: number): void {
161 if (this.type === PoolTypes
.dynamic
) {
163 throw new RangeError(
164 'Cannot instantiate a dynamic pool with a maximum pool size inferior to the minimum pool size'
166 } else if (min
=== 0 && max
=== 0) {
167 throw new RangeError(
168 'Cannot instantiate a dynamic pool with a minimum pool size and a maximum pool size equal to zero'
170 } else if (min
=== max
) {
171 throw new RangeError(
172 'Cannot instantiate a dynamic pool with a minimum pool size equal to the maximum pool size. Use a fixed pool instead'
178 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
179 if (isPlainObject(opts
)) {
180 this.opts
.workerChoiceStrategy
=
181 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
182 this.checkValidWorkerChoiceStrategy(this.opts
.workerChoiceStrategy
)
183 this.opts
.workerChoiceStrategyOptions
=
184 opts
.workerChoiceStrategyOptions
??
185 DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
186 this.checkValidWorkerChoiceStrategyOptions(
187 this.opts
.workerChoiceStrategyOptions
189 this.opts
.restartWorkerOnError
= opts
.restartWorkerOnError
?? true
190 this.opts
.enableEvents
= opts
.enableEvents
?? true
191 this.opts
.enableTasksQueue
= opts
.enableTasksQueue
?? false
192 if (this.opts
.enableTasksQueue
) {
193 this.checkValidTasksQueueOptions(
194 opts
.tasksQueueOptions
as TasksQueueOptions
196 this.opts
.tasksQueueOptions
= this.buildTasksQueueOptions(
197 opts
.tasksQueueOptions
as TasksQueueOptions
201 throw new TypeError('Invalid pool options: must be a plain object')
205 private checkValidWorkerChoiceStrategy (
206 workerChoiceStrategy
: WorkerChoiceStrategy
208 if (!Object.values(WorkerChoiceStrategies
).includes(workerChoiceStrategy
)) {
210 `Invalid worker choice strategy '${workerChoiceStrategy}'`
215 private checkValidWorkerChoiceStrategyOptions (
216 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
218 if (!isPlainObject(workerChoiceStrategyOptions
)) {
220 'Invalid worker choice strategy options: must be a plain object'
224 workerChoiceStrategyOptions
.weights
!= null &&
225 Object.keys(workerChoiceStrategyOptions
.weights
).length
!== this.maxSize
228 'Invalid worker choice strategy options: must have a weight for each worker node'
232 workerChoiceStrategyOptions
.measurement
!= null &&
233 !Object.values(Measurements
).includes(
234 workerChoiceStrategyOptions
.measurement
238 `Invalid worker choice strategy options: invalid measurement '${workerChoiceStrategyOptions.measurement}'`
243 private checkValidTasksQueueOptions (
244 tasksQueueOptions
: TasksQueueOptions
246 if (tasksQueueOptions
!= null && !isPlainObject(tasksQueueOptions
)) {
247 throw new TypeError('Invalid tasks queue options: must be a plain object')
250 tasksQueueOptions
?.concurrency
!= null &&
251 !Number.isSafeInteger(tasksQueueOptions
.concurrency
)
254 'Invalid worker tasks concurrency: must be an integer'
258 tasksQueueOptions
?.concurrency
!= null &&
259 tasksQueueOptions
.concurrency
<= 0
262 `Invalid worker tasks concurrency '${tasksQueueOptions.concurrency}'`
268 public get
info (): PoolInfo
{
274 strategy
: this.opts
.workerChoiceStrategy
as WorkerChoiceStrategy
,
275 minSize
: this.minSize
,
276 maxSize
: this.maxSize
,
277 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
278 .runTime
.aggregate
&&
279 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
280 .waitTime
.aggregate
&& { utilization
: round(this.utilization
) }),
281 workerNodes
: this.workerNodes
.length
,
282 idleWorkerNodes
: this.workerNodes
.reduce(
283 (accumulator
, workerNode
) =>
284 workerNode
.usage
.tasks
.executing
=== 0
289 busyWorkerNodes
: this.workerNodes
.reduce(
290 (accumulator
, workerNode
) =>
291 workerNode
.usage
.tasks
.executing
> 0 ? accumulator
+ 1 : accumulator
,
294 executedTasks
: this.workerNodes
.reduce(
295 (accumulator
, workerNode
) =>
296 accumulator
+ workerNode
.usage
.tasks
.executed
,
299 executingTasks
: this.workerNodes
.reduce(
300 (accumulator
, workerNode
) =>
301 accumulator
+ workerNode
.usage
.tasks
.executing
,
304 queuedTasks
: this.workerNodes
.reduce(
305 (accumulator
, workerNode
) =>
306 accumulator
+ workerNode
.usage
.tasks
.queued
,
309 maxQueuedTasks
: this.workerNodes
.reduce(
310 (accumulator
, workerNode
) =>
311 accumulator
+ workerNode
.usage
.tasks
.maxQueued
,
314 failedTasks
: this.workerNodes
.reduce(
315 (accumulator
, workerNode
) =>
316 accumulator
+ workerNode
.usage
.tasks
.failed
,
319 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
320 .runTime
.aggregate
&& {
324 ...this.workerNodes
.map(
325 workerNode
=> workerNode
.usage
.runTime
?.minimum
?? Infinity
331 ...this.workerNodes
.map(
332 workerNode
=> workerNode
.usage
.runTime
?.maximum
?? -Infinity
337 this.workerNodes
.reduce(
338 (accumulator
, workerNode
) =>
339 accumulator
+ (workerNode
.usage
.runTime
?.aggregate
?? 0),
342 this.workerNodes
.reduce(
343 (accumulator
, workerNode
) =>
344 accumulator
+ (workerNode
.usage
.tasks
?.executed
?? 0),
348 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
352 this.workerNodes
.map(
353 workerNode
=> workerNode
.usage
.runTime
?.median
?? 0
360 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
361 .waitTime
.aggregate
&& {
365 ...this.workerNodes
.map(
366 workerNode
=> workerNode
.usage
.waitTime
?.minimum
?? Infinity
372 ...this.workerNodes
.map(
373 workerNode
=> workerNode
.usage
.waitTime
?.maximum
?? -Infinity
378 this.workerNodes
.reduce(
379 (accumulator
, workerNode
) =>
380 accumulator
+ (workerNode
.usage
.waitTime
?.aggregate
?? 0),
383 this.workerNodes
.reduce(
384 (accumulator
, workerNode
) =>
385 accumulator
+ (workerNode
.usage
.tasks
?.executed
?? 0),
389 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
390 .waitTime
.median
&& {
393 this.workerNodes
.map(
394 workerNode
=> workerNode
.usage
.waitTime
?.median
?? 0
404 private get
starting (): boolean {
406 this.workerNodes
.length
< this.minSize
||
407 (this.workerNodes
.length
>= this.minSize
&&
408 this.workerNodes
.some(workerNode
=> !workerNode
.info
.ready
))
412 private get
ready (): boolean {
414 this.workerNodes
.length
>= this.minSize
&&
415 this.workerNodes
.every(workerNode
=> workerNode
.info
.ready
)
420 * Gets the approximate pool utilization.
422 * @returns The pool utilization.
424 private get
utilization (): number {
425 const poolRunTimeCapacity
=
426 (performance
.now() - this.startTimestamp
) * this.maxSize
427 const totalTasksRunTime
= this.workerNodes
.reduce(
428 (accumulator
, workerNode
) =>
429 accumulator
+ (workerNode
.usage
.runTime
?.aggregate
?? 0),
432 const totalTasksWaitTime
= this.workerNodes
.reduce(
433 (accumulator
, workerNode
) =>
434 accumulator
+ (workerNode
.usage
.waitTime
?.aggregate
?? 0),
437 return (totalTasksRunTime
+ totalTasksWaitTime
) / poolRunTimeCapacity
443 * If it is `'dynamic'`, it provides the `max` property.
445 protected abstract get
type (): PoolType
448 * Gets the worker type.
450 protected abstract get
worker (): WorkerType
455 protected abstract get
minSize (): number
460 protected abstract get
maxSize (): number
463 * Get the worker given its id.
465 * @param workerId - The worker id.
466 * @returns The worker if found in the pool worker nodes, `undefined` otherwise.
468 private getWorkerById (workerId
: number): Worker
| undefined {
469 return this.workerNodes
.find(workerNode
=> workerNode
.info
.id
=== workerId
)
473 private checkMessageWorkerId (message
: MessageValue
<Response
>): void {
475 message
.workerId
!= null &&
476 this.getWorkerById(message
.workerId
) == null
479 `Worker message received from unknown worker '${message.workerId}'`
485 * Gets the given worker its worker node key.
487 * @param worker - The worker.
488 * @returns The worker node key if found in the pool worker nodes, `-1` otherwise.
490 private getWorkerNodeKey (worker
: Worker
): number {
491 return this.workerNodes
.findIndex(
492 workerNode
=> workerNode
.worker
=== worker
497 public setWorkerChoiceStrategy (
498 workerChoiceStrategy
: WorkerChoiceStrategy
,
499 workerChoiceStrategyOptions
?: WorkerChoiceStrategyOptions
501 this.checkValidWorkerChoiceStrategy(workerChoiceStrategy
)
502 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
503 this.workerChoiceStrategyContext
.setWorkerChoiceStrategy(
504 this.opts
.workerChoiceStrategy
506 if (workerChoiceStrategyOptions
!= null) {
507 this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
509 for (const workerNode
of this.workerNodes
) {
510 workerNode
.resetUsage()
511 this.setWorkerStatistics(workerNode
.worker
)
516 public setWorkerChoiceStrategyOptions (
517 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
519 this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
520 this.opts
.workerChoiceStrategyOptions
= workerChoiceStrategyOptions
521 this.workerChoiceStrategyContext
.setOptions(
522 this.opts
.workerChoiceStrategyOptions
527 public enableTasksQueue (
529 tasksQueueOptions
?: TasksQueueOptions
531 if (this.opts
.enableTasksQueue
=== true && !enable
) {
532 this.flushTasksQueues()
534 this.opts
.enableTasksQueue
= enable
535 this.setTasksQueueOptions(tasksQueueOptions
as TasksQueueOptions
)
539 public setTasksQueueOptions (tasksQueueOptions
: TasksQueueOptions
): void {
540 if (this.opts
.enableTasksQueue
=== true) {
541 this.checkValidTasksQueueOptions(tasksQueueOptions
)
542 this.opts
.tasksQueueOptions
=
543 this.buildTasksQueueOptions(tasksQueueOptions
)
544 } else if (this.opts
.tasksQueueOptions
!= null) {
545 delete this.opts
.tasksQueueOptions
549 private buildTasksQueueOptions (
550 tasksQueueOptions
: TasksQueueOptions
551 ): TasksQueueOptions
{
553 concurrency
: tasksQueueOptions
?.concurrency
?? 1
558 * Whether the pool is full or not.
560 * The pool filling boolean status.
562 protected get
full (): boolean {
563 return this.workerNodes
.length
>= this.maxSize
567 * Whether the pool is busy or not.
569 * The pool busyness boolean status.
571 protected abstract get
busy (): boolean
574 * Whether worker nodes are executing at least one task.
576 * @returns Worker nodes busyness boolean status.
578 protected internalBusy (): boolean {
580 this.workerNodes
.findIndex(workerNode
=> {
581 return workerNode
.usage
.tasks
.executing
=== 0
587 public async execute (data
?: Data
, name
?: string): Promise
<Response
> {
588 const timestamp
= performance
.now()
589 const workerNodeKey
= this.chooseWorkerNode()
590 const submittedTask
: Task
<Data
> = {
592 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
593 data
: data
?? ({} as Data
),
595 workerId
: this.getWorkerInfo(workerNodeKey
).id
as number,
598 const res
= new Promise
<Response
>((resolve
, reject
) => {
599 this.promiseResponseMap
.set(submittedTask
.id
as string, {
602 worker
: this.workerNodes
[workerNodeKey
].worker
606 this.opts
.enableTasksQueue
=== true &&
608 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
>=
609 ((this.opts
.tasksQueueOptions
as TasksQueueOptions
)
610 .concurrency
as number))
612 this.enqueueTask(workerNodeKey
, submittedTask
)
614 this.executeTask(workerNodeKey
, submittedTask
)
616 this.checkAndEmitEvents()
617 // eslint-disable-next-line @typescript-eslint/return-await
622 public async destroy (): Promise
<void> {
624 this.workerNodes
.map(async (workerNode
, workerNodeKey
) => {
625 this.flushTasksQueue(workerNodeKey
)
626 // FIXME: wait for tasks to be finished
627 const workerExitPromise
= new Promise
<void>(resolve
=> {
628 workerNode
.worker
.on('exit', () => {
632 await this.destroyWorker(workerNode
.worker
)
633 await workerExitPromise
639 * Terminates the given worker.
641 * @param worker - A worker within `workerNodes`.
643 protected abstract destroyWorker (worker
: Worker
): void | Promise
<void>
646 * Setup hook to execute code before worker nodes are created in the abstract constructor.
651 protected setupHook (): void {
652 // Intentionally empty
656 * Should return whether the worker is the main worker or not.
658 protected abstract isMain (): boolean
661 * Hook executed before the worker task execution.
664 * @param workerNodeKey - The worker node key.
665 * @param task - The task to execute.
667 protected beforeTaskExecutionHook (
668 workerNodeKey
: number,
671 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
672 ++workerUsage
.tasks
.executing
673 this.updateWaitTimeWorkerUsage(workerUsage
, task
)
677 * Hook executed after the worker task execution.
680 * @param worker - The worker.
681 * @param message - The received message.
683 protected afterTaskExecutionHook (
685 message
: MessageValue
<Response
>
687 const workerUsage
= this.workerNodes
[this.getWorkerNodeKey(worker
)].usage
688 this.updateTaskStatisticsWorkerUsage(workerUsage
, message
)
689 this.updateRunTimeWorkerUsage(workerUsage
, message
)
690 this.updateEluWorkerUsage(workerUsage
, message
)
693 private updateTaskStatisticsWorkerUsage (
694 workerUsage
: WorkerUsage
,
695 message
: MessageValue
<Response
>
697 const workerTaskStatistics
= workerUsage
.tasks
698 --workerTaskStatistics
.executing
699 if (message
.taskError
== null) {
700 ++workerTaskStatistics
.executed
702 ++workerTaskStatistics
.failed
706 private updateRunTimeWorkerUsage (
707 workerUsage
: WorkerUsage
,
708 message
: MessageValue
<Response
>
711 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
714 const taskRunTime
= message
.taskPerformance
?.runTime
?? 0
715 workerUsage
.runTime
.aggregate
=
716 (workerUsage
.runTime
.aggregate
?? 0) + taskRunTime
717 workerUsage
.runTime
.minimum
= Math.min(
719 workerUsage
.runTime
?.minimum
?? Infinity
721 workerUsage
.runTime
.maximum
= Math.max(
723 workerUsage
.runTime
?.maximum
?? -Infinity
726 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
728 workerUsage
.tasks
.executed
!== 0
730 workerUsage
.runTime
.average
=
731 workerUsage
.runTime
.aggregate
/ workerUsage
.tasks
.executed
734 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
736 message
.taskPerformance
?.runTime
!= null
738 workerUsage
.runTime
.history
.push(message
.taskPerformance
.runTime
)
739 workerUsage
.runTime
.median
= median(workerUsage
.runTime
.history
)
744 private updateWaitTimeWorkerUsage (
745 workerUsage
: WorkerUsage
,
748 const timestamp
= performance
.now()
749 const taskWaitTime
= timestamp
- (task
.timestamp
?? timestamp
)
751 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().waitTime
754 workerUsage
.waitTime
.aggregate
=
755 (workerUsage
.waitTime
?.aggregate
?? 0) + taskWaitTime
756 workerUsage
.waitTime
.minimum
= Math.min(
758 workerUsage
.waitTime
?.minimum
?? Infinity
760 workerUsage
.waitTime
.maximum
= Math.max(
762 workerUsage
.waitTime
?.maximum
?? -Infinity
765 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
767 workerUsage
.tasks
.executed
!== 0
769 workerUsage
.waitTime
.average
=
770 workerUsage
.waitTime
.aggregate
/ workerUsage
.tasks
.executed
773 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
776 workerUsage
.waitTime
.history
.push(taskWaitTime
)
777 workerUsage
.waitTime
.median
= median(workerUsage
.waitTime
.history
)
782 private updateEluWorkerUsage (
783 workerUsage
: WorkerUsage
,
784 message
: MessageValue
<Response
>
787 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
790 if (message
.taskPerformance
?.elu
!= null) {
791 workerUsage
.elu
.idle
.aggregate
=
792 (workerUsage
.elu
.idle
?.aggregate
?? 0) +
793 message
.taskPerformance
.elu
.idle
794 workerUsage
.elu
.active
.aggregate
=
795 (workerUsage
.elu
.active
?.aggregate
?? 0) +
796 message
.taskPerformance
.elu
.active
797 if (workerUsage
.elu
.utilization
!= null) {
798 workerUsage
.elu
.utilization
=
799 (workerUsage
.elu
.utilization
+
800 message
.taskPerformance
.elu
.utilization
) /
803 workerUsage
.elu
.utilization
= message
.taskPerformance
.elu
.utilization
805 workerUsage
.elu
.idle
.minimum
= Math.min(
806 message
.taskPerformance
.elu
.idle
,
807 workerUsage
.elu
.idle
?.minimum
?? Infinity
809 workerUsage
.elu
.idle
.maximum
= Math.max(
810 message
.taskPerformance
.elu
.idle
,
811 workerUsage
.elu
.idle
?.maximum
?? -Infinity
813 workerUsage
.elu
.active
.minimum
= Math.min(
814 message
.taskPerformance
.elu
.active
,
815 workerUsage
.elu
.active
?.minimum
?? Infinity
817 workerUsage
.elu
.active
.maximum
= Math.max(
818 message
.taskPerformance
.elu
.active
,
819 workerUsage
.elu
.active
?.maximum
?? -Infinity
822 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
824 workerUsage
.tasks
.executed
!== 0
826 workerUsage
.elu
.idle
.average
=
827 workerUsage
.elu
.idle
.aggregate
/ workerUsage
.tasks
.executed
828 workerUsage
.elu
.active
.average
=
829 workerUsage
.elu
.active
.aggregate
/ workerUsage
.tasks
.executed
832 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
835 workerUsage
.elu
.idle
.history
.push(message
.taskPerformance
.elu
.idle
)
836 workerUsage
.elu
.active
.history
.push(
837 message
.taskPerformance
.elu
.active
839 workerUsage
.elu
.idle
.median
= median(workerUsage
.elu
.idle
.history
)
840 workerUsage
.elu
.active
.median
= median(workerUsage
.elu
.active
.history
)
847 * Chooses a worker node for the next task.
849 * The default worker choice strategy uses a round robin algorithm to distribute the tasks.
851 * @returns The worker node key
853 private chooseWorkerNode (): number {
854 if (this.shallCreateDynamicWorker()) {
855 const worker
= this.createAndSetupDynamicWorker()
857 this.workerChoiceStrategyContext
.getStrategyPolicy().useDynamicWorker
859 return this.getWorkerNodeKey(worker
)
862 return this.workerChoiceStrategyContext
.execute()
866 * Conditions for dynamic worker creation.
868 * @returns Whether to create a dynamic worker or not.
870 private shallCreateDynamicWorker (): boolean {
871 return this.type === PoolTypes
.dynamic
&& !this.full
&& this.internalBusy()
875 * Sends a message to the given worker.
877 * @param worker - The worker which should receive the message.
878 * @param message - The message.
880 protected abstract sendToWorker (
882 message
: MessageValue
<Data
>
886 * Registers a listener callback on the given worker.
888 * @param worker - The worker which should register a listener.
889 * @param listener - The message listener callback.
891 private registerWorkerMessageListener
<Message
extends Data
| Response
>(
893 listener
: (message
: MessageValue
<Message
>) => void
895 worker
.on('message', listener
as MessageHandler
<Worker
>)
899 * Creates a new worker.
901 * @returns Newly created worker.
903 protected abstract createWorker (): Worker
906 * Function that can be hooked up when a worker has been newly created and moved to the pool worker nodes.
909 * @param worker - The newly created worker.
911 protected afterWorkerSetup (worker
: Worker
): void {
912 // Listen to worker messages.
913 this.registerWorkerMessageListener(worker
, this.workerListener())
914 // Send startup message to worker.
915 this.sendToWorker(worker
, {
917 workerId
: this.getWorkerInfo(this.getWorkerNodeKey(worker
)).id
as number
919 // Setup worker task statistics computation.
920 this.setWorkerStatistics(worker
)
924 * Creates a new worker and sets it up completely in the pool worker nodes.
926 * @returns New, completely set up worker.
928 protected createAndSetupWorker (): Worker
{
929 const worker
= this.createWorker()
931 worker
.on('message', this.opts
.messageHandler
?? EMPTY_FUNCTION
)
932 worker
.on('error', this.opts
.errorHandler
?? EMPTY_FUNCTION
)
933 worker
.on('error', error
=> {
934 if (this.emitter
!= null) {
935 this.emitter
.emit(PoolEvents
.error
, error
)
937 if (this.opts
.enableTasksQueue
=== true) {
938 this.redistributeQueuedTasks(worker
)
940 if (this.opts
.restartWorkerOnError
=== true && !this.starting
) {
941 if (this.getWorkerInfo(this.getWorkerNodeKey(worker
)).dynamic
) {
942 this.createAndSetupDynamicWorker()
944 this.createAndSetupWorker()
948 worker
.on('online', this.opts
.onlineHandler
?? EMPTY_FUNCTION
)
949 worker
.on('exit', this.opts
.exitHandler
?? EMPTY_FUNCTION
)
950 worker
.once('exit', () => {
951 this.removeWorkerNode(worker
)
954 this.pushWorkerNode(worker
)
956 this.afterWorkerSetup(worker
)
961 private redistributeQueuedTasks (worker
: Worker
): void {
962 const workerNodeKey
= this.getWorkerNodeKey(worker
)
963 while (this.tasksQueueSize(workerNodeKey
) > 0) {
964 let targetWorkerNodeKey
: number = workerNodeKey
965 let minQueuedTasks
= Infinity
966 for (const [workerNodeId
, workerNode
] of this.workerNodes
.entries()) {
968 workerNodeId
!== workerNodeKey
&&
969 workerNode
.usage
.tasks
.queued
=== 0
971 targetWorkerNodeKey
= workerNodeId
975 workerNodeId
!== workerNodeKey
&&
976 workerNode
.usage
.tasks
.queued
< minQueuedTasks
978 minQueuedTasks
= workerNode
.usage
.tasks
.queued
979 targetWorkerNodeKey
= workerNodeId
984 this.dequeueTask(workerNodeKey
) as Task
<Data
>
990 * Creates a new dynamic worker and sets it up completely in the pool worker nodes.
992 * @returns New, completely set up dynamic worker.
994 protected createAndSetupDynamicWorker (): Worker
{
995 const worker
= this.createAndSetupWorker()
996 this.registerWorkerMessageListener(worker
, message
=> {
997 const workerNodeKey
= this.getWorkerNodeKey(worker
)
999 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
1000 (message
.kill
!= null &&
1001 ((this.opts
.enableTasksQueue
=== false &&
1002 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
=== 0) ||
1003 (this.opts
.enableTasksQueue
=== true &&
1004 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
=== 0 &&
1005 this.tasksQueueSize(workerNodeKey
) === 0)))
1007 // Kill message received from the worker: no new tasks are submitted to that worker for a while ( > maxInactiveTime)
1008 void (this.destroyWorker(worker
) as Promise
<void>)
1011 const workerInfo
= this.getWorkerInfo(this.getWorkerNodeKey(worker
))
1012 workerInfo
.dynamic
= true
1013 this.sendToWorker(worker
, {
1015 workerId
: workerInfo
.id
as number
1021 * This function is the listener registered for each worker message.
1023 * @returns The listener function to execute when a message is received from a worker.
1025 protected workerListener (): (message
: MessageValue
<Response
>) => void {
1027 this.checkMessageWorkerId(message
)
1028 if (message
.ready
!= null && message
.workerId
!= null) {
1029 // Worker ready message received
1030 this.handleWorkerReadyMessage(message
)
1031 } else if (message
.id
!= null) {
1032 // Task execution response received
1033 this.handleTaskExecutionResponse(message
)
1038 private handleWorkerReadyMessage (message
: MessageValue
<Response
>): void {
1039 const worker
= this.getWorkerById(message
.workerId
)
1040 this.getWorkerInfo(this.getWorkerNodeKey(worker
as Worker
)).ready
=
1041 message
.ready
as boolean
1042 if (this.emitter
!= null && this.ready
) {
1043 this.emitter
.emit(PoolEvents
.ready
, this.info
)
1047 private handleTaskExecutionResponse (message
: MessageValue
<Response
>): void {
1048 const promiseResponse
= this.promiseResponseMap
.get(message
.id
as string)
1049 if (promiseResponse
!= null) {
1050 if (message
.taskError
!= null) {
1051 if (this.emitter
!= null) {
1052 this.emitter
.emit(PoolEvents
.taskError
, message
.taskError
)
1054 promiseResponse
.reject(message
.taskError
.message
)
1056 promiseResponse
.resolve(message
.data
as Response
)
1058 this.afterTaskExecutionHook(promiseResponse
.worker
, message
)
1059 this.promiseResponseMap
.delete(message
.id
as string)
1060 const workerNodeKey
= this.getWorkerNodeKey(promiseResponse
.worker
)
1062 this.opts
.enableTasksQueue
=== true &&
1063 this.tasksQueueSize(workerNodeKey
) > 0
1067 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1070 this.workerChoiceStrategyContext
.update(workerNodeKey
)
1074 private checkAndEmitEvents (): void {
1075 if (this.emitter
!= null) {
1077 this.emitter
.emit(PoolEvents
.busy
, this.info
)
1079 if (this.type === PoolTypes
.dynamic
&& this.full
) {
1080 this.emitter
.emit(PoolEvents
.full
, this.info
)
1086 * Gets the worker information.
1088 * @param workerNodeKey - The worker node key.
1090 private getWorkerInfo (workerNodeKey
: number): WorkerInfo
{
1091 return this.workerNodes
[workerNodeKey
].info
1095 * Pushes the given worker in the pool worker nodes.
1097 * @param worker - The worker.
1098 * @returns The worker nodes length.
1100 private pushWorkerNode (worker
: Worker
): number {
1101 return this.workerNodes
.push(new WorkerNode(worker
, this.worker
))
1105 * Removes the given worker from the pool worker nodes.
1107 * @param worker - The worker.
1109 private removeWorkerNode (worker
: Worker
): void {
1110 const workerNodeKey
= this.getWorkerNodeKey(worker
)
1111 if (workerNodeKey
!== -1) {
1112 this.workerNodes
.splice(workerNodeKey
, 1)
1113 this.workerChoiceStrategyContext
.remove(workerNodeKey
)
1117 private executeTask (workerNodeKey
: number, task
: Task
<Data
>): void {
1118 this.beforeTaskExecutionHook(workerNodeKey
, task
)
1119 this.sendToWorker(this.workerNodes
[workerNodeKey
].worker
, task
)
1122 private enqueueTask (workerNodeKey
: number, task
: Task
<Data
>): number {
1123 return this.workerNodes
[workerNodeKey
].enqueueTask(task
)
1126 private dequeueTask (workerNodeKey
: number): Task
<Data
> | undefined {
1127 return this.workerNodes
[workerNodeKey
].dequeueTask()
1130 private tasksQueueSize (workerNodeKey
: number): number {
1131 return this.workerNodes
[workerNodeKey
].tasksQueueSize()
1134 private flushTasksQueue (workerNodeKey
: number): void {
1135 while (this.tasksQueueSize(workerNodeKey
) > 0) {
1138 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1141 this.workerNodes
[workerNodeKey
].clearTasksQueue()
1144 private flushTasksQueues (): void {
1145 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
1146 this.flushTasksQueue(workerNodeKey
)
1150 private setWorkerStatistics (worker
: Worker
): void {
1151 this.sendToWorker(worker
, {
1154 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1156 elu
: this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1159 workerId
: this.getWorkerInfo(this.getWorkerNodeKey(worker
)).id
as number