feat: introduce worker node queue back pressure detection
[poolifier.git] / src / pools / pool.ts
1 import { EventEmitter } from 'node:events'
2 import { type TransferListItem } from 'node:worker_threads'
3 import type {
4 ErrorHandler,
5 ExitHandler,
6 IWorker,
7 IWorkerNode,
8 MessageHandler,
9 OnlineHandler,
10 WorkerType
11 } from './worker'
12 import type {
13 WorkerChoiceStrategy,
14 WorkerChoiceStrategyOptions
15 } from './selection-strategies/selection-strategies-types'
16
17 /**
18 * Enumeration of pool types.
19 */
20 export const PoolTypes = Object.freeze({
21 /**
22 * Fixed pool type.
23 */
24 fixed: 'fixed',
25 /**
26 * Dynamic pool type.
27 */
28 dynamic: 'dynamic'
29 } as const)
30
31 /**
32 * Pool type.
33 */
34 export type PoolType = keyof typeof PoolTypes
35
36 /**
37 * Pool events emitter.
38 */
39 export class PoolEmitter extends EventEmitter {}
40
41 /**
42 * Enumeration of pool events.
43 */
44 export const PoolEvents = Object.freeze({
45 ready: 'ready',
46 busy: 'busy',
47 full: 'full',
48 destroy: 'destroy',
49 error: 'error',
50 taskError: 'taskError',
51 backPressure: 'backPressure'
52 } as const)
53
54 /**
55 * Pool event.
56 */
57 export type PoolEvent = keyof typeof PoolEvents
58
59 /**
60 * Pool information.
61 */
62 export interface PoolInfo {
63 readonly version: string
64 readonly type: PoolType
65 readonly worker: WorkerType
66 readonly ready: boolean
67 readonly strategy: WorkerChoiceStrategy
68 readonly minSize: number
69 readonly maxSize: number
70 /** Pool utilization. */
71 readonly utilization?: number
72 /** Pool total worker nodes. */
73 readonly workerNodes: number
74 /** Pool idle worker nodes. */
75 readonly idleWorkerNodes: number
76 /** Pool busy worker nodes. */
77 readonly busyWorkerNodes: number
78 readonly executedTasks: number
79 readonly executingTasks: number
80 readonly queuedTasks?: number
81 readonly maxQueuedTasks?: number
82 readonly failedTasks: number
83 readonly runTime?: {
84 readonly minimum: number
85 readonly maximum: number
86 readonly average: number
87 readonly median?: number
88 }
89 readonly waitTime?: {
90 readonly minimum: number
91 readonly maximum: number
92 readonly average: number
93 readonly median?: number
94 }
95 }
96
97 /**
98 * Worker tasks queue options.
99 */
100 export interface TasksQueueOptions {
101 /**
102 * Maximum number of tasks that can be executed concurrently on a worker.
103 *
104 * @defaultValue 1
105 */
106 readonly concurrency?: number
107 }
108
109 /**
110 * Options for a poolifier pool.
111 *
112 * @typeParam Worker - Type of worker.
113 */
114 export interface PoolOptions<Worker extends IWorker> {
115 /**
116 * A function that will listen for online event on each worker.
117 */
118 onlineHandler?: OnlineHandler<Worker>
119 /**
120 * A function that will listen for message event on each worker.
121 */
122 messageHandler?: MessageHandler<Worker>
123 /**
124 * A function that will listen for error event on each worker.
125 */
126 errorHandler?: ErrorHandler<Worker>
127 /**
128 * A function that will listen for exit event on each worker.
129 */
130 exitHandler?: ExitHandler<Worker>
131 /**
132 * The worker choice strategy to use in this pool.
133 *
134 * @defaultValue WorkerChoiceStrategies.ROUND_ROBIN
135 */
136 workerChoiceStrategy?: WorkerChoiceStrategy
137 /**
138 * The worker choice strategy options.
139 */
140 workerChoiceStrategyOptions?: WorkerChoiceStrategyOptions
141 /**
142 * Restart worker on error.
143 */
144 restartWorkerOnError?: boolean
145 /**
146 * Pool events emission.
147 *
148 * @defaultValue true
149 */
150 enableEvents?: boolean
151 /**
152 * Pool worker tasks queue.
153 *
154 * @defaultValue false
155 */
156 enableTasksQueue?: boolean
157 /**
158 * Pool worker tasks queue options.
159 */
160 tasksQueueOptions?: TasksQueueOptions
161 }
162
163 /**
164 * Contract definition for a poolifier pool.
165 *
166 * @typeParam Worker - Type of worker which manages this pool.
167 * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
168 * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
169 */
170 export interface IPool<
171 Worker extends IWorker,
172 Data = unknown,
173 Response = unknown
174 > {
175 /**
176 * Pool information.
177 */
178 readonly info: PoolInfo
179 /**
180 * Pool worker nodes.
181 */
182 readonly workerNodes: Array<IWorkerNode<Worker, Data>>
183 /**
184 * Emitter on which events can be listened to.
185 *
186 * Events that can currently be listened to:
187 *
188 * - `'ready'`: Emitted when the number of workers created in the pool has reached the minimum size expected and are ready.
189 * - `'busy'`: Emitted when the number of workers created in the pool has reached the maximum size expected and are executing at least one task.
190 * - `'full'`: Emitted when the pool is dynamic and the number of workers created has reached the maximum size expected.
191 * - '`destroy`': Emitted when the pool is destroyed.
192 * - `'error'`: Emitted when an uncaught error occurs.
193 * - `'taskError'`: Emitted when an error occurs while executing a task.
194 */
195 readonly emitter?: PoolEmitter
196 /**
197 * Executes the specified function in the worker constructor with the task data input parameter.
198 *
199 * @param data - The optional task input data for the specified task function. This can only be structured-cloneable data.
200 * @param name - The optional name of the task function to execute. If not specified, the default task function will be executed.
201 * @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.
202 * @returns Promise that will be fulfilled when the task is completed.
203 */
204 readonly execute: (
205 data?: Data,
206 name?: string,
207 transferList?: TransferListItem[]
208 ) => Promise<Response>
209 /**
210 * Terminates all workers in this pool.
211 */
212 readonly destroy: () => Promise<void>
213 /**
214 * Lists the names of task function available in this pool.
215 *
216 * @returns The names of task function available in this pool.
217 */
218 readonly listTaskFunctions: () => string[]
219 /**
220 * Sets the worker choice strategy in this pool.
221 *
222 * @param workerChoiceStrategy - The worker choice strategy.
223 * @param workerChoiceStrategyOptions - The worker choice strategy options.
224 */
225 readonly setWorkerChoiceStrategy: (
226 workerChoiceStrategy: WorkerChoiceStrategy,
227 workerChoiceStrategyOptions?: WorkerChoiceStrategyOptions
228 ) => void
229 /**
230 * Sets the worker choice strategy options in this pool.
231 *
232 * @param workerChoiceStrategyOptions - The worker choice strategy options.
233 */
234 readonly setWorkerChoiceStrategyOptions: (
235 workerChoiceStrategyOptions: WorkerChoiceStrategyOptions
236 ) => void
237 /**
238 * Enables/disables the worker tasks queue in this pool.
239 *
240 * @param enable - Whether to enable or disable the worker tasks queue.
241 * @param tasksQueueOptions - The worker tasks queue options.
242 */
243 readonly enableTasksQueue: (
244 enable: boolean,
245 tasksQueueOptions?: TasksQueueOptions
246 ) => void
247 /**
248 * Sets the worker tasks queue options in this pool.
249 *
250 * @param tasksQueueOptions - The worker tasks queue options.
251 */
252 readonly setTasksQueueOptions: (tasksQueueOptions: TasksQueueOptions) => void
253 }