1 import crypto from
'node:crypto'
2 import type { MessageValue
, PromiseResponseWrapper
} from
'../utility-types'
3 import { EMPTY_FUNCTION
, median
} from
'../utils'
4 import { KillBehaviors
, isKillBehavior
} from
'../worker/worker-options'
5 import { PoolEvents
, type PoolOptions
} from
'./pool'
6 import { PoolEmitter
} from
'./pool'
7 import type { IPoolInternal
} from
'./pool-internal'
8 import { PoolType
} from
'./pool-internal'
9 import type { IWorker
, Task
, TasksUsage
, WorkerNode
} from
'./worker'
11 WorkerChoiceStrategies
,
12 type WorkerChoiceStrategy
13 } from
'./selection-strategies/selection-strategies-types'
14 import { WorkerChoiceStrategyContext
} from
'./selection-strategies/worker-choice-strategy-context'
15 import { CircularArray
} from
'../circular-array'
18 * Base class that implements some shared logic for all poolifier pools.
20 * @typeParam Worker - Type of worker which manages this pool.
21 * @typeParam Data - Type of data sent to the worker. This can only be serializable data.
22 * @typeParam Response - Type of response of execution. This can only be serializable data.
24 export abstract class AbstractPool
<
25 Worker
extends IWorker
,
28 > implements IPoolInternal
<Worker
, Data
, Response
> {
30 public readonly workerNodes
: Array<WorkerNode
<Worker
, Data
>> = []
33 public readonly emitter
?: PoolEmitter
36 * The promise response map.
38 * - `key`: The message id of each submitted task.
39 * - `value`: An object that contains the worker, the promise resolve and reject callbacks.
41 * When we receive a message from the worker we get a map entry with the promise resolve/reject bound to the message.
43 protected promiseResponseMap
: Map
<
45 PromiseResponseWrapper
<Worker
, Response
>
46 > = new Map
<string, PromiseResponseWrapper
<Worker
, Response
>>()
49 * Worker choice strategy context referencing a worker choice algorithm implementation.
51 * Default to a round robin algorithm.
53 protected workerChoiceStrategyContext
: WorkerChoiceStrategyContext
<
60 * Constructs a new poolifier pool.
62 * @param numberOfWorkers - Number of workers that this pool should manage.
63 * @param filePath - Path to the worker-file.
64 * @param opts - Options for the pool.
67 public readonly numberOfWorkers
: number,
68 public readonly filePath
: string,
69 public readonly opts
: PoolOptions
<Worker
>
72 throw new Error('Cannot start a pool from a worker!')
74 this.checkNumberOfWorkers(this.numberOfWorkers
)
75 this.checkFilePath(this.filePath
)
76 this.checkPoolOptions(this.opts
)
78 this.chooseWorkerNode
.bind(this)
79 this.internalExecute
.bind(this)
80 this.checkAndEmitEvents
.bind(this)
81 this.sendToWorker
.bind(this)
85 for (let i
= 1; i
<= this.numberOfWorkers
; i
++) {
86 this.createAndSetupWorker()
89 if (this.opts
.enableEvents
=== true) {
90 this.emitter
= new PoolEmitter()
92 this.workerChoiceStrategyContext
= new WorkerChoiceStrategyContext
<
96 >(this, this.opts
.workerChoiceStrategy
)
99 private checkFilePath (filePath
: string): void {
102 (typeof filePath
=== 'string' && filePath
.trim().length
=== 0)
104 throw new Error('Please specify a file with a worker implementation')
108 private checkNumberOfWorkers (numberOfWorkers
: number): void {
109 if (numberOfWorkers
== null) {
111 'Cannot instantiate a pool without specifying the number of workers'
113 } else if (!Number.isSafeInteger(numberOfWorkers
)) {
115 'Cannot instantiate a pool with a non integer number of workers'
117 } else if (numberOfWorkers
< 0) {
118 throw new RangeError(
119 'Cannot instantiate a pool with a negative number of workers'
121 } else if (this.type === PoolType
.FIXED
&& numberOfWorkers
=== 0) {
122 throw new Error('Cannot instantiate a fixed pool with no worker')
126 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
127 this.opts
.workerChoiceStrategy
=
128 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
129 this.checkValidWorkerChoiceStrategy(this.opts
.workerChoiceStrategy
)
130 this.opts
.enableEvents
= opts
.enableEvents
?? true
131 this.opts
.enableTasksQueue
= opts
.enableTasksQueue
?? false
134 private checkValidWorkerChoiceStrategy (
135 workerChoiceStrategy
: WorkerChoiceStrategy
137 if (!Object.values(WorkerChoiceStrategies
).includes(workerChoiceStrategy
)) {
139 `Invalid worker choice strategy '${workerChoiceStrategy}'`
145 public abstract get
type (): PoolType
148 * Number of tasks running in the pool.
150 private get
numberOfRunningTasks (): number {
151 return this.workerNodes
.reduce(
152 (accumulator
, workerNode
) => accumulator
+ workerNode
.tasksUsage
.running
,
158 * Number of tasks queued in the pool.
160 private get
numberOfQueuedTasks (): number {
161 if (this.opts
.enableTasksQueue
=== false) {
164 return this.workerNodes
.reduce(
165 (accumulator
, workerNode
) => accumulator
+ workerNode
.tasksQueue
.length
,
171 * Gets the given worker its worker node key.
173 * @param worker - The worker.
174 * @returns The worker node key if the worker is found in the pool worker nodes, `-1` otherwise.
176 private getWorkerNodeKey (worker
: Worker
): number {
177 return this.workerNodes
.findIndex(
178 workerNode
=> workerNode
.worker
=== worker
183 public setWorkerChoiceStrategy (
184 workerChoiceStrategy
: WorkerChoiceStrategy
186 this.checkValidWorkerChoiceStrategy(workerChoiceStrategy
)
187 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
188 for (const [index
, workerNode
] of this.workerNodes
.entries()) {
196 runTimeHistory
: new CircularArray(),
201 workerNode
.tasksQueue
204 this.workerChoiceStrategyContext
.setWorkerChoiceStrategy(
210 public abstract get
full (): boolean
213 public abstract get
busy (): boolean
215 protected internalBusy (): boolean {
217 this.numberOfRunningTasks
>= this.numberOfWorkers
&&
218 this.findFreeWorkerNodeKey() === -1
223 public findFreeWorkerNodeKey (): number {
224 return this.workerNodes
.findIndex(workerNode
=> {
225 return workerNode
.tasksUsage
?.running
=== 0
230 public async execute (data
: Data
): Promise
<Response
> {
231 const [workerNodeKey
, workerNode
] = this.chooseWorkerNode()
232 const submittedTask
: Task
<Data
> = {
233 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
234 data
: data
?? ({} as Data
),
235 id
: crypto
.randomUUID()
237 const res
= this.internalExecute(workerNodeKey
, workerNode
, submittedTask
)
238 let currentTask
: Task
<Data
> = submittedTask
240 this.opts
.enableTasksQueue
=== true &&
241 (this.busy
|| this.tasksQueueLength(workerNodeKey
) > 0)
243 this.enqueueTask(workerNodeKey
, submittedTask
)
244 currentTask
= this.dequeueTask(workerNodeKey
) as Task
<Data
>
246 this.sendToWorker(workerNode
.worker
, currentTask
)
247 this.checkAndEmitEvents()
248 // eslint-disable-next-line @typescript-eslint/return-await
253 public async destroy (): Promise
<void> {
255 this.workerNodes
.map(async workerNode
=> {
256 this.flushTasksQueueByWorker(workerNode
.worker
)
257 await this.destroyWorker(workerNode
.worker
)
263 * Shutdowns the given worker.
265 * @param worker - A worker within `workerNodes`.
267 protected abstract destroyWorker (worker
: Worker
): void | Promise
<void>
270 * Setup hook to run code before worker node are created in the abstract constructor.
275 protected setupHook (): void {
276 // Intentionally empty
280 * Should return whether the worker is the main worker or not.
282 protected abstract isMain (): boolean
285 * Hook executed before the worker task promise resolution.
288 * @param workerNodeKey - The worker node key.
290 protected beforePromiseResponseHook (workerNodeKey
: number): void {
291 ++this.workerNodes
[workerNodeKey
].tasksUsage
.running
295 * Hook executed after the worker task promise resolution.
298 * @param worker - The worker.
299 * @param message - The received message.
301 protected afterPromiseResponseHook (
303 message
: MessageValue
<Response
>
305 const workerTasksUsage
= this.getWorkerTasksUsage(worker
) as TasksUsage
306 --workerTasksUsage
.running
307 ++workerTasksUsage
.run
308 if (message
.error
!= null) {
309 ++workerTasksUsage
.error
311 if (this.workerChoiceStrategyContext
.getRequiredStatistics().runTime
) {
312 workerTasksUsage
.runTime
+= message
.runTime
?? 0
314 this.workerChoiceStrategyContext
.getRequiredStatistics().avgRunTime
&&
315 workerTasksUsage
.run
!== 0
317 workerTasksUsage
.avgRunTime
=
318 workerTasksUsage
.runTime
/ workerTasksUsage
.run
320 if (this.workerChoiceStrategyContext
.getRequiredStatistics().medRunTime
) {
321 workerTasksUsage
.runTimeHistory
.push(message
.runTime
?? 0)
322 workerTasksUsage
.medRunTime
= median(workerTasksUsage
.runTimeHistory
)
328 * Chooses a worker node for the next task.
330 * The default uses a round robin algorithm to distribute the load.
332 * @returns [worker node key, worker node].
334 protected chooseWorkerNode (): [number, WorkerNode
<Worker
, Data
>] {
335 let workerNodeKey
: number
337 this.type === PoolType
.DYNAMIC
&&
339 this.findFreeWorkerNodeKey() === -1
341 const workerCreated
= this.createAndSetupWorker()
342 this.registerWorkerMessageListener(workerCreated
, message
=> {
344 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
345 (message
.kill
!= null &&
346 this.getWorkerTasksUsage(workerCreated
)?.running
=== 0)
348 // Kill message received from the worker: no new tasks are submitted to that worker for a while ( > maxInactiveTime)
349 this.flushTasksQueueByWorker(workerCreated
)
350 void this.destroyWorker(workerCreated
)
353 workerNodeKey
= this.getWorkerNodeKey(workerCreated
)
355 workerNodeKey
= this.workerChoiceStrategyContext
.execute()
357 return [workerNodeKey
, this.workerNodes
[workerNodeKey
]]
361 * Sends a message to the given worker.
363 * @param worker - The worker which should receive the message.
364 * @param message - The message.
366 protected abstract sendToWorker (
368 message
: MessageValue
<Data
>
372 * Registers a listener callback on the given worker.
374 * @param worker - The worker which should register a listener.
375 * @param listener - The message listener callback.
377 protected abstract registerWorkerMessageListener
<
378 Message
extends Data
| Response
379 >(worker
: Worker
, listener
: (message
: MessageValue
<Message
>) => void): void
382 * Returns a newly created worker.
384 protected abstract createWorker (): Worker
387 * Function that can be hooked up when a worker has been newly created and moved to the pool worker nodes.
389 * Can be used to update the `maxListeners` or binding the `main-worker`\<-\>`worker` connection if not bind by default.
391 * @param worker - The newly created worker.
393 protected abstract afterWorkerSetup (worker
: Worker
): void
396 * Creates a new worker and sets it up completely in the pool worker nodes.
398 * @returns New, completely set up worker.
400 protected createAndSetupWorker (): Worker
{
401 const worker
= this.createWorker()
403 worker
.on('message', this.opts
.messageHandler
?? EMPTY_FUNCTION
)
404 worker
.on('error', this.opts
.errorHandler
?? EMPTY_FUNCTION
)
405 worker
.on('online', this.opts
.onlineHandler
?? EMPTY_FUNCTION
)
406 worker
.on('exit', this.opts
.exitHandler
?? EMPTY_FUNCTION
)
407 worker
.once('exit', () => {
408 this.removeWorkerNode(worker
)
411 this.pushWorkerNode(worker
)
413 this.afterWorkerSetup(worker
)
419 * This function is the listener registered for each worker message.
421 * @returns The listener function to execute when a message is received from a worker.
423 protected workerListener (): (message
: MessageValue
<Response
>) => void {
425 if (message
.id
!= null) {
426 // Task response received
427 const promiseResponse
= this.promiseResponseMap
.get(message
.id
)
428 if (promiseResponse
!= null) {
429 if (message
.error
!= null) {
430 promiseResponse
.reject(message
.error
)
432 promiseResponse
.resolve(message
.data
as Response
)
434 this.afterPromiseResponseHook(promiseResponse
.worker
, message
)
435 this.promiseResponseMap
.delete(message
.id
)
436 const workerNodeKey
= this.getWorkerNodeKey(promiseResponse
.worker
)
438 this.opts
.enableTasksQueue
=== true &&
439 this.tasksQueueLength(workerNodeKey
) > 0
442 promiseResponse
.worker
,
443 this.dequeueTask(workerNodeKey
) as Task
<Data
>
451 private async internalExecute (
452 workerNodeKey
: number,
453 workerNode
: WorkerNode
<Worker
, Data
>,
455 ): Promise
<Response
> {
456 this.beforePromiseResponseHook(workerNodeKey
)
457 return await new Promise
<Response
>((resolve
, reject
) => {
458 this.promiseResponseMap
.set(task
.id
, {
461 worker
: workerNode
.worker
466 private checkAndEmitEvents (): void {
467 if (this.opts
.enableEvents
=== true) {
469 this.emitter
?.emit(PoolEvents
.busy
)
471 if (this.type === PoolType
.DYNAMIC
&& this.full
) {
472 this.emitter
?.emit(PoolEvents
.full
)
478 * Gets the given worker its tasks usage in the pool.
480 * @param worker - The worker.
481 * @returns The worker tasks usage.
483 private getWorkerTasksUsage (worker
: Worker
): TasksUsage
| undefined {
484 const workerNodeKey
= this.getWorkerNodeKey(worker
)
485 if (workerNodeKey
!== -1) {
486 return this.workerNodes
[workerNodeKey
].tasksUsage
488 throw new Error('Worker could not be found in the pool worker nodes')
492 * Pushes the given worker in the pool worker nodes.
494 * @param worker - The worker.
495 * @returns The worker nodes length.
497 private pushWorkerNode (worker
: Worker
): number {
498 return this.workerNodes
.push({
504 runTimeHistory
: new CircularArray(),
514 * Sets the given worker in the pool worker nodes.
516 * @param workerNodeKey - The worker node key.
517 * @param worker - The worker.
518 * @param tasksUsage - The worker tasks usage.
519 * @param tasksQueue - The worker task queue.
521 private setWorkerNode (
522 workerNodeKey
: number,
524 tasksUsage
: TasksUsage
,
525 tasksQueue
: Array<Task
<Data
>>
527 this.workerNodes
[workerNodeKey
] = {
535 * Removes the given worker from the pool worker nodes.
537 * @param worker - The worker.
539 protected removeWorkerNode (worker
: Worker
): void {
540 const workerNodeKey
= this.getWorkerNodeKey(worker
)
541 this.workerNodes
.splice(workerNodeKey
, 1)
542 this.workerChoiceStrategyContext
.remove(workerNodeKey
)
545 protected enqueueTask (workerNodeKey
: number, task
: Task
<Data
>): void {
546 this.workerNodes
[workerNodeKey
].tasksQueue
.push(task
)
549 protected dequeueTask (workerNodeKey
: number): Task
<Data
> | undefined {
550 return this.workerNodes
[workerNodeKey
].tasksQueue
.shift()
553 protected tasksQueueLength (workerNodeKey
: number): number {
554 return this.workerNodes
[workerNodeKey
].tasksQueue
.length
557 protected flushTasksQueue (workerNodeKey
: number): void {
558 if (this.tasksQueueLength(workerNodeKey
) > 0) {
559 for (const task
of this.workerNodes
[workerNodeKey
].tasksQueue
) {
560 this.sendToWorker(this.workerNodes
[workerNodeKey
].worker
, task
)
562 this.workerNodes
[workerNodeKey
].tasksQueue
= []
566 protected flushTasksQueueByWorker (worker
: Worker
): void {
567 const workerNodeKey
= this.getWorkerNodeKey(worker
)
568 this.flushTasksQueue(workerNodeKey
)