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