feat: check worker inactive time only on dynamic worker
[poolifier.git] / src / worker / abstract-worker.ts
CommitLineData
fc3e6586 1import { AsyncResource } from 'node:async_hooks'
6677a3d3 2import type { Worker } from 'node:cluster'
fc3e6586 3import type { MessagePort } from 'node:worker_threads'
d715b7bc
JB
4import { performance } from 'node:perf_hooks'
5import type {
6 MessageValue,
7 TaskPerformance,
8 WorkerStatistics
9} from '../utility-types'
49d1b48c 10import { EMPTY_FUNCTION, isAsyncFunction, isPlainObject } from '../utils'
241f23c2
JB
11import {
12 type KillBehavior,
13 KillBehaviors,
14 type WorkerOptions
15} from './worker-options'
b6b32453
JB
16import type {
17 TaskFunctions,
18 WorkerAsyncFunction,
19 WorkerFunction,
20 WorkerSyncFunction
21} from './worker-functions'
4c35177b 22
ec8fd331 23const DEFAULT_FUNCTION_NAME = 'default'
978aad6f 24const DEFAULT_MAX_INACTIVE_TIME = 60000
1a81f8af 25const DEFAULT_KILL_BEHAVIOR: KillBehavior = KillBehaviors.SOFT
c97c7edb 26
729c563d 27/**
ea7a90d3 28 * Base class that implements some shared logic for all poolifier workers.
729c563d 29 *
38e795c1 30 * @typeParam MainWorker - Type of main worker.
e102732c
JB
31 * @typeParam Data - Type of data this worker receives from pool's execution. This can only be structured-cloneable data.
32 * @typeParam Response - Type of response the worker sends back to the main worker. This can only be structured-cloneable data.
729c563d 33 */
c97c7edb 34export abstract class AbstractWorker<
6677a3d3 35 MainWorker extends Worker | MessagePort,
d3c8a1a8
S
36 Data = unknown,
37 Response = unknown
c97c7edb 38> extends AsyncResource {
f59e1027 39 /**
83fa0a36 40 * Worker id.
f59e1027
JB
41 */
42 protected abstract id: number
a86b6df1
JB
43 /**
44 * Task function(s) processed by the worker when the pool's `execution` function is invoked.
45 */
46 protected taskFunctions!: Map<string, WorkerFunction<Data, Response>>
729c563d
S
47 /**
48 * Timestamp of the last task processed by this worker.
49 */
a9d9ea34 50 protected lastTaskTimestamp!: number
b6b32453 51 /**
8a970421 52 * Performance statistics computation requirements.
b6b32453
JB
53 */
54 protected statistics!: WorkerStatistics
729c563d 55 /**
aee46736 56 * Handler id of the `aliveInterval` worker alive check.
729c563d 57 */
75d3401a 58 protected aliveInterval?: NodeJS.Timeout
c97c7edb 59 /**
729c563d 60 * Constructs a new poolifier worker.
c97c7edb 61 *
38e795c1
JB
62 * @param type - The type of async event.
63 * @param isMain - Whether this is the main worker or not.
82888165 64 * @param taskFunctions - Task function(s) processed by the worker when the pool's `execution` function is invoked. The first function is the default function.
38e795c1
JB
65 * @param mainWorker - Reference to main worker.
66 * @param opts - Options for the worker.
c97c7edb
S
67 */
68 public constructor (
69 type: string,
c2ade475 70 protected readonly isMain: boolean,
a86b6df1
JB
71 taskFunctions:
72 | WorkerFunction<Data, Response>
73 | TaskFunctions<Data, Response>,
448ad581 74 protected readonly mainWorker: MainWorker,
d99ba5a8 75 protected readonly opts: WorkerOptions = {
e088a00c 76 /**
aee46736 77 * The kill behavior option on this worker or its default value.
e088a00c 78 */
1a81f8af 79 killBehavior: DEFAULT_KILL_BEHAVIOR,
e088a00c
JB
80 /**
81 * The maximum time to keep this worker alive while idle.
82 * The pool automatically checks and terminates this worker when the time expires.
83 */
1a81f8af 84 maxInactiveTime: DEFAULT_MAX_INACTIVE_TIME
4c35177b 85 }
c97c7edb
S
86 ) {
87 super(type)
e088a00c 88 this.checkWorkerOptions(this.opts)
a86b6df1 89 this.checkTaskFunctions(taskFunctions)
1f68cede 90 if (!this.isMain) {
3749facb 91 this.mainWorker?.on('message', this.messageListener.bind(this))
c97c7edb
S
92 }
93 }
94
41aa7dcd
JB
95 private checkWorkerOptions (opts: WorkerOptions): void {
96 this.opts.killBehavior = opts.killBehavior ?? DEFAULT_KILL_BEHAVIOR
97 this.opts.maxInactiveTime =
98 opts.maxInactiveTime ?? DEFAULT_MAX_INACTIVE_TIME
571227f4 99 delete this.opts.async
41aa7dcd
JB
100 }
101
102 /**
a86b6df1 103 * Checks if the `taskFunctions` parameter is passed to the constructor.
41aa7dcd 104 *
82888165 105 * @param taskFunctions - The task function(s) parameter that should be checked.
41aa7dcd 106 */
a86b6df1
JB
107 private checkTaskFunctions (
108 taskFunctions:
109 | WorkerFunction<Data, Response>
110 | TaskFunctions<Data, Response>
111 ): void {
ec8fd331
JB
112 if (taskFunctions == null) {
113 throw new Error('taskFunctions parameter is mandatory')
114 }
a86b6df1 115 this.taskFunctions = new Map<string, WorkerFunction<Data, Response>>()
0d80593b
JB
116 if (typeof taskFunctions === 'function') {
117 this.taskFunctions.set(DEFAULT_FUNCTION_NAME, taskFunctions.bind(this))
118 } else if (isPlainObject(taskFunctions)) {
82888165 119 let firstEntry = true
a86b6df1
JB
120 for (const [name, fn] of Object.entries(taskFunctions)) {
121 if (typeof fn !== 'function') {
0d80593b 122 throw new TypeError(
a86b6df1
JB
123 'A taskFunctions parameter object value is not a function'
124 )
125 }
126 this.taskFunctions.set(name, fn.bind(this))
82888165
JB
127 if (firstEntry) {
128 this.taskFunctions.set(DEFAULT_FUNCTION_NAME, fn.bind(this))
129 firstEntry = false
130 }
a86b6df1 131 }
630f0acf
JB
132 if (firstEntry) {
133 throw new Error('taskFunctions parameter object is empty')
134 }
a86b6df1 135 } else {
f34fdabe
JB
136 throw new TypeError(
137 'taskFunctions parameter is not a function or a plain object'
138 )
41aa7dcd
JB
139 }
140 }
141
aee46736
JB
142 /**
143 * Worker message listener.
144 *
145 * @param message - Message received.
aee46736 146 */
6677a3d3 147 protected messageListener (message: MessageValue<Data, Data>): void {
75d3401a
JB
148 if (message.dynamic === true) {
149 // Worker dynamic message received
150 this.startCheckAlive()
151 } else if (message.statistics != null) {
152 // Statistics message received
153 this.statistics = message.statistics
154 } else if (message.id != null && message.data != null) {
aee46736 155 // Task message received
c7c04698 156 const fn = this.getTaskFunction(message.name)
49d1b48c 157 if (isAsyncFunction(fn)) {
aee46736 158 this.runInAsyncScope(this.runAsync.bind(this), this, fn, message)
cf597bc5 159 } else {
70a4f5ea 160 this.runInAsyncScope(this.runSync.bind(this), this, fn, message)
cf597bc5 161 }
aee46736
JB
162 } else if (message.kill != null) {
163 // Kill message received
73cff87e 164 this.aliveInterval != null && clearInterval(this.aliveInterval)
cf597bc5
JB
165 this.emitDestroy()
166 }
167 }
168
75d3401a
JB
169 private startCheckAlive (): void {
170 this.lastTaskTimestamp = performance.now()
171 this.aliveInterval = setInterval(
172 this.checkAlive.bind(this),
173 (this.opts.maxInactiveTime ?? DEFAULT_MAX_INACTIVE_TIME) / 2
174 )
175 this.checkAlive.bind(this)()
176 }
177
729c563d
S
178 /**
179 * Returns the main worker.
838898f1
S
180 *
181 * @returns Reference to the main worker.
729c563d 182 */
838898f1 183 protected getMainWorker (): MainWorker {
78cea37e 184 if (this.mainWorker == null) {
e102732c 185 throw new Error('Main worker not set')
838898f1
S
186 }
187 return this.mainWorker
188 }
c97c7edb 189
729c563d 190 /**
8accb8d5 191 * Sends a message to the main worker.
729c563d 192 *
38e795c1 193 * @param message - The response message.
729c563d 194 */
82f36766
JB
195 protected abstract sendToMainWorker (
196 message: MessageValue<Response, Data>
197 ): void
c97c7edb 198
729c563d 199 /**
a05c10de 200 * Checks if the worker should be terminated, because its living too long.
729c563d 201 */
c97c7edb 202 protected checkAlive (): void {
e088a00c 203 if (
3fafb1b2 204 performance.now() - this.lastTaskTimestamp >
e088a00c
JB
205 (this.opts.maxInactiveTime ?? DEFAULT_MAX_INACTIVE_TIME)
206 ) {
207 this.sendToMainWorker({ kill: this.opts.killBehavior })
c97c7edb
S
208 }
209 }
210
729c563d 211 /**
8accb8d5 212 * Handles an error and convert it to a string so it can be sent back to the main worker.
729c563d 213 *
38e795c1 214 * @param e - The error raised by the worker.
ab80dc46 215 * @returns The error message.
729c563d 216 */
c97c7edb 217 protected handleError (e: Error | string): string {
985d0e79 218 return e instanceof Error ? e.message : e
c97c7edb
S
219 }
220
729c563d 221 /**
8accb8d5 222 * Runs the given function synchronously.
729c563d 223 *
38e795c1 224 * @param fn - Function that will be executed.
aee46736 225 * @param message - Input data for the given function.
729c563d 226 */
70a4f5ea 227 protected runSync (
48ef9107 228 fn: WorkerSyncFunction<Data, Response>,
aee46736 229 message: MessageValue<Data>
c97c7edb
S
230 ): void {
231 try {
1c6fe997 232 let taskPerformance = this.beginTaskPerformance()
aee46736 233 const res = fn(message.data)
d715b7bc 234 taskPerformance = this.endTaskPerformance(taskPerformance)
3fafb1b2
JB
235 this.sendToMainWorker({
236 data: res,
d715b7bc 237 taskPerformance,
f59e1027 238 workerId: this.id,
91ee39ed 239 id: message.id
3fafb1b2 240 })
c97c7edb 241 } catch (e) {
985d0e79 242 const errorMessage = this.handleError(e as Error | string)
91ee39ed 243 this.sendToMainWorker({
82f36766 244 taskError: {
7ae6fb74 245 workerId: this.id,
985d0e79 246 message: errorMessage,
82f36766
JB
247 data: message.data
248 },
91ee39ed
JB
249 id: message.id
250 })
6e9d10db 251 } finally {
75d3401a
JB
252 if (!this.isMain && this.aliveInterval != null) {
253 this.lastTaskTimestamp = performance.now()
254 }
c97c7edb
S
255 }
256 }
257
729c563d 258 /**
8accb8d5 259 * Runs the given function asynchronously.
729c563d 260 *
38e795c1 261 * @param fn - Function that will be executed.
aee46736 262 * @param message - Input data for the given function.
729c563d 263 */
c97c7edb 264 protected runAsync (
48ef9107 265 fn: WorkerAsyncFunction<Data, Response>,
aee46736 266 message: MessageValue<Data>
c97c7edb 267 ): void {
1c6fe997 268 let taskPerformance = this.beginTaskPerformance()
aee46736 269 fn(message.data)
c97c7edb 270 .then(res => {
d715b7bc 271 taskPerformance = this.endTaskPerformance(taskPerformance)
3fafb1b2
JB
272 this.sendToMainWorker({
273 data: res,
d715b7bc 274 taskPerformance,
f59e1027 275 workerId: this.id,
91ee39ed 276 id: message.id
3fafb1b2 277 })
c97c7edb
S
278 return null
279 })
280 .catch(e => {
985d0e79 281 const errorMessage = this.handleError(e as Error | string)
91ee39ed 282 this.sendToMainWorker({
82f36766 283 taskError: {
7ae6fb74 284 workerId: this.id,
985d0e79 285 message: errorMessage,
82f36766
JB
286 data: message.data
287 },
91ee39ed
JB
288 id: message.id
289 })
6e9d10db
JB
290 })
291 .finally(() => {
75d3401a
JB
292 if (!this.isMain && this.aliveInterval != null) {
293 this.lastTaskTimestamp = performance.now()
294 }
c97c7edb 295 })
6e9d10db 296 .catch(EMPTY_FUNCTION)
c97c7edb 297 }
ec8fd331 298
82888165
JB
299 /**
300 * Gets the task function in the given scope.
301 *
302 * @param name - Name of the function that will be returned.
303 */
ec8fd331
JB
304 private getTaskFunction (name?: string): WorkerFunction<Data, Response> {
305 name = name ?? DEFAULT_FUNCTION_NAME
306 const fn = this.taskFunctions.get(name)
307 if (fn == null) {
ace229a1 308 throw new Error(`Task function '${name}' not found`)
ec8fd331
JB
309 }
310 return fn
311 }
62c15a68 312
1c6fe997 313 private beginTaskPerformance (): TaskPerformance {
8a970421 314 this.checkStatistics()
62c15a68 315 return {
1c6fe997 316 timestamp: performance.now(),
b6b32453 317 ...(this.statistics.elu && { elu: performance.eventLoopUtilization() })
62c15a68
JB
318 }
319 }
320
d9d31201
JB
321 private endTaskPerformance (
322 taskPerformance: TaskPerformance
323 ): TaskPerformance {
8a970421 324 this.checkStatistics()
62c15a68
JB
325 return {
326 ...taskPerformance,
b6b32453
JB
327 ...(this.statistics.runTime && {
328 runTime: performance.now() - taskPerformance.timestamp
329 }),
330 ...(this.statistics.elu && {
62c15a68 331 elu: performance.eventLoopUtilization(taskPerformance.elu)
b6b32453 332 })
62c15a68
JB
333 }
334 }
8a970421
JB
335
336 private checkStatistics (): void {
337 if (this.statistics == null) {
338 throw new Error('Performance statistics computation requirements not set')
339 }
340 }
c97c7edb 341}