3 PromiseWorkerResponseWrapper
4 } from
'../utility-types'
5 import { EMPTY_FUNCTION
, EMPTY_OBJECT_LITERAL
} from
'../utils'
6 import { isKillBehavior
, KillBehaviors
} from
'../worker/worker-options'
7 import type { PoolOptions
} from
'./pool'
8 import { PoolEmitter
} from
'./pool'
9 import type { IPoolInternal
, TasksUsage
} from
'./pool-internal'
10 import { PoolType
} from
'./pool-internal'
11 import type { IPoolWorker
} from
'./pool-worker'
13 WorkerChoiceStrategies
,
14 type WorkerChoiceStrategy
15 } from
'./selection-strategies/selection-strategies-types'
16 import { WorkerChoiceStrategyContext
} from
'./selection-strategies/worker-choice-strategy-context'
19 * Base class that implements some shared logic for all poolifier pools.
21 * @typeParam Worker - Type of worker which manages this pool.
22 * @typeParam Data - Type of data sent to the worker. This can only be serializable data.
23 * @typeParam Response - Type of response of execution. This can only be serializable data.
25 export abstract class AbstractPool
<
26 Worker
extends IPoolWorker
,
29 > implements IPoolInternal
<Worker
, Data
, Response
> {
31 public readonly workers
: Worker
[] = []
34 public readonly workersTasksUsage
: Map
<Worker
, TasksUsage
> = new Map
<
40 public readonly emitter
?: PoolEmitter
45 * - `key`: This is the message Id of each submitted task.
46 * - `value`: An object that contains the worker, the resolve function and the reject function.
48 * When we receive a message from the worker we get a map entry and resolve/reject the promise based on the message.
50 protected promiseMap
: Map
<
52 PromiseWorkerResponseWrapper
<Worker
, Response
>
53 > = new Map
<number, PromiseWorkerResponseWrapper
<Worker
, Response
>>()
56 * Id of the next message.
58 protected nextMessageId
: number = 0
61 * Worker choice strategy instance implementing the worker choice algorithm.
63 * Default to a strategy implementing a round robin algorithm.
65 protected workerChoiceStrategyContext
: WorkerChoiceStrategyContext
<
72 * Constructs a new poolifier pool.
74 * @param numberOfWorkers - Number of workers that this pool should manage.
75 * @param filePath - Path to the worker-file.
76 * @param opts - Options for the pool.
79 public readonly numberOfWorkers
: number,
80 public readonly filePath
: string,
81 public readonly opts
: PoolOptions
<Worker
>
84 throw new Error('Cannot start a pool from a worker!')
86 this.checkNumberOfWorkers(this.numberOfWorkers
)
87 this.checkFilePath(this.filePath
)
88 this.checkPoolOptions(this.opts
)
91 for (let i
= 1; i
<= this.numberOfWorkers
; i
++) {
92 this.createAndSetupWorker()
95 if (this.opts
.enableEvents
=== true) {
96 this.emitter
= new PoolEmitter()
98 this.workerChoiceStrategyContext
= new WorkerChoiceStrategyContext(
101 const workerCreated
= this.createAndSetupWorker()
102 this.registerWorkerMessageListener(workerCreated
, message
=> {
104 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
105 this.getWorkerRunningTasks(workerCreated
) === 0
107 // Kill received from the worker, means that no new tasks are submitted to that worker for a while ( > maxInactiveTime)
108 void (this.destroyWorker(workerCreated
) as Promise
<void>)
113 this.opts
.workerChoiceStrategy
117 private checkFilePath (filePath
: string): void {
118 if (filePath
== null || filePath
.length
=== 0) {
119 throw new Error('Please specify a file with a worker implementation')
123 private checkNumberOfWorkers (numberOfWorkers
: number): void {
124 if (numberOfWorkers
== null) {
126 'Cannot instantiate a pool without specifying the number of workers'
128 } else if (!Number.isSafeInteger(numberOfWorkers
)) {
130 'Cannot instantiate a pool with a non integer number of workers'
132 } else if (numberOfWorkers
< 0) {
133 throw new RangeError(
134 'Cannot instantiate a pool with a negative number of workers'
136 } else if (this.type === PoolType
.FIXED
&& numberOfWorkers
=== 0) {
137 throw new Error('Cannot instantiate a fixed pool with no worker')
141 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
142 this.opts
.workerChoiceStrategy
=
143 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
144 this.opts
.enableEvents
= opts
.enableEvents
?? true
148 public abstract get
type (): PoolType
151 public get
numberOfRunningTasks (): number {
152 return this.promiseMap
.size
156 public getWorkerIndex (worker
: Worker
): number {
157 return this.workers
.indexOf(worker
)
161 public getWorkerRunningTasks (worker
: Worker
): number | undefined {
162 return this.workersTasksUsage
.get(worker
)?.running
166 public getWorkerAverageTasksRunTime (worker
: Worker
): number | undefined {
167 return this.workersTasksUsage
.get(worker
)?.avgRunTime
171 public setWorkerChoiceStrategy (
172 workerChoiceStrategy
: WorkerChoiceStrategy
174 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
175 for (const worker
of this.workers
) {
176 this.resetWorkerTasksUsage(worker
)
178 this.workerChoiceStrategyContext
.setWorkerChoiceStrategy(
184 public abstract get
busy (): boolean
186 protected internalGetBusyStatus (): boolean {
188 this.numberOfRunningTasks
>= this.numberOfWorkers
&&
189 this.findFreeWorker() === false
194 public findFreeWorker (): Worker
| false {
195 for (const worker
of this.workers
) {
196 if (this.getWorkerRunningTasks(worker
) === 0) {
197 // A worker is free, return the matching worker
205 public async execute (data
: Data
): Promise
<Response
> {
206 // Configure worker to handle message with the specified task
207 const worker
= this.chooseWorker()
208 const res
= this.internalExecute(worker
, this.nextMessageId
)
209 this.checkAndEmitBusy()
210 this.sendToWorker(worker
, {
211 data
: data
?? (EMPTY_OBJECT_LITERAL
as Data
),
212 id
: this.nextMessageId
215 // eslint-disable-next-line @typescript-eslint/return-await
220 public async destroy (): Promise
<void> {
222 this.workers
.map(async worker
=> {
223 await this.destroyWorker(worker
)
229 * Shutdowns given worker.
231 * @param worker - A worker within `workers`.
233 protected abstract destroyWorker (worker
: Worker
): void | Promise
<void>
236 * Setup hook that can be overridden by a Poolifier pool implementation
237 * to run code before workers are created in the abstract constructor.
239 protected setupHook (): void {
244 * Should return whether the worker is the main worker or not.
246 protected abstract isMain (): boolean
249 * Hook executed before the worker task promise resolution.
252 * @param worker - The worker.
254 protected beforePromiseWorkerResponseHook (worker
: Worker
): void {
255 this.increaseWorkerRunningTasks(worker
)
259 * Hook executed after the worker task promise resolution.
262 * @param message - The received message.
263 * @param promise - The Promise response.
265 protected afterPromiseWorkerResponseHook (
266 message
: MessageValue
<Response
>,
267 promise
: PromiseWorkerResponseWrapper
<Worker
, Response
>
269 this.decreaseWorkerRunningTasks(promise
.worker
)
270 this.stepWorkerRunTasks(promise
.worker
, 1)
271 this.updateWorkerTasksRunTime(promise
.worker
, message
.taskRunTime
)
275 * Removes the given worker from the pool.
277 * @param worker - The worker that will be removed.
279 protected removeWorker (worker
: Worker
): void {
280 // Clean worker from data structure
281 this.workers
.splice(this.getWorkerIndex(worker
), 1)
282 this.removeWorkerTasksUsage(worker
)
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.workers
.push(worker
)
349 // Init worker tasks usage map
350 this.initWorkerTasksUsage(worker
)
352 this.afterWorkerSetup(worker
)
358 * This function is the listener registered for each worker.
360 * @returns The listener function to execute when a message is received from a worker.
362 protected workerListener (): (message
: MessageValue
<Response
>) => void {
364 if (message
.id
!== undefined) {
365 const promise
= this.promiseMap
.get(message
.id
)
366 if (promise
!== undefined) {
367 if (message
.error
!= null) {
368 promise
.reject(message
.error
)
370 promise
.resolve(message
.data
as Response
)
372 this.afterPromiseWorkerResponseHook(message
, promise
)
373 this.promiseMap
.delete(message
.id
)
379 private async internalExecute (
382 ): Promise
<Response
> {
383 this.beforePromiseWorkerResponseHook(worker
)
384 return await new Promise
<Response
>((resolve
, reject
) => {
385 this.promiseMap
.set(messageId
, { resolve
, reject
, worker
})
389 private checkAndEmitBusy (): void {
390 if (this.opts
.enableEvents
=== true && this.busy
) {
391 this.emitter
?.emit('busy')
396 * Increases the number of tasks that the given worker has applied.
398 * @param worker - Worker which running tasks is increased.
400 private increaseWorkerRunningTasks (worker
: Worker
): void {
401 this.stepWorkerRunningTasks(worker
, 1)
405 * Decreases the number of tasks that the given worker has applied.
407 * @param worker - Worker which running tasks is decreased.
409 private decreaseWorkerRunningTasks (worker
: Worker
): void {
410 this.stepWorkerRunningTasks(worker
, -1)
414 * Steps the number of tasks that the given worker has applied.
416 * @param worker - Worker which running tasks are stepped.
417 * @param step - Number of running tasks step.
419 private stepWorkerRunningTasks (worker
: Worker
, step
: number): void {
420 if (this.checkWorkerTasksUsage(worker
)) {
421 const tasksUsage
= this.workersTasksUsage
.get(worker
) as TasksUsage
422 tasksUsage
.running
= tasksUsage
.running
+ step
423 this.workersTasksUsage
.set(worker
, tasksUsage
)
428 * Steps the number of tasks that the given worker has run.
430 * @param worker - Worker which has run tasks.
431 * @param step - Number of run tasks step.
433 private stepWorkerRunTasks (worker
: Worker
, step
: number): void {
434 if (this.checkWorkerTasksUsage(worker
)) {
435 const tasksUsage
= this.workersTasksUsage
.get(worker
) as TasksUsage
436 tasksUsage
.run
= tasksUsage
.run
+ step
437 this.workersTasksUsage
.set(worker
, tasksUsage
)
442 * Updates tasks runtime for the given worker.
444 * @param worker - Worker which run the task.
445 * @param taskRunTime - Worker task runtime.
447 private updateWorkerTasksRunTime (
449 taskRunTime
: number | undefined
452 this.workerChoiceStrategyContext
.getWorkerChoiceStrategy()
453 .requiredStatistics
.runTime
&&
454 this.checkWorkerTasksUsage(worker
)
456 const tasksUsage
= this.workersTasksUsage
.get(worker
) as TasksUsage
457 tasksUsage
.runTime
+= taskRunTime
?? 0
458 if (tasksUsage
.run
!== 0) {
459 tasksUsage
.avgRunTime
= tasksUsage
.runTime
/ tasksUsage
.run
461 this.workersTasksUsage
.set(worker
, tasksUsage
)
466 * Checks if the given worker is registered in the workers tasks usage map.
468 * @param worker - Worker to check.
469 * @returns `true` if the worker is registered in the workers tasks usage map. `false` otherwise.
471 private checkWorkerTasksUsage (worker
: Worker
): boolean {
472 const hasTasksUsage
= this.workersTasksUsage
.has(worker
)
473 if (!hasTasksUsage
) {
474 throw new Error('Worker could not be found in workers tasks usage map')
480 * Initializes tasks usage statistics.
482 * @param worker - The worker.
484 private initWorkerTasksUsage (worker
: Worker
): void {
485 this.workersTasksUsage
.set(worker
, {
494 * Removes worker tasks usage statistics.
496 * @param worker - The worker.
498 private removeWorkerTasksUsage (worker
: Worker
): void {
499 this.workersTasksUsage
.delete(worker
)
503 * Resets worker tasks usage statistics.
505 * @param worker - The worker.
507 private resetWorkerTasksUsage (worker
: Worker
): void {
508 this.removeWorkerTasksUsage(worker
)
509 this.initWorkerTasksUsage(worker
)