src/pools/worker-node.ts

   1 import { MessageChannel } from 'node:worker_threads'
   2 import { CircularArray } from '../circular-array'
   3 import type { Task } from '../utility-types'
   4 import {
   5   DEFAULT_TASK_NAME,
   6   EMPTY_FUNCTION,
   7   exponentialDelay,
   8   sleep
   9 } from '../utils'
  10 import { Deque } from '../deque'
  11 import {
  12   type IWorker,
  13   type IWorkerNode,
  14   type WorkerInfo,
  15   type WorkerType,
  16   WorkerTypes,
  17   type WorkerUsage
  18 } from './worker'
  19
  20 /**
  21  * Worker node.
  22  *
  23  * @typeParam Worker - Type of worker.
  24  * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
  25  */
  26 export class WorkerNode<Worker extends IWorker, Data = unknown>
  27 implements IWorkerNode<Worker, Data> {
  28   /** @inheritdoc */
  29   public readonly worker: Worker
  30   /** @inheritdoc */
  31   public readonly info: WorkerInfo
  32   /** @inheritdoc */
  33   public messageChannel?: MessageChannel
  34   /** @inheritdoc */
  35   public usage: WorkerUsage
  36   /** @inheritdoc */
  37   public tasksQueueBackPressureSize: number
  38   /** @inheritdoc */
  39   public onBackPressure?: (workerId: number) => void
  40   /** @inheritdoc */
  41   public onEmptyQueue?: (workerId: number) => void
  42   private readonly taskFunctionsUsage: Map<string, WorkerUsage>
  43   private readonly tasksQueue: Deque<Task<Data>>
  44   private onEmptyQueueCount: number
  45
  46   /**
  47    * Constructs a new worker node.
  48    *
  49    * @param worker - The worker.
  50    * @param workerType - The worker type.
  51    * @param tasksQueueBackPressureSize - The tasks queue back pressure size.
  52    */
  53   constructor (
  54     worker: Worker,
  55     workerType: WorkerType,
  56     tasksQueueBackPressureSize: number
  57   ) {
  58     if (worker == null) {
  59       throw new TypeError('Cannot construct a worker node without a worker')
  60     }
  61     if (workerType == null) {
  62       throw new TypeError(
  63         'Cannot construct a worker node without a worker type'
  64       )
  65     }
  66     if (tasksQueueBackPressureSize == null) {
  67       throw new TypeError(
  68         'Cannot construct a worker node without a tasks queue back pressure size'
  69       )
  70     }
  71     if (!Number.isSafeInteger(tasksQueueBackPressureSize)) {
  72       throw new TypeError(
  73         'Cannot construct a worker node with a tasks queue back pressure size that is not an integer'
  74       )
  75     }
  76     this.worker = worker
  77     this.info = this.initWorkerInfo(worker, workerType)
  78     if (workerType === WorkerTypes.thread) {
  79       this.messageChannel = new MessageChannel()
  80     }
  81     this.usage = this.initWorkerUsage()
  82     this.taskFunctionsUsage = new Map<string, WorkerUsage>()
  83     this.tasksQueue = new Deque<Task<Data>>()
  84     this.tasksQueueBackPressureSize = tasksQueueBackPressureSize
  85     this.onEmptyQueueCount = 0
  86   }
  87
  88   /** @inheritdoc */
  89   public tasksQueueSize (): number {
  90     return this.tasksQueue.size
  91   }
  92
  93   /** @inheritdoc */
  94   public enqueueTask (task: Task<Data>): number {
  95     const tasksQueueSize = this.tasksQueue.push(task)
  96     if (this.onBackPressure != null && this.hasBackPressure()) {
  97       this.onBackPressure(this.info.id as number)
  98     }
  99     return tasksQueueSize
 100   }
 101
 102   /** @inheritdoc */
 103   public unshiftTask (task: Task<Data>): number {
 104     const tasksQueueSize = this.tasksQueue.unshift(task)
 105     if (this.onBackPressure != null && this.hasBackPressure()) {
 106       this.onBackPressure(this.info.id as number)
 107     }
 108     return tasksQueueSize
 109   }
 110
 111   /** @inheritdoc */
 112   public dequeueTask (): Task<Data> | undefined {
 113     const task = this.tasksQueue.shift()
 114     if (this.onEmptyQueue != null && this.tasksQueue.size === 0) {
 115       this.startOnEmptyQueue().catch(EMPTY_FUNCTION)
 116     }
 117     return task
 118   }
 119
 120   /** @inheritdoc */
 121   public popTask (): Task<Data> | undefined {
 122     const task = this.tasksQueue.pop()
 123     if (this.onEmptyQueue != null && this.tasksQueue.size === 0) {
 124       this.startOnEmptyQueue().catch(EMPTY_FUNCTION)
 125     }
 126     return task
 127   }
 128
 129   /** @inheritdoc */
 130   public clearTasksQueue (): void {
 131     this.tasksQueue.clear()
 132   }
 133
 134   /** @inheritdoc */
 135   public hasBackPressure (): boolean {
 136     return this.tasksQueue.size >= this.tasksQueueBackPressureSize
 137   }
 138
 139   /** @inheritdoc */
 140   public resetUsage (): void {
 141     this.usage = this.initWorkerUsage()
 142     this.taskFunctionsUsage.clear()
 143   }
 144
 145   /** @inheritdoc */
 146   public closeChannel (): void {
 147     if (this.messageChannel != null) {
 148       this.messageChannel?.port1.unref()
 149       this.messageChannel?.port2.unref()
 150       this.messageChannel?.port1.close()
 151       this.messageChannel?.port2.close()
 152       delete this.messageChannel
 153     }
 154   }
 155
 156   /** @inheritdoc */
 157   public getTaskFunctionWorkerUsage (name: string): WorkerUsage | undefined {
 158     if (!Array.isArray(this.info.taskFunctions)) {
 159       throw new Error(
 160         `Cannot get task function worker usage for task function name '${name}' when task function names list is not yet defined`
 161       )
 162     }
 163     if (
 164       Array.isArray(this.info.taskFunctions) &&
 165       this.info.taskFunctions.length < 3
 166     ) {
 167       throw new Error(
 168         `Cannot get task function worker usage for task function name '${name}' when task function names list has less than 3 elements`
 169       )
 170     }
 171     if (name === DEFAULT_TASK_NAME) {
 172       name = this.info.taskFunctions[1]
 173     }
 174     if (!this.taskFunctionsUsage.has(name)) {
 175       this.taskFunctionsUsage.set(name, this.initTaskFunctionWorkerUsage(name))
 176     }
 177     return this.taskFunctionsUsage.get(name)
 178   }
 179
 180   private async startOnEmptyQueue (): Promise<void> {
 181     if (
 182       this.onEmptyQueueCount > 0 &&
 183       this.usage.tasks.executing > 0 &&
 184       this.tasksQueue.size > 0
 185     ) {
 186       this.onEmptyQueueCount = 0
 187       return
 188     }
 189     (this.onEmptyQueue as (workerId: number) => void)(this.info.id as number)
 190     ++this.onEmptyQueueCount
 191     await sleep(exponentialDelay(this.onEmptyQueueCount))
 192     await this.startOnEmptyQueue()
 193   }
 194
 195   private initWorkerInfo (worker: Worker, workerType: WorkerType): WorkerInfo {
 196     return {
 197       id: this.getWorkerId(worker, workerType),
 198       type: workerType,
 199       dynamic: false,
 200       ready: false
 201     }
 202   }
 203
 204   private initWorkerUsage (): WorkerUsage {
 205     const getTasksQueueSize = (): number => {
 206       return this.tasksQueue.size
 207     }
 208     const getTasksQueueMaxSize = (): number => {
 209       return this.tasksQueue.maxSize
 210     }
 211     return {
 212       tasks: {
 213         executed: 0,
 214         executing: 0,
 215         get queued (): number {
 216           return getTasksQueueSize()
 217         },
 218         get maxQueued (): number {
 219           return getTasksQueueMaxSize()
 220         },
 221         stolen: 0,
 222         failed: 0
 223       },
 224       runTime: {
 225         history: new CircularArray()
 226       },
 227       waitTime: {
 228         history: new CircularArray()
 229       },
 230       elu: {
 231         idle: {
 232           history: new CircularArray()
 233         },
 234         active: {
 235           history: new CircularArray()
 236         }
 237       }
 238     }
 239   }
 240
 241   private initTaskFunctionWorkerUsage (name: string): WorkerUsage {
 242     const getTaskFunctionQueueSize = (): number => {
 243       let taskFunctionQueueSize = 0
 244       for (const task of this.tasksQueue) {
 245         if (
 246           (task.name === DEFAULT_TASK_NAME &&
 247             name === (this.info.taskFunctions as string[])[1]) ||
 248           (task.name !== DEFAULT_TASK_NAME && name === task.name)
 249         ) {
 250           ++taskFunctionQueueSize
 251         }
 252       }
 253       return taskFunctionQueueSize
 254     }
 255     return {
 256       tasks: {
 257         executed: 0,
 258         executing: 0,
 259         get queued (): number {
 260           return getTaskFunctionQueueSize()
 261         },
 262         stolen: 0,
 263         failed: 0
 264       },
 265       runTime: {
 266         history: new CircularArray()
 267       },
 268       waitTime: {
 269         history: new CircularArray()
 270       },
 271       elu: {
 272         idle: {
 273           history: new CircularArray()
 274         },
 275         active: {
 276           history: new CircularArray()
 277         }
 278       }
 279     }
 280   }
 281
 282   /**
 283    * Gets the worker id.
 284    *
 285    * @param worker - The worker.
 286    * @param workerType - The worker type.
 287    * @returns The worker id.
 288    */
 289   private getWorkerId (
 290     worker: Worker,
 291     workerType: WorkerType
 292   ): number | undefined {
 293     if (workerType === WorkerTypes.thread) {
 294       return worker.threadId
 295     } else if (workerType === WorkerTypes.cluster) {
 296       return worker.id
 297     }
 298   }
 299 }