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