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