1 import { randomUUID
} 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
,
37 WorkerChoiceStrategies
,
38 type WorkerChoiceStrategy
,
39 type WorkerChoiceStrategyOptions
40 } from
'./selection-strategies/selection-strategies-types'
41 import { WorkerChoiceStrategyContext
} from
'./selection-strategies/worker-choice-strategy-context'
42 import { version
} from
'./version'
45 * Base class that implements some shared logic for all poolifier pools.
47 * @typeParam Worker - Type of worker which manages this pool.
48 * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
49 * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
51 export abstract class AbstractPool
<
52 Worker
extends IWorker
,
55 > implements IPool
<Worker
, Data
, Response
> {
57 public readonly workerNodes
: Array<WorkerNode
<Worker
, Data
>> = []
60 public readonly emitter
?: PoolEmitter
63 * The execution response promise map.
65 * - `key`: The message id of each submitted task.
66 * - `value`: An object that contains the worker, the execution response promise resolve and reject callbacks.
68 * When we receive a message from the worker, we get a map entry with the promise resolve/reject bound to the message id.
70 protected promiseResponseMap
: Map
<
72 PromiseResponseWrapper
<Worker
, Response
>
73 > = new Map
<string, PromiseResponseWrapper
<Worker
, Response
>>()
76 * Worker choice strategy context referencing a worker choice algorithm implementation.
78 protected workerChoiceStrategyContext
: WorkerChoiceStrategyContext
<
85 * The start timestamp of the pool.
87 private readonly startTimestamp
90 * Constructs a new poolifier pool.
92 * @param numberOfWorkers - Number of workers that this pool should manage.
93 * @param filePath - Path to the worker file.
94 * @param opts - Options for the pool.
97 protected readonly numberOfWorkers
: number,
98 protected readonly filePath
: string,
99 protected readonly opts
: PoolOptions
<Worker
>
101 if (!this.isMain()) {
102 throw new Error('Cannot start a pool from a worker!')
104 this.checkNumberOfWorkers(this.numberOfWorkers
)
105 this.checkFilePath(this.filePath
)
106 this.checkPoolOptions(this.opts
)
108 this.chooseWorkerNode
= this.chooseWorkerNode
.bind(this)
109 this.executeTask
= this.executeTask
.bind(this)
110 this.enqueueTask
= this.enqueueTask
.bind(this)
111 this.checkAndEmitEvents
= this.checkAndEmitEvents
.bind(this)
113 if (this.opts
.enableEvents
=== true) {
114 this.emitter
= new PoolEmitter()
116 this.workerChoiceStrategyContext
= new WorkerChoiceStrategyContext
<
122 this.opts
.workerChoiceStrategy
,
123 this.opts
.workerChoiceStrategyOptions
128 while (this.workerNodes
.length
< this.numberOfWorkers
) {
129 this.createAndSetupWorker()
132 this.startTimestamp
= performance
.now()
135 private checkFilePath (filePath
: string): void {
138 (typeof filePath
=== 'string' && filePath
.trim().length
=== 0)
140 throw new Error('Please specify a file with a worker implementation')
144 private checkNumberOfWorkers (numberOfWorkers
: number): void {
145 if (numberOfWorkers
== null) {
147 'Cannot instantiate a pool without specifying the number of workers'
149 } else if (!Number.isSafeInteger(numberOfWorkers
)) {
151 'Cannot instantiate a pool with a non safe integer number of workers'
153 } else if (numberOfWorkers
< 0) {
154 throw new RangeError(
155 'Cannot instantiate a pool with a negative number of workers'
157 } else if (this.type === PoolTypes
.fixed
&& numberOfWorkers
=== 0) {
158 throw new Error('Cannot instantiate a fixed pool with no worker')
162 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
163 if (isPlainObject(opts
)) {
164 this.opts
.workerChoiceStrategy
=
165 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
166 this.checkValidWorkerChoiceStrategy(this.opts
.workerChoiceStrategy
)
167 this.opts
.workerChoiceStrategyOptions
=
168 opts
.workerChoiceStrategyOptions
??
169 DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
170 this.checkValidWorkerChoiceStrategyOptions(
171 this.opts
.workerChoiceStrategyOptions
173 this.opts
.restartWorkerOnError
= opts
.restartWorkerOnError
?? true
174 this.opts
.enableEvents
= opts
.enableEvents
?? true
175 this.opts
.enableTasksQueue
= opts
.enableTasksQueue
?? false
176 if (this.opts
.enableTasksQueue
) {
177 this.checkValidTasksQueueOptions(
178 opts
.tasksQueueOptions
as TasksQueueOptions
180 this.opts
.tasksQueueOptions
= this.buildTasksQueueOptions(
181 opts
.tasksQueueOptions
as TasksQueueOptions
185 throw new TypeError('Invalid pool options: must be a plain object')
189 private checkValidWorkerChoiceStrategy (
190 workerChoiceStrategy
: WorkerChoiceStrategy
192 if (!Object.values(WorkerChoiceStrategies
).includes(workerChoiceStrategy
)) {
194 `Invalid worker choice strategy '${workerChoiceStrategy}'`
199 private checkValidWorkerChoiceStrategyOptions (
200 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
202 if (!isPlainObject(workerChoiceStrategyOptions
)) {
204 'Invalid worker choice strategy options: must be a plain object'
208 workerChoiceStrategyOptions
.weights
!= null &&
209 Object.keys(workerChoiceStrategyOptions
.weights
).length
!== this.maxSize
212 'Invalid worker choice strategy options: must have a weight for each worker node'
216 workerChoiceStrategyOptions
.measurement
!= null &&
217 !Object.values(Measurements
).includes(
218 workerChoiceStrategyOptions
.measurement
222 `Invalid worker choice strategy options: invalid measurement '${workerChoiceStrategyOptions.measurement}'`
227 private checkValidTasksQueueOptions (
228 tasksQueueOptions
: TasksQueueOptions
230 if (tasksQueueOptions
!= null && !isPlainObject(tasksQueueOptions
)) {
231 throw new TypeError('Invalid tasks queue options: must be a plain object')
234 tasksQueueOptions
?.concurrency
!= null &&
235 !Number.isSafeInteger(tasksQueueOptions
.concurrency
)
238 'Invalid worker tasks concurrency: must be an integer'
242 tasksQueueOptions
?.concurrency
!= null &&
243 tasksQueueOptions
.concurrency
<= 0
246 `Invalid worker tasks concurrency '${tasksQueueOptions.concurrency}'`
252 public get
info (): PoolInfo
{
257 minSize
: this.minSize
,
258 maxSize
: this.maxSize
,
259 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
260 .runTime
.aggregate
&&
261 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
262 .waitTime
.aggregate
&& { utilization
: round(this.utilization
) }),
263 workerNodes
: this.workerNodes
.length
,
264 idleWorkerNodes
: this.workerNodes
.reduce(
265 (accumulator
, workerNode
) =>
266 workerNode
.usage
.tasks
.executing
=== 0
271 busyWorkerNodes
: this.workerNodes
.reduce(
272 (accumulator
, workerNode
) =>
273 workerNode
.usage
.tasks
.executing
> 0 ? accumulator
+ 1 : accumulator
,
276 executedTasks
: this.workerNodes
.reduce(
277 (accumulator
, workerNode
) =>
278 accumulator
+ workerNode
.usage
.tasks
.executed
,
281 executingTasks
: this.workerNodes
.reduce(
282 (accumulator
, workerNode
) =>
283 accumulator
+ workerNode
.usage
.tasks
.executing
,
286 queuedTasks
: this.workerNodes
.reduce(
287 (accumulator
, workerNode
) =>
288 accumulator
+ workerNode
.usage
.tasks
.queued
,
291 maxQueuedTasks
: this.workerNodes
.reduce(
292 (accumulator
, workerNode
) =>
293 accumulator
+ workerNode
.usage
.tasks
.maxQueued
,
296 failedTasks
: this.workerNodes
.reduce(
297 (accumulator
, workerNode
) =>
298 accumulator
+ workerNode
.usage
.tasks
.failed
,
301 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
302 .runTime
.aggregate
&& {
306 ...this.workerNodes
.map(
307 workerNode
=> workerNode
.usage
.runTime
?.minimum
?? Infinity
313 ...this.workerNodes
.map(
314 workerNode
=> workerNode
.usage
.runTime
?.maximum
?? -Infinity
319 this.workerNodes
.reduce(
320 (accumulator
, workerNode
) =>
321 accumulator
+ (workerNode
.usage
.runTime
?.aggregate
?? 0),
324 this.workerNodes
.reduce(
325 (accumulator
, workerNode
) =>
326 accumulator
+ (workerNode
.usage
.tasks
?.executed
?? 0),
330 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
334 this.workerNodes
.map(
335 workerNode
=> workerNode
.usage
.runTime
?.median
?? 0
342 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
343 .waitTime
.aggregate
&& {
347 ...this.workerNodes
.map(
348 workerNode
=> workerNode
.usage
.waitTime
?.minimum
?? Infinity
354 ...this.workerNodes
.map(
355 workerNode
=> workerNode
.usage
.waitTime
?.maximum
?? -Infinity
360 this.workerNodes
.reduce(
361 (accumulator
, workerNode
) =>
362 accumulator
+ (workerNode
.usage
.waitTime
?.aggregate
?? 0),
365 this.workerNodes
.reduce(
366 (accumulator
, workerNode
) =>
367 accumulator
+ (workerNode
.usage
.tasks
?.executed
?? 0),
371 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
372 .waitTime
.median
&& {
375 this.workerNodes
.map(
376 workerNode
=> workerNode
.usage
.waitTime
?.median
?? 0
387 * Gets the approximate pool utilization.
389 * @returns The pool utilization.
391 private get
utilization (): number {
392 const poolRunTimeCapacity
=
393 (performance
.now() - this.startTimestamp
) * this.maxSize
394 const totalTasksRunTime
= this.workerNodes
.reduce(
395 (accumulator
, workerNode
) =>
396 accumulator
+ (workerNode
.usage
.runTime
?.aggregate
?? 0),
399 const totalTasksWaitTime
= this.workerNodes
.reduce(
400 (accumulator
, workerNode
) =>
401 accumulator
+ (workerNode
.usage
.waitTime
?.aggregate
?? 0),
404 return (totalTasksRunTime
+ totalTasksWaitTime
) / poolRunTimeCapacity
410 * If it is `'dynamic'`, it provides the `max` property.
412 protected abstract get
type (): PoolType
415 * Gets the worker type.
417 protected abstract get
worker (): WorkerType
422 protected abstract get
minSize (): number
427 protected abstract get
maxSize (): number
430 * Get the worker given its id.
432 * @param workerId - The worker id.
433 * @returns The worker if found in the pool worker nodes, `undefined` otherwise.
435 private getWorkerById (workerId
: number): Worker
| undefined {
436 return this.workerNodes
.find(workerNode
=> workerNode
.info
.id
=== workerId
)
441 * Gets the given worker its worker node key.
443 * @param worker - The worker.
444 * @returns The worker node key if found in the pool worker nodes, `-1` otherwise.
446 private getWorkerNodeKey (worker
: Worker
): number {
447 return this.workerNodes
.findIndex(
448 workerNode
=> workerNode
.worker
=== worker
453 public setWorkerChoiceStrategy (
454 workerChoiceStrategy
: WorkerChoiceStrategy
,
455 workerChoiceStrategyOptions
?: WorkerChoiceStrategyOptions
457 this.checkValidWorkerChoiceStrategy(workerChoiceStrategy
)
458 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
459 this.workerChoiceStrategyContext
.setWorkerChoiceStrategy(
460 this.opts
.workerChoiceStrategy
462 if (workerChoiceStrategyOptions
!= null) {
463 this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
465 for (const workerNode
of this.workerNodes
) {
466 this.setWorkerNodeTasksUsage(
468 this.getInitialWorkerUsage(workerNode
.worker
)
470 this.setWorkerStatistics(workerNode
.worker
)
475 public setWorkerChoiceStrategyOptions (
476 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
478 this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
479 this.opts
.workerChoiceStrategyOptions
= workerChoiceStrategyOptions
480 this.workerChoiceStrategyContext
.setOptions(
481 this.opts
.workerChoiceStrategyOptions
486 public enableTasksQueue (
488 tasksQueueOptions
?: TasksQueueOptions
490 if (this.opts
.enableTasksQueue
=== true && !enable
) {
491 this.flushTasksQueues()
493 this.opts
.enableTasksQueue
= enable
494 this.setTasksQueueOptions(tasksQueueOptions
as TasksQueueOptions
)
498 public setTasksQueueOptions (tasksQueueOptions
: TasksQueueOptions
): void {
499 if (this.opts
.enableTasksQueue
=== true) {
500 this.checkValidTasksQueueOptions(tasksQueueOptions
)
501 this.opts
.tasksQueueOptions
=
502 this.buildTasksQueueOptions(tasksQueueOptions
)
503 } else if (this.opts
.tasksQueueOptions
!= null) {
504 delete this.opts
.tasksQueueOptions
508 private buildTasksQueueOptions (
509 tasksQueueOptions
: TasksQueueOptions
510 ): TasksQueueOptions
{
512 concurrency
: tasksQueueOptions
?.concurrency
?? 1
517 * Whether the pool is full or not.
519 * The pool filling boolean status.
521 protected get
full (): boolean {
522 return this.workerNodes
.length
>= this.maxSize
526 * Whether the pool is busy or not.
528 * The pool busyness boolean status.
530 protected abstract get
busy (): boolean
533 * Whether worker nodes are executing at least one task.
535 * @returns Worker nodes busyness boolean status.
537 protected internalBusy (): boolean {
539 this.workerNodes
.findIndex(workerNode
=> {
540 return workerNode
.usage
.tasks
.executing
=== 0
546 public async execute (data
?: Data
, name
?: string): Promise
<Response
> {
547 const timestamp
= performance
.now()
548 const workerNodeKey
= this.chooseWorkerNode()
549 const submittedTask
: Task
<Data
> = {
551 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
552 data
: data
?? ({} as Data
),
556 const res
= new Promise
<Response
>((resolve
, reject
) => {
557 this.promiseResponseMap
.set(submittedTask
.id
as string, {
560 worker
: this.workerNodes
[workerNodeKey
].worker
564 this.opts
.enableTasksQueue
=== true &&
566 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
>=
567 ((this.opts
.tasksQueueOptions
as TasksQueueOptions
)
568 .concurrency
as number))
570 this.enqueueTask(workerNodeKey
, submittedTask
)
572 this.executeTask(workerNodeKey
, submittedTask
)
574 this.checkAndEmitEvents()
575 // eslint-disable-next-line @typescript-eslint/return-await
580 public async destroy (): Promise
<void> {
582 this.workerNodes
.map(async (workerNode
, workerNodeKey
) => {
583 this.flushTasksQueue(workerNodeKey
)
584 // FIXME: wait for tasks to be finished
585 const workerExitPromise
= new Promise
<void>(resolve
=> {
586 workerNode
.worker
.on('exit', () => {
590 await this.destroyWorker(workerNode
.worker
)
591 await workerExitPromise
597 * Terminates the given worker.
599 * @param worker - A worker within `workerNodes`.
601 protected abstract destroyWorker (worker
: Worker
): void | Promise
<void>
604 * Setup hook to execute code before worker nodes are created in the abstract constructor.
609 protected setupHook (): void {
610 // Intentionally empty
614 * Should return whether the worker is the main worker or not.
616 protected abstract isMain (): boolean
619 * Hook executed before the worker task execution.
622 * @param workerNodeKey - The worker node key.
623 * @param task - The task to execute.
625 protected beforeTaskExecutionHook (
626 workerNodeKey
: number,
629 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
630 ++workerUsage
.tasks
.executing
631 this.updateWaitTimeWorkerUsage(workerUsage
, task
)
635 * Hook executed after the worker task execution.
638 * @param worker - The worker.
639 * @param message - The received message.
641 protected afterTaskExecutionHook (
643 message
: MessageValue
<Response
>
645 const workerUsage
= this.workerNodes
[this.getWorkerNodeKey(worker
)].usage
646 this.updateTaskStatisticsWorkerUsage(workerUsage
, message
)
647 this.updateRunTimeWorkerUsage(workerUsage
, message
)
648 this.updateEluWorkerUsage(workerUsage
, message
)
651 private updateTaskStatisticsWorkerUsage (
652 workerUsage
: WorkerUsage
,
653 message
: MessageValue
<Response
>
655 const workerTaskStatistics
= workerUsage
.tasks
656 --workerTaskStatistics
.executing
657 if (message
.taskError
== null) {
658 ++workerTaskStatistics
.executed
660 ++workerTaskStatistics
.failed
664 private updateRunTimeWorkerUsage (
665 workerUsage
: WorkerUsage
,
666 message
: MessageValue
<Response
>
669 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
672 const taskRunTime
= message
.taskPerformance
?.runTime
?? 0
673 workerUsage
.runTime
.aggregate
=
674 (workerUsage
.runTime
.aggregate
?? 0) + taskRunTime
675 workerUsage
.runTime
.minimum
= Math.min(
677 workerUsage
.runTime
?.minimum
?? Infinity
679 workerUsage
.runTime
.maximum
= Math.max(
681 workerUsage
.runTime
?.maximum
?? -Infinity
684 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
686 workerUsage
.tasks
.executed
!== 0
688 workerUsage
.runTime
.average
=
689 workerUsage
.runTime
.aggregate
/ workerUsage
.tasks
.executed
692 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
694 message
.taskPerformance
?.runTime
!= null
696 workerUsage
.runTime
.history
.push(message
.taskPerformance
.runTime
)
697 workerUsage
.runTime
.median
= median(workerUsage
.runTime
.history
)
702 private updateWaitTimeWorkerUsage (
703 workerUsage
: WorkerUsage
,
706 const timestamp
= performance
.now()
707 const taskWaitTime
= timestamp
- (task
.timestamp
?? timestamp
)
709 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().waitTime
712 workerUsage
.waitTime
.aggregate
=
713 (workerUsage
.waitTime
?.aggregate
?? 0) + taskWaitTime
714 workerUsage
.waitTime
.minimum
= Math.min(
716 workerUsage
.waitTime
?.minimum
?? Infinity
718 workerUsage
.waitTime
.maximum
= Math.max(
720 workerUsage
.waitTime
?.maximum
?? -Infinity
723 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
725 workerUsage
.tasks
.executed
!== 0
727 workerUsage
.waitTime
.average
=
728 workerUsage
.waitTime
.aggregate
/ workerUsage
.tasks
.executed
731 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
734 workerUsage
.waitTime
.history
.push(taskWaitTime
)
735 workerUsage
.waitTime
.median
= median(workerUsage
.waitTime
.history
)
740 private updateEluWorkerUsage (
741 workerUsage
: WorkerUsage
,
742 message
: MessageValue
<Response
>
745 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
748 if (message
.taskPerformance
?.elu
!= null) {
749 workerUsage
.elu
.idle
.aggregate
=
750 (workerUsage
.elu
.idle
?.aggregate
?? 0) +
751 message
.taskPerformance
.elu
.idle
752 workerUsage
.elu
.active
.aggregate
=
753 (workerUsage
.elu
.active
?.aggregate
?? 0) +
754 message
.taskPerformance
.elu
.active
755 if (workerUsage
.elu
.utilization
!= null) {
756 workerUsage
.elu
.utilization
=
757 (workerUsage
.elu
.utilization
+
758 message
.taskPerformance
.elu
.utilization
) /
761 workerUsage
.elu
.utilization
= message
.taskPerformance
.elu
.utilization
763 workerUsage
.elu
.idle
.minimum
= Math.min(
764 message
.taskPerformance
.elu
.idle
,
765 workerUsage
.elu
.idle
?.minimum
?? Infinity
767 workerUsage
.elu
.idle
.maximum
= Math.max(
768 message
.taskPerformance
.elu
.idle
,
769 workerUsage
.elu
.idle
?.maximum
?? -Infinity
771 workerUsage
.elu
.active
.minimum
= Math.min(
772 message
.taskPerformance
.elu
.active
,
773 workerUsage
.elu
.active
?.minimum
?? Infinity
775 workerUsage
.elu
.active
.maximum
= Math.max(
776 message
.taskPerformance
.elu
.active
,
777 workerUsage
.elu
.active
?.maximum
?? -Infinity
780 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
782 workerUsage
.tasks
.executed
!== 0
784 workerUsage
.elu
.idle
.average
=
785 workerUsage
.elu
.idle
.aggregate
/ workerUsage
.tasks
.executed
786 workerUsage
.elu
.active
.average
=
787 workerUsage
.elu
.active
.aggregate
/ workerUsage
.tasks
.executed
790 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
793 workerUsage
.elu
.idle
.history
.push(message
.taskPerformance
.elu
.idle
)
794 workerUsage
.elu
.active
.history
.push(
795 message
.taskPerformance
.elu
.active
797 workerUsage
.elu
.idle
.median
= median(workerUsage
.elu
.idle
.history
)
798 workerUsage
.elu
.active
.median
= median(workerUsage
.elu
.active
.history
)
805 * Chooses a worker node for the next task.
807 * The default worker choice strategy uses a round robin algorithm to distribute the tasks.
809 * @returns The worker node key
811 private chooseWorkerNode (): number {
812 if (this.shallCreateDynamicWorker()) {
813 const worker
= this.createAndSetupDynamicWorker()
815 this.workerChoiceStrategyContext
.getStrategyPolicy().useDynamicWorker
817 return this.getWorkerNodeKey(worker
)
820 return this.workerChoiceStrategyContext
.execute()
824 * Conditions for dynamic worker creation.
826 * @returns Whether to create a dynamic worker or not.
828 private shallCreateDynamicWorker (): boolean {
829 return this.type === PoolTypes
.dynamic
&& !this.full
&& this.internalBusy()
833 * Sends a message to the given worker.
835 * @param worker - The worker which should receive the message.
836 * @param message - The message.
838 protected abstract sendToWorker (
840 message
: MessageValue
<Data
>
844 * Registers a listener callback on the given worker.
846 * @param worker - The worker which should register a listener.
847 * @param listener - The message listener callback.
849 private registerWorkerMessageListener
<Message
extends Data
| Response
>(
851 listener
: (message
: MessageValue
<Message
>) => void
853 worker
.on('message', listener
as MessageHandler
<Worker
>)
857 * Creates a new worker.
859 * @returns Newly created worker.
861 protected abstract createWorker (): Worker
864 * Function that can be hooked up when a worker has been newly created and moved to the pool worker nodes.
867 * @param worker - The newly created worker.
869 protected afterWorkerSetup (worker
: Worker
): void {
870 // Listen to worker messages.
871 this.registerWorkerMessageListener(worker
, this.workerListener())
875 * Creates a new worker and sets it up completely in the pool worker nodes.
877 * @returns New, completely set up worker.
879 protected createAndSetupWorker (): Worker
{
880 const worker
= this.createWorker()
882 worker
.on('message', this.opts
.messageHandler
?? EMPTY_FUNCTION
)
883 worker
.on('error', this.opts
.errorHandler
?? EMPTY_FUNCTION
)
884 worker
.on('error', error
=> {
885 if (this.emitter
!= null) {
886 this.emitter
.emit(PoolEvents
.error
, error
)
888 if (this.opts
.enableTasksQueue
=== true) {
889 this.redistributeQueuedTasks(worker
)
891 if (this.opts
.restartWorkerOnError
=== true) {
892 if (this.getWorkerInfo(this.getWorkerNodeKey(worker
)).dynamic
) {
893 this.createAndSetupDynamicWorker()
895 this.createAndSetupWorker()
899 worker
.on('online', this.opts
.onlineHandler
?? EMPTY_FUNCTION
)
900 worker
.on('exit', this.opts
.exitHandler
?? EMPTY_FUNCTION
)
901 worker
.once('exit', () => {
902 this.removeWorkerNode(worker
)
905 this.pushWorkerNode(worker
)
907 this.setWorkerStatistics(worker
)
909 this.afterWorkerSetup(worker
)
914 private redistributeQueuedTasks (worker
: Worker
): void {
915 const workerNodeKey
= this.getWorkerNodeKey(worker
)
916 while (this.tasksQueueSize(workerNodeKey
) > 0) {
917 let targetWorkerNodeKey
: number = workerNodeKey
918 let minQueuedTasks
= Infinity
919 for (const [workerNodeId
, workerNode
] of this.workerNodes
.entries()) {
921 workerNodeId
!== workerNodeKey
&&
922 workerNode
.usage
.tasks
.queued
=== 0
924 targetWorkerNodeKey
= workerNodeId
928 workerNodeId
!== workerNodeKey
&&
929 workerNode
.usage
.tasks
.queued
< minQueuedTasks
931 minQueuedTasks
= workerNode
.usage
.tasks
.queued
932 targetWorkerNodeKey
= workerNodeId
937 this.dequeueTask(workerNodeKey
) as Task
<Data
>
943 * Creates a new dynamic worker and sets it up completely in the pool worker nodes.
945 * @returns New, completely set up dynamic worker.
947 protected createAndSetupDynamicWorker (): Worker
{
948 const worker
= this.createAndSetupWorker()
949 this.getWorkerInfo(this.getWorkerNodeKey(worker
)).dynamic
= true
950 this.registerWorkerMessageListener(worker
, message
=> {
951 const workerNodeKey
= this.getWorkerNodeKey(worker
)
953 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
954 (message
.kill
!= null &&
955 ((this.opts
.enableTasksQueue
=== false &&
956 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
=== 0) ||
957 (this.opts
.enableTasksQueue
=== true &&
958 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
=== 0 &&
959 this.tasksQueueSize(workerNodeKey
) === 0)))
961 // Kill message received from the worker: no new tasks are submitted to that worker for a while ( > maxInactiveTime)
962 void (this.destroyWorker(worker
) as Promise
<void>)
965 this.sendToWorker(worker
, { dynamic
: true })
970 * This function is the listener registered for each worker message.
972 * @returns The listener function to execute when a message is received from a worker.
974 protected workerListener (): (message
: MessageValue
<Response
>) => void {
976 if (message
.workerId
!= null && message
.started
!= null) {
977 // Worker started message received
978 this.handleWorkerStartedMessage(message
)
979 } else if (message
.id
!= null) {
980 // Task execution response received
981 this.handleTaskExecutionResponse(message
)
986 private handleWorkerStartedMessage (message
: MessageValue
<Response
>): void {
987 // Worker started message received
988 const worker
= this.getWorkerById(message
.workerId
as number)
989 if (worker
!= null) {
990 this.workerNodes
[this.getWorkerNodeKey(worker
)].info
.started
=
991 message
.started
as boolean
994 `Worker started message received from unknown worker '${
995 message.workerId as number
1001 private handleTaskExecutionResponse (message
: MessageValue
<Response
>): void {
1002 const promiseResponse
= this.promiseResponseMap
.get(message
.id
as string)
1003 if (promiseResponse
!= null) {
1004 if (message
.taskError
!= null) {
1005 if (this.emitter
!= null) {
1006 this.emitter
.emit(PoolEvents
.taskError
, message
.taskError
)
1008 promiseResponse
.reject(message
.taskError
.message
)
1010 promiseResponse
.resolve(message
.data
as Response
)
1012 this.afterTaskExecutionHook(promiseResponse
.worker
, message
)
1013 this.promiseResponseMap
.delete(message
.id
as string)
1014 const workerNodeKey
= this.getWorkerNodeKey(promiseResponse
.worker
)
1016 this.opts
.enableTasksQueue
=== true &&
1017 this.tasksQueueSize(workerNodeKey
) > 0
1021 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1024 this.workerChoiceStrategyContext
.update(workerNodeKey
)
1028 private checkAndEmitEvents (): void {
1029 if (this.emitter
!= null) {
1031 this.emitter
.emit(PoolEvents
.busy
, this.info
)
1033 if (this.type === PoolTypes
.dynamic
&& this.full
) {
1034 this.emitter
.emit(PoolEvents
.full
, this.info
)
1040 * Sets the given worker node its tasks usage in the pool.
1042 * @param workerNode - The worker node.
1043 * @param workerUsage - The worker usage.
1045 private setWorkerNodeTasksUsage (
1046 workerNode
: WorkerNode
<Worker
, Data
>,
1047 workerUsage
: WorkerUsage
1049 workerNode
.usage
= workerUsage
1053 * Gets the worker information.
1055 * @param workerNodeKey - The worker node key.
1057 private getWorkerInfo (workerNodeKey
: number): WorkerInfo
{
1058 return this.workerNodes
[workerNodeKey
].info
1062 * Pushes the given worker in the pool worker nodes.
1064 * @param worker - The worker.
1065 * @returns The worker nodes length.
1067 private pushWorkerNode (worker
: Worker
): number {
1068 this.workerNodes
.push({
1070 info
: this.getInitialWorkerInfo(worker
),
1071 usage
: this.getInitialWorkerUsage(),
1072 tasksQueue
: new Queue
<Task
<Data
>>()
1074 this.setWorkerNodeTasksUsage(
1075 this.workerNodes
[this.getWorkerNodeKey(worker
)],
1076 this.getInitialWorkerUsage(worker
)
1078 return this.workerNodes
.length
1082 * Gets the worker id.
1084 * @param worker - The worker.
1085 * @returns The worker id.
1087 private getWorkerId (worker
: Worker
): number | undefined {
1088 if (this.worker
=== WorkerTypes
.thread
) {
1089 return worker
.threadId
1090 } else if (this.worker
=== WorkerTypes
.cluster
) {
1096 // * Sets the given worker in the pool worker nodes.
1098 // * @param workerNodeKey - The worker node key.
1099 // * @param worker - The worker.
1100 // * @param workerInfo - The worker info.
1101 // * @param workerUsage - The worker usage.
1102 // * @param tasksQueue - The worker task queue.
1104 // private setWorkerNode (
1105 // workerNodeKey: number,
1107 // workerInfo: WorkerInfo,
1108 // workerUsage: WorkerUsage,
1109 // tasksQueue: Queue<Task<Data>>
1111 // this.workerNodes[workerNodeKey] = {
1113 // info: workerInfo,
1114 // usage: workerUsage,
1120 * Removes the given worker from the pool worker nodes.
1122 * @param worker - The worker.
1124 private removeWorkerNode (worker
: Worker
): void {
1125 const workerNodeKey
= this.getWorkerNodeKey(worker
)
1126 if (workerNodeKey
!== -1) {
1127 this.workerNodes
.splice(workerNodeKey
, 1)
1128 this.workerChoiceStrategyContext
.remove(workerNodeKey
)
1132 private executeTask (workerNodeKey
: number, task
: Task
<Data
>): void {
1133 this.beforeTaskExecutionHook(workerNodeKey
, task
)
1134 this.sendToWorker(this.workerNodes
[workerNodeKey
].worker
, task
)
1137 private enqueueTask (workerNodeKey
: number, task
: Task
<Data
>): number {
1138 return this.workerNodes
[workerNodeKey
].tasksQueue
.enqueue(task
)
1141 private dequeueTask (workerNodeKey
: number): Task
<Data
> | undefined {
1142 return this.workerNodes
[workerNodeKey
].tasksQueue
.dequeue()
1145 private tasksQueueSize (workerNodeKey
: number): number {
1146 return this.workerNodes
[workerNodeKey
].tasksQueue
.size
1149 private tasksMaxQueueSize (workerNodeKey
: number): number {
1150 return this.workerNodes
[workerNodeKey
].tasksQueue
.maxSize
1153 private flushTasksQueue (workerNodeKey
: number): void {
1154 while (this.tasksQueueSize(workerNodeKey
) > 0) {
1157 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1160 this.workerNodes
[workerNodeKey
].tasksQueue
.clear()
1163 private flushTasksQueues (): void {
1164 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
1165 this.flushTasksQueue(workerNodeKey
)
1169 private setWorkerStatistics (worker
: Worker
): void {
1170 this.sendToWorker(worker
, {
1173 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1175 elu
: this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1181 private getInitialWorkerUsage (worker
?: Worker
): WorkerUsage
{
1182 const getTasksQueueSize
= (worker
?: Worker
): number => {
1183 if (worker
== null) {
1186 return this.tasksQueueSize(this.getWorkerNodeKey(worker
))
1188 const getTasksMaxQueueSize
= (worker
?: Worker
): number => {
1189 if (worker
== null) {
1192 return this.tasksMaxQueueSize(this.getWorkerNodeKey(worker
))
1198 get
queued (): number {
1199 return getTasksQueueSize(worker
)
1201 get
maxQueued (): number {
1202 return getTasksMaxQueueSize(worker
)
1207 history
: new CircularArray()
1210 history
: new CircularArray()
1214 history
: new CircularArray()
1217 history
: new CircularArray()
1223 private getInitialWorkerInfo (worker
: Worker
): WorkerInfo
{
1224 return { id
: this.getWorkerId(worker
), dynamic
: false, started
: true }