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