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