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