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
,
32 WorkerChoiceStrategies
,
33 type WorkerChoiceStrategy
,
34 type WorkerChoiceStrategyOptions
35 } from
'./selection-strategies/selection-strategies-types'
36 import { WorkerChoiceStrategyContext
} from
'./selection-strategies/worker-choice-strategy-context'
39 * Base class that implements some shared logic for all poolifier pools.
41 * @typeParam Worker - Type of worker which manages this pool.
42 * @typeParam Data - Type of data sent to the worker. This can only be serializable data.
43 * @typeParam Response - Type of execution response. This can only be serializable data.
45 export abstract class AbstractPool
<
46 Worker
extends IWorker
,
49 > implements IPool
<Worker
, Data
, Response
> {
51 public readonly workerNodes
: Array<WorkerNode
<Worker
, Data
>> = []
54 public readonly emitter
?: PoolEmitter
57 * The execution response promise map.
59 * - `key`: The message id of each submitted task.
60 * - `value`: An object that contains the worker, the execution response promise resolve and reject callbacks.
62 * When we receive a message from the worker, we get a map entry with the promise resolve/reject bound to the message id.
64 protected promiseResponseMap
: Map
<
66 PromiseResponseWrapper
<Worker
, Response
>
67 > = new Map
<string, PromiseResponseWrapper
<Worker
, Response
>>()
70 * Worker choice strategy context referencing a worker choice algorithm implementation.
72 protected workerChoiceStrategyContext
: WorkerChoiceStrategyContext
<
79 * Constructs a new poolifier pool.
81 * @param numberOfWorkers - Number of workers that this pool should manage.
82 * @param filePath - Path to the worker file.
83 * @param opts - Options for the pool.
86 protected readonly numberOfWorkers
: number,
87 protected readonly filePath
: string,
88 protected readonly opts
: PoolOptions
<Worker
>
91 throw new Error('Cannot start a pool from a worker!')
93 this.checkNumberOfWorkers(this.numberOfWorkers
)
94 this.checkFilePath(this.filePath
)
95 this.checkPoolOptions(this.opts
)
97 this.chooseWorkerNode
= this.chooseWorkerNode
.bind(this)
98 this.executeTask
= this.executeTask
.bind(this)
99 this.enqueueTask
= this.enqueueTask
.bind(this)
100 this.checkAndEmitEvents
= this.checkAndEmitEvents
.bind(this)
102 if (this.opts
.enableEvents
=== true) {
103 this.emitter
= new PoolEmitter()
105 this.workerChoiceStrategyContext
= new WorkerChoiceStrategyContext
<
111 this.opts
.workerChoiceStrategy
,
112 this.opts
.workerChoiceStrategyOptions
117 for (let i
= 1; i
<= this.numberOfWorkers
; i
++) {
118 this.createAndSetupWorker()
122 private checkFilePath (filePath
: string): void {
125 (typeof filePath
=== 'string' && filePath
.trim().length
=== 0)
127 throw new Error('Please specify a file with a worker implementation')
131 private checkNumberOfWorkers (numberOfWorkers
: number): void {
132 if (numberOfWorkers
== null) {
134 'Cannot instantiate a pool without specifying the number of workers'
136 } else if (!Number.isSafeInteger(numberOfWorkers
)) {
138 'Cannot instantiate a pool with a non safe integer number of workers'
140 } else if (numberOfWorkers
< 0) {
141 throw new RangeError(
142 'Cannot instantiate a pool with a negative number of workers'
144 } else if (this.type === PoolTypes
.fixed
&& numberOfWorkers
=== 0) {
145 throw new Error('Cannot instantiate a fixed pool with no worker')
149 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
150 if (isPlainObject(opts
)) {
151 this.opts
.workerChoiceStrategy
=
152 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
153 this.checkValidWorkerChoiceStrategy(this.opts
.workerChoiceStrategy
)
154 this.opts
.workerChoiceStrategyOptions
=
155 opts
.workerChoiceStrategyOptions
??
156 DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
157 this.checkValidWorkerChoiceStrategyOptions(
158 this.opts
.workerChoiceStrategyOptions
160 this.opts
.restartWorkerOnError
= opts
.restartWorkerOnError
?? true
161 this.opts
.enableEvents
= opts
.enableEvents
?? true
162 this.opts
.enableTasksQueue
= opts
.enableTasksQueue
?? false
163 if (this.opts
.enableTasksQueue
) {
164 this.checkValidTasksQueueOptions(
165 opts
.tasksQueueOptions
as TasksQueueOptions
167 this.opts
.tasksQueueOptions
= this.buildTasksQueueOptions(
168 opts
.tasksQueueOptions
as TasksQueueOptions
172 throw new TypeError('Invalid pool options: must be a plain object')
176 private checkValidWorkerChoiceStrategy (
177 workerChoiceStrategy
: WorkerChoiceStrategy
179 if (!Object.values(WorkerChoiceStrategies
).includes(workerChoiceStrategy
)) {
181 `Invalid worker choice strategy '${workerChoiceStrategy}'`
186 private checkValidWorkerChoiceStrategyOptions (
187 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
189 if (!isPlainObject(workerChoiceStrategyOptions
)) {
191 'Invalid worker choice strategy options: must be a plain object'
195 workerChoiceStrategyOptions
.weights
!= null &&
196 Object.keys(workerChoiceStrategyOptions
.weights
).length
!== this.maxSize
199 'Invalid worker choice strategy options: must have a weight for each worker node'
204 private checkValidTasksQueueOptions (
205 tasksQueueOptions
: TasksQueueOptions
207 if (tasksQueueOptions
!= null && !isPlainObject(tasksQueueOptions
)) {
208 throw new TypeError('Invalid tasks queue options: must be a plain object')
210 if ((tasksQueueOptions
?.concurrency
as number) <= 0) {
212 `Invalid worker tasks concurrency '${
213 tasksQueueOptions.concurrency as number
220 public get
info (): PoolInfo
{
224 minSize
: this.minSize
,
225 maxSize
: this.maxSize
,
226 workerNodes
: this.workerNodes
.length
,
227 idleWorkerNodes
: this.workerNodes
.reduce(
228 (accumulator
, workerNode
) =>
229 workerNode
.workerUsage
.tasks
.executing
=== 0
234 busyWorkerNodes
: this.workerNodes
.reduce(
235 (accumulator
, workerNode
) =>
236 workerNode
.workerUsage
.tasks
.executing
> 0
241 executedTasks
: this.workerNodes
.reduce(
242 (accumulator
, workerNode
) =>
243 accumulator
+ workerNode
.workerUsage
.tasks
.executed
,
246 executingTasks
: this.workerNodes
.reduce(
247 (accumulator
, workerNode
) =>
248 accumulator
+ workerNode
.workerUsage
.tasks
.executing
,
251 queuedTasks
: this.workerNodes
.reduce(
252 (accumulator
, workerNode
) => accumulator
+ workerNode
.tasksQueue
.size
,
255 maxQueuedTasks
: this.workerNodes
.reduce(
256 (accumulator
, workerNode
) =>
257 accumulator
+ workerNode
.tasksQueue
.maxSize
,
260 failedTasks
: this.workerNodes
.reduce(
261 (accumulator
, workerNode
) =>
262 accumulator
+ workerNode
.workerUsage
.tasks
.failed
,
271 * If it is `'dynamic'`, it provides the `max` property.
273 protected abstract get
type (): PoolType
276 * Gets the worker type.
278 protected abstract get
worker (): WorkerType
283 protected abstract get
minSize (): number
288 protected abstract get
maxSize (): number
291 * Gets the given worker its worker node key.
293 * @param worker - The worker.
294 * @returns The worker node key if the worker is found in the pool worker nodes, `-1` otherwise.
296 private getWorkerNodeKey (worker
: Worker
): number {
297 return this.workerNodes
.findIndex(
298 workerNode
=> workerNode
.worker
=== worker
303 public setWorkerChoiceStrategy (
304 workerChoiceStrategy
: WorkerChoiceStrategy
,
305 workerChoiceStrategyOptions
?: WorkerChoiceStrategyOptions
307 this.checkValidWorkerChoiceStrategy(workerChoiceStrategy
)
308 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
309 this.workerChoiceStrategyContext
.setWorkerChoiceStrategy(
310 this.opts
.workerChoiceStrategy
312 if (workerChoiceStrategyOptions
!= null) {
313 this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
315 for (const workerNode
of this.workerNodes
) {
316 this.setWorkerNodeTasksUsage(
318 this.getWorkerUsage(workerNode
.worker
)
320 this.setWorkerStatistics(workerNode
.worker
)
325 public setWorkerChoiceStrategyOptions (
326 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
328 this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
329 this.opts
.workerChoiceStrategyOptions
= workerChoiceStrategyOptions
330 this.workerChoiceStrategyContext
.setOptions(
331 this.opts
.workerChoiceStrategyOptions
336 public enableTasksQueue (
338 tasksQueueOptions
?: TasksQueueOptions
340 if (this.opts
.enableTasksQueue
=== true && !enable
) {
341 this.flushTasksQueues()
343 this.opts
.enableTasksQueue
= enable
344 this.setTasksQueueOptions(tasksQueueOptions
as TasksQueueOptions
)
348 public setTasksQueueOptions (tasksQueueOptions
: TasksQueueOptions
): void {
349 if (this.opts
.enableTasksQueue
=== true) {
350 this.checkValidTasksQueueOptions(tasksQueueOptions
)
351 this.opts
.tasksQueueOptions
=
352 this.buildTasksQueueOptions(tasksQueueOptions
)
353 } else if (this.opts
.tasksQueueOptions
!= null) {
354 delete this.opts
.tasksQueueOptions
358 private buildTasksQueueOptions (
359 tasksQueueOptions
: TasksQueueOptions
360 ): TasksQueueOptions
{
362 concurrency
: tasksQueueOptions
?.concurrency
?? 1
367 * Whether the pool is full or not.
369 * The pool filling boolean status.
371 protected get
full (): boolean {
372 return this.workerNodes
.length
>= this.maxSize
376 * Whether the pool is busy or not.
378 * The pool busyness boolean status.
380 protected abstract get
busy (): boolean
383 * Whether worker nodes are executing at least one task.
385 * @returns Worker nodes busyness boolean status.
387 protected internalBusy (): boolean {
389 this.workerNodes
.findIndex(workerNode
=> {
390 return workerNode
.workerUsage
.tasks
.executing
=== 0
396 public async execute (data
?: Data
, name
?: string): Promise
<Response
> {
397 const timestamp
= performance
.now()
398 const workerNodeKey
= this.chooseWorkerNode()
399 const submittedTask
: Task
<Data
> = {
401 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
402 data
: data
?? ({} as Data
),
404 id
: crypto
.randomUUID()
406 const res
= new Promise
<Response
>((resolve
, reject
) => {
407 this.promiseResponseMap
.set(submittedTask
.id
as string, {
410 worker
: this.workerNodes
[workerNodeKey
].worker
414 this.opts
.enableTasksQueue
=== true &&
416 this.workerNodes
[workerNodeKey
].workerUsage
.tasks
.executing
>=
417 ((this.opts
.tasksQueueOptions
as TasksQueueOptions
)
418 .concurrency
as number))
420 this.enqueueTask(workerNodeKey
, submittedTask
)
422 this.executeTask(workerNodeKey
, submittedTask
)
424 this.checkAndEmitEvents()
425 // eslint-disable-next-line @typescript-eslint/return-await
430 public async destroy (): Promise
<void> {
432 this.workerNodes
.map(async (workerNode
, workerNodeKey
) => {
433 this.flushTasksQueue(workerNodeKey
)
434 // FIXME: wait for tasks to be finished
435 await this.destroyWorker(workerNode
.worker
)
441 * Terminates the given worker.
443 * @param worker - A worker within `workerNodes`.
445 protected abstract destroyWorker (worker
: Worker
): void | Promise
<void>
448 * Setup hook to execute code before worker node are created in the abstract constructor.
453 protected setupHook (): void {
454 // Intentionally empty
458 * Should return whether the worker is the main worker or not.
460 protected abstract isMain (): boolean
463 * Hook executed before the worker task execution.
466 * @param workerNodeKey - The worker node key.
467 * @param task - The task to execute.
469 protected beforeTaskExecutionHook (
470 workerNodeKey
: number,
473 const workerUsage
= this.workerNodes
[workerNodeKey
].workerUsage
474 ++workerUsage
.tasks
.executing
475 this.updateWaitTimeWorkerUsage(workerUsage
, task
)
479 * Hook executed after the worker task execution.
482 * @param worker - The worker.
483 * @param message - The received message.
485 protected afterTaskExecutionHook (
487 message
: MessageValue
<Response
>
490 this.workerNodes
[this.getWorkerNodeKey(worker
)].workerUsage
491 this.updateTaskStatisticsWorkerUsage(workerUsage
, message
)
492 this.updateRunTimeWorkerUsage(workerUsage
, message
)
493 this.updateEluWorkerUsage(workerUsage
, message
)
496 private updateTaskStatisticsWorkerUsage (
497 workerUsage
: WorkerUsage
,
498 message
: MessageValue
<Response
>
500 const workerTaskStatistics
= workerUsage
.tasks
501 --workerTaskStatistics
.executing
502 ++workerTaskStatistics
.executed
503 if (message
.taskError
!= null) {
504 ++workerTaskStatistics
.failed
508 private updateRunTimeWorkerUsage (
509 workerUsage
: WorkerUsage
,
510 message
: MessageValue
<Response
>
513 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
516 workerUsage
.runTime
.aggregate
+= message
.taskPerformance
?.runTime
?? 0
518 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
520 workerUsage
.tasks
.executed
!== 0
522 workerUsage
.runTime
.average
=
523 workerUsage
.runTime
.aggregate
/
524 (workerUsage
.tasks
.executed
- workerUsage
.tasks
.failed
)
527 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
529 message
.taskPerformance
?.runTime
!= null
531 workerUsage
.runTime
.history
.push(message
.taskPerformance
.runTime
)
532 workerUsage
.runTime
.median
= median(workerUsage
.runTime
.history
)
537 private updateWaitTimeWorkerUsage (
538 workerUsage
: WorkerUsage
,
541 const timestamp
= performance
.now()
542 const taskWaitTime
= timestamp
- (task
.timestamp
?? timestamp
)
544 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().waitTime
547 workerUsage
.waitTime
.aggregate
+= taskWaitTime
?? 0
549 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
551 workerUsage
.tasks
.executed
!== 0
553 workerUsage
.waitTime
.average
=
554 workerUsage
.waitTime
.aggregate
/
555 (workerUsage
.tasks
.executed
- workerUsage
.tasks
.failed
)
558 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
562 workerUsage
.waitTime
.history
.push(taskWaitTime
)
563 workerUsage
.waitTime
.median
= median(workerUsage
.waitTime
.history
)
568 private updateEluWorkerUsage (
569 workerUsage
: WorkerUsage
,
570 message
: MessageValue
<Response
>
573 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
576 if (workerUsage
.elu
!= null && message
.taskPerformance
?.elu
!= null) {
577 workerUsage
.elu
.idle
.aggregate
+= message
.taskPerformance
.elu
.idle
578 workerUsage
.elu
.active
.aggregate
+= message
.taskPerformance
.elu
.active
579 workerUsage
.elu
.utilization
=
580 (workerUsage
.elu
.utilization
+
581 message
.taskPerformance
.elu
.utilization
) /
583 } else if (message
.taskPerformance
?.elu
!= null) {
584 workerUsage
.elu
.idle
.aggregate
= message
.taskPerformance
.elu
.idle
585 workerUsage
.elu
.active
.aggregate
= message
.taskPerformance
.elu
.active
586 workerUsage
.elu
.utilization
= message
.taskPerformance
.elu
.utilization
589 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
591 workerUsage
.tasks
.executed
!== 0
593 const executedTasks
=
594 workerUsage
.tasks
.executed
- workerUsage
.tasks
.failed
595 workerUsage
.elu
.idle
.average
=
596 workerUsage
.elu
.idle
.aggregate
/ executedTasks
597 workerUsage
.elu
.active
.average
=
598 workerUsage
.elu
.active
.aggregate
/ executedTasks
601 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
603 message
.taskPerformance
?.elu
!= null
605 workerUsage
.elu
.idle
.history
.push(message
.taskPerformance
.elu
.idle
)
606 workerUsage
.elu
.active
.history
.push(message
.taskPerformance
.elu
.active
)
607 workerUsage
.elu
.idle
.median
= median(workerUsage
.elu
.idle
.history
)
608 workerUsage
.elu
.active
.median
= median(workerUsage
.elu
.active
.history
)
614 * Chooses a worker node for the next task.
616 * The default worker choice strategy uses a round robin algorithm to distribute the tasks.
618 * @returns The worker node key
620 private chooseWorkerNode (): number {
621 if (this.shallCreateDynamicWorker()) {
622 const worker
= this.createAndSetupDynamicWorker()
624 this.workerChoiceStrategyContext
.getStrategyPolicy().useDynamicWorker
626 return this.getWorkerNodeKey(worker
)
629 return this.workerChoiceStrategyContext
.execute()
633 * Conditions for dynamic worker creation.
635 * @returns Whether to create a dynamic worker or not.
637 private shallCreateDynamicWorker (): boolean {
638 return this.type === PoolTypes
.dynamic
&& !this.full
&& this.internalBusy()
642 * Sends a message to the given worker.
644 * @param worker - The worker which should receive the message.
645 * @param message - The message.
647 protected abstract sendToWorker (
649 message
: MessageValue
<Data
>
653 * Registers a listener callback on the given worker.
655 * @param worker - The worker which should register a listener.
656 * @param listener - The message listener callback.
658 protected abstract registerWorkerMessageListener
<
659 Message
extends Data
| Response
660 >(worker
: Worker
, listener
: (message
: MessageValue
<Message
>) => void): void
663 * Creates a new worker.
665 * @returns Newly created worker.
667 protected abstract createWorker (): Worker
670 * Function that can be hooked up when a worker has been newly created and moved to the pool worker nodes.
672 * Can be used to update the `maxListeners` or binding the `main-worker`\<-\>`worker` connection if not bind by default.
674 * @param worker - The newly created worker.
676 protected abstract afterWorkerSetup (worker
: Worker
): void
679 * Creates a new worker and sets it up completely in the pool worker nodes.
681 * @returns New, completely set up worker.
683 protected createAndSetupWorker (): Worker
{
684 const worker
= this.createWorker()
686 worker
.on('message', this.opts
.messageHandler
?? EMPTY_FUNCTION
)
687 worker
.on('error', this.opts
.errorHandler
?? EMPTY_FUNCTION
)
688 worker
.on('error', error
=> {
689 if (this.emitter
!= null) {
690 this.emitter
.emit(PoolEvents
.error
, error
)
692 if (this.opts
.restartWorkerOnError
=== true) {
693 this.createAndSetupWorker()
696 worker
.on('online', this.opts
.onlineHandler
?? EMPTY_FUNCTION
)
697 worker
.on('exit', this.opts
.exitHandler
?? EMPTY_FUNCTION
)
698 worker
.once('exit', () => {
699 this.removeWorkerNode(worker
)
702 this.pushWorkerNode(worker
)
704 this.setWorkerStatistics(worker
)
706 this.afterWorkerSetup(worker
)
712 * Creates a new dynamic worker and sets it up completely in the pool worker nodes.
714 * @returns New, completely set up dynamic worker.
716 protected createAndSetupDynamicWorker (): Worker
{
717 const worker
= this.createAndSetupWorker()
718 this.registerWorkerMessageListener(worker
, message
=> {
719 const workerNodeKey
= this.getWorkerNodeKey(worker
)
721 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
722 (message
.kill
!= null &&
723 ((this.opts
.enableTasksQueue
=== false &&
724 this.workerNodes
[workerNodeKey
].workerUsage
.tasks
.executing
===
726 (this.opts
.enableTasksQueue
=== true &&
727 this.workerNodes
[workerNodeKey
].workerUsage
.tasks
.executing
===
729 this.tasksQueueSize(workerNodeKey
) === 0)))
731 // Kill message received from the worker: no new tasks are submitted to that worker for a while ( > maxInactiveTime)
732 void (this.destroyWorker(worker
) as Promise
<void>)
739 * This function is the listener registered for each worker message.
741 * @returns The listener function to execute when a message is received from a worker.
743 protected workerListener (): (message
: MessageValue
<Response
>) => void {
745 if (message
.id
!= null) {
746 // Task execution response received
747 const promiseResponse
= this.promiseResponseMap
.get(message
.id
)
748 if (promiseResponse
!= null) {
749 if (message
.taskError
!= null) {
750 promiseResponse
.reject(message
.taskError
.message
)
751 if (this.emitter
!= null) {
752 this.emitter
.emit(PoolEvents
.taskError
, message
.taskError
)
755 promiseResponse
.resolve(message
.data
as Response
)
757 this.afterTaskExecutionHook(promiseResponse
.worker
, message
)
758 this.promiseResponseMap
.delete(message
.id
)
759 const workerNodeKey
= this.getWorkerNodeKey(promiseResponse
.worker
)
761 this.opts
.enableTasksQueue
=== true &&
762 this.tasksQueueSize(workerNodeKey
) > 0
766 this.dequeueTask(workerNodeKey
) as Task
<Data
>
769 this.workerChoiceStrategyContext
.update(workerNodeKey
)
775 private checkAndEmitEvents (): void {
776 if (this.emitter
!= null) {
778 this.emitter
?.emit(PoolEvents
.busy
, this.info
)
780 if (this.type === PoolTypes
.dynamic
&& this.full
) {
781 this.emitter
?.emit(PoolEvents
.full
, this.info
)
787 * Sets the given worker node its tasks usage in the pool.
789 * @param workerNode - The worker node.
790 * @param workerUsage - The worker usage.
792 private setWorkerNodeTasksUsage (
793 workerNode
: WorkerNode
<Worker
, Data
>,
794 workerUsage
: WorkerUsage
796 workerNode
.workerUsage
= workerUsage
800 * Pushes the given worker in the pool worker nodes.
802 * @param worker - The worker.
803 * @returns The worker nodes length.
805 private pushWorkerNode (worker
: Worker
): number {
806 return this.workerNodes
.push({
808 workerUsage
: this.getWorkerUsage(worker
),
809 tasksQueue
: new Queue
<Task
<Data
>>()
814 // * Sets the given worker in the pool worker nodes.
816 // * @param workerNodeKey - The worker node key.
817 // * @param worker - The worker.
818 // * @param workerUsage - The worker usage.
819 // * @param tasksQueue - The worker task queue.
821 // private setWorkerNode (
822 // workerNodeKey: number,
824 // workerUsage: WorkerUsage,
825 // tasksQueue: Queue<Task<Data>>
827 // this.workerNodes[workerNodeKey] = {
835 * Removes the given worker from the pool worker nodes.
837 * @param worker - The worker.
839 private removeWorkerNode (worker
: Worker
): void {
840 const workerNodeKey
= this.getWorkerNodeKey(worker
)
841 if (workerNodeKey
!== -1) {
842 this.workerNodes
.splice(workerNodeKey
, 1)
843 this.workerChoiceStrategyContext
.remove(workerNodeKey
)
847 private executeTask (workerNodeKey
: number, task
: Task
<Data
>): void {
848 this.beforeTaskExecutionHook(workerNodeKey
, task
)
849 this.sendToWorker(this.workerNodes
[workerNodeKey
].worker
, task
)
852 private enqueueTask (workerNodeKey
: number, task
: Task
<Data
>): number {
853 return this.workerNodes
[workerNodeKey
].tasksQueue
.enqueue(task
)
856 private dequeueTask (workerNodeKey
: number): Task
<Data
> | undefined {
857 return this.workerNodes
[workerNodeKey
].tasksQueue
.dequeue()
860 private tasksQueueSize (workerNodeKey
: number): number {
861 return this.workerNodes
[workerNodeKey
].tasksQueue
.size
864 private flushTasksQueue (workerNodeKey
: number): void {
865 if (this.tasksQueueSize(workerNodeKey
) > 0) {
866 for (let i
= 0; i
< this.tasksQueueSize(workerNodeKey
); i
++) {
869 this.dequeueTask(workerNodeKey
) as Task
<Data
>
875 private flushTasksQueues (): void {
876 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
877 this.flushTasksQueue(workerNodeKey
)
881 private setWorkerStatistics (worker
: Worker
): void {
882 this.sendToWorker(worker
, {
885 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
887 elu
: this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
893 private getWorkerUsage (worker
: Worker
): WorkerUsage
{
895 tasks
: this.getTaskStatistics(worker
),
900 history
: new CircularArray()
906 history
: new CircularArray()
913 history
: new CircularArray()
919 history
: new CircularArray()
926 private getTaskStatistics (worker
: Worker
): TaskStatistics
{
928 this.workerNodes
[this.getWorkerNodeKey(worker
)]?.tasksQueue
?.size
932 get
queued (): number {
933 return queueSize
?? 0