3 PromiseWorkerResponseWrapper
4 } from
'../utility-types'
5 import { EMPTY_FUNCTION
} from
'../utils'
6 import { isKillBehavior
, KillBehaviors
} from
'../worker/worker-options'
7 import type { PoolOptions
} from
'./pool'
8 import type { IPoolInternal
, TasksUsage
} from
'./pool-internal'
9 import { PoolEmitter
, PoolType
} from
'./pool-internal'
10 import type { IPoolWorker
} from
'./pool-worker'
12 WorkerChoiceStrategies
,
14 } from
'./selection-strategies/selection-strategies-types'
15 import { WorkerChoiceStrategyContext
} from
'./selection-strategies/worker-choice-strategy-context'
17 const WORKER_NOT_FOUND_TASKS_USAGE_MAP
=
18 'Worker could not be found in worker tasks usage map'
21 * Base class that implements some shared logic for all poolifier pools.
23 * @template Worker Type of worker which manages this pool.
24 * @template Data Type of data sent to the worker. This can only be serializable data.
25 * @template Response Type of response of execution. This can only be serializable data.
27 export abstract class AbstractPool
<
28 Worker
extends IPoolWorker
,
31 > implements IPoolInternal
<Worker
, Data
, Response
> {
33 public readonly workers
: Worker
[] = []
36 public readonly workersTasksUsage
: Map
<Worker
, TasksUsage
> = new Map
<
42 public readonly emitter
?: PoolEmitter
45 public readonly max
?: number
50 * - `key`: This is the message Id of each submitted task.
51 * - `value`: An object that contains the worker, the resolve function and the reject function.
53 * When we receive a message from the worker we get a map entry and resolve/reject the promise based on the message.
55 protected promiseMap
: Map
<
57 PromiseWorkerResponseWrapper
<Worker
, Response
>
58 > = new Map
<number, PromiseWorkerResponseWrapper
<Worker
, Response
>>()
61 * Id of the next message.
63 protected nextMessageId
: number = 0
66 * Worker choice strategy instance implementing the worker choice algorithm.
68 * Default to a strategy implementing a round robin algorithm.
70 protected workerChoiceStrategyContext
: WorkerChoiceStrategyContext
<
77 * Constructs a new poolifier pool.
79 * @param numberOfWorkers Number of workers that this pool should manage.
80 * @param filePath Path to the worker-file.
81 * @param opts Options for the pool.
84 public readonly numberOfWorkers
: number,
85 public readonly filePath
: string,
86 public readonly opts
: PoolOptions
<Worker
>
89 throw new Error('Cannot start a pool from a worker!')
91 this.checkNumberOfWorkers(this.numberOfWorkers
)
92 this.checkFilePath(this.filePath
)
93 this.checkPoolOptions(this.opts
)
96 for (let i
= 1; i
<= this.numberOfWorkers
; i
++) {
97 this.createAndSetupWorker()
100 if (this.opts
.enableEvents
) {
101 this.emitter
= new PoolEmitter()
103 this.workerChoiceStrategyContext
= new WorkerChoiceStrategyContext(
106 const workerCreated
= this.createAndSetupWorker()
107 this.registerWorkerMessageListener(workerCreated
, message
=> {
109 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
110 this.getWorkerRunningTasks(workerCreated
) === 0
112 // Kill received from the worker, means that no new tasks are submitted to that worker for a while ( > maxInactiveTime)
113 this.destroyWorker(workerCreated
) as void
118 this.opts
.workerChoiceStrategy
122 private checkFilePath (filePath
: string): void {
124 throw new Error('Please specify a file with a worker implementation')
128 private checkNumberOfWorkers (numberOfWorkers
: number): void {
129 if (numberOfWorkers
== null) {
131 'Cannot instantiate a pool without specifying the number of workers'
133 } else if (Number.isSafeInteger(numberOfWorkers
) === false) {
135 'Cannot instantiate a pool with a non integer number of workers'
137 } else if (numberOfWorkers
< 0) {
139 'Cannot instantiate a pool with a negative number of workers'
141 } else if (this.type === PoolType
.FIXED
&& numberOfWorkers
=== 0) {
142 throw new Error('Cannot instantiate a fixed pool with no worker')
146 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
147 this.opts
.workerChoiceStrategy
=
148 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
149 this.opts
.enableEvents
= opts
.enableEvents
?? true
153 public abstract get
type (): PoolType
156 public get
numberOfRunningTasks (): number {
157 return this.promiseMap
.size
161 public getWorkerIndex (worker
: Worker
): number {
162 return this.workers
.indexOf(worker
)
166 public getWorkerRunningTasks (worker
: Worker
): number | undefined {
167 return this.workersTasksUsage
.get(worker
)?.running
171 public getWorkerAverageTasksRunTime (worker
: Worker
): number | undefined {
172 return this.workersTasksUsage
.get(worker
)?.avgRunTime
176 public setWorkerChoiceStrategy (
177 workerChoiceStrategy
: WorkerChoiceStrategy
179 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
180 for (const worker
of this.workers
) {
181 this.resetWorkerTasksUsage(worker
)
183 this.workerChoiceStrategyContext
.setWorkerChoiceStrategy(
189 public abstract get
busy (): boolean
191 protected internalGetBusyStatus (): boolean {
193 this.numberOfRunningTasks
>= this.numberOfWorkers
&&
194 this.findFreeWorker() === false
199 public findFreeWorker (): Worker
| false {
200 for (const worker
of this.workers
) {
201 if (this.getWorkerRunningTasks(worker
) === 0) {
202 // A worker is free, return the matching worker
210 public execute (data
: Data
): Promise
<Response
> {
211 // Configure worker to handle message with the specified task
212 const worker
= this.chooseWorker()
213 const messageId
= ++this.nextMessageId
214 const res
= this.internalExecute(worker
, messageId
)
215 this.checkAndEmitBusy()
216 data
= data
?? ({} as Data
)
217 this.sendToWorker(worker
, { data
, id
: messageId
})
222 public async destroy (): Promise
<void> {
223 await Promise
.all(this.workers
.map(worker
=> this.destroyWorker(worker
)))
227 * Shutdowns given worker.
229 * @param worker A worker within `workers`.
231 protected abstract destroyWorker (worker
: Worker
): void | Promise
<void>
234 * Setup hook that can be overridden by a Poolifier pool implementation
235 * to run code before workers are created in the abstract constructor.
237 protected setupHook (): void {
242 * Should return whether the worker is the main worker or not.
244 protected abstract isMain (): boolean
247 * Hook executed before the worker task promise resolution.
250 * @param worker The worker.
252 protected beforePromiseWorkerResponseHook (worker
: Worker
): void {
253 this.increaseWorkerRunningTasks(worker
)
257 * Hook executed after the worker task promise resolution.
260 * @param message The received message.
261 * @param promise The Promise response.
263 protected afterPromiseWorkerResponseHook (
264 message
: MessageValue
<Response
>,
265 promise
: PromiseWorkerResponseWrapper
<Worker
, Response
>
267 this.decreaseWorkerRunningTasks(promise
.worker
)
268 this.stepWorkerRunTasks(promise
.worker
, 1)
269 this.updateWorkerTasksRunTime(promise
.worker
, message
.taskRunTime
)
273 * Removes the given worker from the pool.
275 * @param worker Worker that will be removed.
277 protected removeWorker (worker
: Worker
): void {
278 // Clean worker from data structure
279 this.workers
.splice(this.getWorkerIndex(worker
), 1)
280 this.removeWorkerTasksUsage(worker
)
284 * Chooses a worker for the next task.
286 * The default implementation uses a round robin algorithm to distribute the load.
290 protected chooseWorker (): Worker
{
291 return this.workerChoiceStrategyContext
.execute()
295 * Sends a message to the given worker.
297 * @param worker The worker which should receive the message.
298 * @param message The message.
300 protected abstract sendToWorker (
302 message
: MessageValue
<Data
>
306 * Registers a listener callback on a given worker.
308 * @param worker A worker.
309 * @param listener A message listener callback.
311 protected abstract registerWorkerMessageListener
<
312 Message
extends Data
| Response
313 > (worker
: Worker
, listener
: (message
: MessageValue
<Message
>) => void): void
315 protected internalExecute (
318 ): Promise
<Response
> {
319 this.beforePromiseWorkerResponseHook(worker
)
320 return new Promise
<Response
>((resolve
, reject
) => {
321 this.promiseMap
.set(messageId
, { resolve
, reject
, worker
})
326 * Returns a newly created worker.
328 protected abstract createWorker (): Worker
331 * Function that can be hooked up when a worker has been newly created and moved to the workers registry.
333 * Can be used to update the `maxListeners` or binding the `main-worker`<->`worker` connection if not bind by default.
335 * @param worker The newly created worker.
337 protected abstract afterWorkerSetup (worker
: Worker
): void
340 * Creates a new worker for this pool and sets it up completely.
342 * @returns New, completely set up worker.
344 protected createAndSetupWorker (): Worker
{
345 const worker
= this.createWorker()
347 worker
.on('message', this.opts
.messageHandler
?? EMPTY_FUNCTION
)
348 worker
.on('error', this.opts
.errorHandler
?? EMPTY_FUNCTION
)
349 worker
.on('online', this.opts
.onlineHandler
?? EMPTY_FUNCTION
)
350 worker
.on('exit', this.opts
.exitHandler
?? EMPTY_FUNCTION
)
351 worker
.once('exit', () => this.removeWorker(worker
))
353 this.workers
.push(worker
)
355 // Init worker tasks usage map
356 this.initWorkerTasksUsage(worker
)
358 this.afterWorkerSetup(worker
)
364 * This function is the listener registered for each worker.
366 * @returns The listener function to execute when a message is received from a worker.
368 protected workerListener (): (message
: MessageValue
<Response
>) => void {
370 if (message
.id
!== undefined) {
371 const promise
= this.promiseMap
.get(message
.id
)
372 if (promise
!== undefined) {
373 this.afterPromiseWorkerResponseHook(message
, promise
)
374 if (message
.error
) promise
.reject(message
.error
)
375 else promise
.resolve(message
.data
as Response
)
376 this.promiseMap
.delete(message
.id
)
382 private checkAndEmitBusy (): void {
383 if (this.opts
.enableEvents
&& this.busy
) {
384 this.emitter
?.emit('busy')
389 * Increases the number of tasks that the given worker has applied.
391 * @param worker Worker which running tasks is increased.
393 private increaseWorkerRunningTasks (worker
: Worker
): void {
394 this.stepWorkerRunningTasks(worker
, 1)
398 * Decreases the number of tasks that the given worker has applied.
400 * @param worker Worker which running tasks is decreased.
402 private decreaseWorkerRunningTasks (worker
: Worker
): void {
403 this.stepWorkerRunningTasks(worker
, -1)
407 * Steps the number of tasks that the given worker has applied.
409 * @param worker Worker which running tasks are stepped.
410 * @param step Number of running tasks step.
412 private stepWorkerRunningTasks (worker
: Worker
, step
: number): void {
413 const tasksUsage
= this.workersTasksUsage
.get(worker
)
414 if (tasksUsage
!== undefined) {
415 tasksUsage
.running
= tasksUsage
.running
+ step
416 this.workersTasksUsage
.set(worker
, tasksUsage
)
418 throw new Error(WORKER_NOT_FOUND_TASKS_USAGE_MAP
)
423 * Steps the number of tasks that the given worker has run.
425 * @param worker Worker which has run tasks.
426 * @param step Number of run tasks step.
428 private stepWorkerRunTasks (worker
: Worker
, step
: number): void {
429 const tasksUsage
= this.workersTasksUsage
.get(worker
)
430 if (tasksUsage
!== undefined) {
431 tasksUsage
.run
= tasksUsage
.run
+ step
432 this.workersTasksUsage
.set(worker
, tasksUsage
)
434 throw new Error(WORKER_NOT_FOUND_TASKS_USAGE_MAP
)
439 * Updates tasks runtime for the given worker.
441 * @param worker Worker which run the task.
442 * @param taskRunTime Worker task runtime.
444 private updateWorkerTasksRunTime (
446 taskRunTime
: number | undefined
449 this.workerChoiceStrategyContext
.getWorkerChoiceStrategy()
450 .requiredStatistics
.runTime
=== true
452 const tasksUsage
= this.workersTasksUsage
.get(worker
)
453 if (tasksUsage
!== undefined) {
454 tasksUsage
.runTime
+= taskRunTime
?? 0
455 if (tasksUsage
.run
!== 0) {
456 tasksUsage
.avgRunTime
= tasksUsage
.runTime
/ tasksUsage
.run
458 this.workersTasksUsage
.set(worker
, tasksUsage
)
460 throw new Error(WORKER_NOT_FOUND_TASKS_USAGE_MAP
)
466 * Initializes tasks usage statistics.
468 * @param worker The worker.
470 initWorkerTasksUsage (worker
: Worker
): void {
471 this.workersTasksUsage
.set(worker
, {
480 * Removes worker tasks usage statistics.
482 * @param worker The worker.
484 private removeWorkerTasksUsage (worker
: Worker
): void {
485 this.workersTasksUsage
.delete(worker
)
489 * Resets worker tasks usage statistics.
491 * @param worker The worker.
493 private resetWorkerTasksUsage (worker
: Worker
): void {
494 this.removeWorkerTasksUsage(worker
)
495 this.initWorkerTasksUsage(worker
)