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