tests/pools/worker-node.test.mjs

   1 import { MessageChannel } from 'node:worker_threads'
   2 import { expect } from 'expect'
   3 import { WorkerNode } from '../../lib/pools/worker-node.js'
   4 import { WorkerTypes } from '../../lib/index.js'
   5 import { CircularArray } from '../../lib/circular-array.js'
   6 import { Deque } from '../../lib/deque.js'
   7 import { DEFAULT_TASK_NAME } from '../../lib/utils.js'
   8
   9 describe('Worker node test suite', () => {
  10   const threadWorkerNode = new WorkerNode(
  11     WorkerTypes.thread,
  12     './tests/worker-files/thread/testWorker.mjs',
  13     { tasksQueueBackPressureSize: 12 }
  14   )
  15   const clusterWorkerNode = new WorkerNode(
  16     WorkerTypes.cluster,
  17     './tests/worker-files/cluster/testWorker.js',
  18     { tasksQueueBackPressureSize: 12 }
  19   )
  20
  21   it('Worker node instantiation', () => {
  22     expect(() => new WorkerNode()).toThrow(
  23       new TypeError('Cannot construct a worker node without a worker type')
  24     )
  25     expect(
  26       () =>
  27         new WorkerNode(
  28           'invalidWorkerType',
  29           './tests/worker-files/thread/testWorker.mjs',
  30           { tasksQueueBackPressureSize: 12 }
  31         )
  32     ).toThrow(
  33       new TypeError(
  34         "Cannot construct a worker node with an invalid worker type 'invalidWorkerType'"
  35       )
  36     )
  37     expect(
  38       () =>
  39         new WorkerNode(
  40           WorkerTypes.thread,
  41           './tests/worker-files/thread/testWorker.mjs'
  42         )
  43     ).toThrow(
  44       new TypeError(
  45         'Cannot construct a worker node without worker node options'
  46       )
  47     )
  48     expect(
  49       () =>
  50         new WorkerNode(
  51           WorkerTypes.thread,
  52           './tests/worker-files/thread/testWorker.mjs',
  53           ''
  54         )
  55     ).toThrow(
  56       new TypeError(
  57         'Cannot construct a worker node with invalid options: must be a plain object'
  58       )
  59     )
  60     expect(
  61       () =>
  62         new WorkerNode(
  63           WorkerTypes.thread,
  64           './tests/worker-files/thread/testWorker.mjs',
  65           {}
  66         )
  67     ).toThrow(
  68       new TypeError(
  69         'Cannot construct a worker node without a tasks queue back pressure size option'
  70       )
  71     )
  72     expect(
  73       () =>
  74         new WorkerNode(
  75           WorkerTypes.thread,
  76           './tests/worker-files/thread/testWorker.mjs',
  77           { tasksQueueBackPressureSize: 'invalidTasksQueueBackPressureSize' }
  78         )
  79     ).toThrow(
  80       new TypeError(
  81         'Cannot construct a worker node with a tasks queue back pressure size option that is not an integer'
  82       )
  83     )
  84     expect(
  85       () =>
  86         new WorkerNode(
  87           WorkerTypes.thread,
  88           './tests/worker-files/thread/testWorker.mjs',
  89           { tasksQueueBackPressureSize: 0.2 }
  90         )
  91     ).toThrow(
  92       new TypeError(
  93         'Cannot construct a worker node with a tasks queue back pressure size option that is not an integer'
  94       )
  95     )
  96     expect(
  97       () =>
  98         new WorkerNode(
  99           WorkerTypes.thread,
 100           './tests/worker-files/thread/testWorker.mjs',
 101           { tasksQueueBackPressureSize: 0 }
 102         )
 103     ).toThrow(
 104       new RangeError(
 105         'Cannot construct a worker node with a tasks queue back pressure size option that is not a positive integer'
 106       )
 107     )
 108     expect(
 109       () =>
 110         new WorkerNode(
 111           WorkerTypes.thread,
 112           './tests/worker-files/thread/testWorker.mjs',
 113           { tasksQueueBackPressureSize: -1 }
 114         )
 115     ).toThrow(
 116       new RangeError(
 117         'Cannot construct a worker node with a tasks queue back pressure size option that is not a positive integer'
 118       )
 119     )
 120     expect(threadWorkerNode).toBeInstanceOf(WorkerNode)
 121     expect(threadWorkerNode.info).toStrictEqual({
 122       id: threadWorkerNode.worker.threadId,
 123       type: WorkerTypes.thread,
 124       dynamic: false,
 125       ready: false
 126     })
 127     expect(threadWorkerNode.usage).toStrictEqual({
 128       tasks: {
 129         executed: 0,
 130         executing: 0,
 131         queued: 0,
 132         maxQueued: 0,
 133         sequentiallyStolen: 0,
 134         stolen: 0,
 135         failed: 0
 136       },
 137       runTime: {
 138         history: new CircularArray()
 139       },
 140       waitTime: {
 141         history: new CircularArray()
 142       },
 143       elu: {
 144         idle: {
 145           history: new CircularArray()
 146         },
 147         active: {
 148           history: new CircularArray()
 149         }
 150       }
 151     })
 152     expect(threadWorkerNode.messageChannel).toBeInstanceOf(MessageChannel)
 153     expect(threadWorkerNode.tasksQueueBackPressureSize).toBe(12)
 154     expect(threadWorkerNode.tasksQueue).toBeInstanceOf(Deque)
 155     expect(threadWorkerNode.tasksQueue.size).toBe(0)
 156     expect(threadWorkerNode.tasksQueueSize()).toBe(
 157       threadWorkerNode.tasksQueue.size
 158     )
 159     expect(threadWorkerNode.onBackPressureStarted).toBe(false)
 160     expect(threadWorkerNode.taskFunctionsUsage).toBeInstanceOf(Map)
 161
 162     expect(clusterWorkerNode).toBeInstanceOf(WorkerNode)
 163     expect(clusterWorkerNode.info).toStrictEqual({
 164       id: clusterWorkerNode.worker.id,
 165       type: WorkerTypes.cluster,
 166       dynamic: false,
 167       ready: false
 168     })
 169     expect(clusterWorkerNode.usage).toStrictEqual({
 170       tasks: {
 171         executed: 0,
 172         executing: 0,
 173         queued: 0,
 174         maxQueued: 0,
 175         sequentiallyStolen: 0,
 176         stolen: 0,
 177         failed: 0
 178       },
 179       runTime: {
 180         history: new CircularArray()
 181       },
 182       waitTime: {
 183         history: new CircularArray()
 184       },
 185       elu: {
 186         idle: {
 187           history: new CircularArray()
 188         },
 189         active: {
 190           history: new CircularArray()
 191         }
 192       }
 193     })
 194     expect(clusterWorkerNode.messageChannel).toBeUndefined()
 195     expect(clusterWorkerNode.tasksQueueBackPressureSize).toBe(12)
 196     expect(clusterWorkerNode.tasksQueue).toBeInstanceOf(Deque)
 197     expect(clusterWorkerNode.tasksQueue.size).toBe(0)
 198     expect(clusterWorkerNode.tasksQueueSize()).toBe(
 199       clusterWorkerNode.tasksQueue.size
 200     )
 201     expect(clusterWorkerNode.onBackPressureStarted).toBe(false)
 202     expect(clusterWorkerNode.taskFunctionsUsage).toBeInstanceOf(Map)
 203   })
 204
 205   it('Worker node getTaskFunctionWorkerUsage()', () => {
 206     expect(() =>
 207       threadWorkerNode.getTaskFunctionWorkerUsage('invalidTaskFunction')
 208     ).toThrow(
 209       new TypeError(
 210         "Cannot get task function worker usage for task function name 'invalidTaskFunction' when task function names list is not yet defined"
 211       )
 212     )
 213     threadWorkerNode.info.taskFunctionNames = [DEFAULT_TASK_NAME, 'fn1']
 214     expect(() =>
 215       threadWorkerNode.getTaskFunctionWorkerUsage('invalidTaskFunction')
 216     ).toThrow(
 217       new TypeError(
 218         "Cannot get task function worker usage for task function name 'invalidTaskFunction' when task function names list has less than 3 elements"
 219       )
 220     )
 221     threadWorkerNode.info.taskFunctionNames = [DEFAULT_TASK_NAME, 'fn1', 'fn2']
 222     expect(
 223       threadWorkerNode.getTaskFunctionWorkerUsage(DEFAULT_TASK_NAME)
 224     ).toStrictEqual({
 225       tasks: {
 226         executed: 0,
 227         executing: 0,
 228         queued: 0,
 229         sequentiallyStolen: 0,
 230         stolen: 0,
 231         failed: 0
 232       },
 233       runTime: {
 234         history: new CircularArray()
 235       },
 236       waitTime: {
 237         history: new CircularArray()
 238       },
 239       elu: {
 240         idle: {
 241           history: new CircularArray()
 242         },
 243         active: {
 244           history: new CircularArray()
 245         }
 246       }
 247     })
 248     expect(threadWorkerNode.getTaskFunctionWorkerUsage('fn1')).toStrictEqual({
 249       tasks: {
 250         executed: 0,
 251         executing: 0,
 252         queued: 0,
 253         sequentiallyStolen: 0,
 254         stolen: 0,
 255         failed: 0
 256       },
 257       runTime: {
 258         history: new CircularArray()
 259       },
 260       waitTime: {
 261         history: new CircularArray()
 262       },
 263       elu: {
 264         idle: {
 265           history: new CircularArray()
 266         },
 267         active: {
 268           history: new CircularArray()
 269         }
 270       }
 271     })
 272     expect(threadWorkerNode.getTaskFunctionWorkerUsage('fn2')).toStrictEqual({
 273       tasks: {
 274         executed: 0,
 275         executing: 0,
 276         queued: 0,
 277         sequentiallyStolen: 0,
 278         stolen: 0,
 279         failed: 0
 280       },
 281       runTime: {
 282         history: new CircularArray()
 283       },
 284       waitTime: {
 285         history: new CircularArray()
 286       },
 287       elu: {
 288         idle: {
 289           history: new CircularArray()
 290         },
 291         active: {
 292           history: new CircularArray()
 293         }
 294       }
 295     })
 296     expect(threadWorkerNode.taskFunctionsUsage.size).toBe(2)
 297   })
 298
 299   it('Worker node deleteTaskFunctionWorkerUsage()', () => {
 300     expect(threadWorkerNode.info.taskFunctionNames).toStrictEqual([
 301       DEFAULT_TASK_NAME,
 302       'fn1',
 303       'fn2'
 304     ])
 305     expect(threadWorkerNode.taskFunctionsUsage.size).toBe(2)
 306     expect(
 307       threadWorkerNode.deleteTaskFunctionWorkerUsage('invalidTaskFunction')
 308     ).toBe(false)
 309     expect(threadWorkerNode.taskFunctionsUsage.size).toBe(2)
 310     expect(threadWorkerNode.deleteTaskFunctionWorkerUsage('fn1')).toBe(true)
 311     expect(threadWorkerNode.taskFunctionsUsage.size).toBe(1)
 312   })
 313 })