Update dependencies. (#318)
[poolifier.git] / src / pools / abstract-pool.ts
CommitLineData
be0676b3
APA
1import type {
2 MessageValue,
3 PromiseWorkerResponseWrapper
4} from '../utility-types'
6e9d10db 5import { EMPTY_FUNCTION } from '../utils'
4a6952ff 6import { isKillBehavior, KillBehaviors } from '../worker/worker-options'
a35560ba 7import type { IPoolInternal } from './pool-internal'
7c0ba920 8import { PoolEmitter, PoolType } from './pool-internal'
a35560ba
S
9import type { WorkerChoiceStrategy } from './selection-strategies'
10import {
11 WorkerChoiceStrategies,
12 WorkerChoiceStrategyContext
13} from './selection-strategies'
c97c7edb 14
729c563d
S
15/**
16 * Callback invoked if the worker raised an error.
17 */
c97c7edb 18export type ErrorHandler<Worker> = (this: Worker, e: Error) => void
729c563d
S
19
20/**
21 * Callback invoked when the worker has started successfully.
22 */
c97c7edb 23export type OnlineHandler<Worker> = (this: Worker) => void
729c563d
S
24
25/**
26 * Callback invoked when the worker exits successfully.
27 */
c97c7edb
S
28export type ExitHandler<Worker> = (this: Worker, code: number) => void
29
729c563d
S
30/**
31 * Basic interface that describes the minimum required implementation of listener events for a pool-worker.
32 */
c97c7edb 33export interface IWorker {
3832ad95
S
34 /**
35 * Register a listener to the error event.
36 *
37 * @param event `'error'`.
38 * @param handler The error handler.
39 */
c97c7edb 40 on(event: 'error', handler: ErrorHandler<this>): void
3832ad95
S
41 /**
42 * Register a listener to the online event.
43 *
44 * @param event `'online'`.
45 * @param handler The online handler.
46 */
c97c7edb 47 on(event: 'online', handler: OnlineHandler<this>): void
3832ad95
S
48 /**
49 * Register a listener to the exit event.
50 *
51 * @param event `'exit'`.
52 * @param handler The exit handler.
53 */
c97c7edb 54 on(event: 'exit', handler: ExitHandler<this>): void
3832ad95
S
55 /**
56 * Register a listener to the exit event that will only performed once.
57 *
58 * @param event `'exit'`.
59 * @param handler The exit handler.
60 */
45dbbb14 61 once(event: 'exit', handler: ExitHandler<this>): void
c97c7edb
S
62}
63
729c563d
S
64/**
65 * Options for a poolifier pool.
66 */
c97c7edb
S
67export interface PoolOptions<Worker> {
68 /**
69 * A function that will listen for error event on each worker.
70 */
71 errorHandler?: ErrorHandler<Worker>
72 /**
73 * A function that will listen for online event on each worker.
74 */
75 onlineHandler?: OnlineHandler<Worker>
76 /**
77 * A function that will listen for exit event on each worker.
78 */
79 exitHandler?: ExitHandler<Worker>
a35560ba
S
80 /**
81 * The work choice strategy to use in this pool.
82 */
83 workerChoiceStrategy?: WorkerChoiceStrategy
7c0ba920
JB
84 /**
85 * Pool events emission.
86 *
87 * Default to true.
88 */
89 enableEvents?: boolean
c97c7edb
S
90}
91
729c563d
S
92/**
93 * Base class containing some shared logic for all poolifier pools.
94 *
95 * @template Worker Type of worker which manages this pool.
deb85c12
JB
96 * @template Data Type of data sent to the worker. This can only be serializable data.
97 * @template Response Type of response of execution. This can only be serializable data.
729c563d 98 */
c97c7edb
S
99export abstract class AbstractPool<
100 Worker extends IWorker,
d3c8a1a8
S
101 Data = unknown,
102 Response = unknown
a35560ba 103> implements IPoolInternal<Worker, Data, Response> {
4a6952ff
JB
104 /** @inheritdoc */
105 public readonly workers: Worker[] = []
106
107 /** @inheritdoc */
108 public readonly tasks: Map<Worker, number> = new Map<Worker, number>()
109
110 /** @inheritdoc */
7c0ba920
JB
111 public readonly emitter?: PoolEmitter
112
113 /** @inheritdoc */
114 public readonly max?: number
4a6952ff 115
be0676b3
APA
116 /**
117 * The promise map.
118 *
119 * - `key`: This is the message ID of each submitted task.
120 * - `value`: An object that contains the worker, the resolve function and the reject function.
121 *
122 * When we receive a message from the worker we get a map entry and resolve/reject the promise based on the message.
123 */
124 protected promiseMap: Map<
125 number,
126 PromiseWorkerResponseWrapper<Worker, Response>
127 > = new Map<number, PromiseWorkerResponseWrapper<Worker, Response>>()
128
729c563d
S
129 /**
130 * ID of the next message.
131 */
280c2a77 132 protected nextMessageId: number = 0
c97c7edb 133
a35560ba
S
134 /**
135 * Worker choice strategy instance implementing the worker choice algorithm.
136 *
137 * Default to a strategy implementing a round robin algorithm.
138 */
139 protected workerChoiceStrategyContext: WorkerChoiceStrategyContext<
140 Worker,
141 Data,
142 Response
143 >
144
729c563d
S
145 /**
146 * Constructs a new poolifier pool.
147 *
5c5a1fb7 148 * @param numberOfWorkers Number of workers that this pool should manage.
729c563d 149 * @param filePath Path to the worker-file.
1927ee67 150 * @param opts Options for the pool.
729c563d 151 */
c97c7edb 152 public constructor (
5c5a1fb7 153 public readonly numberOfWorkers: number,
c97c7edb 154 public readonly filePath: string,
1927ee67 155 public readonly opts: PoolOptions<Worker>
c97c7edb
S
156 ) {
157 if (!this.isMain()) {
158 throw new Error('Cannot start a pool from a worker!')
159 }
8d3782fa 160 this.checkNumberOfWorkers(this.numberOfWorkers)
c510fea7 161 this.checkFilePath(this.filePath)
7c0ba920 162 this.checkPoolOptions(this.opts)
c97c7edb
S
163 this.setupHook()
164
5c5a1fb7 165 for (let i = 1; i <= this.numberOfWorkers; i++) {
280c2a77 166 this.createAndSetupWorker()
c97c7edb
S
167 }
168
7c0ba920
JB
169 if (this.opts.enableEvents) {
170 this.emitter = new PoolEmitter()
171 }
a35560ba
S
172 this.workerChoiceStrategyContext = new WorkerChoiceStrategyContext(
173 this,
4a6952ff
JB
174 () => {
175 const workerCreated = this.createAndSetupWorker()
176 this.registerWorkerMessageListener(workerCreated, message => {
177 const tasksInProgress = this.tasks.get(workerCreated)
178 if (
179 isKillBehavior(KillBehaviors.HARD, message.kill) ||
180 tasksInProgress === 0
181 ) {
182 // Kill received from the worker, means that no new tasks are submitted to that worker for a while ( > maxInactiveTime)
183 void this.destroyWorker(workerCreated)
184 }
185 })
186 return workerCreated
187 },
e843b904 188 this.opts.workerChoiceStrategy
a35560ba 189 )
c97c7edb
S
190 }
191
a35560ba 192 private checkFilePath (filePath: string): void {
c510fea7
APA
193 if (!filePath) {
194 throw new Error('Please specify a file with a worker implementation')
195 }
196 }
197
8d3782fa
JB
198 private checkNumberOfWorkers (numberOfWorkers: number): void {
199 if (numberOfWorkers == null) {
200 throw new Error(
201 'Cannot instantiate a pool without specifying the number of workers'
202 )
203 } else if (!Number.isSafeInteger(numberOfWorkers)) {
204 throw new Error(
205 'Cannot instantiate a pool with a non integer number of workers'
206 )
207 } else if (numberOfWorkers < 0) {
208 throw new Error(
209 'Cannot instantiate a pool with a negative number of workers'
210 )
7c0ba920 211 } else if (this.type === PoolType.FIXED && numberOfWorkers === 0) {
8d3782fa
JB
212 throw new Error('Cannot instantiate a fixed pool with no worker')
213 }
214 }
215
7c0ba920 216 private checkPoolOptions (opts: PoolOptions<Worker>): void {
e843b904
JB
217 this.opts.workerChoiceStrategy =
218 opts.workerChoiceStrategy ?? WorkerChoiceStrategies.ROUND_ROBIN
7c0ba920
JB
219 this.opts.enableEvents = opts.enableEvents ?? true
220 }
221
a35560ba 222 /** @inheritdoc */
7c0ba920
JB
223 public abstract get type (): PoolType
224
225 /** @inheritdoc */
226 public get numberOfRunningTasks (): number {
227 return this.promiseMap.size
a35560ba
S
228 }
229
230 /** @inheritdoc */
231 public setWorkerChoiceStrategy (
232 workerChoiceStrategy: WorkerChoiceStrategy
233 ): void {
b98ec2e6 234 this.opts.workerChoiceStrategy = workerChoiceStrategy
a35560ba
S
235 this.workerChoiceStrategyContext.setWorkerChoiceStrategy(
236 workerChoiceStrategy
237 )
238 }
239
7c0ba920
JB
240 /** @inheritdoc */
241 public abstract get busy (): boolean
242
243 protected internalGetBusyStatus (): boolean {
244 return (
245 this.numberOfRunningTasks >= this.numberOfWorkers &&
246 this.findFreeTasksMapEntry() === false
247 )
248 }
249
250 /** @inheritdoc */
251 public findFreeTasksMapEntry (): [Worker, number] | false {
252 for (const [worker, numberOfTasks] of this.tasks) {
253 if (numberOfTasks === 0) {
254 // A worker is free, return the matching tasks map entry
255 return [worker, numberOfTasks]
256 }
257 }
258 return false
259 }
260
a35560ba 261 /** @inheritdoc */
280c2a77
S
262 public execute (data: Data): Promise<Response> {
263 // Configure worker to handle message with the specified task
264 const worker = this.chooseWorker()
265 this.increaseWorkersTask(worker)
7c0ba920 266 this.checkAndEmitBusy()
280c2a77
S
267 const messageId = ++this.nextMessageId
268 const res = this.internalExecute(worker, messageId)
269 this.sendToWorker(worker, { data: data || ({} as Data), id: messageId })
270 return res
271 }
c97c7edb 272
a35560ba 273 /** @inheritdoc */
c97c7edb 274 public async destroy (): Promise<void> {
45dbbb14 275 await Promise.all(this.workers.map(worker => this.destroyWorker(worker)))
c97c7edb
S
276 }
277
4a6952ff
JB
278 /**
279 * Shut down given worker.
280 *
281 * @param worker A worker within `workers`.
282 */
283 protected abstract destroyWorker (worker: Worker): void | Promise<void>
c97c7edb 284
729c563d 285 /**
280c2a77
S
286 * Setup hook that can be overridden by a Poolifier pool implementation
287 * to run code before workers are created in the abstract constructor.
729c563d 288 */
280c2a77
S
289 protected setupHook (): void {
290 // Can be overridden
291 }
c97c7edb 292
729c563d 293 /**
280c2a77
S
294 * Should return whether the worker is the main worker or not.
295 */
296 protected abstract isMain (): boolean
297
298 /**
7c0ba920 299 * Increase the number of tasks that the given workers has applied.
729c563d 300 *
f416c098 301 * @param worker Worker whose tasks are increased.
729c563d 302 */
280c2a77 303 protected increaseWorkersTask (worker: Worker): void {
f416c098 304 this.stepWorkerNumberOfTasks(worker, 1)
c97c7edb
S
305 }
306
c01733f1 307 /**
7c0ba920 308 * Decrease the number of tasks that the given workers has applied.
c01733f1 309 *
f416c098 310 * @param worker Worker whose tasks are decreased.
c01733f1 311 */
312 protected decreaseWorkersTasks (worker: Worker): void {
f416c098
JB
313 this.stepWorkerNumberOfTasks(worker, -1)
314 }
315
316 /**
7c0ba920 317 * Step the number of tasks that the given workers has applied.
f416c098
JB
318 *
319 * @param worker Worker whose tasks are set.
320 * @param step Worker number of tasks step.
321 */
a35560ba 322 private stepWorkerNumberOfTasks (worker: Worker, step: number): void {
d63d3be3 323 const numberOfTasksInProgress = this.tasks.get(worker)
324 if (numberOfTasksInProgress !== undefined) {
f416c098 325 this.tasks.set(worker, numberOfTasksInProgress + step)
c01733f1 326 } else {
327 throw Error('Worker could not be found in tasks map')
328 }
329 }
330
729c563d
S
331 /**
332 * Removes the given worker from the pool.
333 *
334 * @param worker Worker that will be removed.
335 */
f2fdaa86
JB
336 protected removeWorker (worker: Worker): void {
337 // Clean worker from data structure
338 const workerIndex = this.workers.indexOf(worker)
339 this.workers.splice(workerIndex, 1)
340 this.tasks.delete(worker)
341 }
342
280c2a77
S
343 /**
344 * Choose a worker for the next task.
345 *
346 * The default implementation uses a round robin algorithm to distribute the load.
347 *
348 * @returns Worker.
349 */
350 protected chooseWorker (): Worker {
a35560ba 351 return this.workerChoiceStrategyContext.execute()
c97c7edb
S
352 }
353
280c2a77
S
354 /**
355 * Send a message to the given worker.
356 *
357 * @param worker The worker which should receive the message.
358 * @param message The message.
359 */
360 protected abstract sendToWorker (
361 worker: Worker,
362 message: MessageValue<Data>
363 ): void
364
4a6952ff
JB
365 /**
366 * Register a listener callback on a given worker.
367 *
368 * @param worker A worker.
369 * @param listener A message listener callback.
370 */
371 protected abstract registerWorkerMessageListener<
4f7fa42a
S
372 Message extends Data | Response
373 > (worker: Worker, listener: (message: MessageValue<Message>) => void): void
c97c7edb 374
280c2a77
S
375 protected internalExecute (
376 worker: Worker,
377 messageId: number
378 ): Promise<Response> {
be0676b3
APA
379 return new Promise<Response>((resolve, reject) => {
380 this.promiseMap.set(messageId, { resolve, reject, worker })
c97c7edb
S
381 })
382 }
383
729c563d
S
384 /**
385 * Returns a newly created worker.
386 */
280c2a77 387 protected abstract createWorker (): Worker
c97c7edb 388
729c563d
S
389 /**
390 * Function that can be hooked up when a worker has been newly created and moved to the workers registry.
391 *
392 * Can be used to update the `maxListeners` or binding the `main-worker`<->`worker` connection if not bind by default.
393 *
394 * @param worker The newly created worker.
395 */
280c2a77 396 protected abstract afterWorkerSetup (worker: Worker): void
c97c7edb 397
4a6952ff
JB
398 /**
399 * Creates a new worker for this pool and sets it up completely.
400 *
401 * @returns New, completely set up worker.
402 */
403 protected createAndSetupWorker (): Worker {
280c2a77
S
404 const worker: Worker = this.createWorker()
405
a35560ba
S
406 worker.on('error', this.opts.errorHandler ?? EMPTY_FUNCTION)
407 worker.on('online', this.opts.onlineHandler ?? EMPTY_FUNCTION)
408 worker.on('exit', this.opts.exitHandler ?? EMPTY_FUNCTION)
45dbbb14 409 worker.once('exit', () => this.removeWorker(worker))
280c2a77 410
c97c7edb 411 this.workers.push(worker)
280c2a77
S
412
413 // Init tasks map
c97c7edb 414 this.tasks.set(worker, 0)
280c2a77
S
415
416 this.afterWorkerSetup(worker)
417
c97c7edb
S
418 return worker
419 }
be0676b3
APA
420
421 /**
422 * This function is the listener registered for each worker.
423 *
424 * @returns The listener function to execute when a message is sent from a worker.
425 */
426 protected workerListener (): (message: MessageValue<Response>) => void {
4a6952ff 427 return message => {
be0676b3
APA
428 if (message.id) {
429 const value = this.promiseMap.get(message.id)
430 if (value) {
431 this.decreaseWorkersTasks(value.worker)
432 if (message.error) value.reject(message.error)
433 else value.resolve(message.data as Response)
434 this.promiseMap.delete(message.id)
435 }
436 }
437 }
be0676b3 438 }
7c0ba920
JB
439
440 private checkAndEmitBusy (): void {
441 if (this.opts.enableEvents && this.busy) {
442 this.emitter?.emit('busy')
443 }
444 }
c97c7edb 445}