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