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
,
23 import type { IWorker
, Task
, TasksUsage
, WorkerNode
} from
'./worker'
25 WorkerChoiceStrategies
,
26 type WorkerChoiceStrategy
,
27 type WorkerChoiceStrategyOptions
28 } from
'./selection-strategies/selection-strategies-types'
29 import { WorkerChoiceStrategyContext
} from
'./selection-strategies/worker-choice-strategy-context'
32 * Base class that implements some shared logic for all poolifier pools.
34 * @typeParam Worker - Type of worker which manages this pool.
35 * @typeParam Data - Type of data sent to the worker. This can only be serializable data.
36 * @typeParam Response - Type of execution response. This can only be serializable data.
38 export abstract class AbstractPool
<
39 Worker
extends IWorker
,
42 > implements IPool
<Worker
, Data
, Response
> {
44 public readonly workerNodes
: Array<WorkerNode
<Worker
, Data
>> = []
47 public readonly emitter
?: PoolEmitter
50 * The execution response promise map.
52 * - `key`: The message id of each submitted task.
53 * - `value`: An object that contains the worker, the execution response promise resolve and reject callbacks.
55 * When we receive a message from the worker, we get a map entry with the promise resolve/reject bound to the message id.
57 protected promiseResponseMap
: Map
<
59 PromiseResponseWrapper
<Worker
, Response
>
60 > = new Map
<string, PromiseResponseWrapper
<Worker
, Response
>>()
63 * Worker choice strategy context referencing a worker choice algorithm implementation.
65 * Default to a round robin algorithm.
67 protected workerChoiceStrategyContext
: WorkerChoiceStrategyContext
<
74 * Constructs a new poolifier pool.
76 * @param numberOfWorkers - Number of workers that this pool should manage.
77 * @param filePath - Path to the worker file.
78 * @param opts - Options for the pool.
81 protected readonly numberOfWorkers
: number,
82 protected readonly filePath
: string,
83 protected readonly opts
: PoolOptions
<Worker
>
86 throw new Error('Cannot start a pool from a worker!')
88 this.checkNumberOfWorkers(this.numberOfWorkers
)
89 this.checkFilePath(this.filePath
)
90 this.checkPoolOptions(this.opts
)
92 this.chooseWorkerNode
= this.chooseWorkerNode
.bind(this)
93 this.executeTask
= this.executeTask
.bind(this)
94 this.enqueueTask
= this.enqueueTask
.bind(this)
95 this.checkAndEmitEvents
= this.checkAndEmitEvents
.bind(this)
99 for (let i
= 1; i
<= this.numberOfWorkers
; i
++) {
100 this.createAndSetupWorker()
103 if (this.opts
.enableEvents
=== true) {
104 this.emitter
= new PoolEmitter()
106 this.workerChoiceStrategyContext
= new WorkerChoiceStrategyContext
<
112 this.opts
.workerChoiceStrategy
,
113 this.opts
.workerChoiceStrategyOptions
117 private checkFilePath (filePath
: string): void {
120 (typeof filePath
=== 'string' && filePath
.trim().length
=== 0)
122 throw new Error('Please specify a file with a worker implementation')
126 private checkNumberOfWorkers (numberOfWorkers
: number): void {
127 if (numberOfWorkers
== null) {
129 'Cannot instantiate a pool without specifying the number of workers'
131 } else if (!Number.isSafeInteger(numberOfWorkers
)) {
133 'Cannot instantiate a pool with a non safe integer number of workers'
135 } else if (numberOfWorkers
< 0) {
136 throw new RangeError(
137 'Cannot instantiate a pool with a negative number of workers'
139 } else if (this.type === PoolTypes
.fixed
&& numberOfWorkers
=== 0) {
140 throw new Error('Cannot instantiate a fixed pool with no worker')
144 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
145 if (isPlainObject(opts
)) {
146 this.opts
.workerChoiceStrategy
=
147 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
148 this.checkValidWorkerChoiceStrategy(this.opts
.workerChoiceStrategy
)
149 this.opts
.workerChoiceStrategyOptions
=
150 opts
.workerChoiceStrategyOptions
??
151 DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
152 this.checkValidWorkerChoiceStrategyOptions(
153 this.opts
.workerChoiceStrategyOptions
155 this.opts
.restartWorkerOnError
= opts
.restartWorkerOnError
?? true
156 this.opts
.enableEvents
= opts
.enableEvents
?? true
157 this.opts
.enableTasksQueue
= opts
.enableTasksQueue
?? false
158 if (this.opts
.enableTasksQueue
) {
159 this.checkValidTasksQueueOptions(
160 opts
.tasksQueueOptions
as TasksQueueOptions
162 this.opts
.tasksQueueOptions
= this.buildTasksQueueOptions(
163 opts
.tasksQueueOptions
as TasksQueueOptions
167 throw new TypeError('Invalid pool options: must be a plain object')
171 private checkValidWorkerChoiceStrategy (
172 workerChoiceStrategy
: WorkerChoiceStrategy
174 if (!Object.values(WorkerChoiceStrategies
).includes(workerChoiceStrategy
)) {
176 `Invalid worker choice strategy '${workerChoiceStrategy}'`
181 private checkValidWorkerChoiceStrategyOptions (
182 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
184 if (!isPlainObject(workerChoiceStrategyOptions
)) {
186 'Invalid worker choice strategy options: must be a plain object'
190 workerChoiceStrategyOptions
.weights
!= null &&
191 Object.keys(workerChoiceStrategyOptions
.weights
).length
!== this.maxSize
194 'Invalid worker choice strategy options: must have a weight for each worker node'
199 private checkValidTasksQueueOptions (
200 tasksQueueOptions
: TasksQueueOptions
202 if (tasksQueueOptions
!= null && !isPlainObject(tasksQueueOptions
)) {
203 throw new TypeError('Invalid tasks queue options: must be a plain object')
205 if ((tasksQueueOptions
?.concurrency
as number) <= 0) {
207 `Invalid worker tasks concurrency '${
208 tasksQueueOptions.concurrency as number
215 public get
info (): PoolInfo
{
219 minSize
: this.minSize
,
220 maxSize
: this.maxSize
,
221 workerNodes
: this.workerNodes
.length
,
222 idleWorkerNodes
: this.workerNodes
.reduce(
223 (accumulator
, workerNode
) =>
224 workerNode
.tasksUsage
.running
=== 0 ? accumulator
+ 1 : accumulator
,
227 busyWorkerNodes
: this.workerNodes
.reduce(
228 (accumulator
, workerNode
) =>
229 workerNode
.tasksUsage
.running
> 0 ? accumulator
+ 1 : accumulator
,
232 runningTasks
: this.workerNodes
.reduce(
233 (accumulator
, workerNode
) =>
234 accumulator
+ workerNode
.tasksUsage
.running
,
237 queuedTasks
: this.workerNodes
.reduce(
238 (accumulator
, workerNode
) => accumulator
+ workerNode
.tasksQueue
.size
,
241 maxQueuedTasks
: this.workerNodes
.reduce(
242 (accumulator
, workerNode
) =>
243 accumulator
+ workerNode
.tasksQueue
.maxSize
,
252 * If it is `'dynamic'`, it provides the `max` property.
254 protected abstract get
type (): PoolType
257 * Gets the worker type.
259 protected abstract get
worker (): WorkerType
264 protected abstract get
minSize (): number
269 protected abstract get
maxSize (): number
272 * Gets the given worker its worker node key.
274 * @param worker - The worker.
275 * @returns The worker node key if the worker is found in the pool worker nodes, `-1` otherwise.
277 private getWorkerNodeKey (worker
: Worker
): number {
278 return this.workerNodes
.findIndex(
279 workerNode
=> workerNode
.worker
=== worker
284 public setWorkerChoiceStrategy (
285 workerChoiceStrategy
: WorkerChoiceStrategy
,
286 workerChoiceStrategyOptions
?: WorkerChoiceStrategyOptions
288 this.checkValidWorkerChoiceStrategy(workerChoiceStrategy
)
289 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
290 for (const workerNode
of this.workerNodes
) {
291 this.setWorkerNodeTasksUsage(workerNode
, {
295 runTimeHistory
: new CircularArray(),
299 waitTimeHistory
: new CircularArray(),
305 this.workerChoiceStrategyContext
.setWorkerChoiceStrategy(
306 this.opts
.workerChoiceStrategy
308 if (workerChoiceStrategyOptions
!= null) {
309 this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
314 public setWorkerChoiceStrategyOptions (
315 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
317 this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
318 this.opts
.workerChoiceStrategyOptions
= workerChoiceStrategyOptions
319 this.workerChoiceStrategyContext
.setOptions(
320 this.opts
.workerChoiceStrategyOptions
325 public enableTasksQueue (
327 tasksQueueOptions
?: TasksQueueOptions
329 if (this.opts
.enableTasksQueue
=== true && !enable
) {
330 this.flushTasksQueues()
332 this.opts
.enableTasksQueue
= enable
333 this.setTasksQueueOptions(tasksQueueOptions
as TasksQueueOptions
)
337 public setTasksQueueOptions (tasksQueueOptions
: TasksQueueOptions
): void {
338 if (this.opts
.enableTasksQueue
=== true) {
339 this.checkValidTasksQueueOptions(tasksQueueOptions
)
340 this.opts
.tasksQueueOptions
=
341 this.buildTasksQueueOptions(tasksQueueOptions
)
343 delete this.opts
.tasksQueueOptions
347 private buildTasksQueueOptions (
348 tasksQueueOptions
: TasksQueueOptions
349 ): TasksQueueOptions
{
351 concurrency
: tasksQueueOptions
?.concurrency
?? 1
356 * Whether the pool is full or not.
358 * The pool filling boolean status.
360 protected get
full (): boolean {
361 return this.workerNodes
.length
>= this.maxSize
365 * Whether the pool is busy or not.
367 * The pool busyness boolean status.
369 protected abstract get
busy (): boolean
371 protected internalBusy (): boolean {
373 this.workerNodes
.findIndex(workerNode
=> {
374 return workerNode
.tasksUsage
.running
=== 0
380 public async execute (data
?: Data
, name
?: string): Promise
<Response
> {
381 const submissionTimestamp
= performance
.now()
382 const workerNodeKey
= this.chooseWorkerNode()
383 const submittedTask
: Task
<Data
> = {
385 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
386 data
: data
?? ({} as Data
),
388 id
: crypto
.randomUUID()
390 const res
= new Promise
<Response
>((resolve
, reject
) => {
391 this.promiseResponseMap
.set(submittedTask
.id
as string, {
394 worker
: this.workerNodes
[workerNodeKey
].worker
398 this.opts
.enableTasksQueue
=== true &&
400 this.workerNodes
[workerNodeKey
].tasksUsage
.running
>=
401 ((this.opts
.tasksQueueOptions
as TasksQueueOptions
)
402 .concurrency
as number))
404 this.enqueueTask(workerNodeKey
, submittedTask
)
406 this.executeTask(workerNodeKey
, submittedTask
)
408 this.workerChoiceStrategyContext
.update(workerNodeKey
)
409 this.checkAndEmitEvents()
410 // eslint-disable-next-line @typescript-eslint/return-await
415 public async destroy (): Promise
<void> {
417 this.workerNodes
.map(async (workerNode
, workerNodeKey
) => {
418 this.flushTasksQueue(workerNodeKey
)
419 // FIXME: wait for tasks to be finished
420 await this.destroyWorker(workerNode
.worker
)
426 * Shutdowns the given worker.
428 * @param worker - A worker within `workerNodes`.
430 protected abstract destroyWorker (worker
: Worker
): void | Promise
<void>
433 * Setup hook to execute code before worker node are created in the abstract constructor.
438 protected setupHook (): void {
439 // Intentionally empty
443 * Should return whether the worker is the main worker or not.
445 protected abstract isMain (): boolean
448 * Hook executed before the worker task execution.
451 * @param workerNodeKey - The worker node key.
453 protected beforeTaskExecutionHook (workerNodeKey
: number): void {
454 ++this.workerNodes
[workerNodeKey
].tasksUsage
.running
458 * Hook executed after the worker task execution.
461 * @param worker - The worker.
462 * @param message - The received message.
464 protected afterTaskExecutionHook (
466 message
: MessageValue
<Response
>
468 const workerTasksUsage
=
469 this.workerNodes
[this.getWorkerNodeKey(worker
)].tasksUsage
470 --workerTasksUsage
.running
471 ++workerTasksUsage
.run
472 if (message
.error
!= null) {
473 ++workerTasksUsage
.error
475 this.updateRunTimeTasksUsage(workerTasksUsage
, message
)
476 this.updateWaitTimeTasksUsage(workerTasksUsage
, message
)
479 private updateRunTimeTasksUsage (
480 workerTasksUsage
: TasksUsage
,
481 message
: MessageValue
<Response
>
483 if (this.workerChoiceStrategyContext
.getRequiredStatistics().runTime
) {
484 workerTasksUsage
.runTime
+= message
.runTime
?? 0
486 this.workerChoiceStrategyContext
.getRequiredStatistics().avgRunTime
&&
487 workerTasksUsage
.run
!== 0
489 workerTasksUsage
.avgRunTime
=
490 workerTasksUsage
.runTime
/ workerTasksUsage
.run
493 this.workerChoiceStrategyContext
.getRequiredStatistics().medRunTime
&&
494 message
.runTime
!= null
496 workerTasksUsage
.runTimeHistory
.push(message
.runTime
)
497 workerTasksUsage
.medRunTime
= median(workerTasksUsage
.runTimeHistory
)
502 private updateWaitTimeTasksUsage (
503 workerTasksUsage
: TasksUsage
,
504 message
: MessageValue
<Response
>
506 if (this.workerChoiceStrategyContext
.getRequiredStatistics().waitTime
) {
507 workerTasksUsage
.waitTime
+= message
.waitTime
?? 0
509 this.workerChoiceStrategyContext
.getRequiredStatistics().avgWaitTime
&&
510 workerTasksUsage
.run
!== 0
512 workerTasksUsage
.avgWaitTime
=
513 workerTasksUsage
.waitTime
/ workerTasksUsage
.run
516 this.workerChoiceStrategyContext
.getRequiredStatistics().medWaitTime
&&
517 message
.waitTime
!= null
519 workerTasksUsage
.waitTimeHistory
.push(message
.waitTime
)
520 workerTasksUsage
.medWaitTime
= median(workerTasksUsage
.waitTimeHistory
)
526 * Chooses a worker node for the next task.
528 * The default worker choice strategy uses a round robin algorithm to distribute the load.
530 * @returns The worker node key
532 protected chooseWorkerNode (): number {
533 let workerNodeKey
: number
534 if (this.type === PoolTypes
.dynamic
&& !this.full
&& this.internalBusy()) {
535 const workerCreated
= this.createAndSetupWorker()
536 this.registerWorkerMessageListener(workerCreated
, message
=> {
537 const currentWorkerNodeKey
= this.getWorkerNodeKey(workerCreated
)
539 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
540 (message
.kill
!= null &&
541 this.workerNodes
[currentWorkerNodeKey
].tasksUsage
.running
=== 0)
543 // Kill message received from the worker: no new tasks are submitted to that worker for a while ( > maxInactiveTime)
544 this.flushTasksQueue(currentWorkerNodeKey
)
545 // FIXME: wait for tasks to be finished
546 void (this.destroyWorker(workerCreated
) as Promise
<void>)
549 workerNodeKey
= this.getWorkerNodeKey(workerCreated
)
551 workerNodeKey
= this.workerChoiceStrategyContext
.execute()
557 * Sends a message to the given worker.
559 * @param worker - The worker which should receive the message.
560 * @param message - The message.
562 protected abstract sendToWorker (
564 message
: MessageValue
<Data
>
568 * Registers a listener callback on the given worker.
570 * @param worker - The worker which should register a listener.
571 * @param listener - The message listener callback.
573 protected abstract registerWorkerMessageListener
<
574 Message
extends Data
| Response
575 >(worker
: Worker
, listener
: (message
: MessageValue
<Message
>) => void): void
578 * Returns a newly created worker.
580 protected abstract createWorker (): Worker
583 * Function that can be hooked up when a worker has been newly created and moved to the pool worker nodes.
585 * Can be used to update the `maxListeners` or binding the `main-worker`\<-\>`worker` connection if not bind by default.
587 * @param worker - The newly created worker.
589 protected abstract afterWorkerSetup (worker
: Worker
): void
592 * Creates a new worker and sets it up completely in the pool worker nodes.
594 * @returns New, completely set up worker.
596 protected createAndSetupWorker (): Worker
{
597 const worker
= this.createWorker()
599 worker
.on('message', this.opts
.messageHandler
?? EMPTY_FUNCTION
)
600 worker
.on('error', this.opts
.errorHandler
?? EMPTY_FUNCTION
)
601 worker
.on('error', error
=> {
602 if (this.emitter
!= null) {
603 this.emitter
.emit(PoolEvents
.error
, error
)
606 if (this.opts
.restartWorkerOnError
=== true) {
607 worker
.on('error', () => {
608 this.createAndSetupWorker()
611 worker
.on('online', this.opts
.onlineHandler
?? EMPTY_FUNCTION
)
612 worker
.on('exit', this.opts
.exitHandler
?? EMPTY_FUNCTION
)
613 worker
.once('exit', () => {
614 this.removeWorkerNode(worker
)
617 this.pushWorkerNode(worker
)
619 this.afterWorkerSetup(worker
)
625 * This function is the listener registered for each worker message.
627 * @returns The listener function to execute when a message is received from a worker.
629 protected workerListener (): (message
: MessageValue
<Response
>) => void {
631 if (message
.id
!= null) {
632 // Task execution response received
633 const promiseResponse
= this.promiseResponseMap
.get(message
.id
)
634 if (promiseResponse
!= null) {
635 if (message
.error
!= null) {
636 promiseResponse
.reject(message
.error
)
637 if (this.emitter
!= null) {
638 this.emitter
.emit(PoolEvents
.taskError
, {
639 error
: message
.error
,
640 errorData
: message
.errorData
644 promiseResponse
.resolve(message
.data
as Response
)
646 this.afterTaskExecutionHook(promiseResponse
.worker
, message
)
647 this.promiseResponseMap
.delete(message
.id
)
648 const workerNodeKey
= this.getWorkerNodeKey(promiseResponse
.worker
)
650 this.opts
.enableTasksQueue
=== true &&
651 this.tasksQueueSize(workerNodeKey
) > 0
655 this.dequeueTask(workerNodeKey
) as Task
<Data
>
663 private checkAndEmitEvents (): void {
664 if (this.emitter
!= null) {
666 this.emitter
?.emit(PoolEvents
.busy
, this.info
)
668 if (this.type === PoolTypes
.dynamic
&& this.full
) {
669 this.emitter
?.emit(PoolEvents
.full
, this.info
)
675 * Sets the given worker node its tasks usage in the pool.
677 * @param workerNode - The worker node.
678 * @param tasksUsage - The worker node tasks usage.
680 private setWorkerNodeTasksUsage (
681 workerNode
: WorkerNode
<Worker
, Data
>,
682 tasksUsage
: TasksUsage
684 workerNode
.tasksUsage
= tasksUsage
688 * Pushes the given worker in the pool worker nodes.
690 * @param worker - The worker.
691 * @returns The worker nodes length.
693 private pushWorkerNode (worker
: Worker
): number {
694 return this.workerNodes
.push({
700 runTimeHistory
: new CircularArray(),
704 waitTimeHistory
: new CircularArray(),
709 tasksQueue
: new Queue
<Task
<Data
>>()
714 * Sets the given worker in the pool worker nodes.
716 * @param workerNodeKey - The worker node key.
717 * @param worker - The worker.
718 * @param tasksUsage - The worker tasks usage.
719 * @param tasksQueue - The worker task queue.
721 private setWorkerNode (
722 workerNodeKey
: number,
724 tasksUsage
: TasksUsage
,
725 tasksQueue
: Queue
<Task
<Data
>>
727 this.workerNodes
[workerNodeKey
] = {
735 * Removes the given worker from the pool worker nodes.
737 * @param worker - The worker.
739 private removeWorkerNode (worker
: Worker
): void {
740 const workerNodeKey
= this.getWorkerNodeKey(worker
)
741 if (workerNodeKey
!== -1) {
742 this.workerNodes
.splice(workerNodeKey
, 1)
743 this.workerChoiceStrategyContext
.remove(workerNodeKey
)
747 private executeTask (workerNodeKey
: number, task
: Task
<Data
>): void {
748 this.beforeTaskExecutionHook(workerNodeKey
)
749 this.sendToWorker(this.workerNodes
[workerNodeKey
].worker
, task
)
752 private enqueueTask (workerNodeKey
: number, task
: Task
<Data
>): number {
753 return this.workerNodes
[workerNodeKey
].tasksQueue
.enqueue(task
)
756 private dequeueTask (workerNodeKey
: number): Task
<Data
> | undefined {
757 return this.workerNodes
[workerNodeKey
].tasksQueue
.dequeue()
760 private tasksQueueSize (workerNodeKey
: number): number {
761 return this.workerNodes
[workerNodeKey
].tasksQueue
.size
764 private flushTasksQueue (workerNodeKey
: number): void {
765 if (this.tasksQueueSize(workerNodeKey
) > 0) {
766 for (let i
= 0; i
< this.tasksQueueSize(workerNodeKey
); i
++) {
769 this.dequeueTask(workerNodeKey
) as Task
<Data
>
775 private flushTasksQueues (): void {
776 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
777 this.flushTasksQueue(workerNodeKey
)