1 import crypto from
'node:crypto'
4 PromiseWorkerResponseWrapper
5 } from
'../utility-types'
6 import { EMPTY_FUNCTION
} from
'../utils'
7 import { KillBehaviors
, isKillBehavior
} from
'../worker/worker-options'
8 import type { PoolOptions
} from
'./pool'
9 import { PoolEmitter
} from
'./pool'
10 import type { IPoolInternal
, TasksUsage
, WorkerType
} from
'./pool-internal'
11 import { PoolType
} from
'./pool-internal'
12 import type { IPoolWorker
} from
'./pool-worker'
14 WorkerChoiceStrategies
,
15 type WorkerChoiceStrategy
16 } from
'./selection-strategies/selection-strategies-types'
17 import { WorkerChoiceStrategyContext
} from
'./selection-strategies/worker-choice-strategy-context'
20 * Base class that implements some shared logic for all poolifier pools.
22 * @typeParam Worker - Type of worker which manages this pool.
23 * @typeParam Data - Type of data sent to the worker. This can only be serializable data.
24 * @typeParam Response - Type of response of execution. This can only be serializable data.
26 export abstract class AbstractPool
<
27 Worker
extends IPoolWorker
,
30 > implements IPoolInternal
<Worker
, Data
, Response
> {
32 public readonly workers
: Map
<number, WorkerType
<Worker
>> = new Map
<
38 public readonly emitter
?: PoolEmitter
41 * Id of the next worker.
43 protected nextWorkerId
: number = 0
48 * - `key`: This is the message id of each submitted task.
49 * - `value`: An object that contains the worker, the resolve function and the reject function.
51 * When we receive a message from the worker we get a map entry and resolve/reject the promise based on the message.
53 protected promiseMap
: Map
<
55 PromiseWorkerResponseWrapper
<Worker
, Response
>
56 > = new Map
<string, PromiseWorkerResponseWrapper
<Worker
, Response
>>()
59 * Worker choice strategy instance implementing the worker choice algorithm.
61 * Default to a strategy implementing 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
)
89 for (let i
= 1; i
<= this.numberOfWorkers
; i
++) {
90 this.createAndSetupWorker()
93 if (this.opts
.enableEvents
=== true) {
94 this.emitter
= new PoolEmitter()
96 this.workerChoiceStrategyContext
= new WorkerChoiceStrategyContext(
99 const workerCreated
= this.createAndSetupWorker()
100 this.registerWorkerMessageListener(workerCreated
, message
=> {
102 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
103 this.getWorkerRunningTasks(workerCreated
) === 0
105 // Kill received from the worker, means that no new tasks are submitted to that worker for a while ( > maxInactiveTime)
106 void this.destroyWorker(workerCreated
)
111 this.opts
.workerChoiceStrategy
115 private checkFilePath (filePath
: string): void {
118 (typeof filePath
=== 'string' && filePath
.trim().length
=== 0)
120 throw new Error('Please specify a file with a worker implementation')
124 private checkNumberOfWorkers (numberOfWorkers
: number): void {
125 if (numberOfWorkers
== null) {
127 'Cannot instantiate a pool without specifying the number of workers'
129 } else if (!Number.isSafeInteger(numberOfWorkers
)) {
131 'Cannot instantiate a pool with a non integer number of workers'
133 } else if (numberOfWorkers
< 0) {
134 throw new RangeError(
135 'Cannot instantiate a pool with a negative number of workers'
137 } else if (this.type === PoolType
.FIXED
&& numberOfWorkers
=== 0) {
138 throw new Error('Cannot instantiate a fixed pool with no worker')
142 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
143 this.opts
.workerChoiceStrategy
=
144 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
145 this.opts
.enableEvents
= opts
.enableEvents
?? true
149 public abstract get
type (): PoolType
152 public get
numberOfRunningTasks (): number {
153 return this.promiseMap
.size
157 * Gets the given worker key.
159 * @param worker - The worker.
160 * @returns The worker key.
162 private getWorkerKey (worker
: Worker
): number | undefined {
163 return [...this.workers
].find(([, value
]) => value
.worker
=== worker
)?.[0]
167 public getWorkerRunningTasks (worker
: Worker
): number | undefined {
168 return this.workers
.get(this.getWorkerKey(worker
) as number)?.tasksUsage
173 public getWorkerAverageTasksRunTime (worker
: Worker
): number | undefined {
174 return this.workers
.get(this.getWorkerKey(worker
) as number)?.tasksUsage
179 public setWorkerChoiceStrategy (
180 workerChoiceStrategy
: WorkerChoiceStrategy
182 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
183 for (const [key
, value
] of this.workers
) {
184 this.setWorker(key
, value
.worker
, {
191 this.workerChoiceStrategyContext
.setWorkerChoiceStrategy(
197 public abstract get
busy (): boolean
199 protected internalGetBusyStatus (): boolean {
201 this.numberOfRunningTasks
>= this.numberOfWorkers
&&
202 this.findFreeWorker() === false
207 public findFreeWorker (): Worker
| false {
208 for (const value
of this.workers
.values()) {
209 if (value
.tasksUsage
.running
=== 0) {
210 // A worker is free, return the matching worker
218 public async execute (data
: Data
): Promise
<Response
> {
219 const worker
= this.chooseWorker()
220 const messageId
= crypto
.randomUUID()
221 const res
= this.internalExecute(worker
, messageId
)
222 this.checkAndEmitBusy()
223 this.sendToWorker(worker
, {
224 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
225 data
: data
?? ({} as Data
),
228 // eslint-disable-next-line @typescript-eslint/return-await
233 public async destroy (): Promise
<void> {
235 [...this.workers
].map(async ([, value
]) => {
236 await this.destroyWorker(value
.worker
)
242 * Shutdowns given worker.
244 * @param worker - A worker within `workers`.
246 protected abstract destroyWorker (worker
: Worker
): void | Promise
<void>
249 * Setup hook that can be overridden by a Poolifier pool implementation
250 * to run code before workers are created in the abstract constructor.
252 protected setupHook (): void {
257 * Should return whether the worker is the main worker or not.
259 protected abstract isMain (): boolean
262 * Hook executed before the worker task promise resolution.
265 * @param worker - The worker.
267 protected beforePromiseWorkerResponseHook (worker
: Worker
): void {
268 this.increaseWorkerRunningTasks(worker
)
272 * Hook executed after the worker task promise resolution.
275 * @param message - The received message.
276 * @param promise - The Promise response.
278 protected afterPromiseWorkerResponseHook (
279 message
: MessageValue
<Response
>,
280 promise
: PromiseWorkerResponseWrapper
<Worker
, Response
>
282 this.decreaseWorkerRunningTasks(promise
.worker
)
283 this.stepWorkerRunTasks(promise
.worker
, 1)
284 this.updateWorkerTasksRunTime(promise
.worker
, message
.taskRunTime
)
288 * Removes the given worker from the pool.
290 * @param worker - The worker that will be removed.
292 protected removeWorker (worker
: Worker
): void {
293 this.workers
.delete(this.getWorkerKey(worker
) as number)
298 * Chooses a worker for the next task.
300 * The default implementation uses a round robin algorithm to distribute the load.
304 protected chooseWorker (): Worker
{
305 return this.workerChoiceStrategyContext
.execute()
309 * Sends a message to the given worker.
311 * @param worker - The worker which should receive the message.
312 * @param message - The message.
314 protected abstract sendToWorker (
316 message
: MessageValue
<Data
>
320 * Registers a listener callback on a given worker.
322 * @param worker - The worker which should register a listener.
323 * @param listener - The message listener callback.
325 protected abstract registerWorkerMessageListener
<
326 Message
extends Data
| Response
327 >(worker
: Worker
, listener
: (message
: MessageValue
<Message
>) => void): void
330 * Returns a newly created worker.
332 protected abstract createWorker (): Worker
335 * Function that can be hooked up when a worker has been newly created and moved to the workers registry.
337 * Can be used to update the `maxListeners` or binding the `main-worker`\<-\>`worker` connection if not bind by default.
339 * @param worker - The newly created worker.
341 protected abstract afterWorkerSetup (worker
: Worker
): void
344 * Creates a new worker for this pool and sets it up completely.
346 * @returns New, completely set up worker.
348 protected createAndSetupWorker (): Worker
{
349 const worker
= this.createWorker()
351 worker
.on('message', this.opts
.messageHandler
?? EMPTY_FUNCTION
)
352 worker
.on('error', this.opts
.errorHandler
?? EMPTY_FUNCTION
)
353 worker
.on('online', this.opts
.onlineHandler
?? EMPTY_FUNCTION
)
354 worker
.on('exit', this.opts
.exitHandler
?? EMPTY_FUNCTION
)
355 worker
.once('exit', () => {
356 this.removeWorker(worker
)
359 this.setWorker(this.nextWorkerId
, worker
, {
367 this.afterWorkerSetup(worker
)
373 * This function is the listener registered for each worker.
375 * @returns The listener function to execute when a message is received from a worker.
377 protected workerListener (): (message
: MessageValue
<Response
>) => void {
379 if (message
.id
!== undefined) {
380 const promise
= this.promiseMap
.get(message
.id
)
381 if (promise
!== undefined) {
382 if (message
.error
!= null) {
383 promise
.reject(message
.error
)
385 promise
.resolve(message
.data
as Response
)
387 this.afterPromiseWorkerResponseHook(message
, promise
)
388 this.promiseMap
.delete(message
.id
)
394 private async internalExecute (
397 ): Promise
<Response
> {
398 this.beforePromiseWorkerResponseHook(worker
)
399 return await new Promise
<Response
>((resolve
, reject
) => {
400 this.promiseMap
.set(messageId
, { resolve
, reject
, worker
})
404 private checkAndEmitBusy (): void {
405 if (this.opts
.enableEvents
=== true && this.busy
) {
406 this.emitter
?.emit('busy')
411 * Increases the number of tasks that the given worker has applied.
413 * @param worker - Worker which running tasks is increased.
415 private increaseWorkerRunningTasks (worker
: Worker
): void {
416 this.stepWorkerRunningTasks(worker
, 1)
420 * Decreases the number of tasks that the given worker has applied.
422 * @param worker - Worker which running tasks is decreased.
424 private decreaseWorkerRunningTasks (worker
: Worker
): void {
425 this.stepWorkerRunningTasks(worker
, -1)
429 * Gets tasks usage of the given worker.
431 * @param worker - Worker which tasks usage is returned.
433 private getWorkerTasksUsage (worker
: Worker
): TasksUsage
| undefined {
434 if (this.checkWorker(worker
)) {
435 const workerKey
= this.getWorkerKey(worker
) as number
436 const workerEntry
= this.workers
.get(workerKey
) as WorkerType
<Worker
>
437 return workerEntry
.tasksUsage
442 * Steps the number of tasks that the given worker has applied.
444 * @param worker - Worker which running tasks are stepped.
445 * @param step - Number of running tasks step.
447 private stepWorkerRunningTasks (worker
: Worker
, step
: number): void {
449 (this.getWorkerTasksUsage(worker
) as TasksUsage
).running
+= step
453 * Steps the number of tasks that the given worker has run.
455 * @param worker - Worker which has run tasks.
456 * @param step - Number of run tasks step.
458 private stepWorkerRunTasks (worker
: Worker
, step
: number): void {
460 (this.getWorkerTasksUsage(worker
) as TasksUsage
).run
+= step
464 * Updates tasks runtime for the given worker.
466 * @param worker - Worker which run the task.
467 * @param taskRunTime - Worker task runtime.
469 private updateWorkerTasksRunTime (
471 taskRunTime
: number | undefined
474 this.workerChoiceStrategyContext
.getWorkerChoiceStrategy()
475 .requiredStatistics
.runTime
477 const workerTasksUsage
= this.getWorkerTasksUsage(worker
) as TasksUsage
478 workerTasksUsage
.runTime
+= taskRunTime
?? 0
479 if (workerTasksUsage
.run
!== 0) {
480 workerTasksUsage
.avgRunTime
=
481 workerTasksUsage
.runTime
/ workerTasksUsage
.run
487 * Sets the given worker.
489 * @param workerKey - The worker key.
490 * @param worker - The worker.
491 * @param tasksUsage - The worker tasks usage.
496 tasksUsage
: TasksUsage
498 this.workers
.set(workerKey
, {
505 * Checks if the given worker is registered in the pool.
507 * @param worker - Worker to check.
508 * @returns `true` if the worker is registered in the pool.
510 private checkWorker (worker
: Worker
): boolean {
511 if (this.getWorkerKey(worker
) == null) {
512 throw new Error('Worker could not be found in the pool')