1 import { AsyncResource
} from
'node:async_hooks'
2 import type { Worker
} from
'node:cluster'
3 import type { MessagePort
} from
'node:worker_threads'
4 import { type EventLoopUtilization
, performance
} from
'node:perf_hooks'
5 import type { MessageValue
, WorkerStatistics
} from
'../utility-types'
6 import { EMPTY_FUNCTION
, isPlainObject
} from
'../utils'
11 } from
'./worker-options'
17 } from
'./worker-functions'
19 const DEFAULT_FUNCTION_NAME
= 'default'
20 const DEFAULT_MAX_INACTIVE_TIME
= 60000
21 const DEFAULT_KILL_BEHAVIOR
: KillBehavior
= KillBehaviors
.SOFT
26 interface TaskPerformance
{
30 elu
?: EventLoopUtilization
34 * Base class that implements some shared logic for all poolifier workers.
36 * @typeParam MainWorker - Type of main worker.
37 * @typeParam Data - Type of data this worker receives from pool's execution. This can only be serializable data.
38 * @typeParam Response - Type of response the worker sends back to the main worker. This can only be serializable data.
40 export abstract class AbstractWorker
<
41 MainWorker
extends Worker
| MessagePort
,
44 > extends AsyncResource
{
46 * Task function(s) processed by the worker when the pool's `execution` function is invoked.
48 protected taskFunctions
!: Map
<string, WorkerFunction
<Data
, Response
>>
50 * Timestamp of the last task processed by this worker.
52 protected lastTaskTimestamp
!: number
54 * Performance statistics computation.
56 protected statistics
!: WorkerStatistics
58 * Handler id of the `aliveInterval` worker alive check.
60 protected readonly aliveInterval
?: NodeJS
.Timeout
62 * Constructs a new poolifier worker.
64 * @param type - The type of async event.
65 * @param isMain - Whether this is the main worker or not.
66 * @param taskFunctions - Task function(s) processed by the worker when the pool's `execution` function is invoked. The first function is the default function.
67 * @param mainWorker - Reference to main worker.
68 * @param opts - Options for the worker.
72 protected readonly isMain
: boolean,
74 | WorkerFunction
<Data
, Response
>
75 | TaskFunctions
<Data
, Response
>,
76 protected mainWorker
: MainWorker
| undefined | null,
77 protected readonly opts
: WorkerOptions
= {
79 * The kill behavior option on this worker or its default value.
81 killBehavior
: DEFAULT_KILL_BEHAVIOR
,
83 * The maximum time to keep this worker alive while idle.
84 * The pool automatically checks and terminates this worker when the time expires.
86 maxInactiveTime
: DEFAULT_MAX_INACTIVE_TIME
90 this.checkWorkerOptions(this.opts
)
91 this.checkTaskFunctions(taskFunctions
)
93 this.lastTaskTimestamp
= performance
.now()
94 this.aliveInterval
= setInterval(
95 this.checkAlive
.bind(this),
96 (this.opts
.maxInactiveTime
?? DEFAULT_MAX_INACTIVE_TIME
) / 2
98 this.checkAlive
.bind(this)()
100 this.mainWorker
?.on('message', this.messageListener
.bind(this))
103 private checkWorkerOptions (opts
: WorkerOptions
): void {
104 this.opts
.killBehavior
= opts
.killBehavior
?? DEFAULT_KILL_BEHAVIOR
105 this.opts
.maxInactiveTime
=
106 opts
.maxInactiveTime
?? DEFAULT_MAX_INACTIVE_TIME
107 delete this.opts
.async
111 * Checks if the `taskFunctions` parameter is passed to the constructor.
113 * @param taskFunctions - The task function(s) parameter that should be checked.
115 private checkTaskFunctions (
117 | WorkerFunction
<Data
, Response
>
118 | TaskFunctions
<Data
, Response
>
120 if (taskFunctions
== null) {
121 throw new Error('taskFunctions parameter is mandatory')
123 this.taskFunctions
= new Map
<string, WorkerFunction
<Data
, Response
>>()
124 if (typeof taskFunctions
=== 'function') {
125 this.taskFunctions
.set(DEFAULT_FUNCTION_NAME
, taskFunctions
.bind(this))
126 } else if (isPlainObject(taskFunctions
)) {
127 let firstEntry
= true
128 for (const [name
, fn
] of Object.entries(taskFunctions
)) {
129 if (typeof fn
!== 'function') {
131 'A taskFunctions parameter object value is not a function'
134 this.taskFunctions
.set(name
, fn
.bind(this))
136 this.taskFunctions
.set(DEFAULT_FUNCTION_NAME
, fn
.bind(this))
141 throw new Error('taskFunctions parameter object is empty')
145 'taskFunctions parameter is not a function or a plain object'
151 * Worker message listener.
153 * @param message - Message received.
155 protected messageListener (message
: MessageValue
<Data
, MainWorker
>): void {
156 if (message
.id
!= null && message
.data
!= null) {
157 // Task message received
158 const fn
= this.getTaskFunction(message
.name
)
159 if (fn
?.constructor
.name
=== 'AsyncFunction') {
160 this.runInAsyncScope(this.runAsync
.bind(this), this, fn
, message
)
162 this.runInAsyncScope(this.runSync
.bind(this), this, fn
, message
)
164 } else if (message
.parent != null) {
165 // Main worker reference message received
166 this.mainWorker
= message
.parent
167 } else if (message
.kill
!= null) {
168 // Kill message received
169 this.aliveInterval
!= null && clearInterval(this.aliveInterval
)
171 } else if (message
.statistics
!= null) {
172 // Statistics message received
173 this.statistics
= message
.statistics
178 * Returns the main worker.
180 * @returns Reference to the main worker.
182 protected getMainWorker (): MainWorker
{
183 if (this.mainWorker
== null) {
184 throw new Error('Main worker was not set')
186 return this.mainWorker
190 * Sends a message to the main worker.
192 * @param message - The response message.
194 protected abstract sendToMainWorker (message
: MessageValue
<Response
>): void
197 * Checks if the worker should be terminated, because its living too long.
199 protected checkAlive (): void {
201 performance
.now() - this.lastTaskTimestamp
>
202 (this.opts
.maxInactiveTime
?? DEFAULT_MAX_INACTIVE_TIME
)
204 this.sendToMainWorker({ kill
: this.opts
.killBehavior
})
209 * Handles an error and convert it to a string so it can be sent back to the main worker.
211 * @param e - The error raised by the worker.
212 * @returns Message of the error.
214 protected handleError (e
: Error | string): string {
219 * Runs the given function synchronously.
221 * @param fn - Function that will be executed.
222 * @param message - Input data for the given function.
225 fn
: WorkerSyncFunction
<Data
, Response
>,
226 message
: MessageValue
<Data
>
229 const taskPerformance
= this.beginTaskPerformance(message
)
230 const res
= fn(message
.data
)
231 const { runTime
, waitTime
, elu
} =
232 this.endTaskPerformance(taskPerformance
)
233 this.sendToMainWorker({
241 const err
= this.handleError(e
as Error)
242 this.sendToMainWorker({
244 errorData
: message
.data
,
248 !this.isMain
&& (this.lastTaskTimestamp
= performance
.now())
253 * Runs the given function asynchronously.
255 * @param fn - Function that will be executed.
256 * @param message - Input data for the given function.
259 fn
: WorkerAsyncFunction
<Data
, Response
>,
260 message
: MessageValue
<Data
>
262 const taskPerformance
= this.beginTaskPerformance(message
)
265 const { runTime
, waitTime
, elu
} =
266 this.endTaskPerformance(taskPerformance
)
267 this.sendToMainWorker({
277 const err
= this.handleError(e
as Error)
278 this.sendToMainWorker({
280 errorData
: message
.data
,
285 !this.isMain
&& (this.lastTaskTimestamp
= performance
.now())
287 .catch(EMPTY_FUNCTION
)
291 * Gets the task function in the given scope.
293 * @param name - Name of the function that will be returned.
295 private getTaskFunction (name
?: string): WorkerFunction
<Data
, Response
> {
296 name
= name
?? DEFAULT_FUNCTION_NAME
297 const fn
= this.taskFunctions
.get(name
)
299 throw new Error(`Task function '${name}' not found`)
304 private beginTaskPerformance (message
: MessageValue
<Data
>): TaskPerformance
{
305 const timestamp
= performance
.now()
308 ...(this.statistics
.waitTime
&& {
309 waitTime
: timestamp
- (message
.timestamp
?? timestamp
)
311 ...(this.statistics
.elu
&& { elu
: performance
.eventLoopUtilization() })
315 private endTaskPerformance (
316 taskPerformance
: TaskPerformance
320 ...(this.statistics
.runTime
&& {
321 runTime
: performance
.now() - taskPerformance
.timestamp
323 ...(this.statistics
.elu
&& {
324 elu
: performance
.eventLoopUtilization(taskPerformance
.elu
)