build(deps-dev): bump rollup
[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 task: Task<Data> = {
776 name: name ?? DEFAULT_TASK_NAME,
777 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
778 data: data ?? ({} as Data),
779 transferList,
780 timestamp,
781 workerId: this.getWorkerInfo(workerNodeKey).id as number,
782 taskId: randomUUID()
783 }
784 this.promiseResponseMap.set(task.taskId as string, {
785 resolve,
786 reject,
787 workerNodeKey
788 })
789 if (
790 this.opts.enableTasksQueue === false ||
791 (this.opts.enableTasksQueue === true &&
792 this.shallExecuteTask(workerNodeKey))
793 ) {
794 this.executeTask(workerNodeKey, task)
795 } else {
796 this.enqueueTask(workerNodeKey, task)
797 }
798 })
799 }
800
801 /** @inheritDoc */
802 public async destroy (): Promise<void> {
803 await Promise.all(
804 this.workerNodes.map(async (_, workerNodeKey) => {
805 await this.destroyWorkerNode(workerNodeKey)
806 })
807 )
808 this.emitter?.emit(PoolEvents.destroy, this.info)
809 this.started = false
810 }
811
812 protected async sendKillMessageToWorker (
813 workerNodeKey: number,
814 workerId: number
815 ): Promise<void> {
816 await new Promise<void>((resolve, reject) => {
817 this.registerWorkerMessageListener(workerNodeKey, message => {
818 if (message.kill === 'success') {
819 resolve()
820 } else if (message.kill === 'failure') {
821 reject(new Error(`Worker ${workerId} kill message handling failed`))
822 }
823 })
824 this.sendToWorker(workerNodeKey, { kill: true, workerId })
825 })
826 }
827
828 /**
829 * Terminates the worker node given its worker node key.
830 *
831 * @param workerNodeKey - The worker node key.
832 */
833 protected abstract destroyWorkerNode (workerNodeKey: number): Promise<void>
834
835 /**
836 * Setup hook to execute code before worker nodes are created in the abstract constructor.
837 * Can be overridden.
838 *
839 * @virtual
840 */
841 protected setupHook (): void {
842 /* Intentionally empty */
843 }
844
845 /**
846 * Should return whether the worker is the main worker or not.
847 */
848 protected abstract isMain (): boolean
849
850 /**
851 * Hook executed before the worker task execution.
852 * Can be overridden.
853 *
854 * @param workerNodeKey - The worker node key.
855 * @param task - The task to execute.
856 */
857 protected beforeTaskExecutionHook (
858 workerNodeKey: number,
859 task: Task<Data>
860 ): void {
861 if (this.workerNodes[workerNodeKey]?.usage != null) {
862 const workerUsage = this.workerNodes[workerNodeKey].usage
863 ++workerUsage.tasks.executing
864 this.updateWaitTimeWorkerUsage(workerUsage, task)
865 }
866 if (
867 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) &&
868 this.workerNodes[workerNodeKey].getTaskFunctionWorkerUsage(
869 task.name as string
870 ) != null
871 ) {
872 const taskFunctionWorkerUsage = this.workerNodes[
873 workerNodeKey
874 ].getTaskFunctionWorkerUsage(task.name as string) as WorkerUsage
875 ++taskFunctionWorkerUsage.tasks.executing
876 this.updateWaitTimeWorkerUsage(taskFunctionWorkerUsage, task)
877 }
878 }
879
880 /**
881 * Hook executed after the worker task execution.
882 * Can be overridden.
883 *
884 * @param workerNodeKey - The worker node key.
885 * @param message - The received message.
886 */
887 protected afterTaskExecutionHook (
888 workerNodeKey: number,
889 message: MessageValue<Response>
890 ): void {
891 if (this.workerNodes[workerNodeKey]?.usage != null) {
892 const workerUsage = this.workerNodes[workerNodeKey].usage
893 this.updateTaskStatisticsWorkerUsage(workerUsage, message)
894 this.updateRunTimeWorkerUsage(workerUsage, message)
895 this.updateEluWorkerUsage(workerUsage, message)
896 }
897 if (
898 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) &&
899 this.workerNodes[workerNodeKey].getTaskFunctionWorkerUsage(
900 message.taskPerformance?.name as string
901 ) != null
902 ) {
903 const taskFunctionWorkerUsage = this.workerNodes[
904 workerNodeKey
905 ].getTaskFunctionWorkerUsage(
906 message.taskPerformance?.name as string
907 ) as WorkerUsage
908 this.updateTaskStatisticsWorkerUsage(taskFunctionWorkerUsage, message)
909 this.updateRunTimeWorkerUsage(taskFunctionWorkerUsage, message)
910 this.updateEluWorkerUsage(taskFunctionWorkerUsage, message)
911 }
912 }
913
914 /**
915 * Whether the worker node shall update its task function worker usage or not.
916 *
917 * @param workerNodeKey - The worker node key.
918 * @returns `true` if the worker node shall update its task function worker usage, `false` otherwise.
919 */
920 private shallUpdateTaskFunctionWorkerUsage (workerNodeKey: number): boolean {
921 const workerInfo = this.getWorkerInfo(workerNodeKey)
922 return (
923 workerInfo != null &&
924 Array.isArray(workerInfo.taskFunctions) &&
925 workerInfo.taskFunctions.length > 2
926 )
927 }
928
929 private updateTaskStatisticsWorkerUsage (
930 workerUsage: WorkerUsage,
931 message: MessageValue<Response>
932 ): void {
933 const workerTaskStatistics = workerUsage.tasks
934 if (
935 workerTaskStatistics.executing != null &&
936 workerTaskStatistics.executing > 0
937 ) {
938 --workerTaskStatistics.executing
939 }
940 if (message.taskError == null) {
941 ++workerTaskStatistics.executed
942 } else {
943 ++workerTaskStatistics.failed
944 }
945 }
946
947 private updateRunTimeWorkerUsage (
948 workerUsage: WorkerUsage,
949 message: MessageValue<Response>
950 ): void {
951 if (message.taskError != null) {
952 return
953 }
954 updateMeasurementStatistics(
955 workerUsage.runTime,
956 this.workerChoiceStrategyContext.getTaskStatisticsRequirements().runTime,
957 message.taskPerformance?.runTime ?? 0
958 )
959 }
960
961 private updateWaitTimeWorkerUsage (
962 workerUsage: WorkerUsage,
963 task: Task<Data>
964 ): void {
965 const timestamp = performance.now()
966 const taskWaitTime = timestamp - (task.timestamp ?? timestamp)
967 updateMeasurementStatistics(
968 workerUsage.waitTime,
969 this.workerChoiceStrategyContext.getTaskStatisticsRequirements().waitTime,
970 taskWaitTime
971 )
972 }
973
974 private updateEluWorkerUsage (
975 workerUsage: WorkerUsage,
976 message: MessageValue<Response>
977 ): void {
978 if (message.taskError != null) {
979 return
980 }
981 const eluTaskStatisticsRequirements: MeasurementStatisticsRequirements =
982 this.workerChoiceStrategyContext.getTaskStatisticsRequirements().elu
983 updateMeasurementStatistics(
984 workerUsage.elu.active,
985 eluTaskStatisticsRequirements,
986 message.taskPerformance?.elu?.active ?? 0
987 )
988 updateMeasurementStatistics(
989 workerUsage.elu.idle,
990 eluTaskStatisticsRequirements,
991 message.taskPerformance?.elu?.idle ?? 0
992 )
993 if (eluTaskStatisticsRequirements.aggregate) {
994 if (message.taskPerformance?.elu != null) {
995 if (workerUsage.elu.utilization != null) {
996 workerUsage.elu.utilization =
997 (workerUsage.elu.utilization +
998 message.taskPerformance.elu.utilization) /
999 2
1000 } else {
1001 workerUsage.elu.utilization = message.taskPerformance.elu.utilization
1002 }
1003 }
1004 }
1005 }
1006
1007 /**
1008 * Chooses a worker node for the next task.
1009 *
1010 * The default worker choice strategy uses a round robin algorithm to distribute the tasks.
1011 *
1012 * @returns The chosen worker node key
1013 */
1014 private chooseWorkerNode (): number {
1015 if (this.shallCreateDynamicWorker()) {
1016 const workerNodeKey = this.createAndSetupDynamicWorkerNode()
1017 if (
1018 this.workerChoiceStrategyContext.getStrategyPolicy().dynamicWorkerUsage
1019 ) {
1020 return workerNodeKey
1021 }
1022 }
1023 return this.workerChoiceStrategyContext.execute()
1024 }
1025
1026 /**
1027 * Conditions for dynamic worker creation.
1028 *
1029 * @returns Whether to create a dynamic worker or not.
1030 */
1031 private shallCreateDynamicWorker (): boolean {
1032 return this.type === PoolTypes.dynamic && !this.full && this.internalBusy()
1033 }
1034
1035 /**
1036 * Sends a message to worker given its worker node key.
1037 *
1038 * @param workerNodeKey - The worker node key.
1039 * @param message - The message.
1040 * @param transferList - The optional array of transferable objects.
1041 */
1042 protected abstract sendToWorker (
1043 workerNodeKey: number,
1044 message: MessageValue<Data>,
1045 transferList?: TransferListItem[]
1046 ): void
1047
1048 /**
1049 * Creates a new worker.
1050 *
1051 * @returns Newly created worker.
1052 */
1053 protected abstract createWorker (): Worker
1054
1055 /**
1056 * Creates a new, completely set up worker node.
1057 *
1058 * @returns New, completely set up worker node key.
1059 */
1060 protected createAndSetupWorkerNode (): number {
1061 const worker = this.createWorker()
1062
1063 worker.on('online', this.opts.onlineHandler ?? EMPTY_FUNCTION)
1064 worker.on('message', this.opts.messageHandler ?? EMPTY_FUNCTION)
1065 worker.on('error', this.opts.errorHandler ?? EMPTY_FUNCTION)
1066 worker.on('error', error => {
1067 const workerNodeKey = this.getWorkerNodeKeyByWorker(worker)
1068 const workerInfo = this.getWorkerInfo(workerNodeKey)
1069 workerInfo.ready = false
1070 this.workerNodes[workerNodeKey].closeChannel()
1071 this.emitter?.emit(PoolEvents.error, error)
1072 if (
1073 this.opts.restartWorkerOnError === true &&
1074 this.started &&
1075 !this.starting
1076 ) {
1077 if (workerInfo.dynamic) {
1078 this.createAndSetupDynamicWorkerNode()
1079 } else {
1080 this.createAndSetupWorkerNode()
1081 }
1082 }
1083 if (this.opts.enableTasksQueue === true) {
1084 this.redistributeQueuedTasks(workerNodeKey)
1085 }
1086 })
1087 worker.on('exit', this.opts.exitHandler ?? EMPTY_FUNCTION)
1088 worker.once('exit', () => {
1089 this.removeWorkerNode(worker)
1090 })
1091
1092 const workerNodeKey = this.addWorkerNode(worker)
1093
1094 this.afterWorkerNodeSetup(workerNodeKey)
1095
1096 return workerNodeKey
1097 }
1098
1099 /**
1100 * Creates a new, completely set up dynamic worker node.
1101 *
1102 * @returns New, completely set up dynamic worker node key.
1103 */
1104 protected createAndSetupDynamicWorkerNode (): number {
1105 const workerNodeKey = this.createAndSetupWorkerNode()
1106 this.registerWorkerMessageListener(workerNodeKey, message => {
1107 const localWorkerNodeKey = this.getWorkerNodeKeyByWorkerId(
1108 message.workerId
1109 )
1110 const workerUsage = this.workerNodes[localWorkerNodeKey].usage
1111 // Kill message received from worker
1112 if (
1113 isKillBehavior(KillBehaviors.HARD, message.kill) ||
1114 (isKillBehavior(KillBehaviors.SOFT, message.kill) &&
1115 ((this.opts.enableTasksQueue === false &&
1116 workerUsage.tasks.executing === 0) ||
1117 (this.opts.enableTasksQueue === true &&
1118 workerUsage.tasks.executing === 0 &&
1119 this.tasksQueueSize(localWorkerNodeKey) === 0)))
1120 ) {
1121 this.destroyWorkerNode(localWorkerNodeKey).catch(error => {
1122 this.emitter?.emit(PoolEvents.error, error)
1123 })
1124 }
1125 })
1126 const workerInfo = this.getWorkerInfo(workerNodeKey)
1127 this.sendToWorker(workerNodeKey, {
1128 checkActive: true,
1129 workerId: workerInfo.id as number
1130 })
1131 workerInfo.dynamic = true
1132 if (
1133 this.workerChoiceStrategyContext.getStrategyPolicy().dynamicWorkerReady ||
1134 this.workerChoiceStrategyContext.getStrategyPolicy().dynamicWorkerUsage
1135 ) {
1136 workerInfo.ready = true
1137 }
1138 this.checkAndEmitDynamicWorkerCreationEvents()
1139 return workerNodeKey
1140 }
1141
1142 /**
1143 * Registers a listener callback on the worker given its worker node key.
1144 *
1145 * @param workerNodeKey - The worker node key.
1146 * @param listener - The message listener callback.
1147 */
1148 protected abstract registerWorkerMessageListener<
1149 Message extends Data | Response
1150 >(
1151 workerNodeKey: number,
1152 listener: (message: MessageValue<Message>) => void
1153 ): void
1154
1155 /**
1156 * Method hooked up after a worker node has been newly created.
1157 * Can be overridden.
1158 *
1159 * @param workerNodeKey - The newly created worker node key.
1160 */
1161 protected afterWorkerNodeSetup (workerNodeKey: number): void {
1162 // Listen to worker messages.
1163 this.registerWorkerMessageListener(workerNodeKey, this.workerListener())
1164 // Send the startup message to worker.
1165 this.sendStartupMessageToWorker(workerNodeKey)
1166 // Send the statistics message to worker.
1167 this.sendStatisticsMessageToWorker(workerNodeKey)
1168 if (this.opts.enableTasksQueue === true) {
1169 this.workerNodes[workerNodeKey].onEmptyQueue =
1170 this.taskStealingOnEmptyQueue.bind(this)
1171 this.workerNodes[workerNodeKey].onBackPressure =
1172 this.tasksStealingOnBackPressure.bind(this)
1173 }
1174 }
1175
1176 /**
1177 * Sends the startup message to worker given its worker node key.
1178 *
1179 * @param workerNodeKey - The worker node key.
1180 */
1181 protected abstract sendStartupMessageToWorker (workerNodeKey: number): void
1182
1183 /**
1184 * Sends the statistics message to worker given its worker node key.
1185 *
1186 * @param workerNodeKey - The worker node key.
1187 */
1188 private sendStatisticsMessageToWorker (workerNodeKey: number): void {
1189 this.sendToWorker(workerNodeKey, {
1190 statistics: {
1191 runTime:
1192 this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
1193 .runTime.aggregate,
1194 elu: this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
1195 .elu.aggregate
1196 },
1197 workerId: this.getWorkerInfo(workerNodeKey).id as number
1198 })
1199 }
1200
1201 private redistributeQueuedTasks (workerNodeKey: number): void {
1202 while (this.tasksQueueSize(workerNodeKey) > 0) {
1203 const destinationWorkerNodeKey = this.workerNodes.reduce(
1204 (minWorkerNodeKey, workerNode, workerNodeKey, workerNodes) => {
1205 return workerNode.info.ready &&
1206 workerNode.usage.tasks.queued <
1207 workerNodes[minWorkerNodeKey].usage.tasks.queued
1208 ? workerNodeKey
1209 : minWorkerNodeKey
1210 },
1211 0
1212 )
1213 const destinationWorkerNode = this.workerNodes[destinationWorkerNodeKey]
1214 const task = {
1215 ...(this.dequeueTask(workerNodeKey) as Task<Data>),
1216 workerId: destinationWorkerNode.info.id as number
1217 }
1218 if (this.shallExecuteTask(destinationWorkerNodeKey)) {
1219 this.executeTask(destinationWorkerNodeKey, task)
1220 } else {
1221 this.enqueueTask(destinationWorkerNodeKey, task)
1222 }
1223 }
1224 }
1225
1226 private updateTaskStolenStatisticsWorkerUsage (
1227 workerNodeKey: number,
1228 taskName: string
1229 ): void {
1230 const workerNode = this.workerNodes[workerNodeKey]
1231 if (workerNode?.usage != null) {
1232 ++workerNode.usage.tasks.stolen
1233 }
1234 if (
1235 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) &&
1236 workerNode.getTaskFunctionWorkerUsage(taskName) != null
1237 ) {
1238 const taskFunctionWorkerUsage = workerNode.getTaskFunctionWorkerUsage(
1239 taskName
1240 ) as WorkerUsage
1241 ++taskFunctionWorkerUsage.tasks.stolen
1242 }
1243 }
1244
1245 private taskStealingOnEmptyQueue (workerId: number): void {
1246 const destinationWorkerNodeKey = this.getWorkerNodeKeyByWorkerId(workerId)
1247 const destinationWorkerNode = this.workerNodes[destinationWorkerNodeKey]
1248 const workerNodes = this.workerNodes
1249 .slice()
1250 .sort(
1251 (workerNodeA, workerNodeB) =>
1252 workerNodeB.usage.tasks.queued - workerNodeA.usage.tasks.queued
1253 )
1254 const sourceWorkerNode = workerNodes.find(
1255 workerNode =>
1256 workerNode.info.ready &&
1257 workerNode.info.id !== workerId &&
1258 workerNode.usage.tasks.queued > 0
1259 )
1260 if (sourceWorkerNode != null) {
1261 const task = {
1262 ...(sourceWorkerNode.popTask() as Task<Data>),
1263 workerId: destinationWorkerNode.info.id as number
1264 }
1265 if (this.shallExecuteTask(destinationWorkerNodeKey)) {
1266 this.executeTask(destinationWorkerNodeKey, task)
1267 } else {
1268 this.enqueueTask(destinationWorkerNodeKey, task)
1269 }
1270 this.updateTaskStolenStatisticsWorkerUsage(
1271 destinationWorkerNodeKey,
1272 task.name as string
1273 )
1274 }
1275 }
1276
1277 private tasksStealingOnBackPressure (workerId: number): void {
1278 const sizeOffset = 1
1279 if ((this.opts.tasksQueueOptions?.size as number) <= sizeOffset) {
1280 return
1281 }
1282 const sourceWorkerNode =
1283 this.workerNodes[this.getWorkerNodeKeyByWorkerId(workerId)]
1284 const workerNodes = this.workerNodes
1285 .slice()
1286 .sort(
1287 (workerNodeA, workerNodeB) =>
1288 workerNodeA.usage.tasks.queued - workerNodeB.usage.tasks.queued
1289 )
1290 for (const [workerNodeKey, workerNode] of workerNodes.entries()) {
1291 if (
1292 sourceWorkerNode.usage.tasks.queued > 0 &&
1293 workerNode.info.ready &&
1294 workerNode.info.id !== workerId &&
1295 workerNode.usage.tasks.queued <
1296 (this.opts.tasksQueueOptions?.size as number) - sizeOffset
1297 ) {
1298 const task = {
1299 ...(sourceWorkerNode.popTask() as Task<Data>),
1300 workerId: workerNode.info.id as number
1301 }
1302 if (this.shallExecuteTask(workerNodeKey)) {
1303 this.executeTask(workerNodeKey, task)
1304 } else {
1305 this.enqueueTask(workerNodeKey, task)
1306 }
1307 this.updateTaskStolenStatisticsWorkerUsage(
1308 workerNodeKey,
1309 task.name as string
1310 )
1311 }
1312 }
1313 }
1314
1315 /**
1316 * This method is the listener registered for each worker message.
1317 *
1318 * @returns The listener function to execute when a message is received from a worker.
1319 */
1320 protected workerListener (): (message: MessageValue<Response>) => void {
1321 return message => {
1322 this.checkMessageWorkerId(message)
1323 if (message.ready != null && message.taskFunctions != null) {
1324 // Worker ready response received from worker
1325 this.handleWorkerReadyResponse(message)
1326 } else if (message.taskId != null) {
1327 // Task execution response received from worker
1328 this.handleTaskExecutionResponse(message)
1329 } else if (message.taskFunctions != null) {
1330 // Task functions message received from worker
1331 this.getWorkerInfo(
1332 this.getWorkerNodeKeyByWorkerId(message.workerId)
1333 ).taskFunctions = message.taskFunctions
1334 }
1335 }
1336 }
1337
1338 private handleWorkerReadyResponse (message: MessageValue<Response>): void {
1339 if (message.ready === false) {
1340 throw new Error(`Worker ${message.workerId} failed to initialize`)
1341 }
1342 const workerInfo = this.getWorkerInfo(
1343 this.getWorkerNodeKeyByWorkerId(message.workerId)
1344 )
1345 workerInfo.ready = message.ready as boolean
1346 workerInfo.taskFunctions = message.taskFunctions
1347 if (this.emitter != null && this.ready) {
1348 this.emitter.emit(PoolEvents.ready, this.info)
1349 }
1350 }
1351
1352 private handleTaskExecutionResponse (message: MessageValue<Response>): void {
1353 const { taskId, taskError, data } = message
1354 const promiseResponse = this.promiseResponseMap.get(taskId as string)
1355 if (promiseResponse != null) {
1356 if (taskError != null) {
1357 this.emitter?.emit(PoolEvents.taskError, taskError)
1358 promiseResponse.reject(taskError.message)
1359 } else {
1360 promiseResponse.resolve(data as Response)
1361 }
1362 const workerNodeKey = promiseResponse.workerNodeKey
1363 this.afterTaskExecutionHook(workerNodeKey, message)
1364 this.workerChoiceStrategyContext.update(workerNodeKey)
1365 this.promiseResponseMap.delete(taskId as string)
1366 if (
1367 this.opts.enableTasksQueue === true &&
1368 this.tasksQueueSize(workerNodeKey) > 0 &&
1369 this.workerNodes[workerNodeKey].usage.tasks.executing <
1370 (this.opts.tasksQueueOptions?.concurrency as number)
1371 ) {
1372 this.executeTask(
1373 workerNodeKey,
1374 this.dequeueTask(workerNodeKey) as Task<Data>
1375 )
1376 }
1377 }
1378 }
1379
1380 private checkAndEmitTaskExecutionEvents (): void {
1381 if (this.busy) {
1382 this.emitter?.emit(PoolEvents.busy, this.info)
1383 }
1384 }
1385
1386 private checkAndEmitTaskQueuingEvents (): void {
1387 if (this.hasBackPressure()) {
1388 this.emitter?.emit(PoolEvents.backPressure, this.info)
1389 }
1390 }
1391
1392 private checkAndEmitDynamicWorkerCreationEvents (): void {
1393 if (this.type === PoolTypes.dynamic) {
1394 if (this.full) {
1395 this.emitter?.emit(PoolEvents.full, this.info)
1396 }
1397 }
1398 }
1399
1400 /**
1401 * Gets the worker information given its worker node key.
1402 *
1403 * @param workerNodeKey - The worker node key.
1404 * @returns The worker information.
1405 */
1406 protected getWorkerInfo (workerNodeKey: number): WorkerInfo {
1407 return this.workerNodes[workerNodeKey].info
1408 }
1409
1410 /**
1411 * Adds the given worker in the pool worker nodes.
1412 *
1413 * @param worker - The worker.
1414 * @returns The added worker node key.
1415 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the added worker node is not found.
1416 */
1417 private addWorkerNode (worker: Worker): number {
1418 const workerNode = new WorkerNode<Worker, Data>(
1419 worker,
1420 this.opts.tasksQueueOptions?.size ?? Math.pow(this.maxSize, 2)
1421 )
1422 // Flag the worker node as ready at pool startup.
1423 if (this.starting) {
1424 workerNode.info.ready = true
1425 }
1426 this.workerNodes.push(workerNode)
1427 const workerNodeKey = this.getWorkerNodeKeyByWorker(worker)
1428 if (workerNodeKey === -1) {
1429 throw new Error('Worker added not found in worker nodes')
1430 }
1431 return workerNodeKey
1432 }
1433
1434 /**
1435 * Removes the given worker from the pool worker nodes.
1436 *
1437 * @param worker - The worker.
1438 */
1439 private removeWorkerNode (worker: Worker): void {
1440 const workerNodeKey = this.getWorkerNodeKeyByWorker(worker)
1441 if (workerNodeKey !== -1) {
1442 this.workerNodes.splice(workerNodeKey, 1)
1443 this.workerChoiceStrategyContext.remove(workerNodeKey)
1444 }
1445 }
1446
1447 /** @inheritDoc */
1448 public hasWorkerNodeBackPressure (workerNodeKey: number): boolean {
1449 return (
1450 this.opts.enableTasksQueue === true &&
1451 this.workerNodes[workerNodeKey].hasBackPressure()
1452 )
1453 }
1454
1455 private hasBackPressure (): boolean {
1456 return (
1457 this.opts.enableTasksQueue === true &&
1458 this.workerNodes.findIndex(
1459 workerNode => !workerNode.hasBackPressure()
1460 ) === -1
1461 )
1462 }
1463
1464 /**
1465 * Executes the given task on the worker given its worker node key.
1466 *
1467 * @param workerNodeKey - The worker node key.
1468 * @param task - The task to execute.
1469 */
1470 private executeTask (workerNodeKey: number, task: Task<Data>): void {
1471 this.beforeTaskExecutionHook(workerNodeKey, task)
1472 this.sendToWorker(workerNodeKey, task, task.transferList)
1473 this.checkAndEmitTaskExecutionEvents()
1474 }
1475
1476 private enqueueTask (workerNodeKey: number, task: Task<Data>): number {
1477 const tasksQueueSize = this.workerNodes[workerNodeKey].enqueueTask(task)
1478 this.checkAndEmitTaskQueuingEvents()
1479 return tasksQueueSize
1480 }
1481
1482 private dequeueTask (workerNodeKey: number): Task<Data> | undefined {
1483 return this.workerNodes[workerNodeKey].dequeueTask()
1484 }
1485
1486 private tasksQueueSize (workerNodeKey: number): number {
1487 return this.workerNodes[workerNodeKey].tasksQueueSize()
1488 }
1489
1490 protected flushTasksQueue (workerNodeKey: number): void {
1491 while (this.tasksQueueSize(workerNodeKey) > 0) {
1492 this.executeTask(
1493 workerNodeKey,
1494 this.dequeueTask(workerNodeKey) as Task<Data>
1495 )
1496 }
1497 this.workerNodes[workerNodeKey].clearTasksQueue()
1498 }
1499
1500 private flushTasksQueues (): void {
1501 for (const [workerNodeKey] of this.workerNodes.entries()) {
1502 this.flushTasksQueue(workerNodeKey)
1503 }
1504 }
1505 }