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