Merge pull request #758 from poolifier/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 workerTasksUsage =
434 this.workerNodes[this.getWorkerNodeKey(worker)].tasksUsage
435 --workerTasksUsage.running
436 ++workerTasksUsage.run
437 if (message.error != null) {
438 ++workerTasksUsage.error
439 }
440 this.updateRunTimeTasksUsage(workerTasksUsage, message)
441 this.updateWaitTasksUsage(workerTasksUsage, message)
442 }
443
444 private updateRunTimeTasksUsage (
445 workerTasksUsage: TasksUsage,
446 message: MessageValue<Response>
447 ): void {
448 if (this.workerChoiceStrategyContext.getRequiredStatistics().runTime) {
449 workerTasksUsage.runTime += message.runTime ?? 0
450 if (
451 this.workerChoiceStrategyContext.getRequiredStatistics().avgRunTime &&
452 workerTasksUsage.run !== 0
453 ) {
454 workerTasksUsage.avgRunTime =
455 workerTasksUsage.runTime / workerTasksUsage.run
456 }
457 if (
458 this.workerChoiceStrategyContext.getRequiredStatistics().medRunTime &&
459 message.runTime != null
460 ) {
461 workerTasksUsage.runTimeHistory.push(message.runTime)
462 workerTasksUsage.medRunTime = median(workerTasksUsage.runTimeHistory)
463 }
464 }
465 }
466
467 private updateWaitTasksUsage (
468 workerTasksUsage: TasksUsage,
469 message: MessageValue<Response>
470 ): void {
471 if (this.workerChoiceStrategyContext.getRequiredStatistics().waitTime) {
472 workerTasksUsage.waitTime += message.waitTime ?? 0
473 if (
474 this.workerChoiceStrategyContext.getRequiredStatistics().avgWaitTime &&
475 workerTasksUsage.run !== 0
476 ) {
477 workerTasksUsage.avgWaitTime =
478 workerTasksUsage.waitTime / workerTasksUsage.run
479 }
480 if (
481 this.workerChoiceStrategyContext.getRequiredStatistics().medWaitTime &&
482 message.waitTime != null
483 ) {
484 workerTasksUsage.waitTimeHistory.push(message.waitTime)
485 workerTasksUsage.medWaitTime = median(workerTasksUsage.waitTimeHistory)
486 }
487 }
488 }
489
490 /**
491 * Chooses a worker node for the next task.
492 *
493 * The default worker choice strategy uses a round robin algorithm to distribute the load.
494 *
495 * @returns The worker node key
496 */
497 protected chooseWorkerNode (): number {
498 let workerNodeKey: number
499 if (this.type === PoolType.DYNAMIC && !this.full && this.internalBusy()) {
500 const workerCreated = this.createAndSetupWorker()
501 this.registerWorkerMessageListener(workerCreated, message => {
502 const currentWorkerNodeKey = this.getWorkerNodeKey(workerCreated)
503 if (
504 isKillBehavior(KillBehaviors.HARD, message.kill) ||
505 (message.kill != null &&
506 this.workerNodes[currentWorkerNodeKey].tasksUsage.running === 0)
507 ) {
508 // Kill message received from the worker: no new tasks are submitted to that worker for a while ( > maxInactiveTime)
509 this.flushTasksQueue(currentWorkerNodeKey)
510 void (this.destroyWorker(workerCreated) as Promise<void>)
511 }
512 })
513 workerNodeKey = this.getWorkerNodeKey(workerCreated)
514 } else {
515 workerNodeKey = this.workerChoiceStrategyContext.execute()
516 }
517 return workerNodeKey
518 }
519
520 /**
521 * Sends a message to the given worker.
522 *
523 * @param worker - The worker which should receive the message.
524 * @param message - The message.
525 */
526 protected abstract sendToWorker (
527 worker: Worker,
528 message: MessageValue<Data>
529 ): void
530
531 /**
532 * Registers a listener callback on the given worker.
533 *
534 * @param worker - The worker which should register a listener.
535 * @param listener - The message listener callback.
536 */
537 protected abstract registerWorkerMessageListener<
538 Message extends Data | Response
539 >(worker: Worker, listener: (message: MessageValue<Message>) => void): void
540
541 /**
542 * Returns a newly created worker.
543 */
544 protected abstract createWorker (): Worker
545
546 /**
547 * Function that can be hooked up when a worker has been newly created and moved to the pool worker nodes.
548 *
549 * Can be used to update the `maxListeners` or binding the `main-worker`\<-\>`worker` connection if not bind by default.
550 *
551 * @param worker - The newly created worker.
552 */
553 protected abstract afterWorkerSetup (worker: Worker): void
554
555 /**
556 * Creates a new worker and sets it up completely in the pool worker nodes.
557 *
558 * @returns New, completely set up worker.
559 */
560 protected createAndSetupWorker (): Worker {
561 const worker = this.createWorker()
562
563 worker.on('message', this.opts.messageHandler ?? EMPTY_FUNCTION)
564 worker.on('error', this.opts.errorHandler ?? EMPTY_FUNCTION)
565 worker.on('online', this.opts.onlineHandler ?? EMPTY_FUNCTION)
566 worker.on('exit', this.opts.exitHandler ?? EMPTY_FUNCTION)
567 worker.once('exit', () => {
568 this.removeWorkerNode(worker)
569 })
570
571 this.pushWorkerNode(worker)
572
573 this.afterWorkerSetup(worker)
574
575 return worker
576 }
577
578 /**
579 * This function is the listener registered for each worker message.
580 *
581 * @returns The listener function to execute when a message is received from a worker.
582 */
583 protected workerListener (): (message: MessageValue<Response>) => void {
584 return message => {
585 if (message.id != null) {
586 // Task execution response received
587 const promiseResponse = this.promiseResponseMap.get(message.id)
588 if (promiseResponse != null) {
589 if (message.error != null) {
590 promiseResponse.reject(message.error)
591 } else {
592 promiseResponse.resolve(message.data as Response)
593 }
594 this.afterTaskExecutionHook(promiseResponse.worker, message)
595 this.promiseResponseMap.delete(message.id)
596 const workerNodeKey = this.getWorkerNodeKey(promiseResponse.worker)
597 if (
598 this.opts.enableTasksQueue === true &&
599 this.tasksQueueSize(workerNodeKey) > 0
600 ) {
601 this.executeTask(
602 workerNodeKey,
603 this.dequeueTask(workerNodeKey) as Task<Data>
604 )
605 }
606 }
607 }
608 }
609 }
610
611 private checkAndEmitEvents (): void {
612 if (this.opts.enableEvents === true) {
613 if (this.busy) {
614 this.emitter?.emit(PoolEvents.busy)
615 }
616 if (this.type === PoolType.DYNAMIC && this.full) {
617 this.emitter?.emit(PoolEvents.full)
618 }
619 }
620 }
621
622 /**
623 * Sets the given worker node its tasks usage in the pool.
624 *
625 * @param workerNode - The worker node.
626 * @param tasksUsage - The worker node tasks usage.
627 */
628 private setWorkerNodeTasksUsage (
629 workerNode: WorkerNode<Worker, Data>,
630 tasksUsage: TasksUsage
631 ): void {
632 workerNode.tasksUsage = tasksUsage
633 }
634
635 /**
636 * Pushes the given worker in the pool worker nodes.
637 *
638 * @param worker - The worker.
639 * @returns The worker nodes length.
640 */
641 private pushWorkerNode (worker: Worker): number {
642 return this.workerNodes.push({
643 worker,
644 tasksUsage: {
645 run: 0,
646 running: 0,
647 runTime: 0,
648 runTimeHistory: new CircularArray(),
649 avgRunTime: 0,
650 medRunTime: 0,
651 waitTime: 0,
652 waitTimeHistory: new CircularArray(),
653 avgWaitTime: 0,
654 medWaitTime: 0,
655 error: 0
656 },
657 tasksQueue: new Queue<Task<Data>>()
658 })
659 }
660
661 /**
662 * Sets the given worker in the pool worker nodes.
663 *
664 * @param workerNodeKey - The worker node key.
665 * @param worker - The worker.
666 * @param tasksUsage - The worker tasks usage.
667 * @param tasksQueue - The worker task queue.
668 */
669 private setWorkerNode (
670 workerNodeKey: number,
671 worker: Worker,
672 tasksUsage: TasksUsage,
673 tasksQueue: Queue<Task<Data>>
674 ): void {
675 this.workerNodes[workerNodeKey] = {
676 worker,
677 tasksUsage,
678 tasksQueue
679 }
680 }
681
682 /**
683 * Removes the given worker from the pool worker nodes.
684 *
685 * @param worker - The worker.
686 */
687 private removeWorkerNode (worker: Worker): void {
688 const workerNodeKey = this.getWorkerNodeKey(worker)
689 this.workerNodes.splice(workerNodeKey, 1)
690 this.workerChoiceStrategyContext.remove(workerNodeKey)
691 }
692
693 private executeTask (workerNodeKey: number, task: Task<Data>): void {
694 this.beforeTaskExecutionHook(workerNodeKey)
695 this.sendToWorker(this.workerNodes[workerNodeKey].worker, task)
696 }
697
698 private enqueueTask (workerNodeKey: number, task: Task<Data>): number {
699 return this.workerNodes[workerNodeKey].tasksQueue.enqueue(task)
700 }
701
702 private dequeueTask (workerNodeKey: number): Task<Data> | undefined {
703 return this.workerNodes[workerNodeKey].tasksQueue.dequeue()
704 }
705
706 private tasksQueueSize (workerNodeKey: number): number {
707 return this.workerNodes[workerNodeKey].tasksQueue.size
708 }
709
710 private flushTasksQueue (workerNodeKey: number): void {
711 if (this.tasksQueueSize(workerNodeKey) > 0) {
712 for (let i = 0; i < this.tasksQueueSize(workerNodeKey); i++) {
713 this.executeTask(
714 workerNodeKey,
715 this.dequeueTask(workerNodeKey) as Task<Data>
716 )
717 }
718 }
719 }
720
721 private flushTasksQueues (): void {
722 for (const [workerNodeKey] of this.workerNodes.entries()) {
723 this.flushTasksQueue(workerNodeKey)
724 }
725 }
726 }