Fix JSDoc (#171)
[poolifier.git] / src / pools / abstract-pool.ts
... / ...
CommitLineData
1import EventEmitter from 'events'
2import type { MessageValue } from '../utility-types'
3import type { IPool } from './pool'
4
5/**
6 * Callback invoked if the worker raised an error.
7 */
8export type ErrorHandler<Worker> = (this: Worker, e: Error) => void
9
10/**
11 * Callback invoked when the worker has started successfully.
12 */
13export type OnlineHandler<Worker> = (this: Worker) => void
14
15/**
16 * Callback invoked when the worker exits successfully.
17 */
18export type ExitHandler<Worker> = (this: Worker, code: number) => void
19
20/**
21 * Basic interface that describes the minimum required implementation of listener events for a pool-worker.
22 */
23export interface IWorker {
24 on(event: 'error', handler: ErrorHandler<this>): void
25 on(event: 'online', handler: OnlineHandler<this>): void
26 on(event: 'exit', handler: ExitHandler<this>): void
27 once(event: 'exit', handler: ExitHandler<this>): void
28}
29
30/**
31 * Options for a poolifier pool.
32 */
33export interface PoolOptions<Worker> {
34 /**
35 * A function that will listen for error event on each worker.
36 */
37 errorHandler?: ErrorHandler<Worker>
38 /**
39 * A function that will listen for online event on each worker.
40 */
41 onlineHandler?: OnlineHandler<Worker>
42 /**
43 * A function that will listen for exit event on each worker.
44 */
45 exitHandler?: ExitHandler<Worker>
46 /**
47 * This is just to avoid non-useful warning messages.
48 *
49 * Will be used to set `maxListeners` on event emitters (workers are event emitters).
50 *
51 * @default 1000
52 * @see [Node events emitter.setMaxListeners(n)](https://nodejs.org/api/events.html#events_emitter_setmaxlisteners_n)
53 */
54 maxTasks?: number
55}
56
57/**
58 * Internal poolifier pool emitter.
59 */
60class PoolEmitter extends EventEmitter {}
61
62/**
63 * Base class containing some shared logic for all poolifier pools.
64 *
65 * @template Worker Type of worker which manages this pool.
66 * @template Data Type of data sent to the worker.
67 * @template Response Type of response of execution.
68 */
69export abstract class AbstractPool<
70 Worker extends IWorker,
71 Data = unknown,
72 Response = unknown
73> implements IPool<Data, Response> {
74 /**
75 * List of currently available workers.
76 */
77 public readonly workers: Worker[] = []
78
79 /**
80 * Index for the next worker.
81 */
82 public nextWorkerIndex: number = 0
83
84 /**
85 * - `key`: The `Worker`
86 * - `value`: Number of tasks currently in progress on the worker.
87 */
88 public readonly tasks: Map<Worker, number> = new Map<Worker, number>()
89
90 /**
91 * Emitter on which events can be listened to.
92 *
93 * Events that can currently be listened to:
94 *
95 * - `'FullPool'`
96 */
97 public readonly emitter: PoolEmitter
98
99 /**
100 * ID of the next message.
101 */
102 protected nextMessageId: number = 0
103
104 /**
105 * Constructs a new poolifier pool.
106 *
107 * @param numberOfWorkers Number of workers that this pool should manage.
108 * @param filePath Path to the worker-file.
109 * @param opts Options for the pool. Default: `{ maxTasks: 1000 }`
110 */
111 public constructor (
112 public readonly numberOfWorkers: number,
113 public readonly filePath: string,
114 public readonly opts: PoolOptions<Worker> = { maxTasks: 1000 }
115 ) {
116 if (!this.isMain()) {
117 throw new Error('Cannot start a pool from a worker!')
118 }
119 // TODO christopher 2021-02-07: Improve this check e.g. with a pattern or blank check
120 if (!this.filePath) {
121 throw new Error('Please specify a file with a worker implementation')
122 }
123 this.setupHook()
124
125 for (let i = 1; i <= this.numberOfWorkers; i++) {
126 this.createAndSetupWorker()
127 }
128
129 this.emitter = new PoolEmitter()
130 }
131
132 /**
133 * Number of workers that this pool should manage.
134 *
135 * @returns Number of workers that this pool manages.
136 * @deprecated Only here for backward compatibility.
137 */
138 // eslint-disable-next-line spellcheck/spell-checker
139 public get numWorkers (): number {
140 return this.numberOfWorkers
141 }
142
143 /**
144 * Index for the next worker.
145 *
146 * @returns Index for the next worker.
147 * @deprecated Only here for backward compatibility.
148 */
149 public get nextWorker (): number {
150 return this.nextWorkerIndex
151 }
152
153 public execute (data: Data): Promise<Response> {
154 // Configure worker to handle message with the specified task
155 const worker = this.chooseWorker()
156 this.increaseWorkersTask(worker)
157 const messageId = ++this.nextMessageId
158 const res = this.internalExecute(worker, messageId)
159 this.sendToWorker(worker, { data: data || ({} as Data), id: messageId })
160 return res
161 }
162
163 public async destroy (): Promise<void> {
164 await Promise.all(this.workers.map(worker => this.destroyWorker(worker)))
165 }
166
167 /**
168 * Shut down given worker.
169 *
170 * @param worker A worker within `workers`.
171 */
172 protected abstract destroyWorker (worker: Worker): void | Promise<void>
173
174 /**
175 * Setup hook that can be overridden by a Poolifier pool implementation
176 * to run code before workers are created in the abstract constructor.
177 */
178 protected setupHook (): void {
179 // Can be overridden
180 }
181
182 /**
183 * Should return whether the worker is the main worker or not.
184 */
185 protected abstract isMain (): boolean
186
187 /**
188 * Increase the number of tasks that the given workers has done.
189 *
190 * @param worker Workers whose tasks are increased.
191 */
192 protected increaseWorkersTask (worker: Worker): void {
193 const numberOfTasksInProgress = this.tasks.get(worker)
194 if (numberOfTasksInProgress !== undefined) {
195 this.tasks.set(worker, numberOfTasksInProgress + 1)
196 } else {
197 throw Error('Worker could not be found in tasks map')
198 }
199 }
200
201 /**
202 * Decrease the number of tasks that the given workers has done.
203 *
204 * @param worker Workers whose tasks are decreased.
205 */
206 protected decreaseWorkersTasks (worker: Worker): void {
207 const numberOfTasksInProgress = this.tasks.get(worker)
208 if (numberOfTasksInProgress !== undefined) {
209 this.tasks.set(worker, numberOfTasksInProgress - 1)
210 } else {
211 throw Error('Worker could not be found in tasks map')
212 }
213 }
214
215 /**
216 * Removes the given worker from the pool.
217 *
218 * @param worker Worker that will be removed.
219 */
220 protected removeWorker (worker: Worker): void {
221 // Clean worker from data structure
222 const workerIndex = this.workers.indexOf(worker)
223 this.workers.splice(workerIndex, 1)
224 this.tasks.delete(worker)
225 }
226
227 /**
228 * Choose a worker for the next task.
229 *
230 * The default implementation uses a round robin algorithm to distribute the load.
231 *
232 * @returns Worker.
233 */
234 protected chooseWorker (): Worker {
235 const chosenWorker = this.workers[this.nextWorkerIndex]
236 this.nextWorkerIndex =
237 this.workers.length - 1 === this.nextWorkerIndex
238 ? 0
239 : this.nextWorkerIndex + 1
240 return chosenWorker
241 }
242
243 /**
244 * Send a message to the given worker.
245 *
246 * @param worker The worker which should receive the message.
247 * @param message The message.
248 */
249 protected abstract sendToWorker (
250 worker: Worker,
251 message: MessageValue<Data>
252 ): void
253
254 protected abstract registerWorkerMessageListener<
255 Message extends Data | Response
256 > (worker: Worker, listener: (message: MessageValue<Message>) => void): void
257
258 protected abstract unregisterWorkerMessageListener<
259 Message extends Data | Response
260 > (worker: Worker, listener: (message: MessageValue<Message>) => void): void
261
262 protected internalExecute (
263 worker: Worker,
264 messageId: number
265 ): Promise<Response> {
266 return new Promise((resolve, reject) => {
267 const listener: (message: MessageValue<Response>) => void = message => {
268 if (message.id === messageId) {
269 this.unregisterWorkerMessageListener(worker, listener)
270 this.decreaseWorkersTasks(worker)
271 if (message.error) reject(message.error)
272 else resolve(message.data as Response)
273 }
274 }
275 this.registerWorkerMessageListener(worker, listener)
276 })
277 }
278
279 /**
280 * Returns a newly created worker.
281 */
282 protected abstract createWorker (): Worker
283
284 /**
285 * Function that can be hooked up when a worker has been newly created and moved to the workers registry.
286 *
287 * Can be used to update the `maxListeners` or binding the `main-worker`<->`worker` connection if not bind by default.
288 *
289 * @param worker The newly created worker.
290 */
291 protected abstract afterWorkerSetup (worker: Worker): void
292
293 /**
294 * Creates a new worker for this pool and sets it up completely.
295 *
296 * @returns New, completely set up worker.
297 */
298 protected createAndSetupWorker (): Worker {
299 const worker: Worker = this.createWorker()
300
301 worker.on('error', this.opts.errorHandler ?? (() => {}))
302 worker.on('online', this.opts.onlineHandler ?? (() => {}))
303 worker.on('exit', this.opts.exitHandler ?? (() => {}))
304 worker.once('exit', () => this.removeWorker(worker))
305
306 this.workers.push(worker)
307
308 // Init tasks map
309 this.tasks.set(worker, 0)
310
311 this.afterWorkerSetup(worker)
312
313 return worker
314 }
315}