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 * Default to a round robin algorithm.
74 protected workerChoiceStrategyContext
: WorkerChoiceStrategyContext
<
81 * Constructs a new poolifier pool.
83 * @param numberOfWorkers - Number of workers that this pool should manage.
84 * @param filePath - Path to the worker file.
85 * @param opts - Options for the pool.
88 protected readonly numberOfWorkers
: number,
89 protected readonly filePath
: string,
90 protected readonly opts
: PoolOptions
<Worker
>
93 throw new Error('Cannot start a pool from a worker!')
95 this.checkNumberOfWorkers(this.numberOfWorkers
)
96 this.checkFilePath(this.filePath
)
97 this.checkPoolOptions(this.opts
)
99 this.chooseWorkerNode
= this.chooseWorkerNode
.bind(this)
100 this.executeTask
= this.executeTask
.bind(this)
101 this.enqueueTask
= this.enqueueTask
.bind(this)
102 this.checkAndEmitEvents
= this.checkAndEmitEvents
.bind(this)
104 if (this.opts
.enableEvents
=== true) {
105 this.emitter
= new PoolEmitter()
107 this.workerChoiceStrategyContext
= new WorkerChoiceStrategyContext
<
113 this.opts
.workerChoiceStrategy
,
114 this.opts
.workerChoiceStrategyOptions
119 for (let i
= 1; i
<= this.numberOfWorkers
; i
++) {
120 this.createAndSetupWorker()
124 private checkFilePath (filePath
: string): void {
127 (typeof filePath
=== 'string' && filePath
.trim().length
=== 0)
129 throw new Error('Please specify a file with a worker implementation')
133 private checkNumberOfWorkers (numberOfWorkers
: number): void {
134 if (numberOfWorkers
== null) {
136 'Cannot instantiate a pool without specifying the number of workers'
138 } else if (!Number.isSafeInteger(numberOfWorkers
)) {
140 'Cannot instantiate a pool with a non safe integer number of workers'
142 } else if (numberOfWorkers
< 0) {
143 throw new RangeError(
144 'Cannot instantiate a pool with a negative number of workers'
146 } else if (this.type === PoolTypes
.fixed
&& numberOfWorkers
=== 0) {
147 throw new Error('Cannot instantiate a fixed pool with no worker')
151 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
152 if (isPlainObject(opts
)) {
153 this.opts
.workerChoiceStrategy
=
154 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
155 this.checkValidWorkerChoiceStrategy(this.opts
.workerChoiceStrategy
)
156 this.opts
.workerChoiceStrategyOptions
=
157 opts
.workerChoiceStrategyOptions
??
158 DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
159 this.checkValidWorkerChoiceStrategyOptions(
160 this.opts
.workerChoiceStrategyOptions
162 this.opts
.restartWorkerOnError
= opts
.restartWorkerOnError
?? true
163 this.opts
.enableEvents
= opts
.enableEvents
?? true
164 this.opts
.enableTasksQueue
= opts
.enableTasksQueue
?? false
165 if (this.opts
.enableTasksQueue
) {
166 this.checkValidTasksQueueOptions(
167 opts
.tasksQueueOptions
as TasksQueueOptions
169 this.opts
.tasksQueueOptions
= this.buildTasksQueueOptions(
170 opts
.tasksQueueOptions
as TasksQueueOptions
174 throw new TypeError('Invalid pool options: must be a plain object')
178 private checkValidWorkerChoiceStrategy (
179 workerChoiceStrategy
: WorkerChoiceStrategy
181 if (!Object.values(WorkerChoiceStrategies
).includes(workerChoiceStrategy
)) {
183 `Invalid worker choice strategy '${workerChoiceStrategy}'`
188 private checkValidWorkerChoiceStrategyOptions (
189 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
191 if (!isPlainObject(workerChoiceStrategyOptions
)) {
193 'Invalid worker choice strategy options: must be a plain object'
197 workerChoiceStrategyOptions
.weights
!= null &&
198 Object.keys(workerChoiceStrategyOptions
.weights
).length
!== this.maxSize
201 'Invalid worker choice strategy options: must have a weight for each worker node'
206 private checkValidTasksQueueOptions (
207 tasksQueueOptions
: TasksQueueOptions
209 if (tasksQueueOptions
!= null && !isPlainObject(tasksQueueOptions
)) {
210 throw new TypeError('Invalid tasks queue options: must be a plain object')
212 if ((tasksQueueOptions
?.concurrency
as number) <= 0) {
214 `Invalid worker tasks concurrency '${
215 tasksQueueOptions.concurrency as number
222 public get
info (): PoolInfo
{
226 minSize
: this.minSize
,
227 maxSize
: this.maxSize
,
228 workerNodes
: this.workerNodes
.length
,
229 idleWorkerNodes
: this.workerNodes
.reduce(
230 (accumulator
, workerNode
) =>
231 workerNode
.workerUsage
.tasks
.executing
=== 0
236 busyWorkerNodes
: this.workerNodes
.reduce(
237 (accumulator
, workerNode
) =>
238 workerNode
.workerUsage
.tasks
.executing
> 0
243 executedTasks
: this.workerNodes
.reduce(
244 (accumulator
, workerNode
) =>
245 accumulator
+ workerNode
.workerUsage
.tasks
.executed
,
248 executingTasks
: this.workerNodes
.reduce(
249 (accumulator
, workerNode
) =>
250 accumulator
+ workerNode
.workerUsage
.tasks
.executing
,
253 queuedTasks
: this.workerNodes
.reduce(
254 (accumulator
, workerNode
) => accumulator
+ workerNode
.tasksQueue
.size
,
257 maxQueuedTasks
: this.workerNodes
.reduce(
258 (accumulator
, workerNode
) =>
259 accumulator
+ workerNode
.tasksQueue
.maxSize
,
262 failedTasks
: this.workerNodes
.reduce(
263 (accumulator
, workerNode
) =>
264 accumulator
+ workerNode
.workerUsage
.tasks
.failed
,
273 * If it is `'dynamic'`, it provides the `max` property.
275 protected abstract get
type (): PoolType
278 * Gets the worker type.
280 protected abstract get
worker (): WorkerType
285 protected abstract get
minSize (): number
290 protected abstract get
maxSize (): number
293 * Gets the given worker its worker node key.
295 * @param worker - The worker.
296 * @returns The worker node key if the worker is found in the pool worker nodes, `-1` otherwise.
298 private getWorkerNodeKey (worker
: Worker
): number {
299 return this.workerNodes
.findIndex(
300 workerNode
=> workerNode
.worker
=== worker
305 public setWorkerChoiceStrategy (
306 workerChoiceStrategy
: WorkerChoiceStrategy
,
307 workerChoiceStrategyOptions
?: WorkerChoiceStrategyOptions
309 this.checkValidWorkerChoiceStrategy(workerChoiceStrategy
)
310 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
311 this.workerChoiceStrategyContext
.setWorkerChoiceStrategy(
312 this.opts
.workerChoiceStrategy
314 if (workerChoiceStrategyOptions
!= null) {
315 this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
317 for (const workerNode
of this.workerNodes
) {
318 this.setWorkerNodeTasksUsage(
320 this.getWorkerUsage(workerNode
.worker
)
322 this.setWorkerStatistics(workerNode
.worker
)
327 public setWorkerChoiceStrategyOptions (
328 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
330 this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
331 this.opts
.workerChoiceStrategyOptions
= workerChoiceStrategyOptions
332 this.workerChoiceStrategyContext
.setOptions(
333 this.opts
.workerChoiceStrategyOptions
338 public enableTasksQueue (
340 tasksQueueOptions
?: TasksQueueOptions
342 if (this.opts
.enableTasksQueue
=== true && !enable
) {
343 this.flushTasksQueues()
345 this.opts
.enableTasksQueue
= enable
346 this.setTasksQueueOptions(tasksQueueOptions
as TasksQueueOptions
)
350 public setTasksQueueOptions (tasksQueueOptions
: TasksQueueOptions
): void {
351 if (this.opts
.enableTasksQueue
=== true) {
352 this.checkValidTasksQueueOptions(tasksQueueOptions
)
353 this.opts
.tasksQueueOptions
=
354 this.buildTasksQueueOptions(tasksQueueOptions
)
355 } else if (this.opts
.tasksQueueOptions
!= null) {
356 delete this.opts
.tasksQueueOptions
360 private buildTasksQueueOptions (
361 tasksQueueOptions
: TasksQueueOptions
362 ): TasksQueueOptions
{
364 concurrency
: tasksQueueOptions
?.concurrency
?? 1
369 * Whether the pool is full or not.
371 * The pool filling boolean status.
373 protected get
full (): boolean {
374 return this.workerNodes
.length
>= this.maxSize
378 * Whether the pool is busy or not.
380 * The pool busyness boolean status.
382 protected abstract get
busy (): boolean
384 protected internalBusy (): boolean {
386 this.workerNodes
.findIndex(workerNode
=> {
387 return workerNode
.workerUsage
.tasks
.executing
=== 0
393 public async execute (data
?: Data
, name
?: string): Promise
<Response
> {
394 const timestamp
= performance
.now()
395 const workerNodeKey
= this.chooseWorkerNode()
396 const submittedTask
: Task
<Data
> = {
398 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
399 data
: data
?? ({} as Data
),
401 id
: crypto
.randomUUID()
403 const res
= new Promise
<Response
>((resolve
, reject
) => {
404 this.promiseResponseMap
.set(submittedTask
.id
as string, {
407 worker
: this.workerNodes
[workerNodeKey
].worker
411 this.opts
.enableTasksQueue
=== true &&
413 this.workerNodes
[workerNodeKey
].workerUsage
.tasks
.executing
>=
414 ((this.opts
.tasksQueueOptions
as TasksQueueOptions
)
415 .concurrency
as number))
417 this.enqueueTask(workerNodeKey
, submittedTask
)
419 this.executeTask(workerNodeKey
, submittedTask
)
421 this.workerChoiceStrategyContext
.update(workerNodeKey
)
422 this.checkAndEmitEvents()
423 // eslint-disable-next-line @typescript-eslint/return-await
428 public async destroy (): Promise
<void> {
430 this.workerNodes
.map(async (workerNode
, workerNodeKey
) => {
431 this.flushTasksQueue(workerNodeKey
)
432 // FIXME: wait for tasks to be finished
433 await this.destroyWorker(workerNode
.worker
)
439 * Shutdowns the given worker.
441 * @param worker - A worker within `workerNodes`.
443 protected abstract destroyWorker (worker
: Worker
): void | Promise
<void>
446 * Setup hook to execute code before worker node are created in the abstract constructor.
451 protected setupHook (): void {
452 // Intentionally empty
456 * Should return whether the worker is the main worker or not.
458 protected abstract isMain (): boolean
461 * Hook executed before the worker task execution.
464 * @param workerNodeKey - The worker node key.
465 * @param task - The task to execute.
467 protected beforeTaskExecutionHook (
468 workerNodeKey
: number,
471 const workerUsage
= this.workerNodes
[workerNodeKey
].workerUsage
472 ++workerUsage
.tasks
.executing
473 this.updateWaitTimeWorkerUsage(workerUsage
, task
)
477 * Hook executed after the worker task execution.
480 * @param worker - The worker.
481 * @param message - The received message.
483 protected afterTaskExecutionHook (
485 message
: MessageValue
<Response
>
488 this.workerNodes
[this.getWorkerNodeKey(worker
)].workerUsage
489 const workerTaskStatistics
= workerUsage
.tasks
490 --workerTaskStatistics
.executing
491 ++workerTaskStatistics
.executed
492 if (message
.taskError
!= null) {
493 ++workerTaskStatistics
.failed
495 this.updateRunTimeWorkerUsage(workerUsage
, message
)
496 this.updateEluWorkerUsage(workerUsage
, message
)
499 private updateRunTimeWorkerUsage (
500 workerUsage
: WorkerUsage
,
501 message
: MessageValue
<Response
>
504 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
507 workerUsage
.runTime
.aggregate
+= message
.taskPerformance
?.runTime
?? 0
509 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
511 workerUsage
.tasks
.executed
!== 0
513 workerUsage
.runTime
.average
=
514 workerUsage
.runTime
.aggregate
/ workerUsage
.tasks
.executed
517 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
519 message
.taskPerformance
?.runTime
!= null
521 workerUsage
.runTime
.history
.push(message
.taskPerformance
.runTime
)
522 workerUsage
.runTime
.median
= median(workerUsage
.runTime
.history
)
527 private updateWaitTimeWorkerUsage (
528 workerUsage
: WorkerUsage
,
531 const timestamp
= performance
.now()
532 const taskWaitTime
= timestamp
- (task
.timestamp
?? timestamp
)
534 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().waitTime
537 workerUsage
.waitTime
.aggregate
+= taskWaitTime
?? 0
539 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
541 workerUsage
.tasks
.executed
!== 0
543 workerUsage
.waitTime
.average
=
544 workerUsage
.waitTime
.aggregate
/ workerUsage
.tasks
.executed
547 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
551 workerUsage
.waitTime
.history
.push(taskWaitTime
)
552 workerUsage
.waitTime
.median
= median(workerUsage
.waitTime
.history
)
557 private updateEluWorkerUsage (
558 workerUsage
: WorkerUsage
,
559 message
: MessageValue
<Response
>
562 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
565 if (workerUsage
.elu
!= null && message
.taskPerformance
?.elu
!= null) {
566 workerUsage
.elu
.idle
.aggregate
=
567 workerUsage
.elu
.idle
.aggregate
+ message
.taskPerformance
.elu
.idle
568 workerUsage
.elu
.active
.aggregate
=
569 workerUsage
.elu
.active
.aggregate
+ message
.taskPerformance
.elu
.active
570 workerUsage
.elu
.utilization
=
571 (workerUsage
.elu
.utilization
+
572 message
.taskPerformance
.elu
.utilization
) /
574 } else if (message
.taskPerformance
?.elu
!= null) {
575 workerUsage
.elu
.idle
.aggregate
= message
.taskPerformance
.elu
.idle
576 workerUsage
.elu
.active
.aggregate
= message
.taskPerformance
.elu
.active
577 workerUsage
.elu
.utilization
= message
.taskPerformance
.elu
.utilization
580 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
582 workerUsage
.tasks
.executed
!== 0
584 workerUsage
.elu
.idle
.average
=
585 workerUsage
.elu
.idle
.aggregate
/ workerUsage
.tasks
.executed
586 workerUsage
.elu
.active
.average
=
587 workerUsage
.elu
.active
.aggregate
/ workerUsage
.tasks
.executed
590 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
592 message
.taskPerformance
?.elu
!= null
594 workerUsage
.elu
.idle
.history
.push(message
.taskPerformance
.elu
.idle
)
595 workerUsage
.elu
.active
.history
.push(message
.taskPerformance
.elu
.active
)
596 workerUsage
.elu
.idle
.median
= median(workerUsage
.elu
.idle
.history
)
597 workerUsage
.elu
.active
.median
= median(workerUsage
.elu
.active
.history
)
603 * Chooses a worker node for the next task.
605 * The default worker choice strategy uses a round robin algorithm to distribute the load.
607 * @returns The worker node key
609 protected chooseWorkerNode (): number {
610 let workerNodeKey
: number
611 if (this.type === PoolTypes
.dynamic
&& !this.full
&& this.internalBusy()) {
612 const workerCreated
= this.createAndSetupWorker()
613 this.registerWorkerMessageListener(workerCreated
, message
=> {
614 const currentWorkerNodeKey
= this.getWorkerNodeKey(workerCreated
)
616 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
617 (message
.kill
!= null &&
618 this.workerNodes
[currentWorkerNodeKey
].workerUsage
.tasks
621 // Kill message received from the worker: no new tasks are submitted to that worker for a while ( > maxInactiveTime)
622 this.flushTasksQueue(currentWorkerNodeKey
)
623 // FIXME: wait for tasks to be finished
624 void (this.destroyWorker(workerCreated
) as Promise
<void>)
627 workerNodeKey
= this.getWorkerNodeKey(workerCreated
)
629 workerNodeKey
= this.workerChoiceStrategyContext
.execute()
635 * Sends a message to the given worker.
637 * @param worker - The worker which should receive the message.
638 * @param message - The message.
640 protected abstract sendToWorker (
642 message
: MessageValue
<Data
>
646 * Registers a listener callback on the given worker.
648 * @param worker - The worker which should register a listener.
649 * @param listener - The message listener callback.
651 protected abstract registerWorkerMessageListener
<
652 Message
extends Data
| Response
653 >(worker
: Worker
, listener
: (message
: MessageValue
<Message
>) => void): void
656 * Returns a newly created worker.
658 protected abstract createWorker (): Worker
661 * Function that can be hooked up when a worker has been newly created and moved to the pool worker nodes.
663 * Can be used to update the `maxListeners` or binding the `main-worker`\<-\>`worker` connection if not bind by default.
665 * @param worker - The newly created worker.
667 protected abstract afterWorkerSetup (worker
: Worker
): void
670 * Creates a new worker and sets it up completely in the pool worker nodes.
672 * @returns New, completely set up worker.
674 protected createAndSetupWorker (): Worker
{
675 const worker
= this.createWorker()
677 worker
.on('message', this.opts
.messageHandler
?? EMPTY_FUNCTION
)
678 worker
.on('error', this.opts
.errorHandler
?? EMPTY_FUNCTION
)
679 worker
.on('error', error
=> {
680 if (this.emitter
!= null) {
681 this.emitter
.emit(PoolEvents
.error
, error
)
684 worker
.on('error', () => {
685 if (this.opts
.restartWorkerOnError
=== true) {
686 this.createAndSetupWorker()
689 worker
.on('online', this.opts
.onlineHandler
?? EMPTY_FUNCTION
)
690 worker
.on('exit', this.opts
.exitHandler
?? EMPTY_FUNCTION
)
691 worker
.once('exit', () => {
692 this.removeWorkerNode(worker
)
695 this.pushWorkerNode(worker
)
697 this.setWorkerStatistics(worker
)
699 this.afterWorkerSetup(worker
)
705 * This function is the listener registered for each worker message.
707 * @returns The listener function to execute when a message is received from a worker.
709 protected workerListener (): (message
: MessageValue
<Response
>) => void {
711 if (message
.id
!= null) {
712 // Task execution response received
713 const promiseResponse
= this.promiseResponseMap
.get(message
.id
)
714 if (promiseResponse
!= null) {
715 if (message
.taskError
!= null) {
716 promiseResponse
.reject(message
.taskError
.message
)
717 if (this.emitter
!= null) {
718 this.emitter
.emit(PoolEvents
.taskError
, message
.taskError
)
721 promiseResponse
.resolve(message
.data
as Response
)
723 this.afterTaskExecutionHook(promiseResponse
.worker
, message
)
724 this.promiseResponseMap
.delete(message
.id
)
725 const workerNodeKey
= this.getWorkerNodeKey(promiseResponse
.worker
)
727 this.opts
.enableTasksQueue
=== true &&
728 this.tasksQueueSize(workerNodeKey
) > 0
732 this.dequeueTask(workerNodeKey
) as Task
<Data
>
740 private checkAndEmitEvents (): void {
741 if (this.emitter
!= null) {
743 this.emitter
?.emit(PoolEvents
.busy
, this.info
)
745 if (this.type === PoolTypes
.dynamic
&& this.full
) {
746 this.emitter
?.emit(PoolEvents
.full
, this.info
)
752 * Sets the given worker node its tasks usage in the pool.
754 * @param workerNode - The worker node.
755 * @param workerUsage - The worker usage.
757 private setWorkerNodeTasksUsage (
758 workerNode
: WorkerNode
<Worker
, Data
>,
759 workerUsage
: WorkerUsage
761 workerNode
.workerUsage
= workerUsage
765 * Pushes the given worker in the pool worker nodes.
767 * @param worker - The worker.
768 * @returns The worker nodes length.
770 private pushWorkerNode (worker
: Worker
): number {
771 return this.workerNodes
.push({
773 workerUsage
: this.getWorkerUsage(worker
),
774 tasksQueue
: new Queue
<Task
<Data
>>()
779 // * Sets the given worker in the pool worker nodes.
781 // * @param workerNodeKey - The worker node key.
782 // * @param worker - The worker.
783 // * @param workerUsage - The worker usage.
784 // * @param tasksQueue - The worker task queue.
786 // private setWorkerNode (
787 // workerNodeKey: number,
789 // workerUsage: WorkerUsage,
790 // tasksQueue: Queue<Task<Data>>
792 // this.workerNodes[workerNodeKey] = {
800 * Removes the given worker from the pool worker nodes.
802 * @param worker - The worker.
804 private removeWorkerNode (worker
: Worker
): void {
805 const workerNodeKey
= this.getWorkerNodeKey(worker
)
806 if (workerNodeKey
!== -1) {
807 this.workerNodes
.splice(workerNodeKey
, 1)
808 this.workerChoiceStrategyContext
.remove(workerNodeKey
)
812 private executeTask (workerNodeKey
: number, task
: Task
<Data
>): void {
813 this.beforeTaskExecutionHook(workerNodeKey
, task
)
814 this.sendToWorker(this.workerNodes
[workerNodeKey
].worker
, task
)
817 private enqueueTask (workerNodeKey
: number, task
: Task
<Data
>): number {
818 return this.workerNodes
[workerNodeKey
].tasksQueue
.enqueue(task
)
821 private dequeueTask (workerNodeKey
: number): Task
<Data
> | undefined {
822 return this.workerNodes
[workerNodeKey
].tasksQueue
.dequeue()
825 private tasksQueueSize (workerNodeKey
: number): number {
826 return this.workerNodes
[workerNodeKey
].tasksQueue
.size
829 private flushTasksQueue (workerNodeKey
: number): void {
830 if (this.tasksQueueSize(workerNodeKey
) > 0) {
831 for (let i
= 0; i
< this.tasksQueueSize(workerNodeKey
); i
++) {
834 this.dequeueTask(workerNodeKey
) as Task
<Data
>
840 private flushTasksQueues (): void {
841 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
842 this.flushTasksQueue(workerNodeKey
)
846 private setWorkerStatistics (worker
: Worker
): void {
847 this.sendToWorker(worker
, {
850 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
852 elu
: this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
858 private getWorkerUsage (worker
: Worker
): WorkerUsage
{
860 tasks
: this.getTaskStatistics(worker
),
865 history
: new CircularArray()
871 history
: new CircularArray()
878 history
: new CircularArray()
884 history
: new CircularArray()
891 private getTaskStatistics (worker
: Worker
): TaskStatistics
{
893 this.workerNodes
[this.getWorkerNodeKey(worker
)]?.tasksQueue
?.size
897 get
queued (): number {
898 return queueSize
?? 0