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