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