src/workers.ts

   1 import { isMainThread, parentPort } from 'worker_threads'
   2 import { AsyncResource } from 'async_hooks'
   3
   4 export interface ThreadWorkerOptions {
   5   /**
   6    * Max time to wait tasks to work on (in ms), after this period the new worker threads will die.
   7    *
   8    * @default 60.000 ms
   9    */
  10   maxInactiveTime?: number
  11   /**
  12    * `true` if your function contains async pieces, else `false`.
  13    *
  14    * @default false
  15    */
  16   async?: boolean
  17 }
  18
  19 /**
  20  * An example worker that will be always alive, you just need to **extend** this class if you want a static pool.
  21  *
  22  * When this worker is inactive for more than 1 minute, it will send this info to the main thread,
  23  * if you are using DynamicThreadPool, the workers created after will be killed, the min num of thread will be guaranteed.
  24  *
  25  * @author [Alessandro Pio Ardizio](https://github.com/pioardi)
  26  * @since 0.0.1
  27  */
  28 export class ThreadWorker<Data = any, Response = any> extends AsyncResource {
  29   protected readonly maxInactiveTime: number
  30   protected readonly async: boolean
  31   protected lastTask: number
  32   protected readonly interval: NodeJS.Timeout
  33   protected parent: any
  34
  35   public constructor (
  36     fn: (data: Data) => Response,
  37     public readonly opts: ThreadWorkerOptions = {}
  38   ) {
  39     super('worker-thread-pool:pioardi')
  40
  41     this.maxInactiveTime = this.opts.maxInactiveTime || 1000 * 60
  42     this.async = !!this.opts.async
  43     this.lastTask = Date.now()
  44     if (!fn) throw new Error('Fn parameter is mandatory')
  45     // keep the worker active
  46     if (!isMainThread) {
  47       this.interval = setInterval(
  48         this._checkAlive.bind(this),
  49         this.maxInactiveTime / 2
  50       )
  51       this._checkAlive.bind(this)()
  52     }
  53     parentPort.on('message', (value) => {
  54       if (value && value.data && value._id) {
  55         // here you will receive messages
  56         // console.log('This is the main thread ' + isMainThread)
  57         if (this.async) {
  58           this.runInAsyncScope(this._runAsync.bind(this), this, fn, value)
  59         } else {
  60           this.runInAsyncScope(this._run.bind(this), this, fn, value)
  61         }
  62       } else if (value.parent) {
  63         // save the port to communicate with the main thread
  64         // this will be received once
  65         this.parent = value.parent
  66       } else if (value.kill) {
  67         // here is time to kill this thread, just clearing the interval
  68         clearInterval(this.interval)
  69         this.emitDestroy()
  70       }
  71     })
  72   }
  73
  74   protected _checkAlive (): void {
  75     if (Date.now() - this.lastTask > this.maxInactiveTime) {
  76       this.parent.postMessage({ kill: 1 })
  77     }
  78   }
  79
  80   protected _run (
  81     fn: (data: Data) => Response,
  82     value: { readonly data: Data, readonly _id: number }
  83   ): void {
  84     try {
  85       const res = fn(value.data)
  86       this.parent.postMessage({ data: res, _id: value._id })
  87       this.lastTask = Date.now()
  88     } catch (e) {
  89       this.parent.postMessage({ error: e, _id: value._id })
  90       this.lastTask = Date.now()
  91     }
  92   }
  93
  94   protected _runAsync (
  95     fn: (data: Data) => Promise<Response>,
  96     value: { readonly data: Data, readonly _id: number }
  97   ): void {
  98     fn(value.data)
  99       .then((res) => {
 100         this.parent.postMessage({ data: res, _id: value._id })
 101         this.lastTask = Date.now()
 102       })
 103       .catch((e) => {
 104         this.parent.postMessage({ error: e, _id: value._id })
 105         this.lastTask = Date.now()
 106       })
 107   }
 108 }