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'
9 import { CircularArray
} from
'../circular-array'
10 import { Queue
} from
'../queue'
17 type TasksQueueOptions
19 import type { IWorker
, Task
, TasksUsage
, WorkerNode
} from
'./worker'
21 WorkerChoiceStrategies
,
22 type WorkerChoiceStrategy
,
23 type WorkerChoiceStrategyOptions
24 } from
'./selection-strategies/selection-strategies-types'
25 import { WorkerChoiceStrategyContext
} from
'./selection-strategies/worker-choice-strategy-context'
28 * Base class that implements some shared logic for all poolifier pools.
30 * @typeParam Worker - Type of worker which manages this pool.
31 * @typeParam Data - Type of data sent to the worker. This can only be serializable data.
32 * @typeParam Response - Type of execution response. This can only be serializable data.
34 export abstract class AbstractPool
<
35 Worker
extends IWorker
,
38 > implements IPool
<Worker
, Data
, Response
> {
40 public readonly workerNodes
: Array<WorkerNode
<Worker
, Data
>> = []
43 public readonly emitter
?: PoolEmitter
46 * The execution response promise map.
48 * - `key`: The message id of each submitted task.
49 * - `value`: An object that contains the worker, the execution response promise resolve and reject callbacks.
51 * When we receive a message from the worker, we get a map entry with the promise resolve/reject bound to the message id.
53 protected promiseResponseMap
: Map
<
55 PromiseResponseWrapper
<Worker
, Response
>
56 > = new Map
<string, PromiseResponseWrapper
<Worker
, Response
>>()
59 * Worker choice strategy context referencing a worker choice algorithm implementation.
61 * Default to a round robin algorithm.
63 protected workerChoiceStrategyContext
: WorkerChoiceStrategyContext
<
70 * Constructs a new poolifier pool.
72 * @param numberOfWorkers - Number of workers that this pool should manage.
73 * @param filePath - Path to the worker file.
74 * @param opts - Options for the pool.
77 public readonly numberOfWorkers
: number,
78 public readonly filePath
: string,
79 public readonly opts
: PoolOptions
<Worker
>
82 throw new Error('Cannot start a pool from a worker!')
84 this.checkNumberOfWorkers(this.numberOfWorkers
)
85 this.checkFilePath(this.filePath
)
86 this.checkPoolOptions(this.opts
)
88 this.chooseWorkerNode
= this.chooseWorkerNode
.bind(this)
89 this.executeTask
= this.executeTask
.bind(this)
90 this.enqueueTask
= this.enqueueTask
.bind(this)
91 this.checkAndEmitEvents
= this.checkAndEmitEvents
.bind(this)
95 for (let i
= 1; i
<= this.numberOfWorkers
; i
++) {
96 this.createAndSetupWorker()
99 if (this.opts
.enableEvents
=== true) {
100 this.emitter
= new PoolEmitter()
102 this.workerChoiceStrategyContext
= new WorkerChoiceStrategyContext
<
108 this.opts
.workerChoiceStrategy
,
109 this.opts
.workerChoiceStrategyOptions
113 private checkFilePath (filePath
: string): void {
116 (typeof filePath
=== 'string' && filePath
.trim().length
=== 0)
118 throw new Error('Please specify a file with a worker implementation')
122 private checkNumberOfWorkers (numberOfWorkers
: number): void {
123 if (numberOfWorkers
== null) {
125 'Cannot instantiate a pool without specifying the number of workers'
127 } else if (!Number.isSafeInteger(numberOfWorkers
)) {
129 'Cannot instantiate a pool with a non integer number of workers'
131 } else if (numberOfWorkers
< 0) {
132 throw new RangeError(
133 'Cannot instantiate a pool with a negative number of workers'
135 } else if (this.type === PoolType
.FIXED
&& numberOfWorkers
=== 0) {
136 throw new Error('Cannot instantiate a fixed pool with no worker')
140 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
141 this.opts
.workerChoiceStrategy
=
142 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
143 this.checkValidWorkerChoiceStrategy(this.opts
.workerChoiceStrategy
)
144 this.opts
.workerChoiceStrategyOptions
=
145 opts
.workerChoiceStrategyOptions
?? DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
146 this.opts
.enableEvents
= opts
.enableEvents
?? true
147 this.opts
.enableTasksQueue
= opts
.enableTasksQueue
?? false
148 if (this.opts
.enableTasksQueue
) {
149 this.checkValidTasksQueueOptions(
150 opts
.tasksQueueOptions
as TasksQueueOptions
152 this.opts
.tasksQueueOptions
= this.buildTasksQueueOptions(
153 opts
.tasksQueueOptions
as TasksQueueOptions
158 private checkValidWorkerChoiceStrategy (
159 workerChoiceStrategy
: WorkerChoiceStrategy
161 if (!Object.values(WorkerChoiceStrategies
).includes(workerChoiceStrategy
)) {
163 `Invalid worker choice strategy '${workerChoiceStrategy}'`
168 private checkValidTasksQueueOptions (
169 tasksQueueOptions
: TasksQueueOptions
171 if ((tasksQueueOptions
?.concurrency
as number) <= 0) {
173 `Invalid worker tasks concurrency '${
174 tasksQueueOptions.concurrency as number
181 public abstract get
type (): PoolType
184 public abstract get
size (): number
187 * Number of tasks running in the pool.
189 private get
numberOfRunningTasks (): number {
190 return this.workerNodes
.reduce(
191 (accumulator
, workerNode
) => accumulator
+ workerNode
.tasksUsage
.running
,
197 * Number of tasks queued in the pool.
199 private get
numberOfQueuedTasks (): number {
200 if (this.opts
.enableTasksQueue
=== false) {
203 return this.workerNodes
.reduce(
204 (accumulator
, workerNode
) => accumulator
+ workerNode
.tasksQueue
.size
,
210 * Gets the given worker its worker node key.
212 * @param worker - The worker.
213 * @returns The worker node key if the worker is found in the pool worker nodes, `-1` otherwise.
215 private getWorkerNodeKey (worker
: Worker
): number {
216 return this.workerNodes
.findIndex(
217 workerNode
=> workerNode
.worker
=== worker
222 public setWorkerChoiceStrategy (
223 workerChoiceStrategy
: WorkerChoiceStrategy
,
224 workerChoiceStrategyOptions
?: WorkerChoiceStrategyOptions
226 this.checkValidWorkerChoiceStrategy(workerChoiceStrategy
)
227 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
228 for (const workerNode
of this.workerNodes
) {
229 this.setWorkerNodeTasksUsage(workerNode
, {
233 runTimeHistory
: new CircularArray(),
239 this.workerChoiceStrategyContext
.setWorkerChoiceStrategy(
240 this.opts
.workerChoiceStrategy
242 if (workerChoiceStrategyOptions
!= null) {
243 this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions
)
248 public setWorkerChoiceStrategyOptions (
249 workerChoiceStrategyOptions
: WorkerChoiceStrategyOptions
251 this.opts
.workerChoiceStrategyOptions
= workerChoiceStrategyOptions
252 this.workerChoiceStrategyContext
.setOptions(
253 this.opts
.workerChoiceStrategyOptions
258 public enableTasksQueue (
260 tasksQueueOptions
?: TasksQueueOptions
262 if (this.opts
.enableTasksQueue
=== true && !enable
) {
263 this.flushTasksQueues()
265 this.opts
.enableTasksQueue
= enable
266 this.setTasksQueueOptions(tasksQueueOptions
as TasksQueueOptions
)
270 public setTasksQueueOptions (tasksQueueOptions
: TasksQueueOptions
): void {
271 if (this.opts
.enableTasksQueue
=== true) {
272 this.checkValidTasksQueueOptions(tasksQueueOptions
)
273 this.opts
.tasksQueueOptions
=
274 this.buildTasksQueueOptions(tasksQueueOptions
)
276 delete this.opts
.tasksQueueOptions
280 private buildTasksQueueOptions (
281 tasksQueueOptions
: TasksQueueOptions
282 ): TasksQueueOptions
{
284 concurrency
: tasksQueueOptions
?.concurrency
?? 1
289 * Whether the pool is full or not.
291 * The pool filling boolean status.
293 protected abstract get
full (): boolean
296 * Whether the pool is busy or not.
298 * The pool busyness boolean status.
300 protected abstract get
busy (): boolean
302 protected internalBusy (): boolean {
304 this.workerNodes
.findIndex(workerNode
=> {
305 return workerNode
.tasksUsage
?.running
=== 0
311 public async execute (data
?: Data
, name
?: string): Promise
<Response
> {
312 const [workerNodeKey
, workerNode
] = this.chooseWorkerNode()
313 const submittedTask
: Task
<Data
> = {
315 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
316 data
: data
?? ({} as Data
),
317 id
: crypto
.randomUUID()
319 const res
= new Promise
<Response
>((resolve
, reject
) => {
320 this.promiseResponseMap
.set(submittedTask
.id
as string, {
323 worker
: workerNode
.worker
327 this.opts
.enableTasksQueue
=== true &&
329 this.workerNodes
[workerNodeKey
].tasksUsage
.running
>=
330 ((this.opts
.tasksQueueOptions
as TasksQueueOptions
)
331 .concurrency
as number))
333 this.enqueueTask(workerNodeKey
, submittedTask
)
335 this.executeTask(workerNodeKey
, submittedTask
)
337 this.checkAndEmitEvents()
338 // eslint-disable-next-line @typescript-eslint/return-await
343 public async destroy (): Promise
<void> {
345 this.workerNodes
.map(async (workerNode
, workerNodeKey
) => {
346 this.flushTasksQueue(workerNodeKey
)
347 await this.destroyWorker(workerNode
.worker
)
353 * Shutdowns the given worker.
355 * @param worker - A worker within `workerNodes`.
357 protected abstract destroyWorker (worker
: Worker
): void | Promise
<void>
360 * Setup hook to execute code before worker node are created in the abstract constructor.
365 protected setupHook (): void {
366 // Intentionally empty
370 * Should return whether the worker is the main worker or not.
372 protected abstract isMain (): boolean
375 * Hook executed before the worker task execution.
378 * @param workerNodeKey - The worker node key.
380 protected beforeTaskExecutionHook (workerNodeKey
: number): void {
381 ++this.workerNodes
[workerNodeKey
].tasksUsage
.running
385 * Hook executed after the worker task execution.
388 * @param worker - The worker.
389 * @param message - The received message.
391 protected afterTaskExecutionHook (
393 message
: MessageValue
<Response
>
395 const workerTasksUsage
= this.getWorkerTasksUsage(worker
)
396 --workerTasksUsage
.running
397 ++workerTasksUsage
.run
398 if (message
.error
!= null) {
399 ++workerTasksUsage
.error
401 if (this.workerChoiceStrategyContext
.getRequiredStatistics().runTime
) {
402 workerTasksUsage
.runTime
+= message
.runTime
?? 0
404 this.workerChoiceStrategyContext
.getRequiredStatistics().avgRunTime
&&
405 workerTasksUsage
.run
!== 0
407 workerTasksUsage
.avgRunTime
=
408 workerTasksUsage
.runTime
/ workerTasksUsage
.run
410 if (this.workerChoiceStrategyContext
.getRequiredStatistics().medRunTime
) {
411 workerTasksUsage
.runTimeHistory
.push(message
.runTime
?? 0)
412 workerTasksUsage
.medRunTime
= median(workerTasksUsage
.runTimeHistory
)
418 * Chooses a worker node for the next task.
420 * The default uses a round robin algorithm to distribute the load.
422 * @returns [worker node key, worker node].
424 protected chooseWorkerNode (): [number, WorkerNode
<Worker
, Data
>] {
425 let workerNodeKey
: number
426 if (this.type === PoolType
.DYNAMIC
&& !this.full
&& this.internalBusy()) {
427 const workerCreated
= this.createAndSetupWorker()
428 this.registerWorkerMessageListener(workerCreated
, message
=> {
430 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
431 (message
.kill
!= null &&
432 this.getWorkerTasksUsage(workerCreated
)?.running
=== 0)
434 // Kill message received from the worker: no new tasks are submitted to that worker for a while ( > maxInactiveTime)
435 this.flushTasksQueueByWorker(workerCreated
)
436 void (this.destroyWorker(workerCreated
) as Promise
<void>)
439 workerNodeKey
= this.getWorkerNodeKey(workerCreated
)
441 workerNodeKey
= this.workerChoiceStrategyContext
.execute()
443 return [workerNodeKey
, this.workerNodes
[workerNodeKey
]]
447 * Sends a message to the given worker.
449 * @param worker - The worker which should receive the message.
450 * @param message - The message.
452 protected abstract sendToWorker (
454 message
: MessageValue
<Data
>
458 * Registers a listener callback on the given worker.
460 * @param worker - The worker which should register a listener.
461 * @param listener - The message listener callback.
463 protected abstract registerWorkerMessageListener
<
464 Message
extends Data
| Response
465 >(worker
: Worker
, listener
: (message
: MessageValue
<Message
>) => void): void
468 * Returns a newly created worker.
470 protected abstract createWorker (): Worker
473 * Function that can be hooked up when a worker has been newly created and moved to the pool worker nodes.
475 * Can be used to update the `maxListeners` or binding the `main-worker`\<-\>`worker` connection if not bind by default.
477 * @param worker - The newly created worker.
479 protected abstract afterWorkerSetup (worker
: Worker
): void
482 * Creates a new worker and sets it up completely in the pool worker nodes.
484 * @returns New, completely set up worker.
486 protected createAndSetupWorker (): Worker
{
487 const worker
= this.createWorker()
489 worker
.on('message', this.opts
.messageHandler
?? EMPTY_FUNCTION
)
490 worker
.on('error', this.opts
.errorHandler
?? EMPTY_FUNCTION
)
491 worker
.on('online', this.opts
.onlineHandler
?? EMPTY_FUNCTION
)
492 worker
.on('exit', this.opts
.exitHandler
?? EMPTY_FUNCTION
)
493 worker
.once('exit', () => {
494 this.removeWorkerNode(worker
)
497 this.pushWorkerNode(worker
)
499 this.afterWorkerSetup(worker
)
505 * This function is the listener registered for each worker message.
507 * @returns The listener function to execute when a message is received from a worker.
509 protected workerListener (): (message
: MessageValue
<Response
>) => void {
511 if (message
.id
!= null) {
512 // Task execution response received
513 const promiseResponse
= this.promiseResponseMap
.get(message
.id
)
514 if (promiseResponse
!= null) {
515 if (message
.error
!= null) {
516 promiseResponse
.reject(message
.error
)
518 promiseResponse
.resolve(message
.data
as Response
)
520 this.afterTaskExecutionHook(promiseResponse
.worker
, message
)
521 this.promiseResponseMap
.delete(message
.id
)
522 const workerNodeKey
= this.getWorkerNodeKey(promiseResponse
.worker
)
524 this.opts
.enableTasksQueue
=== true &&
525 this.tasksQueueSize(workerNodeKey
) > 0
529 this.dequeueTask(workerNodeKey
) as Task
<Data
>
537 private checkAndEmitEvents (): void {
538 if (this.opts
.enableEvents
=== true) {
540 this.emitter
?.emit(PoolEvents
.busy
)
542 if (this.type === PoolType
.DYNAMIC
&& this.full
) {
543 this.emitter
?.emit(PoolEvents
.full
)
549 * Sets the given worker node its tasks usage in the pool.
551 * @param workerNode - The worker node.
552 * @param tasksUsage - The worker node tasks usage.
554 private setWorkerNodeTasksUsage (
555 workerNode
: WorkerNode
<Worker
, Data
>,
556 tasksUsage
: TasksUsage
558 workerNode
.tasksUsage
= tasksUsage
562 * Gets the given worker its tasks usage in the pool.
564 * @param worker - The worker.
565 * @throws Error if the worker is not found in the pool worker nodes.
566 * @returns The worker tasks usage.
568 private getWorkerTasksUsage (worker
: Worker
): TasksUsage
{
569 const workerNodeKey
= this.getWorkerNodeKey(worker
)
570 if (workerNodeKey
!== -1) {
571 return this.workerNodes
[workerNodeKey
].tasksUsage
573 throw new Error('Worker could not be found in the pool worker nodes')
577 * Pushes the given worker in the pool worker nodes.
579 * @param worker - The worker.
580 * @returns The worker nodes length.
582 private pushWorkerNode (worker
: Worker
): number {
583 return this.workerNodes
.push({
589 runTimeHistory
: new CircularArray(),
594 tasksQueue
: new Queue
<Task
<Data
>>()
599 * Sets the given worker in the pool worker nodes.
601 * @param workerNodeKey - The worker node key.
602 * @param worker - The worker.
603 * @param tasksUsage - The worker tasks usage.
604 * @param tasksQueue - The worker task queue.
606 private setWorkerNode (
607 workerNodeKey
: number,
609 tasksUsage
: TasksUsage
,
610 tasksQueue
: Queue
<Task
<Data
>>
612 this.workerNodes
[workerNodeKey
] = {
620 * Removes the given worker from the pool worker nodes.
622 * @param worker - The worker.
624 private removeWorkerNode (worker
: Worker
): void {
625 const workerNodeKey
= this.getWorkerNodeKey(worker
)
626 this.workerNodes
.splice(workerNodeKey
, 1)
627 this.workerChoiceStrategyContext
.remove(workerNodeKey
)
630 private executeTask (workerNodeKey
: number, task
: Task
<Data
>): void {
631 this.beforeTaskExecutionHook(workerNodeKey
)
632 this.sendToWorker(this.workerNodes
[workerNodeKey
].worker
, task
)
635 private enqueueTask (workerNodeKey
: number, task
: Task
<Data
>): number {
636 return this.workerNodes
[workerNodeKey
].tasksQueue
.enqueue(task
)
639 private dequeueTask (workerNodeKey
: number): Task
<Data
> | undefined {
640 return this.workerNodes
[workerNodeKey
].tasksQueue
.dequeue()
643 private tasksQueueSize (workerNodeKey
: number): number {
644 return this.workerNodes
[workerNodeKey
].tasksQueue
.size
647 private flushTasksQueue (workerNodeKey
: number): void {
648 if (this.tasksQueueSize(workerNodeKey
) > 0) {
649 for (let i
= 0; i
< this.tasksQueueSize(workerNodeKey
); i
++) {
652 this.dequeueTask(workerNodeKey
) as Task
<Data
>
658 private flushTasksQueueByWorker (worker
: Worker
): void {
659 const workerNodeKey
= this.getWorkerNodeKey(worker
)
660 this.flushTasksQueue(workerNodeKey
)
663 private flushTasksQueues (): void {
664 for (const [workerNodeKey
] of this.workerNodes
.entries()) {
665 this.flushTasksQueue(workerNodeKey
)