1 import { AsyncResource
} from
'async_hooks'
2 import type { Worker
} from
'cluster'
3 import type { MessagePort
} from
'worker_threads'
4 import type { MessageValue
} from
'../utility-types'
5 import { EMPTY_FUNCTION
} from
'../utils'
6 import type { KillBehavior
, WorkerOptions
} from
'./worker-options'
7 import { KillBehaviors
} from
'./worker-options'
9 const DEFAULT_MAX_INACTIVE_TIME
= 60000
10 const DEFAULT_KILL_BEHAVIOR
: KillBehavior
= KillBehaviors
.SOFT
13 * Base class that implements some shared logic for all poolifier workers.
15 * @typeParam MainWorker - Type of main worker.
16 * @typeParam Data - Type of data this worker receives from pool's execution. This can only be serializable data.
17 * @typeParam Response - Type of response the worker sends back to the main worker. This can only be serializable data.
19 export abstract class AbstractWorker
<
20 MainWorker
extends Worker
| MessagePort
,
23 > extends AsyncResource
{
25 * Timestamp of the last task processed by this worker.
27 protected lastTaskTimestamp
: number
29 * Handler Id of the `aliveInterval` worker alive check.
31 protected readonly aliveInterval
?: NodeJS
.Timeout
33 * Options for the worker.
35 public readonly opts
: WorkerOptions
37 * Constructs a new poolifier worker.
39 * @param type - The type of async event.
40 * @param isMain - Whether this is the main worker or not.
41 * @param fn - Function processed by the worker when the pool's `execution` function is invoked.
42 * @param mainWorker - Reference to main worker.
43 * @param opts - Options for the worker.
48 fn
: (data
: Data
) => Response
,
49 protected mainWorker
: MainWorker
| undefined | null,
50 opts
: WorkerOptions
= {
52 * The kill behavior option on this Worker or its default value.
54 killBehavior
: DEFAULT_KILL_BEHAVIOR
,
56 * The maximum time to keep this worker alive while idle.
57 * The pool automatically checks and terminates this worker when the time expires.
59 maxInactiveTime
: DEFAULT_MAX_INACTIVE_TIME
64 this.checkFunctionInput(fn
)
65 this.checkWorkerOptions(this.opts
)
66 this.lastTaskTimestamp
= Date.now()
67 // Keep the worker active
69 this.aliveInterval
= setInterval(
70 this.checkAlive
.bind(this),
71 (this.opts
.maxInactiveTime
?? DEFAULT_MAX_INACTIVE_TIME
) / 2
73 this.checkAlive
.bind(this)()
76 this.mainWorker
?.on('message', (value
: MessageValue
<Data
, MainWorker
>) => {
77 this.messageListener(value
, fn
)
81 protected messageListener (
82 value
: MessageValue
<Data
, MainWorker
>,
83 fn
: (data
: Data
) => Response
85 if (value
.data
!== undefined && value
.id
!== undefined) {
86 // Here you will receive messages
87 if (this.opts
.async === true) {
88 this.runInAsyncScope(this.runAsync
.bind(this), this, fn
, value
)
90 this.runInAsyncScope(this.run
.bind(this), this, fn
, value
)
92 } else if (value
.parent !== undefined) {
93 // Save a reference of the main worker to communicate with it
94 // This will be received once
95 this.mainWorker
= value
.parent
96 } else if (value
.kill
!== undefined) {
97 // Here is time to kill this worker, just clearing the interval
98 if (this.aliveInterval
!= null) clearInterval(this.aliveInterval
)
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 this.opts
.async = opts
.async ?? false
111 * Checks if the `fn` parameter is passed to the constructor.
113 * @param fn - The function that should be defined.
115 private checkFunctionInput (fn
: (data
: Data
) => Response
): void {
116 if (fn
== null) throw new Error('fn parameter is mandatory')
117 if (typeof fn
!== 'function') {
118 throw new TypeError('fn parameter is not a function')
123 * Returns the main worker.
125 * @returns Reference to the main worker.
127 protected getMainWorker (): MainWorker
{
128 if (this.mainWorker
== null) {
129 throw new Error('Main worker was not set')
131 return this.mainWorker
135 * Sends a message to the main worker.
137 * @param message - The response message.
139 protected abstract sendToMainWorker (message
: MessageValue
<Response
>): void
142 * Checks if the worker should be terminated, because its living too long.
144 protected checkAlive (): void {
146 Date.now() - this.lastTaskTimestamp
>
147 (this.opts
.maxInactiveTime
?? DEFAULT_MAX_INACTIVE_TIME
)
149 this.sendToMainWorker({ kill
: this.opts
.killBehavior
})
154 * Handles an error and convert it to a string so it can be sent back to the main worker.
156 * @param e - The error raised by the worker.
157 * @returns Message of the error.
159 protected handleError (e
: Error | string): string {
164 * Runs the given function synchronously.
166 * @param fn - Function that will be executed.
167 * @param value - Input data for the given function.
170 fn
: (data
?: Data
) => Response
,
171 value
: MessageValue
<Data
>
174 const startTaskTimestamp
= Date.now()
175 const res
= fn(value
.data
)
176 const taskRunTime
= Date.now() - startTaskTimestamp
177 this.sendToMainWorker({ data
: res
, id
: value
.id
, taskRunTime
})
179 const err
= this.handleError(e
as Error)
180 this.sendToMainWorker({ error
: err
, id
: value
.id
})
182 this.lastTaskTimestamp
= Date.now()
187 * Runs the given function asynchronously.
189 * @param fn - Function that will be executed.
190 * @param value - Input data for the given function.
193 fn
: (data
?: Data
) => Promise
<Response
>,
194 value
: MessageValue
<Data
>
196 const startTaskTimestamp
= Date.now()
199 const taskRunTime
= Date.now() - startTaskTimestamp
200 this.sendToMainWorker({ data
: res
, id
: value
.id
, taskRunTime
})
204 const err
= this.handleError(e
as Error)
205 this.sendToMainWorker({ error
: err
, id
: value
.id
})
208 this.lastTaskTimestamp
= Date.now()
210 .catch(EMPTY_FUNCTION
)