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