1 import { EventEmitter
} from
'node:events'
2 import { MessageChannel
} from
'node:worker_threads'
4 import { CircularBuffer
} from
'../circular-buffer.js'
5 import { PriorityQueue
} from
'../queues/priority-queue.js'
6 import type { Task
} from
'../utility-types.js'
7 import { DEFAULT_TASK_NAME
} from
'../utils.js'
9 checkWorkerNodeArguments
,
18 MeasurementHistorySize
,
21 type WorkerNodeOptions
,
29 * @typeParam Worker - Type of worker.
30 * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
32 export class WorkerNode
<Worker
extends IWorker
, Data
= unknown
>
34 implements IWorkerNode
<Worker
, Data
> {
36 public readonly worker
: Worker
38 public readonly info
: WorkerInfo
40 public usage
: WorkerUsage
42 public strategyData
?: StrategyData
44 public messageChannel
?: MessageChannel
46 public tasksQueueBackPressureSize
: number
47 private readonly tasksQueue
: PriorityQueue
<Task
<Data
>>
48 private setBackPressureFlag
: boolean
49 private readonly taskFunctionsUsage
: Map
<string, WorkerUsage
>
52 * Constructs a new worker node.
53 * @param type - The worker type.
54 * @param filePath - Path to the worker file.
55 * @param opts - The worker node options.
57 constructor (type: WorkerType
, filePath
: string, opts
: WorkerNodeOptions
) {
59 checkWorkerNodeArguments(type, filePath
, opts
)
60 this.worker
= createWorker
<Worker
>(type, filePath
, {
62 workerOptions
: opts
.workerOptions
,
64 this.info
= this.initWorkerInfo(this.worker
)
65 this.usage
= this.initWorkerUsage()
66 if (this.info
.type === WorkerTypes
.thread
) {
67 this.messageChannel
= new MessageChannel()
69 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
70 this.tasksQueueBackPressureSize
= opts
.tasksQueueBackPressureSize
!
71 this.tasksQueue
= new PriorityQueue
<Task
<Data
>>(
72 opts
.tasksQueueBucketSize
,
73 opts
.tasksQueuePriority
75 this.setBackPressureFlag
= false
76 this.taskFunctionsUsage
= new Map
<string, WorkerUsage
>()
80 public setTasksQueuePriority (enablePriority
: boolean): void {
81 this.tasksQueue
.enablePriority
= enablePriority
85 public tasksQueueSize (): number {
86 return this.tasksQueue
.size
90 public enqueueTask (task
: Task
<Data
>): number {
91 const tasksQueueSize
= this.tasksQueue
.enqueue(task
, task
.priority
)
93 !this.setBackPressureFlag
&&
94 this.hasBackPressure() &&
95 !this.info
.backPressure
97 this.setBackPressureFlag
= true
98 this.info
.backPressure
= true
99 this.emit('backPressure', { workerId
: this.info
.id
})
100 this.setBackPressureFlag
= false
102 return tasksQueueSize
106 public dequeueTask (bucket
?: number): Task
<Data
> | undefined {
107 const task
= this.tasksQueue
.dequeue(bucket
)
109 !this.setBackPressureFlag
&&
110 !this.hasBackPressure() &&
111 this.info
.backPressure
113 this.setBackPressureFlag
= true
114 this.info
.backPressure
= false
115 this.setBackPressureFlag
= false
121 public dequeueLastPrioritizedTask (): Task
<Data
> | undefined {
122 // Start from the last empty or partially filled bucket
123 return this.dequeueTask(this.tasksQueue
.buckets
+ 1)
127 public clearTasksQueue (): void {
128 this.tasksQueue
.clear()
132 public hasBackPressure (): boolean {
133 return this.tasksQueue
.size
>= this.tasksQueueBackPressureSize
137 public async terminate (): Promise
<void> {
138 const waitWorkerExit
= new Promise
<void>(resolve
=> {
139 this.registerOnceWorkerEventHandler('exit', () => {
143 this.closeMessageChannel()
144 this.removeAllListeners()
145 switch (this.info
.type) {
146 case WorkerTypes
.thread
:
147 this.worker
.unref
?.()
148 await this.worker
.terminate
?.()
150 case WorkerTypes
.cluster
:
151 this.registerOnceWorkerEventHandler('disconnect', () => {
154 this.worker
.disconnect
?.()
161 public registerWorkerEventHandler (
163 handler
: EventHandler
<Worker
>
165 this.worker
.on(event
, handler
)
169 public registerOnceWorkerEventHandler (
171 handler
: EventHandler
<Worker
>
173 this.worker
.once(event
, handler
)
177 public getTaskFunctionWorkerUsage (name
: string): WorkerUsage
| undefined {
178 if (!Array.isArray(this.info
.taskFunctionsProperties
)) {
180 `Cannot get task function worker usage for task function name '${name}' when task function properties list is not yet defined`
184 Array.isArray(this.info
.taskFunctionsProperties
) &&
185 this.info
.taskFunctionsProperties
.length
< 3
188 `Cannot get task function worker usage for task function name '${name}' when task function properties list has less than 3 elements`
191 if (name
=== DEFAULT_TASK_NAME
) {
192 name
= this.info
.taskFunctionsProperties
[1].name
194 if (!this.taskFunctionsUsage
.has(name
)) {
195 this.taskFunctionsUsage
.set(name
, this.initTaskFunctionWorkerUsage(name
))
197 return this.taskFunctionsUsage
.get(name
)
201 public deleteTaskFunctionWorkerUsage (name
: string): boolean {
202 return this.taskFunctionsUsage
.delete(name
)
205 private closeMessageChannel (): void {
206 if (this.messageChannel
!= null) {
207 this.messageChannel
.port1
.unref()
208 this.messageChannel
.port2
.unref()
209 this.messageChannel
.port1
.close()
210 this.messageChannel
.port2
.close()
211 delete this.messageChannel
215 private initWorkerInfo (worker
: Worker
): WorkerInfo
{
217 id
: getWorkerId(worker
),
218 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
219 type: getWorkerType(worker
)!,
227 private initWorkerUsage (): WorkerUsage
{
228 const getTasksQueueSize
= (): number => {
229 return this.tasksQueue
.size
231 const getTasksQueueMaxSize
= (): number => {
232 return this.tasksQueue
.maxSize
238 get
queued (): number {
239 return getTasksQueueSize()
241 get
maxQueued (): number {
242 return getTasksQueueMaxSize()
244 sequentiallyStolen
: 0,
249 history
: new CircularBuffer(MeasurementHistorySize
),
252 history
: new CircularBuffer(MeasurementHistorySize
),
256 history
: new CircularBuffer(MeasurementHistorySize
),
259 history
: new CircularBuffer(MeasurementHistorySize
),
265 private initTaskFunctionWorkerUsage (name
: string): WorkerUsage
{
266 const getTaskFunctionQueueSize
= (): number => {
267 let taskFunctionQueueSize
= 0
268 for (const task
of this.tasksQueue
) {
270 (task
.name
=== DEFAULT_TASK_NAME
&&
271 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
272 name
=== this.info
.taskFunctionsProperties
![1].name
) ||
273 (task
.name
!== DEFAULT_TASK_NAME
&& name
=== task
.name
)
275 ++taskFunctionQueueSize
278 return taskFunctionQueueSize
284 get
queued (): number {
285 return getTaskFunctionQueueSize()
287 sequentiallyStolen
: 0,
292 history
: new CircularBuffer(MeasurementHistorySize
),
295 history
: new CircularBuffer(MeasurementHistorySize
),
299 history
: new CircularBuffer(MeasurementHistorySize
),
302 history
: new CircularBuffer(MeasurementHistorySize
),