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