Add repo vscode extensions recommendation (#99)
[poolifier.git] / src / dynamic.ts
1 import { EventEmitter } from 'events'
2 import {
3 FixedThreadPool,
4 FixedThreadPoolOptions,
5 WorkerWithMessageChannel
6 } from './fixed'
7
8 class MyEmitter extends EventEmitter {}
9
10 export type DynamicThreadPoolOptions = FixedThreadPoolOptions
11
12 /**
13 * A thread pool with a min/max number of threads, is possible to execute tasks in sync or async mode as you prefer.
14 *
15 * This thread pool will create new workers when the other ones are busy, until the max number of threads,
16 * when the max number of threads is reached, an event will be emitted, if you want to listen this event use the emitter method.
17 *
18 * @author [Alessandro Pio Ardizio](https://github.com/pioardi)
19 * @since 0.0.1
20 */
21 /* eslint-disable @typescript-eslint/no-explicit-any */
22 export class DynamicThreadPool<
23 Data = any,
24 Response = any
25 > extends FixedThreadPool<Data, Response> {
26 /* eslint-enable @typescript-eslint/no-explicit-any */
27 public readonly emitter: MyEmitter
28
29 /**
30 * @param min Min number of threads that will be always active
31 * @param max Max number of threads that will be active
32 * @param filename A file path with implementation of `ThreadWorker` class, relative path is fine.
33 * @param opts An object with possible options for example `errorHandler`, `onlineHandler`. Default: `{ maxTasks: 1000 }`
34 */
35 public constructor (
36 public readonly min: number,
37 public readonly max: number,
38 public readonly filename: string,
39 public readonly opts: DynamicThreadPoolOptions = { maxTasks: 1000 }
40 ) {
41 super(min, filename, opts)
42
43 this.emitter = new MyEmitter()
44 }
45
46 protected chooseWorker (): WorkerWithMessageChannel {
47 let worker: WorkerWithMessageChannel | undefined
48 for (const entry of this.tasks) {
49 if (entry[1] === 0) {
50 worker = entry[0]
51 break
52 }
53 }
54
55 if (worker) {
56 // a worker is free, use it
57 return worker
58 } else {
59 if (this.workers.length === this.max) {
60 this.emitter.emit('FullPool')
61 return super.chooseWorker()
62 }
63 // all workers are busy create a new worker
64 const worker = this.newWorker()
65 worker.port2?.on('message', (message: { kill?: number }) => {
66 if (message.kill) {
67 worker.postMessage({ kill: 1 })
68 void worker.terminate()
69 // clean workers from data structures
70 const workerIndex = this.workers.indexOf(worker)
71 this.workers.splice(workerIndex, 1)
72 this.tasks.delete(worker)
73 }
74 })
75 return worker
76 }
77 }
78 }