refactor: sensible defaults for worker choice strategy policy
[poolifier.git] / src / pools / abstract-pool.ts
CommitLineData
2845f2a5 1import { randomUUID } from 'node:crypto'
62c15a68 2import { performance } from 'node:perf_hooks'
3d6dd312 3import { existsSync } from 'node:fs'
7d91a8cd 4import { type TransferListItem } from 'node:worker_threads'
5c4d16da
JB
5import type {
6 MessageValue,
7 PromiseResponseWrapper,
8 Task
9} from '../utility-types'
bbeadd16 10import {
ff128cc9 11 DEFAULT_TASK_NAME,
bbeadd16
JB
12 DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS,
13 EMPTY_FUNCTION,
59317253 14 isKillBehavior,
0d80593b 15 isPlainObject,
afe0d5bf 16 median,
e4f20deb
JB
17 round,
18 updateMeasurementStatistics
bbeadd16 19} from '../utils'
59317253 20import { KillBehaviors } from '../worker/worker-options'
c4855468 21import {
65d7a1c9 22 type IPool,
7c5a1080 23 PoolEmitter,
c4855468 24 PoolEvents,
6b27d407 25 type PoolInfo,
c4855468 26 type PoolOptions,
6b27d407
JB
27 type PoolType,
28 PoolTypes,
4b628b48 29 type TasksQueueOptions
c4855468 30} from './pool'
bbfa38a2
JB
31import type {
32 IWorker,
33 IWorkerNode,
34 WorkerInfo,
35 WorkerType,
36 WorkerUsage
e102732c 37} from './worker'
a35560ba 38import {
008512c7 39 type MeasurementStatisticsRequirements,
f0d7f803 40 Measurements,
a35560ba 41 WorkerChoiceStrategies,
a20f0ba5
JB
42 type WorkerChoiceStrategy,
43 type WorkerChoiceStrategyOptions
bdaf31cd
JB
44} from './selection-strategies/selection-strategies-types'
45import { WorkerChoiceStrategyContext } from './selection-strategies/worker-choice-strategy-context'
92b1feaa 46import { version } from './version'
4b628b48 47import { WorkerNode } from './worker-node'
23ccf9d7 48
729c563d 49/**
ea7a90d3 50 * Base class that implements some shared logic for all poolifier pools.
729c563d 51 *
38e795c1 52 * @typeParam Worker - Type of worker which manages this pool.
e102732c
JB
53 * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
54 * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
729c563d 55 */
c97c7edb 56export abstract class AbstractPool<
f06e48d8 57 Worker extends IWorker,
d3c8a1a8
S
58 Data = unknown,
59 Response = unknown
c4855468 60> implements IPool<Worker, Data, Response> {
afc003b2 61 /** @inheritDoc */
4b628b48 62 public readonly workerNodes: Array<IWorkerNode<Worker, Data>> = []
4a6952ff 63
afc003b2 64 /** @inheritDoc */
7c0ba920
JB
65 public readonly emitter?: PoolEmitter
66
be0676b3 67 /**
52b71763 68 * The task execution response promise map.
be0676b3 69 *
2740a743 70 * - `key`: The message id of each submitted task.
a3445496 71 * - `value`: An object that contains the worker, the execution response promise resolve and reject callbacks.
be0676b3 72 *
a3445496 73 * When we receive a message from the worker, we get a map entry with the promise resolve/reject bound to the message id.
be0676b3 74 */
501aea93
JB
75 protected promiseResponseMap: Map<string, PromiseResponseWrapper<Response>> =
76 new Map<string, PromiseResponseWrapper<Response>>()
c97c7edb 77
a35560ba 78 /**
51fe3d3c 79 * Worker choice strategy context referencing a worker choice algorithm implementation.
a35560ba
S
80 */
81 protected workerChoiceStrategyContext: WorkerChoiceStrategyContext<
78cea37e
JB
82 Worker,
83 Data,
84 Response
a35560ba
S
85 >
86
8735b4e5
JB
87 /**
88 * Dynamic pool maximum size property placeholder.
89 */
90 protected readonly max?: number
91
075e51d1 92 /**
adc9cc64 93 * Whether the pool is starting or not.
075e51d1
JB
94 */
95 private readonly starting: boolean
afe0d5bf
JB
96 /**
97 * The start timestamp of the pool.
98 */
99 private readonly startTimestamp
100
729c563d
S
101 /**
102 * Constructs a new poolifier pool.
103 *
38e795c1 104 * @param numberOfWorkers - Number of workers that this pool should manage.
029715f0 105 * @param filePath - Path to the worker file.
38e795c1 106 * @param opts - Options for the pool.
729c563d 107 */
c97c7edb 108 public constructor (
b4213b7f
JB
109 protected readonly numberOfWorkers: number,
110 protected readonly filePath: string,
111 protected readonly opts: PoolOptions<Worker>
c97c7edb 112 ) {
78cea37e 113 if (!this.isMain()) {
04f45163 114 throw new Error(
8c6d4acf 115 'Cannot start a pool from a worker with the same type as the pool'
04f45163 116 )
c97c7edb 117 }
8d3782fa 118 this.checkNumberOfWorkers(this.numberOfWorkers)
c510fea7 119 this.checkFilePath(this.filePath)
7c0ba920 120 this.checkPoolOptions(this.opts)
1086026a 121
7254e419
JB
122 this.chooseWorkerNode = this.chooseWorkerNode.bind(this)
123 this.executeTask = this.executeTask.bind(this)
124 this.enqueueTask = this.enqueueTask.bind(this)
1086026a 125
6bd72cd0 126 if (this.opts.enableEvents === true) {
7c0ba920
JB
127 this.emitter = new PoolEmitter()
128 }
d59df138
JB
129 this.workerChoiceStrategyContext = new WorkerChoiceStrategyContext<
130 Worker,
131 Data,
132 Response
da309861
JB
133 >(
134 this,
135 this.opts.workerChoiceStrategy,
136 this.opts.workerChoiceStrategyOptions
137 )
b6b32453
JB
138
139 this.setupHook()
140
075e51d1 141 this.starting = true
e761c033 142 this.startPool()
075e51d1 143 this.starting = false
afe0d5bf
JB
144
145 this.startTimestamp = performance.now()
c97c7edb
S
146 }
147
a35560ba 148 private checkFilePath (filePath: string): void {
ffcbbad8
JB
149 if (
150 filePath == null ||
3d6dd312 151 typeof filePath !== 'string' ||
ffcbbad8
JB
152 (typeof filePath === 'string' && filePath.trim().length === 0)
153 ) {
c510fea7
APA
154 throw new Error('Please specify a file with a worker implementation')
155 }
3d6dd312
JB
156 if (!existsSync(filePath)) {
157 throw new Error(`Cannot find the worker file '${filePath}'`)
158 }
c510fea7
APA
159 }
160
8d3782fa
JB
161 private checkNumberOfWorkers (numberOfWorkers: number): void {
162 if (numberOfWorkers == null) {
163 throw new Error(
164 'Cannot instantiate a pool without specifying the number of workers'
165 )
78cea37e 166 } else if (!Number.isSafeInteger(numberOfWorkers)) {
473c717a 167 throw new TypeError(
0d80593b 168 'Cannot instantiate a pool with a non safe integer number of workers'
8d3782fa
JB
169 )
170 } else if (numberOfWorkers < 0) {
473c717a 171 throw new RangeError(
8d3782fa
JB
172 'Cannot instantiate a pool with a negative number of workers'
173 )
6b27d407 174 } else if (this.type === PoolTypes.fixed && numberOfWorkers === 0) {
2431bdb4
JB
175 throw new RangeError('Cannot instantiate a fixed pool with zero worker')
176 }
177 }
178
179 protected checkDynamicPoolSize (min: number, max: number): void {
079de991 180 if (this.type === PoolTypes.dynamic) {
a5ed75b7
JB
181 if (max == null) {
182 throw new Error(
183 'Cannot instantiate a dynamic pool without specifying the maximum pool size'
184 )
185 } else if (!Number.isSafeInteger(max)) {
2761efb4
JB
186 throw new TypeError(
187 'Cannot instantiate a dynamic pool with a non safe integer maximum pool size'
188 )
189 } else if (min > max) {
079de991
JB
190 throw new RangeError(
191 'Cannot instantiate a dynamic pool with a maximum pool size inferior to the minimum pool size'
192 )
b97d82d8 193 } else if (max === 0) {
079de991 194 throw new RangeError(
d640b48b 195 'Cannot instantiate a dynamic pool with a maximum pool size equal to zero'
079de991
JB
196 )
197 } else if (min === max) {
198 throw new RangeError(
199 'Cannot instantiate a dynamic pool with a minimum pool size equal to the maximum pool size. Use a fixed pool instead'
200 )
201 }
8d3782fa
JB
202 }
203 }
204
7c0ba920 205 private checkPoolOptions (opts: PoolOptions<Worker>): void {
0d80593b
JB
206 if (isPlainObject(opts)) {
207 this.opts.workerChoiceStrategy =
208 opts.workerChoiceStrategy ?? WorkerChoiceStrategies.ROUND_ROBIN
209 this.checkValidWorkerChoiceStrategy(this.opts.workerChoiceStrategy)
8990357d
JB
210 this.opts.workerChoiceStrategyOptions = {
211 ...DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS,
212 ...opts.workerChoiceStrategyOptions
213 }
49be33fe
JB
214 this.checkValidWorkerChoiceStrategyOptions(
215 this.opts.workerChoiceStrategyOptions
216 )
1f68cede 217 this.opts.restartWorkerOnError = opts.restartWorkerOnError ?? true
0d80593b
JB
218 this.opts.enableEvents = opts.enableEvents ?? true
219 this.opts.enableTasksQueue = opts.enableTasksQueue ?? false
220 if (this.opts.enableTasksQueue) {
221 this.checkValidTasksQueueOptions(
222 opts.tasksQueueOptions as TasksQueueOptions
223 )
224 this.opts.tasksQueueOptions = this.buildTasksQueueOptions(
225 opts.tasksQueueOptions as TasksQueueOptions
226 )
227 }
228 } else {
229 throw new TypeError('Invalid pool options: must be a plain object')
7171d33f 230 }
aee46736
JB
231 }
232
233 private checkValidWorkerChoiceStrategy (
234 workerChoiceStrategy: WorkerChoiceStrategy
235 ): void {
236 if (!Object.values(WorkerChoiceStrategies).includes(workerChoiceStrategy)) {
b529c323 237 throw new Error(
aee46736 238 `Invalid worker choice strategy '${workerChoiceStrategy}'`
b529c323
JB
239 )
240 }
7c0ba920
JB
241 }
242
0d80593b
JB
243 private checkValidWorkerChoiceStrategyOptions (
244 workerChoiceStrategyOptions: WorkerChoiceStrategyOptions
245 ): void {
246 if (!isPlainObject(workerChoiceStrategyOptions)) {
247 throw new TypeError(
248 'Invalid worker choice strategy options: must be a plain object'
249 )
250 }
8990357d
JB
251 if (
252 workerChoiceStrategyOptions.choiceRetries != null &&
253 !Number.isSafeInteger(workerChoiceStrategyOptions.choiceRetries)
254 ) {
255 throw new TypeError(
256 'Invalid worker choice strategy options: choice retries must be an integer'
257 )
258 }
259 if (
260 workerChoiceStrategyOptions.choiceRetries != null &&
261 workerChoiceStrategyOptions.choiceRetries <= 0
262 ) {
263 throw new RangeError(
264 `Invalid worker choice strategy options: choice retries '${workerChoiceStrategyOptions.choiceRetries}' must be greater than zero`
265 )
266 }
49be33fe
JB
267 if (
268 workerChoiceStrategyOptions.weights != null &&
6b27d407 269 Object.keys(workerChoiceStrategyOptions.weights).length !== this.maxSize
49be33fe
JB
270 ) {
271 throw new Error(
272 'Invalid worker choice strategy options: must have a weight for each worker node'
273 )
274 }
f0d7f803
JB
275 if (
276 workerChoiceStrategyOptions.measurement != null &&
277 !Object.values(Measurements).includes(
278 workerChoiceStrategyOptions.measurement
279 )
280 ) {
281 throw new Error(
282 `Invalid worker choice strategy options: invalid measurement '${workerChoiceStrategyOptions.measurement}'`
283 )
284 }
0d80593b
JB
285 }
286
a20f0ba5
JB
287 private checkValidTasksQueueOptions (
288 tasksQueueOptions: TasksQueueOptions
289 ): void {
0d80593b
JB
290 if (tasksQueueOptions != null && !isPlainObject(tasksQueueOptions)) {
291 throw new TypeError('Invalid tasks queue options: must be a plain object')
292 }
f0d7f803
JB
293 if (
294 tasksQueueOptions?.concurrency != null &&
295 !Number.isSafeInteger(tasksQueueOptions.concurrency)
296 ) {
297 throw new TypeError(
298 'Invalid worker tasks concurrency: must be an integer'
299 )
300 }
301 if (
302 tasksQueueOptions?.concurrency != null &&
303 tasksQueueOptions.concurrency <= 0
304 ) {
a20f0ba5 305 throw new Error(
8735b4e5 306 `Invalid worker tasks concurrency: ${tasksQueueOptions.concurrency} is a negative integer or zero`
a20f0ba5
JB
307 )
308 }
309 }
310
e761c033
JB
311 private startPool (): void {
312 while (
313 this.workerNodes.reduce(
314 (accumulator, workerNode) =>
315 !workerNode.info.dynamic ? accumulator + 1 : accumulator,
316 0
317 ) < this.numberOfWorkers
318 ) {
aa9eede8 319 this.createAndSetupWorkerNode()
e761c033
JB
320 }
321 }
322
08f3f44c 323 /** @inheritDoc */
6b27d407
JB
324 public get info (): PoolInfo {
325 return {
23ccf9d7 326 version,
6b27d407 327 type: this.type,
184855e6 328 worker: this.worker,
2431bdb4
JB
329 ready: this.ready,
330 strategy: this.opts.workerChoiceStrategy as WorkerChoiceStrategy,
6b27d407
JB
331 minSize: this.minSize,
332 maxSize: this.maxSize,
c05f0d50
JB
333 ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
334 .runTime.aggregate &&
1305e9a8
JB
335 this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
336 .waitTime.aggregate && { utilization: round(this.utilization) }),
6b27d407
JB
337 workerNodes: this.workerNodes.length,
338 idleWorkerNodes: this.workerNodes.reduce(
339 (accumulator, workerNode) =>
f59e1027 340 workerNode.usage.tasks.executing === 0
a4e07f72
JB
341 ? accumulator + 1
342 : accumulator,
6b27d407
JB
343 0
344 ),
345 busyWorkerNodes: this.workerNodes.reduce(
346 (accumulator, workerNode) =>
f59e1027 347 workerNode.usage.tasks.executing > 0 ? accumulator + 1 : accumulator,
6b27d407
JB
348 0
349 ),
a4e07f72 350 executedTasks: this.workerNodes.reduce(
6b27d407 351 (accumulator, workerNode) =>
f59e1027 352 accumulator + workerNode.usage.tasks.executed,
a4e07f72
JB
353 0
354 ),
355 executingTasks: this.workerNodes.reduce(
356 (accumulator, workerNode) =>
f59e1027 357 accumulator + workerNode.usage.tasks.executing,
6b27d407
JB
358 0
359 ),
daf86646
JB
360 ...(this.opts.enableTasksQueue === true && {
361 queuedTasks: this.workerNodes.reduce(
362 (accumulator, workerNode) =>
363 accumulator + workerNode.usage.tasks.queued,
364 0
365 )
366 }),
367 ...(this.opts.enableTasksQueue === true && {
368 maxQueuedTasks: this.workerNodes.reduce(
369 (accumulator, workerNode) =>
370 accumulator + (workerNode.usage.tasks?.maxQueued ?? 0),
371 0
372 )
373 }),
a1763c54
JB
374 ...(this.opts.enableTasksQueue === true && {
375 backPressure: this.hasBackPressure()
376 }),
a4e07f72
JB
377 failedTasks: this.workerNodes.reduce(
378 (accumulator, workerNode) =>
f59e1027 379 accumulator + workerNode.usage.tasks.failed,
a4e07f72 380 0
1dcf8b7b
JB
381 ),
382 ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
383 .runTime.aggregate && {
384 runTime: {
98e72cda
JB
385 minimum: round(
386 Math.min(
387 ...this.workerNodes.map(
8ebe6c30 388 (workerNode) => workerNode.usage.runTime?.minimum ?? Infinity
98e72cda 389 )
1dcf8b7b
JB
390 )
391 ),
98e72cda
JB
392 maximum: round(
393 Math.max(
394 ...this.workerNodes.map(
8ebe6c30 395 (workerNode) => workerNode.usage.runTime?.maximum ?? -Infinity
98e72cda 396 )
1dcf8b7b 397 )
98e72cda
JB
398 ),
399 average: round(
400 this.workerNodes.reduce(
401 (accumulator, workerNode) =>
402 accumulator + (workerNode.usage.runTime?.aggregate ?? 0),
403 0
404 ) /
405 this.workerNodes.reduce(
406 (accumulator, workerNode) =>
407 accumulator + (workerNode.usage.tasks?.executed ?? 0),
408 0
409 )
410 ),
411 ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
412 .runTime.median && {
413 median: round(
414 median(
415 this.workerNodes.map(
8ebe6c30 416 (workerNode) => workerNode.usage.runTime?.median ?? 0
98e72cda
JB
417 )
418 )
419 )
420 })
1dcf8b7b
JB
421 }
422 }),
423 ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
424 .waitTime.aggregate && {
425 waitTime: {
98e72cda
JB
426 minimum: round(
427 Math.min(
428 ...this.workerNodes.map(
8ebe6c30 429 (workerNode) => workerNode.usage.waitTime?.minimum ?? Infinity
98e72cda 430 )
1dcf8b7b
JB
431 )
432 ),
98e72cda
JB
433 maximum: round(
434 Math.max(
435 ...this.workerNodes.map(
8ebe6c30 436 (workerNode) => workerNode.usage.waitTime?.maximum ?? -Infinity
98e72cda 437 )
1dcf8b7b 438 )
98e72cda
JB
439 ),
440 average: round(
441 this.workerNodes.reduce(
442 (accumulator, workerNode) =>
443 accumulator + (workerNode.usage.waitTime?.aggregate ?? 0),
444 0
445 ) /
446 this.workerNodes.reduce(
447 (accumulator, workerNode) =>
448 accumulator + (workerNode.usage.tasks?.executed ?? 0),
449 0
450 )
451 ),
452 ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
453 .waitTime.median && {
454 median: round(
455 median(
456 this.workerNodes.map(
8ebe6c30 457 (workerNode) => workerNode.usage.waitTime?.median ?? 0
98e72cda
JB
458 )
459 )
460 )
461 })
1dcf8b7b
JB
462 }
463 })
6b27d407
JB
464 }
465 }
08f3f44c 466
aa9eede8
JB
467 /**
468 * The pool readiness boolean status.
469 */
2431bdb4
JB
470 private get ready (): boolean {
471 return (
b97d82d8
JB
472 this.workerNodes.reduce(
473 (accumulator, workerNode) =>
474 !workerNode.info.dynamic && workerNode.info.ready
475 ? accumulator + 1
476 : accumulator,
477 0
478 ) >= this.minSize
2431bdb4
JB
479 )
480 }
481
afe0d5bf 482 /**
aa9eede8 483 * The approximate pool utilization.
afe0d5bf
JB
484 *
485 * @returns The pool utilization.
486 */
487 private get utilization (): number {
8e5ca040 488 const poolTimeCapacity =
fe7d90db 489 (performance.now() - this.startTimestamp) * this.maxSize
afe0d5bf
JB
490 const totalTasksRunTime = this.workerNodes.reduce(
491 (accumulator, workerNode) =>
71514351 492 accumulator + (workerNode.usage.runTime?.aggregate ?? 0),
afe0d5bf
JB
493 0
494 )
495 const totalTasksWaitTime = this.workerNodes.reduce(
496 (accumulator, workerNode) =>
71514351 497 accumulator + (workerNode.usage.waitTime?.aggregate ?? 0),
afe0d5bf
JB
498 0
499 )
8e5ca040 500 return (totalTasksRunTime + totalTasksWaitTime) / poolTimeCapacity
afe0d5bf
JB
501 }
502
8881ae32 503 /**
aa9eede8 504 * The pool type.
8881ae32
JB
505 *
506 * If it is `'dynamic'`, it provides the `max` property.
507 */
508 protected abstract get type (): PoolType
509
184855e6 510 /**
aa9eede8 511 * The worker type.
184855e6
JB
512 */
513 protected abstract get worker (): WorkerType
514
c2ade475 515 /**
aa9eede8 516 * The pool minimum size.
c2ade475 517 */
8735b4e5
JB
518 protected get minSize (): number {
519 return this.numberOfWorkers
520 }
ff733df7
JB
521
522 /**
aa9eede8 523 * The pool maximum size.
ff733df7 524 */
8735b4e5
JB
525 protected get maxSize (): number {
526 return this.max ?? this.numberOfWorkers
527 }
a35560ba 528
6b813701
JB
529 /**
530 * Checks if the worker id sent in the received message from a worker is valid.
531 *
532 * @param message - The received message.
533 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the worker id is invalid.
534 */
21f710aa 535 private checkMessageWorkerId (message: MessageValue<Response>): void {
310de0aa
JB
536 if (message.workerId == null) {
537 throw new Error('Worker message received without worker id')
538 } else if (
21f710aa 539 message.workerId != null &&
aad6fb64 540 this.getWorkerNodeKeyByWorkerId(message.workerId) === -1
21f710aa
JB
541 ) {
542 throw new Error(
543 `Worker message received from unknown worker '${message.workerId}'`
544 )
545 }
546 }
547
ffcbbad8 548 /**
f06e48d8 549 * Gets the given worker its worker node key.
ffcbbad8
JB
550 *
551 * @param worker - The worker.
f59e1027 552 * @returns The worker node key if found in the pool worker nodes, `-1` otherwise.
ffcbbad8 553 */
aad6fb64 554 private getWorkerNodeKeyByWorker (worker: Worker): number {
f06e48d8 555 return this.workerNodes.findIndex(
8ebe6c30 556 (workerNode) => workerNode.worker === worker
f06e48d8 557 )
bf9549ae
JB
558 }
559
aa9eede8
JB
560 /**
561 * Gets the worker node key given its worker id.
562 *
563 * @param workerId - The worker id.
aad6fb64 564 * @returns The worker node key if the worker id is found in the pool worker nodes, `-1` otherwise.
aa9eede8 565 */
aad6fb64
JB
566 private getWorkerNodeKeyByWorkerId (workerId: number): number {
567 return this.workerNodes.findIndex(
8ebe6c30 568 (workerNode) => workerNode.info.id === workerId
aad6fb64 569 )
aa9eede8
JB
570 }
571
afc003b2 572 /** @inheritDoc */
a35560ba 573 public setWorkerChoiceStrategy (
59219cbb
JB
574 workerChoiceStrategy: WorkerChoiceStrategy,
575 workerChoiceStrategyOptions?: WorkerChoiceStrategyOptions
a35560ba 576 ): void {
aee46736 577 this.checkValidWorkerChoiceStrategy(workerChoiceStrategy)
b98ec2e6 578 this.opts.workerChoiceStrategy = workerChoiceStrategy
b6b32453
JB
579 this.workerChoiceStrategyContext.setWorkerChoiceStrategy(
580 this.opts.workerChoiceStrategy
581 )
582 if (workerChoiceStrategyOptions != null) {
583 this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions)
584 }
aa9eede8 585 for (const [workerNodeKey, workerNode] of this.workerNodes.entries()) {
4b628b48 586 workerNode.resetUsage()
9edb9717 587 this.sendStatisticsMessageToWorker(workerNodeKey)
59219cbb 588 }
a20f0ba5
JB
589 }
590
591 /** @inheritDoc */
592 public setWorkerChoiceStrategyOptions (
593 workerChoiceStrategyOptions: WorkerChoiceStrategyOptions
594 ): void {
0d80593b 595 this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions)
8990357d
JB
596 this.opts.workerChoiceStrategyOptions = {
597 ...DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS,
598 ...workerChoiceStrategyOptions
599 }
a20f0ba5
JB
600 this.workerChoiceStrategyContext.setOptions(
601 this.opts.workerChoiceStrategyOptions
a35560ba
S
602 )
603 }
604
a20f0ba5 605 /** @inheritDoc */
8f52842f
JB
606 public enableTasksQueue (
607 enable: boolean,
608 tasksQueueOptions?: TasksQueueOptions
609 ): void {
a20f0ba5 610 if (this.opts.enableTasksQueue === true && !enable) {
ef41a6e6 611 this.flushTasksQueues()
a20f0ba5
JB
612 }
613 this.opts.enableTasksQueue = enable
8f52842f 614 this.setTasksQueueOptions(tasksQueueOptions as TasksQueueOptions)
a20f0ba5
JB
615 }
616
617 /** @inheritDoc */
8f52842f 618 public setTasksQueueOptions (tasksQueueOptions: TasksQueueOptions): void {
a20f0ba5 619 if (this.opts.enableTasksQueue === true) {
8f52842f
JB
620 this.checkValidTasksQueueOptions(tasksQueueOptions)
621 this.opts.tasksQueueOptions =
622 this.buildTasksQueueOptions(tasksQueueOptions)
5baee0d7 623 } else if (this.opts.tasksQueueOptions != null) {
a20f0ba5
JB
624 delete this.opts.tasksQueueOptions
625 }
626 }
627
628 private buildTasksQueueOptions (
629 tasksQueueOptions: TasksQueueOptions
630 ): TasksQueueOptions {
631 return {
632 concurrency: tasksQueueOptions?.concurrency ?? 1
633 }
634 }
635
c319c66b
JB
636 /**
637 * Whether the pool is full or not.
638 *
639 * The pool filling boolean status.
640 */
dea903a8
JB
641 protected get full (): boolean {
642 return this.workerNodes.length >= this.maxSize
643 }
c2ade475 644
c319c66b
JB
645 /**
646 * Whether the pool is busy or not.
647 *
648 * The pool busyness boolean status.
649 */
650 protected abstract get busy (): boolean
7c0ba920 651
6c6afb84 652 /**
3d76750a 653 * Whether worker nodes are executing concurrently their tasks quota or not.
6c6afb84
JB
654 *
655 * @returns Worker nodes busyness boolean status.
656 */
c2ade475 657 protected internalBusy (): boolean {
3d76750a
JB
658 if (this.opts.enableTasksQueue === true) {
659 return (
660 this.workerNodes.findIndex(
8ebe6c30 661 (workerNode) =>
3d76750a
JB
662 workerNode.info.ready &&
663 workerNode.usage.tasks.executing <
664 (this.opts.tasksQueueOptions?.concurrency as number)
665 ) === -1
666 )
667 } else {
668 return (
669 this.workerNodes.findIndex(
8ebe6c30 670 (workerNode) =>
3d76750a
JB
671 workerNode.info.ready && workerNode.usage.tasks.executing === 0
672 ) === -1
673 )
674 }
cb70b19d
JB
675 }
676
90d7d101
JB
677 /** @inheritDoc */
678 public listTaskFunctions (): string[] {
f2dbbf95
JB
679 for (const workerNode of this.workerNodes) {
680 if (
681 Array.isArray(workerNode.info.taskFunctions) &&
682 workerNode.info.taskFunctions.length > 0
683 ) {
684 return workerNode.info.taskFunctions
685 }
90d7d101 686 }
f2dbbf95 687 return []
90d7d101
JB
688 }
689
afc003b2 690 /** @inheritDoc */
7d91a8cd
JB
691 public async execute (
692 data?: Data,
693 name?: string,
694 transferList?: TransferListItem[]
695 ): Promise<Response> {
52b71763 696 return await new Promise<Response>((resolve, reject) => {
7d91a8cd
JB
697 if (name != null && typeof name !== 'string') {
698 reject(new TypeError('name argument must be a string'))
699 }
90d7d101
JB
700 if (
701 name != null &&
702 typeof name === 'string' &&
703 name.trim().length === 0
704 ) {
f58b60b9 705 reject(new TypeError('name argument must not be an empty string'))
90d7d101 706 }
b558f6b5
JB
707 if (transferList != null && !Array.isArray(transferList)) {
708 reject(new TypeError('transferList argument must be an array'))
709 }
710 const timestamp = performance.now()
711 const workerNodeKey = this.chooseWorkerNode()
94407def 712 const workerInfo = this.getWorkerInfo(workerNodeKey) as WorkerInfo
cea399c8
JB
713 if (
714 name != null &&
a5d15204
JB
715 Array.isArray(workerInfo.taskFunctions) &&
716 !workerInfo.taskFunctions.includes(name)
cea399c8 717 ) {
90d7d101
JB
718 reject(
719 new Error(`Task function '${name}' is not registered in the pool`)
720 )
721 }
501aea93 722 const task: Task<Data> = {
52b71763
JB
723 name: name ?? DEFAULT_TASK_NAME,
724 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
725 data: data ?? ({} as Data),
7d91a8cd 726 transferList,
52b71763 727 timestamp,
a5d15204 728 workerId: workerInfo.id as number,
7629bdf1 729 taskId: randomUUID()
52b71763 730 }
7629bdf1 731 this.promiseResponseMap.set(task.taskId as string, {
2e81254d
JB
732 resolve,
733 reject,
501aea93 734 workerNodeKey
2e81254d 735 })
52b71763 736 if (
4e377863
JB
737 this.opts.enableTasksQueue === false ||
738 (this.opts.enableTasksQueue === true &&
739 this.workerNodes[workerNodeKey].usage.tasks.executing <
b5e113f6 740 (this.opts.tasksQueueOptions?.concurrency as number))
52b71763 741 ) {
501aea93 742 this.executeTask(workerNodeKey, task)
4e377863
JB
743 } else {
744 this.enqueueTask(workerNodeKey, task)
52b71763 745 }
2e81254d 746 })
280c2a77 747 }
c97c7edb 748
afc003b2 749 /** @inheritDoc */
c97c7edb 750 public async destroy (): Promise<void> {
1fbcaa7c 751 await Promise.all(
81c02522 752 this.workerNodes.map(async (_, workerNodeKey) => {
aa9eede8 753 await this.destroyWorkerNode(workerNodeKey)
1fbcaa7c
JB
754 })
755 )
33e6bb4c 756 this.emitter?.emit(PoolEvents.destroy, this.info)
c97c7edb
S
757 }
758
1e3214b6
JB
759 protected async sendKillMessageToWorker (
760 workerNodeKey: number,
761 workerId: number
762 ): Promise<void> {
9edb9717 763 await new Promise<void>((resolve, reject) => {
1e3214b6
JB
764 this.registerWorkerMessageListener(workerNodeKey, (message) => {
765 if (message.kill === 'success') {
766 resolve()
767 } else if (message.kill === 'failure') {
e1af34e6 768 reject(new Error(`Worker ${workerId} kill message handling failed`))
1e3214b6
JB
769 }
770 })
9edb9717 771 this.sendToWorker(workerNodeKey, { kill: true, workerId })
1e3214b6 772 })
1e3214b6
JB
773 }
774
4a6952ff 775 /**
aa9eede8 776 * Terminates the worker node given its worker node key.
4a6952ff 777 *
aa9eede8 778 * @param workerNodeKey - The worker node key.
4a6952ff 779 */
81c02522 780 protected abstract destroyWorkerNode (workerNodeKey: number): Promise<void>
c97c7edb 781
729c563d 782 /**
6677a3d3
JB
783 * Setup hook to execute code before worker nodes are created in the abstract constructor.
784 * Can be overridden.
afc003b2
JB
785 *
786 * @virtual
729c563d 787 */
280c2a77 788 protected setupHook (): void {
d99ba5a8 789 // Intentionally empty
280c2a77 790 }
c97c7edb 791
729c563d 792 /**
280c2a77
S
793 * Should return whether the worker is the main worker or not.
794 */
795 protected abstract isMain (): boolean
796
797 /**
2e81254d 798 * Hook executed before the worker task execution.
bf9549ae 799 * Can be overridden.
729c563d 800 *
f06e48d8 801 * @param workerNodeKey - The worker node key.
1c6fe997 802 * @param task - The task to execute.
729c563d 803 */
1c6fe997
JB
804 protected beforeTaskExecutionHook (
805 workerNodeKey: number,
806 task: Task<Data>
807 ): void {
94407def
JB
808 if (this.workerNodes[workerNodeKey]?.usage != null) {
809 const workerUsage = this.workerNodes[workerNodeKey].usage
810 ++workerUsage.tasks.executing
811 this.updateWaitTimeWorkerUsage(workerUsage, task)
812 }
813 if (
814 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) &&
815 this.workerNodes[workerNodeKey].getTaskFunctionWorkerUsage(
816 task.name as string
817 ) != null
818 ) {
db0e38ee 819 const taskFunctionWorkerUsage = this.workerNodes[
b558f6b5 820 workerNodeKey
db0e38ee 821 ].getTaskFunctionWorkerUsage(task.name as string) as WorkerUsage
5623b8d5
JB
822 ++taskFunctionWorkerUsage.tasks.executing
823 this.updateWaitTimeWorkerUsage(taskFunctionWorkerUsage, task)
b558f6b5 824 }
c97c7edb
S
825 }
826
c01733f1 827 /**
2e81254d 828 * Hook executed after the worker task execution.
bf9549ae 829 * Can be overridden.
c01733f1 830 *
501aea93 831 * @param workerNodeKey - The worker node key.
38e795c1 832 * @param message - The received message.
c01733f1 833 */
2e81254d 834 protected afterTaskExecutionHook (
501aea93 835 workerNodeKey: number,
2740a743 836 message: MessageValue<Response>
bf9549ae 837 ): void {
94407def
JB
838 if (this.workerNodes[workerNodeKey]?.usage != null) {
839 const workerUsage = this.workerNodes[workerNodeKey].usage
840 this.updateTaskStatisticsWorkerUsage(workerUsage, message)
841 this.updateRunTimeWorkerUsage(workerUsage, message)
842 this.updateEluWorkerUsage(workerUsage, message)
843 }
844 if (
845 this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) &&
846 this.workerNodes[workerNodeKey].getTaskFunctionWorkerUsage(
5623b8d5 847 message.taskPerformance?.name as string
94407def
JB
848 ) != null
849 ) {
db0e38ee 850 const taskFunctionWorkerUsage = this.workerNodes[
b558f6b5 851 workerNodeKey
db0e38ee 852 ].getTaskFunctionWorkerUsage(
b558f6b5
JB
853 message.taskPerformance?.name ?? DEFAULT_TASK_NAME
854 ) as WorkerUsage
db0e38ee
JB
855 this.updateTaskStatisticsWorkerUsage(taskFunctionWorkerUsage, message)
856 this.updateRunTimeWorkerUsage(taskFunctionWorkerUsage, message)
857 this.updateEluWorkerUsage(taskFunctionWorkerUsage, message)
b558f6b5
JB
858 }
859 }
860
db0e38ee
JB
861 /**
862 * Whether the worker node shall update its task function worker usage or not.
863 *
864 * @param workerNodeKey - The worker node key.
865 * @returns `true` if the worker node shall update its task function worker usage, `false` otherwise.
866 */
867 private shallUpdateTaskFunctionWorkerUsage (workerNodeKey: number): boolean {
a5d15204 868 const workerInfo = this.getWorkerInfo(workerNodeKey)
b558f6b5 869 return (
94407def 870 workerInfo != null &&
a5d15204 871 Array.isArray(workerInfo.taskFunctions) &&
db0e38ee 872 workerInfo.taskFunctions.length > 2
b558f6b5 873 )
f1c06930
JB
874 }
875
876 private updateTaskStatisticsWorkerUsage (
877 workerUsage: WorkerUsage,
878 message: MessageValue<Response>
879 ): void {
a4e07f72 880 const workerTaskStatistics = workerUsage.tasks
5bb5be17
JB
881 if (
882 workerTaskStatistics.executing != null &&
883 workerTaskStatistics.executing > 0
884 ) {
885 --workerTaskStatistics.executing
886 } else if (
887 workerTaskStatistics.executing != null &&
888 workerTaskStatistics.executing < 0
889 ) {
890 throw new Error(
cf6e3991 891 'Worker usage statistic for tasks executing cannot be negative'
5bb5be17
JB
892 )
893 }
98e72cda
JB
894 if (message.taskError == null) {
895 ++workerTaskStatistics.executed
896 } else {
a4e07f72 897 ++workerTaskStatistics.failed
2740a743 898 }
f8eb0a2a
JB
899 }
900
a4e07f72
JB
901 private updateRunTimeWorkerUsage (
902 workerUsage: WorkerUsage,
f8eb0a2a
JB
903 message: MessageValue<Response>
904 ): void {
e4f20deb
JB
905 updateMeasurementStatistics(
906 workerUsage.runTime,
907 this.workerChoiceStrategyContext.getTaskStatisticsRequirements().runTime,
908 message.taskPerformance?.runTime ?? 0,
909 workerUsage.tasks.executed
910 )
f8eb0a2a
JB
911 }
912
a4e07f72
JB
913 private updateWaitTimeWorkerUsage (
914 workerUsage: WorkerUsage,
1c6fe997 915 task: Task<Data>
f8eb0a2a 916 ): void {
1c6fe997
JB
917 const timestamp = performance.now()
918 const taskWaitTime = timestamp - (task.timestamp ?? timestamp)
e4f20deb
JB
919 updateMeasurementStatistics(
920 workerUsage.waitTime,
921 this.workerChoiceStrategyContext.getTaskStatisticsRequirements().waitTime,
922 taskWaitTime,
923 workerUsage.tasks.executed
924 )
c01733f1 925 }
926
a4e07f72 927 private updateEluWorkerUsage (
5df69fab 928 workerUsage: WorkerUsage,
62c15a68
JB
929 message: MessageValue<Response>
930 ): void {
008512c7
JB
931 const eluTaskStatisticsRequirements: MeasurementStatisticsRequirements =
932 this.workerChoiceStrategyContext.getTaskStatisticsRequirements().elu
e4f20deb
JB
933 updateMeasurementStatistics(
934 workerUsage.elu.active,
008512c7 935 eluTaskStatisticsRequirements,
e4f20deb
JB
936 message.taskPerformance?.elu?.active ?? 0,
937 workerUsage.tasks.executed
938 )
939 updateMeasurementStatistics(
940 workerUsage.elu.idle,
008512c7 941 eluTaskStatisticsRequirements,
e4f20deb
JB
942 message.taskPerformance?.elu?.idle ?? 0,
943 workerUsage.tasks.executed
944 )
008512c7 945 if (eluTaskStatisticsRequirements.aggregate) {
f7510105 946 if (message.taskPerformance?.elu != null) {
f7510105
JB
947 if (workerUsage.elu.utilization != null) {
948 workerUsage.elu.utilization =
949 (workerUsage.elu.utilization +
950 message.taskPerformance.elu.utilization) /
951 2
952 } else {
953 workerUsage.elu.utilization = message.taskPerformance.elu.utilization
954 }
62c15a68
JB
955 }
956 }
957 }
958
280c2a77 959 /**
f06e48d8 960 * Chooses a worker node for the next task.
280c2a77 961 *
6c6afb84 962 * The default worker choice strategy uses a round robin algorithm to distribute the tasks.
280c2a77 963 *
aa9eede8 964 * @returns The chosen worker node key
280c2a77 965 */
6c6afb84 966 private chooseWorkerNode (): number {
930dcf12 967 if (this.shallCreateDynamicWorker()) {
aa9eede8 968 const workerNodeKey = this.createAndSetupDynamicWorkerNode()
6c6afb84 969 if (
b1aae695 970 this.workerChoiceStrategyContext.getStrategyPolicy().dynamicWorkerUsage
6c6afb84 971 ) {
aa9eede8 972 return workerNodeKey
6c6afb84 973 }
17393ac8 974 }
930dcf12
JB
975 return this.workerChoiceStrategyContext.execute()
976 }
977
6c6afb84
JB
978 /**
979 * Conditions for dynamic worker creation.
980 *
981 * @returns Whether to create a dynamic worker or not.
982 */
983 private shallCreateDynamicWorker (): boolean {
930dcf12 984 return this.type === PoolTypes.dynamic && !this.full && this.internalBusy()
c97c7edb
S
985 }
986
280c2a77 987 /**
aa9eede8 988 * Sends a message to worker given its worker node key.
280c2a77 989 *
aa9eede8 990 * @param workerNodeKey - The worker node key.
38e795c1 991 * @param message - The message.
7d91a8cd 992 * @param transferList - The optional array of transferable objects.
280c2a77
S
993 */
994 protected abstract sendToWorker (
aa9eede8 995 workerNodeKey: number,
7d91a8cd
JB
996 message: MessageValue<Data>,
997 transferList?: TransferListItem[]
280c2a77
S
998 ): void
999
729c563d 1000 /**
41344292 1001 * Creates a new worker.
6c6afb84
JB
1002 *
1003 * @returns Newly created worker.
729c563d 1004 */
280c2a77 1005 protected abstract createWorker (): Worker
c97c7edb 1006
4a6952ff 1007 /**
aa9eede8 1008 * Creates a new, completely set up worker node.
4a6952ff 1009 *
aa9eede8 1010 * @returns New, completely set up worker node key.
4a6952ff 1011 */
aa9eede8 1012 protected createAndSetupWorkerNode (): number {
bdacc2d2 1013 const worker = this.createWorker()
280c2a77 1014
fd04474e 1015 worker.on('online', this.opts.onlineHandler ?? EMPTY_FUNCTION)
35cf1c03 1016 worker.on('message', this.opts.messageHandler ?? EMPTY_FUNCTION)
a35560ba 1017 worker.on('error', this.opts.errorHandler ?? EMPTY_FUNCTION)
8ebe6c30 1018 worker.on('error', (error) => {
aad6fb64 1019 const workerNodeKey = this.getWorkerNodeKeyByWorker(worker)
94407def 1020 const workerInfo = this.getWorkerInfo(workerNodeKey) as WorkerInfo
9b106837 1021 workerInfo.ready = false
0dc838e3 1022 this.workerNodes[workerNodeKey].closeChannel()
2a69b8c5 1023 this.emitter?.emit(PoolEvents.error, error)
2431bdb4 1024 if (this.opts.restartWorkerOnError === true && !this.starting) {
9b106837 1025 if (workerInfo.dynamic) {
aa9eede8 1026 this.createAndSetupDynamicWorkerNode()
8a1260a3 1027 } else {
aa9eede8 1028 this.createAndSetupWorkerNode()
8a1260a3 1029 }
5baee0d7 1030 }
19dbc45b 1031 if (this.opts.enableTasksQueue === true) {
9b106837 1032 this.redistributeQueuedTasks(workerNodeKey)
19dbc45b 1033 }
5baee0d7 1034 })
a35560ba 1035 worker.on('exit', this.opts.exitHandler ?? EMPTY_FUNCTION)
a974afa6 1036 worker.once('exit', () => {
f06e48d8 1037 this.removeWorkerNode(worker)
a974afa6 1038 })
280c2a77 1039
aa9eede8 1040 const workerNodeKey = this.addWorkerNode(worker)
280c2a77 1041
aa9eede8 1042 this.afterWorkerNodeSetup(workerNodeKey)
280c2a77 1043
aa9eede8 1044 return workerNodeKey
c97c7edb 1045 }
be0676b3 1046
930dcf12 1047 /**
aa9eede8 1048 * Creates a new, completely set up dynamic worker node.
930dcf12 1049 *
aa9eede8 1050 * @returns New, completely set up dynamic worker node key.
930dcf12 1051 */
aa9eede8
JB
1052 protected createAndSetupDynamicWorkerNode (): number {
1053 const workerNodeKey = this.createAndSetupWorkerNode()
8ebe6c30 1054 this.registerWorkerMessageListener(workerNodeKey, (message) => {
aa9eede8
JB
1055 const localWorkerNodeKey = this.getWorkerNodeKeyByWorkerId(
1056 message.workerId
aad6fb64 1057 )
aa9eede8 1058 const workerUsage = this.workerNodes[localWorkerNodeKey].usage
81c02522 1059 // Kill message received from worker
930dcf12
JB
1060 if (
1061 isKillBehavior(KillBehaviors.HARD, message.kill) ||
1e3214b6 1062 (isKillBehavior(KillBehaviors.SOFT, message.kill) &&
7b56f532 1063 ((this.opts.enableTasksQueue === false &&
aa9eede8 1064 workerUsage.tasks.executing === 0) ||
7b56f532 1065 (this.opts.enableTasksQueue === true &&
aa9eede8
JB
1066 workerUsage.tasks.executing === 0 &&
1067 this.tasksQueueSize(localWorkerNodeKey) === 0)))
930dcf12 1068 ) {
5270d253
JB
1069 this.destroyWorkerNode(localWorkerNodeKey).catch((error) => {
1070 this.emitter?.emit(PoolEvents.error, error)
1071 })
930dcf12
JB
1072 }
1073 })
94407def 1074 const workerInfo = this.getWorkerInfo(workerNodeKey) as WorkerInfo
aa9eede8 1075 this.sendToWorker(workerNodeKey, {
b0a4db63 1076 checkActive: true,
21f710aa
JB
1077 workerId: workerInfo.id as number
1078 })
b5e113f6 1079 workerInfo.dynamic = true
b1aae695
JB
1080 if (
1081 this.workerChoiceStrategyContext.getStrategyPolicy().dynamicWorkerReady ||
1082 this.workerChoiceStrategyContext.getStrategyPolicy().dynamicWorkerUsage
1083 ) {
b5e113f6
JB
1084 workerInfo.ready = true
1085 }
33e6bb4c 1086 this.checkAndEmitDynamicWorkerCreationEvents()
aa9eede8 1087 return workerNodeKey
930dcf12
JB
1088 }
1089
a2ed5053 1090 /**
aa9eede8 1091 * Registers a listener callback on the worker given its worker node key.
a2ed5053 1092 *
aa9eede8 1093 * @param workerNodeKey - The worker node key.
a2ed5053
JB
1094 * @param listener - The message listener callback.
1095 */
85aeb3f3
JB
1096 protected abstract registerWorkerMessageListener<
1097 Message extends Data | Response
aa9eede8
JB
1098 >(
1099 workerNodeKey: number,
1100 listener: (message: MessageValue<Message>) => void
1101 ): void
a2ed5053
JB
1102
1103 /**
aa9eede8 1104 * Method hooked up after a worker node has been newly created.
a2ed5053
JB
1105 * Can be overridden.
1106 *
aa9eede8 1107 * @param workerNodeKey - The newly created worker node key.
a2ed5053 1108 */
aa9eede8 1109 protected afterWorkerNodeSetup (workerNodeKey: number): void {
a2ed5053 1110 // Listen to worker messages.
aa9eede8 1111 this.registerWorkerMessageListener(workerNodeKey, this.workerListener())
85aeb3f3 1112 // Send the startup message to worker.
aa9eede8 1113 this.sendStartupMessageToWorker(workerNodeKey)
9edb9717
JB
1114 // Send the statistics message to worker.
1115 this.sendStatisticsMessageToWorker(workerNodeKey)
d2c73f82
JB
1116 }
1117
85aeb3f3 1118 /**
aa9eede8
JB
1119 * Sends the startup message to worker given its worker node key.
1120 *
1121 * @param workerNodeKey - The worker node key.
1122 */
1123 protected abstract sendStartupMessageToWorker (workerNodeKey: number): void
1124
1125 /**
9edb9717 1126 * Sends the statistics message to worker given its worker node key.
85aeb3f3 1127 *
aa9eede8 1128 * @param workerNodeKey - The worker node key.
85aeb3f3 1129 */
9edb9717 1130 private sendStatisticsMessageToWorker (workerNodeKey: number): void {
aa9eede8
JB
1131 this.sendToWorker(workerNodeKey, {
1132 statistics: {
1133 runTime:
1134 this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
1135 .runTime.aggregate,
1136 elu: this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
1137 .elu.aggregate
1138 },
94407def 1139 workerId: (this.getWorkerInfo(workerNodeKey) as WorkerInfo).id as number
aa9eede8
JB
1140 })
1141 }
a2ed5053
JB
1142
1143 private redistributeQueuedTasks (workerNodeKey: number): void {
1144 while (this.tasksQueueSize(workerNodeKey) > 0) {
1145 let targetWorkerNodeKey: number = workerNodeKey
1146 let minQueuedTasks = Infinity
10ecf8fd 1147 let executeTask = false
a2ed5053 1148 for (const [workerNodeId, workerNode] of this.workerNodes.entries()) {
94407def 1149 const workerInfo = this.getWorkerInfo(workerNodeId) as WorkerInfo
a2ed5053
JB
1150 if (
1151 workerNodeId !== workerNodeKey &&
1152 workerInfo.ready &&
1153 workerNode.usage.tasks.queued === 0
1154 ) {
a5ed75b7
JB
1155 if (
1156 this.workerNodes[workerNodeId].usage.tasks.executing <
1157 (this.opts.tasksQueueOptions?.concurrency as number)
1158 ) {
10ecf8fd
JB
1159 executeTask = true
1160 }
a2ed5053
JB
1161 targetWorkerNodeKey = workerNodeId
1162 break
1163 }
1164 if (
1165 workerNodeId !== workerNodeKey &&
1166 workerInfo.ready &&
1167 workerNode.usage.tasks.queued < minQueuedTasks
1168 ) {
1169 minQueuedTasks = workerNode.usage.tasks.queued
1170 targetWorkerNodeKey = workerNodeId
1171 }
1172 }
10ecf8fd
JB
1173 if (executeTask) {
1174 this.executeTask(
1175 targetWorkerNodeKey,
1176 this.dequeueTask(workerNodeKey) as Task<Data>
1177 )
1178 } else {
1179 this.enqueueTask(
1180 targetWorkerNodeKey,
1181 this.dequeueTask(workerNodeKey) as Task<Data>
1182 )
1183 }
a2ed5053
JB
1184 }
1185 }
1186
be0676b3 1187 /**
aa9eede8 1188 * This method is the listener registered for each worker message.
be0676b3 1189 *
bdacc2d2 1190 * @returns The listener function to execute when a message is received from a worker.
be0676b3
APA
1191 */
1192 protected workerListener (): (message: MessageValue<Response>) => void {
8ebe6c30 1193 return (message) => {
21f710aa 1194 this.checkMessageWorkerId(message)
a5d15204 1195 if (message.ready != null && message.taskFunctions != null) {
81c02522 1196 // Worker ready response received from worker
10e2aa7e 1197 this.handleWorkerReadyResponse(message)
7629bdf1 1198 } else if (message.taskId != null) {
81c02522 1199 // Task execution response received from worker
6b272951 1200 this.handleTaskExecutionResponse(message)
90d7d101
JB
1201 } else if (message.taskFunctions != null) {
1202 // Task functions message received from worker
94407def
JB
1203 (
1204 this.getWorkerInfo(
1205 this.getWorkerNodeKeyByWorkerId(message.workerId)
1206 ) as WorkerInfo
b558f6b5 1207 ).taskFunctions = message.taskFunctions
6b272951
JB
1208 }
1209 }
1210 }
1211
10e2aa7e 1212 private handleWorkerReadyResponse (message: MessageValue<Response>): void {
f05ed93c
JB
1213 if (message.ready === false) {
1214 throw new Error(`Worker ${message.workerId} failed to initialize`)
1215 }
a5d15204 1216 const workerInfo = this.getWorkerInfo(
aad6fb64 1217 this.getWorkerNodeKeyByWorkerId(message.workerId)
94407def 1218 ) as WorkerInfo
a5d15204
JB
1219 workerInfo.ready = message.ready as boolean
1220 workerInfo.taskFunctions = message.taskFunctions
2431bdb4
JB
1221 if (this.emitter != null && this.ready) {
1222 this.emitter.emit(PoolEvents.ready, this.info)
1223 }
6b272951
JB
1224 }
1225
1226 private handleTaskExecutionResponse (message: MessageValue<Response>): void {
5441aea6
JB
1227 const { taskId, taskError, data } = message
1228 const promiseResponse = this.promiseResponseMap.get(taskId as string)
6b272951 1229 if (promiseResponse != null) {
5441aea6
JB
1230 if (taskError != null) {
1231 this.emitter?.emit(PoolEvents.taskError, taskError)
1232 promiseResponse.reject(taskError.message)
6b272951 1233 } else {
5441aea6 1234 promiseResponse.resolve(data as Response)
6b272951 1235 }
501aea93
JB
1236 const workerNodeKey = promiseResponse.workerNodeKey
1237 this.afterTaskExecutionHook(workerNodeKey, message)
5441aea6 1238 this.promiseResponseMap.delete(taskId as string)
6b272951
JB
1239 if (
1240 this.opts.enableTasksQueue === true &&
b5e113f6
JB
1241 this.tasksQueueSize(workerNodeKey) > 0 &&
1242 this.workerNodes[workerNodeKey].usage.tasks.executing <
1243 (this.opts.tasksQueueOptions?.concurrency as number)
6b272951
JB
1244 ) {
1245 this.executeTask(
1246 workerNodeKey,
1247 this.dequeueTask(workerNodeKey) as Task<Data>
1248 )
be0676b3 1249 }
6b272951 1250 this.workerChoiceStrategyContext.update(workerNodeKey)
be0676b3 1251 }
be0676b3 1252 }
7c0ba920 1253
a1763c54 1254 private checkAndEmitTaskExecutionEvents (): void {
33e6bb4c
JB
1255 if (this.busy) {
1256 this.emitter?.emit(PoolEvents.busy, this.info)
a1763c54
JB
1257 }
1258 }
1259
1260 private checkAndEmitTaskQueuingEvents (): void {
1261 if (this.hasBackPressure()) {
1262 this.emitter?.emit(PoolEvents.backPressure, this.info)
164d950a
JB
1263 }
1264 }
1265
33e6bb4c
JB
1266 private checkAndEmitDynamicWorkerCreationEvents (): void {
1267 if (this.type === PoolTypes.dynamic) {
1268 if (this.full) {
1269 this.emitter?.emit(PoolEvents.full, this.info)
1270 }
1271 }
1272 }
1273
8a1260a3 1274 /**
aa9eede8 1275 * Gets the worker information given its worker node key.
8a1260a3
JB
1276 *
1277 * @param workerNodeKey - The worker node key.
3f09ed9f 1278 * @returns The worker information.
8a1260a3 1279 */
94407def
JB
1280 protected getWorkerInfo (workerNodeKey: number): WorkerInfo | undefined {
1281 return this.workerNodes[workerNodeKey]?.info
e221309a
JB
1282 }
1283
a05c10de 1284 /**
b0a4db63 1285 * Adds the given worker in the pool worker nodes.
ea7a90d3 1286 *
38e795c1 1287 * @param worker - The worker.
aa9eede8
JB
1288 * @returns The added worker node key.
1289 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the added worker node is not found.
ea7a90d3 1290 */
b0a4db63 1291 private addWorkerNode (worker: Worker): number {
671d5154
JB
1292 const workerNode = new WorkerNode<Worker, Data>(
1293 worker,
1294 this.worker,
1295 this.maxSize
1296 )
b97d82d8 1297 // Flag the worker node as ready at pool startup.
d2c73f82
JB
1298 if (this.starting) {
1299 workerNode.info.ready = true
1300 }
aa9eede8 1301 this.workerNodes.push(workerNode)
aad6fb64 1302 const workerNodeKey = this.getWorkerNodeKeyByWorker(worker)
aa9eede8 1303 if (workerNodeKey === -1) {
a5d15204 1304 throw new Error('Worker node added not found')
aa9eede8
JB
1305 }
1306 return workerNodeKey
ea7a90d3 1307 }
c923ce56 1308
51fe3d3c 1309 /**
f06e48d8 1310 * Removes the given worker from the pool worker nodes.
51fe3d3c 1311 *
f06e48d8 1312 * @param worker - The worker.
51fe3d3c 1313 */
416fd65c 1314 private removeWorkerNode (worker: Worker): void {
aad6fb64 1315 const workerNodeKey = this.getWorkerNodeKeyByWorker(worker)
1f68cede
JB
1316 if (workerNodeKey !== -1) {
1317 this.workerNodes.splice(workerNodeKey, 1)
1318 this.workerChoiceStrategyContext.remove(workerNodeKey)
1319 }
51fe3d3c 1320 }
adc3c320 1321
e2b31e32
JB
1322 /** @inheritDoc */
1323 public hasWorkerNodeBackPressure (workerNodeKey: number): boolean {
9e844245 1324 return (
e2b31e32
JB
1325 this.opts.enableTasksQueue === true &&
1326 this.workerNodes[workerNodeKey].hasBackPressure()
9e844245
JB
1327 )
1328 }
1329
1330 private hasBackPressure (): boolean {
1331 return (
1332 this.opts.enableTasksQueue === true &&
1333 this.workerNodes.findIndex(
1334 (workerNode) => !workerNode.hasBackPressure()
a1763c54 1335 ) === -1
9e844245 1336 )
e2b31e32
JB
1337 }
1338
b0a4db63 1339 /**
aa9eede8 1340 * Executes the given task on the worker given its worker node key.
b0a4db63 1341 *
aa9eede8 1342 * @param workerNodeKey - The worker node key.
b0a4db63
JB
1343 * @param task - The task to execute.
1344 */
2e81254d 1345 private executeTask (workerNodeKey: number, task: Task<Data>): void {
1c6fe997 1346 this.beforeTaskExecutionHook(workerNodeKey, task)
bbfa38a2 1347 this.sendToWorker(workerNodeKey, task, task.transferList)
a1763c54 1348 this.checkAndEmitTaskExecutionEvents()
2e81254d
JB
1349 }
1350
f9f00b5f 1351 private enqueueTask (workerNodeKey: number, task: Task<Data>): number {
a1763c54
JB
1352 const tasksQueueSize = this.workerNodes[workerNodeKey].enqueueTask(task)
1353 this.checkAndEmitTaskQueuingEvents()
1354 return tasksQueueSize
adc3c320
JB
1355 }
1356
416fd65c 1357 private dequeueTask (workerNodeKey: number): Task<Data> | undefined {
4b628b48 1358 return this.workerNodes[workerNodeKey].dequeueTask()
adc3c320
JB
1359 }
1360
416fd65c 1361 private tasksQueueSize (workerNodeKey: number): number {
4b628b48 1362 return this.workerNodes[workerNodeKey].tasksQueueSize()
df593701
JB
1363 }
1364
81c02522 1365 protected flushTasksQueue (workerNodeKey: number): void {
920278a2
JB
1366 while (this.tasksQueueSize(workerNodeKey) > 0) {
1367 this.executeTask(
1368 workerNodeKey,
1369 this.dequeueTask(workerNodeKey) as Task<Data>
1370 )
ff733df7 1371 }
4b628b48 1372 this.workerNodes[workerNodeKey].clearTasksQueue()
ff733df7
JB
1373 }
1374
ef41a6e6
JB
1375 private flushTasksQueues (): void {
1376 for (const [workerNodeKey] of this.workerNodes.entries()) {
1377 this.flushTasksQueue(workerNodeKey)
1378 }
1379 }
c97c7edb 1380}