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
,
12 import { KillBehaviors
} from
'../worker/worker-options'
13 import { CircularArray
} from
'../circular-array'
14 import { Queue
} from
'../queue'
23 type TasksQueueOptions
,
36 WorkerChoiceStrategies
,
37 type WorkerChoiceStrategy
,
38 type WorkerChoiceStrategyOptions
39 } from
'./selection-strategies/selection-strategies-types'
40 import { WorkerChoiceStrategyContext
} from
'./selection-strategies/worker-choice-strategy-context'
41 import { version
} from
'./version'
44 * Base class that implements some shared logic for all poolifier pools.
46 * @typeParam Worker - Type of worker which manages this pool.
47 * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
48 * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
50 export abstract class AbstractPool
<
51 Worker
extends IWorker
,
54 > implements IPool
<Worker
, Data
, Response
> {
56 public readonly workerNodes
: Array<WorkerNode
<Worker
, Data
>> = []
59 public readonly emitter
?: PoolEmitter
62 * The execution response promise map.
64 * - `key`: The message id of each submitted task.
65 * - `value`: An object that contains the worker, the execution response promise resolve and reject callbacks.
67 * When we receive a message from the worker, we get a map entry with the promise resolve/reject bound to the message id.
69 protected promiseResponseMap
: Map
<
71 PromiseResponseWrapper
<Worker
, Response
>
72 > = new Map
<string, PromiseResponseWrapper
<Worker
, Response
>>()
75 * Worker choice strategy context referencing a worker choice algorithm implementation.
77 protected workerChoiceStrategyContext
: WorkerChoiceStrategyContext
<
84 * The start timestamp of the pool.
86 private readonly startTimestamp
89 * Constructs a new poolifier pool.
91 * @param numberOfWorkers - Number of workers that this pool should manage.
92 * @param filePath - Path to the worker file.
93 * @param opts - Options for the pool.
96 protected readonly numberOfWorkers
: number,
97 protected readonly filePath
: string,
98 protected readonly opts
: PoolOptions
<Worker
>
100 if (!this.isMain()) {
101 throw new Error('Cannot start a pool from a worker!')
103 this.checkNumberOfWorkers(this.numberOfWorkers
)
104 this.checkFilePath(this.filePath
)
105 this.checkPoolOptions(this.opts
)
107 this.chooseWorkerNode
= this.chooseWorkerNode
.bind(this)
108 this.executeTask
= this.executeTask
.bind(this)
109 this.enqueueTask
= this.enqueueTask
.bind(this)
110 this.checkAndEmitEvents
= this.checkAndEmitEvents
.bind(this)
112 if (this.opts
.enableEvents
=== true) {
113 this.emitter
= new PoolEmitter()
115 this.workerChoiceStrategyContext
= new WorkerChoiceStrategyContext
<
121 this.opts
.workerChoiceStrategy
,
122 this.opts
.workerChoiceStrategyOptions
127 while (this.workerNodes
.length
< this.numberOfWorkers
) {
128 this.createAndSetupWorker()
131 this.startTimestamp
= performance
.now()
134 private checkFilePath (filePath
: string): void {
137 (typeof filePath
=== 'string' && filePath
.trim().length
=== 0)
139 throw new Error('Please specify a file with a worker implementation')
143 private checkNumberOfWorkers (numberOfWorkers
: number): void {
144 if (numberOfWorkers
== null) {
146 'Cannot instantiate a pool without specifying the number of workers'
148 } else if (!Number.isSafeInteger(numberOfWorkers
)) {
150 'Cannot instantiate a pool with a non safe integer number of workers'
152 } else if (numberOfWorkers
< 0) {
153 throw new RangeError(
154 'Cannot instantiate a pool with a negative number of workers'
156 } else if (this.type === PoolTypes
.fixed
&& numberOfWorkers
=== 0) {
157 throw new Error('Cannot instantiate a fixed pool with no worker')
161 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
162 if (isPlainObject(opts
)) {
163 this.opts
.workerChoiceStrategy
=
164 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
165 this.checkValidWorkerChoiceStrategy(this.opts
.workerChoiceStrategy
)
166 this.opts
.workerChoiceStrategyOptions
=
167 opts
.workerChoiceStrategyOptions
??
168 DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
169 this.checkValidWorkerChoiceStrategyOptions(
170 this.opts
.workerChoiceStrategyOptions
172 this.opts
.restartWorkerOnError
= opts
.restartWorkerOnError
?? true
173 this.opts
.enableEvents
= opts
.enableEvents
?? true
174 this.opts
.enableTasksQueue
= opts
.enableTasksQueue
?? false
175 if (this.opts
.enableTasksQueue
) {
176 this.checkValidTasksQueueOptions(
177 opts
.tasksQueueOptions
as TasksQueueOptions
179 this.opts
.tasksQueueOptions
= this.buildTasksQueueOptions(
180 opts
.tasksQueueOptions
as TasksQueueOptions
184 throw new TypeError('Invalid pool options: must be a plain object')
188 private checkValidWorkerChoiceStrategy (
189 workerChoiceStrategy
: WorkerChoiceStrategy
191 if (!Object.values(WorkerChoiceStrategies
).includes(workerChoiceStrategy
)) {
193 `Invalid worker choice strategy '${workerChoiceStrategy}'`
198 private checkValidWorkerChoiceStrategyOptions (
199 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
201 if (!isPlainObject(workerChoiceStrategyOptions
)) {
203 'Invalid worker choice strategy options: must be a plain object'
207 workerChoiceStrategyOptions
.weights
!= null &&
208 Object.keys(workerChoiceStrategyOptions
.weights
).length
!== this.maxSize
211 'Invalid worker choice strategy options: must have a weight for each worker node'
215 workerChoiceStrategyOptions
.measurement
!= null &&
216 !Object.values(Measurements
).includes(
217 workerChoiceStrategyOptions
.measurement
221 `Invalid worker choice strategy options: invalid measurement '${workerChoiceStrategyOptions.measurement}'`
226 private checkValidTasksQueueOptions (
227 tasksQueueOptions
: TasksQueueOptions
229 if (tasksQueueOptions
!= null && !isPlainObject(tasksQueueOptions
)) {
230 throw new TypeError('Invalid tasks queue options: must be a plain object')
233 tasksQueueOptions
?.concurrency
!= null &&
234 !Number.isSafeInteger(tasksQueueOptions
.concurrency
)
237 'Invalid worker tasks concurrency: must be an integer'
241 tasksQueueOptions
?.concurrency
!= null &&
242 tasksQueueOptions
.concurrency
<= 0
245 `Invalid worker tasks concurrency '${tasksQueueOptions.concurrency}'`
251 public get
info (): PoolInfo
{
256 minSize
: this.minSize
,
257 maxSize
: this.maxSize
,
258 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
259 .runTime
.aggregate
&&
260 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
261 .waitTime
.aggregate
&& { utilization
: round(this.utilization
) }),
262 workerNodes
: this.workerNodes
.length
,
263 idleWorkerNodes
: this.workerNodes
.reduce(
264 (accumulator
, workerNode
) =>
265 workerNode
.usage
.tasks
.executing
=== 0
270 busyWorkerNodes
: this.workerNodes
.reduce(
271 (accumulator
, workerNode
) =>
272 workerNode
.usage
.tasks
.executing
> 0 ? accumulator
+ 1 : accumulator
,
275 executedTasks
: this.workerNodes
.reduce(
276 (accumulator
, workerNode
) =>
277 accumulator
+ workerNode
.usage
.tasks
.executed
,
280 executingTasks
: this.workerNodes
.reduce(
281 (accumulator
, workerNode
) =>
282 accumulator
+ workerNode
.usage
.tasks
.executing
,
285 queuedTasks
: this.workerNodes
.reduce(
286 (accumulator
, workerNode
) =>
287 accumulator
+ workerNode
.usage
.tasks
.queued
,
290 maxQueuedTasks
: this.workerNodes
.reduce(
291 (accumulator
, workerNode
) =>
292 accumulator
+ workerNode
.usage
.tasks
.maxQueued
,
295 failedTasks
: this.workerNodes
.reduce(
296 (accumulator
, workerNode
) =>
297 accumulator
+ workerNode
.usage
.tasks
.failed
,
304 * Gets the approximate pool utilization.
306 * @returns The pool utilization.
308 private get
utilization (): number {
309 const poolRunTimeCapacity
=
310 (performance
.now() - this.startTimestamp
) * this.maxSize
311 const totalTasksRunTime
= this.workerNodes
.reduce(
312 (accumulator
, workerNode
) =>
313 accumulator
+ workerNode
.usage
.runTime
.aggregate
,
316 const totalTasksWaitTime
= this.workerNodes
.reduce(
317 (accumulator
, workerNode
) =>
318 accumulator
+ workerNode
.usage
.waitTime
.aggregate
,
321 return (totalTasksRunTime
+ totalTasksWaitTime
) / poolRunTimeCapacity
327 * If it is `'dynamic'`, it provides the `max` property.
329 protected abstract get
type (): PoolType
332 * Gets the worker type.
334 protected abstract get
worker (): WorkerType
339 protected abstract get
minSize (): number
344 protected abstract get
maxSize (): number
347 * Get the worker given its id.
349 * @param workerId - The worker id.
350 * @returns The worker if found in the pool worker nodes, `undefined` otherwise.
352 private getWorkerById (workerId
: number): Worker
| undefined {
353 return this.workerNodes
.find(workerNode
=> workerNode
.info
.id
=== workerId
)
358 * Gets the given worker its worker node key.
360 * @param worker - The worker.
361 * @returns The worker node key if found in the pool worker nodes, `-1` otherwise.
363 private getWorkerNodeKey (worker
: Worker
): number {
364 return this.workerNodes
.findIndex(
365 workerNode
=> workerNode
.worker
=== worker
370 public setWorkerChoiceStrategy (
371 workerChoiceStrategy
: WorkerChoiceStrategy
,
372 workerChoiceStrategyOptions
?: WorkerChoiceStrategyOptions
374 this.checkValidWorkerChoiceStrategy(workerChoiceStrategy
)
375 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
376 this.workerChoiceStrategyContext
.setWorkerChoiceStrategy(
377 this.opts
.workerChoiceStrategy
379 if (workerChoiceStrategyOptions
!= null) {
380 this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
382 for (const [workerNodeKey
, workerNode
] of this.workerNodes
.entries()) {
383 this.setWorkerNodeTasksUsage(
385 this.getWorkerUsage(workerNodeKey
)
387 this.setWorkerStatistics(workerNode
.worker
)
392 public setWorkerChoiceStrategyOptions (
393 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
395 this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
396 this.opts
.workerChoiceStrategyOptions
= workerChoiceStrategyOptions
397 this.workerChoiceStrategyContext
.setOptions(
398 this.opts
.workerChoiceStrategyOptions
403 public enableTasksQueue (
405 tasksQueueOptions
?: TasksQueueOptions
407 if (this.opts
.enableTasksQueue
=== true && !enable
) {
408 this.flushTasksQueues()
410 this.opts
.enableTasksQueue
= enable
411 this.setTasksQueueOptions(tasksQueueOptions
as TasksQueueOptions
)
415 public setTasksQueueOptions (tasksQueueOptions
: TasksQueueOptions
): void {
416 if (this.opts
.enableTasksQueue
=== true) {
417 this.checkValidTasksQueueOptions(tasksQueueOptions
)
418 this.opts
.tasksQueueOptions
=
419 this.buildTasksQueueOptions(tasksQueueOptions
)
420 } else if (this.opts
.tasksQueueOptions
!= null) {
421 delete this.opts
.tasksQueueOptions
425 private buildTasksQueueOptions (
426 tasksQueueOptions
: TasksQueueOptions
427 ): TasksQueueOptions
{
429 concurrency
: tasksQueueOptions
?.concurrency
?? 1
434 * Whether the pool is full or not.
436 * The pool filling boolean status.
438 protected get
full (): boolean {
439 return this.workerNodes
.length
>= this.maxSize
443 * Whether the pool is busy or not.
445 * The pool busyness boolean status.
447 protected abstract get
busy (): boolean
450 * Whether worker nodes are executing at least one task.
452 * @returns Worker nodes busyness boolean status.
454 protected internalBusy (): boolean {
456 this.workerNodes
.findIndex(workerNode
=> {
457 return workerNode
.usage
.tasks
.executing
=== 0
463 public async execute (data
?: Data
, name
?: string): Promise
<Response
> {
464 const timestamp
= performance
.now()
465 const workerNodeKey
= this.chooseWorkerNode()
466 const submittedTask
: Task
<Data
> = {
468 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
469 data
: data
?? ({} as Data
),
471 id
: crypto
.randomUUID()
473 const res
= new Promise
<Response
>((resolve
, reject
) => {
474 this.promiseResponseMap
.set(submittedTask
.id
as string, {
477 worker
: this.workerNodes
[workerNodeKey
].worker
481 this.opts
.enableTasksQueue
=== true &&
483 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
>=
484 ((this.opts
.tasksQueueOptions
as TasksQueueOptions
)
485 .concurrency
as number))
487 this.enqueueTask(workerNodeKey
, submittedTask
)
489 this.executeTask(workerNodeKey
, submittedTask
)
491 this.checkAndEmitEvents()
492 // eslint-disable-next-line @typescript-eslint/return-await
497 public async destroy (): Promise
<void> {
499 this.workerNodes
.map(async (workerNode
, workerNodeKey
) => {
500 this.flushTasksQueue(workerNodeKey
)
501 // FIXME: wait for tasks to be finished
502 const workerExitPromise
= new Promise
<void>(resolve
=> {
503 workerNode
.worker
.on('exit', () => {
507 await this.destroyWorker(workerNode
.worker
)
508 await workerExitPromise
514 * Terminates the given worker.
516 * @param worker - A worker within `workerNodes`.
518 protected abstract destroyWorker (worker
: Worker
): void | Promise
<void>
521 * Setup hook to execute code before worker nodes are created in the abstract constructor.
526 protected setupHook (): void {
527 // Intentionally empty
531 * Should return whether the worker is the main worker or not.
533 protected abstract isMain (): boolean
536 * Hook executed before the worker task execution.
539 * @param workerNodeKey - The worker node key.
540 * @param task - The task to execute.
542 protected beforeTaskExecutionHook (
543 workerNodeKey
: number,
546 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
547 ++workerUsage
.tasks
.executing
548 this.updateWaitTimeWorkerUsage(workerUsage
, task
)
552 * Hook executed after the worker task execution.
555 * @param worker - The worker.
556 * @param message - The received message.
558 protected afterTaskExecutionHook (
560 message
: MessageValue
<Response
>
562 const workerUsage
= this.workerNodes
[this.getWorkerNodeKey(worker
)].usage
563 this.updateTaskStatisticsWorkerUsage(workerUsage
, message
)
564 this.updateRunTimeWorkerUsage(workerUsage
, message
)
565 this.updateEluWorkerUsage(workerUsage
, message
)
568 private updateTaskStatisticsWorkerUsage (
569 workerUsage
: WorkerUsage
,
570 message
: MessageValue
<Response
>
572 const workerTaskStatistics
= workerUsage
.tasks
573 --workerTaskStatistics
.executing
574 ++workerTaskStatistics
.executed
575 if (message
.taskError
!= null) {
576 ++workerTaskStatistics
.failed
580 private updateRunTimeWorkerUsage (
581 workerUsage
: WorkerUsage
,
582 message
: MessageValue
<Response
>
585 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
588 const taskRunTime
= message
.taskPerformance
?.runTime
?? 0
589 workerUsage
.runTime
.aggregate
+= taskRunTime
590 workerUsage
.runTime
.minimum
= Math.min(
592 workerUsage
.runTime
.minimum
594 workerUsage
.runTime
.maximum
= Math.max(
596 workerUsage
.runTime
.maximum
599 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
601 workerUsage
.tasks
.executed
!== 0
603 workerUsage
.runTime
.average
=
604 workerUsage
.runTime
.aggregate
/
605 (workerUsage
.tasks
.executed
- workerUsage
.tasks
.failed
)
608 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
610 message
.taskPerformance
?.runTime
!= null
612 workerUsage
.runTime
.history
.push(message
.taskPerformance
.runTime
)
613 workerUsage
.runTime
.median
= median(workerUsage
.runTime
.history
)
618 private updateWaitTimeWorkerUsage (
619 workerUsage
: WorkerUsage
,
622 const timestamp
= performance
.now()
623 const taskWaitTime
= timestamp
- (task
.timestamp
?? timestamp
)
625 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().waitTime
628 workerUsage
.waitTime
.aggregate
+= taskWaitTime
629 workerUsage
.waitTime
.minimum
= Math.min(
631 workerUsage
.waitTime
.minimum
633 workerUsage
.waitTime
.maximum
= Math.max(
635 workerUsage
.waitTime
.maximum
638 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
640 workerUsage
.tasks
.executed
!== 0
642 workerUsage
.waitTime
.average
=
643 workerUsage
.waitTime
.aggregate
/
644 (workerUsage
.tasks
.executed
- workerUsage
.tasks
.failed
)
647 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
651 workerUsage
.waitTime
.history
.push(taskWaitTime
)
652 workerUsage
.waitTime
.median
= median(workerUsage
.waitTime
.history
)
657 private updateEluWorkerUsage (
658 workerUsage
: WorkerUsage
,
659 message
: MessageValue
<Response
>
662 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
665 if (message
.taskPerformance
?.elu
!= null) {
666 workerUsage
.elu
.idle
.aggregate
+= message
.taskPerformance
.elu
.idle
667 workerUsage
.elu
.active
.aggregate
+= message
.taskPerformance
.elu
.active
668 if (workerUsage
.elu
.utilization
!= null) {
669 workerUsage
.elu
.utilization
=
670 (workerUsage
.elu
.utilization
+
671 message
.taskPerformance
.elu
.utilization
) /
674 workerUsage
.elu
.utilization
= message
.taskPerformance
.elu
.utilization
676 workerUsage
.elu
.idle
.minimum
= Math.min(
677 message
.taskPerformance
.elu
.idle
,
678 workerUsage
.elu
.idle
.minimum
680 workerUsage
.elu
.idle
.maximum
= Math.max(
681 message
.taskPerformance
.elu
.idle
,
682 workerUsage
.elu
.idle
.maximum
684 workerUsage
.elu
.active
.minimum
= Math.min(
685 message
.taskPerformance
.elu
.active
,
686 workerUsage
.elu
.active
.minimum
688 workerUsage
.elu
.active
.maximum
= Math.max(
689 message
.taskPerformance
.elu
.active
,
690 workerUsage
.elu
.active
.maximum
694 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
696 workerUsage
.tasks
.executed
!== 0
698 const executedTasks
=
699 workerUsage
.tasks
.executed
- workerUsage
.tasks
.failed
700 workerUsage
.elu
.idle
.average
=
701 workerUsage
.elu
.idle
.aggregate
/ executedTasks
702 workerUsage
.elu
.active
.average
=
703 workerUsage
.elu
.active
.aggregate
/ executedTasks
706 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
708 message
.taskPerformance
?.elu
!= null
710 workerUsage
.elu
.idle
.history
.push(message
.taskPerformance
.elu
.idle
)
711 workerUsage
.elu
.active
.history
.push(message
.taskPerformance
.elu
.active
)
712 workerUsage
.elu
.idle
.median
= median(workerUsage
.elu
.idle
.history
)
713 workerUsage
.elu
.active
.median
= median(workerUsage
.elu
.active
.history
)
719 * Chooses a worker node for the next task.
721 * The default worker choice strategy uses a round robin algorithm to distribute the tasks.
723 * @returns The worker node key
725 private chooseWorkerNode (): number {
726 if (this.shallCreateDynamicWorker()) {
727 const worker
= this.createAndSetupDynamicWorker()
729 this.workerChoiceStrategyContext
.getStrategyPolicy().useDynamicWorker
731 return this.getWorkerNodeKey(worker
)
734 return this.workerChoiceStrategyContext
.execute()
738 * Conditions for dynamic worker creation.
740 * @returns Whether to create a dynamic worker or not.
742 private shallCreateDynamicWorker (): boolean {
743 return this.type === PoolTypes
.dynamic
&& !this.full
&& this.internalBusy()
747 * Sends a message to the given worker.
749 * @param worker - The worker which should receive the message.
750 * @param message - The message.
752 protected abstract sendToWorker (
754 message
: MessageValue
<Data
>
758 * Registers a listener callback on the given worker.
760 * @param worker - The worker which should register a listener.
761 * @param listener - The message listener callback.
763 private registerWorkerMessageListener
<Message
extends Data
| Response
>(
765 listener
: (message
: MessageValue
<Message
>) => void
767 worker
.on('message', listener
as MessageHandler
<Worker
>)
771 * Creates a new worker.
773 * @returns Newly created worker.
775 protected abstract createWorker (): Worker
778 * Function that can be hooked up when a worker has been newly created and moved to the pool worker nodes.
781 * @param worker - The newly created worker.
783 protected afterWorkerSetup (worker
: Worker
): void {
784 // Listen to worker messages.
785 this.registerWorkerMessageListener(worker
, this.workerListener())
789 * Creates a new worker and sets it up completely in the pool worker nodes.
791 * @returns New, completely set up worker.
793 protected createAndSetupWorker (): Worker
{
794 const worker
= this.createWorker()
796 worker
.on('message', this.opts
.messageHandler
?? EMPTY_FUNCTION
)
797 worker
.on('error', this.opts
.errorHandler
?? EMPTY_FUNCTION
)
798 worker
.on('error', error
=> {
799 if (this.emitter
!= null) {
800 this.emitter
.emit(PoolEvents
.error
, error
)
802 if (this.opts
.enableTasksQueue
=== true) {
803 const workerNodeKey
= this.getWorkerNodeKey(worker
)
804 while (this.tasksQueueSize(workerNodeKey
) > 0) {
805 let targetWorkerNodeKey
: number = workerNodeKey
806 let minQueuedTasks
= Infinity
807 for (const [workerNodeId
, workerNode
] of this.workerNodes
.entries()) {
809 workerNodeId
!== workerNodeKey
&&
810 workerNode
.usage
.tasks
.queued
=== 0
812 targetWorkerNodeKey
= workerNodeId
816 workerNodeId
!== workerNodeKey
&&
817 workerNode
.usage
.tasks
.queued
< minQueuedTasks
819 minQueuedTasks
= workerNode
.usage
.tasks
.queued
820 targetWorkerNodeKey
= workerNodeId
825 this.dequeueTask(workerNodeKey
) as Task
<Data
>
829 if (this.opts
.restartWorkerOnError
=== true) {
830 this.createAndSetupWorker()
833 worker
.on('online', this.opts
.onlineHandler
?? EMPTY_FUNCTION
)
834 worker
.on('exit', this.opts
.exitHandler
?? EMPTY_FUNCTION
)
835 worker
.once('exit', () => {
836 this.removeWorkerNode(worker
)
839 this.pushWorkerNode(worker
)
841 this.setWorkerStatistics(worker
)
843 this.afterWorkerSetup(worker
)
849 * Creates a new dynamic worker and sets it up completely in the pool worker nodes.
851 * @returns New, completely set up dynamic worker.
853 protected createAndSetupDynamicWorker (): Worker
{
854 const worker
= this.createAndSetupWorker()
855 this.registerWorkerMessageListener(worker
, message
=> {
856 const workerNodeKey
= this.getWorkerNodeKey(worker
)
858 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
859 (message
.kill
!= null &&
860 ((this.opts
.enableTasksQueue
=== false &&
861 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
=== 0) ||
862 (this.opts
.enableTasksQueue
=== true &&
863 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
=== 0 &&
864 this.tasksQueueSize(workerNodeKey
) === 0)))
866 // Kill message received from the worker: no new tasks are submitted to that worker for a while ( > maxInactiveTime)
867 void (this.destroyWorker(worker
) as Promise
<void>)
874 * This function is the listener registered for each worker message.
876 * @returns The listener function to execute when a message is received from a worker.
878 protected workerListener (): (message
: MessageValue
<Response
>) => void {
880 if (message
.workerId
!= null && message
.started
!= null) {
881 // Worker started message received
882 this.handleWorkerStartedMessage(message
)
883 } else if (message
.id
!= null) {
884 // Task execution response received
885 this.handleTaskExecutionResponse(message
)
890 private handleWorkerStartedMessage (message
: MessageValue
<Response
>): void {
891 // Worker started message received
892 const worker
= this.getWorkerById(message
.workerId
as number)
893 if (worker
!= null) {
894 this.workerNodes
[this.getWorkerNodeKey(worker
)].info
.started
=
895 message
.started
as boolean
898 `Worker started message received from unknown worker '${
899 message.workerId as number
905 private handleTaskExecutionResponse (message
: MessageValue
<Response
>): void {
906 const promiseResponse
= this.promiseResponseMap
.get(message
.id
as string)
907 if (promiseResponse
!= null) {
908 if (message
.taskError
!= null) {
909 if (this.emitter
!= null) {
910 this.emitter
.emit(PoolEvents
.taskError
, message
.taskError
)
912 promiseResponse
.reject(message
.taskError
.message
)
914 promiseResponse
.resolve(message
.data
as Response
)
916 this.afterTaskExecutionHook(promiseResponse
.worker
, message
)
917 this.promiseResponseMap
.delete(message
.id
as string)
918 const workerNodeKey
= this.getWorkerNodeKey(promiseResponse
.worker
)
920 this.opts
.enableTasksQueue
=== true &&
921 this.tasksQueueSize(workerNodeKey
) > 0
925 this.dequeueTask(workerNodeKey
) as Task
<Data
>
928 this.workerChoiceStrategyContext
.update(workerNodeKey
)
932 private checkAndEmitEvents (): void {
933 if (this.emitter
!= null) {
935 this.emitter
?.emit(PoolEvents
.busy
, this.info
)
937 if (this.type === PoolTypes
.dynamic
&& this.full
) {
938 this.emitter
?.emit(PoolEvents
.full
, this.info
)
944 * Sets the given worker node its tasks usage in the pool.
946 * @param workerNode - The worker node.
947 * @param workerUsage - The worker usage.
949 private setWorkerNodeTasksUsage (
950 workerNode
: WorkerNode
<Worker
, Data
>,
951 workerUsage
: WorkerUsage
953 workerNode
.usage
= workerUsage
957 * Pushes the given worker in the pool worker nodes.
959 * @param worker - The worker.
960 * @returns The worker nodes length.
962 private pushWorkerNode (worker
: Worker
): number {
963 this.workerNodes
.push({
965 info
: { id
: this.getWorkerId(worker
), started
: true },
966 usage
: this.getWorkerUsage(),
967 tasksQueue
: new Queue
<Task
<Data
>>()
969 const workerNodeKey
= this.getWorkerNodeKey(worker
)
970 this.setWorkerNodeTasksUsage(
971 this.workerNodes
[workerNodeKey
],
972 this.getWorkerUsage(workerNodeKey
)
974 return this.workerNodes
.length
978 * Gets the worker id.
980 * @param worker - The worker.
981 * @returns The worker id.
983 private getWorkerId (worker
: Worker
): number | undefined {
984 if (this.worker
=== WorkerTypes
.thread
) {
985 return worker
.threadId
986 } else if (this.worker
=== WorkerTypes
.cluster
) {
992 // * Sets the given worker in the pool worker nodes.
994 // * @param workerNodeKey - The worker node key.
995 // * @param worker - The worker.
996 // * @param workerInfo - The worker info.
997 // * @param workerUsage - The worker usage.
998 // * @param tasksQueue - The worker task queue.
1000 // private setWorkerNode (
1001 // workerNodeKey: number,
1003 // workerInfo: WorkerInfo,
1004 // workerUsage: WorkerUsage,
1005 // tasksQueue: Queue<Task<Data>>
1007 // this.workerNodes[workerNodeKey] = {
1009 // info: workerInfo,
1010 // usage: workerUsage,
1016 * Removes the given worker from the pool worker nodes.
1018 * @param worker - The worker.
1020 private removeWorkerNode (worker
: Worker
): void {
1021 const workerNodeKey
= this.getWorkerNodeKey(worker
)
1022 if (workerNodeKey
!== -1) {
1023 this.workerNodes
.splice(workerNodeKey
, 1)
1024 this.workerChoiceStrategyContext
.remove(workerNodeKey
)
1028 private executeTask (workerNodeKey
: number, task
: Task
<Data
>): void {
1029 this.beforeTaskExecutionHook(workerNodeKey
, task
)
1030 this.sendToWorker(this.workerNodes
[workerNodeKey
].worker
, task
)
1033 private enqueueTask (workerNodeKey
: number, task
: Task
<Data
>): number {
1034 return this.workerNodes
[workerNodeKey
].tasksQueue
.enqueue(task
)
1037 private dequeueTask (workerNodeKey
: number): Task
<Data
> | undefined {
1038 return this.workerNodes
[workerNodeKey
].tasksQueue
.dequeue()
1041 private tasksQueueSize (workerNodeKey
: number): number {
1042 return this.workerNodes
[workerNodeKey
].tasksQueue
.size
1045 private tasksMaxQueueSize (workerNodeKey
: number): number {
1046 return this.workerNodes
[workerNodeKey
].tasksQueue
.maxSize
1049 private flushTasksQueue (workerNodeKey
: number): void {
1050 while (this.tasksQueueSize(workerNodeKey
) > 0) {
1053 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1056 this.workerNodes
[workerNodeKey
].tasksQueue
.clear()
1059 private flushTasksQueues (): void {
1060 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
1061 this.flushTasksQueue(workerNodeKey
)
1065 private setWorkerStatistics (worker
: Worker
): void {
1066 this.sendToWorker(worker
, {
1069 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1071 elu
: this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1077 private getWorkerUsage (workerNodeKey
?: number): WorkerUsage
{
1078 const getTasksQueueSize
= (workerNodeKey
?: number): number => {
1079 return workerNodeKey
!= null ? this.tasksQueueSize(workerNodeKey
) : 0
1081 const getTasksMaxQueueSize
= (workerNodeKey
?: number): number => {
1082 return workerNodeKey
!= null ? this.tasksMaxQueueSize(workerNodeKey
) : 0
1088 get
queued (): number {
1089 return getTasksQueueSize(workerNodeKey
)
1091 get
maxQueued (): number {
1092 return getTasksMaxQueueSize(workerNodeKey
)
1102 history
: new CircularArray()
1110 history
: new CircularArray()
1119 history
: new CircularArray()
1127 history
: new CircularArray()