Merge pull request #879 from poolifier/dependabot/npm_and_yarn/examples/typescript...
[poolifier.git] / src / pools / worker-node.ts
... / ...
CommitLineData
1import { MessageChannel } from 'node:worker_threads'
2import { CircularArray } from '../circular-array'
3import { Queue } from '../queue'
4import type { Task } from '../utility-types'
5import { DEFAULT_TASK_NAME } from '../utils'
6import {
7 type IWorker,
8 type IWorkerNode,
9 type WorkerInfo,
10 type WorkerType,
11 WorkerTypes,
12 type WorkerUsage
13} from './worker'
14
15/**
16 * Worker node.
17 *
18 * @typeParam Worker - Type of worker.
19 * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
20 */
21export class WorkerNode<Worker extends IWorker, Data = unknown>
22implements IWorkerNode<Worker, Data> {
23 /** @inheritdoc */
24 public readonly worker: Worker
25 /** @inheritdoc */
26 public readonly info: WorkerInfo
27 /** @inheritdoc */
28 public messageChannel?: MessageChannel
29 /** @inheritdoc */
30 public usage: WorkerUsage
31 private readonly taskFunctionsUsage: Map<string, WorkerUsage>
32 private readonly tasksQueue: Queue<Task<Data>>
33 private readonly tasksQueueBackPressureSize: number
34
35 /**
36 * Constructs a new worker node.
37 *
38 * @param worker - The worker.
39 * @param workerType - The worker type.
40 * @param poolMaxSize - The pool maximum size.
41 */
42 constructor (worker: Worker, workerType: WorkerType, poolMaxSize: number) {
43 if (worker == null) {
44 throw new Error('Cannot construct a worker node without a worker')
45 }
46 if (workerType == null) {
47 throw new Error('Cannot construct a worker node without a worker type')
48 }
49 if (poolMaxSize == null) {
50 throw new Error(
51 'Cannot construct a worker node without a pool maximum size'
52 )
53 }
54 if (isNaN(poolMaxSize)) {
55 throw new Error(
56 'Cannot construct a worker node with a NaN pool maximum size'
57 )
58 }
59 this.worker = worker
60 this.info = this.initWorkerInfo(worker, workerType)
61 if (workerType === WorkerTypes.thread) {
62 this.messageChannel = new MessageChannel()
63 }
64 this.usage = this.initWorkerUsage()
65 this.taskFunctionsUsage = new Map<string, WorkerUsage>()
66 this.tasksQueue = new Queue<Task<Data>>()
67 this.tasksQueueBackPressureSize = Math.pow(poolMaxSize, 2)
68 }
69
70 /** @inheritdoc */
71 public tasksQueueSize (): number {
72 return this.tasksQueue.size
73 }
74
75 /**
76 * Tasks queue maximum size.
77 *
78 * @returns The tasks queue maximum size.
79 */
80 private tasksQueueMaxSize (): number {
81 return this.tasksQueue.maxSize
82 }
83
84 /** @inheritdoc */
85 public enqueueTask (task: Task<Data>): number {
86 return this.tasksQueue.enqueue(task)
87 }
88
89 /** @inheritdoc */
90 public dequeueTask (): Task<Data> | undefined {
91 return this.tasksQueue.dequeue()
92 }
93
94 /** @inheritdoc */
95 public clearTasksQueue (): void {
96 this.tasksQueue.clear()
97 }
98
99 /** @inheritdoc */
100 public hasBackPressure (): boolean {
101 return this.tasksQueue.size >= this.tasksQueueBackPressureSize
102 }
103
104 /** @inheritdoc */
105 public resetUsage (): void {
106 this.usage = this.initWorkerUsage()
107 this.taskFunctionsUsage.clear()
108 }
109
110 /** @inheritdoc */
111 public closeChannel (): void {
112 if (this.messageChannel != null) {
113 this.messageChannel?.port1.unref()
114 this.messageChannel?.port2.unref()
115 this.messageChannel?.port1.close()
116 this.messageChannel?.port2.close()
117 delete this.messageChannel
118 }
119 }
120
121 /** @inheritdoc */
122 public getTaskFunctionWorkerUsage (name: string): WorkerUsage | undefined {
123 if (!Array.isArray(this.info.taskFunctions)) {
124 throw new Error(
125 `Cannot get task function worker usage for task function name '${name}' when task function names list is not yet defined`
126 )
127 }
128 if (
129 Array.isArray(this.info.taskFunctions) &&
130 this.info.taskFunctions.length < 3
131 ) {
132 throw new Error(
133 `Cannot get task function worker usage for task function name '${name}' when task function names list has less than 3 elements`
134 )
135 }
136 if (name === DEFAULT_TASK_NAME) {
137 name = this.info.taskFunctions[1]
138 }
139 if (!this.taskFunctionsUsage.has(name)) {
140 this.taskFunctionsUsage.set(name, this.initTaskFunctionWorkerUsage(name))
141 }
142 return this.taskFunctionsUsage.get(name)
143 }
144
145 private initWorkerInfo (worker: Worker, workerType: WorkerType): WorkerInfo {
146 return {
147 id: this.getWorkerId(worker, workerType),
148 type: workerType,
149 dynamic: false,
150 ready: false
151 }
152 }
153
154 private initWorkerUsage (): WorkerUsage {
155 const getTasksQueueSize = (): number => {
156 return this.tasksQueueSize()
157 }
158 const getTasksQueueMaxSize = (): number => {
159 return this.tasksQueueMaxSize()
160 }
161 return {
162 tasks: {
163 executed: 0,
164 executing: 0,
165 get queued (): number {
166 return getTasksQueueSize()
167 },
168 get maxQueued (): number {
169 return getTasksQueueMaxSize()
170 },
171 failed: 0
172 },
173 runTime: {
174 history: new CircularArray()
175 },
176 waitTime: {
177 history: new CircularArray()
178 },
179 elu: {
180 idle: {
181 history: new CircularArray()
182 },
183 active: {
184 history: new CircularArray()
185 }
186 }
187 }
188 }
189
190 private initTaskFunctionWorkerUsage (name: string): WorkerUsage {
191 const getTaskQueueSize = (): number => {
192 let taskQueueSize = 0
193 for (const task of this.tasksQueue) {
194 if (task.name === name) {
195 ++taskQueueSize
196 }
197 }
198 return taskQueueSize
199 }
200 return {
201 tasks: {
202 executed: 0,
203 executing: 0,
204 get queued (): number {
205 return getTaskQueueSize()
206 },
207 failed: 0
208 },
209 runTime: {
210 history: new CircularArray()
211 },
212 waitTime: {
213 history: new CircularArray()
214 },
215 elu: {
216 idle: {
217 history: new CircularArray()
218 },
219 active: {
220 history: new CircularArray()
221 }
222 }
223 }
224 }
225
226 /**
227 * Gets the worker id.
228 *
229 * @param worker - The worker.
230 * @param workerType - The worker type.
231 * @returns The worker id.
232 */
233 private getWorkerId (
234 worker: Worker,
235 workerType: WorkerType
236 ): number | undefined {
237 if (workerType === WorkerTypes.thread) {
238 return worker.threadId
239 } else if (workerType === WorkerTypes.cluster) {
240 return worker.id
241 }
242 }
243}