docs: refine tasks response handling documentation
[poolifier.git] / src / pools / abstract-pool.ts
1 import crypto from 'node:crypto'
2 import type { MessageValue, PromiseResponseWrapper } from '../utility-types'
3 import { EMPTY_FUNCTION, median } from '../utils'
4 import { KillBehaviors, isKillBehavior } from '../worker/worker-options'
5 import { PoolEvents, type PoolOptions } from './pool'
6 import { PoolEmitter } from './pool'
7 import type { IPoolInternal } from './pool-internal'
8 import { PoolType } from './pool-internal'
9 import type { IWorker, Task, TasksUsage, WorkerNode } from './worker'
10 import {
11 WorkerChoiceStrategies,
12 type WorkerChoiceStrategy
13 } from './selection-strategies/selection-strategies-types'
14 import { WorkerChoiceStrategyContext } from './selection-strategies/worker-choice-strategy-context'
15 import { CircularArray } from '../circular-array'
16
17 /**
18 * Base class that implements some shared logic for all poolifier pools.
19 *
20 * @typeParam Worker - Type of worker which manages this pool.
21 * @typeParam Data - Type of data sent to the worker. This can only be serializable data.
22 * @typeParam Response - Type of response of execution. This can only be serializable data.
23 */
24 export abstract class AbstractPool<
25 Worker extends IWorker,
26 Data = unknown,
27 Response = unknown
28 > implements IPoolInternal<Worker, Data, Response> {
29 /** @inheritDoc */
30 public readonly workerNodes: Array<WorkerNode<Worker, Data>> = []
31
32 /** @inheritDoc */
33 public readonly emitter?: PoolEmitter
34
35 /**
36 * The execution response promise map.
37 *
38 * - `key`: The message id of each submitted task.
39 * - `value`: An object that contains the worker, the execution response promise resolve and reject callbacks.
40 *
41 * When we receive a message from the worker, we get a map entry with the promise resolve/reject bound to the message id.
42 */
43 protected promiseResponseMap: Map<
44 string,
45 PromiseResponseWrapper<Worker, Response>
46 > = new Map<string, PromiseResponseWrapper<Worker, Response>>()
47
48 /**
49 * Worker choice strategy context referencing a worker choice algorithm implementation.
50 *
51 * Default to a round robin algorithm.
52 */
53 protected workerChoiceStrategyContext: WorkerChoiceStrategyContext<
54 Worker,
55 Data,
56 Response
57 >
58
59 /**
60 * Constructs a new poolifier pool.
61 *
62 * @param numberOfWorkers - Number of workers that this pool should manage.
63 * @param filePath - Path to the worker-file.
64 * @param opts - Options for the pool.
65 */
66 public constructor (
67 public readonly numberOfWorkers: number,
68 public readonly filePath: string,
69 public readonly opts: PoolOptions<Worker>
70 ) {
71 if (!this.isMain()) {
72 throw new Error('Cannot start a pool from a worker!')
73 }
74 this.checkNumberOfWorkers(this.numberOfWorkers)
75 this.checkFilePath(this.filePath)
76 this.checkPoolOptions(this.opts)
77
78 this.chooseWorkerNode.bind(this)
79 this.internalExecute.bind(this)
80 this.checkAndEmitEvents.bind(this)
81 this.sendToWorker.bind(this)
82
83 this.setupHook()
84
85 for (let i = 1; i <= this.numberOfWorkers; i++) {
86 this.createAndSetupWorker()
87 }
88
89 if (this.opts.enableEvents === true) {
90 this.emitter = new PoolEmitter()
91 }
92 this.workerChoiceStrategyContext = new WorkerChoiceStrategyContext<
93 Worker,
94 Data,
95 Response
96 >(
97 this,
98 this.opts.workerChoiceStrategy,
99 this.opts.workerChoiceStrategyOptions
100 )
101 }
102
103 private checkFilePath (filePath: string): void {
104 if (
105 filePath == null ||
106 (typeof filePath === 'string' && filePath.trim().length === 0)
107 ) {
108 throw new Error('Please specify a file with a worker implementation')
109 }
110 }
111
112 private checkNumberOfWorkers (numberOfWorkers: number): void {
113 if (numberOfWorkers == null) {
114 throw new Error(
115 'Cannot instantiate a pool without specifying the number of workers'
116 )
117 } else if (!Number.isSafeInteger(numberOfWorkers)) {
118 throw new TypeError(
119 'Cannot instantiate a pool with a non integer number of workers'
120 )
121 } else if (numberOfWorkers < 0) {
122 throw new RangeError(
123 'Cannot instantiate a pool with a negative number of workers'
124 )
125 } else if (this.type === PoolType.FIXED && numberOfWorkers === 0) {
126 throw new Error('Cannot instantiate a fixed pool with no worker')
127 }
128 }
129
130 private checkPoolOptions (opts: PoolOptions<Worker>): void {
131 this.opts.workerChoiceStrategy =
132 opts.workerChoiceStrategy ?? WorkerChoiceStrategies.ROUND_ROBIN
133 this.checkValidWorkerChoiceStrategy(this.opts.workerChoiceStrategy)
134 this.opts.workerChoiceStrategyOptions =
135 opts.workerChoiceStrategyOptions ?? { medRunTime: false }
136 this.opts.enableEvents = opts.enableEvents ?? true
137 this.opts.enableTasksQueue = opts.enableTasksQueue ?? false
138 }
139
140 private checkValidWorkerChoiceStrategy (
141 workerChoiceStrategy: WorkerChoiceStrategy
142 ): void {
143 if (!Object.values(WorkerChoiceStrategies).includes(workerChoiceStrategy)) {
144 throw new Error(
145 `Invalid worker choice strategy '${workerChoiceStrategy}'`
146 )
147 }
148 }
149
150 /** @inheritDoc */
151 public abstract get type (): PoolType
152
153 /**
154 * Number of tasks running in the pool.
155 */
156 private get numberOfRunningTasks (): number {
157 return this.workerNodes.reduce(
158 (accumulator, workerNode) => accumulator + workerNode.tasksUsage.running,
159 0
160 )
161 }
162
163 /**
164 * Number of tasks queued in the pool.
165 */
166 private get numberOfQueuedTasks (): number {
167 if (this.opts.enableTasksQueue === false) {
168 return 0
169 }
170 return this.workerNodes.reduce(
171 (accumulator, workerNode) => accumulator + workerNode.tasksQueue.length,
172 0
173 )
174 }
175
176 /**
177 * Gets the given worker its worker node key.
178 *
179 * @param worker - The worker.
180 * @returns The worker node key if the worker is found in the pool worker nodes, `-1` otherwise.
181 */
182 private getWorkerNodeKey (worker: Worker): number {
183 return this.workerNodes.findIndex(
184 workerNode => workerNode.worker === worker
185 )
186 }
187
188 /** @inheritDoc */
189 public setWorkerChoiceStrategy (
190 workerChoiceStrategy: WorkerChoiceStrategy
191 ): void {
192 this.checkValidWorkerChoiceStrategy(workerChoiceStrategy)
193 this.opts.workerChoiceStrategy = workerChoiceStrategy
194 for (const [index, workerNode] of this.workerNodes.entries()) {
195 this.setWorkerNode(
196 index,
197 workerNode.worker,
198 {
199 run: 0,
200 running: 0,
201 runTime: 0,
202 runTimeHistory: new CircularArray(),
203 avgRunTime: 0,
204 medRunTime: 0,
205 error: 0
206 },
207 workerNode.tasksQueue
208 )
209 }
210 this.workerChoiceStrategyContext.setWorkerChoiceStrategy(
211 workerChoiceStrategy
212 )
213 }
214
215 /** @inheritDoc */
216 public abstract get full (): boolean
217
218 /** @inheritDoc */
219 public abstract get busy (): boolean
220
221 protected internalBusy (): boolean {
222 return (
223 this.numberOfRunningTasks >= this.numberOfWorkers &&
224 this.findFreeWorkerNodeKey() === -1
225 )
226 }
227
228 /** @inheritDoc */
229 public findFreeWorkerNodeKey (): number {
230 return this.workerNodes.findIndex(workerNode => {
231 return workerNode.tasksUsage?.running === 0
232 })
233 }
234
235 /** @inheritDoc */
236 public async execute (data: Data): Promise<Response> {
237 const [workerNodeKey, workerNode] = this.chooseWorkerNode()
238 const submittedTask: Task<Data> = {
239 // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
240 data: data ?? ({} as Data),
241 id: crypto.randomUUID()
242 }
243 const res = this.internalExecute(workerNodeKey, workerNode, submittedTask)
244 let currentTask: Task<Data> = submittedTask
245 if (
246 this.opts.enableTasksQueue === true &&
247 (this.busy || this.tasksQueueSize(workerNodeKey) > 0)
248 ) {
249 currentTask = this.enqueueDequeueTask(
250 workerNodeKey,
251 submittedTask
252 ) as Task<Data>
253 }
254 this.sendToWorker(workerNode.worker, currentTask)
255 this.checkAndEmitEvents()
256 // eslint-disable-next-line @typescript-eslint/return-await
257 return res
258 }
259
260 /** @inheritDoc */
261 public async destroy (): Promise<void> {
262 await Promise.all(
263 this.workerNodes.map(async workerNode => {
264 this.flushTasksQueueByWorker(workerNode.worker)
265 await this.destroyWorker(workerNode.worker)
266 })
267 )
268 }
269
270 /**
271 * Shutdowns the given worker.
272 *
273 * @param worker - A worker within `workerNodes`.
274 */
275 protected abstract destroyWorker (worker: Worker): void | Promise<void>
276
277 /**
278 * Setup hook to run code before worker node are created in the abstract constructor.
279 * Can be overridden
280 *
281 * @virtual
282 */
283 protected setupHook (): void {
284 // Intentionally empty
285 }
286
287 /**
288 * Should return whether the worker is the main worker or not.
289 */
290 protected abstract isMain (): boolean
291
292 /**
293 * Hook executed before the worker task promise resolution.
294 * Can be overridden.
295 *
296 * @param workerNodeKey - The worker node key.
297 */
298 protected beforePromiseResponseHook (workerNodeKey: number): void {
299 ++this.workerNodes[workerNodeKey].tasksUsage.running
300 }
301
302 /**
303 * Hook executed after the worker task promise resolution.
304 * Can be overridden.
305 *
306 * @param worker - The worker.
307 * @param message - The received message.
308 */
309 protected afterPromiseResponseHook (
310 worker: Worker,
311 message: MessageValue<Response>
312 ): void {
313 const workerTasksUsage = this.getWorkerTasksUsage(worker) as TasksUsage
314 --workerTasksUsage.running
315 ++workerTasksUsage.run
316 if (message.error != null) {
317 ++workerTasksUsage.error
318 }
319 if (this.workerChoiceStrategyContext.getRequiredStatistics().runTime) {
320 workerTasksUsage.runTime += message.runTime ?? 0
321 if (
322 this.workerChoiceStrategyContext.getRequiredStatistics().avgRunTime &&
323 workerTasksUsage.run !== 0
324 ) {
325 workerTasksUsage.avgRunTime =
326 workerTasksUsage.runTime / workerTasksUsage.run
327 }
328 if (this.workerChoiceStrategyContext.getRequiredStatistics().medRunTime) {
329 workerTasksUsage.runTimeHistory.push(message.runTime ?? 0)
330 workerTasksUsage.medRunTime = median(workerTasksUsage.runTimeHistory)
331 }
332 }
333 }
334
335 /**
336 * Chooses a worker node for the next task.
337 *
338 * The default uses a round robin algorithm to distribute the load.
339 *
340 * @returns [worker node key, worker node].
341 */
342 protected chooseWorkerNode (): [number, WorkerNode<Worker, Data>] {
343 let workerNodeKey: number
344 if (
345 this.type === PoolType.DYNAMIC &&
346 !this.full &&
347 this.findFreeWorkerNodeKey() === -1
348 ) {
349 const workerCreated = this.createAndSetupWorker()
350 this.registerWorkerMessageListener(workerCreated, message => {
351 if (
352 isKillBehavior(KillBehaviors.HARD, message.kill) ||
353 (message.kill != null &&
354 this.getWorkerTasksUsage(workerCreated)?.running === 0)
355 ) {
356 // Kill message received from the worker: no new tasks are submitted to that worker for a while ( > maxInactiveTime)
357 this.flushTasksQueueByWorker(workerCreated)
358 void this.destroyWorker(workerCreated)
359 }
360 })
361 workerNodeKey = this.getWorkerNodeKey(workerCreated)
362 } else {
363 workerNodeKey = this.workerChoiceStrategyContext.execute()
364 }
365 return [workerNodeKey, this.workerNodes[workerNodeKey]]
366 }
367
368 /**
369 * Sends a message to the given worker.
370 *
371 * @param worker - The worker which should receive the message.
372 * @param message - The message.
373 */
374 protected abstract sendToWorker (
375 worker: Worker,
376 message: MessageValue<Data>
377 ): void
378
379 /**
380 * Registers a listener callback on the given worker.
381 *
382 * @param worker - The worker which should register a listener.
383 * @param listener - The message listener callback.
384 */
385 protected abstract registerWorkerMessageListener<
386 Message extends Data | Response
387 >(worker: Worker, listener: (message: MessageValue<Message>) => void): void
388
389 /**
390 * Returns a newly created worker.
391 */
392 protected abstract createWorker (): Worker
393
394 /**
395 * Function that can be hooked up when a worker has been newly created and moved to the pool worker nodes.
396 *
397 * Can be used to update the `maxListeners` or binding the `main-worker`\<-\>`worker` connection if not bind by default.
398 *
399 * @param worker - The newly created worker.
400 */
401 protected abstract afterWorkerSetup (worker: Worker): void
402
403 /**
404 * Creates a new worker and sets it up completely in the pool worker nodes.
405 *
406 * @returns New, completely set up worker.
407 */
408 protected createAndSetupWorker (): Worker {
409 const worker = this.createWorker()
410
411 worker.on('message', this.opts.messageHandler ?? EMPTY_FUNCTION)
412 worker.on('error', this.opts.errorHandler ?? EMPTY_FUNCTION)
413 worker.on('online', this.opts.onlineHandler ?? EMPTY_FUNCTION)
414 worker.on('exit', this.opts.exitHandler ?? EMPTY_FUNCTION)
415 worker.once('exit', () => {
416 this.removeWorkerNode(worker)
417 })
418
419 this.pushWorkerNode(worker)
420
421 this.afterWorkerSetup(worker)
422
423 return worker
424 }
425
426 /**
427 * This function is the listener registered for each worker message.
428 *
429 * @returns The listener function to execute when a message is received from a worker.
430 */
431 protected workerListener (): (message: MessageValue<Response>) => void {
432 return message => {
433 if (message.id != null) {
434 // Task execution response received
435 const promiseResponse = this.promiseResponseMap.get(message.id)
436 if (promiseResponse != null) {
437 if (message.error != null) {
438 promiseResponse.reject(message.error)
439 } else {
440 promiseResponse.resolve(message.data as Response)
441 }
442 this.afterPromiseResponseHook(promiseResponse.worker, message)
443 this.promiseResponseMap.delete(message.id)
444 const workerNodeKey = this.getWorkerNodeKey(promiseResponse.worker)
445 if (
446 this.opts.enableTasksQueue === true &&
447 this.tasksQueueSize(workerNodeKey) > 0
448 ) {
449 this.sendToWorker(
450 promiseResponse.worker,
451 this.dequeueTask(workerNodeKey) as Task<Data>
452 )
453 }
454 }
455 }
456 }
457 }
458
459 private async internalExecute (
460 workerNodeKey: number,
461 workerNode: WorkerNode<Worker, Data>,
462 task: Task<Data>
463 ): Promise<Response> {
464 this.beforePromiseResponseHook(workerNodeKey)
465 return await new Promise<Response>((resolve, reject) => {
466 this.promiseResponseMap.set(task.id, {
467 resolve,
468 reject,
469 worker: workerNode.worker
470 })
471 })
472 }
473
474 private checkAndEmitEvents (): void {
475 if (this.opts.enableEvents === true) {
476 if (this.busy) {
477 this.emitter?.emit(PoolEvents.busy)
478 }
479 if (this.type === PoolType.DYNAMIC && this.full) {
480 this.emitter?.emit(PoolEvents.full)
481 }
482 }
483 }
484
485 /**
486 * Gets the given worker its tasks usage in the pool.
487 *
488 * @param worker - The worker.
489 * @returns The worker tasks usage.
490 */
491 private getWorkerTasksUsage (worker: Worker): TasksUsage | undefined {
492 const workerNodeKey = this.getWorkerNodeKey(worker)
493 if (workerNodeKey !== -1) {
494 return this.workerNodes[workerNodeKey].tasksUsage
495 }
496 throw new Error('Worker could not be found in the pool worker nodes')
497 }
498
499 /**
500 * Pushes the given worker in the pool worker nodes.
501 *
502 * @param worker - The worker.
503 * @returns The worker nodes length.
504 */
505 private pushWorkerNode (worker: Worker): number {
506 return this.workerNodes.push({
507 worker,
508 tasksUsage: {
509 run: 0,
510 running: 0,
511 runTime: 0,
512 runTimeHistory: new CircularArray(),
513 avgRunTime: 0,
514 medRunTime: 0,
515 error: 0
516 },
517 tasksQueue: []
518 })
519 }
520
521 /**
522 * Sets the given worker in the pool worker nodes.
523 *
524 * @param workerNodeKey - The worker node key.
525 * @param worker - The worker.
526 * @param tasksUsage - The worker tasks usage.
527 * @param tasksQueue - The worker task queue.
528 */
529 private setWorkerNode (
530 workerNodeKey: number,
531 worker: Worker,
532 tasksUsage: TasksUsage,
533 tasksQueue: Array<Task<Data>>
534 ): void {
535 this.workerNodes[workerNodeKey] = {
536 worker,
537 tasksUsage,
538 tasksQueue
539 }
540 }
541
542 /**
543 * Removes the given worker from the pool worker nodes.
544 *
545 * @param worker - The worker.
546 */
547 private removeWorkerNode (worker: Worker): void {
548 const workerNodeKey = this.getWorkerNodeKey(worker)
549 this.workerNodes.splice(workerNodeKey, 1)
550 this.workerChoiceStrategyContext.remove(workerNodeKey)
551 }
552
553 private enqueueDequeueTask (
554 workerNodeKey: number,
555 task: Task<Data>
556 ): Task<Data> | undefined {
557 this.enqueueTask(workerNodeKey, task)
558 return this.dequeueTask(workerNodeKey)
559 }
560
561 private enqueueTask (workerNodeKey: number, task: Task<Data>): void {
562 this.workerNodes[workerNodeKey].tasksQueue.push(task)
563 }
564
565 private dequeueTask (workerNodeKey: number): Task<Data> | undefined {
566 return this.workerNodes[workerNodeKey].tasksQueue.shift()
567 }
568
569 private tasksQueueSize (workerNodeKey: number): number {
570 return this.workerNodes[workerNodeKey].tasksQueue.length
571 }
572
573 private flushTasksQueue (workerNodeKey: number): void {
574 if (this.tasksQueueSize(workerNodeKey) > 0) {
575 for (const task of this.workerNodes[workerNodeKey].tasksQueue) {
576 this.sendToWorker(this.workerNodes[workerNodeKey].worker, task)
577 }
578 this.workerNodes[workerNodeKey].tasksQueue = []
579 }
580 }
581
582 private flushTasksQueueByWorker (worker: Worker): void {
583 const workerNodeKey = this.getWorkerNodeKey(worker)
584 this.flushTasksQueue(workerNodeKey)
585 }
586 }