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 }