refactor: cleanup cluster ESM issue workaround
[poolifier.git] / src / pools / worker-node.ts
1 import { MessageChannel } from 'node:worker_threads'
2 import { CircularArray } from '../circular-array'
3 import { Queue } from '../queue'
4 import type { Task } from '../utility-types'
5 import {
6 type IWorker,
7 type IWorkerNode,
8 type WorkerInfo,
9 type WorkerType,
10 WorkerTypes,
11 type WorkerUsage
12 } from './worker'
13
14 /**
15 * Worker node.
16 *
17 * @typeParam Worker - Type of worker.
18 * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
19 */
20 export class WorkerNode<Worker extends IWorker, Data = unknown>
21 implements IWorkerNode<Worker, Data> {
22 public readonly worker: Worker
23 public readonly info: WorkerInfo
24 public usage: WorkerUsage
25 private readonly tasksUsage: Map<string, WorkerUsage>
26 private readonly tasksQueue: Queue<Task<Data>>
27
28 /**
29 * Constructs a new worker node.
30 *
31 * @param worker - The worker.
32 * @param workerType - The worker type.
33 */
34 constructor (worker: Worker, workerType: WorkerType) {
35 this.worker = worker
36 this.info = this.initWorkerInfo(worker, workerType)
37 this.usage = this.initWorkerUsage()
38 this.tasksUsage = new Map<string, WorkerUsage>()
39 this.tasksQueue = new Queue<Task<Data>>()
40 }
41
42 /** @inheritdoc */
43 public tasksQueueSize (): number {
44 return this.tasksQueue.size
45 }
46
47 /**
48 * Tasks queue maximum size.
49 *
50 * @returns The tasks queue maximum size.
51 */
52 private tasksQueueMaxSize (): number {
53 return this.tasksQueue.maxSize
54 }
55
56 /** @inheritdoc */
57 public enqueueTask (task: Task<Data>): number {
58 return this.tasksQueue.enqueue(task)
59 }
60
61 /** @inheritdoc */
62 public dequeueTask (): Task<Data> | undefined {
63 return this.tasksQueue.dequeue()
64 }
65
66 /** @inheritdoc */
67 public clearTasksQueue (): void {
68 this.tasksQueue.clear()
69 }
70
71 /** @inheritdoc */
72 public resetUsage (): void {
73 this.usage = this.initWorkerUsage()
74 this.tasksUsage.clear()
75 }
76
77 /** @inheritdoc */
78 public closeChannel (): void {
79 if (this.info.messageChannel != null) {
80 this.info.messageChannel?.port1.unref()
81 this.info.messageChannel?.port2.unref()
82 this.info.messageChannel?.port1.close()
83 this.info.messageChannel?.port2.close()
84 delete this.info.messageChannel
85 }
86 }
87
88 /** @inheritdoc */
89 public getTaskWorkerUsage (name: string): WorkerUsage | undefined {
90 if (!this.tasksUsage.has(name)) {
91 this.tasksUsage.set(name, this.initTaskWorkerUsage(name))
92 }
93 return this.tasksUsage.get(name)
94 }
95
96 private initWorkerInfo (worker: Worker, workerType: WorkerType): WorkerInfo {
97 return {
98 id: this.getWorkerId(worker, workerType),
99 type: workerType,
100 dynamic: false,
101 ready: false,
102 ...(workerType === WorkerTypes.thread && {
103 messageChannel: new MessageChannel()
104 })
105 }
106 }
107
108 private initWorkerUsage (): WorkerUsage {
109 const getTasksQueueSize = (): number => {
110 return this.tasksQueueSize()
111 }
112 const getTasksQueueMaxSize = (): number => {
113 return this.tasksQueueMaxSize()
114 }
115 return {
116 tasks: {
117 executed: 0,
118 executing: 0,
119 get queued (): number {
120 return getTasksQueueSize()
121 },
122 get maxQueued (): number {
123 return getTasksQueueMaxSize()
124 },
125 failed: 0
126 },
127 runTime: {
128 history: new CircularArray()
129 },
130 waitTime: {
131 history: new CircularArray()
132 },
133 elu: {
134 idle: {
135 history: new CircularArray()
136 },
137 active: {
138 history: new CircularArray()
139 }
140 }
141 }
142 }
143
144 private initTaskWorkerUsage (name: string): WorkerUsage {
145 const getTaskQueueSize = (): number => {
146 let taskQueueSize = 0
147 for (const task of this.tasksQueue) {
148 if (task.name === name) {
149 ++taskQueueSize
150 }
151 }
152 return taskQueueSize
153 }
154 return {
155 tasks: {
156 executed: 0,
157 executing: 0,
158 get queued (): number {
159 return getTaskQueueSize()
160 },
161 failed: 0
162 },
163 runTime: {
164 history: new CircularArray()
165 },
166 waitTime: {
167 history: new CircularArray()
168 },
169 elu: {
170 idle: {
171 history: new CircularArray()
172 },
173 active: {
174 history: new CircularArray()
175 }
176 }
177 }
178 }
179
180 /**
181 * Gets the worker id.
182 *
183 * @param worker - The worker.
184 * @param workerType - The worker type.
185 * @returns The worker id.
186 */
187 private getWorkerId (
188 worker: Worker,
189 workerType: WorkerType
190 ): number | undefined {
191 if (workerType === WorkerTypes.thread) {
192 return worker.threadId
193 } else if (workerType === WorkerTypes.cluster) {
194 return worker.id
195 }
196 }
197 }