Merge dependabot/npm_and_yarn/examples/typescript/websocket-server-pool/ws-hybrid...
[poolifier.git] / src / pools / abstract-pool.ts
1 import { randomUUID } from 'node:crypto'
2 import { performance } from 'node:perf_hooks'
3 import { existsSync } from 'node:fs'
4 import { type TransferListItem } from 'node:worker_threads'
5 import type {
6 MessageValue,
7 PromiseResponseWrapper,
8 Task
9 } from '../utility-types'
10 import {
11 DEFAULT_TASK_NAME,
12 DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS,
13 EMPTY_FUNCTION,
14 average,
15 isKillBehavior,
16 isPlainObject,
17 max,
18 median,
19 min,
20 round,
21 updateMeasurementStatistics
22 } from '../utils'
23 import { KillBehaviors } from '../worker/worker-options'
24 import {
25 type IPool,
26 PoolEmitter,
27 PoolEvents,
28 type PoolInfo,
29 type PoolOptions,
30 type PoolType,
31 PoolTypes,
32 type TasksQueueOptions
33 } from './pool'
34 import type {
35 IWorker,
36 IWorkerNode,
37 WorkerInfo,
38 WorkerType,
39 WorkerUsage
40 } from './worker'
41 import {
42 type MeasurementStatisticsRequirements,
43 Measurements,
44 WorkerChoiceStrategies,
45 type WorkerChoiceStrategy,
46 type WorkerChoiceStrategyOptions
47 } from './selection-strategies/selection-strategies-types'
48 import { WorkerChoiceStrategyContext } from './selection-strategies/worker-choice-strategy-context'
49 import { version } from './version'
50 import { WorkerNode } from './worker-node'
51
52 /**
53 * Base class that implements some shared logic for all poolifier pools.
54 *
55 * @typeParam Worker - Type of worker which manages this pool.
56 * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
57 * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
58 */
59 export abstract class AbstractPool<
60 Worker extends IWorker,
61 Data = unknown,
62 Response = unknown
63 > implements IPool<Worker, Data, Response> {
64 /** @inheritDoc */
65 public readonly workerNodes: Array<IWorkerNode<Worker, Data>> = []
66
67 /** @inheritDoc */
68 public readonly emitter?: PoolEmitter
69
70 /**
71 * The task execution response promise map.
72 *
73 * - `key`: The message id of each submitted task.
74 * - `value`: An object that contains the worker, the execution response promise resolve and reject callbacks.
75 *
76 * When we receive a message from the worker, we get a map entry with the promise resolve/reject bound to the message id.
77 */
78 protected promiseResponseMap: Map<string, PromiseResponseWrapper<Response>> =
79 new Map<string, PromiseResponseWrapper<Response>>()
80
81 /**
82 * Worker choice strategy context referencing a worker choice algorithm implementation.
83 */
84 protected workerChoiceStrategyContext: WorkerChoiceStrategyContext<
85 Worker,
86 Data,
87 Response
88 >
89
90 /**
91 * Dynamic pool maximum size property placeholder.
92 */
93 protected readonly max?: number
94
95 /**
96 * Whether the pool is starting or not.
97 */
98 private readonly starting: boolean
99 /**
100 * Whether the pool is started or not.
101 */
102 private started: boolean
103 /**
104 * The start timestamp of the pool.
105 */
106 private readonly startTimestamp
107
108 /**
109 * Constructs a new poolifier pool.
110 *
111 * @param numberOfWorkers - Number of workers that this pool should manage.
112 * @param filePath - Path to the worker file.
113 * @param opts - Options for the pool.
114 */
115 public constructor (
116 protected readonly numberOfWorkers: number,
117 protected readonly filePath: string,
118 protected readonly opts: PoolOptions<Worker>
119 ) {
120 if (!this.isMain()) {
121 throw new Error(
122 'Cannot start a pool from a worker with the same type as the pool'
123 )
124 }
125 this.checkNumberOfWorkers(this.numberOfWorkers)
126 this.checkFilePath(this.filePath)
127 this.checkPoolOptions(this.opts)
128
129 this.chooseWorkerNode = this.chooseWorkerNode.bind(this)
130 this.executeTask = this.executeTask.bind(this)
131 this.enqueueTask = this.enqueueTask.bind(this)
132
133 if (this.opts.enableEvents === true) {
134 this.emitter = new PoolEmitter()
135 }
136 this.workerChoiceStrategyContext = new WorkerChoiceStrategyContext<
137 Worker,
138 Data,
139 Response
140 >(
141 this,
142 this.opts.workerChoiceStrategy,
143 this.opts.workerChoiceStrategyOptions
144 )
145
146 this.setupHook()
147
148 this.starting = true
149 this.startPool()
150 this.starting = false
151 this.started = true
152
153 this.startTimestamp = performance.now()
154 }
155
156 private checkFilePath (filePath: string): void {
157 if (
158 filePath == null ||
159 typeof filePath !== 'string' ||
160 (typeof filePath === 'string' && filePath.trim().length === 0)
161 ) {
162 throw new Error('Please specify a file with a worker implementation')
163 }
164 if (!existsSync(filePath)) {
165 throw new Error(`Cannot find the worker file '${filePath}'`)
166 }
167 }
168
169 private checkNumberOfWorkers (numberOfWorkers: number): void {
170 if (numberOfWorkers == null) {
171 throw new Error(
172 'Cannot instantiate a pool without specifying the number of workers'
173 )
174 } else if (!Number.isSafeInteger(numberOfWorkers)) {
175 throw new TypeError(
176 'Cannot instantiate a pool with a non safe integer number of workers'
177 )
178 } else if (numberOfWorkers < 0) {
179 throw new RangeError(
180 'Cannot instantiate a pool with a negative number of workers'
181 )
182 } else if (this.type === PoolTypes.fixed && numberOfWorkers === 0) {
183 throw new RangeError('Cannot instantiate a fixed pool with zero worker')
184 }
185 }
186
187 protected checkDynamicPoolSize (min: number, max: number): void {
188 if (this.type === PoolTypes.dynamic) {
189 if (max == null) {
190 throw new TypeError(
191 'Cannot instantiate a dynamic pool without specifying the maximum pool size'
192 )
193 } else if (!Number.isSafeInteger(max)) {
194 throw new TypeError(
195 'Cannot instantiate a dynamic pool with a non safe integer maximum pool size'
196 )
197 } else if (min > max) {
198 throw new RangeError(
199 'Cannot instantiate a dynamic pool with a maximum pool size inferior to the minimum pool size'
200 )
201 } else if (max === 0) {
202 throw new RangeError(
203 'Cannot instantiate a dynamic pool with a maximum pool size equal to zero'
204 )
205 } else if (min === max) {
206 throw new RangeError(
207 'Cannot instantiate a dynamic pool with a minimum pool size equal to the maximum pool size. Use a fixed pool instead'
208 )
209 }
210 }
211 }
212
213 private checkPoolOptions (opts: PoolOptions<Worker>): void {
214 if (isPlainObject(opts)) {
215 this.opts.workerChoiceStrategy =
216 opts.workerChoiceStrategy ?? WorkerChoiceStrategies.ROUND_ROBIN
217 this.checkValidWorkerChoiceStrategy(this.opts.workerChoiceStrategy)
218 this.opts.workerChoiceStrategyOptions = {
219 ...DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS,
220 ...opts.workerChoiceStrategyOptions
221 }
222 this.checkValidWorkerChoiceStrategyOptions(
223 this.opts.workerChoiceStrategyOptions
224 )
225 this.opts.restartWorkerOnError = opts.restartWorkerOnError ?? true
226 this.opts.enableEvents = opts.enableEvents ?? true
227 this.opts.enableTasksQueue = opts.enableTasksQueue ?? false
228 if (this.opts.enableTasksQueue) {
229 this.checkValidTasksQueueOptions(
230 opts.tasksQueueOptions as TasksQueueOptions
231 )
232 this.opts.tasksQueueOptions = this.buildTasksQueueOptions(
233 opts.tasksQueueOptions as TasksQueueOptions
234 )
235 }
236 } else {
237 throw new TypeError('Invalid pool options: must be a plain object')
238 }
239 }
240
241 private checkValidWorkerChoiceStrategy (
242 workerChoiceStrategy: WorkerChoiceStrategy
243 ): void {
244 if (!Object.values(WorkerChoiceStrategies).includes(workerChoiceStrategy)) {
245 throw new Error(
246 `Invalid worker choice strategy '${workerChoiceStrategy}'`
247 )
248 }
249 }
250
251 private checkValidWorkerChoiceStrategyOptions (
252 workerChoiceStrategyOptions: WorkerChoiceStrategyOptions
253 ): void {
254 if (!isPlainObject(workerChoiceStrategyOptions)) {
255 throw new TypeError(
256 'Invalid worker choice strategy options: must be a plain object'
257 )
258 }
259 if (
260 workerChoiceStrategyOptions.retries != null &&
261 !Number.isSafeInteger(workerChoiceStrategyOptions.retries)
262 ) {
263 throw new TypeError(
264 'Invalid worker choice strategy options: retries must be an integer'
265 )
266 }
267 if (
268 workerChoiceStrategyOptions.retries != null &&
269 workerChoiceStrategyOptions.retries < 0
270 ) {
271 throw new RangeError(
272 `Invalid worker choice strategy options: retries '${workerChoiceStrategyOptions.retries}' must be greater or equal than zero`
273 )
274 }
275 if (
276 workerChoiceStrategyOptions.weights != null &&
277 Object.keys(workerChoiceStrategyOptions.weights).length !== this.maxSize
278 ) {
279 throw new Error(
280 'Invalid worker choice strategy options: must have a weight for each worker node'
281 )
282 }
283 if (
284 workerChoiceStrategyOptions.measurement != null &&
285 !Object.values(Measurements).includes(
286 workerChoiceStrategyOptions.measurement
287 )
288 ) {
289 throw new Error(
290 `Invalid worker choice strategy options: invalid measurement '${workerChoiceStrategyOptions.measurement}'`
291 )
292 }
293 }
294
295 private checkValidTasksQueueOptions (
296 tasksQueueOptions: TasksQueueOptions
297 ): void {
298 if (tasksQueueOptions != null && !isPlainObject(tasksQueueOptions)) {
299 throw new TypeError('Invalid tasks queue options: must be a plain object')
300 }
301 if (
302 tasksQueueOptions?.concurrency != null &&
303 !Number.isSafeInteger(tasksQueueOptions?.concurrency)
304 ) {
305 throw new TypeError(
306 'Invalid worker node tasks concurrency: must be an integer'
307 )
308 }
309 if (
310 tasksQueueOptions?.concurrency != null &&
311 tasksQueueOptions?.concurrency <= 0
312 ) {
313 throw new RangeError(
314 `Invalid worker node tasks concurrency: ${tasksQueueOptions?.concurrency} is a negative integer or zero`
315 )
316 }
317 if (tasksQueueOptions?.queueMaxSize != null) {
318 throw new Error(
319 'Invalid tasks queue options: queueMaxSize is deprecated, please use size instead'
320 )
321 }
322 if (
323 tasksQueueOptions?.size != null &&
324 !Number.isSafeInteger(tasksQueueOptions?.size)
325 ) {
326 throw new TypeError(
327 'Invalid worker node tasks queue size: must be an integer'
328 )
329 }
330 if (tasksQueueOptions?.size != null && tasksQueueOptions?.size <= 0) {
331 throw new RangeError(
332 `Invalid worker node tasks queue size: ${tasksQueueOptions?.size} is a negative integer or zero`
333 )
334 }
335 }
336
337 private startPool (): void {
338 while (
339 this.workerNodes.reduce(
340 (accumulator, workerNode) =>
341 !workerNode.info.dynamic ? accumulator + 1 : accumulator,
342 0
343 ) < this.numberOfWorkers
344 ) {
345 this.createAndSetupWorkerNode()
346 }
347 }
348
349 /** @inheritDoc */
350 public get info (): PoolInfo {
351 return {
352 version,
353 type: this.type,
354 worker: this.worker,
355 ready: this.ready,
356 strategy: this.opts.workerChoiceStrategy as WorkerChoiceStrategy,
357 minSize: this.minSize,
358 maxSize: this.maxSize,
359 ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
360 .runTime.aggregate &&
361 this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
362 .waitTime.aggregate && { utilization: round(this.utilization) }),
363 workerNodes: this.workerNodes.length,
364 idleWorkerNodes: this.workerNodes.reduce(
365 (accumulator, workerNode) =>
366 workerNode.usage.tasks.executing === 0
367 ? accumulator + 1
368 : accumulator,
369 0
370 ),
371 busyWorkerNodes: this.workerNodes.reduce(
372 (accumulator, workerNode) =>
373 workerNode.usage.tasks.executing > 0 ? accumulator + 1 : accumulator,
374 0
375 ),
376 executedTasks: this.workerNodes.reduce(
377 (accumulator, workerNode) =>
378 accumulator + workerNode.usage.tasks.executed,
379 0
380 ),
381 executingTasks: this.workerNodes.reduce(
382 (accumulator, workerNode) =>
383 accumulator + workerNode.usage.tasks.executing,
384 0
385 ),
386 ...(this.opts.enableTasksQueue === true && {
387 queuedTasks: this.workerNodes.reduce(
388 (accumulator, workerNode) =>
389 accumulator + workerNode.usage.tasks.queued,
390 0
391 )
392 }),
393 ...(this.opts.enableTasksQueue === true && {
394 maxQueuedTasks: this.workerNodes.reduce(
395 (accumulator, workerNode) =>
396 accumulator + (workerNode.usage.tasks?.maxQueued ?? 0),
397 0
398 )
399 }),
400 ...(this.opts.enableTasksQueue === true && {
401 backPressure: this.hasBackPressure()
402 }),
403 ...(this.opts.enableTasksQueue === true && {
404 stolenTasks: this.workerNodes.reduce(
405 (accumulator, workerNode) =>
406 accumulator + workerNode.usage.tasks.stolen,
407 0
408 )
409 }),
410 failedTasks: this.workerNodes.reduce(
411 (accumulator, workerNode) =>
412 accumulator + workerNode.usage.tasks.failed,
413 0
414 ),
415 ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
416 .runTime.aggregate && {
417 runTime: {
418 minimum: round(
419 min(
420 ...this.workerNodes.map(
421 (workerNode) => workerNode.usage.runTime?.minimum ?? Infinity
422 )
423 )
424 ),
425 maximum: round(
426 max(
427 ...this.workerNodes.map(
428 (workerNode) => workerNode.usage.runTime?.maximum ?? -Infinity
429 )
430 )
431 ),
432 ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
433 .runTime.average && {
434 average: round(
435 average(
436 this.workerNodes.reduce<number[]>(
437 (accumulator, workerNode) =>
438 accumulator.concat(workerNode.usage.runTime.history),
439 []
440 )
441 )
442 )
443 }),
444 ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
445 .runTime.median && {
446 median: round(
447 median(
448 this.workerNodes.reduce<number[]>(
449 (accumulator, workerNode) =>
450 accumulator.concat(workerNode.usage.runTime.history),
451 []
452 )
453 )
454 )
455 })
456 }
457 }),
458 ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
459 .waitTime.aggregate && {
460 waitTime: {
461 minimum: round(
462 min(
463 ...this.workerNodes.map(
464 (workerNode) => workerNode.usage.waitTime?.minimum ?? Infinity
465 )
466 )
467 ),
468 maximum: round(
469 max(
470 ...this.workerNodes.map(
471 (workerNode) => workerNode.usage.waitTime?.maximum ?? -Infinity
472 )
473 )
474 ),
475 ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
476 .waitTime.average && {
477 average: round(
478 average(
479 this.workerNodes.reduce<number[]>(
480 (accumulator, workerNode) =>
481 accumulator.concat(workerNode.usage.waitTime.history),
482 []
483 )
484 )
485 )
486 }),
487 ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
488 .waitTime.median && {
489 median: round(
490 median(
491 this.workerNodes.reduce<number[]>(
492 (accumulator, workerNode) =>
493 accumulator.concat(workerNode.usage.waitTime.history),
494 []
495 )
496 )
497 )
498 })
499 }
500 })
501 }
502 }
503
504 /**
505 * The pool readiness boolean status.
506 */
507 private get ready (): boolean {
508 return (
509 this.workerNodes.reduce(
510 (accumulator, workerNode) =>
511 !workerNode.info.dynamic && workerNode.info.ready
512 ? accumulator + 1
513 : accumulator,
514 0
515 ) >= this.minSize
516 )
517 }
518
519 /**
520 * The approximate pool utilization.
521 *
522 * @returns The pool utilization.
523 */
524 private get utilization (): number {
525 const poolTimeCapacity =
526 (performance.now() - this.startTimestamp) * this.maxSize
527 const totalTasksRunTime = this.workerNodes.reduce(
528 (accumulator, workerNode) =>
529 accumulator + (workerNode.usage.runTime?.aggregate ?? 0),
530 0
531 )
532 const totalTasksWaitTime = this.workerNodes.reduce(
533 (accumulator, workerNode) =>
534 accumulator + (workerNode.usage.waitTime?.aggregate ?? 0),
535 0
536 )
537 return (totalTasksRunTime + totalTasksWaitTime) / poolTimeCapacity
538 }
539
540 /**
541 * The pool type.
542 *
543 * If it is `'dynamic'`, it provides the `max` property.
544 */
545 protected abstract get type (): PoolType
546
547 /**
548 * The worker type.
549 */
550 protected abstract get worker (): WorkerType
551
552 /**
553 * The pool minimum size.
554 */
555 protected get minSize (): number {
556 return this.numberOfWorkers
557 }
558
559 /**
560 * The pool maximum size.
561 */
562 protected get maxSize (): number {
563 return this.max ?? this.numberOfWorkers
564 }
565
566 /**
567 * Checks if the worker id sent in the received message from a worker is valid.
568 *
569 * @param message - The received message.
570 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the worker id is invalid.
571 */
572 private checkMessageWorkerId (message: MessageValue<Response>): void {
573 if (message.workerId == null) {
574 throw new Error('Worker message received without worker id')
575 } else if (
576 message.workerId != null &&
577 this.getWorkerNodeKeyByWorkerId(message.workerId) === -1
578 ) {
579 throw new Error(
580 `Worker message received from unknown worker '${message.workerId}'`
581 )
582 }
583 }
584
585 /**
586 * Gets the given worker its worker node key.
587 *
588 * @param worker - The worker.
589 * @returns The worker node key if found in the pool worker nodes, `-1` otherwise.
590 */
591 private getWorkerNodeKeyByWorker (worker: Worker): number {
592 return this.workerNodes.findIndex(
593 (workerNode) => workerNode.worker === worker
594 )
595 }
596
597 /**
598 * Gets the worker node key given its worker id.
599 *
600 * @param workerId - The worker id.
601 * @returns The worker node key if the worker id is found in the pool worker nodes, `-1` otherwise.
602 */
603 private getWorkerNodeKeyByWorkerId (workerId: number): number {
604 return this.workerNodes.findIndex(
605 (workerNode) => workerNode.info.id === workerId
606 )
607 }
608
609 /** @inheritDoc */
610 public setWorkerChoiceStrategy (
611 workerChoiceStrategy: WorkerChoiceStrategy,
612 workerChoiceStrategyOptions?: WorkerChoiceStrategyOptions
613 ): void {
614 this.checkValidWorkerChoiceStrategy(workerChoiceStrategy)
615 this.opts.workerChoiceStrategy = workerChoiceStrategy
616 this.workerChoiceStrategyContext.setWorkerChoiceStrategy(
617 this.opts.workerChoiceStrategy
618 )
619 if (workerChoiceStrategyOptions != null) {
620 this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions)
621 }
622 for (const [workerNodeKey, workerNode] of this.workerNodes.entries()) {
623 workerNode.resetUsage()
624 this.sendStatisticsMessageToWorker(workerNodeKey)
625 }
626 }
627
628 /** @inheritDoc */
629 public setWorkerChoiceStrategyOptions (
630 workerChoiceStrategyOptions: WorkerChoiceStrategyOptions
631 ): void {
632 this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions)
633 this.opts.workerChoiceStrategyOptions = {
634 ...DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS,
635 ...workerChoiceStrategyOptions
636 }
637 this.workerChoiceStrategyContext.setOptions(
638 this.opts.workerChoiceStrategyOptions
639 )
640 }
641
642 /** @inheritDoc */
643 public enableTasksQueue (
644 enable: boolean,
645 tasksQueueOptions?: TasksQueueOptions
646 ): void {
647 if (this.opts.enableTasksQueue === true && !enable) {
648 this.flushTasksQueues()
649 }
650 this.opts.enableTasksQueue = enable
651 this.setTasksQueueOptions(tasksQueueOptions as TasksQueueOptions)
652 }
653
654 /** @inheritDoc */
655 public setTasksQueueOptions (tasksQueueOptions: TasksQueueOptions): void {
656 if (this.opts.enableTasksQueue === true) {
657 this.checkValidTasksQueueOptions(tasksQueueOptions)
658 this.opts.tasksQueueOptions =
659 this.buildTasksQueueOptions(tasksQueueOptions)
660 this.setTasksQueueSize(this.opts.tasksQueueOptions.size as number)
661 } else if (this.opts.tasksQueueOptions != null) {
662 delete this.opts.tasksQueueOptions
663 }
664 }
665
666 private setTasksQueueSize (size: number): void {
667 for (const workerNode of this.workerNodes) {
668 workerNode.tasksQueueBackPressureSize = size
669 }
670 }
671
672 private buildTasksQueueOptions (
673 tasksQueueOptions: TasksQueueOptions
674 ): TasksQueueOptions {
675 return {
676 ...{
677 size: Math.pow(this.maxSize, 2),
678 concurrency: 1
679 },
680 ...tasksQueueOptions
681 }
682 }
683
684 /**
685 * Whether the pool is full or not.
686 *
687 * The pool filling boolean status.
688 */
689 protected get full (): boolean {
690 return this.workerNodes.length >= this.maxSize
691 }
692
693 /**
694 * Whether the pool is busy or not.
695 *
696 * The pool busyness boolean status.
697 */
698 protected abstract get busy (): boolean
699
700 /**
701 * Whether worker nodes are executing concurrently their tasks quota or not.
702 *
703 * @returns Worker nodes busyness boolean status.
704 */
705 protected internalBusy (): boolean {
706 if (this.opts.enableTasksQueue === true) {
707 return (
708 this.workerNodes.findIndex(
709 (workerNode) =>
710 workerNode.info.ready &&
711 workerNode.usage.tasks.executing <
712 (this.opts.tasksQueueOptions?.concurrency as number)
713 ) === -1
714 )
715 } else {
716 return (
717 this.workerNodes.findIndex(
718 (workerNode) =>
719 workerNode.info.ready && workerNode.usage.tasks.executing === 0
720 ) === -1
721 )
722 }
723 }
724
725 /** @inheritDoc */
726 public listTaskFunctions (): string[] {
727 for (const workerNode of this.workerNodes) {
728 if (
729 Array.isArray(workerNode.info.taskFunctions) &&
730 workerNode.info.taskFunctions.length > 0
731 ) {
732 return workerNode.info.taskFunctions
733 }
734 }
735 return []
736 }
737
738 private shallExecuteTask (workerNodeKey: number): boolean {
739 return (
740 this.tasksQueueSize(workerNodeKey) === 0 &&
741 this.workerNodes[workerNodeKey].usage.tasks.executing <
742 (this.opts.tasksQueueOptions?.concurrency as number)
743 )
744 }
745
746 /** @inheritDoc */
747 public async execute (
748 data?: Data,
749 name?: string,
750 transferList?: TransferListItem[]
751 ): Promise<Response> {
752 return await new Promise<Response>((resolve, reject) => {
753 if (!this.started) {
754 reject(new Error('Cannot execute a task on destroyed pool'))
755 return
756 }
757 if (name != null && typeof name !== 'string') {
758 reject(new TypeError('name argument must be a string'))
759 return
760 }
761 if (
762 name != null &&
763 typeof name === 'string' &&
764 name.trim().length === 0
765 ) {
766 reject(new TypeError('name argument must not be an empty string'))
767 return
768 }
769 if (transferList != null && !Array.isArray(transferList)) {
770 reject(new TypeError('transferList argument must be an array'))
771 return
772 }
773 const timestamp = performance.now()
774 const workerNodeKey = this.chooseWorkerNode()
775 const workerInfo = this.getWorkerInfo(workerNodeKey)
776 const task: Task<Data> = {
777 name: name ?? DEFAULT_TASK_NAME,
778 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
779 data: data ?? ({} as Data),
780 transferList,
781 timestamp,
782 workerId: workerInfo.id as number,
783 taskId: randomUUID()
784 }
785 this.promiseResponseMap.set(task.taskId as string, {
786 resolve,
787 reject,
788 workerNodeKey
789 })
790 if (
791 this.opts.enableTasksQueue === false ||
792 (this.opts.enableTasksQueue === true &&
793 this.shallExecuteTask(workerNodeKey))
794 ) {
795 this.executeTask(workerNodeKey, task)
796 } else {
797 this.enqueueTask(workerNodeKey, task)
798 }
799 })
800 }
801
802 /** @inheritDoc */
803 public async destroy (): Promise<void> {
804 await Promise.all(
805 this.workerNodes.map(async (_, workerNodeKey) => {
806 await this.destroyWorkerNode(workerNodeKey)
807 })
808 )
809 this.emitter?.emit(PoolEvents.destroy, this.info)
810 this.started = false
811 }
812
813 protected async sendKillMessageToWorker (
814 workerNodeKey: number,
815 workerId: number
816 ): Promise<void> {
817 await new Promise<void>((resolve, reject) => {
818 this.registerWorkerMessageListener(workerNodeKey, (message) => {
819 if (message.kill === 'success') {
820 resolve()
821 } else if (message.kill === 'failure') {
822 reject(new Error(`Worker ${workerId} kill message handling failed`))
823 }
824 })
825 this.sendToWorker(workerNodeKey, { kill: true, workerId })
826 })
827 }
828
829 /**
830 * Terminates the worker node given its worker node key.
831 *
832 * @param workerNodeKey - The worker node key.
833 */
834 protected abstract destroyWorkerNode (workerNodeKey: number): Promise<void>
835
836 /**
837 * Setup hook to execute code before worker nodes are created in the abstract constructor.
838 * Can be overridden.
839 *
840 * @virtual
841 */
842 protected setupHook (): void {
843 /* Intentionally empty */
844 }
845
846 /**
847 * Should return whether the worker is the main worker or not.
848 */
849 protected abstract isMain (): boolean
850
851 /**
852 * Hook executed before the worker task execution.
853 * Can be overridden.
854 *
855 * @param workerNodeKey - The worker node key.
856 * @param task - The task to execute.
857 */
858 protected beforeTaskExecutionHook (
859 workerNodeKey: number,
860 task: Task<Data>
861 ): void {
862 if (this.workerNodes[workerNodeKey]?.usage != null) {
863 const workerUsage = this.workerNodes[workerNodeKey].usage
864 ++workerUsage.tasks.executing
865 this.updateWaitTimeWorkerUsage(workerUsage, task)
866 }
867 if (
868 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) &&
869 this.workerNodes[workerNodeKey].getTaskFunctionWorkerUsage(
870 task.name as string
871 ) != null
872 ) {
873 const taskFunctionWorkerUsage = this.workerNodes[
874 workerNodeKey
875 ].getTaskFunctionWorkerUsage(task.name as string) as WorkerUsage
876 ++taskFunctionWorkerUsage.tasks.executing
877 this.updateWaitTimeWorkerUsage(taskFunctionWorkerUsage, task)
878 }
879 }
880
881 /**
882 * Hook executed after the worker task execution.
883 * Can be overridden.
884 *
885 * @param workerNodeKey - The worker node key.
886 * @param message - The received message.
887 */
888 protected afterTaskExecutionHook (
889 workerNodeKey: number,
890 message: MessageValue<Response>
891 ): void {
892 if (this.workerNodes[workerNodeKey]?.usage != null) {
893 const workerUsage = this.workerNodes[workerNodeKey].usage
894 this.updateTaskStatisticsWorkerUsage(workerUsage, message)
895 this.updateRunTimeWorkerUsage(workerUsage, message)
896 this.updateEluWorkerUsage(workerUsage, message)
897 }
898 if (
899 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) &&
900 this.workerNodes[workerNodeKey].getTaskFunctionWorkerUsage(
901 message.taskPerformance?.name as string
902 ) != null
903 ) {
904 const taskFunctionWorkerUsage = this.workerNodes[
905 workerNodeKey
906 ].getTaskFunctionWorkerUsage(
907 message.taskPerformance?.name as string
908 ) as WorkerUsage
909 this.updateTaskStatisticsWorkerUsage(taskFunctionWorkerUsage, message)
910 this.updateRunTimeWorkerUsage(taskFunctionWorkerUsage, message)
911 this.updateEluWorkerUsage(taskFunctionWorkerUsage, message)
912 }
913 }
914
915 /**
916 * Whether the worker node shall update its task function worker usage or not.
917 *
918 * @param workerNodeKey - The worker node key.
919 * @returns `true` if the worker node shall update its task function worker usage, `false` otherwise.
920 */
921 private shallUpdateTaskFunctionWorkerUsage (workerNodeKey: number): boolean {
922 const workerInfo = this.getWorkerInfo(workerNodeKey)
923 return (
924 workerInfo != null &&
925 Array.isArray(workerInfo.taskFunctions) &&
926 workerInfo.taskFunctions.length > 2
927 )
928 }
929
930 private updateTaskStatisticsWorkerUsage (
931 workerUsage: WorkerUsage,
932 message: MessageValue<Response>
933 ): void {
934 const workerTaskStatistics = workerUsage.tasks
935 if (
936 workerTaskStatistics.executing != null &&
937 workerTaskStatistics.executing > 0
938 ) {
939 --workerTaskStatistics.executing
940 }
941 if (message.taskError == null) {
942 ++workerTaskStatistics.executed
943 } else {
944 ++workerTaskStatistics.failed
945 }
946 }
947
948 private updateRunTimeWorkerUsage (
949 workerUsage: WorkerUsage,
950 message: MessageValue<Response>
951 ): void {
952 if (message.taskError != null) {
953 return
954 }
955 updateMeasurementStatistics(
956 workerUsage.runTime,
957 this.workerChoiceStrategyContext.getTaskStatisticsRequirements().runTime,
958 message.taskPerformance?.runTime ?? 0
959 )
960 }
961
962 private updateWaitTimeWorkerUsage (
963 workerUsage: WorkerUsage,
964 task: Task<Data>
965 ): void {
966 const timestamp = performance.now()
967 const taskWaitTime = timestamp - (task.timestamp ?? timestamp)
968 updateMeasurementStatistics(
969 workerUsage.waitTime,
970 this.workerChoiceStrategyContext.getTaskStatisticsRequirements().waitTime,
971 taskWaitTime
972 )
973 }
974
975 private updateEluWorkerUsage (
976 workerUsage: WorkerUsage,
977 message: MessageValue<Response>
978 ): void {
979 if (message.taskError != null) {
980 return
981 }
982 const eluTaskStatisticsRequirements: MeasurementStatisticsRequirements =
983 this.workerChoiceStrategyContext.getTaskStatisticsRequirements().elu
984 updateMeasurementStatistics(
985 workerUsage.elu.active,
986 eluTaskStatisticsRequirements,
987 message.taskPerformance?.elu?.active ?? 0
988 )
989 updateMeasurementStatistics(
990 workerUsage.elu.idle,
991 eluTaskStatisticsRequirements,
992 message.taskPerformance?.elu?.idle ?? 0
993 )
994 if (eluTaskStatisticsRequirements.aggregate) {
995 if (message.taskPerformance?.elu != null) {
996 if (workerUsage.elu.utilization != null) {
997 workerUsage.elu.utilization =
998 (workerUsage.elu.utilization +
999 message.taskPerformance.elu.utilization) /
1000 2
1001 } else {
1002 workerUsage.elu.utilization = message.taskPerformance.elu.utilization
1003 }
1004 }
1005 }
1006 }
1007
1008 /**
1009 * Chooses a worker node for the next task.
1010 *
1011 * The default worker choice strategy uses a round robin algorithm to distribute the tasks.
1012 *
1013 * @returns The chosen worker node key
1014 */
1015 private chooseWorkerNode (): number {
1016 if (this.shallCreateDynamicWorker()) {
1017 const workerNodeKey = this.createAndSetupDynamicWorkerNode()
1018 if (
1019 this.workerChoiceStrategyContext.getStrategyPolicy().dynamicWorkerUsage
1020 ) {
1021 return workerNodeKey
1022 }
1023 }
1024 return this.workerChoiceStrategyContext.execute()
1025 }
1026
1027 /**
1028 * Conditions for dynamic worker creation.
1029 *
1030 * @returns Whether to create a dynamic worker or not.
1031 */
1032 private shallCreateDynamicWorker (): boolean {
1033 return this.type === PoolTypes.dynamic && !this.full && this.internalBusy()
1034 }
1035
1036 /**
1037 * Sends a message to worker given its worker node key.
1038 *
1039 * @param workerNodeKey - The worker node key.
1040 * @param message - The message.
1041 * @param transferList - The optional array of transferable objects.
1042 */
1043 protected abstract sendToWorker (
1044 workerNodeKey: number,
1045 message: MessageValue<Data>,
1046 transferList?: TransferListItem[]
1047 ): void
1048
1049 /**
1050 * Creates a new worker.
1051 *
1052 * @returns Newly created worker.
1053 */
1054 protected abstract createWorker (): Worker
1055
1056 /**
1057 * Creates a new, completely set up worker node.
1058 *
1059 * @returns New, completely set up worker node key.
1060 */
1061 protected createAndSetupWorkerNode (): number {
1062 const worker = this.createWorker()
1063
1064 worker.on('online', this.opts.onlineHandler ?? EMPTY_FUNCTION)
1065 worker.on('message', this.opts.messageHandler ?? EMPTY_FUNCTION)
1066 worker.on('error', this.opts.errorHandler ?? EMPTY_FUNCTION)
1067 worker.on('error', (error) => {
1068 const workerNodeKey = this.getWorkerNodeKeyByWorker(worker)
1069 const workerInfo = this.getWorkerInfo(workerNodeKey)
1070 workerInfo.ready = false
1071 this.workerNodes[workerNodeKey].closeChannel()
1072 this.emitter?.emit(PoolEvents.error, error)
1073 if (
1074 this.opts.restartWorkerOnError === true &&
1075 !this.starting &&
1076 this.started
1077 ) {
1078 if (workerInfo.dynamic) {
1079 this.createAndSetupDynamicWorkerNode()
1080 } else {
1081 this.createAndSetupWorkerNode()
1082 }
1083 }
1084 if (this.opts.enableTasksQueue === true) {
1085 this.redistributeQueuedTasks(workerNodeKey)
1086 }
1087 })
1088 worker.on('exit', this.opts.exitHandler ?? EMPTY_FUNCTION)
1089 worker.once('exit', () => {
1090 this.removeWorkerNode(worker)
1091 })
1092
1093 const workerNodeKey = this.addWorkerNode(worker)
1094
1095 this.afterWorkerNodeSetup(workerNodeKey)
1096
1097 return workerNodeKey
1098 }
1099
1100 /**
1101 * Creates a new, completely set up dynamic worker node.
1102 *
1103 * @returns New, completely set up dynamic worker node key.
1104 */
1105 protected createAndSetupDynamicWorkerNode (): number {
1106 const workerNodeKey = this.createAndSetupWorkerNode()
1107 this.registerWorkerMessageListener(workerNodeKey, (message) => {
1108 const localWorkerNodeKey = this.getWorkerNodeKeyByWorkerId(
1109 message.workerId
1110 )
1111 const workerUsage = this.workerNodes[localWorkerNodeKey].usage
1112 // Kill message received from worker
1113 if (
1114 isKillBehavior(KillBehaviors.HARD, message.kill) ||
1115 (isKillBehavior(KillBehaviors.SOFT, message.kill) &&
1116 ((this.opts.enableTasksQueue === false &&
1117 workerUsage.tasks.executing === 0) ||
1118 (this.opts.enableTasksQueue === true &&
1119 workerUsage.tasks.executing === 0 &&
1120 this.tasksQueueSize(localWorkerNodeKey) === 0)))
1121 ) {
1122 this.destroyWorkerNode(localWorkerNodeKey).catch((error) => {
1123 this.emitter?.emit(PoolEvents.error, error)
1124 })
1125 }
1126 })
1127 const workerInfo = this.getWorkerInfo(workerNodeKey)
1128 this.sendToWorker(workerNodeKey, {
1129 checkActive: true,
1130 workerId: workerInfo.id as number
1131 })
1132 workerInfo.dynamic = true
1133 if (
1134 this.workerChoiceStrategyContext.getStrategyPolicy().dynamicWorkerReady ||
1135 this.workerChoiceStrategyContext.getStrategyPolicy().dynamicWorkerUsage
1136 ) {
1137 workerInfo.ready = true
1138 }
1139 this.checkAndEmitDynamicWorkerCreationEvents()
1140 return workerNodeKey
1141 }
1142
1143 /**
1144 * Registers a listener callback on the worker given its worker node key.
1145 *
1146 * @param workerNodeKey - The worker node key.
1147 * @param listener - The message listener callback.
1148 */
1149 protected abstract registerWorkerMessageListener<
1150 Message extends Data | Response
1151 >(
1152 workerNodeKey: number,
1153 listener: (message: MessageValue<Message>) => void
1154 ): void
1155
1156 /**
1157 * Method hooked up after a worker node has been newly created.
1158 * Can be overridden.
1159 *
1160 * @param workerNodeKey - The newly created worker node key.
1161 */
1162 protected afterWorkerNodeSetup (workerNodeKey: number): void {
1163 // Listen to worker messages.
1164 this.registerWorkerMessageListener(workerNodeKey, this.workerListener())
1165 // Send the startup message to worker.
1166 this.sendStartupMessageToWorker(workerNodeKey)
1167 // Send the statistics message to worker.
1168 this.sendStatisticsMessageToWorker(workerNodeKey)
1169 if (this.opts.enableTasksQueue === true) {
1170 this.workerNodes[workerNodeKey].onEmptyQueue =
1171 this.taskStealingOnEmptyQueue.bind(this)
1172 this.workerNodes[workerNodeKey].onBackPressure =
1173 this.tasksStealingOnBackPressure.bind(this)
1174 }
1175 }
1176
1177 /**
1178 * Sends the startup message to worker given its worker node key.
1179 *
1180 * @param workerNodeKey - The worker node key.
1181 */
1182 protected abstract sendStartupMessageToWorker (workerNodeKey: number): void
1183
1184 /**
1185 * Sends the statistics message to worker given its worker node key.
1186 *
1187 * @param workerNodeKey - The worker node key.
1188 */
1189 private sendStatisticsMessageToWorker (workerNodeKey: number): void {
1190 this.sendToWorker(workerNodeKey, {
1191 statistics: {
1192 runTime:
1193 this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
1194 .runTime.aggregate,
1195 elu: this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
1196 .elu.aggregate
1197 },
1198 workerId: this.getWorkerInfo(workerNodeKey).id as number
1199 })
1200 }
1201
1202 private redistributeQueuedTasks (workerNodeKey: number): void {
1203 while (this.tasksQueueSize(workerNodeKey) > 0) {
1204 const destinationWorkerNodeKey = this.workerNodes.reduce(
1205 (minWorkerNodeKey, workerNode, workerNodeKey, workerNodes) => {
1206 return workerNode.info.ready &&
1207 workerNode.usage.tasks.queued <
1208 workerNodes[minWorkerNodeKey].usage.tasks.queued
1209 ? workerNodeKey
1210 : minWorkerNodeKey
1211 },
1212 0
1213 )
1214 const destinationWorkerNode = this.workerNodes[destinationWorkerNodeKey]
1215 const task = {
1216 ...(this.dequeueTask(workerNodeKey) as Task<Data>),
1217 workerId: destinationWorkerNode.info.id as number
1218 }
1219 if (this.shallExecuteTask(destinationWorkerNodeKey)) {
1220 this.executeTask(destinationWorkerNodeKey, task)
1221 } else {
1222 this.enqueueTask(destinationWorkerNodeKey, task)
1223 }
1224 }
1225 }
1226
1227 private updateTaskStolenStatisticsWorkerUsage (
1228 workerNodeKey: number,
1229 taskName: string
1230 ): void {
1231 const workerNode = this.workerNodes[workerNodeKey]
1232 if (workerNode?.usage != null) {
1233 ++workerNode.usage.tasks.stolen
1234 }
1235 if (
1236 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) &&
1237 workerNode.getTaskFunctionWorkerUsage(taskName) != null
1238 ) {
1239 const taskFunctionWorkerUsage = workerNode.getTaskFunctionWorkerUsage(
1240 taskName
1241 ) as WorkerUsage
1242 ++taskFunctionWorkerUsage.tasks.stolen
1243 }
1244 }
1245
1246 private taskStealingOnEmptyQueue (workerId: number): void {
1247 const destinationWorkerNodeKey = this.getWorkerNodeKeyByWorkerId(workerId)
1248 const destinationWorkerNode = this.workerNodes[destinationWorkerNodeKey]
1249 const workerNodes = this.workerNodes
1250 .slice()
1251 .sort(
1252 (workerNodeA, workerNodeB) =>
1253 workerNodeB.usage.tasks.queued - workerNodeA.usage.tasks.queued
1254 )
1255 const sourceWorkerNode = workerNodes.find(
1256 (workerNode) =>
1257 workerNode.info.ready &&
1258 workerNode.info.id !== workerId &&
1259 workerNode.usage.tasks.queued > 0
1260 )
1261 if (sourceWorkerNode != null) {
1262 const task = {
1263 ...(sourceWorkerNode.popTask() as Task<Data>),
1264 workerId: destinationWorkerNode.info.id as number
1265 }
1266 if (this.shallExecuteTask(destinationWorkerNodeKey)) {
1267 this.executeTask(destinationWorkerNodeKey, task)
1268 } else {
1269 this.enqueueTask(destinationWorkerNodeKey, task)
1270 }
1271 this.updateTaskStolenStatisticsWorkerUsage(
1272 destinationWorkerNodeKey,
1273 task.name as string
1274 )
1275 }
1276 }
1277
1278 private tasksStealingOnBackPressure (workerId: number): void {
1279 const sizeOffset = 1
1280 if ((this.opts.tasksQueueOptions?.size as number) <= sizeOffset) {
1281 return
1282 }
1283 const sourceWorkerNode =
1284 this.workerNodes[this.getWorkerNodeKeyByWorkerId(workerId)]
1285 const workerNodes = this.workerNodes
1286 .slice()
1287 .sort(
1288 (workerNodeA, workerNodeB) =>
1289 workerNodeA.usage.tasks.queued - workerNodeB.usage.tasks.queued
1290 )
1291 for (const [workerNodeKey, workerNode] of workerNodes.entries()) {
1292 if (
1293 sourceWorkerNode.usage.tasks.queued > 0 &&
1294 workerNode.info.ready &&
1295 workerNode.info.id !== workerId &&
1296 workerNode.usage.tasks.queued <
1297 (this.opts.tasksQueueOptions?.size as number) - sizeOffset
1298 ) {
1299 const task = {
1300 ...(sourceWorkerNode.popTask() as Task<Data>),
1301 workerId: workerNode.info.id as number
1302 }
1303 if (this.shallExecuteTask(workerNodeKey)) {
1304 this.executeTask(workerNodeKey, task)
1305 } else {
1306 this.enqueueTask(workerNodeKey, task)
1307 }
1308 this.updateTaskStolenStatisticsWorkerUsage(
1309 workerNodeKey,
1310 task.name as string
1311 )
1312 }
1313 }
1314 }
1315
1316 /**
1317 * This method is the listener registered for each worker message.
1318 *
1319 * @returns The listener function to execute when a message is received from a worker.
1320 */
1321 protected workerListener (): (message: MessageValue<Response>) => void {
1322 return (message) => {
1323 this.checkMessageWorkerId(message)
1324 if (message.ready != null && message.taskFunctions != null) {
1325 // Worker ready response received from worker
1326 this.handleWorkerReadyResponse(message)
1327 } else if (message.taskId != null) {
1328 // Task execution response received from worker
1329 this.handleTaskExecutionResponse(message)
1330 } else if (message.taskFunctions != null) {
1331 // Task functions message received from worker
1332 this.getWorkerInfo(
1333 this.getWorkerNodeKeyByWorkerId(message.workerId)
1334 ).taskFunctions = message.taskFunctions
1335 }
1336 }
1337 }
1338
1339 private handleWorkerReadyResponse (message: MessageValue<Response>): void {
1340 if (message.ready === false) {
1341 throw new Error(`Worker ${message.workerId} failed to initialize`)
1342 }
1343 const workerInfo = this.getWorkerInfo(
1344 this.getWorkerNodeKeyByWorkerId(message.workerId)
1345 )
1346 workerInfo.ready = message.ready as boolean
1347 workerInfo.taskFunctions = message.taskFunctions
1348 if (this.emitter != null && this.ready) {
1349 this.emitter.emit(PoolEvents.ready, this.info)
1350 }
1351 }
1352
1353 private handleTaskExecutionResponse (message: MessageValue<Response>): void {
1354 const { taskId, taskError, data } = message
1355 const promiseResponse = this.promiseResponseMap.get(taskId as string)
1356 if (promiseResponse != null) {
1357 if (taskError != null) {
1358 this.emitter?.emit(PoolEvents.taskError, taskError)
1359 promiseResponse.reject(taskError.message)
1360 } else {
1361 promiseResponse.resolve(data as Response)
1362 }
1363 const workerNodeKey = promiseResponse.workerNodeKey
1364 this.afterTaskExecutionHook(workerNodeKey, message)
1365 this.workerChoiceStrategyContext.update(workerNodeKey)
1366 this.promiseResponseMap.delete(taskId as string)
1367 if (
1368 this.opts.enableTasksQueue === true &&
1369 this.tasksQueueSize(workerNodeKey) > 0 &&
1370 this.workerNodes[workerNodeKey].usage.tasks.executing <
1371 (this.opts.tasksQueueOptions?.concurrency as number)
1372 ) {
1373 this.executeTask(
1374 workerNodeKey,
1375 this.dequeueTask(workerNodeKey) as Task<Data>
1376 )
1377 }
1378 }
1379 }
1380
1381 private checkAndEmitTaskExecutionEvents (): void {
1382 if (this.busy) {
1383 this.emitter?.emit(PoolEvents.busy, this.info)
1384 }
1385 }
1386
1387 private checkAndEmitTaskQueuingEvents (): void {
1388 if (this.hasBackPressure()) {
1389 this.emitter?.emit(PoolEvents.backPressure, this.info)
1390 }
1391 }
1392
1393 private checkAndEmitDynamicWorkerCreationEvents (): void {
1394 if (this.type === PoolTypes.dynamic) {
1395 if (this.full) {
1396 this.emitter?.emit(PoolEvents.full, this.info)
1397 }
1398 }
1399 }
1400
1401 /**
1402 * Gets the worker information given its worker node key.
1403 *
1404 * @param workerNodeKey - The worker node key.
1405 * @returns The worker information.
1406 */
1407 protected getWorkerInfo (workerNodeKey: number): WorkerInfo {
1408 return this.workerNodes[workerNodeKey].info
1409 }
1410
1411 /**
1412 * Adds the given worker in the pool worker nodes.
1413 *
1414 * @param worker - The worker.
1415 * @returns The added worker node key.
1416 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the added worker node is not found.
1417 */
1418 private addWorkerNode (worker: Worker): number {
1419 const workerNode = new WorkerNode<Worker, Data>(
1420 worker,
1421 this.opts.tasksQueueOptions?.size ?? Math.pow(this.maxSize, 2)
1422 )
1423 // Flag the worker node as ready at pool startup.
1424 if (this.starting) {
1425 workerNode.info.ready = true
1426 }
1427 this.workerNodes.push(workerNode)
1428 const workerNodeKey = this.getWorkerNodeKeyByWorker(worker)
1429 if (workerNodeKey === -1) {
1430 throw new Error('Worker added not found in worker nodes')
1431 }
1432 return workerNodeKey
1433 }
1434
1435 /**
1436 * Removes the given worker from the pool worker nodes.
1437 *
1438 * @param worker - The worker.
1439 */
1440 private removeWorkerNode (worker: Worker): void {
1441 const workerNodeKey = this.getWorkerNodeKeyByWorker(worker)
1442 if (workerNodeKey !== -1) {
1443 this.workerNodes.splice(workerNodeKey, 1)
1444 this.workerChoiceStrategyContext.remove(workerNodeKey)
1445 }
1446 }
1447
1448 /** @inheritDoc */
1449 public hasWorkerNodeBackPressure (workerNodeKey: number): boolean {
1450 return (
1451 this.opts.enableTasksQueue === true &&
1452 this.workerNodes[workerNodeKey].hasBackPressure()
1453 )
1454 }
1455
1456 private hasBackPressure (): boolean {
1457 return (
1458 this.opts.enableTasksQueue === true &&
1459 this.workerNodes.findIndex(
1460 (workerNode) => !workerNode.hasBackPressure()
1461 ) === -1
1462 )
1463 }
1464
1465 /**
1466 * Executes the given task on the worker given its worker node key.
1467 *
1468 * @param workerNodeKey - The worker node key.
1469 * @param task - The task to execute.
1470 */
1471 private executeTask (workerNodeKey: number, task: Task<Data>): void {
1472 this.beforeTaskExecutionHook(workerNodeKey, task)
1473 this.sendToWorker(workerNodeKey, task, task.transferList)
1474 this.checkAndEmitTaskExecutionEvents()
1475 }
1476
1477 private enqueueTask (workerNodeKey: number, task: Task<Data>): number {
1478 const tasksQueueSize = this.workerNodes[workerNodeKey].enqueueTask(task)
1479 this.checkAndEmitTaskQueuingEvents()
1480 return tasksQueueSize
1481 }
1482
1483 private dequeueTask (workerNodeKey: number): Task<Data> | undefined {
1484 return this.workerNodes[workerNodeKey].dequeueTask()
1485 }
1486
1487 private tasksQueueSize (workerNodeKey: number): number {
1488 return this.workerNodes[workerNodeKey].tasksQueueSize()
1489 }
1490
1491 protected flushTasksQueue (workerNodeKey: number): void {
1492 while (this.tasksQueueSize(workerNodeKey) > 0) {
1493 this.executeTask(
1494 workerNodeKey,
1495 this.dequeueTask(workerNodeKey) as Task<Data>
1496 )
1497 }
1498 this.workerNodes[workerNodeKey].clearTasksQueue()
1499 }
1500
1501 private flushTasksQueues (): void {
1502 for (const [workerNodeKey] of this.workerNodes.entries()) {
1503 this.flushTasksQueue(workerNodeKey)
1504 }
1505 }
1506 }