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
<
76 PromiseResponseWrapper
<Worker
, Response
>
77 > = new Map
<string, PromiseResponseWrapper
<Worker
, Response
>>()
80 * Worker choice strategy context referencing a worker choice algorithm implementation.
82 protected workerChoiceStrategyContext
: WorkerChoiceStrategyContext
<
89 * Whether the pool is starting or not.
91 private readonly starting
: boolean
93 * The start timestamp of the pool.
95 private readonly startTimestamp
98 * Constructs a new poolifier pool.
100 * @param numberOfWorkers - Number of workers that this pool should manage.
101 * @param filePath - Path to the worker file.
102 * @param opts - Options for the pool.
105 protected readonly numberOfWorkers
: number,
106 protected readonly filePath
: string,
107 protected readonly opts
: PoolOptions
<Worker
>
109 if (!this.isMain()) {
110 throw new Error('Cannot start a pool from a worker!')
112 this.checkNumberOfWorkers(this.numberOfWorkers
)
113 this.checkFilePath(this.filePath
)
114 this.checkPoolOptions(this.opts
)
116 this.chooseWorkerNode
= this.chooseWorkerNode
.bind(this)
117 this.executeTask
= this.executeTask
.bind(this)
118 this.enqueueTask
= this.enqueueTask
.bind(this)
119 this.checkAndEmitEvents
= this.checkAndEmitEvents
.bind(this)
121 if (this.opts
.enableEvents
=== true) {
122 this.emitter
= new PoolEmitter()
124 this.workerChoiceStrategyContext
= new WorkerChoiceStrategyContext
<
130 this.opts
.workerChoiceStrategy
,
131 this.opts
.workerChoiceStrategyOptions
138 this.starting
= false
140 this.startTimestamp
= performance
.now()
143 private checkFilePath (filePath
: string): void {
146 typeof filePath
!== 'string' ||
147 (typeof filePath
=== 'string' && filePath
.trim().length
=== 0)
149 throw new Error('Please specify a file with a worker implementation')
151 if (!existsSync(filePath
)) {
152 throw new Error(`Cannot find the worker file '${filePath}'`)
156 private checkNumberOfWorkers (numberOfWorkers
: number): void {
157 if (numberOfWorkers
== null) {
159 'Cannot instantiate a pool without specifying the number of workers'
161 } else if (!Number.isSafeInteger(numberOfWorkers
)) {
163 'Cannot instantiate a pool with a non safe integer number of workers'
165 } else if (numberOfWorkers
< 0) {
166 throw new RangeError(
167 'Cannot instantiate a pool with a negative number of workers'
169 } else if (this.type === PoolTypes
.fixed
&& numberOfWorkers
=== 0) {
170 throw new RangeError('Cannot instantiate a fixed pool with zero worker')
174 protected checkDynamicPoolSize (min
: number, max
: number): void {
175 if (this.type === PoolTypes
.dynamic
) {
177 throw new RangeError(
178 'Cannot instantiate a dynamic pool with a maximum pool size inferior to the minimum pool size'
180 } else if (max
=== 0) {
181 throw new RangeError(
182 'Cannot instantiate a dynamic pool with a pool size equal to zero'
184 } else if (min
=== max
) {
185 throw new RangeError(
186 'Cannot instantiate a dynamic pool with a minimum pool size equal to the maximum pool size. Use a fixed pool instead'
192 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
193 if (isPlainObject(opts
)) {
194 this.opts
.workerChoiceStrategy
=
195 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
196 this.checkValidWorkerChoiceStrategy(this.opts
.workerChoiceStrategy
)
197 this.opts
.workerChoiceStrategyOptions
=
198 opts
.workerChoiceStrategyOptions
??
199 DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
200 this.checkValidWorkerChoiceStrategyOptions(
201 this.opts
.workerChoiceStrategyOptions
203 this.opts
.restartWorkerOnError
= opts
.restartWorkerOnError
?? true
204 this.opts
.enableEvents
= opts
.enableEvents
?? true
205 this.opts
.enableTasksQueue
= opts
.enableTasksQueue
?? false
206 if (this.opts
.enableTasksQueue
) {
207 this.checkValidTasksQueueOptions(
208 opts
.tasksQueueOptions
as TasksQueueOptions
210 this.opts
.tasksQueueOptions
= this.buildTasksQueueOptions(
211 opts
.tasksQueueOptions
as TasksQueueOptions
215 throw new TypeError('Invalid pool options: must be a plain object')
219 private checkValidWorkerChoiceStrategy (
220 workerChoiceStrategy
: WorkerChoiceStrategy
222 if (!Object.values(WorkerChoiceStrategies
).includes(workerChoiceStrategy
)) {
224 `Invalid worker choice strategy '${workerChoiceStrategy}'`
229 private checkValidWorkerChoiceStrategyOptions (
230 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
232 if (!isPlainObject(workerChoiceStrategyOptions
)) {
234 'Invalid worker choice strategy options: must be a plain object'
238 workerChoiceStrategyOptions
.weights
!= null &&
239 Object.keys(workerChoiceStrategyOptions
.weights
).length
!== this.maxSize
242 'Invalid worker choice strategy options: must have a weight for each worker node'
246 workerChoiceStrategyOptions
.measurement
!= null &&
247 !Object.values(Measurements
).includes(
248 workerChoiceStrategyOptions
.measurement
252 `Invalid worker choice strategy options: invalid measurement '${workerChoiceStrategyOptions.measurement}'`
257 private checkValidTasksQueueOptions (
258 tasksQueueOptions
: TasksQueueOptions
260 if (tasksQueueOptions
!= null && !isPlainObject(tasksQueueOptions
)) {
261 throw new TypeError('Invalid tasks queue options: must be a plain object')
264 tasksQueueOptions
?.concurrency
!= null &&
265 !Number.isSafeInteger(tasksQueueOptions
.concurrency
)
268 'Invalid worker tasks concurrency: must be an integer'
272 tasksQueueOptions
?.concurrency
!= null &&
273 tasksQueueOptions
.concurrency
<= 0
276 `Invalid worker tasks concurrency '${tasksQueueOptions.concurrency}'`
281 private startPool (): void {
283 this.workerNodes
.reduce(
284 (accumulator
, workerNode
) =>
285 !workerNode
.info
.dynamic
? accumulator
+ 1 : accumulator
,
287 ) < this.numberOfWorkers
289 this.createAndSetupWorker()
294 public get
info (): PoolInfo
{
300 strategy
: this.opts
.workerChoiceStrategy
as WorkerChoiceStrategy
,
301 minSize
: this.minSize
,
302 maxSize
: this.maxSize
,
303 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
304 .runTime
.aggregate
&&
305 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
306 .waitTime
.aggregate
&& { utilization
: round(this.utilization
) }),
307 workerNodes
: this.workerNodes
.length
,
308 idleWorkerNodes
: this.workerNodes
.reduce(
309 (accumulator
, workerNode
) =>
310 workerNode
.usage
.tasks
.executing
=== 0
315 busyWorkerNodes
: this.workerNodes
.reduce(
316 (accumulator
, workerNode
) =>
317 workerNode
.usage
.tasks
.executing
> 0 ? accumulator
+ 1 : accumulator
,
320 executedTasks
: this.workerNodes
.reduce(
321 (accumulator
, workerNode
) =>
322 accumulator
+ workerNode
.usage
.tasks
.executed
,
325 executingTasks
: this.workerNodes
.reduce(
326 (accumulator
, workerNode
) =>
327 accumulator
+ workerNode
.usage
.tasks
.executing
,
330 queuedTasks
: this.workerNodes
.reduce(
331 (accumulator
, workerNode
) =>
332 accumulator
+ workerNode
.usage
.tasks
.queued
,
335 maxQueuedTasks
: this.workerNodes
.reduce(
336 (accumulator
, workerNode
) =>
337 accumulator
+ (workerNode
.usage
.tasks
?.maxQueued
?? 0),
340 failedTasks
: this.workerNodes
.reduce(
341 (accumulator
, workerNode
) =>
342 accumulator
+ workerNode
.usage
.tasks
.failed
,
345 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
346 .runTime
.aggregate
&& {
350 ...this.workerNodes
.map(
351 workerNode
=> workerNode
.usage
.runTime
?.minimum
?? Infinity
357 ...this.workerNodes
.map(
358 workerNode
=> workerNode
.usage
.runTime
?.maximum
?? -Infinity
363 this.workerNodes
.reduce(
364 (accumulator
, workerNode
) =>
365 accumulator
+ (workerNode
.usage
.runTime
?.aggregate
?? 0),
368 this.workerNodes
.reduce(
369 (accumulator
, workerNode
) =>
370 accumulator
+ (workerNode
.usage
.tasks
?.executed
?? 0),
374 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
378 this.workerNodes
.map(
379 workerNode
=> workerNode
.usage
.runTime
?.median
?? 0
386 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
387 .waitTime
.aggregate
&& {
391 ...this.workerNodes
.map(
392 workerNode
=> workerNode
.usage
.waitTime
?.minimum
?? Infinity
398 ...this.workerNodes
.map(
399 workerNode
=> workerNode
.usage
.waitTime
?.maximum
?? -Infinity
404 this.workerNodes
.reduce(
405 (accumulator
, workerNode
) =>
406 accumulator
+ (workerNode
.usage
.waitTime
?.aggregate
?? 0),
409 this.workerNodes
.reduce(
410 (accumulator
, workerNode
) =>
411 accumulator
+ (workerNode
.usage
.tasks
?.executed
?? 0),
415 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
416 .waitTime
.median
&& {
419 this.workerNodes
.map(
420 workerNode
=> workerNode
.usage
.waitTime
?.median
?? 0
430 private get
ready (): boolean {
432 this.workerNodes
.reduce(
433 (accumulator
, workerNode
) =>
434 !workerNode
.info
.dynamic
&& workerNode
.info
.ready
443 * Gets the approximate pool utilization.
445 * @returns The pool utilization.
447 private get
utilization (): number {
448 const poolTimeCapacity
=
449 (performance
.now() - this.startTimestamp
) * this.maxSize
450 const totalTasksRunTime
= this.workerNodes
.reduce(
451 (accumulator
, workerNode
) =>
452 accumulator
+ (workerNode
.usage
.runTime
?.aggregate
?? 0),
455 const totalTasksWaitTime
= this.workerNodes
.reduce(
456 (accumulator
, workerNode
) =>
457 accumulator
+ (workerNode
.usage
.waitTime
?.aggregate
?? 0),
460 return (totalTasksRunTime
+ totalTasksWaitTime
) / poolTimeCapacity
466 * If it is `'dynamic'`, it provides the `max` property.
468 protected abstract get
type (): PoolType
471 * Gets the worker type.
473 protected abstract get
worker (): WorkerType
478 protected abstract get
minSize (): number
483 protected abstract get
maxSize (): number
486 * Get the worker given its id.
488 * @param workerId - The worker id.
489 * @returns The worker if found in the pool worker nodes, `undefined` otherwise.
491 private getWorkerById (workerId
: number): Worker
| undefined {
492 return this.workerNodes
.find(workerNode
=> workerNode
.info
.id
=== workerId
)
497 * Checks if the worker id sent in the received message from a worker is valid.
499 * @param message - The received message.
500 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the worker id is invalid.
502 private checkMessageWorkerId (message
: MessageValue
<Response
>): void {
504 message
.workerId
!= null &&
505 this.getWorkerById(message
.workerId
) == null
508 `Worker message received from unknown worker '${message.workerId}'`
514 * Gets the given worker its worker node key.
516 * @param worker - The worker.
517 * @returns The worker node key if found in the pool worker nodes, `-1` otherwise.
519 protected getWorkerNodeKey (worker
: Worker
): number {
520 return this.workerNodes
.findIndex(
521 workerNode
=> workerNode
.worker
=== worker
526 public setWorkerChoiceStrategy (
527 workerChoiceStrategy
: WorkerChoiceStrategy
,
528 workerChoiceStrategyOptions
?: WorkerChoiceStrategyOptions
530 this.checkValidWorkerChoiceStrategy(workerChoiceStrategy
)
531 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
532 this.workerChoiceStrategyContext
.setWorkerChoiceStrategy(
533 this.opts
.workerChoiceStrategy
535 if (workerChoiceStrategyOptions
!= null) {
536 this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
538 for (const workerNode
of this.workerNodes
) {
539 workerNode
.resetUsage()
540 this.setWorkerStatistics(workerNode
.worker
)
545 public setWorkerChoiceStrategyOptions (
546 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
548 this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
549 this.opts
.workerChoiceStrategyOptions
= workerChoiceStrategyOptions
550 this.workerChoiceStrategyContext
.setOptions(
551 this.opts
.workerChoiceStrategyOptions
556 public enableTasksQueue (
558 tasksQueueOptions
?: TasksQueueOptions
560 if (this.opts
.enableTasksQueue
=== true && !enable
) {
561 this.flushTasksQueues()
563 this.opts
.enableTasksQueue
= enable
564 this.setTasksQueueOptions(tasksQueueOptions
as TasksQueueOptions
)
568 public setTasksQueueOptions (tasksQueueOptions
: TasksQueueOptions
): void {
569 if (this.opts
.enableTasksQueue
=== true) {
570 this.checkValidTasksQueueOptions(tasksQueueOptions
)
571 this.opts
.tasksQueueOptions
=
572 this.buildTasksQueueOptions(tasksQueueOptions
)
573 } else if (this.opts
.tasksQueueOptions
!= null) {
574 delete this.opts
.tasksQueueOptions
578 private buildTasksQueueOptions (
579 tasksQueueOptions
: TasksQueueOptions
580 ): TasksQueueOptions
{
582 concurrency
: tasksQueueOptions
?.concurrency
?? 1
587 * Whether the pool is full or not.
589 * The pool filling boolean status.
591 protected get
full (): boolean {
592 return this.workerNodes
.length
>= this.maxSize
596 * Whether the pool is busy or not.
598 * The pool busyness boolean status.
600 protected abstract get
busy (): boolean
603 * Whether worker nodes are executing at least one task.
605 * @returns Worker nodes busyness boolean status.
607 protected internalBusy (): boolean {
609 this.workerNodes
.findIndex(workerNode
=> {
610 return workerNode
.usage
.tasks
.executing
=== 0
616 public async execute (data
?: Data
, name
?: string): Promise
<Response
> {
617 return await new Promise
<Response
>((resolve
, reject
) => {
618 const timestamp
= performance
.now()
619 const workerNodeKey
= this.chooseWorkerNode()
620 const submittedTask
: Task
<Data
> = {
621 name
: name
?? DEFAULT_TASK_NAME
,
622 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
623 data
: data
?? ({} as Data
),
625 workerId
: this.getWorkerInfo(workerNodeKey
).id
as number,
628 this.promiseResponseMap
.set(submittedTask
.id
as string, {
631 worker
: this.workerNodes
[workerNodeKey
].worker
634 this.opts
.enableTasksQueue
=== true &&
636 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
>=
637 ((this.opts
.tasksQueueOptions
as TasksQueueOptions
)
638 .concurrency
as number))
640 this.enqueueTask(workerNodeKey
, submittedTask
)
642 this.executeTask(workerNodeKey
, submittedTask
)
644 this.checkAndEmitEvents()
649 public async destroy (): Promise
<void> {
651 this.workerNodes
.map(async (workerNode
, workerNodeKey
) => {
652 this.flushTasksQueue(workerNodeKey
)
653 // FIXME: wait for tasks to be finished
654 const workerExitPromise
= new Promise
<void>(resolve
=> {
655 workerNode
.worker
.on('exit', () => {
659 await this.destroyWorker(workerNode
.worker
)
660 await workerExitPromise
666 * Terminates the given worker.
668 * @param worker - A worker within `workerNodes`.
670 protected abstract destroyWorker (worker
: Worker
): void | Promise
<void>
673 * Setup hook to execute code before worker nodes are created in the abstract constructor.
678 protected setupHook (): void {
679 // Intentionally empty
683 * Should return whether the worker is the main worker or not.
685 protected abstract isMain (): boolean
688 * Hook executed before the worker task execution.
691 * @param workerNodeKey - The worker node key.
692 * @param task - The task to execute.
694 protected beforeTaskExecutionHook (
695 workerNodeKey
: number,
698 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
699 ++workerUsage
.tasks
.executing
700 this.updateWaitTimeWorkerUsage(workerUsage
, task
)
701 const taskWorkerUsage
= this.workerNodes
[workerNodeKey
].getTaskWorkerUsage(
704 ++taskWorkerUsage
.tasks
.executing
705 this.updateWaitTimeWorkerUsage(taskWorkerUsage
, task
)
709 * Hook executed after the worker task execution.
712 * @param worker - The worker.
713 * @param message - The received message.
715 protected afterTaskExecutionHook (
717 message
: MessageValue
<Response
>
719 const workerNodeKey
= this.getWorkerNodeKey(worker
)
720 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
721 this.updateTaskStatisticsWorkerUsage(workerUsage
, message
)
722 this.updateRunTimeWorkerUsage(workerUsage
, message
)
723 this.updateEluWorkerUsage(workerUsage
, message
)
724 const taskWorkerUsage
= this.workerNodes
[workerNodeKey
].getTaskWorkerUsage(
725 message
.taskPerformance
?.name
?? DEFAULT_TASK_NAME
727 this.updateTaskStatisticsWorkerUsage(taskWorkerUsage
, message
)
728 this.updateRunTimeWorkerUsage(taskWorkerUsage
, message
)
729 this.updateEluWorkerUsage(taskWorkerUsage
, message
)
732 private updateTaskStatisticsWorkerUsage (
733 workerUsage
: WorkerUsage
,
734 message
: MessageValue
<Response
>
736 const workerTaskStatistics
= workerUsage
.tasks
737 --workerTaskStatistics
.executing
738 if (message
.taskError
== null) {
739 ++workerTaskStatistics
.executed
741 ++workerTaskStatistics
.failed
745 private updateRunTimeWorkerUsage (
746 workerUsage
: WorkerUsage
,
747 message
: MessageValue
<Response
>
749 updateMeasurementStatistics(
751 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
,
752 message
.taskPerformance
?.runTime
?? 0,
753 workerUsage
.tasks
.executed
757 private updateWaitTimeWorkerUsage (
758 workerUsage
: WorkerUsage
,
761 const timestamp
= performance
.now()
762 const taskWaitTime
= timestamp
- (task
.timestamp
?? timestamp
)
763 updateMeasurementStatistics(
764 workerUsage
.waitTime
,
765 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().waitTime
,
767 workerUsage
.tasks
.executed
771 private updateEluWorkerUsage (
772 workerUsage
: WorkerUsage
,
773 message
: MessageValue
<Response
>
775 const eluTaskStatisticsRequirements
: MeasurementStatisticsRequirements
=
776 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
777 updateMeasurementStatistics(
778 workerUsage
.elu
.active
,
779 eluTaskStatisticsRequirements
,
780 message
.taskPerformance
?.elu
?.active
?? 0,
781 workerUsage
.tasks
.executed
783 updateMeasurementStatistics(
784 workerUsage
.elu
.idle
,
785 eluTaskStatisticsRequirements
,
786 message
.taskPerformance
?.elu
?.idle
?? 0,
787 workerUsage
.tasks
.executed
789 if (eluTaskStatisticsRequirements
.aggregate
) {
790 if (message
.taskPerformance
?.elu
!= null) {
791 if (workerUsage
.elu
.utilization
!= null) {
792 workerUsage
.elu
.utilization
=
793 (workerUsage
.elu
.utilization
+
794 message
.taskPerformance
.elu
.utilization
) /
797 workerUsage
.elu
.utilization
= message
.taskPerformance
.elu
.utilization
804 * Chooses a worker node for the next task.
806 * The default worker choice strategy uses a round robin algorithm to distribute the tasks.
808 * @returns The worker node key
810 private chooseWorkerNode (): number {
811 if (this.shallCreateDynamicWorker()) {
812 const worker
= this.createAndSetupDynamicWorker()
814 this.workerChoiceStrategyContext
.getStrategyPolicy().useDynamicWorker
816 return this.getWorkerNodeKey(worker
)
819 return this.workerChoiceStrategyContext
.execute()
823 * Conditions for dynamic worker creation.
825 * @returns Whether to create a dynamic worker or not.
827 private shallCreateDynamicWorker (): boolean {
828 return this.type === PoolTypes
.dynamic
&& !this.full
&& this.internalBusy()
832 * Sends a message to the given worker.
834 * @param worker - The worker which should receive the message.
835 * @param message - The message.
837 protected abstract sendToWorker (
839 message
: MessageValue
<Data
>
843 * Creates a new worker.
845 * @returns Newly created worker.
847 protected abstract createWorker (): Worker
850 * Creates a new worker and sets it up completely in the pool worker nodes.
852 * @returns New, completely set up worker.
854 protected createAndSetupWorker (): Worker
{
855 const worker
= this.createWorker()
857 worker
.on('message', this.opts
.messageHandler
?? EMPTY_FUNCTION
)
858 worker
.on('error', this.opts
.errorHandler
?? EMPTY_FUNCTION
)
859 worker
.on('error', error
=> {
860 const workerNodeKey
= this.getWorkerNodeKey(worker
)
861 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
862 workerInfo
.ready
= false
863 this.workerNodes
[workerNodeKey
].closeChannel()
864 this.emitter
?.emit(PoolEvents
.error
, error
)
865 if (this.opts
.restartWorkerOnError
=== true && !this.starting
) {
866 if (workerInfo
.dynamic
) {
867 this.createAndSetupDynamicWorker()
869 this.createAndSetupWorker()
872 if (this.opts
.enableTasksQueue
=== true) {
873 this.redistributeQueuedTasks(workerNodeKey
)
876 worker
.on('online', this.opts
.onlineHandler
?? EMPTY_FUNCTION
)
877 worker
.on('exit', this.opts
.exitHandler
?? EMPTY_FUNCTION
)
878 worker
.once('exit', () => {
879 this.removeWorkerNode(worker
)
882 this.addWorkerNode(worker
)
884 this.afterWorkerSetup(worker
)
890 * Creates a new dynamic worker and sets it up completely in the pool worker nodes.
892 * @returns New, completely set up dynamic worker.
894 protected createAndSetupDynamicWorker (): Worker
{
895 const worker
= this.createAndSetupWorker()
896 this.registerWorkerMessageListener(worker
, message
=> {
897 const workerNodeKey
= this.getWorkerNodeKey(worker
)
899 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
900 (message
.kill
!= null &&
901 ((this.opts
.enableTasksQueue
=== false &&
902 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
=== 0) ||
903 (this.opts
.enableTasksQueue
=== true &&
904 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
=== 0 &&
905 this.tasksQueueSize(workerNodeKey
) === 0)))
907 // Kill message received from the worker: no new tasks are submitted to that worker for a while ( > maxInactiveTime)
908 void (this.destroyWorker(worker
) as Promise
<void>)
911 const workerInfo
= this.getWorkerInfoByWorker(worker
)
912 workerInfo
.dynamic
= true
913 if (this.workerChoiceStrategyContext
.getStrategyPolicy().useDynamicWorker
) {
914 workerInfo
.ready
= true
916 this.sendToWorker(worker
, {
918 workerId
: workerInfo
.id
as number
924 * Registers a listener callback on the given worker.
926 * @param worker - The worker which should register a listener.
927 * @param listener - The message listener callback.
929 protected abstract registerWorkerMessageListener
<
930 Message
extends Data
| Response
931 >(worker
: Worker
, listener
: (message
: MessageValue
<Message
>) => void): void
934 * Function that can be hooked up when a worker has been newly created and moved to the pool worker nodes.
937 * @param worker - The newly created worker.
939 protected afterWorkerSetup (worker
: Worker
): void {
940 // Listen to worker messages.
941 this.registerWorkerMessageListener(worker
, this.workerListener())
942 // Send the startup message to worker.
943 this.sendStartupMessageToWorker(worker
)
944 // Setup worker task statistics computation.
945 this.setWorkerStatistics(worker
)
949 * Sends the startup message to the given worker.
951 * @param worker - The worker which should receive the startup message.
953 protected abstract sendStartupMessageToWorker (worker
: Worker
): void
955 private redistributeQueuedTasks (workerNodeKey
: number): void {
956 while (this.tasksQueueSize(workerNodeKey
) > 0) {
957 let targetWorkerNodeKey
: number = workerNodeKey
958 let minQueuedTasks
= Infinity
959 for (const [workerNodeId
, workerNode
] of this.workerNodes
.entries()) {
960 const workerInfo
= this.getWorkerInfo(workerNodeId
)
962 workerNodeId
!== workerNodeKey
&&
964 workerNode
.usage
.tasks
.queued
=== 0
966 targetWorkerNodeKey
= workerNodeId
970 workerNodeId
!== workerNodeKey
&&
972 workerNode
.usage
.tasks
.queued
< minQueuedTasks
974 minQueuedTasks
= workerNode
.usage
.tasks
.queued
975 targetWorkerNodeKey
= workerNodeId
980 this.dequeueTask(workerNodeKey
) as Task
<Data
>
986 * This function is the listener registered for each worker message.
988 * @returns The listener function to execute when a message is received from a worker.
990 protected workerListener (): (message
: MessageValue
<Response
>) => void {
992 this.checkMessageWorkerId(message
)
993 if (message
.ready
!= null) {
994 // Worker ready response received
995 this.handleWorkerReadyResponse(message
)
996 } else if (message
.id
!= null) {
997 // Task execution response received
998 this.handleTaskExecutionResponse(message
)
1003 private handleWorkerReadyResponse (message
: MessageValue
<Response
>): void {
1004 this.getWorkerInfoByWorker(
1005 this.getWorkerById(message
.workerId
) as Worker
1006 ).ready
= message
.ready
as boolean
1007 if (this.emitter
!= null && this.ready
) {
1008 this.emitter
.emit(PoolEvents
.ready
, this.info
)
1012 private handleTaskExecutionResponse (message
: MessageValue
<Response
>): void {
1013 const promiseResponse
= this.promiseResponseMap
.get(message
.id
as string)
1014 if (promiseResponse
!= null) {
1015 if (message
.taskError
!= null) {
1016 this.emitter
?.emit(PoolEvents
.taskError
, message
.taskError
)
1017 promiseResponse
.reject(message
.taskError
.message
)
1019 promiseResponse
.resolve(message
.data
as Response
)
1021 this.afterTaskExecutionHook(promiseResponse
.worker
, message
)
1022 this.promiseResponseMap
.delete(message
.id
as string)
1023 const workerNodeKey
= this.getWorkerNodeKey(promiseResponse
.worker
)
1025 this.opts
.enableTasksQueue
=== true &&
1026 this.tasksQueueSize(workerNodeKey
) > 0
1030 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1033 this.workerChoiceStrategyContext
.update(workerNodeKey
)
1037 private checkAndEmitEvents (): void {
1038 if (this.emitter
!= null) {
1040 this.emitter
.emit(PoolEvents
.busy
, this.info
)
1042 if (this.type === PoolTypes
.dynamic
&& this.full
) {
1043 this.emitter
.emit(PoolEvents
.full
, this.info
)
1049 * Gets the worker information from the given worker node key.
1051 * @param workerNodeKey - The worker node key.
1052 * @returns The worker information.
1054 private getWorkerInfo (workerNodeKey
: number): WorkerInfo
{
1055 return this.workerNodes
[workerNodeKey
].info
1059 * Gets the worker information from the given worker.
1061 * @param worker - The worker.
1062 * @returns The worker information.
1063 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the worker is not found.
1065 protected getWorkerInfoByWorker (worker
: Worker
): WorkerInfo
{
1066 const workerNodeKey
= this.getWorkerNodeKey(worker
)
1067 if (workerNodeKey
=== -1) {
1068 throw new Error('Worker not found')
1070 return this.workerNodes
[workerNodeKey
].info
1074 * Adds the given worker in the pool worker nodes.
1076 * @param worker - The worker.
1077 * @returns The worker nodes length.
1079 private addWorkerNode (worker
: Worker
): number {
1080 const workerNode
= new WorkerNode
<Worker
, Data
>(worker
, this.worker
)
1081 // Flag the worker node as ready at pool startup.
1082 if (this.starting
) {
1083 workerNode
.info
.ready
= true
1085 return this.workerNodes
.push(workerNode
)
1089 * Removes the given worker from the pool worker nodes.
1091 * @param worker - The worker.
1093 private removeWorkerNode (worker
: Worker
): void {
1094 const workerNodeKey
= this.getWorkerNodeKey(worker
)
1095 if (workerNodeKey
!== -1) {
1096 this.workerNodes
.splice(workerNodeKey
, 1)
1097 this.workerChoiceStrategyContext
.remove(workerNodeKey
)
1102 * Executes the given task on the given worker.
1104 * @param worker - The worker.
1105 * @param task - The task to execute.
1107 private executeTask (workerNodeKey
: number, task
: Task
<Data
>): void {
1108 this.beforeTaskExecutionHook(workerNodeKey
, task
)
1109 this.sendToWorker(this.workerNodes
[workerNodeKey
].worker
, task
)
1112 private enqueueTask (workerNodeKey
: number, task
: Task
<Data
>): number {
1113 return this.workerNodes
[workerNodeKey
].enqueueTask(task
)
1116 private dequeueTask (workerNodeKey
: number): Task
<Data
> | undefined {
1117 return this.workerNodes
[workerNodeKey
].dequeueTask()
1120 private tasksQueueSize (workerNodeKey
: number): number {
1121 return this.workerNodes
[workerNodeKey
].tasksQueueSize()
1124 private flushTasksQueue (workerNodeKey
: number): void {
1125 while (this.tasksQueueSize(workerNodeKey
) > 0) {
1128 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1131 this.workerNodes
[workerNodeKey
].clearTasksQueue()
1134 private flushTasksQueues (): void {
1135 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
1136 this.flushTasksQueue(workerNodeKey
)
1140 private setWorkerStatistics (worker
: Worker
): void {
1141 this.sendToWorker(worker
, {
1144 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1146 elu
: this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1149 workerId
: this.getWorkerInfoByWorker(worker
).id
as number