1 import crypto 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'
13 import { CircularArray
} from
'../circular-array'
14 import { Queue
} from
'../queue'
23 type TasksQueueOptions
,
36 WorkerChoiceStrategies
,
37 type WorkerChoiceStrategy
,
38 type WorkerChoiceStrategyOptions
39 } from
'./selection-strategies/selection-strategies-types'
40 import { WorkerChoiceStrategyContext
} from
'./selection-strategies/worker-choice-strategy-context'
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<WorkerNode
<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 Error('Cannot instantiate a fixed pool with no worker')
160 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
161 if (isPlainObject(opts
)) {
162 this.opts
.workerChoiceStrategy
=
163 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
164 this.checkValidWorkerChoiceStrategy(this.opts
.workerChoiceStrategy
)
165 this.opts
.workerChoiceStrategyOptions
=
166 opts
.workerChoiceStrategyOptions
??
167 DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
168 this.checkValidWorkerChoiceStrategyOptions(
169 this.opts
.workerChoiceStrategyOptions
171 this.opts
.restartWorkerOnError
= opts
.restartWorkerOnError
?? true
172 this.opts
.enableEvents
= opts
.enableEvents
?? true
173 this.opts
.enableTasksQueue
= opts
.enableTasksQueue
?? false
174 if (this.opts
.enableTasksQueue
) {
175 this.checkValidTasksQueueOptions(
176 opts
.tasksQueueOptions
as TasksQueueOptions
178 this.opts
.tasksQueueOptions
= this.buildTasksQueueOptions(
179 opts
.tasksQueueOptions
as TasksQueueOptions
183 throw new TypeError('Invalid pool options: must be a plain object')
187 private checkValidWorkerChoiceStrategy (
188 workerChoiceStrategy
: WorkerChoiceStrategy
190 if (!Object.values(WorkerChoiceStrategies
).includes(workerChoiceStrategy
)) {
192 `Invalid worker choice strategy '${workerChoiceStrategy}'`
197 private checkValidWorkerChoiceStrategyOptions (
198 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
200 if (!isPlainObject(workerChoiceStrategyOptions
)) {
202 'Invalid worker choice strategy options: must be a plain object'
206 workerChoiceStrategyOptions
.weights
!= null &&
207 Object.keys(workerChoiceStrategyOptions
.weights
).length
!== this.maxSize
210 'Invalid worker choice strategy options: must have a weight for each worker node'
214 workerChoiceStrategyOptions
.measurement
!= null &&
215 !Object.values(Measurements
).includes(
216 workerChoiceStrategyOptions
.measurement
220 `Invalid worker choice strategy options: invalid measurement '${workerChoiceStrategyOptions.measurement}'`
225 private checkValidTasksQueueOptions (
226 tasksQueueOptions
: TasksQueueOptions
228 if (tasksQueueOptions
!= null && !isPlainObject(tasksQueueOptions
)) {
229 throw new TypeError('Invalid tasks queue options: must be a plain object')
232 tasksQueueOptions
?.concurrency
!= null &&
233 !Number.isSafeInteger(tasksQueueOptions
.concurrency
)
236 'Invalid worker tasks concurrency: must be an integer'
240 tasksQueueOptions
?.concurrency
!= null &&
241 tasksQueueOptions
.concurrency
<= 0
244 `Invalid worker tasks concurrency '${tasksQueueOptions.concurrency}'`
250 public get
info (): PoolInfo
{
254 minSize
: this.minSize
,
255 maxSize
: this.maxSize
,
256 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
257 .runTime
.aggregate
&&
258 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
259 .waitTime
.aggregate
&& { utilization
: round(this.utilization
) }),
260 workerNodes
: this.workerNodes
.length
,
261 idleWorkerNodes
: this.workerNodes
.reduce(
262 (accumulator
, workerNode
) =>
263 workerNode
.usage
.tasks
.executing
=== 0
268 busyWorkerNodes
: this.workerNodes
.reduce(
269 (accumulator
, workerNode
) =>
270 workerNode
.usage
.tasks
.executing
> 0 ? accumulator
+ 1 : accumulator
,
273 executedTasks
: this.workerNodes
.reduce(
274 (accumulator
, workerNode
) =>
275 accumulator
+ workerNode
.usage
.tasks
.executed
,
278 executingTasks
: this.workerNodes
.reduce(
279 (accumulator
, workerNode
) =>
280 accumulator
+ workerNode
.usage
.tasks
.executing
,
283 queuedTasks
: this.workerNodes
.reduce(
284 (accumulator
, workerNode
) =>
285 accumulator
+ workerNode
.usage
.tasks
.queued
,
288 maxQueuedTasks
: this.workerNodes
.reduce(
289 (accumulator
, workerNode
) =>
290 accumulator
+ workerNode
.usage
.tasks
.maxQueued
,
293 failedTasks
: this.workerNodes
.reduce(
294 (accumulator
, workerNode
) =>
295 accumulator
+ workerNode
.usage
.tasks
.failed
,
302 * Gets the pool run time.
304 * @returns The pool run time in milliseconds.
306 private get
runTime (): number {
307 return performance
.now() - this.startTimestamp
311 * Gets the approximate pool utilization.
313 * @returns The pool utilization.
315 private get
utilization (): number {
316 const poolRunTimeCapacity
= this.runTime
* this.maxSize
317 const totalTasksRunTime
= this.workerNodes
.reduce(
318 (accumulator
, workerNode
) =>
319 accumulator
+ workerNode
.usage
.runTime
.aggregate
,
322 const totalTasksWaitTime
= this.workerNodes
.reduce(
323 (accumulator
, workerNode
) =>
324 accumulator
+ workerNode
.usage
.waitTime
.aggregate
,
327 return (totalTasksRunTime
+ totalTasksWaitTime
) / poolRunTimeCapacity
333 * If it is `'dynamic'`, it provides the `max` property.
335 protected abstract get
type (): PoolType
338 * Gets the worker type.
340 protected abstract get
worker (): WorkerType
345 protected abstract get
minSize (): number
350 protected abstract get
maxSize (): number
353 * Get the worker given its id.
355 * @param workerId - The worker id.
356 * @returns The worker if found in the pool worker nodes, `undefined` otherwise.
358 private getWorkerById (workerId
: number): Worker
| undefined {
359 return this.workerNodes
.find(workerNode
=> workerNode
.info
.id
=== workerId
)
364 * Gets the given worker its worker node key.
366 * @param worker - The worker.
367 * @returns The worker node key if found in the pool worker nodes, `-1` otherwise.
369 private getWorkerNodeKey (worker
: Worker
): number {
370 return this.workerNodes
.findIndex(
371 workerNode
=> workerNode
.worker
=== worker
376 public setWorkerChoiceStrategy (
377 workerChoiceStrategy
: WorkerChoiceStrategy
,
378 workerChoiceStrategyOptions
?: WorkerChoiceStrategyOptions
380 this.checkValidWorkerChoiceStrategy(workerChoiceStrategy
)
381 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
382 this.workerChoiceStrategyContext
.setWorkerChoiceStrategy(
383 this.opts
.workerChoiceStrategy
385 if (workerChoiceStrategyOptions
!= null) {
386 this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
388 for (const [workerNodeKey
, workerNode
] of this.workerNodes
.entries()) {
389 this.setWorkerNodeTasksUsage(
391 this.getWorkerUsage(workerNodeKey
)
393 this.setWorkerStatistics(workerNode
.worker
)
398 public setWorkerChoiceStrategyOptions (
399 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
401 this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
402 this.opts
.workerChoiceStrategyOptions
= workerChoiceStrategyOptions
403 this.workerChoiceStrategyContext
.setOptions(
404 this.opts
.workerChoiceStrategyOptions
409 public enableTasksQueue (
411 tasksQueueOptions
?: TasksQueueOptions
413 if (this.opts
.enableTasksQueue
=== true && !enable
) {
414 this.flushTasksQueues()
416 this.opts
.enableTasksQueue
= enable
417 this.setTasksQueueOptions(tasksQueueOptions
as TasksQueueOptions
)
421 public setTasksQueueOptions (tasksQueueOptions
: TasksQueueOptions
): void {
422 if (this.opts
.enableTasksQueue
=== true) {
423 this.checkValidTasksQueueOptions(tasksQueueOptions
)
424 this.opts
.tasksQueueOptions
=
425 this.buildTasksQueueOptions(tasksQueueOptions
)
426 } else if (this.opts
.tasksQueueOptions
!= null) {
427 delete this.opts
.tasksQueueOptions
431 private buildTasksQueueOptions (
432 tasksQueueOptions
: TasksQueueOptions
433 ): TasksQueueOptions
{
435 concurrency
: tasksQueueOptions
?.concurrency
?? 1
440 * Whether the pool is full or not.
442 * The pool filling boolean status.
444 protected get
full (): boolean {
445 return this.workerNodes
.length
>= this.maxSize
449 * Whether the pool is busy or not.
451 * The pool busyness boolean status.
453 protected abstract get
busy (): boolean
456 * Whether worker nodes are executing at least one task.
458 * @returns Worker nodes busyness boolean status.
460 protected internalBusy (): boolean {
462 this.workerNodes
.findIndex(workerNode
=> {
463 return workerNode
.usage
.tasks
.executing
=== 0
469 public async execute (data
?: Data
, name
?: string): Promise
<Response
> {
470 const timestamp
= performance
.now()
471 const workerNodeKey
= this.chooseWorkerNode()
472 const submittedTask
: Task
<Data
> = {
474 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
475 data
: data
?? ({} as Data
),
477 id
: crypto
.randomUUID()
479 const res
= new Promise
<Response
>((resolve
, reject
) => {
480 this.promiseResponseMap
.set(submittedTask
.id
as string, {
483 worker
: this.workerNodes
[workerNodeKey
].worker
487 this.opts
.enableTasksQueue
=== true &&
489 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
>=
490 ((this.opts
.tasksQueueOptions
as TasksQueueOptions
)
491 .concurrency
as number))
493 this.enqueueTask(workerNodeKey
, submittedTask
)
495 this.executeTask(workerNodeKey
, submittedTask
)
497 this.checkAndEmitEvents()
498 // eslint-disable-next-line @typescript-eslint/return-await
503 public async destroy (): Promise
<void> {
505 this.workerNodes
.map(async (workerNode
, workerNodeKey
) => {
506 this.flushTasksQueue(workerNodeKey
)
507 // FIXME: wait for tasks to be finished
508 const workerExitPromise
= new Promise
<void>(resolve
=> {
509 workerNode
.worker
.on('exit', () => {
513 await this.destroyWorker(workerNode
.worker
)
514 await workerExitPromise
520 * Terminates the given worker.
522 * @param worker - A worker within `workerNodes`.
524 protected abstract destroyWorker (worker
: Worker
): void | Promise
<void>
527 * Setup hook to execute code before worker nodes are created in the abstract constructor.
532 protected setupHook (): void {
533 // Intentionally empty
537 * Should return whether the worker is the main worker or not.
539 protected abstract isMain (): boolean
542 * Hook executed before the worker task execution.
545 * @param workerNodeKey - The worker node key.
546 * @param task - The task to execute.
548 protected beforeTaskExecutionHook (
549 workerNodeKey
: number,
552 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
553 ++workerUsage
.tasks
.executing
554 this.updateWaitTimeWorkerUsage(workerUsage
, task
)
558 * Hook executed after the worker task execution.
561 * @param worker - The worker.
562 * @param message - The received message.
564 protected afterTaskExecutionHook (
566 message
: MessageValue
<Response
>
568 const workerUsage
= this.workerNodes
[this.getWorkerNodeKey(worker
)].usage
569 this.updateTaskStatisticsWorkerUsage(workerUsage
, message
)
570 this.updateRunTimeWorkerUsage(workerUsage
, message
)
571 this.updateEluWorkerUsage(workerUsage
, message
)
574 private updateTaskStatisticsWorkerUsage (
575 workerUsage
: WorkerUsage
,
576 message
: MessageValue
<Response
>
578 const workerTaskStatistics
= workerUsage
.tasks
579 --workerTaskStatistics
.executing
580 ++workerTaskStatistics
.executed
581 if (message
.taskError
!= null) {
582 ++workerTaskStatistics
.failed
586 private updateRunTimeWorkerUsage (
587 workerUsage
: WorkerUsage
,
588 message
: MessageValue
<Response
>
591 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
594 workerUsage
.runTime
.aggregate
+= message
.taskPerformance
?.runTime
?? 0
596 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
598 workerUsage
.tasks
.executed
!== 0
600 workerUsage
.runTime
.average
=
601 workerUsage
.runTime
.aggregate
/
602 (workerUsage
.tasks
.executed
- workerUsage
.tasks
.failed
)
605 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
607 message
.taskPerformance
?.runTime
!= null
609 workerUsage
.runTime
.history
.push(message
.taskPerformance
.runTime
)
610 workerUsage
.runTime
.median
= median(workerUsage
.runTime
.history
)
615 private updateWaitTimeWorkerUsage (
616 workerUsage
: WorkerUsage
,
619 const timestamp
= performance
.now()
620 const taskWaitTime
= timestamp
- (task
.timestamp
?? timestamp
)
622 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().waitTime
625 workerUsage
.waitTime
.aggregate
+= taskWaitTime
?? 0
627 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
629 workerUsage
.tasks
.executed
!== 0
631 workerUsage
.waitTime
.average
=
632 workerUsage
.waitTime
.aggregate
/
633 (workerUsage
.tasks
.executed
- workerUsage
.tasks
.failed
)
636 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
640 workerUsage
.waitTime
.history
.push(taskWaitTime
)
641 workerUsage
.waitTime
.median
= median(workerUsage
.waitTime
.history
)
646 private updateEluWorkerUsage (
647 workerUsage
: WorkerUsage
,
648 message
: MessageValue
<Response
>
651 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
654 if (workerUsage
.elu
!= null && message
.taskPerformance
?.elu
!= null) {
655 workerUsage
.elu
.idle
.aggregate
+= message
.taskPerformance
.elu
.idle
656 workerUsage
.elu
.active
.aggregate
+= message
.taskPerformance
.elu
.active
657 workerUsage
.elu
.utilization
=
658 (workerUsage
.elu
.utilization
+
659 message
.taskPerformance
.elu
.utilization
) /
661 } else if (message
.taskPerformance
?.elu
!= null) {
662 workerUsage
.elu
.idle
.aggregate
= message
.taskPerformance
.elu
.idle
663 workerUsage
.elu
.active
.aggregate
= message
.taskPerformance
.elu
.active
664 workerUsage
.elu
.utilization
= message
.taskPerformance
.elu
.utilization
667 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
669 workerUsage
.tasks
.executed
!== 0
671 const executedTasks
=
672 workerUsage
.tasks
.executed
- workerUsage
.tasks
.failed
673 workerUsage
.elu
.idle
.average
=
674 workerUsage
.elu
.idle
.aggregate
/ executedTasks
675 workerUsage
.elu
.active
.average
=
676 workerUsage
.elu
.active
.aggregate
/ executedTasks
679 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
681 message
.taskPerformance
?.elu
!= null
683 workerUsage
.elu
.idle
.history
.push(message
.taskPerformance
.elu
.idle
)
684 workerUsage
.elu
.active
.history
.push(message
.taskPerformance
.elu
.active
)
685 workerUsage
.elu
.idle
.median
= median(workerUsage
.elu
.idle
.history
)
686 workerUsage
.elu
.active
.median
= median(workerUsage
.elu
.active
.history
)
692 * Chooses a worker node for the next task.
694 * The default worker choice strategy uses a round robin algorithm to distribute the tasks.
696 * @returns The worker node key
698 private chooseWorkerNode (): number {
699 if (this.shallCreateDynamicWorker()) {
700 const worker
= this.createAndSetupDynamicWorker()
702 this.workerChoiceStrategyContext
.getStrategyPolicy().useDynamicWorker
704 return this.getWorkerNodeKey(worker
)
707 return this.workerChoiceStrategyContext
.execute()
711 * Conditions for dynamic worker creation.
713 * @returns Whether to create a dynamic worker or not.
715 private shallCreateDynamicWorker (): boolean {
716 return this.type === PoolTypes
.dynamic
&& !this.full
&& this.internalBusy()
720 * Sends a message to the given worker.
722 * @param worker - The worker which should receive the message.
723 * @param message - The message.
725 protected abstract sendToWorker (
727 message
: MessageValue
<Data
>
731 * Registers a listener callback on the given worker.
733 * @param worker - The worker which should register a listener.
734 * @param listener - The message listener callback.
736 private registerWorkerMessageListener
<Message
extends Data
| Response
>(
738 listener
: (message
: MessageValue
<Message
>) => void
740 worker
.on('message', listener
as MessageHandler
<Worker
>)
744 * Creates a new worker.
746 * @returns Newly created worker.
748 protected abstract createWorker (): Worker
751 * Function that can be hooked up when a worker has been newly created and moved to the pool worker nodes.
754 * @param worker - The newly created worker.
756 protected afterWorkerSetup (worker
: Worker
): void {
757 // Listen to worker messages.
758 this.registerWorkerMessageListener(worker
, this.workerListener())
762 * Creates a new worker and sets it up completely in the pool worker nodes.
764 * @returns New, completely set up worker.
766 protected createAndSetupWorker (): Worker
{
767 const worker
= this.createWorker()
769 worker
.on('message', this.opts
.messageHandler
?? EMPTY_FUNCTION
)
770 worker
.on('error', this.opts
.errorHandler
?? EMPTY_FUNCTION
)
771 worker
.on('error', error
=> {
772 if (this.emitter
!= null) {
773 this.emitter
.emit(PoolEvents
.error
, error
)
775 if (this.opts
.enableTasksQueue
=== true) {
776 const workerNodeKey
= this.getWorkerNodeKey(worker
)
777 while (this.tasksQueueSize(workerNodeKey
) > 0) {
778 let targetWorkerNodeKey
: number = workerNodeKey
779 let minQueuedTasks
= Infinity
780 for (const [workerNodeId
, workerNode
] of this.workerNodes
.entries()) {
782 workerNodeId
!== workerNodeKey
&&
783 workerNode
.usage
.tasks
.queued
=== 0
785 targetWorkerNodeKey
= workerNodeId
789 workerNodeId
!== workerNodeKey
&&
790 workerNode
.usage
.tasks
.queued
< minQueuedTasks
792 minQueuedTasks
= workerNode
.usage
.tasks
.queued
793 targetWorkerNodeKey
= workerNodeId
798 this.dequeueTask(workerNodeKey
) as Task
<Data
>
802 if (this.opts
.restartWorkerOnError
=== true) {
803 this.createAndSetupWorker()
806 worker
.on('online', this.opts
.onlineHandler
?? EMPTY_FUNCTION
)
807 worker
.on('exit', this.opts
.exitHandler
?? EMPTY_FUNCTION
)
808 worker
.once('exit', () => {
809 this.removeWorkerNode(worker
)
812 this.pushWorkerNode(worker
)
814 this.setWorkerStatistics(worker
)
816 this.afterWorkerSetup(worker
)
822 * Creates a new dynamic worker and sets it up completely in the pool worker nodes.
824 * @returns New, completely set up dynamic worker.
826 protected createAndSetupDynamicWorker (): Worker
{
827 const worker
= this.createAndSetupWorker()
828 this.registerWorkerMessageListener(worker
, message
=> {
829 const workerNodeKey
= this.getWorkerNodeKey(worker
)
831 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
832 (message
.kill
!= null &&
833 ((this.opts
.enableTasksQueue
=== false &&
834 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
=== 0) ||
835 (this.opts
.enableTasksQueue
=== true &&
836 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
=== 0 &&
837 this.tasksQueueSize(workerNodeKey
) === 0)))
839 // Kill message received from the worker: no new tasks are submitted to that worker for a while ( > maxInactiveTime)
840 void (this.destroyWorker(worker
) as Promise
<void>)
847 * This function is the listener registered for each worker message.
849 * @returns The listener function to execute when a message is received from a worker.
851 protected workerListener (): (message
: MessageValue
<Response
>) => void {
853 if (message
.workerId
!= null && message
.started
!= null) {
854 // Worker started message received
855 this.handleWorkerStartedMessage(message
)
856 } else if (message
.id
!= null) {
857 // Task execution response received
858 this.handleTaskExecutionResponse(message
)
863 private handleWorkerStartedMessage (message
: MessageValue
<Response
>): void {
864 // Worker started message received
865 const worker
= this.getWorkerById(message
.workerId
as number)
866 if (worker
!= null) {
867 this.workerNodes
[this.getWorkerNodeKey(worker
)].info
.started
=
868 message
.started
as boolean
871 `Worker started message received from unknown worker '${
872 message.workerId as number
878 private handleTaskExecutionResponse (message
: MessageValue
<Response
>): void {
879 const promiseResponse
= this.promiseResponseMap
.get(message
.id
as string)
880 if (promiseResponse
!= null) {
881 if (message
.taskError
!= null) {
882 if (this.emitter
!= null) {
883 this.emitter
.emit(PoolEvents
.taskError
, message
.taskError
)
885 promiseResponse
.reject(message
.taskError
.message
)
887 promiseResponse
.resolve(message
.data
as Response
)
889 this.afterTaskExecutionHook(promiseResponse
.worker
, message
)
890 this.promiseResponseMap
.delete(message
.id
as string)
891 const workerNodeKey
= this.getWorkerNodeKey(promiseResponse
.worker
)
893 this.opts
.enableTasksQueue
=== true &&
894 this.tasksQueueSize(workerNodeKey
) > 0
898 this.dequeueTask(workerNodeKey
) as Task
<Data
>
901 this.workerChoiceStrategyContext
.update(workerNodeKey
)
905 private checkAndEmitEvents (): void {
906 if (this.emitter
!= null) {
908 this.emitter
?.emit(PoolEvents
.busy
, this.info
)
910 if (this.type === PoolTypes
.dynamic
&& this.full
) {
911 this.emitter
?.emit(PoolEvents
.full
, this.info
)
917 * Sets the given worker node its tasks usage in the pool.
919 * @param workerNode - The worker node.
920 * @param workerUsage - The worker usage.
922 private setWorkerNodeTasksUsage (
923 workerNode
: WorkerNode
<Worker
, Data
>,
924 workerUsage
: WorkerUsage
926 workerNode
.usage
= workerUsage
930 * Pushes the given worker in the pool worker nodes.
932 * @param worker - The worker.
933 * @returns The worker nodes length.
935 private pushWorkerNode (worker
: Worker
): number {
936 this.workerNodes
.push({
938 info
: { id
: this.getWorkerId(worker
), started
: true },
939 usage
: this.getWorkerUsage(),
940 tasksQueue
: new Queue
<Task
<Data
>>()
942 const workerNodeKey
= this.getWorkerNodeKey(worker
)
943 this.setWorkerNodeTasksUsage(
944 this.workerNodes
[workerNodeKey
],
945 this.getWorkerUsage(workerNodeKey
)
947 return this.workerNodes
.length
951 * Gets the worker id.
953 * @param worker - The worker.
954 * @returns The worker id.
956 private getWorkerId (worker
: Worker
): number | undefined {
957 if (this.worker
=== WorkerTypes
.thread
) {
958 return worker
.threadId
959 } else if (this.worker
=== WorkerTypes
.cluster
) {
965 // * Sets the given worker in the pool worker nodes.
967 // * @param workerNodeKey - The worker node key.
968 // * @param worker - The worker.
969 // * @param workerInfo - The worker info.
970 // * @param workerUsage - The worker usage.
971 // * @param tasksQueue - The worker task queue.
973 // private setWorkerNode (
974 // workerNodeKey: number,
976 // workerInfo: WorkerInfo,
977 // workerUsage: WorkerUsage,
978 // tasksQueue: Queue<Task<Data>>
980 // this.workerNodes[workerNodeKey] = {
983 // usage: workerUsage,
989 * Removes the given worker from the pool worker nodes.
991 * @param worker - The worker.
993 private removeWorkerNode (worker
: Worker
): void {
994 const workerNodeKey
= this.getWorkerNodeKey(worker
)
995 if (workerNodeKey
!== -1) {
996 this.workerNodes
.splice(workerNodeKey
, 1)
997 this.workerChoiceStrategyContext
.remove(workerNodeKey
)
1001 private executeTask (workerNodeKey
: number, task
: Task
<Data
>): void {
1002 this.beforeTaskExecutionHook(workerNodeKey
, task
)
1003 this.sendToWorker(this.workerNodes
[workerNodeKey
].worker
, task
)
1006 private enqueueTask (workerNodeKey
: number, task
: Task
<Data
>): number {
1007 return this.workerNodes
[workerNodeKey
].tasksQueue
.enqueue(task
)
1010 private dequeueTask (workerNodeKey
: number): Task
<Data
> | undefined {
1011 return this.workerNodes
[workerNodeKey
].tasksQueue
.dequeue()
1014 private tasksQueueSize (workerNodeKey
: number): number {
1015 return this.workerNodes
[workerNodeKey
].tasksQueue
.size
1018 private tasksMaxQueueSize (workerNodeKey
: number): number {
1019 return this.workerNodes
[workerNodeKey
].tasksQueue
.maxSize
1022 private flushTasksQueue (workerNodeKey
: number): void {
1023 while (this.tasksQueueSize(workerNodeKey
) > 0) {
1026 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1029 this.workerNodes
[workerNodeKey
].tasksQueue
.clear()
1032 private flushTasksQueues (): void {
1033 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
1034 this.flushTasksQueue(workerNodeKey
)
1038 private setWorkerStatistics (worker
: Worker
): void {
1039 this.sendToWorker(worker
, {
1042 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1044 elu
: this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1050 private getWorkerUsage (workerNodeKey
?: number): WorkerUsage
{
1051 const getTasksQueueSize
= (workerNodeKey
?: number): number => {
1052 return workerNodeKey
!= null ? this.tasksQueueSize(workerNodeKey
) : 0
1054 const getTasksMaxQueueSize
= (workerNodeKey
?: number): number => {
1055 return workerNodeKey
!= null ? this.tasksMaxQueueSize(workerNodeKey
) : 0
1061 get
queued (): number {
1062 return getTasksQueueSize(workerNodeKey
)
1064 get
maxQueued (): number {
1065 return getTasksMaxQueueSize(workerNodeKey
)
1073 history
: new CircularArray()
1079 history
: new CircularArray()
1086 history
: new CircularArray()
1092 history
: new CircularArray()