fix: ensure worker node destroy semantic is always the same
[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 {
5 MessageValue,
6 PromiseResponseWrapper,
7 Task
8 } from '../utility-types'
9 import {
10 DEFAULT_TASK_NAME,
11 DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS,
12 EMPTY_FUNCTION,
13 isKillBehavior,
14 isPlainObject,
15 median,
16 round,
17 updateMeasurementStatistics
18 } from '../utils'
19 import { KillBehaviors } from '../worker/worker-options'
20 import {
21 type IPool,
22 PoolEmitter,
23 PoolEvents,
24 type PoolInfo,
25 type PoolOptions,
26 type PoolType,
27 PoolTypes,
28 type TasksQueueOptions
29 } from './pool'
30 import type {
31 IWorker,
32 IWorkerNode,
33 WorkerInfo,
34 WorkerType,
35 WorkerUsage
36 } from './worker'
37 import {
38 type MeasurementStatisticsRequirements,
39 Measurements,
40 WorkerChoiceStrategies,
41 type WorkerChoiceStrategy,
42 type WorkerChoiceStrategyOptions
43 } from './selection-strategies/selection-strategies-types'
44 import { WorkerChoiceStrategyContext } from './selection-strategies/worker-choice-strategy-context'
45 import { version } from './version'
46 import { WorkerNode } from './worker-node'
47
48 /**
49 * Base class that implements some shared logic for all poolifier pools.
50 *
51 * @typeParam Worker - Type of worker which manages this pool.
52 * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
53 * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
54 */
55 export abstract class AbstractPool<
56 Worker extends IWorker,
57 Data = unknown,
58 Response = unknown
59 > implements IPool<Worker, Data, Response> {
60 /** @inheritDoc */
61 public readonly workerNodes: Array<IWorkerNode<Worker, Data>> = []
62
63 /** @inheritDoc */
64 public readonly emitter?: PoolEmitter
65
66 /**
67 * The task execution response promise map.
68 *
69 * - `key`: The message id of each submitted task.
70 * - `value`: An object that contains the worker, the execution response promise resolve and reject callbacks.
71 *
72 * When we receive a message from the worker, we get a map entry with the promise resolve/reject bound to the message id.
73 */
74 protected promiseResponseMap: Map<string, PromiseResponseWrapper<Response>> =
75 new Map<string, PromiseResponseWrapper<Response>>()
76
77 /**
78 * Worker choice strategy context referencing a worker choice algorithm implementation.
79 */
80 protected workerChoiceStrategyContext: WorkerChoiceStrategyContext<
81 Worker,
82 Data,
83 Response
84 >
85
86 /**
87 * Whether the pool is starting or not.
88 */
89 private readonly starting: boolean
90 /**
91 * The start timestamp of the pool.
92 */
93 private readonly startTimestamp
94
95 /**
96 * Constructs a new poolifier pool.
97 *
98 * @param numberOfWorkers - Number of workers that this pool should manage.
99 * @param filePath - Path to the worker file.
100 * @param opts - Options for the pool.
101 */
102 public constructor (
103 protected readonly numberOfWorkers: number,
104 protected readonly filePath: string,
105 protected readonly opts: PoolOptions<Worker>
106 ) {
107 if (!this.isMain()) {
108 throw new Error('Cannot start a pool from a worker!')
109 }
110 this.checkNumberOfWorkers(this.numberOfWorkers)
111 this.checkFilePath(this.filePath)
112 this.checkPoolOptions(this.opts)
113
114 this.chooseWorkerNode = this.chooseWorkerNode.bind(this)
115 this.executeTask = this.executeTask.bind(this)
116 this.enqueueTask = this.enqueueTask.bind(this)
117 this.dequeueTask = this.dequeueTask.bind(this)
118 this.checkAndEmitEvents = this.checkAndEmitEvents.bind(this)
119
120 if (this.opts.enableEvents === true) {
121 this.emitter = new PoolEmitter()
122 }
123 this.workerChoiceStrategyContext = new WorkerChoiceStrategyContext<
124 Worker,
125 Data,
126 Response
127 >(
128 this,
129 this.opts.workerChoiceStrategy,
130 this.opts.workerChoiceStrategyOptions
131 )
132
133 this.setupHook()
134
135 this.starting = true
136 this.startPool()
137 this.starting = false
138
139 this.startTimestamp = performance.now()
140 }
141
142 private checkFilePath (filePath: string): void {
143 if (
144 filePath == null ||
145 typeof filePath !== 'string' ||
146 (typeof filePath === 'string' && filePath.trim().length === 0)
147 ) {
148 throw new Error('Please specify a file with a worker implementation')
149 }
150 if (!existsSync(filePath)) {
151 throw new Error(`Cannot find the worker file '${filePath}'`)
152 }
153 }
154
155 private checkNumberOfWorkers (numberOfWorkers: number): void {
156 if (numberOfWorkers == null) {
157 throw new Error(
158 'Cannot instantiate a pool without specifying the number of workers'
159 )
160 } else if (!Number.isSafeInteger(numberOfWorkers)) {
161 throw new TypeError(
162 'Cannot instantiate a pool with a non safe integer number of workers'
163 )
164 } else if (numberOfWorkers < 0) {
165 throw new RangeError(
166 'Cannot instantiate a pool with a negative number of workers'
167 )
168 } else if (this.type === PoolTypes.fixed && numberOfWorkers === 0) {
169 throw new RangeError('Cannot instantiate a fixed pool with zero worker')
170 }
171 }
172
173 protected checkDynamicPoolSize (min: number, max: number): void {
174 if (this.type === PoolTypes.dynamic) {
175 if (max == null) {
176 throw new Error(
177 'Cannot instantiate a dynamic pool without specifying the maximum pool size'
178 )
179 } else if (!Number.isSafeInteger(max)) {
180 throw new TypeError(
181 'Cannot instantiate a dynamic pool with a non safe integer maximum pool size'
182 )
183 } else if (min > max) {
184 throw new RangeError(
185 'Cannot instantiate a dynamic pool with a maximum pool size inferior to the minimum pool size'
186 )
187 } else if (max === 0) {
188 throw new RangeError(
189 'Cannot instantiate a dynamic pool with a pool size equal to zero'
190 )
191 } else if (min === max) {
192 throw new RangeError(
193 'Cannot instantiate a dynamic pool with a minimum pool size equal to the maximum pool size. Use a fixed pool instead'
194 )
195 }
196 }
197 }
198
199 private checkPoolOptions (opts: PoolOptions<Worker>): void {
200 if (isPlainObject(opts)) {
201 this.opts.workerChoiceStrategy =
202 opts.workerChoiceStrategy ?? WorkerChoiceStrategies.ROUND_ROBIN
203 this.checkValidWorkerChoiceStrategy(this.opts.workerChoiceStrategy)
204 this.opts.workerChoiceStrategyOptions =
205 opts.workerChoiceStrategyOptions ??
206 DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
207 this.checkValidWorkerChoiceStrategyOptions(
208 this.opts.workerChoiceStrategyOptions
209 )
210 this.opts.restartWorkerOnError = opts.restartWorkerOnError ?? true
211 this.opts.enableEvents = opts.enableEvents ?? true
212 this.opts.enableTasksQueue = opts.enableTasksQueue ?? false
213 if (this.opts.enableTasksQueue) {
214 this.checkValidTasksQueueOptions(
215 opts.tasksQueueOptions as TasksQueueOptions
216 )
217 this.opts.tasksQueueOptions = this.buildTasksQueueOptions(
218 opts.tasksQueueOptions as TasksQueueOptions
219 )
220 }
221 } else {
222 throw new TypeError('Invalid pool options: must be a plain object')
223 }
224 }
225
226 private checkValidWorkerChoiceStrategy (
227 workerChoiceStrategy: WorkerChoiceStrategy
228 ): void {
229 if (!Object.values(WorkerChoiceStrategies).includes(workerChoiceStrategy)) {
230 throw new Error(
231 `Invalid worker choice strategy '${workerChoiceStrategy}'`
232 )
233 }
234 }
235
236 private checkValidWorkerChoiceStrategyOptions (
237 workerChoiceStrategyOptions: WorkerChoiceStrategyOptions
238 ): void {
239 if (!isPlainObject(workerChoiceStrategyOptions)) {
240 throw new TypeError(
241 'Invalid worker choice strategy options: must be a plain object'
242 )
243 }
244 if (
245 workerChoiceStrategyOptions.weights != null &&
246 Object.keys(workerChoiceStrategyOptions.weights).length !== this.maxSize
247 ) {
248 throw new Error(
249 'Invalid worker choice strategy options: must have a weight for each worker node'
250 )
251 }
252 if (
253 workerChoiceStrategyOptions.measurement != null &&
254 !Object.values(Measurements).includes(
255 workerChoiceStrategyOptions.measurement
256 )
257 ) {
258 throw new Error(
259 `Invalid worker choice strategy options: invalid measurement '${workerChoiceStrategyOptions.measurement}'`
260 )
261 }
262 }
263
264 private checkValidTasksQueueOptions (
265 tasksQueueOptions: TasksQueueOptions
266 ): void {
267 if (tasksQueueOptions != null && !isPlainObject(tasksQueueOptions)) {
268 throw new TypeError('Invalid tasks queue options: must be a plain object')
269 }
270 if (
271 tasksQueueOptions?.concurrency != null &&
272 !Number.isSafeInteger(tasksQueueOptions.concurrency)
273 ) {
274 throw new TypeError(
275 'Invalid worker tasks concurrency: must be an integer'
276 )
277 }
278 if (
279 tasksQueueOptions?.concurrency != null &&
280 tasksQueueOptions.concurrency <= 0
281 ) {
282 throw new Error(
283 `Invalid worker tasks concurrency '${tasksQueueOptions.concurrency}'`
284 )
285 }
286 }
287
288 private startPool (): void {
289 while (
290 this.workerNodes.reduce(
291 (accumulator, workerNode) =>
292 !workerNode.info.dynamic ? accumulator + 1 : accumulator,
293 0
294 ) < this.numberOfWorkers
295 ) {
296 this.createAndSetupWorkerNode()
297 }
298 }
299
300 /** @inheritDoc */
301 public get info (): PoolInfo {
302 return {
303 version,
304 type: this.type,
305 worker: this.worker,
306 ready: this.ready,
307 strategy: this.opts.workerChoiceStrategy as WorkerChoiceStrategy,
308 minSize: this.minSize,
309 maxSize: this.maxSize,
310 ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
311 .runTime.aggregate &&
312 this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
313 .waitTime.aggregate && { utilization: round(this.utilization) }),
314 workerNodes: this.workerNodes.length,
315 idleWorkerNodes: this.workerNodes.reduce(
316 (accumulator, workerNode) =>
317 workerNode.usage.tasks.executing === 0
318 ? accumulator + 1
319 : accumulator,
320 0
321 ),
322 busyWorkerNodes: this.workerNodes.reduce(
323 (accumulator, workerNode) =>
324 workerNode.usage.tasks.executing > 0 ? accumulator + 1 : accumulator,
325 0
326 ),
327 executedTasks: this.workerNodes.reduce(
328 (accumulator, workerNode) =>
329 accumulator + workerNode.usage.tasks.executed,
330 0
331 ),
332 executingTasks: this.workerNodes.reduce(
333 (accumulator, workerNode) =>
334 accumulator + workerNode.usage.tasks.executing,
335 0
336 ),
337 queuedTasks: this.workerNodes.reduce(
338 (accumulator, workerNode) =>
339 accumulator + workerNode.usage.tasks.queued,
340 0
341 ),
342 maxQueuedTasks: this.workerNodes.reduce(
343 (accumulator, workerNode) =>
344 accumulator + (workerNode.usage.tasks?.maxQueued ?? 0),
345 0
346 ),
347 failedTasks: this.workerNodes.reduce(
348 (accumulator, workerNode) =>
349 accumulator + workerNode.usage.tasks.failed,
350 0
351 ),
352 ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
353 .runTime.aggregate && {
354 runTime: {
355 minimum: round(
356 Math.min(
357 ...this.workerNodes.map(
358 workerNode => workerNode.usage.runTime?.minimum ?? Infinity
359 )
360 )
361 ),
362 maximum: round(
363 Math.max(
364 ...this.workerNodes.map(
365 workerNode => workerNode.usage.runTime?.maximum ?? -Infinity
366 )
367 )
368 ),
369 average: round(
370 this.workerNodes.reduce(
371 (accumulator, workerNode) =>
372 accumulator + (workerNode.usage.runTime?.aggregate ?? 0),
373 0
374 ) /
375 this.workerNodes.reduce(
376 (accumulator, workerNode) =>
377 accumulator + (workerNode.usage.tasks?.executed ?? 0),
378 0
379 )
380 ),
381 ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
382 .runTime.median && {
383 median: round(
384 median(
385 this.workerNodes.map(
386 workerNode => workerNode.usage.runTime?.median ?? 0
387 )
388 )
389 )
390 })
391 }
392 }),
393 ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
394 .waitTime.aggregate && {
395 waitTime: {
396 minimum: round(
397 Math.min(
398 ...this.workerNodes.map(
399 workerNode => workerNode.usage.waitTime?.minimum ?? Infinity
400 )
401 )
402 ),
403 maximum: round(
404 Math.max(
405 ...this.workerNodes.map(
406 workerNode => workerNode.usage.waitTime?.maximum ?? -Infinity
407 )
408 )
409 ),
410 average: round(
411 this.workerNodes.reduce(
412 (accumulator, workerNode) =>
413 accumulator + (workerNode.usage.waitTime?.aggregate ?? 0),
414 0
415 ) /
416 this.workerNodes.reduce(
417 (accumulator, workerNode) =>
418 accumulator + (workerNode.usage.tasks?.executed ?? 0),
419 0
420 )
421 ),
422 ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
423 .waitTime.median && {
424 median: round(
425 median(
426 this.workerNodes.map(
427 workerNode => workerNode.usage.waitTime?.median ?? 0
428 )
429 )
430 )
431 })
432 }
433 })
434 }
435 }
436
437 /**
438 * The pool readiness boolean status.
439 */
440 private get ready (): boolean {
441 return (
442 this.workerNodes.reduce(
443 (accumulator, workerNode) =>
444 !workerNode.info.dynamic && workerNode.info.ready
445 ? accumulator + 1
446 : accumulator,
447 0
448 ) >= this.minSize
449 )
450 }
451
452 /**
453 * The approximate pool utilization.
454 *
455 * @returns The pool utilization.
456 */
457 private get utilization (): number {
458 const poolTimeCapacity =
459 (performance.now() - this.startTimestamp) * this.maxSize
460 const totalTasksRunTime = this.workerNodes.reduce(
461 (accumulator, workerNode) =>
462 accumulator + (workerNode.usage.runTime?.aggregate ?? 0),
463 0
464 )
465 const totalTasksWaitTime = this.workerNodes.reduce(
466 (accumulator, workerNode) =>
467 accumulator + (workerNode.usage.waitTime?.aggregate ?? 0),
468 0
469 )
470 return (totalTasksRunTime + totalTasksWaitTime) / poolTimeCapacity
471 }
472
473 /**
474 * The pool type.
475 *
476 * If it is `'dynamic'`, it provides the `max` property.
477 */
478 protected abstract get type (): PoolType
479
480 /**
481 * The worker type.
482 */
483 protected abstract get worker (): WorkerType
484
485 /**
486 * The pool minimum size.
487 */
488 protected abstract get minSize (): number
489
490 /**
491 * The pool maximum size.
492 */
493 protected abstract get maxSize (): number
494
495 /**
496 * Checks if the worker id sent in the received message from a worker is valid.
497 *
498 * @param message - The received message.
499 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the worker id is invalid.
500 */
501 private checkMessageWorkerId (message: MessageValue<Response>): void {
502 if (
503 message.workerId != null &&
504 this.getWorkerNodeKeyByWorkerId(message.workerId) === -1
505 ) {
506 throw new Error(
507 `Worker message received from unknown worker '${message.workerId}'`
508 )
509 }
510 }
511
512 /**
513 * Gets the given worker its worker node key.
514 *
515 * @param worker - The worker.
516 * @returns The worker node key if found in the pool worker nodes, `-1` otherwise.
517 */
518 private getWorkerNodeKeyByWorker (worker: Worker): number {
519 return this.workerNodes.findIndex(
520 workerNode => workerNode.worker === worker
521 )
522 }
523
524 /**
525 * Gets the worker node key given its worker id.
526 *
527 * @param workerId - The worker id.
528 * @returns The worker node key if the worker id is found in the pool worker nodes, `-1` otherwise.
529 */
530 private getWorkerNodeKeyByWorkerId (workerId: number): number {
531 return this.workerNodes.findIndex(
532 workerNode => workerNode.info.id === workerId
533 )
534 }
535
536 /** @inheritDoc */
537 public setWorkerChoiceStrategy (
538 workerChoiceStrategy: WorkerChoiceStrategy,
539 workerChoiceStrategyOptions?: WorkerChoiceStrategyOptions
540 ): void {
541 this.checkValidWorkerChoiceStrategy(workerChoiceStrategy)
542 this.opts.workerChoiceStrategy = workerChoiceStrategy
543 this.workerChoiceStrategyContext.setWorkerChoiceStrategy(
544 this.opts.workerChoiceStrategy
545 )
546 if (workerChoiceStrategyOptions != null) {
547 this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions)
548 }
549 for (const [workerNodeKey, workerNode] of this.workerNodes.entries()) {
550 workerNode.resetUsage()
551 this.sendWorkerStatisticsMessageToWorker(workerNodeKey)
552 }
553 }
554
555 /** @inheritDoc */
556 public setWorkerChoiceStrategyOptions (
557 workerChoiceStrategyOptions: WorkerChoiceStrategyOptions
558 ): void {
559 this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions)
560 this.opts.workerChoiceStrategyOptions = workerChoiceStrategyOptions
561 this.workerChoiceStrategyContext.setOptions(
562 this.opts.workerChoiceStrategyOptions
563 )
564 }
565
566 /** @inheritDoc */
567 public enableTasksQueue (
568 enable: boolean,
569 tasksQueueOptions?: TasksQueueOptions
570 ): void {
571 if (this.opts.enableTasksQueue === true && !enable) {
572 this.flushTasksQueues()
573 }
574 this.opts.enableTasksQueue = enable
575 this.setTasksQueueOptions(tasksQueueOptions as TasksQueueOptions)
576 }
577
578 /** @inheritDoc */
579 public setTasksQueueOptions (tasksQueueOptions: TasksQueueOptions): void {
580 if (this.opts.enableTasksQueue === true) {
581 this.checkValidTasksQueueOptions(tasksQueueOptions)
582 this.opts.tasksQueueOptions =
583 this.buildTasksQueueOptions(tasksQueueOptions)
584 } else if (this.opts.tasksQueueOptions != null) {
585 delete this.opts.tasksQueueOptions
586 }
587 }
588
589 private buildTasksQueueOptions (
590 tasksQueueOptions: TasksQueueOptions
591 ): TasksQueueOptions {
592 return {
593 concurrency: tasksQueueOptions?.concurrency ?? 1
594 }
595 }
596
597 /**
598 * Whether the pool is full or not.
599 *
600 * The pool filling boolean status.
601 */
602 protected get full (): boolean {
603 return this.workerNodes.length >= this.maxSize
604 }
605
606 /**
607 * Whether the pool is busy or not.
608 *
609 * The pool busyness boolean status.
610 */
611 protected abstract get busy (): boolean
612
613 /**
614 * Whether worker nodes are executing at least one task.
615 *
616 * @returns Worker nodes busyness boolean status.
617 */
618 protected internalBusy (): boolean {
619 return (
620 this.workerNodes.findIndex(workerNode => {
621 return workerNode.usage.tasks.executing === 0
622 }) === -1
623 )
624 }
625
626 /** @inheritDoc */
627 public async execute (data?: Data, name?: string): Promise<Response> {
628 return await new Promise<Response>((resolve, reject) => {
629 const timestamp = performance.now()
630 const workerNodeKey = this.chooseWorkerNode()
631 const task: Task<Data> = {
632 name: name ?? DEFAULT_TASK_NAME,
633 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
634 data: data ?? ({} as Data),
635 timestamp,
636 workerId: this.getWorkerInfo(workerNodeKey).id as number,
637 id: randomUUID()
638 }
639 this.promiseResponseMap.set(task.id as string, {
640 resolve,
641 reject,
642 workerNodeKey
643 })
644 if (
645 this.opts.enableTasksQueue === true &&
646 (this.busy ||
647 this.workerNodes[workerNodeKey].usage.tasks.executing >=
648 (this.opts.tasksQueueOptions?.concurrency as number))
649 ) {
650 this.enqueueTask(workerNodeKey, task)
651 } else {
652 this.executeTask(workerNodeKey, task)
653 }
654 this.checkAndEmitEvents()
655 })
656 }
657
658 /** @inheritDoc */
659 public async destroy (): Promise<void> {
660 await Promise.all(
661 this.workerNodes.map(async (_, workerNodeKey) => {
662 await this.destroyWorkerNode(workerNodeKey)
663 })
664 )
665 }
666
667 /**
668 * Terminates the worker node given its worker node key.
669 *
670 * @param workerNodeKey - The worker node key.
671 */
672 protected abstract destroyWorkerNode (workerNodeKey: number): Promise<void>
673
674 /**
675 * Setup hook to execute code before worker nodes are created in the abstract constructor.
676 * Can be overridden.
677 *
678 * @virtual
679 */
680 protected setupHook (): void {
681 // Intentionally empty
682 }
683
684 /**
685 * Should return whether the worker is the main worker or not.
686 */
687 protected abstract isMain (): boolean
688
689 /**
690 * Hook executed before the worker task execution.
691 * Can be overridden.
692 *
693 * @param workerNodeKey - The worker node key.
694 * @param task - The task to execute.
695 */
696 protected beforeTaskExecutionHook (
697 workerNodeKey: number,
698 task: Task<Data>
699 ): void {
700 const workerUsage = this.workerNodes[workerNodeKey].usage
701 ++workerUsage.tasks.executing
702 this.updateWaitTimeWorkerUsage(workerUsage, task)
703 const taskWorkerUsage = this.workerNodes[workerNodeKey].getTaskWorkerUsage(
704 task.name as string
705 ) as WorkerUsage
706 ++taskWorkerUsage.tasks.executing
707 this.updateWaitTimeWorkerUsage(taskWorkerUsage, task)
708 }
709
710 /**
711 * Hook executed after the worker task execution.
712 * Can be overridden.
713 *
714 * @param workerNodeKey - The worker node key.
715 * @param message - The received message.
716 */
717 protected afterTaskExecutionHook (
718 workerNodeKey: number,
719 message: MessageValue<Response>
720 ): void {
721 const workerUsage = this.workerNodes[workerNodeKey].usage
722 this.updateTaskStatisticsWorkerUsage(workerUsage, message)
723 this.updateRunTimeWorkerUsage(workerUsage, message)
724 this.updateEluWorkerUsage(workerUsage, message)
725 const taskWorkerUsage = this.workerNodes[workerNodeKey].getTaskWorkerUsage(
726 message.taskPerformance?.name ?? DEFAULT_TASK_NAME
727 ) as WorkerUsage
728 this.updateTaskStatisticsWorkerUsage(taskWorkerUsage, message)
729 this.updateRunTimeWorkerUsage(taskWorkerUsage, message)
730 this.updateEluWorkerUsage(taskWorkerUsage, message)
731 }
732
733 private updateTaskStatisticsWorkerUsage (
734 workerUsage: WorkerUsage,
735 message: MessageValue<Response>
736 ): void {
737 const workerTaskStatistics = workerUsage.tasks
738 --workerTaskStatistics.executing
739 if (message.taskError == null) {
740 ++workerTaskStatistics.executed
741 } else {
742 ++workerTaskStatistics.failed
743 }
744 }
745
746 private updateRunTimeWorkerUsage (
747 workerUsage: WorkerUsage,
748 message: MessageValue<Response>
749 ): void {
750 updateMeasurementStatistics(
751 workerUsage.runTime,
752 this.workerChoiceStrategyContext.getTaskStatisticsRequirements().runTime,
753 message.taskPerformance?.runTime ?? 0,
754 workerUsage.tasks.executed
755 )
756 }
757
758 private updateWaitTimeWorkerUsage (
759 workerUsage: WorkerUsage,
760 task: Task<Data>
761 ): void {
762 const timestamp = performance.now()
763 const taskWaitTime = timestamp - (task.timestamp ?? timestamp)
764 updateMeasurementStatistics(
765 workerUsage.waitTime,
766 this.workerChoiceStrategyContext.getTaskStatisticsRequirements().waitTime,
767 taskWaitTime,
768 workerUsage.tasks.executed
769 )
770 }
771
772 private updateEluWorkerUsage (
773 workerUsage: WorkerUsage,
774 message: MessageValue<Response>
775 ): void {
776 const eluTaskStatisticsRequirements: MeasurementStatisticsRequirements =
777 this.workerChoiceStrategyContext.getTaskStatisticsRequirements().elu
778 updateMeasurementStatistics(
779 workerUsage.elu.active,
780 eluTaskStatisticsRequirements,
781 message.taskPerformance?.elu?.active ?? 0,
782 workerUsage.tasks.executed
783 )
784 updateMeasurementStatistics(
785 workerUsage.elu.idle,
786 eluTaskStatisticsRequirements,
787 message.taskPerformance?.elu?.idle ?? 0,
788 workerUsage.tasks.executed
789 )
790 if (eluTaskStatisticsRequirements.aggregate) {
791 if (message.taskPerformance?.elu != null) {
792 if (workerUsage.elu.utilization != null) {
793 workerUsage.elu.utilization =
794 (workerUsage.elu.utilization +
795 message.taskPerformance.elu.utilization) /
796 2
797 } else {
798 workerUsage.elu.utilization = message.taskPerformance.elu.utilization
799 }
800 }
801 }
802 }
803
804 /**
805 * Chooses a worker node for the next task.
806 *
807 * The default worker choice strategy uses a round robin algorithm to distribute the tasks.
808 *
809 * @returns The chosen worker node key
810 */
811 private chooseWorkerNode (): number {
812 if (this.shallCreateDynamicWorker()) {
813 const workerNodeKey = this.createAndSetupDynamicWorkerNode()
814 if (
815 this.workerChoiceStrategyContext.getStrategyPolicy().useDynamicWorker
816 ) {
817 return workerNodeKey
818 }
819 }
820 return this.workerChoiceStrategyContext.execute()
821 }
822
823 /**
824 * Conditions for dynamic worker creation.
825 *
826 * @returns Whether to create a dynamic worker or not.
827 */
828 private shallCreateDynamicWorker (): boolean {
829 return this.type === PoolTypes.dynamic && !this.full && this.internalBusy()
830 }
831
832 /**
833 * Sends a message to worker given its worker node key.
834 *
835 * @param workerNodeKey - The worker node key.
836 * @param message - The message.
837 */
838 protected abstract sendToWorker (
839 workerNodeKey: number,
840 message: MessageValue<Data>
841 ): void
842
843 /**
844 * Creates a new worker.
845 *
846 * @returns Newly created worker.
847 */
848 protected abstract createWorker (): Worker
849
850 /**
851 * Creates a new, completely set up worker node.
852 *
853 * @returns New, completely set up worker node key.
854 */
855 protected createAndSetupWorkerNode (): number {
856 const worker = this.createWorker()
857
858 worker.on('message', this.opts.messageHandler ?? EMPTY_FUNCTION)
859 worker.on('error', this.opts.errorHandler ?? EMPTY_FUNCTION)
860 worker.on('error', error => {
861 const workerNodeKey = this.getWorkerNodeKeyByWorker(worker)
862 const workerInfo = this.getWorkerInfo(workerNodeKey)
863 workerInfo.ready = false
864 this.workerNodes[workerNodeKey].closeChannel()
865 this.emitter?.emit(PoolEvents.error, error)
866 if (this.opts.restartWorkerOnError === true && !this.starting) {
867 if (workerInfo.dynamic) {
868 this.createAndSetupDynamicWorkerNode()
869 } else {
870 this.createAndSetupWorkerNode()
871 }
872 }
873 if (this.opts.enableTasksQueue === true) {
874 this.redistributeQueuedTasks(workerNodeKey)
875 }
876 })
877 worker.on('online', this.opts.onlineHandler ?? EMPTY_FUNCTION)
878 worker.on('exit', this.opts.exitHandler ?? EMPTY_FUNCTION)
879 worker.once('exit', () => {
880 this.removeWorkerNode(worker)
881 })
882
883 const workerNodeKey = this.addWorkerNode(worker)
884
885 this.afterWorkerNodeSetup(workerNodeKey)
886
887 return workerNodeKey
888 }
889
890 /**
891 * Creates a new, completely set up dynamic worker node.
892 *
893 * @returns New, completely set up dynamic worker node key.
894 */
895 protected createAndSetupDynamicWorkerNode (): number {
896 const workerNodeKey = this.createAndSetupWorkerNode()
897 this.registerWorkerMessageListener(workerNodeKey, message => {
898 const localWorkerNodeKey = this.getWorkerNodeKeyByWorkerId(
899 message.workerId
900 )
901 const workerUsage = this.workerNodes[localWorkerNodeKey].usage
902 // Kill message received from worker
903 if (
904 isKillBehavior(KillBehaviors.HARD, message.kill) ||
905 (message.kill != null &&
906 ((this.opts.enableTasksQueue === false &&
907 workerUsage.tasks.executing === 0) ||
908 (this.opts.enableTasksQueue === true &&
909 workerUsage.tasks.executing === 0 &&
910 this.tasksQueueSize(localWorkerNodeKey) === 0)))
911 ) {
912 this.destroyWorkerNode(localWorkerNodeKey).catch(EMPTY_FUNCTION)
913 }
914 })
915 const workerInfo = this.getWorkerInfo(workerNodeKey)
916 this.sendToWorker(workerNodeKey, {
917 checkActive: true,
918 workerId: workerInfo.id as number
919 })
920 workerInfo.dynamic = true
921 if (this.workerChoiceStrategyContext.getStrategyPolicy().useDynamicWorker) {
922 workerInfo.ready = true
923 }
924 return workerNodeKey
925 }
926
927 /**
928 * Registers a listener callback on the worker given its worker node key.
929 *
930 * @param workerNodeKey - The worker node key.
931 * @param listener - The message listener callback.
932 */
933 protected abstract registerWorkerMessageListener<
934 Message extends Data | Response
935 >(
936 workerNodeKey: number,
937 listener: (message: MessageValue<Message>) => void
938 ): void
939
940 /**
941 * Method hooked up after a worker node has been newly created.
942 * Can be overridden.
943 *
944 * @param workerNodeKey - The newly created worker node key.
945 */
946 protected afterWorkerNodeSetup (workerNodeKey: number): void {
947 // Listen to worker messages.
948 this.registerWorkerMessageListener(workerNodeKey, this.workerListener())
949 // Send the startup message to worker.
950 this.sendStartupMessageToWorker(workerNodeKey)
951 // Send the worker statistics message to worker.
952 this.sendWorkerStatisticsMessageToWorker(workerNodeKey)
953 }
954
955 /**
956 * Sends the startup message to worker given its worker node key.
957 *
958 * @param workerNodeKey - The worker node key.
959 */
960 protected abstract sendStartupMessageToWorker (workerNodeKey: number): void
961
962 /**
963 * Sends the worker statistics message to worker given its worker node key.
964 *
965 * @param workerNodeKey - The worker node key.
966 */
967 private sendWorkerStatisticsMessageToWorker (workerNodeKey: number): void {
968 this.sendToWorker(workerNodeKey, {
969 statistics: {
970 runTime:
971 this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
972 .runTime.aggregate,
973 elu: this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
974 .elu.aggregate
975 },
976 workerId: this.getWorkerInfo(workerNodeKey).id as number
977 })
978 }
979
980 private redistributeQueuedTasks (workerNodeKey: number): void {
981 while (this.tasksQueueSize(workerNodeKey) > 0) {
982 let targetWorkerNodeKey: number = workerNodeKey
983 let minQueuedTasks = Infinity
984 let executeTask = false
985 for (const [workerNodeId, workerNode] of this.workerNodes.entries()) {
986 const workerInfo = this.getWorkerInfo(workerNodeId)
987 if (
988 workerNodeId !== workerNodeKey &&
989 workerInfo.ready &&
990 workerNode.usage.tasks.queued === 0
991 ) {
992 if (
993 this.workerNodes[workerNodeId].usage.tasks.executing <
994 (this.opts.tasksQueueOptions?.concurrency as number)
995 ) {
996 executeTask = true
997 }
998 targetWorkerNodeKey = workerNodeId
999 break
1000 }
1001 if (
1002 workerNodeId !== workerNodeKey &&
1003 workerInfo.ready &&
1004 workerNode.usage.tasks.queued < minQueuedTasks
1005 ) {
1006 minQueuedTasks = workerNode.usage.tasks.queued
1007 targetWorkerNodeKey = workerNodeId
1008 }
1009 }
1010 if (executeTask) {
1011 this.executeTask(
1012 targetWorkerNodeKey,
1013 this.dequeueTask(workerNodeKey) as Task<Data>
1014 )
1015 } else {
1016 this.enqueueTask(
1017 targetWorkerNodeKey,
1018 this.dequeueTask(workerNodeKey) as Task<Data>
1019 )
1020 }
1021 }
1022 }
1023
1024 /**
1025 * This method is the listener registered for each worker message.
1026 *
1027 * @returns The listener function to execute when a message is received from a worker.
1028 */
1029 protected workerListener (): (message: MessageValue<Response>) => void {
1030 return message => {
1031 this.checkMessageWorkerId(message)
1032 if (message.ready != null) {
1033 // Worker ready response received from worker
1034 this.handleWorkerReadyResponse(message)
1035 } else if (message.id != null) {
1036 // Task execution response received from worker
1037 this.handleTaskExecutionResponse(message)
1038 }
1039 }
1040 }
1041
1042 private handleWorkerReadyResponse (message: MessageValue<Response>): void {
1043 this.getWorkerInfo(
1044 this.getWorkerNodeKeyByWorkerId(message.workerId)
1045 ).ready = message.ready as boolean
1046 if (this.emitter != null && this.ready) {
1047 this.emitter.emit(PoolEvents.ready, this.info)
1048 }
1049 }
1050
1051 private handleTaskExecutionResponse (message: MessageValue<Response>): void {
1052 const promiseResponse = this.promiseResponseMap.get(message.id as string)
1053 if (promiseResponse != null) {
1054 if (message.taskError != null) {
1055 this.emitter?.emit(PoolEvents.taskError, message.taskError)
1056 promiseResponse.reject(message.taskError.message)
1057 } else {
1058 promiseResponse.resolve(message.data as Response)
1059 }
1060 const workerNodeKey = promiseResponse.workerNodeKey
1061 this.afterTaskExecutionHook(workerNodeKey, message)
1062 this.promiseResponseMap.delete(message.id as string)
1063 if (
1064 this.opts.enableTasksQueue === true &&
1065 this.tasksQueueSize(workerNodeKey) > 0 &&
1066 this.workerNodes[workerNodeKey].usage.tasks.executing <
1067 (this.opts.tasksQueueOptions?.concurrency as number)
1068 ) {
1069 this.executeTask(
1070 workerNodeKey,
1071 this.dequeueTask(workerNodeKey) as Task<Data>
1072 )
1073 }
1074 this.workerChoiceStrategyContext.update(workerNodeKey)
1075 }
1076 }
1077
1078 private checkAndEmitEvents (): void {
1079 if (this.emitter != null) {
1080 if (this.busy) {
1081 this.emitter.emit(PoolEvents.busy, this.info)
1082 }
1083 if (this.type === PoolTypes.dynamic && this.full) {
1084 this.emitter.emit(PoolEvents.full, this.info)
1085 }
1086 }
1087 }
1088
1089 /**
1090 * Gets the worker information given its worker node key.
1091 *
1092 * @param workerNodeKey - The worker node key.
1093 * @returns The worker information.
1094 */
1095 protected getWorkerInfo (workerNodeKey: number): WorkerInfo {
1096 return this.workerNodes[workerNodeKey].info
1097 }
1098
1099 /**
1100 * Adds the given worker in the pool worker nodes.
1101 *
1102 * @param worker - The worker.
1103 * @returns The added worker node key.
1104 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the added worker node is not found.
1105 */
1106 private addWorkerNode (worker: Worker): number {
1107 const workerNode = new WorkerNode<Worker, Data>(worker, this.worker)
1108 // Flag the worker node as ready at pool startup.
1109 if (this.starting) {
1110 workerNode.info.ready = true
1111 }
1112 this.workerNodes.push(workerNode)
1113 const workerNodeKey = this.getWorkerNodeKeyByWorker(worker)
1114 if (workerNodeKey === -1) {
1115 throw new Error('Worker node not found')
1116 }
1117 return workerNodeKey
1118 }
1119
1120 /**
1121 * Removes the given worker from the pool worker nodes.
1122 *
1123 * @param worker - The worker.
1124 */
1125 private removeWorkerNode (worker: Worker): void {
1126 const workerNodeKey = this.getWorkerNodeKeyByWorker(worker)
1127 if (workerNodeKey !== -1) {
1128 this.workerNodes.splice(workerNodeKey, 1)
1129 this.workerChoiceStrategyContext.remove(workerNodeKey)
1130 }
1131 }
1132
1133 /**
1134 * Executes the given task on the worker given its worker node key.
1135 *
1136 * @param workerNodeKey - The worker node key.
1137 * @param task - The task to execute.
1138 */
1139 private executeTask (workerNodeKey: number, task: Task<Data>): void {
1140 this.beforeTaskExecutionHook(workerNodeKey, task)
1141 this.sendToWorker(workerNodeKey, task)
1142 }
1143
1144 private enqueueTask (workerNodeKey: number, task: Task<Data>): number {
1145 return this.workerNodes[workerNodeKey].enqueueTask(task)
1146 }
1147
1148 private dequeueTask (workerNodeKey: number): Task<Data> | undefined {
1149 return this.workerNodes[workerNodeKey].dequeueTask()
1150 }
1151
1152 private tasksQueueSize (workerNodeKey: number): number {
1153 return this.workerNodes[workerNodeKey].tasksQueueSize()
1154 }
1155
1156 protected flushTasksQueue (workerNodeKey: number): void {
1157 while (this.tasksQueueSize(workerNodeKey) > 0) {
1158 this.executeTask(
1159 workerNodeKey,
1160 this.dequeueTask(workerNodeKey) as Task<Data>
1161 )
1162 }
1163 this.workerNodes[workerNodeKey].clearTasksQueue()
1164 }
1165
1166 private flushTasksQueues (): void {
1167 for (const [workerNodeKey] of this.workerNodes.entries()) {
1168 this.flushTasksQueue(workerNodeKey)
1169 }
1170 }
1171 }