Merge dependabot/npm_and_yarn/examples/typescript/websocket-server-pool/ws-hybrid...
[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 const {
328 statistics,
329 checkActive,
330 taskFunctionOperation,
331 taskId,
332 data,
333 kill
334 } = message
335 if (statistics != null) {
336 // Statistics message received
337 this.statistics = statistics
338 } else if (checkActive != null) {
339 // Check active message received
340 checkActive ? this.startCheckActive() : this.stopCheckActive()
341 } else if (taskFunctionOperation != null) {
342 // Task function operation message received
343 this.handleTaskFunctionOperationMessage(message)
344 } else if (taskId != null && data != null) {
345 // Task message received
346 this.run(message)
347 } else if (kill === true) {
348 // Kill message received
349 this.handleKillMessage(message)
350 }
351 }
352
353 protected handleTaskFunctionOperationMessage (
354 message: MessageValue<Data>
355 ): void {
356 const { taskFunctionOperation, taskFunctionProperties, taskFunction } =
357 message
358 if (taskFunctionProperties == null) {
359 throw new Error(
360 'Cannot handle task function operation message without task function properties'
361 )
362 }
363 let response: TaskFunctionOperationResult
364 switch (taskFunctionOperation) {
365 case 'add':
366 response = this.addTaskFunction(taskFunctionProperties.name, {
367 // eslint-disable-next-line @typescript-eslint/no-implied-eval, no-new-func
368 taskFunction: new Function(
369 `return ${taskFunction}`
370 )() as TaskFunction<Data, Response>,
371 ...(taskFunctionProperties.priority != null && {
372 priority: taskFunctionProperties.priority
373 }),
374 ...(taskFunctionProperties.strategy != null && {
375 strategy: taskFunctionProperties.strategy
376 })
377 })
378 break
379 case 'remove':
380 response = this.removeTaskFunction(taskFunctionProperties.name)
381 break
382 case 'default':
383 response = this.setDefaultTaskFunction(taskFunctionProperties.name)
384 break
385 default:
386 response = { status: false, error: new Error('Unknown task operation') }
387 break
388 }
389 this.sendToMainWorker({
390 taskFunctionOperation,
391 taskFunctionOperationStatus: response.status,
392 taskFunctionProperties,
393 ...(!response.status &&
394 response.error != null && {
395 workerError: {
396 name: taskFunctionProperties.name,
397 message: this.handleError(response.error as Error | string)
398 }
399 })
400 })
401 }
402
403 /**
404 * Handles a kill message sent by the main worker.
405 *
406 * @param message - The kill message.
407 */
408 protected handleKillMessage (_message: MessageValue<Data>): void {
409 this.stopCheckActive()
410 if (isAsyncFunction(this.opts.killHandler)) {
411 (this.opts.killHandler() as Promise<void>)
412 .then(() => {
413 this.sendToMainWorker({ kill: 'success' })
414 return undefined
415 })
416 .catch(() => {
417 this.sendToMainWorker({ kill: 'failure' })
418 })
419 } else {
420 try {
421 // eslint-disable-next-line @typescript-eslint/no-invalid-void-type
422 this.opts.killHandler?.() as void
423 this.sendToMainWorker({ kill: 'success' })
424 } catch {
425 this.sendToMainWorker({ kill: 'failure' })
426 }
427 }
428 }
429
430 /**
431 * Check if the message worker id is set and matches the worker id.
432 *
433 * @param message - The message to check.
434 * @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.
435 */
436 private checkMessageWorkerId (message: MessageValue<Data>): void {
437 if (message.workerId == null) {
438 throw new Error('Message worker id is not set')
439 } else if (message.workerId !== this.id) {
440 throw new Error(
441 `Message worker id ${message.workerId} does not match the worker id ${this.id}`
442 )
443 }
444 }
445
446 /**
447 * Starts the worker check active interval.
448 */
449 private startCheckActive (): void {
450 this.lastTaskTimestamp = performance.now()
451 this.activeInterval = setInterval(
452 this.checkActive.bind(this),
453 (this.opts.maxInactiveTime ?? DEFAULT_MAX_INACTIVE_TIME) / 2
454 )
455 }
456
457 /**
458 * Stops the worker check active interval.
459 */
460 private stopCheckActive (): void {
461 if (this.activeInterval != null) {
462 clearInterval(this.activeInterval)
463 delete this.activeInterval
464 }
465 }
466
467 /**
468 * Checks if the worker should be terminated, because its living too long.
469 */
470 private checkActive (): void {
471 if (
472 performance.now() - this.lastTaskTimestamp >
473 (this.opts.maxInactiveTime ?? DEFAULT_MAX_INACTIVE_TIME)
474 ) {
475 this.sendToMainWorker({ kill: this.opts.killBehavior })
476 }
477 }
478
479 /**
480 * Returns the main worker.
481 *
482 * @returns Reference to the main worker.
483 * @throws {@link https://nodejs.org/api/errors.html#class-error} If the main worker is not set.
484 */
485 protected getMainWorker (): MainWorker {
486 if (this.mainWorker == null) {
487 throw new Error('Main worker not set')
488 }
489 return this.mainWorker
490 }
491
492 /**
493 * Sends a message to main worker.
494 *
495 * @param message - The response message.
496 */
497 protected abstract sendToMainWorker (
498 message: MessageValue<Response, Data>
499 ): void
500
501 /**
502 * Sends task functions properties to the main worker.
503 */
504 protected sendTaskFunctionsPropertiesToMainWorker (): void {
505 this.sendToMainWorker({
506 taskFunctionsProperties: this.listTaskFunctionsProperties()
507 })
508 }
509
510 /**
511 * Handles an error and convert it to a string so it can be sent back to the main worker.
512 *
513 * @param error - The error raised by the worker.
514 * @returns The error message.
515 */
516 protected handleError (error: Error | string): string {
517 return error instanceof Error ? error.message : error
518 }
519
520 /**
521 * Runs the given task.
522 *
523 * @param task - The task to execute.
524 */
525 protected readonly run = (task: Task<Data>): void => {
526 const { name, taskId, data } = task
527 const taskFunctionName = name ?? DEFAULT_TASK_NAME
528 if (!this.taskFunctions.has(taskFunctionName)) {
529 this.sendToMainWorker({
530 workerError: {
531 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
532 name: name!,
533 message: `Task function '${name}' not found`,
534 data
535 },
536 taskId
537 })
538 return
539 }
540 const fn = this.taskFunctions.get(taskFunctionName)?.taskFunction
541 if (isAsyncFunction(fn)) {
542 this.runAsync(fn as TaskAsyncFunction<Data, Response>, task)
543 } else {
544 this.runSync(fn as TaskSyncFunction<Data, Response>, task)
545 }
546 }
547
548 /**
549 * Runs the given task function synchronously.
550 *
551 * @param fn - Task function that will be executed.
552 * @param task - Input data for the task function.
553 */
554 protected readonly runSync = (
555 fn: TaskSyncFunction<Data, Response>,
556 task: Task<Data>
557 ): void => {
558 const { name, taskId, data } = task
559 try {
560 let taskPerformance = this.beginTaskPerformance(name)
561 const res = fn(data)
562 taskPerformance = this.endTaskPerformance(taskPerformance)
563 this.sendToMainWorker({
564 data: res,
565 taskPerformance,
566 taskId
567 })
568 } catch (error) {
569 this.sendToMainWorker({
570 workerError: {
571 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
572 name: name!,
573 message: this.handleError(error as Error | string),
574 data
575 },
576 taskId
577 })
578 } finally {
579 this.updateLastTaskTimestamp()
580 }
581 }
582
583 /**
584 * Runs the given task function asynchronously.
585 *
586 * @param fn - Task function that will be executed.
587 * @param task - Input data for the task function.
588 */
589 protected readonly runAsync = (
590 fn: TaskAsyncFunction<Data, Response>,
591 task: Task<Data>
592 ): void => {
593 const { name, taskId, data } = task
594 let taskPerformance = this.beginTaskPerformance(name)
595 fn(data)
596 .then(res => {
597 taskPerformance = this.endTaskPerformance(taskPerformance)
598 this.sendToMainWorker({
599 data: res,
600 taskPerformance,
601 taskId
602 })
603 return undefined
604 })
605 .catch((error: unknown) => {
606 this.sendToMainWorker({
607 workerError: {
608 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
609 name: name!,
610 message: this.handleError(error as Error | string),
611 data
612 },
613 taskId
614 })
615 })
616 .finally(() => {
617 this.updateLastTaskTimestamp()
618 })
619 .catch(EMPTY_FUNCTION)
620 }
621
622 private beginTaskPerformance (name?: string): TaskPerformance {
623 if (this.statistics == null) {
624 throw new Error('Performance statistics computation requirements not set')
625 }
626 return {
627 name: name ?? DEFAULT_TASK_NAME,
628 timestamp: performance.now(),
629 ...(this.statistics.elu && {
630 elu: performance.eventLoopUtilization()
631 })
632 }
633 }
634
635 private endTaskPerformance (
636 taskPerformance: TaskPerformance
637 ): TaskPerformance {
638 if (this.statistics == null) {
639 throw new Error('Performance statistics computation requirements not set')
640 }
641 return {
642 ...taskPerformance,
643 ...(this.statistics.runTime && {
644 runTime: performance.now() - taskPerformance.timestamp
645 }),
646 ...(this.statistics.elu && {
647 elu: performance.eventLoopUtilization(taskPerformance.elu)
648 })
649 }
650 }
651
652 private updateLastTaskTimestamp (): void {
653 if (this.activeInterval != null) {
654 this.lastTaskTimestamp = performance.now()
655 }
656 }
657 }