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