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