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 ready: boolean
68 readonly strategy: WorkerChoiceStrategy
69 readonly minSize: number
70 readonly maxSize: number
71 /** Pool utilization. */
72 readonly utilization?: number
73 /** Pool total worker nodes. */
74 readonly workerNodes: number
75 /** Pool idle worker nodes. */
76 readonly idleWorkerNodes: number
77 /** Pool busy worker nodes. */
78 readonly busyWorkerNodes: number
79 readonly executedTasks: number
80 readonly executingTasks: number
81 readonly queuedTasks?: number
82 readonly maxQueuedTasks?: number
83 readonly backPressure?: boolean
84 readonly stolenTasks?: number
85 readonly failedTasks: number
86 readonly runTime?: {
87 readonly minimum: number
88 readonly maximum: number
89 readonly average?: number
90 readonly median?: number
91 }
92 readonly waitTime?: {
93 readonly minimum: number
94 readonly maximum: number
95 readonly average?: number
96 readonly median?: number
97 }
98 }
99
100 /**
101 * Worker node tasks queue options.
102 */
103 export interface TasksQueueOptions {
104 /**
105 * Maximum tasks queue size per worker node flagging it as back pressured.
106 *
107 * @defaultValue (pool maximum size)^2
108 */
109 readonly size?: number
110 /**
111 * @deprecated Use `size` instead.
112 */
113 readonly queueMaxSize?: number
114 /**
115 * Maximum number of tasks that can be executed concurrently on a worker node.
116 *
117 * @defaultValue 1
118 */
119 readonly concurrency?: number
120 }
121
122 /**
123 * Options for a poolifier pool.
124 *
125 * @typeParam Worker - Type of worker.
126 */
127 export interface PoolOptions<Worker extends IWorker> {
128 /**
129 * A function that will listen for online event on each worker.
130 */
131 onlineHandler?: OnlineHandler<Worker>
132 /**
133 * A function that will listen for message event on each worker.
134 */
135 messageHandler?: MessageHandler<Worker>
136 /**
137 * A function that will listen for error event on each worker.
138 */
139 errorHandler?: ErrorHandler<Worker>
140 /**
141 * A function that will listen for exit event on each worker.
142 */
143 exitHandler?: ExitHandler<Worker>
144 /**
145 * The worker choice strategy to use in this pool.
146 *
147 * @defaultValue WorkerChoiceStrategies.ROUND_ROBIN
148 */
149 workerChoiceStrategy?: WorkerChoiceStrategy
150 /**
151 * The worker choice strategy options.
152 */
153 workerChoiceStrategyOptions?: WorkerChoiceStrategyOptions
154 /**
155 * Restart worker on error.
156 */
157 restartWorkerOnError?: boolean
158 /**
159 * Pool events emission.
160 *
161 * @defaultValue true
162 */
163 enableEvents?: boolean
164 /**
165 * Pool worker node tasks queue.
166 *
167 * @defaultValue false
168 */
169 enableTasksQueue?: boolean
170 /**
171 * Pool worker node tasks queue options.
172 */
173 tasksQueueOptions?: TasksQueueOptions
174 }
175
176 /**
177 * Contract definition for a poolifier pool.
178 *
179 * @typeParam Worker - Type of worker which manages this pool.
180 * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
181 * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
182 */
183 export interface IPool<
184 Worker extends IWorker,
185 Data = unknown,
186 Response = unknown
187 > {
188 /**
189 * Pool information.
190 */
191 readonly info: PoolInfo
192 /**
193 * Pool worker nodes.
194 *
195 * @internal
196 */
197 readonly workerNodes: Array<IWorkerNode<Worker, Data>>
198 /**
199 * Whether the worker node has back pressure (i.e. its tasks queue is full).
200 *
201 * @param workerNodeKey - The worker node key.
202 * @returns `true` if the worker node has back pressure, `false` otherwise.
203 * @internal
204 */
205 readonly hasWorkerNodeBackPressure: (workerNodeKey: number) => boolean
206 /**
207 * Emitter on which events can be listened to.
208 *
209 * Events that can currently be listened to:
210 *
211 * - `'ready'`: Emitted when the number of workers created in the pool has reached the minimum size expected and are ready.
212 * - `'busy'`: Emitted when the number of workers created in the pool has reached the maximum size expected and are executing concurrently their tasks quota.
213 * - `'full'`: Emitted when the pool is dynamic and the number of workers created has reached the maximum size expected.
214 * - `'destroy'`: Emitted when the pool is destroyed.
215 * - `'error'`: Emitted when an uncaught error occurs.
216 * - `'taskError'`: Emitted when an error occurs while executing a task.
217 * - `'backPressure'`: Emitted when all worker nodes have back pressure (i.e. their tasks queue is full: queue size \>= maximum queue size).
218 */
219 readonly emitter?: PoolEmitter
220 /**
221 * Executes the specified function in the worker constructor with the task data input parameter.
222 *
223 * @param data - The optional task input data for the specified task function. This can only be structured-cloneable data.
224 * @param name - The optional name of the task function to execute. If not specified, the default task function will be executed.
225 * @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.
226 * @returns Promise that will be fulfilled when the task is completed.
227 */
228 readonly execute: (
229 data?: Data,
230 name?: string,
231 transferList?: TransferListItem[]
232 ) => Promise<Response>
233 /**
234 * Terminates all workers in this pool.
235 */
236 readonly destroy: () => Promise<void>
237 /**
238 * Whether the specified task function exists in this pool.
239 *
240 * @param name - The name of the task function.
241 * @returns `true` if the task function exists, `false` otherwise.
242 */
243 readonly hasTaskFunction: (name: string) => boolean
244 /**
245 * Adds a task function to this pool.
246 * If a task function with the same name already exists, it will be overwritten.
247 *
248 * @param name - The name of the task function.
249 * @param taskFunction - The task function.
250 * @returns `true` if the task function was added, `false` otherwise.
251 */
252 readonly addTaskFunction: (
253 name: string,
254 taskFunction: TaskFunction
255 ) => boolean
256 /**
257 * Removes a task function from this pool.
258 *
259 * @param name - The name of the task function.
260 * @returns `true` if the task function was removed, `false` otherwise.
261 */
262 readonly removeTaskFunction: (name: string) => boolean
263 /**
264 * Lists the names of task function available in this pool.
265 *
266 * @returns The names of task function available in this pool.
267 */
268 readonly listTaskFunctionNames: () => string[]
269 /**
270 * Sets the default task function in this pool.
271 *
272 * @param name - The name of the task function.
273 * @returns `true` if the default task function was set, `false` otherwise.
274 */
275 readonly setDefaultTaskFunction: (name: string) => boolean
276 /**
277 * Sets the worker choice strategy in this pool.
278 *
279 * @param workerChoiceStrategy - The worker choice strategy.
280 * @param workerChoiceStrategyOptions - The worker choice strategy options.
281 */
282 readonly setWorkerChoiceStrategy: (
283 workerChoiceStrategy: WorkerChoiceStrategy,
284 workerChoiceStrategyOptions?: WorkerChoiceStrategyOptions
285 ) => void
286 /**
287 * Sets the worker choice strategy options in this pool.
288 *
289 * @param workerChoiceStrategyOptions - The worker choice strategy options.
290 */
291 readonly setWorkerChoiceStrategyOptions: (
292 workerChoiceStrategyOptions: WorkerChoiceStrategyOptions
293 ) => void
294 /**
295 * Enables/disables the worker node tasks queue in this pool.
296 *
297 * @param enable - Whether to enable or disable the worker node tasks queue.
298 * @param tasksQueueOptions - The worker node tasks queue options.
299 */
300 readonly enableTasksQueue: (
301 enable: boolean,
302 tasksQueueOptions?: TasksQueueOptions
303 ) => void
304 /**
305 * Sets the worker node tasks queue options in this pool.
306 *
307 * @param tasksQueueOptions - The worker node tasks queue options.
308 */
309 readonly setTasksQueueOptions: (tasksQueueOptions: TasksQueueOptions) => void
310 }