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