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