Merge branch 'master' into feature/task-functions
[poolifier.git] / src / pools / pool.ts
1 import { EventEmitter } from 'node:events'
2 import { type TransferListItem } from 'node:worker_threads'
3 import type { TaskFunction } from '../worker/task-functions'
4 import type {
5 ErrorHandler,
6 ExitHandler,
7 IWorker,
8 IWorkerNode,
9 MessageHandler,
10 OnlineHandler,
11 WorkerType
12 } from './worker'
13 import type {
14 WorkerChoiceStrategy,
15 WorkerChoiceStrategyOptions
16 } from './selection-strategies/selection-strategies-types'
17
18 /**
19 * Enumeration of pool types.
20 */
21 export const PoolTypes = Object.freeze({
22 /**
23 * Fixed pool type.
24 */
25 fixed: 'fixed',
26 /**
27 * Dynamic pool type.
28 */
29 dynamic: 'dynamic'
30 } as const)
31
32 /**
33 * Pool type.
34 */
35 export type PoolType = keyof typeof PoolTypes
36
37 /**
38 * Pool events emitter.
39 */
40 export class PoolEmitter extends EventEmitter {}
41
42 /**
43 * Enumeration of pool events.
44 */
45 export const PoolEvents = Object.freeze({
46 ready: 'ready',
47 busy: 'busy',
48 full: 'full',
49 destroy: 'destroy',
50 error: 'error',
51 taskError: 'taskError',
52 backPressure: 'backPressure'
53 } as const)
54
55 /**
56 * Pool event.
57 */
58 export type PoolEvent = keyof typeof PoolEvents
59
60 /**
61 * Pool information.
62 */
63 export interface PoolInfo {
64 readonly version: string
65 readonly type: PoolType
66 readonly worker: WorkerType
67 readonly started: boolean
68 readonly ready: boolean
69 readonly strategy: WorkerChoiceStrategy
70 readonly minSize: number
71 readonly maxSize: number
72 /** Pool utilization. */
73 readonly utilization?: number
74 /** Pool total worker nodes. */
75 readonly workerNodes: number
76 /** Pool idle worker nodes. */
77 readonly idleWorkerNodes: number
78 /** Pool busy worker nodes. */
79 readonly busyWorkerNodes: number
80 readonly executedTasks: number
81 readonly executingTasks: number
82 readonly queuedTasks?: number
83 readonly maxQueuedTasks?: number
84 readonly backPressure?: boolean
85 readonly stolenTasks?: number
86 readonly failedTasks: number
87 readonly runTime?: {
88 readonly minimum: number
89 readonly maximum: number
90 readonly average?: number
91 readonly median?: number
92 }
93 readonly waitTime?: {
94 readonly minimum: number
95 readonly maximum: number
96 readonly average?: number
97 readonly median?: number
98 }
99 }
100
101 /**
102 * Worker node tasks queue options.
103 */
104 export interface TasksQueueOptions {
105 /**
106 * Maximum tasks queue size per worker node flagging it as back pressured.
107 *
108 * @defaultValue (pool maximum size)^2
109 */
110 readonly size?: number
111 /**
112 * Maximum number of tasks that can be executed concurrently on a worker node.
113 *
114 * @defaultValue 1
115 */
116 readonly concurrency?: number
117 /**
118 * Whether to enable task stealing.
119 *
120 * @defaultValue true
121 */
122 readonly taskStealing?: boolean
123 /**
124 * Whether to enable tasks stealing on back pressure.
125 *
126 * @defaultValue true
127 */
128 readonly tasksStealingOnBackPressure?: boolean
129 }
130
131 /**
132 * Options for a poolifier pool.
133 *
134 * @typeParam Worker - Type of worker.
135 */
136 export interface PoolOptions<Worker extends IWorker> {
137 /**
138 * A function that will listen for online event on each worker.
139 *
140 * @defaultValue `() => {}`
141 */
142 onlineHandler?: OnlineHandler<Worker>
143 /**
144 * A function that will listen for message event on each worker.
145 *
146 * @defaultValue `() => {}`
147 */
148 messageHandler?: MessageHandler<Worker>
149 /**
150 * A function that will listen for error event on each worker.
151 *
152 * @defaultValue `() => {}`
153 */
154 errorHandler?: ErrorHandler<Worker>
155 /**
156 * A function that will listen for exit event on each worker.
157 *
158 * @defaultValue `() => {}`
159 */
160 exitHandler?: ExitHandler<Worker>
161 /**
162 * Whether to start the minimum number of workers at pool initialization.
163 *
164 * @defaultValue true
165 */
166 startWorkers?: boolean
167 /**
168 * The worker choice strategy to use in this pool.
169 *
170 * @defaultValue WorkerChoiceStrategies.ROUND_ROBIN
171 */
172 workerChoiceStrategy?: WorkerChoiceStrategy
173 /**
174 * The worker choice strategy options.
175 */
176 workerChoiceStrategyOptions?: WorkerChoiceStrategyOptions
177 /**
178 * Restart worker on error.
179 */
180 restartWorkerOnError?: boolean
181 /**
182 * Pool events emission.
183 *
184 * @defaultValue true
185 */
186 enableEvents?: boolean
187 /**
188 * Pool worker node tasks queue.
189 *
190 * @defaultValue false
191 */
192 enableTasksQueue?: boolean
193 /**
194 * Pool worker node tasks queue options.
195 */
196 tasksQueueOptions?: TasksQueueOptions
197 }
198
199 /**
200 * Contract definition for a poolifier pool.
201 *
202 * @typeParam Worker - Type of worker which manages this pool.
203 * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
204 * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
205 */
206 export interface IPool<
207 Worker extends IWorker,
208 Data = unknown,
209 Response = unknown
210 > {
211 /**
212 * Pool information.
213 */
214 readonly info: PoolInfo
215 /**
216 * Pool worker nodes.
217 *
218 * @internal
219 */
220 readonly workerNodes: Array<IWorkerNode<Worker, Data>>
221 /**
222 * Whether the worker node has back pressure (i.e. its tasks queue is full).
223 *
224 * @param workerNodeKey - The worker node key.
225 * @returns `true` if the worker node has back pressure, `false` otherwise.
226 * @internal
227 */
228 readonly hasWorkerNodeBackPressure: (workerNodeKey: number) => boolean
229 /**
230 * Emitter on which events can be listened to.
231 *
232 * Events that can currently be listened to:
233 *
234 * - `'ready'`: Emitted when the number of workers created in the pool has reached the minimum size expected and are ready.
235 * - `'busy'`: Emitted when the number of workers created in the pool has reached the maximum size expected and are executing concurrently their tasks quota.
236 * - `'full'`: Emitted when the pool is dynamic and the number of workers created has reached the maximum size expected.
237 * - `'destroy'`: Emitted when the pool is destroyed.
238 * - `'error'`: Emitted when an uncaught error occurs.
239 * - `'taskError'`: Emitted when an error occurs while executing a task.
240 * - `'backPressure'`: Emitted when all worker nodes have back pressure (i.e. their tasks queue is full: queue size \>= maximum queue size).
241 */
242 readonly emitter?: PoolEmitter
243 /**
244 * Executes the specified function in the worker constructor with the task data input parameter.
245 *
246 * @param data - The optional task input data for the specified task function. This can only be structured-cloneable data.
247 * @param name - The optional name of the task function to execute. If not specified, the default task function will be executed.
248 * @param transferList - An optional array of transferable objects to transfer ownership of. Ownership of the transferred objects is given to the pool's worker_threads worker and they should not be used in the main thread afterwards.
249 * @returns Promise that will be fulfilled when the task is completed.
250 */
251 readonly execute: (
252 data?: Data,
253 name?: string,
254 transferList?: TransferListItem[]
255 ) => Promise<Response>
256 /**
257 * Starts the minimum number of workers in this pool.
258 */
259 readonly start: () => void
260 /**
261 * Terminates all workers in this pool.
262 */
263 readonly destroy: () => Promise<void>
264 /**
265 * Whether the specified task function exists in this pool.
266 *
267 * @param name - The name of the task function.
268 * @returns `true` if the task function exists, `false` otherwise.
269 */
270 readonly hasTaskFunction: (name: string) => boolean
271 /**
272 * Adds a task function to this pool.
273 * If a task function with the same name already exists, it will be overwritten.
274 *
275 * @param name - The name of the task function.
276 * @param taskFunction - The task function.
277 * @returns `true` if the task function was added, `false` otherwise.
278 */
279 readonly addTaskFunction: (
280 name: string,
281 taskFunction: TaskFunction<Data, Response>
282 ) => Promise<boolean>
283 /**
284 * Removes a task function from this pool.
285 *
286 * @param name - The name of the task function.
287 * @returns `true` if the task function was removed, `false` otherwise.
288 */
289 readonly removeTaskFunction: (name: string) => Promise<boolean>
290 /**
291 * Lists the names of task function available in this pool.
292 *
293 * @returns The names of task function available in this pool.
294 */
295 readonly listTaskFunctionNames: () => string[]
296 /**
297 * Sets the default task function in this pool.
298 *
299 * @param name - The name of the task function.
300 * @returns `true` if the default task function was set, `false` otherwise.
301 */
302 readonly setDefaultTaskFunction: (name: string) => Promise<boolean>
303 /**
304 * Sets the worker choice strategy in this pool.
305 *
306 * @param workerChoiceStrategy - The worker choice strategy.
307 * @param workerChoiceStrategyOptions - The worker choice strategy options.
308 */
309 readonly setWorkerChoiceStrategy: (
310 workerChoiceStrategy: WorkerChoiceStrategy,
311 workerChoiceStrategyOptions?: WorkerChoiceStrategyOptions
312 ) => void
313 /**
314 * Sets the worker choice strategy options in this pool.
315 *
316 * @param workerChoiceStrategyOptions - The worker choice strategy options.
317 */
318 readonly setWorkerChoiceStrategyOptions: (
319 workerChoiceStrategyOptions: WorkerChoiceStrategyOptions
320 ) => void
321 /**
322 * Enables/disables the worker node tasks queue in this pool.
323 *
324 * @param enable - Whether to enable or disable the worker node tasks queue.
325 * @param tasksQueueOptions - The worker node tasks queue options.
326 */
327 readonly enableTasksQueue: (
328 enable: boolean,
329 tasksQueueOptions?: TasksQueueOptions
330 ) => void
331 /**
332 * Sets the worker node tasks queue options in this pool.
333 *
334 * @param tasksQueueOptions - The worker node tasks queue options.
335 */
336 readonly setTasksQueueOptions: (tasksQueueOptions: TasksQueueOptions) => void
337 }