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