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