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
,
10 import { KillBehaviors
, isKillBehavior
} from
'../worker/worker-options'
11 import { CircularArray
} from
'../circular-array'
12 import { Queue
} from
'../queue'
21 type TasksQueueOptions
,
33 WorkerChoiceStrategies
,
34 type WorkerChoiceStrategy
,
35 type WorkerChoiceStrategyOptions
36 } from
'./selection-strategies/selection-strategies-types'
37 import { WorkerChoiceStrategyContext
} from
'./selection-strategies/worker-choice-strategy-context'
40 * Base class that implements some shared logic for all poolifier pools.
42 * @typeParam Worker - Type of worker which manages this pool.
43 * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
44 * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
46 export abstract class AbstractPool
<
47 Worker
extends IWorker
,
50 > implements IPool
<Worker
, Data
, Response
> {
52 public readonly workerNodes
: Array<WorkerNode
<Worker
, Data
>> = []
55 public readonly emitter
?: PoolEmitter
58 * The execution response promise map.
60 * - `key`: The message id of each submitted task.
61 * - `value`: An object that contains the worker, the execution response promise resolve and reject callbacks.
63 * When we receive a message from the worker, we get a map entry with the promise resolve/reject bound to the message id.
65 protected promiseResponseMap
: Map
<
67 PromiseResponseWrapper
<Worker
, Response
>
68 > = new Map
<string, PromiseResponseWrapper
<Worker
, Response
>>()
71 * Worker choice strategy context referencing a worker choice algorithm implementation.
73 protected workerChoiceStrategyContext
: WorkerChoiceStrategyContext
<
80 * Constructs a new poolifier pool.
82 * @param numberOfWorkers - Number of workers that this pool should manage.
83 * @param filePath - Path to the worker file.
84 * @param opts - Options for the pool.
87 protected readonly numberOfWorkers
: number,
88 protected readonly filePath
: string,
89 protected readonly opts
: PoolOptions
<Worker
>
92 throw new Error('Cannot start a pool from a worker!')
94 this.checkNumberOfWorkers(this.numberOfWorkers
)
95 this.checkFilePath(this.filePath
)
96 this.checkPoolOptions(this.opts
)
98 this.chooseWorkerNode
= this.chooseWorkerNode
.bind(this)
99 this.executeTask
= this.executeTask
.bind(this)
100 this.enqueueTask
= this.enqueueTask
.bind(this)
101 this.checkAndEmitEvents
= this.checkAndEmitEvents
.bind(this)
103 if (this.opts
.enableEvents
=== true) {
104 this.emitter
= new PoolEmitter()
106 this.workerChoiceStrategyContext
= new WorkerChoiceStrategyContext
<
112 this.opts
.workerChoiceStrategy
,
113 this.opts
.workerChoiceStrategyOptions
118 for (let i
= 1; i
<= this.numberOfWorkers
; i
++) {
119 this.createAndSetupWorker()
123 private checkFilePath (filePath
: string): void {
126 (typeof filePath
=== 'string' && filePath
.trim().length
=== 0)
128 throw new Error('Please specify a file with a worker implementation')
132 private checkNumberOfWorkers (numberOfWorkers
: number): void {
133 if (numberOfWorkers
== null) {
135 'Cannot instantiate a pool without specifying the number of workers'
137 } else if (!Number.isSafeInteger(numberOfWorkers
)) {
139 'Cannot instantiate a pool with a non safe integer number of workers'
141 } else if (numberOfWorkers
< 0) {
142 throw new RangeError(
143 'Cannot instantiate a pool with a negative number of workers'
145 } else if (this.type === PoolTypes
.fixed
&& numberOfWorkers
=== 0) {
146 throw new Error('Cannot instantiate a fixed pool with no worker')
150 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
151 if (isPlainObject(opts
)) {
152 this.opts
.workerChoiceStrategy
=
153 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
154 this.checkValidWorkerChoiceStrategy(this.opts
.workerChoiceStrategy
)
155 this.opts
.workerChoiceStrategyOptions
=
156 opts
.workerChoiceStrategyOptions
??
157 DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
158 this.checkValidWorkerChoiceStrategyOptions(
159 this.opts
.workerChoiceStrategyOptions
161 this.opts
.restartWorkerOnError
= opts
.restartWorkerOnError
?? true
162 this.opts
.enableEvents
= opts
.enableEvents
?? true
163 this.opts
.enableTasksQueue
= opts
.enableTasksQueue
?? false
164 if (this.opts
.enableTasksQueue
) {
165 this.checkValidTasksQueueOptions(
166 opts
.tasksQueueOptions
as TasksQueueOptions
168 this.opts
.tasksQueueOptions
= this.buildTasksQueueOptions(
169 opts
.tasksQueueOptions
as TasksQueueOptions
173 throw new TypeError('Invalid pool options: must be a plain object')
177 private checkValidWorkerChoiceStrategy (
178 workerChoiceStrategy
: WorkerChoiceStrategy
180 if (!Object.values(WorkerChoiceStrategies
).includes(workerChoiceStrategy
)) {
182 `Invalid worker choice strategy '${workerChoiceStrategy}'`
187 private checkValidWorkerChoiceStrategyOptions (
188 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
190 if (!isPlainObject(workerChoiceStrategyOptions
)) {
192 'Invalid worker choice strategy options: must be a plain object'
196 workerChoiceStrategyOptions
.weights
!= null &&
197 Object.keys(workerChoiceStrategyOptions
.weights
).length
!== this.maxSize
200 'Invalid worker choice strategy options: must have a weight for each worker node'
204 workerChoiceStrategyOptions
.measurement
!= null &&
205 !Object.values(Measurements
).includes(
206 workerChoiceStrategyOptions
.measurement
210 `Invalid worker choice strategy options: invalid measurement '${workerChoiceStrategyOptions.measurement}'`
215 private checkValidTasksQueueOptions (
216 tasksQueueOptions
: TasksQueueOptions
218 if (tasksQueueOptions
!= null && !isPlainObject(tasksQueueOptions
)) {
219 throw new TypeError('Invalid tasks queue options: must be a plain object')
222 tasksQueueOptions
?.concurrency
!= null &&
223 !Number.isSafeInteger(tasksQueueOptions
.concurrency
)
226 'Invalid worker tasks concurrency: must be an integer'
230 tasksQueueOptions
?.concurrency
!= null &&
231 tasksQueueOptions
.concurrency
<= 0
234 `Invalid worker tasks concurrency '${tasksQueueOptions.concurrency}'`
239 private get
starting (): boolean {
240 return this.workerNodes
.some(workerNode
=> !workerNode
.info
.started
)
243 private get
started (): boolean {
244 return this.workerNodes
.some(workerNode
=> workerNode
.info
.started
)
248 public get
info (): PoolInfo
{
252 minSize
: this.minSize
,
253 maxSize
: this.maxSize
,
254 workerNodes
: this.workerNodes
.length
,
255 idleWorkerNodes
: this.workerNodes
.reduce(
256 (accumulator
, workerNode
) =>
257 workerNode
.usage
.tasks
.executing
=== 0
262 busyWorkerNodes
: this.workerNodes
.reduce(
263 (accumulator
, workerNode
) =>
264 workerNode
.usage
.tasks
.executing
> 0 ? accumulator
+ 1 : accumulator
,
267 executedTasks
: this.workerNodes
.reduce(
268 (accumulator
, workerNode
) =>
269 accumulator
+ workerNode
.usage
.tasks
.executed
,
272 executingTasks
: this.workerNodes
.reduce(
273 (accumulator
, workerNode
) =>
274 accumulator
+ workerNode
.usage
.tasks
.executing
,
277 queuedTasks
: this.workerNodes
.reduce(
278 (accumulator
, workerNode
) =>
279 accumulator
+ workerNode
.usage
.tasks
.queued
,
282 maxQueuedTasks
: this.workerNodes
.reduce(
283 (accumulator
, workerNode
) =>
284 accumulator
+ workerNode
.usage
.tasks
.maxQueued
,
287 failedTasks
: this.workerNodes
.reduce(
288 (accumulator
, workerNode
) =>
289 accumulator
+ workerNode
.usage
.tasks
.failed
,
298 * If it is `'dynamic'`, it provides the `max` property.
300 protected abstract get
type (): PoolType
303 * Gets the worker type.
305 protected abstract get
worker (): WorkerType
310 protected abstract get
minSize (): number
315 protected abstract get
maxSize (): number
318 * Get the worker given its id.
320 * @param workerId - The worker id.
321 * @returns The worker if found in the pool worker nodes, `undefined` otherwise.
323 private getWorkerById (workerId
: number): Worker
| undefined {
324 return this.workerNodes
.find(workerNode
=> workerNode
.info
.id
=== workerId
)
329 * Gets the given worker its worker node key.
331 * @param worker - The worker.
332 * @returns The worker node key if found in the pool worker nodes, `-1` otherwise.
334 private getWorkerNodeKey (worker
: Worker
): number {
335 return this.workerNodes
.findIndex(
336 workerNode
=> workerNode
.worker
=== worker
341 public setWorkerChoiceStrategy (
342 workerChoiceStrategy
: WorkerChoiceStrategy
,
343 workerChoiceStrategyOptions
?: WorkerChoiceStrategyOptions
345 this.checkValidWorkerChoiceStrategy(workerChoiceStrategy
)
346 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
347 this.workerChoiceStrategyContext
.setWorkerChoiceStrategy(
348 this.opts
.workerChoiceStrategy
350 if (workerChoiceStrategyOptions
!= null) {
351 this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
353 for (const [workerNodeKey
, workerNode
] of this.workerNodes
.entries()) {
354 this.setWorkerNodeTasksUsage(
356 this.getWorkerUsage(workerNodeKey
)
358 this.setWorkerStatistics(workerNode
.worker
)
363 public setWorkerChoiceStrategyOptions (
364 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
366 this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
367 this.opts
.workerChoiceStrategyOptions
= workerChoiceStrategyOptions
368 this.workerChoiceStrategyContext
.setOptions(
369 this.opts
.workerChoiceStrategyOptions
374 public enableTasksQueue (
376 tasksQueueOptions
?: TasksQueueOptions
378 if (this.opts
.enableTasksQueue
=== true && !enable
) {
379 this.flushTasksQueues()
381 this.opts
.enableTasksQueue
= enable
382 this.setTasksQueueOptions(tasksQueueOptions
as TasksQueueOptions
)
386 public setTasksQueueOptions (tasksQueueOptions
: TasksQueueOptions
): void {
387 if (this.opts
.enableTasksQueue
=== true) {
388 this.checkValidTasksQueueOptions(tasksQueueOptions
)
389 this.opts
.tasksQueueOptions
=
390 this.buildTasksQueueOptions(tasksQueueOptions
)
391 } else if (this.opts
.tasksQueueOptions
!= null) {
392 delete this.opts
.tasksQueueOptions
396 private buildTasksQueueOptions (
397 tasksQueueOptions
: TasksQueueOptions
398 ): TasksQueueOptions
{
400 concurrency
: tasksQueueOptions
?.concurrency
?? 1
405 * Whether the pool is full or not.
407 * The pool filling boolean status.
409 protected get
full (): boolean {
410 return this.workerNodes
.length
>= this.maxSize
414 * Whether the pool is busy or not.
416 * The pool busyness boolean status.
418 protected abstract get
busy (): boolean
421 * Whether worker nodes are executing at least one task.
423 * @returns Worker nodes busyness boolean status.
425 protected internalBusy (): boolean {
427 this.workerNodes
.findIndex(workerNode
=> {
428 return workerNode
.usage
.tasks
.executing
=== 0
434 public async execute (data
?: Data
, name
?: string): Promise
<Response
> {
435 const timestamp
= performance
.now()
436 const workerNodeKey
= this.chooseWorkerNode()
437 const submittedTask
: Task
<Data
> = {
439 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
440 data
: data
?? ({} as Data
),
442 id
: crypto
.randomUUID()
444 const res
= new Promise
<Response
>((resolve
, reject
) => {
445 this.promiseResponseMap
.set(submittedTask
.id
as string, {
448 worker
: this.workerNodes
[workerNodeKey
].worker
452 this.opts
.enableTasksQueue
=== true &&
454 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
>=
455 ((this.opts
.tasksQueueOptions
as TasksQueueOptions
)
456 .concurrency
as number))
458 this.enqueueTask(workerNodeKey
, submittedTask
)
460 this.executeTask(workerNodeKey
, submittedTask
)
462 this.checkAndEmitEvents()
463 // eslint-disable-next-line @typescript-eslint/return-await
468 public async destroy (): Promise
<void> {
470 this.workerNodes
.map(async (workerNode
, workerNodeKey
) => {
471 this.flushTasksQueue(workerNodeKey
)
472 // FIXME: wait for tasks to be finished
473 await this.destroyWorker(workerNode
.worker
)
479 * Terminates the given worker.
481 * @param worker - A worker within `workerNodes`.
483 protected abstract destroyWorker (worker
: Worker
): void | Promise
<void>
486 * Setup hook to execute code before worker nodes are created in the abstract constructor.
491 protected setupHook (): void {
492 // Intentionally empty
496 * Should return whether the worker is the main worker or not.
498 protected abstract isMain (): boolean
501 * Hook executed before the worker task execution.
504 * @param workerNodeKey - The worker node key.
505 * @param task - The task to execute.
507 protected beforeTaskExecutionHook (
508 workerNodeKey
: number,
511 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
512 ++workerUsage
.tasks
.executing
513 this.updateWaitTimeWorkerUsage(workerUsage
, task
)
517 * Hook executed after the worker task execution.
520 * @param worker - The worker.
521 * @param message - The received message.
523 protected afterTaskExecutionHook (
525 message
: MessageValue
<Response
>
527 const workerUsage
= this.workerNodes
[this.getWorkerNodeKey(worker
)].usage
528 this.updateTaskStatisticsWorkerUsage(workerUsage
, message
)
529 this.updateRunTimeWorkerUsage(workerUsage
, message
)
530 this.updateEluWorkerUsage(workerUsage
, message
)
533 private updateTaskStatisticsWorkerUsage (
534 workerUsage
: WorkerUsage
,
535 message
: MessageValue
<Response
>
537 const workerTaskStatistics
= workerUsage
.tasks
538 --workerTaskStatistics
.executing
539 ++workerTaskStatistics
.executed
540 if (message
.taskError
!= null) {
541 ++workerTaskStatistics
.failed
545 private updateRunTimeWorkerUsage (
546 workerUsage
: WorkerUsage
,
547 message
: MessageValue
<Response
>
550 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
553 workerUsage
.runTime
.aggregate
+= message
.taskPerformance
?.runTime
?? 0
555 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
557 workerUsage
.tasks
.executed
!== 0
559 workerUsage
.runTime
.average
=
560 workerUsage
.runTime
.aggregate
/
561 (workerUsage
.tasks
.executed
- workerUsage
.tasks
.failed
)
564 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
566 message
.taskPerformance
?.runTime
!= null
568 workerUsage
.runTime
.history
.push(message
.taskPerformance
.runTime
)
569 workerUsage
.runTime
.median
= median(workerUsage
.runTime
.history
)
574 private updateWaitTimeWorkerUsage (
575 workerUsage
: WorkerUsage
,
578 const timestamp
= performance
.now()
579 const taskWaitTime
= timestamp
- (task
.timestamp
?? timestamp
)
581 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().waitTime
584 workerUsage
.waitTime
.aggregate
+= taskWaitTime
?? 0
586 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
588 workerUsage
.tasks
.executed
!== 0
590 workerUsage
.waitTime
.average
=
591 workerUsage
.waitTime
.aggregate
/
592 (workerUsage
.tasks
.executed
- workerUsage
.tasks
.failed
)
595 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
599 workerUsage
.waitTime
.history
.push(taskWaitTime
)
600 workerUsage
.waitTime
.median
= median(workerUsage
.waitTime
.history
)
605 private updateEluWorkerUsage (
606 workerUsage
: WorkerUsage
,
607 message
: MessageValue
<Response
>
610 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
613 if (workerUsage
.elu
!= null && message
.taskPerformance
?.elu
!= null) {
614 workerUsage
.elu
.idle
.aggregate
+= message
.taskPerformance
.elu
.idle
615 workerUsage
.elu
.active
.aggregate
+= message
.taskPerformance
.elu
.active
616 workerUsage
.elu
.utilization
=
617 (workerUsage
.elu
.utilization
+
618 message
.taskPerformance
.elu
.utilization
) /
620 } else if (message
.taskPerformance
?.elu
!= null) {
621 workerUsage
.elu
.idle
.aggregate
= message
.taskPerformance
.elu
.idle
622 workerUsage
.elu
.active
.aggregate
= message
.taskPerformance
.elu
.active
623 workerUsage
.elu
.utilization
= message
.taskPerformance
.elu
.utilization
626 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
628 workerUsage
.tasks
.executed
!== 0
630 const executedTasks
=
631 workerUsage
.tasks
.executed
- workerUsage
.tasks
.failed
632 workerUsage
.elu
.idle
.average
=
633 workerUsage
.elu
.idle
.aggregate
/ executedTasks
634 workerUsage
.elu
.active
.average
=
635 workerUsage
.elu
.active
.aggregate
/ executedTasks
638 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
640 message
.taskPerformance
?.elu
!= null
642 workerUsage
.elu
.idle
.history
.push(message
.taskPerformance
.elu
.idle
)
643 workerUsage
.elu
.active
.history
.push(message
.taskPerformance
.elu
.active
)
644 workerUsage
.elu
.idle
.median
= median(workerUsage
.elu
.idle
.history
)
645 workerUsage
.elu
.active
.median
= median(workerUsage
.elu
.active
.history
)
651 * Chooses a worker node for the next task.
653 * The default worker choice strategy uses a round robin algorithm to distribute the tasks.
655 * @returns The worker node key
657 private chooseWorkerNode (): number {
658 if (this.shallCreateDynamicWorker()) {
659 const worker
= this.createAndSetupDynamicWorker()
661 this.workerChoiceStrategyContext
.getStrategyPolicy().useDynamicWorker
663 return this.getWorkerNodeKey(worker
)
666 return this.workerChoiceStrategyContext
.execute()
670 * Conditions for dynamic worker creation.
672 * @returns Whether to create a dynamic worker or not.
674 private shallCreateDynamicWorker (): boolean {
675 return this.type === PoolTypes
.dynamic
&& !this.full
&& this.internalBusy()
679 * Sends a message to the given worker.
681 * @param worker - The worker which should receive the message.
682 * @param message - The message.
684 protected abstract sendToWorker (
686 message
: MessageValue
<Data
>
690 * Registers a listener callback on the given worker.
692 * @param worker - The worker which should register a listener.
693 * @param listener - The message listener callback.
695 private registerWorkerMessageListener
<Message
extends Data
| Response
>(
697 listener
: (message
: MessageValue
<Message
>) => void
699 worker
.on('message', listener
as MessageHandler
<Worker
>)
703 * Creates a new worker.
705 * @returns Newly created worker.
707 protected abstract createWorker (): Worker
710 * Function that can be hooked up when a worker has been newly created and moved to the pool worker nodes.
713 * @param worker - The newly created worker.
715 protected afterWorkerSetup (worker
: Worker
): void {
716 // Listen to worker messages.
717 this.registerWorkerMessageListener(worker
, this.workerListener())
721 * Creates a new worker and sets it up completely in the pool worker nodes.
723 * @returns New, completely set up worker.
725 protected createAndSetupWorker (): Worker
{
726 const worker
= this.createWorker()
728 worker
.on('message', this.opts
.messageHandler
?? EMPTY_FUNCTION
)
729 worker
.on('error', this.opts
.errorHandler
?? EMPTY_FUNCTION
)
730 worker
.on('error', error
=> {
731 if (this.emitter
!= null) {
732 this.emitter
.emit(PoolEvents
.error
, error
)
734 if (this.opts
.restartWorkerOnError
=== true && !this.starting
) {
735 this.createAndSetupWorker()
738 worker
.on('online', this.opts
.onlineHandler
?? EMPTY_FUNCTION
)
739 worker
.on('exit', this.opts
.exitHandler
?? EMPTY_FUNCTION
)
740 worker
.once('exit', () => {
741 this.removeWorkerNode(worker
)
744 this.pushWorkerNode(worker
)
746 this.setWorkerStatistics(worker
)
748 this.afterWorkerSetup(worker
)
754 * Creates a new dynamic worker and sets it up completely in the pool worker nodes.
756 * @returns New, completely set up dynamic worker.
758 protected createAndSetupDynamicWorker (): Worker
{
759 const worker
= this.createAndSetupWorker()
760 this.registerWorkerMessageListener(worker
, message
=> {
761 const workerNodeKey
= this.getWorkerNodeKey(worker
)
763 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
764 (message
.kill
!= null &&
765 ((this.opts
.enableTasksQueue
=== false &&
766 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
=== 0) ||
767 (this.opts
.enableTasksQueue
=== true &&
768 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
=== 0 &&
769 this.tasksQueueSize(workerNodeKey
) === 0)))
771 // Kill message received from the worker: no new tasks are submitted to that worker for a while ( > maxInactiveTime)
772 void (this.destroyWorker(worker
) as Promise
<void>)
779 * This function is the listener registered for each worker message.
781 * @returns The listener function to execute when a message is received from a worker.
783 protected workerListener (): (message
: MessageValue
<Response
>) => void {
785 if (message
.workerId
!= null && message
.started
!= null) {
786 // Worker started message received
788 this.getWorkerNodeKey(this.getWorkerById(message
.workerId
) as Worker
)
789 ].info
.started
= message
.started
790 } else if (message
.id
!= null) {
791 // Task execution response received
792 const promiseResponse
= this.promiseResponseMap
.get(message
.id
)
793 if (promiseResponse
!= null) {
794 if (message
.taskError
!= null) {
795 if (this.emitter
!= null) {
796 this.emitter
.emit(PoolEvents
.taskError
, message
.taskError
)
798 promiseResponse
.reject(message
.taskError
.message
)
800 promiseResponse
.resolve(message
.data
as Response
)
802 this.afterTaskExecutionHook(promiseResponse
.worker
, message
)
803 this.promiseResponseMap
.delete(message
.id
)
804 const workerNodeKey
= this.getWorkerNodeKey(promiseResponse
.worker
)
806 this.opts
.enableTasksQueue
=== true &&
807 this.tasksQueueSize(workerNodeKey
) > 0
811 this.dequeueTask(workerNodeKey
) as Task
<Data
>
814 this.workerChoiceStrategyContext
.update(workerNodeKey
)
820 private checkAndEmitEvents (): void {
821 if (this.emitter
!= null) {
823 this.emitter
?.emit(PoolEvents
.busy
, this.info
)
825 if (this.type === PoolTypes
.dynamic
&& this.full
) {
826 this.emitter
?.emit(PoolEvents
.full
, this.info
)
832 * Sets the given worker node its tasks usage in the pool.
834 * @param workerNode - The worker node.
835 * @param workerUsage - The worker usage.
837 private setWorkerNodeTasksUsage (
838 workerNode
: WorkerNode
<Worker
, Data
>,
839 workerUsage
: WorkerUsage
841 workerNode
.usage
= workerUsage
845 * Pushes the given worker in the pool worker nodes.
847 * @param worker - The worker.
848 * @returns The worker nodes length.
850 private pushWorkerNode (worker
: Worker
): number {
851 this.workerNodes
.push({
853 info
: { id
: worker
.threadId
?? worker
.id
, started
: false },
854 usage
: this.getWorkerUsage(),
855 tasksQueue
: new Queue
<Task
<Data
>>()
857 const workerNodeKey
= this.getWorkerNodeKey(worker
)
858 this.setWorkerNodeTasksUsage(
859 this.workerNodes
[workerNodeKey
],
860 this.getWorkerUsage(workerNodeKey
)
862 return this.workerNodes
.length
866 // * Sets the given worker in the pool worker nodes.
868 // * @param workerNodeKey - The worker node key.
869 // * @param worker - The worker.
870 // * @param workerInfo - The worker info.
871 // * @param workerUsage - The worker usage.
872 // * @param tasksQueue - The worker task queue.
874 // private setWorkerNode (
875 // workerNodeKey: number,
877 // workerInfo: WorkerInfo,
878 // workerUsage: WorkerUsage,
879 // tasksQueue: Queue<Task<Data>>
881 // this.workerNodes[workerNodeKey] = {
884 // usage: workerUsage,
890 * Removes the given worker from the pool worker nodes.
892 * @param worker - The worker.
894 private removeWorkerNode (worker
: Worker
): void {
895 const workerNodeKey
= this.getWorkerNodeKey(worker
)
896 if (workerNodeKey
!== -1) {
897 this.workerNodes
.splice(workerNodeKey
, 1)
898 this.workerChoiceStrategyContext
.remove(workerNodeKey
)
902 private executeTask (workerNodeKey
: number, task
: Task
<Data
>): void {
903 this.beforeTaskExecutionHook(workerNodeKey
, task
)
904 this.sendToWorker(this.workerNodes
[workerNodeKey
].worker
, task
)
907 private enqueueTask (workerNodeKey
: number, task
: Task
<Data
>): number {
908 return this.workerNodes
[workerNodeKey
].tasksQueue
.enqueue(task
)
911 private dequeueTask (workerNodeKey
: number): Task
<Data
> | undefined {
912 return this.workerNodes
[workerNodeKey
].tasksQueue
.dequeue()
915 private tasksQueueSize (workerNodeKey
: number): number {
916 return this.workerNodes
[workerNodeKey
].tasksQueue
.size
919 private tasksMaxQueueSize (workerNodeKey
: number): number {
920 return this.workerNodes
[workerNodeKey
].tasksQueue
.maxSize
923 private flushTasksQueue (workerNodeKey
: number): void {
924 if (this.tasksQueueSize(workerNodeKey
) > 0) {
925 for (let i
= 0; i
< this.tasksQueueSize(workerNodeKey
); i
++) {
928 this.dequeueTask(workerNodeKey
) as Task
<Data
>
932 this.workerNodes
[workerNodeKey
].tasksQueue
.clear()
935 private flushTasksQueues (): void {
936 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
937 this.flushTasksQueue(workerNodeKey
)
941 private setWorkerStatistics (worker
: Worker
): void {
942 this.sendToWorker(worker
, {
945 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
947 elu
: this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
953 private getWorkerUsage (workerNodeKey
?: number): WorkerUsage
{
954 const getTasksQueueSize
= (workerNodeKey
?: number): number => {
955 return workerNodeKey
!= null ? this.tasksQueueSize(workerNodeKey
) : 0
957 const getTasksMaxQueueSize
= (workerNodeKey
?: number): number => {
958 return workerNodeKey
!= null ? this.tasksMaxQueueSize(workerNodeKey
) : 0
964 get
queued (): number {
965 return getTasksQueueSize(workerNodeKey
)
967 get
maxQueued (): number {
968 return getTasksMaxQueueSize(workerNodeKey
)
976 history
: new CircularArray()
982 history
: new CircularArray()
989 history
: new CircularArray()
995 history
: new CircularArray()