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