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
,
34 WorkerChoiceStrategies
,
35 type WorkerChoiceStrategy
,
36 type WorkerChoiceStrategyOptions
37 } from
'./selection-strategies/selection-strategies-types'
38 import { WorkerChoiceStrategyContext
} from
'./selection-strategies/worker-choice-strategy-context'
41 * Base class that implements some shared logic for all poolifier pools.
43 * @typeParam Worker - Type of worker which manages this pool.
44 * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
45 * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
47 export abstract class AbstractPool
<
48 Worker
extends IWorker
,
51 > implements IPool
<Worker
, Data
, Response
> {
53 public readonly workerNodes
: Array<WorkerNode
<Worker
, Data
>> = []
56 public readonly emitter
?: PoolEmitter
59 * The execution response promise map.
61 * - `key`: The message id of each submitted task.
62 * - `value`: An object that contains the worker, the execution response promise resolve and reject callbacks.
64 * When we receive a message from the worker, we get a map entry with the promise resolve/reject bound to the message id.
66 protected promiseResponseMap
: Map
<
68 PromiseResponseWrapper
<Worker
, Response
>
69 > = new Map
<string, PromiseResponseWrapper
<Worker
, Response
>>()
72 * Worker choice strategy context referencing a worker choice algorithm implementation.
74 protected workerChoiceStrategyContext
: WorkerChoiceStrategyContext
<
81 * 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'
205 workerChoiceStrategyOptions
.measurement
!= null &&
206 !Object.values(Measurements
).includes(
207 workerChoiceStrategyOptions
.measurement
211 `Invalid worker choice strategy options: invalid measurement '${workerChoiceStrategyOptions.measurement}'`
216 private checkValidTasksQueueOptions (
217 tasksQueueOptions
: TasksQueueOptions
219 if (tasksQueueOptions
!= null && !isPlainObject(tasksQueueOptions
)) {
220 throw new TypeError('Invalid tasks queue options: must be a plain object')
223 tasksQueueOptions
?.concurrency
!= null &&
224 !Number.isSafeInteger(tasksQueueOptions
.concurrency
)
227 'Invalid worker tasks concurrency: must be an integer'
231 tasksQueueOptions
?.concurrency
!= null &&
232 tasksQueueOptions
.concurrency
<= 0
235 `Invalid worker tasks concurrency '${tasksQueueOptions.concurrency}'`
240 private get
starting (): boolean {
241 return this.workerNodes
.some(workerNode
=> !workerNode
.info
.started
)
244 private get
started (): boolean {
245 return this.workerNodes
.some(workerNode
=> workerNode
.info
.started
)
249 public get
info (): PoolInfo
{
253 minSize
: this.minSize
,
254 maxSize
: this.maxSize
,
255 workerNodes
: this.workerNodes
.length
,
256 idleWorkerNodes
: this.workerNodes
.reduce(
257 (accumulator
, workerNode
) =>
258 workerNode
.usage
.tasks
.executing
=== 0
263 busyWorkerNodes
: this.workerNodes
.reduce(
264 (accumulator
, workerNode
) =>
265 workerNode
.usage
.tasks
.executing
> 0 ? accumulator
+ 1 : accumulator
,
268 executedTasks
: this.workerNodes
.reduce(
269 (accumulator
, workerNode
) =>
270 accumulator
+ workerNode
.usage
.tasks
.executed
,
273 executingTasks
: this.workerNodes
.reduce(
274 (accumulator
, workerNode
) =>
275 accumulator
+ workerNode
.usage
.tasks
.executing
,
278 queuedTasks
: this.workerNodes
.reduce(
279 (accumulator
, workerNode
) =>
280 accumulator
+ workerNode
.usage
.tasks
.queued
,
283 maxQueuedTasks
: this.workerNodes
.reduce(
284 (accumulator
, workerNode
) =>
285 accumulator
+ workerNode
.usage
.tasks
.maxQueued
,
288 failedTasks
: this.workerNodes
.reduce(
289 (accumulator
, workerNode
) =>
290 accumulator
+ workerNode
.usage
.tasks
.failed
,
299 * If it is `'dynamic'`, it provides the `max` property.
301 protected abstract get
type (): PoolType
304 * Gets the worker type.
306 protected abstract get
worker (): WorkerType
311 protected abstract get
minSize (): number
316 protected abstract get
maxSize (): number
319 * Get the worker given its id.
321 * @param workerId - The worker id.
322 * @returns The worker if found in the pool worker nodes, `undefined` otherwise.
324 private getWorkerById (workerId
: number): Worker
| undefined {
325 return this.workerNodes
.find(workerNode
=> workerNode
.info
.id
=== workerId
)
330 * Gets the given worker its worker node key.
332 * @param worker - The worker.
333 * @returns The worker node key if found in the pool worker nodes, `-1` otherwise.
335 private getWorkerNodeKey (worker
: Worker
): number {
336 return this.workerNodes
.findIndex(
337 workerNode
=> workerNode
.worker
=== worker
342 public setWorkerChoiceStrategy (
343 workerChoiceStrategy
: WorkerChoiceStrategy
,
344 workerChoiceStrategyOptions
?: WorkerChoiceStrategyOptions
346 this.checkValidWorkerChoiceStrategy(workerChoiceStrategy
)
347 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
348 this.workerChoiceStrategyContext
.setWorkerChoiceStrategy(
349 this.opts
.workerChoiceStrategy
351 if (workerChoiceStrategyOptions
!= null) {
352 this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
354 for (const [workerNodeKey
, workerNode
] of this.workerNodes
.entries()) {
355 this.setWorkerNodeTasksUsage(
357 this.getWorkerUsage(workerNodeKey
)
359 this.setWorkerStatistics(workerNode
.worker
)
364 public setWorkerChoiceStrategyOptions (
365 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
367 this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
368 this.opts
.workerChoiceStrategyOptions
= workerChoiceStrategyOptions
369 this.workerChoiceStrategyContext
.setOptions(
370 this.opts
.workerChoiceStrategyOptions
375 public enableTasksQueue (
377 tasksQueueOptions
?: TasksQueueOptions
379 if (this.opts
.enableTasksQueue
=== true && !enable
) {
380 this.flushTasksQueues()
382 this.opts
.enableTasksQueue
= enable
383 this.setTasksQueueOptions(tasksQueueOptions
as TasksQueueOptions
)
387 public setTasksQueueOptions (tasksQueueOptions
: TasksQueueOptions
): void {
388 if (this.opts
.enableTasksQueue
=== true) {
389 this.checkValidTasksQueueOptions(tasksQueueOptions
)
390 this.opts
.tasksQueueOptions
=
391 this.buildTasksQueueOptions(tasksQueueOptions
)
392 } else if (this.opts
.tasksQueueOptions
!= null) {
393 delete this.opts
.tasksQueueOptions
397 private buildTasksQueueOptions (
398 tasksQueueOptions
: TasksQueueOptions
399 ): TasksQueueOptions
{
401 concurrency
: tasksQueueOptions
?.concurrency
?? 1
406 * Whether the pool is full or not.
408 * The pool filling boolean status.
410 protected get
full (): boolean {
411 return this.workerNodes
.length
>= this.maxSize
415 * Whether the pool is busy or not.
417 * The pool busyness boolean status.
419 protected abstract get
busy (): boolean
422 * Whether worker nodes are executing at least one task.
424 * @returns Worker nodes busyness boolean status.
426 protected internalBusy (): boolean {
428 this.workerNodes
.findIndex(workerNode
=> {
429 return workerNode
.usage
.tasks
.executing
=== 0
435 public async execute (data
?: Data
, name
?: string): Promise
<Response
> {
436 const timestamp
= performance
.now()
437 const workerNodeKey
= this.chooseWorkerNode()
438 const submittedTask
: Task
<Data
> = {
440 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
441 data
: data
?? ({} as Data
),
443 id
: crypto
.randomUUID()
445 const res
= new Promise
<Response
>((resolve
, reject
) => {
446 this.promiseResponseMap
.set(submittedTask
.id
as string, {
449 worker
: this.workerNodes
[workerNodeKey
].worker
453 this.opts
.enableTasksQueue
=== true &&
455 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
>=
456 ((this.opts
.tasksQueueOptions
as TasksQueueOptions
)
457 .concurrency
as number))
459 this.enqueueTask(workerNodeKey
, submittedTask
)
461 this.executeTask(workerNodeKey
, submittedTask
)
463 this.checkAndEmitEvents()
464 // eslint-disable-next-line @typescript-eslint/return-await
469 public async destroy (): Promise
<void> {
471 this.workerNodes
.map(async (workerNode
, workerNodeKey
) => {
472 this.flushTasksQueue(workerNodeKey
)
473 // FIXME: wait for tasks to be finished
474 await this.destroyWorker(workerNode
.worker
)
480 * Terminates the given worker.
482 * @param worker - A worker within `workerNodes`.
484 protected abstract destroyWorker (worker
: Worker
): void | Promise
<void>
487 * Setup hook to execute code before worker nodes are created in the abstract constructor.
492 protected setupHook (): void {
493 // Intentionally empty
497 * Should return whether the worker is the main worker or not.
499 protected abstract isMain (): boolean
502 * Hook executed before the worker task execution.
505 * @param workerNodeKey - The worker node key.
506 * @param task - The task to execute.
508 protected beforeTaskExecutionHook (
509 workerNodeKey
: number,
512 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
513 ++workerUsage
.tasks
.executing
514 this.updateWaitTimeWorkerUsage(workerUsage
, task
)
518 * Hook executed after the worker task execution.
521 * @param worker - The worker.
522 * @param message - The received message.
524 protected afterTaskExecutionHook (
526 message
: MessageValue
<Response
>
528 const workerUsage
= this.workerNodes
[this.getWorkerNodeKey(worker
)].usage
529 this.updateTaskStatisticsWorkerUsage(workerUsage
, message
)
530 this.updateRunTimeWorkerUsage(workerUsage
, message
)
531 this.updateEluWorkerUsage(workerUsage
, message
)
534 private updateTaskStatisticsWorkerUsage (
535 workerUsage
: WorkerUsage
,
536 message
: MessageValue
<Response
>
538 const workerTaskStatistics
= workerUsage
.tasks
539 --workerTaskStatistics
.executing
540 ++workerTaskStatistics
.executed
541 if (message
.taskError
!= null) {
542 ++workerTaskStatistics
.failed
546 private updateRunTimeWorkerUsage (
547 workerUsage
: WorkerUsage
,
548 message
: MessageValue
<Response
>
551 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
554 workerUsage
.runTime
.aggregate
+= message
.taskPerformance
?.runTime
?? 0
556 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
558 workerUsage
.tasks
.executed
!== 0
560 workerUsage
.runTime
.average
=
561 workerUsage
.runTime
.aggregate
/
562 (workerUsage
.tasks
.executed
- workerUsage
.tasks
.failed
)
565 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
567 message
.taskPerformance
?.runTime
!= null
569 workerUsage
.runTime
.history
.push(message
.taskPerformance
.runTime
)
570 workerUsage
.runTime
.median
= median(workerUsage
.runTime
.history
)
575 private updateWaitTimeWorkerUsage (
576 workerUsage
: WorkerUsage
,
579 const timestamp
= performance
.now()
580 const taskWaitTime
= timestamp
- (task
.timestamp
?? timestamp
)
582 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().waitTime
585 workerUsage
.waitTime
.aggregate
+= taskWaitTime
?? 0
587 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
589 workerUsage
.tasks
.executed
!== 0
591 workerUsage
.waitTime
.average
=
592 workerUsage
.waitTime
.aggregate
/
593 (workerUsage
.tasks
.executed
- workerUsage
.tasks
.failed
)
596 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
600 workerUsage
.waitTime
.history
.push(taskWaitTime
)
601 workerUsage
.waitTime
.median
= median(workerUsage
.waitTime
.history
)
606 private updateEluWorkerUsage (
607 workerUsage
: WorkerUsage
,
608 message
: MessageValue
<Response
>
611 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
614 if (workerUsage
.elu
!= null && message
.taskPerformance
?.elu
!= null) {
615 workerUsage
.elu
.idle
.aggregate
+= message
.taskPerformance
.elu
.idle
616 workerUsage
.elu
.active
.aggregate
+= message
.taskPerformance
.elu
.active
617 workerUsage
.elu
.utilization
=
618 (workerUsage
.elu
.utilization
+
619 message
.taskPerformance
.elu
.utilization
) /
621 } else if (message
.taskPerformance
?.elu
!= null) {
622 workerUsage
.elu
.idle
.aggregate
= message
.taskPerformance
.elu
.idle
623 workerUsage
.elu
.active
.aggregate
= message
.taskPerformance
.elu
.active
624 workerUsage
.elu
.utilization
= message
.taskPerformance
.elu
.utilization
627 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
629 workerUsage
.tasks
.executed
!== 0
631 const executedTasks
=
632 workerUsage
.tasks
.executed
- workerUsage
.tasks
.failed
633 workerUsage
.elu
.idle
.average
=
634 workerUsage
.elu
.idle
.aggregate
/ executedTasks
635 workerUsage
.elu
.active
.average
=
636 workerUsage
.elu
.active
.aggregate
/ executedTasks
639 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
641 message
.taskPerformance
?.elu
!= null
643 workerUsage
.elu
.idle
.history
.push(message
.taskPerformance
.elu
.idle
)
644 workerUsage
.elu
.active
.history
.push(message
.taskPerformance
.elu
.active
)
645 workerUsage
.elu
.idle
.median
= median(workerUsage
.elu
.idle
.history
)
646 workerUsage
.elu
.active
.median
= median(workerUsage
.elu
.active
.history
)
652 * Chooses a worker node for the next task.
654 * The default worker choice strategy uses a round robin algorithm to distribute the tasks.
656 * @returns The worker node key
658 private chooseWorkerNode (): number {
659 if (this.shallCreateDynamicWorker()) {
660 const worker
= this.createAndSetupDynamicWorker()
662 this.workerChoiceStrategyContext
.getStrategyPolicy().useDynamicWorker
664 return this.getWorkerNodeKey(worker
)
667 return this.workerChoiceStrategyContext
.execute()
671 * Conditions for dynamic worker creation.
673 * @returns Whether to create a dynamic worker or not.
675 private shallCreateDynamicWorker (): boolean {
676 return this.type === PoolTypes
.dynamic
&& !this.full
&& this.internalBusy()
680 * Sends a message to the given worker.
682 * @param worker - The worker which should receive the message.
683 * @param message - The message.
685 protected abstract sendToWorker (
687 message
: MessageValue
<Data
>
691 * Registers a listener callback on the given worker.
693 * @param worker - The worker which should register a listener.
694 * @param listener - The message listener callback.
696 private registerWorkerMessageListener
<Message
extends Data
| Response
>(
698 listener
: (message
: MessageValue
<Message
>) => void
700 worker
.on('message', listener
as MessageHandler
<Worker
>)
704 * Creates a new worker.
706 * @returns Newly created worker.
708 protected abstract createWorker (): Worker
711 * Function that can be hooked up when a worker has been newly created and moved to the pool worker nodes.
714 * @param worker - The newly created worker.
716 protected afterWorkerSetup (worker
: Worker
): void {
717 // Listen to worker messages.
718 this.registerWorkerMessageListener(worker
, this.workerListener())
722 * Creates a new worker and sets it up completely in the pool worker nodes.
724 * @returns New, completely set up worker.
726 protected createAndSetupWorker (): Worker
{
727 const worker
= this.createWorker()
729 worker
.on('message', this.opts
.messageHandler
?? EMPTY_FUNCTION
)
730 worker
.on('error', this.opts
.errorHandler
?? EMPTY_FUNCTION
)
731 worker
.on('error', error
=> {
732 if (this.emitter
!= null) {
733 this.emitter
.emit(PoolEvents
.error
, error
)
735 if (this.opts
.restartWorkerOnError
=== true && !this.starting
) {
736 this.createAndSetupWorker()
739 worker
.on('online', this.opts
.onlineHandler
?? EMPTY_FUNCTION
)
740 worker
.on('exit', this.opts
.exitHandler
?? EMPTY_FUNCTION
)
741 worker
.once('exit', () => {
742 this.removeWorkerNode(worker
)
745 this.pushWorkerNode(worker
)
747 this.setWorkerStatistics(worker
)
749 this.afterWorkerSetup(worker
)
755 * Creates a new dynamic worker and sets it up completely in the pool worker nodes.
757 * @returns New, completely set up dynamic worker.
759 protected createAndSetupDynamicWorker (): Worker
{
760 const worker
= this.createAndSetupWorker()
761 this.registerWorkerMessageListener(worker
, message
=> {
762 const workerNodeKey
= this.getWorkerNodeKey(worker
)
764 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
765 (message
.kill
!= null &&
766 ((this.opts
.enableTasksQueue
=== false &&
767 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
=== 0) ||
768 (this.opts
.enableTasksQueue
=== true &&
769 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
=== 0 &&
770 this.tasksQueueSize(workerNodeKey
) === 0)))
772 // Kill message received from the worker: no new tasks are submitted to that worker for a while ( > maxInactiveTime)
773 void (this.destroyWorker(worker
) as Promise
<void>)
780 * This function is the listener registered for each worker message.
782 * @returns The listener function to execute when a message is received from a worker.
784 protected workerListener (): (message
: MessageValue
<Response
>) => void {
786 if (message
.workerId
!= null && message
.started
!= null) {
787 // Worker started message received
788 const worker
= this.getWorkerById(message
.workerId
)
789 if (worker
!= null) {
790 this.workerNodes
[this.getWorkerNodeKey(worker
)].info
.started
=
793 throw new Error('Worker started message received from unknown worker')
795 } else if (message
.id
!= null) {
796 // Task execution response received
797 const promiseResponse
= this.promiseResponseMap
.get(message
.id
)
798 if (promiseResponse
!= null) {
799 if (message
.taskError
!= null) {
800 if (this.emitter
!= null) {
801 this.emitter
.emit(PoolEvents
.taskError
, message
.taskError
)
803 promiseResponse
.reject(message
.taskError
.message
)
805 promiseResponse
.resolve(message
.data
as Response
)
807 this.afterTaskExecutionHook(promiseResponse
.worker
, message
)
808 this.promiseResponseMap
.delete(message
.id
)
809 const workerNodeKey
= this.getWorkerNodeKey(promiseResponse
.worker
)
811 this.opts
.enableTasksQueue
=== true &&
812 this.tasksQueueSize(workerNodeKey
) > 0
816 this.dequeueTask(workerNodeKey
) as Task
<Data
>
819 this.workerChoiceStrategyContext
.update(workerNodeKey
)
825 private checkAndEmitEvents (): void {
826 if (this.emitter
!= null) {
828 this.emitter
?.emit(PoolEvents
.busy
, this.info
)
830 if (this.type === PoolTypes
.dynamic
&& this.full
) {
831 this.emitter
?.emit(PoolEvents
.full
, this.info
)
837 * Sets the given worker node its tasks usage in the pool.
839 * @param workerNode - The worker node.
840 * @param workerUsage - The worker usage.
842 private setWorkerNodeTasksUsage (
843 workerNode
: WorkerNode
<Worker
, Data
>,
844 workerUsage
: WorkerUsage
846 workerNode
.usage
= workerUsage
850 * Pushes the given worker in the pool worker nodes.
852 * @param worker - The worker.
853 * @returns The worker nodes length.
855 private pushWorkerNode (worker
: Worker
): number {
856 this.workerNodes
.push({
858 info
: { id
: this.getWorkerId(worker
), started
: false },
859 usage
: this.getWorkerUsage(),
860 tasksQueue
: new Queue
<Task
<Data
>>()
862 const workerNodeKey
= this.getWorkerNodeKey(worker
)
863 this.setWorkerNodeTasksUsage(
864 this.workerNodes
[workerNodeKey
],
865 this.getWorkerUsage(workerNodeKey
)
867 return this.workerNodes
.length
871 * Gets the worker id.
873 * @param worker - The worker.
874 * @returns The worker id.
876 private getWorkerId (worker
: Worker
): number | undefined {
877 if (this.worker
=== WorkerTypes
.thread
) {
878 return worker
.threadId
879 } else if (this.worker
=== WorkerTypes
.cluster
) {
885 // * Sets the given worker in the pool worker nodes.
887 // * @param workerNodeKey - The worker node key.
888 // * @param worker - The worker.
889 // * @param workerInfo - The worker info.
890 // * @param workerUsage - The worker usage.
891 // * @param tasksQueue - The worker task queue.
893 // private setWorkerNode (
894 // workerNodeKey: number,
896 // workerInfo: WorkerInfo,
897 // workerUsage: WorkerUsage,
898 // tasksQueue: Queue<Task<Data>>
900 // this.workerNodes[workerNodeKey] = {
903 // usage: workerUsage,
909 * Removes the given worker from the pool worker nodes.
911 * @param worker - The worker.
913 private removeWorkerNode (worker
: Worker
): void {
914 const workerNodeKey
= this.getWorkerNodeKey(worker
)
915 if (workerNodeKey
!== -1) {
916 this.workerNodes
.splice(workerNodeKey
, 1)
917 this.workerChoiceStrategyContext
.remove(workerNodeKey
)
921 private executeTask (workerNodeKey
: number, task
: Task
<Data
>): void {
922 this.beforeTaskExecutionHook(workerNodeKey
, task
)
923 this.sendToWorker(this.workerNodes
[workerNodeKey
].worker
, task
)
926 private enqueueTask (workerNodeKey
: number, task
: Task
<Data
>): number {
927 return this.workerNodes
[workerNodeKey
].tasksQueue
.enqueue(task
)
930 private dequeueTask (workerNodeKey
: number): Task
<Data
> | undefined {
931 return this.workerNodes
[workerNodeKey
].tasksQueue
.dequeue()
934 private tasksQueueSize (workerNodeKey
: number): number {
935 return this.workerNodes
[workerNodeKey
].tasksQueue
.size
938 private tasksMaxQueueSize (workerNodeKey
: number): number {
939 return this.workerNodes
[workerNodeKey
].tasksQueue
.maxSize
942 private flushTasksQueue (workerNodeKey
: number): void {
943 if (this.tasksQueueSize(workerNodeKey
) > 0) {
944 for (let i
= 0; i
< this.tasksQueueSize(workerNodeKey
); i
++) {
947 this.dequeueTask(workerNodeKey
) as Task
<Data
>
951 this.workerNodes
[workerNodeKey
].tasksQueue
.clear()
954 private flushTasksQueues (): void {
955 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
956 this.flushTasksQueue(workerNodeKey
)
960 private setWorkerStatistics (worker
: Worker
): void {
961 this.sendToWorker(worker
, {
964 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
966 elu
: this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
972 private getWorkerUsage (workerNodeKey
?: number): WorkerUsage
{
973 const getTasksQueueSize
= (workerNodeKey
?: number): number => {
974 return workerNodeKey
!= null ? this.tasksQueueSize(workerNodeKey
) : 0
976 const getTasksMaxQueueSize
= (workerNodeKey
?: number): number => {
977 return workerNodeKey
!= null ? this.tasksMaxQueueSize(workerNodeKey
) : 0
983 get
queued (): number {
984 return getTasksQueueSize(workerNodeKey
)
986 get
maxQueued (): number {
987 return getTasksMaxQueueSize(workerNodeKey
)
995 history
: new CircularArray()
1001 history
: new CircularArray()
1008 history
: new CircularArray()
1014 history
: new CircularArray()