1 import { randomUUID
} from
'node:crypto'
2 import { performance
} from
'node:perf_hooks'
3 import { existsSync
} from
'node:fs'
4 import { type TransferListItem
} from
'node:worker_threads'
7 PromiseResponseWrapper
,
9 } from
'../utility-types'
12 DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
,
21 updateMeasurementStatistics
23 import { KillBehaviors
} from
'../worker/worker-options'
24 import type { TaskFunction
} from
'../worker/task-functions'
33 type TasksQueueOptions
43 type MeasurementStatisticsRequirements
,
45 WorkerChoiceStrategies
,
46 type WorkerChoiceStrategy
,
47 type WorkerChoiceStrategyOptions
48 } from
'./selection-strategies/selection-strategies-types'
49 import { WorkerChoiceStrategyContext
} from
'./selection-strategies/worker-choice-strategy-context'
50 import { version
} from
'./version'
51 import { WorkerNode
} from
'./worker-node'
54 * Base class that implements some shared logic for all poolifier pools.
56 * @typeParam Worker - Type of worker which manages this pool.
57 * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
58 * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
60 export abstract class AbstractPool
<
61 Worker
extends IWorker
,
64 > implements IPool
<Worker
, Data
, Response
> {
66 public readonly workerNodes
: Array<IWorkerNode
<Worker
, Data
>> = []
69 public readonly emitter
?: PoolEmitter
72 * The task execution response promise map:
73 * - `key`: The message id of each submitted task.
74 * - `value`: An object that contains the worker, the execution response promise resolve and reject callbacks.
76 * When we receive a message from the worker, we get a map entry with the promise resolve/reject bound to the message id.
78 protected promiseResponseMap
: Map
<string, PromiseResponseWrapper
<Response
>> =
79 new Map
<string, PromiseResponseWrapper
<Response
>>()
82 * Worker choice strategy context referencing a worker choice algorithm implementation.
84 protected workerChoiceStrategyContext
: WorkerChoiceStrategyContext
<
91 * Dynamic pool maximum size property placeholder.
93 protected readonly max
?: number
96 * The task functions added at runtime map:
97 * - `key`: The task function name.
98 * - `value`: The task function itself.
100 private readonly taskFunctions
: Map
<string, TaskFunction
<Data
, Response
>>
103 * Whether the pool is started or not.
105 private started
: boolean
107 * Whether the pool is starting or not.
109 private starting
: boolean
111 * The start timestamp of the pool.
113 private readonly startTimestamp
116 * Constructs a new poolifier pool.
118 * @param numberOfWorkers - Number of workers that this pool should manage.
119 * @param filePath - Path to the worker file.
120 * @param opts - Options for the pool.
123 protected readonly numberOfWorkers
: number,
124 protected readonly filePath
: string,
125 protected readonly opts
: PoolOptions
<Worker
>
127 if (!this.isMain()) {
129 'Cannot start a pool from a worker with the same type as the pool'
132 this.checkNumberOfWorkers(this.numberOfWorkers
)
133 this.checkFilePath(this.filePath
)
134 this.checkPoolOptions(this.opts
)
136 this.chooseWorkerNode
= this.chooseWorkerNode
.bind(this)
137 this.executeTask
= this.executeTask
.bind(this)
138 this.enqueueTask
= this.enqueueTask
.bind(this)
140 if (this.opts
.enableEvents
=== true) {
141 this.emitter
= new PoolEmitter()
143 this.workerChoiceStrategyContext
= new WorkerChoiceStrategyContext
<
149 this.opts
.workerChoiceStrategy
,
150 this.opts
.workerChoiceStrategyOptions
155 this.taskFunctions
= new Map
<string, TaskFunction
<Data
, Response
>>()
158 this.starting
= false
159 if (this.opts
.startWorkers
=== true) {
163 this.startTimestamp
= performance
.now()
166 private checkFilePath (filePath
: string): void {
169 typeof filePath
!== 'string' ||
170 (typeof filePath
=== 'string' && filePath
.trim().length
=== 0)
172 throw new Error('Please specify a file with a worker implementation')
174 if (!existsSync(filePath
)) {
175 throw new Error(`Cannot find the worker file '${filePath}'`)
179 private checkNumberOfWorkers (numberOfWorkers
: number): void {
180 if (numberOfWorkers
== null) {
182 'Cannot instantiate a pool without specifying the number of workers'
184 } else if (!Number.isSafeInteger(numberOfWorkers
)) {
186 'Cannot instantiate a pool with a non safe integer number of workers'
188 } else if (numberOfWorkers
< 0) {
189 throw new RangeError(
190 'Cannot instantiate a pool with a negative number of workers'
192 } else if (this.type === PoolTypes
.fixed
&& numberOfWorkers
=== 0) {
193 throw new RangeError('Cannot instantiate a fixed pool with zero worker')
197 protected checkDynamicPoolSize (min
: number, max
: number): void {
198 if (this.type === PoolTypes
.dynamic
) {
201 'Cannot instantiate a dynamic pool without specifying the maximum pool size'
203 } else if (!Number.isSafeInteger(max
)) {
205 'Cannot instantiate a dynamic pool with a non safe integer maximum pool size'
207 } else if (min
> max
) {
208 throw new RangeError(
209 'Cannot instantiate a dynamic pool with a maximum pool size inferior to the minimum pool size'
211 } else if (max
=== 0) {
212 throw new RangeError(
213 'Cannot instantiate a dynamic pool with a maximum pool size equal to zero'
215 } else if (min
=== max
) {
216 throw new RangeError(
217 'Cannot instantiate a dynamic pool with a minimum pool size equal to the maximum pool size. Use a fixed pool instead'
223 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
224 if (isPlainObject(opts
)) {
225 this.opts
.startWorkers
= opts
.startWorkers
?? true
226 this.checkValidWorkerChoiceStrategy(
227 opts
.workerChoiceStrategy
as WorkerChoiceStrategy
229 this.opts
.workerChoiceStrategy
=
230 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
231 this.checkValidWorkerChoiceStrategyOptions(
232 opts
.workerChoiceStrategyOptions
as WorkerChoiceStrategyOptions
234 this.opts
.workerChoiceStrategyOptions
= {
235 ...DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
,
236 ...opts
.workerChoiceStrategyOptions
238 this.opts
.restartWorkerOnError
= opts
.restartWorkerOnError
?? true
239 this.opts
.enableEvents
= opts
.enableEvents
?? true
240 this.opts
.enableTasksQueue
= opts
.enableTasksQueue
?? false
241 if (this.opts
.enableTasksQueue
) {
242 this.checkValidTasksQueueOptions(
243 opts
.tasksQueueOptions
as TasksQueueOptions
245 this.opts
.tasksQueueOptions
= this.buildTasksQueueOptions(
246 opts
.tasksQueueOptions
as TasksQueueOptions
250 throw new TypeError('Invalid pool options: must be a plain object')
254 private checkValidWorkerChoiceStrategy (
255 workerChoiceStrategy
: WorkerChoiceStrategy
258 workerChoiceStrategy
!= null &&
259 !Object.values(WorkerChoiceStrategies
).includes(workerChoiceStrategy
)
262 `Invalid worker choice strategy '${workerChoiceStrategy}'`
267 private checkValidWorkerChoiceStrategyOptions (
268 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
271 workerChoiceStrategyOptions
!= null &&
272 !isPlainObject(workerChoiceStrategyOptions
)
275 'Invalid worker choice strategy options: must be a plain object'
279 workerChoiceStrategyOptions
?.retries
!= null &&
280 !Number.isSafeInteger(workerChoiceStrategyOptions
.retries
)
283 'Invalid worker choice strategy options: retries must be an integer'
287 workerChoiceStrategyOptions
?.retries
!= null &&
288 workerChoiceStrategyOptions
.retries
< 0
290 throw new RangeError(
291 `Invalid worker choice strategy options: retries '${workerChoiceStrategyOptions.retries}' must be greater or equal than zero`
295 workerChoiceStrategyOptions
?.weights
!= null &&
296 Object.keys(workerChoiceStrategyOptions
.weights
).length
!== this.maxSize
299 'Invalid worker choice strategy options: must have a weight for each worker node'
303 workerChoiceStrategyOptions
?.measurement
!= null &&
304 !Object.values(Measurements
).includes(
305 workerChoiceStrategyOptions
.measurement
309 `Invalid worker choice strategy options: invalid measurement '${workerChoiceStrategyOptions.measurement}'`
314 private checkValidTasksQueueOptions (
315 tasksQueueOptions
: TasksQueueOptions
317 if (tasksQueueOptions
!= null && !isPlainObject(tasksQueueOptions
)) {
318 throw new TypeError('Invalid tasks queue options: must be a plain object')
321 tasksQueueOptions
?.concurrency
!= null &&
322 !Number.isSafeInteger(tasksQueueOptions
.concurrency
)
325 'Invalid worker node tasks concurrency: must be an integer'
329 tasksQueueOptions
?.concurrency
!= null &&
330 tasksQueueOptions
.concurrency
<= 0
332 throw new RangeError(
333 `Invalid worker node tasks concurrency: ${tasksQueueOptions.concurrency} is a negative integer or zero`
337 tasksQueueOptions
?.size
!= null &&
338 !Number.isSafeInteger(tasksQueueOptions
.size
)
341 'Invalid worker node tasks queue size: must be an integer'
344 if (tasksQueueOptions
?.size
!= null && tasksQueueOptions
.size
<= 0) {
345 throw new RangeError(
346 `Invalid worker node tasks queue size: ${tasksQueueOptions.size} is a negative integer or zero`
352 public get
info (): PoolInfo
{
357 started
: this.started
,
359 strategy
: this.opts
.workerChoiceStrategy
as WorkerChoiceStrategy
,
360 minSize
: this.minSize
,
361 maxSize
: this.maxSize
,
362 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
363 .runTime
.aggregate
&&
364 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
365 .waitTime
.aggregate
&& { utilization
: round(this.utilization
) }),
366 workerNodes
: this.workerNodes
.length
,
367 idleWorkerNodes
: this.workerNodes
.reduce(
368 (accumulator
, workerNode
) =>
369 workerNode
.usage
.tasks
.executing
=== 0
374 busyWorkerNodes
: this.workerNodes
.reduce(
375 (accumulator
, workerNode
) =>
376 workerNode
.usage
.tasks
.executing
> 0 ? accumulator
+ 1 : accumulator
,
379 executedTasks
: this.workerNodes
.reduce(
380 (accumulator
, workerNode
) =>
381 accumulator
+ workerNode
.usage
.tasks
.executed
,
384 executingTasks
: this.workerNodes
.reduce(
385 (accumulator
, workerNode
) =>
386 accumulator
+ workerNode
.usage
.tasks
.executing
,
389 ...(this.opts
.enableTasksQueue
=== true && {
390 queuedTasks
: this.workerNodes
.reduce(
391 (accumulator
, workerNode
) =>
392 accumulator
+ workerNode
.usage
.tasks
.queued
,
396 ...(this.opts
.enableTasksQueue
=== true && {
397 maxQueuedTasks
: this.workerNodes
.reduce(
398 (accumulator
, workerNode
) =>
399 accumulator
+ (workerNode
.usage
.tasks
?.maxQueued
?? 0),
403 ...(this.opts
.enableTasksQueue
=== true && {
404 backPressure
: this.hasBackPressure()
406 ...(this.opts
.enableTasksQueue
=== true && {
407 stolenTasks
: this.workerNodes
.reduce(
408 (accumulator
, workerNode
) =>
409 accumulator
+ workerNode
.usage
.tasks
.stolen
,
413 failedTasks
: this.workerNodes
.reduce(
414 (accumulator
, workerNode
) =>
415 accumulator
+ workerNode
.usage
.tasks
.failed
,
418 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
419 .runTime
.aggregate
&& {
423 ...this.workerNodes
.map(
424 workerNode
=> workerNode
.usage
.runTime
?.minimum
?? Infinity
430 ...this.workerNodes
.map(
431 workerNode
=> workerNode
.usage
.runTime
?.maximum
?? -Infinity
435 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
436 .runTime
.average
&& {
439 this.workerNodes
.reduce
<number[]>(
440 (accumulator
, workerNode
) =>
441 accumulator
.concat(workerNode
.usage
.runTime
.history
),
447 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
451 this.workerNodes
.reduce
<number[]>(
452 (accumulator
, workerNode
) =>
453 accumulator
.concat(workerNode
.usage
.runTime
.history
),
461 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
462 .waitTime
.aggregate
&& {
466 ...this.workerNodes
.map(
467 workerNode
=> workerNode
.usage
.waitTime
?.minimum
?? Infinity
473 ...this.workerNodes
.map(
474 workerNode
=> workerNode
.usage
.waitTime
?.maximum
?? -Infinity
478 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
479 .waitTime
.average
&& {
482 this.workerNodes
.reduce
<number[]>(
483 (accumulator
, workerNode
) =>
484 accumulator
.concat(workerNode
.usage
.waitTime
.history
),
490 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
491 .waitTime
.median
&& {
494 this.workerNodes
.reduce
<number[]>(
495 (accumulator
, workerNode
) =>
496 accumulator
.concat(workerNode
.usage
.waitTime
.history
),
508 * The pool readiness boolean status.
510 private get
ready (): boolean {
512 this.workerNodes
.reduce(
513 (accumulator
, workerNode
) =>
514 !workerNode
.info
.dynamic
&& workerNode
.info
.ready
523 * The approximate pool utilization.
525 * @returns The pool utilization.
527 private get
utilization (): number {
528 const poolTimeCapacity
=
529 (performance
.now() - this.startTimestamp
) * this.maxSize
530 const totalTasksRunTime
= this.workerNodes
.reduce(
531 (accumulator
, workerNode
) =>
532 accumulator
+ (workerNode
.usage
.runTime
?.aggregate
?? 0),
535 const totalTasksWaitTime
= this.workerNodes
.reduce(
536 (accumulator
, workerNode
) =>
537 accumulator
+ (workerNode
.usage
.waitTime
?.aggregate
?? 0),
540 return (totalTasksRunTime
+ totalTasksWaitTime
) / poolTimeCapacity
546 * If it is `'dynamic'`, it provides the `max` property.
548 protected abstract get
type (): PoolType
553 protected abstract get
worker (): WorkerType
556 * The pool minimum size.
558 protected get
minSize (): number {
559 return this.numberOfWorkers
563 * The pool maximum size.
565 protected get
maxSize (): number {
566 return this.max
?? this.numberOfWorkers
570 * Checks if the worker id sent in the received message from a worker is valid.
572 * @param message - The received message.
573 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the worker id is invalid.
575 private checkMessageWorkerId (message
: MessageValue
<Response
>): void {
576 if (message
.workerId
== null) {
577 throw new Error('Worker message received without worker id')
579 message
.workerId
!= null &&
580 this.getWorkerNodeKeyByWorkerId(message
.workerId
) === -1
583 `Worker message received from unknown worker '${message.workerId}'`
589 * Gets the given worker its worker node key.
591 * @param worker - The worker.
592 * @returns The worker node key if found in the pool worker nodes, `-1` otherwise.
594 private getWorkerNodeKeyByWorker (worker
: Worker
): number {
595 return this.workerNodes
.findIndex(
596 workerNode
=> workerNode
.worker
=== worker
601 * Gets the worker node key given its worker id.
603 * @param workerId - The worker id.
604 * @returns The worker node key if the worker id is found in the pool worker nodes, `-1` otherwise.
606 private getWorkerNodeKeyByWorkerId (workerId
: number | undefined): number {
607 return this.workerNodes
.findIndex(
608 workerNode
=> workerNode
.info
.id
=== workerId
613 public setWorkerChoiceStrategy (
614 workerChoiceStrategy
: WorkerChoiceStrategy
,
615 workerChoiceStrategyOptions
?: WorkerChoiceStrategyOptions
617 this.checkValidWorkerChoiceStrategy(workerChoiceStrategy
)
618 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
619 this.workerChoiceStrategyContext
.setWorkerChoiceStrategy(
620 this.opts
.workerChoiceStrategy
622 if (workerChoiceStrategyOptions
!= null) {
623 this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
625 for (const [workerNodeKey
, workerNode
] of this.workerNodes
.entries()) {
626 workerNode
.resetUsage()
627 this.sendStatisticsMessageToWorker(workerNodeKey
)
632 public setWorkerChoiceStrategyOptions (
633 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
635 this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
636 this.opts
.workerChoiceStrategyOptions
= {
637 ...DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
,
638 ...workerChoiceStrategyOptions
640 this.workerChoiceStrategyContext
.setOptions(
641 this.opts
.workerChoiceStrategyOptions
646 public enableTasksQueue (
648 tasksQueueOptions
?: TasksQueueOptions
650 if (this.opts
.enableTasksQueue
=== true && !enable
) {
651 this.unsetTaskStealing()
652 this.unsetTasksStealingOnBackPressure()
653 this.flushTasksQueues()
655 this.opts
.enableTasksQueue
= enable
656 this.setTasksQueueOptions(tasksQueueOptions
as TasksQueueOptions
)
660 public setTasksQueueOptions (tasksQueueOptions
: TasksQueueOptions
): void {
661 if (this.opts
.enableTasksQueue
=== true) {
662 this.checkValidTasksQueueOptions(tasksQueueOptions
)
663 this.opts
.tasksQueueOptions
=
664 this.buildTasksQueueOptions(tasksQueueOptions
)
665 this.setTasksQueueSize(this.opts
.tasksQueueOptions
.size
as number)
666 if (this.opts
.tasksQueueOptions
.taskStealing
=== true) {
667 this.setTaskStealing()
669 this.unsetTaskStealing()
671 if (this.opts
.tasksQueueOptions
.tasksStealingOnBackPressure
=== true) {
672 this.setTasksStealingOnBackPressure()
674 this.unsetTasksStealingOnBackPressure()
676 } else if (this.opts
.tasksQueueOptions
!= null) {
677 delete this.opts
.tasksQueueOptions
681 private buildTasksQueueOptions (
682 tasksQueueOptions
: TasksQueueOptions
683 ): TasksQueueOptions
{
686 size
: Math.pow(this.maxSize
, 2),
689 tasksStealingOnBackPressure
: true
695 private setTasksQueueSize (size
: number): void {
696 for (const workerNode
of this.workerNodes
) {
697 workerNode
.tasksQueueBackPressureSize
= size
701 private setTaskStealing (): void {
702 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
703 this.workerNodes
[workerNodeKey
].onEmptyQueue
=
704 this.taskStealingOnEmptyQueue
.bind(this)
708 private unsetTaskStealing (): void {
709 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
710 delete this.workerNodes
[workerNodeKey
].onEmptyQueue
714 private setTasksStealingOnBackPressure (): void {
715 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
716 this.workerNodes
[workerNodeKey
].onBackPressure
=
717 this.tasksStealingOnBackPressure
.bind(this)
721 private unsetTasksStealingOnBackPressure (): void {
722 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
723 delete this.workerNodes
[workerNodeKey
].onBackPressure
728 * Whether the pool is full or not.
730 * The pool filling boolean status.
732 protected get
full (): boolean {
733 return this.workerNodes
.length
>= this.maxSize
737 * Whether the pool is busy or not.
739 * The pool busyness boolean status.
741 protected abstract get
busy (): boolean
744 * Whether worker nodes are executing concurrently their tasks quota or not.
746 * @returns Worker nodes busyness boolean status.
748 protected internalBusy (): boolean {
749 if (this.opts
.enableTasksQueue
=== true) {
751 this.workerNodes
.findIndex(
753 workerNode
.info
.ready
&&
754 workerNode
.usage
.tasks
.executing
<
755 (this.opts
.tasksQueueOptions
?.concurrency
as number)
760 this.workerNodes
.findIndex(
762 workerNode
.info
.ready
&& workerNode
.usage
.tasks
.executing
=== 0
767 private async sendTaskFunctionOperationToWorker (
768 workerNodeKey
: number,
769 message
: MessageValue
<Data
>
770 ): Promise
<boolean> {
771 return await new Promise
<boolean>((resolve
, reject
) => {
772 const workerId
= this.getWorkerInfo(workerNodeKey
).id
as number
773 this.registerWorkerMessageListener(workerNodeKey
, message
=> {
775 message
.workerId
=== workerId
&&
776 message
.taskFunctionOperationStatus
=== true
780 message
.workerId
=== workerId
&&
781 message
.taskFunctionOperationStatus
=== false
785 `Task function operation ${
786 message.taskFunctionOperation as string
787 } failed on worker ${message.workerId}`
792 this.sendToWorker(workerNodeKey
, message
)
796 private async sendTaskFunctionOperationToWorkers (
797 message
: MessageValue
<Data
>
798 ): Promise
<boolean> {
799 return await new Promise
<boolean>((resolve
, reject
) => {
800 const responsesReceived
= new Array<MessageValue
<Data
| Response
>>()
801 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
802 this.registerWorkerMessageListener(workerNodeKey
, message
=> {
803 if (message
.taskFunctionOperationStatus
!= null) {
804 responsesReceived
.push(message
)
806 responsesReceived
.length
=== this.workerNodes
.length
&&
807 responsesReceived
.every(
808 message
=> message
.taskFunctionOperationStatus
=== true
813 responsesReceived
.length
=== this.workerNodes
.length
&&
814 responsesReceived
.some(
815 message
=> message
.taskFunctionOperationStatus
=== false
820 `Task function operation ${
821 message.taskFunctionOperation as string
822 } failed on worker ${message.workerId as number}`
828 this.sendToWorker(workerNodeKey
, message
)
834 public hasTaskFunction (name
: string): boolean {
835 for (const workerNode
of this.workerNodes
) {
837 Array.isArray(workerNode
.info
.taskFunctionNames
) &&
838 workerNode
.info
.taskFunctionNames
.includes(name
)
847 public async addTaskFunction (
849 fn
: TaskFunction
<Data
, Response
>
850 ): Promise
<boolean> {
851 if (typeof name
!== 'string') {
852 throw new TypeError('name argument must be a string')
854 if (typeof name
=== 'string' && name
.trim().length
=== 0) {
855 throw new TypeError('name argument must not be an empty string')
857 if (typeof fn
!== 'function') {
858 throw new TypeError('fn argument must be a function')
860 const opResult
= await this.sendTaskFunctionOperationToWorkers({
861 taskFunctionOperation
: 'add',
862 taskFunctionName
: name
,
863 taskFunction
: fn
.toString()
865 this.taskFunctions
.set(name
, fn
)
870 public async removeTaskFunction (name
: string): Promise
<boolean> {
871 if (!this.taskFunctions
.has(name
)) {
873 'Cannot remove a task function not handled on the pool side'
876 const opResult
= await this.sendTaskFunctionOperationToWorkers({
877 taskFunctionOperation
: 'remove',
878 taskFunctionName
: name
880 this.deleteTaskFunctionWorkerUsages(name
)
881 this.taskFunctions
.delete(name
)
886 public listTaskFunctionNames (): string[] {
887 for (const workerNode
of this.workerNodes
) {
889 Array.isArray(workerNode
.info
.taskFunctionNames
) &&
890 workerNode
.info
.taskFunctionNames
.length
> 0
892 return workerNode
.info
.taskFunctionNames
899 public async setDefaultTaskFunction (name
: string): Promise
<boolean> {
900 return await this.sendTaskFunctionOperationToWorkers({
901 taskFunctionOperation
: 'default',
902 taskFunctionName
: name
906 private deleteTaskFunctionWorkerUsages (name
: string): void {
907 for (const workerNode
of this.workerNodes
) {
908 workerNode
.deleteTaskFunctionWorkerUsage(name
)
912 private shallExecuteTask (workerNodeKey
: number): boolean {
914 this.tasksQueueSize(workerNodeKey
) === 0 &&
915 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
<
916 (this.opts
.tasksQueueOptions
?.concurrency
as number)
921 public async execute (
924 transferList
?: TransferListItem
[]
925 ): Promise
<Response
> {
926 return await new Promise
<Response
>((resolve
, reject
) => {
928 reject(new Error('Cannot execute a task on not started pool'))
931 if (name
!= null && typeof name
!== 'string') {
932 reject(new TypeError('name argument must be a string'))
937 typeof name
=== 'string' &&
938 name
.trim().length
=== 0
940 reject(new TypeError('name argument must not be an empty string'))
943 if (transferList
!= null && !Array.isArray(transferList
)) {
944 reject(new TypeError('transferList argument must be an array'))
947 const timestamp
= performance
.now()
948 const workerNodeKey
= this.chooseWorkerNode()
949 const task
: Task
<Data
> = {
950 name
: name
?? DEFAULT_TASK_NAME
,
951 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
952 data
: data
?? ({} as Data
),
957 this.promiseResponseMap
.set(task
.taskId
as string, {
963 this.opts
.enableTasksQueue
=== false ||
964 (this.opts
.enableTasksQueue
=== true &&
965 this.shallExecuteTask(workerNodeKey
))
967 this.executeTask(workerNodeKey
, task
)
969 this.enqueueTask(workerNodeKey
, task
)
975 public start (): void {
978 this.workerNodes
.reduce(
979 (accumulator
, workerNode
) =>
980 !workerNode
.info
.dynamic
? accumulator
+ 1 : accumulator
,
982 ) < this.numberOfWorkers
984 this.createAndSetupWorkerNode()
986 this.starting
= false
991 public async destroy (): Promise
<void> {
993 this.workerNodes
.map(async (_
, workerNodeKey
) => {
994 await this.destroyWorkerNode(workerNodeKey
)
997 this.emitter
?.emit(PoolEvents
.destroy
, this.info
)
1001 protected async sendKillMessageToWorker (
1002 workerNodeKey
: number
1004 await new Promise
<void>((resolve
, reject
) => {
1005 this.registerWorkerMessageListener(workerNodeKey
, message
=> {
1006 if (message
.kill
=== 'success') {
1008 } else if (message
.kill
=== 'failure') {
1012 message.workerId as number
1013 } kill message handling failed`
1018 this.sendToWorker(workerNodeKey
, { kill
: true })
1023 * Terminates the worker node given its worker node key.
1025 * @param workerNodeKey - The worker node key.
1027 protected abstract destroyWorkerNode (workerNodeKey
: number): Promise
<void>
1030 * Setup hook to execute code before worker nodes are created in the abstract constructor.
1031 * Can be overridden.
1035 protected setupHook (): void {
1036 /* Intentionally empty */
1040 * Should return whether the worker is the main worker or not.
1042 protected abstract isMain (): boolean
1045 * Hook executed before the worker task execution.
1046 * Can be overridden.
1048 * @param workerNodeKey - The worker node key.
1049 * @param task - The task to execute.
1051 protected beforeTaskExecutionHook (
1052 workerNodeKey
: number,
1055 if (this.workerNodes
[workerNodeKey
]?.usage
!= null) {
1056 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
1057 ++workerUsage
.tasks
.executing
1058 this.updateWaitTimeWorkerUsage(workerUsage
, task
)
1061 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1062 this.workerNodes
[workerNodeKey
].getTaskFunctionWorkerUsage(
1066 const taskFunctionWorkerUsage
= this.workerNodes
[
1068 ].getTaskFunctionWorkerUsage(task
.name
as string) as WorkerUsage
1069 ++taskFunctionWorkerUsage
.tasks
.executing
1070 this.updateWaitTimeWorkerUsage(taskFunctionWorkerUsage
, task
)
1075 * Hook executed after the worker task execution.
1076 * Can be overridden.
1078 * @param workerNodeKey - The worker node key.
1079 * @param message - The received message.
1081 protected afterTaskExecutionHook (
1082 workerNodeKey
: number,
1083 message
: MessageValue
<Response
>
1085 if (this.workerNodes
[workerNodeKey
]?.usage
!= null) {
1086 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
1087 this.updateTaskStatisticsWorkerUsage(workerUsage
, message
)
1088 this.updateRunTimeWorkerUsage(workerUsage
, message
)
1089 this.updateEluWorkerUsage(workerUsage
, message
)
1092 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1093 this.workerNodes
[workerNodeKey
].getTaskFunctionWorkerUsage(
1094 message
.taskPerformance
?.name
as string
1097 const taskFunctionWorkerUsage
= this.workerNodes
[
1099 ].getTaskFunctionWorkerUsage(
1100 message
.taskPerformance
?.name
as string
1102 this.updateTaskStatisticsWorkerUsage(taskFunctionWorkerUsage
, message
)
1103 this.updateRunTimeWorkerUsage(taskFunctionWorkerUsage
, message
)
1104 this.updateEluWorkerUsage(taskFunctionWorkerUsage
, message
)
1109 * Whether the worker node shall update its task function worker usage or not.
1111 * @param workerNodeKey - The worker node key.
1112 * @returns `true` if the worker node shall update its task function worker usage, `false` otherwise.
1114 private shallUpdateTaskFunctionWorkerUsage (workerNodeKey
: number): boolean {
1115 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1117 workerInfo
!= null &&
1118 Array.isArray(workerInfo
.taskFunctionNames
) &&
1119 workerInfo
.taskFunctionNames
.length
> 2
1123 private updateTaskStatisticsWorkerUsage (
1124 workerUsage
: WorkerUsage
,
1125 message
: MessageValue
<Response
>
1127 const workerTaskStatistics
= workerUsage
.tasks
1129 workerTaskStatistics
.executing
!= null &&
1130 workerTaskStatistics
.executing
> 0
1132 --workerTaskStatistics
.executing
1134 if (message
.workerError
== null) {
1135 ++workerTaskStatistics
.executed
1137 ++workerTaskStatistics
.failed
1141 private updateRunTimeWorkerUsage (
1142 workerUsage
: WorkerUsage
,
1143 message
: MessageValue
<Response
>
1145 if (message
.workerError
!= null) {
1148 updateMeasurementStatistics(
1149 workerUsage
.runTime
,
1150 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
,
1151 message
.taskPerformance
?.runTime
?? 0
1155 private updateWaitTimeWorkerUsage (
1156 workerUsage
: WorkerUsage
,
1159 const timestamp
= performance
.now()
1160 const taskWaitTime
= timestamp
- (task
.timestamp
?? timestamp
)
1161 updateMeasurementStatistics(
1162 workerUsage
.waitTime
,
1163 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().waitTime
,
1168 private updateEluWorkerUsage (
1169 workerUsage
: WorkerUsage
,
1170 message
: MessageValue
<Response
>
1172 if (message
.workerError
!= null) {
1175 const eluTaskStatisticsRequirements
: MeasurementStatisticsRequirements
=
1176 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
1177 updateMeasurementStatistics(
1178 workerUsage
.elu
.active
,
1179 eluTaskStatisticsRequirements
,
1180 message
.taskPerformance
?.elu
?.active
?? 0
1182 updateMeasurementStatistics(
1183 workerUsage
.elu
.idle
,
1184 eluTaskStatisticsRequirements
,
1185 message
.taskPerformance
?.elu
?.idle
?? 0
1187 if (eluTaskStatisticsRequirements
.aggregate
) {
1188 if (message
.taskPerformance
?.elu
!= null) {
1189 if (workerUsage
.elu
.utilization
!= null) {
1190 workerUsage
.elu
.utilization
=
1191 (workerUsage
.elu
.utilization
+
1192 message
.taskPerformance
.elu
.utilization
) /
1195 workerUsage
.elu
.utilization
= message
.taskPerformance
.elu
.utilization
1202 * Chooses a worker node for the next task.
1204 * The default worker choice strategy uses a round robin algorithm to distribute the tasks.
1206 * @returns The chosen worker node key
1208 private chooseWorkerNode (): number {
1209 if (this.shallCreateDynamicWorker()) {
1210 const workerNodeKey
= this.createAndSetupDynamicWorkerNode()
1212 this.workerChoiceStrategyContext
.getStrategyPolicy().dynamicWorkerUsage
1214 return workerNodeKey
1217 return this.workerChoiceStrategyContext
.execute()
1221 * Conditions for dynamic worker creation.
1223 * @returns Whether to create a dynamic worker or not.
1225 private shallCreateDynamicWorker (): boolean {
1226 return this.type === PoolTypes
.dynamic
&& !this.full
&& this.internalBusy()
1230 * Sends a message to worker given its worker node key.
1232 * @param workerNodeKey - The worker node key.
1233 * @param message - The message.
1234 * @param transferList - The optional array of transferable objects.
1236 protected abstract sendToWorker (
1237 workerNodeKey
: number,
1238 message
: MessageValue
<Data
>,
1239 transferList
?: TransferListItem
[]
1243 * Creates a new worker.
1245 * @returns Newly created worker.
1247 protected abstract createWorker (): Worker
1250 * Creates a new, completely set up worker node.
1252 * @returns New, completely set up worker node key.
1254 protected createAndSetupWorkerNode (): number {
1255 const worker
= this.createWorker()
1257 worker
.on('online', this.opts
.onlineHandler
?? EMPTY_FUNCTION
)
1258 worker
.on('message', this.opts
.messageHandler
?? EMPTY_FUNCTION
)
1259 worker
.on('error', this.opts
.errorHandler
?? EMPTY_FUNCTION
)
1260 worker
.on('error', error
=> {
1261 const workerNodeKey
= this.getWorkerNodeKeyByWorker(worker
)
1262 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1263 workerInfo
.ready
= false
1264 this.workerNodes
[workerNodeKey
].closeChannel()
1265 this.emitter
?.emit(PoolEvents
.error
, error
)
1269 this.opts
.restartWorkerOnError
=== true
1271 if (workerInfo
.dynamic
) {
1272 this.createAndSetupDynamicWorkerNode()
1274 this.createAndSetupWorkerNode()
1277 if (this.started
&& this.opts
.enableTasksQueue
=== true) {
1278 this.redistributeQueuedTasks(workerNodeKey
)
1281 worker
.on('exit', this.opts
.exitHandler
?? EMPTY_FUNCTION
)
1282 worker
.once('exit', () => {
1283 this.removeWorkerNode(worker
)
1286 const workerNodeKey
= this.addWorkerNode(worker
)
1288 this.afterWorkerNodeSetup(workerNodeKey
)
1290 return workerNodeKey
1294 * Creates a new, completely set up dynamic worker node.
1296 * @returns New, completely set up dynamic worker node key.
1298 protected createAndSetupDynamicWorkerNode (): number {
1299 const workerNodeKey
= this.createAndSetupWorkerNode()
1300 this.registerWorkerMessageListener(workerNodeKey
, message
=> {
1301 const localWorkerNodeKey
= this.getWorkerNodeKeyByWorkerId(
1304 const workerUsage
= this.workerNodes
[localWorkerNodeKey
].usage
1305 // Kill message received from worker
1307 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
1308 (isKillBehavior(KillBehaviors
.SOFT
, message
.kill
) &&
1309 ((this.opts
.enableTasksQueue
=== false &&
1310 workerUsage
.tasks
.executing
=== 0) ||
1311 (this.opts
.enableTasksQueue
=== true &&
1312 workerUsage
.tasks
.executing
=== 0 &&
1313 this.tasksQueueSize(localWorkerNodeKey
) === 0)))
1315 this.destroyWorkerNode(localWorkerNodeKey
).catch(error
=> {
1316 this.emitter
?.emit(PoolEvents
.error
, error
)
1320 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1321 this.sendToWorker(workerNodeKey
, {
1324 if (this.taskFunctions
.size
> 0) {
1325 for (const [taskFunctionName
, taskFunction
] of this.taskFunctions
) {
1326 this.sendTaskFunctionOperationToWorker(workerNodeKey
, {
1327 taskFunctionOperation
: 'add',
1329 taskFunction
: taskFunction
.toString()
1331 this.emitter
?.emit(PoolEvents
.error
, error
)
1335 workerInfo
.dynamic
= true
1337 this.workerChoiceStrategyContext
.getStrategyPolicy().dynamicWorkerReady
||
1338 this.workerChoiceStrategyContext
.getStrategyPolicy().dynamicWorkerUsage
1340 workerInfo
.ready
= true
1342 this.checkAndEmitDynamicWorkerCreationEvents()
1343 return workerNodeKey
1347 * Registers a listener callback on the worker given its worker node key.
1349 * @param workerNodeKey - The worker node key.
1350 * @param listener - The message listener callback.
1352 protected abstract registerWorkerMessageListener
<
1353 Message
extends Data
| Response
1355 workerNodeKey
: number,
1356 listener
: (message
: MessageValue
<Message
>) => void
1360 * Method hooked up after a worker node has been newly created.
1361 * Can be overridden.
1363 * @param workerNodeKey - The newly created worker node key.
1365 protected afterWorkerNodeSetup (workerNodeKey
: number): void {
1366 // Listen to worker messages.
1367 this.registerWorkerMessageListener(workerNodeKey
, this.workerListener())
1368 // Send the startup message to worker.
1369 this.sendStartupMessageToWorker(workerNodeKey
)
1370 // Send the statistics message to worker.
1371 this.sendStatisticsMessageToWorker(workerNodeKey
)
1372 if (this.opts
.enableTasksQueue
=== true) {
1373 if (this.opts
.tasksQueueOptions
?.taskStealing
=== true) {
1374 this.workerNodes
[workerNodeKey
].onEmptyQueue
=
1375 this.taskStealingOnEmptyQueue
.bind(this)
1377 if (this.opts
.tasksQueueOptions
?.tasksStealingOnBackPressure
=== true) {
1378 this.workerNodes
[workerNodeKey
].onBackPressure
=
1379 this.tasksStealingOnBackPressure
.bind(this)
1385 * Sends the startup message to worker given its worker node key.
1387 * @param workerNodeKey - The worker node key.
1389 protected abstract sendStartupMessageToWorker (workerNodeKey
: number): void
1392 * Sends the statistics message to worker given its worker node key.
1394 * @param workerNodeKey - The worker node key.
1396 private sendStatisticsMessageToWorker (workerNodeKey
: number): void {
1397 this.sendToWorker(workerNodeKey
, {
1400 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1402 elu
: this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1408 private redistributeQueuedTasks (workerNodeKey
: number): void {
1409 while (this.tasksQueueSize(workerNodeKey
) > 0) {
1410 const destinationWorkerNodeKey
= this.workerNodes
.reduce(
1411 (minWorkerNodeKey
, workerNode
, workerNodeKey
, workerNodes
) => {
1412 return workerNode
.info
.ready
&&
1413 workerNode
.usage
.tasks
.queued
<
1414 workerNodes
[minWorkerNodeKey
].usage
.tasks
.queued
1420 const task
= this.dequeueTask(workerNodeKey
) as Task
<Data
>
1421 if (this.shallExecuteTask(destinationWorkerNodeKey
)) {
1422 this.executeTask(destinationWorkerNodeKey
, task
)
1424 this.enqueueTask(destinationWorkerNodeKey
, task
)
1429 private updateTaskStolenStatisticsWorkerUsage (
1430 workerNodeKey
: number,
1433 const workerNode
= this.workerNodes
[workerNodeKey
]
1434 if (workerNode
?.usage
!= null) {
1435 ++workerNode
.usage
.tasks
.stolen
1438 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1439 workerNode
.getTaskFunctionWorkerUsage(taskName
) != null
1441 const taskFunctionWorkerUsage
= workerNode
.getTaskFunctionWorkerUsage(
1444 ++taskFunctionWorkerUsage
.tasks
.stolen
1448 private taskStealingOnEmptyQueue (workerId
: number): void {
1449 const destinationWorkerNodeKey
= this.getWorkerNodeKeyByWorkerId(workerId
)
1450 const workerNodes
= this.workerNodes
1453 (workerNodeA
, workerNodeB
) =>
1454 workerNodeB
.usage
.tasks
.queued
- workerNodeA
.usage
.tasks
.queued
1456 const sourceWorkerNode
= workerNodes
.find(
1458 workerNode
.info
.ready
&&
1459 workerNode
.info
.id
!== workerId
&&
1460 workerNode
.usage
.tasks
.queued
> 0
1462 if (sourceWorkerNode
!= null) {
1463 const task
= sourceWorkerNode
.popTask() as Task
<Data
>
1464 if (this.shallExecuteTask(destinationWorkerNodeKey
)) {
1465 this.executeTask(destinationWorkerNodeKey
, task
)
1467 this.enqueueTask(destinationWorkerNodeKey
, task
)
1469 this.updateTaskStolenStatisticsWorkerUsage(
1470 destinationWorkerNodeKey
,
1476 private tasksStealingOnBackPressure (workerId
: number): void {
1477 const sizeOffset
= 1
1478 if ((this.opts
.tasksQueueOptions
?.size
as number) <= sizeOffset
) {
1481 const sourceWorkerNode
=
1482 this.workerNodes
[this.getWorkerNodeKeyByWorkerId(workerId
)]
1483 const workerNodes
= this.workerNodes
1486 (workerNodeA
, workerNodeB
) =>
1487 workerNodeA
.usage
.tasks
.queued
- workerNodeB
.usage
.tasks
.queued
1489 for (const [workerNodeKey
, workerNode
] of workerNodes
.entries()) {
1491 sourceWorkerNode
.usage
.tasks
.queued
> 0 &&
1492 workerNode
.info
.ready
&&
1493 workerNode
.info
.id
!== workerId
&&
1494 workerNode
.usage
.tasks
.queued
<
1495 (this.opts
.tasksQueueOptions
?.size
as number) - sizeOffset
1497 const task
= sourceWorkerNode
.popTask() as Task
<Data
>
1498 if (this.shallExecuteTask(workerNodeKey
)) {
1499 this.executeTask(workerNodeKey
, task
)
1501 this.enqueueTask(workerNodeKey
, task
)
1503 this.updateTaskStolenStatisticsWorkerUsage(
1512 * This method is the listener registered for each worker message.
1514 * @returns The listener function to execute when a message is received from a worker.
1516 protected workerListener (): (message
: MessageValue
<Response
>) => void {
1518 this.checkMessageWorkerId(message
)
1519 if (message
.ready
!= null && message
.taskFunctionNames
!= null) {
1520 // Worker ready response received from worker
1521 this.handleWorkerReadyResponse(message
)
1522 } else if (message
.taskId
!= null) {
1523 // Task execution response received from worker
1524 this.handleTaskExecutionResponse(message
)
1525 } else if (message
.taskFunctionNames
!= null) {
1526 // Task function names message received from worker
1528 this.getWorkerNodeKeyByWorkerId(message
.workerId
)
1529 ).taskFunctionNames
= message
.taskFunctionNames
1534 private handleWorkerReadyResponse (message
: MessageValue
<Response
>): void {
1535 if (message
.ready
=== false) {
1537 `Worker ${message.workerId as number} failed to initialize`
1540 const workerInfo
= this.getWorkerInfo(
1541 this.getWorkerNodeKeyByWorkerId(message
.workerId
)
1543 workerInfo
.ready
= message
.ready
as boolean
1544 workerInfo
.taskFunctionNames
= message
.taskFunctionNames
1546 this.emitter
?.emit(PoolEvents
.ready
, this.info
)
1550 private handleTaskExecutionResponse (message
: MessageValue
<Response
>): void {
1551 const { taskId
, workerError
, data
} = message
1552 const promiseResponse
= this.promiseResponseMap
.get(taskId
as string)
1553 if (promiseResponse
!= null) {
1554 if (workerError
!= null) {
1555 this.emitter
?.emit(PoolEvents
.taskError
, workerError
)
1556 promiseResponse
.reject(workerError
.message
)
1558 promiseResponse
.resolve(data
as Response
)
1560 const workerNodeKey
= promiseResponse
.workerNodeKey
1561 this.afterTaskExecutionHook(workerNodeKey
, message
)
1562 this.workerChoiceStrategyContext
.update(workerNodeKey
)
1563 this.promiseResponseMap
.delete(taskId
as string)
1565 this.opts
.enableTasksQueue
=== true &&
1566 this.tasksQueueSize(workerNodeKey
) > 0 &&
1567 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
<
1568 (this.opts
.tasksQueueOptions
?.concurrency
as number)
1572 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1578 private checkAndEmitTaskExecutionEvents (): void {
1580 this.emitter
?.emit(PoolEvents
.busy
, this.info
)
1584 private checkAndEmitTaskQueuingEvents (): void {
1585 if (this.hasBackPressure()) {
1586 this.emitter
?.emit(PoolEvents
.backPressure
, this.info
)
1590 private checkAndEmitDynamicWorkerCreationEvents (): void {
1591 if (this.type === PoolTypes
.dynamic
) {
1593 this.emitter
?.emit(PoolEvents
.full
, this.info
)
1599 * Gets the worker information given its worker node key.
1601 * @param workerNodeKey - The worker node key.
1602 * @returns The worker information.
1604 protected getWorkerInfo (workerNodeKey
: number): WorkerInfo
{
1605 return this.workerNodes
[workerNodeKey
].info
1609 * Adds the given worker in the pool worker nodes.
1611 * @param worker - The worker.
1612 * @returns The added worker node key.
1613 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the added worker node is not found.
1615 private addWorkerNode (worker
: Worker
): number {
1616 const workerNode
= new WorkerNode
<Worker
, Data
>(
1618 this.opts
.tasksQueueOptions
?.size
?? Math.pow(this.maxSize
, 2)
1620 // Flag the worker node as ready at pool startup.
1621 if (this.starting
) {
1622 workerNode
.info
.ready
= true
1624 this.workerNodes
.push(workerNode
)
1625 const workerNodeKey
= this.getWorkerNodeKeyByWorker(worker
)
1626 if (workerNodeKey
=== -1) {
1627 throw new Error('Worker added not found in worker nodes')
1629 return workerNodeKey
1633 * Removes the given worker from the pool worker nodes.
1635 * @param worker - The worker.
1637 private removeWorkerNode (worker
: Worker
): void {
1638 const workerNodeKey
= this.getWorkerNodeKeyByWorker(worker
)
1639 if (workerNodeKey
!== -1) {
1640 this.workerNodes
.splice(workerNodeKey
, 1)
1641 this.workerChoiceStrategyContext
.remove(workerNodeKey
)
1646 public hasWorkerNodeBackPressure (workerNodeKey
: number): boolean {
1648 this.opts
.enableTasksQueue
=== true &&
1649 this.workerNodes
[workerNodeKey
].hasBackPressure()
1653 private hasBackPressure (): boolean {
1655 this.opts
.enableTasksQueue
=== true &&
1656 this.workerNodes
.findIndex(
1657 workerNode
=> !workerNode
.hasBackPressure()
1663 * Executes the given task on the worker given its worker node key.
1665 * @param workerNodeKey - The worker node key.
1666 * @param task - The task to execute.
1668 private executeTask (workerNodeKey
: number, task
: Task
<Data
>): void {
1669 this.beforeTaskExecutionHook(workerNodeKey
, task
)
1670 this.sendToWorker(workerNodeKey
, task
, task
.transferList
)
1671 this.checkAndEmitTaskExecutionEvents()
1674 private enqueueTask (workerNodeKey
: number, task
: Task
<Data
>): number {
1675 const tasksQueueSize
= this.workerNodes
[workerNodeKey
].enqueueTask(task
)
1676 this.checkAndEmitTaskQueuingEvents()
1677 return tasksQueueSize
1680 private dequeueTask (workerNodeKey
: number): Task
<Data
> | undefined {
1681 return this.workerNodes
[workerNodeKey
].dequeueTask()
1684 private tasksQueueSize (workerNodeKey
: number): number {
1685 return this.workerNodes
[workerNodeKey
].tasksQueueSize()
1688 protected flushTasksQueue (workerNodeKey
: number): void {
1689 while (this.tasksQueueSize(workerNodeKey
) > 0) {
1692 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1695 this.workerNodes
[workerNodeKey
].clearTasksQueue()
1698 private flushTasksQueues (): void {
1699 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
1700 this.flushTasksQueue(workerNodeKey
)