1 /* eslint-disable @typescript-eslint/strict-boolean-expressions */
3 import { isMainThread
, parentPort
} from
'worker_threads'
5 import { AsyncResource
} from
'async_hooks'
7 export interface ThreadWorkerOptions
{
9 * Max time to wait tasks to work on (in ms), after this period the new worker threads will die.
13 maxInactiveTime
?: number
15 * `true` if your function contains async pieces, else `false`.
23 * An example worker that will be always alive, you just need to **extend** this class if you want a static pool.
25 * When this worker is inactive for more than 1 minute, it will send this info to the main thread,
26 * if you are using DynamicThreadPool, the workers created after will be killed, the min num of thread will be guaranteed.
28 * @author [Alessandro Pio Ardizio](https://github.com/pioardi)
31 export class ThreadWorker
<Data
= any, Response
= any> extends AsyncResource
{
32 protected readonly maxInactiveTime
: number
33 protected readonly async: boolean
34 protected lastTask
: number
35 protected readonly interval
?: NodeJS
.Timeout
36 protected parent?: MessagePort
39 fn
: (data
: Data
) => Response
,
40 public readonly opts
: ThreadWorkerOptions
= {}
42 super('worker-thread-pool:pioardi')
44 this.maxInactiveTime
= this.opts
.maxInactiveTime
?? 1000 * 60
45 this.async = !!this.opts
.async
46 this.lastTask
= Date.now()
47 if (!fn
) throw new Error('Fn parameter is mandatory')
48 // keep the worker active
50 this.interval
= setInterval(
51 this._checkAlive
.bind(this),
52 this.maxInactiveTime
/ 2
54 this._checkAlive
.bind(this)()
64 if (value
?.data
&& value
._id
) {
65 // here you will receive messages
66 // console.log('This is the main thread ' + isMainThread)
68 this.runInAsyncScope(this._runAsync
.bind(this), this, fn
, value
)
70 this.runInAsyncScope(this._run
.bind(this), this, fn
, value
)
72 } else if (value
.parent) {
73 // save the port to communicate with the main thread
74 // this will be received once
75 this.parent = value
.parent
76 } else if (value
.kill
) {
77 // here is time to kill this thread, just clearing the interval
78 if (this.interval
) clearInterval(this.interval
)
85 protected _checkAlive (): void {
86 if (Date.now() - this.lastTask
> this.maxInactiveTime
) {
87 this.parent?.postMessage({ kill
: 1 })
92 fn
: (data
: Data
) => Response
,
93 value
: { readonly data
: Data
, readonly _id
: number }
96 const res
= fn(value
.data
)
97 this.parent?.postMessage({ data
: res
, _id
: value
._id
})
98 this.lastTask
= Date.now()
100 this.parent?.postMessage({ error
: e
, _id
: value
._id
})
101 this.lastTask
= Date.now()
105 protected _runAsync (
106 fn
: (data
: Data
) => Promise
<Response
>,
107 value
: { readonly data
: Data
, readonly _id
: number }
111 this.parent?.postMessage({ data
: res
, _id
: value
._id
})
112 this.lastTask
= Date.now()
115 this.parent?.postMessage({ error
: e
, _id
: value
._id
})
116 this.lastTask
= Date.now()