1 import { MessageChannel
} from
'node:worker_threads'
2 import { CircularArray
} from
'../circular-array'
3 import type { Task
} from
'../utility-types'
12 import { Deque
} from
'../deque'
17 type WorkerNodeEventCallback
,
26 * @typeParam Worker - Type of worker.
27 * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
29 export class WorkerNode
<Worker
extends IWorker
, Data
= unknown
>
30 implements IWorkerNode
<Worker
, Data
> {
32 public readonly worker
: Worker
34 public readonly info
: WorkerInfo
36 public usage
: WorkerUsage
38 public messageChannel
?: MessageChannel
40 public tasksQueueBackPressureSize
: number
42 public onBackPressure
?: WorkerNodeEventCallback
44 public onEmptyQueue
?: WorkerNodeEventCallback
45 private readonly tasksQueue
: Deque
<Task
<Data
>>
46 private onEmptyQueueCount
: number
47 private readonly taskFunctionsUsage
: Map
<string, WorkerUsage
>
50 * Constructs a new worker node.
52 * @param worker - The worker.
53 * @param tasksQueueBackPressureSize - The tasks queue back pressure size.
55 constructor (worker
: Worker
, tasksQueueBackPressureSize
: number) {
57 throw new TypeError('Cannot construct a worker node without a worker')
59 if (tasksQueueBackPressureSize
== null) {
61 'Cannot construct a worker node without a tasks queue back pressure size'
64 if (!Number.isSafeInteger(tasksQueueBackPressureSize
)) {
66 'Cannot construct a worker node with a tasks queue back pressure size that is not an integer'
70 this.info
= this.initWorkerInfo(worker
)
71 this.usage
= this.initWorkerUsage()
72 if (this.info
.type === WorkerTypes
.thread
) {
73 this.messageChannel
= new MessageChannel()
75 this.tasksQueueBackPressureSize
= tasksQueueBackPressureSize
76 this.tasksQueue
= new Deque
<Task
<Data
>>()
77 this.onEmptyQueueCount
= 0
78 this.taskFunctionsUsage
= new Map
<string, WorkerUsage
>()
82 public tasksQueueSize (): number {
83 return this.tasksQueue
.size
87 public enqueueTask (task
: Task
<Data
>): number {
88 const tasksQueueSize
= this.tasksQueue
.push(task
)
89 if (this.onBackPressure
!= null && this.hasBackPressure()) {
90 this.onBackPressure(this.info
.id
as number)
96 public unshiftTask (task
: Task
<Data
>): number {
97 const tasksQueueSize
= this.tasksQueue
.unshift(task
)
98 if (this.onBackPressure
!= null && this.hasBackPressure()) {
99 this.onBackPressure(this.info
.id
as number)
101 return tasksQueueSize
105 public dequeueTask (): Task
<Data
> | undefined {
106 const task
= this.tasksQueue
.shift()
107 if (this.onEmptyQueue
!= null && this.tasksQueue
.size
=== 0) {
108 this.startOnEmptyQueue().catch(EMPTY_FUNCTION
)
114 public popTask (): Task
<Data
> | undefined {
115 const task
= this.tasksQueue
.pop()
116 if (this.onEmptyQueue
!= null && this.tasksQueue
.size
=== 0) {
117 this.startOnEmptyQueue().catch(EMPTY_FUNCTION
)
123 public clearTasksQueue (): void {
124 this.tasksQueue
.clear()
128 public hasBackPressure (): boolean {
129 return this.tasksQueue
.size
>= this.tasksQueueBackPressureSize
133 public resetUsage (): void {
134 this.usage
= this.initWorkerUsage()
135 this.taskFunctionsUsage
.clear()
139 public closeChannel (): void {
140 if (this.messageChannel
!= null) {
141 this.messageChannel
?.port1
.unref()
142 this.messageChannel
?.port2
.unref()
143 this.messageChannel
?.port1
.close()
144 this.messageChannel
?.port2
.close()
145 delete this.messageChannel
150 public getTaskFunctionWorkerUsage (name
: string): WorkerUsage
| undefined {
151 if (!Array.isArray(this.info
.taskFunctions
)) {
153 `Cannot get task function worker usage for task function name '${name}' when task function names list is not yet defined`
157 Array.isArray(this.info
.taskFunctions
) &&
158 this.info
.taskFunctions
.length
< 3
161 `Cannot get task function worker usage for task function name '${name}' when task function names list has less than 3 elements`
164 if (name
=== DEFAULT_TASK_NAME
) {
165 name
= this.info
.taskFunctions
[1]
167 if (!this.taskFunctionsUsage
.has(name
)) {
168 this.taskFunctionsUsage
.set(name
, this.initTaskFunctionWorkerUsage(name
))
170 return this.taskFunctionsUsage
.get(name
)
173 private async startOnEmptyQueue (): Promise
<void> {
175 this.onEmptyQueueCount
> 0 &&
176 (this.usage
.tasks
.executing
> 0 || this.tasksQueue
.size
> 0)
178 this.onEmptyQueueCount
= 0
181 (this.onEmptyQueue
as WorkerNodeEventCallback
)(this.info
.id
as number)
182 ++this.onEmptyQueueCount
183 await sleep(exponentialDelay(this.onEmptyQueueCount
))
184 await this.startOnEmptyQueue()
187 private initWorkerInfo (worker
: Worker
): WorkerInfo
{
189 id
: getWorkerId(worker
),
190 type: getWorkerType(worker
) as WorkerType
,
196 private initWorkerUsage (): WorkerUsage
{
197 const getTasksQueueSize
= (): number => {
198 return this.tasksQueue
.size
200 const getTasksQueueMaxSize
= (): number => {
201 return this.tasksQueue
.maxSize
207 get
queued (): number {
208 return getTasksQueueSize()
210 get
maxQueued (): number {
211 return getTasksQueueMaxSize()
217 history
: new CircularArray()
220 history
: new CircularArray()
224 history
: new CircularArray()
227 history
: new CircularArray()
233 private initTaskFunctionWorkerUsage (name
: string): WorkerUsage
{
234 const getTaskFunctionQueueSize
= (): number => {
235 let taskFunctionQueueSize
= 0
236 for (const task
of this.tasksQueue
) {
238 (task
.name
=== DEFAULT_TASK_NAME
&&
239 name
=== (this.info
.taskFunctions
as string[])[1]) ||
240 (task
.name
!== DEFAULT_TASK_NAME
&& name
=== task
.name
)
242 ++taskFunctionQueueSize
245 return taskFunctionQueueSize
251 get
queued (): number {
252 return getTaskFunctionQueueSize()
258 history
: new CircularArray()
261 history
: new CircularArray()
265 history
: new CircularArray()
268 history
: new CircularArray()