perf: remove unneeded class indirection for dynamic pool in worker
[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'
7c24d88b 6import { type KillBehavior, type 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 /**
e088a00c 29 * Handler Id of the `aliveInterval` worker alive check.
729c563d 30 */
e088a00c 31 protected readonly aliveInterval?: NodeJS.Timeout
dc5d0cb3
JB
32 /**
33 * Options for the worker.
34 */
35 public readonly opts: WorkerOptions
c97c7edb 36 /**
729c563d 37 * Constructs a new poolifier worker.
c97c7edb 38 *
38e795c1
JB
39 * @param type - The type of async event.
40 * @param isMain - Whether this is the main worker or not.
41 * @param fn - Function processed by the worker when the pool's `execution` function is invoked.
42 * @param mainWorker - Reference to main worker.
43 * @param opts - Options for the worker.
c97c7edb
S
44 */
45 public constructor (
46 type: string,
c2ade475 47 protected readonly isMain: boolean,
c97c7edb 48 fn: (data: Data) => Response,
7e0d447f 49 protected mainWorker: MainWorker | undefined | null,
59a11fd2 50 opts: WorkerOptions = {
e088a00c
JB
51 /**
52 * The kill behavior option on this Worker or its default value.
53 */
1a81f8af 54 killBehavior: DEFAULT_KILL_BEHAVIOR,
e088a00c
JB
55 /**
56 * The maximum time to keep this worker alive while idle.
57 * The pool automatically checks and terminates this worker when the time expires.
58 */
1a81f8af 59 maxInactiveTime: DEFAULT_MAX_INACTIVE_TIME
4c35177b 60 }
c97c7edb
S
61 ) {
62 super(type)
59a11fd2 63 this.opts = opts
c510fea7 64 this.checkFunctionInput(fn)
e088a00c 65 this.checkWorkerOptions(this.opts)
7c24d88b 66 if (!this.isMain) {
a9d9ea34 67 this.lastTaskTimestamp = Date.now()
e088a00c 68 this.aliveInterval = setInterval(
c97c7edb 69 this.checkAlive.bind(this),
e088a00c 70 (this.opts.maxInactiveTime ?? DEFAULT_MAX_INACTIVE_TIME) / 2
c97c7edb
S
71 )
72 this.checkAlive.bind(this)()
73 }
838898f1
S
74
75 this.mainWorker?.on('message', (value: MessageValue<Data, MainWorker>) => {
cf597bc5 76 this.messageListener(value, fn)
838898f1 77 })
c97c7edb
S
78 }
79
cf597bc5
JB
80 protected messageListener (
81 value: MessageValue<Data, MainWorker>,
82 fn: (data: Data) => Response
83 ): void {
84 if (value.data !== undefined && value.id !== undefined) {
85 // Here you will receive messages
6bd72cd0 86 if (this.opts.async === true) {
cf597bc5
JB
87 this.runInAsyncScope(this.runAsync.bind(this), this, fn, value)
88 } else {
89 this.runInAsyncScope(this.run.bind(this), this, fn, value)
90 }
91 } else if (value.parent !== undefined) {
92 // Save a reference of the main worker to communicate with it
93 // This will be received once
94 this.mainWorker = value.parent
95 } else if (value.kill !== undefined) {
96 // Here is time to kill this worker, just clearing the interval
73cff87e 97 this.aliveInterval != null && clearInterval(this.aliveInterval)
cf597bc5
JB
98 this.emitDestroy()
99 }
100 }
101
78cea37e 102 private checkWorkerOptions (opts: WorkerOptions): void {
e088a00c
JB
103 this.opts.killBehavior = opts.killBehavior ?? DEFAULT_KILL_BEHAVIOR
104 this.opts.maxInactiveTime =
105 opts.maxInactiveTime ?? DEFAULT_MAX_INACTIVE_TIME
78cea37e 106 this.opts.async = opts.async ?? false
e088a00c
JB
107 }
108
c510fea7 109 /**
8accb8d5 110 * Checks if the `fn` parameter is passed to the constructor.
c510fea7 111 *
38e795c1 112 * @param fn - The function that should be defined.
c510fea7 113 */
a35560ba 114 private checkFunctionInput (fn: (data: Data) => Response): void {
78cea37e 115 if (fn == null) throw new Error('fn parameter is mandatory')
af5204ed
JB
116 if (typeof fn !== 'function') {
117 throw new TypeError('fn parameter is not a function')
118 }
c510fea7
APA
119 }
120
729c563d
S
121 /**
122 * Returns the main worker.
838898f1
S
123 *
124 * @returns Reference to the main worker.
729c563d 125 */
838898f1 126 protected getMainWorker (): MainWorker {
78cea37e 127 if (this.mainWorker == null) {
838898f1
S
128 throw new Error('Main worker was not set')
129 }
130 return this.mainWorker
131 }
c97c7edb 132
729c563d 133 /**
8accb8d5 134 * Sends a message to the main worker.
729c563d 135 *
38e795c1 136 * @param message - The response message.
729c563d 137 */
c97c7edb
S
138 protected abstract sendToMainWorker (message: MessageValue<Response>): void
139
729c563d 140 /**
a05c10de 141 * Checks if the worker should be terminated, because its living too long.
729c563d 142 */
c97c7edb 143 protected checkAlive (): void {
e088a00c
JB
144 if (
145 Date.now() - this.lastTaskTimestamp >
146 (this.opts.maxInactiveTime ?? DEFAULT_MAX_INACTIVE_TIME)
147 ) {
148 this.sendToMainWorker({ kill: this.opts.killBehavior })
c97c7edb
S
149 }
150 }
151
729c563d 152 /**
8accb8d5 153 * Handles an error and convert it to a string so it can be sent back to the main worker.
729c563d 154 *
38e795c1 155 * @param e - The error raised by the worker.
50eceb07 156 * @returns Message of the error.
729c563d 157 */
c97c7edb 158 protected handleError (e: Error | string): string {
4a34343d 159 return e as string
c97c7edb
S
160 }
161
729c563d 162 /**
8accb8d5 163 * Runs the given function synchronously.
729c563d 164 *
38e795c1
JB
165 * @param fn - Function that will be executed.
166 * @param value - Input data for the given function.
729c563d 167 */
c97c7edb
S
168 protected run (
169 fn: (data?: Data) => Response,
170 value: MessageValue<Data>
171 ): void {
172 try {
bf9549ae 173 const startTaskTimestamp = Date.now()
c97c7edb 174 const res = fn(value.data)
bf9549ae
JB
175 const taskRunTime = Date.now() - startTaskTimestamp
176 this.sendToMainWorker({ data: res, id: value.id, taskRunTime })
c97c7edb 177 } catch (e) {
0a23f635 178 const err = this.handleError(e as Error)
c97c7edb 179 this.sendToMainWorker({ error: err, id: value.id })
6e9d10db 180 } finally {
7c24d88b 181 !this.isMain && (this.lastTaskTimestamp = Date.now())
c97c7edb
S
182 }
183 }
184
729c563d 185 /**
8accb8d5 186 * Runs the given function asynchronously.
729c563d 187 *
38e795c1
JB
188 * @param fn - Function that will be executed.
189 * @param value - Input data for the given function.
729c563d 190 */
c97c7edb
S
191 protected runAsync (
192 fn: (data?: Data) => Promise<Response>,
193 value: MessageValue<Data>
194 ): void {
bf9549ae 195 const startTaskTimestamp = Date.now()
c97c7edb
S
196 fn(value.data)
197 .then(res => {
bf9549ae
JB
198 const taskRunTime = Date.now() - startTaskTimestamp
199 this.sendToMainWorker({ data: res, id: value.id, taskRunTime })
c97c7edb
S
200 return null
201 })
202 .catch(e => {
f3636726 203 const err = this.handleError(e as Error)
c97c7edb 204 this.sendToMainWorker({ error: err, id: value.id })
6e9d10db
JB
205 })
206 .finally(() => {
7c24d88b 207 !this.isMain && (this.lastTaskTimestamp = Date.now())
c97c7edb 208 })
6e9d10db 209 .catch(EMPTY_FUNCTION)
c97c7edb
S
210 }
211}