1 import { randomUUID
} from
'node:crypto'
2 import { performance
} from
'node:perf_hooks'
3 import type { TransferListItem
} from
'node:worker_threads'
4 import { EventEmitterAsyncResource
} from
'node:events'
5 import { AsyncResource
} from
'node:async_hooks'
8 PromiseResponseWrapper
,
10 } from
'../utility-types'
13 DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
,
25 import { KillBehaviors
} from
'../worker/worker-options'
26 import type { TaskFunction
} from
'../worker/task-functions'
34 type TasksQueueOptions
41 WorkerNodeEventDetail
,
46 type MeasurementStatisticsRequirements
,
48 WorkerChoiceStrategies
,
49 type WorkerChoiceStrategy
,
50 type WorkerChoiceStrategyOptions
51 } from
'./selection-strategies/selection-strategies-types'
52 import { WorkerChoiceStrategyContext
} from
'./selection-strategies/worker-choice-strategy-context'
53 import { version
} from
'./version'
54 import { WorkerNode
} from
'./worker-node'
57 checkValidTasksQueueOptions
,
58 checkValidWorkerChoiceStrategy
,
59 updateMeasurementStatistics
63 * Base class that implements some shared logic for all poolifier pools.
65 * @typeParam Worker - Type of worker which manages this pool.
66 * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
67 * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
69 export abstract class AbstractPool
<
70 Worker
extends IWorker
,
73 > implements IPool
<Worker
, Data
, Response
> {
75 public readonly workerNodes
: Array<IWorkerNode
<Worker
, Data
>> = []
78 public emitter
?: EventEmitterAsyncResource
81 * Dynamic pool maximum size property placeholder.
83 protected readonly max
?: number
86 * The task execution response promise map:
87 * - `key`: The message id of each submitted task.
88 * - `value`: An object that contains the worker, the execution response promise resolve and reject callbacks.
90 * When we receive a message from the worker, we get a map entry with the promise resolve/reject bound to the message id.
92 protected promiseResponseMap
: Map
<string, PromiseResponseWrapper
<Response
>> =
93 new Map
<string, PromiseResponseWrapper
<Response
>>()
96 * Worker choice strategy context referencing a worker choice algorithm implementation.
98 protected workerChoiceStrategyContext
: WorkerChoiceStrategyContext
<
105 * The task functions added at runtime map:
106 * - `key`: The task function name.
107 * - `value`: The task function itself.
109 private readonly taskFunctions
: Map
<string, TaskFunction
<Data
, Response
>>
112 * Whether the pool is started or not.
114 private started
: boolean
116 * Whether the pool is starting or not.
118 private starting
: boolean
120 * Whether the pool is destroying or not.
122 private destroying
: boolean
124 * Whether the pool ready event has been emitted or not.
126 private readyEventEmitted
: boolean
128 * The start timestamp of the pool.
130 private readonly startTimestamp
133 * Constructs a new poolifier pool.
135 * @param numberOfWorkers - Number of workers that this pool should manage.
136 * @param filePath - Path to the worker file.
137 * @param opts - Options for the pool.
140 protected readonly numberOfWorkers
: number,
141 protected readonly filePath
: string,
142 protected readonly opts
: PoolOptions
<Worker
>
144 if (!this.isMain()) {
146 'Cannot start a pool from a worker with the same type as the pool'
149 checkFilePath(this.filePath
)
150 this.checkNumberOfWorkers(this.numberOfWorkers
)
151 this.checkPoolOptions(this.opts
)
153 this.chooseWorkerNode
= this.chooseWorkerNode
.bind(this)
154 this.executeTask
= this.executeTask
.bind(this)
155 this.enqueueTask
= this.enqueueTask
.bind(this)
157 if (this.opts
.enableEvents
=== true) {
158 this.initializeEventEmitter()
160 this.workerChoiceStrategyContext
= new WorkerChoiceStrategyContext
<
166 this.opts
.workerChoiceStrategy
,
167 this.opts
.workerChoiceStrategyOptions
172 this.taskFunctions
= new Map
<string, TaskFunction
<Data
, Response
>>()
175 this.starting
= false
176 this.destroying
= false
177 this.readyEventEmitted
= false
178 if (this.opts
.startWorkers
=== true) {
182 this.startTimestamp
= performance
.now()
185 private checkNumberOfWorkers (numberOfWorkers
: number): void {
186 if (numberOfWorkers
== null) {
188 'Cannot instantiate a pool without specifying the number of workers'
190 } else if (!Number.isSafeInteger(numberOfWorkers
)) {
192 'Cannot instantiate a pool with a non safe integer number of workers'
194 } else if (numberOfWorkers
< 0) {
195 throw new RangeError(
196 'Cannot instantiate a pool with a negative number of workers'
198 } else if (this.type === PoolTypes
.fixed
&& numberOfWorkers
=== 0) {
199 throw new RangeError('Cannot instantiate a fixed pool with zero worker')
203 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
204 if (isPlainObject(opts
)) {
205 this.opts
.startWorkers
= opts
.startWorkers
?? true
206 checkValidWorkerChoiceStrategy(
207 opts
.workerChoiceStrategy
as WorkerChoiceStrategy
209 this.opts
.workerChoiceStrategy
=
210 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
211 this.checkValidWorkerChoiceStrategyOptions(
212 opts
.workerChoiceStrategyOptions
as WorkerChoiceStrategyOptions
214 this.opts
.workerChoiceStrategyOptions
= {
215 ...DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
,
216 ...opts
.workerChoiceStrategyOptions
218 this.opts
.restartWorkerOnError
= opts
.restartWorkerOnError
?? true
219 this.opts
.enableEvents
= opts
.enableEvents
?? true
220 this.opts
.enableTasksQueue
= opts
.enableTasksQueue
?? false
221 if (this.opts
.enableTasksQueue
) {
222 checkValidTasksQueueOptions(opts
.tasksQueueOptions
as TasksQueueOptions
)
223 this.opts
.tasksQueueOptions
= this.buildTasksQueueOptions(
224 opts
.tasksQueueOptions
as TasksQueueOptions
228 throw new TypeError('Invalid pool options: must be a plain object')
232 private checkValidWorkerChoiceStrategyOptions (
233 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
236 workerChoiceStrategyOptions
!= null &&
237 !isPlainObject(workerChoiceStrategyOptions
)
240 'Invalid worker choice strategy options: must be a plain object'
244 workerChoiceStrategyOptions
?.retries
!= null &&
245 !Number.isSafeInteger(workerChoiceStrategyOptions
.retries
)
248 'Invalid worker choice strategy options: retries must be an integer'
252 workerChoiceStrategyOptions
?.retries
!= null &&
253 workerChoiceStrategyOptions
.retries
< 0
255 throw new RangeError(
256 `Invalid worker choice strategy options: retries '${workerChoiceStrategyOptions.retries}' must be greater or equal than zero`
260 workerChoiceStrategyOptions
?.weights
!= null &&
261 Object.keys(workerChoiceStrategyOptions
.weights
).length
!== this.maxSize
264 'Invalid worker choice strategy options: must have a weight for each worker node'
268 workerChoiceStrategyOptions
?.measurement
!= null &&
269 !Object.values(Measurements
).includes(
270 workerChoiceStrategyOptions
.measurement
274 `Invalid worker choice strategy options: invalid measurement '${workerChoiceStrategyOptions.measurement}'`
279 private initializeEventEmitter (): void {
280 this.emitter
= new EventEmitterAsyncResource({
281 name
: `poolifier:${this.type}-${this.worker}-pool`
286 public get
info (): PoolInfo
{
291 started
: this.started
,
293 strategy
: this.opts
.workerChoiceStrategy
as WorkerChoiceStrategy
,
294 minSize
: this.minSize
,
295 maxSize
: this.maxSize
,
296 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
297 .runTime
.aggregate
&&
298 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
299 .waitTime
.aggregate
&& { utilization
: round(this.utilization
) }),
300 workerNodes
: this.workerNodes
.length
,
301 idleWorkerNodes
: this.workerNodes
.reduce(
302 (accumulator
, workerNode
) =>
303 workerNode
.usage
.tasks
.executing
=== 0
308 busyWorkerNodes
: this.workerNodes
.reduce(
309 (accumulator
, workerNode
) =>
310 workerNode
.usage
.tasks
.executing
> 0 ? accumulator
+ 1 : accumulator
,
313 executedTasks
: this.workerNodes
.reduce(
314 (accumulator
, workerNode
) =>
315 accumulator
+ workerNode
.usage
.tasks
.executed
,
318 executingTasks
: this.workerNodes
.reduce(
319 (accumulator
, workerNode
) =>
320 accumulator
+ workerNode
.usage
.tasks
.executing
,
323 ...(this.opts
.enableTasksQueue
=== true && {
324 queuedTasks
: this.workerNodes
.reduce(
325 (accumulator
, workerNode
) =>
326 accumulator
+ workerNode
.usage
.tasks
.queued
,
330 ...(this.opts
.enableTasksQueue
=== true && {
331 maxQueuedTasks
: this.workerNodes
.reduce(
332 (accumulator
, workerNode
) =>
333 accumulator
+ (workerNode
.usage
.tasks
?.maxQueued
?? 0),
337 ...(this.opts
.enableTasksQueue
=== true && {
338 backPressure
: this.hasBackPressure()
340 ...(this.opts
.enableTasksQueue
=== true && {
341 stolenTasks
: this.workerNodes
.reduce(
342 (accumulator
, workerNode
) =>
343 accumulator
+ workerNode
.usage
.tasks
.stolen
,
347 failedTasks
: this.workerNodes
.reduce(
348 (accumulator
, workerNode
) =>
349 accumulator
+ workerNode
.usage
.tasks
.failed
,
352 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
353 .runTime
.aggregate
&& {
357 ...this.workerNodes
.map(
358 workerNode
=> workerNode
.usage
.runTime
?.minimum
?? Infinity
364 ...this.workerNodes
.map(
365 workerNode
=> workerNode
.usage
.runTime
?.maximum
?? -Infinity
369 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
370 .runTime
.average
&& {
373 this.workerNodes
.reduce
<number[]>(
374 (accumulator
, workerNode
) =>
375 accumulator
.concat(workerNode
.usage
.runTime
.history
),
381 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
385 this.workerNodes
.reduce
<number[]>(
386 (accumulator
, workerNode
) =>
387 accumulator
.concat(workerNode
.usage
.runTime
.history
),
395 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
396 .waitTime
.aggregate
&& {
400 ...this.workerNodes
.map(
401 workerNode
=> workerNode
.usage
.waitTime
?.minimum
?? Infinity
407 ...this.workerNodes
.map(
408 workerNode
=> workerNode
.usage
.waitTime
?.maximum
?? -Infinity
412 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
413 .waitTime
.average
&& {
416 this.workerNodes
.reduce
<number[]>(
417 (accumulator
, workerNode
) =>
418 accumulator
.concat(workerNode
.usage
.waitTime
.history
),
424 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
425 .waitTime
.median
&& {
428 this.workerNodes
.reduce
<number[]>(
429 (accumulator
, workerNode
) =>
430 accumulator
.concat(workerNode
.usage
.waitTime
.history
),
442 * The pool readiness boolean status.
444 private get
ready (): boolean {
446 this.workerNodes
.reduce(
447 (accumulator
, workerNode
) =>
448 !workerNode
.info
.dynamic
&& workerNode
.info
.ready
457 * The approximate pool utilization.
459 * @returns The pool utilization.
461 private get
utilization (): number {
462 const poolTimeCapacity
=
463 (performance
.now() - this.startTimestamp
) * this.maxSize
464 const totalTasksRunTime
= this.workerNodes
.reduce(
465 (accumulator
, workerNode
) =>
466 accumulator
+ (workerNode
.usage
.runTime
?.aggregate
?? 0),
469 const totalTasksWaitTime
= this.workerNodes
.reduce(
470 (accumulator
, workerNode
) =>
471 accumulator
+ (workerNode
.usage
.waitTime
?.aggregate
?? 0),
474 return (totalTasksRunTime
+ totalTasksWaitTime
) / poolTimeCapacity
480 * If it is `'dynamic'`, it provides the `max` property.
482 protected abstract get
type (): PoolType
487 protected abstract get
worker (): WorkerType
490 * The pool minimum size.
492 protected get
minSize (): number {
493 return this.numberOfWorkers
497 * The pool maximum size.
499 protected get
maxSize (): number {
500 return this.max
?? this.numberOfWorkers
504 * Checks if the worker id sent in the received message from a worker is valid.
506 * @param message - The received message.
507 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the worker id is invalid.
509 private checkMessageWorkerId (message
: MessageValue
<Data
| Response
>): void {
510 if (message
.workerId
== null) {
511 throw new Error('Worker message received without worker id')
512 } else if (this.getWorkerNodeKeyByWorkerId(message
.workerId
) === -1) {
514 `Worker message received from unknown worker '${message.workerId}'`
520 * Gets the given worker its worker node key.
522 * @param worker - The worker.
523 * @returns The worker node key if found in the pool worker nodes, `-1` otherwise.
525 private getWorkerNodeKeyByWorker (worker
: Worker
): number {
526 return this.workerNodes
.findIndex(
527 workerNode
=> workerNode
.worker
=== worker
532 * Gets the worker node key given its worker id.
534 * @param workerId - The worker id.
535 * @returns The worker node key if the worker id is found in the pool worker nodes, `-1` otherwise.
537 private getWorkerNodeKeyByWorkerId (workerId
: number | undefined): number {
538 return this.workerNodes
.findIndex(
539 workerNode
=> workerNode
.info
.id
=== workerId
544 public setWorkerChoiceStrategy (
545 workerChoiceStrategy
: WorkerChoiceStrategy
,
546 workerChoiceStrategyOptions
?: WorkerChoiceStrategyOptions
548 checkValidWorkerChoiceStrategy(workerChoiceStrategy
)
549 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
550 this.workerChoiceStrategyContext
.setWorkerChoiceStrategy(
551 this.opts
.workerChoiceStrategy
553 if (workerChoiceStrategyOptions
!= null) {
554 this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
556 for (const [workerNodeKey
, workerNode
] of this.workerNodes
.entries()) {
557 workerNode
.resetUsage()
558 this.sendStatisticsMessageToWorker(workerNodeKey
)
563 public setWorkerChoiceStrategyOptions (
564 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
566 this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
567 this.opts
.workerChoiceStrategyOptions
= {
568 ...DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
,
569 ...workerChoiceStrategyOptions
571 this.workerChoiceStrategyContext
.setOptions(
572 this.opts
.workerChoiceStrategyOptions
577 public enableTasksQueue (
579 tasksQueueOptions
?: TasksQueueOptions
581 if (this.opts
.enableTasksQueue
=== true && !enable
) {
582 this.unsetTaskStealing()
583 this.unsetTasksStealingOnBackPressure()
584 this.flushTasksQueues()
586 this.opts
.enableTasksQueue
= enable
587 this.setTasksQueueOptions(tasksQueueOptions
as TasksQueueOptions
)
591 public setTasksQueueOptions (tasksQueueOptions
: TasksQueueOptions
): void {
592 if (this.opts
.enableTasksQueue
=== true) {
593 checkValidTasksQueueOptions(tasksQueueOptions
)
594 this.opts
.tasksQueueOptions
=
595 this.buildTasksQueueOptions(tasksQueueOptions
)
596 this.setTasksQueueSize(this.opts
.tasksQueueOptions
.size
as number)
597 if (this.opts
.tasksQueueOptions
.taskStealing
=== true) {
598 this.unsetTaskStealing()
599 this.setTaskStealing()
601 this.unsetTaskStealing()
603 if (this.opts
.tasksQueueOptions
.tasksStealingOnBackPressure
=== true) {
604 this.unsetTasksStealingOnBackPressure()
605 this.setTasksStealingOnBackPressure()
607 this.unsetTasksStealingOnBackPressure()
609 } else if (this.opts
.tasksQueueOptions
!= null) {
610 delete this.opts
.tasksQueueOptions
614 private buildTasksQueueOptions (
615 tasksQueueOptions
: TasksQueueOptions
616 ): TasksQueueOptions
{
619 size
: Math.pow(this.maxSize
, 2),
622 tasksStealingOnBackPressure
: true
628 private setTasksQueueSize (size
: number): void {
629 for (const workerNode
of this.workerNodes
) {
630 workerNode
.tasksQueueBackPressureSize
= size
634 private setTaskStealing (): void {
635 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
636 this.workerNodes
[workerNodeKey
].on(
638 this.handleIdleWorkerNodeEvent
643 private unsetTaskStealing (): void {
644 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
645 this.workerNodes
[workerNodeKey
].off(
647 this.handleIdleWorkerNodeEvent
652 private setTasksStealingOnBackPressure (): void {
653 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
654 this.workerNodes
[workerNodeKey
].on(
656 this.handleBackPressureEvent
661 private unsetTasksStealingOnBackPressure (): void {
662 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
663 this.workerNodes
[workerNodeKey
].off(
665 this.handleBackPressureEvent
671 * Whether the pool is full or not.
673 * The pool filling boolean status.
675 protected get
full (): boolean {
676 return this.workerNodes
.length
>= this.maxSize
680 * Whether the pool is busy or not.
682 * The pool busyness boolean status.
684 protected abstract get
busy (): boolean
687 * Whether worker nodes are executing concurrently their tasks quota or not.
689 * @returns Worker nodes busyness boolean status.
691 protected internalBusy (): boolean {
692 if (this.opts
.enableTasksQueue
=== true) {
694 this.workerNodes
.findIndex(
696 workerNode
.info
.ready
&&
697 workerNode
.usage
.tasks
.executing
<
698 (this.opts
.tasksQueueOptions
?.concurrency
as number)
703 this.workerNodes
.findIndex(
705 workerNode
.info
.ready
&& workerNode
.usage
.tasks
.executing
=== 0
710 private async sendTaskFunctionOperationToWorker (
711 workerNodeKey
: number,
712 message
: MessageValue
<Data
>
713 ): Promise
<boolean> {
714 return await new Promise
<boolean>((resolve
, reject
) => {
715 const taskFunctionOperationListener
= (
716 message
: MessageValue
<Response
>
718 this.checkMessageWorkerId(message
)
719 const workerId
= this.getWorkerInfo(workerNodeKey
).id
as number
721 message
.taskFunctionOperationStatus
!= null &&
722 message
.workerId
=== workerId
724 if (message
.taskFunctionOperationStatus
) {
726 } else if (!message
.taskFunctionOperationStatus
) {
729 `Task function operation '${
730 message.taskFunctionOperation as string
731 }' failed on worker ${message.workerId} with error: '${
732 message.workerError?.message as string
737 this.deregisterWorkerMessageListener(
738 this.getWorkerNodeKeyByWorkerId(message
.workerId
),
739 taskFunctionOperationListener
743 this.registerWorkerMessageListener(
745 taskFunctionOperationListener
747 this.sendToWorker(workerNodeKey
, message
)
751 private async sendTaskFunctionOperationToWorkers (
752 message
: MessageValue
<Data
>
753 ): Promise
<boolean> {
754 return await new Promise
<boolean>((resolve
, reject
) => {
755 const responsesReceived
= new Array<MessageValue
<Response
>>()
756 const taskFunctionOperationsListener
= (
757 message
: MessageValue
<Response
>
759 this.checkMessageWorkerId(message
)
760 if (message
.taskFunctionOperationStatus
!= null) {
761 responsesReceived
.push(message
)
762 if (responsesReceived
.length
=== this.workerNodes
.length
) {
764 responsesReceived
.every(
765 message
=> message
.taskFunctionOperationStatus
=== true
770 responsesReceived
.some(
771 message
=> message
.taskFunctionOperationStatus
=== false
774 const errorResponse
= responsesReceived
.find(
775 response
=> response
.taskFunctionOperationStatus
=== false
779 `Task function operation '${
780 message.taskFunctionOperation as string
781 }' failed on worker ${
782 errorResponse?.workerId as number
784 errorResponse?.workerError?.message as string
789 this.deregisterWorkerMessageListener(
790 this.getWorkerNodeKeyByWorkerId(message
.workerId
),
791 taskFunctionOperationsListener
796 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
797 this.registerWorkerMessageListener(
799 taskFunctionOperationsListener
801 this.sendToWorker(workerNodeKey
, message
)
807 public hasTaskFunction (name
: string): boolean {
808 for (const workerNode
of this.workerNodes
) {
810 Array.isArray(workerNode
.info
.taskFunctionNames
) &&
811 workerNode
.info
.taskFunctionNames
.includes(name
)
820 public async addTaskFunction (
822 fn
: TaskFunction
<Data
, Response
>
823 ): Promise
<boolean> {
824 if (typeof name
!== 'string') {
825 throw new TypeError('name argument must be a string')
827 if (typeof name
=== 'string' && name
.trim().length
=== 0) {
828 throw new TypeError('name argument must not be an empty string')
830 if (typeof fn
!== 'function') {
831 throw new TypeError('fn argument must be a function')
833 const opResult
= await this.sendTaskFunctionOperationToWorkers({
834 taskFunctionOperation
: 'add',
835 taskFunctionName
: name
,
836 taskFunction
: fn
.toString()
838 this.taskFunctions
.set(name
, fn
)
843 public async removeTaskFunction (name
: string): Promise
<boolean> {
844 if (!this.taskFunctions
.has(name
)) {
846 'Cannot remove a task function not handled on the pool side'
849 const opResult
= await this.sendTaskFunctionOperationToWorkers({
850 taskFunctionOperation
: 'remove',
851 taskFunctionName
: name
853 this.deleteTaskFunctionWorkerUsages(name
)
854 this.taskFunctions
.delete(name
)
859 public listTaskFunctionNames (): string[] {
860 for (const workerNode
of this.workerNodes
) {
862 Array.isArray(workerNode
.info
.taskFunctionNames
) &&
863 workerNode
.info
.taskFunctionNames
.length
> 0
865 return workerNode
.info
.taskFunctionNames
872 public async setDefaultTaskFunction (name
: string): Promise
<boolean> {
873 return await this.sendTaskFunctionOperationToWorkers({
874 taskFunctionOperation
: 'default',
875 taskFunctionName
: name
879 private deleteTaskFunctionWorkerUsages (name
: string): void {
880 for (const workerNode
of this.workerNodes
) {
881 workerNode
.deleteTaskFunctionWorkerUsage(name
)
885 private shallExecuteTask (workerNodeKey
: number): boolean {
887 this.tasksQueueSize(workerNodeKey
) === 0 &&
888 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
<
889 (this.opts
.tasksQueueOptions
?.concurrency
as number)
894 public async execute (
897 transferList
?: TransferListItem
[]
898 ): Promise
<Response
> {
899 return await new Promise
<Response
>((resolve
, reject
) => {
901 reject(new Error('Cannot execute a task on not started pool'))
904 if (this.destroying
) {
905 reject(new Error('Cannot execute a task on destroying pool'))
908 if (name
!= null && typeof name
!== 'string') {
909 reject(new TypeError('name argument must be a string'))
914 typeof name
=== 'string' &&
915 name
.trim().length
=== 0
917 reject(new TypeError('name argument must not be an empty string'))
920 if (transferList
!= null && !Array.isArray(transferList
)) {
921 reject(new TypeError('transferList argument must be an array'))
924 const timestamp
= performance
.now()
925 const workerNodeKey
= this.chooseWorkerNode()
926 const task
: Task
<Data
> = {
927 name
: name
?? DEFAULT_TASK_NAME
,
928 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
929 data
: data
?? ({} as Data
),
934 this.promiseResponseMap
.set(task
.taskId
as string, {
938 ...(this.emitter
!= null && {
939 asyncResource
: new AsyncResource('poolifier:task', {
940 triggerAsyncId
: this.emitter
.asyncId
,
941 requireManualDestroy
: true
946 this.opts
.enableTasksQueue
=== false ||
947 (this.opts
.enableTasksQueue
=== true &&
948 this.shallExecuteTask(workerNodeKey
))
950 this.executeTask(workerNodeKey
, task
)
952 this.enqueueTask(workerNodeKey
, task
)
958 public start (): void {
960 throw new Error('Cannot start an already started pool')
963 throw new Error('Cannot start an already starting pool')
965 if (this.destroying
) {
966 throw new Error('Cannot start a destroying pool')
970 this.workerNodes
.reduce(
971 (accumulator
, workerNode
) =>
972 !workerNode
.info
.dynamic
? accumulator
+ 1 : accumulator
,
974 ) < this.numberOfWorkers
976 this.createAndSetupWorkerNode()
978 this.starting
= false
983 public async destroy (): Promise
<void> {
985 throw new Error('Cannot destroy an already destroyed pool')
988 throw new Error('Cannot destroy an starting pool')
990 if (this.destroying
) {
991 throw new Error('Cannot destroy an already destroying pool')
993 this.destroying
= true
995 this.workerNodes
.map(async (_workerNode
, workerNodeKey
) => {
996 await this.destroyWorkerNode(workerNodeKey
)
999 this.emitter
?.emit(PoolEvents
.destroy
, this.info
)
1000 this.emitter
?.emitDestroy()
1001 this.emitter
?.removeAllListeners()
1002 this.readyEventEmitted
= false
1003 this.destroying
= false
1004 this.started
= false
1007 protected async sendKillMessageToWorker (
1008 workerNodeKey
: number
1010 await new Promise
<void>((resolve
, reject
) => {
1011 const killMessageListener
= (message
: MessageValue
<Response
>): void => {
1012 this.checkMessageWorkerId(message
)
1013 if (message
.kill
=== 'success') {
1015 } else if (message
.kill
=== 'failure') {
1018 `Kill message handling failed on worker ${
1019 message.workerId as number
1025 // FIXME: should be registered only once
1026 this.registerWorkerMessageListener(workerNodeKey
, killMessageListener
)
1027 this.sendToWorker(workerNodeKey
, { kill
: true })
1032 * Terminates the worker node given its worker node key.
1034 * @param workerNodeKey - The worker node key.
1036 protected abstract destroyWorkerNode (workerNodeKey
: number): Promise
<void>
1039 * Setup hook to execute code before worker nodes are created in the abstract constructor.
1040 * Can be overridden.
1044 protected setupHook (): void {
1045 /* Intentionally empty */
1049 * Should return whether the worker is the main worker or not.
1051 protected abstract isMain (): boolean
1054 * Hook executed before the worker task execution.
1055 * Can be overridden.
1057 * @param workerNodeKey - The worker node key.
1058 * @param task - The task to execute.
1060 protected beforeTaskExecutionHook (
1061 workerNodeKey
: number,
1064 if (this.workerNodes
[workerNodeKey
]?.usage
!= null) {
1065 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
1066 ++workerUsage
.tasks
.executing
1067 this.updateWaitTimeWorkerUsage(workerUsage
, task
)
1070 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1071 this.workerNodes
[workerNodeKey
].getTaskFunctionWorkerUsage(
1075 const taskFunctionWorkerUsage
= this.workerNodes
[
1077 ].getTaskFunctionWorkerUsage(task
.name
as string) as WorkerUsage
1078 ++taskFunctionWorkerUsage
.tasks
.executing
1079 this.updateWaitTimeWorkerUsage(taskFunctionWorkerUsage
, task
)
1084 * Hook executed after the worker task execution.
1085 * Can be overridden.
1087 * @param workerNodeKey - The worker node key.
1088 * @param message - The received message.
1090 protected afterTaskExecutionHook (
1091 workerNodeKey
: number,
1092 message
: MessageValue
<Response
>
1094 if (this.workerNodes
[workerNodeKey
]?.usage
!= null) {
1095 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
1096 this.updateTaskStatisticsWorkerUsage(workerUsage
, message
)
1097 this.updateRunTimeWorkerUsage(workerUsage
, message
)
1098 this.updateEluWorkerUsage(workerUsage
, message
)
1101 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1102 this.workerNodes
[workerNodeKey
].getTaskFunctionWorkerUsage(
1103 message
.taskPerformance
?.name
as string
1106 const taskFunctionWorkerUsage
= this.workerNodes
[
1108 ].getTaskFunctionWorkerUsage(
1109 message
.taskPerformance
?.name
as string
1111 this.updateTaskStatisticsWorkerUsage(taskFunctionWorkerUsage
, message
)
1112 this.updateRunTimeWorkerUsage(taskFunctionWorkerUsage
, message
)
1113 this.updateEluWorkerUsage(taskFunctionWorkerUsage
, message
)
1118 * Whether the worker node shall update its task function worker usage or not.
1120 * @param workerNodeKey - The worker node key.
1121 * @returns `true` if the worker node shall update its task function worker usage, `false` otherwise.
1123 private shallUpdateTaskFunctionWorkerUsage (workerNodeKey
: number): boolean {
1124 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1126 workerInfo
!= null &&
1127 Array.isArray(workerInfo
.taskFunctionNames
) &&
1128 workerInfo
.taskFunctionNames
.length
> 2
1132 private updateTaskStatisticsWorkerUsage (
1133 workerUsage
: WorkerUsage
,
1134 message
: MessageValue
<Response
>
1136 const workerTaskStatistics
= workerUsage
.tasks
1138 workerTaskStatistics
.executing
!= null &&
1139 workerTaskStatistics
.executing
> 0
1141 --workerTaskStatistics
.executing
1143 if (message
.workerError
== null) {
1144 ++workerTaskStatistics
.executed
1146 ++workerTaskStatistics
.failed
1150 private updateRunTimeWorkerUsage (
1151 workerUsage
: WorkerUsage
,
1152 message
: MessageValue
<Response
>
1154 if (message
.workerError
!= null) {
1157 updateMeasurementStatistics(
1158 workerUsage
.runTime
,
1159 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
,
1160 message
.taskPerformance
?.runTime
?? 0
1164 private updateWaitTimeWorkerUsage (
1165 workerUsage
: WorkerUsage
,
1168 const timestamp
= performance
.now()
1169 const taskWaitTime
= timestamp
- (task
.timestamp
?? timestamp
)
1170 updateMeasurementStatistics(
1171 workerUsage
.waitTime
,
1172 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().waitTime
,
1177 private updateEluWorkerUsage (
1178 workerUsage
: WorkerUsage
,
1179 message
: MessageValue
<Response
>
1181 if (message
.workerError
!= null) {
1184 const eluTaskStatisticsRequirements
: MeasurementStatisticsRequirements
=
1185 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
1186 updateMeasurementStatistics(
1187 workerUsage
.elu
.active
,
1188 eluTaskStatisticsRequirements
,
1189 message
.taskPerformance
?.elu
?.active
?? 0
1191 updateMeasurementStatistics(
1192 workerUsage
.elu
.idle
,
1193 eluTaskStatisticsRequirements
,
1194 message
.taskPerformance
?.elu
?.idle
?? 0
1196 if (eluTaskStatisticsRequirements
.aggregate
) {
1197 if (message
.taskPerformance
?.elu
!= null) {
1198 if (workerUsage
.elu
.utilization
!= null) {
1199 workerUsage
.elu
.utilization
=
1200 (workerUsage
.elu
.utilization
+
1201 message
.taskPerformance
.elu
.utilization
) /
1204 workerUsage
.elu
.utilization
= message
.taskPerformance
.elu
.utilization
1211 * Chooses a worker node for the next task.
1213 * The default worker choice strategy uses a round robin algorithm to distribute the tasks.
1215 * @returns The chosen worker node key
1217 private chooseWorkerNode (): number {
1218 if (this.shallCreateDynamicWorker()) {
1219 const workerNodeKey
= this.createAndSetupDynamicWorkerNode()
1221 this.workerChoiceStrategyContext
.getStrategyPolicy().dynamicWorkerUsage
1223 return workerNodeKey
1226 return this.workerChoiceStrategyContext
.execute()
1230 * Conditions for dynamic worker creation.
1232 * @returns Whether to create a dynamic worker or not.
1234 private shallCreateDynamicWorker (): boolean {
1235 return this.type === PoolTypes
.dynamic
&& !this.full
&& this.internalBusy()
1239 * Sends a message to worker given its worker node key.
1241 * @param workerNodeKey - The worker node key.
1242 * @param message - The message.
1243 * @param transferList - The optional array of transferable objects.
1245 protected abstract sendToWorker (
1246 workerNodeKey
: number,
1247 message
: MessageValue
<Data
>,
1248 transferList
?: TransferListItem
[]
1252 * Creates a new worker.
1254 * @returns Newly created worker.
1256 protected abstract createWorker (): Worker
1259 * Creates a new, completely set up worker node.
1261 * @returns New, completely set up worker node key.
1263 protected createAndSetupWorkerNode (): number {
1264 const worker
= this.createWorker()
1266 worker
.on('online', this.opts
.onlineHandler
?? EMPTY_FUNCTION
)
1267 worker
.on('message', this.opts
.messageHandler
?? EMPTY_FUNCTION
)
1268 worker
.on('error', this.opts
.errorHandler
?? EMPTY_FUNCTION
)
1269 worker
.on('error', error
=> {
1270 const workerNodeKey
= this.getWorkerNodeKeyByWorker(worker
)
1271 this.flagWorkerNodeAsNotReady(workerNodeKey
)
1272 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1273 this.emitter
?.emit(PoolEvents
.error
, error
)
1274 this.workerNodes
[workerNodeKey
].closeChannel()
1279 this.opts
.restartWorkerOnError
=== true
1281 if (workerInfo
.dynamic
) {
1282 this.createAndSetupDynamicWorkerNode()
1284 this.createAndSetupWorkerNode()
1287 if (this.started
&& this.opts
.enableTasksQueue
=== true) {
1288 this.redistributeQueuedTasks(workerNodeKey
)
1291 worker
.on('exit', this.opts
.exitHandler
?? EMPTY_FUNCTION
)
1292 worker
.once('exit', () => {
1293 this.removeWorkerNode(worker
)
1296 const workerNodeKey
= this.addWorkerNode(worker
)
1298 this.afterWorkerNodeSetup(workerNodeKey
)
1300 return workerNodeKey
1304 * Creates a new, completely set up dynamic worker node.
1306 * @returns New, completely set up dynamic worker node key.
1308 protected createAndSetupDynamicWorkerNode (): number {
1309 const workerNodeKey
= this.createAndSetupWorkerNode()
1310 this.registerWorkerMessageListener(workerNodeKey
, message
=> {
1311 this.checkMessageWorkerId(message
)
1312 const localWorkerNodeKey
= this.getWorkerNodeKeyByWorkerId(
1315 const workerUsage
= this.workerNodes
[localWorkerNodeKey
].usage
1316 // Kill message received from worker
1318 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
1319 (isKillBehavior(KillBehaviors
.SOFT
, message
.kill
) &&
1320 ((this.opts
.enableTasksQueue
=== false &&
1321 workerUsage
.tasks
.executing
=== 0) ||
1322 (this.opts
.enableTasksQueue
=== true &&
1323 workerUsage
.tasks
.executing
=== 0 &&
1324 this.tasksQueueSize(localWorkerNodeKey
) === 0)))
1326 // Flag the worker node as not ready immediately
1327 this.flagWorkerNodeAsNotReady(localWorkerNodeKey
)
1328 this.destroyWorkerNode(localWorkerNodeKey
).catch(error
=> {
1329 this.emitter
?.emit(PoolEvents
.error
, error
)
1333 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1334 this.sendToWorker(workerNodeKey
, {
1337 if (this.taskFunctions
.size
> 0) {
1338 for (const [taskFunctionName
, taskFunction
] of this.taskFunctions
) {
1339 this.sendTaskFunctionOperationToWorker(workerNodeKey
, {
1340 taskFunctionOperation
: 'add',
1342 taskFunction
: taskFunction
.toString()
1344 this.emitter
?.emit(PoolEvents
.error
, error
)
1348 workerInfo
.dynamic
= true
1350 this.workerChoiceStrategyContext
.getStrategyPolicy().dynamicWorkerReady
||
1351 this.workerChoiceStrategyContext
.getStrategyPolicy().dynamicWorkerUsage
1353 workerInfo
.ready
= true
1355 this.checkAndEmitDynamicWorkerCreationEvents()
1356 return workerNodeKey
1360 * Registers a listener callback on the worker given its worker node key.
1362 * @param workerNodeKey - The worker node key.
1363 * @param listener - The message listener callback.
1365 protected abstract registerWorkerMessageListener
<
1366 Message
extends Data
| Response
1368 workerNodeKey
: number,
1369 listener
: (message
: MessageValue
<Message
>) => void
1373 * Registers once a listener callback on the worker given its worker node key.
1375 * @param workerNodeKey - The worker node key.
1376 * @param listener - The message listener callback.
1378 protected abstract registerOnceWorkerMessageListener
<
1379 Message
extends Data
| Response
1381 workerNodeKey
: number,
1382 listener
: (message
: MessageValue
<Message
>) => void
1386 * Deregisters a listener callback on the worker given its worker node key.
1388 * @param workerNodeKey - The worker node key.
1389 * @param listener - The message listener callback.
1391 protected abstract deregisterWorkerMessageListener
<
1392 Message
extends Data
| Response
1394 workerNodeKey
: number,
1395 listener
: (message
: MessageValue
<Message
>) => void
1399 * Method hooked up after a worker node has been newly created.
1400 * Can be overridden.
1402 * @param workerNodeKey - The newly created worker node key.
1404 protected afterWorkerNodeSetup (workerNodeKey
: number): void {
1405 // Listen to worker messages.
1406 this.registerWorkerMessageListener(
1408 this.workerMessageListener
1410 // Send the startup message to worker.
1411 this.sendStartupMessageToWorker(workerNodeKey
)
1412 // Send the statistics message to worker.
1413 this.sendStatisticsMessageToWorker(workerNodeKey
)
1414 if (this.opts
.enableTasksQueue
=== true) {
1415 if (this.opts
.tasksQueueOptions
?.taskStealing
=== true) {
1416 this.workerNodes
[workerNodeKey
].on(
1418 this.handleIdleWorkerNodeEvent
1421 if (this.opts
.tasksQueueOptions
?.tasksStealingOnBackPressure
=== true) {
1422 this.workerNodes
[workerNodeKey
].on(
1424 this.handleBackPressureEvent
1431 * Sends the startup message to worker given its worker node key.
1433 * @param workerNodeKey - The worker node key.
1435 protected abstract sendStartupMessageToWorker (workerNodeKey
: number): void
1438 * Sends the statistics message to worker given its worker node key.
1440 * @param workerNodeKey - The worker node key.
1442 private sendStatisticsMessageToWorker (workerNodeKey
: number): void {
1443 this.sendToWorker(workerNodeKey
, {
1446 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1448 elu
: this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1454 private redistributeQueuedTasks (workerNodeKey
: number): void {
1455 if (this.workerNodes
.length
<= 1) {
1458 while (this.tasksQueueSize(workerNodeKey
) > 0) {
1459 const destinationWorkerNodeKey
= this.workerNodes
.reduce(
1460 (minWorkerNodeKey
, workerNode
, workerNodeKey
, workerNodes
) => {
1461 return workerNode
.info
.ready
&&
1462 workerNode
.usage
.tasks
.queued
<
1463 workerNodes
[minWorkerNodeKey
].usage
.tasks
.queued
1469 const task
= this.dequeueTask(workerNodeKey
) as Task
<Data
>
1470 if (this.shallExecuteTask(destinationWorkerNodeKey
)) {
1471 this.executeTask(destinationWorkerNodeKey
, task
)
1473 this.enqueueTask(destinationWorkerNodeKey
, task
)
1478 private updateTaskStolenStatisticsWorkerUsage (
1479 workerNodeKey
: number,
1482 const workerNode
= this.workerNodes
[workerNodeKey
]
1483 if (workerNode
?.usage
!= null) {
1484 ++workerNode
.usage
.tasks
.stolen
1487 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1488 workerNode
.getTaskFunctionWorkerUsage(taskName
) != null
1490 const taskFunctionWorkerUsage
= workerNode
.getTaskFunctionWorkerUsage(
1493 ++taskFunctionWorkerUsage
.tasks
.stolen
1497 private updateTaskSequentiallyStolenStatisticsWorkerUsage (
1498 workerNodeKey
: number
1500 const workerNode
= this.workerNodes
[workerNodeKey
]
1501 if (workerNode
?.usage
!= null) {
1502 ++workerNode
.usage
.tasks
.sequentiallyStolen
1506 private updateTaskSequentiallyStolenStatisticsTaskFunctionWorkerUsage (
1507 workerNodeKey
: number,
1510 const workerNode
= this.workerNodes
[workerNodeKey
]
1512 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1513 workerNode
.getTaskFunctionWorkerUsage(taskName
) != null
1515 const taskFunctionWorkerUsage
= workerNode
.getTaskFunctionWorkerUsage(
1518 ++taskFunctionWorkerUsage
.tasks
.sequentiallyStolen
1522 private resetTaskSequentiallyStolenStatisticsWorkerUsage (
1523 workerNodeKey
: number
1525 const workerNode
= this.workerNodes
[workerNodeKey
]
1526 if (workerNode
?.usage
!= null) {
1527 workerNode
.usage
.tasks
.sequentiallyStolen
= 0
1531 private resetTaskSequentiallyStolenStatisticsTaskFunctionWorkerUsage (
1532 workerNodeKey
: number,
1535 const workerNode
= this.workerNodes
[workerNodeKey
]
1537 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1538 workerNode
.getTaskFunctionWorkerUsage(taskName
) != null
1540 const taskFunctionWorkerUsage
= workerNode
.getTaskFunctionWorkerUsage(
1543 taskFunctionWorkerUsage
.tasks
.sequentiallyStolen
= 0
1547 private readonly handleIdleWorkerNodeEvent
= (
1548 eventDetail
: WorkerNodeEventDetail
,
1549 previousStolenTask
?: Task
<Data
>
1551 if (this.workerNodes
.length
<= 1) {
1554 const { workerNodeKey
} = eventDetail
1555 if (workerNodeKey
== null) {
1557 'WorkerNode event detail workerNodeKey attribute must be defined'
1560 const workerNodeTasksUsage
= this.workerNodes
[workerNodeKey
].usage
.tasks
1562 previousStolenTask
!= null &&
1563 workerNodeTasksUsage
.sequentiallyStolen
> 0 &&
1564 (workerNodeTasksUsage
.executing
> 0 ||
1565 this.tasksQueueSize(workerNodeKey
) > 0)
1567 for (const taskName
of this.workerNodes
[workerNodeKey
].info
1568 .taskFunctionNames
as string[]) {
1569 this.resetTaskSequentiallyStolenStatisticsTaskFunctionWorkerUsage(
1574 this.resetTaskSequentiallyStolenStatisticsWorkerUsage(workerNodeKey
)
1577 const stolenTask
= this.workerNodeStealTask(workerNodeKey
)
1579 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1582 const taskFunctionTasksWorkerUsage
= this.workerNodes
[
1584 ].getTaskFunctionWorkerUsage(stolenTask
.name
as string)
1585 ?.tasks
as TaskStatistics
1587 taskFunctionTasksWorkerUsage
.sequentiallyStolen
=== 0 ||
1588 (previousStolenTask
!= null &&
1589 previousStolenTask
.name
=== stolenTask
.name
&&
1590 taskFunctionTasksWorkerUsage
.sequentiallyStolen
> 0)
1592 this.updateTaskSequentiallyStolenStatisticsTaskFunctionWorkerUsage(
1594 stolenTask
.name
as string
1597 this.resetTaskSequentiallyStolenStatisticsTaskFunctionWorkerUsage(
1599 stolenTask
.name
as string
1603 sleep(exponentialDelay(workerNodeTasksUsage
.sequentiallyStolen
))
1605 this.handleIdleWorkerNodeEvent(eventDetail
, stolenTask
)
1608 .catch(EMPTY_FUNCTION
)
1611 private readonly workerNodeStealTask
= (
1612 workerNodeKey
: number
1613 ): Task
<Data
> | undefined => {
1614 const workerNodes
= this.workerNodes
1617 (workerNodeA
, workerNodeB
) =>
1618 workerNodeB
.usage
.tasks
.queued
- workerNodeA
.usage
.tasks
.queued
1620 const sourceWorkerNode
= workerNodes
.find(
1621 (sourceWorkerNode
, sourceWorkerNodeKey
) =>
1622 sourceWorkerNode
.info
.ready
&&
1623 sourceWorkerNodeKey
!== workerNodeKey
&&
1624 sourceWorkerNode
.usage
.tasks
.queued
> 0
1626 if (sourceWorkerNode
!= null) {
1627 const task
= sourceWorkerNode
.popTask() as Task
<Data
>
1628 if (this.shallExecuteTask(workerNodeKey
)) {
1629 this.executeTask(workerNodeKey
, task
)
1631 this.enqueueTask(workerNodeKey
, task
)
1633 this.updateTaskSequentiallyStolenStatisticsWorkerUsage(workerNodeKey
)
1634 this.updateTaskStolenStatisticsWorkerUsage(
1642 private readonly handleBackPressureEvent
= (
1643 eventDetail
: WorkerNodeEventDetail
1645 if (this.workerNodes
.length
<= 1) {
1648 const { workerId
} = eventDetail
1649 const sizeOffset
= 1
1650 if ((this.opts
.tasksQueueOptions
?.size
as number) <= sizeOffset
) {
1653 const sourceWorkerNode
=
1654 this.workerNodes
[this.getWorkerNodeKeyByWorkerId(workerId
)]
1655 const workerNodes
= this.workerNodes
1658 (workerNodeA
, workerNodeB
) =>
1659 workerNodeA
.usage
.tasks
.queued
- workerNodeB
.usage
.tasks
.queued
1661 for (const [workerNodeKey
, workerNode
] of workerNodes
.entries()) {
1663 sourceWorkerNode
.usage
.tasks
.queued
> 0 &&
1664 workerNode
.info
.ready
&&
1665 workerNode
.info
.id
!== workerId
&&
1666 workerNode
.usage
.tasks
.queued
<
1667 (this.opts
.tasksQueueOptions
?.size
as number) - sizeOffset
1669 const task
= sourceWorkerNode
.popTask() as Task
<Data
>
1670 if (this.shallExecuteTask(workerNodeKey
)) {
1671 this.executeTask(workerNodeKey
, task
)
1673 this.enqueueTask(workerNodeKey
, task
)
1675 this.updateTaskStolenStatisticsWorkerUsage(
1684 * This method is the message listener registered on each worker.
1686 protected readonly workerMessageListener
= (
1687 message
: MessageValue
<Response
>
1689 this.checkMessageWorkerId(message
)
1690 const { workerId
, ready
, taskId
, taskFunctionNames
} = message
1691 if (ready
!= null && taskFunctionNames
!= null) {
1692 // Worker ready response received from worker
1693 this.handleWorkerReadyResponse(message
)
1694 } else if (taskId
!= null) {
1695 // Task execution response received from worker
1696 this.handleTaskExecutionResponse(message
)
1697 } else if (taskFunctionNames
!= null) {
1698 // Task function names message received from worker
1700 this.getWorkerNodeKeyByWorkerId(workerId
)
1701 ).taskFunctionNames
= taskFunctionNames
1705 private handleWorkerReadyResponse (message
: MessageValue
<Response
>): void {
1706 const { workerId
, ready
, taskFunctionNames
} = message
1707 if (ready
=== false) {
1708 throw new Error(`Worker ${workerId as number} failed to initialize`)
1710 const workerInfo
= this.getWorkerInfo(
1711 this.getWorkerNodeKeyByWorkerId(workerId
)
1713 workerInfo
.ready
= ready
as boolean
1714 workerInfo
.taskFunctionNames
= taskFunctionNames
1715 if (!this.readyEventEmitted
&& this.ready
) {
1716 this.readyEventEmitted
= true
1717 this.emitter
?.emit(PoolEvents
.ready
, this.info
)
1721 private handleTaskExecutionResponse (message
: MessageValue
<Response
>): void {
1722 const { workerId
, taskId
, workerError
, data
} = message
1723 const promiseResponse
= this.promiseResponseMap
.get(taskId
as string)
1724 if (promiseResponse
!= null) {
1725 const { resolve
, reject
, workerNodeKey
, asyncResource
} = promiseResponse
1726 if (workerError
!= null) {
1727 this.emitter
?.emit(PoolEvents
.taskError
, workerError
)
1728 asyncResource
!= null
1729 ? asyncResource
.runInAsyncScope(
1734 : reject(workerError
.message
)
1736 asyncResource
!= null
1737 ? asyncResource
.runInAsyncScope(resolve
, this.emitter
, data
)
1738 : resolve(data
as Response
)
1740 asyncResource
?.emitDestroy()
1741 this.afterTaskExecutionHook(workerNodeKey
, message
)
1742 this.workerChoiceStrategyContext
.update(workerNodeKey
)
1743 this.promiseResponseMap
.delete(taskId
as string)
1744 if (this.opts
.enableTasksQueue
=== true) {
1745 const workerNodeTasksUsage
= this.workerNodes
[workerNodeKey
].usage
.tasks
1747 this.tasksQueueSize(workerNodeKey
) > 0 &&
1748 workerNodeTasksUsage
.executing
<
1749 (this.opts
.tasksQueueOptions
?.concurrency
as number)
1753 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1757 workerNodeTasksUsage
.executing
=== 0 &&
1758 this.tasksQueueSize(workerNodeKey
) === 0 &&
1759 workerNodeTasksUsage
.sequentiallyStolen
=== 0
1761 this.workerNodes
[workerNodeKey
].emit('idleWorkerNode', {
1762 workerId
: workerId
as number,
1770 private checkAndEmitTaskExecutionEvents (): void {
1772 this.emitter
?.emit(PoolEvents
.busy
, this.info
)
1776 private checkAndEmitTaskQueuingEvents (): void {
1777 if (this.hasBackPressure()) {
1778 this.emitter
?.emit(PoolEvents
.backPressure
, this.info
)
1782 private checkAndEmitDynamicWorkerCreationEvents (): void {
1783 if (this.type === PoolTypes
.dynamic
) {
1785 this.emitter
?.emit(PoolEvents
.full
, this.info
)
1791 * Gets the worker information given its worker node key.
1793 * @param workerNodeKey - The worker node key.
1794 * @returns The worker information.
1796 protected getWorkerInfo (workerNodeKey
: number): WorkerInfo
{
1797 return this.workerNodes
[workerNodeKey
]?.info
1801 * Adds the given worker in the pool worker nodes.
1803 * @param worker - The worker.
1804 * @returns The added worker node key.
1805 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the added worker node is not found.
1807 private addWorkerNode (worker
: Worker
): number {
1808 const workerNode
= new WorkerNode
<Worker
, Data
>(
1810 this.opts
.tasksQueueOptions
?.size
?? Math.pow(this.maxSize
, 2)
1812 // Flag the worker node as ready at pool startup.
1813 if (this.starting
) {
1814 workerNode
.info
.ready
= true
1816 this.workerNodes
.push(workerNode
)
1817 const workerNodeKey
= this.getWorkerNodeKeyByWorker(worker
)
1818 if (workerNodeKey
=== -1) {
1819 throw new Error('Worker added not found in worker nodes')
1821 return workerNodeKey
1825 * Removes the given worker from the pool worker nodes.
1827 * @param worker - The worker.
1829 private removeWorkerNode (worker
: Worker
): void {
1830 const workerNodeKey
= this.getWorkerNodeKeyByWorker(worker
)
1831 if (workerNodeKey
!== -1) {
1832 this.workerNodes
.splice(workerNodeKey
, 1)
1833 this.workerChoiceStrategyContext
.remove(workerNodeKey
)
1837 protected flagWorkerNodeAsNotReady (workerNodeKey
: number): void {
1838 this.getWorkerInfo(workerNodeKey
).ready
= false
1842 public hasWorkerNodeBackPressure (workerNodeKey
: number): boolean {
1844 this.opts
.enableTasksQueue
=== true &&
1845 this.workerNodes
[workerNodeKey
].hasBackPressure()
1849 private hasBackPressure (): boolean {
1851 this.opts
.enableTasksQueue
=== true &&
1852 this.workerNodes
.findIndex(
1853 workerNode
=> !workerNode
.hasBackPressure()
1859 * Executes the given task on the worker given its worker node key.
1861 * @param workerNodeKey - The worker node key.
1862 * @param task - The task to execute.
1864 private executeTask (workerNodeKey
: number, task
: Task
<Data
>): void {
1865 this.beforeTaskExecutionHook(workerNodeKey
, task
)
1866 this.sendToWorker(workerNodeKey
, task
, task
.transferList
)
1867 this.checkAndEmitTaskExecutionEvents()
1870 private enqueueTask (workerNodeKey
: number, task
: Task
<Data
>): number {
1871 const tasksQueueSize
= this.workerNodes
[workerNodeKey
].enqueueTask(task
)
1872 this.checkAndEmitTaskQueuingEvents()
1873 return tasksQueueSize
1876 private dequeueTask (workerNodeKey
: number): Task
<Data
> | undefined {
1877 return this.workerNodes
[workerNodeKey
].dequeueTask()
1880 private tasksQueueSize (workerNodeKey
: number): number {
1881 return this.workerNodes
[workerNodeKey
].tasksQueueSize()
1884 protected flushTasksQueue (workerNodeKey
: number): void {
1885 while (this.tasksQueueSize(workerNodeKey
) > 0) {
1888 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1891 this.workerNodes
[workerNodeKey
].clearTasksQueue()
1894 private flushTasksQueues (): void {
1895 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
1896 this.flushTasksQueue(workerNodeKey
)