refactor: cleanup cluster ESM issue workaround
[poolifier.git] / src / worker / abstract-worker.ts
1 import { AsyncResource } from 'node:async_hooks'
2 import type { Worker } from 'node:cluster'
3 import type { MessagePort } from 'node:worker_threads'
4 import { performance } from 'node:perf_hooks'
5 import type {
6 MessageValue,
7 Task,
8 TaskPerformance,
9 WorkerStatistics
10 } from '../utility-types'
11 import {
12 DEFAULT_TASK_NAME,
13 EMPTY_FUNCTION,
14 isAsyncFunction,
15 isPlainObject
16 } from '../utils'
17 import {
18 type KillBehavior,
19 KillBehaviors,
20 type WorkerOptions
21 } from './worker-options'
22 import type {
23 TaskFunctions,
24 WorkerAsyncFunction,
25 WorkerFunction,
26 WorkerSyncFunction
27 } from './worker-functions'
28
29 const DEFAULT_MAX_INACTIVE_TIME = 60000
30 const DEFAULT_KILL_BEHAVIOR: KillBehavior = KillBehaviors.SOFT
31
32 /**
33 * Base class that implements some shared logic for all poolifier workers.
34 *
35 * @typeParam MainWorker - Type of main worker.
36 * @typeParam Data - Type of data this worker receives from pool's execution. This can only be structured-cloneable data.
37 * @typeParam Response - Type of response the worker sends back to the main worker. This can only be structured-cloneable data.
38 */
39 export abstract class AbstractWorker<
40 MainWorker extends Worker | MessagePort,
41 Data = unknown,
42 Response = unknown
43 > extends AsyncResource {
44 /**
45 * Worker id.
46 */
47 protected abstract id: number
48 /**
49 * Task function(s) processed by the worker when the pool's `execution` function is invoked.
50 */
51 protected taskFunctions!: Map<string, WorkerFunction<Data, Response>>
52 /**
53 * Timestamp of the last task processed by this worker.
54 */
55 protected lastTaskTimestamp!: number
56 /**
57 * Performance statistics computation requirements.
58 */
59 protected statistics!: WorkerStatistics
60 /**
61 * Handler id of the `activeInterval` worker activity check.
62 */
63 protected activeInterval?: NodeJS.Timeout
64 /**
65 * Constructs a new poolifier worker.
66 *
67 * @param type - The type of async event.
68 * @param isMain - Whether this is the main worker or not.
69 * @param mainWorker - Reference to main worker.
70 * @param taskFunctions - Task function(s) processed by the worker when the pool's `execution` function is invoked. The first function is the default function.
71 * @param opts - Options for the worker.
72 */
73 public constructor (
74 type: string,
75 protected readonly isMain: boolean,
76 private readonly mainWorker: MainWorker,
77 taskFunctions:
78 | WorkerFunction<Data, Response>
79 | TaskFunctions<Data, Response>,
80 protected readonly opts: WorkerOptions = {
81 /**
82 * The kill behavior option on this worker or its default value.
83 */
84 killBehavior: DEFAULT_KILL_BEHAVIOR,
85 /**
86 * The maximum time to keep this worker active while idle.
87 * The pool automatically checks and terminates this worker when the time expires.
88 */
89 maxInactiveTime: DEFAULT_MAX_INACTIVE_TIME
90 }
91 ) {
92 super(type)
93 this.checkWorkerOptions(this.opts)
94 this.checkTaskFunctions(taskFunctions)
95 if (!this.isMain) {
96 this.getMainWorker()?.on('message', this.handleReadyMessage.bind(this))
97 }
98 }
99
100 private checkWorkerOptions (opts: WorkerOptions): void {
101 this.opts.killBehavior = opts.killBehavior ?? DEFAULT_KILL_BEHAVIOR
102 this.opts.maxInactiveTime =
103 opts.maxInactiveTime ?? DEFAULT_MAX_INACTIVE_TIME
104 delete this.opts.async
105 }
106
107 /**
108 * Checks if the `taskFunctions` parameter is passed to the constructor.
109 *
110 * @param taskFunctions - The task function(s) parameter that should be checked.
111 */
112 private checkTaskFunctions (
113 taskFunctions:
114 | WorkerFunction<Data, Response>
115 | TaskFunctions<Data, Response>
116 ): void {
117 if (taskFunctions == null) {
118 throw new Error('taskFunctions parameter is mandatory')
119 }
120 this.taskFunctions = new Map<string, WorkerFunction<Data, Response>>()
121 if (typeof taskFunctions === 'function') {
122 const boundFn = taskFunctions.bind(this)
123 this.taskFunctions.set(DEFAULT_TASK_NAME, boundFn)
124 this.taskFunctions.set(
125 typeof taskFunctions.name === 'string' &&
126 taskFunctions.name.trim().length > 0
127 ? taskFunctions.name
128 : 'fn1',
129 boundFn
130 )
131 } else if (isPlainObject(taskFunctions)) {
132 let firstEntry = true
133 for (const [name, fn] of Object.entries(taskFunctions)) {
134 if (typeof name !== 'string') {
135 throw new TypeError(
136 'A taskFunctions parameter object key is not a string'
137 )
138 }
139 if (typeof fn !== 'function') {
140 throw new TypeError(
141 'A taskFunctions parameter object value is not a function'
142 )
143 }
144 const boundFn = fn.bind(this)
145 if (firstEntry) {
146 this.taskFunctions.set(DEFAULT_TASK_NAME, boundFn)
147 firstEntry = false
148 }
149 this.taskFunctions.set(name, boundFn)
150 }
151 if (firstEntry) {
152 throw new Error('taskFunctions parameter object is empty')
153 }
154 } else {
155 throw new TypeError(
156 'taskFunctions parameter is not a function or a plain object'
157 )
158 }
159 }
160
161 /**
162 * Checks if the worker has a task function with the given name.
163 *
164 * @param name - The name of the task function to check.
165 * @returns Whether the worker has a task function with the given name or not.
166 * @throws {@link https://nodejs.org/api/errors.html#class-typeerror} If the `name` parameter is not a string.
167 */
168 public hasTaskFunction (name: string): boolean {
169 if (typeof name !== 'string') {
170 throw new TypeError('name parameter is not a string')
171 }
172 return this.taskFunctions.has(name)
173 }
174
175 /**
176 * Adds a task function to the worker.
177 * If a task function with the same name already exists, it is replaced.
178 *
179 * @param name - The name of the task function to add.
180 * @param fn - The task function to add.
181 * @returns Whether the task function was added or not.
182 * @throws {@link https://nodejs.org/api/errors.html#class-typeerror} If the `name` parameter is not a string.
183 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the `name` parameter is the default task function reserved name.
184 * @throws {@link https://nodejs.org/api/errors.html#class-typeerror} If the `fn` parameter is not a function.
185 */
186 public addTaskFunction (
187 name: string,
188 fn: WorkerFunction<Data, Response>
189 ): boolean {
190 if (typeof name !== 'string') {
191 throw new TypeError('name parameter is not a string')
192 }
193 if (name === DEFAULT_TASK_NAME) {
194 throw new Error(
195 'Cannot add a task function with the default reserved name'
196 )
197 }
198 if (typeof fn !== 'function') {
199 throw new TypeError('fn parameter is not a function')
200 }
201 try {
202 const boundFn = fn.bind(this)
203 if (
204 this.taskFunctions.get(name) ===
205 this.taskFunctions.get(DEFAULT_TASK_NAME)
206 ) {
207 this.taskFunctions.set(DEFAULT_TASK_NAME, boundFn)
208 }
209 this.taskFunctions.set(name, boundFn)
210 return true
211 } catch {
212 return false
213 }
214 }
215
216 /**
217 * Removes a task function from the worker.
218 *
219 * @param name - The name of the task function to remove.
220 * @returns Whether the task function existed and was removed or not.
221 * @throws {@link https://nodejs.org/api/errors.html#class-typeerror} If the `name` parameter is not a string.
222 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the `name` parameter is the default task function reserved name.
223 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the `name` parameter is the task function used as default task function.
224 */
225 public removeTaskFunction (name: string): boolean {
226 if (typeof name !== 'string') {
227 throw new TypeError('name parameter is not a string')
228 }
229 if (name === DEFAULT_TASK_NAME) {
230 throw new Error(
231 'Cannot remove the task function with the default reserved name'
232 )
233 }
234 if (
235 this.taskFunctions.get(name) === this.taskFunctions.get(DEFAULT_TASK_NAME)
236 ) {
237 throw new Error(
238 'Cannot remove the task function used as the default task function'
239 )
240 }
241 return this.taskFunctions.delete(name)
242 }
243
244 /**
245 * Lists the names of the worker's task functions.
246 *
247 * @returns The names of the worker's task functions.
248 */
249 public listTaskFunctions (): string[] {
250 return Array.from(this.taskFunctions.keys())
251 }
252
253 /**
254 * Sets the default task function to use in the worker.
255 *
256 * @param name - The name of the task function to use as default task function.
257 * @returns Whether the default task function was set or not.
258 * @throws {@link https://nodejs.org/api/errors.html#class-typeerror} If the `name` parameter is not a string.
259 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the `name` parameter is the default task function reserved name.
260 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the `name` parameter is a non-existing task function.
261 */
262 public setDefaultTaskFunction (name: string): boolean {
263 if (typeof name !== 'string') {
264 throw new TypeError('name parameter is not a string')
265 }
266 if (name === DEFAULT_TASK_NAME) {
267 throw new Error(
268 'Cannot set the default task function reserved name as the default task function'
269 )
270 }
271 if (!this.taskFunctions.has(name)) {
272 throw new Error(
273 'Cannot set the default task function to a non-existing task function'
274 )
275 }
276 try {
277 this.taskFunctions.set(
278 DEFAULT_TASK_NAME,
279 this.taskFunctions.get(name) as WorkerFunction<Data, Response>
280 )
281 return true
282 } catch {
283 return false
284 }
285 }
286
287 /**
288 * Handles the ready message sent by the main worker.
289 *
290 * @param message - The ready message.
291 */
292 protected abstract handleReadyMessage (message: MessageValue<Data>): void
293
294 /**
295 * Worker message listener.
296 *
297 * @param message - The received message.
298 */
299 protected messageListener (message: MessageValue<Data>): void {
300 if (message.workerId === this.id) {
301 if (message.statistics != null) {
302 // Statistics message received
303 this.statistics = message.statistics
304 } else if (message.checkActive != null) {
305 // Check active message received
306 !this.isMain && message.checkActive
307 ? this.startCheckActive()
308 : this.stopCheckActive()
309 } else if (message.id != null && message.data != null) {
310 // Task message received
311 this.run(message)
312 } else if (message.kill === true) {
313 // Kill message received
314 this.handleKillMessage(message)
315 }
316 }
317 }
318
319 /**
320 * Handles a kill message sent by the main worker.
321 *
322 * @param message - The kill message.
323 */
324 protected handleKillMessage (message: MessageValue<Data>): void {
325 !this.isMain && this.stopCheckActive()
326 this.emitDestroy()
327 }
328
329 /**
330 * Starts the worker check active interval.
331 */
332 private startCheckActive (): void {
333 this.lastTaskTimestamp = performance.now()
334 this.activeInterval = setInterval(
335 this.checkActive.bind(this),
336 (this.opts.maxInactiveTime ?? DEFAULT_MAX_INACTIVE_TIME) / 2
337 )
338 }
339
340 /**
341 * Stops the worker check active interval.
342 */
343 private stopCheckActive (): void {
344 if (this.activeInterval != null) {
345 clearInterval(this.activeInterval)
346 delete this.activeInterval
347 }
348 }
349
350 /**
351 * Checks if the worker should be terminated, because its living too long.
352 */
353 private checkActive (): void {
354 if (
355 performance.now() - this.lastTaskTimestamp >
356 (this.opts.maxInactiveTime ?? DEFAULT_MAX_INACTIVE_TIME)
357 ) {
358 this.sendToMainWorker({ kill: this.opts.killBehavior, workerId: this.id })
359 }
360 }
361
362 /**
363 * Returns the main worker.
364 *
365 * @returns Reference to the main worker.
366 */
367 protected getMainWorker (): MainWorker {
368 if (this.mainWorker == null) {
369 throw new Error('Main worker not set')
370 }
371 return this.mainWorker
372 }
373
374 /**
375 * Sends a message to main worker.
376 *
377 * @param message - The response message.
378 */
379 protected abstract sendToMainWorker (
380 message: MessageValue<Response, Data>
381 ): void
382
383 /**
384 * Handles an error and convert it to a string so it can be sent back to the main worker.
385 *
386 * @param e - The error raised by the worker.
387 * @returns The error message.
388 */
389 protected handleError (e: Error | string): string {
390 return e instanceof Error ? e.message : e
391 }
392
393 /**
394 * Runs the given task.
395 *
396 * @param task - The task to execute.
397 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the task function is not found.
398 */
399 protected run (task: Task<Data>): void {
400 if (this.isMain) {
401 throw new Error('Cannot run a task in the main worker')
402 }
403 const fn = this.getTaskFunction(task.name)
404 if (isAsyncFunction(fn)) {
405 this.runInAsyncScope(this.runAsync.bind(this), this, fn, task)
406 } else {
407 this.runInAsyncScope(this.runSync.bind(this), this, fn, task)
408 }
409 }
410
411 /**
412 * Runs the given task function synchronously.
413 *
414 * @param fn - Task function that will be executed.
415 * @param task - Input data for the task function.
416 */
417 protected runSync (
418 fn: WorkerSyncFunction<Data, Response>,
419 task: Task<Data>
420 ): void {
421 try {
422 let taskPerformance = this.beginTaskPerformance(task.name)
423 const res = fn(task.data)
424 taskPerformance = this.endTaskPerformance(taskPerformance)
425 this.sendToMainWorker({
426 data: res,
427 taskPerformance,
428 workerId: this.id,
429 id: task.id
430 })
431 } catch (e) {
432 const errorMessage = this.handleError(e as Error | string)
433 this.sendToMainWorker({
434 taskError: {
435 name: task.name ?? DEFAULT_TASK_NAME,
436 message: errorMessage,
437 data: task.data
438 },
439 workerId: this.id,
440 id: task.id
441 })
442 } finally {
443 this.updateLastTaskTimestamp()
444 }
445 }
446
447 /**
448 * Runs the given task function asynchronously.
449 *
450 * @param fn - Task function that will be executed.
451 * @param task - Input data for the task function.
452 */
453 protected runAsync (
454 fn: WorkerAsyncFunction<Data, Response>,
455 task: Task<Data>
456 ): void {
457 let taskPerformance = this.beginTaskPerformance(task.name)
458 fn(task.data)
459 .then(res => {
460 taskPerformance = this.endTaskPerformance(taskPerformance)
461 this.sendToMainWorker({
462 data: res,
463 taskPerformance,
464 workerId: this.id,
465 id: task.id
466 })
467 return null
468 })
469 .catch(e => {
470 const errorMessage = this.handleError(e as Error | string)
471 this.sendToMainWorker({
472 taskError: {
473 name: task.name ?? DEFAULT_TASK_NAME,
474 message: errorMessage,
475 data: task.data
476 },
477 workerId: this.id,
478 id: task.id
479 })
480 })
481 .finally(() => {
482 this.updateLastTaskTimestamp()
483 })
484 .catch(EMPTY_FUNCTION)
485 }
486
487 /**
488 * Gets the task function with the given name.
489 *
490 * @param name - Name of the task function that will be returned.
491 * @returns The task function.
492 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the task function is not found.
493 */
494 private getTaskFunction (name?: string): WorkerFunction<Data, Response> {
495 name = name ?? DEFAULT_TASK_NAME
496 const fn = this.taskFunctions.get(name)
497 if (fn == null) {
498 throw new Error(`Task function '${name}' not found`)
499 }
500 return fn
501 }
502
503 private beginTaskPerformance (name?: string): TaskPerformance {
504 this.checkStatistics()
505 return {
506 name: name ?? DEFAULT_TASK_NAME,
507 timestamp: performance.now(),
508 ...(this.statistics.elu && { elu: performance.eventLoopUtilization() })
509 }
510 }
511
512 private endTaskPerformance (
513 taskPerformance: TaskPerformance
514 ): TaskPerformance {
515 this.checkStatistics()
516 return {
517 ...taskPerformance,
518 ...(this.statistics.runTime && {
519 runTime: performance.now() - taskPerformance.timestamp
520 }),
521 ...(this.statistics.elu && {
522 elu: performance.eventLoopUtilization(taskPerformance.elu)
523 })
524 }
525 }
526
527 private checkStatistics (): void {
528 if (this.statistics == null) {
529 throw new Error('Performance statistics computation requirements not set')
530 }
531 }
532
533 private updateLastTaskTimestamp (): void {
534 if (!this.isMain && this.activeInterval != null) {
535 this.lastTaskTimestamp = performance.now()
536 }
537 }
538 }