Merge branch 'master' of github.com:poolifier/poolifier
[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 { KillBehaviors, type WorkerOptions } from './worker-options'
18 import type {
19 TaskAsyncFunction,
20 TaskFunction,
21 TaskFunctionOperationResult,
22 TaskFunctions,
23 TaskSyncFunction
24 } from './task-functions'
25
26 const DEFAULT_MAX_INACTIVE_TIME = 60000
27 const DEFAULT_WORKER_OPTIONS: WorkerOptions = {
28 /**
29 * The kill behavior option on this worker or its default value.
30 */
31 killBehavior: KillBehaviors.SOFT,
32 /**
33 * The maximum time to keep this worker active while idle.
34 * The pool automatically checks and terminates this worker when the time expires.
35 */
36 maxInactiveTime: DEFAULT_MAX_INACTIVE_TIME,
37 /**
38 * The function to call when the worker is killed.
39 */
40 killHandler: EMPTY_FUNCTION
41 }
42
43 /**
44 * Base class that implements some shared logic for all poolifier workers.
45 *
46 * @typeParam MainWorker - Type of main worker.
47 * @typeParam Data - Type of data this worker receives from pool's execution. This can only be structured-cloneable data.
48 * @typeParam Response - Type of response the worker sends back to the main worker. This can only be structured-cloneable data.
49 */
50 export abstract class AbstractWorker<
51 MainWorker extends Worker | MessagePort,
52 Data = unknown,
53 Response = unknown
54 > extends AsyncResource {
55 /**
56 * Worker id.
57 */
58 protected abstract id: number
59 /**
60 * Task function(s) processed by the worker when the pool's `execution` function is invoked.
61 */
62 protected taskFunctions!: Map<string, TaskFunction<Data, Response>>
63 /**
64 * Timestamp of the last task processed by this worker.
65 */
66 protected lastTaskTimestamp!: number
67 /**
68 * Performance statistics computation requirements.
69 */
70 protected statistics!: WorkerStatistics
71 /**
72 * Handler id of the `activeInterval` worker activity check.
73 */
74 protected activeInterval?: NodeJS.Timeout
75 /**
76 * Constructs a new poolifier worker.
77 *
78 * @param type - The type of async event.
79 * @param isMain - Whether this is the main worker or not.
80 * @param mainWorker - Reference to main worker.
81 * @param taskFunctions - Task function(s) processed by the worker when the pool's `execution` function is invoked. The first function is the default function.
82 * @param opts - Options for the worker.
83 */
84 public constructor (
85 type: string,
86 protected readonly isMain: boolean,
87 private readonly mainWorker: MainWorker,
88 taskFunctions: TaskFunction<Data, Response> | TaskFunctions<Data, Response>,
89 protected opts: WorkerOptions = DEFAULT_WORKER_OPTIONS
90 ) {
91 super(type)
92 if (this.isMain == null) {
93 throw new Error('isMain parameter is mandatory')
94 }
95 this.checkTaskFunctions(taskFunctions)
96 this.checkWorkerOptions(this.opts)
97 if (!this.isMain) {
98 this.getMainWorker().on('message', this.handleReadyMessage.bind(this))
99 }
100 }
101
102 private checkWorkerOptions (opts: WorkerOptions): void {
103 if (opts != null && !isPlainObject(opts)) {
104 throw new TypeError('opts worker options parameter is not a plain object')
105 }
106 if (
107 opts?.killBehavior != null &&
108 !Object.values(KillBehaviors).includes(opts.killBehavior)
109 ) {
110 throw new TypeError(
111 `killBehavior option '${opts.killBehavior}' is not valid`
112 )
113 }
114 if (
115 opts?.maxInactiveTime != null &&
116 !Number.isSafeInteger(opts.maxInactiveTime)
117 ) {
118 throw new TypeError('maxInactiveTime option is not an integer')
119 }
120 if (opts?.maxInactiveTime != null && opts.maxInactiveTime < 5) {
121 throw new TypeError(
122 'maxInactiveTime option is not a positive integer greater or equal than 5'
123 )
124 }
125 if (opts?.killHandler != null && typeof opts.killHandler !== 'function') {
126 throw new TypeError('killHandler option is not a function')
127 }
128 if (opts?.async != null) {
129 throw new Error('async option is deprecated')
130 }
131 this.opts = { ...DEFAULT_WORKER_OPTIONS, ...opts }
132 }
133
134 private checkValidTaskFunctionEntry (
135 name: string,
136 fn: TaskFunction<Data, Response>
137 ): void {
138 if (typeof name !== 'string') {
139 throw new TypeError(
140 'A taskFunctions parameter object key is not a string'
141 )
142 }
143 if (typeof name === 'string' && name.trim().length === 0) {
144 throw new TypeError(
145 'A taskFunctions parameter object key is an empty string'
146 )
147 }
148 if (typeof fn !== 'function') {
149 throw new TypeError(
150 'A taskFunctions parameter object value is not a function'
151 )
152 }
153 }
154
155 /**
156 * Checks if the `taskFunctions` parameter is passed to the constructor and valid.
157 *
158 * @param taskFunctions - The task function(s) parameter that should be checked.
159 */
160 private checkTaskFunctions (
161 taskFunctions: TaskFunction<Data, Response> | TaskFunctions<Data, Response>
162 ): void {
163 if (taskFunctions == null) {
164 throw new Error('taskFunctions parameter is mandatory')
165 }
166 this.taskFunctions = new Map<string, TaskFunction<Data, Response>>()
167 if (typeof taskFunctions === 'function') {
168 const boundFn = taskFunctions.bind(this)
169 this.taskFunctions.set(DEFAULT_TASK_NAME, boundFn)
170 this.taskFunctions.set(
171 typeof taskFunctions.name === 'string' &&
172 taskFunctions.name.trim().length > 0
173 ? taskFunctions.name
174 : 'fn1',
175 boundFn
176 )
177 } else if (isPlainObject(taskFunctions)) {
178 let firstEntry = true
179 for (const [name, fn] of Object.entries(taskFunctions)) {
180 this.checkValidTaskFunctionEntry(name, fn)
181 const boundFn = fn.bind(this)
182 if (firstEntry) {
183 this.taskFunctions.set(DEFAULT_TASK_NAME, boundFn)
184 firstEntry = false
185 }
186 this.taskFunctions.set(name, boundFn)
187 }
188 if (firstEntry) {
189 throw new Error('taskFunctions parameter object is empty')
190 }
191 } else {
192 throw new TypeError(
193 'taskFunctions parameter is not a function or a plain object'
194 )
195 }
196 }
197
198 /**
199 * Checks if the worker has a task function with the given name.
200 *
201 * @param name - The name of the task function to check.
202 * @returns Whether the worker has a task function with the given name or not.
203 */
204 public hasTaskFunction (name: string): TaskFunctionOperationResult {
205 try {
206 this.checkTaskFunctionName(name)
207 } catch (error) {
208 return { status: false, error: error as Error }
209 }
210 return { status: this.taskFunctions.has(name) }
211 }
212
213 /**
214 * Adds a task function to the worker.
215 * If a task function with the same name already exists, it is replaced.
216 *
217 * @param name - The name of the task function to add.
218 * @param fn - The task function to add.
219 * @returns Whether the task function was added or not.
220 */
221 public addTaskFunction (
222 name: string,
223 fn: TaskFunction<Data, Response>
224 ): TaskFunctionOperationResult {
225 try {
226 this.checkTaskFunctionName(name)
227 if (name === DEFAULT_TASK_NAME) {
228 throw new Error(
229 'Cannot add a task function with the default reserved name'
230 )
231 }
232 if (typeof fn !== 'function') {
233 throw new TypeError('fn parameter is not a function')
234 }
235 const boundFn = fn.bind(this)
236 if (
237 this.taskFunctions.get(name) ===
238 this.taskFunctions.get(DEFAULT_TASK_NAME)
239 ) {
240 this.taskFunctions.set(DEFAULT_TASK_NAME, boundFn)
241 }
242 this.taskFunctions.set(name, boundFn)
243 this.sendTaskFunctionNamesToMainWorker()
244 return { status: true }
245 } catch (error) {
246 return { status: false, error: error as Error }
247 }
248 }
249
250 /**
251 * Removes a task function from the worker.
252 *
253 * @param name - The name of the task function to remove.
254 * @returns Whether the task function existed and was removed or not.
255 */
256 public removeTaskFunction (name: string): TaskFunctionOperationResult {
257 try {
258 this.checkTaskFunctionName(name)
259 if (name === DEFAULT_TASK_NAME) {
260 throw new Error(
261 'Cannot remove the task function with the default reserved name'
262 )
263 }
264 if (
265 this.taskFunctions.get(name) ===
266 this.taskFunctions.get(DEFAULT_TASK_NAME)
267 ) {
268 throw new Error(
269 'Cannot remove the task function used as the default task function'
270 )
271 }
272 const deleteStatus = this.taskFunctions.delete(name)
273 this.sendTaskFunctionNamesToMainWorker()
274 return { status: deleteStatus }
275 } catch (error) {
276 return { status: false, error: error as Error }
277 }
278 }
279
280 /**
281 * Lists the names of the worker's task functions.
282 *
283 * @returns The names of the worker's task functions.
284 */
285 public listTaskFunctionNames (): string[] {
286 const names: string[] = [...this.taskFunctions.keys()]
287 let defaultTaskFunctionName: string = DEFAULT_TASK_NAME
288 for (const [name, fn] of this.taskFunctions) {
289 if (
290 name !== DEFAULT_TASK_NAME &&
291 fn === this.taskFunctions.get(DEFAULT_TASK_NAME)
292 ) {
293 defaultTaskFunctionName = name
294 break
295 }
296 }
297 return [
298 names[names.indexOf(DEFAULT_TASK_NAME)],
299 defaultTaskFunctionName,
300 ...names.filter(
301 name => name !== DEFAULT_TASK_NAME && name !== defaultTaskFunctionName
302 )
303 ]
304 }
305
306 /**
307 * Sets the default task function to use in the worker.
308 *
309 * @param name - The name of the task function to use as default task function.
310 * @returns Whether the default task function was set or not.
311 */
312 public setDefaultTaskFunction (name: string): TaskFunctionOperationResult {
313 try {
314 this.checkTaskFunctionName(name)
315 if (name === DEFAULT_TASK_NAME) {
316 throw new Error(
317 'Cannot set the default task function reserved name as the default task function'
318 )
319 }
320 if (!this.taskFunctions.has(name)) {
321 throw new Error(
322 'Cannot set the default task function to a non-existing task function'
323 )
324 }
325 this.taskFunctions.set(
326 DEFAULT_TASK_NAME,
327 this.taskFunctions.get(name) as TaskFunction<Data, Response>
328 )
329 this.sendTaskFunctionNamesToMainWorker()
330 return { status: true }
331 } catch (error) {
332 return { status: false, error: error as Error }
333 }
334 }
335
336 private checkTaskFunctionName (name: string): void {
337 if (typeof name !== 'string') {
338 throw new TypeError('name parameter is not a string')
339 }
340 if (typeof name === 'string' && name.trim().length === 0) {
341 throw new TypeError('name parameter is an empty string')
342 }
343 }
344
345 /**
346 * Handles the ready message sent by the main worker.
347 *
348 * @param message - The ready message.
349 */
350 protected abstract handleReadyMessage (message: MessageValue<Data>): void
351
352 /**
353 * Worker message listener.
354 *
355 * @param message - The received message.
356 */
357 protected messageListener (message: MessageValue<Data>): void {
358 this.checkMessageWorkerId(message)
359 if (message.statistics != null) {
360 // Statistics message received
361 this.statistics = message.statistics
362 } else if (message.checkActive != null) {
363 // Check active message received
364 message.checkActive ? this.startCheckActive() : this.stopCheckActive()
365 } else if (message.taskFunctionOperation != null) {
366 // Task function operation message received
367 this.handleTaskFunctionOperationMessage(message)
368 } else if (message.taskId != null && message.data != null) {
369 // Task message received
370 this.run(message)
371 } else if (message.kill === true) {
372 // Kill message received
373 this.handleKillMessage(message)
374 }
375 }
376
377 protected handleTaskFunctionOperationMessage (
378 message: MessageValue<Data>
379 ): void {
380 const { taskFunctionOperation, taskFunctionName, taskFunction } = message
381 let response!: TaskFunctionOperationResult
382 if (taskFunctionOperation === 'add') {
383 response = this.addTaskFunction(
384 taskFunctionName as string,
385 // eslint-disable-next-line @typescript-eslint/no-implied-eval, no-new-func
386 new Function(`return ${taskFunction as string}`)() as TaskFunction<
387 Data,
388 Response
389 >
390 )
391 } else if (taskFunctionOperation === 'remove') {
392 response = this.removeTaskFunction(taskFunctionName as string)
393 } else if (taskFunctionOperation === 'default') {
394 response = this.setDefaultTaskFunction(taskFunctionName as string)
395 }
396 this.sendToMainWorker({
397 taskFunctionOperation,
398 taskFunctionOperationStatus: response.status,
399 taskFunctionName,
400 ...(!response.status &&
401 response?.error != null && {
402 workerError: {
403 name: taskFunctionName as string,
404 message: this.handleError(response.error as Error | string)
405 }
406 })
407 })
408 }
409
410 /**
411 * Handles a kill message sent by the main worker.
412 *
413 * @param message - The kill message.
414 */
415 protected handleKillMessage (message: MessageValue<Data>): void {
416 this.stopCheckActive()
417 if (isAsyncFunction(this.opts.killHandler)) {
418 (this.opts.killHandler?.() as Promise<void>)
419 .then(() => {
420 this.sendToMainWorker({ kill: 'success' })
421 return null
422 })
423 .catch(() => {
424 this.sendToMainWorker({ kill: 'failure' })
425 })
426 .finally(() => {
427 this.emitDestroy()
428 })
429 .catch(EMPTY_FUNCTION)
430 } else {
431 try {
432 // eslint-disable-next-line @typescript-eslint/no-invalid-void-type
433 this.opts.killHandler?.() as void
434 this.sendToMainWorker({ kill: 'success' })
435 } catch {
436 this.sendToMainWorker({ kill: 'failure' })
437 } finally {
438 this.emitDestroy()
439 }
440 }
441 }
442
443 /**
444 * Check if the message worker id is set and matches the worker id.
445 *
446 * @param message - The message to check.
447 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the message worker id is not set or does not match the worker id.
448 */
449 private checkMessageWorkerId (message: MessageValue<Data>): void {
450 if (message.workerId == null) {
451 throw new Error('Message worker id is not set')
452 } else if (message.workerId != null && message.workerId !== this.id) {
453 throw new Error(
454 `Message worker id ${message.workerId} does not match the worker id ${this.id}`
455 )
456 }
457 }
458
459 /**
460 * Starts the worker check active interval.
461 */
462 private startCheckActive (): void {
463 this.lastTaskTimestamp = performance.now()
464 this.activeInterval = setInterval(
465 this.checkActive.bind(this),
466 (this.opts.maxInactiveTime ?? DEFAULT_MAX_INACTIVE_TIME) / 2
467 )
468 }
469
470 /**
471 * Stops the worker check active interval.
472 */
473 private stopCheckActive (): void {
474 if (this.activeInterval != null) {
475 clearInterval(this.activeInterval)
476 delete this.activeInterval
477 }
478 }
479
480 /**
481 * Checks if the worker should be terminated, because its living too long.
482 */
483 private checkActive (): void {
484 if (
485 performance.now() - this.lastTaskTimestamp >
486 (this.opts.maxInactiveTime ?? DEFAULT_MAX_INACTIVE_TIME)
487 ) {
488 this.sendToMainWorker({ kill: this.opts.killBehavior })
489 }
490 }
491
492 /**
493 * Returns the main worker.
494 *
495 * @returns Reference to the main worker.
496 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the main worker is not set.
497 */
498 protected getMainWorker (): MainWorker {
499 if (this.mainWorker == null) {
500 throw new Error('Main worker not set')
501 }
502 return this.mainWorker
503 }
504
505 /**
506 * Sends a message to main worker.
507 *
508 * @param message - The response message.
509 */
510 protected abstract sendToMainWorker (
511 message: MessageValue<Response, Data>
512 ): void
513
514 /**
515 * Sends task function names to the main worker.
516 */
517 protected sendTaskFunctionNamesToMainWorker (): void {
518 this.sendToMainWorker({
519 taskFunctionNames: this.listTaskFunctionNames()
520 })
521 }
522
523 /**
524 * Handles an error and convert it to a string so it can be sent back to the main worker.
525 *
526 * @param error - The error raised by the worker.
527 * @returns The error message.
528 */
529 protected handleError (error: Error | string): string {
530 return error instanceof Error ? error.message : error
531 }
532
533 /**
534 * Runs the given task.
535 *
536 * @param task - The task to execute.
537 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the task function is not found.
538 */
539 protected run (task: Task<Data>): void {
540 const { name, taskId, data } = task
541 const fn = this.taskFunctions.get(name ?? DEFAULT_TASK_NAME)
542 if (fn == null) {
543 this.sendToMainWorker({
544 workerError: {
545 name: name as string,
546 message: `Task function '${name as string}' not found`,
547 data
548 },
549 taskId
550 })
551 return
552 }
553 if (isAsyncFunction(fn)) {
554 this.runInAsyncScope(this.runAsync.bind(this), this, fn, task)
555 } else {
556 this.runInAsyncScope(this.runSync.bind(this), this, fn, task)
557 }
558 }
559
560 /**
561 * Runs the given task function synchronously.
562 *
563 * @param fn - Task function that will be executed.
564 * @param task - Input data for the task function.
565 */
566 protected runSync (
567 fn: TaskSyncFunction<Data, Response>,
568 task: Task<Data>
569 ): void {
570 const { name, taskId, data } = task
571 try {
572 let taskPerformance = this.beginTaskPerformance(name)
573 const res = fn(data)
574 taskPerformance = this.endTaskPerformance(taskPerformance)
575 this.sendToMainWorker({
576 data: res,
577 taskPerformance,
578 taskId
579 })
580 } catch (error) {
581 this.sendToMainWorker({
582 workerError: {
583 name: name as string,
584 message: this.handleError(error as Error | string),
585 data
586 },
587 taskId
588 })
589 } finally {
590 this.updateLastTaskTimestamp()
591 }
592 }
593
594 /**
595 * Runs the given task function asynchronously.
596 *
597 * @param fn - Task function that will be executed.
598 * @param task - Input data for the task function.
599 */
600 protected runAsync (
601 fn: TaskAsyncFunction<Data, Response>,
602 task: Task<Data>
603 ): void {
604 const { name, taskId, data } = task
605 let taskPerformance = this.beginTaskPerformance(name)
606 fn(data)
607 .then(res => {
608 taskPerformance = this.endTaskPerformance(taskPerformance)
609 this.sendToMainWorker({
610 data: res,
611 taskPerformance,
612 taskId
613 })
614 return undefined
615 })
616 .catch(error => {
617 this.sendToMainWorker({
618 workerError: {
619 name: name as string,
620 message: this.handleError(error as Error | string),
621 data
622 },
623 taskId
624 })
625 })
626 .finally(() => {
627 this.updateLastTaskTimestamp()
628 })
629 .catch(EMPTY_FUNCTION)
630 }
631
632 private beginTaskPerformance (name?: string): TaskPerformance {
633 this.checkStatistics()
634 return {
635 name: name ?? DEFAULT_TASK_NAME,
636 timestamp: performance.now(),
637 ...(this.statistics.elu && { elu: performance.eventLoopUtilization() })
638 }
639 }
640
641 private endTaskPerformance (
642 taskPerformance: TaskPerformance
643 ): TaskPerformance {
644 this.checkStatistics()
645 return {
646 ...taskPerformance,
647 ...(this.statistics.runTime && {
648 runTime: performance.now() - taskPerformance.timestamp
649 }),
650 ...(this.statistics.elu && {
651 elu: performance.eventLoopUtilization(taskPerformance.elu)
652 })
653 }
654 }
655
656 private checkStatistics (): void {
657 if (this.statistics == null) {
658 throw new Error('Performance statistics computation requirements not set')
659 }
660 }
661
662 private updateLastTaskTimestamp (): void {
663 if (this.activeInterval != null) {
664 this.lastTaskTimestamp = performance.now()
665 }
666 }
667 }