refactor: cleanup handleWorkerNodeIdleEvent()
[poolifier.git] / src / pools / abstract-pool.ts
CommitLineData
ded253e2 1import { AsyncResource } from 'node:async_hooks'
2845f2a5 2import { randomUUID } from 'node:crypto'
ded253e2 3import { EventEmitterAsyncResource } from 'node:events'
62c15a68 4import { performance } from 'node:perf_hooks'
ef083f7b 5import type { TransferListItem } from 'node:worker_threads'
ded253e2 6
5c4d16da
JB
7import type {
8 MessageValue,
9 PromiseResponseWrapper,
31847469
JB
10 Task,
11 TaskFunctionProperties
d35e5717 12} from '../utility-types.js'
bbeadd16 13import {
ded253e2 14 average,
31847469 15 buildTaskFunctionProperties,
ff128cc9 16 DEFAULT_TASK_NAME,
bbeadd16 17 EMPTY_FUNCTION,
463226a4 18 exponentialDelay,
59317253 19 isKillBehavior,
0d80593b 20 isPlainObject,
90d6701c 21 max,
afe0d5bf 22 median,
90d6701c 23 min,
463226a4
JB
24 round,
25 sleep
d35e5717 26} from '../utils.js'
31847469
JB
27import type {
28 TaskFunction,
29 TaskFunctionObject
30} from '../worker/task-functions.js'
ded253e2 31import { KillBehaviors } from '../worker/worker-options.js'
c4855468 32import {
65d7a1c9 33 type IPool,
c4855468 34 PoolEvents,
6b27d407 35 type PoolInfo,
c4855468 36 type PoolOptions,
6b27d407
JB
37 type PoolType,
38 PoolTypes,
4b628b48 39 type TasksQueueOptions
d35e5717 40} from './pool.js'
a35560ba 41import {
f0d7f803 42 Measurements,
a35560ba 43 WorkerChoiceStrategies,
a20f0ba5
JB
44 type WorkerChoiceStrategy,
45 type WorkerChoiceStrategyOptions
d35e5717 46} from './selection-strategies/selection-strategies-types.js'
bcfb06ce 47import { WorkerChoiceStrategiesContext } from './selection-strategies/worker-choice-strategies-context.js'
bde6b5d7
JB
48import {
49 checkFilePath,
95d1a734 50 checkValidPriority,
bde6b5d7 51 checkValidTasksQueueOptions,
bfc75cca 52 checkValidWorkerChoiceStrategy,
32b141fd 53 getDefaultTasksQueueOptions,
c329fd41
JB
54 updateEluWorkerUsage,
55 updateRunTimeWorkerUsage,
56 updateTaskStatisticsWorkerUsage,
57 updateWaitTimeWorkerUsage,
87347ea8 58 waitWorkerNodeEvents
d35e5717 59} from './utils.js'
ded253e2
JB
60import { version } from './version.js'
61import type {
62 IWorker,
63 IWorkerNode,
64 WorkerInfo,
65 WorkerNodeEventDetail,
66 WorkerType
67} from './worker.js'
68import { WorkerNode } from './worker-node.js'
23ccf9d7 69
729c563d 70/**
ea7a90d3 71 * Base class that implements some shared logic for all poolifier pools.
729c563d 72 *
38e795c1 73 * @typeParam Worker - Type of worker which manages this pool.
e102732c
JB
74 * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
75 * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
729c563d 76 */
c97c7edb 77export abstract class AbstractPool<
f06e48d8 78 Worker extends IWorker,
d3c8a1a8
S
79 Data = unknown,
80 Response = unknown
c4855468 81> implements IPool<Worker, Data, Response> {
afc003b2 82 /** @inheritDoc */
4b628b48 83 public readonly workerNodes: Array<IWorkerNode<Worker, Data>> = []
4a6952ff 84
afc003b2 85 /** @inheritDoc */
f80125ca 86 public emitter?: EventEmitterAsyncResource
7c0ba920 87
be0676b3 88 /**
419e8121 89 * The task execution response promise map:
2740a743 90 * - `key`: The message id of each submitted task.
bcfb06ce 91 * - `value`: An object that contains task's worker node key, execution response promise resolve and reject callbacks, async resource.
be0676b3 92 *
a3445496 93 * When we receive a message from the worker, we get a map entry with the promise resolve/reject bound to the message id.
be0676b3 94 */
84781cd6
JB
95 protected promiseResponseMap: Map<
96 `${string}-${string}-${string}-${string}-${string}`,
97 PromiseResponseWrapper<Response>
98 > = new Map<
99 `${string}-${string}-${string}-${string}-${string}`,
100 PromiseResponseWrapper<Response>
101 >()
c97c7edb 102
a35560ba 103 /**
bcfb06ce 104 * Worker choice strategies context referencing worker choice algorithms implementation.
a35560ba 105 */
bcfb06ce 106 protected workerChoiceStrategiesContext?: WorkerChoiceStrategiesContext<
78cea37e
JB
107 Worker,
108 Data,
109 Response
a35560ba
S
110 >
111
72ae84a2
JB
112 /**
113 * The task functions added at runtime map:
114 * - `key`: The task function name.
31847469 115 * - `value`: The task function object.
72ae84a2 116 */
31847469
JB
117 private readonly taskFunctions: Map<
118 string,
119 TaskFunctionObject<Data, Response>
120 >
72ae84a2 121
15b176e0
JB
122 /**
123 * Whether the pool is started or not.
124 */
125 private started: boolean
47352846
JB
126 /**
127 * Whether the pool is starting or not.
128 */
129 private starting: boolean
711623b8
JB
130 /**
131 * Whether the pool is destroying or not.
132 */
133 private destroying: boolean
b362a929
JB
134 /**
135 * Whether the minimum number of workers is starting or not.
136 */
137 private startingMinimumNumberOfWorkers: boolean
55082af9
JB
138 /**
139 * Whether the pool ready event has been emitted or not.
140 */
141 private readyEventEmitted: boolean
afe0d5bf
JB
142 /**
143 * The start timestamp of the pool.
144 */
97dc65d9 145 private startTimestamp?: number
afe0d5bf 146
729c563d
S
147 /**
148 * Constructs a new poolifier pool.
149 *
00e1bdeb 150 * @param minimumNumberOfWorkers - Minimum number of workers that this pool manages.
029715f0 151 * @param filePath - Path to the worker file.
38e795c1 152 * @param opts - Options for the pool.
00e1bdeb 153 * @param maximumNumberOfWorkers - Maximum number of workers that this pool manages.
729c563d 154 */
c97c7edb 155 public constructor (
26ce26ca 156 protected readonly minimumNumberOfWorkers: number,
b4213b7f 157 protected readonly filePath: string,
26ce26ca
JB
158 protected readonly opts: PoolOptions<Worker>,
159 protected readonly maximumNumberOfWorkers?: number
c97c7edb 160 ) {
78cea37e 161 if (!this.isMain()) {
04f45163 162 throw new Error(
8c6d4acf 163 'Cannot start a pool from a worker with the same type as the pool'
04f45163 164 )
c97c7edb 165 }
26ce26ca 166 this.checkPoolType()
bde6b5d7 167 checkFilePath(this.filePath)
26ce26ca 168 this.checkMinimumNumberOfWorkers(this.minimumNumberOfWorkers)
7c0ba920 169 this.checkPoolOptions(this.opts)
1086026a 170
7254e419
JB
171 this.chooseWorkerNode = this.chooseWorkerNode.bind(this)
172 this.executeTask = this.executeTask.bind(this)
173 this.enqueueTask = this.enqueueTask.bind(this)
1086026a 174
6bd72cd0 175 if (this.opts.enableEvents === true) {
e0843544 176 this.initEventEmitter()
7c0ba920 177 }
bcfb06ce 178 this.workerChoiceStrategiesContext = new WorkerChoiceStrategiesContext<
d59df138
JB
179 Worker,
180 Data,
181 Response
da309861
JB
182 >(
183 this,
bcfb06ce
JB
184 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
185 [this.opts.workerChoiceStrategy!],
da309861
JB
186 this.opts.workerChoiceStrategyOptions
187 )
b6b32453
JB
188
189 this.setupHook()
190
31847469 191 this.taskFunctions = new Map<string, TaskFunctionObject<Data, Response>>()
72ae84a2 192
47352846 193 this.started = false
075e51d1 194 this.starting = false
711623b8 195 this.destroying = false
55082af9 196 this.readyEventEmitted = false
b362a929 197 this.startingMinimumNumberOfWorkers = false
47352846
JB
198 if (this.opts.startWorkers === true) {
199 this.start()
200 }
c97c7edb
S
201 }
202
26ce26ca
JB
203 private checkPoolType (): void {
204 if (this.type === PoolTypes.fixed && this.maximumNumberOfWorkers != null) {
205 throw new Error(
206 'Cannot instantiate a fixed pool with a maximum number of workers specified at initialization'
207 )
208 }
209 }
210
c63a35a0
JB
211 private checkMinimumNumberOfWorkers (
212 minimumNumberOfWorkers: number | undefined
213 ): void {
af7f2788 214 if (minimumNumberOfWorkers == null) {
8d3782fa
JB
215 throw new Error(
216 'Cannot instantiate a pool without specifying the number of workers'
217 )
af7f2788 218 } else if (!Number.isSafeInteger(minimumNumberOfWorkers)) {
473c717a 219 throw new TypeError(
0d80593b 220 'Cannot instantiate a pool with a non safe integer number of workers'
8d3782fa 221 )
af7f2788 222 } else if (minimumNumberOfWorkers < 0) {
473c717a 223 throw new RangeError(
8d3782fa
JB
224 'Cannot instantiate a pool with a negative number of workers'
225 )
af7f2788 226 } else if (this.type === PoolTypes.fixed && minimumNumberOfWorkers === 0) {
2431bdb4
JB
227 throw new RangeError('Cannot instantiate a fixed pool with zero worker')
228 }
229 }
230
7c0ba920 231 private checkPoolOptions (opts: PoolOptions<Worker>): void {
0d80593b 232 if (isPlainObject(opts)) {
47352846 233 this.opts.startWorkers = opts.startWorkers ?? true
c63a35a0 234 checkValidWorkerChoiceStrategy(opts.workerChoiceStrategy)
0d80593b
JB
235 this.opts.workerChoiceStrategy =
236 opts.workerChoiceStrategy ?? WorkerChoiceStrategies.ROUND_ROBIN
2324f8c9 237 this.checkValidWorkerChoiceStrategyOptions(
c63a35a0 238 opts.workerChoiceStrategyOptions
2324f8c9 239 )
26ce26ca
JB
240 if (opts.workerChoiceStrategyOptions != null) {
241 this.opts.workerChoiceStrategyOptions = opts.workerChoiceStrategyOptions
8990357d 242 }
1f68cede 243 this.opts.restartWorkerOnError = opts.restartWorkerOnError ?? true
0d80593b
JB
244 this.opts.enableEvents = opts.enableEvents ?? true
245 this.opts.enableTasksQueue = opts.enableTasksQueue ?? false
246 if (this.opts.enableTasksQueue) {
c63a35a0 247 checkValidTasksQueueOptions(opts.tasksQueueOptions)
0d80593b 248 this.opts.tasksQueueOptions = this.buildTasksQueueOptions(
c63a35a0 249 opts.tasksQueueOptions
0d80593b
JB
250 )
251 }
252 } else {
253 throw new TypeError('Invalid pool options: must be a plain object')
7171d33f 254 }
aee46736
JB
255 }
256
0d80593b 257 private checkValidWorkerChoiceStrategyOptions (
c63a35a0 258 workerChoiceStrategyOptions: WorkerChoiceStrategyOptions | undefined
0d80593b 259 ): void {
2324f8c9
JB
260 if (
261 workerChoiceStrategyOptions != null &&
262 !isPlainObject(workerChoiceStrategyOptions)
263 ) {
0d80593b
JB
264 throw new TypeError(
265 'Invalid worker choice strategy options: must be a plain object'
266 )
267 }
49be33fe 268 if (
2324f8c9 269 workerChoiceStrategyOptions?.weights != null &&
26ce26ca
JB
270 Object.keys(workerChoiceStrategyOptions.weights).length !==
271 (this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers)
49be33fe
JB
272 ) {
273 throw new Error(
274 'Invalid worker choice strategy options: must have a weight for each worker node'
275 )
276 }
f0d7f803 277 if (
2324f8c9 278 workerChoiceStrategyOptions?.measurement != null &&
f0d7f803
JB
279 !Object.values(Measurements).includes(
280 workerChoiceStrategyOptions.measurement
281 )
282 ) {
283 throw new Error(
284 `Invalid worker choice strategy options: invalid measurement '${workerChoiceStrategyOptions.measurement}'`
285 )
286 }
0d80593b
JB
287 }
288
e0843544 289 private initEventEmitter (): void {
5b7d00b4 290 this.emitter = new EventEmitterAsyncResource({
b5604034
JB
291 name: `poolifier:${this.type}-${this.worker}-pool`
292 })
293 }
294
08f3f44c 295 /** @inheritDoc */
6b27d407
JB
296 public get info (): PoolInfo {
297 return {
23ccf9d7 298 version,
6b27d407 299 type: this.type,
184855e6 300 worker: this.worker,
47352846 301 started: this.started,
2431bdb4 302 ready: this.ready,
67f3f2d6 303 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
bcfb06ce
JB
304 defaultStrategy: this.opts.workerChoiceStrategy!,
305 strategyRetries: this.workerChoiceStrategiesContext?.retriesCount ?? 0,
26ce26ca
JB
306 minSize: this.minimumNumberOfWorkers,
307 maxSize: this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers,
bcfb06ce 308 ...(this.workerChoiceStrategiesContext?.getTaskStatisticsRequirements()
f4d0a470 309 .runTime.aggregate === true &&
bcfb06ce 310 this.workerChoiceStrategiesContext.getTaskStatisticsRequirements()
c63a35a0
JB
311 .waitTime.aggregate && {
312 utilization: round(this.utilization)
313 }),
6b27d407
JB
314 workerNodes: this.workerNodes.length,
315 idleWorkerNodes: this.workerNodes.reduce(
316 (accumulator, workerNode) =>
f59e1027 317 workerNode.usage.tasks.executing === 0
a4e07f72
JB
318 ? accumulator + 1
319 : accumulator,
6b27d407
JB
320 0
321 ),
5eb72b9e
JB
322 ...(this.opts.enableTasksQueue === true && {
323 stealingWorkerNodes: this.workerNodes.reduce(
324 (accumulator, workerNode) =>
325 workerNode.info.stealing ? accumulator + 1 : accumulator,
326 0
327 )
328 }),
6b27d407 329 busyWorkerNodes: this.workerNodes.reduce(
533a8e22 330 (accumulator, _, workerNodeKey) =>
42c677c1 331 this.isWorkerNodeBusy(workerNodeKey) ? accumulator + 1 : accumulator,
6b27d407
JB
332 0
333 ),
a4e07f72 334 executedTasks: this.workerNodes.reduce(
6b27d407 335 (accumulator, workerNode) =>
f59e1027 336 accumulator + workerNode.usage.tasks.executed,
a4e07f72
JB
337 0
338 ),
339 executingTasks: this.workerNodes.reduce(
340 (accumulator, workerNode) =>
f59e1027 341 accumulator + workerNode.usage.tasks.executing,
6b27d407
JB
342 0
343 ),
daf86646
JB
344 ...(this.opts.enableTasksQueue === true && {
345 queuedTasks: this.workerNodes.reduce(
346 (accumulator, workerNode) =>
347 accumulator + workerNode.usage.tasks.queued,
348 0
349 )
350 }),
351 ...(this.opts.enableTasksQueue === true && {
352 maxQueuedTasks: this.workerNodes.reduce(
353 (accumulator, workerNode) =>
c63a35a0 354 accumulator + (workerNode.usage.tasks.maxQueued ?? 0),
daf86646
JB
355 0
356 )
357 }),
a1763c54
JB
358 ...(this.opts.enableTasksQueue === true && {
359 backPressure: this.hasBackPressure()
360 }),
68cbdc84
JB
361 ...(this.opts.enableTasksQueue === true && {
362 stolenTasks: this.workerNodes.reduce(
363 (accumulator, workerNode) =>
364 accumulator + workerNode.usage.tasks.stolen,
365 0
366 )
367 }),
a4e07f72
JB
368 failedTasks: this.workerNodes.reduce(
369 (accumulator, workerNode) =>
f59e1027 370 accumulator + workerNode.usage.tasks.failed,
a4e07f72 371 0
1dcf8b7b 372 ),
bcfb06ce 373 ...(this.workerChoiceStrategiesContext?.getTaskStatisticsRequirements()
f4d0a470 374 .runTime.aggregate === true && {
1dcf8b7b 375 runTime: {
98e72cda 376 minimum: round(
90d6701c 377 min(
98e72cda 378 ...this.workerNodes.map(
c63a35a0 379 workerNode => workerNode.usage.runTime.minimum ?? Infinity
98e72cda 380 )
1dcf8b7b
JB
381 )
382 ),
98e72cda 383 maximum: round(
90d6701c 384 max(
98e72cda 385 ...this.workerNodes.map(
c63a35a0 386 workerNode => workerNode.usage.runTime.maximum ?? -Infinity
98e72cda 387 )
1dcf8b7b 388 )
98e72cda 389 ),
bcfb06ce 390 ...(this.workerChoiceStrategiesContext.getTaskStatisticsRequirements()
3baa0837
JB
391 .runTime.average && {
392 average: round(
393 average(
394 this.workerNodes.reduce<number[]>(
395 (accumulator, workerNode) =>
396 accumulator.concat(workerNode.usage.runTime.history),
397 []
398 )
98e72cda 399 )
dc021bcc 400 )
3baa0837 401 }),
bcfb06ce 402 ...(this.workerChoiceStrategiesContext.getTaskStatisticsRequirements()
98e72cda
JB
403 .runTime.median && {
404 median: round(
405 median(
3baa0837
JB
406 this.workerNodes.reduce<number[]>(
407 (accumulator, workerNode) =>
408 accumulator.concat(workerNode.usage.runTime.history),
409 []
98e72cda
JB
410 )
411 )
412 )
413 })
1dcf8b7b
JB
414 }
415 }),
bcfb06ce 416 ...(this.workerChoiceStrategiesContext?.getTaskStatisticsRequirements()
f4d0a470 417 .waitTime.aggregate === true && {
1dcf8b7b 418 waitTime: {
98e72cda 419 minimum: round(
90d6701c 420 min(
98e72cda 421 ...this.workerNodes.map(
c63a35a0 422 workerNode => workerNode.usage.waitTime.minimum ?? Infinity
98e72cda 423 )
1dcf8b7b
JB
424 )
425 ),
98e72cda 426 maximum: round(
90d6701c 427 max(
98e72cda 428 ...this.workerNodes.map(
c63a35a0 429 workerNode => workerNode.usage.waitTime.maximum ?? -Infinity
98e72cda 430 )
1dcf8b7b 431 )
98e72cda 432 ),
bcfb06ce 433 ...(this.workerChoiceStrategiesContext.getTaskStatisticsRequirements()
3baa0837
JB
434 .waitTime.average && {
435 average: round(
436 average(
437 this.workerNodes.reduce<number[]>(
438 (accumulator, workerNode) =>
439 accumulator.concat(workerNode.usage.waitTime.history),
440 []
441 )
98e72cda 442 )
dc021bcc 443 )
3baa0837 444 }),
bcfb06ce 445 ...(this.workerChoiceStrategiesContext.getTaskStatisticsRequirements()
98e72cda
JB
446 .waitTime.median && {
447 median: round(
448 median(
3baa0837
JB
449 this.workerNodes.reduce<number[]>(
450 (accumulator, workerNode) =>
451 accumulator.concat(workerNode.usage.waitTime.history),
452 []
98e72cda
JB
453 )
454 )
455 )
456 })
1dcf8b7b 457 }
533a8e22
JB
458 }),
459 ...(this.workerChoiceStrategiesContext?.getTaskStatisticsRequirements()
460 .elu.aggregate === true && {
461 elu: {
462 idle: {
463 minimum: round(
464 min(
465 ...this.workerNodes.map(
466 workerNode => workerNode.usage.elu.idle.minimum ?? Infinity
467 )
468 )
469 ),
470 maximum: round(
471 max(
472 ...this.workerNodes.map(
473 workerNode => workerNode.usage.elu.idle.maximum ?? -Infinity
474 )
475 )
476 ),
477 ...(this.workerChoiceStrategiesContext.getTaskStatisticsRequirements()
478 .elu.average && {
479 average: round(
480 average(
481 this.workerNodes.reduce<number[]>(
482 (accumulator, workerNode) =>
483 accumulator.concat(workerNode.usage.elu.idle.history),
484 []
485 )
486 )
487 )
488 }),
489 ...(this.workerChoiceStrategiesContext.getTaskStatisticsRequirements()
490 .elu.median && {
491 median: round(
492 median(
493 this.workerNodes.reduce<number[]>(
494 (accumulator, workerNode) =>
495 accumulator.concat(workerNode.usage.elu.idle.history),
496 []
497 )
498 )
499 )
500 })
501 },
502 active: {
503 minimum: round(
504 min(
505 ...this.workerNodes.map(
506 workerNode => workerNode.usage.elu.active.minimum ?? Infinity
507 )
508 )
509 ),
510 maximum: round(
511 max(
512 ...this.workerNodes.map(
513 workerNode => workerNode.usage.elu.active.maximum ?? -Infinity
514 )
515 )
516 ),
517 ...(this.workerChoiceStrategiesContext.getTaskStatisticsRequirements()
518 .elu.average && {
519 average: round(
520 average(
521 this.workerNodes.reduce<number[]>(
522 (accumulator, workerNode) =>
523 accumulator.concat(workerNode.usage.elu.active.history),
524 []
525 )
526 )
527 )
528 }),
529 ...(this.workerChoiceStrategiesContext.getTaskStatisticsRequirements()
530 .elu.median && {
531 median: round(
532 median(
533 this.workerNodes.reduce<number[]>(
534 (accumulator, workerNode) =>
535 accumulator.concat(workerNode.usage.elu.active.history),
536 []
537 )
538 )
539 )
540 })
541 }
542 }
1dcf8b7b 543 })
6b27d407
JB
544 }
545 }
08f3f44c 546
aa9eede8
JB
547 /**
548 * The pool readiness boolean status.
549 */
2431bdb4 550 private get ready (): boolean {
8e8d9101
JB
551 if (this.empty) {
552 return false
553 }
2431bdb4 554 return (
b97d82d8
JB
555 this.workerNodes.reduce(
556 (accumulator, workerNode) =>
557 !workerNode.info.dynamic && workerNode.info.ready
558 ? accumulator + 1
559 : accumulator,
560 0
26ce26ca 561 ) >= this.minimumNumberOfWorkers
2431bdb4
JB
562 )
563 }
564
8e8d9101
JB
565 /**
566 * The pool emptiness boolean status.
567 */
568 protected get empty (): boolean {
fb43035c 569 return this.minimumNumberOfWorkers === 0 && this.workerNodes.length === 0
8e8d9101
JB
570 }
571
afe0d5bf 572 /**
aa9eede8 573 * The approximate pool utilization.
afe0d5bf
JB
574 *
575 * @returns The pool utilization.
576 */
577 private get utilization (): number {
97dc65d9
JB
578 if (this.startTimestamp == null) {
579 return 0
580 }
8e5ca040 581 const poolTimeCapacity =
26ce26ca
JB
582 (performance.now() - this.startTimestamp) *
583 (this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers)
afe0d5bf
JB
584 const totalTasksRunTime = this.workerNodes.reduce(
585 (accumulator, workerNode) =>
c63a35a0 586 accumulator + (workerNode.usage.runTime.aggregate ?? 0),
afe0d5bf
JB
587 0
588 )
589 const totalTasksWaitTime = this.workerNodes.reduce(
590 (accumulator, workerNode) =>
c63a35a0 591 accumulator + (workerNode.usage.waitTime.aggregate ?? 0),
afe0d5bf
JB
592 0
593 )
8e5ca040 594 return (totalTasksRunTime + totalTasksWaitTime) / poolTimeCapacity
afe0d5bf
JB
595 }
596
8881ae32 597 /**
aa9eede8 598 * The pool type.
8881ae32
JB
599 *
600 * If it is `'dynamic'`, it provides the `max` property.
601 */
602 protected abstract get type (): PoolType
603
184855e6 604 /**
aa9eede8 605 * The worker type.
184855e6
JB
606 */
607 protected abstract get worker (): WorkerType
608
6b813701
JB
609 /**
610 * Checks if the worker id sent in the received message from a worker is valid.
611 *
612 * @param message - The received message.
613 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the worker id is invalid.
614 */
4d159167 615 private checkMessageWorkerId (message: MessageValue<Data | Response>): void {
310de0aa
JB
616 if (message.workerId == null) {
617 throw new Error('Worker message received without worker id')
8e5a7cf9 618 } else if (this.getWorkerNodeKeyByWorkerId(message.workerId) === -1) {
21f710aa
JB
619 throw new Error(
620 `Worker message received from unknown worker '${message.workerId}'`
621 )
622 }
623 }
624
aa9eede8
JB
625 /**
626 * Gets the worker node key given its worker id.
627 *
628 * @param workerId - The worker id.
aad6fb64 629 * @returns The worker node key if the worker id is found in the pool worker nodes, `-1` otherwise.
aa9eede8 630 */
72ae84a2 631 private getWorkerNodeKeyByWorkerId (workerId: number | undefined): number {
aad6fb64 632 return this.workerNodes.findIndex(
041dc05b 633 workerNode => workerNode.info.id === workerId
aad6fb64 634 )
aa9eede8
JB
635 }
636
afc003b2 637 /** @inheritDoc */
a35560ba 638 public setWorkerChoiceStrategy (
59219cbb
JB
639 workerChoiceStrategy: WorkerChoiceStrategy,
640 workerChoiceStrategyOptions?: WorkerChoiceStrategyOptions
a35560ba 641 ): void {
19b8be8b 642 let requireSync = false
bde6b5d7 643 checkValidWorkerChoiceStrategy(workerChoiceStrategy)
85bbc7ab 644 if (workerChoiceStrategyOptions != null) {
423f3b7a 645 requireSync = !this.setWorkerChoiceStrategyOptions(
19b8be8b
JB
646 workerChoiceStrategyOptions
647 )
85bbc7ab
JB
648 }
649 if (workerChoiceStrategy !== this.opts.workerChoiceStrategy) {
650 this.opts.workerChoiceStrategy = workerChoiceStrategy
651 this.workerChoiceStrategiesContext?.setDefaultWorkerChoiceStrategy(
652 this.opts.workerChoiceStrategy,
653 this.opts.workerChoiceStrategyOptions
654 )
19b8be8b
JB
655 requireSync = true
656 }
657 if (requireSync) {
85bbc7ab
JB
658 this.workerChoiceStrategiesContext?.syncWorkerChoiceStrategies(
659 this.getWorkerWorkerChoiceStrategies(),
660 this.opts.workerChoiceStrategyOptions
661 )
19b8be8b 662 for (const workerNodeKey of this.workerNodes.keys()) {
85bbc7ab
JB
663 this.sendStatisticsMessageToWorker(workerNodeKey)
664 }
59219cbb 665 }
a20f0ba5
JB
666 }
667
668 /** @inheritDoc */
669 public setWorkerChoiceStrategyOptions (
c63a35a0 670 workerChoiceStrategyOptions: WorkerChoiceStrategyOptions | undefined
19b8be8b 671 ): boolean {
0d80593b 672 this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions)
26ce26ca
JB
673 if (workerChoiceStrategyOptions != null) {
674 this.opts.workerChoiceStrategyOptions = workerChoiceStrategyOptions
19b8be8b
JB
675 this.workerChoiceStrategiesContext?.setOptions(
676 this.opts.workerChoiceStrategyOptions
677 )
678 this.workerChoiceStrategiesContext?.syncWorkerChoiceStrategies(
679 this.getWorkerWorkerChoiceStrategies(),
680 this.opts.workerChoiceStrategyOptions
681 )
682 for (const workerNodeKey of this.workerNodes.keys()) {
683 this.sendStatisticsMessageToWorker(workerNodeKey)
684 }
685 return true
8990357d 686 }
19b8be8b 687 return false
a35560ba
S
688 }
689
a20f0ba5 690 /** @inheritDoc */
8f52842f
JB
691 public enableTasksQueue (
692 enable: boolean,
693 tasksQueueOptions?: TasksQueueOptions
694 ): void {
a20f0ba5 695 if (this.opts.enableTasksQueue === true && !enable) {
d6ca1416
JB
696 this.unsetTaskStealing()
697 this.unsetTasksStealingOnBackPressure()
ef41a6e6 698 this.flushTasksQueues()
a20f0ba5
JB
699 }
700 this.opts.enableTasksQueue = enable
c63a35a0 701 this.setTasksQueueOptions(tasksQueueOptions)
a20f0ba5
JB
702 }
703
704 /** @inheritDoc */
c63a35a0
JB
705 public setTasksQueueOptions (
706 tasksQueueOptions: TasksQueueOptions | undefined
707 ): void {
a20f0ba5 708 if (this.opts.enableTasksQueue === true) {
bde6b5d7 709 checkValidTasksQueueOptions(tasksQueueOptions)
8f52842f
JB
710 this.opts.tasksQueueOptions =
711 this.buildTasksQueueOptions(tasksQueueOptions)
67f3f2d6
JB
712 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
713 this.setTasksQueueSize(this.opts.tasksQueueOptions.size!)
d6ca1416 714 if (this.opts.tasksQueueOptions.taskStealing === true) {
9f99eb9b 715 this.unsetTaskStealing()
d6ca1416
JB
716 this.setTaskStealing()
717 } else {
718 this.unsetTaskStealing()
719 }
720 if (this.opts.tasksQueueOptions.tasksStealingOnBackPressure === true) {
9f99eb9b 721 this.unsetTasksStealingOnBackPressure()
d6ca1416
JB
722 this.setTasksStealingOnBackPressure()
723 } else {
724 this.unsetTasksStealingOnBackPressure()
725 }
5baee0d7 726 } else if (this.opts.tasksQueueOptions != null) {
a20f0ba5
JB
727 delete this.opts.tasksQueueOptions
728 }
729 }
730
9b38ab2d 731 private buildTasksQueueOptions (
c63a35a0 732 tasksQueueOptions: TasksQueueOptions | undefined
9b38ab2d
JB
733 ): TasksQueueOptions {
734 return {
26ce26ca
JB
735 ...getDefaultTasksQueueOptions(
736 this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers
737 ),
9b38ab2d
JB
738 ...tasksQueueOptions
739 }
740 }
741
5b49e864 742 private setTasksQueueSize (size: number): void {
20c6f652 743 for (const workerNode of this.workerNodes) {
ff3f866a 744 workerNode.tasksQueueBackPressureSize = size
20c6f652
JB
745 }
746 }
747
d6ca1416 748 private setTaskStealing (): void {
19b8be8b 749 for (const workerNodeKey of this.workerNodes.keys()) {
e44639e9 750 this.workerNodes[workerNodeKey].on('idle', this.handleWorkerNodeIdleEvent)
d6ca1416
JB
751 }
752 }
753
754 private unsetTaskStealing (): void {
19b8be8b 755 for (const workerNodeKey of this.workerNodes.keys()) {
e1c2dba7 756 this.workerNodes[workerNodeKey].off(
e44639e9
JB
757 'idle',
758 this.handleWorkerNodeIdleEvent
9f95d5eb 759 )
d6ca1416
JB
760 }
761 }
762
763 private setTasksStealingOnBackPressure (): void {
19b8be8b 764 for (const workerNodeKey of this.workerNodes.keys()) {
e1c2dba7 765 this.workerNodes[workerNodeKey].on(
b5e75be8 766 'backPressure',
e44639e9 767 this.handleWorkerNodeBackPressureEvent
9f95d5eb 768 )
d6ca1416
JB
769 }
770 }
771
772 private unsetTasksStealingOnBackPressure (): void {
19b8be8b 773 for (const workerNodeKey of this.workerNodes.keys()) {
e1c2dba7 774 this.workerNodes[workerNodeKey].off(
b5e75be8 775 'backPressure',
e44639e9 776 this.handleWorkerNodeBackPressureEvent
9f95d5eb 777 )
d6ca1416
JB
778 }
779 }
780
c319c66b
JB
781 /**
782 * Whether the pool is full or not.
783 *
784 * The pool filling boolean status.
785 */
dea903a8 786 protected get full (): boolean {
26ce26ca
JB
787 return (
788 this.workerNodes.length >=
789 (this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers)
790 )
dea903a8 791 }
c2ade475 792
c319c66b
JB
793 /**
794 * Whether the pool is busy or not.
795 *
796 * The pool busyness boolean status.
797 */
798 protected abstract get busy (): boolean
7c0ba920 799
6c6afb84 800 /**
3d76750a 801 * Whether worker nodes are executing concurrently their tasks quota or not.
6c6afb84
JB
802 *
803 * @returns Worker nodes busyness boolean status.
804 */
c2ade475 805 protected internalBusy (): boolean {
3d76750a
JB
806 if (this.opts.enableTasksQueue === true) {
807 return (
808 this.workerNodes.findIndex(
041dc05b 809 workerNode =>
3d76750a
JB
810 workerNode.info.ready &&
811 workerNode.usage.tasks.executing <
67f3f2d6
JB
812 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
813 this.opts.tasksQueueOptions!.concurrency!
3d76750a
JB
814 ) === -1
815 )
3d76750a 816 }
419e8121
JB
817 return (
818 this.workerNodes.findIndex(
819 workerNode =>
820 workerNode.info.ready && workerNode.usage.tasks.executing === 0
821 ) === -1
822 )
cb70b19d
JB
823 }
824
42c677c1
JB
825 private isWorkerNodeBusy (workerNodeKey: number): boolean {
826 if (this.opts.enableTasksQueue === true) {
827 return (
828 this.workerNodes[workerNodeKey].usage.tasks.executing >=
67f3f2d6
JB
829 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
830 this.opts.tasksQueueOptions!.concurrency!
42c677c1
JB
831 )
832 }
833 return this.workerNodes[workerNodeKey].usage.tasks.executing > 0
834 }
835
e81c38f2 836 private async sendTaskFunctionOperationToWorker (
72ae84a2
JB
837 workerNodeKey: number,
838 message: MessageValue<Data>
839 ): Promise<boolean> {
72ae84a2 840 return await new Promise<boolean>((resolve, reject) => {
ae036c3e
JB
841 const taskFunctionOperationListener = (
842 message: MessageValue<Response>
843 ): void => {
844 this.checkMessageWorkerId(message)
1851fed0 845 const workerId = this.getWorkerInfo(workerNodeKey)?.id
72ae84a2 846 if (
ae036c3e
JB
847 message.taskFunctionOperationStatus != null &&
848 message.workerId === workerId
72ae84a2 849 ) {
ae036c3e
JB
850 if (message.taskFunctionOperationStatus) {
851 resolve(true)
c63a35a0 852 } else {
ae036c3e
JB
853 reject(
854 new Error(
c63a35a0 855 `Task function operation '${message.taskFunctionOperation}' failed on worker ${message.workerId} with error: '${message.workerError?.message}'`
ae036c3e 856 )
72ae84a2 857 )
ae036c3e
JB
858 }
859 this.deregisterWorkerMessageListener(
860 this.getWorkerNodeKeyByWorkerId(message.workerId),
861 taskFunctionOperationListener
72ae84a2
JB
862 )
863 }
ae036c3e
JB
864 }
865 this.registerWorkerMessageListener(
866 workerNodeKey,
867 taskFunctionOperationListener
868 )
72ae84a2
JB
869 this.sendToWorker(workerNodeKey, message)
870 })
871 }
872
873 private async sendTaskFunctionOperationToWorkers (
adee6053 874 message: MessageValue<Data>
e81c38f2
JB
875 ): Promise<boolean> {
876 return await new Promise<boolean>((resolve, reject) => {
ae036c3e
JB
877 const responsesReceived = new Array<MessageValue<Response>>()
878 const taskFunctionOperationsListener = (
879 message: MessageValue<Response>
880 ): void => {
881 this.checkMessageWorkerId(message)
882 if (message.taskFunctionOperationStatus != null) {
883 responsesReceived.push(message)
884 if (responsesReceived.length === this.workerNodes.length) {
e81c38f2 885 if (
e81c38f2
JB
886 responsesReceived.every(
887 message => message.taskFunctionOperationStatus === true
888 )
889 ) {
890 resolve(true)
891 } else if (
e81c38f2
JB
892 responsesReceived.some(
893 message => message.taskFunctionOperationStatus === false
894 )
895 ) {
b0b55f57
JB
896 const errorResponse = responsesReceived.find(
897 response => response.taskFunctionOperationStatus === false
898 )
e81c38f2
JB
899 reject(
900 new Error(
b0b55f57 901 `Task function operation '${
e81c38f2 902 message.taskFunctionOperation as string
c63a35a0
JB
903 }' failed on worker ${errorResponse?.workerId} with error: '${
904 errorResponse?.workerError?.message
b0b55f57 905 }'`
e81c38f2
JB
906 )
907 )
908 }
ae036c3e
JB
909 this.deregisterWorkerMessageListener(
910 this.getWorkerNodeKeyByWorkerId(message.workerId),
911 taskFunctionOperationsListener
912 )
e81c38f2 913 }
ae036c3e
JB
914 }
915 }
19b8be8b 916 for (const workerNodeKey of this.workerNodes.keys()) {
ae036c3e
JB
917 this.registerWorkerMessageListener(
918 workerNodeKey,
919 taskFunctionOperationsListener
920 )
72ae84a2 921 this.sendToWorker(workerNodeKey, message)
e81c38f2
JB
922 }
923 })
6703b9f4
JB
924 }
925
926 /** @inheritDoc */
927 public hasTaskFunction (name: string): boolean {
85bbc7ab
JB
928 return this.listTaskFunctionsProperties().some(
929 taskFunctionProperties => taskFunctionProperties.name === name
930 )
6703b9f4
JB
931 }
932
933 /** @inheritDoc */
e81c38f2
JB
934 public async addTaskFunction (
935 name: string,
31847469 936 fn: TaskFunction<Data, Response> | TaskFunctionObject<Data, Response>
e81c38f2 937 ): Promise<boolean> {
3feeab69
JB
938 if (typeof name !== 'string') {
939 throw new TypeError('name argument must be a string')
940 }
941 if (typeof name === 'string' && name.trim().length === 0) {
942 throw new TypeError('name argument must not be an empty string')
943 }
31847469
JB
944 if (typeof fn === 'function') {
945 fn = { taskFunction: fn } satisfies TaskFunctionObject<Data, Response>
946 }
947 if (typeof fn.taskFunction !== 'function') {
948 throw new TypeError('taskFunction property must be a function')
3feeab69 949 }
95d1a734
JB
950 checkValidPriority(fn.priority)
951 checkValidWorkerChoiceStrategy(fn.strategy)
adee6053 952 const opResult = await this.sendTaskFunctionOperationToWorkers({
6703b9f4 953 taskFunctionOperation: 'add',
31847469
JB
954 taskFunctionProperties: buildTaskFunctionProperties(name, fn),
955 taskFunction: fn.taskFunction.toString()
6703b9f4 956 })
adee6053 957 this.taskFunctions.set(name, fn)
85bbc7ab
JB
958 this.workerChoiceStrategiesContext?.syncWorkerChoiceStrategies(
959 this.getWorkerWorkerChoiceStrategies()
960 )
9a55fa8c
JB
961 for (const workerNodeKey of this.workerNodes.keys()) {
962 this.sendStatisticsMessageToWorker(workerNodeKey)
963 }
adee6053 964 return opResult
6703b9f4
JB
965 }
966
967 /** @inheritDoc */
e81c38f2 968 public async removeTaskFunction (name: string): Promise<boolean> {
9eae3c69
JB
969 if (!this.taskFunctions.has(name)) {
970 throw new Error(
16248b23 971 'Cannot remove a task function not handled on the pool side'
9eae3c69
JB
972 )
973 }
adee6053 974 const opResult = await this.sendTaskFunctionOperationToWorkers({
6703b9f4 975 taskFunctionOperation: 'remove',
31847469
JB
976 taskFunctionProperties: buildTaskFunctionProperties(
977 name,
978 this.taskFunctions.get(name)
979 )
6703b9f4 980 })
9a55fa8c
JB
981 for (const workerNode of this.workerNodes) {
982 workerNode.deleteTaskFunctionWorkerUsage(name)
983 }
adee6053 984 this.taskFunctions.delete(name)
85bbc7ab
JB
985 this.workerChoiceStrategiesContext?.syncWorkerChoiceStrategies(
986 this.getWorkerWorkerChoiceStrategies()
987 )
9a55fa8c
JB
988 for (const workerNodeKey of this.workerNodes.keys()) {
989 this.sendStatisticsMessageToWorker(workerNodeKey)
990 }
adee6053 991 return opResult
6703b9f4
JB
992 }
993
90d7d101 994 /** @inheritDoc */
31847469 995 public listTaskFunctionsProperties (): TaskFunctionProperties[] {
f2dbbf95
JB
996 for (const workerNode of this.workerNodes) {
997 if (
31847469
JB
998 Array.isArray(workerNode.info.taskFunctionsProperties) &&
999 workerNode.info.taskFunctionsProperties.length > 0
f2dbbf95 1000 ) {
31847469 1001 return workerNode.info.taskFunctionsProperties
f2dbbf95 1002 }
90d7d101 1003 }
f2dbbf95 1004 return []
90d7d101
JB
1005 }
1006
bcfb06ce
JB
1007 /**
1008 * Gets task function strategy, if any.
1009 *
1010 * @param name - The task function name.
1011 * @returns The task function worker choice strategy if the task function worker choice strategy is defined, `undefined` otherwise.
1012 */
1013 private readonly getTaskFunctionWorkerWorkerChoiceStrategy = (
1014 name?: string
1015 ): WorkerChoiceStrategy | undefined => {
1016 if (name != null) {
1017 return this.listTaskFunctionsProperties().find(
1018 (taskFunctionProperties: TaskFunctionProperties) =>
1019 taskFunctionProperties.name === name
1020 )?.strategy
1021 }
1022 }
1023
95d1a734
JB
1024 /**
1025 * Gets worker node task function priority, if any.
1026 *
1027 * @param workerNodeKey - The worker node key.
1028 * @param name - The task function name.
1029 * @returns The task function worker choice priority if the task function worker choice priority is defined, `undefined` otherwise.
1030 */
1031 private readonly getWorkerNodeTaskFunctionPriority = (
1032 workerNodeKey: number,
1033 name?: string
1034 ): number | undefined => {
1035 if (name != null) {
1036 return this.getWorkerInfo(workerNodeKey)?.taskFunctionsProperties?.find(
1037 (taskFunctionProperties: TaskFunctionProperties) =>
1038 taskFunctionProperties.name === name
1039 )?.priority
1040 }
1041 }
1042
85bbc7ab
JB
1043 /**
1044 * Gets the worker choice strategies registered in this pool.
1045 *
1046 * @returns The worker choice strategies.
1047 */
1048 private readonly getWorkerWorkerChoiceStrategies =
1049 (): Set<WorkerChoiceStrategy> => {
1050 return new Set([
1051 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1052 this.opts.workerChoiceStrategy!,
1053 ...(this.listTaskFunctionsProperties()
1054 .map(
1055 (taskFunctionProperties: TaskFunctionProperties) =>
1056 taskFunctionProperties.strategy
1057 )
1058 .filter(
1059 (strategy: WorkerChoiceStrategy | undefined) => strategy != null
1060 ) as WorkerChoiceStrategy[])
1061 ])
1062 }
1063
6703b9f4 1064 /** @inheritDoc */
e81c38f2 1065 public async setDefaultTaskFunction (name: string): Promise<boolean> {
72ae84a2 1066 return await this.sendTaskFunctionOperationToWorkers({
6703b9f4 1067 taskFunctionOperation: 'default',
31847469
JB
1068 taskFunctionProperties: buildTaskFunctionProperties(
1069 name,
1070 this.taskFunctions.get(name)
1071 )
6703b9f4 1072 })
6703b9f4
JB
1073 }
1074
375f7504
JB
1075 private shallExecuteTask (workerNodeKey: number): boolean {
1076 return (
1077 this.tasksQueueSize(workerNodeKey) === 0 &&
1078 this.workerNodes[workerNodeKey].usage.tasks.executing <
67f3f2d6
JB
1079 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1080 this.opts.tasksQueueOptions!.concurrency!
375f7504
JB
1081 )
1082 }
1083
afc003b2 1084 /** @inheritDoc */
7d91a8cd
JB
1085 public async execute (
1086 data?: Data,
1087 name?: string,
6a3ecc50 1088 transferList?: readonly TransferListItem[]
7d91a8cd 1089 ): Promise<Response> {
52b71763 1090 return await new Promise<Response>((resolve, reject) => {
15b176e0 1091 if (!this.started) {
47352846 1092 reject(new Error('Cannot execute a task on not started pool'))
9d2d0da1 1093 return
15b176e0 1094 }
711623b8
JB
1095 if (this.destroying) {
1096 reject(new Error('Cannot execute a task on destroying pool'))
1097 return
1098 }
7d91a8cd
JB
1099 if (name != null && typeof name !== 'string') {
1100 reject(new TypeError('name argument must be a string'))
9d2d0da1 1101 return
7d91a8cd 1102 }
90d7d101
JB
1103 if (
1104 name != null &&
1105 typeof name === 'string' &&
1106 name.trim().length === 0
1107 ) {
f58b60b9 1108 reject(new TypeError('name argument must not be an empty string'))
9d2d0da1 1109 return
90d7d101 1110 }
b558f6b5
JB
1111 if (transferList != null && !Array.isArray(transferList)) {
1112 reject(new TypeError('transferList argument must be an array'))
9d2d0da1 1113 return
b558f6b5
JB
1114 }
1115 const timestamp = performance.now()
95d1a734 1116 const taskFunctionStrategy =
bcfb06ce 1117 this.getTaskFunctionWorkerWorkerChoiceStrategy(name)
95d1a734 1118 const workerNodeKey = this.chooseWorkerNode(taskFunctionStrategy)
501aea93 1119 const task: Task<Data> = {
52b71763
JB
1120 name: name ?? DEFAULT_TASK_NAME,
1121 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
1122 data: data ?? ({} as Data),
95d1a734
JB
1123 priority: this.getWorkerNodeTaskFunctionPriority(workerNodeKey, name),
1124 strategy: taskFunctionStrategy,
7d91a8cd 1125 transferList,
52b71763 1126 timestamp,
7629bdf1 1127 taskId: randomUUID()
52b71763 1128 }
67f3f2d6
JB
1129 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1130 this.promiseResponseMap.set(task.taskId!, {
2e81254d
JB
1131 resolve,
1132 reject,
f18fd12b
JB
1133 workerNodeKey,
1134 ...(this.emitter != null && {
1135 asyncResource: new AsyncResource('poolifier:task', {
1136 triggerAsyncId: this.emitter.asyncId,
1137 requireManualDestroy: true
1138 })
1139 })
2e81254d 1140 })
52b71763 1141 if (
4e377863
JB
1142 this.opts.enableTasksQueue === false ||
1143 (this.opts.enableTasksQueue === true &&
375f7504 1144 this.shallExecuteTask(workerNodeKey))
52b71763 1145 ) {
501aea93 1146 this.executeTask(workerNodeKey, task)
4e377863
JB
1147 } else {
1148 this.enqueueTask(workerNodeKey, task)
52b71763 1149 }
2e81254d 1150 })
280c2a77 1151 }
c97c7edb 1152
60dc0a9c
JB
1153 /**
1154 * Starts the minimum number of workers.
1155 */
e0843544 1156 private startMinimumNumberOfWorkers (initWorkerNodeUsage = false): void {
b362a929 1157 this.startingMinimumNumberOfWorkers = true
60dc0a9c
JB
1158 while (
1159 this.workerNodes.reduce(
1160 (accumulator, workerNode) =>
1161 !workerNode.info.dynamic ? accumulator + 1 : accumulator,
1162 0
1163 ) < this.minimumNumberOfWorkers
1164 ) {
e0843544
JB
1165 const workerNodeKey = this.createAndSetupWorkerNode()
1166 initWorkerNodeUsage &&
1167 this.initWorkerNodeUsage(this.workerNodes[workerNodeKey])
60dc0a9c 1168 }
b362a929 1169 this.startingMinimumNumberOfWorkers = false
60dc0a9c
JB
1170 }
1171
47352846
JB
1172 /** @inheritdoc */
1173 public start (): void {
711623b8
JB
1174 if (this.started) {
1175 throw new Error('Cannot start an already started pool')
1176 }
1177 if (this.starting) {
1178 throw new Error('Cannot start an already starting pool')
1179 }
1180 if (this.destroying) {
1181 throw new Error('Cannot start a destroying pool')
1182 }
47352846 1183 this.starting = true
60dc0a9c 1184 this.startMinimumNumberOfWorkers()
2008bccd 1185 this.startTimestamp = performance.now()
47352846
JB
1186 this.starting = false
1187 this.started = true
1188 }
1189
afc003b2 1190 /** @inheritDoc */
c97c7edb 1191 public async destroy (): Promise<void> {
711623b8
JB
1192 if (!this.started) {
1193 throw new Error('Cannot destroy an already destroyed pool')
1194 }
1195 if (this.starting) {
1196 throw new Error('Cannot destroy an starting pool')
1197 }
1198 if (this.destroying) {
1199 throw new Error('Cannot destroy an already destroying pool')
1200 }
1201 this.destroying = true
1fbcaa7c 1202 await Promise.all(
533a8e22 1203 this.workerNodes.map(async (_, workerNodeKey) => {
aa9eede8 1204 await this.destroyWorkerNode(workerNodeKey)
1fbcaa7c
JB
1205 })
1206 )
33e6bb4c 1207 this.emitter?.emit(PoolEvents.destroy, this.info)
f80125ca 1208 this.emitter?.emitDestroy()
55082af9 1209 this.readyEventEmitted = false
2008bccd 1210 delete this.startTimestamp
711623b8 1211 this.destroying = false
15b176e0 1212 this.started = false
c97c7edb
S
1213 }
1214
26ce26ca 1215 private async sendKillMessageToWorker (workerNodeKey: number): Promise<void> {
9edb9717 1216 await new Promise<void>((resolve, reject) => {
c63a35a0
JB
1217 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1218 if (this.workerNodes[workerNodeKey] == null) {
ad3836ed 1219 resolve()
85b2561d
JB
1220 return
1221 }
ae036c3e
JB
1222 const killMessageListener = (message: MessageValue<Response>): void => {
1223 this.checkMessageWorkerId(message)
1e3214b6
JB
1224 if (message.kill === 'success') {
1225 resolve()
1226 } else if (message.kill === 'failure') {
72ae84a2
JB
1227 reject(
1228 new Error(
c63a35a0 1229 `Kill message handling failed on worker ${message.workerId}`
72ae84a2
JB
1230 )
1231 )
1e3214b6 1232 }
ae036c3e 1233 }
51d9cfbd 1234 // FIXME: should be registered only once
ae036c3e 1235 this.registerWorkerMessageListener(workerNodeKey, killMessageListener)
72ae84a2 1236 this.sendToWorker(workerNodeKey, { kill: true })
1e3214b6 1237 })
1e3214b6
JB
1238 }
1239
4a6952ff 1240 /**
aa9eede8 1241 * Terminates the worker node given its worker node key.
4a6952ff 1242 *
aa9eede8 1243 * @param workerNodeKey - The worker node key.
4a6952ff 1244 */
07e0c9e5
JB
1245 protected async destroyWorkerNode (workerNodeKey: number): Promise<void> {
1246 this.flagWorkerNodeAsNotReady(workerNodeKey)
87347ea8 1247 const flushedTasks = this.flushTasksQueue(workerNodeKey)
07e0c9e5 1248 const workerNode = this.workerNodes[workerNodeKey]
32b141fd
JB
1249 await waitWorkerNodeEvents(
1250 workerNode,
1251 'taskFinished',
1252 flushedTasks,
1253 this.opts.tasksQueueOptions?.tasksFinishedTimeout ??
26ce26ca
JB
1254 getDefaultTasksQueueOptions(
1255 this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers
1256 ).tasksFinishedTimeout
32b141fd 1257 )
07e0c9e5
JB
1258 await this.sendKillMessageToWorker(workerNodeKey)
1259 await workerNode.terminate()
1260 }
c97c7edb 1261
729c563d 1262 /**
6677a3d3
JB
1263 * Setup hook to execute code before worker nodes are created in the abstract constructor.
1264 * Can be overridden.
afc003b2
JB
1265 *
1266 * @virtual
729c563d 1267 */
280c2a77 1268 protected setupHook (): void {
965df41c 1269 /* Intentionally empty */
280c2a77 1270 }
c97c7edb 1271
729c563d 1272 /**
5bec72fd
JB
1273 * Returns whether the worker is the main worker or not.
1274 *
1275 * @returns `true` if the worker is the main worker, `false` otherwise.
280c2a77
S
1276 */
1277 protected abstract isMain (): boolean
1278
1279 /**
2e81254d 1280 * Hook executed before the worker task execution.
bf9549ae 1281 * Can be overridden.
729c563d 1282 *
f06e48d8 1283 * @param workerNodeKey - The worker node key.
1c6fe997 1284 * @param task - The task to execute.
729c563d 1285 */
1c6fe997
JB
1286 protected beforeTaskExecutionHook (
1287 workerNodeKey: number,
1288 task: Task<Data>
1289 ): void {
c63a35a0 1290 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
efabc98d 1291 if (this.workerNodes[workerNodeKey]?.usage != null) {
94407def
JB
1292 const workerUsage = this.workerNodes[workerNodeKey].usage
1293 ++workerUsage.tasks.executing
c329fd41 1294 updateWaitTimeWorkerUsage(
bcfb06ce 1295 this.workerChoiceStrategiesContext,
c329fd41
JB
1296 workerUsage,
1297 task
1298 )
94407def
JB
1299 }
1300 if (
1301 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) &&
67f3f2d6
JB
1302 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1303 this.workerNodes[workerNodeKey].getTaskFunctionWorkerUsage(task.name!) !=
1304 null
94407def 1305 ) {
67f3f2d6 1306 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
db0e38ee 1307 const taskFunctionWorkerUsage = this.workerNodes[
b558f6b5 1308 workerNodeKey
67f3f2d6
JB
1309 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1310 ].getTaskFunctionWorkerUsage(task.name!)!
5623b8d5 1311 ++taskFunctionWorkerUsage.tasks.executing
c329fd41 1312 updateWaitTimeWorkerUsage(
bcfb06ce 1313 this.workerChoiceStrategiesContext,
c329fd41
JB
1314 taskFunctionWorkerUsage,
1315 task
1316 )
b558f6b5 1317 }
c97c7edb
S
1318 }
1319
c01733f1 1320 /**
2e81254d 1321 * Hook executed after the worker task execution.
bf9549ae 1322 * Can be overridden.
c01733f1 1323 *
501aea93 1324 * @param workerNodeKey - The worker node key.
38e795c1 1325 * @param message - The received message.
c01733f1 1326 */
2e81254d 1327 protected afterTaskExecutionHook (
501aea93 1328 workerNodeKey: number,
2740a743 1329 message: MessageValue<Response>
bf9549ae 1330 ): void {
9a55fa8c 1331 let needWorkerChoiceStrategiesUpdate = false
c63a35a0 1332 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
efabc98d 1333 if (this.workerNodes[workerNodeKey]?.usage != null) {
94407def 1334 const workerUsage = this.workerNodes[workerNodeKey].usage
c329fd41
JB
1335 updateTaskStatisticsWorkerUsage(workerUsage, message)
1336 updateRunTimeWorkerUsage(
bcfb06ce 1337 this.workerChoiceStrategiesContext,
c329fd41
JB
1338 workerUsage,
1339 message
1340 )
1341 updateEluWorkerUsage(
bcfb06ce 1342 this.workerChoiceStrategiesContext,
c329fd41
JB
1343 workerUsage,
1344 message
1345 )
9a55fa8c 1346 needWorkerChoiceStrategiesUpdate = true
94407def
JB
1347 }
1348 if (
1349 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) &&
1350 this.workerNodes[workerNodeKey].getTaskFunctionWorkerUsage(
67f3f2d6
JB
1351 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1352 message.taskPerformance!.name
94407def
JB
1353 ) != null
1354 ) {
67f3f2d6 1355 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
db0e38ee 1356 const taskFunctionWorkerUsage = this.workerNodes[
b558f6b5 1357 workerNodeKey
67f3f2d6
JB
1358 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1359 ].getTaskFunctionWorkerUsage(message.taskPerformance!.name)!
c329fd41
JB
1360 updateTaskStatisticsWorkerUsage(taskFunctionWorkerUsage, message)
1361 updateRunTimeWorkerUsage(
bcfb06ce 1362 this.workerChoiceStrategiesContext,
c329fd41
JB
1363 taskFunctionWorkerUsage,
1364 message
1365 )
1366 updateEluWorkerUsage(
bcfb06ce 1367 this.workerChoiceStrategiesContext,
c329fd41
JB
1368 taskFunctionWorkerUsage,
1369 message
1370 )
9a55fa8c 1371 needWorkerChoiceStrategiesUpdate = true
c329fd41 1372 }
9a55fa8c 1373 if (needWorkerChoiceStrategiesUpdate) {
bcfb06ce 1374 this.workerChoiceStrategiesContext?.update(workerNodeKey)
b558f6b5
JB
1375 }
1376 }
1377
db0e38ee
JB
1378 /**
1379 * Whether the worker node shall update its task function worker usage or not.
1380 *
1381 * @param workerNodeKey - The worker node key.
1382 * @returns `true` if the worker node shall update its task function worker usage, `false` otherwise.
1383 */
1384 private shallUpdateTaskFunctionWorkerUsage (workerNodeKey: number): boolean {
a5d15204 1385 const workerInfo = this.getWorkerInfo(workerNodeKey)
b558f6b5 1386 return (
1851fed0 1387 workerInfo != null &&
31847469
JB
1388 Array.isArray(workerInfo.taskFunctionsProperties) &&
1389 workerInfo.taskFunctionsProperties.length > 2
b558f6b5 1390 )
f1c06930
JB
1391 }
1392
280c2a77 1393 /**
2be9b405 1394 * Chooses a worker node for the next task given the worker choice strategy.
280c2a77 1395 *
bcfb06ce 1396 * @param workerChoiceStrategy - The worker choice strategy.
aa9eede8 1397 * @returns The chosen worker node key
280c2a77 1398 */
bcfb06ce
JB
1399 private chooseWorkerNode (
1400 workerChoiceStrategy?: WorkerChoiceStrategy
1401 ): number {
930dcf12 1402 if (this.shallCreateDynamicWorker()) {
aa9eede8 1403 const workerNodeKey = this.createAndSetupDynamicWorkerNode()
6c6afb84 1404 if (
bcfb06ce
JB
1405 this.workerChoiceStrategiesContext?.getPolicy().dynamicWorkerUsage ===
1406 true
6c6afb84 1407 ) {
aa9eede8 1408 return workerNodeKey
6c6afb84 1409 }
17393ac8 1410 }
c63a35a0 1411 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
bcfb06ce 1412 return this.workerChoiceStrategiesContext!.execute(workerChoiceStrategy)
930dcf12
JB
1413 }
1414
6c6afb84
JB
1415 /**
1416 * Conditions for dynamic worker creation.
1417 *
1418 * @returns Whether to create a dynamic worker or not.
1419 */
9d9fb7b6 1420 protected abstract shallCreateDynamicWorker (): boolean
c97c7edb 1421
280c2a77 1422 /**
aa9eede8 1423 * Sends a message to worker given its worker node key.
280c2a77 1424 *
aa9eede8 1425 * @param workerNodeKey - The worker node key.
38e795c1 1426 * @param message - The message.
7d91a8cd 1427 * @param transferList - The optional array of transferable objects.
280c2a77
S
1428 */
1429 protected abstract sendToWorker (
aa9eede8 1430 workerNodeKey: number,
7d91a8cd 1431 message: MessageValue<Data>,
6a3ecc50 1432 transferList?: readonly TransferListItem[]
280c2a77
S
1433 ): void
1434
e0843544
JB
1435 /**
1436 * Initializes the worker node usage with sensible default values gathered during runtime.
1437 *
1438 * @param workerNode - The worker node.
1439 */
1440 private initWorkerNodeUsage (workerNode: IWorkerNode<Worker, Data>): void {
1441 if (
1442 this.workerChoiceStrategiesContext?.getTaskStatisticsRequirements()
1443 .runTime.aggregate === true
1444 ) {
1445 workerNode.usage.runTime.aggregate = min(
1446 ...this.workerNodes.map(
1447 workerNode => workerNode.usage.runTime.aggregate ?? Infinity
1448 )
1449 )
1450 }
1451 if (
1452 this.workerChoiceStrategiesContext?.getTaskStatisticsRequirements()
1453 .waitTime.aggregate === true
1454 ) {
1455 workerNode.usage.waitTime.aggregate = min(
1456 ...this.workerNodes.map(
1457 workerNode => workerNode.usage.waitTime.aggregate ?? Infinity
1458 )
1459 )
1460 }
1461 if (
1462 this.workerChoiceStrategiesContext?.getTaskStatisticsRequirements().elu
1463 .aggregate === true
1464 ) {
1465 workerNode.usage.elu.active.aggregate = min(
1466 ...this.workerNodes.map(
1467 workerNode => workerNode.usage.elu.active.aggregate ?? Infinity
1468 )
1469 )
1470 }
1471 }
1472
4a6952ff 1473 /**
aa9eede8 1474 * Creates a new, completely set up worker node.
4a6952ff 1475 *
aa9eede8 1476 * @returns New, completely set up worker node key.
4a6952ff 1477 */
aa9eede8 1478 protected createAndSetupWorkerNode (): number {
c3719753
JB
1479 const workerNode = this.createWorkerNode()
1480 workerNode.registerWorkerEventHandler(
1481 'online',
1482 this.opts.onlineHandler ?? EMPTY_FUNCTION
1483 )
1484 workerNode.registerWorkerEventHandler(
1485 'message',
1486 this.opts.messageHandler ?? EMPTY_FUNCTION
1487 )
1488 workerNode.registerWorkerEventHandler(
1489 'error',
1490 this.opts.errorHandler ?? EMPTY_FUNCTION
1491 )
b271fce0 1492 workerNode.registerOnceWorkerEventHandler('error', (error: Error) => {
07e0c9e5 1493 workerNode.info.ready = false
2a69b8c5 1494 this.emitter?.emit(PoolEvents.error, error)
15b176e0 1495 if (
b6bfca01 1496 this.started &&
711623b8 1497 !this.destroying &&
9b38ab2d 1498 this.opts.restartWorkerOnError === true
15b176e0 1499 ) {
07e0c9e5 1500 if (workerNode.info.dynamic) {
aa9eede8 1501 this.createAndSetupDynamicWorkerNode()
b362a929 1502 } else if (!this.startingMinimumNumberOfWorkers) {
e0843544 1503 this.startMinimumNumberOfWorkers(true)
8a1260a3 1504 }
5baee0d7 1505 }
3c9123c7
JB
1506 if (
1507 this.started &&
1508 !this.destroying &&
1509 this.opts.enableTasksQueue === true
1510 ) {
9974369e 1511 this.redistributeQueuedTasks(this.workerNodes.indexOf(workerNode))
19dbc45b 1512 }
b4f8aca8 1513 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
07f6f7b1 1514 workerNode?.terminate().catch((error: unknown) => {
07e0c9e5
JB
1515 this.emitter?.emit(PoolEvents.error, error)
1516 })
5baee0d7 1517 })
c3719753
JB
1518 workerNode.registerWorkerEventHandler(
1519 'exit',
1520 this.opts.exitHandler ?? EMPTY_FUNCTION
1521 )
1522 workerNode.registerOnceWorkerEventHandler('exit', () => {
9974369e 1523 this.removeWorkerNode(workerNode)
b362a929
JB
1524 if (
1525 this.started &&
1526 !this.startingMinimumNumberOfWorkers &&
1527 !this.destroying
1528 ) {
e0843544 1529 this.startMinimumNumberOfWorkers(true)
60dc0a9c 1530 }
a974afa6 1531 })
c3719753 1532 const workerNodeKey = this.addWorkerNode(workerNode)
aa9eede8 1533 this.afterWorkerNodeSetup(workerNodeKey)
aa9eede8 1534 return workerNodeKey
c97c7edb 1535 }
be0676b3 1536
930dcf12 1537 /**
aa9eede8 1538 * Creates a new, completely set up dynamic worker node.
930dcf12 1539 *
aa9eede8 1540 * @returns New, completely set up dynamic worker node key.
930dcf12 1541 */
aa9eede8
JB
1542 protected createAndSetupDynamicWorkerNode (): number {
1543 const workerNodeKey = this.createAndSetupWorkerNode()
041dc05b 1544 this.registerWorkerMessageListener(workerNodeKey, message => {
4d159167 1545 this.checkMessageWorkerId(message)
aa9eede8
JB
1546 const localWorkerNodeKey = this.getWorkerNodeKeyByWorkerId(
1547 message.workerId
aad6fb64 1548 )
2b6b412f 1549 const workerUsage = this.workerNodes[localWorkerNodeKey]?.usage
81c02522 1550 // Kill message received from worker
930dcf12
JB
1551 if (
1552 isKillBehavior(KillBehaviors.HARD, message.kill) ||
1e3214b6 1553 (isKillBehavior(KillBehaviors.SOFT, message.kill) &&
7b56f532 1554 ((this.opts.enableTasksQueue === false &&
aa9eede8 1555 workerUsage.tasks.executing === 0) ||
7b56f532 1556 (this.opts.enableTasksQueue === true &&
aa9eede8
JB
1557 workerUsage.tasks.executing === 0 &&
1558 this.tasksQueueSize(localWorkerNodeKey) === 0)))
930dcf12 1559 ) {
f3827d5d 1560 // Flag the worker node as not ready immediately
ae3ab61d 1561 this.flagWorkerNodeAsNotReady(localWorkerNodeKey)
07f6f7b1 1562 this.destroyWorkerNode(localWorkerNodeKey).catch((error: unknown) => {
5270d253
JB
1563 this.emitter?.emit(PoolEvents.error, error)
1564 })
930dcf12
JB
1565 }
1566 })
aa9eede8 1567 this.sendToWorker(workerNodeKey, {
e9dd5b66 1568 checkActive: true
21f710aa 1569 })
72ae84a2 1570 if (this.taskFunctions.size > 0) {
31847469 1571 for (const [taskFunctionName, taskFunctionObject] of this.taskFunctions) {
72ae84a2
JB
1572 this.sendTaskFunctionOperationToWorker(workerNodeKey, {
1573 taskFunctionOperation: 'add',
31847469
JB
1574 taskFunctionProperties: buildTaskFunctionProperties(
1575 taskFunctionName,
1576 taskFunctionObject
1577 ),
1578 taskFunction: taskFunctionObject.taskFunction.toString()
07f6f7b1 1579 }).catch((error: unknown) => {
72ae84a2
JB
1580 this.emitter?.emit(PoolEvents.error, error)
1581 })
1582 }
1583 }
e44639e9
JB
1584 const workerNode = this.workerNodes[workerNodeKey]
1585 workerNode.info.dynamic = true
b1aae695 1586 if (
bcfb06ce
JB
1587 this.workerChoiceStrategiesContext?.getPolicy().dynamicWorkerReady ===
1588 true ||
1589 this.workerChoiceStrategiesContext?.getPolicy().dynamicWorkerUsage ===
1590 true
b1aae695 1591 ) {
e44639e9 1592 workerNode.info.ready = true
b5e113f6 1593 }
e0843544 1594 this.initWorkerNodeUsage(workerNode)
33e6bb4c 1595 this.checkAndEmitDynamicWorkerCreationEvents()
aa9eede8 1596 return workerNodeKey
930dcf12
JB
1597 }
1598
a2ed5053 1599 /**
aa9eede8 1600 * Registers a listener callback on the worker given its worker node key.
a2ed5053 1601 *
aa9eede8 1602 * @param workerNodeKey - The worker node key.
a2ed5053
JB
1603 * @param listener - The message listener callback.
1604 */
85aeb3f3
JB
1605 protected abstract registerWorkerMessageListener<
1606 Message extends Data | Response
aa9eede8
JB
1607 >(
1608 workerNodeKey: number,
1609 listener: (message: MessageValue<Message>) => void
1610 ): void
a2ed5053 1611
ae036c3e
JB
1612 /**
1613 * Registers once a listener callback on the worker given its worker node key.
1614 *
1615 * @param workerNodeKey - The worker node key.
1616 * @param listener - The message listener callback.
1617 */
1618 protected abstract registerOnceWorkerMessageListener<
1619 Message extends Data | Response
1620 >(
1621 workerNodeKey: number,
1622 listener: (message: MessageValue<Message>) => void
1623 ): void
1624
1625 /**
1626 * Deregisters a listener callback on the worker given its worker node key.
1627 *
1628 * @param workerNodeKey - The worker node key.
1629 * @param listener - The message listener callback.
1630 */
1631 protected abstract deregisterWorkerMessageListener<
1632 Message extends Data | Response
1633 >(
1634 workerNodeKey: number,
1635 listener: (message: MessageValue<Message>) => void
1636 ): void
1637
a2ed5053 1638 /**
aa9eede8 1639 * Method hooked up after a worker node has been newly created.
a2ed5053
JB
1640 * Can be overridden.
1641 *
aa9eede8 1642 * @param workerNodeKey - The newly created worker node key.
a2ed5053 1643 */
aa9eede8 1644 protected afterWorkerNodeSetup (workerNodeKey: number): void {
a2ed5053 1645 // Listen to worker messages.
fcd39179
JB
1646 this.registerWorkerMessageListener(
1647 workerNodeKey,
3a776b6d 1648 this.workerMessageListener
fcd39179 1649 )
85aeb3f3 1650 // Send the startup message to worker.
aa9eede8 1651 this.sendStartupMessageToWorker(workerNodeKey)
9edb9717
JB
1652 // Send the statistics message to worker.
1653 this.sendStatisticsMessageToWorker(workerNodeKey)
72695f86 1654 if (this.opts.enableTasksQueue === true) {
dbd73092 1655 if (this.opts.tasksQueueOptions?.taskStealing === true) {
e1c2dba7 1656 this.workerNodes[workerNodeKey].on(
e44639e9
JB
1657 'idle',
1658 this.handleWorkerNodeIdleEvent
9f95d5eb 1659 )
47352846
JB
1660 }
1661 if (this.opts.tasksQueueOptions?.tasksStealingOnBackPressure === true) {
e1c2dba7 1662 this.workerNodes[workerNodeKey].on(
b5e75be8 1663 'backPressure',
e44639e9 1664 this.handleWorkerNodeBackPressureEvent
9f95d5eb 1665 )
47352846 1666 }
72695f86 1667 }
d2c73f82
JB
1668 }
1669
85aeb3f3 1670 /**
aa9eede8
JB
1671 * Sends the startup message to worker given its worker node key.
1672 *
1673 * @param workerNodeKey - The worker node key.
1674 */
1675 protected abstract sendStartupMessageToWorker (workerNodeKey: number): void
1676
1677 /**
9edb9717 1678 * Sends the statistics message to worker given its worker node key.
85aeb3f3 1679 *
aa9eede8 1680 * @param workerNodeKey - The worker node key.
85aeb3f3 1681 */
9edb9717 1682 private sendStatisticsMessageToWorker (workerNodeKey: number): void {
aa9eede8
JB
1683 this.sendToWorker(workerNodeKey, {
1684 statistics: {
1685 runTime:
bcfb06ce 1686 this.workerChoiceStrategiesContext?.getTaskStatisticsRequirements()
f4d0a470 1687 .runTime.aggregate ?? false,
c63a35a0 1688 elu:
bcfb06ce
JB
1689 this.workerChoiceStrategiesContext?.getTaskStatisticsRequirements()
1690 .elu.aggregate ?? false
72ae84a2 1691 }
aa9eede8
JB
1692 })
1693 }
a2ed5053 1694
5eb72b9e
JB
1695 private cannotStealTask (): boolean {
1696 return this.workerNodes.length <= 1 || this.info.queuedTasks === 0
1697 }
1698
42c677c1
JB
1699 private handleTask (workerNodeKey: number, task: Task<Data>): void {
1700 if (this.shallExecuteTask(workerNodeKey)) {
1701 this.executeTask(workerNodeKey, task)
1702 } else {
1703 this.enqueueTask(workerNodeKey, task)
1704 }
1705 }
1706
2d1d3b2d
JB
1707 private redistributeQueuedTasks (sourceWorkerNodeKey: number): void {
1708 if (sourceWorkerNodeKey === -1 || this.cannotStealTask()) {
cb71d660
JB
1709 return
1710 }
2d1d3b2d 1711 while (this.tasksQueueSize(sourceWorkerNodeKey) > 0) {
f201a0cd
JB
1712 const destinationWorkerNodeKey = this.workerNodes.reduce(
1713 (minWorkerNodeKey, workerNode, workerNodeKey, workerNodes) => {
2d1d3b2d
JB
1714 return sourceWorkerNodeKey !== workerNodeKey &&
1715 workerNode.info.ready &&
852ed3e4
JB
1716 workerNode.usage.tasks.queued <
1717 workerNodes[minWorkerNodeKey].usage.tasks.queued
f201a0cd
JB
1718 ? workerNodeKey
1719 : minWorkerNodeKey
1720 },
1721 0
1722 )
42c677c1
JB
1723 this.handleTask(
1724 destinationWorkerNodeKey,
67f3f2d6 1725 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2d1d3b2d 1726 this.dequeueTask(sourceWorkerNodeKey)!
42c677c1 1727 )
dd951876
JB
1728 }
1729 }
1730
b1838604
JB
1731 private updateTaskStolenStatisticsWorkerUsage (
1732 workerNodeKey: number,
b1838604
JB
1733 taskName: string
1734 ): void {
1a880eca 1735 const workerNode = this.workerNodes[workerNodeKey]
c63a35a0 1736 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
efabc98d 1737 if (workerNode?.usage != null) {
b1838604
JB
1738 ++workerNode.usage.tasks.stolen
1739 }
1740 if (
1741 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) &&
1742 workerNode.getTaskFunctionWorkerUsage(taskName) != null
1743 ) {
8bd0556f
JB
1744 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1745 ++workerNode.getTaskFunctionWorkerUsage(taskName)!.tasks.stolen
b1838604
JB
1746 }
1747 }
1748
463226a4 1749 private updateTaskSequentiallyStolenStatisticsWorkerUsage (
2d1d3b2d
JB
1750 workerNodeKey: number,
1751 taskName: string,
8bd0556f 1752 previousTaskName?: string
463226a4
JB
1753 ): void {
1754 const workerNode = this.workerNodes[workerNodeKey]
c63a35a0 1755 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
7aa2f476 1756 if (workerNode?.usage != null) {
463226a4
JB
1757 ++workerNode.usage.tasks.sequentiallyStolen
1758 }
1759 if (
1760 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) &&
3b33cd2a
JB
1761 workerNode.getTaskFunctionWorkerUsage(taskName) != null
1762 ) {
1763 const taskFunctionWorkerUsage =
1764 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1765 workerNode.getTaskFunctionWorkerUsage(taskName)!
1766 if (
1767 taskFunctionWorkerUsage.tasks.sequentiallyStolen === 0 ||
8bd0556f
JB
1768 (previousTaskName != null &&
1769 previousTaskName === taskName &&
3b33cd2a
JB
1770 taskFunctionWorkerUsage.tasks.sequentiallyStolen > 0)
1771 ) {
1772 ++taskFunctionWorkerUsage.tasks.sequentiallyStolen
1773 } else if (taskFunctionWorkerUsage.tasks.sequentiallyStolen > 0) {
1774 taskFunctionWorkerUsage.tasks.sequentiallyStolen = 0
1775 }
463226a4
JB
1776 }
1777 }
1778
1779 private resetTaskSequentiallyStolenStatisticsWorkerUsage (
2d1d3b2d
JB
1780 workerNodeKey: number,
1781 taskName: string
463226a4
JB
1782 ): void {
1783 const workerNode = this.workerNodes[workerNodeKey]
c63a35a0 1784 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
efabc98d 1785 if (workerNode?.usage != null) {
463226a4
JB
1786 workerNode.usage.tasks.sequentiallyStolen = 0
1787 }
1788 if (
1789 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) &&
1790 workerNode.getTaskFunctionWorkerUsage(taskName) != null
1791 ) {
8bd0556f
JB
1792 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1793 workerNode.getTaskFunctionWorkerUsage(
1794 taskName
1795 )!.tasks.sequentiallyStolen = 0
463226a4
JB
1796 }
1797 }
1798
e44639e9 1799 private readonly handleWorkerNodeIdleEvent = (
e1c2dba7 1800 eventDetail: WorkerNodeEventDetail,
463226a4 1801 previousStolenTask?: Task<Data>
9f95d5eb 1802 ): void => {
e1c2dba7 1803 const { workerNodeKey } = eventDetail
463226a4
JB
1804 if (workerNodeKey == null) {
1805 throw new Error(
c63a35a0 1806 "WorkerNode event detail 'workerNodeKey' property must be defined"
463226a4
JB
1807 )
1808 }
c63a35a0 1809 const workerInfo = this.getWorkerInfo(workerNodeKey)
010f158c
JB
1810 if (workerInfo == null) {
1811 throw new Error(
1812 `Worker node with key '${workerNodeKey}' not found in pool`
1813 )
1814 }
5eb72b9e
JB
1815 if (
1816 this.cannotStealTask() ||
c63a35a0
JB
1817 (this.info.stealingWorkerNodes ?? 0) >
1818 Math.floor(this.workerNodes.length / 2)
5eb72b9e 1819 ) {
010f158c 1820 if (previousStolenTask != null) {
c63a35a0 1821 workerInfo.stealing = false
2d1d3b2d
JB
1822 this.resetTaskSequentiallyStolenStatisticsWorkerUsage(
1823 workerNodeKey,
1824 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1825 previousStolenTask.name!
1826 )
5eb72b9e
JB
1827 }
1828 return
1829 }
463226a4
JB
1830 const workerNodeTasksUsage = this.workerNodes[workerNodeKey].usage.tasks
1831 if (
1832 previousStolenTask != null &&
463226a4
JB
1833 (workerNodeTasksUsage.executing > 0 ||
1834 this.tasksQueueSize(workerNodeKey) > 0)
1835 ) {
c63a35a0 1836 workerInfo.stealing = false
2d1d3b2d
JB
1837 this.resetTaskSequentiallyStolenStatisticsWorkerUsage(
1838 workerNodeKey,
1839 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1840 previousStolenTask.name!
1841 )
463226a4
JB
1842 return
1843 }
c63a35a0 1844 workerInfo.stealing = true
463226a4 1845 const stolenTask = this.workerNodeStealTask(workerNodeKey)
8bd0556f 1846 if (stolenTask != null) {
2d1d3b2d
JB
1847 this.updateTaskSequentiallyStolenStatisticsWorkerUsage(
1848 workerNodeKey,
67f3f2d6 1849 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2d1d3b2d 1850 stolenTask.name!,
8bd0556f 1851 previousStolenTask?.name
2d1d3b2d 1852 )
f1f77f45 1853 }
463226a4
JB
1854 sleep(exponentialDelay(workerNodeTasksUsage.sequentiallyStolen))
1855 .then(() => {
e44639e9 1856 this.handleWorkerNodeIdleEvent(eventDetail, stolenTask)
463226a4
JB
1857 return undefined
1858 })
07f6f7b1 1859 .catch((error: unknown) => {
2e8cfbb7
JB
1860 this.emitter?.emit(PoolEvents.error, error)
1861 })
463226a4
JB
1862 }
1863
1864 private readonly workerNodeStealTask = (
1865 workerNodeKey: number
1866 ): Task<Data> | undefined => {
dd951876 1867 const workerNodes = this.workerNodes
a6b3272b 1868 .slice()
dd951876
JB
1869 .sort(
1870 (workerNodeA, workerNodeB) =>
1871 workerNodeB.usage.tasks.queued - workerNodeA.usage.tasks.queued
1872 )
f201a0cd 1873 const sourceWorkerNode = workerNodes.find(
463226a4
JB
1874 (sourceWorkerNode, sourceWorkerNodeKey) =>
1875 sourceWorkerNode.info.ready &&
5eb72b9e 1876 !sourceWorkerNode.info.stealing &&
463226a4
JB
1877 sourceWorkerNodeKey !== workerNodeKey &&
1878 sourceWorkerNode.usage.tasks.queued > 0
f201a0cd
JB
1879 )
1880 if (sourceWorkerNode != null) {
67f3f2d6 1881 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2eee7220 1882 const task = sourceWorkerNode.dequeueLastPrioritizedTask()!
42c677c1 1883 this.handleTask(workerNodeKey, task)
67f3f2d6
JB
1884 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1885 this.updateTaskStolenStatisticsWorkerUsage(workerNodeKey, task.name!)
463226a4 1886 return task
72695f86
JB
1887 }
1888 }
1889
e44639e9 1890 private readonly handleWorkerNodeBackPressureEvent = (
e1c2dba7 1891 eventDetail: WorkerNodeEventDetail
9f95d5eb 1892 ): void => {
5eb72b9e
JB
1893 if (
1894 this.cannotStealTask() ||
2eee7220 1895 this.hasBackPressure() ||
c63a35a0
JB
1896 (this.info.stealingWorkerNodes ?? 0) >
1897 Math.floor(this.workerNodes.length / 2)
5eb72b9e 1898 ) {
cb71d660
JB
1899 return
1900 }
f778c355 1901 const sizeOffset = 1
67f3f2d6
JB
1902 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1903 if (this.opts.tasksQueueOptions!.size! <= sizeOffset) {
68dbcdc0
JB
1904 return
1905 }
7b2c5627 1906 const { workerId } = eventDetail
72695f86 1907 const sourceWorkerNode =
b5e75be8 1908 this.workerNodes[this.getWorkerNodeKeyByWorkerId(workerId)]
72695f86 1909 const workerNodes = this.workerNodes
a6b3272b 1910 .slice()
72695f86
JB
1911 .sort(
1912 (workerNodeA, workerNodeB) =>
1913 workerNodeA.usage.tasks.queued - workerNodeB.usage.tasks.queued
1914 )
1915 for (const [workerNodeKey, workerNode] of workerNodes.entries()) {
1916 if (
0bc68267 1917 sourceWorkerNode.usage.tasks.queued > 0 &&
a6b3272b 1918 workerNode.info.ready &&
5eb72b9e 1919 !workerNode.info.stealing &&
b5e75be8 1920 workerNode.info.id !== workerId &&
0bc68267 1921 workerNode.usage.tasks.queued <
67f3f2d6
JB
1922 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1923 this.opts.tasksQueueOptions!.size! - sizeOffset
72695f86 1924 ) {
c63a35a0 1925 const workerInfo = this.getWorkerInfo(workerNodeKey)
1851fed0
JB
1926 if (workerInfo == null) {
1927 throw new Error(
1928 `Worker node with key '${workerNodeKey}' not found in pool`
1929 )
1930 }
c63a35a0 1931 workerInfo.stealing = true
67f3f2d6 1932 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2eee7220 1933 const task = sourceWorkerNode.dequeueLastPrioritizedTask()!
42c677c1 1934 this.handleTask(workerNodeKey, task)
67f3f2d6
JB
1935 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1936 this.updateTaskStolenStatisticsWorkerUsage(workerNodeKey, task.name!)
c63a35a0 1937 workerInfo.stealing = false
10ecf8fd 1938 }
a2ed5053
JB
1939 }
1940 }
1941
be0676b3 1942 /**
fcd39179 1943 * This method is the message listener registered on each worker.
be0676b3 1944 */
3a776b6d
JB
1945 protected readonly workerMessageListener = (
1946 message: MessageValue<Response>
1947 ): void => {
fcd39179 1948 this.checkMessageWorkerId(message)
31847469
JB
1949 const { workerId, ready, taskId, taskFunctionsProperties } = message
1950 if (ready != null && taskFunctionsProperties != null) {
fcd39179
JB
1951 // Worker ready response received from worker
1952 this.handleWorkerReadyResponse(message)
31847469
JB
1953 } else if (taskFunctionsProperties != null) {
1954 // Task function properties message received from worker
9a55fa8c
JB
1955 const workerNodeKey = this.getWorkerNodeKeyByWorkerId(workerId)
1956 const workerInfo = this.getWorkerInfo(workerNodeKey)
1851fed0 1957 if (workerInfo != null) {
31847469 1958 workerInfo.taskFunctionsProperties = taskFunctionsProperties
9a55fa8c 1959 this.sendStatisticsMessageToWorker(workerNodeKey)
1851fed0 1960 }
31847469
JB
1961 } else if (taskId != null) {
1962 // Task execution response received from worker
1963 this.handleTaskExecutionResponse(message)
6b272951
JB
1964 }
1965 }
1966
8e8d9101
JB
1967 private checkAndEmitReadyEvent (): void {
1968 if (!this.readyEventEmitted && this.ready) {
1969 this.emitter?.emit(PoolEvents.ready, this.info)
1970 this.readyEventEmitted = true
1971 }
1972 }
1973
10e2aa7e 1974 private handleWorkerReadyResponse (message: MessageValue<Response>): void {
31847469 1975 const { workerId, ready, taskFunctionsProperties } = message
e44639e9 1976 if (ready == null || !ready) {
c63a35a0 1977 throw new Error(`Worker ${workerId} failed to initialize`)
f05ed93c 1978 }
9a55fa8c
JB
1979 const workerNodeKey = this.getWorkerNodeKeyByWorkerId(workerId)
1980 const workerNode = this.workerNodes[workerNodeKey]
e44639e9 1981 workerNode.info.ready = ready
31847469 1982 workerNode.info.taskFunctionsProperties = taskFunctionsProperties
9a55fa8c 1983 this.sendStatisticsMessageToWorker(workerNodeKey)
8e8d9101 1984 this.checkAndEmitReadyEvent()
6b272951
JB
1985 }
1986
1987 private handleTaskExecutionResponse (message: MessageValue<Response>): void {
463226a4 1988 const { workerId, taskId, workerError, data } = message
67f3f2d6
JB
1989 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1990 const promiseResponse = this.promiseResponseMap.get(taskId!)
6b272951 1991 if (promiseResponse != null) {
f18fd12b 1992 const { resolve, reject, workerNodeKey, asyncResource } = promiseResponse
9b358e72 1993 const workerNode = this.workerNodes[workerNodeKey]
6703b9f4
JB
1994 if (workerError != null) {
1995 this.emitter?.emit(PoolEvents.taskError, workerError)
f18fd12b
JB
1996 asyncResource != null
1997 ? asyncResource.runInAsyncScope(
1998 reject,
1999 this.emitter,
2000 workerError.message
2001 )
2002 : reject(workerError.message)
6b272951 2003 } else {
f18fd12b
JB
2004 asyncResource != null
2005 ? asyncResource.runInAsyncScope(resolve, this.emitter, data)
2006 : resolve(data as Response)
6b272951 2007 }
f18fd12b 2008 asyncResource?.emitDestroy()
501aea93 2009 this.afterTaskExecutionHook(workerNodeKey, message)
67f3f2d6
JB
2010 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2011 this.promiseResponseMap.delete(taskId!)
cdaecaee
JB
2012 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
2013 workerNode?.emit('taskFinished', taskId)
16d7e943
JB
2014 if (
2015 this.opts.enableTasksQueue === true &&
2016 !this.destroying &&
2017 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
2018 workerNode != null
2019 ) {
9b358e72 2020 const workerNodeTasksUsage = workerNode.usage.tasks
463226a4
JB
2021 if (
2022 this.tasksQueueSize(workerNodeKey) > 0 &&
2023 workerNodeTasksUsage.executing <
67f3f2d6
JB
2024 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2025 this.opts.tasksQueueOptions!.concurrency!
463226a4 2026 ) {
67f3f2d6
JB
2027 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2028 this.executeTask(workerNodeKey, this.dequeueTask(workerNodeKey)!)
463226a4
JB
2029 }
2030 if (
2031 workerNodeTasksUsage.executing === 0 &&
2032 this.tasksQueueSize(workerNodeKey) === 0 &&
2033 workerNodeTasksUsage.sequentiallyStolen === 0
2034 ) {
e44639e9 2035 workerNode.emit('idle', {
7f0e1334 2036 workerId,
e1c2dba7
JB
2037 workerNodeKey
2038 })
463226a4 2039 }
be0676b3
APA
2040 }
2041 }
be0676b3 2042 }
7c0ba920 2043
a1763c54 2044 private checkAndEmitTaskExecutionEvents (): void {
33e6bb4c
JB
2045 if (this.busy) {
2046 this.emitter?.emit(PoolEvents.busy, this.info)
a1763c54
JB
2047 }
2048 }
2049
2050 private checkAndEmitTaskQueuingEvents (): void {
2051 if (this.hasBackPressure()) {
2052 this.emitter?.emit(PoolEvents.backPressure, this.info)
164d950a
JB
2053 }
2054 }
2055
d0878034
JB
2056 /**
2057 * Emits dynamic worker creation events.
2058 */
2059 protected abstract checkAndEmitDynamicWorkerCreationEvents (): void
33e6bb4c 2060
8a1260a3 2061 /**
aa9eede8 2062 * Gets the worker information given its worker node key.
8a1260a3
JB
2063 *
2064 * @param workerNodeKey - The worker node key.
3f09ed9f 2065 * @returns The worker information.
8a1260a3 2066 */
1851fed0
JB
2067 protected getWorkerInfo (workerNodeKey: number): WorkerInfo | undefined {
2068 return this.workerNodes[workerNodeKey]?.info
e221309a
JB
2069 }
2070
a05c10de 2071 /**
c3719753 2072 * Creates a worker node.
ea7a90d3 2073 *
c3719753 2074 * @returns The created worker node.
ea7a90d3 2075 */
c3719753 2076 private createWorkerNode (): IWorkerNode<Worker, Data> {
671d5154 2077 const workerNode = new WorkerNode<Worker, Data>(
c3719753
JB
2078 this.worker,
2079 this.filePath,
2080 {
2081 env: this.opts.env,
2082 workerOptions: this.opts.workerOptions,
2083 tasksQueueBackPressureSize:
32b141fd 2084 this.opts.tasksQueueOptions?.size ??
26ce26ca
JB
2085 getDefaultTasksQueueOptions(
2086 this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers
95d1a734
JB
2087 ).size,
2088 tasksQueueBucketSize:
2089 (this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers) * 2
c3719753 2090 }
671d5154 2091 )
b97d82d8 2092 // Flag the worker node as ready at pool startup.
d2c73f82
JB
2093 if (this.starting) {
2094 workerNode.info.ready = true
2095 }
c3719753
JB
2096 return workerNode
2097 }
2098
2099 /**
2100 * Adds the given worker node in the pool worker nodes.
2101 *
2102 * @param workerNode - The worker node.
2103 * @returns The added worker node key.
2104 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the added worker node is not found.
2105 */
2106 private addWorkerNode (workerNode: IWorkerNode<Worker, Data>): number {
aa9eede8 2107 this.workerNodes.push(workerNode)
c3719753 2108 const workerNodeKey = this.workerNodes.indexOf(workerNode)
aa9eede8 2109 if (workerNodeKey === -1) {
86ed0598 2110 throw new Error('Worker added not found in worker nodes')
aa9eede8
JB
2111 }
2112 return workerNodeKey
ea7a90d3 2113 }
c923ce56 2114
8e8d9101
JB
2115 private checkAndEmitEmptyEvent (): void {
2116 if (this.empty) {
2117 this.emitter?.emit(PoolEvents.empty, this.info)
2118 this.readyEventEmitted = false
2119 }
2120 }
2121
51fe3d3c 2122 /**
9974369e 2123 * Removes the worker node from the pool worker nodes.
51fe3d3c 2124 *
9974369e 2125 * @param workerNode - The worker node.
51fe3d3c 2126 */
9974369e
JB
2127 private removeWorkerNode (workerNode: IWorkerNode<Worker, Data>): void {
2128 const workerNodeKey = this.workerNodes.indexOf(workerNode)
1f68cede
JB
2129 if (workerNodeKey !== -1) {
2130 this.workerNodes.splice(workerNodeKey, 1)
bcfb06ce 2131 this.workerChoiceStrategiesContext?.remove(workerNodeKey)
1f68cede 2132 }
8e8d9101 2133 this.checkAndEmitEmptyEvent()
51fe3d3c 2134 }
adc3c320 2135
ae3ab61d 2136 protected flagWorkerNodeAsNotReady (workerNodeKey: number): void {
1851fed0
JB
2137 const workerInfo = this.getWorkerInfo(workerNodeKey)
2138 if (workerInfo != null) {
2139 workerInfo.ready = false
2140 }
ae3ab61d
JB
2141 }
2142
9e844245
JB
2143 private hasBackPressure (): boolean {
2144 return (
2145 this.opts.enableTasksQueue === true &&
2146 this.workerNodes.findIndex(
041dc05b 2147 workerNode => !workerNode.hasBackPressure()
a1763c54 2148 ) === -1
9e844245 2149 )
e2b31e32
JB
2150 }
2151
b0a4db63 2152 /**
aa9eede8 2153 * Executes the given task on the worker given its worker node key.
b0a4db63 2154 *
aa9eede8 2155 * @param workerNodeKey - The worker node key.
b0a4db63
JB
2156 * @param task - The task to execute.
2157 */
2e81254d 2158 private executeTask (workerNodeKey: number, task: Task<Data>): void {
1c6fe997 2159 this.beforeTaskExecutionHook(workerNodeKey, task)
bbfa38a2 2160 this.sendToWorker(workerNodeKey, task, task.transferList)
a1763c54 2161 this.checkAndEmitTaskExecutionEvents()
2e81254d
JB
2162 }
2163
f9f00b5f 2164 private enqueueTask (workerNodeKey: number, task: Task<Data>): number {
a1763c54
JB
2165 const tasksQueueSize = this.workerNodes[workerNodeKey].enqueueTask(task)
2166 this.checkAndEmitTaskQueuingEvents()
2167 return tasksQueueSize
adc3c320
JB
2168 }
2169
0d4e88b3
JB
2170 private dequeueTask (workerNodeKey: number): Task<Data> | undefined {
2171 return this.workerNodes[workerNodeKey].dequeueTask()
adc3c320
JB
2172 }
2173
416fd65c 2174 private tasksQueueSize (workerNodeKey: number): number {
4b628b48 2175 return this.workerNodes[workerNodeKey].tasksQueueSize()
df593701
JB
2176 }
2177
87347ea8
JB
2178 protected flushTasksQueue (workerNodeKey: number): number {
2179 let flushedTasks = 0
920278a2 2180 while (this.tasksQueueSize(workerNodeKey) > 0) {
67f3f2d6
JB
2181 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2182 this.executeTask(workerNodeKey, this.dequeueTask(workerNodeKey)!)
87347ea8 2183 ++flushedTasks
ff733df7 2184 }
4b628b48 2185 this.workerNodes[workerNodeKey].clearTasksQueue()
87347ea8 2186 return flushedTasks
ff733df7
JB
2187 }
2188
ef41a6e6 2189 private flushTasksQueues (): void {
19b8be8b 2190 for (const workerNodeKey of this.workerNodes.keys()) {
ef41a6e6
JB
2191 this.flushTasksQueue(workerNodeKey)
2192 }
2193 }
c97c7edb 2194}