923fe9b222a4ea9dc4279294d51d0755f67f1ba7
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 { DEFAULT_TASK_NAME
} from
'../utils'
18 * @typeParam Worker - Type of worker.
19 * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
21 export class WorkerNode
<Worker
extends IWorker
, Data
= unknown
>
22 implements IWorkerNode
<Worker
, Data
> {
23 public readonly worker
: Worker
24 public readonly info
: WorkerInfo
25 public usage
: WorkerUsage
26 public taskFunctions
!: string[]
27 private readonly tasksUsage
: Map
<string, WorkerUsage
>
28 private readonly tasksQueue
: Queue
<Task
<Data
>>
31 * Constructs a new worker node.
33 * @param worker - The worker.
34 * @param workerType - The worker type.
36 constructor (worker
: Worker
, workerType
: WorkerType
) {
38 this.info
= this.initWorkerInfo(worker
, workerType
)
39 this.usage
= this.initWorkerUsage()
40 this.tasksUsage
= new Map
<string, WorkerUsage
>()
41 this.tasksQueue
= new Queue
<Task
<Data
>>()
45 public tasksQueueSize (): number {
46 return this.tasksQueue
.size
50 * Tasks queue maximum size.
52 * @returns The tasks queue maximum size.
54 private tasksQueueMaxSize (): number {
55 return this.tasksQueue
.maxSize
59 public enqueueTask (task
: Task
<Data
>): number {
60 return this.tasksQueue
.enqueue(task
)
64 public dequeueTask (): Task
<Data
> | undefined {
65 return this.tasksQueue
.dequeue()
69 public clearTasksQueue (): void {
70 this.tasksQueue
.clear()
74 public resetUsage (): void {
75 this.usage
= this.initWorkerUsage()
76 this.tasksUsage
.clear()
80 public closeChannel (): void {
81 if (this.info
.messageChannel
!= null) {
82 this.info
.messageChannel
?.port1
.unref()
83 this.info
.messageChannel
?.port2
.unref()
84 this.info
.messageChannel
?.port1
.close()
85 this.info
.messageChannel
?.port2
.close()
86 delete this.info
.messageChannel
91 public getTaskWorkerUsage (name
: string): WorkerUsage
| undefined {
93 name
=== DEFAULT_TASK_NAME
&&
94 Array.isArray(this.taskFunctions
) &&
95 this.taskFunctions
.length
> 1
97 name
= this.taskFunctions
[1]
99 if (!this.tasksUsage
.has(name
)) {
100 this.tasksUsage
.set(name
, this.initTaskWorkerUsage(name
))
102 return this.tasksUsage
.get(name
)
105 private initWorkerInfo (worker
: Worker
, workerType
: WorkerType
): WorkerInfo
{
107 id
: this.getWorkerId(worker
, workerType
),
111 ...(workerType
=== WorkerTypes
.thread
&& {
112 messageChannel
: new MessageChannel()
117 private initWorkerUsage (): WorkerUsage
{
118 const getTasksQueueSize
= (): number => {
119 return this.tasksQueueSize()
121 const getTasksQueueMaxSize
= (): number => {
122 return this.tasksQueueMaxSize()
128 get
queued (): number {
129 return getTasksQueueSize()
131 get
maxQueued (): number {
132 return getTasksQueueMaxSize()
137 history
: new CircularArray()
140 history
: new CircularArray()
144 history
: new CircularArray()
147 history
: new CircularArray()
153 private initTaskWorkerUsage (name
: string): WorkerUsage
{
154 const getTaskQueueSize
= (): number => {
155 let taskQueueSize
= 0
156 for (const task
of this.tasksQueue
) {
157 if (task
.name
=== name
) {
167 get
queued (): number {
168 return getTaskQueueSize()
173 history
: new CircularArray()
176 history
: new CircularArray()
180 history
: new CircularArray()
183 history
: new CircularArray()
190 * Gets the worker id.
192 * @param worker - The worker.
193 * @param workerType - The worker type.
194 * @returns The worker id.
196 private getWorkerId (
198 workerType
: WorkerType
199 ): number | undefined {
200 if (workerType
=== WorkerTypes
.thread
) {
201 return worker
.threadId
202 } else if (workerType
=== WorkerTypes
.cluster
) {