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'
24 import { KillBehaviors
} from
'../worker/worker-options'
25 import type { TaskFunction
} from
'../worker/task-functions'
33 type TasksQueueOptions
40 WorkerNodeEventDetail
,
46 WorkerChoiceStrategies
,
47 type WorkerChoiceStrategy
,
48 type WorkerChoiceStrategyOptions
49 } from
'./selection-strategies/selection-strategies-types'
50 import { WorkerChoiceStrategyContext
} from
'./selection-strategies/worker-choice-strategy-context'
51 import { version
} from
'./version'
52 import { WorkerNode
} from
'./worker-node'
55 checkValidTasksQueueOptions
,
56 checkValidWorkerChoiceStrategy
,
57 getDefaultTasksQueueOptions
,
59 updateRunTimeWorkerUsage
,
60 updateTaskStatisticsWorkerUsage
,
61 updateWaitTimeWorkerUsage
,
66 * Base class that implements some shared logic for all poolifier pools.
68 * @typeParam Worker - Type of worker which manages this pool.
69 * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
70 * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
72 export abstract class AbstractPool
<
73 Worker
extends IWorker
,
76 > implements IPool
<Worker
, Data
, Response
> {
78 public readonly workerNodes
: Array<IWorkerNode
<Worker
, Data
>> = []
81 public emitter
?: EventEmitterAsyncResource
84 * The task execution response promise map:
85 * - `key`: The message id of each submitted task.
86 * - `value`: An object that contains the worker, the execution response promise resolve and reject callbacks.
88 * When we receive a message from the worker, we get a map entry with the promise resolve/reject bound to the message id.
90 protected promiseResponseMap
: Map
<string, PromiseResponseWrapper
<Response
>> =
91 new Map
<string, PromiseResponseWrapper
<Response
>>()
94 * Worker choice strategy context referencing a worker choice algorithm implementation.
96 protected workerChoiceStrategyContext
: WorkerChoiceStrategyContext
<
103 * The task functions added at runtime map:
104 * - `key`: The task function name.
105 * - `value`: The task function itself.
107 private readonly taskFunctions
: Map
<string, TaskFunction
<Data
, Response
>>
110 * Whether the pool is started or not.
112 private started
: boolean
114 * Whether the pool is starting or not.
116 private starting
: boolean
118 * Whether the pool is destroying or not.
120 private destroying
: boolean
122 * Whether the pool ready event has been emitted or not.
124 private readyEventEmitted
: boolean
126 * The start timestamp of the pool.
128 private readonly startTimestamp
131 * Constructs a new poolifier pool.
133 * @param minimumNumberOfWorkers - Minimum number of workers that this pool manages.
134 * @param filePath - Path to the worker file.
135 * @param opts - Options for the pool.
136 * @param maximumNumberOfWorkers - Maximum number of workers that this pool manages.
139 protected readonly minimumNumberOfWorkers
: number,
140 protected readonly filePath
: string,
141 protected readonly opts
: PoolOptions
<Worker
>,
142 protected readonly maximumNumberOfWorkers
?: number
144 if (!this.isMain()) {
146 'Cannot start a pool from a worker with the same type as the pool'
150 checkFilePath(this.filePath
)
151 this.checkMinimumNumberOfWorkers(this.minimumNumberOfWorkers
)
152 this.checkPoolOptions(this.opts
)
154 this.chooseWorkerNode
= this.chooseWorkerNode
.bind(this)
155 this.executeTask
= this.executeTask
.bind(this)
156 this.enqueueTask
= this.enqueueTask
.bind(this)
158 if (this.opts
.enableEvents
=== true) {
159 this.initializeEventEmitter()
161 this.workerChoiceStrategyContext
= new WorkerChoiceStrategyContext
<
167 this.opts
.workerChoiceStrategy
,
168 this.opts
.workerChoiceStrategyOptions
173 this.taskFunctions
= new Map
<string, TaskFunction
<Data
, Response
>>()
176 this.starting
= false
177 this.destroying
= false
178 this.readyEventEmitted
= false
179 if (this.opts
.startWorkers
=== true) {
183 this.startTimestamp
= performance
.now()
186 private checkPoolType (): void {
187 if (this.type === PoolTypes
.fixed
&& this.maximumNumberOfWorkers
!= null) {
189 'Cannot instantiate a fixed pool with a maximum number of workers specified at initialization'
194 private checkMinimumNumberOfWorkers (minimumNumberOfWorkers
: number): void {
195 if (minimumNumberOfWorkers
== null) {
197 'Cannot instantiate a pool without specifying the number of workers'
199 } else if (!Number.isSafeInteger(minimumNumberOfWorkers
)) {
201 'Cannot instantiate a pool with a non safe integer number of workers'
203 } else if (minimumNumberOfWorkers
< 0) {
204 throw new RangeError(
205 'Cannot instantiate a pool with a negative number of workers'
207 } else if (this.type === PoolTypes
.fixed
&& minimumNumberOfWorkers
=== 0) {
208 throw new RangeError('Cannot instantiate a fixed pool with zero worker')
212 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
213 if (isPlainObject(opts
)) {
214 this.opts
.startWorkers
= opts
.startWorkers
?? true
215 checkValidWorkerChoiceStrategy(
216 opts
.workerChoiceStrategy
as WorkerChoiceStrategy
218 this.opts
.workerChoiceStrategy
=
219 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
220 this.checkValidWorkerChoiceStrategyOptions(
221 opts
.workerChoiceStrategyOptions
as WorkerChoiceStrategyOptions
223 if (opts
.workerChoiceStrategyOptions
!= null) {
224 this.opts
.workerChoiceStrategyOptions
= opts
.workerChoiceStrategyOptions
226 this.opts
.restartWorkerOnError
= opts
.restartWorkerOnError
?? true
227 this.opts
.enableEvents
= opts
.enableEvents
?? true
228 this.opts
.enableTasksQueue
= opts
.enableTasksQueue
?? false
229 if (this.opts
.enableTasksQueue
) {
230 checkValidTasksQueueOptions(opts
.tasksQueueOptions
as TasksQueueOptions
)
231 this.opts
.tasksQueueOptions
= this.buildTasksQueueOptions(
232 opts
.tasksQueueOptions
as TasksQueueOptions
236 throw new TypeError('Invalid pool options: must be a plain object')
240 private checkValidWorkerChoiceStrategyOptions (
241 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
244 workerChoiceStrategyOptions
!= null &&
245 !isPlainObject(workerChoiceStrategyOptions
)
248 'Invalid worker choice strategy options: must be a plain object'
252 workerChoiceStrategyOptions
?.weights
!= null &&
253 Object.keys(workerChoiceStrategyOptions
.weights
).length
!==
254 (this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
)
257 'Invalid worker choice strategy options: must have a weight for each worker node'
261 workerChoiceStrategyOptions
?.measurement
!= null &&
262 !Object.values(Measurements
).includes(
263 workerChoiceStrategyOptions
.measurement
267 `Invalid worker choice strategy options: invalid measurement '${workerChoiceStrategyOptions.measurement}'`
272 private initializeEventEmitter (): void {
273 this.emitter
= new EventEmitterAsyncResource({
274 name
: `poolifier:${this.type}-${this.worker}-pool`
279 public get
info (): PoolInfo
{
284 started
: this.started
,
286 strategy
: this.opts
.workerChoiceStrategy
as WorkerChoiceStrategy
,
287 minSize
: this.minimumNumberOfWorkers
,
288 maxSize
: this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
,
289 ...(this.workerChoiceStrategyContext
?.getTaskStatisticsRequirements()
290 .runTime
.aggregate
&&
291 this.workerChoiceStrategyContext
?.getTaskStatisticsRequirements()
292 .waitTime
.aggregate
&& { utilization
: round(this.utilization
) }),
293 workerNodes
: this.workerNodes
.length
,
294 idleWorkerNodes
: this.workerNodes
.reduce(
295 (accumulator
, workerNode
) =>
296 workerNode
.usage
.tasks
.executing
=== 0
301 busyWorkerNodes
: this.workerNodes
.reduce(
302 (accumulator
, _workerNode
, workerNodeKey
) =>
303 this.isWorkerNodeBusy(workerNodeKey
) ? accumulator
+ 1 : accumulator
,
306 executedTasks
: this.workerNodes
.reduce(
307 (accumulator
, workerNode
) =>
308 accumulator
+ workerNode
.usage
.tasks
.executed
,
311 executingTasks
: this.workerNodes
.reduce(
312 (accumulator
, workerNode
) =>
313 accumulator
+ workerNode
.usage
.tasks
.executing
,
316 ...(this.opts
.enableTasksQueue
=== true && {
317 queuedTasks
: this.workerNodes
.reduce(
318 (accumulator
, workerNode
) =>
319 accumulator
+ workerNode
.usage
.tasks
.queued
,
323 ...(this.opts
.enableTasksQueue
=== true && {
324 maxQueuedTasks
: this.workerNodes
.reduce(
325 (accumulator
, workerNode
) =>
326 accumulator
+ (workerNode
.usage
.tasks
?.maxQueued
?? 0),
330 ...(this.opts
.enableTasksQueue
=== true && {
331 backPressure
: this.hasBackPressure()
333 ...(this.opts
.enableTasksQueue
=== true && {
334 stolenTasks
: this.workerNodes
.reduce(
335 (accumulator
, workerNode
) =>
336 accumulator
+ workerNode
.usage
.tasks
.stolen
,
340 failedTasks
: this.workerNodes
.reduce(
341 (accumulator
, workerNode
) =>
342 accumulator
+ workerNode
.usage
.tasks
.failed
,
345 ...(this.workerChoiceStrategyContext
?.getTaskStatisticsRequirements()
346 .runTime
.aggregate
&& {
350 ...this.workerNodes
.map(
351 workerNode
=> workerNode
.usage
.runTime
?.minimum
?? Infinity
357 ...this.workerNodes
.map(
358 workerNode
=> workerNode
.usage
.runTime
?.maximum
?? -Infinity
362 ...(this.workerChoiceStrategyContext
?.getTaskStatisticsRequirements()
363 .runTime
.average
&& {
366 this.workerNodes
.reduce
<number[]>(
367 (accumulator
, workerNode
) =>
368 accumulator
.concat(workerNode
.usage
.runTime
.history
),
374 ...(this.workerChoiceStrategyContext
?.getTaskStatisticsRequirements()
378 this.workerNodes
.reduce
<number[]>(
379 (accumulator
, workerNode
) =>
380 accumulator
.concat(workerNode
.usage
.runTime
.history
),
388 ...(this.workerChoiceStrategyContext
?.getTaskStatisticsRequirements()
389 .waitTime
.aggregate
&& {
393 ...this.workerNodes
.map(
394 workerNode
=> workerNode
.usage
.waitTime
?.minimum
?? Infinity
400 ...this.workerNodes
.map(
401 workerNode
=> workerNode
.usage
.waitTime
?.maximum
?? -Infinity
405 ...(this.workerChoiceStrategyContext
?.getTaskStatisticsRequirements()
406 .waitTime
.average
&& {
409 this.workerNodes
.reduce
<number[]>(
410 (accumulator
, workerNode
) =>
411 accumulator
.concat(workerNode
.usage
.waitTime
.history
),
417 ...(this.workerChoiceStrategyContext
?.getTaskStatisticsRequirements()
418 .waitTime
.median
&& {
421 this.workerNodes
.reduce
<number[]>(
422 (accumulator
, workerNode
) =>
423 accumulator
.concat(workerNode
.usage
.waitTime
.history
),
435 * The pool readiness boolean status.
437 private get
ready (): boolean {
439 this.workerNodes
.reduce(
440 (accumulator
, workerNode
) =>
441 !workerNode
.info
.dynamic
&& workerNode
.info
.ready
445 ) >= this.minimumNumberOfWorkers
450 * The approximate pool utilization.
452 * @returns The pool utilization.
454 private get
utilization (): number {
455 const poolTimeCapacity
=
456 (performance
.now() - this.startTimestamp
) *
457 (this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
)
458 const totalTasksRunTime
= this.workerNodes
.reduce(
459 (accumulator
, workerNode
) =>
460 accumulator
+ (workerNode
.usage
.runTime
?.aggregate
?? 0),
463 const totalTasksWaitTime
= this.workerNodes
.reduce(
464 (accumulator
, workerNode
) =>
465 accumulator
+ (workerNode
.usage
.waitTime
?.aggregate
?? 0),
468 return (totalTasksRunTime
+ totalTasksWaitTime
) / poolTimeCapacity
474 * If it is `'dynamic'`, it provides the `max` property.
476 protected abstract get
type (): PoolType
481 protected abstract get
worker (): WorkerType
484 * Checks if the worker id sent in the received message from a worker is valid.
486 * @param message - The received message.
487 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the worker id is invalid.
489 private checkMessageWorkerId (message
: MessageValue
<Data
| Response
>): void {
490 if (message
.workerId
== null) {
491 throw new Error('Worker message received without worker id')
492 } else if (this.getWorkerNodeKeyByWorkerId(message
.workerId
) === -1) {
494 `Worker message received from unknown worker '${message.workerId}'`
500 * Gets the worker node key given its worker id.
502 * @param workerId - The worker id.
503 * @returns The worker node key if the worker id is found in the pool worker nodes, `-1` otherwise.
505 private getWorkerNodeKeyByWorkerId (workerId
: number | undefined): number {
506 return this.workerNodes
.findIndex(
507 workerNode
=> workerNode
.info
.id
=== workerId
512 public setWorkerChoiceStrategy (
513 workerChoiceStrategy
: WorkerChoiceStrategy
,
514 workerChoiceStrategyOptions
?: WorkerChoiceStrategyOptions
516 checkValidWorkerChoiceStrategy(workerChoiceStrategy
)
517 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
518 this.workerChoiceStrategyContext
.setWorkerChoiceStrategy(
519 this.opts
.workerChoiceStrategy
521 if (workerChoiceStrategyOptions
!= null) {
522 this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
524 for (const [workerNodeKey
, workerNode
] of this.workerNodes
.entries()) {
525 workerNode
.resetUsage()
526 this.sendStatisticsMessageToWorker(workerNodeKey
)
531 public setWorkerChoiceStrategyOptions (
532 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
534 this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
535 if (workerChoiceStrategyOptions
!= null) {
536 this.opts
.workerChoiceStrategyOptions
= workerChoiceStrategyOptions
538 this.workerChoiceStrategyContext
.setOptions(
540 this.opts
.workerChoiceStrategyOptions
545 public enableTasksQueue (
547 tasksQueueOptions
?: TasksQueueOptions
549 if (this.opts
.enableTasksQueue
=== true && !enable
) {
550 this.unsetTaskStealing()
551 this.unsetTasksStealingOnBackPressure()
552 this.flushTasksQueues()
554 this.opts
.enableTasksQueue
= enable
555 this.setTasksQueueOptions(tasksQueueOptions
as TasksQueueOptions
)
559 public setTasksQueueOptions (tasksQueueOptions
: TasksQueueOptions
): void {
560 if (this.opts
.enableTasksQueue
=== true) {
561 checkValidTasksQueueOptions(tasksQueueOptions
)
562 this.opts
.tasksQueueOptions
=
563 this.buildTasksQueueOptions(tasksQueueOptions
)
564 this.setTasksQueueSize(this.opts
.tasksQueueOptions
.size
as number)
565 if (this.opts
.tasksQueueOptions
.taskStealing
=== true) {
566 this.unsetTaskStealing()
567 this.setTaskStealing()
569 this.unsetTaskStealing()
571 if (this.opts
.tasksQueueOptions
.tasksStealingOnBackPressure
=== true) {
572 this.unsetTasksStealingOnBackPressure()
573 this.setTasksStealingOnBackPressure()
575 this.unsetTasksStealingOnBackPressure()
577 } else if (this.opts
.tasksQueueOptions
!= null) {
578 delete this.opts
.tasksQueueOptions
582 private buildTasksQueueOptions (
583 tasksQueueOptions
: TasksQueueOptions
584 ): TasksQueueOptions
{
586 ...getDefaultTasksQueueOptions(
587 this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
593 private setTasksQueueSize (size
: number): void {
594 for (const workerNode
of this.workerNodes
) {
595 workerNode
.tasksQueueBackPressureSize
= size
599 private setTaskStealing (): void {
600 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
601 this.workerNodes
[workerNodeKey
].on(
603 this.handleIdleWorkerNodeEvent
608 private unsetTaskStealing (): void {
609 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
610 this.workerNodes
[workerNodeKey
].off(
612 this.handleIdleWorkerNodeEvent
617 private setTasksStealingOnBackPressure (): void {
618 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
619 this.workerNodes
[workerNodeKey
].on(
621 this.handleBackPressureEvent
626 private unsetTasksStealingOnBackPressure (): void {
627 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
628 this.workerNodes
[workerNodeKey
].off(
630 this.handleBackPressureEvent
636 * Whether the pool is full or not.
638 * The pool filling boolean status.
640 protected get
full (): boolean {
642 this.workerNodes
.length
>=
643 (this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
)
648 * Whether the pool is busy or not.
650 * The pool busyness boolean status.
652 protected abstract get
busy (): boolean
655 * Whether worker nodes are executing concurrently their tasks quota or not.
657 * @returns Worker nodes busyness boolean status.
659 protected internalBusy (): boolean {
660 if (this.opts
.enableTasksQueue
=== true) {
662 this.workerNodes
.findIndex(
664 workerNode
.info
.ready
&&
665 workerNode
.usage
.tasks
.executing
<
666 (this.opts
.tasksQueueOptions
?.concurrency
as number)
671 this.workerNodes
.findIndex(
673 workerNode
.info
.ready
&& workerNode
.usage
.tasks
.executing
=== 0
678 private isWorkerNodeBusy (workerNodeKey
: number): boolean {
679 if (this.opts
.enableTasksQueue
=== true) {
681 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
>=
682 (this.opts
.tasksQueueOptions
?.concurrency
as number)
685 return this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
> 0
688 private async sendTaskFunctionOperationToWorker (
689 workerNodeKey
: number,
690 message
: MessageValue
<Data
>
691 ): Promise
<boolean> {
692 return await new Promise
<boolean>((resolve
, reject
) => {
693 const taskFunctionOperationListener
= (
694 message
: MessageValue
<Response
>
696 this.checkMessageWorkerId(message
)
697 const workerId
= this.getWorkerInfo(workerNodeKey
).id
as number
699 message
.taskFunctionOperationStatus
!= null &&
700 message
.workerId
=== workerId
702 if (message
.taskFunctionOperationStatus
) {
704 } else if (!message
.taskFunctionOperationStatus
) {
707 `Task function operation '${
708 message.taskFunctionOperation as string
709 }' failed on worker ${message.workerId} with error: '${
710 message.workerError?.message as string
715 this.deregisterWorkerMessageListener(
716 this.getWorkerNodeKeyByWorkerId(message
.workerId
),
717 taskFunctionOperationListener
721 this.registerWorkerMessageListener(
723 taskFunctionOperationListener
725 this.sendToWorker(workerNodeKey
, message
)
729 private async sendTaskFunctionOperationToWorkers (
730 message
: MessageValue
<Data
>
731 ): Promise
<boolean> {
732 return await new Promise
<boolean>((resolve
, reject
) => {
733 const responsesReceived
= new Array<MessageValue
<Response
>>()
734 const taskFunctionOperationsListener
= (
735 message
: MessageValue
<Response
>
737 this.checkMessageWorkerId(message
)
738 if (message
.taskFunctionOperationStatus
!= null) {
739 responsesReceived
.push(message
)
740 if (responsesReceived
.length
=== this.workerNodes
.length
) {
742 responsesReceived
.every(
743 message
=> message
.taskFunctionOperationStatus
=== true
748 responsesReceived
.some(
749 message
=> message
.taskFunctionOperationStatus
=== false
752 const errorResponse
= responsesReceived
.find(
753 response
=> response
.taskFunctionOperationStatus
=== false
757 `Task function operation '${
758 message.taskFunctionOperation as string
759 }' failed on worker ${
760 errorResponse?.workerId as number
762 errorResponse?.workerError?.message as string
767 this.deregisterWorkerMessageListener(
768 this.getWorkerNodeKeyByWorkerId(message
.workerId
),
769 taskFunctionOperationsListener
774 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
775 this.registerWorkerMessageListener(
777 taskFunctionOperationsListener
779 this.sendToWorker(workerNodeKey
, message
)
785 public hasTaskFunction (name
: string): boolean {
786 for (const workerNode
of this.workerNodes
) {
788 Array.isArray(workerNode
.info
.taskFunctionNames
) &&
789 workerNode
.info
.taskFunctionNames
.includes(name
)
798 public async addTaskFunction (
800 fn
: TaskFunction
<Data
, Response
>
801 ): Promise
<boolean> {
802 if (typeof name
!== 'string') {
803 throw new TypeError('name argument must be a string')
805 if (typeof name
=== 'string' && name
.trim().length
=== 0) {
806 throw new TypeError('name argument must not be an empty string')
808 if (typeof fn
!== 'function') {
809 throw new TypeError('fn argument must be a function')
811 const opResult
= await this.sendTaskFunctionOperationToWorkers({
812 taskFunctionOperation
: 'add',
813 taskFunctionName
: name
,
814 taskFunction
: fn
.toString()
816 this.taskFunctions
.set(name
, fn
)
821 public async removeTaskFunction (name
: string): Promise
<boolean> {
822 if (!this.taskFunctions
.has(name
)) {
824 'Cannot remove a task function not handled on the pool side'
827 const opResult
= await this.sendTaskFunctionOperationToWorkers({
828 taskFunctionOperation
: 'remove',
829 taskFunctionName
: name
831 this.deleteTaskFunctionWorkerUsages(name
)
832 this.taskFunctions
.delete(name
)
837 public listTaskFunctionNames (): string[] {
838 for (const workerNode
of this.workerNodes
) {
840 Array.isArray(workerNode
.info
.taskFunctionNames
) &&
841 workerNode
.info
.taskFunctionNames
.length
> 0
843 return workerNode
.info
.taskFunctionNames
850 public async setDefaultTaskFunction (name
: string): Promise
<boolean> {
851 return await this.sendTaskFunctionOperationToWorkers({
852 taskFunctionOperation
: 'default',
853 taskFunctionName
: name
857 private deleteTaskFunctionWorkerUsages (name
: string): void {
858 for (const workerNode
of this.workerNodes
) {
859 workerNode
.deleteTaskFunctionWorkerUsage(name
)
863 private shallExecuteTask (workerNodeKey
: number): boolean {
865 this.tasksQueueSize(workerNodeKey
) === 0 &&
866 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
<
867 (this.opts
.tasksQueueOptions
?.concurrency
as number)
872 public async execute (
875 transferList
?: TransferListItem
[]
876 ): Promise
<Response
> {
877 return await new Promise
<Response
>((resolve
, reject
) => {
879 reject(new Error('Cannot execute a task on not started pool'))
882 if (this.destroying
) {
883 reject(new Error('Cannot execute a task on destroying pool'))
886 if (name
!= null && typeof name
!== 'string') {
887 reject(new TypeError('name argument must be a string'))
892 typeof name
=== 'string' &&
893 name
.trim().length
=== 0
895 reject(new TypeError('name argument must not be an empty string'))
898 if (transferList
!= null && !Array.isArray(transferList
)) {
899 reject(new TypeError('transferList argument must be an array'))
902 const timestamp
= performance
.now()
903 const workerNodeKey
= this.chooseWorkerNode()
904 const task
: Task
<Data
> = {
905 name
: name
?? DEFAULT_TASK_NAME
,
906 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
907 data
: data
?? ({} as Data
),
912 this.promiseResponseMap
.set(task
.taskId
as string, {
916 ...(this.emitter
!= null && {
917 asyncResource
: new AsyncResource('poolifier:task', {
918 triggerAsyncId
: this.emitter
.asyncId
,
919 requireManualDestroy
: true
924 this.opts
.enableTasksQueue
=== false ||
925 (this.opts
.enableTasksQueue
=== true &&
926 this.shallExecuteTask(workerNodeKey
))
928 this.executeTask(workerNodeKey
, task
)
930 this.enqueueTask(workerNodeKey
, task
)
936 public start (): void {
938 throw new Error('Cannot start an already started pool')
941 throw new Error('Cannot start an already starting pool')
943 if (this.destroying
) {
944 throw new Error('Cannot start a destroying pool')
948 this.workerNodes
.reduce(
949 (accumulator
, workerNode
) =>
950 !workerNode
.info
.dynamic
? accumulator
+ 1 : accumulator
,
952 ) < this.minimumNumberOfWorkers
954 this.createAndSetupWorkerNode()
956 this.starting
= false
961 public async destroy (): Promise
<void> {
963 throw new Error('Cannot destroy an already destroyed pool')
966 throw new Error('Cannot destroy an starting pool')
968 if (this.destroying
) {
969 throw new Error('Cannot destroy an already destroying pool')
971 this.destroying
= true
973 this.workerNodes
.map(async (_workerNode
, workerNodeKey
) => {
974 await this.destroyWorkerNode(workerNodeKey
)
977 this.emitter
?.emit(PoolEvents
.destroy
, this.info
)
978 this.emitter
?.emitDestroy()
979 this.emitter
?.removeAllListeners()
980 this.readyEventEmitted
= false
981 this.destroying
= false
985 private async sendKillMessageToWorker (workerNodeKey
: number): Promise
<void> {
986 await new Promise
<void>((resolve
, reject
) => {
987 if (this.workerNodes
?.[workerNodeKey
] == null) {
991 const killMessageListener
= (message
: MessageValue
<Response
>): void => {
992 this.checkMessageWorkerId(message
)
993 if (message
.kill
=== 'success') {
995 } else if (message
.kill
=== 'failure') {
998 `Kill message handling failed on worker ${
999 message.workerId as number
1005 // FIXME: should be registered only once
1006 this.registerWorkerMessageListener(workerNodeKey
, killMessageListener
)
1007 this.sendToWorker(workerNodeKey
, { kill
: true })
1012 * Terminates the worker node given its worker node key.
1014 * @param workerNodeKey - The worker node key.
1016 protected async destroyWorkerNode (workerNodeKey
: number): Promise
<void> {
1017 this.flagWorkerNodeAsNotReady(workerNodeKey
)
1018 const flushedTasks
= this.flushTasksQueue(workerNodeKey
)
1019 const workerNode
= this.workerNodes
[workerNodeKey
]
1020 await waitWorkerNodeEvents(
1024 this.opts
.tasksQueueOptions
?.tasksFinishedTimeout
??
1025 getDefaultTasksQueueOptions(
1026 this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
1027 ).tasksFinishedTimeout
1029 await this.sendKillMessageToWorker(workerNodeKey
)
1030 await workerNode
.terminate()
1034 * Setup hook to execute code before worker nodes are created in the abstract constructor.
1035 * Can be overridden.
1039 protected setupHook (): void {
1040 /* Intentionally empty */
1044 * Should return whether the worker is the main worker or not.
1046 protected abstract isMain (): boolean
1049 * Hook executed before the worker task execution.
1050 * Can be overridden.
1052 * @param workerNodeKey - The worker node key.
1053 * @param task - The task to execute.
1055 protected beforeTaskExecutionHook (
1056 workerNodeKey
: number,
1059 if (this.workerNodes
[workerNodeKey
]?.usage
!= null) {
1060 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
1061 ++workerUsage
.tasks
.executing
1062 updateWaitTimeWorkerUsage(
1063 this.workerChoiceStrategyContext
,
1069 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1070 this.workerNodes
[workerNodeKey
].getTaskFunctionWorkerUsage(
1074 const taskFunctionWorkerUsage
= this.workerNodes
[
1076 ].getTaskFunctionWorkerUsage(task
.name
as string) as WorkerUsage
1077 ++taskFunctionWorkerUsage
.tasks
.executing
1078 updateWaitTimeWorkerUsage(
1079 this.workerChoiceStrategyContext
,
1080 taskFunctionWorkerUsage
,
1087 * Hook executed after the worker task execution.
1088 * Can be overridden.
1090 * @param workerNodeKey - The worker node key.
1091 * @param message - The received message.
1093 protected afterTaskExecutionHook (
1094 workerNodeKey
: number,
1095 message
: MessageValue
<Response
>
1097 let needWorkerChoiceStrategyUpdate
= false
1098 if (this.workerNodes
[workerNodeKey
]?.usage
!= null) {
1099 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
1100 updateTaskStatisticsWorkerUsage(workerUsage
, message
)
1101 updateRunTimeWorkerUsage(
1102 this.workerChoiceStrategyContext
,
1106 updateEluWorkerUsage(
1107 this.workerChoiceStrategyContext
,
1111 needWorkerChoiceStrategyUpdate
= true
1114 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1115 this.workerNodes
[workerNodeKey
].getTaskFunctionWorkerUsage(
1116 message
.taskPerformance
?.name
as string
1119 const taskFunctionWorkerUsage
= this.workerNodes
[
1121 ].getTaskFunctionWorkerUsage(
1122 message
.taskPerformance
?.name
as string
1124 updateTaskStatisticsWorkerUsage(taskFunctionWorkerUsage
, message
)
1125 updateRunTimeWorkerUsage(
1126 this.workerChoiceStrategyContext
,
1127 taskFunctionWorkerUsage
,
1130 updateEluWorkerUsage(
1131 this.workerChoiceStrategyContext
,
1132 taskFunctionWorkerUsage
,
1135 needWorkerChoiceStrategyUpdate
= true
1137 if (needWorkerChoiceStrategyUpdate
) {
1138 this.workerChoiceStrategyContext
.update(workerNodeKey
)
1143 * Whether the worker node shall update its task function worker usage or not.
1145 * @param workerNodeKey - The worker node key.
1146 * @returns `true` if the worker node shall update its task function worker usage, `false` otherwise.
1148 private shallUpdateTaskFunctionWorkerUsage (workerNodeKey
: number): boolean {
1149 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1151 workerInfo
!= null &&
1152 Array.isArray(workerInfo
.taskFunctionNames
) &&
1153 workerInfo
.taskFunctionNames
.length
> 2
1158 * Chooses a worker node for the next task.
1160 * The default worker choice strategy uses a round robin algorithm to distribute the tasks.
1162 * @returns The chosen worker node key
1164 private chooseWorkerNode (): number {
1165 if (this.shallCreateDynamicWorker()) {
1166 const workerNodeKey
= this.createAndSetupDynamicWorkerNode()
1168 this.workerChoiceStrategyContext
.getStrategyPolicy().dynamicWorkerUsage
1170 return workerNodeKey
1173 return this.workerChoiceStrategyContext
.execute()
1177 * Conditions for dynamic worker creation.
1179 * @returns Whether to create a dynamic worker or not.
1181 protected abstract shallCreateDynamicWorker (): boolean
1184 * Sends a message to worker given its worker node key.
1186 * @param workerNodeKey - The worker node key.
1187 * @param message - The message.
1188 * @param transferList - The optional array of transferable objects.
1190 protected abstract sendToWorker (
1191 workerNodeKey
: number,
1192 message
: MessageValue
<Data
>,
1193 transferList
?: TransferListItem
[]
1197 * Creates a new, completely set up worker node.
1199 * @returns New, completely set up worker node key.
1201 protected createAndSetupWorkerNode (): number {
1202 const workerNode
= this.createWorkerNode()
1203 workerNode
.registerWorkerEventHandler(
1205 this.opts
.onlineHandler
?? EMPTY_FUNCTION
1207 workerNode
.registerWorkerEventHandler(
1209 this.opts
.messageHandler
?? EMPTY_FUNCTION
1211 workerNode
.registerWorkerEventHandler(
1213 this.opts
.errorHandler
?? EMPTY_FUNCTION
1215 workerNode
.registerWorkerEventHandler('error', (error
: Error) => {
1216 workerNode
.info
.ready
= false
1217 this.emitter
?.emit(PoolEvents
.error
, error
)
1222 this.opts
.restartWorkerOnError
=== true
1224 if (workerNode
.info
.dynamic
) {
1225 this.createAndSetupDynamicWorkerNode()
1227 this.createAndSetupWorkerNode()
1230 if (this.started
&& this.opts
.enableTasksQueue
=== true) {
1231 this.redistributeQueuedTasks(this.workerNodes
.indexOf(workerNode
))
1233 workerNode
?.terminate().catch(error
=> {
1234 this.emitter
?.emit(PoolEvents
.error
, error
)
1237 workerNode
.registerWorkerEventHandler(
1239 this.opts
.exitHandler
?? EMPTY_FUNCTION
1241 workerNode
.registerOnceWorkerEventHandler('exit', () => {
1242 this.removeWorkerNode(workerNode
)
1244 const workerNodeKey
= this.addWorkerNode(workerNode
)
1245 this.afterWorkerNodeSetup(workerNodeKey
)
1246 return workerNodeKey
1250 * Creates a new, completely set up dynamic worker node.
1252 * @returns New, completely set up dynamic worker node key.
1254 protected createAndSetupDynamicWorkerNode (): number {
1255 const workerNodeKey
= this.createAndSetupWorkerNode()
1256 this.registerWorkerMessageListener(workerNodeKey
, message
=> {
1257 this.checkMessageWorkerId(message
)
1258 const localWorkerNodeKey
= this.getWorkerNodeKeyByWorkerId(
1261 const workerUsage
= this.workerNodes
[localWorkerNodeKey
].usage
1262 // Kill message received from worker
1264 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
1265 (isKillBehavior(KillBehaviors
.SOFT
, message
.kill
) &&
1266 ((this.opts
.enableTasksQueue
=== false &&
1267 workerUsage
.tasks
.executing
=== 0) ||
1268 (this.opts
.enableTasksQueue
=== true &&
1269 workerUsage
.tasks
.executing
=== 0 &&
1270 this.tasksQueueSize(localWorkerNodeKey
) === 0)))
1272 // Flag the worker node as not ready immediately
1273 this.flagWorkerNodeAsNotReady(localWorkerNodeKey
)
1274 this.destroyWorkerNode(localWorkerNodeKey
).catch(error
=> {
1275 this.emitter
?.emit(PoolEvents
.error
, error
)
1279 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1280 this.sendToWorker(workerNodeKey
, {
1283 if (this.taskFunctions
.size
> 0) {
1284 for (const [taskFunctionName
, taskFunction
] of this.taskFunctions
) {
1285 this.sendTaskFunctionOperationToWorker(workerNodeKey
, {
1286 taskFunctionOperation
: 'add',
1288 taskFunction
: taskFunction
.toString()
1290 this.emitter
?.emit(PoolEvents
.error
, error
)
1294 workerInfo
.dynamic
= true
1296 this.workerChoiceStrategyContext
.getStrategyPolicy().dynamicWorkerReady
||
1297 this.workerChoiceStrategyContext
.getStrategyPolicy().dynamicWorkerUsage
1299 workerInfo
.ready
= true
1301 this.checkAndEmitDynamicWorkerCreationEvents()
1302 return workerNodeKey
1306 * Registers a listener callback on the worker given its worker node key.
1308 * @param workerNodeKey - The worker node key.
1309 * @param listener - The message listener callback.
1311 protected abstract registerWorkerMessageListener
<
1312 Message
extends Data
| Response
1314 workerNodeKey
: number,
1315 listener
: (message
: MessageValue
<Message
>) => void
1319 * Registers once a listener callback on the worker given its worker node key.
1321 * @param workerNodeKey - The worker node key.
1322 * @param listener - The message listener callback.
1324 protected abstract registerOnceWorkerMessageListener
<
1325 Message
extends Data
| Response
1327 workerNodeKey
: number,
1328 listener
: (message
: MessageValue
<Message
>) => void
1332 * Deregisters a listener callback on the worker given its worker node key.
1334 * @param workerNodeKey - The worker node key.
1335 * @param listener - The message listener callback.
1337 protected abstract deregisterWorkerMessageListener
<
1338 Message
extends Data
| Response
1340 workerNodeKey
: number,
1341 listener
: (message
: MessageValue
<Message
>) => void
1345 * Method hooked up after a worker node has been newly created.
1346 * Can be overridden.
1348 * @param workerNodeKey - The newly created worker node key.
1350 protected afterWorkerNodeSetup (workerNodeKey
: number): void {
1351 // Listen to worker messages.
1352 this.registerWorkerMessageListener(
1354 this.workerMessageListener
1356 // Send the startup message to worker.
1357 this.sendStartupMessageToWorker(workerNodeKey
)
1358 // Send the statistics message to worker.
1359 this.sendStatisticsMessageToWorker(workerNodeKey
)
1360 if (this.opts
.enableTasksQueue
=== true) {
1361 if (this.opts
.tasksQueueOptions
?.taskStealing
=== true) {
1362 this.workerNodes
[workerNodeKey
].on(
1364 this.handleIdleWorkerNodeEvent
1367 if (this.opts
.tasksQueueOptions
?.tasksStealingOnBackPressure
=== true) {
1368 this.workerNodes
[workerNodeKey
].on(
1370 this.handleBackPressureEvent
1377 * Sends the startup message to worker given its worker node key.
1379 * @param workerNodeKey - The worker node key.
1381 protected abstract sendStartupMessageToWorker (workerNodeKey
: number): void
1384 * Sends the statistics message to worker given its worker node key.
1386 * @param workerNodeKey - The worker node key.
1388 private sendStatisticsMessageToWorker (workerNodeKey
: number): void {
1389 this.sendToWorker(workerNodeKey
, {
1392 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1394 elu
: this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1400 private handleTask (workerNodeKey
: number, task
: Task
<Data
>): void {
1401 if (this.shallExecuteTask(workerNodeKey
)) {
1402 this.executeTask(workerNodeKey
, task
)
1404 this.enqueueTask(workerNodeKey
, task
)
1408 private redistributeQueuedTasks (workerNodeKey
: number): void {
1409 if (workerNodeKey
=== -1) {
1412 if (this.workerNodes
.length
<= 1) {
1415 while (this.tasksQueueSize(workerNodeKey
) > 0) {
1416 const destinationWorkerNodeKey
= this.workerNodes
.reduce(
1417 (minWorkerNodeKey
, workerNode
, workerNodeKey
, workerNodes
) => {
1418 return workerNode
.info
.ready
&&
1419 workerNode
.usage
.tasks
.queued
<
1420 workerNodes
[minWorkerNodeKey
].usage
.tasks
.queued
1427 destinationWorkerNodeKey
,
1428 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1433 private updateTaskStolenStatisticsWorkerUsage (
1434 workerNodeKey
: number,
1437 const workerNode
= this.workerNodes
[workerNodeKey
]
1438 if (workerNode
?.usage
!= null) {
1439 ++workerNode
.usage
.tasks
.stolen
1442 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1443 workerNode
.getTaskFunctionWorkerUsage(taskName
) != null
1445 const taskFunctionWorkerUsage
= workerNode
.getTaskFunctionWorkerUsage(
1448 ++taskFunctionWorkerUsage
.tasks
.stolen
1452 private updateTaskSequentiallyStolenStatisticsWorkerUsage (
1453 workerNodeKey
: number
1455 const workerNode
= this.workerNodes
[workerNodeKey
]
1456 if (workerNode
?.usage
!= null) {
1457 ++workerNode
.usage
.tasks
.sequentiallyStolen
1461 private updateTaskSequentiallyStolenStatisticsTaskFunctionWorkerUsage (
1462 workerNodeKey
: number,
1465 const workerNode
= this.workerNodes
[workerNodeKey
]
1467 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1468 workerNode
.getTaskFunctionWorkerUsage(taskName
) != null
1470 const taskFunctionWorkerUsage
= workerNode
.getTaskFunctionWorkerUsage(
1473 ++taskFunctionWorkerUsage
.tasks
.sequentiallyStolen
1477 private resetTaskSequentiallyStolenStatisticsWorkerUsage (
1478 workerNodeKey
: number
1480 const workerNode
= this.workerNodes
[workerNodeKey
]
1481 if (workerNode
?.usage
!= null) {
1482 workerNode
.usage
.tasks
.sequentiallyStolen
= 0
1486 private resetTaskSequentiallyStolenStatisticsTaskFunctionWorkerUsage (
1487 workerNodeKey
: number,
1490 const workerNode
= this.workerNodes
[workerNodeKey
]
1492 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1493 workerNode
.getTaskFunctionWorkerUsage(taskName
) != null
1495 const taskFunctionWorkerUsage
= workerNode
.getTaskFunctionWorkerUsage(
1498 taskFunctionWorkerUsage
.tasks
.sequentiallyStolen
= 0
1502 private readonly handleIdleWorkerNodeEvent
= (
1503 eventDetail
: WorkerNodeEventDetail
,
1504 previousStolenTask
?: Task
<Data
>
1506 if (this.workerNodes
.length
<= 1) {
1509 const { workerNodeKey
} = eventDetail
1510 if (workerNodeKey
== null) {
1512 'WorkerNode event detail workerNodeKey attribute must be defined'
1515 const workerNodeTasksUsage
= this.workerNodes
[workerNodeKey
].usage
.tasks
1517 previousStolenTask
!= null &&
1518 workerNodeTasksUsage
.sequentiallyStolen
> 0 &&
1519 (workerNodeTasksUsage
.executing
> 0 ||
1520 this.tasksQueueSize(workerNodeKey
) > 0)
1522 for (const taskName
of this.workerNodes
[workerNodeKey
].info
1523 .taskFunctionNames
as string[]) {
1524 this.resetTaskSequentiallyStolenStatisticsTaskFunctionWorkerUsage(
1529 this.resetTaskSequentiallyStolenStatisticsWorkerUsage(workerNodeKey
)
1532 const stolenTask
= this.workerNodeStealTask(workerNodeKey
)
1534 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1537 const taskFunctionTasksWorkerUsage
= this.workerNodes
[
1539 ].getTaskFunctionWorkerUsage(stolenTask
.name
as string)
1540 ?.tasks
as TaskStatistics
1542 taskFunctionTasksWorkerUsage
.sequentiallyStolen
=== 0 ||
1543 (previousStolenTask
!= null &&
1544 previousStolenTask
.name
=== stolenTask
.name
&&
1545 taskFunctionTasksWorkerUsage
.sequentiallyStolen
> 0)
1547 this.updateTaskSequentiallyStolenStatisticsTaskFunctionWorkerUsage(
1549 stolenTask
.name
as string
1552 this.resetTaskSequentiallyStolenStatisticsTaskFunctionWorkerUsage(
1554 stolenTask
.name
as string
1558 sleep(exponentialDelay(workerNodeTasksUsage
.sequentiallyStolen
))
1560 this.handleIdleWorkerNodeEvent(eventDetail
, stolenTask
)
1563 .catch(EMPTY_FUNCTION
)
1566 private readonly workerNodeStealTask
= (
1567 workerNodeKey
: number
1568 ): Task
<Data
> | undefined => {
1569 const workerNodes
= this.workerNodes
1572 (workerNodeA
, workerNodeB
) =>
1573 workerNodeB
.usage
.tasks
.queued
- workerNodeA
.usage
.tasks
.queued
1575 const sourceWorkerNode
= workerNodes
.find(
1576 (sourceWorkerNode
, sourceWorkerNodeKey
) =>
1577 sourceWorkerNode
.info
.ready
&&
1578 sourceWorkerNodeKey
!== workerNodeKey
&&
1579 sourceWorkerNode
.usage
.tasks
.queued
> 0
1581 if (sourceWorkerNode
!= null) {
1582 const task
= sourceWorkerNode
.popTask() as Task
<Data
>
1583 this.handleTask(workerNodeKey
, task
)
1584 this.updateTaskSequentiallyStolenStatisticsWorkerUsage(workerNodeKey
)
1585 this.updateTaskStolenStatisticsWorkerUsage(
1593 private readonly handleBackPressureEvent
= (
1594 eventDetail
: WorkerNodeEventDetail
1596 if (this.workerNodes
.length
<= 1) {
1599 const { workerId
} = eventDetail
1600 const sizeOffset
= 1
1601 if ((this.opts
.tasksQueueOptions
?.size
as number) <= sizeOffset
) {
1604 const sourceWorkerNode
=
1605 this.workerNodes
[this.getWorkerNodeKeyByWorkerId(workerId
)]
1606 const workerNodes
= this.workerNodes
1609 (workerNodeA
, workerNodeB
) =>
1610 workerNodeA
.usage
.tasks
.queued
- workerNodeB
.usage
.tasks
.queued
1612 for (const [workerNodeKey
, workerNode
] of workerNodes
.entries()) {
1614 sourceWorkerNode
.usage
.tasks
.queued
> 0 &&
1615 workerNode
.info
.ready
&&
1616 workerNode
.info
.id
!== workerId
&&
1617 workerNode
.usage
.tasks
.queued
<
1618 (this.opts
.tasksQueueOptions
?.size
as number) - sizeOffset
1620 const task
= sourceWorkerNode
.popTask() as Task
<Data
>
1621 this.handleTask(workerNodeKey
, task
)
1622 this.updateTaskStolenStatisticsWorkerUsage(
1631 * This method is the message listener registered on each worker.
1633 protected readonly workerMessageListener
= (
1634 message
: MessageValue
<Response
>
1636 this.checkMessageWorkerId(message
)
1637 const { workerId
, ready
, taskId
, taskFunctionNames
} = message
1638 if (ready
!= null && taskFunctionNames
!= null) {
1639 // Worker ready response received from worker
1640 this.handleWorkerReadyResponse(message
)
1641 } else if (taskId
!= null) {
1642 // Task execution response received from worker
1643 this.handleTaskExecutionResponse(message
)
1644 } else if (taskFunctionNames
!= null) {
1645 // Task function names message received from worker
1647 this.getWorkerNodeKeyByWorkerId(workerId
)
1648 ).taskFunctionNames
= taskFunctionNames
1652 private handleWorkerReadyResponse (message
: MessageValue
<Response
>): void {
1653 const { workerId
, ready
, taskFunctionNames
} = message
1654 if (ready
=== false) {
1655 throw new Error(`Worker ${workerId as number} failed to initialize`)
1657 const workerInfo
= this.getWorkerInfo(
1658 this.getWorkerNodeKeyByWorkerId(workerId
)
1660 workerInfo
.ready
= ready
as boolean
1661 workerInfo
.taskFunctionNames
= taskFunctionNames
1662 if (!this.readyEventEmitted
&& this.ready
) {
1663 this.readyEventEmitted
= true
1664 this.emitter
?.emit(PoolEvents
.ready
, this.info
)
1668 private handleTaskExecutionResponse (message
: MessageValue
<Response
>): void {
1669 const { workerId
, taskId
, workerError
, data
} = message
1670 const promiseResponse
= this.promiseResponseMap
.get(taskId
as string)
1671 if (promiseResponse
!= null) {
1672 const { resolve
, reject
, workerNodeKey
, asyncResource
} = promiseResponse
1673 const workerNode
= this.workerNodes
[workerNodeKey
]
1674 if (workerError
!= null) {
1675 this.emitter
?.emit(PoolEvents
.taskError
, workerError
)
1676 asyncResource
!= null
1677 ? asyncResource
.runInAsyncScope(
1682 : reject(workerError
.message
)
1684 asyncResource
!= null
1685 ? asyncResource
.runInAsyncScope(resolve
, this.emitter
, data
)
1686 : resolve(data
as Response
)
1688 asyncResource
?.emitDestroy()
1689 this.afterTaskExecutionHook(workerNodeKey
, message
)
1690 this.promiseResponseMap
.delete(taskId
as string)
1691 workerNode
?.emit('taskFinished', taskId
)
1692 if (this.opts
.enableTasksQueue
=== true && !this.destroying
) {
1693 const workerNodeTasksUsage
= workerNode
.usage
.tasks
1695 this.tasksQueueSize(workerNodeKey
) > 0 &&
1696 workerNodeTasksUsage
.executing
<
1697 (this.opts
.tasksQueueOptions
?.concurrency
as number)
1701 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1705 workerNodeTasksUsage
.executing
=== 0 &&
1706 this.tasksQueueSize(workerNodeKey
) === 0 &&
1707 workerNodeTasksUsage
.sequentiallyStolen
=== 0
1709 workerNode
.emit('idleWorkerNode', {
1710 workerId
: workerId
as number,
1718 private checkAndEmitTaskExecutionEvents (): void {
1720 this.emitter
?.emit(PoolEvents
.busy
, this.info
)
1724 private checkAndEmitTaskQueuingEvents (): void {
1725 if (this.hasBackPressure()) {
1726 this.emitter
?.emit(PoolEvents
.backPressure
, this.info
)
1730 private checkAndEmitDynamicWorkerCreationEvents (): void {
1731 if (this.type === PoolTypes
.dynamic
) {
1733 this.emitter
?.emit(PoolEvents
.full
, this.info
)
1739 * Gets the worker information given its worker node key.
1741 * @param workerNodeKey - The worker node key.
1742 * @returns The worker information.
1744 protected getWorkerInfo (workerNodeKey
: number): WorkerInfo
{
1745 return this.workerNodes
[workerNodeKey
]?.info
1749 * Creates a worker node.
1751 * @returns The created worker node.
1753 private createWorkerNode (): IWorkerNode
<Worker
, Data
> {
1754 const workerNode
= new WorkerNode
<Worker
, Data
>(
1759 workerOptions
: this.opts
.workerOptions
,
1760 tasksQueueBackPressureSize
:
1761 this.opts
.tasksQueueOptions
?.size
??
1762 getDefaultTasksQueueOptions(
1763 this.maximumNumberOfWorkers
?? this.minimumNumberOfWorkers
1767 // Flag the worker node as ready at pool startup.
1768 if (this.starting
) {
1769 workerNode
.info
.ready
= true
1775 * Adds the given worker node in the pool worker nodes.
1777 * @param workerNode - The worker node.
1778 * @returns The added worker node key.
1779 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the added worker node is not found.
1781 private addWorkerNode (workerNode
: IWorkerNode
<Worker
, Data
>): number {
1782 this.workerNodes
.push(workerNode
)
1783 const workerNodeKey
= this.workerNodes
.indexOf(workerNode
)
1784 if (workerNodeKey
=== -1) {
1785 throw new Error('Worker added not found in worker nodes')
1787 return workerNodeKey
1791 * Removes the worker node from the pool worker nodes.
1793 * @param workerNode - The worker node.
1795 private removeWorkerNode (workerNode
: IWorkerNode
<Worker
, Data
>): void {
1796 const workerNodeKey
= this.workerNodes
.indexOf(workerNode
)
1797 if (workerNodeKey
!== -1) {
1798 this.workerNodes
.splice(workerNodeKey
, 1)
1799 this.workerChoiceStrategyContext
.remove(workerNodeKey
)
1803 protected flagWorkerNodeAsNotReady (workerNodeKey
: number): void {
1804 this.getWorkerInfo(workerNodeKey
).ready
= false
1808 public hasWorkerNodeBackPressure (workerNodeKey
: number): boolean {
1810 this.opts
.enableTasksQueue
=== true &&
1811 this.workerNodes
[workerNodeKey
].hasBackPressure()
1815 private hasBackPressure (): boolean {
1817 this.opts
.enableTasksQueue
=== true &&
1818 this.workerNodes
.findIndex(
1819 workerNode
=> !workerNode
.hasBackPressure()
1825 * Executes the given task on the worker given its worker node key.
1827 * @param workerNodeKey - The worker node key.
1828 * @param task - The task to execute.
1830 private executeTask (workerNodeKey
: number, task
: Task
<Data
>): void {
1831 this.beforeTaskExecutionHook(workerNodeKey
, task
)
1832 this.sendToWorker(workerNodeKey
, task
, task
.transferList
)
1833 this.checkAndEmitTaskExecutionEvents()
1836 private enqueueTask (workerNodeKey
: number, task
: Task
<Data
>): number {
1837 const tasksQueueSize
= this.workerNodes
[workerNodeKey
].enqueueTask(task
)
1838 this.checkAndEmitTaskQueuingEvents()
1839 return tasksQueueSize
1842 private dequeueTask (workerNodeKey
: number): Task
<Data
> | undefined {
1843 return this.workerNodes
[workerNodeKey
].dequeueTask()
1846 private tasksQueueSize (workerNodeKey
: number): number {
1847 return this.workerNodes
[workerNodeKey
].tasksQueueSize()
1850 protected flushTasksQueue (workerNodeKey
: number): number {
1851 let flushedTasks
= 0
1852 while (this.tasksQueueSize(workerNodeKey
) > 0) {
1855 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1859 this.workerNodes
[workerNodeKey
].clearTasksQueue()
1863 private flushTasksQueues (): void {
1864 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
1865 this.flushTasksQueue(workerNodeKey
)