benchmarks/benchmarks-utils.mjs

   1 import { strictEqual } from 'node:assert'
   2
   3 import Benchmark from 'benchmark'
   4 import { bench, clear, group, run } from 'tatami-ng'
   5
   6 import {
   7   DynamicClusterPool,
   8   DynamicThreadPool,
   9   FixedClusterPool,
  10   FixedThreadPool,
  11   Measurements,
  12   PoolTypes,
  13   WorkerChoiceStrategies,
  14   WorkerTypes
  15 } from '../lib/index.mjs'
  16 import { executeTaskFunction } from './benchmarks-utils.cjs'
  17
  18 const buildPoolifierPool = (workerType, poolType, poolSize, poolOptions) => {
  19   switch (poolType) {
  20     case PoolTypes.fixed:
  21       switch (workerType) {
  22         case WorkerTypes.thread:
  23           return new FixedThreadPool(
  24             poolSize,
  25             './benchmarks/internal/thread-worker.mjs',
  26             poolOptions
  27           )
  28         case WorkerTypes.cluster:
  29           return new FixedClusterPool(
  30             poolSize,
  31             './benchmarks/internal/cluster-worker.cjs',
  32             poolOptions
  33           )
  34       }
  35       break
  36     case PoolTypes.dynamic:
  37       switch (workerType) {
  38         case WorkerTypes.thread:
  39           return new DynamicThreadPool(
  40             Math.floor(poolSize / 2),
  41             poolSize,
  42             './benchmarks/internal/thread-worker.mjs',
  43             poolOptions
  44           )
  45         case WorkerTypes.cluster:
  46           return new DynamicClusterPool(
  47             Math.floor(poolSize / 2),
  48             poolSize,
  49             './benchmarks/internal/cluster-worker.cjs',
  50             poolOptions
  51           )
  52       }
  53       break
  54   }
  55 }
  56
  57 const runPoolifierPool = async (pool, { taskExecutions, workerData }) => {
  58   return await new Promise((resolve, reject) => {
  59     let executions = 0
  60     for (let i = 1; i <= taskExecutions; i++) {
  61       pool
  62         .execute(workerData)
  63         .then(() => {
  64           ++executions
  65           if (executions === taskExecutions) {
  66             resolve({ ok: 1 })
  67           }
  68           return undefined
  69         })
  70         .catch(err => {
  71           console.error(err)
  72           reject(err)
  73         })
  74     }
  75   })
  76 }
  77
  78 export const runPoolifierBenchmarkBenchmarkJs = async (
  79   name,
  80   workerType,
  81   poolType,
  82   poolSize,
  83   poolOptions,
  84   { taskExecutions, workerData }
  85 ) => {
  86   return await new Promise((resolve, reject) => {
  87     const pool = buildPoolifierPool(workerType, poolType, poolSize, poolOptions)
  88     let workerChoiceStrategy
  89     let enableTasksQueue
  90     let workerChoiceStrategyOptions
  91     if (poolOptions != null) {
  92       ({
  93         workerChoiceStrategy,
  94         enableTasksQueue,
  95         workerChoiceStrategyOptions
  96       } = poolOptions)
  97     }
  98     const measurement = workerChoiceStrategyOptions?.measurement
  99     new Benchmark(
 100       `${name} with ${workerChoiceStrategy ?? pool.opts.workerChoiceStrategy}${
 101         measurement != null ? `, with ${measurement}` : ''
 102       } and ${enableTasksQueue ? 'with' : 'without'} tasks queue`,
 103       async () => {
 104         await runPoolifierPool(pool, {
 105           taskExecutions,
 106           workerData
 107         })
 108       },
 109       {
 110         onStart: () => {
 111           if (workerChoiceStrategy != null) {
 112             strictEqual(pool.opts.workerChoiceStrategy, workerChoiceStrategy)
 113           }
 114           if (enableTasksQueue != null) {
 115             strictEqual(pool.opts.enableTasksQueue, enableTasksQueue)
 116           }
 117           if (measurement != null) {
 118             strictEqual(
 119               pool.opts.workerChoiceStrategyOptions.measurement,
 120               measurement
 121             )
 122           }
 123         },
 124         onComplete: event => {
 125           console.info(event.target.toString())
 126           if (pool.started && !pool.destroying) {
 127             pool.destroy().then(resolve).catch(reject)
 128           } else {
 129             resolve()
 130           }
 131         },
 132         onError: event => {
 133           if (pool.started && !pool.destroying) {
 134             pool
 135               .destroy()
 136               .then(() => {
 137                 return reject(event.target.error)
 138               })
 139               .catch(() => {})
 140           } else {
 141             reject(event.target.error)
 142           }
 143         }
 144       }
 145     ).run({ async: true })
 146   })
 147 }
 148
 149 export const runPoolifierBenchmarkBenchmarkJsSuite = async (
 150   name,
 151   workerType,
 152   poolType,
 153   poolSize,
 154   { taskExecutions, workerData }
 155 ) => {
 156   return await new Promise((resolve, reject) => {
 157     const pool = buildPoolifierPool(workerType, poolType, poolSize)
 158     const suite = new Benchmark.Suite(name, {
 159       onComplete: () => {
 160         if (pool.started && !pool.destroying) {
 161           pool.destroy().then(resolve).catch(reject)
 162         } else {
 163           resolve()
 164         }
 165       },
 166       onCycle: event => {
 167         console.info(event.target.toString())
 168       },
 169       onError: event => {
 170         if (pool.started && !pool.destroying) {
 171           pool
 172             .destroy()
 173             .then(() => {
 174               return reject(event.target.error)
 175             })
 176             .catch(() => {})
 177         } else {
 178           reject(event.target.error)
 179         }
 180       }
 181     })
 182     for (const workerChoiceStrategy of Object.values(WorkerChoiceStrategies)) {
 183       for (const enableTasksQueue of [false, true]) {
 184         if (workerChoiceStrategy === WorkerChoiceStrategies.FAIR_SHARE) {
 185           for (const measurement of [Measurements.runTime, Measurements.elu]) {
 186             suite.add(
 187               `${name} with ${workerChoiceStrategy}, with ${measurement} and ${
 188                 enableTasksQueue ? 'with' : 'without'
 189               } tasks queue`,
 190               async () => {
 191                 await runPoolifierPool(pool, {
 192                   taskExecutions,
 193                   workerData
 194                 })
 195               },
 196               {
 197                 onStart: () => {
 198                   pool.setWorkerChoiceStrategy(workerChoiceStrategy, {
 199                     measurement
 200                   })
 201                   pool.enableTasksQueue(enableTasksQueue)
 202                   strictEqual(
 203                     pool.opts.workerChoiceStrategy,
 204                     workerChoiceStrategy
 205                   )
 206                   strictEqual(pool.opts.enableTasksQueue, enableTasksQueue)
 207                   strictEqual(
 208                     pool.opts.workerChoiceStrategyOptions.measurement,
 209                     measurement
 210                   )
 211                 }
 212               }
 213             )
 214           }
 215         } else {
 216           suite.add(
 217             `${name} with ${workerChoiceStrategy} and ${
 218               enableTasksQueue ? 'with' : 'without'
 219             } tasks queue`,
 220             async () => {
 221               await runPoolifierPool(pool, {
 222                 taskExecutions,
 223                 workerData
 224               })
 225             },
 226             {
 227               onStart: () => {
 228                 pool.setWorkerChoiceStrategy(workerChoiceStrategy)
 229                 pool.enableTasksQueue(enableTasksQueue)
 230                 strictEqual(
 231                   pool.opts.workerChoiceStrategy,
 232                   workerChoiceStrategy
 233                 )
 234                 strictEqual(pool.opts.enableTasksQueue, enableTasksQueue)
 235               }
 236             }
 237           )
 238         }
 239       }
 240     }
 241     suite
 242       .on('complete', function () {
 243         console.info(
 244           'Fastest is ' +
 245             LIST_FORMATTER.format(this.filter('fastest').map('name'))
 246         )
 247       })
 248       .run({ async: true })
 249   })
 250 }
 251
 252 export const runPoolifierBenchmarkTatamiNg = async (
 253   name,
 254   workerType,
 255   poolType,
 256   poolSize,
 257   { taskExecutions, workerData }
 258 ) => {
 259   try {
 260     const pool = buildPoolifierPool(workerType, poolType, poolSize)
 261     for (const workerChoiceStrategy of Object.values(WorkerChoiceStrategies)) {
 262       for (const enableTasksQueue of [false, true]) {
 263         if (workerChoiceStrategy === WorkerChoiceStrategies.FAIR_SHARE) {
 264           for (const measurement of [Measurements.runTime, Measurements.elu]) {
 265             group(name, () => {
 266               bench(
 267                 `${name} with ${workerChoiceStrategy}, with ${measurement} and ${
 268                   enableTasksQueue ? 'with' : 'without'
 269                 } tasks queue`,
 270                 async () => {
 271                   await runPoolifierPool(pool, {
 272                     taskExecutions,
 273                     workerData
 274                   })
 275                 },
 276                 {
 277                   before: () => {
 278                     pool.setWorkerChoiceStrategy(workerChoiceStrategy, {
 279                       measurement
 280                     })
 281                     pool.enableTasksQueue(enableTasksQueue)
 282                     strictEqual(
 283                       pool.opts.workerChoiceStrategy,
 284                       workerChoiceStrategy
 285                     )
 286                     strictEqual(pool.opts.enableTasksQueue, enableTasksQueue)
 287                     strictEqual(
 288                       pool.opts.workerChoiceStrategyOptions.measurement,
 289                       measurement
 290                     )
 291                   }
 292                 }
 293               )
 294             })
 295           }
 296         } else {
 297           group(name, () => {
 298             bench(
 299               `${name} with ${workerChoiceStrategy} and ${
 300                 enableTasksQueue ? 'with' : 'without'
 301               } tasks queue`,
 302               async () => {
 303                 await runPoolifierPool(pool, {
 304                   taskExecutions,
 305                   workerData
 306                 })
 307               },
 308               {
 309                 before: () => {
 310                   pool.setWorkerChoiceStrategy(workerChoiceStrategy)
 311                   pool.enableTasksQueue(enableTasksQueue)
 312                   strictEqual(
 313                     pool.opts.workerChoiceStrategy,
 314                     workerChoiceStrategy
 315                   )
 316                   strictEqual(pool.opts.enableTasksQueue, enableTasksQueue)
 317                 }
 318               }
 319             )
 320           })
 321         }
 322       }
 323     }
 324     await run()
 325     clear()
 326     await pool.destroy()
 327   } catch (error) {
 328     console.error(error)
 329   }
 330 }
 331
 332 const LIST_FORMATTER = new Intl.ListFormat('en-US', {
 333   style: 'long',
 334   type: 'conjunction'
 335 })
 336
 337 export { executeTaskFunction }