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
: Array<WorkerType
<Worker
>> = []
35 public readonly emitter
?: PoolEmitter
40 * - `key`: This is the message id of each submitted task.
41 * - `value`: An object that contains the worker, the resolve function and the reject function.
43 * When we receive a message from the worker we get a map entry and resolve/reject the promise based on the message.
45 protected promiseMap
: Map
<
47 PromiseWorkerResponseWrapper
<Worker
, Response
>
48 > = new Map
<string, PromiseWorkerResponseWrapper
<Worker
, Response
>>()
51 * Worker choice strategy instance implementing the worker choice algorithm.
53 * Default to a strategy implementing a round robin algorithm.
55 protected workerChoiceStrategyContext
: WorkerChoiceStrategyContext
<
62 * Constructs a new poolifier pool.
64 * @param numberOfWorkers - Number of workers that this pool should manage.
65 * @param filePath - Path to the worker-file.
66 * @param opts - Options for the pool.
69 public readonly numberOfWorkers
: number,
70 public readonly filePath
: string,
71 public readonly opts
: PoolOptions
<Worker
>
74 throw new Error('Cannot start a pool from a worker!')
76 this.checkNumberOfWorkers(this.numberOfWorkers
)
77 this.checkFilePath(this.filePath
)
78 this.checkPoolOptions(this.opts
)
81 for (let i
= 1; i
<= this.numberOfWorkers
; i
++) {
82 this.createAndSetupWorker()
85 if (this.opts
.enableEvents
=== true) {
86 this.emitter
= new PoolEmitter()
88 this.workerChoiceStrategyContext
= new WorkerChoiceStrategyContext(
91 const workerCreated
= this.createAndSetupWorker()
92 this.registerWorkerMessageListener(workerCreated
, message
=> {
94 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
95 this.getWorkerTasksUsage(workerCreated
)?.running
=== 0
97 // Kill received from the worker, means that no new tasks are submitted to that worker for a while ( > maxInactiveTime)
98 void this.destroyWorker(workerCreated
)
103 this.opts
.workerChoiceStrategy
107 private checkFilePath (filePath
: string): void {
110 (typeof filePath
=== 'string' && filePath
.trim().length
=== 0)
112 throw new Error('Please specify a file with a worker implementation')
116 private checkNumberOfWorkers (numberOfWorkers
: number): void {
117 if (numberOfWorkers
== null) {
119 'Cannot instantiate a pool without specifying the number of workers'
121 } else if (!Number.isSafeInteger(numberOfWorkers
)) {
123 'Cannot instantiate a pool with a non integer number of workers'
125 } else if (numberOfWorkers
< 0) {
126 throw new RangeError(
127 'Cannot instantiate a pool with a negative number of workers'
129 } else if (this.type === PoolType
.FIXED
&& numberOfWorkers
=== 0) {
130 throw new Error('Cannot instantiate a fixed pool with no worker')
134 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
135 this.opts
.workerChoiceStrategy
=
136 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
137 this.opts
.enableEvents
= opts
.enableEvents
?? true
141 public abstract get
type (): PoolType
144 public get
numberOfRunningTasks (): number {
145 return this.promiseMap
.size
149 * Gets the given worker key.
151 * @param worker - The worker.
152 * @returns The worker key.
154 private getWorkerKey (worker
: Worker
): number {
155 return this.workers
.findIndex(workerItem
=> workerItem
.worker
=== worker
)
159 public setWorkerChoiceStrategy (
160 workerChoiceStrategy
: WorkerChoiceStrategy
162 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
163 for (const workerItem
of this.workers
) {
164 this.setWorker(workerItem
.worker
, {
171 this.workerChoiceStrategyContext
.setWorkerChoiceStrategy(
177 public abstract get
busy (): boolean
179 protected internalGetBusyStatus (): boolean {
181 this.numberOfRunningTasks
>= this.numberOfWorkers
&&
182 this.findFreeWorker() === false
187 public findFreeWorker (): Worker
| false {
188 for (const workerItem
of this.workers
) {
189 if (workerItem
.tasksUsage
.running
=== 0) {
190 // A worker is free, return the matching worker
191 return workerItem
.worker
198 public async execute (data
: Data
): Promise
<Response
> {
199 const worker
= this.chooseWorker()
200 const messageId
= crypto
.randomUUID()
201 const res
= this.internalExecute(worker
, messageId
)
202 this.checkAndEmitBusy()
203 this.sendToWorker(worker
, {
204 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
205 data
: data
?? ({} as Data
),
208 // eslint-disable-next-line @typescript-eslint/return-await
213 public async destroy (): Promise
<void> {
215 this.workers
.map(async workerItem
=> {
216 await this.destroyWorker(workerItem
.worker
)
222 * Shutdowns given worker.
224 * @param worker - A worker within `workers`.
226 protected abstract destroyWorker (worker
: Worker
): void | Promise
<void>
229 * Setup hook that can be overridden by a Poolifier pool implementation
230 * to run code before workers are created in the abstract constructor.
232 protected setupHook (): void {
237 * Should return whether the worker is the main worker or not.
239 protected abstract isMain (): boolean
242 * Hook executed before the worker task promise resolution.
245 * @param worker - The worker.
247 protected beforePromiseWorkerResponseHook (worker
: Worker
): void {
248 ++(this.getWorkerTasksUsage(worker
) as TasksUsage
).running
252 * Hook executed after the worker task promise resolution.
255 * @param message - The received message.
256 * @param promise - The Promise response.
258 protected afterPromiseWorkerResponseHook (
259 message
: MessageValue
<Response
>,
260 promise
: PromiseWorkerResponseWrapper
<Worker
, Response
>
262 const workerTasksUsage
= this.getWorkerTasksUsage(
265 --workerTasksUsage
.running
266 ++workerTasksUsage
.run
268 this.workerChoiceStrategyContext
.getWorkerChoiceStrategy()
269 .requiredStatistics
.runTime
271 workerTasksUsage
.runTime
+= message
.taskRunTime
?? 0
272 if (workerTasksUsage
.run
!== 0) {
273 workerTasksUsage
.avgRunTime
=
274 workerTasksUsage
.runTime
/ workerTasksUsage
.run
280 * Removes the given worker from the pool.
282 * @param worker - The worker that will be removed.
284 protected removeWorker (worker
: Worker
): void {
285 this.workers
.splice(this.getWorkerKey(worker
), 1)
289 * Chooses a worker for the next task.
291 * The default implementation uses a round robin algorithm to distribute the load.
295 protected chooseWorker (): Worker
{
296 return this.workerChoiceStrategyContext
.execute()
300 * Sends a message to the given worker.
302 * @param worker - The worker which should receive the message.
303 * @param message - The message.
305 protected abstract sendToWorker (
307 message
: MessageValue
<Data
>
311 * Registers a listener callback on a given worker.
313 * @param worker - The worker which should register a listener.
314 * @param listener - The message listener callback.
316 protected abstract registerWorkerMessageListener
<
317 Message
extends Data
| Response
318 >(worker
: Worker
, listener
: (message
: MessageValue
<Message
>) => void): void
321 * Returns a newly created worker.
323 protected abstract createWorker (): Worker
326 * Function that can be hooked up when a worker has been newly created and moved to the workers registry.
328 * Can be used to update the `maxListeners` or binding the `main-worker`\<-\>`worker` connection if not bind by default.
330 * @param worker - The newly created worker.
332 protected abstract afterWorkerSetup (worker
: Worker
): void
335 * Creates a new worker for this pool and sets it up completely.
337 * @returns New, completely set up worker.
339 protected createAndSetupWorker (): Worker
{
340 const worker
= this.createWorker()
342 worker
.on('message', this.opts
.messageHandler
?? EMPTY_FUNCTION
)
343 worker
.on('error', this.opts
.errorHandler
?? EMPTY_FUNCTION
)
344 worker
.on('online', this.opts
.onlineHandler
?? EMPTY_FUNCTION
)
345 worker
.on('exit', this.opts
.exitHandler
?? EMPTY_FUNCTION
)
346 worker
.once('exit', () => {
347 this.removeWorker(worker
)
350 this.setWorker(worker
, {
357 this.afterWorkerSetup(worker
)
363 * This function is the listener registered for each worker.
365 * @returns The listener function to execute when a message is received from a worker.
367 protected workerListener (): (message
: MessageValue
<Response
>) => void {
369 if (message
.id
!== undefined) {
370 const promise
= this.promiseMap
.get(message
.id
)
371 if (promise
!== undefined) {
372 if (message
.error
!= null) {
373 promise
.reject(message
.error
)
375 promise
.resolve(message
.data
as Response
)
377 this.afterPromiseWorkerResponseHook(message
, promise
)
378 this.promiseMap
.delete(message
.id
)
384 private async internalExecute (
387 ): Promise
<Response
> {
388 this.beforePromiseWorkerResponseHook(worker
)
389 return await new Promise
<Response
>((resolve
, reject
) => {
390 this.promiseMap
.set(messageId
, { resolve
, reject
, worker
})
394 private checkAndEmitBusy (): void {
395 if (this.opts
.enableEvents
=== true && this.busy
) {
396 this.emitter
?.emit('busy')
401 public getWorkerTasksUsage (worker
: Worker
): TasksUsage
| undefined {
402 const workerKey
= this.getWorkerKey(worker
)
403 if (workerKey
!== -1) {
404 return this.workers
[workerKey
].tasksUsage
406 throw new Error('Worker could not be found in the pool')
410 * Sets the given worker.
412 * @param worker - The worker.
413 * @param tasksUsage - The worker tasks usage.
415 private setWorker (worker
: Worker
, tasksUsage
: TasksUsage
): void {