1 import { randomUUID
} from
'node:crypto'
2 import { performance
} from
'node:perf_hooks'
3 import { existsSync
} from
'node:fs'
4 import { type TransferListItem
} from
'node:worker_threads'
7 PromiseResponseWrapper
,
9 } from
'../utility-types'
12 DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
,
18 updateMeasurementStatistics
20 import { KillBehaviors
} from
'../worker/worker-options'
29 type TasksQueueOptions
39 type MeasurementStatisticsRequirements
,
41 WorkerChoiceStrategies
,
42 type WorkerChoiceStrategy
,
43 type WorkerChoiceStrategyOptions
44 } from
'./selection-strategies/selection-strategies-types'
45 import { WorkerChoiceStrategyContext
} from
'./selection-strategies/worker-choice-strategy-context'
46 import { version
} from
'./version'
47 import { WorkerNode
} from
'./worker-node'
50 * Base class that implements some shared logic for all poolifier pools.
52 * @typeParam Worker - Type of worker which manages this pool.
53 * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
54 * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
56 export abstract class AbstractPool
<
57 Worker
extends IWorker
,
60 > implements IPool
<Worker
, Data
, Response
> {
62 public readonly workerNodes
: Array<IWorkerNode
<Worker
, Data
>> = []
65 public readonly emitter
?: PoolEmitter
68 * The task execution response promise map.
70 * - `key`: The message id of each submitted task.
71 * - `value`: An object that contains the worker, the execution response promise resolve and reject callbacks.
73 * When we receive a message from the worker, we get a map entry with the promise resolve/reject bound to the message id.
75 protected promiseResponseMap
: Map
<string, PromiseResponseWrapper
<Response
>> =
76 new Map
<string, PromiseResponseWrapper
<Response
>>()
79 * Worker choice strategy context referencing a worker choice algorithm implementation.
81 protected workerChoiceStrategyContext
: WorkerChoiceStrategyContext
<
88 * Whether the pool is starting or not.
90 private readonly starting
: boolean
92 * The start timestamp of the pool.
94 private readonly startTimestamp
96 * The task function names.
98 private taskFunctions
!: string[]
101 * Constructs a new poolifier pool.
103 * @param numberOfWorkers - Number of workers that this pool should manage.
104 * @param filePath - Path to the worker file.
105 * @param opts - Options for the pool.
108 protected readonly numberOfWorkers
: number,
109 protected readonly filePath
: string,
110 protected readonly opts
: PoolOptions
<Worker
>
112 if (!this.isMain()) {
114 'Cannot start a pool from a worker with the same type as the pool'
117 this.checkNumberOfWorkers(this.numberOfWorkers
)
118 this.checkFilePath(this.filePath
)
119 this.checkPoolOptions(this.opts
)
121 this.chooseWorkerNode
= this.chooseWorkerNode
.bind(this)
122 this.executeTask
= this.executeTask
.bind(this)
123 this.enqueueTask
= this.enqueueTask
.bind(this)
124 this.dequeueTask
= this.dequeueTask
.bind(this)
125 this.checkAndEmitEvents
= this.checkAndEmitEvents
.bind(this)
127 if (this.opts
.enableEvents
=== true) {
128 this.emitter
= new PoolEmitter()
130 this.workerChoiceStrategyContext
= new WorkerChoiceStrategyContext
<
136 this.opts
.workerChoiceStrategy
,
137 this.opts
.workerChoiceStrategyOptions
144 this.starting
= false
146 this.startTimestamp
= performance
.now()
149 private checkFilePath (filePath
: string): void {
152 typeof filePath
!== 'string' ||
153 (typeof filePath
=== 'string' && filePath
.trim().length
=== 0)
155 throw new Error('Please specify a file with a worker implementation')
157 if (!existsSync(filePath
)) {
158 throw new Error(`Cannot find the worker file '${filePath}'`)
162 private checkNumberOfWorkers (numberOfWorkers
: number): void {
163 if (numberOfWorkers
== null) {
165 'Cannot instantiate a pool without specifying the number of workers'
167 } else if (!Number.isSafeInteger(numberOfWorkers
)) {
169 'Cannot instantiate a pool with a non safe integer number of workers'
171 } else if (numberOfWorkers
< 0) {
172 throw new RangeError(
173 'Cannot instantiate a pool with a negative number of workers'
175 } else if (this.type === PoolTypes
.fixed
&& numberOfWorkers
=== 0) {
176 throw new RangeError('Cannot instantiate a fixed pool with zero worker')
180 protected checkDynamicPoolSize (min
: number, max
: number): void {
181 if (this.type === PoolTypes
.dynamic
) {
184 'Cannot instantiate a dynamic pool without specifying the maximum pool size'
186 } else if (!Number.isSafeInteger(max
)) {
188 'Cannot instantiate a dynamic pool with a non safe integer maximum pool size'
190 } else if (min
> max
) {
191 throw new RangeError(
192 'Cannot instantiate a dynamic pool with a maximum pool size inferior to the minimum pool size'
194 } else if (max
=== 0) {
195 throw new RangeError(
196 'Cannot instantiate a dynamic pool with a maximum pool size equal to zero'
198 } else if (min
=== max
) {
199 throw new RangeError(
200 'Cannot instantiate a dynamic pool with a minimum pool size equal to the maximum pool size. Use a fixed pool instead'
206 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
207 if (isPlainObject(opts
)) {
208 this.opts
.workerChoiceStrategy
=
209 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
210 this.checkValidWorkerChoiceStrategy(this.opts
.workerChoiceStrategy
)
211 this.opts
.workerChoiceStrategyOptions
=
212 opts
.workerChoiceStrategyOptions
??
213 DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
214 this.checkValidWorkerChoiceStrategyOptions(
215 this.opts
.workerChoiceStrategyOptions
217 this.opts
.restartWorkerOnError
= opts
.restartWorkerOnError
?? true
218 this.opts
.enableEvents
= opts
.enableEvents
?? true
219 this.opts
.enableTasksQueue
= opts
.enableTasksQueue
?? false
220 if (this.opts
.enableTasksQueue
) {
221 this.checkValidTasksQueueOptions(
222 opts
.tasksQueueOptions
as TasksQueueOptions
224 this.opts
.tasksQueueOptions
= this.buildTasksQueueOptions(
225 opts
.tasksQueueOptions
as TasksQueueOptions
229 throw new TypeError('Invalid pool options: must be a plain object')
233 private checkValidWorkerChoiceStrategy (
234 workerChoiceStrategy
: WorkerChoiceStrategy
236 if (!Object.values(WorkerChoiceStrategies
).includes(workerChoiceStrategy
)) {
238 `Invalid worker choice strategy '${workerChoiceStrategy}'`
243 private checkValidWorkerChoiceStrategyOptions (
244 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
246 if (!isPlainObject(workerChoiceStrategyOptions
)) {
248 'Invalid worker choice strategy options: must be a plain object'
252 workerChoiceStrategyOptions
.weights
!= null &&
253 Object.keys(workerChoiceStrategyOptions
.weights
).length
!== this.maxSize
256 'Invalid worker choice strategy options: must have a weight for each worker node'
260 workerChoiceStrategyOptions
.measurement
!= null &&
261 !Object.values(Measurements
).includes(
262 workerChoiceStrategyOptions
.measurement
266 `Invalid worker choice strategy options: invalid measurement '${workerChoiceStrategyOptions.measurement}'`
271 private checkValidTasksQueueOptions (
272 tasksQueueOptions
: TasksQueueOptions
274 if (tasksQueueOptions
!= null && !isPlainObject(tasksQueueOptions
)) {
275 throw new TypeError('Invalid tasks queue options: must be a plain object')
278 tasksQueueOptions
?.concurrency
!= null &&
279 !Number.isSafeInteger(tasksQueueOptions
.concurrency
)
282 'Invalid worker tasks concurrency: must be an integer'
286 tasksQueueOptions
?.concurrency
!= null &&
287 tasksQueueOptions
.concurrency
<= 0
290 `Invalid worker tasks concurrency '${tasksQueueOptions.concurrency}'`
295 private startPool (): void {
297 this.workerNodes
.reduce(
298 (accumulator
, workerNode
) =>
299 !workerNode
.info
.dynamic
? accumulator
+ 1 : accumulator
,
301 ) < this.numberOfWorkers
303 this.createAndSetupWorkerNode()
308 public get
info (): PoolInfo
{
314 strategy
: this.opts
.workerChoiceStrategy
as WorkerChoiceStrategy
,
315 minSize
: this.minSize
,
316 maxSize
: this.maxSize
,
317 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
318 .runTime
.aggregate
&&
319 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
320 .waitTime
.aggregate
&& { utilization
: round(this.utilization
) }),
321 workerNodes
: this.workerNodes
.length
,
322 idleWorkerNodes
: this.workerNodes
.reduce(
323 (accumulator
, workerNode
) =>
324 workerNode
.usage
.tasks
.executing
=== 0
329 busyWorkerNodes
: this.workerNodes
.reduce(
330 (accumulator
, workerNode
) =>
331 workerNode
.usage
.tasks
.executing
> 0 ? accumulator
+ 1 : accumulator
,
334 executedTasks
: this.workerNodes
.reduce(
335 (accumulator
, workerNode
) =>
336 accumulator
+ workerNode
.usage
.tasks
.executed
,
339 executingTasks
: this.workerNodes
.reduce(
340 (accumulator
, workerNode
) =>
341 accumulator
+ workerNode
.usage
.tasks
.executing
,
344 ...(this.opts
.enableTasksQueue
=== true && {
345 queuedTasks
: this.workerNodes
.reduce(
346 (accumulator
, workerNode
) =>
347 accumulator
+ workerNode
.usage
.tasks
.queued
,
351 ...(this.opts
.enableTasksQueue
=== true && {
352 maxQueuedTasks
: this.workerNodes
.reduce(
353 (accumulator
, workerNode
) =>
354 accumulator
+ (workerNode
.usage
.tasks
?.maxQueued
?? 0),
358 failedTasks
: this.workerNodes
.reduce(
359 (accumulator
, workerNode
) =>
360 accumulator
+ workerNode
.usage
.tasks
.failed
,
363 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
364 .runTime
.aggregate
&& {
368 ...this.workerNodes
.map(
369 (workerNode
) => workerNode
.usage
.runTime
?.minimum
?? Infinity
375 ...this.workerNodes
.map(
376 (workerNode
) => workerNode
.usage
.runTime
?.maximum
?? -Infinity
381 this.workerNodes
.reduce(
382 (accumulator
, workerNode
) =>
383 accumulator
+ (workerNode
.usage
.runTime
?.aggregate
?? 0),
386 this.workerNodes
.reduce(
387 (accumulator
, workerNode
) =>
388 accumulator
+ (workerNode
.usage
.tasks
?.executed
?? 0),
392 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
396 this.workerNodes
.map(
397 (workerNode
) => workerNode
.usage
.runTime
?.median
?? 0
404 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
405 .waitTime
.aggregate
&& {
409 ...this.workerNodes
.map(
410 (workerNode
) => workerNode
.usage
.waitTime
?.minimum
?? Infinity
416 ...this.workerNodes
.map(
417 (workerNode
) => workerNode
.usage
.waitTime
?.maximum
?? -Infinity
422 this.workerNodes
.reduce(
423 (accumulator
, workerNode
) =>
424 accumulator
+ (workerNode
.usage
.waitTime
?.aggregate
?? 0),
427 this.workerNodes
.reduce(
428 (accumulator
, workerNode
) =>
429 accumulator
+ (workerNode
.usage
.tasks
?.executed
?? 0),
433 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
434 .waitTime
.median
&& {
437 this.workerNodes
.map(
438 (workerNode
) => workerNode
.usage
.waitTime
?.median
?? 0
449 * The pool readiness boolean status.
451 private get
ready (): boolean {
453 this.workerNodes
.reduce(
454 (accumulator
, workerNode
) =>
455 !workerNode
.info
.dynamic
&& workerNode
.info
.ready
464 * The approximate pool utilization.
466 * @returns The pool utilization.
468 private get
utilization (): number {
469 const poolTimeCapacity
=
470 (performance
.now() - this.startTimestamp
) * this.maxSize
471 const totalTasksRunTime
= this.workerNodes
.reduce(
472 (accumulator
, workerNode
) =>
473 accumulator
+ (workerNode
.usage
.runTime
?.aggregate
?? 0),
476 const totalTasksWaitTime
= this.workerNodes
.reduce(
477 (accumulator
, workerNode
) =>
478 accumulator
+ (workerNode
.usage
.waitTime
?.aggregate
?? 0),
481 return (totalTasksRunTime
+ totalTasksWaitTime
) / poolTimeCapacity
487 * If it is `'dynamic'`, it provides the `max` property.
489 protected abstract get
type (): PoolType
494 protected abstract get
worker (): WorkerType
497 * The pool minimum size.
499 protected abstract get
minSize (): number
502 * The pool maximum size.
504 protected abstract get
maxSize (): number
507 * Checks if the worker id sent in the received message from a worker is valid.
509 * @param message - The received message.
510 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the worker id is invalid.
512 private checkMessageWorkerId (message
: MessageValue
<Response
>): void {
514 message
.workerId
!= null &&
515 this.getWorkerNodeKeyByWorkerId(message
.workerId
) === -1
518 `Worker message received from unknown worker '${message.workerId}'`
524 * Gets the given worker its worker node key.
526 * @param worker - The worker.
527 * @returns The worker node key if found in the pool worker nodes, `-1` otherwise.
529 private getWorkerNodeKeyByWorker (worker
: Worker
): number {
530 return this.workerNodes
.findIndex(
531 (workerNode
) => workerNode
.worker
=== worker
536 * Gets the worker node key given its worker id.
538 * @param workerId - The worker id.
539 * @returns The worker node key if the worker id is found in the pool worker nodes, `-1` otherwise.
541 private getWorkerNodeKeyByWorkerId (workerId
: number): number {
542 return this.workerNodes
.findIndex(
543 (workerNode
) => workerNode
.info
.id
=== workerId
548 public setWorkerChoiceStrategy (
549 workerChoiceStrategy
: WorkerChoiceStrategy
,
550 workerChoiceStrategyOptions
?: WorkerChoiceStrategyOptions
552 this.checkValidWorkerChoiceStrategy(workerChoiceStrategy
)
553 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
554 this.workerChoiceStrategyContext
.setWorkerChoiceStrategy(
555 this.opts
.workerChoiceStrategy
557 if (workerChoiceStrategyOptions
!= null) {
558 this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
560 for (const [workerNodeKey
, workerNode
] of this.workerNodes
.entries()) {
561 workerNode
.resetUsage()
562 this.sendStatisticsMessageToWorker(workerNodeKey
)
567 public setWorkerChoiceStrategyOptions (
568 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
570 this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
571 this.opts
.workerChoiceStrategyOptions
= workerChoiceStrategyOptions
572 this.workerChoiceStrategyContext
.setOptions(
573 this.opts
.workerChoiceStrategyOptions
578 public enableTasksQueue (
580 tasksQueueOptions
?: TasksQueueOptions
582 if (this.opts
.enableTasksQueue
=== true && !enable
) {
583 this.flushTasksQueues()
585 this.opts
.enableTasksQueue
= enable
586 this.setTasksQueueOptions(tasksQueueOptions
as TasksQueueOptions
)
590 public setTasksQueueOptions (tasksQueueOptions
: TasksQueueOptions
): void {
591 if (this.opts
.enableTasksQueue
=== true) {
592 this.checkValidTasksQueueOptions(tasksQueueOptions
)
593 this.opts
.tasksQueueOptions
=
594 this.buildTasksQueueOptions(tasksQueueOptions
)
595 } else if (this.opts
.tasksQueueOptions
!= null) {
596 delete this.opts
.tasksQueueOptions
600 private buildTasksQueueOptions (
601 tasksQueueOptions
: TasksQueueOptions
602 ): TasksQueueOptions
{
604 concurrency
: tasksQueueOptions
?.concurrency
?? 1
609 * Whether the pool is full or not.
611 * The pool filling boolean status.
613 protected get
full (): boolean {
614 return this.workerNodes
.length
>= this.maxSize
618 * Whether the pool is busy or not.
620 * The pool busyness boolean status.
622 protected abstract get
busy (): boolean
625 * Whether worker nodes are executing concurrently their tasks quota or not.
627 * @returns Worker nodes busyness boolean status.
629 protected internalBusy (): boolean {
630 if (this.opts
.enableTasksQueue
=== true) {
632 this.workerNodes
.findIndex(
634 workerNode
.info
.ready
&&
635 workerNode
.usage
.tasks
.executing
<
636 (this.opts
.tasksQueueOptions
?.concurrency
as number)
641 this.workerNodes
.findIndex(
643 workerNode
.info
.ready
&& workerNode
.usage
.tasks
.executing
=== 0
650 public listTaskFunctions (): string[] {
651 if (this.taskFunctions
!= null) {
652 return this.taskFunctions
659 public async execute (
662 transferList
?: TransferListItem
[]
663 ): Promise
<Response
> {
664 return await new Promise
<Response
>((resolve
, reject
) => {
665 if (name
!= null && typeof name
!== 'string') {
666 reject(new TypeError('name argument must be a string'))
670 typeof name
=== 'string' &&
671 name
.trim().length
=== 0
673 reject(new TypeError('name argument must not be an empty string'))
677 this.taskFunctions
!= null &&
678 !this.taskFunctions
.includes(name
)
681 new Error(`Task function '${name}' is not registered in the pool`)
684 if (transferList
!= null && !Array.isArray(transferList
)) {
685 reject(new TypeError('transferList argument must be an array'))
687 const timestamp
= performance
.now()
688 const workerNodeKey
= this.chooseWorkerNode()
689 const task
: Task
<Data
> = {
690 name
: name
?? DEFAULT_TASK_NAME
,
691 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
692 data
: data
?? ({} as Data
),
695 workerId
: this.getWorkerInfo(workerNodeKey
).id
as number,
698 this.promiseResponseMap
.set(task
.taskId
as string, {
704 this.opts
.enableTasksQueue
=== false ||
705 (this.opts
.enableTasksQueue
=== true &&
706 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
<
707 (this.opts
.tasksQueueOptions
?.concurrency
as number))
709 this.executeTask(workerNodeKey
, task
)
711 this.enqueueTask(workerNodeKey
, task
)
713 this.checkAndEmitEvents()
718 public async destroy (): Promise
<void> {
720 this.workerNodes
.map(async (_
, workerNodeKey
) => {
721 await this.destroyWorkerNode(workerNodeKey
)
724 this.emitter
?.emit(PoolEvents
.destroy
)
727 protected async sendKillMessageToWorker (
728 workerNodeKey
: number,
731 await new Promise
<void>((resolve
, reject
) => {
732 this.registerWorkerMessageListener(workerNodeKey
, (message
) => {
733 if (message
.kill
=== 'success') {
735 } else if (message
.kill
=== 'failure') {
736 reject(new Error(`Worker ${workerId} kill message handling failed`))
739 this.sendToWorker(workerNodeKey
, { kill
: true, workerId
})
744 * Terminates the worker node given its worker node key.
746 * @param workerNodeKey - The worker node key.
748 protected abstract destroyWorkerNode (workerNodeKey
: number): Promise
<void>
751 * Setup hook to execute code before worker nodes are created in the abstract constructor.
756 protected setupHook (): void {
757 // Intentionally empty
761 * Should return whether the worker is the main worker or not.
763 protected abstract isMain (): boolean
766 * Hook executed before the worker task execution.
769 * @param workerNodeKey - The worker node key.
770 * @param task - The task to execute.
772 protected beforeTaskExecutionHook (
773 workerNodeKey
: number,
776 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
777 ++workerUsage
.tasks
.executing
778 this.updateWaitTimeWorkerUsage(workerUsage
, task
)
779 const taskWorkerUsage
= this.workerNodes
[workerNodeKey
].getTaskWorkerUsage(
782 ++taskWorkerUsage
.tasks
.executing
783 this.updateWaitTimeWorkerUsage(taskWorkerUsage
, task
)
787 * Hook executed after the worker task execution.
790 * @param workerNodeKey - The worker node key.
791 * @param message - The received message.
793 protected afterTaskExecutionHook (
794 workerNodeKey
: number,
795 message
: MessageValue
<Response
>
797 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
798 this.updateTaskStatisticsWorkerUsage(workerUsage
, message
)
799 this.updateRunTimeWorkerUsage(workerUsage
, message
)
800 this.updateEluWorkerUsage(workerUsage
, message
)
801 const taskWorkerUsage
= this.workerNodes
[workerNodeKey
].getTaskWorkerUsage(
802 message
.taskPerformance
?.name
?? DEFAULT_TASK_NAME
804 this.updateTaskStatisticsWorkerUsage(taskWorkerUsage
, message
)
805 this.updateRunTimeWorkerUsage(taskWorkerUsage
, message
)
806 this.updateEluWorkerUsage(taskWorkerUsage
, message
)
809 private updateTaskStatisticsWorkerUsage (
810 workerUsage
: WorkerUsage
,
811 message
: MessageValue
<Response
>
813 const workerTaskStatistics
= workerUsage
.tasks
814 --workerTaskStatistics
.executing
815 if (message
.taskError
== null) {
816 ++workerTaskStatistics
.executed
818 ++workerTaskStatistics
.failed
822 private updateRunTimeWorkerUsage (
823 workerUsage
: WorkerUsage
,
824 message
: MessageValue
<Response
>
826 updateMeasurementStatistics(
828 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
,
829 message
.taskPerformance
?.runTime
?? 0,
830 workerUsage
.tasks
.executed
834 private updateWaitTimeWorkerUsage (
835 workerUsage
: WorkerUsage
,
838 const timestamp
= performance
.now()
839 const taskWaitTime
= timestamp
- (task
.timestamp
?? timestamp
)
840 updateMeasurementStatistics(
841 workerUsage
.waitTime
,
842 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().waitTime
,
844 workerUsage
.tasks
.executed
848 private updateEluWorkerUsage (
849 workerUsage
: WorkerUsage
,
850 message
: MessageValue
<Response
>
852 const eluTaskStatisticsRequirements
: MeasurementStatisticsRequirements
=
853 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
854 updateMeasurementStatistics(
855 workerUsage
.elu
.active
,
856 eluTaskStatisticsRequirements
,
857 message
.taskPerformance
?.elu
?.active
?? 0,
858 workerUsage
.tasks
.executed
860 updateMeasurementStatistics(
861 workerUsage
.elu
.idle
,
862 eluTaskStatisticsRequirements
,
863 message
.taskPerformance
?.elu
?.idle
?? 0,
864 workerUsage
.tasks
.executed
866 if (eluTaskStatisticsRequirements
.aggregate
) {
867 if (message
.taskPerformance
?.elu
!= null) {
868 if (workerUsage
.elu
.utilization
!= null) {
869 workerUsage
.elu
.utilization
=
870 (workerUsage
.elu
.utilization
+
871 message
.taskPerformance
.elu
.utilization
) /
874 workerUsage
.elu
.utilization
= message
.taskPerformance
.elu
.utilization
881 * Chooses a worker node for the next task.
883 * The default worker choice strategy uses a round robin algorithm to distribute the tasks.
885 * @returns The chosen worker node key
887 private chooseWorkerNode (): number {
888 if (this.shallCreateDynamicWorker()) {
889 const workerNodeKey
= this.createAndSetupDynamicWorkerNode()
891 this.workerChoiceStrategyContext
.getStrategyPolicy().useDynamicWorker
896 return this.workerChoiceStrategyContext
.execute()
900 * Conditions for dynamic worker creation.
902 * @returns Whether to create a dynamic worker or not.
904 private shallCreateDynamicWorker (): boolean {
905 return this.type === PoolTypes
.dynamic
&& !this.full
&& this.internalBusy()
909 * Sends a message to worker given its worker node key.
911 * @param workerNodeKey - The worker node key.
912 * @param message - The message.
913 * @param transferList - The optional array of transferable objects.
915 protected abstract sendToWorker (
916 workerNodeKey
: number,
917 message
: MessageValue
<Data
>,
918 transferList
?: TransferListItem
[]
922 * Creates a new worker.
924 * @returns Newly created worker.
926 protected abstract createWorker (): Worker
929 * Creates a new, completely set up worker node.
931 * @returns New, completely set up worker node key.
933 protected createAndSetupWorkerNode (): number {
934 const worker
= this.createWorker()
936 worker
.on('online', this.opts
.onlineHandler
?? EMPTY_FUNCTION
)
937 worker
.on('message', this.opts
.messageHandler
?? EMPTY_FUNCTION
)
938 worker
.on('error', this.opts
.errorHandler
?? EMPTY_FUNCTION
)
939 worker
.on('error', (error
) => {
940 const workerNodeKey
= this.getWorkerNodeKeyByWorker(worker
)
941 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
942 workerInfo
.ready
= false
943 this.workerNodes
[workerNodeKey
].closeChannel()
944 this.emitter
?.emit(PoolEvents
.error
, error
)
945 if (this.opts
.restartWorkerOnError
=== true && !this.starting
) {
946 if (workerInfo
.dynamic
) {
947 this.createAndSetupDynamicWorkerNode()
949 this.createAndSetupWorkerNode()
952 if (this.opts
.enableTasksQueue
=== true) {
953 this.redistributeQueuedTasks(workerNodeKey
)
956 worker
.on('exit', this.opts
.exitHandler
?? EMPTY_FUNCTION
)
957 worker
.once('exit', () => {
958 this.removeWorkerNode(worker
)
961 const workerNodeKey
= this.addWorkerNode(worker
)
963 this.afterWorkerNodeSetup(workerNodeKey
)
969 * Creates a new, completely set up dynamic worker node.
971 * @returns New, completely set up dynamic worker node key.
973 protected createAndSetupDynamicWorkerNode (): number {
974 const workerNodeKey
= this.createAndSetupWorkerNode()
975 this.registerWorkerMessageListener(workerNodeKey
, (message
) => {
976 const localWorkerNodeKey
= this.getWorkerNodeKeyByWorkerId(
979 const workerUsage
= this.workerNodes
[localWorkerNodeKey
].usage
980 // Kill message received from worker
982 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
983 (isKillBehavior(KillBehaviors
.SOFT
, message
.kill
) &&
984 ((this.opts
.enableTasksQueue
=== false &&
985 workerUsage
.tasks
.executing
=== 0) ||
986 (this.opts
.enableTasksQueue
=== true &&
987 workerUsage
.tasks
.executing
=== 0 &&
988 this.tasksQueueSize(localWorkerNodeKey
) === 0)))
990 this.destroyWorkerNode(localWorkerNodeKey
).catch((error
) => {
991 this.emitter
?.emit(PoolEvents
.error
, error
)
995 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
996 this.sendToWorker(workerNodeKey
, {
998 workerId
: workerInfo
.id
as number
1000 workerInfo
.dynamic
= true
1001 if (this.workerChoiceStrategyContext
.getStrategyPolicy().useDynamicWorker
) {
1002 workerInfo
.ready
= true
1004 return workerNodeKey
1008 * Registers a listener callback on the worker given its worker node key.
1010 * @param workerNodeKey - The worker node key.
1011 * @param listener - The message listener callback.
1013 protected abstract registerWorkerMessageListener
<
1014 Message
extends Data
| Response
1016 workerNodeKey
: number,
1017 listener
: (message
: MessageValue
<Message
>) => void
1021 * Method hooked up after a worker node has been newly created.
1022 * Can be overridden.
1024 * @param workerNodeKey - The newly created worker node key.
1026 protected afterWorkerNodeSetup (workerNodeKey
: number): void {
1027 // Listen to worker messages.
1028 this.registerWorkerMessageListener(workerNodeKey
, this.workerListener())
1029 // Send the startup message to worker.
1030 this.sendStartupMessageToWorker(workerNodeKey
)
1031 // Send the statistics message to worker.
1032 this.sendStatisticsMessageToWorker(workerNodeKey
)
1036 * Sends the startup message to worker given its worker node key.
1038 * @param workerNodeKey - The worker node key.
1040 protected abstract sendStartupMessageToWorker (workerNodeKey
: number): void
1043 * Sends the statistics message to worker given its worker node key.
1045 * @param workerNodeKey - The worker node key.
1047 private sendStatisticsMessageToWorker (workerNodeKey
: number): void {
1048 this.sendToWorker(workerNodeKey
, {
1051 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1053 elu
: this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1056 workerId
: this.getWorkerInfo(workerNodeKey
).id
as number
1060 private redistributeQueuedTasks (workerNodeKey
: number): void {
1061 while (this.tasksQueueSize(workerNodeKey
) > 0) {
1062 let targetWorkerNodeKey
: number = workerNodeKey
1063 let minQueuedTasks
= Infinity
1064 let executeTask
= false
1065 for (const [workerNodeId
, workerNode
] of this.workerNodes
.entries()) {
1066 const workerInfo
= this.getWorkerInfo(workerNodeId
)
1068 workerNodeId
!== workerNodeKey
&&
1070 workerNode
.usage
.tasks
.queued
=== 0
1073 this.workerNodes
[workerNodeId
].usage
.tasks
.executing
<
1074 (this.opts
.tasksQueueOptions
?.concurrency
as number)
1078 targetWorkerNodeKey
= workerNodeId
1082 workerNodeId
!== workerNodeKey
&&
1084 workerNode
.usage
.tasks
.queued
< minQueuedTasks
1086 minQueuedTasks
= workerNode
.usage
.tasks
.queued
1087 targetWorkerNodeKey
= workerNodeId
1092 targetWorkerNodeKey
,
1093 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1097 targetWorkerNodeKey
,
1098 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1105 * This method is the listener registered for each worker message.
1107 * @returns The listener function to execute when a message is received from a worker.
1109 protected workerListener (): (message
: MessageValue
<Response
>) => void {
1110 return (message
) => {
1111 this.checkMessageWorkerId(message
)
1112 if (message
.ready
!= null) {
1113 // Worker ready response received from worker
1114 this.handleWorkerReadyResponse(message
)
1115 } else if (message
.taskId
!= null) {
1116 // Task execution response received from worker
1117 this.handleTaskExecutionResponse(message
)
1118 } else if (message
.taskFunctions
!= null) {
1119 // Task functions message received from worker
1120 this.taskFunctions
= message
.taskFunctions
1125 private handleWorkerReadyResponse (message
: MessageValue
<Response
>): void {
1127 this.getWorkerNodeKeyByWorkerId(message
.workerId
)
1128 ).ready
= message
.ready
as boolean
1129 if (this.emitter
!= null && this.ready
) {
1130 this.emitter
.emit(PoolEvents
.ready
, this.info
)
1134 private handleTaskExecutionResponse (message
: MessageValue
<Response
>): void {
1135 const promiseResponse
= this.promiseResponseMap
.get(
1136 message
.taskId
as string
1138 if (promiseResponse
!= null) {
1139 if (message
.taskError
!= null) {
1140 this.emitter
?.emit(PoolEvents
.taskError
, message
.taskError
)
1141 promiseResponse
.reject(message
.taskError
.message
)
1143 promiseResponse
.resolve(message
.data
as Response
)
1145 const workerNodeKey
= promiseResponse
.workerNodeKey
1146 this.afterTaskExecutionHook(workerNodeKey
, message
)
1147 this.promiseResponseMap
.delete(message
.taskId
as string)
1149 this.opts
.enableTasksQueue
=== true &&
1150 this.tasksQueueSize(workerNodeKey
) > 0 &&
1151 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
<
1152 (this.opts
.tasksQueueOptions
?.concurrency
as number)
1156 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1159 this.workerChoiceStrategyContext
.update(workerNodeKey
)
1163 private checkAndEmitEvents (): void {
1164 if (this.emitter
!= null) {
1166 this.emitter
.emit(PoolEvents
.busy
, this.info
)
1168 if (this.type === PoolTypes
.dynamic
&& this.full
) {
1169 this.emitter
.emit(PoolEvents
.full
, this.info
)
1175 * Gets the worker information given its worker node key.
1177 * @param workerNodeKey - The worker node key.
1178 * @returns The worker information.
1180 protected getWorkerInfo (workerNodeKey
: number): WorkerInfo
{
1181 return this.workerNodes
[workerNodeKey
].info
1185 * Adds the given worker in the pool worker nodes.
1187 * @param worker - The worker.
1188 * @returns The added worker node key.
1189 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the added worker node is not found.
1191 private addWorkerNode (worker
: Worker
): number {
1192 const workerNode
= new WorkerNode
<Worker
, Data
>(worker
, this.worker
)
1193 // Flag the worker node as ready at pool startup.
1194 if (this.starting
) {
1195 workerNode
.info
.ready
= true
1197 this.workerNodes
.push(workerNode
)
1198 const workerNodeKey
= this.getWorkerNodeKeyByWorker(worker
)
1199 if (workerNodeKey
=== -1) {
1200 throw new Error('Worker node not found')
1202 return workerNodeKey
1206 * Removes the given worker from the pool worker nodes.
1208 * @param worker - The worker.
1210 private removeWorkerNode (worker
: Worker
): void {
1211 const workerNodeKey
= this.getWorkerNodeKeyByWorker(worker
)
1212 if (workerNodeKey
!== -1) {
1213 this.workerNodes
.splice(workerNodeKey
, 1)
1214 this.workerChoiceStrategyContext
.remove(workerNodeKey
)
1219 * Executes the given task on the worker given its worker node key.
1221 * @param workerNodeKey - The worker node key.
1222 * @param task - The task to execute.
1224 private executeTask (workerNodeKey
: number, task
: Task
<Data
>): void {
1225 this.beforeTaskExecutionHook(workerNodeKey
, task
)
1226 this.sendToWorker(workerNodeKey
, task
, task
.transferList
)
1229 private enqueueTask (workerNodeKey
: number, task
: Task
<Data
>): number {
1230 return this.workerNodes
[workerNodeKey
].enqueueTask(task
)
1233 private dequeueTask (workerNodeKey
: number): Task
<Data
> | undefined {
1234 return this.workerNodes
[workerNodeKey
].dequeueTask()
1237 private tasksQueueSize (workerNodeKey
: number): number {
1238 return this.workerNodes
[workerNodeKey
].tasksQueueSize()
1241 protected flushTasksQueue (workerNodeKey
: number): void {
1242 while (this.tasksQueueSize(workerNodeKey
) > 0) {
1245 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1248 this.workerNodes
[workerNodeKey
].clearTasksQueue()
1251 private flushTasksQueues (): void {
1252 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
1253 this.flushTasksQueue(workerNodeKey
)