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'
17 * @typeParam Worker - Type of worker.
18 * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
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
>>
29 * Constructs a new worker node.
31 * @param worker - The worker.
32 * @param workerType - The worker type.
34 constructor (worker
: Worker
, workerType
: WorkerType
) {
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
>>()
43 public tasksQueueSize (): number {
44 return this.tasksQueue
.size
48 * Tasks queue maximum size.
50 * @returns The tasks queue maximum size.
52 private tasksQueueMaxSize (): number {
53 return this.tasksQueue
.maxSize
57 public enqueueTask (task
: Task
<Data
>): number {
58 return this.tasksQueue
.enqueue(task
)
62 public dequeueTask (): Task
<Data
> | undefined {
63 return this.tasksQueue
.dequeue()
67 public clearTasksQueue (): void {
68 this.tasksQueue
.clear()
72 public resetUsage (): void {
73 this.usage
= this.initWorkerUsage()
74 this.tasksUsage
.clear()
78 public closeChannel (): void {
79 if (this.info
.messageChannel
!= null) {
80 this.info
.messageChannel
?.port1
.close()
81 this.info
.messageChannel
?.port2
.close()
82 delete this.info
.messageChannel
87 public getTaskWorkerUsage (name
: string): WorkerUsage
| undefined {
88 if (!this.tasksUsage
.has(name
)) {
89 this.tasksUsage
.set(name
, this.initTaskWorkerUsage(name
))
91 return this.tasksUsage
.get(name
)
94 private initWorkerInfo (worker
: Worker
, workerType
: WorkerType
): WorkerInfo
{
96 id
: this.getWorkerId(worker
, workerType
),
100 ...(workerType
=== WorkerTypes
.thread
&& {
101 messageChannel
: new MessageChannel()
106 private initWorkerUsage (): WorkerUsage
{
107 const getTasksQueueSize
= (): number => {
108 return this.tasksQueueSize()
110 const getTasksQueueMaxSize
= (): number => {
111 return this.tasksQueueMaxSize()
117 get
queued (): number {
118 return getTasksQueueSize()
120 get
maxQueued (): number {
121 return getTasksQueueMaxSize()
126 history
: new CircularArray()
129 history
: new CircularArray()
133 history
: new CircularArray()
136 history
: new CircularArray()
142 private initTaskWorkerUsage (name
: string): WorkerUsage
{
143 const getTaskQueueSize
= (): number => {
144 let taskQueueSize
= 0
145 for (const task
of this.tasksQueue
) {
146 if (task
.name
=== name
) {
156 get
queued (): number {
157 return getTaskQueueSize()
162 history
: new CircularArray()
165 history
: new CircularArray()
169 history
: new CircularArray()
172 history
: new CircularArray()
179 * Gets the worker id.
181 * @param worker - The worker.
182 * @param workerType - The worker type.
183 * @returns The worker id.
185 private getWorkerId (
187 workerType
: WorkerType
188 ): number | undefined {
189 if (workerType
=== WorkerTypes
.thread
) {
190 return worker
.threadId
191 } else if (workerType
=== WorkerTypes
.cluster
) {