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'
20 type TasksQueueOptions
22 import type { IWorker
, Task
, TasksUsage
, WorkerNode
} from
'./worker'
24 WorkerChoiceStrategies
,
25 type WorkerChoiceStrategy
,
26 type WorkerChoiceStrategyOptions
27 } from
'./selection-strategies/selection-strategies-types'
28 import { WorkerChoiceStrategyContext
} from
'./selection-strategies/worker-choice-strategy-context'
31 * Base class that implements some shared logic for all poolifier pools.
33 * @typeParam Worker - Type of worker which manages this pool.
34 * @typeParam Data - Type of data sent to the worker. This can only be serializable data.
35 * @typeParam Response - Type of execution response. This can only be serializable data.
37 export abstract class AbstractPool
<
38 Worker
extends IWorker
,
41 > implements IPool
<Worker
, Data
, Response
> {
43 public readonly workerNodes
: Array<WorkerNode
<Worker
, Data
>> = []
46 public readonly emitter
?: PoolEmitter
49 * The execution response promise map.
51 * - `key`: The message id of each submitted task.
52 * - `value`: An object that contains the worker, the execution response promise resolve and reject callbacks.
54 * When we receive a message from the worker, we get a map entry with the promise resolve/reject bound to the message id.
56 protected promiseResponseMap
: Map
<
58 PromiseResponseWrapper
<Worker
, Response
>
59 > = new Map
<string, PromiseResponseWrapper
<Worker
, Response
>>()
62 * Worker choice strategy context referencing a worker choice algorithm implementation.
64 * Default to a round robin algorithm.
66 protected workerChoiceStrategyContext
: WorkerChoiceStrategyContext
<
73 * Constructs a new poolifier pool.
75 * @param numberOfWorkers - Number of workers that this pool should manage.
76 * @param filePath - Path to the worker file.
77 * @param opts - Options for the pool.
80 public readonly numberOfWorkers
: number,
81 public readonly filePath
: string,
82 public readonly opts
: PoolOptions
<Worker
>
85 throw new Error('Cannot start a pool from a worker!')
87 this.checkNumberOfWorkers(this.numberOfWorkers
)
88 this.checkFilePath(this.filePath
)
89 this.checkPoolOptions(this.opts
)
91 this.chooseWorkerNode
= this.chooseWorkerNode
.bind(this)
92 this.executeTask
= this.executeTask
.bind(this)
93 this.enqueueTask
= this.enqueueTask
.bind(this)
94 this.checkAndEmitEvents
= this.checkAndEmitEvents
.bind(this)
98 for (let i
= 1; i
<= this.numberOfWorkers
; i
++) {
99 this.createAndSetupWorker()
102 if (this.opts
.enableEvents
=== true) {
103 this.emitter
= new PoolEmitter()
105 this.workerChoiceStrategyContext
= new WorkerChoiceStrategyContext
<
111 this.opts
.workerChoiceStrategy
,
112 this.opts
.workerChoiceStrategyOptions
116 private checkFilePath (filePath
: string): void {
119 (typeof filePath
=== 'string' && filePath
.trim().length
=== 0)
121 throw new Error('Please specify a file with a worker implementation')
125 private checkNumberOfWorkers (numberOfWorkers
: number): void {
126 if (numberOfWorkers
== null) {
128 'Cannot instantiate a pool without specifying the number of workers'
130 } else if (!Number.isSafeInteger(numberOfWorkers
)) {
132 'Cannot instantiate a pool with a non safe integer number of workers'
134 } else if (numberOfWorkers
< 0) {
135 throw new RangeError(
136 'Cannot instantiate a pool with a negative number of workers'
138 } else if (this.type === PoolTypes
.fixed
&& numberOfWorkers
=== 0) {
139 throw new Error('Cannot instantiate a fixed pool with no worker')
143 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
144 if (isPlainObject(opts
)) {
145 this.opts
.workerChoiceStrategy
=
146 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
147 this.checkValidWorkerChoiceStrategy(this.opts
.workerChoiceStrategy
)
148 this.opts
.workerChoiceStrategyOptions
=
149 opts
.workerChoiceStrategyOptions
??
150 DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
151 this.checkValidWorkerChoiceStrategyOptions(
152 this.opts
.workerChoiceStrategyOptions
154 this.opts
.restartWorkerOnError
= opts
.restartWorkerOnError
?? true
155 this.opts
.enableEvents
= opts
.enableEvents
?? true
156 this.opts
.enableTasksQueue
= opts
.enableTasksQueue
?? false
157 if (this.opts
.enableTasksQueue
) {
158 this.checkValidTasksQueueOptions(
159 opts
.tasksQueueOptions
as TasksQueueOptions
161 this.opts
.tasksQueueOptions
= this.buildTasksQueueOptions(
162 opts
.tasksQueueOptions
as TasksQueueOptions
166 throw new TypeError('Invalid pool options: must be a plain object')
170 private checkValidWorkerChoiceStrategy (
171 workerChoiceStrategy
: WorkerChoiceStrategy
173 if (!Object.values(WorkerChoiceStrategies
).includes(workerChoiceStrategy
)) {
175 `Invalid worker choice strategy '${workerChoiceStrategy}'`
180 private checkValidWorkerChoiceStrategyOptions (
181 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
183 if (!isPlainObject(workerChoiceStrategyOptions
)) {
185 'Invalid worker choice strategy options: must be a plain object'
189 workerChoiceStrategyOptions
.weights
!= null &&
190 Object.keys(workerChoiceStrategyOptions
.weights
).length
!== this.maxSize
193 'Invalid worker choice strategy options: must have a weight for each worker node'
198 private checkValidTasksQueueOptions (
199 tasksQueueOptions
: TasksQueueOptions
201 if (tasksQueueOptions
!= null && !isPlainObject(tasksQueueOptions
)) {
202 throw new TypeError('Invalid tasks queue options: must be a plain object')
204 if ((tasksQueueOptions
?.concurrency
as number) <= 0) {
206 `Invalid worker tasks concurrency '${
207 tasksQueueOptions.concurrency as number
214 public abstract get
type (): PoolType
217 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
.tasksUsage
.running
=== 0 ? accumulator
+ 1 : accumulator
,
228 busyWorkerNodes
: this.workerNodes
.reduce(
229 (accumulator
, workerNode
) =>
230 workerNode
.tasksUsage
.running
> 0 ? accumulator
+ 1 : accumulator
,
233 runningTasks
: this.workerNodes
.reduce(
234 (accumulator
, workerNode
) =>
235 accumulator
+ workerNode
.tasksUsage
.running
,
238 queuedTasks
: this.workerNodes
.reduce(
239 (accumulator
, workerNode
) => accumulator
+ workerNode
.tasksQueue
.size
,
242 maxQueuedTasks
: this.workerNodes
.reduce(
243 (accumulator
, workerNode
) =>
244 accumulator
+ workerNode
.tasksQueue
.maxSize
,
253 protected abstract get
minSize (): number
258 protected abstract get
maxSize (): number
261 * Gets the given worker its worker node key.
263 * @param worker - The worker.
264 * @returns The worker node key if the worker is found in the pool worker nodes, `-1` otherwise.
266 private getWorkerNodeKey (worker
: Worker
): number {
267 return this.workerNodes
.findIndex(
268 workerNode
=> workerNode
.worker
=== worker
273 public setWorkerChoiceStrategy (
274 workerChoiceStrategy
: WorkerChoiceStrategy
,
275 workerChoiceStrategyOptions
?: WorkerChoiceStrategyOptions
277 this.checkValidWorkerChoiceStrategy(workerChoiceStrategy
)
278 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
279 for (const workerNode
of this.workerNodes
) {
280 this.setWorkerNodeTasksUsage(workerNode
, {
284 runTimeHistory
: new CircularArray(),
288 waitTimeHistory
: new CircularArray(),
294 this.workerChoiceStrategyContext
.setWorkerChoiceStrategy(
295 this.opts
.workerChoiceStrategy
297 if (workerChoiceStrategyOptions
!= null) {
298 this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
303 public setWorkerChoiceStrategyOptions (
304 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
306 this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
307 this.opts
.workerChoiceStrategyOptions
= workerChoiceStrategyOptions
308 this.workerChoiceStrategyContext
.setOptions(
309 this.opts
.workerChoiceStrategyOptions
314 public enableTasksQueue (
316 tasksQueueOptions
?: TasksQueueOptions
318 if (this.opts
.enableTasksQueue
=== true && !enable
) {
319 this.flushTasksQueues()
321 this.opts
.enableTasksQueue
= enable
322 this.setTasksQueueOptions(tasksQueueOptions
as TasksQueueOptions
)
326 public setTasksQueueOptions (tasksQueueOptions
: TasksQueueOptions
): void {
327 if (this.opts
.enableTasksQueue
=== true) {
328 this.checkValidTasksQueueOptions(tasksQueueOptions
)
329 this.opts
.tasksQueueOptions
=
330 this.buildTasksQueueOptions(tasksQueueOptions
)
332 delete this.opts
.tasksQueueOptions
336 private buildTasksQueueOptions (
337 tasksQueueOptions
: TasksQueueOptions
338 ): TasksQueueOptions
{
340 concurrency
: tasksQueueOptions
?.concurrency
?? 1
345 * Whether the pool is full or not.
347 * The pool filling boolean status.
349 protected abstract get
full (): boolean
352 * Whether the pool is busy or not.
354 * The pool busyness boolean status.
356 protected abstract get
busy (): boolean
358 protected internalBusy (): boolean {
360 this.workerNodes
.findIndex(workerNode
=> {
361 return workerNode
.tasksUsage
.running
=== 0
367 public async execute (data
?: Data
, name
?: string): Promise
<Response
> {
368 const submissionTimestamp
= performance
.now()
369 const workerNodeKey
= this.chooseWorkerNode()
370 const submittedTask
: Task
<Data
> = {
372 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
373 data
: data
?? ({} as Data
),
375 id
: crypto
.randomUUID()
377 const res
= new Promise
<Response
>((resolve
, reject
) => {
378 this.promiseResponseMap
.set(submittedTask
.id
as string, {
381 worker
: this.workerNodes
[workerNodeKey
].worker
385 this.opts
.enableTasksQueue
=== true &&
387 this.workerNodes
[workerNodeKey
].tasksUsage
.running
>=
388 ((this.opts
.tasksQueueOptions
as TasksQueueOptions
)
389 .concurrency
as number))
391 this.enqueueTask(workerNodeKey
, submittedTask
)
393 this.executeTask(workerNodeKey
, submittedTask
)
395 this.workerChoiceStrategyContext
.update(workerNodeKey
)
396 this.checkAndEmitEvents()
397 // eslint-disable-next-line @typescript-eslint/return-await
402 public async destroy (): Promise
<void> {
404 this.workerNodes
.map(async (workerNode
, workerNodeKey
) => {
405 this.flushTasksQueue(workerNodeKey
)
406 await this.destroyWorker(workerNode
.worker
)
412 * Shutdowns the given worker.
414 * @param worker - A worker within `workerNodes`.
416 protected abstract destroyWorker (worker
: Worker
): void | Promise
<void>
419 * Setup hook to execute code before worker node are created in the abstract constructor.
424 protected setupHook (): void {
425 // Intentionally empty
429 * Should return whether the worker is the main worker or not.
431 protected abstract isMain (): boolean
434 * Hook executed before the worker task execution.
437 * @param workerNodeKey - The worker node key.
439 protected beforeTaskExecutionHook (workerNodeKey
: number): void {
440 ++this.workerNodes
[workerNodeKey
].tasksUsage
.running
444 * Hook executed after the worker task execution.
447 * @param worker - The worker.
448 * @param message - The received message.
450 protected afterTaskExecutionHook (
452 message
: MessageValue
<Response
>
454 const workerTasksUsage
=
455 this.workerNodes
[this.getWorkerNodeKey(worker
)].tasksUsage
456 --workerTasksUsage
.running
457 ++workerTasksUsage
.run
458 if (message
.error
!= null) {
459 ++workerTasksUsage
.error
461 this.updateRunTimeTasksUsage(workerTasksUsage
, message
)
462 this.updateWaitTimeTasksUsage(workerTasksUsage
, message
)
465 private updateRunTimeTasksUsage (
466 workerTasksUsage
: TasksUsage
,
467 message
: MessageValue
<Response
>
469 if (this.workerChoiceStrategyContext
.getRequiredStatistics().runTime
) {
470 workerTasksUsage
.runTime
+= message
.runTime
?? 0
472 this.workerChoiceStrategyContext
.getRequiredStatistics().avgRunTime
&&
473 workerTasksUsage
.run
!== 0
475 workerTasksUsage
.avgRunTime
=
476 workerTasksUsage
.runTime
/ workerTasksUsage
.run
479 this.workerChoiceStrategyContext
.getRequiredStatistics().medRunTime
&&
480 message
.runTime
!= null
482 workerTasksUsage
.runTimeHistory
.push(message
.runTime
)
483 workerTasksUsage
.medRunTime
= median(workerTasksUsage
.runTimeHistory
)
488 private updateWaitTimeTasksUsage (
489 workerTasksUsage
: TasksUsage
,
490 message
: MessageValue
<Response
>
492 if (this.workerChoiceStrategyContext
.getRequiredStatistics().waitTime
) {
493 workerTasksUsage
.waitTime
+= message
.waitTime
?? 0
495 this.workerChoiceStrategyContext
.getRequiredStatistics().avgWaitTime
&&
496 workerTasksUsage
.run
!== 0
498 workerTasksUsage
.avgWaitTime
=
499 workerTasksUsage
.waitTime
/ workerTasksUsage
.run
502 this.workerChoiceStrategyContext
.getRequiredStatistics().medWaitTime
&&
503 message
.waitTime
!= null
505 workerTasksUsage
.waitTimeHistory
.push(message
.waitTime
)
506 workerTasksUsage
.medWaitTime
= median(workerTasksUsage
.waitTimeHistory
)
512 * Chooses a worker node for the next task.
514 * The default worker choice strategy uses a round robin algorithm to distribute the load.
516 * @returns The worker node key
518 protected chooseWorkerNode (): number {
519 let workerNodeKey
: number
520 if (this.type === PoolTypes
.dynamic
&& !this.full
&& this.internalBusy()) {
521 const workerCreated
= this.createAndSetupWorker()
522 this.registerWorkerMessageListener(workerCreated
, message
=> {
523 const currentWorkerNodeKey
= this.getWorkerNodeKey(workerCreated
)
525 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
526 (message
.kill
!= null &&
527 this.workerNodes
[currentWorkerNodeKey
].tasksUsage
.running
=== 0)
529 // Kill message received from the worker: no new tasks are submitted to that worker for a while ( > maxInactiveTime)
530 this.flushTasksQueue(currentWorkerNodeKey
)
531 void (this.destroyWorker(workerCreated
) as Promise
<void>)
534 workerNodeKey
= this.getWorkerNodeKey(workerCreated
)
536 workerNodeKey
= this.workerChoiceStrategyContext
.execute()
542 * Sends a message to the given worker.
544 * @param worker - The worker which should receive the message.
545 * @param message - The message.
547 protected abstract sendToWorker (
549 message
: MessageValue
<Data
>
553 * Registers a listener callback on the given worker.
555 * @param worker - The worker which should register a listener.
556 * @param listener - The message listener callback.
558 protected abstract registerWorkerMessageListener
<
559 Message
extends Data
| Response
560 >(worker
: Worker
, listener
: (message
: MessageValue
<Message
>) => void): void
563 * Returns a newly created worker.
565 protected abstract createWorker (): Worker
568 * Function that can be hooked up when a worker has been newly created and moved to the pool worker nodes.
570 * Can be used to update the `maxListeners` or binding the `main-worker`\<-\>`worker` connection if not bind by default.
572 * @param worker - The newly created worker.
574 protected abstract afterWorkerSetup (worker
: Worker
): void
577 * Creates a new worker and sets it up completely in the pool worker nodes.
579 * @returns New, completely set up worker.
581 protected createAndSetupWorker (): Worker
{
582 const worker
= this.createWorker()
584 worker
.on('message', this.opts
.messageHandler
?? EMPTY_FUNCTION
)
585 worker
.on('error', this.opts
.errorHandler
?? EMPTY_FUNCTION
)
586 worker
.on('error', error
=> {
587 if (this.emitter
!= null) {
588 this.emitter
.emit(PoolEvents
.error
, error
)
591 if (this.opts
.restartWorkerOnError
=== true) {
592 worker
.on('error', () => {
593 this.createAndSetupWorker()
596 worker
.on('online', this.opts
.onlineHandler
?? EMPTY_FUNCTION
)
597 worker
.on('exit', this.opts
.exitHandler
?? EMPTY_FUNCTION
)
598 worker
.once('exit', () => {
599 this.removeWorkerNode(worker
)
602 this.pushWorkerNode(worker
)
604 this.afterWorkerSetup(worker
)
610 * This function is the listener registered for each worker message.
612 * @returns The listener function to execute when a message is received from a worker.
614 protected workerListener (): (message
: MessageValue
<Response
>) => void {
616 if (message
.id
!= null) {
617 // Task execution response received
618 const promiseResponse
= this.promiseResponseMap
.get(message
.id
)
619 if (promiseResponse
!= null) {
620 if (message
.error
!= null) {
621 promiseResponse
.reject(message
.error
)
622 if (this.emitter
!= null) {
623 this.emitter
.emit(PoolEvents
.taskError
, {
624 error
: message
.error
,
625 errorData
: message
.errorData
629 promiseResponse
.resolve(message
.data
as Response
)
631 this.afterTaskExecutionHook(promiseResponse
.worker
, message
)
632 this.promiseResponseMap
.delete(message
.id
)
633 const workerNodeKey
= this.getWorkerNodeKey(promiseResponse
.worker
)
635 this.opts
.enableTasksQueue
=== true &&
636 this.tasksQueueSize(workerNodeKey
) > 0
640 this.dequeueTask(workerNodeKey
) as Task
<Data
>
648 private checkAndEmitEvents (): void {
649 if (this.emitter
!= null) {
651 this.emitter
?.emit(PoolEvents
.busy
, this.info
)
653 if (this.type === PoolTypes
.dynamic
&& this.full
) {
654 this.emitter
?.emit(PoolEvents
.full
, this.info
)
660 * Sets the given worker node its tasks usage in the pool.
662 * @param workerNode - The worker node.
663 * @param tasksUsage - The worker node tasks usage.
665 private setWorkerNodeTasksUsage (
666 workerNode
: WorkerNode
<Worker
, Data
>,
667 tasksUsage
: TasksUsage
669 workerNode
.tasksUsage
= tasksUsage
673 * Pushes the given worker in the pool worker nodes.
675 * @param worker - The worker.
676 * @returns The worker nodes length.
678 private pushWorkerNode (worker
: Worker
): number {
679 return this.workerNodes
.push({
685 runTimeHistory
: new CircularArray(),
689 waitTimeHistory
: new CircularArray(),
694 tasksQueue
: new Queue
<Task
<Data
>>()
699 * Sets the given worker in the pool worker nodes.
701 * @param workerNodeKey - The worker node key.
702 * @param worker - The worker.
703 * @param tasksUsage - The worker tasks usage.
704 * @param tasksQueue - The worker task queue.
706 private setWorkerNode (
707 workerNodeKey
: number,
709 tasksUsage
: TasksUsage
,
710 tasksQueue
: Queue
<Task
<Data
>>
712 this.workerNodes
[workerNodeKey
] = {
720 * Removes the given worker from the pool worker nodes.
722 * @param worker - The worker.
724 private removeWorkerNode (worker
: Worker
): void {
725 const workerNodeKey
= this.getWorkerNodeKey(worker
)
726 if (workerNodeKey
!== -1) {
727 this.workerNodes
.splice(workerNodeKey
, 1)
728 this.workerChoiceStrategyContext
.remove(workerNodeKey
)
732 private executeTask (workerNodeKey
: number, task
: Task
<Data
>): void {
733 this.beforeTaskExecutionHook(workerNodeKey
)
734 this.sendToWorker(this.workerNodes
[workerNodeKey
].worker
, task
)
737 private enqueueTask (workerNodeKey
: number, task
: Task
<Data
>): number {
738 return this.workerNodes
[workerNodeKey
].tasksQueue
.enqueue(task
)
741 private dequeueTask (workerNodeKey
: number): Task
<Data
> | undefined {
742 return this.workerNodes
[workerNodeKey
].tasksQueue
.dequeue()
745 private tasksQueueSize (workerNodeKey
: number): number {
746 return this.workerNodes
[workerNodeKey
].tasksQueue
.size
749 private flushTasksQueue (workerNodeKey
: number): void {
750 if (this.tasksQueueSize(workerNodeKey
) > 0) {
751 for (let i
= 0; i
< this.tasksQueueSize(workerNodeKey
); i
++) {
754 this.dequeueTask(workerNodeKey
) as Task
<Data
>
760 private flushTasksQueues (): void {
761 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
762 this.flushTasksQueue(workerNodeKey
)