Merge dependabot/npm_and_yarn/types/node-20.12.10 into combined-prs-branch
[poolifier.git] / src / pools / abstract-pool.ts
1 import { AsyncResource } from 'node:async_hooks'
2 import { randomUUID } from 'node:crypto'
3 import { EventEmitterAsyncResource } from 'node:events'
4 import { performance } from 'node:perf_hooks'
5 import type { TransferListItem } from 'node:worker_threads'
6
7 import type {
8 MessageValue,
9 PromiseResponseWrapper,
10 Task,
11 TaskFunctionProperties
12 } from '../utility-types.js'
13 import {
14 average,
15 buildTaskFunctionProperties,
16 DEFAULT_TASK_NAME,
17 EMPTY_FUNCTION,
18 exponentialDelay,
19 isKillBehavior,
20 isPlainObject,
21 max,
22 median,
23 min,
24 round,
25 sleep
26 } from '../utils.js'
27 import type {
28 TaskFunction,
29 TaskFunctionObject
30 } from '../worker/task-functions.js'
31 import { KillBehaviors } from '../worker/worker-options.js'
32 import {
33 type IPool,
34 PoolEvents,
35 type PoolInfo,
36 type PoolOptions,
37 type PoolType,
38 PoolTypes,
39 type TasksQueueOptions
40 } from './pool.js'
41 import {
42 Measurements,
43 WorkerChoiceStrategies,
44 type WorkerChoiceStrategy,
45 type WorkerChoiceStrategyOptions
46 } from './selection-strategies/selection-strategies-types.js'
47 import { WorkerChoiceStrategiesContext } from './selection-strategies/worker-choice-strategies-context.js'
48 import {
49 checkFilePath,
50 checkValidPriority,
51 checkValidTasksQueueOptions,
52 checkValidWorkerChoiceStrategy,
53 getDefaultTasksQueueOptions,
54 updateEluWorkerUsage,
55 updateRunTimeWorkerUsage,
56 updateTaskStatisticsWorkerUsage,
57 updateWaitTimeWorkerUsage,
58 waitWorkerNodeEvents
59 } from './utils.js'
60 import { version } from './version.js'
61 import type {
62 IWorker,
63 IWorkerNode,
64 WorkerInfo,
65 WorkerNodeEventDetail,
66 WorkerType
67 } from './worker.js'
68 import { WorkerNode } from './worker-node.js'
69
70 /**
71 * Base class that implements some shared logic for all poolifier pools.
72 *
73 * @typeParam Worker - Type of worker which manages this pool.
74 * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
75 * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
76 */
77 export abstract class AbstractPool<
78 Worker extends IWorker,
79 Data = unknown,
80 Response = unknown
81 > implements IPool<Worker, Data, Response> {
82 /** @inheritDoc */
83 public readonly workerNodes: Array<IWorkerNode<Worker, Data>> = []
84
85 /** @inheritDoc */
86 public emitter?: EventEmitterAsyncResource
87
88 /**
89 * The task execution response promise map:
90 * - `key`: The message id of each submitted task.
91 * - `value`: An object that contains task's worker node key, execution response promise resolve and reject callbacks, async resource.
92 *
93 * When we receive a message from the worker, we get a map entry with the promise resolve/reject bound to the message id.
94 */
95 protected promiseResponseMap: Map<string, PromiseResponseWrapper<Response>> =
96 new Map<string, PromiseResponseWrapper<Response>>()
97
98 /**
99 * Worker choice strategies context referencing worker choice algorithms implementation.
100 */
101 protected workerChoiceStrategiesContext?: WorkerChoiceStrategiesContext<
102 Worker,
103 Data,
104 Response
105 >
106
107 /**
108 * The task functions added at runtime map:
109 * - `key`: The task function name.
110 * - `value`: The task function object.
111 */
112 private readonly taskFunctions: Map<
113 string,
114 TaskFunctionObject<Data, Response>
115 >
116
117 /**
118 * Whether the pool is started or not.
119 */
120 private started: boolean
121 /**
122 * Whether the pool is starting or not.
123 */
124 private starting: boolean
125 /**
126 * Whether the pool is destroying or not.
127 */
128 private destroying: boolean
129 /**
130 * Whether the minimum number of workers is starting or not.
131 */
132 private startingMinimumNumberOfWorkers: boolean
133 /**
134 * Whether the pool ready event has been emitted or not.
135 */
136 private readyEventEmitted: boolean
137 /**
138 * The start timestamp of the pool.
139 */
140 private startTimestamp?: number
141
142 /**
143 * Constructs a new poolifier pool.
144 *
145 * @param minimumNumberOfWorkers - Minimum number of workers that this pool manages.
146 * @param filePath - Path to the worker file.
147 * @param opts - Options for the pool.
148 * @param maximumNumberOfWorkers - Maximum number of workers that this pool manages.
149 */
150 public constructor (
151 protected readonly minimumNumberOfWorkers: number,
152 protected readonly filePath: string,
153 protected readonly opts: PoolOptions<Worker>,
154 protected readonly maximumNumberOfWorkers?: number
155 ) {
156 if (!this.isMain()) {
157 throw new Error(
158 'Cannot start a pool from a worker with the same type as the pool'
159 )
160 }
161 this.checkPoolType()
162 checkFilePath(this.filePath)
163 this.checkMinimumNumberOfWorkers(this.minimumNumberOfWorkers)
164 this.checkPoolOptions(this.opts)
165
166 this.chooseWorkerNode = this.chooseWorkerNode.bind(this)
167 this.executeTask = this.executeTask.bind(this)
168 this.enqueueTask = this.enqueueTask.bind(this)
169
170 if (this.opts.enableEvents === true) {
171 this.initEventEmitter()
172 }
173 this.workerChoiceStrategiesContext = new WorkerChoiceStrategiesContext<
174 Worker,
175 Data,
176 Response
177 >(
178 this,
179 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
180 [this.opts.workerChoiceStrategy!],
181 this.opts.workerChoiceStrategyOptions
182 )
183
184 this.setupHook()
185
186 this.taskFunctions = new Map<string, TaskFunctionObject<Data, Response>>()
187
188 this.started = false
189 this.starting = false
190 this.destroying = false
191 this.readyEventEmitted = false
192 this.startingMinimumNumberOfWorkers = false
193 if (this.opts.startWorkers === true) {
194 this.start()
195 }
196 }
197
198 private checkPoolType (): void {
199 if (this.type === PoolTypes.fixed && this.maximumNumberOfWorkers != null) {
200 throw new Error(
201 'Cannot instantiate a fixed pool with a maximum number of workers specified at initialization'
202 )
203 }
204 }
205
206 private checkMinimumNumberOfWorkers (
207 minimumNumberOfWorkers: number | undefined
208 ): void {
209 if (minimumNumberOfWorkers == null) {
210 throw new Error(
211 'Cannot instantiate a pool without specifying the number of workers'
212 )
213 } else if (!Number.isSafeInteger(minimumNumberOfWorkers)) {
214 throw new TypeError(
215 'Cannot instantiate a pool with a non safe integer number of workers'
216 )
217 } else if (minimumNumberOfWorkers < 0) {
218 throw new RangeError(
219 'Cannot instantiate a pool with a negative number of workers'
220 )
221 } else if (this.type === PoolTypes.fixed && minimumNumberOfWorkers === 0) {
222 throw new RangeError('Cannot instantiate a fixed pool with zero worker')
223 }
224 }
225
226 private checkPoolOptions (opts: PoolOptions<Worker>): void {
227 if (isPlainObject(opts)) {
228 this.opts.startWorkers = opts.startWorkers ?? true
229 checkValidWorkerChoiceStrategy(opts.workerChoiceStrategy)
230 this.opts.workerChoiceStrategy =
231 opts.workerChoiceStrategy ?? WorkerChoiceStrategies.ROUND_ROBIN
232 this.checkValidWorkerChoiceStrategyOptions(
233 opts.workerChoiceStrategyOptions
234 )
235 if (opts.workerChoiceStrategyOptions != null) {
236 this.opts.workerChoiceStrategyOptions = opts.workerChoiceStrategyOptions
237 }
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 checkValidTasksQueueOptions(opts.tasksQueueOptions)
243 this.opts.tasksQueueOptions = this.buildTasksQueueOptions(
244 opts.tasksQueueOptions
245 )
246 }
247 } else {
248 throw new TypeError('Invalid pool options: must be a plain object')
249 }
250 }
251
252 private checkValidWorkerChoiceStrategyOptions (
253 workerChoiceStrategyOptions: WorkerChoiceStrategyOptions | undefined
254 ): void {
255 if (
256 workerChoiceStrategyOptions != null &&
257 !isPlainObject(workerChoiceStrategyOptions)
258 ) {
259 throw new TypeError(
260 'Invalid worker choice strategy options: must be a plain object'
261 )
262 }
263 if (
264 workerChoiceStrategyOptions?.weights != null &&
265 Object.keys(workerChoiceStrategyOptions.weights).length !==
266 (this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers)
267 ) {
268 throw new Error(
269 'Invalid worker choice strategy options: must have a weight for each worker node'
270 )
271 }
272 if (
273 workerChoiceStrategyOptions?.measurement != null &&
274 !Object.values(Measurements).includes(
275 workerChoiceStrategyOptions.measurement
276 )
277 ) {
278 throw new Error(
279 `Invalid worker choice strategy options: invalid measurement '${workerChoiceStrategyOptions.measurement}'`
280 )
281 }
282 }
283
284 private initEventEmitter (): void {
285 this.emitter = new EventEmitterAsyncResource({
286 name: `poolifier:${this.type}-${this.worker}-pool`
287 })
288 }
289
290 /** @inheritDoc */
291 public get info (): PoolInfo {
292 return {
293 version,
294 type: this.type,
295 worker: this.worker,
296 started: this.started,
297 ready: this.ready,
298 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
299 defaultStrategy: this.opts.workerChoiceStrategy!,
300 strategyRetries: this.workerChoiceStrategiesContext?.retriesCount ?? 0,
301 minSize: this.minimumNumberOfWorkers,
302 maxSize: this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers,
303 ...(this.workerChoiceStrategiesContext?.getTaskStatisticsRequirements()
304 .runTime.aggregate === true &&
305 this.workerChoiceStrategiesContext.getTaskStatisticsRequirements()
306 .waitTime.aggregate && {
307 utilization: round(this.utilization)
308 }),
309 workerNodes: this.workerNodes.length,
310 idleWorkerNodes: this.workerNodes.reduce(
311 (accumulator, workerNode) =>
312 workerNode.usage.tasks.executing === 0
313 ? accumulator + 1
314 : accumulator,
315 0
316 ),
317 ...(this.opts.enableTasksQueue === true && {
318 stealingWorkerNodes: this.workerNodes.reduce(
319 (accumulator, workerNode) =>
320 workerNode.info.stealing ? accumulator + 1 : accumulator,
321 0
322 )
323 }),
324 busyWorkerNodes: this.workerNodes.reduce(
325 (accumulator, _workerNode, workerNodeKey) =>
326 this.isWorkerNodeBusy(workerNodeKey) ? accumulator + 1 : accumulator,
327 0
328 ),
329 executedTasks: this.workerNodes.reduce(
330 (accumulator, workerNode) =>
331 accumulator + workerNode.usage.tasks.executed,
332 0
333 ),
334 executingTasks: this.workerNodes.reduce(
335 (accumulator, workerNode) =>
336 accumulator + workerNode.usage.tasks.executing,
337 0
338 ),
339 ...(this.opts.enableTasksQueue === true && {
340 queuedTasks: this.workerNodes.reduce(
341 (accumulator, workerNode) =>
342 accumulator + workerNode.usage.tasks.queued,
343 0
344 )
345 }),
346 ...(this.opts.enableTasksQueue === true && {
347 maxQueuedTasks: this.workerNodes.reduce(
348 (accumulator, workerNode) =>
349 accumulator + (workerNode.usage.tasks.maxQueued ?? 0),
350 0
351 )
352 }),
353 ...(this.opts.enableTasksQueue === true && {
354 backPressure: this.hasBackPressure()
355 }),
356 ...(this.opts.enableTasksQueue === true && {
357 stolenTasks: this.workerNodes.reduce(
358 (accumulator, workerNode) =>
359 accumulator + workerNode.usage.tasks.stolen,
360 0
361 )
362 }),
363 failedTasks: this.workerNodes.reduce(
364 (accumulator, workerNode) =>
365 accumulator + workerNode.usage.tasks.failed,
366 0
367 ),
368 ...(this.workerChoiceStrategiesContext?.getTaskStatisticsRequirements()
369 .runTime.aggregate === true && {
370 runTime: {
371 minimum: round(
372 min(
373 ...this.workerNodes.map(
374 workerNode => workerNode.usage.runTime.minimum ?? Infinity
375 )
376 )
377 ),
378 maximum: round(
379 max(
380 ...this.workerNodes.map(
381 workerNode => workerNode.usage.runTime.maximum ?? -Infinity
382 )
383 )
384 ),
385 ...(this.workerChoiceStrategiesContext.getTaskStatisticsRequirements()
386 .runTime.average && {
387 average: round(
388 average(
389 this.workerNodes.reduce<number[]>(
390 (accumulator, workerNode) =>
391 accumulator.concat(workerNode.usage.runTime.history),
392 []
393 )
394 )
395 )
396 }),
397 ...(this.workerChoiceStrategiesContext.getTaskStatisticsRequirements()
398 .runTime.median && {
399 median: round(
400 median(
401 this.workerNodes.reduce<number[]>(
402 (accumulator, workerNode) =>
403 accumulator.concat(workerNode.usage.runTime.history),
404 []
405 )
406 )
407 )
408 })
409 }
410 }),
411 ...(this.workerChoiceStrategiesContext?.getTaskStatisticsRequirements()
412 .waitTime.aggregate === true && {
413 waitTime: {
414 minimum: round(
415 min(
416 ...this.workerNodes.map(
417 workerNode => workerNode.usage.waitTime.minimum ?? Infinity
418 )
419 )
420 ),
421 maximum: round(
422 max(
423 ...this.workerNodes.map(
424 workerNode => workerNode.usage.waitTime.maximum ?? -Infinity
425 )
426 )
427 ),
428 ...(this.workerChoiceStrategiesContext.getTaskStatisticsRequirements()
429 .waitTime.average && {
430 average: round(
431 average(
432 this.workerNodes.reduce<number[]>(
433 (accumulator, workerNode) =>
434 accumulator.concat(workerNode.usage.waitTime.history),
435 []
436 )
437 )
438 )
439 }),
440 ...(this.workerChoiceStrategiesContext.getTaskStatisticsRequirements()
441 .waitTime.median && {
442 median: round(
443 median(
444 this.workerNodes.reduce<number[]>(
445 (accumulator, workerNode) =>
446 accumulator.concat(workerNode.usage.waitTime.history),
447 []
448 )
449 )
450 )
451 })
452 }
453 })
454 }
455 }
456
457 /**
458 * The pool readiness boolean status.
459 */
460 private get ready (): boolean {
461 if (this.empty) {
462 return false
463 }
464 return (
465 this.workerNodes.reduce(
466 (accumulator, workerNode) =>
467 !workerNode.info.dynamic && workerNode.info.ready
468 ? accumulator + 1
469 : accumulator,
470 0
471 ) >= this.minimumNumberOfWorkers
472 )
473 }
474
475 /**
476 * The pool emptiness boolean status.
477 */
478 protected get empty (): boolean {
479 return this.minimumNumberOfWorkers === 0 && this.workerNodes.length === 0
480 }
481
482 /**
483 * The approximate pool utilization.
484 *
485 * @returns The pool utilization.
486 */
487 private get utilization (): number {
488 if (this.startTimestamp == null) {
489 return 0
490 }
491 const poolTimeCapacity =
492 (performance.now() - this.startTimestamp) *
493 (this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers)
494 const totalTasksRunTime = this.workerNodes.reduce(
495 (accumulator, workerNode) =>
496 accumulator + (workerNode.usage.runTime.aggregate ?? 0),
497 0
498 )
499 const totalTasksWaitTime = this.workerNodes.reduce(
500 (accumulator, workerNode) =>
501 accumulator + (workerNode.usage.waitTime.aggregate ?? 0),
502 0
503 )
504 return (totalTasksRunTime + totalTasksWaitTime) / poolTimeCapacity
505 }
506
507 /**
508 * The pool type.
509 *
510 * If it is `'dynamic'`, it provides the `max` property.
511 */
512 protected abstract get type (): PoolType
513
514 /**
515 * The worker type.
516 */
517 protected abstract get worker (): WorkerType
518
519 /**
520 * Checks if the worker id sent in the received message from a worker is valid.
521 *
522 * @param message - The received message.
523 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the worker id is invalid.
524 */
525 private checkMessageWorkerId (message: MessageValue<Data | Response>): void {
526 if (message.workerId == null) {
527 throw new Error('Worker message received without worker id')
528 } else if (this.getWorkerNodeKeyByWorkerId(message.workerId) === -1) {
529 throw new Error(
530 `Worker message received from unknown worker '${message.workerId}'`
531 )
532 }
533 }
534
535 /**
536 * Gets the worker node key given its worker id.
537 *
538 * @param workerId - The worker id.
539 * @returns The worker node key if the worker id is found in the pool worker nodes, `-1` otherwise.
540 */
541 private getWorkerNodeKeyByWorkerId (workerId: number | undefined): number {
542 return this.workerNodes.findIndex(
543 workerNode => workerNode.info.id === workerId
544 )
545 }
546
547 /** @inheritDoc */
548 public setWorkerChoiceStrategy (
549 workerChoiceStrategy: WorkerChoiceStrategy,
550 workerChoiceStrategyOptions?: WorkerChoiceStrategyOptions
551 ): void {
552 let requireSync = false
553 checkValidWorkerChoiceStrategy(workerChoiceStrategy)
554 if (workerChoiceStrategyOptions != null) {
555 requireSync = !this.setWorkerChoiceStrategyOptions(
556 workerChoiceStrategyOptions
557 )
558 }
559 if (workerChoiceStrategy !== this.opts.workerChoiceStrategy) {
560 this.opts.workerChoiceStrategy = workerChoiceStrategy
561 this.workerChoiceStrategiesContext?.setDefaultWorkerChoiceStrategy(
562 this.opts.workerChoiceStrategy,
563 this.opts.workerChoiceStrategyOptions
564 )
565 requireSync = true
566 }
567 if (requireSync) {
568 this.workerChoiceStrategiesContext?.syncWorkerChoiceStrategies(
569 this.getWorkerWorkerChoiceStrategies(),
570 this.opts.workerChoiceStrategyOptions
571 )
572 for (const workerNodeKey of this.workerNodes.keys()) {
573 this.sendStatisticsMessageToWorker(workerNodeKey)
574 }
575 }
576 }
577
578 /** @inheritDoc */
579 public setWorkerChoiceStrategyOptions (
580 workerChoiceStrategyOptions: WorkerChoiceStrategyOptions | undefined
581 ): boolean {
582 this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions)
583 if (workerChoiceStrategyOptions != null) {
584 this.opts.workerChoiceStrategyOptions = workerChoiceStrategyOptions
585 this.workerChoiceStrategiesContext?.setOptions(
586 this.opts.workerChoiceStrategyOptions
587 )
588 this.workerChoiceStrategiesContext?.syncWorkerChoiceStrategies(
589 this.getWorkerWorkerChoiceStrategies(),
590 this.opts.workerChoiceStrategyOptions
591 )
592 for (const workerNodeKey of this.workerNodes.keys()) {
593 this.sendStatisticsMessageToWorker(workerNodeKey)
594 }
595 return true
596 }
597 return false
598 }
599
600 /** @inheritDoc */
601 public enableTasksQueue (
602 enable: boolean,
603 tasksQueueOptions?: TasksQueueOptions
604 ): void {
605 if (this.opts.enableTasksQueue === true && !enable) {
606 this.unsetTaskStealing()
607 this.unsetTasksStealingOnBackPressure()
608 this.flushTasksQueues()
609 }
610 this.opts.enableTasksQueue = enable
611 this.setTasksQueueOptions(tasksQueueOptions)
612 }
613
614 /** @inheritDoc */
615 public setTasksQueueOptions (
616 tasksQueueOptions: TasksQueueOptions | undefined
617 ): void {
618 if (this.opts.enableTasksQueue === true) {
619 checkValidTasksQueueOptions(tasksQueueOptions)
620 this.opts.tasksQueueOptions =
621 this.buildTasksQueueOptions(tasksQueueOptions)
622 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
623 this.setTasksQueueSize(this.opts.tasksQueueOptions.size!)
624 if (this.opts.tasksQueueOptions.taskStealing === true) {
625 this.unsetTaskStealing()
626 this.setTaskStealing()
627 } else {
628 this.unsetTaskStealing()
629 }
630 if (this.opts.tasksQueueOptions.tasksStealingOnBackPressure === true) {
631 this.unsetTasksStealingOnBackPressure()
632 this.setTasksStealingOnBackPressure()
633 } else {
634 this.unsetTasksStealingOnBackPressure()
635 }
636 } else if (this.opts.tasksQueueOptions != null) {
637 delete this.opts.tasksQueueOptions
638 }
639 }
640
641 private buildTasksQueueOptions (
642 tasksQueueOptions: TasksQueueOptions | undefined
643 ): TasksQueueOptions {
644 return {
645 ...getDefaultTasksQueueOptions(
646 this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers
647 ),
648 ...tasksQueueOptions
649 }
650 }
651
652 private setTasksQueueSize (size: number): void {
653 for (const workerNode of this.workerNodes) {
654 workerNode.tasksQueueBackPressureSize = size
655 }
656 }
657
658 private setTaskStealing (): void {
659 for (const workerNodeKey of this.workerNodes.keys()) {
660 this.workerNodes[workerNodeKey].on('idle', this.handleWorkerNodeIdleEvent)
661 }
662 }
663
664 private unsetTaskStealing (): void {
665 for (const workerNodeKey of this.workerNodes.keys()) {
666 this.workerNodes[workerNodeKey].off(
667 'idle',
668 this.handleWorkerNodeIdleEvent
669 )
670 }
671 }
672
673 private setTasksStealingOnBackPressure (): void {
674 for (const workerNodeKey of this.workerNodes.keys()) {
675 this.workerNodes[workerNodeKey].on(
676 'backPressure',
677 this.handleWorkerNodeBackPressureEvent
678 )
679 }
680 }
681
682 private unsetTasksStealingOnBackPressure (): void {
683 for (const workerNodeKey of this.workerNodes.keys()) {
684 this.workerNodes[workerNodeKey].off(
685 'backPressure',
686 this.handleWorkerNodeBackPressureEvent
687 )
688 }
689 }
690
691 /**
692 * Whether the pool is full or not.
693 *
694 * The pool filling boolean status.
695 */
696 protected get full (): boolean {
697 return (
698 this.workerNodes.length >=
699 (this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers)
700 )
701 }
702
703 /**
704 * Whether the pool is busy or not.
705 *
706 * The pool busyness boolean status.
707 */
708 protected abstract get busy (): boolean
709
710 /**
711 * Whether worker nodes are executing concurrently their tasks quota or not.
712 *
713 * @returns Worker nodes busyness boolean status.
714 */
715 protected internalBusy (): boolean {
716 if (this.opts.enableTasksQueue === true) {
717 return (
718 this.workerNodes.findIndex(
719 workerNode =>
720 workerNode.info.ready &&
721 workerNode.usage.tasks.executing <
722 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
723 this.opts.tasksQueueOptions!.concurrency!
724 ) === -1
725 )
726 }
727 return (
728 this.workerNodes.findIndex(
729 workerNode =>
730 workerNode.info.ready && workerNode.usage.tasks.executing === 0
731 ) === -1
732 )
733 }
734
735 private isWorkerNodeBusy (workerNodeKey: number): boolean {
736 if (this.opts.enableTasksQueue === true) {
737 return (
738 this.workerNodes[workerNodeKey].usage.tasks.executing >=
739 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
740 this.opts.tasksQueueOptions!.concurrency!
741 )
742 }
743 return this.workerNodes[workerNodeKey].usage.tasks.executing > 0
744 }
745
746 private async sendTaskFunctionOperationToWorker (
747 workerNodeKey: number,
748 message: MessageValue<Data>
749 ): Promise<boolean> {
750 return await new Promise<boolean>((resolve, reject) => {
751 const taskFunctionOperationListener = (
752 message: MessageValue<Response>
753 ): void => {
754 this.checkMessageWorkerId(message)
755 const workerId = this.getWorkerInfo(workerNodeKey)?.id
756 if (
757 message.taskFunctionOperationStatus != null &&
758 message.workerId === workerId
759 ) {
760 if (message.taskFunctionOperationStatus) {
761 resolve(true)
762 } else {
763 reject(
764 new Error(
765 `Task function operation '${message.taskFunctionOperation}' failed on worker ${message.workerId} with error: '${message.workerError?.message}'`
766 )
767 )
768 }
769 this.deregisterWorkerMessageListener(
770 this.getWorkerNodeKeyByWorkerId(message.workerId),
771 taskFunctionOperationListener
772 )
773 }
774 }
775 this.registerWorkerMessageListener(
776 workerNodeKey,
777 taskFunctionOperationListener
778 )
779 this.sendToWorker(workerNodeKey, message)
780 })
781 }
782
783 private async sendTaskFunctionOperationToWorkers (
784 message: MessageValue<Data>
785 ): Promise<boolean> {
786 return await new Promise<boolean>((resolve, reject) => {
787 const responsesReceived = new Array<MessageValue<Response>>()
788 const taskFunctionOperationsListener = (
789 message: MessageValue<Response>
790 ): void => {
791 this.checkMessageWorkerId(message)
792 if (message.taskFunctionOperationStatus != null) {
793 responsesReceived.push(message)
794 if (responsesReceived.length === this.workerNodes.length) {
795 if (
796 responsesReceived.every(
797 message => message.taskFunctionOperationStatus === true
798 )
799 ) {
800 resolve(true)
801 } else if (
802 responsesReceived.some(
803 message => message.taskFunctionOperationStatus === false
804 )
805 ) {
806 const errorResponse = responsesReceived.find(
807 response => response.taskFunctionOperationStatus === false
808 )
809 reject(
810 new Error(
811 `Task function operation '${
812 message.taskFunctionOperation as string
813 }' failed on worker ${errorResponse?.workerId} with error: '${
814 errorResponse?.workerError?.message
815 }'`
816 )
817 )
818 }
819 this.deregisterWorkerMessageListener(
820 this.getWorkerNodeKeyByWorkerId(message.workerId),
821 taskFunctionOperationsListener
822 )
823 }
824 }
825 }
826 for (const workerNodeKey of this.workerNodes.keys()) {
827 this.registerWorkerMessageListener(
828 workerNodeKey,
829 taskFunctionOperationsListener
830 )
831 this.sendToWorker(workerNodeKey, message)
832 }
833 })
834 }
835
836 /** @inheritDoc */
837 public hasTaskFunction (name: string): boolean {
838 return this.listTaskFunctionsProperties().some(
839 taskFunctionProperties => taskFunctionProperties.name === name
840 )
841 }
842
843 /** @inheritDoc */
844 public async addTaskFunction (
845 name: string,
846 fn: TaskFunction<Data, Response> | TaskFunctionObject<Data, Response>
847 ): Promise<boolean> {
848 if (typeof name !== 'string') {
849 throw new TypeError('name argument must be a string')
850 }
851 if (typeof name === 'string' && name.trim().length === 0) {
852 throw new TypeError('name argument must not be an empty string')
853 }
854 if (typeof fn === 'function') {
855 fn = { taskFunction: fn } satisfies TaskFunctionObject<Data, Response>
856 }
857 if (typeof fn.taskFunction !== 'function') {
858 throw new TypeError('taskFunction property must be a function')
859 }
860 checkValidPriority(fn.priority)
861 checkValidWorkerChoiceStrategy(fn.strategy)
862 const opResult = await this.sendTaskFunctionOperationToWorkers({
863 taskFunctionOperation: 'add',
864 taskFunctionProperties: buildTaskFunctionProperties(name, fn),
865 taskFunction: fn.taskFunction.toString()
866 })
867 this.taskFunctions.set(name, fn)
868 this.workerChoiceStrategiesContext?.syncWorkerChoiceStrategies(
869 this.getWorkerWorkerChoiceStrategies()
870 )
871 for (const workerNodeKey of this.workerNodes.keys()) {
872 this.sendStatisticsMessageToWorker(workerNodeKey)
873 }
874 return opResult
875 }
876
877 /** @inheritDoc */
878 public async removeTaskFunction (name: string): Promise<boolean> {
879 if (!this.taskFunctions.has(name)) {
880 throw new Error(
881 'Cannot remove a task function not handled on the pool side'
882 )
883 }
884 const opResult = await this.sendTaskFunctionOperationToWorkers({
885 taskFunctionOperation: 'remove',
886 taskFunctionProperties: buildTaskFunctionProperties(
887 name,
888 this.taskFunctions.get(name)
889 )
890 })
891 for (const workerNode of this.workerNodes) {
892 workerNode.deleteTaskFunctionWorkerUsage(name)
893 }
894 this.taskFunctions.delete(name)
895 this.workerChoiceStrategiesContext?.syncWorkerChoiceStrategies(
896 this.getWorkerWorkerChoiceStrategies()
897 )
898 for (const workerNodeKey of this.workerNodes.keys()) {
899 this.sendStatisticsMessageToWorker(workerNodeKey)
900 }
901 return opResult
902 }
903
904 /** @inheritDoc */
905 public listTaskFunctionsProperties (): TaskFunctionProperties[] {
906 for (const workerNode of this.workerNodes) {
907 if (
908 Array.isArray(workerNode.info.taskFunctionsProperties) &&
909 workerNode.info.taskFunctionsProperties.length > 0
910 ) {
911 return workerNode.info.taskFunctionsProperties
912 }
913 }
914 return []
915 }
916
917 /**
918 * Gets task function strategy, if any.
919 *
920 * @param name - The task function name.
921 * @returns The task function worker choice strategy if the task function worker choice strategy is defined, `undefined` otherwise.
922 */
923 private readonly getTaskFunctionWorkerWorkerChoiceStrategy = (
924 name?: string
925 ): WorkerChoiceStrategy | undefined => {
926 if (name != null) {
927 return this.listTaskFunctionsProperties().find(
928 (taskFunctionProperties: TaskFunctionProperties) =>
929 taskFunctionProperties.name === name
930 )?.strategy
931 }
932 }
933
934 /**
935 * Gets worker node task function priority, if any.
936 *
937 * @param workerNodeKey - The worker node key.
938 * @param name - The task function name.
939 * @returns The task function worker choice priority if the task function worker choice priority is defined, `undefined` otherwise.
940 */
941 private readonly getWorkerNodeTaskFunctionPriority = (
942 workerNodeKey: number,
943 name?: string
944 ): number | undefined => {
945 if (name != null) {
946 return this.getWorkerInfo(workerNodeKey)?.taskFunctionsProperties?.find(
947 (taskFunctionProperties: TaskFunctionProperties) =>
948 taskFunctionProperties.name === name
949 )?.priority
950 }
951 }
952
953 /**
954 * Gets the worker choice strategies registered in this pool.
955 *
956 * @returns The worker choice strategies.
957 */
958 private readonly getWorkerWorkerChoiceStrategies =
959 (): Set<WorkerChoiceStrategy> => {
960 return new Set([
961 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
962 this.opts.workerChoiceStrategy!,
963 ...(this.listTaskFunctionsProperties()
964 .map(
965 (taskFunctionProperties: TaskFunctionProperties) =>
966 taskFunctionProperties.strategy
967 )
968 .filter(
969 (strategy: WorkerChoiceStrategy | undefined) => strategy != null
970 ) as WorkerChoiceStrategy[])
971 ])
972 }
973
974 /** @inheritDoc */
975 public async setDefaultTaskFunction (name: string): Promise<boolean> {
976 return await this.sendTaskFunctionOperationToWorkers({
977 taskFunctionOperation: 'default',
978 taskFunctionProperties: buildTaskFunctionProperties(
979 name,
980 this.taskFunctions.get(name)
981 )
982 })
983 }
984
985 private shallExecuteTask (workerNodeKey: number): boolean {
986 return (
987 this.tasksQueueSize(workerNodeKey) === 0 &&
988 this.workerNodes[workerNodeKey].usage.tasks.executing <
989 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
990 this.opts.tasksQueueOptions!.concurrency!
991 )
992 }
993
994 /** @inheritDoc */
995 public async execute (
996 data?: Data,
997 name?: string,
998 transferList?: readonly TransferListItem[]
999 ): Promise<Response> {
1000 return await new Promise<Response>((resolve, reject) => {
1001 if (!this.started) {
1002 reject(new Error('Cannot execute a task on not started pool'))
1003 return
1004 }
1005 if (this.destroying) {
1006 reject(new Error('Cannot execute a task on destroying pool'))
1007 return
1008 }
1009 if (name != null && typeof name !== 'string') {
1010 reject(new TypeError('name argument must be a string'))
1011 return
1012 }
1013 if (
1014 name != null &&
1015 typeof name === 'string' &&
1016 name.trim().length === 0
1017 ) {
1018 reject(new TypeError('name argument must not be an empty string'))
1019 return
1020 }
1021 if (transferList != null && !Array.isArray(transferList)) {
1022 reject(new TypeError('transferList argument must be an array'))
1023 return
1024 }
1025 const timestamp = performance.now()
1026 const taskFunctionStrategy =
1027 this.getTaskFunctionWorkerWorkerChoiceStrategy(name)
1028 const workerNodeKey = this.chooseWorkerNode(taskFunctionStrategy)
1029 const task: Task<Data> = {
1030 name: name ?? DEFAULT_TASK_NAME,
1031 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
1032 data: data ?? ({} as Data),
1033 priority: this.getWorkerNodeTaskFunctionPriority(workerNodeKey, name),
1034 strategy: taskFunctionStrategy,
1035 transferList,
1036 timestamp,
1037 taskId: randomUUID()
1038 }
1039 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1040 this.promiseResponseMap.set(task.taskId!, {
1041 resolve,
1042 reject,
1043 workerNodeKey,
1044 ...(this.emitter != null && {
1045 asyncResource: new AsyncResource('poolifier:task', {
1046 triggerAsyncId: this.emitter.asyncId,
1047 requireManualDestroy: true
1048 })
1049 })
1050 })
1051 if (
1052 this.opts.enableTasksQueue === false ||
1053 (this.opts.enableTasksQueue === true &&
1054 this.shallExecuteTask(workerNodeKey))
1055 ) {
1056 this.executeTask(workerNodeKey, task)
1057 } else {
1058 this.enqueueTask(workerNodeKey, task)
1059 }
1060 })
1061 }
1062
1063 /**
1064 * Starts the minimum number of workers.
1065 */
1066 private startMinimumNumberOfWorkers (initWorkerNodeUsage = false): void {
1067 this.startingMinimumNumberOfWorkers = true
1068 while (
1069 this.workerNodes.reduce(
1070 (accumulator, workerNode) =>
1071 !workerNode.info.dynamic ? accumulator + 1 : accumulator,
1072 0
1073 ) < this.minimumNumberOfWorkers
1074 ) {
1075 const workerNodeKey = this.createAndSetupWorkerNode()
1076 initWorkerNodeUsage &&
1077 this.initWorkerNodeUsage(this.workerNodes[workerNodeKey])
1078 }
1079 this.startingMinimumNumberOfWorkers = false
1080 }
1081
1082 /** @inheritdoc */
1083 public start (): void {
1084 if (this.started) {
1085 throw new Error('Cannot start an already started pool')
1086 }
1087 if (this.starting) {
1088 throw new Error('Cannot start an already starting pool')
1089 }
1090 if (this.destroying) {
1091 throw new Error('Cannot start a destroying pool')
1092 }
1093 this.starting = true
1094 this.startMinimumNumberOfWorkers()
1095 this.startTimestamp = performance.now()
1096 this.starting = false
1097 this.started = true
1098 }
1099
1100 /** @inheritDoc */
1101 public async destroy (): Promise<void> {
1102 if (!this.started) {
1103 throw new Error('Cannot destroy an already destroyed pool')
1104 }
1105 if (this.starting) {
1106 throw new Error('Cannot destroy an starting pool')
1107 }
1108 if (this.destroying) {
1109 throw new Error('Cannot destroy an already destroying pool')
1110 }
1111 this.destroying = true
1112 await Promise.all(
1113 this.workerNodes.map(async (_workerNode, workerNodeKey) => {
1114 await this.destroyWorkerNode(workerNodeKey)
1115 })
1116 )
1117 this.emitter?.emit(PoolEvents.destroy, this.info)
1118 this.emitter?.emitDestroy()
1119 this.readyEventEmitted = false
1120 delete this.startTimestamp
1121 this.destroying = false
1122 this.started = false
1123 }
1124
1125 private async sendKillMessageToWorker (workerNodeKey: number): Promise<void> {
1126 await new Promise<void>((resolve, reject) => {
1127 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1128 if (this.workerNodes[workerNodeKey] == null) {
1129 resolve()
1130 return
1131 }
1132 const killMessageListener = (message: MessageValue<Response>): void => {
1133 this.checkMessageWorkerId(message)
1134 if (message.kill === 'success') {
1135 resolve()
1136 } else if (message.kill === 'failure') {
1137 reject(
1138 new Error(
1139 `Kill message handling failed on worker ${message.workerId}`
1140 )
1141 )
1142 }
1143 }
1144 // FIXME: should be registered only once
1145 this.registerWorkerMessageListener(workerNodeKey, killMessageListener)
1146 this.sendToWorker(workerNodeKey, { kill: true })
1147 })
1148 }
1149
1150 /**
1151 * Terminates the worker node given its worker node key.
1152 *
1153 * @param workerNodeKey - The worker node key.
1154 */
1155 protected async destroyWorkerNode (workerNodeKey: number): Promise<void> {
1156 this.flagWorkerNodeAsNotReady(workerNodeKey)
1157 const flushedTasks = this.flushTasksQueue(workerNodeKey)
1158 const workerNode = this.workerNodes[workerNodeKey]
1159 await waitWorkerNodeEvents(
1160 workerNode,
1161 'taskFinished',
1162 flushedTasks,
1163 this.opts.tasksQueueOptions?.tasksFinishedTimeout ??
1164 getDefaultTasksQueueOptions(
1165 this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers
1166 ).tasksFinishedTimeout
1167 )
1168 await this.sendKillMessageToWorker(workerNodeKey)
1169 await workerNode.terminate()
1170 }
1171
1172 /**
1173 * Setup hook to execute code before worker nodes are created in the abstract constructor.
1174 * Can be overridden.
1175 *
1176 * @virtual
1177 */
1178 protected setupHook (): void {
1179 /* Intentionally empty */
1180 }
1181
1182 /**
1183 * Returns whether the worker is the main worker or not.
1184 *
1185 * @returns `true` if the worker is the main worker, `false` otherwise.
1186 */
1187 protected abstract isMain (): boolean
1188
1189 /**
1190 * Hook executed before the worker task execution.
1191 * Can be overridden.
1192 *
1193 * @param workerNodeKey - The worker node key.
1194 * @param task - The task to execute.
1195 */
1196 protected beforeTaskExecutionHook (
1197 workerNodeKey: number,
1198 task: Task<Data>
1199 ): void {
1200 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1201 if (this.workerNodes[workerNodeKey]?.usage != null) {
1202 const workerUsage = this.workerNodes[workerNodeKey].usage
1203 ++workerUsage.tasks.executing
1204 updateWaitTimeWorkerUsage(
1205 this.workerChoiceStrategiesContext,
1206 workerUsage,
1207 task
1208 )
1209 }
1210 if (
1211 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) &&
1212 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1213 this.workerNodes[workerNodeKey].getTaskFunctionWorkerUsage(task.name!) !=
1214 null
1215 ) {
1216 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1217 const taskFunctionWorkerUsage = this.workerNodes[
1218 workerNodeKey
1219 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1220 ].getTaskFunctionWorkerUsage(task.name!)!
1221 ++taskFunctionWorkerUsage.tasks.executing
1222 updateWaitTimeWorkerUsage(
1223 this.workerChoiceStrategiesContext,
1224 taskFunctionWorkerUsage,
1225 task
1226 )
1227 }
1228 }
1229
1230 /**
1231 * Hook executed after the worker task execution.
1232 * Can be overridden.
1233 *
1234 * @param workerNodeKey - The worker node key.
1235 * @param message - The received message.
1236 */
1237 protected afterTaskExecutionHook (
1238 workerNodeKey: number,
1239 message: MessageValue<Response>
1240 ): void {
1241 let needWorkerChoiceStrategiesUpdate = false
1242 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1243 if (this.workerNodes[workerNodeKey]?.usage != null) {
1244 const workerUsage = this.workerNodes[workerNodeKey].usage
1245 updateTaskStatisticsWorkerUsage(workerUsage, message)
1246 updateRunTimeWorkerUsage(
1247 this.workerChoiceStrategiesContext,
1248 workerUsage,
1249 message
1250 )
1251 updateEluWorkerUsage(
1252 this.workerChoiceStrategiesContext,
1253 workerUsage,
1254 message
1255 )
1256 needWorkerChoiceStrategiesUpdate = true
1257 }
1258 if (
1259 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) &&
1260 this.workerNodes[workerNodeKey].getTaskFunctionWorkerUsage(
1261 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1262 message.taskPerformance!.name
1263 ) != null
1264 ) {
1265 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1266 const taskFunctionWorkerUsage = this.workerNodes[
1267 workerNodeKey
1268 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1269 ].getTaskFunctionWorkerUsage(message.taskPerformance!.name)!
1270 updateTaskStatisticsWorkerUsage(taskFunctionWorkerUsage, message)
1271 updateRunTimeWorkerUsage(
1272 this.workerChoiceStrategiesContext,
1273 taskFunctionWorkerUsage,
1274 message
1275 )
1276 updateEluWorkerUsage(
1277 this.workerChoiceStrategiesContext,
1278 taskFunctionWorkerUsage,
1279 message
1280 )
1281 needWorkerChoiceStrategiesUpdate = true
1282 }
1283 if (needWorkerChoiceStrategiesUpdate) {
1284 this.workerChoiceStrategiesContext?.update(workerNodeKey)
1285 }
1286 }
1287
1288 /**
1289 * Whether the worker node shall update its task function worker usage or not.
1290 *
1291 * @param workerNodeKey - The worker node key.
1292 * @returns `true` if the worker node shall update its task function worker usage, `false` otherwise.
1293 */
1294 private shallUpdateTaskFunctionWorkerUsage (workerNodeKey: number): boolean {
1295 const workerInfo = this.getWorkerInfo(workerNodeKey)
1296 return (
1297 workerInfo != null &&
1298 Array.isArray(workerInfo.taskFunctionsProperties) &&
1299 workerInfo.taskFunctionsProperties.length > 2
1300 )
1301 }
1302
1303 /**
1304 * Chooses a worker node for the next task given the worker choice strategy.
1305 *
1306 * @param workerChoiceStrategy - The worker choice strategy.
1307 * @returns The chosen worker node key
1308 */
1309 private chooseWorkerNode (
1310 workerChoiceStrategy?: WorkerChoiceStrategy
1311 ): number {
1312 if (this.shallCreateDynamicWorker()) {
1313 const workerNodeKey = this.createAndSetupDynamicWorkerNode()
1314 if (
1315 this.workerChoiceStrategiesContext?.getPolicy().dynamicWorkerUsage ===
1316 true
1317 ) {
1318 return workerNodeKey
1319 }
1320 }
1321 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1322 return this.workerChoiceStrategiesContext!.execute(workerChoiceStrategy)
1323 }
1324
1325 /**
1326 * Conditions for dynamic worker creation.
1327 *
1328 * @returns Whether to create a dynamic worker or not.
1329 */
1330 protected abstract shallCreateDynamicWorker (): boolean
1331
1332 /**
1333 * Sends a message to worker given its worker node key.
1334 *
1335 * @param workerNodeKey - The worker node key.
1336 * @param message - The message.
1337 * @param transferList - The optional array of transferable objects.
1338 */
1339 protected abstract sendToWorker (
1340 workerNodeKey: number,
1341 message: MessageValue<Data>,
1342 transferList?: readonly TransferListItem[]
1343 ): void
1344
1345 /**
1346 * Initializes the worker node usage with sensible default values gathered during runtime.
1347 *
1348 * @param workerNode - The worker node.
1349 */
1350 private initWorkerNodeUsage (workerNode: IWorkerNode<Worker, Data>): void {
1351 if (
1352 this.workerChoiceStrategiesContext?.getTaskStatisticsRequirements()
1353 .runTime.aggregate === true
1354 ) {
1355 workerNode.usage.runTime.aggregate = min(
1356 ...this.workerNodes.map(
1357 workerNode => workerNode.usage.runTime.aggregate ?? Infinity
1358 )
1359 )
1360 }
1361 if (
1362 this.workerChoiceStrategiesContext?.getTaskStatisticsRequirements()
1363 .waitTime.aggregate === true
1364 ) {
1365 workerNode.usage.waitTime.aggregate = min(
1366 ...this.workerNodes.map(
1367 workerNode => workerNode.usage.waitTime.aggregate ?? Infinity
1368 )
1369 )
1370 }
1371 if (
1372 this.workerChoiceStrategiesContext?.getTaskStatisticsRequirements().elu
1373 .aggregate === true
1374 ) {
1375 workerNode.usage.elu.active.aggregate = min(
1376 ...this.workerNodes.map(
1377 workerNode => workerNode.usage.elu.active.aggregate ?? Infinity
1378 )
1379 )
1380 }
1381 }
1382
1383 /**
1384 * Creates a new, completely set up worker node.
1385 *
1386 * @returns New, completely set up worker node key.
1387 */
1388 protected createAndSetupWorkerNode (): number {
1389 const workerNode = this.createWorkerNode()
1390 workerNode.registerWorkerEventHandler(
1391 'online',
1392 this.opts.onlineHandler ?? EMPTY_FUNCTION
1393 )
1394 workerNode.registerWorkerEventHandler(
1395 'message',
1396 this.opts.messageHandler ?? EMPTY_FUNCTION
1397 )
1398 workerNode.registerWorkerEventHandler(
1399 'error',
1400 this.opts.errorHandler ?? EMPTY_FUNCTION
1401 )
1402 workerNode.registerOnceWorkerEventHandler('error', (error: Error) => {
1403 workerNode.info.ready = false
1404 this.emitter?.emit(PoolEvents.error, error)
1405 if (
1406 this.started &&
1407 !this.destroying &&
1408 this.opts.restartWorkerOnError === true
1409 ) {
1410 if (workerNode.info.dynamic) {
1411 this.createAndSetupDynamicWorkerNode()
1412 } else if (!this.startingMinimumNumberOfWorkers) {
1413 this.startMinimumNumberOfWorkers(true)
1414 }
1415 }
1416 if (
1417 this.started &&
1418 !this.destroying &&
1419 this.opts.enableTasksQueue === true
1420 ) {
1421 this.redistributeQueuedTasks(this.workerNodes.indexOf(workerNode))
1422 }
1423 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1424 workerNode?.terminate().catch((error: unknown) => {
1425 this.emitter?.emit(PoolEvents.error, error)
1426 })
1427 })
1428 workerNode.registerWorkerEventHandler(
1429 'exit',
1430 this.opts.exitHandler ?? EMPTY_FUNCTION
1431 )
1432 workerNode.registerOnceWorkerEventHandler('exit', () => {
1433 this.removeWorkerNode(workerNode)
1434 if (
1435 this.started &&
1436 !this.startingMinimumNumberOfWorkers &&
1437 !this.destroying
1438 ) {
1439 this.startMinimumNumberOfWorkers(true)
1440 }
1441 })
1442 const workerNodeKey = this.addWorkerNode(workerNode)
1443 this.afterWorkerNodeSetup(workerNodeKey)
1444 return workerNodeKey
1445 }
1446
1447 /**
1448 * Creates a new, completely set up dynamic worker node.
1449 *
1450 * @returns New, completely set up dynamic worker node key.
1451 */
1452 protected createAndSetupDynamicWorkerNode (): number {
1453 const workerNodeKey = this.createAndSetupWorkerNode()
1454 this.registerWorkerMessageListener(workerNodeKey, message => {
1455 this.checkMessageWorkerId(message)
1456 const localWorkerNodeKey = this.getWorkerNodeKeyByWorkerId(
1457 message.workerId
1458 )
1459 const workerUsage = this.workerNodes[localWorkerNodeKey]?.usage
1460 // Kill message received from worker
1461 if (
1462 isKillBehavior(KillBehaviors.HARD, message.kill) ||
1463 (isKillBehavior(KillBehaviors.SOFT, message.kill) &&
1464 ((this.opts.enableTasksQueue === false &&
1465 workerUsage.tasks.executing === 0) ||
1466 (this.opts.enableTasksQueue === true &&
1467 workerUsage.tasks.executing === 0 &&
1468 this.tasksQueueSize(localWorkerNodeKey) === 0)))
1469 ) {
1470 // Flag the worker node as not ready immediately
1471 this.flagWorkerNodeAsNotReady(localWorkerNodeKey)
1472 this.destroyWorkerNode(localWorkerNodeKey).catch((error: unknown) => {
1473 this.emitter?.emit(PoolEvents.error, error)
1474 })
1475 }
1476 })
1477 this.sendToWorker(workerNodeKey, {
1478 checkActive: true
1479 })
1480 if (this.taskFunctions.size > 0) {
1481 for (const [taskFunctionName, taskFunctionObject] of this.taskFunctions) {
1482 this.sendTaskFunctionOperationToWorker(workerNodeKey, {
1483 taskFunctionOperation: 'add',
1484 taskFunctionProperties: buildTaskFunctionProperties(
1485 taskFunctionName,
1486 taskFunctionObject
1487 ),
1488 taskFunction: taskFunctionObject.taskFunction.toString()
1489 }).catch((error: unknown) => {
1490 this.emitter?.emit(PoolEvents.error, error)
1491 })
1492 }
1493 }
1494 const workerNode = this.workerNodes[workerNodeKey]
1495 workerNode.info.dynamic = true
1496 if (
1497 this.workerChoiceStrategiesContext?.getPolicy().dynamicWorkerReady ===
1498 true ||
1499 this.workerChoiceStrategiesContext?.getPolicy().dynamicWorkerUsage ===
1500 true
1501 ) {
1502 workerNode.info.ready = true
1503 }
1504 this.initWorkerNodeUsage(workerNode)
1505 this.checkAndEmitDynamicWorkerCreationEvents()
1506 return workerNodeKey
1507 }
1508
1509 /**
1510 * Registers a listener callback on the worker given its worker node key.
1511 *
1512 * @param workerNodeKey - The worker node key.
1513 * @param listener - The message listener callback.
1514 */
1515 protected abstract registerWorkerMessageListener<
1516 Message extends Data | Response
1517 >(
1518 workerNodeKey: number,
1519 listener: (message: MessageValue<Message>) => void
1520 ): void
1521
1522 /**
1523 * Registers once a listener callback on the worker given its worker node key.
1524 *
1525 * @param workerNodeKey - The worker node key.
1526 * @param listener - The message listener callback.
1527 */
1528 protected abstract registerOnceWorkerMessageListener<
1529 Message extends Data | Response
1530 >(
1531 workerNodeKey: number,
1532 listener: (message: MessageValue<Message>) => void
1533 ): void
1534
1535 /**
1536 * Deregisters a listener callback on the worker given its worker node key.
1537 *
1538 * @param workerNodeKey - The worker node key.
1539 * @param listener - The message listener callback.
1540 */
1541 protected abstract deregisterWorkerMessageListener<
1542 Message extends Data | Response
1543 >(
1544 workerNodeKey: number,
1545 listener: (message: MessageValue<Message>) => void
1546 ): void
1547
1548 /**
1549 * Method hooked up after a worker node has been newly created.
1550 * Can be overridden.
1551 *
1552 * @param workerNodeKey - The newly created worker node key.
1553 */
1554 protected afterWorkerNodeSetup (workerNodeKey: number): void {
1555 // Listen to worker messages.
1556 this.registerWorkerMessageListener(
1557 workerNodeKey,
1558 this.workerMessageListener
1559 )
1560 // Send the startup message to worker.
1561 this.sendStartupMessageToWorker(workerNodeKey)
1562 // Send the statistics message to worker.
1563 this.sendStatisticsMessageToWorker(workerNodeKey)
1564 if (this.opts.enableTasksQueue === true) {
1565 if (this.opts.tasksQueueOptions?.taskStealing === true) {
1566 this.workerNodes[workerNodeKey].on(
1567 'idle',
1568 this.handleWorkerNodeIdleEvent
1569 )
1570 }
1571 if (this.opts.tasksQueueOptions?.tasksStealingOnBackPressure === true) {
1572 this.workerNodes[workerNodeKey].on(
1573 'backPressure',
1574 this.handleWorkerNodeBackPressureEvent
1575 )
1576 }
1577 }
1578 }
1579
1580 /**
1581 * Sends the startup message to worker given its worker node key.
1582 *
1583 * @param workerNodeKey - The worker node key.
1584 */
1585 protected abstract sendStartupMessageToWorker (workerNodeKey: number): void
1586
1587 /**
1588 * Sends the statistics message to worker given its worker node key.
1589 *
1590 * @param workerNodeKey - The worker node key.
1591 */
1592 private sendStatisticsMessageToWorker (workerNodeKey: number): void {
1593 this.sendToWorker(workerNodeKey, {
1594 statistics: {
1595 runTime:
1596 this.workerChoiceStrategiesContext?.getTaskStatisticsRequirements()
1597 .runTime.aggregate ?? false,
1598 elu:
1599 this.workerChoiceStrategiesContext?.getTaskStatisticsRequirements()
1600 .elu.aggregate ?? false
1601 }
1602 })
1603 }
1604
1605 private cannotStealTask (): boolean {
1606 return this.workerNodes.length <= 1 || this.info.queuedTasks === 0
1607 }
1608
1609 private handleTask (workerNodeKey: number, task: Task<Data>): void {
1610 if (this.shallExecuteTask(workerNodeKey)) {
1611 this.executeTask(workerNodeKey, task)
1612 } else {
1613 this.enqueueTask(workerNodeKey, task)
1614 }
1615 }
1616
1617 private redistributeQueuedTasks (workerNodeKey: number): void {
1618 if (workerNodeKey === -1 || this.cannotStealTask()) {
1619 return
1620 }
1621 while (this.tasksQueueSize(workerNodeKey) > 0) {
1622 const destinationWorkerNodeKey = this.workerNodes.reduce(
1623 (minWorkerNodeKey, workerNode, workerNodeKey, workerNodes) => {
1624 return workerNode.info.ready &&
1625 workerNode.usage.tasks.queued <
1626 workerNodes[minWorkerNodeKey].usage.tasks.queued
1627 ? workerNodeKey
1628 : minWorkerNodeKey
1629 },
1630 0
1631 )
1632 this.handleTask(
1633 destinationWorkerNodeKey,
1634 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1635 this.dequeueTask(workerNodeKey)!
1636 )
1637 }
1638 }
1639
1640 private updateTaskStolenStatisticsWorkerUsage (
1641 workerNodeKey: number,
1642 taskName: string
1643 ): void {
1644 const workerNode = this.workerNodes[workerNodeKey]
1645 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1646 if (workerNode?.usage != null) {
1647 ++workerNode.usage.tasks.stolen
1648 }
1649 if (
1650 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) &&
1651 workerNode.getTaskFunctionWorkerUsage(taskName) != null
1652 ) {
1653 const taskFunctionWorkerUsage =
1654 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1655 workerNode.getTaskFunctionWorkerUsage(taskName)!
1656 ++taskFunctionWorkerUsage.tasks.stolen
1657 }
1658 }
1659
1660 private updateTaskSequentiallyStolenStatisticsWorkerUsage (
1661 workerNodeKey: number
1662 ): void {
1663 const workerNode = this.workerNodes[workerNodeKey]
1664 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1665 if (workerNode?.usage != null) {
1666 ++workerNode.usage.tasks.sequentiallyStolen
1667 }
1668 }
1669
1670 private updateTaskSequentiallyStolenStatisticsTaskFunctionWorkerUsage (
1671 workerNodeKey: number,
1672 taskName: string
1673 ): void {
1674 const workerNode = this.workerNodes[workerNodeKey]
1675 if (
1676 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) &&
1677 workerNode.getTaskFunctionWorkerUsage(taskName) != null
1678 ) {
1679 const taskFunctionWorkerUsage =
1680 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1681 workerNode.getTaskFunctionWorkerUsage(taskName)!
1682 ++taskFunctionWorkerUsage.tasks.sequentiallyStolen
1683 }
1684 }
1685
1686 private resetTaskSequentiallyStolenStatisticsWorkerUsage (
1687 workerNodeKey: number
1688 ): void {
1689 const workerNode = this.workerNodes[workerNodeKey]
1690 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1691 if (workerNode?.usage != null) {
1692 workerNode.usage.tasks.sequentiallyStolen = 0
1693 }
1694 }
1695
1696 private resetTaskSequentiallyStolenStatisticsTaskFunctionWorkerUsage (
1697 workerNodeKey: number,
1698 taskName: string
1699 ): void {
1700 const workerNode = this.workerNodes[workerNodeKey]
1701 if (
1702 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) &&
1703 workerNode.getTaskFunctionWorkerUsage(taskName) != null
1704 ) {
1705 const taskFunctionWorkerUsage =
1706 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1707 workerNode.getTaskFunctionWorkerUsage(taskName)!
1708 taskFunctionWorkerUsage.tasks.sequentiallyStolen = 0
1709 }
1710 }
1711
1712 private readonly handleWorkerNodeIdleEvent = (
1713 eventDetail: WorkerNodeEventDetail,
1714 previousStolenTask?: Task<Data>
1715 ): void => {
1716 const { workerNodeKey } = eventDetail
1717 if (workerNodeKey == null) {
1718 throw new Error(
1719 "WorkerNode event detail 'workerNodeKey' property must be defined"
1720 )
1721 }
1722 const workerInfo = this.getWorkerInfo(workerNodeKey)
1723 if (
1724 this.cannotStealTask() ||
1725 (this.info.stealingWorkerNodes ?? 0) >
1726 Math.floor(this.workerNodes.length / 2)
1727 ) {
1728 if (workerInfo != null && previousStolenTask != null) {
1729 workerInfo.stealing = false
1730 }
1731 return
1732 }
1733 const workerNodeTasksUsage = this.workerNodes[workerNodeKey].usage.tasks
1734 if (
1735 workerInfo != null &&
1736 previousStolenTask != null &&
1737 workerNodeTasksUsage.sequentiallyStolen > 0 &&
1738 (workerNodeTasksUsage.executing > 0 ||
1739 this.tasksQueueSize(workerNodeKey) > 0)
1740 ) {
1741 workerInfo.stealing = false
1742 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1743 for (const taskFunctionProperties of workerInfo.taskFunctionsProperties!) {
1744 this.resetTaskSequentiallyStolenStatisticsTaskFunctionWorkerUsage(
1745 workerNodeKey,
1746 taskFunctionProperties.name
1747 )
1748 }
1749 this.resetTaskSequentiallyStolenStatisticsWorkerUsage(workerNodeKey)
1750 return
1751 }
1752 if (workerInfo == null) {
1753 throw new Error(
1754 `Worker node with key '${workerNodeKey}' not found in pool`
1755 )
1756 }
1757 workerInfo.stealing = true
1758 const stolenTask = this.workerNodeStealTask(workerNodeKey)
1759 if (
1760 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) &&
1761 stolenTask != null
1762 ) {
1763 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1764 const taskFunctionTasksWorkerUsage = this.workerNodes[
1765 workerNodeKey
1766 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1767 ].getTaskFunctionWorkerUsage(stolenTask.name!)!.tasks
1768 if (
1769 taskFunctionTasksWorkerUsage.sequentiallyStolen === 0 ||
1770 (previousStolenTask != null &&
1771 previousStolenTask.name === stolenTask.name &&
1772 taskFunctionTasksWorkerUsage.sequentiallyStolen > 0)
1773 ) {
1774 this.updateTaskSequentiallyStolenStatisticsTaskFunctionWorkerUsage(
1775 workerNodeKey,
1776 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1777 stolenTask.name!
1778 )
1779 } else {
1780 this.resetTaskSequentiallyStolenStatisticsTaskFunctionWorkerUsage(
1781 workerNodeKey,
1782 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1783 stolenTask.name!
1784 )
1785 }
1786 }
1787 sleep(exponentialDelay(workerNodeTasksUsage.sequentiallyStolen))
1788 .then(() => {
1789 this.handleWorkerNodeIdleEvent(eventDetail, stolenTask)
1790 return undefined
1791 })
1792 .catch((error: unknown) => {
1793 this.emitter?.emit(PoolEvents.error, error)
1794 })
1795 }
1796
1797 private readonly workerNodeStealTask = (
1798 workerNodeKey: number
1799 ): Task<Data> | undefined => {
1800 const workerNodes = this.workerNodes
1801 .slice()
1802 .sort(
1803 (workerNodeA, workerNodeB) =>
1804 workerNodeB.usage.tasks.queued - workerNodeA.usage.tasks.queued
1805 )
1806 const sourceWorkerNode = workerNodes.find(
1807 (sourceWorkerNode, sourceWorkerNodeKey) =>
1808 sourceWorkerNode.info.ready &&
1809 !sourceWorkerNode.info.stealing &&
1810 sourceWorkerNodeKey !== workerNodeKey &&
1811 sourceWorkerNode.usage.tasks.queued > 0
1812 )
1813 if (sourceWorkerNode != null) {
1814 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1815 const task = sourceWorkerNode.dequeueTask(1)!
1816 this.handleTask(workerNodeKey, task)
1817 this.updateTaskSequentiallyStolenStatisticsWorkerUsage(workerNodeKey)
1818 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1819 this.updateTaskStolenStatisticsWorkerUsage(workerNodeKey, task.name!)
1820 return task
1821 }
1822 }
1823
1824 private readonly handleWorkerNodeBackPressureEvent = (
1825 eventDetail: WorkerNodeEventDetail
1826 ): void => {
1827 if (
1828 this.cannotStealTask() ||
1829 (this.info.stealingWorkerNodes ?? 0) >
1830 Math.floor(this.workerNodes.length / 2)
1831 ) {
1832 return
1833 }
1834 const { workerId } = eventDetail
1835 const sizeOffset = 1
1836 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1837 if (this.opts.tasksQueueOptions!.size! <= sizeOffset) {
1838 return
1839 }
1840 const sourceWorkerNode =
1841 this.workerNodes[this.getWorkerNodeKeyByWorkerId(workerId)]
1842 const workerNodes = this.workerNodes
1843 .slice()
1844 .sort(
1845 (workerNodeA, workerNodeB) =>
1846 workerNodeA.usage.tasks.queued - workerNodeB.usage.tasks.queued
1847 )
1848 for (const [workerNodeKey, workerNode] of workerNodes.entries()) {
1849 if (
1850 sourceWorkerNode.usage.tasks.queued > 0 &&
1851 workerNode.info.ready &&
1852 !workerNode.info.stealing &&
1853 workerNode.info.id !== workerId &&
1854 workerNode.usage.tasks.queued <
1855 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1856 this.opts.tasksQueueOptions!.size! - sizeOffset
1857 ) {
1858 const workerInfo = this.getWorkerInfo(workerNodeKey)
1859 if (workerInfo == null) {
1860 throw new Error(
1861 `Worker node with key '${workerNodeKey}' not found in pool`
1862 )
1863 }
1864 workerInfo.stealing = true
1865 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1866 const task = sourceWorkerNode.dequeueTask(1)!
1867 this.handleTask(workerNodeKey, task)
1868 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1869 this.updateTaskStolenStatisticsWorkerUsage(workerNodeKey, task.name!)
1870 workerInfo.stealing = false
1871 }
1872 }
1873 }
1874
1875 /**
1876 * This method is the message listener registered on each worker.
1877 */
1878 protected readonly workerMessageListener = (
1879 message: MessageValue<Response>
1880 ): void => {
1881 this.checkMessageWorkerId(message)
1882 const { workerId, ready, taskId, taskFunctionsProperties } = message
1883 if (ready != null && taskFunctionsProperties != null) {
1884 // Worker ready response received from worker
1885 this.handleWorkerReadyResponse(message)
1886 } else if (taskFunctionsProperties != null) {
1887 // Task function properties message received from worker
1888 const workerNodeKey = this.getWorkerNodeKeyByWorkerId(workerId)
1889 const workerInfo = this.getWorkerInfo(workerNodeKey)
1890 if (workerInfo != null) {
1891 workerInfo.taskFunctionsProperties = taskFunctionsProperties
1892 this.sendStatisticsMessageToWorker(workerNodeKey)
1893 }
1894 } else if (taskId != null) {
1895 // Task execution response received from worker
1896 this.handleTaskExecutionResponse(message)
1897 }
1898 }
1899
1900 private checkAndEmitReadyEvent (): void {
1901 if (!this.readyEventEmitted && this.ready) {
1902 this.emitter?.emit(PoolEvents.ready, this.info)
1903 this.readyEventEmitted = true
1904 }
1905 }
1906
1907 private handleWorkerReadyResponse (message: MessageValue<Response>): void {
1908 const { workerId, ready, taskFunctionsProperties } = message
1909 if (ready == null || !ready) {
1910 throw new Error(`Worker ${workerId} failed to initialize`)
1911 }
1912 const workerNodeKey = this.getWorkerNodeKeyByWorkerId(workerId)
1913 const workerNode = this.workerNodes[workerNodeKey]
1914 workerNode.info.ready = ready
1915 workerNode.info.taskFunctionsProperties = taskFunctionsProperties
1916 this.sendStatisticsMessageToWorker(workerNodeKey)
1917 this.checkAndEmitReadyEvent()
1918 }
1919
1920 private handleTaskExecutionResponse (message: MessageValue<Response>): void {
1921 const { workerId, taskId, workerError, data } = message
1922 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1923 const promiseResponse = this.promiseResponseMap.get(taskId!)
1924 if (promiseResponse != null) {
1925 const { resolve, reject, workerNodeKey, asyncResource } = promiseResponse
1926 const workerNode = this.workerNodes[workerNodeKey]
1927 if (workerError != null) {
1928 this.emitter?.emit(PoolEvents.taskError, workerError)
1929 asyncResource != null
1930 ? asyncResource.runInAsyncScope(
1931 reject,
1932 this.emitter,
1933 workerError.message
1934 )
1935 : reject(workerError.message)
1936 } else {
1937 asyncResource != null
1938 ? asyncResource.runInAsyncScope(resolve, this.emitter, data)
1939 : resolve(data as Response)
1940 }
1941 asyncResource?.emitDestroy()
1942 this.afterTaskExecutionHook(workerNodeKey, message)
1943 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1944 this.promiseResponseMap.delete(taskId!)
1945 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1946 workerNode?.emit('taskFinished', taskId)
1947 if (
1948 this.opts.enableTasksQueue === true &&
1949 !this.destroying &&
1950 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1951 workerNode != null
1952 ) {
1953 const workerNodeTasksUsage = workerNode.usage.tasks
1954 if (
1955 this.tasksQueueSize(workerNodeKey) > 0 &&
1956 workerNodeTasksUsage.executing <
1957 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1958 this.opts.tasksQueueOptions!.concurrency!
1959 ) {
1960 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1961 this.executeTask(workerNodeKey, this.dequeueTask(workerNodeKey)!)
1962 }
1963 if (
1964 workerNodeTasksUsage.executing === 0 &&
1965 this.tasksQueueSize(workerNodeKey) === 0 &&
1966 workerNodeTasksUsage.sequentiallyStolen === 0
1967 ) {
1968 workerNode.emit('idle', {
1969 workerId,
1970 workerNodeKey
1971 })
1972 }
1973 }
1974 }
1975 }
1976
1977 private checkAndEmitTaskExecutionEvents (): void {
1978 if (this.busy) {
1979 this.emitter?.emit(PoolEvents.busy, this.info)
1980 }
1981 }
1982
1983 private checkAndEmitTaskQueuingEvents (): void {
1984 if (this.hasBackPressure()) {
1985 this.emitter?.emit(PoolEvents.backPressure, this.info)
1986 }
1987 }
1988
1989 /**
1990 * Emits dynamic worker creation events.
1991 */
1992 protected abstract checkAndEmitDynamicWorkerCreationEvents (): void
1993
1994 /**
1995 * Gets the worker information given its worker node key.
1996 *
1997 * @param workerNodeKey - The worker node key.
1998 * @returns The worker information.
1999 */
2000 protected getWorkerInfo (workerNodeKey: number): WorkerInfo | undefined {
2001 return this.workerNodes[workerNodeKey]?.info
2002 }
2003
2004 /**
2005 * Creates a worker node.
2006 *
2007 * @returns The created worker node.
2008 */
2009 private createWorkerNode (): IWorkerNode<Worker, Data> {
2010 const workerNode = new WorkerNode<Worker, Data>(
2011 this.worker,
2012 this.filePath,
2013 {
2014 env: this.opts.env,
2015 workerOptions: this.opts.workerOptions,
2016 tasksQueueBackPressureSize:
2017 this.opts.tasksQueueOptions?.size ??
2018 getDefaultTasksQueueOptions(
2019 this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers
2020 ).size,
2021 tasksQueueBucketSize:
2022 (this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers) * 2
2023 }
2024 )
2025 // Flag the worker node as ready at pool startup.
2026 if (this.starting) {
2027 workerNode.info.ready = true
2028 }
2029 return workerNode
2030 }
2031
2032 /**
2033 * Adds the given worker node in the pool worker nodes.
2034 *
2035 * @param workerNode - The worker node.
2036 * @returns The added worker node key.
2037 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the added worker node is not found.
2038 */
2039 private addWorkerNode (workerNode: IWorkerNode<Worker, Data>): number {
2040 this.workerNodes.push(workerNode)
2041 const workerNodeKey = this.workerNodes.indexOf(workerNode)
2042 if (workerNodeKey === -1) {
2043 throw new Error('Worker added not found in worker nodes')
2044 }
2045 return workerNodeKey
2046 }
2047
2048 private checkAndEmitEmptyEvent (): void {
2049 if (this.empty) {
2050 this.emitter?.emit(PoolEvents.empty, this.info)
2051 this.readyEventEmitted = false
2052 }
2053 }
2054
2055 /**
2056 * Removes the worker node from the pool worker nodes.
2057 *
2058 * @param workerNode - The worker node.
2059 */
2060 private removeWorkerNode (workerNode: IWorkerNode<Worker, Data>): void {
2061 const workerNodeKey = this.workerNodes.indexOf(workerNode)
2062 if (workerNodeKey !== -1) {
2063 this.workerNodes.splice(workerNodeKey, 1)
2064 this.workerChoiceStrategiesContext?.remove(workerNodeKey)
2065 }
2066 this.checkAndEmitEmptyEvent()
2067 }
2068
2069 protected flagWorkerNodeAsNotReady (workerNodeKey: number): void {
2070 const workerInfo = this.getWorkerInfo(workerNodeKey)
2071 if (workerInfo != null) {
2072 workerInfo.ready = false
2073 }
2074 }
2075
2076 private hasBackPressure (): boolean {
2077 return (
2078 this.opts.enableTasksQueue === true &&
2079 this.workerNodes.findIndex(
2080 workerNode => !workerNode.hasBackPressure()
2081 ) === -1
2082 )
2083 }
2084
2085 /**
2086 * Executes the given task on the worker given its worker node key.
2087 *
2088 * @param workerNodeKey - The worker node key.
2089 * @param task - The task to execute.
2090 */
2091 private executeTask (workerNodeKey: number, task: Task<Data>): void {
2092 this.beforeTaskExecutionHook(workerNodeKey, task)
2093 this.sendToWorker(workerNodeKey, task, task.transferList)
2094 this.checkAndEmitTaskExecutionEvents()
2095 }
2096
2097 private enqueueTask (workerNodeKey: number, task: Task<Data>): number {
2098 const tasksQueueSize = this.workerNodes[workerNodeKey].enqueueTask(task)
2099 this.checkAndEmitTaskQueuingEvents()
2100 return tasksQueueSize
2101 }
2102
2103 private dequeueTask (
2104 workerNodeKey: number,
2105 bucket?: number
2106 ): Task<Data> | undefined {
2107 return this.workerNodes[workerNodeKey].dequeueTask(bucket)
2108 }
2109
2110 private tasksQueueSize (workerNodeKey: number): number {
2111 return this.workerNodes[workerNodeKey].tasksQueueSize()
2112 }
2113
2114 protected flushTasksQueue (workerNodeKey: number): number {
2115 let flushedTasks = 0
2116 while (this.tasksQueueSize(workerNodeKey) > 0) {
2117 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2118 this.executeTask(workerNodeKey, this.dequeueTask(workerNodeKey)!)
2119 ++flushedTasks
2120 }
2121 this.workerNodes[workerNodeKey].clearTasksQueue()
2122 return flushedTasks
2123 }
2124
2125 private flushTasksQueues (): void {
2126 for (const workerNodeKey of this.workerNodes.keys()) {
2127 this.flushTasksQueue(workerNodeKey)
2128 }
2129 }
2130 }