benchmarks/benchmarks-utils.cjs

   1 const { randomInt } = require('node:crypto')
   2 const { strictEqual } = require('node:assert')
   3 const {
   4   existsSync,
   5   mkdirSync,
   6   readFileSync,
   7   rmSync,
   8   writeFileSync
   9 } = require('node:fs')
  10 const Benchmark = require('benchmark')
  11 const {
  12   DynamicClusterPool,
  13   DynamicThreadPool,
  14   FixedClusterPool,
  15   FixedThreadPool,
  16   Measurements,
  17   PoolTypes,
  18   WorkerChoiceStrategies,
  19   WorkerTypes
  20 } = require('../lib/index.cjs')
  21 const { TaskFunctions } = require('./benchmarks-types.cjs')
  22
  23 const buildPoolifierPool = (workerType, poolType, poolSize, poolOptions) => {
  24   switch (poolType) {
  25     case PoolTypes.fixed:
  26       switch (workerType) {
  27         case WorkerTypes.thread:
  28           return new FixedThreadPool(
  29             poolSize,
  30             './benchmarks/internal/thread-worker.mjs',
  31             poolOptions
  32           )
  33         case WorkerTypes.cluster:
  34           return new FixedClusterPool(
  35             poolSize,
  36             './benchmarks/internal/cluster-worker.cjs',
  37             poolOptions
  38           )
  39       }
  40       break
  41     case PoolTypes.dynamic:
  42       switch (workerType) {
  43         case WorkerTypes.thread:
  44           return new DynamicThreadPool(
  45             Math.floor(poolSize / 2),
  46             poolSize,
  47             './benchmarks/internal/thread-worker.mjs',
  48             poolOptions
  49           )
  50         case WorkerTypes.cluster:
  51           return new DynamicClusterPool(
  52             Math.floor(poolSize / 2),
  53             poolSize,
  54             './benchmarks/internal/cluster-worker.cjs',
  55             poolOptions
  56           )
  57       }
  58       break
  59   }
  60 }
  61
  62 const runPoolifierPool = async (pool, { taskExecutions, workerData }) => {
  63   return await new Promise((resolve, reject) => {
  64     let executions = 0
  65     for (let i = 1; i <= taskExecutions; i++) {
  66       pool
  67         .execute(workerData)
  68         .then(() => {
  69           ++executions
  70           if (executions === taskExecutions) {
  71             resolve({ ok: 1 })
  72           }
  73           return undefined
  74         })
  75         .catch(err => {
  76           console.error(err)
  77           reject(err)
  78         })
  79     }
  80   })
  81 }
  82
  83 const runPoolifierPoolBenchmark = async (
  84   name,
  85   workerType,
  86   poolType,
  87   poolSize,
  88   { taskExecutions, workerData }
  89 ) => {
  90   return await new Promise((resolve, reject) => {
  91     const pool = buildPoolifierPool(workerType, poolType, poolSize)
  92     try {
  93       const suite = new Benchmark.Suite(name)
  94       for (const workerChoiceStrategy of Object.values(
  95         WorkerChoiceStrategies
  96       )) {
  97         for (const enableTasksQueue of [false, true]) {
  98           if (workerChoiceStrategy === WorkerChoiceStrategies.FAIR_SHARE) {
  99             for (const measurement of [
 100               Measurements.runTime,
 101               Measurements.elu
 102             ]) {
 103               suite.add(
 104                 `${name} with ${workerChoiceStrategy}, with ${measurement} and ${
 105                   enableTasksQueue ? 'with' : 'without'
 106                 } tasks queue`,
 107                 async () => {
 108                   pool.setWorkerChoiceStrategy(workerChoiceStrategy, {
 109                     measurement
 110                   })
 111                   pool.enableTasksQueue(enableTasksQueue)
 112                   strictEqual(
 113                     pool.opts.workerChoiceStrategy,
 114                     workerChoiceStrategy
 115                   )
 116                   strictEqual(pool.opts.enableTasksQueue, enableTasksQueue)
 117                   strictEqual(
 118                     pool.opts.workerChoiceStrategyOptions.measurement,
 119                     measurement
 120                   )
 121                   await runPoolifierPool(pool, {
 122                     taskExecutions,
 123                     workerData
 124                   })
 125                 }
 126               )
 127             }
 128           } else {
 129             suite.add(
 130               `${name} with ${workerChoiceStrategy} and ${
 131                 enableTasksQueue ? 'with' : 'without'
 132               } tasks queue`,
 133               async () => {
 134                 pool.setWorkerChoiceStrategy(workerChoiceStrategy)
 135                 pool.enableTasksQueue(enableTasksQueue)
 136                 strictEqual(
 137                   pool.opts.workerChoiceStrategy,
 138                   workerChoiceStrategy
 139                 )
 140                 strictEqual(pool.opts.enableTasksQueue, enableTasksQueue)
 141                 await runPoolifierPool(pool, {
 142                   taskExecutions,
 143                   workerData
 144                 })
 145               }
 146             )
 147           }
 148         }
 149       }
 150       suite
 151         .on('cycle', event => {
 152           console.info(event.target.toString())
 153         })
 154         .on('complete', function () {
 155           console.info(
 156             'Fastest is ' +
 157               LIST_FORMATTER.format(this.filter('fastest').map('name'))
 158           )
 159           const destroyTimeout = setTimeout(() => {
 160             console.error('Pool destroy timeout reached (30s)')
 161             resolve()
 162           }, 30000)
 163           pool
 164             .destroy()
 165             .then(resolve)
 166             .catch(reject)
 167             .finally(() => {
 168               clearTimeout(destroyTimeout)
 169             })
 170             .catch(() => {})
 171         })
 172         .run({ async: true })
 173     } catch (error) {
 174       pool
 175         .destroy()
 176         .then(() => {
 177           return reject(error)
 178         })
 179         .catch(() => {})
 180     }
 181   })
 182 }
 183
 184 const LIST_FORMATTER = new Intl.ListFormat('en-US', {
 185   style: 'long',
 186   type: 'conjunction'
 187 })
 188
 189 const jsonIntegerSerialization = n => {
 190   for (let i = 0; i < n; i++) {
 191     const o = {
 192       a: i
 193     }
 194     JSON.stringify(o)
 195   }
 196   return { ok: 1 }
 197 }
 198
 199 /**
 200  * @param {number} n - The number of fibonacci numbers to generate.
 201  * @returns {number} - The nth fibonacci number.
 202  */
 203 const fibonacci = n => {
 204   n = BigInt(n)
 205   let current = 1n
 206   let previous = 0n
 207   while (--n) {
 208     const tmp = current
 209     current += previous
 210     previous = tmp
 211   }
 212   return current
 213 }
 214
 215 /**
 216  * @param {number} n - The number to calculate the factorial of.
 217  * @returns {number} - The factorial of n.
 218  */
 219 const factorial = n => {
 220   if (n === 0 || n === 1) {
 221     return 1n
 222   } else {
 223     n = BigInt(n)
 224     let factorial = 1n
 225     for (let i = 1n; i <= n; i++) {
 226       factorial *= i
 227     }
 228     return factorial
 229   }
 230 }
 231
 232 const readWriteFiles = (
 233   n,
 234   baseDirectory = `/tmp/poolifier-benchmarks/${randomInt(281474976710655)}`
 235 ) => {
 236   if (existsSync(baseDirectory) === true) {
 237     rmSync(baseDirectory, { recursive: true })
 238   }
 239   mkdirSync(baseDirectory, { recursive: true })
 240   for (let i = 0; i < n; i++) {
 241     const filePath = `${baseDirectory}/${i}`
 242     writeFileSync(filePath, i.toString(), {
 243       encoding: 'utf8',
 244       flag: 'a'
 245     })
 246     readFileSync(filePath, 'utf8')
 247   }
 248   rmSync(baseDirectory, { recursive: true })
 249   return { ok: 1 }
 250 }
 251
 252 const executeTaskFunction = data => {
 253   switch (data.function) {
 254     case TaskFunctions.jsonIntegerSerialization:
 255       return jsonIntegerSerialization(data.taskSize || 1000)
 256     case TaskFunctions.fibonacci:
 257       return fibonacci(data.taskSize || 1000)
 258     case TaskFunctions.factorial:
 259       return factorial(data.taskSize || 1000)
 260     case TaskFunctions.readWriteFiles:
 261       return readWriteFiles(data.taskSize || 1000)
 262     default:
 263       throw new Error('Unknown task function')
 264   }
 265 }
 266
 267 module.exports = {
 268   LIST_FORMATTER,
 269   executeTaskFunction,
 270   runPoolifierPoolBenchmark
 271 }