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