1 import crypto from
'node:crypto'
2 import type { MessageValue
, PromiseResponseWrapper
} from
'../utility-types'
4 DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
,
9 import { KillBehaviors
, isKillBehavior
} from
'../worker/worker-options'
10 import { CircularArray
} from
'../circular-array'
11 import { Queue
} from
'../queue'
18 type TasksQueueOptions
20 import type { IWorker
, Task
, TasksUsage
, WorkerNode
} from
'./worker'
22 WorkerChoiceStrategies
,
23 type WorkerChoiceStrategy
,
24 type WorkerChoiceStrategyOptions
25 } from
'./selection-strategies/selection-strategies-types'
26 import { WorkerChoiceStrategyContext
} from
'./selection-strategies/worker-choice-strategy-context'
29 * Base class that implements some shared logic for all poolifier pools.
31 * @typeParam Worker - Type of worker which manages this pool.
32 * @typeParam Data - Type of data sent to the worker. This can only be serializable data.
33 * @typeParam Response - Type of execution response. This can only be serializable data.
35 export abstract class AbstractPool
<
36 Worker
extends IWorker
,
39 > implements IPool
<Worker
, Data
, Response
> {
41 public readonly workerNodes
: Array<WorkerNode
<Worker
, Data
>> = []
44 public readonly emitter
?: PoolEmitter
47 * The execution response promise map.
49 * - `key`: The message id of each submitted task.
50 * - `value`: An object that contains the worker, the execution response promise resolve and reject callbacks.
52 * When we receive a message from the worker, we get a map entry with the promise resolve/reject bound to the message id.
54 protected promiseResponseMap
: Map
<
56 PromiseResponseWrapper
<Worker
, Response
>
57 > = new Map
<string, PromiseResponseWrapper
<Worker
, Response
>>()
60 * Worker choice strategy context referencing a worker choice algorithm implementation.
62 * Default to a round robin algorithm.
64 protected workerChoiceStrategyContext
: WorkerChoiceStrategyContext
<
71 * Constructs a new poolifier pool.
73 * @param numberOfWorkers - Number of workers that this pool should manage.
74 * @param filePath - Path to the worker file.
75 * @param opts - Options for the pool.
78 public readonly numberOfWorkers
: number,
79 public readonly filePath
: string,
80 public readonly opts
: PoolOptions
<Worker
>
83 throw new Error('Cannot start a pool from a worker!')
85 this.checkNumberOfWorkers(this.numberOfWorkers
)
86 this.checkFilePath(this.filePath
)
87 this.checkPoolOptions(this.opts
)
89 this.chooseWorkerNode
= this.chooseWorkerNode
.bind(this)
90 this.executeTask
= this.executeTask
.bind(this)
91 this.enqueueTask
= this.enqueueTask
.bind(this)
92 this.checkAndEmitEvents
= this.checkAndEmitEvents
.bind(this)
96 for (let i
= 1; i
<= this.numberOfWorkers
; i
++) {
97 this.createAndSetupWorker()
100 if (this.opts
.enableEvents
=== true) {
101 this.emitter
= new PoolEmitter()
103 this.workerChoiceStrategyContext
= new WorkerChoiceStrategyContext
<
109 this.opts
.workerChoiceStrategy
,
110 this.opts
.workerChoiceStrategyOptions
114 private checkFilePath (filePath
: string): void {
117 (typeof filePath
=== 'string' && filePath
.trim().length
=== 0)
119 throw new Error('Please specify a file with a worker implementation')
123 private checkNumberOfWorkers (numberOfWorkers
: number): void {
124 if (numberOfWorkers
== null) {
126 'Cannot instantiate a pool without specifying the number of workers'
128 } else if (!Number.isSafeInteger(numberOfWorkers
)) {
130 'Cannot instantiate a pool with a non safe integer number of workers'
132 } else if (numberOfWorkers
< 0) {
133 throw new RangeError(
134 'Cannot instantiate a pool with a negative number of workers'
136 } else if (this.type === PoolType
.FIXED
&& numberOfWorkers
=== 0) {
137 throw new Error('Cannot instantiate a fixed pool with no worker')
141 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
142 if (isPlainObject(opts
)) {
143 this.opts
.workerChoiceStrategy
=
144 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
145 this.checkValidWorkerChoiceStrategy(this.opts
.workerChoiceStrategy
)
146 this.opts
.workerChoiceStrategyOptions
=
147 opts
.workerChoiceStrategyOptions
??
148 DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
149 this.checkValidWorkerChoiceStrategyOptions(
150 this.opts
.workerChoiceStrategyOptions
152 this.opts
.restartWorkerOnError
= opts
.restartWorkerOnError
?? true
153 this.opts
.enableEvents
= opts
.enableEvents
?? true
154 this.opts
.enableTasksQueue
= opts
.enableTasksQueue
?? false
155 if (this.opts
.enableTasksQueue
) {
156 this.checkValidTasksQueueOptions(
157 opts
.tasksQueueOptions
as TasksQueueOptions
159 this.opts
.tasksQueueOptions
= this.buildTasksQueueOptions(
160 opts
.tasksQueueOptions
as TasksQueueOptions
164 throw new TypeError('Invalid pool options: must be a plain object')
168 private checkValidWorkerChoiceStrategy (
169 workerChoiceStrategy
: WorkerChoiceStrategy
171 if (!Object.values(WorkerChoiceStrategies
).includes(workerChoiceStrategy
)) {
173 `Invalid worker choice strategy '${workerChoiceStrategy}'`
178 private checkValidWorkerChoiceStrategyOptions (
179 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
181 if (!isPlainObject(workerChoiceStrategyOptions
)) {
183 'Invalid worker choice strategy options: must be a plain object'
187 workerChoiceStrategyOptions
.weights
!= null &&
188 Object.keys(workerChoiceStrategyOptions
.weights
).length
!== this.size
191 'Invalid worker choice strategy options: must have a weight for each worker node'
196 private checkValidTasksQueueOptions (
197 tasksQueueOptions
: TasksQueueOptions
199 if (tasksQueueOptions
!= null && !isPlainObject(tasksQueueOptions
)) {
200 throw new TypeError('Invalid tasks queue options: must be a plain object')
202 if ((tasksQueueOptions
?.concurrency
as number) <= 0) {
204 `Invalid worker tasks concurrency '${
205 tasksQueueOptions.concurrency as number
212 public abstract get
type (): PoolType
215 public abstract get
size (): number
218 * Number of tasks running in the pool.
220 private get
numberOfRunningTasks (): number {
221 return this.workerNodes
.reduce(
222 (accumulator
, workerNode
) => accumulator
+ workerNode
.tasksUsage
.running
,
228 * Number of tasks queued in the pool.
230 private get
numberOfQueuedTasks (): number {
231 if (this.opts
.enableTasksQueue
=== false) {
234 return this.workerNodes
.reduce(
235 (accumulator
, workerNode
) => accumulator
+ workerNode
.tasksQueue
.size
,
241 * Gets the given worker its worker node key.
243 * @param worker - The worker.
244 * @returns The worker node key if the worker is found in the pool worker nodes, `-1` otherwise.
246 private getWorkerNodeKey (worker
: Worker
): number {
247 return this.workerNodes
.findIndex(
248 workerNode
=> workerNode
.worker
=== worker
253 public setWorkerChoiceStrategy (
254 workerChoiceStrategy
: WorkerChoiceStrategy
,
255 workerChoiceStrategyOptions
?: WorkerChoiceStrategyOptions
257 this.checkValidWorkerChoiceStrategy(workerChoiceStrategy
)
258 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
259 for (const workerNode
of this.workerNodes
) {
260 this.setWorkerNodeTasksUsage(workerNode
, {
264 runTimeHistory
: new CircularArray(),
268 waitTimeHistory
: new CircularArray(),
274 this.workerChoiceStrategyContext
.setWorkerChoiceStrategy(
275 this.opts
.workerChoiceStrategy
277 if (workerChoiceStrategyOptions
!= null) {
278 this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
283 public setWorkerChoiceStrategyOptions (
284 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
286 this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
287 this.opts
.workerChoiceStrategyOptions
= workerChoiceStrategyOptions
288 this.workerChoiceStrategyContext
.setOptions(
289 this.opts
.workerChoiceStrategyOptions
294 public enableTasksQueue (
296 tasksQueueOptions
?: TasksQueueOptions
298 if (this.opts
.enableTasksQueue
=== true && !enable
) {
299 this.flushTasksQueues()
301 this.opts
.enableTasksQueue
= enable
302 this.setTasksQueueOptions(tasksQueueOptions
as TasksQueueOptions
)
306 public setTasksQueueOptions (tasksQueueOptions
: TasksQueueOptions
): void {
307 if (this.opts
.enableTasksQueue
=== true) {
308 this.checkValidTasksQueueOptions(tasksQueueOptions
)
309 this.opts
.tasksQueueOptions
=
310 this.buildTasksQueueOptions(tasksQueueOptions
)
312 delete this.opts
.tasksQueueOptions
316 private buildTasksQueueOptions (
317 tasksQueueOptions
: TasksQueueOptions
318 ): TasksQueueOptions
{
320 concurrency
: tasksQueueOptions
?.concurrency
?? 1
325 * Whether the pool is full or not.
327 * The pool filling boolean status.
329 protected abstract get
full (): boolean
332 * Whether the pool is busy or not.
334 * The pool busyness boolean status.
336 protected abstract get
busy (): boolean
338 protected internalBusy (): boolean {
340 this.workerNodes
.findIndex(workerNode
=> {
341 return workerNode
.tasksUsage
.running
=== 0
347 public async execute (data
?: Data
, name
?: string): Promise
<Response
> {
348 const submissionTimestamp
= performance
.now()
349 const workerNodeKey
= this.chooseWorkerNode()
350 const submittedTask
: Task
<Data
> = {
352 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
353 data
: data
?? ({} as Data
),
355 id
: crypto
.randomUUID()
357 const res
= new Promise
<Response
>((resolve
, reject
) => {
358 this.promiseResponseMap
.set(submittedTask
.id
as string, {
361 worker
: this.workerNodes
[workerNodeKey
].worker
365 this.opts
.enableTasksQueue
=== true &&
367 this.workerNodes
[workerNodeKey
].tasksUsage
.running
>=
368 ((this.opts
.tasksQueueOptions
as TasksQueueOptions
)
369 .concurrency
as number))
371 this.enqueueTask(workerNodeKey
, submittedTask
)
373 this.executeTask(workerNodeKey
, submittedTask
)
375 this.workerChoiceStrategyContext
.update(workerNodeKey
)
376 this.checkAndEmitEvents()
377 // eslint-disable-next-line @typescript-eslint/return-await
382 public async destroy (): Promise
<void> {
384 this.workerNodes
.map(async (workerNode
, workerNodeKey
) => {
385 this.flushTasksQueue(workerNodeKey
)
386 await this.destroyWorker(workerNode
.worker
)
392 * Shutdowns the given worker.
394 * @param worker - A worker within `workerNodes`.
396 protected abstract destroyWorker (worker
: Worker
): void | Promise
<void>
399 * Setup hook to execute code before worker node are created in the abstract constructor.
404 protected setupHook (): void {
405 // Intentionally empty
409 * Should return whether the worker is the main worker or not.
411 protected abstract isMain (): boolean
414 * Hook executed before the worker task execution.
417 * @param workerNodeKey - The worker node key.
419 protected beforeTaskExecutionHook (workerNodeKey
: number): void {
420 ++this.workerNodes
[workerNodeKey
].tasksUsage
.running
424 * Hook executed after the worker task execution.
427 * @param worker - The worker.
428 * @param message - The received message.
430 protected afterTaskExecutionHook (
432 message
: MessageValue
<Response
>
434 const workerTasksUsage
=
435 this.workerNodes
[this.getWorkerNodeKey(worker
)].tasksUsage
436 --workerTasksUsage
.running
437 ++workerTasksUsage
.run
438 if (message
.error
!= null) {
439 ++workerTasksUsage
.error
441 this.updateRunTimeTasksUsage(workerTasksUsage
, message
)
442 this.updateWaitTimeTasksUsage(workerTasksUsage
, message
)
445 private updateRunTimeTasksUsage (
446 workerTasksUsage
: TasksUsage
,
447 message
: MessageValue
<Response
>
449 if (this.workerChoiceStrategyContext
.getRequiredStatistics().runTime
) {
450 workerTasksUsage
.runTime
+= message
.runTime
?? 0
452 this.workerChoiceStrategyContext
.getRequiredStatistics().avgRunTime
&&
453 workerTasksUsage
.run
!== 0
455 workerTasksUsage
.avgRunTime
=
456 workerTasksUsage
.runTime
/ workerTasksUsage
.run
459 this.workerChoiceStrategyContext
.getRequiredStatistics().medRunTime
&&
460 message
.runTime
!= null
462 workerTasksUsage
.runTimeHistory
.push(message
.runTime
)
463 workerTasksUsage
.medRunTime
= median(workerTasksUsage
.runTimeHistory
)
468 private updateWaitTimeTasksUsage (
469 workerTasksUsage
: TasksUsage
,
470 message
: MessageValue
<Response
>
472 if (this.workerChoiceStrategyContext
.getRequiredStatistics().waitTime
) {
473 workerTasksUsage
.waitTime
+= message
.waitTime
?? 0
475 this.workerChoiceStrategyContext
.getRequiredStatistics().avgWaitTime
&&
476 workerTasksUsage
.run
!== 0
478 workerTasksUsage
.avgWaitTime
=
479 workerTasksUsage
.waitTime
/ workerTasksUsage
.run
482 this.workerChoiceStrategyContext
.getRequiredStatistics().medWaitTime
&&
483 message
.waitTime
!= null
485 workerTasksUsage
.waitTimeHistory
.push(message
.waitTime
)
486 workerTasksUsage
.medWaitTime
= median(workerTasksUsage
.waitTimeHistory
)
492 * Chooses a worker node for the next task.
494 * The default worker choice strategy uses a round robin algorithm to distribute the load.
496 * @returns The worker node key
498 protected chooseWorkerNode (): number {
499 let workerNodeKey
: number
500 if (this.type === PoolType
.DYNAMIC
&& !this.full
&& this.internalBusy()) {
501 const workerCreated
= this.createAndSetupWorker()
502 this.registerWorkerMessageListener(workerCreated
, message
=> {
503 const currentWorkerNodeKey
= this.getWorkerNodeKey(workerCreated
)
505 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
506 (message
.kill
!= null &&
507 this.workerNodes
[currentWorkerNodeKey
].tasksUsage
.running
=== 0)
509 // Kill message received from the worker: no new tasks are submitted to that worker for a while ( > maxInactiveTime)
510 this.flushTasksQueue(currentWorkerNodeKey
)
511 void (this.destroyWorker(workerCreated
) as Promise
<void>)
514 workerNodeKey
= this.getWorkerNodeKey(workerCreated
)
516 workerNodeKey
= this.workerChoiceStrategyContext
.execute()
522 * Sends a message to the given worker.
524 * @param worker - The worker which should receive the message.
525 * @param message - The message.
527 protected abstract sendToWorker (
529 message
: MessageValue
<Data
>
533 * Registers a listener callback on the given worker.
535 * @param worker - The worker which should register a listener.
536 * @param listener - The message listener callback.
538 protected abstract registerWorkerMessageListener
<
539 Message
extends Data
| Response
540 >(worker
: Worker
, listener
: (message
: MessageValue
<Message
>) => void): void
543 * Returns a newly created worker.
545 protected abstract createWorker (): Worker
548 * Function that can be hooked up when a worker has been newly created and moved to the pool worker nodes.
550 * Can be used to update the `maxListeners` or binding the `main-worker`\<-\>`worker` connection if not bind by default.
552 * @param worker - The newly created worker.
554 protected abstract afterWorkerSetup (worker
: Worker
): void
557 * Creates a new worker and sets it up completely in the pool worker nodes.
559 * @returns New, completely set up worker.
561 protected createAndSetupWorker (): Worker
{
562 const worker
= this.createWorker()
564 worker
.on('message', this.opts
.messageHandler
?? EMPTY_FUNCTION
)
565 worker
.on('error', this.opts
.errorHandler
?? EMPTY_FUNCTION
)
566 worker
.on('error', error
=> {
567 if (this.emitter
!= null) {
568 this.emitter
.emit(PoolEvents
.error
, error
)
571 if (this.opts
.restartWorkerOnError
=== true) {
572 worker
.on('error', () => {
573 this.createAndSetupWorker()
576 worker
.on('online', this.opts
.onlineHandler
?? EMPTY_FUNCTION
)
577 worker
.on('exit', this.opts
.exitHandler
?? EMPTY_FUNCTION
)
578 worker
.once('exit', () => {
579 this.removeWorkerNode(worker
)
582 this.pushWorkerNode(worker
)
584 this.afterWorkerSetup(worker
)
590 * This function is the listener registered for each worker message.
592 * @returns The listener function to execute when a message is received from a worker.
594 protected workerListener (): (message
: MessageValue
<Response
>) => void {
596 if (message
.id
!= null) {
597 // Task execution response received
598 const promiseResponse
= this.promiseResponseMap
.get(message
.id
)
599 if (promiseResponse
!= null) {
600 if (message
.error
!= null) {
601 promiseResponse
.reject(message
.error
)
603 promiseResponse
.resolve(message
.data
as Response
)
605 this.afterTaskExecutionHook(promiseResponse
.worker
, message
)
606 this.promiseResponseMap
.delete(message
.id
)
607 const workerNodeKey
= this.getWorkerNodeKey(promiseResponse
.worker
)
609 this.opts
.enableTasksQueue
=== true &&
610 this.tasksQueueSize(workerNodeKey
) > 0
614 this.dequeueTask(workerNodeKey
) as Task
<Data
>
622 private checkAndEmitEvents (): void {
623 if (this.emitter
!= null) {
625 this.emitter
?.emit(PoolEvents
.busy
)
627 if (this.type === PoolType
.DYNAMIC
&& this.full
) {
628 this.emitter
?.emit(PoolEvents
.full
)
634 * Sets the given worker node its tasks usage in the pool.
636 * @param workerNode - The worker node.
637 * @param tasksUsage - The worker node tasks usage.
639 private setWorkerNodeTasksUsage (
640 workerNode
: WorkerNode
<Worker
, Data
>,
641 tasksUsage
: TasksUsage
643 workerNode
.tasksUsage
= tasksUsage
647 * Pushes the given worker in the pool worker nodes.
649 * @param worker - The worker.
650 * @returns The worker nodes length.
652 private pushWorkerNode (worker
: Worker
): number {
653 return this.workerNodes
.push({
659 runTimeHistory
: new CircularArray(),
663 waitTimeHistory
: new CircularArray(),
668 tasksQueue
: new Queue
<Task
<Data
>>()
673 * Sets the given worker in the pool worker nodes.
675 * @param workerNodeKey - The worker node key.
676 * @param worker - The worker.
677 * @param tasksUsage - The worker tasks usage.
678 * @param tasksQueue - The worker task queue.
680 private setWorkerNode (
681 workerNodeKey
: number,
683 tasksUsage
: TasksUsage
,
684 tasksQueue
: Queue
<Task
<Data
>>
686 this.workerNodes
[workerNodeKey
] = {
694 * Removes the given worker from the pool worker nodes.
696 * @param worker - The worker.
698 private removeWorkerNode (worker
: Worker
): void {
699 const workerNodeKey
= this.getWorkerNodeKey(worker
)
700 if (workerNodeKey
!== -1) {
701 this.workerNodes
.splice(workerNodeKey
, 1)
702 this.workerChoiceStrategyContext
.remove(workerNodeKey
)
706 private executeTask (workerNodeKey
: number, task
: Task
<Data
>): void {
707 this.beforeTaskExecutionHook(workerNodeKey
)
708 this.sendToWorker(this.workerNodes
[workerNodeKey
].worker
, task
)
711 private enqueueTask (workerNodeKey
: number, task
: Task
<Data
>): number {
712 return this.workerNodes
[workerNodeKey
].tasksQueue
.enqueue(task
)
715 private dequeueTask (workerNodeKey
: number): Task
<Data
> | undefined {
716 return this.workerNodes
[workerNodeKey
].tasksQueue
.dequeue()
719 private tasksQueueSize (workerNodeKey
: number): number {
720 return this.workerNodes
[workerNodeKey
].tasksQueue
.size
723 private flushTasksQueue (workerNodeKey
: number): void {
724 if (this.tasksQueueSize(workerNodeKey
) > 0) {
725 for (let i
= 0; i
< this.tasksQueueSize(workerNodeKey
); i
++) {
728 this.dequeueTask(workerNodeKey
) as Task
<Data
>
734 private flushTasksQueues (): void {
735 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
736 this.flushTasksQueue(workerNodeKey
)