1 import { AsyncResource
} from
'node:async_hooks'
2 import { randomUUID
} from
'node:crypto'
3 import { EventEmitterAsyncResource
} from
'node:events'
4 import { performance
} from
'node:perf_hooks'
5 import type { TransferListItem
} from
'node:worker_threads'
9 PromiseResponseWrapper
,
11 TaskFunctionProperties
12 } from
'../utility-types.js'
15 buildTaskFunctionProperties
,
30 } from
'../worker/task-functions.js'
31 import { KillBehaviors
} from
'../worker/worker-options.js'
39 type TasksQueueOptions
43 WorkerChoiceStrategies
,
44 type WorkerChoiceStrategy
,
45 type WorkerChoiceStrategyOptions
46 } from
'./selection-strategies/selection-strategies-types.js'
47 import { WorkerChoiceStrategiesContext
} from
'./selection-strategies/worker-choice-strategies-context.js'
51 checkValidTasksQueueOptions
,
52 checkValidWorkerChoiceStrategy
,
53 getDefaultTasksQueueOptions
,
55 updateRunTimeWorkerUsage
,
56 updateTaskStatisticsWorkerUsage
,
57 updateWaitTimeWorkerUsage
,
60 import { version
} from
'./version.js'
65 WorkerNodeEventDetail
,
68 import { WorkerNode
} from
'./worker-node.js'
71 * Base class that implements some shared logic for all poolifier pools.
73 * @typeParam Worker - Type of worker which manages this pool.
74 * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
75 * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
77 export abstract class AbstractPool
<
78 Worker
extends IWorker
,
81 > implements IPool
<Worker
, Data
, Response
> {
83 public readonly workerNodes
: Array<IWorkerNode
<Worker
, Data
>> = []
86 public emitter
?: EventEmitterAsyncResource
89 * The task execution response promise map:
90 * - `key`: The message id of each submitted task.
91 * - `value`: An object that contains task's worker node key, execution response promise resolve and reject callbacks, async resource.
93 * When we receive a message from the worker, we get a map entry with the promise resolve/reject bound to the message id.
95 protected promiseResponseMap
: Map
<string, PromiseResponseWrapper
<Response
>> =
96 new Map
<string, PromiseResponseWrapper
<Response
>>()
99 * Worker choice strategies context referencing worker choice algorithms implementation.
101 protected workerChoiceStrategiesContext
?: WorkerChoiceStrategiesContext
<
108 * The task functions added at runtime map:
109 * - `key`: The task function name.
110 * - `value`: The task function object.
112 private readonly taskFunctions
: Map
<
114 TaskFunctionObject
<Data
, Response
>
118 * Whether the pool is started or not.
120 private started
: boolean
122 * Whether the pool is starting or not.
124 private starting
: boolean
126 * Whether the pool is destroying or not.
128 private destroying
: boolean
130 * Whether the minimum number of workers is starting or not.
132 private startingMinimumNumberOfWorkers
: boolean
134 * Whether the pool ready event has been emitted or not.
136 private readyEventEmitted
: boolean
138 * The start timestamp of the pool.
140 private readonly startTimestamp
143 * Constructs a new poolifier pool.
145 * @param minimumNumberOfWorkers - Minimum number of workers that this pool manages.
146 * @param filePath - Path to the worker file.
147 * @param opts - Options for the pool.
148 * @param maximumNumberOfWorkers - Maximum number of workers that this pool manages.
151 protected readonly minimumNumberOfWorkers
: number,
152 protected readonly filePath
: string,
153 protected readonly opts
: PoolOptions
<Worker
>,
154 protected readonly maximumNumberOfWorkers
?: number
156 if (!this.isMain()) {
158 'Cannot start a pool from a worker with the same type as the pool'
162 checkFilePath(this.filePath
)
163 this.checkMinimumNumberOfWorkers(this.minimumNumberOfWorkers
)
164 this.checkPoolOptions(this.opts
)
166 this.chooseWorkerNode
= this.chooseWorkerNode
.bind(this)
167 this.executeTask
= this.executeTask
.bind(this)
168 this.enqueueTask
= this.enqueueTask
.bind(this)
170 if (this.opts
.enableEvents
=== true) {
171 this.initializeEventEmitter()
173 this.workerChoiceStrategiesContext
= new WorkerChoiceStrategiesContext
<
179 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
180 [this.opts
.workerChoiceStrategy
!],
181 this.opts
.workerChoiceStrategyOptions
186 this.taskFunctions
= new Map
<string, TaskFunctionObject
<Data
, Response
>>()
189 this.starting
= false
190 this.destroying
= false
191 this.readyEventEmitted
= false
192 this.startingMinimumNumberOfWorkers
= false
193 if (this.opts
.startWorkers
=== true) {
197 this.startTimestamp
= performance
.now()
200 private checkPoolType (): void {
201 if (this.type === PoolTypes
.fixed
&& this.maximumNumberOfWorkers
!= null) {
203 'Cannot instantiate a fixed pool with a maximum number of workers specified at initialization'
208 private checkMinimumNumberOfWorkers (
209 minimumNumberOfWorkers
: number | undefined
211 if (minimumNumberOfWorkers
== null) {
213 'Cannot instantiate a pool without specifying the number of workers'
215 } else if (!Number.isSafeInteger(minimumNumberOfWorkers
)) {
217 'Cannot instantiate a pool with a non safe integer number of workers'
219 } else if (minimumNumberOfWorkers
< 0) {
220 throw new RangeError(
221 'Cannot instantiate a pool with a negative number of workers'
223 } else if (this.type === PoolTypes
.fixed
&& minimumNumberOfWorkers
=== 0) {
224 throw new RangeError('Cannot instantiate a fixed pool with zero worker')
228 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
229 if (isPlainObject(opts
)) {
230 this.opts
.startWorkers
= opts
.startWorkers
?? true
231 checkValidWorkerChoiceStrategy(opts
.workerChoiceStrategy
)
232 this.opts
.workerChoiceStrategy
=
233 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
234 this.checkValidWorkerChoiceStrategyOptions(
235 opts
.workerChoiceStrategyOptions
237 if (opts
.workerChoiceStrategyOptions
!= null) {
238 this.opts
.workerChoiceStrategyOptions
= opts
.workerChoiceStrategyOptions
240 this.opts
.restartWorkerOnError
= opts
.restartWorkerOnError
?? true
241 this.opts
.enableEvents
= opts
.enableEvents
?? true
242 this.opts
.enableTasksQueue
= opts
.enableTasksQueue
?? false
243 if (this.opts
.enableTasksQueue
) {
244 checkValidTasksQueueOptions(opts
.tasksQueueOptions
)
245 this.opts
.tasksQueueOptions
= this.buildTasksQueueOptions(
246 opts
.tasksQueueOptions
250 throw new TypeError('Invalid pool options: must be a plain object')
254 private checkValidWorkerChoiceStrategyOptions (
255 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
| undefined
258 workerChoiceStrategyOptions
!= null &&
259 !isPlainObject(workerChoiceStrategyOptions
)
262 'Invalid worker choice strategy options: must be a plain object'
266 workerChoiceStrategyOptions
?.weights
!= null &&
267 Object.keys(workerChoiceStrategyOptions
.weights
).length
!==
268 (this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
)
271 'Invalid worker choice strategy options: must have a weight for each worker node'
275 workerChoiceStrategyOptions
?.measurement
!= null &&
276 !Object.values(Measurements
).includes(
277 workerChoiceStrategyOptions
.measurement
281 `Invalid worker choice strategy options: invalid measurement '${workerChoiceStrategyOptions.measurement}'`
286 private initializeEventEmitter (): void {
287 this.emitter
= new EventEmitterAsyncResource({
288 name
: `poolifier:${this.type}-${this.worker}-pool`
293 public get
info (): PoolInfo
{
298 started
: this.started
,
300 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
301 defaultStrategy
: this.opts
.workerChoiceStrategy
!,
302 strategyRetries
: this.workerChoiceStrategiesContext
?.retriesCount
?? 0,
303 minSize
: this.minimumNumberOfWorkers
,
304 maxSize
: this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
,
305 ...(this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
306 .runTime
.aggregate
=== true &&
307 this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
308 .waitTime
.aggregate
&& {
309 utilization
: round(this.utilization
)
311 workerNodes
: this.workerNodes
.length
,
312 idleWorkerNodes
: this.workerNodes
.reduce(
313 (accumulator
, workerNode
) =>
314 workerNode
.usage
.tasks
.executing
=== 0
319 ...(this.opts
.enableTasksQueue
=== true && {
320 stealingWorkerNodes
: this.workerNodes
.reduce(
321 (accumulator
, workerNode
) =>
322 workerNode
.info
.stealing
? accumulator
+ 1 : accumulator
,
326 busyWorkerNodes
: this.workerNodes
.reduce(
327 (accumulator
, _workerNode
, workerNodeKey
) =>
328 this.isWorkerNodeBusy(workerNodeKey
) ? accumulator
+ 1 : accumulator
,
331 executedTasks
: this.workerNodes
.reduce(
332 (accumulator
, workerNode
) =>
333 accumulator
+ workerNode
.usage
.tasks
.executed
,
336 executingTasks
: this.workerNodes
.reduce(
337 (accumulator
, workerNode
) =>
338 accumulator
+ workerNode
.usage
.tasks
.executing
,
341 ...(this.opts
.enableTasksQueue
=== true && {
342 queuedTasks
: this.workerNodes
.reduce(
343 (accumulator
, workerNode
) =>
344 accumulator
+ workerNode
.usage
.tasks
.queued
,
348 ...(this.opts
.enableTasksQueue
=== true && {
349 maxQueuedTasks
: this.workerNodes
.reduce(
350 (accumulator
, workerNode
) =>
351 accumulator
+ (workerNode
.usage
.tasks
.maxQueued
?? 0),
355 ...(this.opts
.enableTasksQueue
=== true && {
356 backPressure
: this.hasBackPressure()
358 ...(this.opts
.enableTasksQueue
=== true && {
359 stolenTasks
: this.workerNodes
.reduce(
360 (accumulator
, workerNode
) =>
361 accumulator
+ workerNode
.usage
.tasks
.stolen
,
365 failedTasks
: this.workerNodes
.reduce(
366 (accumulator
, workerNode
) =>
367 accumulator
+ workerNode
.usage
.tasks
.failed
,
370 ...(this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
371 .runTime
.aggregate
=== true && {
375 ...this.workerNodes
.map(
376 workerNode
=> workerNode
.usage
.runTime
.minimum
?? Infinity
382 ...this.workerNodes
.map(
383 workerNode
=> workerNode
.usage
.runTime
.maximum
?? -Infinity
387 ...(this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
388 .runTime
.average
&& {
391 this.workerNodes
.reduce
<number[]>(
392 (accumulator
, workerNode
) =>
393 accumulator
.concat(workerNode
.usage
.runTime
.history
),
399 ...(this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
403 this.workerNodes
.reduce
<number[]>(
404 (accumulator
, workerNode
) =>
405 accumulator
.concat(workerNode
.usage
.runTime
.history
),
413 ...(this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
414 .waitTime
.aggregate
=== true && {
418 ...this.workerNodes
.map(
419 workerNode
=> workerNode
.usage
.waitTime
.minimum
?? Infinity
425 ...this.workerNodes
.map(
426 workerNode
=> workerNode
.usage
.waitTime
.maximum
?? -Infinity
430 ...(this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
431 .waitTime
.average
&& {
434 this.workerNodes
.reduce
<number[]>(
435 (accumulator
, workerNode
) =>
436 accumulator
.concat(workerNode
.usage
.waitTime
.history
),
442 ...(this.workerChoiceStrategiesContext
.getTaskStatisticsRequirements()
443 .waitTime
.median
&& {
446 this.workerNodes
.reduce
<number[]>(
447 (accumulator
, workerNode
) =>
448 accumulator
.concat(workerNode
.usage
.waitTime
.history
),
460 * The pool readiness boolean status.
462 private get
ready (): boolean {
467 this.workerNodes
.reduce(
468 (accumulator
, workerNode
) =>
469 !workerNode
.info
.dynamic
&& workerNode
.info
.ready
473 ) >= this.minimumNumberOfWorkers
478 * The pool emptiness boolean status.
480 protected get
empty (): boolean {
481 return this.minimumNumberOfWorkers
=== 0 && this.workerNodes
.length
=== 0
485 * The approximate pool utilization.
487 * @returns The pool utilization.
489 private get
utilization (): number {
490 const poolTimeCapacity
=
491 (performance
.now() - this.startTimestamp
) *
492 (this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
)
493 const totalTasksRunTime
= this.workerNodes
.reduce(
494 (accumulator
, workerNode
) =>
495 accumulator
+ (workerNode
.usage
.runTime
.aggregate
?? 0),
498 const totalTasksWaitTime
= this.workerNodes
.reduce(
499 (accumulator
, workerNode
) =>
500 accumulator
+ (workerNode
.usage
.waitTime
.aggregate
?? 0),
503 return (totalTasksRunTime
+ totalTasksWaitTime
) / poolTimeCapacity
509 * If it is `'dynamic'`, it provides the `max` property.
511 protected abstract get
type (): PoolType
516 protected abstract get
worker (): WorkerType
519 * Checks if the worker id sent in the received message from a worker is valid.
521 * @param message - The received message.
522 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the worker id is invalid.
524 private checkMessageWorkerId (message
: MessageValue
<Data
| Response
>): void {
525 if (message
.workerId
== null) {
526 throw new Error('Worker message received without worker id')
527 } else if (this.getWorkerNodeKeyByWorkerId(message
.workerId
) === -1) {
529 `Worker message received from unknown worker '${message.workerId}'`
535 * Gets the worker node key given its worker id.
537 * @param workerId - The worker id.
538 * @returns The worker node key if the worker id is found in the pool worker nodes, `-1` otherwise.
540 private getWorkerNodeKeyByWorkerId (workerId
: number | undefined): number {
541 return this.workerNodes
.findIndex(
542 workerNode
=> workerNode
.info
.id
=== workerId
547 public setWorkerChoiceStrategy (
548 workerChoiceStrategy
: WorkerChoiceStrategy
,
549 workerChoiceStrategyOptions
?: WorkerChoiceStrategyOptions
551 let requireSync
= false
552 checkValidWorkerChoiceStrategy(workerChoiceStrategy
)
553 if (workerChoiceStrategyOptions
!= null) {
554 requireSync
= this.setWorkerChoiceStrategyOptions(
555 workerChoiceStrategyOptions
558 if (workerChoiceStrategy
!== this.opts
.workerChoiceStrategy
) {
559 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
560 this.workerChoiceStrategiesContext
?.setDefaultWorkerChoiceStrategy(
561 this.opts
.workerChoiceStrategy
,
562 this.opts
.workerChoiceStrategyOptions
567 this.workerChoiceStrategiesContext
?.syncWorkerChoiceStrategies(
568 this.getWorkerWorkerChoiceStrategies(),
569 this.opts
.workerChoiceStrategyOptions
571 for (const workerNodeKey
of this.workerNodes
.keys()) {
572 this.sendStatisticsMessageToWorker(workerNodeKey
)
578 public setWorkerChoiceStrategyOptions (
579 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
| undefined
581 this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
582 if (workerChoiceStrategyOptions
!= null) {
583 this.opts
.workerChoiceStrategyOptions
= workerChoiceStrategyOptions
584 this.workerChoiceStrategiesContext
?.setOptions(
585 this.opts
.workerChoiceStrategyOptions
587 this.workerChoiceStrategiesContext
?.syncWorkerChoiceStrategies(
588 this.getWorkerWorkerChoiceStrategies(),
589 this.opts
.workerChoiceStrategyOptions
591 for (const workerNodeKey
of this.workerNodes
.keys()) {
592 this.sendStatisticsMessageToWorker(workerNodeKey
)
600 public enableTasksQueue (
602 tasksQueueOptions
?: TasksQueueOptions
604 if (this.opts
.enableTasksQueue
=== true && !enable
) {
605 this.unsetTaskStealing()
606 this.unsetTasksStealingOnBackPressure()
607 this.flushTasksQueues()
609 this.opts
.enableTasksQueue
= enable
610 this.setTasksQueueOptions(tasksQueueOptions
)
614 public setTasksQueueOptions (
615 tasksQueueOptions
: TasksQueueOptions
| undefined
617 if (this.opts
.enableTasksQueue
=== true) {
618 checkValidTasksQueueOptions(tasksQueueOptions
)
619 this.opts
.tasksQueueOptions
=
620 this.buildTasksQueueOptions(tasksQueueOptions
)
621 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
622 this.setTasksQueueSize(this.opts
.tasksQueueOptions
.size
!)
623 if (this.opts
.tasksQueueOptions
.taskStealing
=== true) {
624 this.unsetTaskStealing()
625 this.setTaskStealing()
627 this.unsetTaskStealing()
629 if (this.opts
.tasksQueueOptions
.tasksStealingOnBackPressure
=== true) {
630 this.unsetTasksStealingOnBackPressure()
631 this.setTasksStealingOnBackPressure()
633 this.unsetTasksStealingOnBackPressure()
635 } else if (this.opts
.tasksQueueOptions
!= null) {
636 delete this.opts
.tasksQueueOptions
640 private buildTasksQueueOptions (
641 tasksQueueOptions
: TasksQueueOptions
| undefined
642 ): TasksQueueOptions
{
644 ...getDefaultTasksQueueOptions(
645 this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
651 private setTasksQueueSize (size
: number): void {
652 for (const workerNode
of this.workerNodes
) {
653 workerNode
.tasksQueueBackPressureSize
= size
657 private setTaskStealing (): void {
658 for (const workerNodeKey
of this.workerNodes
.keys()) {
659 this.workerNodes
[workerNodeKey
].on('idle', this.handleWorkerNodeIdleEvent
)
663 private unsetTaskStealing (): void {
664 for (const workerNodeKey
of this.workerNodes
.keys()) {
665 this.workerNodes
[workerNodeKey
].off(
667 this.handleWorkerNodeIdleEvent
672 private setTasksStealingOnBackPressure (): void {
673 for (const workerNodeKey
of this.workerNodes
.keys()) {
674 this.workerNodes
[workerNodeKey
].on(
676 this.handleWorkerNodeBackPressureEvent
681 private unsetTasksStealingOnBackPressure (): void {
682 for (const workerNodeKey
of this.workerNodes
.keys()) {
683 this.workerNodes
[workerNodeKey
].off(
685 this.handleWorkerNodeBackPressureEvent
691 * Whether the pool is full or not.
693 * The pool filling boolean status.
695 protected get
full (): boolean {
697 this.workerNodes
.length
>=
698 (this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
)
703 * Whether the pool is busy or not.
705 * The pool busyness boolean status.
707 protected abstract get
busy (): boolean
710 * Whether worker nodes are executing concurrently their tasks quota or not.
712 * @returns Worker nodes busyness boolean status.
714 protected internalBusy (): boolean {
715 if (this.opts
.enableTasksQueue
=== true) {
717 this.workerNodes
.findIndex(
719 workerNode
.info
.ready
&&
720 workerNode
.usage
.tasks
.executing
<
721 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
722 this.opts
.tasksQueueOptions
!.concurrency
!
727 this.workerNodes
.findIndex(
729 workerNode
.info
.ready
&& workerNode
.usage
.tasks
.executing
=== 0
734 private isWorkerNodeBusy (workerNodeKey
: number): boolean {
735 if (this.opts
.enableTasksQueue
=== true) {
737 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
>=
738 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
739 this.opts
.tasksQueueOptions
!.concurrency
!
742 return this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
> 0
745 private async sendTaskFunctionOperationToWorker (
746 workerNodeKey
: number,
747 message
: MessageValue
<Data
>
748 ): Promise
<boolean> {
749 return await new Promise
<boolean>((resolve
, reject
) => {
750 const taskFunctionOperationListener
= (
751 message
: MessageValue
<Response
>
753 this.checkMessageWorkerId(message
)
754 const workerId
= this.getWorkerInfo(workerNodeKey
)?.id
756 message
.taskFunctionOperationStatus
!= null &&
757 message
.workerId
=== workerId
759 if (message
.taskFunctionOperationStatus
) {
764 `Task function operation '${message.taskFunctionOperation}' failed on worker ${message.workerId} with error: '${message.workerError?.message}'`
768 this.deregisterWorkerMessageListener(
769 this.getWorkerNodeKeyByWorkerId(message
.workerId
),
770 taskFunctionOperationListener
774 this.registerWorkerMessageListener(
776 taskFunctionOperationListener
778 this.sendToWorker(workerNodeKey
, message
)
782 private async sendTaskFunctionOperationToWorkers (
783 message
: MessageValue
<Data
>
784 ): Promise
<boolean> {
785 return await new Promise
<boolean>((resolve
, reject
) => {
786 const responsesReceived
= new Array<MessageValue
<Response
>>()
787 const taskFunctionOperationsListener
= (
788 message
: MessageValue
<Response
>
790 this.checkMessageWorkerId(message
)
791 if (message
.taskFunctionOperationStatus
!= null) {
792 responsesReceived
.push(message
)
793 if (responsesReceived
.length
=== this.workerNodes
.length
) {
795 responsesReceived
.every(
796 message
=> message
.taskFunctionOperationStatus
=== true
801 responsesReceived
.some(
802 message
=> message
.taskFunctionOperationStatus
=== false
805 const errorResponse
= responsesReceived
.find(
806 response
=> response
.taskFunctionOperationStatus
=== false
810 `Task function operation '${
811 message.taskFunctionOperation as string
812 }' failed on worker ${errorResponse?.workerId} with error: '${
813 errorResponse?.workerError?.message
818 this.deregisterWorkerMessageListener(
819 this.getWorkerNodeKeyByWorkerId(message
.workerId
),
820 taskFunctionOperationsListener
825 for (const workerNodeKey
of this.workerNodes
.keys()) {
826 this.registerWorkerMessageListener(
828 taskFunctionOperationsListener
830 this.sendToWorker(workerNodeKey
, message
)
836 public hasTaskFunction (name
: string): boolean {
837 return this.listTaskFunctionsProperties().some(
838 taskFunctionProperties
=> taskFunctionProperties
.name
=== name
843 public async addTaskFunction (
845 fn
: TaskFunction
<Data
, Response
> | TaskFunctionObject
<Data
, Response
>
846 ): Promise
<boolean> {
847 if (typeof name
!== 'string') {
848 throw new TypeError('name argument must be a string')
850 if (typeof name
=== 'string' && name
.trim().length
=== 0) {
851 throw new TypeError('name argument must not be an empty string')
853 if (typeof fn
=== 'function') {
854 fn
= { taskFunction
: fn
} satisfies TaskFunctionObject
<Data
, Response
>
856 if (typeof fn
.taskFunction
!== 'function') {
857 throw new TypeError('taskFunction property must be a function')
859 checkValidPriority(fn
.priority
)
860 checkValidWorkerChoiceStrategy(fn
.strategy
)
861 const opResult
= await this.sendTaskFunctionOperationToWorkers({
862 taskFunctionOperation
: 'add',
863 taskFunctionProperties
: buildTaskFunctionProperties(name
, fn
),
864 taskFunction
: fn
.taskFunction
.toString()
866 this.taskFunctions
.set(name
, fn
)
867 this.workerChoiceStrategiesContext
?.syncWorkerChoiceStrategies(
868 this.getWorkerWorkerChoiceStrategies()
874 public async removeTaskFunction (name
: string): Promise
<boolean> {
875 if (!this.taskFunctions
.has(name
)) {
877 'Cannot remove a task function not handled on the pool side'
880 const opResult
= await this.sendTaskFunctionOperationToWorkers({
881 taskFunctionOperation
: 'remove',
882 taskFunctionProperties
: buildTaskFunctionProperties(
884 this.taskFunctions
.get(name
)
887 this.deleteTaskFunctionWorkerUsages(name
)
888 this.taskFunctions
.delete(name
)
889 this.workerChoiceStrategiesContext
?.syncWorkerChoiceStrategies(
890 this.getWorkerWorkerChoiceStrategies()
896 public listTaskFunctionsProperties (): TaskFunctionProperties
[] {
897 for (const workerNode
of this.workerNodes
) {
899 Array.isArray(workerNode
.info
.taskFunctionsProperties
) &&
900 workerNode
.info
.taskFunctionsProperties
.length
> 0
902 return workerNode
.info
.taskFunctionsProperties
909 * Gets task function strategy, if any.
911 * @param name - The task function name.
912 * @returns The task function worker choice strategy if the task function worker choice strategy is defined, `undefined` otherwise.
914 private readonly getTaskFunctionWorkerWorkerChoiceStrategy
= (
916 ): WorkerChoiceStrategy
| undefined => {
918 return this.listTaskFunctionsProperties().find(
919 (taskFunctionProperties
: TaskFunctionProperties
) =>
920 taskFunctionProperties
.name
=== name
926 * Gets worker node task function priority, if any.
928 * @param workerNodeKey - The worker node key.
929 * @param name - The task function name.
930 * @returns The task function worker choice priority if the task function worker choice priority is defined, `undefined` otherwise.
932 private readonly getWorkerNodeTaskFunctionPriority
= (
933 workerNodeKey
: number,
935 ): number | undefined => {
937 return this.getWorkerInfo(workerNodeKey
)?.taskFunctionsProperties
?.find(
938 (taskFunctionProperties
: TaskFunctionProperties
) =>
939 taskFunctionProperties
.name
=== name
945 * Gets the worker choice strategies registered in this pool.
947 * @returns The worker choice strategies.
949 private readonly getWorkerWorkerChoiceStrategies
=
950 (): Set
<WorkerChoiceStrategy
> => {
952 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
953 this.opts
.workerChoiceStrategy
!,
954 ...(this.listTaskFunctionsProperties()
956 (taskFunctionProperties
: TaskFunctionProperties
) =>
957 taskFunctionProperties
.strategy
960 (strategy
: WorkerChoiceStrategy
| undefined) => strategy
!= null
961 ) as WorkerChoiceStrategy
[])
966 public async setDefaultTaskFunction (name
: string): Promise
<boolean> {
967 return await this.sendTaskFunctionOperationToWorkers({
968 taskFunctionOperation
: 'default',
969 taskFunctionProperties
: buildTaskFunctionProperties(
971 this.taskFunctions
.get(name
)
976 private deleteTaskFunctionWorkerUsages (name
: string): void {
977 for (const workerNode
of this.workerNodes
) {
978 workerNode
.deleteTaskFunctionWorkerUsage(name
)
982 private shallExecuteTask (workerNodeKey
: number): boolean {
984 this.tasksQueueSize(workerNodeKey
) === 0 &&
985 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
<
986 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
987 this.opts
.tasksQueueOptions
!.concurrency
!
992 public async execute (
995 transferList
?: readonly TransferListItem
[]
996 ): Promise
<Response
> {
997 return await new Promise
<Response
>((resolve
, reject
) => {
999 reject(new Error('Cannot execute a task on not started pool'))
1002 if (this.destroying
) {
1003 reject(new Error('Cannot execute a task on destroying pool'))
1006 if (name
!= null && typeof name
!== 'string') {
1007 reject(new TypeError('name argument must be a string'))
1012 typeof name
=== 'string' &&
1013 name
.trim().length
=== 0
1015 reject(new TypeError('name argument must not be an empty string'))
1018 if (transferList
!= null && !Array.isArray(transferList
)) {
1019 reject(new TypeError('transferList argument must be an array'))
1022 const timestamp
= performance
.now()
1023 const taskFunctionStrategy
=
1024 this.getTaskFunctionWorkerWorkerChoiceStrategy(name
)
1025 const workerNodeKey
= this.chooseWorkerNode(taskFunctionStrategy
)
1026 const task
: Task
<Data
> = {
1027 name
: name
?? DEFAULT_TASK_NAME
,
1028 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
1029 data
: data
?? ({} as Data
),
1030 priority
: this.getWorkerNodeTaskFunctionPriority(workerNodeKey
, name
),
1031 strategy
: taskFunctionStrategy
,
1034 taskId
: randomUUID()
1036 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1037 this.promiseResponseMap
.set(task
.taskId
!, {
1041 ...(this.emitter
!= null && {
1042 asyncResource
: new AsyncResource('poolifier:task', {
1043 triggerAsyncId
: this.emitter
.asyncId
,
1044 requireManualDestroy
: true
1049 this.opts
.enableTasksQueue
=== false ||
1050 (this.opts
.enableTasksQueue
=== true &&
1051 this.shallExecuteTask(workerNodeKey
))
1053 this.executeTask(workerNodeKey
, task
)
1055 this.enqueueTask(workerNodeKey
, task
)
1061 * Starts the minimum number of workers.
1063 private startMinimumNumberOfWorkers (): void {
1064 this.startingMinimumNumberOfWorkers
= true
1066 this.workerNodes
.reduce(
1067 (accumulator
, workerNode
) =>
1068 !workerNode
.info
.dynamic
? accumulator
+ 1 : accumulator
,
1070 ) < this.minimumNumberOfWorkers
1072 this.createAndSetupWorkerNode()
1074 this.startingMinimumNumberOfWorkers
= false
1078 public start (): void {
1080 throw new Error('Cannot start an already started pool')
1082 if (this.starting
) {
1083 throw new Error('Cannot start an already starting pool')
1085 if (this.destroying
) {
1086 throw new Error('Cannot start a destroying pool')
1088 this.starting
= true
1089 this.startMinimumNumberOfWorkers()
1090 this.starting
= false
1095 public async destroy (): Promise
<void> {
1096 if (!this.started
) {
1097 throw new Error('Cannot destroy an already destroyed pool')
1099 if (this.starting
) {
1100 throw new Error('Cannot destroy an starting pool')
1102 if (this.destroying
) {
1103 throw new Error('Cannot destroy an already destroying pool')
1105 this.destroying
= true
1107 this.workerNodes
.map(async (_workerNode
, workerNodeKey
) => {
1108 await this.destroyWorkerNode(workerNodeKey
)
1111 this.emitter
?.emit(PoolEvents
.destroy
, this.info
)
1112 this.emitter
?.emitDestroy()
1113 this.readyEventEmitted
= false
1114 this.destroying
= false
1115 this.started
= false
1118 private async sendKillMessageToWorker (workerNodeKey
: number): Promise
<void> {
1119 await new Promise
<void>((resolve
, reject
) => {
1120 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1121 if (this.workerNodes
[workerNodeKey
] == null) {
1125 const killMessageListener
= (message
: MessageValue
<Response
>): void => {
1126 this.checkMessageWorkerId(message
)
1127 if (message
.kill
=== 'success') {
1129 } else if (message
.kill
=== 'failure') {
1132 `Kill message handling failed on worker ${message.workerId}`
1137 // FIXME: should be registered only once
1138 this.registerWorkerMessageListener(workerNodeKey
, killMessageListener
)
1139 this.sendToWorker(workerNodeKey
, { kill
: true })
1144 * Terminates the worker node given its worker node key.
1146 * @param workerNodeKey - The worker node key.
1148 protected async destroyWorkerNode (workerNodeKey
: number): Promise
<void> {
1149 this.flagWorkerNodeAsNotReady(workerNodeKey
)
1150 const flushedTasks
= this.flushTasksQueue(workerNodeKey
)
1151 const workerNode
= this.workerNodes
[workerNodeKey
]
1152 await waitWorkerNodeEvents(
1156 this.opts
.tasksQueueOptions
?.tasksFinishedTimeout
??
1157 getDefaultTasksQueueOptions(
1158 this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
1159 ).tasksFinishedTimeout
1161 await this.sendKillMessageToWorker(workerNodeKey
)
1162 await workerNode
.terminate()
1166 * Setup hook to execute code before worker nodes are created in the abstract constructor.
1167 * Can be overridden.
1171 protected setupHook (): void {
1172 /* Intentionally empty */
1176 * Returns whether the worker is the main worker or not.
1178 * @returns `true` if the worker is the main worker, `false` otherwise.
1180 protected abstract isMain (): boolean
1183 * Hook executed before the worker task execution.
1184 * Can be overridden.
1186 * @param workerNodeKey - The worker node key.
1187 * @param task - The task to execute.
1189 protected beforeTaskExecutionHook (
1190 workerNodeKey
: number,
1193 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1194 if (this.workerNodes
[workerNodeKey
]?.usage
!= null) {
1195 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
1196 ++workerUsage
.tasks
.executing
1197 updateWaitTimeWorkerUsage(
1198 this.workerChoiceStrategiesContext
,
1204 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1205 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1206 this.workerNodes
[workerNodeKey
].getTaskFunctionWorkerUsage(task
.name
!) !=
1209 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1210 const taskFunctionWorkerUsage
= this.workerNodes
[
1212 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1213 ].getTaskFunctionWorkerUsage(task
.name
!)!
1214 ++taskFunctionWorkerUsage
.tasks
.executing
1215 updateWaitTimeWorkerUsage(
1216 this.workerChoiceStrategiesContext
,
1217 taskFunctionWorkerUsage
,
1224 * Hook executed after the worker task execution.
1225 * Can be overridden.
1227 * @param workerNodeKey - The worker node key.
1228 * @param message - The received message.
1230 protected afterTaskExecutionHook (
1231 workerNodeKey
: number,
1232 message
: MessageValue
<Response
>
1234 let needWorkerChoiceStrategyUpdate
= false
1235 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1236 if (this.workerNodes
[workerNodeKey
]?.usage
!= null) {
1237 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
1238 updateTaskStatisticsWorkerUsage(workerUsage
, message
)
1239 updateRunTimeWorkerUsage(
1240 this.workerChoiceStrategiesContext
,
1244 updateEluWorkerUsage(
1245 this.workerChoiceStrategiesContext
,
1249 needWorkerChoiceStrategyUpdate
= true
1252 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1253 this.workerNodes
[workerNodeKey
].getTaskFunctionWorkerUsage(
1254 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1255 message
.taskPerformance
!.name
1258 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1259 const taskFunctionWorkerUsage
= this.workerNodes
[
1261 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1262 ].getTaskFunctionWorkerUsage(message
.taskPerformance
!.name
)!
1263 updateTaskStatisticsWorkerUsage(taskFunctionWorkerUsage
, message
)
1264 updateRunTimeWorkerUsage(
1265 this.workerChoiceStrategiesContext
,
1266 taskFunctionWorkerUsage
,
1269 updateEluWorkerUsage(
1270 this.workerChoiceStrategiesContext
,
1271 taskFunctionWorkerUsage
,
1274 needWorkerChoiceStrategyUpdate
= true
1276 if (needWorkerChoiceStrategyUpdate
) {
1277 this.workerChoiceStrategiesContext
?.update(workerNodeKey
)
1282 * Whether the worker node shall update its task function worker usage or not.
1284 * @param workerNodeKey - The worker node key.
1285 * @returns `true` if the worker node shall update its task function worker usage, `false` otherwise.
1287 private shallUpdateTaskFunctionWorkerUsage (workerNodeKey
: number): boolean {
1288 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1290 workerInfo
!= null &&
1291 Array.isArray(workerInfo
.taskFunctionsProperties
) &&
1292 workerInfo
.taskFunctionsProperties
.length
> 2
1297 * Chooses a worker node for the next task.
1299 * @param workerChoiceStrategy - The worker choice strategy.
1300 * @returns The chosen worker node key
1302 private chooseWorkerNode (
1303 workerChoiceStrategy
?: WorkerChoiceStrategy
1305 if (this.shallCreateDynamicWorker()) {
1306 const workerNodeKey
= this.createAndSetupDynamicWorkerNode()
1308 this.workerChoiceStrategiesContext
?.getPolicy().dynamicWorkerUsage
===
1311 return workerNodeKey
1314 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1315 return this.workerChoiceStrategiesContext
!.execute(workerChoiceStrategy
)
1319 * Conditions for dynamic worker creation.
1321 * @returns Whether to create a dynamic worker or not.
1323 protected abstract shallCreateDynamicWorker (): boolean
1326 * Sends a message to worker given its worker node key.
1328 * @param workerNodeKey - The worker node key.
1329 * @param message - The message.
1330 * @param transferList - The optional array of transferable objects.
1332 protected abstract sendToWorker (
1333 workerNodeKey
: number,
1334 message
: MessageValue
<Data
>,
1335 transferList
?: readonly TransferListItem
[]
1339 * Creates a new, completely set up worker node.
1341 * @returns New, completely set up worker node key.
1343 protected createAndSetupWorkerNode (): number {
1344 const workerNode
= this.createWorkerNode()
1345 workerNode
.registerWorkerEventHandler(
1347 this.opts
.onlineHandler
?? EMPTY_FUNCTION
1349 workerNode
.registerWorkerEventHandler(
1351 this.opts
.messageHandler
?? EMPTY_FUNCTION
1353 workerNode
.registerWorkerEventHandler(
1355 this.opts
.errorHandler
?? EMPTY_FUNCTION
1357 workerNode
.registerOnceWorkerEventHandler('error', (error
: Error) => {
1358 workerNode
.info
.ready
= false
1359 this.emitter
?.emit(PoolEvents
.error
, error
)
1363 this.opts
.restartWorkerOnError
=== true
1365 if (workerNode
.info
.dynamic
) {
1366 this.createAndSetupDynamicWorkerNode()
1367 } else if (!this.startingMinimumNumberOfWorkers
) {
1368 this.startMinimumNumberOfWorkers()
1374 this.opts
.enableTasksQueue
=== true
1376 this.redistributeQueuedTasks(this.workerNodes
.indexOf(workerNode
))
1378 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1379 workerNode
?.terminate().catch((error
: unknown
) => {
1380 this.emitter
?.emit(PoolEvents
.error
, error
)
1383 workerNode
.registerWorkerEventHandler(
1385 this.opts
.exitHandler
?? EMPTY_FUNCTION
1387 workerNode
.registerOnceWorkerEventHandler('exit', () => {
1388 this.removeWorkerNode(workerNode
)
1391 !this.startingMinimumNumberOfWorkers
&&
1394 this.startMinimumNumberOfWorkers()
1397 const workerNodeKey
= this.addWorkerNode(workerNode
)
1398 this.afterWorkerNodeSetup(workerNodeKey
)
1399 return workerNodeKey
1403 * Creates a new, completely set up dynamic worker node.
1405 * @returns New, completely set up dynamic worker node key.
1407 protected createAndSetupDynamicWorkerNode (): number {
1408 const workerNodeKey
= this.createAndSetupWorkerNode()
1409 this.registerWorkerMessageListener(workerNodeKey
, message
=> {
1410 this.checkMessageWorkerId(message
)
1411 const localWorkerNodeKey
= this.getWorkerNodeKeyByWorkerId(
1414 const workerUsage
= this.workerNodes
[localWorkerNodeKey
]?.usage
1415 // Kill message received from worker
1417 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
1418 (isKillBehavior(KillBehaviors
.SOFT
, message
.kill
) &&
1419 ((this.opts
.enableTasksQueue
=== false &&
1420 workerUsage
.tasks
.executing
=== 0) ||
1421 (this.opts
.enableTasksQueue
=== true &&
1422 workerUsage
.tasks
.executing
=== 0 &&
1423 this.tasksQueueSize(localWorkerNodeKey
) === 0)))
1425 // Flag the worker node as not ready immediately
1426 this.flagWorkerNodeAsNotReady(localWorkerNodeKey
)
1427 this.destroyWorkerNode(localWorkerNodeKey
).catch((error
: unknown
) => {
1428 this.emitter
?.emit(PoolEvents
.error
, error
)
1432 this.sendToWorker(workerNodeKey
, {
1435 if (this.taskFunctions
.size
> 0) {
1436 for (const [taskFunctionName
, taskFunctionObject
] of this.taskFunctions
) {
1437 this.sendTaskFunctionOperationToWorker(workerNodeKey
, {
1438 taskFunctionOperation
: 'add',
1439 taskFunctionProperties
: buildTaskFunctionProperties(
1443 taskFunction
: taskFunctionObject
.taskFunction
.toString()
1444 }).catch((error
: unknown
) => {
1445 this.emitter
?.emit(PoolEvents
.error
, error
)
1449 const workerNode
= this.workerNodes
[workerNodeKey
]
1450 workerNode
.info
.dynamic
= true
1452 this.workerChoiceStrategiesContext
?.getPolicy().dynamicWorkerReady
===
1454 this.workerChoiceStrategiesContext
?.getPolicy().dynamicWorkerUsage
===
1457 workerNode
.info
.ready
= true
1459 this.checkAndEmitDynamicWorkerCreationEvents()
1460 return workerNodeKey
1464 * Registers a listener callback on the worker given its worker node key.
1466 * @param workerNodeKey - The worker node key.
1467 * @param listener - The message listener callback.
1469 protected abstract registerWorkerMessageListener
<
1470 Message
extends Data
| Response
1472 workerNodeKey
: number,
1473 listener
: (message
: MessageValue
<Message
>) => void
1477 * Registers once a listener callback on the worker given its worker node key.
1479 * @param workerNodeKey - The worker node key.
1480 * @param listener - The message listener callback.
1482 protected abstract registerOnceWorkerMessageListener
<
1483 Message
extends Data
| Response
1485 workerNodeKey
: number,
1486 listener
: (message
: MessageValue
<Message
>) => void
1490 * Deregisters a listener callback on the worker given its worker node key.
1492 * @param workerNodeKey - The worker node key.
1493 * @param listener - The message listener callback.
1495 protected abstract deregisterWorkerMessageListener
<
1496 Message
extends Data
| Response
1498 workerNodeKey
: number,
1499 listener
: (message
: MessageValue
<Message
>) => void
1503 * Method hooked up after a worker node has been newly created.
1504 * Can be overridden.
1506 * @param workerNodeKey - The newly created worker node key.
1508 protected afterWorkerNodeSetup (workerNodeKey
: number): void {
1509 // Listen to worker messages.
1510 this.registerWorkerMessageListener(
1512 this.workerMessageListener
1514 // Send the startup message to worker.
1515 this.sendStartupMessageToWorker(workerNodeKey
)
1516 // Send the statistics message to worker.
1517 this.sendStatisticsMessageToWorker(workerNodeKey
)
1518 if (this.opts
.enableTasksQueue
=== true) {
1519 if (this.opts
.tasksQueueOptions
?.taskStealing
=== true) {
1520 this.workerNodes
[workerNodeKey
].on(
1522 this.handleWorkerNodeIdleEvent
1525 if (this.opts
.tasksQueueOptions
?.tasksStealingOnBackPressure
=== true) {
1526 this.workerNodes
[workerNodeKey
].on(
1528 this.handleWorkerNodeBackPressureEvent
1535 * Sends the startup message to worker given its worker node key.
1537 * @param workerNodeKey - The worker node key.
1539 protected abstract sendStartupMessageToWorker (workerNodeKey
: number): void
1542 * Sends the statistics message to worker given its worker node key.
1544 * @param workerNodeKey - The worker node key.
1546 private sendStatisticsMessageToWorker (workerNodeKey
: number): void {
1547 this.sendToWorker(workerNodeKey
, {
1550 this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
1551 .runTime
.aggregate
?? false,
1553 this.workerChoiceStrategiesContext
?.getTaskStatisticsRequirements()
1554 .elu
.aggregate
?? false
1559 private cannotStealTask (): boolean {
1560 return this.workerNodes
.length
<= 1 || this.info
.queuedTasks
=== 0
1563 private handleTask (workerNodeKey
: number, task
: Task
<Data
>): void {
1564 if (this.shallExecuteTask(workerNodeKey
)) {
1565 this.executeTask(workerNodeKey
, task
)
1567 this.enqueueTask(workerNodeKey
, task
)
1571 private redistributeQueuedTasks (workerNodeKey
: number): void {
1572 if (workerNodeKey
=== -1 || this.cannotStealTask()) {
1575 while (this.tasksQueueSize(workerNodeKey
) > 0) {
1576 const destinationWorkerNodeKey
= this.workerNodes
.reduce(
1577 (minWorkerNodeKey
, workerNode
, workerNodeKey
, workerNodes
) => {
1578 return workerNode
.info
.ready
&&
1579 workerNode
.usage
.tasks
.queued
<
1580 workerNodes
[minWorkerNodeKey
].usage
.tasks
.queued
1587 destinationWorkerNodeKey
,
1588 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1589 this.dequeueTask(workerNodeKey
)!
1594 private updateTaskStolenStatisticsWorkerUsage (
1595 workerNodeKey
: number,
1598 const workerNode
= this.workerNodes
[workerNodeKey
]
1599 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1600 if (workerNode
?.usage
!= null) {
1601 ++workerNode
.usage
.tasks
.stolen
1604 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1605 workerNode
.getTaskFunctionWorkerUsage(taskName
) != null
1607 const taskFunctionWorkerUsage
=
1608 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1609 workerNode
.getTaskFunctionWorkerUsage(taskName
)!
1610 ++taskFunctionWorkerUsage
.tasks
.stolen
1614 private updateTaskSequentiallyStolenStatisticsWorkerUsage (
1615 workerNodeKey
: number
1617 const workerNode
= this.workerNodes
[workerNodeKey
]
1618 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1619 if (workerNode
?.usage
!= null) {
1620 ++workerNode
.usage
.tasks
.sequentiallyStolen
1624 private updateTaskSequentiallyStolenStatisticsTaskFunctionWorkerUsage (
1625 workerNodeKey
: number,
1628 const workerNode
= this.workerNodes
[workerNodeKey
]
1630 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1631 workerNode
.getTaskFunctionWorkerUsage(taskName
) != null
1633 const taskFunctionWorkerUsage
=
1634 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1635 workerNode
.getTaskFunctionWorkerUsage(taskName
)!
1636 ++taskFunctionWorkerUsage
.tasks
.sequentiallyStolen
1640 private resetTaskSequentiallyStolenStatisticsWorkerUsage (
1641 workerNodeKey
: number
1643 const workerNode
= this.workerNodes
[workerNodeKey
]
1644 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1645 if (workerNode
?.usage
!= null) {
1646 workerNode
.usage
.tasks
.sequentiallyStolen
= 0
1650 private resetTaskSequentiallyStolenStatisticsTaskFunctionWorkerUsage (
1651 workerNodeKey
: number,
1654 const workerNode
= this.workerNodes
[workerNodeKey
]
1656 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1657 workerNode
.getTaskFunctionWorkerUsage(taskName
) != null
1659 const taskFunctionWorkerUsage
=
1660 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1661 workerNode
.getTaskFunctionWorkerUsage(taskName
)!
1662 taskFunctionWorkerUsage
.tasks
.sequentiallyStolen
= 0
1666 private readonly handleWorkerNodeIdleEvent
= (
1667 eventDetail
: WorkerNodeEventDetail
,
1668 previousStolenTask
?: Task
<Data
>
1670 const { workerNodeKey
} = eventDetail
1671 if (workerNodeKey
== null) {
1673 "WorkerNode event detail 'workerNodeKey' property must be defined"
1676 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1678 this.cannotStealTask() ||
1679 (this.info
.stealingWorkerNodes
?? 0) >
1680 Math.floor(this.workerNodes
.length
/ 2)
1682 if (workerInfo
!= null && previousStolenTask
!= null) {
1683 workerInfo
.stealing
= false
1687 const workerNodeTasksUsage
= this.workerNodes
[workerNodeKey
].usage
.tasks
1689 workerInfo
!= null &&
1690 previousStolenTask
!= null &&
1691 workerNodeTasksUsage
.sequentiallyStolen
> 0 &&
1692 (workerNodeTasksUsage
.executing
> 0 ||
1693 this.tasksQueueSize(workerNodeKey
) > 0)
1695 workerInfo
.stealing
= false
1696 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1697 for (const taskFunctionProperties
of workerInfo
.taskFunctionsProperties
!) {
1698 this.resetTaskSequentiallyStolenStatisticsTaskFunctionWorkerUsage(
1700 taskFunctionProperties
.name
1703 this.resetTaskSequentiallyStolenStatisticsWorkerUsage(workerNodeKey
)
1706 if (workerInfo
== null) {
1708 `Worker node with key '${workerNodeKey}' not found in pool`
1711 workerInfo
.stealing
= true
1712 const stolenTask
= this.workerNodeStealTask(workerNodeKey
)
1714 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1717 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1718 const taskFunctionTasksWorkerUsage
= this.workerNodes
[
1720 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1721 ].getTaskFunctionWorkerUsage(stolenTask
.name
!)!.tasks
1723 taskFunctionTasksWorkerUsage
.sequentiallyStolen
=== 0 ||
1724 (previousStolenTask
!= null &&
1725 previousStolenTask
.name
=== stolenTask
.name
&&
1726 taskFunctionTasksWorkerUsage
.sequentiallyStolen
> 0)
1728 this.updateTaskSequentiallyStolenStatisticsTaskFunctionWorkerUsage(
1730 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1734 this.resetTaskSequentiallyStolenStatisticsTaskFunctionWorkerUsage(
1736 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1741 sleep(exponentialDelay(workerNodeTasksUsage
.sequentiallyStolen
))
1743 this.handleWorkerNodeIdleEvent(eventDetail
, stolenTask
)
1746 .catch((error
: unknown
) => {
1747 this.emitter
?.emit(PoolEvents
.error
, error
)
1751 private readonly workerNodeStealTask
= (
1752 workerNodeKey
: number
1753 ): Task
<Data
> | undefined => {
1754 const workerNodes
= this.workerNodes
1757 (workerNodeA
, workerNodeB
) =>
1758 workerNodeB
.usage
.tasks
.queued
- workerNodeA
.usage
.tasks
.queued
1760 const sourceWorkerNode
= workerNodes
.find(
1761 (sourceWorkerNode
, sourceWorkerNodeKey
) =>
1762 sourceWorkerNode
.info
.ready
&&
1763 !sourceWorkerNode
.info
.stealing
&&
1764 sourceWorkerNodeKey
!== workerNodeKey
&&
1765 sourceWorkerNode
.usage
.tasks
.queued
> 0
1767 if (sourceWorkerNode
!= null) {
1768 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1769 const task
= sourceWorkerNode
.dequeueTask(1)!
1770 this.handleTask(workerNodeKey
, task
)
1771 this.updateTaskSequentiallyStolenStatisticsWorkerUsage(workerNodeKey
)
1772 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1773 this.updateTaskStolenStatisticsWorkerUsage(workerNodeKey
, task
.name
!)
1778 private readonly handleWorkerNodeBackPressureEvent
= (
1779 eventDetail
: WorkerNodeEventDetail
1782 this.cannotStealTask() ||
1783 (this.info
.stealingWorkerNodes
?? 0) >
1784 Math.floor(this.workerNodes
.length
/ 2)
1788 const { workerId
} = eventDetail
1789 const sizeOffset
= 1
1790 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1791 if (this.opts
.tasksQueueOptions
!.size
! <= sizeOffset
) {
1794 const sourceWorkerNode
=
1795 this.workerNodes
[this.getWorkerNodeKeyByWorkerId(workerId
)]
1796 const workerNodes
= this.workerNodes
1799 (workerNodeA
, workerNodeB
) =>
1800 workerNodeA
.usage
.tasks
.queued
- workerNodeB
.usage
.tasks
.queued
1802 for (const [workerNodeKey
, workerNode
] of workerNodes
.entries()) {
1804 sourceWorkerNode
.usage
.tasks
.queued
> 0 &&
1805 workerNode
.info
.ready
&&
1806 !workerNode
.info
.stealing
&&
1807 workerNode
.info
.id
!== workerId
&&
1808 workerNode
.usage
.tasks
.queued
<
1809 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1810 this.opts
.tasksQueueOptions
!.size
! - sizeOffset
1812 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1813 if (workerInfo
== null) {
1815 `Worker node with key '${workerNodeKey}' not found in pool`
1818 workerInfo
.stealing
= true
1819 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1820 const task
= sourceWorkerNode
.dequeueTask(1)!
1821 this.handleTask(workerNodeKey
, task
)
1822 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1823 this.updateTaskStolenStatisticsWorkerUsage(workerNodeKey
, task
.name
!)
1824 workerInfo
.stealing
= false
1830 * This method is the message listener registered on each worker.
1832 protected readonly workerMessageListener
= (
1833 message
: MessageValue
<Response
>
1835 this.checkMessageWorkerId(message
)
1836 const { workerId
, ready
, taskId
, taskFunctionsProperties
} = message
1837 if (ready
!= null && taskFunctionsProperties
!= null) {
1838 // Worker ready response received from worker
1839 this.handleWorkerReadyResponse(message
)
1840 } else if (taskFunctionsProperties
!= null) {
1841 // Task function properties message received from worker
1842 const workerInfo
= this.getWorkerInfo(
1843 this.getWorkerNodeKeyByWorkerId(workerId
)
1845 if (workerInfo
!= null) {
1846 workerInfo
.taskFunctionsProperties
= taskFunctionsProperties
1848 } else if (taskId
!= null) {
1849 // Task execution response received from worker
1850 this.handleTaskExecutionResponse(message
)
1854 private checkAndEmitReadyEvent (): void {
1855 if (!this.readyEventEmitted
&& this.ready
) {
1856 this.emitter
?.emit(PoolEvents
.ready
, this.info
)
1857 this.readyEventEmitted
= true
1861 private handleWorkerReadyResponse (message
: MessageValue
<Response
>): void {
1862 const { workerId
, ready
, taskFunctionsProperties
} = message
1863 if (ready
== null || !ready
) {
1864 throw new Error(`Worker ${workerId} failed to initialize`)
1867 this.workerNodes
[this.getWorkerNodeKeyByWorkerId(workerId
)]
1868 workerNode
.info
.ready
= ready
1869 workerNode
.info
.taskFunctionsProperties
= taskFunctionsProperties
1870 this.checkAndEmitReadyEvent()
1873 private handleTaskExecutionResponse (message
: MessageValue
<Response
>): void {
1874 const { workerId
, taskId
, workerError
, data
} = message
1875 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1876 const promiseResponse
= this.promiseResponseMap
.get(taskId
!)
1877 if (promiseResponse
!= null) {
1878 const { resolve
, reject
, workerNodeKey
, asyncResource
} = promiseResponse
1879 const workerNode
= this.workerNodes
[workerNodeKey
]
1880 if (workerError
!= null) {
1881 this.emitter
?.emit(PoolEvents
.taskError
, workerError
)
1882 asyncResource
!= null
1883 ? asyncResource
.runInAsyncScope(
1888 : reject(workerError
.message
)
1890 asyncResource
!= null
1891 ? asyncResource
.runInAsyncScope(resolve
, this.emitter
, data
)
1892 : resolve(data
as Response
)
1894 asyncResource
?.emitDestroy()
1895 this.afterTaskExecutionHook(workerNodeKey
, message
)
1896 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1897 this.promiseResponseMap
.delete(taskId
!)
1898 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1899 workerNode
?.emit('taskFinished', taskId
)
1901 this.opts
.enableTasksQueue
=== true &&
1903 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1906 const workerNodeTasksUsage
= workerNode
.usage
.tasks
1908 this.tasksQueueSize(workerNodeKey
) > 0 &&
1909 workerNodeTasksUsage
.executing
<
1910 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1911 this.opts
.tasksQueueOptions
!.concurrency
!
1913 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1914 this.executeTask(workerNodeKey
, this.dequeueTask(workerNodeKey
)!)
1917 workerNodeTasksUsage
.executing
=== 0 &&
1918 this.tasksQueueSize(workerNodeKey
) === 0 &&
1919 workerNodeTasksUsage
.sequentiallyStolen
=== 0
1921 workerNode
.emit('idle', {
1930 private checkAndEmitTaskExecutionEvents (): void {
1932 this.emitter
?.emit(PoolEvents
.busy
, this.info
)
1936 private checkAndEmitTaskQueuingEvents (): void {
1937 if (this.hasBackPressure()) {
1938 this.emitter
?.emit(PoolEvents
.backPressure
, this.info
)
1943 * Emits dynamic worker creation events.
1945 protected abstract checkAndEmitDynamicWorkerCreationEvents (): void
1948 * Gets the worker information given its worker node key.
1950 * @param workerNodeKey - The worker node key.
1951 * @returns The worker information.
1953 protected getWorkerInfo (workerNodeKey
: number): WorkerInfo
| undefined {
1954 return this.workerNodes
[workerNodeKey
]?.info
1958 * Creates a worker node.
1960 * @returns The created worker node.
1962 private createWorkerNode (): IWorkerNode
<Worker
, Data
> {
1963 const workerNode
= new WorkerNode
<Worker
, Data
>(
1968 workerOptions
: this.opts
.workerOptions
,
1969 tasksQueueBackPressureSize
:
1970 this.opts
.tasksQueueOptions
?.size
??
1971 getDefaultTasksQueueOptions(
1972 this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
1974 tasksQueueBucketSize
:
1975 (this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
) * 2
1978 // Flag the worker node as ready at pool startup.
1979 if (this.starting
) {
1980 workerNode
.info
.ready
= true
1986 * Adds the given worker node in the pool worker nodes.
1988 * @param workerNode - The worker node.
1989 * @returns The added worker node key.
1990 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the added worker node is not found.
1992 private addWorkerNode (workerNode
: IWorkerNode
<Worker
, Data
>): number {
1993 this.workerNodes
.push(workerNode
)
1994 const workerNodeKey
= this.workerNodes
.indexOf(workerNode
)
1995 if (workerNodeKey
=== -1) {
1996 throw new Error('Worker added not found in worker nodes')
1998 return workerNodeKey
2001 private checkAndEmitEmptyEvent (): void {
2003 this.emitter
?.emit(PoolEvents
.empty
, this.info
)
2004 this.readyEventEmitted
= false
2009 * Removes the worker node from the pool worker nodes.
2011 * @param workerNode - The worker node.
2013 private removeWorkerNode (workerNode
: IWorkerNode
<Worker
, Data
>): void {
2014 const workerNodeKey
= this.workerNodes
.indexOf(workerNode
)
2015 if (workerNodeKey
!== -1) {
2016 this.workerNodes
.splice(workerNodeKey
, 1)
2017 this.workerChoiceStrategiesContext
?.remove(workerNodeKey
)
2019 this.checkAndEmitEmptyEvent()
2022 protected flagWorkerNodeAsNotReady (workerNodeKey
: number): void {
2023 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
2024 if (workerInfo
!= null) {
2025 workerInfo
.ready
= false
2029 private hasBackPressure (): boolean {
2031 this.opts
.enableTasksQueue
=== true &&
2032 this.workerNodes
.findIndex(
2033 workerNode
=> !workerNode
.hasBackPressure()
2039 * Executes the given task on the worker given its worker node key.
2041 * @param workerNodeKey - The worker node key.
2042 * @param task - The task to execute.
2044 private executeTask (workerNodeKey
: number, task
: Task
<Data
>): void {
2045 this.beforeTaskExecutionHook(workerNodeKey
, task
)
2046 this.sendToWorker(workerNodeKey
, task
, task
.transferList
)
2047 this.checkAndEmitTaskExecutionEvents()
2050 private enqueueTask (workerNodeKey
: number, task
: Task
<Data
>): number {
2051 const tasksQueueSize
= this.workerNodes
[workerNodeKey
].enqueueTask(task
)
2052 this.checkAndEmitTaskQueuingEvents()
2053 return tasksQueueSize
2056 private dequeueTask (
2057 workerNodeKey
: number,
2059 ): Task
<Data
> | undefined {
2060 return this.workerNodes
[workerNodeKey
].dequeueTask(bucket
)
2063 private tasksQueueSize (workerNodeKey
: number): number {
2064 return this.workerNodes
[workerNodeKey
].tasksQueueSize()
2067 protected flushTasksQueue (workerNodeKey
: number): number {
2068 let flushedTasks
= 0
2069 while (this.tasksQueueSize(workerNodeKey
) > 0) {
2070 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2071 this.executeTask(workerNodeKey
, this.dequeueTask(workerNodeKey
)!)
2074 this.workerNodes
[workerNodeKey
].clearTasksQueue()
2078 private flushTasksQueues (): void {
2079 for (const workerNodeKey
of this.workerNodes
.keys()) {
2080 this.flushTasksQueue(workerNodeKey
)