1 import { randomUUID
} from
'node:crypto'
2 import { performance
} from
'node:perf_hooks'
3 import type { TransferListItem
} from
'node:worker_threads'
4 import { EventEmitterAsyncResource
} from
'node:events'
5 import { AsyncResource
} from
'node:async_hooks'
8 PromiseResponseWrapper
,
10 } from
'../utility-types'
13 DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
,
25 import { KillBehaviors
} from
'../worker/worker-options'
26 import type { TaskFunction
} from
'../worker/task-functions'
34 type TasksQueueOptions
41 WorkerNodeEventDetail
,
46 type MeasurementStatisticsRequirements
,
48 WorkerChoiceStrategies
,
49 type WorkerChoiceStrategy
,
50 type WorkerChoiceStrategyOptions
51 } from
'./selection-strategies/selection-strategies-types'
52 import { WorkerChoiceStrategyContext
} from
'./selection-strategies/worker-choice-strategy-context'
53 import { version
} from
'./version'
54 import { WorkerNode
} from
'./worker-node'
57 checkValidTasksQueueOptions
,
58 checkValidWorkerChoiceStrategy
,
59 updateMeasurementStatistics
63 * Base class that implements some shared logic for all poolifier pools.
65 * @typeParam Worker - Type of worker which manages this pool.
66 * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
67 * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
69 export abstract class AbstractPool
<
70 Worker
extends IWorker
,
73 > implements IPool
<Worker
, Data
, Response
> {
75 public readonly workerNodes
: Array<IWorkerNode
<Worker
, Data
>> = []
78 public emitter
?: EventEmitterAsyncResource
81 * Dynamic pool maximum size property placeholder.
83 protected readonly max
?: number
86 * The task execution response promise map:
87 * - `key`: The message id of each submitted task.
88 * - `value`: An object that contains the worker, the execution response promise resolve and reject callbacks.
90 * When we receive a message from the worker, we get a map entry with the promise resolve/reject bound to the message id.
92 protected promiseResponseMap
: Map
<string, PromiseResponseWrapper
<Response
>> =
93 new Map
<string, PromiseResponseWrapper
<Response
>>()
96 * Worker choice strategy context referencing a worker choice algorithm implementation.
98 protected workerChoiceStrategyContext
: WorkerChoiceStrategyContext
<
105 * The task functions added at runtime map:
106 * - `key`: The task function name.
107 * - `value`: The task function itself.
109 private readonly taskFunctions
: Map
<string, TaskFunction
<Data
, Response
>>
112 * Whether the pool is started or not.
114 private started
: boolean
116 * Whether the pool is starting or not.
118 private starting
: boolean
120 * Whether the pool is destroying or not.
122 private destroying
: boolean
124 * Whether the pool ready event has been emitted or not.
126 private readyEventEmitted
: boolean
128 * The start timestamp of the pool.
130 private readonly startTimestamp
133 * Constructs a new poolifier pool.
135 * @param numberOfWorkers - Number of workers that this pool should manage.
136 * @param filePath - Path to the worker file.
137 * @param opts - Options for the pool.
140 protected readonly numberOfWorkers
: number,
141 protected readonly filePath
: string,
142 protected readonly opts
: PoolOptions
<Worker
>
144 if (!this.isMain()) {
146 'Cannot start a pool from a worker with the same type as the pool'
149 checkFilePath(this.filePath
)
150 this.checkNumberOfWorkers(this.numberOfWorkers
)
151 this.checkPoolOptions(this.opts
)
153 this.chooseWorkerNode
= this.chooseWorkerNode
.bind(this)
154 this.executeTask
= this.executeTask
.bind(this)
155 this.enqueueTask
= this.enqueueTask
.bind(this)
157 if (this.opts
.enableEvents
=== true) {
158 this.initializeEventEmitter()
160 this.workerChoiceStrategyContext
= new WorkerChoiceStrategyContext
<
166 this.opts
.workerChoiceStrategy
,
167 this.opts
.workerChoiceStrategyOptions
172 this.taskFunctions
= new Map
<string, TaskFunction
<Data
, Response
>>()
175 this.starting
= false
176 this.destroying
= false
177 this.readyEventEmitted
= false
178 if (this.opts
.startWorkers
=== true) {
182 this.startTimestamp
= performance
.now()
185 private checkNumberOfWorkers (numberOfWorkers
: number): void {
186 if (numberOfWorkers
== null) {
188 'Cannot instantiate a pool without specifying the number of workers'
190 } else if (!Number.isSafeInteger(numberOfWorkers
)) {
192 'Cannot instantiate a pool with a non safe integer number of workers'
194 } else if (numberOfWorkers
< 0) {
195 throw new RangeError(
196 'Cannot instantiate a pool with a negative number of workers'
198 } else if (this.type === PoolTypes
.fixed
&& numberOfWorkers
=== 0) {
199 throw new RangeError('Cannot instantiate a fixed pool with zero worker')
203 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
204 if (isPlainObject(opts
)) {
205 this.opts
.startWorkers
= opts
.startWorkers
?? true
206 checkValidWorkerChoiceStrategy(
207 opts
.workerChoiceStrategy
as WorkerChoiceStrategy
209 this.opts
.workerChoiceStrategy
=
210 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
211 this.checkValidWorkerChoiceStrategyOptions(
212 opts
.workerChoiceStrategyOptions
as WorkerChoiceStrategyOptions
214 this.opts
.workerChoiceStrategyOptions
= {
215 ...DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
,
216 ...opts
.workerChoiceStrategyOptions
218 this.opts
.restartWorkerOnError
= opts
.restartWorkerOnError
?? true
219 this.opts
.enableEvents
= opts
.enableEvents
?? true
220 this.opts
.enableTasksQueue
= opts
.enableTasksQueue
?? false
221 if (this.opts
.enableTasksQueue
) {
222 checkValidTasksQueueOptions(opts
.tasksQueueOptions
as TasksQueueOptions
)
223 this.opts
.tasksQueueOptions
= this.buildTasksQueueOptions(
224 opts
.tasksQueueOptions
as TasksQueueOptions
228 throw new TypeError('Invalid pool options: must be a plain object')
232 private checkValidWorkerChoiceStrategyOptions (
233 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
236 workerChoiceStrategyOptions
!= null &&
237 !isPlainObject(workerChoiceStrategyOptions
)
240 'Invalid worker choice strategy options: must be a plain object'
244 workerChoiceStrategyOptions
?.retries
!= null &&
245 !Number.isSafeInteger(workerChoiceStrategyOptions
.retries
)
248 'Invalid worker choice strategy options: retries must be an integer'
252 workerChoiceStrategyOptions
?.retries
!= null &&
253 workerChoiceStrategyOptions
.retries
< 0
255 throw new RangeError(
256 `Invalid worker choice strategy options: retries '${workerChoiceStrategyOptions.retries}' must be greater or equal than zero`
260 workerChoiceStrategyOptions
?.weights
!= null &&
261 Object.keys(workerChoiceStrategyOptions
.weights
).length
!== this.maxSize
264 'Invalid worker choice strategy options: must have a weight for each worker node'
268 workerChoiceStrategyOptions
?.measurement
!= null &&
269 !Object.values(Measurements
).includes(
270 workerChoiceStrategyOptions
.measurement
274 `Invalid worker choice strategy options: invalid measurement '${workerChoiceStrategyOptions.measurement}'`
279 private initializeEventEmitter (): void {
280 this.emitter
= new EventEmitterAsyncResource({
281 name
: `poolifier:${this.type}-${this.worker}-pool`
286 public get
info (): PoolInfo
{
291 started
: this.started
,
293 strategy
: this.opts
.workerChoiceStrategy
as WorkerChoiceStrategy
,
294 minSize
: this.minSize
,
295 maxSize
: this.maxSize
,
296 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
297 .runTime
.aggregate
&&
298 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
299 .waitTime
.aggregate
&& { utilization
: round(this.utilization
) }),
300 workerNodes
: this.workerNodes
.length
,
301 idleWorkerNodes
: this.workerNodes
.reduce(
302 (accumulator
, workerNode
) =>
303 workerNode
.usage
.tasks
.executing
=== 0
308 busyWorkerNodes
: this.workerNodes
.reduce(
309 (accumulator
, _workerNode
, workerNodeKey
) =>
310 this.isWorkerNodeBusy(workerNodeKey
) ? accumulator
+ 1 : accumulator
,
313 executedTasks
: this.workerNodes
.reduce(
314 (accumulator
, workerNode
) =>
315 accumulator
+ workerNode
.usage
.tasks
.executed
,
318 executingTasks
: this.workerNodes
.reduce(
319 (accumulator
, workerNode
) =>
320 accumulator
+ workerNode
.usage
.tasks
.executing
,
323 ...(this.opts
.enableTasksQueue
=== true && {
324 queuedTasks
: this.workerNodes
.reduce(
325 (accumulator
, workerNode
) =>
326 accumulator
+ workerNode
.usage
.tasks
.queued
,
330 ...(this.opts
.enableTasksQueue
=== true && {
331 maxQueuedTasks
: this.workerNodes
.reduce(
332 (accumulator
, workerNode
) =>
333 accumulator
+ (workerNode
.usage
.tasks
?.maxQueued
?? 0),
337 ...(this.opts
.enableTasksQueue
=== true && {
338 backPressure
: this.hasBackPressure()
340 ...(this.opts
.enableTasksQueue
=== true && {
341 stolenTasks
: this.workerNodes
.reduce(
342 (accumulator
, workerNode
) =>
343 accumulator
+ workerNode
.usage
.tasks
.stolen
,
347 failedTasks
: this.workerNodes
.reduce(
348 (accumulator
, workerNode
) =>
349 accumulator
+ workerNode
.usage
.tasks
.failed
,
352 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
353 .runTime
.aggregate
&& {
357 ...this.workerNodes
.map(
358 workerNode
=> workerNode
.usage
.runTime
?.minimum
?? Infinity
364 ...this.workerNodes
.map(
365 workerNode
=> workerNode
.usage
.runTime
?.maximum
?? -Infinity
369 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
370 .runTime
.average
&& {
373 this.workerNodes
.reduce
<number[]>(
374 (accumulator
, workerNode
) =>
375 accumulator
.concat(workerNode
.usage
.runTime
.history
),
381 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
385 this.workerNodes
.reduce
<number[]>(
386 (accumulator
, workerNode
) =>
387 accumulator
.concat(workerNode
.usage
.runTime
.history
),
395 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
396 .waitTime
.aggregate
&& {
400 ...this.workerNodes
.map(
401 workerNode
=> workerNode
.usage
.waitTime
?.minimum
?? Infinity
407 ...this.workerNodes
.map(
408 workerNode
=> workerNode
.usage
.waitTime
?.maximum
?? -Infinity
412 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
413 .waitTime
.average
&& {
416 this.workerNodes
.reduce
<number[]>(
417 (accumulator
, workerNode
) =>
418 accumulator
.concat(workerNode
.usage
.waitTime
.history
),
424 ...(this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
425 .waitTime
.median
&& {
428 this.workerNodes
.reduce
<number[]>(
429 (accumulator
, workerNode
) =>
430 accumulator
.concat(workerNode
.usage
.waitTime
.history
),
442 * The pool readiness boolean status.
444 private get
ready (): boolean {
446 this.workerNodes
.reduce(
447 (accumulator
, workerNode
) =>
448 !workerNode
.info
.dynamic
&& workerNode
.info
.ready
457 * The approximate pool utilization.
459 * @returns The pool utilization.
461 private get
utilization (): number {
462 const poolTimeCapacity
=
463 (performance
.now() - this.startTimestamp
) * this.maxSize
464 const totalTasksRunTime
= this.workerNodes
.reduce(
465 (accumulator
, workerNode
) =>
466 accumulator
+ (workerNode
.usage
.runTime
?.aggregate
?? 0),
469 const totalTasksWaitTime
= this.workerNodes
.reduce(
470 (accumulator
, workerNode
) =>
471 accumulator
+ (workerNode
.usage
.waitTime
?.aggregate
?? 0),
474 return (totalTasksRunTime
+ totalTasksWaitTime
) / poolTimeCapacity
480 * If it is `'dynamic'`, it provides the `max` property.
482 protected abstract get
type (): PoolType
487 protected abstract get
worker (): WorkerType
490 * The pool minimum size.
492 protected get
minSize (): number {
493 return this.numberOfWorkers
497 * The pool maximum size.
499 protected get
maxSize (): number {
500 return this.max
?? this.numberOfWorkers
504 * Checks if the worker id sent in the received message from a worker is valid.
506 * @param message - The received message.
507 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the worker id is invalid.
509 private checkMessageWorkerId (message
: MessageValue
<Data
| Response
>): void {
510 if (message
.workerId
== null) {
511 throw new Error('Worker message received without worker id')
512 } else if (this.getWorkerNodeKeyByWorkerId(message
.workerId
) === -1) {
514 `Worker message received from unknown worker '${message.workerId}'`
520 * Gets the given worker its worker node key.
522 * @param worker - The worker.
523 * @returns The worker node key if found in the pool worker nodes, `-1` otherwise.
525 private getWorkerNodeKeyByWorker (worker
: Worker
): number {
526 return this.workerNodes
.findIndex(
527 workerNode
=> workerNode
.worker
=== worker
532 * Gets the worker node key given its worker id.
534 * @param workerId - The worker id.
535 * @returns The worker node key if the worker id is found in the pool worker nodes, `-1` otherwise.
537 private getWorkerNodeKeyByWorkerId (workerId
: number | undefined): number {
538 return this.workerNodes
.findIndex(
539 workerNode
=> workerNode
.info
.id
=== workerId
544 public setWorkerChoiceStrategy (
545 workerChoiceStrategy
: WorkerChoiceStrategy
,
546 workerChoiceStrategyOptions
?: WorkerChoiceStrategyOptions
548 checkValidWorkerChoiceStrategy(workerChoiceStrategy
)
549 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
550 this.workerChoiceStrategyContext
.setWorkerChoiceStrategy(
551 this.opts
.workerChoiceStrategy
553 if (workerChoiceStrategyOptions
!= null) {
554 this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
556 for (const [workerNodeKey
, workerNode
] of this.workerNodes
.entries()) {
557 workerNode
.resetUsage()
558 this.sendStatisticsMessageToWorker(workerNodeKey
)
563 public setWorkerChoiceStrategyOptions (
564 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
566 this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
567 this.opts
.workerChoiceStrategyOptions
= {
568 ...DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
,
569 ...workerChoiceStrategyOptions
571 this.workerChoiceStrategyContext
.setOptions(
572 this.opts
.workerChoiceStrategyOptions
577 public enableTasksQueue (
579 tasksQueueOptions
?: TasksQueueOptions
581 if (this.opts
.enableTasksQueue
=== true && !enable
) {
582 this.unsetTaskStealing()
583 this.unsetTasksStealingOnBackPressure()
584 this.flushTasksQueues()
586 this.opts
.enableTasksQueue
= enable
587 this.setTasksQueueOptions(tasksQueueOptions
as TasksQueueOptions
)
591 public setTasksQueueOptions (tasksQueueOptions
: TasksQueueOptions
): void {
592 if (this.opts
.enableTasksQueue
=== true) {
593 checkValidTasksQueueOptions(tasksQueueOptions
)
594 this.opts
.tasksQueueOptions
=
595 this.buildTasksQueueOptions(tasksQueueOptions
)
596 this.setTasksQueueSize(this.opts
.tasksQueueOptions
.size
as number)
597 if (this.opts
.tasksQueueOptions
.taskStealing
=== true) {
598 this.unsetTaskStealing()
599 this.setTaskStealing()
601 this.unsetTaskStealing()
603 if (this.opts
.tasksQueueOptions
.tasksStealingOnBackPressure
=== true) {
604 this.unsetTasksStealingOnBackPressure()
605 this.setTasksStealingOnBackPressure()
607 this.unsetTasksStealingOnBackPressure()
609 } else if (this.opts
.tasksQueueOptions
!= null) {
610 delete this.opts
.tasksQueueOptions
614 private buildTasksQueueOptions (
615 tasksQueueOptions
: TasksQueueOptions
616 ): TasksQueueOptions
{
619 size
: Math.pow(this.maxSize
, 2),
622 tasksStealingOnBackPressure
: true
628 private setTasksQueueSize (size
: number): void {
629 for (const workerNode
of this.workerNodes
) {
630 workerNode
.tasksQueueBackPressureSize
= size
634 private setTaskStealing (): void {
635 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
636 this.workerNodes
[workerNodeKey
].on(
638 this.handleIdleWorkerNodeEvent
643 private unsetTaskStealing (): void {
644 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
645 this.workerNodes
[workerNodeKey
].off(
647 this.handleIdleWorkerNodeEvent
652 private setTasksStealingOnBackPressure (): void {
653 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
654 this.workerNodes
[workerNodeKey
].on(
656 this.handleBackPressureEvent
661 private unsetTasksStealingOnBackPressure (): void {
662 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
663 this.workerNodes
[workerNodeKey
].off(
665 this.handleBackPressureEvent
671 * Whether the pool is full or not.
673 * The pool filling boolean status.
675 protected get
full (): boolean {
676 return this.workerNodes
.length
>= this.maxSize
680 * Whether the pool is busy or not.
682 * The pool busyness boolean status.
684 protected abstract get
busy (): boolean
687 * Whether worker nodes are executing concurrently their tasks quota or not.
689 * @returns Worker nodes busyness boolean status.
691 protected internalBusy (): boolean {
692 if (this.opts
.enableTasksQueue
=== true) {
694 this.workerNodes
.findIndex(
696 workerNode
.info
.ready
&&
697 workerNode
.usage
.tasks
.executing
<
698 (this.opts
.tasksQueueOptions
?.concurrency
as number)
703 this.workerNodes
.findIndex(
705 workerNode
.info
.ready
&& workerNode
.usage
.tasks
.executing
=== 0
710 private isWorkerNodeBusy (workerNodeKey
: number): boolean {
711 if (this.opts
.enableTasksQueue
=== true) {
713 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
>=
714 (this.opts
.tasksQueueOptions
?.concurrency
as number)
717 return this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
> 0
720 private async sendTaskFunctionOperationToWorker (
721 workerNodeKey
: number,
722 message
: MessageValue
<Data
>
723 ): Promise
<boolean> {
724 return await new Promise
<boolean>((resolve
, reject
) => {
725 const taskFunctionOperationListener
= (
726 message
: MessageValue
<Response
>
728 this.checkMessageWorkerId(message
)
729 const workerId
= this.getWorkerInfo(workerNodeKey
).id
as number
731 message
.taskFunctionOperationStatus
!= null &&
732 message
.workerId
=== workerId
734 if (message
.taskFunctionOperationStatus
) {
736 } else if (!message
.taskFunctionOperationStatus
) {
739 `Task function operation '${
740 message.taskFunctionOperation as string
741 }' failed on worker ${message.workerId} with error: '${
742 message.workerError?.message as string
747 this.deregisterWorkerMessageListener(
748 this.getWorkerNodeKeyByWorkerId(message
.workerId
),
749 taskFunctionOperationListener
753 this.registerWorkerMessageListener(
755 taskFunctionOperationListener
757 this.sendToWorker(workerNodeKey
, message
)
761 private async sendTaskFunctionOperationToWorkers (
762 message
: MessageValue
<Data
>
763 ): Promise
<boolean> {
764 return await new Promise
<boolean>((resolve
, reject
) => {
765 const responsesReceived
= new Array<MessageValue
<Response
>>()
766 const taskFunctionOperationsListener
= (
767 message
: MessageValue
<Response
>
769 this.checkMessageWorkerId(message
)
770 if (message
.taskFunctionOperationStatus
!= null) {
771 responsesReceived
.push(message
)
772 if (responsesReceived
.length
=== this.workerNodes
.length
) {
774 responsesReceived
.every(
775 message
=> message
.taskFunctionOperationStatus
=== true
780 responsesReceived
.some(
781 message
=> message
.taskFunctionOperationStatus
=== false
784 const errorResponse
= responsesReceived
.find(
785 response
=> response
.taskFunctionOperationStatus
=== false
789 `Task function operation '${
790 message.taskFunctionOperation as string
791 }' failed on worker ${
792 errorResponse?.workerId as number
794 errorResponse?.workerError?.message as string
799 this.deregisterWorkerMessageListener(
800 this.getWorkerNodeKeyByWorkerId(message
.workerId
),
801 taskFunctionOperationsListener
806 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
807 this.registerWorkerMessageListener(
809 taskFunctionOperationsListener
811 this.sendToWorker(workerNodeKey
, message
)
817 public hasTaskFunction (name
: string): boolean {
818 for (const workerNode
of this.workerNodes
) {
820 Array.isArray(workerNode
.info
.taskFunctionNames
) &&
821 workerNode
.info
.taskFunctionNames
.includes(name
)
830 public async addTaskFunction (
832 fn
: TaskFunction
<Data
, Response
>
833 ): Promise
<boolean> {
834 if (typeof name
!== 'string') {
835 throw new TypeError('name argument must be a string')
837 if (typeof name
=== 'string' && name
.trim().length
=== 0) {
838 throw new TypeError('name argument must not be an empty string')
840 if (typeof fn
!== 'function') {
841 throw new TypeError('fn argument must be a function')
843 const opResult
= await this.sendTaskFunctionOperationToWorkers({
844 taskFunctionOperation
: 'add',
845 taskFunctionName
: name
,
846 taskFunction
: fn
.toString()
848 this.taskFunctions
.set(name
, fn
)
853 public async removeTaskFunction (name
: string): Promise
<boolean> {
854 if (!this.taskFunctions
.has(name
)) {
856 'Cannot remove a task function not handled on the pool side'
859 const opResult
= await this.sendTaskFunctionOperationToWorkers({
860 taskFunctionOperation
: 'remove',
861 taskFunctionName
: name
863 this.deleteTaskFunctionWorkerUsages(name
)
864 this.taskFunctions
.delete(name
)
869 public listTaskFunctionNames (): string[] {
870 for (const workerNode
of this.workerNodes
) {
872 Array.isArray(workerNode
.info
.taskFunctionNames
) &&
873 workerNode
.info
.taskFunctionNames
.length
> 0
875 return workerNode
.info
.taskFunctionNames
882 public async setDefaultTaskFunction (name
: string): Promise
<boolean> {
883 return await this.sendTaskFunctionOperationToWorkers({
884 taskFunctionOperation
: 'default',
885 taskFunctionName
: name
889 private deleteTaskFunctionWorkerUsages (name
: string): void {
890 for (const workerNode
of this.workerNodes
) {
891 workerNode
.deleteTaskFunctionWorkerUsage(name
)
895 private shallExecuteTask (workerNodeKey
: number): boolean {
897 this.tasksQueueSize(workerNodeKey
) === 0 &&
898 this.workerNodes
[workerNodeKey
].usage
.tasks
.executing
<
899 (this.opts
.tasksQueueOptions
?.concurrency
as number)
904 public async execute (
907 transferList
?: TransferListItem
[]
908 ): Promise
<Response
> {
909 return await new Promise
<Response
>((resolve
, reject
) => {
911 reject(new Error('Cannot execute a task on not started pool'))
914 if (this.destroying
) {
915 reject(new Error('Cannot execute a task on destroying pool'))
918 if (name
!= null && typeof name
!== 'string') {
919 reject(new TypeError('name argument must be a string'))
924 typeof name
=== 'string' &&
925 name
.trim().length
=== 0
927 reject(new TypeError('name argument must not be an empty string'))
930 if (transferList
!= null && !Array.isArray(transferList
)) {
931 reject(new TypeError('transferList argument must be an array'))
934 const timestamp
= performance
.now()
935 const workerNodeKey
= this.chooseWorkerNode()
936 const task
: Task
<Data
> = {
937 name
: name
?? DEFAULT_TASK_NAME
,
938 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
939 data
: data
?? ({} as Data
),
944 this.promiseResponseMap
.set(task
.taskId
as string, {
948 ...(this.emitter
!= null && {
949 asyncResource
: new AsyncResource('poolifier:task', {
950 triggerAsyncId
: this.emitter
.asyncId
,
951 requireManualDestroy
: true
956 this.opts
.enableTasksQueue
=== false ||
957 (this.opts
.enableTasksQueue
=== true &&
958 this.shallExecuteTask(workerNodeKey
))
960 this.executeTask(workerNodeKey
, task
)
962 this.enqueueTask(workerNodeKey
, task
)
968 public start (): void {
970 throw new Error('Cannot start an already started pool')
973 throw new Error('Cannot start an already starting pool')
975 if (this.destroying
) {
976 throw new Error('Cannot start a destroying pool')
980 this.workerNodes
.reduce(
981 (accumulator
, workerNode
) =>
982 !workerNode
.info
.dynamic
? accumulator
+ 1 : accumulator
,
984 ) < this.numberOfWorkers
986 this.createAndSetupWorkerNode()
988 this.starting
= false
993 public async destroy (): Promise
<void> {
995 throw new Error('Cannot destroy an already destroyed pool')
998 throw new Error('Cannot destroy an starting pool')
1000 if (this.destroying
) {
1001 throw new Error('Cannot destroy an already destroying pool')
1003 this.destroying
= true
1005 this.workerNodes
.map(async (_workerNode
, workerNodeKey
) => {
1006 await this.destroyWorkerNode(workerNodeKey
)
1009 this.emitter
?.emit(PoolEvents
.destroy
, this.info
)
1010 this.emitter
?.emitDestroy()
1011 this.emitter
?.removeAllListeners()
1012 this.readyEventEmitted
= false
1013 this.destroying
= false
1014 this.started
= false
1017 protected async sendKillMessageToWorker (
1018 workerNodeKey
: number
1020 await new Promise
<void>((resolve
, reject
) => {
1021 const killMessageListener
= (message
: MessageValue
<Response
>): void => {
1022 this.checkMessageWorkerId(message
)
1023 if (message
.kill
=== 'success') {
1025 } else if (message
.kill
=== 'failure') {
1028 `Kill message handling failed on worker ${
1029 message.workerId as number
1035 // FIXME: should be registered only once
1036 this.registerWorkerMessageListener(workerNodeKey
, killMessageListener
)
1037 this.sendToWorker(workerNodeKey
, { kill
: true })
1042 * Terminates the worker node given its worker node key.
1044 * @param workerNodeKey - The worker node key.
1046 protected async destroyWorkerNode (workerNodeKey
: number): Promise
<void> {
1047 this.flagWorkerNodeAsNotReady(workerNodeKey
)
1048 this.flushTasksQueue(workerNodeKey
)
1049 // FIXME: wait for tasks to be finished
1050 const workerNode
= this.workerNodes
[workerNodeKey
]
1051 await this.sendKillMessageToWorker(workerNodeKey
)
1052 await workerNode
.terminate()
1056 * Setup hook to execute code before worker nodes are created in the abstract constructor.
1057 * Can be overridden.
1061 protected setupHook (): void {
1062 /* Intentionally empty */
1066 * Should return whether the worker is the main worker or not.
1068 protected abstract isMain (): boolean
1071 * Hook executed before the worker task execution.
1072 * Can be overridden.
1074 * @param workerNodeKey - The worker node key.
1075 * @param task - The task to execute.
1077 protected beforeTaskExecutionHook (
1078 workerNodeKey
: number,
1081 if (this.workerNodes
[workerNodeKey
]?.usage
!= null) {
1082 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
1083 ++workerUsage
.tasks
.executing
1084 this.updateWaitTimeWorkerUsage(workerUsage
, task
)
1087 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1088 this.workerNodes
[workerNodeKey
].getTaskFunctionWorkerUsage(
1092 const taskFunctionWorkerUsage
= this.workerNodes
[
1094 ].getTaskFunctionWorkerUsage(task
.name
as string) as WorkerUsage
1095 ++taskFunctionWorkerUsage
.tasks
.executing
1096 this.updateWaitTimeWorkerUsage(taskFunctionWorkerUsage
, task
)
1101 * Hook executed after the worker task execution.
1102 * Can be overridden.
1104 * @param workerNodeKey - The worker node key.
1105 * @param message - The received message.
1107 protected afterTaskExecutionHook (
1108 workerNodeKey
: number,
1109 message
: MessageValue
<Response
>
1111 if (this.workerNodes
[workerNodeKey
]?.usage
!= null) {
1112 const workerUsage
= this.workerNodes
[workerNodeKey
].usage
1113 this.updateTaskStatisticsWorkerUsage(workerUsage
, message
)
1114 this.updateRunTimeWorkerUsage(workerUsage
, message
)
1115 this.updateEluWorkerUsage(workerUsage
, message
)
1118 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1119 this.workerNodes
[workerNodeKey
].getTaskFunctionWorkerUsage(
1120 message
.taskPerformance
?.name
as string
1123 const taskFunctionWorkerUsage
= this.workerNodes
[
1125 ].getTaskFunctionWorkerUsage(
1126 message
.taskPerformance
?.name
as string
1128 this.updateTaskStatisticsWorkerUsage(taskFunctionWorkerUsage
, message
)
1129 this.updateRunTimeWorkerUsage(taskFunctionWorkerUsage
, message
)
1130 this.updateEluWorkerUsage(taskFunctionWorkerUsage
, message
)
1135 * Whether the worker node shall update its task function worker usage or not.
1137 * @param workerNodeKey - The worker node key.
1138 * @returns `true` if the worker node shall update its task function worker usage, `false` otherwise.
1140 private shallUpdateTaskFunctionWorkerUsage (workerNodeKey
: number): boolean {
1141 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1143 workerInfo
!= null &&
1144 Array.isArray(workerInfo
.taskFunctionNames
) &&
1145 workerInfo
.taskFunctionNames
.length
> 2
1149 private updateTaskStatisticsWorkerUsage (
1150 workerUsage
: WorkerUsage
,
1151 message
: MessageValue
<Response
>
1153 const workerTaskStatistics
= workerUsage
.tasks
1155 workerTaskStatistics
.executing
!= null &&
1156 workerTaskStatistics
.executing
> 0
1158 --workerTaskStatistics
.executing
1160 if (message
.workerError
== null) {
1161 ++workerTaskStatistics
.executed
1163 ++workerTaskStatistics
.failed
1167 private updateRunTimeWorkerUsage (
1168 workerUsage
: WorkerUsage
,
1169 message
: MessageValue
<Response
>
1171 if (message
.workerError
!= null) {
1174 updateMeasurementStatistics(
1175 workerUsage
.runTime
,
1176 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
,
1177 message
.taskPerformance
?.runTime
?? 0
1181 private updateWaitTimeWorkerUsage (
1182 workerUsage
: WorkerUsage
,
1185 const timestamp
= performance
.now()
1186 const taskWaitTime
= timestamp
- (task
.timestamp
?? timestamp
)
1187 updateMeasurementStatistics(
1188 workerUsage
.waitTime
,
1189 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().waitTime
,
1194 private updateEluWorkerUsage (
1195 workerUsage
: WorkerUsage
,
1196 message
: MessageValue
<Response
>
1198 if (message
.workerError
!= null) {
1201 const eluTaskStatisticsRequirements
: MeasurementStatisticsRequirements
=
1202 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
1203 updateMeasurementStatistics(
1204 workerUsage
.elu
.active
,
1205 eluTaskStatisticsRequirements
,
1206 message
.taskPerformance
?.elu
?.active
?? 0
1208 updateMeasurementStatistics(
1209 workerUsage
.elu
.idle
,
1210 eluTaskStatisticsRequirements
,
1211 message
.taskPerformance
?.elu
?.idle
?? 0
1213 if (eluTaskStatisticsRequirements
.aggregate
) {
1214 if (message
.taskPerformance
?.elu
!= null) {
1215 if (workerUsage
.elu
.utilization
!= null) {
1216 workerUsage
.elu
.utilization
=
1217 (workerUsage
.elu
.utilization
+
1218 message
.taskPerformance
.elu
.utilization
) /
1221 workerUsage
.elu
.utilization
= message
.taskPerformance
.elu
.utilization
1228 * Chooses a worker node for the next task.
1230 * The default worker choice strategy uses a round robin algorithm to distribute the tasks.
1232 * @returns The chosen worker node key
1234 private chooseWorkerNode (): number {
1235 if (this.shallCreateDynamicWorker()) {
1236 const workerNodeKey
= this.createAndSetupDynamicWorkerNode()
1238 this.workerChoiceStrategyContext
.getStrategyPolicy().dynamicWorkerUsage
1240 return workerNodeKey
1243 return this.workerChoiceStrategyContext
.execute()
1247 * Conditions for dynamic worker creation.
1249 * @returns Whether to create a dynamic worker or not.
1251 private shallCreateDynamicWorker (): boolean {
1252 return this.type === PoolTypes
.dynamic
&& !this.full
&& this.internalBusy()
1256 * Sends a message to worker given its worker node key.
1258 * @param workerNodeKey - The worker node key.
1259 * @param message - The message.
1260 * @param transferList - The optional array of transferable objects.
1262 protected abstract sendToWorker (
1263 workerNodeKey
: number,
1264 message
: MessageValue
<Data
>,
1265 transferList
?: TransferListItem
[]
1269 * Creates a new, completely set up worker node.
1271 * @returns New, completely set up worker node key.
1273 protected createAndSetupWorkerNode (): number {
1274 const workerNode
= this.createWorkerNode()
1275 workerNode
.registerWorkerEventHandler(
1277 this.opts
.onlineHandler
?? EMPTY_FUNCTION
1279 workerNode
.registerWorkerEventHandler(
1281 this.opts
.messageHandler
?? EMPTY_FUNCTION
1283 workerNode
.registerWorkerEventHandler(
1285 this.opts
.errorHandler
?? EMPTY_FUNCTION
1287 workerNode
.registerWorkerEventHandler('error', (error
: Error) => {
1288 workerNode
.info
.ready
= false
1289 this.emitter
?.emit(PoolEvents
.error
, error
)
1294 this.opts
.restartWorkerOnError
=== true
1296 if (workerNode
.info
.dynamic
) {
1297 this.createAndSetupDynamicWorkerNode()
1299 this.createAndSetupWorkerNode()
1302 if (this.started
&& this.opts
.enableTasksQueue
=== true) {
1303 this.redistributeQueuedTasks(this.workerNodes
.indexOf(workerNode
))
1305 workerNode
.terminate().catch(error
=> {
1306 this.emitter
?.emit(PoolEvents
.error
, error
)
1309 workerNode
.registerWorkerEventHandler(
1311 this.opts
.exitHandler
?? EMPTY_FUNCTION
1313 workerNode
.registerOnceWorkerEventHandler('exit', () => {
1314 this.removeWorkerNode(workerNode
)
1316 const workerNodeKey
= this.addWorkerNode(workerNode
)
1317 this.afterWorkerNodeSetup(workerNodeKey
)
1318 return workerNodeKey
1322 * Creates a new, completely set up dynamic worker node.
1324 * @returns New, completely set up dynamic worker node key.
1326 protected createAndSetupDynamicWorkerNode (): number {
1327 const workerNodeKey
= this.createAndSetupWorkerNode()
1328 this.registerWorkerMessageListener(workerNodeKey
, message
=> {
1329 this.checkMessageWorkerId(message
)
1330 const localWorkerNodeKey
= this.getWorkerNodeKeyByWorkerId(
1333 const workerUsage
= this.workerNodes
[localWorkerNodeKey
].usage
1334 // Kill message received from worker
1336 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
1337 (isKillBehavior(KillBehaviors
.SOFT
, message
.kill
) &&
1338 ((this.opts
.enableTasksQueue
=== false &&
1339 workerUsage
.tasks
.executing
=== 0) ||
1340 (this.opts
.enableTasksQueue
=== true &&
1341 workerUsage
.tasks
.executing
=== 0 &&
1342 this.tasksQueueSize(localWorkerNodeKey
) === 0)))
1344 // Flag the worker node as not ready immediately
1345 this.flagWorkerNodeAsNotReady(localWorkerNodeKey
)
1346 this.destroyWorkerNode(localWorkerNodeKey
).catch(error
=> {
1347 this.emitter
?.emit(PoolEvents
.error
, error
)
1351 const workerInfo
= this.getWorkerInfo(workerNodeKey
)
1352 this.sendToWorker(workerNodeKey
, {
1355 if (this.taskFunctions
.size
> 0) {
1356 for (const [taskFunctionName
, taskFunction
] of this.taskFunctions
) {
1357 this.sendTaskFunctionOperationToWorker(workerNodeKey
, {
1358 taskFunctionOperation
: 'add',
1360 taskFunction
: taskFunction
.toString()
1362 this.emitter
?.emit(PoolEvents
.error
, error
)
1366 workerInfo
.dynamic
= true
1368 this.workerChoiceStrategyContext
.getStrategyPolicy().dynamicWorkerReady
||
1369 this.workerChoiceStrategyContext
.getStrategyPolicy().dynamicWorkerUsage
1371 workerInfo
.ready
= true
1373 this.checkAndEmitDynamicWorkerCreationEvents()
1374 return workerNodeKey
1378 * Registers a listener callback on the worker given its worker node key.
1380 * @param workerNodeKey - The worker node key.
1381 * @param listener - The message listener callback.
1383 protected abstract registerWorkerMessageListener
<
1384 Message
extends Data
| Response
1386 workerNodeKey
: number,
1387 listener
: (message
: MessageValue
<Message
>) => void
1391 * Registers once a listener callback on the worker given its worker node key.
1393 * @param workerNodeKey - The worker node key.
1394 * @param listener - The message listener callback.
1396 protected abstract registerOnceWorkerMessageListener
<
1397 Message
extends Data
| Response
1399 workerNodeKey
: number,
1400 listener
: (message
: MessageValue
<Message
>) => void
1404 * Deregisters a listener callback on the worker given its worker node key.
1406 * @param workerNodeKey - The worker node key.
1407 * @param listener - The message listener callback.
1409 protected abstract deregisterWorkerMessageListener
<
1410 Message
extends Data
| Response
1412 workerNodeKey
: number,
1413 listener
: (message
: MessageValue
<Message
>) => void
1417 * Method hooked up after a worker node has been newly created.
1418 * Can be overridden.
1420 * @param workerNodeKey - The newly created worker node key.
1422 protected afterWorkerNodeSetup (workerNodeKey
: number): void {
1423 // Listen to worker messages.
1424 this.registerWorkerMessageListener(
1426 this.workerMessageListener
1428 // Send the startup message to worker.
1429 this.sendStartupMessageToWorker(workerNodeKey
)
1430 // Send the statistics message to worker.
1431 this.sendStatisticsMessageToWorker(workerNodeKey
)
1432 if (this.opts
.enableTasksQueue
=== true) {
1433 if (this.opts
.tasksQueueOptions
?.taskStealing
=== true) {
1434 this.workerNodes
[workerNodeKey
].on(
1436 this.handleIdleWorkerNodeEvent
1439 if (this.opts
.tasksQueueOptions
?.tasksStealingOnBackPressure
=== true) {
1440 this.workerNodes
[workerNodeKey
].on(
1442 this.handleBackPressureEvent
1449 * Sends the startup message to worker given its worker node key.
1451 * @param workerNodeKey - The worker node key.
1453 protected abstract sendStartupMessageToWorker (workerNodeKey
: number): void
1456 * Sends the statistics message to worker given its worker node key.
1458 * @param workerNodeKey - The worker node key.
1460 private sendStatisticsMessageToWorker (workerNodeKey
: number): void {
1461 this.sendToWorker(workerNodeKey
, {
1464 this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1466 elu
: this.workerChoiceStrategyContext
.getTaskStatisticsRequirements()
1472 private handleTask (workerNodeKey
: number, task
: Task
<Data
>): void {
1473 if (this.shallExecuteTask(workerNodeKey
)) {
1474 this.executeTask(workerNodeKey
, task
)
1476 this.enqueueTask(workerNodeKey
, task
)
1480 private redistributeQueuedTasks (workerNodeKey
: number): void {
1481 if (this.workerNodes
.length
<= 1) {
1484 while (this.tasksQueueSize(workerNodeKey
) > 0) {
1485 const destinationWorkerNodeKey
= this.workerNodes
.reduce(
1486 (minWorkerNodeKey
, workerNode
, workerNodeKey
, workerNodes
) => {
1487 return workerNode
.info
.ready
&&
1488 workerNode
.usage
.tasks
.queued
<
1489 workerNodes
[minWorkerNodeKey
].usage
.tasks
.queued
1496 destinationWorkerNodeKey
,
1497 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1502 private updateTaskStolenStatisticsWorkerUsage (
1503 workerNodeKey
: number,
1506 const workerNode
= this.workerNodes
[workerNodeKey
]
1507 if (workerNode
?.usage
!= null) {
1508 ++workerNode
.usage
.tasks
.stolen
1511 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1512 workerNode
.getTaskFunctionWorkerUsage(taskName
) != null
1514 const taskFunctionWorkerUsage
= workerNode
.getTaskFunctionWorkerUsage(
1517 ++taskFunctionWorkerUsage
.tasks
.stolen
1521 private updateTaskSequentiallyStolenStatisticsWorkerUsage (
1522 workerNodeKey
: number
1524 const workerNode
= this.workerNodes
[workerNodeKey
]
1525 if (workerNode
?.usage
!= null) {
1526 ++workerNode
.usage
.tasks
.sequentiallyStolen
1530 private updateTaskSequentiallyStolenStatisticsTaskFunctionWorkerUsage (
1531 workerNodeKey
: number,
1534 const workerNode
= this.workerNodes
[workerNodeKey
]
1536 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1537 workerNode
.getTaskFunctionWorkerUsage(taskName
) != null
1539 const taskFunctionWorkerUsage
= workerNode
.getTaskFunctionWorkerUsage(
1542 ++taskFunctionWorkerUsage
.tasks
.sequentiallyStolen
1546 private resetTaskSequentiallyStolenStatisticsWorkerUsage (
1547 workerNodeKey
: number
1549 const workerNode
= this.workerNodes
[workerNodeKey
]
1550 if (workerNode
?.usage
!= null) {
1551 workerNode
.usage
.tasks
.sequentiallyStolen
= 0
1555 private resetTaskSequentiallyStolenStatisticsTaskFunctionWorkerUsage (
1556 workerNodeKey
: number,
1559 const workerNode
= this.workerNodes
[workerNodeKey
]
1561 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1562 workerNode
.getTaskFunctionWorkerUsage(taskName
) != null
1564 const taskFunctionWorkerUsage
= workerNode
.getTaskFunctionWorkerUsage(
1567 taskFunctionWorkerUsage
.tasks
.sequentiallyStolen
= 0
1571 private readonly handleIdleWorkerNodeEvent
= (
1572 eventDetail
: WorkerNodeEventDetail
,
1573 previousStolenTask
?: Task
<Data
>
1575 if (this.workerNodes
.length
<= 1) {
1578 const { workerNodeKey
} = eventDetail
1579 if (workerNodeKey
== null) {
1581 'WorkerNode event detail workerNodeKey attribute must be defined'
1584 const workerNodeTasksUsage
= this.workerNodes
[workerNodeKey
].usage
.tasks
1586 previousStolenTask
!= null &&
1587 workerNodeTasksUsage
.sequentiallyStolen
> 0 &&
1588 (workerNodeTasksUsage
.executing
> 0 ||
1589 this.tasksQueueSize(workerNodeKey
) > 0)
1591 for (const taskName
of this.workerNodes
[workerNodeKey
].info
1592 .taskFunctionNames
as string[]) {
1593 this.resetTaskSequentiallyStolenStatisticsTaskFunctionWorkerUsage(
1598 this.resetTaskSequentiallyStolenStatisticsWorkerUsage(workerNodeKey
)
1601 const stolenTask
= this.workerNodeStealTask(workerNodeKey
)
1603 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey
) &&
1606 const taskFunctionTasksWorkerUsage
= this.workerNodes
[
1608 ].getTaskFunctionWorkerUsage(stolenTask
.name
as string)
1609 ?.tasks
as TaskStatistics
1611 taskFunctionTasksWorkerUsage
.sequentiallyStolen
=== 0 ||
1612 (previousStolenTask
!= null &&
1613 previousStolenTask
.name
=== stolenTask
.name
&&
1614 taskFunctionTasksWorkerUsage
.sequentiallyStolen
> 0)
1616 this.updateTaskSequentiallyStolenStatisticsTaskFunctionWorkerUsage(
1618 stolenTask
.name
as string
1621 this.resetTaskSequentiallyStolenStatisticsTaskFunctionWorkerUsage(
1623 stolenTask
.name
as string
1627 sleep(exponentialDelay(workerNodeTasksUsage
.sequentiallyStolen
))
1629 this.handleIdleWorkerNodeEvent(eventDetail
, stolenTask
)
1632 .catch(EMPTY_FUNCTION
)
1635 private readonly workerNodeStealTask
= (
1636 workerNodeKey
: number
1637 ): Task
<Data
> | undefined => {
1638 const workerNodes
= this.workerNodes
1641 (workerNodeA
, workerNodeB
) =>
1642 workerNodeB
.usage
.tasks
.queued
- workerNodeA
.usage
.tasks
.queued
1644 const sourceWorkerNode
= workerNodes
.find(
1645 (sourceWorkerNode
, sourceWorkerNodeKey
) =>
1646 sourceWorkerNode
.info
.ready
&&
1647 sourceWorkerNodeKey
!== workerNodeKey
&&
1648 sourceWorkerNode
.usage
.tasks
.queued
> 0
1650 if (sourceWorkerNode
!= null) {
1651 const task
= sourceWorkerNode
.popTask() as Task
<Data
>
1652 this.handleTask(workerNodeKey
, task
)
1653 this.updateTaskSequentiallyStolenStatisticsWorkerUsage(workerNodeKey
)
1654 this.updateTaskStolenStatisticsWorkerUsage(
1662 private readonly handleBackPressureEvent
= (
1663 eventDetail
: WorkerNodeEventDetail
1665 if (this.workerNodes
.length
<= 1) {
1668 const { workerId
} = eventDetail
1669 const sizeOffset
= 1
1670 if ((this.opts
.tasksQueueOptions
?.size
as number) <= sizeOffset
) {
1673 const sourceWorkerNode
=
1674 this.workerNodes
[this.getWorkerNodeKeyByWorkerId(workerId
)]
1675 const workerNodes
= this.workerNodes
1678 (workerNodeA
, workerNodeB
) =>
1679 workerNodeA
.usage
.tasks
.queued
- workerNodeB
.usage
.tasks
.queued
1681 for (const [workerNodeKey
, workerNode
] of workerNodes
.entries()) {
1683 sourceWorkerNode
.usage
.tasks
.queued
> 0 &&
1684 workerNode
.info
.ready
&&
1685 workerNode
.info
.id
!== workerId
&&
1686 workerNode
.usage
.tasks
.queued
<
1687 (this.opts
.tasksQueueOptions
?.size
as number) - sizeOffset
1689 const task
= sourceWorkerNode
.popTask() as Task
<Data
>
1690 this.handleTask(workerNodeKey
, task
)
1691 this.updateTaskStolenStatisticsWorkerUsage(
1700 * This method is the message listener registered on each worker.
1702 protected readonly workerMessageListener
= (
1703 message
: MessageValue
<Response
>
1705 this.checkMessageWorkerId(message
)
1706 const { workerId
, ready
, taskId
, taskFunctionNames
} = message
1707 if (ready
!= null && taskFunctionNames
!= null) {
1708 // Worker ready response received from worker
1709 this.handleWorkerReadyResponse(message
)
1710 } else if (taskId
!= null) {
1711 // Task execution response received from worker
1712 this.handleTaskExecutionResponse(message
)
1713 } else if (taskFunctionNames
!= null) {
1714 // Task function names message received from worker
1716 this.getWorkerNodeKeyByWorkerId(workerId
)
1717 ).taskFunctionNames
= taskFunctionNames
1721 private handleWorkerReadyResponse (message
: MessageValue
<Response
>): void {
1722 const { workerId
, ready
, taskFunctionNames
} = message
1723 if (ready
=== false) {
1724 throw new Error(`Worker ${workerId as number} failed to initialize`)
1726 const workerInfo
= this.getWorkerInfo(
1727 this.getWorkerNodeKeyByWorkerId(workerId
)
1729 workerInfo
.ready
= ready
as boolean
1730 workerInfo
.taskFunctionNames
= taskFunctionNames
1731 if (!this.readyEventEmitted
&& this.ready
) {
1732 this.readyEventEmitted
= true
1733 this.emitter
?.emit(PoolEvents
.ready
, this.info
)
1737 private handleTaskExecutionResponse (message
: MessageValue
<Response
>): void {
1738 const { workerId
, taskId
, workerError
, data
} = message
1739 const promiseResponse
= this.promiseResponseMap
.get(taskId
as string)
1740 if (promiseResponse
!= null) {
1741 const { resolve
, reject
, workerNodeKey
, asyncResource
} = promiseResponse
1742 if (workerError
!= null) {
1743 this.emitter
?.emit(PoolEvents
.taskError
, workerError
)
1744 asyncResource
!= null
1745 ? asyncResource
.runInAsyncScope(
1750 : reject(workerError
.message
)
1752 asyncResource
!= null
1753 ? asyncResource
.runInAsyncScope(resolve
, this.emitter
, data
)
1754 : resolve(data
as Response
)
1756 asyncResource
?.emitDestroy()
1757 this.afterTaskExecutionHook(workerNodeKey
, message
)
1758 this.workerChoiceStrategyContext
.update(workerNodeKey
)
1759 this.promiseResponseMap
.delete(taskId
as string)
1760 if (this.opts
.enableTasksQueue
=== true) {
1761 const workerNodeTasksUsage
= this.workerNodes
[workerNodeKey
].usage
.tasks
1763 this.tasksQueueSize(workerNodeKey
) > 0 &&
1764 workerNodeTasksUsage
.executing
<
1765 (this.opts
.tasksQueueOptions
?.concurrency
as number)
1769 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1773 workerNodeTasksUsage
.executing
=== 0 &&
1774 this.tasksQueueSize(workerNodeKey
) === 0 &&
1775 workerNodeTasksUsage
.sequentiallyStolen
=== 0
1777 this.workerNodes
[workerNodeKey
].emit('idleWorkerNode', {
1778 workerId
: workerId
as number,
1786 private checkAndEmitTaskExecutionEvents (): void {
1788 this.emitter
?.emit(PoolEvents
.busy
, this.info
)
1792 private checkAndEmitTaskQueuingEvents (): void {
1793 if (this.hasBackPressure()) {
1794 this.emitter
?.emit(PoolEvents
.backPressure
, this.info
)
1798 private checkAndEmitDynamicWorkerCreationEvents (): void {
1799 if (this.type === PoolTypes
.dynamic
) {
1801 this.emitter
?.emit(PoolEvents
.full
, this.info
)
1807 * Gets the worker information given its worker node key.
1809 * @param workerNodeKey - The worker node key.
1810 * @returns The worker information.
1812 protected getWorkerInfo (workerNodeKey
: number): WorkerInfo
{
1813 return this.workerNodes
[workerNodeKey
]?.info
1817 * Creates a worker node.
1819 * @returns The created worker node.
1821 private createWorkerNode (): IWorkerNode
<Worker
, Data
> {
1822 const workerNode
= new WorkerNode
<Worker
, Data
>(
1827 workerOptions
: this.opts
.workerOptions
,
1828 tasksQueueBackPressureSize
:
1829 this.opts
.tasksQueueOptions
?.size
?? Math.pow(this.maxSize
, 2)
1832 // Flag the worker node as ready at pool startup.
1833 if (this.starting
) {
1834 workerNode
.info
.ready
= true
1840 * Adds the given worker node in the pool worker nodes.
1842 * @param workerNode - The worker node.
1843 * @returns The added worker node key.
1844 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the added worker node is not found.
1846 private addWorkerNode (workerNode
: IWorkerNode
<Worker
, Data
>): number {
1847 this.workerNodes
.push(workerNode
)
1848 const workerNodeKey
= this.workerNodes
.indexOf(workerNode
)
1849 if (workerNodeKey
=== -1) {
1850 throw new Error('Worker added not found in worker nodes')
1852 return workerNodeKey
1856 * Removes the worker node from the pool worker nodes.
1858 * @param workerNode - The worker node.
1860 private removeWorkerNode (workerNode
: IWorkerNode
<Worker
, Data
>): void {
1861 const workerNodeKey
= this.workerNodes
.indexOf(workerNode
)
1862 if (workerNodeKey
!== -1) {
1863 this.workerNodes
.splice(workerNodeKey
, 1)
1864 this.workerChoiceStrategyContext
.remove(workerNodeKey
)
1868 protected flagWorkerNodeAsNotReady (workerNodeKey
: number): void {
1869 this.getWorkerInfo(workerNodeKey
).ready
= false
1873 public hasWorkerNodeBackPressure (workerNodeKey
: number): boolean {
1875 this.opts
.enableTasksQueue
=== true &&
1876 this.workerNodes
[workerNodeKey
].hasBackPressure()
1880 private hasBackPressure (): boolean {
1882 this.opts
.enableTasksQueue
=== true &&
1883 this.workerNodes
.findIndex(
1884 workerNode
=> !workerNode
.hasBackPressure()
1890 * Executes the given task on the worker given its worker node key.
1892 * @param workerNodeKey - The worker node key.
1893 * @param task - The task to execute.
1895 private executeTask (workerNodeKey
: number, task
: Task
<Data
>): void {
1896 this.beforeTaskExecutionHook(workerNodeKey
, task
)
1897 this.sendToWorker(workerNodeKey
, task
, task
.transferList
)
1898 this.checkAndEmitTaskExecutionEvents()
1901 private enqueueTask (workerNodeKey
: number, task
: Task
<Data
>): number {
1902 const tasksQueueSize
= this.workerNodes
[workerNodeKey
].enqueueTask(task
)
1903 this.checkAndEmitTaskQueuingEvents()
1904 return tasksQueueSize
1907 private dequeueTask (workerNodeKey
: number): Task
<Data
> | undefined {
1908 return this.workerNodes
[workerNodeKey
].dequeueTask()
1911 private tasksQueueSize (workerNodeKey
: number): number {
1912 return this.workerNodes
[workerNodeKey
].tasksQueueSize()
1915 protected flushTasksQueue (workerNodeKey
: number): void {
1916 while (this.tasksQueueSize(workerNodeKey
) > 0) {
1919 this.dequeueTask(workerNodeKey
) as Task
<Data
>
1922 this.workerNodes
[workerNodeKey
].clearTasksQueue()
1925 private flushTasksQueues (): void {
1926 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
1927 this.flushTasksQueue(workerNodeKey
)