1 import { randomUUID
} from
'node:crypto'
2 import { performance
} from
'node:perf_hooks'
3 import { existsSync
} from
'node:fs'
6 PromiseResponseWrapper
,
8 } from
'../utility-types'
11 DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
,
17 updateMeasurementStatistics
19 import { KillBehaviors
} from
'../worker/worker-options'
28 type TasksQueueOptions
38 type MeasurementStatisticsRequirements
,
40 WorkerChoiceStrategies
,
41 type WorkerChoiceStrategy
,
42 type WorkerChoiceStrategyOptions
43 } from
'./selection-strategies/selection-strategies-types'
44 import { WorkerChoiceStrategyContext
} from
'./selection-strategies/worker-choice-strategy-context'
45 import { version
} from
'./version'
46 import { WorkerNode
} from
'./worker-node'
49 * Base class that implements some shared logic for all poolifier pools.
51 * @typeParam Worker - Type of worker which manages this pool.
52 * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
53 * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
55 export abstract class AbstractPool
<
56 Worker
extends IWorker
,
59 > implements IPool
<Worker
, Data
, Response
> {
61 public readonly workerNodes
: Array<IWorkerNode
<Worker
, Data
>> = []
64 public readonly emitter
?: PoolEmitter
67 * The task execution response promise map.
69 * - `key`: The message id of each submitted task.
70 * - `value`: An object that contains the worker, the execution response promise resolve and reject callbacks.
72 * When we receive a message from the worker, we get a map entry with the promise resolve/reject bound to the message id.
74 protected promiseResponseMap
: Map
<string, PromiseResponseWrapper
<Response
>> =
75 new Map
<string, PromiseResponseWrapper
<Response
>>()
78 * Worker choice strategy context referencing a worker choice algorithm implementation.
80 protected workerChoiceStrategyContext
: WorkerChoiceStrategyContext
<
87 * Whether the pool is starting or not.
89 private readonly starting
: boolean
91 * The start timestamp of the pool.
93 private readonly startTimestamp
96 * Constructs a new poolifier pool.
98 * @param numberOfWorkers - Number of workers that this pool should manage.
99 * @param filePath - Path to the worker file.
100 * @param opts - Options for the pool.
103 protected readonly numberOfWorkers
: number,
104 protected readonly filePath
: string,
105 protected readonly opts
: PoolOptions
<Worker
>
107 if (!this.isMain()) {
108 throw new Error('Cannot start a pool from a worker!')
110 this.checkNumberOfWorkers(this.numberOfWorkers
)
111 this.checkFilePath(this.filePath
)
112 this.checkPoolOptions(this.opts
)
114 this.chooseWorkerNode
= this.chooseWorkerNode
.bind(this)
115 this.executeTask
= this.executeTask
.bind(this)
116 this.enqueueTask
= this.enqueueTask
.bind(this)
117 this.checkAndEmitEvents
= this.checkAndEmitEvents
.bind(this)
119 if (this.opts
.enableEvents
=== true) {
120 this.emitter
= new PoolEmitter()
122 this.workerChoiceStrategyContext
= new WorkerChoiceStrategyContext
<
128 this.opts
.workerChoiceStrategy
,
129 this.opts
.workerChoiceStrategyOptions
136 this.starting
= false
138 this.startTimestamp
= performance
.now()
141 private checkFilePath (filePath
: string): void {
144 typeof filePath
!== 'string' ||
145 (typeof filePath
=== 'string' && filePath
.trim().length
=== 0)
147 throw new Error('Please specify a file with a worker implementation')
149 if (!existsSync(filePath
)) {
150 throw new Error(`Cannot find the worker file '${filePath}'`)
154 private checkNumberOfWorkers (numberOfWorkers
: number): void {
155 if (numberOfWorkers
== null) {
157 'Cannot instantiate a pool without specifying the number of workers'
159 } else if (!Number.isSafeInteger(numberOfWorkers
)) {
161 'Cannot instantiate a pool with a non safe integer number of workers'
163 } else if (numberOfWorkers
< 0) {
164 throw new RangeError(
165 'Cannot instantiate a pool with a negative number of workers'
167 } else if (this.type === PoolTypes
.fixed
&& numberOfWorkers
=== 0) {
168 throw new RangeError('Cannot instantiate a fixed pool with zero worker')
172 protected checkDynamicPoolSize (min
: number, max
: number): void {
173 if (this.type === PoolTypes
.dynamic
) {
175 throw new RangeError(
176 'Cannot instantiate a dynamic pool with a maximum pool size inferior to the minimum pool size'
178 } else if (max
=== 0) {
179 throw new RangeError(
180 'Cannot instantiate a dynamic pool with a pool size equal to zero'
182 } else if (min
=== max
) {
183 throw new RangeError(
184 'Cannot instantiate a dynamic pool with a minimum pool size equal to the maximum pool size. Use a fixed pool instead'
190 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
191 if (isPlainObject(opts
)) {
192 this.opts
.workerChoiceStrategy
=
193 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
194 this.checkValidWorkerChoiceStrategy(this.opts
.workerChoiceStrategy
)
195 this.opts
.workerChoiceStrategyOptions
=
196 opts
.workerChoiceStrategyOptions
??
197 DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
198 this.checkValidWorkerChoiceStrategyOptions(
199 this.opts
.workerChoiceStrategyOptions
201 this.opts
.restartWorkerOnError
= opts
.restartWorkerOnError
?? true
202 this.opts
.enableEvents
= opts
.enableEvents
?? true
203 this.opts
.enableTasksQueue
= opts
.enableTasksQueue
?? false
204 if (this.opts
.enableTasksQueue
) {
205 this.checkValidTasksQueueOptions(
206 opts
.tasksQueueOptions
as TasksQueueOptions
208 this.opts
.tasksQueueOptions
= this.buildTasksQueueOptions(
209 opts
.tasksQueueOptions
as TasksQueueOptions
213 throw new TypeError('Invalid pool options: must be a plain object')
217 private checkValidWorkerChoiceStrategy (
218 workerChoiceStrategy
: WorkerChoiceStrategy
220 if (!Object.values(WorkerChoiceStrategies
).includes(workerChoiceStrategy
)) {
222 `Invalid worker choice strategy '${workerChoiceStrategy}'`
227 private checkValidWorkerChoiceStrategyOptions (
228 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
230 if (!isPlainObject(workerChoiceStrategyOptions
)) {
232 'Invalid worker choice strategy options: must be a plain object'
236 workerChoiceStrategyOptions
.weights
!= null &&
237 Object.keys(workerChoiceStrategyOptions
.weights
).length
!== this.maxSize
240 'Invalid worker choice strategy options: must have a weight for each worker node'
244 workerChoiceStrategyOptions
.measurement
!= null &&
245 !Object.values(Measurements
).includes(
246 workerChoiceStrategyOptions
.measurement
250 `Invalid worker choice strategy options: invalid measurement '${workerChoiceStrategyOptions.measurement}'`
255 private checkValidTasksQueueOptions (
256 tasksQueueOptions
: TasksQueueOptions
258 if (tasksQueueOptions
!= null && !isPlainObject(tasksQueueOptions
)) {
259 throw new TypeError('Invalid tasks queue options: must be a plain object')
262 tasksQueueOptions
?.concurrency
!= null &&
263 !Number.isSafeInteger(tasksQueueOptions
.concurrency
)
266 'Invalid worker tasks concurrency: must be an integer'
270 tasksQueueOptions
?.concurrency
!= null &&
271 tasksQueueOptions
.concurrency
<= 0
274 `Invalid worker tasks concurrency '${tasksQueueOptions.concurrency}'`
279 private startPool (): void {
281 this.workerNodes
.reduce(
282 (accumulator
, workerNode
) =>
283 !workerNode
.info
.dynamic
? accumulator
+ 1 : accumulator
,
285 ) < this.numberOfWorkers
287 this.createAndSetupWorkerNode()
292 public get
info (): PoolInfo
{
298 strategy
: this.opts
.workerChoiceStrategy
as WorkerChoiceStrategy
,
299 minSize
: this.minSize
,
300 maxSize
: this.maxSize
,
301 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
302 .runTime
.aggregate
&&
303 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
304 .waitTime
.aggregate
&& { utilization
: round(this.utilization
) }),
305 workerNodes
: this.workerNodes
.length
,
306 idleWorkerNodes
: this.workerNodes
.reduce(
307 (accumulator
, workerNode
) =>
308 workerNode
.usage
.tasks
.executing
=== 0
313 busyWorkerNodes
: this.workerNodes
.reduce(
314 (accumulator
, workerNode
) =>
315 workerNode
.usage
.tasks
.executing
> 0 ? accumulator
+ 1 : accumulator
,
318 executedTasks
: this.workerNodes
.reduce(
319 (accumulator
, workerNode
) =>
320 accumulator
+ workerNode
.usage
.tasks
.executed
,
323 executingTasks
: this.workerNodes
.reduce(
324 (accumulator
, workerNode
) =>
325 accumulator
+ workerNode
.usage
.tasks
.executing
,
328 queuedTasks
: this.workerNodes
.reduce(
329 (accumulator
, workerNode
) =>
330 accumulator
+ workerNode
.usage
.tasks
.queued
,
333 maxQueuedTasks
: this.workerNodes
.reduce(
334 (accumulator
, workerNode
) =>
335 accumulator
+ (workerNode
.usage
.tasks
?.maxQueued
?? 0),
338 failedTasks
: this.workerNodes
.reduce(
339 (accumulator
, workerNode
) =>
340 accumulator
+ workerNode
.usage
.tasks
.failed
,
343 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
344 .runTime
.aggregate
&& {
348 ...this.workerNodes
.map(
349 workerNode
=> workerNode
.usage
.runTime
?.minimum
?? Infinity
355 ...this.workerNodes
.map(
356 workerNode
=> workerNode
.usage
.runTime
?.maximum
?? -Infinity
361 this.workerNodes
.reduce(
362 (accumulator
, workerNode
) =>
363 accumulator
+ (workerNode
.usage
.runTime
?.aggregate
?? 0),
366 this.workerNodes
.reduce(
367 (accumulator
, workerNode
) =>
368 accumulator
+ (workerNode
.usage
.tasks
?.executed
?? 0),
372 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
376 this.workerNodes
.map(
377 workerNode
=> workerNode
.usage
.runTime
?.median
?? 0
384 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
385 .waitTime
.aggregate
&& {
389 ...this.workerNodes
.map(
390 workerNode
=> workerNode
.usage
.waitTime
?.minimum
?? Infinity
396 ...this.workerNodes
.map(
397 workerNode
=> workerNode
.usage
.waitTime
?.maximum
?? -Infinity
402 this.workerNodes
.reduce(
403 (accumulator
, workerNode
) =>
404 accumulator
+ (workerNode
.usage
.waitTime
?.aggregate
?? 0),
407 this.workerNodes
.reduce(
408 (accumulator
, workerNode
) =>
409 accumulator
+ (workerNode
.usage
.tasks
?.executed
?? 0),
413 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
414 .waitTime
.median
&& {
417 this.workerNodes
.map(
418 workerNode
=> workerNode
.usage
.waitTime
?.median
?? 0
429 * The pool readiness boolean status.
431 private get
ready (): boolean {
433 this.workerNodes
.reduce(
434 (accumulator
, workerNode
) =>
435 !workerNode
.info
.dynamic
&& workerNode
.info
.ready
444 * The approximate pool utilization.
446 * @returns The pool utilization.
448 private get
utilization (): number {
449 const poolTimeCapacity
=
450 (performance
.now() - this.startTimestamp
) * this.maxSize
451 const totalTasksRunTime
= this.workerNodes
.reduce(
452 (accumulator
, workerNode
) =>
453 accumulator
+ (workerNode
.usage
.runTime
?.aggregate
?? 0),
456 const totalTasksWaitTime
= this.workerNodes
.reduce(
457 (accumulator
, workerNode
) =>
458 accumulator
+ (workerNode
.usage
.waitTime
?.aggregate
?? 0),
461 return (totalTasksRunTime
+ totalTasksWaitTime
) / poolTimeCapacity
467 * If it is `'dynamic'`, it provides the `max` property.
469 protected abstract get
type (): PoolType
474 protected abstract get
worker (): WorkerType
477 * The pool minimum size.
479 protected abstract get
minSize (): number
482 * The pool maximum size.
484 protected abstract get
maxSize (): number
487 * Checks if the worker id sent in the received message from a worker is valid.
489 * @param message - The received message.
490 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the worker id is invalid.
492 private checkMessageWorkerId (message
: MessageValue
<Response
>): void {
494 message
.workerId
!= null &&
495 this.getWorkerNodeKeyByWorkerId(message
.workerId
) == null
498 `Worker message received from unknown worker '${message.workerId}'`
504 * Gets the given worker its worker node key.
506 * @param worker - The worker.
507 * @returns The worker node key if found in the pool worker nodes, `-1` otherwise.
509 private getWorkerNodeKey (worker
: Worker
): number {
510 return this.workerNodes
.findIndex(
511 workerNode
=> workerNode
.worker
=== worker
516 * Gets the worker node key given its worker id.
518 * @param workerId - The worker id.
519 * @returns The worker node key if the worker id is found in the pool worker nodes, `undefined` otherwise.
521 private getWorkerNodeKeyByWorkerId (workerId
: number): number | undefined {
522 for (const [workerNodeKey
, workerNode
] of this.workerNodes
.entries()) {
523 if (workerNode
.info
.id
=== workerId
) {
530 public setWorkerChoiceStrategy (
531 workerChoiceStrategy
: WorkerChoiceStrategy
,
532 workerChoiceStrategyOptions
?: WorkerChoiceStrategyOptions
534 this.checkValidWorkerChoiceStrategy(workerChoiceStrategy
)
535 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
536 this.workerChoiceStrategyContext
.setWorkerChoiceStrategy(
537 this.opts
.workerChoiceStrategy
539 if (workerChoiceStrategyOptions
!= null) {
540 this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
542 for (const [workerNodeKey
, workerNode
] of this.workerNodes
.entries()) {
543 workerNode
.resetUsage()
544 this.sendWorkerStatisticsMessageToWorker(workerNodeKey
)
549 public setWorkerChoiceStrategyOptions (
550 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
552 this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
553 this.opts
.workerChoiceStrategyOptions
= workerChoiceStrategyOptions
554 this.workerChoiceStrategyContext
.setOptions(
555 this.opts
.workerChoiceStrategyOptions
560 public enableTasksQueue (
562 tasksQueueOptions
?: TasksQueueOptions
564 if (this.opts
.enableTasksQueue
=== true && !enable
) {
565 this.flushTasksQueues()
567 this.opts
.enableTasksQueue
= enable
568 this.setTasksQueueOptions(tasksQueueOptions
as TasksQueueOptions
)
572 public setTasksQueueOptions (tasksQueueOptions
: TasksQueueOptions
): void {
573 if (this.opts
.enableTasksQueue
=== true) {
574 this.checkValidTasksQueueOptions(tasksQueueOptions
)
575 this.opts
.tasksQueueOptions
=
576 this.buildTasksQueueOptions(tasksQueueOptions
)
577 } else if (this.opts
.tasksQueueOptions
!= null) {
578 delete this.opts
.tasksQueueOptions
582 private buildTasksQueueOptions (
583 tasksQueueOptions
: TasksQueueOptions
584 ): TasksQueueOptions
{
586 concurrency
: tasksQueueOptions
?.concurrency
?? 1
591 * Whether the pool is full or not.
593 * The pool filling boolean status.
595 protected get
full (): boolean {
596 return this.workerNodes
.length
>= this.maxSize
600 * Whether the pool is busy or not.
602 * The pool busyness boolean status.
604 protected abstract get
busy (): boolean
607 * Whether worker nodes are executing at least one task.
609 * @returns Worker nodes busyness boolean status.
611 protected internalBusy (): boolean {
613 this.workerNodes
.findIndex(workerNode
=> {
614 return workerNode
.usage
.tasks
.executing
=== 0
620 public async execute (data
?: Data
, name
?: string): Promise
<Response
> {
621 return await new Promise
<Response
>((resolve
, reject
) => {
622 const timestamp
= performance
.now()
623 const workerNodeKey
= this.chooseWorkerNode()
624 const task
: Task
<Data
> = {
625 name
: name
?? DEFAULT_TASK_NAME
,
626 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
627 data
: data
?? ({} as Data
),
629 workerId
: this.getWorkerInfo(workerNodeKey
).id
as number,
632 this.promiseResponseMap
.set(task
.id
as string, {
638 this.opts
.enableTasksQueue
=== true &&
640 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
>=
641 ((this.opts
.tasksQueueOptions
as TasksQueueOptions
)
642 .concurrency
as number))
644 this.enqueueTask(workerNodeKey
, task
)
646 this.executeTask(workerNodeKey
, task
)
648 this.checkAndEmitEvents()
653 public async destroy (): Promise
<void> {
655 this.workerNodes
.map(async (workerNode
, workerNodeKey
) => {
656 this.flushTasksQueue(workerNodeKey
)
657 // FIXME: wait for tasks to be finished
658 const workerExitPromise
= new Promise
<void>(resolve
=> {
659 workerNode
.worker
.on('exit', () => {
663 await this.destroyWorkerNode(workerNodeKey
)
664 await workerExitPromise
670 * Terminates the worker node given its worker node key.
672 * @param workerNodeKey - The worker node key.
674 protected abstract destroyWorkerNode (
675 workerNodeKey
: number
676 ): void | Promise
<void>
679 * Setup hook to execute code before worker nodes are created in the abstract constructor.
684 protected setupHook (): void {
685 // Intentionally empty
689 * Should return whether the worker is the main worker or not.
691 protected abstract isMain (): boolean
694 * Hook executed before the worker task execution.
697 * @param workerNodeKey - The worker node key.
698 * @param task - The task to execute.
700 protected beforeTaskExecutionHook (
701 workerNodeKey
: number,
704 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
705 ++workerUsage
.tasks
.executing
706 this.updateWaitTimeWorkerUsage(workerUsage
, task
)
707 const taskWorkerUsage
= this.workerNodes
[workerNodeKey
].getTaskWorkerUsage(
710 ++taskWorkerUsage
.tasks
.executing
711 this.updateWaitTimeWorkerUsage(taskWorkerUsage
, task
)
715 * Hook executed after the worker task execution.
718 * @param workerNodeKey - The worker node key.
719 * @param message - The received message.
721 protected afterTaskExecutionHook (
722 workerNodeKey
: number,
723 message
: MessageValue
<Response
>
725 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
726 this.updateTaskStatisticsWorkerUsage(workerUsage
, message
)
727 this.updateRunTimeWorkerUsage(workerUsage
, message
)
728 this.updateEluWorkerUsage(workerUsage
, message
)
729 const taskWorkerUsage
= this.workerNodes
[workerNodeKey
].getTaskWorkerUsage(
730 message
.taskPerformance
?.name
?? DEFAULT_TASK_NAME
732 this.updateTaskStatisticsWorkerUsage(taskWorkerUsage
, message
)
733 this.updateRunTimeWorkerUsage(taskWorkerUsage
, message
)
734 this.updateEluWorkerUsage(taskWorkerUsage
, message
)
737 private updateTaskStatisticsWorkerUsage (
738 workerUsage
: WorkerUsage
,
739 message
: MessageValue
<Response
>
741 const workerTaskStatistics
= workerUsage
.tasks
742 --workerTaskStatistics
.executing
743 if (message
.taskError
== null) {
744 ++workerTaskStatistics
.executed
746 ++workerTaskStatistics
.failed
750 private updateRunTimeWorkerUsage (
751 workerUsage
: WorkerUsage
,
752 message
: MessageValue
<Response
>
754 updateMeasurementStatistics(
756 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
,
757 message
.taskPerformance
?.runTime
?? 0,
758 workerUsage
.tasks
.executed
762 private updateWaitTimeWorkerUsage (
763 workerUsage
: WorkerUsage
,
766 const timestamp
= performance
.now()
767 const taskWaitTime
= timestamp
- (task
.timestamp
?? timestamp
)
768 updateMeasurementStatistics(
769 workerUsage
.waitTime
,
770 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().waitTime
,
772 workerUsage
.tasks
.executed
776 private updateEluWorkerUsage (
777 workerUsage
: WorkerUsage
,
778 message
: MessageValue
<Response
>
780 const eluTaskStatisticsRequirements
: MeasurementStatisticsRequirements
=
781 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
782 updateMeasurementStatistics(
783 workerUsage
.elu
.active
,
784 eluTaskStatisticsRequirements
,
785 message
.taskPerformance
?.elu
?.active
?? 0,
786 workerUsage
.tasks
.executed
788 updateMeasurementStatistics(
789 workerUsage
.elu
.idle
,
790 eluTaskStatisticsRequirements
,
791 message
.taskPerformance
?.elu
?.idle
?? 0,
792 workerUsage
.tasks
.executed
794 if (eluTaskStatisticsRequirements
.aggregate
) {
795 if (message
.taskPerformance
?.elu
!= null) {
796 if (workerUsage
.elu
.utilization
!= null) {
797 workerUsage
.elu
.utilization
=
798 (workerUsage
.elu
.utilization
+
799 message
.taskPerformance
.elu
.utilization
) /
802 workerUsage
.elu
.utilization
= message
.taskPerformance
.elu
.utilization
809 * Chooses a worker node for the next task.
811 * The default worker choice strategy uses a round robin algorithm to distribute the tasks.
813 * @returns The chosen worker node key
815 private chooseWorkerNode (): number {
816 if (this.shallCreateDynamicWorker()) {
817 const workerNodeKey
= this.createAndSetupDynamicWorkerNode()
819 this.workerChoiceStrategyContext
.getStrategyPolicy().useDynamicWorker
824 return this.workerChoiceStrategyContext
.execute()
828 * Conditions for dynamic worker creation.
830 * @returns Whether to create a dynamic worker or not.
832 private shallCreateDynamicWorker (): boolean {
833 return this.type === PoolTypes
.dynamic
&& !this.full
&& this.internalBusy()
837 * Sends a message to worker given its worker node key.
839 * @param workerNodeKey - The worker node key.
840 * @param message - The message.
842 protected abstract sendToWorker (
843 workerNodeKey
: number,
844 message
: MessageValue
<Data
>
848 * Creates a new worker.
850 * @returns Newly created worker.
852 protected abstract createWorker (): Worker
855 * Creates a new, completely set up worker node.
857 * @returns New, completely set up worker node key.
859 protected createAndSetupWorkerNode (): number {
860 const worker
= this.createWorker()
862 worker
.on('message', this.opts
.messageHandler
?? EMPTY_FUNCTION
)
863 worker
.on('error', this.opts
.errorHandler
?? EMPTY_FUNCTION
)
864 worker
.on('error', error
=> {
865 const workerNodeKey
= this.getWorkerNodeKey(worker
)
866 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
867 workerInfo
.ready
= false
868 this.workerNodes
[workerNodeKey
].closeChannel()
869 this.emitter
?.emit(PoolEvents
.error
, error
)
870 if (this.opts
.restartWorkerOnError
=== true && !this.starting
) {
871 if (workerInfo
.dynamic
) {
872 this.createAndSetupDynamicWorkerNode()
874 this.createAndSetupWorkerNode()
877 if (this.opts
.enableTasksQueue
=== true) {
878 this.redistributeQueuedTasks(workerNodeKey
)
881 worker
.on('online', this.opts
.onlineHandler
?? EMPTY_FUNCTION
)
882 worker
.on('exit', this.opts
.exitHandler
?? EMPTY_FUNCTION
)
883 worker
.once('exit', () => {
884 this.removeWorkerNode(worker
)
887 const workerNodeKey
= this.addWorkerNode(worker
)
889 this.afterWorkerNodeSetup(workerNodeKey
)
895 * Creates a new, completely set up dynamic worker node.
897 * @returns New, completely set up dynamic worker node key.
899 protected createAndSetupDynamicWorkerNode (): number {
900 const workerNodeKey
= this.createAndSetupWorkerNode()
901 this.registerWorkerMessageListener(workerNodeKey
, message
=> {
902 const localWorkerNodeKey
= this.getWorkerNodeKeyByWorkerId(
905 const workerUsage
= this.workerNodes
[localWorkerNodeKey
].usage
907 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
908 (message
.kill
!= null &&
909 ((this.opts
.enableTasksQueue
=== false &&
910 workerUsage
.tasks
.executing
=== 0) ||
911 (this.opts
.enableTasksQueue
=== true &&
912 workerUsage
.tasks
.executing
=== 0 &&
913 this.tasksQueueSize(localWorkerNodeKey
) === 0)))
915 // Kill message received from the worker: no new tasks are submitted to that worker for a while ( > maxInactiveTime)
916 void (this.destroyWorkerNode(localWorkerNodeKey
) as Promise
<void>)
919 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
920 workerInfo
.dynamic
= true
921 if (this.workerChoiceStrategyContext
.getStrategyPolicy().useDynamicWorker
) {
922 workerInfo
.ready
= true
924 this.sendToWorker(workerNodeKey
, {
926 workerId
: workerInfo
.id
as number
932 * Registers a listener callback on the worker given its worker node key.
934 * @param workerNodeKey - The worker node key.
935 * @param listener - The message listener callback.
937 protected abstract registerWorkerMessageListener
<
938 Message
extends Data
| Response
940 workerNodeKey
: number,
941 listener
: (message
: MessageValue
<Message
>) => void
945 * Method hooked up after a worker node has been newly created.
948 * @param workerNodeKey - The newly created worker node key.
950 protected afterWorkerNodeSetup (workerNodeKey
: number): void {
951 // Listen to worker messages.
952 this.registerWorkerMessageListener(workerNodeKey
, this.workerListener())
953 // Send the startup message to worker.
954 this.sendStartupMessageToWorker(workerNodeKey
)
955 // Send the worker statistics message to worker.
956 this.sendWorkerStatisticsMessageToWorker(workerNodeKey
)
960 * Sends the startup message to worker given its worker node key.
962 * @param workerNodeKey - The worker node key.
964 protected abstract sendStartupMessageToWorker (workerNodeKey
: number): void
967 * Sends the worker statistics message to worker given its worker node key.
969 * @param workerNodeKey - The worker node key.
971 private sendWorkerStatisticsMessageToWorker (workerNodeKey
: number): void {
972 this.sendToWorker(workerNodeKey
, {
975 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
977 elu
: this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
980 workerId
: this.getWorkerInfo(workerNodeKey
).id
as number
984 private redistributeQueuedTasks (workerNodeKey
: number): void {
985 while (this.tasksQueueSize(workerNodeKey
) > 0) {
986 let targetWorkerNodeKey
: number = workerNodeKey
987 let minQueuedTasks
= Infinity
988 for (const [workerNodeId
, workerNode
] of this.workerNodes
.entries()) {
989 const workerInfo
= this.getWorkerInfo(workerNodeId
)
991 workerNodeId
!== workerNodeKey
&&
993 workerNode
.usage
.tasks
.queued
=== 0
995 targetWorkerNodeKey
= workerNodeId
999 workerNodeId
!== workerNodeKey
&&
1001 workerNode
.usage
.tasks
.queued
< minQueuedTasks
1003 minQueuedTasks
= workerNode
.usage
.tasks
.queued
1004 targetWorkerNodeKey
= workerNodeId
1008 targetWorkerNodeKey
,
1009 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1015 * This method is the listener registered for each worker message.
1017 * @returns The listener function to execute when a message is received from a worker.
1019 protected workerListener (): (message
: MessageValue
<Response
>) => void {
1021 this.checkMessageWorkerId(message
)
1022 if (message
.ready
!= null) {
1023 // Worker ready response received
1024 this.handleWorkerReadyResponse(message
)
1025 } else if (message
.id
!= null) {
1026 // Task execution response received
1027 this.handleTaskExecutionResponse(message
)
1032 private handleWorkerReadyResponse (message
: MessageValue
<Response
>): void {
1034 this.getWorkerNodeKeyByWorkerId(message
.workerId
) as number
1035 ).ready
= message
.ready
as boolean
1036 if (this.emitter
!= null && this.ready
) {
1037 this.emitter
.emit(PoolEvents
.ready
, this.info
)
1041 private handleTaskExecutionResponse (message
: MessageValue
<Response
>): void {
1042 const promiseResponse
= this.promiseResponseMap
.get(message
.id
as string)
1043 if (promiseResponse
!= null) {
1044 if (message
.taskError
!= null) {
1045 this.emitter
?.emit(PoolEvents
.taskError
, message
.taskError
)
1046 promiseResponse
.reject(message
.taskError
.message
)
1048 promiseResponse
.resolve(message
.data
as Response
)
1050 const workerNodeKey
= promiseResponse
.workerNodeKey
1051 this.afterTaskExecutionHook(workerNodeKey
, message
)
1052 this.promiseResponseMap
.delete(message
.id
as string)
1054 this.opts
.enableTasksQueue
=== true &&
1055 this.tasksQueueSize(workerNodeKey
) > 0
1059 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1062 this.workerChoiceStrategyContext
.update(workerNodeKey
)
1066 private checkAndEmitEvents (): void {
1067 if (this.emitter
!= null) {
1069 this.emitter
.emit(PoolEvents
.busy
, this.info
)
1071 if (this.type === PoolTypes
.dynamic
&& this.full
) {
1072 this.emitter
.emit(PoolEvents
.full
, this.info
)
1078 * Gets the worker information given its worker node key.
1080 * @param workerNodeKey - The worker node key.
1081 * @returns The worker information.
1083 protected getWorkerInfo (workerNodeKey
: number): WorkerInfo
{
1084 return this.workerNodes
[workerNodeKey
].info
1088 * Adds the given worker in the pool worker nodes.
1090 * @param worker - The worker.
1091 * @returns The added worker node key.
1092 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the added worker node is not found.
1094 private addWorkerNode (worker
: Worker
): number {
1095 const workerNode
= new WorkerNode
<Worker
, Data
>(worker
, this.worker
)
1096 // Flag the worker node as ready at pool startup.
1097 if (this.starting
) {
1098 workerNode
.info
.ready
= true
1100 this.workerNodes
.push(workerNode
)
1101 const workerNodeKey
= this.getWorkerNodeKey(worker
)
1102 if (workerNodeKey
=== -1) {
1103 throw new Error('Worker node not found')
1105 return workerNodeKey
1109 * Removes the given worker from the pool worker nodes.
1111 * @param worker - The worker.
1113 private removeWorkerNode (worker
: Worker
): void {
1114 const workerNodeKey
= this.getWorkerNodeKey(worker
)
1115 if (workerNodeKey
!== -1) {
1116 this.workerNodes
.splice(workerNodeKey
, 1)
1117 this.workerChoiceStrategyContext
.remove(workerNodeKey
)
1122 * Executes the given task on the worker given its worker node key.
1124 * @param workerNodeKey - The worker node key.
1125 * @param task - The task to execute.
1127 private executeTask (workerNodeKey
: number, task
: Task
<Data
>): void {
1128 this.beforeTaskExecutionHook(workerNodeKey
, task
)
1129 this.sendToWorker(workerNodeKey
, task
)
1132 private enqueueTask (workerNodeKey
: number, task
: Task
<Data
>): number {
1133 return this.workerNodes
[workerNodeKey
].enqueueTask(task
)
1136 private dequeueTask (workerNodeKey
: number): Task
<Data
> | undefined {
1137 return this.workerNodes
[workerNodeKey
].dequeueTask()
1140 private tasksQueueSize (workerNodeKey
: number): number {
1141 return this.workerNodes
[workerNodeKey
].tasksQueueSize()
1144 private flushTasksQueue (workerNodeKey
: number): void {
1145 while (this.tasksQueueSize(workerNodeKey
) > 0) {
1148 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1151 this.workerNodes
[workerNodeKey
].clearTasksQueue()
1154 private flushTasksQueues (): void {
1155 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
1156 this.flushTasksQueue(workerNodeKey
)