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