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
,
21 updateMeasurementStatistics
23 import { KillBehaviors
} from
'../worker/worker-options'
24 import type { TaskFunction
} from
'../worker/task-functions'
33 type TasksQueueOptions
43 type MeasurementStatisticsRequirements
,
45 WorkerChoiceStrategies
,
46 type WorkerChoiceStrategy
,
47 type WorkerChoiceStrategyOptions
48 } from
'./selection-strategies/selection-strategies-types'
49 import { WorkerChoiceStrategyContext
} from
'./selection-strategies/worker-choice-strategy-context'
50 import { version
} from
'./version'
51 import { WorkerNode
} from
'./worker-node'
54 * Base class that implements some shared logic for all poolifier pools.
56 * @typeParam Worker - Type of worker which manages this pool.
57 * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
58 * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
60 export abstract class AbstractPool
<
61 Worker
extends IWorker
,
64 > implements IPool
<Worker
, Data
, Response
> {
66 public readonly workerNodes
: Array<IWorkerNode
<Worker
, Data
>> = []
69 public readonly emitter
?: PoolEmitter
72 * The task execution response promise map.
74 * - `key`: The message id of each submitted task.
75 * - `value`: An object that contains the worker, the execution response promise resolve and reject callbacks.
77 * When we receive a message from the worker, we get a map entry with the promise resolve/reject bound to the message id.
79 protected promiseResponseMap
: Map
<string, PromiseResponseWrapper
<Response
>> =
80 new Map
<string, PromiseResponseWrapper
<Response
>>()
83 * Worker choice strategy context referencing a worker choice algorithm implementation.
85 protected workerChoiceStrategyContext
: WorkerChoiceStrategyContext
<
92 * Dynamic pool maximum size property placeholder.
94 protected readonly max
?: number
97 * Whether the pool is starting or not.
99 private readonly starting
: boolean
101 * Whether the pool is started or not.
103 private started
: boolean
105 * The start timestamp of the pool.
107 private readonly startTimestamp
110 * Constructs a new poolifier pool.
112 * @param numberOfWorkers - Number of workers that this pool should manage.
113 * @param filePath - Path to the worker file.
114 * @param opts - Options for the pool.
117 protected readonly numberOfWorkers
: number,
118 protected readonly filePath
: string,
119 protected readonly opts
: PoolOptions
<Worker
>
121 if (!this.isMain()) {
123 'Cannot start a pool from a worker with the same type as the pool'
126 this.checkNumberOfWorkers(this.numberOfWorkers
)
127 this.checkFilePath(this.filePath
)
128 this.checkPoolOptions(this.opts
)
130 this.chooseWorkerNode
= this.chooseWorkerNode
.bind(this)
131 this.executeTask
= this.executeTask
.bind(this)
132 this.enqueueTask
= this.enqueueTask
.bind(this)
134 if (this.opts
.enableEvents
=== true) {
135 this.emitter
= new PoolEmitter()
137 this.workerChoiceStrategyContext
= new WorkerChoiceStrategyContext
<
143 this.opts
.workerChoiceStrategy
,
144 this.opts
.workerChoiceStrategyOptions
151 this.starting
= false
154 this.startTimestamp
= performance
.now()
157 private checkFilePath (filePath
: string): void {
160 typeof filePath
!== 'string' ||
161 (typeof filePath
=== 'string' && filePath
.trim().length
=== 0)
163 throw new Error('Please specify a file with a worker implementation')
165 if (!existsSync(filePath
)) {
166 throw new Error(`Cannot find the worker file '${filePath}'`)
170 private checkNumberOfWorkers (numberOfWorkers
: number): void {
171 if (numberOfWorkers
== null) {
173 'Cannot instantiate a pool without specifying the number of workers'
175 } else if (!Number.isSafeInteger(numberOfWorkers
)) {
177 'Cannot instantiate a pool with a non safe integer number of workers'
179 } else if (numberOfWorkers
< 0) {
180 throw new RangeError(
181 'Cannot instantiate a pool with a negative number of workers'
183 } else if (this.type === PoolTypes
.fixed
&& numberOfWorkers
=== 0) {
184 throw new RangeError('Cannot instantiate a fixed pool with zero worker')
188 protected checkDynamicPoolSize (min
: number, max
: number): void {
189 if (this.type === PoolTypes
.dynamic
) {
192 'Cannot instantiate a dynamic pool without specifying the maximum pool size'
194 } else if (!Number.isSafeInteger(max
)) {
196 'Cannot instantiate a dynamic pool with a non safe integer maximum pool size'
198 } else if (min
> max
) {
199 throw new RangeError(
200 'Cannot instantiate a dynamic pool with a maximum pool size inferior to the minimum pool size'
202 } else if (max
=== 0) {
203 throw new RangeError(
204 'Cannot instantiate a dynamic pool with a maximum pool size equal to zero'
206 } else if (min
=== max
) {
207 throw new RangeError(
208 'Cannot instantiate a dynamic pool with a minimum pool size equal to the maximum pool size. Use a fixed pool instead'
214 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
215 if (isPlainObject(opts
)) {
216 this.opts
.workerChoiceStrategy
=
217 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
218 this.checkValidWorkerChoiceStrategy(this.opts
.workerChoiceStrategy
)
219 this.opts
.workerChoiceStrategyOptions
= {
220 ...DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
,
221 ...opts
.workerChoiceStrategyOptions
223 this.checkValidWorkerChoiceStrategyOptions(
224 this.opts
.workerChoiceStrategyOptions
226 this.opts
.restartWorkerOnError
= opts
.restartWorkerOnError
?? true
227 this.opts
.enableEvents
= opts
.enableEvents
?? true
228 this.opts
.enableTasksQueue
= opts
.enableTasksQueue
?? false
229 if (this.opts
.enableTasksQueue
) {
230 this.checkValidTasksQueueOptions(
231 opts
.tasksQueueOptions
as TasksQueueOptions
233 this.opts
.tasksQueueOptions
= this.buildTasksQueueOptions(
234 opts
.tasksQueueOptions
as TasksQueueOptions
238 throw new TypeError('Invalid pool options: must be a plain object')
242 private checkValidWorkerChoiceStrategy (
243 workerChoiceStrategy
: WorkerChoiceStrategy
245 if (!Object.values(WorkerChoiceStrategies
).includes(workerChoiceStrategy
)) {
247 `Invalid worker choice strategy '${workerChoiceStrategy}'`
252 private checkValidWorkerChoiceStrategyOptions (
253 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
255 if (!isPlainObject(workerChoiceStrategyOptions
)) {
257 'Invalid worker choice strategy options: must be a plain object'
261 workerChoiceStrategyOptions
.retries
!= null &&
262 !Number.isSafeInteger(workerChoiceStrategyOptions
.retries
)
265 'Invalid worker choice strategy options: retries must be an integer'
269 workerChoiceStrategyOptions
.retries
!= null &&
270 workerChoiceStrategyOptions
.retries
< 0
272 throw new RangeError(
273 `Invalid worker choice strategy options: retries '${workerChoiceStrategyOptions.retries}' must be greater or equal than zero`
277 workerChoiceStrategyOptions
.weights
!= null &&
278 Object.keys(workerChoiceStrategyOptions
.weights
).length
!== this.maxSize
281 'Invalid worker choice strategy options: must have a weight for each worker node'
285 workerChoiceStrategyOptions
.measurement
!= null &&
286 !Object.values(Measurements
).includes(
287 workerChoiceStrategyOptions
.measurement
291 `Invalid worker choice strategy options: invalid measurement '${workerChoiceStrategyOptions.measurement}'`
296 private checkValidTasksQueueOptions (
297 tasksQueueOptions
: TasksQueueOptions
299 if (tasksQueueOptions
!= null && !isPlainObject(tasksQueueOptions
)) {
300 throw new TypeError('Invalid tasks queue options: must be a plain object')
303 tasksQueueOptions
?.concurrency
!= null &&
304 !Number.isSafeInteger(tasksQueueOptions
?.concurrency
)
307 'Invalid worker node tasks concurrency: must be an integer'
311 tasksQueueOptions
?.concurrency
!= null &&
312 tasksQueueOptions
?.concurrency
<= 0
314 throw new RangeError(
315 `Invalid worker node tasks concurrency: ${tasksQueueOptions?.concurrency} is a negative integer or zero`
318 if (tasksQueueOptions
?.queueMaxSize
!= null) {
320 'Invalid tasks queue options: queueMaxSize is deprecated, please use size instead'
324 tasksQueueOptions
?.size
!= null &&
325 !Number.isSafeInteger(tasksQueueOptions
?.size
)
328 'Invalid worker node tasks queue size: must be an integer'
331 if (tasksQueueOptions
?.size
!= null && tasksQueueOptions
?.size
<= 0) {
332 throw new RangeError(
333 `Invalid worker node tasks queue size: ${tasksQueueOptions?.size} is a negative integer or zero`
338 private startPool (): void {
340 this.workerNodes
.reduce(
341 (accumulator
, workerNode
) =>
342 !workerNode
.info
.dynamic
? accumulator
+ 1 : accumulator
,
344 ) < this.numberOfWorkers
346 this.createAndSetupWorkerNode()
351 public get
info (): PoolInfo
{
357 strategy
: this.opts
.workerChoiceStrategy
as WorkerChoiceStrategy
,
358 minSize
: this.minSize
,
359 maxSize
: this.maxSize
,
360 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
361 .runTime
.aggregate
&&
362 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
363 .waitTime
.aggregate
&& { utilization
: round(this.utilization
) }),
364 workerNodes
: this.workerNodes
.length
,
365 idleWorkerNodes
: this.workerNodes
.reduce(
366 (accumulator
, workerNode
) =>
367 workerNode
.usage
.tasks
.executing
=== 0
372 busyWorkerNodes
: this.workerNodes
.reduce(
373 (accumulator
, workerNode
) =>
374 workerNode
.usage
.tasks
.executing
> 0 ? accumulator
+ 1 : accumulator
,
377 executedTasks
: this.workerNodes
.reduce(
378 (accumulator
, workerNode
) =>
379 accumulator
+ workerNode
.usage
.tasks
.executed
,
382 executingTasks
: this.workerNodes
.reduce(
383 (accumulator
, workerNode
) =>
384 accumulator
+ workerNode
.usage
.tasks
.executing
,
387 ...(this.opts
.enableTasksQueue
=== true && {
388 queuedTasks
: this.workerNodes
.reduce(
389 (accumulator
, workerNode
) =>
390 accumulator
+ workerNode
.usage
.tasks
.queued
,
394 ...(this.opts
.enableTasksQueue
=== true && {
395 maxQueuedTasks
: this.workerNodes
.reduce(
396 (accumulator
, workerNode
) =>
397 accumulator
+ (workerNode
.usage
.tasks
?.maxQueued
?? 0),
401 ...(this.opts
.enableTasksQueue
=== true && {
402 backPressure
: this.hasBackPressure()
404 ...(this.opts
.enableTasksQueue
=== true && {
405 stolenTasks
: this.workerNodes
.reduce(
406 (accumulator
, workerNode
) =>
407 accumulator
+ workerNode
.usage
.tasks
.stolen
,
411 failedTasks
: this.workerNodes
.reduce(
412 (accumulator
, workerNode
) =>
413 accumulator
+ workerNode
.usage
.tasks
.failed
,
416 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
417 .runTime
.aggregate
&& {
421 ...this.workerNodes
.map(
422 workerNode
=> workerNode
.usage
.runTime
?.minimum
?? Infinity
428 ...this.workerNodes
.map(
429 workerNode
=> workerNode
.usage
.runTime
?.maximum
?? -Infinity
433 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
434 .runTime
.average
&& {
437 this.workerNodes
.reduce
<number[]>(
438 (accumulator
, workerNode
) =>
439 accumulator
.concat(workerNode
.usage
.runTime
.history
),
445 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
449 this.workerNodes
.reduce
<number[]>(
450 (accumulator
, workerNode
) =>
451 accumulator
.concat(workerNode
.usage
.runTime
.history
),
459 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
460 .waitTime
.aggregate
&& {
464 ...this.workerNodes
.map(
465 workerNode
=> workerNode
.usage
.waitTime
?.minimum
?? Infinity
471 ...this.workerNodes
.map(
472 workerNode
=> workerNode
.usage
.waitTime
?.maximum
?? -Infinity
476 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
477 .waitTime
.average
&& {
480 this.workerNodes
.reduce
<number[]>(
481 (accumulator
, workerNode
) =>
482 accumulator
.concat(workerNode
.usage
.waitTime
.history
),
488 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
489 .waitTime
.median
&& {
492 this.workerNodes
.reduce
<number[]>(
493 (accumulator
, workerNode
) =>
494 accumulator
.concat(workerNode
.usage
.waitTime
.history
),
506 * The pool readiness boolean status.
508 private get
ready (): boolean {
510 this.workerNodes
.reduce(
511 (accumulator
, workerNode
) =>
512 !workerNode
.info
.dynamic
&& workerNode
.info
.ready
521 * The approximate pool utilization.
523 * @returns The pool utilization.
525 private get
utilization (): number {
526 const poolTimeCapacity
=
527 (performance
.now() - this.startTimestamp
) * this.maxSize
528 const totalTasksRunTime
= this.workerNodes
.reduce(
529 (accumulator
, workerNode
) =>
530 accumulator
+ (workerNode
.usage
.runTime
?.aggregate
?? 0),
533 const totalTasksWaitTime
= this.workerNodes
.reduce(
534 (accumulator
, workerNode
) =>
535 accumulator
+ (workerNode
.usage
.waitTime
?.aggregate
?? 0),
538 return (totalTasksRunTime
+ totalTasksWaitTime
) / poolTimeCapacity
544 * If it is `'dynamic'`, it provides the `max` property.
546 protected abstract get
type (): PoolType
551 protected abstract get
worker (): WorkerType
554 * The pool minimum size.
556 protected get
minSize (): number {
557 return this.numberOfWorkers
561 * The pool maximum size.
563 protected get
maxSize (): number {
564 return this.max
?? this.numberOfWorkers
568 * Checks if the worker id sent in the received message from a worker is valid.
570 * @param message - The received message.
571 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the worker id is invalid.
573 private checkMessageWorkerId (message
: MessageValue
<Response
>): void {
574 if (message
.workerId
== null) {
575 throw new Error('Worker message received without worker id')
577 message
.workerId
!= null &&
578 this.getWorkerNodeKeyByWorkerId(message
.workerId
) === -1
581 `Worker message received from unknown worker '${message.workerId}'`
587 * Gets the given worker its worker node key.
589 * @param worker - The worker.
590 * @returns The worker node key if found in the pool worker nodes, `-1` otherwise.
592 private getWorkerNodeKeyByWorker (worker
: Worker
): number {
593 return this.workerNodes
.findIndex(
594 workerNode
=> workerNode
.worker
=== worker
599 * Gets the worker node key given its worker id.
601 * @param workerId - The worker id.
602 * @returns The worker node key if the worker id is found in the pool worker nodes, `-1` otherwise.
604 private getWorkerNodeKeyByWorkerId (workerId
: number): number {
605 return this.workerNodes
.findIndex(
606 workerNode
=> workerNode
.info
.id
=== workerId
611 public setWorkerChoiceStrategy (
612 workerChoiceStrategy
: WorkerChoiceStrategy
,
613 workerChoiceStrategyOptions
?: WorkerChoiceStrategyOptions
615 this.checkValidWorkerChoiceStrategy(workerChoiceStrategy
)
616 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
617 this.workerChoiceStrategyContext
.setWorkerChoiceStrategy(
618 this.opts
.workerChoiceStrategy
620 if (workerChoiceStrategyOptions
!= null) {
621 this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
623 for (const [workerNodeKey
, workerNode
] of this.workerNodes
.entries()) {
624 workerNode
.resetUsage()
625 this.sendStatisticsMessageToWorker(workerNodeKey
)
630 public setWorkerChoiceStrategyOptions (
631 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
633 this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
634 this.opts
.workerChoiceStrategyOptions
= {
635 ...DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
,
636 ...workerChoiceStrategyOptions
638 this.workerChoiceStrategyContext
.setOptions(
639 this.opts
.workerChoiceStrategyOptions
644 public enableTasksQueue (
646 tasksQueueOptions
?: TasksQueueOptions
648 if (this.opts
.enableTasksQueue
=== true && !enable
) {
649 this.flushTasksQueues()
651 this.opts
.enableTasksQueue
= enable
652 this.setTasksQueueOptions(tasksQueueOptions
as TasksQueueOptions
)
656 public setTasksQueueOptions (tasksQueueOptions
: TasksQueueOptions
): void {
657 if (this.opts
.enableTasksQueue
=== true) {
658 this.checkValidTasksQueueOptions(tasksQueueOptions
)
659 this.opts
.tasksQueueOptions
=
660 this.buildTasksQueueOptions(tasksQueueOptions
)
661 this.setTasksQueueSize(this.opts
.tasksQueueOptions
.size
as number)
662 } else if (this.opts
.tasksQueueOptions
!= null) {
663 delete this.opts
.tasksQueueOptions
667 private setTasksQueueSize (size
: number): void {
668 for (const workerNode
of this.workerNodes
) {
669 workerNode
.tasksQueueBackPressureSize
= size
673 private buildTasksQueueOptions (
674 tasksQueueOptions
: TasksQueueOptions
675 ): TasksQueueOptions
{
678 size
: Math.pow(this.maxSize
, 2),
686 * Whether the pool is full or not.
688 * The pool filling boolean status.
690 protected get
full (): boolean {
691 return this.workerNodes
.length
>= this.maxSize
695 * Whether the pool is busy or not.
697 * The pool busyness boolean status.
699 protected abstract get
busy (): boolean
702 * Whether worker nodes are executing concurrently their tasks quota or not.
704 * @returns Worker nodes busyness boolean status.
706 protected internalBusy (): boolean {
707 if (this.opts
.enableTasksQueue
=== true) {
709 this.workerNodes
.findIndex(
711 workerNode
.info
.ready
&&
712 workerNode
.usage
.tasks
.executing
<
713 (this.opts
.tasksQueueOptions
?.concurrency
as number)
718 this.workerNodes
.findIndex(
720 workerNode
.info
.ready
&& workerNode
.usage
.tasks
.executing
=== 0
726 private async sendTaskFunctionOperationToWorker (
727 message
: Omit
<MessageValue
<Data
>, 'workerId'>
728 ): Promise
<boolean> {
729 return await new Promise
<boolean>((resolve
, reject
) => {
730 const responsesReceived
= new Array<MessageValue
<Data
| Response
>>()
731 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
732 this.registerWorkerMessageListener(workerNodeKey
, message
=> {
733 if (message
.taskFunctionOperationStatus
!= null) {
734 responsesReceived
.push(message
)
736 responsesReceived
.length
=== this.workerNodes
.length
&&
737 responsesReceived
.every(
738 message
=> message
.taskFunctionOperationStatus
=== true
743 responsesReceived
.length
=== this.workerNodes
.length
&&
744 responsesReceived
.some(
745 message
=> message
.taskFunctionOperationStatus
=== false
750 `Task function operation ${
751 message.taskFunctionOperation as string
752 } failed on worker ${message.workerId}`
758 this.sendToWorker(workerNodeKey
, {
760 workerId
: this.getWorkerInfo(workerNodeKey
).id
as number
767 public hasTaskFunction (name
: string): boolean {
768 for (const workerNode
of this.workerNodes
) {
770 Array.isArray(workerNode
.info
.taskFunctionNames
) &&
771 workerNode
.info
.taskFunctionNames
.includes(name
)
780 public async addTaskFunction (
782 taskFunction
: TaskFunction
783 ): Promise
<boolean> {
784 return await this.sendTaskFunctionOperationToWorker({
785 taskFunctionOperation
: 'add',
786 taskFunctionName
: name
,
787 taskFunction
: taskFunction
.toString()
792 public async removeTaskFunction (name
: string): Promise
<boolean> {
793 return await this.sendTaskFunctionOperationToWorker({
794 taskFunctionOperation
: 'remove',
795 taskFunctionName
: name
800 public listTaskFunctionNames (): string[] {
801 for (const workerNode
of this.workerNodes
) {
803 Array.isArray(workerNode
.info
.taskFunctionNames
) &&
804 workerNode
.info
.taskFunctionNames
.length
> 0
806 return workerNode
.info
.taskFunctionNames
813 public async setDefaultTaskFunction (name
: string): Promise
<boolean> {
814 return await this.sendTaskFunctionOperationToWorker({
815 taskFunctionOperation
: 'default',
816 taskFunctionName
: name
820 private shallExecuteTask (workerNodeKey
: number): boolean {
822 this.tasksQueueSize(workerNodeKey
) === 0 &&
823 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
<
824 (this.opts
.tasksQueueOptions
?.concurrency
as number)
829 public async execute (
832 transferList
?: TransferListItem
[]
833 ): Promise
<Response
> {
834 return await new Promise
<Response
>((resolve
, reject
) => {
836 reject(new Error('Cannot execute a task on destroyed pool'))
839 if (name
!= null && typeof name
!== 'string') {
840 reject(new TypeError('name argument must be a string'))
845 typeof name
=== 'string' &&
846 name
.trim().length
=== 0
848 reject(new TypeError('name argument must not be an empty string'))
851 if (transferList
!= null && !Array.isArray(transferList
)) {
852 reject(new TypeError('transferList argument must be an array'))
855 const timestamp
= performance
.now()
856 const workerNodeKey
= this.chooseWorkerNode()
857 const task
: Task
<Data
> = {
858 name
: name
?? DEFAULT_TASK_NAME
,
859 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
860 data
: data
?? ({} as Data
),
863 workerId
: this.getWorkerInfo(workerNodeKey
).id
as number,
866 this.promiseResponseMap
.set(task
.taskId
as string, {
872 this.opts
.enableTasksQueue
=== false ||
873 (this.opts
.enableTasksQueue
=== true &&
874 this.shallExecuteTask(workerNodeKey
))
876 this.executeTask(workerNodeKey
, task
)
878 this.enqueueTask(workerNodeKey
, task
)
884 public async destroy (): Promise
<void> {
886 this.workerNodes
.map(async (_
, workerNodeKey
) => {
887 await this.destroyWorkerNode(workerNodeKey
)
890 this.emitter
?.emit(PoolEvents
.destroy
, this.info
)
894 protected async sendKillMessageToWorker (
895 workerNodeKey
: number,
898 await new Promise
<void>((resolve
, reject
) => {
899 this.registerWorkerMessageListener(workerNodeKey
, message
=> {
900 if (message
.kill
=== 'success') {
902 } else if (message
.kill
=== 'failure') {
903 reject(new Error(`Worker ${workerId} kill message handling failed`))
906 this.sendToWorker(workerNodeKey
, { kill
: true, workerId
})
911 * Terminates the worker node given its worker node key.
913 * @param workerNodeKey - The worker node key.
915 protected abstract destroyWorkerNode (workerNodeKey
: number): Promise
<void>
918 * Setup hook to execute code before worker nodes are created in the abstract constructor.
923 protected setupHook (): void {
924 /* Intentionally empty */
928 * Should return whether the worker is the main worker or not.
930 protected abstract isMain (): boolean
933 * Hook executed before the worker task execution.
936 * @param workerNodeKey - The worker node key.
937 * @param task - The task to execute.
939 protected beforeTaskExecutionHook (
940 workerNodeKey
: number,
943 if (this.workerNodes
[workerNodeKey
]?.usage
!= null) {
944 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
945 ++workerUsage
.tasks
.executing
946 this.updateWaitTimeWorkerUsage(workerUsage
, task
)
949 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
950 this.workerNodes
[workerNodeKey
].getTaskFunctionWorkerUsage(
954 const taskFunctionWorkerUsage
= this.workerNodes
[
956 ].getTaskFunctionWorkerUsage(task
.name
as string) as WorkerUsage
957 ++taskFunctionWorkerUsage
.tasks
.executing
958 this.updateWaitTimeWorkerUsage(taskFunctionWorkerUsage
, task
)
963 * Hook executed after the worker task execution.
966 * @param workerNodeKey - The worker node key.
967 * @param message - The received message.
969 protected afterTaskExecutionHook (
970 workerNodeKey
: number,
971 message
: MessageValue
<Response
>
973 if (this.workerNodes
[workerNodeKey
]?.usage
!= null) {
974 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
975 this.updateTaskStatisticsWorkerUsage(workerUsage
, message
)
976 this.updateRunTimeWorkerUsage(workerUsage
, message
)
977 this.updateEluWorkerUsage(workerUsage
, message
)
980 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
981 this.workerNodes
[workerNodeKey
].getTaskFunctionWorkerUsage(
982 message
.taskPerformance
?.name
as string
985 const taskFunctionWorkerUsage
= this.workerNodes
[
987 ].getTaskFunctionWorkerUsage(
988 message
.taskPerformance
?.name
as string
990 this.updateTaskStatisticsWorkerUsage(taskFunctionWorkerUsage
, message
)
991 this.updateRunTimeWorkerUsage(taskFunctionWorkerUsage
, message
)
992 this.updateEluWorkerUsage(taskFunctionWorkerUsage
, message
)
997 * Whether the worker node shall update its task function worker usage or not.
999 * @param workerNodeKey - The worker node key.
1000 * @returns `true` if the worker node shall update its task function worker usage, `false` otherwise.
1002 private shallUpdateTaskFunctionWorkerUsage (workerNodeKey
: number): boolean {
1003 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1005 workerInfo
!= null &&
1006 Array.isArray(workerInfo
.taskFunctionNames
) &&
1007 workerInfo
.taskFunctionNames
.length
> 2
1011 private updateTaskStatisticsWorkerUsage (
1012 workerUsage
: WorkerUsage
,
1013 message
: MessageValue
<Response
>
1015 const workerTaskStatistics
= workerUsage
.tasks
1017 workerTaskStatistics
.executing
!= null &&
1018 workerTaskStatistics
.executing
> 0
1020 --workerTaskStatistics
.executing
1022 if (message
.workerError
== null) {
1023 ++workerTaskStatistics
.executed
1025 ++workerTaskStatistics
.failed
1029 private updateRunTimeWorkerUsage (
1030 workerUsage
: WorkerUsage
,
1031 message
: MessageValue
<Response
>
1033 if (message
.workerError
!= null) {
1036 updateMeasurementStatistics(
1037 workerUsage
.runTime
,
1038 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
,
1039 message
.taskPerformance
?.runTime
?? 0
1043 private updateWaitTimeWorkerUsage (
1044 workerUsage
: WorkerUsage
,
1047 const timestamp
= performance
.now()
1048 const taskWaitTime
= timestamp
- (task
.timestamp
?? timestamp
)
1049 updateMeasurementStatistics(
1050 workerUsage
.waitTime
,
1051 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().waitTime
,
1056 private updateEluWorkerUsage (
1057 workerUsage
: WorkerUsage
,
1058 message
: MessageValue
<Response
>
1060 if (message
.workerError
!= null) {
1063 const eluTaskStatisticsRequirements
: MeasurementStatisticsRequirements
=
1064 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
1065 updateMeasurementStatistics(
1066 workerUsage
.elu
.active
,
1067 eluTaskStatisticsRequirements
,
1068 message
.taskPerformance
?.elu
?.active
?? 0
1070 updateMeasurementStatistics(
1071 workerUsage
.elu
.idle
,
1072 eluTaskStatisticsRequirements
,
1073 message
.taskPerformance
?.elu
?.idle
?? 0
1075 if (eluTaskStatisticsRequirements
.aggregate
) {
1076 if (message
.taskPerformance
?.elu
!= null) {
1077 if (workerUsage
.elu
.utilization
!= null) {
1078 workerUsage
.elu
.utilization
=
1079 (workerUsage
.elu
.utilization
+
1080 message
.taskPerformance
.elu
.utilization
) /
1083 workerUsage
.elu
.utilization
= message
.taskPerformance
.elu
.utilization
1090 * Chooses a worker node for the next task.
1092 * The default worker choice strategy uses a round robin algorithm to distribute the tasks.
1094 * @returns The chosen worker node key
1096 private chooseWorkerNode (): number {
1097 if (this.shallCreateDynamicWorker()) {
1098 const workerNodeKey
= this.createAndSetupDynamicWorkerNode()
1100 this.workerChoiceStrategyContext
.getStrategyPolicy().dynamicWorkerUsage
1102 return workerNodeKey
1105 return this.workerChoiceStrategyContext
.execute()
1109 * Conditions for dynamic worker creation.
1111 * @returns Whether to create a dynamic worker or not.
1113 private shallCreateDynamicWorker (): boolean {
1114 return this.type === PoolTypes
.dynamic
&& !this.full
&& this.internalBusy()
1118 * Sends a message to worker given its worker node key.
1120 * @param workerNodeKey - The worker node key.
1121 * @param message - The message.
1122 * @param transferList - The optional array of transferable objects.
1124 protected abstract sendToWorker (
1125 workerNodeKey
: number,
1126 message
: MessageValue
<Data
>,
1127 transferList
?: TransferListItem
[]
1131 * Creates a new worker.
1133 * @returns Newly created worker.
1135 protected abstract createWorker (): Worker
1138 * Creates a new, completely set up worker node.
1140 * @returns New, completely set up worker node key.
1142 protected createAndSetupWorkerNode (): number {
1143 const worker
= this.createWorker()
1145 worker
.on('online', this.opts
.onlineHandler
?? EMPTY_FUNCTION
)
1146 worker
.on('message', this.opts
.messageHandler
?? EMPTY_FUNCTION
)
1147 worker
.on('error', this.opts
.errorHandler
?? EMPTY_FUNCTION
)
1148 worker
.on('error', error
=> {
1149 const workerNodeKey
= this.getWorkerNodeKeyByWorker(worker
)
1150 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1151 workerInfo
.ready
= false
1152 this.workerNodes
[workerNodeKey
].closeChannel()
1153 this.emitter
?.emit(PoolEvents
.error
, error
)
1155 this.opts
.restartWorkerOnError
=== true &&
1159 if (workerInfo
.dynamic
) {
1160 this.createAndSetupDynamicWorkerNode()
1162 this.createAndSetupWorkerNode()
1165 if (this.opts
.enableTasksQueue
=== true) {
1166 this.redistributeQueuedTasks(workerNodeKey
)
1169 worker
.on('exit', this.opts
.exitHandler
?? EMPTY_FUNCTION
)
1170 worker
.once('exit', () => {
1171 this.removeWorkerNode(worker
)
1174 const workerNodeKey
= this.addWorkerNode(worker
)
1176 this.afterWorkerNodeSetup(workerNodeKey
)
1178 return workerNodeKey
1182 * Creates a new, completely set up dynamic worker node.
1184 * @returns New, completely set up dynamic worker node key.
1186 protected createAndSetupDynamicWorkerNode (): number {
1187 const workerNodeKey
= this.createAndSetupWorkerNode()
1188 this.registerWorkerMessageListener(workerNodeKey
, message
=> {
1189 const localWorkerNodeKey
= this.getWorkerNodeKeyByWorkerId(
1192 const workerUsage
= this.workerNodes
[localWorkerNodeKey
].usage
1193 // Kill message received from worker
1195 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
1196 (isKillBehavior(KillBehaviors
.SOFT
, message
.kill
) &&
1197 ((this.opts
.enableTasksQueue
=== false &&
1198 workerUsage
.tasks
.executing
=== 0) ||
1199 (this.opts
.enableTasksQueue
=== true &&
1200 workerUsage
.tasks
.executing
=== 0 &&
1201 this.tasksQueueSize(localWorkerNodeKey
) === 0)))
1203 this.destroyWorkerNode(localWorkerNodeKey
).catch(error
=> {
1204 this.emitter
?.emit(PoolEvents
.error
, error
)
1208 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1209 this.sendToWorker(workerNodeKey
, {
1211 workerId
: workerInfo
.id
as number
1213 workerInfo
.dynamic
= true
1215 this.workerChoiceStrategyContext
.getStrategyPolicy().dynamicWorkerReady
||
1216 this.workerChoiceStrategyContext
.getStrategyPolicy().dynamicWorkerUsage
1218 workerInfo
.ready
= true
1220 this.checkAndEmitDynamicWorkerCreationEvents()
1221 return workerNodeKey
1225 * Registers a listener callback on the worker given its worker node key.
1227 * @param workerNodeKey - The worker node key.
1228 * @param listener - The message listener callback.
1230 protected abstract registerWorkerMessageListener
<
1231 Message
extends Data
| Response
1233 workerNodeKey
: number,
1234 listener
: (message
: MessageValue
<Message
>) => void
1238 * Method hooked up after a worker node has been newly created.
1239 * Can be overridden.
1241 * @param workerNodeKey - The newly created worker node key.
1243 protected afterWorkerNodeSetup (workerNodeKey
: number): void {
1244 // Listen to worker messages.
1245 this.registerWorkerMessageListener(workerNodeKey
, this.workerListener())
1246 // Send the startup message to worker.
1247 this.sendStartupMessageToWorker(workerNodeKey
)
1248 // Send the statistics message to worker.
1249 this.sendStatisticsMessageToWorker(workerNodeKey
)
1250 if (this.opts
.enableTasksQueue
=== true) {
1251 this.workerNodes
[workerNodeKey
].onEmptyQueue
=
1252 this.taskStealingOnEmptyQueue
.bind(this)
1253 this.workerNodes
[workerNodeKey
].onBackPressure
=
1254 this.tasksStealingOnBackPressure
.bind(this)
1259 * Sends the startup message to worker given its worker node key.
1261 * @param workerNodeKey - The worker node key.
1263 protected abstract sendStartupMessageToWorker (workerNodeKey
: number): void
1266 * Sends the statistics message to worker given its worker node key.
1268 * @param workerNodeKey - The worker node key.
1270 private sendStatisticsMessageToWorker (workerNodeKey
: number): void {
1271 this.sendToWorker(workerNodeKey
, {
1274 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1276 elu
: this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1279 workerId
: this.getWorkerInfo(workerNodeKey
).id
as number
1283 private redistributeQueuedTasks (workerNodeKey
: number): void {
1284 while (this.tasksQueueSize(workerNodeKey
) > 0) {
1285 const destinationWorkerNodeKey
= this.workerNodes
.reduce(
1286 (minWorkerNodeKey
, workerNode
, workerNodeKey
, workerNodes
) => {
1287 return workerNode
.info
.ready
&&
1288 workerNode
.usage
.tasks
.queued
<
1289 workerNodes
[minWorkerNodeKey
].usage
.tasks
.queued
1295 const destinationWorkerNode
= this.workerNodes
[destinationWorkerNodeKey
]
1297 ...(this.dequeueTask(workerNodeKey
) as Task
<Data
>),
1298 workerId
: destinationWorkerNode
.info
.id
as number
1300 if (this.shallExecuteTask(destinationWorkerNodeKey
)) {
1301 this.executeTask(destinationWorkerNodeKey
, task
)
1303 this.enqueueTask(destinationWorkerNodeKey
, task
)
1308 private updateTaskStolenStatisticsWorkerUsage (
1309 workerNodeKey
: number,
1312 const workerNode
= this.workerNodes
[workerNodeKey
]
1313 if (workerNode
?.usage
!= null) {
1314 ++workerNode
.usage
.tasks
.stolen
1317 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1318 workerNode
.getTaskFunctionWorkerUsage(taskName
) != null
1320 const taskFunctionWorkerUsage
= workerNode
.getTaskFunctionWorkerUsage(
1323 ++taskFunctionWorkerUsage
.tasks
.stolen
1327 private taskStealingOnEmptyQueue (workerId
: number): void {
1328 const destinationWorkerNodeKey
= this.getWorkerNodeKeyByWorkerId(workerId
)
1329 const destinationWorkerNode
= this.workerNodes
[destinationWorkerNodeKey
]
1330 const workerNodes
= this.workerNodes
1333 (workerNodeA
, workerNodeB
) =>
1334 workerNodeB
.usage
.tasks
.queued
- workerNodeA
.usage
.tasks
.queued
1336 const sourceWorkerNode
= workerNodes
.find(
1338 workerNode
.info
.ready
&&
1339 workerNode
.info
.id
!== workerId
&&
1340 workerNode
.usage
.tasks
.queued
> 0
1342 if (sourceWorkerNode
!= null) {
1344 ...(sourceWorkerNode
.popTask() as Task
<Data
>),
1345 workerId
: destinationWorkerNode
.info
.id
as number
1347 if (this.shallExecuteTask(destinationWorkerNodeKey
)) {
1348 this.executeTask(destinationWorkerNodeKey
, task
)
1350 this.enqueueTask(destinationWorkerNodeKey
, task
)
1352 this.updateTaskStolenStatisticsWorkerUsage(
1353 destinationWorkerNodeKey
,
1359 private tasksStealingOnBackPressure (workerId
: number): void {
1360 const sizeOffset
= 1
1361 if ((this.opts
.tasksQueueOptions
?.size
as number) <= sizeOffset
) {
1364 const sourceWorkerNode
=
1365 this.workerNodes
[this.getWorkerNodeKeyByWorkerId(workerId
)]
1366 const workerNodes
= this.workerNodes
1369 (workerNodeA
, workerNodeB
) =>
1370 workerNodeA
.usage
.tasks
.queued
- workerNodeB
.usage
.tasks
.queued
1372 for (const [workerNodeKey
, workerNode
] of workerNodes
.entries()) {
1374 sourceWorkerNode
.usage
.tasks
.queued
> 0 &&
1375 workerNode
.info
.ready
&&
1376 workerNode
.info
.id
!== workerId
&&
1377 workerNode
.usage
.tasks
.queued
<
1378 (this.opts
.tasksQueueOptions
?.size
as number) - sizeOffset
1381 ...(sourceWorkerNode
.popTask() as Task
<Data
>),
1382 workerId
: workerNode
.info
.id
as number
1384 if (this.shallExecuteTask(workerNodeKey
)) {
1385 this.executeTask(workerNodeKey
, task
)
1387 this.enqueueTask(workerNodeKey
, task
)
1389 this.updateTaskStolenStatisticsWorkerUsage(
1398 * This method is the listener registered for each worker message.
1400 * @returns The listener function to execute when a message is received from a worker.
1402 protected workerListener (): (message
: MessageValue
<Response
>) => void {
1404 this.checkMessageWorkerId(message
)
1405 if (message
.ready
!= null && message
.taskFunctionNames
!= null) {
1406 // Worker ready response received from worker
1407 this.handleWorkerReadyResponse(message
)
1408 } else if (message
.taskId
!= null) {
1409 // Task execution response received from worker
1410 this.handleTaskExecutionResponse(message
)
1411 } else if (message
.taskFunctionNames
!= null) {
1412 // Task function names message received from worker
1414 this.getWorkerNodeKeyByWorkerId(message
.workerId
)
1415 ).taskFunctionNames
= message
.taskFunctionNames
1416 } else if (message
.taskFunctionOperation
!= null) {
1417 // Task function operation response received from worker
1422 private handleWorkerReadyResponse (message
: MessageValue
<Response
>): void {
1423 if (message
.ready
=== false) {
1424 throw new Error(`Worker ${message.workerId} failed to initialize`)
1426 const workerInfo
= this.getWorkerInfo(
1427 this.getWorkerNodeKeyByWorkerId(message
.workerId
)
1429 workerInfo
.ready
= message
.ready
as boolean
1430 workerInfo
.taskFunctionNames
= message
.taskFunctionNames
1431 if (this.emitter
!= null && this.ready
) {
1432 this.emitter
.emit(PoolEvents
.ready
, this.info
)
1436 private handleTaskExecutionResponse (message
: MessageValue
<Response
>): void {
1437 const { taskId
, workerError
, data
} = message
1438 const promiseResponse
= this.promiseResponseMap
.get(taskId
as string)
1439 if (promiseResponse
!= null) {
1440 if (workerError
!= null) {
1441 this.emitter
?.emit(PoolEvents
.taskError
, workerError
)
1442 promiseResponse
.reject(workerError
.message
)
1444 promiseResponse
.resolve(data
as Response
)
1446 const workerNodeKey
= promiseResponse
.workerNodeKey
1447 this.afterTaskExecutionHook(workerNodeKey
, message
)
1448 this.workerChoiceStrategyContext
.update(workerNodeKey
)
1449 this.promiseResponseMap
.delete(taskId
as string)
1451 this.opts
.enableTasksQueue
=== true &&
1452 this.tasksQueueSize(workerNodeKey
) > 0 &&
1453 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
<
1454 (this.opts
.tasksQueueOptions
?.concurrency
as number)
1458 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1464 private checkAndEmitTaskExecutionEvents (): void {
1466 this.emitter
?.emit(PoolEvents
.busy
, this.info
)
1470 private checkAndEmitTaskQueuingEvents (): void {
1471 if (this.hasBackPressure()) {
1472 this.emitter
?.emit(PoolEvents
.backPressure
, this.info
)
1476 private checkAndEmitDynamicWorkerCreationEvents (): void {
1477 if (this.type === PoolTypes
.dynamic
) {
1479 this.emitter
?.emit(PoolEvents
.full
, this.info
)
1485 * Gets the worker information given its worker node key.
1487 * @param workerNodeKey - The worker node key.
1488 * @returns The worker information.
1490 protected getWorkerInfo (workerNodeKey
: number): WorkerInfo
{
1491 return this.workerNodes
[workerNodeKey
].info
1495 * Adds the given worker in the pool worker nodes.
1497 * @param worker - The worker.
1498 * @returns The added worker node key.
1499 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the added worker node is not found.
1501 private addWorkerNode (worker
: Worker
): number {
1502 const workerNode
= new WorkerNode
<Worker
, Data
>(
1504 this.opts
.tasksQueueOptions
?.size
?? Math.pow(this.maxSize
, 2)
1506 // Flag the worker node as ready at pool startup.
1507 if (this.starting
) {
1508 workerNode
.info
.ready
= true
1510 this.workerNodes
.push(workerNode
)
1511 const workerNodeKey
= this.getWorkerNodeKeyByWorker(worker
)
1512 if (workerNodeKey
=== -1) {
1513 throw new Error('Worker added not found in worker nodes')
1515 return workerNodeKey
1519 * Removes the given worker from the pool worker nodes.
1521 * @param worker - The worker.
1523 private removeWorkerNode (worker
: Worker
): void {
1524 const workerNodeKey
= this.getWorkerNodeKeyByWorker(worker
)
1525 if (workerNodeKey
!== -1) {
1526 this.workerNodes
.splice(workerNodeKey
, 1)
1527 this.workerChoiceStrategyContext
.remove(workerNodeKey
)
1532 public hasWorkerNodeBackPressure (workerNodeKey
: number): boolean {
1534 this.opts
.enableTasksQueue
=== true &&
1535 this.workerNodes
[workerNodeKey
].hasBackPressure()
1539 private hasBackPressure (): boolean {
1541 this.opts
.enableTasksQueue
=== true &&
1542 this.workerNodes
.findIndex(
1543 workerNode
=> !workerNode
.hasBackPressure()
1549 * Executes the given task on the worker given its worker node key.
1551 * @param workerNodeKey - The worker node key.
1552 * @param task - The task to execute.
1554 private executeTask (workerNodeKey
: number, task
: Task
<Data
>): void {
1555 this.beforeTaskExecutionHook(workerNodeKey
, task
)
1556 this.sendToWorker(workerNodeKey
, task
, task
.transferList
)
1557 this.checkAndEmitTaskExecutionEvents()
1560 private enqueueTask (workerNodeKey
: number, task
: Task
<Data
>): number {
1561 const tasksQueueSize
= this.workerNodes
[workerNodeKey
].enqueueTask(task
)
1562 this.checkAndEmitTaskQueuingEvents()
1563 return tasksQueueSize
1566 private dequeueTask (workerNodeKey
: number): Task
<Data
> | undefined {
1567 return this.workerNodes
[workerNodeKey
].dequeueTask()
1570 private tasksQueueSize (workerNodeKey
: number): number {
1571 return this.workerNodes
[workerNodeKey
].tasksQueueSize()
1574 protected flushTasksQueue (workerNodeKey
: number): void {
1575 while (this.tasksQueueSize(workerNodeKey
) > 0) {
1578 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1581 this.workerNodes
[workerNodeKey
].clearTasksQueue()
1584 private flushTasksQueues (): void {
1585 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
1586 this.flushTasksQueue(workerNodeKey
)