tests/pools/worker-node.test.mjs

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