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