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'
7 PromiseResponseWrapper
,
9 } from
'../utility-types'
12 DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
,
23 import { KillBehaviors
} from
'../worker/worker-options'
24 import type { TaskFunction
} from
'../worker/task-functions'
32 type TasksQueueOptions
38 WorkerNodeEventDetail
,
43 type MeasurementStatisticsRequirements
,
45 WorkerChoiceStrategies
,
46 type WorkerChoiceStrategy
,
47 type WorkerChoiceStrategyOptions
48 } from
'./selection-strategies/selection-strategies-types'
49 import { WorkerChoiceStrategyContext
} from
'./selection-strategies/worker-choice-strategy-context'
50 import { version
} from
'./version'
51 import { WorkerNode
} from
'./worker-node'
54 checkValidTasksQueueOptions
,
55 checkValidWorkerChoiceStrategy
,
56 updateMeasurementStatistics
60 * Base class that implements some shared logic for all poolifier pools.
62 * @typeParam Worker - Type of worker which manages this pool.
63 * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
64 * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
66 export abstract class AbstractPool
<
67 Worker
extends IWorker
,
70 > implements IPool
<Worker
, Data
, Response
> {
72 public readonly workerNodes
: Array<IWorkerNode
<Worker
, Data
>> = []
75 public emitter
?: EventEmitterAsyncResource
78 * Dynamic pool maximum size property placeholder.
80 protected readonly max
?: number
83 * The task execution response promise map:
84 * - `key`: The message id of each submitted task.
85 * - `value`: An object that contains the worker, the execution response promise resolve and reject callbacks.
87 * When we receive a message from the worker, we get a map entry with the promise resolve/reject bound to the message id.
89 protected promiseResponseMap
: Map
<string, PromiseResponseWrapper
<Response
>> =
90 new Map
<string, PromiseResponseWrapper
<Response
>>()
93 * Worker choice strategy context referencing a worker choice algorithm implementation.
95 protected workerChoiceStrategyContext
: WorkerChoiceStrategyContext
<
102 * The task functions added at runtime map:
103 * - `key`: The task function name.
104 * - `value`: The task function itself.
106 private readonly taskFunctions
: Map
<string, TaskFunction
<Data
, Response
>>
109 * Whether the pool is started or not.
111 private started
: boolean
113 * Whether the pool is starting or not.
115 private starting
: boolean
117 * Whether the pool is destroying or not.
119 private destroying
: boolean
121 * The start timestamp of the pool.
123 private readonly startTimestamp
126 * Constructs a new poolifier pool.
128 * @param numberOfWorkers - Number of workers that this pool should manage.
129 * @param filePath - Path to the worker file.
130 * @param opts - Options for the pool.
133 protected readonly numberOfWorkers
: number,
134 protected readonly filePath
: string,
135 protected readonly opts
: PoolOptions
<Worker
>
137 if (!this.isMain()) {
139 'Cannot start a pool from a worker with the same type as the pool'
142 checkFilePath(this.filePath
)
143 this.checkNumberOfWorkers(this.numberOfWorkers
)
144 this.checkPoolOptions(this.opts
)
146 this.chooseWorkerNode
= this.chooseWorkerNode
.bind(this)
147 this.executeTask
= this.executeTask
.bind(this)
148 this.enqueueTask
= this.enqueueTask
.bind(this)
150 if (this.opts
.enableEvents
=== true) {
151 this.initializeEventEmitter()
153 this.workerChoiceStrategyContext
= new WorkerChoiceStrategyContext
<
159 this.opts
.workerChoiceStrategy
,
160 this.opts
.workerChoiceStrategyOptions
165 this.taskFunctions
= new Map
<string, TaskFunction
<Data
, Response
>>()
168 this.starting
= false
169 this.destroying
= false
170 if (this.opts
.startWorkers
=== true) {
174 this.startTimestamp
= performance
.now()
177 private checkNumberOfWorkers (numberOfWorkers
: number): void {
178 if (numberOfWorkers
== null) {
180 'Cannot instantiate a pool without specifying the number of workers'
182 } else if (!Number.isSafeInteger(numberOfWorkers
)) {
184 'Cannot instantiate a pool with a non safe integer number of workers'
186 } else if (numberOfWorkers
< 0) {
187 throw new RangeError(
188 'Cannot instantiate a pool with a negative number of workers'
190 } else if (this.type === PoolTypes
.fixed
&& numberOfWorkers
=== 0) {
191 throw new RangeError('Cannot instantiate a fixed pool with zero worker')
195 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
196 if (isPlainObject(opts
)) {
197 this.opts
.startWorkers
= opts
.startWorkers
?? true
198 checkValidWorkerChoiceStrategy(
199 opts
.workerChoiceStrategy
as WorkerChoiceStrategy
201 this.opts
.workerChoiceStrategy
=
202 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
203 this.checkValidWorkerChoiceStrategyOptions(
204 opts
.workerChoiceStrategyOptions
as WorkerChoiceStrategyOptions
206 this.opts
.workerChoiceStrategyOptions
= {
207 ...DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
,
208 ...opts
.workerChoiceStrategyOptions
210 this.opts
.restartWorkerOnError
= opts
.restartWorkerOnError
?? true
211 this.opts
.enableEvents
= opts
.enableEvents
?? true
212 this.opts
.enableTasksQueue
= opts
.enableTasksQueue
?? false
213 if (this.opts
.enableTasksQueue
) {
214 checkValidTasksQueueOptions(opts
.tasksQueueOptions
as TasksQueueOptions
)
215 this.opts
.tasksQueueOptions
= this.buildTasksQueueOptions(
216 opts
.tasksQueueOptions
as TasksQueueOptions
220 throw new TypeError('Invalid pool options: must be a plain object')
224 private checkValidWorkerChoiceStrategyOptions (
225 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
228 workerChoiceStrategyOptions
!= null &&
229 !isPlainObject(workerChoiceStrategyOptions
)
232 'Invalid worker choice strategy options: must be a plain object'
236 workerChoiceStrategyOptions
?.retries
!= null &&
237 !Number.isSafeInteger(workerChoiceStrategyOptions
.retries
)
240 'Invalid worker choice strategy options: retries must be an integer'
244 workerChoiceStrategyOptions
?.retries
!= null &&
245 workerChoiceStrategyOptions
.retries
< 0
247 throw new RangeError(
248 `Invalid worker choice strategy options: retries '${workerChoiceStrategyOptions.retries}' must be greater or equal than zero`
252 workerChoiceStrategyOptions
?.weights
!= null &&
253 Object.keys(workerChoiceStrategyOptions
.weights
).length
!== this.maxSize
256 'Invalid worker choice strategy options: must have a weight for each worker node'
260 workerChoiceStrategyOptions
?.measurement
!= null &&
261 !Object.values(Measurements
).includes(
262 workerChoiceStrategyOptions
.measurement
266 `Invalid worker choice strategy options: invalid measurement '${workerChoiceStrategyOptions.measurement}'`
271 private initializeEventEmitter (): void {
272 this.emitter
= new EventEmitterAsyncResource({
273 name
: `poolifier:${this.type}-${this.worker}-pool`
278 public get
info (): PoolInfo
{
283 started
: this.started
,
285 strategy
: this.opts
.workerChoiceStrategy
as WorkerChoiceStrategy
,
286 minSize
: this.minSize
,
287 maxSize
: this.maxSize
,
288 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
289 .runTime
.aggregate
&&
290 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
291 .waitTime
.aggregate
&& { utilization
: round(this.utilization
) }),
292 workerNodes
: this.workerNodes
.length
,
293 idleWorkerNodes
: this.workerNodes
.reduce(
294 (accumulator
, workerNode
) =>
295 workerNode
.usage
.tasks
.executing
=== 0
300 busyWorkerNodes
: this.workerNodes
.reduce(
301 (accumulator
, workerNode
) =>
302 workerNode
.usage
.tasks
.executing
> 0 ? accumulator
+ 1 : accumulator
,
305 executedTasks
: this.workerNodes
.reduce(
306 (accumulator
, workerNode
) =>
307 accumulator
+ workerNode
.usage
.tasks
.executed
,
310 executingTasks
: this.workerNodes
.reduce(
311 (accumulator
, workerNode
) =>
312 accumulator
+ workerNode
.usage
.tasks
.executing
,
315 ...(this.opts
.enableTasksQueue
=== true && {
316 queuedTasks
: this.workerNodes
.reduce(
317 (accumulator
, workerNode
) =>
318 accumulator
+ workerNode
.usage
.tasks
.queued
,
322 ...(this.opts
.enableTasksQueue
=== true && {
323 maxQueuedTasks
: this.workerNodes
.reduce(
324 (accumulator
, workerNode
) =>
325 accumulator
+ (workerNode
.usage
.tasks
?.maxQueued
?? 0),
329 ...(this.opts
.enableTasksQueue
=== true && {
330 backPressure
: this.hasBackPressure()
332 ...(this.opts
.enableTasksQueue
=== true && {
333 stolenTasks
: this.workerNodes
.reduce(
334 (accumulator
, workerNode
) =>
335 accumulator
+ workerNode
.usage
.tasks
.stolen
,
339 failedTasks
: this.workerNodes
.reduce(
340 (accumulator
, workerNode
) =>
341 accumulator
+ workerNode
.usage
.tasks
.failed
,
344 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
345 .runTime
.aggregate
&& {
349 ...this.workerNodes
.map(
350 workerNode
=> workerNode
.usage
.runTime
?.minimum
?? Infinity
356 ...this.workerNodes
.map(
357 workerNode
=> workerNode
.usage
.runTime
?.maximum
?? -Infinity
361 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
362 .runTime
.average
&& {
365 this.workerNodes
.reduce
<number[]>(
366 (accumulator
, workerNode
) =>
367 accumulator
.concat(workerNode
.usage
.runTime
.history
),
373 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
377 this.workerNodes
.reduce
<number[]>(
378 (accumulator
, workerNode
) =>
379 accumulator
.concat(workerNode
.usage
.runTime
.history
),
387 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
388 .waitTime
.aggregate
&& {
392 ...this.workerNodes
.map(
393 workerNode
=> workerNode
.usage
.waitTime
?.minimum
?? Infinity
399 ...this.workerNodes
.map(
400 workerNode
=> workerNode
.usage
.waitTime
?.maximum
?? -Infinity
404 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
405 .waitTime
.average
&& {
408 this.workerNodes
.reduce
<number[]>(
409 (accumulator
, workerNode
) =>
410 accumulator
.concat(workerNode
.usage
.waitTime
.history
),
416 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
417 .waitTime
.median
&& {
420 this.workerNodes
.reduce
<number[]>(
421 (accumulator
, workerNode
) =>
422 accumulator
.concat(workerNode
.usage
.waitTime
.history
),
434 * The pool readiness boolean status.
436 private get
ready (): boolean {
438 this.workerNodes
.reduce(
439 (accumulator
, workerNode
) =>
440 !workerNode
.info
.dynamic
&& workerNode
.info
.ready
449 * The approximate pool utilization.
451 * @returns The pool utilization.
453 private get
utilization (): number {
454 const poolTimeCapacity
=
455 (performance
.now() - this.startTimestamp
) * this.maxSize
456 const totalTasksRunTime
= this.workerNodes
.reduce(
457 (accumulator
, workerNode
) =>
458 accumulator
+ (workerNode
.usage
.runTime
?.aggregate
?? 0),
461 const totalTasksWaitTime
= this.workerNodes
.reduce(
462 (accumulator
, workerNode
) =>
463 accumulator
+ (workerNode
.usage
.waitTime
?.aggregate
?? 0),
466 return (totalTasksRunTime
+ totalTasksWaitTime
) / poolTimeCapacity
472 * If it is `'dynamic'`, it provides the `max` property.
474 protected abstract get
type (): PoolType
479 protected abstract get
worker (): WorkerType
482 * The pool minimum size.
484 protected get
minSize (): number {
485 return this.numberOfWorkers
489 * The pool maximum size.
491 protected get
maxSize (): number {
492 return this.max
?? this.numberOfWorkers
496 * Checks if the worker id sent in the received message from a worker is valid.
498 * @param message - The received message.
499 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the worker id is invalid.
501 private checkMessageWorkerId (message
: MessageValue
<Data
| Response
>): void {
502 if (message
.workerId
== null) {
503 throw new Error('Worker message received without worker id')
504 } else if (this.getWorkerNodeKeyByWorkerId(message
.workerId
) === -1) {
506 `Worker message received from unknown worker '${message.workerId}'`
512 * Gets the given worker its worker node key.
514 * @param worker - The worker.
515 * @returns The worker node key if found in the pool worker nodes, `-1` otherwise.
517 private getWorkerNodeKeyByWorker (worker
: Worker
): number {
518 return this.workerNodes
.findIndex(
519 workerNode
=> workerNode
.worker
=== worker
524 * Gets the worker node key given its worker id.
526 * @param workerId - The worker id.
527 * @returns The worker node key if the worker id is found in the pool worker nodes, `-1` otherwise.
529 private getWorkerNodeKeyByWorkerId (workerId
: number | undefined): number {
530 return this.workerNodes
.findIndex(
531 workerNode
=> workerNode
.info
.id
=== workerId
536 public setWorkerChoiceStrategy (
537 workerChoiceStrategy
: WorkerChoiceStrategy
,
538 workerChoiceStrategyOptions
?: WorkerChoiceStrategyOptions
540 checkValidWorkerChoiceStrategy(workerChoiceStrategy
)
541 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
542 this.workerChoiceStrategyContext
.setWorkerChoiceStrategy(
543 this.opts
.workerChoiceStrategy
545 if (workerChoiceStrategyOptions
!= null) {
546 this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
548 for (const [workerNodeKey
, workerNode
] of this.workerNodes
.entries()) {
549 workerNode
.resetUsage()
550 this.sendStatisticsMessageToWorker(workerNodeKey
)
555 public setWorkerChoiceStrategyOptions (
556 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
558 this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
559 this.opts
.workerChoiceStrategyOptions
= {
560 ...DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
,
561 ...workerChoiceStrategyOptions
563 this.workerChoiceStrategyContext
.setOptions(
564 this.opts
.workerChoiceStrategyOptions
569 public enableTasksQueue (
571 tasksQueueOptions
?: TasksQueueOptions
573 if (this.opts
.enableTasksQueue
=== true && !enable
) {
574 this.unsetTaskStealing()
575 this.unsetTasksStealingOnBackPressure()
576 this.flushTasksQueues()
578 this.opts
.enableTasksQueue
= enable
579 this.setTasksQueueOptions(tasksQueueOptions
as TasksQueueOptions
)
583 public setTasksQueueOptions (tasksQueueOptions
: TasksQueueOptions
): void {
584 if (this.opts
.enableTasksQueue
=== true) {
585 checkValidTasksQueueOptions(tasksQueueOptions
)
586 this.opts
.tasksQueueOptions
=
587 this.buildTasksQueueOptions(tasksQueueOptions
)
588 this.setTasksQueueSize(this.opts
.tasksQueueOptions
.size
as number)
589 if (this.opts
.tasksQueueOptions
.taskStealing
=== true) {
590 this.setTaskStealing()
592 this.unsetTaskStealing()
594 if (this.opts
.tasksQueueOptions
.tasksStealingOnBackPressure
=== true) {
595 this.setTasksStealingOnBackPressure()
597 this.unsetTasksStealingOnBackPressure()
599 } else if (this.opts
.tasksQueueOptions
!= null) {
600 delete this.opts
.tasksQueueOptions
604 private buildTasksQueueOptions (
605 tasksQueueOptions
: TasksQueueOptions
606 ): TasksQueueOptions
{
609 size
: Math.pow(this.maxSize
, 2),
612 tasksStealingOnBackPressure
: true
618 private setTasksQueueSize (size
: number): void {
619 for (const workerNode
of this.workerNodes
) {
620 workerNode
.tasksQueueBackPressureSize
= size
624 private setTaskStealing (): void {
625 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
626 this.workerNodes
[workerNodeKey
].addEventListener(
628 this.handleEmptyQueueEvent
as EventListener
633 private unsetTaskStealing (): void {
634 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
635 this.workerNodes
[workerNodeKey
].removeEventListener(
637 this.handleEmptyQueueEvent
as EventListener
642 private setTasksStealingOnBackPressure (): void {
643 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
644 this.workerNodes
[workerNodeKey
].addEventListener(
646 this.handleBackPressureEvent
as EventListener
651 private unsetTasksStealingOnBackPressure (): void {
652 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
653 this.workerNodes
[workerNodeKey
].removeEventListener(
655 this.handleBackPressureEvent
as EventListener
661 * Whether the pool is full or not.
663 * The pool filling boolean status.
665 protected get
full (): boolean {
666 return this.workerNodes
.length
>= this.maxSize
670 * Whether the pool is busy or not.
672 * The pool busyness boolean status.
674 protected abstract get
busy (): boolean
677 * Whether worker nodes are executing concurrently their tasks quota or not.
679 * @returns Worker nodes busyness boolean status.
681 protected internalBusy (): boolean {
682 if (this.opts
.enableTasksQueue
=== true) {
684 this.workerNodes
.findIndex(
686 workerNode
.info
.ready
&&
687 workerNode
.usage
.tasks
.executing
<
688 (this.opts
.tasksQueueOptions
?.concurrency
as number)
693 this.workerNodes
.findIndex(
695 workerNode
.info
.ready
&& workerNode
.usage
.tasks
.executing
=== 0
700 private async sendTaskFunctionOperationToWorker (
701 workerNodeKey
: number,
702 message
: MessageValue
<Data
>
703 ): Promise
<boolean> {
704 return await new Promise
<boolean>((resolve
, reject
) => {
705 const taskFunctionOperationListener
= (
706 message
: MessageValue
<Response
>
708 this.checkMessageWorkerId(message
)
709 const workerId
= this.getWorkerInfo(workerNodeKey
).id
as number
711 message
.taskFunctionOperationStatus
!= null &&
712 message
.workerId
=== workerId
714 if (message
.taskFunctionOperationStatus
) {
716 } else if (!message
.taskFunctionOperationStatus
) {
719 `Task function operation '${
720 message.taskFunctionOperation as string
721 }' failed on worker ${message.workerId} with error: '${
722 message.workerError?.message as string
727 this.deregisterWorkerMessageListener(
728 this.getWorkerNodeKeyByWorkerId(message
.workerId
),
729 taskFunctionOperationListener
733 this.registerWorkerMessageListener(
735 taskFunctionOperationListener
737 this.sendToWorker(workerNodeKey
, message
)
741 private async sendTaskFunctionOperationToWorkers (
742 message
: MessageValue
<Data
>
743 ): Promise
<boolean> {
744 return await new Promise
<boolean>((resolve
, reject
) => {
745 const responsesReceived
= new Array<MessageValue
<Response
>>()
746 const taskFunctionOperationsListener
= (
747 message
: MessageValue
<Response
>
749 this.checkMessageWorkerId(message
)
750 if (message
.taskFunctionOperationStatus
!= null) {
751 responsesReceived
.push(message
)
752 if (responsesReceived
.length
=== this.workerNodes
.length
) {
754 responsesReceived
.every(
755 message
=> message
.taskFunctionOperationStatus
=== true
760 responsesReceived
.some(
761 message
=> message
.taskFunctionOperationStatus
=== false
764 const errorResponse
= responsesReceived
.find(
765 response
=> response
.taskFunctionOperationStatus
=== false
769 `Task function operation '${
770 message.taskFunctionOperation as string
771 }' failed on worker ${
772 errorResponse?.workerId as number
774 errorResponse?.workerError?.message as string
779 this.deregisterWorkerMessageListener(
780 this.getWorkerNodeKeyByWorkerId(message
.workerId
),
781 taskFunctionOperationsListener
786 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
787 this.registerWorkerMessageListener(
789 taskFunctionOperationsListener
791 this.sendToWorker(workerNodeKey
, message
)
797 public hasTaskFunction (name
: string): boolean {
798 for (const workerNode
of this.workerNodes
) {
800 Array.isArray(workerNode
.info
.taskFunctionNames
) &&
801 workerNode
.info
.taskFunctionNames
.includes(name
)
810 public async addTaskFunction (
812 fn
: TaskFunction
<Data
, Response
>
813 ): Promise
<boolean> {
814 if (typeof name
!== 'string') {
815 throw new TypeError('name argument must be a string')
817 if (typeof name
=== 'string' && name
.trim().length
=== 0) {
818 throw new TypeError('name argument must not be an empty string')
820 if (typeof fn
!== 'function') {
821 throw new TypeError('fn argument must be a function')
823 const opResult
= await this.sendTaskFunctionOperationToWorkers({
824 taskFunctionOperation
: 'add',
825 taskFunctionName
: name
,
826 taskFunction
: fn
.toString()
828 this.taskFunctions
.set(name
, fn
)
833 public async removeTaskFunction (name
: string): Promise
<boolean> {
834 if (!this.taskFunctions
.has(name
)) {
836 'Cannot remove a task function not handled on the pool side'
839 const opResult
= await this.sendTaskFunctionOperationToWorkers({
840 taskFunctionOperation
: 'remove',
841 taskFunctionName
: name
843 this.deleteTaskFunctionWorkerUsages(name
)
844 this.taskFunctions
.delete(name
)
849 public listTaskFunctionNames (): string[] {
850 for (const workerNode
of this.workerNodes
) {
852 Array.isArray(workerNode
.info
.taskFunctionNames
) &&
853 workerNode
.info
.taskFunctionNames
.length
> 0
855 return workerNode
.info
.taskFunctionNames
862 public async setDefaultTaskFunction (name
: string): Promise
<boolean> {
863 return await this.sendTaskFunctionOperationToWorkers({
864 taskFunctionOperation
: 'default',
865 taskFunctionName
: name
869 private deleteTaskFunctionWorkerUsages (name
: string): void {
870 for (const workerNode
of this.workerNodes
) {
871 workerNode
.deleteTaskFunctionWorkerUsage(name
)
875 private shallExecuteTask (workerNodeKey
: number): boolean {
877 this.tasksQueueSize(workerNodeKey
) === 0 &&
878 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
<
879 (this.opts
.tasksQueueOptions
?.concurrency
as number)
884 public async execute (
887 transferList
?: TransferListItem
[]
888 ): Promise
<Response
> {
889 return await new Promise
<Response
>((resolve
, reject
) => {
891 reject(new Error('Cannot execute a task on not started pool'))
894 if (this.destroying
) {
895 reject(new Error('Cannot execute a task on destroying pool'))
898 if (name
!= null && typeof name
!== 'string') {
899 reject(new TypeError('name argument must be a string'))
904 typeof name
=== 'string' &&
905 name
.trim().length
=== 0
907 reject(new TypeError('name argument must not be an empty string'))
910 if (transferList
!= null && !Array.isArray(transferList
)) {
911 reject(new TypeError('transferList argument must be an array'))
914 const timestamp
= performance
.now()
915 const workerNodeKey
= this.chooseWorkerNode()
916 const task
: Task
<Data
> = {
917 name
: name
?? DEFAULT_TASK_NAME
,
918 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
919 data
: data
?? ({} as Data
),
924 this.promiseResponseMap
.set(task
.taskId
as string, {
930 this.opts
.enableTasksQueue
=== false ||
931 (this.opts
.enableTasksQueue
=== true &&
932 this.shallExecuteTask(workerNodeKey
))
934 this.executeTask(workerNodeKey
, task
)
936 this.enqueueTask(workerNodeKey
, task
)
942 public start (): void {
944 throw new Error('Cannot start an already started pool')
947 throw new Error('Cannot start an already starting pool')
949 if (this.destroying
) {
950 throw new Error('Cannot start a destroying pool')
954 this.workerNodes
.reduce(
955 (accumulator
, workerNode
) =>
956 !workerNode
.info
.dynamic
? accumulator
+ 1 : accumulator
,
958 ) < this.numberOfWorkers
960 this.createAndSetupWorkerNode()
962 this.starting
= false
967 public async destroy (): Promise
<void> {
969 throw new Error('Cannot destroy an already destroyed pool')
972 throw new Error('Cannot destroy an starting pool')
974 if (this.destroying
) {
975 throw new Error('Cannot destroy an already destroying pool')
977 this.destroying
= true
979 this.workerNodes
.map(async (_
, workerNodeKey
) => {
980 await this.destroyWorkerNode(workerNodeKey
)
983 this.emitter
?.emit(PoolEvents
.destroy
, this.info
)
984 this.emitter
?.emitDestroy()
985 this.destroying
= false
989 protected async sendKillMessageToWorker (
990 workerNodeKey
: number
992 await new Promise
<void>((resolve
, reject
) => {
993 const killMessageListener
= (message
: MessageValue
<Response
>): void => {
994 this.checkMessageWorkerId(message
)
995 if (message
.kill
=== 'success') {
997 } else if (message
.kill
=== 'failure') {
1000 `Kill message handling failed on worker ${
1001 message.workerId as number
1007 // FIXME: should be registered only once
1008 this.registerWorkerMessageListener(workerNodeKey
, killMessageListener
)
1009 this.sendToWorker(workerNodeKey
, { kill
: true })
1014 * Terminates the worker node given its worker node key.
1016 * @param workerNodeKey - The worker node key.
1018 protected abstract destroyWorkerNode (workerNodeKey
: number): Promise
<void>
1021 * Setup hook to execute code before worker nodes are created in the abstract constructor.
1022 * Can be overridden.
1026 protected setupHook (): void {
1027 /* Intentionally empty */
1031 * Should return whether the worker is the main worker or not.
1033 protected abstract isMain (): boolean
1036 * Hook executed before the worker task execution.
1037 * Can be overridden.
1039 * @param workerNodeKey - The worker node key.
1040 * @param task - The task to execute.
1042 protected beforeTaskExecutionHook (
1043 workerNodeKey
: number,
1046 if (this.workerNodes
[workerNodeKey
]?.usage
!= null) {
1047 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
1048 ++workerUsage
.tasks
.executing
1049 this.updateWaitTimeWorkerUsage(workerUsage
, task
)
1052 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1053 this.workerNodes
[workerNodeKey
].getTaskFunctionWorkerUsage(
1057 const taskFunctionWorkerUsage
= this.workerNodes
[
1059 ].getTaskFunctionWorkerUsage(task
.name
as string) as WorkerUsage
1060 ++taskFunctionWorkerUsage
.tasks
.executing
1061 this.updateWaitTimeWorkerUsage(taskFunctionWorkerUsage
, task
)
1066 * Hook executed after the worker task execution.
1067 * Can be overridden.
1069 * @param workerNodeKey - The worker node key.
1070 * @param message - The received message.
1072 protected afterTaskExecutionHook (
1073 workerNodeKey
: number,
1074 message
: MessageValue
<Response
>
1076 if (this.workerNodes
[workerNodeKey
]?.usage
!= null) {
1077 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
1078 this.updateTaskStatisticsWorkerUsage(workerUsage
, message
)
1079 this.updateRunTimeWorkerUsage(workerUsage
, message
)
1080 this.updateEluWorkerUsage(workerUsage
, message
)
1083 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1084 this.workerNodes
[workerNodeKey
].getTaskFunctionWorkerUsage(
1085 message
.taskPerformance
?.name
as string
1088 const taskFunctionWorkerUsage
= this.workerNodes
[
1090 ].getTaskFunctionWorkerUsage(
1091 message
.taskPerformance
?.name
as string
1093 this.updateTaskStatisticsWorkerUsage(taskFunctionWorkerUsage
, message
)
1094 this.updateRunTimeWorkerUsage(taskFunctionWorkerUsage
, message
)
1095 this.updateEluWorkerUsage(taskFunctionWorkerUsage
, message
)
1100 * Whether the worker node shall update its task function worker usage or not.
1102 * @param workerNodeKey - The worker node key.
1103 * @returns `true` if the worker node shall update its task function worker usage, `false` otherwise.
1105 private shallUpdateTaskFunctionWorkerUsage (workerNodeKey
: number): boolean {
1106 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1108 workerInfo
!= null &&
1109 Array.isArray(workerInfo
.taskFunctionNames
) &&
1110 workerInfo
.taskFunctionNames
.length
> 2
1114 private updateTaskStatisticsWorkerUsage (
1115 workerUsage
: WorkerUsage
,
1116 message
: MessageValue
<Response
>
1118 const workerTaskStatistics
= workerUsage
.tasks
1120 workerTaskStatistics
.executing
!= null &&
1121 workerTaskStatistics
.executing
> 0
1123 --workerTaskStatistics
.executing
1125 if (message
.workerError
== null) {
1126 ++workerTaskStatistics
.executed
1128 ++workerTaskStatistics
.failed
1132 private updateRunTimeWorkerUsage (
1133 workerUsage
: WorkerUsage
,
1134 message
: MessageValue
<Response
>
1136 if (message
.workerError
!= null) {
1139 updateMeasurementStatistics(
1140 workerUsage
.runTime
,
1141 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
,
1142 message
.taskPerformance
?.runTime
?? 0
1146 private updateWaitTimeWorkerUsage (
1147 workerUsage
: WorkerUsage
,
1150 const timestamp
= performance
.now()
1151 const taskWaitTime
= timestamp
- (task
.timestamp
?? timestamp
)
1152 updateMeasurementStatistics(
1153 workerUsage
.waitTime
,
1154 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().waitTime
,
1159 private updateEluWorkerUsage (
1160 workerUsage
: WorkerUsage
,
1161 message
: MessageValue
<Response
>
1163 if (message
.workerError
!= null) {
1166 const eluTaskStatisticsRequirements
: MeasurementStatisticsRequirements
=
1167 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
1168 updateMeasurementStatistics(
1169 workerUsage
.elu
.active
,
1170 eluTaskStatisticsRequirements
,
1171 message
.taskPerformance
?.elu
?.active
?? 0
1173 updateMeasurementStatistics(
1174 workerUsage
.elu
.idle
,
1175 eluTaskStatisticsRequirements
,
1176 message
.taskPerformance
?.elu
?.idle
?? 0
1178 if (eluTaskStatisticsRequirements
.aggregate
) {
1179 if (message
.taskPerformance
?.elu
!= null) {
1180 if (workerUsage
.elu
.utilization
!= null) {
1181 workerUsage
.elu
.utilization
=
1182 (workerUsage
.elu
.utilization
+
1183 message
.taskPerformance
.elu
.utilization
) /
1186 workerUsage
.elu
.utilization
= message
.taskPerformance
.elu
.utilization
1193 * Chooses a worker node for the next task.
1195 * The default worker choice strategy uses a round robin algorithm to distribute the tasks.
1197 * @returns The chosen worker node key
1199 private chooseWorkerNode (): number {
1200 if (this.shallCreateDynamicWorker()) {
1201 const workerNodeKey
= this.createAndSetupDynamicWorkerNode()
1203 this.workerChoiceStrategyContext
.getStrategyPolicy().dynamicWorkerUsage
1205 return workerNodeKey
1208 return this.workerChoiceStrategyContext
.execute()
1212 * Conditions for dynamic worker creation.
1214 * @returns Whether to create a dynamic worker or not.
1216 private shallCreateDynamicWorker (): boolean {
1217 return this.type === PoolTypes
.dynamic
&& !this.full
&& this.internalBusy()
1221 * Sends a message to worker given its worker node key.
1223 * @param workerNodeKey - The worker node key.
1224 * @param message - The message.
1225 * @param transferList - The optional array of transferable objects.
1227 protected abstract sendToWorker (
1228 workerNodeKey
: number,
1229 message
: MessageValue
<Data
>,
1230 transferList
?: TransferListItem
[]
1234 * Creates a new worker.
1236 * @returns Newly created worker.
1238 protected abstract createWorker (): Worker
1241 * Creates a new, completely set up worker node.
1243 * @returns New, completely set up worker node key.
1245 protected createAndSetupWorkerNode (): number {
1246 const worker
= this.createWorker()
1248 worker
.on('online', this.opts
.onlineHandler
?? EMPTY_FUNCTION
)
1249 worker
.on('message', this.opts
.messageHandler
?? EMPTY_FUNCTION
)
1250 worker
.on('error', this.opts
.errorHandler
?? EMPTY_FUNCTION
)
1251 worker
.on('error', error
=> {
1252 const workerNodeKey
= this.getWorkerNodeKeyByWorker(worker
)
1253 this.flagWorkerNodeAsNotReady(workerNodeKey
)
1254 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1255 this.emitter
?.emit(PoolEvents
.error
, error
)
1256 this.workerNodes
[workerNodeKey
].closeChannel()
1261 this.opts
.restartWorkerOnError
=== true
1263 if (workerInfo
.dynamic
) {
1264 this.createAndSetupDynamicWorkerNode()
1266 this.createAndSetupWorkerNode()
1272 this.opts
.enableTasksQueue
=== true
1274 this.redistributeQueuedTasks(workerNodeKey
)
1277 worker
.on('exit', this.opts
.exitHandler
?? EMPTY_FUNCTION
)
1278 worker
.once('exit', () => {
1279 this.removeWorkerNode(worker
)
1282 const workerNodeKey
= this.addWorkerNode(worker
)
1284 this.afterWorkerNodeSetup(workerNodeKey
)
1286 return workerNodeKey
1290 * Creates a new, completely set up dynamic worker node.
1292 * @returns New, completely set up dynamic worker node key.
1294 protected createAndSetupDynamicWorkerNode (): number {
1295 const workerNodeKey
= this.createAndSetupWorkerNode()
1296 this.registerWorkerMessageListener(workerNodeKey
, message
=> {
1297 this.checkMessageWorkerId(message
)
1298 const localWorkerNodeKey
= this.getWorkerNodeKeyByWorkerId(
1301 const workerUsage
= this.workerNodes
[localWorkerNodeKey
].usage
1302 // Kill message received from worker
1304 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
1305 (isKillBehavior(KillBehaviors
.SOFT
, message
.kill
) &&
1306 ((this.opts
.enableTasksQueue
=== false &&
1307 workerUsage
.tasks
.executing
=== 0) ||
1308 (this.opts
.enableTasksQueue
=== true &&
1309 workerUsage
.tasks
.executing
=== 0 &&
1310 this.tasksQueueSize(localWorkerNodeKey
) === 0)))
1312 // Flag the worker node as not ready immediately
1313 this.flagWorkerNodeAsNotReady(localWorkerNodeKey
)
1314 this.destroyWorkerNode(localWorkerNodeKey
).catch(error
=> {
1315 this.emitter
?.emit(PoolEvents
.error
, error
)
1319 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1320 this.sendToWorker(workerNodeKey
, {
1323 if (this.taskFunctions
.size
> 0) {
1324 for (const [taskFunctionName
, taskFunction
] of this.taskFunctions
) {
1325 this.sendTaskFunctionOperationToWorker(workerNodeKey
, {
1326 taskFunctionOperation
: 'add',
1328 taskFunction
: taskFunction
.toString()
1330 this.emitter
?.emit(PoolEvents
.error
, error
)
1334 workerInfo
.dynamic
= true
1336 this.workerChoiceStrategyContext
.getStrategyPolicy().dynamicWorkerReady
||
1337 this.workerChoiceStrategyContext
.getStrategyPolicy().dynamicWorkerUsage
1339 workerInfo
.ready
= true
1341 this.checkAndEmitDynamicWorkerCreationEvents()
1342 return workerNodeKey
1346 * Registers a listener callback on the worker given its worker node key.
1348 * @param workerNodeKey - The worker node key.
1349 * @param listener - The message listener callback.
1351 protected abstract registerWorkerMessageListener
<
1352 Message
extends Data
| Response
1354 workerNodeKey
: number,
1355 listener
: (message
: MessageValue
<Message
>) => void
1359 * Registers once a listener callback on the worker given its worker node key.
1361 * @param workerNodeKey - The worker node key.
1362 * @param listener - The message listener callback.
1364 protected abstract registerOnceWorkerMessageListener
<
1365 Message
extends Data
| Response
1367 workerNodeKey
: number,
1368 listener
: (message
: MessageValue
<Message
>) => void
1372 * Deregisters a listener callback on the worker given its worker node key.
1374 * @param workerNodeKey - The worker node key.
1375 * @param listener - The message listener callback.
1377 protected abstract deregisterWorkerMessageListener
<
1378 Message
extends Data
| Response
1380 workerNodeKey
: number,
1381 listener
: (message
: MessageValue
<Message
>) => void
1385 * Method hooked up after a worker node has been newly created.
1386 * Can be overridden.
1388 * @param workerNodeKey - The newly created worker node key.
1390 protected afterWorkerNodeSetup (workerNodeKey
: number): void {
1391 // Listen to worker messages.
1392 this.registerWorkerMessageListener(
1394 this.workerMessageListener
.bind(this)
1396 // Send the startup message to worker.
1397 this.sendStartupMessageToWorker(workerNodeKey
)
1398 // Send the statistics message to worker.
1399 this.sendStatisticsMessageToWorker(workerNodeKey
)
1400 if (this.opts
.enableTasksQueue
=== true) {
1401 if (this.opts
.tasksQueueOptions
?.taskStealing
=== true) {
1402 this.workerNodes
[workerNodeKey
].addEventListener(
1404 this.handleEmptyQueueEvent
as EventListener
1407 if (this.opts
.tasksQueueOptions
?.tasksStealingOnBackPressure
=== true) {
1408 this.workerNodes
[workerNodeKey
].addEventListener(
1410 this.handleBackPressureEvent
as EventListener
1417 * Sends the startup message to worker given its worker node key.
1419 * @param workerNodeKey - The worker node key.
1421 protected abstract sendStartupMessageToWorker (workerNodeKey
: number): void
1424 * Sends the statistics message to worker given its worker node key.
1426 * @param workerNodeKey - The worker node key.
1428 private sendStatisticsMessageToWorker (workerNodeKey
: number): void {
1429 this.sendToWorker(workerNodeKey
, {
1432 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1434 elu
: this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1440 private redistributeQueuedTasks (workerNodeKey
: number): void {
1441 while (this.tasksQueueSize(workerNodeKey
) > 0) {
1442 const destinationWorkerNodeKey
= this.workerNodes
.reduce(
1443 (minWorkerNodeKey
, workerNode
, workerNodeKey
, workerNodes
) => {
1444 return workerNode
.info
.ready
&&
1445 workerNode
.usage
.tasks
.queued
<
1446 workerNodes
[minWorkerNodeKey
].usage
.tasks
.queued
1452 const task
= this.dequeueTask(workerNodeKey
) as Task
<Data
>
1453 if (this.shallExecuteTask(destinationWorkerNodeKey
)) {
1454 this.executeTask(destinationWorkerNodeKey
, task
)
1456 this.enqueueTask(destinationWorkerNodeKey
, task
)
1461 private updateTaskStolenStatisticsWorkerUsage (
1462 workerNodeKey
: number,
1465 const workerNode
= this.workerNodes
[workerNodeKey
]
1466 if (workerNode
?.usage
!= null) {
1467 ++workerNode
.usage
.tasks
.stolen
1470 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1471 workerNode
.getTaskFunctionWorkerUsage(taskName
) != null
1473 const taskFunctionWorkerUsage
= workerNode
.getTaskFunctionWorkerUsage(
1476 ++taskFunctionWorkerUsage
.tasks
.stolen
1480 private readonly handleEmptyQueueEvent
= (
1481 event
: CustomEvent
<WorkerNodeEventDetail
>
1483 const destinationWorkerNodeKey
= this.getWorkerNodeKeyByWorkerId(
1484 event
.detail
.workerId
1486 const workerNodes
= this.workerNodes
1489 (workerNodeA
, workerNodeB
) =>
1490 workerNodeB
.usage
.tasks
.queued
- workerNodeA
.usage
.tasks
.queued
1492 const sourceWorkerNode
= workerNodes
.find(
1494 workerNode
.info
.ready
&&
1495 workerNode
.info
.id
!== event
.detail
.workerId
&&
1496 workerNode
.usage
.tasks
.queued
> 0
1498 if (sourceWorkerNode
!= null) {
1499 const task
= sourceWorkerNode
.popTask() as Task
<Data
>
1500 if (this.shallExecuteTask(destinationWorkerNodeKey
)) {
1501 this.executeTask(destinationWorkerNodeKey
, task
)
1503 this.enqueueTask(destinationWorkerNodeKey
, task
)
1505 this.updateTaskStolenStatisticsWorkerUsage(
1506 destinationWorkerNodeKey
,
1512 private readonly handleBackPressureEvent
= (
1513 event
: CustomEvent
<WorkerNodeEventDetail
>
1515 const sizeOffset
= 1
1516 if ((this.opts
.tasksQueueOptions
?.size
as number) <= sizeOffset
) {
1519 const sourceWorkerNode
=
1520 this.workerNodes
[this.getWorkerNodeKeyByWorkerId(event
.detail
.workerId
)]
1521 const workerNodes
= this.workerNodes
1524 (workerNodeA
, workerNodeB
) =>
1525 workerNodeA
.usage
.tasks
.queued
- workerNodeB
.usage
.tasks
.queued
1527 for (const [workerNodeKey
, workerNode
] of workerNodes
.entries()) {
1529 sourceWorkerNode
.usage
.tasks
.queued
> 0 &&
1530 workerNode
.info
.ready
&&
1531 workerNode
.info
.id
!== event
.detail
.workerId
&&
1532 workerNode
.usage
.tasks
.queued
<
1533 (this.opts
.tasksQueueOptions
?.size
as number) - sizeOffset
1535 const task
= sourceWorkerNode
.popTask() as Task
<Data
>
1536 if (this.shallExecuteTask(workerNodeKey
)) {
1537 this.executeTask(workerNodeKey
, task
)
1539 this.enqueueTask(workerNodeKey
, task
)
1541 this.updateTaskStolenStatisticsWorkerUsage(
1550 * This method is the message listener registered on each worker.
1552 protected workerMessageListener (message
: MessageValue
<Response
>): void {
1553 this.checkMessageWorkerId(message
)
1554 if (message
.ready
!= null && message
.taskFunctionNames
!= null) {
1555 // Worker ready response received from worker
1556 this.handleWorkerReadyResponse(message
)
1557 } else if (message
.taskId
!= null) {
1558 // Task execution response received from worker
1559 this.handleTaskExecutionResponse(message
)
1560 } else if (message
.taskFunctionNames
!= null) {
1561 // Task function names message received from worker
1563 this.getWorkerNodeKeyByWorkerId(message
.workerId
)
1564 ).taskFunctionNames
= message
.taskFunctionNames
1568 private handleWorkerReadyResponse (message
: MessageValue
<Response
>): void {
1569 if (message
.ready
=== false) {
1571 `Worker ${message.workerId as number} failed to initialize`
1574 const workerInfo
= this.getWorkerInfo(
1575 this.getWorkerNodeKeyByWorkerId(message
.workerId
)
1577 workerInfo
.ready
= message
.ready
as boolean
1578 workerInfo
.taskFunctionNames
= message
.taskFunctionNames
1580 const emitPoolReadyEventOnce
= once(
1581 () => this.emitter
?.emit(PoolEvents
.ready
, this.info
),
1584 emitPoolReadyEventOnce()
1588 private handleTaskExecutionResponse (message
: MessageValue
<Response
>): void {
1589 const { taskId
, workerError
, data
} = message
1590 const promiseResponse
= this.promiseResponseMap
.get(taskId
as string)
1591 if (promiseResponse
!= null) {
1592 if (workerError
!= null) {
1593 this.emitter
?.emit(PoolEvents
.taskError
, workerError
)
1594 promiseResponse
.reject(workerError
.message
)
1596 promiseResponse
.resolve(data
as Response
)
1598 const workerNodeKey
= promiseResponse
.workerNodeKey
1599 this.afterTaskExecutionHook(workerNodeKey
, message
)
1600 this.workerChoiceStrategyContext
.update(workerNodeKey
)
1601 this.promiseResponseMap
.delete(taskId
as string)
1603 this.opts
.enableTasksQueue
=== true &&
1604 this.tasksQueueSize(workerNodeKey
) > 0 &&
1605 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
<
1606 (this.opts
.tasksQueueOptions
?.concurrency
as number)
1610 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1616 private checkAndEmitTaskExecutionEvents (): void {
1618 this.emitter
?.emit(PoolEvents
.busy
, this.info
)
1622 private checkAndEmitTaskQueuingEvents (): void {
1623 if (this.hasBackPressure()) {
1624 this.emitter
?.emit(PoolEvents
.backPressure
, this.info
)
1628 private checkAndEmitDynamicWorkerCreationEvents (): void {
1629 if (this.type === PoolTypes
.dynamic
) {
1631 this.emitter
?.emit(PoolEvents
.full
, this.info
)
1637 * Gets the worker information given its worker node key.
1639 * @param workerNodeKey - The worker node key.
1640 * @returns The worker information.
1642 protected getWorkerInfo (workerNodeKey
: number): WorkerInfo
{
1643 return this.workerNodes
[workerNodeKey
]?.info
1647 * Adds the given worker in the pool worker nodes.
1649 * @param worker - The worker.
1650 * @returns The added worker node key.
1651 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the added worker node is not found.
1653 private addWorkerNode (worker
: Worker
): number {
1654 const workerNode
= new WorkerNode
<Worker
, Data
>(
1656 this.opts
.tasksQueueOptions
?.size
?? Math.pow(this.maxSize
, 2)
1658 // Flag the worker node as ready at pool startup.
1659 if (this.starting
) {
1660 workerNode
.info
.ready
= true
1662 this.workerNodes
.push(workerNode
)
1663 const workerNodeKey
= this.getWorkerNodeKeyByWorker(worker
)
1664 if (workerNodeKey
=== -1) {
1665 throw new Error('Worker added not found in worker nodes')
1667 return workerNodeKey
1671 * Removes the given worker from the pool worker nodes.
1673 * @param worker - The worker.
1675 private removeWorkerNode (worker
: Worker
): void {
1676 const workerNodeKey
= this.getWorkerNodeKeyByWorker(worker
)
1677 if (workerNodeKey
!== -1) {
1678 this.workerNodes
.splice(workerNodeKey
, 1)
1679 this.workerChoiceStrategyContext
.remove(workerNodeKey
)
1683 protected flagWorkerNodeAsNotReady (workerNodeKey
: number): void {
1684 this.getWorkerInfo(workerNodeKey
).ready
= false
1688 public hasWorkerNodeBackPressure (workerNodeKey
: number): boolean {
1690 this.opts
.enableTasksQueue
=== true &&
1691 this.workerNodes
[workerNodeKey
].hasBackPressure()
1695 private hasBackPressure (): boolean {
1697 this.opts
.enableTasksQueue
=== true &&
1698 this.workerNodes
.findIndex(
1699 workerNode
=> !workerNode
.hasBackPressure()
1705 * Executes the given task on the worker given its worker node key.
1707 * @param workerNodeKey - The worker node key.
1708 * @param task - The task to execute.
1710 private executeTask (workerNodeKey
: number, task
: Task
<Data
>): void {
1711 this.beforeTaskExecutionHook(workerNodeKey
, task
)
1712 this.sendToWorker(workerNodeKey
, task
, task
.transferList
)
1713 this.checkAndEmitTaskExecutionEvents()
1716 private enqueueTask (workerNodeKey
: number, task
: Task
<Data
>): number {
1717 const tasksQueueSize
= this.workerNodes
[workerNodeKey
].enqueueTask(task
)
1718 this.checkAndEmitTaskQueuingEvents()
1719 return tasksQueueSize
1722 private dequeueTask (workerNodeKey
: number): Task
<Data
> | undefined {
1723 return this.workerNodes
[workerNodeKey
].dequeueTask()
1726 private tasksQueueSize (workerNodeKey
: number): number {
1727 return this.workerNodes
[workerNodeKey
].tasksQueueSize()
1730 protected flushTasksQueue (workerNodeKey
: number): void {
1731 while (this.tasksQueueSize(workerNodeKey
) > 0) {
1734 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1737 this.workerNodes
[workerNodeKey
].clearTasksQueue()
1740 private flushTasksQueues (): void {
1741 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
1742 this.flushTasksQueue(workerNodeKey
)