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