perf: alternate worker selection between start and end of worker nodes
[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'
c4855468
JB
9import {
10 PoolEvents,
11 type IPool,
12 type PoolOptions,
13 type TasksQueueOptions,
14 PoolType
15} from './pool'
b4904890 16import { PoolEmitter } from './pool'
f06e48d8 17import type { IWorker, Task, TasksUsage, WorkerNode } from './worker'
a35560ba
S
18import {
19 WorkerChoiceStrategies,
a20f0ba5
JB
20 type WorkerChoiceStrategy,
21 type WorkerChoiceStrategyOptions
bdaf31cd
JB
22} from './selection-strategies/selection-strategies-types'
23import { WorkerChoiceStrategyContext } from './selection-strategies/worker-choice-strategy-context'
f82cd357 24import { CircularArray } from '../circular-array'
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 {
24990079 299 return this.findFreeWorkerNodeKey() === -1
7c0ba920
JB
300 }
301
afc003b2 302 /** @inheritDoc */
f06e48d8
JB
303 public findFreeWorkerNodeKey (): number {
304 return this.workerNodes.findIndex(workerNode => {
305 return workerNode.tasksUsage?.running === 0
c923ce56 306 })
7c0ba920
JB
307 }
308
cb70b19d
JB
309 /** @inheritDoc */
310 public findLastFreeWorkerNodeKey (): number {
311 // It requires node >= 18.0.0
312 // return this.workerNodes.findLastIndex(workerNode => {
313 // return workerNode.tasksUsage?.running === 0
314 // })
315 for (let i = this.workerNodes.length - 1; i >= 0; i--) {
316 if (this.workerNodes[i].tasksUsage?.running === 0) {
317 return i
318 }
319 }
320 return -1
321 }
322
afc003b2 323 /** @inheritDoc */
ef41a6e6 324 public async execute (data?: Data): Promise<Response> {
adc3c320
JB
325 const [workerNodeKey, workerNode] = this.chooseWorkerNode()
326 const submittedTask: Task<Data> = {
e5a5c0fc
JB
327 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
328 data: data ?? ({} as Data),
adc3c320
JB
329 id: crypto.randomUUID()
330 }
2e81254d 331 const res = new Promise<Response>((resolve, reject) => {
02706357 332 this.promiseResponseMap.set(submittedTask.id as string, {
2e81254d
JB
333 resolve,
334 reject,
335 worker: workerNode.worker
336 })
337 })
ff733df7
JB
338 if (
339 this.opts.enableTasksQueue === true &&
7171d33f 340 (this.busy ||
3528c992 341 this.workerNodes[workerNodeKey].tasksUsage.running >=
7171d33f 342 ((this.opts.tasksQueueOptions as TasksQueueOptions)
3528c992 343 .concurrency as number))
ff733df7 344 ) {
26a929d7
JB
345 this.enqueueTask(workerNodeKey, submittedTask)
346 } else {
2e81254d 347 this.executeTask(workerNodeKey, submittedTask)
adc3c320 348 }
ff733df7 349 this.checkAndEmitEvents()
78cea37e 350 // eslint-disable-next-line @typescript-eslint/return-await
280c2a77
S
351 return res
352 }
c97c7edb 353
afc003b2 354 /** @inheritDoc */
c97c7edb 355 public async destroy (): Promise<void> {
1fbcaa7c 356 await Promise.all(
875a7c37
JB
357 this.workerNodes.map(async (workerNode, workerNodeKey) => {
358 this.flushTasksQueue(workerNodeKey)
f06e48d8 359 await this.destroyWorker(workerNode.worker)
1fbcaa7c
JB
360 })
361 )
c97c7edb
S
362 }
363
4a6952ff 364 /**
f06e48d8 365 * Shutdowns the given worker.
4a6952ff 366 *
f06e48d8 367 * @param worker - A worker within `workerNodes`.
4a6952ff
JB
368 */
369 protected abstract destroyWorker (worker: Worker): void | Promise<void>
c97c7edb 370
729c563d 371 /**
2e81254d 372 * Setup hook to execute code before worker node are created in the abstract constructor.
d99ba5a8 373 * Can be overridden
afc003b2
JB
374 *
375 * @virtual
729c563d 376 */
280c2a77 377 protected setupHook (): void {
d99ba5a8 378 // Intentionally empty
280c2a77 379 }
c97c7edb 380
729c563d 381 /**
280c2a77
S
382 * Should return whether the worker is the main worker or not.
383 */
384 protected abstract isMain (): boolean
385
386 /**
2e81254d 387 * Hook executed before the worker task execution.
bf9549ae 388 * Can be overridden.
729c563d 389 *
f06e48d8 390 * @param workerNodeKey - The worker node key.
729c563d 391 */
2e81254d 392 protected beforeTaskExecutionHook (workerNodeKey: number): void {
f06e48d8 393 ++this.workerNodes[workerNodeKey].tasksUsage.running
c97c7edb
S
394 }
395
c01733f1 396 /**
2e81254d 397 * Hook executed after the worker task execution.
bf9549ae 398 * Can be overridden.
c01733f1 399 *
c923ce56 400 * @param worker - The worker.
38e795c1 401 * @param message - The received message.
c01733f1 402 */
2e81254d 403 protected afterTaskExecutionHook (
c923ce56 404 worker: Worker,
2740a743 405 message: MessageValue<Response>
bf9549ae 406 ): void {
c923ce56 407 const workerTasksUsage = this.getWorkerTasksUsage(worker) as TasksUsage
3032893a
JB
408 --workerTasksUsage.running
409 ++workerTasksUsage.run
2740a743
JB
410 if (message.error != null) {
411 ++workerTasksUsage.error
412 }
97a2abc3 413 if (this.workerChoiceStrategyContext.getRequiredStatistics().runTime) {
aee46736 414 workerTasksUsage.runTime += message.runTime ?? 0
c6bd2650
JB
415 if (
416 this.workerChoiceStrategyContext.getRequiredStatistics().avgRunTime &&
417 workerTasksUsage.run !== 0
418 ) {
3032893a
JB
419 workerTasksUsage.avgRunTime =
420 workerTasksUsage.runTime / workerTasksUsage.run
421 }
78099a15
JB
422 if (this.workerChoiceStrategyContext.getRequiredStatistics().medRunTime) {
423 workerTasksUsage.runTimeHistory.push(message.runTime ?? 0)
424 workerTasksUsage.medRunTime = median(workerTasksUsage.runTimeHistory)
425 }
3032893a 426 }
c01733f1 427 }
428
280c2a77 429 /**
f06e48d8 430 * Chooses a worker node for the next task.
280c2a77 431 *
51fe3d3c 432 * The default uses a round robin algorithm to distribute the load.
280c2a77 433 *
adc3c320 434 * @returns [worker node key, worker node].
280c2a77 435 */
adc3c320 436 protected chooseWorkerNode (): [number, WorkerNode<Worker, Data>] {
f06e48d8 437 let workerNodeKey: number
0527b6db 438 if (this.type === PoolType.DYNAMIC && !this.full && this.internalBusy()) {
adc3c320
JB
439 const workerCreated = this.createAndSetupWorker()
440 this.registerWorkerMessageListener(workerCreated, message => {
17393ac8
JB
441 if (
442 isKillBehavior(KillBehaviors.HARD, message.kill) ||
d2097c13 443 (message.kill != null &&
adc3c320 444 this.getWorkerTasksUsage(workerCreated)?.running === 0)
17393ac8 445 ) {
ff733df7
JB
446 // Kill message received from the worker: no new tasks are submitted to that worker for a while ( > maxInactiveTime)
447 this.flushTasksQueueByWorker(workerCreated)
adc3c320 448 void this.destroyWorker(workerCreated)
17393ac8
JB
449 }
450 })
adc3c320 451 workerNodeKey = this.getWorkerNodeKey(workerCreated)
17393ac8 452 } else {
f06e48d8 453 workerNodeKey = this.workerChoiceStrategyContext.execute()
17393ac8 454 }
adc3c320 455 return [workerNodeKey, this.workerNodes[workerNodeKey]]
c97c7edb
S
456 }
457
280c2a77 458 /**
675bb809 459 * Sends a message to the given worker.
280c2a77 460 *
38e795c1
JB
461 * @param worker - The worker which should receive the message.
462 * @param message - The message.
280c2a77
S
463 */
464 protected abstract sendToWorker (
465 worker: Worker,
466 message: MessageValue<Data>
467 ): void
468
4a6952ff 469 /**
f06e48d8 470 * Registers a listener callback on the given worker.
4a6952ff 471 *
38e795c1
JB
472 * @param worker - The worker which should register a listener.
473 * @param listener - The message listener callback.
4a6952ff
JB
474 */
475 protected abstract registerWorkerMessageListener<
4f7fa42a 476 Message extends Data | Response
78cea37e 477 >(worker: Worker, listener: (message: MessageValue<Message>) => void): void
c97c7edb 478
729c563d
S
479 /**
480 * Returns a newly created worker.
481 */
280c2a77 482 protected abstract createWorker (): Worker
c97c7edb 483
729c563d 484 /**
f06e48d8 485 * Function that can be hooked up when a worker has been newly created and moved to the pool worker nodes.
729c563d 486 *
38e795c1 487 * Can be used to update the `maxListeners` or binding the `main-worker`\<-\>`worker` connection if not bind by default.
729c563d 488 *
38e795c1 489 * @param worker - The newly created worker.
729c563d 490 */
280c2a77 491 protected abstract afterWorkerSetup (worker: Worker): void
c97c7edb 492
4a6952ff 493 /**
f06e48d8 494 * Creates a new worker and sets it up completely in the pool worker nodes.
4a6952ff
JB
495 *
496 * @returns New, completely set up worker.
497 */
498 protected createAndSetupWorker (): Worker {
bdacc2d2 499 const worker = this.createWorker()
280c2a77 500
35cf1c03 501 worker.on('message', this.opts.messageHandler ?? EMPTY_FUNCTION)
a35560ba
S
502 worker.on('error', this.opts.errorHandler ?? EMPTY_FUNCTION)
503 worker.on('online', this.opts.onlineHandler ?? EMPTY_FUNCTION)
504 worker.on('exit', this.opts.exitHandler ?? EMPTY_FUNCTION)
a974afa6 505 worker.once('exit', () => {
f06e48d8 506 this.removeWorkerNode(worker)
a974afa6 507 })
280c2a77 508
f06e48d8 509 this.pushWorkerNode(worker)
280c2a77
S
510
511 this.afterWorkerSetup(worker)
512
c97c7edb
S
513 return worker
514 }
be0676b3
APA
515
516 /**
ff733df7 517 * This function is the listener registered for each worker message.
be0676b3 518 *
bdacc2d2 519 * @returns The listener function to execute when a message is received from a worker.
be0676b3
APA
520 */
521 protected workerListener (): (message: MessageValue<Response>) => void {
4a6952ff 522 return message => {
b1989cfd 523 if (message.id != null) {
a3445496 524 // Task execution response received
2740a743 525 const promiseResponse = this.promiseResponseMap.get(message.id)
b1989cfd 526 if (promiseResponse != null) {
78cea37e 527 if (message.error != null) {
2740a743 528 promiseResponse.reject(message.error)
a05c10de 529 } else {
2740a743 530 promiseResponse.resolve(message.data as Response)
a05c10de 531 }
2e81254d 532 this.afterTaskExecutionHook(promiseResponse.worker, message)
2740a743 533 this.promiseResponseMap.delete(message.id)
ff733df7
JB
534 const workerNodeKey = this.getWorkerNodeKey(promiseResponse.worker)
535 if (
536 this.opts.enableTasksQueue === true &&
416fd65c 537 this.tasksQueueSize(workerNodeKey) > 0
ff733df7 538 ) {
2e81254d
JB
539 this.executeTask(
540 workerNodeKey,
ff733df7
JB
541 this.dequeueTask(workerNodeKey) as Task<Data>
542 )
543 }
be0676b3
APA
544 }
545 }
546 }
be0676b3 547 }
7c0ba920 548
ff733df7
JB
549 private checkAndEmitEvents (): void {
550 if (this.opts.enableEvents === true) {
551 if (this.busy) {
552 this.emitter?.emit(PoolEvents.busy)
553 }
554 if (this.type === PoolType.DYNAMIC && this.full) {
555 this.emitter?.emit(PoolEvents.full)
556 }
164d950a
JB
557 }
558 }
559
0ebe2a9f
JB
560 /**
561 * Sets the given worker node its tasks usage in the pool.
562 *
563 * @param workerNode - The worker node.
564 * @param tasksUsage - The worker node tasks usage.
565 */
566 private setWorkerNodeTasksUsage (
567 workerNode: WorkerNode<Worker, Data>,
568 tasksUsage: TasksUsage
569 ): void {
570 workerNode.tasksUsage = tasksUsage
571 }
572
c923ce56 573 /**
f06e48d8 574 * Gets the given worker its tasks usage in the pool.
c923ce56
JB
575 *
576 * @param worker - The worker.
e2473f60 577 * @throws Error if the worker is not found in the pool worker nodes.
c923ce56
JB
578 * @returns The worker tasks usage.
579 */
580 private getWorkerTasksUsage (worker: Worker): TasksUsage | undefined {
f06e48d8
JB
581 const workerNodeKey = this.getWorkerNodeKey(worker)
582 if (workerNodeKey !== -1) {
583 return this.workerNodes[workerNodeKey].tasksUsage
ffcbbad8 584 }
f06e48d8 585 throw new Error('Worker could not be found in the pool worker nodes')
a05c10de
JB
586 }
587
588 /**
f06e48d8 589 * Pushes the given worker in the pool worker nodes.
ea7a90d3 590 *
38e795c1 591 * @param worker - The worker.
f06e48d8 592 * @returns The worker nodes length.
ea7a90d3 593 */
f06e48d8
JB
594 private pushWorkerNode (worker: Worker): number {
595 return this.workerNodes.push({
ffcbbad8 596 worker,
f82cd357
JB
597 tasksUsage: {
598 run: 0,
599 running: 0,
600 runTime: 0,
601 runTimeHistory: new CircularArray(),
602 avgRunTime: 0,
603 medRunTime: 0,
604 error: 0
605 },
f06e48d8 606 tasksQueue: []
ea7a90d3
JB
607 })
608 }
c923ce56
JB
609
610 /**
f06e48d8 611 * Sets the given worker in the pool worker nodes.
c923ce56 612 *
f06e48d8 613 * @param workerNodeKey - The worker node key.
c923ce56
JB
614 * @param worker - The worker.
615 * @param tasksUsage - The worker tasks usage.
f06e48d8 616 * @param tasksQueue - The worker task queue.
c923ce56 617 */
f06e48d8
JB
618 private setWorkerNode (
619 workerNodeKey: number,
c923ce56 620 worker: Worker,
f06e48d8
JB
621 tasksUsage: TasksUsage,
622 tasksQueue: Array<Task<Data>>
c923ce56 623 ): void {
f06e48d8 624 this.workerNodes[workerNodeKey] = {
c923ce56 625 worker,
f06e48d8
JB
626 tasksUsage,
627 tasksQueue
c923ce56
JB
628 }
629 }
51fe3d3c
JB
630
631 /**
f06e48d8 632 * Removes the given worker from the pool worker nodes.
51fe3d3c 633 *
f06e48d8 634 * @param worker - The worker.
51fe3d3c 635 */
416fd65c 636 private removeWorkerNode (worker: Worker): void {
f06e48d8
JB
637 const workerNodeKey = this.getWorkerNodeKey(worker)
638 this.workerNodes.splice(workerNodeKey, 1)
639 this.workerChoiceStrategyContext.remove(workerNodeKey)
51fe3d3c 640 }
adc3c320 641
2e81254d
JB
642 private executeTask (workerNodeKey: number, task: Task<Data>): void {
643 this.beforeTaskExecutionHook(workerNodeKey)
644 this.sendToWorker(this.workerNodes[workerNodeKey].worker, task)
645 }
646
f9f00b5f
JB
647 private enqueueTask (workerNodeKey: number, task: Task<Data>): number {
648 return this.workerNodes[workerNodeKey].tasksQueue.push(task)
adc3c320
JB
649 }
650
416fd65c 651 private dequeueTask (workerNodeKey: number): Task<Data> | undefined {
adc3c320
JB
652 return this.workerNodes[workerNodeKey].tasksQueue.shift()
653 }
654
416fd65c 655 private tasksQueueSize (workerNodeKey: number): number {
adc3c320
JB
656 return this.workerNodes[workerNodeKey].tasksQueue.length
657 }
ff733df7 658
416fd65c
JB
659 private flushTasksQueue (workerNodeKey: number): void {
660 if (this.tasksQueueSize(workerNodeKey) > 0) {
ff733df7 661 for (const task of this.workerNodes[workerNodeKey].tasksQueue) {
2e81254d 662 this.executeTask(workerNodeKey, task)
ff733df7 663 }
ff733df7
JB
664 }
665 }
666
416fd65c 667 private flushTasksQueueByWorker (worker: Worker): void {
ff733df7
JB
668 const workerNodeKey = this.getWorkerNodeKey(worker)
669 this.flushTasksQueue(workerNodeKey)
670 }
ef41a6e6
JB
671
672 private flushTasksQueues (): void {
673 for (const [workerNodeKey] of this.workerNodes.entries()) {
674 this.flushTasksQueue(workerNodeKey)
675 }
676 }
c97c7edb 677}