src/pools/worker-node.ts

   1 import { CircularArray } from '../circular-array'
   2 import { Queue } from '../queue'
   3 import {
   4   type IWorker,
   5   type IWorkerNode,
   6   type Task,
   7   type WorkerInfo,
   8   type WorkerType,
   9   WorkerTypes,
  10   type WorkerUsage
  11 } from './worker'
  12
  13 export class WorkerNode<Worker extends IWorker, Data = unknown>
  14 implements IWorkerNode<Worker, Data> {
  15   public readonly worker: Worker
  16   public readonly info: WorkerInfo
  17   public usage: WorkerUsage
  18   private readonly tasksQueue: Queue<Task<Data>>
  19
  20   constructor (worker: Worker, workerType: WorkerType) {
  21     this.worker = worker
  22     this.info = this.initWorkerInfo(worker, workerType)
  23     this.usage = this.initWorkerUsage()
  24     this.tasksQueue = new Queue<Task<Data>>()
  25   }
  26
  27   /** @inheritdoc */
  28   public tasksQueueSize (): number {
  29     return this.tasksQueue.size
  30   }
  31
  32   /**
  33    * Worker node tasks queue maximum size.
  34    *
  35    * @returns The tasks queue maximum size.
  36    */
  37   private tasksQueueMaxSize (): number {
  38     return this.tasksQueue.maxSize
  39   }
  40
  41   /** @inheritdoc */
  42   public enqueueTask (task: Task<Data>): number {
  43     return this.tasksQueue.enqueue(task)
  44   }
  45
  46   /** @inheritdoc */
  47   public dequeueTask (): Task<Data> | undefined {
  48     return this.tasksQueue.dequeue()
  49   }
  50
  51   /** @inheritdoc */
  52   public clearTasksQueue (): void {
  53     this.tasksQueue.clear()
  54   }
  55
  56   public resetUsage (): void {
  57     this.usage = this.initWorkerUsage()
  58   }
  59
  60   private initWorkerInfo (worker: Worker, workerType: WorkerType): WorkerInfo {
  61     return {
  62       id: this.getWorkerId(worker, workerType),
  63       type: workerType,
  64       dynamic: false,
  65       started: true
  66     }
  67   }
  68
  69   private initWorkerUsage (): WorkerUsage {
  70     const getTasksQueueSize = (): number => {
  71       return this.tasksQueueSize()
  72     }
  73     const getTasksMaxQueueSize = (): number => {
  74       return this.tasksQueueMaxSize()
  75     }
  76     return {
  77       tasks: {
  78         executed: 0,
  79         executing: 0,
  80         get queued (): number {
  81           return getTasksQueueSize()
  82         },
  83         get maxQueued (): number {
  84           return getTasksMaxQueueSize()
  85         },
  86         failed: 0
  87       },
  88       runTime: {
  89         history: new CircularArray()
  90       },
  91       waitTime: {
  92         history: new CircularArray()
  93       },
  94       elu: {
  95         idle: {
  96           history: new CircularArray()
  97         },
  98         active: {
  99           history: new CircularArray()
 100         }
 101       }
 102     }
 103   }
 104
 105   /**
 106    * Gets the worker id.
 107    *
 108    * @param worker - The worker.
 109    * @returns The worker id.
 110    */
 111   private getWorkerId (
 112     worker: Worker,
 113     workerType: WorkerType
 114   ): number | undefined {
 115     if (workerType === WorkerTypes.thread) {
 116       return worker.threadId
 117     } else if (workerType === WorkerTypes.cluster) {
 118       return worker.id
 119     }
 120   }
 121 }