refactor: use eslint plugin to sort imports
[poolifier.git] / src / pools / abstract-pool.ts
CommitLineData
fc3e6586 1import crypto from 'node:crypto'
2740a743 2import type { MessageValue, PromiseResponseWrapper } from '../utility-types'
bbeadd16
JB
3import {
4 DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS,
5 EMPTY_FUNCTION,
6 median
7} from '../utils'
34a0cfab 8import { KillBehaviors, isKillBehavior } from '../worker/worker-options'
65d7a1c9 9import { CircularArray } from '../circular-array'
c4855468 10import {
65d7a1c9 11 type IPool,
7c5a1080 12 PoolEmitter,
c4855468 13 PoolEvents,
c4855468 14 type PoolOptions,
65d7a1c9
JB
15 PoolType,
16 type TasksQueueOptions
c4855468 17} from './pool'
f06e48d8 18import type { IWorker, Task, TasksUsage, WorkerNode } from './worker'
a35560ba
S
19import {
20 WorkerChoiceStrategies,
a20f0ba5
JB
21 type WorkerChoiceStrategy,
22 type WorkerChoiceStrategyOptions
bdaf31cd
JB
23} from './selection-strategies/selection-strategies-types'
24import { WorkerChoiceStrategyContext } from './selection-strategies/worker-choice-strategy-context'
c97c7edb 25
729c563d 26/**
ea7a90d3 27 * Base class that implements some shared logic for all poolifier pools.
729c563d 28 *
38e795c1
JB
29 * @typeParam Worker - Type of worker which manages this pool.
30 * @typeParam Data - Type of data sent to the worker. This can only be serializable data.
02706357 31 * @typeParam Response - Type of execution response. This can only be serializable data.
729c563d 32 */
c97c7edb 33export abstract class AbstractPool<
f06e48d8 34 Worker extends IWorker,
d3c8a1a8
S
35 Data = unknown,
36 Response = unknown
c4855468 37> implements IPool<Worker, Data, Response> {
afc003b2 38 /** @inheritDoc */
f06e48d8 39 public readonly workerNodes: Array<WorkerNode<Worker, Data>> = []
4a6952ff 40
afc003b2 41 /** @inheritDoc */
7c0ba920
JB
42 public readonly emitter?: PoolEmitter
43
be0676b3 44 /**
a3445496 45 * The execution response promise map.
be0676b3 46 *
2740a743 47 * - `key`: The message id of each submitted task.
a3445496 48 * - `value`: An object that contains the worker, the execution response promise resolve and reject callbacks.
be0676b3 49 *
a3445496 50 * When we receive a message from the worker, we get a map entry with the promise resolve/reject bound to the message id.
be0676b3 51 */
c923ce56
JB
52 protected promiseResponseMap: Map<
53 string,
54 PromiseResponseWrapper<Worker, Response>
55 > = new Map<string, PromiseResponseWrapper<Worker, Response>>()
c97c7edb 56
a35560ba 57 /**
51fe3d3c 58 * Worker choice strategy context referencing a worker choice algorithm implementation.
a35560ba 59 *
51fe3d3c 60 * Default to a round robin algorithm.
a35560ba
S
61 */
62 protected workerChoiceStrategyContext: WorkerChoiceStrategyContext<
78cea37e
JB
63 Worker,
64 Data,
65 Response
a35560ba
S
66 >
67
729c563d
S
68 /**
69 * Constructs a new poolifier pool.
70 *
38e795c1 71 * @param numberOfWorkers - Number of workers that this pool should manage.
029715f0 72 * @param filePath - Path to the worker file.
38e795c1 73 * @param opts - Options for the pool.
729c563d 74 */
c97c7edb 75 public constructor (
5c5a1fb7 76 public readonly numberOfWorkers: number,
c97c7edb 77 public readonly filePath: string,
1927ee67 78 public readonly opts: PoolOptions<Worker>
c97c7edb 79 ) {
78cea37e 80 if (!this.isMain()) {
c97c7edb
S
81 throw new Error('Cannot start a pool from a worker!')
82 }
8d3782fa 83 this.checkNumberOfWorkers(this.numberOfWorkers)
c510fea7 84 this.checkFilePath(this.filePath)
7c0ba920 85 this.checkPoolOptions(this.opts)
1086026a 86
7254e419
JB
87 this.chooseWorkerNode = this.chooseWorkerNode.bind(this)
88 this.executeTask = this.executeTask.bind(this)
89 this.enqueueTask = this.enqueueTask.bind(this)
90 this.checkAndEmitEvents = this.checkAndEmitEvents.bind(this)
1086026a 91
c97c7edb
S
92 this.setupHook()
93
5c5a1fb7 94 for (let i = 1; i <= this.numberOfWorkers; i++) {
280c2a77 95 this.createAndSetupWorker()
c97c7edb
S
96 }
97
6bd72cd0 98 if (this.opts.enableEvents === true) {
7c0ba920
JB
99 this.emitter = new PoolEmitter()
100 }
d59df138
JB
101 this.workerChoiceStrategyContext = new WorkerChoiceStrategyContext<
102 Worker,
103 Data,
104 Response
da309861
JB
105 >(
106 this,
107 this.opts.workerChoiceStrategy,
108 this.opts.workerChoiceStrategyOptions
109 )
c97c7edb
S
110 }
111
a35560ba 112 private checkFilePath (filePath: string): void {
ffcbbad8
JB
113 if (
114 filePath == null ||
115 (typeof filePath === 'string' && filePath.trim().length === 0)
116 ) {
c510fea7
APA
117 throw new Error('Please specify a file with a worker implementation')
118 }
119 }
120
8d3782fa
JB
121 private checkNumberOfWorkers (numberOfWorkers: number): void {
122 if (numberOfWorkers == null) {
123 throw new Error(
124 'Cannot instantiate a pool without specifying the number of workers'
125 )
78cea37e 126 } else if (!Number.isSafeInteger(numberOfWorkers)) {
473c717a 127 throw new TypeError(
8d3782fa
JB
128 'Cannot instantiate a pool with a non integer number of workers'
129 )
130 } else if (numberOfWorkers < 0) {
473c717a 131 throw new RangeError(
8d3782fa
JB
132 'Cannot instantiate a pool with a negative number of workers'
133 )
7c0ba920 134 } else if (this.type === PoolType.FIXED && numberOfWorkers === 0) {
8d3782fa
JB
135 throw new Error('Cannot instantiate a fixed pool with no worker')
136 }
137 }
138
7c0ba920 139 private checkPoolOptions (opts: PoolOptions<Worker>): void {
e843b904
JB
140 this.opts.workerChoiceStrategy =
141 opts.workerChoiceStrategy ?? WorkerChoiceStrategies.ROUND_ROBIN
aee46736 142 this.checkValidWorkerChoiceStrategy(this.opts.workerChoiceStrategy)
da309861 143 this.opts.workerChoiceStrategyOptions =
bbeadd16 144 opts.workerChoiceStrategyOptions ?? DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
aee46736 145 this.opts.enableEvents = opts.enableEvents ?? true
ff733df7 146 this.opts.enableTasksQueue = opts.enableTasksQueue ?? false
7171d33f 147 if (this.opts.enableTasksQueue) {
a20f0ba5
JB
148 this.checkValidTasksQueueOptions(
149 opts.tasksQueueOptions as TasksQueueOptions
150 )
151 this.opts.tasksQueueOptions = this.buildTasksQueueOptions(
152 opts.tasksQueueOptions as TasksQueueOptions
153 )
7171d33f 154 }
aee46736
JB
155 }
156
157 private checkValidWorkerChoiceStrategy (
158 workerChoiceStrategy: WorkerChoiceStrategy
159 ): void {
160 if (!Object.values(WorkerChoiceStrategies).includes(workerChoiceStrategy)) {
b529c323 161 throw new Error(
aee46736 162 `Invalid worker choice strategy '${workerChoiceStrategy}'`
b529c323
JB
163 )
164 }
7c0ba920
JB
165 }
166
a20f0ba5
JB
167 private checkValidTasksQueueOptions (
168 tasksQueueOptions: TasksQueueOptions
169 ): void {
170 if ((tasksQueueOptions?.concurrency as number) <= 0) {
171 throw new Error(
172 `Invalid worker tasks concurrency '${
173 tasksQueueOptions.concurrency as number
174 }'`
175 )
176 }
177 }
178
afc003b2 179 /** @inheritDoc */
7c0ba920
JB
180 public abstract get type (): PoolType
181
c2ade475 182 /**
ff733df7 183 * Number of tasks running in the pool.
c2ade475
JB
184 */
185 private get numberOfRunningTasks (): number {
ff733df7
JB
186 return this.workerNodes.reduce(
187 (accumulator, workerNode) => accumulator + workerNode.tasksUsage.running,
188 0
189 )
190 }
191
192 /**
193 * Number of tasks queued in the pool.
194 */
195 private get numberOfQueuedTasks (): number {
196 if (this.opts.enableTasksQueue === false) {
197 return 0
198 }
199 return this.workerNodes.reduce(
200 (accumulator, workerNode) => accumulator + workerNode.tasksQueue.length,
201 0
202 )
a35560ba
S
203 }
204
ffcbbad8 205 /**
f06e48d8 206 * Gets the given worker its worker node key.
ffcbbad8
JB
207 *
208 * @param worker - The worker.
f06e48d8 209 * @returns The worker node key if the worker is found in the pool worker nodes, `-1` otherwise.
ffcbbad8 210 */
f06e48d8
JB
211 private getWorkerNodeKey (worker: Worker): number {
212 return this.workerNodes.findIndex(
213 workerNode => workerNode.worker === worker
214 )
bf9549ae
JB
215 }
216
afc003b2 217 /** @inheritDoc */
a35560ba 218 public setWorkerChoiceStrategy (
59219cbb
JB
219 workerChoiceStrategy: WorkerChoiceStrategy,
220 workerChoiceStrategyOptions?: WorkerChoiceStrategyOptions
a35560ba 221 ): void {
aee46736 222 this.checkValidWorkerChoiceStrategy(workerChoiceStrategy)
b98ec2e6 223 this.opts.workerChoiceStrategy = workerChoiceStrategy
0ebe2a9f 224 for (const workerNode of this.workerNodes) {
f82cd357
JB
225 this.setWorkerNodeTasksUsage(workerNode, {
226 run: 0,
227 running: 0,
228 runTime: 0,
229 runTimeHistory: new CircularArray(),
230 avgRunTime: 0,
231 medRunTime: 0,
232 error: 0
233 })
ea7a90d3 234 }
a35560ba 235 this.workerChoiceStrategyContext.setWorkerChoiceStrategy(
a20f0ba5
JB
236 this.opts.workerChoiceStrategy
237 )
59219cbb
JB
238 if (workerChoiceStrategyOptions != null) {
239 this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions)
240 }
a20f0ba5
JB
241 }
242
243 /** @inheritDoc */
244 public setWorkerChoiceStrategyOptions (
245 workerChoiceStrategyOptions: WorkerChoiceStrategyOptions
246 ): void {
247 this.opts.workerChoiceStrategyOptions = workerChoiceStrategyOptions
248 this.workerChoiceStrategyContext.setOptions(
249 this.opts.workerChoiceStrategyOptions
a35560ba
S
250 )
251 }
252
a20f0ba5 253 /** @inheritDoc */
8f52842f
JB
254 public enableTasksQueue (
255 enable: boolean,
256 tasksQueueOptions?: TasksQueueOptions
257 ): void {
a20f0ba5 258 if (this.opts.enableTasksQueue === true && !enable) {
ef41a6e6 259 this.flushTasksQueues()
a20f0ba5
JB
260 }
261 this.opts.enableTasksQueue = enable
8f52842f 262 this.setTasksQueueOptions(tasksQueueOptions as TasksQueueOptions)
a20f0ba5
JB
263 }
264
265 /** @inheritDoc */
8f52842f 266 public setTasksQueueOptions (tasksQueueOptions: TasksQueueOptions): void {
a20f0ba5 267 if (this.opts.enableTasksQueue === true) {
8f52842f
JB
268 this.checkValidTasksQueueOptions(tasksQueueOptions)
269 this.opts.tasksQueueOptions =
270 this.buildTasksQueueOptions(tasksQueueOptions)
a20f0ba5
JB
271 } else {
272 delete this.opts.tasksQueueOptions
273 }
274 }
275
276 private buildTasksQueueOptions (
277 tasksQueueOptions: TasksQueueOptions
278 ): TasksQueueOptions {
279 return {
280 concurrency: tasksQueueOptions?.concurrency ?? 1
281 }
282 }
283
c319c66b
JB
284 /**
285 * Whether the pool is full or not.
286 *
287 * The pool filling boolean status.
288 */
289 protected abstract get full (): boolean
c2ade475 290
c319c66b
JB
291 /**
292 * Whether the pool is busy or not.
293 *
294 * The pool busyness boolean status.
295 */
296 protected abstract get busy (): boolean
7c0ba920 297
c2ade475 298 protected internalBusy (): boolean {
e0ae6100
JB
299 return (
300 this.workerNodes.findIndex(workerNode => {
301 return workerNode.tasksUsage?.running === 0
302 }) === -1
303 )
cb70b19d
JB
304 }
305
afc003b2 306 /** @inheritDoc */
ef41a6e6 307 public async execute (data?: Data): Promise<Response> {
adc3c320
JB
308 const [workerNodeKey, workerNode] = this.chooseWorkerNode()
309 const submittedTask: Task<Data> = {
e5a5c0fc
JB
310 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
311 data: data ?? ({} as Data),
adc3c320
JB
312 id: crypto.randomUUID()
313 }
2e81254d 314 const res = new Promise<Response>((resolve, reject) => {
02706357 315 this.promiseResponseMap.set(submittedTask.id as string, {
2e81254d
JB
316 resolve,
317 reject,
318 worker: workerNode.worker
319 })
320 })
ff733df7
JB
321 if (
322 this.opts.enableTasksQueue === true &&
7171d33f 323 (this.busy ||
3528c992 324 this.workerNodes[workerNodeKey].tasksUsage.running >=
7171d33f 325 ((this.opts.tasksQueueOptions as TasksQueueOptions)
3528c992 326 .concurrency as number))
ff733df7 327 ) {
26a929d7
JB
328 this.enqueueTask(workerNodeKey, submittedTask)
329 } else {
2e81254d 330 this.executeTask(workerNodeKey, submittedTask)
adc3c320 331 }
ff733df7 332 this.checkAndEmitEvents()
78cea37e 333 // eslint-disable-next-line @typescript-eslint/return-await
280c2a77
S
334 return res
335 }
c97c7edb 336
afc003b2 337 /** @inheritDoc */
c97c7edb 338 public async destroy (): Promise<void> {
1fbcaa7c 339 await Promise.all(
875a7c37
JB
340 this.workerNodes.map(async (workerNode, workerNodeKey) => {
341 this.flushTasksQueue(workerNodeKey)
f06e48d8 342 await this.destroyWorker(workerNode.worker)
1fbcaa7c
JB
343 })
344 )
c97c7edb
S
345 }
346
4a6952ff 347 /**
f06e48d8 348 * Shutdowns the given worker.
4a6952ff 349 *
f06e48d8 350 * @param worker - A worker within `workerNodes`.
4a6952ff
JB
351 */
352 protected abstract destroyWorker (worker: Worker): void | Promise<void>
c97c7edb 353
729c563d 354 /**
2e81254d 355 * Setup hook to execute code before worker node are created in the abstract constructor.
d99ba5a8 356 * Can be overridden
afc003b2
JB
357 *
358 * @virtual
729c563d 359 */
280c2a77 360 protected setupHook (): void {
d99ba5a8 361 // Intentionally empty
280c2a77 362 }
c97c7edb 363
729c563d 364 /**
280c2a77
S
365 * Should return whether the worker is the main worker or not.
366 */
367 protected abstract isMain (): boolean
368
369 /**
2e81254d 370 * Hook executed before the worker task execution.
bf9549ae 371 * Can be overridden.
729c563d 372 *
f06e48d8 373 * @param workerNodeKey - The worker node key.
729c563d 374 */
2e81254d 375 protected beforeTaskExecutionHook (workerNodeKey: number): void {
f06e48d8 376 ++this.workerNodes[workerNodeKey].tasksUsage.running
c97c7edb
S
377 }
378
c01733f1 379 /**
2e81254d 380 * Hook executed after the worker task execution.
bf9549ae 381 * Can be overridden.
c01733f1 382 *
c923ce56 383 * @param worker - The worker.
38e795c1 384 * @param message - The received message.
c01733f1 385 */
2e81254d 386 protected afterTaskExecutionHook (
c923ce56 387 worker: Worker,
2740a743 388 message: MessageValue<Response>
bf9549ae 389 ): void {
05ca3d0d 390 const workerTasksUsage = this.getWorkerTasksUsage(worker)
3032893a
JB
391 --workerTasksUsage.running
392 ++workerTasksUsage.run
2740a743
JB
393 if (message.error != null) {
394 ++workerTasksUsage.error
395 }
97a2abc3 396 if (this.workerChoiceStrategyContext.getRequiredStatistics().runTime) {
aee46736 397 workerTasksUsage.runTime += message.runTime ?? 0
c6bd2650
JB
398 if (
399 this.workerChoiceStrategyContext.getRequiredStatistics().avgRunTime &&
400 workerTasksUsage.run !== 0
401 ) {
3032893a
JB
402 workerTasksUsage.avgRunTime =
403 workerTasksUsage.runTime / workerTasksUsage.run
404 }
78099a15
JB
405 if (this.workerChoiceStrategyContext.getRequiredStatistics().medRunTime) {
406 workerTasksUsage.runTimeHistory.push(message.runTime ?? 0)
407 workerTasksUsage.medRunTime = median(workerTasksUsage.runTimeHistory)
408 }
3032893a 409 }
c01733f1 410 }
411
280c2a77 412 /**
f06e48d8 413 * Chooses a worker node for the next task.
280c2a77 414 *
51fe3d3c 415 * The default uses a round robin algorithm to distribute the load.
280c2a77 416 *
adc3c320 417 * @returns [worker node key, worker node].
280c2a77 418 */
adc3c320 419 protected chooseWorkerNode (): [number, WorkerNode<Worker, Data>] {
f06e48d8 420 let workerNodeKey: number
0527b6db 421 if (this.type === PoolType.DYNAMIC && !this.full && this.internalBusy()) {
adc3c320
JB
422 const workerCreated = this.createAndSetupWorker()
423 this.registerWorkerMessageListener(workerCreated, message => {
17393ac8
JB
424 if (
425 isKillBehavior(KillBehaviors.HARD, message.kill) ||
d2097c13 426 (message.kill != null &&
adc3c320 427 this.getWorkerTasksUsage(workerCreated)?.running === 0)
17393ac8 428 ) {
ff733df7
JB
429 // Kill message received from the worker: no new tasks are submitted to that worker for a while ( > maxInactiveTime)
430 this.flushTasksQueueByWorker(workerCreated)
7c5a1080 431 void (this.destroyWorker(workerCreated) as Promise<void>)
17393ac8
JB
432 }
433 })
adc3c320 434 workerNodeKey = this.getWorkerNodeKey(workerCreated)
17393ac8 435 } else {
f06e48d8 436 workerNodeKey = this.workerChoiceStrategyContext.execute()
17393ac8 437 }
adc3c320 438 return [workerNodeKey, this.workerNodes[workerNodeKey]]
c97c7edb
S
439 }
440
280c2a77 441 /**
675bb809 442 * Sends a message to the given worker.
280c2a77 443 *
38e795c1
JB
444 * @param worker - The worker which should receive the message.
445 * @param message - The message.
280c2a77
S
446 */
447 protected abstract sendToWorker (
448 worker: Worker,
449 message: MessageValue<Data>
450 ): void
451
4a6952ff 452 /**
f06e48d8 453 * Registers a listener callback on the given worker.
4a6952ff 454 *
38e795c1
JB
455 * @param worker - The worker which should register a listener.
456 * @param listener - The message listener callback.
4a6952ff
JB
457 */
458 protected abstract registerWorkerMessageListener<
4f7fa42a 459 Message extends Data | Response
78cea37e 460 >(worker: Worker, listener: (message: MessageValue<Message>) => void): void
c97c7edb 461
729c563d
S
462 /**
463 * Returns a newly created worker.
464 */
280c2a77 465 protected abstract createWorker (): Worker
c97c7edb 466
729c563d 467 /**
f06e48d8 468 * Function that can be hooked up when a worker has been newly created and moved to the pool worker nodes.
729c563d 469 *
38e795c1 470 * Can be used to update the `maxListeners` or binding the `main-worker`\<-\>`worker` connection if not bind by default.
729c563d 471 *
38e795c1 472 * @param worker - The newly created worker.
729c563d 473 */
280c2a77 474 protected abstract afterWorkerSetup (worker: Worker): void
c97c7edb 475
4a6952ff 476 /**
f06e48d8 477 * Creates a new worker and sets it up completely in the pool worker nodes.
4a6952ff
JB
478 *
479 * @returns New, completely set up worker.
480 */
481 protected createAndSetupWorker (): Worker {
bdacc2d2 482 const worker = this.createWorker()
280c2a77 483
35cf1c03 484 worker.on('message', this.opts.messageHandler ?? EMPTY_FUNCTION)
a35560ba
S
485 worker.on('error', this.opts.errorHandler ?? EMPTY_FUNCTION)
486 worker.on('online', this.opts.onlineHandler ?? EMPTY_FUNCTION)
487 worker.on('exit', this.opts.exitHandler ?? EMPTY_FUNCTION)
a974afa6 488 worker.once('exit', () => {
f06e48d8 489 this.removeWorkerNode(worker)
a974afa6 490 })
280c2a77 491
f06e48d8 492 this.pushWorkerNode(worker)
280c2a77
S
493
494 this.afterWorkerSetup(worker)
495
c97c7edb
S
496 return worker
497 }
be0676b3
APA
498
499 /**
ff733df7 500 * This function is the listener registered for each worker message.
be0676b3 501 *
bdacc2d2 502 * @returns The listener function to execute when a message is received from a worker.
be0676b3
APA
503 */
504 protected workerListener (): (message: MessageValue<Response>) => void {
4a6952ff 505 return message => {
b1989cfd 506 if (message.id != null) {
a3445496 507 // Task execution response received
2740a743 508 const promiseResponse = this.promiseResponseMap.get(message.id)
b1989cfd 509 if (promiseResponse != null) {
78cea37e 510 if (message.error != null) {
2740a743 511 promiseResponse.reject(message.error)
a05c10de 512 } else {
2740a743 513 promiseResponse.resolve(message.data as Response)
a05c10de 514 }
2e81254d 515 this.afterTaskExecutionHook(promiseResponse.worker, message)
2740a743 516 this.promiseResponseMap.delete(message.id)
ff733df7
JB
517 const workerNodeKey = this.getWorkerNodeKey(promiseResponse.worker)
518 if (
519 this.opts.enableTasksQueue === true &&
416fd65c 520 this.tasksQueueSize(workerNodeKey) > 0
ff733df7 521 ) {
2e81254d
JB
522 this.executeTask(
523 workerNodeKey,
ff733df7
JB
524 this.dequeueTask(workerNodeKey) as Task<Data>
525 )
526 }
be0676b3
APA
527 }
528 }
529 }
be0676b3 530 }
7c0ba920 531
ff733df7
JB
532 private checkAndEmitEvents (): void {
533 if (this.opts.enableEvents === true) {
534 if (this.busy) {
535 this.emitter?.emit(PoolEvents.busy)
536 }
537 if (this.type === PoolType.DYNAMIC && this.full) {
538 this.emitter?.emit(PoolEvents.full)
539 }
164d950a
JB
540 }
541 }
542
0ebe2a9f
JB
543 /**
544 * Sets the given worker node its tasks usage in the pool.
545 *
546 * @param workerNode - The worker node.
547 * @param tasksUsage - The worker node tasks usage.
548 */
549 private setWorkerNodeTasksUsage (
550 workerNode: WorkerNode<Worker, Data>,
551 tasksUsage: TasksUsage
552 ): void {
553 workerNode.tasksUsage = tasksUsage
554 }
555
c923ce56 556 /**
f06e48d8 557 * Gets the given worker its tasks usage in the pool.
c923ce56
JB
558 *
559 * @param worker - The worker.
e2473f60 560 * @throws Error if the worker is not found in the pool worker nodes.
c923ce56
JB
561 * @returns The worker tasks usage.
562 */
05ca3d0d 563 private getWorkerTasksUsage (worker: Worker): TasksUsage {
f06e48d8
JB
564 const workerNodeKey = this.getWorkerNodeKey(worker)
565 if (workerNodeKey !== -1) {
566 return this.workerNodes[workerNodeKey].tasksUsage
ffcbbad8 567 }
f06e48d8 568 throw new Error('Worker could not be found in the pool worker nodes')
a05c10de
JB
569 }
570
571 /**
f06e48d8 572 * Pushes the given worker in the pool worker nodes.
ea7a90d3 573 *
38e795c1 574 * @param worker - The worker.
f06e48d8 575 * @returns The worker nodes length.
ea7a90d3 576 */
f06e48d8
JB
577 private pushWorkerNode (worker: Worker): number {
578 return this.workerNodes.push({
ffcbbad8 579 worker,
f82cd357
JB
580 tasksUsage: {
581 run: 0,
582 running: 0,
583 runTime: 0,
584 runTimeHistory: new CircularArray(),
585 avgRunTime: 0,
586 medRunTime: 0,
587 error: 0
588 },
f06e48d8 589 tasksQueue: []
ea7a90d3
JB
590 })
591 }
c923ce56
JB
592
593 /**
f06e48d8 594 * Sets the given worker in the pool worker nodes.
c923ce56 595 *
f06e48d8 596 * @param workerNodeKey - The worker node key.
c923ce56
JB
597 * @param worker - The worker.
598 * @param tasksUsage - The worker tasks usage.
f06e48d8 599 * @param tasksQueue - The worker task queue.
c923ce56 600 */
f06e48d8
JB
601 private setWorkerNode (
602 workerNodeKey: number,
c923ce56 603 worker: Worker,
f06e48d8
JB
604 tasksUsage: TasksUsage,
605 tasksQueue: Array<Task<Data>>
c923ce56 606 ): void {
f06e48d8 607 this.workerNodes[workerNodeKey] = {
c923ce56 608 worker,
f06e48d8
JB
609 tasksUsage,
610 tasksQueue
c923ce56
JB
611 }
612 }
51fe3d3c
JB
613
614 /**
f06e48d8 615 * Removes the given worker from the pool worker nodes.
51fe3d3c 616 *
f06e48d8 617 * @param worker - The worker.
51fe3d3c 618 */
416fd65c 619 private removeWorkerNode (worker: Worker): void {
f06e48d8
JB
620 const workerNodeKey = this.getWorkerNodeKey(worker)
621 this.workerNodes.splice(workerNodeKey, 1)
622 this.workerChoiceStrategyContext.remove(workerNodeKey)
51fe3d3c 623 }
adc3c320 624
2e81254d
JB
625 private executeTask (workerNodeKey: number, task: Task<Data>): void {
626 this.beforeTaskExecutionHook(workerNodeKey)
627 this.sendToWorker(this.workerNodes[workerNodeKey].worker, task)
628 }
629
f9f00b5f
JB
630 private enqueueTask (workerNodeKey: number, task: Task<Data>): number {
631 return this.workerNodes[workerNodeKey].tasksQueue.push(task)
adc3c320
JB
632 }
633
416fd65c 634 private dequeueTask (workerNodeKey: number): Task<Data> | undefined {
adc3c320
JB
635 return this.workerNodes[workerNodeKey].tasksQueue.shift()
636 }
637
416fd65c 638 private tasksQueueSize (workerNodeKey: number): number {
adc3c320
JB
639 return this.workerNodes[workerNodeKey].tasksQueue.length
640 }
ff733df7 641
416fd65c
JB
642 private flushTasksQueue (workerNodeKey: number): void {
643 if (this.tasksQueueSize(workerNodeKey) > 0) {
ff733df7 644 for (const task of this.workerNodes[workerNodeKey].tasksQueue) {
2e81254d 645 this.executeTask(workerNodeKey, task)
ff733df7 646 }
ff733df7
JB
647 }
648 }
649
416fd65c 650 private flushTasksQueueByWorker (worker: Worker): void {
ff733df7
JB
651 const workerNodeKey = this.getWorkerNodeKey(worker)
652 this.flushTasksQueue(workerNodeKey)
653 }
ef41a6e6
JB
654
655 private flushTasksQueues (): void {
656 for (const [workerNodeKey] of this.workerNodes.entries()) {
657 this.flushTasksQueue(workerNodeKey)
658 }
659 }
c97c7edb 660}