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