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