1 import crypto from
'node:crypto'
2 import { performance
} from
'node:perf_hooks'
3 import type { MessageValue
, PromiseResponseWrapper
} from
'../utility-types'
5 DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
,
10 import { KillBehaviors
, isKillBehavior
} from
'../worker/worker-options'
11 import { CircularArray
} from
'../circular-array'
12 import { Queue
} from
'../queue'
21 type TasksQueueOptions
,
24 import type { IWorker
, Task
, WorkerNode
, WorkerUsage
} from
'./worker'
26 WorkerChoiceStrategies
,
27 type WorkerChoiceStrategy
,
28 type WorkerChoiceStrategyOptions
29 } from
'./selection-strategies/selection-strategies-types'
30 import { WorkerChoiceStrategyContext
} from
'./selection-strategies/worker-choice-strategy-context'
33 * Base class that implements some shared logic for all poolifier pools.
35 * @typeParam Worker - Type of worker which manages this pool.
36 * @typeParam Data - Type of data sent to the worker. This can only be serializable data.
37 * @typeParam Response - Type of execution response. This can only be serializable data.
39 export abstract class AbstractPool
<
40 Worker
extends IWorker
,
43 > implements IPool
<Worker
, Data
, Response
> {
45 public readonly workerNodes
: Array<WorkerNode
<Worker
, Data
>> = []
48 public readonly emitter
?: PoolEmitter
51 * The execution response promise map.
53 * - `key`: The message id of each submitted task.
54 * - `value`: An object that contains the worker, the execution response promise resolve and reject callbacks.
56 * When we receive a message from the worker, we get a map entry with the promise resolve/reject bound to the message id.
58 protected promiseResponseMap
: Map
<
60 PromiseResponseWrapper
<Worker
, Response
>
61 > = new Map
<string, PromiseResponseWrapper
<Worker
, Response
>>()
64 * Worker choice strategy context referencing a worker choice algorithm implementation.
66 * Default to a round robin algorithm.
68 protected workerChoiceStrategyContext
: WorkerChoiceStrategyContext
<
75 * Constructs a new poolifier pool.
77 * @param numberOfWorkers - Number of workers that this pool should manage.
78 * @param filePath - Path to the worker file.
79 * @param opts - Options for the pool.
82 protected readonly numberOfWorkers
: number,
83 protected readonly filePath
: string,
84 protected readonly opts
: PoolOptions
<Worker
>
87 throw new Error('Cannot start a pool from a worker!')
89 this.checkNumberOfWorkers(this.numberOfWorkers
)
90 this.checkFilePath(this.filePath
)
91 this.checkPoolOptions(this.opts
)
93 this.chooseWorkerNode
= this.chooseWorkerNode
.bind(this)
94 this.executeTask
= this.executeTask
.bind(this)
95 this.enqueueTask
= this.enqueueTask
.bind(this)
96 this.checkAndEmitEvents
= this.checkAndEmitEvents
.bind(this)
98 if (this.opts
.enableEvents
=== true) {
99 this.emitter
= new PoolEmitter()
101 this.workerChoiceStrategyContext
= new WorkerChoiceStrategyContext
<
107 this.opts
.workerChoiceStrategy
,
108 this.opts
.workerChoiceStrategyOptions
113 for (let i
= 1; i
<= this.numberOfWorkers
; i
++) {
114 this.createAndSetupWorker()
118 private checkFilePath (filePath
: string): void {
121 (typeof filePath
=== 'string' && filePath
.trim().length
=== 0)
123 throw new Error('Please specify a file with a worker implementation')
127 private checkNumberOfWorkers (numberOfWorkers
: number): void {
128 if (numberOfWorkers
== null) {
130 'Cannot instantiate a pool without specifying the number of workers'
132 } else if (!Number.isSafeInteger(numberOfWorkers
)) {
134 'Cannot instantiate a pool with a non safe integer number of workers'
136 } else if (numberOfWorkers
< 0) {
137 throw new RangeError(
138 'Cannot instantiate a pool with a negative number of workers'
140 } else if (this.type === PoolTypes
.fixed
&& numberOfWorkers
=== 0) {
141 throw new Error('Cannot instantiate a fixed pool with no worker')
145 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
146 if (isPlainObject(opts
)) {
147 this.opts
.workerChoiceStrategy
=
148 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
149 this.checkValidWorkerChoiceStrategy(this.opts
.workerChoiceStrategy
)
150 this.opts
.workerChoiceStrategyOptions
=
151 opts
.workerChoiceStrategyOptions
??
152 DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
153 this.checkValidWorkerChoiceStrategyOptions(
154 this.opts
.workerChoiceStrategyOptions
156 this.opts
.restartWorkerOnError
= opts
.restartWorkerOnError
?? true
157 this.opts
.enableEvents
= opts
.enableEvents
?? true
158 this.opts
.enableTasksQueue
= opts
.enableTasksQueue
?? false
159 if (this.opts
.enableTasksQueue
) {
160 this.checkValidTasksQueueOptions(
161 opts
.tasksQueueOptions
as TasksQueueOptions
163 this.opts
.tasksQueueOptions
= this.buildTasksQueueOptions(
164 opts
.tasksQueueOptions
as TasksQueueOptions
168 throw new TypeError('Invalid pool options: must be a plain object')
172 private checkValidWorkerChoiceStrategy (
173 workerChoiceStrategy
: WorkerChoiceStrategy
175 if (!Object.values(WorkerChoiceStrategies
).includes(workerChoiceStrategy
)) {
177 `Invalid worker choice strategy '${workerChoiceStrategy}'`
182 private checkValidWorkerChoiceStrategyOptions (
183 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
185 if (!isPlainObject(workerChoiceStrategyOptions
)) {
187 'Invalid worker choice strategy options: must be a plain object'
191 workerChoiceStrategyOptions
.weights
!= null &&
192 Object.keys(workerChoiceStrategyOptions
.weights
).length
!== this.maxSize
195 'Invalid worker choice strategy options: must have a weight for each worker node'
200 private checkValidTasksQueueOptions (
201 tasksQueueOptions
: TasksQueueOptions
203 if (tasksQueueOptions
!= null && !isPlainObject(tasksQueueOptions
)) {
204 throw new TypeError('Invalid tasks queue options: must be a plain object')
206 if ((tasksQueueOptions
?.concurrency
as number) <= 0) {
208 `Invalid worker tasks concurrency '${
209 tasksQueueOptions.concurrency as number
216 public get
info (): PoolInfo
{
220 minSize
: this.minSize
,
221 maxSize
: this.maxSize
,
222 workerNodes
: this.workerNodes
.length
,
223 idleWorkerNodes
: this.workerNodes
.reduce(
224 (accumulator
, workerNode
) =>
225 workerNode
.workerUsage
.tasks
.executing
=== 0
230 busyWorkerNodes
: this.workerNodes
.reduce(
231 (accumulator
, workerNode
) =>
232 workerNode
.workerUsage
.tasks
.executing
> 0
237 executedTasks
: this.workerNodes
.reduce(
238 (accumulator
, workerNode
) =>
239 accumulator
+ workerNode
.workerUsage
.tasks
.executed
,
242 executingTasks
: this.workerNodes
.reduce(
243 (accumulator
, workerNode
) =>
244 accumulator
+ workerNode
.workerUsage
.tasks
.executing
,
247 queuedTasks
: this.workerNodes
.reduce(
248 (accumulator
, workerNode
) => accumulator
+ workerNode
.tasksQueue
.size
,
251 maxQueuedTasks
: this.workerNodes
.reduce(
252 (accumulator
, workerNode
) =>
253 accumulator
+ workerNode
.tasksQueue
.maxSize
,
256 failedTasks
: this.workerNodes
.reduce(
257 (accumulator
, workerNode
) =>
258 accumulator
+ workerNode
.workerUsage
.tasks
.failed
,
267 * If it is `'dynamic'`, it provides the `max` property.
269 protected abstract get
type (): PoolType
272 * Gets the worker type.
274 protected abstract get
worker (): WorkerType
279 protected abstract get
minSize (): number
284 protected abstract get
maxSize (): number
287 * Gets the given worker its worker node key.
289 * @param worker - The worker.
290 * @returns The worker node key if the worker is found in the pool worker nodes, `-1` otherwise.
292 private getWorkerNodeKey (worker
: Worker
): number {
293 return this.workerNodes
.findIndex(
294 workerNode
=> workerNode
.worker
=== worker
299 public setWorkerChoiceStrategy (
300 workerChoiceStrategy
: WorkerChoiceStrategy
,
301 workerChoiceStrategyOptions
?: WorkerChoiceStrategyOptions
303 this.checkValidWorkerChoiceStrategy(workerChoiceStrategy
)
304 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
305 this.workerChoiceStrategyContext
.setWorkerChoiceStrategy(
306 this.opts
.workerChoiceStrategy
308 if (workerChoiceStrategyOptions
!= null) {
309 this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
311 for (const workerNode
of this.workerNodes
) {
312 this.setWorkerNodeTasksUsage(workerNode
, {
317 this.opts
.enableTasksQueue
=== true
318 ? workerNode
.tasksQueue
.size
326 history
: new CircularArray()
332 history
: new CircularArray()
336 this.setWorkerStatistics(workerNode
.worker
)
341 public setWorkerChoiceStrategyOptions (
342 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
344 this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
345 this.opts
.workerChoiceStrategyOptions
= workerChoiceStrategyOptions
346 this.workerChoiceStrategyContext
.setOptions(
347 this.opts
.workerChoiceStrategyOptions
352 public enableTasksQueue (
354 tasksQueueOptions
?: TasksQueueOptions
356 if (this.opts
.enableTasksQueue
=== true && !enable
) {
357 this.flushTasksQueues()
359 this.opts
.enableTasksQueue
= enable
360 this.setTasksQueueOptions(tasksQueueOptions
as TasksQueueOptions
)
364 public setTasksQueueOptions (tasksQueueOptions
: TasksQueueOptions
): void {
365 if (this.opts
.enableTasksQueue
=== true) {
366 this.checkValidTasksQueueOptions(tasksQueueOptions
)
367 this.opts
.tasksQueueOptions
=
368 this.buildTasksQueueOptions(tasksQueueOptions
)
369 } else if (this.opts
.tasksQueueOptions
!= null) {
370 delete this.opts
.tasksQueueOptions
374 private buildTasksQueueOptions (
375 tasksQueueOptions
: TasksQueueOptions
376 ): TasksQueueOptions
{
378 concurrency
: tasksQueueOptions
?.concurrency
?? 1
383 * Whether the pool is full or not.
385 * The pool filling boolean status.
387 protected get
full (): boolean {
388 return this.workerNodes
.length
>= this.maxSize
392 * Whether the pool is busy or not.
394 * The pool busyness boolean status.
396 protected abstract get
busy (): boolean
398 protected internalBusy (): boolean {
400 this.workerNodes
.findIndex(workerNode
=> {
401 return workerNode
.workerUsage
.tasks
.executing
=== 0
407 public async execute (data
?: Data
, name
?: string): Promise
<Response
> {
408 const timestamp
= performance
.now()
409 const workerNodeKey
= this.chooseWorkerNode()
410 const submittedTask
: Task
<Data
> = {
412 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
413 data
: data
?? ({} as Data
),
415 id
: crypto
.randomUUID()
417 const res
= new Promise
<Response
>((resolve
, reject
) => {
418 this.promiseResponseMap
.set(submittedTask
.id
as string, {
421 worker
: this.workerNodes
[workerNodeKey
].worker
425 this.opts
.enableTasksQueue
=== true &&
427 this.workerNodes
[workerNodeKey
].workerUsage
.tasks
.executing
>=
428 ((this.opts
.tasksQueueOptions
as TasksQueueOptions
)
429 .concurrency
as number))
431 this.enqueueTask(workerNodeKey
, submittedTask
)
433 this.executeTask(workerNodeKey
, submittedTask
)
435 this.workerChoiceStrategyContext
.update(workerNodeKey
)
436 this.checkAndEmitEvents()
437 // eslint-disable-next-line @typescript-eslint/return-await
442 public async destroy (): Promise
<void> {
444 this.workerNodes
.map(async (workerNode
, workerNodeKey
) => {
445 this.flushTasksQueue(workerNodeKey
)
446 // FIXME: wait for tasks to be finished
447 await this.destroyWorker(workerNode
.worker
)
453 * Shutdowns the given worker.
455 * @param worker - A worker within `workerNodes`.
457 protected abstract destroyWorker (worker
: Worker
): void | Promise
<void>
460 * Setup hook to execute code before worker node are created in the abstract constructor.
465 protected setupHook (): void {
466 // Intentionally empty
470 * Should return whether the worker is the main worker or not.
472 protected abstract isMain (): boolean
475 * Hook executed before the worker task execution.
478 * @param workerNodeKey - The worker node key.
480 protected beforeTaskExecutionHook (workerNodeKey
: number): void {
481 ++this.workerNodes
[workerNodeKey
].workerUsage
.tasks
.executing
482 if (this.opts
.enableTasksQueue
=== true) {
483 this.workerNodes
[workerNodeKey
].workerUsage
.tasks
.queued
=
484 this.tasksQueueSize(workerNodeKey
)
489 * Hook executed after the worker task execution.
492 * @param worker - The worker.
493 * @param message - The received message.
495 protected afterTaskExecutionHook (
497 message
: MessageValue
<Response
>
500 this.workerNodes
[this.getWorkerNodeKey(worker
)].workerUsage
501 const workerTaskStatistics
= workerUsage
.tasks
502 --workerTaskStatistics
.executing
503 ++workerTaskStatistics
.executed
504 if (message
.taskError
!= null) {
505 ++workerTaskStatistics
.failed
508 this.updateRunTimeWorkerUsage(workerUsage
, message
)
509 this.updateWaitTimeWorkerUsage(workerUsage
, message
)
510 this.updateEluWorkerUsage(workerUsage
, message
)
513 private updateRunTimeWorkerUsage (
514 workerUsage
: WorkerUsage
,
515 message
: MessageValue
<Response
>
517 if (this.workerChoiceStrategyContext
.getTaskStatistics().runTime
) {
518 workerUsage
.runTime
.aggregation
+= message
.taskPerformance
?.runTime
?? 0
520 this.workerChoiceStrategyContext
.getTaskStatistics().avgRunTime
&&
521 workerUsage
.tasks
.executed
!== 0
523 workerUsage
.runTime
.average
=
524 workerUsage
.runTime
.aggregation
/ workerUsage
.tasks
.executed
527 this.workerChoiceStrategyContext
.getTaskStatistics().medRunTime
&&
528 message
.taskPerformance
?.runTime
!= null
530 workerUsage
.runTime
.history
.push(message
.taskPerformance
.runTime
)
531 workerUsage
.runTime
.median
= median(workerUsage
.runTime
.history
)
536 private updateWaitTimeWorkerUsage (
537 workerUsage
: WorkerUsage
,
538 message
: MessageValue
<Response
>
540 if (this.workerChoiceStrategyContext
.getTaskStatistics().waitTime
) {
541 workerUsage
.waitTime
.aggregation
+= message
.taskPerformance
?.waitTime
?? 0
543 this.workerChoiceStrategyContext
.getTaskStatistics().avgWaitTime
&&
544 workerUsage
.tasks
.executed
!== 0
546 workerUsage
.waitTime
.average
=
547 workerUsage
.waitTime
.aggregation
/ workerUsage
.tasks
.executed
550 this.workerChoiceStrategyContext
.getTaskStatistics().medWaitTime
&&
551 message
.taskPerformance
?.waitTime
!= null
553 workerUsage
.waitTime
.history
.push(message
.taskPerformance
.waitTime
)
554 workerUsage
.waitTime
.median
= median(workerUsage
.waitTime
.history
)
559 private updateEluWorkerUsage (
560 workerTasksUsage
: WorkerUsage
,
561 message
: MessageValue
<Response
>
563 if (this.workerChoiceStrategyContext
.getTaskStatistics().elu
) {
565 workerTasksUsage
.elu
!= null &&
566 message
.taskPerformance
?.elu
!= null
568 workerTasksUsage
.elu
= {
569 idle
: workerTasksUsage
.elu
.idle
+ message
.taskPerformance
.elu
.idle
,
571 workerTasksUsage
.elu
.active
+ message
.taskPerformance
.elu
.active
,
573 (workerTasksUsage
.elu
.utilization
+
574 message
.taskPerformance
.elu
.utilization
) /
577 } else if (message
.taskPerformance
?.elu
!= null) {
578 workerTasksUsage
.elu
= message
.taskPerformance
.elu
584 * Chooses a worker node for the next task.
586 * The default worker choice strategy uses a round robin algorithm to distribute the load.
588 * @returns The worker node key
590 protected chooseWorkerNode (): number {
591 let workerNodeKey
: number
592 if (this.type === PoolTypes
.dynamic
&& !this.full
&& this.internalBusy()) {
593 const workerCreated
= this.createAndSetupWorker()
594 this.registerWorkerMessageListener(workerCreated
, message
=> {
595 const currentWorkerNodeKey
= this.getWorkerNodeKey(workerCreated
)
597 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
598 (message
.kill
!= null &&
599 this.workerNodes
[currentWorkerNodeKey
].workerUsage
.tasks
602 // Kill message received from the worker: no new tasks are submitted to that worker for a while ( > maxInactiveTime)
603 this.flushTasksQueue(currentWorkerNodeKey
)
604 // FIXME: wait for tasks to be finished
605 void (this.destroyWorker(workerCreated
) as Promise
<void>)
608 workerNodeKey
= this.getWorkerNodeKey(workerCreated
)
610 workerNodeKey
= this.workerChoiceStrategyContext
.execute()
616 * Sends a message to the given worker.
618 * @param worker - The worker which should receive the message.
619 * @param message - The message.
621 protected abstract sendToWorker (
623 message
: MessageValue
<Data
>
627 * Registers a listener callback on the given worker.
629 * @param worker - The worker which should register a listener.
630 * @param listener - The message listener callback.
632 protected abstract registerWorkerMessageListener
<
633 Message
extends Data
| Response
634 >(worker
: Worker
, listener
: (message
: MessageValue
<Message
>) => void): void
637 * Returns a newly created worker.
639 protected abstract createWorker (): Worker
642 * Function that can be hooked up when a worker has been newly created and moved to the pool worker nodes.
644 * Can be used to update the `maxListeners` or binding the `main-worker`\<-\>`worker` connection if not bind by default.
646 * @param worker - The newly created worker.
648 protected abstract afterWorkerSetup (worker
: Worker
): void
651 * Creates a new worker and sets it up completely in the pool worker nodes.
653 * @returns New, completely set up worker.
655 protected createAndSetupWorker (): Worker
{
656 const worker
= this.createWorker()
658 worker
.on('message', this.opts
.messageHandler
?? EMPTY_FUNCTION
)
659 worker
.on('error', this.opts
.errorHandler
?? EMPTY_FUNCTION
)
660 worker
.on('error', error
=> {
661 if (this.emitter
!= null) {
662 this.emitter
.emit(PoolEvents
.error
, error
)
665 worker
.on('error', () => {
666 if (this.opts
.restartWorkerOnError
=== true) {
667 this.createAndSetupWorker()
670 worker
.on('online', this.opts
.onlineHandler
?? EMPTY_FUNCTION
)
671 worker
.on('exit', this.opts
.exitHandler
?? EMPTY_FUNCTION
)
672 worker
.once('exit', () => {
673 this.removeWorkerNode(worker
)
676 this.pushWorkerNode(worker
)
678 this.setWorkerStatistics(worker
)
680 this.afterWorkerSetup(worker
)
686 * This function is the listener registered for each worker message.
688 * @returns The listener function to execute when a message is received from a worker.
690 protected workerListener (): (message
: MessageValue
<Response
>) => void {
692 if (message
.id
!= null) {
693 // Task execution response received
694 const promiseResponse
= this.promiseResponseMap
.get(message
.id
)
695 if (promiseResponse
!= null) {
696 if (message
.taskError
!= null) {
697 promiseResponse
.reject(message
.taskError
.message
)
698 if (this.emitter
!= null) {
699 this.emitter
.emit(PoolEvents
.taskError
, message
.taskError
)
702 promiseResponse
.resolve(message
.data
as Response
)
704 this.afterTaskExecutionHook(promiseResponse
.worker
, message
)
705 this.promiseResponseMap
.delete(message
.id
)
706 const workerNodeKey
= this.getWorkerNodeKey(promiseResponse
.worker
)
708 this.opts
.enableTasksQueue
=== true &&
709 this.tasksQueueSize(workerNodeKey
) > 0
713 this.dequeueTask(workerNodeKey
) as Task
<Data
>
721 private checkAndEmitEvents (): void {
722 if (this.emitter
!= null) {
724 this.emitter
?.emit(PoolEvents
.busy
, this.info
)
726 if (this.type === PoolTypes
.dynamic
&& this.full
) {
727 this.emitter
?.emit(PoolEvents
.full
, this.info
)
733 * Sets the given worker node its tasks usage in the pool.
735 * @param workerNode - The worker node.
736 * @param workerUsage - The worker usage.
738 private setWorkerNodeTasksUsage (
739 workerNode
: WorkerNode
<Worker
, Data
>,
740 workerUsage
: WorkerUsage
742 workerNode
.workerUsage
= workerUsage
746 * Pushes the given worker in the pool worker nodes.
748 * @param worker - The worker.
749 * @returns The worker nodes length.
751 private pushWorkerNode (worker
: Worker
): number {
752 return this.workerNodes
.push({
765 history
: new CircularArray()
772 history
: new CircularArray()
776 tasksQueue
: new Queue
<Task
<Data
>>()
781 * Sets the given worker in the pool worker nodes.
783 * @param workerNodeKey - The worker node key.
784 * @param worker - The worker.
785 * @param workerUsage - The worker usage.
786 * @param tasksQueue - The worker task queue.
788 private setWorkerNode (
789 workerNodeKey
: number,
791 workerUsage
: WorkerUsage
,
792 tasksQueue
: Queue
<Task
<Data
>>
794 this.workerNodes
[workerNodeKey
] = {
802 * Removes the given worker from the pool worker nodes.
804 * @param worker - The worker.
806 private removeWorkerNode (worker
: Worker
): void {
807 const workerNodeKey
= this.getWorkerNodeKey(worker
)
808 if (workerNodeKey
!== -1) {
809 this.workerNodes
.splice(workerNodeKey
, 1)
810 this.workerChoiceStrategyContext
.remove(workerNodeKey
)
814 private executeTask (workerNodeKey
: number, task
: Task
<Data
>): void {
815 this.beforeTaskExecutionHook(workerNodeKey
)
816 this.sendToWorker(this.workerNodes
[workerNodeKey
].worker
, task
)
819 private enqueueTask (workerNodeKey
: number, task
: Task
<Data
>): number {
820 return this.workerNodes
[workerNodeKey
].tasksQueue
.enqueue(task
)
823 private dequeueTask (workerNodeKey
: number): Task
<Data
> | undefined {
824 return this.workerNodes
[workerNodeKey
].tasksQueue
.dequeue()
827 private tasksQueueSize (workerNodeKey
: number): number {
828 return this.workerNodes
[workerNodeKey
].tasksQueue
.size
831 private flushTasksQueue (workerNodeKey
: number): void {
832 if (this.tasksQueueSize(workerNodeKey
) > 0) {
833 for (let i
= 0; i
< this.tasksQueueSize(workerNodeKey
); i
++) {
836 this.dequeueTask(workerNodeKey
) as Task
<Data
>
842 private flushTasksQueues (): void {
843 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
844 this.flushTasksQueue(workerNodeKey
)
848 private setWorkerStatistics (worker
: Worker
): void {
849 this.sendToWorker(worker
, {
851 runTime
: this.workerChoiceStrategyContext
.getTaskStatistics().runTime
,
852 waitTime
: this.workerChoiceStrategyContext
.getTaskStatistics().waitTime
,
853 elu
: this.workerChoiceStrategyContext
.getTaskStatistics().elu