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
,
11 import { KillBehaviors
, isKillBehavior
} from
'../worker/worker-options'
12 import { CircularArray
} from
'../circular-array'
13 import { Queue
} from
'../queue'
22 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'
41 * Base class that implements some shared logic for all poolifier pools.
43 * @typeParam Worker - Type of worker which manages this pool.
44 * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
45 * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
47 export abstract class AbstractPool
<
48 Worker
extends IWorker
,
51 > implements IPool
<Worker
, Data
, Response
> {
53 public readonly workerNodes
: Array<WorkerNode
<Worker
, Data
>> = []
56 public readonly emitter
?: PoolEmitter
59 * The execution response promise map.
61 * - `key`: The message id of each submitted task.
62 * - `value`: An object that contains the worker, the execution response promise resolve and reject callbacks.
64 * When we receive a message from the worker, we get a map entry with the promise resolve/reject bound to the message id.
66 protected promiseResponseMap
: Map
<
68 PromiseResponseWrapper
<Worker
, Response
>
69 > = new Map
<string, PromiseResponseWrapper
<Worker
, Response
>>()
72 * Worker choice strategy context referencing a worker choice algorithm implementation.
74 protected workerChoiceStrategyContext
: WorkerChoiceStrategyContext
<
81 * The start timestamp of the pool.
83 private readonly startTimestamp
86 * Constructs a new poolifier pool.
88 * @param numberOfWorkers - Number of workers that this pool should manage.
89 * @param filePath - Path to the worker file.
90 * @param opts - Options for the pool.
93 protected readonly numberOfWorkers
: number,
94 protected readonly filePath
: string,
95 protected readonly opts
: PoolOptions
<Worker
>
98 throw new Error('Cannot start a pool from a worker!')
100 this.checkNumberOfWorkers(this.numberOfWorkers
)
101 this.checkFilePath(this.filePath
)
102 this.checkPoolOptions(this.opts
)
104 this.chooseWorkerNode
= this.chooseWorkerNode
.bind(this)
105 this.executeTask
= this.executeTask
.bind(this)
106 this.enqueueTask
= this.enqueueTask
.bind(this)
107 this.checkAndEmitEvents
= this.checkAndEmitEvents
.bind(this)
109 if (this.opts
.enableEvents
=== true) {
110 this.emitter
= new PoolEmitter()
112 this.workerChoiceStrategyContext
= new WorkerChoiceStrategyContext
<
118 this.opts
.workerChoiceStrategy
,
119 this.opts
.workerChoiceStrategyOptions
124 while (this.workerNodes
.length
< this.numberOfWorkers
) {
125 this.createAndSetupWorker()
128 this.startTimestamp
= performance
.now()
131 private checkFilePath (filePath
: string): void {
134 (typeof filePath
=== 'string' && filePath
.trim().length
=== 0)
136 throw new Error('Please specify a file with a worker implementation')
140 private checkNumberOfWorkers (numberOfWorkers
: number): void {
141 if (numberOfWorkers
== null) {
143 'Cannot instantiate a pool without specifying the number of workers'
145 } else if (!Number.isSafeInteger(numberOfWorkers
)) {
147 'Cannot instantiate a pool with a non safe integer number of workers'
149 } else if (numberOfWorkers
< 0) {
150 throw new RangeError(
151 'Cannot instantiate a pool with a negative number of workers'
153 } else if (this.type === PoolTypes
.fixed
&& numberOfWorkers
=== 0) {
154 throw new Error('Cannot instantiate a fixed pool with no worker')
158 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
159 if (isPlainObject(opts
)) {
160 this.opts
.workerChoiceStrategy
=
161 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
162 this.checkValidWorkerChoiceStrategy(this.opts
.workerChoiceStrategy
)
163 this.opts
.workerChoiceStrategyOptions
=
164 opts
.workerChoiceStrategyOptions
??
165 DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
166 this.checkValidWorkerChoiceStrategyOptions(
167 this.opts
.workerChoiceStrategyOptions
169 this.opts
.restartWorkerOnError
= opts
.restartWorkerOnError
?? true
170 this.opts
.enableEvents
= opts
.enableEvents
?? true
171 this.opts
.enableTasksQueue
= opts
.enableTasksQueue
?? false
172 if (this.opts
.enableTasksQueue
) {
173 this.checkValidTasksQueueOptions(
174 opts
.tasksQueueOptions
as TasksQueueOptions
176 this.opts
.tasksQueueOptions
= this.buildTasksQueueOptions(
177 opts
.tasksQueueOptions
as TasksQueueOptions
181 throw new TypeError('Invalid pool options: must be a plain object')
185 private checkValidWorkerChoiceStrategy (
186 workerChoiceStrategy
: WorkerChoiceStrategy
188 if (!Object.values(WorkerChoiceStrategies
).includes(workerChoiceStrategy
)) {
190 `Invalid worker choice strategy '${workerChoiceStrategy}'`
195 private checkValidWorkerChoiceStrategyOptions (
196 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
198 if (!isPlainObject(workerChoiceStrategyOptions
)) {
200 'Invalid worker choice strategy options: must be a plain object'
204 workerChoiceStrategyOptions
.weights
!= null &&
205 Object.keys(workerChoiceStrategyOptions
.weights
).length
!== this.maxSize
208 'Invalid worker choice strategy options: must have a weight for each worker node'
212 workerChoiceStrategyOptions
.measurement
!= null &&
213 !Object.values(Measurements
).includes(
214 workerChoiceStrategyOptions
.measurement
218 `Invalid worker choice strategy options: invalid measurement '${workerChoiceStrategyOptions.measurement}'`
223 private checkValidTasksQueueOptions (
224 tasksQueueOptions
: TasksQueueOptions
226 if (tasksQueueOptions
!= null && !isPlainObject(tasksQueueOptions
)) {
227 throw new TypeError('Invalid tasks queue options: must be a plain object')
230 tasksQueueOptions
?.concurrency
!= null &&
231 !Number.isSafeInteger(tasksQueueOptions
.concurrency
)
234 'Invalid worker tasks concurrency: must be an integer'
238 tasksQueueOptions
?.concurrency
!= null &&
239 tasksQueueOptions
.concurrency
<= 0
242 `Invalid worker tasks concurrency '${tasksQueueOptions.concurrency}'`
248 public get
info (): PoolInfo
{
252 minSize
: this.minSize
,
253 maxSize
: this.maxSize
,
254 utilization
: round(this.utilization
),
255 workerNodes
: this.workerNodes
.length
,
256 idleWorkerNodes
: this.workerNodes
.reduce(
257 (accumulator
, workerNode
) =>
258 workerNode
.usage
.tasks
.executing
=== 0
263 busyWorkerNodes
: this.workerNodes
.reduce(
264 (accumulator
, workerNode
) =>
265 workerNode
.usage
.tasks
.executing
> 0 ? accumulator
+ 1 : accumulator
,
268 executedTasks
: this.workerNodes
.reduce(
269 (accumulator
, workerNode
) =>
270 accumulator
+ workerNode
.usage
.tasks
.executed
,
273 executingTasks
: this.workerNodes
.reduce(
274 (accumulator
, workerNode
) =>
275 accumulator
+ workerNode
.usage
.tasks
.executing
,
278 queuedTasks
: this.workerNodes
.reduce(
279 (accumulator
, workerNode
) =>
280 accumulator
+ workerNode
.usage
.tasks
.queued
,
283 maxQueuedTasks
: this.workerNodes
.reduce(
284 (accumulator
, workerNode
) =>
285 accumulator
+ workerNode
.usage
.tasks
.maxQueued
,
288 failedTasks
: this.workerNodes
.reduce(
289 (accumulator
, workerNode
) =>
290 accumulator
+ workerNode
.usage
.tasks
.failed
,
297 * Gets the pool run time.
299 * @returns The pool run time in milliseconds.
301 private get
runTime (): number {
302 return performance
.now() - this.startTimestamp
306 * Gets the approximate pool utilization.
308 * @returns The pool utilization.
310 private get
utilization (): number {
311 const poolRunTimeCapacity
= this.runTime
* this.maxSize
312 const totalTasksRunTime
= this.workerNodes
.reduce(
313 (accumulator
, workerNode
) =>
314 accumulator
+ workerNode
.usage
.runTime
.aggregate
,
317 const totalTasksWaitTime
= this.workerNodes
.reduce(
318 (accumulator
, workerNode
) =>
319 accumulator
+ workerNode
.usage
.waitTime
.aggregate
,
322 return (totalTasksRunTime
+ totalTasksWaitTime
) / poolRunTimeCapacity
328 * If it is `'dynamic'`, it provides the `max` property.
330 protected abstract get
type (): PoolType
333 * Gets the worker type.
335 protected abstract get
worker (): WorkerType
340 protected abstract get
minSize (): number
345 protected abstract get
maxSize (): number
348 * Gets the given worker its worker node key.
350 * @param worker - The worker.
351 * @returns The worker node key if found in the pool worker nodes, `-1` otherwise.
353 private getWorkerNodeKey (worker
: Worker
): number {
354 return this.workerNodes
.findIndex(
355 workerNode
=> workerNode
.worker
=== worker
360 public setWorkerChoiceStrategy (
361 workerChoiceStrategy
: WorkerChoiceStrategy
,
362 workerChoiceStrategyOptions
?: WorkerChoiceStrategyOptions
364 this.checkValidWorkerChoiceStrategy(workerChoiceStrategy
)
365 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
366 this.workerChoiceStrategyContext
.setWorkerChoiceStrategy(
367 this.opts
.workerChoiceStrategy
369 if (workerChoiceStrategyOptions
!= null) {
370 this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
372 for (const [workerNodeKey
, workerNode
] of this.workerNodes
.entries()) {
373 this.setWorkerNodeTasksUsage(
375 this.getWorkerUsage(workerNodeKey
)
377 this.setWorkerStatistics(workerNode
.worker
)
382 public setWorkerChoiceStrategyOptions (
383 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
385 this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
386 this.opts
.workerChoiceStrategyOptions
= workerChoiceStrategyOptions
387 this.workerChoiceStrategyContext
.setOptions(
388 this.opts
.workerChoiceStrategyOptions
393 public enableTasksQueue (
395 tasksQueueOptions
?: TasksQueueOptions
397 if (this.opts
.enableTasksQueue
=== true && !enable
) {
398 this.flushTasksQueues()
400 this.opts
.enableTasksQueue
= enable
401 this.setTasksQueueOptions(tasksQueueOptions
as TasksQueueOptions
)
405 public setTasksQueueOptions (tasksQueueOptions
: TasksQueueOptions
): void {
406 if (this.opts
.enableTasksQueue
=== true) {
407 this.checkValidTasksQueueOptions(tasksQueueOptions
)
408 this.opts
.tasksQueueOptions
=
409 this.buildTasksQueueOptions(tasksQueueOptions
)
410 } else if (this.opts
.tasksQueueOptions
!= null) {
411 delete this.opts
.tasksQueueOptions
415 private buildTasksQueueOptions (
416 tasksQueueOptions
: TasksQueueOptions
417 ): TasksQueueOptions
{
419 concurrency
: tasksQueueOptions
?.concurrency
?? 1
424 * Whether the pool is full or not.
426 * The pool filling boolean status.
428 protected get
full (): boolean {
429 return this.workerNodes
.length
>= this.maxSize
433 * Whether the pool is busy or not.
435 * The pool busyness boolean status.
437 protected abstract get
busy (): boolean
440 * Whether worker nodes are executing at least one task.
442 * @returns Worker nodes busyness boolean status.
444 protected internalBusy (): boolean {
446 this.workerNodes
.findIndex(workerNode
=> {
447 return workerNode
.usage
.tasks
.executing
=== 0
453 public async execute (data
?: Data
, name
?: string): Promise
<Response
> {
454 const timestamp
= performance
.now()
455 const workerNodeKey
= this.chooseWorkerNode()
456 const submittedTask
: Task
<Data
> = {
458 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
459 data
: data
?? ({} as Data
),
461 id
: crypto
.randomUUID()
463 const res
= new Promise
<Response
>((resolve
, reject
) => {
464 this.promiseResponseMap
.set(submittedTask
.id
as string, {
467 worker
: this.workerNodes
[workerNodeKey
].worker
471 this.opts
.enableTasksQueue
=== true &&
473 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
>=
474 ((this.opts
.tasksQueueOptions
as TasksQueueOptions
)
475 .concurrency
as number))
477 this.enqueueTask(workerNodeKey
, submittedTask
)
479 this.executeTask(workerNodeKey
, submittedTask
)
481 this.checkAndEmitEvents()
482 // eslint-disable-next-line @typescript-eslint/return-await
487 public async destroy (): Promise
<void> {
489 this.workerNodes
.map(async (workerNode
, workerNodeKey
) => {
490 this.flushTasksQueue(workerNodeKey
)
491 // FIXME: wait for tasks to be finished
492 await this.destroyWorker(workerNode
.worker
)
498 * Terminates the given worker.
500 * @param worker - A worker within `workerNodes`.
502 protected abstract destroyWorker (worker
: Worker
): void | Promise
<void>
505 * Setup hook to execute code before worker nodes are created in the abstract constructor.
510 protected setupHook (): void {
511 // Intentionally empty
515 * Should return whether the worker is the main worker or not.
517 protected abstract isMain (): boolean
520 * Hook executed before the worker task execution.
523 * @param workerNodeKey - The worker node key.
524 * @param task - The task to execute.
526 protected beforeTaskExecutionHook (
527 workerNodeKey
: number,
530 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
531 ++workerUsage
.tasks
.executing
532 this.updateWaitTimeWorkerUsage(workerUsage
, task
)
536 * Hook executed after the worker task execution.
539 * @param worker - The worker.
540 * @param message - The received message.
542 protected afterTaskExecutionHook (
544 message
: MessageValue
<Response
>
546 const workerUsage
= this.workerNodes
[this.getWorkerNodeKey(worker
)].usage
547 this.updateTaskStatisticsWorkerUsage(workerUsage
, message
)
548 this.updateRunTimeWorkerUsage(workerUsage
, message
)
549 this.updateEluWorkerUsage(workerUsage
, message
)
552 private updateTaskStatisticsWorkerUsage (
553 workerUsage
: WorkerUsage
,
554 message
: MessageValue
<Response
>
556 const workerTaskStatistics
= workerUsage
.tasks
557 --workerTaskStatistics
.executing
558 ++workerTaskStatistics
.executed
559 if (message
.taskError
!= null) {
560 ++workerTaskStatistics
.failed
564 private updateRunTimeWorkerUsage (
565 workerUsage
: WorkerUsage
,
566 message
: MessageValue
<Response
>
569 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
572 workerUsage
.runTime
.aggregate
+= message
.taskPerformance
?.runTime
?? 0
574 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
576 workerUsage
.tasks
.executed
!== 0
578 workerUsage
.runTime
.average
=
579 workerUsage
.runTime
.aggregate
/
580 (workerUsage
.tasks
.executed
- workerUsage
.tasks
.failed
)
583 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
585 message
.taskPerformance
?.runTime
!= null
587 workerUsage
.runTime
.history
.push(message
.taskPerformance
.runTime
)
588 workerUsage
.runTime
.median
= median(workerUsage
.runTime
.history
)
593 private updateWaitTimeWorkerUsage (
594 workerUsage
: WorkerUsage
,
597 const timestamp
= performance
.now()
598 const taskWaitTime
= timestamp
- (task
.timestamp
?? timestamp
)
600 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().waitTime
603 workerUsage
.waitTime
.aggregate
+= taskWaitTime
?? 0
605 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
607 workerUsage
.tasks
.executed
!== 0
609 workerUsage
.waitTime
.average
=
610 workerUsage
.waitTime
.aggregate
/
611 (workerUsage
.tasks
.executed
- workerUsage
.tasks
.failed
)
614 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
618 workerUsage
.waitTime
.history
.push(taskWaitTime
)
619 workerUsage
.waitTime
.median
= median(workerUsage
.waitTime
.history
)
624 private updateEluWorkerUsage (
625 workerUsage
: WorkerUsage
,
626 message
: MessageValue
<Response
>
629 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
632 if (workerUsage
.elu
!= null && message
.taskPerformance
?.elu
!= null) {
633 workerUsage
.elu
.idle
.aggregate
+= message
.taskPerformance
.elu
.idle
634 workerUsage
.elu
.active
.aggregate
+= message
.taskPerformance
.elu
.active
635 workerUsage
.elu
.utilization
=
636 (workerUsage
.elu
.utilization
+
637 message
.taskPerformance
.elu
.utilization
) /
639 } else if (message
.taskPerformance
?.elu
!= null) {
640 workerUsage
.elu
.idle
.aggregate
= message
.taskPerformance
.elu
.idle
641 workerUsage
.elu
.active
.aggregate
= message
.taskPerformance
.elu
.active
642 workerUsage
.elu
.utilization
= message
.taskPerformance
.elu
.utilization
645 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
647 workerUsage
.tasks
.executed
!== 0
649 const executedTasks
=
650 workerUsage
.tasks
.executed
- workerUsage
.tasks
.failed
651 workerUsage
.elu
.idle
.average
=
652 workerUsage
.elu
.idle
.aggregate
/ executedTasks
653 workerUsage
.elu
.active
.average
=
654 workerUsage
.elu
.active
.aggregate
/ executedTasks
657 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
659 message
.taskPerformance
?.elu
!= null
661 workerUsage
.elu
.idle
.history
.push(message
.taskPerformance
.elu
.idle
)
662 workerUsage
.elu
.active
.history
.push(message
.taskPerformance
.elu
.active
)
663 workerUsage
.elu
.idle
.median
= median(workerUsage
.elu
.idle
.history
)
664 workerUsage
.elu
.active
.median
= median(workerUsage
.elu
.active
.history
)
670 * Chooses a worker node for the next task.
672 * The default worker choice strategy uses a round robin algorithm to distribute the tasks.
674 * @returns The worker node key
676 private chooseWorkerNode (): number {
677 if (this.shallCreateDynamicWorker()) {
678 const worker
= this.createAndSetupDynamicWorker()
680 this.workerChoiceStrategyContext
.getStrategyPolicy().useDynamicWorker
682 return this.getWorkerNodeKey(worker
)
685 return this.workerChoiceStrategyContext
.execute()
689 * Conditions for dynamic worker creation.
691 * @returns Whether to create a dynamic worker or not.
693 private shallCreateDynamicWorker (): boolean {
694 return this.type === PoolTypes
.dynamic
&& !this.full
&& this.internalBusy()
698 * Sends a message to the given worker.
700 * @param worker - The worker which should receive the message.
701 * @param message - The message.
703 protected abstract sendToWorker (
705 message
: MessageValue
<Data
>
709 * Registers a listener callback on the given worker.
711 * @param worker - The worker which should register a listener.
712 * @param listener - The message listener callback.
714 private registerWorkerMessageListener
<Message
extends Data
| Response
>(
716 listener
: (message
: MessageValue
<Message
>) => void
718 worker
.on('message', listener
as MessageHandler
<Worker
>)
722 * Creates a new worker.
724 * @returns Newly created worker.
726 protected abstract createWorker (): Worker
729 * Function that can be hooked up when a worker has been newly created and moved to the pool worker nodes.
732 * @param worker - The newly created worker.
734 protected afterWorkerSetup (worker
: Worker
): void {
735 // Listen to worker messages.
736 this.registerWorkerMessageListener(worker
, this.workerListener())
740 * Creates a new worker and sets it up completely in the pool worker nodes.
742 * @returns New, completely set up worker.
744 protected createAndSetupWorker (): Worker
{
745 const worker
= this.createWorker()
747 worker
.on('message', this.opts
.messageHandler
?? EMPTY_FUNCTION
)
748 worker
.on('error', this.opts
.errorHandler
?? EMPTY_FUNCTION
)
749 worker
.on('error', error
=> {
750 if (this.emitter
!= null) {
751 this.emitter
.emit(PoolEvents
.error
, error
)
753 if (this.opts
.restartWorkerOnError
=== true) {
754 this.createAndSetupWorker()
757 worker
.on('online', this.opts
.onlineHandler
?? EMPTY_FUNCTION
)
758 worker
.on('exit', this.opts
.exitHandler
?? EMPTY_FUNCTION
)
759 worker
.once('exit', () => {
760 this.removeWorkerNode(worker
)
763 this.pushWorkerNode(worker
)
765 this.setWorkerStatistics(worker
)
767 this.afterWorkerSetup(worker
)
773 * Creates a new dynamic worker and sets it up completely in the pool worker nodes.
775 * @returns New, completely set up dynamic worker.
777 protected createAndSetupDynamicWorker (): Worker
{
778 const worker
= this.createAndSetupWorker()
779 this.registerWorkerMessageListener(worker
, message
=> {
780 const workerNodeKey
= this.getWorkerNodeKey(worker
)
782 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
783 (message
.kill
!= null &&
784 ((this.opts
.enableTasksQueue
=== false &&
785 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
=== 0) ||
786 (this.opts
.enableTasksQueue
=== true &&
787 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
=== 0 &&
788 this.tasksQueueSize(workerNodeKey
) === 0)))
790 // Kill message received from the worker: no new tasks are submitted to that worker for a while ( > maxInactiveTime)
791 void (this.destroyWorker(worker
) as Promise
<void>)
798 * This function is the listener registered for each worker message.
800 * @returns The listener function to execute when a message is received from a worker.
802 protected workerListener (): (message
: MessageValue
<Response
>) => void {
804 if (message
.id
!= null) {
805 // Task execution response received
806 const promiseResponse
= this.promiseResponseMap
.get(message
.id
)
807 if (promiseResponse
!= null) {
808 if (message
.taskError
!= null) {
809 if (this.emitter
!= null) {
810 this.emitter
.emit(PoolEvents
.taskError
, message
.taskError
)
812 promiseResponse
.reject(message
.taskError
.message
)
814 promiseResponse
.resolve(message
.data
as Response
)
816 this.afterTaskExecutionHook(promiseResponse
.worker
, message
)
817 this.promiseResponseMap
.delete(message
.id
)
818 const workerNodeKey
= this.getWorkerNodeKey(promiseResponse
.worker
)
820 this.opts
.enableTasksQueue
=== true &&
821 this.tasksQueueSize(workerNodeKey
) > 0
825 this.dequeueTask(workerNodeKey
) as Task
<Data
>
828 this.workerChoiceStrategyContext
.update(workerNodeKey
)
834 private checkAndEmitEvents (): void {
835 if (this.emitter
!= null) {
837 this.emitter
?.emit(PoolEvents
.busy
, this.info
)
839 if (this.type === PoolTypes
.dynamic
&& this.full
) {
840 this.emitter
?.emit(PoolEvents
.full
, this.info
)
846 * Sets the given worker node its tasks usage in the pool.
848 * @param workerNode - The worker node.
849 * @param workerUsage - The worker usage.
851 private setWorkerNodeTasksUsage (
852 workerNode
: WorkerNode
<Worker
, Data
>,
853 workerUsage
: WorkerUsage
855 workerNode
.usage
= workerUsage
859 * Pushes the given worker in the pool worker nodes.
861 * @param worker - The worker.
862 * @returns The worker nodes length.
864 private pushWorkerNode (worker
: Worker
): number {
865 this.workerNodes
.push({
867 usage
: this.getWorkerUsage(),
868 tasksQueue
: new Queue
<Task
<Data
>>()
870 const workerNodeKey
= this.getWorkerNodeKey(worker
)
871 this.setWorkerNodeTasksUsage(
872 this.workerNodes
[workerNodeKey
],
873 this.getWorkerUsage(workerNodeKey
)
875 return this.workerNodes
.length
879 // * Sets the given worker in the pool worker nodes.
881 // * @param workerNodeKey - The worker node key.
882 // * @param worker - The worker.
883 // * @param workerUsage - The worker usage.
884 // * @param tasksQueue - The worker task queue.
886 // private setWorkerNode (
887 // workerNodeKey: number,
889 // workerUsage: WorkerUsage,
890 // tasksQueue: Queue<Task<Data>>
892 // this.workerNodes[workerNodeKey] = {
894 // usage: workerUsage,
900 * Removes the given worker from the pool worker nodes.
902 * @param worker - The worker.
904 private removeWorkerNode (worker
: Worker
): void {
905 const workerNodeKey
= this.getWorkerNodeKey(worker
)
906 if (workerNodeKey
!== -1) {
907 this.workerNodes
.splice(workerNodeKey
, 1)
908 this.workerChoiceStrategyContext
.remove(workerNodeKey
)
912 private executeTask (workerNodeKey
: number, task
: Task
<Data
>): void {
913 this.beforeTaskExecutionHook(workerNodeKey
, task
)
914 this.sendToWorker(this.workerNodes
[workerNodeKey
].worker
, task
)
917 private enqueueTask (workerNodeKey
: number, task
: Task
<Data
>): number {
918 return this.workerNodes
[workerNodeKey
].tasksQueue
.enqueue(task
)
921 private dequeueTask (workerNodeKey
: number): Task
<Data
> | undefined {
922 return this.workerNodes
[workerNodeKey
].tasksQueue
.dequeue()
925 private tasksQueueSize (workerNodeKey
: number): number {
926 return this.workerNodes
[workerNodeKey
].tasksQueue
.size
929 private tasksMaxQueueSize (workerNodeKey
: number): number {
930 return this.workerNodes
[workerNodeKey
].tasksQueue
.maxSize
933 private flushTasksQueue (workerNodeKey
: number): void {
934 if (this.tasksQueueSize(workerNodeKey
) > 0) {
935 for (let i
= 0; i
< this.tasksQueueSize(workerNodeKey
); i
++) {
938 this.dequeueTask(workerNodeKey
) as Task
<Data
>
942 this.workerNodes
[workerNodeKey
].tasksQueue
.clear()
945 private flushTasksQueues (): void {
946 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
947 this.flushTasksQueue(workerNodeKey
)
951 private setWorkerStatistics (worker
: Worker
): void {
952 this.sendToWorker(worker
, {
955 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
957 elu
: this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
963 private getWorkerUsage (workerNodeKey
?: number): WorkerUsage
{
964 const getTasksQueueSize
= (workerNodeKey
?: number): number => {
965 return workerNodeKey
!= null ? this.tasksQueueSize(workerNodeKey
) : 0
967 const getTasksMaxQueueSize
= (workerNodeKey
?: number): number => {
968 return workerNodeKey
!= null ? this.tasksMaxQueueSize(workerNodeKey
) : 0
974 get
queued (): number {
975 return getTasksQueueSize(workerNodeKey
)
977 get
maxQueued (): number {
978 return getTasksMaxQueueSize(workerNodeKey
)
986 history
: new CircularArray()
992 history
: new CircularArray()
999 history
: new CircularArray()
1005 history
: new CircularArray()