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
,
35 WorkerChoiceStrategies
,
36 type WorkerChoiceStrategy
,
37 type WorkerChoiceStrategyOptions
38 } from
'./selection-strategies/selection-strategies-types'
39 import { WorkerChoiceStrategyContext
} from
'./selection-strategies/worker-choice-strategy-context'
42 * Base class that implements some shared logic for all poolifier pools.
44 * @typeParam Worker - Type of worker which manages this pool.
45 * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
46 * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
48 export abstract class AbstractPool
<
49 Worker
extends IWorker
,
52 > implements IPool
<Worker
, Data
, Response
> {
54 public readonly workerNodes
: Array<WorkerNode
<Worker
, Data
>> = []
57 public readonly emitter
?: PoolEmitter
60 * The execution response promise map.
62 * - `key`: The message id of each submitted task.
63 * - `value`: An object that contains the worker, the execution response promise resolve and reject callbacks.
65 * When we receive a message from the worker, we get a map entry with the promise resolve/reject bound to the message id.
67 protected promiseResponseMap
: Map
<
69 PromiseResponseWrapper
<Worker
, Response
>
70 > = new Map
<string, PromiseResponseWrapper
<Worker
, Response
>>()
73 * Worker choice strategy context referencing a worker choice algorithm implementation.
75 protected workerChoiceStrategyContext
: WorkerChoiceStrategyContext
<
82 * The start timestamp of the pool.
84 private readonly startTimestamp
87 * Constructs a new poolifier pool.
89 * @param numberOfWorkers - Number of workers that this pool should manage.
90 * @param filePath - Path to the worker file.
91 * @param opts - Options for the pool.
94 protected readonly numberOfWorkers
: number,
95 protected readonly filePath
: string,
96 protected readonly opts
: PoolOptions
<Worker
>
99 throw new Error('Cannot start a pool from a worker!')
101 this.checkNumberOfWorkers(this.numberOfWorkers
)
102 this.checkFilePath(this.filePath
)
103 this.checkPoolOptions(this.opts
)
105 this.chooseWorkerNode
= this.chooseWorkerNode
.bind(this)
106 this.executeTask
= this.executeTask
.bind(this)
107 this.enqueueTask
= this.enqueueTask
.bind(this)
108 this.checkAndEmitEvents
= this.checkAndEmitEvents
.bind(this)
110 if (this.opts
.enableEvents
=== true) {
111 this.emitter
= new PoolEmitter()
113 this.workerChoiceStrategyContext
= new WorkerChoiceStrategyContext
<
119 this.opts
.workerChoiceStrategy
,
120 this.opts
.workerChoiceStrategyOptions
125 while (this.workerNodes
.length
< this.numberOfWorkers
) {
126 this.createAndSetupWorker()
129 this.startTimestamp
= performance
.now()
132 private checkFilePath (filePath
: string): void {
135 (typeof filePath
=== 'string' && filePath
.trim().length
=== 0)
137 throw new Error('Please specify a file with a worker implementation')
141 private checkNumberOfWorkers (numberOfWorkers
: number): void {
142 if (numberOfWorkers
== null) {
144 'Cannot instantiate a pool without specifying the number of workers'
146 } else if (!Number.isSafeInteger(numberOfWorkers
)) {
148 'Cannot instantiate a pool with a non safe integer number of workers'
150 } else if (numberOfWorkers
< 0) {
151 throw new RangeError(
152 'Cannot instantiate a pool with a negative number of workers'
154 } else if (this.type === PoolTypes
.fixed
&& numberOfWorkers
=== 0) {
155 throw new Error('Cannot instantiate a fixed pool with no worker')
159 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
160 if (isPlainObject(opts
)) {
161 this.opts
.workerChoiceStrategy
=
162 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
163 this.checkValidWorkerChoiceStrategy(this.opts
.workerChoiceStrategy
)
164 this.opts
.workerChoiceStrategyOptions
=
165 opts
.workerChoiceStrategyOptions
??
166 DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
167 this.checkValidWorkerChoiceStrategyOptions(
168 this.opts
.workerChoiceStrategyOptions
170 this.opts
.restartWorkerOnError
= opts
.restartWorkerOnError
?? true
171 this.opts
.enableEvents
= opts
.enableEvents
?? true
172 this.opts
.enableTasksQueue
= opts
.enableTasksQueue
?? false
173 if (this.opts
.enableTasksQueue
) {
174 this.checkValidTasksQueueOptions(
175 opts
.tasksQueueOptions
as TasksQueueOptions
177 this.opts
.tasksQueueOptions
= this.buildTasksQueueOptions(
178 opts
.tasksQueueOptions
as TasksQueueOptions
182 throw new TypeError('Invalid pool options: must be a plain object')
186 private checkValidWorkerChoiceStrategy (
187 workerChoiceStrategy
: WorkerChoiceStrategy
189 if (!Object.values(WorkerChoiceStrategies
).includes(workerChoiceStrategy
)) {
191 `Invalid worker choice strategy '${workerChoiceStrategy}'`
196 private checkValidWorkerChoiceStrategyOptions (
197 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
199 if (!isPlainObject(workerChoiceStrategyOptions
)) {
201 'Invalid worker choice strategy options: must be a plain object'
205 workerChoiceStrategyOptions
.weights
!= null &&
206 Object.keys(workerChoiceStrategyOptions
.weights
).length
!== this.maxSize
209 'Invalid worker choice strategy options: must have a weight for each worker node'
213 workerChoiceStrategyOptions
.measurement
!= null &&
214 !Object.values(Measurements
).includes(
215 workerChoiceStrategyOptions
.measurement
219 `Invalid worker choice strategy options: invalid measurement '${workerChoiceStrategyOptions.measurement}'`
224 private checkValidTasksQueueOptions (
225 tasksQueueOptions
: TasksQueueOptions
227 if (tasksQueueOptions
!= null && !isPlainObject(tasksQueueOptions
)) {
228 throw new TypeError('Invalid tasks queue options: must be a plain object')
231 tasksQueueOptions
?.concurrency
!= null &&
232 !Number.isSafeInteger(tasksQueueOptions
.concurrency
)
235 'Invalid worker tasks concurrency: must be an integer'
239 tasksQueueOptions
?.concurrency
!= null &&
240 tasksQueueOptions
.concurrency
<= 0
243 `Invalid worker tasks concurrency '${tasksQueueOptions.concurrency}'`
249 public get
info (): PoolInfo
{
253 minSize
: this.minSize
,
254 maxSize
: this.maxSize
,
255 utilization
: round(this.utilization
),
256 workerNodes
: this.workerNodes
.length
,
257 idleWorkerNodes
: this.workerNodes
.reduce(
258 (accumulator
, workerNode
) =>
259 workerNode
.usage
.tasks
.executing
=== 0
264 busyWorkerNodes
: this.workerNodes
.reduce(
265 (accumulator
, workerNode
) =>
266 workerNode
.usage
.tasks
.executing
> 0 ? accumulator
+ 1 : accumulator
,
269 executedTasks
: this.workerNodes
.reduce(
270 (accumulator
, workerNode
) =>
271 accumulator
+ workerNode
.usage
.tasks
.executed
,
274 executingTasks
: this.workerNodes
.reduce(
275 (accumulator
, workerNode
) =>
276 accumulator
+ workerNode
.usage
.tasks
.executing
,
279 queuedTasks
: this.workerNodes
.reduce(
280 (accumulator
, workerNode
) =>
281 accumulator
+ workerNode
.usage
.tasks
.queued
,
284 maxQueuedTasks
: this.workerNodes
.reduce(
285 (accumulator
, workerNode
) =>
286 accumulator
+ workerNode
.usage
.tasks
.maxQueued
,
289 failedTasks
: this.workerNodes
.reduce(
290 (accumulator
, workerNode
) =>
291 accumulator
+ workerNode
.usage
.tasks
.failed
,
298 * Gets the pool run time.
300 * @returns The pool run time in milliseconds.
302 private get
runTime (): number {
303 return performance
.now() - this.startTimestamp
307 * Gets the approximate pool utilization.
309 * @returns The pool utilization.
311 private get
utilization (): number {
312 const poolRunTimeCapacity
= this.runTime
* this.maxSize
313 const totalTasksRunTime
= this.workerNodes
.reduce(
314 (accumulator
, workerNode
) =>
315 accumulator
+ workerNode
.usage
.runTime
.aggregate
,
318 const totalTasksWaitTime
= this.workerNodes
.reduce(
319 (accumulator
, workerNode
) =>
320 accumulator
+ workerNode
.usage
.waitTime
.aggregate
,
323 return (totalTasksRunTime
+ totalTasksWaitTime
) / poolRunTimeCapacity
329 * If it is `'dynamic'`, it provides the `max` property.
331 protected abstract get
type (): PoolType
334 * Gets the worker type.
336 protected abstract get
worker (): WorkerType
341 protected abstract get
minSize (): number
346 protected abstract get
maxSize (): number
349 * Get the worker given its id.
351 * @param workerId - The worker id.
352 * @returns The worker if found in the pool worker nodes, `undefined` otherwise.
354 private getWorkerById (workerId
: number): Worker
| undefined {
355 return this.workerNodes
.find(workerNode
=> workerNode
.info
.id
=== workerId
)
360 * Gets the given worker its worker node key.
362 * @param worker - The worker.
363 * @returns The worker node key if found in the pool worker nodes, `-1` otherwise.
365 private getWorkerNodeKey (worker
: Worker
): number {
366 return this.workerNodes
.findIndex(
367 workerNode
=> workerNode
.worker
=== worker
372 public setWorkerChoiceStrategy (
373 workerChoiceStrategy
: WorkerChoiceStrategy
,
374 workerChoiceStrategyOptions
?: WorkerChoiceStrategyOptions
376 this.checkValidWorkerChoiceStrategy(workerChoiceStrategy
)
377 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
378 this.workerChoiceStrategyContext
.setWorkerChoiceStrategy(
379 this.opts
.workerChoiceStrategy
381 if (workerChoiceStrategyOptions
!= null) {
382 this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
384 for (const [workerNodeKey
, workerNode
] of this.workerNodes
.entries()) {
385 this.setWorkerNodeTasksUsage(
387 this.getWorkerUsage(workerNodeKey
)
389 this.setWorkerStatistics(workerNode
.worker
)
394 public setWorkerChoiceStrategyOptions (
395 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
397 this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
398 this.opts
.workerChoiceStrategyOptions
= workerChoiceStrategyOptions
399 this.workerChoiceStrategyContext
.setOptions(
400 this.opts
.workerChoiceStrategyOptions
405 public enableTasksQueue (
407 tasksQueueOptions
?: TasksQueueOptions
409 if (this.opts
.enableTasksQueue
=== true && !enable
) {
410 this.flushTasksQueues()
412 this.opts
.enableTasksQueue
= enable
413 this.setTasksQueueOptions(tasksQueueOptions
as TasksQueueOptions
)
417 public setTasksQueueOptions (tasksQueueOptions
: TasksQueueOptions
): void {
418 if (this.opts
.enableTasksQueue
=== true) {
419 this.checkValidTasksQueueOptions(tasksQueueOptions
)
420 this.opts
.tasksQueueOptions
=
421 this.buildTasksQueueOptions(tasksQueueOptions
)
422 } else if (this.opts
.tasksQueueOptions
!= null) {
423 delete this.opts
.tasksQueueOptions
427 private buildTasksQueueOptions (
428 tasksQueueOptions
: TasksQueueOptions
429 ): TasksQueueOptions
{
431 concurrency
: tasksQueueOptions
?.concurrency
?? 1
436 * Whether the pool is full or not.
438 * The pool filling boolean status.
440 protected get
full (): boolean {
441 return this.workerNodes
.length
>= this.maxSize
445 * Whether the pool is busy or not.
447 * The pool busyness boolean status.
449 protected abstract get
busy (): boolean
452 * Whether worker nodes are executing at least one task.
454 * @returns Worker nodes busyness boolean status.
456 protected internalBusy (): boolean {
458 this.workerNodes
.findIndex(workerNode
=> {
459 return workerNode
.usage
.tasks
.executing
=== 0
465 public async execute (data
?: Data
, name
?: string): Promise
<Response
> {
466 const timestamp
= performance
.now()
467 const workerNodeKey
= this.chooseWorkerNode()
468 const submittedTask
: Task
<Data
> = {
470 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
471 data
: data
?? ({} as Data
),
473 id
: crypto
.randomUUID()
475 const res
= new Promise
<Response
>((resolve
, reject
) => {
476 this.promiseResponseMap
.set(submittedTask
.id
as string, {
479 worker
: this.workerNodes
[workerNodeKey
].worker
483 this.opts
.enableTasksQueue
=== true &&
485 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
>=
486 ((this.opts
.tasksQueueOptions
as TasksQueueOptions
)
487 .concurrency
as number))
489 this.enqueueTask(workerNodeKey
, submittedTask
)
491 this.executeTask(workerNodeKey
, submittedTask
)
493 this.checkAndEmitEvents()
494 // eslint-disable-next-line @typescript-eslint/return-await
499 public async destroy (): Promise
<void> {
501 this.workerNodes
.map(async (workerNode
, workerNodeKey
) => {
502 this.flushTasksQueue(workerNodeKey
)
503 // FIXME: wait for tasks to be finished
504 await this.destroyWorker(workerNode
.worker
)
510 * Terminates the given worker.
512 * @param worker - A worker within `workerNodes`.
514 protected abstract destroyWorker (worker
: Worker
): void | Promise
<void>
517 * Setup hook to execute code before worker nodes are created in the abstract constructor.
522 protected setupHook (): void {
523 // Intentionally empty
527 * Should return whether the worker is the main worker or not.
529 protected abstract isMain (): boolean
532 * Hook executed before the worker task execution.
535 * @param workerNodeKey - The worker node key.
536 * @param task - The task to execute.
538 protected beforeTaskExecutionHook (
539 workerNodeKey
: number,
542 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
543 ++workerUsage
.tasks
.executing
544 this.updateWaitTimeWorkerUsage(workerUsage
, task
)
548 * Hook executed after the worker task execution.
551 * @param worker - The worker.
552 * @param message - The received message.
554 protected afterTaskExecutionHook (
556 message
: MessageValue
<Response
>
558 const workerUsage
= this.workerNodes
[this.getWorkerNodeKey(worker
)].usage
559 this.updateTaskStatisticsWorkerUsage(workerUsage
, message
)
560 this.updateRunTimeWorkerUsage(workerUsage
, message
)
561 this.updateEluWorkerUsage(workerUsage
, message
)
564 private updateTaskStatisticsWorkerUsage (
565 workerUsage
: WorkerUsage
,
566 message
: MessageValue
<Response
>
568 const workerTaskStatistics
= workerUsage
.tasks
569 --workerTaskStatistics
.executing
570 ++workerTaskStatistics
.executed
571 if (message
.taskError
!= null) {
572 ++workerTaskStatistics
.failed
576 private updateRunTimeWorkerUsage (
577 workerUsage
: WorkerUsage
,
578 message
: MessageValue
<Response
>
581 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
584 workerUsage
.runTime
.aggregate
+= message
.taskPerformance
?.runTime
?? 0
586 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
588 workerUsage
.tasks
.executed
!== 0
590 workerUsage
.runTime
.average
=
591 workerUsage
.runTime
.aggregate
/
592 (workerUsage
.tasks
.executed
- workerUsage
.tasks
.failed
)
595 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
597 message
.taskPerformance
?.runTime
!= null
599 workerUsage
.runTime
.history
.push(message
.taskPerformance
.runTime
)
600 workerUsage
.runTime
.median
= median(workerUsage
.runTime
.history
)
605 private updateWaitTimeWorkerUsage (
606 workerUsage
: WorkerUsage
,
609 const timestamp
= performance
.now()
610 const taskWaitTime
= timestamp
- (task
.timestamp
?? timestamp
)
612 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().waitTime
615 workerUsage
.waitTime
.aggregate
+= taskWaitTime
?? 0
617 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
619 workerUsage
.tasks
.executed
!== 0
621 workerUsage
.waitTime
.average
=
622 workerUsage
.waitTime
.aggregate
/
623 (workerUsage
.tasks
.executed
- workerUsage
.tasks
.failed
)
626 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
630 workerUsage
.waitTime
.history
.push(taskWaitTime
)
631 workerUsage
.waitTime
.median
= median(workerUsage
.waitTime
.history
)
636 private updateEluWorkerUsage (
637 workerUsage
: WorkerUsage
,
638 message
: MessageValue
<Response
>
641 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
644 if (workerUsage
.elu
!= null && message
.taskPerformance
?.elu
!= null) {
645 workerUsage
.elu
.idle
.aggregate
+= message
.taskPerformance
.elu
.idle
646 workerUsage
.elu
.active
.aggregate
+= message
.taskPerformance
.elu
.active
647 workerUsage
.elu
.utilization
=
648 (workerUsage
.elu
.utilization
+
649 message
.taskPerformance
.elu
.utilization
) /
651 } else if (message
.taskPerformance
?.elu
!= null) {
652 workerUsage
.elu
.idle
.aggregate
= message
.taskPerformance
.elu
.idle
653 workerUsage
.elu
.active
.aggregate
= message
.taskPerformance
.elu
.active
654 workerUsage
.elu
.utilization
= message
.taskPerformance
.elu
.utilization
657 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
659 workerUsage
.tasks
.executed
!== 0
661 const executedTasks
=
662 workerUsage
.tasks
.executed
- workerUsage
.tasks
.failed
663 workerUsage
.elu
.idle
.average
=
664 workerUsage
.elu
.idle
.aggregate
/ executedTasks
665 workerUsage
.elu
.active
.average
=
666 workerUsage
.elu
.active
.aggregate
/ executedTasks
669 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
671 message
.taskPerformance
?.elu
!= null
673 workerUsage
.elu
.idle
.history
.push(message
.taskPerformance
.elu
.idle
)
674 workerUsage
.elu
.active
.history
.push(message
.taskPerformance
.elu
.active
)
675 workerUsage
.elu
.idle
.median
= median(workerUsage
.elu
.idle
.history
)
676 workerUsage
.elu
.active
.median
= median(workerUsage
.elu
.active
.history
)
682 * Chooses a worker node for the next task.
684 * The default worker choice strategy uses a round robin algorithm to distribute the tasks.
686 * @returns The worker node key
688 private chooseWorkerNode (): number {
689 if (this.shallCreateDynamicWorker()) {
690 const worker
= this.createAndSetupDynamicWorker()
692 this.workerChoiceStrategyContext
.getStrategyPolicy().useDynamicWorker
694 return this.getWorkerNodeKey(worker
)
697 return this.workerChoiceStrategyContext
.execute()
701 * Conditions for dynamic worker creation.
703 * @returns Whether to create a dynamic worker or not.
705 private shallCreateDynamicWorker (): boolean {
706 return this.type === PoolTypes
.dynamic
&& !this.full
&& this.internalBusy()
710 * Sends a message to the given worker.
712 * @param worker - The worker which should receive the message.
713 * @param message - The message.
715 protected abstract sendToWorker (
717 message
: MessageValue
<Data
>
721 * Registers a listener callback on the given worker.
723 * @param worker - The worker which should register a listener.
724 * @param listener - The message listener callback.
726 private registerWorkerMessageListener
<Message
extends Data
| Response
>(
728 listener
: (message
: MessageValue
<Message
>) => void
730 worker
.on('message', listener
as MessageHandler
<Worker
>)
734 * Creates a new worker.
736 * @returns Newly created worker.
738 protected abstract createWorker (): Worker
741 * Function that can be hooked up when a worker has been newly created and moved to the pool worker nodes.
744 * @param worker - The newly created worker.
746 protected afterWorkerSetup (worker
: Worker
): void {
747 // Listen to worker messages.
748 this.registerWorkerMessageListener(worker
, this.workerListener())
752 * Creates a new worker and sets it up completely in the pool worker nodes.
754 * @returns New, completely set up worker.
756 protected createAndSetupWorker (): Worker
{
757 const worker
= this.createWorker()
759 worker
.on('message', this.opts
.messageHandler
?? EMPTY_FUNCTION
)
760 worker
.on('error', this.opts
.errorHandler
?? EMPTY_FUNCTION
)
761 worker
.on('error', error
=> {
762 if (this.emitter
!= null) {
763 this.emitter
.emit(PoolEvents
.error
, error
)
765 if (this.opts
.restartWorkerOnError
=== true) {
766 this.createAndSetupWorker()
769 worker
.on('online', this.opts
.onlineHandler
?? EMPTY_FUNCTION
)
770 worker
.on('exit', this.opts
.exitHandler
?? EMPTY_FUNCTION
)
771 worker
.once('exit', () => {
772 this.removeWorkerNode(worker
)
775 this.pushWorkerNode(worker
)
777 this.setWorkerStatistics(worker
)
779 this.afterWorkerSetup(worker
)
785 * Creates a new dynamic worker and sets it up completely in the pool worker nodes.
787 * @returns New, completely set up dynamic worker.
789 protected createAndSetupDynamicWorker (): Worker
{
790 const worker
= this.createAndSetupWorker()
791 this.registerWorkerMessageListener(worker
, message
=> {
792 const workerNodeKey
= this.getWorkerNodeKey(worker
)
794 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
795 (message
.kill
!= null &&
796 ((this.opts
.enableTasksQueue
=== false &&
797 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
=== 0) ||
798 (this.opts
.enableTasksQueue
=== true &&
799 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
=== 0 &&
800 this.tasksQueueSize(workerNodeKey
) === 0)))
802 // Kill message received from the worker: no new tasks are submitted to that worker for a while ( > maxInactiveTime)
803 void (this.destroyWorker(worker
) as Promise
<void>)
810 * This function is the listener registered for each worker message.
812 * @returns The listener function to execute when a message is received from a worker.
814 protected workerListener (): (message
: MessageValue
<Response
>) => void {
816 if (message
.workerId
!= null && message
.started
!= null) {
817 // Worker started message received
818 const worker
= this.getWorkerById(message
.workerId
)
819 if (worker
!= null) {
820 this.workerNodes
[this.getWorkerNodeKey(worker
)].info
.started
=
824 `Worker started message received from unknown worker '${message.workerId}'`
827 } else if (message
.id
!= null) {
828 // Task execution response received
829 const promiseResponse
= this.promiseResponseMap
.get(message
.id
)
830 if (promiseResponse
!= null) {
831 if (message
.taskError
!= null) {
832 if (this.emitter
!= null) {
833 this.emitter
.emit(PoolEvents
.taskError
, message
.taskError
)
835 promiseResponse
.reject(
836 `${message.taskError.message} on worker '${message.taskError.workerId}'`
839 promiseResponse
.resolve(message
.data
as Response
)
841 this.afterTaskExecutionHook(promiseResponse
.worker
, message
)
842 this.promiseResponseMap
.delete(message
.id
)
843 const workerNodeKey
= this.getWorkerNodeKey(promiseResponse
.worker
)
845 this.opts
.enableTasksQueue
=== true &&
846 this.tasksQueueSize(workerNodeKey
) > 0
850 this.dequeueTask(workerNodeKey
) as Task
<Data
>
853 this.workerChoiceStrategyContext
.update(workerNodeKey
)
859 private checkAndEmitEvents (): void {
860 if (this.emitter
!= null) {
862 this.emitter
?.emit(PoolEvents
.busy
, this.info
)
864 if (this.type === PoolTypes
.dynamic
&& this.full
) {
865 this.emitter
?.emit(PoolEvents
.full
, this.info
)
871 * Sets the given worker node its tasks usage in the pool.
873 * @param workerNode - The worker node.
874 * @param workerUsage - The worker usage.
876 private setWorkerNodeTasksUsage (
877 workerNode
: WorkerNode
<Worker
, Data
>,
878 workerUsage
: WorkerUsage
880 workerNode
.usage
= workerUsage
884 * Pushes the given worker in the pool worker nodes.
886 * @param worker - The worker.
887 * @returns The worker nodes length.
889 private pushWorkerNode (worker
: Worker
): number {
890 this.workerNodes
.push({
892 info
: { id
: this.getWorkerId(worker
), started
: true },
893 usage
: this.getWorkerUsage(),
894 tasksQueue
: new Queue
<Task
<Data
>>()
896 const workerNodeKey
= this.getWorkerNodeKey(worker
)
897 this.setWorkerNodeTasksUsage(
898 this.workerNodes
[workerNodeKey
],
899 this.getWorkerUsage(workerNodeKey
)
901 return this.workerNodes
.length
905 * Gets the worker id.
907 * @param worker - The worker.
908 * @returns The worker id.
910 private getWorkerId (worker
: Worker
): number | undefined {
911 if (this.worker
=== WorkerTypes
.thread
) {
912 return worker
.threadId
913 } else if (this.worker
=== WorkerTypes
.cluster
) {
919 // * Sets the given worker in the pool worker nodes.
921 // * @param workerNodeKey - The worker node key.
922 // * @param worker - The worker.
923 // * @param workerInfo - The worker info.
924 // * @param workerUsage - The worker usage.
925 // * @param tasksQueue - The worker task queue.
927 // private setWorkerNode (
928 // workerNodeKey: number,
930 // workerInfo: WorkerInfo,
931 // workerUsage: WorkerUsage,
932 // tasksQueue: Queue<Task<Data>>
934 // this.workerNodes[workerNodeKey] = {
937 // usage: workerUsage,
943 * Removes the given worker from the pool worker nodes.
945 * @param worker - The worker.
947 private removeWorkerNode (worker
: Worker
): void {
948 const workerNodeKey
= this.getWorkerNodeKey(worker
)
949 if (workerNodeKey
!== -1) {
950 this.workerNodes
.splice(workerNodeKey
, 1)
951 this.workerChoiceStrategyContext
.remove(workerNodeKey
)
955 private executeTask (workerNodeKey
: number, task
: Task
<Data
>): void {
956 this.beforeTaskExecutionHook(workerNodeKey
, task
)
957 this.sendToWorker(this.workerNodes
[workerNodeKey
].worker
, task
)
960 private enqueueTask (workerNodeKey
: number, task
: Task
<Data
>): number {
961 return this.workerNodes
[workerNodeKey
].tasksQueue
.enqueue(task
)
964 private dequeueTask (workerNodeKey
: number): Task
<Data
> | undefined {
965 return this.workerNodes
[workerNodeKey
].tasksQueue
.dequeue()
968 private tasksQueueSize (workerNodeKey
: number): number {
969 return this.workerNodes
[workerNodeKey
].tasksQueue
.size
972 private tasksMaxQueueSize (workerNodeKey
: number): number {
973 return this.workerNodes
[workerNodeKey
].tasksQueue
.maxSize
976 private flushTasksQueue (workerNodeKey
: number): void {
977 if (this.tasksQueueSize(workerNodeKey
) > 0) {
978 for (let i
= 0; i
< this.tasksQueueSize(workerNodeKey
); i
++) {
981 this.dequeueTask(workerNodeKey
) as Task
<Data
>
985 this.workerNodes
[workerNodeKey
].tasksQueue
.clear()
988 private flushTasksQueues (): void {
989 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
990 this.flushTasksQueue(workerNodeKey
)
994 private setWorkerStatistics (worker
: Worker
): void {
995 this.sendToWorker(worker
, {
998 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1000 elu
: this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1006 private getWorkerUsage (workerNodeKey
?: number): WorkerUsage
{
1007 const getTasksQueueSize
= (workerNodeKey
?: number): number => {
1008 return workerNodeKey
!= null ? this.tasksQueueSize(workerNodeKey
) : 0
1010 const getTasksMaxQueueSize
= (workerNodeKey
?: number): number => {
1011 return workerNodeKey
!= null ? this.tasksMaxQueueSize(workerNodeKey
) : 0
1017 get
queued (): number {
1018 return getTasksQueueSize(workerNodeKey
)
1020 get
maxQueued (): number {
1021 return getTasksMaxQueueSize(workerNodeKey
)
1029 history
: new CircularArray()
1035 history
: new CircularArray()
1042 history
: new CircularArray()
1048 history
: new CircularArray()