docs: enhance documentation and update changelog entries
[poolifier.git] / src / worker / abstract-worker.ts
CommitLineData
fc3e6586
JB
1import { AsyncResource } from 'node:async_hooks'
2import type { Worker } from 'node:cluster'
3import type { MessagePort } from 'node:worker_threads'
1a81f8af 4import type { MessageValue } from '../utility-types'
6e9d10db 5import { EMPTY_FUNCTION } from '../utils'
5919b303 6import type { KillBehavior, WorkerOptions } from './worker-options'
1a81f8af 7import { KillBehaviors } from './worker-options'
4c35177b 8
978aad6f 9const DEFAULT_MAX_INACTIVE_TIME = 60000
1a81f8af 10const DEFAULT_KILL_BEHAVIOR: KillBehavior = KillBehaviors.SOFT
c97c7edb 11
729c563d 12/**
ea7a90d3 13 * Base class that implements some shared logic for all poolifier workers.
729c563d 14 *
38e795c1
JB
15 * @typeParam MainWorker - Type of main worker.
16 * @typeParam Data - Type of data this worker receives from pool's execution. This can only be serializable data.
17 * @typeParam Response - Type of response the worker sends back to the main worker. This can only be serializable data.
729c563d 18 */
c97c7edb 19export abstract class AbstractWorker<
838898f1 20 MainWorker extends Worker | MessagePort,
d3c8a1a8
S
21 Data = unknown,
22 Response = unknown
c97c7edb 23> extends AsyncResource {
729c563d
S
24 /**
25 * Timestamp of the last task processed by this worker.
26 */
a9d9ea34 27 protected lastTaskTimestamp!: number
729c563d 28 /**
aee46736 29 * Handler id of the `aliveInterval` worker alive check.
729c563d 30 */
e088a00c 31 protected readonly aliveInterval?: NodeJS.Timeout
c97c7edb 32 /**
729c563d 33 * Constructs a new poolifier worker.
c97c7edb 34 *
38e795c1
JB
35 * @param type - The type of async event.
36 * @param isMain - Whether this is the main worker or not.
37 * @param fn - Function processed by the worker when the pool's `execution` function is invoked.
38 * @param mainWorker - Reference to main worker.
39 * @param opts - Options for the worker.
c97c7edb
S
40 */
41 public constructor (
42 type: string,
c2ade475 43 protected readonly isMain: boolean,
c97c7edb 44 fn: (data: Data) => Response,
7e0d447f 45 protected mainWorker: MainWorker | undefined | null,
d99ba5a8 46 protected readonly opts: WorkerOptions = {
e088a00c 47 /**
aee46736 48 * The kill behavior option on this worker or its default value.
e088a00c 49 */
1a81f8af 50 killBehavior: DEFAULT_KILL_BEHAVIOR,
e088a00c
JB
51 /**
52 * The maximum time to keep this worker alive while idle.
53 * The pool automatically checks and terminates this worker when the time expires.
54 */
1a81f8af 55 maxInactiveTime: DEFAULT_MAX_INACTIVE_TIME
4c35177b 56 }
c97c7edb
S
57 ) {
58 super(type)
c510fea7 59 this.checkFunctionInput(fn)
e088a00c 60 this.checkWorkerOptions(this.opts)
7c24d88b 61 if (!this.isMain) {
3fafb1b2 62 this.lastTaskTimestamp = performance.now()
e088a00c 63 this.aliveInterval = setInterval(
c97c7edb 64 this.checkAlive.bind(this),
e088a00c 65 (this.opts.maxInactiveTime ?? DEFAULT_MAX_INACTIVE_TIME) / 2
c97c7edb
S
66 )
67 this.checkAlive.bind(this)()
68 }
838898f1 69
aee46736
JB
70 this.mainWorker?.on(
71 'message',
72 (message: MessageValue<Data, MainWorker>) => {
73 this.messageListener(message, fn)
74 }
75 )
c97c7edb
S
76 }
77
aee46736
JB
78 /**
79 * Worker message listener.
80 *
81 * @param message - Message received.
82 * @param fn - Function processed by the worker when the pool's `execution` function is invoked.
83 */
cf597bc5 84 protected messageListener (
aee46736 85 message: MessageValue<Data, MainWorker>,
cf597bc5
JB
86 fn: (data: Data) => Response
87 ): void {
aee46736
JB
88 if (message.data != null && message.id != null) {
89 // Task message received
6bd72cd0 90 if (this.opts.async === true) {
aee46736 91 this.runInAsyncScope(this.runAsync.bind(this), this, fn, message)
cf597bc5 92 } else {
aee46736 93 this.runInAsyncScope(this.run.bind(this), this, fn, message)
cf597bc5 94 }
aee46736
JB
95 } else if (message.parent != null) {
96 // Main worker reference message received
97 this.mainWorker = message.parent
98 } else if (message.kill != null) {
99 // Kill message received
73cff87e 100 this.aliveInterval != null && clearInterval(this.aliveInterval)
cf597bc5
JB
101 this.emitDestroy()
102 }
103 }
104
78cea37e 105 private checkWorkerOptions (opts: WorkerOptions): void {
e088a00c
JB
106 this.opts.killBehavior = opts.killBehavior ?? DEFAULT_KILL_BEHAVIOR
107 this.opts.maxInactiveTime =
108 opts.maxInactiveTime ?? DEFAULT_MAX_INACTIVE_TIME
78cea37e 109 this.opts.async = opts.async ?? false
e088a00c
JB
110 }
111
c510fea7 112 /**
8accb8d5 113 * Checks if the `fn` parameter is passed to the constructor.
c510fea7 114 *
38e795c1 115 * @param fn - The function that should be defined.
c510fea7 116 */
a35560ba 117 private checkFunctionInput (fn: (data: Data) => Response): void {
78cea37e 118 if (fn == null) throw new Error('fn parameter is mandatory')
af5204ed
JB
119 if (typeof fn !== 'function') {
120 throw new TypeError('fn parameter is not a function')
121 }
c510fea7
APA
122 }
123
729c563d
S
124 /**
125 * Returns the main worker.
838898f1
S
126 *
127 * @returns Reference to the main worker.
729c563d 128 */
838898f1 129 protected getMainWorker (): MainWorker {
78cea37e 130 if (this.mainWorker == null) {
838898f1
S
131 throw new Error('Main worker was not set')
132 }
133 return this.mainWorker
134 }
c97c7edb 135
729c563d 136 /**
8accb8d5 137 * Sends a message to the main worker.
729c563d 138 *
38e795c1 139 * @param message - The response message.
729c563d 140 */
c97c7edb
S
141 protected abstract sendToMainWorker (message: MessageValue<Response>): void
142
729c563d 143 /**
a05c10de 144 * Checks if the worker should be terminated, because its living too long.
729c563d 145 */
c97c7edb 146 protected checkAlive (): void {
e088a00c 147 if (
3fafb1b2 148 performance.now() - this.lastTaskTimestamp >
e088a00c
JB
149 (this.opts.maxInactiveTime ?? DEFAULT_MAX_INACTIVE_TIME)
150 ) {
151 this.sendToMainWorker({ kill: this.opts.killBehavior })
c97c7edb
S
152 }
153 }
154
729c563d 155 /**
8accb8d5 156 * Handles an error and convert it to a string so it can be sent back to the main worker.
729c563d 157 *
38e795c1 158 * @param e - The error raised by the worker.
50eceb07 159 * @returns Message of the error.
729c563d 160 */
c97c7edb 161 protected handleError (e: Error | string): string {
4a34343d 162 return e as string
c97c7edb
S
163 }
164
729c563d 165 /**
8accb8d5 166 * Runs the given function synchronously.
729c563d 167 *
38e795c1 168 * @param fn - Function that will be executed.
aee46736 169 * @param message - Input data for the given function.
729c563d 170 */
c97c7edb
S
171 protected run (
172 fn: (data?: Data) => Response,
aee46736 173 message: MessageValue<Data>
c97c7edb
S
174 ): void {
175 try {
3fafb1b2 176 const startTimestamp = performance.now()
aee46736 177 const res = fn(message.data)
3fafb1b2
JB
178 const runTime = performance.now() - startTimestamp
179 this.sendToMainWorker({
180 data: res,
181 id: message.id,
182 runTime
183 })
c97c7edb 184 } catch (e) {
0a23f635 185 const err = this.handleError(e as Error)
aee46736 186 this.sendToMainWorker({ error: err, id: message.id })
6e9d10db 187 } finally {
3fafb1b2 188 !this.isMain && (this.lastTaskTimestamp = performance.now())
c97c7edb
S
189 }
190 }
191
729c563d 192 /**
8accb8d5 193 * Runs the given function asynchronously.
729c563d 194 *
38e795c1 195 * @param fn - Function that will be executed.
aee46736 196 * @param message - Input data for the given function.
729c563d 197 */
c97c7edb
S
198 protected runAsync (
199 fn: (data?: Data) => Promise<Response>,
aee46736 200 message: MessageValue<Data>
c97c7edb 201 ): void {
3fafb1b2 202 const startTimestamp = performance.now()
aee46736 203 fn(message.data)
c97c7edb 204 .then(res => {
3fafb1b2
JB
205 const runTime = performance.now() - startTimestamp
206 this.sendToMainWorker({
207 data: res,
208 id: message.id,
209 runTime
210 })
c97c7edb
S
211 return null
212 })
213 .catch(e => {
f3636726 214 const err = this.handleError(e as Error)
aee46736 215 this.sendToMainWorker({ error: err, id: message.id })
6e9d10db
JB
216 })
217 .finally(() => {
3fafb1b2 218 !this.isMain && (this.lastTaskTimestamp = performance.now())
c97c7edb 219 })
6e9d10db 220 .catch(EMPTY_FUNCTION)
c97c7edb
S
221 }
222}