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
97 * Constructs a new poolifier pool.
99 * @param numberOfWorkers - Number of workers that this pool should manage.
100 * @param filePath - Path to the worker file.
101 * @param opts - Options for the pool.
104 protected readonly numberOfWorkers
: number,
105 protected readonly filePath
: string,
106 protected readonly opts
: PoolOptions
<Worker
>
108 if (!this.isMain()) {
109 throw new Error('Cannot start a pool from a worker!')
111 this.checkNumberOfWorkers(this.numberOfWorkers
)
112 this.checkFilePath(this.filePath
)
113 this.checkPoolOptions(this.opts
)
115 this.chooseWorkerNode
= this.chooseWorkerNode
.bind(this)
116 this.executeTask
= this.executeTask
.bind(this)
117 this.enqueueTask
= this.enqueueTask
.bind(this)
118 this.dequeueTask
= this.dequeueTask
.bind(this)
119 this.checkAndEmitEvents
= this.checkAndEmitEvents
.bind(this)
121 if (this.opts
.enableEvents
=== true) {
122 this.emitter
= new PoolEmitter()
124 this.workerChoiceStrategyContext
= new WorkerChoiceStrategyContext
<
130 this.opts
.workerChoiceStrategy
,
131 this.opts
.workerChoiceStrategyOptions
138 this.starting
= false
140 this.startTimestamp
= performance
.now()
143 private checkFilePath (filePath
: string): void {
146 typeof filePath
!== 'string' ||
147 (typeof filePath
=== 'string' && filePath
.trim().length
=== 0)
149 throw new Error('Please specify a file with a worker implementation')
151 if (!existsSync(filePath
)) {
152 throw new Error(`Cannot find the worker file '${filePath}'`)
156 private checkNumberOfWorkers (numberOfWorkers
: number): void {
157 if (numberOfWorkers
== null) {
159 'Cannot instantiate a pool without specifying the number of workers'
161 } else if (!Number.isSafeInteger(numberOfWorkers
)) {
163 'Cannot instantiate a pool with a non safe integer number of workers'
165 } else if (numberOfWorkers
< 0) {
166 throw new RangeError(
167 'Cannot instantiate a pool with a negative number of workers'
169 } else if (this.type === PoolTypes
.fixed
&& numberOfWorkers
=== 0) {
170 throw new RangeError('Cannot instantiate a fixed pool with zero worker')
174 protected checkDynamicPoolSize (min
: number, max
: number): void {
175 if (this.type === PoolTypes
.dynamic
) {
178 'Cannot instantiate a dynamic pool without specifying the maximum pool size'
180 } else if (!Number.isSafeInteger(max
)) {
182 'Cannot instantiate a dynamic pool with a non safe integer maximum pool size'
184 } else if (min
> max
) {
185 throw new RangeError(
186 'Cannot instantiate a dynamic pool with a maximum pool size inferior to the minimum pool size'
188 } else if (max
=== 0) {
189 throw new RangeError(
190 'Cannot instantiate a dynamic pool with a maximum pool size equal to zero'
192 } else if (min
=== max
) {
193 throw new RangeError(
194 'Cannot instantiate a dynamic pool with a minimum pool size equal to the maximum pool size. Use a fixed pool instead'
200 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
201 if (isPlainObject(opts
)) {
202 this.opts
.workerChoiceStrategy
=
203 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
204 this.checkValidWorkerChoiceStrategy(this.opts
.workerChoiceStrategy
)
205 this.opts
.workerChoiceStrategyOptions
=
206 opts
.workerChoiceStrategyOptions
??
207 DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
208 this.checkValidWorkerChoiceStrategyOptions(
209 this.opts
.workerChoiceStrategyOptions
211 this.opts
.restartWorkerOnError
= opts
.restartWorkerOnError
?? true
212 this.opts
.enableEvents
= opts
.enableEvents
?? true
213 this.opts
.enableTasksQueue
= opts
.enableTasksQueue
?? false
214 if (this.opts
.enableTasksQueue
) {
215 this.checkValidTasksQueueOptions(
216 opts
.tasksQueueOptions
as TasksQueueOptions
218 this.opts
.tasksQueueOptions
= this.buildTasksQueueOptions(
219 opts
.tasksQueueOptions
as TasksQueueOptions
223 throw new TypeError('Invalid pool options: must be a plain object')
227 private checkValidWorkerChoiceStrategy (
228 workerChoiceStrategy
: WorkerChoiceStrategy
230 if (!Object.values(WorkerChoiceStrategies
).includes(workerChoiceStrategy
)) {
232 `Invalid worker choice strategy '${workerChoiceStrategy}'`
237 private checkValidWorkerChoiceStrategyOptions (
238 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
240 if (!isPlainObject(workerChoiceStrategyOptions
)) {
242 'Invalid worker choice strategy options: must be a plain object'
246 workerChoiceStrategyOptions
.weights
!= null &&
247 Object.keys(workerChoiceStrategyOptions
.weights
).length
!== this.maxSize
250 'Invalid worker choice strategy options: must have a weight for each worker node'
254 workerChoiceStrategyOptions
.measurement
!= null &&
255 !Object.values(Measurements
).includes(
256 workerChoiceStrategyOptions
.measurement
260 `Invalid worker choice strategy options: invalid measurement '${workerChoiceStrategyOptions.measurement}'`
265 private checkValidTasksQueueOptions (
266 tasksQueueOptions
: TasksQueueOptions
268 if (tasksQueueOptions
!= null && !isPlainObject(tasksQueueOptions
)) {
269 throw new TypeError('Invalid tasks queue options: must be a plain object')
272 tasksQueueOptions
?.concurrency
!= null &&
273 !Number.isSafeInteger(tasksQueueOptions
.concurrency
)
276 'Invalid worker tasks concurrency: must be an integer'
280 tasksQueueOptions
?.concurrency
!= null &&
281 tasksQueueOptions
.concurrency
<= 0
284 `Invalid worker tasks concurrency '${tasksQueueOptions.concurrency}'`
289 private startPool (): void {
291 this.workerNodes
.reduce(
292 (accumulator
, workerNode
) =>
293 !workerNode
.info
.dynamic
? accumulator
+ 1 : accumulator
,
295 ) < this.numberOfWorkers
297 this.createAndSetupWorkerNode()
302 public get
info (): PoolInfo
{
308 strategy
: this.opts
.workerChoiceStrategy
as WorkerChoiceStrategy
,
309 minSize
: this.minSize
,
310 maxSize
: this.maxSize
,
311 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
312 .runTime
.aggregate
&&
313 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
314 .waitTime
.aggregate
&& { utilization
: round(this.utilization
) }),
315 workerNodes
: this.workerNodes
.length
,
316 idleWorkerNodes
: this.workerNodes
.reduce(
317 (accumulator
, workerNode
) =>
318 workerNode
.usage
.tasks
.executing
=== 0
323 busyWorkerNodes
: this.workerNodes
.reduce(
324 (accumulator
, workerNode
) =>
325 workerNode
.usage
.tasks
.executing
> 0 ? accumulator
+ 1 : accumulator
,
328 executedTasks
: this.workerNodes
.reduce(
329 (accumulator
, workerNode
) =>
330 accumulator
+ workerNode
.usage
.tasks
.executed
,
333 executingTasks
: this.workerNodes
.reduce(
334 (accumulator
, workerNode
) =>
335 accumulator
+ workerNode
.usage
.tasks
.executing
,
338 ...(this.opts
.enableTasksQueue
=== true && {
339 queuedTasks
: this.workerNodes
.reduce(
340 (accumulator
, workerNode
) =>
341 accumulator
+ workerNode
.usage
.tasks
.queued
,
345 ...(this.opts
.enableTasksQueue
=== true && {
346 maxQueuedTasks
: this.workerNodes
.reduce(
347 (accumulator
, workerNode
) =>
348 accumulator
+ (workerNode
.usage
.tasks
?.maxQueued
?? 0),
352 failedTasks
: this.workerNodes
.reduce(
353 (accumulator
, workerNode
) =>
354 accumulator
+ workerNode
.usage
.tasks
.failed
,
357 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
358 .runTime
.aggregate
&& {
362 ...this.workerNodes
.map(
363 (workerNode
) => workerNode
.usage
.runTime
?.minimum
?? Infinity
369 ...this.workerNodes
.map(
370 (workerNode
) => workerNode
.usage
.runTime
?.maximum
?? -Infinity
375 this.workerNodes
.reduce(
376 (accumulator
, workerNode
) =>
377 accumulator
+ (workerNode
.usage
.runTime
?.aggregate
?? 0),
380 this.workerNodes
.reduce(
381 (accumulator
, workerNode
) =>
382 accumulator
+ (workerNode
.usage
.tasks
?.executed
?? 0),
386 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
390 this.workerNodes
.map(
391 (workerNode
) => workerNode
.usage
.runTime
?.median
?? 0
398 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
399 .waitTime
.aggregate
&& {
403 ...this.workerNodes
.map(
404 (workerNode
) => workerNode
.usage
.waitTime
?.minimum
?? Infinity
410 ...this.workerNodes
.map(
411 (workerNode
) => workerNode
.usage
.waitTime
?.maximum
?? -Infinity
416 this.workerNodes
.reduce(
417 (accumulator
, workerNode
) =>
418 accumulator
+ (workerNode
.usage
.waitTime
?.aggregate
?? 0),
421 this.workerNodes
.reduce(
422 (accumulator
, workerNode
) =>
423 accumulator
+ (workerNode
.usage
.tasks
?.executed
?? 0),
427 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
428 .waitTime
.median
&& {
431 this.workerNodes
.map(
432 (workerNode
) => workerNode
.usage
.waitTime
?.median
?? 0
443 * The pool readiness boolean status.
445 private get
ready (): boolean {
447 this.workerNodes
.reduce(
448 (accumulator
, workerNode
) =>
449 !workerNode
.info
.dynamic
&& workerNode
.info
.ready
458 * The approximate pool utilization.
460 * @returns The pool utilization.
462 private get
utilization (): number {
463 const poolTimeCapacity
=
464 (performance
.now() - this.startTimestamp
) * this.maxSize
465 const totalTasksRunTime
= this.workerNodes
.reduce(
466 (accumulator
, workerNode
) =>
467 accumulator
+ (workerNode
.usage
.runTime
?.aggregate
?? 0),
470 const totalTasksWaitTime
= this.workerNodes
.reduce(
471 (accumulator
, workerNode
) =>
472 accumulator
+ (workerNode
.usage
.waitTime
?.aggregate
?? 0),
475 return (totalTasksRunTime
+ totalTasksWaitTime
) / poolTimeCapacity
481 * If it is `'dynamic'`, it provides the `max` property.
483 protected abstract get
type (): PoolType
488 protected abstract get
worker (): WorkerType
491 * The pool minimum size.
493 protected abstract get
minSize (): number
496 * The pool maximum size.
498 protected abstract get
maxSize (): number
501 * Checks if the worker id sent in the received message from a worker is valid.
503 * @param message - The received message.
504 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the worker id is invalid.
506 private checkMessageWorkerId (message
: MessageValue
<Response
>): void {
508 message
.workerId
!= null &&
509 this.getWorkerNodeKeyByWorkerId(message
.workerId
) === -1
512 `Worker message received from unknown worker '${message.workerId}'`
518 * Gets the given worker its worker node key.
520 * @param worker - The worker.
521 * @returns The worker node key if found in the pool worker nodes, `-1` otherwise.
523 private getWorkerNodeKeyByWorker (worker
: Worker
): number {
524 return this.workerNodes
.findIndex(
525 (workerNode
) => workerNode
.worker
=== worker
530 * Gets the worker node key given its worker id.
532 * @param workerId - The worker id.
533 * @returns The worker node key if the worker id is found in the pool worker nodes, `-1` otherwise.
535 private getWorkerNodeKeyByWorkerId (workerId
: number): number {
536 return this.workerNodes
.findIndex(
537 (workerNode
) => workerNode
.info
.id
=== workerId
542 public setWorkerChoiceStrategy (
543 workerChoiceStrategy
: WorkerChoiceStrategy
,
544 workerChoiceStrategyOptions
?: WorkerChoiceStrategyOptions
546 this.checkValidWorkerChoiceStrategy(workerChoiceStrategy
)
547 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
548 this.workerChoiceStrategyContext
.setWorkerChoiceStrategy(
549 this.opts
.workerChoiceStrategy
551 if (workerChoiceStrategyOptions
!= null) {
552 this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
554 for (const [workerNodeKey
, workerNode
] of this.workerNodes
.entries()) {
555 workerNode
.resetUsage()
556 this.sendWorkerStatisticsMessageToWorker(workerNodeKey
)
561 public setWorkerChoiceStrategyOptions (
562 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
564 this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
565 this.opts
.workerChoiceStrategyOptions
= workerChoiceStrategyOptions
566 this.workerChoiceStrategyContext
.setOptions(
567 this.opts
.workerChoiceStrategyOptions
572 public enableTasksQueue (
574 tasksQueueOptions
?: TasksQueueOptions
576 if (this.opts
.enableTasksQueue
=== true && !enable
) {
577 this.flushTasksQueues()
579 this.opts
.enableTasksQueue
= enable
580 this.setTasksQueueOptions(tasksQueueOptions
as TasksQueueOptions
)
584 public setTasksQueueOptions (tasksQueueOptions
: TasksQueueOptions
): void {
585 if (this.opts
.enableTasksQueue
=== true) {
586 this.checkValidTasksQueueOptions(tasksQueueOptions
)
587 this.opts
.tasksQueueOptions
=
588 this.buildTasksQueueOptions(tasksQueueOptions
)
589 } else if (this.opts
.tasksQueueOptions
!= null) {
590 delete this.opts
.tasksQueueOptions
594 private buildTasksQueueOptions (
595 tasksQueueOptions
: TasksQueueOptions
596 ): TasksQueueOptions
{
598 concurrency
: tasksQueueOptions
?.concurrency
?? 1
603 * Whether the pool is full or not.
605 * The pool filling boolean status.
607 protected get
full (): boolean {
608 return this.workerNodes
.length
>= this.maxSize
612 * Whether the pool is busy or not.
614 * The pool busyness boolean status.
616 protected abstract get
busy (): boolean
619 * Whether worker nodes are executing concurrently their tasks quota or not.
621 * @returns Worker nodes busyness boolean status.
623 protected internalBusy (): boolean {
624 if (this.opts
.enableTasksQueue
=== true) {
626 this.workerNodes
.findIndex(
628 workerNode
.info
.ready
&&
629 workerNode
.usage
.tasks
.executing
<
630 (this.opts
.tasksQueueOptions
?.concurrency
as number)
635 this.workerNodes
.findIndex(
637 workerNode
.info
.ready
&& workerNode
.usage
.tasks
.executing
=== 0
644 public async execute (
647 transferList
?: TransferListItem
[]
648 ): Promise
<Response
> {
649 return await new Promise
<Response
>((resolve
, reject
) => {
650 if (name
!= null && typeof name
!== 'string') {
651 reject(new TypeError('name argument must be a string'))
653 if (transferList
!= null && !Array.isArray(transferList
)) {
654 reject(new TypeError('transferList argument must be an array'))
656 const timestamp
= performance
.now()
657 const workerNodeKey
= this.chooseWorkerNode()
658 const task
: Task
<Data
> = {
659 name
: name
?? DEFAULT_TASK_NAME
,
660 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
661 data
: data
?? ({} as Data
),
664 workerId
: this.getWorkerInfo(workerNodeKey
).id
as number,
667 this.promiseResponseMap
.set(task
.taskId
as string, {
673 this.opts
.enableTasksQueue
=== false ||
674 (this.opts
.enableTasksQueue
=== true &&
675 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
<
676 (this.opts
.tasksQueueOptions
?.concurrency
as number))
678 this.executeTask(workerNodeKey
, task
)
680 this.enqueueTask(workerNodeKey
, task
)
682 this.checkAndEmitEvents()
687 public async destroy (): Promise
<void> {
689 this.workerNodes
.map(async (_
, workerNodeKey
) => {
690 await this.destroyWorkerNode(workerNodeKey
)
696 * Terminates the worker node given its worker node key.
698 * @param workerNodeKey - The worker node key.
700 protected abstract destroyWorkerNode (workerNodeKey
: number): Promise
<void>
703 * Setup hook to execute code before worker nodes are created in the abstract constructor.
708 protected setupHook (): void {
709 // Intentionally empty
713 * Should return whether the worker is the main worker or not.
715 protected abstract isMain (): boolean
718 * Hook executed before the worker task execution.
721 * @param workerNodeKey - The worker node key.
722 * @param task - The task to execute.
724 protected beforeTaskExecutionHook (
725 workerNodeKey
: number,
728 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
729 ++workerUsage
.tasks
.executing
730 this.updateWaitTimeWorkerUsage(workerUsage
, task
)
731 const taskWorkerUsage
= this.workerNodes
[workerNodeKey
].getTaskWorkerUsage(
734 ++taskWorkerUsage
.tasks
.executing
735 this.updateWaitTimeWorkerUsage(taskWorkerUsage
, task
)
739 * Hook executed after the worker task execution.
742 * @param workerNodeKey - The worker node key.
743 * @param message - The received message.
745 protected afterTaskExecutionHook (
746 workerNodeKey
: number,
747 message
: MessageValue
<Response
>
749 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
750 this.updateTaskStatisticsWorkerUsage(workerUsage
, message
)
751 this.updateRunTimeWorkerUsage(workerUsage
, message
)
752 this.updateEluWorkerUsage(workerUsage
, message
)
753 const taskWorkerUsage
= this.workerNodes
[workerNodeKey
].getTaskWorkerUsage(
754 message
.taskPerformance
?.name
?? DEFAULT_TASK_NAME
756 this.updateTaskStatisticsWorkerUsage(taskWorkerUsage
, message
)
757 this.updateRunTimeWorkerUsage(taskWorkerUsage
, message
)
758 this.updateEluWorkerUsage(taskWorkerUsage
, message
)
761 private updateTaskStatisticsWorkerUsage (
762 workerUsage
: WorkerUsage
,
763 message
: MessageValue
<Response
>
765 const workerTaskStatistics
= workerUsage
.tasks
766 --workerTaskStatistics
.executing
767 if (message
.taskError
== null) {
768 ++workerTaskStatistics
.executed
770 ++workerTaskStatistics
.failed
774 private updateRunTimeWorkerUsage (
775 workerUsage
: WorkerUsage
,
776 message
: MessageValue
<Response
>
778 updateMeasurementStatistics(
780 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
,
781 message
.taskPerformance
?.runTime
?? 0,
782 workerUsage
.tasks
.executed
786 private updateWaitTimeWorkerUsage (
787 workerUsage
: WorkerUsage
,
790 const timestamp
= performance
.now()
791 const taskWaitTime
= timestamp
- (task
.timestamp
?? timestamp
)
792 updateMeasurementStatistics(
793 workerUsage
.waitTime
,
794 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().waitTime
,
796 workerUsage
.tasks
.executed
800 private updateEluWorkerUsage (
801 workerUsage
: WorkerUsage
,
802 message
: MessageValue
<Response
>
804 const eluTaskStatisticsRequirements
: MeasurementStatisticsRequirements
=
805 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
806 updateMeasurementStatistics(
807 workerUsage
.elu
.active
,
808 eluTaskStatisticsRequirements
,
809 message
.taskPerformance
?.elu
?.active
?? 0,
810 workerUsage
.tasks
.executed
812 updateMeasurementStatistics(
813 workerUsage
.elu
.idle
,
814 eluTaskStatisticsRequirements
,
815 message
.taskPerformance
?.elu
?.idle
?? 0,
816 workerUsage
.tasks
.executed
818 if (eluTaskStatisticsRequirements
.aggregate
) {
819 if (message
.taskPerformance
?.elu
!= null) {
820 if (workerUsage
.elu
.utilization
!= null) {
821 workerUsage
.elu
.utilization
=
822 (workerUsage
.elu
.utilization
+
823 message
.taskPerformance
.elu
.utilization
) /
826 workerUsage
.elu
.utilization
= message
.taskPerformance
.elu
.utilization
833 * Chooses a worker node for the next task.
835 * The default worker choice strategy uses a round robin algorithm to distribute the tasks.
837 * @returns The chosen worker node key
839 private chooseWorkerNode (): number {
840 if (this.shallCreateDynamicWorker()) {
841 const workerNodeKey
= this.createAndSetupDynamicWorkerNode()
843 this.workerChoiceStrategyContext
.getStrategyPolicy().useDynamicWorker
848 return this.workerChoiceStrategyContext
.execute()
852 * Conditions for dynamic worker creation.
854 * @returns Whether to create a dynamic worker or not.
856 private shallCreateDynamicWorker (): boolean {
857 return this.type === PoolTypes
.dynamic
&& !this.full
&& this.internalBusy()
861 * Sends a message to worker given its worker node key.
863 * @param workerNodeKey - The worker node key.
864 * @param message - The message.
865 * @param transferList - The optional array of transferable objects.
867 protected abstract sendToWorker (
868 workerNodeKey
: number,
869 message
: MessageValue
<Data
>,
870 transferList
?: TransferListItem
[]
874 * Creates a new worker.
876 * @returns Newly created worker.
878 protected abstract createWorker (): Worker
881 * Creates a new, completely set up worker node.
883 * @returns New, completely set up worker node key.
885 protected createAndSetupWorkerNode (): number {
886 const worker
= this.createWorker()
888 worker
.on('message', this.opts
.messageHandler
?? EMPTY_FUNCTION
)
889 worker
.on('error', this.opts
.errorHandler
?? EMPTY_FUNCTION
)
890 worker
.on('error', (error
) => {
891 const workerNodeKey
= this.getWorkerNodeKeyByWorker(worker
)
892 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
893 workerInfo
.ready
= false
894 this.workerNodes
[workerNodeKey
].closeChannel()
895 this.emitter
?.emit(PoolEvents
.error
, error
)
896 if (this.opts
.restartWorkerOnError
=== true && !this.starting
) {
897 if (workerInfo
.dynamic
) {
898 this.createAndSetupDynamicWorkerNode()
900 this.createAndSetupWorkerNode()
903 if (this.opts
.enableTasksQueue
=== true) {
904 this.redistributeQueuedTasks(workerNodeKey
)
907 worker
.on('online', this.opts
.onlineHandler
?? EMPTY_FUNCTION
)
908 worker
.on('exit', this.opts
.exitHandler
?? EMPTY_FUNCTION
)
909 worker
.once('exit', () => {
910 this.removeWorkerNode(worker
)
913 const workerNodeKey
= this.addWorkerNode(worker
)
915 this.afterWorkerNodeSetup(workerNodeKey
)
921 * Creates a new, completely set up dynamic worker node.
923 * @returns New, completely set up dynamic worker node key.
925 protected createAndSetupDynamicWorkerNode (): number {
926 const workerNodeKey
= this.createAndSetupWorkerNode()
927 this.registerWorkerMessageListener(workerNodeKey
, (message
) => {
928 const localWorkerNodeKey
= this.getWorkerNodeKeyByWorkerId(
931 const workerUsage
= this.workerNodes
[localWorkerNodeKey
].usage
932 // Kill message received from worker
934 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
935 (message
.kill
!= null &&
936 ((this.opts
.enableTasksQueue
=== false &&
937 workerUsage
.tasks
.executing
=== 0) ||
938 (this.opts
.enableTasksQueue
=== true &&
939 workerUsage
.tasks
.executing
=== 0 &&
940 this.tasksQueueSize(localWorkerNodeKey
) === 0)))
942 this.destroyWorkerNode(localWorkerNodeKey
).catch(EMPTY_FUNCTION
)
945 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
946 this.sendToWorker(workerNodeKey
, {
948 workerId
: workerInfo
.id
as number
950 workerInfo
.dynamic
= true
951 if (this.workerChoiceStrategyContext
.getStrategyPolicy().useDynamicWorker
) {
952 workerInfo
.ready
= true
958 * Registers a listener callback on the worker given its worker node key.
960 * @param workerNodeKey - The worker node key.
961 * @param listener - The message listener callback.
963 protected abstract registerWorkerMessageListener
<
964 Message
extends Data
| Response
966 workerNodeKey
: number,
967 listener
: (message
: MessageValue
<Message
>) => void
971 * Method hooked up after a worker node has been newly created.
974 * @param workerNodeKey - The newly created worker node key.
976 protected afterWorkerNodeSetup (workerNodeKey
: number): void {
977 // Listen to worker messages.
978 this.registerWorkerMessageListener(workerNodeKey
, this.workerListener())
979 // Send the startup message to worker.
980 this.sendStartupMessageToWorker(workerNodeKey
)
981 // Send the worker statistics message to worker.
982 this.sendWorkerStatisticsMessageToWorker(workerNodeKey
)
986 * Sends the startup message to worker given its worker node key.
988 * @param workerNodeKey - The worker node key.
990 protected abstract sendStartupMessageToWorker (workerNodeKey
: number): void
993 * Sends the worker statistics message to worker given its worker node key.
995 * @param workerNodeKey - The worker node key.
997 private sendWorkerStatisticsMessageToWorker (workerNodeKey
: number): void {
998 this.sendToWorker(workerNodeKey
, {
1001 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1003 elu
: this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1006 workerId
: this.getWorkerInfo(workerNodeKey
).id
as number
1010 private redistributeQueuedTasks (workerNodeKey
: number): void {
1011 while (this.tasksQueueSize(workerNodeKey
) > 0) {
1012 let targetWorkerNodeKey
: number = workerNodeKey
1013 let minQueuedTasks
= Infinity
1014 let executeTask
= false
1015 for (const [workerNodeId
, workerNode
] of this.workerNodes
.entries()) {
1016 const workerInfo
= this.getWorkerInfo(workerNodeId
)
1018 workerNodeId
!== workerNodeKey
&&
1020 workerNode
.usage
.tasks
.queued
=== 0
1023 this.workerNodes
[workerNodeId
].usage
.tasks
.executing
<
1024 (this.opts
.tasksQueueOptions
?.concurrency
as number)
1028 targetWorkerNodeKey
= workerNodeId
1032 workerNodeId
!== workerNodeKey
&&
1034 workerNode
.usage
.tasks
.queued
< minQueuedTasks
1036 minQueuedTasks
= workerNode
.usage
.tasks
.queued
1037 targetWorkerNodeKey
= workerNodeId
1042 targetWorkerNodeKey
,
1043 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1047 targetWorkerNodeKey
,
1048 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1055 * This method is the listener registered for each worker message.
1057 * @returns The listener function to execute when a message is received from a worker.
1059 protected workerListener (): (message
: MessageValue
<Response
>) => void {
1060 return (message
) => {
1061 this.checkMessageWorkerId(message
)
1062 if (message
.ready
!= null) {
1063 // Worker ready response received from worker
1064 this.handleWorkerReadyResponse(message
)
1065 } else if (message
.taskId
!= null) {
1066 // Task execution response received from worker
1067 this.handleTaskExecutionResponse(message
)
1072 private handleWorkerReadyResponse (message
: MessageValue
<Response
>): void {
1074 this.getWorkerNodeKeyByWorkerId(message
.workerId
)
1075 ).ready
= message
.ready
as boolean
1076 if (this.emitter
!= null && this.ready
) {
1077 this.emitter
.emit(PoolEvents
.ready
, this.info
)
1081 private handleTaskExecutionResponse (message
: MessageValue
<Response
>): void {
1082 const promiseResponse
= this.promiseResponseMap
.get(
1083 message
.taskId
as string
1085 if (promiseResponse
!= null) {
1086 if (message
.taskError
!= null) {
1087 this.emitter
?.emit(PoolEvents
.taskError
, message
.taskError
)
1088 promiseResponse
.reject(message
.taskError
.message
)
1090 promiseResponse
.resolve(message
.data
as Response
)
1092 const workerNodeKey
= promiseResponse
.workerNodeKey
1093 this.afterTaskExecutionHook(workerNodeKey
, message
)
1094 this.promiseResponseMap
.delete(message
.taskId
as string)
1096 this.opts
.enableTasksQueue
=== true &&
1097 this.tasksQueueSize(workerNodeKey
) > 0 &&
1098 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
<
1099 (this.opts
.tasksQueueOptions
?.concurrency
as number)
1103 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1106 this.workerChoiceStrategyContext
.update(workerNodeKey
)
1110 private checkAndEmitEvents (): void {
1111 if (this.emitter
!= null) {
1113 this.emitter
.emit(PoolEvents
.busy
, this.info
)
1115 if (this.type === PoolTypes
.dynamic
&& this.full
) {
1116 this.emitter
.emit(PoolEvents
.full
, this.info
)
1122 * Gets the worker information given its worker node key.
1124 * @param workerNodeKey - The worker node key.
1125 * @returns The worker information.
1127 protected getWorkerInfo (workerNodeKey
: number): WorkerInfo
{
1128 return this.workerNodes
[workerNodeKey
].info
1132 * Adds the given worker in the pool worker nodes.
1134 * @param worker - The worker.
1135 * @returns The added worker node key.
1136 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the added worker node is not found.
1138 private addWorkerNode (worker
: Worker
): number {
1139 const workerNode
= new WorkerNode
<Worker
, Data
>(worker
, this.worker
)
1140 // Flag the worker node as ready at pool startup.
1141 if (this.starting
) {
1142 workerNode
.info
.ready
= true
1144 this.workerNodes
.push(workerNode
)
1145 const workerNodeKey
= this.getWorkerNodeKeyByWorker(worker
)
1146 if (workerNodeKey
=== -1) {
1147 throw new Error('Worker node not found')
1149 return workerNodeKey
1153 * Removes the given worker from the pool worker nodes.
1155 * @param worker - The worker.
1157 private removeWorkerNode (worker
: Worker
): void {
1158 const workerNodeKey
= this.getWorkerNodeKeyByWorker(worker
)
1159 if (workerNodeKey
!== -1) {
1160 this.workerNodes
.splice(workerNodeKey
, 1)
1161 this.workerChoiceStrategyContext
.remove(workerNodeKey
)
1166 * Executes the given task on the worker given its worker node key.
1168 * @param workerNodeKey - The worker node key.
1169 * @param task - The task to execute.
1171 private executeTask (workerNodeKey
: number, task
: Task
<Data
>): void {
1172 this.beforeTaskExecutionHook(workerNodeKey
, task
)
1173 this.sendToWorker(workerNodeKey
, task
, task
.transferList
)
1176 private enqueueTask (workerNodeKey
: number, task
: Task
<Data
>): number {
1177 return this.workerNodes
[workerNodeKey
].enqueueTask(task
)
1180 private dequeueTask (workerNodeKey
: number): Task
<Data
> | undefined {
1181 return this.workerNodes
[workerNodeKey
].dequeueTask()
1184 private tasksQueueSize (workerNodeKey
: number): number {
1185 return this.workerNodes
[workerNodeKey
].tasksQueueSize()
1188 protected flushTasksQueue (workerNodeKey
: number): void {
1189 while (this.tasksQueueSize(workerNodeKey
) > 0) {
1192 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1195 this.workerNodes
[workerNodeKey
].clearTasksQueue()
1198 private flushTasksQueues (): void {
1199 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
1200 this.flushTasksQueue(workerNodeKey
)