3 PromiseWorkerResponseWrapper
4 } from
'../utility-types'
5 import { EMPTY_FUNCTION
} from
'../utils'
6 import { isKillBehavior
, KillBehaviors
} from
'../worker/worker-options'
7 import type { AbstractPoolWorker
} from
'./abstract-pool-worker'
8 import type { PoolOptions
} from
'./pool'
9 import type { IPoolInternal
} from
'./pool-internal'
10 import { PoolEmitter
, PoolType
} from
'./pool-internal'
12 WorkerChoiceStrategies
,
14 } from
'./selection-strategies/selection-strategies-types'
15 import { WorkerChoiceStrategyContext
} from
'./selection-strategies/worker-choice-strategy-context'
18 * Base class containing some shared logic for all poolifier pools.
20 * @template Worker Type of worker which manages this pool.
21 * @template Data Type of data sent to the worker. This can only be serializable data.
22 * @template Response Type of response of execution. This can only be serializable data.
24 export abstract class AbstractPool
<
25 Worker
extends AbstractPoolWorker
,
28 > implements IPoolInternal
<Worker
, Data
, Response
> {
30 public readonly workers
: Worker
[] = []
33 public readonly tasks
: Map
<Worker
, number> = new Map
<Worker
, number>()
36 public readonly emitter
?: PoolEmitter
39 public readonly max
?: number
44 * - `key`: This is the message Id of each submitted task.
45 * - `value`: An object that contains the worker, the resolve function and the reject function.
47 * When we receive a message from the worker we get a map entry and resolve/reject the promise based on the message.
49 protected promiseMap
: Map
<
51 PromiseWorkerResponseWrapper
<Worker
, Response
>
52 > = new Map
<number, PromiseWorkerResponseWrapper
<Worker
, Response
>>()
55 * Id of the next message.
57 protected nextMessageId
: number = 0
60 * Worker choice strategy instance implementing the worker choice algorithm.
62 * Default to a strategy implementing a round robin algorithm.
64 protected workerChoiceStrategyContext
: WorkerChoiceStrategyContext
<
71 * Constructs a new poolifier pool.
73 * @param numberOfWorkers Number of workers that this pool should manage.
74 * @param filePath Path to the worker-file.
75 * @param opts Options for the pool.
78 public readonly numberOfWorkers
: number,
79 public readonly filePath
: string,
80 public readonly opts
: PoolOptions
<Worker
>
83 throw new Error('Cannot start a pool from a worker!')
85 this.checkNumberOfWorkers(this.numberOfWorkers
)
86 this.checkFilePath(this.filePath
)
87 this.checkPoolOptions(this.opts
)
90 for (let i
= 1; i
<= this.numberOfWorkers
; i
++) {
91 this.createAndSetupWorker()
94 if (this.opts
.enableEvents
) {
95 this.emitter
= new PoolEmitter()
97 this.workerChoiceStrategyContext
= new WorkerChoiceStrategyContext(
100 const workerCreated
= this.createAndSetupWorker()
101 this.registerWorkerMessageListener(workerCreated
, message
=> {
103 isKillBehavior(KillBehaviors
.HARD
, message
.kill
) ||
104 this.getWorkerRunningTasks(workerCreated
) === 0
106 // Kill received from the worker, means that no new tasks are submitted to that worker for a while ( > maxInactiveTime)
107 this.destroyWorker(workerCreated
) as void
112 this.opts
.workerChoiceStrategy
116 private checkFilePath (filePath
: string): void {
118 throw new Error('Please specify a file with a worker implementation')
122 private checkNumberOfWorkers (numberOfWorkers
: number): void {
123 if (numberOfWorkers
== null) {
125 'Cannot instantiate a pool without specifying the number of workers'
127 } else if (!Number.isSafeInteger(numberOfWorkers
)) {
129 'Cannot instantiate a pool with a non integer number of workers'
131 } else if (numberOfWorkers
< 0) {
133 'Cannot instantiate a pool with a negative number of workers'
135 } else if (this.type === PoolType
.FIXED
&& numberOfWorkers
=== 0) {
136 throw new Error('Cannot instantiate a fixed pool with no worker')
140 private checkPoolOptions (opts
: PoolOptions
<Worker
>): void {
141 this.opts
.workerChoiceStrategy
=
142 opts
.workerChoiceStrategy
?? WorkerChoiceStrategies
.ROUND_ROBIN
143 this.opts
.enableEvents
= opts
.enableEvents
?? true
147 public abstract get
type (): PoolType
150 public get
numberOfRunningTasks (): number {
151 return this.promiseMap
.size
155 public getWorkerRunningTasks (worker
: Worker
): number | undefined {
156 return this.tasks
.get(worker
)
160 public getWorkerIndex (worker
: Worker
): number {
161 return this.workers
.indexOf(worker
)
165 public setWorkerChoiceStrategy (
166 workerChoiceStrategy
: WorkerChoiceStrategy
168 this.opts
.workerChoiceStrategy
= workerChoiceStrategy
169 this.workerChoiceStrategyContext
.setWorkerChoiceStrategy(
175 public abstract get
busy (): boolean
177 protected internalGetBusyStatus (): boolean {
179 this.numberOfRunningTasks
>= this.numberOfWorkers
&&
180 this.findFreeWorker() === false
185 public findFreeWorker (): Worker
| false {
186 for (const worker
of this.workers
) {
187 if (this.getWorkerRunningTasks(worker
) === 0) {
188 // A worker is free, return the matching worker
196 public execute (data
: Data
): Promise
<Response
> {
197 // Configure worker to handle message with the specified task
198 const worker
= this.chooseWorker()
199 const messageId
= ++this.nextMessageId
200 const res
= this.internalExecute(worker
, messageId
)
201 this.checkAndEmitBusy()
202 data
= data
?? ({} as Data
)
203 this.sendToWorker(worker
, { data
, id
: messageId
})
208 public async destroy (): Promise
<void> {
209 await Promise
.all(this.workers
.map(worker
=> this.destroyWorker(worker
)))
213 * Shut down given worker.
215 * @param worker A worker within `workers`.
217 protected abstract destroyWorker (worker
: Worker
): void | Promise
<void>
220 * Setup hook that can be overridden by a Poolifier pool implementation
221 * to run code before workers are created in the abstract constructor.
223 protected setupHook (): void {
228 * Should return whether the worker is the main worker or not.
230 protected abstract isMain (): boolean
233 * Increase the number of tasks that the given worker has applied.
235 * @param worker Worker whose tasks are increased.
237 protected increaseWorkersTask (worker
: Worker
): void {
238 this.stepWorkerNumberOfTasks(worker
, 1)
242 * Decrease the number of tasks that the given worker has applied.
244 * @param worker Worker whose tasks are decreased.
246 protected decreaseWorkersTasks (worker
: Worker
): void {
247 this.stepWorkerNumberOfTasks(worker
, -1)
251 * Step the number of tasks that the given worker has applied.
253 * @param worker Worker whose tasks are set.
254 * @param step Worker number of tasks step.
256 private stepWorkerNumberOfTasks (worker
: Worker
, step
: number): void {
257 const numberOfTasksInProgress
= this.tasks
.get(worker
)
258 if (numberOfTasksInProgress
!== undefined) {
259 this.tasks
.set(worker
, numberOfTasksInProgress
+ step
)
261 throw Error('Worker could not be found in tasks map')
266 * Removes the given worker from the pool.
268 * @param worker Worker that will be removed.
270 protected removeWorker (worker
: Worker
): void {
271 // Clean worker from data structure
272 this.workers
.splice(this.getWorkerIndex(worker
), 1)
273 this.tasks
.delete(worker
)
277 * Choose a worker for the next task.
279 * The default implementation uses a round robin algorithm to distribute the load.
283 protected chooseWorker (): Worker
{
284 return this.workerChoiceStrategyContext
.execute()
288 * Send a message to the given worker.
290 * @param worker The worker which should receive the message.
291 * @param message The message.
293 protected abstract sendToWorker (
295 message
: MessageValue
<Data
>
299 * Register a listener callback on a given worker.
301 * @param worker A worker.
302 * @param listener A message listener callback.
304 protected abstract registerWorkerMessageListener
<
305 Message
extends Data
| Response
306 > (worker
: Worker
, listener
: (message
: MessageValue
<Message
>) => void): void
308 protected internalExecute (
311 ): Promise
<Response
> {
312 this.increaseWorkersTask(worker
)
313 return new Promise
<Response
>((resolve
, reject
) => {
314 this.promiseMap
.set(messageId
, { resolve
, reject
, worker
})
319 * Returns a newly created worker.
321 protected abstract createWorker (): Worker
324 * Function that can be hooked up when a worker has been newly created and moved to the workers registry.
326 * Can be used to update the `maxListeners` or binding the `main-worker`<->`worker` connection if not bind by default.
328 * @param worker The newly created worker.
330 protected abstract afterWorkerSetup (worker
: Worker
): void
333 * Creates a new worker for this pool and sets it up completely.
335 * @returns New, completely set up worker.
337 protected createAndSetupWorker (): Worker
{
338 const worker
= this.createWorker()
340 worker
.on('message', this.opts
.messageHandler
?? EMPTY_FUNCTION
)
341 worker
.on('error', this.opts
.errorHandler
?? EMPTY_FUNCTION
)
342 worker
.on('online', this.opts
.onlineHandler
?? EMPTY_FUNCTION
)
343 worker
.on('exit', this.opts
.exitHandler
?? EMPTY_FUNCTION
)
344 worker
.once('exit', () => this.removeWorker(worker
))
346 this.workers
.push(worker
)
349 this.tasks
.set(worker
, 0)
351 this.afterWorkerSetup(worker
)
357 * This function is the listener registered for each worker.
359 * @returns The listener function to execute when a message is received from a worker.
361 protected workerListener (): (message
: MessageValue
<Response
>) => void {
363 if (message
.id
!== undefined) {
364 const promise
= this.promiseMap
.get(message
.id
)
365 if (promise
!== undefined) {
366 this.decreaseWorkersTasks(promise
.worker
)
367 if (message
.error
) promise
.reject(message
.error
)
368 else promise
.resolve(message
.data
as Response
)
369 this.promiseMap
.delete(message
.id
)
375 private checkAndEmitBusy (): void {
376 if (this.opts
.enableEvents
&& this.busy
) {
377 this.emitter
?.emit('busy')