1 import crypto from
'node:crypto'
2 import type { MessageValue
, PromiseResponseWrapper
} from
'../utility-types'
4 DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
,
8 import { KillBehaviors
, isKillBehavior
} from
'../worker/worker-options'
13 type TasksQueueOptions
,
16 import { PoolEmitter
} from
'./pool'
17 import type { IWorker
, Task
, TasksUsage
, WorkerNode
} from
'./worker'
19 WorkerChoiceStrategies
,
20 type WorkerChoiceStrategy
,
21 type WorkerChoiceStrategyOptions
22 } from
'./selection-strategies/selection-strategies-types'
23 import { WorkerChoiceStrategyContext
} from
'./selection-strategies/worker-choice-strategy-context'
24 import { CircularArray
} from
'../circular-array'
27 * Base class that implements some shared logic for all poolifier pools.
29 * @typeParam Worker - Type of worker which manages this pool.
30 * @typeParam Data - Type of data sent to the worker. This can only be serializable data.
31 * @typeParam Response - Type of execution response. This can only be serializable data.
33 export abstract class AbstractPool
<
34 Worker
extends IWorker
,
37 > implements IPool
<Worker
, Data
, Response
> {
39 public readonly workerNodes
: Array<WorkerNode
<Worker
, Data
>> = []
42 public readonly emitter
?: PoolEmitter
45 * The execution response promise map.
47 * - `key`: The message id of each submitted task.
48 * - `value`: An object that contains the worker, the execution response promise resolve and reject callbacks.
50 * When we receive a message from the worker, we get a map entry with the promise resolve/reject bound to the message id.
52 protected promiseResponseMap
: Map
<
54 PromiseResponseWrapper
<Worker
, Response
>
55 > = new Map
<string, PromiseResponseWrapper
<Worker
, Response
>>()
58 * Worker choice strategy context referencing a worker choice algorithm implementation.
60 * Default to a round robin algorithm.
62 protected workerChoiceStrategyContext
: WorkerChoiceStrategyContext
<
69 * Constructs a new poolifier pool.
71 * @param numberOfWorkers - Number of workers that this pool should manage.
72 * @param filePath - Path to the worker-file.
73 * @param opts - Options for the pool.
76 public readonly numberOfWorkers
: number,
77 public readonly filePath
: string,
78 public readonly opts
: PoolOptions
<Worker
>
81 throw new Error('Cannot start a pool from a worker!')
83 this.checkNumberOfWorkers(this.numberOfWorkers
)
84 this.checkFilePath(this.filePath
)
85 this.checkPoolOptions(this.opts
)
87 this.chooseWorkerNode
.bind(this)
88 this.executeTask
.bind(this)
89 this.enqueueTask
.bind(this)
90 this.checkAndEmitEvents
.bind(this)
94 for (let i
= 1; i
<= this.numberOfWorkers
; i
++) {
95 this.createAndSetupWorker()
98 if (this.opts
.enableEvents
=== true) {
99 this.emitter
= new PoolEmitter()
101 this.workerChoiceStrategyContext
= new WorkerChoiceStrategyContext
<
107 this.opts
.workerChoiceStrategy
,
108 this.opts
.workerChoiceStrategyOptions
112 private checkFilePath (filePath
: string): void {
115 (typeof filePath
=== 'string' && filePath
.trim().length
=== 0)
117 throw new Error('Please specify a file with a worker implementation')
121 private checkNumberOfWorkers (numberOfWorkers
: number): void {
122 if (numberOfWorkers
== null) {
124 'Cannot instantiate a pool without specifying the number of workers'
126 } else if (!Number.isSafeInteger(numberOfWorkers
)) {
128 'Cannot instantiate a pool with a non integer number of workers'
130 } else if (numberOfWorkers
< 0) {
131 throw new RangeError(
132 'Cannot instantiate a pool with a negative number of workers'
134 } else if (this.type === PoolType
.FIXED
&& numberOfWorkers
=== 0) {
135 throw new Error('Cannot instantiate a fixed pool with no worker')
139 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
140 this.opts
.workerChoiceStrategy
=
141 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
142 this.checkValidWorkerChoiceStrategy(this.opts
.workerChoiceStrategy
)
143 this.opts
.workerChoiceStrategyOptions
=
144 opts
.workerChoiceStrategyOptions
?? DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
145 this.opts
.enableEvents
= opts
.enableEvents
?? true
146 this.opts
.enableTasksQueue
= opts
.enableTasksQueue
?? false
147 if (this.opts
.enableTasksQueue
) {
148 this.checkValidTasksQueueOptions(
149 opts
.tasksQueueOptions
as TasksQueueOptions
151 this.opts
.tasksQueueOptions
= this.buildTasksQueueOptions(
152 opts
.tasksQueueOptions
as TasksQueueOptions
157 private checkValidWorkerChoiceStrategy (
158 workerChoiceStrategy
: WorkerChoiceStrategy
160 if (!Object.values(WorkerChoiceStrategies
).includes(workerChoiceStrategy
)) {
162 `Invalid worker choice strategy '${workerChoiceStrategy}'`
167 private checkValidTasksQueueOptions (
168 tasksQueueOptions
: TasksQueueOptions
170 if ((tasksQueueOptions
?.concurrency
as number) <= 0) {
172 `Invalid worker tasks concurrency '${
173 tasksQueueOptions.concurrency as number
180 public abstract get
type (): PoolType
183 * Number of tasks running in the pool.
185 private get
numberOfRunningTasks (): number {
186 return this.workerNodes
.reduce(
187 (accumulator
, workerNode
) => accumulator
+ workerNode
.tasksUsage
.running
,
193 * Number of tasks queued in the pool.
195 private get
numberOfQueuedTasks (): number {
196 if (this.opts
.enableTasksQueue
=== false) {
199 return this.workerNodes
.reduce(
200 (accumulator
, workerNode
) => accumulator
+ workerNode
.tasksQueue
.length
,
206 * Gets the given worker its worker node key.
208 * @param worker - The worker.
209 * @returns The worker node key if the worker is found in the pool worker nodes, `-1` otherwise.
211 private getWorkerNodeKey (worker
: Worker
): number {
212 return this.workerNodes
.findIndex(
213 workerNode
=> workerNode
.worker
=== worker
218 public setWorkerChoiceStrategy (
219 workerChoiceStrategy
: WorkerChoiceStrategy
221 this.checkValidWorkerChoiceStrategy(workerChoiceStrategy
)
222 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
223 for (const workerNode
of this.workerNodes
) {
224 this.setWorkerNodeTasksUsage(workerNode
, {
228 runTimeHistory
: new CircularArray(),
234 this.workerChoiceStrategyContext
.setWorkerChoiceStrategy(
235 this.opts
.workerChoiceStrategy
240 public setWorkerChoiceStrategyOptions (
241 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
243 this.opts
.workerChoiceStrategyOptions
= workerChoiceStrategyOptions
244 this.workerChoiceStrategyContext
.setOptions(
245 this.opts
.workerChoiceStrategyOptions
250 public enableTasksQueue (enable
: boolean, opts
?: TasksQueueOptions
): void {
251 if (this.opts
.enableTasksQueue
=== true && !enable
) {
252 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
253 this.flushTasksQueue(workerNodeKey
)
256 this.opts
.enableTasksQueue
= enable
257 this.setTasksQueueOptions(opts
as TasksQueueOptions
)
261 public setTasksQueueOptions (opts
: TasksQueueOptions
): void {
262 if (this.opts
.enableTasksQueue
=== true) {
263 this.checkValidTasksQueueOptions(opts
)
264 this.opts
.tasksQueueOptions
= this.buildTasksQueueOptions(opts
)
266 delete this.opts
.tasksQueueOptions
270 private buildTasksQueueOptions (
271 tasksQueueOptions
: TasksQueueOptions
272 ): TasksQueueOptions
{
274 concurrency
: tasksQueueOptions
?.concurrency
?? 1
279 * Whether the pool is full or not.
281 * The pool filling boolean status.
283 protected abstract get
full (): boolean
286 * Whether the pool is busy or not.
288 * The pool busyness boolean status.
290 protected abstract get
busy (): boolean
292 protected internalBusy (): boolean {
293 return this.findFreeWorkerNodeKey() === -1
297 public findFreeWorkerNodeKey (): number {
298 return this.workerNodes
.findIndex(workerNode
=> {
299 return workerNode
.tasksUsage
?.running
=== 0
304 public async execute (data
: Data
): Promise
<Response
> {
305 const [workerNodeKey
, workerNode
] = this.chooseWorkerNode()
306 const submittedTask
: Task
<Data
> = {
307 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
308 data
: data
?? ({} as Data
),
309 id
: crypto
.randomUUID()
311 const res
= new Promise
<Response
>((resolve
, reject
) => {
312 this.promiseResponseMap
.set(submittedTask
.id
as string, {
315 worker
: workerNode
.worker
319 this.opts
.enableTasksQueue
=== true &&
321 this.workerNodes
[workerNodeKey
].tasksUsage
.running
>=
322 ((this.opts
.tasksQueueOptions
as TasksQueueOptions
)
323 .concurrency
as number))
325 this.enqueueTask(workerNodeKey
, submittedTask
)
327 this.executeTask(workerNodeKey
, submittedTask
)
329 this.checkAndEmitEvents()
330 // eslint-disable-next-line @typescript-eslint/return-await
335 public async destroy (): Promise
<void> {
337 this.workerNodes
.map(async (workerNode
, workerNodeKey
) => {
338 this.flushTasksQueue(workerNodeKey
)
339 await this.destroyWorker(workerNode
.worker
)
345 * Shutdowns the given worker.
347 * @param worker - A worker within `workerNodes`.
349 protected abstract destroyWorker (worker
: Worker
): void | Promise
<void>
352 * Setup hook to execute code before worker node are created in the abstract constructor.
357 protected setupHook (): void {
358 // Intentionally empty
362 * Should return whether the worker is the main worker or not.
364 protected abstract isMain (): boolean
367 * Hook executed before the worker task execution.
370 * @param workerNodeKey - The worker node key.
372 protected beforeTaskExecutionHook (workerNodeKey
: number): void {
373 ++this.workerNodes
[workerNodeKey
].tasksUsage
.running
377 * Hook executed after the worker task execution.
380 * @param worker - The worker.
381 * @param message - The received message.
383 protected afterTaskExecutionHook (
385 message
: MessageValue
<Response
>
387 const workerTasksUsage
= this.getWorkerTasksUsage(worker
) as TasksUsage
388 --workerTasksUsage
.running
389 ++workerTasksUsage
.run
390 if (message
.error
!= null) {
391 ++workerTasksUsage
.error
393 if (this.workerChoiceStrategyContext
.getRequiredStatistics().runTime
) {
394 workerTasksUsage
.runTime
+= message
.runTime
?? 0
396 this.workerChoiceStrategyContext
.getRequiredStatistics().avgRunTime
&&
397 workerTasksUsage
.run
!== 0
399 workerTasksUsage
.avgRunTime
=
400 workerTasksUsage
.runTime
/ workerTasksUsage
.run
402 if (this.workerChoiceStrategyContext
.getRequiredStatistics().medRunTime
) {
403 workerTasksUsage
.runTimeHistory
.push(message
.runTime
?? 0)
404 workerTasksUsage
.medRunTime
= median(workerTasksUsage
.runTimeHistory
)
410 * Chooses a worker node for the next task.
412 * The default uses a round robin algorithm to distribute the load.
414 * @returns [worker node key, worker node].
416 protected chooseWorkerNode (): [number, WorkerNode
<Worker
, Data
>] {
417 let workerNodeKey
: number
418 if (this.type === PoolType
.DYNAMIC
&& !this.full
&& this.internalBusy()) {
419 const workerCreated
= this.createAndSetupWorker()
420 this.registerWorkerMessageListener(workerCreated
, message
=> {
422 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
423 (message
.kill
!= null &&
424 this.getWorkerTasksUsage(workerCreated
)?.running
=== 0)
426 // Kill message received from the worker: no new tasks are submitted to that worker for a while ( > maxInactiveTime)
427 this.flushTasksQueueByWorker(workerCreated
)
428 void this.destroyWorker(workerCreated
)
431 workerNodeKey
= this.getWorkerNodeKey(workerCreated
)
433 workerNodeKey
= this.workerChoiceStrategyContext
.execute()
435 return [workerNodeKey
, this.workerNodes
[workerNodeKey
]]
439 * Sends a message to the given worker.
441 * @param worker - The worker which should receive the message.
442 * @param message - The message.
444 protected abstract sendToWorker (
446 message
: MessageValue
<Data
>
450 * Registers a listener callback on the given worker.
452 * @param worker - The worker which should register a listener.
453 * @param listener - The message listener callback.
455 protected abstract registerWorkerMessageListener
<
456 Message
extends Data
| Response
457 >(worker
: Worker
, listener
: (message
: MessageValue
<Message
>) => void): void
460 * Returns a newly created worker.
462 protected abstract createWorker (): Worker
465 * Function that can be hooked up when a worker has been newly created and moved to the pool worker nodes.
467 * Can be used to update the `maxListeners` or binding the `main-worker`\<-\>`worker` connection if not bind by default.
469 * @param worker - The newly created worker.
471 protected abstract afterWorkerSetup (worker
: Worker
): void
474 * Creates a new worker and sets it up completely in the pool worker nodes.
476 * @returns New, completely set up worker.
478 protected createAndSetupWorker (): Worker
{
479 const worker
= this.createWorker()
481 worker
.on('message', this.opts
.messageHandler
?? EMPTY_FUNCTION
)
482 worker
.on('error', this.opts
.errorHandler
?? EMPTY_FUNCTION
)
483 worker
.on('online', this.opts
.onlineHandler
?? EMPTY_FUNCTION
)
484 worker
.on('exit', this.opts
.exitHandler
?? EMPTY_FUNCTION
)
485 worker
.once('exit', () => {
486 this.removeWorkerNode(worker
)
489 this.pushWorkerNode(worker
)
491 this.afterWorkerSetup(worker
)
497 * This function is the listener registered for each worker message.
499 * @returns The listener function to execute when a message is received from a worker.
501 protected workerListener (): (message
: MessageValue
<Response
>) => void {
503 if (message
.id
!= null) {
504 // Task execution response received
505 const promiseResponse
= this.promiseResponseMap
.get(message
.id
)
506 if (promiseResponse
!= null) {
507 if (message
.error
!= null) {
508 promiseResponse
.reject(message
.error
)
510 promiseResponse
.resolve(message
.data
as Response
)
512 this.afterTaskExecutionHook(promiseResponse
.worker
, message
)
513 this.promiseResponseMap
.delete(message
.id
)
514 const workerNodeKey
= this.getWorkerNodeKey(promiseResponse
.worker
)
516 this.opts
.enableTasksQueue
=== true &&
517 this.tasksQueueSize(workerNodeKey
) > 0
521 this.dequeueTask(workerNodeKey
) as Task
<Data
>
529 private checkAndEmitEvents (): void {
530 if (this.opts
.enableEvents
=== true) {
532 this.emitter
?.emit(PoolEvents
.busy
)
534 if (this.type === PoolType
.DYNAMIC
&& this.full
) {
535 this.emitter
?.emit(PoolEvents
.full
)
541 * Sets the given worker node its tasks usage in the pool.
543 * @param workerNode - The worker node.
544 * @param tasksUsage - The worker node tasks usage.
546 private setWorkerNodeTasksUsage (
547 workerNode
: WorkerNode
<Worker
, Data
>,
548 tasksUsage
: TasksUsage
550 workerNode
.tasksUsage
= tasksUsage
554 * Gets the given worker its tasks usage in the pool.
556 * @param worker - The worker.
557 * @throws {@link Error} if the worker is not found in the pool worker nodes.
558 * @returns The worker tasks usage.
560 private getWorkerTasksUsage (worker
: Worker
): TasksUsage
| undefined {
561 const workerNodeKey
= this.getWorkerNodeKey(worker
)
562 if (workerNodeKey
!== -1) {
563 return this.workerNodes
[workerNodeKey
].tasksUsage
565 throw new Error('Worker could not be found in the pool worker nodes')
569 * Pushes the given worker in the pool worker nodes.
571 * @param worker - The worker.
572 * @returns The worker nodes length.
574 private pushWorkerNode (worker
: Worker
): number {
575 return this.workerNodes
.push({
581 runTimeHistory
: new CircularArray(),
591 * Sets the given worker in the pool worker nodes.
593 * @param workerNodeKey - The worker node key.
594 * @param worker - The worker.
595 * @param tasksUsage - The worker tasks usage.
596 * @param tasksQueue - The worker task queue.
598 private setWorkerNode (
599 workerNodeKey
: number,
601 tasksUsage
: TasksUsage
,
602 tasksQueue
: Array<Task
<Data
>>
604 this.workerNodes
[workerNodeKey
] = {
612 * Removes the given worker from the pool worker nodes.
614 * @param worker - The worker.
616 private removeWorkerNode (worker
: Worker
): void {
617 const workerNodeKey
= this.getWorkerNodeKey(worker
)
618 this.workerNodes
.splice(workerNodeKey
, 1)
619 this.workerChoiceStrategyContext
.remove(workerNodeKey
)
622 private executeTask (workerNodeKey
: number, task
: Task
<Data
>): void {
623 this.beforeTaskExecutionHook(workerNodeKey
)
624 this.sendToWorker(this.workerNodes
[workerNodeKey
].worker
, task
)
627 private enqueueTask (workerNodeKey
: number, task
: Task
<Data
>): number {
628 return this.workerNodes
[workerNodeKey
].tasksQueue
.push(task
)
631 private dequeueTask (workerNodeKey
: number): Task
<Data
> | undefined {
632 return this.workerNodes
[workerNodeKey
].tasksQueue
.shift()
635 private tasksQueueSize (workerNodeKey
: number): number {
636 return this.workerNodes
[workerNodeKey
].tasksQueue
.length
639 private flushTasksQueue (workerNodeKey
: number): void {
640 if (this.tasksQueueSize(workerNodeKey
) > 0) {
641 for (const task
of this.workerNodes
[workerNodeKey
].tasksQueue
) {
642 this.executeTask(workerNodeKey
, task
)
647 private flushTasksQueueByWorker (worker
: Worker
): void {
648 const workerNodeKey
= this.getWorkerNodeKey(worker
)
649 this.flushTasksQueue(workerNodeKey
)