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