1 import crypto from
'node:crypto'
2 import type { MessageValue
, PromiseResponseWrapper
} from
'../utility-types'
3 import { EMPTY_FUNCTION
} from
'../utils'
4 import { KillBehaviors
, isKillBehavior
} from
'../worker/worker-options'
5 import type { PoolOptions
} from
'./pool'
6 import { PoolEmitter
} from
'./pool'
7 import type { IPoolInternal
, TasksUsage
, WorkerType
} from
'./pool-internal'
8 import { PoolType
} from
'./pool-internal'
9 import type { IPoolWorker
} from
'./pool-worker'
11 WorkerChoiceStrategies
,
12 type WorkerChoiceStrategy
13 } from
'./selection-strategies/selection-strategies-types'
14 import { WorkerChoiceStrategyContext
} from
'./selection-strategies/worker-choice-strategy-context'
17 * Base class that implements some shared logic for all poolifier pools.
19 * @typeParam Worker - Type of worker which manages this pool.
20 * @typeParam Data - Type of data sent to the worker. This can only be serializable data.
21 * @typeParam Response - Type of response of execution. This can only be serializable data.
23 export abstract class AbstractPool
<
24 Worker
extends IPoolWorker
,
27 > implements IPoolInternal
<Worker
, Data
, Response
> {
29 public readonly workers
: Array<WorkerType
<Worker
>> = []
32 public readonly emitter
?: PoolEmitter
35 * The promise response map.
37 * - `key`: The message id of each submitted task.
38 * - `value`: An object that contains the worker key, the promise resolve and reject callbacks.
40 * When we receive a message from the worker we get a map entry with the promise resolve/reject bound to the message.
42 protected promiseResponseMap
: Map
<string, PromiseResponseWrapper
<Response
>> =
43 new Map
<string, PromiseResponseWrapper
<Response
>>()
46 * Worker choice strategy instance implementing the worker choice algorithm.
48 * Default to a strategy implementing a round robin algorithm.
50 protected workerChoiceStrategyContext
: WorkerChoiceStrategyContext
<
57 * Constructs a new poolifier pool.
59 * @param numberOfWorkers - Number of workers that this pool should manage.
60 * @param filePath - Path to the worker-file.
61 * @param opts - Options for the pool.
64 public readonly numberOfWorkers
: number,
65 public readonly filePath
: string,
66 public readonly opts
: PoolOptions
<Worker
>
69 throw new Error('Cannot start a pool from a worker!')
71 this.checkNumberOfWorkers(this.numberOfWorkers
)
72 this.checkFilePath(this.filePath
)
73 this.checkPoolOptions(this.opts
)
76 for (let i
= 1; i
<= this.numberOfWorkers
; i
++) {
77 this.createAndSetupWorker()
80 if (this.opts
.enableEvents
=== true) {
81 this.emitter
= new PoolEmitter()
83 this.workerChoiceStrategyContext
= new WorkerChoiceStrategyContext(
86 const workerCreated
= this.createAndSetupWorker()
87 this.registerWorkerMessageListener(workerCreated
, message
=> {
89 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
90 this.getWorkerTasksUsage(workerCreated
)?.running
=== 0
92 // Kill received from the worker, means that no new tasks are submitted to that worker for a while ( > maxInactiveTime)
93 void this.destroyWorker(workerCreated
)
98 this.opts
.workerChoiceStrategy
102 private checkFilePath (filePath
: string): void {
105 (typeof filePath
=== 'string' && filePath
.trim().length
=== 0)
107 throw new Error('Please specify a file with a worker implementation')
111 private checkNumberOfWorkers (numberOfWorkers
: number): void {
112 if (numberOfWorkers
== null) {
114 'Cannot instantiate a pool without specifying the number of workers'
116 } else if (!Number.isSafeInteger(numberOfWorkers
)) {
118 'Cannot instantiate a pool with a non integer number of workers'
120 } else if (numberOfWorkers
< 0) {
121 throw new RangeError(
122 'Cannot instantiate a pool with a negative number of workers'
124 } else if (this.type === PoolType
.FIXED
&& numberOfWorkers
=== 0) {
125 throw new Error('Cannot instantiate a fixed pool with no worker')
129 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
130 this.opts
.workerChoiceStrategy
=
131 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
132 this.opts
.enableEvents
= opts
.enableEvents
?? true
136 public abstract get
type (): PoolType
139 public get
numberOfRunningTasks (): number {
140 return this.promiseResponseMap
.size
144 * Gets the given worker key.
146 * @param worker - The worker.
147 * @returns The worker key.
149 private getWorkerKey (worker
: Worker
): number {
150 return this.workers
.findIndex(workerItem
=> workerItem
.worker
=== worker
)
154 public setWorkerChoiceStrategy (
155 workerChoiceStrategy
: WorkerChoiceStrategy
157 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
158 for (const workerItem
of this.workers
) {
159 this.setWorker(workerItem
.worker
, {
167 this.workerChoiceStrategyContext
.setWorkerChoiceStrategy(
173 public abstract get
busy (): boolean
175 protected internalGetBusyStatus (): boolean {
177 this.numberOfRunningTasks
>= this.numberOfWorkers
&&
178 this.findFreeWorker() === false
183 public findFreeWorker (): Worker
| false {
184 for (const workerItem
of this.workers
) {
185 if (workerItem
.tasksUsage
.running
=== 0) {
186 // A worker is free, return the matching worker
187 return workerItem
.worker
194 public async execute (data
: Data
): Promise
<Response
> {
195 const worker
= this.chooseWorker()
196 const messageId
= crypto
.randomUUID()
197 const res
= this.internalExecute(this.getWorkerKey(worker
), messageId
)
198 this.checkAndEmitBusy()
199 this.sendToWorker(worker
, {
200 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
201 data
: data
?? ({} as Data
),
204 // eslint-disable-next-line @typescript-eslint/return-await
209 public async destroy (): Promise
<void> {
211 this.workers
.map(async workerItem
=> {
212 await this.destroyWorker(workerItem
.worker
)
218 * Shutdowns given worker.
220 * @param worker - A worker within `workers`.
222 protected abstract destroyWorker (worker
: Worker
): void | Promise
<void>
225 * Setup hook that can be overridden by a Poolifier pool implementation
226 * to run code before workers are created in the abstract constructor.
228 protected setupHook (): void {
233 * Should return whether the worker is the main worker or not.
235 protected abstract isMain (): boolean
238 * Hook executed before the worker task promise resolution.
241 * @param workerKey - The worker key.
243 protected beforePromiseResponseHook (workerKey
: number): void {
244 ++this.workers
[workerKey
].tasksUsage
.running
248 * Hook executed after the worker task promise resolution.
251 * @param workerKey - The worker key.
252 * @param message - The received message.
254 protected afterPromiseResponseHook (
256 message
: MessageValue
<Response
>
258 const workerTasksUsage
= this.workers
[workerKey
].tasksUsage
259 --workerTasksUsage
.running
260 ++workerTasksUsage
.run
261 if (message
.error
!= null) {
262 ++workerTasksUsage
.error
265 this.workerChoiceStrategyContext
.getWorkerChoiceStrategy()
266 .requiredStatistics
.runTime
268 workerTasksUsage
.runTime
+= message
.taskRunTime
?? 0
269 if (workerTasksUsage
.run
!== 0) {
270 workerTasksUsage
.avgRunTime
=
271 workerTasksUsage
.runTime
/ workerTasksUsage
.run
277 * Removes the given worker from the pool.
279 * @param worker - The worker that will be removed.
281 protected removeWorker (worker
: Worker
): void {
282 this.workers
.splice(this.getWorkerKey(worker
), 1)
286 * Chooses a worker for the next task.
288 * The default implementation uses a round robin algorithm to distribute the load.
292 protected chooseWorker (): Worker
{
293 return this.workerChoiceStrategyContext
.execute()
297 * Sends a message to the given worker.
299 * @param worker - The worker which should receive the message.
300 * @param message - The message.
302 protected abstract sendToWorker (
304 message
: MessageValue
<Data
>
308 * Registers a listener callback on a given worker.
310 * @param worker - The worker which should register a listener.
311 * @param listener - The message listener callback.
313 protected abstract registerWorkerMessageListener
<
314 Message
extends Data
| Response
315 >(worker
: Worker
, listener
: (message
: MessageValue
<Message
>) => void): void
318 * Returns a newly created worker.
320 protected abstract createWorker (): Worker
323 * Function that can be hooked up when a worker has been newly created and moved to the workers registry.
325 * Can be used to update the `maxListeners` or binding the `main-worker`\<-\>`worker` connection if not bind by default.
327 * @param worker - The newly created worker.
329 protected abstract afterWorkerSetup (worker
: Worker
): void
332 * Creates a new worker for this pool and sets it up completely.
334 * @returns New, completely set up worker.
336 protected createAndSetupWorker (): Worker
{
337 const worker
= this.createWorker()
339 worker
.on('message', this.opts
.messageHandler
?? EMPTY_FUNCTION
)
340 worker
.on('error', this.opts
.errorHandler
?? EMPTY_FUNCTION
)
341 worker
.on('online', this.opts
.onlineHandler
?? EMPTY_FUNCTION
)
342 worker
.on('exit', this.opts
.exitHandler
?? EMPTY_FUNCTION
)
343 worker
.once('exit', () => {
344 this.removeWorker(worker
)
347 this.setWorker(worker
, {
355 this.afterWorkerSetup(worker
)
361 * This function is the listener registered for each worker.
363 * @returns The listener function to execute when a message is received from a worker.
365 protected workerListener (): (message
: MessageValue
<Response
>) => void {
367 if (message
.id
!== undefined) {
368 const promiseResponse
= this.promiseResponseMap
.get(message
.id
)
369 if (promiseResponse
!== undefined) {
370 if (message
.error
!= null) {
371 promiseResponse
.reject(message
.error
)
373 promiseResponse
.resolve(message
.data
as Response
)
375 this.afterPromiseResponseHook(promiseResponse
.workerKey
, message
)
376 this.promiseResponseMap
.delete(message
.id
)
382 private async internalExecute (
385 ): Promise
<Response
> {
386 this.beforePromiseResponseHook(workerKey
)
387 return await new Promise
<Response
>((resolve
, reject
) => {
388 this.promiseResponseMap
.set(messageId
, { resolve
, reject
, workerKey
})
392 private checkAndEmitBusy (): void {
393 if (this.opts
.enableEvents
=== true && this.busy
) {
394 this.emitter
?.emit('busy')
399 public getWorkerTasksUsage (worker
: Worker
): TasksUsage
| undefined {
400 const workerKey
= this.getWorkerKey(worker
)
401 if (workerKey
!== -1) {
402 return this.workers
[workerKey
].tasksUsage
404 throw new Error('Worker could not be found in the pool')
408 * Sets the given worker.
410 * @param worker - The worker.
411 * @param tasksUsage - The worker tasks usage.
413 private setWorker (worker
: Worker
, tasksUsage
: TasksUsage
): void {