feat: automate benchmarking (#1035)

* feat: automate benchmark testing

* group averages

* add percentages to report.js

* fix missing counter reset in bombingQoS1

* add README.md and examples

* update bombingQoS1

* remove benchmark examples

* improve benchmarking

* put summary first

Author: seriousme

benchmarks/README.md

@@ -0,0 +1,33 @@
+# Benchmarks
+
+This folder contains a number of scripts to perform benchmark testing and
+reporting on it:
+
+- sender.js publishes messages with a serialnumber as payload (add '-q 1' to get QoS1)
+- receiver.js subscribes and validates the serialnumber (add '-q 1' to get QoS1)
+- pingpong.js measures latency between sending and receiving the same
+message
+- server.js starts the Aedes server to use in the test
+- runBenchmarks.js starts the server and runs Publish/Subscribe tests
+with QoS0 and QoS1 using the scripts above it produces CSV data, this
+ data includes the current git branch name.
+- report.js reads the CSV data from STDIN and turns it into a Markdown
+report.
+
+Examples:
+
+```bash
+node runBenchmark.js > result.csv
+cat result.csv | node report.csv > result.md
+```
+
+```bash
+node runBenchmark.js > result.branch.csv
+cat result.main.csv result.branch.csv | node report.csv > result.combined.md
+```
+
+## WARNING
+
+Running benchmarks and especially interpreting results can be misleading
+E.g. performance of the benchmark run might depend on the presence (or absence)
+ of other, unrelated, activity in the system.

benchmarks/bombing.js

@@ -1,36 +1,116 @@
 #! /usr/bin/env node
-
+const { parseArgs } = require('node:util')
+const { hrtime } = require('node:process')
 const mqtt = require('mqtt')
-const convertHrtime = require('convert-hrtime')
-const mode = require('compute-mode')
-const client = mqtt.connect({ port: 1883, host: 'localhost', clean: true, keepalive: 0 })
+
 const interval = 5000
 
-let sent = 0
-const latencies = []
+function processsLatencies (latencies, counter) {
+  let total = 0
+  let count = 0
+  let median
+  let perc95
+  let perc99
+  const posMedian = Math.floor(counter / 2)
+  const pos95 = Math.floor(counter * 0.95)
+  const pos99 = Math.floor(counter * 0.99)
+  // sort keys from smallest to largest
+  const keys = Object.keys(latencies).sort((a, b) => a - b)
+  for (const key of keys) {
+    const value = latencies[key]
+    total += value * key
+    count += value
+
+    if (count >= posMedian && median === undefined) {
+      median = key
+    }
+    if (count >= pos95 && perc95 === undefined) {
+      perc95 = key
+    }
+    if (count >= pos99 && perc99 === undefined) {
+      perc99 = key
+    }
+  }
+  return {
+    buckets: keys.length,
+    mean: Math.floor(total / counter),
+    minimum: keys[0],
+    maximum: keys.pop(),
+    median,
+    perc95,
+    perc99,
+  }
+}
+
+let counter = 0
+let latencies = {}
+
+const { values } = parseArgs({
+  options: {
+    qos: {
+      type: 'string',
+      default: '0',
+      choices: ['0', '1'],
+      description: 'QoS level to use for publishing messages',
+      short: 'q'
+    },
+    help: {
+      type: 'boolean',
+      default: false,
+      description: 'Show this help message',
+      short: 'h'
+    }
+  }
+})
+
+if (values.help) {
+  console.log('Usage: node pingpong.js [options]')
+  console.log('Options:')
+  console.log('  -q, --qos <0|1>       QoS level to use for publishing messages (default: 0)')
+  console.log('  -h, --help            Show this help message')
+  process.exit(0)
+}
+
+if (!process.send) {
+  console.error(`Starting pingpong with options: qos=${values.qos}`)
+}
+const client = mqtt.connect({ port: 1883, host: 'localhost', clean: true, encoding: 'binary', keepalive: 0 })
+
+const qosOpts = {
+  qos: parseInt(values.qos, 10),
+}
 
 function count () {
-  console.log('sent/s', sent / interval * 1000)
-  sent = 0
+  // reset latencies while keeping the counts
+
+  const latencyResult = processsLatencies(latencies, counter)
+  latencies = {}
+  if (process.send) {
+    process.send({ type: 'latency', data: latencyResult })
+  } else {
+    console.log('latencies', latencyResult)
+  }
+  counter = 0
 }
 
 setInterval(count, interval)
 
 function publish () {
-  sent++
-  client.publish('test', JSON.stringify(process.hrtime()), { qos: 1 })
+  counter++
+  client.publish('test', process.hrtime.bigint().toString(), qosOpts)
 }
 
 function subscribe () {
-  client.subscribe('test', { qos: 1 }, publish)
+  client.subscribe('test', qosOpts, publish)
 }
 
 client.on('connect', subscribe)
 client.on('message', publish)
 client.on('message', function (topic, payload) {
-  const sentAt = JSON.parse(payload)
-  const diff = process.hrtime(sentAt)
-  latencies.push(convertHrtime(diff).ms)
+  const receivedAt = hrtime.bigint()
+  const sentAt = BigInt(payload.toString())
+  const msDiff = Math.floor(Number(receivedAt - sentAt) / 1e6) // Convert from nanoseconds to milliseconds
+  latencies[msDiff] = (latencies[msDiff] || 0) + 1
 })
 
 client.on('offline', function () {
@@ -41,13 +121,3 @@ client.on('error', function () {
   console.log('reconnect!')
   client.stream.end()
 })
-
-process.on('SIGINT', function () {
-  const total = latencies.reduce(function (acc, num) {
-    return acc + num
-  })
-  console.log('total', total)
-  console.log('average', total / latencies.length)
-  console.log('mode', mode(latencies))
-  process.exit(0)
-})

benchmarks/bombingQoS1.js

@@ -0,0 +1,78 @@
+#! /usr/bin/env node
+const { parseArgs } = require('node:util')
+const mqtt = require('mqtt')
+
+const interval = 5000
+
+let counter = 0
+let previousSerial
+
+const { values } = parseArgs({
+  options: {
+    qos: {
+      type: 'string',
+      default: '0',
+      choices: ['0', '1'],
+      description: 'QoS level to use for publishing messages',
+      short: 'q'
+    },
+    help: {
+      type: 'boolean',
+      default: false,
+      description: 'Show this help message',
+      short: 'h'
+    }
+  }
+})
+
+if (values.help) {
+  console.log('Usage: node receiver.js [options]')
+  console.log('Options:')
+  console.log('  -q, --qos <0|1>       QoS level to use for publishing messages (default: 0)')
+  console.log('  -h, --help            Show this help message')
+  process.exit(0)
+}
+
+const client = mqtt.connect({ port: 1883, host: 'localhost', clean: true, encoding: 'binary', keepalive: 0 })
+
+if (!process.send) {
+  console.error(`Starting receiver with options: qos=${values.qos}`)
+}
+const subscribeOpts = {
+  qos: parseInt(values.qos, 10),
+}
+
+function count () {
+  if (process.send) {
+    const rate = Math.floor(counter / interval * 1000)
+    process.send({ type: 'rate', data: rate })
+  } else {
+    console.log('received/s', counter / interval * 1000)
+  }
+  counter = 0
+}
+
+setInterval(count, interval)
+
+client.on('connect', function () {
+  this.subscribe('test', subscribeOpts)
+})
+
+client.handleMessage = function (packet, done) {
+  const serial = BigInt(packet.payload.toString())
+  if (previousSerial !== undefined && (serial !== (previousSerial + 1n))) {
+    console.error(`Received out of order message: expected ${previousSerial}, got ${serial}`)
+  }
+  previousSerial = serial
+  counter++
+  done()
+}
+
+client.on('offline', function () {
+  console.log('offline')
+})
+
+client.on('error', function () {
+  console.log('reconnect!')
+  client.stream.end()
+})