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