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