Merge dependabot/npm_and_yarn/examples/typescript/websocket-server-pool/ws-hybrid...
[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 })
be202c2c
JB
541 },
542 utilization: {
543 average: round(
544 average(
545 this.workerNodes.map(
546 workerNode => workerNode.usage.elu.utilization ?? 0
547 )
548 )
549 ),
550 median: round(
551 median(
552 this.workerNodes.map(
553 workerNode => workerNode.usage.elu.utilization ?? 0
554 )
555 )
556 )
533a8e22
JB
557 }
558 }
1dcf8b7b 559 })
6b27d407
JB
560 }
561 }
08f3f44c 562
aa9eede8
JB
563 /**
564 * The pool readiness boolean status.
565 */
2431bdb4 566 private get ready (): boolean {
8e8d9101
JB
567 if (this.empty) {
568 return false
569 }
2431bdb4 570 return (
b97d82d8
JB
571 this.workerNodes.reduce(
572 (accumulator, workerNode) =>
573 !workerNode.info.dynamic && workerNode.info.ready
574 ? accumulator + 1
575 : accumulator,
576 0
26ce26ca 577 ) >= this.minimumNumberOfWorkers
2431bdb4
JB
578 )
579 }
580
8e8d9101
JB
581 /**
582 * The pool emptiness boolean status.
583 */
584 protected get empty (): boolean {
fb43035c 585 return this.minimumNumberOfWorkers === 0 && this.workerNodes.length === 0
8e8d9101
JB
586 }
587
afe0d5bf 588 /**
aa9eede8 589 * The approximate pool utilization.
afe0d5bf
JB
590 *
591 * @returns The pool utilization.
592 */
593 private get utilization (): number {
97dc65d9
JB
594 if (this.startTimestamp == null) {
595 return 0
596 }
8e5ca040 597 const poolTimeCapacity =
26ce26ca
JB
598 (performance.now() - this.startTimestamp) *
599 (this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers)
afe0d5bf
JB
600 const totalTasksRunTime = this.workerNodes.reduce(
601 (accumulator, workerNode) =>
c63a35a0 602 accumulator + (workerNode.usage.runTime.aggregate ?? 0),
afe0d5bf
JB
603 0
604 )
605 const totalTasksWaitTime = this.workerNodes.reduce(
606 (accumulator, workerNode) =>
c63a35a0 607 accumulator + (workerNode.usage.waitTime.aggregate ?? 0),
afe0d5bf
JB
608 0
609 )
8e5ca040 610 return (totalTasksRunTime + totalTasksWaitTime) / poolTimeCapacity
afe0d5bf
JB
611 }
612
8881ae32 613 /**
aa9eede8 614 * The pool type.
8881ae32
JB
615 *
616 * If it is `'dynamic'`, it provides the `max` property.
617 */
618 protected abstract get type (): PoolType
619
184855e6 620 /**
aa9eede8 621 * The worker type.
184855e6
JB
622 */
623 protected abstract get worker (): WorkerType
624
6b813701
JB
625 /**
626 * Checks if the worker id sent in the received message from a worker is valid.
627 *
628 * @param message - The received message.
629 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the worker id is invalid.
630 */
4d159167 631 private checkMessageWorkerId (message: MessageValue<Data | Response>): void {
310de0aa
JB
632 if (message.workerId == null) {
633 throw new Error('Worker message received without worker id')
8e5a7cf9 634 } else if (this.getWorkerNodeKeyByWorkerId(message.workerId) === -1) {
21f710aa
JB
635 throw new Error(
636 `Worker message received from unknown worker '${message.workerId}'`
637 )
638 }
639 }
640
aa9eede8
JB
641 /**
642 * Gets the worker node key given its worker id.
643 *
644 * @param workerId - The worker id.
aad6fb64 645 * @returns The worker node key if the worker id is found in the pool worker nodes, `-1` otherwise.
aa9eede8 646 */
72ae84a2 647 private getWorkerNodeKeyByWorkerId (workerId: number | undefined): number {
aad6fb64 648 return this.workerNodes.findIndex(
041dc05b 649 workerNode => workerNode.info.id === workerId
aad6fb64 650 )
aa9eede8
JB
651 }
652
afc003b2 653 /** @inheritDoc */
a35560ba 654 public setWorkerChoiceStrategy (
59219cbb
JB
655 workerChoiceStrategy: WorkerChoiceStrategy,
656 workerChoiceStrategyOptions?: WorkerChoiceStrategyOptions
a35560ba 657 ): void {
19b8be8b 658 let requireSync = false
bde6b5d7 659 checkValidWorkerChoiceStrategy(workerChoiceStrategy)
85bbc7ab 660 if (workerChoiceStrategyOptions != null) {
423f3b7a 661 requireSync = !this.setWorkerChoiceStrategyOptions(
19b8be8b
JB
662 workerChoiceStrategyOptions
663 )
85bbc7ab
JB
664 }
665 if (workerChoiceStrategy !== this.opts.workerChoiceStrategy) {
666 this.opts.workerChoiceStrategy = workerChoiceStrategy
667 this.workerChoiceStrategiesContext?.setDefaultWorkerChoiceStrategy(
668 this.opts.workerChoiceStrategy,
669 this.opts.workerChoiceStrategyOptions
670 )
19b8be8b
JB
671 requireSync = true
672 }
673 if (requireSync) {
85bbc7ab 674 this.workerChoiceStrategiesContext?.syncWorkerChoiceStrategies(
91041638 675 this.getWorkerChoiceStrategies(),
85bbc7ab
JB
676 this.opts.workerChoiceStrategyOptions
677 )
19b8be8b 678 for (const workerNodeKey of this.workerNodes.keys()) {
85bbc7ab
JB
679 this.sendStatisticsMessageToWorker(workerNodeKey)
680 }
59219cbb 681 }
a20f0ba5
JB
682 }
683
684 /** @inheritDoc */
685 public setWorkerChoiceStrategyOptions (
c63a35a0 686 workerChoiceStrategyOptions: WorkerChoiceStrategyOptions | undefined
19b8be8b 687 ): boolean {
0d80593b 688 this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions)
26ce26ca
JB
689 if (workerChoiceStrategyOptions != null) {
690 this.opts.workerChoiceStrategyOptions = workerChoiceStrategyOptions
19b8be8b
JB
691 this.workerChoiceStrategiesContext?.setOptions(
692 this.opts.workerChoiceStrategyOptions
693 )
694 this.workerChoiceStrategiesContext?.syncWorkerChoiceStrategies(
91041638 695 this.getWorkerChoiceStrategies(),
19b8be8b
JB
696 this.opts.workerChoiceStrategyOptions
697 )
698 for (const workerNodeKey of this.workerNodes.keys()) {
699 this.sendStatisticsMessageToWorker(workerNodeKey)
700 }
701 return true
8990357d 702 }
19b8be8b 703 return false
a35560ba
S
704 }
705
a20f0ba5 706 /** @inheritDoc */
8f52842f
JB
707 public enableTasksQueue (
708 enable: boolean,
709 tasksQueueOptions?: TasksQueueOptions
710 ): void {
a20f0ba5 711 if (this.opts.enableTasksQueue === true && !enable) {
d6ca1416
JB
712 this.unsetTaskStealing()
713 this.unsetTasksStealingOnBackPressure()
ef41a6e6 714 this.flushTasksQueues()
a20f0ba5
JB
715 }
716 this.opts.enableTasksQueue = enable
c63a35a0 717 this.setTasksQueueOptions(tasksQueueOptions)
a20f0ba5
JB
718 }
719
720 /** @inheritDoc */
c63a35a0
JB
721 public setTasksQueueOptions (
722 tasksQueueOptions: TasksQueueOptions | undefined
723 ): void {
a20f0ba5 724 if (this.opts.enableTasksQueue === true) {
bde6b5d7 725 checkValidTasksQueueOptions(tasksQueueOptions)
8f52842f
JB
726 this.opts.tasksQueueOptions =
727 this.buildTasksQueueOptions(tasksQueueOptions)
67f3f2d6
JB
728 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
729 this.setTasksQueueSize(this.opts.tasksQueueOptions.size!)
d6ca1416 730 if (this.opts.tasksQueueOptions.taskStealing === true) {
9f99eb9b 731 this.unsetTaskStealing()
d6ca1416
JB
732 this.setTaskStealing()
733 } else {
734 this.unsetTaskStealing()
735 }
736 if (this.opts.tasksQueueOptions.tasksStealingOnBackPressure === true) {
9f99eb9b 737 this.unsetTasksStealingOnBackPressure()
d6ca1416
JB
738 this.setTasksStealingOnBackPressure()
739 } else {
740 this.unsetTasksStealingOnBackPressure()
741 }
5baee0d7 742 } else if (this.opts.tasksQueueOptions != null) {
a20f0ba5
JB
743 delete this.opts.tasksQueueOptions
744 }
745 }
746
9b38ab2d 747 private buildTasksQueueOptions (
c63a35a0 748 tasksQueueOptions: TasksQueueOptions | undefined
9b38ab2d
JB
749 ): TasksQueueOptions {
750 return {
26ce26ca
JB
751 ...getDefaultTasksQueueOptions(
752 this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers
753 ),
9b38ab2d
JB
754 ...tasksQueueOptions
755 }
756 }
757
5b49e864 758 private setTasksQueueSize (size: number): void {
20c6f652 759 for (const workerNode of this.workerNodes) {
ff3f866a 760 workerNode.tasksQueueBackPressureSize = size
20c6f652
JB
761 }
762 }
763
d6ca1416 764 private setTaskStealing (): void {
19b8be8b 765 for (const workerNodeKey of this.workerNodes.keys()) {
e44639e9 766 this.workerNodes[workerNodeKey].on('idle', this.handleWorkerNodeIdleEvent)
d6ca1416
JB
767 }
768 }
769
770 private unsetTaskStealing (): void {
19b8be8b 771 for (const workerNodeKey of this.workerNodes.keys()) {
e1c2dba7 772 this.workerNodes[workerNodeKey].off(
e44639e9
JB
773 'idle',
774 this.handleWorkerNodeIdleEvent
9f95d5eb 775 )
d6ca1416
JB
776 }
777 }
778
779 private setTasksStealingOnBackPressure (): void {
19b8be8b 780 for (const workerNodeKey of this.workerNodes.keys()) {
e1c2dba7 781 this.workerNodes[workerNodeKey].on(
b5e75be8 782 'backPressure',
e44639e9 783 this.handleWorkerNodeBackPressureEvent
9f95d5eb 784 )
d6ca1416
JB
785 }
786 }
787
788 private unsetTasksStealingOnBackPressure (): void {
19b8be8b 789 for (const workerNodeKey of this.workerNodes.keys()) {
e1c2dba7 790 this.workerNodes[workerNodeKey].off(
b5e75be8 791 'backPressure',
e44639e9 792 this.handleWorkerNodeBackPressureEvent
9f95d5eb 793 )
d6ca1416
JB
794 }
795 }
796
c319c66b
JB
797 /**
798 * Whether the pool is full or not.
799 *
800 * The pool filling boolean status.
801 */
dea903a8 802 protected get full (): boolean {
26ce26ca
JB
803 return (
804 this.workerNodes.length >=
805 (this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers)
806 )
dea903a8 807 }
c2ade475 808
c319c66b
JB
809 /**
810 * Whether the pool is busy or not.
811 *
812 * The pool busyness boolean status.
813 */
814 protected abstract get busy (): boolean
7c0ba920 815
6c6afb84 816 /**
3d76750a 817 * Whether worker nodes are executing concurrently their tasks quota or not.
6c6afb84
JB
818 *
819 * @returns Worker nodes busyness boolean status.
820 */
c2ade475 821 protected internalBusy (): boolean {
3d76750a
JB
822 if (this.opts.enableTasksQueue === true) {
823 return (
824 this.workerNodes.findIndex(
041dc05b 825 workerNode =>
3d76750a
JB
826 workerNode.info.ready &&
827 workerNode.usage.tasks.executing <
67f3f2d6
JB
828 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
829 this.opts.tasksQueueOptions!.concurrency!
3d76750a
JB
830 ) === -1
831 )
3d76750a 832 }
419e8121
JB
833 return (
834 this.workerNodes.findIndex(
835 workerNode =>
836 workerNode.info.ready && workerNode.usage.tasks.executing === 0
837 ) === -1
838 )
cb70b19d
JB
839 }
840
42c677c1
JB
841 private isWorkerNodeBusy (workerNodeKey: number): boolean {
842 if (this.opts.enableTasksQueue === true) {
843 return (
844 this.workerNodes[workerNodeKey].usage.tasks.executing >=
67f3f2d6
JB
845 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
846 this.opts.tasksQueueOptions!.concurrency!
42c677c1
JB
847 )
848 }
849 return this.workerNodes[workerNodeKey].usage.tasks.executing > 0
850 }
851
e81c38f2 852 private async sendTaskFunctionOperationToWorker (
72ae84a2
JB
853 workerNodeKey: number,
854 message: MessageValue<Data>
855 ): Promise<boolean> {
72ae84a2 856 return await new Promise<boolean>((resolve, reject) => {
ae036c3e
JB
857 const taskFunctionOperationListener = (
858 message: MessageValue<Response>
859 ): void => {
860 this.checkMessageWorkerId(message)
1851fed0 861 const workerId = this.getWorkerInfo(workerNodeKey)?.id
72ae84a2 862 if (
ae036c3e
JB
863 message.taskFunctionOperationStatus != null &&
864 message.workerId === workerId
72ae84a2 865 ) {
ae036c3e
JB
866 if (message.taskFunctionOperationStatus) {
867 resolve(true)
c63a35a0 868 } else {
ae036c3e
JB
869 reject(
870 new Error(
c63a35a0 871 `Task function operation '${message.taskFunctionOperation}' failed on worker ${message.workerId} with error: '${message.workerError?.message}'`
ae036c3e 872 )
72ae84a2 873 )
ae036c3e
JB
874 }
875 this.deregisterWorkerMessageListener(
876 this.getWorkerNodeKeyByWorkerId(message.workerId),
877 taskFunctionOperationListener
72ae84a2
JB
878 )
879 }
ae036c3e
JB
880 }
881 this.registerWorkerMessageListener(
882 workerNodeKey,
883 taskFunctionOperationListener
884 )
72ae84a2
JB
885 this.sendToWorker(workerNodeKey, message)
886 })
887 }
888
889 private async sendTaskFunctionOperationToWorkers (
adee6053 890 message: MessageValue<Data>
e81c38f2
JB
891 ): Promise<boolean> {
892 return await new Promise<boolean>((resolve, reject) => {
ae036c3e
JB
893 const responsesReceived = new Array<MessageValue<Response>>()
894 const taskFunctionOperationsListener = (
895 message: MessageValue<Response>
896 ): void => {
897 this.checkMessageWorkerId(message)
898 if (message.taskFunctionOperationStatus != null) {
899 responsesReceived.push(message)
900 if (responsesReceived.length === this.workerNodes.length) {
e81c38f2 901 if (
e81c38f2
JB
902 responsesReceived.every(
903 message => message.taskFunctionOperationStatus === true
904 )
905 ) {
906 resolve(true)
907 } else if (
e81c38f2
JB
908 responsesReceived.some(
909 message => message.taskFunctionOperationStatus === false
910 )
911 ) {
b0b55f57
JB
912 const errorResponse = responsesReceived.find(
913 response => response.taskFunctionOperationStatus === false
914 )
e81c38f2
JB
915 reject(
916 new Error(
b0b55f57 917 `Task function operation '${
e81c38f2 918 message.taskFunctionOperation as string
c63a35a0
JB
919 }' failed on worker ${errorResponse?.workerId} with error: '${
920 errorResponse?.workerError?.message
b0b55f57 921 }'`
e81c38f2
JB
922 )
923 )
924 }
ae036c3e
JB
925 this.deregisterWorkerMessageListener(
926 this.getWorkerNodeKeyByWorkerId(message.workerId),
927 taskFunctionOperationsListener
928 )
e81c38f2 929 }
ae036c3e
JB
930 }
931 }
19b8be8b 932 for (const workerNodeKey of this.workerNodes.keys()) {
ae036c3e
JB
933 this.registerWorkerMessageListener(
934 workerNodeKey,
935 taskFunctionOperationsListener
936 )
72ae84a2 937 this.sendToWorker(workerNodeKey, message)
e81c38f2
JB
938 }
939 })
6703b9f4
JB
940 }
941
942 /** @inheritDoc */
943 public hasTaskFunction (name: string): boolean {
85bbc7ab
JB
944 return this.listTaskFunctionsProperties().some(
945 taskFunctionProperties => taskFunctionProperties.name === name
946 )
6703b9f4
JB
947 }
948
949 /** @inheritDoc */
e81c38f2
JB
950 public async addTaskFunction (
951 name: string,
31847469 952 fn: TaskFunction<Data, Response> | TaskFunctionObject<Data, Response>
e81c38f2 953 ): Promise<boolean> {
3feeab69
JB
954 if (typeof name !== 'string') {
955 throw new TypeError('name argument must be a string')
956 }
957 if (typeof name === 'string' && name.trim().length === 0) {
958 throw new TypeError('name argument must not be an empty string')
959 }
31847469
JB
960 if (typeof fn === 'function') {
961 fn = { taskFunction: fn } satisfies TaskFunctionObject<Data, Response>
962 }
963 if (typeof fn.taskFunction !== 'function') {
964 throw new TypeError('taskFunction property must be a function')
3feeab69 965 }
95d1a734
JB
966 checkValidPriority(fn.priority)
967 checkValidWorkerChoiceStrategy(fn.strategy)
adee6053 968 const opResult = await this.sendTaskFunctionOperationToWorkers({
6703b9f4 969 taskFunctionOperation: 'add',
31847469
JB
970 taskFunctionProperties: buildTaskFunctionProperties(name, fn),
971 taskFunction: fn.taskFunction.toString()
6703b9f4 972 })
adee6053 973 this.taskFunctions.set(name, fn)
85bbc7ab 974 this.workerChoiceStrategiesContext?.syncWorkerChoiceStrategies(
91041638 975 this.getWorkerChoiceStrategies()
85bbc7ab 976 )
9a55fa8c
JB
977 for (const workerNodeKey of this.workerNodes.keys()) {
978 this.sendStatisticsMessageToWorker(workerNodeKey)
979 }
adee6053 980 return opResult
6703b9f4
JB
981 }
982
983 /** @inheritDoc */
e81c38f2 984 public async removeTaskFunction (name: string): Promise<boolean> {
9eae3c69
JB
985 if (!this.taskFunctions.has(name)) {
986 throw new Error(
16248b23 987 'Cannot remove a task function not handled on the pool side'
9eae3c69
JB
988 )
989 }
adee6053 990 const opResult = await this.sendTaskFunctionOperationToWorkers({
6703b9f4 991 taskFunctionOperation: 'remove',
31847469
JB
992 taskFunctionProperties: buildTaskFunctionProperties(
993 name,
994 this.taskFunctions.get(name)
995 )
6703b9f4 996 })
9a55fa8c
JB
997 for (const workerNode of this.workerNodes) {
998 workerNode.deleteTaskFunctionWorkerUsage(name)
999 }
adee6053 1000 this.taskFunctions.delete(name)
85bbc7ab 1001 this.workerChoiceStrategiesContext?.syncWorkerChoiceStrategies(
91041638 1002 this.getWorkerChoiceStrategies()
85bbc7ab 1003 )
9a55fa8c
JB
1004 for (const workerNodeKey of this.workerNodes.keys()) {
1005 this.sendStatisticsMessageToWorker(workerNodeKey)
1006 }
adee6053 1007 return opResult
6703b9f4
JB
1008 }
1009
90d7d101 1010 /** @inheritDoc */
31847469 1011 public listTaskFunctionsProperties (): TaskFunctionProperties[] {
f2dbbf95
JB
1012 for (const workerNode of this.workerNodes) {
1013 if (
31847469
JB
1014 Array.isArray(workerNode.info.taskFunctionsProperties) &&
1015 workerNode.info.taskFunctionsProperties.length > 0
f2dbbf95 1016 ) {
31847469 1017 return workerNode.info.taskFunctionsProperties
f2dbbf95 1018 }
90d7d101 1019 }
f2dbbf95 1020 return []
90d7d101
JB
1021 }
1022
bcfb06ce 1023 /**
91041638 1024 * Gets task function worker choice strategy, if any.
bcfb06ce
JB
1025 *
1026 * @param name - The task function name.
1027 * @returns The task function worker choice strategy if the task function worker choice strategy is defined, `undefined` otherwise.
1028 */
91041638 1029 private readonly getTaskFunctionWorkerChoiceStrategy = (
bcfb06ce
JB
1030 name?: string
1031 ): WorkerChoiceStrategy | undefined => {
91041638
JB
1032 name = name ?? DEFAULT_TASK_NAME
1033 const taskFunctionsProperties = this.listTaskFunctionsProperties()
1034 if (name === DEFAULT_TASK_NAME) {
1035 name = taskFunctionsProperties[1]?.name
bcfb06ce 1036 }
91041638
JB
1037 return taskFunctionsProperties.find(
1038 (taskFunctionProperties: TaskFunctionProperties) =>
1039 taskFunctionProperties.name === name
1040 )?.strategy
1041 }
1042
1043 /**
1044 * Gets worker node task function worker choice strategy, if any.
1045 *
1046 * @param workerNodeKey - The worker node key.
1047 * @param name - The task function name.
1048 * @returns The worker node task function worker choice strategy if the worker node task function worker choice strategy is defined, `undefined` otherwise.
1049 */
1050 private readonly getWorkerNodeTaskFunctionWorkerChoiceStrategy = (
1051 workerNodeKey: number,
1052 name?: string
1053 ): WorkerChoiceStrategy | undefined => {
1054 const workerInfo = this.getWorkerInfo(workerNodeKey)
1055 if (workerInfo == null) {
1056 return
1057 }
1058 name = name ?? DEFAULT_TASK_NAME
1059 if (name === DEFAULT_TASK_NAME) {
1060 name = workerInfo.taskFunctionsProperties?.[1]?.name
1061 }
1062 return workerInfo.taskFunctionsProperties?.find(
1063 (taskFunctionProperties: TaskFunctionProperties) =>
1064 taskFunctionProperties.name === name
1065 )?.strategy
bcfb06ce
JB
1066 }
1067
95d1a734
JB
1068 /**
1069 * Gets worker node task function priority, if any.
1070 *
1071 * @param workerNodeKey - The worker node key.
1072 * @param name - The task function name.
a20062fa 1073 * @returns The worker node task function priority if the worker node task function priority is defined, `undefined` otherwise.
95d1a734
JB
1074 */
1075 private readonly getWorkerNodeTaskFunctionPriority = (
1076 workerNodeKey: number,
1077 name?: string
1078 ): number | undefined => {
91041638
JB
1079 const workerInfo = this.getWorkerInfo(workerNodeKey)
1080 if (workerInfo == null) {
1081 return
95d1a734 1082 }
91041638
JB
1083 name = name ?? DEFAULT_TASK_NAME
1084 if (name === DEFAULT_TASK_NAME) {
1085 name = workerInfo.taskFunctionsProperties?.[1]?.name
1086 }
1087 return workerInfo.taskFunctionsProperties?.find(
1088 (taskFunctionProperties: TaskFunctionProperties) =>
1089 taskFunctionProperties.name === name
1090 )?.priority
95d1a734
JB
1091 }
1092
85bbc7ab
JB
1093 /**
1094 * Gets the worker choice strategies registered in this pool.
1095 *
1096 * @returns The worker choice strategies.
1097 */
91041638 1098 private readonly getWorkerChoiceStrategies =
85bbc7ab
JB
1099 (): Set<WorkerChoiceStrategy> => {
1100 return new Set([
1101 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1102 this.opts.workerChoiceStrategy!,
1103 ...(this.listTaskFunctionsProperties()
1104 .map(
1105 (taskFunctionProperties: TaskFunctionProperties) =>
1106 taskFunctionProperties.strategy
1107 )
1108 .filter(
1109 (strategy: WorkerChoiceStrategy | undefined) => strategy != null
1110 ) as WorkerChoiceStrategy[])
1111 ])
1112 }
1113
6703b9f4 1114 /** @inheritDoc */
e81c38f2 1115 public async setDefaultTaskFunction (name: string): Promise<boolean> {
72ae84a2 1116 return await this.sendTaskFunctionOperationToWorkers({
6703b9f4 1117 taskFunctionOperation: 'default',
31847469
JB
1118 taskFunctionProperties: buildTaskFunctionProperties(
1119 name,
1120 this.taskFunctions.get(name)
1121 )
6703b9f4 1122 })
6703b9f4
JB
1123 }
1124
375f7504
JB
1125 private shallExecuteTask (workerNodeKey: number): boolean {
1126 return (
1127 this.tasksQueueSize(workerNodeKey) === 0 &&
1128 this.workerNodes[workerNodeKey].usage.tasks.executing <
67f3f2d6
JB
1129 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1130 this.opts.tasksQueueOptions!.concurrency!
375f7504
JB
1131 )
1132 }
1133
afc003b2 1134 /** @inheritDoc */
7d91a8cd
JB
1135 public async execute (
1136 data?: Data,
1137 name?: string,
6a3ecc50 1138 transferList?: readonly TransferListItem[]
7d91a8cd 1139 ): Promise<Response> {
52b71763 1140 return await new Promise<Response>((resolve, reject) => {
15b176e0 1141 if (!this.started) {
47352846 1142 reject(new Error('Cannot execute a task on not started pool'))
9d2d0da1 1143 return
15b176e0 1144 }
711623b8
JB
1145 if (this.destroying) {
1146 reject(new Error('Cannot execute a task on destroying pool'))
1147 return
1148 }
7d91a8cd
JB
1149 if (name != null && typeof name !== 'string') {
1150 reject(new TypeError('name argument must be a string'))
9d2d0da1 1151 return
7d91a8cd 1152 }
90d7d101
JB
1153 if (
1154 name != null &&
1155 typeof name === 'string' &&
1156 name.trim().length === 0
1157 ) {
f58b60b9 1158 reject(new TypeError('name argument must not be an empty string'))
9d2d0da1 1159 return
90d7d101 1160 }
b558f6b5
JB
1161 if (transferList != null && !Array.isArray(transferList)) {
1162 reject(new TypeError('transferList argument must be an array'))
9d2d0da1 1163 return
b558f6b5
JB
1164 }
1165 const timestamp = performance.now()
91041638 1166 const workerNodeKey = this.chooseWorkerNode(name)
501aea93 1167 const task: Task<Data> = {
52b71763
JB
1168 name: name ?? DEFAULT_TASK_NAME,
1169 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
1170 data: data ?? ({} as Data),
95d1a734 1171 priority: this.getWorkerNodeTaskFunctionPriority(workerNodeKey, name),
91041638
JB
1172 strategy: this.getWorkerNodeTaskFunctionWorkerChoiceStrategy(
1173 workerNodeKey,
1174 name
1175 ),
7d91a8cd 1176 transferList,
52b71763 1177 timestamp,
7629bdf1 1178 taskId: randomUUID()
52b71763 1179 }
67f3f2d6
JB
1180 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1181 this.promiseResponseMap.set(task.taskId!, {
2e81254d
JB
1182 resolve,
1183 reject,
f18fd12b
JB
1184 workerNodeKey,
1185 ...(this.emitter != null && {
1186 asyncResource: new AsyncResource('poolifier:task', {
1187 triggerAsyncId: this.emitter.asyncId,
1188 requireManualDestroy: true
1189 })
1190 })
2e81254d 1191 })
52b71763 1192 if (
4e377863
JB
1193 this.opts.enableTasksQueue === false ||
1194 (this.opts.enableTasksQueue === true &&
375f7504 1195 this.shallExecuteTask(workerNodeKey))
52b71763 1196 ) {
501aea93 1197 this.executeTask(workerNodeKey, task)
4e377863
JB
1198 } else {
1199 this.enqueueTask(workerNodeKey, task)
52b71763 1200 }
2e81254d 1201 })
280c2a77 1202 }
c97c7edb 1203
60dc0a9c
JB
1204 /**
1205 * Starts the minimum number of workers.
1206 */
e0843544 1207 private startMinimumNumberOfWorkers (initWorkerNodeUsage = false): void {
b362a929 1208 this.startingMinimumNumberOfWorkers = true
60dc0a9c
JB
1209 while (
1210 this.workerNodes.reduce(
1211 (accumulator, workerNode) =>
1212 !workerNode.info.dynamic ? accumulator + 1 : accumulator,
1213 0
1214 ) < this.minimumNumberOfWorkers
1215 ) {
e0843544
JB
1216 const workerNodeKey = this.createAndSetupWorkerNode()
1217 initWorkerNodeUsage &&
1218 this.initWorkerNodeUsage(this.workerNodes[workerNodeKey])
60dc0a9c 1219 }
b362a929 1220 this.startingMinimumNumberOfWorkers = false
60dc0a9c
JB
1221 }
1222
47352846
JB
1223 /** @inheritdoc */
1224 public start (): void {
711623b8
JB
1225 if (this.started) {
1226 throw new Error('Cannot start an already started pool')
1227 }
1228 if (this.starting) {
1229 throw new Error('Cannot start an already starting pool')
1230 }
1231 if (this.destroying) {
1232 throw new Error('Cannot start a destroying pool')
1233 }
47352846 1234 this.starting = true
60dc0a9c 1235 this.startMinimumNumberOfWorkers()
2008bccd 1236 this.startTimestamp = performance.now()
47352846
JB
1237 this.starting = false
1238 this.started = true
1239 }
1240
afc003b2 1241 /** @inheritDoc */
c97c7edb 1242 public async destroy (): Promise<void> {
711623b8
JB
1243 if (!this.started) {
1244 throw new Error('Cannot destroy an already destroyed pool')
1245 }
1246 if (this.starting) {
1247 throw new Error('Cannot destroy an starting pool')
1248 }
1249 if (this.destroying) {
1250 throw new Error('Cannot destroy an already destroying pool')
1251 }
1252 this.destroying = true
1fbcaa7c 1253 await Promise.all(
533a8e22 1254 this.workerNodes.map(async (_, workerNodeKey) => {
aa9eede8 1255 await this.destroyWorkerNode(workerNodeKey)
1fbcaa7c
JB
1256 })
1257 )
33e6bb4c 1258 this.emitter?.emit(PoolEvents.destroy, this.info)
f80125ca 1259 this.emitter?.emitDestroy()
55082af9 1260 this.readyEventEmitted = false
2008bccd 1261 delete this.startTimestamp
711623b8 1262 this.destroying = false
15b176e0 1263 this.started = false
c97c7edb
S
1264 }
1265
26ce26ca 1266 private async sendKillMessageToWorker (workerNodeKey: number): Promise<void> {
9edb9717 1267 await new Promise<void>((resolve, reject) => {
c63a35a0
JB
1268 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
1269 if (this.workerNodes[workerNodeKey] == null) {
ad3836ed 1270 resolve()
85b2561d
JB
1271 return
1272 }
ae036c3e
JB
1273 const killMessageListener = (message: MessageValue<Response>): void => {
1274 this.checkMessageWorkerId(message)
1e3214b6
JB
1275 if (message.kill === 'success') {
1276 resolve()
1277 } else if (message.kill === 'failure') {
72ae84a2
JB
1278 reject(
1279 new Error(
c63a35a0 1280 `Kill message handling failed on worker ${message.workerId}`
72ae84a2
JB
1281 )
1282 )
1e3214b6 1283 }
ae036c3e 1284 }
51d9cfbd 1285 // FIXME: should be registered only once
ae036c3e 1286 this.registerWorkerMessageListener(workerNodeKey, killMessageListener)
72ae84a2 1287 this.sendToWorker(workerNodeKey, { kill: true })
1e3214b6 1288 })
1e3214b6
JB
1289 }
1290
4a6952ff 1291 /**
aa9eede8 1292 * Terminates the worker node given its worker node key.
4a6952ff 1293 *
aa9eede8 1294 * @param workerNodeKey - The worker node key.
4a6952ff 1295 */
07e0c9e5
JB
1296 protected async destroyWorkerNode (workerNodeKey: number): Promise<void> {
1297 this.flagWorkerNodeAsNotReady(workerNodeKey)
87347ea8 1298 const flushedTasks = this.flushTasksQueue(workerNodeKey)
07e0c9e5 1299 const workerNode = this.workerNodes[workerNodeKey]
32b141fd
JB
1300 await waitWorkerNodeEvents(
1301 workerNode,
1302 'taskFinished',
1303 flushedTasks,
1304 this.opts.tasksQueueOptions?.tasksFinishedTimeout ??
26ce26ca
JB
1305 getDefaultTasksQueueOptions(
1306 this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers
1307 ).tasksFinishedTimeout
32b141fd 1308 )
07e0c9e5
JB
1309 await this.sendKillMessageToWorker(workerNodeKey)
1310 await workerNode.terminate()
1311 }
c97c7edb 1312
729c563d 1313 /**
6677a3d3
JB
1314 * Setup hook to execute code before worker nodes are created in the abstract constructor.
1315 * Can be overridden.
afc003b2
JB
1316 *
1317 * @virtual
729c563d 1318 */
280c2a77 1319 protected setupHook (): void {
965df41c 1320 /* Intentionally empty */
280c2a77 1321 }
c97c7edb 1322
729c563d 1323 /**
5bec72fd
JB
1324 * Returns whether the worker is the main worker or not.
1325 *
1326 * @returns `true` if the worker is the main worker, `false` otherwise.
280c2a77
S
1327 */
1328 protected abstract isMain (): boolean
1329
1330 /**
2e81254d 1331 * Hook executed before the worker task execution.
bf9549ae 1332 * Can be overridden.
729c563d 1333 *
f06e48d8 1334 * @param workerNodeKey - The worker node key.
1c6fe997 1335 * @param task - The task to execute.
729c563d 1336 */
1c6fe997
JB
1337 protected beforeTaskExecutionHook (
1338 workerNodeKey: number,
1339 task: Task<Data>
1340 ): void {
c63a35a0 1341 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
efabc98d 1342 if (this.workerNodes[workerNodeKey]?.usage != null) {
94407def
JB
1343 const workerUsage = this.workerNodes[workerNodeKey].usage
1344 ++workerUsage.tasks.executing
c329fd41 1345 updateWaitTimeWorkerUsage(
bcfb06ce 1346 this.workerChoiceStrategiesContext,
c329fd41
JB
1347 workerUsage,
1348 task
1349 )
94407def
JB
1350 }
1351 if (
1352 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) &&
67f3f2d6
JB
1353 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1354 this.workerNodes[workerNodeKey].getTaskFunctionWorkerUsage(task.name!) !=
1355 null
94407def 1356 ) {
67f3f2d6 1357 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
db0e38ee 1358 const taskFunctionWorkerUsage = this.workerNodes[
b558f6b5 1359 workerNodeKey
67f3f2d6
JB
1360 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1361 ].getTaskFunctionWorkerUsage(task.name!)!
5623b8d5 1362 ++taskFunctionWorkerUsage.tasks.executing
c329fd41 1363 updateWaitTimeWorkerUsage(
bcfb06ce 1364 this.workerChoiceStrategiesContext,
c329fd41
JB
1365 taskFunctionWorkerUsage,
1366 task
1367 )
b558f6b5 1368 }
c97c7edb
S
1369 }
1370
c01733f1 1371 /**
2e81254d 1372 * Hook executed after the worker task execution.
bf9549ae 1373 * Can be overridden.
c01733f1 1374 *
501aea93 1375 * @param workerNodeKey - The worker node key.
38e795c1 1376 * @param message - The received message.
c01733f1 1377 */
2e81254d 1378 protected afterTaskExecutionHook (
501aea93 1379 workerNodeKey: number,
2740a743 1380 message: MessageValue<Response>
bf9549ae 1381 ): void {
9a55fa8c 1382 let needWorkerChoiceStrategiesUpdate = false
c63a35a0 1383 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
efabc98d 1384 if (this.workerNodes[workerNodeKey]?.usage != null) {
94407def 1385 const workerUsage = this.workerNodes[workerNodeKey].usage
c329fd41
JB
1386 updateTaskStatisticsWorkerUsage(workerUsage, message)
1387 updateRunTimeWorkerUsage(
bcfb06ce 1388 this.workerChoiceStrategiesContext,
c329fd41
JB
1389 workerUsage,
1390 message
1391 )
1392 updateEluWorkerUsage(
bcfb06ce 1393 this.workerChoiceStrategiesContext,
c329fd41
JB
1394 workerUsage,
1395 message
1396 )
9a55fa8c 1397 needWorkerChoiceStrategiesUpdate = true
94407def
JB
1398 }
1399 if (
1400 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) &&
1401 this.workerNodes[workerNodeKey].getTaskFunctionWorkerUsage(
67f3f2d6
JB
1402 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1403 message.taskPerformance!.name
94407def
JB
1404 ) != null
1405 ) {
67f3f2d6 1406 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
db0e38ee 1407 const taskFunctionWorkerUsage = this.workerNodes[
b558f6b5 1408 workerNodeKey
67f3f2d6
JB
1409 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1410 ].getTaskFunctionWorkerUsage(message.taskPerformance!.name)!
c329fd41
JB
1411 updateTaskStatisticsWorkerUsage(taskFunctionWorkerUsage, message)
1412 updateRunTimeWorkerUsage(
bcfb06ce 1413 this.workerChoiceStrategiesContext,
c329fd41
JB
1414 taskFunctionWorkerUsage,
1415 message
1416 )
1417 updateEluWorkerUsage(
bcfb06ce 1418 this.workerChoiceStrategiesContext,
c329fd41
JB
1419 taskFunctionWorkerUsage,
1420 message
1421 )
9a55fa8c 1422 needWorkerChoiceStrategiesUpdate = true
c329fd41 1423 }
9a55fa8c 1424 if (needWorkerChoiceStrategiesUpdate) {
bcfb06ce 1425 this.workerChoiceStrategiesContext?.update(workerNodeKey)
b558f6b5
JB
1426 }
1427 }
1428
db0e38ee
JB
1429 /**
1430 * Whether the worker node shall update its task function worker usage or not.
1431 *
1432 * @param workerNodeKey - The worker node key.
1433 * @returns `true` if the worker node shall update its task function worker usage, `false` otherwise.
1434 */
1435 private shallUpdateTaskFunctionWorkerUsage (workerNodeKey: number): boolean {
a5d15204 1436 const workerInfo = this.getWorkerInfo(workerNodeKey)
b558f6b5 1437 return (
1851fed0 1438 workerInfo != null &&
31847469
JB
1439 Array.isArray(workerInfo.taskFunctionsProperties) &&
1440 workerInfo.taskFunctionsProperties.length > 2
b558f6b5 1441 )
f1c06930
JB
1442 }
1443
280c2a77 1444 /**
91041638 1445 * Chooses a worker node for the next task.
280c2a77 1446 *
91041638 1447 * @param name - The task function name.
aa9eede8 1448 * @returns The chosen worker node key
280c2a77 1449 */
91041638 1450 private chooseWorkerNode (name?: string): number {
930dcf12 1451 if (this.shallCreateDynamicWorker()) {
aa9eede8 1452 const workerNodeKey = this.createAndSetupDynamicWorkerNode()
6c6afb84 1453 if (
bcfb06ce
JB
1454 this.workerChoiceStrategiesContext?.getPolicy().dynamicWorkerUsage ===
1455 true
6c6afb84 1456 ) {
aa9eede8 1457 return workerNodeKey
6c6afb84 1458 }
17393ac8 1459 }
c63a35a0 1460 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
91041638
JB
1461 return this.workerChoiceStrategiesContext!.execute(
1462 this.getTaskFunctionWorkerChoiceStrategy(name)
1463 )
930dcf12
JB
1464 }
1465
6c6afb84
JB
1466 /**
1467 * Conditions for dynamic worker creation.
1468 *
1469 * @returns Whether to create a dynamic worker or not.
1470 */
9d9fb7b6 1471 protected abstract shallCreateDynamicWorker (): boolean
c97c7edb 1472
280c2a77 1473 /**
aa9eede8 1474 * Sends a message to worker given its worker node key.
280c2a77 1475 *
aa9eede8 1476 * @param workerNodeKey - The worker node key.
38e795c1 1477 * @param message - The message.
7d91a8cd 1478 * @param transferList - The optional array of transferable objects.
280c2a77
S
1479 */
1480 protected abstract sendToWorker (
aa9eede8 1481 workerNodeKey: number,
7d91a8cd 1482 message: MessageValue<Data>,
6a3ecc50 1483 transferList?: readonly TransferListItem[]
280c2a77
S
1484 ): void
1485
e0843544
JB
1486 /**
1487 * Initializes the worker node usage with sensible default values gathered during runtime.
1488 *
1489 * @param workerNode - The worker node.
1490 */
1491 private initWorkerNodeUsage (workerNode: IWorkerNode<Worker, Data>): void {
1492 if (
1493 this.workerChoiceStrategiesContext?.getTaskStatisticsRequirements()
1494 .runTime.aggregate === true
1495 ) {
1496 workerNode.usage.runTime.aggregate = min(
1497 ...this.workerNodes.map(
1498 workerNode => workerNode.usage.runTime.aggregate ?? Infinity
1499 )
1500 )
1501 }
1502 if (
1503 this.workerChoiceStrategiesContext?.getTaskStatisticsRequirements()
1504 .waitTime.aggregate === true
1505 ) {
1506 workerNode.usage.waitTime.aggregate = min(
1507 ...this.workerNodes.map(
1508 workerNode => workerNode.usage.waitTime.aggregate ?? Infinity
1509 )
1510 )
1511 }
1512 if (
1513 this.workerChoiceStrategiesContext?.getTaskStatisticsRequirements().elu
1514 .aggregate === true
1515 ) {
1516 workerNode.usage.elu.active.aggregate = min(
1517 ...this.workerNodes.map(
1518 workerNode => workerNode.usage.elu.active.aggregate ?? Infinity
1519 )
1520 )
1521 }
1522 }
1523
4a6952ff 1524 /**
aa9eede8 1525 * Creates a new, completely set up worker node.
4a6952ff 1526 *
aa9eede8 1527 * @returns New, completely set up worker node key.
4a6952ff 1528 */
aa9eede8 1529 protected createAndSetupWorkerNode (): number {
c3719753
JB
1530 const workerNode = this.createWorkerNode()
1531 workerNode.registerWorkerEventHandler(
1532 'online',
1533 this.opts.onlineHandler ?? EMPTY_FUNCTION
1534 )
1535 workerNode.registerWorkerEventHandler(
1536 'message',
1537 this.opts.messageHandler ?? EMPTY_FUNCTION
1538 )
1539 workerNode.registerWorkerEventHandler(
1540 'error',
1541 this.opts.errorHandler ?? EMPTY_FUNCTION
1542 )
b271fce0 1543 workerNode.registerOnceWorkerEventHandler('error', (error: Error) => {
07e0c9e5 1544 workerNode.info.ready = false
2a69b8c5 1545 this.emitter?.emit(PoolEvents.error, error)
15b176e0 1546 if (
b6bfca01 1547 this.started &&
711623b8 1548 !this.destroying &&
9b38ab2d 1549 this.opts.restartWorkerOnError === true
15b176e0 1550 ) {
07e0c9e5 1551 if (workerNode.info.dynamic) {
aa9eede8 1552 this.createAndSetupDynamicWorkerNode()
b362a929 1553 } else if (!this.startingMinimumNumberOfWorkers) {
e0843544 1554 this.startMinimumNumberOfWorkers(true)
8a1260a3 1555 }
5baee0d7 1556 }
3c9123c7
JB
1557 if (
1558 this.started &&
1559 !this.destroying &&
1560 this.opts.enableTasksQueue === true
1561 ) {
9974369e 1562 this.redistributeQueuedTasks(this.workerNodes.indexOf(workerNode))
19dbc45b 1563 }
b4f8aca8 1564 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
07f6f7b1 1565 workerNode?.terminate().catch((error: unknown) => {
07e0c9e5
JB
1566 this.emitter?.emit(PoolEvents.error, error)
1567 })
5baee0d7 1568 })
c3719753
JB
1569 workerNode.registerWorkerEventHandler(
1570 'exit',
1571 this.opts.exitHandler ?? EMPTY_FUNCTION
1572 )
1573 workerNode.registerOnceWorkerEventHandler('exit', () => {
9974369e 1574 this.removeWorkerNode(workerNode)
b362a929
JB
1575 if (
1576 this.started &&
1577 !this.startingMinimumNumberOfWorkers &&
1578 !this.destroying
1579 ) {
e0843544 1580 this.startMinimumNumberOfWorkers(true)
60dc0a9c 1581 }
a974afa6 1582 })
c3719753 1583 const workerNodeKey = this.addWorkerNode(workerNode)
aa9eede8 1584 this.afterWorkerNodeSetup(workerNodeKey)
aa9eede8 1585 return workerNodeKey
c97c7edb 1586 }
be0676b3 1587
930dcf12 1588 /**
aa9eede8 1589 * Creates a new, completely set up dynamic worker node.
930dcf12 1590 *
aa9eede8 1591 * @returns New, completely set up dynamic worker node key.
930dcf12 1592 */
aa9eede8
JB
1593 protected createAndSetupDynamicWorkerNode (): number {
1594 const workerNodeKey = this.createAndSetupWorkerNode()
041dc05b 1595 this.registerWorkerMessageListener(workerNodeKey, message => {
4d159167 1596 this.checkMessageWorkerId(message)
aa9eede8
JB
1597 const localWorkerNodeKey = this.getWorkerNodeKeyByWorkerId(
1598 message.workerId
aad6fb64 1599 )
152e87a8 1600 const workerInfo = this.getWorkerInfo(localWorkerNodeKey)
2b6b412f 1601 const workerUsage = this.workerNodes[localWorkerNodeKey]?.usage
81c02522 1602 // Kill message received from worker
930dcf12
JB
1603 if (
1604 isKillBehavior(KillBehaviors.HARD, message.kill) ||
1e3214b6 1605 (isKillBehavior(KillBehaviors.SOFT, message.kill) &&
7b56f532 1606 ((this.opts.enableTasksQueue === false &&
aa9eede8 1607 workerUsage.tasks.executing === 0) ||
7b56f532 1608 (this.opts.enableTasksQueue === true &&
152e87a8
JB
1609 workerInfo != null &&
1610 !workerInfo.stealing &&
aa9eede8
JB
1611 workerUsage.tasks.executing === 0 &&
1612 this.tasksQueueSize(localWorkerNodeKey) === 0)))
930dcf12 1613 ) {
f3827d5d 1614 // Flag the worker node as not ready immediately
ae3ab61d 1615 this.flagWorkerNodeAsNotReady(localWorkerNodeKey)
07f6f7b1 1616 this.destroyWorkerNode(localWorkerNodeKey).catch((error: unknown) => {
5270d253
JB
1617 this.emitter?.emit(PoolEvents.error, error)
1618 })
930dcf12
JB
1619 }
1620 })
aa9eede8 1621 this.sendToWorker(workerNodeKey, {
e9dd5b66 1622 checkActive: true
21f710aa 1623 })
72ae84a2 1624 if (this.taskFunctions.size > 0) {
31847469 1625 for (const [taskFunctionName, taskFunctionObject] of this.taskFunctions) {
72ae84a2
JB
1626 this.sendTaskFunctionOperationToWorker(workerNodeKey, {
1627 taskFunctionOperation: 'add',
31847469
JB
1628 taskFunctionProperties: buildTaskFunctionProperties(
1629 taskFunctionName,
1630 taskFunctionObject
1631 ),
1632 taskFunction: taskFunctionObject.taskFunction.toString()
07f6f7b1 1633 }).catch((error: unknown) => {
72ae84a2
JB
1634 this.emitter?.emit(PoolEvents.error, error)
1635 })
1636 }
1637 }
e44639e9
JB
1638 const workerNode = this.workerNodes[workerNodeKey]
1639 workerNode.info.dynamic = true
b1aae695 1640 if (
bcfb06ce
JB
1641 this.workerChoiceStrategiesContext?.getPolicy().dynamicWorkerReady ===
1642 true ||
1643 this.workerChoiceStrategiesContext?.getPolicy().dynamicWorkerUsage ===
1644 true
b1aae695 1645 ) {
e44639e9 1646 workerNode.info.ready = true
b5e113f6 1647 }
e0843544 1648 this.initWorkerNodeUsage(workerNode)
33e6bb4c 1649 this.checkAndEmitDynamicWorkerCreationEvents()
aa9eede8 1650 return workerNodeKey
930dcf12
JB
1651 }
1652
a2ed5053 1653 /**
aa9eede8 1654 * Registers a listener callback on the worker given its worker node key.
a2ed5053 1655 *
aa9eede8 1656 * @param workerNodeKey - The worker node key.
a2ed5053
JB
1657 * @param listener - The message listener callback.
1658 */
85aeb3f3
JB
1659 protected abstract registerWorkerMessageListener<
1660 Message extends Data | Response
aa9eede8
JB
1661 >(
1662 workerNodeKey: number,
1663 listener: (message: MessageValue<Message>) => void
1664 ): void
a2ed5053 1665
ae036c3e
JB
1666 /**
1667 * Registers once a listener callback on the worker given its worker node key.
1668 *
1669 * @param workerNodeKey - The worker node key.
1670 * @param listener - The message listener callback.
1671 */
1672 protected abstract registerOnceWorkerMessageListener<
1673 Message extends Data | Response
1674 >(
1675 workerNodeKey: number,
1676 listener: (message: MessageValue<Message>) => void
1677 ): void
1678
1679 /**
1680 * Deregisters a listener callback on the worker given its worker node key.
1681 *
1682 * @param workerNodeKey - The worker node key.
1683 * @param listener - The message listener callback.
1684 */
1685 protected abstract deregisterWorkerMessageListener<
1686 Message extends Data | Response
1687 >(
1688 workerNodeKey: number,
1689 listener: (message: MessageValue<Message>) => void
1690 ): void
1691
a2ed5053 1692 /**
aa9eede8 1693 * Method hooked up after a worker node has been newly created.
a2ed5053
JB
1694 * Can be overridden.
1695 *
aa9eede8 1696 * @param workerNodeKey - The newly created worker node key.
a2ed5053 1697 */
aa9eede8 1698 protected afterWorkerNodeSetup (workerNodeKey: number): void {
a2ed5053 1699 // Listen to worker messages.
fcd39179
JB
1700 this.registerWorkerMessageListener(
1701 workerNodeKey,
3a776b6d 1702 this.workerMessageListener
fcd39179 1703 )
85aeb3f3 1704 // Send the startup message to worker.
aa9eede8 1705 this.sendStartupMessageToWorker(workerNodeKey)
9edb9717
JB
1706 // Send the statistics message to worker.
1707 this.sendStatisticsMessageToWorker(workerNodeKey)
72695f86 1708 if (this.opts.enableTasksQueue === true) {
dbd73092 1709 if (this.opts.tasksQueueOptions?.taskStealing === true) {
e1c2dba7 1710 this.workerNodes[workerNodeKey].on(
e44639e9
JB
1711 'idle',
1712 this.handleWorkerNodeIdleEvent
9f95d5eb 1713 )
47352846
JB
1714 }
1715 if (this.opts.tasksQueueOptions?.tasksStealingOnBackPressure === true) {
e1c2dba7 1716 this.workerNodes[workerNodeKey].on(
b5e75be8 1717 'backPressure',
e44639e9 1718 this.handleWorkerNodeBackPressureEvent
9f95d5eb 1719 )
47352846 1720 }
72695f86 1721 }
d2c73f82
JB
1722 }
1723
85aeb3f3 1724 /**
aa9eede8
JB
1725 * Sends the startup message to worker given its worker node key.
1726 *
1727 * @param workerNodeKey - The worker node key.
1728 */
1729 protected abstract sendStartupMessageToWorker (workerNodeKey: number): void
1730
1731 /**
9edb9717 1732 * Sends the statistics message to worker given its worker node key.
85aeb3f3 1733 *
aa9eede8 1734 * @param workerNodeKey - The worker node key.
85aeb3f3 1735 */
9edb9717 1736 private sendStatisticsMessageToWorker (workerNodeKey: number): void {
aa9eede8
JB
1737 this.sendToWorker(workerNodeKey, {
1738 statistics: {
1739 runTime:
bcfb06ce 1740 this.workerChoiceStrategiesContext?.getTaskStatisticsRequirements()
f4d0a470 1741 .runTime.aggregate ?? false,
c63a35a0 1742 elu:
bcfb06ce
JB
1743 this.workerChoiceStrategiesContext?.getTaskStatisticsRequirements()
1744 .elu.aggregate ?? false
72ae84a2 1745 }
aa9eede8
JB
1746 })
1747 }
a2ed5053 1748
5eb72b9e
JB
1749 private cannotStealTask (): boolean {
1750 return this.workerNodes.length <= 1 || this.info.queuedTasks === 0
1751 }
1752
42c677c1
JB
1753 private handleTask (workerNodeKey: number, task: Task<Data>): void {
1754 if (this.shallExecuteTask(workerNodeKey)) {
1755 this.executeTask(workerNodeKey, task)
1756 } else {
1757 this.enqueueTask(workerNodeKey, task)
1758 }
1759 }
1760
2d1d3b2d
JB
1761 private redistributeQueuedTasks (sourceWorkerNodeKey: number): void {
1762 if (sourceWorkerNodeKey === -1 || this.cannotStealTask()) {
cb71d660
JB
1763 return
1764 }
2d1d3b2d 1765 while (this.tasksQueueSize(sourceWorkerNodeKey) > 0) {
f201a0cd
JB
1766 const destinationWorkerNodeKey = this.workerNodes.reduce(
1767 (minWorkerNodeKey, workerNode, workerNodeKey, workerNodes) => {
2d1d3b2d
JB
1768 return sourceWorkerNodeKey !== workerNodeKey &&
1769 workerNode.info.ready &&
852ed3e4
JB
1770 workerNode.usage.tasks.queued <
1771 workerNodes[minWorkerNodeKey].usage.tasks.queued
f201a0cd
JB
1772 ? workerNodeKey
1773 : minWorkerNodeKey
1774 },
1775 0
1776 )
42c677c1
JB
1777 this.handleTask(
1778 destinationWorkerNodeKey,
67f3f2d6 1779 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2d1d3b2d 1780 this.dequeueTask(sourceWorkerNodeKey)!
42c677c1 1781 )
dd951876
JB
1782 }
1783 }
1784
b1838604
JB
1785 private updateTaskStolenStatisticsWorkerUsage (
1786 workerNodeKey: number,
b1838604
JB
1787 taskName: string
1788 ): void {
1a880eca 1789 const workerNode = this.workerNodes[workerNodeKey]
c63a35a0 1790 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
efabc98d 1791 if (workerNode?.usage != null) {
b1838604
JB
1792 ++workerNode.usage.tasks.stolen
1793 }
1794 if (
1795 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) &&
1796 workerNode.getTaskFunctionWorkerUsage(taskName) != null
1797 ) {
8bd0556f
JB
1798 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1799 ++workerNode.getTaskFunctionWorkerUsage(taskName)!.tasks.stolen
b1838604
JB
1800 }
1801 }
1802
463226a4 1803 private updateTaskSequentiallyStolenStatisticsWorkerUsage (
2d1d3b2d
JB
1804 workerNodeKey: number,
1805 taskName: string,
8bd0556f 1806 previousTaskName?: string
463226a4
JB
1807 ): void {
1808 const workerNode = this.workerNodes[workerNodeKey]
c63a35a0 1809 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
7aa2f476 1810 if (workerNode?.usage != null) {
463226a4
JB
1811 ++workerNode.usage.tasks.sequentiallyStolen
1812 }
1813 if (
1814 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) &&
3b33cd2a
JB
1815 workerNode.getTaskFunctionWorkerUsage(taskName) != null
1816 ) {
1817 const taskFunctionWorkerUsage =
1818 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1819 workerNode.getTaskFunctionWorkerUsage(taskName)!
1820 if (
1821 taskFunctionWorkerUsage.tasks.sequentiallyStolen === 0 ||
8bd0556f
JB
1822 (previousTaskName != null &&
1823 previousTaskName === taskName &&
3b33cd2a
JB
1824 taskFunctionWorkerUsage.tasks.sequentiallyStolen > 0)
1825 ) {
1826 ++taskFunctionWorkerUsage.tasks.sequentiallyStolen
1827 } else if (taskFunctionWorkerUsage.tasks.sequentiallyStolen > 0) {
1828 taskFunctionWorkerUsage.tasks.sequentiallyStolen = 0
1829 }
463226a4
JB
1830 }
1831 }
1832
1833 private resetTaskSequentiallyStolenStatisticsWorkerUsage (
2d1d3b2d
JB
1834 workerNodeKey: number,
1835 taskName: string
463226a4
JB
1836 ): void {
1837 const workerNode = this.workerNodes[workerNodeKey]
c63a35a0 1838 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
efabc98d 1839 if (workerNode?.usage != null) {
463226a4
JB
1840 workerNode.usage.tasks.sequentiallyStolen = 0
1841 }
1842 if (
1843 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) &&
1844 workerNode.getTaskFunctionWorkerUsage(taskName) != null
1845 ) {
8bd0556f
JB
1846 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1847 workerNode.getTaskFunctionWorkerUsage(
1848 taskName
1849 )!.tasks.sequentiallyStolen = 0
463226a4
JB
1850 }
1851 }
1852
e44639e9 1853 private readonly handleWorkerNodeIdleEvent = (
e1c2dba7 1854 eventDetail: WorkerNodeEventDetail,
463226a4 1855 previousStolenTask?: Task<Data>
9f95d5eb 1856 ): void => {
e1c2dba7 1857 const { workerNodeKey } = eventDetail
463226a4
JB
1858 if (workerNodeKey == null) {
1859 throw new Error(
c63a35a0 1860 "WorkerNode event detail 'workerNodeKey' property must be defined"
463226a4
JB
1861 )
1862 }
c63a35a0 1863 const workerInfo = this.getWorkerInfo(workerNodeKey)
010f158c
JB
1864 if (workerInfo == null) {
1865 throw new Error(
1866 `Worker node with key '${workerNodeKey}' not found in pool`
1867 )
1868 }
5eb72b9e
JB
1869 if (
1870 this.cannotStealTask() ||
c63a35a0
JB
1871 (this.info.stealingWorkerNodes ?? 0) >
1872 Math.floor(this.workerNodes.length / 2)
5eb72b9e 1873 ) {
010f158c 1874 if (previousStolenTask != null) {
c63a35a0 1875 workerInfo.stealing = false
2d1d3b2d
JB
1876 this.resetTaskSequentiallyStolenStatisticsWorkerUsage(
1877 workerNodeKey,
1878 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1879 previousStolenTask.name!
1880 )
5eb72b9e
JB
1881 }
1882 return
1883 }
463226a4
JB
1884 const workerNodeTasksUsage = this.workerNodes[workerNodeKey].usage.tasks
1885 if (
1886 previousStolenTask != null &&
463226a4
JB
1887 (workerNodeTasksUsage.executing > 0 ||
1888 this.tasksQueueSize(workerNodeKey) > 0)
1889 ) {
c63a35a0 1890 workerInfo.stealing = false
2d1d3b2d
JB
1891 this.resetTaskSequentiallyStolenStatisticsWorkerUsage(
1892 workerNodeKey,
1893 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1894 previousStolenTask.name!
1895 )
463226a4
JB
1896 return
1897 }
c63a35a0 1898 workerInfo.stealing = true
463226a4 1899 const stolenTask = this.workerNodeStealTask(workerNodeKey)
8bd0556f 1900 if (stolenTask != null) {
2d1d3b2d
JB
1901 this.updateTaskSequentiallyStolenStatisticsWorkerUsage(
1902 workerNodeKey,
67f3f2d6 1903 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2d1d3b2d 1904 stolenTask.name!,
8bd0556f 1905 previousStolenTask?.name
2d1d3b2d 1906 )
f1f77f45 1907 }
463226a4
JB
1908 sleep(exponentialDelay(workerNodeTasksUsage.sequentiallyStolen))
1909 .then(() => {
e44639e9 1910 this.handleWorkerNodeIdleEvent(eventDetail, stolenTask)
463226a4
JB
1911 return undefined
1912 })
07f6f7b1 1913 .catch((error: unknown) => {
2e8cfbb7
JB
1914 this.emitter?.emit(PoolEvents.error, error)
1915 })
463226a4
JB
1916 }
1917
1918 private readonly workerNodeStealTask = (
1919 workerNodeKey: number
1920 ): Task<Data> | undefined => {
dd951876 1921 const workerNodes = this.workerNodes
a6b3272b 1922 .slice()
dd951876
JB
1923 .sort(
1924 (workerNodeA, workerNodeB) =>
1925 workerNodeB.usage.tasks.queued - workerNodeA.usage.tasks.queued
1926 )
f201a0cd 1927 const sourceWorkerNode = workerNodes.find(
463226a4
JB
1928 (sourceWorkerNode, sourceWorkerNodeKey) =>
1929 sourceWorkerNode.info.ready &&
5eb72b9e 1930 !sourceWorkerNode.info.stealing &&
463226a4
JB
1931 sourceWorkerNodeKey !== workerNodeKey &&
1932 sourceWorkerNode.usage.tasks.queued > 0
f201a0cd
JB
1933 )
1934 if (sourceWorkerNode != null) {
67f3f2d6 1935 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2eee7220 1936 const task = sourceWorkerNode.dequeueLastPrioritizedTask()!
42c677c1 1937 this.handleTask(workerNodeKey, task)
67f3f2d6
JB
1938 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1939 this.updateTaskStolenStatisticsWorkerUsage(workerNodeKey, task.name!)
463226a4 1940 return task
72695f86
JB
1941 }
1942 }
1943
e44639e9 1944 private readonly handleWorkerNodeBackPressureEvent = (
e1c2dba7 1945 eventDetail: WorkerNodeEventDetail
9f95d5eb 1946 ): void => {
5eb72b9e
JB
1947 if (
1948 this.cannotStealTask() ||
2eee7220 1949 this.hasBackPressure() ||
c63a35a0
JB
1950 (this.info.stealingWorkerNodes ?? 0) >
1951 Math.floor(this.workerNodes.length / 2)
5eb72b9e 1952 ) {
cb71d660
JB
1953 return
1954 }
f778c355 1955 const sizeOffset = 1
67f3f2d6
JB
1956 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1957 if (this.opts.tasksQueueOptions!.size! <= sizeOffset) {
68dbcdc0
JB
1958 return
1959 }
7b2c5627 1960 const { workerId } = eventDetail
72695f86 1961 const sourceWorkerNode =
b5e75be8 1962 this.workerNodes[this.getWorkerNodeKeyByWorkerId(workerId)]
72695f86 1963 const workerNodes = this.workerNodes
a6b3272b 1964 .slice()
72695f86
JB
1965 .sort(
1966 (workerNodeA, workerNodeB) =>
1967 workerNodeA.usage.tasks.queued - workerNodeB.usage.tasks.queued
1968 )
1969 for (const [workerNodeKey, workerNode] of workerNodes.entries()) {
1970 if (
0bc68267 1971 sourceWorkerNode.usage.tasks.queued > 0 &&
a6b3272b 1972 workerNode.info.ready &&
5eb72b9e 1973 !workerNode.info.stealing &&
b5e75be8 1974 workerNode.info.id !== workerId &&
0bc68267 1975 workerNode.usage.tasks.queued <
67f3f2d6
JB
1976 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1977 this.opts.tasksQueueOptions!.size! - sizeOffset
72695f86 1978 ) {
c63a35a0 1979 const workerInfo = this.getWorkerInfo(workerNodeKey)
1851fed0
JB
1980 if (workerInfo == null) {
1981 throw new Error(
1982 `Worker node with key '${workerNodeKey}' not found in pool`
1983 )
1984 }
c63a35a0 1985 workerInfo.stealing = true
67f3f2d6 1986 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2eee7220 1987 const task = sourceWorkerNode.dequeueLastPrioritizedTask()!
42c677c1 1988 this.handleTask(workerNodeKey, task)
67f3f2d6
JB
1989 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
1990 this.updateTaskStolenStatisticsWorkerUsage(workerNodeKey, task.name!)
c63a35a0 1991 workerInfo.stealing = false
10ecf8fd 1992 }
a2ed5053
JB
1993 }
1994 }
1995
be0676b3 1996 /**
fcd39179 1997 * This method is the message listener registered on each worker.
be0676b3 1998 */
3a776b6d
JB
1999 protected readonly workerMessageListener = (
2000 message: MessageValue<Response>
2001 ): void => {
fcd39179 2002 this.checkMessageWorkerId(message)
31847469
JB
2003 const { workerId, ready, taskId, taskFunctionsProperties } = message
2004 if (ready != null && taskFunctionsProperties != null) {
fcd39179
JB
2005 // Worker ready response received from worker
2006 this.handleWorkerReadyResponse(message)
31847469
JB
2007 } else if (taskFunctionsProperties != null) {
2008 // Task function properties message received from worker
9a55fa8c
JB
2009 const workerNodeKey = this.getWorkerNodeKeyByWorkerId(workerId)
2010 const workerInfo = this.getWorkerInfo(workerNodeKey)
1851fed0 2011 if (workerInfo != null) {
31847469 2012 workerInfo.taskFunctionsProperties = taskFunctionsProperties
9a55fa8c 2013 this.sendStatisticsMessageToWorker(workerNodeKey)
1851fed0 2014 }
31847469
JB
2015 } else if (taskId != null) {
2016 // Task execution response received from worker
2017 this.handleTaskExecutionResponse(message)
6b272951
JB
2018 }
2019 }
2020
8e8d9101
JB
2021 private checkAndEmitReadyEvent (): void {
2022 if (!this.readyEventEmitted && this.ready) {
2023 this.emitter?.emit(PoolEvents.ready, this.info)
2024 this.readyEventEmitted = true
2025 }
2026 }
2027
10e2aa7e 2028 private handleWorkerReadyResponse (message: MessageValue<Response>): void {
31847469 2029 const { workerId, ready, taskFunctionsProperties } = message
e44639e9 2030 if (ready == null || !ready) {
c63a35a0 2031 throw new Error(`Worker ${workerId} failed to initialize`)
f05ed93c 2032 }
9a55fa8c
JB
2033 const workerNodeKey = this.getWorkerNodeKeyByWorkerId(workerId)
2034 const workerNode = this.workerNodes[workerNodeKey]
e44639e9 2035 workerNode.info.ready = ready
31847469 2036 workerNode.info.taskFunctionsProperties = taskFunctionsProperties
9a55fa8c 2037 this.sendStatisticsMessageToWorker(workerNodeKey)
8e8d9101 2038 this.checkAndEmitReadyEvent()
6b272951
JB
2039 }
2040
2041 private handleTaskExecutionResponse (message: MessageValue<Response>): void {
463226a4 2042 const { workerId, taskId, workerError, data } = message
67f3f2d6
JB
2043 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2044 const promiseResponse = this.promiseResponseMap.get(taskId!)
6b272951 2045 if (promiseResponse != null) {
f18fd12b 2046 const { resolve, reject, workerNodeKey, asyncResource } = promiseResponse
9b358e72 2047 const workerNode = this.workerNodes[workerNodeKey]
6703b9f4
JB
2048 if (workerError != null) {
2049 this.emitter?.emit(PoolEvents.taskError, workerError)
f18fd12b
JB
2050 asyncResource != null
2051 ? asyncResource.runInAsyncScope(
2052 reject,
2053 this.emitter,
2054 workerError.message
2055 )
2056 : reject(workerError.message)
6b272951 2057 } else {
f18fd12b
JB
2058 asyncResource != null
2059 ? asyncResource.runInAsyncScope(resolve, this.emitter, data)
2060 : resolve(data as Response)
6b272951 2061 }
f18fd12b 2062 asyncResource?.emitDestroy()
501aea93 2063 this.afterTaskExecutionHook(workerNodeKey, message)
67f3f2d6
JB
2064 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2065 this.promiseResponseMap.delete(taskId!)
cdaecaee
JB
2066 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
2067 workerNode?.emit('taskFinished', taskId)
16d7e943
JB
2068 if (
2069 this.opts.enableTasksQueue === true &&
2070 !this.destroying &&
2071 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
2072 workerNode != null
2073 ) {
9b358e72 2074 const workerNodeTasksUsage = workerNode.usage.tasks
463226a4
JB
2075 if (
2076 this.tasksQueueSize(workerNodeKey) > 0 &&
2077 workerNodeTasksUsage.executing <
67f3f2d6
JB
2078 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2079 this.opts.tasksQueueOptions!.concurrency!
463226a4 2080 ) {
67f3f2d6
JB
2081 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2082 this.executeTask(workerNodeKey, this.dequeueTask(workerNodeKey)!)
463226a4
JB
2083 }
2084 if (
2085 workerNodeTasksUsage.executing === 0 &&
2086 this.tasksQueueSize(workerNodeKey) === 0 &&
2087 workerNodeTasksUsage.sequentiallyStolen === 0
2088 ) {
e44639e9 2089 workerNode.emit('idle', {
7f0e1334 2090 workerId,
e1c2dba7
JB
2091 workerNodeKey
2092 })
463226a4 2093 }
be0676b3
APA
2094 }
2095 }
be0676b3 2096 }
7c0ba920 2097
a1763c54 2098 private checkAndEmitTaskExecutionEvents (): void {
33e6bb4c
JB
2099 if (this.busy) {
2100 this.emitter?.emit(PoolEvents.busy, this.info)
a1763c54
JB
2101 }
2102 }
2103
2104 private checkAndEmitTaskQueuingEvents (): void {
2105 if (this.hasBackPressure()) {
2106 this.emitter?.emit(PoolEvents.backPressure, this.info)
164d950a
JB
2107 }
2108 }
2109
d0878034
JB
2110 /**
2111 * Emits dynamic worker creation events.
2112 */
2113 protected abstract checkAndEmitDynamicWorkerCreationEvents (): void
33e6bb4c 2114
8a1260a3 2115 /**
aa9eede8 2116 * Gets the worker information given its worker node key.
8a1260a3
JB
2117 *
2118 * @param workerNodeKey - The worker node key.
3f09ed9f 2119 * @returns The worker information.
8a1260a3 2120 */
1851fed0
JB
2121 protected getWorkerInfo (workerNodeKey: number): WorkerInfo | undefined {
2122 return this.workerNodes[workerNodeKey]?.info
e221309a
JB
2123 }
2124
a05c10de 2125 /**
c3719753 2126 * Creates a worker node.
ea7a90d3 2127 *
c3719753 2128 * @returns The created worker node.
ea7a90d3 2129 */
c3719753 2130 private createWorkerNode (): IWorkerNode<Worker, Data> {
671d5154 2131 const workerNode = new WorkerNode<Worker, Data>(
c3719753
JB
2132 this.worker,
2133 this.filePath,
2134 {
2135 env: this.opts.env,
2136 workerOptions: this.opts.workerOptions,
2137 tasksQueueBackPressureSize:
32b141fd 2138 this.opts.tasksQueueOptions?.size ??
26ce26ca
JB
2139 getDefaultTasksQueueOptions(
2140 this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers
95d1a734
JB
2141 ).size,
2142 tasksQueueBucketSize:
2143 (this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers) * 2
c3719753 2144 }
671d5154 2145 )
b97d82d8 2146 // Flag the worker node as ready at pool startup.
d2c73f82
JB
2147 if (this.starting) {
2148 workerNode.info.ready = true
2149 }
c3719753
JB
2150 return workerNode
2151 }
2152
2153 /**
2154 * Adds the given worker node in the pool worker nodes.
2155 *
2156 * @param workerNode - The worker node.
2157 * @returns The added worker node key.
2158 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the added worker node is not found.
2159 */
2160 private addWorkerNode (workerNode: IWorkerNode<Worker, Data>): number {
aa9eede8 2161 this.workerNodes.push(workerNode)
c3719753 2162 const workerNodeKey = this.workerNodes.indexOf(workerNode)
aa9eede8 2163 if (workerNodeKey === -1) {
86ed0598 2164 throw new Error('Worker added not found in worker nodes')
aa9eede8
JB
2165 }
2166 return workerNodeKey
ea7a90d3 2167 }
c923ce56 2168
8e8d9101
JB
2169 private checkAndEmitEmptyEvent (): void {
2170 if (this.empty) {
2171 this.emitter?.emit(PoolEvents.empty, this.info)
2172 this.readyEventEmitted = false
2173 }
2174 }
2175
51fe3d3c 2176 /**
9974369e 2177 * Removes the worker node from the pool worker nodes.
51fe3d3c 2178 *
9974369e 2179 * @param workerNode - The worker node.
51fe3d3c 2180 */
9974369e
JB
2181 private removeWorkerNode (workerNode: IWorkerNode<Worker, Data>): void {
2182 const workerNodeKey = this.workerNodes.indexOf(workerNode)
1f68cede
JB
2183 if (workerNodeKey !== -1) {
2184 this.workerNodes.splice(workerNodeKey, 1)
bcfb06ce 2185 this.workerChoiceStrategiesContext?.remove(workerNodeKey)
1f68cede 2186 }
8e8d9101 2187 this.checkAndEmitEmptyEvent()
51fe3d3c 2188 }
adc3c320 2189
ae3ab61d 2190 protected flagWorkerNodeAsNotReady (workerNodeKey: number): void {
1851fed0
JB
2191 const workerInfo = this.getWorkerInfo(workerNodeKey)
2192 if (workerInfo != null) {
2193 workerInfo.ready = false
2194 }
ae3ab61d
JB
2195 }
2196
9e844245
JB
2197 private hasBackPressure (): boolean {
2198 return (
2199 this.opts.enableTasksQueue === true &&
2200 this.workerNodes.findIndex(
041dc05b 2201 workerNode => !workerNode.hasBackPressure()
a1763c54 2202 ) === -1
9e844245 2203 )
e2b31e32
JB
2204 }
2205
b0a4db63 2206 /**
aa9eede8 2207 * Executes the given task on the worker given its worker node key.
b0a4db63 2208 *
aa9eede8 2209 * @param workerNodeKey - The worker node key.
b0a4db63
JB
2210 * @param task - The task to execute.
2211 */
2e81254d 2212 private executeTask (workerNodeKey: number, task: Task<Data>): void {
1c6fe997 2213 this.beforeTaskExecutionHook(workerNodeKey, task)
bbfa38a2 2214 this.sendToWorker(workerNodeKey, task, task.transferList)
a1763c54 2215 this.checkAndEmitTaskExecutionEvents()
2e81254d
JB
2216 }
2217
f9f00b5f 2218 private enqueueTask (workerNodeKey: number, task: Task<Data>): number {
a1763c54
JB
2219 const tasksQueueSize = this.workerNodes[workerNodeKey].enqueueTask(task)
2220 this.checkAndEmitTaskQueuingEvents()
2221 return tasksQueueSize
adc3c320
JB
2222 }
2223
0d4e88b3
JB
2224 private dequeueTask (workerNodeKey: number): Task<Data> | undefined {
2225 return this.workerNodes[workerNodeKey].dequeueTask()
adc3c320
JB
2226 }
2227
416fd65c 2228 private tasksQueueSize (workerNodeKey: number): number {
4b628b48 2229 return this.workerNodes[workerNodeKey].tasksQueueSize()
df593701
JB
2230 }
2231
87347ea8
JB
2232 protected flushTasksQueue (workerNodeKey: number): number {
2233 let flushedTasks = 0
920278a2 2234 while (this.tasksQueueSize(workerNodeKey) > 0) {
67f3f2d6
JB
2235 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
2236 this.executeTask(workerNodeKey, this.dequeueTask(workerNodeKey)!)
87347ea8 2237 ++flushedTasks
ff733df7 2238 }
4b628b48 2239 this.workerNodes[workerNodeKey].clearTasksQueue()
87347ea8 2240 return flushedTasks
ff733df7
JB
2241 }
2242
ef41a6e6 2243 private flushTasksQueues (): void {
19b8be8b 2244 for (const workerNodeKey of this.workerNodes.keys()) {
ef41a6e6
JB
2245 this.flushTasksQueue(workerNodeKey)
2246 }
2247 }
c97c7edb 2248}