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  * Intentionally inefficient implementation.
 201  * @param {number} n - The number of fibonacci numbers to generate.
 202  * @returns {number} - The nth fibonacci number.
 203  */
 204 const fibonacci = n => {
 205   if (n <= 1) return n
 206   return fibonacci(n - 1) + fibonacci(n - 2)
 207 }
 208
 209 /**
 210  * Intentionally inefficient implementation.
 211  * @param {number} n - The number to calculate the factorial of.
 212  * @returns {number} - The factorial of n.
 213  */
 214 const factorial = n => {
 215   if (n === 0) {
 216     return 1
 217   }
 218   return factorial(n - 1) * n
 219 }
 220
 221 const readWriteFiles = (
 222   n,
 223   baseDirectory = `/tmp/poolifier-benchmarks/${randomInt(281474976710655)}`
 224 ) => {
 225   if (existsSync(baseDirectory) === true) {
 226     rmSync(baseDirectory, { recursive: true })
 227   }
 228   mkdirSync(baseDirectory, { recursive: true })
 229   for (let i = 0; i < n; i++) {
 230     const filePath = `${baseDirectory}/${i}`
 231     writeFileSync(filePath, i.toString(), {
 232       encoding: 'utf8',
 233       flag: 'a'
 234     })
 235     readFileSync(filePath, 'utf8')
 236   }
 237   rmSync(baseDirectory, { recursive: true })
 238   return { ok: 1 }
 239 }
 240
 241 const executeTaskFunction = data => {
 242   switch (data.function) {
 243     case TaskFunctions.jsonIntegerSerialization:
 244       return jsonIntegerSerialization(data.taskSize || 1000)
 245     case TaskFunctions.fibonacci:
 246       return fibonacci(data.taskSize || 1000)
 247     case TaskFunctions.factorial:
 248       return factorial(data.taskSize || 1000)
 249     case TaskFunctions.readWriteFiles:
 250       return readWriteFiles(data.taskSize || 1000)
 251     default:
 252       throw new Error('Unknown task function')
 253   }
 254 }
 255
 256 module.exports = {
 257   LIST_FORMATTER,
 258   executeTaskFunction,
 259   runPoolifierPoolBenchmark
 260 }